删除测试完成

后续: 修改 删除外键的情况, 修改主键的情况适配

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java

@@ -38,6 +38,9 @@ public class ESSyncConfigLoader {
     public static synchronized void load() {
         logger.info("## Start loading mapping config ... ");
         Collection<String> configs = AdapterConfigs.get("es");
+        if (configs == null) {
+            return;
+        }
         for (String c : configs) {
             if (c == null) {
                 continue;
@@ -67,7 +70,7 @@ public class ESSyncConfigLoader {
                 }
                 Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
                 Matcher matcher = pattern.matcher(dataSource.getUrl());
-                if (!matcher.find()){
+                if (!matcher.find()) {
                     throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
                 }
                 String schema = matcher.group(2);

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java

@@ -99,7 +99,88 @@ public class ESSyncService {
     }
 
     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())) {
+                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);
+                    }
+                    boolean result = esTemplate.delete(mapping, idVal);
+                    if (!result) {
+                        logger.error("Main table delete es index error, destination:{}, table: {}, index: {}, id: {}",
+                            config.getDestination(),
+                            dml.getTable(),
+                            mapping.get_index(),
+                            idVal);
+                    }
+                } 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(fieldItem.getFieldName(), null);
+                        }
+                        joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
+                    } else {
+                        // ------关联子表简单字段更新------
+                        subTableSimpleFieldOperation(config, dml, data, null, tableItem);
+                    }
+                } else {
+                    // ------关联子表复杂字段更新 执行全sql更新es------
+                     jonTableWholeSqlOperation(config, dml, data, null, tableItem);
+                }
+            }
+        }
     }
 
     /**
@@ -293,7 +374,7 @@ public class ESSyncService {
                 mapping.get_index(),
                 idVal);
         }
-        boolean result = esTemplate.insert(config, idVal, esFieldData);
+        boolean result = esTemplate.insert(mapping, idVal, esFieldData);
         if (!result) {
             logger.error("Single table insert to es index error, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
@@ -337,7 +418,7 @@ public class ESSyncService {
                             mapping.get_index(),
                             idVal);
                     }
-                    boolean result = esTemplate.insert(config, idVal, esFieldData);
+                    boolean result = esTemplate.insert(mapping, idVal, esFieldData);
                     if (!result) {
                         logger.error(
                             "Main table insert to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
@@ -354,6 +435,49 @@ public class ESSyncService {
         });
     }
 
+    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", " "));
+        }
+        ESSyncUtil.sqlRS(ds, sql, rs -> {
+            try {
+                while (rs.next()) {
+                    Object idVal = esTemplate.getIdValFromRS(mapping, rs);
+
+                    if (logger.isTraceEnabled()) {
+                        logger.trace(
+                            "Main table delete ot es index by query sql, destination:{}, table: {}, index: {}, id: {}",
+                            config.getDestination(),
+                            dml.getTable(),
+                            mapping.get_index(),
+                            idVal);
+                    }
+                    boolean result = esTemplate.delete(mapping, idVal);
+                    if (!result) {
+                        logger.error(
+                            "Main table delete to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
+                            config.getDestination(),
+                            dml.getTable(),
+                            mapping.get_index(),
+                            idVal);
+                    }
+                }
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+            return 0;
+        });
+    }
+
     /**
      * 关联表主表简单字段operation
      *
@@ -626,15 +750,15 @@ public class ESSyncService {
         Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData);
 
         if (logger.isTraceEnabled()) {
-            logger.trace("Single table update ot es index, destination:{}, table: {}, index: {}, id: {}",
+            logger.trace("Main table update ot es index, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
                 dml.getTable(),
                 mapping.get_index(),
                 idVal);
         }
-        boolean result = esTemplate.update(config, idVal, esFieldData);
+        boolean result = esTemplate.update(mapping, idVal, esFieldData);
         if (!result) {
-            logger.error("Single table update to es index error, destination:{}, table: {}, index: {}, id: {}",
+            logger.error("Main table update to es index error, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
                 dml.getTable(),
                 mapping.get_index(),
@@ -676,7 +800,7 @@ public class ESSyncService {
                             mapping.get_index(),
                             idVal);
                     }
-                    boolean result = esTemplate.update(config, idVal, esFieldData);
+                    boolean result = esTemplate.update(mapping, idVal, esFieldData);
                     if (!result) {
                         logger.error(
                             "Main table update to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java

@@ -31,7 +31,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.CollectionUtils;
 
-import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
 import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
 import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem;
@@ -56,14 +55,13 @@ public class ESTemplate {
     /**
      * 插入数据
      * 
-     * @param esSyncConfig
+     * @param mapping
      * @param pkVal
      * @param esFieldData
      * @return
      */
