Adding support for UUID

mzitnik · mzitnik · commit b9a7553a16b4 · 2025-09-04T09:57:28.000+03:00
diff --git a/README.md b/README.md
@@ -164,6 +164,7 @@ Planned for a future release — a complete end-to-end example will be added onc
 | long/Long       | Time64          | ❌         | N/A                        |
 | byte/Byte       | Enum8           | ✅         | Serialize.writeInt8        |
 | int/Integer     | Enum16          | ✅         | Serialize.writeInt16       |
+| java.util.UUID  | UUID            | ✅         | Serialize.writeIntUUID     |
 | String          | JSON            | ❌         | N/A                        |
 | Array<Type>     | Array<Type>     | ❌         | N/A                        |
 | Map<K,V>        | Map<K,V>        | ❌         | N/A                        |
diff --git a/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java b/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java
@@ -46,6 +46,56 @@ public class ClickHouseSinkTests extends FlinkClusterTests {
 
     static final int STREAM_PARALLELISM = 5;
 
+    private String createSimplePOJOTableSQL(String database, String tableName, int parts_to_throw_insert) {
+        String createTable = createSimplePOJOTableSQL(database, tableName);
+        return createTable.trim().substring(0, createTable.trim().length() - 1) + " " + String.format("SETTINGS parts_to_throw_insert = %d;",  parts_to_throw_insert);
+    }
+
+    private String createSimplePOJOTableSQL(String database, String tableName) {
+        return "CREATE TABLE `" + database + "`.`" + tableName + "` (" +
+                "bytePrimitive Int8," +
+                "byteObject Int8," +
+                "shortPrimitive Int16," +
+                "shortObject Int16," +
+                "intPrimitive Int32," +
+                "integerObject Int32," +
+                "longPrimitive Int64," +
+                "longObject Int64," +
+                "bigInteger128 Int128," +
+                "bigInteger256 Int256," +
+                "uint8Primitive  UInt8," +
+                "uint8Object UInt8," +
+                "uint16Primitive  UInt16," +
+                "uint16Object UInt16," +
+                "uint32Primitive  UInt32," +
+                "uint32Object UInt32," +
+                "uint64Primitive  UInt64," +
+                "uint64Object UInt64," +
+                "uint128Object UInt128," +
+                "uint256Object UInt256," +
+                "decimal Decimal(10,5)," +
+                "decimal32 Decimal32(9)," +
+                "decimal64 Decimal64(18)," +
+                "decimal128 Decimal128(38)," +
+                "decimal256 Decimal256(76)," +
+                "floatPrimitive Float," +
+                "floatObject Float," +
+                "doublePrimitive Double," +
+                "doubleObject Double," +
+                "booleanPrimitive Boolean," +
+                "booleanObject Boolean," +
+                "str String," +
+                "fixedStr FixedString(10)," +
+                "v_date Date," +
+                "v_date32 Date32," +
+                "v_dateTime DateTime," +
+                "v_dateTime64 DateTime64," +
+                "uuid UUID," +
+                ") " +
+                "ENGINE = MergeTree " +
+                "ORDER BY (longPrimitive); ";
+    }
+
     private int executeAsyncJob(StreamExecutionEnvironment env, String tableName, int numIterations, int expectedRows) throws Exception {
         JobClient jobClient = env.executeAsync("Read GZipped CSV with FileSource");
         int rows = 0;
@@ -197,47 +247,7 @@ void SimplePOJODataTest() throws Exception {
         String dropTable = String.format("DROP TABLE IF EXISTS `%s`.`%s`", getDatabase(), tableName);
         ClickHouseServerForTests.executeSql(dropTable);
         // create table
-        String tableSql = "CREATE TABLE `" + getDatabase() + "`.`" + tableName + "` (" +
-                "bytePrimitive Int8," +
-                "byteObject Int8," +
-                "shortPrimitive Int16," +
-                "shortObject Int16," +
-                "intPrimitive Int32," +
-                "integerObject Int32," +
-                "longPrimitive Int64," +
-                "longObject Int64," +
-                "bigInteger128 Int128," +
-                "bigInteger256 Int256," +
-                "uint8Primitive  UInt8," +
-                "uint8Object UInt8," +
-                "uint16Primitive  UInt16," +
-                "uint16Object UInt16," +
-                "uint32Primitive  UInt32," +
-                "uint32Object UInt32," +
-                "uint64Primitive  UInt64," +
-                "uint64Object UInt64," +
-                "uint128Object UInt128," +
-                "uint256Object UInt256," +
-                "decimal Decimal(10,5)," +
-                "decimal32 Decimal32(9)," +
-                "decimal64 Decimal64(18)," +
-                "decimal128 Decimal128(38)," +
-                "decimal256 Decimal256(76)," +
-                "floatPrimitive Float," +
-                "floatObject Float," +
-                "doublePrimitive Double," +
-                "doubleObject Double," +
-                "booleanPrimitive Boolean," +
-                "booleanObject Boolean," +
-                "str String," +
-                "fixedStr FixedString(10)," +
-                "v_date Date," +
-                "v_date32 Date32," +
-                "v_dateTime DateTime," +
-                "v_dateTime64 DateTime64," +
-                ") " +
-                "ENGINE = MergeTree " +
-                "ORDER BY (longPrimitive); ";
+        String tableSql = createSimplePOJOTableSQL(getDatabase(),  tableName);
         ClickHouseServerForTests.executeSql(tableSql);
 
 
@@ -485,48 +495,7 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
         String dropTable = String.format("DROP TABLE IF EXISTS `%s`.`%s`", getDatabase(), tableName);
         ClickHouseServerForTests.executeSql(dropTable);
         // create table
-        String tableSql = "CREATE TABLE `" + getDatabase() + "`.`" + tableName + "` (" +
-                "bytePrimitive Int8," +
-                "byteObject Int8," +
-                "shortPrimitive Int16," +
-                "shortObject Int16," +
-                "intPrimitive Int32," +
-                "integerObject Int32," +
-                "longPrimitive Int64," +
-                "longObject Int64," +
-                "bigInteger128 Int128," +
-                "bigInteger256 Int256," +
-                "uint8Primitive  UInt8," +
-                "uint8Object UInt8," +
-                "uint16Primitive  UInt16," +
-                "uint16Object UInt16," +
-                "uint32Primitive  UInt32," +
-                "uint32Object UInt32," +
-                "uint64Primitive  UInt64," +
-                "uint64Object UInt64," +
-                "uint128Object UInt128," +
-                "uint256Object UInt256," +
-                "decimal Decimal(10,5)," +
-                "decimal32 Decimal32(9)," +
-                "decimal64 Decimal64(18)," +
-                "decimal128 Decimal128(38)," +
-                "decimal256 Decimal256(76)," +
-                "floatPrimitive Float," +
-                "floatObject Float," +
-                "doublePrimitive Double," +
-                "doubleObject Double," +
-                "booleanPrimitive Boolean," +
-                "booleanObject Boolean," +
-                "str String," +
-                "fixedStr FixedString(10)," +
-                "v_date Date," +
-                "v_date32 Date32," +
-                "v_dateTime DateTime," +
-                "v_dateTime64 DateTime64," +
-                ") " +
-                "ENGINE = MergeTree " +
-                "ORDER BY (longPrimitive) " +
-                "SETTINGS parts_to_throw_insert = 10;";
+        String tableSql = createSimplePOJOTableSQL(getDatabase(),  tableName, 10);
         ClickHouseServerForTests.executeSql(tableSql);
         //ClickHouseServerForTests.executeSql(String.format("SYSTEM STOP MERGES `%s.%s`", getDatabase(), tableName));
 
diff --git a/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/convertor/SimplePOJOConvertor.java b/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/convertor/SimplePOJOConvertor.java
@@ -65,5 +65,6 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {
         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);
 
+        Serialize.writeUUID(out, input.getUuid(), false, false, ClickHouseDataType.UUID, false, "uuid");
     }
 }
diff --git a/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/pojo/SimplePOJO.java b/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/pojo/SimplePOJO.java
@@ -4,6 +4,7 @@
 import java.math.BigInteger;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.util.UUID;
 
 public class SimplePOJO {
 
@@ -61,6 +62,8 @@ public class SimplePOJO {
     private LocalDateTime dateTime;
     private LocalDateTime dateTime64;
 
+    private UUID uuid;
+
     public SimplePOJO(int index) {
         this.bytePrimitive = Byte.MIN_VALUE;
         this.byteObject = Byte.MAX_VALUE;
@@ -115,6 +118,7 @@ public SimplePOJO(int index) {
         this.dateTime = LocalDateTime.now();
         this.dateTime64 = LocalDateTime.now();
 
+        this.uuid = UUID.randomUUID();
     }
 
     public byte getBytePrimitive() {
@@ -215,4 +219,6 @@ public Double getDoubleObject() {
 
     public LocalDateTime getDateTime64() { return dateTime64; }
 
+    public UUID getUuid() { return uuid; }
+
 }
diff --git a/flink-connector-clickhouse-base/src/main/java/com/clickhouse/utils/Serialize.java b/flink-connector-clickhouse-base/src/main/java/com/clickhouse/utils/Serialize.java
@@ -18,6 +18,7 @@
 import java.time.ZonedDateTime;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 public class Serialize {
     private static final Logger LOG = LoggerFactory.getLogger(Serialize.class);
@@ -286,4 +287,11 @@ public static void writeBoolean(OutputStream out, Boolean value, boolean default
         }
     }
 
+    // UUID
+    public static void writeUUID(OutputStream out, UUID value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeUuid(out, value);
+        }
+    }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -65,5 +65,6 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {`
`65`	`65`	`Serialize.writeTimeDate(out, input.getDateTime(), false, false, ClickHouseDataType.DateTime, false, "v_dateTime");`
`66`	`66`	`Serialize.writeTimeDate64(out, input.getDateTime64(), false, false, ClickHouseDataType.DateTime64, false, "v_dateTime64", 1);`
`67`	`67`
	`68`	`+ Serialize.writeUUID(out, input.getUuid(), false, false, ClickHouseDataType.UUID, false, "uuid");`
`68`	`69`	`}`
`69`	`70`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`import java.time.ZonedDateTime;`
`19`	`19`	`import java.util.HashMap;`
`20`	`20`	`import java.util.Map;`
	`21`	`+import java.util.UUID;`
`21`	`22`
`22`	`23`	`public class Serialize {`
`23`	`24`	`private static final Logger LOG = LoggerFactory.getLogger(Serialize.class);`
`@@ -286,4 +287,11 @@ public static void writeBoolean(OutputStream out, Boolean value, boolean default`
`286`	`287`	`}`
`287`	`288`	`}`
`288`	`289`
	`290`	`+ // UUID`
	`291`	`+ public static void writeUUID(OutputStream out, UUID value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {`
	`292`	`+ if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {`
	`293`	`+ BinaryStreamUtils.writeUuid(out, value);`
	`294`	`+ }`
	`295`	`+ }`
	`296`	`+`
`289`	`297`	`}`