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 ad1e6c1878 [python] Remove legacy native read option forwarding
(#10120)
ad1e6c1878 is described below
commit ad1e6c187817f23c2e20c15f6c59d5923fc5b6b0
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Sep 23 09:24:56 2026 +0800
[python] Remove legacy native read option forwarding (#10120)
---
paimon-python/README.md | 8 +-
paimon-python/pypaimon/read/native_plan.py | 64 ++------
.../pypaimon/tests/native_plan_integration_test.py | 7 +-
paimon-python/pypaimon/tests/native_plan_test.py | 170 ++++++++++++---------
4 files changed, 109 insertions(+), 140 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 7da6734823..3295d70df3 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -219,9 +219,11 @@ interrupts native reads that are still in flight. A
missing reader capability,
unsupported route, or native-reader construction failure falls back to Python;
I/O and data errors raised after streaming starts surface to the caller.
-With Rust main's `Table.from_resolved_schema()` binding, filesystem and JDBC
catalog
-tables preserve the Python table's resolved schema and complete effective
-options. Stale table objects, historical schemas, and `copy()` overrides or
+Native planning and reading require Rust's resolved-schema bindings:
+`Table.from_resolved_schema()` for filesystem, JDBC and path-based tables, and
+`Table.copy_with_resolved_schema()` for REST tables. The adapter passes the
Python
+table's resolved schema and complete effective options without an option
whitelist.
+Stale table objects, historical schemas, and `copy()` overrides or
option removals no longer require catalog reloading or Python planning.
Tables opened with `FileStoreTable.from_path(path, file_io_options=None)` use
the same path with standard local, PyArrow or resolving FileIO. Storage options
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 6eac710aaa..ca655dea07 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -27,7 +27,6 @@ from typing import List, Optional, Tuple
from packaging.version import InvalidVersion, Version
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
from pypaimon.read.plan import Plan
@@ -167,45 +166,8 @@ def _catalog_context_options(table) -> dict:
return normalized
-def _read_options(table) -> dict:
- """Effective Rust read options, including FileStoreTable.copy overrides."""
- options = {
- CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(): str(
- table.options.source_split_target_size()),
- CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): str(
- table.options.source_split_open_file_cost()),
- CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key():
_option_value_to_string(
-
table.options.options.get(CoreOptions.DELETION_VECTORS_MERGE_ON_READ)),
- }
- table_options = table.options.options
- for option in (
- CoreOptions.BLOB_AS_DESCRIPTOR,
- CoreOptions.SCAN_VERSION,
- CoreOptions.SCAN_SNAPSHOT_ID,
- CoreOptions.SCAN_TAG_NAME,
- CoreOptions.SCAN_TIMESTAMP_MILLIS,
- CoreOptions.SCAN_WATERMARK,
- 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.to_map()[option.key()])
-
- # Rust takes epoch millis but PyPaimon also accepts a timestamp string.
- if table_options.contains_key(CoreOptions.SCAN_TIMESTAMP.key()):
- from pypaimon.snapshot.time_travel_util import
_parse_timestamp_to_millis
- options[CoreOptions.SCAN_TIMESTAMP_MILLIS.key()] = str(
- _parse_timestamp_to_millis(
- table_options.get(CoreOptions.SCAN_TIMESTAMP)))
- return options
-
-
def _resolved_schema_file_io_options(table) -> Optional[dict]:
"""FileIO properties for tables whose metadata needs no catalog
resolution."""
- if not native_method_available('Table', 'from_resolved_schema'):
- return None
environment = table.catalog_environment
loader = environment.catalog_loader
if loader is None:
@@ -234,12 +196,10 @@ def _resolved_schema_file_io_options(table) ->
Optional[dict]:
def _resolved_schema_json(table) -> str:
+ """Preserve all effective table options as strings for Rust."""
from pypaimon.common.json_util import JSON
options = {str(key): _option_value_to_string(value)
for key, value in table.table_schema.options.items() if value
is not None}
- options.update(_read_options(table))
- # The timestamp string has already been converted to epoch millis.
- options.pop(CoreOptions.SCAN_TIMESTAMP.key(), None)
return JSON.to_json(table.table_schema.copy(new_options=options))
@@ -301,27 +261,21 @@ def _native_read_builder(table):
database=table.identifier.get_database_name(),
table=table.identifier.get_table_name(),
branch=table.current_branch(), options=file_io_options)
- builder = rt.new_read_builder()
else:
from pypaimon_rust.datafusion import PaimonCatalog
catalog = PaimonCatalog(_catalog_options(table))
- if native_method_available('Table', 'copy_with_resolved_schema'):
- # REST may keep branch schemas in the catalog only. Load the base
- # environment, then attach the schema/branch already resolved here.
- rt = catalog.get_table((
- table.identifier.get_database_name(),
table.identifier.get_table_name()))
- if rt.location() != table.table_path:
- raise RuntimeError('Native catalog resolved a different table
location')
- rt = rt.copy_with_resolved_schema(_resolved_schema_json(table),
branch=table.current_branch())
- builder = rt.new_read_builder()
- else:
- rt = catalog.get_table(table.identifier.get_full_name())
- builder = rt.new_read_builder(_read_options(table))
+ # REST may keep branch schemas in the catalog only. Load the base
+ # environment, then attach the schema/branch already resolved here.
+ rt = catalog.get_table((
+ table.identifier.get_database_name(),
table.identifier.get_table_name()))
+ if rt.location() != table.table_path:
+ raise RuntimeError('Native catalog resolved a different table
location')
+ rt = rt.copy_with_resolved_schema(_resolved_schema_json(table),
branch=table.current_branch())
if table.current_branch() != 'main':
branch = getattr(rt, 'branch', None)
if not callable(branch) or branch() != table.current_branch():
raise RuntimeError("Native table did not resolve the requested
branch")
- return builder
+ return rt.new_read_builder()
def _configure_native_read_builder(builder, predicate, limit, projection,
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 646ea5eb68..aa22c9a7ba 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -1130,12 +1130,7 @@ class NativePlanIntegrationTest(unittest.TestCase):
@unittest.skipUnless(native_reader_available(),
"pypaimon-rust native reader API not installed")
- @patch('pypaimon.read.native_plan.native_method_available',
- side_effect=lambda type_name, method: (
- False if method in ('from_resolved_schema',
'copy_with_resolved_schema')
- else native_method_available(type_name, method)))
- def test_native_read_dynamic_blob_as_descriptor(self, capabilities):
- # Exercise the catalog path used by Rust versions without resolved
schemas.
+ def test_native_read_dynamic_blob_as_descriptor(self):
schema = pa.schema([('id', pa.int32()), ('payload',
pa.large_binary())])
self.cat.create_table(
'default.native_dynamic_descriptor',
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 124a64c4c2..285c0a40fa 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -33,8 +33,8 @@ from pypaimon.globalindex.global_index_result import
GlobalIndexResult
from pypaimon.globalindex.vector_search_result import ScoredGlobalIndexResult
from pypaimon.read.native_plan import (
_catalog_options,
+ _native_read_builder,
_predicate_to_native,
- _read_options,
_resolved_schema_json,
_restore_python_partition_paths,
native_family_search_modes_available,
@@ -120,13 +120,11 @@ class NativePlanTest(unittest.TestCase):
False, AtomicType('STRING', False), AtomicType('INT'))))
])))
])
- table = SimpleNamespace(table_schema=schema,
options=CoreOptions(Options({})))
+ table = SimpleNamespace(table_schema=schema)
self.assertEqual(json.loads(_resolved_schema_json(table)), {
'version': 3, 'id': 7, 'highestFieldId': 3, 'timeMillis': 0,
'partitionKeys': [], 'primaryKeys': [], 'comment': None,
- 'options': {'source.split.target-size': '134217728',
- 'source.split.open-file-cost': '4194304',
- 'deletion-vectors.merge-on-read': 'false'},
+ 'options': {},
'fields': [{'id': 0, 'name': 'attributes', 'type': {
'type': 'MAP NOT NULL', 'nullable': False, 'key': 'STRING NOT
NULL',
'value': {'type': 'ROW', 'nullable': True, 'fields': [
@@ -160,24 +158,23 @@ class NativePlanTest(unittest.TestCase):
pass
table = Mock(file_io=LocalFileIO(),
catalog_environment=CatalogEnvironment.empty())
- with patch('pypaimon.read.native_plan.native_method_available',
return_value=True):
- self.assertEqual(_resolved_schema_file_io_options(table), {})
- table.file_io = CustomIO()
+ self.assertEqual(_resolved_schema_file_io_options(table), {})
+ table.file_io = CustomIO()
+ self.assertIsNone(_resolved_schema_file_io_options(table))
+ table.file_io = LocalFileIO()
+ table.catalog_environment = CustomEnvironment()
+ self.assertIsNone(_resolved_schema_file_io_options(table))
+ table.catalog_environment = CatalogEnvironment.empty()
+ for loader_type in (RESTCatalogLoader, CustomLoader, CustomJdbcLoader):
+ table.catalog_environment.catalog_loader = loader_type(
+ CatalogContext.create_from_options(Options({})))
self.assertIsNone(_resolved_schema_file_io_options(table))
- table.file_io = LocalFileIO()
- table.catalog_environment = CustomEnvironment()
- self.assertIsNone(_resolved_schema_file_io_options(table))
- table.catalog_environment = CatalogEnvironment.empty()
- for loader_type in (RESTCatalogLoader, CustomLoader,
CustomJdbcLoader):
- table.catalog_environment.catalog_loader = loader_type(
- CatalogContext.create_from_options(Options({})))
+ for loader_type in (FileSystemCatalogLoader, JdbcCatalogLoader):
+ for attr in ('hadoop_conf', 'prefer_io_loader',
'fallback_io_loader'):
+ context = CatalogContext.create_from_options(Options({}))
+ setattr(context, attr, object())
+ table.catalog_environment.catalog_loader = loader_type(context)
self.assertIsNone(_resolved_schema_file_io_options(table))
- for loader_type in (FileSystemCatalogLoader, JdbcCatalogLoader):
- for attr in ('hadoop_conf', 'prefer_io_loader',
'fallback_io_loader'):
- context = CatalogContext.create_from_options(Options({}))
- setattr(context, attr, object())
- table.catalog_environment.catalog_loader =
loader_type(context)
- self.assertIsNone(_resolved_schema_file_io_options(table))
def test_switch_defaults_off(self):
defaults = CoreOptions(Options({}))
@@ -200,9 +197,8 @@ class NativePlanTest(unittest.TestCase):
arrow.properties = properties
for file_io in (arrow, ResolvingFileIO(properties)):
table = Mock(file_io=file_io,
catalog_environment=CatalogEnvironment.empty())
- with patch('pypaimon.read.native_plan.native_method_available',
return_value=True):
- self.assertEqual(_resolved_schema_file_io_options(table), {
- 's3.path-style-access': 'true', 's3.endpoint':
'http://localhost:9000'})
+ self.assertEqual(_resolved_schema_file_io_options(table), {
+ 's3.path-style-access': 'true', 's3.endpoint':
'http://localhost:9000'})
def test_plan_uses_file_scanner_when_switch_off(self):
fs = Mock()
@@ -387,8 +383,13 @@ class NativePlanTest(unittest.TestCase):
check(lambda s, fs: (setattr(s.table, 'is_primary_key_table', True),
setattr(s.table, 'trimmed_primary_keys', [])))
check(lambda s, fs: setattr(s.table.options, 'query_auth_enabled',
True))
- check(lambda s, fs: s.table.identifier.get_database_name.__setattr__(
- 'return_value', 'unknown'))
+
+ def unknown_rest_database(scan, fs):
+ scan.table.catalog_environment.catalog_loader = RESTCatalogLoader(
+ CatalogContext.create_from_options(Options({})))
+ scan.table.identifier.get_database_name.return_value = 'unknown'
+
+ check(unknown_rest_database)
check(lambda s, fs: setattr(
s.table.catalog_environment, 'catalog_loader', object())) # no
context()
for attr in ('hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'):
@@ -494,19 +495,17 @@ class NativePlanTest(unittest.TestCase):
self.assertIs(scan.plan(), sentinel)
fs.scan.assert_called_once_with()
- def test_plan_falls_back_for_jdbc_catalog_loader(self):
+ def test_plan_uses_resolved_schema_for_jdbc_catalog_loader(self):
fs = Mock(partition_key_predicate=None)
- sentinel = object()
- fs.scan.return_value = sentinel
scan = _scan(native_enabled=True, file_scanner=fs)
scan.table.catalog_environment.catalog_loader = JdbcCatalogLoader(
CatalogContext.create_from_options(Options({})))
- with patch('pypaimon.read.native_plan.native_plan') as np:
- self.assertIs(scan.plan(), sentinel)
+ with patch('pypaimon.read.native_plan.native_plan',
return_value=Plan([], 1)) as np:
+ self.assertEqual(scan.plan().snapshot_id, 1)
- np.assert_not_called()
- fs.scan.assert_called_once_with()
+ np.assert_called_once()
+ fs.scan.assert_not_called()
def test_plan_falls_back_for_builtin_catalog_loader_subclasses(self):
class RoutedFileSystemLoader(FileSystemCatalogLoader):
@@ -627,13 +626,14 @@ class NativePlanTest(unittest.TestCase):
def test_blob_as_descriptor_is_forwarded_to_rust(self):
for value in ('true', 'false', True, False):
with self.subTest(value=value):
- table = Mock()
- table.options = CoreOptions(Options({'blob-as-descriptor':
value}))
+ options = {'blob-as-descriptor': value}
+ table = SimpleNamespace(
+ table_schema=TableSchema(0, [], options=options))
self.assertEqual(
- _read_options(table)['blob-as-descriptor'],
+
json.loads(_resolved_schema_json(table))['options']['blob-as-descriptor'],
str(value).lower())
- def test_predicate_and_time_travel_are_converted_for_rust(self):
+ def test_predicate_is_converted_for_rust(self):
predicate = PredicateBuilder.and_predicates([
Predicate('greaterOrEqual', 0, 'k', [10]),
Predicate('in', 1, 'v', ['a', 'b']),
@@ -646,26 +646,48 @@ class NativePlanTest(unittest.TestCase):
],
})
- table = Mock()
- table.options.source_split_target_size.return_value = 1024
- table.options.source_split_open_file_cost.return_value = 128
- table.options.options = Options({
+ def test_resolved_schema_preserves_options_and_stringifies_values(self):
+ options = {
+ 'source.split.target-size': '1 kb',
+ 'source.split.open-file-cost': '128 b',
+ 'deletion-vectors.merge-on-read': 'true',
'scan.snapshot-id': '9',
+ 'scan.watermark': 200,
'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',
- 'deletion-vectors.merge-on-read': 'false',
+ 'read.batch-size': 32,
+ 'custom.read-option': True,
+ 'removed.option': None,
+ }
+ table = SimpleNamespace(
+ table_schema=TableSchema(0, [], options=options))
+ self.assertEqual(json.loads(_resolved_schema_json(table))['options'], {
+ 'source.split.target-size': '1 kb',
+ 'source.split.open-file-cost': '128 b',
+ 'deletion-vectors.merge-on-read': 'true',
'scan.snapshot-id': '9',
+ 'scan.watermark': '200',
'global-index.search-mode': 'detail',
'scalar-index.search-mode': 'full',
'vector-index.search-mode': 'fast',
'full-text-index.search-mode': 'fast',
+ 'read.batch-size': '32',
+ 'custom.read-option': 'true',
})
+ self.assertEqual(table.table_schema.options, options)
+
+ def test_resolved_schema_preserves_timestamp_selectors(self):
+ for key, value in (('scan.timestamp', '2026-09-22T00:00:00'),
+ ('scan.timestamp-millis', 1790035200000)):
+ with self.subTest(key=key):
+ options = {key: value}
+ table = SimpleNamespace(
+ table_schema=TableSchema(0, [], options=options))
+ resolved = json.loads(_resolved_schema_json(table))['options']
+ self.assertEqual(resolved, {key: str(value)})
+ self.assertEqual(table.table_schema.options, options)
@unittest.skipIf(sys.version_info < (3, 8),
"importlib.metadata requires Python 3.8")
@@ -771,29 +793,14 @@ class NativePlanTest(unittest.TestCase):
table.current_branch.return_value = 'main'
table.table_schema = Mock(fields=[], partition_keys=[])
table.partition_keys = []
- table.options.source_split_target_size.return_value = 1024
- table.options.source_split_open_file_cost.return_value = 128
- table.options.options = Options({})
- table._applied_dynamic_options = {}
split = Mock()
split.serialize.return_value = b'bytes'
- rt = Mock()
- builder = rt.new_read_builder.return_value
+ builder = Mock()
builder.with_row_ranges.return_value = builder
builder.new_scan.return_value.plan.return_value.splits.return_value =
[split]
builder.new_scan.return_value.plan.return_value.snapshot_id.return_value = 3
- catalog = Mock()
- catalog.get_table.return_value = rt
- fake_df = ModuleType('pypaimon_rust.datafusion')
- fake_df.PaimonCatalog = Mock(return_value=catalog)
- fake_df.Split = type('Split', (), {'serialize': lambda self: b''})
- fake_mod = ModuleType('pypaimon_rust')
- fake_mod.datafusion = fake_df
-
- with patch.dict(sys.modules,
- {'pypaimon_rust': fake_mod,
'pypaimon_rust.datafusion': fake_df}), \
- patch('pypaimon.read.native_plan._catalog_options',
return_value={}), \
+ with patch('pypaimon.read.native_plan._native_read_builder',
return_value=builder), \
patch('pypaimon.read.native_plan.deserialize_split_v1') as des:
decoded = Mock()
des.return_value = decoded
@@ -802,11 +809,6 @@ class NativePlanTest(unittest.TestCase):
self.assertEqual(result.splits(), [decoded])
self.assertIs(decoded._native_split, split)
self.assertEqual(result.snapshot_id, 3)
- rt.new_read_builder.assert_called_once_with({
- CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(): '1024',
- CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): '128',
- CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(): 'false',
- })
builder.with_row_ranges.assert_called_once_with([(1, 2)])
des.assert_called_once_with(b'bytes', [], kfields)
@@ -819,10 +821,9 @@ class NativePlanTest(unittest.TestCase):
if not legacy:
rust_plan.snapshot_id = lambda: snapshot_id
scan = SimpleNamespace(plan=lambda: rust_plan)
- rt = Mock()
- rt.new_read_builder.return_value.new_scan.return_value = scan
- with patch('pypaimon_rust.datafusion.PaimonCatalog') as
catalog:
- catalog.return_value.get_table.return_value = rt
+ builder = Mock()
+ builder.new_scan.return_value = scan
+ with patch('pypaimon.read.native_plan._native_read_builder',
return_value=builder):
if legacy:
with self.assertRaisesRegex(RuntimeError, "empty
plan's snapshot"):
native_plan(table)
@@ -833,7 +834,8 @@ class NativePlanTest(unittest.TestCase):
def test_native_plan_branch_resolution(self):
table = _scan(True, Mock()).table
- table.table_schema = Mock(fields=[], partition_keys=[])
+ table.table_schema = TableSchema(0, [])
+ table.options = CoreOptions(Options({}))
table.current_branch.return_value = 'b1'
rust_plan = SimpleNamespace(splits=lambda: [], snapshot_id=lambda: 7)
scan = Mock()
@@ -841,8 +843,8 @@ class NativePlanTest(unittest.TestCase):
rt = Mock()
rt.branch.return_value = 'b1'
rt.new_read_builder.return_value.new_scan.return_value = scan
- with patch('pypaimon_rust.datafusion.PaimonCatalog') as catalog:
- catalog.return_value.get_table.return_value = rt
+ with patch('pypaimon_rust.datafusion.Table', create=True) as
native_table:
+ native_table.from_resolved_schema.return_value = rt
plan = native_plan(table)
self.assertEqual(plan.snapshot_id, 7)
scan.plan.assert_called_once_with()
@@ -850,6 +852,23 @@ class NativePlanTest(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, 'requested branch'):
native_plan(table)
+ def test_native_read_builder_requires_resolved_schema(self):
+ for loader_type in (FileSystemCatalogLoader, RESTCatalogLoader):
+ with self.subTest(loader=loader_type):
+ table = _scan(True, Mock()).table
+ table.table_schema = TableSchema(0, [],
options={'read.batch-size': 32})
+ table.options =
CoreOptions(Options(table.table_schema.options))
+ table.catalog_environment.catalog_loader = loader_type(
+ CatalogContext.create_from_options(Options({})))
+ legacy_table = SimpleNamespace(
+ location=lambda: table.table_path, new_read_builder=Mock())
+ with patch('pypaimon_rust.datafusion.Table', type('Table', (),
{}), create=True), \
+ patch('pypaimon_rust.datafusion.PaimonCatalog') as
catalog:
+ catalog.return_value.get_table.return_value = legacy_table
+ with self.assertRaises(AttributeError):
+ _native_read_builder(table)
+ legacy_table.new_read_builder.assert_not_called()
+
def test_explicit_row_ranges_are_forwarded(self):
for ranges in ([], [Range(1, 2), Range(5, 8)]):
with self.subTest(ranges=ranges):
@@ -882,7 +901,6 @@ class NativePlanTest(unittest.TestCase):
scan.table.options.options = Options({'scan.watermark': '200'})
scan.table._applied_dynamic_options = {'scan.watermark': '200'}
scan.table.schema_manager.latest.return_value.id = 2
- self.assertEqual(_read_options(scan.table)['scan.watermark'], '200')
with patch('pypaimon.read.native_plan.native_plan',
return_value=Plan([], 1)) as native:
self.assertEqual(scan.plan().snapshot_id, 1)
