[python] Support per-request endpoint routing for io-cache

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

@@ -22,19 +22,24 @@
 import weakref
 from datetime import timedelta
 from typing import Optional, Union
+from urllib.parse import urlparse
 
 from cachetools import TTLCache
 
 from pypaimon.api.auth.bearer import BearTokenAuthProvider
 from pypaimon.api.auth.dlf_provider import DLFAuthProvider
 from pypaimon.api.rest_api import RESTApi
-from pypaimon.api.rest_util import RESTUtil
 from pypaimon.catalog.rest.rest_token import RESTToken
 from pypaimon.common.file_io import FileIO
 from pypaimon.common.identifier import Identifier
 from pypaimon.common.options import Options
-from pypaimon.common.options.config import CatalogOptions, OssOptions
 from pypaimon.common.uri_reader import UriReaderFactory
+from pypaimon.filesystem.io_cache_file_io import (IoCacheRoutingFileIO,
+                                                  table_view_spec)
+from pypaimon.filesystem.io_cache_routing import (IoCacheRouting,
+                                                  effective_options,
+                                                  schema_file_prefixes,
+                                                  with_catalog_overrides)
 
 
 class RESTTokenFileIO(FileIO):
@@ -65,6 +70,10 @@ class RESTTokenFileIO(FileIO):
                       "fds": None, "next_scan": 0.0}
     _SHORT_TOKEN_STATS = {"count": 0, "next_log": 0.0}
 
+    # The table view each FileIO is built for, see FileIO.origin_only and for_table.
+    _origin_only = False
+    _table: Optional[tuple] = None
+
     def __init__(self, identifier: Identifier, path: str,
                  catalog_options: Optional[Union[dict, Options]] = None):
         self.identifier = identifier