-    public boolean insert(ESSyncConfig esSyncConfig, Object pkVal, Map<String, Object> esFieldData) {
+    public boolean insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
         BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
-        ESMapping mapping = esSyncConfig.getEsMapping();
         if (mapping.get_id() != null) {
             bulkRequestBuilder
                 .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
@@ -87,14 +85,14 @@ public class ESTemplate {
     /**
      * 根据主键更新数据
      * 
-     * @param esSyncConfig
+     * @param mapping
      * @param pkVal
      * @param esFieldData
      * @return
      */
-    public boolean update(ESSyncConfig esSyncConfig, Object pkVal, Map<String, Object> esFieldData) {
+    public boolean update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
         BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
-        append4Update(bulkRequestBuilder, esSyncConfig, pkVal, esFieldData);
+        append4Update(bulkRequestBuilder, mapping, pkVal, esFieldData);
         return commitBulkRequest(bulkRequestBuilder);
     }
 
@@ -110,9 +108,8 @@ public class ESTemplate {
         return commitBulkRequest(bulkRequestBuilder);
     }
 
-    public void append4Update(BulkRequestBuilder bulkRequestBuilder, ESSyncConfig esSyncConfig, Object pkVal,
+    public void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal,
                               Map<String, Object> esFieldData) {
-        ESMapping mapping = esSyncConfig.getEsMapping();
         if (mapping.get_id() != null) {
             bulkRequestBuilder
                 .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
@@ -223,13 +220,12 @@ public class ESTemplate {
     /**
      * 通过主键删除数据
      *
-     * @param esSyncConfig
+     * @param mapping
      * @param pkVal
      * @return
      */
-    public boolean delete(ESSyncConfig esSyncConfig, Object pkVal) {
+    public boolean delete(ESMapping mapping, Object pkVal) {
         BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
-        ESMapping mapping = esSyncConfig.getEsMapping();
         if (mapping.get_id() != null) {
             bulkRequestBuilder
                 .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString()));
@@ -308,6 +304,21 @@ public class ESTemplate {
         return resultIdVal;
     }
 
+    public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
+        SchemaItem schemaItem = mapping.getSchemaItem();
+        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        Object resultIdVal = null;
+        for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
+            Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
+
+            if (fieldItem.getFieldName().equals(idFieldName)) {
+                resultIdVal = value;
+                break;
+            }
+        }
+        return resultIdVal;
+    }
+
     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();

client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml

@@ -3,15 +3,12 @@ destination: example
 esMapping:
   _index: mytest_user
   _type: _doc
-  _id: id
+  _id: _id
 #  pk: id
-#  sql: "select a.id, concat(a.name,'_test') as name, a.role_id, b.name as role_name, c.labels from user a
-#        left join role b on a.role_id=b.id
-#        left join (select user_id, group_concat(label,',') as labels from user_label
-#        group by user_id) c on c.user_id=a.id"
-#  sql: "select id, name, c_time from user "
-#  sql: "select id, concat(name, '_') as name, c_time from user "
-  sql: "select a.id, a.name, a.role_id, b.role_name, a.c_time from user a
-        left join role b on b.id=a.role_id"
+  sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
+        a.c_time as _c_time, c.labels as _labels from user a
+        left join role b on b.id=a.role_id
+        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
+        group by user_id) c on c.user_id=a.id"
   commitBatch: 3000
-  etlCondition: "where a.c_time>='{0}' or b.c_time>='{0}' or c.c_time>='{0}'"
+  etlCondition: "where a.c_time>='{0}'"