From a2360efff9465243944b81af5f930c7a8e336054 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 19 Mar 2019 19:13:15 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=AF=B9sql=E5=AD=97?= =?UTF-8?q?=E6=AE=B5=20``=20=E7=9A=84=E6=94=AF=E6=8C=81=20fix=20#1559?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/es/service/ESEtlService.java | 23 +++++++++++-------- .../adapter/es/service/ESSyncService.java | 16 ++++++------- .../client/adapter/es/support/ESTemplate.java | 14 +++++++---- 3 files changed, 30 insertions(+), 23 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java index 646bcc47..8b1b2c5c 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java @@ -202,20 +202,23 @@ public class ESEtlService { long batchBegin = System.currentTimeMillis(); while (rs.next()) { Map esFieldData = new LinkedHashMap<>(); + Map idEsFieldData = new HashMap<>(); for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) { - // 如果是主键字段则不插入 - if (fieldItem.getFieldName().equals(mapping.get_id())) { - continue; - } - String fieldName = fieldItem.getFieldName(); if (mapping.getSkips().contains(fieldName)) { continue; } - Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); - esFieldData.put(fieldName, val); + // 如果是主键字段则不插入 + if (fieldItem.getFieldName().equals(mapping.get_id())) { + Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); + idEsFieldData.put(Util.cleanColumn(fieldName), val); + } else { + Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); + esFieldData.put(Util.cleanColumn(fieldName), val); + } + } if (!mapping.getRelations().isEmpty()) { @@ -241,13 +244,13 @@ public class ESEtlService { } } - esFieldData.put(relationField, relations); + esFieldData.put(Util.cleanColumn(relationField), relations); }); } Object idVal = null; if (mapping.get_id() != null) { - idVal = esFieldData.get(mapping.get_id()); + idVal = idEsFieldData.get(mapping.get_id()); } if (idVal != null) { @@ -271,7 +274,7 @@ public class ESEtlService { bulkRequestBuilder.add(indexRequestBuilder); } } else { - idVal = esFieldData.get(mapping.getPk()); + idVal = idEsFieldData.get(mapping.getPk()); SearchResponse response = transportClient.prepareSearch(mapping.get_index()) .setTypes(mapping.get_type()) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index 3d472a4d..ace00cba 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -167,7 +167,7 @@ public class ESSyncService { data, fieldItem.getFieldName(), fieldItem.getColumn().getColumnName()); - esFieldData.put(fieldItem.getFieldName(), value); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); @@ -296,7 +296,7 @@ public class ESSyncService { data, fieldItem.getFieldName(), fieldItem.getColumn().getColumnName()); - esFieldData.put(fieldItem.getFieldName(), value); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); @@ -408,7 +408,7 @@ public class ESSyncService { // ------关联表简单字段更新为null------ Map esFieldData = new LinkedHashMap<>(); for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) { - esFieldData.put(fieldItem.getFieldName(), null); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null); } joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); } else { @@ -509,7 +509,7 @@ public class ESSyncService { esTemplate.getESDataFromDmlData(mapping, data, esFieldData); esFieldData.remove(mapping.getPk()); for (String key : esFieldData.keySet()) { - esFieldData.put(key, null); + esFieldData.put(Util.cleanColumn(key), null); } } while (rs.next()) { @@ -618,7 +618,7 @@ public class ESSyncService { rs, fieldItem.getFieldName(), fieldItem.getColumn().getColumnName()); - esFieldData.put(fieldItem.getFieldName(), val); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val); break out; } } @@ -629,7 +629,7 @@ public class ESSyncService { rs, fieldItem.getFieldName(), fieldItem.getColumn().getColumnName()); - esFieldData.put(fieldItem.getFieldName(), val); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val); } } @@ -725,7 +725,7 @@ public class ESSyncService { rs, fieldItem.getFieldName(), fieldItem.getFieldName()); - esFieldData.put(fieldItem.getFieldName(), val); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val); break; } } @@ -734,7 +734,7 @@ public class ESSyncService { } else { Object val = esTemplate .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName()); - esFieldData.put(fieldItem.getFieldName(), val); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val); } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java index 45eeb57f..8a083479 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java @@ -112,7 +112,9 @@ public class ESTemplate { * @param esFieldData 数据Map */ public void update(ESMapping mapping, Object pkVal, Map esFieldData) { - append4Update(mapping, pkVal, esFieldData); + Map esFieldDataTmp = new LinkedHashMap<>(esFieldData.size()); + esFieldData.forEach((k, v) -> esFieldDataTmp.put(Util.cleanColumn(k), v)); + append4Update(mapping, pkVal, esFieldDataTmp); commitBulk(); } @@ -250,6 +252,8 @@ public class ESTemplate { public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, String columnName) throws SQLException { + fieldName = Util.cleanColumn(fieldName); + columnName = Util.cleanColumn(columnName); String esType = getEsType(mapping, fieldName); Object value = resultSet.getObject(columnName); @@ -281,7 +285,7 @@ public class ESTemplate { if (!fieldItem.getFieldName().equals(mapping.get_id()) && !mapping.getSkips().contains(fieldItem.getFieldName())) { - esFieldData.put(fieldItem.getFieldName(), value); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } @@ -319,7 +323,7 @@ public class ESTemplate { for (ColumnItem columnItem : fieldItem.getColumnItems()) { if (dmlOld.containsKey(columnItem.getColumnName()) && !mapping.getSkips().contains(fieldItem.getFieldName())) { - esFieldData.put(fieldItem.getFieldName(), + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); break; } @@ -372,7 +376,7 @@ public class ESTemplate { if (!fieldItem.getFieldName().equals(mapping.get_id()) && !mapping.getSkips().contains(fieldItem.getFieldName())) { - esFieldData.put(fieldItem.getFieldName(), value); + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } @@ -402,7 +406,7 @@ public class ESTemplate { } if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { - esFieldData.put(fieldItem.getFieldName(), + esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); } }