Add support Map & Tuple

Author: mzitnik

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

@@ -93,6 +93,8 @@ private String createSimplePOJOTableSQL(String database, String tableName) {
                 "uuid UUID," +
                 "stringList Array(String)," +
                 "longList Array(Int64)," +
+                "mapOfStrings Map(String,String)," +
+                "tupleOfObjects Tuple(String,Int64,Boolean)," +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive); ";
@@ -285,6 +287,7 @@ void SimplePOJODataTest() throws Exception {
         simplePOJOs.sinkTo(simplePOJOSink);
         int rows = executeAsyncJob(env, tableName, 10, EXPECTED_ROWS);
         Assertions.assertEquals(EXPECTED_ROWS, rows);
+        ClickHouseServerForTests.showData(tableName);
     }
 
     @Test

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

@@ -71,5 +71,10 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {
         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())));
+
+        Serialize.writeMap(out, input.getMapOfStrings(), ClickHouseColumn.of("mapOfStrings", ClickHouseDataType.Map, false, ClickHouseColumn.of("", ClickHouseDataType.String.toString()), ClickHouseColumn.of("", ClickHouseDataType.String.toString())));
+
+        Serialize.writeTuple(out, input.getTupleOfObjects(), ClickHouseColumn.of("tupleOfObjects", ClickHouseDataType.Tuple, false, ClickHouseColumn.of("", ClickHouseDataType.String.toString()), ClickHouseColumn.of("", ClickHouseDataType.Int64.toString()), ClickHouseColumn.of("", ClickHouseDataType.Bool.toString()) ));
     }
+
 }

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

@@ -4,9 +4,7 @@
 import java.math.BigInteger;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
 
 public class SimplePOJO {
 
@@ -69,6 +67,10 @@ public class SimplePOJO {
     private List<String> stringList;
     private List<Long> longList;
 
+    private Map<String, String> mapOfStrings;
+
+    private List<Object> tupleOfObjects;
+
     public SimplePOJO(int index) {
         this.bytePrimitive = Byte.MIN_VALUE;
         this.byteObject = Byte.MAX_VALUE;
@@ -136,6 +138,15 @@ public SimplePOJO(int index) {
         this.longList.add(2L);
         this.longList.add(3L);
         this.longList.add(4L);
+
+        this.mapOfStrings = new HashMap<>();
+        this.mapOfStrings.put("a", "a");
+        this.mapOfStrings.put("b", "b");
+
+        this.tupleOfObjects =  new ArrayList<>();
+        this.tupleOfObjects.add("test");
+        this.tupleOfObjects.add(1L);
+        this.tupleOfObjects.add(true);
     }
 
     public byte getBytePrimitive() {
@@ -242,4 +253,7 @@ public Double getDoubleObject() {
 
     public List<Long> getLongList() { return longList; }
 
+    public Map<String, String> getMapOfStrings() { return mapOfStrings; }
+
+    public List<Object> getTupleOfObjects() { return tupleOfObjects; }
 }

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

@@ -300,4 +300,14 @@ public static void writeArray(OutputStream out, Object value, ClickHouseColumn c
         SerializerUtils.serializeArrayData(out, value, column);
     }
 
+    // Map
+    public static void writeMap(OutputStream out, Object value, ClickHouseColumn column) throws IOException {
+        SerializerUtils.serializeData(out, value, column);
+    }
+
+    // Tuple
+    public static void writeTuple(OutputStream out, Object value, ClickHouseColumn column) throws IOException {
+        SerializerUtils.serializeData(out, value, column);
+    }
+
 }