Skip to content

[common][core] Preserve identity-mapped iterators safely #9702

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jianguotian wants to merge 6 commits into
base: master
Choose a base branch
from
Open

[common][core] Preserve identity-mapped iterators safely #9702

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d53ff92
[common] Preserve specialized iterators for full identity mappings
Sep 9, 2026
151f38b
[core] Isolate row tracking from reusable column batches
Sep 10, 2026
3c2c993
[core] Avoid covariant array row-tracking failure
Sep 10, 2026
4788f51
[core] Preserve identity iterators without tracking fields
Sep 10, 2026
53b5ad9
[common][core] Strengthen identity row-tracking tests
Sep 10, 2026
c6b3ac0
Merge upstream master into identity-mapping fix
Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions paimon-common/src/main/java/org/apache/paimon/data/columnar/ColumnarRowIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,10 @@ public ColumnarRowIterator copy(ColumnVector[] vectors) {

public ColumnarRowIterator mapping(
@Nullable PartitionInfo partitionInfo, @Nullable int[] indexMapping) {
if (partitionInfo == null && isIdentityMapping(indexMapping, row.batch().getArity())) {
return this;
Comment on lines +118 to +119

@JingsongLi JingsongLi Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Keep row-tracking wrappers out of the reader's reusable batch

For an unpartitioned table with row tracking enabled, reading t$row_tracking can reach this branch with a full identity mapping: FormatReaderMapping.Builder.trimKeyFields() returns an explicit identity array even when the schema mapping is null. DataFileRecordReader then calls assignRowTracking(), which replaces entries in batch.columns in place. Returning the original iterator exposes the Parquet/ORC reader's reusable column array to those mutations, so each reused batch wraps the previous batch's wrappers. Since their isNullAt() always returns false, each metadata getLong() recursively traverses the accumulated wrappers, causing progressively slower reads and eventually StackOverflowError. Previously, createMappedVectors() plus copy() isolated these mutations in a separate column array.

I reproduced this on JDK 8 with a real Parquet file, batch size 1, and the same mapping(...).assignRowTracking(...) sequence: this implementation overflows when checking the row at approximately 20,000 batches, while the identical test with the parent implementation completes all 100,000 batches. The four existing PR tests pass but do not exercise this reuse path.

Please keep row-tracking decoration isolated from the format reader's column array, or retain the copy path when row tracking needs to modify the vectors, and add a regression covering repeated batch reuse.

}

if (partitionInfo != null || indexMapping != null) {
VectorizedColumnBatch vectorizedColumnBatch = row.batch();
ColumnVector[] vectors = vectorizedColumnBatch.columns;
Expand All @@ -129,6 +133,19 @@ public ColumnarRowIterator mapping(
return this;
}

private static boolean isIdentityMapping(@Nullable int[] indexMapping, int arity) {
if (indexMapping == null || indexMapping.length != arity) {
return false;
}

for (int i = 0; i < indexMapping.length; i++) {
if (indexMapping[i] != i) {
return false;
}
}
return true;
}

/**
* Strips a row-tracking wrapper previously installed by {@link #assignRowTracking}, so repeated
* assignment re-wraps the base vector instead of nesting.
Expand Down
71 changes: 71 additions & 0 deletions paimon-common/src/test/java/org/apache/paimon/data/columnar/ColumnarRowIteratorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,15 @@

package org.apache.paimon.data.columnar;

import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.PartitionInfo;
import org.apache.paimon.data.columnar.heap.HeapIntVector;
import org.apache.paimon.data.columnar.heap.HeapLongVector;
import org.apache.paimon.fs.Path;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.LongIterator;

import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -115,4 +120,70 @@ public void testRepeatedAssignRowTrackingDoesNotNest() {
assertThat(tracked.next()).isNotNull();
assertThat(tracked.row.getLong(0)).isEqualTo(300L);
}

@Test
public void testIdentityMappingPreservesSpecializedIterator() {
HeapIntVector firstVector = new HeapIntVector(1);
HeapIntVector secondVector = new HeapIntVector(1);
VectorizedColumnBatch batch =
new VectorizedColumnBatch(new ColumnVector[] {firstVector, secondVector});
batch.setNumRows(1);
ColumnarRowIterator rowIterator = new TestingSpecializedIterator(batch);
rowIterator.reset(0);

assertThat(rowIterator.mapping(null, new int[] {0, 1})).isSameAs(rowIterator);
}

@Test
public void testNonIdentityMappingCopiesIterator() {
HeapIntVector firstVector = new HeapIntVector(1);
HeapIntVector secondVector = new HeapIntVector(1);
VectorizedColumnBatch batch =
new VectorizedColumnBatch(new ColumnVector[] {firstVector, secondVector});
batch.setNumRows(1);
ColumnarRowIterator rowIterator = new TestingSpecializedIterator(batch);
rowIterator.reset(0);

ColumnarRowIterator reordered = rowIterator.mapping(null, new int[] {1, 0});
assertThat(reordered).isNotSameAs(rowIterator);
assertThat(reordered.batch().columns).containsExactly(secondVector, firstVector);

ColumnarRowIterator duplicated = rowIterator.mapping(null, new int[] {0, 0});
assertThat(duplicated).isNotSameAs(rowIterator);
assertThat(duplicated.batch().columns).containsExactly(firstVector, firstVector);

ColumnarRowIterator projected = rowIterator.mapping(null, new int[] {0});
assertThat(projected).isNotSameAs(rowIterator);
assertThat(projected.batch().columns).containsExactly(firstVector);
}

@Test
public void testPartitionMappingCopiesIterator() {
HeapIntVector dataVector = new HeapIntVector(1);
dataVector.setInt(0, 7);
VectorizedColumnBatch batch = new VectorizedColumnBatch(new ColumnVector[] {dataVector});
batch.setNumRows(1);
ColumnarRowIterator rowIterator = new TestingSpecializedIterator(batch);
rowIterator.reset(0);

BinaryRow partition = new BinaryRow(1);
BinaryRowWriter writer = new BinaryRowWriter(partition);
writer.writeInt(0, 42);
writer.complete();
PartitionInfo partitionInfo =
new PartitionInfo(new int[] {1, -1, 0}, RowType.of(DataTypes.INT()), partition);

ColumnarRowIterator mapped = rowIterator.mapping(partitionInfo, new int[] {0, 1});
assertThat(mapped).isNotSameAs(rowIterator);
assertThat(mapped.batch().getArity()).isEqualTo(2);
assertThat(mapped.batch().getInt(0, 0)).isEqualTo(7);
assertThat(mapped.batch().getInt(0, 1)).isEqualTo(42);
}

private static class TestingSpecializedIterator extends ColumnarRowIterator {

private TestingSpecializedIterator(VectorizedColumnBatch batch) {
super(new Path("test"), new ColumnarRow(batch), null);
}
}
}
23 changes: 21 additions & 2 deletions paimon-core/src/main/java/org/apache/paimon/io/DataFileRecordReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.PartitionInfo;
import org.apache.paimon.data.columnar.ColumnVector;
import org.apache.paimon.data.columnar.ColumnarRowIterator;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.fs.Path;
Expand Down Expand Up @@ -178,8 +179,26 @@ private FileRecordIterator<InternalRow> readBatchInternal() throws IOException {
}

if (iterator instanceof ColumnarRowIterator) {
iterator = ((ColumnarRowIterator) iterator).mapping(partitionInfo, indexMapping);
if (rowTrackingEnabled) {
ColumnarRowIterator sourceIterator = (ColumnarRowIterator) iterator;
iterator = sourceIterator.mapping(partitionInfo, indexMapping);
boolean assignRowTracking =
rowTrackingEnabled
&& (systemFields.containsKey(SpecialFields.SEQUENCE_NUMBER.name())
|| (firstRowId != null
&& systemFields.containsKey(
SpecialFields.ROW_ID.name())));
if (assignRowTracking) {
if (iterator == sourceIterator) {
// Copy to a ColumnVector[] because cloning a subtype array preserves its
// runtime type and cannot accept row-tracking wrapper vectors.
ColumnarRowIterator columnarIterator = (ColumnarRowIterator) iterator;
iterator =
columnarIterator.copy(
Arrays.copyOf(
columnarIterator.batch().columns,
columnarIterator.batch().columns.length,
ColumnVector[].class));
}
iterator =
((ColumnarRowIterator) iterator)
.assignRowTracking(firstRowId, maxSequenceNumber, systemFields);
Expand Down
Loading
Loading