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 9a6f807581df545759073ed38bafcf7551b6ba31 Author: Philipp Zehnder <[email protected]> AuthorDate: Fri Dec 1 11:54:00 2023 +0100 feat(#2252): WIP --- streampipes-data-explorer/pom.xml | 6 +- .../dataexplorer/DataExplorerSchemaManagement.java | 114 +++++++++++++-------- .../DataExplorerSchemaManagementTest.java | 78 ++++++++++++++ .../sinks/internal/jvm/datalake/DataLakeSink.java | 9 +- 4 files changed, 161 insertions(+), 46 deletions(-) diff --git a/streampipes-data-explorer/pom.xml b/streampipes-data-explorer/pom.xml index c68a436d7..6cbfbbb62 100644 --- a/streampipes-data-explorer/pom.xml +++ b/streampipes-data-explorer/pom.xml @@ -54,7 +54,11 @@ <artifactId>junit</artifactId> <scope>test</scope> </dependency> - + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-core</artifactId> + <scope>test</scope> + </dependency> <dependency> <groupId>org.influxdb</groupId> <artifactId>influxdb-java</artifactId> 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 8c4dc4a34..28ebd35aa 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 @@ -39,6 +39,14 @@ import java.util.Optional; public class DataExplorerSchemaManagement implements IDataExplorerSchemaManagement { + IDataLakeStorage dataLakeStorage; + + public DataExplorerSchemaManagement() {} + + public DataExplorerSchemaManagement(IDataLakeStorage dataLakeStorage) { + this.dataLakeStorage = dataLakeStorage; + } + @Override public List<DataLakeMeasure> getAllMeasurements() { return DataExplorerUtils.getInfos(); @@ -51,24 +59,30 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme @Override public DataLakeMeasure createMeasurement(DataLakeMeasure measure) { - List<DataLakeMeasure> dataLakeMeasureList = getDataLakeStorage().getAllDataLakeMeasures(); + List<DataLakeMeasure> dataLakeMeasureList = dataLakeStorage.getAllDataLakeMeasures(); Optional<DataLakeMeasure> optional = - dataLakeMeasureList.stream().filter(entry -> entry.getMeasureName().equals(measure.getMeasureName())) - .findFirst(); + 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())) { + if (!compareEventProperties( + oldEntry.getEventSchema() + .getEventProperties(), + measure.getEventSchema() + .getEventProperties() + )) { oldEntry.setEventSchema(measure.getEventSchema()); oldEntry.setTimestampField(measure.getTimestampField()); oldEntry.setPipelineName(measure.getPipelineName()); - getDataLakeStorage().updateDataLakeMeasure(oldEntry); + dataLakeStorage.updateDataLakeMeasure(oldEntry); return oldEntry; } } else { measure.setSchemaVersion(DataLakeMeasure.CURRENT_SCHEMA_VERSION); - getDataLakeStorage().storeDataLakeMeasure(measure); + dataLakeStorage.storeDataLakeMeasure(measure); return measure; } @@ -88,12 +102,23 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme public boolean deleteMeasurementByName(String measureName) { boolean isSuccess = false; CouchDbClient couchDbClient = Utils.getCouchDbDataLakeClient(); - List<JsonObject> docs = couchDbClient.view("_all_docs").includeDocs(true).query(JsonObject.class); + List<JsonObject> docs = couchDbClient.view("_all_docs") + .includeDocs(true) + .query(JsonObject.class); for (JsonObject document : docs) { - if (document.get("measureName").toString().replace("\"", "").equals(measureName)) { - couchDbClient.remove(document.get("_id").toString().replace("\"", ""), - document.get("_rev").toString().replace("\"", "")); + if (document.get("measureName") + .toString() + .replace("\"", "") + .equals(measureName)) { + couchDbClient.remove( + document.get("_id") + .toString() + .replace("\"", ""), + document.get("_rev") + .toString() + .replace("\"", "") + ); isSuccess = true; break; } @@ -118,8 +143,9 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme } } - private IDataLakeStorage getDataLakeStorage() { - return StorageDispatcher.INSTANCE.getNoSqlStore().getDataLakeStorage(); + protected IDataLakeStorage getDataLakeStorage() { + return StorageDispatcher.INSTANCE.getNoSqlStore() + .getDataLakeStorage(); } private boolean compareEventProperties(List<EventProperty> prop1, List<EventProperty> prop2) { @@ -127,33 +153,39 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme 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; - - }); + 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; + + }); } } 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 new file mode 100644 index 000000000..2fe400e51 --- /dev/null +++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.streampipes.dataexplorer; + +import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.storage.api.IDataLakeStorage; + +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.List; + +import static org.junit.Assert.assertEquals; +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 { + + @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); + + var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorage); + + // Act + var updatedMeasure = schemaManagement.createMeasurement(measure); + + // Assert + assertEquals(measure.getMeasureName(), updatedMeasure.getMeasureName()); + verify(dataLakeStorage, Mockito.times(1)) + .storeDataLakeMeasure(any()); + + } + + @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); + + var schemaManagement = new DataExplorerSchemaManagement(dataLakeStorage); + + // Act + var updatedMeasure = schemaManagement.createMeasurement(measure); + + // Assert + assertEquals(measure.getMeasureName(), updatedMeasure.getMeasureName()); + verify(dataLakeStorage, Mockito.times(1)) + .storeDataLakeMeasure(any()); + + } +} \ No newline at end of file diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java index 06f63e0d2..396b791d0 100644 --- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java +++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java @@ -74,14 +74,15 @@ public class DataLakeSink extends StreamPipesDataSink { @Override public void onInvocation(SinkParams parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException { - var timestampField = parameters.extractor() - .mappingPropertyValue(TIMESTAMP_MAPPING_KEY); - var measureName = parameters.extractor() - .singleValueParameter(DATABASE_MEASUREMENT_KEY, String.class); + var extractor = parameters.extractor(); + var timestampField = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY); + var measureName = extractor.singleValueParameter(DATABASE_MEASUREMENT_KEY, String.class); var eventSchema = parameters.getInputSchemaInfos() .get(0) .getEventSchema(); + String schemaUpdateOption = extractor.selectedSingleValue(SCHEMA_UPDATE_KEY, String.class); + var measure = new DataLakeMeasure(measureName, timestampField, eventSchema); this.timeSeriesStore = new TimeSeriesStore(
