Merge pull request #416 from ClickHouse/fix-comma-column-name

Adding more nuanced split method and tests

CHANGELOG.md

@@ -1,3 +1,6 @@
+# 1.1.3
+* Bugfix to address commas in the column name of enums
+
 ## 1.1.2
 * Adding a "dateTimeFormat" configuration option to allow for custom date/time formatting with String schema values
 * Adding ephemeral column support and adding an error message

VERSION

@@ -1 +1 @@
-v1.1.2
+v1.1.3

src/main/java/com/clickhouse/kafka/connect/sink/db/mapping/Column.java

@@ -1,6 +1,7 @@
 package com.clickhouse.kafka.connect.sink.db.mapping;
 
 import com.clickhouse.kafka.connect.sink.db.helper.ClickHouseFieldDescriptor;
+import com.clickhouse.kafka.connect.util.Utils;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.Setter;
@@ -328,7 +329,7 @@ public static List<String> splitUnlessInBrackets(String input, char delimiter) {
 
     private static Map<String, Integer> extractEnumValues(String valueType) {
         Map <String, Integer> data = new HashMap<>();
-        String[] values = valueType.substring(valueType.indexOf("(") + 1, valueType.indexOf(")")).split(",");
+        List<String> values = Utils.splitIgnoringQuotes(valueType.substring(valueType.indexOf("(") + 1, valueType.indexOf(")")), ',');
         for (String value : values) {
             String[] val = value.split("=", 2);
             String key = val[0].trim();

src/main/java/com/clickhouse/kafka/connect/util/Utils.java

@@ -14,6 +14,7 @@
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -169,4 +170,29 @@ public static String getOffsets(Collection<SinkRecord> records) {
 
         return String.format("minOffset: %d, maxOffset: %d", minOffset, maxOffset);
     }
+
+    public static List<String> splitIgnoringQuotes(String input, char separator) {
+        List<String> result = new ArrayList<>();
+        StringBuilder sb = new StringBuilder();
+        boolean inSingleQuotes = false;
+        boolean inDoubleQuotes = false;
+
+        for (char c : input.toCharArray()) {
+            if (c == '\'' && !inDoubleQuotes) {
+                inSingleQuotes = !inSingleQuotes;
+                sb.append(c);
+            } else if (c == '"' && !inSingleQuotes) {
+                inDoubleQuotes = !inDoubleQuotes;
+                sb.append(c);
+            } else if (c == separator && !inSingleQuotes && !inDoubleQuotes) {
+                result.add(sb.toString().trim());
+                sb.setLength(0);
+            } else {
+                sb.append(c);
+            }
+        }
+        result.add(sb.toString().trim());
+
+        return result;
+    }
 }

src/test/java/com/clickhouse/kafka/connect/sink/db/mapping/ColumnTest.java

@@ -156,5 +156,14 @@ public void extractVariantOfPrimitives() {
             assertEquals(expectedSubtype.getScale(), actualSubtype.getScale());
         }
     }
+
+    @Test
+    public void extractEnumOfPrimitives() {
+        Column col = Column.extractColumn(newDescriptor("Enum8('a, valid' = 1, 'b' = 2)"));
+        assertEquals(Type.Enum8, col.getType());
+        assertEquals(2, col.getEnumValues().size());
+        assertTrue(col.getEnumValues().containsKey("a, valid"));
+        assertTrue(col.getEnumValues().containsKey("b"));
+    }
 }