@@ -78,6 +87,8 @@ def __init__(self, identifier: Identifier, path: str,
             self.catalog_options = catalog_options
         self.properties = self.catalog_options or Options({})  # For compatibility with refresh_token()
         self.token: Optional[RESTToken] = None
+        # The token as the REST server vended it, without this catalog's overrides.
+        self._vended_token: Optional[RESTToken] = None
         self.api_instance: Optional[RESTApi] = None
         self.log = logging.getLogger(__name__)
         self._uri_reader_factory_cache: Optional[UriReaderFactory] = None
@@ -126,8 +137,42 @@ def _after_fork_in_child(cls):
         for file_io in list(cls._INSTANCES):
             file_io._token_lock = threading.Lock()
 
+    def origin_only(self) -> FileIO:
+        return self if self._origin_only else self._with_view(True, None)
+
+    def for_table(self, table_path: str, file_prefixes, branch: str) -> FileIO:
+        if self._origin_only:
+            return self
+        table = table_view_spec(self._table, table_path, file_prefixes, branch)
+        return self if table == self._table else self._with_view(False, table)
+
+    def _with_view(self, origin_only: bool, table) -> 'RESTTokenFileIO':
+        # Same token and caches; each FileIO it returns is the view of the cached one.
+        view = RESTTokenFileIO.__new__(RESTTokenFileIO)
+        view.__setstate__(self.__getstate__())
+        view.api_instance = self.api_instance
+        view._origin_only = origin_only
+        view._table = table
+        return view
+
+    def _table_view(self, file_io: FileIO) -> FileIO:
+        if self._origin_only:
+            return file_io.origin_only()
+        if self._table is None or not isinstance(file_io, IoCacheRoutingFileIO):
+            return file_io
+        return file_io.table_view(self._table, self._catalog_schema_prefixes)
+
+    def _catalog_schema_prefixes(self, branch: str):
+        # A REST catalog may keep a branch's schema without a schema directory in storage.
+        if self.api_instance is None:
+            self.api_instance = RESTApi(self.properties, False)
+        identifier = Identifier.create(self.identifier.get_database_name(), self.identifier.get_table_name(),
+                                       branch=None if branch == "main" else branch)
+        schema = self.api_instance.get_table(identifier).schema
+        return schema_file_prefixes(schema.options) if schema is not None else []
+
     def file_io(self) -> FileIO:
-        return self._file_io_with_token()[0]
+        return self._table_view(self._file_io_with_token()[0])
 
     def _file_io_with_token(self, minimum_validity_millis: int = 0):
         with self._token_lock:
@@ -151,7 +196,7 @@ def _file_io_with_token(self, minimum_validity_millis: int = 0):
                 cache = self._FILE_IO_CACHE
                 file_io = cache.get(cache_key)
                 if file_io is None:
-                    file_io = FileIO.get(self.path, self._merged_options(token))
+                    file_io = self._create_file_io(self._merged_options(token))
                     self._log_file_io_created(cache)
                 # Re-inserting on every access expires idle entries only, like expireAfterAccess.
                 cache[cache_key] = file_io
@@ -199,24 +244,32 @@ def _get_file_io_scope(self) -> str:
         return self._file_io_scope
 
     def _merged_options(self, token: RESTToken) -> Options:
-        merged_properties = RESTUtil.merge(
-            self.catalog_options.to_map() if self.catalog_options else {},
-            token.token
-        )
-        if self.catalog_options:
-            dlf_oss_endpoint = self.catalog_options.get(CatalogOptions.DLF_OSS_ENDPOINT)
-            if dlf_oss_endpoint and dlf_oss_endpoint.strip():
-                merged_properties[OssOptions.OSS_ENDPOINT.key()] = dlf_oss_endpoint
-        return Options(merged_properties)
+        return Options(effective_options(self._catalog_map(), token.token))
+
+    def _catalog_map(self) -> dict:
+        return self.catalog_options.to_map() if self.catalog_options else {}
+
+    def _create_file_io(self, options: Options) -> FileIO:
+        properties = options.to_map()
+        routing = IoCacheRouting.create(properties, root=self.path)
+        if routing is None or urlparse(self.path).scheme != "oss":
+            return FileIO.get(self.path, options)
+        self.log.info(
+            "io-cache routing enabled for [%s], targets %s, origin endpoint %s",
+            self.identifier, routing.targets(), routing.origin_endpoint())
+        origin = FileIO.get(self.path, Options(routing.origin_options(properties)))
+        return IoCacheRoutingFileIO(
+            routing, origin,
+            lambda name: FileIO.get(self.path, Options(routing.target_options(properties, name))))
 
     def _merge_token_with_catalog_options(self, token: dict) -> dict:
         """Merge token with catalog options, DLF OSS endpoint should override the standard OSS endpoint."""
-        merged_token = dict(token)
-        if self.catalog_options:
-            dlf_oss_endpoint = self.catalog_options.get(CatalogOptions.DLF_OSS_ENDPOINT)
-            if dlf_oss_endpoint and dlf_oss_endpoint.strip():
-                merged_token[OssOptions.OSS_ENDPOINT.key()] = dlf_oss_endpoint
-        return merged_token
+        return with_catalog_overrides(token, self._catalog_map())
+
+    def _adopt(self, vended: RESTToken):
+        # The shared token cache keeps the vended token; this instance applies its own catalog's overrides.
+        self._vended_token = vended
+        self.token = RESTToken(self._merge_token_with_catalog_options(vended.token), vended.expire_at_millis)
 
     def new_input_stream(self, path: str):
         return self.file_io().new_input_stream(path)
@@ -328,21 +381,21 @@ def try_to_refresh_token(self, minimum_validity_millis: int = 0):
 
         cached_token = self._get_cached_token(identifier_str)
         if cached_token and not self._is_token_expired(cached_token, minimum_validity_millis):
-            self.token = cached_token
+            self._adopt(cached_token)
             return
 
         global_lock = self._get_global_token_lock(identifier_str)
 
         with global_lock:
             cached_token = self._get_cached_token(identifier_str)
             if cached_token and not self._is_token_expired(cached_token, minimum_validity_millis):
-                self.token = cached_token
+                self._adopt(cached_token)
                 return
 
             token_to_check = cached_token if cached_token else self.token
             if token_to_check is None or self._is_token_expired(token_to_check, minimum_validity_millis):
                 self.refresh_token()
-                self._set_cached_token(identifier_str, self.token)
+                self._set_cached_token(identifier_str, self._vended_token)
 
     def _get_cached_token(self, identifier_str: str) -> Optional[RESTToken]:
         with self._TOKEN_LOCKS_LOCK:
@@ -413,11 +466,9 @@ def refresh_token(self):
             f"end refresh data token for identifier [{self.identifier}] expiresAtMillis [{response.expires_at_millis}]"
         )
 
-        merged_token_dict = self._merge_token_with_catalog_options(response.token)
-        new_token = RESTToken(merged_token_dict, response.expires_at_millis)
-        self.token = new_token
-        if self._is_token_expired(new_token):
-            self._log_short_token(new_token)
+        self._adopt(RESTToken(response.token, response.expires_at_millis))
+        if self._is_token_expired(self.token):
+            self._log_short_token(self.token)
 
     def _log_short_token(self, token: RESTToken):
         with self._TOKEN_LOCKS_LOCK:

