diff --git a/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java b/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java index 1255328847d..40fc6054fd8 100644 --- a/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java +++ b/ingestion/src/main/java/feast/ingestion/transform/fn/ValidateFeatureRowsDoFn.java @@ -46,6 +46,7 @@ public class ValidateFeatureRowsDoFn extends BaseFeatureDoFn { + private static final String EMPTY_STORE = ""; private final List featureIds = new ArrayList<>(); private Set supportedServingTypes = new HashSet<>(); @@ -108,10 +109,13 @@ public void processElementImpl(ProcessContext context) { String.format("Serving storage type=%s not supported", servingStorageSpec.getType())); String warehouseStoreId = featureSpec.getDataStores().getWarehouse().getId(); - StorageSpec warehouseStorageSpec = specs.getStorageSpec(warehouseStoreId); - checkArgument( - supportedWarehouseTypes.contains(warehouseStorageSpec.getType()), - String.format("Warehouse storage type=%s not supported", servingStorageSpec.getType())); + if (!warehouseStoreId.equals(EMPTY_STORE)) { + StorageSpec warehouseStorageSpec = specs.getStorageSpec(warehouseStoreId); + checkArgument( + supportedWarehouseTypes.contains(warehouseStorageSpec.getType()), + String.format( + "Warehouse storage type=%s not supported", servingStorageSpec.getType())); + } checkArgument( featureSpec.getEntity().equals(row.getEntityName()), diff --git a/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java b/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java index 82133a382d8..61f44f4d44b 100644 --- a/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java +++ b/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java @@ -276,4 +276,59 @@ public void testImportWithErrors() throws IOException { PAssert.that(writtenToServing).satisfies(hasCount(0)); testPipeline.run(); } + + + @Test + public void testImportWithoutWarehouseStore() throws IOException { + ImportSpec importSpec = + ProtoUtil.decodeProtoYaml( + "---\n" + + "type: file\n" + + "options:\n" + + " format: csv\n" + + " # path: to be overwritten in tests\n" + + "entities:\n" + + " - testEntity\n" + + "schema:\n" + + " entityIdColumn: id\n" + + " timestampValue: 2018-09-25T00:00:00.000Z\n" + + " fields:\n" + + " - name: id\n" + + " - featureId: testEntity.none.testInt64NoWarehouse\n" + + " - featureId: testEntity.none.testStringNoWarehouse\n" + + "\n", + ImportSpec.getDefaultInstance()); + + File csvFile = folder.newFile("data.csv"); + + // Note the string and integer features are in the wrong positions for the import spec. + Files.asCharSink(csvFile, Charsets.UTF_8).write("1,101,a\n2,202,b\n3,303,c\n"); + importSpec = initImportSpec(importSpec, csvFile.toString()); + + ImportJobOptions options = initOptions(); + options.setErrorsStoreType(MOCK_ERRORS_STORE_TYPE); + + Injector injector = + Guice.createInjector( + new ImportJobModule(options, importSpec), new TestPipelineModule(testPipeline)); + + ImportJob job = injector.getInstance(ImportJob.class); + + injector.getInstance(ImportJob.class); + job.expand(); + + PCollection writtenToServing = + PCollectionList.of(ServingStoreService.get(MockServingStore.class).getWrite().getInputs()) + .apply("flatten serving input", Flatten.pCollections()); + + PCollection writtenToErrors = + PCollectionList.of(ErrorsStoreService.get(MockErrorsStore.class).getWrite().getInputs()) + .apply("flatten errors input", Flatten.pCollections()); + + PAssert.that(writtenToErrors) + .satisfies(hasCount(0)); + + PAssert.that(writtenToServing).satisfies(hasCount(3)); + testPipeline.run(); + } } diff --git a/ingestion/src/test/resources/core_specs/feature/testEntity.none.testInt64NoWarehouse.json b/ingestion/src/test/resources/core_specs/feature/testEntity.none.testInt64NoWarehouse.json new file mode 100644 index 00000000000..e12d5fdcb49 --- /dev/null +++ b/ingestion/src/test/resources/core_specs/feature/testEntity.none.testInt64NoWarehouse.json @@ -0,0 +1,17 @@ +{ + "id": "testEntity.none.testInt64NoWarehouse", + "entity": "testEntity", + "granularity": "NONE", + "name": "testInt64NoWarehouse", + "owner": "feast@example.com", + "description": "This is test feature of type integer", + "uri": "https://example.com/", + "valueType": "INT64", + "tags": [], + "options": {}, + "dataStores": { + "serving": { + "id": "TEST_SERVING" + } + } +} \ No newline at end of file diff --git a/ingestion/src/test/resources/core_specs/feature/testEntity.none.testStringNoWarehouse.json b/ingestion/src/test/resources/core_specs/feature/testEntity.none.testStringNoWarehouse.json new file mode 100644 index 00000000000..2b7f7f87e49 --- /dev/null +++ b/ingestion/src/test/resources/core_specs/feature/testEntity.none.testStringNoWarehouse.json @@ -0,0 +1,15 @@ +{ + "id": "testEntity.none.testStringNoWarehouse", + "entity": "testEntity", + "granularity": "NONE", + "name": "testStringNoWarehouse", + "owner": "feast@example.com", + "description": "This is test feature of type integer", + "uri": "https://example.com/", + "valueType": "STRING", + "dataStores": { + "serving": { + "id": "TEST_SERVING" + } + } +}