From 165cefa8346c65c46417d7a514fdf0088e9eeb8a Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Wed, 7 Oct 2026 18:53:53 +0800 Subject: [PATCH 1/7] [core] Select row sidecars using the final Data Evolution selection --- .../operation/DataEvolutionSplitRead.java | 74 +++++++++++++------ .../operation/DataEvolutionSplitReadTest.java | 16 ++++ .../table/DataEvolutionFileIndexTest.java | 56 ++++++++++++++ 3 files changed, 124 insertions(+), 22 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index b05383463303..76f677c8d80b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -471,7 +471,8 @@ private RecordReader createUnionReader( } FieldBunch bunch = fieldsFiles.get(i); DataFileMeta firstFile = bunch.files().get(0); - FileReadTarget readTarget = readTarget(firstFile, dataFilePathFactory, rowRanges); + FileReadTarget readTarget = + readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); String formatIdentifier = readTarget.formatIdentifier; long schemaId = firstFile.schemaId(); TableSchema dataSchema = bunchDataSchemas[i]; @@ -525,7 +526,8 @@ private RecordReader createMissingFieldsReader( // Use the physical schema: the full table schema may declare columns this file never wrote. FormatReaderMapping mapping = formatBuilder.build( - readTarget(firstFile, dataFilePathFactory, rowRanges).formatIdentifier, + readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection) + .formatIdentifier, schema, dataSchema, readRowType.getFields(), @@ -722,8 +724,6 @@ private FileRecordReader createFileReader( @Nullable DeletionVectorWithRange deletionVector, @Nullable FileIndexResult precomputedFileIndexResult) throws IOException { - FileReadTarget readTarget = readTarget(file, dataFilePathFactory, rowRanges); - String formatIdentifier = readTarget.formatIdentifier; long schemaId = file.schemaId(); TableSchema dataSchema = schemaId == schema.id() ? schema : schemaFetcher.apply(schemaId); boolean nestedFieldEnabled = nestedFieldEnabledFor(Collections.singletonList(file)); @@ -731,18 +731,6 @@ private FileRecordReader createFileReader( // no column merge here, so the filters this file can answer reach both the file index and // the format reader List fileFilters = fileFilters(filters, file); - FormatReaderMapping formatReaderMapping = - singleFileReaderMappings.computeIfAbsent( - new SingleFileKey( - schemaId, - formatIdentifier, - file.writeCols(), - readRowType, - nestedFieldEnabled), - key -> - formatBuilder(readRowType, fileFilters, nestedFieldEnabled) - .build(formatIdentifier, schema, dataSchema)); - FileIndexResult fileIndexResult = precomputedFileIndexResult; if (fileIndexResult == null && fileIndexReadEnabled) { DeletionVector dv = deletionVector == null ? null : deletionVector.deletionVector; @@ -767,6 +755,21 @@ private FileRecordReader createFileReader( } } + FileReadTarget readTarget = + readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult); + String formatIdentifier = readTarget.formatIdentifier; + FormatReaderMapping formatReaderMapping = + singleFileReaderMappings.computeIfAbsent( + new SingleFileKey( + schemaId, + formatIdentifier, + file.writeCols(), + readRowType, + nestedFieldEnabled), + key -> + formatBuilder(readRowType, fileFilters, nestedFieldEnabled) + .build(formatIdentifier, schema, dataSchema)); + return createFileReader( partition, file, @@ -814,7 +817,7 @@ private FileRecordReader createFileReader( formatReaderMapping, rowRanges, readRowType, - readTarget(file, dataFilePathFactory, rowRanges), + readTarget(file, dataFilePathFactory, rowRanges, groupSelection), deletionVector, groupSelection); } @@ -1173,13 +1176,17 @@ private DeletionVectorWithRange readDeletionVector( } private FileReadTarget readTarget( - DataFileMeta file, DataFilePathFactory dataFilePathFactory, List rowRanges) + DataFileMeta file, + DataFilePathFactory dataFilePathFactory, + List rowRanges, + @Nullable FileIndexResult fileIndexResult) throws IOException { String rowSidecar = rowSidecarFileName(file); if (rowSidecar != null && shouldReadRowSidecar( file, rowRanges, + fileIndexResult, coreOptions.dataEvolutionRowSidecarMaxSelectedRows(), coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) { Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file); @@ -1213,6 +1220,7 @@ static boolean shouldReadRowSidecar(DataFileMeta file, @Nullable List row return shouldReadRowSidecar( file, rowRanges, + null, CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(), CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue()); } @@ -1223,16 +1231,24 @@ static boolean shouldReadRowSidecar( @Nullable List rowRanges, long maxSelectedRows, double maxSelectionRatio) { - if (rowRanges == null - || rowRanges.isEmpty() - || file.rowCount() <= 0 + return shouldReadRowSidecar(file, rowRanges, null, maxSelectedRows, maxSelectionRatio); + } + + @VisibleForTesting + static boolean shouldReadRowSidecar( + DataFileMeta file, + @Nullable List rowRanges, + @Nullable FileIndexResult fileIndexResult, + long maxSelectedRows, + double maxSelectionRatio) { + if (file.rowCount() <= 0 || isBlobFile(file.fileName()) || isVectorStoreFile(file.fileName()) || rowSidecarFileName(file) == null) { return false; } - long selectedRowCount = selectedRowCount(file, rowRanges); + long selectedRowCount = selectedRowCount(file, rowRanges, fileIndexResult); if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) { return false; } @@ -1256,6 +1272,20 @@ static long selectedRowCount(DataFileMeta file, List rowRanges) { .sum(); } + private static long selectedRowCount( + DataFileMeta file, + @Nullable List rowRanges, + @Nullable FileIndexResult fileIndexResult) { + if (fileIndexResult instanceof BitmapIndexResult) { + RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get(); + RoaringBitmap32 rangeSelection = file.toFileSelection(rowRanges); + return rangeSelection == null + ? indexSelection.getCardinality() + : RoaringBitmap32.and(rangeSelection, indexSelection).getCardinality(); + } + return isNullOrEmpty(rowRanges) ? file.rowCount() : selectedRowCount(file, rowRanges); + } + private static boolean isRowSidecarFile(String fileName) { try { return ROW_SIDECAR_FORMAT.equals(DataFilePathFactory.formatIdentifier(fileName)); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java index cd60d981643a..530ef8e4917a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java @@ -23,6 +23,7 @@ import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fileindex.bitmap.BitmapIndexResult; import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.FormatWriter; import org.apache.paimon.fs.Path; @@ -42,6 +43,7 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringBitmap32; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -225,6 +227,20 @@ public void testShouldReadRowSidecarRequiresSmallCountAndLowRatio() { smallFile, Collections.singletonList(new Range(10L, 15L)), 64L, 0.25d)); } + @Test + public void testShouldReadRowSidecarUsesFinalBitmapSelection() { + DataFileMeta file = + createFile("file1.parquet", 10L, 100, 1, Collections.singletonList("file1.row")); + BitmapIndexResult bitmap = new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(0, 5, 20)); + List rowRanges = Collections.singletonList(new Range(10L, 19L)); + + assertTrue(DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.02d)); + assertFalse( + DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 1L, 0.02d)); + assertFalse( + DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.01d)); + } + @Test public void testShouldNotReadRowSidecarWithoutSparseSelection() { DataFileMeta file = diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index c2629f4f9817..16894cb29dac 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -40,6 +40,7 @@ import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.io.DataIncrement; import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.IndexManifestEntry; @@ -679,6 +680,32 @@ public void testMergedGroupDvAwareBitmapSelectionIsPushedDown() throws Exception assertRow(assertSingleRow(rows), 50); } + @Test + public void testMergedGroupFinalSelectionChoosesRowSidecar() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("merged_bitmap_dv_row_sidecar", options); + writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"), Collections.emptyMap()); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter)), 50); + } + + @Test + public void testSingleFileFinalSelectionChoosesRowSidecar() throws Exception { + Map options = bitmapOptions("f1"); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("single_bitmap_dv_row_sidecar", options); + writeAllColumns(table, ROW_COUNT); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter)), 50); + } + @Test public void testMergedGroupFileIndexSkipsBeforeReadingDeletionVector() throws Exception { Map options = new HashMap<>(); @@ -1040,6 +1067,27 @@ private List readWithFilter(FileStoreTable table, Predicate predica return readWithFilter(table, predicate, null); } + private List readAfterDeletingDataFiles(FileStoreTable table, Predicate predicate) + throws Exception { + FileStoreTable latest = getTable(identifier(table.name())); + ReadBuilder readBuilder = latest.newReadBuilder().withFilter(predicate); + DataSplit split = (DataSplit) readBuilder.newScan().plan().splits().get(0); + DataFilePathFactory pathFactory = + latest.store() + .pathFactory() + .createDataFilePathFactory(split.partition(), split.bucket()); + for (DataFileMeta file : split.dataFiles()) { + assertThat(latest.fileIO().delete(pathFactory.toPath(file), false)).isTrue(); + } + + List rows = new ArrayList<>(); + InternalRowSerializer serializer = new InternalRowSerializer(latest.rowType()); + try (RecordReader reader = readBuilder.newRead().createReader(split)) { + reader.forEachRemaining(row -> rows.add(serializer.copy(row))); + } + return rows; + } + private List readWithFilter( FileStoreTable table, Predicate predicate, @Nullable RowType readType) throws Exception { @@ -1195,6 +1243,14 @@ private static Predicate equalF1(String value) { return new PredicateBuilder(rowType()).equal(1, BinaryString.fromString(value)); } + private static Predicate equalAnyF1(int... values) { + Predicate[] predicates = new Predicate[values.length]; + for (int i = 0; i < values.length; i++) { + predicates[i] = equalF1(f1(values[i])); + } + return PredicateBuilder.or(predicates); + } + private static Predicate equalF2(String value) { return new PredicateBuilder(rowType()).equal(2, BinaryString.fromString(value)); } From f6eb605a0f785d9ec4c4102ccba9f13d8360c779 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Thu, 8 Oct 2026 12:42:12 +0800 Subject: [PATCH 2/7] Avoid row sidecar for bitmap-only reads without row ranges --- .../operation/DataEvolutionSplitRead.java | 3 +- .../operation/DataEvolutionSplitReadTest.java | 3 + .../table/DataEvolutionFileIndexTest.java | 115 +++++++++--------- 3 files changed, 65 insertions(+), 56 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index 76f677c8d80b..1a48b1760fd8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -1241,7 +1241,8 @@ static boolean shouldReadRowSidecar( @Nullable FileIndexResult fileIndexResult, long maxSelectedRows, double maxSelectionRatio) { - if (file.rowCount() <= 0 + if (isNullOrEmpty(rowRanges) + || file.rowCount() <= 0 || isBlobFile(file.fileName()) || isVectorStoreFile(file.fileName()) || rowSidecarFileName(file) == null) { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java index 530ef8e4917a..998ab24e501f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java @@ -235,6 +235,9 @@ public void testShouldReadRowSidecarUsesFinalBitmapSelection() { List rowRanges = Collections.singletonList(new Range(10L, 19L)); assertTrue(DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.02d)); + assertFalse( + DataEvolutionSplitRead.shouldReadRowSidecar( + file, null, bitmap, 2L, 0.02d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 1L, 0.02d)); assertFalse( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index 16894cb29dac..a81882f68298 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -114,6 +114,66 @@ public class DataEvolutionFileIndexTest extends DataEvolutionTestBase { /** Inside the min/max range of f2 ("b000".."b099"), but not written. */ private static final String MISSING_F2 = "b050x"; + @Test + public void testBitmapFilterWithProjectedWriteSchemaAndRowSidecar() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("projected_schema_row_sidecar", options); + writeSplitColumns(table, ROW_COUNT, Collections.emptyMap(), bitmapOptions("f2")); + + FileStoreTable latest = getTable(identifier(table.name())); + RowType projection = latest.rowType().project("f2"); + Predicate filter = equalF2(f2(50)); + ReadBuilder readBuilder = + latest.newReadBuilder().withReadType(projection).withFilter(filter); + List rows = + collect( + readBuilder.newRead().executeFilter(), + readBuilder.newScan().plan(), + projection); + + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(0).toString()).isEqualTo(f2(50)); + } + + @Test + public void testBitmapFilterWithWideWriteSchemaAndRowSidecar() throws Exception { + Schema.Builder schemaBuilder = Schema.newBuilder(); + for (int i = 0; i < 7; i++) { + schemaBuilder.column("c" + i, DataTypes.INT()); + } + schemaBuilder + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + bitmapOptions("c6").forEach(schemaBuilder::option); + Identifier identifier = identifier("wide_schema_row_sidecar"); + catalog.createTable(identifier, schemaBuilder.build(), false); + FileStoreTable table = getTable(identifier); + + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (int i = 0; i < ROW_COUNT; i++) { + write.write(GenericRow.of(i, i, i, i, i, i, i)); + } + commit.commit(write.prepareCommit()); + } + + Predicate filter = new PredicateBuilder(table.rowType()).equal(6, 50); + ReadBuilder readBuilder = table.newReadBuilder().withFilter(filter); + List rows = + collect( + readBuilder.newRead().executeFilter(), + readBuilder.newScan().plan(), + table.rowType()); + + InternalRow row = assertSingleRow(rows); + for (int i = 0; i < 7; i++) { + assertThat(row.getInt(i)).isEqualTo(50); + } + } + @Test public void testSingleFileSkippedByFileIndex() throws Exception { // a standalone .index file, only the reader can evaluate it @@ -680,32 +740,6 @@ public void testMergedGroupDvAwareBitmapSelectionIsPushedDown() throws Exception assertRow(assertSingleRow(rows), 50); } - @Test - public void testMergedGroupFinalSelectionChoosesRowSidecar() throws Exception { - Map options = new HashMap<>(); - options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); - options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); - FileStoreTable table = createTable("merged_bitmap_dv_row_sidecar", options); - writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"), Collections.emptyMap()); - deleteRows(table, 51, 52, 53, 54, 55); - - Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); - assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter)), 50); - } - - @Test - public void testSingleFileFinalSelectionChoosesRowSidecar() throws Exception { - Map options = bitmapOptions("f1"); - options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); - options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); - FileStoreTable table = createTable("single_bitmap_dv_row_sidecar", options); - writeAllColumns(table, ROW_COUNT); - deleteRows(table, 51, 52, 53, 54, 55); - - Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); - assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter)), 50); - } - @Test public void testMergedGroupFileIndexSkipsBeforeReadingDeletionVector() throws Exception { Map options = new HashMap<>(); @@ -1067,27 +1101,6 @@ private List readWithFilter(FileStoreTable table, Predicate predica return readWithFilter(table, predicate, null); } - private List readAfterDeletingDataFiles(FileStoreTable table, Predicate predicate) - throws Exception { - FileStoreTable latest = getTable(identifier(table.name())); - ReadBuilder readBuilder = latest.newReadBuilder().withFilter(predicate); - DataSplit split = (DataSplit) readBuilder.newScan().plan().splits().get(0); - DataFilePathFactory pathFactory = - latest.store() - .pathFactory() - .createDataFilePathFactory(split.partition(), split.bucket()); - for (DataFileMeta file : split.dataFiles()) { - assertThat(latest.fileIO().delete(pathFactory.toPath(file), false)).isTrue(); - } - - List rows = new ArrayList<>(); - InternalRowSerializer serializer = new InternalRowSerializer(latest.rowType()); - try (RecordReader reader = readBuilder.newRead().createReader(split)) { - reader.forEachRemaining(row -> rows.add(serializer.copy(row))); - } - return rows; - } - private List readWithFilter( FileStoreTable table, Predicate predicate, @Nullable RowType readType) throws Exception { @@ -1243,14 +1256,6 @@ private static Predicate equalF1(String value) { return new PredicateBuilder(rowType()).equal(1, BinaryString.fromString(value)); } - private static Predicate equalAnyF1(int... values) { - Predicate[] predicates = new Predicate[values.length]; - for (int i = 0; i < values.length; i++) { - predicates[i] = equalF1(f1(values[i])); - } - return PredicateBuilder.or(predicates); - } - private static Predicate equalF2(String value) { return new PredicateBuilder(rowType()).equal(2, BinaryString.fromString(value)); } From 5ee90ac7506f8a6f7c7df150d499a7153b407fe2 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Thu, 8 Oct 2026 12:48:46 +0800 Subject: [PATCH 3/7] fix code format --- .../apache/paimon/operation/DataEvolutionSplitReadTest.java | 4 +--- .../org/apache/paimon/table/DataEvolutionFileIndexTest.java | 1 - 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java index 998ab24e501f..aeec3492f691 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java @@ -235,9 +235,7 @@ public void testShouldReadRowSidecarUsesFinalBitmapSelection() { List rowRanges = Collections.singletonList(new Range(10L, 19L)); assertTrue(DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.02d)); - assertFalse( - DataEvolutionSplitRead.shouldReadRowSidecar( - file, null, bitmap, 2L, 0.02d)); + assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null, bitmap, 2L, 0.02d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 1L, 0.02d)); assertFalse( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index a81882f68298..a119b80c4dd6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -40,7 +40,6 @@ import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; -import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.io.DataIncrement; import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.IndexManifestEntry; From 506ac4a93bbbb4040621c89588e137d1cb9707a6 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Thu, 8 Oct 2026 17:29:55 +0800 Subject: [PATCH 4/7] fix row sidecar schema consistency --- .../operation/DataEvolutionSplitRead.java | 53 +++++++++++---- .../operation/DataEvolutionSplitReadTest.java | 13 +++- .../table/DataEvolutionFileIndexTest.java | 64 +++++++++++++++++++ 3 files changed, 117 insertions(+), 13 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index 1a48b1760fd8..d89106eeb4cd 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -455,7 +455,7 @@ private RecordReader createUnionReader( fieldsFiles.get(0), bunchDataSchemas[0], dataFilePathFactory, - formatBuilder, + nestedFieldEnabled, rowRanges, readRowType, deletionVector, @@ -474,6 +474,10 @@ private RecordReader createUnionReader( FileReadTarget readTarget = readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); String formatIdentifier = readTarget.formatIdentifier; + Builder targetFormatBuilder = + ROW_SIDECAR_FORMAT.equals(formatIdentifier) + ? rowSidecarFormatBuilder(readRowType, null, nestedFieldEnabled) + : formatBuilder; long schemaId = firstFile.schemaId(); TableSchema dataSchema = bunchDataSchemas[i]; RowType partialReadRowType = new RowType(readFields); @@ -485,7 +489,7 @@ private RecordReader createUnionReader( formatReaderMappings.computeIfAbsent( new FormatKey(schemaId, formatIdentifier, cacheKey), key -> - formatBuilder.build( + targetFormatBuilder.build( formatIdentifier, schema, dataSchema, @@ -516,18 +520,23 @@ private RecordReader createMissingFieldsReader( FieldBunch bunch, TableSchema dataSchema, DataFilePathFactory dataFilePathFactory, - Builder formatBuilder, + boolean nestedFieldEnabled, List rowRanges, RowType readRowType, @Nullable DeletionVectorWithRange deletionVector, @Nullable BitmapIndexResult groupSelection) throws IOException { DataFileMeta firstFile = bunch.files().get(0); + FileReadTarget readTarget = + readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); + Builder formatBuilder = + ROW_SIDECAR_FORMAT.equals(readTarget.formatIdentifier) + ? rowSidecarFormatBuilder(readRowType, null, nestedFieldEnabled) + : formatBuilder(readRowType, null, nestedFieldEnabled); // Use the physical schema: the full table schema may declare columns this file never wrote. FormatReaderMapping mapping = formatBuilder.build( - readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection) - .formatIdentifier, + readTarget.formatIdentifier, schema, dataSchema, readRowType.getFields(), @@ -725,7 +734,8 @@ private FileRecordReader createFileReader( @Nullable FileIndexResult precomputedFileIndexResult) throws IOException { long schemaId = file.schemaId(); - TableSchema dataSchema = schemaId == schema.id() ? schema : schemaFetcher.apply(schemaId); + TableSchema fileSchema = schemaId == schema.id() ? schema : schemaFetcher.apply(schemaId); + TableSchema physicalDataSchema = fileSchema.dataFileSchema(file.writeCols()); boolean nestedFieldEnabled = nestedFieldEnabledFor(Collections.singletonList(file)); // no column merge here, so the filters this file can answer reach both the file index and @@ -742,8 +752,8 @@ private FileRecordReader createFileReader( fileIndexResult = FileIndexEvaluator.evaluate( fileIO, - dataSchema, - devolveFilters(fileFilters, dataSchema), + physicalDataSchema, + devolveFilters(fileFilters, physicalDataSchema), null, null, dataFilePathFactory, @@ -758,6 +768,12 @@ private FileRecordReader createFileReader( FileReadTarget readTarget = readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult); String formatIdentifier = readTarget.formatIdentifier; + boolean readRowSidecar = ROW_SIDECAR_FORMAT.equals(formatIdentifier); + Builder targetFormatBuilder = + readRowSidecar + ? rowSidecarFormatBuilder(readRowType, fileFilters, nestedFieldEnabled) + : formatBuilder(readRowType, fileFilters, nestedFieldEnabled); + TableSchema readerDataSchema = readRowSidecar ? physicalDataSchema : fileSchema; FormatReaderMapping formatReaderMapping = singleFileReaderMappings.computeIfAbsent( new SingleFileKey( @@ -767,8 +783,8 @@ private FileRecordReader createFileReader( readRowType, nestedFieldEnabled), key -> - formatBuilder(readRowType, fileFilters, nestedFieldEnabled) - .build(formatIdentifier, schema, dataSchema)); + targetFormatBuilder.build( + formatIdentifier, schema, readerDataSchema)); return createFileReader( partition, @@ -1081,6 +1097,20 @@ private Builder formatBuilder( nestedFieldEnabled); } + // Row sidecars use positional decoding and contain exactly the physical write schema. + // Row tracking fields are supplied by DataFileRecordReader from the manifest entry. + private Builder rowSidecarFormatBuilder( + RowType readRowType, @Nullable List filters, boolean nestedFieldEnabled) { + return new Builder( + formatDiscover, + readRowType.getFields(), + TableSchema::fields, + filters, + null, + null, + nestedFieldEnabled); + } + /** * Filters the file can answer. A column missing from a data evolution file is not null, its * values live in another file of the same row id range, so a predicate on it must not be pushed @@ -1241,8 +1271,7 @@ static boolean shouldReadRowSidecar( @Nullable FileIndexResult fileIndexResult, long maxSelectedRows, double maxSelectionRatio) { - if (isNullOrEmpty(rowRanges) - || file.rowCount() <= 0 + if (file.rowCount() <= 0 || isBlobFile(file.fileName()) || isVectorStoreFile(file.fileName()) || rowSidecarFileName(file) == null) { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java index aeec3492f691..2a1c5c4472eb 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java @@ -232,10 +232,21 @@ public void testShouldReadRowSidecarUsesFinalBitmapSelection() { DataFileMeta file = createFile("file1.parquet", 10L, 100, 1, Collections.singletonList("file1.row")); BitmapIndexResult bitmap = new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(0, 5, 20)); + BitmapIndexResult singleRowBitmap = + new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(50)); List rowRanges = Collections.singletonList(new Range(10L, 19L)); assertTrue(DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.02d)); - assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null, bitmap, 2L, 0.02d)); + assertTrue( + DataEvolutionSplitRead.shouldReadRowSidecar( + file, null, singleRowBitmap, 1L, 0.01d)); + assertTrue( + DataEvolutionSplitRead.shouldReadRowSidecar( + file, + Collections.singletonList(new Range(10L, 109L)), + singleRowBitmap, + 1L, + 0.01d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 1L, 0.02d)); assertFalse( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index a119b80c4dd6..91e36e27c1cf 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -62,6 +62,7 @@ import org.apache.paimon.table.source.TableScan; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Range; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -133,6 +134,37 @@ public void testBitmapFilterWithProjectedWriteSchemaAndRowSidecar() throws Excep assertThat(rows).hasSize(1); assertThat(rows.get(0).getString(0).toString()).isEqualTo(f2(50)); + + ReadBuilder rangeReadBuilder = + latest.newReadBuilder() + .withReadType(projection) + .withFilter(filter) + .withRowRanges(Collections.singletonList(new Range(0L, ROW_COUNT - 1L))); + rows = + collect( + rangeReadBuilder.newRead().executeFilter(), + rangeReadBuilder.newScan().plan(), + projection); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(0).toString()).isEqualTo(f2(50)); + + RowType rowIdType = rowTypeWithRowId(latest.rowType()); + RowType rowIdProjection = rowIdType.project(SpecialFields.ROW_ID.name(), "f2"); + PredicateBuilder rowIdBuilder = new PredicateBuilder(rowIdType); + Predicate rowIdFilter = + PredicateBuilder.and( + rowIdBuilder.between(3, 0L, ROW_COUNT - 1L), + rowIdBuilder.equal(2, BinaryString.fromString(f2(50)))); + ReadBuilder rowIdReadBuilder = + latest.newReadBuilder().withReadType(rowIdProjection).withFilter(rowIdFilter); + rows = + collect( + rowIdReadBuilder.newRead().executeFilter(), + rowIdReadBuilder.newScan().plan(), + rowIdProjection); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(0)).isEqualTo(50L); + assertThat(rows.get(0).getString(1).toString()).isEqualTo(f2(50)); } @Test @@ -171,6 +203,38 @@ public void testBitmapFilterWithWideWriteSchemaAndRowSidecar() throws Exception for (int i = 0; i < 7; i++) { assertThat(row.getInt(i)).isEqualTo(50); } + + ReadBuilder rangeReadBuilder = + table.newReadBuilder() + .withFilter(filter) + .withRowRanges(Collections.singletonList(new Range(0L, ROW_COUNT - 1L))); + rows = + collect( + rangeReadBuilder.newRead().executeFilter(), + rangeReadBuilder.newScan().plan(), + table.rowType()); + row = assertSingleRow(rows); + for (int i = 0; i < 7; i++) { + assertThat(row.getInt(i)).isEqualTo(50); + } + + RowType rowIdType = rowTypeWithRowId(table.rowType()); + PredicateBuilder rowIdBuilder = new PredicateBuilder(rowIdType); + Predicate rowIdFilter = + PredicateBuilder.and( + rowIdBuilder.between(7, 0L, ROW_COUNT - 1L), rowIdBuilder.equal(6, 50)); + ReadBuilder rowIdReadBuilder = + table.newReadBuilder().withReadType(rowIdType).withFilter(rowIdFilter); + rows = + collect( + rowIdReadBuilder.newRead().executeFilter(), + rowIdReadBuilder.newScan().plan(), + rowIdType); + row = assertSingleRow(rows); + for (int i = 0; i < 7; i++) { + assertThat(row.getInt(i)).isEqualTo(50); + } + assertThat(row.getLong(7)).isEqualTo(50L); } @Test From 1369458e785f47a1fdf30e72e28ec2f5df2a85dd Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Sat, 10 Oct 2026 19:36:13 +0800 Subject: [PATCH 5/7] fix row sidecar reader cache key --- .../operation/DataEvolutionSplitRead.java | 4 +- .../table/DataEvolutionFileIndexTest.java | 48 +++++++++++++++++++ 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index d89106eeb4cd..e3699fbfab2f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -482,9 +482,7 @@ private RecordReader createUnionReader( TableSchema dataSchema = bunchDataSchemas[i]; RowType partialReadRowType = new RowType(readFields); List cacheKey = - nestedFieldEnabled - ? readerCacheKey(readFields, dataSchema.fields(), true) - : readFields.stream().map(DataField::name).collect(Collectors.toList()); + readerCacheKey(readFields, dataSchema.fields(), nestedFieldEnabled); FormatReaderMapping formatReaderMapping = formatReaderMappings.computeIfAbsent( new FormatKey(schemaId, formatIdentifier, cacheKey), diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index 91e36e27c1cf..f4f49e25c2c8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -167,6 +167,25 @@ public void testBitmapFilterWithProjectedWriteSchemaAndRowSidecar() throws Excep assertThat(rows.get(0).getString(1).toString()).isEqualTo(f2(50)); } + @Test + public void testFormatReaderCacheKeyIncludesPhysicalSchema() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("row_sidecar_physical_schema_key", options); + writeSplitColumns(table, ROW_COUNT, Collections.emptyMap(), bitmapOptions("f2")); + writeSecondSplitColumns(table, ROW_COUNT, bitmapOptions("f2")); + + FileStoreTable latest = getTable(identifier(table.name())); + RowType projection = latest.rowType().project(Arrays.asList("f1", "f2")); + List rows = readWithFilter(table, equalF2(f2(50)), projection); + + assertThat(rows).hasSize(2); + assertThat(rows) + .extracting(row -> row.getString(0).toString()) + .containsExactlyInAnyOrder(f1(50), "second-" + f1(50)); + assertThat(rows).extracting(row -> row.getString(1).toString()).containsOnly(f2(50)); + } + @Test public void testBitmapFilterWithWideWriteSchemaAndRowSidecar() throws Exception { Schema.Builder schemaBuilder = Schema.newBuilder(); @@ -1115,6 +1134,35 @@ private void writeSplitColumns( } } + /** Writes f1 and f2 into a second merged group with a different physical schema. */ + private void writeSecondSplitColumns( + FileStoreTable table, int count, Map secondOptions) throws Exception { + FileStoreTable latest = getTable(identifier(table.name())); + RowType writeType0 = latest.rowType().project(Collections.singletonList("f1")); + BatchWriteBuilder builder = latest.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite().withWriteType(writeType0); + BatchTableCommit commit = builder.newCommit()) { + for (int i = 0; i < count; i++) { + write.write(GenericRow.of(BinaryString.fromString("second-" + f1(i)))); + } + commit.commit(write.prepareCommit()); + } + + latest = getTable(identifier(table.name())); + long firstRowId = latest.snapshotManager().latestSnapshot().nextRowId() - count; + RowType writeType1 = latest.rowType().project(Collections.singletonList("f2")); + builder = latest.copy(secondOptions).newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite().withWriteType(writeType1); + BatchTableCommit commit = builder.newCommit()) { + for (int i = 0; i < count; i++) { + write.write(GenericRow.of(BinaryString.fromString(f2(i)))); + } + List commitables = write.prepareCommit(); + setSingleFileFirstRowId(commitables, firstRowId); + commit.commit(commitables); + } + } + /** * Writes f0/f1 (a*) with a bloom index on f1, then overwrites f1 with c* in a second file that * shares the same row id range and has a higher sequence number, so the column merge takes f1 From fa8fbe02d36865aa0d707db4a4cedbca5fd9012f Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Thu, 15 Oct 2026 10:38:13 +0800 Subject: [PATCH 6/7] Preserve tracking and partition fields in data evolution reads --- .../operation/DataEvolutionSplitRead.java | 82 +++- .../table/DataEvolutionFileIndexTest.java | 396 ++++++++++++++++++ 2 files changed, 464 insertions(+), 14 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index e3699fbfab2f..966563aa865e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -19,10 +19,14 @@ package org.apache.paimon.operation; import org.apache.paimon.CoreOptions; +import org.apache.paimon.PartitionSettedRow; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.append.ForceSingleBatchReader; +import org.apache.paimon.casting.CastExecutors; +import org.apache.paimon.casting.CastFieldGetter; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.PartitionInfo; import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader; import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.disk.IOManager; @@ -61,6 +65,8 @@ import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.FormatReaderMapping; import org.apache.paimon.utils.FormatReaderMapping.Builder; +import org.apache.paimon.utils.InternalRowUtils; +import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.Preconditions; import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RangeHelper; @@ -212,7 +218,15 @@ private RecordReader createReader(DataSplit dataSplit) throws IOExc } private RecordReader createReader( - DataSplit dataSplit, List rowRanges, RowType readRowType) throws IOException { + DataSplit dataSplit, List rowRanges, RowType outputRowType) throws IOException { + // Partition values belong to the split and may be absent from every physical file. + // Plan and merge only data columns, then restore partitions in the requested order. + Pair, List> partitionMapping = + PartitionUtils.getPartitionMapping2fieldsWithoutPartition( + schema.partitionKeys(), + outputRowType.getFields(), + schema.logicalPartitionType()); + RowType readRowType = new RowType(partitionMapping.getRight()); List files = dataSplit.dataFiles(); List deletionFiles = dataSplit.deletionFiles().orElse(null); @@ -228,7 +242,7 @@ private RecordReader createReader( // the suppliers below run lazily, so take the filters now, the same way the read type is // already taken by the caller - List filters = readTypeFilters(this.filters, readRowType); + List filters = readTypeFilters(this.filters, outputRowType); List> splitByRowId = mergeRangesAndSort(files); for (List needMergeFiles : splitByRowId) { @@ -308,7 +322,13 @@ private RecordReader createReader( } } - return ConcatRecordReader.create(suppliers); + RecordReader reader = ConcatRecordReader.create(suppliers); + PartitionInfo partitionInfo = PartitionUtils.create(partitionMapping.getLeft(), partition); + if (partitionInfo != null) { + PartitionSettedRow partitionRow = PartitionSettedRow.from(partitionInfo); + return reader.transform(partitionRow::replaceRow); + } + return reader; } private RecordReader createReader(IndexedSplit indexedSplit) throws IOException { @@ -475,8 +495,8 @@ private RecordReader createUnionReader( readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); String formatIdentifier = readTarget.formatIdentifier; Builder targetFormatBuilder = - ROW_SIDECAR_FORMAT.equals(formatIdentifier) - ? rowSidecarFormatBuilder(readRowType, null, nestedFieldEnabled) + readTarget.isPositionalRowFormat() + ? positionalRowFormatBuilder(readRowType, null, nestedFieldEnabled) : formatBuilder; long schemaId = firstFile.schemaId(); TableSchema dataSchema = bunchDataSchemas[i]; @@ -528,8 +548,8 @@ private RecordReader createMissingFieldsReader( FileReadTarget readTarget = readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); Builder formatBuilder = - ROW_SIDECAR_FORMAT.equals(readTarget.formatIdentifier) - ? rowSidecarFormatBuilder(readRowType, null, nestedFieldEnabled) + readTarget.isPositionalRowFormat() + ? positionalRowFormatBuilder(readRowType, null, nestedFieldEnabled) : formatBuilder(readRowType, null, nestedFieldEnabled); // Use the physical schema: the full table schema may declare columns this file never wrote. FormatReaderMapping mapping = @@ -766,12 +786,12 @@ private FileRecordReader createFileReader( FileReadTarget readTarget = readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult); String formatIdentifier = readTarget.formatIdentifier; - boolean readRowSidecar = ROW_SIDECAR_FORMAT.equals(formatIdentifier); + boolean positionalRowFormat = readTarget.isPositionalRowFormat(); Builder targetFormatBuilder = - readRowSidecar - ? rowSidecarFormatBuilder(readRowType, fileFilters, nestedFieldEnabled) + positionalRowFormat + ? positionalRowFormatBuilder(readRowType, fileFilters, nestedFieldEnabled) : formatBuilder(readRowType, fileFilters, nestedFieldEnabled); - TableSchema readerDataSchema = readRowSidecar ? physicalDataSchema : fileSchema; + TableSchema readerDataSchema = positionalRowFormat ? physicalDataSchema : fileSchema; FormatReaderMapping formatReaderMapping = singleFileReaderMappings.computeIfAbsent( new SingleFileKey( @@ -873,7 +893,7 @@ private FileRecordReader createFileReader( coreOptions.scanIgnoreCorruptFile(), coreOptions.scanIgnoreLostFile(), formatReaderMapping.getIndexMapping(), - formatReaderMapping.getCastMapping(), + trackingCastMapping(readTarget, formatReaderMapping), PartitionUtils.create(formatReaderMapping.getPartitionPair(), partition), true, file.firstRowId(), @@ -889,6 +909,32 @@ private FileRecordReader createFileReader( fileRecordReader, file.nonNullRowIdRange(), rowRanges, deletionVector); } + @Nullable + private CastFieldGetter[] trackingCastMapping( + FileReadTarget target, FormatReaderMapping mapping) { + CastFieldGetter[] casts = mapping.getCastMapping(); + if (!target.isPositionalRowFormat() + || casts == null + || mapping.getSystemFields().isEmpty()) { + return casts; + } + // The physical schema omits tracking fields, but DataFileRecordReader supplies them + // before casting. Preserve those values without changing the cached cast mapping. + CastFieldGetter[] result = casts.clone(); + for (DataField field : + new DataField[] {SpecialFields.ROW_ID, SpecialFields.SEQUENCE_NUMBER}) { + Integer index = mapping.getSystemFields().get(field.name()); + if (index != null) { + result[index] = + new CastFieldGetter( + InternalRowUtils.createNullCheckingFieldGetter(field.type(), index), + CastExecutors.identityCastExecutor(), + field.name()); + } + } + return result; + } + private FileRecordReader applyDeletionVector( FileRecordReader reader, Range readerRange, @@ -1095,9 +1141,9 @@ private Builder formatBuilder( nestedFieldEnabled); } - // Row sidecars use positional decoding and contain exactly the physical write schema. + // Row sidecars and ordinary row-format data files decode against the physical write schema. // Row tracking fields are supplied by DataFileRecordReader from the manifest entry. - private Builder rowSidecarFormatBuilder( + private Builder positionalRowFormatBuilder( RowType readRowType, @Nullable List filters, boolean nestedFieldEnabled) { return new Builder( formatDiscover, @@ -1333,6 +1379,14 @@ private FileReadTarget(String formatIdentifier, Path path, long fileSize) { this.path = path; this.fileSize = fileSize; } + + /** + * Sidecars and ordinary "row" data files use the same positional decoder and physical + * schema mapping. They can share a cache entry for the same physical and requested fields. + */ + private boolean isPositionalRowFormat() { + return ROW_SIDECAR_FORMAT.equals(formatIdentifier); + } } /** diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index f4f49e25c2c8..4b0029f88516 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -37,9 +37,12 @@ import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory; import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory; import org.apache.paimon.fs.Path; +import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.BundleRecords; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.io.DataIncrement; import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.IndexManifestEntry; @@ -75,6 +78,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; @@ -167,6 +171,343 @@ public void testBitmapFilterWithProjectedWriteSchemaAndRowSidecar() throws Excep assertThat(rows.get(0).getString(1).toString()).isEqualTo(f2(50)); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testPositionalRowPreservesTrackingFieldsAfterSchemaCast(boolean ordinaryRowFormat) + throws Exception { + Map options = bitmapOptions("f1"); + options.put( + CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), + Boolean.toString(!ordinaryRowFormat)); + if (ordinaryRowFormat) { + options.put(CoreOptions.FILE_FORMAT.key(), "row"); + } + FileStoreTable table = createTable("row_sidecar_tracking_cast", options); + writeAllColumns(table, ROW_COUNT); + catalog.alterTable( + identifier("row_sidecar_tracking_cast"), + SchemaChange.updateColumnType("f0", DataTypes.BIGINT()), + false); + + FileStoreTable latest = getTable(identifier("row_sidecar_tracking_cast")); + RowType readType = + SpecialFields.rowTypeWithRowTracking(latest.rowType(), true, true) + .project( + "f0", + SpecialFields.ROW_ID.name(), + SpecialFields.SEQUENCE_NUMBER.name()); + Predicate filter = + new PredicateBuilder(latest.rowType()).equal(1, BinaryString.fromString(f1(50))); + + ReadBuilder readBuilder = latest.newReadBuilder().withReadType(readType).withFilter(filter); + TableScan.Plan plan = readBuilder.newScan().plan(); + long maxSequenceNumber = + ((DataSplit) plan.splits().get(0)).dataFiles().get(0).maxSequenceNumber(); + List rows = collect(readBuilder.newRead().executeFilter(), plan, readType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(0)).isEqualTo(50L); + assertThat(rows.get(0).getLong(1)).isEqualTo(50L); + assertThat(rows.get(0).isNullAt(2)).isFalse(); + assertThat(rows.get(0).getLong(2)).isEqualTo(maxSequenceNumber); + + ReadBuilder rangeReadBuilder = + latest.newReadBuilder() + .withReadType(readType) + .withFilter(filter) + .withRowRanges(Collections.singletonList(new Range(0L, ROW_COUNT - 1L))); + rows = + collect( + rangeReadBuilder.newRead().executeFilter(), + rangeReadBuilder.newScan().plan(), + readType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(0)).isEqualTo(50L); + assertThat(rows.get(0).getLong(1)).isEqualTo(50L); + assertThat(rows.get(0).isNullAt(2)).isFalse(); + assertThat(rows.get(0).getLong(2)).isEqualTo(maxSequenceNumber); + + RowType trackedType = SpecialFields.rowTypeWithRowTracking(latest.rowType(), true, true); + PredicateBuilder trackedBuilder = new PredicateBuilder(trackedType); + Predicate rowIdFilter = + PredicateBuilder.and( + filter, + trackedBuilder.between( + trackedType.getFieldIndex(SpecialFields.ROW_ID.name()), + 0L, + ROW_COUNT - 1L)); + ReadBuilder rowIdReadBuilder = + latest.newReadBuilder().withReadType(readType).withFilter(rowIdFilter); + rows = + collect( + rowIdReadBuilder.newRead().executeFilter(), + rowIdReadBuilder.newScan().plan(), + readType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(0)).isEqualTo(50L); + assertThat(rows.get(0).getLong(1)).isEqualTo(50L); + assertThat(rows.get(0).getLong(2)).isEqualTo(maxSequenceNumber); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testRowSidecarRestoresMissingPartitionField(boolean merged) throws Exception { + Schema.Builder schemaBuilder = + Schema.newBuilder() + .column("pt", DataTypes.STRING()) + .column("f1", DataTypes.STRING()) + .column("f2", DataTypes.STRING()) + .partitionKeys("pt") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + Identifier identifier = identifier("row_sidecar_partition_restore"); + catalog.createTable(identifier, schemaBuilder.build(), false); + FileStoreTable table = getTable(identifier); + + RowType writeType = table.rowType().project("f1"); + BatchWriteBuilder writeBuilder = table.copy(bitmapOptions("f1")).newBatchWriteBuilder(); + List writtenRows = new ArrayList<>(); + for (int i = 0; i < ROW_COUNT; i++) { + writtenRows.add(GenericRow.of(BinaryString.fromString(f1(i)))); + } + try (BatchTableWrite write = writeBuilder.newWrite().withWriteType(writeType); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.writeBundle( + new InternalRowSerializer(table.schema().logicalPartitionType()) + .toBinaryRow(GenericRow.of(BinaryString.fromString("p1"))), + 0, + new BundleRecords() { + @Override + public long rowCount() { + return writtenRows.size(); + } + + @Override + public Iterator iterator() { + return writtenRows.iterator(); + } + }); + commit.commit(write.prepareCommit()); + } + + if (merged) { + BatchWriteBuilder mergeWriteBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = + mergeWriteBuilder + .newWrite() + .withWriteType(table.rowType().project("f2")); + BatchTableCommit commit = mergeWriteBuilder.newCommit()) { + write.writeBundle( + new InternalRowSerializer(table.schema().logicalPartitionType()) + .toBinaryRow(GenericRow.of(BinaryString.fromString("p1"))), + 0, + new BundleRecords() { + @Override + public long rowCount() { + return ROW_COUNT; + } + + @Override + public Iterator iterator() { + return writtenRows.stream() + .map(row -> (InternalRow) GenericRow.of(row.getString(0))) + .iterator(); + } + }); + List messages = write.prepareCommit(); + setFirstRowId(messages, 0L); + commit.commit(messages); + } + } + + FileStoreTable latest = getTable(identifier); + RowType readType = + latest.rowType() + .project( + merged + ? new String[] {"f2", "pt", "f1"} + : new String[] {"pt", "f1"}); + Predicate filter = + new PredicateBuilder(latest.rowType()).equal(1, BinaryString.fromString(f1(50))); + for (List ranges : Arrays.>asList(null, fullRowRanges())) { + ReadBuilder readBuilder = + latest.newReadBuilder().withReadType(readType).withFilter(filter); + if (ranges != null) { + readBuilder.withRowRanges(ranges); + } + List rows = + collect( + readBuilder.newRead().executeFilter(), + readBuilder.newScan().plan(), + readType); + + InternalRow row = assertSingleRow(rows); + assertThat(row.getString(readType.getFieldIndex("pt")).toString()).isEqualTo("p1"); + assertThat(row.getString(readType.getFieldIndex("f1")).toString()).isEqualTo(f1(50)); + if (merged) { + assertThat(row.getString(0).toString()).isEqualTo(f1(50)); + } + } + + RowType partitionOnly = latest.rowType().project("pt"); + ReadBuilder partitionReadBuilder = latest.newReadBuilder().withReadType(partitionOnly); + List partitions = + collect( + partitionReadBuilder.newRead(), + partitionReadBuilder.newScan().plan(), + partitionOnly); + assertThat(partitions).hasSize(ROW_COUNT); + assertThat(partitions) + .allSatisfy(row -> assertThat(row.getString(0).toString()).isEqualTo("p1")); + } + + // Deleting the data files proves the rows come from the row sidecars: without them the read + // would fail on the missing files. + @Test + public void testMergedGroupFinalSelectionChoosesRowSidecar() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("merged_bitmap_dv_row_sidecar", options); + writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"), Collections.emptyMap()); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter, null)), 50); + } + + @Test + public void testSingleFileFinalSelectionChoosesRowSidecar() throws Exception { + Map options = bitmapOptions("f1"); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("single_bitmap_dv_row_sidecar", options); + writeAllColumns(table, ROW_COUNT); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter, null)), 50); + } + + @Test + public void testMergedGroupFinalSelectionChoosesRowSidecarWithRowRanges() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("merged_bitmap_dv_row_sidecar_ranges", options); + writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"), Collections.emptyMap()); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter, fullRowRanges())), 50); + } + + @Test + public void testSingleFileFinalSelectionChoosesRowSidecarWithRowRanges() throws Exception { + Map options = bitmapOptions("f1"); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("single_bitmap_dv_row_sidecar_ranges", options); + writeAllColumns(table, ROW_COUNT); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow(assertSingleRow(readAfterDeletingDataFiles(table, filter, fullRowRanges())), 50); + } + + private static List fullRowRanges() { + return Collections.singletonList(new Range(0L, ROW_COUNT - 1L)); + } + + // The f0/f1 file is read from its row sidecar (its data file is deleted), the f2 file has no + // sidecar and is read from its data file. + @Test + public void testMergedGroupMixedSidecarAndDataFileRead() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("merged_mixed_sidecar_data_file", options); + Map withoutSidecar = new HashMap<>(); + withoutSidecar.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "false"); + writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"), withoutSidecar); + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + assertRow( + assertSingleRow( + readAfterDeletingDataFiles( + table, + filter, + null, + file -> + file.writeCols() != null + && file.writeCols().contains("f1"))), + 50); + } + + // The first file is read from its row sidecar (its data file is deleted), the second file has + // no sidecar and is read from its data file. + @Test + public void testSingleFileMixedSidecarAndDataFileRead() throws Exception { + Map options = bitmapOptions("f1"); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "true"); + FileStoreTable table = createTable("single_mixed_sidecar_data_file", options); + writeAllColumns(table, ROW_COUNT); + Map withoutSidecar = new HashMap<>(bitmapOptions("f1")); + withoutSidecar.put(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED.key(), "false"); + writeAllColumns(getTable(identifier(table.name())).copy(withoutSidecar), ROW_COUNT); + // deleting the overlapping rows leaves the first file with a single-row bitmap selection, + // sparse enough to be read from its sidecar, while the second file has no sidecar at all + deleteRows(table, 51, 52, 53, 54, 55); + + Predicate filter = equalAnyF1(50, 51, 52, 53, 54, 55); + List rows = + readAfterDeletingDataFiles( + table, + filter, + null, + file -> file.firstRowId() != null && file.firstRowId() == 0L); + // row 50 of the first file comes from its sidecar, rows 50-55 of the second file come + // from its data file + assertThat(rows).hasSize(7); + for (InternalRow row : rows) { + int index = row.getInt(0); + assertThat(index).isBetween(50, 55); + assertThat(row.getString(1).toString()).isEqualTo(f1(index)); + assertThat(row.getString(2).toString()).isEqualTo(f2(index)); + } + + // the same mixed read with row tracking: the row id of the sidecar row comes from the + // manifest entry, the row ids of the data file rows from its own range + RowType readType = + rowTypeWithRowId(getTable(identifier(table.name())).rowType()) + .project(SpecialFields.ROW_ID.name(), "f0", "f1"); + List trackedRows = readWithFilter(table, filter, readType); + assertThat(trackedRows).hasSize(7); + assertThat(trackedRows.stream().map(row -> row.getLong(0))) + .containsExactlyInAnyOrder(50L, 150L, 151L, 152L, 153L, 154L, 155L); + } + + // Ordinary row-format data files use the same physical schema mapping as row sidecars. The + // row format has no format-level filter pushdown, so the bitmap index provides the selection. + @Test + public void testOrdinaryRowFormatDataFile() throws Exception { + Map options = bitmapOptions("f1"); + options.put(CoreOptions.FILE_FORMAT.key(), "row"); + FileStoreTable table = createTable("ordinary_row_format_file", options); + writeAllColumns(table, ROW_COUNT); + + assertRow(assertSingleRow(readWithFilter(table, equalF1(f1(50)), null)), 50); + + RowType readType = + rowTypeWithRowId(table.rowType()).project(SpecialFields.ROW_ID.name(), "f1"); + List rows = readWithFilter(table, equalF1(f1(50)), readType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(0)).isEqualTo(50L); + assertThat(rows.get(0).getString(1).toString()).isEqualTo(f1(50)); + } + @Test public void testFormatReaderCacheKeyIncludesPhysicalSchema() throws Exception { Map options = new HashMap<>(); @@ -1212,6 +1553,53 @@ private List readWithFilter(FileStoreTable table, Predicate predica return readWithFilter(table, predicate, null); } + private List readAfterDeletingDataFiles( + FileStoreTable table, Predicate predicate, @Nullable List rowRanges) + throws Exception { + return readAfterDeletingDataFiles(table, predicate, rowRanges, null); + } + + /** + * Reads after deleting the data files. A non-null delete filter restricts the deletion to the + * matching files, so part of the split is read from the row sidecars and the rest from the + * remaining data files. + */ + private List readAfterDeletingDataFiles( + FileStoreTable table, + Predicate predicate, + @Nullable List rowRanges, + @Nullable java.util.function.Predicate deleteFilter) + throws Exception { + FileStoreTable latest = getTable(identifier(table.name())); + ReadBuilder readBuilder = latest.newReadBuilder().withFilter(predicate); + if (rowRanges != null) { + readBuilder.withRowRanges(rowRanges); + } + TableScan.Plan plan = readBuilder.newScan().plan(); + // with row ranges the scan wraps the split into an IndexedSplit + Split first = plan.splits().get(0); + DataSplit split = + first instanceof IndexedSplit + ? ((IndexedSplit) first).dataSplit() + : (DataSplit) first; + DataFilePathFactory pathFactory = + latest.store() + .pathFactory() + .createDataFilePathFactory(split.partition(), split.bucket()); + for (DataFileMeta file : split.dataFiles()) { + if (deleteFilter == null || deleteFilter.test(file)) { + assertThat(latest.fileIO().delete(pathFactory.toPath(file), false)).isTrue(); + } + } + + List rows = new ArrayList<>(); + InternalRowSerializer serializer = new InternalRowSerializer(latest.rowType()); + try (RecordReader reader = readBuilder.newRead().createReader(plan)) { + reader.forEachRemaining(row -> rows.add(serializer.copy(row))); + } + return rows; + } + private List readWithFilter( FileStoreTable table, Predicate predicate, @Nullable RowType readType) throws Exception { @@ -1367,6 +1755,14 @@ private static Predicate equalF1(String value) { return new PredicateBuilder(rowType()).equal(1, BinaryString.fromString(value)); } + private static Predicate equalAnyF1(int... values) { + Predicate[] predicates = new Predicate[values.length]; + for (int i = 0; i < values.length; i++) { + predicates[i] = equalF1(f1(values[i])); + } + return PredicateBuilder.or(predicates); + } + private static Predicate equalF2(String value) { return new PredicateBuilder(rowType()).equal(2, BinaryString.fromString(value)); } From bab496f51ff47df19fe462a2b53d04d576e536e5 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Thu, 15 Oct 2026 13:02:52 +0800 Subject: [PATCH 7/7] chore: trigger ci