check es pk

dbsyncer-connector/src/main/java/org/dbsyncer/connector/Connector.java

@@ -3,7 +3,6 @@ package org.dbsyncer.connector;
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.connector.config.*;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -105,9 +104,7 @@ public interface Connector<M, C> {
      * @param commandConfig
      * @return
      */
-    default Map<String, String> getSourceCommand(CommandConfig commandConfig) {
-        return Collections.EMPTY_MAP;
-    }
+    Map<String, String> getSourceCommand(CommandConfig commandConfig);
 
     /**
      * 获取目标源同步参数
@@ -115,7 +112,5 @@ public interface Connector<M, C> {
      * @param commandConfig
      * @return
      */
-    default Map<String, String> getTargetCommand(CommandConfig commandConfig) {
-        return Collections.EMPTY_MAP;
-    }
+    Map<String, String> getTargetCommand(CommandConfig commandConfig);
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnector.java

@@ -39,10 +39,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
@@ -263,6 +260,15 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
         return command;
     }
 
+    @Override
+    public Map<String, String> getTargetCommand(CommandConfig commandConfig) {
+        Table table = commandConfig.getTable();
+        if(!CollectionUtils.isEmpty(table.getColumn())){
+            getPrimaryKeyField(table.getColumn());
+        }
+        return Collections.EMPTY_MAP;
+    }
+
     private void genSearchSourceBuilder(SearchSourceBuilder builder, Map<String, String> command) {
         // 查询字段
         String fieldNamesJson = command.get(ConnectorConstant.OPERTION_QUERY);