From 6ca88a92a9f7fa0f7a040ff7255a9c44e97a7919 Mon Sep 17 00:00:00 2001 From: chaoyang Date: Sun, 20 Sep 2026 14:34:40 +0800 Subject: [PATCH 1/2] [python] Verify candidate-only scalar filters before vector top-k --- 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 +- .../globalindex/global_index_evaluator.py | 5 + .../globalindex/global_index_result.py | 22 +- .../table/source/vector_search_read.py | 39 +++- .../tests/ray_vector_filter_exactness_test.py | 60 ++++++ .../tests/vector_filter_exactness_test.py | 190 ++++++++++++++++++ 9 files changed, 334 insertions(+), 14 deletions(-) create mode 100644 paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py create mode 100644 paimon-python/pypaimon/tests/vector_filter_exactness_test.py diff --git a/docs/docs/pypaimon/multimodal-search.md b/docs/docs/pypaimon/multimodal-search.md index 2431e4877213..2fc052bd7f4d 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 801b73e5310f..5d4cfa34dd40 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 @@ def global_index_external_path(self, default=None): 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 c1d0b6007e73..f553960886ed 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 @@ def visit_not_in(self, literals: List[object]) -> Optional[GlobalIndexResult]: 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 e7acdee35c39..1dbbfb0154b8 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 @@ def add_file(self, index_type, range_key, io_meta): 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 b0c8a0ed2ca7..c33b6d1bae46 100644 --- a/paimon-python/pypaimon/globalindex/global_index_evaluator.py +++ b/paimon-python/pypaimon/globalindex/global_index_evaluator.py @@ -195,6 +195,11 @@ def _combine_results( 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 709e2d43934b..4a63103eb593 100644 --- a/paimon-python/pypaimon/globalindex/global_index_result.py +++ b/paimon-python/pypaimon/globalindex/global_index_result.py @@ -30,6 +30,10 @@ def results(self) -> RoaringBitmap64: """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 @@ def offset(self, start_offset: int) -> 'GlobalIndexResult': 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 @@ def create_empty() -> 'GlobalIndexResult': 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 @@ def from_ranges(ranges: List[Range]) -> 'GlobalIndexResult': 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 f48f1b10f4be..59909966d0f5 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 @@ def _scalar_matched_rows(self, splits, snapshot=None): 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 000000000000..eae220840a1d --- /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 + + +@pytest.mark.parametrize("batch", [False, True]) +@pytest.mark.parametrize("mode", ["full", "fast"]) +@pytest.mark.parametrize("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 000000000000..61b24dfcff85 --- /dev/null +++ b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py @@ -0,0 +1,190 @@ +# 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 + + +@pytest.fixture +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) + + +@pytest.mark.parametrize("pattern", ["%zeta%", "beta%"]) +@pytest.mark.parametrize("batch", [False, True]) +@pytest.mark.parametrize("mode", ["full", "fast"]) +@pytest.mark.parametrize("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) + + +@pytest.mark.parametrize("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}] + + +@pytest.mark.parametrize("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() == [] + + +@pytest.mark.parametrize("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() + + +@pytest.mark.parametrize("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() == [] + + +@pytest.mark.parametrize("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] From 2d535e58eb7af4ae515501093e87624082756573 Mon Sep 17 00:00:00 2001 From: chaoyang Date: Sun, 20 Sep 2026 15:59:18 +0800 Subject: [PATCH 2/2] [python] Preserve exactness for same-predicate index intersections --- .../globalindex/global_index_evaluator.py | 6 +++- .../tests/vector_filter_exactness_test.py | 33 +++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/globalindex/global_index_evaluator.py b/paimon-python/pypaimon/globalindex/global_index_evaluator.py index c33b6d1bae46..fe786e604b8b 100644 --- a/paimon-python/pypaimon/globalindex/global_index_evaluator.py +++ b/paimon-python/pypaimon/globalindex/global_index_evaluator.py @@ -130,7 +130,11 @@ def _combine_reader_results( 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(): diff --git a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py index 61b24dfcff85..b4fa8be79c56 100644 --- a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py +++ b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py @@ -84,6 +84,39 @@ def test_exact_indexes_do_not_read_filter_columns(table, kind, predicate): assert query(table, predicate).to_list() == [{"id": 1}] +@pytest.mark.parametrize("batch", [False, True]) +@pytest.mark.parametrize("mode", ["full", "fast"]) +@pytest.mark.parametrize("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) + + +@pytest.mark.parametrize("method", ["leaf", "and"]) +@pytest.mark.parametrize("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)) + + @pytest.mark.parametrize("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)