mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
fixed testcase
This commit is contained in:
agapple
49
client-adapter/escore/pom.xml
Normal file
49
client-adapter/escore/pom.xml
Normal file
@@ -0,0 +1,49 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<artifactId>canal.client-adapter</artifactId>
|
||||
<groupId>com.alibaba.otter</groupId>
|
||||
<version>1.1.5-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<groupId>com.alibaba.otter</groupId>
|
||||
<artifactId>client-adapter.escore</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>canal client adapter escore module for otter ${project.version}</name>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.alibaba.otter</groupId>
|
||||
<artifactId>client-adapter.common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.alibaba.fastsql</groupId>
|
||||
<artifactId>fastsql</artifactId>
|
||||
<version>2.0.0_preview_855</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>4.12</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>mysql</groupId>
|
||||
<artifactId>mysql-connector-java</artifactId>
|
||||
<version>5.1.40</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.h2database</groupId>
|
||||
<artifactId>h2</artifactId>
|
||||
<version>1.4.192</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -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<String, ESSyncConfig> esSyncConfig = new ConcurrentHashMap<>(); // 文件名对应配置
|
||||
protected Map<String, Map<String, ESSyncConfig>> dbTableEsSyncConfig = new ConcurrentHashMap<>(); // schema-table对应配置
|
||||
|
||||
protected ESTemplate esTemplate;
|
||||
|
||||
protected ESSyncService esSyncService;
|
||||
|
||||
protected ESConfigMonitor esConfigMonitor;
|
||||
|
||||
protected Properties envProperties;
|
||||
|
||||
public ESSyncService getEsSyncService() {
|
||||
return esSyncService;
|
||||
}
|
||||
|
||||
public Map<String, ESSyncConfig> getEsSyncConfig() {
|
||||
return esSyncConfig;
|
||||
}
|
||||
|
||||
public Map<String, Map<String, ESSyncConfig>> getDbTableEsSyncConfig() {
|
||||
return dbTableEsSyncConfig;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void init(OuterAdapterConfig configuration, Properties envProperties) {
|
||||
try {
|
||||
this.envProperties = envProperties;
|
||||
Map<String, ESSyncConfig> 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<String, ESSyncConfig> 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<Dml> 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<String, ESSyncConfig> 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<String> params);
|
||||
|
||||
@Override
|
||||
public abstract Map<String, Object> 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<String, ESSyncConfig> 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);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<String, RelationMapping> relations = new LinkedHashMap<>();
|
||||
private String sql;
|
||||
// 对象字段, 例: objFields:
|
||||
// - _labels: array:;
|
||||
private Map<String, String> objFields = new LinkedHashMap<>();
|
||||
private List<String> 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<String, String> getObjFields() {
|
||||
return objFields;
|
||||
}
|
||||
|
||||
public void setObjFields(Map<String, String> objFields) {
|
||||
this.objFields = objFields;
|
||||
}
|
||||
|
||||
public List<String> getSkips() {
|
||||
return skips;
|
||||
}
|
||||
|
||||
public void setSkips(List<String> skips) {
|
||||
this.skips = skips;
|
||||
}
|
||||
|
||||
public Map<String, RelationMapping> getRelations() {
|
||||
return relations;
|
||||
}
|
||||
|
||||
public void setRelations(Map<String, RelationMapping> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String, ESSyncConfig> load(Properties envProperties) {
|
||||
logger.info("## Start loading es mapping config ... ");
|
||||
|
||||
Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>();
|
||||
|
||||
String esv = envProperties.getProperty("es.version");
|
||||
Map<String, String> 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;
|
||||
}
|
||||
}
|
||||
@@ -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<String, TableItem> aliasTableItems = new LinkedHashMap<>(); // 别名对应表名
|
||||
private Map<String, FieldItem> selectFields = new LinkedHashMap<>(); // 查询字段
|
||||
private String sql;
|
||||
|
||||
private volatile Map<String, List<TableItem>> tableItemAliases;
|
||||
private volatile Map<String, List<FieldItem>> columnFields;
|
||||
private volatile Boolean allFieldsSimple;
|
||||
|
||||
public void init() {
|
||||
this.getTableItemAliases();
|
||||
this.getColumnFields();
|
||||
this.isAllFieldsSimple();
|
||||
aliasTableItems.values().forEach(tableItem -> {
|
||||
tableItem.getRelationTableFields();
|
||||
tableItem.getRelationSelectFieldItems();
|
||||
});
|
||||
}
|
||||
|
||||
public Map<String, TableItem> getAliasTableItems() {
|
||||
return aliasTableItems;
|
||||
}
|
||||
|
||||
public void setAliasTableItems(Map<String, TableItem> aliasTableItems) {
|
||||
this.aliasTableItems = aliasTableItems;
|
||||
}
|
||||
|
||||
public String getSql() {
|
||||
return sql;
|
||||
}
|
||||
|
||||
public void setSql(String sql) {
|
||||
this.sql = sql;
|
||||
}
|
||||
|
||||
public Map<String, FieldItem> getSelectFields() {
|
||||
return selectFields;
|
||||
}
|
||||
|
||||
public void setSelectFields(Map<String, FieldItem> selectFields) {
|
||||
this.selectFields = selectFields;
|
||||
}
|
||||
|
||||
public String toSql() {
|
||||
// todo
|
||||
return null;
|
||||
}
|
||||
|
||||
public Map<String, List<TableItem>> getTableItemAliases() {
|
||||
if (tableItemAliases == null) {
|
||||
synchronized (SchemaItem.class) {
|
||||
if (tableItemAliases == null) {
|
||||
tableItemAliases = new LinkedHashMap<>();
|
||||
aliasTableItems.forEach((alias, tableItem) -> {
|
||||
List<TableItem> aliases = tableItemAliases
|
||||
.computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>());
|
||||
aliases.add(tableItem);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
return tableItemAliases;
|
||||
}
|
||||
|
||||
public Map<String, List<FieldItem>> 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<FieldItem> fieldItems = columnFields.computeIfAbsent(
|
||||
columnItem.getOwner() + "." + columnItem.getColumnName(),
|
||||
k -> new ArrayList<>());
|
||||
fieldItems.add(fieldItem);
|
||||
}
|
||||
// } else {
|
||||
// tableItem.getSubQueryFields().forEach(subQueryField -> {
|
||||
// List<FieldItem> 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<FieldItem> subQueryFields = new ArrayList<>();
|
||||
private List<RelationFieldsPair> relationFields = new ArrayList<>();
|
||||
|
||||
private boolean main;
|
||||
private boolean subQuery;
|
||||
|
||||
private volatile Map<FieldItem, List<FieldItem>> relationTableFields; // 当前表关联条件字段对应主表查询字段
|
||||
private volatile List<FieldItem> 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<FieldItem> getSubQueryFields() {
|
||||
return subQueryFields;
|
||||
}
|
||||
|
||||
public void setSubQueryFields(List<FieldItem> subQueryFields) {
|
||||
this.subQueryFields = subQueryFields;
|
||||
}
|
||||
|
||||
public List<RelationFieldsPair> getRelationFields() {
|
||||
return relationFields;
|
||||
}
|
||||
|
||||
public void setRelationFields(List<RelationFieldsPair> relationFields) {
|
||||
this.relationFields = relationFields;
|
||||
}
|
||||
|
||||
public Map<FieldItem, List<FieldItem>> 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<FieldItem> 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<FieldItem> getRelationSelectFieldItems() {
|
||||
if (relationSelectFieldItems == null) {
|
||||
synchronized (SchemaItem.class) {
|
||||
if (relationSelectFieldItems == null) {
|
||||
List<FieldItem> 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<ColumnItem> columnItems = new ArrayList<>();
|
||||
private List<String> 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<ColumnItem> getColumnItems() {
|
||||
return columnItems;
|
||||
}
|
||||
|
||||
public void setColumnItems(List<ColumnItem> 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<String> getOwners() {
|
||||
return owners;
|
||||
}
|
||||
|
||||
public void setOwners(List<String> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<TableItem> tableItems = new ArrayList<>();
|
||||
SqlParser.visitSelectTable(schemaItem, sqlTableSource, tableItems, null);
|
||||
tableItems.forEach(tableItem -> schemaItem.getAliasTableItems().put(tableItem.getAlias(), tableItem));
|
||||
|
||||
List<FieldItem> 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<FieldItem> 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<TableItem> 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<SQLSelectItem> 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;
|
||||
}
|
||||
}
|
||||
@@ -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<String, ESSyncConfig> configMap : esAdapter.getDbTableEsSyncConfig().values()) {
|
||||
if (configMap != null) {
|
||||
configMap.remove(file.getName());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<ESSyncConfig> 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<Map<String, Object>> dataList = dml.getData();
|
||||
if (dataList == null || dataList.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
|
||||
for (Map<String, Object> 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<String, Object> 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<Map<String, Object>> dataList = dml.getData();
|
||||
List<Map<String, Object>> oldList = dml.getOld();
|
||||
if (dataList == null || dataList.isEmpty() || oldList == null || oldList.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
|
||||
int i = 0;
|
||||
for (Map<String, Object> data : dataList) {
|
||||
Map<String, Object> 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<FieldItem> 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<String, Object> 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<Map<String, Object>> dataList = dml.getData();
|
||||
if (dataList == null || dataList.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
|
||||
|
||||
for (Map<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> data) {
|
||||
ESMapping mapping = config.getEsMapping();
|
||||
Map<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> data,
|
||||
TableItem tableItem, Map<String, Object> esFieldData) {
|
||||
ESMapping mapping = config.getEsMapping();
|
||||
|
||||
Map<String, Object> paramsTmp = new LinkedHashMap<>();
|
||||
for (Map.Entry<FieldItem, List<FieldItem>> 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<String, Object> data,
|
||||
Map<String, Object> 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<Object> values = new ArrayList<>();
|
||||
|
||||
for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) {
|
||||
String columnName = fkFieldItem.getColumn().getColumnName();
|
||||
Object value = esTemplate.getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName);
|
||||
sql.append(" AND ").append(columnName).append("=? ");
|
||||
values.add(value);
|
||||
}
|
||||
|
||||
String groupSql = SqlParser.parse4GroupBy(queryBlock);
|
||||
if (groupSql != null) {
|
||||
sql.append(groupSql);
|
||||
}
|
||||
|
||||
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
|
||||
config.getDestination(),
|
||||
dml.getTable(),
|
||||
mapping.get_index(),
|
||||
sql.toString().replace("\n", " "));
|
||||
}
|
||||
Util.sqlRS(ds, sql.toString(), values, rs -> {
|
||||
try {
|
||||
while (rs.next()) {
|
||||
Map<String, Object> esFieldData = new LinkedHashMap<>();
|
||||
|
||||
for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
|
||||
if (old != null) {
|
||||
out: for (FieldItem fieldItem1 : tableItem.getSubQueryFields()) {
|
||||
for (ColumnItem columnItem0 : fieldItem.getColumnItems()) {
|
||||
if (fieldItem1.getFieldName().equals(columnItem0.getColumnName()))
|
||||
for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
|
||||
if (old.containsKey(columnItem.getColumnName())) {
|
||||
Object val = esTemplate.getValFromRS(mapping,
|
||||
rs,
|
||||
fieldItem.getFieldName(),
|
||||
fieldItem.getColumn().getColumnName());
|
||||
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
|
||||
break out;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
Object val = esTemplate.getValFromRS(mapping,
|
||||
rs,
|
||||
fieldItem.getFieldName(),
|
||||
fieldItem.getColumn().getColumnName());
|
||||
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
|
||||
}
|
||||
}
|
||||
|
||||
Map<String, Object> paramsTmp = new LinkedHashMap<>();
|
||||
for (Map.Entry<FieldItem, List<FieldItem>> entry : tableItem.getRelationTableFields().entrySet()) {
|
||||
for (FieldItem fieldItem : entry.getValue()) {
|
||||
if (fieldItem.getColumnItems().size() == 1) {
|
||||
Object value = esTemplate.getValFromRS(mapping,
|
||||
rs,
|
||||
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 query sql, destination:{}, table: {}, index: {}",
|
||||
config.getDestination(),
|
||||
dml.getTable(),
|
||||
mapping.get_index());
|
||||
}
|
||||
esTemplate.updateByQuery(config, paramsTmp, esFieldData);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return 0;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 关联(子查询), 主表复杂字段operation, 全sql执行
|
||||
*
|
||||
* @param config es配置
|
||||
* @param dml dml信息
|
||||
* @param data 单行dml数据
|
||||
* @param tableItem 当前表配置
|
||||
*/
|
||||
private void wholeSqlOperation(ESSyncConfig config, Dml dml, Map<String, Object> data, Map<String, Object> old,
|
||||
TableItem tableItem) {
|
||||
ESMapping mapping = config.getEsMapping();
|
||||
// 防止最后出现groupby 导致sql解析异常
|
||||
String[] sqlSplit = mapping.getSql().split("GROUP\\ BY(?!(.*)ON)");
|
||||
String sqlNoWhere = sqlSplit[0];
|
||||
|
||||
String sqlGroupBy = "";
|
||||
|
||||
if (sqlSplit.length > 1) {
|
||||
sqlGroupBy = "GROUP BY " + sqlSplit[1];
|
||||
}
|
||||
|
||||
StringBuilder sql = new StringBuilder(sqlNoWhere + " WHERE ");
|
||||
|
||||
for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) {
|
||||
String columnName = fkFieldItem.getColumn().getColumnName();
|
||||
Object value = esTemplate.getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName);
|
||||
ESSyncUtil.appendCondition(sql, value, tableItem.getAlias(), columnName);
|
||||
}
|
||||
int len = sql.length();
|
||||
sql.delete(len - 5, len);
|
||||
sql.append(sqlGroupBy);
|
||||
|
||||
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("Join table update es index by query whole sql, destination:{}, table: {}, index: {}, sql: {}",
|
||||
config.getDestination(),
|
||||
dml.getTable(),
|
||||
mapping.get_index(),
|
||||
sql.toString().replace("\n", " "));
|
||||
}
|
||||
Util.sqlRS(ds, sql.toString(), rs -> {
|
||||
try {
|
||||
while (rs.next()) {
|
||||
Map<String, Object> esFieldData = new LinkedHashMap<>();
|
||||
for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
|
||||
if (old != null) {
|
||||
// 从表子查询
|
||||
out: for (FieldItem fieldItem1 : tableItem.getSubQueryFields()) {
|
||||
for (ColumnItem columnItem0 : fieldItem.getColumnItems()) {
|
||||
if (fieldItem1.getFieldName().equals(columnItem0.getColumnName()))
|
||||
for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
|
||||
if (old.containsKey(columnItem.getColumnName())) {
|
||||
Object val = esTemplate.getValFromRS(mapping,
|
||||
rs,
|
||||
fieldItem.getFieldName(),
|
||||
fieldItem.getFieldName());
|
||||
esFieldData.put(fieldItem.getFieldName(), val);
|
||||
break out;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// 从表非子查询
|
||||
for (FieldItem fieldItem1 : tableItem.getRelationSelectFieldItems()) {
|
||||
if (fieldItem1.equals(fieldItem)) {
|
||||
for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
|
||||
if (old.containsKey(columnItem.getColumnName())) {
|
||||
Object val = esTemplate.getValFromRS(mapping,
|
||||
rs,
|
||||
fieldItem.getFieldName(),
|
||||
fieldItem.getFieldName());
|
||||
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
Object val = esTemplate
|
||||
.getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
|
||||
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
|
||||
}
|
||||
}
|
||||
|
||||
Map<String, Object> paramsTmp = new LinkedHashMap<>();
|
||||
for (Map.Entry<FieldItem, List<FieldItem>> entry : tableItem.getRelationTableFields().entrySet()) {
|
||||
for (FieldItem fieldItem : entry.getValue()) {
|
||||
Object value = esTemplate
|
||||
.getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
|
||||
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 query whole sql, destination:{}, table: {}, index: {}",
|
||||
config.getDestination(),
|
||||
dml.getTable(),
|
||||
mapping.get_index());
|
||||
}
|
||||
esTemplate.updateByQuery(config, paramsTmp, esFieldData);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return 0;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 单表简单字段update
|
||||
*
|
||||
* @param config es配置
|
||||
* @param dml dml信息
|
||||
* @param data 单行data数据
|
||||
* @param old 单行old数据
|
||||
*/
|
||||
private void singleTableSimpleFiledUpdate(ESSyncConfig config, Dml dml, Map<String, Object> data,
|
||||
Map<String, Object> old) {
|
||||
ESMapping mapping = config.getEsMapping();
|
||||
Map<String, Object> esFieldData = new LinkedHashMap<>();
|
||||
|
||||
Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData);
|
||||
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",
|
||||
config.getDestination(),
|
||||
dml.getTable(),
|
||||
mapping.get_index(),
|
||||
idVal);
|
||||
}
|
||||
esTemplate.update(mapping, idVal, esFieldData);
|
||||
}
|
||||
|
||||
/**
|
||||
* 主表(单表)复杂字段update
|
||||
*
|
||||
* @param config es配置
|
||||
* @param dml dml信息
|
||||
* @param data 单行dml数据
|
||||
*/
|
||||
private void mainTableUpdate(ESSyncConfig config, Dml dml, Map<String, Object> data, Map<String, Object> old) {
|
||||
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 update 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<String, Object> esFieldData = new LinkedHashMap<>();
|
||||
Object idVal = esTemplate.getESDataFromRS(mapping, rs, old, esFieldData);
|
||||
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace(
|
||||
"Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
|
||||
config.getDestination(),
|
||||
dml.getTable(),
|
||||
mapping.get_index(),
|
||||
idVal);
|
||||
}
|
||||
esTemplate.update(mapping, idVal, esFieldData);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return 0;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 提交批次
|
||||
*/
|
||||
public void commit() {
|
||||
esTemplate.commit();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package com.alibaba.otter.canal.client.adapter.es.core.support;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
public interface ESBulkRequest {
|
||||
|
||||
void resetBulk();
|
||||
|
||||
ESBulkRequest add(ESIndexRequest esIndexRequest);
|
||||
|
||||
ESBulkRequest add(ESUpdateRequest esUpdateRequest);
|
||||
|
||||
ESBulkRequest add(ESDeleteRequest esDeleteRequest);
|
||||
|
||||
int numberOfActions();
|
||||
|
||||
ESBulkResponse bulk();
|
||||
|
||||
interface ESIndexRequest {
|
||||
|
||||
ESIndexRequest setSource(Map<String, ?> source);
|
||||
|
||||
ESIndexRequest setRouting(String routing);
|
||||
}
|
||||
|
||||
interface ESUpdateRequest {
|
||||
|
||||
ESUpdateRequest setDoc(Map source);
|
||||
|
||||
ESUpdateRequest setDocAsUpsert(boolean shouldUpsertDoc);
|
||||
|
||||
ESUpdateRequest setRouting(String routing);
|
||||
}
|
||||
|
||||
interface ESDeleteRequest {
|
||||
}
|
||||
|
||||
interface ESBulkResponse {
|
||||
boolean hasFailures();
|
||||
|
||||
void processFailBulkResponse(String errorMsg);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,303 @@
|
||||
package com.alibaba.otter.canal.client.adapter.es.core.support;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.sql.Blob;
|
||||
import java.sql.SQLException;
|
||||
import java.util.*;
|
||||
|
||||
import org.apache.commons.codec.binary.Base64;
|
||||
import org.joda.time.DateTime;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
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.TableItem;
|
||||
import com.alibaba.otter.canal.client.adapter.support.Util;
|
||||
|
||||
/**
|
||||
* ES 同步工具同类
|
||||
*
|
||||
* @author rewerma 2018-11-01
|
||||
* @version 1.0.0
|
||||
*/
|
||||
public class ESSyncUtil {
|
||||
|
||||
private static Logger logger = LoggerFactory.getLogger(ESSyncUtil.class);
|
||||
|
||||
public static Object convertToEsObj(Object val, String fieldInfo) {
|
||||
if (val == null) {
|
||||
return null;
|
||||
}
|
||||
if (fieldInfo.startsWith("array:")) {
|
||||
String separator = fieldInfo.substring("array:".length()).trim();
|
||||
String[] values = val.toString().split(separator);
|
||||
return Arrays.asList(values);
|
||||
} else if (fieldInfo.startsWith("object")) {
|
||||
if (val instanceof String){
|
||||
return JSON.parse(val.toString());
|
||||
}
|
||||
return JSON.parse(new String((byte[])val));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 类型转换为Mapping中对应的类型
|
||||
*/
|
||||
public static Object typeConvert(Object val, String esType) {
|
||||
if (val == null) {
|
||||
return null;
|
||||
}
|
||||
if (esType == null) {
|
||||
return val;
|
||||
}
|
||||
Object res = null;
|
||||
if ("integer".equals(esType)) {
|
||||
if (val instanceof Number) {
|
||||
res = ((Number) val).intValue();
|
||||
} else {
|
||||
res = Integer.parseInt(val.toString());
|
||||
}
|
||||
} else if ("long".equals(esType)) {
|
||||
if (val instanceof Number) {
|
||||
res = ((Number) val).longValue();
|
||||
} else {
|
||||
res = Long.parseLong(val.toString());
|
||||
}
|
||||
} else if ("short".equals(esType)) {
|
||||
if (val instanceof Number) {
|
||||
res = ((Number) val).shortValue();
|
||||
} else {
|
||||
res = Short.parseShort(val.toString());
|
||||
}
|
||||
} else if ("byte".equals(esType)) {
|
||||
if (val instanceof Number) {
|
||||
res = ((Number) val).byteValue();
|
||||
} else {
|
||||
res = Byte.parseByte(val.toString());
|
||||
}
|
||||
} else if ("double".equals(esType)) {
|
||||
if (val instanceof Number) {
|
||||
res = ((Number) val).doubleValue();
|
||||
} else {
|
||||
res = Double.parseDouble(val.toString());
|
||||
}
|
||||
} else if ("float".equals(esType) || "half_float".equals(esType) || "scaled_float".equals(esType)) {
|
||||
if (val instanceof Number) {
|
||||
res = ((Number) val).floatValue();
|
||||
} else {
|
||||
res = Float.parseFloat(val.toString());
|
||||
}
|
||||
} else if ("boolean".equals(esType)) {
|
||||
if (val instanceof Boolean) {
|
||||
res = val;
|
||||
} else if (val instanceof Number) {
|
||||
int v = ((Number) val).intValue();
|
||||
res = v != 0;
|
||||
} else {
|
||||
res = Boolean.parseBoolean(val.toString());
|
||||
}
|
||||
} else if ("date".equals(esType)) {
|
||||
if (val instanceof java.sql.Time) {
|
||||
DateTime dateTime = new DateTime(((java.sql.Time) val).getTime());
|
||||
if (dateTime.getMillisOfSecond() != 0) {
|
||||
res = dateTime.toString("HH:mm:ss.SSS");
|
||||
} else {
|
||||
res = dateTime.toString("HH:mm:ss");
|
||||
}
|
||||
} else if (val instanceof java.sql.Timestamp) {
|
||||
DateTime dateTime = new DateTime(((java.sql.Timestamp) val).getTime());
|
||||
if (dateTime.getMillisOfSecond() != 0) {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
|
||||
} else {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
|
||||
}
|
||||
} else if (val instanceof java.sql.Date || val instanceof Date) {
|
||||
DateTime dateTime;
|
||||
if (val instanceof java.sql.Date) {
|
||||
dateTime = new DateTime(((java.sql.Date) val).getTime());
|
||||
} else {
|
||||
dateTime = new DateTime(((Date) val).getTime());
|
||||
}
|
||||
if (dateTime.getHourOfDay() == 0 && dateTime.getMinuteOfHour() == 0 && dateTime.getSecondOfMinute() == 0
|
||||
&& dateTime.getMillisOfSecond() == 0) {
|
||||
res = dateTime.toString("yyyy-MM-dd");
|
||||
} else {
|
||||
if (dateTime.getMillisOfSecond() != 0) {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
|
||||
} else {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
|
||||
}
|
||||
}
|
||||
} else if (val instanceof Long) {
|
||||
DateTime dateTime = new DateTime(((Long) val).longValue());
|
||||
if (dateTime.getHourOfDay() == 0 && dateTime.getMinuteOfHour() == 0 && dateTime.getSecondOfMinute() == 0
|
||||
&& dateTime.getMillisOfSecond() == 0) {
|
||||
res = dateTime.toString("yyyy-MM-dd");
|
||||
} else if (dateTime.getMillisOfSecond() != 0) {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
|
||||
} else {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
|
||||
}
|
||||
} else if (val instanceof String) {
|
||||
String v = ((String) val).trim();
|
||||
if (v.length() > 18 && v.charAt(4) == '-' && v.charAt(7) == '-' && v.charAt(10) == ' '
|
||||
&& v.charAt(13) == ':' && v.charAt(16) == ':') {
|
||||
String dt = v.substring(0, 10) + "T" + v.substring(11);
|
||||
Date date = Util.parseDate(dt);
|
||||
if (date != null) {
|
||||
DateTime dateTime = new DateTime(date);
|
||||
if (dateTime.getMillisOfSecond() != 0) {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
|
||||
} else {
|
||||
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
|
||||
}
|
||||
}
|
||||
} else if (v.length() == 10 && v.charAt(4) == '-' && v.charAt(7) == '-') {
|
||||
Date date = Util.parseDate(v);
|
||||
if (date != null) {
|
||||
DateTime dateTime = new DateTime(date);
|
||||
res = dateTime.toString("yyyy-MM-dd");
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if ("binary".equals(esType)) {
|
||||
if (val instanceof byte[]) {
|
||||
Base64 base64 = new Base64();
|
||||
res = base64.encodeAsString((byte[]) val);
|
||||
} else if (val instanceof Blob) {
|
||||
byte[] b = blobToBytes((Blob) val);
|
||||
Base64 base64 = new Base64();
|
||||
res = base64.encodeAsString(b);
|
||||
} else if (val instanceof String) {
|
||||
// 对应canal中的单字节编码
|
||||
byte[] b = ((String) val).getBytes(StandardCharsets.ISO_8859_1);
|
||||
Base64 base64 = new Base64();
|
||||
res = base64.encodeAsString(b);
|
||||
}
|
||||
} else if ("geo_point".equals(esType)) {
|
||||
if (!(val instanceof String)) {
|
||||
logger.error("es type is geo_point, but source type is not String");
|
||||
return val;
|
||||
}
|
||||
|
||||
if (!((String) val).contains(",")) {
|
||||
logger.error("es type is geo_point, source value not contains ',' separator");
|
||||
return val;
|
||||
}
|
||||
|
||||
String[] point = ((String) val).split(",");
|
||||
Map<String, Double> location = new HashMap<>();
|
||||
location.put("lat", Double.valueOf(point[0].trim()));
|
||||
location.put("lon", Double.valueOf(point[1].trim()));
|
||||
return location;
|
||||
} else if ("array".equals(esType)) {
|
||||
if ("".equals(val.toString().trim())) {
|
||||
res = new ArrayList<>();
|
||||
} else {
|
||||
String value = val.toString();
|
||||
String separator = ",";
|
||||
if (!value.contains(",")) {
|
||||
if (value.contains(";")) {
|
||||
separator = ";";
|
||||
} else if (value.contains("|")) {
|
||||
separator = "\\|";
|
||||
} else if (value.contains("-")) {
|
||||
separator = "-";
|
||||
}
|
||||
}
|
||||
String[] values = value.split(separator);
|
||||
return Arrays.asList(values);
|
||||
}
|
||||
} else if ("object".equals(esType)) {
|
||||
if ("".equals(val.toString().trim())) {
|
||||
res = new HashMap<>();
|
||||
} else {
|
||||
res = JSON.parseObject(val.toString(), Map.class);
|
||||
}
|
||||
} else {
|
||||
// 其他类全以字符串处理
|
||||
res = val.toString();
|
||||
}
|
||||
|
||||
return res;
|
||||
}
|
||||
|
||||
/**
|
||||
* Blob转byte[]
|
||||
*/
|
||||
private static byte[] blobToBytes(Blob blob) {
|
||||
try (InputStream is = blob.getBinaryStream()) {
|
||||
byte[] b = new byte[(int) blob.length()];
|
||||
if (is.read(b) != -1) {
|
||||
return b;
|
||||
} else {
|
||||
return new byte[0];
|
||||
}
|
||||
} catch (IOException | SQLException e) {
|
||||
logger.error(e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 拼接主键条件
|
||||
*
|
||||
* @param mapping
|
||||
* @param data
|
||||
* @return
|
||||
*/
|
||||
public static String pkConditionSql(ESMapping mapping, Map<String, Object> data) {
|
||||
Set<ColumnItem> idColumns = new LinkedHashSet<>();
|
||||
SchemaItem schemaItem = mapping.getSchemaItem();
|
||||
|
||||
TableItem mainTable = schemaItem.getMainTable();
|
||||
|
||||
for (ColumnItem idColumnItem : schemaItem.getIdFieldItem(mapping).getColumnItems()) {
|
||||
if ((mainTable.getAlias() == null && idColumnItem.getOwner() == null)
|
||||
|| (mainTable.getAlias() != null && mainTable.getAlias().equals(idColumnItem.getOwner()))) {
|
||||
idColumns.add(idColumnItem);
|
||||
}
|
||||
}
|
||||
|
||||
if (idColumns.isEmpty()) {
|
||||
throw new RuntimeException("Not found primary key field in main table");
|
||||
}
|
||||
|
||||
// 拼接condition
|
||||
StringBuilder condition = new StringBuilder(" ");
|
||||
for (ColumnItem idColumn : idColumns) {
|
||||
Object idVal = data.get(idColumn.getColumnName());
|
||||
if (mainTable.getAlias() != null) condition.append(mainTable.getAlias()).append(".");
|
||||
condition.append(idColumn.getColumnName()).append("=");
|
||||
if (idVal instanceof String) {
|
||||
condition.append("'").append(idVal).append("' AND ");
|
||||
} else {
|
||||
condition.append(idVal).append(" AND ");
|
||||
}
|
||||
}
|
||||
|
||||
if (condition.toString().endsWith("AND ")) {
|
||||
int len2 = condition.length();
|
||||
condition.delete(len2 - 4, len2);
|
||||
}
|
||||
return condition.toString();
|
||||
}
|
||||
|
||||
public static String appendCondition(String sql, String condition) {
|
||||
return sql + " WHERE " + condition + " ";
|
||||
}
|
||||
|
||||
public static void appendCondition(StringBuilder sql, Object value, String owner, String columnName) {
|
||||
if (value instanceof String) {
|
||||
sql.append(owner).append(".").append(columnName).append("='").append(value).append("' AND ");
|
||||
} else {
|
||||
sql.append(owner).append(".").append(columnName).append("=").append(value).append(" AND ");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package com.alibaba.otter.canal.client.adapter.es.core.support;
|
||||
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.util.Map;
|
||||
|
||||
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
|
||||
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
|
||||
|
||||
public interface ESTemplate {
|
||||
|
||||
/**
|
||||
* 插入数据
|
||||
*
|
||||
* @param mapping 配置对象
|
||||
* @param pkVal 主键值
|
||||
* @param esFieldData 数据Map
|
||||
*/
|
||||
void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData);
|
||||
|
||||
/**
|
||||
* 根据主键更新数据
|
||||
*
|
||||
* @param mapping 配置对象
|
||||
* @param pkVal 主键值
|
||||
* @param esFieldData 数据Map
|
||||
*/
|
||||
void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData);
|
||||
|
||||
/**
|
||||
* update by query
|
||||
*
|
||||
* @param config 配置对象
|
||||
* @param paramsTmp sql查询条件
|
||||
* @param esFieldData 数据Map
|
||||
*/
|
||||
void updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData);
|
||||
|
||||
/**
|
||||
* 通过主键删除数据
|
||||
*
|
||||
* @param mapping 配置对象
|
||||
* @param pkVal 主键值
|
||||
* @param esFieldData 数据Map
|
||||
*/
|
||||
void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData);
|
||||
|
||||
/**
|
||||
* 提交批次
|
||||
*/
|
||||
void commit();
|
||||
|
||||
Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
|
||||
String columnName) throws SQLException;
|
||||
|
||||
Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> esFieldData) throws SQLException;
|
||||
|
||||
Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException;
|
||||
|
||||
Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
|
||||
Map<String, Object> esFieldData) throws SQLException;
|
||||
|
||||
Object getValFromData(ESMapping mapping, Map<String, Object> dmlData, String fieldName, String columnName);
|
||||
|
||||
Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData);
|
||||
|
||||
Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld,
|
||||
Map<String, Object> esFieldData);
|
||||
}
|
||||
Reference in New Issue
Block a user