This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch
4367-warn-on-incompatible-dataset-schema-changes-affecting-measurements-in-pipelines
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/4367-warn-on-incompatible-dataset-schema-changes-affecting-measurements-in-pipelines
by this push:
new 6e30262d3e Detect data type change of measurements
6e30262d3e is described below
commit 6e30262d3e2ad7580105867b2e316e38bcedd5ab
Author: Sven Oehler <[email protected]>
AuthorDate: Tue May 5 10:33:30 2026 +0200
Detect data type change of measurements
---
.../update/PipelineMeasurementChangeDetector.java | 135 +++++++++++++++++++++
.../pipeline/update/PipelineUpdateCoordinator.java | 18 ++-
2 files changed, 151 insertions(+), 2 deletions(-)
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineMeasurementChangeDetector.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineMeasurementChangeDetector.java
new file mode 100644
index 0000000000..700c104f6f
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineMeasurementChangeDetector.java
@@ -0,0 +1,135 @@
+/*
+ * 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.manager.pipeline.update;
+
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.vocabulary.SO;
+import org.apache.streampipes.vocabulary.XSD;
+
+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 PipelineMeasurementChangeDetector {
+
+ private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+
+ public boolean hasCriticalMeasurementFieldChange(Pipeline pipeline,
+ String affectedElementId,
+ EventSchema
updatedEventSchema) {
+ if (!hasDatabaseSink(pipeline)) {
+ return false;
+ }
+
+ return getEventSchema(pipeline, affectedElementId)
+ .map(existingEventSchema ->
+ hasCriticalMeasurementFieldChange(existingEventSchema,
updatedEventSchema))
+ .orElse(false);
+ }
+
+ private boolean hasDatabaseSink(Pipeline pipeline) {
+ return pipeline.getActions().stream()
+ .anyMatch(this::isDatabaseSink);
+ }
+
+ private boolean isDatabaseSink(DataSinkInvocation dataSink) {
+ return DATA_LAKE_SINK_APP_ID.equals(dataSink.getAppId())
+ ||
dataSink.getCategory().stream().anyMatch(DataSinkType.DATABASE.name()::equals);
+ }
+
+ private Optional<EventSchema> getEventSchema(Pipeline pipeline,
+ String affectedElementId) {
+ return pipeline
+ .getStreams()
+ .stream()
+ .filter(stream -> stream.getElementId().equals(affectedElementId))
+ .findFirst()
+ .map(SpDataStream::getEventSchema);
+ }
+
+ private boolean hasCriticalMeasurementFieldChange(EventSchema
existingEventSchema,
+ EventSchema
updatedEventSchema) {
+ var existingMeasurementFields = existingEventSchema
+ .getEventProperties()
+ .stream()
+ .filter(this::isMeasurementField)
+ .collect(Collectors.toMap(
+ EventProperty::getRuntimeName,
+ Function.identity(),
+ (existingProperty, duplicateProperty) -> existingProperty
+ ));
+
+ return updatedEventSchema
+ .getEventProperties()
+ .stream()
+ .filter(this::isMeasurementField)
+ .anyMatch(updatedProperty -> {
+ var existingProperty =
existingMeasurementFields.get(updatedProperty.getRuntimeName());
+ return existingProperty != null &&
hasCriticalFieldTypeChange(existingProperty, updatedProperty);
+ });
+ }
+
+ private boolean isMeasurementField(EventProperty eventProperty) {
+ return
!PropertyScope.DIMENSION_PROPERTY.name().equals(eventProperty.getPropertyScope());
+ }
+
+ private boolean hasCriticalFieldTypeChange(EventProperty existingProperty,
+ EventProperty updatedProperty) {
+ return toStorageType(existingProperty) != toStorageType(updatedProperty);
+ }
+
+ private StorageType toStorageType(EventProperty eventProperty) {
+ if (eventProperty instanceof EventPropertyPrimitive primitiveProperty) {
+ return toPrimitiveStorageType(primitiveProperty.getRuntimeType());
+ } else {
+ return StorageType.STRING;
+ }
+ }
+
+ private StorageType toPrimitiveStorageType(String runtimeType) {
+ if (XSD.INTEGER.toString().equals(runtimeType)
+ || XSD.LONG.toString().equals(runtimeType)) {
+ return StorageType.INTEGER;
+ } else if (XSD.FLOAT.toString().equals(runtimeType)
+ || XSD.DOUBLE.toString().equals(runtimeType)
+ || SO.NUMBER.equals(runtimeType)) {
+ return StorageType.FLOAT;
+ } else if (XSD.BOOLEAN.toString().equals(runtimeType)) {
+ return StorageType.BOOLEAN;
+ } else {
+ return StorageType.STRING;
+ }
+ }
+
+ private enum StorageType {
+ INTEGER,
+ FLOAT,
+ BOOLEAN,
+ STRING
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
index 6de0a17945..1f81c988ff 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
@@ -49,9 +49,11 @@ public class PipelineUpdateCoordinator {
private static final PipelinesStats PIPELINES_STATS = new PipelinesStats();
private final ExtensionServiceRequestManager requestManager;
+ private final PipelineMeasurementChangeDetector
measurementFieldChangeDetector;
public PipelineUpdateCoordinator(ExtensionServiceRequestManager
requestManager) {
this.requestManager = requestManager;
+ this.measurementFieldChangeDetector = new
PipelineMeasurementChangeDetector();
}
public void updatePipelines(SpDataStream dataStream) {
@@ -101,6 +103,11 @@ public class PipelineUpdateCoordinator {
}
var storedPipeline =
PipelineManager.getPipeline(pipeline.getPipelineId());
+ var hasChangedMeasurementField =
measurementFieldChangeDetector.hasCriticalMeasurementFieldChange(
+ storedPipeline,
+ affectedElementId,
+ updatedEventSchema
+ );
var updatedPipeline = updatePipeline(storedPipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
try {
@@ -108,7 +115,7 @@ public class PipelineUpdateCoordinator {
var modificationMessage = verificationHandler.verifyPipeline();
var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
var modifiedPipeline =
verificationHandler.makeModifiedPipeline().pipeline();
- var canAutoMigrate = canAutoMigrate(modificationMessage);
+ var canAutoMigrate = canAutoMigrate(modificationMessage) &&
!hasChangedMeasurementField;
if (!canAutoMigrate) {
modifiedPipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
@@ -139,10 +146,17 @@ public class PipelineUpdateCoordinator {
var updateInfos = new ArrayList<PipelineUpdateInfo>();
affectedPipelines.forEach(pipeline -> {
+ var hasChangedMeasurementField =
measurementFieldChangeDetector.hasCriticalMeasurementFieldChange(
+ pipeline,
+ affectedElementId,
+ updatedEventSchema
+ );
var updatedPipeline = updatePipeline(pipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
try {
var modificationMessage = new
PipelineVerificationHandlerV2(updatedPipeline, requestManager).verifyPipeline();
- updateInfos.add(makeUpdateInfo(modificationMessage, updatedPipeline));
+ var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
+ updateInfo.setCanAutoMigrate(updateInfo.isCanAutoMigrate() &&
!hasChangedMeasurementField);
+ updateInfos.add(updateInfo);
} catch (Exception e) {
throw new RuntimeException(e);
}
