Altinity · subkanthi · Jan 11, 2025 · Jan 12, 2025 · Jan 16, 2025 · Mar 13, 2025
diff --git a/sink-connector-lightweight/pom.xml b/sink-connector-lightweight/pom.xml
@@ -209,7 +209,7 @@
         <dependency>
             <groupId>org.yaml</groupId>
             <artifactId>snakeyaml</artifactId>
-            <version>1.33</version>
+            <version>2.0</version>
         </dependency>
         <!-- VERSION COMPARE LIBRARY -->
         <dependency>
@@ -427,6 +427,11 @@
             <artifactId>jackson-databind</artifactId>
             <version>2.12.6</version>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>2.12.6</version>
+        </dependency>
 
         <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
         <dependency>

diff --git a/...java/com/altinity/clickhouse/debezium/embedded/ClickHouseDebeziumEmbeddedApplication.java b/...java/com/altinity/clickhouse/debezium/embedded/ClickHouseDebeziumEmbeddedApplication.java
@@ -7,7 +7,6 @@
 import com.altinity.clickhouse.debezium.embedded.common.PropertiesHelper;
 import com.altinity.clickhouse.debezium.embedded.config.ConfigLoader;
 import com.altinity.clickhouse.debezium.embedded.config.ConfigurationService;
-import com.altinity.clickhouse.debezium.embedded.ddl.parser.DDLParserService;
 import com.altinity.clickhouse.debezium.embedded.parser.DebeziumRecordParserService;
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig;
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;

diff --git a/...c/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java b/...c/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java
@@ -395,7 +395,7 @@ private Connection setSystemDbConnection(DBCredentials dbCredentials, ClickHouse
 
         Connection conn = BaseDbWriter.createConnection(jdbcUrl, BaseDbWriter.DATABASE_CLIENT_NAME,
                 dbCredentials.getUserName(), dbCredentials.getPassword(), BaseDbWriter.SYSTEM_DB, config);
-                writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
+        writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                 BaseDbWriter.SYSTEM_DB, dbCredentials.getUserName(),
                 dbCredentials.getPassword(), config, conn);
         return conn;
@@ -449,10 +449,10 @@ private void updateMetrics(String DDL) {
      * @param record ChangeEvent Record
      */
     private ClickHouseStruct processEveryChangeRecord(Properties props, ChangeEvent<SourceRecord, SourceRecord> record,
-                                          DebeziumRecordParserService debeziumRecordParserService,
-                                          ClickHouseSinkConnectorConfig config,
-                                          DebeziumEngine.RecordCommitter<ChangeEvent<SourceRecord, SourceRecord>>
-                                                  recordCommitter, boolean lastRecordInBatch) {
+                                                      DebeziumRecordParserService debeziumRecordParserService,
+                                                      ClickHouseSinkConnectorConfig config,
+                                                      DebeziumEngine.RecordCommitter<ChangeEvent<SourceRecord, SourceRecord>>
+                                                              recordCommitter, boolean lastRecordInBatch) {
         ClickHouseStruct chStruct = null;
 
         try {
@@ -602,13 +602,13 @@ private void setupProcessingThread(ClickHouseSinkConnectorConfig config) {
             singleThreadedWriter = new ClickHouseBatchWriter(config, new HashMap());
         }
 
-            ThreadFactory namedThreadFactory =
-                    new ThreadFactoryBuilder().setNameFormat("Sink Connector thread-pool-%d").build();
-            this.executor = new ClickHouseBatchExecutor(config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()), namedThreadFactory);
-            for (int i = 0; i < config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()); i++) {
-                this.executor.scheduleAtFixedRate(new ClickHouseBatchRunnable(this.records, config, new HashMap()), 0,
-                        config.getLong(ClickHouseSinkConnectorConfigVariables.BUFFER_FLUSH_TIME.toString()), TimeUnit.MILLISECONDS);
-            }
+        ThreadFactory namedThreadFactory =
+                new ThreadFactoryBuilder().setNameFormat("Sink Connector thread-pool-%d").build();
+        this.executor = new ClickHouseBatchExecutor(config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()), namedThreadFactory);
+        for (int i = 0; i < config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()); i++) {
+            this.executor.scheduleAtFixedRate(new ClickHouseBatchRunnable(this.records, config, new HashMap()), 0,
+                    config.getLong(ClickHouseSinkConnectorConfigVariables.BUFFER_FLUSH_TIME.toString()), TimeUnit.MILLISECONDS);
+        }
 
         //this.executor.scheduleAtFixedRate(this.runnable, 0, config.getLong(ClickHouseSinkConnectorConfigVariables.BUFFER_FLUSH_TIME.toString()), TimeUnit.MILLISECONDS);
     }

