Fix validating feature row when the associated feature spec has no warehouse store

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<String> featureIds = new ArrayList<>();
 
   private Set<String> 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()),

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<FeatureRowExtended> writtenToServing =
+        PCollectionList.of(ServingStoreService.get(MockServingStore.class).getWrite().getInputs())
+            .apply("flatten serving input", Flatten.pCollections());
+
+    PCollection<FeatureRowExtended> 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();
+  }
 }

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": "[email protected]",
+  "description": "This is test feature of type integer",
+  "uri": "https://example.com/",
+  "valueType": "INT64",
+  "tags": [],
+  "options": {},
+  "dataStores": {
+    "serving": {
+      "id": "TEST_SERVING"
+    }
+  }
+}

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": "[email protected]",
+  "description": "This is test feature of type integer",
+  "uri": "https://example.com/",
+  "valueType": "STRING",
+  "dataStores": {
+    "serving": {
+      "id": "TEST_SERVING"
+    }
+  }
+}