From 202db48caf3bf84de4910dc74e85024613567aa8 Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Tue, 22 Sep 2026 09:57:01 -0700
Subject: [PATCH 1/7] [python] Reuse resolved REST metadata for native readers

---
 .../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    | 33 ++++++++-
 .../pypaimon/tests/native_plan_test.py        | 74 +++++++++++++++++++
 5 files changed, 119 insertions(+), 6 deletions(-)

diff --git a/paimon-python/pypaimon/catalog/catalog_environment.py b/paimon-python/pypaimon/catalog/catalog_environment.py
index 84530fbc0a8d..badb14734083 100644
--- a/paimon-python/pypaimon/catalog/catalog_environment.py
+++ b/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
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index a572c947f9f3..1b7f58e4f6b9 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.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())
diff --git a/paimon-python/pypaimon/catalog/rest/table_metadata.py b/paimon-python/pypaimon/catalog/rest/table_metadata.py
index e9a7221f3736..3226081ce4b0 100644
--- a/paimon-python/pypaimon/catalog/rest/table_metadata.py
+++ b/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:
diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py
index 6eac710aaa1b..63cd3ed8f352 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -22,6 +22,7 @@
 still applies them while reading, so pushdown remains an optimization.
 """
 
+import json
 from typing import List, Optional, Tuple
 
 from packaging.version import InvalidVersion, Version
@@ -291,10 +292,40 @@ 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
+    if json.loads(response).get('path') != table.table_path:
+        return None
+    return response
+
+
 def _native_read_builder(table):
     """Reconstruct the Rust table and return a builder for the same schema."""
+    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:
+        from pypaimon_rust.datafusion import Table
+        rt = Table.from_rest_response(
+            rest_response,
+            database=table.identifier.get_database_name(),
+            table=table.identifier.get_table_name(), options=_catalog_options(table))
+        rt = rt.copy_with_resolved_schema(_resolved_schema_json(table), branch=table.current_branch())
+        builder = rt.new_read_builder()
+    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/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py
index 124a64c4c2f0..ca2bba737ed4 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -763,6 +763,80 @@ def test_partition_path_listing_failure_is_not_hidden(self):
         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_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({'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.create('db', 't')
+        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):
+            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', options=_catalog_options(table))
+        native_table.copy_with_resolved_schema.assert_called_once_with(resolved, branch='dev')
+
     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.

From 372b2e5ef74abf794b43f6bbcc48e09fe70ea4d0 Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 24 Sep 2026 02:14:32 -0700
Subject: [PATCH 2/7] [python] Align resolved REST table binding contract

---
 paimon-python/pypaimon/read/native_plan.py               | 3 ++-
 .../pypaimon/tests/native_plan_resolved_schema_test.py   | 5 +++--
 paimon-python/pypaimon/tests/native_plan_test.py         | 5 +++--
 paimon-python/pypaimon/tests/rest/rest_branch_test.py    | 4 ++++
 paimon-python/pypaimon/tests/rest/rest_server.py         | 9 ++++++---
 5 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py
index 63cd3ed8f352..cf45ad10aeb7 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -322,7 +322,8 @@ def _native_read_builder(table):
         rt = Table.from_rest_response(
             rest_response,
             database=table.identifier.get_database_name(),
-            table=table.identifier.get_table_name(), options=_catalog_options(table))
+            table=table.identifier.get_object_name(),
+            rest_options=_catalog_options(table))
         rt = rt.copy_with_resolved_schema(_resolved_schema_json(table), branch=table.current_branch())
         builder = rt.new_read_builder()
     elif file_io_options is not None:
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 cbae972b0907..bfa0e0ce8a31 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 @@
 
 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_test.py b/paimon-python/pypaimon/tests/native_plan_test.py
index ca2bba737ed4..50b754bec537 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -811,7 +811,7 @@ def test_rest_native_builder_reuses_loaded_metadata(self):
         loader = RESTCatalogLoader(CatalogContext.create_from_options(Options({
             'uri': 'http://localhost:1', 'warehouse': 'test', 'data-token.enabled': 'true'})))
         table = Mock()
-        table.identifier = Identifier.create('db', 't')
+        table.identifier = Identifier('db', 't', branch='dev')
         table.table_path = '/warehouse/t'
         table.current_branch.return_value = 'dev'
         table.catalog_environment = CatalogEnvironment(
@@ -834,7 +834,8 @@ def test_rest_native_builder_reuses_loaded_metadata(self):
             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', options=_catalog_options(table))
+            response, database='db', table='t$branch_dev',
+            rest_options=_catalog_options(table))
         native_table.copy_with_resolved_schema.assert_called_once_with(resolved, branch='dev')
 
     def test_native_plan_threads_trimmed_keys_to_deserializer(self):
diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
index e3c8951fa97d..5b13d6f195d0 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 @@ def test_branch_table_uses_branch_schema_manager(self):
 
         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 612bf7393b57..1c5826b03fd0 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -537,7 +537,8 @@ def _handle_table_resource(self, method: str, path_parts: List[str],
 
         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 @@ def _tables_handle(self, method: str = None, data: str = None, database_name: st
                 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 @@ def _table_handle(self, method: str, data: str, identifier: Identifier) -> Tuple
             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

From d05b38d95935dd86f481a95785bfc0d16e68332a Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 24 Sep 2026 05:09:53 -0700
Subject: [PATCH 3/7] [python] Create resolved REST read builder only once

---
 paimon-python/pypaimon/read/native_plan.py       | 1 -
 paimon-python/pypaimon/tests/native_plan_test.py | 1 +
 2 files changed, 1 insertion(+), 1 deletion(-)

diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py
index 5e5ba0314607..683c657b78e6 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -285,7 +285,6 @@ def _native_read_builder(table):
             table=table.identifier.get_object_name(),
             rest_options=_catalog_options(table))
         rt = rt.copy_with_resolved_schema(_resolved_schema_json(table), branch=table.current_branch())
-        builder = rt.new_read_builder()
     elif file_io_options is not None:
         from pypaimon_rust.datafusion import Table
         rt = Table.from_resolved_schema(
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py
index 420d531b248c..4c49dd29153a 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -859,6 +859,7 @@ def test_rest_native_builder_reuses_loaded_metadata(self):
             response, database='db', table='t$branch_dev',
             rest_options=_catalog_options(table))
         native_table.copy_with_resolved_schema.assert_called_once_with(resolved, branch='dev')
+        native_table.new_read_builder.assert_called_once_with()
 
     def test_native_plan_threads_trimmed_keys_to_deserializer(self):
         # PK tables route through: the trimmed primary keys must reach the

From ce325937be9af9fd09f12bf8e4a53fb66c2ce880 Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 24 Sep 2026 06:53:19 -0700
Subject: [PATCH 4/7] [python] Validate cached REST table identity before
 native reuse

---
 paimon-python/pypaimon/read/native_plan.py    |  6 ++++-
 .../pypaimon/tests/native_plan_rest_test.py   | 19 +++++++++++++++
 .../pypaimon/tests/native_plan_test.py        | 23 ++++++++++++++++++-
 3 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py
index 683c657b78e6..b522d61ef51c 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -268,7 +268,11 @@ def _resolved_rest_table_response(table):
     if any(getattr(context, attr, None) is not None for attr in (
             'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')):
         return None
-    if json.loads(response).get('path') != table.table_path:
+    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
 
diff --git a/paimon-python/pypaimon/tests/native_plan_rest_test.py b/paimon-python/pypaimon/tests/native_plan_rest_test.py
index a2c2a92fde5d..063d22b9668f 100644
--- a/paimon-python/pypaimon/tests/native_plan_rest_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_rest_test.py
@@ -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
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py
index 4c49dd29153a..55eca4f9a931 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -824,12 +824,33 @@ def test_rest_response_requires_native_support_and_matching_path(self):
             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({'path': '/warehouse/t', 'id': 'uuid', 'isExternal': False})
+        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()

From 71ac5afeeba406b87baa0b45ad0d15a2fc436fac Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 24 Sep 2026 06:55:03 -0700
Subject: [PATCH 5/7] [python] Supply path factory in split grouping test

---
 paimon-python/pypaimon/tests/interval_partition_test.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/paimon-python/pypaimon/tests/interval_partition_test.py b/paimon-python/pypaimon/tests/interval_partition_test.py
index 173ed6c9eee3..822d3e64d1b3 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)

From 61796897c891b9115aef2681252a00d2c679e360 Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 25 Sep 2026 00:58:17 -0700
Subject: [PATCH 6/7] [python] Retain native REST environments across scans

---
 paimon-python/README.md                       |  2 +
 paimon-python/pypaimon/read/native_plan.py    | 39 +++++++-
 .../pypaimon/tests/native_plan_rest_test.py   | 90 ++++++++++++++++++-
 .../pypaimon/tests/native_plan_test.py        | 74 ++++++++++++++-
 4 files changed, 196 insertions(+), 9 deletions(-)

diff --git a/paimon-python/README.md b/paimon-python/README.md
index 96f7ba0d9f75..d83b94751866 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/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py
index b522d61ef51c..f7e202dece84 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -23,6 +23,8 @@
 """
 
 import json
