diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedColumnReader.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedColumnReader.java
index 5641f8e073ae..8f20be275447 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedColumnReader.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedColumnReader.java
@@ -141,9 +141,7 @@ private Pair<LevelDelegation, WritableColumnVector> readRow(
         boolean hasNull = false;
         for (int i = 0; i < len; i++) {
             for (WritableColumnVector child : finalChildrenVectors) {
-                if (((ElementCountable) child).getLen() - 1 >= i) {
-                    isNull[i] = isNull[i] && child.isNullAt(i);
-                }
+                isNull[i] = isNull[i] && child.isNullAt(i);
             }
             if (isNull[i]) {
                 hasNull = true;
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPositionUtil.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPositionUtil.java
index 473ef332c296..2d73ac07a7bd 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPositionUtil.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPositionUtil.java
@@ -99,10 +99,12 @@ public static CollectionPosition calculateCollectionOffsets(
             } else {
                 // else when definitionLevels[i] < collectionDefinitionLevel - 1, it means the
                 // collection is not defined, no need to increase offset.
-                nullCollectionFlags.add(true);
-                nullValuesCount++;
-                emptyCollectionFlags.add(false);
-                valueCount++;
+                if (readRowField) {
+                    nullCollectionFlags.add(true);
+                    nullValuesCount++;
+                    emptyCollectionFlags.add(false);
+                    valueCount++;
+                }
             }
         }
         long[] offsetsArray = offsets.toArray();
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPrimitiveColumnReader.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPrimitiveColumnReader.java
index dd115698ecdc..0aaa41d5b6c9 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPrimitiveColumnReader.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/NestedPrimitiveColumnReader.java
@@ -107,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(
@@ -200,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++;
                 }
-                valueIndex++;
+                readState.valuesToReadInPage = readState.valuesToReadInPage - 1;
             } while (readValue() && (repetitionLevel != 0));
 
             if (pageRowId == readState.rowId) {
@@ -216,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;
     }
 
@@ -226,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);
     }
 
@@ -289,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);
     }
diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
index d7d16450a50f..e73e52f9d992 100644
--- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
+++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
@@ -61,7 +61,6 @@
 import org.apache.parquet.hadoop.ParquetFileWriter;
 import org.apache.parquet.hadoop.ParquetWriter;
 import org.apache.parquet.hadoop.example.ExampleParquetWriter;
-import org.apache.parquet.hadoop.metadata.CompressionCodecName;
 import org.apache.parquet.hadoop.util.HadoopOutputFile;
 import org.apache.parquet.schema.ConversionPatterns;
 import org.apache.parquet.schema.GroupType;
@@ -840,14 +839,11 @@ private Path createNestedDataByOriginWriter(int rowNum, File tmpDir, int rowGrou
         MessageType schema =
                 ParquetSchemaConverter.convertToParquetMessageType(
                         "paimon-parquet", NESTED_ARRAY_MAP_TYPE);
-        String[] candidates = new String[] {"snappy", "zstd", "gzip"};
-        String compress = candidates[new Random().nextInt(3)];
         try (ParquetWriter<Group> writer =
                 ExampleParquetWriter.builder(
                                 HadoopOutputFile.fromPath(
                                         new org.apache.hadoop.fs.Path(path.toString()), conf))
                         .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
-                        .withCompressionCodec(CompressionCodecName.fromConf(compress))
                         .withConf(new Configuration())
                         .withType(schema)
                         .build()) {