From ce3071e6fb87a4f9b328466ef36ab2a848c04650 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 04:53:17 -0700 Subject: [PATCH 01/10] [python] Default scalar global-index search-mode to full to stop silent row loss Scalar (sorted/bitmap) global-index queries used the vector/full-text `global-index.search-mode` = fast, which returns no unindexed ranges, so rows committed but not yet covered by the index were pruned before being read and `WHERE = X` silently returned incomplete results. - Add `scalar-index.search-mode` (default `full`). Resolution priority: explicit `scalar-index.search-mode` -> an explicit `global-index.search-mode` (inherited) -> default `full`. - Thread an optional `search_mode` through DataEvolutionGlobalIndexCoverage and DataEvolutionGlobalIndexScanner.unindexed_rows. - Only the pure scalar scan (file_scanner) passes the scalar mode. Vector / full-text callers leave it unset and keep the fast default, so a vector scan with a scalar filter is not widened into a full-table brute-force scan. --- .../pypaimon/common/options/core_options.py | 21 +++- .../data_evolution_global_index_coverage.py | 12 ++- .../data_evolution_global_index_scanner.py | 6 +- .../pypaimon/read/scanner/file_scanner.py | 4 +- .../global_index_scalar_search_mode_test.py | 100 ++++++++++++++++++ 5 files changed, 136 insertions(+), 7 deletions(-) create mode 100644 paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 20ee9ddb8c25..5222a02c40f4 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -733,7 +733,19 @@ class CoreOptions: .enum_type(GlobalIndexSearchMode) .default_value(GlobalIndexSearchMode.FAST) .with_description( - "Search mode for global index queries. " + "Search mode for vector and full-text global index queries. " + "Scalar index queries use 'scalar-index.search-mode'. " + "Supported values are 'fast', 'full', and 'detail'." + ) + ) + + GLOBAL_INDEX_SCALAR_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + ConfigOptions.key("scalar-index.search-mode") + .enum_type(GlobalIndexSearchMode) + .default_value(GlobalIndexSearchMode.FULL) + .with_description( + "Search mode for scalar (sorted/bitmap) global index queries. " + "Defaults to 'full' to include rows not covered by the index. " "Supported values are 'fast', 'full', and 'detail'." ) ) @@ -1372,6 +1384,13 @@ def global_index_enabled(self, default=None): def global_index_search_mode(self): return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) + def global_index_scalar_search_mode(self): + if self.options.contains(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE): + return self.options.get(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE) + if self.options.contains(CoreOptions.GLOBAL_INDEX_SEARCH_MODE): + return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) + return GlobalIndexSearchMode.FULL + def global_index_external_path(self, default=None): value = self.options.get(CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH, default) if value is None: diff --git a/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py b/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py index 07a9213bb830..878a87a5680a 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py @@ -55,6 +55,7 @@ def unindexed_ranges( self, fields_or_field_id: Union[List[DataField], Collection[int], int], predicate: Optional[Predicate] = None, + search_mode: Optional[GlobalIndexSearchMode] = None, ) -> List[Range]: if isinstance(fields_or_field_id, int): field_ids = {fields_or_field_id} @@ -67,7 +68,7 @@ def unindexed_ranges( field = field_by_name.get(name) if field is not None: field_ids.add(field.id) - return self._unindexed_ranges(field_ids) + return self._unindexed_ranges(field_ids, search_mode) def _add_coverage(self, field_id: int, row_range: Range) -> None: self._coverage_by_field.setdefault(field_id, []).append(row_range) @@ -84,8 +85,13 @@ def _indexed_ranges(self, field_ids: Collection[int]) -> List[Range]: return [] return Range.sort_and_merge_overlap(ranges, True) - def _unindexed_ranges(self, field_ids: Collection[int]) -> List[Range]: - search_mode = _global_index_search_mode(self._table) + def _unindexed_ranges( + self, + field_ids: Collection[int], + search_mode: Optional[GlobalIndexSearchMode] = None, + ) -> List[Range]: + if search_mode is None: + search_mode = _global_index_search_mode(self._table) if search_mode == GlobalIndexSearchMode.FAST: return [] next_row_id = getattr(self._snapshot, "next_row_id", None) 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 4ee6b94ced30..192c2e99bb8d 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -209,12 +209,14 @@ def scan(self, predicate: Optional[Predicate]) -> Optional[GlobalIndexResult]: """Scan the global index with the given predicate.""" return self._evaluator.evaluate(predicate) - def unindexed_rows(self, predicate: Optional[Predicate]) -> GlobalIndexResult: + def unindexed_rows(self, predicate: Optional[Predicate], + search_mode=None) -> GlobalIndexResult: """Return coarse row ids not covered by global indexes.""" if self._coverage is None: return GlobalIndexResult.create_empty() return GlobalIndexResult.from_ranges( - self._coverage.unindexed_ranges(self._fields, predicate)) + self._coverage.unindexed_ranges( + self._fields, predicate, search_mode=search_mode)) def close(self): """Close the scanner and release resources.""" diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index e952a9f3a4c8..ac70b5e00b68 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -498,7 +498,9 @@ def _eval_global_index(self, snapshot=None): result = scanner.scan(self.predicate) if result is None: return None - return result.or_(scanner.unindexed_rows(self.predicate)) + scalar_mode = self.table.options.global_index_scalar_search_mode() + return result.or_( + scanner.unindexed_rows(self.predicate, search_mode=scalar_mode)) except Exception: return None diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py new file mode 100644 index 000000000000..3071afee83d1 --- /dev/null +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -0,0 +1,100 @@ +# 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. + +import unittest +from types import SimpleNamespace + +from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode +from pypaimon.common.options.options import Options +from pypaimon.globalindex.data_evolution_global_index_coverage import ( + DataEvolutionGlobalIndexCoverage, +) +from pypaimon.globalindex.data_evolution_global_index_scanner import ( + DataEvolutionGlobalIndexScanner, +) + + +def _ranges(result): + inner = result.results() if hasattr(result, "results") else result + for attr in ("to_range_list", "to_ranges", "ranges"): + if hasattr(inner, attr): + value = getattr(inner, attr) + value = value() if callable(value) else value + return [(r.from_, r.to) for r in value] + raise AssertionError("cannot extract ranges from %r" % (result,)) + + +def _coverage(options): + meta = SimpleNamespace( + row_range_start=0, row_range_end=99, index_field_id=1, extra_field_ids=None) + snapshot = SimpleNamespace(next_row_id=200) + table = SimpleNamespace(options=options) + return DataEvolutionGlobalIndexCoverage( + table, snapshot, None, [SimpleNamespace(global_index_meta=meta)]) + + +class ScalarGlobalIndexSearchModeTest(unittest.TestCase): + + def test_default_values(self): + options = CoreOptions(Options.from_none()) + self.assertEqual( + GlobalIndexSearchMode.FULL, options.global_index_scalar_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.global_index_search_mode()) + + def test_scalar_option_override(self): + options = CoreOptions(Options({"scalar-index.search-mode": "detail"})) + self.assertEqual( + GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) + + def test_inherits_explicit_global_index_search_mode(self): + options = CoreOptions(Options({"global-index.search-mode": "detail"})) + self.assertEqual( + GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) + + def test_scalar_key_wins_over_global_index_key(self): + options = CoreOptions(Options({ + "scalar-index.search-mode": "detail", + "global-index.search-mode": "fast", + })) + self.assertEqual( + GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) + + def test_coverage_honours_search_mode_override(self): + coverage = _coverage(CoreOptions(Options.from_none())) + self.assertEqual( + [], coverage.unindexed_ranges(1, search_mode=GlobalIndexSearchMode.FAST)) + full = coverage.unindexed_ranges(1, search_mode=GlobalIndexSearchMode.FULL) + self.assertEqual([(100, 199)], [(r.from_, r.to) for r in full]) + self.assertEqual([], coverage.unindexed_ranges(1)) + + def test_scanner_applies_passed_scalar_mode(self): + coverage = _coverage(CoreOptions(Options.from_none())) + scanner = SimpleNamespace(_coverage=coverage, _fields=[1]) + result = DataEvolutionGlobalIndexScanner.unindexed_rows( + scanner, None, search_mode=GlobalIndexSearchMode.FULL) + self.assertEqual([(100, 199)], _ranges(result)) + + def test_scanner_default_is_general_mode(self): + coverage = _coverage(CoreOptions(Options.from_none())) + scanner = SimpleNamespace(_coverage=coverage, _fields=[1]) + result = DataEvolutionGlobalIndexScanner.unindexed_rows(scanner, None) + self.assertEqual([], _ranges(result)) + + +if __name__ == "__main__": + unittest.main() From 419590eb2e624555855b9f3a95b3d298a67c778e Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 09:31:34 +0800 Subject: [PATCH 02/10] [python] Fix scalar search mode fallback compatibility --- paimon-python/pypaimon/read/scanner/file_scanner.py | 9 ++++++++- paimon-python/pypaimon/tests/global_index_test.py | 2 ++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index ac70b5e00b68..f37b54e4ce5c 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -22,6 +22,7 @@ logger = logging.getLogger(__name__) +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.common.predicate import Predicate from pypaimon.globalindex import ScoredGlobalIndexResult from pypaimon.manifest.index_manifest_file import IndexManifestFile @@ -498,7 +499,13 @@ def _eval_global_index(self, snapshot=None): result = scanner.scan(self.predicate) if result is None: return None - scalar_mode = self.table.options.global_index_scalar_search_mode() + scalar_mode_getter = getattr( + self.table.options, "global_index_scalar_search_mode", None) + scalar_mode = ( + scalar_mode_getter() + if scalar_mode_getter is not None + else GlobalIndexSearchMode.FULL + ) return result.or_( scanner.unindexed_rows(self.predicate, search_mode=scalar_mode)) except Exception: diff --git a/paimon-python/pypaimon/tests/global_index_test.py b/paimon-python/pypaimon/tests/global_index_test.py index 4d7a1b142bc2..2c96e4e40fbb 100644 --- a/paimon-python/pypaimon/tests/global_index_test.py +++ b/paimon-python/pypaimon/tests/global_index_test.py @@ -249,6 +249,8 @@ class _Table: [Range(1, 1), Range(5, 6)], result.results().to_range_list(), ) + fake_scanner.unindexed_rows.assert_called_once_with( + predicate, search_mode=GlobalIndexSearchMode.FULL) def test_eval_global_index_keeps_none_as_full_scan(self): from pypaimon.read.scanner.file_scanner import FileScanner From 3b229d53a07c68d12f3ee7a122926865f786d26b Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 10:05:29 +0800 Subject: [PATCH 03/10] [python] Force full scalar scans for predicate mutations --- .../pypaimon/common/options/core_options.py | 5 ++++- .../tests/e2e/java_py_read_write_test.py | 6 +++--- .../global_index_scalar_search_mode_test.py | 5 +++++ .../pypaimon/tests/table_update_test.py | 19 +++++++++++++++++-- paimon-python/pypaimon/write/table_update.py | 2 ++ 5 files changed, 31 insertions(+), 6 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 5222a02c40f4..64121de5fe6d 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -734,7 +734,8 @@ class CoreOptions: .default_value(GlobalIndexSearchMode.FAST) .with_description( "Search mode for vector and full-text global index queries. " - "Scalar index queries use 'scalar-index.search-mode'. " + "An explicitly configured value is also used by scalar index " + "queries when 'scalar-index.search-mode' is not set. " "Supported values are 'fast', 'full', and 'detail'." ) ) @@ -746,6 +747,8 @@ class CoreOptions: .with_description( "Search mode for scalar (sorted/bitmap) global index queries. " "Defaults to 'full' to include rows not covered by the index. " + "An explicitly configured 'global-index.search-mode' is used " + "when this option is not set. " "Supported values are 'fast', 'full', and 'detail'." ) ) diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py index bc48ef3ccc4c..80a3e7934c8d 100644 --- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py +++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py @@ -463,15 +463,15 @@ def test_read_btree_index_table(self): def test_read_btree_raw_fallback(self): table = self.catalog.get_table('default.test_btree_raw_fallback') - fast_builder = table.new_read_builder() + fast_table = table.copy({'scalar-index.search-mode': 'fast'}) + fast_builder = fast_table.new_read_builder() fast_predicate = fast_builder.new_predicate_builder().equal('k', 'k4') fast_builder.with_filter(fast_predicate) fast_result = fast_builder.new_read().to_arrow( fast_builder.new_scan().plan().splits()) self.assertEqual(0, fast_result.num_rows) - full_table = table.copy({'global-index.search-mode': 'full'}) - read_builder = full_table.new_read_builder() + read_builder = table.new_read_builder() read_builder.with_filter( read_builder.new_predicate_builder().equal('k', 'k4')) actual = read_builder.new_read().to_arrow( diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py index 3071afee83d1..8979f0a751db 100644 --- a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -66,6 +66,11 @@ def test_inherits_explicit_global_index_search_mode(self): self.assertEqual( GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) + def test_inherits_explicit_global_fast_mode(self): + options = CoreOptions(Options({"global-index.search-mode": "fast"})) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.global_index_scalar_search_mode()) + def test_scalar_key_wins_over_global_index_key(self): options = CoreOptions(Options({ "scalar-index.search-mode": "detail", diff --git a/paimon-python/pypaimon/tests/table_update_test.py b/paimon-python/pypaimon/tests/table_update_test.py index afbb18b469db..40ceaf2489f6 100644 --- a/paimon-python/pypaimon/tests/table_update_test.py +++ b/paimon-python/pypaimon/tests/table_update_test.py @@ -94,13 +94,14 @@ def _create_seeded_table(self, partition_keys=None, options=None): }, schema=self.pa_schema)) return table - def _create_global_indexed_table_for_predicate_update(self): + def _create_global_indexed_table_for_predicate_update(self, extra_options=None): options = dict(self.table_options) options.update({ 'global-index.enabled': 'true', 'bucket': '-1', 'file.format': 'parquet', }) + options.update(extra_options or {}) table = self._create_table(options=options) self._write_arrow(table, pa.Table.from_pydict({ 'id': [1, 2], @@ -329,7 +330,9 @@ def test_update_by_predicate_rejects_uncastable_assignment(self): ) def test_update_by_predicate_with_global_index_updates_unindexed_rows(self): - table = self._create_global_indexed_table_for_predicate_update() + table = self._create_global_indexed_table_for_predicate_update({ + 'scalar-index.search-mode': 'fast', + }) pb = table.new_read_builder().new_predicate_builder() self._do_update_by_predicate( @@ -345,6 +348,18 @@ def test_update_by_predicate_with_global_index_updates_unindexed_rows(self): )) self.assertEqual({1: 10, 2: 15, 3: 21, 4: 30}, ages_by_id) + def test_delete_by_predicate_with_global_index_deletes_unindexed_rows(self): + table = self._create_global_indexed_table_for_predicate_update({ + 'deletion-vectors.enabled': 'true', + 'scalar-index.search-mode': 'fast', + }) + + pb = table.new_read_builder().new_predicate_builder() + self._do_delete_by_predicate(table, pb.equal('name', 'new')) + + result = self._read_all(table).sort_by('id') + self.assertEqual([1, 2, 4], result['id'].to_pylist()) + def test_update_by_predicate_with_global_index_falls_back_to_full_scan(self): table = self._create_global_indexed_table_for_predicate_update() diff --git a/paimon-python/pypaimon/write/table_update.py b/paimon-python/pypaimon/write/table_update.py index a08418591efd..90b784224526 100644 --- a/paimon-python/pypaimon/write/table_update.py +++ b/paimon-python/pypaimon/write/table_update.py @@ -275,6 +275,8 @@ def _matched_update_scan_table(self): dynamic_options = { CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(): GlobalIndexSearchMode.FULL.value, + CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE.key(): + GlobalIndexSearchMode.FULL.value, CoreOptions.SCAN_MODE.key(): StartupMode.DEFAULT.value, CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id), } From 5d2058185112afccad381cc4d7d37e874f1b255c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 10:27:53 +0800 Subject: [PATCH 04/10] [python] Isolate scalar index search mode --- .../pypaimon/common/options/core_options.py | 11 ++--------- .../pypaimon/read/scanner/file_scanner.py | 9 +-------- .../global_index_scalar_search_mode_test.py | 19 ++++++++----------- .../pypaimon/tests/global_index_test.py | 3 +++ 4 files changed, 14 insertions(+), 28 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 64121de5fe6d..818c1f0a9ac4 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -734,8 +734,7 @@ class CoreOptions: .default_value(GlobalIndexSearchMode.FAST) .with_description( "Search mode for vector and full-text global index queries. " - "An explicitly configured value is also used by scalar index " - "queries when 'scalar-index.search-mode' is not set. " + "Scalar index queries use 'scalar-index.search-mode'. " "Supported values are 'fast', 'full', and 'detail'." ) ) @@ -747,8 +746,6 @@ class CoreOptions: .with_description( "Search mode for scalar (sorted/bitmap) global index queries. " "Defaults to 'full' to include rows not covered by the index. " - "An explicitly configured 'global-index.search-mode' is used " - "when this option is not set. " "Supported values are 'fast', 'full', and 'detail'." ) ) @@ -1388,11 +1385,7 @@ def global_index_search_mode(self): return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) def global_index_scalar_search_mode(self): - if self.options.contains(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE): - return self.options.get(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE) - if self.options.contains(CoreOptions.GLOBAL_INDEX_SEARCH_MODE): - return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) - return GlobalIndexSearchMode.FULL + return self.options.get(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE) def global_index_external_path(self, default=None): value = self.options.get(CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH, default) diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index f37b54e4ce5c..ac70b5e00b68 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -22,7 +22,6 @@ logger = logging.getLogger(__name__) -from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.common.predicate import Predicate from pypaimon.globalindex import ScoredGlobalIndexResult from pypaimon.manifest.index_manifest_file import IndexManifestFile @@ -499,13 +498,7 @@ def _eval_global_index(self, snapshot=None): result = scanner.scan(self.predicate) if result is None: return None - scalar_mode_getter = getattr( - self.table.options, "global_index_scalar_search_mode", None) - scalar_mode = ( - scalar_mode_getter() - if scalar_mode_getter is not None - else GlobalIndexSearchMode.FULL - ) + scalar_mode = self.table.options.global_index_scalar_search_mode() return result.or_( scanner.unindexed_rows(self.predicate, search_mode=scalar_mode)) except Exception: diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py index 8979f0a751db..886c9c2f5a5f 100644 --- a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -57,20 +57,17 @@ def test_default_values(self): GlobalIndexSearchMode.FAST, options.global_index_search_mode()) def test_scalar_option_override(self): - options = CoreOptions(Options({"scalar-index.search-mode": "detail"})) - self.assertEqual( - GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) - - def test_inherits_explicit_global_index_search_mode(self): - options = CoreOptions(Options({"global-index.search-mode": "detail"})) - self.assertEqual( - GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) - - def test_inherits_explicit_global_fast_mode(self): - options = CoreOptions(Options({"global-index.search-mode": "fast"})) + options = CoreOptions(Options({"scalar-index.search-mode": "fast"})) self.assertEqual( GlobalIndexSearchMode.FAST, options.global_index_scalar_search_mode()) + def test_does_not_inherit_global_index_search_mode(self): + for mode in ("fast", "detail"): + options = CoreOptions(Options({"global-index.search-mode": mode})) + self.assertEqual( + GlobalIndexSearchMode.FULL, + options.global_index_scalar_search_mode()) + def test_scalar_key_wins_over_global_index_key(self): options = CoreOptions(Options({ "scalar-index.search-mode": "detail", diff --git a/paimon-python/pypaimon/tests/global_index_test.py b/paimon-python/pypaimon/tests/global_index_test.py index 2c96e4e40fbb..22d57f6d4118 100644 --- a/paimon-python/pypaimon/tests/global_index_test.py +++ b/paimon-python/pypaimon/tests/global_index_test.py @@ -223,6 +223,9 @@ class _Options: def global_index_enabled(self): return True + def global_index_scalar_search_mode(self): + return GlobalIndexSearchMode.FULL + class _Table: options = _Options() From 0d17bdf7d14c6503861c3a4d89134f1b1b0c77c8 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 10:31:51 +0800 Subject: [PATCH 05/10] [python] Fall back when scalar search mode is unset --- paimon-python/pypaimon/common/options/core_options.py | 5 ++++- .../pypaimon/tests/global_index_scalar_search_mode_test.py | 7 +++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 818c1f0a9ac4..ec56a6d67b1d 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1385,7 +1385,10 @@ def global_index_search_mode(self): return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) def global_index_scalar_search_mode(self): - return self.options.get(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE) + scalar_mode = self.options.get(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE) + if scalar_mode is not None: + return scalar_mode + return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) def global_index_external_path(self, default=None): value = self.options.get(CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH, default) diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py index 886c9c2f5a5f..b58804d544fb 100644 --- a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -68,6 +68,13 @@ def test_does_not_inherit_global_index_search_mode(self): GlobalIndexSearchMode.FULL, options.global_index_scalar_search_mode()) + def test_falls_back_when_scalar_mode_has_no_value(self): + options = unittest.mock.Mock() + options.get.side_effect = [None, GlobalIndexSearchMode.DETAIL] + self.assertEqual( + GlobalIndexSearchMode.DETAIL, + CoreOptions(options).global_index_scalar_search_mode()) + def test_scalar_key_wins_over_global_index_key(self): options = CoreOptions(Options({ "scalar-index.search-mode": "detail", From d2367366feb60a58a6ef83fddcd068e0f4ffb2f1 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 10:41:30 +0800 Subject: [PATCH 06/10] [python] Fix standalone scalar search mode test --- .../pypaimon/tests/global_index_scalar_search_mode_test.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py index b58804d544fb..6db1b820ee75 100644 --- a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -17,6 +17,7 @@ import unittest from types import SimpleNamespace +from unittest import mock from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode from pypaimon.common.options.options import Options @@ -69,7 +70,7 @@ def test_does_not_inherit_global_index_search_mode(self): options.global_index_scalar_search_mode()) def test_falls_back_when_scalar_mode_has_no_value(self): - options = unittest.mock.Mock() + options = mock.Mock() options.get.side_effect = [None, GlobalIndexSearchMode.DETAIL] self.assertEqual( GlobalIndexSearchMode.DETAIL, From 54fd9599b48c81fe2c6975a874db3c127ba871f2 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 11:46:41 +0800 Subject: [PATCH 07/10] [core][python] Add index-specific search modes --- docs/docs/multimodal-table/global-index.mdx | 20 ++- .../global-index/full-text.mdx | 4 +- docs/docs/primary-key-table/global-index.mdx | 14 +- docs/docs/pypaimon/data-evolution.md | 2 +- docs/generated/core_configuration.html | 22 ++- .../java/org/apache/paimon/CoreOptions.java | 44 +++++- .../globalindex/DataEvolutionBatchScan.java | 2 +- .../DataEvolutionGlobalIndexCoverage.java | 17 ++- .../DataEvolutionGlobalIndexScanner.java | 7 +- .../source/DataEvolutionFullTextScan.java | 6 +- .../table/source/DataEvolutionVectorScan.java | 9 +- .../table/source/PrimaryKeyFullTextRead.java | 4 +- .../table/source/PrimaryKeyVectorRead.java | 2 +- .../org/apache/paimon/CoreOptionsTest.java | 33 ++++ .../table/BtreeGlobalIndexTableTest.java | 14 +- .../source/FullTextSearchBuilderTest.java | 2 +- .../source/PrimaryKeyFullTextReadTest.java | 2 +- .../table/source/VectorSearchBuilderTest.java | 21 +-- paimon-full-text/README.md | 2 +- .../pypaimon/common/options/core_options.py | 49 +++--- .../data_evolution_global_index_coverage.py | 12 +- .../pypaimon/read/scanner/file_scanner.py | 2 +- .../pypaimon/table/source/full_text_scan.py | 5 +- .../table/source/full_text_search_builder.py | 4 +- .../source/primary_key_full_text_read.py | 4 +- .../table/source/primary_key_vector_read.py | 6 +- .../table/source/vector_search_read.py | 5 +- .../table/source/vector_search_scan.py | 11 +- .../global_index_scalar_search_mode_test.py | 55 ++++--- .../pypaimon/tests/global_index_test.py | 6 +- .../primary_key_index_definitions_test.py | 39 ++++- .../tests/vector_search_filter_test.py | 142 ++++++++++++------ paimon-python/pypaimon/write/table_update.py | 4 +- .../MergeIntoPaimonDataEvolutionTable.scala | 5 +- .../MergeIntoPaimonDataEvolutionTable.scala | 5 +- .../spark/sql/RowTrackingTestBase.scala | 2 + 36 files changed, 417 insertions(+), 166 deletions(-) diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx index ace4770ba7e8..731fb8c078ab 100644 --- a/docs/docs/multimodal-table/global-index.mdx +++ b/docs/docs/multimodal-table/global-index.mdx @@ -311,8 +311,8 @@ catalog.alter_table( Global index files cover row-id ranges. If more rows are appended after an index is built, those new rows are not automatically covered by the existing index files. Build the global index again -to create index files for newly uncovered data. By default, queries use fast search and only read -indexed row ranges; rows in uncovered ranges are not returned for that indexed query. +to create index files for newly uncovered data. Scalar queries default to `full` search, while +vector and full-text queries default to `fast` and only search indexed row ranges. To improve freshness for query types that support raw-data search, set: @@ -321,7 +321,7 @@ To improve freshness for query types that support raw-data search, set: ```sql -ALTER TABLE my_table SET ('global-index.search-mode' = 'full'); +ALTER TABLE my_table SET ('vector-index.search-mode' = 'full'); ``` @@ -333,14 +333,14 @@ from pypaimon.schema.schema_change import SchemaChange catalog.alter_table( "db.my_table", - [SchemaChange.set_option("global-index.search-mode", "full")], + [SchemaChange.set_option("vector-index.search-mode", "full")], ) ``` For a read-only override on an existing `Table` instance: ```python -full_table = table.copy({"global-index.search-mode": "full"}) +full_table = table.copy({"vector-index.search-mode": "full"}) ``` @@ -353,6 +353,11 @@ only when such a gap exists. Use `detail` search when data files may have been r after index creation; it scans data file metadata to find the exact unindexed row ranges and can handle index invalidation caused by updates or rewrites. +Use `scalar-index.search-mode`, `vector-index.search-mode`, or +`full-text-index.search-mode` for one index family. The legacy +`global-index.search-mode` has no default and is used as a fallback when the corresponding +family-specific option is not explicitly configured. + To temporarily disable global-index scan acceleration while keeping the index files, set: @@ -389,7 +394,10 @@ These table options affect global index build and read behavior: | Option | Default | Description | |---|---|---| | `global-index.enabled` | `true` | Whether scans can use global indexes. | -| `global-index.search-mode` | `fast` | Search mode for global-index queries. `fast` searches indexed data only. `full` checks snapshot `nextRowId` against global index row-id coverage and scans raw data only if a gap exists. `detail` scans data file metadata to find exact unindexed rows and can handle index invalidation caused by updates or rewrites. | +| `global-index.search-mode` | Not set | Legacy fallback search mode for global-index queries. Family-specific options take precedence. | +| `scalar-index.search-mode` | `full` | Search mode for BTree and Bitmap queries. | +| `vector-index.search-mode` | `fast` | Search mode for vector queries. | +| `full-text-index.search-mode` | `fast` | Search mode for full-text queries. | | `global-index.external-path` | Not set | Root directory for global index files. If not set, files are stored under the table index directory. | | `global-index.column-update-action` | `THROW_ERROR` | Action for updates to indexed columns. `THROW_ERROR` rejects the update, `DROP_PARTITION_INDEX` drops affected partition indexes, and `IGNORE` keeps existing index files unchanged. Use `IGNORE` only when a stale index is acceptable; for example, a vector updated from `NULL` to a value remains invisible until the index is rebuilt. | | `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree and Bitmap builds. | diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx index ae8ea59d61de..5886ef2d7fcc 100644 --- a/docs/docs/multimodal-table/global-index/full-text.mdx +++ b/docs/docs/multimodal-table/global-index/full-text.mdx @@ -282,8 +282,8 @@ print(pa_table) -By default, `global-index.search-mode=fast` searches indexed row ranges only. -When `global-index.search-mode` is `full` or `detail`, Paimon also covers +By default, `full-text-index.search-mode=fast` searches indexed row ranges only. +When `full-text-index.search-mode` is `full` or `detail`, Paimon also covers unindexed row ranges by reading raw rows and building a temporary native full-text index for the searched column. diff --git a/docs/docs/primary-key-table/global-index.mdx b/docs/docs/primary-key-table/global-index.mdx index 3098d71c52e4..e76845e6c390 100644 --- a/docs/docs/primary-key-table/global-index.mdx +++ b/docs/docs/primary-key-table/global-index.mdx @@ -234,7 +234,9 @@ schema validation. | `fields..pk-btree.index.options` | Not set | JSON object containing BTree build options. Unqualified keys are scoped to `btree-index`. | | `pk-bitmap.index.columns` | Not set | Comma-separated columns which own independent Bitmap indexes. | | `fields..pk-bitmap.index.options` | Not set | JSON object containing Bitmap build options. Unqualified keys are scoped to `bitmap-index`. | -| `global-index.search-mode` | `fast` | Search mode for primary-key Vector and Full Text queries. `fast` searches indexed data only, so uncovered files are omitted. For Vector, `full` and `detail` search uncovered files exactly. Primary-key Full Text supports only `fast`. | +| `global-index.search-mode` | Not set | Legacy fallback search mode. Family-specific options take precedence. | +| `vector-index.search-mode` | `fast` | Search mode for primary-key Vector queries. `full` and `detail` search uncovered files exactly. | +| `full-text-index.search-mode` | `fast` | Search mode for primary-key Full Text queries. Only `fast` is currently supported. | For algorithm-specific options, see the corresponding [BTree](../multimodal-table/global-index/btree), @@ -272,10 +274,10 @@ Index construction can execute asynchronously inside the writer. A writer which compaction also waits for active index maintenance; a non-blocking writer can complete maintenance in a later commit. Coverage can therefore be temporarily partial. -Coverage behavior depends on the index family and search mode. `global-index.search-mode` defaults -to `fast`. For primary-key Vector and Full Text queries, `fast` searches only data covered by an -active index group. Uncovered files are not searched, so partial coverage can make results -incomplete. +Coverage behavior depends on the index family and search mode. `vector-index.search-mode` and +`full-text-index.search-mode` default to `fast`. In `fast` mode, primary-key Vector and Full Text +queries search only data covered by an active index group. Uncovered files are not searched, so +partial coverage can make results incomplete. - BTree and Bitmap scans always read uncovered files through the ordinary data path. Partial coverage affects their acceleration, not result completeness. The original scalar predicate and @@ -283,7 +285,7 @@ incomplete. - Vector search in `full` or `detail` mode evaluates files without an active ANN group exactly and merges those results with ANN candidates. In the default `fast` mode, those files are omitted. -Primary-key Full Text currently supports only `global-index.search-mode = fast`. It searches +Primary-key Full Text currently supports only `full-text-index.search-mode = fast`. It searches persistent archives and ignores uncovered files; `full` and `detail` are rejected because a merge-aware logical-row fallback is not implemented. Consequently, newly appended Level-0 rows become full-text searchable only after compaction publishes an eligible data file and archive. diff --git a/docs/docs/pypaimon/data-evolution.md b/docs/docs/pypaimon/data-evolution.md index f5d58d75e4e9..43f31c63b845 100644 --- a/docs/docs/pypaimon/data-evolution.md +++ b/docs/docs/pypaimon/data-evolution.md @@ -99,7 +99,7 @@ You can use `update_by_predicate` for SQL-like `UPDATE ... SET ... WHERE ...` operations. The `Predicate` identifies rows to update, and the assignment map contains literal values for updated columns. When global indexes are available, `update_by_predicate` discovers matching -`_ROW_ID` values with `global-index.search-mode=full` on the configured +`_ROW_ID` values with `scalar-index.search-mode=full` on the configured point-in-time scan snapshot or, if none is configured, the latest snapshot. ```python diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 2841618e7c34..539ceaef90c6 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -752,6 +752,12 @@ Integer For streaming write, full compaction will be constantly triggered after delta commits. For batch write, full compaction will be triggered with each commit as long as this value is greater than 0. + +
full-text-index.search-mode
+ fast +

