This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ec922f63aa [format] Give each row-format reader its own projected-row
wrapper (#9561)
ec922f63aa is described below
commit ec922f63aae33ce8bd7cd57aa2b78dda239ea222
Author: YangJie <[email protected]>
AuthorDate: Thu Sep 3 01:31:35 2026 -0400
[format] Give each row-format reader its own projected-row wrapper (#9561)
---
.../apache/paimon/format/row/RowFileFormat.java | 5 +--
.../paimon/format/row/RowFormatReaderFactory.java | 18 +++++---
.../paimon/format/row/RowFormatReadWriteTest.java | 52 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 10 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFormat.java
index 8683dd4fde..26ad03369b 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFormat.java
@@ -25,7 +25,6 @@ import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.NestedProjectedRow;
import javax.annotation.Nullable;
@@ -51,9 +50,7 @@ public class RowFileFormat extends FileFormat {
RowType dataSchemaRowType,
RowType projectedRowType,
@Nullable List<Predicate> filters) {
- NestedProjectedRow projection =
- NestedProjectedRow.create(dataSchemaRowType, projectedRowType);
- return new RowFormatReaderFactory(dataSchemaRowType, projection);
+ return new RowFormatReaderFactory(dataSchemaRowType, projectedRowType);
}
@Override
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
index bda5a428bb..476dffb9a4 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
@@ -28,8 +28,6 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.NestedProjectedRow;
-import javax.annotation.Nullable;
-
import java.io.IOException;
/** Factory for creating {@link RowFormatReader}. */
@@ -38,11 +36,11 @@ public class RowFormatReaderFactory implements
FormatReaderFactory {
private static final int TAIL_PREFETCH_SIZE = 64 * 1024;
private final RowType rowType;
- @Nullable private final NestedProjectedRow projection;
+ private final RowType projectedRowType;
- public RowFormatReaderFactory(RowType rowType, @Nullable
NestedProjectedRow projection) {
+ public RowFormatReaderFactory(RowType rowType, RowType projectedRowType) {
this.rowType = rowType;
- this.projection = projection;
+ this.projectedRowType = projectedRowType;
}
@Override
@@ -77,7 +75,15 @@ public class RowFormatReaderFactory implements
FormatReaderFactory {
}
return new RowFormatReader(
- in, path, footer, blockIndex, rowType, projection,
context.selection());
+ in,
+ path,
+ footer,
+ blockIndex,
+ rowType,
+ // Each reader needs its own wrapper: it is mutated in
place per row,
+ // so sharing one instance across readers corrupts
interleaved reads.
+ NestedProjectedRow.create(rowType, projectedRowType),
+ context.selection());
} catch (Throwable t) {
IOUtils.closeQuietly(in);
throw t;
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReadWriteTest.java
index 25a46020d7..b48bba6e68 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReadWriteTest.java
@@ -605,6 +605,58 @@ public class RowFormatReadWriteTest {
assertThat(result.get(2).getInt(0)).isEqualTo(300);
}
+ @Test
+ public void testInterleavedReadersDoNotShareProjectedRow() throws
IOException {
+ RowType fullType =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "a", new IntType()),
+ new DataField(1, "b", new VarCharType(100))));
+ // The projection has to drop a column: for an identical schema
+ // NestedProjectedRow.create returns null and no wrapper is involved.
+ RowType projectedType = new RowType(Arrays.asList(new DataField(0,
"a", new IntType())));
+
+ Path pathA = new Path(tempDir.toUri().toString(), "interleaved_a.row");
+ Path pathB = new Path(tempDir.toUri().toString(), "interleaved_b.row");
+ FileFormat format = FileFormat.fromIdentifier("row", new Options());
+ writeRows(
+ format,
+ fullType,
+ pathA,
+ Arrays.asList(GenericRow.of(1, BinaryString.fromString("A"))));
+ writeRows(
+ format,
+ fullType,
+ pathB,
+ Arrays.asList(GenericRow.of(2, BinaryString.fromString("B"))));
+
+ LocalFileIO fileIO = new LocalFileIO();
+ FormatReaderFactory readerFactory =
+ format.createReaderFactory(fullType, projectedType, new
ArrayList<>());
+ try (FileRecordReader<InternalRow> readerA =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, pathA,
fileIO.getFileSize(pathA), null, null));
+ FileRecordReader<InternalRow> readerB =
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, pathB,
fileIO.getFileSize(pathB), null, null))) {
+ FileRecordIterator<InternalRow> batchA = readerA.readBatch();
+ assertThat(batchA).isNotNull();
+ InternalRow rowA = batchA.next();
+ assertThat(rowA.getInt(0)).isEqualTo(1);
+
+ FileRecordIterator<InternalRow> batchB = readerB.readBatch();
+ assertThat(batchB).isNotNull();
+ InternalRow rowB = batchB.next();
+ assertThat(rowB.getInt(0)).isEqualTo(2);
+
+ // Reading from B must leave the row A handed out alone.
+ assertThat(rowA).isNotSameAs(rowB);
+ assertThat(rowA.getInt(0)).isEqualTo(1);
+ }
+ }
+
@Test
public void testProjectionMultipleColumns() throws IOException {
RowType fullType =
