[python] Use conditional OSS puts for atomic metadata writes

.github/workflows/paimon-python-checks.yml

@@ -173,6 +173,7 @@ jobs:
               python -m pip check
             fi
           fi
+          python -m pip install 'oss2>=2.18,<3'
           df -h
 
       - name: Run lint-python.sh

paimon-python/README.md

@@ -15,6 +15,80 @@ Pypaimon requires Python 3.6+.
 The core dependencies are listed in `dev/requirements.txt`.
 The development dependencies are listed in `dev/requirements-dev.txt`.
 
+# OSS metadata commits
+
+Install `pypaimon[oss]` (legacy PyArrow data access) or `pypaimon[jindo]`
+(Jindo data access). Both include `oss2` for atomic metadata writes.
+Configure `fs.oss.accessKeyId`, `fs.oss.accessKeySecret` and `fs.oss.endpoint`,
+plus `fs.oss.securityToken` when using STS. An endpoint without a scheme uses
+HTTPS for metadata writes. Credentials supplied only through an underlying
+filesystem's credential provider must also be supplied through these options.
+
+Metadata requests always use OSS Signature V4 for both AK and STS credentials,
+independently of the Jindo data-access signer setting. Set `fs.oss.region` to the
+bucket's region ID, such as `cn-hangzhou`. If unset, it is inferred from standard
+`oss-<region>.aliyuncs.com` or `oss-<region>-internal.aliyuncs.com` endpoints.
+Other endpoints, including acceleration endpoints, require an explicit region.
+
+Atomic metadata PUTs also forward the OSS server-side encryption options, using
+the same resolution as Java `OSSFileIO`:
+
+| Option | Behavior |
+| --- | --- |
+| `fs.oss.server-side-encryption` | `AES256`, `KMS` or `SM4` (case-insensitive) |
+| `fs.oss.server-side-encryption-key-id` | KMS key ID; implies `KMS` if the method is unset |
+| `fs.oss.server-side-data-encryption` | `SM4` with `KMS`; implies `KMS` if the method is unset |
+| `fs.oss.server-side-encryption-algorithm` | Legacy method fallback, used only if all three options above are unset |
+
+The first three options reject blank values and invalid combinations before any
+request is sent. If no encryption options are set, no encryption headers are
+added and OSS applies the bucket's default policy. These settings cover the
+atomic metadata PUT path; ordinary data writes still use the configured
+PyArrow/Jindo filesystem and its encryption capabilities.
+
+For `oss://` paths, `FileIO.get` selects `OssFileIO`, a thin `PyArrowFileIO`
+subclass that overrides atomic creation. Filesystem initialization, path handling,
+and ordinary PyArrow/Jindo file operations are inherited unchanged.
+Use `FileIO.get(path, options)` or construct `OssFileIO` explicitly for OSS atomic
+writes. REST token refresh and `ResolvingFileIO` also route atomic writes through
+this implementation.
+
+When the bucket is confirmed unversioned, `OssFileIO.try_to_write_atomic` uses a single OSS PUT
+with `x-oss-forbid-overwrite=true`. Exactly one writer can create a given object;
+`FileAlreadyExists` returns `False` so snapshot commits can retry. Other SDK errors
+are raised as `OSError`, retaining their cause for diagnostics. A lost PUT response
+is not retried by the SDK; the snapshot
+commit loop checks the commit user and identifier before retrying.
+
+The vendor SDK is an optional backend dependency, imported only for OSS atomic
+writes. The common FileIO API and Paimon table format remain independent of it.
+This implementation uses OSS-specific conditional creation; it does not provide
+the same atomic-write capability for every object store.
+
+Conditional creation requires a bucket that has **never enabled versioning**.
+The first atomic write on each `OssFileIO` instance checks `GetBucketVersioning`
+and caches the result, including the query-denied fallback. Concurrent first
+writes may repeat the check and warning. Query errors other than `403 AccessDenied`
+are not cached.
+If versioning is Enabled/Suspended,
+the state is unrecognized, or the query returns `403 AccessDenied`, the operation
+logs a warning when caching the fallback and uses the inherited PyArrow/Jindo
+temporary-file-and-rename path.
+This preserves legacy writes without making version-query permission mandatory,
+but the fallback does **not** guarantee safe concurrent commits. It also retains
+the existing backend's encryption behavior rather than applying the conditional
+PUT's OSS SSE headers. Invalid credentials, expired tokens, missing buckets, and
+other query failures still propagate as errors.
+
+Grant `oss:GetBucketVersioning` and keep versioning disabled to use conditional
+creation. Keep bucket versioning and version-query permissions unchanged for the
+instance's lifetime; recreate the FileIO after changing them. A configuration
+change is not guaranteed to produce an error and can invalidate the conditional-write guarantee.
+
+All concurrent writers must use conditional creation. Older Python clients or
+other clients that overwrite snapshot objects can still overwrite a successful
+commit. This change does not add conditional writes for other object stores.
+
 # Build
 
 You can build the source package by executing the following command:

