This is an automated email from the ASF dual-hosted git repository. zehnder pushed a commit to branch 2252-enhance-data-lake-sink-schema-management-options in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 14abf009d0e83efececb96de0397a5301788af9b Author: Philipp Zehnder <[email protected]> AuthorDate: Sun Dec 3 12:11:19 2023 +0100 refactor(#2246): The backend REST API does now handle schema updates according to the new strategies --- streampipes-data-explorer/pom.xml | 6 + .../dataexplorer/DataExplorerSchemaManagement.java | 182 +++++++++++++-------- .../api/IDataExplorerSchemaManagement.java | 2 +- .../DataExplorerSchemaManagementTest.java | 182 +++++++++++++++++---- .../model/datalake/DataLakeMeasure.java | 10 ++ .../DataLakeMeasureSchemaUpdateStrategy.java | 34 ++-- .../streampipes/ps/DataLakeMeasureResourceV4.java | 2 +- 7 files changed, 298 insertions(+), 120 deletions(-) diff --git a/streampipes-data-explorer/pom.xml b/streampipes-data-explorer/pom.xml index 6cbfbbb62..2b18b0790 100644 --- a/streampipes-data-explorer/pom.xml +++ b/streampipes-data-explorer/pom.xml @@ -48,6 +48,12 @@ <artifactId>streampipes-storage-management</artifactId> <version>0.95.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-test-utils</artifactId> + <version>0.95.0-SNAPSHOT</version> + <scope>test</scope> + </dependency> <dependency> <groupId>junit</groupId> diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java index 28ebd35aa..b2c980623 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java @@ -21,24 +21,29 @@ package org.apache.streampipes.dataexplorer; import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement; import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils; import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.model.datalake.DataLakeMeasureSchemaUpdateStrategy; import org.apache.streampipes.model.schema.EventProperty; -import org.apache.streampipes.model.schema.EventPropertyList; -import org.apache.streampipes.model.schema.EventPropertyNested; -import org.apache.streampipes.model.schema.EventPropertyPrimitive; import org.apache.streampipes.storage.api.IDataLakeStorage; import org.apache.streampipes.storage.couchdb.utils.Utils; import org.apache.streampipes.storage.management.StorageDispatcher; import com.google.gson.JsonObject; import org.lightcouch.CouchDbClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; -import java.util.Collections; +import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; public class DataExplorerSchemaManagement implements IDataExplorerSchemaManagement { + private static final Logger LOG = LoggerFactory.getLogger(DataExplorerSchemaManagement.class); + IDataLakeStorage dataLakeStorage; public DataExplorerSchemaManagement() {} @@ -57,38 +62,60 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme return getDataLakeStorage().findOne(elementId); } + /** + * For new measurements an entry is generated in the database. For existing measurements the schema is updated + * according to the provided update strategy. + */ @Override - public DataLakeMeasure createMeasurement(DataLakeMeasure measure) { - List<DataLakeMeasure> dataLakeMeasureList = dataLakeStorage.getAllDataLakeMeasures(); - Optional<DataLakeMeasure> optional = - dataLakeMeasureList.stream() - .filter(entry -> entry.getMeasureName() - .equals(measure.getMeasureName())) - .findFirst(); - - if (optional.isPresent()) { - DataLakeMeasure oldEntry = optional.get(); - if (!compareEventProperties( - oldEntry.getEventSchema() - .getEventProperties(), - measure.getEventSchema() - .getEventProperties() - )) { - oldEntry.setEventSchema(measure.getEventSchema()); - oldEntry.setTimestampField(measure.getTimestampField()); - oldEntry.setPipelineName(measure.getPipelineName()); - dataLakeStorage.updateDataLakeMeasure(oldEntry); - return oldEntry; - } + public DataLakeMeasure createOrUpdateMeasurement(DataLakeMeasure measure) { + + setDefaultUpdateStrategyIfNoneProvided(measure); + + var existingMeasure = getExistingMeasureByName(measure.getMeasureName()); + + if (existingMeasure.isEmpty()) { + setSchemaVersionAndStoreMeasurement(measure); } else { - measure.setSchemaVersion(DataLakeMeasure.CURRENT_SCHEMA_VERSION); - dataLakeStorage.storeDataLakeMeasure(measure); - return measure; + handleExistingMeasurement(measure, existingMeasure.get()); } return measure; } + /** + * Destiguishes between the update straregy for existing measurments + */ + private void handleExistingMeasurement( + DataLakeMeasure measure, + DataLakeMeasure existingMeasure + ) { + if (DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA.equals(measure.getSchemaUpdateStrategy())) { + // For the update schema strategy the old schema is overwritten with the new one + updateMeasurement(measure); + } else { + // For the extent existing schema strategy the old schema is merged with the new one + unifyEventSchemaAndUpdateMeasure(measure, existingMeasure); + } + } + + + /** + * Returns the existing measure that has the provided measure name + */ + private Optional<DataLakeMeasure> getExistingMeasureByName(String measureName) { + return dataLakeStorage.getAllDataLakeMeasures() + .stream() + .filter(m -> m.getMeasureName() + .equals(measureName)) + .findFirst(); + } + + private static void setDefaultUpdateStrategyIfNoneProvided(DataLakeMeasure measure) { + if (measure.getSchemaUpdateStrategy() == null) { + measure.setSchemaUpdateStrategy(DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA); + } + } + @Override public void deleteMeasurement(String elementId) { if (getDataLakeStorage().findOne(elementId) != null) { @@ -127,19 +154,19 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme try { couchDbClient.close(); } catch (IOException e) { - e.printStackTrace(); + LOG.error("Could not close CouchDB client", e); } return isSuccess; } @Override public void updateMeasurement(DataLakeMeasure measure) { - var existingMeasure = getDataLakeStorage().findOne(measure.getElementId()); + var existingMeasure = dataLakeStorage.findOne(measure.getElementId()); if (existingMeasure != null) { measure.setRev(existingMeasure.getRev()); - getDataLakeStorage().updateDataLakeMeasure(measure); + dataLakeStorage.updateDataLakeMeasure(measure); } else { - getDataLakeStorage().storeDataLakeMeasure(measure); + dataLakeStorage.storeDataLakeMeasure(measure); } } @@ -148,44 +175,57 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme .getDataLakeStorage(); } - private boolean compareEventProperties(List<EventProperty> prop1, List<EventProperty> prop2) { - if (prop1.size() != prop2.size()) { - return false; - } - return prop1.stream() - .allMatch(prop -> { - - for (EventProperty property : prop2) { - if (prop.getRuntimeName() - .equals(property.getRuntimeName())) { - - //primitive - if (prop instanceof EventPropertyPrimitive && property instanceof EventPropertyPrimitive) { - if (((EventPropertyPrimitive) prop) - .getRuntimeType() - .equals(((EventPropertyPrimitive) property).getRuntimeType())) { - return true; - } - - //list - } else if (prop instanceof EventPropertyList && property instanceof EventPropertyList) { - return compareEventProperties( - Collections.singletonList(((EventPropertyList) prop).getEventProperty()), - Collections.singletonList(((EventPropertyList) property).getEventProperty()) - ); - - //nested - } else if (prop instanceof EventPropertyNested && property instanceof EventPropertyNested) { - return compareEventProperties( - ((EventPropertyNested) prop).getEventProperties(), - ((EventPropertyNested) property).getEventProperties() - ); - } - } - } - return false; - - }); + private void setSchemaVersionAndStoreMeasurement(DataLakeMeasure measure) { + measure.setSchemaVersion(DataLakeMeasure.CURRENT_SCHEMA_VERSION); + dataLakeStorage.storeDataLakeMeasure(measure); + } + + /** + * First the event schemas of the measurements are merged and then the measure is updated in the database + */ + private void unifyEventSchemaAndUpdateMeasure( + DataLakeMeasure measure, + DataLakeMeasure existingMeasure + ) { + var properties = getUnifiedEventProperties( + existingMeasure, + measure + ); + + measure + .getEventSchema() + .setEventProperties(properties); + + updateMeasurement(measure); + } + + /** + * Returns the union of the unique event properties of the two measures. + * They are unique by runtime name. + */ + private List<EventProperty> getUnifiedEventProperties( + DataLakeMeasure measure1, + DataLakeMeasure measure2 + ) { + return new ArrayList<>( + Stream + .concat( + measure1 + .getEventSchema() + .getEventProperties() + .stream(), + measure2 + .getEventSchema() + .getEventProperties() + .stream() + ) + .collect(Collectors.toMap( + EventProperty::getRuntimeName, + Function.identity(), + (eventProperty, eventProperty2) -> eventProperty + )) + .values()); + } } diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java index d8dc29648..a210b515b 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java @@ -28,7 +28,7 @@ public interface IDataExplorerSchemaManagement { DataLakeMeasure getById(String elementId); - DataLakeMeasure createMeasurement(DataLakeMeasure measure); + DataLakeMeasure createOrUpdateMeasurement(DataLakeMeasure measure); void deleteMeasurement(String elementId); diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java index 2fe400e51..baa6053b9 100644 --- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java +++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java @@ -19,60 +19,186 @@ package org.apache.streampipes.dataexplorer; import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.model.datalake.DataLakeMeasureSchemaUpdateStrategy; +import org.apache.streampipes.model.schema.EventProperty; import org.apache.streampipes.storage.api.IDataLakeStorage; +import org.apache.streampipes.test.generator.EventPropertyPrimitiveTestBuilder; +import org.apache.streampipes.test.generator.EventSchemaTestBuilder; +import org.apache.streampipes.vocabulary.XSD; +import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; +import java.net.URI; import java.util.List; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class DataExplorerSchemaManagementTest { + public static final String NEW_PROPERTY = "newProperty"; + public static final String OLD_PROPERTY = "oldProperty"; + + private IDataLakeStorage dataLakeStorageMock; + + @Before + public void setUp() { + dataLakeStorageMock = mock(IDataLakeStorage.class); + } @Test - public void createMeasurement() { - // Arrange - var dataLakeStorage = mock(IDataLakeStorage.class); - DataLakeMeasure measure = new DataLakeMeasure(); - measure.setMeasureName("testMeasure"); - when(dataLakeStorage.getAllDataLakeMeasures()).thenReturn(List.of()); - when(dataLakeStorage.storeDataLakeMeasure(any())).thenReturn(true); + public void createMeasurementThatNotExisted() { + when(dataLakeStorageMock.getAllDataLakeMeasures()).thenReturn(List.of()); + when(dataLakeStorageMock.storeDataLakeMeasure(any())).thenReturn(true); + var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorageMock); + + var oldMeasure = getSampleMeasure( + DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA, + List.of() + ); + var resultingMeasure = schemaManagement.createOrUpdateMeasurement(oldMeasure); + + assertEquals(oldMeasure.getMeasureName(), resultingMeasure.getMeasureName()); + verify(dataLakeStorageMock, Mockito.times(1)) + .storeDataLakeMeasure(any()); + } - var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorage); - // Act - var updatedMeasure = schemaManagement.createMeasurement(measure); + @Test + public void createMeasurementWithUpdateStrategy() { - // Assert - assertEquals(measure.getMeasureName(), updatedMeasure.getMeasureName()); - verify(dataLakeStorage, Mockito.times(1)) - .storeDataLakeMeasure(any()); + var oldMeasure = getSampleMeasure( + DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA, + List.of( + getEventProperty(OLD_PROPERTY, XSD.STRING) + ) + ); + + when(dataLakeStorageMock.getAllDataLakeMeasures()).thenReturn(List.of(oldMeasure)); + when(dataLakeStorageMock.findOne(any())).thenReturn(oldMeasure); + var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorageMock); + + var newMeasure = getNewMeasure(DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA); + + var resultMeasure = schemaManagement.createOrUpdateMeasurement(newMeasure); + + assertEquals(newMeasure.getMeasureName(), resultMeasure.getMeasureName()); + verify(dataLakeStorageMock, Mockito.times(1)) + .updateDataLakeMeasure(any()); + assertFalse(containsPropertyWithName(resultMeasure, OLD_PROPERTY)); + assertTrue(containsPropertyWithName(resultMeasure, NEW_PROPERTY)); } + @Test - public void createMeasurementUpdate() { - // Arrange - var dataLakeStorage = mock(IDataLakeStorage.class); - DataLakeMeasure measure = new DataLakeMeasure(); - measure.setMeasureName("testMeasure"); - when(dataLakeStorage.getAllDataLakeMeasures()).thenReturn(List.of(measure)); - when(dataLakeStorage.storeDataLakeMeasure(any())).thenReturn(true); + public void createMeasurementWithExtendSchemaStrategy() { + + var oldMeasure = getSampleMeasure( + DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA, + List.of( + getEventProperty(OLD_PROPERTY, XSD.STRING) + ) + ); + when(dataLakeStorageMock.getAllDataLakeMeasures()).thenReturn(List.of(oldMeasure)); + when(dataLakeStorageMock.findOne(any())).thenReturn(oldMeasure); + var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorageMock); + var newMeasure = getNewMeasure(DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA); + + var resultMeasure = schemaManagement.createOrUpdateMeasurement(newMeasure); + + assertEquals(newMeasure.getMeasureName(), resultMeasure.getMeasureName()); + verify(dataLakeStorageMock, Mockito.times(1)).updateDataLakeMeasure(any()); + assertTrue(containsPropertyWithName(resultMeasure, OLD_PROPERTY)); + assertTrue(containsPropertyWithName(resultMeasure, NEW_PROPERTY)); + } - var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorage); - // Act - var updatedMeasure = schemaManagement.createMeasurement(measure); + @Test + public void createMeasurementWithExtendSchemaStrategyAndDifferentPropertyTypes() { + var oldMeasure = getSampleMeasure( + DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA, + List.of( + getEventProperty(OLD_PROPERTY, XSD.STRING), + getEventProperty(NEW_PROPERTY, XSD.INTEGER) + ) + ); + + when(dataLakeStorageMock.getAllDataLakeMeasures()).thenReturn(List.of(oldMeasure)); + when(dataLakeStorageMock.findOne(any())).thenReturn(oldMeasure); + + var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorageMock); + + var newMeasure = getNewMeasure(DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA); + + var resultMeasure = schemaManagement.createOrUpdateMeasurement(newMeasure); + assertEquals(newMeasure.getMeasureName(), resultMeasure.getMeasureName()); + verify(dataLakeStorageMock, Mockito.times(1)).updateDataLakeMeasure(any()); + assertEquals( + 2, + resultMeasure.getEventSchema() + .getEventProperties() + .size() + ); + assertTrue(containsPropertyWithName(resultMeasure, OLD_PROPERTY)); + assertTrue(containsPropertyWithName(resultMeasure, NEW_PROPERTY)); + } - // Assert - assertEquals(measure.getMeasureName(), updatedMeasure.getMeasureName()); - verify(dataLakeStorage, Mockito.times(1)) - .storeDataLakeMeasure(any()); + private EventProperty getEventProperty( + String runtimeName, + URI runtimeType + ) { + return EventPropertyPrimitiveTestBuilder + .create() + .withRuntimeName(runtimeName) + .withRuntimeType(runtimeType) + .build(); + } + + private DataLakeMeasure getNewMeasure(DataLakeMeasureSchemaUpdateStrategy updateStrategy) { + return getSampleMeasure( + updateStrategy, + List.of(getEventProperty(NEW_PROPERTY, XSD.STRING)) + ); + } + + private DataLakeMeasure getSampleMeasure( + DataLakeMeasureSchemaUpdateStrategy updateStrategy, + List<EventProperty> eventProperties + ) { + var measure = new DataLakeMeasure(); + measure.setMeasureName("testMeasure"); + measure.setSchemaUpdateStrategy(updateStrategy); + + measure.setEventSchema( + EventSchemaTestBuilder + .create() + .withEventProperties( + eventProperties + ) + .build() + ); + + return measure; + } + private boolean containsPropertyWithName( + DataLakeMeasure measure, + String runtimeName + ) { + return measure + .getEventSchema() + .getEventProperties() + .stream() + .anyMatch( + eventProperty -> eventProperty.getRuntimeName() + .equals(runtimeName) + ); } } \ No newline at end of file diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasure.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasure.java index 44076e605..2d9441b28 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasure.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasure.java @@ -51,6 +51,8 @@ public class DataLakeMeasure { private String schemaVersion; + private DataLakeMeasureSchemaUpdateStrategy schemaUpdateStrategy = DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA; + public DataLakeMeasure() { super(); } @@ -129,6 +131,14 @@ public class DataLakeMeasure { this.timestampField = timestampField; } + public DataLakeMeasureSchemaUpdateStrategy getSchemaUpdateStrategy() { + return schemaUpdateStrategy; + } + + public void setSchemaUpdateStrategy(DataLakeMeasureSchemaUpdateStrategy schemaUpdateStrategy) { + this.schemaUpdateStrategy = schemaUpdateStrategy; + } + /** * This can be used to get the name of the timestamp property without the stream prefix * diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasureSchemaUpdateStrategy.java similarity index 61% copy from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java copy to streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasureSchemaUpdateStrategy.java index d8dc29648..acd5bc232 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeMeasureSchemaUpdateStrategy.java @@ -16,23 +16,19 @@ * */ -package org.apache.streampipes.dataexplorer.api; +package org.apache.streampipes.model.datalake; -import org.apache.streampipes.model.datalake.DataLakeMeasure; - -import java.util.List; - -public interface IDataExplorerSchemaManagement { - - List<DataLakeMeasure> getAllMeasurements(); - - DataLakeMeasure getById(String elementId); - - DataLakeMeasure createMeasurement(DataLakeMeasure measure); - - void deleteMeasurement(String elementId); - - boolean deleteMeasurementByName(String measureName); - - void updateMeasurement(DataLakeMeasure measure); -} +/** + * This enum contains the different options that are available for the update strategy of the data lake measure + */ +public enum DataLakeMeasureSchemaUpdateStrategy { + /** + * This strategy will update the schema of the data lake measure when it changes + */ + UPDATE_SCHEMA, + /** + * This strategy will extend the schema of the data lake measure when it changes and keep the old fields + */ + EXTEND_EXISTING_SCHEMA + +} \ No newline at end of file diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java index c40f2e56b..0a5b97e16 100644 --- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java +++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java @@ -54,7 +54,7 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource { @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) public Response addDataLake(DataLakeMeasure dataLakeMeasure) { - DataLakeMeasure result = this.dataLakeMeasureManagement.createMeasurement(dataLakeMeasure); + DataLakeMeasure result = this.dataLakeMeasureManagement.createOrUpdateMeasurement(dataLakeMeasure); return ok(result); }
