[python] Forward scan options to native planner

.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:

paimon-python/pypaimon/read/native_plan.py

@@ -22,9 +22,10 @@
 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
+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
@@ -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
@@ -93,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
 
 
@@ -108,7 +133,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))

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

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,27 @@ 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)
+        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()),

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)
@@ -425,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
@@ -454,13 +536,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 = (