SinkWriter implementation of String data type

.github/workflows/tests-scala.yaml

@@ -0,0 +1,40 @@
+name: Apache Flink ClickHouse Connector Tests CI (Scala)
+
+on: [push]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        clickhouse: [ "23.7", "24.3", "latest", "cloud" ]
+    name: Apache Flink ClickHouse Connector tests with ClickHouse ${{ matrix.clickhouse }}
+    steps:
+      - name: Check for Cloud Credentials
+        id: check-cloud-credentials
+        run: |
+          if [[ "${{ matrix.clickhouse }}" == "cloud" && (-z "${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_HOST_SMT }}" || -z "${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_PASSWORD_SMT }}") ]]; then
+            echo "SKIP_STEP=true" >> $GITHUB_ENV
+          else
+            echo "SKIP_STEP=false" >> $GITHUB_ENV
+          fi
+        shell: bash
+      - uses: actions/checkout@v3
+        if: env.SKIP_STEP != 'true'
+      - name: Set up JDK 21
+        if: env.SKIP_STEP != 'true'
+        uses: actions/setup-java@v3
+        with:
+          java-version: '21'
+          distribution: 'adopt'
+          architecture: x64
+      - name: Setup and execute Gradle 'runScalaTests' task
+        if: env.SKIP_STEP != 'true'
+        uses: gradle/gradle-build-action@v2
+        env:
+          CLICKHOUSE_VERSION: ${{ matrix.clickhouse }}
+          CLICKHOUSE_CLOUD_HOST: ${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_HOST_SMT }}
+          CLICKHOUSE_CLOUD_PASSWORD: ${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_PASSWORD_SMT }}
+        with:
+          arguments: runScalaTests

.github/workflows/tests.yaml

@@ -1,4 +1,4 @@
-name: Apache Flink ClickHouse Connector Tests CI
+name: Apache Flink ClickHouse Connector Tests CI (Java)
 
 on: [push]
 
@@ -22,7 +22,7 @@ jobs:
         shell: bash
       - uses: actions/checkout@v3
         if: env.SKIP_STEP != 'true'
-      - name: Set up JDK 17
+      - name: Set up JDK 21
         if: env.SKIP_STEP != 'true'
         uses: actions/setup-java@v3
         with:

flink-connector-clickhouse-base/build.gradle.kts

@@ -4,42 +4,46 @@
  */
 
 plugins {
-    kotlin("jvm") version "1.9.0"
+    scala
     java
 }
 
+val scalaVersion = "2.13.12"
+
 repositories {
     // Use Maven Central for resolving dependencies.
+    mavenLocal()
     mavenCentral()
 }
 
 extra.apply {
-    set("clickHouseDriverVersion", "0.8.5")
+    set("clickHouseDriverVersion", "0.8.6")
     set("flinkVersion", "2.0.0")
     set("log4jVersion","2.17.2")
     set("testContainersVersion", "1.21.0")
+    set("byteBuddyVersion", "1.17.5")
 }
 
