This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ab97a37e43 [python] Handle unsupported composite BTree indexes safely
(#10352)
ab97a37e43 is described below
commit ab97a37e43e0c7d13a44f48894dda8f24c4e63fe
Author: QuakeWang <[email protected]>
AuthorDate: Mon Oct 5 21:13:40 2026 +0800
[python] Handle unsupported composite BTree indexes safely (#10352)
---
.../data_evolution_global_index_scanner.py | 5 +-
.../tests/composite_btree_fallback_test.py | 182 +++++++++++++++++++++
.../pypaimon/tests/vector_search_filter_test.py | 103 +++++++-----
3 files changed, 246 insertions(+), 44 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 1dbbfb0154..1174df8919 100644
--- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py
+++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py
@@ -371,9 +371,12 @@ def _resolve_snapshot(table, snapshot):
def is_supported_scalar_index(index_file):
+ meta = index_file.global_index_meta
return (
- index_file.global_index_meta is not None
+ meta is not None
and index_file.index_type in _SUPPORTED_SCALAR_INDEX_TYPES
+ # Composite BTree keys cannot use scalar readers or their coverage.
+ and not (index_file.index_type == 'btree' and meta.extra_field_ids)
)
diff --git a/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
b/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
new file mode 100644
index 0000000000..d94aab1cf4
--- /dev/null
+++ b/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
@@ -0,0 +1,182 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+from dataclasses import replace
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.globalindex.data_evolution_global_index_scanner import (
+ DataEvolutionGlobalIndexScanner,
+ is_supported_scalar_index,
+)
+from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.read.scanner.file_scanner import FileScanner
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.tests.data_evolution_test_helpers import (
+ BatchModeMixin,
+ DataEvolutionTestBase,
+)
+from pypaimon.utils.range import Range
+
+
+def _composite_file():
+ # Java compacted keys (NULL, NULL) and (9, 1). Component nulls do not
+ # set the scalar has_nulls flag; treating this as a scalar index loses
rows.
+ metadata = SortedIndexFileMeta(
+ b'\x00\x03', b'\x00\x00\x09\x00\x00\x00\x01\x00\x00\x00', False)
+ return IndexFileMeta(
+ index_type='btree', file_name='unsupported-composite',
+ file_size=259, row_count=10,
+ global_index_meta=GlobalIndexMeta(0, 9, 0, [1], metadata.serialize()),
+ )
+
+
+class CompositeBTreeFallbackTest(
+ BatchModeMixin, DataEvolutionTestBase, unittest.TestCase):
+
+ pa_schema = pa.schema([('a', pa.int32()), ('b', pa.int32())])
+ table_options = {
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'global-index.enabled': 'true',
+ 'bucket': '-1',
+ 'file.format': 'parquet',
+ }
+
+ def _table_with_partial_index(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table({
+ 'a': [None, 1, 2, 3, 4], 'b': [None, 1, 0, 1, 0],
+ }, schema=self.pa_schema))
+ table.create_global_index('a')
+ self._write_arrow(table, pa.table({
+ 'a': [5, 6, 7, 8, 9], 'b': [1, 0, 1, 0, 1],
+ }, schema=self.pa_schema))
+ return table
+
+ def test_scalar_candidate_types(self):
+ composite = _composite_file()
+ self.assertFalse(is_supported_scalar_index(composite))
+ for extra_fields in (None, []):
+ single = replace(composite, global_index_meta=replace(
+ composite.global_index_meta, extra_field_ids=extra_fields))
+ self.assertTrue(is_supported_scalar_index(single))
+ self.assertTrue(is_supported_scalar_index(replace(single,
index_type='bitmap')))
+ self.assertFalse(is_supported_scalar_index(replace(composite,
global_index_meta=None)))
+ self.assertFalse(is_supported_scalar_index(replace(composite,
index_type='unknown')))
+ # Keep the existing policy for other index types unchanged.
+ self.assertTrue(is_supported_scalar_index(replace(composite,
index_type='bitmap')))
+
+ def test_explicit_files_and_constructor_share_usable_coverage(self):
+ table =
self._table_with_partial_index().copy({'scalar-index.search-mode': 'full'})
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ files = [e.index_file for e in IndexFileHandler(table).scan(snapshot)]
+ composite = _composite_file()
+ predicate =
table.new_read_builder().new_predicate_builder().equal('a', 2)
+ self.assertIsNone(DataEvolutionGlobalIndexScanner.create(
+ table, index_files=[composite], snapshot=snapshot))
+ for ordered in (files + [composite], [composite] + files):
+ for use_factory in (True, False):
+ with self.subTest(factory=use_factory,
composite_first=ordered[0] is composite):
+ if use_factory:
+ scanner = DataEvolutionGlobalIndexScanner.create(
+ table, index_files=ordered, snapshot=snapshot)
+ else:
+ scanner = DataEvolutionGlobalIndexScanner(
+ table.fields, table.file_io,
+
table.path_factory().global_index_path_factory().index_path(),
+ ordered, options=table.options, table=table,
snapshot=snapshot)
+ with scanner:
+ evaluation = scanner.scan_with_coverage(predicate)
+ self.assertEqual([Range(2, 2)],
evaluation.result.results().to_range_list())
+ self.assertEqual(frozenset([0]),
evaluation.contributing_field_ids)
+ self.assertEqual([Range(5, 9)],
scanner.unindexed_ranges(
+ predicate,
contributing_field_ids=evaluation.contributing_field_ids))
+
+ @pytest.mark.python_plan
+ def test_queries_preserve_fallback_and_fast_mode(self):
+ table = self._table_with_partial_index()
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ scalar_entries = IndexFileHandler(table).scan(snapshot)
+ composite_entry = IndexManifestEntry(0, GenericRow([], []), 0,
_composite_file())
+ b = table.new_read_builder().new_predicate_builder()
+ cases = [
+ (b.equal('a', 2), True, [2]),
+ (b.equal('a', 7), True, []),
+ (b.greater_or_equal('a', 2), True, [2, 3, 4]),
+ (b.is_in('a', [2, 7]), True, [2]),
+ (b.is_null('a'), True, [None]),
+ (b.equal('b', 1), False, [1, 3, 5, 7, 9]),
+ (b.is_null('b'), False, [None]),
+ (PredicateBuilder.and_predicates([
+ b.greater_or_equal('a', 2), b.equal('b', 1)]), True, [3]),
+ (PredicateBuilder.or_predicates([
+ b.equal('a', 2), b.equal('b', 1)]), False, [1, 2, 3, 5, 7, 9]),
+ (PredicateBuilder.and_predicates([
+ b.greater_or_equal('a', 2), PredicateBuilder.or_predicates([
+ b.equal('a', 7), b.equal('b', 1)])]), True, [3]),
+ ]
+ original_eval = FileScanner._eval_global_index
+ for predicate, uses_scalar, fast_expected in cases:
+ baseline = self._query(table.copy({'global-index.enabled':
'false'}), predicate)
+ for coexist in (False, True):
+ entries = (scalar_entries if coexist else []) +
[composite_entry]
+ for reverse in (False, True):
+ ordered = list(reversed(entries)) if reverse else entries
+
+ # Model Java-produced manifests without changing Python's
write validation.
+ def scan_indexes(handler, snapshot, entry_filter=None):
+ return [e for e in ordered if entry_filter is None or
entry_filter(e)]
+
+ for mode in ('fast', 'full', 'detail'):
+ with self.subTest(predicate=predicate, coexist=coexist,
+ reverse=reverse, mode=mode):
+ plans = []
+
+ def observe(scanner, snapshot=None):
+ plan = original_eval(scanner, snapshot)
+ plans.append(plan)
+ return plan
+
+ view = table.copy({'scalar-index.search-mode':
mode})
+ with patch.object(IndexFileHandler, 'scan',
scan_indexes), patch.object(
+ FileScanner, '_eval_global_index',
observe):
+ actual = self._query(view, predicate)
+ indexed = coexist and uses_scalar
+ expected = fast_expected if mode == 'fast' and
indexed else baseline
+ self.assertEqual(expected, actual)
+ self.assertEqual(1, len(plans))
+ if indexed:
+ self.assertIsNotNone(plans[0])
+ self.assertEqual([] if mode == 'fast' else
[Range(5, 9)],
+ plans[0].unindexed_ranges)
+ else:
+ self.assertIsNone(plans[0])
+
+ @staticmethod
+ def _query(table, predicate):
+ builder = table.new_read_builder().with_filter(predicate)
+ rows = builder.new_read().to_arrow(builder.new_scan().plan().splits())
+ return sorted(rows.column('a').to_pylist(), key=lambda value: (value
is not None, value))
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py
b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 837e792ea7..f707189fae 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -1051,6 +1051,43 @@ class VectorSearchFilterTest(unittest.TestCase):
self.assertEqual([Range(0, 9)], raw[0].row_ranges)
self.assertEqual([], raw[0].scalar_index_files)
+ def test_composite_btree_does_not_cover_scalar_filter(self):
+ from pypaimon.table.source.vector_search_split import (
+ IndexVectorSearchSplit,
+ RawVectorSearchSplit,
+ )
+
+ self.table.fields.append(_field(2, "other"))
+ self.entries[2].index_file.global_index_meta.row_range_end = 4
+ composite = _entry(
+ None, field_id=0, index_type="btree", file_name="composite.index",
+ row_range_start=0, row_range_end=9)
+ composite.index_file.global_index_meta.extra_field_ids = [2]
+ self.entries.append(composite)
+ predicate = Predicate(method="equal", index=0, field="id",
literals=[7])
+
+ for mode in ("fast", "full", "detail"):
+ with self.subTest(mode=mode):
+ self.table.options = CoreOptions(Options({
+ "scalar-index.search-mode": mode,
+ "vector-index.search-mode": "fast",
+ }))
+
self.table.data_ranges_for_data_evolution_global_index_coverage = (
+ lambda snapshot, partition_filter: [Range(0, 9)])
+ splits =
self._builder(predicate).new_vector_search_scan().scan().splits()
+ indexed = [s for s in splits if isinstance(s,
IndexVectorSearchSplit)]
+ raw = [s for s in splits if isinstance(s,
RawVectorSearchSplit)]
+ self.assertEqual(2, len(indexed))
+ attached = [f.file_name for s in indexed for f in
s.scalar_index_files]
+ self.assertEqual(["id-btree-0.index"], attached)
+ if mode == "fast":
+ self.assertEqual([], raw)
+ else:
+ self.assertEqual(1, len(raw))
+ self.assertEqual([Range(5, 9)], raw[0].row_ranges)
+ self.assertNotIn("composite.index",
+ [f.file_name for f in
raw[0].scalar_index_files])
+
def test_read_threads_prefilter_bitmap_as_include_row_ids(self):
"""preFilter bitmap from scanner.scan(filter) must reach each split's
VectorSearch, offset-rebased to local coords by
OffsetGlobalIndexReader.
@@ -1766,7 +1803,7 @@ class VectorSearchFilterTest(unittest.TestCase):
self.assertEqual(1, len(raw))
self.assertEqual("ivf-flat", raw[0].index_type)
- def test_scan_attaches_scalar_index_when_filter_hits_extra_field(self):
+ def test_scan_excludes_composite_btree_when_filter_hits_extra_field(self):
id_name_index = _entry(None, field_id=2, index_type="btree",
file_name="name-id.index",
row_range_start=0,
@@ -1797,8 +1834,7 @@ class VectorSearchFilterTest(unittest.TestCase):
.splits()
)
- self.assertEqual(["name-id.index"],
- [f.file_name for f in splits[0].scalar_index_files])
+ self.assertEqual([], splits[0].scalar_index_files)
class VectorSearchMultiShardScalarTest(unittest.TestCase):
@@ -1886,7 +1922,7 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
# Must not be empty despite shard_a being empty (no short-circuit).
self.assertEqual([7], hits)
- def test_primary_and_extra_field_indexes_share_coverage(self):
+ def test_composite_btree_does_not_extend_primary_coverage(self):
from pypaimon.globalindex.global_index_reader import GlobalIndexReader
from pypaimon.globalindex.data_evolution_global_index_scanner import (
DataEvolutionGlobalIndexScanner,
@@ -1948,9 +1984,8 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
finally:
scanner.close()
- result = evaluation.result.or_(fallback)
- self.assertTrue(fallback.results().is_empty())
- self.assertEqual([1, 7], sorted(result.results()))
+ self.assertEqual([Range(5, 9)], fallback.results().to_range_list())
+ self.assertEqual([1], sorted(evaluation.result.results()))
def test_unsupported_extra_field_index_does_not_poison_primary(self):
from pypaimon.globalindex.global_index_reader import GlobalIndexReader
@@ -2024,11 +2059,12 @@ class
VectorSearchMultiShardScalarTest(unittest.TestCase):
b_field = _field(1, "b")
c_field = _field(2, "c")
- short = _entry(None, field_id=0, index_type="btree",
+ # Mocked non-BTree readers retain coverage of generic extra-field
padding.
+ short = _entry(None, field_id=0, index_type="bitmap",
file_name="a-c.index",
row_range_start=0, row_range_end=4).index_file
short.global_index_meta.extra_field_ids = [2]
- long = _entry(None, field_id=1, index_type="btree",
+ long = _entry(None, field_id=1, index_type="bitmap",
file_name="b-c.index",
row_range_start=0, row_range_end=9).index_file
long.global_index_meta.extra_field_ids = [2]
@@ -2087,15 +2123,16 @@ class
VectorSearchMultiShardScalarTest(unittest.TestCase):
b_field = _field(1, "b")
c_field = _field(2, "c")
- a_early = _entry(None, field_id=0, index_type="btree",
+ # Reader stubs test padding independently of composite 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
a_early.global_index_meta.extra_field_ids = [2]
- a_late = _entry(None, field_id=0, index_type="btree",
+ a_late = _entry(None, field_id=0, index_type="bitmap",
file_name="a-c-late.index",
row_range_start=7, row_range_end=9).index_file
a_late.global_index_meta.extra_field_ids = [2]
- b_full = _entry(None, field_id=1, index_type="btree",
+ b_full = _entry(None, field_id=1, index_type="bitmap",
file_name="b-c-full.index",
row_range_start=0, row_range_end=9).index_file
b_full.global_index_meta.extra_field_ids = [2]
@@ -2155,11 +2192,11 @@ class
VectorSearchMultiShardScalarTest(unittest.TestCase):
b_field = _field(1, "b", "STRING")
c_field = _field(2, "c", "STRING")
- short = _entry(None, field_id=0, index_type="btree",
+ short = _entry(None, field_id=0, index_type="bitmap",
file_name="a-c.index",
row_range_start=0, row_range_end=4).index_file
short.global_index_meta.extra_field_ids = [2]
- long = _entry(None, field_id=1, index_type="btree",
+ long = _entry(None, field_id=1, index_type="bitmap",
file_name="b-c.index",
row_range_start=0, row_range_end=9).index_file
long.global_index_meta.extra_field_ids = [2]
@@ -2337,7 +2374,7 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
self.assertEqual([Range(5, 9)], result.results().to_range_list())
- def test_scanner_create_selects_extra_field_indexes(self):
+ def test_scanner_create_excludes_composite_btree_extra_fields(self):
from pypaimon.globalindex.data_evolution_global_index_scanner import (
DataEvolutionGlobalIndexScanner,
)
@@ -2358,35 +2395,15 @@ class
VectorSearchMultiShardScalarTest(unittest.TestCase):
)
_patch_snapshot(self, table._entries)
- class _StubBTreeReader:
- def __init__(self_inner, key_serializer, file_io, index_path,
- io_meta):
- pass
-
- def visit_equal(self_inner, literal):
- return GlobalIndexResult.create_empty()
-
- def close(self_inner):
- pass
-
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=BTreeIndexMeta(first_key=b'',
last_key=b'zzzz', has_nulls=False)):
- scanner = DataEvolutionGlobalIndexScanner.create(
- table,
- predicate=Predicate(method="equal", index=1, field="id",
- literals=[3]),
- )
- try:
- self.assertIsNotNone(scanner)
- readers = scanner._evaluator._readers_function(id_field)
- self.assertTrue(readers)
- finally:
- if scanner is not None:
- scanner.close()
+ "pypaimon.globalindex.data_evolution_global_index_scanner."
+ "_create_inner_readers") as create_readers:
+ scanner = DataEvolutionGlobalIndexScanner.create(
+ table,
+ predicate=Predicate(method="equal", index=1, field="id",
literals=[3]),
+ )
+ self.assertIsNone(scanner)
+ create_readers.assert_not_called()
class VectorSearchPartitionedFilterTest(unittest.TestCase):
