[format] Follow Parquet list backward-compatibility rules in schema clipping

paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetListLayoutResolver.java

@@ -0,0 +1,216 @@
+/*
+ * 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.paimon.format.parquet;
+
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.Type;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.format.parquet.ParquetSchemaConverter.LIST_ELEMENT_NAME;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Resolves Parquet list layouts, following the backward-compatibility rules in the Parquet spec: <a
+ * href="https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules">LogicalTypes#Backward-compatibility-rules</a>.
+ *
+ * <p>All list layout decisions should be made through this class so that schema inference,
+ * requested-schema clipping and reader construction share a single interpretation.
+ */
+public final class ParquetListLayoutResolver {
+
+    private static final String LIST_WRAPPER_NAME = "list";
+    private static final String LEGACY_LIST_ARRAY_NAME = "array";
+
+    private ParquetListLayoutResolver() {}
+
+    /** Returns true if the given group is annotated as a Parquet LIST logical type. */
+    public static boolean isList(GroupType listType) {
+        return listType.getLogicalTypeAnnotation()
+                instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation;
+    }
+
+    /**
+     * Whether the given group has the legacy nested-list shape: an unannotated {@code REPEATED}
+     * group whose single child is also {@code REPEATED}.
+     *
+     * <p>This is the Parquet spec's backward-compatibility <b>Rule 3</b> and appears as the element
+     * of an annotated list, e.g. {@code repeated group element { repeated int32 array; }}. The
+     * repeated child is the element of the nested list.
+     */
+    public static boolean isLegacyNestedList(GroupType groupType) {
+        return groupType.isRepetition(Type.Repetition.REPEATED)
+                && groupType.getFieldCount() == 1
+                && groupType.getType(0).getRepetition() == Type.Repetition.REPEATED;
+    }
+
+    /**
+     * Returns true if the given group is a three-level Parquet list.
+     *
+     * <p>In a three-level list the immediate repeated child is a wrapper group whose single
+     * non-repeated child is the actual element type. This covers the canonical layout ({@code list
+     * -> element}) as well as legacy wrappers such as Hive's {@code bag} layout.
+     *
+     * <p>This corresponds to the Parquet spec's backward-compatibility <b>Rule 5</b>: a repeated
+     * group that contains exactly one non-repeated child is a wrapper, unless it matches one of
+     * Rules 1-4.
+     *
+     * <p>The compatibility encodings that are <em>not</em> three-level are:
+     *
+     * <ul>
+     *   <li><b>Rule 1</b>: the repeated field is a primitive and is itself the element type.
+     *   <li><b>Rule 2</b>: the repeated field is a group with multiple fields and is itself the
+     *       element type.
+     *   <li><b>Rule 3</b>: the repeated field is a group whose single child is also repeated; the
+     *       group itself is the element type.
+     *   <li><b>Rule 4</b>: the repeated field is a group named {@code "array"} or {@code
+     *       "<list>_tuple"} with a single child; the group itself is the element type.
+     * </ul>
+     */
+    public static boolean isThreeLevelList(Type type) {
+        if (type.isPrimitive()) {
+            return false;
+        }
+
+        GroupType listType = type.asGroupType();
+        if (!isList(listType)) {
+            return false;
+        }
+
+        // A list must have exactly one repeated child (the middle level).
+        if (listType.getFieldCount() != 1) {
+            return false;
+        }
+        Type middle = listType.getType(0);
+        if (middle.isPrimitive() || middle.getRepetition() != Type.Repetition.REPEATED) {
+            return false;
+        }
+        GroupType repeatedGroup = middle.asGroupType();
+
+        // Rule 5: the repeated group is a wrapper containing exactly one non-repeated child.
+        if (repeatedGroup.getFieldCount() != 1
+                || repeatedGroup.getType(0).getRepetition() == Type.Repetition.REPEATED) {
+            return false;
+        }
+
+        // Rule 4: legacy "array" and "<list>_tuple" encodings are not wrappers; the repeated
+        // group itself is the element type.
+        return !LEGACY_LIST_ARRAY_NAME.equals(repeatedGroup.getName())
+                && !(listType.getName() + "_tuple").equals(repeatedGroup.getName());
+    }
+
+    /**
+     * Returns true if the given group follows the canonical three-level Parquet list layout ({@code
+     * list -> element}).
+     *
+     * <p>The canonical layout is described in the Parquet spec: <a
+     * href="https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#lists">LogicalTypes#Lists</a>
+     */
+    public static boolean isCanonicalList(Type type) {
+        if (!isThreeLevelList(type)) {
+            return false;
+        }
+
+        Type middle = type.asGroupType().getType(0);
+        Type element = middle.asGroupType().getType(0);
+        return LIST_WRAPPER_NAME.equals(middle.getName())
+                && LIST_ELEMENT_NAME.equals(element.getName());
+    }
+
+    /**
+     * Resolves the element type of the given LIST-annotated group according to the Parquet spec's
+     * backward-compatibility rules for lists.
+     *
+     * <p>For a three-level list (Rule 5) the returned type is the single child of the repeated
+     * wrapper. For Rules 1-4 the repeated field itself is returned because it is the element type.
+     */
+    public static Type resolveElementType(GroupType listType) {
+        checkArgument(
+                isList(listType) || isLegacyNestedList(listType),
+                "Expected LIST-annotated group but got: %s",
+                listType);
+
+        if (isThreeLevelList(listType)) {
+            return listType.getType(0).asGroupType().getType(0);
+        }
+
+        return listType.getType(0);
+    }
+
+    /**
+     * A schema-level manifest of list layouts, modeled after parquet-cpp's {@code SchemaManifest}.
+     *
+     * <p>Built once from the file schema, it resolves the three-level verdict of every
+     * LIST-annotated node in the file and records it by field path, so that reader construction
+     * interprets list layouts against the file schema rather than the (possibly reshaped) requested
+     * schema. Field paths are the identifier shared by the file schema, the requested schema and
+     * the ColumnIO tree: clipping may rebuild nodes, but it preserves names, so paths stay stable
+     * where node identity would not.
+     *
+     * <p>Paths absent from this context belong to nodes that are not LIST-annotated in the file
+     * schema (or synthetic fill fields, whose requested shape is Paimon-canonical); the verdict
+     * falls back to interpreting the requested node itself.
+     */
+    public static final class LayoutContext {
+        private final Map<List<String>, Boolean> threeLevelMapping = new HashMap<>();
+
+        /** Builds the manifest by resolving every LIST-annotated node of the file schema. */
+        public static LayoutContext fromFileSchema(GroupType fileSchema) {
+            LayoutContext context = new LayoutContext();
+            for (Type field : fileSchema.getFields()) {
+                collect(field, Collections.singletonList(field.getName()), context);
+            }
+            return context;
+        }
+
+        private static void collect(Type node, List<String> path, LayoutContext context) {
+            if (node.isPrimitive()) {
+                return;
+            }
+            GroupType group = node.asGroupType();
+            if (isList(group)) {
+                context.threeLevelMapping.put(
+                        path, ParquetListLayoutResolver.isThreeLevelList(group));
+            }
+            for (Type child : group.getFields()) {
+                List<String> childPath = new ArrayList<>(path);
+                childPath.add(child.getName());
+                collect(child, childPath, context);
+            }
+        }
+
+        /**
+         * Returns whether the list at {@code path} is a three-level list, as resolved against the
+         * file schema; falls back to interpreting {@code requestedGroup} itself when the path is
+         * not an annotated list in the file schema.
+         */
+        public boolean isThreeLevelList(GroupType requestedGroup, String[] path) {
+            Boolean threeLevel = threeLevelMapping.get(Arrays.asList(path));
+            return threeLevel != null
+                    ? threeLevel
+                    : ParquetListLayoutResolver.isThreeLevelList(requestedGroup);
+        }
+    }
+}

paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java

@@ -26,6 +26,7 @@
 import org.apache.paimon.data.shredding.ShreddingReadPlan;
 import org.apache.paimon.format.FormatMetadataUtils;
 import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.format.parquet.ParquetListLayoutResolver.LayoutContext;
 import org.apache.paimon.format.parquet.reader.VectorizedParquetRecordReader;
 import org.apache.paimon.format.parquet.type.ParquetField;
 import org.apache.paimon.format.shredding.ShreddingFormatReader;
@@ -56,7 +57,6 @@
 import org.apache.parquet.schema.GroupType;
 import org.apache.parquet.schema.LogicalTypeAnnotation;
 import org.apache.parquet.schema.MessageType;
-import org.apache.parquet.schema.OriginalType;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.parquet.schema.Type;
 import org.apache.parquet.schema.Types;
@@ -76,8 +76,11 @@
 import java.util.function.IntFunction;
 
 import static org.apache.paimon.data.columnar.ColumnVectorUtils.createParquetWritableColumnVector;
+import static org.apache.paimon.format.parquet.ParquetListLayoutResolver.isLegacyNestedList;
+import static org.apache.paimon.format.parquet.ParquetListLayoutResolver.isList;
+import static org.apache.paimon.format.parquet.ParquetListLayoutResolver.isThreeLevelList;
+import static org.apache.paimon.format.parquet.ParquetListLayoutResolver.resolveElementType;
 import static org.apache.paimon.format.parquet.ParquetSchemaConverter.PAIMON_SCHEMA;
