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 1c838e11bf [python] Reuse resolved REST metadata for native readers
(#10123)
1c838e11bf is described below
commit 1c838e11bf3fbd2925c1a13f46bda9ad90479507
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Sep 25 21:12:22 2026 +0800
[python] Reuse resolved REST metadata for native readers (#10123)
---
paimon-python/README.md | 2 +
.../pypaimon/catalog/catalog_environment.py | 7 +-
.../pypaimon/catalog/rest/rest_catalog.py | 7 +-
.../pypaimon/catalog/rest/table_metadata.py | 4 +-
paimon-python/pypaimon/read/native_plan.py | 74 ++++++-
.../pypaimon/tests/interval_partition_test.py | 6 +-
.../tests/native_plan_resolved_schema_test.py | 5 +-
.../pypaimon/tests/native_plan_rest_test.py | 109 +++++++++-
paimon-python/pypaimon/tests/native_plan_test.py | 237 ++++++++++++++++++++-
.../pypaimon/tests/rest/rest_branch_test.py | 4 +
paimon-python/pypaimon/tests/rest/rest_server.py | 9 +-
11 files changed, 449 insertions(+), 15 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 96f7ba0d9f..d83b947518 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -262,6 +262,8 @@ JDBC planning uses the resolved table location and storage
properties without
opening another database connection.
REST tables use `Table.copy_with_resolved_schema()` to preserve the same schema
and option semantics, including branches whose schemas are catalog-managed.
+Matching REST tables retain the native environment across scans and read-option
+copies, preserving FileIO caches. Worker deserialization creates a fresh
environment.
The native table retains REST credentials, token refresh and catalog snapshot
resolution. Database and table names containing dots are passed as separate
identifier components. REST snapshot results (including empty results) take
precedence over
diff --git a/paimon-python/pypaimon/catalog/catalog_environment.py
b/paimon-python/pypaimon/catalog/catalog_environment.py
index 84530fbc0a..badb147340 100644
--- a/paimon-python/pypaimon/catalog/catalog_environment.py
+++ b/paimon-python/pypaimon/catalog/catalog_environment.py
@@ -39,12 +39,14 @@ class CatalogEnvironment:
identifier: Optional[Identifier] = None,
uuid: Optional[str] = None,
catalog_loader: Optional[CatalogLoader] = None,
- supports_version_management: bool = False
+ supports_version_management: bool = False,
+ rest_table_response: Optional[str] = None
):
self.identifier = identifier
self.uuid = uuid
self.catalog_loader = catalog_loader
self.supports_version_management = supports_version_management
+ self.rest_table_response = rest_table_response
def snapshot_commit(self, snapshot_manager) -> Optional[SnapshotCommit]:
"""
@@ -142,7 +144,8 @@ class CatalogEnvironment:
identifier=identifier,
uuid=self.uuid,
catalog_loader=self.catalog_loader,
- supports_version_management=self.supports_version_management
+ supports_version_management=self.supports_version_management,
+ rest_table_response=getattr(self, 'rest_table_response', None)
)
@staticmethod
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index a572c947f9..1b7f58e4f6 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -45,6 +45,7 @@ from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.file_io import FileIO
from pypaimon.filesystem.caching_file_io import CachingFileIO
from pypaimon.common.identifier import Identifier
+from pypaimon.common.json_util import JSON
from pypaimon.schema.schema import Schema
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.table_schema import TableSchema
@@ -672,7 +673,8 @@ class RESTCatalog(Catalog):
return TableMetadata(
schema=schema.copy(options),
is_external=response.get_is_external(),
- uuid=response.get_id()
+ uuid=response.get_id(),
+ rest_table_response=JSON.to_json(response)
)
def file_io_from_options(self, table_path: str) -> FileIO:
@@ -718,7 +720,8 @@ class RESTCatalog(Catalog):
identifier=identifier,
uuid=metadata.uuid,
catalog_loader=self.catalog_loader(),
- supports_version_management=True # REST catalogs support version
management
+ supports_version_management=True,
+ rest_table_response=getattr(metadata, 'rest_table_response', None)
)
# Use the path from server response directly (do not trim scheme)
table_path = schema.options.get(CoreOptions.PATH.key())
diff --git a/paimon-python/pypaimon/catalog/rest/table_metadata.py
b/paimon-python/pypaimon/catalog/rest/table_metadata.py
index e9a7221f37..3226081ce4 100644
--- a/paimon-python/pypaimon/catalog/rest/table_metadata.py
+++ b/paimon-python/pypaimon/catalog/rest/table_metadata.py
@@ -22,10 +22,12 @@ from pypaimon.schema.table_schema import TableSchema
class TableMetadata:
- def __init__(self, schema: TableSchema, is_external: bool, uuid:
Optional[str] = None):
+ def __init__(self, schema: TableSchema, is_external: bool, uuid:
Optional[str] = None,
+ rest_table_response: Optional[str] = None):
self._schema = schema
self._is_external = is_external
self._uuid = uuid
+ self.rest_table_response = rest_table_response
@property
def schema(self) -> TableSchema:
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index ca655dea07..c314a75e5c 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -22,6 +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 json
+import os
+from threading import Lock
+from types import SimpleNamespace
from typing import List, Optional, Tuple
from packaging.version import InvalidVersion, Version
@@ -251,10 +255,76 @@ def _restore_python_partition_paths(table, splits:
List[Split]) -> None:
split._native_split = None
+def _resolved_rest_table_response(table):
+ """Reuse REST metadata only when the standard loader can be reproduced."""
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+
+ environment = table.catalog_environment
+ response = getattr(environment, 'rest_table_response', None)
+ if (type(environment) is not CatalogEnvironment
+ or type(environment.catalog_loader) is not RESTCatalogLoader
+ or not isinstance(response, str)
+ or not native_method_available('Table', 'from_rest_response')):
+ return None
+ context = environment.catalog_loader.context()
+ if any(getattr(context, attr, None) is not None for attr in (
+ 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')):
+ return None
+ metadata = json.loads(response)
+ if (metadata.get('path') != table.table_path
+ or metadata.get('name') != table.identifier.get_object_name()
+ or ('database' in metadata
+ and metadata['database'] !=
table.identifier.get_database_name())):
+ return None
+ return response
+
+
+class _NativeRestTableCache:
+ """One native environment per Python environment, never sent to workers."""
+
+ def __init__(self):
+ self._states = {}
+
+ def __getstate__(self):
+ return {}
+
+ def __setstate__(self, state):
+ self.__init__()
+
+ def get(self, response, database, table, options):
+ from pypaimon_rust.datafusion import Table
+
+ pid = os.getpid()
+ # Publish a complete state atomically; never touch an inherited lock.
+ states = self._states
+ state = states.get(pid)
+ if state is None:
+ state = states.setdefault(pid, SimpleNamespace(lock=Lock(),
entry=None))
+ self._states = {pid: state}
+ key = (response, database, table, tuple(sorted(options.items())))
+ with state.lock:
+ if state.entry is None or state.entry[0] != key:
+ native_table = Table.from_rest_response(
+ response, database=database, table=table,
rest_options=options)
+ state.entry = (key, native_table)
+ return state.entry[1]
+
+
def _native_read_builder(table):
- """Reconstruct the Rust table and return a builder for the same schema."""
+ """Return a fresh builder with the current schema and shared REST
FileIO."""
+ rest_response = _resolved_rest_table_response(table)
file_io_options = _resolved_schema_file_io_options(table)
- if file_io_options is not None:
+ if rest_response is not None:
+ cache = table.catalog_environment.__dict__.setdefault(
+ '_native_rest_table_cache', _NativeRestTableCache())
+ rt = cache.get(
+ rest_response,
+ database=table.identifier.get_database_name(),
+ table=table.identifier.get_object_name(),
+ options=_catalog_options(table))
+ rt = rt.copy_with_resolved_schema(_resolved_schema_json(table),
branch=table.current_branch())
+ elif file_io_options is not None:
from pypaimon_rust.datafusion import Table
rt = Table.from_resolved_schema(
table.table_path, _resolved_schema_json(table),
diff --git a/paimon-python/pypaimon/tests/interval_partition_test.py
b/paimon-python/pypaimon/tests/interval_partition_test.py
index 173ed6c9ee..822d3e64d1 100644
--- a/paimon-python/pypaimon/tests/interval_partition_test.py
+++ b/paimon-python/pypaimon/tests/interval_partition_test.py
@@ -19,6 +19,7 @@
from decimal import Decimal
from types import SimpleNamespace
+from unittest.mock import Mock
import pytest
@@ -60,7 +61,10 @@ def
test_signed_zero_key_ranges_keep_versions_in_one_split(type_name):
assert len(sections) == 1
assert sorted([f.file_name for f in run.files] for run in sections[0]) ==
[['broad'], ['point']]
- table = SimpleNamespace(table_path='/tmp/interval-test',
options=CoreOptions(Options({})))
+ path_factory = Mock()
+ path_factory.bucket_path.return_value = '/tmp/interval-test/bucket-0'
+ table = SimpleNamespace(table_path='/tmp/interval-test',
options=CoreOptions(Options({})),
+ path_factory=lambda: path_factory)
entries = [ManifestEntry(0, GenericRow([], []), 0, 1, file) for file in
files]
splits = PrimaryKeyTableSplitGenerator(
table, 1, 1, snapshot_id=7).create_splits(entries)
diff --git a/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py
b/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py
index cbae972b09..bfa0e0ce8a 100644
--- a/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py
@@ -23,7 +23,7 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.identifier import Identifier
-from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.read.native_plan import native_method_available,
native_runtime_available
from pypaimon.schema.data_types import AtomicType
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.table.file_store_table import FileStoreTable
@@ -91,7 +91,8 @@ def _read(table, native, predicate=None, projection=None):
side_effect=AssertionError('native fallback')))
stack.enter_context(patch.object(table.schema_manager, 'latest',
side_effect=AssertionError('schema reload')))
- if type(table.catalog_environment.catalog_loader) is not
RESTCatalogLoader:
+ if (type(table.catalog_environment.catalog_loader) is not
RESTCatalogLoader
+ or native_method_available('Table', 'from_rest_response')):
stack.enter_context(patch('pypaimon_rust.datafusion.PaimonCatalog',
side_effect=AssertionError('catalog
reload')))
if type(table.catalog_environment.catalog_loader) is
JdbcCatalogLoader:
diff --git a/paimon-python/pypaimon/tests/native_plan_rest_test.py
b/paimon-python/pypaimon/tests/native_plan_rest_test.py
index a2c2a92fde..320bfd61bc 100644
--- a/paimon-python/pypaimon/tests/native_plan_rest_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_rest_test.py
@@ -23,7 +23,7 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.api.api_response import ConfigResponse, ErrorResponse,
GetTableSnapshotResponse
from pypaimon.api.auth import BearTokenAuthProvider
-from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.read.native_plan import native_method_available,
native_runtime_available
from pypaimon.snapshot.table_snapshot import TableSnapshot
from pypaimon.table.row.blob import BlobViewStruct
from pypaimon.tests.rest.rest_server import RESTCatalogServer
@@ -110,6 +110,25 @@ def
test_rest_branch_keeps_catalog_snapshot_and_schema(rest_source, rest_catalog
assert all(call.args[2] == 'dev' for call in load.call_args_list)
[email protected]('from_tag', [False, True], ids=['empty-branch',
'tagged-branch'])
+def test_dynamic_branch_uses_native_catalog(rest_source, rest_catalog,
from_tag):
+ from pypaimon.read.native_plan import _resolved_rest_table_response,
native_plan
+ table, _, _ = rest_source
+ catalog, _ = rest_catalog
+ if from_tag:
+ catalog.create_tag(table.identifier, 'first', 1)
+ catalog.create_branch(table.identifier, 'dev', tag_name='first' if
from_tag else None)
+ branch = table.copy({'branch': 'dev', 'read.native.enabled': 'true'})
+ assert branch.catalog_environment.rest_table_response ==
table.catalog_environment.rest_table_response
+ assert _resolved_rest_table_response(branch) is None
+ plan = native_plan(branch)
+ assert plan.snapshot_id == (1 if from_tag else None)
+ with patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('native read fell back')):
+ rows =
branch.new_read_builder().new_read().to_arrow(plan.splits()).to_pylist()
+ assert rows == ([{'id': 1, 'value': 'old'}] if from_tag else [])
+
+
def test_resolved_rest_table_keeps_refreshable_file_io(rest_source,
rest_catalog):
from pypaimon.api.api_response import GetTableTokenResponse
from pypaimon.read.native_plan import _resolved_schema_json
@@ -196,3 +215,91 @@ def
test_rest_blob_view_limit_filters_before_resolving_unselected_view(rest_cata
side_effect=AssertionError('native view read fell back')):
assert builder.new_read().to_arrow(plan.splits()).to_pylist() == [
{'id': 11, 'payload': b'selected'}]
+
+
+def test_reused_rest_environment_sees_new_snapshot(rest_source):
+ from pypaimon.tests.native_plan_resolved_schema_test import
_assert_parity, _write
+ table, _, _ = rest_source
+ rows = [{'id': 1, 'value': 'old'}, {'id': 2, 'value': 'new'}]
+ _assert_parity(table, rows, 2)
+ _write(table, [{'id': 3, 'value': 'latest'}])
+ _assert_parity(table.copy({'read.batch-size': '1'}), rows + [{'id': 3,
'value': 'latest'}], 3)
+
+
[email protected](not native_method_available('Table', 'from_rest_response'),
+ reason='REST response binding required')
+def test_repeated_plans_reuse_remote_file_sizes(rest_source, tmp_path):
+ import json
+ from collections import Counter
+ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+ from pathlib import Path
+ from threading import Thread
+ from urllib.parse import urlparse
+
+ from pypaimon.catalog.catalog_context import CatalogContext
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+ from pypaimon.common.options.options import Options
+ from pypaimon.read.native_plan import native_plan
+
+ table, _, _ = rest_source
+ root = Path(urlparse(table.table_path).path)
+ requests = Counter()
+
+ class ObjectStore(BaseHTTPRequestHandler):
+ def do_HEAD(self):
+ self.serve()
+
+ def do_GET(self):
+ self.serve()
+
+ def serve(self):
+ path = urlparse(self.path).path
+ requests[self.command, path] += 1
+ file = root / path[len('/bucket/t/'):]
+ if not file.is_file():
+ self.send_error(404)
+ return
+ data = file.read_bytes()
+ size = len(data)
+ byte_range = self.headers.get('Range')
+ self.send_response(206 if byte_range else 200)
+ if byte_range:
+ start, end = byte_range[6:].split('-')
+ start, end = int(start), int(end) if end else size - 1
+ data = data[start:end + 1]
+ self.send_header('Content-Range', 'bytes %s-%s/%s' % (start,
end, size))
+ self.send_header('Content-Length', str(len(data)))
+ self.end_headers()
+ if self.command == 'GET':
+ self.wfile.write(data)
+
+ def log_message(self, *args):
+ pass
+
+ server = ThreadingHTTPServer(('127.0.0.1', 0), ObjectStore)
+ thread = Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ try:
+ options =
dict(table.catalog_environment.catalog_loader.context().options.to_map(), **{
+ 's3.endpoint': 'http://127.0.0.1:%s' % server.server_port,
+ 's3.region': 'us-east-1', 's3.path.style.access': 'true',
's3.anonymous': 'true',
+ 'local-cache.enabled': 'true', 'local-cache.dir': str(tmp_path /
'cache')})
+ response = json.loads(table.catalog_environment.rest_table_response)
+ response['path'] = 's3://bucket/t'
+ table.table_path = response['path']
+ table.catalog_environment = CatalogEnvironment(
+ identifier=table.identifier, uuid=response['id'],
supports_version_management=True,
+
catalog_loader=RESTCatalogLoader(CatalogContext.create_from_options(Options(options))),
+ rest_table_response=json.dumps(response))
+ first = native_plan(table)
+ initial = requests.copy()
+ assert sum(n for (method, _), n in initial.items() if method ==
'HEAD') > 0
+ second = native_plan(table.copy({'read.batch-size': '1'}))
+ assert second.snapshot_id == first.snapshot_id == 2
+ assert len(second.splits()) == len(first.splits()) > 0
+ assert requests == initial
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join()
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 5c6c6078d6..ccbc92e160 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -19,7 +19,7 @@ import json
import sys
import unittest
from types import ModuleType, SimpleNamespace
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, call, patch
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.filesystem_catalog_loader import FileSystemCatalogLoader
@@ -786,6 +786,241 @@ class NativePlanTest(unittest.TestCase):
with self.assertRaises(PermissionError):
_restore_python_partition_paths(table, [split])
+ def test_rest_catalog_retains_response_for_native_reads(self):
+ from pypaimon.api.api_response import GetTableResponse
+ from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.schema.schema import Schema
+
+ catalog = RESTCatalog.__new__(RESTCatalog)
+ catalog.context =
CatalogContext.create_from_options(Options({'warehouse': 'test'}))
+ catalog.create = Mock()
+ identifier = Identifier.create('db', 't')
+ response = GetTableResponse(
+ 'uuid', 't', '/warehouse/t', True, 3,
+ Schema(fields=[DataField(4, 'id', AtomicType('INT'))]))
+ metadata = catalog.to_table_metadata('db', response)
+ catalog.load_table(identifier, Mock(), Mock(), lambda _: metadata)
+ environment = catalog.create.call_args[0][3]
+ saved = json.loads(environment.rest_table_response)
+ self.assertEqual(saved['id'], 'uuid')
+ self.assertTrue(saved['isExternal'])
+ self.assertEqual(saved['schemaId'], 3)
+ self.assertEqual(saved['schema']['fields'][0]['id'], 4)
+
+ def test_rest_response_requires_native_support_and_matching_path(self):
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.read.native_plan import _resolved_rest_table_response
+
+ table = Mock(table_path='/warehouse/t')
+ loader =
RESTCatalogLoader(CatalogContext.create_from_options(Options({})))
+ table.catalog_environment = CatalogEnvironment(
+ catalog_loader=loader, rest_table_response='{"path":
"/warehouse/t"}')
+ with patch('pypaimon.read.native_plan.native_method_available',
return_value=False):
+ self.assertIsNone(_resolved_rest_table_response(table))
+ with patch('pypaimon.read.native_plan.native_method_available',
return_value=True):
+ table.table_path = '/another/table'
+ self.assertIsNone(_resolved_rest_table_response(table))
+ table.table_path = '/warehouse/t'
+ table.catalog_environment.rest_table_response = None
+ self.assertIsNone(_resolved_rest_table_response(table))
+
+ def test_rest_response_requires_matching_identity(self):
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.read.native_plan import _resolved_rest_table_response
+
+ table = Mock(table_path='/warehouse/t', identifier=Identifier('db',
't', branch='dev'))
+ loader =
RESTCatalogLoader(CatalogContext.create_from_options(Options({})))
+ table.catalog_environment = CatalogEnvironment(catalog_loader=loader)
+ for identity, matches in [
+ ({'name': 't'}, False),
+ ({}, False),
+ ({'name': 't$branch_dev'}, True),
+ ({'name': 't$branch_dev', 'database': 'db'}, True),
+ ({'name': 't$branch_dev', 'database': 'other'}, False)]:
+ with self.subTest(identity=identity), patch(
+ 'pypaimon.read.native_plan.native_method_available',
return_value=True):
+ response = json.dumps(dict(identity, path=table.table_path))
+ table.catalog_environment.rest_table_response = response
+ self.assertEqual(_resolved_rest_table_response(table),
response if matches else None)
+
+ def test_rest_native_builder_reuses_loaded_metadata(self):
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.read.native_plan import _native_read_builder
+
+ response = json.dumps({'name': 't$branch_dev', 'path': '/warehouse/t',
+ 'id': 'uuid', 'isExternal': False})
+ loader = RESTCatalogLoader(CatalogContext.create_from_options(Options({
+ 'uri': 'http://localhost:1', 'warehouse': 'test',
'data-token.enabled': 'true'})))
+ table = Mock()
+ table.identifier = Identifier('db', 't', branch='dev')
+ table.table_path = '/warehouse/t'
+ table.current_branch.return_value = 'dev'
+ table.catalog_environment = CatalogEnvironment(
+ identifier=table.identifier, uuid='uuid', catalog_loader=loader,
+ supports_version_management=True, rest_table_response=response)
+
self.assertEqual(table.catalog_environment.copy(table.identifier).rest_table_response,
response)
+ resolved = '{"id": 2, "options": {"blob-as-descriptor": "true"}}'
+ native_table = Mock()
+ native_table.copy_with_resolved_schema.return_value = native_table
+ native_table.branch.return_value = 'dev'
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.Table = Mock()
+ fake_df.Table.from_rest_response.return_value = native_table
+ fake_df.PaimonCatalog = Mock()
+ fake_module = ModuleType('pypaimon_rust')
+ fake_module.datafusion = fake_df
+ with patch.dict(sys.modules, {'pypaimon_rust': fake_module,
+ 'pypaimon_rust.datafusion': fake_df}), \
+ patch('pypaimon.read.native_plan._resolved_schema_json',
return_value=resolved):
+ for _ in range(3):
+ self.assertIs(_native_read_builder(table),
native_table.new_read_builder.return_value)
+ fake_df.PaimonCatalog.assert_not_called()
+ fake_df.Table.from_rest_response.assert_called_once_with(
+ response, database='db', table='t$branch_dev',
+ rest_options=_catalog_options(table))
+ self.assertEqual(native_table.copy_with_resolved_schema.call_args_list,
+ [call(resolved, branch='dev')] * 3)
+ self.assertEqual(native_table.new_read_builder.call_count, 3)
+
+ def test_native_rest_cache_invalidation_and_serialization(self):
+ import pickle
+ from pypaimon.read.native_plan import _NativeRestTableCache
+
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.Table = Mock()
+ fake_df.Table.from_rest_response.side_effect = lambda *args, **kwargs:
object()
+ cache = _NativeRestTableCache()
+ with patch.dict(sys.modules, {'pypaimon_rust.datafusion': fake_df}):
+ original = cache.get('response', 'db', 't', {'token': 'first'})
+ self.assertIs(cache.get('response', 'db', 't', {'token':
'first'}), original)
+ for response, db, table, options in [
+ ('response', 'db', 't', {'token': 'second'}),
+ ('new-response', 'db', 't', {'token': 'second'}),
+ ('new-response', 'db', 't$branch_dev', {'token':
'second'}),
+ ('new-response', 'other', 't$branch_dev', {'token':
'second'})]:
+ replacement = cache.get(response, db, table, options)
+ self.assertIsNot(replacement, original)
+ original = replacement
+ restored = pickle.loads(pickle.dumps(cache))
+ self.assertEqual(restored._states, {})
+ self.assertIsNot(restored.get(response, db, table, options),
original)
+ with patch('pypaimon.read.native_plan.os.getpid', return_value=-1):
+ self.assertIsNot(cache.get(response, db, table, options),
original)
+
+ def test_native_rest_cache_lifetime_and_concurrent_access(self):
+ import gc
+ import weakref
+ from concurrent.futures import ThreadPoolExecutor
+ from pypaimon.read.native_plan import _NativeRestTableCache
+
+ class NativeTable:
+ pass
+
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.Table = Mock()
+ fake_df.Table.from_rest_response.side_effect = lambda *args, **kwargs:
NativeTable()
+ cache = _NativeRestTableCache()
+ with patch.dict(sys.modules, {'pypaimon_rust.datafusion': fake_df}):
+ with ThreadPoolExecutor(max_workers=4) as pool:
+ tables = list(pool.map(lambda _: cache.get('response', 'db',
't', {}), range(8)))
+ self.assertTrue(all(table is tables[0] for table in tables))
+ fake_df.Table.from_rest_response.assert_called_once()
+ reference = weakref.ref(tables[0])
+ del tables
+ gc.collect()
+ self.assertIsNotNone(reference())
+ del cache
+ gc.collect()
+ self.assertIsNone(reference())
+
+ def test_native_rest_cache_concurrent_first_access_after_fork(self):
+ import multiprocessing
+ import os
+ from threading import Event, Lock, Thread, current_thread
+ from pypaimon.read.native_plan import _NativeRestTableCache
+
+ if 'fork' not in multiprocessing.get_all_start_methods():
+ self.skipTest('fork required')
+ context = multiprocessing.get_context('fork')
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.Table = Mock()
+ fake_df.Table.from_rest_response.side_effect = lambda *args, **kwargs:
object()
+ cache = _NativeRestTableCache()
+ held, release = Event(), Event()
+
+ def hold_parent_lock():
+ with cache._states[os.getpid()].lock:
+ held.set()
+ release.wait()
+
+ def child(connection):
+ creating, resume = Event(), Event()
+ results = []
+ fake_df.Table.from_rest_response.reset_mock()
+
+ def new_lock():
+ if current_thread().name == 'first':
+ creating.set()
+ assert resume.wait(5)
+ return Lock()
+
+ def access():
+ results.append(cache.get('response', 'db', 't', {}))
+
+ with patch('pypaimon.read.native_plan.Lock', side_effect=new_lock):
+ first = Thread(target=access, name='first', daemon=True)
+ second = Thread(target=access, name='second', daemon=True)
+ first.start()
+ assert creating.wait(5)
+ second.start()
+ second.join(2)
+ second_completed = not second.is_alive()
+ resume.set()
+ first.join(2)
+ connection.send((second_completed, not first.is_alive(),
+ len(results) == 2 and results[0] is
results[1],
+ fake_df.Table.from_rest_response.call_count))
+ connection.close()
+
+ with patch.dict(sys.modules, {'pypaimon_rust.datafusion': fake_df}):
+ cache.get('response', 'db', 't', {})
+ holder = Thread(target=hold_parent_lock, daemon=True)
+ holder.start()
+ self.assertTrue(held.wait(5))
+ receiving, sending = context.Pipe(duplex=False)
+ process = context.Process(target=child, args=(sending,))
+ try:
+ process.start()
+ sending.close()
+ self.assertTrue(receiving.poll(10), 'child deadlocked on
inherited lock')
+ self.assertEqual(receiving.recv(), (True, True, True, 1))
+ process.join(5)
+ self.assertEqual(process.exitcode, 0)
+ finally:
+ if process.is_alive():
+ process.terminate()
+ process.join(5)
+ receiving.close()
+ sending.close()
+ release.set()
+ holder.join(5)
+
+ def test_native_rest_cache_retries_failed_construction(self):
+ from pypaimon.read.native_plan import _NativeRestTableCache
+
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.Table = Mock()
+ native_table = object()
+ fake_df.Table.from_rest_response.side_effect =
[RuntimeError('unavailable'), native_table]
+ cache = _NativeRestTableCache()
+ with patch.dict(sys.modules, {'pypaimon_rust.datafusion': fake_df}):
+ with self.assertRaisesRegex(RuntimeError, 'unavailable'):
+ cache.get('response', 'db', 't', {})
+ self.assertIs(cache.get('response', 'db', 't', {}), native_table)
+
def test_native_plan_threads_trimmed_keys_to_deserializer(self):
# PK tables route through: the trimmed primary keys must reach the
# deserializer so per-file min/max keys are decoded for merge-on-read.
diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
index e3c8951fa9..5b13d6f195 100644
--- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import json
import unittest
import pyarrow as pa
@@ -91,6 +92,9 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
self.assertEqual(table.current_branch(), "b1")
self.assertEqual(table.schema_manager.branch, "b1")
+ self.assertEqual(
+ json.loads(table.catalog_environment.rest_table_response)['name'],
+ identifier.get_object_name() + '$branch_b1')
def test_write_blob_to_data_evolution_branch(self):
schema = pa.schema([
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index 612bf7393b..1c5826b03f 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -537,7 +537,8 @@ class RESTCatalogServer:
if len(path_parts) == 3:
# Basic table operations (GET, DELETE, etc.)
- return self._table_handle(method, data, lookup_identifier)
+ return self._table_handle(
+ method, data, lookup_identifier,
response_identifier=identifier)
elif len(path_parts) == 4:
# Extended operations (e.g., commit, token, snapshot)
operation = path_parts[3]
@@ -1028,7 +1029,8 @@ class RESTCatalogServer:
return self._mock_response("", 200)
return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
- def _table_handle(self, method: str, data: str, identifier: Identifier) ->
Tuple[str, int]:
+ def _table_handle(self, method: str, data: str, identifier: Identifier,
+ response_identifier: Identifier = None) -> Tuple[str,
int]:
"""Handle individual table operations"""
if method == "GET":
if identifier.get_full_name() not in self.table_metadata_store:
@@ -1037,7 +1039,8 @@ class RESTCatalogServer:
table_path = (f'file://{self.data_path}/{self.warehouse}/'
f'{identifier.get_database_name()}/{identifier.get_object_name()}')
schema = table_metadata.schema.to_schema()
- response = self.mock_table(identifier, table_metadata, table_path,
schema)
+ response = self.mock_table(
+ response_identifier or identifier, table_metadata, table_path,
schema)
return self._mock_response(response, 200)
elif method == "POST":
# Alter table
