From 497ff3442fa31b832f096076ed8404dc14bbf77a Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Sat, 24 Oct 2026 16:27:43 +0800 Subject: [PATCH 1/3] [core] Consider deletion vectors when selecting row sidecars --- .../operation/DataEvolutionSplitRead.java | 145 ++++++++++++++++-- .../operation/DataEvolutionSplitReadTest.java | 73 +++++++++ 2 files changed, 209 insertions(+), 9 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 953f73f93508..cda3a34ee7e1 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 @@ -83,6 +83,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; @@ -543,7 +544,12 @@ private RecordReader createUnionReader( FieldBunch bunch = fieldsFiles.get(i); DataFileMeta firstFile = bunch.files().get(0); FileReadTarget readTarget = - readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); + readTarget( + firstFile, + dataFilePathFactory, + rowRanges, + groupSelection, + deletionVector); String formatIdentifier = readTarget.formatIdentifier; Builder targetFormatBuilder = readTarget.isPositionalRowFormat() @@ -600,7 +606,8 @@ private RecordReader createMissingFieldsReader( throws IOException { DataFileMeta firstFile = bunch.files().get(0); FileReadTarget readTarget = - readTarget(firstFile, dataFilePathFactory, rowRanges, groupSelection); + readTarget( + firstFile, dataFilePathFactory, rowRanges, groupSelection, deletionVector); Builder formatBuilder = readTarget.isPositionalRowFormat() ? positionalRowFormatBuilder(readRowType, null, nestedFieldEnabled) @@ -845,7 +852,7 @@ private FileRecordReader createFileReader( } FileReadTarget readTarget = - readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult); + readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult, deletionVector); String formatIdentifier = readTarget.formatIdentifier; boolean positionalRowFormat = readTarget.isPositionalRowFormat(); Builder targetFormatBuilder = @@ -915,7 +922,7 @@ private FileRecordReader createFileReader( formatReaderMapping, rowRanges, readRowType, - readTarget(file, dataFilePathFactory, rowRanges, groupSelection), + readTarget(file, dataFilePathFactory, rowRanges, groupSelection, deletionVector), deletionVector, groupSelection, metadataCache); @@ -1530,7 +1537,8 @@ private FileReadTarget readTarget( DataFileMeta file, DataFilePathFactory dataFilePathFactory, List rowRanges, - @Nullable FileIndexResult fileIndexResult) + @Nullable FileIndexResult fileIndexResult, + @Nullable DeletionVectorWithRange deletionVector) throws IOException { String rowSidecar = rowSidecarFileName(file); if (rowSidecar != null @@ -1538,6 +1546,7 @@ && shouldReadRowSidecar( file, rowRanges, fileIndexResult, + deletionVector, coreOptions.dataEvolutionRowSidecarMaxSelectedRows(), coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) { Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file); @@ -1572,6 +1581,7 @@ static boolean shouldReadRowSidecar(DataFileMeta file, @Nullable List row file, rowRanges, null, + (DeletionVector) null, CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(), CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue()); } @@ -1582,7 +1592,41 @@ static boolean shouldReadRowSidecar( @Nullable List rowRanges, long maxSelectedRows, double maxSelectionRatio) { - return shouldReadRowSidecar(file, rowRanges, null, maxSelectedRows, maxSelectionRatio); + return shouldReadRowSidecar( + file, rowRanges, null, (DeletionVector) null, maxSelectedRows, maxSelectionRatio); + } + + @VisibleForTesting + static boolean shouldReadRowSidecar( + DataFileMeta file, + @Nullable List rowRanges, + @Nullable FileIndexResult fileIndexResult, + long maxSelectedRows, + double maxSelectionRatio) { + return shouldReadRowSidecar( + file, + rowRanges, + fileIndexResult, + (DeletionVector) null, + maxSelectedRows, + maxSelectionRatio); + } + + private static boolean shouldReadRowSidecar( + DataFileMeta file, + @Nullable List rowRanges, + @Nullable FileIndexResult fileIndexResult, + @Nullable DeletionVectorWithRange deletionVector, + long maxSelectedRows, + double maxSelectionRatio) { + return shouldReadRowSidecar( + file, + rowRanges, + fileIndexResult, + deletionVector == null ? null : deletionVector.deletionVector, + deletionVector == null ? null : deletionVector.range, + maxSelectedRows, + maxSelectionRatio); } @VisibleForTesting @@ -1590,6 +1634,25 @@ static boolean shouldReadRowSidecar( DataFileMeta file, @Nullable List rowRanges, @Nullable FileIndexResult fileIndexResult, + @Nullable DeletionVector deletionVector, + long maxSelectedRows, + double maxSelectionRatio) { + return shouldReadRowSidecar( + file, + rowRanges, + fileIndexResult, + deletionVector, + null, + maxSelectedRows, + maxSelectionRatio); + } + + private static boolean shouldReadRowSidecar( + DataFileMeta file, + @Nullable List rowRanges, + @Nullable FileIndexResult fileIndexResult, + @Nullable DeletionVector deletionVector, + @Nullable Range deletionVectorRange, long maxSelectedRows, double maxSelectionRatio) { if (file.rowCount() <= 0 @@ -1599,7 +1662,9 @@ static boolean shouldReadRowSidecar( return false; } - long selectedRowCount = selectedRowCount(file, rowRanges, fileIndexResult); + long selectedRowCount = + selectedRowCount( + file, rowRanges, fileIndexResult, deletionVector, deletionVectorRange); if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) { return false; } @@ -1623,10 +1688,72 @@ static long selectedRowCount(DataFileMeta file, List rowRanges) { .sum(); } - private static long selectedRowCount( + @VisibleForTesting + static long selectedRowCount( DataFileMeta file, @Nullable List rowRanges, - @Nullable FileIndexResult fileIndexResult) { + @Nullable FileIndexResult fileIndexResult, + @Nullable DeletionVector deletionVector, + @Nullable Range deletionVectorRange) { + if (deletionVector != null && !deletionVector.isEmpty()) { + RoaringBitmap32 selection = file.toFileSelection(rowRanges); + if (fileIndexResult instanceof BitmapIndexResult) { + RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get(); + selection = + selection == null + ? indexSelection.clone() + : RoaringBitmap32.and(selection, indexSelection); + } + long rangeStart = + deletionVectorRange == null + ? file.nonNullFirstRowId() + : deletionVectorRange.from; + if (selection == null) { + if (deletionVectorRange != null + && deletionVectorRange.equals(file.nonNullRowIdRange())) { + return file.rowCount() - deletionVector.getCardinality(); + } + // Iterate the deletion vector instead of every row of the file: the + // deletion vector is usually far smaller than the row count. A deleted + // position outside this file's range simply does not count. + long firstRowId = file.nonNullFirstRowId(); + long rowCount = file.rowCount(); + long[] deletedInRange = new long[1]; + deletionVector.forEachDeletedPosition( + deletedPosition -> { + long position = deletedPosition - firstRowId + rangeStart; + if (position >= 0 && position < rowCount) { + deletedInRange[0]++; + } + }); + return rowCount - deletedInRange[0]; + } + long selected = selection.getCardinality(); + long deleted = deletionVector.getCardinality(); + long offset = file.nonNullFirstRowId() - rangeStart; + if (selected <= deleted) { + long live = 0; + Iterator iterator = selection.iterator(); + while (iterator.hasNext()) { + if (!deletionVector.isDeleted(iterator.next() + offset)) { + live++; + } + } + return live; + } + RoaringBitmap32 finalSelection = selection; + long[] deletedInSelection = new long[1]; + deletionVector.forEachDeletedPosition( + deletedPosition -> { + long localPosition = deletedPosition - offset; + if (localPosition >= 0 + && localPosition <= Integer.MAX_VALUE + && finalSelection.contains((int) localPosition)) { + deletedInSelection[0]++; + } + }); + return selected - deletedInSelection[0]; + } if (fileIndexResult instanceof BitmapIndexResult) { RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get(); RoaringBitmap32 rangeSelection = file.toFileSelection(rowRanges); 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 2a1c5c4472eb..7ac28b932cc5 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.deletionvectors.BitmapDeletionVector; import org.apache.paimon.fileindex.bitmap.BitmapIndexResult; import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.FormatWriter; @@ -53,6 +54,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.function.LongConsumer; import static org.apache.paimon.data.BinaryRow.EMPTY_ROW; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -253,6 +255,77 @@ public void testShouldReadRowSidecarUsesFinalBitmapSelection() { DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.01d)); } + @Test + public void testShouldReadRowSidecarUsesDvLiveRowsWithoutFileIndex() { + DataFileMeta file = + createFile("file1.parquet", 0L, 20_000, 1, Collections.singletonList("file1.row")); + BitmapDeletionVector deletionVector = new BitmapDeletionVector(); + for (int i = 0; i < 19_000; i++) { + deletionVector.delete(i); + } + + assertTrue( + DataEvolutionSplitRead.shouldReadRowSidecar( + file, null, null, deletionVector, 20_000L, 1.0d)); + } + + @Test + public void testSelectedRowCountUsesDeletionVectorAnchorRange() { + // The file starts at row 100, while the DV positions are relative to an anchor at row 0. + DataFileMeta file = + createFile("file1.parquet", 100L, 100, 1, Collections.singletonList("file1.row")); + BitmapDeletionVector deletionVector = new BitmapDeletionVector(); + deletionVector.delete(100L); + deletionVector.delete(150L); + deletionVector.delete(250L); // outside the file range + + assertEquals( + 98L, + DataEvolutionSplitRead.selectedRowCount( + file, null, null, deletionVector, new Range(0L, 299L))); + } + + @Test + public void testSelectedRowCountUsesDvCardinalityForMatchingAnchorRange() { + DataFileMeta file = + createFile("file1.parquet", 100L, 100, 1, Collections.singletonList("file1.row")); + BitmapDeletionVector deletionVector = + new BitmapDeletionVector() { + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + throw new AssertionError("matching anchor range should use cardinality"); + } + }; + deletionVector.delete(0L); + deletionVector.delete(50L); + + assertEquals( + 98L, + DataEvolutionSplitRead.selectedRowCount( + file, null, null, deletionVector, file.nonNullRowIdRange())); + } + + @Test + public void testSelectedRowCountIteratesSparseDvInsteadOfSelection() { + DataFileMeta file = + createFile("file1.parquet", 0L, 10_000, 1, Collections.singletonList("file1.row")); + BitmapIndexResult bitmap = + new BitmapIndexResult(() -> RoaringBitmap32.bitmapOfRange(0L, 10_000L)); + BitmapDeletionVector deletionVector = + new BitmapDeletionVector() { + @Override + public boolean isDeleted(long position) { + throw new AssertionError("sparse DV should not probe every selected row"); + } + }; + deletionVector.delete(5L); + + assertEquals( + 9_999L, + DataEvolutionSplitRead.selectedRowCount( + file, null, bitmap, deletionVector, file.nonNullRowIdRange())); + } + @Test public void testShouldNotReadRowSidecarWithoutSparseSelection() { DataFileMeta file = From 9c94a4768b397e573a7988d7fbe168b249aba24f Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Sat, 24 Oct 2026 18:28:00 +0800 Subject: [PATCH 2/3] chore: trigger ci From 9dffc5dd4a14ae45dc8dd4a31dc5e587538eecdf Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Mon, 26 Oct 2026 11:45:15 +0800 Subject: [PATCH 3/3] push final selection to row sidecar reader --- .../operation/DataEvolutionSplitRead.java | 355 ++++++++---------- .../operation/DataEvolutionSplitReadTest.java | 332 ++++++++++------ 2 files changed, 366 insertions(+), 321 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 cda3a34ee7e1..7eb5026e104a 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 @@ -28,6 +28,7 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.PartitionInfo; import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader; +import org.apache.paimon.deletionvectors.BitmapDeletionVector; import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.disk.IOManager; import org.apache.paimon.fileindex.FileIndexResult; @@ -761,20 +762,23 @@ private RecordReader sequentialReadFiles( List> readerSuppliers = new ArrayList<>(); for (DataFileMeta file : files) { readerSuppliers.add( - () -> - createFileReader( - partition, - file, - formatReaderMapping, - rowRanges, - readRowType, - new FileReadTarget( - DataFilePathFactory.formatIdentifier(file.fileName()), - dataFilePathFactory.toPath(file), - file.fileSize()), - deletionVector, - groupSelection, - null)); + () -> { + RoaringBitmap32 selection = fileSelection(file, rowRanges, groupSelection); + return createFileReader( + partition, + file, + formatReaderMapping, + rowRanges, + readRowType, + new FileReadTarget( + DataFilePathFactory.formatIdentifier(file.fileName()), + dataFilePathFactory.toPath(file), + file.fileSize(), + selection), + deletionVector, + groupSelection, + null); + }); } return ConcatRecordReader.create(readerSuppliers); } @@ -939,20 +943,15 @@ private FileRecordReader createFileReader( @Nullable FileIndexResult fileIndexResult, @Nullable FileMetadataCache metadataCache) throws IOException { - RoaringBitmap32 selection = file.toFileSelection(rowRanges); + RoaringBitmap32 selection = readTarget.selection; BitmapIndexResult bitmapIndexResult = fileIndexResult instanceof BitmapIndexResult ? (BitmapIndexResult) fileIndexResult : null; - if (bitmapIndexResult != null) { - RoaringBitmap32 indexSelection = bitmapIndexResult.get(); - selection = - selection == null - ? indexSelection.clone() - : RoaringBitmap32.and(selection, indexSelection); - if (selection.isEmpty()) { - return new EmptyFileRecordReader<>(); - } + if (selection != null && selection.isEmpty()) { + // Data-file targets never include DV-only filtering in their selection, so a file + // fully deleted by its DV still reaches applyDeletionVector below. + return new EmptyFileRecordReader<>(); } FormatReaderContext formatReaderContext = @@ -1540,119 +1539,54 @@ private FileReadTarget readTarget( @Nullable FileIndexResult fileIndexResult, @Nullable DeletionVectorWithRange deletionVector) throws IOException { + RoaringBitmap32 selection = fileSelection(file, rowRanges, fileIndexResult); String rowSidecar = rowSidecarFileName(file); - if (rowSidecar != null - && shouldReadRowSidecar( - file, - rowRanges, - fileIndexResult, - deletionVector, - coreOptions.dataEvolutionRowSidecarMaxSelectedRows(), - coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) { - Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file); - try { - return new FileReadTarget( - ROW_SIDECAR_FORMAT, rowPath, fileIO.getFileStatus(rowPath).getLen()); - } catch (IOException e) { - if (!coreOptions.scanIgnoreLostFile()) { - throw e; + if (rowSidecar != null) { + RoaringBitmap32 rowSidecarSelection = + sidecarSelection( + file, + selection, + deletionVector == null ? null : deletionVector.deletionVector, + deletionVector == null ? null : deletionVector.range); + if (shouldReadRowSidecar( + file, + rowSidecarSelection, + coreOptions.dataEvolutionRowSidecarMaxSelectedRows(), + coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) { + Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file); + try { + return new FileReadTarget( + ROW_SIDECAR_FORMAT, + rowPath, + fileIO.getFileStatus(rowPath).getLen(), + rowSidecarSelection); + } catch (IOException e) { + if (!coreOptions.scanIgnoreLostFile()) { + throw e; + } } } } return new FileReadTarget( DataFilePathFactory.formatIdentifier(file.fileName()), dataFilePathFactory.toPath(file), - file.fileSize()); - } - - @Nullable - @VisibleForTesting - static String rowSidecarFileName(DataFileMeta file) { - List rowFiles = - file.extraFiles().stream() - .filter(DataEvolutionSplitRead::isRowSidecarFile) - .collect(Collectors.toList()); - return rowFiles.size() == 1 ? rowFiles.get(0) : null; - } - - @VisibleForTesting - static boolean shouldReadRowSidecar(DataFileMeta file, @Nullable List rowRanges) { - return shouldReadRowSidecar( - file, - rowRanges, - null, - (DeletionVector) null, - CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(), - CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue()); - } - - @VisibleForTesting - static boolean shouldReadRowSidecar( - DataFileMeta file, - @Nullable List rowRanges, - long maxSelectedRows, - double maxSelectionRatio) { - return shouldReadRowSidecar( - file, rowRanges, null, (DeletionVector) null, maxSelectedRows, maxSelectionRatio); - } - - @VisibleForTesting - static boolean shouldReadRowSidecar( - DataFileMeta file, - @Nullable List rowRanges, - @Nullable FileIndexResult fileIndexResult, - long maxSelectedRows, - double maxSelectionRatio) { - return shouldReadRowSidecar( - file, - rowRanges, - fileIndexResult, - (DeletionVector) null, - maxSelectedRows, - maxSelectionRatio); - } - - private static boolean shouldReadRowSidecar( - DataFileMeta file, - @Nullable List rowRanges, - @Nullable FileIndexResult fileIndexResult, - @Nullable DeletionVectorWithRange deletionVector, - long maxSelectedRows, - double maxSelectionRatio) { - return shouldReadRowSidecar( - file, - rowRanges, - fileIndexResult, - deletionVector == null ? null : deletionVector.deletionVector, - deletionVector == null ? null : deletionVector.range, - maxSelectedRows, - maxSelectionRatio); + file.fileSize(), + selection); } @VisibleForTesting static boolean shouldReadRowSidecar( DataFileMeta file, - @Nullable List rowRanges, - @Nullable FileIndexResult fileIndexResult, - @Nullable DeletionVector deletionVector, + @Nullable RoaringBitmap32 selection, long maxSelectedRows, double maxSelectionRatio) { - return shouldReadRowSidecar( - file, - rowRanges, - fileIndexResult, - deletionVector, - null, - maxSelectedRows, - maxSelectionRatio); + long selectedRowCount = selection == null ? file.rowCount() : selection.getCardinality(); + return shouldReadRowSidecar(file, selectedRowCount, maxSelectedRows, maxSelectionRatio); } private static boolean shouldReadRowSidecar( DataFileMeta file, - @Nullable List rowRanges, - @Nullable FileIndexResult fileIndexResult, - @Nullable DeletionVector deletionVector, - @Nullable Range deletionVectorRange, + long selectedRowCount, long maxSelectedRows, double maxSelectionRatio) { if (file.rowCount() <= 0 @@ -1662,9 +1596,6 @@ private static boolean shouldReadRowSidecar( return false; } - long selectedRowCount = - selectedRowCount( - file, rowRanges, fileIndexResult, deletionVector, deletionVectorRange); if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) { return false; } @@ -1673,95 +1604,115 @@ private static boolean shouldReadRowSidecar( return selectedRowCount <= maxSelectedRows && selectionRatio <= maxSelectionRatio; } + @Nullable @VisibleForTesting - static long selectedRowCount(DataFileMeta file, List rowRanges) { - Range fileRange = file.nonNullRowIdRange(); - List intersections = new ArrayList<>(); - for (Range range : rowRanges) { - Range intersection = Range.intersection(fileRange, range); - if (intersection != null) { - intersections.add(intersection); - } - } - return Range.sortAndMergeOverlap(intersections, true).stream() - .mapToLong(Range::count) - .sum(); + static String rowSidecarFileName(DataFileMeta file) { + List rowFiles = + file.extraFiles().stream() + .filter(DataEvolutionSplitRead::isRowSidecarFile) + .collect(Collectors.toList()); + return rowFiles.size() == 1 ? rowFiles.get(0) : null; } + /** + * Builds the rows read from one physical file in file-local positions. Null means every row. + */ + @Nullable @VisibleForTesting - static long selectedRowCount( + static RoaringBitmap32 fileSelection( DataFileMeta file, @Nullable List rowRanges, - @Nullable FileIndexResult fileIndexResult, + @Nullable FileIndexResult fileIndexResult) { + RoaringBitmap32 selection = file.toFileSelection(rowRanges); + if (fileIndexResult instanceof BitmapIndexResult) { + RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get(); + selection = + selection == null + ? indexSelection.clone() + : RoaringBitmap32.and(selection, indexSelection); + } + return selection; + } + + @Nullable + @VisibleForTesting + static RoaringBitmap32 sidecarSelection( + DataFileMeta file, + @Nullable RoaringBitmap32 selection, @Nullable DeletionVector deletionVector, @Nullable Range deletionVectorRange) { - if (deletionVector != null && !deletionVector.isEmpty()) { - RoaringBitmap32 selection = file.toFileSelection(rowRanges); - if (fileIndexResult instanceof BitmapIndexResult) { - RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get(); - selection = - selection == null - ? indexSelection.clone() - : RoaringBitmap32.and(selection, indexSelection); - } - long rangeStart = - deletionVectorRange == null - ? file.nonNullFirstRowId() - : deletionVectorRange.from; - if (selection == null) { - if (deletionVectorRange != null - && deletionVectorRange.equals(file.nonNullRowIdRange())) { - return file.rowCount() - deletionVector.getCardinality(); - } - // Iterate the deletion vector instead of every row of the file: the - // deletion vector is usually far smaller than the row count. A deleted - // position outside this file's range simply does not count. - long firstRowId = file.nonNullFirstRowId(); - long rowCount = file.rowCount(); - long[] deletedInRange = new long[1]; - deletionVector.forEachDeletedPosition( - deletedPosition -> { - long position = deletedPosition - firstRowId + rangeStart; - if (position >= 0 && position < rowCount) { - deletedInRange[0]++; - } - }); - return rowCount - deletedInRange[0]; + if (deletionVector == null + || deletionVector.isEmpty() + || (selection != null && selection.isEmpty())) { + return selection; + } + if (file.rowCount() > RoaringBitmap32.MAX_VALUE) { + // A 32-bit selection cannot represent all positions in this file. Keep the existing + // range/index selection and let ApplyDeletionVectorReader enforce the DV. This + // conservatively ignores the DV when deciding whether a row sidecar is selective. + return selection; + } + return liveSelection(file, selection, deletionVector, deletionVectorRange); + } + + private static RoaringBitmap32 liveSelection( + DataFileMeta file, + @Nullable RoaringBitmap32 selection, + DeletionVector deletionVector, + @Nullable Range deletionVectorRange) { + long rangeStart = + deletionVectorRange == null ? file.nonNullFirstRowId() : deletionVectorRange.from; + long offset = file.nonNullFirstRowId() - rangeStart; + + if (offset == 0 && deletionVector instanceof BitmapDeletionVector) { + RoaringBitmap32 deleted = ((BitmapDeletionVector) deletionVector).get(); + if (!deleted.intersects(0, file.rowCount())) { + return selection; } - long selected = selection.getCardinality(); - long deleted = deletionVector.getCardinality(); - long offset = file.nonNullFirstRowId() - rangeStart; - if (selected <= deleted) { - long live = 0; - Iterator iterator = selection.iterator(); - while (iterator.hasNext()) { - if (!deletionVector.isDeleted(iterator.next() + offset)) { - live++; - } + return RoaringBitmap32.andNot( + selection == null + ? RoaringBitmap32.bitmapOfRange(0, file.rowCount()) + : selection, + deleted); + } + + if (selection != null && selection.getCardinality() <= deletionVector.getCardinality()) { + RoaringBitmap32 live = new RoaringBitmap32(); + Iterator iterator = selection.iterator(); + while (iterator.hasNext()) { + int localPosition = iterator.next(); + long deletionPosition = localPosition + offset; + if (deletionPosition < 0 + || (deletionVector instanceof BitmapDeletionVector + && deletionPosition > RoaringBitmap32.MAX_VALUE) + || !deletionVector.isDeleted(deletionPosition)) { + live.add(localPosition); } - return live; } - RoaringBitmap32 finalSelection = selection; - long[] deletedInSelection = new long[1]; - deletionVector.forEachDeletedPosition( - deletedPosition -> { - long localPosition = deletedPosition - offset; - if (localPosition >= 0 - && localPosition <= Integer.MAX_VALUE - && finalSelection.contains((int) localPosition)) { - deletedInSelection[0]++; - } - }); - return selected - deletedInSelection[0]; + return live; } - if (fileIndexResult instanceof BitmapIndexResult) { - RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get(); - RoaringBitmap32 rangeSelection = file.toFileSelection(rowRanges); - return rangeSelection == null - ? indexSelection.getCardinality() - : RoaringBitmap32.and(rangeSelection, indexSelection).getCardinality(); + + RoaringBitmap32 deleted = deletedPositions(file, deletionVector, offset); + if (deleted.isEmpty()) { + return selection; } - return isNullOrEmpty(rowRanges) ? file.rowCount() : selectedRowCount(file, rowRanges); + return RoaringBitmap32.andNot( + selection == null ? RoaringBitmap32.bitmapOfRange(0, file.rowCount()) : selection, + deleted); + } + + private static RoaringBitmap32 deletedPositions( + DataFileMeta file, DeletionVector deletionVector, long offset) { + long rowCount = file.rowCount(); + RoaringBitmap32 deleted = new RoaringBitmap32(); + deletionVector.forEachDeletedPosition( + position -> { + long localPosition = position - offset; + if (localPosition >= 0 && localPosition < rowCount) { + deleted.add((int) localPosition); + } + }); + return deleted; } private static boolean isRowSidecarFile(String fileName) { @@ -1777,11 +1728,17 @@ private static class FileReadTarget { private final String formatIdentifier; private final Path path; private final long fileSize; + @Nullable private final RoaringBitmap32 selection; - private FileReadTarget(String formatIdentifier, Path path, long fileSize) { + private FileReadTarget( + String formatIdentifier, + Path path, + long fileSize, + @Nullable RoaringBitmap32 selection) { this.formatIdentifier = formatIdentifier; this.path = path; this.fileSize = fileSize; + this.selection = selection; } /** 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 7ac28b932cc5..8e29490970b8 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,16 +23,21 @@ import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.deletionvectors.Bitmap64DeletionVector; import org.apache.paimon.deletionvectors.BitmapDeletionVector; +import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.fileindex.bitmap.BitmapIndexResult; import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatReaderContext; import org.apache.paimon.format.FormatWriter; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.FileSystemSchemaManager; @@ -40,6 +45,7 @@ import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.DeletionFile; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.FileStorePathFactory; @@ -48,13 +54,16 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.io.DataOutputStream; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.function.LongConsumer; +import java.util.concurrent.atomic.AtomicLong; import static org.apache.paimon.data.BinaryRow.EMPTY_ROW; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -197,16 +206,6 @@ public void testRowSidecarFileNameWithNoOrAmbiguousSidecar() { assertNull(DataEvolutionSplitRead.rowSidecarFileName(ambiguousSidecar)); } - @Test - public void testShouldReadRowSidecarForSparseRowSelection() { - DataFileMeta file = - createFile("file1.parquet", 10L, 100, 1, Collections.singletonList("file1.row")); - - assertTrue( - DataEvolutionSplitRead.shouldReadRowSidecar( - file, Arrays.asList(new Range(10L, 10L), new Range(42L, 42L)))); - } - @Test public void testShouldReadRowSidecarRequiresSmallCountAndLowRatio() { DataFileMeta smallFile = @@ -217,113 +216,37 @@ public void testShouldReadRowSidecarRequiresSmallCountAndLowRatio() { assertTrue( DataEvolutionSplitRead.shouldReadRowSidecar( - largeFile, Collections.singletonList(new Range(10L, 4105L)))); + largeFile, RoaringBitmap32.bitmapOfRange(0L, 4096L), 4096L, 0.05d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar( - smallFile, Collections.singletonList(new Range(10L, 15L)))); + smallFile, RoaringBitmap32.bitmapOfRange(0L, 6L), 4096L, 0.05d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar( - largeFile, Collections.singletonList(new Range(10L, 4106L)))); + largeFile, RoaringBitmap32.bitmapOfRange(0L, 4097L), 4096L, 0.05d)); assertTrue( DataEvolutionSplitRead.shouldReadRowSidecar( - smallFile, Collections.singletonList(new Range(10L, 15L)), 64L, 0.25d)); + smallFile, RoaringBitmap32.bitmapOfRange(0L, 6L), 64L, 0.25d)); } @Test - public void testShouldReadRowSidecarUsesFinalBitmapSelection() { + public void testDeletionVectorOnlyAffectsSidecarSelection() { 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)); - 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( - DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges, bitmap, 2L, 0.01d)); - } - - @Test - public void testShouldReadRowSidecarUsesDvLiveRowsWithoutFileIndex() { - DataFileMeta file = - createFile("file1.parquet", 0L, 20_000, 1, Collections.singletonList("file1.row")); - BitmapDeletionVector deletionVector = new BitmapDeletionVector(); - for (int i = 0; i < 19_000; i++) { - deletionVector.delete(i); - } - - assertTrue( - DataEvolutionSplitRead.shouldReadRowSidecar( - file, null, null, deletionVector, 20_000L, 1.0d)); - } - - @Test - public void testSelectedRowCountUsesDeletionVectorAnchorRange() { - // The file starts at row 100, while the DV positions are relative to an anchor at row 0. - DataFileMeta file = - createFile("file1.parquet", 100L, 100, 1, Collections.singletonList("file1.row")); - BitmapDeletionVector deletionVector = new BitmapDeletionVector(); - deletionVector.delete(100L); - deletionVector.delete(150L); - deletionVector.delete(250L); // outside the file range - - assertEquals( - 98L, - DataEvolutionSplitRead.selectedRowCount( - file, null, null, deletionVector, new Range(0L, 299L))); - } - - @Test - public void testSelectedRowCountUsesDvCardinalityForMatchingAnchorRange() { - DataFileMeta file = - createFile("file1.parquet", 100L, 100, 1, Collections.singletonList("file1.row")); - BitmapDeletionVector deletionVector = - new BitmapDeletionVector() { - @Override - public void forEachDeletedPosition(LongConsumer consumer) { - throw new AssertionError("matching anchor range should use cardinality"); - } - }; - deletionVector.delete(0L); - deletionVector.delete(50L); - - assertEquals( - 98L, - DataEvolutionSplitRead.selectedRowCount( - file, null, null, deletionVector, file.nonNullRowIdRange())); - } - - @Test - public void testSelectedRowCountIteratesSparseDvInsteadOfSelection() { - DataFileMeta file = - createFile("file1.parquet", 0L, 10_000, 1, Collections.singletonList("file1.row")); BitmapIndexResult bitmap = - new BitmapIndexResult(() -> RoaringBitmap32.bitmapOfRange(0L, 10_000L)); - BitmapDeletionVector deletionVector = - new BitmapDeletionVector() { - @Override - public boolean isDeleted(long position) { - throw new AssertionError("sparse DV should not probe every selected row"); - } - }; - deletionVector.delete(5L); - + new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(1, 2, 4, 6, 8, 9)); + BitmapDeletionVector deletionVector = new BitmapDeletionVector(); + deletionVector.delete(14L); // file-local position 4 for anchor range starting at 0 + deletionVector.delete(18L); // file-local position 8 + deletionVector.delete(150L); // outside this file + + RoaringBitmap32 fileSelection = + DataEvolutionSplitRead.fileSelection( + file, Collections.singletonList(new Range(12L, 18L)), bitmap); + assertEquals(RoaringBitmap32.bitmapOf(2, 4, 6, 8), fileSelection); assertEquals( - 9_999L, - DataEvolutionSplitRead.selectedRowCount( - file, null, bitmap, deletionVector, file.nonNullRowIdRange())); + RoaringBitmap32.bitmapOf(2, 6), + DataEvolutionSplitRead.sidecarSelection( + file, fileSelection, deletionVector, new Range(0L, 199L))); } @Test @@ -331,32 +254,23 @@ public void testShouldNotReadRowSidecarWithoutSparseSelection() { DataFileMeta file = createFile("file1.parquet", 10L, 100, 1, Collections.singletonList("file1.row")); - assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null)); - assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, Collections.emptyList())); + assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null, 4096L, 0.05d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar( - file, Collections.singletonList(new Range(10L, 109L)))); + file, new RoaringBitmap32(), 4096L, 0.05d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar( createFile("file2.parquet", 10L, 100, 1), - Collections.singletonList(new Range(10L, 10L)))); + RoaringBitmap32.bitmapOf(0), + 4096L, + 0.05d)); assertFalse( DataEvolutionSplitRead.shouldReadRowSidecar( createFile( "file3.blob", 10L, 100, 1, Collections.singletonList("file3.row")), - Collections.singletonList(new Range(10L, 10L)))); - } - - @Test - public void testSelectedRowCountMergesOverlappingRanges() { - DataFileMeta file = createFile("file1.parquet", 10L, 100, 1); - - assertEquals( - 12, - DataEvolutionSplitRead.selectedRowCount( - file, - Arrays.asList( - new Range(5L, 12L), new Range(12L, 15L), new Range(20L, 25L)))); + RoaringBitmap32.bitmapOf(0), + 4096L, + 0.05d)); } @Test @@ -505,6 +419,180 @@ public void testSparseRowIdReadUsesParquetRowsAtSelectedPositions() throws Excep assertEquals(Arrays.asList("1000:[0,1]", "5200:[4200,4201]"), actual); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testDvSelectionNarrowsRowSidecarRead(boolean bitmap64) throws Exception { + LocalFileIO fileIO = new LocalFileIO(); + Path tableRoot = new Path(tempDir.toUri().toString(), "dv-sidecar-" + bitmap64); + CoreOptions coreOptions = new CoreOptions(new Options()); + FileStorePathFactory pathFactory = + new FileStorePathFactory( + tableRoot, + RowType.of(), + coreOptions.partitionDefaultName(), + CoreOptions.FILE_FORMAT.defaultValue(), + CoreOptions.DATA_FILE_PREFIX.defaultValue(), + CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), + CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(), + CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), + CoreOptions.FILE_COMPRESSION.defaultValue(), + null, + null, + CoreOptions.ExternalPathStrategy.NONE, + null, + false, + null); + + Schema schema = + Schema.newBuilder() + .column("f0", DataTypes.INT()) + .column("f1", DataTypes.STRING()) + .build(); + SchemaManager schemaManager = new FileSystemSchemaManager(fileIO, tableRoot); + TableSchema tableSchema = schemaManager.createTable(schema); + RowType rowType = tableSchema.logicalRowType(); + + Path bucketPath = pathFactory.bucketPath(EMPTY_ROW, 0); + fileIO.mkdirs(bucketPath); + int rowCount = 60_000; + int liveRowCount = 500; + String rowSidecarName = "data-0.row"; + Path rowSidecarPath = new Path(bucketPath, rowSidecarName); + Options rowFormatOptions = new Options(); + rowFormatOptions.set(CoreOptions.FILE_BLOCK_SIZE, MemorySize.ofKibiBytes(4)); + writeFormatFile(fileIO, rowSidecarPath, rowType, rowCount, "row", rowFormatOptions); + + DeletionVector deletionVector = + bitmap64 ? new Bitmap64DeletionVector() : new BitmapDeletionVector(); + for (int i = 0; i < rowCount - liveRowCount; i++) { + deletionVector.delete(i); + } + Path dvPath = new Path(bucketPath, "data-0.parquet.dv"); + long dvSize; + try (PositionOutputStream out = fileIO.newOutputStream(dvPath, false); + DataOutputStream dataOutput = new DataOutputStream(out)) { + dvSize = deletionVector.serializeTo(dataOutput); + } + + DataFileMeta dataFile = + createFile( + "data-0.parquet", + 0L, + rowCount, + 1, + Collections.singletonList(rowSidecarName)); + DataSplit split = + DataSplit.builder() + .withPartition(EMPTY_ROW) + .withBucket(0) + .withBucketPath(bucketPath.toString()) + .withDataFiles(Collections.singletonList(dataFile)) + .withDataDeletionFiles( + Collections.singletonList( + new DeletionFile( + dvPath.toString(), + 0, + dvSize, + (long) (rowCount - liveRowCount)))) + .rawConvertible(false) + .build(); + + CountingFileIO countingFileIO = new CountingFileIO(fileIO, rowSidecarPath); + DataEvolutionSplitRead splitRead = + new DataEvolutionSplitRead( + countingFileIO, + schemaManager, + tableSchema, + rowType, + coreOptions, + pathFactory); + + List actual = new ArrayList<>(); + try (RecordReader reader = splitRead.createReader(split)) { + reader.forEachRemaining(row -> actual.add(row.getInt(0))); + } + + assertEquals(liveRowCount, actual.size()); + assertEquals(1000 + rowCount - liveRowCount, (int) actual.get(0)); + assertEquals(1000 + rowCount - 1, (int) actual.get(actual.size() - 1)); + long sidecarSize = fileIO.getFileStatus(rowSidecarPath).getLen(); + + CountingFileIO fullScanFileIO = new CountingFileIO(fileIO, rowSidecarPath); + RecordReader fullScanReader = + FileFormat.fromIdentifier("row", rowFormatOptions) + .createReaderFactory(rowType, rowType, null) + .createReader( + new FormatReaderContext( + fullScanFileIO, rowSidecarPath, sidecarSize, null, null)); + fullScanReader.forEachRemaining(ignored -> {}); + + assertTrue( + countingFileIO.bytesRead() < fullScanFileIO.bytesRead(), + "Selection read " + + countingFileIO.bytesRead() + + " sidecar bytes; full scan read " + + fullScanFileIO.bytesRead() + + "."); + } + + private static final class CountingFileIO extends LocalFileIO { + + private final LocalFileIO fileIO; + private final Path countedPath; + private final AtomicLong bytesRead = new AtomicLong(); + + private CountingFileIO(LocalFileIO fileIO, Path countedPath) { + this.fileIO = fileIO; + this.countedPath = countedPath; + } + + private long bytesRead() { + return bytesRead.get(); + } + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + SeekableInputStream input = fileIO.newInputStream(path); + if (!countedPath.equals(path)) { + return input; + } + return new SeekableInputStream() { + @Override + public void seek(long desired) throws IOException { + input.seek(desired); + } + + @Override + public long getPos() throws IOException { + return input.getPos(); + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + int read = input.read(buffer, offset, length); + if (read > 0) { + bytesRead.addAndGet(read); + } + return read; + } + + @Override + public int read() throws IOException { + int read = input.read(); + if (read >= 0) { + bytesRead.incrementAndGet(); + } + return read; + } + + @Override + public void close() throws IOException { + input.close(); + } + }; + } + } + private static void writeNestedParquetFile( LocalFileIO fileIO, Path path, RowType rowType, int rowCount, Options options) throws IOException {