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 663e598035 [python] Verify candidate-only scalar filters before vector
top-k (#9999)
663e598035 is described below
commit 663e59803555aaf57cab1b022959e4e73a10a843
Author: chaoyang <[email protected]>
AuthorDate: Sun Sep 20 16:10:51 2026 +0800
[python] Verify candidate-only scalar filters before vector top-k (#9999)
---
docs/docs/pypaimon/multimodal-search.md | 9 +
.../pypaimon/common/options/core_options.py | 12 ++
.../globalindex/btree/btree_index_reader.py | 8 +-
.../data_evolution_global_index_scanner.py | 3 +-
.../pypaimon/globalindex/global_index_evaluator.py | 11 +-
.../pypaimon/globalindex/global_index_result.py | 22 +-
.../pypaimon/table/source/vector_search_read.py | 39 +++-
.../tests/ray_vector_filter_exactness_test.py | 60 ++++++
.../pypaimon/tests/vector_filter_exactness_test.py | 223 +++++++++++++++++++++
9 files changed, 372 insertions(+), 15 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-search.md
b/docs/docs/pypaimon/multimodal-search.md
index 2431e48772..2fc052bd7f 100644
--- a/docs/docs/pypaimon/multimodal-search.md
+++ b/docs/docs/pypaimon/multimodal-search.md
@@ -63,6 +63,15 @@ filter the rows read from the search result. Both
`pre_filter` and `where()`
accept SQL-like predicate strings. For full-text search, `pre_filter` must only
reference partition columns.
+For data-evolution vector search, a scalar index may return candidates rather
than exact
+matches, for example for BTree string-prefix or substring predicates, or when
+part of a conjunction is unsupported. Such index candidates are excluded with
+a warning by default, so the result can contain fewer than the requested rows.
+Set the table option `global-index.filter.refine-from-data=true` to verify
those
+candidates before vector top-k selection. This reads the filter columns at the
+search snapshot and may scan every candidate row; exact index matches need no
+extra read. This applies to single and batch vector queries, locally and on
Ray.
+
Each execution of `search`, `search_vectors`, or `search_hybrid` reads one
snapshot across candidate search, filtering, reranking, and result lookup.
Concurrent commits become visible on the next execution, including when reusing
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 801b73e531..5d4cfa34dd 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -956,6 +956,15 @@ class CoreOptions:
)
)
+ GLOBAL_INDEX_FILTER_REFINE_FROM_DATA: ConfigOption[bool] = (
+ ConfigOptions.key("global-index.filter.refine-from-data")
+ .boolean_type()
+ .default_value(False)
+ .with_description(
+ "Whether vector search may read filter columns to verify
candidate-only scalar index matches. "
+ "When false, inexact index candidates are excluded from the
search.")
+ )
+
GLOBAL_INDEX_THREAD_NUM: ConfigOption[int] = (
ConfigOptions.key("global-index.thread-num")
.int_type()
@@ -1742,6 +1751,9 @@ class CoreOptions:
def global_index_thread_num(self) -> Optional[int]:
return self.options.get(CoreOptions.GLOBAL_INDEX_THREAD_NUM)
+ def global_index_filter_refine_from_data(self) -> bool:
+ return
self.options.get(CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA)
+
def global_index_row_count_per_shard(self) -> int:
return self.options.get(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD)
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
index c1d0b6007e..f553960886 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
@@ -239,16 +239,16 @@ class BTreeIndexReader:
return GlobalIndexResult.create(result)
def visit_starts_with(self, literal: object) ->
Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(self._all_non_null_rows())
+ return GlobalIndexResult.create(self._all_non_null_rows(),
is_exact=False)
def visit_ends_with(self, literal: object) -> Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(self._all_non_null_rows())
+ return GlobalIndexResult.create(self._all_non_null_rows(),
is_exact=False)
def visit_contains(self, literal: object) -> Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(self._all_non_null_rows())
+ return GlobalIndexResult.create(self._all_non_null_rows(),
is_exact=False)
def visit_like(self, literal: object) -> Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(self._all_non_null_rows())
+ return GlobalIndexResult.create(self._all_non_null_rows(),
is_exact=False)
def visit_between(self, min_v: object, max_v: object) ->
Optional[GlobalIndexResult]:
return GlobalIndexResult.create(
diff --git
a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py
b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py
index e7acdee35c..1dbbfb0154 100644
--- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py
+++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py
@@ -282,7 +282,8 @@ class _IndexMetaFileGroup:
class _PaddingGlobalIndexReader(GlobalIndexReader):
def __init__(self, wrapped, padding):
self._wrapped = wrapped
- self._padding = padding
+ # Padding rows have not been tested by this index.
+ self._padding = GlobalIndexResult.create(padding.results(),
is_exact=False)
def _pad(self, future):
return _map_future(
diff --git a/paimon-python/pypaimon/globalindex/global_index_evaluator.py
b/paimon-python/pypaimon/globalindex/global_index_evaluator.py
index b0c8a0ed2c..fe786e604b 100644
--- a/paimon-python/pypaimon/globalindex/global_index_evaluator.py
+++ b/paimon-python/pypaimon/globalindex/global_index_evaluator.py
@@ -130,7 +130,11 @@ class GlobalIndexEvaluator:
if child_result is None:
continue
if compound_result is not None:
- compound_result = compound_result.and_(child_result)
+ # Readers answer the same predicate: an exact result
intersected
+ # with a candidate superset remains exact.
+ is_exact = compound_result.is_exact() or
child_result.is_exact()
+ compound_result = GlobalIndexResult.create(
+ compound_result.and_(child_result).results(),
is_exact=is_exact)
else:
compound_result = child_result
if compound_result.is_empty():
@@ -195,6 +199,11 @@ class GlobalIndexEvaluator:
break
if compound_result is None:
return None
+ if any(child is None for child in results):
+ # A dropped AND child can share a field with a supported child.
+ # Contributing field ids alone therefore cannot prove
exactness.
+ compound_result = GlobalIndexResult.create(
+ compound_result.results(), is_exact=False)
return GlobalIndexEvaluation(compound_result,
frozenset(contributing_field_ids))
diff --git a/paimon-python/pypaimon/globalindex/global_index_result.py
b/paimon-python/pypaimon/globalindex/global_index_result.py
index 709e2d4393..4a63103eb5 100644
--- a/paimon-python/pypaimon/globalindex/global_index_result.py
+++ b/paimon-python/pypaimon/globalindex/global_index_result.py
@@ -30,6 +30,10 @@ class GlobalIndexResult(ABC):
"""Returns the bitmap representing row ids."""
pass
+ def is_exact(self) -> bool:
+ """Whether these row ids are matches rather than a candidate
superset."""
+ return True
+
def offset(self, start_offset: int) -> 'GlobalIndexResult':
"""Returns a new result with row IDs offset by the given amount."""
if start_offset == 0:
@@ -38,18 +42,20 @@ class GlobalIndexResult(ABC):
offset_bitmap = RoaringBitmap64()
for row_id in bitmap:
offset_bitmap.add(row_id + start_offset)
- return SimpleGlobalIndexResult(offset_bitmap)
+ return SimpleGlobalIndexResult(offset_bitmap, self.is_exact())
def and_(self, other: 'GlobalIndexResult') -> 'GlobalIndexResult':
"""Returns the intersection of this result and the other result."""
return SimpleGlobalIndexResult(
- RoaringBitmap64.and_(self.results(), other.results())
+ RoaringBitmap64.and_(self.results(), other.results()),
+ self.is_exact() and other.is_exact(),
)
def or_(self, other: 'GlobalIndexResult') -> 'GlobalIndexResult':
"""Returns the union of this result and the other result."""
return SimpleGlobalIndexResult(
- RoaringBitmap64.or_(self.results(), other.results())
+ RoaringBitmap64.or_(self.results(), other.results()),
+ self.is_exact() and other.is_exact(),
)
def is_empty(self) -> bool:
@@ -62,9 +68,9 @@ class GlobalIndexResult(ABC):
return SimpleGlobalIndexResult(RoaringBitmap64())
@staticmethod
- def create(bitmap: RoaringBitmap64) -> 'GlobalIndexResult':
+ def create(bitmap: RoaringBitmap64, is_exact: bool = True) ->
'GlobalIndexResult':
"""Returns a new GlobalIndexResult wrapping the given bitmap."""
- return SimpleGlobalIndexResult(bitmap)
+ return SimpleGlobalIndexResult(bitmap, is_exact)
@staticmethod
def from_range(range_: Range) -> 'GlobalIndexResult':
@@ -84,8 +90,12 @@ class GlobalIndexResult(ABC):
class SimpleGlobalIndexResult(GlobalIndexResult):
- def __init__(self, result: RoaringBitmap64):
+ def __init__(self, result: RoaringBitmap64, is_exact: bool = True):
self._result = result
+ self._is_exact = is_exact
+
+ def is_exact(self) -> bool:
+ return self._is_exact or self.is_empty()
def results(self) -> RoaringBitmap64:
return self._result
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py
b/paimon-python/pypaimon/table/source/vector_search_read.py
index f48f1b10f4..59909966d0 100644
--- a/paimon-python/pypaimon/table/source/vector_search_read.py
+++ b/paimon-python/pypaimon/table/source/vector_search_read.py
@@ -17,6 +17,7 @@
"""Vector search read to read index files."""
+import logging
from abc import ABC, abstractmethod
from collections import deque
from concurrent.futures import ThreadPoolExecutor
@@ -181,11 +182,43 @@ class AbstractVectorSearchReadImpl:
return RoaringBitmap64()
try:
result = scanner.scan(self._filter)
- if result is None:
- return RoaringBitmap64()
- return result.results()
finally:
scanner.close()
+ if result is not None and result.is_exact():
+ return result.results()
+ if not self._table.options.global_index_filter_refine_from_data():
+ logging.getLogger(__name__).warning(
+ "Scalar index candidates are excluded because the row filter
%s cannot be "
+ "evaluated exactly. Set
global-index.filter.refine-from-data=true to verify "
+ "candidates against the data; otherwise vector search may
return fewer rows.",
+ self._filter)
+ return RoaringBitmap64()
+
+ candidates = RoaringBitmap64()
+ for split in splits:
+ candidates.add_range(split.row_range_start, split.row_range_end)
+ if result is not None:
+ candidates = RoaringBitmap64.and_(candidates, result.results())
+ return self._matching_candidate_rows(candidates, snapshot)
+
+ def _matching_candidate_rows(self, candidates, snapshot):
+ from pypaimon.read.table_read import _ClosableArrowBatchReader
+
+ matched = RoaringBitmap64()
+ if candidates.is_empty():
+ return matched
+ table = global_index_live_row_filter.table_at_snapshot(self._table,
snapshot)
+ builder = (table.new_read_builder().with_filter(self._filter)
+ .with_projection([SpecialFields.ROW_ID.name]))
+ if self._partition_filter is not None:
+ builder = builder.with_partition_filter(self._partition_filter)
+ splits =
builder.new_scan().with_row_ranges(candidates.to_range_list()).plan().splits()
+ reader, batches = builder.new_read()._new_arrow_batch_reader(splits)
+ with _ClosableArrowBatchReader(reader, batches) as batch_reader:
+ for batch in batch_reader:
+ for row_id in
batch.column(SpecialFields.ROW_ID.name).to_pylist():
+ matched.add(row_id)
+ return matched
def _pre_filter(self, splits, snapshot=None):
# Backwards-compatible helper used by older tests/callers.
diff --git a/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
b/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
new file mode 100644
index 0000000000..eae220840a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
@@ -0,0 +1,60 @@
+# 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.
+
+from unittest.mock import patch
+
+import pytest
+
+pytest.importorskip("ray")
+
+from pypaimon.read.table_read import TableRead
+from pypaimon.table.source.vector_search_read import
AbstractVectorSearchReadImpl
+from pypaimon.tests import ray_vector_search_test as ray_fixtures
+from pypaimon.tests import vector_filter_exactness_test as fixtures
+from pypaimon.tests.vector_filter_exactness_test import query, scalar_index
+
+ray_cluster = ray_fixtures.ray_cluster
+table = fixtures.table
+
+
[email protected]("batch", [False, True])
[email protected]("mode", ["full", "fast"])
[email protected]("refine", [False, True])
+def test_ray_applies_exact_row_filter_before_worker_top_k(table, ray_cluster,
batch, mode, refine):
+ scalar_index(table)
+ table.raw_table = table.raw_table.copy({
+ "vector-index.search-mode": mode,
"global-index.filter.refine-from-data": str(refine).lower()})
+ original = AbstractVectorSearchReadImpl._matching_candidate_rows
+ arrow_read = TableRead._new_arrow_batch_reader
+ calls = []
+
+ def no_vectors(read, *args, **kwargs):
+ assert "embedding" not in [field.name for field in read.read_type]
+ return arrow_read(read, *args, **kwargs)
+
+ def verify(reader, candidates, snapshot):
+ calls.append(list(candidates))
+ assert snapshot is not None
+ with patch.object(TableRead, "_new_arrow_batch_reader", no_vectors):
+ return original(reader, candidates, snapshot)
+
+ with patch.object(AbstractVectorSearchReadImpl,
"_matching_candidate_rows", verify):
+ result = query(table, "name LIKE '%zeta%'",
batch).to_arrow(execution="ray", concurrency=2)
+ actual = [value.to_pylist() for value in result] if batch else
result.to_pylist()
+ expected = [{"id": 1}] if refine else []
+ assert actual == ([expected, expected] if batch else expected)
+ assert calls == ([[0, 1, 2]] if refine else [])
diff --git a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
new file mode 100644
index 0000000000..b4fa8be79c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
@@ -0,0 +1,223 @@
+# 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.
+
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+import pypaimon.multimodal as pm
+from pypaimon.common.predicate import Predicate
+from pypaimon.globalindex.data_evolution_global_index_scanner import
_PaddingGlobalIndexReader
+from pypaimon.globalindex.global_index_evaluator import GlobalIndexEvaluator
+from pypaimon.globalindex.global_index_reader import FieldRef,
_completed_future
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.offset_global_index_reader import
OffsetGlobalIndexReader
+from pypaimon.globalindex.union_global_index_reader import
UnionGlobalIndexReader
+from pypaimon.table.source.vector_search_read import
AbstractVectorSearchReadImpl, DataEvolutionVectorRead
+from pypaimon.tests.vector_search_filter_test import _StubTable, _field
+from pypaimon.tests.global_index_evaluator_test import StubGlobalIndexReader,
_make_fields
+from pypaimon.utils.range import Range
+
+
[email protected]
+def table(tmp_path):
+ pytest.importorskip("paimon_vindex")
+ schema = pa.schema([("id", pa.int64()), ("name", pa.string()),
("embedding", pa.list_(pa.float32(), 2))])
+ table = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+ "vectors", schema=schema, options={"file.format": "parquet",
"vector.file.format": "parquet",
+ "read.batch-size": "1"})
+ table.add(pa.table({"id": [0, 1, 2], "name": ["alpha", "beta zeta",
"gamma"],
+ "embedding": [[0., 1.], [1., 1.], [2., 1.]]},
schema=schema))
+ table.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index(
+ "embedding", "ivf-flat", options={"ivf-flat.nlist": "1",
"ivf-flat.distance.metric": "l2"})
+ return table
+
+
+def scalar_index(table, kind="btree"):
+ table.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("name", kind)
+
+
+def query(table, predicate, batch=False):
+ search = table.search_vectors([[0., 1.], [0., 1.]], pre_filter=predicate)
if batch else table.search(
+ [0., 1.], pre_filter=predicate)
+ return search.select(["id"]).limit(1)
+
+
[email protected]("pattern", ["%zeta%", "beta%"])
[email protected]("batch", [False, True])
[email protected]("mode", ["full", "fast"])
[email protected]("refine", [False, True])
+def test_btree_candidates_are_verified_before_top_k(table, caplog, pattern,
batch, mode, refine):
+ scalar_index(table)
+ table.raw_table = table.raw_table.copy({
+ "vector-index.search-mode": mode,
"global-index.filter.refine-from-data": str(refine).lower()})
+ result = query(table, "name LIKE '%s'" % pattern, batch).to_list()
+ expected = [{"id": 1}] if refine else []
+ assert result == ([expected, expected] if batch else expected)
+ assert ("global-index.filter.refine-from-data=true" in caplog.text) ==
(not refine)
+
+
[email protected]("kind, predicate", [
+ ("btree", "name = 'beta zeta'"), ("btree", "name >= 'beta' AND name <
'gamma'"),
+ ("btree", "name LIKE 'beta zeta'"), ("bitmap", "name LIKE '%zeta%'"),
+])
+def test_exact_indexes_do_not_read_filter_columns(table, kind, predicate):
+ scalar_index(table, kind)
+ table.raw_table =
table.raw_table.copy({"global-index.filter.refine-from-data": "true"})
+ with patch.object(AbstractVectorSearchReadImpl, "_matching_candidate_rows",
+ side_effect=AssertionError("exact index recheck")):
+ assert query(table, predicate).to_list() == [{"id": 1}]
+
+
[email protected]("batch", [False, True])
[email protected]("mode", ["full", "fast"])
[email protected]("refine", [False, True])
+def test_mixed_btree_and_bitmap_preserve_exact_matches(table, batch, mode,
refine):
+ scalar_index(table, "btree")
+ scalar_index(table, "bitmap")
+ table.raw_table = table.raw_table.copy({
+ "vector-index.search-mode": mode,
"global-index.filter.refine-from-data": str(refine).lower()})
+ with patch.object(AbstractVectorSearchReadImpl, "_matching_candidate_rows",
+ side_effect=AssertionError("exact index recheck")):
+ result = query(table, "name LIKE '%zeta%'", batch).to_list()
+ expected = [{"id": 1}]
+ assert result == ([expected, expected] if batch else expected)
+
+
[email protected]("method", ["leaf", "and"])
[email protected]("first_exact, second_exact", [(False, False), (False,
True), (True, False), (True, True)])
+def test_reader_intersection_and_predicate_conjunction_exactness(method,
first_exact, second_exact):
+ results = [GlobalIndexResult.create(
+ GlobalIndexResult.from_range(Range(1, 2) if exact else Range(0,
3)).results(), is_exact=exact)
+ for exact in (first_exact, second_exact)]
+ readers = [StubGlobalIndexReader(result) for result in results]
+ leaves = [Predicate(method="equal", index=i, field=field, literals=[1])
for i, field in enumerate(("a", "b"))]
+ predicate = leaves[0] if method == "leaf" else Predicate(
+ method="and", index=None, field=None, literals=leaves)
+ with GlobalIndexEvaluator(_make_fields(), lambda field: readers +
[StubGlobalIndexReader(None)]
+ if method == "leaf" else [readers[field.id]]) as
evaluator:
+ result = evaluator.evaluate(predicate)
+ assert list(result.results()) == ([1, 2] if first_exact or second_exact
else [0, 1, 2, 3])
+ assert result.is_exact() == ((first_exact or second_exact) if method ==
"leaf" else (
+ first_exact and second_exact))
+
+
[email protected]("predicate", ["name LIKE '%zeta%'", "name >= 'a' AND
name LIKE '%zeta%'"])
+def test_unsupported_leaf_and_same_field_conjunction_can_be_refined(table,
predicate):
+ scalar_index(table)
+ table.raw_table =
table.raw_table.copy({"global-index.filter.refine-from-data": "true",
+
"btree-index.fallback-scan-max-size": "0 b"})
+ assert query(table, predicate).to_list() == [{"id": 1}]
+
+
+def test_commit_before_candidate_verification_keeps_snapshot(table):
+ scalar_index(table)
+ table.raw_table =
table.raw_table.copy({"global-index.filter.refine-from-data": "true",
+
"global-index.column-update-action": "DROP_PARTITION_INDEX"})
+ search = query(table, "name LIKE '%zeta%'")
+ original = AbstractVectorSearchReadImpl._matching_candidate_rows
+
+ def update_before_read(reader, candidates, snapshot):
+ table.update("id = 1", {"name": "changed"})
+ return original(reader, candidates, snapshot)
+
+ with patch.object(AbstractVectorSearchReadImpl,
"_matching_candidate_rows", update_before_read):
+ assert search.to_list() == [{"id": 1}]
+ assert search.to_list() == []
+
+
[email protected]("method", ["and", "or"])
+def test_exactness_follows_predicate_tree_and_reader_wrappers(method):
+ candidate = GlobalIndexResult.create(GlobalIndexResult.from_range(Range(0,
3)).results(), is_exact=False)
+ exact = GlobalIndexResult.from_range(Range(1, 2))
+ assert not candidate.offset(10).is_exact()
+ assert not candidate.and_(exact).is_exact()
+ assert not candidate.or_(exact).is_exact()
+ assert candidate.and_(GlobalIndexResult.create_empty()).is_exact()
+ reader =
UnionGlobalIndexReader([OffsetGlobalIndexReader(StubGlobalIndexReader(candidate),
10, 13)])
+ assert not reader.visit_equal(FieldRef(0, "a", "INT"),
1).result().is_exact()
+ padded = _PaddingGlobalIndexReader(StubGlobalIndexReader(exact),
GlobalIndexResult.from_range(Range(5, 6)))
+ assert not padded.visit_equal(FieldRef(0, "a", "INT"),
1).result().is_exact()
+
+ class PartialReader(StubGlobalIndexReader):
+ def visit_greater_than(self, field_ref, literal):
+ return _completed_future(None)
+
+ evaluator = GlobalIndexEvaluator(_make_fields(), lambda field:
[PartialReader(exact)])
+ predicate = Predicate(method=method, index=None, field=None, literals=[
+ Predicate(method="equal", index=0, field="a", literals=[1]),
+ Predicate(method="greaterThan", index=0, field="a", literals=[0]),
+ ])
+ try:
+ result = evaluator.evaluate(predicate)
+ if method == "or":
+ assert result is None
+ else:
+ assert list(result.results()) == [1, 2]
+ assert not result.is_exact()
+ finally:
+ evaluator.close()
+
+
[email protected]("selector", ["snapshot", "tag"])
+def test_candidate_verification_preserves_historical_deletions(table,
selector):
+ scalar_index(table)
+ table.raw_table =
table.raw_table.copy({"global-index.filter.refine-from-data": "true"})
+ saved = table.raw_table.snapshot_manager().get_latest_snapshot()
+ table.raw_table.create_tag("saved", snapshot_id=saved.id)
+ table.delete("id = 1")
+ options = {"snapshot_id": saved.id} if selector == "snapshot" else
{"tag_name": "saved"}
+ search = table.search([0., 1.], pre_filter="name LIKE '%zeta%'",
**options).select(["id"]).limit(1)
+ assert search.to_list() == [{"id": 1}]
+ assert query(table, "name LIKE '%zeta%'").to_list() == []
+
+
[email protected]("fail_read", [False, True])
+def
test_candidate_verification_projects_only_row_ids_and_closes_stream(fail_read):
+ field = _field(1, "embedding", "FLOAT")
+ table = _StubTable([field], [])
+ read = DataEvolutionVectorRead(table, 1, field, [0.], filter_="filter")
+ builder = Mock()
+ builder.with_filter.return_value = builder
+ builder.with_projection.return_value = builder
+ table.new_read_builder = Mock(return_value=builder)
+ schema = pa.schema([("_ROW_ID", pa.int64())])
+ closed = []
+
+ def batches():
+ try:
+ yield pa.record_batch([[1]], schema=schema)
+ if fail_read:
+ raise ValueError("filter read failed")
+ finally:
+ closed.append(True)
+
+ source = batches()
+ resource = Mock()
+ resource.read_next_batch.side_effect = lambda: next(source)
+ builder.new_read.return_value._new_arrow_batch_reader.return_value =
resource, source
+ candidates = GlobalIndexResult.from_range(Range(0, 2)).results()
+ if fail_read:
+ with pytest.raises(ValueError, match="filter read failed"):
+ read._matching_candidate_rows(candidates, None)
+ else:
+ assert list(read._matching_candidate_rows(candidates, None)) == [1]
+ builder.with_filter.assert_called_once_with("filter")
+ builder.with_projection.assert_called_once_with(["_ROW_ID"])
+ resource.close.assert_called_once_with()
+ assert closed == [True]
