You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/10/23 01:58:46 UTC

(hudi) branch master updated: fix(flink): apply the nested array reader fix to Flink 1.19 and later (#20399)

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 617be49f0ab5 fix(flink): apply the nested array reader fix to Flink 1.19 and later (#20399)
617be49f0ab5 is described below

commit 617be49f0ab5497174dcab098e258c648491f4ee
Author: Shihuan Liu <sk...@gmail.com>
AuthorDate: Thu Oct 22 18:58:40 2026 -0700

    fix(flink): apply the nested array reader fix to Flink 1.19 and later (#20399)
    
    Port #20337 from hudi-flink1.18.x to hudi-flink1.19.x, 1.20.x, 2.0.x, 2.1.x and 2.2.x,
    which have the same COW Parquet reader code:
    
    - createWritableColumnVector unwraps the Parquet LIST element of an ARRAY with the same
      backward-compatibility rules as getArrayElementColumn, now shared through
      isThreeLevelListRepeatedType, so reading array<row<..., array<row<...>>>> no longer fails
      with a NullPointerException in NestedColumnReader.readRow, and arrays of rows read their
      field values instead of nulls.
    - INT32/INT64-encoded decimals are allocated with int/long vectors.
    - NestedColumnReader grows the null placeholders of absent fields to the row's position
      count, and NestedPositionUtil.calculateRowOffsets returns the row's own position count
      when it has no nulls.
    
    The change and its tests are the same as in hudi-flink1.18.x. The 2.1.x and 2.2.x modules
    also support VARIANT, which the change leaves as is.
---
 .../table/format/cow/ParquetSplitReaderUtil.java   |  85 ++++-
 .../table/format/cow/utils/NestedPositionUtil.java |   4 +-
 .../cow/vector/reader/NestedColumnReader.java      |  13 +
 .../reader/ParquetColumnarRowSplitReader.java      |   1 +
 .../format/cow/TestParquetSplitReaderUtil.java     | 415 +++++++++++++++++++++
 .../format/cow/utils/TestNestedPositionUtil.java   |  66 ++++
 .../table/format/cow/ParquetSplitReaderUtil.java   |  85 ++++-
 .../table/format/cow/utils/NestedPositionUtil.java |   4 +-
 .../cow/vector/reader/NestedColumnReader.java      |  13 +
 .../reader/ParquetColumnarRowSplitReader.java      |   1 +
 .../format/cow/TestParquetSplitReaderUtil.java     | 415 +++++++++++++++++++++
 .../format/cow/utils/TestNestedPositionUtil.java   |  66 ++++
 .../table/format/cow/ParquetSplitReaderUtil.java   |  85 ++++-
 .../table/format/cow/utils/NestedPositionUtil.java |   4 +-
 .../cow/vector/reader/NestedColumnReader.java      |  13 +
 .../reader/ParquetColumnarRowSplitReader.java      |   1 +
 .../format/cow/TestParquetSplitReaderUtil.java     | 415 +++++++++++++++++++++
 .../format/cow/utils/TestNestedPositionUtil.java   |  66 ++++
 .../table/format/cow/ParquetSplitReaderUtil.java   |  85 ++++-
 .../table/format/cow/utils/NestedPositionUtil.java |   4 +-
 .../cow/vector/reader/NestedColumnReader.java      |  13 +
 .../reader/ParquetColumnarRowSplitReader.java      |   1 +
 .../format/cow/TestParquetSplitReaderUtil.java     | 415 +++++++++++++++++++++
 .../format/cow/utils/TestNestedPositionUtil.java   |  66 ++++
 .../table/format/cow/ParquetSplitReaderUtil.java   |  85 ++++-
 .../table/format/cow/utils/NestedPositionUtil.java |   4 +-
 .../cow/vector/reader/NestedColumnReader.java      |  13 +
 .../reader/ParquetColumnarRowSplitReader.java      |   1 +
 .../format/cow/TestParquetSplitReaderUtil.java     | 415 +++++++++++++++++++++
 .../format/cow/utils/TestNestedPositionUtil.java   |  66 ++++
 30 files changed, 2840 insertions(+), 80 deletions(-)

diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index 33da58dd3ad5..e702461b5203 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -19,6 +19,7 @@
 package org.apache.hudi.table.format.cow;
 
 import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
 import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
@@ -483,19 +484,24 @@ public class ParquetSplitReaderUtil {
    * matches {@code fieldType}; for ROW types missing physical fields are slotted with null-filled
    * vectors (sourced from {@link #createVectorFromConstant}) so that the Dremel assembler in
    * {@link NestedColumnReader} can pass them through unchanged.
+   *
+   * @param parentName name of the group that contains {@code physicalType}, which decides whether a
+   *     repeated group named {@code <parentName>_tuple} is a list element
    */
   public static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> descriptors) {
-    return createWritableColumnVector(batchSize, fieldType, physicalType, descriptors, 0);
+    return createWritableColumnVector(batchSize, fieldType, physicalType, parentName, descriptors, 0);
   }
 
   private static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> columns,
       int depth) {
     List<ColumnDescriptor> descriptors = filterDescriptors(depth, physicalType, columns);
@@ -555,26 +561,35 @@ public class ParquetSplitReaderUtil {
         return new HeapTimestampVector(batchSize);
       case DECIMAL:
         checkArgument(
-            (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
-                    || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
-                && primitiveType.getOriginalType() == OriginalType.DECIMAL,
+            primitiveType.getOriginalType() == OriginalType.DECIMAL,
             getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
-        return new HeapDecimalVector(batchSize);
+        switch (typeName) {
+          case INT32:
+            return new HeapIntVector(batchSize);
+          case INT64:
+            return new HeapLongVector(batchSize);
+          case BINARY:
+          case FIXED_LEN_BYTE_ARRAY:
+            return new HeapDecimalVector(batchSize);
+          default:
+            throw new IllegalArgumentException(getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+        }
       case ARRAY:
         ArrayType arrayType = (ArrayType) fieldType;
+        Triple<Type, String, Integer> element = unwrapListElementType(physicalType, parentName, depth);
         return new HeapArrayVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+            createWritableColumnVector(batchSize, arrayType.getElementType(), element.getLeft(),
+                element.getMiddle(), descriptors, element.getRight()));
       case MAP: {
         MapType mapType = (MapType) fieldType;
         GroupType repeatedType = unwrapMapRepeatedType(physicalType);
         return new HeapMapColumnVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, mapType.getKeyType(), repeatedType.getType(0), descriptors, depth + 2),
-            createWritableColumnVector(
-                batchSize, mapType.getValueType(), repeatedType.getType(1), descriptors, depth + 2));
+            createWritableColumnVector(batchSize, mapType.getKeyType(), repeatedType.getType(0),
+                repeatedType.getName(), descriptors, depth + 2),
+            createWritableColumnVector(batchSize, mapType.getValueType(), repeatedType.getType(1),
+                repeatedType.getName(), descriptors, depth + 2));
       }
       case MULTISET: {
         MultisetType multisetType = (MultisetType) fieldType;
@@ -585,12 +600,14 @@ public class ParquetSplitReaderUtil {
                 batchSize,
                 multisetType.getElementType(),
                 repeatedType.getType(0),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2),
             createWritableColumnVector(
                 batchSize,
                 new IntType(false),
                 repeatedType.getType(1),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2));
       }
@@ -612,6 +629,7 @@ public class ParquetSplitReaderUtil {
                     batchSize,
                     rowType.getTypeAt(i),
                     groupType.getType(fieldIndex),
+                    groupType.getName(),
                     descriptors,
                     depth + 1);
           }
@@ -622,6 +640,46 @@ public class ParquetSplitReaderUtil {
     }
   }
 
+  /**
+   * Returns the element type of a LIST physical type, the name of the group that contains it and its
+   * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+   * so that the vectors line up with the {@code ParquetField} tree.
+   *
+   * @param parentName name of the group that contains {@code listType}
+   */
+  private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+    Type elementType = listType;
+    String elementParentName = parentName;
+    int elementDepth = depth;
+    while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    return Triple.of(elementType, elementParentName, elementDepth);
+  }
+
+  /**
+   * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+   * rather than the element itself. Shared by {@link #unwrapListElementType} and
+   * {@link #getArrayElementColumn}, so that the vectors and the {@code ParquetField} tree pick the
+   * same element.
+   *
+   * @param parentName name of the group that contains {@code repeatedType}
+   */
+  private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
+    return !repeatedType.isPrimitive()
+        && repeatedType.getLogicalTypeAnnotation() == null
+        && repeatedType.asGroupType().getFieldCount() == 1
+        && !repeatedType.getName().equals("array")
+        && !repeatedType.getName().equals(parentName + "_tuple");
+  }
+
   /**
    * Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET physical type, matching
    * Parquet's canonical 3-level map encoding.
@@ -823,10 +881,7 @@ public class ParquetSplitReaderUtil {
 
     // Three-level list: skip the synthetic `element` / `list` wrapper when present.
     if (columnIO instanceof GroupColumnIO
-        && columnIO.getType().getLogicalTypeAnnotation() == null
-        && ((GroupColumnIO) columnIO).getChildrenCount() == 1
-        && !columnIO.getName().equals("array")
-        && !columnIO.getName().equals(columnIO.getParent().getName() + "_tuple")) {
+        && isThreeLevelListRepeatedType(columnIO.getType(), columnIO.getParent().getName())) {
       return ((GroupColumnIO) columnIO).getChild(0);
     }
     return columnIO;
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index 3f2f8976b69b..52745f747d25 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -69,7 +69,9 @@ public class NestedPositionUtil {
       }
     }
     if (nullValuesCount == 0) {
-      return new RowPosition(null, fieldDefinitionLevels.length);
+      // Diverges from Flink, which returns fieldDefinitionLevels.length here: that also counts the
+      // skipped values of a last field that is an array, one per element instead of one per row.
+      return new RowPosition(null, nullRowFlags.size());
     }
     return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
   }
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
index 60575f148cc4..bbfe5fcff31f 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -49,6 +49,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createVectorFromConstant;
+
 /**
  * ColumnReader used to read a {@code Group} type in Parquet ({@code Map}, {@code Array}, {@code
  * Row}). Resolves nested structures using Dremel striping/assembly; see <a
@@ -139,6 +141,17 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> {
             levelDelegation.getDefinitionLevel(),
             levelDelegation.getRepetitionLevel());
 
+    // Inside an array the row has one position per element, which can exceed the batch size the
+    // placeholders of absent fields were sized with, so grow them to the position count.
+    int positionsCount = rowPosition.getPositionsCount();
+    RowType rowType = (RowType) field.getType();
+    for (int i = 0; i < children.size(); i++) {
+      if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
+        finalChildrenVectors[i] = (WritableColumnVector) createVectorFromConstant(
+            rowType.getTypeAt(i), null, positionsCount);
+      }
+    }
+
     // If row was inside the structure, then we need to renew the vector to reset the
     // capacity.
     if (inside) {
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index 1826419db5d4..b9f7527a91a8 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -250,6 +250,7 @@ public class ParquetColumnarRowSplitReader implements Closeable {
                 batchSize,
                 requestedTypes[i],
                 types.get(i),
+                requestedSchema.getName(),
                 descriptors);
       } catch (IllegalArgumentException e) {
         String fieldName = requestedSchema.getFieldName(i);
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
new file mode 100644
index 000000000000..3763003e554d
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
@@ -0,0 +1,415 @@
+/*
+ * 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.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+  private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+  private static final int DEFAULT_BATCH_SIZE = 2048;
+
+  @TempDir
+  File tempDir;
+
+  /**
+   * Physical encodings of a list of single- or multi-field element groups.
+   */
+  enum ListEncoding {
+    THREE_LEVEL {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+            + elementFields + " } } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("list").addGroup("element");
+      }
+    },
+    TWO_LEVEL_ARRAY {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("array");
+      }
+    },
+    TWO_LEVEL_TUPLE {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+            + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup(name + "_tuple");
+      }
+    };
+
+    abstract String schema(String name, String elementFields);
+
+    abstract Group addElement(Group list, String name);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+    MessageType schema = itemsSchema(encoding);
+    LogicalType itemsType = new ArrayType(
+        rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+    List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+        row -> formatItems(row, false));
+
+    assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), DEFAULT_BATCH_SIZE);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount(ListEncoding encoding)
+      throws IOException {
+    // The first batch holds two rows with three items and three tags, so the placeholders of the
+    // absent fields must grow beyond the batch size.
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), 2);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @Test
+  void testReadBareRepeatedGroupNamedAfterItsParentAsArrayOfRow() throws IOException {
+    // A repeated group with one field named <parent>_tuple is the element itself, not a wrapper.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { repeated group info_tuple { optional int32 x; } } }");
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.addGroup("info_tuple").append("x", 1);
+      info.addGroup("info_tuple").append("x", 2);
+    };
+    LogicalType infoType = rowType(new String[] {"info_tuple"}, new ArrayType(rowType(new String[] {"x"}, new IntType())));
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      ArrayData tuples = row.getRow(0, 1).getArray(0);
+      List<Integer> xs = new ArrayList<>();
+      for (int i = 0; i < tuples.size(); i++) {
+        xs.add(tuples.getRow(i, 1).getInt(0));
+      }
+      return xs.toString();
+    });
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadRowWithFieldMissingFromFileAfterLongArray() throws IOException {
+    // The row's last field in the file is an array with more elements than the batch size, which
+    // must not count as more rows when sizing the placeholder of the absent field.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { optional binary name (UTF8); "
+            + "optional group nums (LIST) { repeated group list { optional int32 element; } } } }");
+    int[] recordIndex = {0};
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.append("name", "r" + recordIndex[0]++);
+      Group nums = info.addGroup("nums");
+      for (int i = 0; i < 100; i++) {
+        nums.addGroup("list").append("element", i);
+      }
+    };
+    LogicalType infoType = rowType(new String[] {"name", "nums", "added"}, STRING, new ArrayType(new IntType()), STRING);
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      RowData info = row.getRow(0, 3);
+      return info.getString(0) + ":" + info.getArray(1).size() + ":" + (info.isNullAt(2) ? "null" : info.getString(2));
+    }, 2);
+
+    assertEquals(Arrays.asList("r0:100:null", "r1:100:null", "r2:100:null"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadArrayOfRowWithIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group items (LIST) { repeated group list { optional group element { "
+            + "optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); } } } }");
+    Consumer<Group> writer = record -> {
+      Group items = record.addGroup("items");
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, 12345);
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, -1);
+    };
+    LogicalType itemsType = new ArrayType(rowType(new String[] {"amount"}, new DecimalType(precision, 2)));
+
+    List<String> values = writeAndRead(schema, writer, "items", itemsType, row -> {
+      ArrayData items = row.getArray(0);
+      List<String> amounts = new ArrayList<>();
+      for (int i = 0; i < items.size(); i++) {
+        amounts.add(items.getRow(i, 1).getDecimal(0, precision, 2).toString());
+      }
+      return amounts.toString();
+    });
+
+    assertEquals(Arrays.asList("[123.45, -0.01]", "[123.45, -0.01]", "[123.45, -0.01]"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); }");
+
+    List<String> values = writeAndRead(schema, record -> appendUnscaled(record, physicalType, 12345), "amount",
+        new DecimalType(precision, 2), row -> row.getDecimal(0, precision, 2).toString());
+
+    assertEquals(Arrays.asList("123.45", "123.45", "123.45"), values);
+  }
+
+  @Test
+  void testReadThreeLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.addGroup("list").append("element", 1);
+      list.addGroup("list").append("element", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadTwoLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated int32 array; } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.append("array", 1);
+      list.append("array", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  /**
+   * Appends the unscaled value of a decimal to the {@code amount} field, as the given physical type.
+   */
+  private static void appendUnscaled(Group group, String physicalType, int unscaled) {
+    if (physicalType.equals("int32")) {
+      group.append("amount", unscaled);
+    } else {
+      group.append("amount", (long) unscaled);
+    }
+  }
+
+  private static MessageType itemsSchema(ListEncoding encoding) {
+    String tags = encoding.schema("tags", "optional binary value (UTF8);");
+    String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+    return MessageTypeParser.parseMessageType("message m { " + items + " }");
+  }
+
+  /**
+   * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+   */
+  private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+    int[] recordIndex = {0};
+    return record -> {
+      int index = recordIndex[0]++;
+      if (index == 0) {
+        Group items = record.addGroup("items");
+        addItem(encoding, items, "a", "x", "y");
+        addItem(encoding, items, "b", "z");
+      } else if (index == 1) {
+        addItem(encoding, record.addGroup("items"), "c");
+      }
+    };
+  }
+
+  private static void addItem(ListEncoding encoding, Group items, String name, String... tags) {
+    Group item = encoding.addElement(items, "items");
+    item.append("name", name);
+    Group tagList = item.addGroup("tags");
+    for (String tag : tags) {
+      encoding.addElement(tagList, "tags").append("value", tag);
+    }
+  }
+
+  /**
+   * Returns the items type with fields that the written files don't have: {@code added_scores} on
+   * the item and {@code added_tag_field} on the nested tag.
+   */
+  private static LogicalType evolvedItemsType() {
+    RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+    return new ArrayType(
+        rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+            new ArrayType(tagType)));
+  }
+
+  private static RowType rowType(String[] names, LogicalType... types) {
+    return RowType.of(types, names);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter) throws IOException {
+    return writeAndRead(schema, recordWriter, columnName, columnType, formatter, DEFAULT_BATCH_SIZE);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter,
+      int batchSize) throws IOException {
+    Path path = new Path(new File(tempDir, "data.parquet").getAbsolutePath());
+    Configuration conf = new Configuration();
+    SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(path).withType(schema).withConf(conf).build()) {
+      for (int i = 0; i < 3; i++) {
+        Group record = factory.newGroup();
+        recordWriter.accept(record);
+        writer.write(record);
+      }
+    }
+
+    List<String> values = new ArrayList<>();
+    try (ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+        true,
+        true,
+        conf,
+        new LogicalType[] {columnType},
+        new String[] {columnName},
+        VectorizedColumnBatch::new,
+        batchSize,
+        path,
+        0,
+        new File(tempDir, "data.parquet").length(),
+        null,
+        null)) {
+      while (!reader.reachedEnd()) {
+        values.add(formatter.apply(reader.nextRecord()));
+      }
+    }
+    return values;
+  }
+
+  /**
+   * Formats {@code array<row<name, [added_scores,] array<row<value[, added_tag_field]>>>>}. With
+   * {@code evolved}, the added fields are printed in parentheses after the existing ones.
+   */
+  private static String formatItems(RowData row, boolean evolved) {
+    if (row.isNullAt(0)) {
+      return "null";
+    }
+    ArrayData items = row.getArray(0);
+    int tagsPos = evolved ? 2 : 1;
+    List<String> formatted = new ArrayList<>();
+    for (int i = 0; i < items.size(); i++) {
+      RowData item = items.getRow(i, tagsPos + 1);
+      String name = item.getString(0).toString();
+      if (evolved) {
+        name += "(" + (item.isNullAt(1) ? "null" : formatInts(item.getArray(1))) + ")";
+      }
+      ArrayData tags = item.getArray(tagsPos);
+      List<String> tagValues = new ArrayList<>();
+      for (int j = 0; j < tags.size(); j++) {
+        RowData tag = tags.getRow(j, evolved ? 2 : 1);
+        String value = tag.getString(0).toString();
+        if (evolved) {
+          value += "(" + (tag.isNullAt(1) ? "null" : tag.getString(1).toString()) + ")";
+        }
+        tagValues.add(value);
+      }
+      formatted.add(name + ":" + tagValues);
+    }
+    return formatted.toString();
+  }
+
+  private static String formatInts(ArrayData array) {
+    List<Integer> ints = new ArrayList<>();
+    for (int i = 0; i < array.size(); i++) {
+      ints.add(array.getInt(i));
+    }
+    return ints.toString();
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
new file mode 100644
index 000000000000..0bfb1a8aa668
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.hudi.table.format.cow.utils;
+
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link NestedPositionUtil}.
+ */
+class TestNestedPositionUtil {
+
+  /** An optional top-level row whose last field is an optional list of optional ints. */
+  private static final ParquetGroupField ROW =
+      new ParquetGroupField(RowType.of(new IntType()), 0, 1, false, Collections.emptyList());
+
+  @Test
+  void testRowOffsetsCountOneValuePerRowWhenTheLastFieldIsAnArray() {
+    // two rows of three elements each, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 4, 4, 4, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 1, 1, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertNull(position.getIsNull());
+  }
+
+  @Test
+  void testRowOffsetsWithNullRows() {
+    // a row of three elements, a null row, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 0, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertArrayEquals(new boolean[] {false, true, false}, position.getIsNull());
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index 33da58dd3ad5..e702461b5203 100644
--- a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++ b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -19,6 +19,7 @@
 package org.apache.hudi.table.format.cow;
 
 import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
 import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
@@ -483,19 +484,24 @@ public class ParquetSplitReaderUtil {
    * matches {@code fieldType}; for ROW types missing physical fields are slotted with null-filled
    * vectors (sourced from {@link #createVectorFromConstant}) so that the Dremel assembler in
    * {@link NestedColumnReader} can pass them through unchanged.
+   *
+   * @param parentName name of the group that contains {@code physicalType}, which decides whether a
+   *     repeated group named {@code <parentName>_tuple} is a list element
    */
   public static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> descriptors) {
-    return createWritableColumnVector(batchSize, fieldType, physicalType, descriptors, 0);
+    return createWritableColumnVector(batchSize, fieldType, physicalType, parentName, descriptors, 0);
   }
 
   private static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> columns,
       int depth) {
     List<ColumnDescriptor> descriptors = filterDescriptors(depth, physicalType, columns);
@@ -555,26 +561,35 @@ public class ParquetSplitReaderUtil {
         return new HeapTimestampVector(batchSize);
       case DECIMAL:
         checkArgument(
-            (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
-                    || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
-                && primitiveType.getOriginalType() == OriginalType.DECIMAL,
+            primitiveType.getOriginalType() == OriginalType.DECIMAL,
             getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
-        return new HeapDecimalVector(batchSize);
+        switch (typeName) {
+          case INT32:
+            return new HeapIntVector(batchSize);
+          case INT64:
+            return new HeapLongVector(batchSize);
+          case BINARY:
+          case FIXED_LEN_BYTE_ARRAY:
+            return new HeapDecimalVector(batchSize);
+          default:
+            throw new IllegalArgumentException(getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+        }
       case ARRAY:
         ArrayType arrayType = (ArrayType) fieldType;
+        Triple<Type, String, Integer> element = unwrapListElementType(physicalType, parentName, depth);
         return new HeapArrayVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+            createWritableColumnVector(batchSize, arrayType.getElementType(), element.getLeft(),
+                element.getMiddle(), descriptors, element.getRight()));
       case MAP: {
         MapType mapType = (MapType) fieldType;
         GroupType repeatedType = unwrapMapRepeatedType(physicalType);
         return new HeapMapColumnVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, mapType.getKeyType(), repeatedType.getType(0), descriptors, depth + 2),
-            createWritableColumnVector(
-                batchSize, mapType.getValueType(), repeatedType.getType(1), descriptors, depth + 2));
+            createWritableColumnVector(batchSize, mapType.getKeyType(), repeatedType.getType(0),
+                repeatedType.getName(), descriptors, depth + 2),
+            createWritableColumnVector(batchSize, mapType.getValueType(), repeatedType.getType(1),
+                repeatedType.getName(), descriptors, depth + 2));
       }
       case MULTISET: {
         MultisetType multisetType = (MultisetType) fieldType;
@@ -585,12 +600,14 @@ public class ParquetSplitReaderUtil {
                 batchSize,
                 multisetType.getElementType(),
                 repeatedType.getType(0),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2),
             createWritableColumnVector(
                 batchSize,
                 new IntType(false),
                 repeatedType.getType(1),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2));
       }
@@ -612,6 +629,7 @@ public class ParquetSplitReaderUtil {
                     batchSize,
                     rowType.getTypeAt(i),
                     groupType.getType(fieldIndex),
+                    groupType.getName(),
                     descriptors,
                     depth + 1);
           }
@@ -622,6 +640,46 @@ public class ParquetSplitReaderUtil {
     }
   }
 
+  /**
+   * Returns the element type of a LIST physical type, the name of the group that contains it and its
+   * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+   * so that the vectors line up with the {@code ParquetField} tree.
+   *
+   * @param parentName name of the group that contains {@code listType}
+   */
+  private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+    Type elementType = listType;
+    String elementParentName = parentName;
+    int elementDepth = depth;
+    while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    return Triple.of(elementType, elementParentName, elementDepth);
+  }
+
+  /**
+   * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+   * rather than the element itself. Shared by {@link #unwrapListElementType} and
+   * {@link #getArrayElementColumn}, so that the vectors and the {@code ParquetField} tree pick the
+   * same element.
+   *
+   * @param parentName name of the group that contains {@code repeatedType}
+   */
+  private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
+    return !repeatedType.isPrimitive()
+        && repeatedType.getLogicalTypeAnnotation() == null
+        && repeatedType.asGroupType().getFieldCount() == 1
+        && !repeatedType.getName().equals("array")
+        && !repeatedType.getName().equals(parentName + "_tuple");
+  }
+
   /**
    * Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET physical type, matching
    * Parquet's canonical 3-level map encoding.
@@ -823,10 +881,7 @@ public class ParquetSplitReaderUtil {
 
     // Three-level list: skip the synthetic `element` / `list` wrapper when present.
     if (columnIO instanceof GroupColumnIO
-        && columnIO.getType().getLogicalTypeAnnotation() == null
-        && ((GroupColumnIO) columnIO).getChildrenCount() == 1
-        && !columnIO.getName().equals("array")
-        && !columnIO.getName().equals(columnIO.getParent().getName() + "_tuple")) {
+        && isThreeLevelListRepeatedType(columnIO.getType(), columnIO.getParent().getName())) {
       return ((GroupColumnIO) columnIO).getChild(0);
     }
     return columnIO;
diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index 3f2f8976b69b..52745f747d25 100644
--- a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
+++ b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -69,7 +69,9 @@ public class NestedPositionUtil {
       }
     }
     if (nullValuesCount == 0) {
-      return new RowPosition(null, fieldDefinitionLevels.length);
+      // Diverges from Flink, which returns fieldDefinitionLevels.length here: that also counts the
+      // skipped values of a last field that is an array, one per element instead of one per row.
+      return new RowPosition(null, nullRowFlags.size());
     }
     return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
   }
diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
index 60575f148cc4..bbfe5fcff31f 100644
--- a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
+++ b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -49,6 +49,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createVectorFromConstant;
+
 /**
  * ColumnReader used to read a {@code Group} type in Parquet ({@code Map}, {@code Array}, {@code
  * Row}). Resolves nested structures using Dremel striping/assembly; see <a
@@ -139,6 +141,17 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> {
             levelDelegation.getDefinitionLevel(),
             levelDelegation.getRepetitionLevel());
 
+    // Inside an array the row has one position per element, which can exceed the batch size the
+    // placeholders of absent fields were sized with, so grow them to the position count.
+    int positionsCount = rowPosition.getPositionsCount();
+    RowType rowType = (RowType) field.getType();
+    for (int i = 0; i < children.size(); i++) {
+      if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
+        finalChildrenVectors[i] = (WritableColumnVector) createVectorFromConstant(
+            rowType.getTypeAt(i), null, positionsCount);
+      }
+    }
+
     // If row was inside the structure, then we need to renew the vector to reset the
     // capacity.
     if (inside) {
diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index 1826419db5d4..b9f7527a91a8 100644
--- a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -250,6 +250,7 @@ public class ParquetColumnarRowSplitReader implements Closeable {
                 batchSize,
                 requestedTypes[i],
                 types.get(i),
+                requestedSchema.getName(),
                 descriptors);
       } catch (IllegalArgumentException e) {
         String fieldName = requestedSchema.getFieldName(i);
diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
new file mode 100644
index 000000000000..3763003e554d
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
@@ -0,0 +1,415 @@
+/*
+ * 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.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+  private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+  private static final int DEFAULT_BATCH_SIZE = 2048;
+
+  @TempDir
+  File tempDir;
+
+  /**
+   * Physical encodings of a list of single- or multi-field element groups.
+   */
+  enum ListEncoding {
+    THREE_LEVEL {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+            + elementFields + " } } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("list").addGroup("element");
+      }
+    },
+    TWO_LEVEL_ARRAY {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("array");
+      }
+    },
+    TWO_LEVEL_TUPLE {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+            + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup(name + "_tuple");
+      }
+    };
+
+    abstract String schema(String name, String elementFields);
+
+    abstract Group addElement(Group list, String name);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+    MessageType schema = itemsSchema(encoding);
+    LogicalType itemsType = new ArrayType(
+        rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+    List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+        row -> formatItems(row, false));
+
+    assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), DEFAULT_BATCH_SIZE);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount(ListEncoding encoding)
+      throws IOException {
+    // The first batch holds two rows with three items and three tags, so the placeholders of the
+    // absent fields must grow beyond the batch size.
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), 2);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @Test
+  void testReadBareRepeatedGroupNamedAfterItsParentAsArrayOfRow() throws IOException {
+    // A repeated group with one field named <parent>_tuple is the element itself, not a wrapper.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { repeated group info_tuple { optional int32 x; } } }");
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.addGroup("info_tuple").append("x", 1);
+      info.addGroup("info_tuple").append("x", 2);
+    };
+    LogicalType infoType = rowType(new String[] {"info_tuple"}, new ArrayType(rowType(new String[] {"x"}, new IntType())));
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      ArrayData tuples = row.getRow(0, 1).getArray(0);
+      List<Integer> xs = new ArrayList<>();
+      for (int i = 0; i < tuples.size(); i++) {
+        xs.add(tuples.getRow(i, 1).getInt(0));
+      }
+      return xs.toString();
+    });
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadRowWithFieldMissingFromFileAfterLongArray() throws IOException {
+    // The row's last field in the file is an array with more elements than the batch size, which
+    // must not count as more rows when sizing the placeholder of the absent field.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { optional binary name (UTF8); "
+            + "optional group nums (LIST) { repeated group list { optional int32 element; } } } }");
+    int[] recordIndex = {0};
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.append("name", "r" + recordIndex[0]++);
+      Group nums = info.addGroup("nums");
+      for (int i = 0; i < 100; i++) {
+        nums.addGroup("list").append("element", i);
+      }
+    };
+    LogicalType infoType = rowType(new String[] {"name", "nums", "added"}, STRING, new ArrayType(new IntType()), STRING);
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      RowData info = row.getRow(0, 3);
+      return info.getString(0) + ":" + info.getArray(1).size() + ":" + (info.isNullAt(2) ? "null" : info.getString(2));
+    }, 2);
+
+    assertEquals(Arrays.asList("r0:100:null", "r1:100:null", "r2:100:null"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadArrayOfRowWithIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group items (LIST) { repeated group list { optional group element { "
+            + "optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); } } } }");
+    Consumer<Group> writer = record -> {
+      Group items = record.addGroup("items");
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, 12345);
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, -1);
+    };
+    LogicalType itemsType = new ArrayType(rowType(new String[] {"amount"}, new DecimalType(precision, 2)));
+
+    List<String> values = writeAndRead(schema, writer, "items", itemsType, row -> {
+      ArrayData items = row.getArray(0);
+      List<String> amounts = new ArrayList<>();
+      for (int i = 0; i < items.size(); i++) {
+        amounts.add(items.getRow(i, 1).getDecimal(0, precision, 2).toString());
+      }
+      return amounts.toString();
+    });
+
+    assertEquals(Arrays.asList("[123.45, -0.01]", "[123.45, -0.01]", "[123.45, -0.01]"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); }");
+
+    List<String> values = writeAndRead(schema, record -> appendUnscaled(record, physicalType, 12345), "amount",
+        new DecimalType(precision, 2), row -> row.getDecimal(0, precision, 2).toString());
+
+    assertEquals(Arrays.asList("123.45", "123.45", "123.45"), values);
+  }
+
+  @Test
+  void testReadThreeLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.addGroup("list").append("element", 1);
+      list.addGroup("list").append("element", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadTwoLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated int32 array; } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.append("array", 1);
+      list.append("array", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  /**
+   * Appends the unscaled value of a decimal to the {@code amount} field, as the given physical type.
+   */
+  private static void appendUnscaled(Group group, String physicalType, int unscaled) {
+    if (physicalType.equals("int32")) {
+      group.append("amount", unscaled);
+    } else {
+      group.append("amount", (long) unscaled);
+    }
+  }
+
+  private static MessageType itemsSchema(ListEncoding encoding) {
+    String tags = encoding.schema("tags", "optional binary value (UTF8);");
+    String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+    return MessageTypeParser.parseMessageType("message m { " + items + " }");
+  }
+
+  /**
+   * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+   */
+  private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+    int[] recordIndex = {0};
+    return record -> {
+      int index = recordIndex[0]++;
+      if (index == 0) {
+        Group items = record.addGroup("items");
+        addItem(encoding, items, "a", "x", "y");
+        addItem(encoding, items, "b", "z");
+      } else if (index == 1) {
+        addItem(encoding, record.addGroup("items"), "c");
+      }
+    };
+  }
+
+  private static void addItem(ListEncoding encoding, Group items, String name, String... tags) {
+    Group item = encoding.addElement(items, "items");
+    item.append("name", name);
+    Group tagList = item.addGroup("tags");
+    for (String tag : tags) {
+      encoding.addElement(tagList, "tags").append("value", tag);
+    }
+  }
+
+  /**
+   * Returns the items type with fields that the written files don't have: {@code added_scores} on
+   * the item and {@code added_tag_field} on the nested tag.
+   */
+  private static LogicalType evolvedItemsType() {
+    RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+    return new ArrayType(
+        rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+            new ArrayType(tagType)));
+  }
+
+  private static RowType rowType(String[] names, LogicalType... types) {
+    return RowType.of(types, names);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter) throws IOException {
+    return writeAndRead(schema, recordWriter, columnName, columnType, formatter, DEFAULT_BATCH_SIZE);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter,
+      int batchSize) throws IOException {
+    Path path = new Path(new File(tempDir, "data.parquet").getAbsolutePath());
+    Configuration conf = new Configuration();
+    SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(path).withType(schema).withConf(conf).build()) {
+      for (int i = 0; i < 3; i++) {
+        Group record = factory.newGroup();
+        recordWriter.accept(record);
+        writer.write(record);
+      }
+    }
+
+    List<String> values = new ArrayList<>();
+    try (ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+        true,
+        true,
+        conf,
+        new LogicalType[] {columnType},
+        new String[] {columnName},
+        VectorizedColumnBatch::new,
+        batchSize,
+        path,
+        0,
+        new File(tempDir, "data.parquet").length(),
+        null,
+        null)) {
+      while (!reader.reachedEnd()) {
+        values.add(formatter.apply(reader.nextRecord()));
+      }
+    }
+    return values;
+  }
+
+  /**
+   * Formats {@code array<row<name, [added_scores,] array<row<value[, added_tag_field]>>>>}. With
+   * {@code evolved}, the added fields are printed in parentheses after the existing ones.
+   */
+  private static String formatItems(RowData row, boolean evolved) {
+    if (row.isNullAt(0)) {
+      return "null";
+    }
+    ArrayData items = row.getArray(0);
+    int tagsPos = evolved ? 2 : 1;
+    List<String> formatted = new ArrayList<>();
+    for (int i = 0; i < items.size(); i++) {
+      RowData item = items.getRow(i, tagsPos + 1);
+      String name = item.getString(0).toString();
+      if (evolved) {
+        name += "(" + (item.isNullAt(1) ? "null" : formatInts(item.getArray(1))) + ")";
+      }
+      ArrayData tags = item.getArray(tagsPos);
+      List<String> tagValues = new ArrayList<>();
+      for (int j = 0; j < tags.size(); j++) {
+        RowData tag = tags.getRow(j, evolved ? 2 : 1);
+        String value = tag.getString(0).toString();
+        if (evolved) {
+          value += "(" + (tag.isNullAt(1) ? "null" : tag.getString(1).toString()) + ")";
+        }
+        tagValues.add(value);
+      }
+      formatted.add(name + ":" + tagValues);
+    }
+    return formatted.toString();
+  }
+
+  private static String formatInts(ArrayData array) {
+    List<Integer> ints = new ArrayList<>();
+    for (int i = 0; i < array.size(); i++) {
+      ints.add(array.getInt(i));
+    }
+    return ints.toString();
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java b/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
new file mode 100644
index 000000000000..0bfb1a8aa668
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.hudi.table.format.cow.utils;
+
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link NestedPositionUtil}.
+ */
+class TestNestedPositionUtil {
+
+  /** An optional top-level row whose last field is an optional list of optional ints. */
+  private static final ParquetGroupField ROW =
+      new ParquetGroupField(RowType.of(new IntType()), 0, 1, false, Collections.emptyList());
+
+  @Test
+  void testRowOffsetsCountOneValuePerRowWhenTheLastFieldIsAnArray() {
+    // two rows of three elements each, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 4, 4, 4, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 1, 1, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertNull(position.getIsNull());
+  }
+
+  @Test
+  void testRowOffsetsWithNullRows() {
+    // a row of three elements, a null row, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 0, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertArrayEquals(new boolean[] {false, true, false}, position.getIsNull());
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index 33da58dd3ad5..e702461b5203 100644
--- a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++ b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -19,6 +19,7 @@
 package org.apache.hudi.table.format.cow;
 
 import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
 import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
@@ -483,19 +484,24 @@ public class ParquetSplitReaderUtil {
    * matches {@code fieldType}; for ROW types missing physical fields are slotted with null-filled
    * vectors (sourced from {@link #createVectorFromConstant}) so that the Dremel assembler in
    * {@link NestedColumnReader} can pass them through unchanged.
+   *
+   * @param parentName name of the group that contains {@code physicalType}, which decides whether a
+   *     repeated group named {@code <parentName>_tuple} is a list element
    */
   public static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> descriptors) {
-    return createWritableColumnVector(batchSize, fieldType, physicalType, descriptors, 0);
+    return createWritableColumnVector(batchSize, fieldType, physicalType, parentName, descriptors, 0);
   }
 
   private static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> columns,
       int depth) {
     List<ColumnDescriptor> descriptors = filterDescriptors(depth, physicalType, columns);
@@ -555,26 +561,35 @@ public class ParquetSplitReaderUtil {
         return new HeapTimestampVector(batchSize);
       case DECIMAL:
         checkArgument(
-            (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
-                    || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
-                && primitiveType.getOriginalType() == OriginalType.DECIMAL,
+            primitiveType.getOriginalType() == OriginalType.DECIMAL,
             getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
-        return new HeapDecimalVector(batchSize);
+        switch (typeName) {
+          case INT32:
+            return new HeapIntVector(batchSize);
+          case INT64:
+            return new HeapLongVector(batchSize);
+          case BINARY:
+          case FIXED_LEN_BYTE_ARRAY:
+            return new HeapDecimalVector(batchSize);
+          default:
+            throw new IllegalArgumentException(getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+        }
       case ARRAY:
         ArrayType arrayType = (ArrayType) fieldType;
+        Triple<Type, String, Integer> element = unwrapListElementType(physicalType, parentName, depth);
         return new HeapArrayVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+            createWritableColumnVector(batchSize, arrayType.getElementType(), element.getLeft(),
+                element.getMiddle(), descriptors, element.getRight()));
       case MAP: {
         MapType mapType = (MapType) fieldType;
         GroupType repeatedType = unwrapMapRepeatedType(physicalType);
         return new HeapMapColumnVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, mapType.getKeyType(), repeatedType.getType(0), descriptors, depth + 2),
-            createWritableColumnVector(
-                batchSize, mapType.getValueType(), repeatedType.getType(1), descriptors, depth + 2));
+            createWritableColumnVector(batchSize, mapType.getKeyType(), repeatedType.getType(0),
+                repeatedType.getName(), descriptors, depth + 2),
+            createWritableColumnVector(batchSize, mapType.getValueType(), repeatedType.getType(1),
+                repeatedType.getName(), descriptors, depth + 2));
       }
       case MULTISET: {
         MultisetType multisetType = (MultisetType) fieldType;
@@ -585,12 +600,14 @@ public class ParquetSplitReaderUtil {
                 batchSize,
                 multisetType.getElementType(),
                 repeatedType.getType(0),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2),
             createWritableColumnVector(
                 batchSize,
                 new IntType(false),
                 repeatedType.getType(1),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2));
       }
@@ -612,6 +629,7 @@ public class ParquetSplitReaderUtil {
                     batchSize,
                     rowType.getTypeAt(i),
                     groupType.getType(fieldIndex),
+                    groupType.getName(),
                     descriptors,
                     depth + 1);
           }
@@ -622,6 +640,46 @@ public class ParquetSplitReaderUtil {
     }
   }
 
+  /**
+   * Returns the element type of a LIST physical type, the name of the group that contains it and its
+   * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+   * so that the vectors line up with the {@code ParquetField} tree.
+   *
+   * @param parentName name of the group that contains {@code listType}
+   */
+  private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+    Type elementType = listType;
+    String elementParentName = parentName;
+    int elementDepth = depth;
+    while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    return Triple.of(elementType, elementParentName, elementDepth);
+  }
+
+  /**
+   * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+   * rather than the element itself. Shared by {@link #unwrapListElementType} and
+   * {@link #getArrayElementColumn}, so that the vectors and the {@code ParquetField} tree pick the
+   * same element.
+   *
+   * @param parentName name of the group that contains {@code repeatedType}
+   */
+  private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
+    return !repeatedType.isPrimitive()
+        && repeatedType.getLogicalTypeAnnotation() == null
+        && repeatedType.asGroupType().getFieldCount() == 1
+        && !repeatedType.getName().equals("array")
+        && !repeatedType.getName().equals(parentName + "_tuple");
+  }
+
   /**
    * Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET physical type, matching
    * Parquet's canonical 3-level map encoding.
@@ -823,10 +881,7 @@ public class ParquetSplitReaderUtil {
 
     // Three-level list: skip the synthetic `element` / `list` wrapper when present.
     if (columnIO instanceof GroupColumnIO
-        && columnIO.getType().getLogicalTypeAnnotation() == null
-        && ((GroupColumnIO) columnIO).getChildrenCount() == 1
-        && !columnIO.getName().equals("array")
-        && !columnIO.getName().equals(columnIO.getParent().getName() + "_tuple")) {
+        && isThreeLevelListRepeatedType(columnIO.getType(), columnIO.getParent().getName())) {
       return ((GroupColumnIO) columnIO).getChild(0);
     }
     return columnIO;
diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index 3f2f8976b69b..52745f747d25 100644
--- a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
+++ b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -69,7 +69,9 @@ public class NestedPositionUtil {
       }
     }
     if (nullValuesCount == 0) {
-      return new RowPosition(null, fieldDefinitionLevels.length);
+      // Diverges from Flink, which returns fieldDefinitionLevels.length here: that also counts the
+      // skipped values of a last field that is an array, one per element instead of one per row.
+      return new RowPosition(null, nullRowFlags.size());
     }
     return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
   }
diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
index 60575f148cc4..bbfe5fcff31f 100644
--- a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
+++ b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -49,6 +49,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createVectorFromConstant;
+
 /**
  * ColumnReader used to read a {@code Group} type in Parquet ({@code Map}, {@code Array}, {@code
  * Row}). Resolves nested structures using Dremel striping/assembly; see <a
@@ -139,6 +141,17 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> {
             levelDelegation.getDefinitionLevel(),
             levelDelegation.getRepetitionLevel());
 
+    // Inside an array the row has one position per element, which can exceed the batch size the
+    // placeholders of absent fields were sized with, so grow them to the position count.
+    int positionsCount = rowPosition.getPositionsCount();
+    RowType rowType = (RowType) field.getType();
+    for (int i = 0; i < children.size(); i++) {
+      if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
+        finalChildrenVectors[i] = (WritableColumnVector) createVectorFromConstant(
+            rowType.getTypeAt(i), null, positionsCount);
+      }
+    }
+
     // If row was inside the structure, then we need to renew the vector to reset the
     // capacity.
     if (inside) {
diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index 1826419db5d4..b9f7527a91a8 100644
--- a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -250,6 +250,7 @@ public class ParquetColumnarRowSplitReader implements Closeable {
                 batchSize,
                 requestedTypes[i],
                 types.get(i),
+                requestedSchema.getName(),
                 descriptors);
       } catch (IllegalArgumentException e) {
         String fieldName = requestedSchema.getFieldName(i);
diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
new file mode 100644
index 000000000000..3763003e554d
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
@@ -0,0 +1,415 @@
+/*
+ * 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.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+  private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+  private static final int DEFAULT_BATCH_SIZE = 2048;
+
+  @TempDir
+  File tempDir;
+
+  /**
+   * Physical encodings of a list of single- or multi-field element groups.
+   */
+  enum ListEncoding {
+    THREE_LEVEL {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+            + elementFields + " } } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("list").addGroup("element");
+      }
+    },
+    TWO_LEVEL_ARRAY {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("array");
+      }
+    },
+    TWO_LEVEL_TUPLE {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+            + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup(name + "_tuple");
+      }
+    };
+
+    abstract String schema(String name, String elementFields);
+
+    abstract Group addElement(Group list, String name);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+    MessageType schema = itemsSchema(encoding);
+    LogicalType itemsType = new ArrayType(
+        rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+    List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+        row -> formatItems(row, false));
+
+    assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), DEFAULT_BATCH_SIZE);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount(ListEncoding encoding)
+      throws IOException {
+    // The first batch holds two rows with three items and three tags, so the placeholders of the
+    // absent fields must grow beyond the batch size.
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), 2);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @Test
+  void testReadBareRepeatedGroupNamedAfterItsParentAsArrayOfRow() throws IOException {
+    // A repeated group with one field named <parent>_tuple is the element itself, not a wrapper.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { repeated group info_tuple { optional int32 x; } } }");
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.addGroup("info_tuple").append("x", 1);
+      info.addGroup("info_tuple").append("x", 2);
+    };
+    LogicalType infoType = rowType(new String[] {"info_tuple"}, new ArrayType(rowType(new String[] {"x"}, new IntType())));
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      ArrayData tuples = row.getRow(0, 1).getArray(0);
+      List<Integer> xs = new ArrayList<>();
+      for (int i = 0; i < tuples.size(); i++) {
+        xs.add(tuples.getRow(i, 1).getInt(0));
+      }
+      return xs.toString();
+    });
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadRowWithFieldMissingFromFileAfterLongArray() throws IOException {
+    // The row's last field in the file is an array with more elements than the batch size, which
+    // must not count as more rows when sizing the placeholder of the absent field.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { optional binary name (UTF8); "
+            + "optional group nums (LIST) { repeated group list { optional int32 element; } } } }");
+    int[] recordIndex = {0};
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.append("name", "r" + recordIndex[0]++);
+      Group nums = info.addGroup("nums");
+      for (int i = 0; i < 100; i++) {
+        nums.addGroup("list").append("element", i);
+      }
+    };
+    LogicalType infoType = rowType(new String[] {"name", "nums", "added"}, STRING, new ArrayType(new IntType()), STRING);
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      RowData info = row.getRow(0, 3);
+      return info.getString(0) + ":" + info.getArray(1).size() + ":" + (info.isNullAt(2) ? "null" : info.getString(2));
+    }, 2);
+
+    assertEquals(Arrays.asList("r0:100:null", "r1:100:null", "r2:100:null"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadArrayOfRowWithIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group items (LIST) { repeated group list { optional group element { "
+            + "optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); } } } }");
+    Consumer<Group> writer = record -> {
+      Group items = record.addGroup("items");
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, 12345);
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, -1);
+    };
+    LogicalType itemsType = new ArrayType(rowType(new String[] {"amount"}, new DecimalType(precision, 2)));
+
+    List<String> values = writeAndRead(schema, writer, "items", itemsType, row -> {
+      ArrayData items = row.getArray(0);
+      List<String> amounts = new ArrayList<>();
+      for (int i = 0; i < items.size(); i++) {
+        amounts.add(items.getRow(i, 1).getDecimal(0, precision, 2).toString());
+      }
+      return amounts.toString();
+    });
+
+    assertEquals(Arrays.asList("[123.45, -0.01]", "[123.45, -0.01]", "[123.45, -0.01]"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); }");
+
+    List<String> values = writeAndRead(schema, record -> appendUnscaled(record, physicalType, 12345), "amount",
+        new DecimalType(precision, 2), row -> row.getDecimal(0, precision, 2).toString());
+
+    assertEquals(Arrays.asList("123.45", "123.45", "123.45"), values);
+  }
+
+  @Test
+  void testReadThreeLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.addGroup("list").append("element", 1);
+      list.addGroup("list").append("element", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadTwoLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated int32 array; } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.append("array", 1);
+      list.append("array", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  /**
+   * Appends the unscaled value of a decimal to the {@code amount} field, as the given physical type.
+   */
+  private static void appendUnscaled(Group group, String physicalType, int unscaled) {
+    if (physicalType.equals("int32")) {
+      group.append("amount", unscaled);
+    } else {
+      group.append("amount", (long) unscaled);
+    }
+  }
+
+  private static MessageType itemsSchema(ListEncoding encoding) {
+    String tags = encoding.schema("tags", "optional binary value (UTF8);");
+    String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+    return MessageTypeParser.parseMessageType("message m { " + items + " }");
+  }
+
+  /**
+   * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+   */
+  private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+    int[] recordIndex = {0};
+    return record -> {
+      int index = recordIndex[0]++;
+      if (index == 0) {
+        Group items = record.addGroup("items");
+        addItem(encoding, items, "a", "x", "y");
+        addItem(encoding, items, "b", "z");
+      } else if (index == 1) {
+        addItem(encoding, record.addGroup("items"), "c");
+      }
+    };
+  }
+
+  private static void addItem(ListEncoding encoding, Group items, String name, String... tags) {
+    Group item = encoding.addElement(items, "items");
+    item.append("name", name);
+    Group tagList = item.addGroup("tags");
+    for (String tag : tags) {
+      encoding.addElement(tagList, "tags").append("value", tag);
+    }
+  }
+
+  /**
+   * Returns the items type with fields that the written files don't have: {@code added_scores} on
+   * the item and {@code added_tag_field} on the nested tag.
+   */
+  private static LogicalType evolvedItemsType() {
+    RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+    return new ArrayType(
+        rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+            new ArrayType(tagType)));
+  }
+
+  private static RowType rowType(String[] names, LogicalType... types) {
+    return RowType.of(types, names);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter) throws IOException {
+    return writeAndRead(schema, recordWriter, columnName, columnType, formatter, DEFAULT_BATCH_SIZE);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter,
+      int batchSize) throws IOException {
+    Path path = new Path(new File(tempDir, "data.parquet").getAbsolutePath());
+    Configuration conf = new Configuration();
+    SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(path).withType(schema).withConf(conf).build()) {
+      for (int i = 0; i < 3; i++) {
+        Group record = factory.newGroup();
+        recordWriter.accept(record);
+        writer.write(record);
+      }
+    }
+
+    List<String> values = new ArrayList<>();
+    try (ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+        true,
+        true,
+        conf,
+        new LogicalType[] {columnType},
+        new String[] {columnName},
+        VectorizedColumnBatch::new,
+        batchSize,
+        path,
+        0,
+        new File(tempDir, "data.parquet").length(),
+        null,
+        null)) {
+      while (!reader.reachedEnd()) {
+        values.add(formatter.apply(reader.nextRecord()));
+      }
+    }
+    return values;
+  }
+
+  /**
+   * Formats {@code array<row<name, [added_scores,] array<row<value[, added_tag_field]>>>>}. With
+   * {@code evolved}, the added fields are printed in parentheses after the existing ones.
+   */
+  private static String formatItems(RowData row, boolean evolved) {
+    if (row.isNullAt(0)) {
+      return "null";
+    }
+    ArrayData items = row.getArray(0);
+    int tagsPos = evolved ? 2 : 1;
+    List<String> formatted = new ArrayList<>();
+    for (int i = 0; i < items.size(); i++) {
+      RowData item = items.getRow(i, tagsPos + 1);
+      String name = item.getString(0).toString();
+      if (evolved) {
+        name += "(" + (item.isNullAt(1) ? "null" : formatInts(item.getArray(1))) + ")";
+      }
+      ArrayData tags = item.getArray(tagsPos);
+      List<String> tagValues = new ArrayList<>();
+      for (int j = 0; j < tags.size(); j++) {
+        RowData tag = tags.getRow(j, evolved ? 2 : 1);
+        String value = tag.getString(0).toString();
+        if (evolved) {
+          value += "(" + (tag.isNullAt(1) ? "null" : tag.getString(1).toString()) + ")";
+        }
+        tagValues.add(value);
+      }
+      formatted.add(name + ":" + tagValues);
+    }
+    return formatted.toString();
+  }
+
+  private static String formatInts(ArrayData array) {
+    List<Integer> ints = new ArrayList<>();
+    for (int i = 0; i < array.size(); i++) {
+      ints.add(array.getInt(i));
+    }
+    return ints.toString();
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
new file mode 100644
index 000000000000..0bfb1a8aa668
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.hudi.table.format.cow.utils;
+
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link NestedPositionUtil}.
+ */
+class TestNestedPositionUtil {
+
+  /** An optional top-level row whose last field is an optional list of optional ints. */
+  private static final ParquetGroupField ROW =
+      new ParquetGroupField(RowType.of(new IntType()), 0, 1, false, Collections.emptyList());
+
+  @Test
+  void testRowOffsetsCountOneValuePerRowWhenTheLastFieldIsAnArray() {
+    // two rows of three elements each, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 4, 4, 4, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 1, 1, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertNull(position.getIsNull());
+  }
+
+  @Test
+  void testRowOffsetsWithNullRows() {
+    // a row of three elements, a null row, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 0, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertArrayEquals(new boolean[] {false, true, false}, position.getIsNull());
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index 756e7e7069da..d74f4d71e1ca 100644
--- a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -20,6 +20,7 @@ package org.apache.hudi.table.format.cow;
 
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
 import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
@@ -498,19 +499,24 @@ public class ParquetSplitReaderUtil {
    * matches {@code fieldType}; for ROW types missing physical fields are slotted with null-filled
    * vectors (sourced from {@link #createVectorFromConstant}) so that the Dremel assembler in
    * {@link NestedColumnReader} can pass them through unchanged.
+   *
+   * @param parentName name of the group that contains {@code physicalType}, which decides whether a
+   *     repeated group named {@code <parentName>_tuple} is a list element
    */
   public static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> descriptors) {
-    return createWritableColumnVector(batchSize, fieldType, physicalType, descriptors, 0);
+    return createWritableColumnVector(batchSize, fieldType, physicalType, parentName, descriptors, 0);
   }
 
   private static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> columns,
       int depth) {
     List<ColumnDescriptor> descriptors = filterDescriptors(depth, physicalType, columns);
@@ -570,26 +576,35 @@ public class ParquetSplitReaderUtil {
         return new HeapTimestampVector(batchSize);
       case DECIMAL:
         checkArgument(
-            (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
-                    || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
-                && primitiveType.getOriginalType() == OriginalType.DECIMAL,
+            primitiveType.getOriginalType() == OriginalType.DECIMAL,
             getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
-        return new HeapDecimalVector(batchSize);
+        switch (typeName) {
+          case INT32:
+            return new HeapIntVector(batchSize);
+          case INT64:
+            return new HeapLongVector(batchSize);
+          case BINARY:
+          case FIXED_LEN_BYTE_ARRAY:
+            return new HeapDecimalVector(batchSize);
+          default:
+            throw new IllegalArgumentException(getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+        }
       case ARRAY:
         ArrayType arrayType = (ArrayType) fieldType;
+        Triple<Type, String, Integer> element = unwrapListElementType(physicalType, parentName, depth);
         return new HeapArrayVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+            createWritableColumnVector(batchSize, arrayType.getElementType(), element.getLeft(),
+                element.getMiddle(), descriptors, element.getRight()));
       case MAP: {
         MapType mapType = (MapType) fieldType;
         GroupType repeatedType = unwrapMapRepeatedType(physicalType);
         return new HeapMapColumnVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, mapType.getKeyType(), repeatedType.getType(0), descriptors, depth + 2),
-            createWritableColumnVector(
-                batchSize, mapType.getValueType(), repeatedType.getType(1), descriptors, depth + 2));
+            createWritableColumnVector(batchSize, mapType.getKeyType(), repeatedType.getType(0),
+                repeatedType.getName(), descriptors, depth + 2),
+            createWritableColumnVector(batchSize, mapType.getValueType(), repeatedType.getType(1),
+                repeatedType.getName(), descriptors, depth + 2));
       }
       case MULTISET: {
         MultisetType multisetType = (MultisetType) fieldType;
@@ -600,12 +615,14 @@ public class ParquetSplitReaderUtil {
                 batchSize,
                 multisetType.getElementType(),
                 repeatedType.getType(0),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2),
             createWritableColumnVector(
                 batchSize,
                 new IntType(false),
                 repeatedType.getType(1),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2));
       }
@@ -627,6 +644,7 @@ public class ParquetSplitReaderUtil {
                     batchSize,
                     rowType.getTypeAt(i),
                     groupType.getType(fieldIndex),
+                    groupType.getName(),
                     descriptors,
                     depth + 1);
           }
@@ -681,6 +699,46 @@ public class ParquetSplitReaderUtil {
         "Invalid Variant Parquet schema: missing binary field '" + fieldName + "'.");
   }
 
+  /**
+   * Returns the element type of a LIST physical type, the name of the group that contains it and its
+   * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+   * so that the vectors line up with the {@code ParquetField} tree.
+   *
+   * @param parentName name of the group that contains {@code listType}
+   */
+  private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+    Type elementType = listType;
+    String elementParentName = parentName;
+    int elementDepth = depth;
+    while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    return Triple.of(elementType, elementParentName, elementDepth);
+  }
+
+  /**
+   * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+   * rather than the element itself. Shared by {@link #unwrapListElementType} and
+   * {@link #getArrayElementColumn}, so that the vectors and the {@code ParquetField} tree pick the
+   * same element.
+   *
+   * @param parentName name of the group that contains {@code repeatedType}
+   */
+  private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
+    return !repeatedType.isPrimitive()
+        && repeatedType.getLogicalTypeAnnotation() == null
+        && repeatedType.asGroupType().getFieldCount() == 1
+        && !repeatedType.getName().equals("array")
+        && !repeatedType.getName().equals(parentName + "_tuple");
+  }
+
   /**
    * Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET physical type, matching
    * Parquet's canonical 3-level map encoding.
@@ -906,10 +964,7 @@ public class ParquetSplitReaderUtil {
 
     // Three-level list: skip the synthetic `element` / `list` wrapper when present.
     if (columnIO instanceof GroupColumnIO
-        && columnIO.getType().getLogicalTypeAnnotation() == null
-        && ((GroupColumnIO) columnIO).getChildrenCount() == 1
-        && !columnIO.getName().equals("array")
-        && !columnIO.getName().equals(columnIO.getParent().getName() + "_tuple")) {
+        && isThreeLevelListRepeatedType(columnIO.getType(), columnIO.getParent().getName())) {
       return ((GroupColumnIO) columnIO).getChild(0);
     }
     return columnIO;
diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index 3f2f8976b69b..52745f747d25 100644
--- a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
+++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -69,7 +69,9 @@ public class NestedPositionUtil {
       }
     }
     if (nullValuesCount == 0) {
-      return new RowPosition(null, fieldDefinitionLevels.length);
+      // Diverges from Flink, which returns fieldDefinitionLevels.length here: that also counts the
+      // skipped values of a last field that is an array, one per element instead of one per row.
+      return new RowPosition(null, nullRowFlags.size());
     }
     return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
   }
diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
index 7be03dacc218..1a0f676c22ec 100644
--- a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
+++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -50,6 +50,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createVectorFromConstant;
+
 /**
  * ColumnReader used to read a {@code Group} type in Parquet ({@code Map}, {@code Array}, {@code
  * Row}). Resolves nested structures using Dremel striping/assembly; see <a
@@ -142,6 +144,17 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> {
             levelDelegation.getDefinitionLevel(),
             levelDelegation.getRepetitionLevel());
 
+    // Inside an array the row has one position per element, which can exceed the batch size the
+    // placeholders of absent fields were sized with, so grow them to the position count.
+    // Only a ROW can have absent fields: a VARIANT, also read here, always has both of its fields.
+    int positionsCount = rowPosition.getPositionsCount();
+    for (int i = 0; i < children.size(); i++) {
+      if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
+        finalChildrenVectors[i] = (WritableColumnVector) createVectorFromConstant(
+            ((RowType) field.getType()).getTypeAt(i), null, positionsCount);
+      }
+    }
+
     // If row was inside the structure, then we need to renew the vector to reset the
     // capacity.
     if (inside) {
diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index 1826419db5d4..b9f7527a91a8 100644
--- a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -250,6 +250,7 @@ public class ParquetColumnarRowSplitReader implements Closeable {
                 batchSize,
                 requestedTypes[i],
                 types.get(i),
+                requestedSchema.getName(),
                 descriptors);
       } catch (IllegalArgumentException e) {
         String fieldName = requestedSchema.getFieldName(i);
diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
new file mode 100644
index 000000000000..3763003e554d
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
@@ -0,0 +1,415 @@
+/*
+ * 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.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+  private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+  private static final int DEFAULT_BATCH_SIZE = 2048;
+
+  @TempDir
+  File tempDir;
+
+  /**
+   * Physical encodings of a list of single- or multi-field element groups.
+   */
+  enum ListEncoding {
+    THREE_LEVEL {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+            + elementFields + " } } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("list").addGroup("element");
+      }
+    },
+    TWO_LEVEL_ARRAY {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("array");
+      }
+    },
+    TWO_LEVEL_TUPLE {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+            + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup(name + "_tuple");
+      }
+    };
+
+    abstract String schema(String name, String elementFields);
+
+    abstract Group addElement(Group list, String name);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+    MessageType schema = itemsSchema(encoding);
+    LogicalType itemsType = new ArrayType(
+        rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+    List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+        row -> formatItems(row, false));
+
+    assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), DEFAULT_BATCH_SIZE);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount(ListEncoding encoding)
+      throws IOException {
+    // The first batch holds two rows with three items and three tags, so the placeholders of the
+    // absent fields must grow beyond the batch size.
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), 2);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @Test
+  void testReadBareRepeatedGroupNamedAfterItsParentAsArrayOfRow() throws IOException {
+    // A repeated group with one field named <parent>_tuple is the element itself, not a wrapper.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { repeated group info_tuple { optional int32 x; } } }");
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.addGroup("info_tuple").append("x", 1);
+      info.addGroup("info_tuple").append("x", 2);
+    };
+    LogicalType infoType = rowType(new String[] {"info_tuple"}, new ArrayType(rowType(new String[] {"x"}, new IntType())));
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      ArrayData tuples = row.getRow(0, 1).getArray(0);
+      List<Integer> xs = new ArrayList<>();
+      for (int i = 0; i < tuples.size(); i++) {
+        xs.add(tuples.getRow(i, 1).getInt(0));
+      }
+      return xs.toString();
+    });
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadRowWithFieldMissingFromFileAfterLongArray() throws IOException {
+    // The row's last field in the file is an array with more elements than the batch size, which
+    // must not count as more rows when sizing the placeholder of the absent field.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { optional binary name (UTF8); "
+            + "optional group nums (LIST) { repeated group list { optional int32 element; } } } }");
+    int[] recordIndex = {0};
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.append("name", "r" + recordIndex[0]++);
+      Group nums = info.addGroup("nums");
+      for (int i = 0; i < 100; i++) {
+        nums.addGroup("list").append("element", i);
+      }
+    };
+    LogicalType infoType = rowType(new String[] {"name", "nums", "added"}, STRING, new ArrayType(new IntType()), STRING);
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      RowData info = row.getRow(0, 3);
+      return info.getString(0) + ":" + info.getArray(1).size() + ":" + (info.isNullAt(2) ? "null" : info.getString(2));
+    }, 2);
+
+    assertEquals(Arrays.asList("r0:100:null", "r1:100:null", "r2:100:null"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadArrayOfRowWithIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group items (LIST) { repeated group list { optional group element { "
+            + "optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); } } } }");
+    Consumer<Group> writer = record -> {
+      Group items = record.addGroup("items");
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, 12345);
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, -1);
+    };
+    LogicalType itemsType = new ArrayType(rowType(new String[] {"amount"}, new DecimalType(precision, 2)));
+
+    List<String> values = writeAndRead(schema, writer, "items", itemsType, row -> {
+      ArrayData items = row.getArray(0);
+      List<String> amounts = new ArrayList<>();
+      for (int i = 0; i < items.size(); i++) {
+        amounts.add(items.getRow(i, 1).getDecimal(0, precision, 2).toString());
+      }
+      return amounts.toString();
+    });
+
+    assertEquals(Arrays.asList("[123.45, -0.01]", "[123.45, -0.01]", "[123.45, -0.01]"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); }");
+
+    List<String> values = writeAndRead(schema, record -> appendUnscaled(record, physicalType, 12345), "amount",
+        new DecimalType(precision, 2), row -> row.getDecimal(0, precision, 2).toString());
+
+    assertEquals(Arrays.asList("123.45", "123.45", "123.45"), values);
+  }
+
+  @Test
+  void testReadThreeLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.addGroup("list").append("element", 1);
+      list.addGroup("list").append("element", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadTwoLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated int32 array; } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.append("array", 1);
+      list.append("array", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  /**
+   * Appends the unscaled value of a decimal to the {@code amount} field, as the given physical type.
+   */
+  private static void appendUnscaled(Group group, String physicalType, int unscaled) {
+    if (physicalType.equals("int32")) {
+      group.append("amount", unscaled);
+    } else {
+      group.append("amount", (long) unscaled);
+    }
+  }
+
+  private static MessageType itemsSchema(ListEncoding encoding) {
+    String tags = encoding.schema("tags", "optional binary value (UTF8);");
+    String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+    return MessageTypeParser.parseMessageType("message m { " + items + " }");
+  }
+
+  /**
+   * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+   */
+  private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+    int[] recordIndex = {0};
+    return record -> {
+      int index = recordIndex[0]++;
+      if (index == 0) {
+        Group items = record.addGroup("items");
+        addItem(encoding, items, "a", "x", "y");
+        addItem(encoding, items, "b", "z");
+      } else if (index == 1) {
+        addItem(encoding, record.addGroup("items"), "c");
+      }
+    };
+  }
+
+  private static void addItem(ListEncoding encoding, Group items, String name, String... tags) {
+    Group item = encoding.addElement(items, "items");
+    item.append("name", name);
+    Group tagList = item.addGroup("tags");
+    for (String tag : tags) {
+      encoding.addElement(tagList, "tags").append("value", tag);
+    }
+  }
+
+  /**
+   * Returns the items type with fields that the written files don't have: {@code added_scores} on
+   * the item and {@code added_tag_field} on the nested tag.
+   */
+  private static LogicalType evolvedItemsType() {
+    RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+    return new ArrayType(
+        rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+            new ArrayType(tagType)));
+  }
+
+  private static RowType rowType(String[] names, LogicalType... types) {
+    return RowType.of(types, names);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter) throws IOException {
+    return writeAndRead(schema, recordWriter, columnName, columnType, formatter, DEFAULT_BATCH_SIZE);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter,
+      int batchSize) throws IOException {
+    Path path = new Path(new File(tempDir, "data.parquet").getAbsolutePath());
+    Configuration conf = new Configuration();
+    SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(path).withType(schema).withConf(conf).build()) {
+      for (int i = 0; i < 3; i++) {
+        Group record = factory.newGroup();
+        recordWriter.accept(record);
+        writer.write(record);
+      }
+    }
+
+    List<String> values = new ArrayList<>();
+    try (ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+        true,
+        true,
+        conf,
+        new LogicalType[] {columnType},
+        new String[] {columnName},
+        VectorizedColumnBatch::new,
+        batchSize,
+        path,
+        0,
+        new File(tempDir, "data.parquet").length(),
+        null,
+        null)) {
+      while (!reader.reachedEnd()) {
+        values.add(formatter.apply(reader.nextRecord()));
+      }
+    }
+    return values;
+  }
+
+  /**
+   * Formats {@code array<row<name, [added_scores,] array<row<value[, added_tag_field]>>>>}. With
+   * {@code evolved}, the added fields are printed in parentheses after the existing ones.
+   */
+  private static String formatItems(RowData row, boolean evolved) {
+    if (row.isNullAt(0)) {
+      return "null";
+    }
+    ArrayData items = row.getArray(0);
+    int tagsPos = evolved ? 2 : 1;
+    List<String> formatted = new ArrayList<>();
+    for (int i = 0; i < items.size(); i++) {
+      RowData item = items.getRow(i, tagsPos + 1);
+      String name = item.getString(0).toString();
+      if (evolved) {
+        name += "(" + (item.isNullAt(1) ? "null" : formatInts(item.getArray(1))) + ")";
+      }
+      ArrayData tags = item.getArray(tagsPos);
+      List<String> tagValues = new ArrayList<>();
+      for (int j = 0; j < tags.size(); j++) {
+        RowData tag = tags.getRow(j, evolved ? 2 : 1);
+        String value = tag.getString(0).toString();
+        if (evolved) {
+          value += "(" + (tag.isNullAt(1) ? "null" : tag.getString(1).toString()) + ")";
+        }
+        tagValues.add(value);
+      }
+      formatted.add(name + ":" + tagValues);
+    }
+    return formatted.toString();
+  }
+
+  private static String formatInts(ArrayData array) {
+    List<Integer> ints = new ArrayList<>();
+    for (int i = 0; i < array.size(); i++) {
+      ints.add(array.getInt(i));
+    }
+    return ints.toString();
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
new file mode 100644
index 000000000000..0bfb1a8aa668
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.hudi.table.format.cow.utils;
+
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link NestedPositionUtil}.
+ */
+class TestNestedPositionUtil {
+
+  /** An optional top-level row whose last field is an optional list of optional ints. */
+  private static final ParquetGroupField ROW =
+      new ParquetGroupField(RowType.of(new IntType()), 0, 1, false, Collections.emptyList());
+
+  @Test
+  void testRowOffsetsCountOneValuePerRowWhenTheLastFieldIsAnArray() {
+    // two rows of three elements each, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 4, 4, 4, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 1, 1, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertNull(position.getIsNull());
+  }
+
+  @Test
+  void testRowOffsetsWithNullRows() {
+    // a row of three elements, a null row, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 0, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertArrayEquals(new boolean[] {false, true, false}, position.getIsNull());
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index 756e7e7069da..d74f4d71e1ca 100644
--- a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++ b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -20,6 +20,7 @@ package org.apache.hudi.table.format.cow;
 
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
 import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
@@ -498,19 +499,24 @@ public class ParquetSplitReaderUtil {
    * matches {@code fieldType}; for ROW types missing physical fields are slotted with null-filled
    * vectors (sourced from {@link #createVectorFromConstant}) so that the Dremel assembler in
    * {@link NestedColumnReader} can pass them through unchanged.
+   *
+   * @param parentName name of the group that contains {@code physicalType}, which decides whether a
+   *     repeated group named {@code <parentName>_tuple} is a list element
    */
   public static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> descriptors) {
-    return createWritableColumnVector(batchSize, fieldType, physicalType, descriptors, 0);
+    return createWritableColumnVector(batchSize, fieldType, physicalType, parentName, descriptors, 0);
   }
 
   private static WritableColumnVector createWritableColumnVector(
       int batchSize,
       LogicalType fieldType,
       Type physicalType,
+      String parentName,
       List<ColumnDescriptor> columns,
       int depth) {
     List<ColumnDescriptor> descriptors = filterDescriptors(depth, physicalType, columns);
@@ -570,26 +576,35 @@ public class ParquetSplitReaderUtil {
         return new HeapTimestampVector(batchSize);
       case DECIMAL:
         checkArgument(
-            (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
-                    || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
-                && primitiveType.getOriginalType() == OriginalType.DECIMAL,
+            primitiveType.getOriginalType() == OriginalType.DECIMAL,
             getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
-        return new HeapDecimalVector(batchSize);
+        switch (typeName) {
+          case INT32:
+            return new HeapIntVector(batchSize);
+          case INT64:
+            return new HeapLongVector(batchSize);
+          case BINARY:
+          case FIXED_LEN_BYTE_ARRAY:
+            return new HeapDecimalVector(batchSize);
+          default:
+            throw new IllegalArgumentException(getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+        }
       case ARRAY:
         ArrayType arrayType = (ArrayType) fieldType;
+        Triple<Type, String, Integer> element = unwrapListElementType(physicalType, parentName, depth);
         return new HeapArrayVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+            createWritableColumnVector(batchSize, arrayType.getElementType(), element.getLeft(),
+                element.getMiddle(), descriptors, element.getRight()));
       case MAP: {
         MapType mapType = (MapType) fieldType;
         GroupType repeatedType = unwrapMapRepeatedType(physicalType);
         return new HeapMapColumnVector(
             batchSize,
-            createWritableColumnVector(
-                batchSize, mapType.getKeyType(), repeatedType.getType(0), descriptors, depth + 2),
-            createWritableColumnVector(
-                batchSize, mapType.getValueType(), repeatedType.getType(1), descriptors, depth + 2));
+            createWritableColumnVector(batchSize, mapType.getKeyType(), repeatedType.getType(0),
+                repeatedType.getName(), descriptors, depth + 2),
+            createWritableColumnVector(batchSize, mapType.getValueType(), repeatedType.getType(1),
+                repeatedType.getName(), descriptors, depth + 2));
       }
       case MULTISET: {
         MultisetType multisetType = (MultisetType) fieldType;
@@ -600,12 +615,14 @@ public class ParquetSplitReaderUtil {
                 batchSize,
                 multisetType.getElementType(),
                 repeatedType.getType(0),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2),
             createWritableColumnVector(
                 batchSize,
                 new IntType(false),
                 repeatedType.getType(1),
+                repeatedType.getName(),
                 descriptors,
                 depth + 2));
       }
@@ -627,6 +644,7 @@ public class ParquetSplitReaderUtil {
                     batchSize,
                     rowType.getTypeAt(i),
                     groupType.getType(fieldIndex),
+                    groupType.getName(),
                     descriptors,
                     depth + 1);
           }
@@ -681,6 +699,46 @@ public class ParquetSplitReaderUtil {
         "Invalid Variant Parquet schema: missing binary field '" + fieldName + "'.");
   }
 
+  /**
+   * Returns the element type of a LIST physical type, the name of the group that contains it and its
+   * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+   * so that the vectors line up with the {@code ParquetField} tree.
+   *
+   * @param parentName name of the group that contains {@code listType}
+   */
+  private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+    Type elementType = listType;
+    String elementParentName = parentName;
+    int elementDepth = depth;
+    while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+      elementParentName = elementType.getName();
+      elementType = elementType.asGroupType().getType(0);
+      elementDepth++;
+    }
+    return Triple.of(elementType, elementParentName, elementDepth);
+  }
+
+  /**
+   * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+   * rather than the element itself. Shared by {@link #unwrapListElementType} and
+   * {@link #getArrayElementColumn}, so that the vectors and the {@code ParquetField} tree pick the
+   * same element.
+   *
+   * @param parentName name of the group that contains {@code repeatedType}
+   */
+  private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
+    return !repeatedType.isPrimitive()
+        && repeatedType.getLogicalTypeAnnotation() == null
+        && repeatedType.asGroupType().getFieldCount() == 1
+        && !repeatedType.getName().equals("array")
+        && !repeatedType.getName().equals(parentName + "_tuple");
+  }
+
   /**
    * Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET physical type, matching
    * Parquet's canonical 3-level map encoding.
@@ -906,10 +964,7 @@ public class ParquetSplitReaderUtil {
 
     // Three-level list: skip the synthetic `element` / `list` wrapper when present.
     if (columnIO instanceof GroupColumnIO
-        && columnIO.getType().getLogicalTypeAnnotation() == null
-        && ((GroupColumnIO) columnIO).getChildrenCount() == 1
-        && !columnIO.getName().equals("array")
-        && !columnIO.getName().equals(columnIO.getParent().getName() + "_tuple")) {
+        && isThreeLevelListRepeatedType(columnIO.getType(), columnIO.getParent().getName())) {
       return ((GroupColumnIO) columnIO).getChild(0);
     }
     return columnIO;
diff --git a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index 3f2f8976b69b..52745f747d25 100644
--- a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
+++ b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -69,7 +69,9 @@ public class NestedPositionUtil {
       }
     }
     if (nullValuesCount == 0) {
-      return new RowPosition(null, fieldDefinitionLevels.length);
+      // Diverges from Flink, which returns fieldDefinitionLevels.length here: that also counts the
+      // skipped values of a last field that is an array, one per element instead of one per row.
+      return new RowPosition(null, nullRowFlags.size());
     }
     return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
   }
diff --git a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
index 7be03dacc218..1a0f676c22ec 100644
--- a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
+++ b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -50,6 +50,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createVectorFromConstant;
+
 /**
  * ColumnReader used to read a {@code Group} type in Parquet ({@code Map}, {@code Array}, {@code
  * Row}). Resolves nested structures using Dremel striping/assembly; see <a
@@ -142,6 +144,17 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> {
             levelDelegation.getDefinitionLevel(),
             levelDelegation.getRepetitionLevel());
 
+    // Inside an array the row has one position per element, which can exceed the batch size the
+    // placeholders of absent fields were sized with, so grow them to the position count.
+    // Only a ROW can have absent fields: a VARIANT, also read here, always has both of its fields.
+    int positionsCount = rowPosition.getPositionsCount();
+    for (int i = 0; i < children.size(); i++) {
+      if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
+        finalChildrenVectors[i] = (WritableColumnVector) createVectorFromConstant(
+            ((RowType) field.getType()).getTypeAt(i), null, positionsCount);
+      }
+    }
+
     // If row was inside the structure, then we need to renew the vector to reset the
     // capacity.
     if (inside) {
diff --git a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index d077630b41cf..eca0fceae6aa 100644
--- a/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink2.2.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -250,6 +250,7 @@ public class ParquetColumnarRowSplitReader implements Closeable {
                 batchSize,
                 requestedTypes[i],
                 types.get(i),
+                requestedSchema.getName(),
                 descriptors);
       } catch (IllegalArgumentException e) {
         String fieldName = requestedSchema.getFieldName(i);
diff --git a/hudi-flink-datasource/hudi-flink2.2.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.2.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
new file mode 100644
index 000000000000..3763003e554d
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink2.2.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
@@ -0,0 +1,415 @@
+/*
+ * 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.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+  private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+  private static final int DEFAULT_BATCH_SIZE = 2048;
+
+  @TempDir
+  File tempDir;
+
+  /**
+   * Physical encodings of a list of single- or multi-field element groups.
+   */
+  enum ListEncoding {
+    THREE_LEVEL {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+            + elementFields + " } } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("list").addGroup("element");
+      }
+    },
+    TWO_LEVEL_ARRAY {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup("array");
+      }
+    },
+    TWO_LEVEL_TUPLE {
+      @Override
+      String schema(String name, String elementFields) {
+        return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+            + elementFields + " } }";
+      }
+
+      @Override
+      Group addElement(Group list, String name) {
+        return list.addGroup(name + "_tuple");
+      }
+    };
+
+    abstract String schema(String name, String elementFields);
+
+    abstract Group addElement(Group list, String name);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+    MessageType schema = itemsSchema(encoding);
+    LogicalType itemsType = new ArrayType(
+        rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+    List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+        row -> formatItems(row, false));
+
+    assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), DEFAULT_BATCH_SIZE);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @ParameterizedTest
+  @EnumSource(ListEncoding.class)
+  void testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount(ListEncoding encoding)
+      throws IOException {
+    // The first batch holds two rows with three items and three tags, so the placeholders of the
+    // absent fields must grow beyond the batch size.
+    List<String> values = writeAndRead(itemsSchema(encoding), itemsWriter(encoding), "items",
+        evolvedItemsType(), row -> formatItems(row, true), 2);
+
+    assertEquals(
+        Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+  }
+
+  @Test
+  void testReadBareRepeatedGroupNamedAfterItsParentAsArrayOfRow() throws IOException {
+    // A repeated group with one field named <parent>_tuple is the element itself, not a wrapper.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { repeated group info_tuple { optional int32 x; } } }");
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.addGroup("info_tuple").append("x", 1);
+      info.addGroup("info_tuple").append("x", 2);
+    };
+    LogicalType infoType = rowType(new String[] {"info_tuple"}, new ArrayType(rowType(new String[] {"x"}, new IntType())));
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      ArrayData tuples = row.getRow(0, 1).getArray(0);
+      List<Integer> xs = new ArrayList<>();
+      for (int i = 0; i < tuples.size(); i++) {
+        xs.add(tuples.getRow(i, 1).getInt(0));
+      }
+      return xs.toString();
+    });
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadRowWithFieldMissingFromFileAfterLongArray() throws IOException {
+    // The row's last field in the file is an array with more elements than the batch size, which
+    // must not count as more rows when sizing the placeholder of the absent field.
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group info { optional binary name (UTF8); "
+            + "optional group nums (LIST) { repeated group list { optional int32 element; } } } }");
+    int[] recordIndex = {0};
+    Consumer<Group> writer = record -> {
+      Group info = record.addGroup("info");
+      info.append("name", "r" + recordIndex[0]++);
+      Group nums = info.addGroup("nums");
+      for (int i = 0; i < 100; i++) {
+        nums.addGroup("list").append("element", i);
+      }
+    };
+    LogicalType infoType = rowType(new String[] {"name", "nums", "added"}, STRING, new ArrayType(new IntType()), STRING);
+
+    List<String> values = writeAndRead(schema, writer, "info", infoType, row -> {
+      RowData info = row.getRow(0, 3);
+      return info.getString(0) + ":" + info.getArray(1).size() + ":" + (info.isNullAt(2) ? "null" : info.getString(2));
+    }, 2);
+
+    assertEquals(Arrays.asList("r0:100:null", "r1:100:null", "r2:100:null"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadArrayOfRowWithIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group items (LIST) { repeated group list { optional group element { "
+            + "optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); } } } }");
+    Consumer<Group> writer = record -> {
+      Group items = record.addGroup("items");
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, 12345);
+      appendUnscaled(items.addGroup("list").addGroup("element"), physicalType, -1);
+    };
+    LogicalType itemsType = new ArrayType(rowType(new String[] {"amount"}, new DecimalType(precision, 2)));
+
+    List<String> values = writeAndRead(schema, writer, "items", itemsType, row -> {
+      ArrayData items = row.getArray(0);
+      List<String> amounts = new ArrayList<>();
+      for (int i = 0; i < items.size(); i++) {
+        amounts.add(items.getRow(i, 1).getDecimal(0, precision, 2).toString());
+      }
+      return amounts.toString();
+    });
+
+    assertEquals(Arrays.asList("[123.45, -0.01]", "[123.45, -0.01]", "[123.45, -0.01]"), values);
+  }
+
+  @ParameterizedTest
+  @CsvSource({"int32, 9", "int64, 18"})
+  void testReadIntegerEncodedDecimal(String physicalType, int precision) throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional " + physicalType + " amount (DECIMAL(" + precision + ",2)); }");
+
+    List<String> values = writeAndRead(schema, record -> appendUnscaled(record, physicalType, 12345), "amount",
+        new DecimalType(precision, 2), row -> row.getDecimal(0, precision, 2).toString());
+
+    assertEquals(Arrays.asList("123.45", "123.45", "123.45"), values);
+  }
+
+  @Test
+  void testReadThreeLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.addGroup("list").append("element", 1);
+      list.addGroup("list").append("element", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  @Test
+  void testReadTwoLevelArrayOfPrimitive() throws IOException {
+    MessageType schema = MessageTypeParser.parseMessageType(
+        "message m { optional group nums (LIST) { repeated int32 array; } }");
+    Consumer<Group> writer = record -> {
+      Group list = record.addGroup("nums");
+      list.append("array", 1);
+      list.append("array", 2);
+    };
+
+    List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+        row -> formatInts(row.getArray(0)));
+
+    assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+  }
+
+  /**
+   * Appends the unscaled value of a decimal to the {@code amount} field, as the given physical type.
+   */
+  private static void appendUnscaled(Group group, String physicalType, int unscaled) {
+    if (physicalType.equals("int32")) {
+      group.append("amount", unscaled);
+    } else {
+      group.append("amount", (long) unscaled);
+    }
+  }
+
+  private static MessageType itemsSchema(ListEncoding encoding) {
+    String tags = encoding.schema("tags", "optional binary value (UTF8);");
+    String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+    return MessageTypeParser.parseMessageType("message m { " + items + " }");
+  }
+
+  /**
+   * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+   */
+  private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+    int[] recordIndex = {0};
+    return record -> {
+      int index = recordIndex[0]++;
+      if (index == 0) {
+        Group items = record.addGroup("items");
+        addItem(encoding, items, "a", "x", "y");
+        addItem(encoding, items, "b", "z");
+      } else if (index == 1) {
+        addItem(encoding, record.addGroup("items"), "c");
+      }
+    };
+  }
+
+  private static void addItem(ListEncoding encoding, Group items, String name, String... tags) {
+    Group item = encoding.addElement(items, "items");
+    item.append("name", name);
+    Group tagList = item.addGroup("tags");
+    for (String tag : tags) {
+      encoding.addElement(tagList, "tags").append("value", tag);
+    }
+  }
+
+  /**
+   * Returns the items type with fields that the written files don't have: {@code added_scores} on
+   * the item and {@code added_tag_field} on the nested tag.
+   */
+  private static LogicalType evolvedItemsType() {
+    RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+    return new ArrayType(
+        rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+            new ArrayType(tagType)));
+  }
+
+  private static RowType rowType(String[] names, LogicalType... types) {
+    return RowType.of(types, names);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter) throws IOException {
+    return writeAndRead(schema, recordWriter, columnName, columnType, formatter, DEFAULT_BATCH_SIZE);
+  }
+
+  private List<String> writeAndRead(
+      MessageType schema,
+      Consumer<Group> recordWriter,
+      String columnName,
+      LogicalType columnType,
+      Function<RowData, String> formatter,
+      int batchSize) throws IOException {
+    Path path = new Path(new File(tempDir, "data.parquet").getAbsolutePath());
+    Configuration conf = new Configuration();
+    SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(path).withType(schema).withConf(conf).build()) {
+      for (int i = 0; i < 3; i++) {
+        Group record = factory.newGroup();
+        recordWriter.accept(record);
+        writer.write(record);
+      }
+    }
+
+    List<String> values = new ArrayList<>();
+    try (ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+        true,
+        true,
+        conf,
+        new LogicalType[] {columnType},
+        new String[] {columnName},
+        VectorizedColumnBatch::new,
+        batchSize,
+        path,
+        0,
+        new File(tempDir, "data.parquet").length(),
+        null,
+        null)) {
+      while (!reader.reachedEnd()) {
+        values.add(formatter.apply(reader.nextRecord()));
+      }
+    }
+    return values;
+  }
+
+  /**
+   * Formats {@code array<row<name, [added_scores,] array<row<value[, added_tag_field]>>>>}. With
+   * {@code evolved}, the added fields are printed in parentheses after the existing ones.
+   */
+  private static String formatItems(RowData row, boolean evolved) {
+    if (row.isNullAt(0)) {
+      return "null";
+    }
+    ArrayData items = row.getArray(0);
+    int tagsPos = evolved ? 2 : 1;
+    List<String> formatted = new ArrayList<>();
+    for (int i = 0; i < items.size(); i++) {
+      RowData item = items.getRow(i, tagsPos + 1);
+      String name = item.getString(0).toString();
+      if (evolved) {
+        name += "(" + (item.isNullAt(1) ? "null" : formatInts(item.getArray(1))) + ")";
+      }
+      ArrayData tags = item.getArray(tagsPos);
+      List<String> tagValues = new ArrayList<>();
+      for (int j = 0; j < tags.size(); j++) {
+        RowData tag = tags.getRow(j, evolved ? 2 : 1);
+        String value = tag.getString(0).toString();
+        if (evolved) {
+          value += "(" + (tag.isNullAt(1) ? "null" : tag.getString(1).toString()) + ")";
+        }
+        tagValues.add(value);
+      }
+      formatted.add(name + ":" + tagValues);
+    }
+    return formatted.toString();
+  }
+
+  private static String formatInts(ArrayData array) {
+    List<Integer> ints = new ArrayList<>();
+    for (int i = 0; i < array.size(); i++) {
+      ints.add(array.getInt(i));
+    }
+    return ints.toString();
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink2.2.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java b/hudi-flink-datasource/hudi-flink2.2.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
new file mode 100644
index 000000000000..0bfb1a8aa668
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink2.2.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.hudi.table.format.cow.utils;
+
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link NestedPositionUtil}.
+ */
+class TestNestedPositionUtil {
+
+  /** An optional top-level row whose last field is an optional list of optional ints. */
+  private static final ParquetGroupField ROW =
+      new ParquetGroupField(RowType.of(new IntType()), 0, 1, false, Collections.emptyList());
+
+  @Test
+  void testRowOffsetsCountOneValuePerRowWhenTheLastFieldIsAnArray() {
+    // two rows of three elements each, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 4, 4, 4, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 1, 1, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertNull(position.getIsNull());
+  }
+
+  @Test
+  void testRowOffsetsWithNullRows() {
+    // a row of three elements, a null row, then the first element of the next row
+    int[] definitionLevels = {4, 4, 4, 0, 4};
+    int[] repetitionLevels = {0, 1, 1, 0, 0};
+
+    RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW, definitionLevels, repetitionLevels);
+
+    assertEquals(3, position.getPositionsCount());
+    assertArrayEquals(new boolean[] {false, true, false}, position.getIsNull());
+  }
+}