CanalAdapter两张表join字段同名bug

环境信息

canal version canal-adapter1.1.3-alpha-2 mysql version mysql

问题描述

在两张表join的时候，如果两张表有同名列名，在update的时候，会错误更新两个字段 表A：school_phase 有一个字段是phase_code 表B：school 有一个字段也是phase_code 表A和表B left join的时候 把表a的phase_code 命名为 school_phase_code，表b的还是phase_code 但是测试在测试的过程发现，更新表A的phase_code 会也把es索引里的phase_code也更新了

现执行情况

这边看了下代码是

public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld, Map<String, Object> esFieldData)

这个函数有问题，里面

if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); }

没有考虑表的关系，我这边是重新了这段逻辑，在判断更新的时候，加了一段判断是否是同一张表

//处理两张表字段同名的bug String aliceTableName = mapping.getSchemaItem().getAliasesName(tableName); if(StringUtils.isNotEmpty(aliceTableName)){

       for ( Map.Entry<String, FieldItem> fieldItemEntry : schemaItem.getSelectFields().entrySet()) {
           FieldItem fieldItem =fieldItemEntry.getValue();
           String ownTable = fieldItem.getOwner();
           String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
           String aliceColumnName = fieldItemEntry.getKey();

           if (fieldItem.getFieldName().equals(idFieldName)) {
               resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
           }

           if (StringUtils.equals(ownTable,aliceTableName) && dmlOld.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) {
               esFieldData.put(fieldItem.getFieldName(),
                       getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
           }
       }
   }else{
       for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
           String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();

           if (fieldItem.getFieldName().equals(idFieldName)) {
               resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
           }

           if (dmlOld.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) {
               esFieldData.put(fieldItem.getFieldName(),
                       getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
           }
       }
   }

    return resultIdVal;
}

原提问者GitHub用户house0128