diff --git a/.../src/main/java/com/altinity/clickhouse/debezium/embedded/config/ColumnOverrideParser.java b/.../src/main/java/com/altinity/clickhouse/debezium/embedded/config/ColumnOverrideParser.java
@@ -0,0 +1,57 @@
+package com.altinity.clickhouse.debezium.embedded.config;
+
+import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
+import org.yaml.snakeyaml.Yaml;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.LinkedHashMap;
+import com.clickhouse.data.ClickHouseDataType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class ColumnOverrideParser {
+
+    private static final Logger log = LogManager.getLogger(ColumnOverrideParser.class);
+    public static Map<String, String> parseColumnOverrides(String yamlFile) throws FileNotFoundException {
+
+
+        Yaml yaml = new Yaml();
+        FileInputStream inputStream = new FileInputStream(yamlFile);
+
+        Map<String, Object> data = yaml.load(inputStream);
+
+        Object result = data.get(ClickHouseSinkConnectorConfigVariables.DEFAULT_COLUMN_DATATYPE_MAPPING.toString());
+
+
+        // if result is instance of LinkedHashMap , then cast it to LinkedHashMap
+        if (result instanceof LinkedHashMap) {
+            result = (LinkedHashMap<String, String>) result;
+        }
+        else {
+            return new HashMap<>();
+        }
+        // Iterate through the map and convert values to ClickHouse data types
+        Map<String, String> columnOverrides = new HashMap<>();
+        for (Map.Entry<String, String> entry : ((Map<String, String>) result).entrySet()) {
+            String key = entry.getKey();
+            Object value = entry.getValue();
+
+            // Match to ClickHouseDataType
+            ClickHouseDataType clickHouseDataType = ClickHouseDataType.valueOf(value.toString());   
+
+            // if clickhouseDataType is null, then log an error.
+            if(clickHouseDataType == null) {
+                log.error("*********** Invalid ClickHouse data type passed by user in yaml file for column override:******** " + value.toString());
+            }
+            columnOverrides.put(key, clickHouseDataType.toString());
+        }
+
+        return columnOverrides;
+    }
+
+
+
+}
diff --git a/...htweight/src/main/java/com/altinity/clickhouse/debezium/embedded/config/ConfigLoader.java b/...htweight/src/main/java/com/altinity/clickhouse/debezium/embedded/config/ConfigLoader.java
@@ -1,15 +1,20 @@
 package com.altinity.clickhouse.debezium.embedded.config;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.yaml.snakeyaml.Yaml;
 
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.InputStream;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Properties;
 
 public class ConfigLoader {
 
+    private static final Logger log = LogManager.getLogger(ConfigLoader.class);
+
     public Properties load(String resourceFileName) {
         InputStream fis = this.getClass()
                 .getClassLoader()
@@ -24,8 +29,27 @@ public Properties load(String resourceFileName) {
             if(entry.getValue() instanceof Integer) {
                 props.setProperty(entry.getKey(), Integer.toString((Integer) entry.getValue()));
             } else {
-                String value = (String) entry.getValue();
-                props.setProperty(entry.getKey(), value.replace("\"", ""));
+                Object entryValue = entry.getValue();
+                // Check if value is an instance of String.
+                if (entryValue instanceof String) {
+                    entryValue = (String) entryValue;
+                }
+                else {
+                    // Additional
+                    log.info("entryValue is not a String");
+                    if (entryValue instanceof LinkedHashMap) {
+                        // iterate through the map and add the properties to the props.
+                        for (Map.Entry<String, Object> mapEntry : ((LinkedHashMap<String, Object>) entryValue).entrySet()) {
+                            // prfix the key with the entry key.
+                            String key = entry.getKey() + "." + mapEntry.getKey();
+                            props.setProperty(key, mapEntry.getValue().toString());
+                        }
+                    }
+
+                }
+                if (entryValue instanceof String) {
+                    props.setProperty(entry.getKey(), ((String) entryValue).replace("\"", ""));
+                }
             }
         }
 

diff --git a/...ght/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverter.java b/...ght/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverter.java
@@ -16,7 +16,6 @@
 import io.debezium.jdbc.JdbcValueConverters;
 import io.debezium.jdbc.TemporalPrecisionMode;
 import io.debezium.relational.Column;
-import io.debezium.relational.RelationalDatabaseConnectorConfig;
 import io.debezium.relational.ddl.DataType;
 import io.debezium.service.DefaultServiceRegistry;
 import io.debezium.service.spi.ServiceRegistry;
@@ -25,7 +24,6 @@
 import java.sql.Types;
 import java.time.ZoneId;
 import java.util.Arrays;
-import java.util.Map;
 
 /**
  *

diff --git a/.../test/java/com/altinity/clickhouse/debezium/embedded/config/ColumnOverrideParserTest.java b/.../test/java/com/altinity/clickhouse/debezium/embedded/config/ColumnOverrideParserTest.java
@@ -0,0 +1,21 @@
+package com.altinity.clickhouse.debezium.embedded.config;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Assertions;
+
+import java.util.Map;
+import java.io.FileNotFoundException;   
+
+public class ColumnOverrideParserTest {
+
+    @Test
+    public void testParseColumnOverrides() {
+        String yamlFile = "src/test/resources/config.yml";
+        try {
+            Map<String, String> result = ColumnOverrideParser.parseColumnOverrides(yamlFile);
+            Assertions.assertEquals(result.size(), 7);
+        } catch (FileNotFoundException e) {
+            e.printStackTrace();
+        }
+    }       
+}
diff --git a/...ight/src/test/java/com/altinity/clickhouse/debezium/embedded/config/ConfigLoaderTest.java b/...ight/src/test/java/com/altinity/clickhouse/debezium/embedded/config/ConfigLoaderTest.java
@@ -4,6 +4,8 @@
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
+import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
+
 import java.util.Properties;
 
 public class ConfigLoaderTest {
@@ -16,4 +18,24 @@ public void testLoad() {
 
         Assertions.assertNotNull(props);
     }
+
+
+    @Test
+    @DisplayName("Unit test to validate loading of nested entries in config.yml")
+    public void testLoadNestedEntries() {
+        ConfigLoader loader = new ConfigLoader();
+        Properties props = loader.load("config.yml");
+
+        int defaultColumnDataTypeMappingCount = 0;
+        // iterate through the properties and check if the nested entries are loaded correctly
+        // the nested entries have the prefix ClickHouseSinkConnectorConfigVariables.DEFAULT_COLUMN_DATATYPE_MAPPING
+        for (Object key : props.keySet()) {
+            if (key.toString().startsWith(ClickHouseSinkConnectorConfigVariables.DEFAULT_COLUMN_DATATYPE_MAPPING.toString())) {
+                Assertions.assertNotNull(props.getProperty(key.toString()));
+                defaultColumnDataTypeMappingCount++;
+            }
+        }
+
+        Assertions.assertEquals(defaultColumnDataTypeMappingCount, 7);
+    }
 }
diff --git a/sink-connector-lightweight/src/test/resources/config.yml b/sink-connector-lightweight/src/test/resources/config.yml
@@ -30,4 +30,20 @@ schema.history.internal.jdbc.schema.history.table.name: "altinity_sink_connector
 enable.snapshot.ddl: "true"
 auto.create.tables: "true"
 metrics.enable: "false"
-database.connectionTimeZone: "America/Chicago"
+database.connectionTimeZone: "America/Chicago"
+default_column_datatype_mapping:
+  # we are no longer turning Date/DateTime/Timestamp as a String
+  transaction_id: String
+  exchange_transaction_id: String
+  unique_transaction_id: String
+  account_ref: String
+  otm_identifier: String
+  tag_reserved_4: String
+  initiator: String
+databases:
+  dbo:
+    tables:
+      tr_live:
+        partition_by: tr_date_id
+        primary_key: gmt_time
+        settings: allow_nullable_key=1
diff --git a/sink-connector/pom.xml b/sink-connector/pom.xml
@@ -442,6 +442,31 @@
             <version>1.18.2</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.83</version>
+        </dependency>
+
+        <!-- Jackson Core -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>2.12.6</version>
+        </dependency>
+        <!-- Jackson Databind -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.12.6</version>
+        </dependency>
+        <!-- Jackson YAML Dataformat -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>2.12.6</version>
+        </dependency>
+
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>

diff --git a/...n/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java b/...n/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java
@@ -3,6 +3,7 @@
 
 public enum ClickHouseSinkConnectorConfigVariables {
 
+    DEFAULT_COLUMN_DATATYPE_MAPPING("default_column_datatype_mapping"),
     IGNORE_DELETE("ignore_delete"),
     THREAD_POOL_SIZE("thread.pool.size"),
     BUFFER_COUNT("buffer.count"),