[Feature][Doris] Add Doris type converter (#6354)

---------

Co-authored-by: Jia Fan <fanjiaeminem@qq.com>

Author: EricJoy2048

docs/en/connector-v2/sink/Doris.md

@@ -116,6 +116,7 @@ You can use the following placeholders
   description of Doris
 - rowtype_primary_key: Used to get the primary key in the upstream schema (maybe a list)
 - rowtype_unique_key: Used to get the unique key in the upstream schema (maybe a list)
+- rowtype_duplicate_key: Used to get the duplicate key in the upstream schema (only for doris source, maybe a list)
 
 ## Data Type Mapping
 

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SaveModePlaceHolder.java

@@ -23,6 +23,7 @@
 public enum SaveModePlaceHolder {
     ROWTYPE_PRIMARY_KEY("rowtype_primary_key", "primary keys"),
     ROWTYPE_UNIQUE_KEY("rowtype_unique_key", "unique keys"),
+    ROWTYPE_DUPLICATE_KEY("rowtype_duplicate_key", "duplicate keys"),
     ROWTYPE_FIELDS("rowtype_fields", "fields"),
     TABLE_NAME("table_name", "table name"),
     DATABASE("database", "database");

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Catalog.java

@@ -131,6 +131,21 @@ default Optional<Factory> getFactory() {
      */
     CatalogTable getTable(TablePath tablePath) throws CatalogException, TableNotExistException;
 
+    /**
+     * Return a {@link CatalogTable} identified by the given {@link TablePath} and field names. The
+     * framework will resolve the metadata objects when necessary.
+     *
+     * @param tablePath Path of the table
+     * @param fieldNames The field names need read
+     * @return The requested table
+     * @throws CatalogException in case of any runtime exception
+     */
+    default CatalogTable getTable(TablePath tablePath, List<String> fieldNames)
+            throws CatalogException, TableNotExistException {
+        throw CommonError.unsupportedOperation(
+                name(), "get table with tablePath " + tablePath + ", fieldNames: " + fieldNames);
+    }
+
     default List<CatalogTable> getTables(ReadonlyConfig config) throws CatalogException {
         // Get the list of specified tables
         List<String> tableNames = config.get(CatalogOptions.TABLE_NAMES);

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/converter/BasicDataConverter.java

@@ -19,7 +19,6 @@
 
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.type.ArrayType;
-import org.apache.seatunnel.api.table.type.BasicType;
 import org.apache.seatunnel.api.table.type.MapType;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -164,7 +163,7 @@ default Object[] convertArray(T typeDefine, Column columnDefine, Object value)
     default Object[] convertArray(ArrayType typeDefine, Object value)
             throws UnsupportedOperationException {
         if (value.getClass().isArray()) {
-            BasicType elementType = typeDefine.getElementType();
+            SeaTunnelDataType elementType = typeDefine.getElementType();
 
             Object[] array = (Object[]) value;
             for (int i = 0; i < array.length; i++) {
@@ -173,7 +172,7 @@ default Object[] convertArray(ArrayType typeDefine, Object value)
             return array;
         }
         if (value instanceof List) {
-            BasicType elementType = typeDefine.getElementType();
+            SeaTunnelDataType elementType = typeDefine.getElementType();
 
             List<Object> list = (List<Object>) value;
             int elements = list.size();
@@ -183,7 +182,7 @@ default Object[] convertArray(ArrayType typeDefine, Object value)
             return list.toArray();
         }
         if (value instanceof Set) {
-            BasicType elementType = typeDefine.getElementType();
+            SeaTunnelDataType elementType = typeDefine.getElementType();
 
             return ((Set) value).stream().map(e -> convert(elementType, e)).toArray();
         }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/ArrayType.java

@@ -39,17 +39,26 @@ public class ArrayType<T, E> implements SeaTunnelDataType<T> {
     public static final ArrayType<Double[], Double> DOUBLE_ARRAY_TYPE =
             new ArrayType<>(Double[].class, BasicType.DOUBLE_TYPE);
 
+    public static final ArrayType<LocalTimeType[], LocalTimeType> LOCAL_DATE_ARRAY_TYPE =
+            new ArrayType(LocalTimeType[].class, LocalTimeType.LOCAL_DATE_TYPE);
+
+    public static final ArrayType<LocalTimeType[], LocalTimeType> LOCAL_TIME_ARRAY_TYPE =
+            new ArrayType(LocalTimeType[].class, LocalTimeType.LOCAL_TIME_TYPE);
+
+    public static final ArrayType<LocalTimeType[], LocalTimeType> LOCAL_DATE_TIME_ARRAY_TYPE =
+            new ArrayType(LocalTimeType[].class, LocalTimeType.LOCAL_DATE_TIME_TYPE);
+
     // --------------------------------------------------------------------------------------------
 
     private final Class<T> arrayClass;
-    private final BasicType<E> elementType;
+    private final SeaTunnelDataType<E> elementType;
 
-    private ArrayType(Class<T> arrayClass, BasicType<E> elementType) {
+    protected ArrayType(Class<T> arrayClass, SeaTunnelDataType<E> elementType) {
         this.arrayClass = arrayClass;
         this.elementType = elementType;
     }
 
-    public BasicType<E> getElementType() {
+    public SeaTunnelDataType<E> getElementType() {
         return elementType;
     }
 

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/DecimalArrayType.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.table.type;
+
+public class DecimalArrayType extends ArrayType {
+    private static final long serialVersionUID = 1L;
+
+    public static final Class arrayClass = DecimalType[].class;
+
+    public DecimalArrayType(DecimalType elementType) {
+        super(arrayClass, elementType);
+    }
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java

@@ -142,6 +142,26 @@ private int getBytesForValue(Object v, SeaTunnelDataType<?> dataType) {
             case TIMESTAMP:
                 return 48;
             case ARRAY:
+                SeaTunnelDataType elementType = ((ArrayType) dataType).getElementType();
+                if (elementType instanceof DecimalType) {
+                    return ((Object[]) v).length * 36;
+                }
+
+                if (elementType instanceof LocalTimeType) {
+                    SqlType eleSqlType = elementType.getSqlType();
+                    switch (eleSqlType) {
+                        case DATE:
+                            return ((Object[]) v).length * 24;
+                        case TIME:
+                            return ((Object[]) v).length * 12;
+                        case TIMESTAMP:
+                            return ((Object[]) v).length * 48;
+                        default:
+                            throw new UnsupportedOperationException(
+                                    "Unsupported type in LocalTimeArrayType: " + eleSqlType);
+                    }
+                }
+
                 return getBytesForArray(v, ((ArrayType) dataType).getElementType());
             case MAP:
                 int size = 0;
@@ -166,7 +186,7 @@ private int getBytesForValue(Object v, SeaTunnelDataType<?> dataType) {
         }
     }
 
-    private int getBytesForArray(Object v, BasicType<?> dataType) {
+    private int getBytesForArray(Object v, SeaTunnelDataType<?> dataType) {
         switch (dataType.getSqlType()) {
             case STRING:
                 int s = 0;

seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/CommonError.java

@@ -36,9 +36,11 @@
 import static org.apache.seatunnel.common.exception.CommonErrorCode.GET_CATALOG_TABLES_WITH_UNSUPPORTED_TYPE_ERROR;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.GET_CATALOG_TABLE_WITH_UNSUPPORTED_TYPE_ERROR;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.JSON_OPERATION_FAILED;
+import static org.apache.seatunnel.common.exception.CommonErrorCode.OPERATION_NOT_SUPPORTED;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.SQL_TEMPLATE_HANDLED_ERROR;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.UNSUPPORTED_DATA_TYPE;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.UNSUPPORTED_ENCODING;
+import static org.apache.seatunnel.common.exception.CommonErrorCode.VERSION_NOT_SUPPORTED;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.WRITE_SEATUNNEL_ROW_ERROR;
 
 /**
@@ -96,6 +98,13 @@ public static SeaTunnelRuntimeException unsupportedDataType(
         return new SeaTunnelRuntimeException(UNSUPPORTED_DATA_TYPE, params);
     }
 
+    public static SeaTunnelRuntimeException unsupportedVersion(String identifier, String version) {
+        Map<String, String> params = new HashMap<>();
+        params.put("identifier", identifier);
+        params.put("version", version);
+        return new SeaTunnelRuntimeException(VERSION_NOT_SUPPORTED, params);
+    }
+
     public static SeaTunnelRuntimeException unsupportedEncoding(String encoding) {
         Map<String, String> params = new SingletonMap<>("encoding", encoding);
         return new SeaTunnelRuntimeException(UNSUPPORTED_ENCODING, params);
@@ -185,6 +194,14 @@ public static SeaTunnelRuntimeException jsonOperationError(
         }
     }
 
+    public static SeaTunnelRuntimeException unsupportedOperation(
+            String identifier, String operation) {
+        Map<String, String> params = new HashMap<>();
+        params.put("identifier", identifier);
+        params.put("operation", operation);
+        return new SeaTunnelRuntimeException(OPERATION_NOT_SUPPORTED, params);
+    }
+
     public static SeaTunnelRuntimeException sqlTemplateHandledError(
             String tableName,
             String keyName,

seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/CommonErrorCode.java

@@ -49,9 +49,14 @@ public enum CommonErrorCode implements SeaTunnelErrorCode {
     WRITE_SEATUNNEL_ROW_ERROR(
             "COMMON-23",
             "<connector> write SeaTunnelRow failed, the SeaTunnelRow value is '<seaTunnelRow>'."),
+
     SQL_TEMPLATE_HANDLED_ERROR(
             "COMMON-24",
-            "The table of <tableName> has no <keyName>, but the template \n <template> \n which has the place holder named <placeholder>. Please use the option named <optionName> to specify sql template");
+            "The table of <tableName> has no <keyName>, but the template \n <template> \n which has the place holder named <placeholder>. Please use the option named <optionName> to specify sql template"),
+
+    VERSION_NOT_SUPPORTED("COMMON-25", "<identifier> <version> is unsupported."),
+
+    OPERATION_NOT_SUPPORTED("COMMON-26", "<identifier> <operation> is unsupported.");
 
     private final String code;
     private final String description;

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/serialize/DefaultSeaTunnelRowSerializer.java

@@ -18,7 +18,6 @@
 package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize;
 
 import org.apache.seatunnel.api.table.type.ArrayType;
-import org.apache.seatunnel.api.table.type.BasicType;
 import org.apache.seatunnel.api.table.type.MapType;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -171,7 +170,7 @@ private AttributeValue convertItem(
                 return AttributeValue.builder().m(resultMap).build();
             case L:
                 ArrayType<?, ?> arrayType = (ArrayType<?, ?>) seaTunnelDataType;
-                BasicType<?> elementType = arrayType.getElementType();
+                SeaTunnelDataType<?> elementType = arrayType.getElementType();
                 Object[] l = (Object[]) value;
                 return AttributeValue.builder()
                         .l(