+import os
+from threading import Lock
 from typing import List, Optional, Tuple
 
 from packaging.version import InvalidVersion, Version
@@ -277,17 +279,46 @@ def _resolved_rest_table_response(table):
     return response
 
 
+class _NativeRestTableCache:
+    """One native environment per Python environment, never sent to workers."""
+
+    def __init__(self):
+        self.pid = os.getpid()
+        self.lock = Lock()
+        self.entry = None
+
+    def __getstate__(self):
+        return {}
+
+    def __setstate__(self, state):
+        self.__init__()
+
+    def get(self, response, database, table, options):
+        from pypaimon_rust.datafusion import Table
+
+        if self.pid != os.getpid():
+            self.__init__()
+        key = (response, database, table, tuple(sorted(options.items())))
+        with self.lock:
+            if self.entry is None or self.entry[0] != key:
+                native_table = Table.from_rest_response(
+                    response, database=database, table=table, rest_options=options)
+                self.entry = (key, native_table)
+            return self.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 rest_response is not None:
-        from pypaimon_rust.datafusion import Table
-        rt = Table.from_rest_response(
+        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(),
-            rest_options=_catalog_options(table))
+            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
diff --git a/paimon-python/pypaimon/tests/native_plan_rest_test.py b/paimon-python/pypaimon/tests/native_plan_rest_test.py
index 063d22b9668f..320bfd61bc00 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 @@
 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
@@ -215,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()
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py
index 55eca4f9a931..0133a1303e97 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -19,7 +19,7 @@
 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
@@ -874,13 +874,79 @@ def test_rest_native_builder_reuses_loaded_metadata(self):
         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):
-            self.assertIs(_native_read_builder(table), native_table.new_read_builder.return_value)
+            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))
-        native_table.copy_with_resolved_schema.assert_called_once_with(resolved, branch='dev')
-        native_table.new_read_builder.assert_called_once_with()
+        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.assertIsNone(restored.entry)
+            self.assertIsNot(restored.get(response, db, table, options), original)
+            cache.pid = -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_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

From 2c37f97578868b14f5b1a73774576633f2f35746 Mon Sep 17 00:00:00 2001
From: "xiaohongbo.xhb" <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 25 Sep 2026 01:19:15 -0700
Subject: [PATCH 7/7] [python] Publish native cache state atomically after fork

---
 paimon-python/pypaimon/read/native_plan.py    | 22 +++---
 .../pypaimon/tests/native_plan_test.py        | 78 ++++++++++++++++++-
 2 files changed, 88 insertions(+), 12 deletions(-)

diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py
index f7e202dece84..c314a75e5cc2 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -25,6 +25,7 @@
 import json
 import os
 from threading import Lock
+from types import SimpleNamespace
 from typing import List, Optional, Tuple
 
 from packaging.version import InvalidVersion, Version
@@ -283,9 +284,7 @@ class _NativeRestTableCache:
     """One native environment per Python environment, never sent to workers."""
 
     def __init__(self):
-        self.pid = os.getpid()
-        self.lock = Lock()
-        self.entry = None
+        self._states = {}
 
     def __getstate__(self):
         return {}
@@ -296,15 +295,20 @@ def __setstate__(self, state):
     def get(self, response, database, table, options):
         from pypaimon_rust.datafusion import Table
 
-        if self.pid != os.getpid():
-            self.__init__()
+        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 self.lock:
-            if self.entry is None or self.entry[0] != key:
+        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)
-                self.entry = (key, native_table)
-            return self.entry[1]
+                state.entry = (key, native_table)
+            return state.entry[1]
 
 
 def _native_read_builder(table):
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py
index 0133a1303e97..1bd926dffa79 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -904,10 +904,10 @@ def test_native_rest_cache_invalidation_and_serialization(self):
                 self.assertIsNot(replacement, original)
                 original = replacement
             restored = pickle.loads(pickle.dumps(cache))
-            self.assertIsNone(restored.entry)
+            self.assertEqual(restored._states, {})
             self.assertIsNot(restored.get(response, db, table, options), original)
-            cache.pid = -1
-            self.assertIsNot(cache.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
@@ -935,6 +935,78 @@ class NativeTable:
             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