[INLONG-2614][Sort] Support array and map data structures in Hive sink and ClickHouse sink

inlong-sort/sort-connectors/src/main/java/org/apache/inlong/sort/flink/clickhouse/ClickHouseRowConverter.java

@@ -18,6 +18,9 @@
 
 package org.apache.inlong.sort.flink.clickhouse;
 
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.flink.shaded.guava18.com.google.common.annotations.VisibleForTesting;
+import org.apache.inlong.sort.formats.common.ArrayTypeInfo;
 import org.apache.inlong.sort.formats.common.BooleanTypeInfo;
 import org.apache.inlong.sort.formats.common.ByteTypeInfo;
 import org.apache.inlong.sort.formats.common.DateTypeInfo;
@@ -27,19 +30,24 @@
 import org.apache.inlong.sort.formats.common.FormatInfo;
 import org.apache.inlong.sort.formats.common.IntTypeInfo;
 import org.apache.inlong.sort.formats.common.LongTypeInfo;
+import org.apache.inlong.sort.formats.common.MapTypeInfo;
 import org.apache.inlong.sort.formats.common.ShortTypeInfo;
 import org.apache.inlong.sort.formats.common.StringTypeInfo;
 import org.apache.inlong.sort.formats.common.TimeTypeInfo;
 import org.apache.inlong.sort.formats.common.TimestampTypeInfo;
 import org.apache.inlong.sort.formats.common.TypeInfo;
 import org.apache.flink.types.Row;
+import ru.yandex.clickhouse.ClickHouseArray;
+import ru.yandex.clickhouse.domain.ClickHouseDataType;
+import ru.yandex.clickhouse.util.Utils;
 
 import java.math.BigDecimal;
 import java.sql.Date;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Time;
 import java.sql.Timestamp;
+import java.util.Map;
 
 public class ClickHouseRowConverter {
 
@@ -83,8 +91,70 @@ private static void setField(
             statement.setDate(index + 1, (Date) value);
         } else if (typeInfo instanceof TimestampTypeInfo) {
             statement.setTimestamp(index + 1, (Timestamp) value);
+        } else if (typeInfo instanceof ArrayTypeInfo) {
+            TypeInfo elementTypeInfo = ((ArrayTypeInfo) typeInfo).getElementTypeInfo();
+            statement.setArray(index + 1, new ClickHouseArray(
+                    getClickHouseDataTypeFromTypeInfo(elementTypeInfo), toObjectArray(elementTypeInfo, value)));
+        } else if (typeInfo instanceof MapTypeInfo) {
+            statement.setObject(index + 1, Utils.mapOf(toKeyValuePairObjectArray(value)));
         } else {
             throw new IllegalArgumentException("Unsupported TypeInfo " + typeInfo.getClass().getName());
         }
     }
+
+    private static ClickHouseDataType getClickHouseDataTypeFromTypeInfo(TypeInfo typeInfo) {
+        if (typeInfo instanceof StringTypeInfo) {
+            return ClickHouseDataType.String;
+        } else if (typeInfo instanceof BooleanTypeInfo || typeInfo instanceof ByteTypeInfo) {
+            return ClickHouseDataType.Int8;
+        } else if (typeInfo instanceof ShortTypeInfo) {
+            return ClickHouseDataType.Int16;
+        } else if (typeInfo instanceof IntTypeInfo) {
+            return ClickHouseDataType.Int32;
+        } else if (typeInfo instanceof LongTypeInfo) {
+            return ClickHouseDataType.Int64;
+        } else if (typeInfo instanceof FloatTypeInfo) {
+            return ClickHouseDataType.Float32;
+        } else if (typeInfo instanceof DoubleTypeInfo) {
+            return ClickHouseDataType.Float64;
+        } else {
+            throw new IllegalArgumentException("Unsupported TypeInfo " + typeInfo.getClass().getName());
+        }
+    }
+
+    @VisibleForTesting
+    static Object toObjectArray(TypeInfo typeInfo, Object object) {
+        if (typeInfo instanceof BooleanTypeInfo && object instanceof boolean[]) {
+            return ArrayUtils.toObject((boolean[]) object);
+        } else if (typeInfo instanceof ByteTypeInfo && object instanceof byte[]) {
+            return ArrayUtils.toObject((byte[]) object);
+        } else if (typeInfo instanceof ShortTypeInfo && object instanceof short[]) {
+            return ArrayUtils.toObject((short[]) object);
+        } else if (typeInfo instanceof IntTypeInfo && object instanceof int[]) {
+            return ArrayUtils.toObject((int[]) object);
+        } else if (typeInfo instanceof LongTypeInfo && object instanceof long[]) {
+            return ArrayUtils.toObject((long[]) object);
+        } else if (typeInfo instanceof FloatTypeInfo && object instanceof float[]) {
+            return ArrayUtils.toObject((float[]) object);
+        } else if (typeInfo instanceof DoubleTypeInfo && object instanceof double[]) {
+            return ArrayUtils.toObject((double[]) object);
+        } else {
+            return object;
+        }
+    }
+
+    @VisibleForTesting
+    static Object[] toKeyValuePairObjectArray(Object input) {
+        Map<?, ?> mapValue = (Map<?, ?>) input;
+        int size = mapValue.size();
+        Object[] kvps = new Object[size * 2];
+        int i = 0;
+        for (Map.Entry<?, ?> entry : mapValue.entrySet()) {
+            kvps[i] = entry.getKey();
+            kvps[i + 1] = entry.getValue();
+            i += 2;
+        }
+
+        return kvps;
+    }
 }

inlong-sort/sort-connectors/src/main/java/org/apache/inlong/sort/flink/hive/HiveSinkHelper.java

@@ -51,7 +51,7 @@ public static BulkWriter.Factory<Row> createBulkWriterFactory(
             return ParquetRowWriterBuilder.createWriterFactory(
                     rowType, (ParquetFileFormat) hiveFileFormat);
         } else if (hiveFileFormat instanceof TextFileFormat) {
-            return new TextRowWriter.Factory((TextFileFormat) hiveFileFormat, config);
+            return new TextRowWriter.Factory((TextFileFormat) hiveFileFormat, fieldTypes, config);
         } else if (hiveFileFormat instanceof OrcFileFormat) {
             return OrcBulkWriterFactory.createWriterFactory(rowType, fieldTypes, config);
         } else {