Enum

+ Search mode for full-text index queries.

Possible values:
  • "fast": Only search indexed data.
  • "full": Use snapshot next row id and global index coverage to detect missing row ids, and scan raw data only when a gap exists.
  • "detail": Scan data files to find exact unindexed rows. This can handle index invalidation caused by updates or rewrites.
+
global-index.build.max-parallelism
4096 @@ -790,9 +796,9 @@
global-index.search-mode
- fast + (none)

Enum

- Search mode for global index queries. Supported values are 'fast', 'full', and 'detail'.

Possible values:
  • "fast": Only search indexed data.
  • "full": Use snapshot next row id and global index coverage to detect missing row ids, and scan raw data only when a gap exists.
  • "detail": Scan data files to find exact unindexed rows. This can handle index invalidation caused by updates or rewrites.
+ Fallback search mode for global index queries.

Possible values:
  • "fast": Only search indexed data.
  • "full": Use snapshot next row id and global index coverage to detect missing row ids, and scan raw data only when a gap exists.
  • "detail": Scan data files to find exact unindexed rows. This can handle index invalidation caused by updates or rewrites.
global-index.thread-num
@@ -1320,6 +1326,12 @@ String The field that generates the row kind for primary key table, the row kind determines which data is '+I', '-U', '+U' or '-D'. + +
scalar-index.search-mode
+ full +

