This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new ee3692cf3142 fix(flink): avoid AIOOBE in NestedColumnReader across the
read batch boundary (#19210)
ee3692cf3142 is described below
commit ee3692cf3142c293b2ec36e520c252936a8c7af3
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
---
.../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 9cf41d1e47fd..4d03c3d9cc82 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
@@ -119,6 +119,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;
@@ -2474,6 +2475,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 7ec70490dc45..60575f148cc4 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;
@@ -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-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 7ec70490dc45..60575f148cc4 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;
@@ -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-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]) {
