This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 077788e12a [core][python] filter deleted rows from global index search
(#8459)
077788e12a is described below
commit 077788e12aaafe266eccc4f3e04abda81c1fe69a
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jul 4 21:51:32 2026 +0800
[core][python] filter deleted rows from global index search (#8459)
Filter deletion-vector deleted rows out of global-index based vector and
full-text search results without rebuilding existing index files. The
search path now derives live row ids from the current snapshot and
applies them as include-row filters so deleted rows cannot affect topK
results.
---
.../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 | 32 +-
.../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 | 321 ++++++++++++++++++++-
.../index/TantivyFullTextGlobalIndexReader.java | 24 +-
.../index/TantivyFullTextGlobalIndexTest.java | 30 ++
paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs | 11 +
.../org/apache/paimon/tantivy/TantivySearcher.java | 7 +
.../org/apache/paimon/tantivy/TantivyJniTest.java | 2 +
25 files changed, 1075 insertions(+), 90 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..8e3d22c7be 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,21 @@ class TantivyFullTextGlobalIndexReader(GlobalIndexReader):
import tantivy
+ include_row_ids = full_text_search.include_row_ids
+ search_limit = limit
+ if include_row_ids is not None:
+ if include_row_ids.is_empty():
+ return _completed_future(DictBasedScoredIndexResult({}))
+ search_limit = self._required_shard_doc_count()
+
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:
+ 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))
@@ -377,11 +390,26 @@ class TantivyFullTextGlobalIndexReader(GlobalIndexReader):
self._parse_structured_query(tantivy, query), limit)
def _child_query_limit(self, limit):
- num_docs = getattr(self._searcher, "num_docs", None)
+ num_docs = self._shard_doc_count()
if num_docs is None:
return limit
return max(limit, int(num_docs))
+ def _required_shard_doc_count(self):
+ num_docs = self._shard_doc_count()
+ if num_docs is None:
+ raise RuntimeError(
+ "PyPaimon Tantivy full-text search with include_row_ids "
+ "requires a tantivy-py Searcher with num_docs support."
+ )
+ return max(0, int(num_docs))
+
+ def _shard_doc_count(self):
+ num_docs = getattr(self._searcher, "num_docs", None)
+ if callable(num_docs):
+ num_docs = num_docs()
+ return None if num_docs is None else int(num_docs)
+
def _search_boolean_query(self, tantivy, query, limit):
from pypaimon.globalindex.full_text_query import Occur
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..b74a9f3d64 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."""
@@ -386,15 +393,25 @@ class _FakeSearcher:
def __init__(self):
self.query = None
self.queries = []
+ self.num_docs = 3
def search(self, query, limit):
self.query = query
self.queries.append(query)
query_text = _fake_query_text(query)
if query_text == "positive":
- return _FakeSearchResults([(10.0, ("addr", 1)), (5.0, ("addr",
2))])
+ hits = [(10.0, ("addr", 1)), (5.0, ("addr", 2))]
+ return _FakeSearchResults(hits[:limit])
if query_text == "negative":
- return _FakeSearchResults([(7.0, ("addr", 2))])
+ hits = [(7.0, ("addr", 2))]
+ return _FakeSearchResults(hits[:limit])
+ if query_text == "ranked":
+ hits = [
+ (10.0, ("addr", 1)),
+ (9.0, ("addr", 2)),
+ (8.0, ("addr", 3)),
+ ]
+ return _FakeSearchResults(hits[:limit])
return _FakeSearchResults()
def fast_field_values(self, name, addresses):
@@ -462,6 +479,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 +1010,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("ranked", "content"), 1
+ ).with_include_row_ids(_bitmap(3))
+ 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([3], sorted(list(result.results())))
+ self.assertEqual(8.0, result.score_getter()(3))
+
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 +1659,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 +1676,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 +3202,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..1dd9d81fe9 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;
@@ -93,10 +95,22 @@ public class TantivyFullTextGlobalIndexReader implements
GlobalIndexReader {
() -> {
try {
ensureLoaded();
+ RoaringNavigableMap64 includeRowIds =
fullTextSearch.includeRowIds();
+ if (includeRowIds != null && includeRowIds.isEmpty()) {
+ return
Optional.of(ScoredGlobalIndexResult.createEmpty());
+ }
+ int searchLimit =
+ includeRowIds == null
+ ? fullTextSearch.limit()
+ : borrowed.searcher.numDocs();
+ if (searchLimit <= 0) {
+ return
Optional.of(ScoredGlobalIndexResult.createEmpty());
+ }
SearchResult result =
borrowed.searcher.searchJson(
- fullTextSearch.queryJson(),
fullTextSearch.limit());
- return Optional.of(toScoredResult(result));
+ fullTextSearch.queryJson(),
searchLimit);
+ ScoredGlobalIndexResult scored =
toScoredResult(result, includeRowIds);
+ return
Optional.of(scored.topK(fullTextSearch.limit()));
} catch (IOException e) {
throw new RuntimeException("Failed to search Tantivy
full-text index", e);
}
@@ -104,11 +118,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]);
}
diff --git
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
index f259646557..a2f4068297 100644
---
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
+++
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
@@ -345,6 +345,36 @@ public class TantivyFullTextGlobalIndexTest {
}
}
+ @Test
+ public void testIncludeRowIdsSearchesFullShardBeforeTopK() throws
IOException {
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ TantivyFullTextGlobalIndexWriter writer = new
TantivyFullTextGlobalIndexWriter(fileWriter);
+
+ writer.write(BinaryString.fromString("paimon paimon paimon paimon"),
0);
+ writer.write(BinaryString.fromString("paimon paimon"), 1);
+ writer.write(BinaryString.fromString("paimon"), 2);
+
+ List<ResultEntry> results = writer.finish();
+ List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
+ GlobalIndexFileReader fileReader = createFileReader();
+
+ RoaringNavigableMap64 includeRowIds = new RoaringNavigableMap64();
+ includeRowIds.add(2L);
+ FullTextSearch search =
+ new FullTextSearch(FullTextQuery.match("paimon", "text"), 1)
+ .withIncludeRowIds(includeRowIds);
+
+ try (TantivyFullTextGlobalIndexReader reader =
createReader(fileReader, metas)) {
+ Optional<ScoredGlobalIndexResult> searchResult =
+ reader.visitFullTextSearch(search).join();
+ assertThat(searchResult).isPresent();
+
+ RoaringNavigableMap64 rowIds = searchResult.get().results();
+ assertThat(rowIds.getLongCardinality()).isEqualTo(1);
+ assertThat(rowIds.contains(2L)).isTrue();
+ }
+ }
+
@Test
public void testPoolReuse() throws IOException {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
diff --git a/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
b/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
index 7f111efcf8..d6dacbca92 100644
--- a/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
+++ b/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
@@ -1280,6 +1280,17 @@ pub extern "system" fn
Java_org_apache_paimon_tantivy_TantivySearcher_searchInde
search_with_query(&mut env, handle, query, limit)
}
+#[no_mangle]
+pub extern "system" fn Java_org_apache_paimon_tantivy_TantivySearcher_numDocs(
+ _env: JNIEnv,
+ _class: JClass,
+ searcher_ptr: jlong,
+) -> jint {
+ let handle = unsafe { &*(searcher_ptr as *const TantivySearcherHandle) };
+ let num_docs = handle.reader.searcher().num_docs() as u64;
+ std::cmp::min(num_docs, i32::MAX as u64) as jint
+}
+
#[no_mangle]
pub extern "system" fn
Java_org_apache_paimon_tantivy_TantivySearcher_freeSearcher(
_env: JNIEnv,
diff --git
a/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
b/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
index 8962e5b99b..85421f536f 100644
---
a/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
+++
b/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
@@ -126,6 +126,11 @@ public class TantivySearcher implements AutoCloseable {
return searchIndexJson(searcherPtr, queryJson, limit);
}
+ public int numDocs() {
+ checkNotClosed();
+ return numDocs(searcherPtr);
+ }
+
@Override
public void close() {
if (!closed) {
@@ -185,6 +190,8 @@ public class TantivySearcher implements AutoCloseable {
static native SearchResult searchIndexJson(long searcherPtr, String
queryJson, int limit);
+ static native int numDocs(long searcherPtr);
+
static native void freeSearcher(long searcherPtr);
}
diff --git
a/paimon-tantivy/paimon-tantivy-jni/src/test/java/org/apache/paimon/tantivy/TantivyJniTest.java
b/paimon-tantivy/paimon-tantivy-jni/src/test/java/org/apache/paimon/tantivy/TantivyJniTest.java
index 80300c24a8..8e45a133c7 100644
---
a/paimon-tantivy/paimon-tantivy-jni/src/test/java/org/apache/paimon/tantivy/TantivyJniTest.java
+++
b/paimon-tantivy/paimon-tantivy-jni/src/test/java/org/apache/paimon/tantivy/TantivyJniTest.java
@@ -55,6 +55,8 @@ public class TantivyJniTest {
}
try (TantivySearcher searcher = new TantivySearcher(indexPath)) {
+ assertEquals(3, searcher.numDocs());
+
SearchResult result = searcher.search("paimon", 10);
assertTrue(result.size() > 0, "Should find at least one result");
assertEquals(result.getRowIds().length, result.getScores().length);
