Add support for Date/Date32/DateTime/DateTime64

Author: mzitnik

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

@@ -215,7 +215,11 @@ void SimplePOJODataTest() throws Exception {
                 "booleanPrimitive Boolean," +
                 "booleanObject Boolean," +
                 "str String," +
-                "fixedStr FixedString(10)" +
+                "fixedStr FixedString(10)," +
+                "v_date Date," +
+                "v_date32 Date32," +
+                "v_dateTime DateTime," +
+                "v_dateTime64 DateTime64," +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive); ";
@@ -484,7 +488,11 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
                 "booleanPrimitive Boolean," +
                 "booleanObject Boolean," +
                 "str String," +
-                "fixedStr FixedString(10)" +
+                "fixedStr FixedString(10)," +
+                "v_date Date," +
+                "v_date32 Date32," +
+                "v_dateTime DateTime," +
+                "v_dateTime64 DateTime64," +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive) " +

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

@@ -38,5 +38,10 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {
         Serialize.writeString(out, input.getStr(), false, false, ClickHouseDataType.String, false, "str");
         Serialize.writeFixedString(out, input.getFixedStr(), false, false, ClickHouseDataType.FixedString, false, 10, "fixedStr");
 
+        Serialize.writeDate(out, input.getDate(), false, false, ClickHouseDataType.Date, false, "v_date");
+        Serialize.writeDate32(out, input.getDate32(), false, false, ClickHouseDataType.Date32, false, "v_date32");
+        Serialize.writeTimeDate(out, input.getDateTime(), false, false, ClickHouseDataType.DateTime, false, "v_dateTime");
+        Serialize.writeTimeDate64(out, input.getDateTime64(), false, false, ClickHouseDataType.DateTime64, false, "v_dateTime64", 1);
+
     }
 }

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

@@ -1,6 +1,8 @@
 package org.apache.flink.connector.clickhouse.sink.pojo;
 
 import java.math.BigInteger;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
 
 public class SimplePOJO {
 
@@ -32,6 +34,11 @@ public class SimplePOJO {
     private BigInteger bigInteger128;
     private BigInteger bigInteger256;
 
+    private LocalDate date;
+    private LocalDate date32;
+    private LocalDateTime dateTime;
+    private LocalDateTime dateTime64;
+
     public SimplePOJO(int index) {
         this.bytePrimitive = Byte.MIN_VALUE;
         this.byteObject = Byte.MAX_VALUE;
@@ -60,6 +67,10 @@ public SimplePOJO(int index) {
         this.bigInteger128 = BigInteger.valueOf(longPrimitive);
         this.bigInteger256 = BigInteger.valueOf(longPrimitive);
 
+        this.date = LocalDate.ofEpochDay(0);
+        this.date32 = LocalDate.ofEpochDay(0);
+        this.dateTime = LocalDateTime.now();
+        this.dateTime64 = LocalDateTime.now();
 
     }
 
@@ -122,4 +133,13 @@ public Double getDoubleObject() {
     public BigInteger getBigInteger256() { return bigInteger256; }
 
     public String getFixedStr() { return fixedStr; }
+
+    public LocalDate getDate() { return date; }
+
+    public LocalDate getDate32() { return date32; }
+
+    public LocalDateTime getDateTime() { return dateTime; }
+
+    public LocalDateTime getDateTime64() { return dateTime64; }
+
 }

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

@@ -12,6 +12,7 @@
 import java.lang.reflect.Method;
 import java.math.BigInteger;
 import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.util.HashMap;
@@ -128,6 +129,43 @@ public static void writeDate(OutputStream out, ZonedDateTime value, boolean defa
         }
     }
 
+    public static void writeDate32(OutputStream out, LocalDate value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            SerializerUtils.writeDate32(out, value, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
+    public static void writeDate32(OutputStream out, ZonedDateTime value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            SerializerUtils.writeDate32(out, value, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
+    // Support for DateTime section
+    public static void writeTimeDate(OutputStream out, LocalDateTime value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            SerializerUtils.writeDateTime(out, value, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
+    public static void writeTimeDate(OutputStream out, ZonedDateTime value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            SerializerUtils.writeDateTime(out, value, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
+    public static void writeTimeDate64(OutputStream out, LocalDateTime value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column, int scale) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            SerializerUtils.writeDateTime64(out, value, scale, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
+    public static void writeTimeDate64(OutputStream out, ZonedDateTime value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column, int scale) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            SerializerUtils.writeDateTime64(out, value, scale, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
     // clickhouse type String support
     public static void writeString(OutputStream out, String value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
         if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {