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 9afaf4a589db2ecb85e01cac6978c17424b08647 Author: Philipp Zehnder <[email protected]> AuthorDate: Mon Dec 4 17:43:42 2023 +0100 refactor(#2246): Fix update bug --- .../streampipes/dataexplorer/DataExplorerSchemaManagement.java | 1 + .../streampipes/sinks/internal/jvm/datalake/DataLakeSink.java | 10 +++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) 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 858cd1f8c..53c1b68b2 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 @@ -86,6 +86,7 @@ public class DataExplorerSchemaManagement implements IDataExplorerSchemaManageme DataLakeMeasure measure, DataLakeMeasure existingMeasure ) { + measure.setElementId(existingMeasure.getElementId()); if (DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA.equals(measure.getSchemaUpdateStrategy())) { // For the update schema strategy the old schema is overwritten with the new one updateMeasurement(measure); 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 b95b183dc..56aa41676 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 @@ -24,6 +24,7 @@ import org.apache.streampipes.dataexplorer.commons.TimeSeriesStore; import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext; import org.apache.streampipes.model.DataSinkType; import org.apache.streampipes.model.datalake.DataLakeMeasure; +import org.apache.streampipes.model.datalake.DataLakeMeasureSchemaUpdateStrategy; import org.apache.streampipes.model.graph.DataSinkDescription; import org.apache.streampipes.model.runtime.Event; import org.apache.streampipes.model.schema.PropertyScope; @@ -80,10 +81,17 @@ public class DataLakeSink extends StreamPipesDataSink { .get(0) .getEventSchema(); - String schemaUpdateOption = extractor.selectedSingleValue(SCHEMA_UPDATE_KEY, String.class); var measure = new DataLakeMeasure(measureName, timestampField, eventSchema); + var schemaUpdateOptionString = extractor.selectedSingleValue(SCHEMA_UPDATE_KEY, String.class); + + if (schemaUpdateOptionString.equals(DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA.toString())) { + measure.setSchemaUpdateStrategy(DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA); + } else { + measure.setSchemaUpdateStrategy(DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA); + } + this.timeSeriesStore = new TimeSeriesStore( Environments.getEnvironment(), runtimeContext.getStreamPipesClient(),
