diff --git a/client-adapter/es-core/pom.xml b/client-adapter/es-core/pom.xml
new file mode 100644
index 00000000..af8aaacc
--- /dev/null
+++ b/client-adapter/es-core/pom.xml
@@ -0,0 +1,49 @@
+
+
+
+ canal.client-adapter
+ com.alibaba.otter
+ 1.1.5-SNAPSHOT
+
+ 4.0.0
+ com.alibaba.otter
+ client-adapter.es-core
+ jar
+ canal client adapter es-core module for otter ${project.version}
+
+
+
+ com.alibaba.otter
+ client-adapter.common
+ ${project.version}
+ provided
+
+
+ com.alibaba.fastsql
+ fastsql
+ 2.0.0_preview_855
+
+
+
+ junit
+ junit
+ 4.12
+ test
+
+
+ mysql
+ mysql-connector-java
+ 5.1.40
+ test
+
+
+ com.h2database
+ h2
+ 1.4.192
+ test
+
+
+
+
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java
new file mode 100644
index 00000000..d5bf39dc
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java
@@ -0,0 +1,177 @@
+package com.alibaba.otter.canal.client.adapter.es.core;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate;
+import org.apache.commons.lang.StringUtils;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.OuterAdapter;
+import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
+import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfigLoader;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser;
+import com.alibaba.otter.canal.client.adapter.es.core.monitor.ESConfigMonitor;
+import com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
+
+/**
+ * ES外部适配器
+ *
+ * @author rewerma 2018-10-20
+ * @version 1.0.0
+ */
+public abstract class ESAdapter implements OuterAdapter {
+
+ protected Map esSyncConfig = new ConcurrentHashMap<>(); // 文件名对应配置
+ protected Map> dbTableEsSyncConfig = new ConcurrentHashMap<>(); // schema-table对应配置
+
+ protected ESTemplate esTemplate;
+
+ protected ESSyncService esSyncService;
+
+ protected ESConfigMonitor esConfigMonitor;
+
+ protected Properties envProperties;
+
+ public ESSyncService getEsSyncService() {
+ return esSyncService;
+ }
+
+ public Map getEsSyncConfig() {
+ return esSyncConfig;
+ }
+
+ public Map> getDbTableEsSyncConfig() {
+ return dbTableEsSyncConfig;
+ }
+
+ @Override
+ public void init(OuterAdapterConfig configuration, Properties envProperties) {
+ try {
+ this.envProperties = envProperties;
+ Map esSyncConfigTmp = ESSyncConfigLoader.load(envProperties);
+ // 过滤不匹配的key的配置
+ esSyncConfigTmp.forEach((key, config) -> {
+ if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
+ || (config.getOuterAdapterKey() != null
+ && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
+ esSyncConfig.put(key, config);
+ }
+ });
+
+ for (Map.Entry entry : esSyncConfig.entrySet()) {
+ String configName = entry.getKey();
+ ESSyncConfig config = entry.getValue();
+
+ addSyncConfigToCache(configName, config);
+ }
+
+ esSyncService = new ESSyncService(esTemplate);
+
+ esConfigMonitor = new ESConfigMonitor();
+ esConfigMonitor.init(this, envProperties);
+ } catch (Throwable e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void sync(List dmls) {
+ if (dmls == null || dmls.isEmpty()) {
+ return;
+ }
+ for (Dml dml : dmls) {
+ if (!dml.getIsDdl()) {
+ sync(dml);
+ }
+ }
+ esSyncService.commit(); // 批次统一提交
+
+ }
+
+ private void sync(Dml dml) {
+ String database = dml.getDatabase();
+ String table = dml.getTable();
+ Map configMap;
+ if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+ configMap = dbTableEsSyncConfig
+ .get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + StringUtils.trimToEmpty(dml.getGroupId())
+ + "_" + database + "-" + table);
+ } else {
+ configMap = dbTableEsSyncConfig
+ .get(StringUtils.trimToEmpty(dml.getDestination()) + "_" + database + "-" + table);
+ }
+
+ if (configMap != null && !configMap.values().isEmpty()) {
+ esSyncService.sync(configMap.values(), dml);
+ }
+ }
+
+ @Override
+ public abstract EtlResult etl(String task, List params);
+
+ @Override
+ public abstract Map count(String task);
+
+ @Override
+ public void destroy() {
+ if (esConfigMonitor != null) {
+ esConfigMonitor.destroy();
+ }
+ }
+
+ @Override
+ public String getDestination(String task) {
+ ESSyncConfig config = esSyncConfig.get(task);
+ if (config != null) {
+ return config.getDestination();
+ }
+ return null;
+ }
+
+ public void addSyncConfigToCache(String configName, ESSyncConfig config) {
+ Properties envProperties = this.envProperties;
+ SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
+ config.getEsMapping().setSchemaItem(schemaItem);
+
+ DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+ if (dataSource == null || dataSource.getUrl() == null) {
+ throw new RuntimeException("No data source found: " + config.getDataSourceKey());
+ }
+ Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
+ Matcher matcher = pattern.matcher(dataSource.getUrl());
+ if (!matcher.find()) {
+ throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
+ }
+ String schema = matcher.group(2);
+
+ schemaItem.getAliasTableItems().values().forEach(tableItem -> {
+ Map esSyncConfigMap;
+ if (envProperties != null
+ && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+ esSyncConfigMap = dbTableEsSyncConfig
+ .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "-"
+ + StringUtils.trimToEmpty(config.getGroupId()) + "_" + schema + "-"
+ + tableItem.getTableName(),
+ k -> new ConcurrentHashMap<>());
+ } else {
+ esSyncConfigMap = dbTableEsSyncConfig
+ .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_" + schema + "-"
+ + tableItem.getTableName(),
+ k -> new ConcurrentHashMap<>());
+ }
+
+ esSyncConfigMap.put(configName, config);
+ });
+ }
+}
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java
new file mode 100644
index 00000000..1277ebbc
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java
@@ -0,0 +1,251 @@
+package com.alibaba.otter.canal.client.adapter.es.core.config;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+
+/**
+ * ES 映射配置
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESSyncConfig implements AdapterConfig {
+
+ private String dataSourceKey; // 数据源key
+
+ private String outerAdapterKey; // adapter key
+
+ private String groupId; // group id
+
+ private String destination; // canal destination
+
+ private ESMapping esMapping;
+
+ private String esVersion = "es6";
+
+ public void validate() {
+ if (esMapping._index == null) {
+ throw new NullPointerException("esMapping._index");
+ }
+ if ("es6".equals(esVersion) && esMapping._type == null) {
+ throw new NullPointerException("esMapping._type");
+ }
+ if (esMapping._id == null && esMapping.getPk() == null) {
+ throw new NullPointerException("esMapping._id or esMapping.pk");
+ }
+ if (esMapping.sql == null) {
+ throw new NullPointerException("esMapping.sql");
+ }
+ }
+
+ public String getDataSourceKey() {
+ return dataSourceKey;
+ }
+
+ public void setDataSourceKey(String dataSourceKey) {
+ this.dataSourceKey = dataSourceKey;
+ }
+
+ public String getOuterAdapterKey() {
+ return outerAdapterKey;
+ }
+
+ public void setOuterAdapterKey(String outerAdapterKey) {
+ this.outerAdapterKey = outerAdapterKey;
+ }
+
+ public String getGroupId() {
+ return groupId;
+ }
+
+ public void setGroupId(String groupId) {
+ this.groupId = groupId;
+ }
+
+ public String getDestination() {
+ return destination;
+ }
+
+ public void setDestination(String destination) {
+ this.destination = destination;
+ }
+
+ public ESMapping getEsMapping() {
+ return esMapping;
+ }
+
+ public void setEsMapping(ESMapping esMapping) {
+ this.esMapping = esMapping;
+ }
+
+ public ESMapping getMapping() {
+ return esMapping;
+ }
+
+ public String getEsVersion() {
+ return esVersion;
+ }
+
+ public void setEsVersion(String esVersion) {
+ this.esVersion = esVersion;
+ }
+
+ public static class ESMapping implements AdapterMapping {
+
+ private String _index;
+ private String _type;
+ private String _id;
+ private boolean upsert = false;
+ private String pk;
+ private Map relations = new LinkedHashMap<>();
+ private String sql;
+ // 对象字段, 例: objFields:
+ // - _labels: array:;
+ private Map objFields = new LinkedHashMap<>();
+ private List skips = new ArrayList<>();
+ private int commitBatch = 1000;
+ private String etlCondition;
+ private boolean syncByTimestamp = false; // 是否按时间戳定时同步
+ private Long syncInterval; // 同步时间间隔
+
+ private SchemaItem schemaItem; // sql解析结果模型
+
+ public String get_index() {
+ return _index;
+ }
+
+ public void set_index(String _index) {
+ this._index = _index;
+ }
+
+ public String get_type() {
+ return _type;
+ }
+
+ public void set_type(String _type) {
+ this._type = _type;
+ }
+
+ public String get_id() {
+ return _id;
+ }
+
+ public void set_id(String _id) {
+ this._id = _id;
+ }
+
+ public boolean isUpsert() {
+ return upsert;
+ }
+
+ public void setUpsert(boolean upsert) {
+ this.upsert = upsert;
+ }
+
+ public String getPk() {
+ return pk;
+ }
+
+ public void setPk(String pk) {
+ this.pk = pk;
+ }
+
+ public Map getObjFields() {
+ return objFields;
+ }
+
+ public void setObjFields(Map objFields) {
+ this.objFields = objFields;
+ }
+
+ public List getSkips() {
+ return skips;
+ }
+
+ public void setSkips(List skips) {
+ this.skips = skips;
+ }
+
+ public Map getRelations() {
+ return relations;
+ }
+
+ public void setRelations(Map relations) {
+ this.relations = relations;
+ }
+
+ public String getSql() {
+ return sql;
+ }
+
+ public void setSql(String sql) {
+ this.sql = sql;
+ }
+
+ public int getCommitBatch() {
+ return commitBatch;
+ }
+
+ public void setCommitBatch(int commitBatch) {
+ this.commitBatch = commitBatch;
+ }
+
+ public String getEtlCondition() {
+ return etlCondition;
+ }
+
+ public void setEtlCondition(String etlCondition) {
+ this.etlCondition = etlCondition;
+ }
+
+ public Long getSyncInterval() {
+ return syncInterval;
+ }
+
+ public void setSyncInterval(Long syncInterval) {
+ this.syncInterval = syncInterval;
+ }
+
+ public boolean isSyncByTimestamp() {
+ return syncByTimestamp;
+ }
+
+ public void setSyncByTimestamp(boolean syncByTimestamp) {
+ this.syncByTimestamp = syncByTimestamp;
+ }
+
+ public SchemaItem getSchemaItem() {
+ return schemaItem;
+ }
+
+ public void setSchemaItem(SchemaItem schemaItem) {
+ this.schemaItem = schemaItem;
+ }
+ }
+
+ public static class RelationMapping {
+
+ private String name;
+ private String parent;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getParent() {
+ return parent;
+ }
+
+ public void setParent(String parent) {
+ this.parent = parent;
+ }
+ }
+}
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfigLoader.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfigLoader.java
new file mode 100644
index 00000000..6b717a5d
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfigLoader.java
@@ -0,0 +1,47 @@
+package com.alibaba.otter.canal.client.adapter.es.core.config;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
+
+/**
+ * ES 配置装载器
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESSyncConfigLoader {
+
+ private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class);
+
+ public static synchronized Map load(Properties envProperties) {
+ logger.info("## Start loading es mapping config ... ");
+
+ Map esSyncConfig = new LinkedHashMap<>();
+
+ String esv = envProperties.getProperty("es.version");
+ Map configContentMap = MappingConfigsLoader.loadConfigs(esv);
+ configContentMap.forEach((fileName, content) -> {
+ ESSyncConfig config = YmlConfigBinder.bindYmlToObj(null, content, ESSyncConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
+ config.setEsVersion(esv);
+ try {
+ config.validate();
+ } catch (Exception e) {
+ throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e);
+ }
+ esSyncConfig.put(fileName, config);
+ });
+
+ logger.info("## ES mapping config loaded");
+ return esSyncConfig;
+ }
+}
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java
new file mode 100644
index 00000000..441e440f
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java
@@ -0,0 +1,439 @@
+package com.alibaba.otter.canal.client.adapter.es.core.config;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * ES 映射配置视图
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class SchemaItem {
+
+ private Map aliasTableItems = new LinkedHashMap<>(); // 别名对应表名
+ private Map selectFields = new LinkedHashMap<>(); // 查询字段
+ private String sql;
+
+ private volatile Map> tableItemAliases;
+ private volatile Map> columnFields;
+ private volatile Boolean allFieldsSimple;
+
+ public void init() {
+ this.getTableItemAliases();
+ this.getColumnFields();
+ this.isAllFieldsSimple();
+ aliasTableItems.values().forEach(tableItem -> {
+ tableItem.getRelationTableFields();
+ tableItem.getRelationSelectFieldItems();
+ });
+ }
+
+ public Map getAliasTableItems() {
+ return aliasTableItems;
+ }
+
+ public void setAliasTableItems(Map aliasTableItems) {
+ this.aliasTableItems = aliasTableItems;
+ }
+
+ public String getSql() {
+ return sql;
+ }
+
+ public void setSql(String sql) {
+ this.sql = sql;
+ }
+
+ public Map getSelectFields() {
+ return selectFields;
+ }
+
+ public void setSelectFields(Map selectFields) {
+ this.selectFields = selectFields;
+ }
+
+ public String toSql() {
+ // todo
+ return null;
+ }
+
+ public Map> getTableItemAliases() {
+ if (tableItemAliases == null) {
+ synchronized (SchemaItem.class) {
+ if (tableItemAliases == null) {
+ tableItemAliases = new LinkedHashMap<>();
+ aliasTableItems.forEach((alias, tableItem) -> {
+ List aliases = tableItemAliases
+ .computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>());
+ aliases.add(tableItem);
+ });
+ }
+ }
+ }
+ return tableItemAliases;
+ }
+
+ public Map> getColumnFields() {
+ if (columnFields == null) {
+ synchronized (SchemaItem.class) {
+ if (columnFields == null) {
+ columnFields = new LinkedHashMap<>();
+ getSelectFields()
+ .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> {
+ // TableItem tableItem = getAliasTableItems().get(columnItem.getOwner());
+ // if (!tableItem.isSubQuery()) {
+ //当数据列并非原始列时，columnName是空的，例如concat('px',id)
+ if(columnItem.getColumnName() != null) {
+ List fieldItems = columnFields.computeIfAbsent(
+ columnItem.getOwner() + "." + columnItem.getColumnName(),
+ k -> new ArrayList<>());
+ fieldItems.add(fieldItem);
+ }
+ // } else {
+ // tableItem.getSubQueryFields().forEach(subQueryField -> {
+ // List fieldItems = columnFields.computeIfAbsent(
+ // columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(),
+ // k -> new ArrayList<>());
+ // fieldItems.add(fieldItem);
+ // });
+ // }
+ }));
+ }
+ }
+ }
+ return columnFields;
+ }
+
+ public boolean isAllFieldsSimple() {
+ if (allFieldsSimple == null) {
+ synchronized (SchemaItem.class) {
+ if (allFieldsSimple == null) {
+ allFieldsSimple = true;
+
+ for (FieldItem fieldItem : getSelectFields().values()) {
+ if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
+ allFieldsSimple = false;
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ return allFieldsSimple;
+ }
+
+ public TableItem getMainTable() {
+ if (!aliasTableItems.isEmpty()) {
+ return aliasTableItems.values().iterator().next();
+ } else {
+ return null;
+ }
+ }
+
+ public FieldItem getIdFieldItem(ESSyncConfig.ESMapping mapping) {
+ if (mapping.get_id() != null) {
+ return getSelectFields().get(mapping.get_id());
+ } else {
+ return getSelectFields().get(mapping.getPk());
+ }
+ }
+
+ public static class TableItem {
+
+ private SchemaItem schemaItem;
+
+ private String schema;
+ private String tableName;
+ private String alias;
+ private String subQuerySql;
+ private List subQueryFields = new ArrayList<>();
+ private List relationFields = new ArrayList<>();
+
+ private boolean main;
+ private boolean subQuery;
+
+ private volatile Map> relationTableFields; // 当前表关联条件字段对应主表查询字段
+ private volatile List relationSelectFieldItems; // 子表所在主表的查询字段
+
+ public TableItem(SchemaItem schemaItem){
+ this.schemaItem = schemaItem;
+ }
+
+ public SchemaItem getSchemaItem() {
+ return schemaItem;
+ }
+
+ public void setSchemaItem(SchemaItem schemaItem) {
+ this.schemaItem = schemaItem;
+ }
+
+ public String getSchema() {
+ return schema;
+ }
+
+ public void setSchema(String schema) {
+ this.schema = schema;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = tableName;
+ }
+
+ public String getAlias() {
+ return alias;
+ }
+
+ public void setAlias(String alias) {
+ this.alias = alias;
+ }
+
+ public String getSubQuerySql() {
+ return subQuerySql;
+ }
+
+ public void setSubQuerySql(String subQuerySql) {
+ this.subQuerySql = subQuerySql;
+ }
+
+ public boolean isMain() {
+ return main;
+ }
+
+ public void setMain(boolean main) {
+ this.main = main;
+ }
+
+ public boolean isSubQuery() {
+ return subQuery;
+ }
+
+ public void setSubQuery(boolean subQuery) {
+ this.subQuery = subQuery;
+ }
+
+ public List getSubQueryFields() {
+ return subQueryFields;
+ }
+
+ public void setSubQueryFields(List subQueryFields) {
+ this.subQueryFields = subQueryFields;
+ }
+
+ public List getRelationFields() {
+ return relationFields;
+ }
+
+ public void setRelationFields(List relationFields) {
+ this.relationFields = relationFields;
+ }
+
+ public Map> getRelationTableFields() {
+ if (relationTableFields == null) {
+ synchronized (SchemaItem.class) {
+ if (relationTableFields == null) {
+ relationTableFields = new LinkedHashMap<>();
+
+ getRelationFields().forEach(relationFieldsPair -> {
+ FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem();
+ FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem();
+ FieldItem currentTableRelField = null;
+ if (getAlias().equals(leftFieldItem.getOwner())) {
+ currentTableRelField = leftFieldItem;
+ } else if (getAlias().equals(rightFieldItem.getOwner())) {
+ currentTableRelField = rightFieldItem;
+ }
+
+ if (currentTableRelField != null) {
+ List selectFieldItem = getSchemaItem().getColumnFields()
+ .get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName());
+ if (selectFieldItem != null && !selectFieldItem.isEmpty()) {
+ relationTableFields.put(currentTableRelField, selectFieldItem);
+ } else {
+ selectFieldItem = getSchemaItem().getColumnFields()
+ .get(rightFieldItem.getOwner() + "."
+ + rightFieldItem.getColumn().getColumnName());
+ if (selectFieldItem != null && !selectFieldItem.isEmpty()) {
+ relationTableFields.put(currentTableRelField, selectFieldItem);
+ } else {
+ throw new UnsupportedOperationException(
+ "Relation condition column must in select columns.");
+ }
+ }
+ }
+ });
+ }
+ }
+ }
+ return relationTableFields;
+ }
+
+ public List getRelationSelectFieldItems() {
+ if (relationSelectFieldItems == null) {
+ synchronized (SchemaItem.class) {
+ if (relationSelectFieldItems == null) {
+ List relationSelectFieldItemsTmp = new ArrayList<>();
+ for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
+ if (fieldItem.getOwners().contains(getAlias())) {
+ relationSelectFieldItemsTmp.add(fieldItem);
+ }
+ }
+ relationSelectFieldItems = relationSelectFieldItemsTmp;
+ }
+ }
+ }
+ return relationSelectFieldItems;
+ }
+ }
+
+ public static class RelationFieldsPair {
+
+ private FieldItem leftFieldItem;
+ private FieldItem rightFieldItem;
+
+ public RelationFieldsPair(FieldItem leftFieldItem, FieldItem rightFieldItem){
+ this.leftFieldItem = leftFieldItem;
+ this.rightFieldItem = rightFieldItem;
+ }
+
+ public FieldItem getLeftFieldItem() {
+ return leftFieldItem;
+ }
+
+ public void setLeftFieldItem(FieldItem leftFieldItem) {
+ this.leftFieldItem = leftFieldItem;
+ }
+
+ public FieldItem getRightFieldItem() {
+ return rightFieldItem;
+ }
+
+ public void setRightFieldItem(FieldItem rightFieldItem) {
+ this.rightFieldItem = rightFieldItem;
+ }
+ }
+
+ public static class FieldItem {
+
+ private String fieldName;
+ private String expr;
+ private List columnItems = new ArrayList<>();
+ private List owners = new ArrayList<>();
+
+ private boolean method;
+ private boolean binaryOp;
+
+ public String getFieldName() {
+ return fieldName;
+ }
+
+ public void setFieldName(String fieldName) {
+ this.fieldName = fieldName;
+ }
+
+ public String getExpr() {
+ return expr;
+ }
+
+ public void setExpr(String expr) {
+ this.expr = expr;
+ }
+
+ public List getColumnItems() {
+ return columnItems;
+ }
+
+ public void setColumnItems(List columnItems) {
+ this.columnItems = columnItems;
+ }
+
+ public boolean isMethod() {
+ return method;
+ }
+
+ public void setMethod(boolean method) {
+ this.method = method;
+ }
+
+ public boolean isBinaryOp() {
+ return binaryOp;
+ }
+
+ public void setBinaryOp(boolean binaryOp) {
+ this.binaryOp = binaryOp;
+ }
+
+ public List getOwners() {
+ return owners;
+ }
+
+ public void setOwners(List owners) {
+ this.owners = owners;
+ }
+
+ public void addColumn(ColumnItem columnItem) {
+ columnItems.add(columnItem);
+ }
+
+ public ColumnItem getColumn() {
+ if (!columnItems.isEmpty()) {
+ return columnItems.get(0);
+ } else {
+ return null;
+ }
+ }
+
+ public String getOwner() {
+ if (!owners.isEmpty()) {
+ return owners.get(0);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ FieldItem fieldItem = (FieldItem) o;
+
+ return fieldName != null ? fieldName.equals(fieldItem.fieldName) : fieldItem.fieldName == null;
+ }
+
+ @Override
+ public int hashCode() {
+ return fieldName != null ? fieldName.hashCode() : 0;
+ }
+ }
+
+ public static class ColumnItem {
+
+ private String owner;
+ private String columnName;
+
+ public String getOwner() {
+ return owner;
+ }
+
+ public void setOwner(String owner) {
+ this.owner = owner;
+ }
+
+ public String getColumnName() {
+ return columnName;
+ }
+
+ public void setColumnName(String columnName) {
+ this.columnName = columnName;
+ }
+ }
+}
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java
new file mode 100644
index 00000000..59d94512
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java
@@ -0,0 +1,276 @@
+package com.alibaba.otter.canal.client.adapter.es.core.config;
+
+import static com.alibaba.fastsql.sql.ast.expr.SQLBinaryOperator.BooleanAnd;
+import static com.alibaba.fastsql.sql.ast.expr.SQLBinaryOperator.Equality;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import com.alibaba.fastsql.sql.SQLUtils;
+import com.alibaba.fastsql.sql.ast.SQLExpr;
+import com.alibaba.fastsql.sql.ast.expr.*;
+import com.alibaba.fastsql.sql.ast.statement.*;
+import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
+import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser;
+import com.alibaba.fastsql.sql.parser.ParserException;
+import com.alibaba.fastsql.sql.parser.SQLStatementParser;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.RelationFieldsPair;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.TableItem;
+
+/**
+ * ES同步指定sql格式解析
+ *
+ * @author rewerma 2018-10-26 下午03:45:49
+ * @version 1.0.0
+ */
+public class SqlParser {
+
+ /**
+ * 解析sql
+ *
+ * @param sql sql
+ * @return 视图对象
+ */
+ public static SchemaItem parse(String sql) {
+ try {
+ SQLStatementParser parser = new MySqlStatementParser(sql);
+ SQLSelectStatement statement = (SQLSelectStatement) parser.parseStatement();
+ MySqlSelectQueryBlock sqlSelectQueryBlock = (MySqlSelectQueryBlock) statement.getSelect().getQuery();
+
+ SchemaItem schemaItem = new SchemaItem();
+ schemaItem.setSql(SQLUtils.toMySqlString(sqlSelectQueryBlock));
+ SQLTableSource sqlTableSource = sqlSelectQueryBlock.getFrom();
+ List tableItems = new ArrayList<>();
+ SqlParser.visitSelectTable(schemaItem, sqlTableSource, tableItems, null);
+ tableItems.forEach(tableItem -> schemaItem.getAliasTableItems().put(tableItem.getAlias(), tableItem));
+
+ List fieldItems = collectSelectQueryFields(sqlSelectQueryBlock);
+ fieldItems.forEach(fieldItem -> schemaItem.getSelectFields().put(fieldItem.getFieldName(), fieldItem));
+
+ schemaItem.init();
+
+ if (schemaItem.getAliasTableItems().isEmpty() || schemaItem.getSelectFields().isEmpty()) {
+ throw new ParserException("Parse sql error");
+ }
+ return schemaItem;
+ } catch (Exception e) {
+ throw new ParserException();
+ }
+ }
+
+ /**
+ * 归集字段
+ *
+ * @param sqlSelectQueryBlock sqlSelectQueryBlock
+ * @return 字段属性列表
+ */
+ private static List collectSelectQueryFields(MySqlSelectQueryBlock sqlSelectQueryBlock) {
+ return sqlSelectQueryBlock.getSelectList().stream().map(selectItem -> {
+ FieldItem fieldItem = new FieldItem();
+ fieldItem.setFieldName(selectItem.getAlias());
+ fieldItem.setExpr(selectItem.toString());
+ visitColumn(selectItem.getExpr(), fieldItem);
+ return fieldItem;
+ }).collect(Collectors.toList());
+ }
+
+ /**
+ * 解析字段
+ *
+ * @param expr sql expr
+ * @param fieldItem 字段属性
+ */
+ private static void visitColumn(SQLExpr expr, FieldItem fieldItem) {
+ if (expr instanceof SQLIdentifierExpr) {
+ // 无owner
+ SQLIdentifierExpr identifierExpr = (SQLIdentifierExpr) expr;
+ if (fieldItem.getFieldName() == null) {
+ fieldItem.setFieldName(identifierExpr.getName());
+ fieldItem.setExpr(identifierExpr.toString());
+ }
+ ColumnItem columnItem = new ColumnItem();
+ columnItem.setColumnName(identifierExpr.getName());
+ fieldItem.getOwners().add(null);
+ fieldItem.addColumn(columnItem);
+ } else if (expr instanceof SQLPropertyExpr) {
+ // 有owner
+ SQLPropertyExpr sqlPropertyExpr = (SQLPropertyExpr) expr;
+ if (fieldItem.getFieldName() == null) {
+ fieldItem.setFieldName(sqlPropertyExpr.getName());
+ fieldItem.setExpr(sqlPropertyExpr.toString());
+ }
+ fieldItem.getOwners().add(sqlPropertyExpr.getOwnernName());
+ ColumnItem columnItem = new ColumnItem();
+ columnItem.setColumnName(sqlPropertyExpr.getName());
+ columnItem.setOwner(sqlPropertyExpr.getOwnernName());
+ fieldItem.addColumn(columnItem);
+ } else if (expr instanceof SQLMethodInvokeExpr) {
+ SQLMethodInvokeExpr methodInvokeExpr = (SQLMethodInvokeExpr) expr;
+ fieldItem.setMethod(true);
+ for (SQLExpr sqlExpr : methodInvokeExpr.getArguments()) {
+ visitColumn(sqlExpr, fieldItem);
+ }
+ } else if (expr instanceof SQLBinaryOpExpr) {
+ SQLBinaryOpExpr sqlBinaryOpExpr = (SQLBinaryOpExpr) expr;
+ fieldItem.setBinaryOp(true);
+ visitColumn(sqlBinaryOpExpr.getLeft(), fieldItem);
+ visitColumn(sqlBinaryOpExpr.getRight(), fieldItem);
+ } else if (expr instanceof SQLCaseExpr) {
+ SQLCaseExpr sqlCaseExpr = (SQLCaseExpr) expr;
+ fieldItem.setMethod(true);
+ sqlCaseExpr.getItems().forEach(item -> visitColumn(item.getConditionExpr(), fieldItem));
+ } else if (expr instanceof SQLCharExpr) {
+ SQLCharExpr sqlCharExpr = (SQLCharExpr) expr;
+ String owner = null;
+ String columnName = null;
+ if (sqlCharExpr.getParent() instanceof SQLCaseExpr.Item) {
+ owner = ((SQLPropertyExpr) ((SQLCaseExpr.Item) sqlCharExpr.getParent()).getValueExpr()).getOwnernName();
+ columnName = ((SQLPropertyExpr) ((SQLCaseExpr.Item) sqlCharExpr.getParent()).getValueExpr()).getName();
+ }
+ if (fieldItem.getFieldName() == null) {
+ fieldItem.setFieldName(columnName);
+ fieldItem.setExpr(sqlCharExpr.toString());
+ }
+ ColumnItem columnItem = new ColumnItem();
+ columnItem.setColumnName(columnName);
+ columnItem.setOwner(owner);
+ fieldItem.getOwners().add(owner);
+ fieldItem.addColumn(columnItem);
+ }
+ }
+
+ /**
+ * 解析表
+ *
+ * @param schemaItem 视图对象
+ * @param sqlTableSource sqlTableSource
+ * @param tableItems 表对象列表
+ * @param tableItemTmp 表对象(临时)
+ */
+ private static void visitSelectTable(SchemaItem schemaItem, SQLTableSource sqlTableSource,
+ List tableItems, TableItem tableItemTmp) {
+ if (sqlTableSource instanceof SQLExprTableSource) {
+ SQLExprTableSource sqlExprTableSource = (SQLExprTableSource) sqlTableSource;
+ TableItem tableItem;
+ if (tableItemTmp != null) {
+ tableItem = tableItemTmp;
+ } else {
+ tableItem = new TableItem(schemaItem);
+ }
+ tableItem.setSchema(sqlExprTableSource.getSchema());
+ tableItem.setTableName(sqlExprTableSource.getTableName());
+ if (tableItem.getAlias() == null) {
+ tableItem.setAlias(sqlExprTableSource.getAlias());
+ }
+ if (tableItems.isEmpty()) {
+ // 第一张表为主表
+ tableItem.setMain(true);
+ }
+ tableItems.add(tableItem);
+ } else if (sqlTableSource instanceof SQLJoinTableSource) {
+ SQLJoinTableSource sqlJoinTableSource = (SQLJoinTableSource) sqlTableSource;
+ SQLTableSource leftTableSource = sqlJoinTableSource.getLeft();
+ visitSelectTable(schemaItem, leftTableSource, tableItems, null);
+ SQLTableSource rightTableSource = sqlJoinTableSource.getRight();
+ TableItem rightTableItem = new TableItem(schemaItem);
+ // 解析on条件字段
+ visitOnCondition(sqlJoinTableSource.getCondition(), rightTableItem);
+ visitSelectTable(schemaItem, rightTableSource, tableItems, rightTableItem);
+
+ } else if (sqlTableSource instanceof SQLSubqueryTableSource) {
+ SQLSubqueryTableSource subQueryTableSource = (SQLSubqueryTableSource) sqlTableSource;
+ MySqlSelectQueryBlock sqlSelectQuery = (MySqlSelectQueryBlock) subQueryTableSource.getSelect().getQuery();
+ TableItem tableItem;
+ if (tableItemTmp != null) {
+ tableItem = tableItemTmp;
+ } else {
+ tableItem = new TableItem(schemaItem);
+ }
+ tableItem.setAlias(subQueryTableSource.getAlias());
+ tableItem.setSubQuerySql(SQLUtils.toMySqlString(sqlSelectQuery));
+ tableItem.setSubQuery(true);
+ tableItem.setSubQueryFields(collectSelectQueryFields(sqlSelectQuery));
+ visitSelectTable(schemaItem, sqlSelectQuery.getFrom(), tableItems, tableItem);
+ }
+ }
+
+ /**
+ * 解析on条件
+ *
+ * @param expr sql expr
+ * @param tableItem 表对象
+ */
+ private static void visitOnCondition(SQLExpr expr, TableItem tableItem) {
+ if (!(expr instanceof SQLBinaryOpExpr)) {
+ throw new UnsupportedOperationException();
+ }
+ SQLBinaryOpExpr sqlBinaryOpExpr = (SQLBinaryOpExpr) expr;
+ if (sqlBinaryOpExpr.getOperator() == BooleanAnd) {
+ visitOnCondition(sqlBinaryOpExpr.getLeft(), tableItem);
+ visitOnCondition(sqlBinaryOpExpr.getRight(), tableItem);
+ } else if (sqlBinaryOpExpr.getOperator() == Equality) {
+ FieldItem leftFieldItem = new FieldItem();
+ visitColumn(sqlBinaryOpExpr.getLeft(), leftFieldItem);
+ if (leftFieldItem.getColumnItems().size() != 1 || leftFieldItem.isMethod() || leftFieldItem.isBinaryOp()) {
+ throw new UnsupportedOperationException("Unsupported for complex of on-condition");
+ }
+ FieldItem rightFieldItem = new FieldItem();
+ visitColumn(sqlBinaryOpExpr.getRight(), rightFieldItem);
+ if (rightFieldItem.getColumnItems().size() != 1 || rightFieldItem.isMethod()
+ || rightFieldItem.isBinaryOp()) {
+ throw new UnsupportedOperationException("Unsupported for complex of on-condition");
+ }
+ tableItem.getRelationFields().add(new RelationFieldsPair(leftFieldItem, rightFieldItem));
+ } else {
+ throw new UnsupportedOperationException("Unsupported for complex of on-condition");
+ }
+ }
+
+ public static MySqlSelectQueryBlock parseSQLSelectQueryBlock(String sql) {
+ if (sql == null || "".equals(sql)) {
+ return null;
+ }
+ SQLStatementParser parser = new MySqlStatementParser(sql);
+ SQLSelectStatement statement = (SQLSelectStatement) parser.parseStatement();
+ return (MySqlSelectQueryBlock) statement.getSelect().getQuery();
+ }
+
+ public static String parse4SQLSelectItem(MySqlSelectQueryBlock sqlSelectQueryBlock) {
+ List selectItems = sqlSelectQueryBlock.getSelectList();
+ StringBuilder subSql = new StringBuilder();
+ int i = 0;
+ for (SQLSelectItem sqlSelectItem : selectItems) {
+ if (i != 0) {
+ subSql.append(",");
+ } else {
+ i++;
+ }
+ subSql.append(SQLUtils.toMySqlString(sqlSelectItem));
+ }
+ return subSql.toString();
+ }
+
+ public static String parse4FromTableSource(MySqlSelectQueryBlock sqlSelectQueryBlock) {
+ SQLTableSource sqlTableSource = sqlSelectQueryBlock.getFrom();
+ return SQLUtils.toMySqlString(sqlTableSource);
+ }
+
+ public static String parse4WhereItem(MySqlSelectQueryBlock sqlSelectQueryBlock) {
+ SQLExpr sqlExpr = sqlSelectQueryBlock.getWhere();
+ if (sqlExpr != null) {
+ return SQLUtils.toMySqlString(sqlExpr);
+ }
+ return null;
+ }
+
+ public static String parse4GroupBy(MySqlSelectQueryBlock sqlSelectQueryBlock) {
+ SQLSelectGroupByClause expr = sqlSelectQueryBlock.getGroupBy();
+ if (expr != null) {
+ return SQLUtils.toMySqlString(expr);
+ }
+ return null;
+ }
+}
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java
new file mode 100644
index 00000000..527d292b
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java
@@ -0,0 +1,146 @@
+package com.alibaba.otter.canal.client.adapter.es.core.monitor;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter;
+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 com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
+import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+
+public class ESConfigMonitor {
+
+ private static final Logger logger = LoggerFactory.getLogger(ESConfigMonitor.class);
+
+ private static final String adapterName = "es";
+
+ private ESAdapter esAdapter;
+
+ private Properties envProperties;
+
+ private FileAlterationMonitor fileMonitor;
+
+ public void init(ESAdapter esAdapter, Properties envProperties) {
+ this.esAdapter = esAdapter;
+ 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());
+ ESSyncConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
+ if (config != null) {
+ config.validate();
+ addConfigToCache(file, config);
+ logger.info("Add a new es 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 (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
+ // 加载配置文件
+ String configContent = MappingConfigsLoader
+ .loadConfig(adapterName + File.separator + file.getName());
+ if (configContent == null) {
+ onFileDelete(file);
+ return;
+ }
+ ESSyncConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
+ config.validate();
+ if (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
+ deleteConfigFromCache(file);
+ }
+ addConfigToCache(file, config);
+
+ logger.info("Change a es 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 (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
+ deleteConfigFromCache(file);
+
+ logger.info("Delete a es mapping config: {} of canal adapter", file.getName());
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ private void addConfigToCache(File file, ESSyncConfig config) {
+ esAdapter.getEsSyncConfig().put(file.getName(), config);
+
+ esAdapter.addSyncConfigToCache(file.getName(),config);
+ }
+
+ private void deleteConfigFromCache(File file) {
+ esAdapter.getEsSyncConfig().remove(file.getName());
+ for (Map configMap : esAdapter.getDbTableEsSyncConfig().values()) {
+ if (configMap != null) {
+ configMap.remove(file.getName());
+ }
+ }
+
+ }
+ }
+}
diff --git a/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java
new file mode 100644
index 00000000..349093bb
--- /dev/null
+++ b/client-adapter/es-core/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java
@@ -0,0 +1,875 @@
+package com.alibaba.otter.canal.client.adapter.es.core.service;
+
+import java.util.*;
+
+import javax.sql.DataSource;
+
+import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
+import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem;
+import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.TableItem;
+import com.alibaba.otter.canal.client.adapter.es.core.support.ESSyncUtil;
+import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+
+/**
+ * ES 同步 Service
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESSyncService {
+
+ private static Logger logger = LoggerFactory.getLogger(ESSyncService.class);
+
+ private ESTemplate esTemplate;
+
+ public ESSyncService(ESTemplate esTemplate){
+ this.esTemplate = esTemplate;
+ }
+
+ public void sync(Collection esSyncConfigs, Dml dml) {
+ long begin = System.currentTimeMillis();
+ if (esSyncConfigs != null) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Destination: {}, database:{}, table:{}, type:{}, affected index count: {}",
+ dml.getDestination(),
+ dml.getDatabase(),
+ dml.getTable(),
+ dml.getType(),
+ esSyncConfigs.size());
+ }
+
+ for (ESSyncConfig config : esSyncConfigs) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Prepared to sync index: {}, destination: {}",
+ config.getEsMapping().get_index(),
+ dml.getDestination());
+ }
+ this.sync(config, dml);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sync completed: {}, destination: {}",
+ config.getEsMapping().get_index(),
+ dml.getDestination());
+ }
+ }
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sync elapsed time: {} ms, affected indexes count：{}, destination: {}",
+ (System.currentTimeMillis() - begin),
+ esSyncConfigs.size(),
+ dml.getDestination());
+ }
+ if (logger.isDebugEnabled()) {
+ StringBuilder configIndexes = new StringBuilder();
+ esSyncConfigs
+ .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
+ logger.debug("DML: {} \nAffected indexes: {}",
+ JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
+ configIndexes.toString());
+ }
+ }
+ }
+
+ public void sync(ESSyncConfig config, Dml dml) {
+ try {
+ // 如果是按时间戳定时更新则返回
+ if (config.getEsMapping().isSyncByTimestamp()) {
+ return;
+ }
+
+ long begin = System.currentTimeMillis();
+
+ String type = dml.getType();
+ if (type != null && type.equalsIgnoreCase("INSERT")) {
+ insert(config, dml);
+ } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
+ update(config, dml);
+ } else if (type != null && type.equalsIgnoreCase("DELETE")) {
+ delete(config, dml);
+ } else {
+ return;
+ }
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sync elapsed time: {} ms,destination: {}, es index: {}",
+ (System.currentTimeMillis() - begin),
+ dml.getDestination(),
+ config.getEsMapping().get_index());
+ }
+ } catch (Throwable e) {
+ logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().get_index(), dml);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * 插入操作dml
+ *
+ * @param config es配置
+ * @param dml dml数据
+ */
+ private void insert(ESSyncConfig config, Dml dml) {
+ List> dataList = dml.getData();
+ if (dataList == null || dataList.isEmpty()) {
+ return;
+ }
+ SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
+ for (Map data : dataList) {
+ if (data == null || data.isEmpty()) {
+ continue;
+ }
+
+ if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) {
+ // ------单表 & 所有字段都为简单字段------
+ singleTableSimpleFiledInsert(config, dml, data);
+ } else {
+ // ------是主表 查询sql来插入------
+ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
+ mainTableInsert(config, dml, data);
+ }
+
+ // 从表的操作
+ for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
+ if (tableItem.isMain()) {
+ continue;
+ }
+ if (!tableItem.getTableName().equals(dml.getTable())) {
+ continue;
+ }
+ // 关联条件出现在主表查询条件是否为简单字段
+ boolean allFieldsSimple = true;
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
+ allFieldsSimple = false;
+ break;
+ }
+ }
+ // 所有查询字段均为简单字段
+ if (allFieldsSimple) {
+ // 不是子查询
+ if (!tableItem.isSubQuery()) {
+ // ------关联表简单字段插入------
+ Map esFieldData = new LinkedHashMap<>();
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ Object value = esTemplate.getValFromData(config.getEsMapping(),
+ data,
+ fieldItem.getFieldName(),
+ fieldItem.getColumn().getColumnName());
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
+ }
+
+ joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
+ } else {
+ // ------关联子表简单字段插入------
+ subTableSimpleFieldOperation(config, dml, data, null, tableItem);
+ }
+ } else {
+ // ------关联子表复杂字段插入 执行全sql更新es------
+ wholeSqlOperation(config, dml, data, null, tableItem);
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * 更新操作dml
+ *
+ * @param config es配置
+ * @param dml dml数据
+ */
+ private void update(ESSyncConfig config, Dml dml) {
+ List> dataList = dml.getData();
+ List> oldList = dml.getOld();
+ if (dataList == null || dataList.isEmpty() || oldList == null || oldList.isEmpty()) {
+ return;
+ }
+ SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
+ int i = 0;
+ for (Map data : dataList) {
+ Map old = oldList.get(i);
+ if (data == null || data.isEmpty() || old == null || old.isEmpty()) {
+ continue;
+ }
+
+ if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) {
+ // ------单表 & 所有字段都为简单字段------
+ singleTableSimpleFiledUpdate(config, dml, data, old);
+ } else {
+ // ------主表 查询sql来更新------
+ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
+ ESMapping mapping = config.getEsMapping();
+ String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+ FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName);
+
+ boolean idFieldSimple = true;
+ if (idFieldItem.isMethod() || idFieldItem.isBinaryOp()) {
+ idFieldSimple = false;
+ }
+
+ boolean allUpdateFieldSimple = true;
+ out: for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
+ for (ColumnItem columnItem : fieldItem.getColumnItems()) {
+ if (old.containsKey(columnItem.getColumnName())) {
+ if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
+ allUpdateFieldSimple = false;
+ break out;
+ }
+ }
+ }
+ }
+
+ // 不支持主键更新!!
+
+ // 判断是否有外键更新
+ boolean fkChanged = false;
+ for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
+ if (tableItem.isMain()) {
+ continue;
+ }
+ boolean changed = false;
+ for (List fieldItems : tableItem.getRelationTableFields().values()) {
+ for (FieldItem fieldItem : fieldItems) {
+ if (old.containsKey(fieldItem.getColumn().getColumnName())) {
+ fkChanged = true;
+ changed = true;
+ break;
+ }
+ }
+ }
+ // 如果外键有修改,则更新所对应该表的所有查询条件数据
+ if (changed) {
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ fieldItem.getColumnItems()
+ .forEach(columnItem -> old.put(columnItem.getColumnName(), null));
+ }
+ }
+ }
+
+ // 判断主键和所更新的字段是否全为简单字段
+ if (idFieldSimple && allUpdateFieldSimple && !fkChanged) {
+ singleTableSimpleFiledUpdate(config, dml, data, old);
+ } else {
+ mainTableUpdate(config, dml, data, old);
+ }
+ }
+
+ // 从表的操作
+ for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
+ if (tableItem.isMain()) {
+ continue;
+ }
+ if (!tableItem.getTableName().equals(dml.getTable())) {
+ continue;
+ }
+
+ // 关联条件出现在主表查询条件是否为简单字段
+ boolean allFieldsSimple = true;
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
+ allFieldsSimple = false;
+ break;
+ }
+ }
+
+ // 所有查询字段均为简单字段
+ if (allFieldsSimple) {
+ // 不是子查询
+ if (!tableItem.isSubQuery()) {
+ // ------关联表简单字段更新------
+ Map esFieldData = new LinkedHashMap<>();
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ if (old.containsKey(fieldItem.getColumn().getColumnName())) {
+ Object value = esTemplate.getValFromData(config.getEsMapping(),
+ data,
+ fieldItem.getFieldName(),
+ fieldItem.getColumn().getColumnName());
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
+ }
+ }
+ joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
+ } else {
+ // ------关联子表简单字段更新------
+ subTableSimpleFieldOperation(config, dml, data, old, tableItem);
+ }
+ } else {
+ // ------关联子表复杂字段更新 执行全sql更新es------
+ wholeSqlOperation(config, dml, data, old, tableItem);
+ }
+ }
+ }
+
+ i++;
+ }
+ }
+
+ /**
+ * 删除操作dml
+ *
+ * @param config es配置
+ * @param dml dml数据
+ */
+ private void delete(ESSyncConfig config, Dml dml) {
+ List> dataList = dml.getData();
+ if (dataList == null || dataList.isEmpty()) {
+ return;
+ }
+ SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
+
+ for (Map data : dataList) {
+ if (data == null || data.isEmpty()) {
+ continue;
+ }
+
+ ESMapping mapping = config.getEsMapping();
+
+ // ------是主表------
+ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
+ if (mapping.get_id() != null) {
+ FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping);
+ // 主键为简单字段
+ if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) {
+ Object idVal = esTemplate.getValFromData(mapping,
+ data,
+ idFieldItem.getFieldName(),
+ idFieldItem.getColumn().getColumnName());
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ idVal);
+ }
+ esTemplate.delete(mapping, idVal, null);
+ } else {
+ // ------主键带函数, 查询sql获取主键删除------
+ // FIXME 删除时反查sql为空记录, 无法获获取 id field 值
+ mainTableDelete(config, dml, data);
+ }
+ } else {
+ FieldItem pkFieldItem = schemaItem.getIdFieldItem(mapping);
+ if (!pkFieldItem.isMethod() && !pkFieldItem.isBinaryOp()) {
+ Map esFieldData = new LinkedHashMap<>();
+ Object pkVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ pkVal);
+ }
+ esFieldData.remove(pkFieldItem.getFieldName());
+ esFieldData.keySet().forEach(key -> esFieldData.put(key, null));
+ esTemplate.delete(mapping, pkVal, esFieldData);
+ } else {
+ // ------主键带函数, 查询sql获取主键删除------
+ mainTableDelete(config, dml, data);
+ }
+ }
+
+ }
+
+ // 从表的操作
+ for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
+ if (tableItem.isMain()) {
+ continue;
+ }
+ if (!tableItem.getTableName().equals(dml.getTable())) {
+ continue;
+ }
+
+ // 关联条件出现在主表查询条件是否为简单字段
+ boolean allFieldsSimple = true;
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
+ allFieldsSimple = false;
+ break;
+ }
+ }
+
+ // 所有查询字段均为简单字段
+ if (allFieldsSimple) {
+ // 不是子查询
+ if (!tableItem.isSubQuery()) {
+ // ------关联表简单字段更新为null------
+ Map esFieldData = new LinkedHashMap<>();
+ for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null);
+ }
+ joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
+ } else {
+ // ------关联子表简单字段更新------
+ subTableSimpleFieldOperation(config, dml, data, null, tableItem);
+ }
+ } else {
+ // ------关联子表复杂字段更新 执行全sql更新es------
+ wholeSqlOperation(config, dml, data, null, tableItem);
+ }
+ }
+ }
+ }
+
+ /**
+ * 单表简单字段insert
+ *
+ * @param config es配置
+ * @param dml dml信息
+ * @param data 单行dml数据
+ */
+ private void singleTableSimpleFiledInsert(ESSyncConfig config, Dml dml, Map data) {
+ ESMapping mapping = config.getEsMapping();
+ Map esFieldData = new LinkedHashMap<>();
+ Object idVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Single table insert to es index, destination:{}, table: {}, index: {}, id: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ idVal);
+ }
+ esTemplate.insert(mapping, idVal, esFieldData);
+ }
+
+ /**
+ * 主表(单表)复杂字段insert
+ *
+ * @param config es配置
+ * @param dml dml信息
+ * @param data 单行dml数据
+ */
+ private void mainTableInsert(ESSyncConfig config, Dml dml, Map data) {
+ ESMapping mapping = config.getEsMapping();
+ String sql = mapping.getSql();
+ String condition = ESSyncUtil.pkConditionSql(mapping, data);
+ sql = ESSyncUtil.appendCondition(sql, condition);
+ DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+ if (logger.isTraceEnabled()) {
+ logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ sql.replace("\n", " "));
+ }
+ Util.sqlRS(ds, sql, rs -> {
+ try {
+ while (rs.next()) {
+ Map esFieldData = new LinkedHashMap<>();
+ Object idVal = esTemplate.getESDataFromRS(mapping, rs, esFieldData);
+
+ if (logger.isTraceEnabled()) {
+ logger.trace(
+ "Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ idVal);
+ }
+ esTemplate.insert(mapping, idVal, esFieldData);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return 0;
+ });
+ }
+
+ private void mainTableDelete(ESSyncConfig config, Dml dml, Map data) {
+ ESMapping mapping = config.getEsMapping();
+ String sql = mapping.getSql();
+ String condition = ESSyncUtil.pkConditionSql(mapping, data);
+ sql = ESSyncUtil.appendCondition(sql, condition);
+ DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+ if (logger.isTraceEnabled()) {
+ logger.trace("Main table delete es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ sql.replace("\n", " "));
+ }
+ Util.sqlRS(ds, sql, rs -> {
+ try {
+ Map esFieldData = null;
+ if (mapping.getPk() != null) {
+ esFieldData = new LinkedHashMap<>();
+ esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
+ esFieldData.remove(mapping.getPk());
+ for (String key : esFieldData.keySet()) {
+ esFieldData.put(Util.cleanColumn(key), null);
+ }
+ }
+ while (rs.next()) {
+ Object idVal = esTemplate.getIdValFromRS(mapping, rs);
+
+ if (logger.isTraceEnabled()) {
+ logger.trace(
+ "Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index(),
+ idVal);
+ }
+ esTemplate.delete(mapping, idVal, esFieldData);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return 0;
+ });
+ }
+
+ /**
+ * 关联表主表简单字段operation
+ *
+ * @param config es配置
+ * @param dml dml信息
+ * @param data 单行dml数据
+ * @param tableItem 当前表配置
+ */
+ private void joinTableSimpleFieldOperation(ESSyncConfig config, Dml dml, Map data,
+ TableItem tableItem, Map esFieldData) {
+ ESMapping mapping = config.getEsMapping();
+
+ Map paramsTmp = new LinkedHashMap<>();
+ for (Map.Entry> entry : tableItem.getRelationTableFields().entrySet()) {
+ for (FieldItem fieldItem : entry.getValue()) {
+ if (fieldItem.getColumnItems().size() == 1) {
+ Object value = esTemplate.getValFromData(mapping,
+ data,
+ fieldItem.getFieldName(),
+ entry.getKey().getColumn().getColumnName());
+
+ String fieldName = fieldItem.getFieldName();
+ // 判断是否是主键
+ if (fieldName.equals(mapping.get_id())) {
+ fieldName = "_id";
+ }
+ paramsTmp.put(fieldName, value);
+ }
+ }
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.trace("Join table update es index by foreign key, destination:{}, table: {}, index: {}",
+ config.getDestination(),
+ dml.getTable(),
+ mapping.get_index());
+ }
+ esTemplate.updateByQuery(config, paramsTmp, esFieldData);
+ }
+
+ /**
+ * 关联子查询, 主表简单字段operation
+ *
+ * @param config es配置
+ * @param dml dml信息
+ * @param data 单行dml数据
+ * @param old 单行old数据
+ * @param tableItem 当前表配置
+ */
+ private void subTableSimpleFieldOperation(ESSyncConfig config, Dml dml, Map data,
+ Map old, TableItem tableItem) {
+ ESMapping mapping = config.getEsMapping();
+
+ MySqlSelectQueryBlock queryBlock = SqlParser.parseSQLSelectQueryBlock(tableItem.getSubQuerySql());
+ StringBuilder sql = new StringBuilder();
+ sql.append("SELECT ")
+ .append(SqlParser.parse4SQLSelectItem(queryBlock))
+ .append(" FROM ")
+ .append(SqlParser.parse4FromTableSource(queryBlock));
+
+ String whereSql = SqlParser.parse4WhereItem(queryBlock);
+ if (whereSql != null) {
+ sql.append(" WHERE ").append(whereSql);
+ } else {
+ sql.append(" WHERE 1=1 ");
+ }
+
+ List