You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/05 17:30:55 UTC
(hudi) 25/40: fix(flink): avoid AIOOBE in NestedColumnReader across the read batch boundary (#19210)
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 <77...@users.noreply.github.com>
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]) {