-
 dependencies {
     // Use JUnit Jupiter for testing.
     testImplementation(libs.junit.jupiter)
 
     testRuntimeOnly("org.junit.platform:junit-platform-launcher")
 
+    implementation("net.bytebuddy:byte-buddy:${project.extra["byteBuddyVersion"]}")
+    implementation("net.bytebuddy:byte-buddy-agent:${project.extra["byteBuddyVersion"]}")
     // This dependency is used by the application.
     implementation(libs.guava)
+    implementation("org.scala-lang:scala-library:$scalaVersion")
+    implementation("org.scala-lang:scala-compiler:$scalaVersion")
     // logger
     implementation("org.apache.logging.log4j:log4j-slf4j-impl:${project.extra["log4jVersion"]}")
     implementation("org.apache.logging.log4j:log4j-api:${project.extra["log4jVersion"]}")
     implementation("org.apache.logging.log4j:log4j-1.2-api:${project.extra["log4jVersion"]}")
     implementation("org.apache.logging.log4j:log4j-core:${project.extra["log4jVersion"]}")
 
     // ClickHouse Client Libraries
-    implementation("com.clickhouse:clickhouse-client:${project.extra["clickHouseDriverVersion"]}")
-    implementation("com.clickhouse:clickhouse-http-client:${project.extra["clickHouseDriverVersion"]}")
-    implementation("com.clickhouse:clickhouse-data:${project.extra["clickHouseDriverVersion"]}")
-    implementation("com.clickhouse:client-v2:${project.extra["clickHouseDriverVersion"]}")
+    implementation("com.clickhouse:client-v2:${project.extra["clickHouseDriverVersion"]}:all")
     // Apache Flink Libraries
     implementation("org.apache.flink:flink-connector-base:${project.extra["flinkVersion"]}")
     implementation("org.apache.flink:flink-streaming-java:${project.extra["flinkVersion"]}")
@@ -60,6 +64,28 @@ dependencies {
     //
     testImplementation("org.testcontainers:testcontainers:${project.extra["testContainersVersion"]}")
     testImplementation("org.testcontainers:clickhouse:${project.extra["testContainersVersion"]}")
+    testImplementation("org.scalatest:scalatest_2.13:3.2.19")
+    testRuntimeOnly("org.scalatestplus:junit-4-13_2.13:3.2.18.0")
+//    testRuntimeOnly("org.pegdown:pegdown:1.6.0") // sometimes required by ScalaTest
+}
+
+sourceSets {
+    main {
+        scala {
+            srcDirs("src/main/scala")
+        }
+        java {
+            srcDirs("src/main/java")
+        }
+    }
+    test {
+        scala {
+            srcDirs("src/test/scala")
+        }
+        java {
+            srcDirs("src/test/java")
+        }
+    }
 }
 
 // Apply a specific Java toolchain to ease working on different environments.
@@ -69,7 +95,36 @@ java {
     }
 }
 
+tasks.test {
+    useJUnitPlatform()
+
+    include("**/*Test.class", "**/*Tests.class", "**/*Spec.class")
+    testLogging {
+        events("passed", "failed", "skipped")
+        //showStandardStreams = true - , "standardOut", "standardError"
+    }
+}
+
+tasks.withType<ScalaCompile> {
+    scalaCompileOptions.apply {
+        encoding = "UTF-8"
+        isDeprecation = true
+        additionalParameters = listOf("-feature", "-unchecked")
+    }
+}
+
 tasks.named<Test>("test") {
     // Use JUnit Platform for unit tests.
     useJUnitPlatform()
 }
+
+tasks.register<JavaExec>("runScalaTests") {
+    group = "verification"
+    mainClass.set("org.scalatest.tools.Runner")
+    classpath = sourceSets["test"].runtimeClasspath
+    args = listOf(
+        "-R", "build/classes/scala/test",
+        "-oD", // show durations
+        "-s", "org.apache.flink.connector.clickhouse.test.scala.ClickHouseSinkTests"
+    )
+}

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

