From 68d2edcfe613044c163a1ab33c79ef06d462e5ee Mon Sep 17 00:00:00 2001
From: Paultagoras <paul.moore@clickhouse.com>
Date: Fri, 19 Jul 2024 18:12:14 -0400
Subject: [PATCH 1/2] Adding more nuanced split method and tests

---
 .../kafka/connect/sink/db/mapping/Column.java |  3 ++-
 .../clickhouse/kafka/connect/util/Utils.java  | 26 +++++++++++++++++++
 .../connect/sink/db/mapping/ColumnTest.java   |  9 +++++++
 3 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/src/main/java/com/clickhouse/kafka/connect/sink/db/mapping/Column.java b/src/main/java/com/clickhouse/kafka/connect/sink/db/mapping/Column.java
index d60e4706..4e23b361 100644
--- a/src/main/java/com/clickhouse/kafka/connect/sink/db/mapping/Column.java
+++ b/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();
diff --git a/src/main/java/com/clickhouse/kafka/connect/util/Utils.java b/src/main/java/com/clickhouse/kafka/connect/util/Utils.java
index fe45b567..c180eca7 100644
--- a/src/main/java/com/clickhouse/kafka/connect/util/Utils.java
+++ b/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;
+    }
 }
diff --git a/src/test/java/com/clickhouse/kafka/connect/sink/db/mapping/ColumnTest.java b/src/test/java/com/clickhouse/kafka/connect/sink/db/mapping/ColumnTest.java
index 16642687..d933eddc 100644
--- a/src/test/java/com/clickhouse/kafka/connect/sink/db/mapping/ColumnTest.java
+++ b/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"));
+    }
 }
 

From d6da10c55d63e27f394d2c440021871cd0857ddf Mon Sep 17 00:00:00 2001
From: Paultagoras <paul.moore@clickhouse.com>
Date: Fri, 19 Jul 2024 18:13:56 -0400
Subject: [PATCH 2/2] Updated version

---
 CHANGELOG.md | 3 +++
 VERSION      | 2 +-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 59068c78..e89d6aa0 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/VERSION b/VERSION
index 0f1acbd5..99a4aef0 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-v1.1.2
+v1.1.3