[core] fix parquet can not read empty row with first column is array.

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/AbstractHeapVector.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.data.columnar.writable.AbstractWritableVector;
 import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.utils.Preconditions;
 
 import java.nio.ByteOrder;
 import java.util.Arrays;
@@ -54,6 +55,23 @@ public AbstractHeapVector(int len) {
         this.len = len;
     }
 
+    // This will be called only when inner vectors don't have data.
+    public AbstractHeapVector(int len, boolean[] isNull) {
+        Preconditions.checkArgument(
+                len == isNull.length, "len should be equal to isNull's length.");
+
+        for (boolean element : isNull) {
+            if (!element) {
+                throw new UnsupportedOperationException(
+                        "This constructor can only be called when the vector is all null.");
+            }
+        }
+
+        this.len = len;
+        this.isNull = isNull;
+        this.noNulls = false;
+    }
+
     /**
      * Resets the column to default state. - fills the isNull array with false. - sets noNulls to
      * true.

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapBooleanVector.java

@@ -34,6 +34,10 @@ public HeapBooleanVector(int len) {
         vector = new boolean[len];
     }
 
+    public HeapBooleanVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public HeapIntVector reserveDictionaryIds(int capacity) {
         throw new RuntimeException("HeapBooleanVector has no dictionary.");

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapByteVector.java

@@ -39,6 +39,10 @@ public HeapByteVector(int len) {
         vector = new byte[len];
     }
 
+    public HeapByteVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public byte getByte(int i) {
         if (dictionary == null) {

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapBytesVector.java

@@ -66,6 +66,10 @@ public HeapBytesVector(int size) {
         length = new int[size];
     }
 
+    public HeapBytesVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public void reset() {
         super.reset();

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapDoubleVector.java

@@ -45,6 +45,10 @@ public HeapDoubleVector(int len) {
         vector = new double[len];
     }
 
+    public HeapDoubleVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public double getDouble(int i) {
         if (dictionary == null) {

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapFloatVector.java

@@ -44,6 +44,10 @@ public HeapFloatVector(int len) {
         vector = new float[len];
     }
 
+    public HeapFloatVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public float getFloat(int i) {
         if (dictionary == null) {

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapIntVector.java

@@ -39,6 +39,10 @@ public HeapIntVector(int len) {
         vector = new int[len];
     }
 
+    public HeapIntVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public int getInt(int i) {
         if (dictionary == null) {

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapLongVector.java

@@ -39,6 +39,10 @@ public HeapLongVector(int len) {
         vector = new long[len];
     }
 
+    public HeapLongVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public long getLong(int i) {
         if (dictionary == null) {

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapShortVector.java

@@ -39,6 +39,10 @@ public HeapShortVector(int len) {
         vector = new short[len];
     }
 
+    public HeapShortVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public short getShort(int i) {
         if (dictionary == null) {

paimon-common/src/main/java/org/apache/paimon/data/columnar/heap/HeapTimestampVector.java

@@ -28,15 +28,19 @@ public class HeapTimestampVector extends AbstractHeapVector implements WritableT
 
     private static final long serialVersionUID = 1L;
 
-    private final long[] milliseconds;
-    private final int[] nanoOfMilliseconds;
+    private long[] milliseconds;
+    private int[] nanoOfMilliseconds;
 
     public HeapTimestampVector(int len) {
         super(len);
         this.milliseconds = new long[len];
         this.nanoOfMilliseconds = new int[len];
     }
 
+    public HeapTimestampVector(int len, boolean[] isNull) {
+        super(len, isNull);
+    }
+
     @Override
     public Timestamp getTimestamp(int i, int precision) {
         if (dictionary == null) {

paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPositionUtil.java

@@ -96,9 +96,16 @@ public static CollectionPosition calculateCollectionOffsets(
 
                 offsets.add(offset);
                 valueCount++;
+            } else {
+                // else when definitionLevels[i] < collectionDefinitionLevel - 1, it means the
+                // collection is not defined, no need to increase offset.
+                if (readRowField) {
+                    nullCollectionFlags.add(true);
+                    nullValuesCount++;
+                    emptyCollectionFlags.add(false);
+                    valueCount++;
+                }
             }
-            // else when definitionLevels[i] < collectionDefinitionLevel - 1, it means the
-            // collection is not defined, just ignore it
         }
         long[] offsetsArray = offsets.toArray();
         long[] length = calculateLengthByOffsets(emptyCollectionFlags.toArray(), offsetsArray);

paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPrimitiveColumnReader.java

@@ -57,6 +57,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.apache.parquet.column.ValuesType.DEFINITION_LEVEL;
@@ -106,6 +107,11 @@ public class NestedPrimitiveColumnReader implements ColumnReader<WritableColumnV
 
     private boolean isFirstRow = true;
 
+    // When reading array, we need to read the next value's repetition level to know whether it's a
+    // new row. This is a flag to tell whether we need to cut the repetition level when getting
+    // LevelDelegation.
+    private boolean cutLevel = false;
+
     private final LastValueContainer lastValue = new LastValueContainer();
 
     public NestedPrimitiveColumnReader(
@@ -168,7 +174,10 @@ public WritableColumnVector readAndNewVector(int readNumber, WritableColumnVecto
 
         int valueIndex = collectDataFromParquetPage(readNumber, valueList);
 
-        return fillColumnVector(valueIndex, valueList);
+        if (!valueList.isEmpty()) {
+            return fillColumnVector(valueIndex, valueList);
+        }
+        return fillColumnVectorWithNone(valueIndex);
     }
 
     private int collectDataFromParquetPage(int total, List<Object> valueList) throws IOException {
@@ -196,11 +205,13 @@ private int collectDataFromParquetPage(int total, List<Object> valueList) throws
             boolean needFilterSkip = pageRowId < rangeStart;
 
             do {
-
                 if (!lastValue.shouldSkip && !needFilterSkip) {
                     valueList.add(lastValue.value);
                     valueIndex++;
+                } else if (readRowField) {
+                    valueIndex++;
                 }
+                readState.valuesToReadInPage = readState.valuesToReadInPage - 1;
             } while (readValue() && (repetitionLevel != 0));
 
             if (pageRowId == readState.rowId) {
@@ -212,6 +223,12 @@ private int collectDataFromParquetPage(int total, List<Object> valueList) throws
             }
         }
 
+        // When the values to read in page > 0 and row to read in batch == 0, it means the
+        // repetition level contains the next value's, so need to set the cutLevel flag to true.
+        if (readState.valuesToReadInPage > 0 && readState.rowsToReadInBatch == 0) {
+            cutLevel = true;
+        }
+
         return valueIndex;
     }
 
@@ -222,6 +239,11 @@ public LevelDelegation getLevelDelegation() {
         definitionLevelList.clear();
         repetitionLevelList.add(repetitionLevel);
         definitionLevelList.add(definitionLevel);
+        if (cutLevel) {
+            repetition = Arrays.copyOf(repetition, repetition.length - 1);
+            definition = Arrays.copyOf(definition, definition.length - 1);
+            cutLevel = false;
+        }
         return new LevelDelegation(repetition, definition);
     }
 
@@ -285,7 +307,6 @@ private void readAndSaveRepetitionAndDefinitionLevels() {
         // get the values of repetition and definitionLevel
         repetitionLevel = repetitionLevelColumn.nextInt();
         definitionLevel = definitionLevelColumn.nextInt();
-        readState.valuesToReadInPage = readState.valuesToReadInPage - 1;
         repetitionLevelList.add(repetitionLevel);
         definitionLevelList.add(definitionLevel);
     }
@@ -549,6 +570,53 @@ private WritableColumnVector fillColumnVector(int total, List valueList) {
         }
     }
 
+    private WritableColumnVector fillColumnVectorWithNone(int total) {
+        boolean[] isNull = new boolean[total];
+        Arrays.fill(isNull, true);
+        switch (dataType.getTypeRoot()) {
+            case CHAR:
+            case VARCHAR:
+            case BINARY:
+            case VARBINARY:
+                return new HeapBytesVector(total, isNull);
+            case BOOLEAN:
+                return new HeapBooleanVector(total, isNull);
+            case TINYINT:
+                return new HeapByteVector(total, isNull);
+            case SMALLINT:
+                return new HeapShortVector(total, isNull);
+            case INTEGER:
+            case DATE:
+            case TIME_WITHOUT_TIME_ZONE:
+                return new HeapIntVector(total, isNull);
+            case FLOAT:
+                return new HeapFloatVector(total, isNull);
+            case BIGINT:
+                return new HeapLongVector(total, isNull);
+            case DOUBLE:
+                return new HeapDoubleVector(total, isNull);
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                return new HeapTimestampVector(total, isNull);
+            case DECIMAL:
+                PrimitiveType.PrimitiveTypeName primitiveTypeName =
+                        descriptor.getPrimitiveType().getPrimitiveTypeName();
+                switch (primitiveTypeName) {
+                    case INT32:
+                        HeapIntVector phiv = new HeapIntVector(total, isNull);
+                        return new ParquetDecimalVector(phiv, total);
+                    case INT64:
+                        HeapLongVector phlv = new HeapLongVector(total, isNull);
+                        return new ParquetDecimalVector(phlv, total);
+                    default:
+                        HeapBytesVector phbv = new HeapBytesVector(total, isNull);
+                        return new ParquetDecimalVector(phbv, total);
+                }
+            default:
+                throw new RuntimeException("Unsupported type in the list: " + type);
+        }
+    }
+
     private static HeapBytesVector getHeapBytesVector(int total, List valueList) {
         HeapBytesVector phbv = new HeapBytesVector(total);
         for (int i = 0; i < valueList.size(); i++) {

paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java

@@ -176,7 +176,11 @@ public class ParquetReadWriteTest {
                                                             new ArrayType(true, new IntType())))
                                             .field("c", new IntType())
                                             .build()),
-                            new IntType()));
+                            new IntType()),
+                    RowType.of(
+                            new ArrayType(RowType.of(new VarCharType(255))),
+                            RowType.of(new IntType()),
+                            new VarCharType(255)));
 
     @TempDir public File folder;
 
@@ -822,7 +826,8 @@ null, new GenericMap(mp1), new GenericMap(mp2)
                                                                 }),
                                                         i)
                                             }),
-                                    i)));
+                                    i),
+                            null));
         }
         return rows;
     }
@@ -1011,6 +1016,10 @@ private void compareNestedRow(
                     origin.getRow(5, 2).getArray(0).getRow(0, 2).getInt(1),
                     result.getRow(5, 2).getArray(0).getRow(0, 2).getInt(1));
             Assertions.assertEquals(origin.getRow(5, 2).getInt(1), result.getRow(5, 2).getInt(1));
+            Assertions.assertTrue(result.isNullAt(6));
+            Assertions.assertTrue(result.getRow(6, 2).isNullAt(0));
+            Assertions.assertTrue(result.getRow(6, 2).isNullAt(1));
+            Assertions.assertTrue(result.getRow(6, 2).isNullAt(2));
         }
         assertThat(iterator.hasNext()).isFalse();
         iterator.close();