fix(filters): fix ExplodeFilter should support dotted notation

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/TypedStruct.java

@@ -21,7 +21,6 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.LinkedList;
-
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -502,8 +501,14 @@ public int hashCode() {
     @Override
     public String toString() {
         return StreamSupport.stream(schema.spliterator(), false)
-                .map( field -> field.name() + "=" + get(field))
-                .collect(Collectors.joining(",", "[", "]"));
+            .map( field -> {
+                TypedValue value = get(field);
+                return
+                    "name: " + field.name()
+                    + ", type: " +  value.schema().type()
+                    + ", value: " +  value.value();
+            })
+            .collect(Collectors.joining(", ", "[", "]"));
     }
 
     /**

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/merger/DefaultTypeValueMerger.java

@@ -28,7 +28,9 @@
 
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Default {@link TypeValueMerger} implementation which automatically
@@ -42,37 +44,54 @@ public class DefaultTypeValueMerger implements TypeValueMerger {
      * @param left      the left {@link TypedStruct} to be merged.
      * @param right     the right {@link TypedStruct} to be merged.
      * @param overwrite the left field that must overwritten.
-     *
-     * @return       the new {@link TypedStruct} instance.
+     * @return the new {@link TypedStruct} instance.
      */
     public TypedStruct merge(final TypedStruct left,
                              final TypedStruct right,
                              final Set<String> overwrite) {
+        return mergeObjects(left, right, overwrite.stream().map(Path::new).collect(Collectors.toSet()));
+    }
 
+    private static TypedStruct mergeObjects(final TypedStruct left,
+                                            final TypedStruct right,
+                                            final Set<Path> overwrite) {
         if (left == null) return right;
         if (right == null) return left;
 
         final TypedStruct struct = TypedStruct.create();
 
         for (TypedField leftField : left) {
-            final String leftFieldName = leftField.name();
-
-            boolean isOverwrite = overwrite.contains(leftFieldName);
+            final String fieldName = leftField.name();
 
-            TypedValue typedValue = left.get(leftField);
+            final TypedValue leftValue = left.get(leftField);
 
-            if (!right.has(leftFieldName)) {
-                struct.put(leftFieldName, typedValue);
+            if (!right.has(fieldName)) {
+                struct.put(fieldName, leftValue);
                 continue;
             }
 
+            boolean isOverwrite = overwrite.stream().anyMatch(it -> it.matches(fieldName));
+
             if (isOverwrite) {
                 continue; // skip the left field
             }
 
-            final TypedField rightField = right.field(leftFieldName);
+            final TypedField rightField = right.field(fieldName);
             checkIfTypesAreCompatibleForMerge(leftField, rightField);
-            TypedValue merged = merge(typedValue, right.get(rightField));
+
+            final TypedValue rightValue = right.get(rightField);
+            final TypedValue merged;
+
+            if (leftField.type() == Type.STRUCT) {
+                final TypedStruct mergedStruct = mergeObjects(
+                    leftValue.getStruct(),
+                    rightValue.getStruct(),
+                    overwrite.stream().map(Path::forward).filter(Objects::nonNull).collect(Collectors.toSet())
+                );
+                merged = TypedValue.struct(mergedStruct);
+            } else {
+                merged = merge(leftValue, rightValue);
+            }
             struct.put(leftField.name(), merged);
         }
 
@@ -84,6 +103,33 @@ public TypedStruct merge(final TypedStruct left,
         return struct;
     }
 
+    private static class Path {
+
+        private final String path;
+        private final String field;
+        private final String remaining;
+
+        Path(final String path) {
+            this.path = path;
+            if (path.contains(".")) {
+                String[] split = path.split("\\.", 2);
+                field = split[0];
+                remaining = split[1];
+            } else {
+                field = path;
+                remaining = null;
+            }
+        }
+
+        public boolean matches(final String field) {
+            return this.field.equals(field) || path.equals(field);
+        }
+
+        private Path forward() {
+            return remaining == null ? null : new Path(remaining);
+        }
+    }
+
     private static TypedValue merge(final TypedValue left, final TypedValue right) {
         List<Object> values = new LinkedList<>();
 
@@ -170,4 +216,5 @@ private static void checkIfTypesAreCompatibleForMerge(final TypedField left, fin
     private static boolean isTypeNotEqual(final Schema left, final Schema right) {
         return left != null && right != null && left.type() != right.type();
     }
+
 }

connect-file-pulse-api/src/test/java/io/streamthoughts/kafka/connect/filepulse/data/merger/TypeValueMergerTest.java

@@ -132,4 +132,20 @@ public void shouldMergeStructGivenTwoArrayFieldsWithEqualsValueType() {
         assertEquals(VALUE_A,  merged.getArray(FIELD_VALUE_A).get(0));
         assertEquals(VALUE_B,  merged.getArray(FIELD_VALUE_A).get(1));
     }
+
+    @Test
+    public void shouldMergedGivenTwoStructWithCommonChildFields() {
+        final TypedStruct structLeft = TypedStruct.create()
+            .insert("a.b1.c", VALUE_A);
+
+        final TypedStruct structRight = TypedStruct.create()
+            .insert("a.b2.c", VALUE_B);
+
+        final TypedStruct merged = merger.merge(structLeft, structRight, Collections.emptySet());
+
+        assertNotNull(merged);
+
+        assertEquals(VALUE_A,  merged.find("a.b1.c").getString());
+        assertEquals(VALUE_B,  merged.find("a.b2.c").getString());
+    }
 }

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/ExplodeFilter.java

@@ -63,15 +63,16 @@ public ConfigDef configDef() {
     @Override
     protected RecordsIterable<TypedStruct> apply(final FilterContext context,
                                                  final TypedStruct record) throws FilterException {
-        TypedValue value = checkIsNotNull(record.get(config.source()));
+        final TypedValue value = checkIsNotNull(record.find(config.source()));
+
         if (value.type() != Type.ARRAY) {
             throw new FilterException(
-                    "Invalid type for field '" + config.source()  + "', expected ARRAY, was " + value.type());
+                "Invalid type for field '" + config.source()  + "', expected ARRAY, was " + value.type());
         }
 
         final List<TypedStruct> explode = value.getArray()
                 .stream()
-                .map(it -> TypedStruct.create().put(config.source(), TypedValue.any(it)))
+                .map(it -> TypedStruct.create().insert(config.source(), TypedValue.any(it)))
                 .collect(Collectors.toList());
 
         return new RecordsIterable<>(explode);

connect-file-pulse-filters/src/test/java/io/streamthoughts/kafka/connect/filepulse/filter/ExplodeFilterTest.java

@@ -80,6 +80,19 @@ public void should_explode_input_given_sub_field_source() {
         Assert.assertEquals(PRIMITIVE_VALUES, extractStringValues(result, it -> it.getString("values")));
     }
 
+    @Test
+    public void should_explode_input_given_sub_dotted_field_source() {
+        filter.configure(new HashMap<String, String>() {{
+            put(CommonFilterConfig.FILTER_SOURCE_FIELD_CONFIG, "field.values");
+        }});
+
+        RecordsIterable<TypedStruct> result = filter.apply(null, create()
+                .insert("message", "dummy")
+                .insert("field.values", PRIMITIVE_VALUES), false);
+        Assert.assertEquals(PRIMITIVE_VALUES.size(), result.size());
+        Assert.assertEquals(PRIMITIVE_VALUES, extractStringValues(result, it -> it.find("field.values").getString()));
+    }
+
     private <T> List<T> extractStringValues(final RecordsIterable<TypedStruct> result,
                                              final Function<TypedStruct, T> extractor) {
         return result.stream().map(extractor).collect(Collectors.toList());