Add support for FixedString

Author: mzitnik

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

@@ -215,6 +215,7 @@ void SimplePOJODataTest() throws Exception {
                 "booleanPrimitive Boolean," +
                 "booleanObject Boolean," +
                 "str String," +
+                "fixedStr FixedString(10)" +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive); ";
@@ -483,6 +484,7 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
                 "booleanPrimitive Boolean," +
                 "booleanObject Boolean," +
                 "str String," +
+                "fixedStr FixedString(10)" +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive) " +

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

@@ -35,7 +35,8 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {
         Serialize.writeBoolean(out, input.isBooleanPrimitive(), false, false, ClickHouseDataType.Bool, false, "booleanPrimitive");
         Serialize.writeBoolean(out, input.getBooleanObject(), false, false, ClickHouseDataType.Bool, false, "booleanObject");
 
-        Serialize.writeString(out, input.getStr(), false, false, ClickHouseDataType.String, false, "String");
+        Serialize.writeString(out, input.getStr(), false, false, ClickHouseDataType.String, false, "str");
+        Serialize.writeFixedString(out, input.getFixedStr(), false, false, ClickHouseDataType.FixedString, false, 10, "fixedStr");
 
     }
 }

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

@@ -27,6 +27,8 @@ public class SimplePOJO {
 
     private String str;
 
+    private String fixedStr;
+
     private BigInteger bigInteger128;
     private BigInteger bigInteger256;
 
@@ -53,9 +55,12 @@ public SimplePOJO(int index) {
         this.booleanObject = Boolean.FALSE;
 
         this.str = "str" + longPrimitive;
+        this.fixedStr = (str + "_FixedString").substring(0, 10);
 
         this.bigInteger128 = BigInteger.valueOf(longPrimitive);
         this.bigInteger256 = BigInteger.valueOf(longPrimitive);
+
+
     }
 
     public byte getBytePrimitive() {
@@ -115,4 +120,6 @@ public Double getDoubleObject() {
     public BigInteger getBigInteger128() { return bigInteger128; }
 
     public BigInteger getBigInteger256() { return bigInteger256; }
+
+    public String getFixedStr() { return fixedStr; }
 }

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

@@ -135,6 +135,7 @@ public static void writeString(OutputStream out, String value, boolean defaultsS
         }
     }
 
+    // Add a boundary check before inserting
     public static void writeFixedString(OutputStream out, String value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, int size, String column) throws IOException {
         if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
             BinaryStreamUtils.writeFixedString(out, convertToString(value), size);