Adding String & Boolean support

Author: mzitnik

README.md

@@ -117,14 +117,16 @@ data.sinkTo(csvSink);
 
 ## Supported ClickHouse Types
 
-| Java Type     | ClickHouse Type | Supported |
-|---------------|-----------------|-----------|
-| byte/Byte     | Int8            | ✅         |
-| short/Short   | Int16           | ✅         |
-| int/Integer   | Int32           | ✅         |
-| long/Long     | Int64           | ✅         |
-| float/Float   | Float           | ✅         |
-| double/Double | Double          | ✅         |
+| Java Type       | ClickHouse Type | Supported |
+|-----------------|-----------------|-----------|
+| byte/Byte       | Int8            | ✅         |
+| short/Short     | Int16           | ✅         |
+| int/Integer     | Int32           | ✅         |
+| long/Long       | Int64           | ✅         |
+| float/Float     | Float           | ✅         |
+| double/Double   | Double          | ✅         |
+| boolean/Boolean | Boolean         | ✅         |
+| String          | String          | ✅         |
 
 ## Configuration Options
 

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

@@ -210,6 +210,9 @@ void SimplePOJODataTest() throws Exception {
                 "floatObject Float," +
                 "doublePrimitive Double," +
                 "doubleObject Double," +
+                "booleanPrimitive Boolean," +
+                "booleanObject Boolean," +
+                "str String," +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive); ";
@@ -468,6 +471,9 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
                 "floatObject Float," +
                 "doublePrimitive Double," +
                 "doubleObject Double," +
+                "booleanPrimitive Boolean," +
+                "booleanObject Boolean," +
+                "str String," +
                 ") " +
                 "ENGINE = MergeTree " +
                 "ORDER BY (longPrimitive) " +

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

@@ -28,5 +28,11 @@ public void instrument(OutputStream out, SimplePOJO input) throws IOException {
 
         Serialize.writeFloat64(out, input.getDoublePrimitive(), false, false, ClickHouseDataType.Float64, false, "doublePrimitive");
         Serialize.writeFloat64(out, input.getDoubleObject(), false, false, ClickHouseDataType.Float64, false, "doubleObject");
+
+        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");
+
     }
 }

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

@@ -20,6 +20,11 @@ public class SimplePOJO {
     private double doublePrimitive;
     private Double doubleObject;
 
+    private boolean booleanPrimitive;
+    private Boolean booleanObject;
+
+    private String str;
+
     public SimplePOJO(int index) {
         this.bytePrimitive = Byte.MIN_VALUE;
         this.byteObject = Byte.MAX_VALUE;
@@ -38,6 +43,11 @@ public SimplePOJO(int index) {
 
         this.doublePrimitive = Double.MIN_VALUE;
         this.doubleObject = Double.MAX_VALUE;
+
+        this.booleanPrimitive = true;
+        this.booleanObject = Boolean.FALSE;
+
+        this.str = "str" + longPrimitive;
     }
 
     public byte getBytePrimitive() {
@@ -87,4 +97,10 @@ public double getDoublePrimitive() {
     public Double getDoubleObject() {
         return doubleObject;
     }
+
+    public boolean isBooleanPrimitive() { return booleanPrimitive; }
+
+    public Boolean getBooleanObject() { return booleanObject; }
+
+    public String getStr() { return str; }
 }

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

@@ -182,4 +182,11 @@ public static void writeFloat64(OutputStream out, Double value, boolean defaults
         }
     }
 
+    // Boolean
+    public static void writeBoolean(OutputStream out, Boolean value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeBoolean(out, value);
+        }
+    }
+
 }