diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java index f0505aaa..f092bf78 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java @@ -23,7 +23,6 @@ public class SchemaItem { this.isAllFieldsSimple(); aliasTableItems.values().forEach(tableItem -> { tableItem.getRelationTableFields(); - // tableItem.getRelationKeyFieldItems(); tableItem.getRelationSelectFieldItems(); }); } @@ -76,19 +75,19 @@ public class SchemaItem { getSelectFields() .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> { TableItem tableItem = getAliasTableItems().get(columnItem.getOwner()); - if (!tableItem.isSubQuery()) { - List fieldItems = columnFields.computeIfAbsent( - columnItem.getOwner() + "." + columnItem.getColumnName(), - k -> new ArrayList<>()); - fieldItems.add(fieldItem); - } else { - tableItem.getSubQueryFields().forEach(subQueryField -> { - List fieldItems = columnFields.computeIfAbsent( - columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(), - k -> new ArrayList<>()); - fieldItems.add(fieldItem); - }); - } + // if (!tableItem.isSubQuery()) { + List fieldItems = columnFields.computeIfAbsent( + columnItem.getOwner() + "." + columnItem.getColumnName(), + k -> new ArrayList<>()); + fieldItems.add(fieldItem); + // } else { + // tableItem.getSubQueryFields().forEach(subQueryField -> { + // List fieldItems = columnFields.computeIfAbsent( + // columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(), + // k -> new ArrayList<>()); + // fieldItems.add(fieldItem); + // }); + // } })); } } @@ -146,8 +145,6 @@ public class SchemaItem { private boolean subQuery; private volatile Map> relationTableFields; // 当前表关联条件字段对应主表查询字段 - // private volatile List relationKeyFieldItems; // - // 关联条件字段在select中的对应字段 private volatile List relationSelectFieldItems; // 子表所在主表的查询字段 public TableItem(SchemaItem schemaItem){ @@ -237,10 +234,8 @@ public class SchemaItem { FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); FieldItem currentTableRelField = null; if (getAlias().equals(leftFieldItem.getOwner())) { - // relationTableFields.add(leftFieldItem); currentTableRelField = leftFieldItem; } else if (getAlias().equals(rightFieldItem.getOwner())) { - // relationTableFields.add(rightFieldItem); currentTableRelField = rightFieldItem; } @@ -268,37 +263,6 @@ public class SchemaItem { return relationTableFields; } - // public List getRelationKeyFieldItems() { - // if (relationKeyFieldItems == null) { - // synchronized (SchemaItem.class) { - // if (relationKeyFieldItems == null) { - // relationKeyFieldItems = new ArrayList<>(); - // getRelationFields().forEach(relationFieldsPair -> { - // FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); - // List selectFieldItem = getSchemaItem().getColumnFields() - // .get(leftFieldItem.getOwner() + "." + - // leftFieldItem.getColumn().getColumnName()); - // if (selectFieldItem != null && !selectFieldItem.isEmpty()) { - // relationKeyFieldItems.addAll(selectFieldItem); - // } else { - // FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); - // selectFieldItem = getSchemaItem().getColumnFields() - // .get(rightFieldItem.getOwner() + "." + - // rightFieldItem.getColumn().getColumnName()); - // if (selectFieldItem != null && !selectFieldItem.isEmpty()) { - // relationKeyFieldItems.addAll(selectFieldItem); - // } else { - // throw new UnsupportedOperationException( - // "Relation condition column must in select columns."); - // } - // } - // }); - // } - // } - // } - // return relationKeyFieldItems; - // } - public List getRelationSelectFieldItems() { if (relationSelectFieldItems == null) { synchronized (SchemaItem.class) { 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 4e86383e..9cdff29a 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 @@ -133,12 +133,20 @@ public class ESSyncService { continue; // 单表插入完成 } - // ------是否主表 查询sql来插入------ + // ------是主表 查询sql来插入------ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { 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( + "Single table insert ot 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()) { @@ -192,23 +200,30 @@ public class ESSyncService { if (!tableItem.isSubQuery()) { Map esFieldData = new LinkedHashMap<>(); for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) { - Object value = esTemplate - .getValFromData(mapping, data, fieldItem.getColumn().getColumnName()); + Object value = esTemplate.getValFromData(mapping, + data, + fieldItem.getFieldName(), + fieldItem.getColumn().getColumnName()); esFieldData.put(fieldItem.getFieldName(), value); } BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); for (Map.Entry> entry : tableItem.getRelationTableFields() .entrySet()) { - Object value = esTemplate - .getValFromData(mapping, data, entry.getKey().getColumn().getColumnName()); for (FieldItem fieldItem : entry.getValue()) { - String fieldName = fieldItem.getFieldName(); - // 判断是否是主键 - if (fieldName.equals(mapping.get_id())) { - fieldName = "_id"; + 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"; + } + queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)); } - queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)); } } @@ -228,7 +243,95 @@ public class ESSyncService { mapping.get_index()); } } else { - // TODO + StringBuilder sql = new StringBuilder( + "SELECT * FROM (" + tableItem.getSubQuerySql() + ") " + tableItem.getAlias() + " WHERE "); + + for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) { + String columnName = fkFieldItem.getColumn().getColumnName(); + Object value = esTemplate + .getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName); + if (value instanceof String) { + sql.append(tableItem.getAlias()) + .append(".") + .append(columnName) + .append("='") + .append(value) + .append("' ") + .append(" AND "); + } else { + sql.append(tableItem.getAlias()) + .append(".") + .append(columnName) + .append("=") + .append(value) + .append(" ") + .append(" AND "); + } + } + int len = sql.length(); + sql.delete(len - 5, len); + 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", " ")); + } + ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { + try { + while (rs.next()) { + Map esFieldData = new LinkedHashMap<>(); + for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) { + Object val = esTemplate.getValFromRS(mapping, + rs, + fieldItem.getFieldName(), + fieldItem.getColumn().getColumnName()); + esFieldData.put(fieldItem.getFieldName(), val); + } + + BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); + for (Map.Entry> 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"; + } + queryBuilder.must(QueryBuilders.termsQuery(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()); + } + boolean result = esTemplate.updateByQuery(mapping, queryBuilder, esFieldData); + if (!result) { + logger.error( + "Join table update es index by foreign key error, destination:{}, table: {}, index: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index()); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } + return 0; + }); } } else { // TODO 查询总sql diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java index f09641d7..88f7b496 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java @@ -2,7 +2,6 @@ package com.alibaba.otter.canal.client.adapter.es.support; import java.io.IOException; import java.io.InputStream; -import java.io.UnsupportedEncodingException; import java.nio.charset.StandardCharsets; import java.sql.*; import java.util.*; @@ -20,6 +19,7 @@ import com.alibaba.fastjson.JSON; 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; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem; public class ESSyncUtil { @@ -249,7 +249,7 @@ public class ESSyncUtil { /** * 执行查询sql */ - public static Object sqlRS(DataSource ds, String sql, Function fun) { + public static Object sqlRS(DataSource ds, String sql, Function fun) { Connection conn = null; Statement smt = null; ResultSet rs = null; 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 fec06dd1..7df0c77e 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 @@ -274,44 +274,14 @@ public class ESTemplate { return true; } - // public Object getRSDataValue(ESMapping mapping, ResultSet rs, String - // fieldName) throws SQLException { - // String esType = getEsType(mapping, fieldName); - // Object value = rs.getObject(fieldName); - // if (value instanceof Boolean) { - // // 判断es类型 - // if (!"boolean".equals(esType)) { - // value = rs.getByte(fieldName); - // } - // } - // return value; - // } - // - // public Object getDataValue(ESMapping mapping, Map data, - // String fieldName) { - // String esType = getEsType(mapping, fieldName); - // Object value = data.get(fieldName); - // if (value instanceof Byte) { - // if ("boolean".equals(esType)) { - // value = ((Byte) value).intValue() != 0; - // } - // } - // return value; - // } - - // public Object convertType(ESMapping mapping, String fieldName, Object val) { - // // 从mapping中获取类型来转换 - // String esType = getEsType(mapping, fieldName); - // return ESSyncUtil.typeConvert(val, esType); - // } - - public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName) throws SQLException { + public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, + String columnName) throws SQLException { String esType = getEsType(mapping, fieldName); - Object value = resultSet.getObject(fieldName); + Object value = resultSet.getObject(columnName); if (value instanceof Boolean) { if (!"boolean".equals(esType)) { - value = resultSet.getByte(fieldName); + value = resultSet.getByte(columnName); } } return ESSyncUtil.typeConvert(value, esType); @@ -323,48 +293,23 @@ public class ESTemplate { String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); Object resultIdVal = null; for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { - String fieldName = Util.cleanColumn(fieldItem.getColumnItems().iterator().next().getColumnName()); - Object value = getValFromRS(mapping, resultSet, fieldName); + Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); if (fieldItem.getFieldName().equals(idFieldName)) { resultIdVal = value; } - if (!fieldItem.getFieldName().equals(mapping.get_id()) && !mapping.getSkips().contains(fieldName)) { - esFieldData.put(fieldName, value); + if (!fieldItem.getFieldName().equals(mapping.get_id()) + && !mapping.getSkips().contains(fieldItem.getFieldName())) { + esFieldData.put(fieldItem.getFieldName(), value); } } return resultIdVal; } - // public Object getIdValFromDmlData(ESMapping mapping, Map - // dmlData) { - // SchemaItem schemaItem = mapping.getSchemaItem(); - // String idFieldName = mapping.get_id() == null ? mapping.getPk() : - // mapping.get_id(); - // - // for (SchemaItem.FieldItem fieldItem : schemaItem.getSelectFields().values()) - // { - // if (fieldItem.getFieldName().equals(idFieldName)) { - // String fieldName = - // Util.cleanColumn(fieldItem.getColumnItems().iterator().next().getColumnName()); - // String esType = getEsType(mapping, fieldName); - // Object value = dmlData.get(fieldName); - // if (value instanceof Byte) { - // if ("boolean".equals(esType)) { - // value = ((Byte) value).intValue() != 0; - // } - // } - // - // return ESSyncUtil.typeConvert(value, esType); - // } - // } - // return null; - // } - - public Object getValFromData(ESMapping mapping, Map dmlData, String fieldName) { + public Object getValFromData(ESMapping mapping, Map dmlData, String fieldName, String columnName) { String esType = getEsType(mapping, fieldName); - Object value = dmlData.get(fieldName); + Object value = dmlData.get(columnName); if (value instanceof Byte) { if ("boolean".equals(esType)) { value = ((Byte) value).intValue() != 0; @@ -387,15 +332,16 @@ public class ESTemplate { String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); Object resultIdVal = null; for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { - String fieldName = Util.cleanColumn(fieldItem.getColumnItems().iterator().next().getColumnName()); - Object value = getValFromData(mapping, dmlData, fieldName); + String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); + Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); if (fieldItem.getFieldName().equals(idFieldName)) { resultIdVal = value; } - if (!fieldItem.getFieldName().equals(mapping.get_id()) && !mapping.getSkips().contains(fieldName)) { - esFieldData.put(fieldName, value); + if (!fieldItem.getFieldName().equals(mapping.get_id()) + && !mapping.getSkips().contains(fieldItem.getFieldName())) { + esFieldData.put(fieldItem.getFieldName(), value); } } return resultIdVal; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java new file mode 100644 index 00000000..e5b31cbd --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java @@ -0,0 +1,27 @@ +package com.alibaba.otter.canal.client.adapter.es.test.sync; + +import com.alibaba.otter.canal.client.adapter.es.ESAdapter; +import com.alibaba.otter.canal.client.adapter.es.test.TestConstant; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; + +import java.util.HashMap; +import java.util.Map; + +public class Common { + + public static ESAdapter init() { + DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); + + OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); + outerAdapterConfig.setName("es"); + outerAdapterConfig.setHosts(TestConstant.esHosts); + Map properties = new HashMap<>(); + properties.put("cluster.name", TestConstant.clusterNmae); + outerAdapterConfig.setProperties(properties); + + ESAdapter esAdapter = new ESAdapter(); + esAdapter.init(outerAdapterConfig); + return esAdapter; + } +} diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java new file mode 100644 index 00000000..dde3351d --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java @@ -0,0 +1,57 @@ +package com.alibaba.otter.canal.client.adapter.es.test.sync; + +import java.util.*; + +import org.elasticsearch.action.get.GetResponse; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.es.ESAdapter; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.Dml; + +public class RoleSyncJoinOneTest { + + private ESAdapter esAdapter; + + @Before + public void init() { + AdapterConfigs.put("es", "mytest_user_join_one.yml"); + esAdapter = Common.init(); + } + + /** + * 非子查询从表插入 (确保主表记录必须有数据) + */ + @Test + public void insertTest01() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("INSERT"); + dml.setDatabase("mytest"); + dml.setTable("role"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("role_name", "admin"); + + dml.setData(dataList); + + esAdapter.getEsSyncService().sync(dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("admin", response.getSource().get("_role_name")); + } + + @After + public void after() { + esAdapter.destroy(); + DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close); + } +} diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinSubTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinSubTest.java new file mode 100644 index 00000000..31387898 --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinSubTest.java @@ -0,0 +1,57 @@ +package com.alibaba.otter.canal.client.adapter.es.test.sync; + +import java.util.*; + +import org.elasticsearch.action.get.GetResponse; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.es.ESAdapter; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.Dml; + +public class RoleSyncJoinSubTest { + + private ESAdapter esAdapter; + + @Before + public void init() { + AdapterConfigs.put("es", "mytest_user_join_sub.yml"); + esAdapter = Common.init(); + } + + /** + * 非子查询从表插入 (确保主表记录必须有数据) + */ + @Test + public void insertTest01() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("INSERT"); + dml.setDatabase("mytest"); + dml.setTable("role"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("role_name", "admin2"); + + dml.setData(dataList); + + esAdapter.getEsSyncService().sync(dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("admin2", response.getSource().get("_role_name")); + } + + @After + public void after() { + esAdapter.destroy(); + DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close); + } +} diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java new file mode 100644 index 00000000..f2f9f4e3 --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java @@ -0,0 +1,59 @@ +package com.alibaba.otter.canal.client.adapter.es.test.sync; + +import java.util.*; + +import org.elasticsearch.action.get.GetResponse; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.es.ESAdapter; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.Dml; + +public class UserSyncJoinOneTest { + + private ESAdapter esAdapter; + + @Before + public void init() { + AdapterConfigs.put("es", "mytest_user_join_one.yml"); + esAdapter = Common.init(); + } + + /** + * 主表带函数插入, 数据库里内容必须和单测一致 + */ + @Test + public void insertTest01() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("INSERT"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("name", "Eric"); + data.put("role_id", 1L); + data.put("c_time", new Date()); + + dml.setData(dataList); + + esAdapter.getEsSyncService().sync(dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("Eric_", response.getSource().get("_name")); + } + + @After + public void after() { + esAdapter.destroy(); + DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close); + } +} diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java new file mode 100644 index 00000000..16a25bea --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java @@ -0,0 +1,75 @@ +package com.alibaba.otter.canal.client.adapter.es.test.sync; + +import java.util.*; + +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.es.ESAdapter; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.Dml; + +public class UserSyncSingleTest { + + private ESAdapter esAdapter; + + @Before + public void init() { + AdapterConfigs.put("es", "mytest_user_single.yml"); + esAdapter = Common.init(); + } + + /** + * 单表插入 + */ + @Test + public void insertTest01() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("INSERT"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("name", "Eric"); + data.put("role_id", 1L); + data.put("c_time", new Date()); + + dml.setData(dataList); + + esAdapter.getEsSyncService().sync(dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("Eric", response.getSource().get("_name")); + } + + @Test + public void ttt() { + SearchResponse response = esAdapter.getTransportClient() + .prepareSearch("mytest_user") + .setTypes("_doc") + .setQuery(QueryBuilders.termQuery("_id", 1L)) + .setSize(100) + .get(); + for (SearchHit hit : response.getHits()) { + System.out.println(hit.getSourceAsMap().get("name")); + } + } + + @After + public void after() { + esAdapter.destroy(); + DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close); + } +} diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java deleted file mode 100644 index c69fa995..00000000 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java +++ /dev/null @@ -1,133 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.es.test.sync; - -import java.util.*; - -import org.elasticsearch.action.get.GetResponse; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.search.SearchHit; -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; - -import com.alibaba.druid.pool.DruidDataSource; -import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.es.test.TestConstant; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; -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.OuterAdapterConfig; - -public class UserSyncTest { - - private ESAdapter esAdapter; - - @Before - public void init() { - AdapterConfigs.put("es", "mytest_user.yml"); - DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); - - OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); - outerAdapterConfig.setName("es"); - outerAdapterConfig.setHosts(TestConstant.esHosts); - Map properties = new HashMap<>(); - properties.put("cluster.name", TestConstant.clusterNmae); - outerAdapterConfig.setProperties(properties); - - esAdapter = new ESAdapter(); - esAdapter.init(outerAdapterConfig); - - } - - /** - * 单表插入 - */ - @Test - public void insertTest01() { - Dml dml = new Dml(); - dml.setDestination("example"); - dml.setTs(new Date().getTime()); - dml.setType("INSERT"); - dml.setDatabase("mytest"); - dml.setTable("user"); - List> dataList = new ArrayList<>(); - Map data = new LinkedHashMap<>(); - dataList.add(data); - data.put("id", 1L); - data.put("name", "Eric"); - data.put("role_id", 1L); - data.put("c_time", new Date()); - - dml.setData(dataList); - - esAdapter.getEsSyncService().sync(dml); - - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); - Assert.assertEquals("Eric", response.getSource().get("name")); - } - - /** - * 主表带函数插入, 数据库里内容必须和单测一致 - */ - @Test - public void insertTest02() { - Dml dml = new Dml(); - dml.setDestination("example"); - dml.setTs(new Date().getTime()); - dml.setType("INSERT"); - dml.setDatabase("mytest"); - dml.setTable("user"); - List> dataList = new ArrayList<>(); - Map data = new LinkedHashMap<>(); - dataList.add(data); - data.put("id", 1L); - data.put("name", "Eric"); - data.put("role_id", 1L); - data.put("c_time", new Date()); - - dml.setData(dataList); - - esAdapter.getEsSyncService().sync(dml); - - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); - Assert.assertEquals("Eric_", response.getSource().get("name")); - } - - @Test - public void insertTest03() { - Dml dml = new Dml(); - dml.setDestination("example"); - dml.setTs(new Date().getTime()); - dml.setType("INSERT"); - dml.setDatabase("mytest"); - dml.setTable("role"); - List> dataList = new ArrayList<>(); - Map data = new LinkedHashMap<>(); - dataList.add(data); - data.put("id", 1L); - data.put("role_name", "admin"); - - dml.setData(dataList); - - esAdapter.getEsSyncService().sync(dml); - } - - @Test - public void ttt(){ - SearchResponse response = esAdapter.getTransportClient().prepareSearch("mytest_user") - .setTypes("_doc") - .setQuery(QueryBuilders.termQuery("_id", 1L)) - .setSize(100) - .get(); - for (SearchHit hit : response.getHits()) { - System.out.println(hit.getSourceAsMap().get("name")); - } - } - - @After - public void after() { - esAdapter.destroy(); - DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close); - } -} diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml new file mode 100644 index 00000000..8e2f4efa --- /dev/null +++ b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml @@ -0,0 +1,10 @@ +dataSourceKey: defaultDS +destination: example +esMapping: + _index: mytest_user + _type: _doc + _id: _id + sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id, + b.role_name as _role_name, a.c_time as _c_time from user a + left join role b on b.id=a.role_id" + commitBatch: 3000 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml new file mode 100644 index 00000000..bde1139f --- /dev/null +++ b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml @@ -0,0 +1,10 @@ +dataSourceKey: defaultDS +destination: example +esMapping: + _index: mytest_user + _type: _doc + _id: _id + sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id, + b.role_name as _role_name, a.c_time as _c_time from user a + left join (select id, role_name from role ) b on b.id=a.role_id" + commitBatch: 3000 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml new file mode 100644 index 00000000..98fad232 --- /dev/null +++ b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml @@ -0,0 +1,8 @@ +dataSourceKey: defaultDS +destination: example +esMapping: + _index: mytest_user + _type: _doc + _id: _id + sql: "select a.id as _id, a.names as _name, a.role_id as _role_id, a.c_time as _c_time from user a" + commitBatch: 3000 \ No newline at end of file