适配多索引配置

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/tablegroup/TableGroupChecker.java

@@ -159,7 +159,7 @@ public class TableGroupChecker extends AbstractChecker {
                 }
             });
         }
-        return new Table(tableName, metaInfo.getTableType(), metaInfo.getColumn(), metaInfo.getSql());
+        return new Table(tableName, metaInfo.getTableType(), metaInfo.getColumn(), metaInfo.getSql(), metaInfo.getIndexType());
     }
 
     private void checkRepeatedTable(String mappingId, String sourceTable, String targetTable) {

dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/ElasticsearchConnector.java

@@ -67,6 +67,7 @@ import java.sql.Types;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -188,8 +189,19 @@ public final class ElasticsearchConnector extends AbstractConnector implements C
             // 6.x 版本
             if (Version.V_7_0_0.after(connectorInstance.getVersion())) {
                 Map<String, Object> sourceMap = XContentHelper.convertToMap(mappingMetaData.source().compressedReference(), true, null).v2();
-                parseProperties(fields, (Map) sourceMap.get(INDEX_TYPE));
-                return new MetaInfo().setColumn(fields);
+                if (CollectionUtils.isEmpty(sourceMap)) {
+                    throw new ElasticsearchException("未获取到索引配置");
+                }
+                Iterator<String> iterator = sourceMap.keySet().iterator();
+                String indexType = null;
+                if (iterator.hasNext()) {
+                    indexType = iterator.next();
+                    parseProperties(fields, (Map) sourceMap.get(indexType));
+                }
+                if (StringUtil.isBlank(indexType)) {
+                    throw new ElasticsearchException("索引type为空");
+                }
+                return new MetaInfo().setColumn(fields).setIndexType(indexType);
             }
 
             // 7.x 版本以上
@@ -265,13 +277,11 @@ public final class ElasticsearchConnector extends AbstractConnector implements C
         final Result result = new Result();
         final List<Field> pkFields = PrimaryKeyUtil.findConfigPrimaryKeyFields(config);
         try {
-            BulkRequest request = new BulkRequest();
-            // 默认取第一个主键
+            final BulkRequest request = new BulkRequest();
             final String pk = pkFields.get(0).getName();
-            String indexName = config.getCommand().get(TARGET_INDEX_NAME);
-            // 6.x 版本
-            final String type = Version.V_7_0_0.after(connectorInstance.getVersion()) ? INDEX_TYPE : null;
-            data.forEach(row -> addRequest(request, indexName, type, config.getEvent(), String.valueOf(row.get(pk)), row));
+            final String indexName = config.getCommand().get(TARGET_INDEX_NAME);
+            final String indexType = config.getCommand().get(INDEX_TYPE);
+            data.forEach(row -> addRequest(request, indexName, indexType, config.getEvent(), String.valueOf(row.get(pk)), row));
 
             BulkResponse response = connectorInstance.getConnection().bulkWithVersion(request, RequestOptions.DEFAULT);
             RestStatus restStatus = response.status();
@@ -314,6 +324,7 @@ public final class ElasticsearchConnector extends AbstractConnector implements C
         PrimaryKeyUtil.findTablePrimaryKeys(table);
         Map<String, String> command = new HashMap<>();
         command.put(TARGET_INDEX_NAME, table.getName());
+        command.put(INDEX_TYPE, table.getIndexType());
         return command;
     }
 

dbsyncer-parser/src/main/java/org/dbsyncer/parser/impl/ParserComponentImpl.java

@@ -95,8 +95,8 @@ public class ParserComponentImpl implements ParserComponent {
         ConnectorConfig tConnConfig = getConnectorConfig(mapping.getTargetConnectorId());
         Table sourceTable = tableGroup.getSourceTable();
         Table targetTable = tableGroup.getTargetTable();
-        Table sTable = new Table(sourceTable.getName(), sourceTable.getType(), new ArrayList<>(), sourceTable.getSql());
-        Table tTable = new Table(targetTable.getName(), targetTable.getType(), new ArrayList<>(), sourceTable.getSql());
+        Table sTable = sourceTable.clone().setColumn(new ArrayList<>());
+        Table tTable = targetTable.clone().setColumn(new ArrayList<>());
         List<FieldMapping> fieldMapping = tableGroup.getFieldMapping();
         if (!CollectionUtils.isEmpty(fieldMapping)) {
             fieldMapping.forEach(m -> {
@@ -217,7 +217,7 @@ public class ParserComponentImpl implements ParserComponent {
         int total = dataList.size();
         // 单次任务
         if (total <= batchSize) {
-            return connectorFactory.writer(batchWriter.getConnectorInstance(), new WriterBatchConfig(tableName, event, command, fields, dataList,batchWriter.isEnableOnlyUpdate()));
+            return connectorFactory.writer(batchWriter.getConnectorInstance(), new WriterBatchConfig(tableName, event, command, fields, dataList, batchWriter.isEnableOnlyUpdate()));
         }
 
         // 批量任务, 拆分

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/connector/database/AbstractDQLConnector.java

@@ -35,7 +35,7 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
         List<Table> tables = new ArrayList<>();
         if (!CollectionUtils.isEmpty(sqlTables)) {
             sqlTables.forEach(s ->
-                tables.add(new Table(s.getSqlName(), TableTypeEnum.TABLE.getCode(), Collections.EMPTY_LIST, s.getSql()))
+                tables.add(new Table(s.getSqlName(), TableTypeEnum.TABLE.getCode(), Collections.EMPTY_LIST, s.getSql(), null))
             );
         }
         return tables;

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/model/MetaInfo.java

@@ -28,6 +28,11 @@ public class MetaInfo {
      */
     private String sql;
 
+    /**
+     * 索引类型（ES）
+     */
+    private String indexType;
+
     public String getTableType() {
         return tableType;
     }
@@ -55,6 +60,15 @@ public class MetaInfo {
         return this;
     }
 
+    public String getIndexType() {
+        return indexType;
+    }
+
+    public MetaInfo setIndexType(String indexType) {
+        this.indexType = indexType;
+        return this;
+    }
+
     @Override
     public String toString() {
         return new StringBuilder().append("MetaInfo{").append("tableType=").append(tableType).append(", ").append("column=").append(column).append('}').toString();

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/model/Table.java

@@ -36,6 +36,11 @@ public class Table {
     // 总数
     private long count;
 
+    /**
+     * 索引类型（ES）
+     */
+    private String indexType;
+
     public Table() {
     }
 
@@ -44,14 +49,15 @@ public class Table {
     }
 
     public Table(String name, String type) {
-        this(name, type, null, null);
+        this(name, type, null, null, null);
     }
 
-    public Table(String name, String type, List<Field> column, String sql) {
+    public Table(String name, String type, List<Field> column, String sql, String indexType) {
         this.name = name;
         this.type = type;
         this.column = column;
         this.sql = sql;
+        this.indexType = indexType;
     }
 
     public String getName() {
@@ -74,8 +80,9 @@ public class Table {
         return column;
     }
 
-    public void setColumn(List<Field> column) {
+    public Table setColumn(List<Field> column) {
         this.column = column;
+        return this;
     }
 
     public String getSql() {
@@ -93,4 +100,17 @@ public class Table {
     public void setCount(long count) {
         this.count = count;
     }
+
+    public String getIndexType() {
+        return indexType;
+    }
+
+    public void setIndexType(String indexType) {
+        this.indexType = indexType;
+    }
+
+    @Override
+    public Table clone() {
+        return new Table(name, type, column, sql, indexType);
+    }
 }