This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 78523e9265 [python] Use conditional OSS puts for atomic metadata
writes (#9715)
78523e9265 is described below
commit 78523e9265185575618f8b7c8000485d8c9eeb9f
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 a706fe0661..267218a44d 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': [
