This is an automated email from the ASF dual-hosted git repository. JingsongLi pushed a commit to branch codex/filter-deleted-index-search in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 77670dbd4787959acf03995beaf2dbd0d14d62ae Author: JingsongLi <[email protected]> AuthorDate: Sat Jul 4 15:23:26 2026 +0800 fix: filter deleted rows from global index search --- .../globalindex/OffsetGlobalIndexReader.java | 2 +- .../apache/paimon/predicate/FullTextSearch.java | 24 ++ .../TestFullTextGlobalIndexReader.java | 4 + .../deletionvectors/Bitmap64DeletionVector.java | 6 + .../deletionvectors/BitmapDeletionVector.java | 10 + .../paimon/deletionvectors/DeletionVector.java | 4 + .../paimon/table/source/AbstractVectorRead.java | 59 ++-- .../paimon/table/source/FullTextReadImpl.java | 74 +++-- .../table/source/GlobalIndexLiveRowFilter.java | 113 ++++++++ .../table/source/DeletionVectorTestUtils.java | 118 ++++++++ .../table/source/FullTextSearchBuilderTest.java | 35 +++ .../table/source/VectorSearchBuilderTest.java | 58 ++++ .../pypaimon/globalindex/full_text_search.py | 25 ++ .../globalindex/offset_global_index_reader.py | 3 +- .../tantivy_full_text_global_index_reader.py | 16 +- .../pypaimon/table/source/full_text_read.py | 57 ++-- .../table/source/full_text_search_builder.py | 6 +- .../table/source/global_index_live_row_filter.py | 85 ++++++ .../pypaimon/table/source/vector_search_read.py | 55 +++- .../pypaimon/tests/vector_search_filter_test.py | 307 ++++++++++++++++++++- .../index/TantivyFullTextGlobalIndexReader.java | 10 +- 21 files changed, 985 insertions(+), 86 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java index f740223d5a..646e1f6ca2 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java @@ -143,7 +143,7 @@ public class OffsetGlobalIndexReader implements GlobalIndexReader { @Override public CompletableFuture<Optional<ScoredGlobalIndexResult>> visitFullTextSearch( FullTextSearch fullTextSearch) { - return wrapped.visitFullTextSearch(fullTextSearch) + return wrapped.visitFullTextSearch(fullTextSearch.offsetRange(this.offset, this.to)) .thenApply(opt -> opt.map(r -> r.offset(offset))); } diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java index b9731434c2..1f9b11fe1e 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java @@ -18,6 +18,10 @@ package org.apache.paimon.predicate; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import javax.annotation.Nullable; + import java.io.Serializable; import java.util.List; @@ -29,6 +33,8 @@ public class FullTextSearch implements Serializable { private final FullTextQuery query; private final int limit; + @Nullable private RoaringNavigableMap64 includeRowIds; + public FullTextSearch(FullTextQuery query, int limit) { if (query == null) { throw new IllegalArgumentException("Query cannot be null"); @@ -56,6 +62,24 @@ public class FullTextSearch implements Serializable { return query; } + public RoaringNavigableMap64 includeRowIds() { + return includeRowIds; + } + + public FullTextSearch withIncludeRowIds(RoaringNavigableMap64 includeRowIds) { + this.includeRowIds = includeRowIds; + return this; + } + + public FullTextSearch offsetRange(long from, long to) { + if (includeRowIds != null) { + FullTextSearch target = new FullTextSearch(query, limit); + target.withIncludeRowIds(VectorSearchUtils.offsetRowIds(includeRowIds, from, to)); + return target; + } + return this; + } + public String queryJson() { return query.toJson(); } diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java index 1f6240f529..b68bc6a4f0 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java @@ -82,8 +82,12 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader { // Min-heap: smallest score at head, so we evict the weakest candidate. PriorityQueue<ScoredRow> topK = new PriorityQueue<>(effectiveK + 1, Comparator.comparingDouble(s -> s.score)); + RoaringNavigableMap64 includeRowIds = fullTextSearch.includeRowIds(); for (int i = 0; i < count; i++) { + if (includeRowIds != null && !includeRowIds.contains(rowIds[i])) { + continue; + } float score = computeScore(documents[i], fullTextSearch.query()); if (score <= 0) { continue; 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 fdd6955638..ee06d7232b 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 @@ -27,6 +27,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.Objects; +import java.util.function.LongConsumer; import java.util.zip.CRC32; /** @@ -89,6 +90,11 @@ public class Bitmap64DeletionVector implements DeletionVector { return roaringBitmap.cardinality(); } + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + roaringBitmap.forEach(consumer); + } + @Override public int serializeTo(DataOutputStream out) throws IOException { roaringBitmap.runLengthEncode(); // run-length encode the bitmap before serializing diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BitmapDeletionVector.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BitmapDeletionVector.java index 40681a7772..191f314283 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BitmapDeletionVector.java +++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BitmapDeletionVector.java @@ -24,7 +24,9 @@ import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Iterator; import java.util.Objects; +import java.util.function.LongConsumer; import java.util.zip.CRC32; /** @@ -83,6 +85,14 @@ public class BitmapDeletionVector implements DeletionVector { return roaringBitmap.getCardinality(); } + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + Iterator<Integer> iterator = roaringBitmap.iterator(); + while (iterator.hasNext()) { + consumer.accept(iterator.next()); + } + } + @Override public int serializeTo(DataOutputStream out) { try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java index e06dc767a9..3526ae2a1d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java +++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java @@ -34,6 +34,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.List; import java.util.Optional; +import java.util.function.LongConsumer; import static org.apache.paimon.deletionvectors.Bitmap64DeletionVector.toLittleEndianInt; @@ -82,6 +83,9 @@ public interface DeletionVector extends DeletionVectorJudger { /** @return the number of distinct integers added to the DeletionVector. */ long getCardinality(); + /** Iterates over all deleted positions in this deletion vector. */ + void forEachDeletedPosition(LongConsumer consumer); + /** Serializes the deletion vector. */ int serializeTo(DataOutputStream out) throws IOException; 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 a1996f9203..0e3adf3b6a 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 @@ -120,8 +120,38 @@ public abstract class AbstractVectorRead implements Serializable { } protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> splits) { + RoaringNavigableMap64 liveRows = GlobalIndexLiveRowFilter.liveRows(table, partitionFilter); + RoaringNavigableMap64 matchedRows = scalarMatchedRows(splits); + + List<RoaringNavigableMap64> includeRowIds = new ArrayList<>(splits.size()); + boolean hasFilter = false; + for (IndexVectorSearchSplit split : splits) { + Range splitRange = new Range(split.rowRangeStart(), split.rowRangeEnd()); + RoaringNavigableMap64 splitRows = bitmapOf(splitRange); + + RoaringNavigableMap64 include = new RoaringNavigableMap64(); + include.or(splitRows); + if (liveRows != null) { + include.and(liveRows); + } + if (matchedRows != null) { + include.and(matchedRows); + } + + if (include.getLongCardinality() == splitRange.count()) { + includeRowIds.add(null); + } else { + includeRowIds.add(include); + hasFilter = true; + } + } + return hasFilter ? includeRowIds : Collections.emptyList(); + } + + @Nullable + private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> splits) { if (filter == null) { - return Collections.emptyList(); + return null; } Set<IndexFileMeta> scalarIndexFiles = @@ -133,39 +163,18 @@ public abstract class AbstractVectorRead implements Serializable { Optional<GlobalIndexScanner> optionalScanner = GlobalIndexScanner.create(table, partitionFilter, scalarIndexFiles); if (!optionalScanner.isPresent()) { - return emptyPreFilters(splits.size()); + return new RoaringNavigableMap64(); } - RoaringNavigableMap64 matchedRows; try (GlobalIndexScanner scanner = optionalScanner.get()) { Optional<GlobalIndexResult> result = scanner.scan(filter); if (!result.isPresent()) { - return emptyPreFilters(splits.size()); + return new RoaringNavigableMap64(); } - matchedRows = result.get().results(); + return result.get().results(); } catch (IOException e) { throw new RuntimeException(e); } - - List<RoaringNavigableMap64> includeRowIds = new ArrayList<>(splits.size()); - for (IndexVectorSearchSplit split : splits) { - Range splitRange = new Range(split.rowRangeStart(), split.rowRangeEnd()); - RoaringNavigableMap64 splitRows = bitmapOf(splitRange); - - RoaringNavigableMap64 include = new RoaringNavigableMap64(); - include.or(matchedRows); - include.and(splitRows); - includeRowIds.add(include); - } - return includeRowIds; - } - - private List<RoaringNavigableMap64> emptyPreFilters(int size) { - List<RoaringNavigableMap64> preFilters = new ArrayList<>(size); - for (int i = 0; i < size; i++) { - preFilters.add(new RoaringNavigableMap64()); - } - return preFilters; } @Nullable 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 97609ba564..58cfb75e95 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 @@ -115,6 +115,7 @@ public class FullTextReadImpl implements FullTextRead { } GlobalIndexFileReader indexFileReader = m -> table.fileIO().newInputStream(m.filePath()); + RoaringNavigableMap64 liveRows = GlobalIndexLiveRowFilter.liveRows(table, partitionFilter); ScoredGlobalIndexResult result = evalQuery( query, @@ -122,7 +123,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); if (!rawRowRanges.isEmpty()) { result = new RawFullTextReadImpl(table, partitionFilter, limit, query, this::evalQuery) @@ -139,6 +141,24 @@ public class FullTextReadImpl implements FullTextRead { IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, ExecutorService executor) { + return evalQuery( + query, + fieldsByName, + splitsByColumn, + indexPathFactory, + indexFileReader, + executor, + null); + } + + private ScoredGlobalIndexResult evalQuery( + FullTextQuery query, + Map<String, DataField> fieldsByName, + Map<String, List<IndexFullTextSearchSplit>> splitsByColumn, + IndexPathFactory indexPathFactory, + GlobalIndexFileReader indexFileReader, + ExecutorService executor, + @Nullable RoaringNavigableMap64 liveRows) { if (query instanceof FullTextQuery.Match) { return evalColumnQuery( query, @@ -147,7 +167,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); } if (query instanceof FullTextQuery.Phrase) { return evalColumnQuery( @@ -157,7 +178,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); } if (query instanceof FullTextQuery.MultiMatch) { return evalMultiMatch( @@ -166,7 +188,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); } if (query instanceof FullTextQuery.Boost) { FullTextQuery.Boost boost = (FullTextQuery.Boost) query; @@ -177,14 +200,16 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor), + executor, + liveRows), evalQuery( boost.negative(), fieldsByName, splitsByColumn, indexPathFactory, indexFileReader, - executor), + executor, + liveRows), boost.negativeBoost()); } if (query instanceof FullTextQuery.BooleanQuery) { @@ -194,7 +219,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); } throw new IllegalArgumentException("Unsupported full-text query: " + query); } @@ -205,7 +231,8 @@ public class FullTextReadImpl implements FullTextRead { Map<String, List<IndexFullTextSearchSplit>> splitsByColumn, IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, - ExecutorService executor) { + ExecutorService executor, + @Nullable RoaringNavigableMap64 liveRows) { List<String> columns = query.columns(); List<Float> boosts = query.boosts(); List<ScoredGlobalIndexResult> results = new ArrayList<>(columns.size()); @@ -227,7 +254,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor)); + executor, + liveRows)); } return or(results).topK(limit); } @@ -238,7 +266,8 @@ public class FullTextReadImpl implements FullTextRead { Map<String, List<IndexFullTextSearchSplit>> splitsByColumn, IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, - ExecutorService executor) { + ExecutorService executor, + @Nullable RoaringNavigableMap64 liveRows) { ScoredGlobalIndexResult result = null; for (FullTextQuery child : query.must()) { ScoredGlobalIndexResult childResult = @@ -248,7 +277,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); result = result == null ? childResult : and(result, childResult); } @@ -261,7 +291,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor)); + executor, + liveRows)); } if (!shouldResults.isEmpty()) { ScoredGlobalIndexResult shouldResult = or(shouldResults); @@ -279,7 +310,8 @@ public class FullTextReadImpl implements FullTextRead { splitsByColumn, indexPathFactory, indexFileReader, - executor); + executor, + liveRows); result = andNot(result, childResult); } return result.topK(limit); @@ -292,7 +324,8 @@ public class FullTextReadImpl implements FullTextRead { Map<String, List<IndexFullTextSearchSplit>> splitsByColumn, IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, - ExecutorService executor) { + ExecutorService executor, + @Nullable RoaringNavigableMap64 liveRows) { List<IndexFullTextSearchSplit> columnSplits = splitsByColumn.get(column); if (columnSplits == null || columnSplits.isEmpty()) { return ScoredGlobalIndexResult.createEmpty(); @@ -328,7 +361,9 @@ public class FullTextReadImpl implements FullTextRead { split.fullTextIndexFiles(), query, indexFileReader, - executor)); + executor, + GlobalIndexLiveRowFilter.forRange( + liveRows, split.rowRangeStart(), split.rowRangeEnd()))); } CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); @@ -352,7 +387,11 @@ public class FullTextReadImpl implements FullTextRead { List<IndexFileMeta> fullTextIndexFiles, FullTextQuery query, GlobalIndexFileReader indexFileReader, - ExecutorService executor) { + ExecutorService executor, + @Nullable RoaringNavigableMap64 includeRowIds) { + if (includeRowIds != null && includeRowIds.isEmpty()) { + return CompletableFuture.completedFuture(Optional.empty()); + } List<GlobalIndexIOMeta> indexIOMetaList = new ArrayList<>(); for (IndexFileMeta indexFile : fullTextIndexFiles) { GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta()); @@ -365,7 +404,8 @@ public class FullTextReadImpl implements FullTextRead { GlobalIndexReader reader = globalIndexer.createReader(indexFileReader, indexIOMetaList, executor); FullTextSearch fullTextSearch = - new FullTextSearch(query, candidateLimit(rowRangeStart, rowRangeEnd)); + new FullTextSearch(query, candidateLimit(rowRangeStart, rowRangeEnd)) + .withIncludeRowIds(includeRowIds); return new OffsetGlobalIndexReader(reader, rowRangeStart, rowRangeEnd) .visitFullTextSearch(fullTextSearch) .whenComplete((r, t) -> IOUtils.closeQuietly(reader)); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/GlobalIndexLiveRowFilter.java b/paimon-core/src/main/java/org/apache/paimon/table/source/GlobalIndexLiveRowFilter.java new file mode 100644 index 0000000000..63a33479ad --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/GlobalIndexLiveRowFilter.java @@ -0,0 +1,113 @@ +/* + * 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.Snapshot; +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.snapshot.TimeTravelUtil; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; + +/** Live-row filtering shared by global-index based search readers. */ +class GlobalIndexLiveRowFilter { + + @Nullable + static RoaringNavigableMap64 liveRows( + @Nullable FileStoreTable table, @Nullable PartitionPredicate partitionFilter) { + if (table == null || !table.coreOptions().deletionVectorsEnabled()) { + return null; + } + + @Nullable Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table); + if (snapshot == null) { + return null; + } + + RoaringNavigableMap64 liveRows = new RoaringNavigableMap64(); + for (Split split : + table.newSnapshotReader() + .withPartitionFilter(partitionFilter) + .withMode(ScanMode.ALL) + .withSnapshot(snapshot) + .read() + .splits()) { + if (split instanceof DataSplit) { + addLiveRows(table, liveRows, (DataSplit) split); + } + } + return liveRows; + } + + @Nullable + static RoaringNavigableMap64 forRange( + @Nullable RoaringNavigableMap64 liveRows, long from, long to) { + if (liveRows == null) { + return null; + } + + Range range = new Range(from, to); + RoaringNavigableMap64 includeRows = new RoaringNavigableMap64(); + includeRows.addRange(range); + includeRows.and(liveRows); + return includeRows.getLongCardinality() == range.count() ? null : includeRows; + } + + private static void addLiveRows( + FileStoreTable table, RoaringNavigableMap64 liveRows, DataSplit split) { + List<DataFileMeta> files = split.dataFiles(); + List<DeletionFile> deletionFiles = split.deletionFiles().orElse(null); + DeletionVector.Factory deletionVectorFactory = + DeletionVector.factory(table.fileIO(), files, deletionFiles); + for (DataFileMeta file : files) { + if (file.firstRowId() == null) { + continue; + } + long firstRowId = file.nonNullFirstRowId(); + liveRows.addRange(file.nonNullRowIdRange()); + + Optional<DeletionVector> deletionVector; + try { + deletionVector = deletionVectorFactory.create(file.fileName()); + } catch (IOException e) { + throw new RuntimeException( + "Failed to read deletion vector for file " + file.fileName(), e); + } + if (!deletionVector.isPresent() || deletionVector.get().isEmpty()) { + continue; + } + + RoaringNavigableMap64 deletedRows = new RoaringNavigableMap64(); + deletionVector + .get() + .forEachDeletedPosition(position -> deletedRows.add(firstRowId + position)); + liveRows.andNot(deletedRows); + } + } + + private GlobalIndexLiveRowFilter() {} +} 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 0000000000..55ab396c77 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/DeletionVectorTestUtils.java @@ -0,0 +1,118 @@ +/* + * 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.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.utils.Range; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET; + +/** Test utilities for committing deletion vectors. */ +class DeletionVectorTestUtils { + + static void commitDeletionVectors(FileStoreTable table, long... deletedRowIds) + throws Exception { + Map<String, DeletionVector> deletionVectors = new HashMap<>(); + List<DataFileMeta> dataFiles = dataFiles(table); + for (long rowId : deletedRowIds) { + DataFileMeta file = dataFileContaining(dataFiles, rowId); + deletionVectors + .computeIfAbsent(file.fileName(), ignored -> new BitmapDeletionVector()) + .delete(rowId - file.nonNullFirstRowId()); + } + + BaseAppendDeleteFileMaintainer maintainer = + BaseAppendDeleteFileMaintainer.forUnawareAppend( + table.store().newIndexFileHandler(), + table.latestSnapshot().get(), + BinaryRow.EMPTY_ROW); + for (Map.Entry<String, DeletionVector> entry : deletionVectors.entrySet()) { + maintainer.notifyNewDeletionVector(entry.getKey(), entry.getValue()); + } + + List<IndexFileMeta> newIndexFiles = new ArrayList<>(); + List<IndexFileMeta> 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()); + } + } + + CommitMessage message = + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + UNAWARE_BUCKET, + null, + new DataIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + newIndexFiles, + deletedIndexFiles), + CompactIncrement.emptyIncrement()); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(Collections.singletonList(message)); + } + } + + private static List<DataFileMeta> dataFiles(FileStoreTable table) { + List<DataFileMeta> dataFiles = new ArrayList<>(); + for (ManifestEntry entry : table.store().newScan().plan().files()) { + dataFiles.add(entry.file()); + } + return dataFiles; + } + + private static DataFileMeta dataFileContaining(List<DataFileMeta> dataFiles, long rowId) { + for (DataFileMeta file : dataFiles) { + if (file.firstRowId() == null) { + continue; + } + Range range = file.nonNullRowIdRange(); + if (range.from <= rowId && rowId <= range.to) { + return file; + } + } + throw new IllegalArgumentException("No data file contains row id " + rowId); + } + + private DeletionVectorTestUtils() {} +} 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 5f61bc5de6..a9d8e2475b 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 @@ -65,6 +65,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import static org.apache.paimon.table.source.DeletionVectorTestUtils.commitDeletionVectors; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -125,6 +126,40 @@ public class FullTextSearchBuilderTest extends TableTestBase { assertThat(ids).containsAnyOf(0, 1, 3); } + @Test + public void testFullTextSearchExcludesDeletedIndexedRows() throws Exception { + Identifier identifier = identifier("full_text_deleted_indexed_rows"); + Schema schema = + 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(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = getTable(identifier); + + String[] documents = { + "paimon keyword", "paimon keyword", "paimon keyword", "paimon keyword" + }; + writeDocuments(table, documents); + buildAndCommitIndex(table, documents); + commitDeletionVectors(table, 0L, 1L); + + GlobalIndexResult result = + table.newFullTextSearchBuilder() + .withQuery(FullTextQuery.match("keyword", TEXT_FIELD_NAME)) + .withLimit(2) + .executeLocal(); + + assertThat(result.results().getLongCardinality()).isEqualTo(2); + assertThat(result.results()).contains(2L, 3L); + assertThat(result.results()).doesNotContain(0L, 1L); + assertThat(readIds(table, result)).containsExactlyInAnyOrder(2, 3); + } + @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 b937efe223..fdac8eef8a 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 @@ -71,6 +71,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import static org.apache.paimon.table.source.DeletionVectorTestUtils.commitDeletionVectors; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -186,6 +187,63 @@ public class VectorSearchBuilderTest extends TableTestBase { assertThat(ids).contains(0); } + @Test + public void testVectorSearchExcludesDeletedIndexedRows() throws Exception { + catalog.createTable( + identifier("vector_search_deleted_indexed_rows"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier("vector_search_deleted_indexed_rows")); + + float[][] vectors = {{0.0f, 0.0f}, {1.0f, 0.0f}, {2.0f, 0.0f}, {3.0f, 0.0f}}; + writeVectors(table, vectors); + buildAndCommitIndex(table, vectors); + commitDeletionVectors(table, 0L, 1L); + + GlobalIndexResult result = + table.newVectorSearchBuilder() + .withVector(new float[] {0.0f, 0.0f}) + .withLimit(2) + .withVectorColumn(VECTOR_FIELD_NAME) + .executeLocal(); + + assertThat(result.results().getLongCardinality()).isEqualTo(2); + assertThat(result.results()).contains(2L, 3L); + assertThat(result.results()).doesNotContain(0L, 1L); + assertThat(readIds(table, result)).containsExactly(2, 3); + } + + @Test + public void testBatchVectorSearchExcludesDeletedIndexedRows() throws Exception { + catalog.createTable( + identifier("batch_vector_search_deleted_indexed_rows"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier("batch_vector_search_deleted_indexed_rows")); + + float[][] vectors = {{0.0f, 0.0f}, {1.0f, 0.0f}, {2.0f, 0.0f}, {3.0f, 0.0f}}; + writeVectors(table, vectors); + buildAndCommitIndex(table, vectors); + commitDeletionVectors(table, 0L, 3L); + + List<GlobalIndexResult> results = + table.newBatchVectorSearchBuilder() + .withVectors(new float[][] {{0.0f, 0.0f}, {3.0f, 0.0f}}) + .withLimit(1) + .withVectorColumn(VECTOR_FIELD_NAME) + .executeBatchLocal(); + + assertThat(results).hasSize(2); + assertThat(results.get(0).results()).contains(1L); + assertThat(results.get(0).results()).doesNotContain(0L); + assertThat(results.get(1).results()).contains(2L); + assertThat(results.get(1).results()).doesNotContain(3L); + } + @Test public void testVectorSearchWithCosineMetric() throws Exception { // Create a table with cosine metric diff --git a/paimon-python/pypaimon/globalindex/full_text_search.py b/paimon-python/pypaimon/globalindex/full_text_search.py index d1511df2e3..dc6a3dd02d 100644 --- a/paimon-python/pypaimon/globalindex/full_text_search.py +++ b/paimon-python/pypaimon/globalindex/full_text_search.py @@ -32,10 +32,12 @@ class FullTextSearch: Attributes: query: The structured full-text query limit: Maximum number of results to return + include_row_ids: Optional bitmap of row IDs to include in search """ query: FullTextQuery limit: int + include_row_ids: Optional['RoaringBitmap64'] = None def __post_init__(self): if self.query is None: @@ -54,6 +56,29 @@ class FullTextSearch: def query_json(self) -> str: return self.query.to_json() + def with_include_row_ids(self, include_row_ids: 'RoaringBitmap64') -> 'FullTextSearch': + """Return a new FullTextSearch with the specified include_row_ids.""" + return FullTextSearch( + query=self.query, + limit=self.limit, + include_row_ids=include_row_ids, + ) + + def offset_range(self, from_: int, to: int) -> 'FullTextSearch': + """Offset include_row_ids into the given range.""" + if self.include_row_ids is None: + return self + + from pypaimon.utils.roaring_bitmap import RoaringBitmap64 + + range_bitmap = RoaringBitmap64() + range_bitmap.add_range(from_, to) + and_result = RoaringBitmap64.and_(range_bitmap, self.include_row_ids) + offset_bitmap = RoaringBitmap64() + for row_id in and_result: + offset_bitmap.add(row_id - from_) + return self.with_include_row_ids(offset_bitmap) + def visit(self, visitor: 'GlobalIndexReader') -> 'Future[Optional[ScoredGlobalIndexResult]]': """Visit the global index reader with this full-text search.""" return visitor.visit_full_text_search(self) diff --git a/paimon-python/pypaimon/globalindex/offset_global_index_reader.py b/paimon-python/pypaimon/globalindex/offset_global_index_reader.py index f13145ff50..ce370f1336 100644 --- a/paimon-python/pypaimon/globalindex/offset_global_index_reader.py +++ b/paimon-python/pypaimon/globalindex/offset_global_index_reader.py @@ -61,7 +61,8 @@ class OffsetGlobalIndexReader(GlobalIndexReader): def visit_full_text_search(self, full_text_search) -> 'Future[Optional[GlobalIndexResult]]': return self._apply_offset_future( - self._wrapped.visit_full_text_search(full_text_search)) + self._wrapped.visit_full_text_search( + full_text_search.offset_range(self._offset, self._to))) def visit_equal(self, field_ref: FieldRef, literal: object) -> 'Future[Optional[GlobalIndexResult]]': return self._apply_offset_future(self._wrapped.visit_equal(field_ref, literal)) diff --git a/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py b/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py index 441420a872..d4b1a88182 100644 --- a/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py +++ b/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py @@ -346,8 +346,22 @@ class TantivyFullTextGlobalIndexReader(GlobalIndexReader): import tantivy + include_row_ids = full_text_search.include_row_ids + search_limit = limit + if include_row_ids is not None and not include_row_ids.is_empty(): + search_limit = max(limit, include_row_ids.cardinality()) + id_to_scores = self._search_full_text_query( - tantivy, full_text_search.query, limit) + tantivy, full_text_search.query, search_limit) + if include_row_ids is not None: + if include_row_ids.is_empty(): + id_to_scores = {} + else: + id_to_scores = { + row_id: score + for row_id, score in id_to_scores.items() + if include_row_ids.contains(row_id) + } return _completed_future( DictBasedScoredIndexResult(id_to_scores).top_k(limit)) diff --git a/paimon-python/pypaimon/table/source/full_text_read.py b/paimon-python/pypaimon/table/source/full_text_read.py index e35b081d9f..87fa4d5362 100644 --- a/paimon-python/pypaimon/table/source/full_text_read.py +++ b/paimon-python/pypaimon/table/source/full_text_read.py @@ -37,6 +37,7 @@ from pypaimon.globalindex.vector_search_result import ( DictBasedScoredIndexResult, ScoredGlobalIndexResult, ) +from pypaimon.table.source import global_index_live_row_filter from pypaimon.table.source.full_text_search_split import FullTextSearchSplit from pypaimon.table.source.full_text_scan import FullTextScanPlan from pypaimon.utils.roaring_bitmap import RoaringBitmap64 @@ -61,12 +62,14 @@ class FullTextReadImpl(FullTextRead): table: 'FileStoreTable', limit: int, text_column, - query: FullTextQuery + query: FullTextQuery, + partition_filter=None, ): self._table = table self._limit = limit self._text_columns = text_column if isinstance(text_column, list) else [text_column] self._query = query + self._partition_filter = partition_filter def read(self, splits: List[FullTextSearchSplit]) -> GlobalIndexResult: if not splits: @@ -75,38 +78,45 @@ class FullTextReadImpl(FullTextRead): splits_by_column: Dict[str, List[FullTextSearchSplit]] = {} for split in splits: splits_by_column.setdefault(split.column_name, []).append(split) - return self._eval_query(self._query, splits_by_column).top_k(self._limit) + live_rows = global_index_live_row_filter.live_rows( + self._table, self._partition_filter) + return self._eval_query( + self._query, splits_by_column, live_rows).top_k(self._limit) def _eval_query( self, query: FullTextQuery, - splits_by_column: Dict[str, List[FullTextSearchSplit]] + splits_by_column: Dict[str, List[FullTextSearchSplit]], + live_rows ) -> ScoredGlobalIndexResult: if isinstance(query, MatchQuery): - return self._eval_column_query(query, query.column, splits_by_column) + return self._eval_column_query( + query, query.column, splits_by_column, live_rows) if isinstance(query, PhraseQuery): - return self._eval_column_query(query, query.column, splits_by_column) + return self._eval_column_query( + query, query.column, splits_by_column, live_rows) if isinstance(query, MultiMatchQuery): results = [] for column, boost in zip(query.columns, query.boosts): match = MatchQuery( query.query, column, boost=boost, operator=query.operator) results.append( - self._eval_column_query(match, column, splits_by_column)) + self._eval_column_query(match, column, splits_by_column, live_rows)) return _or(results).top_k(self._limit) if isinstance(query, BoostQuery): - positive = self._eval_query(query.positive, splits_by_column) - negative = self._eval_query(query.negative, splits_by_column) + positive = self._eval_query(query.positive, splits_by_column, live_rows) + negative = self._eval_query(query.negative, splits_by_column, live_rows) return _boost(positive, negative, query.negative_boost) if isinstance(query, BooleanQuery): result = None for child in query.must(): - child_result = self._eval_query(child, splits_by_column) + child_result = self._eval_query(child, splits_by_column, live_rows) result = child_result if result is None else _and(result, child_result) should_results = [] for child in query.should(): - should_results.append(self._eval_query(child, splits_by_column)) + should_results.append( + self._eval_query(child, splits_by_column, live_rows)) if should_results: should_result = _or(should_results) result = should_result if result is None else _and_with_bonus( @@ -115,7 +125,8 @@ class FullTextReadImpl(FullTextRead): if result is None: return ScoredGlobalIndexResult.create_empty() for child in query.must_not(): - result = _and_not(result, self._eval_query(child, splits_by_column)) + result = _and_not( + result, self._eval_query(child, splits_by_column, live_rows)) return result raise ValueError("Unsupported full-text query type: %s" % type(query).__name__) @@ -123,19 +134,26 @@ class FullTextReadImpl(FullTextRead): self, query: FullTextQuery, column: str, - splits_by_column: Dict[str, List[FullTextSearchSplit]] + splits_by_column: Dict[str, List[FullTextSearchSplit]], + live_rows ) -> ScoredGlobalIndexResult: splits = splits_by_column.get(column, []) if not splits: return ScoredGlobalIndexResult.create_empty() - futures = [ - self._eval( + futures = [] + for split in splits: + include_row_ids = global_index_live_row_filter.for_range( + live_rows, split.row_range_start, split.row_range_end) + if include_row_ids is not None and include_row_ids.is_empty(): + continue + futures.append(self._eval( split.row_range_start, split.row_range_end, split.full_text_index_files, query, - ) - for split in splits - ] + include_row_ids, + )) + if not futures: + return ScoredGlobalIndexResult.create_empty() wait(futures) @@ -150,7 +168,8 @@ class FullTextReadImpl(FullTextRead): return DictBasedScoredIndexResult(merged_scores).top_k(self._limit) - def _eval(self, row_range_start, row_range_end, full_text_index_files, query): + def _eval(self, row_range_start, row_range_end, full_text_index_files, + query, include_row_ids): index_io_meta_list = [] for index_file in full_text_index_files: meta = index_file.global_index_meta @@ -177,6 +196,8 @@ class FullTextReadImpl(FullTextRead): query=query, limit=_candidate_limit(row_range_start, row_range_end), ) + if include_row_ids is not None: + full_text_search = full_text_search.with_include_row_ids(include_row_ids) offset_reader = OffsetGlobalIndexReader(reader, row_range_start, row_range_end) future = offset_reader.visit_full_text_search(full_text_search) diff --git a/paimon-python/pypaimon/table/source/full_text_search_builder.py b/paimon-python/pypaimon/table/source/full_text_search_builder.py index 49393d63da..dcc252c54b 100644 --- a/paimon-python/pypaimon/table/source/full_text_search_builder.py +++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py @@ -123,7 +123,11 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder): if self._limit <= 0: raise ValueError("Limit must be positive, set via with_limit()") return FullTextReadImpl( - self._table, self._limit, self._text_columns(), self._query + self._table, + self._limit, + self._text_columns(), + self._query, + partition_filter=self._partition_filter, ) def _text_columns(self): diff --git a/paimon-python/pypaimon/table/source/global_index_live_row_filter.py b/paimon-python/pypaimon/table/source/global_index_live_row_filter.py new file mode 100644 index 0000000000..68b08a80d0 --- /dev/null +++ b/paimon-python/pypaimon/table/source/global_index_live_row_filter.py @@ -0,0 +1,85 @@ +# 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. + +"""Live-row filtering shared by global-index based search readers.""" + +from typing import Optional + +from pypaimon.deletionvectors.deletion_vector import DeletionVector +from pypaimon.read.split import DataSplit +from pypaimon.utils.range import Range +from pypaimon.utils.roaring_bitmap import RoaringBitmap64 + + +def live_rows(table, partition_filter=None) -> Optional[RoaringBitmap64]: + """Return current live global row ids for deletion-vector tables. + + ``None`` means no live-row filter is needed. This keeps tables without + deletion vectors on the old zero-overhead path. + """ + options = getattr(table, "options", None) + deletion_vectors_enabled = getattr(options, "deletion_vectors_enabled", None) + if (table is None + or not callable(deletion_vectors_enabled) + or not deletion_vectors_enabled(False)): + return None + + read_builder = table.new_read_builder() + if partition_filter is not None: + read_builder = read_builder.with_partition_filter(partition_filter) + + rows = RoaringBitmap64() + for split in read_builder.new_scan().plan().splits(): + if isinstance(split, DataSplit): + rows = _add_live_rows(table, rows, split) + return rows + + +def for_range(live_row_ids: Optional[RoaringBitmap64], + from_: int, to: int) -> Optional[RoaringBitmap64]: + if live_row_ids is None: + return None + + row_range = Range(from_, to) + include = RoaringBitmap64() + include.add_range(row_range.from_, row_range.to) + include = RoaringBitmap64.and_(include, live_row_ids) + return None if include.cardinality() == row_range.count() else include + + +def _add_live_rows(table, rows: RoaringBitmap64, split: DataSplit) -> RoaringBitmap64: + deletion_files = split.data_deletion_files or [] + for i, data_file in enumerate(split.files): + row_id_range = data_file.row_id_range() + if row_id_range is None: + continue + + rows.add_range(row_id_range.from_, row_id_range.to) + deletion_file = deletion_files[i] if i < len(deletion_files) else None + if deletion_file is None or deletion_file.cardinality == 0: + continue + + deletion_vector = DeletionVector.read(table.file_io, deletion_file) + if deletion_vector.is_empty(): + continue + + deleted_rows = RoaringBitmap64() + first_row_id = data_file.first_row_id + for position in deletion_vector.bit_map(): + deleted_rows.add(first_row_id + position) + rows = RoaringBitmap64.remove_all(rows, deleted_rows) + return rows diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py b/paimon-python/pypaimon/table/source/vector_search_read.py index b0a229f043..bcb142e4f1 100644 --- a/paimon-python/pypaimon/table/source/vector_search_read.py +++ b/paimon-python/pypaimon/table/source/vector_search_read.py @@ -30,6 +30,7 @@ from pypaimon.table.source.vector_search_split import ( IndexVectorSearchSplit, RawVectorSearchSplit, ) +from pypaimon.table.source import global_index_live_row_filter from pypaimon.utils.range import Range from pypaimon.utils.roaring_bitmap import RoaringBitmap64 @@ -81,10 +82,38 @@ class AbstractVectorSearchReadImpl: def _pre_filters(self, splits): # type: (list) -> List[RoaringBitmap64] - """Evaluate scalar indexes and return one include bitmap per index split.""" - if self._filter is None: + """Evaluate live-row/scalar filters and return one bitmap per index split.""" + if not splits: + return [] + + live_rows = global_index_live_row_filter.live_rows( + self._table, self._partition_filter) + matched_rows = self._scalar_matched_rows(splits) + if live_rows is None and matched_rows is None: return [] + include_row_ids = [] + has_filter = False + for split in splits: + split_range = Range(split.row_range_start, split.row_range_end) + include = _bitmap_of_range(split_range) + if live_rows is not None: + include = RoaringBitmap64.and_(include, live_rows) + if matched_rows is not None: + include = RoaringBitmap64.and_(include, matched_rows) + + if include.cardinality() == split_range.count(): + include_row_ids.append(None) + else: + include_row_ids.append(include) + has_filter = True + return include_row_ids if has_filter else [] + + def _scalar_matched_rows(self, splits): + """Evaluate scalar indexes and return matching global row ids.""" + if self._filter is None: + return None + # Collect scalar index files across splits, deduplicated by file name. seen = set() scalar_files = [] @@ -96,7 +125,7 @@ class AbstractVectorSearchReadImpl: scalar_files.append(index_file) if not scalar_files: - return _empty_bitmaps(len(splits)) + return RoaringBitmap64() from pypaimon.globalindex.global_index_scanner import GlobalIndexScanner scanner = GlobalIndexScanner.create( @@ -105,30 +134,26 @@ class AbstractVectorSearchReadImpl: partition_filter=self._partition_filter, ) if scanner is None: - return _empty_bitmaps(len(splits)) + return RoaringBitmap64() try: result = scanner.scan(self._filter) if result is None: - return _empty_bitmaps(len(splits)) - matched_rows = result.results() + return RoaringBitmap64() + return result.results() finally: scanner.close() - include_row_ids = [] - for split in splits: - split_rows = _bitmap_of_range( - Range(split.row_range_start, split.row_range_end)) - include_row_ids.append(RoaringBitmap64.and_(matched_rows, split_rows)) - return include_row_ids - def _pre_filter(self, splits): # Backwards-compatible helper used by older tests/callers. pre_filters = self._pre_filters(splits) if not pre_filters: return None merged = RoaringBitmap64() - for bitmap in pre_filters: - merged = RoaringBitmap64.or_(merged, bitmap) + for split, bitmap in zip(splits, pre_filters): + if bitmap is None: + merged.add_range(split.row_range_start, split.row_range_end) + else: + merged = RoaringBitmap64.or_(merged, bitmap) return merged def _raw_pre_filter(self, splits): diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 47bebcd21f..3331d269ea 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -121,6 +121,13 @@ def _entry(partition_row, field_id, index_type, file_name, index_file=index_file) +def _bitmap(*row_ids): + bitmap = RoaringBitmap64() + for row_id in row_ids: + bitmap.add(row_id) + return bitmap + + def _install_raw_vector_read_builder(table, vector_column_name, row_id_to_vector, calls=None): """Install a fake raw read builder which honors GlobalIndexResult ranges.""" @@ -462,6 +469,99 @@ class _FakeTantivy(types.SimpleNamespace): # ----------------------------- tests --------------------------------------- +class GlobalIndexLiveRowFilterTest(unittest.TestCase): + + def tearDown(self): + mock.patch.stopall() + + def test_live_rows_noops_without_deletion_vectors(self): + from pypaimon.table.source import global_index_live_row_filter + + class _Options: + def deletion_vectors_enabled(self_inner, default=False): + return False + + class _Table: + options = _Options() + + def new_read_builder(self_inner): + raise AssertionError("non-DV table must not be scanned") + + self.assertIsNone(global_index_live_row_filter.live_rows(_Table())) + + def test_live_rows_subtracts_deletion_vector_positions(self): + from pypaimon.read.split import DataSplit + from pypaimon.table.source import global_index_live_row_filter + from pypaimon.table.source.deletion_file import DeletionFile + + calls = {} + + class _Options: + def deletion_vectors_enabled(self_inner, default=False): + return True + + class _File: + first_row_id = 10 + row_count = 5 + + def row_id_range(self_inner): + return Range(10, 14) + + deletion_file = DeletionFile("dv", 0, 1, cardinality=2) + split = DataSplit( + files=[_File()], + partition=None, + bucket=0, + data_deletion_files=[deletion_file], + ) + + class _Plan: + def splits(self_inner): + return [split] + + class _Scan: + def plan(self_inner): + return _Plan() + + class _Builder: + def with_partition_filter(self_inner, predicate): + calls["partition_filter"] = predicate + return self_inner + + def new_scan(self_inner): + calls["new_scan"] = True + return _Scan() + + class _Table: + options = _Options() + file_io = object() + + def new_read_builder(self_inner): + calls["new_read_builder"] = True + return _Builder() + + class _DeletionVector: + def is_empty(self_inner): + return False + + def bit_map(self_inner): + return [1, 3] + + partition_filter = object() + with mock.patch( + "pypaimon.table.source.global_index_live_row_filter." + "DeletionVector.read", + return_value=_DeletionVector()) as read: + rows = global_index_live_row_filter.live_rows( + _Table(), partition_filter) + + read.assert_called_once_with(_Table.file_io, deletion_file) + self.assertIs(partition_filter, calls["partition_filter"]) + self.assertTrue(calls["new_read_builder"]) + self.assertTrue(calls["new_scan"]) + self.assertEqual([10, 12, 14], rows.to_list()) + + class VectorReaderFactoryTest(unittest.TestCase): """Vector reader factory compatibility.""" @@ -900,6 +1000,36 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase): [_fake_query_text(q) for q in tantivy.last_index.searcher_instance.queries], ) + def test_reader_applies_include_row_ids_before_top_k(self): + from pypaimon.globalindex.full_text_search import FullTextSearch + from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import ( + TantivyFullTextGlobalIndexReader, + ) + + tantivy = _FakeTantivy() + old_tantivy = sys.modules.get("tantivy") + sys.modules["tantivy"] = tantivy + try: + reader = TantivyFullTextGlobalIndexReader( + _FakeFileIO(), + "/unused", + [GlobalIndexIOMeta(file_name="ft.index", file_size=1)]) + try: + search = FullTextSearch( + MatchQuery("positive", "content"), 10 + ).with_include_row_ids(_bitmap(2)) + result = reader.visit_full_text_search(search).result() + finally: + reader.close() + finally: + if old_tantivy is None: + sys.modules.pop("tantivy", None) + else: + sys.modules["tantivy"] = old_tantivy + + self.assertEqual([2], sorted(list(result.results()))) + self.assertEqual(5.0, result.score_getter()(2)) + def test_reader_rejects_java_unsupported_match_options(self): from pypaimon.globalindex.full_text_search import FullTextSearch from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import ( @@ -1519,11 +1649,14 @@ class VectorSearchFilterTest(unittest.TestCase): self.assertEqual(1, scanner.scan.call_count) self.assertIs(filter_pred, scanner.scan.call_args[0][0]) - # [0,4] sees empty local bitmap; [5,9] sees {0..4}. - self.assertEqual( - [0, 5], - sorted(vs.include_row_ids.cardinality() - for vs in captured_searches)) + # [0,4] sees empty local bitmap; [5,9] is fully included and stays None. + include_summary = sorted( + ("all", None) + if vs.include_row_ids is None + else ("bitmap", vs.include_row_ids.cardinality()) + for vs in captured_searches + ) + self.assertEqual([("all", None), ("bitmap", 0)], include_summary) # Vector reader io_meta carries external_path from IndexFileMeta. seen_paths = {meta.external_path @@ -1533,6 +1666,112 @@ class VectorSearchFilterTest(unittest.TestCase): {"oss://bucket/vec-0.index", "oss://bucket/vec-1.index"}, seen_paths) + def test_indexed_vector_search_filters_deleted_rows(self): + from pypaimon.globalindex.vector_search_result import ( + DictBasedScoredIndexResult, + ) + from pypaimon.table.source.vector_search_read import VectorSearchReadImpl + from pypaimon.table.source.vector_search_split import IndexVectorSearchSplit + + entry = _entry(None, field_id=1, index_type="lumina-vector-ann", + file_name="vec.index", row_range_start=10, + row_range_end=14) + table = _StubTable(fields=[self.embedding_field], entries=[entry]) + split = IndexVectorSearchSplit( + row_range_start=10, + row_range_end=14, + vector_index_files=[entry.index_file], + ) + live_rows = _bitmap(10, 12, 13, 14) + captured = [] + + def _fake_create(index_type, file_io, index_path, + index_io_meta_list, options=None): + class _FakeReader: + def visit_vector_search(self_inner, vs): + captured.append(vs.include_row_ids) + return _completed_future( + DictBasedScoredIndexResult({2: 1.0})) + + def close(self_inner): + pass + + return _FakeReader() + + with mock.patch( + "pypaimon.table.source.vector_search_read." + "global_index_live_row_filter.live_rows", + return_value=live_rows), \ + mock.patch( + "pypaimon.table.source.vector_search_read._create_vector_reader", + side_effect=_fake_create): + result = VectorSearchReadImpl( + table, + limit=3, + vector_column=self.embedding_field, + query_vector=[1.0], + ).read([split]) + + self.assertEqual([[0, 2, 3, 4]], [b.to_list() for b in captured]) + self.assertEqual([12], sorted(list(result.results()))) + + def test_batch_indexed_vector_search_filters_deleted_rows(self): + from pypaimon.globalindex.global_index_reader import GlobalIndexReader + from pypaimon.globalindex.vector_search_result import ( + DictBasedScoredIndexResult, + ) + from pypaimon.table.source.vector_search_read import ( + BatchVectorSearchReadImpl, + ) + from pypaimon.table.source.vector_search_split import IndexVectorSearchSplit + + entry = _entry(None, field_id=1, index_type="lumina-vector-ann", + file_name="vec.index", row_range_start=10, + row_range_end=14) + table = _StubTable(fields=[self.embedding_field], entries=[entry]) + split = IndexVectorSearchSplit( + row_range_start=10, + row_range_end=14, + vector_index_files=[entry.index_file], + ) + live_rows = _bitmap(10, 12, 13, 14) + captured = [] + + def _fake_create(index_type, file_io, index_path, + index_io_meta_list, options=None): + class _FakeReader(GlobalIndexReader): + def visit_batch_vector_search(self_inner, bvs): + captured.append(bvs.include_row_ids) + return _completed_future([ + DictBasedScoredIndexResult({2: 1.0}) + for _ in range(bvs.vector_count) + ]) + + def close(self_inner): + pass + + return _FakeReader() + + with mock.patch( + "pypaimon.table.source.vector_search_read." + "global_index_live_row_filter.live_rows", + return_value=live_rows), \ + mock.patch( + "pypaimon.table.source.vector_search_read._create_vector_reader", + side_effect=_fake_create): + results = BatchVectorSearchReadImpl( + table, + limit=3, + vector_column=self.embedding_field, + query_vectors=[[1.0], [2.0]], + ).read_batch([split]) + + self.assertEqual([[0, 2, 3, 4]], [b.to_list() for b in captured]) + self.assertEqual([[12], [12]], [ + sorted(list(result.results())) + for result in results + ]) + def test_read_threads_options_to_vector_search(self): scan_plan = self._builder().new_vector_search_scan().scan() @@ -2953,6 +3192,64 @@ class FullTextSearchManySplitsTest(unittest.TestCase): self.assertEqual("oss://bucket/ft.index", captured_io_metas[0][0].external_path) + def test_full_text_read_filters_deleted_rows(self): + from pypaimon.globalindex.vector_search_result import ( + DictBasedScoredIndexResult, + ) + from pypaimon.table.source.full_text_read import FullTextReadImpl + from pypaimon.table.source.full_text_search_split import ( + FullTextSearchSplit, + ) + + text_field = _field(1, "content", "STRING") + entry = _entry(None, field_id=1, index_type="tantivy-fulltext", + file_name="ft.index", row_range_start=10, + row_range_end=14) + table = _StubTable(fields=[text_field], entries=[entry]) + split = FullTextSearchSplit( + column_name="content", + row_range_start=10, + row_range_end=14, + full_text_index_files=[entry.index_file], + ) + live_rows = _bitmap(10, 12, 13, 14) + partition_filter = object() + captured = [] + + def _fake_create(index_type, file_io, index_path, index_io_meta_list): + class _FakeReader: + def visit_full_text_search(self_inner, fts): + captured.append(fts.include_row_ids) + return _completed_future( + DictBasedScoredIndexResult({ + row_id: float(row_id) + for row_id in fts.include_row_ids + })) + + def close(self_inner): + pass + + return _FakeReader() + + with mock.patch( + "pypaimon.table.source.full_text_read." + "global_index_live_row_filter.live_rows", + return_value=live_rows) as live_rows_fn, \ + mock.patch( + "pypaimon.table.source.full_text_read._create_full_text_reader", + side_effect=_fake_create): + result = FullTextReadImpl( + table, + limit=10, + text_column=text_field, + query=MatchQuery("test", "content"), + partition_filter=partition_filter, + ).read([split]) + + live_rows_fn.assert_called_once_with(table, partition_filter) + self.assertEqual([[0, 2, 3, 4]], [b.to_list() for b in captured]) + self.assertEqual([10, 12, 13, 14], sorted(list(result.results()))) + def test_full_text_search_with_many_splits(self): from pypaimon.globalindex.vector_search_result import ( DictBasedScoredIndexResult, diff --git a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java index 4c45df6111..1bce33d07c 100644 --- a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java +++ b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java @@ -31,6 +31,8 @@ import org.apache.paimon.tantivy.StreamFileInput; import org.apache.paimon.tantivy.TantivySearcher; import org.apache.paimon.utils.RoaringNavigableMap64; +import javax.annotation.Nullable; + import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -96,7 +98,7 @@ public class TantivyFullTextGlobalIndexReader implements GlobalIndexReader { SearchResult result = borrowed.searcher.searchJson( fullTextSearch.queryJson(), fullTextSearch.limit()); - return Optional.of(toScoredResult(result)); + return Optional.of(toScoredResult(result, fullTextSearch.includeRowIds())); } catch (IOException e) { throw new RuntimeException("Failed to search Tantivy full-text index", e); } @@ -104,11 +106,15 @@ public class TantivyFullTextGlobalIndexReader implements GlobalIndexReader { executor); } - private ScoredGlobalIndexResult toScoredResult(SearchResult result) { + private ScoredGlobalIndexResult toScoredResult( + SearchResult result, @Nullable RoaringNavigableMap64 includeRowIds) { RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); HashMap<Long, Float> id2scores = new HashMap<>(result.size()); for (int i = 0; i < result.size(); i++) { long rowId = result.getRowIds()[i]; + if (includeRowIds != null && !includeRowIds.contains(rowId)) { + continue; + } bitmap.add(rowId); id2scores.put(rowId, result.getScores()[i]); }
