From 6d804d947f7b4aa27680018d6a70f9a7ca5c61d6 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Thu, 10 Sep 2026 13:33:50 +0800 Subject: [PATCH 1/7] [python] Use conditional OSS puts for atomic metadata writes --- paimon-python/README.md | 61 ++++ .../catalog/rest/rest_token_file_io.py | 7 +- paimon-python/pypaimon/common/file_io.py | 8 +- .../pypaimon/common/options/config.py | 12 + .../pypaimon/filesystem/oss_file_io.py | 124 ++++++++ .../pypaimon/filesystem/resolving_file_io.py | 3 + .../rest_catalog_blob_as_descriptor_sample.py | 23 +- paimon-python/pypaimon/tests/file_io_test.py | 29 +- .../pypaimon/tests/lance_utils_test.py | 6 +- .../pypaimon/tests/oss_atomic_write_test.py | 290 ++++++++++++++++++ .../pypaimon/tests/oss_file_io_test.py | 80 ++--- .../pypaimon/tests/oss_legacy_mode_test.py | 19 +- .../pypaimon/tests/py36/ao_simple_test.py | 8 +- paimon-python/setup.py | 3 + 14 files changed, 584 insertions(+), 89 deletions(-) create mode 100644 paimon-python/pypaimon/filesystem/oss_file_io.py create mode 100644 paimon-python/pypaimon/tests/oss_atomic_write_test.py diff --git a/paimon-python/README.md b/paimon-python/README.md index f864ff265bf0..6abc1cf056b3 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -15,6 +15,67 @@ 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. + +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**. +Each call first checks `GetBucketVersioning`. If versioning is Enabled/Suspended, +the state is unrecognized, or the query returns `403 AccessDenied`, the operation +logs a warning 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. The check and PUT cannot be made atomic with a bucket configuration change. + +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: diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index 42dabb268cb2..633cc7ae8242 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py +++ b/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 @@ -127,7 +126,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 diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 10406849b341..720f9ae5101f 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/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) diff --git a/paimon-python/pypaimon/common/options/config.py b/paimon-python/pypaimon/common/options/config.py index 5684ab62c3c6..98a466496b03 100644 --- a/paimon-python/pypaimon/common/options/config.py +++ b/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: diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py b/paimon-python/pypaimon/filesystem/oss_file_io.py new file mode 100644 index 000000000000..a31097ee3121 --- /dev/null +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -0,0 +1,124 @@ +# 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. + +"""OSS conditional metadata creation, reusing the existing Arrow/Jindo FileIO.""" + +import re +from urllib.parse import urlparse + +from pypaimon.common.options.config import OssOptions +from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO + + +class OssFileIO(PyArrowFileIO): + """Override atomic metadata creation; inherit all ordinary file operations.""" + + def try_to_write_atomic(self, path: str, content: str) -> bool: + uri = urlparse(path) + if uri.scheme: + if uri.scheme != 'oss' or self._extract_oss_bucket(path) != self._oss_bucket: + raise ValueError("Atomic write must target the configured OSS bucket") + key = re.sub(r'/+', '/', uri.path).lstrip('/') + else: + key = path + if not self._use_jindo and not self._oss_bucket_in_endpoint: + bucket, _, key = path.partition('/') + if bucket != self._oss_bucket: + raise ValueError("Atomic write must target the configured OSS bucket") + if not key or key == '.' or key.endswith('/'): + return False + + try: + import oss2 + except ImportError as error: + raise ImportError( + "OSS atomic writes require oss2. Install pypaimon[oss] or pypaimon[jindo]." + ) from error + + access_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_ID) + secret_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_SECRET) + token = self.properties.get(OssOptions.OSS_SECURITY_TOKEN) + endpoint = self.properties.get(OssOptions.OSS_ENDPOINT) + if not access_key or not secret_key or not endpoint: + raise ValueError( + "OSS atomic writes require fs.oss.accessKeyId, fs.oss.accessKeySecret " + "and fs.oss.endpoint; pass fs.oss.securityToken for STS credentials.") + if '://' not in endpoint: + endpoint = 'https://' + endpoint + headers = self._sse_headers() + headers['x-oss-forbid-overwrite'] = 'true' + auth = oss2.StsAuth(access_key, secret_key, token) if token else oss2.Auth(access_key, secret_key) + + session = oss2.Session() + try: + bucket = oss2.Bucket(auth, endpoint, self._oss_bucket, session=session, + region=self.properties.get(OssOptions.OSS_REGION)) + try: + versioning = bucket.get_bucket_versioning().status + except oss2.exceptions.ServerError as error: + if error.status != 403 or error.code != 'AccessDenied': + raise + versioning = 'unknown (GetBucketVersioning denied)' + if versioning is not None: + self.logger.warning( + "Using legacy temporary-file-and-rename writes for OSS bucket %s " + "(versioning: %s). Concurrent commits are not protected against overwrites.", + self._oss_bucket, versioning) + return super().try_to_write_atomic(path, content) + try: + bucket.put_object(key, content.encode('utf-8'), headers=headers) + return True + except oss2.exceptions.ServerError as error: + if error.code == 'FileAlreadyExists': + return False + raise + except oss2.exceptions.OssError as error: + raise OSError("Failed to atomically write oss://{}/{}".format(self._oss_bucket, key)) from error + finally: + session.session.close() + + def _sse_headers(self): + """Match Java OSSFileIO's SSE resolution, including the native option fallback.""" + settings = (OssOptions.OSS_SSE_METHOD, OssOptions.OSS_SSE_KMS_KEY_ID, + OssOptions.OSS_SSE_DATA_ENCRYPTION) + values = [self.properties.get(setting) for setting in settings] + for setting, value in zip(settings, values): + if value is not None and not value.strip(): + raise ValueError("'{}' is set but blank".format(setting.key())) + method, key_id, data_encryption = [value.strip() if value is not None else None for value in values] + if method is None and key_id is None and data_encryption is None: + # Java forwards a nonempty native value as-is; the OSS service validates it. + algorithm = self.properties.get(OssOptions.OSS_SSE_ALGORITHM) + return {'x-oss-server-side-encryption': algorithm} if algorithm else {} + + method = (method or 'KMS').upper() + if method not in ('AES256', 'KMS', 'SM4'): + raise ValueError("'{}' must be one of AES256/KMS/SM4".format(OssOptions.OSS_SSE_METHOD.key())) + headers = {'x-oss-server-side-encryption': method} + if key_id is not None: + if any(character.isspace() for character in key_id): + raise ValueError("'{}' must not contain whitespace".format(OssOptions.OSS_SSE_KMS_KEY_ID.key())) + if method != 'KMS': + raise ValueError("'{}' requires KMS".format(OssOptions.OSS_SSE_KMS_KEY_ID.key())) + headers['x-oss-server-side-encryption-key-id'] = key_id + if data_encryption is not None: + if method != 'KMS': + raise ValueError("'{}' requires KMS".format(OssOptions.OSS_SSE_DATA_ENCRYPTION.key())) + if data_encryption.upper() != 'SM4': + raise ValueError("'{}' only supports SM4".format(OssOptions.OSS_SSE_DATA_ENCRYPTION.key())) + headers['x-oss-server-side-data-encryption'] = 'SM4' + return headers diff --git a/paimon-python/pypaimon/filesystem/resolving_file_io.py b/paimon-python/pypaimon/filesystem/resolving_file_io.py index 576d160c8155..e8f1c79d87d2 100644 --- a/paimon-python/pypaimon/filesystem/resolving_file_io.py +++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py @@ -97,6 +97,9 @@ def mkdirs(self, path: str) -> bool: def rename(self, src: str, dst: str) -> bool: return self._get_fileio(src).rename(src, dst) + def try_to_write_atomic(self, path: str, content: str) -> bool: + return self._get_fileio(path).try_to_write_atomic(path, content) + def get_file_size(self, path: str) -> int: return self._get_fileio(path).get_file_size(path) diff --git a/paimon-python/pypaimon/sample/rest_catalog_blob_as_descriptor_sample.py b/paimon-python/pypaimon/sample/rest_catalog_blob_as_descriptor_sample.py index 639b139d9f57..1e90fa1bdfe5 100644 --- a/paimon-python/pypaimon/sample/rest_catalog_blob_as_descriptor_sample.py +++ b/paimon-python/pypaimon/sample/rest_catalog_blob_as_descriptor_sample.py @@ -18,13 +18,12 @@ """ Sample demonstrating descriptor-stored blob fields with REST catalog. """ -from pypaimon import CatalogFactory import pyarrow as pa -from pypaimon import Schema -from pypaimon.table.row.blob import BlobDescriptor, Blob +from pypaimon import CatalogFactory, Schema +from pypaimon.common.file_io import FileIO from pypaimon.common.options import Options -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO +from pypaimon.table.row.blob import Blob, BlobDescriptor def write_table_with_blob(catalog, video_file_path: str, external_oss_options: dict): @@ -66,7 +65,7 @@ def write_table_with_blob(catalog, video_file_path: str, external_oss_options: d # Access external OSS file to get file size try: - external_file_io = PyArrowFileIO(video_file_path, Options(external_oss_options)) + external_file_io = FileIO.get(video_file_path, Options(external_oss_options)) video_file_size = external_file_io.get_file_size(video_file_path) except Exception as e: raise FileNotFoundError( @@ -92,7 +91,7 @@ def write_table_with_blob(catalog, video_file_path: str, external_oss_options: d print("✓ Data committed successfully") table_write.close() table_commit.close() - + return f'{database_name}.{table_name}' @@ -103,10 +102,10 @@ def read_table_with_blob(catalog, table_name: str): table_scan = read_builder.new_scan() splits = table_scan.plan().splits() table_read = read_builder.new_read() - + result = table_read.to_arrow(splits) print(f"✓ Read {result.num_rows} rows") - + video_bytes_list = result.column('video').to_pylist() for video_bytes in video_bytes_list: if video_bytes is None: @@ -118,7 +117,7 @@ def read_table_with_blob(catalog, table_name: str): blob_data = blob.to_data() print(f"✓ Blob data verified: {len(blob_data) / 1024 / 1024:.2f} MB") break - + return result @@ -129,9 +128,9 @@ def read_table_with_blob(catalog, table_name: str): 'fs.oss.endpoint': "oss-cn-hangzhou.aliyuncs.com", 'fs.oss.region': "cn-hangzhou", } - + video_file_path = "oss://your-bucket/blob_test/video.mov" - + catalog_options = { 'metastore': 'rest', 'uri': "http://your-rest-catalog-uri", @@ -145,7 +144,7 @@ def read_table_with_blob(catalog, table_name: str): } catalog = CatalogFactory.create(catalog_options) - + try: table_name = write_table_with_blob(catalog, video_file_path, external_oss_options) result = read_table_with_blob(catalog, table_name) diff --git a/paimon-python/pypaimon/tests/file_io_test.py b/paimon-python/pypaimon/tests/file_io_test.py index 9803a5379018..8f8d722bc807 100644 --- a/paimon-python/pypaimon/tests/file_io_test.py +++ b/paimon-python/pypaimon/tests/file_io_test.py @@ -29,6 +29,7 @@ from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions from pypaimon.filesystem.local_file_io import LocalFileIO, _file_uri_path +from pypaimon.filesystem.oss_file_io import OssFileIO from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO, _pyarrow_lt_7 @@ -95,7 +96,7 @@ def test_filesystem_path_conversion(self): parent_str = str(Path(converted_path).parent) self.assertEqual(file_io.to_filesystem_path(parent_str), parent_str) - oss_io = PyArrowFileIO("oss://test-bucket/warehouse", Options({ + oss_io = OssFileIO("oss://test-bucket/warehouse", Options({ OssOptions.OSS_ENDPOINT.key(): 'oss-cn-hangzhou.aliyuncs.com', OssOptions.OSS_ACCESS_KEY_ID.key(): 'test-key', OssOptions.OSS_ACCESS_KEY_SECRET.key(): 'test-secret', @@ -348,7 +349,7 @@ def test_exists_does_not_catch_exception(self): file_io.delete_quietly("file:///some/path") file_io.delete_directory_quietly("file:///some/path") - oss_io = PyArrowFileIO("oss://test-bucket/warehouse", Options({ + oss_io = OssFileIO("oss://test-bucket/warehouse", Options({ OssOptions.OSS_ENDPOINT.key(): 'oss-cn-hangzhou.aliyuncs.com', OssOptions.OSS_ACCESS_KEY_ID.key(): 'test-key', OssOptions.OSS_ACCESS_KEY_SECRET.key(): 'test-secret', @@ -441,7 +442,7 @@ def test_get_file_status_raises_error_when_file_not_exists(self): test_file = os.path.join(temp_dir, "test_file.txt") with open(test_file, "w") as f: f.write("test content") - + file_info = file_io.get_file_status(f"file://{test_file}") self.assertEqual(file_info.type, pafs.FileType.File) self.assertIsNotNone(file_info.size) @@ -464,26 +465,26 @@ def test_copy_file(self): source_file = os.path.join(temp_dir, "source.txt") target_file = os.path.join(temp_dir, "target.txt") - + with open(source_file, "w") as f: f.write("source content") - + # Test 1: Raises FileExistsError when target exists and overwrite=False with open(target_file, "w") as f: f.write("target content") - + with self.assertRaises(FileExistsError) as context: file_io.copy_file(f"file://{source_file}", f"file://{target_file}", overwrite=False) self.assertIn("already exists", str(context.exception)) - + with open(target_file, "r") as f: self.assertEqual(f.read(), "target content") - + # Test 2: Overwrites when overwrite=True file_io.copy_file(f"file://{source_file}", f"file://{target_file}", overwrite=True) with open(target_file, "r") as f: self.assertEqual(f.read(), "source content") - + # Test 3: Creates parent directory if it doesn't exist target_file_in_subdir = os.path.join(temp_dir, "subdir", "target.txt") file_io.copy_file(f"file://{source_file}", f"file://{target_file_in_subdir}", overwrite=False) @@ -498,7 +499,7 @@ def test_try_to_write_atomic(self): try: target_dir = os.path.join(temp_dir, "target_dir") normal_file = os.path.join(temp_dir, "normal_file.txt") - + from pypaimon.filesystem.local_file_io import LocalFileIO local_file_io = LocalFileIO(f"file://{temp_dir}", Options({})) os.makedirs(target_dir) @@ -506,18 +507,18 @@ def test_try_to_write_atomic(self): local_file_io.try_to_write_atomic(f"file://{target_dir}", "test content"), "LocalFileIO should return False when target is a directory") self.assertEqual(len(os.listdir(target_dir)), 0, "No file should be created inside the directory") - + self.assertTrue(local_file_io.try_to_write_atomic(f"file://{normal_file}", "test content")) with open(normal_file, "r") as f: self.assertEqual(f.read(), "test content") - + os.remove(normal_file) local_file_io = LocalFileIO(f"file://{temp_dir}", Options({})) self.assertFalse( local_file_io.try_to_write_atomic(f"file://{target_dir}", "test content"), "LocalFileIO should return False when target is a directory") self.assertEqual(len(os.listdir(target_dir)), 0, "No file should be created inside the directory") - + self.assertTrue(local_file_io.try_to_write_atomic(f"file://{normal_file}", "test content")) with open(normal_file, "r") as f: self.assertEqual(f.read(), "test content") @@ -525,7 +526,7 @@ def test_try_to_write_atomic(self): shutil.rmtree(temp_dir, ignore_errors=True) def test_path_on_windows(self): - oss_io = PyArrowFileIO("oss://test-bucket/warehouse", Options({ + oss_io = OssFileIO("oss://test-bucket/warehouse", Options({ OssOptions.OSS_ENDPOINT.key(): 'oss-cn-hangzhou.aliyuncs.com', OssOptions.OSS_ACCESS_KEY_ID.key(): 'test-key', OssOptions.OSS_ACCESS_KEY_SECRET.key(): 'test-secret', diff --git a/paimon-python/pypaimon/tests/lance_utils_test.py b/paimon-python/pypaimon/tests/lance_utils_test.py index 9eb12566b52b..2ffb220efa46 100644 --- a/paimon-python/pypaimon/tests/lance_utils_test.py +++ b/paimon-python/pypaimon/tests/lance_utils_test.py @@ -17,9 +17,9 @@ import unittest +from pypaimon.common.file_io import FileIO from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO from pypaimon.read.reader.lance_utils import to_lance_specified @@ -34,7 +34,7 @@ def test_oss_url_bucket_extraction_correctness(self): OssOptions.OSS_ACCESS_KEY_SECRET.key(): "test-secret", }) - file_io = PyArrowFileIO(file_path, properties) + file_io = FileIO.get(file_path, properties) file_path_for_lance, storage_options = to_lance_specified(file_io, file_path) self.assertEqual( @@ -60,7 +60,7 @@ def test_oss_url_with_security_token(self): OssOptions.OSS_SECURITY_TOKEN.key(): "test-token", }) - file_io = PyArrowFileIO(file_path, properties) + file_io = FileIO.get(file_path, properties) file_path_for_lance, storage_options = to_lance_specified(file_io, file_path) self.assertEqual(file_path_for_lance, "oss://my-bucket/path/to/file.lance") diff --git a/paimon-python/pypaimon/tests/oss_atomic_write_test.py b/paimon-python/pypaimon/tests/oss_atomic_write_test.py new file mode 100644 index 000000000000..3ca19a91b148 --- /dev/null +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -0,0 +1,290 @@ +# 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. + +"""Exercise conditional OSS writes through the real SDK against a local HTTP server.""" + +import socket +import threading +from concurrent.futures import ThreadPoolExecutor +from http.server import BaseHTTPRequestHandler, HTTPServer +from socketserver import ThreadingMixIn +from unittest import mock +from urllib.parse import unquote, urlsplit + +import pyarrow.fs as pafs +import pytest + +from pypaimon.catalog.rest.rest_token import RESTToken +from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO +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 +from pypaimon.filesystem.oss_file_io import OssFileIO + +oss2 = pytest.importorskip("oss2") + + +@pytest.fixture +def oss_server(): + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def respond(self, status, body): + self.send_response(status) + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + def error(self, status, code): + self.respond(status, ('' + code + '').encode()) + + def do_GET(self): + server.gets += 1 + assert urlsplit(self.path).query in ('versioning', 'versioning=') + if server.fail_method == 'GET': + self.error(*server.failure) + return + status = '' if server.versioning is None else '' + server.versioning + '' + self.respond(200, ('' + status + + '').encode()) + + def do_PUT(self): + key = unquote(urlsplit(self.path).path) + data = self.rfile.read(int(self.headers['Content-Length'])) + with server.lock: + server.puts += 1 + server.token = self.headers.get('x-oss-security-token') + server.sse_headers = {key.lower()[len('x-oss-'):]: value + for key, value in self.headers.items() + if key.lower().startswith('x-oss-server-side-')} + if server.fail_method == 'PUT': + self.error(*server.failure) + return + if key in server.objects and self.headers.get('x-oss-forbid-overwrite') == 'true': + self.error(409, 'FileAlreadyExists') + return + server.objects[key] = data + if server.lose_response: + self.connection.shutdown(socket.SHUT_RDWR) + self.connection.close() + return + self.respond(200, b'') + + class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): + daemon_threads = True + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + server.objects = {} + server.lock = threading.Lock() + server.versioning = None + server.fail_method = None + server.failure = (403, 'AccessDenied') + server.lose_response = False + server.puts = 0 + server.gets = 0 + server.token = None + server.sse_headers = {} + thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.01), daemon=True) + thread.start() + yield server + server.shutdown() + server.server_close() + thread.join() + + +def options_for(server, token=None): + return Options({ + 'fs.oss.impl': 'legacy', + 'fs.oss.endpoint': 'http://127.0.0.1:{}'.format(server.server_port), + 'fs.oss.accessKeyId': 'test-ak', + 'fs.oss.accessKeySecret': 'test-sk', + 'fs.oss.securityToken': token, + }) + + +def file_io(server, resolving=False): + options = options_for(server) + if resolving: + options = Options(dict(options.to_map(), **{ + CatalogOptions.RESOLVING_FILE_IO_ENABLED.key(): 'true'})) + with mock.patch.object(OssFileIO, '_initialize_oss_fs'): + return FileIO.get('oss://test-bucket/', options) + + +@pytest.mark.parametrize('resolving', [False, True]) +def test_atomic_competition_and_existing_content(oss_server, resolving): + io = file_io(oss_server, resolving) + path = 'oss://test-bucket/table/snapshot-1' + barrier = threading.Barrier(2) + + def write(content): + barrier.wait(timeout=10) + return io.try_to_write_atomic(path, content) + + contents = ['提交者一', '提交者二'] + with mock.patch.object(OssFileIO, '_initialize_oss_fs'), ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(write, contents)) + assert sorted(results) == [False, True] + assert list(oss_server.objects.values()) == [contents[results.index(True)].encode()] + assert io.try_to_write_atomic(path, 'overwrite') is False + assert list(oss_server.objects.values()) == [contents[results.index(True)].encode()] + + +def test_rest_token_refresh_keeps_oss_atomic_creation(oss_server): + options = dict(options_for(oss_server).to_map()) + options[CatalogOptions.RESOLVING_FILE_IO_ENABLED.key()] = 'true' + path = 'oss://test-bucket/table/snapshot-1' + with mock.patch.object(RESTTokenFileIO, 'try_to_refresh_token'), \ + mock.patch.object(RESTTokenFileIO, '_FILE_IO_CACHE', {}), \ + mock.patch.object(OssFileIO, '_initialize_oss_fs'): + io = RESTTokenFileIO(Identifier.from_string('default.table'), path, Options(options)) + io.token = RESTToken({'fs.oss.securityToken': 'first-token'}, 1) + assert io.try_to_write_atomic(path, 'first') + assert oss_server.token == 'first-token' + io.token = RESTToken({'fs.oss.securityToken': 'refreshed-token'}, 2) + assert io.try_to_write_atomic(path, 'overwrite') is False + assert io.try_to_write_atomic(path.replace('snapshot-1', 'snapshot-2'), 'second') + assert oss_server.token == 'refreshed-token' + assert sorted(oss_server.objects.values()) == [b'first', b'second'] + + +@pytest.mark.parametrize('versioning', ['Enabled', 'Suspended', 'Unexpected', None]) +def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, tmp_path, caplog): + oss_server.versioning = versioning + if versioning is None: + oss_server.fail_method = 'GET' + oss_server.failure = (403, 'AccessDenied') + io = file_io(oss_server) + # Run the inherited stream/rename operations against an actual Arrow filesystem. + io.filesystem = pafs.LocalFileSystem() + path = 'oss://test-bucket{}/snapshot-1'.format(tmp_path) + assert io.try_to_write_atomic(path, '兼容写入') is True + assert io.try_to_write_atomic(path, 'overwrite') is False + assert (tmp_path / 'snapshot-1').read_text() == '兼容写入' + assert sorted(p.name for p in tmp_path.iterdir()) == ['snapshot-1'] + assert oss_server.puts == 0 + assert 'Concurrent commits are not protected' in caplog.text + + +@pytest.mark.parametrize('method,status,code', [ + ('GET', 403, 'InvalidAccessKeyId'), + ('GET', 403, 'SecurityTokenExpired'), + ('GET', 404, 'NoSuchBucket'), + ('GET', 500, 'InternalError'), + ('PUT', 403, 'AccessDenied'), + ('PUT', 500, 'InternalError'), + ('PUT', 409, 'OtherConflict'), +]) +def test_errors_are_not_competition(oss_server, method, status, code): + oss_server.fail_method, oss_server.failure = method, (status, code) + with pytest.raises(OSError) as caught: + file_io(oss_server).try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + assert isinstance(caught.value.__cause__, oss2.exceptions.ServerError) + assert caught.value.__cause__.code == code + assert oss_server.objects == {} + + +def test_lost_response_is_not_replayed_or_reported_as_conflict(oss_server): + oss_server.lose_response = True + with pytest.raises(OSError) as caught: + file_io(oss_server).try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + assert isinstance(caught.value.__cause__, oss2.exceptions.RequestError) + assert oss_server.puts == 1 + assert list(oss_server.objects.values()) == [b'data'] + + +@pytest.mark.parametrize('jindo,legacy,path', [ + (False, False, 'oss://test-bucket/table/p=a%2Fb/snapshot-1'), + (False, False, 'test-bucket/table/p=a%2Fb/snapshot-1'), + (False, True, 'table/p=a%2Fb/snapshot-1'), + (True, False, 'table/p=a%2Fb/snapshot-1'), +]) +def test_path_modes_and_sts(oss_server, jindo, legacy, path): + io = file_io(oss_server) + io.properties = options_for(oss_server, token='test-sts') + io._use_jindo, io._oss_bucket_in_endpoint = jindo, legacy + assert io.try_to_write_atomic(path, 'data') is True + assert oss_server.token == 'test-sts' + assert list(oss_server.objects) == ['/test-bucket/table/p=a%2Fb/snapshot-1'] + + +def test_wrong_bucket_rejected(oss_server): + with pytest.raises(ValueError, match='configured OSS bucket'): + file_io(oss_server).try_to_write_atomic('oss://other/snapshot-1', 'data') + assert oss_server.puts == 0 + + +def test_missing_credentials_do_not_fall_back(oss_server): + io = file_io(oss_server) + io.properties = Options({'fs.oss.endpoint': 'oss.example.com'}) + with pytest.raises(ValueError, match='fs.oss.accessKeyId'): + io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + assert oss_server.puts == 0 + + +def test_versioning_is_rechecked_for_a_long_lived_file_io(oss_server, tmp_path): + io = file_io(oss_server) + io.filesystem = pafs.LocalFileSystem() + assert io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'first') + oss_server.versioning = 'Enabled' + assert io.try_to_write_atomic('oss://test-bucket{}/snapshot-2'.format(tmp_path), 'second') + assert (tmp_path / 'snapshot-2').read_text() == 'second' + assert list(oss_server.objects.values()) == [b'first'] + + +@pytest.mark.parametrize('settings,expected', [ + ({'server-side-encryption': ' aes256 '}, {'server-side-encryption': 'AES256'}), + ({'server-side-encryption': 'sm4'}, {'server-side-encryption': 'SM4'}), + ({'server-side-encryption-key-id': ' my-cmk '}, + {'server-side-encryption': 'KMS', 'server-side-encryption-key-id': 'my-cmk'}), + ({'server-side-data-encryption': ' sm4 '}, + {'server-side-encryption': 'KMS', 'server-side-data-encryption': 'SM4'}), + ({'server-side-encryption': ' kms ', 'server-side-encryption-key-id': 'my-cmk', + 'server-side-data-encryption': 'sm4', 'server-side-encryption-algorithm': 'AES256'}, + {'server-side-encryption': 'KMS', 'server-side-encryption-key-id': 'my-cmk', + 'server-side-data-encryption': 'SM4'}), + ({'server-side-encryption-algorithm': 'AES256'}, {'server-side-encryption': 'AES256'}), +]) +def test_sse_headers_and_conditional_creation(oss_server, settings, expected): + io = file_io(oss_server) + io.properties = Options(dict(io.properties.to_map(), **{ + 'fs.oss.' + key: value for key, value in settings.items()})) + path = 'oss://test-bucket/snapshot-1' + assert io.try_to_write_atomic(path, 'encrypted metadata') + assert oss_server.sse_headers == expected + assert io.try_to_write_atomic(path, 'overwrite') is False + assert list(oss_server.objects.values()) == [b'encrypted metadata'] + + +@pytest.mark.parametrize('settings', [ + {'server-side-encryption': 'AES-256'}, + {'server-side-encryption': 'AES256', 'server-side-encryption-key-id': 'my-cmk'}, + {'server-side-encryption': 'SM4', 'server-side-data-encryption': 'SM4'}, + {'server-side-encryption': 'KMS', 'server-side-data-encryption': 'AES256'}, + {'server-side-encryption-key-id': 'my cmk'}, + {'server-side-encryption': '', 'server-side-encryption-algorithm': 'AES256'}, +]) +def test_invalid_sse_is_rejected_before_io(oss_server, settings): + io = file_io(oss_server) + io.properties = Options(dict(io.properties.to_map(), **{ + 'fs.oss.' + key: value for key, value in settings.items()})) + with pytest.raises(ValueError, match='fs.oss.server-side-'): + io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + assert oss_server.gets == 0 + assert oss_server.puts == 0 diff --git a/paimon-python/pypaimon/tests/oss_file_io_test.py b/paimon-python/pypaimon/tests/oss_file_io_test.py index d1537607853b..e14124ae369b 100644 --- a/paimon-python/pypaimon/tests/oss_file_io_test.py +++ b/paimon-python/pypaimon/tests/oss_file_io_test.py @@ -23,11 +23,11 @@ from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO +from pypaimon.filesystem.oss_file_io import OssFileIO class OSSFileIOTest(unittest.TestCase): - """Test cases for PyArrowFileIO with OSS.""" + """Test cases for OssFileIO with OSS.""" def setUp(self): """Set up test fixtures.""" @@ -50,19 +50,19 @@ def setUp(self): if not endpoint: self.skipTest("test endpoint is not configured") return - + self.root_path = f"oss://{self.bucket}/" - + self.catalog_options = Options({ OssOptions.OSS_ACCESS_KEY_ID.key(): access_key_id, OssOptions.OSS_ACCESS_KEY_SECRET.key(): access_key_secret, OssOptions.OSS_ENDPOINT.key(): endpoint, OssOptions.OSS_IMPL.key(): oss_impl, }) - - # Create PyArrowFileIO instance - self.file_io = PyArrowFileIO(self.root_path, self.catalog_options) - + + # Create OssFileIO instance + self.file_io = OssFileIO(self.root_path, self.catalog_options) + # Create unique test prefix to avoid conflicts self.test_prefix = f"test-{uuid.uuid4().hex[:8]}/" @@ -96,41 +96,41 @@ def test_new_input_stream_read(self): # Create test data test_data = b"Hello, World! This is a test file for OSS input stream." test_file = self._get_test_path("test-input-stream.txt") - + # Write test data to file with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(test_data) - + # Test new_input_stream input_stream = self.file_io.new_input_stream(test_file) self.assertIsNotNone(input_stream) - + # Test read without nbytes (read all) input_stream.seek(0) read_data = input_stream.read() self.assertEqual(read_data, test_data) - + # Test read with nbytes input_stream.seek(0) read_partial = input_stream.read(5) self.assertEqual(read_partial, b"Hello") - + # Test read more bytes read_partial2 = input_stream.read(7) self.assertEqual(read_partial2, b", World") - + # Test read remaining read_remaining = input_stream.read() self.assertEqual(read_remaining, b"! This is a test file for OSS input stream.") - + # Verify complete data input_stream.seek(0) complete_data = input_stream.read() self.assertEqual(complete_data, test_data) - + # Close the stream input_stream.close() - + # Test context manager with self.file_io.new_input_stream(test_file) as input_stream2: data = input_stream2.read() @@ -141,11 +141,11 @@ def test_new_input_stream_read_large_file(self): # Create larger test data (1MB) test_data = b"X" * (1024 * 1024) test_file = self._get_test_path("test-large-input-stream.bin") - + # Write test data with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(test_data) - + # Test reading in chunks chunk_size = 64 * 1024 # 64KB chunks with self.file_io.new_input_stream(test_file) as input_stream: @@ -155,12 +155,12 @@ def test_new_input_stream_read_large_file(self): if not chunk: break read_chunks.append(chunk) - + # Verify all data was read read_data = b''.join(read_chunks) self.assertEqual(len(read_data), len(test_data)) self.assertEqual(read_data, test_data) - + # Test read_at method if available with self.file_io.new_input_stream(test_file) as input_stream: if hasattr(input_stream, 'read_at'): @@ -173,7 +173,7 @@ def test_new_input_stream_read_large_file(self): def test_new_input_stream_file_not_found(self): """Test new_input_stream with non-existent file.""" non_existent_file = self._get_test_path("non-existent-file.txt") - + with self.assertRaises(FileNotFoundError): self.file_io.new_input_stream(non_existent_file) @@ -210,7 +210,7 @@ def test_write_file_with_overwrite_flag(self): def test_exists_does_not_catch_exception(self): """Test that exists does not catch exceptions.""" test_file = self._get_test_path("test_file.txt") - + # Write a test file with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(b"test") @@ -242,7 +242,7 @@ def test_delete_returns_false_when_file_not_exists(self): def test_mkdirs_raises_error_when_path_is_file(self): """Test that mkdirs raises error when path is a file.""" test_file = self._get_test_path("test_file.txt") - + # Create a file with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(b"test") @@ -255,7 +255,7 @@ def test_rename_returns_false_when_dst_exists(self): """Test that rename returns False when destination exists.""" src_file = self._get_test_path("src.txt") dst_file = self._get_test_path("dst.txt") - + # Create source and destination files with self.file_io.new_output_stream(src_file) as out_stream: out_stream.write(b"src") @@ -274,7 +274,7 @@ def test_get_file_status_raises_error_when_file_not_exists(self): test_file = self._get_test_path("test_file.txt") with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(b"test content") - + file_info = self.file_io.get_file_status(test_file) self.assertEqual(file_info.type, pafs.FileType.File) self.assertIsNotNone(file_info.size) @@ -291,30 +291,30 @@ def test_copy_file(self): """Test copy_file method.""" source_file = self._get_test_path("source.txt") target_file = self._get_test_path("target.txt") - + # Create source file with self.file_io.new_output_stream(source_file) as out_stream: out_stream.write(b"source content") - + # Test 1: Raises FileExistsError when target exists and overwrite=False with self.file_io.new_output_stream(target_file) as out_stream: out_stream.write(b"target content") - + with self.assertRaises(FileExistsError) as context: self.file_io.copy_file(source_file, target_file, overwrite=False) self.assertIn("already exists", str(context.exception)) - + # Verify target content unchanged with self.file_io.new_input_stream(target_file) as in_stream: content = in_stream.read() self.assertEqual(content, b"target content") - + # Test 2: Overwrites when overwrite=True self.file_io.copy_file(source_file, target_file, overwrite=True) with self.file_io.new_input_stream(target_file) as in_stream: content = in_stream.read() self.assertEqual(content, b"source content") - + # Test 3: Creates parent directory if it doesn't exist target_file_in_subdir = self._get_test_path("subdir/target.txt") self.file_io.copy_file(source_file, target_file_in_subdir, overwrite=False) @@ -327,33 +327,33 @@ def test_try_to_write_atomic(self): """Test try_to_write_atomic method.""" target_dir = self._get_test_path("target_dir/") normal_file = self._get_test_path("normal_file.txt") - + # Create target directory self.file_io.mkdirs(target_dir) self.assertFalse( self.file_io.try_to_write_atomic(target_dir, "test content"), - "PyArrowFileIO should return False when target is a directory") - + "OssFileIO should return False when target is a directory") + # Verify no file was created inside the directory # List directory contents to verify it's empty selector = pafs.FileSelector(self.file_io.to_filesystem_path(target_dir), recursive=False, allow_not_found=True) dir_contents = self.file_io.filesystem.get_file_info(selector) self.assertEqual(len(dir_contents), 0, "No file should be created inside the directory") - + self.assertTrue(self.file_io.try_to_write_atomic(normal_file, "test content")) content = self.file_io.read_file_utf8(normal_file) self.assertEqual(content, "test content") - + # Delete and test again self.file_io.delete(normal_file) self.assertFalse( self.file_io.try_to_write_atomic(target_dir, "test content"), - "PyArrowFileIO should return False when target is a directory") - + "OssFileIO should return False when target is a directory") + # Verify no file was created inside the directory dir_contents = self.file_io.filesystem.get_file_info(selector) self.assertEqual(len(dir_contents), 0, "No file should be created inside the directory") - + self.assertTrue(self.file_io.try_to_write_atomic(normal_file, "test content")) content = self.file_io.read_file_utf8(normal_file) self.assertEqual(content, "test content") diff --git a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py index 43c4972f0cf1..dc300cb391dc 100644 --- a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py +++ b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py @@ -15,9 +15,9 @@ # specific language governing permissions and limitations # under the License. -"""Unit tests for the OSS bucket-in-endpoint mode (PyArrow < 16) of PyArrowFileIO. +"""Unit tests for the OSS bucket-in-endpoint mode (PyArrow < 16) of OssFileIO. -See ``PyArrowFileIO._legacy_oss_mode`` for why bucket-level operations +See ``OssFileIO._legacy_oss_mode`` for why bucket-level operations must be guarded in this mode. No real OSS access is required. """ @@ -28,11 +28,8 @@ from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions -from pypaimon.filesystem.pyarrow_file_io import ( - LegacyOssDirectoryListingError, - PyArrowFileIO, -) - +from pypaimon.filesystem.oss_file_io import OssFileIO +from pypaimon.filesystem.pyarrow_file_io import LegacyOssDirectoryListingError TABLE_PATH = "oss://test-bucket/db-uuid.db/tbl-uuid" @@ -48,7 +45,7 @@ def _probe_response(status_code, body): class OssLegacyModeTest(unittest.TestCase): - """Behavior of PyArrowFileIO when OSS runs on PyArrow < 16.""" + """Behavior of OssFileIO when OSS runs on PyArrow < 16.""" def _new_file_io(self, legacy): options = Options({ @@ -59,8 +56,8 @@ def _new_file_io(self, legacy): OssOptions.OSS_IMPL.key(): "legacy", }) with mock.patch.object( - PyArrowFileIO, "_initialize_oss_fs", return_value=mock.Mock()): - file_io = PyArrowFileIO("oss://test-bucket/", options) + OssFileIO, "_initialize_oss_fs", return_value=mock.Mock()): + file_io = OssFileIO("oss://test-bucket/", options) # _legacy_oss_mode() keys off the bucket-in-endpoint flag (PyArrow < 16). file_io._oss_bucket_in_endpoint = legacy file_io.filesystem = mock.Mock() @@ -214,7 +211,7 @@ def test_file_io_pickle_roundtrip_recreates_lock(self): OssOptions.OSS_REGION.key(): "cn-test", OssOptions.OSS_IMPL.key(): "legacy", }) - file_io = PyArrowFileIO("oss://test-bucket/wh", options) + file_io = OssFileIO("oss://test-bucket/wh", options) file_io._legacy_bucket_checked = True file_io._legacy_bucket_error = "OSS bucket 'test-bucket' does not exist" diff --git a/paimon-python/pypaimon/tests/py36/ao_simple_test.py b/paimon-python/pypaimon/tests/py36/ao_simple_test.py index 792387179de4..662d0b720e03 100644 --- a/paimon-python/pypaimon/tests/py36/ao_simple_test.py +++ b/paimon-python/pypaimon/tests/py36/ao_simple_test.py @@ -26,7 +26,7 @@ TableNotExistException) from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO +from pypaimon.filesystem.oss_file_io import OssFileIO from pypaimon.tests.py36.pyarrow_compat import table_sort_by from pypaimon.tests.rest.rest_base_test import RESTBaseTest @@ -404,19 +404,19 @@ def test_initialize_oss_fs_pyarrow_lt_7(self): with patch("pypaimon.common.file_io.pyarrow.__version__", "6.0.0"), \ patch("pyarrow.fs.S3FileSystem") as mock_s3fs: - PyArrowFileIO("oss://oss-bucket/paimon-database/paimon-table", Options(props)) + OssFileIO("oss://oss-bucket/paimon-database/paimon-table", Options(props)) mock_s3fs.assert_called_once_with(access_key="AKID", secret_key="SECRET", session_token="TOKEN", region="cn-hangzhou", endpoint_override="oss-bucket." + props[OssOptions.OSS_ENDPOINT.key()]) - PyArrowFileIO("oss://oss-bucket.endpoint/paimon-database/paimon-table", Options(props)) + OssFileIO("oss://oss-bucket.endpoint/paimon-database/paimon-table", Options(props)) mock_s3fs.assert_called_with(access_key="AKID", secret_key="SECRET", session_token="TOKEN", region="cn-hangzhou", endpoint_override="oss-bucket." + props[OssOptions.OSS_ENDPOINT.key()]) - PyArrowFileIO("oss://access_id:secret_key@Endpoint/oss-bucket/paimon-database/paimon-table", Options(props)) + OssFileIO("oss://access_id:secret_key@Endpoint/oss-bucket/paimon-database/paimon-table", Options(props)) mock_s3fs.assert_called_with(access_key="AKID", secret_key="SECRET", session_token="TOKEN", diff --git a/paimon-python/setup.py b/paimon-python/setup.py index 0a08d7acca20..60cfe979d32d 100644 --- a/paimon-python/setup.py +++ b/paimon-python/setup.py @@ -22,6 +22,7 @@ import sys import tarfile import tempfile + from setuptools import find_packages, setup from setuptools.command.build_py import build_py from setuptools.command.sdist import sdist @@ -270,10 +271,12 @@ def read_requirements(): 'daft>=0.7.6; python_version>="3.10"', ], 'oss': [ + 'oss2>=2.18,<3', 'ossfs>=2021.8; python_version<"3.8"', 'ossfs>=2023; python_version>="3.8"' ], 'jindo': [ + 'oss2>=2.18,<3', 'pyjindosdk>=6.10.4', ], 'lance': [ From a1d6b62691f31b95c2d3126975e4d5113db9ee4b Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Thu, 10 Sep 2026 13:50:15 +0800 Subject: [PATCH 2/7] [python] Correct OSS atomic keys for credential URIs --- .../pypaimon/filesystem/oss_file_io.py | 2 ++ .../pypaimon/tests/oss_atomic_write_test.py | 17 +++++++++++++++-- 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py b/paimon-python/pypaimon/filesystem/oss_file_io.py index a31097ee3121..fe6b5af8d446 100644 --- a/paimon-python/pypaimon/filesystem/oss_file_io.py +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -33,6 +33,8 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: if uri.scheme != 'oss' or self._extract_oss_bucket(path) != self._oss_bucket: raise ValueError("Atomic write must target the configured OSS bucket") key = re.sub(r'/+', '/', uri.path).lstrip('/') + if '@' in uri.netloc: + key = key.partition('/')[2] else: key = path if not self._use_jindo and not self._oss_bucket_in_endpoint: diff --git a/paimon-python/pypaimon/tests/oss_atomic_write_test.py b/paimon-python/pypaimon/tests/oss_atomic_write_test.py index 3ca19a91b148..8d47f2d18c35 100644 --- a/paimon-python/pypaimon/tests/oss_atomic_write_test.py +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -214,6 +214,9 @@ def test_lost_response_is_not_replayed_or_reported_as_conflict(oss_server): (False, False, 'test-bucket/table/p=a%2Fb/snapshot-1'), (False, True, 'table/p=a%2Fb/snapshot-1'), (True, False, 'table/p=a%2Fb/snapshot-1'), + (False, False, 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1'), + (False, True, 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1'), + (True, False, 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1'), ]) def test_path_modes_and_sts(oss_server, jindo, legacy, path): io = file_io(oss_server) @@ -222,11 +225,21 @@ def test_path_modes_and_sts(oss_server, jindo, legacy, path): assert io.try_to_write_atomic(path, 'data') is True assert oss_server.token == 'test-sts' assert list(oss_server.objects) == ['/test-bucket/table/p=a%2Fb/snapshot-1'] + assert io.try_to_write_atomic('oss://test-bucket/table/p=a%2Fb/snapshot-1', 'overwrite') is False + assert list(oss_server.objects.values()) == [b'data'] -def test_wrong_bucket_rejected(oss_server): +@pytest.mark.parametrize('path', ['oss://other/snapshot-1', 'oss://AK:SK@endpoint/other/snapshot-1']) +def test_wrong_bucket_rejected(oss_server, path): with pytest.raises(ValueError, match='configured OSS bucket'): - file_io(oss_server).try_to_write_atomic('oss://other/snapshot-1', 'data') + file_io(oss_server).try_to_write_atomic(path, 'data') + assert oss_server.puts == 0 + + +@pytest.mark.parametrize('path', ['oss://AK:SK@endpoint/test-bucket', 'oss://AK:SK@endpoint/test-bucket/']) +def test_credential_uri_bucket_root_is_not_an_object(oss_server, path): + assert file_io(oss_server).try_to_write_atomic(path, 'data') is False + assert oss_server.gets == 0 assert oss_server.puts == 0 From df23c20cc5b306fa66d124df573e5c54c806a149 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Thu, 10 Sep 2026 19:08:55 +0800 Subject: [PATCH 3/7] [python] Isolate REST FileIO caches and normalize OSS fallback paths --- .../catalog/rest/rest_token_file_io.py | 40 ++--- .../pypaimon/filesystem/oss_file_io.py | 1 + .../pypaimon/tests/oss_atomic_write_test.py | 29 +++- .../rest/rest_token_file_io_cache_test.py | 154 ++++++++++++++++++ .../tests/rest/rest_token_file_io_test.py | 23 +++ 5 files changed, 216 insertions(+), 31 deletions(-) create mode 100644 paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index 633cc7ae8242..6a87d004ebdf 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py +++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py @@ -39,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 @@ -76,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 @@ -90,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 diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py b/paimon-python/pypaimon/filesystem/oss_file_io.py index fe6b5af8d446..8b506343e00e 100644 --- a/paimon-python/pypaimon/filesystem/oss_file_io.py +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -35,6 +35,7 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: key = re.sub(r'/+', '/', uri.path).lstrip('/') if '@' in uri.netloc: key = key.partition('/')[2] + path = 'oss://{}/{}'.format(self._oss_bucket, key) else: key = path if not self._use_jindo and not self._oss_bucket_in_endpoint: diff --git a/paimon-python/pypaimon/tests/oss_atomic_write_test.py b/paimon-python/pypaimon/tests/oss_atomic_write_test.py index 8d47f2d18c35..140a2cf2e1fa 100644 --- a/paimon-python/pypaimon/tests/oss_atomic_write_test.py +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -151,7 +151,6 @@ def test_rest_token_refresh_keeps_oss_atomic_creation(oss_server): options[CatalogOptions.RESOLVING_FILE_IO_ENABLED.key()] = 'true' path = 'oss://test-bucket/table/snapshot-1' with mock.patch.object(RESTTokenFileIO, 'try_to_refresh_token'), \ - mock.patch.object(RESTTokenFileIO, '_FILE_IO_CACHE', {}), \ mock.patch.object(OssFileIO, '_initialize_oss_fs'): io = RESTTokenFileIO(Identifier.from_string('default.table'), path, Options(options)) io.token = RESTToken({'fs.oss.securityToken': 'first-token'}, 1) @@ -164,8 +163,29 @@ def test_rest_token_refresh_keeps_oss_atomic_creation(oss_server): assert sorted(oss_server.objects.values()) == [b'first', b'second'] +@pytest.mark.parametrize('second_path,method', [ + ('oss://other-bucket/table', 'AES256'), + ('oss://test-bucket/table', 'KMS'), +]) +def test_rest_file_io_isolates_bucket_and_encryption(oss_server, second_path, method): + with mock.patch.object(RESTTokenFileIO, 'try_to_refresh_token'), \ + mock.patch.object(OssFileIO, '_initialize_oss_fs'): + for index, (path, encryption) in enumerate([ + ('oss://test-bucket/table', 'AES256'), (second_path, method)]): + options = dict(options_for(oss_server).to_map()) + options['fs.oss.server-side-encryption'] = encryption + io = RESTTokenFileIO(Identifier.from_string('default.table'), path, Options(options)) + io.token = RESTToken({'fs.oss.securityToken': 'shared-token'}, oss_server.server_port) + target = path + '/snapshot-' + str(index) + assert io.try_to_write_atomic(target, str(index)) + assert oss_server.sse_headers == {'server-side-encryption': encryption} + assert oss_server.objects['/' + target[len('oss://'):]] == str(index).encode() + assert io.try_to_write_atomic(target, 'overwrite') is False + + @pytest.mark.parametrize('versioning', ['Enabled', 'Suspended', 'Unexpected', None]) -def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, tmp_path, caplog): +@pytest.mark.parametrize('credential_uri', [False, True]) +def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, credential_uri, tmp_path, caplog): oss_server.versioning = versioning if versioning is None: oss_server.fail_method = 'GET' @@ -174,6 +194,11 @@ def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, tmp # Run the inherited stream/rename operations against an actual Arrow filesystem. io.filesystem = pafs.LocalFileSystem() path = 'oss://test-bucket{}/snapshot-1'.format(tmp_path) + if credential_uri: + # Jindo uses key-only paths; keep all fallback writes inside the temporary directory. + io._use_jindo = True + io.filesystem = pafs.SubTreeFileSystem(str(tmp_path), pafs.LocalFileSystem()) + path = 'oss://AK:SK@endpoint/test-bucket/snapshot-1' assert io.try_to_write_atomic(path, '兼容写入') is True assert io.try_to_write_atomic(path, 'overwrite') is False assert (tmp_path / 'snapshot-1').read_text() == '兼容写入' diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py new file mode 100644 index 000000000000..d7cc0f0c906b --- /dev/null +++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py @@ -0,0 +1,154 @@ +# 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. + +"""Local lifecycle tests for REST FileIO caching; only the remote token API is stubbed.""" + +import pickle +import subprocess +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from unittest import mock + +import pytest +from cachetools import TTLCache + +from pypaimon.api.rest_api import RESTApi +from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO +from pypaimon.common.file_io import FileIO +from pypaimon.common.identifier import Identifier +from pypaimon.common.options import Options + + +@pytest.fixture +def options(): + with mock.patch.object(RESTTokenFileIO, '_TOKEN_CACHE', {}), \ + mock.patch.object(RESTTokenFileIO, '_TOKEN_LOCKS', {}): + yield Options({'uri': 'http://127.0.0.1:1', 'token.provider': 'bear', 'token': 'test-user'}) + + +def new_io(path, options): + return RESTTokenFileIO(Identifier.from_string('db.table'), str(path), options) + + +def token_response(now, value): + return SimpleNamespace(token={'test.credential': value}, expires_at_millis=int(now * 1000) + 7_200_000) + + +def test_concurrent_expiry_refresh_keeps_instance_backends_isolated(tmp_path, options): + clock = [1_000_000.0] + instances = [new_io(tmp_path / str(index), options) for index in range(2)] + previous = [None, None] + with mock.patch('pypaimon.catalog.rest.rest_token_file_io.time.time', side_effect=lambda: clock[0]), \ + mock.patch.object(RESTApi, 'load_table_token') as load: + for generation in range(3): + load.return_value = token_response(clock[0], str(generation)) + barrier = threading.Barrier(16) + + def write(index): + barrier.wait(timeout=10) + io = instances[index % 2] + backend = io.file_io() + path = io.path + '/generation-{}-writer-{}'.format(generation, index) + backend.write_file(path, str(generation)) + return backend, path + + with ThreadPoolExecutor(max_workers=16) as pool: + results = list(pool.map(write, range(16))) + assert load.call_count == generation + 1 + for index, io in enumerate(instances): + backend = results[index][0] + assert backend is not previous[index] + assert all(result[0] is backend for result in results[index::2]) + assert backend.path == io.path + assert backend.properties.to_map()['test.credential'] == str(generation) + assert all(io.read_file_utf8(path) == str(generation) for _, path in results[index::2]) + previous[index] = backend + assert previous[0] is not previous[1] + clock[0] += 7_200 + + +def test_refresh_and_initialization_failures_do_not_poison_cache(tmp_path, options): + io = new_io(tmp_path, options) + clock = [1_000_000.0] + responses = [token_response(clock[0], 'old'), OSError('token unavailable'), + token_response(clock[0] + 7_200, 'new')] + with mock.patch('pypaimon.catalog.rest.rest_token_file_io.time.time', side_effect=lambda: clock[0]), \ + mock.patch.object(RESTApi, 'load_table_token', side_effect=responses) as load: + old = io.file_io() + clock[0] += 7_200 + with pytest.raises(OSError, match='token unavailable'): + io.file_io() + with mock.patch.object(FileIO, 'get', side_effect=OSError('backend unavailable')): + with pytest.raises(OSError, match='backend unavailable'): + io.file_io() + backend = io.file_io() + assert backend is not old + assert io.file_io() is backend + assert load.call_count == 3 + path = str(tmp_path / 'recovered') + io.write_file(path, 'ok') + assert io.read_file_utf8(path) == 'ok' + assert backend.properties.to_map()['test.credential'] == 'new' + + +def test_cache_ttl_rebuilds_backend_without_refetching_valid_token(tmp_path, options): + clock = [0.0] + with mock.patch('pypaimon.catalog.rest.rest_token_file_io.TTLCache', + side_effect=lambda **kwargs: TTLCache(timer=lambda: clock[0], **kwargs)), \ + mock.patch.object(RESTTokenFileIO, '_FILE_IO_CACHE_TTL', 10), \ + mock.patch('pypaimon.catalog.rest.rest_token_file_io.time.time', return_value=1_000_000), \ + mock.patch.object(RESTApi, 'load_table_token', return_value=token_response(1_000_000, 'valid')) as load: + io = new_io(tmp_path, options) + first = io.file_io() + path = str(tmp_path / 'before-expiry') + io.write_file(path, 'preserved') + clock[0] = 9 + assert io.file_io() is first + clock[0] = 11 + assert io.file_io() is not first + assert io.read_file_utf8(path) == 'preserved' + assert load.call_count == 1 + + +def test_initialized_file_io_can_be_used_in_fresh_processes(tmp_path, options): + import time + + io = new_io(tmp_path, options) + with mock.patch.object(RESTApi, 'load_table_token', return_value=token_response(time.time(), 'valid')): + io.write_file(str(tmp_path / 'parent'), 'parent data') + payload = pickle.dumps(io) + code = ''' +import pickle, sys +io = pickle.loads(sys.stdin.buffer.read()) +assert io.read_file_utf8(io.path + '/parent') == 'parent data' +backend = io.file_io() +assert io.file_io() is backend +assert backend.properties.to_map()['test.credential'] == 'valid' +io.write_file(io.path + '/child-' + sys.argv[1], 'child data') +''' + + def run_child(index): + result = subprocess.run([sys.executable, '-c', code, str(index)], input=payload, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30) + assert result.returncode == 0, result.stderr.decode() + + with ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(run_child, range(4))) + for index in range(4): + assert io.read_file_utf8(str(tmp_path / ('child-' + str(index)))) == 'child data' diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py index dbb919eed28b..ed0e20b848e3 100644 --- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py @@ -18,8 +18,10 @@ import os import pickle import tempfile +import threading import time import unittest +from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch, MagicMock from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO @@ -147,6 +149,8 @@ def test_pickle_serialization(self): self.warehouse_path, self.catalog_options ) + original_file_io.token = RESTToken({}, 1) + original_backend = original_file_io.file_io() pickled = pickle.dumps(original_file_io) @@ -157,6 +161,7 @@ def test_pickle_serialization(self): self.assertEqual(deserialized_file_io.properties.data, original_file_io.properties.data) self.assertIsNone(deserialized_file_io.api_instance) + self.assertIsNot(deserialized_file_io.file_io(), original_backend) test_file_path = f"file://{self.temp_dir}/pickle_test.txt" test_content = b"pickle test content" @@ -169,6 +174,24 @@ def test_pickle_serialization(self): with open(expected_path, 'rb') as f: self.assertEqual(f.read(), test_content) + def test_concurrent_file_io_reuse_and_token_refresh(self): + with patch.object(RESTTokenFileIO, 'try_to_refresh_token'): + file_io = RESTTokenFileIO(self.identifier, self.warehouse_path, self.catalog_options) + file_io.token = RESTToken({}, 1) + barrier = threading.Barrier(8) + + def get_backend(_): + barrier.wait(timeout=10) + return file_io.file_io() + + with ThreadPoolExecutor(max_workers=8) as pool: + backends = list(pool.map(get_backend, range(8))) + self.assertTrue(all(backend is backends[0] for backend in backends)) + file_io.token = RESTToken({}, 2) + refreshed = file_io.file_io() + self.assertIsNot(refreshed, backends[0]) + self.assertIs(file_io.file_io(), refreshed) + def test_dlf_oss_endpoint_overrides_token_endpoint(self): """Test that DLF OSS endpoint overrides the standard OSS endpoint in token.""" dlf_oss_endpoint = "https://dlf-custom-endpoint.oss-cn-hangzhou.aliyuncs.com" From e7b45a0c25629a4f467a8885ab775b8d81bc5a7d Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Thu, 10 Sep 2026 19:18:42 +0800 Subject: [PATCH 4/7] [python] Keep essential OSS atomic write regression tests --- paimon-python/pypaimon/tests/file_io_test.py | 22 +-- .../pypaimon/tests/oss_atomic_write_test.py | 116 ++----------- .../pypaimon/tests/oss_file_io_test.py | 68 ++++---- .../rest/rest_token_file_io_cache_test.py | 154 ------------------ .../tests/rest/rest_token_file_io_test.py | 20 --- 5 files changed, 61 insertions(+), 319 deletions(-) delete mode 100644 paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py diff --git a/paimon-python/pypaimon/tests/file_io_test.py b/paimon-python/pypaimon/tests/file_io_test.py index 8f8d722bc807..53c8e8d29698 100644 --- a/paimon-python/pypaimon/tests/file_io_test.py +++ b/paimon-python/pypaimon/tests/file_io_test.py @@ -442,7 +442,7 @@ def test_get_file_status_raises_error_when_file_not_exists(self): test_file = os.path.join(temp_dir, "test_file.txt") with open(test_file, "w") as f: f.write("test content") - + file_info = file_io.get_file_status(f"file://{test_file}") self.assertEqual(file_info.type, pafs.FileType.File) self.assertIsNotNone(file_info.size) @@ -465,26 +465,26 @@ def test_copy_file(self): source_file = os.path.join(temp_dir, "source.txt") target_file = os.path.join(temp_dir, "target.txt") - + with open(source_file, "w") as f: f.write("source content") - + # Test 1: Raises FileExistsError when target exists and overwrite=False with open(target_file, "w") as f: f.write("target content") - + with self.assertRaises(FileExistsError) as context: file_io.copy_file(f"file://{source_file}", f"file://{target_file}", overwrite=False) self.assertIn("already exists", str(context.exception)) - + with open(target_file, "r") as f: self.assertEqual(f.read(), "target content") - + # Test 2: Overwrites when overwrite=True file_io.copy_file(f"file://{source_file}", f"file://{target_file}", overwrite=True) with open(target_file, "r") as f: self.assertEqual(f.read(), "source content") - + # Test 3: Creates parent directory if it doesn't exist target_file_in_subdir = os.path.join(temp_dir, "subdir", "target.txt") file_io.copy_file(f"file://{source_file}", f"file://{target_file_in_subdir}", overwrite=False) @@ -499,7 +499,7 @@ def test_try_to_write_atomic(self): try: target_dir = os.path.join(temp_dir, "target_dir") normal_file = os.path.join(temp_dir, "normal_file.txt") - + from pypaimon.filesystem.local_file_io import LocalFileIO local_file_io = LocalFileIO(f"file://{temp_dir}", Options({})) os.makedirs(target_dir) @@ -507,18 +507,18 @@ def test_try_to_write_atomic(self): local_file_io.try_to_write_atomic(f"file://{target_dir}", "test content"), "LocalFileIO should return False when target is a directory") self.assertEqual(len(os.listdir(target_dir)), 0, "No file should be created inside the directory") - + self.assertTrue(local_file_io.try_to_write_atomic(f"file://{normal_file}", "test content")) with open(normal_file, "r") as f: self.assertEqual(f.read(), "test content") - + os.remove(normal_file) local_file_io = LocalFileIO(f"file://{temp_dir}", Options({})) self.assertFalse( local_file_io.try_to_write_atomic(f"file://{target_dir}", "test content"), "LocalFileIO should return False when target is a directory") self.assertEqual(len(os.listdir(target_dir)), 0, "No file should be created inside the directory") - + self.assertTrue(local_file_io.try_to_write_atomic(f"file://{normal_file}", "test content")) with open(normal_file, "r") as f: self.assertEqual(f.read(), "test content") diff --git a/paimon-python/pypaimon/tests/oss_atomic_write_test.py b/paimon-python/pypaimon/tests/oss_atomic_write_test.py index 140a2cf2e1fa..b1a3fbc06f8d 100644 --- a/paimon-python/pypaimon/tests/oss_atomic_write_test.py +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -130,39 +130,23 @@ def file_io(server, resolving=False): @pytest.mark.parametrize('resolving', [False, True]) def test_atomic_competition_and_existing_content(oss_server, resolving): io = file_io(oss_server, resolving) - path = 'oss://test-bucket/table/snapshot-1' + path = 'oss://test-bucket/table/p=a%2Fb/snapshot-1' barrier = threading.Barrier(2) - def write(content): + def write(index): barrier.wait(timeout=10) - return io.try_to_write_atomic(path, content) + target = 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1' if index == 0 else path + return io.try_to_write_atomic(target, contents[index]) contents = ['提交者一', '提交者二'] with mock.patch.object(OssFileIO, '_initialize_oss_fs'), ThreadPoolExecutor(max_workers=2) as pool: - results = list(pool.map(write, contents)) + results = list(pool.map(write, range(2))) assert sorted(results) == [False, True] - assert list(oss_server.objects.values()) == [contents[results.index(True)].encode()] + assert oss_server.objects == {'/test-bucket/table/p=a%2Fb/snapshot-1': contents[results.index(True)].encode()} assert io.try_to_write_atomic(path, 'overwrite') is False assert list(oss_server.objects.values()) == [contents[results.index(True)].encode()] -def test_rest_token_refresh_keeps_oss_atomic_creation(oss_server): - options = dict(options_for(oss_server).to_map()) - options[CatalogOptions.RESOLVING_FILE_IO_ENABLED.key()] = 'true' - path = 'oss://test-bucket/table/snapshot-1' - with mock.patch.object(RESTTokenFileIO, 'try_to_refresh_token'), \ - mock.patch.object(OssFileIO, '_initialize_oss_fs'): - io = RESTTokenFileIO(Identifier.from_string('default.table'), path, Options(options)) - io.token = RESTToken({'fs.oss.securityToken': 'first-token'}, 1) - assert io.try_to_write_atomic(path, 'first') - assert oss_server.token == 'first-token' - io.token = RESTToken({'fs.oss.securityToken': 'refreshed-token'}, 2) - assert io.try_to_write_atomic(path, 'overwrite') is False - assert io.try_to_write_atomic(path.replace('snapshot-1', 'snapshot-2'), 'second') - assert oss_server.token == 'refreshed-token' - assert sorted(oss_server.objects.values()) == [b'first', b'second'] - - @pytest.mark.parametrize('second_path,method', [ ('oss://other-bucket/table', 'AES256'), ('oss://test-bucket/table', 'KMS'), @@ -181,24 +165,22 @@ def test_rest_file_io_isolates_bucket_and_encryption(oss_server, second_path, me assert oss_server.sse_headers == {'server-side-encryption': encryption} assert oss_server.objects['/' + target[len('oss://'):]] == str(index).encode() assert io.try_to_write_atomic(target, 'overwrite') is False + io.token = RESTToken({'fs.oss.securityToken': 'refreshed-token'}, oss_server.server_port + 1) + assert io.try_to_write_atomic(target + '-next', 'next') + assert oss_server.token == 'refreshed-token' -@pytest.mark.parametrize('versioning', ['Enabled', 'Suspended', 'Unexpected', None]) -@pytest.mark.parametrize('credential_uri', [False, True]) -def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, credential_uri, tmp_path, caplog): +@pytest.mark.parametrize('versioning', ['Enabled', 'Suspended', None]) +def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, tmp_path, caplog): oss_server.versioning = versioning if versioning is None: oss_server.fail_method = 'GET' oss_server.failure = (403, 'AccessDenied') io = file_io(oss_server) - # Run the inherited stream/rename operations against an actual Arrow filesystem. - io.filesystem = pafs.LocalFileSystem() - path = 'oss://test-bucket{}/snapshot-1'.format(tmp_path) - if credential_uri: - # Jindo uses key-only paths; keep all fallback writes inside the temporary directory. - io._use_jindo = True - io.filesystem = pafs.SubTreeFileSystem(str(tmp_path), pafs.LocalFileSystem()) - path = 'oss://AK:SK@endpoint/test-bucket/snapshot-1' + # Exercise inherited stream/rename operations using Jindo's key-only path convention. + io._use_jindo = True + io.filesystem = pafs.SubTreeFileSystem(str(tmp_path), pafs.LocalFileSystem()) + path = 'oss://AK:SK@endpoint/test-bucket/snapshot-1' assert io.try_to_write_atomic(path, '兼容写入') is True assert io.try_to_write_atomic(path, 'overwrite') is False assert (tmp_path / 'snapshot-1').read_text() == '兼容写入' @@ -208,12 +190,8 @@ def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, cre @pytest.mark.parametrize('method,status,code', [ - ('GET', 403, 'InvalidAccessKeyId'), ('GET', 403, 'SecurityTokenExpired'), - ('GET', 404, 'NoSuchBucket'), - ('GET', 500, 'InternalError'), ('PUT', 403, 'AccessDenied'), - ('PUT', 500, 'InternalError'), ('PUT', 409, 'OtherConflict'), ]) def test_errors_are_not_competition(oss_server, method, status, code): @@ -234,66 +212,8 @@ def test_lost_response_is_not_replayed_or_reported_as_conflict(oss_server): assert list(oss_server.objects.values()) == [b'data'] -@pytest.mark.parametrize('jindo,legacy,path', [ - (False, False, 'oss://test-bucket/table/p=a%2Fb/snapshot-1'), - (False, False, 'test-bucket/table/p=a%2Fb/snapshot-1'), - (False, True, 'table/p=a%2Fb/snapshot-1'), - (True, False, 'table/p=a%2Fb/snapshot-1'), - (False, False, 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1'), - (False, True, 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1'), - (True, False, 'oss://AK:SK@endpoint/test-bucket/table/p=a%2Fb/snapshot-1'), -]) -def test_path_modes_and_sts(oss_server, jindo, legacy, path): - io = file_io(oss_server) - io.properties = options_for(oss_server, token='test-sts') - io._use_jindo, io._oss_bucket_in_endpoint = jindo, legacy - assert io.try_to_write_atomic(path, 'data') is True - assert oss_server.token == 'test-sts' - assert list(oss_server.objects) == ['/test-bucket/table/p=a%2Fb/snapshot-1'] - assert io.try_to_write_atomic('oss://test-bucket/table/p=a%2Fb/snapshot-1', 'overwrite') is False - assert list(oss_server.objects.values()) == [b'data'] - - -@pytest.mark.parametrize('path', ['oss://other/snapshot-1', 'oss://AK:SK@endpoint/other/snapshot-1']) -def test_wrong_bucket_rejected(oss_server, path): - with pytest.raises(ValueError, match='configured OSS bucket'): - file_io(oss_server).try_to_write_atomic(path, 'data') - assert oss_server.puts == 0 - - -@pytest.mark.parametrize('path', ['oss://AK:SK@endpoint/test-bucket', 'oss://AK:SK@endpoint/test-bucket/']) -def test_credential_uri_bucket_root_is_not_an_object(oss_server, path): - assert file_io(oss_server).try_to_write_atomic(path, 'data') is False - assert oss_server.gets == 0 - assert oss_server.puts == 0 - - -def test_missing_credentials_do_not_fall_back(oss_server): - io = file_io(oss_server) - io.properties = Options({'fs.oss.endpoint': 'oss.example.com'}) - with pytest.raises(ValueError, match='fs.oss.accessKeyId'): - io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') - assert oss_server.puts == 0 - - -def test_versioning_is_rechecked_for_a_long_lived_file_io(oss_server, tmp_path): - io = file_io(oss_server) - io.filesystem = pafs.LocalFileSystem() - assert io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'first') - oss_server.versioning = 'Enabled' - assert io.try_to_write_atomic('oss://test-bucket{}/snapshot-2'.format(tmp_path), 'second') - assert (tmp_path / 'snapshot-2').read_text() == 'second' - assert list(oss_server.objects.values()) == [b'first'] - - @pytest.mark.parametrize('settings,expected', [ - ({'server-side-encryption': ' aes256 '}, {'server-side-encryption': 'AES256'}), - ({'server-side-encryption': 'sm4'}, {'server-side-encryption': 'SM4'}), - ({'server-side-encryption-key-id': ' my-cmk '}, - {'server-side-encryption': 'KMS', 'server-side-encryption-key-id': 'my-cmk'}), - ({'server-side-data-encryption': ' sm4 '}, - {'server-side-encryption': 'KMS', 'server-side-data-encryption': 'SM4'}), - ({'server-side-encryption': ' kms ', 'server-side-encryption-key-id': 'my-cmk', + ({'server-side-encryption-key-id': ' my-cmk ', 'server-side-data-encryption': 'sm4', 'server-side-encryption-algorithm': 'AES256'}, {'server-side-encryption': 'KMS', 'server-side-encryption-key-id': 'my-cmk', 'server-side-data-encryption': 'SM4'}), @@ -311,11 +231,7 @@ def test_sse_headers_and_conditional_creation(oss_server, settings, expected): @pytest.mark.parametrize('settings', [ - {'server-side-encryption': 'AES-256'}, {'server-side-encryption': 'AES256', 'server-side-encryption-key-id': 'my-cmk'}, - {'server-side-encryption': 'SM4', 'server-side-data-encryption': 'SM4'}, - {'server-side-encryption': 'KMS', 'server-side-data-encryption': 'AES256'}, - {'server-side-encryption-key-id': 'my cmk'}, {'server-side-encryption': '', 'server-side-encryption-algorithm': 'AES256'}, ]) def test_invalid_sse_is_rejected_before_io(oss_server, settings): diff --git a/paimon-python/pypaimon/tests/oss_file_io_test.py b/paimon-python/pypaimon/tests/oss_file_io_test.py index e14124ae369b..bb8506d0e7d9 100644 --- a/paimon-python/pypaimon/tests/oss_file_io_test.py +++ b/paimon-python/pypaimon/tests/oss_file_io_test.py @@ -50,19 +50,19 @@ def setUp(self): if not endpoint: self.skipTest("test endpoint is not configured") return - + self.root_path = f"oss://{self.bucket}/" - + self.catalog_options = Options({ OssOptions.OSS_ACCESS_KEY_ID.key(): access_key_id, OssOptions.OSS_ACCESS_KEY_SECRET.key(): access_key_secret, OssOptions.OSS_ENDPOINT.key(): endpoint, OssOptions.OSS_IMPL.key(): oss_impl, }) - + # Create OssFileIO instance self.file_io = OssFileIO(self.root_path, self.catalog_options) - + # Create unique test prefix to avoid conflicts self.test_prefix = f"test-{uuid.uuid4().hex[:8]}/" @@ -96,41 +96,41 @@ def test_new_input_stream_read(self): # Create test data test_data = b"Hello, World! This is a test file for OSS input stream." test_file = self._get_test_path("test-input-stream.txt") - + # Write test data to file with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(test_data) - + # Test new_input_stream input_stream = self.file_io.new_input_stream(test_file) self.assertIsNotNone(input_stream) - + # Test read without nbytes (read all) input_stream.seek(0) read_data = input_stream.read() self.assertEqual(read_data, test_data) - + # Test read with nbytes input_stream.seek(0) read_partial = input_stream.read(5) self.assertEqual(read_partial, b"Hello") - + # Test read more bytes read_partial2 = input_stream.read(7) self.assertEqual(read_partial2, b", World") - + # Test read remaining read_remaining = input_stream.read() self.assertEqual(read_remaining, b"! This is a test file for OSS input stream.") - + # Verify complete data input_stream.seek(0) complete_data = input_stream.read() self.assertEqual(complete_data, test_data) - + # Close the stream input_stream.close() - + # Test context manager with self.file_io.new_input_stream(test_file) as input_stream2: data = input_stream2.read() @@ -141,11 +141,11 @@ def test_new_input_stream_read_large_file(self): # Create larger test data (1MB) test_data = b"X" * (1024 * 1024) test_file = self._get_test_path("test-large-input-stream.bin") - + # Write test data with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(test_data) - + # Test reading in chunks chunk_size = 64 * 1024 # 64KB chunks with self.file_io.new_input_stream(test_file) as input_stream: @@ -155,12 +155,12 @@ def test_new_input_stream_read_large_file(self): if not chunk: break read_chunks.append(chunk) - + # Verify all data was read read_data = b''.join(read_chunks) self.assertEqual(len(read_data), len(test_data)) self.assertEqual(read_data, test_data) - + # Test read_at method if available with self.file_io.new_input_stream(test_file) as input_stream: if hasattr(input_stream, 'read_at'): @@ -173,7 +173,7 @@ def test_new_input_stream_read_large_file(self): def test_new_input_stream_file_not_found(self): """Test new_input_stream with non-existent file.""" non_existent_file = self._get_test_path("non-existent-file.txt") - + with self.assertRaises(FileNotFoundError): self.file_io.new_input_stream(non_existent_file) @@ -210,7 +210,7 @@ def test_write_file_with_overwrite_flag(self): def test_exists_does_not_catch_exception(self): """Test that exists does not catch exceptions.""" test_file = self._get_test_path("test_file.txt") - + # Write a test file with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(b"test") @@ -242,7 +242,7 @@ def test_delete_returns_false_when_file_not_exists(self): def test_mkdirs_raises_error_when_path_is_file(self): """Test that mkdirs raises error when path is a file.""" test_file = self._get_test_path("test_file.txt") - + # Create a file with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(b"test") @@ -255,7 +255,7 @@ def test_rename_returns_false_when_dst_exists(self): """Test that rename returns False when destination exists.""" src_file = self._get_test_path("src.txt") dst_file = self._get_test_path("dst.txt") - + # Create source and destination files with self.file_io.new_output_stream(src_file) as out_stream: out_stream.write(b"src") @@ -274,7 +274,7 @@ def test_get_file_status_raises_error_when_file_not_exists(self): test_file = self._get_test_path("test_file.txt") with self.file_io.new_output_stream(test_file) as out_stream: out_stream.write(b"test content") - + file_info = self.file_io.get_file_status(test_file) self.assertEqual(file_info.type, pafs.FileType.File) self.assertIsNotNone(file_info.size) @@ -291,30 +291,30 @@ def test_copy_file(self): """Test copy_file method.""" source_file = self._get_test_path("source.txt") target_file = self._get_test_path("target.txt") - + # Create source file with self.file_io.new_output_stream(source_file) as out_stream: out_stream.write(b"source content") - + # Test 1: Raises FileExistsError when target exists and overwrite=False with self.file_io.new_output_stream(target_file) as out_stream: out_stream.write(b"target content") - + with self.assertRaises(FileExistsError) as context: self.file_io.copy_file(source_file, target_file, overwrite=False) self.assertIn("already exists", str(context.exception)) - + # Verify target content unchanged with self.file_io.new_input_stream(target_file) as in_stream: content = in_stream.read() self.assertEqual(content, b"target content") - + # Test 2: Overwrites when overwrite=True self.file_io.copy_file(source_file, target_file, overwrite=True) with self.file_io.new_input_stream(target_file) as in_stream: content = in_stream.read() self.assertEqual(content, b"source content") - + # Test 3: Creates parent directory if it doesn't exist target_file_in_subdir = self._get_test_path("subdir/target.txt") self.file_io.copy_file(source_file, target_file_in_subdir, overwrite=False) @@ -327,33 +327,33 @@ def test_try_to_write_atomic(self): """Test try_to_write_atomic method.""" target_dir = self._get_test_path("target_dir/") normal_file = self._get_test_path("normal_file.txt") - + # Create target directory self.file_io.mkdirs(target_dir) self.assertFalse( self.file_io.try_to_write_atomic(target_dir, "test content"), "OssFileIO should return False when target is a directory") - + # Verify no file was created inside the directory # List directory contents to verify it's empty selector = pafs.FileSelector(self.file_io.to_filesystem_path(target_dir), recursive=False, allow_not_found=True) dir_contents = self.file_io.filesystem.get_file_info(selector) self.assertEqual(len(dir_contents), 0, "No file should be created inside the directory") - + self.assertTrue(self.file_io.try_to_write_atomic(normal_file, "test content")) content = self.file_io.read_file_utf8(normal_file) self.assertEqual(content, "test content") - + # Delete and test again self.file_io.delete(normal_file) self.assertFalse( self.file_io.try_to_write_atomic(target_dir, "test content"), "OssFileIO should return False when target is a directory") - + # Verify no file was created inside the directory dir_contents = self.file_io.filesystem.get_file_info(selector) self.assertEqual(len(dir_contents), 0, "No file should be created inside the directory") - + self.assertTrue(self.file_io.try_to_write_atomic(normal_file, "test content")) content = self.file_io.read_file_utf8(normal_file) self.assertEqual(content, "test content") diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py deleted file mode 100644 index d7cc0f0c906b..000000000000 --- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_cache_test.py +++ /dev/null @@ -1,154 +0,0 @@ -# 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. - -"""Local lifecycle tests for REST FileIO caching; only the remote token API is stubbed.""" - -import pickle -import subprocess -import sys -import threading -from concurrent.futures import ThreadPoolExecutor -from types import SimpleNamespace -from unittest import mock - -import pytest -from cachetools import TTLCache - -from pypaimon.api.rest_api import RESTApi -from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO -from pypaimon.common.file_io import FileIO -from pypaimon.common.identifier import Identifier -from pypaimon.common.options import Options - - -@pytest.fixture -def options(): - with mock.patch.object(RESTTokenFileIO, '_TOKEN_CACHE', {}), \ - mock.patch.object(RESTTokenFileIO, '_TOKEN_LOCKS', {}): - yield Options({'uri': 'http://127.0.0.1:1', 'token.provider': 'bear', 'token': 'test-user'}) - - -def new_io(path, options): - return RESTTokenFileIO(Identifier.from_string('db.table'), str(path), options) - - -def token_response(now, value): - return SimpleNamespace(token={'test.credential': value}, expires_at_millis=int(now * 1000) + 7_200_000) - - -def test_concurrent_expiry_refresh_keeps_instance_backends_isolated(tmp_path, options): - clock = [1_000_000.0] - instances = [new_io(tmp_path / str(index), options) for index in range(2)] - previous = [None, None] - with mock.patch('pypaimon.catalog.rest.rest_token_file_io.time.time', side_effect=lambda: clock[0]), \ - mock.patch.object(RESTApi, 'load_table_token') as load: - for generation in range(3): - load.return_value = token_response(clock[0], str(generation)) - barrier = threading.Barrier(16) - - def write(index): - barrier.wait(timeout=10) - io = instances[index % 2] - backend = io.file_io() - path = io.path + '/generation-{}-writer-{}'.format(generation, index) - backend.write_file(path, str(generation)) - return backend, path - - with ThreadPoolExecutor(max_workers=16) as pool: - results = list(pool.map(write, range(16))) - assert load.call_count == generation + 1 - for index, io in enumerate(instances): - backend = results[index][0] - assert backend is not previous[index] - assert all(result[0] is backend for result in results[index::2]) - assert backend.path == io.path - assert backend.properties.to_map()['test.credential'] == str(generation) - assert all(io.read_file_utf8(path) == str(generation) for _, path in results[index::2]) - previous[index] = backend - assert previous[0] is not previous[1] - clock[0] += 7_200 - - -def test_refresh_and_initialization_failures_do_not_poison_cache(tmp_path, options): - io = new_io(tmp_path, options) - clock = [1_000_000.0] - responses = [token_response(clock[0], 'old'), OSError('token unavailable'), - token_response(clock[0] + 7_200, 'new')] - with mock.patch('pypaimon.catalog.rest.rest_token_file_io.time.time', side_effect=lambda: clock[0]), \ - mock.patch.object(RESTApi, 'load_table_token', side_effect=responses) as load: - old = io.file_io() - clock[0] += 7_200 - with pytest.raises(OSError, match='token unavailable'): - io.file_io() - with mock.patch.object(FileIO, 'get', side_effect=OSError('backend unavailable')): - with pytest.raises(OSError, match='backend unavailable'): - io.file_io() - backend = io.file_io() - assert backend is not old - assert io.file_io() is backend - assert load.call_count == 3 - path = str(tmp_path / 'recovered') - io.write_file(path, 'ok') - assert io.read_file_utf8(path) == 'ok' - assert backend.properties.to_map()['test.credential'] == 'new' - - -def test_cache_ttl_rebuilds_backend_without_refetching_valid_token(tmp_path, options): - clock = [0.0] - with mock.patch('pypaimon.catalog.rest.rest_token_file_io.TTLCache', - side_effect=lambda **kwargs: TTLCache(timer=lambda: clock[0], **kwargs)), \ - mock.patch.object(RESTTokenFileIO, '_FILE_IO_CACHE_TTL', 10), \ - mock.patch('pypaimon.catalog.rest.rest_token_file_io.time.time', return_value=1_000_000), \ - mock.patch.object(RESTApi, 'load_table_token', return_value=token_response(1_000_000, 'valid')) as load: - io = new_io(tmp_path, options) - first = io.file_io() - path = str(tmp_path / 'before-expiry') - io.write_file(path, 'preserved') - clock[0] = 9 - assert io.file_io() is first - clock[0] = 11 - assert io.file_io() is not first - assert io.read_file_utf8(path) == 'preserved' - assert load.call_count == 1 - - -def test_initialized_file_io_can_be_used_in_fresh_processes(tmp_path, options): - import time - - io = new_io(tmp_path, options) - with mock.patch.object(RESTApi, 'load_table_token', return_value=token_response(time.time(), 'valid')): - io.write_file(str(tmp_path / 'parent'), 'parent data') - payload = pickle.dumps(io) - code = ''' -import pickle, sys -io = pickle.loads(sys.stdin.buffer.read()) -assert io.read_file_utf8(io.path + '/parent') == 'parent data' -backend = io.file_io() -assert io.file_io() is backend -assert backend.properties.to_map()['test.credential'] == 'valid' -io.write_file(io.path + '/child-' + sys.argv[1], 'child data') -''' - - def run_child(index): - result = subprocess.run([sys.executable, '-c', code, str(index)], input=payload, - stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30) - assert result.returncode == 0, result.stderr.decode() - - with ThreadPoolExecutor(max_workers=4) as pool: - list(pool.map(run_child, range(4))) - for index in range(4): - assert io.read_file_utf8(str(tmp_path / ('child-' + str(index)))) == 'child data' diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py index ed0e20b848e3..cf80fc95d1dd 100644 --- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py @@ -18,10 +18,8 @@ import os import pickle import tempfile -import threading import time import unittest -from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch, MagicMock from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO @@ -174,24 +172,6 @@ def test_pickle_serialization(self): with open(expected_path, 'rb') as f: self.assertEqual(f.read(), test_content) - def test_concurrent_file_io_reuse_and_token_refresh(self): - with patch.object(RESTTokenFileIO, 'try_to_refresh_token'): - file_io = RESTTokenFileIO(self.identifier, self.warehouse_path, self.catalog_options) - file_io.token = RESTToken({}, 1) - barrier = threading.Barrier(8) - - def get_backend(_): - barrier.wait(timeout=10) - return file_io.file_io() - - with ThreadPoolExecutor(max_workers=8) as pool: - backends = list(pool.map(get_backend, range(8))) - self.assertTrue(all(backend is backends[0] for backend in backends)) - file_io.token = RESTToken({}, 2) - refreshed = file_io.file_io() - self.assertIsNot(refreshed, backends[0]) - self.assertIs(file_io.file_io(), refreshed) - def test_dlf_oss_endpoint_overrides_token_endpoint(self): """Test that DLF OSS endpoint overrides the standard OSS endpoint in token.""" dlf_oss_endpoint = "https://dlf-custom-endpoint.oss-cn-hangzhou.aliyuncs.com" From a1fb4d60517d33dc1eea1075bcfe61be0a2428a3 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Fri, 11 Sep 2026 13:55:03 +0800 Subject: [PATCH 5/7] [python] Use OSS V4 signing for conditional metadata writes --- paimon-python/README.md | 6 ++ .../pypaimon/filesystem/oss_file_io.py | 14 ++++- .../pypaimon/tests/oss_atomic_write_test.py | 56 +++++++++++++++++++ 3 files changed, 73 insertions(+), 3 deletions(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index 6abc1cf056b3..03e58f5c1dde 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -24,6 +24,12 @@ 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-.aliyuncs.com` or `oss--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`: diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py b/paimon-python/pypaimon/filesystem/oss_file_io.py index 8b506343e00e..ecf04a300601 100644 --- a/paimon-python/pypaimon/filesystem/oss_file_io.py +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -62,14 +62,22 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: "and fs.oss.endpoint; pass fs.oss.securityToken for STS credentials.") if '://' not in endpoint: endpoint = 'https://' + endpoint + region = (self.properties.get(OssOptions.OSS_REGION) or '').strip() + if not region: + match = re.fullmatch( + r'oss-(?!accelerate(?:[.-]))([a-z0-9-]+?)(?:-internal)?\.aliyuncs\.com', + urlparse(endpoint).hostname or '') + region = match.group(1) if match else None + if not region: + raise ValueError("Set fs.oss.region for OSS V4 signing when the endpoint is not regional") headers = self._sse_headers() headers['x-oss-forbid-overwrite'] = 'true' - auth = oss2.StsAuth(access_key, secret_key, token) if token else oss2.Auth(access_key, secret_key) + auth = (oss2.StsAuth(access_key, secret_key, token, auth_version='v4') + if token else oss2.AuthV4(access_key, secret_key)) session = oss2.Session() try: - bucket = oss2.Bucket(auth, endpoint, self._oss_bucket, session=session, - region=self.properties.get(OssOptions.OSS_REGION)) + bucket = oss2.Bucket(auth, endpoint, self._oss_bucket, session=session, region=region) try: versioning = bucket.get_bucket_versioning().status except oss2.exceptions.ServerError as error: diff --git a/paimon-python/pypaimon/tests/oss_atomic_write_test.py b/paimon-python/pypaimon/tests/oss_atomic_write_test.py index b1a3fbc06f8d..3dcd4e2cfb49 100644 --- a/paimon-python/pypaimon/tests/oss_atomic_write_test.py +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -54,8 +54,19 @@ def respond(self, status, body): def error(self, status, code): self.respond(status, ('' + code + '').encode()) + def authenticate(self): + authorization = self.headers.get('Authorization', '') + server.auth_headers.append((self.command, authorization, self.headers.get('x-oss-security-token'))) + if not (authorization.startswith('OSS4-HMAC-SHA256 ') and + '/cn-hangzhou/oss/aliyun_v4_request' in authorization): + self.error(403, 'AccessDenied') + return False + return True + def do_GET(self): server.gets += 1 + if not self.authenticate(): + return assert urlsplit(self.path).query in ('versioning', 'versioning=') if server.fail_method == 'GET': self.error(*server.failure) @@ -65,6 +76,8 @@ def do_GET(self): '').encode()) def do_PUT(self): + if not self.authenticate(): + return key = unquote(urlsplit(self.path).path) data = self.rfile.read(int(self.headers['Content-Length'])) with server.lock: @@ -100,6 +113,7 @@ class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): server.gets = 0 server.token = None server.sse_headers = {} + server.auth_headers = [] thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.01), daemon=True) thread.start() yield server @@ -112,6 +126,7 @@ def options_for(server, token=None): return Options({ 'fs.oss.impl': 'legacy', 'fs.oss.endpoint': 'http://127.0.0.1:{}'.format(server.server_port), + 'fs.oss.region': 'cn-hangzhou', 'fs.oss.accessKeyId': 'test-ak', 'fs.oss.accessKeySecret': 'test-sk', 'fs.oss.securityToken': token, @@ -127,6 +142,47 @@ def file_io(server, resolving=False): return FileIO.get('oss://test-bucket/', options) +@pytest.mark.parametrize('token,endpoint,region', [ + (None, 'https://oss-cn-beijing.aliyuncs.com', 'cn-hangzhou'), + ('sts-token', None, 'cn-hangzhou'), + (None, 'https://oss-cn-hangzhou.aliyuncs.com', None), + ('sts-token', 'https://oss-cn-hangzhou-internal.aliyuncs.com', None), +]) +def test_v4_authentication_reaches_conditional_put(oss_server, monkeypatch, tmp_path, token, endpoint, region): + io = file_io(oss_server) + io._use_jindo = True + io.filesystem = pafs.SubTreeFileSystem(str(tmp_path), pafs.LocalFileSystem()) + local_endpoint = io.properties.to_map()['fs.oss.endpoint'] + io.properties = Options(dict(options_for(oss_server, token).to_map(), **{ + 'fs.oss.endpoint': endpoint or local_endpoint, + 'fs.oss.region': region, + 'fs.oss.signer.version': '4', + })) + send = oss2.Session.do_request + + def redirect(session, request, timeout): + # Sign the original endpoint, then send the real SDK request to the local server. + request.url = local_endpoint + urlsplit(request.url).path + return send(session, request, timeout) + + monkeypatch.setattr(oss2.Session, 'do_request', redirect) + path = 'oss://test-bucket/snapshot-1' + assert io.try_to_write_atomic(path, 'data') is True + assert io.try_to_write_atomic(path, 'overwrite') is False + assert list(oss_server.objects.values()) == [b'data'] + assert [method for method, _, _ in oss_server.auth_headers] == ['GET', 'PUT', 'GET', 'PUT'] + assert all(header_token == token for _, _, header_token in oss_server.auth_headers) + + +@pytest.mark.parametrize('endpoint', ['http://127.0.0.1', 'https://oss-accelerate.aliyuncs.com']) +def test_v4_requires_region_for_non_regional_endpoints(oss_server, endpoint): + io = file_io(oss_server) + io.properties = Options(dict(io.properties.to_map(), **{'fs.oss.endpoint': endpoint, 'fs.oss.region': None})) + with pytest.raises(ValueError, match='fs.oss.region'): + io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + assert oss_server.gets == oss_server.puts == 0 + + @pytest.mark.parametrize('resolving', [False, True]) def test_atomic_competition_and_existing_content(oss_server, resolving): io = file_io(oss_server, resolving) From 4121be1ccda005c463cf8f5a7bdd246a7b432a0a Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Fri, 11 Sep 2026 14:03:49 +0800 Subject: [PATCH 6/7] [python] Centralize OSS metadata client construction --- .../pypaimon/filesystem/oss_file_io.py | 54 ++++++++++--------- 1 file changed, 30 insertions(+), 24 deletions(-) diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py b/paimon-python/pypaimon/filesystem/oss_file_io.py index ecf04a300601..2346969d3626 100644 --- a/paimon-python/pypaimon/filesystem/oss_file_io.py +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -52,32 +52,11 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: "OSS atomic writes require oss2. Install pypaimon[oss] or pypaimon[jindo]." ) from error - access_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_ID) - secret_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_SECRET) - token = self.properties.get(OssOptions.OSS_SECURITY_TOKEN) - endpoint = self.properties.get(OssOptions.OSS_ENDPOINT) - if not access_key or not secret_key or not endpoint: - raise ValueError( - "OSS atomic writes require fs.oss.accessKeyId, fs.oss.accessKeySecret " - "and fs.oss.endpoint; pass fs.oss.securityToken for STS credentials.") - if '://' not in endpoint: - endpoint = 'https://' + endpoint - region = (self.properties.get(OssOptions.OSS_REGION) or '').strip() - if not region: - match = re.fullmatch( - r'oss-(?!accelerate(?:[.-]))([a-z0-9-]+?)(?:-internal)?\.aliyuncs\.com', - urlparse(endpoint).hostname or '') - region = match.group(1) if match else None - if not region: - raise ValueError("Set fs.oss.region for OSS V4 signing when the endpoint is not regional") - headers = self._sse_headers() - headers['x-oss-forbid-overwrite'] = 'true' - auth = (oss2.StsAuth(access_key, secret_key, token, auth_version='v4') - if token else oss2.AuthV4(access_key, secret_key)) - session = oss2.Session() try: - bucket = oss2.Bucket(auth, endpoint, self._oss_bucket, session=session, region=region) + bucket = self._create_oss_bucket(session) + headers = self._sse_headers() + headers['x-oss-forbid-overwrite'] = 'true' try: versioning = bucket.get_bucket_versioning().status except oss2.exceptions.ServerError as error: @@ -102,6 +81,33 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: finally: session.session.close() + def _create_oss_bucket(self, session): + """Build the metadata client with one V4 credential path for both AK and STS.""" + import oss2 + from oss2.credentials import StaticCredentialsProvider + + access_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_ID) + secret_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_SECRET) + token = self.properties.get(OssOptions.OSS_SECURITY_TOKEN) + endpoint = self.properties.get(OssOptions.OSS_ENDPOINT) + if not access_key or not secret_key or not endpoint: + raise ValueError( + "OSS atomic writes require fs.oss.accessKeyId, fs.oss.accessKeySecret " + "and fs.oss.endpoint; pass fs.oss.securityToken for STS credentials.") + if '://' not in endpoint: + endpoint = 'https://' + endpoint + region = (self.properties.get(OssOptions.OSS_REGION) or '').strip() + if not region: + match = re.fullmatch( + r'oss-(?!accelerate(?:[.-]))([a-z0-9-]+?)(?:-internal)?\.aliyuncs\.com', + urlparse(endpoint).hostname or '') + region = match.group(1) if match else None + if not region: + raise ValueError("Set fs.oss.region for OSS V4 signing when the endpoint is not regional") + provider = StaticCredentialsProvider(access_key, secret_key, token) + auth = oss2.ProviderAuthV4(provider) + return oss2.Bucket(auth, endpoint, self._oss_bucket, session=session, region=region) + def _sse_headers(self): """Match Java OSSFileIO's SSE resolution, including the native option fallback.""" settings = (OssOptions.OSS_SSE_METHOD, OssOptions.OSS_SSE_KMS_KEY_ID, From 62721920a16eea1881d1470ad5b6c26df94437f4 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Sat, 12 Sep 2026 10:26:27 +0800 Subject: [PATCH 7/7] [python] Cache OSS bucket publication mode per FileIO instance --- .github/workflows/paimon-python-checks.yml | 1 + paimon-python/README.md | 13 +++++-- paimon-python/dev/requirements-dev.txt | 1 + .../pypaimon/filesystem/oss_file_io.py | 38 +++++++++++++------ .../pypaimon/tests/oss_atomic_write_test.py | 14 +++++-- 5 files changed, 50 insertions(+), 17 deletions(-) diff --git a/.github/workflows/paimon-python-checks.yml b/.github/workflows/paimon-python-checks.yml index ac48f730fc75..a24546ca8f3b 100755 --- a/.github/workflows/paimon-python-checks.yml +++ b/.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 diff --git a/paimon-python/README.md b/paimon-python/README.md index 03e58f5c1dde..7cfd59d05893 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -66,9 +66,14 @@ 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**. -Each call first checks `GetBucketVersioning`. If versioning is Enabled/Suspended, +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 and uses the inherited PyArrow/Jindo temporary-file-and-rename path. +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 @@ -76,7 +81,9 @@ 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. The check and PUT cannot be made atomic with a bucket configuration change. +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 diff --git a/paimon-python/dev/requirements-dev.txt b/paimon-python/dev/requirements-dev.txt index a1b143383ca2..a122dd0bbeaa 100644 --- a/paimon-python/dev/requirements-dev.txt +++ b/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" diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py b/paimon-python/pypaimon/filesystem/oss_file_io.py index 2346969d3626..23b5859738b1 100644 --- a/paimon-python/pypaimon/filesystem/oss_file_io.py +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -27,6 +27,12 @@ class OssFileIO(PyArrowFileIO): """Override atomic metadata creation; inherit all ordinary file operations.""" + # Intentionally no lock: stable bucket configuration makes duplicate first + # queries acceptable. Without a lock, no additional __getstate__/__setstate__ + # hooks are needed; the parent's pickle support already handles this None/bool + # cache, preserving its value under the same stability assumption. + _atomic_write_supported = None + def try_to_write_atomic(self, path: str, content: str) -> bool: uri = urlparse(path) if uri.scheme: @@ -57,17 +63,7 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: bucket = self._create_oss_bucket(session) headers = self._sse_headers() headers['x-oss-forbid-overwrite'] = 'true' - try: - versioning = bucket.get_bucket_versioning().status - except oss2.exceptions.ServerError as error: - if error.status != 403 or error.code != 'AccessDenied': - raise - versioning = 'unknown (GetBucketVersioning denied)' - if versioning is not None: - self.logger.warning( - "Using legacy temporary-file-and-rename writes for OSS bucket %s " - "(versioning: %s). Concurrent commits are not protected against overwrites.", - self._oss_bucket, versioning) + if not self._supports_atomic_write(bucket): return super().try_to_write_atomic(path, content) try: bucket.put_object(key, content.encode('utf-8'), headers=headers) @@ -81,6 +77,26 @@ def try_to_write_atomic(self, path: str, content: str) -> bool: finally: session.session.close() + def _supports_atomic_write(self, bucket): + """Cache the publication mode for this instance; failed queries remain retryable.""" + import oss2 + + # Intentionally lock-free: duplicate first queries are acceptable with stable bucket configuration. + if self._atomic_write_supported is None: + try: + versioning = bucket.get_bucket_versioning().status + except oss2.exceptions.ServerError as error: + if error.status != 403 or error.code != 'AccessDenied': + raise + versioning = 'unknown (GetBucketVersioning denied)' + self._atomic_write_supported = versioning is None + if not self._atomic_write_supported: + self.logger.warning( + "Using legacy temporary-file-and-rename writes for OSS bucket %s " + "(versioning: %s). Concurrent commits are not protected against overwrites.", + self._oss_bucket, versioning) + return self._atomic_write_supported + def _create_oss_bucket(self, session): """Build the metadata client with one V4 credential path for both AK and STS.""" import oss2 diff --git a/paimon-python/pypaimon/tests/oss_atomic_write_test.py b/paimon-python/pypaimon/tests/oss_atomic_write_test.py index 3dcd4e2cfb49..e90995700758 100644 --- a/paimon-python/pypaimon/tests/oss_atomic_write_test.py +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -170,7 +170,7 @@ def redirect(session, request, timeout): assert io.try_to_write_atomic(path, 'data') is True assert io.try_to_write_atomic(path, 'overwrite') is False assert list(oss_server.objects.values()) == [b'data'] - assert [method for method, _, _ in oss_server.auth_headers] == ['GET', 'PUT', 'GET', 'PUT'] + assert [method for method, _, _ in oss_server.auth_headers] == ['GET', 'PUT', 'PUT'] assert all(header_token == token for _, _, header_token in oss_server.auth_headers) @@ -199,8 +199,10 @@ def write(index): results = list(pool.map(write, range(2))) assert sorted(results) == [False, True] assert oss_server.objects == {'/test-bucket/table/p=a%2Fb/snapshot-1': contents[results.index(True)].encode()} + versioning_queries = oss_server.gets assert io.try_to_write_atomic(path, 'overwrite') is False assert list(oss_server.objects.values()) == [contents[results.index(True)].encode()] + assert oss_server.gets == versioning_queries @pytest.mark.parametrize('second_path,method', [ @@ -224,6 +226,7 @@ def test_rest_file_io_isolates_bucket_and_encryption(oss_server, second_path, me io.token = RESTToken({'fs.oss.securityToken': 'refreshed-token'}, oss_server.server_port + 1) assert io.try_to_write_atomic(target + '-next', 'next') assert oss_server.token == 'refreshed-token' + assert oss_server.gets == 2 * (index + 1) @pytest.mark.parametrize('versioning', ['Enabled', 'Suspended', None]) @@ -242,7 +245,8 @@ def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, tmp assert (tmp_path / 'snapshot-1').read_text() == '兼容写入' assert sorted(p.name for p in tmp_path.iterdir()) == ['snapshot-1'] assert oss_server.puts == 0 - assert 'Concurrent commits are not protected' in caplog.text + assert oss_server.gets == 1 + assert caplog.text.count('Concurrent commits are not protected') == 1 @pytest.mark.parametrize('method,status,code', [ @@ -252,11 +256,15 @@ def test_versioning_fallback_preserves_legacy_writes(oss_server, versioning, tmp ]) def test_errors_are_not_competition(oss_server, method, status, code): oss_server.fail_method, oss_server.failure = method, (status, code) + io = file_io(oss_server) with pytest.raises(OSError) as caught: - file_io(oss_server).try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') assert isinstance(caught.value.__cause__, oss2.exceptions.ServerError) assert caught.value.__cause__.code == code assert oss_server.objects == {} + oss_server.fail_method = None + assert io.try_to_write_atomic('oss://test-bucket/snapshot-1', 'data') + assert oss_server.gets == (2 if method == 'GET' else 1) def test_lost_response_is_not_replayed_or_reported_as_conflict(oss_server):