(streampipes) branch 4781-preserve-missing-values-during-adapter-preprocessing updated: fix(#4781): preserve missing values in preprocessing

[ze...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tenthe pushed a commit to branch 4781-preserve-missing-values-during-adapter-preprocessing
in repository https://gitbox.apache.org/repos/asf/streampipes.git


The following commit(s) were added to refs/heads/4781-preserve-missing-values-during-adapter-preprocessing by this push:
     new 7700b3d5ca fix(#4781): preserve missing values in preprocessing
7700b3d5ca is described below

commit 7700b3d5caf1c599a9e9a283936569e80ef412bd
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Tue Jul 28 10:24:13 2026 +0200

    fix(#4781): preserve missing values in preprocessing
---
 .../SupportsNestedTransformationRule.java          |   4 +
 .../value/DatatypeTransformationRule.java          |   4 +
 .../transform/value/UnitTransformationRule.java    |  18 +++-
 .../value/DatatypeTransformationRuleTest.java      |  58 ++++++++++++
 .../transform/value/UnitTransformRuleTest.java     | 102 +++++++++++++++++++++
 5 files changed, 184 insertions(+), 2 deletions(-)

diff --git a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/SupportsNestedTransformationRule.java b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/SupportsNestedTransformationRule.java
index d4ee006249..d771f0dbfa 100644
--- a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/SupportsNestedTransformationRule.java
+++ b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/SupportsNestedTransformationRule.java
@@ -38,6 +38,10 @@ public abstract class SupportsNestedTransformationRule implements Transformation
       String key = eventKey.get(0);
       List<String> newKeysTmpList = eventKey.subList(1, eventKey.size());
 
+      if (!(event.get(key) instanceof Map<?, ?>)) {
+        return event;
+      }
+
       Map<String, Object> newSubEvent =
           applyNested((Map<String, Object>) event.get(eventKey.get(0)), newKeysTmpList);
 
diff --git a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRule.java b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRule.java
index 534dcc73f3..d3fdc8a539 100644
--- a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRule.java
+++ b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRule.java
@@ -44,6 +44,10 @@ public class DatatypeTransformationRule implements TransformationRule {
 
   @Override
   public Map<String, Object> apply(Map<String, Object> event) {
+    if (!event.containsKey(eventKey)) {
+      return event;
+    }
+
     Object value = event.get(eventKey);
     Object transformedValue = transformDatatype(value);
     event.put(eventKey, transformedValue);
diff --git a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformationRule.java b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformationRule.java
index 4eecf58cfa..592652fe5c 100644
--- a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformationRule.java
+++ b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformationRule.java
@@ -60,13 +60,27 @@ public class UnitTransformationRule extends SupportsNestedTransformationRule {
 
   @Override
   protected void applyTransformation(Map<String, Object> event, List<String> eventKey) {
+    var key = eventKey.get(0);
+    if (!event.containsKey(key) || event.get(key) == null) {
+      return;
+    }
+
+    var originalValue = event.get(key);
     try {
-      double value = Double.parseDouble(String.valueOf(event.get(eventKey.get(0))));
+      double value = Double.parseDouble(String.valueOf(originalValue));
 
       Quantity obs = new Quantity(value, unitTypeFrom);
       double newValue = obs.convertTo(unitTypeTo).getValue();
 
-      event.put(eventKey.get(0), newValue);
+      event.put(key, newValue);
+    } catch (NumberFormatException e) {
+      logger.debug(
+          "Could not convert value '{}' from unit '{}' to '{}' for event property '{}'",
+          originalValue,
+          unitTypeFrom,
+          unitTypeTo,
+          key
+      );
     } catch (ClassCastException | IllegalAccessException e) {
       logger.error(e.toString());
     }
diff --git a/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRuleTest.java b/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRuleTest.java
new file mode 100644
index 0000000000..055bf318ae
--- /dev/null
+++ b/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRuleTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.connect.shared.preprocessing.transform.value;
+
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DatatypeTransformationRuleTest {
+
+  private static final String PROPERTY = "property";
+
+  private final DatatypeTransformationRule rule =
+      new DatatypeTransformationRule("test-adapter", PROPERTY, XSD.DOUBLE.toString());
+
+  @Test
+  public void preserveMissingValue() {
+    Map<String, Object> event = new HashMap<>();
+
+    var result = rule.apply(event);
+
+    assertFalse(result.containsKey(PROPERTY));
+  }
+
+  @Test
+  public void preserveNullValue() {
+    Map<String, Object> event = new HashMap<>();
+    event.put(PROPERTY, null);
+
+    var result = rule.apply(event);
+
+    assertTrue(result.containsKey(PROPERTY));
+    assertNull(result.get(PROPERTY));
+  }
+}
diff --git a/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformRuleTest.java b/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformRuleTest.java
index 37416809ac..de7c6098c6 100644
--- a/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformRuleTest.java
+++ b/streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/UnitTransformRuleTest.java
@@ -25,6 +25,9 @@ import java.util.List;
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class UnitTransformRuleTest {
 
@@ -129,6 +132,105 @@ public class UnitTransformRuleTest {
     assertEquals(273.15, result.get(PROPERTY_TWO));
   }
 
+  @Test
+  public void preserveMissingValue() {
+    var unitTransformationRule = getTopLevelRule();
+    Map<String, Object> event = new HashMap<>();
+
+    var result = unitTransformationRule.apply(event);
+
+    assertFalse(result.containsKey(PROPERTY_TWO));
+  }
+
+  @Test
+  public void preserveNullValue() {
+    var unitTransformationRule = getTopLevelRule();
+    Map<String, Object> event = new HashMap<>();
+    event.put(PROPERTY_TWO, null);
+
+    var result = unitTransformationRule.apply(event);
+
+    assertTrue(result.containsKey(PROPERTY_TWO));
+    assertNull(result.get(PROPERTY_TWO));
+  }
+
+  @Test
+  public void preserveInvalidValue() {
+    var unitTransformationRule = getTopLevelRule();
+    Map<String, Object> event = new HashMap<>();
+    event.put(PROPERTY_TWO, "invalid");
+
+    var result = unitTransformationRule.apply(event);
+
+    assertEquals("invalid", result.get(PROPERTY_TWO));
+  }
+
+  @Test
+  public void preserveMissingNestedParent() {
+    var unitTransformationRule = getNestedRule();
+    Map<String, Object> event = new HashMap<>();
+
+    var result = unitTransformationRule.apply(event);
+
+    assertFalse(result.containsKey(PARENT_PROPERTY));
+  }
+
+  @Test
+  public void preserveNullNestedParent() {
+    var unitTransformationRule = getNestedRule();
+    Map<String, Object> event = new HashMap<>();
+    event.put(PARENT_PROPERTY, null);
+
+    var result = unitTransformationRule.apply(event);
+
+    assertTrue(result.containsKey(PARENT_PROPERTY));
+    assertNull(result.get(PARENT_PROPERTY));
+  }
+
+  @Test
+  public void preserveMissingNestedChild() {
+    var unitTransformationRule = getNestedRule();
+    Map<String, Object> subEvent = new HashMap<>();
+    Map<String, Object> event = new HashMap<>();
+    event.put(PARENT_PROPERTY, subEvent);
+
+    var result = unitTransformationRule.apply(event);
+
+    var resultSubEvent = (Map<?, ?>) result.get(PARENT_PROPERTY);
+    assertFalse(resultSubEvent.containsKey(CHILD_PROPERTY));
+  }
+
+  @Test
+  public void preserveNullNestedChild() {
+    var unitTransformationRule = getNestedRule();
+    Map<String, Object> subEvent = new HashMap<>();
+    subEvent.put(CHILD_PROPERTY, null);
+    Map<String, Object> event = new HashMap<>();
+    event.put(PARENT_PROPERTY, subEvent);
+
+    var result = unitTransformationRule.apply(event);
+
+    var resultSubEvent = (Map<?, ?>) result.get(PARENT_PROPERTY);
+    assertTrue(resultSubEvent.containsKey(CHILD_PROPERTY));
+    assertNull(resultSubEvent.get(CHILD_PROPERTY));
+  }
+
+  private UnitTransformationRule getTopLevelRule() {
+    return new UnitTransformationRule(
+        List.of(PROPERTY_TWO),
+        UNIT_DEGREE_CELSIUS,
+        UNIT_KELVIN
+    );
+  }
+
+  private UnitTransformationRule getNestedRule() {
+    return new UnitTransformationRule(
+        List.of(PARENT_PROPERTY, CHILD_PROPERTY),
+        UNIT_DEGREE_CELSIUS,
+        UNIT_KELVIN
+    );
+  }
+
   private Map<String, Object> getEventWithTwoProperties(double value1, double value2) {
     Map<String, Object> event = new HashMap<>();
     event.put(PROPERTY_ONE, value1);