Skip to content

[core] Consider deletion vectors when selecting row sidecars #10792

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 3 commits into from
Oct 26, 2026
Merged

[core] Consider deletion vectors when selecting row sidecars #10792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
497ff34
[core] Consider deletion vectors when selecting row sidecars
lilei1128 Oct 24, 2026
9c94a47
chore: trigger ci
lilei1128 Oct 24, 2026
9dffc5d
push final selection to row sidecar reader
lilei1128 Oct 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
264 changes: 174 additions & 90 deletions paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -83,6 +84,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;
Expand Down Expand Up @@ -543,7 +545,12 @@ private RecordReader<InternalRow> 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()
Expand Down Expand Up @@ -600,7 +607,8 @@ private RecordReader<InternalRow> 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)
Expand Down Expand Up @@ -754,20 +762,23 @@ private RecordReader<InternalRow> sequentialReadFiles(
List<ReaderSupplier<InternalRow>> 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);
}
Expand Down Expand Up @@ -845,7 +856,7 @@ private FileRecordReader<InternalRow> createFileReader(
}

FileReadTarget readTarget =
readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult);
readTarget(file, dataFilePathFactory, rowRanges, fileIndexResult, deletionVector);
String formatIdentifier = readTarget.formatIdentifier;
boolean positionalRowFormat = readTarget.isPositionalRowFormat();
Builder targetFormatBuilder =
Expand Down Expand Up @@ -915,7 +926,7 @@ private FileRecordReader<InternalRow> createFileReader(
formatReaderMapping,
rowRanges,
readRowType,
readTarget(file, dataFilePathFactory, rowRanges, groupSelection),
readTarget(file, dataFilePathFactory, rowRanges, groupSelection, deletionVector),
deletionVector,
groupSelection,
metadataCache);
Expand All @@ -932,20 +943,15 @@ private FileRecordReader<InternalRow> 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 =
Expand Down Expand Up @@ -1530,66 +1536,57 @@ private FileReadTarget readTarget(
DataFileMeta file,
DataFilePathFactory dataFilePathFactory,
List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult)
@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,
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<String> 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<Range> rowRanges) {
return shouldReadRowSidecar(
file,
rowRanges,
null,
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(),
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue());
file.fileSize(),
selection);
}

@VisibleForTesting
static boolean shouldReadRowSidecar(
DataFileMeta file,
@Nullable List<Range> rowRanges,
@Nullable RoaringBitmap32 selection,
long maxSelectedRows,
double maxSelectionRatio) {
return shouldReadRowSidecar(file, rowRanges, null, maxSelectedRows, maxSelectionRatio);
long selectedRowCount = selection == null ? file.rowCount() : selection.getCardinality();
return shouldReadRowSidecar(file, selectedRowCount, maxSelectedRows, maxSelectionRatio);
}

@VisibleForTesting
static boolean shouldReadRowSidecar(
private static boolean shouldReadRowSidecar(
DataFileMeta file,
@Nullable List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult,
long selectedRowCount,
long maxSelectedRows,
double maxSelectionRatio) {
if (file.rowCount() <= 0
Expand All @@ -1599,7 +1596,6 @@ static boolean shouldReadRowSidecar(
return false;
}

long selectedRowCount = selectedRowCount(file, rowRanges, fileIndexResult);
if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) {
return false;
}
Expand All @@ -1608,33 +1604,115 @@ static boolean shouldReadRowSidecar(
return selectedRowCount <= maxSelectedRows && selectionRatio <= maxSelectionRatio;
}

@Nullable
@VisibleForTesting
static long selectedRowCount(DataFileMeta file, List<Range> rowRanges) {
Range fileRange = file.nonNullRowIdRange();
List<Range> 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<String> rowFiles =
file.extraFiles().stream()
.filter(DataEvolutionSplitRead::isRowSidecarFile)
.collect(Collectors.toList());
return rowFiles.size() == 1 ? rowFiles.get(0) : null;
}

private static long selectedRowCount(
/**
* Builds the rows read from one physical file in file-local positions. Null means every row.
*/
@Nullable
@VisibleForTesting
static RoaringBitmap32 fileSelection(
DataFileMeta file,
@Nullable List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult) {
RoaringBitmap32 selection = file.toFileSelection(rowRanges);
if (fileIndexResult instanceof BitmapIndexResult) {
RoaringBitmap32 indexSelection = ((BitmapIndexResult) fileIndexResult).get();
RoaringBitmap32 rangeSelection = file.toFileSelection(rowRanges);
return rangeSelection == null
? indexSelection.getCardinality()
: RoaringBitmap32.and(rangeSelection, indexSelection).getCardinality();
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()
|| (selection != null && selection.isEmpty())) {
return selection;
}
return isNullOrEmpty(rowRanges) ? file.rowCount() : selectedRowCount(file, rowRanges);
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;
}
return RoaringBitmap32.andNot(
selection == null
? RoaringBitmap32.bitmapOfRange(0, file.rowCount())
: selection,
deleted);
}

if (selection != null && selection.getCardinality() <= deletionVector.getCardinality()) {
RoaringBitmap32 live = new RoaringBitmap32();
Iterator<Integer> 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 deleted = deletedPositions(file, deletionVector, offset);
if (deleted.isEmpty()) {
return selection;
}
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) {
Expand All @@ -1650,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;
}

/**
Expand Down
Loading
Loading