This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 32f0a9261e88fac86d039741452f055289b52315 Author: zhigang <[email protected]> AuthorDate: Wed Sep 16 19:21:14 2026 +0800 [python] Use conditional OSS puts for atomic metadata writes (#9715) --- .github/workflows/ci-python.yml | 1 + paimon-python/README.md | 74 +++++ paimon-python/dev/requirements-dev.txt | 1 + .../pypaimon/catalog/rest/rest_token_file_io.py | 47 +--- paimon-python/pypaimon/common/file_io.py | 8 +- paimon-python/pypaimon/common/options/config.py | 12 + paimon-python/pypaimon/filesystem/oss_file_io.py | 157 +++++++++++ .../pypaimon/filesystem/resolving_file_io.py | 3 + .../rest_catalog_blob_as_descriptor_sample.py | 23 +- paimon-python/pypaimon/tests/file_io_test.py | 7 +- paimon-python/pypaimon/tests/lance_utils_test.py | 6 +- .../pypaimon/tests/oss_atomic_write_test.py | 308 +++++++++++++++++++++ paimon-python/pypaimon/tests/oss_file_io_test.py | 12 +- .../pypaimon/tests/oss_legacy_mode_test.py | 19 +- .../pypaimon/tests/py36/ao_simple_test.py | 8 +- .../pypaimon/tests/rest/rest_token_file_io_test.py | 3 + paimon-python/setup.py | 3 + 17 files changed, 619 insertions(+), 73 deletions(-) diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 637bcad9dc..f7e127f074 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -167,6 +167,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 2af01d988c..c745cdc261 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -15,6 +15,80 @@ Pypaimon requires Python 3.6+. The core dependencies are listed in `dev/requirements.txt`. The development dependencies are listed in `dev/requirements-dev.txt`. +# OSS metadata commits + +Install `pypaimon[oss]` (legacy PyArrow data access) or `pypaimon[jindo]` +(Jindo data access). Both include `oss2` for atomic metadata writes. +Configure `fs.oss.accessKeyId`, `fs.oss.accessKeySecret` and `fs.oss.endpoint`, +plus `fs.oss.securityToken` when using STS. An endpoint without a scheme uses +HTTPS for metadata writes. Credentials supplied only through an underlying +filesystem's credential provider must also be supplied through these options. + +Metadata requests always use OSS Signature V4 for both AK and STS credentials, +independently of the Jindo data-access signer setting. Set `fs.oss.region` to the +bucket's region ID, such as `cn-hangzhou`. If unset, it is inferred from standard +`oss-<region>.aliyuncs.com` or `oss-<region>-internal.aliyuncs.com` endpoints. +Other endpoints, including acceleration endpoints, require an explicit region. + +Atomic metadata PUTs also forward the OSS server-side encryption options, using +the same resolution as Java `OSSFileIO`: + +| Option | Behavior | +| --- | --- | +| `fs.oss.server-side-encryption` | `AES256`, `KMS` or `SM4` (case-insensitive) | +| `fs.oss.server-side-encryption-key-id` | KMS key ID; implies `KMS` if the method is unset | +| `fs.oss.server-side-data-encryption` | `SM4` with `KMS`; implies `KMS` if the method is unset | +| `fs.oss.server-side-encryption-algorithm` | Legacy method fallback, used only if all three options above are unset | + +The first three options reject blank values and invalid combinations before any +request is sent. If no encryption options are set, no encryption headers are +added and OSS applies the bucket's default policy. These settings cover the +atomic metadata PUT path; ordinary data writes still use the configured +PyArrow/Jindo filesystem and its encryption capabilities. + +For `oss://` paths, `FileIO.get` selects `OssFileIO`, a thin `PyArrowFileIO` +subclass that overrides atomic creation. Filesystem initialization, path handling, +and ordinary PyArrow/Jindo file operations are inherited unchanged. +Use `FileIO.get(path, options)` or construct `OssFileIO` explicitly for OSS atomic +writes. REST token refresh and `ResolvingFileIO` also route atomic writes through +this implementation. + +When the bucket is confirmed unversioned, `OssFileIO.try_to_write_atomic` uses a single OSS PUT +with `x-oss-forbid-overwrite=true`. Exactly one writer can create a given object; +`FileAlreadyExists` returns `False` so snapshot commits can retry. Other SDK errors +are raised as `OSError`, retaining their cause for diagnostics. A lost PUT response +is not retried by the SDK; the snapshot +commit loop checks the commit user and identifier before retrying. + +The vendor SDK is an optional backend dependency, imported only for OSS atomic +writes. The common FileIO API and Paimon table format remain independent of it. +This implementation uses OSS-specific conditional creation; it does not provide +the same atomic-write capability for every object store. + +Conditional creation requires a bucket that has **never enabled versioning**. +The first atomic write on each `OssFileIO` instance checks `GetBucketVersioning` +and caches the result, including the query-denied fallback. Concurrent first +writes may repeat the check and warning. Query errors other than `403 AccessDenied` +are not cached. +If versioning is Enabled/Suspended, +the state is unrecognized, or the query returns `403 AccessDenied`, the operation +logs a warning when caching the fallback and uses the inherited PyArrow/Jindo +temporary-file-and-rename path. +This preserves legacy writes without making version-query permission mandatory, +but the fallback does **not** guarantee safe concurrent commits. It also retains +the existing backend's encryption behavior rather than applying the conditional +PUT's OSS SSE headers. Invalid credentials, expired tokens, missing buckets, and +other query failures still propagate as errors. + +Grant `oss:GetBucketVersioning` and keep versioning disabled to use conditional +creation. Keep bucket versioning and version-query permissions unchanged for the +instance's lifetime; recreate the FileIO after changing them. A configuration +change is not guaranteed to produce an error and can invalidate the conditional-write guarantee. + +All concurrent writers must use conditional creation. Older Python clients or +other clients that overwrite snapshot objects can still overwrite a successful +commit. This change does not add conditional writes for other object stores. + # Build You can build the source package by executing the following command: diff --git a/paimon-python/dev/requirements-dev.txt b/paimon-python/dev/requirements-dev.txt index a1b143383c..a122dd0bbe 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/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index 42dabb268c..6a87d004eb 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 typing import Optional, Union from cachetools import TTLCache +from pypaimon.api.auth.bearer import BearTokenAuthProvider +from pypaimon.api.auth.dlf_provider import DLFAuthProvider from pypaimon.api.rest_api import RESTApi from pypaimon.api.rest_util import RESTUtil from pypaimon.catalog.rest.rest_token import RESTToken from pypaimon.common.file_io import FileIO -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO -from pypaimon.api.auth.bearer import BearTokenAuthProvider -from pypaimon.api.auth.dlf_provider import DLFAuthProvider from pypaimon.common.identifier import Identifier from pypaimon.common.options import Options from pypaimon.common.options.config import CatalogOptions, OssOptions @@ -40,27 +39,12 @@ class RESTTokenFileIO(FileIO): A FileIO to support getting token from REST Server. """ - _FILE_IO_CACHE_MAXSIZE = 1000 _FILE_IO_CACHE_TTL = 36000 # 10 hours in seconds - _FILE_IO_CACHE: TTLCache = None - _FILE_IO_CACHE_LOCK = threading.Lock() - _TOKEN_CACHE: dict = {} _TOKEN_LOCKS: dict = {} _TOKEN_LOCKS_LOCK = threading.Lock() - @classmethod - def _get_file_io_cache(cls) -> TTLCache: - if cls._FILE_IO_CACHE is None: - with cls._FILE_IO_CACHE_LOCK: - if cls._FILE_IO_CACHE is None: - cls._FILE_IO_CACHE = TTLCache( - maxsize=cls._FILE_IO_CACHE_MAXSIZE, - ttl=cls._FILE_IO_CACHE_TTL - ) - return cls._FILE_IO_CACHE - def __init__(self, identifier: Identifier, path: str, catalog_options: Optional[Union[dict, Options]] = None): self.identifier = identifier @@ -77,12 +61,20 @@ class RESTTokenFileIO(FileIO): self.api_instance: Optional[RESTApi] = None self.log = logging.getLogger(__name__) self._uri_reader_factory_cache: Optional[UriReaderFactory] = None + self._init_file_io_cache() + + def _init_file_io_cache(self): + # FileIO is bound to this instance's path and catalog options. + self._file_io_cache = TTLCache(maxsize=1, ttl=self._FILE_IO_CACHE_TTL) + self._file_io_cache_lock = threading.Lock() def __getstate__(self): state = self.__dict__.copy() # Remove non-serializable objects state.pop('api_instance', None) state.pop('_uri_reader_factory_cache', None) + state.pop('_file_io_cache', None) + state.pop('_file_io_cache_lock', None) # token can be serialized, but we'll refresh it on deserialization return state @@ -91,28 +83,17 @@ class RESTTokenFileIO(FileIO): self._uri_reader_factory_cache = None # api_instance will be recreated when needed self.api_instance = None + self._init_file_io_cache() def file_io(self) -> FileIO: - self.try_to_refresh_token() - - if self.token is None: - return FileIO.get(self.path, self.catalog_options or Options({})) - - cache_key = self.token - cache = self._get_file_io_cache() - - file_io = cache.get(cache_key) - if file_io is not None: - return file_io - - with self._FILE_IO_CACHE_LOCK: + with self._file_io_cache_lock: self.try_to_refresh_token() if self.token is None: return FileIO.get(self.path, self.catalog_options or Options({})) cache_key = self.token - cache = self._get_file_io_cache() + cache = self._file_io_cache file_io = cache.get(cache_key) if file_io is not None: return file_io @@ -127,7 +108,7 @@ class RESTTokenFileIO(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 10406849b3..720f9ae510 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -223,6 +223,7 @@ class FileIO(ABC): 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 @@ class FileIO(ABC): 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 @@ class FileIO(ABC): 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 5684ab62c3..98a466496b 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 0000000000..23b5859738 --- /dev/null +++ b/paimon-python/pypaimon/filesystem/oss_file_io.py @@ -0,0 +1,157 @@ +# 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.""" + + # 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: + 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] + path = 'oss://{}/{}'.format(self._oss_bucket, key) + 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 + + session = oss2.Session() + try: + bucket = self._create_oss_bucket(session) + headers = self._sse_headers() + headers['x-oss-forbid-overwrite'] = 'true' + 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) + 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 _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 + 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, + 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 576d160c81..e8f1c79d87 100644 --- a/paimon-python/pypaimon/filesystem/resolving_file_io.py +++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py @@ -97,6 +97,9 @@ class ResolvingFileIO(FileIO): 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 639b139d9f..1e90fa1bdf 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 @@ if __name__ == '__main__': '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 @@ if __name__ == '__main__': } 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 9803a53790..53c8e8d296 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.file_io import create_temp_path 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 @@ class FileIOTest(unittest.TestCase): 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 @@ class FileIOTest(unittest.TestCase): 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', @@ -525,7 +526,7 @@ class FileIOTest(unittest.TestCase): 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 9eb12566b5..2ffb220efa 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 @@ class LanceUtilsTest(unittest.TestCase): 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 @@ class LanceUtilsTest(unittest.TestCase): 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 0000000000..e909957007 --- /dev/null +++ b/paimon-python/pypaimon/tests/oss_atomic_write_test.py @@ -0,0 +1,308 @@ +# 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") + + [email protected] +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, ('<Error><Code>' + code + '</Code></Error>').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) + return + status = '' if server.versioning is None else '<Status>' + server.versioning + '</Status>' + self.respond(200, ('<VersioningConfiguration>' + status + + '</VersioningConfiguration>').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: + 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 = {} + server.auth_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.region': 'cn-hangzhou', + '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) + + [email protected]('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', 'PUT'] + assert all(header_token == token for _, _, header_token in oss_server.auth_headers) + + [email protected]('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 + + [email protected]('resolving', [False, True]) +def test_atomic_competition_and_existing_content(oss_server, resolving): + io = file_io(oss_server, resolving) + path = 'oss://test-bucket/table/p=a%2Fb/snapshot-1' + barrier = threading.Barrier(2) + + def write(index): + barrier.wait(timeout=10) + 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, 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 + + [email protected]('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 + 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) + + [email protected]('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) + # 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() == '兼容写入' + assert sorted(p.name for p in tmp_path.iterdir()) == ['snapshot-1'] + assert oss_server.puts == 0 + assert oss_server.gets == 1 + assert caplog.text.count('Concurrent commits are not protected') == 1 + + [email protected]('method,status,code', [ + ('GET', 403, 'SecurityTokenExpired'), + ('PUT', 403, 'AccessDenied'), + ('PUT', 409, 'OtherConflict'), +]) +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: + 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): + 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'] + + [email protected]('settings,expected', [ + ({'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'] + + [email protected]('settings', [ + {'server-side-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 d153760785..bb8506d0e7 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 @@ import pyarrow.fs as pafs 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.""" @@ -60,8 +60,8 @@ class OSSFileIOTest(unittest.TestCase): 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]}/" @@ -332,7 +332,7 @@ class OSSFileIOTest(unittest.TestCase): 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 @@ -348,7 +348,7 @@ class OSSFileIOTest(unittest.TestCase): 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) diff --git a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py index 43c4972f0c..dc300cb391 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 @@ import pyarrow.fs as pafs 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 @@ class OssLegacyModeTest(unittest.TestCase): 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 @@ class OssLegacyModeTest(unittest.TestCase): 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 792387179d..662d0b720e 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 @@ from pypaimon.catalog.catalog_exception import (DatabaseAlreadyExistException, 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 @@ class AOSimpleTest(RESTBaseTest): 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/pypaimon/tests/rest/rest_token_file_io_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py index dbb919eed2..cf80fc95d1 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 @@ -147,6 +147,8 @@ class RESTTokenFileIOTest(unittest.TestCase): 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 +159,7 @@ class RESTTokenFileIOTest(unittest.TestCase): 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" diff --git a/paimon-python/setup.py b/paimon-python/setup.py index 0a08d7acca..60cfe979d3 100644 --- a/paimon-python/setup.py +++ b/paimon-python/setup.py @@ -22,6 +22,7 @@ import subprocess 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 @@ setup( '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': [
