[python] Reuse resolved REST metadata for native readers

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

paimon-python/pypaimon/catalog/catalog_environment.py

@@ -39,12 +39,14 @@ def __init__(
             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 @@ def copy(self, identifier: Identifier) -> '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

paimon-python/pypaimon/catalog/rest/rest_catalog.py

@@ -45,6 +45,7 @@
 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 @@ def to_table_metadata(self, db: str, response: GetTableResponse) -> TableMetadat
         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 @@ def load_table(self,
             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())

paimon-python/pypaimon/catalog/rest/table_metadata.py

@@ -22,10 +22,12 @@
 
 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:

paimon-python/pypaimon/read/native_plan.py

@@ -22,6 +22,10 @@
 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),

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)

paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py

@@ -23,7 +23,7 @@
 
 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:

paimon-python/pypaimon/tests/native_plan_rest_test.py

@@ -23,7 +23,7 @@
 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)
 
 
+@pytest.mark.parametrize('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)
+
+
+@pytest.mark.skipif(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()