Fix NPE on V1 when table has array(nested) field (#497)

Closes #496
ClickHouse · Feb 9, 2025 · a95f74b · a95f74b
1 parent 0e49af7
commit a95f74b
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 1 deletion.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,6 @@
+# 1.2.8
+* Fix NPE on V1 client when Clickhouse has table with Array(Nested...) field
+
 # 1.2.7
 * Bump clickhouse-java to 0.8.0
 * Fix for handling of nullable fields in tuples ClickHouse data type

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
@@ -186,6 +186,9 @@ public static Column extractColumn(String name, String valueType, boolean isNull
                     .build();
         } else if (valueType.startsWith("Array")) {
             Column arrayType = extractColumn(name, valueType.substring("Array".length() + 1, valueType.length() - 1), false, isSubColumn, hasDefaultValue, arrayDepth + 1);
+            if (arrayType == null) {
+                return null;
+            }
 
             Column array = builder.type(Type.ARRAY)
                     .arrayType(arrayType)

diff --git a/src/test/java/com/clickhouse/kafka/connect/sink/ClickHouseBase.java b/src/test/java/com/clickhouse/kafka/connect/sink/ClickHouseBase.java
@@ -78,14 +78,16 @@ protected ClickHouseHelperClient createClient(Map<String,String> props, boolean
         String password = csc.getPassword();
         boolean sslEnabled = csc.isSslEnabled();
         int timeout = csc.getTimeout();
-
+        String clientVersion = csc.getClientVersion();
+        boolean useClientV2 = clientVersion.equals("V1") ? false : true;
         ClickHouseHelperClient tmpChc = new ClickHouseHelperClient.ClickHouseClientBuilder(hostname, port, csc.getProxyType(), csc.getProxyHost(), csc.getProxyPort())
                 .setDatabase(database)
                 .setUsername(username)
                 .setPassword(password)
                 .sslEnable(sslEnabled)
                 .setTimeout(timeout)
                 .setRetry(csc.getRetry())
+                .useClientV2(useClientV2)
                 .build();
 
         if (withDatabase) {
@@ -98,6 +100,7 @@ protected ClickHouseHelperClient createClient(Map<String,String> props, boolean
                     .sslEnable(sslEnabled)
                     .setTimeout(timeout)
                     .setRetry(csc.getRetry())
+                    .useClientV2(useClientV2)
                     .build();
             }
 

diff --git a/src/test/java/com/clickhouse/kafka/connect/sink/db/helper/ClickHouseHelperClientTest.java b/src/test/java/com/clickhouse/kafka/connect/sink/db/helper/ClickHouseHelperClientTest.java
@@ -63,6 +63,22 @@ public void describeNestedFlattenedTable() {
         }
     }
 
+    @Test
+    public void ignoreArrayWithNestedTable() {
+        String topic = createTopicName("nested_table_test");
+        ClickHouseTestHelpers.createTable(chc, topic,
+                "CREATE TABLE %s ( `num` String, " +
+                        "`nested` Array(Nested (innerInt Int32, innerString String))) " +
+                        "Engine = MergeTree ORDER BY num");
+
+        try {
+            Table table = chc.describeTable(chc.getDatabase(), topic);
+            Assertions.assertNull(table);
+        } finally {
+            ClickHouseTestHelpers.dropTable(chc, topic);
+        }
+    }
+
     @Test
     @SinceClickHouseVersion("24.1")
     public void describeNestedUnFlattenedTable() {