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 7cd451fae1 [python] Keep native planning for Blob descriptor reads
(#9144)a
7cd451fae1 is described below
commit 7cd451fae1d4cad039675e4763c683617d504d8e
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Aug 10 20:09:47 2026 +0800
[python] Keep native planning for Blob descriptor reads (#9144)a
---
paimon-python/pypaimon/read/table_scan.py | 14 ++++++---
.../pypaimon/tests/native_plan_integration_test.py | 16 +++++++++++
paimon-python/pypaimon/tests/native_plan_test.py | 33 ++++++++++++++++++++++
3 files changed, 59 insertions(+), 4 deletions(-)
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 9dbcd04dd9..884d33b3cb 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -40,7 +40,6 @@ _NATIVE_FAMILY_SEARCH_MODE_OPTIONS = frozenset({
_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(),
CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(),
@@ -50,6 +49,11 @@ _NATIVE_FORWARDED_OPTIONS = frozenset({
CoreOptions.SCAN_TIMESTAMP.key(),
CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
}) | _NATIVE_SEARCH_MODE_OPTIONS
+_NATIVE_PLAN_INDEPENDENT_OPTIONS = frozenset({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(),
+ CoreOptions.READ_BATCH_SIZE.key(),
+ CoreOptions.READ_PARALLELISM.key(),
+})
_NATIVE_TIME_TRAVEL_OPTIONS = frozenset({
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
@@ -109,8 +113,8 @@ class TableScan:
a primary-key table whose trimmed PK is empty (PK equals the partition
key; native may mark splits raw-convertible and skip merge), dynamic
bucket / cross-partition PK tables (unconfirmed Rust parity), a stale
- schema without time travel, copy() overrides Rust does not see (notably
- removing a persisted scan option), unsupported time travel selectors,
+ schema without time travel, removed copy() options which Rust cannot
+ represent, unsupported time travel selectors,
query auth, non-main branch, incremental scans, a missing/old
pypaimon-rust, or a catalog / identifier Rust cannot reconstruct. Keep
this capability gate in sync when adding scan features."""
@@ -175,7 +179,9 @@ class TableScan:
return False
# 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
+ allowed_options = (
+ _NATIVE_FORWARDED_OPTIONS | _NATIVE_PLAN_INDEPENDENT_OPTIONS)
+ if (set(applied_options) - allowed_options
or any(key in (_NATIVE_TIME_TRAVEL_OPTIONS
| _NATIVE_SEARCH_MODE_OPTIONS) and value is None
for key, value in applied_options.items())):
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 6ed414cd81..7c0112d596 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -24,6 +24,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.table.row.blob import BlobDescriptor
from pypaimon.utils.range import Range
@@ -212,6 +213,21 @@ class NativePlanIntegrationTest(unittest.TestCase):
for data_file in split.files
))
+ descriptor_table = native_table.copy({'blob-as-descriptor': 'true'})
+ descriptor_builder = (
+ descriptor_table.new_read_builder()
+ .with_projection(['media.camera'])
+ .with_limit(1))
+ descriptor_plan = descriptor_builder.new_scan().plan()
+ descriptor_rows = descriptor_builder.new_read().to_arrow(
+ descriptor_plan.splits()).to_pylist()
+ self.assertEqual(len(descriptor_plan.splits()), 1)
+ self.assertEqual(
+
BlobDescriptor.deserialize(descriptor_rows[0]['media.camera']).length,
+ 1,
+ )
+ self.assertTrue(descriptor_builder.explain().native_planned)
+
@unittest.skipUnless(_has_native_row_ranges(),
"pypaimon_rust row-range API not installed")
def test_data_evolution_global_index_row_ranges(self):
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index eecbf2899f..acd624a6a7 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -373,6 +373,38 @@ class NativePlanTest(unittest.TestCase):
native.assert_not_called()
fs.scan.assert_called_once_with()
+ def test_dynamic_read_option_uses_native_plan(self):
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ scan.table._applied_dynamic_options = {
+ 'blob-as-descriptor': 'true',
+ 'read.batch-size': '2048',
+ 'read.parallelism': '2',
+ }
+ split = Mock(partition=Mock(values=[]), snapshot_id=1)
+
+ with patch(
+ 'pypaimon.read.native_plan.native_plan',
+ return_value=[split]) as native:
+ self.assertEqual(scan.plan().splits(), [split])
+
+ native.assert_called_once()
+ fs.scan.assert_not_called()
+
+ def test_unknown_dynamic_option_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 = {
+ 'future.scan-option': 'value',
+ }
+
+ 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)
@@ -655,6 +687,7 @@ class NativePlanTest(unittest.TestCase):
table.options.source_split_target_size.return_value = 1024
table.options.source_split_open_file_cost.return_value = 128
table.options.options.contains_key.return_value = False
+ table._applied_dynamic_options = {}
split = Mock()
split.serialize.return_value = b'bytes'
rt = Mock()
