[FLINK-36709][parquet] fix parquet can not read row with last column is array. (apache#25651)

Author: Stephen0421

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/NestedPositionUtil.java

@@ -50,12 +50,10 @@ public static RowPosition calculateRowOffsets(
         int nullValuesCount = 0;
         BooleanArrayList nullRowFlags = new BooleanArrayList(0);
         for (int i = 0; i < fieldDefinitionLevels.length; i++) {
+            // If a row's last field is an array, the repetition levels for the array's items will
+            // be larger than the parent row's repetition level, so we need to skip those values.
             if (fieldRepetitionLevels[i] > rowRepetitionLevel) {
-                throw new IllegalStateException(
-                        format(
-                                "In parquet's row type field repetition level should not larger than row's repetition level. "
-                                        + "Row repetition level is %s, row field repetition level is %s.",
-                                rowRepetitionLevel, fieldRepetitionLevels[i]));
+                continue;
             }
 
             if (fieldDefinitionLevels[i] >= rowDefinitionLevel) {

flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java

@@ -151,6 +151,7 @@ class ParquetColumnarRowSplitReaderTest {
                                     Collections.singletonList(
                                             new RowType.RowField("a", new IntType())))),
                     RowType.of(
+                            new IntType(),
                             new ArrayType(
                                     true,
                                     new RowType(
@@ -161,8 +162,7 @@ class ParquetColumnarRowSplitReaderTest {
                                                                     true,
                                                                     new ArrayType(
                                                                             true, new IntType()))),
-                                                    new RowType.RowField("c", new IntType())))),
-                            new IntType()));
+                                                    new RowType.RowField("c", new IntType()))))));
 
     @SuppressWarnings("unchecked")
     private static final DataFormatConverters.DataFormatConverter<RowData, Row>
@@ -829,15 +829,15 @@ private List<Row> prepareNestedData(int rowNum) {
                             new Map[] {null, mp1, mp2},
                             new Row[] {Row.of(i), Row.of(i + 1), null},
                             Row.of(
+                                    i,
                                     new Row[] {
                                         Row.of(
                                                 new Integer[][] {
                                                     {i, i + 1, null}, {i, i + 2, null}, {}, null
                                                 },
                                                 i),
                                         null
-                                    },
-                                    i)));
+                                    })));
         }
         return rows;
     }
@@ -888,15 +888,15 @@ private Path createNestedDataByOriginWriter(int rowNum, File tmpDir, int rowGrou
                 row2.add(0, i + 1);
                 f4.addGroup(0);
 
-                // add ROW<`f0` ARRAY<ROW<`b` ARRAY<ARRAY<INT>>, `c` INT>>, `f1` INT>>
+                // add ROW<`f0` INT , `f1` INTARRAY<ROW<`b` ARRAY<ARRAY<INT>>, `c` INT>>>>
                 Group f5 = row.addGroup("f5");
-                Group arrayRow = f5.addGroup(0);
+                f5.add(0, i);
+                Group arrayRow = f5.addGroup(1);
                 Group insideRow = arrayRow.addGroup(0).addGroup(0);
                 Group insideArray = insideRow.addGroup(0);
                 createParquetDoubleNestedArray(insideArray, i);
                 insideRow.add(1, i);
                 arrayRow.addGroup(0);
-                f5.add(1, i);
                 writer.write(row);
             }
         } catch (Exception e) {