From 7e3084b6d37fbce3fd37ea4b2825ca56aaee2a5e Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 15 Oct 2026 23:47:49 +0800 Subject: [PATCH 1/4] [python] Use native index builds on deletion-vector tables --- .../pypaimon/globalindex/build_plan.py | 10 +-- .../globalindex/native_index_build.py | 2 +- .../pypaimon/tests/global_index_build_test.py | 37 ++++++++++ .../tests/native_sorted_index_build_test.py | 50 ++++++++++++++ .../tests/native_vector_index_build_test.py | 67 +++++++++++++++++++ 5 files changed, 157 insertions(+), 9 deletions(-) diff --git a/paimon-python/pypaimon/globalindex/build_plan.py b/paimon-python/pypaimon/globalindex/build_plan.py index f4e33dd2375b..5f30ab5ad0ed 100644 --- a/paimon-python/pypaimon/globalindex/build_plan.py +++ b/paimon-python/pypaimon/globalindex/build_plan.py @@ -318,17 +318,11 @@ def split_one_by_contiguous_row_range(split): def copy_split_with_files(split, files): - data_deletion_files = None - if getattr(split, "data_deletion_files", None) is not None: - index_by_file = {id(file): i for i, file in enumerate(split.files)} - data_deletion_files = [ - split.data_deletion_files[index_by_file[id(file)]] - for file in files - ] + # Like Java's splitByContiguousRowRange, build over physical row IDs. + # Query readers apply the deletion vectors from their own snapshot. return DataSplit( files=list(files), partition=split.partition, bucket=split.bucket, raw_convertible=getattr(split, "raw_convertible", False), - data_deletion_files=data_deletion_files, ) diff --git a/paimon-python/pypaimon/globalindex/native_index_build.py b/paimon-python/pypaimon/globalindex/native_index_build.py index ac45480095f7..914193640722 100644 --- a/paimon-python/pypaimon/globalindex/native_index_build.py +++ b/paimon-python/pypaimon/globalindex/native_index_build.py @@ -34,7 +34,7 @@ def build_native_global_index(builder, partition_filter): or not table.options.native_write_enabled() or not table.options.data_evolution_enabled() or not table.options.global_index_enabled() - or table.options.deletion_vectors_enabled() + or (table.options.deletion_vectors_enabled() and builder._index_type == 'full-text') or table.options.query_auth_enabled or table.is_primary_key_table or table.options.file_format() != 'parquet' diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index 687f45fcd9b7..8a006a7359b4 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/paimon-python/pypaimon/tests/global_index_build_test.py @@ -226,6 +226,33 @@ class GlobalIndexBuildTest( 'file.format': 'parquet', } + def test_sorted_build_indexes_deleted_physical_rows_like_java(self): + for kind in (BTREE_IDENTIFIER, BITMAP_IDENTIFIER): + with self.subTest(kind=kind): + options = dict(self.table_options) + options.update({'deletion-vectors.enabled': 'true', 'write.native.enabled': 'false'}) + table = self._create_table(options=options) + self._write_arrow(table, pa.table({ + 'id': [1, 2, 3], 'name': ['a', 'b', 'a'], + 'age': [10, 20, 30], 'city': ['x', 'y', 'z'], + }, schema=self.pa_schema)) + builder = table.new_batch_write_builder() + commit = builder.new_commit() + try: + commit.commit(builder.new_update().delete_by_row_id([0, 2])) + finally: + commit.close() + # Both deleted keys and the live key belong to index coverage. + self.assertEqual(1, table.create_global_index('name', index_type=kind)) + entries = [e for e in IndexFileHandler(table).scan(None) if e.index_file.index_type == kind] + self.assertEqual(3, sum(e.index_file.row_count for e in entries)) + self.assertEqual([(0, 2)], [(e.index_file.global_index_meta.row_range_start, + e.index_file.global_index_meta.row_range_end) for e in entries]) + read = table.new_read_builder() + rows = read.new_read().to_arrow(read.new_scan().plan().splits()) + self.assertEqual([2], rows.column('id').to_pylist()) + self.assertEqual(0, table.create_global_index('name', index_type=kind)) + def test_create_btree_global_index_from_python(self): table = self._create_table() self._write_arrow(table, pa.table( @@ -1252,6 +1279,16 @@ def test_split_by_contiguous_row_range_matches_java_builder(self): [[file.file_name for file in s.files] for s in splits], ) + def test_sorted_build_splits_omit_deletion_vectors_like_java(self): + split = _FakeSplit([_FakeFile('a', 0, 2), _FakeFile('b', 4, 2)]) + deletion_files = [object(), object()] + split.data_deletion_files = deletion_files + splits = _split_one_by_contiguous_row_range(split) + self.assertEqual([['a'], ['b']], [[file.file_name for file in s.files] for s in splits]) + self.assertTrue(all(s.data_deletion_files is None for s in splits)) + # Planning a physical build must not alter the original query split. + self.assertIs(split.data_deletion_files, deletion_files) + def test_split_by_global_index_shard_matches_java_default_builder(self): split = _FakeSplit([ _FakeFile('a', 0, 3), diff --git a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py index b2849ee60e21..76a52899119d 100644 --- a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py @@ -109,6 +109,56 @@ def _build(table, kind, column='name', **kwargs): return GlobalIndexBuilder(table, column, index_type=kind, **kwargs).build() +def _delete(table, ids): + builder = table.new_batch_write_builder() + _commit(table, builder.new_update().delete_by_row_id(ids)) + + +@pytest.mark.parametrize('kind', ['btree', 'bitmap']) +@pytest.mark.parametrize('deleted', [[0, 2], [0, 1, 2, 3]]) +@pytest.mark.parametrize('native_commit', [False, True]) +def test_deletion_vectors_do_not_change_physical_index_coverage(rest_catalog, kind, deleted, native_commit): + catalog, _ = rest_catalog + table = _create(catalog, options={'deletion-vectors.enabled': 'true', + 'commit.native.enabled': str(native_commit).lower()}) + rows = [{'id': i + 1, 'name': 'b' if i % 2 else 'a', 'pt': 0} for i in range(4)] + _append(table, rows) + _delete(table, deleted) + before = table.snapshot_manager().get_latest_snapshot().id + messages = _build(table, kind) + files = _message_files(messages) + assert sum(file.row_count for file in files) == 4 + assert sorted((file.global_index_meta.row_range_start, + file.global_index_meta.row_range_end) for file in files) == [(0, 1), (2, 3)] + classic = GlobalIndexBuilder(table.copy({'write.native.enabled': 'false'}), 'name', index_type=kind).build() + assert sum(file.row_count for file in _message_files(classic)) == 4 + assert table.snapshot_manager().get_latest_snapshot().id == before + _commit(table, messages) + predicate = table.new_read_builder().new_predicate_builder().equal('name', 'b') + + def assert_readers(expected, source=table): + for native in [False, True]: + reader = source.copy({'read.native.enabled': str(native).lower(), + 'scan.native-plan.enabled': str(native).lower()}) + assert _ids(reader, predicate) == expected + + expected = [2, 4] if len(deleted) == 2 else [] + assert_readers(expected) + indexed_snapshot = table.snapshot_manager().get_latest_snapshot().id + assert _build(table, kind) == [] + _append(table, [{'id': 5, 'name': 'b', 'pt': 0}, {'id': 6, 'name': None, 'pt': 0}]) + messages = _build(table, kind) + files = _message_files(messages) + assert sum(file.row_count for file in files) == 2 + assert all(file.global_index_meta.row_range_start == 4 for file in files) + # A concurrent deletion changes live rows, not the staged index's source values. + _delete(table, [1, 4]) + _commit(table, messages) + assert_readers([4] if len(deleted) == 2 else []) + assert_readers(expected, table.copy({'scan.snapshot-id': str(indexed_snapshot)})) + assert _build(table, kind) == [] + + @pytest.mark.parametrize('kind', ['btree', 'bitmap']) @pytest.mark.parametrize('partitioned', [False, True]) @pytest.mark.parametrize('native_commit', [False, True]) diff --git a/paimon-python/pypaimon/tests/native_vector_index_build_test.py b/paimon-python/pypaimon/tests/native_vector_index_build_test.py index 4e6531150d7d..9a7216b366ad 100644 --- a/paimon-python/pypaimon/tests/native_vector_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_vector_index_build_test.py @@ -81,6 +81,73 @@ def _assert_ids(table, expected, kind): assert sorted(result.row_ids()) == expected +@pytest.mark.parametrize('kind', VINDEX_IDENTIFIERS) +@pytest.mark.parametrize('granule', [False, True]) +@pytest.mark.parametrize('all_deleted', [False, True]) +def test_deletion_vectors_preserve_training_rows_and_live_search(rest_catalog, kind, granule, all_deleted): + catalog, _ = rest_catalog + vectors = [[1., 0.], [0., 1.], None, [1., 1.], [2., 2.]] + table = _table(catalog, vectors, {'deletion-vectors.enabled': 'true'}) + indexes._delete(table, list(range(5)) if all_deleted else [0, 2, 3]) + options = generic._options(kind) + if kind != 'diskann': + options['vindex.build.granule.enabled'] = str(granule).lower() + before = table.snapshot_manager().get_latest_snapshot().id + messages = _build(table, kind, options) + files = indexes._message_files(messages) + assert len(files) == 1 + assert files[0].row_count == 5 + assert (files[0].global_index_meta.row_range_start, files[0].global_index_meta.row_range_end) == (0, 4) + # Java indexes the stable physical rows; deleted vectors still participate in training. + assert _metadata(table, files[0]).total_vectors == 4 + classic = GlobalIndexBuilder(table.copy({'write.native.enabled': 'false'}), 'embedding', + index_type=kind, options=options).build() + assert _bytes(table, files[0]) == _bytes(table, indexes._message_files(classic)[0]) + assert table.snapshot_manager().get_latest_snapshot().id == before + indexes._commit(table, messages) + expected = [] if all_deleted else [1, 4] + _assert_ids(table, expected, kind) + indexed_snapshot = table.snapshot_manager().get_latest_snapshot().id + assert _build(table, kind, options) == [] + + indexes._append(table, _rows([[3., 3.], [4., 4.]], start=5), SCHEMA) + messages = _build(table, kind, options) + file = indexes._message_files(messages)[0] + assert file.row_count == 2 + assert (file.global_index_meta.row_range_start, file.global_index_meta.row_range_end) == (5, 6) + indexes._delete(table, [1, 5]) + indexes._commit(table, messages) + _assert_ids(table, [6] if all_deleted else [4, 6], kind) + _assert_ids(table.copy({'scan.snapshot-id': str(indexed_snapshot)}), expected, kind) + assert _build(table, kind, options) == [] + + +@pytest.mark.parametrize('kind', VINDEX_IDENTIFIERS) +def test_native_search_filters_live_null_vectors_with_deletion_vectors(rest_catalog, kind): + catalog, _ = rest_catalog + table = _table(catalog, [[1., 0.], [0., 1.], None, [1., 1.], [2., 2.]], + {'deletion-vectors.enabled': 'true'}) + indexes._delete(table, [0, 3]) + indexes._commit(table, _build(table, kind, generic._options(kind))) + options = {'diskann.l_search': '64'} if kind == 'diskann' else {'ivf.nprobe': '64'} + result = (_native_table(table).new_vector_search_builder().with_vector_column('embedding') + .with_query_vector([1., 1.]).with_limit(64).with_options(options).execute_local()) + assert sorted(result.row_ids()) == [1, 4] + + +@pytest.mark.parametrize('kind', VINDEX_IDENTIFIERS) +def test_deleted_vectors_still_validate_physical_dimensions(rest_catalog, kind): + catalog, _ = rest_catalog + table = _table(catalog, [[1., 0., 3.], [0., 1.]], {'deletion-vectors.enabled': 'true'}) + indexes._delete(table, [0]) + before = table.snapshot_manager().get_latest_snapshot().id + deletion_files = [file.file_name for file in indexes._files(table)] + with pytest.raises(Exception, match='dimension'): + _build(table, kind, generic._options(kind)) + assert table.snapshot_manager().get_latest_snapshot().id == before + assert [file.file_name for file in indexes._files(table)] == deletion_files + + @pytest.mark.parametrize('kind', VINDEX_IDENTIFIERS) @pytest.mark.parametrize('granule', [False, True]) def test_nullable_vectors_preserve_row_ids_and_logical_cardinality(rest_catalog, kind, granule): From 3054672bff2fb4a0f7757a92350c15b244131c70 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 00:44:16 +0800 Subject: [PATCH 2/4] [python] Complete native index lifecycle and explicit search plans --- .../globalindex/create_global_index.py | 2 +- .../data_evolution_global_index_scanner.py | 10 +- .../pypaimon/globalindex/drop_global_index.py | 27 +- .../globalindex/native_index_build.py | 2 - .../vindex_vector_global_index_reader.py | 57 ++- .../pypaimon/ray/batch_vector_search.py | 3 +- .../pypaimon/ray/full_text_search.py | 3 +- paimon-python/pypaimon/ray/vector_search.py | 3 +- .../source/batch_vector_search_builder.py | 6 +- .../table/source/full_text_search_builder.py | 14 +- .../table/source/native_full_text_search.py | 21 +- .../table/source/native_hybrid_search.py | 1 - .../table/source/native_search_plan.py | 254 +++++++++++ .../table/source/native_vector_search.py | 14 +- .../table/source/vector_search_builder.py | 10 +- .../table/source/vector_search_read.py | 3 +- .../pypaimon/tests/global_index_build_test.py | 14 + .../tests/global_index_family_parity_test.py | 96 ++++ .../tests/native_append_index_search_test.py | 137 ++++++ .../tests/native_generic_index_build_test.py | 25 + .../tests/native_global_index_drop_test.py | 164 +++++++ .../pypaimon/tests/native_search_plan_test.py | 430 ++++++++++++++++++ .../tests/native_sorted_index_build_test.py | 2 +- .../tests/native_vector_index_build_test.py | 3 + .../tests/native_vindex_reader_test.py | 215 +++++++++ .../tests/vector_metric_consistency_test.py | 16 +- .../pypaimon/tests/vindex_input_test.py | 4 +- .../tests/vindex_vector_index_test.py | 16 +- 28 files changed, 1504 insertions(+), 48 deletions(-) create mode 100644 paimon-python/pypaimon/table/source/native_search_plan.py create mode 100644 paimon-python/pypaimon/tests/global_index_family_parity_test.py create mode 100644 paimon-python/pypaimon/tests/native_append_index_search_test.py create mode 100644 paimon-python/pypaimon/tests/native_global_index_drop_test.py create mode 100644 paimon-python/pypaimon/tests/native_search_plan_test.py create mode 100644 paimon-python/pypaimon/tests/native_vindex_reader_test.py diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py b/paimon-python/pypaimon/globalindex/create_global_index.py index 98da19c5a99d..f71ff18cce95 100644 --- a/paimon-python/pypaimon/globalindex/create_global_index.py +++ b/paimon-python/pypaimon/globalindex/create_global_index.py @@ -518,7 +518,7 @@ def _validate_unaware_bucket_table(self) -> None: ) def _validate_generic_index_table(self) -> None: - if (self._core_options.deletion_vectors_enabled() + if (self._table.options.deletion_vectors_enabled() and not (self._index_type in VINDEX_IDENTIFIERS and self._table.options.data_evolution_enabled())): raise ValueError( 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 1174df8919ea..c5efd1535c67 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -110,6 +110,8 @@ def _create_evaluator(self, fields, file_io, index_path, index_files): options = self._options def readers_function(field: DataField) -> Collection[GlobalIndexReader]: + from pypaimon.globalindex.union_global_index_reader import UnionGlobalIndexReader + groups = [] group = index_metas.get(field.id) if group is not None: @@ -125,8 +127,10 @@ def readers_function(field: DataField) -> Collection[GlobalIndexReader]: if not groups: return [] if len(groups) == 1: - return _create_readers( - file_io, index_path, groups[0].metas, field, executor, options) + # Families may cover different row ranges. Like Java, union + # their matches before the evaluator combines field predicates. + return [UnionGlobalIndexReader(_create_readers( + file_io, index_path, groups[0].metas, field, executor, options))] union_coverage = Range.sort_and_merge_overlap( [ range_key @@ -142,7 +146,7 @@ def readers_function(field: DataField) -> Collection[GlobalIndexReader]: _create_readers( file_io, index_path, group.metas, field, executor, options, pad_ranges=pad_ranges)) - return readers + return [UnionGlobalIndexReader(readers)] return GlobalIndexEvaluator(fields, readers_function) diff --git a/paimon-python/pypaimon/globalindex/drop_global_index.py b/paimon-python/pypaimon/globalindex/drop_global_index.py index 6be45499e8b3..b36639bed7ea 100644 --- a/paimon-python/pypaimon/globalindex/drop_global_index.py +++ b/paimon-python/pypaimon/globalindex/drop_global_index.py @@ -89,6 +89,11 @@ def __init__( ) def build(self) -> List[CommitMessage]: + partition_filter = self._resolve_partition_filter() + messages = self._native_build(partition_filter) + if messages is not None: + return messages + snapshot = self._table.snapshot_manager().get_latest_snapshot() if snapshot is None: return [] @@ -96,7 +101,6 @@ def build(self) -> List[CommitMessage]: target_field_ids = [ self._table.field_dict[column].id for column in self._index_columns ] - partition_filter = self._resolve_partition_filter() def should_delete(entry: IndexManifestEntry) -> bool: index_file = entry.index_file @@ -133,6 +137,27 @@ def should_delete(entry: IndexManifestEntry) -> bool: for partition, deletes in by_partition.items() ] + def _native_build(self, partition_filter): + from pypaimon.read.native_plan import _native_table, _predicate_to_native, native_method_available + from pypaimon.write.native_commit import _rest_catalog_supported, from_native_commit_messages + + table = self._table + if (not table.options.native_write_enabled() + or table.options.query_auth_enabled + or not native_method_available('Table', 'new_global_index_drop_builder') + or not _rest_catalog_supported(table)): + return None + from pypaimon_rust.datafusion import GlobalIndexDropBuilder + if not GlobalIndexDropBuilder.supports_index_type(self._index_type): + return None + native = _native_table(table).new_global_index_drop_builder() + native.with_index_columns(self._index_columns).with_index_type(self._index_type) + if partition_filter is not None: + native.with_partition_filter(_predicate_to_native(partition_filter)) + # Preparing deletion messages is read-only. Python retains explicit + # commit and dry-run ownership; failures must not remove index files. + return from_native_commit_messages(table, native.build()) + def _resolve_partition_filter(self) -> Optional[Predicate]: if self._partition_filter is not None: return self._partition_filter diff --git a/paimon-python/pypaimon/globalindex/native_index_build.py b/paimon-python/pypaimon/globalindex/native_index_build.py index 914193640722..3a5c6948a743 100644 --- a/paimon-python/pypaimon/globalindex/native_index_build.py +++ b/paimon-python/pypaimon/globalindex/native_index_build.py @@ -32,8 +32,6 @@ def build_native_global_index(builder, partition_filter): from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import VINDEX_IDENTIFIERS if (builder._index_type not in ('btree', 'bitmap', 'full-text') + tuple(VINDEX_IDENTIFIERS) or not table.options.native_write_enabled() - or not table.options.data_evolution_enabled() - or not table.options.global_index_enabled() or (table.options.deletion_vectors_enabled() and builder._index_type == 'full-text') or table.options.query_auth_enabled or table.is_primary_key_table diff --git a/paimon-python/pypaimon/globalindex/vindex/vindex_vector_global_index_reader.py b/paimon-python/pypaimon/globalindex/vindex/vindex_vector_global_index_reader.py index a8e8aee60c35..a82b618f41b7 100644 --- a/paimon-python/pypaimon/globalindex/vindex/vindex_vector_global_index_reader.py +++ b/paimon-python/pypaimon/globalindex/vindex/vindex_vector_global_index_reader.py @@ -111,6 +111,7 @@ def __init__(self, file_io, index_path, io_metas, options=None): self._reader = None self._metadata = None self._search_params_type = None + self._native_reader = False self._load_lock = threading.Lock() def visit_vector_search(self, vector_search): @@ -134,6 +135,12 @@ def visit_batch_vector_search(self, batch_vector_search): def _run_search(self, vectors, limit, include_row_ids, query_options): self._ensure_loaded() + if self._native_reader: + results = self._reader.search( + vectors, limit, filter_bytes=_filter_bytes(include_row_ids), + options={str(key): str(value) for key, value in (query_options or {}).items()}) + return [DictBasedScoredIndexResult(scores) if scores else None for scores in results] + queries = self._validate_queries(vectors) n = len(queries) effective_k = self._effective_k(limit, include_row_ids) @@ -180,6 +187,8 @@ def _validate_queries(self, vectors): def vector_metric(self): self._ensure_loaded() + if self._native_reader: + return self._reader.metric() return self._metadata.metric def _effective_k(self, limit, include_row_ids): @@ -200,13 +209,6 @@ def _ensure_loaded(self): if self._reader is not None: return - try: - from paimon_vindex import SearchParams, VectorIndexReader - except ImportError as e: - raise ImportError( - "paimon-vindex is required to read vindex vector indexes. " - "Install paimon-vindex==0.5.0 or pypaimon[vindex].") from e - file_path = (self._io_meta.external_path if self._io_meta.external_path else os.path.join(self._index_path, self._io_meta.file_name)) @@ -216,12 +218,26 @@ def _ensure_loaded(self): index_input = None try: index_input = PaimonVindexInput(stream, parallelism) - reader = VectorIndexReader(index_input) - self._metadata = reader.metadata() - reader.optimize_for_search() + native_reader = _native_reader_type() + if native_reader is not None: + reader = native_reader( + index_input, file_path, self._io_meta.file_size, + {str(key): str(value) for key, value in self._options.items()}) + self._native_reader = True + else: + try: + from paimon_vindex import SearchParams, VectorIndexReader + except ImportError as e: + raise ImportError( + "paimon-vindex is required to read vindex vector indexes. " + "Install paimon-vindex==0.5.0 or pypaimon[vindex].") from e + reader = VectorIndexReader(index_input) + self._metadata = reader.metadata() + reader.optimize_for_search() + self._search_params_type = SearchParams + self._native_reader = False self._index_input = index_input self._reader = reader - self._search_params_type = SearchParams self._stream = stream except BaseException: try: @@ -274,11 +290,17 @@ def _read_parallelism(options, file_path): return parallelism +def _native_reader_type(): + try: + from pypaimon_rust import datafusion + except ImportError: + return None + return getattr(datafusion, 'VindexVectorGlobalIndexReader', None) + + def _filter_bytes(include_row_ids): if include_row_ids is None: return None - if include_row_ids.cardinality() == 0: - return None return include_row_ids.serialize() @@ -359,7 +381,14 @@ def _int_parameter(options, key): if value is None: return None try: - return int(value) + text = str(value) + digits = text[1:] if text[:1] in ('+', '-') else text + if not digits or any(ord(char) > 0xffff or not char.isdecimal() for char in digits): + raise ValueError('Invalid Java integer text') + number = int(text) + if not -(1 << 31) <= number < (1 << 31): + raise ValueError('Java integer overflow') + return number except (TypeError, ValueError) as e: raise ValueError( "Invalid value for '%s': %s. Must be an integer." % (key, value)) from e diff --git a/paimon-python/pypaimon/ray/batch_vector_search.py b/paimon-python/pypaimon/ray/batch_vector_search.py index 164bd5793ad4..9bcaa1606036 100644 --- a/paimon-python/pypaimon/ray/batch_vector_search.py +++ b/paimon-python/pypaimon/ray/batch_vector_search.py @@ -31,7 +31,8 @@ def _execute_batch_vector_search(builder, *, concurrency=None, ray_remote_args=None): concurrency, remote_args = _execution_options(concurrency, ray_remote_args) - reader = builder.new_batch_vector_search_read() + from pypaimon.table.source.native_search_plan import python_search_reader + reader = python_search_reader(builder.new_batch_vector_search_read()) if (type(reader) is not BatchVectorSearchReadImpl or not reader._table.options.data_evolution_enabled()): raise ValueError("Ray vector search supports only data-evolution tables.") diff --git a/paimon-python/pypaimon/ray/full_text_search.py b/paimon-python/pypaimon/ray/full_text_search.py index 6a068f6fabf2..f4028202ac24 100644 --- a/paimon-python/pypaimon/ray/full_text_search.py +++ b/paimon-python/pypaimon/ray/full_text_search.py @@ -28,7 +28,8 @@ def _execute_full_text_search(builder, *, concurrency=None, ray_remote_args=None): concurrency, remote_args = _execution_options(concurrency, ray_remote_args) - reader = builder.new_full_text_read() + from pypaimon.table.source.native_search_plan import python_search_reader + reader = python_search_reader(builder.new_full_text_read()) if (type(reader) is not DataEvolutionFullTextRead or not reader._table.options.data_evolution_enabled()): raise ValueError("Ray full-text search supports only data-evolution tables.") diff --git a/paimon-python/pypaimon/ray/vector_search.py b/paimon-python/pypaimon/ray/vector_search.py index 1a6bfad857d8..7752da972eab 100644 --- a/paimon-python/pypaimon/ray/vector_search.py +++ b/paimon-python/pypaimon/ray/vector_search.py @@ -38,7 +38,8 @@ def _execute_vector_search(builder, *, concurrency=None, ray_remote_args=None): """Execute the builder of an already snapshot-pinned ``VectorQuery``.""" concurrency, remote_args = _execution_options(concurrency, ray_remote_args) - reader = builder.new_vector_search_read() + from pypaimon.table.source.native_search_plan import python_search_reader + reader = python_search_reader(builder.new_vector_search_read()) if (type(reader) is not DataEvolutionVectorRead or not reader._table.options.data_evolution_enabled()): raise ValueError("Ray vector search supports only data-evolution tables.") diff --git a/paimon-python/pypaimon/table/source/batch_vector_search_builder.py b/paimon-python/pypaimon/table/source/batch_vector_search_builder.py index f6fd739a5c64..45e94eacb7ee 100644 --- a/paimon-python/pypaimon/table/source/batch_vector_search_builder.py +++ b/paimon-python/pypaimon/table/source/batch_vector_search_builder.py @@ -100,6 +100,8 @@ class BatchVectorSearchBuilderImpl(AbstractVectorSearchBuilderImpl, BatchVectorSearchBuilder): """Implementation for BatchVectorSearchBuilder.""" + _native_batch_search = True + def __init__(self, table): super().__init__(table) self._query_vectors = None @@ -126,7 +128,7 @@ def new_batch_vector_search_read(self): if not self._query_vectors: raise ValueError( "Query vectors must be set via with_query_vectors()") - return BatchVectorSearchReadImpl( + reader = BatchVectorSearchReadImpl( self._table, self._limit, self._vector_column, @@ -135,3 +137,5 @@ def new_batch_vector_search_read(self): partition_filter=self._partition_filter, options=self._options, ) + from pypaimon.table.source.native_search_plan import native_vector_read + return native_vector_read(self, reader, batch=True) 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 3d2d4dd80211..cb434215a0c5 100644 --- a/paimon-python/pypaimon/table/source/full_text_search_builder.py +++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py @@ -147,20 +147,24 @@ def _rebuild_leaf_indices_by_name(cls, predicate, name_to_idx): return predicate.new_index(name_to_idx[predicate.field]) def new_full_text_scan(self) -> FullTextScan: + from pypaimon.table.source.native_search_plan import native_full_text_scan reject_search_under_query_auth(self._table) definition = self._primary_key_full_text_definition() if definition is not None: from pypaimon.table.source.primary_key_full_text_scan import PrimaryKeyFullTextScan - return PrimaryKeyFullTextScan( + scan = PrimaryKeyFullTextScan( self._table, definition, partition_filter=self._partition_filter) - return DataEvolutionFullTextScan( + return native_full_text_scan(self, scan) + scan = DataEvolutionFullTextScan( self._table, self._text_columns(), partition_filter=self._partition_filter, filter_=self._filter, ) + return native_full_text_scan(self, scan) def new_full_text_read(self) -> FullTextRead: + from pypaimon.table.source.native_search_plan import native_full_text_read reject_search_under_query_auth(self._table) if self._limit <= 0: raise ValueError("Limit must be positive, set via with_limit()") @@ -178,10 +182,11 @@ def new_full_text_read(self) -> FullTextRead: "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( + reader = PrimaryKeyFullTextRead( self._table, self._limit, self._text_columns(), self._query, definition=definition, partition_filter=self._partition_filter) - return DataEvolutionFullTextRead( + return native_full_text_read(self, reader) + reader = DataEvolutionFullTextRead( self._table, self._limit, self._text_columns(), @@ -189,6 +194,7 @@ def new_full_text_read(self) -> FullTextRead: partition_filter=self._partition_filter, filter_=self._filter, ) + return native_full_text_read(self, reader) def _text_columns(self): if self._query is None: diff --git a/paimon-python/pypaimon/table/source/native_full_text_search.py b/paimon-python/pypaimon/table/source/native_full_text_search.py index 3665c11ebc3d..e2d3906a9354 100644 --- a/paimon-python/pypaimon/table/source/native_full_text_search.py +++ b/paimon-python/pypaimon/table/source/native_full_text_search.py @@ -27,7 +27,7 @@ from pypaimon.write.native_commit import _rest_catalog_supported -def try_native_full_text_search(builder): +def _native_full_text_builder(builder, require_limit=True): """Qualify the route before execution; Rust owns scan, filtering and ranking.""" table = builder._table reject_search_under_query_auth(table) @@ -38,13 +38,12 @@ def try_native_full_text_search(builder): or not _rest_catalog_supported(table)): return None primary_key = builder._primary_key_full_text_definition() is not None - if not primary_key and (not table.options.data_evolution_enabled() - or not table.options.row_tracking_enabled() + if not primary_key and (not table.options.row_tracking_enabled() or table.is_primary_key_table): return None - if primary_key and table.options.full_text_index_search_mode() != GlobalIndexSearchMode.FAST: + if require_limit and primary_key and table.options.full_text_index_search_mode() != GlobalIndexSearchMode.FAST: raise NotImplementedError('Primary-key full-text search only supports the FAST full-text index search mode') - if builder._limit is None or builder._limit <= 0: + if require_limit and (builder._limit is None or builder._limit <= 0): raise ValueError('Limit must be positive, set via with_limit()') # The full-text adapter accepts typed lists and booleans. Reuse its mapping # before the generic table bridge stringifies options for Rust. @@ -53,7 +52,8 @@ def try_native_full_text_search(builder): native = _native_table(table.copy(options)).new_full_text_search_builder() if builder._field_name is not None and builder._query is not None: native.with_query(builder._field_name, builder._query) - native.with_limit(builder._limit) + if builder._limit is not None: + native.with_limit(builder._limit) try: if builder._partition_filter is not None: native.with_partition_filter(_predicate_to_native(builder._partition_filter)) @@ -63,6 +63,15 @@ def try_native_full_text_search(builder): # Qualify literal support before searching. Some Python literals cannot # be represented by the strict typed predicate FFI. return None + return native + + +def try_native_full_text_search(builder): + native = _native_full_text_builder(builder) + if native is None: + return None + table = builder._table + primary_key = builder._primary_key_full_text_definition() is not None # Once execution starts, propagate errors instead of running a second search. result = native.execute_local() if primary_key: diff --git a/paimon-python/pypaimon/table/source/native_hybrid_search.py b/paimon-python/pypaimon/table/source/native_hybrid_search.py index a820660ab0ab..19e3769b74ff 100644 --- a/paimon-python/pypaimon/table/source/native_hybrid_search.py +++ b/paimon-python/pypaimon/table/source/native_hybrid_search.py @@ -32,7 +32,6 @@ def try_native_hybrid_search(builder): return None reject_search_under_query_auth(table) if (not table.options.native_read_enabled() - or not table.options.data_evolution_enabled() or not table.options.row_tracking_enabled() or table.is_primary_key_table or table.options.file_format() != 'parquet' diff --git a/paimon-python/pypaimon/table/source/native_search_plan.py b/paimon-python/pypaimon/table/source/native_search_plan.py new file mode 100644 index 000000000000..df14f527f332 --- /dev/null +++ b/paimon-python/pypaimon/table/source/native_search_plan.py @@ -0,0 +1,254 @@ +# 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. + +"""Native search plans with ordinary, transportable Python source metadata.""" + +from copy import copy +import hashlib +import json +import pickle + +from pypaimon.common.json_util import JSON +from pypaimon.common.options.core_options import CoreOptions +from pypaimon.common.options.options import Options +from pypaimon.globalindex.vector_search_result import DictBasedScoredIndexResult +from pypaimon.read.native_plan import deserialize_split_v1, native_method_available, _partition_fields +from pypaimon.snapshot.snapshot import Snapshot +from pypaimon.table.source.full_text_scan import FullTextScanPlan +from pypaimon.table.source.primary_key_full_text_scan import ( + PrimaryKeyFullTextScanPlan, PrimaryKeyFullTextSearchSplit, +) +from pypaimon.table.source.primary_key_scored_result import PrimaryKeyScoredResult +from pypaimon.table.source.primary_key_vector_scan import PrimaryKeyVectorScanPlan, PrimaryKeyVectorSearchSplit +from pypaimon.table.source.vector_search_scan import VectorSearchScanPlan +from pypaimon.utils.range import Range +from pypaimon.write.commit_message_serializer import _index_from_row + + +def _metadata_fingerprint(plan): + # Only captured metadata participates: never pickle a Table, catalog or + # live native handle. This also detects edits through a shallow copy. + metadata = (plan.snapshot(), plan.splits(), getattr(plan, 'snapshot_id', None)) + return hashlib.sha256(pickle.dumps(metadata, protocol=4)).digest() + + +class _NativePlan: + def __getstate__(self): + # Transport immutable snapshot data and ordinary Python splits. The + # original result and its metadata remain unchanged by serialization. + state = self.__dict__.copy() + state.pop('_native_plan', None) + state.pop('_native_fingerprint', None) + return state + + def _unchanged_native_plan(self): + native = getattr(self, '_native_plan', None) + return native if native is not None and self._native_fingerprint == _metadata_fingerprint(self) else None + + +class _NativeVectorPlan(_NativePlan, VectorSearchScanPlan): + pass + + +class _NativeFullTextPlan(_NativePlan, FullTextScanPlan): + pass + + +class _NativePkVectorPlan(_NativePlan, PrimaryKeyVectorScanPlan): + pass + + +class _NativePkFullTextPlan(_NativePlan, PrimaryKeyFullTextScanPlan): + pass + + +def _range(value): + return Range(value['from'], value['to']) + + +def _files(values): + return [_index_from_row(bytes(value)) for value in values] + + +def _plan_from_native(table, native, full_text, primary_key, source_modes): + from pypaimon.table.source.full_text_search_split import IndexFullTextSearchSplit, RawFullTextSearchSplit + from pypaimon.table.source.vector_search_split import IndexVectorSearchSplit, RawVectorSearchSplit + + metadata = json.loads(native._metadata()) + snapshot = JSON.from_json(json.dumps(metadata['snapshot']), Snapshot) if metadata['snapshot'] is not None else None + splits = [] + for value in metadata['splits']: + if value['kind'] == 'primary-key': + split = deserialize_split_v1(bytes(value['data_split']), _partition_fields(table), + table.trimmed_primary_keys_fields) + ranges = {name: tuple(_range(range_) for range_ in ranges) + for name, ranges in value['row_ranges_by_file'].items()} + cls = PrimaryKeyFullTextSearchSplit if full_text else PrimaryKeyVectorSearchSplit + splits.append(cls(split, tuple(_files(value['payloads'])), tuple(value['uncovered_data_files']), ranges)) + elif value['kind'] == 'index': + row_range = _range(value['row_range']) + args = (row_range.from_, row_range.to, _files(value['index_files']), _files(value['scalar_index_files'])) + splits.append(IndexFullTextSearchSplit(value['column'], *args) + if full_text else IndexVectorSearchSplit(*args)) + else: + ranges = [_range(range_) for range_ in value['row_ranges']] + splits.append(RawFullTextSearchSplit(ranges) if full_text else RawVectorSearchSplit( + ranges, _files(value['scalar_index_files']), value['index_type'])) + if primary_key: + cls = _NativePkFullTextPlan if full_text else _NativePkVectorPlan + plan = cls(metadata['snapshot_id'] or 0, splits) + plan._snapshot = snapshot + else: + cls = _NativeFullTextPlan if full_text else _NativeVectorPlan + plan = cls(splits, snapshot) + plan._native_plan = native + plan._source_modes = source_modes + plan._table_options = dict(table.options.options.to_map()) + plan._native_fingerprint = _metadata_fingerprint(plan) + return plan + + +def _result(table, value, primary_key): + return (PrimaryKeyScoredResult._from_native(table, value) + if primary_key else DictBasedScoredIndexResult(value.row_ids())) + + +class _NativeScan: + def __init__(self, table, scan, full_text, primary_key, python, source_modes): + self._table = table + self._scan = scan + self._full_text = full_text + self._primary_key = primary_key + self._python = python + self._source_modes = source_modes + + def __getstate__(self): + state = self.__dict__.copy() + state['_scan'] = None + return state + + def scan(self): + if self._scan is None: + return self._python.scan() + return _plan_from_native(self._table, self._scan.scan(), self._full_text, self._primary_key, self._source_modes) + + +class _NativeRead: + def __init__(self, table, native, python, primary_key, source_modes): + self._table = table + self._native = native + self._python = python + self._primary_key = primary_key + self._source_modes = source_modes + self._table_options = dict(table.options.options.to_map()) + + def __getattr__(self, name): + python = self.__dict__.get('_python') + if python is None: + raise AttributeError(name) + return getattr(python, name) + + def __getstate__(self): + state = self.__dict__.copy() + state['_native'] = None + return state + + def read_plan(self, plan): + native = self._native_plan(plan) + if native is not None and self._native is not None: + # Once execution starts, propagate errors without a second search. + return _result(self._table, self._native.read_plan(native), self._primary_key) + reader = copy(self._python) + if plan.snapshot() is not None: + reader._table = reader._table._copy_with_snapshot(plan.snapshot()) + return reader.read_plan(plan) + + def read_batch_plan(self, plan): + native = self._native_plan(plan) + if native is not None and self._native is not None: + return [_result(self._table, result, False) for result in self._native.read_batch_plan(native)] + reader = copy(self._python) + if plan.snapshot() is not None: + reader._table = reader._table._copy_with_snapshot(plan.snapshot()) + return reader.read_batch_plan(plan) + + def _native_plan(self, plan): + # Search modes choose source work at scan time. A new reader may use + # another query or limit, but must still consume these captured splits. + if (isinstance(plan, _NativePlan) and plan._source_modes == self._source_modes + and plan._table_options == self._table_options): + return plan._unchanged_native_plan() + return None + + +def _source_modes(builder, full_text=False): + options = dict(builder._table.options.options.to_map()) + options.update(getattr(builder, '_options', {})) + core = CoreOptions(Options(options)) + return ((core.full_text_index_search_mode(),) if full_text else + (core.vector_index_search_mode(), core.scalar_index_search_mode())) + + +def python_search_reader(reader): + """Keep the existing Ray split workers independent of a live native reader.""" + return reader._python if isinstance(reader, _NativeRead) else reader + + +def native_vector_scan(builder, python, batch=False): + if not native_method_available('VectorScanPlan', '_metadata'): + return python + from pypaimon.table.source.native_vector_search import _native_vector_builder + native = _native_vector_builder(builder, batch) + return python if native is None else _NativeScan( + builder._table, native.new_vector_search_scan(), False, builder._is_primary_key_vector_search(), python, + _source_modes(builder)) + + +def native_vector_read(builder, python, batch=False): + if not native_method_available('VectorScanPlan', '_metadata'): + return python + from pypaimon.table.source.native_vector_search import _native_vector_builder + native = _native_vector_builder(builder, batch) + if native is None: + return python + if batch: + native.with_query_vectors(builder._query_vectors) + reader = native.new_batch_vector_search_read() + else: + native.with_query_vector(builder._query_vector) + reader = native.new_vector_search_read() + return _NativeRead(builder._table, reader, python, builder._is_primary_key_vector_search(), _source_modes(builder)) + + +def native_full_text_scan(builder, python): + if not native_method_available('FullTextScanPlan', '_metadata'): + return python + from pypaimon.table.source.native_full_text_search import _native_full_text_builder + native = _native_full_text_builder(builder, require_limit=False) + return python if native is None else _NativeScan( + builder._table, native.new_full_text_scan(), True, + builder._primary_key_full_text_definition() is not None, python, _source_modes(builder, True)) + + +def native_full_text_read(builder, python): + if not native_method_available('FullTextScanPlan', '_metadata'): + return python + from pypaimon.table.source.native_full_text_search import _native_full_text_builder + native = _native_full_text_builder(builder) + return python if native is None else _NativeRead( + builder._table, native.new_full_text_read(), python, builder._primary_key_full_text_definition() is not None, + _source_modes(builder, True)) diff --git a/paimon-python/pypaimon/table/source/native_vector_search.py b/paimon-python/pypaimon/table/source/native_vector_search.py index b0522e4f2d0d..56ca2e626ba1 100644 --- a/paimon-python/pypaimon/table/source/native_vector_search.py +++ b/paimon-python/pypaimon/table/source/native_vector_search.py @@ -25,11 +25,11 @@ from pypaimon.table.source.primary_key_scored_result import PrimaryKeyScoredResult -def try_native_vector_search(builder, queries, batch=False): - """Return existing Python result types, or None to use the Python reader. +def _native_vector_builder(builder, batch=False): + """Configure a native builder, or return None before execution. Search, scalar filtering, refinement and physical-position selection all - belong to Rust core. Python only converts the scored result metadata. + belong to Rust core. Python only translates options and typed predicates. """ table = builder._table reject_search_under_query_auth(table) @@ -64,6 +64,14 @@ def try_native_vector_search(builder, queries, batch=False): except ValueError: # Qualify unsupported typed literals before searching. return None + return native + + +def try_native_vector_search(builder, queries, batch=False): + native = _native_vector_builder(builder, batch) + if native is None: + return None + table = builder._table # Once the Native route is chosen, errors must not trigger another search # against a potentially newer snapshot. if batch: diff --git a/paimon-python/pypaimon/table/source/vector_search_builder.py b/paimon-python/pypaimon/table/source/vector_search_builder.py index 76ce6575e9bf..fe8eebd5b460 100644 --- a/paimon-python/pypaimon/table/source/vector_search_builder.py +++ b/paimon-python/pypaimon/table/source/vector_search_builder.py @@ -97,6 +97,8 @@ def execute_local(self): class AbstractVectorSearchBuilderImpl: """Shared state and filter/partition handling for the vector search builders.""" + _native_batch_search = False + def __init__(self, table): self._table = table self._limit = 0 @@ -223,13 +225,15 @@ def new_vector_search_scan(self): if self._is_primary_key_vector_search(): from pypaimon.table.source.primary_key_vector_scan import PrimaryKeyVectorScan scan_class = PrimaryKeyVectorScan - return scan_class( + scan = scan_class( self._table, self._vector_column, filter_=self._filter, partition_filter=self._partition_filter, options=self._options, ) + from pypaimon.table.source.native_search_plan import native_vector_scan + return native_vector_scan(self, scan, self._native_batch_search) def _is_primary_key_vector_search(self): if self._vector_column is None: @@ -270,7 +274,7 @@ def new_vector_search_read(self): if self._is_primary_key_vector_search(): from pypaimon.table.source.primary_key_vector_read import PrimaryKeyVectorRead read_class = PrimaryKeyVectorRead - return read_class( + reader = read_class( self._table, self._limit, self._vector_column, @@ -279,3 +283,5 @@ def new_vector_search_read(self): partition_filter=self._partition_filter, options=self._options, ) + from pypaimon.table.source.native_search_plan import native_vector_read + return native_vector_read(self, reader) diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py b/paimon-python/pypaimon/table/source/vector_search_read.py index 20d661f7094f..607aff799879 100644 --- a/paimon-python/pypaimon/table/source/vector_search_read.py +++ b/paimon-python/pypaimon/table/source/vector_search_read.py @@ -1021,6 +1021,7 @@ def _configured_vector_metric(options, vector_column, index_type=None): def _raw_search_metric(table, vector_column, options, index_type=None): + from pypaimon.globalindex.lumina.lumina_vector_global_index_reader import LUMINA_IDENTIFIERS from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import VINDEX_IDENTIFIERS table_map = _table_options_map(table) @@ -1030,7 +1031,7 @@ def _raw_search_metric(table, vector_column, options, index_type=None): return configured # Before an index exists, use its writer's default, not another column's metric. - if index_type in VINDEX_IDENTIFIERS: + if index_type in VINDEX_IDENTIFIERS or index_type in LUMINA_IDENTIFIERS: return "inner_product" inferred = None diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index 8a006a7359b4..e5fc257cb855 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/paimon-python/pypaimon/tests/global_index_build_test.py @@ -1121,6 +1121,20 @@ def test_create_vindex_global_index_rejects_generic_unsupported_tables(self): with self.assertRaisesRegex(ValueError, 'deletion vectors'): dv_table.create_global_index('embedding', index_type=FULL_TEXT_IDENTIFIER) + def test_build_options_cannot_change_table_deletion_vector_eligibility(self): + for enabled in (False, True): + with self.subTest(deletion_vectors=enabled): + table_options = dict(self.table_options) + table_options['deletion-vectors.enabled'] = str(enabled).lower() + table = self._create_table(options=table_options) + build_options = {'deletion-vectors.enabled': str(not enabled).lower()} + if enabled: + with self.assertRaisesRegex(ValueError, 'deletion vectors'): + GlobalIndexBuilder(table, 'name', index_type=FULL_TEXT_IDENTIFIER, options=build_options) + else: + self.assertEqual([], GlobalIndexBuilder( + table, 'name', index_type=FULL_TEXT_IDENTIFIER, options=build_options).build()) + def test_vindex_native_options_follow_java_mapping(self): data_type = ArrayType(True, AtomicType('FLOAT')) options = { diff --git a/paimon-python/pypaimon/tests/global_index_family_parity_test.py b/paimon-python/pypaimon/tests/global_index_family_parity_test.py new file mode 100644 index 000000000000..eaf1efffd736 --- /dev/null +++ b/paimon-python/pypaimon/tests/global_index_family_parity_test.py @@ -0,0 +1,96 @@ +# 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. + +"""Java parity when one scalar column has several index families.""" + +import unittest + +import pyarrow as pa + +from pypaimon.globalindex.data_evolution_global_index_scanner import DataEvolutionGlobalIndexScanner +from pypaimon.index.index_file_handler import IndexFileHandler +from pypaimon.tests.data_evolution_test_helpers import BatchModeMixin, DataEvolutionTestBase +from pypaimon.utils.range import Range + + +class GlobalIndexFamilyParityTest(BatchModeMixin, DataEvolutionTestBase, unittest.TestCase): + table_options = { + 'row-tracking.enabled': 'true', + 'data-evolution.enabled': 'true', + 'global-index.enabled': 'true', + 'bucket': '-1', + 'file.format': 'parquet', + 'scan.native-plan.enabled': 'false', + 'read.native.enabled': 'false', + 'write.native.enabled': 'false', + 'commit.native.enabled': 'false', + } + + def _append(self, table, ids, names): + self._write_arrow(table, pa.table({ + 'id': ids, + 'name': names, + 'age': [10] * len(ids), + 'city': ['x'] * len(ids), + }, schema=self.pa_schema)) + + def _assert_scan(self, table, indexed_ranges, expected_ids): + read = table.new_read_builder() + predicate = read.new_predicate_builder().equal('name', 'match') + with DataEvolutionGlobalIndexScanner.create(table, predicate=predicate) as scanner: + result = scanner.scan(predicate) + self.assertEqual(indexed_ranges, result.results().to_range_list()) + read.with_filter(predicate) + rows = read.new_read().to_arrow(read.new_scan().plan().splits()) + actual = [] if rows is None else sorted(rows.column('id').to_pylist()) + self.assertEqual(expected_ids, actual) + + def test_families_union_matches_across_unequal_coverage(self): + for first, second in [('btree', 'bitmap'), ('bitmap', 'btree')]: + for mode in ['fast', 'full', 'detail']: + with self.subTest(first=first, second=second, mode=mode): + options = dict(self.table_options, **{'global-index.search-mode': mode}) + table = self._create_table(options=options) + self._append(table, [1, 2], ['old', 'other']) + self.assertGreater(table.create_global_index('name', index_type=first), 0) + self._append(table, [3, 4], ['match', 'other']) + self.assertGreater(table.create_global_index('name', index_type=second), 0) + # A third append is outside both families. Only FULL/DETAIL + # extend the indexed selection with these unindexed rows. + self._append(table, [5], ['match']) + coverage = {} + for entry in IndexFileHandler(table).scan(None): + meta = entry.index_file.global_index_meta + if meta is not None: + coverage.setdefault(entry.index_file.index_type, []).append( + (meta.row_range_start, meta.row_range_end)) + self.assertEqual([(0, 1)], coverage[first]) + self.assertEqual([(0, 3)], coverage[second]) + # The later family owns the only indexed match. Intersecting + # it with the earlier family's empty result loses that row. + self._assert_scan(table, [Range(2, 2)], [3] if mode == 'fast' else [3, 5]) + + def test_one_family_retains_matches_and_unindexed_fallback(self): + for family in ['btree', 'bitmap']: + for mode in ['fast', 'full', 'detail']: + with self.subTest(family=family, mode=mode): + options = dict(self.table_options, **{'global-index.search-mode': mode}) + table = self._create_table(options=options) + self._append(table, [1, 2], ['match', 'other']) + self.assertGreater(table.create_global_index('name', index_type=family), 0) + self._append(table, [3], ['match']) + self._assert_scan(table, [Range(0, 0)], [1] if mode == 'fast' else [1, 3]) diff --git a/paimon-python/pypaimon/tests/native_append_index_search_test.py b/paimon-python/pypaimon/tests/native_append_index_search_test.py new file mode 100644 index 000000000000..03f8f5c10603 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_append_index_search_test.py @@ -0,0 +1,137 @@ +# 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. + +"""Indexed searches on row-tracked append tables share the native search path.""" + +from contextlib import ExitStack +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon.read.native_plan import native_method_available +from pypaimon.table.source.full_text_search_builder import FullTextSearchBuilderImpl +from pypaimon.table.source.hybrid_search_builder import HybridSearchBuilderImpl +from pypaimon.table.source.vector_search_read import DataEvolutionVectorRead +from pypaimon.tests import native_sorted_index_build_test as indexes +from pypaimon.tests.vector_search_filter_test import match_query + +rest_catalog = indexes.rest_catalog +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_method_available('Table', 'new_hybrid_search_builder'), reason='Rust search API required')] + +SCHEMA = pa.schema([ + ('id', pa.int32()), ('content', pa.string()), ('embedding', pa.list_(pa.float32())), + ('label', pa.string()), ('pt', pa.int32()), +]) +ROWS = [ + {'id': 0, 'content': 'paimon lake', 'embedding': [0., 0.], 'label': 'keep', 'pt': 0}, + {'id': 1, 'content': 'paimon', 'embedding': [1., 0.], 'label': 'drop', 'pt': 0}, + {'id': 2, 'content': 'lake', 'embedding': [2., 0.], 'label': 'keep', 'pt': 1}, + {'id': 3, 'content': 'paimon', 'embedding': [3., 0.], 'label': 'keep', 'pt': 1}, +] + + +def _table(catalog, global_index): + table = indexes._create(catalog, schema=SCHEMA, partitioned=True, options={ + 'data-evolution.enabled': 'false', 'global-index.enabled': str(global_index).lower(), + 'write.native.enabled': 'false', 'read.native.enabled': 'false', + 'scan.native-plan.enabled': 'false', 'vector-index.search-mode': 'fast', + 'full-text-index.search-mode': 'fast', 'scalar-index.search-mode': 'fast', + }) + indexes._append(table, ROWS, SCHEMA) + # The existing Python builders already support row-tracked append tables. + # Keep these builds independent from the native builder being extended. + table.create_global_index('content', index_type='full-text') + table.create_global_index('embedding', index_type='ivf-flat', options={ + 'ivf-flat.dimension': '2', 'ivf-flat.nlist': '1', 'ivf-flat.distance.metric': 'l2'}) + table.create_global_index('label', index_type='btree') + return table + + +def _builder(table, family, partition=False, row_filter=False): + if family == 'full-text': + builder = table.new_full_text_search_builder().with_query('content', match_query('paimon')).with_limit(4) + elif family == 'vector': + builder = (table.new_vector_search_builder().with_vector_column('embedding') + .with_query_vector([0., 0.]).with_limit(4).with_options({'ivf.nprobe': '1'})) + else: + builder = table.new_hybrid_search_builder().with_limit(4) + if family in ('hybrid-vector', 'hybrid-mixed'): + builder.add_vector_route('embedding', [0., 0.], 4, options={'ivf.nprobe': '1'}) + if family in ('hybrid-text', 'hybrid-mixed'): + builder.add_full_text_route('content', match_query('paimon'), 4) + predicate = table.new_read_builder().new_predicate_builder() + if partition: + builder.with_partition_filter(predicate.equal('pt', 1)) + if row_filter: + builder.with_filter(predicate.equal('label', 'keep')) + return builder + + +def _scores(result): + getter = result.score_getter() if not result.is_empty() else None + return {row_id: getter(row_id) for row_id in result.results()} + + +def _native_only(): + stack = ExitStack() + for target, name in [ + (FullTextSearchBuilderImpl, 'new_full_text_scan'), + (DataEvolutionVectorRead, 'read_plan'), + (HybridSearchBuilderImpl, 'route_builders'), + (HybridSearchBuilderImpl, 'rank'), + ]: + stack.enter_context(patch.object(target, name, side_effect=AssertionError('Python search ran'))) + return stack + + +@pytest.mark.parametrize('global_index', [False, True]) +@pytest.mark.parametrize('partition', [False, True]) +@pytest.mark.parametrize('family', ['full-text', 'vector', 'hybrid-text', 'hybrid-vector', 'hybrid-mixed']) +def test_indexed_append_searches_use_native_without_data_evolution(rest_catalog, family, partition, global_index): + table = _table(rest_catalog[0], global_index) + expected = _scores(_builder(table, family, partition).execute_local()) + with _native_only(): + actual = _scores(_builder(table.copy({'read.native.enabled': 'true'}), family, partition).execute_local()) + assert actual == pytest.approx(expected, abs=1e-6) + wanted = {3} if partition else {0, 1, 3} + if family in ('vector', 'hybrid-vector', 'hybrid-mixed'): + wanted = {2, 3} if partition else {0, 1, 2, 3} + assert set(actual) == wanted + # Exercise the ordinary native planner too in the native-plan CI lane. + reader = table.copy({'scan.native-plan.enabled': 'true'}).new_read_builder() + rows = reader.new_read().to_arrow(reader.new_scan().plan().splits()) + assert sorted(rows.column('id').to_pylist()) == [0, 1, 2, 3] + + +@pytest.mark.parametrize('family', ['vector', 'hybrid-vector']) +def test_exact_scalar_index_filters_work_on_append_vector_routes(rest_catalog, family): + table = _table(rest_catalog[0], True) + expected = _scores(_builder(table, family, row_filter=True).execute_local()) + with _native_only(): + actual = _scores(_builder(table.copy({'read.native.enabled': 'true'}), family, row_filter=True).execute_local()) + assert actual == pytest.approx(expected, abs=1e-6) + assert set(actual) == {0, 2, 3} + + +@pytest.mark.parametrize('family', ['full-text', 'hybrid-text', 'hybrid-mixed']) +@pytest.mark.parametrize('native', [False, True]) +def test_append_full_text_row_filters_remain_unsupported(rest_catalog, family, native): + table = _table(rest_catalog[0], True).copy({'read.native.enabled': str(native).lower()}) + with pytest.raises(Exception, match='[Ff]ull-text row filters require a data-evolution table'): + _builder(table, family, row_filter=True).execute_local() diff --git a/paimon-python/pypaimon/tests/native_generic_index_build_test.py b/paimon-python/pypaimon/tests/native_generic_index_build_test.py index 425f92ee3c3d..7fc13009368b 100644 --- a/paimon-python/pypaimon/tests/native_generic_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_generic_index_build_test.py @@ -72,6 +72,31 @@ def _build(table, builder): return from_native_commit_messages(table, builder.build()) +@pytest.mark.parametrize('kind', ['btree', 'bitmap'] + KINDS) +@pytest.mark.parametrize('evolution, enabled', [(False, True), (True, False), (False, False)]) +def test_native_build_requires_row_tracking_not_read_optimization_options(rest_catalog, kind, evolution, enabled): + catalog, _ = rest_catalog + table = _table(catalog, {'data-evolution.enabled': str(evolution).lower(), + 'global-index.enabled': str(enabled).lower()}) + column = 'id' if kind in ('btree', 'bitmap') else 'name' if kind == 'full-text' else 'embedding' + options = ({'sorted-index.records-per-file': 100} if kind in ('btree', 'bitmap') else _options(kind)) + with patch.object(GlobalIndexBuilder, '_build_sorted_index', side_effect=AssertionError('Python sorted build ran')): + with patch.object(GlobalIndexBuilder, '_build_generic_index', + side_effect=AssertionError('Python generic build ran')): + messages = GlobalIndexBuilder(table, column, index_type=kind, options=options).build() + files = index_tests._message_files(messages) + assert sum(file.row_count for file in files) == 32 + assert {file.index_type for file in files} == {kind} + index_tests._commit(table, messages) + native = _native_table(table).new_global_index_build_builder().with_index_column(column).with_index_type(kind) + assert native.with_options({str(key): str(value) for key, value in options.items()}).build() == [] + # Build messages remain usable after the caller enables query optimizations. + readable = table.copy({'global-index.enabled': 'true', 'read.native.enabled': 'true'}) + if kind in ('btree', 'bitmap'): + predicate = readable.new_read_builder().new_predicate_builder().equal('id', 10) + assert index_tests._ids(readable, predicate) == [10] + + @pytest.mark.parametrize('kind', KINDS) @pytest.mark.parametrize('native_commit', [False, True]) def test_prepare_partition_then_commit_and_query(rest_catalog, kind, native_commit): diff --git a/paimon-python/pypaimon/tests/native_global_index_drop_test.py b/paimon-python/pypaimon/tests/native_global_index_drop_test.py new file mode 100644 index 000000000000..a863f3e93aba --- /dev/null +++ b/paimon-python/pypaimon/tests/native_global_index_drop_test.py @@ -0,0 +1,164 @@ +# 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. + +"""Rust plans index drops; Python retains commit and dry-run ownership.""" + +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import pytest + +from pypaimon.globalindex.drop_global_index import GlobalIndexDropper +from pypaimon.index.index_file_handler import IndexFileHandler +from pypaimon.read.native_plan import _native_table, _predicate_to_native, native_method_available +from pypaimon.tests import native_sorted_index_build_test as indexes +from pypaimon.write.table_commit import BatchTableCommit + +rest_catalog = indexes.rest_catalog +native_drop = pytest.mark.skipif( + not native_method_available('Table', 'new_global_index_drop_builder'), + reason='Rust global index drop builder required') + + +@native_drop +def test_preparation_failure_does_not_retry_python_planning(): + table = SimpleNamespace( + field_dict={'id': None}, + options=SimpleNamespace(native_write_enabled=lambda: True, query_auth_enabled=False)) + native = Mock() + native.with_index_columns.return_value = native + native.with_index_type.return_value = native + native.build.side_effect = OSError('metadata read failed') + factory = SimpleNamespace(supports_index_type=lambda kind: True) + with patch('pypaimon.read.native_plan.native_method_available', return_value=True), \ + patch('pypaimon.write.native_commit._rest_catalog_supported', return_value=True), \ + patch('pypaimon.read.native_plan._native_table') as native_table, \ + patch('pypaimon_rust.datafusion.GlobalIndexDropBuilder', factory, create=True), \ + patch.object(IndexFileHandler, 'scan', side_effect=AssertionError('Python fallback')): + native_table.return_value.new_global_index_drop_builder.return_value = native + with pytest.raises(OSError, match='metadata read failed'): + GlobalIndexDropper(table, 'id').build() + native.build.assert_called_once_with() + native.execute.assert_not_called() + + +@pytest.mark.native_plan +@native_drop +@pytest.mark.parametrize('kind', ['btree', 'bitmap']) +@pytest.mark.parametrize('selection', ['all', 'dict', 'list', 'null', 'predicate']) +def test_prepare_and_dry_run_preserve_files_until_explicit_commit(rest_catalog, kind, selection): + catalog, _ = rest_catalog + table = indexes._create(catalog, partitioned=True) + indexes._append(table, indexes.ROWS) + table.create_global_index('name', index_type=kind) + table.create_global_index('id', index_type=kind) + entries = IndexFileHandler(table).scan(None) + all_files = [entry.index_file for entry in entries] + predicates = table.new_read_builder().new_predicate_builder() + kwargs = { + 'all': {}, + 'dict': {'partitions': {'pt': 1}}, + 'list': {'partitions': [{'pt': 0}, {'pt': 1}]}, + 'null': {'partitions': {'pt': None}}, + 'predicate': {'partition_filter': predicates.equal('pt', 0)}, + }[selection] + selected_partitions = { + 'all': {0, 1, None}, 'dict': {1}, 'list': {0, 1}, 'null': {None}, 'predicate': {0}, + }[selection] + expected = {entry.index_file.file_name for entry in entries + if entry.index_file.global_index_meta.index_field_id == table.field_dict['name'].id + and entry.partition.values[0] in selected_partitions} + before = table.snapshot_manager().get_latest_snapshot().id + with patch.object(IndexFileHandler, 'scan', side_effect=AssertionError('Python drop planning')): + messages = GlobalIndexDropper(table, 'name', index_type=kind, **kwargs).build() + with patch.object(BatchTableCommit, 'commit', side_effect=AssertionError('dry run committed')): + assert table.drop_global_index('name', index_type=kind, dry_run=True, **kwargs) == len(expected) + assert {entry.index_file.file_name for message in messages for entry in message.index_deletes} == expected + assert all(not message.new_files and not message.index_adds for message in messages) + assert table.snapshot_manager().get_latest_snapshot().id == before + assert {file.file_name for file in indexes._files(table)} == {file.file_name for file in all_files} + indexes._commit(table, messages) + assert {file.file_name for file in indexes._files(table)} == { + file.file_name for file in all_files if file.file_name not in expected} + # Old snapshots still refer to the removed metadata's files. + assert all(table.file_io.exists(indexes._path(table, file)) for file in all_files) + assert table.drop_global_index('name', index_type=kind, **kwargs) == 0 + assert table.snapshot_manager().get_latest_snapshot().id == before + 1 + + +@pytest.mark.native_plan +@native_drop +def test_native_drop_binding_conjunction_and_execute_dry_run(rest_catalog): + catalog, _ = rest_catalog + table = indexes._create(catalog, partitioned=True) + indexes._append(table, indexes.ROWS) + table.create_global_index('name') + before = table.snapshot_manager().get_latest_snapshot().id + predicates = table.new_read_builder().new_predicate_builder() + native = _native_table(table).new_global_index_drop_builder().with_index_column('name') + native.with_partition_filter(_predicate_to_native(predicates.equal('pt', 0))) + native.with_partition_filter(_predicate_to_native(predicates.equal('pt', 1))) + assert native.build() == [] + assert native.execute() == 0 + assert table.snapshot_manager().get_latest_snapshot().id == before + with pytest.raises(Exception, match='only partition keys'): + native.with_partition_filter(_predicate_to_native(predicates.equal('id', 1))) + native = _native_table(table).new_global_index_drop_builder().with_index_column('name') + files = indexes._files(table) + assert native.with_dry_run(True).execute() == len(files) + assert table.snapshot_manager().get_latest_snapshot().id == before + assert len(native.build()) > 0 + assert native.with_dry_run(False).execute() == len(files) + assert indexes._files(table) == [] + assert all(table.file_io.exists(indexes._path(table, file)) for file in files) + + +@pytest.mark.native_plan +@native_drop +def test_lost_drop_commit_response_preserves_index_files(rest_catalog): + catalog, _ = rest_catalog + table = indexes._create(catalog) + indexes._append(table, indexes.ROWS) + table.create_global_index('name') + files = indexes._files(table) + before = table.snapshot_manager().get_latest_snapshot().id + original = BatchTableCommit.commit + + def publish_then_raise(commit, messages, *args, **kwargs): + original(commit, messages, *args, **kwargs) + raise OSError('commit response lost') + + with patch.object(BatchTableCommit, 'commit', publish_then_raise): + with pytest.raises(OSError, match='commit response lost'): + table.drop_global_index('name') + assert table.snapshot_manager().get_latest_snapshot().id == before + 1 + assert indexes._files(table) == [] + assert all(table.file_io.exists(indexes._path(table, file)) for file in files) + + +@pytest.mark.native_plan +@native_drop +def test_empty_drop_validates_columns_and_publishes_nothing(rest_catalog): + catalog, _ = rest_catalog + table = indexes._create(catalog) + assert GlobalIndexDropper(table, 'name').build() == [] + assert table.drop_global_index('name') == 0 + assert table.drop_global_index('name', dry_run=True) == 0 + assert table.snapshot_manager().get_latest_snapshot() is None + native = _native_table(table).new_global_index_drop_builder() + with pytest.raises(Exception, match='column is required'): + native.build() diff --git a/paimon-python/pypaimon/tests/native_search_plan_test.py b/paimon-python/pypaimon/tests/native_search_plan_test.py new file mode 100644 index 000000000000..5dffe6639f4f --- /dev/null +++ b/paimon-python/pypaimon/tests/native_search_plan_test.py @@ -0,0 +1,430 @@ +# 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. + +"""Explicit search plans keep native work and transportable source metadata.""" + +from contextlib import ExitStack, contextmanager +from copy import copy +import pickle +from unittest.mock import patch + +import pytest +import pyarrow as pa + +from pypaimon.read.native_plan import native_method_available +from pypaimon.table.source.full_text_read import DataEvolutionFullTextRead +from pypaimon.table.source.full_text_scan import DataEvolutionFullTextScan +from pypaimon.table.source.primary_key_full_text_read import PrimaryKeyFullTextRead +from pypaimon.table.source.primary_key_full_text_scan import PrimaryKeyFullTextScan +from pypaimon.table.source.primary_key_vector_read import PrimaryKeyVectorRead +from pypaimon.table.source.primary_key_vector_scan import PrimaryKeyVectorScan +from pypaimon.table.source.vector_search_read import BatchVectorSearchReadImpl, DataEvolutionVectorRead +from pypaimon.table.source.vector_search_scan import DataEvolutionVectorScan +from pypaimon.tests import native_hybrid_search_test as hybrid +from pypaimon.tests import native_pk_full_text_search_test as text +from pypaimon.tests import native_sorted_index_build_test as indexes +from pypaimon.tests import native_vector_search_test as vectors + +rest_catalog = indexes.rest_catalog +golden_catalog = text.golden_catalog +source = text.source +pk_vector_table = vectors.pk_vector_table +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_method_available('VectorScanPlan', '_metadata'), reason='Rust search plan metadata required')] + + +@pytest.fixture +def table(rest_catalog): + table = hybrid._table(rest_catalog[0], options={ + 'write.native.enabled': 'false', 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false', + }, partial=True) + table.create_global_index('label', index_type='btree') + # Count actual table planning/reading in the native CI lanes, in addition + # to the separate search APIs exercised below. + read = table.copy({'read.native.enabled': 'true', 'scan.native-plan.enabled': 'true'}).new_read_builder() + assert read.new_read().to_arrow(read.new_scan().plan().splits()).num_rows == 12 + return table + + +def _builder(table, family, *, native=False, filtered=False, partition=False, limit=3): + table = table.copy({'read.native.enabled': str(native).lower(), + 'scan.native-plan.enabled': str(native).lower()}) + if family == 'full-text': + builder = text._builder(table, text='paimon', limit=limit) + elif family == 'batch': + builder = (table.new_batch_vector_search_builder().with_vector_column('embedding') + .with_query_vectors([[0., 0.], [7., 0.]]).with_limit(limit)) + else: + builder = (table.new_vector_search_builder().with_vector_column('embedding') + .with_query_vector([0., 0.]).with_limit(limit)) + predicate = table.new_read_builder().new_predicate_builder() + if filtered: + builder.with_filter(predicate.equal('label', 'keep')) + if partition: + builder.with_partition_filter(predicate.equal('pt', 1)) + return builder + + +def _scan(builder, family): + return builder.new_full_text_scan() if family == 'full-text' else builder.new_vector_search_scan() + + +def _reader(builder, family): + if family == 'full-text': + return builder.new_full_text_read() + return builder.new_batch_vector_search_read() if family == 'batch' else builder.new_vector_search_read() + + +def _read(reader, family, plan): + return reader.read_batch_plan(plan) if family == 'batch' else [reader.read_plan(plan)] + + +def _scores(result): + if hasattr(result, 'positions'): + return {(p.partition_bytes, p.bucket, p.data_file_name, p.row_position): p.score for p in result.positions} + getter = result.score_getter() if not result.is_empty() else None + return {row_id: getter(row_id) for row_id in result.results()} + + +def _assert_results(actual, expected): + assert len(actual) == len(expected) + for a, b in zip(actual, expected): + assert _scores(a) == pytest.approx(_scores(b), abs=1e-6) + + +def _native_only(): + stack = ExitStack() + for cls in [DataEvolutionVectorScan, DataEvolutionFullTextScan, + PrimaryKeyVectorScan, PrimaryKeyFullTextScan]: + stack.enter_context(patch.object(cls, 'scan', side_effect=AssertionError('Python search scan ran'))) + for cls, method in [(DataEvolutionVectorRead, 'read_plan'), (DataEvolutionFullTextRead, 'read_plan'), + (BatchVectorSearchReadImpl, 'read_batch_plan'), (PrimaryKeyVectorRead, 'read_plan'), + (PrimaryKeyFullTextRead, 'read_plan')]: + stack.enter_context(patch.object(cls, method, side_effect=AssertionError('Python search read ran'))) + return stack + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +@pytest.mark.parametrize('mode', ['fast', 'full', 'detail']) +@pytest.mark.parametrize('filtered,partition', [(False, False), (True, False), (False, True)]) +def test_explicit_native_plan_and_read_avoid_python_search(table, family, mode, filtered, partition): + table = table.copy({'vector-index.search-mode': mode, 'full-text-index.search-mode': mode}) + classic = _builder(table, family, filtered=filtered, partition=partition) + expected_plan = _scan(classic, family).scan() + expected = _read(_reader(classic, family), family, expected_plan) + native = _builder(table, family, native=True, filtered=filtered, partition=partition) + with _native_only(): + plan = _scan(native, family).scan() + actual = _read(_reader(native, family), family, plan) + assert plan.snapshot().id == expected_plan.snapshot().id + _assert_results(actual, expected) + # Ordinary Python readers can consume the exported metadata directly. + _assert_results(_read(_reader(classic, family), family, pickle.loads(pickle.dumps(plan))), expected) + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +@pytest.mark.parametrize('transport', ['pickle-plan', 'pickle-reader', 'split-list']) +def test_transported_plan_uses_captured_snapshot_without_replanning(table, family, transport): + native = _builder(table, family, native=True, limit=20) + plan = _scan(native, family).scan() + reader = _reader(native, family) + expected = _read(reader, family, plan) + snapshot = plan.snapshot().id + captured = pickle.dumps(plan.splits()) + indexes._append(table, hybrid._rows(12, 4), hybrid.SCHEMA) + if transport == 'pickle-plan': + plan = pickle.loads(pickle.dumps(plan)) + elif transport == 'pickle-reader': + reader = pickle.loads(pickle.dumps(reader)) + with patch.object(DataEvolutionVectorScan, 'scan', side_effect=AssertionError('Replanned vector search')), \ + patch.object(DataEvolutionFullTextScan, 'scan', side_effect=AssertionError('Replanned full-text search')): + if transport == 'split-list': + classic = _reader(_builder(table._copy_with_snapshot(plan.snapshot()), family, limit=20), family) + actual = classic.read_batch(plan.splits()) if family == 'batch' else [classic.read(plan.splits())] + else: + actual = _read(reader, family, plan) + assert plan.snapshot().id == snapshot + assert pickle.dumps(plan.splits()) == captured + _assert_results(actual, expected) + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +def test_modified_plan_and_shallow_alias_observe_current_split_selection(table, family): + builder = _builder(table, family, native=True) + plan = _scan(builder, family).scan() + alias = copy(plan) + alias.splits().clear() + results = _read(_reader(builder, family), family, plan) + assert len(results) == (2 if family == 'batch' else 1) + assert all(result.is_empty() for result in results) + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +def test_read_plan_context_errors_do_not_retry_python(table, family): + plan = _scan(_builder(table, family, native=True), family).scan() + reader = _reader(_builder(table, family, native=True, filtered=True), family) + with _native_only(), pytest.raises(Exception, match='same table|same.*filter|different table|different.*filter'): + _read(reader, family, plan) + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +def test_reader_modes_do_not_expand_the_captured_plan(table, family): + fast = table.copy({'vector-index.search-mode': 'fast', 'full-text-index.search-mode': 'fast'}) + native = _builder(fast, family, native=True, limit=20) + plan = _scan(native, family).scan() + read_builder = _builder(table, family, native=True, limit=20) + if family != 'full-text': + read_builder.with_options({'vector-index.search-mode': 'full'}) + reader = _reader(read_builder, family) + transported = pickle.loads(pickle.dumps(plan)) + expected = _read(reader, family, transported) + actual = _read(reader, family, plan) + _assert_results(actual, expected) + assert all(set(_scores(result)) <= set(range(8)) for result in actual) + + +@pytest.mark.parametrize('family', ['vector', 'batch']) +def test_reader_table_options_apply_to_captured_sources(rest_catalog, family): + table = hybrid._table(rest_catalog[0], indexed=False, options={ + 'write.native.enabled': 'false', 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false'}) + scan_builder = _builder(table.copy({'fields.embedding.distance.metric': 'l2'}), family, native=True, limit=2) + plan = _scan(scan_builder, family).scan() + read_builder = _builder(table.copy({'fields.embedding.distance.metric': 'inner_product'}), + family, native=True, limit=2) + if family == 'vector': + read_builder.with_query_vector([1., 0.]) + else: + read_builder.with_query_vectors([[1., 0.], [2., 0.]]) + reader = _reader(read_builder, family) + expected = _read(reader, family, pickle.loads(pickle.dumps(plan))) + _assert_results(_read(reader, family, plan), expected) + assert all(set(_scores(result)) == {6, 7} for result in expected) + + +@pytest.mark.parametrize('family', ['vector', 'batch']) +@pytest.mark.parametrize('metric', [None, 'cosine']) +@pytest.mark.parametrize('index_type', [' IVF-FLAT ', 'lumina', 'lumina-vector-ann']) +def test_raw_native_plans_use_java_configured_indexer_default(rest_catalog, family, metric, index_type): + options = { + 'fields.embedding.index-type': index_type, 'write.native.enabled': 'false', + 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false', + 'fields.other.distance.metric': 'l2', + } + if metric is not None: + options[index_type.strip().lower() + '.distance.metric'] = metric + table = hybrid._table(rest_catalog[0], indexed=False, options=options) + builder = _builder(table, family, native=True, limit=2) + builder.with_options({'fields.other.metric': 'l2'}) + if family == 'vector': + builder.with_query_vector([1., 0.]) + else: + builder.with_query_vectors([[1., 0.], [2., 0.]]) + with _native_only(): + plan = _scan(builder, family).scan() + actual = _read(_reader(builder, family), family, plan) + assert all(split.index_type == index_type.strip().lower() for split in plan.splits()) + expected = _read(_reader(builder, family)._python, family, pickle.loads(pickle.dumps(plan))) + _assert_results(actual, expected) + assert all(set(_scores(result)) == ({6, 7} if metric is None else {1, 2}) for result in actual) + + +def test_native_scan_and_execute_reject_mixed_persisted_vector_formats(table): + table.create_global_index('embedding', index_type='ivf-sq', options={ + 'ivf-sq.dimension': '2', 'ivf-sq.nlist': '1', 'ivf-sq.distance.metric': 'l2'}) + classic = _builder(table, 'vector') + with pytest.raises(ValueError, match='multiple index types'): + classic.new_vector_search_scan().scan() + for family in ('vector', 'batch'): + builder = _builder(table, family, native=True) + with _native_only(), pytest.raises(Exception, match='multiple index types'): + builder.new_vector_search_scan().scan() + with _native_only(), pytest.raises(Exception, match='multiple index types'): + (builder.execute_batch_local() if family == 'batch' else builder.execute_local()) + + +@pytest.mark.parametrize('family', ['vector', 'batch']) +def test_persisted_vector_family_ignores_other_columns_metric(table, family): + assert table.options.options.to_map().get('fields.embedding.index-type') is None + builder = _builder(table, family, native=True, limit=20).with_options({ + 'fields.other.metric': 'inner_product', 'ivf-flat.distance.metric': 'l2'}) + with _native_only(): + plan = _scan(builder, family).scan() + actual = _read(_reader(builder, family), family, plan) + implicit = builder.execute_batch_local() if family == 'batch' else [builder.execute_local()] + expected = _read(_reader(builder, family)._python, family, pickle.loads(pickle.dumps(plan))) + _assert_results(actual, expected) + _assert_results(implicit, expected) + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +def test_empty_native_plans_remain_transportable(rest_catalog, family): + catalog, _ = rest_catalog + table = indexes._create(catalog, schema=hybrid.SCHEMA) + builder = _builder(table, family, native=True) + with _native_only(): + plan = _scan(builder, family).scan() + results = _read(_reader(builder, family), family, plan) + assert plan.snapshot() is None + assert plan.splits() == [] + assert all(result.is_empty() for result in results) + _assert_results(_read(_reader(builder, family), family, pickle.loads(pickle.dumps(plan))), results) + + +def test_vector_scan_does_not_require_query_or_limit(table): + builder = table.copy({'read.native.enabled': 'true'}).new_vector_search_builder().with_vector_column('embedding') + with _native_only(): + plan = builder.new_vector_search_scan().scan() + assert plan.snapshot() is not None and plan.splits() + builder.with_query_vector([0., 0.]).with_limit(3) + with _native_only(): + assert len(builder.new_vector_search_read().read_plan(plan).results()) == 3 + + +def test_full_text_scan_does_not_require_limit(table): + builder = text._builder(table.copy({'read.native.enabled': 'true'}), limit=None) + with _native_only(): + plan = builder.new_full_text_scan().scan() + assert plan.snapshot() is not None and plan.splits() + builder.with_limit(3) + with _native_only(): + assert len(builder.new_full_text_read().read_plan(plan).results()) == 3 + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +def test_native_plan_is_query_independent_and_pickle_keeps_original_handle(table, family): + builder = _builder(table, family, native=True, limit=1) + plan = _scan(builder, family).scan() + pickle.dumps(plan) + assert plan._unchanged_native_plan() is not None + if family == 'full-text': + builder.with_query('content', text.match_query('lake')) + elif family == 'batch': + builder.with_query_vectors([[7., 0.], [0., 0.], [4., 0.]]) + else: + builder.with_query_vector([7., 0.]).with_options({'ivf.nprobe': '1'}) + builder.with_limit(4) + classic = _reader(builder, family)._python + expected = _read(classic, family, pickle.loads(pickle.dumps(plan))) + with _native_only(): + actual = _read(_reader(builder, family), family, plan) + _assert_results(actual, expected) + + +@pytest.mark.parametrize('family', ['vector', 'batch', 'full-text']) +def test_ray_accepts_native_reader_and_transports_captured_sources(table, family): + # Run the real Ray adapters and workers with in-process serialized task + # dispatch. This exercises their reader qualification and metadata contract + # without starting another cluster in the native CI jobs. + from pypaimon.ray import vector_search, batch_vector_search, full_text_search + module, reader_type, execute, method = { + 'vector': (vector_search, vector_search._RayVectorSearchRead, + vector_search._execute_vector_search, 'read_plan'), + 'batch': (batch_vector_search, batch_vector_search._RayBatchVectorSearchRead, + batch_vector_search._execute_batch_vector_search, 'read_batch_plan'), + 'full-text': (full_text_search, full_text_search._RayFullTextRead, + full_text_search._execute_full_text_search, 'read_plan'), + }[family] + classic = _builder(table, family, limit=20) + expected = _read(_reader(classic, family), family, _scan(classic, family).scan()) + original_read = getattr(reader_type, method) + + def read_after_append(reader, plan): + captured = plan.snapshot().id + transported = pickle.loads(pickle.dumps(plan)) + assert plan._unchanged_native_plan() is not None + indexes._append(table, hybrid._rows(12, 4), hybrid.SCHEMA) + assert transported.snapshot().id == captured + return original_read(reader, transported) + + def dispatch(worker, context, items, *args): + for ordinal, item in enumerate(items): + yield ordinal, worker(pickle.loads(pickle.dumps(context)), pickle.loads(pickle.dumps(item))) + + @contextmanager + def scalar_filters(*args): + yield None + + with ExitStack() as stack: + stack.enter_context(patch.object(module, '_require_ray')) + stack.enter_context(patch.object(module, '_map_tasks', dispatch)) + stack.enter_context(patch.object(reader_type, method, read_after_append)) + if family != 'full-text': + stack.enter_context(patch.object(module, '_shared_scalar_filters', scalar_filters)) + for cls in (DataEvolutionVectorScan, DataEvolutionFullTextScan): + stack.enter_context(patch.object(cls, 'scan', side_effect=AssertionError('Python search scan ran'))) + actual = execute(_builder(table, family, native=True, limit=20)) + _assert_results(actual if family == 'batch' else [actual], expected) + + +@pytest.mark.parametrize('family', ['vector', 'full-text']) +@pytest.mark.parametrize('filtered', [False, True]) +def test_pk_native_plans_export_physical_sources_for_python_readers(source, pk_vector_table, family, filtered): + table = source[0] if family == 'full-text' else pk_vector_table + + def builder(native): + value = table.copy({'read.native.enabled': str(native).lower(), 'scan.native-plan.enabled': 'false'}) + query = text._builder(value) if family == 'full-text' else vectors._single(value, [1., 0., 0., 0.], 2) + if filtered and family == 'vector': + query.with_filter(value.new_read_builder().new_predicate_builder().equal('id', 3)) + return query + + classic = builder(False) + expected = _read(_reader(classic, family), family, _scan(classic, family).scan()) + native = builder(True) + with _native_only(): + plan = _scan(native, family).scan() + actual = _read(_reader(native, family), family, plan) + assert plan.snapshot_id == expected[0].snapshot_id + assert plan.snapshot().id == plan.snapshot_id + _assert_results(actual, expected) + transported = pickle.loads(pickle.dumps(plan)) + _assert_results(_read(_reader(classic, family), family, transported), expected) + + +@pytest.mark.parametrize('family', ['vector', 'full-text']) +def test_pk_plan_retains_snapshot_and_sources_after_primary_key_update(source, pk_vector_table, family): + table = source[0] if family == 'full-text' else pk_vector_table + native = table.copy({'read.native.enabled': 'true', 'scan.native-plan.enabled': 'false'}) + builder = text._builder(native) if family == 'full-text' else vectors._single(native, [1., 0., 0., 0.], 2) + plan = _scan(builder, family).scan() + expected = _read(_reader(builder, family), family, plan) + expected_rows = vectors._rows(table, expected[0]).to_pylist() + captured = plan.snapshot_id + if family == 'full-text': + schema = pa.schema([pa.field('id', pa.int32(), False), pa.field('content', pa.string())]) + rows = [{'id': 1, 'content': 'changed after the captured search'}, {'id': 100, 'content': 'new document'}] + else: + schema = pa.schema([pa.field('id', pa.int32(), False), + pa.field('embedding', pa.list_(pa.float32(), 4))]) + rows = [{'id': 1, 'embedding': [0., 0., 0., 1.]}, {'id': 100, 'embedding': [0., 0., 0., 1.]}] + write = table.copy({'write.native.enabled': 'true', 'pk-vector.index.columns': '', + 'pk-full-text.index.columns': ''}).new_batch_write_builder() + writer, commit = write.new_write(), write.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist(rows, schema=schema)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + assert table.snapshot_manager().get_latest_snapshot().id > captured + with _native_only(): + actual = _read(_reader(builder, family), family, plan) + transported = _read(_reader(builder, family), family, pickle.loads(pickle.dumps(plan))) + _assert_results(actual, expected) + _assert_results(transported, expected) + assert vectors._rows(table, actual[0]).to_pylist() == expected_rows + assert vectors._rows(table, transported[0]).to_pylist() == expected_rows diff --git a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py index 76a52899119d..4219d765be95 100644 --- a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py @@ -403,7 +403,7 @@ def test_java_records_per_file_option_overrides_table_alias(rest_catalog, kind): assert _ids(table, table.new_read_builder().new_predicate_builder().equal('name', 'a')) == [1, 5] -@pytest.mark.parametrize('options', [{'write.native.enabled': 'false'}, {'data-evolution.enabled': 'false'}]) +@pytest.mark.parametrize('options', [{'write.native.enabled': 'false'}]) def test_ineligible_tables_choose_python_before_building(rest_catalog, options): catalog, _ = rest_catalog table = _create(catalog, options=options) diff --git a/paimon-python/pypaimon/tests/native_vector_index_build_test.py b/paimon-python/pypaimon/tests/native_vector_index_build_test.py index 9a7216b366ad..363d7461dcc8 100644 --- a/paimon-python/pypaimon/tests/native_vector_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_vector_index_build_test.py @@ -133,6 +133,9 @@ def test_native_search_filters_live_null_vectors_with_deletion_vectors(rest_cata result = (_native_table(table).new_vector_search_builder().with_vector_column('embedding') .with_query_vector([1., 1.]).with_limit(64).with_options(options).execute_local()) assert sorted(result.row_ids()) == [1, 4] + # Python table-level planning also uses the core low-level index reader, + # avoiding the third-party scalar FFI's fixed-result-length restriction. + _assert_ids(table, [1, 4], kind) @pytest.mark.parametrize('kind', VINDEX_IDENTIFIERS) diff --git a/paimon-python/pypaimon/tests/native_vindex_reader_test.py b/paimon-python/pypaimon/tests/native_vindex_reader_test.py new file mode 100644 index 000000000000..ef54e3e726a8 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_vindex_reader_test.py @@ -0,0 +1,215 @@ +# 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. + +"""Python positional I/O delegates vector-index queries and scores to Rust core.""" + +import io +from unittest.mock import patch + +import numpy as np +import pytest + +from pypaimon.globalindex.batch_vector_search import BatchVectorSearch +from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta +from pypaimon.globalindex.vector_search import VectorSearch +from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import ( + PaimonVindexInput, VINDEX_IDENTIFIERS, VindexVectorGlobalIndexReader, +) +from pypaimon.read.native_plan import native_method_available +from pypaimon.utils.roaring_bitmap import RoaringBitmap64 + +pytestmark = pytest.mark.skipif( + not native_method_available('VindexVectorGlobalIndexReader', 'search'), + reason='Rust low-level vector-index reader required') + + +def _native_type(): + from pypaimon_rust.datafusion import VindexVectorGlobalIndexReader + return VindexVectorGlobalIndexReader + + +def _bitmap(ids): + bitmap = RoaringBitmap64() + for row_id in ids: + bitmap.add(row_id) + return bitmap + + +def _options(kind): + return {'diskann.l_search': '64'} if kind == 'diskann' else {'ivf.nprobe': '64'} + + +def _reference_scores(kind, data, queries): + from paimon_vindex import SearchParams, VectorIndexReader + params = SearchParams.diskann(4, 64) if kind == 'diskann' else SearchParams.ivf(4, 64) + input_ = PaimonVindexInput(io.BytesIO(data)) + try: + with VectorIndexReader(input_) as reader: + reader.optimize_for_search() + if len(queries) == 1: + ids, distances = reader.search(queries[0], params) + ids, distances = [ids], [distances] + else: + ids, distances = reader.search_batch(queries, params) + # Query the same encoded index without a sparse filter to avoid the + # FFI length bug. Quantizers may approximate the original distances; + # Java converts their estimates using float32 arithmetic. + return [{int(row_id): float(np.float32(1.) / (np.float32(1.) + distance)) + for row_id, distance in zip(query_ids, query_distances) if row_id >= 0} + for query_ids, query_distances in zip(ids, distances)] + finally: + input_.close() + + +@pytest.fixture(params=VINDEX_IDENTIFIERS) +def index_bytes(request): + vindex = pytest.importorskip('paimon_vindex') + kind = request.param + options = {'index.type': kind.replace('-', '_'), 'dimension': '2', 'metric': 'l2'} + if kind != 'diskann': + options['nlist'] = '1' + if kind == 'ivf-pq': + options['pq.m'] = '1' + if kind == 'diskann': + options['pq.bits'] = '4' + vectors = np.array([[1., 0.], [0., 1.], [1., 1.], [2., 2.]], dtype=np.float32) + output = io.BytesIO() + with vindex.VectorIndexTrainer.train(options, vectors) as training: + with vindex.VectorIndexWriter(training) as writer: + writer.add_vectors(np.array([0, 1, 3, 4], dtype=np.int64), vectors) + writer.write(output) + return kind, output.getvalue() + + +@pytest.mark.parametrize('batch', [False, True]) +def test_sparse_filter_allows_fewer_hits_than_limit(index_bytes, batch): + kind, data = index_bytes + stream = io.BytesIO(data) + input_ = PaimonVindexInput(stream) + reader = _native_type()(input_, 'index', len(data)) + try: + assert reader.metric() == 'l2' + queries = [[1., 1.], [0., 1.]] if batch else [[1., 1.]] + # Row 2 is a live NULL vector. It counts toward the filter's cardinality, + # but cannot yield a hit; deleted rows 0 and 3 must never be returned. + results = reader.search(queries, 64, _bitmap([1, 2, 4]).serialize(), _options(kind)) + reference = _reference_scores(kind, data, queries) + assert [sorted(result) for result in results] == [[1, 4]] * len(queries) + for result, scores in zip(results, reference): + assert result == pytest.approx({row_id: scores[row_id] for row_id in [1, 4]}) + for selected in ([], [2]): + assert reader.search(queries, 64, _bitmap(selected).serialize(), _options(kind)) == [ + None] * len(queries) + # Filters and options belong to a call, rather than leaking across calls. + assert sorted(reader.search([[1., 1.]], 64, options=_options(kind))[0]) == [0, 1, 3, 4] + wrong_family = {'ivf.nprobe': '1'} if kind == 'diskann' else {'diskann.l_search': '64'} + with pytest.raises(ValueError, match='cannot be used'): + reader.search([[1., 1.]], 1, options=wrong_family) + finally: + reader.close() + reader.close() + input_.close() + # The adapter's owner closes its stream, as with the existing Python reader. + assert not stream.closed + stream.close() + with pytest.raises(ValueError, match='closed'): + reader.search([[1., 1.]], 1) + + +@pytest.mark.parametrize('batch', [False, True]) +def test_python_reader_uses_core_without_third_party_search(index_bytes, batch): + kind, data = index_bytes + reference = _reference_scores(kind, data, [[1., 1.]])[0] + + class FileIO: + def __init__(self): + self.stream = io.BytesIO(data) + + def new_input_stream(self, path): + assert path == '/indexes/file' + return self.stream + + file_io = FileIO() + reader = VindexVectorGlobalIndexReader( + file_io, '/indexes', [GlobalIndexIOMeta('file', len(data))]) + with patch('paimon_vindex.VectorIndexReader', side_effect=AssertionError('third-party FFI used')): + try: + assert reader.vector_metric() == 'l2' + include = _bitmap([1, 2, 4]) + if batch: + query = BatchVectorSearch( + [[1., 1.], [0., 1.]], 64, 'embedding', options=_options(kind)) + results = reader.visit_batch_vector_search(query.with_include_row_ids(include)).result() + else: + query = VectorSearch([1., 1.], 64, 'embedding', options=_options(kind)) + results = [reader.visit_vector_search(query.with_include_row_ids(include)).result()] + assert [sorted(result.results()) for result in results] == [[1, 4]] * len(results) + assert results[0].score_getter()(1) == pytest.approx(reference[1]) + finally: + reader.close() + assert file_io.stream.closed + assert reader._reader is None + assert reader._index_input is None + + +def test_core_errors_and_query_option_overrides(index_bytes): + kind, data = index_bytes + input_ = PaimonVindexInput(io.BytesIO(data)) + # An invalid reader default must be superseded by a per-query value. + reader = _native_type()(input_, 'index', len(data), { + 'diskann.l_search' if kind == 'diskann' else 'ivf.nprobe': 'invalid'}) + try: + assert sorted(reader.search([[1., 1.]], 64, options=_options(kind))[0]) == [0, 1, 3, 4] + with pytest.raises(ValueError, match='integer'): + reader.search([[1., 1.]], 1) + with pytest.raises(ValueError, match='dimension'): + reader.search([[1.]], 1, options=_options(kind)) + with pytest.raises(ValueError, match='Limit'): + reader.search([[1., 1.]], 0, options=_options(kind)) + with pytest.raises(ValueError, match='Cannot set both'): + reader.search([[1., 1.]], 1, options={'ivf.nprobe': '1', 'diskann.l_search': '64'}) + with pytest.raises(ValueError): + reader.search([[1., 1.]], 1, filter_bytes=b'invalid') + finally: + reader.close() + input_.close() + + +def test_callback_exception_preserves_python_identity_and_short_reads_fail(): + failure = OSError('index input unavailable') + + class FailingInput: + def pread_many(self, ranges): + raise failure + + with pytest.raises(OSError) as raised: + _native_type()(FailingInput(), 'index', 100) + assert raised.value is failure + + class ShortInput: + def pread_many(self, ranges): + return [b'' for _ in ranges] + + with pytest.raises(ValueError, match='short read'): + _native_type()(ShortInput(), 'index', 100) + + class MissingRange: + def pread_many(self, ranges): + return [] + + with pytest.raises(ValueError, match='number of ranges'): + _native_type()(MissingRange(), 'index', 100) diff --git a/paimon-python/pypaimon/tests/vector_metric_consistency_test.py b/paimon-python/pypaimon/tests/vector_metric_consistency_test.py index bdb29735a383..3989d28768f3 100644 --- a/paimon-python/pypaimon/tests/vector_metric_consistency_test.py +++ b/paimon-python/pypaimon/tests/vector_metric_consistency_test.py @@ -167,13 +167,23 @@ def test_other_columns_do_not_override_persisted_metric(self): self.assertEqual('inner_product', _raw_search_metric( self.table, self.column, {'fields.other.metric': 'l2'}, 'ivf-flat')) - def test_raw_only_defaults_match_vindex_writers(self): - for kind in ('ivf-flat', 'ivf-pq', 'ivf-sq', 'ivf-rq', 'diskann'): + def test_raw_only_defaults_match_index_writers(self): + for kind in ('ivf-flat', 'ivf-pq', 'ivf-sq', 'ivf-rq', 'diskann', 'lumina', 'lumina-vector-ann'): self.assertEqual('inner_product', _raw_search_metric(self.table, self.column, {}, kind)) self.assertEqual('cosine', _raw_search_metric( self.table, self.column, {'metric': 'cosine'}, kind)) self.assertEqual('l2', _raw_search_metric(self.table, self.column, {})) - self.assertEqual('l2', _raw_search_metric(self.table, self.column, {}, 'lumina')) + + def test_known_families_ignore_other_columns_metric(self): + for kind in ('ivf-flat', 'lumina', 'lumina-vector-ann'): + with self.subTest(index_type=kind): + self.table.options.options.to_map()['fields.other.metric'] = 'l2' + self.assertEqual('inner_product', _raw_search_metric( + self.table, self.column, {'fields.other.metric': 'cosine'}, kind)) + self.assertEqual('cosine', _raw_search_metric( + self.table, self.column, {kind + '.distance.metric': 'cosine'}, kind)) + self.assertEqual('l2', _raw_search_metric( + self.table, self.column, {'fields.embedding.distance.metric': 'l2'}, kind)) def test_reader_closes_on_metadata_and_metric_errors(self): entry = _entry(None, field_id=1, index_type='ivf-flat', file_name='vectors.index', diff --git a/paimon-python/pypaimon/tests/vindex_input_test.py b/paimon-python/pypaimon/tests/vindex_input_test.py index 8e5935425b77..9e62672cb75a 100644 --- a/paimon-python/pypaimon/tests/vindex_input_test.py +++ b/paimon-python/pypaimon/tests/vindex_input_test.py @@ -211,7 +211,9 @@ def open_reader(input_): native.close.side_effect = error module = mock.Mock() module.VectorIndexReader.side_effect = open_reader - with mock.patch.dict("sys.modules", {"paimon_vindex": module}): + with mock.patch.dict("sys.modules", {"paimon_vindex": module}), mock.patch( + 'pypaimon.globalindex.vindex.vindex_vector_global_index_reader._native_reader_type', + return_value=None): reader = VindexVectorGlobalIndexReader( io_, "s3://bucket", [GlobalIndexIOMeta(file_name="index", file_size=2)]) if phase in ("constructor", "initialize"): diff --git a/paimon-python/pypaimon/tests/vindex_vector_index_test.py b/paimon-python/pypaimon/tests/vindex_vector_index_test.py index 76715d253f72..b68052f4ad47 100644 --- a/paimon-python/pypaimon/tests/vindex_vector_index_test.py +++ b/paimon-python/pypaimon/tests/vindex_vector_index_test.py @@ -19,6 +19,7 @@ import sys import types import unittest +from unittest.mock import patch from pypaimon.globalindex.batch_vector_search import BatchVectorSearch from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta @@ -26,6 +27,7 @@ from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import ( VindexVectorGlobalIndexReader, _build_scores, + _search_params, ) from pypaimon.utils.roaring_bitmap import RoaringBitmap64 @@ -109,6 +111,16 @@ def new_input_stream(self, path): class VindexVectorIndexTest(unittest.TestCase): + def test_query_widths_match_java_integer_text_and_bounds(self): + for key in ('ivf.nprobe', 'diskann.l_search'): + for value, expected in (('+12', 12), ('１２', 12), ('٢', 2), ('2147483647', 2147483647), (12, 12)): + with self.subTest(key=key, value=value): + self.assertEqual(expected, _search_params(_FakeSearchParams, {key: value}, 4).width) + for value in ('2147483648', '-2147483649', ' 2', '2 ', '2.0', '1_0', '𝟙', '+', '', 1.5, True): + with self.subTest(key=key, value=value): + with self.assertRaisesRegex(ValueError, 'integer'): + _search_params(_FakeSearchParams, {key: value}, 4) + def test_inner_product_distance_converted_to_higher_is_better_score(self): id_to_scores = _build_scores( [10, 20, 30], @@ -130,7 +142,9 @@ def test_top_k_breaks_boundary_ties_by_row_id(self): DictBasedScoredIndexResult(scores).top_k(2).results().to_list(), ) - def test_batch_search_uses_native_batch_api(self): + @patch('pypaimon.globalindex.vindex.vindex_vector_global_index_reader._native_reader_type', + return_value=None) + def test_batch_search_uses_native_batch_api(self, _native_reader): old_module = sys.modules.get("paimon_vindex") sys.modules["paimon_vindex"] = types.SimpleNamespace( SearchParams=_FakeSearchParams, From b1caec921dc7ef2754fe23f2b48b4fbb28c406fe Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 09:45:14 +0800 Subject: [PATCH 3/4] [python] Align global index unions with Java and fix CI --- .../data_evolution_global_index_scanner.py | 102 ++---------------- .../tests/vector_filter_exactness_test.py | 3 - .../tests/vector_search_filter_test.py | 63 ++++++----- 3 files changed, 36 insertions(+), 132 deletions(-) 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 c5efd1535c67..bfb54354f8a4 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -25,7 +25,7 @@ GlobalIndexEvaluator, ) from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta -from pypaimon.globalindex.global_index_reader import GlobalIndexReader, _map_future +from pypaimon.globalindex.global_index_reader import GlobalIndexReader from pypaimon.globalindex.global_index_result import GlobalIndexResult from pypaimon.common.options.core_options import CoreOptions from pypaimon.common.options.options import Options @@ -126,26 +126,13 @@ def readers_function(field: DataField) -> Collection[GlobalIndexReader]: ) if not groups: return [] - if len(groups) == 1: - # Families may cover different row ranges. Like Java, union - # their matches before the evaluator combines field predicates. - return [UnionGlobalIndexReader(_create_readers( - file_io, index_path, groups[0].metas, field, executor, options))] - union_coverage = Range.sort_and_merge_overlap( - [ - range_key - for group in groups - for range_key in group.coverage_ranges - ], - True, - ) + # Like Java, union every reader for the field. Missing coverage + # belongs to raw fallback, not to the matches returned by an index. readers = [] for group in groups: - pad_ranges = _exclude_ranges(union_coverage, group.coverage_ranges) readers.extend( _create_readers( - file_io, index_path, group.metas, field, executor, - options, pad_ranges=pad_ranges)) + file_io, index_path, group.metas, field, executor, options)) return [UnionGlobalIndexReader(readers)] return GlobalIndexEvaluator(fields, readers_function) @@ -276,76 +263,11 @@ def __init__(self, index_field_id, field_ids): self.index_field_id = index_field_id self.field_ids = tuple(field_ids) self.metas = {} - self.coverage_ranges = [] def add_file(self, index_type, range_key, io_meta): - self.coverage_ranges.append(range_key) self.metas.setdefault(index_type, {}).setdefault(range_key, []).append(io_meta) -class _PaddingGlobalIndexReader(GlobalIndexReader): - def __init__(self, wrapped, padding): - self._wrapped = wrapped - # Padding rows have not been tested by this index. - self._padding = GlobalIndexResult.create(padding.results(), is_exact=False) - - def _pad(self, future): - return _map_future( - future, - lambda result: None if result is None else result.or_(self._padding)) - - def visit_equal(self, field_ref, literal): - return self._pad(self._wrapped.visit_equal(field_ref, literal)) - - def visit_not_equal(self, field_ref, literal): - return self._pad(self._wrapped.visit_not_equal(field_ref, literal)) - - def visit_less_than(self, field_ref, literal): - return self._pad(self._wrapped.visit_less_than(field_ref, literal)) - - def visit_less_or_equal(self, field_ref, literal): - return self._pad(self._wrapped.visit_less_or_equal(field_ref, literal)) - - def visit_greater_than(self, field_ref, literal): - return self._pad(self._wrapped.visit_greater_than(field_ref, literal)) - - def visit_greater_or_equal(self, field_ref, literal): - return self._pad(self._wrapped.visit_greater_or_equal(field_ref, literal)) - - def visit_is_null(self, field_ref): - return self._pad(self._wrapped.visit_is_null(field_ref)) - - def visit_is_not_null(self, field_ref): - return self._pad(self._wrapped.visit_is_not_null(field_ref)) - - def visit_in(self, field_ref, literals): - return self._pad(self._wrapped.visit_in(field_ref, literals)) - - def visit_not_in(self, field_ref, literals): - return self._pad(self._wrapped.visit_not_in(field_ref, literals)) - - def visit_starts_with(self, field_ref, literal): - return self._pad(self._wrapped.visit_starts_with(field_ref, literal)) - - def visit_ends_with(self, field_ref, literal): - return self._pad(self._wrapped.visit_ends_with(field_ref, literal)) - - def visit_contains(self, field_ref, literal): - return self._pad(self._wrapped.visit_contains(field_ref, literal)) - - def visit_like(self, field_ref, literal): - return self._pad(self._wrapped.visit_like(field_ref, literal)) - - def visit_between(self, field_ref, from_v, to_v): - return self._pad(self._wrapped.visit_between(field_ref, from_v, to_v)) - - def visit_not_between(self, field_ref, from_v, to_v): - return self._pad(self._wrapped.visit_not_between(field_ref, from_v, to_v)) - - def close(self): - self._wrapped.close() - - def _resolve_snapshot(table, snapshot): if snapshot is not None: return snapshot @@ -396,17 +318,9 @@ def _core_options(table): return options -def _exclude_ranges(base_ranges, excluded_ranges): - excluded_ranges = Range.sort_and_merge_overlap(excluded_ranges, True) - result = [] - for base_range in Range.sort_and_merge_overlap(base_ranges, True): - result.extend(base_range.exclude(excluded_ranges)) - return Range.sort_and_merge_overlap(result, True) - - def _create_readers( file_io, index_path, index_type_metas, field, executor=None, - options=None, pad_ranges=None): + options=None): """Create readers for a specific field, dispatched by index_type. Unknown indexTypes raise — a silent skip would make @@ -434,11 +348,7 @@ def _create_readers( OffsetGlobalIndexReader( inner, range_key.from_, range_key.to)) if offset_readers: - reader = UnionGlobalIndexReader(offset_readers) - if pad_ranges: - padding = GlobalIndexResult.from_ranges(pad_ranges) - reader = _PaddingGlobalIndexReader(reader, padding) - readers.append(reader) + readers.append(UnionGlobalIndexReader(offset_readers)) return readers diff --git a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py index 86e1e78c91ff..923de8133fd1 100644 --- a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py +++ b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py @@ -22,7 +22,6 @@ import pypaimon.multimodal as pm from pypaimon.common.predicate import Predicate -from pypaimon.globalindex.data_evolution_global_index_scanner import _PaddingGlobalIndexReader from pypaimon.globalindex.global_index_evaluator import GlobalIndexEvaluator from pypaimon.globalindex.global_index_reader import FieldRef, _completed_future from pypaimon.globalindex.global_index_result import GlobalIndexResult @@ -178,8 +177,6 @@ def test_exactness_follows_predicate_tree_and_reader_wrappers(method): assert candidate.and_(GlobalIndexResult.create_empty()).is_exact() reader = UnionGlobalIndexReader([OffsetGlobalIndexReader(StubGlobalIndexReader(candidate), 10, 13)]) assert not reader.visit_equal(FieldRef(0, "a", "INT"), 1).result().is_exact() - padded = _PaddingGlobalIndexReader(StubGlobalIndexReader(exact), GlobalIndexResult.from_range(Range(5, 6))) - assert not padded.visit_equal(FieldRef(0, "a", "INT"), 1).result().is_exact() class PartialReader(StubGlobalIndexReader): def visit_greater_than(self, field_ref, literal): diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 578f252b8399..b64625dc273a 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -1950,29 +1950,25 @@ def close(self_inner): has_nulls=False) with mock.patch( - "pypaimon.globalindex.data_evolution_global_index_scanner." - "_exclude_ranges", - side_effect=AssertionError("single group must not compute padding")): + "pypaimon.globalindex.btree.lazy_filtered_btree_reader." + "BTreeIndexReader", + _StubBTreeReader): with mock.patch( - "pypaimon.globalindex.btree.lazy_filtered_btree_reader." - "BTreeIndexReader", - _StubBTreeReader): - with mock.patch( - "pypaimon.globalindex.sorted_file_global_index_reader." - "SortedIndexFileMeta.deserialize", - return_value=wide_meta): - scanner = DataEvolutionGlobalIndexScanner( - fields=table.fields, - file_io=table.file_io, - index_path="/unused", - index_files=[shard_a, shard_b], - ) - try: - result = scanner.scan( - Predicate(method="equal", index=0, field="id", - literals=[7])) - finally: - scanner.close() + "pypaimon.globalindex.sorted_file_global_index_reader." + "SortedIndexFileMeta.deserialize", + return_value=wide_meta): + scanner = DataEvolutionGlobalIndexScanner( + fields=table.fields, + file_io=table.file_io, + index_path="/unused", + index_files=[shard_a, shard_b], + ) + try: + result = scanner.scan( + Predicate(method="equal", index=0, field="id", + literals=[7])) + finally: + scanner.close() self.assertIsNotNone(result) hits = sorted(list(result.results())) @@ -2107,7 +2103,7 @@ def _stub_create_inner_readers( self.assertEqual([1], sorted(evaluation.result.results())) self.assertEqual([Range(5, 9)], fallback) - def test_extra_field_groups_are_padded_before_and(self): + def test_extra_field_groups_union_matches_without_padding(self): from pypaimon.globalindex.global_index_reader import GlobalIndexReader from pypaimon.globalindex.data_evolution_global_index_scanner import ( DataEvolutionGlobalIndexScanner, @@ -2117,7 +2113,7 @@ def test_extra_field_groups_are_padded_before_and(self): b_field = _field(1, "b") c_field = _field(2, "c") - # Mocked non-BTree readers retain coverage of generic extra-field padding. + # Mocked non-BTree readers exercise shared extra-field reader grouping. short = _entry(None, field_id=0, index_type="bitmap", file_name="a-c.index", row_range_start=0, row_range_end=4).index_file @@ -2166,12 +2162,12 @@ def _stub_create_inner_readers( scanner.close() self.assertIsNotNone(result) - # The short group is all-hit padded for rows 5..9 before AND-ing with - # the long group. Row 4 is filtered out by the long group, while tail - # rows 7 and 8 survive because the short group has not indexed them. - self.assertEqual([1, 3, 7, 8], sorted(list(result.results()))) + # Java unions both groups: row 4 survives from the short group. Rows + # 5, 6 and 9 were not returned by either index and must not be added. + self.assertEqual([1, 3, 4, 7, 8], sorted(list(result.results()))) + self.assertTrue(result.is_exact()) - def test_extra_field_groups_pad_missing_coverage_before_and(self): + def test_extra_field_groups_do_not_turn_missing_coverage_into_hits(self): from pypaimon.globalindex.global_index_reader import GlobalIndexReader from pypaimon.globalindex.data_evolution_global_index_scanner import ( DataEvolutionGlobalIndexScanner, @@ -2181,7 +2177,7 @@ def test_extra_field_groups_pad_missing_coverage_before_and(self): b_field = _field(1, "b") c_field = _field(2, "c") - # Reader stubs test padding independently of composite BTree support. + # Reader stubs exercise disjoint coverage independently of BTree support. a_early = _entry(None, field_id=0, index_type="bitmap", file_name="a-c-early.index", row_range_start=2, row_range_end=3).index_file @@ -2236,11 +2232,12 @@ def _stub_create_inner_readers( scanner.close() self.assertIsNotNone(result) - # The first group has not indexed [0,1] and [4,6], so these ranges are - # neutral under AND. Its indexed ranges still filter normally. + # Coverage gaps in the first group are not matches. Only real hits + # from either group participate in the union, as in Java. self.assertEqual([1, 2, 5, 7, 8], sorted(list(result.results()))) + self.assertTrue(result.is_exact()) - def test_extra_field_padding_does_not_convert_none_to_hits(self): + def test_extra_field_groups_preserve_unknown_results(self): from pypaimon.globalindex.global_index_reader import GlobalIndexReader from pypaimon.globalindex.data_evolution_global_index_scanner import ( DataEvolutionGlobalIndexScanner, From 9e7facf9c3d115f92a1fffc9a5f4b5f56a47ed83 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 11:41:14 +0800 Subject: [PATCH 4/4] [python] Guard native vector qualification for schema-only tables --- .../pypaimon/table/source/native_vector_search.py | 2 +- .../tests/primary_key_index_definitions_test.py | 13 +++++++++---- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/table/source/native_vector_search.py b/paimon-python/pypaimon/table/source/native_vector_search.py index 56ca2e626ba1..650a830148b1 100644 --- a/paimon-python/pypaimon/table/source/native_vector_search.py +++ b/paimon-python/pypaimon/table/source/native_vector_search.py @@ -33,7 +33,7 @@ def _native_vector_builder(builder, batch=False): """ table = builder._table reject_search_under_query_auth(table) - native_enabled = getattr(table.options, 'native_read_enabled', None) + native_enabled = getattr(getattr(table, 'options', None), 'native_read_enabled', None) if not callable(native_enabled) or not native_enabled(): return None loader = getattr(getattr(table, 'catalog_environment', None), 'catalog_loader', None) 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 7e345f709413..f430c2558b6e 100644 --- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py +++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py @@ -111,10 +111,15 @@ def test_builders_route_primary_key_searches(self): .with_query("content", "paimon") .with_limit(2)) - self.assertIsInstance(vector.new_vector_search_scan(), PrimaryKeyVectorScan) - self.assertIsInstance(vector.new_vector_search_read(), PrimaryKeyVectorRead) - self.assertIsInstance(full_text.new_full_text_scan(), PrimaryKeyFullTextScan) - self.assertIsInstance(full_text.new_full_text_read(), PrimaryKeyFullTextRead) + # Exercise native qualification even without an installed binding. A + # schema-only table cannot delegate, but still selects the PK readers. + with mock.patch( + "pypaimon.table.source.native_search_plan.native_method_available", + return_value=True): + self.assertIsInstance(vector.new_vector_search_scan(), PrimaryKeyVectorScan) + self.assertIsInstance(vector.new_vector_search_read(), PrimaryKeyVectorRead) + self.assertIsInstance(full_text.new_full_text_scan(), PrimaryKeyFullTextScan) + self.assertIsInstance(full_text.new_full_text_read(), PrimaryKeyFullTextRead) def test_primary_key_full_text_only_supports_fast_mode(self): fields = [DataField(0, "id", AtomicType("INT")),