[python] Propagate dependency read context in REST headers

paimon-python/pypaimon/api/rest_api.py

@@ -58,6 +58,7 @@
 
 class RESTApi:
     HEADER_PREFIX = "header."
+    READ_VIA_HEADER = "X-Paimon-Read-Via"
     MAX_RESULTS = "maxResults"
     PAGE_TOKEN = "pageToken"
     DATABASE_NAME_PATTERN = "databaseNamePattern"

paimon-python/pypaimon/catalog/catalog_environment.py

@@ -17,8 +17,13 @@
 
 from typing import Optional
 
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.api.rest_util import RESTUtil
+from pypaimon.catalog.catalog_context import CatalogContext
 from pypaimon.catalog.catalog_loader import CatalogLoader
 from pypaimon.common.identifier import Identifier
+from pypaimon.common.json_util import JSON
+from pypaimon.common.options.config import CatalogOptions
 from pypaimon.snapshot.catalog_snapshot_commit import CatalogSnapshotCommit
 from pypaimon.snapshot.renaming_snapshot_commit import RenamingSnapshotCommit
 from pypaimon.snapshot.snapshot_commit import SnapshotCommit
@@ -27,6 +32,8 @@
 
 class CatalogEnvironment:
 
+    _READ_VIA_OPTION = RESTApi.HEADER_PREFIX + RESTApi.READ_VIA_HEADER
+
     def __init__(
             self,
             identifier: Optional[Identifier] = None,
@@ -86,6 +93,41 @@ def snapshot_loader(self) -> Optional[SnapshotLoader]:
             return SnapshotLoader(self.catalog_loader, self.identifier)
         return None
 
+    def catalog_context(self) -> Optional[CatalogContext]:
+        if self.catalog_loader is None:
+            return None
+        context = getattr(self.catalog_loader, "context", None)
+        return context() if callable(context) else None
+
+    def dependency_read_context(self) -> Optional[CatalogContext]:
+        context = self.catalog_context()
+        if self.identifier is None or context is None:
+            return context
+
+        from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+
+        rest_catalog = (
+            isinstance(self.catalog_loader, RESTCatalogLoader)
+            or context.options.get(CatalogOptions.METASTORE) == "rest"
+        )
+        if not rest_catalog:
+            return context
+        if context.options.contains_key(self._READ_VIA_OPTION):
+            return context
+
+        dependency_options = context.options.copy()
+        if not dependency_options.contains(CatalogOptions.METASTORE):
+            dependency_options.set(CatalogOptions.METASTORE, "rest")
+        dependency_options.to_map()[self._READ_VIA_OPTION] = RESTUtil.encode_string(
+            JSON.to_json(self.identifier, separators=(",", ":"))
+        )
+        return CatalogContext.create(
+            dependency_options,
+            context.hadoop_conf,
+            context.prefer_io_loader,
+            context.fallback_io_loader,
+        )
+
     def copy(self, identifier: Identifier) -> 'CatalogEnvironment':
         """
         Create a copy of this CatalogEnvironment with a different identifier.

paimon-python/pypaimon/catalog/catalog_factory.py

@@ -44,3 +44,20 @@ def create(catalog_options: Dict) -> Catalog:
         if identifier in ("jdbc", "rest"):
             return catalog_class(CatalogContext.create_from_options(Options(catalog_options)))
         return catalog_class(Options(catalog_options))
+
+    @staticmethod
+    def create_from_context(
+            context: CatalogContext,
+            config_required: bool = True
+    ) -> Catalog:
+        identifier = context.options.get(CatalogOptions.METASTORE)
+        catalog_class = CatalogFactory.CATALOG_REGISTRY.get(identifier)
+        if catalog_class is None:
+            raise ValueError(
+                "Unknown catalog identifier: {}. Available types: {}".format(
+                    identifier, list(CatalogFactory.CATALOG_REGISTRY.keys())))
+        if identifier == "filesystem":
+            return catalog_class(context.options)
+        if identifier == "rest":
+            return catalog_class(context, config_required=config_required)
+        return catalog_class(context)

paimon-python/pypaimon/tests/catalog_environment_test.py

@@ -0,0 +1,202 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.api.rest_util import RESTUtil
+from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.catalog.catalog_factory import CatalogFactory
+from pypaimon.catalog.catalog_environment import CatalogEnvironment
+from pypaimon.catalog.filesystem_catalog_loader import FileSystemCatalogLoader
+from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+from pypaimon.common.identifier import Identifier
+from pypaimon.common.json_util import JSON
+from pypaimon.common.options import Options
+from pypaimon.common.options.config import CatalogOptions
+from pypaimon.utils.blob_view_lookup import BlobViewLookup
+
+
+class CatalogEnvironmentTest(unittest.TestCase):
+
+    _READ_VIA_OPTION = RESTApi.HEADER_PREFIX + RESTApi.READ_VIA_HEADER
+
+    def test_dependency_read_context_for_rest_catalog(self):
+        root = Identifier.create("db", "root", branch="dev")
+        options = Options({"other-option": "value"})
+        context = CatalogContext.create_from_options(options)
+        environment = CatalogEnvironment(
+            identifier=root,
+            catalog_loader=RESTCatalogLoader(context),
+        )
+
+        dependency_context = environment.dependency_read_context()
+
+        self.assertIsNot(dependency_context, context)
+        self.assertFalse(context.options.contains_key(self._READ_VIA_OPTION))
+        self.assertEqual(
+            dependency_context.options.get(CatalogOptions.METASTORE), "rest")
+        self.assertEqual(
+            dependency_context.options.to_map()["other-option"], "value")
+        read_via = JSON.from_json(
+            RESTUtil.decode_string(
+                dependency_context.options.to_map()[self._READ_VIA_OPTION]),
+            Identifier,
+        )
+        self.assertEqual(read_via, root)
+
+    def test_dependency_read_context_preserves_outermost_table(self):
+        outermost = Identifier.create("db", "outermost")
+        read_via = RESTUtil.encode_string(
+            JSON.to_json(outermost, separators=(",", ":")))
+        context = CatalogContext.create_from_options(Options({
+            CatalogOptions.METASTORE.key(): "rest",
+            self._READ_VIA_OPTION: read_via,
+        }))
+        environment = CatalogEnvironment(
+            identifier=Identifier.create("db", "intermediate"),
+            catalog_loader=RESTCatalogLoader(context),
+        )
+
+        self.assertIs(environment.dependency_read_context(), context)
+        self.assertEqual(
+            context.options.to_map()[self._READ_VIA_OPTION], read_via)
+
+    def test_dependency_read_context_does_not_affect_other_catalogs(self):
+        context = CatalogContext.create_from_options(Options({}))
+        environment = CatalogEnvironment(
+            identifier=Identifier.create("db", "table"),
+            catalog_loader=FileSystemCatalogLoader(context),
+        )
+
+        self.assertIs(environment.dependency_read_context(), context)
+        self.assertFalse(context.options.contains_key(self._READ_VIA_OPTION))
+
+    def test_dependency_read_context_for_external_rest_table(self):
+        context = CatalogContext.create_from_options(Options({
+            CatalogOptions.METASTORE.key(): "rest",
+        }))
+        environment = CatalogEnvironment(
+            identifier=Identifier.create("db", "external"),
+            catalog_loader=FileSystemCatalogLoader(context),
+        )
+
+        self.assertIsNot(environment.dependency_read_context(), context)
+
+    def test_dependency_read_context_preserves_custom_rest_metastore(self):
+        context = CatalogContext.create_from_options(Options({
+            CatalogOptions.METASTORE.key(): "custom-rest",
+        }))
+        environment = CatalogEnvironment(
+            identifier=Identifier.create("db", "table"),
+            catalog_loader=RESTCatalogLoader(context),
+        )
+
+        dependency_context = environment.dependency_read_context()
+
+        self.assertIsNot(dependency_context, context)
+        self.assertEqual(
+            dependency_context.options.get(CatalogOptions.METASTORE),
+            "custom-rest",
+        )
+
+    def test_catalog_factory_creates_rest_from_context(self):
+        context = CatalogContext.create_from_options(Options({
+            CatalogOptions.METASTORE.key(): "rest",
+        }))
+        dependency_catalog = mock.sentinel.dependency_catalog
+        rest_catalog = mock.Mock(return_value=dependency_catalog)
+
+        with mock.patch.dict(
+                CatalogFactory.CATALOG_REGISTRY, {"rest": rest_catalog}):
+            result = CatalogFactory.create_from_context(
+                context, config_required=False)
+
+        rest_catalog.assert_called_once_with(context, config_required=False)
+        self.assertIs(result, dependency_catalog)
+
+    def test_blob_view_lookup_loads_dependency_catalog(self):
+        root = Identifier.create("db", "root")
+        target = Identifier.create("db", "target")
+        context = CatalogContext.create_from_options(Options({
+            CatalogOptions.METASTORE.key(): "rest",
+        }))
+        original_loader = RESTCatalogLoader(context)
+        environment = CatalogEnvironment(
+            identifier=root,
+            catalog_loader=original_loader,
+        )
+        table = SimpleNamespace(catalog_environment=environment)
+        dependency_catalog = mock.MagicMock()
+        dependency_table = mock.sentinel.dependency_table
+        dependency_catalog.get_table.return_value = dependency_table
+
+        with mock.patch.object(
+                CatalogFactory,
+                "create_from_context",
+                return_value=dependency_catalog) as create_catalog:
+            result = BlobViewLookup(table)._load_table(target)
+
+        dependency_context = create_catalog.call_args.args[0]
+        create_catalog.assert_called_once_with(
+            dependency_context, config_required=False)
+        self.assertTrue(
+            dependency_context.options.contains_key(self._READ_VIA_OPTION))
+        dependency_catalog.get_table.assert_called_once_with(target)
+        self.assertIs(result, dependency_table)
+
+    def test_blob_view_lookup_preserves_custom_rest_catalog(self):
+        target = Identifier.create("db", "target")
+        dependency_table = mock.sentinel.dependency_table
+
+        class CustomRESTCatalog:
+            context = None
+
+            def __init__(self, context):
+                CustomRESTCatalog.context = context
+
+            def get_table(self, identifier):
+                self.identifier = identifier
+                return dependency_table
+
+        context = CatalogContext.create_from_options(Options({
+            CatalogOptions.METASTORE.key(): "custom-rest",
+        }))
+        environment = CatalogEnvironment(
+            identifier=Identifier.create("db", "root"),
+            catalog_loader=RESTCatalogLoader(context),
+        )
+        table = SimpleNamespace(catalog_environment=environment)
+
+        with mock.patch.dict(
+                CatalogFactory.CATALOG_REGISTRY,
+                {"custom-rest": CustomRESTCatalog}):
+            result = BlobViewLookup(table)._load_table(target)
+
+        dependency_context = CustomRESTCatalog.context
+        self.assertEqual(
+            dependency_context.options.get(CatalogOptions.METASTORE),
+            "custom-rest",
+        )
+        self.assertTrue(
+            dependency_context.options.contains_key(self._READ_VIA_OPTION))
+        self.assertIs(result, dependency_table)
+
+
+if __name__ == "__main__":
+    unittest.main()

paimon-python/pypaimon/utils/blob_view_lookup.py

@@ -290,7 +290,15 @@ def _target_rows_per_task(plans: List[TableReadPlan]) -> int:
         return max(_MIN_ROWS_PER_TASK, (total_rows + _PRELOAD_THREAD_NUM - 1) // _PRELOAD_THREAD_NUM)
 
     def _load_table(self, identifier: Identifier):
-        catalog = self._table.catalog_environment.catalog_loader.load()
+        catalog_environment = self._table.catalog_environment
+        catalog_loader = catalog_environment.catalog_loader
+        dependency_context = catalog_environment.dependency_read_context()
+        if dependency_context is catalog_environment.catalog_context():
+            catalog = catalog_loader.load()
+        else:
+            from pypaimon.catalog.catalog_factory import CatalogFactory
+            catalog = CatalogFactory.create_from_context(
+                dependency_context, config_required=False)
         return catalog.get_table(identifier)
 
     @staticmethod