fix(filters): DelimtedRowFileInputFilter should compute schema for each record (#171)

Author: fhussonnois

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

@@ -88,8 +88,8 @@ public void configure(final Map<String, ?> configs) {
 
     private boolean isMandatoryConfigsMissing() {
         return configs.schema() == null &&
-                configs.extractColumnName() == null &&
-                !configs.isAutoGenerateColumnNames();
+               configs.extractColumnName() == null &&
+               !configs.isAutoGenerateColumnNames();
     }
 
     /**
@@ -111,13 +111,22 @@ public RecordsIterable<TypedStruct> apply(final FilterContext context,
         final String source = record.first(DEFAULT_SOURCE_FIELD).getString();
 
         String[] columnValues = splitColumnValues(source);
-        if (schema == null) {
+
+        if (schema == null || isSchemaDynamic()) {
             inferSchemaFromRecord(record, columnValues.length);
         }
         final TypedStruct struct = buildStructForFields(columnValues);
         return RecordsIterable.of(struct);
     }
 
+    public boolean isSchemaDynamic() {
+        // Schema SHOULD be inferred for each record when columns name are auto generate.
+        // This rule is used to handle cases where records may have different number of columns.
+        return configs.extractColumnName() == null &&
+               configs.schema() == null &&
+               configs.isAutoGenerateColumnNames();
+    }
+
     private void inferSchemaFromRecord(final TypedStruct record, int numColumns) {
         schema = Schema.struct();
 
@@ -126,7 +135,8 @@ private void inferSchemaFromRecord(final TypedStruct record, int numColumns) {
             String field = record.first(fieldName).getString();
             if (field == null) {
                 throw new FilterException(
-                        "Can't found field for name '" + fieldName + "' to determine columns names");
+                    "Cannot find field for name '" + fieldName + "' to determine columns names"
+                );
             }
             final List<String> columns = Arrays
                     .stream(splitColumnValues(field))
@@ -168,7 +178,8 @@ private String[] splitColumnValues(final String value) {
     private TypedStruct buildStructForFields(final String[] fieldValues) {
         if (fieldValues.length > columnsTypesByIndex.size()) {
             throw new FilterException(
-                    "Error while reading delimited input row. Too large number of fields (" + fieldValues.length + ")");
+                "Error while reading delimited input row. Too large number of fields (" + fieldValues.length + ")"
+            );
         }
 
         TypedStruct struct = TypedStruct.create();

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

@@ -31,6 +31,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import static io.streamthoughts.kafka.connect.filepulse.config.DelimitedRowFilterConfig.READER_AUTO_GENERATE_COLUMN_NAME_CONFIG;
 import static io.streamthoughts.kafka.connect.filepulse.config.DelimitedRowFilterConfig.READER_EXTRACT_COLUMN_NAME_CONFIG;
 import static io.streamthoughts.kafka.connect.filepulse.config.DelimitedRowFilterConfig.READER_FIELD_COLUMNS_CONFIG;
 import static io.streamthoughts.kafka.connect.filepulse.config.DelimitedRowFilterConfig.READER_FIELD_DUPLICATE_COLUMNS_AS_ARRAY_CONFIG;
@@ -42,12 +43,10 @@ public class DelimitedRowFileInputFilterTest {
 
     private DelimitedRowFilter filter;
 
-
     private static final TypedStruct DEFAULT_STRUCT = TypedStruct.create()
         .put("message", "value1;2;true")
         .put("headers", Collections.singletonList("col1;col2;col3"));
 
-
     @Before
     public void setUp() {
         filter = new DelimitedRowFilter();
@@ -101,6 +100,32 @@ public void should_extract_repeated_columns_names_from_given_field() {
         Assert.assertEquals("value3", output.getString("col3"));
     }
 
+    @Test
+    public void should_generate_column_names_given_records_with_different_size() {
+        configs.put(READER_AUTO_GENERATE_COLUMN_NAME_CONFIG, "true");
+        filter.configure(configs, alias -> null);
+
+        TypedStruct input, output;
+
+        input = TypedStruct.create().put("message", "value1;value2;");
+        RecordsIterable<TypedStruct> iterable1 = filter.apply(null, input, false);
+        Assert.assertNotNull(iterable1);
+        Assert.assertEquals(1, iterable1.size());
+
+        output = iterable1.iterator().next();
+        Assert.assertNotNull(output.schema().field("column1"));
+        Assert.assertNotNull(output.schema().field("column2"));
+
+        input = TypedStruct.create().put("message", "value1;value2;value3");
+        RecordsIterable<TypedStruct> iterable2 = filter.apply(null, input, false);
+        Assert.assertNotNull(iterable2);
+        Assert.assertEquals(1, iterable2.size());
+
+        output = iterable2.iterator().next();
+        Assert.assertNotNull(output.schema().field("column1"));
+        Assert.assertNotNull(output.schema().field("column2"));
+        Assert.assertNotNull(output.schema().field("column3"));
+    }
 
     @Test(expected = DataException.class)
     public void should_fail_given_repeated_columns_names_and_duplicate_not_allowed() {