diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java index fdd6955638b5..f06716098c6a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java +++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java @@ -21,6 +21,7 @@ import org.apache.paimon.utils.OptimizedRoaringBitmap64; import org.apache.paimon.utils.Preconditions; import org.apache.paimon.utils.RoaringBitmap32; +import org.apache.paimon.utils.RoaringNavigableMap64; import java.io.DataOutputStream; import java.io.IOException; @@ -60,6 +61,17 @@ public static Bitmap64DeletionVector fromBitmapDeletionVector( OptimizedRoaringBitmap64.fromRoaringBitmap32(roaringBitmap32)); } + public static Bitmap64DeletionVector fromDeletionVector(DeletionVector deletionVector) { + if (deletionVector instanceof Bitmap64DeletionVector) { + return (Bitmap64DeletionVector) deletionVector; + } + if (deletionVector instanceof BitmapDeletionVector) { + return fromBitmapDeletionVector((BitmapDeletionVector) deletionVector); + } + throw new RuntimeException( + "Unsupported deletion vector type: " + deletionVector.getClass()); + } + @Override public void delete(long position) { roaringBitmap.add(position); @@ -112,6 +124,12 @@ public static DeletionVector deserializeFromBitmapDataBytes(byte[] bytes) { return new Bitmap64DeletionVector(bitmap); } + public RoaringNavigableMap64 toRoaringNavigableMap64(long offset) { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + roaringBitmap.forEach(position -> result.add(offset + position)); + return result; + } + // computes and validates the length of the bitmap data (magic bytes + bitmap) private static int computeBitmapDataLength(OptimizedRoaringBitmap64 bitmap) { long length = MAGIC_NUMBER_SIZE_BYTES + bitmap.serializedSizeInBytes(); diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DeletionVectorRowIdFilter.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DeletionVectorRowIdFilter.java new file mode 100644 index 000000000000..ceaf5c1032d4 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DeletionVectorRowIdFilter.java @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.globalindex; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.deletionvectors.Bitmap64DeletionVector; +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.DeletionFile; +import org.apache.paimon.table.source.snapshot.SnapshotReader; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +/** Filters global row ids with deletion vectors from a snapshot. */ +public class DeletionVectorRowIdFilter { + + private final FileStoreTable table; + @Nullable private final Snapshot snapshot; + @Nullable private final PartitionPredicate partitionFilter; + + public DeletionVectorRowIdFilter( + FileStoreTable table, + @Nullable Snapshot snapshot, + @Nullable PartitionPredicate partitionFilter) { + this.table = table; + this.snapshot = snapshot; + this.partitionFilter = partitionFilter; + } + + public GlobalIndexResult filter(GlobalIndexResult result) { + if (!shouldFilter(result.results())) { + return result; + } + return GlobalIndexResult.create(filter(result.results())); + } + + public ScoredGlobalIndexResult filter(ScoredGlobalIndexResult result) { + if (!shouldFilter(result.results())) { + return result; + } + return ScoredGlobalIndexResult.create(filter(result.results()), result.scoreGetter()); + } + + public RoaringNavigableMap64 deletedRows(List ranges) { + RoaringNavigableMap64 deletedRows = new RoaringNavigableMap64(); + if (!table.coreOptions().deletionVectorsEnabled() || snapshot == null || ranges.isEmpty()) { + return deletedRows; + } + + List normalizedRanges = Range.sortAndMergeOverlap(ranges, true); + if (normalizedRanges.isEmpty()) { + return deletedRows; + } + + RoaringNavigableMap64 candidateRows = bitmapOf(normalizedRanges); + SnapshotReader reader = table.newSnapshotReader().withSnapshot(snapshot); + if (partitionFilter != null) { + reader = reader.withPartitionFilter(partitionFilter); + } + reader = reader.withRowRanges(normalizedRanges); + + try { + for (DataSplit split : reader.read().dataSplits()) { + addDeletedRows(deletedRows, candidateRows, split); + } + } catch (IOException e) { + throw new RuntimeException("Failed to read deletion vectors for global row ids.", e); + } + return deletedRows; + } + + private boolean shouldFilter(RoaringNavigableMap64 rowIds) { + return table.coreOptions().deletionVectorsEnabled() + && snapshot != null + && !rowIds.isEmpty(); + } + + private RoaringNavigableMap64 filter(RoaringNavigableMap64 rowIds) { + RoaringNavigableMap64 filtered = + RoaringNavigableMap64.or(new RoaringNavigableMap64(), rowIds); + filtered.andNot(deletedRows(rowIds.toRangeList())); + return filtered; + } + + private void addDeletedRows( + RoaringNavigableMap64 deletedRows, RoaringNavigableMap64 candidateRows, DataSplit split) + throws IOException { + List files = split.dataFiles(); + Optional> optionalDeletionFiles = split.deletionFiles(); + if (!optionalDeletionFiles.isPresent()) { + return; + } + + DeletionVector.Factory factory = + DeletionVector.factory(table.fileIO(), files, optionalDeletionFiles.get()); + for (DataFileMeta file : files) { + if (file.firstRowId() == null) { + continue; + } + Optional optionalDeletionVector = factory.create(file.fileName()); + if (!optionalDeletionVector.isPresent()) { + continue; + } + + RoaringNavigableMap64 fileDeletedRows = + Bitmap64DeletionVector.fromDeletionVector(optionalDeletionVector.get()) + .toRoaringNavigableMap64(file.nonNullFirstRowId()); + fileDeletedRows.and(bitmapOf(file.nonNullRowIdRange())); + fileDeletedRows.and(candidateRows); + deletedRows.or(fileDeletedRows); + } + } + + private static RoaringNavigableMap64 bitmapOf(Range range) { + return bitmapOf(Collections.singletonList(range)); + } + + private static RoaringNavigableMap64 bitmapOf(List ranges) { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + for (Range range : ranges) { + result.addRange(range); + } + return result; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java index 0537e50114e8..1241a582f203 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java @@ -34,6 +34,8 @@ import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.DeletionFile; +import org.apache.paimon.table.source.snapshot.SnapshotReader; import org.apache.paimon.types.DataField; import org.apache.paimon.utils.Range; @@ -51,6 +53,8 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.Optional; import java.util.function.BiFunction; import static org.apache.paimon.utils.Preconditions.checkArgument; @@ -178,7 +182,30 @@ public static List createShardIndexedSplits( rowsPerShard, (partition, bucket) -> table.store().pathFactory().bucketPath(partition, bucket).toString(), - rowRangesToBuild); + rowRangesToBuild, + Collections.emptyMap()); + } + + public static List createShardIndexedSplits( + FileStoreTable table, + @Nullable Snapshot snapshot, + @Nullable PartitionPredicate partitionPredicate, + List entries, + long rowsPerShard, + @Nullable List rowRangesToBuild) { + checkRowsPerShard(rowsPerShard); + rowRangesToBuild = normalizeRowRanges(rowRangesToBuild); + if (rowRangesToBuild != null && rowRangesToBuild.isEmpty()) { + return Collections.emptyList(); + } + + return createShardIndexedSplits( + entries, + rowsPerShard, + (partition, bucket) -> + table.store().pathFactory().bucketPath(partition, bucket).toString(), + rowRangesToBuild, + deletionFilesByFile(table, snapshot, partitionPredicate, rowRangesToBuild)); } public static List createShardIndexedSplits( @@ -186,14 +213,20 @@ public static List createShardIndexedSplits( long rowsPerShard, BiFunction bucketPathFactory, @Nullable List rowRangesToBuild) { - checkArgument( - rowsPerShard > 0, - "Option 'global-index.row-count-per-shard' must be greater than 0."); - if (rowRangesToBuild != null) { - rowRangesToBuild = Range.sortAndMergeOverlap(rowRangesToBuild, true); - if (rowRangesToBuild.isEmpty()) { - return Collections.emptyList(); - } + return createShardIndexedSplits( + entries, rowsPerShard, bucketPathFactory, rowRangesToBuild, Collections.emptyMap()); + } + + private static List createShardIndexedSplits( + List entries, + long rowsPerShard, + BiFunction bucketPathFactory, + @Nullable List rowRangesToBuild, + Map deletionFilesByFile) { + checkRowsPerShard(rowsPerShard); + rowRangesToBuild = normalizeRowRanges(rowRangesToBuild); + if (rowRangesToBuild != null && rowRangesToBuild.isEmpty()) { + return Collections.emptyList(); } Map>> entriesByPartitionAndBucket = @@ -218,12 +251,24 @@ public static List createShardIndexedSplits( bucketEntry.getValue(), rowsPerShard, bucketPathFactory, - rowRangesToBuild); + rowRangesToBuild, + deletionFilesByFile); } } return result; } + private static void checkRowsPerShard(long rowsPerShard) { + checkArgument( + rowsPerShard > 0, + "Option 'global-index.row-count-per-shard' must be greater than 0."); + } + + @Nullable + private static List normalizeRowRanges(@Nullable List rowRangesToBuild) { + return rowRangesToBuild == null ? null : Range.sortAndMergeOverlap(rowRangesToBuild, true); + } + private static void addShardIndexedSplits( List result, BinaryRow partition, @@ -231,7 +276,8 @@ private static void addShardIndexedSplits( List entries, long rowsPerShard, BiFunction bucketPathFactory, - @Nullable List rowRangesToBuild) { + @Nullable List rowRangesToBuild, + Map deletionFilesByFile) { Map> filesByShard = new LinkedHashMap<>(); for (ManifestEntry entry : entries) { DataFileMeta file = entry.file(); @@ -283,7 +329,8 @@ private static void addShardIndexedSplits( bucket, entries.get(0).totalBuckets(), bucketPathFactory.apply(partition, bucket), - rowRangesToBuild); + rowRangesToBuild, + deletionFilesByFile); currentGroup = new ArrayList<>(); currentGroup.add(file); currentGroupEnd = fileEnd; @@ -299,7 +346,8 @@ private static void addShardIndexedSplits( bucket, entries.get(0).totalBuckets(), bucketPathFactory.apply(partition, bucket), - rowRangesToBuild); + rowRangesToBuild, + deletionFilesByFile); } } } @@ -313,7 +361,8 @@ private static void addIndexedSplitForFileGroup( int bucket, int totalBuckets, String bucketPath, - @Nullable List rowRangesToBuild) { + @Nullable List rowRangesToBuild, + Map deletionFilesByFile) { long groupMinRowId = files.get(0).nonNullFirstRowId(); long groupMaxRowId = files.stream().mapToLong(file -> file.nonNullRowIdRange().to).max().getAsLong(); @@ -327,20 +376,118 @@ private static void addIndexedSplitForFileGroup( return; } - DataSplit dataSplit = + DataSplit.Builder builder = DataSplit.builder() .withPartition(partition) .withBucket(bucket) .withTotalBuckets(totalBuckets) .withDataFiles(files) .withBucketPath(bucketPath) - .rawConvertible(false) - .build(); + .rawConvertible(false); + List dataDeletionFiles = + dataDeletionFiles(files, partition, bucket, deletionFilesByFile); + if (dataDeletionFiles != null) { + builder.withDataDeletionFiles(dataDeletionFiles); + } + DataSplit dataSplit = builder.build(); for (Range taskRange : taskRanges) { result.add(new IndexedSplit(dataSplit, Collections.singletonList(taskRange), null)); } } + @Nullable + private static List dataDeletionFiles( + List files, + BinaryRow partition, + int bucket, + Map deletionFilesByFile) { + if (deletionFilesByFile.isEmpty()) { + return null; + } + + List result = new ArrayList<>(files.size()); + boolean hasDeletionFile = false; + for (DataFileMeta file : files) { + DeletionFile deletionFile = + deletionFilesByFile.get(new FileKey(partition, bucket, file.fileName())); + if (deletionFile != null) { + hasDeletionFile = true; + } + result.add(deletionFile); + } + return hasDeletionFile ? result : null; + } + + private static Map deletionFilesByFile( + FileStoreTable table, + @Nullable Snapshot snapshot, + @Nullable PartitionPredicate partitionPredicate, + @Nullable List rowRangesToBuild) { + if (!table.coreOptions().deletionVectorsEnabled() || snapshot == null) { + return Collections.emptyMap(); + } + + SnapshotReader reader = table.newSnapshotReader().withSnapshot(snapshot); + if (partitionPredicate != null) { + reader = reader.withPartitionFilter(partitionPredicate); + } + if (rowRangesToBuild != null) { + reader = reader.withRowRanges(rowRangesToBuild); + } + + Map result = new HashMap<>(); + for (DataSplit split : reader.read().dataSplits()) { + Optional> optionalDeletionFiles = split.deletionFiles(); + if (!optionalDeletionFiles.isPresent()) { + continue; + } + + List files = split.dataFiles(); + List deletionFiles = optionalDeletionFiles.get(); + for (int i = 0; i < files.size(); i++) { + DeletionFile deletionFile = deletionFiles.get(i); + if (deletionFile != null) { + result.put( + new FileKey(split.partition(), split.bucket(), files.get(i).fileName()), + deletionFile); + } + } + } + return result; + } + + private static class FileKey { + + private final BinaryRow partition; + private final int bucket; + private final String fileName; + + private FileKey(BinaryRow partition, int bucket, String fileName) { + this.partition = partition; + this.bucket = bucket; + this.fileName = fileName; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + FileKey fileKey = (FileKey) o; + return bucket == fileKey.bucket + && Objects.equals(partition, fileKey.partition) + && Objects.equals(fileName, fileKey.fileName); + } + + @Override + public int hashCode() { + return Objects.hash(partition, bucket, fileName); + } + } + private static List toIndexFileMetas( FileIO fileIO, IndexPathFactory indexPathFactory, diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java index 7edb68365b6b..19ff5061ed84 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java @@ -70,6 +70,7 @@ public class GlobalIndexScanner implements Closeable { private final IndexPathFactory indexPathFactory; private final GlobalIndexCoverage coverage; private final FileStoreTable table; + private final DeletionVectorRowIdFilter rowIdFilter; private GlobalIndexScanner( FileStoreTable table, @@ -87,6 +88,7 @@ private GlobalIndexScanner( GlobalIndexReadThreadPool.getExecutorService(options.get(GLOBAL_INDEX_THREAD_NUM)); this.indexPathFactory = indexPathFactory; this.coverage = new GlobalIndexCoverage(table, snapshot, partitionFilter, indexFiles); + this.rowIdFilter = new DeletionVectorRowIdFilter(table, snapshot, partitionFilter); GlobalIndexFileReader indexFileReader = meta -> fileIO.newInputStream(meta.filePath()); Map indexMetas = new HashMap<>(); Map> extraIndexMetas = new HashMap<>(); @@ -262,7 +264,7 @@ private static Filter indexFileFilter( } public Optional scan(Predicate predicate) { - return globalIndexEvaluator.evaluate(predicate); + return globalIndexEvaluator.evaluate(predicate).map(rowIdFilter::filter); } public GlobalIndexResult unindexedRows(Predicate predicate) { @@ -270,7 +272,7 @@ public GlobalIndexResult unindexedRows(Predicate predicate) { for (Range range : coverage.unindexedRanges(rowType, predicate)) { rows.addRange(range); } - return GlobalIndexResult.create(rows); + return rowIdFilter.filter(GlobalIndexResult.create(rows)); } private Collection createReaders( diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java index 6ee00b3bb8c8..1629f8f2c2ce 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java @@ -44,6 +44,7 @@ import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.DeletionFile; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.snapshot.SnapshotReader; import org.apache.paimon.types.DataField; @@ -417,6 +418,7 @@ private static List splitByContiguousRowRange(DataSplit split) { List input = split.dataFiles(); RangeHelper rangeHelper = new RangeHelper<>(DataFileMeta::nonNullRowIdRange); List> ranges = rangeHelper.mergeOverlappingRanges(input); + Map deletionFiles = deletionFilesByName(split); Supplier builderSupplier = () -> @@ -428,11 +430,13 @@ private static List splitByContiguousRowRange(DataSplit split) { .withTotalBuckets(split.totalBuckets()) .isStreaming(split.isStreaming()) .rawConvertible(split.rawConvertible()); - return packByContiguousRanges(builderSupplier, ranges); + return packByContiguousRanges(builderSupplier, ranges, deletionFiles); } private static List packByContiguousRanges( - Supplier builderFactory, List> ranges) { + Supplier builderFactory, + List> ranges, + Map deletionFiles) { if (ranges.isEmpty()) { return new ArrayList<>(); } @@ -449,7 +453,7 @@ private static List packByContiguousRanges( currentMaxRowId = maxRowId; } else { DataSplit.Builder builder = builderFactory.get(); - builder.withDataFiles(currentSegment); + withDataFilesAndDeletionFiles(builder, currentSegment, deletionFiles); result.add(builder.build()); currentSegment = new ArrayList<>(rangeFiles); currentMaxRowId = maxRowId; @@ -457,11 +461,52 @@ private static List packByContiguousRanges( } DataSplit.Builder builder = builderFactory.get(); - builder.withDataFiles(currentSegment); + withDataFilesAndDeletionFiles(builder, currentSegment, deletionFiles); result.add(builder.build()); return result; } + private static Map deletionFilesByName(DataSplit split) { + Optional> optionalDeletionFiles = split.deletionFiles(); + if (!optionalDeletionFiles.isPresent()) { + return Collections.emptyMap(); + } + + List files = split.dataFiles(); + List deletionFiles = optionalDeletionFiles.get(); + Map result = new HashMap<>(); + for (int i = 0; i < files.size(); i++) { + DeletionFile deletionFile = deletionFiles.get(i); + if (deletionFile != null) { + result.put(files.get(i).fileName(), deletionFile); + } + } + return result; + } + + private static void withDataFilesAndDeletionFiles( + DataSplit.Builder builder, + List files, + Map deletionFiles) { + builder.withDataFiles(files); + if (deletionFiles.isEmpty()) { + return; + } + + List alignedDeletionFiles = new ArrayList<>(files.size()); + boolean hasDeletionFile = false; + for (DataFileMeta file : files) { + DeletionFile deletionFile = deletionFiles.get(file.fileName()); + if (deletionFile != null) { + hasDeletionFile = true; + } + alignedDeletionFiles.add(deletionFile); + } + if (hasDeletionFile) { + builder.withDataDeletionFiles(alignedDeletionFiles); + } + } + private static long minRowId(List files) { return files.stream() .mapToLong(f -> f.nonNullRowIdRange().from) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java index a1996f920301..fde4717eae6a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java @@ -18,10 +18,12 @@ package org.apache.paimon.table.source; +import org.apache.paimon.Snapshot; import org.apache.paimon.data.InternalArray; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.InternalVector; import org.apache.paimon.fs.FileIO; +import org.apache.paimon.globalindex.DeletionVectorRowIdFilter; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; @@ -65,6 +67,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; +import static org.apache.paimon.table.source.snapshot.TimeTravelUtil.tryTravelOrLatest; import static org.apache.paimon.utils.Preconditions.checkNotNull; /** Base implementation for vector reads. */ @@ -160,6 +163,73 @@ protected List preFilters(List sp return includeRowIds; } + protected List liveRowFilters(List splits) { + if (!table.coreOptions().deletionVectorsEnabled() || splits.isEmpty()) { + return Collections.emptyList(); + } + + Snapshot snapshot = tryTravelOrLatest(table); + if (snapshot == null) { + return Collections.emptyList(); + } + + List splitRanges = new ArrayList<>(splits.size()); + for (IndexVectorSearchSplit split : splits) { + splitRanges.add(new Range(split.rowRangeStart(), split.rowRangeEnd())); + } + + DeletionVectorRowIdFilter rowIdFilter = + new DeletionVectorRowIdFilter(table, snapshot, partitionFilter); + RoaringNavigableMap64 deletedRows = rowIdFilter.deletedRows(splitRanges); + if (deletedRows.isEmpty()) { + return Collections.emptyList(); + } + + List liveRows = new ArrayList<>(splits.size()); + for (Range splitRange : splitRanges) { + RoaringNavigableMap64 splitLiveRows = bitmapOf(splitRange); + splitLiveRows.andNot(deletedRows); + liveRows.add(splitLiveRows); + } + return liveRows; + } + + @Nullable + protected RoaringNavigableMap64 includeRowIds( + List preFilters, + List liveRowFilters, + int index) { + boolean hasPreFilter = !preFilters.isEmpty(); + boolean hasLiveRowFilter = !liveRowFilters.isEmpty(); + if (!hasPreFilter && !hasLiveRowFilter) { + return null; + } + + if (hasLiveRowFilter) { + RoaringNavigableMap64 include = + RoaringNavigableMap64.or( + new RoaringNavigableMap64(), liveRowFilters.get(index)); + if (hasPreFilter) { + include.and(preFilters.get(index)); + } + return include; + } + + return preFilters.get(index); + } + + protected ScoredGlobalIndexResult filterDeletedRows(ScoredGlobalIndexResult result) { + if (!table.coreOptions().deletionVectorsEnabled() || result.results().isEmpty()) { + return result; + } + + Snapshot snapshot = tryTravelOrLatest(table); + if (snapshot == null) { + return result; + } + return new DeletionVectorRowIdFilter(table, snapshot, partitionFilter).filter(result); + } + private List emptyPreFilters(int size) { List preFilters = new ArrayList<>(size); for (int i = 0; i < size; i++) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java index 739b16260de8..75ef4404c4d6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java @@ -98,6 +98,7 @@ protected ScoredGlobalIndexResult[] readIndexedBatch( List splits, GlobalIndexer globalIndexer) { int n = vectors.length; List preFilters = preFilters(splits); + List liveRowFilters = liveRowFilters(splits); String indexType = vectorIndexType(splits); int searchLimit = indexedSearchLimit(indexType); @@ -119,7 +120,7 @@ protected ScoredGlobalIndexResult[] readIndexedBatch( split.vectorIndexFiles(), vectors, searchLimit, - preFilters.isEmpty() ? null : preFilters.get(i), + includeRowIds(preFilters, liveRowFilters, i), executor)); } @@ -139,7 +140,10 @@ protected ScoredGlobalIndexResult[] readIndexedBatch( } } for (int i = 0; i < n; i++) { - merged[i] = maybeRerankIndexedResult(merged[i], indexType, globalIndexer, vectors[i]); + merged[i] = + filterDeletedRows( + maybeRerankIndexedResult( + merged[i], indexType, globalIndexer, vectors[i])); } return merged; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java index 97609ba56495..3709f85a9fdc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java @@ -18,6 +18,8 @@ package org.apache.paimon.table.source; +import org.apache.paimon.Snapshot; +import org.apache.paimon.globalindex.DeletionVectorRowIdFilter; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReadThreadPool; import org.apache.paimon.globalindex.GlobalIndexReader; @@ -51,6 +53,7 @@ import java.util.concurrent.ExecutorService; import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM; +import static org.apache.paimon.table.source.snapshot.TimeTravelUtil.tryTravelOrLatest; import static org.apache.paimon.utils.Preconditions.checkNotNull; /** Implementation for {@link FullTextRead}. */ @@ -123,13 +126,26 @@ public GlobalIndexResult read(List splits) { indexPathFactory, indexFileReader, executor); + result = filterDeletedRows(result); if (!rawRowRanges.isEmpty()) { result = new RawFullTextReadImpl(table, partitionFilter, limit, query, this::evalQuery) .withRawSearch( result, rawRowRanges, fieldsByName, splitsByColumn, executor); } - return result.topK(limit); + return filterDeletedRows(result).topK(limit); + } + + private ScoredGlobalIndexResult filterDeletedRows(ScoredGlobalIndexResult result) { + if (!table.coreOptions().deletionVectorsEnabled() || result.results().isEmpty()) { + return result; + } + + Snapshot snapshot = tryTravelOrLatest(table); + if (snapshot == null) { + return result; + } + return new DeletionVectorRowIdFilter(table, snapshot, partitionFilter).filter(result); } ScoredGlobalIndexResult evalQuery( @@ -229,7 +245,7 @@ private ScoredGlobalIndexResult evalMultiMatch( indexFileReader, executor)); } - return or(results).topK(limit); + return filterDeletedRows(or(results)).topK(limit); } private ScoredGlobalIndexResult evalBoolean( @@ -282,7 +298,7 @@ private ScoredGlobalIndexResult evalBoolean( executor); result = andNot(result, childResult); } - return result.topK(limit); + return filterDeletedRows(result).topK(limit); } private ScoredGlobalIndexResult evalColumnQuery( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java index 9d8bd1541ed7..8c98d703b45c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java @@ -84,6 +84,7 @@ protected GlobalIndexResult readSplits(List splits) protected ScoredGlobalIndexResult readIndexed( List splits, GlobalIndexer globalIndexer) { List preFilters = preFilters(splits); + List liveRowFilters = liveRowFilters(splits); String indexType = vectorIndexType(splits); int searchLimit = indexedSearchLimit(indexType); @@ -105,7 +106,7 @@ protected ScoredGlobalIndexResult readIndexed( split.vectorIndexFiles(), vector, searchLimit, - preFilters.isEmpty() ? null : preFilters.get(i), + includeRowIds(preFilters, liveRowFilters, i), executor)); } @@ -118,6 +119,7 @@ protected ScoredGlobalIndexResult readIndexed( merged = merged.or(splitResult.get()); } } - return maybeRerankIndexedResult(merged, indexType, globalIndexer, vector); + return filterDeletedRows( + maybeRerankIndexedResult(merged, indexType, globalIndexer, vector)); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java index 1800f4c3328e..374e00b5b997 100644 --- a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java @@ -23,6 +23,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.reader.FileRecordIterator; import org.apache.paimon.reader.FileRecordReader; +import org.apache.paimon.utils.RoaringNavigableMap64; import org.junit.jupiter.api.Test; @@ -185,14 +186,19 @@ public void testBitmapDeletionVectorTo64() { Bitmap64DeletionVector.fromBitmapDeletionVector(deletionVector); assertThat(bitmap64DeletionVector.isEmpty()).isFalse(); + RoaringNavigableMap64 deletionBitmap = + ((Bitmap64DeletionVector) bitmap64DeletionVector).toRoaringNavigableMap64(100); + assertThat(deletionBitmap.getLongCardinality()).isEqualTo(toDelete.size()); for (Integer i : toDelete) { assertThat(deletionVector.isDeleted(i)).isTrue(); assertThat(bitmap64DeletionVector.isDeleted(i)).isTrue(); + assertThat(deletionBitmap.contains(i + 100L)).isTrue(); } for (Integer i : notDelete) { assertThat(deletionVector.isDeleted(i)).isFalse(); assertThat(bitmap64DeletionVector.isDeleted(i)).isFalse(); + assertThat(deletionBitmap.contains(i + 100L)).isFalse(); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java index 2d64ce9db541..eaac55974841 100644 --- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java @@ -24,6 +24,7 @@ import org.apache.paimon.io.PojoDataFileMeta; import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.DeletionFile; import org.apache.paimon.table.source.Split; import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RowRangeIndex; @@ -70,6 +71,37 @@ public void testSplitByContiguousRowRangeFromDataFiles() { .isEqualTo(new Range(300, 399)); } + @Test + public void testSplitByContiguousRowRangePreservesDeletionFiles() { + DataFileMeta file1 = createDataFileMeta(0L, 100L); + DataFileMeta file2 = createDataFileMeta(300L, 100L); + DataFileMeta file3 = createDataFileMeta(100L, 100L); + DeletionFile deletionFile1 = new DeletionFile("dv-1", 0L, 1L, 1L); + DeletionFile deletionFile3 = new DeletionFile("dv-3", 1L, 1L, 1L); + DataSplit split = + DataSplit.builder() + .withSnapshot(1L) + .withPartition(BinaryRow.EMPTY_ROW) + .withBucket(0) + .withBucketPath("bucket-0") + .withDataFiles(Arrays.asList(file1, file2, file3)) + .withDataDeletionFiles(Arrays.asList(deletionFile1, null, deletionFile3)) + .isStreaming(false) + .rawConvertible(false) + .build(); + + List rebuilt = + SortedGlobalIndexBuilder.splitByContiguousRowRange( + Collections.singletonList(split)); + + assertThat(rebuilt).hasSize(2); + assertThat(rebuilt.get(0).dataFiles()).containsExactly(file1, file3); + assertThat(rebuilt.get(0).deletionFiles()) + .hasValue(Arrays.asList(deletionFile1, deletionFile3)); + assertThat(rebuilt.get(1).dataFiles()).containsExactly(file2); + assertThat(rebuilt.get(1).deletionFiles()).isNotPresent(); + } + @Test public void testGroupSplitsByDiscontiguousRowRangeIndex() { DataFileMeta file1 = createDataFileMeta(4750L, 151L); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/DeletionVectorTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/table/source/DeletionVectorTestUtils.java new file mode 100644 index 000000000000..512ce6bcbc26 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/DeletionVectorTestUtils.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.source; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.deletionvectors.BitmapDeletionVector; +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RangeHelper; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET; +import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile; + +final class DeletionVectorTestUtils { + + private DeletionVectorTestUtils() {} + + static void commitDeletionVector(FileStoreTable table, Range range, long... deletedRowIds) + throws Exception { + BaseAppendDeleteFileMaintainer maintainer = + BaseAppendDeleteFileMaintainer.forUnawareAppend( + table.store().newIndexFileHandler(), + table.latestSnapshot().get(), + BinaryRow.EMPTY_ROW); + DeletionVector deletionVector = new BitmapDeletionVector(); + for (long rowId : deletedRowIds) { + deletionVector.delete(rowId - range.from); + } + maintainer.notifyNewDeletionVector(anchorFilesByRange(table).get(range), deletionVector); + + List newIndexFiles = new ArrayList<>(); + List deletedIndexFiles = new ArrayList<>(); + for (IndexManifestEntry entry : maintainer.persist()) { + if (entry.kind() == FileKind.ADD) { + newIndexFiles.add(entry.indexFile()); + } else if (entry.kind() == FileKind.DELETE) { + deletedIndexFiles.add(entry.indexFile()); + } + } + + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit( + Collections.singletonList( + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + UNAWARE_BUCKET, + null, + new DataIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + newIndexFiles, + deletedIndexFiles), + CompactIncrement.emptyIncrement()))); + } + } + + private static Map anchorFilesByRange(FileStoreTable table) { + List dataFiles = + table.store().newScan().plan().files().stream() + .map(ManifestEntry::file) + .collect(Collectors.toList()); + RangeHelper rangeHelper = new RangeHelper<>(DataFileMeta::nonNullRowIdRange); + Map result = new HashMap<>(); + for (List group : rangeHelper.mergeOverlappingRanges(dataFiles)) { + DataFileMeta anchor = retrieveAnchorFile(group, file -> file); + result.put(anchor.nonNullRowIdRange(), anchor.fileName()); + } + return result; + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java index 5f61bc5de69c..e99aa56e7c94 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java @@ -125,6 +125,36 @@ public void testFullTextSearchEndToEnd() throws Exception { assertThat(ids).containsAnyOf(0, 1, 3); } + @Test + public void testFullTextSearchExcludesDeletionVectorRowsBeforeTopK() throws Exception { + catalog.createTable( + identifier("full_text_deletion_vector_table"), + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.STRING()) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier("full_text_deletion_vector_table")); + + String[] documents = {"paimon", "paimon", "paimon"}; + writeDocuments(table, documents); + buildAndCommitIndex(table, documents); + DeletionVectorTestUtils.commitDeletionVector(table, new Range(0, 2), 0); + + GlobalIndexResult result = + table.newFullTextSearchBuilder() + .withQuery(FullTextQuery.match("paimon", TEXT_FIELD_NAME)) + .withLimit(1) + .executeLocal(); + + assertThat(result.results()).containsExactly(1L); + assertThat(readIds(table, result)).containsExactly(1); + } + @Test public void testFullTextSearchNonFastModesScanUnindexedData() throws Exception { createTableDefault(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java index b937efe223c6..3f78b954eecf 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java @@ -26,7 +26,9 @@ import org.apache.paimon.data.serializer.InternalRowSerializer; import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.GlobalIndexScanner; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.globalindex.ResultEntry; import org.apache.paimon.globalindex.ScoredGlobalIndexResult; import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory; @@ -35,6 +37,7 @@ import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.FieldRef; @@ -186,6 +189,94 @@ public void testVectorSearchEndToEnd() throws Exception { assertThat(ids).contains(0); } + @Test + public void testVectorSearchExcludesDeletionVectorRowsBeforeTopK() throws Exception { + catalog.createTable( + identifier("vector_deletion_vector_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier("vector_deletion_vector_table")); + + float[][] vectors = {{1.0f, 0.0f}, {0.95f, 0.05f}, {0.0f, 1.0f}}; + writeVectors(table, vectors); + buildAndCommitIndex(table, vectors); + DeletionVectorTestUtils.commitDeletionVector(table, new Range(0, 2), 0); + + GlobalIndexResult result = + table.newVectorSearchBuilder() + .withVector(new float[] {1.0f, 0.0f}) + .withLimit(1) + .withVectorColumn(VECTOR_FIELD_NAME) + .executeLocal(); + + assertThat(result.results()).containsExactly(1L); + assertThat(readIds(table, result)).containsExactly(1); + + List batchResults = + table.newBatchVectorSearchBuilder() + .withVectors(new float[][] {new float[] {1.0f, 0.0f}}) + .withLimit(1) + .withVectorColumn(VECTOR_FIELD_NAME) + .executeBatchLocal(); + assertThat(batchResults).hasSize(1); + assertThat(batchResults.get(0).results()).containsExactly(1L); + } + + @Test + public void testGlobalIndexScannerExcludesDeletionVectorRows() throws Exception { + catalog.createTable( + identifier("global_index_deletion_vector_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier("global_index_deletion_vector_table")); + + float[][] vectors = {{1.0f, 0.0f}, {0.95f, 0.05f}, {0.0f, 1.0f}}; + writeVectors(table, vectors); + buildAndCommitBTreeIndex(table, new int[] {0, 1, 2}, new Range(0, 2)); + DeletionVectorTestUtils.commitDeletionVector(table, new Range(0, 2), 0); + + Predicate filter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 0); + try (GlobalIndexScanner scanner = GlobalIndexScanner.create(table, null, filter).get()) { + GlobalIndexResult result = scanner.scan(filter).get(); + + assertThat(result.results()).containsExactly(1L, 2L); + } + } + + @Test + public void testShardIndexedSplitsPreserveDeletionFiles() throws Exception { + catalog.createTable( + identifier("shard_indexed_split_deletion_vector_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier("shard_indexed_split_deletion_vector_table")); + + float[][] vectors = {{1.0f, 0.0f}, {0.95f, 0.05f}, {0.0f, 1.0f}}; + writeVectors(table, vectors); + DeletionVectorTestUtils.commitDeletionVector(table, new Range(0, 2), 0); + + List entries = + table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files(); + List splits = + GlobalIndexBuilderUtils.createShardIndexedSplits( + table, + table.latestSnapshot().get(), + null, + entries, + 10, + Collections.singletonList(new Range(0, 2))); + + assertThat(splits).hasSize(1); + assertThat(splits.get(0).dataSplit().deletionFiles()).isPresent(); + assertThat(splits.get(0).dataSplit().deletionFiles().get()).anyMatch(file -> file != null); + } + @Test public void testVectorSearchWithCosineMetric() throws Exception { // Create a table with cosine metric diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java index c8dc0bc01193..0bb8c7f5dcce 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java @@ -134,7 +134,12 @@ public List buildIndex( // generate splits for each partition && shard List splits = GlobalIndexBuilderUtils.createShardIndexedSplits( - table, entries, rowsPerShard, rowRangesToBuild); + table, + snapshot, + partitionPredicate, + entries, + rowsPerShard, + rowRangesToBuild); JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext()); List> taskList = new ArrayList<>();