This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 82c8fd9249da3ee93eb1c7345efc5f3235c83975 Author: ericyuan915 <[email protected]> AuthorDate: Tue Jul 7 04:18:50 2026 -0700 fix(flink): avoid AIOOBE in NestedColumnReader across the read batch boundary (#19210) NestedColumnReader#readRow collapses a present row whose children are all null into a NULL row. That loop iterated up to rowPosition.getPositionsCount(), which on a full, non-final vectorized batch is one larger than the materialized child vectors: the Dremel level stream carries a one-record lookahead (NestedPrimitiveColumnReader#readAndNewVector reads one value past the batch, and #getLevelDelegation keeps that trailing level for the next batch). Reading a nested ROW column from a C [...] Clamp the collapse loop to the shortest vector it indexes (the row vector and every child), unwrapping ParquetDecimalVector (a non-AbstractHeapVector DECIMAL child) via a new vectorLength helper; the phantom trailing position is never surfaced downstream (ParquetColumnarRowSplitReader caps the batch at num). Applied identically to all five flink modules (1.18.x/1.19.x/1.20.x/2.0.x/2.1.x). Adds integration test ITTestHoodieDataSource#testParquetNestedRowExceedingReadBatch covering both [...] closes #19208 (cherry picked from commit ee3692cf3142c293b2ec36e520c252936a8c7af3) --- .../apache/hudi/table/ITTestHoodieDataSource.java | 119 +++++++++++++++++++++ .../cow/vector/reader/NestedColumnReader.java | 33 +++++- .../cow/vector/reader/NestedColumnReader.java | 33 +++++- .../cow/vector/reader/NestedColumnReader.java | 33 +++++- .../cow/vector/reader/NestedColumnReader.java | 33 +++++- .../cow/vector/reader/NestedColumnReader.java | 33 +++++- 6 files changed, 279 insertions(+), 5 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java index 148e14f30b2f..c5e38bde4883 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java @@ -120,6 +120,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertLinesMatch; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -2374,6 +2375,124 @@ public class ITTestHoodieDataSource { assertRowsEqualsUnordered(expected, result); } + @Test + void testParquetNestedRowExceedingReadBatch() { + // Regression for NestedColumnReader#readRow throwing ArrayIndexOutOfBoundsException when a COW + // base file holds more rows than the 2048-row vectorized read batch + // (RecordIterators.DEFAULT_BATCH_SIZE) and a nested ROW column is read. On a full, non-final + // batch the Dremel level stream carries a one-record lookahead, so NestedPositionUtil + // #calculateRowOffsets returns positionsCount = batchSize + 1 = 2049 while the materialized + // child column vectors are sized to their value count = 2048. The Hudi-specific null-row-collapse + // loop iterates to positionsCount and reads child.isNullAt(2048), one past a length-2048 vector. + // + // Two conditions are both required to surface it, and drove this schema and data: + // 1. The bad index is only reached through AbstractHeapVector#isNullAt, which short-circuits to + // false without touching isNull[] when the vector has no nulls. So a child vector must + // actually carry a null. Odd-id rows therefore store a present ROW with all-null children + // (row(null, ...)); the row stays present (its own isNullAt(2048) short-circuits) but the + // child leaf vectors get noNulls=false and overrun at the phantom index. Half the rows are + // null-children so the first full batch is guaranteed to contain them regardless of how + // bulk_insert orders keys. + // 2. The nullable leaves must be *direct* children of the collapsed row. A sub-row child would + // be renewed to positionsCount (length 2049) and not overrun, so the two nested rows are + // top-level columns: f_scalar row(f0 int, f1 varchar(10)) covers heap-vector children, and + // f_dec row(d decimal(10, 2)) covers a decimal child, whose ParquetDecimalVector is not an + // AbstractHeapVector and must be unwrapped by NestedColumnReader#vectorLength. + // See ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes for the collapse behaviour. + TableEnvironment tableEnv = batchTableEnv; + + // More rows than one 2048-row read batch, so the first batch is full and non-final -- that is + // what makes the level stream carry the trailing lookahead that overshoots the vectors. The + // rows are generated by cross joining two small VALUES lists rather than a single 2000+-row + // VALUES literal: Calcite plans the latter pathologically slowly (minutes to hours), while two + // ~50-element lists plan instantly and the row count is simply their product. + final int outer = 43; + final int inner = 50; + final int numRows = outer * inner; // 2150 > 2048 + + String hoodieTableDDL = sql("t1") + .field("f_int int") + .field("f_scalar row(f0 int, f1 varchar(10))") + .field("f_dec row(d decimal(10, 2))") + .pkField("f_int") + .noPartition() + .option(FlinkOptions.PATH, tempFile.getAbsolutePath()) + .option(FlinkOptions.OPERATION, "bulk_insert") + // Single write task => all rows land in one base file, so one read split crosses the + // 2048-row batch boundary. + .option(FlinkOptions.WRITE_TASKS, 1) + .end(); + tableEnv.executeSql(hoodieTableDDL); + + // id = blk * inner + pos is unique over blk in [0, outer), pos in [0, inner) => 0 .. numRows-1. + // Both nested rows stay present; even ids get populated leaves, odd ids get all-null leaves + // (which the reader collapses back to a NULL row). Each ROW is cast to its named type so the + // query output type matches the sink column exactly. + String insert = "insert into t1 select\n" + + " g.id,\n" + + " cast(row(\n" + + " case when mod(g.id, 2) = 0 then g.id else cast(null as int) end,\n" + + " case when mod(g.id, 2) = 0 then concat('v', cast(g.id as varchar)) else cast(null as varchar(10)) end\n" + + " ) as row<f0 int, f1 varchar(10)>),\n" + + " cast(row(\n" + + " case when mod(g.id, 2) = 0 then cast(g.id as decimal(10, 2)) else cast(null as decimal(10, 2)) end\n" + + " ) as row<d decimal(10, 2)>)\n" + + "from (\n" + + " select blk.b * " + inner + " + pos.p as id\n" + + " from (values " + valuesList(outer) + ") as blk(b)\n" + + " cross join (values " + valuesList(inner) + ") as pos(p)\n" + + ") g"; + execInsertSql(tableEnv, insert); + + List<Row> result = CollectionUtil.iterableToList( + () -> tableEnv.sqlQuery("select * from t1").execute().collect()); + + // The read completes (no AIOOBE across the batch boundary) and every row is returned. Without + // the fix the vectorized read throws while materializing the first full batch, so this fails. + assertEquals(numRows, result.size()); + + // bulk_insert does not preserve order, so index by pk. + Map<Integer, Row> byId = new HashMap<>(); + for (Row r : result) { + byId.put((Integer) r.getField(0), r); + } + // Populated rows (even id) round-trip both nested rows -- one from the first (full) batch and + // one with a large id past the boundary. + assertPopulatedRow(byId.get(0), 0); + assertPopulatedRow(byId.get(numRows - 2), numRows - 2); + // All-null-children rows (odd id) collapse both nested rows back to NULL, including a large id. + assertCollapsedRow(byId.get(1)); + assertCollapsedRow(byId.get(numRows - 1)); + } + + /** Builds the VALUES row list {@code (0), (1), ..., (n-1)} for the generator cross join. */ + private static String valuesList(int n) { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < n; i++) { + if (i > 0) { + sb.append(", "); + } + sb.append('(').append(i).append(')'); + } + return sb.toString(); + } + + /** Asserts the row keyed by an even {@code id} round-trips its populated nested rows. */ + private static void assertPopulatedRow(Row row, int id) { + assertNotNull(row, "row with pk " + id + " was not read back"); + Row scalar = (Row) row.getField(1); + assertEquals(id, scalar.getField(0)); + assertEquals("v" + id, scalar.getField(1)); + assertNotNull(((Row) row.getField(2)).getField(0)); // decimal leaf present, not null + } + + /** Asserts the row keyed by an odd {@code id} had both all-null nested rows collapsed to NULL. */ + private static void assertCollapsedRow(Row row) { + assertNotNull(row, "expected an odd-id row to be read back"); + assertNull(row.getField(1)); // f_scalar collapsed to null + assertNull(row.getField(2)); // f_dec collapsed to null + } + @ParameterizedTest @ValueSource(strings = {"insert", "upsert", "bulk_insert"}) void testParquetNullChildColumnsRowTypes(String operation) { diff --git a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java index 2e0bd744cab1..ac94292c3315 100644 --- a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java +++ b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java @@ -22,6 +22,7 @@ import org.apache.hudi.table.format.cow.utils.NestedPositionUtil; import org.apache.hudi.table.format.cow.vector.HeapArrayVector; import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector; import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector; +import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector; import org.apache.hudi.table.format.cow.vector.position.CollectionPosition; import org.apache.hudi.table.format.cow.vector.position.LevelDelegation; import org.apache.hudi.table.format.cow.vector.position.RowPosition; @@ -156,7 +157,21 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { // legacy RowColumnReader (deleted alongside the Dremel rewire) and existing Hudi tables rely // on it. Diverges from Flink 2.1, which would surface it as Row(null, null). Pinned by the // integration test ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes. - int rowCount = rowPosition.getPositionsCount(); + // positionsCount comes from the Dremel definition/repetition level stream + // (NestedPositionUtil#calculateRowOffsets). On a full, non-final batch that stream carries a + // one-record lookahead (NestedPrimitiveColumnReader#readAndNewVector reads one value past the + // batch in its do/while, and #getLevelDelegation keeps that trailing level for the next batch), + // so positionsCount can be one larger than the materialized vector lengths. When inside==true + // the row vector is renewed to positionsCount but its children are sized to their value count; + // when inside==false the row vector keeps its batch capacity. Either way, iterating all the way + // to positionsCount can read one element past a shorter vector and throw + // ArrayIndexOutOfBoundsException. Clamp to the shortest vector this loop indexes -- the phantom + // trailing position is never surfaced downstream (ParquetColumnarRowSplitReader caps the batch + // at num). + int rowCount = Math.min(rowPosition.getPositionsCount(), heapRowVector.getLen()); + for (WritableColumnVector child : finalChildrenVectors) { + rowCount = Math.min(rowCount, vectorLength(child)); + } for (int j = 0; j < rowCount; j++) { if (heapRowVector.isNullAt(j)) { continue; @@ -272,6 +287,22 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { return Tuple2.of(reader.getLevelDelegation(), writableColumnVector); } + /** + * The length of the {@code isNull}-backed storage that {@code vector} (a row child) is indexed + * against by the null-collapse loop in {@link #readRow}. Every row child is an {@link + * AbstractHeapVector} (nested rows/arrays/maps and all non-decimal primitives) or a {@link + * ParquetDecimalVector} wrapping one (DECIMAL leaves; see {@code + * NestedPrimitiveColumnReader#fillColumnVector}); unwrapping the latter yields an {@code + * AbstractHeapVector} in all cases. + */ + private static int vectorLength(ColumnVector vector) { + ColumnVector storage = + vector instanceof ParquetDecimalVector + ? ((ParquetDecimalVector) vector).getVector() + : vector; + return ((AbstractHeapVector) storage).getLen(); + } + private static void setFieldNullFlag(boolean[] nullFlags, AbstractHeapVector vector) { for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) { if (nullFlags[index]) { 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 2e0bd744cab1..ac94292c3315 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 @@ -22,6 +22,7 @@ import org.apache.hudi.table.format.cow.utils.NestedPositionUtil; import org.apache.hudi.table.format.cow.vector.HeapArrayVector; import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector; import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector; +import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector; import org.apache.hudi.table.format.cow.vector.position.CollectionPosition; import org.apache.hudi.table.format.cow.vector.position.LevelDelegation; import org.apache.hudi.table.format.cow.vector.position.RowPosition; @@ -156,7 +157,21 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { // legacy RowColumnReader (deleted alongside the Dremel rewire) and existing Hudi tables rely // on it. Diverges from Flink 2.1, which would surface it as Row(null, null). Pinned by the // integration test ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes. - int rowCount = rowPosition.getPositionsCount(); + // positionsCount comes from the Dremel definition/repetition level stream + // (NestedPositionUtil#calculateRowOffsets). On a full, non-final batch that stream carries a + // one-record lookahead (NestedPrimitiveColumnReader#readAndNewVector reads one value past the + // batch in its do/while, and #getLevelDelegation keeps that trailing level for the next batch), + // so positionsCount can be one larger than the materialized vector lengths. When inside==true + // the row vector is renewed to positionsCount but its children are sized to their value count; + // when inside==false the row vector keeps its batch capacity. Either way, iterating all the way + // to positionsCount can read one element past a shorter vector and throw + // ArrayIndexOutOfBoundsException. Clamp to the shortest vector this loop indexes -- the phantom + // trailing position is never surfaced downstream (ParquetColumnarRowSplitReader caps the batch + // at num). + int rowCount = Math.min(rowPosition.getPositionsCount(), heapRowVector.getLen()); + for (WritableColumnVector child : finalChildrenVectors) { + rowCount = Math.min(rowCount, vectorLength(child)); + } for (int j = 0; j < rowCount; j++) { if (heapRowVector.isNullAt(j)) { continue; @@ -272,6 +287,22 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { return Tuple2.of(reader.getLevelDelegation(), writableColumnVector); } + /** + * The length of the {@code isNull}-backed storage that {@code vector} (a row child) is indexed + * against by the null-collapse loop in {@link #readRow}. Every row child is an {@link + * AbstractHeapVector} (nested rows/arrays/maps and all non-decimal primitives) or a {@link + * ParquetDecimalVector} wrapping one (DECIMAL leaves; see {@code + * NestedPrimitiveColumnReader#fillColumnVector}); unwrapping the latter yields an {@code + * AbstractHeapVector} in all cases. + */ + private static int vectorLength(ColumnVector vector) { + ColumnVector storage = + vector instanceof ParquetDecimalVector + ? ((ParquetDecimalVector) vector).getVector() + : vector; + return ((AbstractHeapVector) storage).getLen(); + } + private static void setFieldNullFlag(boolean[] nullFlags, AbstractHeapVector vector) { for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) { if (nullFlags[index]) { 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 7ec70490dc45..60575f148cc4 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 @@ -22,6 +22,7 @@ import org.apache.hudi.table.format.cow.utils.NestedPositionUtil; import org.apache.hudi.table.format.cow.vector.HeapArrayVector; import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector; import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector; +import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector; import org.apache.hudi.table.format.cow.vector.position.CollectionPosition; import org.apache.hudi.table.format.cow.vector.position.LevelDelegation; import org.apache.hudi.table.format.cow.vector.position.RowPosition; @@ -155,7 +156,21 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { // legacy RowColumnReader (deleted alongside the Dremel rewire) and existing Hudi tables rely // on it. Diverges from Flink 2.1, which would surface it as Row(null, null). Pinned by the // integration test ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes. - int rowCount = rowPosition.getPositionsCount(); + // positionsCount comes from the Dremel definition/repetition level stream + // (NestedPositionUtil#calculateRowOffsets). On a full, non-final batch that stream carries a + // one-record lookahead (NestedPrimitiveColumnReader#readAndNewVector reads one value past the + // batch in its do/while, and #getLevelDelegation keeps that trailing level for the next batch), + // so positionsCount can be one larger than the materialized vector lengths. When inside==true + // the row vector is renewed to positionsCount but its children are sized to their value count; + // when inside==false the row vector keeps its batch capacity. Either way, iterating all the way + // to positionsCount can read one element past a shorter vector and throw + // ArrayIndexOutOfBoundsException. Clamp to the shortest vector this loop indexes -- the phantom + // trailing position is never surfaced downstream (ParquetColumnarRowSplitReader caps the batch + // at num). + int rowCount = Math.min(rowPosition.getPositionsCount(), heapRowVector.getLen()); + for (WritableColumnVector child : finalChildrenVectors) { + rowCount = Math.min(rowCount, vectorLength(child)); + } for (int j = 0; j < rowCount; j++) { if (heapRowVector.isNullAt(j)) { continue; @@ -271,6 +286,22 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { return Tuple2.of(reader.getLevelDelegation(), writableColumnVector); } + /** + * The length of the {@code isNull}-backed storage that {@code vector} (a row child) is indexed + * against by the null-collapse loop in {@link #readRow}. Every row child is an {@link + * AbstractHeapVector} (nested rows/arrays/maps and all non-decimal primitives) or a {@link + * ParquetDecimalVector} wrapping one (DECIMAL leaves; see {@code + * NestedPrimitiveColumnReader#fillColumnVector}); unwrapping the latter yields an {@code + * AbstractHeapVector} in all cases. + */ + private static int vectorLength(ColumnVector vector) { + ColumnVector storage = + vector instanceof ParquetDecimalVector + ? ((ParquetDecimalVector) vector).getVector() + : vector; + return ((AbstractHeapVector) storage).getLen(); + } + private static void setFieldNullFlag(boolean[] nullFlags, AbstractHeapVector vector) { for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) { if (nullFlags[index]) { 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 7ec70490dc45..60575f148cc4 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 @@ -22,6 +22,7 @@ import org.apache.hudi.table.format.cow.utils.NestedPositionUtil; import org.apache.hudi.table.format.cow.vector.HeapArrayVector; import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector; import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector; +import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector; import org.apache.hudi.table.format.cow.vector.position.CollectionPosition; import org.apache.hudi.table.format.cow.vector.position.LevelDelegation; import org.apache.hudi.table.format.cow.vector.position.RowPosition; @@ -155,7 +156,21 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { // legacy RowColumnReader (deleted alongside the Dremel rewire) and existing Hudi tables rely // on it. Diverges from Flink 2.1, which would surface it as Row(null, null). Pinned by the // integration test ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes. - int rowCount = rowPosition.getPositionsCount(); + // positionsCount comes from the Dremel definition/repetition level stream + // (NestedPositionUtil#calculateRowOffsets). On a full, non-final batch that stream carries a + // one-record lookahead (NestedPrimitiveColumnReader#readAndNewVector reads one value past the + // batch in its do/while, and #getLevelDelegation keeps that trailing level for the next batch), + // so positionsCount can be one larger than the materialized vector lengths. When inside==true + // the row vector is renewed to positionsCount but its children are sized to their value count; + // when inside==false the row vector keeps its batch capacity. Either way, iterating all the way + // to positionsCount can read one element past a shorter vector and throw + // ArrayIndexOutOfBoundsException. Clamp to the shortest vector this loop indexes -- the phantom + // trailing position is never surfaced downstream (ParquetColumnarRowSplitReader caps the batch + // at num). + int rowCount = Math.min(rowPosition.getPositionsCount(), heapRowVector.getLen()); + for (WritableColumnVector child : finalChildrenVectors) { + rowCount = Math.min(rowCount, vectorLength(child)); + } for (int j = 0; j < rowCount; j++) { if (heapRowVector.isNullAt(j)) { continue; @@ -271,6 +286,22 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { return Tuple2.of(reader.getLevelDelegation(), writableColumnVector); } + /** + * The length of the {@code isNull}-backed storage that {@code vector} (a row child) is indexed + * against by the null-collapse loop in {@link #readRow}. Every row child is an {@link + * AbstractHeapVector} (nested rows/arrays/maps and all non-decimal primitives) or a {@link + * ParquetDecimalVector} wrapping one (DECIMAL leaves; see {@code + * NestedPrimitiveColumnReader#fillColumnVector}); unwrapping the latter yields an {@code + * AbstractHeapVector} in all cases. + */ + private static int vectorLength(ColumnVector vector) { + ColumnVector storage = + vector instanceof ParquetDecimalVector + ? ((ParquetDecimalVector) vector).getVector() + : vector; + return ((AbstractHeapVector) storage).getLen(); + } + private static void setFieldNullFlag(boolean[] nullFlags, AbstractHeapVector vector) { for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) { if (nullFlags[index]) { 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 00abceeee4fa..7be03dacc218 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 @@ -22,6 +22,7 @@ import org.apache.hudi.table.format.cow.utils.NestedPositionUtil; import org.apache.hudi.table.format.cow.vector.HeapArrayVector; import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector; import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector; +import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector; import org.apache.hudi.table.format.cow.vector.position.CollectionPosition; import org.apache.hudi.table.format.cow.vector.position.LevelDelegation; import org.apache.hudi.table.format.cow.vector.position.RowPosition; @@ -158,7 +159,21 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { // legacy RowColumnReader (deleted alongside the Dremel rewire) and existing Hudi tables rely // on it. Diverges from Flink 2.1, which would surface it as Row(null, null). Pinned by the // integration test ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes. - int rowCount = rowPosition.getPositionsCount(); + // positionsCount comes from the Dremel definition/repetition level stream + // (NestedPositionUtil#calculateRowOffsets). On a full, non-final batch that stream carries a + // one-record lookahead (NestedPrimitiveColumnReader#readAndNewVector reads one value past the + // batch in its do/while, and #getLevelDelegation keeps that trailing level for the next batch), + // so positionsCount can be one larger than the materialized vector lengths. When inside==true + // the row vector is renewed to positionsCount but its children are sized to their value count; + // when inside==false the row vector keeps its batch capacity. Either way, iterating all the way + // to positionsCount can read one element past a shorter vector and throw + // ArrayIndexOutOfBoundsException. Clamp to the shortest vector this loop indexes -- the phantom + // trailing position is never surfaced downstream (ParquetColumnarRowSplitReader caps the batch + // at num). + int rowCount = Math.min(rowPosition.getPositionsCount(), heapRowVector.getLen()); + for (WritableColumnVector child : finalChildrenVectors) { + rowCount = Math.min(rowCount, vectorLength(child)); + } for (int j = 0; j < rowCount; j++) { if (heapRowVector.isNullAt(j)) { continue; @@ -274,6 +289,22 @@ public class NestedColumnReader implements ColumnReader<WritableColumnVector> { return Tuple2.of(reader.getLevelDelegation(), writableColumnVector); } + /** + * The length of the {@code isNull}-backed storage that {@code vector} (a row child) is indexed + * against by the null-collapse loop in {@link #readRow}. Every row child is an {@link + * AbstractHeapVector} (nested rows/arrays/maps and all non-decimal primitives) or a {@link + * ParquetDecimalVector} wrapping one (DECIMAL leaves; see {@code + * NestedPrimitiveColumnReader#fillColumnVector}); unwrapping the latter yields an {@code + * AbstractHeapVector} in all cases. + */ + private static int vectorLength(ColumnVector vector) { + ColumnVector storage = + vector instanceof ParquetDecimalVector + ? ((ParquetDecimalVector) vector).getVector() + : vector; + return ((AbstractHeapVector) storage).getLen(); + } + private static void setFieldNullFlag(boolean[] nullFlags, AbstractHeapVector vector) { for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) { if (nullFlags[index]) {
