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 179abba9be [python] Fix OSS bucket-in-endpoint mode (PyArrow < 16)
corrupting directories via CreateBucket (#8710)
179abba9be is described below
commit 179abba9beb4b7efedb970c5d8c65dc43c5021bd
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Jul 23 21:08:20 2026 +0800
[python] Fix OSS bucket-in-endpoint mode (PyArrow < 16) corrupting
directories via CreateBucket (#8710)
---
.../pypaimon/filesystem/pyarrow_file_io.py | 78 ++++++-
paimon-python/pypaimon/tests/file_io_test.py | 27 ++-
.../pypaimon/tests/oss_legacy_mode_test.py | 252 +++++++++++++++++++++
3 files changed, 343 insertions(+), 14 deletions(-)
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 8b0b4a7446..b789a5f819 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -19,6 +19,7 @@ import logging
import os
import re
import subprocess
+import threading
import uuid
from datetime import datetime, timezone
from pathlib import PurePosixPath
@@ -61,6 +62,9 @@ class PyArrowFileIO(FileIO):
self._oss_bucket = None
_oss_impl = self.properties.get(OssOptions.OSS_IMPL)
self._use_jindo = False
+ self._legacy_bucket_checked = False
+ self._legacy_bucket_error = None
+ self._legacy_bucket_lock = threading.Lock()
if self._is_oss:
self._oss_bucket = self._extract_oss_bucket(path)
@@ -87,6 +91,16 @@ class PyArrowFileIO(FileIO):
else:
raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")
+ def __getstate__(self):
+ state = self.__dict__.copy()
+ # threading.Lock cannot be pickled; recreated in __setstate__.
+ state.pop("_legacy_bucket_lock", None)
+ return state
+
+ def __setstate__(self, state):
+ self.__dict__.update(state)
+ self._legacy_bucket_lock = threading.Lock()
+
@staticmethod
def parse_location(location: str):
uri = urlparse(location)
@@ -388,6 +402,11 @@ class PyArrowFileIO(FileIO):
return file_info
def list_status(self, path: str):
+ if self._legacy_oss_mode():
+ raise RuntimeError(
+ "Listing OSS directories is not supported with PyArrow < 16 "
+ "(it parses the first key segment as a bucket). Upgrade to "
+ "pyarrow >= 16, or install pyjindosdk and set
fs.oss.impl=jindo.")
path_str = self.to_filesystem_path(path)
selector = pafs.FileSelector(path_str, recursive=False,
allow_not_found=True)
return self.filesystem.get_file_info(selector)
@@ -396,7 +415,16 @@ class PyArrowFileIO(FileIO):
file_infos = self.list_status(path)
return [info for info in file_infos if info.type ==
pafs.FileType.Directory]
+ def _legacy_oss_mode(self) -> bool:
+ """OSS in bucket-in-endpoint mode (PyArrow < 16): paths are key-only
+ (the bucket is embedded in endpoint_override), but PyArrow still
+ parses the first key segment as a bucket, so bucket-level operations
+ target the wrong bucket."""
+ return self._is_oss and self._oss_bucket_in_endpoint and not
self._use_jindo
+
def exists(self, path: str) -> bool:
+ # Legacy OSS mode limitation: directories always report NotFound
+ # (see _legacy_oss_mode); plain objects are probed correctly.
path_str = self.to_filesystem_path(path)
return self._get_file_info(path_str).type != pafs.FileType.NotFound
@@ -438,17 +466,59 @@ class PyArrowFileIO(FileIO):
path_str = self.to_filesystem_path(path)
file_info = self._get_file_info(path_str)
- if file_info.type == pafs.FileType.NotFound:
- self.filesystem.create_dir(path_str, recursive=True)
- return True
if file_info.type == pafs.FileType.Directory:
return True
- elif file_info.type == pafs.FileType.File:
+ if file_info.type == pafs.FileType.File:
raise FileExistsError(f"Path exists but is not a directory:
{path}")
+ if self._legacy_oss_mode():
+ # create_dir would CreateBucket the first key segment and corrupt
+ # the parent directory; object stores need no directories. Only
+ # validate that the real bucket exists.
+ self._check_legacy_bucket_exists()
+ return True
+
self.filesystem.create_dir(path_str, recursive=True)
return True
+ def _check_legacy_bucket_exists(self):
+ """Raise if the real OSS bucket does not exist (legacy mode only).
+
+ PyArrow < 16 folds NoSuchBucket into the same NotFound as a missing
+ key, so probe the bucket root anonymously, at most once per
+ instance (the lock serializes concurrent writers). Reject only on
+ an OSS NoSuchBucket error body - a bare 404 may come from a proxy
+ or custom endpoint. Any probe setup or transport failure
+ (Requests-only TLS/proxy settings do not apply to PyArrow's S3
+ client) is indeterminate: fail open."""
+ with self._legacy_bucket_lock:
+ if not self._legacy_bucket_checked:
+ self._legacy_bucket_error = self._probe_legacy_bucket()
+ self._legacy_bucket_checked = True
+ if self._legacy_bucket_error:
+ raise OSError(self._legacy_bucket_error)
+
+ def _probe_legacy_bucket(self) -> Optional[str]:
+ """Return an error message if the bucket definitely does not exist."""
+ import requests
+
+ endpoint = self.properties.get(OssOptions.OSS_ENDPOINT) or ""
+ scheme, _, host = endpoint.rpartition("://")
+ url = f"{scheme or 'https'}://{self._oss_bucket}.{host}/"
+ try:
+ response = requests.get(url, timeout=5, allow_redirects=False,
stream=True)
+ try:
+ status = response.status_code
+ body = next(response.iter_content(2048), b"") if status == 404
else b""
+ finally:
+ response.close()
+ except (requests.RequestException, OSError):
+ return None
+ if status == 404 and b"NoSuchBucket" in body:
+ return (f"OSS bucket '{self._oss_bucket}' does not exist "
+ f"(NoSuchBucket from {url})")
+ return None
+
def rename(self, src: str, dst: str) -> bool:
dst_str = self.to_filesystem_path(dst)
dst_parent = PurePosixPath(dst_str).parent
diff --git a/paimon-python/pypaimon/tests/file_io_test.py
b/paimon-python/pypaimon/tests/file_io_test.py
index fd5f00306c..c7d2ebc6c7 100644
--- a/paimon-python/pypaimon/tests/file_io_test.py
+++ b/paimon-python/pypaimon/tests/file_io_test.py
@@ -97,14 +97,19 @@ class FileIOTest(unittest.TestCase):
mock_fs.create_dir = MagicMock()
mock_fs.open_output_stream.return_value = MagicMock()
oss_io.filesystem = mock_fs
- oss_io.new_output_stream("oss://test-bucket/path/to/file.txt")
- mock_fs.create_dir.assert_called_once()
- path_str =
oss_io.to_filesystem_path("oss://test-bucket/path/to/file.txt")
+ # Bucket-in-endpoint mode (PyArrow < 16) mkdirs probes the real bucket
+ # over HTTP; keep the test offline.
+ with patch("requests.get", return_value=MagicMock(status_code=403)):
+ oss_io.new_output_stream("oss://test-bucket/path/to/file.txt")
if bucket_stripped:
- expected_parent = '/'.join(path_str.split('/')[:-1]) if '/' in
path_str else ''
+ # Legacy mkdirs must not create_dir (it would CreateBucket the
+ # first key segment and corrupt the parent directory).
+ mock_fs.create_dir.assert_not_called()
else:
+ mock_fs.create_dir.assert_called_once()
+ path_str =
oss_io.to_filesystem_path("oss://test-bucket/path/to/file.txt")
expected_parent = "/".join(path_str.split("/")[:-1]) if "/" in
path_str else str(Path(path_str).parent)
- self.assertEqual(mock_fs.create_dir.call_args[0][0], expected_parent)
+ self.assertEqual(mock_fs.create_dir.call_args[0][0],
expected_parent)
if bucket_stripped:
for call_paths in get_file_info_calls:
for p in call_paths:
@@ -480,11 +485,13 @@ class FileIOTest(unittest.TestCase):
mock_fs.copy_file = MagicMock()
oss_io.filesystem = mock_fs
-
oss_io.new_output_stream("oss://test-bucket/db.db/tbl/bucket-0/data.parquet")
- oss_io.rename("oss://test-bucket/db.db/tbl/old.parquet",
- "oss://test-bucket/db.db/tbl/new.parquet")
- oss_io.copy_file("oss://test-bucket/db.db/tbl/src.parquet",
- "oss://test-bucket/db.db/tbl/dst.parquet")
+ # PyArrow < 16 mkdirs probes the real bucket over HTTP; keep the test
offline.
+ with patch("requests.get", return_value=MagicMock(status_code=403)):
+
oss_io.new_output_stream("oss://test-bucket/db.db/tbl/bucket-0/data.parquet")
+ oss_io.rename("oss://test-bucket/db.db/tbl/old.parquet",
+ "oss://test-bucket/db.db/tbl/new.parquet")
+ oss_io.copy_file("oss://test-bucket/db.db/tbl/src.parquet",
+ "oss://test-bucket/db.db/tbl/dst.parquet")
for call in mock_fs.create_dir.call_args_list:
self.assertNotIn("\\", call[0][0], f"backslash in path:
{call[0][0]}")
diff --git a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
new file mode 100644
index 0000000000..02f71f1af8
--- /dev/null
+++ b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
@@ -0,0 +1,252 @@
+# 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.
+
+"""Unit tests for the OSS bucket-in-endpoint mode (PyArrow < 16) of
PyArrowFileIO.
+
+See ``PyArrowFileIO._legacy_oss_mode`` for why bucket-level operations
+must be guarded in this mode. No real OSS access is required.
+"""
+
+import unittest
+from unittest import mock
+
+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
+
+
+TABLE_PATH = "oss://test-bucket/db-uuid.db/tbl-uuid"
+
+
+def _file_info(path, file_type):
+ return pafs.FileInfo(path, file_type)
+
+
+def _probe_response(status_code, body):
+ response = mock.MagicMock(status_code=status_code)
+ response.iter_content.return_value = iter([body])
+ return response
+
+
+class OssLegacyModeTest(unittest.TestCase):
+ """Behavior of PyArrowFileIO when OSS runs on PyArrow < 16."""
+
+ def _new_file_io(self, legacy):
+ options = Options({
+ OssOptions.OSS_ACCESS_KEY_ID.key(): "ak",
+ OssOptions.OSS_ACCESS_KEY_SECRET.key(): "sk",
+ OssOptions.OSS_ENDPOINT.key(): "oss-cn-test.example.com",
+ OssOptions.OSS_REGION.key(): "cn-test",
+ OssOptions.OSS_IMPL.key(): "legacy",
+ })
+ with mock.patch.object(
+ PyArrowFileIO, "_initialize_oss_fs", return_value=mock.Mock()):
+ file_io = PyArrowFileIO("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()
+ return file_io
+
+ def test_legacy_mkdirs_skips_create_dir(self):
+ """create_dir would CreateBucket and corrupt the parent directory."""
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ with mock.patch("requests.get") as get:
+ get.return_value = mock.MagicMock(status_code=403)
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ get.assert_called_once_with(
+ "https://test-bucket.oss-cn-test.example.com/";,
+ timeout=5, allow_redirects=False, stream=True)
+
+ file_io.filesystem.create_dir.assert_not_called()
+
+ def test_legacy_mkdirs_raises_when_bucket_missing(self):
+ """mkdirs must not report success for a missing real bucket."""
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ with mock.patch("requests.get") as get:
+ get.return_value = _probe_response(
+ 404, b"<Error><Code>NoSuchBucket</Code></Error>")
+ with self.assertRaises(OSError) as ctx:
+ file_io.mkdirs(TABLE_PATH)
+ self.assertIn("does not exist", str(ctx.exception))
+ file_io.filesystem.create_dir.assert_not_called()
+
+ def test_legacy_mkdirs_allows_bare_404_from_custom_endpoint(self):
+ """A 404 without the OSS NoSuchBucket body must not reject the
bucket."""
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ with mock.patch("requests.get") as get:
+ get.return_value = _probe_response(404, b"not found")
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ # The indeterminate probe is cached; no repeat per mkdirs.
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ self.assertEqual(get.call_count, 1)
+
+ def test_legacy_mkdirs_fails_open_and_caches_on_transport_error(self):
+ """A probe transport failure must neither fail mkdirs nor repeat
+ the probe (and its timeout) on every subsequent write."""
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ import requests
+ with mock.patch("requests.get") as get:
+ get.side_effect = requests.ConnectionError("boom")
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ self.assertEqual(get.call_count, 1)
+
+ def test_legacy_mkdirs_fails_open_on_plain_oserror(self):
+ """Requests-only setup errors (e.g. a broken REQUESTS_CA_BUNDLE)
+ raise plain OSError; they must not abort legacy writes."""
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ with mock.patch("requests.get") as get:
+ get.side_effect = OSError(
+ "Could not find a suitable TLS CA certificate bundle")
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+ self.assertEqual(get.call_count, 1)
+
+ def test_legacy_mkdirs_probe_is_serialized_across_threads(self):
+ """Concurrent writers must share one probe and all observe its
+ published result."""
+ import threading
+ import time
+
+ for body, expect_error in [
+ (b"<Error><Code>NoSuchBucket</Code></Error>", True),
+ (b"not found", False)]:
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ def slow_get(*args, **kwargs):
+ time.sleep(0.05)
+ return _probe_response(404, body)
+
+ results = []
+
+ def call_mkdirs():
+ try:
+ results.append(file_io.mkdirs(TABLE_PATH))
+ except OSError:
+ results.append("raised")
+
+ with mock.patch("requests.get", side_effect=slow_get) as get:
+ threads = [threading.Thread(target=call_mkdirs) for _ in
range(4)]
+ for t in threads:
+ t.start()
+ for t in threads:
+ t.join()
+ self.assertEqual(get.call_count, 1)
+ expected = "raised" if expect_error else True
+ self.assertEqual(results, [expected] * 4)
+
+ def test_legacy_mkdirs_missing_bucket_keeps_raising_without_reprobe(self):
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.NotFound)]
+
+ with mock.patch("requests.get") as get:
+ get.return_value = _probe_response(
+ 404, b"<Error><Code>NoSuchBucket</Code></Error>")
+ for _ in range(2):
+ with self.assertRaises(OSError):
+ file_io.mkdirs(TABLE_PATH)
+ self.assertEqual(get.call_count, 1)
+
+ def test_legacy_mkdirs_still_rejects_file_conflict(self):
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("db-uuid.db/tbl-uuid", pafs.FileType.File)]
+
+ with self.assertRaises(FileExistsError):
+ file_io.mkdirs(TABLE_PATH)
+ file_io.filesystem.create_dir.assert_not_called()
+
+ def test_modern_mkdirs_still_creates_dir(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("test-bucket/db-uuid.db/tbl-uuid",
pafs.FileType.NotFound)]
+
+ self.assertTrue(file_io.mkdirs(TABLE_PATH))
+
+ file_io.filesystem.create_dir.assert_called_once()
+
+ def test_file_io_pickle_roundtrip_recreates_lock(self):
+ """The probe lock must not break pickling (FileIO travels to Ray or
+ multiprocessing workers); probe state is carried over."""
+ import pickle
+
+ options = Options({
+ OssOptions.OSS_ACCESS_KEY_ID.key(): "ak",
+ OssOptions.OSS_ACCESS_KEY_SECRET.key(): "sk",
+ OssOptions.OSS_ENDPOINT.key(): "oss-cn-test.example.com",
+ OssOptions.OSS_REGION.key(): "cn-test",
+ OssOptions.OSS_IMPL.key(): "legacy",
+ })
+ file_io = PyArrowFileIO("oss://test-bucket/wh", options)
+ file_io._legacy_bucket_checked = True
+ file_io._legacy_bucket_error = "OSS bucket 'test-bucket' does not
exist"
+
+ restored = pickle.loads(pickle.dumps(file_io))
+
+ self.assertIsNotNone(restored._legacy_bucket_lock)
+ # Cached probe verdict survives; no re-probe in the worker.
+ with mock.patch("requests.get") as get:
+ with self.assertRaises(OSError):
+ restored._check_legacy_bucket_exists()
+ get.assert_not_called()
+
+ def test_legacy_exists_true_for_plain_object(self):
+ file_io = self._new_file_io(legacy=True)
+ file_io.filesystem.get_file_info.side_effect = lambda paths: [
+ _file_info(paths[0], pafs.FileType.File)]
+
+ self.assertTrue(file_io.exists(TABLE_PATH + "/data-1.parquet"))
+
+ def test_legacy_list_status_raises_actionable_error(self):
+ """Fail fast instead of the misleading raw NoSuchKey selector error."""
+ file_io = self._new_file_io(legacy=True)
+
+ with self.assertRaises(RuntimeError) as ctx:
+ file_io.list_status(TABLE_PATH)
+ self.assertIn("pyarrow >= 16", str(ctx.exception))
+ file_io.filesystem.get_file_info.assert_not_called()
+
+ def test_modern_list_status_uses_selector(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io.filesystem.get_file_info.return_value = []
+
+ self.assertEqual(file_io.list_status(TABLE_PATH), [])
+ file_io.filesystem.get_file_info.assert_called_once()
+
+
+if __name__ == "__main__":
+ unittest.main()