@@ -0,0 +1,183 @@
+package com.clickhouse.utils;
+
+import com.clickhouse.client.api.data_formats.internal.SerializerUtils;
+import com.clickhouse.data.ClickHouseDataType;
+import com.clickhouse.data.format.BinaryStreamUtils;
+import org.jline.utils.Log;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.reflect.Method;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.Map;
+
+public class Serialize {
+
+    public static boolean writePrimitiveValuePreamble(OutputStream out, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        // since it is primitive we always have a value that is not null
+        if (defaultsSupport) {
+            // Add indicator since the table has default values
+            SerializerUtils.writeNonNull(out);
+        }
+        // if the column is Nullable need to add an indicator for nullable
+        if (isNullable) {
+            SerializerUtils.writeNonNull(out);
+        }
+        return true;
+    }
+    public static boolean writeValuePreamble(OutputStream out, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column, Object value) throws IOException {
+        Log.debug("writeValuePreamble[defaultsSupport='%s', isNullable='%s', dataType='%s', column='%s', value='%s']");
+        if (defaultsSupport) {
+            if (value != null) {
+                SerializerUtils.writeNonNull(out);
+                if (isNullable) {
+                    SerializerUtils.writeNonNull(out);
+                }
+            } else {
+                if (hasDefault) {
+                    SerializerUtils.writeNull(out);
+                    return false;
+                }
+
+                if (isNullable) {
+                    SerializerUtils.writeNonNull(out);
+                    SerializerUtils.writeNull(out);
+                    return false;
+                }
+
+                if (dataType == ClickHouseDataType.Array) {
+                    SerializerUtils.writeNonNull(out);
+                } else if (dataType != ClickHouseDataType.Dynamic) {
+                    throw new IllegalArgumentException(String.format("An attempt to write null into not nullable column '%s' of type '%s'", column, dataType));
+                }
+            }
+        } else if (isNullable) {
+            if (value == null) {
+                SerializerUtils.writeNull(out);
+                return false;
+            }
+
+            SerializerUtils.writeNonNull(out);
+        } else if (value == null) {
+            if (dataType == ClickHouseDataType.Array) {
+                SerializerUtils.writeNonNull(out);
+            } else if (dataType != ClickHouseDataType.Dynamic) {
+                throw new IllegalArgumentException(String.format("An attempt to write null into not nullable column '%s' of type '%s'", column, dataType));
+            }
+        }
+        return true;
+    }
+
+    public static String convertToString(Object value) {
+        return java.lang.String.valueOf(value);
+    }
+
+    public static Integer convertToInteger(Object value) {
+        if (value instanceof Integer) {
+            return (Integer) value;
+        } else if (value instanceof Number) {
+            return ((Number) value).intValue();
+        } else if (value instanceof String) {
+            return Integer.parseInt((String) value);
+        } else if (value instanceof Boolean) {
+            return ((Boolean) value) ? 1 : 0;
+        } else {
+            throw new IllegalArgumentException("Cannot convert object of type " +
+                    value.getClass().getName() + " to Integer: " + value);
+        }
+    }
+
+    public static Map<ClickHouseDataType, Method> mapClickHouseTypeToMethod() {
+        Map<ClickHouseDataType, Method> map = new HashMap<>();
+        for (Method method : Serialize.class.getMethods()) {
+            String name = method.getName();
+            if (name.startsWith("write")) {
+                String chType = name.substring("write".length());
+                try {
+                    ClickHouseDataType type = ClickHouseDataType.valueOf(chType);
+                    map.put(type, method);
+                } catch (IllegalArgumentException e) {
+                    System.out.println(e.getMessage());
+                }
+            }
+        }
+        return map;
+    }
+
+    /**
+     *
+     */
+
+    // Method structure write[ClickHouse Type](OutputStream, Java type, ... )
+    // Date support
+    public static void writeDate(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.writeDate(out, value, ZoneId.of("UTC")); // TODO: check
+        }
+    }
+
+    public static void writeDate(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.writeDate(out, value, 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)) {
+            BinaryStreamUtils.writeString(out, convertToString(value));
+        }
+    }
+
+    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);
+        }
+    }
+
+    // Int8
+    public static void writeInt8(OutputStream out, Byte value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeInt8(out, convertToInteger(value));
+        }
+    }
+
+    // Int16
+    public static void writeInt16(OutputStream out, Short value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeInt16(out, convertToInteger(value));
+        }
+    }
+
+    // Int32
+    public static void writeInt32(OutputStream out, Integer value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeInt32(out, convertToInteger(value));
+        }
+    }
+
+    // Int64
+    public static void writeInt64(OutputStream out, Long value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeInt64(out, convertToInteger(value));
+        }
+    }
+
+    // Float32
+    public static void writeFloat32(OutputStream out, Float value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeFloat32(out, value);
+        }
+    }
+
+    // Float64
+    public static void writeFloat64(OutputStream out, Double value, boolean defaultsSupport, boolean isNullable, ClickHouseDataType dataType, boolean hasDefault, String column) throws IOException {
+        if (writeValuePreamble(out, defaultsSupport, isNullable, dataType, hasDefault, column, value)) {
+            BinaryStreamUtils.writeFloat64(out,  value);
+        }
+    }
+
+}