Tweak to differentiate between Nullable and regular arrays

ClickHouse · Dec 19, 2023 · 3b05323 · 3b05323
1 parent 435db77
commit 3b05323
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 6 deletions.
diff --git a/src/main/java/com/clickhouse/kafka/connect/sink/db/ClickHouseWriter.java b/src/main/java/com/clickhouse/kafka/connect/sink/db/ClickHouseWriter.java
@@ -350,10 +350,10 @@ private void doWriteCol(Record record, Column col, ClickHousePipedOutputStream s
                 BinaryStreamUtils.writeNonNull(stream);
             }
             if (!col.isNullable() && value.getObject() == null) {
-                if (colType != Type.ARRAY) {//Arrays can't be nullable so we ignore those
-                    // this the situation when the col is not isNullable, but the data is null here we need to drop the records
+                if (colType == Type.ARRAY && col.getSubType().isNullable())
+                    BinaryStreamUtils.writeNonNull(stream);
+                else
                     throw new RuntimeException(String.format("An attempt to write null into not nullable column '%s'", col.getName()));
-                }
             }
             switch (colType) {
                 case INT8:
@@ -403,7 +403,6 @@ private void doWriteCol(Record record, Column col, ClickHousePipedOutputStream s
                     List<?> arrObject = (List<?>) value.getObject();
 
                     if (arrObject == null) {
-                        BinaryStreamUtils.writeNonNull(stream);
                         doWritePrimitive(colType, value.getFieldType(), stream, new ArrayList<>());
                     } else {
                         int sizeArrObject = arrObject.size();

diff --git a/src/test/java/com/clickhouse/kafka/connect/sink/ClickHouseSinkTaskWithSchemaTest.java b/src/test/java/com/clickhouse/kafka/connect/sink/ClickHouseSinkTaskWithSchemaTest.java
@@ -120,7 +120,7 @@ public void mapTypesTest() {
     }
 
     @Test
-    public void nullableArrayTypeTest() {
+    public void nullArrayTypeTest() {
         Map<String, String> props = getTestProperties();
         ClickHouseHelperClient chc = createClient(props);
 
@@ -131,7 +131,22 @@ public void nullableArrayTypeTest() {
 
         ClickHouseSinkTask chst = new ClickHouseSinkTask();
         chst.start(props);
-//        assertThrowsExactly(RuntimeException.class, () -> chst.put(sr), "An attempt to write null into not nullable column 'arr'");
+        assertThrowsExactly(RuntimeException.class, () -> chst.put(sr), "An attempt to write null into not nullable column 'arr'");
+        chst.stop();
+    }
+
+    @Test
+    public void nullableArrayTypeTest() {
+        Map<String, String> props = getTestProperties();
+        ClickHouseHelperClient chc = createClient(props);
+
+        String topic = "nullable_array_string_table_test";
+        ClickHouseTestHelpers.dropTable(chc, topic);
+        ClickHouseTestHelpers.createTable(chc, topic, "CREATE TABLE %s ( `off16` Int16, `arr` Array(Nullable(String))  ) Engine = MergeTree ORDER BY off16");
+        Collection<SinkRecord> sr = SchemaTestData.createNullableArrayType(topic, 1);
+
+        ClickHouseSinkTask chst = new ClickHouseSinkTask();
+        chst.start(props);
         chst.put(sr);
         chst.stop();