chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

增加对sql字段 `` 的支持

Browse Source 
fix #1559
This commit is contained in:
mcy
2019-03-19 19:13:15 +08:00
parent 51c469aa6f
commit a2360efff9
3 changed files with 30 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
View File

@@ -202,20 +202,23 @@ public class ESEtlService {
long batchBegin = System.currentTimeMillis(); long batchBegin = System.currentTimeMillis();
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
Map<String, Object> idEsFieldData = new HashMap<>();
for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) { for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) {
// 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) {
continue;
}
String fieldName = fieldItem.getFieldName(); String fieldName = fieldItem.getFieldName();
if (mapping.getSkips().contains(fieldName)) { if (mapping.getSkips().contains(fieldName)) {
continue; 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()) { if (!mapping.getRelations().isEmpty()) {
@@ -241,13 +244,13 @@ public class ESEtlService {
} }
} }
esFieldData.put(relationField, relations); esFieldData.put(Util.cleanColumn(relationField), relations);
}); });
} }
Object idVal = null; Object idVal = null;
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
idVal = esFieldData.get(mapping.get_id()); idVal = idEsFieldData.get(mapping.get_id());
} }
if (idVal != null) { if (idVal != null) {
@@ -271,7 +274,7 @@ public class ESEtlService {
bulkRequestBuilder.add(indexRequestBuilder); bulkRequestBuilder.add(indexRequestBuilder);
} }
} else { } else {
idVal = esFieldData.get(mapping.getPk()); idVal = idEsFieldData.get(mapping.getPk());
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) .setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))

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

@@ -167,7 +167,7 @@ public class ESSyncService {
data, data,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
@@ -296,7 +296,7 @@ public class ESSyncService {
data, data,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
} }
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
@@ -408,7 +408,7 @@ public class ESSyncService {
// ------关联表简单字段更新为null------ // ------关联表简单字段更新为null------
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) { for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
esFieldData.put(fieldItem.getFieldName(), null); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null);
} }
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
} else { } else {
@@ -509,7 +509,7 @@ public class ESSyncService {
esTemplate.getESDataFromDmlData(mapping, data, esFieldData); esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
esFieldData.remove(mapping.getPk()); esFieldData.remove(mapping.getPk());
for (String key : esFieldData.keySet()) { for (String key : esFieldData.keySet()) {
esFieldData.put(key, null); esFieldData.put(Util.cleanColumn(key), null);
} }
} }
while (rs.next()) { while (rs.next()) {
@@ -618,7 +618,7 @@ public class ESSyncService {
rs, rs,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
break out; break out;
} }
} }
@@ -629,7 +629,7 @@ public class ESSyncService {
rs, rs,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
} }
} }
@@ -725,7 +725,7 @@ public class ESSyncService {
rs, rs,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getFieldName()); fieldItem.getFieldName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
break; break;
} }
} }
@@ -734,7 +734,7 @@ public class ESSyncService {
} else { } else {
Object val = esTemplate Object val = esTemplate
.getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName()); .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
} }
} }

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

@@ -112,7 +112,9 @@ public class ESTemplate {
* @param esFieldData 数据Map * @param esFieldData 数据Map
*/ */
public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
append4Update(mapping, pkVal, esFieldData); Map<String, Object> esFieldDataTmp = new LinkedHashMap<>(esFieldData.size());
esFieldData.forEach((k, v) -> esFieldDataTmp.put(Util.cleanColumn(k), v));
append4Update(mapping, pkVal, esFieldDataTmp);
commitBulk(); commitBulk();
} }
@@ -250,6 +252,8 @@ public class ESTemplate {
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
String columnName) throws SQLException { String columnName) throws SQLException {
fieldName = Util.cleanColumn(fieldName);
columnName = Util.cleanColumn(columnName);
String esType = getEsType(mapping, fieldName); String esType = getEsType(mapping, fieldName);
Object value = resultSet.getObject(columnName); Object value = resultSet.getObject(columnName);
@@ -281,7 +285,7 @@ public class ESTemplate {
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !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()) { for (ColumnItem columnItem : fieldItem.getColumnItems()) {
if (dmlOld.containsKey(columnItem.getColumnName()) if (dmlOld.containsKey(columnItem.getColumnName())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
break; break;
} }
@@ -372,7 +376,7 @@ public class ESTemplate {
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !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())) { 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)); getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
} }
} }