-import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetListElementType;
 import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetMapKeyValueType;
 import static org.apache.paimon.format.parquet.reader.ParquetReaderUtil.buildFieldsList;
 
@@ -239,9 +242,10 @@ private static DataField[] readFields(RowType readType) {
     }
 
     private RequestedSchema createRequestedSchema(MessageType fileSchema, DataField[] readFields) {
+        LayoutContext listLayout = LayoutContext.fromFileSchema(fileSchema);
         MessageType rs = clipParquetSchema(fileSchema, readFields);
         MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(rs);
-        List<ParquetField> f = buildFieldsList(readFields, columnIO, rs);
+        List<ParquetField> f = buildFieldsList(readFields, columnIO, rs, listLayout);
         return new RequestedSchema(rs, f);
     }
 
@@ -343,31 +347,37 @@ private Type clipParquetType(DataType readType, Type parquetType) {
                 Preconditions.checkArgument(
                         listSubFields == 1,
                         "Parquet list group type should only have one middle level REPEATED field.");
-                // There are two representations for array type in parquet.
-                // See link:
-                // https://impala.apache.org/docs/build/html/topics/impala_parquet_array_resolution.html.
-                int level = arrayGroup.getType(0) instanceof GroupType ? 3 : 2;
-                Type elementType =
-                        clipParquetType(elementReadType, parquetListElementType(arrayGroup));
-
-                if (level == 3) {
-                    // In case that the name in middle level is not "list".
-                    Type groupMiddle =
-                            new GroupType(
-                                    Type.Repetition.REPEATED,
-                                    arrayGroup.getType(0).getName(),
-                                    elementType);
-                    return new GroupType(
-                            arrayGroup.getRepetition(),
-                            arrayGroup.getName(),
-                            OriginalType.LIST,
-                            groupMiddle);
+                if (isList(arrayGroup)) {
+                    boolean threeLevel = isThreeLevelList(arrayGroup);
+                    Type originalElement = resolveElementType(arrayGroup);
+                    Type elementType = clipParquetType(elementReadType, originalElement);
+                    if (threeLevel) {
+                        Type clippedMiddle =
+                                arrayGroup
+                                        .getType(0)
+                                        .asGroupType()
+                                        .withNewFields(Collections.singletonList(elementType));
+                        return arrayGroup.withNewFields(Collections.singletonList(clippedMiddle));
+                    } else {
+                        // Rules 1-4: the repeated field itself is the element. Keep it (clipped)
+                        // in place so that reader construction unwraps by path against the file
+                        // schema instead of guessing from this reshaped requested shape.
+                        return arrayGroup.withNewFields(Collections.singletonList(elementType));
+                    }
+                } else if (isLegacyNestedList(arrayGroup)) {
+                    // Rule 3: an unannotated repeated group is itself the element of the annotated
+                    // outer list, and its single repeated child is the element of the nested inner
+                    // list.
+                    Type originalElement = arrayGroup.getType(0);
+                    Type elementType = clipParquetType(elementReadType, originalElement);
+                    return arrayGroup.withNewFields(Collections.singletonList(elementType));
                 } else {
-                    return new GroupType(
-                            arrayGroup.getRepetition(),
-                            arrayGroup.getName(),
-                            OriginalType.LIST,
-                            elementType);
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "Cannot read Parquet group '%s' as an ARRAY: it is neither "
+                                            + "LIST-annotated nor a legacy nested list. Parquet type: %s, "
+                                            + "read type: %s",
+                                    arrayGroup.getName(), arrayGroup, readType));
                 }
             default:
                 return parquetType;

paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java

@@ -462,10 +462,18 @@ public static DataField convertToPaimonField(Type parquetType) {
                     parquetType.getId().intValue(), parquetType.getName(), paimonDataType);
         } else {
             GroupType groupType = parquetType.asGroupType();
-            if (logicalType instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation) {
+            if (ParquetListLayoutResolver.isList(groupType)) {
+                Type parquetElementType = ParquetListLayoutResolver.resolveElementType(groupType);
+                DataType elementDataType = convertToPaimonField(parquetElementType).type();
+                if (!ParquetListLayoutResolver.isThreeLevelList(groupType)) {
+                    // Rules 1-4: the repeated node itself is the element. A REPEATED node is
+                    // never null, so the element type is not nullable.
+                    elementDataType = elementDataType.notNull();
+                }
+                paimonDataType = new ArrayType(elementDataType);
+            } else if (ParquetListLayoutResolver.isLegacyNestedList(groupType)) {
                 paimonDataType =
-                        new ArrayType(
-                                convertToPaimonField(parquetListElementType(groupType)).type());
+                        new ArrayType(convertToPaimonField(groupType.getType(0)).type().notNull());
             } else if (logicalType instanceof LogicalTypeAnnotation.MapLogicalTypeAnnotation) {
                 Pair<Type, Type> keyValueType = parquetMapKeyValueType(groupType);
                 paimonDataType =
@@ -490,22 +498,6 @@ public static DataField convertToPaimonField(Type parquetType) {
         return new DataField(parquetType.getId().intValue(), parquetType.getName(), paimonDataType);
     }
 
-    public static Type parquetListElementType(GroupType listType) {
-        int level = listType.getType(0) instanceof GroupType ? 3 : 2;
-        if (level == 3) {
-            // Level 3 representation of list type.
-            // List type should only have one middle group type, which is repeated, and one element
-            // type, which is optional.
-            return listType.getType(0).asGroupType().getType(0);
-        } else if (level == 2) {
-            // Level 2 representation of list type
-            return listType.getType(0);
-        } else {
-            throw new UnsupportedOperationException(
-                    "Parquet list type only have two level representation and three level representation.");
-        }
-    }
-
     public static Pair<Type, Type> parquetMapKeyValueType(GroupType mapType) {
         GroupType keyValue = mapType.getType(0).asGroupType();
         return Pair.of(keyValue.getType(MAP_KEY_NAME), keyValue.getType(MAP_VALUE_NAME));

paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantShreddingReadPlanFactory.java

@@ -56,7 +56,7 @@
 
 import static org.apache.paimon.data.variant.Variant.METADATA;
 import static org.apache.paimon.data.variant.Variant.VALUE;
-import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetListElementType;
+import static org.apache.paimon.format.parquet.ParquetListLayoutResolver.resolveElementType;
 import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetMapKeyValueType;
 
 /**
@@ -194,7 +194,7 @@ private static DataType buildPhysicalType(
                         arrayType.isNullable(),
                         buildPhysicalType(
                                 arrayType.getElementType(),
-                                parquetListElementType(fileType.asGroupType()),
+                                resolveElementType(fileType.asGroupType()),
                                 caseSensitive));
             case VECTOR:
                 VectorType vectorType = (VectorType) logicalType;
@@ -203,7 +203,7 @@ private static DataType buildPhysicalType(
                         vectorType.getLength(),
                         buildPhysicalType(
                                 vectorType.getElementType(),
-                                parquetListElementType(fileType.asGroupType()),
+                                resolveElementType(fileType.asGroupType()),
                                 caseSensitive));
             case MAP:
                 MapType mapType = (MapType) logicalType;