paimon-python/dev/requirements-dev.txt

@@ -20,6 +20,7 @@
 # Test dependencies for pypaimon are as follows
 duckdb==1.3.2
 flake8==4.0.1
+oss2>=2.18,<3
 pytest~=7.0
 # HDF5 ingestion tests run in the supported h5py wheel lanes (Python 3.8+).
 h5py>=3,<4; python_version >= "3.8"

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

@@ -22,13 +22,12 @@
 
 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.filesystem.pyarrow_file_io import PyArrowFileIO
-from pypaimon.api.auth.bearer import BearTokenAuthProvider
-from pypaimon.api.auth.dlf_provider import DLFAuthProvider
 from pypaimon.common.identifier import Identifier
 from pypaimon.common.options import Options
 from pypaimon.common.options.config import CatalogOptions, OssOptions
@@ -40,27 +39,12 @@ class RESTTokenFileIO(FileIO):
     A FileIO to support getting token from REST Server.
     """
 
-    _FILE_IO_CACHE_MAXSIZE = 1000
     _FILE_IO_CACHE_TTL = 36000  # 10 hours in seconds
 
-    _FILE_IO_CACHE: TTLCache = None
-    _FILE_IO_CACHE_LOCK = threading.Lock()
-
     _TOKEN_CACHE: dict = {}
     _TOKEN_LOCKS: dict = {}
     _TOKEN_LOCKS_LOCK = threading.Lock()
 
-    @classmethod
-    def _get_file_io_cache(cls) -> TTLCache:
-        if cls._FILE_IO_CACHE is None:
-            with cls._FILE_IO_CACHE_LOCK:
-                if cls._FILE_IO_CACHE is None:
-                    cls._FILE_IO_CACHE = TTLCache(
-                        maxsize=cls._FILE_IO_CACHE_MAXSIZE,
-                        ttl=cls._FILE_IO_CACHE_TTL
-                    )
-        return cls._FILE_IO_CACHE
-
     def __init__(self, identifier: Identifier, path: str,
                  catalog_options: Optional[Union[dict, Options]] = None):
         self.identifier = identifier
@@ -77,12 +61,20 @@ def __init__(self, identifier: Identifier, path: str,
         self.api_instance: Optional[RESTApi] = None
         self.log = logging.getLogger(__name__)
         self._uri_reader_factory_cache: Optional[UriReaderFactory] = None
+        self._init_file_io_cache()
+
+    def _init_file_io_cache(self):
+        # FileIO is bound to this instance's path and catalog options.
+        self._file_io_cache = TTLCache(maxsize=1, ttl=self._FILE_IO_CACHE_TTL)
+        self._file_io_cache_lock = threading.Lock()
 
     def __getstate__(self):
         state = self.__dict__.copy()
         # Remove non-serializable objects
         state.pop('api_instance', None)
         state.pop('_uri_reader_factory_cache', None)
+        state.pop('_file_io_cache', None)
+        state.pop('_file_io_cache_lock', None)
         # token can be serialized, but we'll refresh it on deserialization
         return state
 
@@ -91,28 +83,17 @@ def __setstate__(self, state):
         self._uri_reader_factory_cache = None
         # api_instance will be recreated when needed
         self.api_instance = None
+        self._init_file_io_cache()
 
     def file_io(self) -> FileIO:
-        self.try_to_refresh_token()
-
-        if self.token is None:
-            return FileIO.get(self.path, self.catalog_options or Options({}))
-
-        cache_key = self.token
-        cache = self._get_file_io_cache()
-
-        file_io = cache.get(cache_key)
-        if file_io is not None:
-            return file_io
-
-        with self._FILE_IO_CACHE_LOCK:
+        with self._file_io_cache_lock:
             self.try_to_refresh_token()
 
             if self.token is None:
                 return FileIO.get(self.path, self.catalog_options or Options({}))
 
             cache_key = self.token
-            cache = self._get_file_io_cache()
+            cache = self._file_io_cache
             file_io = cache.get(cache_key)
             if file_io is not None:
                 return file_io
@@ -127,7 +108,7 @@ def file_io(self) -> FileIO:
                     merged_properties[OssOptions.OSS_ENDPOINT.key()] = dlf_oss_endpoint
             merged_options = Options(merged_properties)
 
-            file_io = PyArrowFileIO(self.path, merged_options)
+            file_io = FileIO.get(self.path, merged_options)
             cache[cache_key] = file_io
             return file_io
 

paimon-python/pypaimon/common/file_io.py

@@ -223,6 +223,7 @@ def read_ranges_coalesced_views(self, ranges, parallelism,
     def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span,
                                max_retained_amplification, return_views):
         from concurrent.futures import ThreadPoolExecutor
+
         # Threads write disjoint results[idx]; safe under the GIL (no list resize).
         results = [None] * len(ranges)
         coalescible, singletons = [], []
@@ -598,7 +599,8 @@ def get(path: str, catalog_options: Optional[Options] = None) -> 'FileIO':
         Returns a FileIO instance for accessing the file system identified by the given path.
         - LocalFileIO for local file system (file:// or no scheme)
         - HdfsNativeFileIO for HDFS/ViewFS (default; pure protocol client, no Hadoop install)
-        - PyArrowFileIO for other remote file systems (oss://, s3://, gs://, ...),
+        - OssFileIO for OSS (oss://)
+        - PyArrowFileIO for other remote file systems (s3://, gs://, ...),
           and for HDFS when explicitly requested via hdfs.client.impl=pyarrow
         """
         import os as _os
@@ -651,5 +653,9 @@ def get(path: str, catalog_options: Optional[Options] = None) -> 'FileIO':
                     f"(from {impl_source}). Supported: 'native', 'pyarrow'."
                 )
 
+        if scheme == "oss":
+            from pypaimon.filesystem.oss_file_io import OssFileIO
+            return OssFileIO(path, opts)
+
         from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO
         return PyArrowFileIO(path, opts)

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

@@ -31,6 +31,18 @@ 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_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")
+    OSS_SSE_KMS_KEY_ID = ConfigOptions.key(
+        "fs.oss.server-side-encryption-key-id").string_type().no_default_value().with_description(
+        "KMS key ID for OSS atomic metadata encryption")
+    OSS_SSE_DATA_ENCRYPTION = ConfigOptions.key(
+        "fs.oss.server-side-data-encryption").string_type().no_default_value().with_description(
+        "Data encryption algorithm for OSS atomic metadata encryption: SM4, with KMS only")
+    OSS_SSE_ALGORITHM = ConfigOptions.key(
+        "fs.oss.server-side-encryption-algorithm").string_type().no_default_value().with_description(
+        "Legacy OSS atomic metadata encryption method; used when the other SSE options are unset")
 
 
 class S3Options: