feat(filters): add support for dot-notation to RenameFilter

Author: fhussonnois

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

@@ -78,17 +78,6 @@ private TypedStruct(final StructSchema schema) {
         this.values = new LinkedList<>();
     }
 
-    public TypedStruct rename(final String field, final String newField) {
-        schema.rename(field, newField);
-        return this;
-    }
-
-    public TypedStruct remove(final String field) {
-        TypedField removed = schema.remove(field);
-        if (removed != null) values.remove(removed.index());
-        return this;
-    }
-
     /**
      * {@inheritDoc}
      */
@@ -319,7 +308,57 @@ public <K, V> Map<K, V> getMap(final String field) throws DataException {
         return getCheckedType(field, Type.MAP);
     }
 
+
+    /**
+     * Renames the field present to the given path.
+     *
+     * @param path      the path of the field.
+     * @param newField  the new field name.
+     * @return          return this.
+     */
+    public TypedStruct rename(final String path, final String newField) {
+        if (has(path)) {
+            schema.rename(path, newField);
+            return this;
+        }
+
+        if (isDotPropertyAccessPath(path)) {
+            String[] split = path.split("\\.", 2);
+            if (has(split[0])) {
+                TypedValue child = get(split[0]);
+                if (child.schema().type() == Type.STRUCT) {
+                    return child.getStruct().rename(split[1], newField);
+                }
+            }
+        }
+        return this;
+    }
+
+    public TypedStruct remove(final String field) {
+        TypedField removed = schema.remove(field);
+        if (removed != null) values.remove(removed.index());
+        return this;
+    }
+
+    /**
+     * Checks if a field exist for the given path.
+     *
+     * @param path  the path to check.
+     * @return      {@code true} if the path exists.
+     */
+    public boolean exists(final String path) {
+        Objects.requireNonNull(path, "path cannot be null");
+        return find(path) != null;
+    }
+
+    /**
+     * Finds the value for the given path.
+     *
+     * @param path  the path.
+     * @return      the value or {@code null} if no value exists.
+     */
     public TypedValue find(final String path) {
+        Objects.requireNonNull(path, "path cannot be null");
         if (has(path)) return get(path);
 
         if (isDotPropertyAccessPath(path)) {
@@ -334,6 +373,13 @@ public TypedValue find(final String path) {
         return null;
     }
 
+    /**
+     * Inserts the given object value to the given path.
+     *
+     * @param path      the object path.
+     * @param value     the object value.
+     * @return          return this?
+     */
     public TypedStruct insert(final String path, final Object value) {
         if (path == null || path.isEmpty()) {
             throw new IllegalArgumentException("Cannot insert value given null or empty path");

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

@@ -54,7 +54,7 @@ public RecordsIterable<TypedStruct> apply(final FilterContext context,
                                               final TypedStruct record,
                                               final boolean hasNext) throws FilterException {
 
-        if (record.has(configs.field())) {
+        if (record.exists(configs.field())) {
             record.rename(configs.field(), configs.target());
             return new RecordsIterable<>(record);
 

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

@@ -41,19 +41,36 @@ public void setUp() {
     }
 
     @Test
-    public void shouldRenameGivenExistingField() {
+    public void should_rename_given_existing_fields() {
 
         configs.put(RenameFilterConfig.RENAME_FIELD_CONFIG, "foo");
         configs.put(RenameFilterConfig.RENAME_TARGET_CONFIG, "bar");
         filter.configure(configs);
 
         TypedStruct record = TypedStruct.create().put("foo", "dummy-value");
-        List<TypedStruct> results = this.filter.apply(null, record, false).collect();
+        List<TypedStruct> results = filter.apply(null, record, false).collect();
 
         Assert.assertNotNull(results);
         Assert.assertEquals(1, results.size());
 
         TypedStruct result = results.get(0);
         Assert.assertEquals("dummy-value", result.getString("bar"));
     }
+
+    @Test
+    public void should_rename_given_existing_path() {
+
+        configs.put(RenameFilterConfig.RENAME_FIELD_CONFIG, "foo.bar");
+        configs.put(RenameFilterConfig.RENAME_TARGET_CONFIG, "foo");
+        filter.configure(configs);
+
+        TypedStruct record = TypedStruct.create().insert("foo.bar", "dummy-value");
+        List<TypedStruct> results = filter.apply(null, record, false).collect();
+
+        Assert.assertNotNull(results);
+        Assert.assertEquals(1, results.size());
+
+        TypedStruct result = results.get(0);
+        Assert.assertEquals("dummy-value", result.find("foo.foo").getString());
+    }
 }