From f480bad74e23a097348d80f175e33f9ab731f2e6 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 8 Aug 2026 19:23:00 -0700 Subject: [PATCH 1/3] [python] Forward global-index search modes to native planner --- .github/workflows/paimon-python-checks.yml | 2 +- paimon-python/pypaimon/read/native_plan.py | 23 ++++++- paimon-python/pypaimon/read/table_scan.py | 18 ++++- .../tests/native_plan_integration_test.py | 19 ++++++ .../pypaimon/tests/native_plan_test.py | 68 +++++++++++++++++++ 5 files changed, 126 insertions(+), 4 deletions(-) diff --git a/.github/workflows/paimon-python-checks.yml b/.github/workflows/paimon-python-checks.yml index 1621d9e12469..f28e61fe71d7 100755 --- a/.github/workflows/paimon-python-checks.yml +++ b/.github/workflows/paimon-python-checks.yml @@ -34,7 +34,7 @@ env: JDK_VERSION: 8 MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true LUMINA_DATA_VERSION: 0.1.0 - PYPAIMON_RUST_REV: b27c30054e17ee11f7400bf07a8fd41cf264f08b + PYPAIMON_RUST_REV: 7a8512f18f47a0634ee02ae8a09f2fff76c12d37 concurrency: diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py index 55b91d93b895..f6941bda2a1b 100644 --- a/paimon-python/pypaimon/read/native_plan.py +++ b/paimon-python/pypaimon/read/native_plan.py @@ -22,6 +22,7 @@ still applies them while reading, so pushdown remains an optimization. """ +import re from typing import List, Optional, Tuple from pypaimon.common.options.config import CatalogOptions @@ -46,6 +47,22 @@ def native_runtime_available() -> bool: return hasattr(PaimonCatalog, 'get_table') and hasattr(Split, 'serialize') +def native_family_search_modes_available() -> bool: + """Whether Rust supports family-specific global-index search modes.""" + if not native_runtime_available(): + return False + try: + from importlib.metadata import PackageNotFoundError, version + except ImportError: + return False + try: + rust_version = version('pypaimon-rust') + except PackageNotFoundError: + return False + match = re.match(r'^(\d+)\.(\d+)', rust_version) + return match is not None and tuple(map(int, match.groups())) >= (0, 4) + + def _partition_fields(table): """Ordered partition DataFields, used to decode the split partition bytes.""" schema = table.table_schema @@ -108,7 +125,11 @@ def _read_options(table) -> dict: for option in ( CoreOptions.SCAN_SNAPSHOT_ID, CoreOptions.SCAN_TAG_NAME, - CoreOptions.SCAN_TIMESTAMP_MILLIS): + CoreOptions.SCAN_TIMESTAMP_MILLIS, + CoreOptions.GLOBAL_INDEX_SEARCH_MODE, + CoreOptions.SCALAR_INDEX_SEARCH_MODE, + CoreOptions.VECTOR_INDEX_SEARCH_MODE, + CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE): if table_options.contains_key(option.key()): options[option.key()] = _option_value_to_string( table_options.get(option)) diff --git a/paimon-python/pypaimon/read/table_scan.py b/paimon-python/pypaimon/read/table_scan.py index fc50ece0f8ab..9dbcd04dd97e 100755 --- a/paimon-python/pypaimon/read/table_scan.py +++ b/paimon-python/pypaimon/read/table_scan.py @@ -32,6 +32,14 @@ logger = logging.getLogger(__name__) +_NATIVE_FAMILY_SEARCH_MODE_OPTIONS = frozenset({ + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), + CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(), + CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(), +}) +_NATIVE_SEARCH_MODE_OPTIONS = _NATIVE_FAMILY_SEARCH_MODE_OPTIONS | { + CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), +} # Options native forwards to Rust; any other copy() override is invisible to Rust. _NATIVE_FORWARDED_OPTIONS = frozenset({ CoreOptions.SCAN_NATIVE_PLAN_ENABLED.key(), @@ -41,7 +49,7 @@ CoreOptions.SCAN_TAG_NAME.key(), CoreOptions.SCAN_TIMESTAMP.key(), CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), -}) +}) | _NATIVE_SEARCH_MODE_OPTIONS _NATIVE_TIME_TRAVEL_OPTIONS = frozenset({ CoreOptions.SCAN_SNAPSHOT_ID.key(), CoreOptions.SCAN_TAG_NAME.key(), @@ -152,6 +160,11 @@ def _native_plan_supported_impl(self) -> bool: if self.table.bucket_mode() in (BucketMode.HASH_DYNAMIC, BucketMode.CROSS_PARTITION): return False options = self.table.options.options + if (any(options.contains_key(key) + for key in _NATIVE_FAMILY_SEARCH_MODE_OPTIONS)): + from pypaimon.read.native_plan import native_family_search_modes_available + if not native_family_search_modes_available(): + return False supported_time_travel = any( options.contains_key(key) for key in _NATIVE_TIME_TRAVEL_OPTIONS) # Time travel intentionally carries a historical schema; other stale @@ -163,7 +176,8 @@ def _native_plan_supported_impl(self) -> bool: # Rust cannot remove an option persisted in the catalog-loaded schema. applied_options = getattr(self.table, '_applied_dynamic_options', {}) or {} if (set(applied_options) - _NATIVE_FORWARDED_OPTIONS - or any(key in _NATIVE_TIME_TRAVEL_OPTIONS and value is None + or any(key in (_NATIVE_TIME_TRAVEL_OPTIONS + | _NATIVE_SEARCH_MODE_OPTIONS) and value is None for key, value in applied_options.items())): return False from pypaimon.snapshot.time_travel_util import SCAN_KEYS diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py b/paimon-python/pypaimon/tests/native_plan_integration_test.py index 5eb062afad9c..2a0c14a039d1 100644 --- a/paimon-python/pypaimon/tests/native_plan_integration_test.py +++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py @@ -131,6 +131,25 @@ def test_append_matches_normal_plan(self): self._write('ap_t', [{'k': 3, 'v': 'c'}]) self._assert_matches('ap_t') + def test_dynamic_family_search_mode_uses_native_plan(self): + self.cat.create_table( + 'default.search_mode_t', Schema.from_pyarrow_schema(self.schema), False) + self._write('search_mode_t', [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}]) + + table = self.cat.get_table('default.search_mode_t').copy({ + 'scan.native-plan.enabled': 'true', + 'scalar-index.search-mode': 'full', + }) + builder = table.new_read_builder() + plan = builder.new_scan().plan() + + self.assertEqual( + sorted(builder.new_read().to_arrow(plan.splits()).to_pylist(), + key=lambda row: row['k']), + [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}], + ) + self.assertTrue(builder.explain().native_planned) + def test_data_evolution_blob_projection_filter_limit(self): schema = pa.schema([ ('k', pa.int64()), diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py index 3f74a3fb860f..ce134a1b97c5 100644 --- a/paimon-python/pypaimon/tests/native_plan_test.py +++ b/paimon-python/pypaimon/tests/native_plan_test.py @@ -35,6 +35,7 @@ _predicate_to_native, _read_options, _restore_python_partition_paths, + native_family_search_modes_available, native_plan, ) from pypaimon.read.scan_stats import ScanStats @@ -328,6 +329,50 @@ def test_plan_falls_back_when_rust_unavailable(self): np.assert_not_called() fs.scan.assert_called_once_with() + def test_family_search_modes_require_rust_0_4(self): + for available, expect_native in ((False, False), (True, True)): + with self.subTest(available=available): + fs = Mock(partition_key_predicate=None) + fs.scan.return_value = fallback = object() + scan = _scan(native_enabled=True, file_scanner=fs) + scan.table.options.options.contains_key.side_effect = ( + lambda key: key == 'scalar-index.search-mode') + scan.table._applied_dynamic_options = { + 'scalar-index.search-mode': 'full', + } + split = Mock(partition=Mock(values=[]), snapshot_id=1) + + with patch( + 'pypaimon.read.native_plan.' + 'native_family_search_modes_available', + return_value=available), patch( + 'pypaimon.read.native_plan.native_plan', + return_value=[split]) as native: + plan = scan.plan() + + if expect_native: + self.assertEqual(plan.splits(), [split]) + native.assert_called_once() + fs.scan.assert_not_called() + else: + self.assertIs(plan, fallback) + native.assert_not_called() + fs.scan.assert_called_once_with() + + def test_removing_search_mode_falls_back(self): + fs = Mock(partition_key_predicate=None) + fs.scan.return_value = fallback = object() + scan = _scan(native_enabled=True, file_scanner=fs) + scan.table._applied_dynamic_options = { + 'scalar-index.search-mode': None, + } + + with patch('pypaimon.read.native_plan.native_plan') as native: + self.assertIs(scan.plan(), fallback) + + native.assert_not_called() + fs.scan.assert_called_once_with() + def test_plan_falls_back_when_native_plan_raises(self): # A native planning failure (e.g. unsupported scheme) must fall back, not crash. fs = Mock(partition_key_predicate=None) @@ -454,13 +499,36 @@ def test_predicate_and_time_travel_are_converted_for_rust(self): table.options.source_split_open_file_cost.return_value = 128 table.options.options = Options({ 'scan.snapshot-id': '9', + 'global-index.search-mode': 'detail', + 'scalar-index.search-mode': 'full', + 'vector-index.search-mode': 'fast', + 'full-text-index.search-mode': 'fast', }) self.assertEqual(_read_options(table), { 'source.split.target-size': '1024', 'source.split.open-file-cost': '128', 'scan.snapshot-id': '9', + 'global-index.search-mode': 'detail', + 'scalar-index.search-mode': 'full', + 'vector-index.search-mode': 'fast', + 'full-text-index.search-mode': 'fast', }) + @unittest.skipIf(sys.version_info < (3, 8), + "importlib.metadata requires Python 3.8") + def test_family_search_mode_version_gate(self): + cases = { + '0.3.0': False, + '0.4.0': True, + '0.4.0.dev20260808': True, + '1.0.0': True, + } + for version, expected in cases.items(): + with self.subTest(version=version), patch( + 'importlib.metadata.version', return_value=version): + self.assertEqual( + native_family_search_modes_available(), expected) + def test_partition_path_prefers_existing_python_legacy_path(self): table = Mock(partition_keys=['p']) table.path_factory.return_value.bucket_path.return_value = ( From 2f758d37175a49365d48b353d4eb3ff26114c058 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 11:05:17 +0800 Subject: [PATCH 2/3] [python] Skip native family mode test on old Rust --- paimon-python/pypaimon/tests/native_plan_integration_test.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py b/paimon-python/pypaimon/tests/native_plan_integration_test.py index 2a0c14a039d1..6ed414cd8199 100644 --- a/paimon-python/pypaimon/tests/native_plan_integration_test.py +++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py @@ -23,6 +23,7 @@ from pypaimon import CatalogFactory, Schema from pypaimon.globalindex.global_index_result import GlobalIndexResult +from pypaimon.read.native_plan import native_family_search_modes_available from pypaimon.utils.range import Range @@ -131,6 +132,8 @@ def test_append_matches_normal_plan(self): self._write('ap_t', [{'k': 3, 'v': 'c'}]) self._assert_matches('ap_t') + @unittest.skipUnless(native_family_search_modes_available(), + "pypaimon-rust 0.4+ required") def test_dynamic_family_search_mode_uses_native_plan(self): self.cat.create_table( 'default.search_mode_t', Schema.from_pyarrow_schema(self.schema), False) From 160bb0fcfda6eb64eb1806a932b3c21bca8453eb Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 8 Aug 2026 21:42:42 -0700 Subject: [PATCH 3/3] feat(python): prefer Jindo for native scan planning --- paimon-python/pypaimon/read/native_plan.py | 10 ++++- .../pypaimon/tests/native_plan_test.py | 37 +++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py index f6941bda2a1b..3d3732a4f41e 100644 --- a/paimon-python/pypaimon/read/native_plan.py +++ b/paimon-python/pypaimon/read/native_plan.py @@ -25,7 +25,7 @@ import re from typing import List, Optional, Tuple -from pypaimon.common.options.config import CatalogOptions +from pypaimon.common.options.config import CatalogOptions, OssOptions from pypaimon.common.options.core_options import CoreOptions from pypaimon.common.options.options_utils import OptionsUtils from pypaimon.common.predicate import Predicate @@ -110,6 +110,14 @@ def _catalog_options(table) -> dict: if metastore is None: raise ValueError("native_plan requires an exact built-in catalog loader") normalized[CatalogOptions.METASTORE.key()] = metastore + if str(getattr(table, 'table_path', '')).startswith('oss://'): + from pypaimon.filesystem.jindo_file_system_handler import ( + JINDO_AVAILABLE, + ) + impl = normalized.get(OssOptions.OSS_IMPL.key()) + if JINDO_AVAILABLE and (impl is None or impl.lower() == 'jindo'): + # This catalog is only used for Rust scan planning. + normalized[OssOptions.OSS_IMPL.key()] = 'jindo' return normalized diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py index ce134a1b97c5..eecbf2899f29 100644 --- a/paimon-python/pypaimon/tests/native_plan_test.py +++ b/paimon-python/pypaimon/tests/native_plan_test.py @@ -470,6 +470,43 @@ def test_catalog_options_use_actual_rest_loader_type(self): 'metastore': 'rest', }) + @patch( + 'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', True) + def test_native_plan_prefers_installed_jindo_for_oss(self): + table = Mock(table_path='oss://bucket/table') + table.catalog_environment.catalog_loader = FileSystemCatalogLoader( + CatalogContext.create_from_options(Options({}))) + + self.assertEqual(_catalog_options(table), { + 'metastore': 'filesystem', + 'fs.oss.impl': 'jindo', + }) + + @patch( + 'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', False) + def test_native_plan_uses_opendal_without_jindo(self): + table = Mock(table_path='oss://bucket/table') + table.catalog_environment.catalog_loader = FileSystemCatalogLoader( + CatalogContext.create_from_options(Options({}))) + + self.assertEqual(_catalog_options(table), { + 'metastore': 'filesystem', + }) + + @patch( + 'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', True) + def test_native_plan_respects_explicit_legacy_oss(self): + table = Mock(table_path='oss://bucket/table') + table.catalog_environment.catalog_loader = FileSystemCatalogLoader( + CatalogContext.create_from_options(Options({ + 'fs.oss.impl': 'legacy', + }))) + + self.assertEqual(_catalog_options(table), { + 'fs.oss.impl': 'legacy', + 'metastore': 'filesystem', + }) + def test_catalog_options_reject_loader_subclass(self): class RoutedFileSystemLoader(FileSystemCatalogLoader): pass