Add Array support

Author: mzitnik

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java

@@ -91,6 +91,8 @@ private String createSimplePOJOTableSQL(String database, String tableName) {
                 "v_dateTime DateTime," +
                 "v_dateTime64 DateTime64," +
                 "uuid UUID," +
+                "stringList Array(String)," +
+                "longList Array(Int64)," +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive); ";
@@ -532,6 +534,7 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
         simplePOJOs.sinkTo(simplePOJOSink);
         int rows = executeAsyncJob(env, tableName, 100, EXPECTED_ROWS);
         Assertions.assertEquals(EXPECTED_ROWS, rows);
+        ClickHouseServerForTests.showData("simple_too_many_parts_pojo");
         //ClickHouseServerForTests.executeSql(String.format("SYSTEM START MERGES `%s.%s`", getDatabase(), tableName));
     }
 

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/convertor/SimplePOJOConvertor.java

@@ -1,5 +1,6 @@
 package org.apache.flink.connector.clickhouse.sink.convertor;
 
+import com.clickhouse.data.ClickHouseColumn;
 import com.clickhouse.data.ClickHouseDataType;
 import com.clickhouse.utils.Serialize;
 import org.apache.flink.connector.clickhouse.convertor.POJOConvertor;
@@ -66,5 +67,9 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {
         Serialize.writeTimeDate64(out, input.getDateTime64(), false, false, ClickHouseDataType.DateTime64, false, "v_dateTime64", 1);
 
         Serialize.writeUUID(out, input.getUuid(), false, false, ClickHouseDataType.UUID, false, "uuid");
+
+        Serialize.writeArray(out, input.getStringList(), ClickHouseColumn.of("stringList", ClickHouseDataType.Array, false, ClickHouseColumn.of("", ClickHouseDataType.String.toString())));
+
+        Serialize.writeArray(out, input.getLongList(), ClickHouseColumn.of("longList", ClickHouseDataType.Array, false, ClickHouseColumn.of("", ClickHouseDataType.Int64.toString())));
     }
 }

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/pojo/SimplePOJO.java

@@ -4,6 +4,8 @@
 import java.math.BigInteger;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 
 public class SimplePOJO {
@@ -64,6 +66,9 @@ public class SimplePOJO {
 
     private UUID uuid;
 
+    private List<String> stringList;
+    private List<Long> longList;
+
     public SimplePOJO(int index) {
         this.bytePrimitive = Byte.MIN_VALUE;
         this.byteObject = Byte.MAX_VALUE;
@@ -119,6 +124,18 @@ public SimplePOJO(int index) {
         this.dateTime64 = LocalDateTime.now();
 
         this.uuid = UUID.randomUUID();
+
+        this.stringList = new ArrayList<>();
+        this.stringList.add("a");
+        this.stringList.add("b");
+        this.stringList.add("c");
+        this.stringList.add("d");
+
+        this.longList = new ArrayList<>();
+        this.longList.add(1L);
+        this.longList.add(2L);
+        this.longList.add(3L);
+        this.longList.add(4L);
     }
 
     public byte getBytePrimitive() {
@@ -221,4 +238,8 @@ public Double getDoubleObject() {
 
     public UUID getUuid() { return uuid; }
 
+    public List<String> getStringList() { return stringList; }
+
+    public List<Long> getLongList() { return longList; }
+
 }

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/test/embedded/clickhouse/ClickHouseServerForTests.java

@@ -102,6 +102,18 @@ public static void executeSql(String sql) throws ExecutionException, Interrupted
         }
     }
 
+    public static void showData(String tableName) throws ExecutionException, InterruptedException {
+        String showDataSql = String.format("select * from '%s'", tableName);
+        Client client = ClickHouseTestHelpers.getClient(host, port, isSSL, username, password);
+        List<GenericRecord> content = client.queryAll(showDataSql);
+        for (GenericRecord record : content) {
+            System.out.println();
+            for (int i = 0; i< record.getSchema().getColumns().toArray().length; i++) {
+                System.out.print(" | " + record.getObject(i +1));
+            }
+        }
+    }
+
     public static int countParts(String table) {
         String countPartsSql = String.format("SELECT count(*) FROM system.parts WHERE table = '%s' and active = 1", table);
         Client client = ClickHouseTestHelpers.getClient(host, port, isSSL, username, password);

flink-connector-clickhouse-base/src/main/java/com/clickhouse/utils/Serialize.java

@@ -1,6 +1,7 @@
 package com.clickhouse.utils;
 
 import com.clickhouse.client.api.data_formats.internal.SerializerUtils;
+import com.clickhouse.data.ClickHouseColumn;
 import com.clickhouse.data.ClickHouseDataType;
 import com.clickhouse.data.format.BinaryStreamUtils;
 import org.slf4j.Logger;
@@ -294,4 +295,9 @@ public static void writeUUID(OutputStream out, UUID value, boolean defaultsSuppo
         }
     }
 
+    // Array
+    public static void writeArray(OutputStream out, Object value, ClickHouseColumn column) throws IOException {
+        SerializerUtils.serializeArrayData(out, value, column);
+    }
+
 }