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 c67e38ba9b [python] Forward scan options to native planner (#9124)
c67e38ba9b is described below
commit c67e38ba9b671f88783d15706547ad83665b84a3
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Aug 9 13:33:26 2026 +0800
[python] Forward scan options to native planner (#9124)
---
.github/workflows/paimon-python-checks.yml | 2 +-
paimon-python/pypaimon/read/native_plan.py | 33 ++++++-
paimon-python/pypaimon/read/table_scan.py | 18 +++-
.../pypaimon/tests/native_plan_integration_test.py | 22 +++++
paimon-python/pypaimon/tests/native_plan_test.py | 105 +++++++++++++++++++++
5 files changed, 175 insertions(+), 5 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml
b/.github/workflows/paimon-python-checks.yml
index 1621d9e124..f28e61fe71 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 55b91d93b8..3d3732a4f4 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -22,9 +22,10 @@ Predicates and limits are pushed into Rust planning. The
normal pypaimon reader
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))
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index fc50ece0f8..9dbcd04dd9 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -32,6 +32,14 @@ from pypaimon.read.scanner.file_scanner import FileScanner
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 @@ _NATIVE_FORWARDED_OPTIONS = frozenset({
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 @@ class TableScan:
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 @@ class TableScan:
# 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 5eb062afad..6ed414cd81 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 @@ import pyarrow as pa
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 @@ class NativePlanIntegrationTest(unittest.TestCase):
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()),
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 3f74a3fb86..eecbf2899f 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -35,6 +35,7 @@ from pypaimon.read.native_plan import (
_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 @@ class NativePlanTest(unittest.TestCase):
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 @@ class NativePlanTest(unittest.TestCase):
'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 @@ class NativePlanTest(unittest.TestCase):
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 = (
