diff --git a/client-adapter/clickhouse/pom.xml b/client-adapter/clickhouse/pom.xml
new file mode 100644
index 00000000..def41cce
--- /dev/null
+++ b/client-adapter/clickhouse/pom.xml
@@ -0,0 +1,84 @@
+
+
+
+ canal.client-adapter
+ com.alibaba.otter
+ 1.1.8-SNAPSHOT
+ ../pom.xml
+
+ 4.0.0
+ com.alibaba.otter
+ client-adapter.clickhouse
+ jar
+ canal client adapter clickhouse module for otter ${project.version}
+
+
+
+ com.alibaba.otter
+ client-adapter.common
+ ${project.version}
+ provided
+
+
+ ru.yandex.clickhouse
+ clickhouse-jdbc
+ test
+
+
+ junit
+ junit
+ test
+
+
+ org.yaml
+ snakeyaml
+ test
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+ 2.4
+
+
+ jar-with-dependencies
+
+
+
+
+ make-assembly
+ package
+
+ single
+
+
+
+
+
+ maven-antrun-plugin
+
+
+ package
+
+ run
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java
new file mode 100644
index 00000000..e4a57cbf
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java
@@ -0,0 +1,334 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse;
+
+import com.alibaba.druid.filter.stat.StatFilter;
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.util.JdbcUtils;
+import com.alibaba.otter.canal.client.adapter.OuterAdapter;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.ConfigLoader;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MirrorDbConfig;
+import com.alibaba.otter.canal.client.adapter.clickhouse.monitor.ClickHouseConfigMonitor;
+import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseBatchSyncService;
+import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseEtlService;
+import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseMirrorDbBatchSyncService;
+import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
+import com.alibaba.otter.canal.client.adapter.support.*;
+import org.apache.commons.lang.BooleanUtils;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * ClickHouse Adapter implementation
+ *
+ * @author: Xander
+ * @date: Created in 2023/11/10 1:13
+ * @email: zhrunxin33@gmail.com
+ */
+@SPI
+public class ClickHouseAdapter implements OuterAdapter {
+
+ private static final Logger logger = LoggerFactory.getLogger(ClickHouseAdapter.class);
+
+ private Map clickHouseMapping = new ConcurrentHashMap<>(); // Store the mapping of filename and configuration, load yml files below resource path
+
+ private Map> mappingConfigCache = new ConcurrentHashMap<>(); // Schema -> Table -> MappingConfig
+
+ private Map mirrorDbConfigCache = new ConcurrentHashMap<>(); // Mirror DB Configuration, don't need to load column mapping
+
+ private DruidDataSource dataSource;
+
+ private ClickHouseBatchSyncService clickHouseBatchSyncService;
+
+ private ClickHouseMirrorDbBatchSyncService clickHouseMirrorDbBatchSyncService;
+
+ private Properties envProperties;
+
+ private OuterAdapterConfig configuration; // Launch configuration
+
+ private ClickHouseConfigMonitor clickHouseConfigMonitor;
+
+ public Map getClickHouseMapping() {
+ return clickHouseMapping;
+ }
+
+ @Override
+ public void init(OuterAdapterConfig configuration, Properties envProperties) {
+ this.envProperties = envProperties;
+ this.configuration = configuration;
+ // Load DB type from adapter.launch/bootstrap.yml
+ Map properties = configuration.getProperties();
+ String dbType = JdbcUtils.getDbType(properties.get("jdbc.url"), null);
+ // 当.yml文件编码格式存在问题，此处clickhouse yml文件构建 可能会抛出异常
+ Map clickHouseMappingTmp = ConfigLoader.load(envProperties);
+ // 过滤不匹配的key的配置
+ clickHouseMappingTmp.forEach((key, config) -> {
+ addConfig(key, config);
+ });
+
+ if (clickHouseMapping.isEmpty()) {
+ throw new RuntimeException("No clickhouse adapter found for config key: " + configuration.getKey());
+ }
+
+ // 初始化连接池
+ dataSource = new DruidDataSource();
+ dataSource.setDriverClassName(properties.get("jdbc.driverClassName"));
+ dataSource.setUrl(properties.get("jdbc.url"));
+ dataSource.setUsername(properties.get("jdbc.username"));
+ dataSource.setPassword(properties.get("jdbc.password"));
+ dataSource.setInitialSize(1);
+ dataSource.setMinIdle(1);
+ dataSource.setMaxActive(30);
+ dataSource.setMaxWait(60000);
+ dataSource.setTimeBetweenEvictionRunsMillis(60000);
+ dataSource.setMinEvictableIdleTimeMillis(300000);
+ dataSource.setUseUnfairLock(true);
+ dataSource.setDefaultAutoCommit(false); // disable auto commit or disable Transactional
+ dataSource.setDbType(dbType);
+
+ if ("true".equals(properties.getOrDefault("druid.stat.enable", "true"))) {
+ StatFilter statFilter = new StatFilter();
+ statFilter.setSlowSqlMillis(Long.parseLong(properties.getOrDefault("druid.stat.slowSqlMillis", "1000")));
+ statFilter.setMergeSql(true);
+ statFilter.setLogSlowSql(true);
+ dataSource.setProxyFilters(Collections.singletonList(statFilter));
+ }
+
+ try {
+ dataSource.init();
+ } catch (SQLException e) {
+ logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e);
+ }
+
+ String threads = properties.get("threads");
+ String scheduleTime = properties.get("scheduleTime");
+ String batchSize = properties.get("batchSize");
+
+ boolean skipDupException = BooleanUtils.toBoolean(configuration.getProperties()
+ .getOrDefault("skipDupException", "true"));
+ clickHouseBatchSyncService = new ClickHouseBatchSyncService(dataSource,
+ threads != null ? Integer.valueOf(threads) : null,
+ batchSize != null ? Integer.valueOf(batchSize) : null,
+ scheduleTime != null ? Long.valueOf(scheduleTime) : null,
+ skipDupException);
+
+ clickHouseMirrorDbBatchSyncService = new ClickHouseMirrorDbBatchSyncService(mirrorDbConfigCache,
+ dataSource,
+ threads != null ? Integer.valueOf(threads) : null,
+ batchSize != null ? Integer.valueOf(batchSize) : null,
+ scheduleTime != null ? Long.valueOf(scheduleTime) : null,
+ clickHouseBatchSyncService.getColumnsTypeCache(),
+ skipDupException);
+
+ clickHouseConfigMonitor = new ClickHouseConfigMonitor();
+ clickHouseConfigMonitor.init(configuration.getKey(), this, envProperties);
+ }
+
+ /**
+ * Sync main entrance
+ *
+ * @param dmls 数据包
+ */
+ @Override
+ public void sync(List dmls) {
+ if (dmls == null || dmls.isEmpty()) {
+ return;
+ }
+ try {
+ // If mappingConfigCache(column mapping) is empty, that must be mirroring synchronize
+ if (!mappingConfigCache.isEmpty()) {
+ clickHouseBatchSyncService.sync(mappingConfigCache, dmls, envProperties);
+ }
+ clickHouseMirrorDbBatchSyncService.sync(dmls);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Destroy
+ */
+ @Override
+ public void destroy() {
+ if (clickHouseConfigMonitor != null) {
+ clickHouseConfigMonitor.destroy();
+ }
+
+ if (clickHouseBatchSyncService != null) {
+ clickHouseBatchSyncService.close();
+ }
+
+ if (dataSource != null) {
+ dataSource.close();
+ }
+ }
+
+
+ /**
+ * ETL方法
+ *
+ * @param task 任务名, 对应配置名
+ * @param params etl筛选条件
+ * @return ETL结果
+ */
+ @Override
+ public EtlResult etl(String task, List params) {
+ EtlResult etlResult = new EtlResult();
+ MappingConfig config = clickHouseMapping.get(task);
+ ClickHouseEtlService clickhouseEtlService = new ClickHouseEtlService(dataSource, config);
+ if (config != null) {
+ return clickhouseEtlService.importData(params);
+ } else {
+ StringBuilder resultMsg = new StringBuilder();
+ boolean resSucc = true;
+ for (MappingConfig configTmp : clickHouseMapping.values()) {
+ // 取所有的destination为task的配置
+ if (configTmp.getDestination().equals(task)) {
+ EtlResult etlRes = clickhouseEtlService.importData(params);
+ if (!etlRes.getSucceeded()) {
+ resSucc = false;
+ resultMsg.append(etlRes.getErrorMessage()).append("\n");
+ } else {
+ resultMsg.append(etlRes.getResultMessage()).append("\n");
+ }
+ }
+ }
+ if (resultMsg.length() > 0) {
+ etlResult.setSucceeded(resSucc);
+ if (resSucc) {
+ etlResult.setResultMessage(resultMsg.toString());
+ } else {
+ etlResult.setErrorMessage(resultMsg.toString());
+ }
+ return etlResult;
+ }
+ }
+ etlResult.setSucceeded(false);
+ etlResult.setErrorMessage("Task not found");
+ return etlResult;
+ }
+
+ /**
+ * 获取总数方法
+ *
+ * @param task 任务名, 对应配置名
+ * @return 总数
+ */
+ @Override
+ public Map count(String task) {
+ MappingConfig config = clickHouseMapping.get(task);
+ MappingConfig.DbMapping dbMapping = config.getDbMapping();
+ String sql = "SELECT COUNT(1) AS cnt FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
+ Connection conn = null;
+ Map res = new LinkedHashMap<>();
+ try {
+ conn = dataSource.getConnection();
+ Util.sqlRS(conn, sql, rs -> {
+ try {
+ if (rs.next()) {
+ Long rowCount = rs.getLong("cnt");
+ res.put("count", rowCount);
+ }
+ } catch (SQLException e) {
+ logger.error(e.getMessage(), e);
+ }
+ });
+ } catch (SQLException e) {
+ logger.error(e.getMessage(), e);
+ } finally {
+ if (conn != null) {
+ try {
+ conn.close();
+ } catch (SQLException e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ }
+ res.put("targetTable", SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()));
+
+ return res;
+ }
+
+ /**
+ * 获取对应canal instance name 或 mq topic
+ *
+ * @param task 任务名, 对应配置名
+ * @return destination
+ */
+ @Override
+ public String getDestination(String task) {
+ MappingConfig config = clickHouseMapping.get(task);
+ if (config != null) {
+ return config.getDestination();
+ }
+ return null;
+ }
+
+ private void addSyncConfigToCache(String configName, MappingConfig mappingConfig) {
+ if (!mappingConfig.getDbMapping().getMirrorDb()) {
+ String key;
+ if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+ key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-"
+ + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_"
+ + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable();
+ } else {
+ key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_"
+ + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable();
+ }
+ Map configMap = mappingConfigCache.computeIfAbsent(key,
+ k1 -> new ConcurrentHashMap<>());
+ configMap.put(configName, mappingConfig);
+ } else {
+ // mirrorDB
+ String key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
+ + mappingConfig.getDbMapping().getDatabase();
+ mirrorDbConfigCache.put(key, MirrorDbConfig.create(configName, mappingConfig));
+ }
+ }
+
+ public boolean addConfig(String fileName, MappingConfig config) {
+ if (match(config)) {
+ clickHouseMapping.put(fileName, config);
+ addSyncConfigToCache(fileName, config);
+ FileName2KeyMapping.register(getClass().getAnnotation(SPI.class).value(), fileName,
+ configuration.getKey());
+ return true;
+ }
+ return false;
+ }
+
+ public void updateConfig(String fileName, MappingConfig config) {
+ if (config.getOuterAdapterKey() != null && !config.getOuterAdapterKey()
+ .equals(configuration.getKey())) {
+ // 理论上不允许改这个 因为本身就是通过这个关联起Adapter和Config的
+ throw new RuntimeException("not allow to change outAdapterKey");
+ }
+ clickHouseMapping.put(fileName, config);
+ addSyncConfigToCache(fileName, config);
+ }
+
+ public void deleteConfig(String fileName) {
+ clickHouseMapping.remove(fileName);
+ for (Map configMap : mappingConfigCache.values()) {
+ if (configMap != null) {
+ configMap.remove(fileName);
+ }
+ }
+ FileName2KeyMapping.unregister(getClass().getAnnotation(SPI.class).value(), fileName);
+ }
+
+ private boolean match(MappingConfig config) {
+ boolean sameMatch = config.getOuterAdapterKey() != null && config.getOuterAdapterKey()
+ .equalsIgnoreCase(configuration.getKey());
+ boolean prefixMatch = config.getOuterAdapterKey() == null && configuration.getKey()
+ .startsWith(StringUtils
+ .join(new String[]{Util.AUTO_GENERATED_PREFIX, config.getDestination(),
+ config.getGroupId()}, '-'));
+ return sameMatch || prefixMatch;
+ }
+}
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java
new file mode 100644
index 00000000..15933b5d
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java
@@ -0,0 +1,49 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse.config;
+
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
+import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * CLICKHOUSE表映射配置加载器
+ *
+ * @author rewerma 2018-11-07 下午02:41:34
+ * @version 1.0.0
+ */
+public class ConfigLoader {
+
+ private static Logger logger = LoggerFactory.getLogger(ConfigLoader.class);
+
+ /**
+ * 加载CLICKHOUSE表映射配置
+ *
+ * @return 配置名/配置文件名--对象
+ */
+ public static Map load(Properties envProperties) {
+ logger.info("## Start loading clickhouse mapping config ... ");
+
+ Map result = new LinkedHashMap<>();
+
+ Map configContentMap = MappingConfigsLoader.loadConfigs("clickhouse");
+ configContentMap.forEach((fileName, content) -> {
+ MappingConfig config = YamlUtils.ymlToObj(null, content, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
+ try {
+ config.validate();
+ } catch (Exception e) {
+ throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e);
+ }
+ result.put(fileName, config);
+ });
+
+ logger.info("## ClickHouse mapping config loaded");
+ return result;
+ }
+}
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java
new file mode 100644
index 00000000..8d230160
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java
@@ -0,0 +1,224 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse.config;
+
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+import org.apache.commons.lang.StringUtils;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * CLICKHOUSE表映射配置
+ *
+ * @author rewerma 2018-11-07 下午02:41:34
+ * @version 1.0.0
+ */
+public class MappingConfig implements AdapterConfig {
+
+ private String dataSourceKey; // 数据源key
+
+ private String destination; // canal实例或MQ的topic
+
+ private String groupId; // groupId
+
+ private String outerAdapterKey; // 对应适配器的key
+
+ private boolean concurrent = false; // 是否并行同步
+
+ private DbMapping dbMapping; // db映射配置
+
+ public String getDataSourceKey() {
+ return dataSourceKey;
+ }
+
+ public void setDataSourceKey(String dataSourceKey) {
+ this.dataSourceKey = dataSourceKey;
+ }
+
+ public String getGroupId() {
+ return groupId;
+ }
+
+ public void setGroupId(String groupId) {
+ this.groupId = groupId;
+ }
+
+ public String getOuterAdapterKey() {
+ return outerAdapterKey;
+ }
+
+ public void setOuterAdapterKey(String outerAdapterKey) {
+ this.outerAdapterKey = outerAdapterKey;
+ }
+
+ public boolean getConcurrent() {
+ return concurrent;
+ }
+
+ public void setConcurrent(boolean concurrent) {
+ this.concurrent = concurrent;
+ }
+
+ public DbMapping getDbMapping() {
+ return dbMapping;
+ }
+
+ public void setDbMapping(DbMapping dbMapping) {
+ this.dbMapping = dbMapping;
+ }
+
+ public String getDestination() {
+ return destination;
+ }
+
+ public void setDestination(String destination) {
+ this.destination = destination;
+ }
+
+ public AdapterMapping getMapping() {
+ return dbMapping;
+ }
+
+ public void validate() {
+ if (dbMapping.database == null || dbMapping.database.isEmpty()) {
+ throw new NullPointerException("dbMapping.database");
+ }
+ if (!dbMapping.getMirrorDb() && (dbMapping.table == null || dbMapping.table.isEmpty())) {
+ throw new NullPointerException("dbMapping.table");
+ }
+ if (!dbMapping.getMirrorDb() && (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty())) {
+ throw new NullPointerException("dbMapping.targetTable");
+ }
+ }
+
+ public static class DbMapping implements AdapterMapping {
+
+ private boolean mirrorDb = false; // 是否镜像库
+ private String database; // 数据库名或schema名
+ private String table; // 表名
+ private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段
+ private boolean mapAll = false; // 映射所有字段
+ private String targetDb; // 目标库名
+ private String targetTable; // 目标表名
+ private Map targetColumns; // 目标表字段映射
+
+ private boolean caseInsensitive = false; // 目标表不区分大小写，默认是否
+
+ private String etlCondition; // etl条件sql
+
+ private int readBatch = 5000;
+ private int commitBatch = 5000; // etl等批量提交大小
+
+ private Map allMapColumns;
+
+ public boolean getMirrorDb() {
+ return mirrorDb;
+ }
+
+ public void setMirrorDb(boolean mirrorDb) {
+ this.mirrorDb = mirrorDb;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public void setDatabase(String database) {
+ this.database = database;
+ }
+
+ public String getTable() {
+ return table;
+ }
+
+ public void setTable(String table) {
+ this.table = table;
+ }
+
+ public Map getTargetPk() {
+ return targetPk;
+ }
+
+ public void setTargetPk(Map targetPk) {
+ this.targetPk = targetPk;
+ }
+
+ public Boolean getMapAll() {
+ return mapAll;
+ }
+
+ public void setMapAll(Boolean mapAll) {
+ this.mapAll = mapAll;
+ }
+
+ public String getTargetDb() {
+ return targetDb;
+ }
+
+ public void setTargetDb(String targetDb) {
+ this.targetDb = targetDb;
+ }
+
+ public String getTargetTable() {
+ return targetTable;
+ }
+
+ public void setTargetTable(String targetTable) {
+ this.targetTable = targetTable;
+ }
+
+ public Map getTargetColumns() {
+ if (targetColumns != null) {
+ targetColumns.forEach((key, value) -> {
+ if (StringUtils.isEmpty(value)) {
+ targetColumns.put(key, key);
+ }
+ });
+ }
+ return targetColumns;
+ }
+
+ public void setTargetColumns(Map targetColumns) {
+ this.targetColumns = targetColumns;
+ }
+
+ public boolean isCaseInsensitive() {
+ return caseInsensitive;
+ }
+
+ public void setCaseInsensitive(boolean caseInsensitive) {
+ this.caseInsensitive = caseInsensitive;
+ }
+
+ public String getEtlCondition() {
+ return etlCondition;
+ }
+
+ public void setEtlCondition(String etlCondition) {
+ this.etlCondition = etlCondition;
+ }
+
+ public int getReadBatch() {
+ return readBatch;
+ }
+
+ public void setReadBatch(int readBatch) {
+ this.readBatch = readBatch;
+ }
+
+ public int getCommitBatch() {
+ return commitBatch;
+ }
+
+ public void setCommitBatch(int commitBatch) {
+ this.commitBatch = commitBatch;
+ }
+
+ public Map getAllMapColumns() {
+ return allMapColumns;
+ }
+
+ public void setAllMapColumns(Map allMapColumns) {
+ this.allMapColumns = allMapColumns;
+ }
+ }
+}
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java
new file mode 100644
index 00000000..5f53b7d5
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java
@@ -0,0 +1,44 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse.config;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class MirrorDbConfig {
+
+ private String fileName;
+ private MappingConfig mappingConfig;
+ private Map tableConfig = new ConcurrentHashMap<>();
+
+ public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) {
+ return new MirrorDbConfig(fileName, mappingConfig);
+ }
+
+ public MirrorDbConfig(String fileName, MappingConfig mappingConfig){
+ this.fileName = fileName;
+ this.mappingConfig = mappingConfig;
+ }
+
+ public String getFileName() {
+ return fileName;
+ }
+
+ public void setFileName(String fileName) {
+ this.fileName = fileName;
+ }
+
+ public MappingConfig getMappingConfig() {
+ return mappingConfig;
+ }
+
+ public void setMappingConfig(MappingConfig mappingConfig) {
+ this.mappingConfig = mappingConfig;
+ }
+
+ public Map getTableConfig() {
+ return tableConfig;
+ }
+
+ public void setTableConfig(Map tableConfig) {
+ this.tableConfig = tableConfig;
+ }
+}
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java
new file mode 100644
index 00000000..ff048235
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java
@@ -0,0 +1,123 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse.monitor;
+
+import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
+import org.apache.commons.io.monitor.FileAlterationMonitor;
+import org.apache.commons.io.monitor.FileAlterationObserver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.Properties;
+
+public class ClickHouseConfigMonitor {
+
+ private static final Logger logger = LoggerFactory.getLogger(ClickHouseConfigMonitor.class);
+
+ private static final String adapterName = "clickhouse";
+
+ private String key;
+
+ private ClickHouseAdapter clickHouseAdapter;
+
+ private Properties envProperties;
+
+ private FileAlterationMonitor fileMonitor;
+
+ public void init(String key, ClickHouseAdapter clickHouseAdapter, Properties envProperties) {
+ this.key = key;
+ this.clickHouseAdapter = clickHouseAdapter;
+ this.envProperties = envProperties;
+ File confDir = Util.getConfDirPath(adapterName);
+ try {
+ FileAlterationObserver observer = new FileAlterationObserver(confDir,
+ FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
+ FileListener listener = new FileListener();
+ observer.addListener(listener);
+ fileMonitor = new FileAlterationMonitor(3000, observer);
+ fileMonitor.start();
+
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ public void destroy() {
+ try {
+ fileMonitor.stop();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ private class FileListener extends FileAlterationListenerAdaptor {
+
+ @Override
+ public void onFileCreate(File file) {
+ super.onFileCreate(file);
+ try {
+ // 加载新增的配置文件
+ String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
+ MappingConfig config = YamlUtils
+ .ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
+ config.validate();
+ boolean result = clickHouseAdapter.addConfig(file.getName(), config);
+ if (result) {
+ logger.info("Add a new clickhouse mapping config: {} to canal adapter", file.getName());
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void onFileChange(File file) {
+ super.onFileChange(file);
+
+ try {
+ if (clickHouseAdapter.getClickHouseMapping().containsKey(file.getName())) {
+ // 加载配置文件
+ String configContent = MappingConfigsLoader
+ .loadConfig(adapterName + File.separator + file.getName());
+ if (configContent == null) {
+ onFileDelete(file);
+ return;
+ }
+ MappingConfig config = YamlUtils
+ .ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
+ config.validate();
+ clickHouseAdapter.updateConfig(file.getName(), config);
+ logger.info("Change a clickhouse mapping config: {} of canal adapter", file.getName());
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void onFileDelete(File file) {
+ super.onFileDelete(file);
+
+ try {
+ if (clickHouseAdapter.getClickHouseMapping().containsKey(file.getName())) {
+ clickHouseAdapter.deleteConfig(file.getName());
+
+ logger.info("Delete a clickhouse mapping config: {} of canal adapter", file.getName());
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ }
+}
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java
new file mode 100644
index 00000000..6265d1bf
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java
@@ -0,0 +1,553 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse.service;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONWriter.Feature;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping;
+import com.alibaba.otter.canal.client.adapter.clickhouse.support.BatchExecutor;
+import com.alibaba.otter.canal.client.adapter.clickhouse.support.SingleDml;
+import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * ClickHouse batch synchronize
+ *
+ * @author: Xander
+ * @date: Created in 2023/11/10 22:23
+ * @email: zhrunxin33@gmail.com
+ */
+public class ClickHouseBatchSyncService {
+
+ private static final Logger logger = LoggerFactory.getLogger(ClickHouseBatchSyncService.class);
+
+ private DruidDataSource dataSource;
+
+ private Map> columnsTypeCache; // Cache of instance.schema.table ->
+
+ private Map>[] bufferPools; // Data buffer pool store sync data, List dispersed as arrays according to hash value
+
+ private BatchExecutor[] batchExecutors; // Batch Executor
+
+ private BatchExecutor alterExecutors; // Alter Single Executor(update/delete/truncate)
+
+ private ExecutorService[] executorThreads; // Be initialized once
+
+ private ScheduledExecutorService[] scheduledExecutors;
+
+ private int threads = 3; // Default parallel thread count
+ private int batchSize = 1000;
+ private long scheduleTime = 10;
+ private boolean skipDupException;
+
+ public Map> getColumnsTypeCache() {
+ return columnsTypeCache;
+ }
+
+ public ClickHouseBatchSyncService(DruidDataSource dataSource, Integer threads, Integer batchSize, Long scheduleTime, boolean skipDupException){
+ this(dataSource, threads, batchSize, scheduleTime, new ConcurrentHashMap<>(), skipDupException);
+ }
+
+ @SuppressWarnings("unchecked")
+ public ClickHouseBatchSyncService(DruidDataSource dataSource, Integer threads, Integer batchSize, Long scheduleTime, Map> columnsTypeCache,
+ boolean skipDupException){
+ this.dataSource = dataSource;
+ this.columnsTypeCache = columnsTypeCache;
+ this.skipDupException = skipDupException;
+ try {
+ if (threads != null) {
+ this.threads = threads;
+ }
+ if (batchSize != null) {
+ this.batchSize = batchSize;
+ }
+ if (scheduleTime != null) {
+ this.scheduleTime = scheduleTime;
+ }
+ this.alterExecutors = new BatchExecutor(dataSource);
+ this.bufferPools = new ConcurrentHashMap[this.threads];
+ this.batchExecutors = new BatchExecutor[this.threads];
+ this.executorThreads = new ExecutorService[this.threads];
+ this.scheduledExecutors = new ScheduledExecutorService[this.threads];
+ for (int i = 0; i < this.threads; i++) {
+ bufferPools[i] = new ConcurrentHashMap<>();
+ batchExecutors[i] = new BatchExecutor(dataSource);
+ executorThreads[i] = Executors.newSingleThreadExecutor();
+ scheduledExecutors[i] = Executors.newSingleThreadScheduledExecutor();
+ }
+ scheduleBatchSync();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Timing-driven event
+ * start schedule batch sync threadPool
+ */
+ private void scheduleBatchSync() {
+ for (int i = 0; i < scheduledExecutors.length; i++) {
+ int index = i;
+ scheduledExecutors[i].scheduleAtFixedRate(()->{
+ List> futures = new ArrayList<>();
+ for (MappingConfig mappingConfig : bufferPools[index].keySet()) {
+ List dmls = bufferPools[index].get(mappingConfig);
+ if (dmls == null || dmls.isEmpty()) {
+ return;
+ }
+ List tempDmls;
+ synchronized (dmls) {
+ tempDmls = new ArrayList<>(dmls);
+ dmls.clear();
+ }
+ futures.add(executorThreads[index].submit(()->{
+ try {
+ insert(batchExecutors[index], mappingConfig, tempDmls);
+ batchExecutors[index].commit();
+ return true;
+ } catch (Exception e){
+ batchExecutors[index].rollback();
+ throw new RuntimeException(e);
+ }
+ }));
+ }
+ }, 0, scheduleTime, TimeUnit.SECONDS);
+ }
+ logger.info("Schedule batch executors has started successfully!");
+ }
+
+ /**
+ * 批量同步回调
+ *
+ * @param dmls 批量 DML
+ * @param function 回调方法
+ */
+ public void sync(List dmls, Function function) {
+ boolean toExecute = false;
+ for (Dml dml : dmls) {
+ if (!toExecute) {
+ toExecute = function.apply(dml);
+ } else {
+ function.apply(dml);
+ }
+ }
+ }
+
+ /**
+ * Distribute dmls into different partition
+ *
+ * @param mappingConfig {@link com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter#mappingConfigCache }
+ * @param dmls received DML
+ */
+ public void sync(Map> mappingConfig, List dmls, Properties envProperties) {
+ sync(dmls, dml -> {
+ if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
+ // DDL(Cache need to update when DDL was executed)
+ columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
+ return false;
+ } else {
+ // DML
+ String destination = StringUtils.trimToEmpty(dml.getDestination());
+ String groupId = StringUtils.trimToEmpty(dml.getGroupId());
+ String database = dml.getDatabase();
+ String table = dml.getTable();
+ Map configMap;
+ if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+ configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
+ } else {
+ configMap = mappingConfig.get(destination + "_" + database + "-" + table);
+ }
+
+ if (configMap == null) {
+ return false;
+ }
+
+ if (configMap.values().isEmpty()) {
+ return false;
+ }
+
+ for (MappingConfig config : configMap.values()) {
+ distributeDml(config, dml);
+ }
+ return true;
+ }
+ } );
+ }
+
+ /**
+ * Dml distributor
+ */
+ private void distributeDml(MappingConfig config, Dml dml) {
+ if (config != null) {
+ try {
+ String type = dml.getType();
+ if (type == null) return;
+
+ if (type.equalsIgnoreCase("INSERT")) {
+ appendDmlBufferPartition(config, dml);
+ } else {
+ boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
+ List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
+
+ if (type.equalsIgnoreCase("UPDATE")) {
+ for (SingleDml singleDml : singleDmls) {
+ update(alterExecutors, config, singleDml);
+ }
+ } else if (type.equalsIgnoreCase("DELETE")) {
+ for (SingleDml singleDml : singleDmls) {
+ delete(alterExecutors, config, singleDml);
+ }
+ } else if (type.equalsIgnoreCase("TRUNCATE")) {
+ truncate(alterExecutors, config);
+ }
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public void appendDmlBufferPartition(MappingConfig config, Dml dml) {
+ boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
+ List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
+
+ singleDmls.forEach(singleDml -> {
+ int hash = mappingHash(config.getDbMapping());
+ if (!config.getConcurrent()) {
+ hash = 0;
+ }
+ List dmls = bufferPools[hash].computeIfAbsent(config, k -> new ArrayList<>());
+ synchronized (dmls) {
+ dmls.add(singleDml);
+ logger.info("Append one data into pool, id {}", singleDml.getData().get("id"));
+ }
+ // Check the size of the List, achieve when it reaches the maximum value
+ if (dmls.size() >= batchSize) syncToClickHouse(config, hash);
+ });
+ }
+
+ /**
+ * sync when size of list{@link #bufferPools } reaches the maximum value
+ *
+ * @param config key
+ * @param index parallel thread index
+ */
+ private void syncToClickHouse(MappingConfig config, int index) {
+ List dmls = bufferPools[index].get(config);
+ logger.info("schema:{} table:{} reaches the maximum value, ready to synchronize, size {}", config.getDbMapping().getDatabase(), config.getDbMapping().getTable(), dmls.size());
+ if (dmls ==null || dmls.isEmpty()) {
+ return;
+ }
+ List tempDmls;
+ synchronized (dmls) {
+ tempDmls = new ArrayList<>(dmls);
+ dmls.clear();
+ }
+ executorThreads[index].submit(() -> {
+ try {
+ insert(batchExecutors[index], config, tempDmls);
+ batchExecutors[index].commit();
+ return true;
+ } catch (Exception e) {
+ batchExecutors[index].rollback();
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ /**
+ * Insert
+ *
+ * @param batchExecutor batch translational executor
+ * @param config corresponding configuration object
+ * @param dmls DMLs
+ */
+ private void insert(BatchExecutor batchExecutor, MappingConfig config, List dmls) throws SQLException {
+ if (dmls == null || dmls.isEmpty()) {
+ return;
+ }
+ List clearDmls = dmls.stream().filter(e -> e.getData() != null && !e.getData().isEmpty()).collect(Collectors.toList());
+ if (clearDmls == null || clearDmls.isEmpty()) {
+ return;
+ }
+
+ DbMapping dbMapping = config.getDbMapping();
+ String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
+ Map columnsMap = SyncUtil.getColumnsMap(dbMapping, clearDmls.get(0).getData());
+
+ StringBuilder insertSql = new StringBuilder();
+ insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" (");
+
+ columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick)
+ .append(targetColumnName)
+ .append(backtick)
+ .append(","));
+ int len = insertSql.length();
+ insertSql.delete(len - 1, len).append(") VALUES (");
+ int mapLen = columnsMap.size();
+ for (int i = 0; i < mapLen; i++) {
+ insertSql.append("?,");
+ }
+ len = insertSql.length();
+ insertSql.delete(len - 1, len).append(")");
+
+ Map ctype = getTargetColumnType(batchExecutor.getConn(), config);
+
+ List>> values = new ArrayList<>();
+ boolean flag = false;
+ for (Map.Entry entry : columnsMap.entrySet()) {
+ String targetColumnName = entry.getKey();
+ String srcColumnName = entry.getValue();
+ if (srcColumnName == null) {
+ srcColumnName = Util.cleanColumn(targetColumnName);
+ }
+
+ Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
+ if (type == null) {
+ throw new RuntimeException("Target column: " + targetColumnName + " not matched");
+ }
+ for (int i = 0; i < clearDmls.size(); i++) {
+ Map dmlData = clearDmls.get(i).getData();
+ List> item;
+ if (flag == false) {
+ item = new ArrayList<>();
+ values.add(item);
+ } else {
+ item = values.get(i);
+ }
+ Object value = dmlData.get(srcColumnName);
+ BatchExecutor.setValue(item, type, value);
+ }
+ flag = true;
+ }
+
+ try {
+ batchExecutor.batchExecute(insertSql.toString(), values);
+ } catch (SQLException e) {
+ if (skipDupException
+ && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) {
+ // ignore
+ // TODO 增加更多关系数据库的主键冲突的错误码
+ } else {
+ throw e;
+ }
+ }
+ if (logger.isTraceEnabled()) {
+ logger.trace("Insert into target table, sql: {}", insertSql);
+ }
+
+ }
+
+ /**
+ * 更新操作
+ *
+ * @param config 配置项
+ * @param dml DML数据
+ */
+ private void update(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException {
+ Map data = dml.getData();
+ if (data == null || data.isEmpty()) {
+ return;
+ }
+
+ Map old = dml.getOld();
+ if (old == null || old.isEmpty()) {
+ return;
+ }
+
+ DbMapping dbMapping = config.getDbMapping();
+ String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
+ Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
+
+ Map ctype = getTargetColumnType(batchExecutor.getConn(), config);
+
+ StringBuilder updateSql = new StringBuilder();
+ updateSql.append("ALTER TABLE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" UPDATE ");
+ List> values = new ArrayList<>();
+ boolean hasMatched = false;
+ for (String srcColumnName : old.keySet()) {
+ List targetColumnNames = new ArrayList<>();
+ columnsMap.forEach((targetColumn, srcColumn) -> {
+ if (srcColumnName.equalsIgnoreCase(srcColumn)) {
+ targetColumnNames.add(targetColumn);
+ }
+ });
+ if (!targetColumnNames.isEmpty()) {
+ hasMatched = true;
+ for (String targetColumnName : targetColumnNames) {
+ updateSql.append(backtick).append(targetColumnName).append(backtick).append("=?, ");
+ Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
+ if (type == null) {
+ throw new RuntimeException("Target column: " + targetColumnName + " not matched");
+ }
+ BatchExecutor.setValue(values, type, data.get(srcColumnName));
+ }
+ }
+ }
+ if (!hasMatched) {
+ logger.warn("Did not matched any columns to update ");
+ return;
+ }
+ int len = updateSql.length();
+ updateSql.delete(len - 2, len).append(" WHERE ");
+
+ // 拼接主键
+ appendCondition(dbMapping, updateSql, ctype, values, data, old);
+ batchExecutor.execute(updateSql.toString(), values);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Alter update target table, sql: {}", updateSql);
+ }
+ }
+
+ /**
+ * 删除操作
+ *
+ * @param config
+ * @param dml
+ */
+ private void delete(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException {
+ Map data = dml.getData();
+ if (data == null || data.isEmpty()) {
+ return;
+ }
+
+ DbMapping dbMapping = config.getDbMapping();
+ Map ctype = getTargetColumnType(batchExecutor.getConn(), config);
+
+ StringBuilder sql = new StringBuilder();
+ sql.append("ALTER TABLE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" DELETE WHERE ");
+
+ List> values = new ArrayList<>();
+ // 拼接主键
+ appendCondition(dbMapping, sql, ctype, values, data);
+ batchExecutor.execute(sql.toString(), values);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Alter delete from target table, sql: {}", sql);
+ }
+ }
+
+ /**
+ * truncate操作
+ *
+ * @param config
+ */
+ private void truncate(BatchExecutor batchExecutor, MappingConfig config) throws SQLException {
+ DbMapping dbMapping = config.getDbMapping();
+ StringBuilder sql = new StringBuilder();
+ sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()));
+ batchExecutor.execute(sql.toString(), new ArrayList<>());
+ if (logger.isTraceEnabled()) {
+ logger.trace("Truncate target table, sql: {}", sql);
+ }
+ }
+
+ /**
+ * 获取目标字段类型
+ *
+ * @param conn sql connection
+ * @param config 映射配置
+ * @return 字段sqlType
+ */
+ private Map getTargetColumnType(Connection conn, MappingConfig config) {
+ DbMapping dbMapping = config.getDbMapping();
+ String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable();
+ Map columnType = columnsTypeCache.get(cacheKey);
+ if (columnType == null) {
+ synchronized (ClickHouseBatchSyncService.class) {
+ columnType = columnsTypeCache.get(cacheKey);
+ if (columnType == null) {
+ columnType = new LinkedHashMap<>();
+ final Map columnTypeTmp = columnType;
+ String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " WHERE 1=2";
+ Util.sqlRS(conn, sql, rs -> {
+ try {
+ ResultSetMetaData rsd = rs.getMetaData();
+ int columnCount = rsd.getColumnCount();
+ for (int i = 1; i <= columnCount; i++) {
+ int colType = rsd.getColumnType(i);
+ // 修复year类型作为date处理时的data truncated问题
+ if ("YEAR".equals(rsd.getColumnTypeName(i))) {
+ colType = Types.VARCHAR;
+ }
+ columnTypeTmp.put(rsd.getColumnName(i).toLowerCase(), colType);
+ }
+ columnsTypeCache.put(cacheKey, columnTypeTmp);
+ } catch (SQLException e) {
+ logger.error(e.getMessage(), e);
+ }
+ });
+ }
+ }
+ }
+ return columnType;
+ }
+
+ /**
+ * 拼接主键 where条件
+ */
+ private void appendCondition(MappingConfig.DbMapping dbMapping, StringBuilder sql, Map ctype,
+ List> values, Map d) {
+ appendCondition(dbMapping, sql, ctype, values, d, null);
+ }
+
+ private void appendCondition(MappingConfig.DbMapping dbMapping, StringBuilder sql, Map ctype,
+ List> values, Map d, Map o) {
+ String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
+
+ // 拼接主键
+ for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) {
+ String targetColumnName = entry.getKey();
+ String srcColumnName = entry.getValue();
+ if (srcColumnName == null) {
+ srcColumnName = Util.cleanColumn(targetColumnName);
+ }
+ sql.append(backtick).append(targetColumnName).append(backtick).append("=? AND ");
+ Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
+ if (type == null) {
+ throw new RuntimeException("Target column: " + targetColumnName + " not matched");
+ }
+ // 如果有修改主键的情况
+ if (o != null && o.containsKey(srcColumnName)) {
+ BatchExecutor.setValue(values, type, o.get(srcColumnName));
+ } else {
+ BatchExecutor.setValue(values, type, d.get(srcColumnName));
+ }
+ }
+ int len = sql.length();
+ sql.delete(len - 4, len);
+ }
+
+ /**
+ * make sure the same table in one index
+ *
+ * @param dbMapping
+ * @return
+ */
+ private int mappingHash(MappingConfig.DbMapping dbMapping) {
+ int hash = dbMapping.getDatabase().toLowerCase().hashCode() + dbMapping.getTable().toLowerCase().hashCode();
+ hash = Math.abs(hash) % threads;
+ return Math.abs(hash);
+ }
+
+ public void close() {
+ for (int i = 0; i < threads; i++) {
+ executorThreads[i].shutdown();
+ }
+ }
+}
diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java
new file mode 100644
index 00000000..70e09da4
--- /dev/null
+++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java
@@ -0,0 +1,193 @@
+package com.alibaba.otter.canal.client.adapter.clickhouse.service;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping;
+import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
+import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService;
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+
+import javax.sql.DataSource;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * ClickHouse ETL 操作业务类
+ *
+ * @author rewerma @ 2018-11-7
+ * @version 1.0.0
+ */
+public class ClickHouseEtlService extends AbstractEtlService {
+
+ private DataSource targetDS;
+ private MappingConfig config;
+
+ public ClickHouseEtlService(DataSource targetDS, MappingConfig config){
+ super("CLICKHOUSE", config);
+ this.targetDS = targetDS;
+ this.config = config;
+ }
+
+ /**
+ * 导入数据
+ */
+ public EtlResult importData(List params) {
+ DbMapping dbMapping = config.getDbMapping();
+ String sql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable();
+ return importData(sql, params);
+ }
+
+ /**
+ * 执行导入
+ */
+ protected boolean executeSqlImport(DataSource srcDS, String sql, List