paimon-python/pypaimon/common/file_io.py

@@ -92,6 +92,15 @@ class FileIO(ABC):
     File IO interface to read and write files.
     """
 
+    def origin_only(self) -> 'FileIO':
+        """A view that sends no request to an io-cache target, for files other writers may replace."""
+        return self
+
+    def for_table(self, table_path: str, file_prefixes: List[str], branch: str) -> 'FileIO':
+        """A view for one Paimon table: io-cache targets serve only its files under table_path, named
+        with these prefixes or those of the branch's schemas."""
+        return self
+
     @abstractmethod
     def new_input_stream(self, path: str):
         pass
@@ -100,6 +109,10 @@ def new_input_stream(self, path: str):
     def new_output_stream(self, path: str):
         pass
 
+    def create_parent_dirs(self, path: str):
+        """Runs only the parent directory step of new_output_stream(path); a no-op unless a FileIO splits it out."""
+        pass
+
     @abstractmethod
     def get_file_status(self, path: str):
         pass
@@ -683,7 +696,12 @@ def get(path: str, catalog_options: Optional[Options] = None) -> 'FileIO':
                 )
 
         if scheme == "oss":
+            from pypaimon.filesystem.io_cache_routing import IoCacheRouting
             from pypaimon.filesystem.oss_file_io import OssFileIO
+            # This FileIO sends every request to one endpoint, so under io-cache routing it is the origin's.
+            routing = IoCacheRouting.create(opts.to_map())
+            if routing is not None:
+                opts = Options(routing.origin_options(opts.to_map()))
             return OssFileIO(path, opts)
 
         from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO

paimon-python/pypaimon/common/options/config.py

@@ -33,6 +33,10 @@ class OssOptions:
     OSS_ENDPOINT = ConfigOptions.key("fs.oss.endpoint").string_type().no_default_value().with_description(
         "OSS endpoint")
     OSS_REGION = ConfigOptions.key("fs.oss.region").string_type().no_default_value().with_description("OSS region")
+    OSS_SECOND_LEVEL_DOMAIN_ENABLE = ConfigOptions.key(
+        "fs.oss.second.level.domain.enable").boolean_type().default_value(False).with_description(
+        "Path-style addressing (endpoint/bucket/key) instead of bucket.endpoint, "
+        "for IP or non-wildcard endpoints")
     OSS_SSE_METHOD = ConfigOptions.key(
         "fs.oss.server-side-encryption").string_type().no_default_value().with_description(
         "OSS atomic metadata encryption method: AES256, KMS or SM4")
@@ -129,6 +133,9 @@ class CatalogOptions:
         "DLF token file path")
     DLF_OSS_ENDPOINT = ConfigOptions.key("dlf.oss-endpoint").string_type().no_default_value().with_description(
         "DLF OSS endpoint")