Enum

+ Search mode for scalar index queries.

Possible values:
  • "fast": Only search indexed data.
  • "full": Use snapshot next row id and global index coverage to detect missing row ids, and scan raw data only when a gap exists.
  • "detail": Scan data files to find exact unindexed rows. This can handle index invalidation caused by updates or rewrites.
+
scan.bounded.watermark
(none) @@ -1747,6 +1759,12 @@ String Specifies column names that should be stored as vector type. This is used when you want to treat a ARRAY column as a VECTOR. + +
vector-index.search-mode
+ fast +

Enum

+ Search mode for vector index queries.

Possible values:
  • "fast": Only search indexed data.
  • "full": Use snapshot next row id and global index coverage to detect missing row ids, and scan raw data only when a gap exists.
  • "detail": Scan data files to find exact unindexed rows. This can handle index invalidation caused by updates or rewrites.
+
vector-search.distribute.enabled
false diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index e0c7fd4004c7..39c0d16e14db 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2705,11 +2705,27 @@ public String toString() { public static final ConfigOption GLOBAL_INDEX_SEARCH_MODE = key("global-index.search-mode") + .enumType(GlobalIndexSearchMode.class) + .noDefaultValue() + .withDescription("Fallback search mode for global index queries."); + + public static final ConfigOption SCALAR_INDEX_SEARCH_MODE = + key("scalar-index.search-mode") + .enumType(GlobalIndexSearchMode.class) + .defaultValue(GlobalIndexSearchMode.FULL) + .withDescription("Search mode for scalar index queries."); + + public static final ConfigOption VECTOR_INDEX_SEARCH_MODE = + key("vector-index.search-mode") .enumType(GlobalIndexSearchMode.class) .defaultValue(GlobalIndexSearchMode.FAST) - .withDescription( - "Search mode for global index queries. " - + "Supported values are 'fast', 'full', and 'detail'."); + .withDescription("Search mode for vector index queries."); + + public static final ConfigOption FULL_TEXT_INDEX_SEARCH_MODE = + key("full-text-index.search-mode") + .enumType(GlobalIndexSearchMode.class) + .defaultValue(GlobalIndexSearchMode.FAST) + .withDescription("Search mode for full-text index queries."); public static final ConfigOption GLOBAL_INDEX_THREAD_NUM = key("global-index.thread-num") @@ -4312,10 +4328,32 @@ public boolean globalIndexEnabled() { return options.get(GLOBAL_INDEX_ENABLED); } + @Nullable public GlobalIndexSearchMode globalIndexSearchMode() { return options.get(GLOBAL_INDEX_SEARCH_MODE); } + public GlobalIndexSearchMode scalarIndexSearchMode() { + return indexSearchMode(SCALAR_INDEX_SEARCH_MODE); + } + + public GlobalIndexSearchMode vectorIndexSearchMode() { + return indexSearchMode(VECTOR_INDEX_SEARCH_MODE); + } + + public GlobalIndexSearchMode fullTextIndexSearchMode() { + return indexSearchMode(FULL_TEXT_INDEX_SEARCH_MODE); + } + + private GlobalIndexSearchMode indexSearchMode( + ConfigOption familySearchMode) { + if (options.contains(familySearchMode)) { + return options.get(familySearchMode); + } + return options.getOptional(GLOBAL_INDEX_SEARCH_MODE) + .orElseGet(() -> options.get(familySearchMode)); + } + public Integer globalIndexThreadNum() { return options.get(GLOBAL_INDEX_THREAD_NUM); } diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java index aacae5ce681b..6dbbca8e7705 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java @@ -315,7 +315,7 @@ private Optional evalGlobalIndex() { LOG.info( "Scan table '{}' with global index. searchMode='{}', total={} ms, metadata={} ms, lookup={} ms, coverage={} ms.", table.name(), - options.globalIndexSearchMode(), + options.scalarIndexSearchMode(), totalDuration / 1_000_000, metadataDuration / 1_000_000, lookupDuration / 1_000_000, diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java index 55678824e62c..122c5f404e81 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java @@ -51,6 +51,7 @@ public class DataEvolutionGlobalIndexCoverage { private final FileStoreTable table; @Nullable private final Snapshot snapshot; @Nullable private final PartitionPredicate partitionFilter; + private final GlobalIndexSearchMode searchMode; private final Map> coverageByField; public DataEvolutionGlobalIndexCoverage( @@ -58,9 +59,24 @@ public DataEvolutionGlobalIndexCoverage( @Nullable Snapshot snapshot, @Nullable PartitionPredicate partitionFilter, Collection indexFiles) { + this( + table, + snapshot, + partitionFilter, + indexFiles, + table.coreOptions().scalarIndexSearchMode()); + } + + public DataEvolutionGlobalIndexCoverage( + FileStoreTable table, + @Nullable Snapshot snapshot, + @Nullable PartitionPredicate partitionFilter, + Collection indexFiles, + GlobalIndexSearchMode searchMode) { this.table = table; this.snapshot = snapshot; this.partitionFilter = partitionFilter; + this.searchMode = checkNotNull(searchMode); this.coverageByField = new HashMap<>(); for (IndexFileMeta indexFile : indexFiles) { GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta()); @@ -83,7 +99,6 @@ public List unindexedRanges(int fieldId) { } public List unindexedRanges(Collection fieldIds) { - GlobalIndexSearchMode searchMode = table.coreOptions().globalIndexSearchMode(); if (searchMode == GlobalIndexSearchMode.FAST) { return Collections.emptyList(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java index 5f07e8a17e5d..215e1b31c56d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java @@ -93,7 +93,12 @@ private DataEvolutionGlobalIndexScanner( GlobalIndexReadThreadPool.getExecutorService(options.get(GLOBAL_INDEX_THREAD_NUM)); this.indexPathFactory = indexPathFactory; this.coverage = - new DataEvolutionGlobalIndexCoverage(table, snapshot, partitionFilter, indexFiles); + new DataEvolutionGlobalIndexCoverage( + table, + snapshot, + partitionFilter, + indexFiles, + table.coreOptions().scalarIndexSearchMode()); GlobalIndexFileReader indexFileReader = meta -> fileIO.newInputStream(meta.filePath()); Map indexMetas = new HashMap<>(); Map> extraIndexMetas = new HashMap<>(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java index 1b6ef7adc71d..3dc1f36812f0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java @@ -117,7 +117,11 @@ public Plan scan() { if (!allIndexFiles.isEmpty()) { List rawRowRanges = new DataEvolutionGlobalIndexCoverage( - table, snapshot, partitionFilter, allIndexFiles) + table, + snapshot, + partitionFilter, + allIndexFiles, + table.coreOptions().fullTextIndexSearchMode()) .unindexedRanges(textColumnIds); if (!rawRowRanges.isEmpty()) { splits.add(new RawFullTextSearchSplit(rawRowRanges)); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java index 317ffc6d5235..83a9a98debff 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java @@ -146,7 +146,11 @@ public Plan scan() { List rawRowRanges = new DataEvolutionGlobalIndexCoverage( - table, snapshot, partitionFilter, vectorIndexFiles) + table, + snapshot, + partitionFilter, + vectorIndexFiles, + table.coreOptions().vectorIndexSearchMode()) .unindexedRanges(vectorColumn.id()); if (filter != null) { rawRowRanges = @@ -157,7 +161,8 @@ public Plan scan() { table, snapshot, partitionFilter, - scalarIndexFiles(allIndexFiles)) + scalarIndexFiles(allIndexFiles), + table.coreOptions().scalarIndexSearchMode()) .unindexedRanges(table.rowType(), filter)), true); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java index 3374141a33e7..c514301be05d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java @@ -69,7 +69,7 @@ public PrimaryKeyFullTextRead( definition.fieldId() == textField.id(), "Full-text definition does not match field %s.", textField.name()); - checkFastSearchMode(table.coreOptions().globalIndexSearchMode()); + checkFastSearchMode(table.coreOptions().fullTextIndexSearchMode()); ProductionSearch production = new ProductionSearch(table, definition, textField, query, limit); this.limit = limit; @@ -87,7 +87,7 @@ public PrimaryKeyFullTextRead( private static void checkFastSearchMode(GlobalIndexSearchMode searchMode) { if (searchMode != GlobalIndexSearchMode.FAST) { throw new UnsupportedOperationException( - "Primary-key full-text search only supports the FAST global-index search mode; " + "Primary-key full-text search only supports the FAST full-text-index search mode; " + "FULL and DETAIL require merge-aware logical-row fallback."); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java index 46232947f3b1..6a632ed64607 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java @@ -313,7 +313,7 @@ CompletableFuture> searchBatchAsync( annSearcher, searchOptions, metric, - table.coreOptions().globalIndexSearchMode()); + table.coreOptions().vectorIndexSearchMode()); return bucketSearch .searchBatchAsync( state, diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index b3e91a3a4b53..96643624a3a0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -107,6 +107,39 @@ public void testIgnoreGlobalIndexColumnUpdateAction() { .isEqualTo(CoreOptions.GlobalIndexColumnUpdateAction.IGNORE); } + @Test + public void testIndexSearchModes() { + Options conf = new Options(); + CoreOptions options = new CoreOptions(conf); + assertThat(options.globalIndexSearchMode()).isNull(); + assertThat(options.scalarIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL); + assertThat(options.vectorIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FAST); + assertThat(options.fullTextIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FAST); + + conf.set(CoreOptions.GLOBAL_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.DETAIL); + options = new CoreOptions(conf); + assertThat(options.scalarIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.DETAIL); + assertThat(options.vectorIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.DETAIL); + assertThat(options.fullTextIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.DETAIL); + + conf.set(CoreOptions.SCALAR_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.FAST); + conf.set(CoreOptions.VECTOR_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.FULL); + conf.set(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.FULL); + options = new CoreOptions(conf); + assertThat(options.scalarIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FAST); + assertThat(options.vectorIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL); + assertThat(options.fullTextIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL); + } + @Test public void testBlobSplitByFileSizeDefault() { Options conf = new Options(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java index c0d8ea01aaa6..ff1d993cf8eb 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java @@ -208,7 +208,7 @@ public void testGlobalIndexDiagnosticLogs() throws Exception { assertThat(logs) .containsPattern( "INFO Scan table '[^']+' with global index\\. " - + "searchMode='fast', total=\\d+ ms, metadata=\\d+ ms, " + + "searchMode='full', total=\\d+ ms, metadata=\\d+ ms, " + "lookup=\\d+ ms, coverage=\\d+ ms\\.") .containsPattern( "INFO Global index lookup table='[^']+', type='btree', " @@ -238,8 +238,9 @@ public void testBTreeGlobalIndexSearchModeControlsUnindexedData() throws Excepti BinaryString.fromString("a100"), BinaryString.fromString("a700"))); - assertThat(readF1(table, predicate)).containsExactly("a100"); + assertThat(readF1(table, predicate)).containsExactly("a100", "a700"); + assertThat(readF1(tableWithSearchMode(table, "fast"), predicate)).containsExactly("a100"); assertThat(readF1(tableWithSearchMode(table, "full"), predicate)) .containsExactly("a100", "a700"); assertThat(readF1(tableWithSearchMode(table, "detail"), predicate)) @@ -251,7 +252,8 @@ public void testBTreeGlobalIndexSearchModeControlsUnindexedData() throws Excepti builder.equal(1, BinaryString.fromString("a700")), builder.equal(2, BinaryString.fromString("b700"))); - assertThat(readF1(table, andWithUnindexedField)).isEmpty(); + assertThat(readF1(table, andWithUnindexedField)).containsExactly("a700"); + assertThat(readF1(tableWithSearchMode(table, "fast"), andWithUnindexedField)).isEmpty(); assertThat(readF1(tableWithSearchMode(table, "full"), andWithUnindexedField)) .containsExactly("a700"); assertThat(readF1(tableWithSearchMode(table, "detail"), andWithUnindexedField)) @@ -358,7 +360,8 @@ public void testBTreeGlobalIndexSearchModeUsesAllPredicateFieldCoverage() throws builder.equal(1, BinaryString.fromString("a700")), builder.equal(2, BinaryString.fromString("b700"))); - assertThat(readF1(table, andPredicate)).isEmpty(); + assertThat(readF1(table, andPredicate)).containsExactly("a700"); + assertThat(readF1(tableWithSearchMode(table, "fast"), andPredicate)).isEmpty(); assertThat(readF1(tableWithSearchMode(table, "full"), andPredicate)) .containsExactly("a700"); assertThat(readF1(tableWithSearchMode(table, "detail"), andPredicate)) @@ -369,7 +372,8 @@ public void testBTreeGlobalIndexSearchModeUsesAllPredicateFieldCoverage() throws builder.equal(1, BinaryString.fromString("a700")), builder.equal(2, BinaryString.fromString("b701"))); - assertThat(readF1(table, orPredicate)).containsExactly("a701"); + assertThat(readF1(table, orPredicate)).containsExactly("a700", "a701"); + assertThat(readF1(tableWithSearchMode(table, "fast"), orPredicate)).containsExactly("a701"); assertThat(readF1(tableWithSearchMode(table, "full"), orPredicate)) .containsExactly("a700", "a701"); assertThat(readF1(tableWithSearchMode(table, "detail"), orPredicate)) 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 12af712f1d36..16c418dc64c5 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 @@ -190,7 +190,7 @@ public void testFullTextSearchNonFastModesScanUnindexedData() throws Exception { (FileStoreTable) table.copy( Collections.singletonMap( - CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), + CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(), searchMode)); GlobalIndexResult result = nonFastModeTable diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java index c00ecb18a27b..a791a605b279 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java @@ -100,7 +100,7 @@ void testRejectsIncompleteSearchModes(GlobalIndexSearchMode mode) { new PrimaryKeyFullTextRead( mode, 10, split -> Collections.emptyList())) .isInstanceOf(UnsupportedOperationException.class) - .hasMessageContaining("only supports the FAST global-index search mode"); + .hasMessageContaining("only supports the FAST full-text-index search mode"); } private static PrimaryKeySearchPosition position( 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 2153b44f2b3a..7c91bce60840 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 @@ -366,7 +366,7 @@ public void testFullModeRawOnlyUsesConfiguredMetric() throws Exception { catalog.createTable( identifier("full_search_raw_only_cosine_table"), vectorSchemaBuilder(VECTOR_FIELD_NAME) - .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full") + .option(CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(), "full") .option("test.vector.metric", "cosine") .build(), false); @@ -1067,12 +1067,10 @@ public void testPreFilterMatchesZeroRows() throws Exception { @Test public void testPartialScalarPreFilterMustNotDropUnindexedScalarRows() throws Exception { catalog.createTable( - identifier("full_search_partial_scalar_unindexed_table"), - vectorSchemaBuilder(VECTOR_FIELD_NAME) - .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full") - .build(), + identifier("default_scalar_full_partial_index_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME).build(), false); - FileStoreTable table = getTable(identifier("full_search_partial_scalar_unindexed_table")); + FileStoreTable table = getTable(identifier("default_scalar_full_partial_index_table")); float[][] vectors = new float[10][]; for (int i = 0; i < vectors.length; i++) { @@ -1144,9 +1142,14 @@ public void testFullModeFilterWithoutScalarIndexMustNotLetVectorIndexPolluteTopK } @Test - public void testFastModePartialScalarPreFilterOnlyUsesIndexedRows() throws Exception { - createTableDefault(); - FileStoreTable table = getTableDefault(); + public void testFastScalarModePartialPreFilterOnlyUsesIndexedRows() throws Exception { + catalog.createTable( + identifier("fast_scalar_partial_index_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "fast") + .build(), + false); + FileStoreTable table = getTable(identifier("fast_scalar_partial_index_table")); float[][] vectors = new float[10][]; for (int i = 0; i < vectors.length; i++) { diff --git a/paimon-full-text/README.md b/paimon-full-text/README.md index 2ac1ae0e9b34..2d17cccfb95d 100644 --- a/paimon-full-text/README.md +++ b/paimon-full-text/README.md @@ -119,7 +119,7 @@ Level-1-or-higher data level. Data compaction replaces the affected level archiv archive can cover multiple ordered source files; its row IDs concatenate their physical row positions. -Primary-key full-text search currently supports only `global-index.search-mode=fast`. Level-0 and +Primary-key full-text search currently supports only `full-text-index.search-mode=fast`. Level-0 and other uncovered files are not searched; their rows become searchable after compaction publishes an eligible data file and persistent archive. Search applies each source file's deletion vector, preserves native relevance scores, and selects a global Top-K. Only Hybrid search rewrites route diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index ec56a6d67b1d..392ea653a094 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -731,23 +731,29 @@ class CoreOptions: GLOBAL_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( ConfigOptions.key("global-index.search-mode") .enum_type(GlobalIndexSearchMode) - .default_value(GlobalIndexSearchMode.FAST) - .with_description( - "Search mode for vector and full-text global index queries. " - "Scalar index queries use 'scalar-index.search-mode'. " - "Supported values are 'fast', 'full', and 'detail'." - ) + .no_default_value() + .with_description("Fallback search mode for global index queries.") ) - GLOBAL_INDEX_SCALAR_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + SCALAR_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( ConfigOptions.key("scalar-index.search-mode") .enum_type(GlobalIndexSearchMode) .default_value(GlobalIndexSearchMode.FULL) - .with_description( - "Search mode for scalar (sorted/bitmap) global index queries. " - "Defaults to 'full' to include rows not covered by the index. " - "Supported values are 'fast', 'full', and 'detail'." - ) + .with_description("Search mode for scalar index queries.") + ) + + VECTOR_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + ConfigOptions.key("vector-index.search-mode") + .enum_type(GlobalIndexSearchMode) + .default_value(GlobalIndexSearchMode.FAST) + .with_description("Search mode for vector index queries.") + ) + + FULL_TEXT_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + ConfigOptions.key("full-text-index.search-mode") + .enum_type(GlobalIndexSearchMode) + .default_value(GlobalIndexSearchMode.FAST) + .with_description("Search mode for full-text index queries.") ) GLOBAL_INDEX_EXTERNAL_PATH: ConfigOption[str] = ( @@ -1384,11 +1390,20 @@ def global_index_enabled(self, default=None): def global_index_search_mode(self): return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) - def global_index_scalar_search_mode(self): - scalar_mode = self.options.get(CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE) - if scalar_mode is not None: - return scalar_mode - return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) + def scalar_index_search_mode(self): + return self._family_index_search_mode(CoreOptions.SCALAR_INDEX_SEARCH_MODE) + + def vector_index_search_mode(self): + return self._family_index_search_mode(CoreOptions.VECTOR_INDEX_SEARCH_MODE) + + def full_text_index_search_mode(self): + return self._family_index_search_mode(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE) + + def _family_index_search_mode(self, option): + if self.options.contains(option): + return self.options.get(option) + global_mode = self.global_index_search_mode() + return global_mode if global_mode is not None else self.options.get(option) def global_index_external_path(self, default=None): value = self.options.get(CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH, default) diff --git a/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py b/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py index 878a87a5680a..19aa1c251fe2 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py @@ -91,7 +91,7 @@ def _unindexed_ranges( search_mode: Optional[GlobalIndexSearchMode] = None, ) -> List[Range]: if search_mode is None: - search_mode = _global_index_search_mode(self._table) + search_mode = _scalar_index_search_mode(self._table) if search_mode == GlobalIndexSearchMode.FAST: return [] next_row_id = getattr(self._snapshot, "next_row_id", None) @@ -144,13 +144,13 @@ def _entry_matches_partition(self, entry) -> bool: return self._partition_filter.test(entry.partition) -def _global_index_search_mode(table): +def _scalar_index_search_mode(table): options = getattr(table, "options", None) if options is None: - return GlobalIndexSearchMode.FAST - if hasattr(options, "global_index_search_mode"): - return options.global_index_search_mode() - return CoreOptions(Options.from_none()).global_index_search_mode() + return GlobalIndexSearchMode.FULL + if hasattr(options, "scalar_index_search_mode"): + return options.scalar_index_search_mode() + return CoreOptions(Options.from_none()).scalar_index_search_mode() def _is_field_id_collection(fields_or_field_id): diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index ac70b5e00b68..c2fcff26a9c9 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -498,7 +498,7 @@ def _eval_global_index(self, snapshot=None): result = scanner.scan(self.predicate) if result is None: return None - scalar_mode = self.table.options.global_index_scalar_search_mode() + scalar_mode = self.table.options.scalar_index_search_mode() return result.or_( scanner.unindexed_rows(self.predicate, search_mode=scalar_mode)) except Exception: diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py index 5e0038a1841c..1387b02536bc 100644 --- a/paimon-python/pypaimon/table/source/full_text_scan.py +++ b/paimon-python/pypaimon/table/source/full_text_scan.py @@ -122,7 +122,10 @@ def index_file_filter(entry): snapshot, partition_filter, all_index_files, - ).unindexed_ranges(list(text_column_ids)) + ).unindexed_ranges( + list(text_column_ids), + search_mode=self._table.options.full_text_index_search_mode(), + ) if raw_row_ranges: splits.append(RawFullTextSearchSplit(raw_row_ranges)) 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 e98a74cef351..1e094e543f62 100644 --- a/paimon-python/pypaimon/table/source/full_text_search_builder.py +++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py @@ -137,11 +137,11 @@ def new_full_text_read(self) -> FullTextRead: from pypaimon.common.options.core_options import GlobalIndexSearchMode mode = CoreOptions( Options(dict(self._table.table_schema.options)) - ).global_index_search_mode() + ).full_text_index_search_mode() if mode != GlobalIndexSearchMode.FAST: raise NotImplementedError( "Primary-key full-text search only supports the FAST " - "global-index search mode; FULL and DETAIL require " + "full-text index search mode; FULL and DETAIL require " "merge-aware logical-row fallback.") from pypaimon.table.source.primary_key_full_text_read import PrimaryKeyFullTextRead return PrimaryKeyFullTextRead( diff --git a/paimon-python/pypaimon/table/source/primary_key_full_text_read.py b/paimon-python/pypaimon/table/source/primary_key_full_text_read.py index 7b447c37a319..2386cdf2d2f2 100644 --- a/paimon-python/pypaimon/table/source/primary_key_full_text_read.py +++ b/paimon-python/pypaimon/table/source/primary_key_full_text_read.py @@ -40,11 +40,11 @@ def __init__(self, table, limit, text_column, query, definition, partition_filter=None): mode = CoreOptions( Options(dict(table.table_schema.options)) - ).global_index_search_mode() + ).full_text_index_search_mode() if mode != GlobalIndexSearchMode.FAST: raise NotImplementedError( "Primary-key full-text search only supports the FAST " - "global-index search mode; FULL and DETAIL require " + "full-text index search mode; FULL and DETAIL require " "merge-aware logical-row fallback.") self._definition = definition super().__init__(table, limit, text_column, query, partition_filter) diff --git a/paimon-python/pypaimon/table/source/primary_key_vector_read.py b/paimon-python/pypaimon/table/source/primary_key_vector_read.py index 8850840e3943..fd6a65e83ec4 100644 --- a/paimon-python/pypaimon/table/source/primary_key_vector_read.py +++ b/paimon-python/pypaimon/table/source/primary_key_vector_read.py @@ -18,6 +18,7 @@ from concurrent.futures import wait from heapq import nsmallest +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.index.pk.primary_key_index_source_meta import PrimaryKeyIndexSourceMeta from pypaimon.table.source.primary_key_scored_result import ( PrimaryKeyScoredResult, PrimaryKeySearchPosition, _partition_bytes) @@ -87,7 +88,10 @@ def indexed_candidate_iter(): plan, indexed_candidates, index_type) else: indexed_candidates = _top_k(indexed_candidates, self._limit) - exact_candidates = _top_k(self._raw_candidates(plan), self._limit) + exact_candidates = [] + if (self._table.options.vector_index_search_mode() + != GlobalIndexSearchMode.FAST): + exact_candidates = _top_k(self._raw_candidates(plan), self._limit) candidates = _top_k( (candidate for group in (indexed_candidates, exact_candidates) diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py b/paimon-python/pypaimon/table/source/vector_search_read.py index eba11094f3fd..84b0ae7c0014 100644 --- a/paimon-python/pypaimon/table/source/vector_search_read.py +++ b/paimon-python/pypaimon/table/source/vector_search_read.py @@ -190,7 +190,10 @@ def _raw_pre_filter(self, splits): include = result.results() include = RoaringBitmap64.or_( include, - scanner.unindexed_rows(self._filter).results()) + scanner.unindexed_rows( + self._filter, + search_mode=self._table.options.scalar_index_search_mode(), + ).results()) return RoaringBitmap64.and_(include, raw_rows) finally: scanner.close() diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py b/paimon-python/pypaimon/table/source/vector_search_scan.py index e9f65a5b6147..1cbdb19c1492 100644 --- a/paimon-python/pypaimon/table/source/vector_search_scan.py +++ b/paimon-python/pypaimon/table/source/vector_search_scan.py @@ -168,7 +168,10 @@ def index_file_filter(entry): snapshot, partition_filter, vector_index_files, - ).unindexed_ranges(vector_column.id) + ).unindexed_ranges( + vector_column.id, + search_mode=self._table.options.vector_index_search_mode(), + ) scalar_index_files = [ f for f in all_index_files if f.global_index_meta is not None @@ -182,7 +185,11 @@ def index_file_filter(entry): snapshot, partition_filter, scalar_index_files, - ).unindexed_ranges(self._table.fields, self._filter), + ).unindexed_ranges( + self._table.fields, + self._filter, + search_mode=self._table.options.scalar_index_search_mode(), + ), True, ) if raw_row_ranges: diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py index 6db1b820ee75..b900c52ca0c1 100644 --- a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -17,7 +17,6 @@ import unittest from types import SimpleNamespace -from unittest import mock from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode from pypaimon.common.options.options import Options @@ -52,37 +51,35 @@ class ScalarGlobalIndexSearchModeTest(unittest.TestCase): def test_default_values(self): options = CoreOptions(Options.from_none()) + self.assertIsNone(options.global_index_search_mode()) self.assertEqual( - GlobalIndexSearchMode.FULL, options.global_index_scalar_search_mode()) + GlobalIndexSearchMode.FULL, options.scalar_index_search_mode()) self.assertEqual( - GlobalIndexSearchMode.FAST, options.global_index_search_mode()) - - def test_scalar_option_override(self): - options = CoreOptions(Options({"scalar-index.search-mode": "fast"})) - self.assertEqual( - GlobalIndexSearchMode.FAST, options.global_index_scalar_search_mode()) - - def test_does_not_inherit_global_index_search_mode(self): - for mode in ("fast", "detail"): - options = CoreOptions(Options({"global-index.search-mode": mode})) - self.assertEqual( - GlobalIndexSearchMode.FULL, - options.global_index_scalar_search_mode()) - - def test_falls_back_when_scalar_mode_has_no_value(self): - options = mock.Mock() - options.get.side_effect = [None, GlobalIndexSearchMode.DETAIL] + GlobalIndexSearchMode.FAST, options.vector_index_search_mode()) self.assertEqual( - GlobalIndexSearchMode.DETAIL, - CoreOptions(options).global_index_scalar_search_mode()) + GlobalIndexSearchMode.FAST, options.full_text_index_search_mode()) - def test_scalar_key_wins_over_global_index_key(self): + def test_legacy_global_mode_is_family_fallback(self): + options = CoreOptions(Options({"global-index.search-mode": "detail"})) + for getter in ( + options.scalar_index_search_mode, + options.vector_index_search_mode, + options.full_text_index_search_mode): + self.assertEqual(GlobalIndexSearchMode.DETAIL, getter()) + + def test_family_modes_override_global_mode(self): options = CoreOptions(Options({ - "scalar-index.search-mode": "detail", - "global-index.search-mode": "fast", + "global-index.search-mode": "detail", + "scalar-index.search-mode": "fast", + "vector-index.search-mode": "full", + "full-text-index.search-mode": "fast", })) self.assertEqual( - GlobalIndexSearchMode.DETAIL, options.global_index_scalar_search_mode()) + GlobalIndexSearchMode.FAST, options.scalar_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FULL, options.vector_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.full_text_index_search_mode()) def test_coverage_honours_search_mode_override(self): coverage = _coverage(CoreOptions(Options.from_none())) @@ -90,7 +87,9 @@ def test_coverage_honours_search_mode_override(self): [], coverage.unindexed_ranges(1, search_mode=GlobalIndexSearchMode.FAST)) full = coverage.unindexed_ranges(1, search_mode=GlobalIndexSearchMode.FULL) self.assertEqual([(100, 199)], [(r.from_, r.to) for r in full]) - self.assertEqual([], coverage.unindexed_ranges(1)) + self.assertEqual( + [(100, 199)], + [(r.from_, r.to) for r in coverage.unindexed_ranges(1)]) def test_scanner_applies_passed_scalar_mode(self): coverage = _coverage(CoreOptions(Options.from_none())) @@ -99,11 +98,11 @@ def test_scanner_applies_passed_scalar_mode(self): scanner, None, search_mode=GlobalIndexSearchMode.FULL) self.assertEqual([(100, 199)], _ranges(result)) - def test_scanner_default_is_general_mode(self): + def test_scanner_default_is_scalar_mode(self): coverage = _coverage(CoreOptions(Options.from_none())) scanner = SimpleNamespace(_coverage=coverage, _fields=[1]) result = DataEvolutionGlobalIndexScanner.unindexed_rows(scanner, None) - self.assertEqual([], _ranges(result)) + self.assertEqual([(100, 199)], _ranges(result)) if __name__ == "__main__": diff --git a/paimon-python/pypaimon/tests/global_index_test.py b/paimon-python/pypaimon/tests/global_index_test.py index 22d57f6d4118..627da08af0bc 100644 --- a/paimon-python/pypaimon/tests/global_index_test.py +++ b/paimon-python/pypaimon/tests/global_index_test.py @@ -80,8 +80,8 @@ class _CoverageOptions: def __init__(self, mode): self.options = Options({"global-index.search-mode": mode}) - def global_index_search_mode(self): - return CoreOptions(self.options).global_index_search_mode() + def scalar_index_search_mode(self): + return CoreOptions(self.options).scalar_index_search_mode() class _CoverageTable: @@ -223,7 +223,7 @@ class _Options: def global_index_enabled(self): return True - def global_index_scalar_search_mode(self): + def scalar_index_search_mode(self): return GlobalIndexSearchMode.FULL class _Table: diff --git a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py index bf1e56009752..5a599f2699e1 100644 --- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py +++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py @@ -18,6 +18,7 @@ from unittest import mock from types import SimpleNamespace +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.index.pk.primary_key_index_definitions import PrimaryKeyIndexDefinitions from pypaimon.index.pk.primary_key_index_source_file import PrimaryKeyIndexSourceFile from pypaimon.index.pk.primary_key_index_source_meta import PrimaryKeyIndexSourceMeta @@ -129,6 +130,22 @@ def builder(mode): self.assertIsInstance( builder("fast").new_full_text_read(), PrimaryKeyFullTextRead) + schema = TableSchema( + fields=fields, + options={ + "pk-full-text.index.columns": "content", + "global-index.search-mode": "full", + "full-text-index.search-mode": "fast", + }) + table = SimpleNamespace( + fields=fields, table_schema=schema, partition_keys=[]) + self.assertIsInstance( + FullTextSearchBuilderImpl(table) + .with_query("content", "paimon") + .with_limit(2) + .new_full_text_read(), + PrimaryKeyFullTextRead, + ) for mode in ("full", "detail"): with self.subTest(mode=mode), self.assertRaisesRegex( NotImplementedError, "only supports the FAST"): @@ -194,7 +211,8 @@ def test_empty_pk_vector_ranges_mean_no_candidates(self): def test_pk_vector_refine_factor_overflow_is_rejected_before_search(self): table = SimpleNamespace(options=SimpleNamespace( - primary_key_vector_index_type=lambda column: "ivf-pq")) + primary_key_vector_index_type=lambda column: "ivf-pq", + vector_index_search_mode=lambda: GlobalIndexSearchMode.FAST)) reader = PrimaryKeyVectorRead( table, 2147483647, DataField(1, "embedding", AtomicType("VECTOR(1)")), [0.0], options={"ivf.refine_factor": "2"}) @@ -205,6 +223,25 @@ def test_pk_vector_refine_factor_overflow_is_rejected_before_search(self): with self.assertRaisesRegex(ValueError, "limit overflow"): reader.read_plan(plan) + def test_pk_vector_raw_fallback_uses_vector_search_mode(self): + from pypaimon.table.source.primary_key_vector_scan import ( + PrimaryKeyVectorScanPlan) + + field = DataField(1, "embedding", AtomicType("VECTOR(1)")) + plan = PrimaryKeyVectorScanPlan(1, []) + for mode, expected_calls in ( + (GlobalIndexSearchMode.FAST, 0), + (GlobalIndexSearchMode.FULL, 1), + (GlobalIndexSearchMode.DETAIL, 1)): + table = SimpleNamespace(options=SimpleNamespace( + primary_key_vector_index_type=lambda column: "ivf-pq", + vector_index_search_mode=lambda mode=mode: mode)) + reader = PrimaryKeyVectorRead(table, 1, field, [0.0]) + with self.subTest(mode=mode), mock.patch.object( + reader, "_raw_candidates", return_value=[]) as raw_candidates: + reader.read_plan(plan) + self.assertEqual(expected_calls, raw_candidates.call_count) + def test_full_text_payload_planner_rejects_duplicate_and_stale_levels(self): files = [SimpleNamespace(file_name="a.parquet", row_count=2, level=1)] source_meta = PrimaryKeyIndexSourceMeta( diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 55fe9bb07bc4..683d64b1de9a 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -30,6 +30,8 @@ from typing import List from unittest import mock +from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode +from pypaimon.common.options.options import Options from pypaimon.common.predicate import Predicate from pypaimon.common.predicate_builder import PredicateBuilder from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta @@ -95,6 +97,7 @@ def __init__(self, fields, entries, partition_fields=None): self.partition_keys: List[str] = [ f.name for f in self.partition_keys_fields] self.table_schema = _StubSchema() + self.options = CoreOptions(Options.from_none()) self.file_io = object() self._entries = entries @@ -805,26 +808,18 @@ def test_full_text_scan_ignores_other_index_types_on_same_column(self): [f.file_name for f in splits[0].full_text_index_files]) def test_full_text_scan_adds_raw_split_for_uncovered_ranges(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options from pypaimon.table.source.full_text_scan import DataEvolutionFullTextScan from pypaimon.table.source.full_text_search_split import ( IndexFullTextSearchSplit, RawFullTextSearchSplit, ) - class _Options: - options = Options({"global-index.search-mode": "full"}) - - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() - text_field = _field(1, "content", "STRING") entry = _entry( None, field_id=1, index_type="full-text", file_name="ft.index", row_range_start=0, row_range_end=4) table = _StubTable(fields=[text_field], entries=[entry]) - table.options = _Options() + table.options = CoreOptions(Options({"global-index.search-mode": "full"})) _patch_snapshot(self, [entry], types.SimpleNamespace(next_row_id=10)) splits = DataEvolutionFullTextScan(table, [text_field]).scan().splits() @@ -835,6 +830,26 @@ def global_index_search_mode(self_inner): self.assertEqual(1, len(raw)) self.assertEqual([Range(5, 9)], raw[0].row_ranges) + def test_full_text_mode_overrides_global_mode_for_uncovered_ranges(self): + from pypaimon.table.source.full_text_scan import DataEvolutionFullTextScan + from pypaimon.table.source.full_text_search_split import RawFullTextSearchSplit + + text_field = _field(1, "content", "STRING") + entry = _entry( + None, field_id=1, index_type="full-text", + file_name="ft.index", row_range_start=0, row_range_end=4) + table = _StubTable(fields=[text_field], entries=[entry]) + table.options = CoreOptions(Options({ + "global-index.search-mode": "full", + "full-text-index.search-mode": "fast", + })) + _patch_snapshot(self, [entry], types.SimpleNamespace(next_row_id=10)) + + splits = DataEvolutionFullTextScan(table, [text_field]).scan().splits() + + self.assertFalse(any(isinstance(split, RawFullTextSearchSplit) + for split in splits)) + class VectorSearchFilterTest(unittest.TestCase): """Non-partitioned wiring: scan + read + external_path plumbing.""" @@ -1224,6 +1239,9 @@ def test_refine_factor_query_options_override_table_options(self): class _Options: options = Options({"ivf.refine_factor": "2"}) + def vector_index_search_mode(self_inner): + return CoreOptions(self_inner.options).vector_index_search_mode() + entry = _entry(None, field_id=1, index_type="ivf-pq", file_name="vec.index", row_range_start=0, row_range_end=9) @@ -1317,26 +1335,18 @@ def close(self_inner): captured_io_metas[0][0].external_path) def test_full_mode_scan_adds_raw_split_for_unindexed_vector_rows(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options from pypaimon.table.source.vector_search_split import ( IndexVectorSearchSplit, RawVectorSearchSplit, ) - class _Options: - options = Options({"global-index.search-mode": "full"}) - - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() - class _Snapshots: def get_latest_snapshot(self_inner): return types.SimpleNamespace(next_row_id=10) table = _StubTable(fields=[self.id_field, self.embedding_field], entries=[self.entries[0]]) - table.options = _Options() + table.options = CoreOptions(Options({"global-index.search-mode": "full"})) table.snapshot_manager = lambda: _Snapshots() self._scan_patch.stop() self._travel_patch.stop() @@ -1360,16 +1370,40 @@ def get_latest_snapshot(self_inner): self.assertEqual(1, len(raw)) self.assertEqual([Range(5, 9)], raw[0].row_ranges) - def test_full_mode_scan_adds_raw_split_for_uncovered_scalar_filter(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options + def test_vector_mode_overrides_global_mode_for_unindexed_vector_rows(self): from pypaimon.table.source.vector_search_split import RawVectorSearchSplit - class _Options: - options = Options({"global-index.search-mode": "full"}) + class _Snapshots: + def get_latest_snapshot(self_inner): + return types.SimpleNamespace(next_row_id=10) - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() + table = _StubTable(fields=[self.id_field, self.embedding_field], + entries=[self.entries[0]]) + table.options = CoreOptions(Options({ + "global-index.search-mode": "full", + "vector-index.search-mode": "fast", + })) + table.snapshot_manager = lambda: _Snapshots() + self._scan_patch.stop() + self._travel_patch.stop() + _patch_snapshot(self, [self.entries[0]]) + self._travel_patch.stop() + + splits = ( + VectorSearchBuilderImpl(table) + .with_vector_column("embedding") + .with_query_vector([1.0, 0.0, 0.0, 0.0]) + .with_limit(3) + .new_vector_search_scan() + .scan() + .splits() + ) + + self.assertFalse(any(isinstance(split, RawVectorSearchSplit) + for split in splits)) + + def test_full_mode_scan_adds_raw_split_for_uncovered_scalar_filter(self): + from pypaimon.table.source.vector_search_split import RawVectorSearchSplit class _Snapshots: def get_latest_snapshot(self_inner): @@ -1387,7 +1421,10 @@ def get_latest_snapshot(self_inner): row_range_end=9) ], ) - table.options = _Options() + table.options = CoreOptions(Options({ + "scalar-index.search-mode": "full", + "vector-index.search-mode": "fast", + })) table.snapshot_manager = lambda: _Snapshots() self._scan_patch.stop() self._travel_patch.stop() @@ -1411,16 +1448,40 @@ def get_latest_snapshot(self_inner): self.assertEqual(1, len(raw)) self.assertEqual([Range(0, 9)], raw[0].row_ranges) - def test_scan_threads_builder_options_to_raw_split_index_type(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options + def test_raw_vector_pre_filter_uses_scalar_search_mode(self): + from pypaimon.table.source.vector_search_read import DataEvolutionVectorRead from pypaimon.table.source.vector_search_split import RawVectorSearchSplit - class _Options: - options = Options({"global-index.search-mode": "full"}) + predicate = Predicate(method="equal", index=0, field="id", literals=[5]) + scalar_file = self.entries[2].index_file + table = _StubTable(fields=[self.id_field, self.embedding_field], entries=[]) + table.options = CoreOptions(Options({ + "global-index.search-mode": "fast", + "scalar-index.search-mode": "detail", + })) + scanner = mock.MagicMock() + scanner.scan.return_value = GlobalIndexResult.create_empty() + scanner.unindexed_rows.return_value = GlobalIndexResult.create_empty() + reader = DataEvolutionVectorRead( + table, + limit=3, + vector_column=self.embedding_field, + query_vector=[1.0, 0.0, 0.0, 0.0], + filter_=predicate, + ) + + with mock.patch( + "pypaimon.globalindex.data_evolution_global_index_scanner." + "DataEvolutionGlobalIndexScanner.create", + return_value=scanner): + reader._raw_pre_filter([ + RawVectorSearchSplit([Range(0, 9)], [scalar_file])]) + + scanner.unindexed_rows.assert_called_once_with( + predicate, search_mode=GlobalIndexSearchMode.DETAIL) - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() + def test_scan_threads_builder_options_to_raw_split_index_type(self): + from pypaimon.table.source.vector_search_split import RawVectorSearchSplit class _Snapshots: def get_latest_snapshot(self_inner): @@ -1428,7 +1489,7 @@ def get_latest_snapshot(self_inner): table = _StubTable(fields=[self.id_field, self.embedding_field], entries=[]) - table.options = _Options() + table.options = CoreOptions(Options({"vector-index.search-mode": "full"})) table.snapshot_manager = lambda: _Snapshots() self._scan_patch.stop() self._travel_patch.stop() @@ -1865,21 +1926,10 @@ def close(self_inner): self.assertEqual([3], sorted(list(result.results()))) def test_scanner_reports_unindexed_rows_for_full_mode(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options from pypaimon.globalindex.data_evolution_global_index_scanner import ( DataEvolutionGlobalIndexScanner, ) - class _Options: - options = Options({"global-index.search-mode": "full"}) - - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() - - def global_index_thread_num(self_inner): - return 32 - class _Snapshots: def get_latest_snapshot(self_inner): return types.SimpleNamespace(next_row_id=10) @@ -1890,7 +1940,7 @@ def get_latest_snapshot(self_inner): file_name="id-0.index", row_range_start=0, row_range_end=4).index_file table = _StubTable(fields=[id_field, emb_field], entries=[]) - table.options = _Options() + table.options = CoreOptions(Options({"scalar-index.search-mode": "full"})) table.snapshot_manager = lambda: _Snapshots() scanner = DataEvolutionGlobalIndexScanner.create(table, index_files=[indexed]) diff --git a/paimon-python/pypaimon/write/table_update.py b/paimon-python/pypaimon/write/table_update.py index 90b784224526..8095c16c2ac4 100644 --- a/paimon-python/pypaimon/write/table_update.py +++ b/paimon-python/pypaimon/write/table_update.py @@ -273,9 +273,7 @@ def _matched_update_scan_table(self): return self.table dynamic_options = { - CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(): - GlobalIndexSearchMode.FULL.value, - CoreOptions.GLOBAL_INDEX_SCALAR_SEARCH_MODE.key(): + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(): GlobalIndexSearchMode.FULL.value, CoreOptions.SCAN_MODE.key(): StartupMode.DEFAULT.value, CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id), diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala index 5690a65d5fc7..79e09bc38029 100644 --- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala +++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala @@ -1047,14 +1047,13 @@ object MergeIntoPaimonDataEvolutionTable { val snapshotId = snapshot.id().toString if ( configuredSnapshotId.contains(snapshotId) && - fullSearchMode.equalsIgnoreCase( - table.options().get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key())) + table.coreOptions().scalarIndexSearchMode() == CoreOptions.GlobalIndexSearchMode.FULL ) { (v2Table, relation) } else { val dynamicOptions = new JHashMap[String, String]() timeTravelOptionKeys.foreach(dynamicOptions.put(_, null)) - dynamicOptions.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), fullSearchMode) + dynamicOptions.put(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), fullSearchMode) dynamicOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId) val scanTable = SparkTable.of(table.copy(dynamicOptions)) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala index f287d31959d0..479515ca4669 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala @@ -1046,14 +1046,13 @@ object MergeIntoPaimonDataEvolutionTable { val snapshotId = snapshot.id().toString if ( configuredSnapshotId.contains(snapshotId) && - fullSearchMode.equalsIgnoreCase( - table.options().get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key())) + table.coreOptions().scalarIndexSearchMode() == CoreOptions.GlobalIndexSearchMode.FULL ) { (v2Table, relation) } else { val dynamicOptions = new JHashMap[String, String]() timeTravelOptionKeys.foreach(dynamicOptions.put(_, null)) - dynamicOptions.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), fullSearchMode) + dynamicOptions.put(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), fullSearchMode) dynamicOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId) val scanTable = SparkTable.of(table.copy(dynamicOptions)) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala index 72258ddbec5d..f88993dba846 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala @@ -1035,6 +1035,7 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar |CREATE TABLE t (id INT, name STRING, b INT) TBLPROPERTIES ( | 'row-tracking.enabled' = 'true', | 'data-evolution.enabled' = 'true', + | 'scalar-index.search-mode' = 'fast', | 'btree-index.records-per-range' = '1000') |""".stripMargin) sql("INSERT INTO t VALUES (1, 'old', 10)") @@ -1062,6 +1063,7 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar |CREATE TABLE t (id INT, name STRING, b INT) TBLPROPERTIES ( | 'row-tracking.enabled' = 'true', | 'data-evolution.enabled' = 'true', + | 'scalar-index.search-mode' = 'fast', | 'btree-index.records-per-range' = '1000') |""".stripMargin) sql("INSERT INTO t VALUES (1, 'old', 10)") From c8eb633e255e70aca5da6076e8bb6804f6c7f8c0 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 12:42:55 +0800 Subject: [PATCH 08/10] [core][flink][python] Fix search mode handling --- .../table/source/DataEvolutionVectorScan.java | 31 ++++++++++------ .../table/source/VectorSearchBuilderTest.java | 35 ++++++++++++++++++ .../flink/action/DataEvolutionDelete.java | 3 +- .../action/DataEvolutionMergeIntoAction.java | 4 +- .../DataEvolutionMergeIntoActionITCase.java | 23 ++++++++++++ .../DeleteActionDataEvolutionITCase.java | 19 ++++++++++ .../table/source/vector_search_scan.py | 32 ++++++++++------ .../tests/vector_search_filter_test.py | 37 +++++++++++++++++++ 8 files changed, 158 insertions(+), 26 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java index 83a9a98debff..e9a26020d5b1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java @@ -18,6 +18,7 @@ package org.apache.paimon.table.source; +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; import org.apache.paimon.Snapshot; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; import org.apache.paimon.index.GlobalIndexMeta; @@ -144,27 +145,33 @@ public Plan scan() { splits.add(new IndexVectorSearchSplit(range.from, range.to, vectorFiles, scalarFiles)); } + GlobalIndexSearchMode vectorSearchMode = table.coreOptions().vectorIndexSearchMode(); List rawRowRanges = new DataEvolutionGlobalIndexCoverage( table, snapshot, partitionFilter, vectorIndexFiles, - table.coreOptions().vectorIndexSearchMode()) + vectorSearchMode) .unindexedRanges(vectorColumn.id()); if (filter != null) { + List scalarUnindexedRanges = + new DataEvolutionGlobalIndexCoverage( + table, + snapshot, + partitionFilter, + scalarIndexFiles(allIndexFiles), + table.coreOptions().scalarIndexSearchMode()) + .unindexedRanges(table.rowType(), filter); + if (vectorSearchMode == GlobalIndexSearchMode.FAST) { + scalarUnindexedRanges = + Range.and( + scalarUnindexedRanges, + Range.sortAndMergeOverlap( + new ArrayList<>(vectorByRange.keySet()), true)); + } rawRowRanges = - Range.sortAndMergeOverlap( - addAll( - rawRowRanges, - new DataEvolutionGlobalIndexCoverage( - table, - snapshot, - partitionFilter, - scalarIndexFiles(allIndexFiles), - table.coreOptions().scalarIndexSearchMode()) - .unindexedRanges(table.rowType(), filter)), - true); + Range.sortAndMergeOverlap(addAll(rawRowRanges, scalarUnindexedRanges), true); } if (!rawRowRanges.isEmpty()) { splits.add( 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 7c91bce60840..9f747807c03a 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 @@ -1102,6 +1102,41 @@ public void testPartialScalarPreFilterMustNotDropUnindexedScalarRows() throws Ex assertThat(ids).containsExactly(8); } + @Test + public void testFastVectorModeLimitsScalarFallbackToVectorCoverage() throws Exception { + catalog.createTable( + identifier("fast_vector_partial_coverage_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(), "fast") + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full") + .build(), + false); + FileStoreTable table = getTable(identifier("fast_vector_partial_coverage_table")); + + float[][] vectors = new float[10][]; + for (int i = 0; i < vectors.length; i++) { + vectors[i] = new float[] {Math.abs(i - 8), 0.0f}; + } + writeVectors(table, vectors); + + buildAndCommitVectorIndex(table, Arrays.copyOf(vectors, 5), new Range(0, 4)); + buildAndCommitBTreeIndex(table, new int[] {2, 3, 4}, new Range(2, 4)); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 5); + VectorScan.Plan plan = + table.newVectorSearchBuilder() + .withVector(new float[] {0.0f, 0.0f}) + .withLimit(1) + .withVectorColumn(VECTOR_FIELD_NAME) + .withFilter(idFilter) + .newVectorScan() + .scan(); + + assertThat(rawVectorSearchSplits(plan.splits())).hasSize(1); + assertThat(rawVectorSearchSplits(plan.splits()).get(0).rowRanges()) + .containsExactly(new Range(0, 1)); + } + @Test public void testFullModeFilterWithoutScalarIndexMustNotLetVectorIndexPolluteTopK() throws Exception { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java index 03ea04d144e5..fc0bd93697d7 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java @@ -168,11 +168,12 @@ TableResult runInternal() { String query = String.format( "SELECT `_ROW_ID` FROM `%s`.`%s`.`%s$row_tracking` " - + "/*+ OPTIONS('scan.snapshot-id'='%d') */ WHERE %s", + + "/*+ OPTIONS('scan.snapshot-id'='%d', '%s'='full') */ WHERE %s", action.catalogName, action.identifier.getDatabaseName(), action.identifier.getObjectName(), baseSnapshotId, + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), filter); LOG.info("Data-evolution delete source query: {}", query); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java index b3a57fe7329b..5c5cce749109 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java @@ -304,11 +304,13 @@ public Tuple2, RowType> buildSource() { // _ROW_ID is the first field of joined table. query = String.format( - "SELECT %s, %s FROM %s INNER JOIN %s AS RT ON %s", + "SELECT %s, %s FROM %s INNER JOIN %s " + + "/*+ OPTIONS('%s'='full') */ AS RT ON %s", "`RT`.`_ROW_ID` as `_ROW_ID`", String.join(",", project), escapedSourceName(), escapedRowTrackingTargetName(), + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), rewriteMergeCondition(mergeCondition)); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java index 2deff25d13b9..bcc25cc1bf3a 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java @@ -621,6 +621,29 @@ public void testUpdateAction() throws Exception { assertFalse(indexFileExists("T")); } + @Test + public void testMergeUnindexedRowWithFastSearchMode() throws Exception { + executeSQL( + "CALL sys.create_global_index(`table` => 'default.T', " + + "index_column => 'name', index_type => 'btree')", + false, + true); + insertInto("T", "(21, 'new', 2.1, '01-22')"); + executeSQL("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')", false, true); + + builder(warehouse, database, "T") + .withMergeCondition("T.name = 'new' AND S.id = 21") + .withMatchedUpdateSet("T.value = S.`value`") + .withSourceTable("S") + .withSinkParallelism(1) + .build() + .run(); + + testBatchRead( + "SELECT id, name, `value` FROM T WHERE id = 21", + Collections.singletonList(changelogRow("+I", 21, "new", 102.1))); + } + private boolean indexFileExists(String tableName) throws Exception { FileStoreTable table = getFileStoreTable(tableName); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java index 08d3e58b118d..f220f848089b 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java @@ -118,6 +118,25 @@ public void testNoMatchedRowsDoesNotCreateSnapshot() throws Exception { assertThat(deletionVectorCardinality(table)).isZero(); } + @Test + public void testDeleteUnindexedRowWithFastSearchMode() throws Exception { + createDataEvolutionTable(TABLE, false, true); + insertInto(TABLE, "(1, 'old', 'A')"); + executeSQL( + "CALL sys.create_global_index(`table` => 'default.T', " + + "index_column => 'name', index_type => 'btree')", + false, + true); + insertInto(TABLE, "(2, 'new', 'A')"); + executeSQL("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')", false, true); + + action(TABLE, "name = 'new'", 1).run(); + + testBatchRead( + "SELECT id, name, dt FROM T ORDER BY id", + Collections.singletonList(changelogRow("+I", 1, "old", "A"))); + } + @Test public void testDeleteWithBoundedExternalCandidates() throws Exception { createDataEvolutionTable(TABLE, false, true); diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py b/paimon-python/pypaimon/table/source/vector_search_scan.py index 1cbdb19c1492..7e002064dbd7 100644 --- a/paimon-python/pypaimon/table/source/vector_search_scan.py +++ b/paimon-python/pypaimon/table/source/vector_search_scan.py @@ -20,6 +20,7 @@ from abc import ABC, abstractmethod from collections import defaultdict +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.globalindex.data_evolution_global_index_coverage import DataEvolutionGlobalIndexCoverage from pypaimon.table.source.vector_search_split import ( IndexVectorSearchSplit, @@ -163,6 +164,7 @@ def index_file_filter(entry): ) ) + vector_search_mode = self._table.options.vector_index_search_mode() raw_row_ranges = DataEvolutionGlobalIndexCoverage( self._table, snapshot, @@ -170,7 +172,7 @@ def index_file_filter(entry): vector_index_files, ).unindexed_ranges( vector_column.id, - search_mode=self._table.options.vector_index_search_mode(), + search_mode=vector_search_mode, ) scalar_index_files = [ f for f in all_index_files @@ -178,18 +180,24 @@ def index_file_filter(entry): and f.global_index_meta.index_field_id != vector_column.id ] if self._filter is not None: + scalar_unindexed_ranges = DataEvolutionGlobalIndexCoverage( + self._table, + snapshot, + partition_filter, + scalar_index_files, + ).unindexed_ranges( + self._table.fields, + self._filter, + search_mode=self._table.options.scalar_index_search_mode(), + ) + if vector_search_mode == GlobalIndexSearchMode.FAST: + scalar_unindexed_ranges = Range.and_( + scalar_unindexed_ranges, + Range.sort_and_merge_overlap( + list(vector_by_range.keys()), True), + ) raw_row_ranges = Range.sort_and_merge_overlap( - raw_row_ranges - + DataEvolutionGlobalIndexCoverage( - self._table, - snapshot, - partition_filter, - scalar_index_files, - ).unindexed_ranges( - self._table.fields, - self._filter, - search_mode=self._table.options.scalar_index_search_mode(), - ), + raw_row_ranges + scalar_unindexed_ranges, True, ) if raw_row_ranges: diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 683d64b1de9a..d3c75613c502 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -1402,6 +1402,43 @@ def get_latest_snapshot(self_inner): self.assertFalse(any(isinstance(split, RawVectorSearchSplit) for split in splits)) + def test_fast_vector_mode_limits_scalar_fallback_to_vector_coverage(self): + from pypaimon.table.source.vector_search_split import RawVectorSearchSplit + + entries = [ + _entry(None, field_id=1, index_type="lumina-vector-ann", + file_name="vec.index", row_range_start=0, + row_range_end=4), + _entry(None, field_id=0, index_type="btree", + file_name="id.index", row_range_start=2, + row_range_end=4), + ] + table = _StubTable(fields=[self.id_field, self.embedding_field], + entries=entries) + table.options = CoreOptions(Options({ + "vector-index.search-mode": "fast", + "scalar-index.search-mode": "full", + })) + _patch_snapshot(self, entries, types.SimpleNamespace(next_row_id=10)) + filter_pred = Predicate(method="greaterOrEqual", index=0, field="id", + literals=[5]) + + splits = ( + VectorSearchBuilderImpl(table) + .with_vector_column("embedding") + .with_query_vector([1.0, 0.0, 0.0, 0.0]) + .with_limit(3) + .with_filter(filter_pred) + .new_vector_search_scan() + .scan() + .splits() + ) + + raw = [split for split in splits + if isinstance(split, RawVectorSearchSplit)] + self.assertEqual(1, len(raw)) + self.assertEqual([Range(0, 1)], raw[0].row_ranges) + def test_full_mode_scan_adds_raw_split_for_uncovered_scalar_filter(self): from pypaimon.table.source.vector_search_split import RawVectorSearchSplit From 26828f735d1dc7500d4088e0083c4535fd31aad7 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 13:57:35 +0800 Subject: [PATCH 09/10] [python] Fix partition-only vector filters --- .../table/source/vector_search_builder.py | 51 ++++++----- .../tests/vector_search_filter_test.py | 88 ++++++++++++++++++- 2 files changed, 112 insertions(+), 27 deletions(-) diff --git a/paimon-python/pypaimon/table/source/vector_search_builder.py b/paimon-python/pypaimon/table/source/vector_search_builder.py index 5675f94c10b5..53322a8f81da 100644 --- a/paimon-python/pypaimon/table/source/vector_search_builder.py +++ b/paimon-python/pypaimon/table/source/vector_search_builder.py @@ -132,21 +132,19 @@ def with_filter(self, predicate): # type: (Predicate) -> VectorSearchBuilder if predicate is None: return self - if self._filter is None: - self._filter = predicate - else: - self._filter = PredicateBuilder.and_predicates([self._filter, predicate]) - # split out the partition-only conjuncts and store them as _partition_filter for - # manifest pruning. Non-partition conjuncts remain in self._filter; - # the silent drop of non-partition conjuncts *in the extracted copy* - # is intentional — nothing is lost overall. - extracted = self._extract_partition_only_conjuncts(predicate) - if extracted is not None: + partition_filter, data_filter = self._split_partition_filter(predicate) + if partition_filter is not None: if self._partition_filter is None: - self._partition_filter = extracted + self._partition_filter = partition_filter else: self._partition_filter = PredicateBuilder.and_predicates( - [self._partition_filter, extracted]) + [self._partition_filter, partition_filter]) + if data_filter is not None: + if self._filter is None: + self._filter = data_filter + else: + self._filter = PredicateBuilder.and_predicates( + [self._filter, data_filter]) return self def with_partition_filter(self, partition_filter): @@ -175,25 +173,26 @@ def with_partition_filter(self, partition_filter): ) return self - def _extract_partition_only_conjuncts(self, predicate): - """AND-split ``predicate``, keep conjuncts that reference ONLY - partition keys, and rebuild their leaf indices against the - partition-only row by field name (so the caller's PredicateBuilder - convention — full-row or partition-row — doesn't matter). - """ + def _split_partition_filter(self, predicate): + """Split partition-only and data conjuncts.""" partition_keys = list(self._table.partition_keys or []) if not partition_keys: - return None + return None, predicate from pypaimon.read.push_down_utils import _split_and, _get_all_fields partition_key_set = set(partition_keys) pk_to_idx = {name: idx for idx, name in enumerate(partition_keys)} - kept = [p for p in _split_and(predicate) - if _get_all_fields(p).issubset(partition_key_set)] - if not kept: - return None - rebuilt = [self._rebuild_leaf_indices_by_name(p, pk_to_idx) - for p in kept] - return PredicateBuilder.and_predicates(rebuilt) + partition_parts = [] + data_parts = [] + for part in _split_and(predicate): + if _get_all_fields(part).issubset(partition_key_set): + partition_parts.append( + self._rebuild_leaf_indices_by_name(part, pk_to_idx)) + else: + data_parts.append(part) + return ( + PredicateBuilder.and_predicates(partition_parts), + PredicateBuilder.and_predicates(data_parts), + ) @classmethod def _rebuild_leaf_indices_by_name(cls, predicate, pk_to_idx): diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index d3c75613c502..8fc16d9bd595 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -2075,6 +2075,8 @@ def test_with_filter_auto_splits_and_prunes_wrong_partition(self): """A normal full-row predicate ``pt == 2`` must (a) be auto-split into _partition_filter with indices re-based to the partition-only row, and (b) drop pt=1 entries during manifest pruning.""" + _patch_snapshot( + self, self.entries, types.SimpleNamespace(next_row_id=10)) pb = PredicateBuilder(self.table.fields) builder = (VectorSearchBuilderImpl(self.table) .with_vector_column("embedding") @@ -2082,6 +2084,7 @@ def test_with_filter_auto_splits_and_prunes_wrong_partition(self): .with_limit(3) .with_filter(pb.equal("pt", 2))) + self.assertIsNone(builder._filter) # Partition filter's leaf index points into the partition-only row. self.assertEqual("pt", builder._partition_filter.field) self.assertEqual(0, builder._partition_filter.index) @@ -2091,8 +2094,91 @@ def test_with_filter_auto_splits_and_prunes_wrong_partition(self): self.assertEqual(["vec-pt2.index"], [f.file_name for f in splits[0].vector_index_files]) + def test_partition_only_filter_is_not_scalar_prefilter(self): + from pypaimon.globalindex.vector_search_result import ( + DictBasedScoredIndexResult, + ) + from pypaimon.table.source.batch_vector_search_builder import ( + BatchVectorSearchBuilderImpl, + ) + + self.table.options = CoreOptions(Options({ + "scalar-index.search-mode": "fast", + })) + _patch_snapshot( + self, self.entries, types.SimpleNamespace(next_row_id=10)) + searches = [] + + def _fake_create(index_type, file_io, index_path, + index_io_meta_list, options=None): + class _FakeReader: + def visit_vector_search(self_inner, search): + searches.append(search) + scores = {} if (search.include_row_ids is not None and + search.include_row_ids.is_empty()) else {0: 1.0} + return _completed_future( + DictBasedScoredIndexResult(scores)) + + def visit_batch_vector_search(self_inner, search): + searches.append(search) + scores = {} if (search.include_row_ids is not None and + search.include_row_ids.is_empty()) else {0: 1.0} + return _completed_future([ + DictBasedScoredIndexResult(scores) + for _ in range(search.vector_count) + ]) + + def close(self_inner): + pass + + return _FakeReader() + + partition_filter = PredicateBuilder(self.table.fields).equal("pt", 2) + with mock.patch( + "pypaimon.table.source.vector_search_read._create_vector_reader", + side_effect=_fake_create): + direct = ( + VectorSearchBuilderImpl(self.table) + .with_vector_column("embedding") + .with_query_vector([1.0, 0.0, 0.0, 0.0]) + .with_limit(3) + .with_filter(partition_filter) + .execute_local() + ) + batch = ( + BatchVectorSearchBuilderImpl(self.table) + .with_vector_column("embedding") + .with_query_vectors([[1.0, 0.0, 0.0, 0.0]]) + .with_limit(3) + .with_filter(partition_filter) + .execute_batch_local() + ) + + self.assertEqual([5], list(direct.results())) + self.assertEqual([[5]], [list(result.results()) for result in batch]) + self.assertTrue(all(search.include_row_ids is None for search in searches)) + + def test_with_filter_keeps_only_data_conjuncts_as_scalar_filter(self): + pb = PredicateBuilder(self.table.fields) + builder = VectorSearchBuilderImpl(self.table).with_filter( + PredicateBuilder.and_predicates([ + pb.equal("pt", 2), + pb.greater_or_equal("id", 5), + ])) + + self.assertEqual("pt", builder._partition_filter.field) + self.assertEqual("id", builder._filter.field) + + cross_field_or = PredicateBuilder.or_predicates([ + pb.equal("pt", 2), + pb.greater_or_equal("id", 5), + ]) + builder = VectorSearchBuilderImpl(self.table).with_filter(cross_field_or) + self.assertIsNone(builder._partition_filter) + self.assertIs(cross_field_or, builder._filter) + def test_with_partition_filter_rejects_non_partition_field(self): - """Non-partition conjuncts would be silently dropped by the extractor, + """Non-partition conjuncts would be silently dropped by the splitter, producing wrong results; the API must refuse them up front.""" pb = PredicateBuilder(self.table.fields) builder = VectorSearchBuilderImpl(self.table) From b8df671abde8067864e2fb09ae66c51ffa8a3ec2 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 26 Jul 2026 14:11:04 +0800 Subject: [PATCH 10/10] [python] Fix vector partition filter accumulation --- .../table/source/vector_search_builder.py | 25 +++++---- .../tests/vector_search_filter_test.py | 55 +++++++++++++++++++ 2 files changed, 70 insertions(+), 10 deletions(-) diff --git a/paimon-python/pypaimon/table/source/vector_search_builder.py b/paimon-python/pypaimon/table/source/vector_search_builder.py index 53322a8f81da..af823e58c2fb 100644 --- a/paimon-python/pypaimon/table/source/vector_search_builder.py +++ b/paimon-python/pypaimon/table/source/vector_search_builder.py @@ -134,11 +134,7 @@ def with_filter(self, predicate): return self partition_filter, data_filter = self._split_partition_filter(predicate) if partition_filter is not None: - if self._partition_filter is None: - self._partition_filter = partition_filter - else: - self._partition_filter = PredicateBuilder.and_predicates( - [self._partition_filter, partition_filter]) + self._add_partition_filter(partition_filter) if data_filter is not None: if self._filter is None: self._filter = data_filter @@ -150,7 +146,6 @@ def with_filter(self, predicate): def with_partition_filter(self, partition_filter): # type: (Predicate) -> VectorSearchBuilder if partition_filter is None: - self._partition_filter = None return self # Strict: every referenced field must be a partition key, otherwise a # non-partition conjunct would be silently dropped (with_filter has @@ -167,10 +162,11 @@ def with_partition_filter(self, partition_filter): "Partition filter must reference only partition keys " "(%s); got non-partition field(s): %s" % (partition_keys, sorted(extras))) - self._partition_filter = self._rebuild_leaf_indices_by_name( - partition_filter, - {name: idx for idx, name in enumerate(partition_keys)}, - ) + self._add_partition_filter( + self._rebuild_leaf_indices_by_name( + partition_filter, + {name: idx for idx, name in enumerate(partition_keys)}, + )) return self def _split_partition_filter(self, predicate): @@ -194,6 +190,15 @@ def _split_partition_filter(self, predicate): PredicateBuilder.and_predicates(data_parts), ) + def _add_partition_filter(self, partition_filter): + if partition_filter is None: + return + if self._partition_filter is None: + self._partition_filter = partition_filter + else: + self._partition_filter = PredicateBuilder.and_predicates( + [self._partition_filter, partition_filter]) + @classmethod def _rebuild_leaf_indices_by_name(cls, predicate, pk_to_idx): """Return a copy of ``predicate`` with every leaf's ``index`` set to diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 8fc16d9bd595..d677ba2a81dd 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -2177,6 +2177,61 @@ def test_with_filter_keeps_only_data_conjuncts_as_scalar_filter(self): self.assertIsNone(builder._partition_filter) self.assertIs(cross_field_or, builder._filter) + def test_partition_filters_accumulate_in_any_order(self): + from pypaimon.table.source.batch_vector_search_builder import ( + BatchVectorSearchBuilderImpl, + ) + + _patch_snapshot( + self, self.entries, types.SimpleNamespace(next_row_id=10)) + pb = PredicateBuilder(self.table.fields) + auto_filter = pb.equal("pt", 2) + explicit_filter = pb.equal("pt", 1) + + for builder_class in ( + VectorSearchBuilderImpl, BatchVectorSearchBuilderImpl): + builders = [ + builder_class(self.table) + .with_vector_column("embedding") + .with_filter(auto_filter) + .with_partition_filter(explicit_filter), + builder_class(self.table) + .with_vector_column("embedding") + .with_partition_filter(explicit_filter) + .with_filter(auto_filter), + ] + for builder in builders: + self.assertEqual( + [], builder.new_vector_search_scan().scan().splits()) + + def test_none_partition_filter_is_noop(self): + from pypaimon.table.source.batch_vector_search_builder import ( + BatchVectorSearchBuilderImpl, + ) + + _patch_snapshot( + self, self.entries, types.SimpleNamespace(next_row_id=10)) + partition_filter = PredicateBuilder(self.table.fields).equal("pt", 2) + + for builder_class in ( + VectorSearchBuilderImpl, BatchVectorSearchBuilderImpl): + builders = [ + builder_class(self.table) + .with_vector_column("embedding") + .with_partition_filter(partition_filter) + .with_partition_filter(None), + builder_class(self.table) + .with_vector_column("embedding") + .with_partition_filter(None) + .with_partition_filter(partition_filter), + ] + for builder in builders: + splits = builder.new_vector_search_scan().scan().splits() + self.assertEqual(1, len(splits)) + self.assertEqual( + ["vec-pt2.index"], + [f.file_name for f in splits[0].vector_index_files]) + def test_with_partition_filter_rejects_non_partition_field(self): """Non-partition conjuncts would be silently dropped by the splitter, producing wrong results; the API must refuse them up front."""