add es field type

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/ElasticSearchConfigChecker.java

@@ -23,6 +23,7 @@ public class ElasticSearchConfigChecker implements ConnectorConfigChecker<ESConf
         String type = params.get("type");
         String url = params.get("url");
         String schema = params.get("schema");
+        String primaryKey = params.get("primaryKey");
         Assert.hasText(username, "Username is empty.");
         Assert.hasText(password, "Password is empty.");
         Assert.hasText(index, "Index is empty.");
@@ -36,5 +37,6 @@ public class ElasticSearchConfigChecker implements ConnectorConfigChecker<ESConf
         connectorConfig.setType(type);
         connectorConfig.setUrl(url);
         connectorConfig.setSchema(schema);
+        connectorConfig.setPrimaryKey(primaryKey);
     }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/ESConfig.java

@@ -38,6 +38,11 @@ public class ESConfig extends ConnectorConfig {
      */
     private String type;
 
+    /**
+     * 主键
+     */
+    private String primaryKey;
+
     public String getUrl() {
         return url;
     }
@@ -85,4 +90,12 @@ public class ESConfig extends ConnectorConfig {
     public void setType(String type) {
         this.type = type;
     }
+
+    public String getPrimaryKey() {
+        return primaryKey;
+    }
+
+    public void setPrimaryKey(String primaryKey) {
+        this.primaryKey = primaryKey;
+    }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/ESFieldTypeEnum.java

@@ -0,0 +1,100 @@
+package org.dbsyncer.connector.enums;
+
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.ConnectorException;
+
+import java.sql.Types;
+
+/**
+ * ES字段类型
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2021/08/26 21:13
+ */
+public enum ESFieldTypeEnum {
+
+    // 字符类型
+    KEYWORD("keyword", Types.VARCHAR),
+    TEXT("text", Types.LONGVARCHAR),
+    /**
+     * ES 5.X之后不再支持string, 由text或keyword取代
+     */
+    @Deprecated
+    STRING("string", Types.VARCHAR),
+
+    // 数字类型
+    /**
+     * 源库中的类型为unsigned int，建议使用long
+     */
+    INTEGER("integer", Types.INTEGER),
+    /**
+     * bit只有一位，建议使用boolean
+     */
+    LONG("long", Types.BIGINT),
+    /**
+     * 源库中的类型为unsigned tinyint 或 unsigned smallint，建议使用integer
+     */
+    SHORT("short", Types.TINYINT),
+    BYTE("byte", Types.BIT),
+    /**
+     * 为保证精度，建议使用text
+     */
+    DOUBLE("double", Types.DOUBLE),
+    FLOAT("float", Types.FLOAT),
+    HALF_FLOAT("half_float", Types.FLOAT),
+    SCALED_FLOAT("scaled_float", Types.FLOAT),
+    BOOLEAN("boolean", Types.BOOLEAN),
+
+    // 日期类型
+    /**
+     * 默认格式："strict_date_optional_time||epoch_millis"
+     * "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"  “2021/01/01 12:10:30” or “2021-01-01”
+     */
+    DAte("date", Types.DATE),
+
+    // 范围类型
+    INTEGER_RANGE("integer_range", Types.INTEGER),
+    FLOAT_RANGE("float_range", Types.FLOAT),
+    LONG_RANGE("long_range", Types.BIGINT),
+    DOUBLE_RANGE("double_range", Types.DOUBLE),
+    DATE_RANGE("date_range", Types.DATE),
+
+    // 其他类型
+    /**
+     * 弥补object类型不足，格式出现list放object会变为array："test": [{"a":"b},{}]
+     */
+    NESTED("nested", Types.VARCHAR),
+    OBJECT("object", Types.VARCHAR),
+    IP("ip", Types.VARCHAR),
+    TOKEN_COUNT("token_count", Types.BIGINT),
+    GEO_POINT("geo_point", Types.VARCHAR),
+    GEO_SHAPE("geo_shape", Types.VARCHAR),
+    BINARY("binary", Types.BINARY);
+
+    private String code;
+    private int type;
+
+    ESFieldTypeEnum(String code, int type) {
+        this.code = code;
+        this.type = type;
+    }
+
+    public static int getType(String code) throws ConnectorException {
+        for (ESFieldTypeEnum e : ESFieldTypeEnum.values()) {
+            if (StringUtil.equals(e.getCode(), code)) {
+                return e.getType();
+            }
+        }
+        return Types.VARCHAR;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public int getType() {
+        return type;
+    }
+
+}

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

@@ -2,11 +2,13 @@ package org.dbsyncer.connector.es;
 
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.AbstractConnector;
 import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.enums.ESFieldTypeEnum;
 import org.dbsyncer.connector.util.ESUtil;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
@@ -29,6 +31,7 @@ import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -91,10 +94,9 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
             if (CollectionUtils.isEmpty(properties)) {
                 throw new ConnectorException("查询字段不能为空.");
             }
-            properties.keySet().forEach(k -> {
-                // TODO 获取实现类型
-                //String columnType = (String) v.get("type");
-                fields.add(new Field(k, k, 12, false));
+            properties.forEach((k, v) -> {
+                String columnType = (String) v.get("type");
+                fields.add(new Field(k, columnType, ESFieldTypeEnum.getType(columnType), StringUtil.equals(config.getPrimaryKey(), k)));
             });
         } catch (IOException e) {
             logger.error(e.getMessage());
@@ -105,12 +107,12 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
 
     @Override
     public Map<String, String> getSourceCommand(CommandConfig commandConfig) {
-        return null;
+        return Collections.EMPTY_MAP;
     }
 
     @Override
     public Map<String, String> getTargetCommand(CommandConfig commandConfig) {
-        return null;
+        return Collections.EMPTY_MAP;
     }
 
     @Override

dbsyncer-web/src/main/resources/public/connector/addDqlMysql.html

@@ -48,7 +48,7 @@
 <script type="text/javascript">
 $(function () {
     // 初始化select2插件
-    var $connectorTypeSelect = $(".select-control").select2({
+    $(".select-control").select2({
         width: "100%",
         theme: "classic"
     });

dbsyncer-web/src/main/resources/public/connector/addElasticSearch.html

@@ -36,12 +36,19 @@
             </select>
         </div>
     </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">主键 </label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" placeholder="id" th:value="${connector?.config?.primaryKey}?:'id'"/>
+        </div>
+        <div class="col-sm-6"></div>
+    </div>
 </div>
 
 <script type="text/javascript">
 $(function () {
     // 初始化select2插件
-    var $connectorTypeSelect = $(".select-control").select2({
+    $(".select-control").select2({
         width: "100%",
         theme: "classic"
     });

dbsyncer-web/src/main/resources/public/connector/addMysql.html

@@ -33,7 +33,7 @@
 <script type="text/javascript">
 $(function () {
     // 初始化select2插件
-    var $connectorTypeSelect = $(".select-control").select2({
+    $(".select-control").select2({
         width: "100%",
         theme: "classic"
     });