+    IO_CACHE_ENABLED = ConfigOptions.key("io-cache.enabled").boolean_type().default_value(False).with_description(
+        "Route reads to the io-cache targets vended with the table token; "
+        "a value set in the catalog options overrides the token")
     DLF_TOKEN_LOADER = ConfigOptions.key("dlf.token-loader").string_type().no_default_value().with_description(
         "DLF token loader")
     DLF_TOKEN_ECS_ROLE_NAME = ConfigOptions.key(

paimon-python/pypaimon/daft/daft_paimon.py

@@ -33,6 +33,10 @@
 from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 from urllib.parse import urlparse
 
+from pypaimon.filesystem.io_cache_routing import (IoCacheRouting,
+                                                  effective_options,
+                                                  single_endpoint_options)
+
 if TYPE_CHECKING:
     import daft
 
@@ -46,20 +50,25 @@ def _enrich_options_with_rest_token(
     # REST catalogs (DLF) keep OSS STS tokens on table.file_io and the bucket on table.table_path,
     # not in catalog_options; fold both in so Daft's IOConfig routes to OSS with valid credentials.
     if catalog_options.get("metastore") != "rest":
-        return catalog_options
+        return _origin_only(catalog_options)
     file_io = getattr(table, "file_io", None)
     if file_io is None or not hasattr(file_io, "try_to_refresh_token"):
-        return catalog_options
+        return _origin_only(catalog_options)
     file_io.try_to_refresh_token()
     if file_io.token is None:
-        return catalog_options
-    enriched = {**catalog_options, **file_io.token.token}
+        return _origin_only(catalog_options)
+    enriched = single_endpoint_options(effective_options(catalog_options, file_io.token.token))
     parsed = urlparse(getattr(table, "table_path", "") or "")
     if parsed.scheme and parsed.netloc:
         enriched["warehouse"] = f"{parsed.scheme}://{parsed.netloc}"
     return enriched
 
 
+def _origin_only(options: Dict[str, str]) -> Dict[str, str]:
+    # Daft reads with its own client, which cannot use io-cache targets per request.
+    return options if IoCacheRouting.create(options) is None else single_endpoint_options(options)
+
+
 # Time-travel targets are mutually exclusive: at most one may be set.
 TimeTravelTimestamp = Union[int, str, "datetime.datetime"]
 

paimon-python/pypaimon/execution/native/table.py

@@ -25,6 +25,7 @@
 
 from pypaimon.common.options.config import CatalogOptions, FuseOptions, OssOptions
 from pypaimon.common.options.options_utils import OptionsUtils
+from pypaimon.filesystem.io_cache_routing import single_endpoint_options
 
 
 def _native_blob_view_supported(table, read_names) -> bool:
@@ -199,11 +200,18 @@ def get(self, response, database, table, options, token_loader=None):
         def create():
             token = token_loader() if token_loader else None
             if token is None:
+                # A REST table can reach this path before its data token is available (for
+                # example a copied branch or a token refresh race).  The native core sends
+                # every OSS request to one endpoint, so routing keys in the catalog must not
+                # leave its legacy fs.oss.endpoint pointed at a cache target.
                 return Table.from_rest_response(
-                    response, database=database, table=table, rest_options=options)
+                    response, database=database, table=table,
+                    rest_options=single_endpoint_options(options))
+            # The native core cannot use io-cache targets per request, and fs.oss.endpoint may name the cache.
             return Table.from_rest_response_with_token(
                 response, database=database, table=table, rest_options=options,
-                data_token=dict(token.token), expires_at_millis=token.expire_at_millis)
+                data_token=single_endpoint_options(token.token, options),
+                expires_at_millis=token.expire_at_millis)
 
         key = (response, database, table, tuple(sorted(options.items())))
         return self.get_or_create(key, create)
@@ -304,16 +312,21 @@ def _load_native_table(table):
             token_loader=lambda: _rest_data_token(table))
     elif file_io_options is not None:
         from pypaimon_rust.datafusion import Table
+        # As with a token: the core cannot use io-cache targets per request, so it gets the origin's options.
         rt = Table.from_resolved_schema(
             table.table_path, _resolved_schema_json(table),
             database=table.identifier.get_database_name(),
             table=table.identifier.get_table_name(),
-            branch=table.current_branch(), options=file_io_options)
+            branch=table.current_branch(), options=single_endpoint_options(file_io_options))
     else:
         from pypaimon_rust.datafusion import PaimonCatalog
         # The Python loader has already resolved /config, including typed
         # FUSE options. Match RESTCatalogLoader.load() and reuse that context.
-        catalog = PaimonCatalog(_catalog_options(table), config_required=False)
+        # This catalog is also a single-endpoint client.  Keep its metadata/catalog
+        # configuration, but make its storage endpoint the origin when catalog-only
+        # io-cache routing is enabled.
+        catalog = PaimonCatalog(
+            single_endpoint_options(_catalog_options(table)), config_required=False)
         # REST may keep branch schemas in the catalog only. Load the base
         # environment, then attach the schema/branch already resolved here.
         rt = catalog.get_table((

paimon-python/pypaimon/filesystem/caching_file_io.py

@@ -414,6 +414,16 @@ def __getstate__(self):
         # Python 3.10 otherwise finds the delegate's hook through __getattr__.
         return self.__dict__.copy()
 
+    def origin_only(self) -> FileIO:
+        return self._with_delegate(self._delegate.origin_only())
+
+    def for_table(self, table_path: str, file_prefixes, branch: str) -> FileIO:
+        return self._with_delegate(self._delegate.for_table(table_path, file_prefixes, branch))
+
+    def _with_delegate(self, view: FileIO) -> FileIO:
+        # A view shares this wrapper's cache.
+        return self if view is self._delegate else CachingFileIO(view, self._cache, self._whitelist)
+
     # Fallback caps when local-cache.max-size is unset (memory shares the heap).
     _DEFAULT_MEMORY_CACHE_MAX_SIZE = 256 * 1024 * 1024
     _DEFAULT_DISK_CACHE_MAX_SIZE = 10 * 1024 * 1024 * 1024