diff --git a/core/src/main/java/feast/core/validators/SpecValidator.java b/core/src/main/java/feast/core/validators/SpecValidator.java index b91c4923ea8..08aaab7ad5d 100644 --- a/core/src/main/java/feast/core/validators/SpecValidator.java +++ b/core/src/main/java/feast/core/validators/SpecValidator.java @@ -17,10 +17,6 @@ package feast.core.validators; -import static com.google.common.base.Preconditions.checkArgument; -import static com.google.common.base.Preconditions.checkNotNull; -import static feast.core.validators.Matchers.checkLowerSnakeCase; - import com.google.common.base.Strings; import com.google.common.collect.Lists; import feast.core.dao.EntityInfoRepository; @@ -28,6 +24,7 @@ import feast.core.dao.FeatureInfoRepository; import feast.core.dao.StorageInfoRepository; import feast.core.model.FeatureGroupInfo; +import feast.core.model.StorageInfo; import feast.core.storage.BigQueryStorageManager; import feast.core.storage.BigTableStorageManager; import feast.core.storage.PostgresStorageManager; @@ -38,9 +35,16 @@ import feast.specs.ImportSpecProto.Field; import feast.specs.ImportSpecProto.ImportSpec; import feast.specs.StorageSpecProto.StorageSpec; +import org.springframework.beans.factory.annotation.Autowired; + import java.util.Arrays; import java.util.Optional; -import org.springframework.beans.factory.annotation.Autowired; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; +import static feast.core.validators.Matchers.checkLowerSnakeCase; public class SpecValidator { @@ -50,14 +54,14 @@ public class SpecValidator { private FeatureInfoRepository featureInfoRepository; private static final String FILE_ERROR_STORE_TYPE = "file.json"; - private String[] supportedStorageTypes = - new String[]{ - BigQueryStorageManager.TYPE, - BigTableStorageManager.TYPE, - PostgresStorageManager.TYPE, - RedisStorageManager.TYPE, - BigQueryStorageManager.TYPE, - FILE_ERROR_STORE_TYPE + private static String[] SUPPORTED_WAREHOUSE_STORES = + new String[] { + BigQueryStorageManager.TYPE, FILE_ERROR_STORE_TYPE, + }; + + private static String[] SUPPORTED_SERVING_STORES = + new String[] { + BigTableStorageManager.TYPE, PostgresStorageManager.TYPE, RedisStorageManager.TYPE, }; @Autowired @@ -126,12 +130,21 @@ public void validateFeatureSpec(FeatureSpec spec) throws IllegalArgumentExceptio warehouseStoreId = warehouseStoreId.equals("") ? group.getWarehouseStore().getId() : warehouseStoreId; } + Optional servingStore = storageInfoRepository.findById(servingStoreId); + Optional warehouseStore = storageInfoRepository.findById(warehouseStoreId); checkArgument( - storageInfoRepository.existsById(servingStoreId), + servingStore.isPresent(), Strings.lenientFormat("Serving store with id %s does not exist", servingStoreId)); checkArgument( - storageInfoRepository.existsById(warehouseStoreId), + Arrays.asList(SUPPORTED_SERVING_STORES).contains(servingStore.get().getType()), + Strings.lenientFormat("Unsupported serving store type", servingStore.get().getType())); + checkArgument( + warehouseStore.isPresent(), Strings.lenientFormat("Warehouse store with id %s does not exist", warehouseStoreId)); + checkArgument( + Arrays.asList(SUPPORTED_WAREHOUSE_STORES).contains(warehouseStore.get().getType()), + Strings.lenientFormat( + "Unsupported warehouse store type", warehouseStore.get().getType())); } catch (NullPointerException | IllegalArgumentException e) { throw new IllegalArgumentException( @@ -186,7 +199,12 @@ public void validateStorageSpec(StorageSpec spec) throws IllegalArgumentExceptio try { checkArgument(!spec.getId().equals(""), "Id field cannot be empty"); Matchers.checkUpperSnakeCase(spec.getId(), "Id"); - checkArgument(Arrays.asList(supportedStorageTypes).contains(spec.getType())); + checkArgument( + Stream.concat( + Arrays.stream(SUPPORTED_SERVING_STORES), + Arrays.stream(SUPPORTED_WAREHOUSE_STORES)) + .collect(Collectors.toList()) + .contains(spec.getType())); } catch (NullPointerException | IllegalArgumentException e) { throw new IllegalArgumentException( Strings.lenientFormat( @@ -275,4 +293,4 @@ private void checkBigqueryImportSpecOption(ImportSpec spec) throws IllegalArgume Strings.lenientFormat("Invalid options: %s", e.getMessage())); } } -} +} \ No newline at end of file diff --git a/core/src/test/java/feast/core/validators/SpecValidatorTest.java b/core/src/test/java/feast/core/validators/SpecValidatorTest.java index bde01ea22f5..baad06941e8 100644 --- a/core/src/test/java/feast/core/validators/SpecValidatorTest.java +++ b/core/src/test/java/feast/core/validators/SpecValidatorTest.java @@ -17,6 +17,9 @@ package feast.core.validators; +import static org.mockito.Mockito.when; + +import com.google.common.base.Strings; import feast.core.dao.EntityInfoRepository; import feast.core.dao.FeatureGroupInfoRepository; import feast.core.dao.FeatureInfoRepository; @@ -33,16 +36,13 @@ import feast.specs.ImportSpecProto.Schema; import feast.specs.StorageSpecProto.StorageSpec; import feast.types.GranularityProto.Granularity; +import java.util.Optional; import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import org.mockito.Mockito; -import java.util.Optional; - -import static org.mockito.Mockito.when; - public class SpecValidatorTest { private FeatureInfoRepository featureInfoRepository; private FeatureGroupInfoRepository featureGroupInfoRepository; @@ -329,7 +329,9 @@ public void featureSpecWithoutServingStoreShouldInheritServingStoreIdFromGroup() FeatureGroupInfo fgi = new FeatureGroupInfo(); StorageInfo redis1 = new StorageInfo(); redis1.setId("REDIS1"); + redis1.setType("redis"); fgi.setServingStore(redis1); + when(storageInfoRepository.findById("REDIS1")).thenReturn(Optional.of(redis1)); when(featureGroupInfoRepository.existsById("group")).thenReturn(true); when(featureGroupInfoRepository.findById("group")).thenReturn(Optional.of(fgi)); SpecValidator validator = @@ -358,17 +360,25 @@ public void featureSpecWithoutServingStoreShouldInheritServingStoreIdFromGroup() @Test public void featureSpecWithoutExistingWarehouseStoreShouldThrowIllegalArgumentException() { + String servingStoreId = "REDIS1"; + String warehouseStoreId = "BIGQUERY"; when(entityInfoRepository.existsById("entity")).thenReturn(true); - when(storageInfoRepository.existsById("REDIS1")).thenReturn(true); - when(storageInfoRepository.existsById("REDIS2")).thenReturn(false); + when(storageInfoRepository.existsById(servingStoreId)).thenReturn(true); + when(storageInfoRepository.existsById(warehouseStoreId)).thenReturn(false); + + StorageInfo redis1 = new StorageInfo(); + redis1.setId(servingStoreId); + redis1.setType("redis"); + when(storageInfoRepository.findById( servingStoreId)).thenReturn(Optional.of(redis1)); + SpecValidator validator = new SpecValidator( storageInfoRepository, entityInfoRepository, featureGroupInfoRepository, featureInfoRepository); - DataStore servingStore = DataStore.newBuilder().setId("REDIS1").build(); - DataStore warehouseStore = DataStore.newBuilder().setId("REDIS2").build(); + DataStore servingStore = DataStore.newBuilder().setId(servingStoreId).build(); + DataStore warehouseStore = DataStore.newBuilder().setId(warehouseStoreId).build(); DataStores dataStores = DataStores.newBuilder().setServing(servingStore).setWarehouse(warehouseStore).build(); FeatureSpec input = @@ -382,7 +392,93 @@ public void featureSpecWithoutExistingWarehouseStoreShouldThrowIllegalArgumentEx .setDataStores(dataStores) .build(); exception.expect(IllegalArgumentException.class); - exception.expectMessage("Warehouse store with id REDIS2 does not exist"); + exception.expectMessage(String.format("Warehouse store with id %s does not exist", warehouseStoreId)); + validator.validateFeatureSpec(input); + } + + @Test + public void featureSpecWithUnsupportedWarehouseStoreShouldThrowIllegalArgumentException() { + String servingStoreId = "REDIS1"; + StorageSpec servingStoreSpec = StorageSpec.newBuilder().setId(servingStoreId).setType("redis").build(); + StorageInfo servingStoreInfo = new StorageInfo(servingStoreSpec); + + String warehouseStoreId = "REDIS2"; + StorageSpec warehouseStoreSpec = StorageSpec.newBuilder().setId(warehouseStoreId).setType("redis").build(); + StorageInfo warehouseStoreInfo = new StorageInfo(warehouseStoreSpec); + + when(entityInfoRepository.existsById("entity")).thenReturn(true); + when(storageInfoRepository.existsById(servingStoreId)).thenReturn(true); + when(storageInfoRepository.existsById(warehouseStoreId)).thenReturn(true); + when(storageInfoRepository.findById(servingStoreId)).thenReturn(Optional.of(servingStoreInfo)); + when(storageInfoRepository.findById(warehouseStoreId)).thenReturn(Optional.of(warehouseStoreInfo)); + SpecValidator validator = + new SpecValidator( + storageInfoRepository, + entityInfoRepository, + featureGroupInfoRepository, + featureInfoRepository); + DataStore servingStore = DataStore.newBuilder().setId(servingStoreId).build(); + DataStore warehouseStore = DataStore.newBuilder().setId(warehouseStoreId).build(); + DataStores dataStores = + DataStores.newBuilder().setServing(servingStore).setWarehouse(warehouseStore).build(); + FeatureSpec input = + FeatureSpec.newBuilder() + .setId("entity.none.name") + .setName("name") + .setOwner("owner") + .setDescription("dasdad") + .setEntity("entity") + .setGranularity(Granularity.Enum.forNumber(0)) + .setDataStores(dataStores) + .build(); + exception.expect(IllegalArgumentException.class); + exception.expectMessage(Strings.lenientFormat("Unsupported warehouse store type", "redis")); + validator.validateFeatureSpec(input); + } + + @Test + public void featureSpecWithUnsupportedServingStoreShouldThrowIllegalArgumentException() { + String servingStoreName = "CASSANDRA"; + StorageSpec redis1Spec = StorageSpec.newBuilder() + .setId(servingStoreName) + .setType("cassandra") + .build(); + StorageInfo redis1StorageInfo = new StorageInfo(redis1Spec); + + String warehouseStorageName = "BIGQUERY"; + StorageSpec bqSpec = StorageSpec.newBuilder() + .setId(warehouseStorageName) + .setType("bigquery") + .build(); + StorageInfo bqInfo = new StorageInfo(bqSpec); + + when(entityInfoRepository.existsById("entity")).thenReturn(true); + when(storageInfoRepository.existsById(servingStoreName)).thenReturn(true); + when(storageInfoRepository.existsById(warehouseStorageName)).thenReturn(true); + when(storageInfoRepository.findById(servingStoreName)).thenReturn(Optional.of(redis1StorageInfo)); + when(storageInfoRepository.findById(warehouseStorageName)).thenReturn(Optional.of(bqInfo)); + SpecValidator validator = + new SpecValidator( + storageInfoRepository, + entityInfoRepository, + featureGroupInfoRepository, + featureInfoRepository); + DataStore servingStore = DataStore.newBuilder().setId(servingStoreName).build(); + DataStore warehouseStore = DataStore.newBuilder().setId(warehouseStorageName).build(); + DataStores dataStores = + DataStores.newBuilder().setServing(servingStore).setWarehouse(warehouseStore).build(); + FeatureSpec input = + FeatureSpec.newBuilder() + .setId("entity.none.name") + .setName("name") + .setOwner("owner") + .setDescription("dasdad") + .setEntity("entity") + .setGranularity(Granularity.Enum.forNumber(0)) + .setDataStores(dataStores) + .build(); + exception.expect(IllegalArgumentException.class); + exception.expectMessage(Strings.lenientFormat("Unsupported serving store type", "cassandra")); validator.validateFeatureSpec(input); }