[Improve][Connector-V2] Add ElasticSearch type converter (#6546)

Author: Hisoka-X

Diff for: seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalog.java

@@ -32,7 +32,9 @@
 import org.apache.seatunnel.api.table.catalog.exception.DatabaseNotExistException;
 import org.apache.seatunnel.api.table.catalog.exception.TableAlreadyExistException;
 import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
+import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.ElasticsearchClusterInfo;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.source.IndexDocsCount;
 
@@ -146,10 +148,8 @@ public CatalogTable getTable(TablePath tablePath)
             throws CatalogException, TableNotExistException {
         // Get the index mapping?
         checkNotNull(tablePath, "tablePath cannot be null");
-        ElasticSearchDataTypeConvertor elasticSearchDataTypeConvertor =
-                new ElasticSearchDataTypeConvertor();
         TableSchema.Builder builder = TableSchema.builder();
-        Map<String, String> fieldTypeMapping =
+        Map<String, BasicTypeDefine<EsType>> fieldTypeMapping =
                 esRestClient.getFieldTypeMapping(tablePath.getTableName(), Collections.emptyList());
         buildColumnsWithErrorCheck(
                 tablePath,
@@ -159,8 +159,9 @@ public CatalogTable getTable(TablePath tablePath)
                     // todo: we need to add a new type TEXT or add length in STRING type
                     return PhysicalColumn.of(
                             nameAndType.getKey(),
-                            elasticSearchDataTypeConvertor.toSeaTunnelType(
-                                    nameAndType.getKey(), nameAndType.getValue()),
+                            ElasticSearchTypeConverter.INSTANCE
+                                    .convert(nameAndType.getValue())
+                                    .getDataType(),
                             (Long) null,
                             true,
                             null,

Diff for: seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchDataTypeConvertor.java

@@ -17,34 +17,24 @@
 
 package org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog;
 
+import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.DataTypeConvertor;
-import org.apache.seatunnel.api.table.type.BasicType;
-import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
-import org.apache.seatunnel.api.table.type.SqlType;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType;
 
 import com.google.auto.service.AutoService;
 
 import java.util.Map;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+/** @deprecated instead by {@link ElasticSearchTypeConverter} */
+@Deprecated
 @AutoService(DataTypeConvertor.class)
 public class ElasticSearchDataTypeConvertor implements DataTypeConvertor<String> {
 
-    public static final String STRING = "string";
-    public static final String KEYWORD = "keyword";
-    public static final String TEXT = "text";
-    public static final String BOOLEAN = "boolean";
-    public static final String BYTE = "byte";
-    public static final String SHORT = "short";
-    public static final String INTEGER = "integer";
-    public static final String LONG = "long";
-    public static final String FLOAT = "float";
-    public static final String HALF_FLOAT = "half_float";
-    public static final String DOUBLE = "double";
-    public static final String DATE = "date";
-
     @Override
     public SeaTunnelDataType<?> toSeaTunnelType(String field, String connectorDataType) {
         return toSeaTunnelType(field, connectorDataType, null);
@@ -54,34 +44,14 @@ public SeaTunnelDataType<?> toSeaTunnelType(String field, String connectorDataTy
     public SeaTunnelDataType<?> toSeaTunnelType(
             String field, String connectorDataType, Map<String, Object> dataTypeProperties) {
         checkNotNull(connectorDataType, "connectorDataType can not be null");
-        switch (connectorDataType) {
-            case STRING:
-                return BasicType.STRING_TYPE;
-            case KEYWORD:
-                return BasicType.STRING_TYPE;
-            case TEXT:
-                return BasicType.STRING_TYPE;
-            case BOOLEAN:
-                return BasicType.BOOLEAN_TYPE;
-            case BYTE:
-                return BasicType.BYTE_TYPE;
-            case SHORT:
-                return BasicType.SHORT_TYPE;
-            case INTEGER:
-                return BasicType.INT_TYPE;
-            case LONG:
-                return BasicType.LONG_TYPE;
-            case FLOAT:
-                return BasicType.FLOAT_TYPE;
-            case HALF_FLOAT:
-                return BasicType.FLOAT_TYPE;
-            case DOUBLE:
-                return BasicType.DOUBLE_TYPE;
-            case DATE:
-                return LocalTimeType.LOCAL_DATE_TIME_TYPE;
-            default:
-                return BasicType.STRING_TYPE;
-        }
+        BasicTypeDefine<EsType> typeDefine =
+                BasicTypeDefine.<EsType>builder()
+                        .name(field)
+                        .columnType(connectorDataType)
+                        .dataType(connectorDataType)
+                        .build();
+
+        return ElasticSearchTypeConverter.INSTANCE.convert(typeDefine).getDataType();
     }
 
     @Override
@@ -90,29 +60,14 @@ public String toConnectorType(
             SeaTunnelDataType<?> seaTunnelDataType,
             Map<String, Object> dataTypeProperties) {
         checkNotNull(seaTunnelDataType, "seaTunnelDataType can not be null");
-        SqlType sqlType = seaTunnelDataType.getSqlType();
-        switch (sqlType) {
-            case STRING:
-                return STRING;
-            case BOOLEAN:
-                return BOOLEAN;
-            case BYTES:
-                return BYTE;
-            case TINYINT:
-                return SHORT;
-            case INT:
-                return INTEGER;
-            case BIGINT:
-                return LONG;
-            case FLOAT:
-                return FLOAT;
-            case DOUBLE:
-                return DOUBLE;
-            case TIMESTAMP:
-                return DATE;
-            default:
-                return STRING;
-        }
+        Column column =
+                PhysicalColumn.builder()
+                        .name(field)
+                        .dataType(seaTunnelDataType)
+                        .nullable(true)
+                        .build();
+        BasicTypeDefine<EsType> typeDefine = ElasticSearchTypeConverter.INSTANCE.reconvert(column);
+        return typeDefine.getColumnType();
     }
 
     @Override