[ISSUE-154] Support Array[Tuple] clickhouse type (#155)

* [ISSUE-154] Support Array[Tuple] clickhouse type

* [ISSUE-154] Support Array[Tuple] clickhouse type

* [ISSUE-154] Support Array[Tuple] clickhouse type for clickhousecatalog

---------

Co-authored-by: jennychen <jennyhq.chen@huolala.cn>

Author: qingfei1994

flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/converter/ClickHouseConverterUtils.java

@@ -21,13 +21,16 @@
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.GenericArrayData;
 import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.RowType;
 
 import com.clickhouse.data.value.UnsignedByte;
 import com.clickhouse.data.value.UnsignedInteger;
@@ -46,7 +49,9 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
@@ -119,9 +124,18 @@ public static Object toExternal(Object value, LogicalType type) {
                             toExternal(valueGetter.getElementOrNull(valueArrayData, i), valueType));
                 }
                 return objectMap;
-            case MULTISET:
             case ROW:
+                List<Object> result = new ArrayList<>();
+                for (int i = 0; i < ((RowData) value).getArity(); i++) {
+                    result.add(
+                            toExternal(
+                                    RowData.createFieldGetter(((RowType) type).getTypeAt(i), i)
+                                            .getFieldOrNull((RowData) value),
+                                    ((RowType) type).getTypeAt(i)));
+                }
+                return result;
             case RAW:
+            case MULTISET:
             default:
                 throw new UnsupportedOperationException("Unsupported type:" + type);
         }
@@ -209,6 +223,12 @@ public static Object toInternal(Object value, LogicalType type) throws SQLExcept
                 }
                 return new GenericMapData(internalMap);
             case ROW:
+                List<Object> row = (List<Object>) value;
+                GenericRowData rowData = new GenericRowData(row.size());
+                for (int i = 0; i < row.size(); i++) {
+                    rowData.setField(i, toInternal(row.get(i), type.getChildren().get(i)));
+                }
+                return rowData;
             case MULTISET:
             case RAW:
             default:

flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/util/DataTypeUtil.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.connector.clickhouse.util;
 
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.table.types.DataType;
@@ -25,6 +26,7 @@
 
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static org.apache.flink.table.types.logical.DecimalType.MAX_PRECISION;
 
@@ -110,6 +112,12 @@ public static DataType toFlinkType(ClickHouseColumn clickHouseColumnInfo) {
                         toFlinkType(clickHouseColumnInfo.getKeyInfo()),
                         toFlinkType(clickHouseColumnInfo.getValueInfo()));
             case Tuple:
+                return DataTypes.ROW(
+                        clickHouseColumnInfo.getNestedColumns().stream()
+                                .map((col) -> new Tuple2<>(col, toFlinkType(col)))
+                                .map(tuple -> DataTypes.FIELD(tuple.f0.getColumnName(), tuple.f1))
+                                .collect(Collectors.toList()));
+
             case Nested:
             case AggregateFunction:
             default: