[Improvement](multi-catalog)support max compute complex type parse

fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeColumnValue.java

@@ -32,8 +32,12 @@
 import org.apache.arrow.vector.SmallIntVector;
 import org.apache.arrow.vector.TimeStampNanoVector;
 import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.ValueVector;
 import org.apache.arrow.vector.VarBinaryVector;
 import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.complex.StructVector;
 import org.apache.log4j.Logger;
 
 import java.math.BigDecimal;
@@ -49,13 +53,18 @@
 public class MaxComputeColumnValue implements ColumnValue {
     private static final Logger LOG = Logger.getLogger(MaxComputeColumnValue.class);
     private int idx;
-    private FieldVector column;
+    private ValueVector column;
 
     public MaxComputeColumnValue() {
         idx = 0;
     }
 
-    public void reset(FieldVector column) {
+    public MaxComputeColumnValue(ValueVector valueVector, int i) {
+        this.column = valueVector;
+        this.idx = i;
+    }
+
+    public void reset(ValueVector column) {
         this.column = column;
         this.idx = 0;
     }
@@ -222,16 +231,38 @@ public byte[] getBytes() {
 
     @Override
     public void unpackArray(List<ColumnValue> values) {
-
+        skippedIfNull();
+        ListVector listCol = (ListVector) column;
+        for (int i = 0; i < listCol.getDataVector().getValueCount(); i++) {
+            MaxComputeColumnValue val = new MaxComputeColumnValue(listCol.getDataVector(), i);
+            values.add(val);
+        }
     }
 
     @Override
     public void unpackMap(List<ColumnValue> keys, List<ColumnValue> values) {
-
+        skippedIfNull();
+        MapVector mapCol = (MapVector) column;
+        FieldVector keyList = mapCol.getDataVector().getChildrenFromFields().get(0);
+        for (int i = 0; i < keyList.getValueCount(); i++) {
+            MaxComputeColumnValue val = new MaxComputeColumnValue(keyList, i);
+            keys.add(val);
+        }
+        FieldVector valList = mapCol.getDataVector().getChildrenFromFields().get(1);
+        for (int i = 0; i < valList.getValueCount(); i++) {
+            MaxComputeColumnValue val = new MaxComputeColumnValue(valList, i);
+            values.add(val);
+        }
     }
 
     @Override
     public void unpackStruct(List<Integer> structFieldIndex, List<ColumnValue> values) {
-
+        skippedIfNull();
+        StructVector structCol = (StructVector) column;
+        for (Integer fieldIndex : structFieldIndex) {
+            MaxComputeColumnValue val = new MaxComputeColumnValue(structCol.getChildByOrdinal(fieldIndex), idx);
+            values.add(val);
+        }
+        idx++;
     }
 }

fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java

@@ -190,6 +190,11 @@ public void open() throws IOException {
     }
 
     private Column createOdpsColumn(int colIdx, ColumnType dorisType) {
+        TypeInfo odpsType = getOdpsType(dorisType);
+        return new Column(fields[colIdx], odpsType);
+    }
+
+    private static TypeInfo getOdpsType(ColumnType dorisType) {
         TypeInfo odpsType;
         switch (dorisType.getType()) {
             case BOOLEAN:
@@ -236,10 +241,27 @@ private Column createOdpsColumn(int colIdx, ColumnType dorisType) {
             case STRING:
                 odpsType = TypeInfoFactory.getPrimitiveTypeInfo(OdpsType.STRING);
                 break;
+            case ARRAY:
+                TypeInfo elementType = getOdpsType(dorisType.getChildTypes().get(0));
+                odpsType = TypeInfoFactory.getArrayTypeInfo(elementType);
+                break;
+            case MAP:
+                TypeInfo keyType = getOdpsType(dorisType.getChildTypes().get(0));
+                TypeInfo valueType = getOdpsType(dorisType.getChildTypes().get(1));
+                odpsType = TypeInfoFactory.getMapTypeInfo(keyType, valueType);
+                break;
+            case STRUCT:
+                List<String> names = dorisType.getChildNames();
+                List<TypeInfo> typeInfos = new ArrayList<>();
+                for (ColumnType childType : dorisType.getChildTypes()) {
+                    typeInfos.add(getOdpsType(childType));
+                }
+                odpsType = TypeInfoFactory.getStructTypeInfo(names, typeInfos);
+                break;
             default:
                 throw new RuntimeException("Unsupported transform for column type: " + dorisType.getType());
         }
-        return new Column(fields[colIdx], odpsType);
+        return odpsType;
     }
 
     @Override

regression-test/pipeline/external/conf/regression-conf.groovy

@@ -145,6 +145,9 @@ aliYunSk="***********"
 txYunAk="***********"
 txYunSk="***********"
 
+// max compute catalog test config
+enableMaxComputeTest=true
+
 max_failure_num=50
 
 externalEnvIp="127.0.0.1"

regression-test/suites/external_table_p0/mc/test_max_compute_complex_type.groovy

@@ -0,0 +1,48 @@
+// 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.
+
+suite("test_max_compute_complex_type", "p0,external,doris,external_docker,external_docker_doris") {
+    String enabled = context.config.otherConfigs.get("enableMaxComputeTest")
+    if (enabled != null && enabled.equalsIgnoreCase("true")) {
+        String ak = context.config.otherConfigs.get("aliYunAk")
+        String sk = context.config.otherConfigs.get("aliYunSk")
+        String mc_catalog_name = "test_max_compute_complex_type"
+        sql """drop catalog if exists ${mc_catalog_name} """
+        sql """
+        CREATE CATALOG IF NOT EXISTS ${mc_catalog_name} PROPERTIES (
+                "type" = "max_compute",
+                "mc.default.project" = "mc_datalake",
+                "mc.region" = "cn-beijing",
+                "mc.access_key" = "${ak}",
+                "mc.secret_key" = "${sk}",
+                "mc.public_access" = "true"
+        );
+        """
+
+        logger.info("catalog " + mc_catalog_name + " created")
+        sql """switch ${mc_catalog_name};"""
+        logger.info("switched to catalog " + mc_catalog_name)
+        sql """ use mc_datalake """
+
+        qt_mc_q1 """ select id,arr3,arr1,arr5,arr2 from array_table order by id desc """
+        qt_mc_q2 """ select arr2,arr1 from map_table order by id limit 2 """
+        qt_mc_q3 """ select contact_info,user_info from struct_table order by id limit 2 """
+        qt_mc_q4 """ select user_id,activity_log from nested_complex_table order by user_id limit 2 """
+
+        sql """drop catalog ${mc_catalog_name};"""
+    }
+}