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 bfc84c8758 [python] Fix OSS directory deletion with PyArrow 22+
(#10456)
bfc84c8758 is described below
commit bfc84c8758dec6d686919f6cdc456b592a7fb187
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Oct 11 22:57:11 2026 +0800
[python] Fix OSS directory deletion with PyArrow 22+ (#10456)
---
.github/workflows/ci-python.yml | 26 +
.../pypaimon/filesystem/pyarrow_file_io.py | 219 ++++-
.../pypaimon/tests/s3_delete_compatibility_test.py | 913 +++++++++++++++++++++
paimon-python/pyproject.toml | 1 +
4 files changed, 1158 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 952a1173a5..9596257545 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -145,6 +145,32 @@ jobs:
chmod +x paimon-python/dev/lint-python.sh
./paimon-python/dev/lint-python.sh -e pytest_torch
+ s3-delete-compatibility:
+ name: S3 deletion / PyArrow ${{ matrix.pyarrow }}
+ runs-on: ubuntu-latest
+ timeout-minutes: 15
+ strategy:
+ fail-fast: false
+ matrix:
+ pyarrow: ['22.0.0', '23.0.0']
+ steps:
+ - uses: actions/checkout@v6
+ - uses: actions/setup-python@v5
+ with:
+ python-version: '3.11'
+ - name: Install test dependencies
+ run: |
+ python -m pip install --upgrade pip
+ python -m pip install -e ./paimon-python --group
'./paimon-python/pyproject.toml:test'
+ # Exercise the compatibility path before widening the package's
Arrow range.
+ python -m pip install --no-deps 'pyarrow==${{ matrix.pyarrow }}'
+ - name: Test S3 compatibility
+ working-directory: paimon-python
+ run: |
+ python -m pytest -q pypaimon/tests/s3_delete_compatibility_test.py \
+ pypaimon/tests/s3_client_compatibility_test.py
pypaimon/tests/oss_legacy_mode_test.py \
+ pypaimon/tests/file_io_test.py
pypaimon/tests/filesystem_catalog_test.py
+
rust-plan:
name: Native CI
timeout-minutes: 90
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 446bb64ed5..1a4362aedb 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -15,14 +15,20 @@
# specific language governing permissions and limitations
# under the License.
+import base64
+import hashlib
+import json
import logging
import os
import re
import subprocess
+import tempfile
import threading
+import time
from datetime import datetime, timezone
+from itertools import islice
from pathlib import PurePosixPath
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
from urllib.parse import splitport, urlparse
import pyarrow
@@ -46,6 +52,9 @@ def _pyarrow_lt_7():
return parse(pyarrow.__version__) < parse("7.0.0")
+_S3_DELETE_TIMEOUT_SECONDS = 3600
+
+
class LegacyOssDirectoryListingError(RuntimeError):
"""Raised when legacy PyArrow OSS cannot enumerate a directory."""
@@ -63,6 +72,7 @@ class PyArrowFileIO(FileIO):
self._get_s3_property("endpoint", S3Options.S3_ENDPOINT.key())
if self._is_s3 else None
)
+ self._s3_delete_client = None
self._oss_bucket = None
_oss_impl = self.properties.get(OssOptions.OSS_IMPL)
self._use_jindo = False
@@ -96,6 +106,7 @@ class PyArrowFileIO(FileIO):
raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")
def _set_pyarrow_version(self):
+ self._pyarrow_gte_22 = parse(pyarrow.__version__) >= parse("22.0.0")
self._pyarrow_gte_8 = parse(pyarrow.__version__) >= parse("8.0.0")
# force_virtual_addressing landed in PyArrow 16; below it the OSS
bucket
# goes into endpoint_override, so keys must omit it (init + path share
@@ -107,6 +118,9 @@ class PyArrowFileIO(FileIO):
return (not self._use_jindo
and (self._is_oss or bool(self._s3_endpoint)))
+ def _uses_s3_delete_fallback(self) -> bool:
+ return self._uses_s3_compatibility() and self._pyarrow_gte_22
+
def _configure_s3_checksums(self):
if parse(pyarrow.__version__) < parse("22.0.0"):
return
@@ -119,6 +133,7 @@ class PyArrowFileIO(FileIO):
# threading.Lock cannot be pickled; recreated in __setstate__.
state.pop("_legacy_bucket_lock", None)
state.pop("logger", None)
+ state.pop("_s3_delete_client", None)
# Recreate S3-compatible clients with the worker's AWS SDK settings.
if self._uses_s3_compatibility():
state.pop("filesystem", None)
@@ -126,6 +141,7 @@ class PyArrowFileIO(FileIO):
def __setstate__(self, state):
self.__dict__.update(state)
+ self._s3_delete_client = None
self.logger = logging.getLogger(__name__)
self._set_pyarrow_version()
if "_is_s3" not in state:
@@ -300,6 +316,25 @@ class PyArrowFileIO(FileIO):
return pafs.S3FileSystem(**client_kwargs)
+ def _s3_connection_options(self):
+ return {
+ "access_key": self._get_property(
+ S3Options.S3_ACCESS_KEY_ID.key(),
+ *self._s3_key_variants("access-key", "access.key")),
+ "secret_key": self._get_property(
+ S3Options.S3_ACCESS_KEY_SECRET.key(),
+ *self._s3_key_variants("secret-key", "secret.key")),
+ "session_token": self._get_property(
+ S3Options.S3_SECURITY_TOKEN.key(),
+ *self._s3_key_variants(
+ "session-token", "session.token",
+ "security-token", "security.token")),
+ "region": self._get_s3_property("region",
S3Options.S3_REGION.key()),
+ "path_style": (self._pyarrow_gte_16 and (
+ self._get_s3_boolean_property("path-style-access") or
+ self._get_s3_boolean_property("path.style.access"))),
+ }
+
def _initialize_hdfs_fs(self, scheme: str, netloc: Optional[str]) ->
FileSystem:
if 'HADOOP_HOME' not in os.environ:
raise RuntimeError("HADOOP_HOME environment variable is not set.")
@@ -486,13 +521,38 @@ class PyArrowFileIO(FileIO):
}
def delete(self, path: str, recursive: bool = False) -> bool:
+ use_fallback = self._uses_s3_delete_fallback()
+ if use_fallback and self._is_oss:
+ parsed = urlparse(path)
+ if parsed.scheme == "oss" and parsed.netloc:
+ bucket = self._extract_oss_bucket(path)
+ key = parsed.path.lstrip("/")
+ if "@" in parsed.netloc:
+ _, _, key = key.partition("/")
+ path = "oss://" + bucket + "/" + key
path_str = self.to_filesystem_path(path)
+ if use_fallback:
+ bucket, key = self._split_s3_path(path_str)
+ if not bucket or not key.strip("/"):
+ raise OSError(f"Refusing to delete bucket root: {path}")
file_info = self._get_file_info(path_str)
if file_info.type == pafs.FileType.NotFound:
return False
if file_info.type == pafs.FileType.Directory:
+ if use_fallback:
+ if recursive:
+ return self._delete_s3_compatible_directory(path_str)
+ selector = pafs.FileSelector(
+ path_str, recursive=False, allow_not_found=True)
+ if self.filesystem.get_file_info(selector):
+ raise OSError(f"Directory {path} is not empty")
+ bucket, key = self._split_s3_path(path_str)
+ if key:
+ self._get_s3_delete_client().delete_object(
+ Bucket=bucket, Key=key.rstrip("/") + "/")
+ return True
if not recursive:
selector = pafs.FileSelector(path_str, recursive=False,
allow_not_found=True)
dir_contents = self.filesystem.get_file_info(selector)
@@ -507,6 +567,163 @@ class PyArrowFileIO(FileIO):
self.filesystem.delete_file(path_str)
return True
+ def _delete_s3_compatible_directory(self, path_str: str) -> bool:
+ client = self._get_s3_delete_client()
+ bucket, key = self._split_s3_path(path_str)
+ prefix = key.rstrip("/")
+ if prefix:
+ prefix += "/"
+ deadline = time.monotonic() + _S3_DELETE_TIMEOUT_SECONDS
+ # Finish listing before deleting: mutating a paginated listing may
skip keys.
+ with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as listed:
+ for name in self._list_s3_keys(client, bucket, prefix, deadline,
path_str):
+ listed.write(json.dumps(name) + "\n")
+ listed.seek(0)
+ self._delete_s3_objects(
+ client, bucket, (json.loads(line) for line in listed),
+ deadline, path_str)
+ if any(self._list_s3_keys(client, bucket, prefix, deadline, path_str)):
+ raise OSError(f"S3 directory {path_str} changed during deletion")
+ return True
+
+ def _list_s3_keys(self, client, bucket: str, prefix: str,
+ deadline: float, path_str: str):
+ token = None
+ while True:
+ self._check_s3_delete_deadline(deadline, path_str)
+ params = {"Bucket": bucket, "Prefix": prefix, "MaxKeys": 1000}
+ if token is not None:
+ params["ContinuationToken"] = token
+ response = client.list_objects_v2(**params)
+ yield from self._listed_s3_keys(response, bucket, prefix)
+ if not response.get("IsTruncated"):
+ return
+ next_token = response.get("NextContinuationToken")
+ if not next_token or next_token == token:
+ raise OSError("S3 listing did not advance")
+ token = next_token
+
+ @staticmethod
+ def _listed_s3_keys(response, bucket: str, prefix: str):
+ if (response.get("Name", bucket) != bucket
+ or response.get("Prefix", prefix) != prefix):
+ raise OSError("S3 listing returned a different bucket or prefix")
+ keys = [item["Key"] for item in response.get("Contents", ())]
+ if any(not key.startswith(prefix) for key in keys):
+ raise OSError(f"S3 listing returned a key outside prefix {prefix}")
+ return keys
+
+ @staticmethod
+ def _check_s3_delete_deadline(deadline: float, path_str: str):
+ if time.monotonic() >= deadline:
+ raise TimeoutError(f"Timed out deleting S3 directory {path_str}")
+
+ @staticmethod
+ def _delete_s3_objects(
+ client, bucket: str, keys: Iterable[str],
+ deadline: float, path_str: str):
+ keys = iter(keys)
+ while True:
+ batch = list(islice(keys, 1000))
+ if not batch:
+ return
+ PyArrowFileIO._check_s3_delete_deadline(deadline, path_str)
+ ordinary = []
+ for key in batch:
+ if key.isprintable():
+ ordinary.append(key)
+ else:
+ # Non-printable keys may not round-trip through
DeleteObjects XML.
+ PyArrowFileIO._check_s3_delete_deadline(deadline, path_str)
+ client.delete_object(Bucket=bucket, Key=key)
+ if not ordinary:
+ continue
+ PyArrowFileIO._check_s3_delete_deadline(deadline, path_str)
+ response = client.delete_objects(
+ Bucket=bucket,
+ Delete={"Objects": [{"Key": key} for key in ordinary],
+ "Quiet": False})
+ deleted = [item["Key"] for item in response.get("Deleted", ())]
+ errors = response.get("Errors", ())
+ if errors:
+ examples = ", ".join(
+ "{}: {!r}".format(item.get("Code"), item.get("Key"))
+ for item in errors[:3])
+ raise OSError(
+ f"S3 batch delete incomplete for {path_str}: {examples}")
+ if set(deleted) != set(ordinary) or len(deleted) != len(ordinary):
+ raise OSError(f"S3 batch delete incomplete for {path_str}")
+
+ @staticmethod
+ def _split_s3_path(path_str: str):
+ bucket, _, key = path_str.partition("/")
+ return bucket, key
+
+ def _get_s3_delete_client(self):
+ if self._s3_delete_client is not None:
+ return self._s3_delete_client
+
+ import boto3
+ from botocore.config import Config
+
+ if self._is_oss:
+ endpoint = self.properties.get(OssOptions.OSS_ENDPOINT)
+ access_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_ID)
+ secret_key = self.properties.get(OssOptions.OSS_ACCESS_KEY_SECRET)
+ session_token = self.properties.get(OssOptions.OSS_SECURITY_TOKEN)
+ region = self.properties.get(OssOptions.OSS_REGION)
+ addressing_style = "virtual"
+ else:
+ endpoint = self._s3_endpoint
+ connection = self._s3_connection_options()
+ access_key = connection["access_key"]
+ secret_key = connection["secret_key"]
+ session_token = connection["session_token"]
+ region = connection["region"]
+ addressing_style = "path" if connection["path_style"] else
"virtual"
+
+ region = region or self.filesystem.region
+ if endpoint and "://" not in endpoint:
+ endpoint = "https://"; + endpoint
+ config_args = {
+ "retries": {"max_attempts": 10, "mode": "standard"},
+ "s3": {"addressing_style": addressing_style},
+ "connect_timeout": 60,
+ "read_timeout": 60,
+ }
+ try:
+ config = Config(request_checksum_calculation="when_required",
+ **config_args)
+ uses_new_checksums = True
+ except TypeError:
+ config = Config(**config_args)
+ uses_new_checksums = False
+
+ client = boto3.session.Session().client(
+ "s3",
+ endpoint_url=endpoint,
+ aws_access_key_id=access_key,
+ aws_secret_access_key=secret_key,
+ aws_session_token=session_token,
+ region_name=region,
+ config=config,
+ )
+ if uses_new_checksums:
+ # OSS requires Content-MD5; newer Botocore defaults to CRC32.
+ def use_content_md5(request, **kwargs):
+ request.headers["Content-MD5"] = base64.b64encode(
+ hashlib.md5(request.body, usedforsecurity=False).digest()
+ ).decode("ascii")
+ for name in list(request.headers):
+ if (name.lower().startswith("x-amz-checksum-") or
+ name.lower() == "x-amz-sdk-checksum-algorithm"):
+ del request.headers[name]
+
+ client.meta.events.register(
+ "before-sign.s3.DeleteObjects", use_content_md5)
+ self._s3_delete_client = client
+ return client
+
def mkdirs(self, path: str) -> bool:
path_str = self.to_filesystem_path(path)
file_info = self._get_file_info(path_str)
diff --git a/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
new file mode 100644
index 0000000000..f5dd6d08ec
--- /dev/null
+++ b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
@@ -0,0 +1,913 @@
+# 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 PyArrow-backed OSS and S3-compatible storage.
+
+No real OSS access is required.
+"""
+
+import base64
+import hashlib
+import os
+import socketserver
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from unittest import mock
+from urllib.parse import parse_qs, quote, unquote, urlsplit
+from xml.etree import ElementTree
+from xml.sax.saxutils import escape
+
+import pyarrow
+import pyarrow.fs as pafs
+from packaging.version import parse
+
+from pypaimon.common.options import Options
+from pypaimon.common.options.config import OssOptions, S3Options
+from pypaimon.filesystem.oss_file_io import OssFileIO
+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 _set_listed_keys(file_io, *passes):
+ file_io._s3_delete_client.list_objects_v2.side_effect = [
+ {"Contents": [{"Key": key} for key in keys]}
+ for keys in passes
+ ] + [{"Contents": []}]
+
+
+def _all_deleted_keys(client):
+ keys = []
+ for call in client.method_calls:
+ if call[0] == "delete_objects":
+ keys.extend(item["Key"] for item in call[2]["Delete"]["Objects"])
+ elif call[0] == "delete_object":
+ keys.append(call[2]["Key"])
+ return keys
+
+
+class _ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer):
+ daemon_threads = True
+
+
+class _DeleteRequestHandler(BaseHTTPRequestHandler):
+ protocol_version = "HTTP/1.1"
+
+ def _target(self):
+ path = unquote(urlsplit(self.path).path).lstrip("/")
+ bucket, _, key = path.partition("/")
+ return bucket, key
+
+ def _respond(self, status, body=b""):
+ self.send_response(status)
+ self.send_header("Content-Type", "application/xml")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ if self.command != "HEAD":
+ self.wfile.write(body)
+
+ def do_HEAD(self):
+ self.server.requests.append((self.command, self.path))
+ bucket, key = self._target()
+ objects = (self.server.bucket_objects.get(bucket, ())
+ if hasattr(self.server, "bucket_objects") else
self.server.objects)
+ exists = not key or key in objects
+ self._respond(200 if exists else 404)
+
+ def do_GET(self):
+ self.server.requests.append((self.command, self.path))
+ required_region = getattr(self.server, "required_region", None)
+ if required_region and "/{}/s3/aws4_request".format(required_region)
not in \
+ self.headers.get("Authorization", ""):
+ return self._respond(
+ 400,
b"<Error><Code>AuthorizationHeaderMalformed</Code></Error>")
+ query = parse_qs(urlsplit(self.path).query)
+ prefix = query.get("prefix", [""])[0]
+ delimiter = query.get("delimiter", [""])[0]
+ max_keys = int(query.get("max-keys", ["1000"])[0])
+ bucket, _ = self._target()
+ objects = (
+ self.server.bucket_objects.get(bucket, ())
+ if hasattr(self.server, "bucket_objects")
+ else self.server.objects
+ )
+ keys = sorted(
+ key for key in objects if key.startswith(prefix)
+ )
+ if max_keys == 1000 and hasattr(self.server, "forced_list_keys"):
+ keys = self.server.forced_list_keys
+ contents = []
+ common_prefixes = set()
+ for key in keys:
+ suffix = key[len(prefix):]
+ if delimiter and delimiter in suffix:
+ common_prefixes.add(
+ prefix + suffix.split(delimiter, 1)[0] + delimiter)
+ else:
+ contents.append(key)
+ contents = contents[:max_keys]
+ encode_keys = getattr(self.server, "encode_list_keys", False)
+ response_prefix = quote(prefix, safe="/") if encode_keys else prefix
+ response_delimiter = quote(delimiter, safe="/") if encode_keys else
delimiter
+ response_contents = (
+ [quote(key, safe="/") for key in contents]
+ if encode_keys else contents)
+ response_common_prefixes = (
+ [quote(key, safe="/") for key in sorted(common_prefixes)]
+ if encode_keys else sorted(common_prefixes))
+ body = (
+ '<?xml version="1.0" encoding="UTF-8"?>'
+ '<ListBucketResult
xmlns="http://s3.amazonaws.com/doc/2006-03-01/";>'
+ '<Name>{}</Name><Prefix>{}</Prefix><Delimiter>{}</Delimiter>'
+ '<KeyCount>{}</KeyCount><MaxKeys>{}</MaxKeys>{}'
+ '<IsTruncated>false</IsTruncated>{}{}</ListBucketResult>'.format(
+ escape(bucket), escape(response_prefix),
escape(response_delimiter),
+ len(contents) + len(common_prefixes), max_keys,
+ '<EncodingType>url</EncodingType>' if encode_keys else '',
+ "".join(
+ "<Contents><Key>{}</Key>"
+ "<LastModified>2026-01-01T00:00:00Z</LastModified>"
+ "<Size>{}</Size><StorageClass>STANDARD</StorageClass>"
+ "</Contents>".format(
+ escape(key), 0 if key.endswith("/") else 1)
+ for key in response_contents),
+ "".join(
+
"<CommonPrefixes><Prefix>{}</Prefix></CommonPrefixes>".format(
+ escape(child))
+ for child in response_common_prefixes)))
+ if (not hasattr(self.server, "bucket_objects")
+ and keys == [self.server.prefix]
+ and not self.server.late_object_added):
+ self.server.objects.add(self.server.prefix + "late.parquet")
+ self.server.late_object_added = True
+ first = getattr(self.server, "prefix", "") + "first.parquet"
+ if (not hasattr(self.server, "bucket_objects")
+ and first in keys and not self.server.missing_object_removed):
+ self.server.objects.discard(first)
+ self.server.missing_object_removed = True
+ encoded = body.encode("utf-8")
+ self._respond(200, encoded)
+
+ def do_DELETE(self):
+ self.server.requests.append((self.command, self.path))
+ bucket, key = self._target()
+ if hasattr(self.server, "bucket_objects"):
+ self.server.bucket_objects.setdefault(bucket, set()).discard(key)
+ else:
+ self.server.objects.discard(key)
+ if key == getattr(self.server, "inject_late_after_delete", None):
+ self.server.objects.add(self.server.prefix + "late.parquet")
+ self._respond(204)
+
+ def do_PUT(self):
+ self.server.requests.append((self.command, self.path))
+ if hasattr(self.server, "put_headers"):
+ self.server.put_headers.append((self.path, dict(self.headers)))
+ bucket, key = self._target()
+ if hasattr(self.server, "bucket_objects"):
+ self.server.bucket_objects.setdefault(bucket, set()).add(key)
+ else:
+ self.server.objects.add(key)
+ self._respond(200)
+
+ def _unexpected(self):
+ self.server.requests.append((self.command, self.path))
+ self.send_response(501)
+ self.send_header("Content-Length", "0")
+ self.end_headers()
+
+ def do_POST(self):
+ self.server.requests.append((self.command, self.path))
+ if "delete" not in urlsplit(self.path).query:
+ return self._respond(501)
+ body = self.rfile.read(int(self.headers["Content-Length"]))
+ if any(name.lower().startswith("x-amz-checksum-")
+ for name in self.headers):
+ return self._respond(
+ 400, b"<Error><Code>InvalidRequest</Code></Error>")
+ content_md5 = base64.b64encode(
+ hashlib.md5(body, usedforsecurity=False).digest()).decode()
+ if self.headers.get("Content-MD5") != content_md5:
+ return self._respond(
+ 400, b"<Error><Code>MissingArgument</Code></Error>")
+ if b"\x01" in body:
+ return self._respond(400,
b"<Error><Code>MalformedXML</Code></Error>")
+ bucket, _ = self._target()
+ keys = [item.text for item in ElementTree.fromstring(body).iter()
+ if item.tag.rsplit("}", 1)[-1] == "Key"]
+ objects = (self.server.bucket_objects.setdefault(bucket, set())
+ if hasattr(self.server, "bucket_objects") else
self.server.objects)
+ failed = getattr(self.server, "fail_delete_key_once", None)
+ if failed in keys:
+ self.server.fail_delete_key_once = None
+ deleted = []
+ for key in keys:
+ if key != failed:
+ objects.discard(key)
+ deleted.append(key)
+ if key == getattr(self.server, "inject_late_after_delete",
None):
+ objects.add(self.server.prefix + "late.parquet")
+ result = "<DeleteResult>{}{}</DeleteResult>".format(
+ "".join("<Deleted><Key>{}</Key></Deleted>".format(escape(key))
+ for key in deleted),
+ "<Error><Key>{}</Key><Code>AccessDenied</Code></Error>".format(
+ escape(failed)) if failed in keys else "")
+ self._respond(200, result.encode())
+
+ def log_message(self, *args):
+ pass
+
+
+def _successful_batch_delete(**kwargs):
+ return {"Deleted": kwargs["Delete"]["Objects"]}
+
+
+class OssDeleteCompatibilityTest(unittest.TestCase):
+ 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(
+ 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(spec=pafs.S3FileSystem)
+ file_io._s3_delete_client = mock.Mock()
+ file_io._s3_delete_client.delete_objects.side_effect = \
+ _successful_batch_delete
+ return file_io
+
+ def test_delete_normalizes_oss_uri_before_selecting_bucket(self):
+ for path in ("oss://other-bucket/table",
"oss://other-bucket.oss-cn-test.example.com/table",
+ "oss://AK:SK@endpoint/other-bucket/table"):
+ with self.subTest(path=path):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("other-bucket/table", pafs.FileType.Directory)]
+ _set_listed_keys(file_io, ["table/data"], [])
+ self.assertTrue(file_io.delete(path, recursive=True))
+
file_io.filesystem.get_file_info.assert_called_once_with(["other-bucket/table"])
+ client = file_io._s3_delete_client
+ self.assertTrue(all(call[1]["Bucket"] == "other-bucket"
+ for call in
client.delete_objects.call_args_list))
+ client.delete_object.assert_not_called()
+
+ def test_pyarrow_22_recursive_delete_batches_objects(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ data_dir = directory.rstrip("/") + "/data"
+ data_file = directory.rstrip("/") + "/data/data.parquet"
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ _set_listed_keys(
+ file_io, [data_dir.split("/", 1)[1] + "/", data_file.split("/",
1)[1]], [])
+
+ self.assertTrue(file_io.delete(TABLE_PATH, recursive=True))
+
+ client = file_io._s3_delete_client
+ self.assertCountEqual([
+ "db-uuid.db/tbl-uuid/data/data.parquet",
+ "db-uuid.db/tbl-uuid/data/",
+ ], [item["Key"] for item in
+ client.delete_objects.call_args[1]["Delete"]["Objects"]])
+ client.delete_object.assert_not_called()
+ client.put_object.assert_not_called()
+ file_io.filesystem.delete_file.assert_not_called()
+ file_io.filesystem.delete_dir_contents.assert_not_called()
+ file_io.filesystem.delete_dir.assert_not_called()
+
+ def test_recursive_delete_lists_all_pages_before_deleting(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ client = file_io._s3_delete_client
+
+ def list_page(**kwargs):
+ if kwargs.get("ContinuationToken") == "next":
+ self.assertEqual(0, client.delete_objects.call_count)
+ return {"Contents": [{"Key": "db-uuid.db/tbl-uuid/b"}]}
+ if client.delete_objects.call_count:
+ return {"Contents": []}
+ return {
+ "Contents": [{"Key": "db-uuid.db/tbl-uuid/a"}],
+ "IsTruncated": True,
+ "NextContinuationToken": "next",
+ }
+
+ client.list_objects_v2.side_effect = list_page
+ self.assertTrue(file_io.delete(TABLE_PATH, recursive=True))
+ self.assertCountEqual(
+ ["db-uuid.db/tbl-uuid/a", "db-uuid.db/tbl-uuid/b"],
+ _all_deleted_keys(client))
+ self.assertEqual("next", client.list_objects_v2.call_args_list[1][1][
+ "ContinuationToken"])
+
+ def test_recursive_delete_treats_schema_as_regular_key(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ prefix = "db-uuid.db/tbl-uuid/"
+ _set_listed_keys(file_io, [
+ prefix + "schema/schema-0",
+ prefix + "snapshot/snapshot-1",
+ prefix + "schema/schema-1",
+ prefix + "data/file.parquet",
+ ], [])
+
+ self.assertTrue(file_io.delete(TABLE_PATH, recursive=True))
+
+ calls = _all_deleted_keys(file_io._s3_delete_client)
+ self.assertCountEqual([
+ prefix + "schema/schema-0",
+ prefix + "snapshot/snapshot-1",
+ prefix + "schema/schema-1",
+ prefix + "data/file.parquet",
+ ], calls)
+
+ def test_recursive_delete_times_out_when_directory_keeps_changing(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ file_io._s3_delete_client.list_objects_v2.return_value = {
+ "Contents": [{"Key": "db-uuid.db/tbl-uuid/data.parquet"}],
+ "IsTruncated": True,
+ "NextContinuationToken": "next",
+ }
+ clock = mock.Mock()
+ clock.monotonic.side_effect = [0, 0, 2]
+
+ with mock.patch("pypaimon.filesystem.pyarrow_file_io.time", clock), \
+ mock.patch(
+
"pypaimon.filesystem.pyarrow_file_io._S3_DELETE_TIMEOUT_SECONDS", 1):
+ with self.assertRaisesRegex(TimeoutError, "deleting S3 directory"):
+ file_io.delete(TABLE_PATH, recursive=True)
+
+ file_io._s3_delete_client.list_objects_v2.assert_called_once()
+ file_io._s3_delete_client.delete_objects.assert_not_called()
+ file_io._s3_delete_client.delete_object.assert_not_called()
+ file_io._s3_delete_client.put_object.assert_not_called()
+
+ def test_recursive_delete_stops_before_next_batch_after_deadline(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ prefix = "db-uuid.db/tbl-uuid/"
+ keys = [prefix + "data/file-{}".format(i) for i in range(1001)]
+ schema_zero = prefix + "schema/schema-0"
+ client = file_io._s3_delete_client
+ client.list_objects_v2.return_value = {
+ "Contents": [{"Key": key} for key in keys + [schema_zero]]}
+ clock = mock.Mock()
+ clock.monotonic.side_effect = \
+ lambda: 2 if client.delete_objects.call_count >= 1 else 0
+
+ with mock.patch("pypaimon.filesystem.pyarrow_file_io.time", clock), \
+ mock.patch(
+
"pypaimon.filesystem.pyarrow_file_io._S3_DELETE_TIMEOUT_SECONDS", 1):
+ with self.assertRaisesRegex(TimeoutError, "deleting S3 directory"):
+ file_io.delete(TABLE_PATH, recursive=True)
+
+ self.assertEqual(1, client.delete_objects.call_count)
+ self.assertEqual(1000, len(
+ client.delete_objects.call_args[1]["Delete"]["Objects"]))
+ client.delete_object.assert_not_called()
+ client.put_object.assert_not_called()
+
+ def test_mixed_delete_stops_before_batch_after_deadline(self):
+ client = mock.Mock()
+ ordinary = "db-uuid.db/tbl-uuid/data/file.parquet"
+ special = "db-uuid.db/tbl-uuid/data/part\rfile.parquet"
+ client.delete_objects.return_value = {"Deleted": [{"Key": ordinary}]}
+ clock = mock.Mock()
+ clock.monotonic.side_effect = \
+ lambda: 2 if client.delete_object.called else 0
+
+ with mock.patch("pypaimon.filesystem.pyarrow_file_io.time", clock):
+ with self.assertRaisesRegex(TimeoutError, "deleting S3 directory"):
+ PyArrowFileIO._delete_s3_objects(
+ client, "test-bucket", [ordinary, special], 1,
+ TABLE_PATH)
+
+ client.delete_object.assert_called_once_with(
+ Bucket="test-bucket", Key=special)
+ client.delete_objects.assert_not_called()
+
+ def test_recursive_delete_batches_at_most_1000_keys(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ prefix = "db-uuid.db/tbl-uuid/"
+ keys = [prefix + "file-{}".format(i) for i in range(1001)]
+ _set_listed_keys(file_io, keys, [])
+
+ self.assertTrue(file_io.delete(TABLE_PATH, recursive=True))
+
+ calls = file_io._s3_delete_client.delete_objects.call_args_list
+ self.assertEqual([1000, 1], [
+ len(call[1]["Delete"]["Objects"]) for call in calls])
+ self.assertCountEqual(keys, [
+ item["Key"] for call in calls
+ for item in call[1]["Delete"]["Objects"]])
+
+ def test_pre_pyarrow_22_recursive_delete_keeps_native_batch(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = False
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ file_io.filesystem.get_file_info.side_effect = [
+ [_file_info(directory, pafs.FileType.Directory)],
+ ]
+
+ self.assertTrue(file_io.delete(TABLE_PATH, recursive=True))
+
+
file_io.filesystem.delete_dir_contents.assert_called_once_with(directory)
+ file_io.filesystem.delete_dir.assert_called_once_with(directory)
+ file_io.filesystem.delete_file.assert_not_called()
+
+ def test_delete_rejects_bucket_root(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = True
+ for recursive in (False, True):
+ with self.subTest(recursive=recursive):
+ with self.assertRaisesRegex(OSError, "bucket root"):
+ file_io.delete("oss://test-bucket/", recursive)
+
+ file_io.filesystem.get_file_info.assert_not_called()
+ file_io._s3_delete_client.delete_object.assert_not_called()
+
+ def test_old_pyarrow_delete_keeps_native_path(self):
+ file_io = self._new_file_io(legacy=False)
+ file_io._pyarrow_gte_22 = False
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("test-bucket/", pafs.FileType.NotFound)]
+
+ self.assertFalse(file_io.delete("oss://test-bucket/", recursive=True))
+ file_io.filesystem.get_file_info.assert_called_once()
+ file_io._s3_delete_client.delete_object.assert_not_called()
+
+ @unittest.skipUnless(
+ parse(pyarrow.__version__) >= parse("22.0.0"),
+ "requires PyArrow 22+ and boto3",
+ )
+ def test_delete_client_uses_pyarrow_region_with_fips_md5(self):
+ server = _ThreadingHTTPServer(
+ ("127.0.0.1", 0), _DeleteRequestHandler)
+ server.requests = []
+ server.bucket_objects = {
+ "test-bucket": {"table/", "table/data.parquet"}}
+ server.required_region = "eu-west-1"
+ server_thread = threading.Thread(target=server.serve_forever)
+ server_thread.start()
+ try:
+ options = Options({
+ S3Options.S3_ACCESS_KEY_ID.key(): "ak",
+ S3Options.S3_ACCESS_KEY_SECRET.key(): "sk",
+ S3Options.S3_ENDPOINT.key():
+ "http://127.0.0.1:{}".format(server.server_port),
+ "fs.s3.path.style.access": "true",
+ })
+ with mock.patch.dict(os.environ, {
+ "AWS_REGION": "eu-west-1",
+ "NO_PROXY": "127.0.0.1,localhost",
+ "no_proxy": "127.0.0.1,localhost",
+ }, clear=True):
+ file_io = PyArrowFileIO("s3://test-bucket/table", options)
+ self.assertEqual("eu-west-1", file_io.filesystem.region)
+ self.assertTrue(file_io.exists("s3://test-bucket/table"))
+ file_io._get_s3_delete_client()
+ real_md5 = hashlib.md5
+
+ def fips_md5(data=b"", *, usedforsecurity=True):
+ if usedforsecurity:
+ raise ValueError("md5 disabled for FIPS")
+ return real_md5(data, usedforsecurity=False)
+
+ with mock.patch.object(hashlib, "md5", side_effect=fips_md5):
+ self.assertTrue(file_io.delete(
+ "s3://test-bucket/table", recursive=True))
+ file_io._s3_delete_client.close()
+ self.assertEqual(set(), server.bucket_objects["test-bucket"])
+ self.assertTrue(any(method == "POST" for method, _ in
server.requests))
+ finally:
+ server.shutdown()
+ server.server_close()
+ server_thread.join()
+
+ def test_modern_non_recursive_delete_rejects_non_empty_directory(self):
+ file_io = self._new_file_io(legacy=False)
+ directory = file_io.to_filesystem_path(TABLE_PATH)
+ data_file = directory.rstrip("/") + "/data.parquet"
+ file_io.filesystem.get_file_info.side_effect = [
+ [_file_info(directory, pafs.FileType.Directory)],
+ [_file_info(data_file, pafs.FileType.File)],
+ ]
+
+ with self.assertRaisesRegex(OSError, "is not empty"):
+ file_io.delete(TABLE_PATH)
+
+ file_io.filesystem.delete_file.assert_not_called()
+
+
+class CustomS3EndpointTest(unittest.TestCase):
+ def _new_file_io(self, scheme="s3"):
+ options = Options({
+ S3Options.S3_ACCESS_KEY_ID.key(): "ak",
+ S3Options.S3_ACCESS_KEY_SECRET.key(): "sk",
+ S3Options.S3_ENDPOINT.key(): "http://minio:9000";,
+ S3Options.S3_REGION.key(): "us-east-1",
+ })
+ with mock.patch.object(
+ PyArrowFileIO, "_initialize_s3_fs", return_value=mock.Mock()):
+ file_io = PyArrowFileIO(
+ "{}://test-bucket/warehouse".format(scheme), options)
+ file_io.filesystem = mock.Mock(spec=pafs.S3FileSystem)
+ file_io._s3_delete_client = mock.Mock()
+ file_io._s3_delete_client.delete_objects.side_effect = \
+ _successful_batch_delete
+ return file_io
+
+ def test_pickle_discards_cached_delete_client(self):
+ file_io = self._new_file_io()
+ state = file_io.__getstate__()
+ self.assertNotIn("_s3_delete_client", state)
+ restored = object.__new__(PyArrowFileIO)
+ with mock.patch.object(PyArrowFileIO, "_initialize_s3_fs",
return_value=mock.Mock()):
+ restored.__setstate__(state)
+ self.assertIsNone(restored._s3_delete_client)
+
+ def test_jindo_and_native_s3_do_not_use_delete_fallback(self):
+ file_io = self._new_file_io()
+ file_io._pyarrow_gte_22 = True
+ file_io._use_jindo = True
+ self.assertFalse(file_io._uses_s3_delete_fallback())
+ file_io._use_jindo = False
+ file_io._s3_endpoint = None
+ self.assertFalse(file_io._uses_s3_delete_fallback())
+
+ def test_recursive_delete_rejects_mismatched_list_scope(self):
+ for scope in ({"Name": "other-bucket"},
+ {"Prefix": "table-other/"}):
+ file_io = self._new_file_io()
+ file_io._pyarrow_gte_22 = True
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("test-bucket/table", pafs.FileType.Directory)]
+ file_io._s3_delete_client.list_objects_v2.return_value = {
+ "Contents": [{"Key": "table/data.parquet"}], **scope}
+
+ with self.subTest(scope=scope):
+ with self.assertRaisesRegex(OSError, "different bucket or
prefix"):
+ file_io.delete("s3://test-bucket/table", recursive=True)
+
+ file_io._s3_delete_client.delete_object.assert_not_called()
+ file_io._s3_delete_client.put_object.assert_not_called()
+
+ def test_pyarrow_22_recursive_delete_uses_batch(self):
+ for scheme in ("s3", "s3a", "s3n"):
+ with self.subTest(scheme=scheme):
+ file_io = self._new_file_io(scheme)
+ file_io._pyarrow_gte_22 = True
+ path = "{}://test-bucket/table".format(scheme)
+ directory = file_io.to_filesystem_path(path)
+ data_file = directory + "/data.parquet"
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info(directory, pafs.FileType.Directory)]
+ _set_listed_keys(file_io, [data_file.split("/", 1)[1]], [])
+
+ self.assertTrue(file_io.delete(path, recursive=True))
+
+ file_io.filesystem.delete_file.assert_not_called()
+ file_io.filesystem.delete_dir_contents.assert_not_called()
+
file_io._s3_delete_client.delete_objects.assert_called_once_with(
+ Bucket="test-bucket", Delete={
+ "Objects": [{"Key": "table/data.parquet"}],
+ "Quiet": False})
+ file_io._s3_delete_client.delete_object.assert_not_called()
+
+ def test_pre_pyarrow_22_cross_bucket_delete_keeps_native_path(self):
+ file_io = self._new_file_io()
+ file_io._pyarrow_gte_22 = False
+ file_io.to_filesystem_path = mock.Mock(
+ return_value="target-bucket/table")
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("target-bucket/table", pafs.FileType.Directory)]
+
+ self.assertTrue(file_io.delete(
+ "s3://target-bucket/table", recursive=True))
+
+ file_io.filesystem.delete_dir_contents.assert_called_once_with(
+ "target-bucket/table")
+ file_io.filesystem.delete_dir.assert_called_once_with(
+ "target-bucket/table")
+ file_io._s3_delete_client.delete_object.assert_not_called()
+
+ @unittest.skipUnless(
+ parse(pyarrow.__version__) >= parse("22.0.0"),
+ "requires PyArrow 22+ and boto3",
+ )
+ def test_delete_uses_cross_bucket_list_status_path(self):
+ server = _ThreadingHTTPServer(
+ ("127.0.0.1", 0), _DeleteRequestHandler)
+ server.requests = []
+ server.put_headers = []
+ source_objects = {
+ "parent/child/", "parent/child/keep.parquet"}
+ server.bucket_objects = {
+ "source-bucket": set(source_objects),
+ "target-bucket": {
+ "parent/", "parent/child/", "parent/child/delete.parquet",
+ "parent-other/keep.parquet"},
+ }
+ server_thread = threading.Thread(target=server.serve_forever)
+ server_thread.start()
+ try:
+ options = Options({
+ S3Options.S3_ACCESS_KEY_ID.key(): "ak",
+ S3Options.S3_ACCESS_KEY_SECRET.key(): "sk",
+ S3Options.S3_ENDPOINT.key():
+ "http://127.0.0.1:{}".format(server.server_port),
+ S3Options.S3_REGION.key(): "us-east-1",
+ "fs.s3.path.style.access": "true",
+ })
+ with mock.patch.dict(os.environ, {
+ "AWS_REQUEST_CHECKSUM_CALCULATION": "WHEN_SUPPORTED",
+ "NO_PROXY": "127.0.0.1,localhost",
+ "no_proxy": "127.0.0.1,localhost",
+ }):
+ file_io = PyArrowFileIO(
+ "s3://source-bucket/warehouse", options)
+ for bucket in ("source-bucket", "target-bucket"):
+ with self.assertRaisesRegex(OSError, "bucket root"):
+ file_io.delete(
+ "s3://{}/".format(bucket), recursive=True)
+ self.assertFalse(any(
+ method in ("DELETE", "PUT", "POST")
+ for method, _ in server.requests))
+ before = {
+ bucket: set(keys)
+ for bucket, keys in server.bucket_objects.items()}
+ server.forced_list_keys = [
+ "parent/child/delete.parquet",
+ "parent-other/keep.parquet"]
+ with self.assertRaisesRegex(OSError, "outside prefix"):
+ file_io.delete(
+ "s3://target-bucket/parent/child", recursive=True)
+ self.assertEqual(before, server.bucket_objects)
+ self.assertFalse(any(
+ method in ("DELETE", "PUT", "POST")
+ for method, _ in server.requests))
+ del server.forced_list_keys
+ statuses = file_io.list_status(
+ "s3://target-bucket/parent")
+ target = next(
+ status for status in statuses
+ if status.type == pafs.FileType.Directory)
+
+ self.assertEqual(
+ "target-bucket/parent/child", target.path)
+ for path in (
+ target.path,
+ "s3://target-bucket/parent/child",
+ "s3:/target-bucket/parent/child",
+ "s3:target-bucket/parent/child"):
+ for recursive in (False, True):
+ with self.subTest(path=path, recursive=recursive):
+ server.bucket_objects["target-bucket"] = {
+ "parent/", "parent/child/"}
+ if recursive:
+ server.bucket_objects["target-bucket"].add(
+ "parent/child/delete.parquet")
+ self.assertTrue(file_io.exists(path))
+ self.assertTrue(file_io.delete(path, recursive))
+ self.assertEqual(
+ {"parent/"},
server.bucket_objects["target-bucket"])
+ self.assertEqual(
+ pafs.FileType.Directory,
+ file_io.filesystem.get_file_info(
+ "target-bucket/parent").type)
+ self.assertEqual(
+ source_objects,
+ server.bucket_objects["source-bucket"])
+ file_io._s3_delete_client.close()
+
+ self.assertEqual(
+ source_objects, server.bucket_objects["source-bucket"])
+ self.assertEqual({"parent/"},
server.bucket_objects["target-bucket"])
+ self.assertTrue(all(
+ urlsplit(path).path.startswith("/target-bucket/")
+ for method, path in server.requests
+ if method == "DELETE"))
+ self.assertEqual(
+ ["/target-bucket"] * 4,
+ [urlsplit(path).path for method, path in server.requests
+ if method == "POST"])
+ self.assertEqual([], server.put_headers)
+ finally:
+ server.shutdown()
+ server.server_close()
+ server_thread.join()
+
+ @unittest.skipUnless(
+ parse(pyarrow.__version__) >= parse("22.0.0"),
+ "requires PyArrow 22+ and boto3",
+ )
+ def test_recursive_delete_control_character_keys(self):
+ server = _ThreadingHTTPServer(
+ ("127.0.0.1", 0), _DeleteRequestHandler)
+ server.requests = []
+ server.prefix = "parent/table/"
+ server.encode_list_keys = True
+ carriage_return = server.prefix + "data/part\rfile.parquet"
+ control = server.prefix + "data/part\x01file.parquet"
+ server.objects = {
+ "parent/", server.prefix, carriage_return, control,
+ server.prefix + "schema/schema-0",
+ }
+ server.late_object_added = False
+ server.missing_object_removed = False
+ server_thread = threading.Thread(target=server.serve_forever)
+ server_thread.start()
+ try:
+ options = Options({
+ S3Options.S3_ACCESS_KEY_ID.key(): "ak",
+ S3Options.S3_ACCESS_KEY_SECRET.key(): "sk",
+ S3Options.S3_ENDPOINT.key():
+ "http://127.0.0.1:{}".format(server.server_port),
+ S3Options.S3_REGION.key(): "us-east-1",
+ "fs.s3.path.style.access": "true",
+ })
+ with mock.patch.object(
+ PyArrowFileIO, "_initialize_s3_fs",
return_value=mock.Mock()), \
+ mock.patch.dict(os.environ, {
+ "NO_PROXY": "127.0.0.1,localhost",
+ "no_proxy": "127.0.0.1,localhost",
+ }):
+ file_io = PyArrowFileIO("s3://test-bucket/parent/table",
options)
+ file_io.filesystem = mock.Mock(spec=pafs.S3FileSystem)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("test-bucket/parent/table",
+ pafs.FileType.Directory)]
+ self.assertTrue(file_io.delete(
+ "s3://test-bucket/parent/table", recursive=True))
+ file_io._s3_delete_client.close()
+
+ self.assertEqual({"parent/"}, server.objects)
+ individual = [unquote(urlsplit(path).path)
+ for method, path in server.requests
+ if method == "DELETE"]
+ self.assertIn("/test-bucket/" + carriage_return, individual)
+ self.assertIn("/test-bucket/" + control, individual)
+ self.assertEqual(1, sum(
+ method == "POST" for method, _ in server.requests))
+ finally:
+ server.shutdown()
+ server.server_close()
+ server_thread.join()
+
+ @unittest.skipUnless(
+ parse(pyarrow.__version__) >= parse("22.0.0"),
+ "requires PyArrow 22+ and boto3",
+ )
+ def test_recursive_delete_retries_partial_batch_failure(self):
+ server = _ThreadingHTTPServer(
+ ("127.0.0.1", 0), _DeleteRequestHandler)
+ server.requests = []
+ server.prefix = "parent/table/"
+ deleted = server.prefix + "data/deleted.parquet"
+ failed = server.prefix + "data/failed.parquet"
+ schema_zero = server.prefix + "schema/schema-0"
+ server.objects = {"parent/", server.prefix, deleted, failed,
schema_zero}
+ server.fail_delete_key_once = failed
+ server.late_object_added = False
+ server.missing_object_removed = False
+ server_thread = threading.Thread(target=server.serve_forever)
+ server_thread.start()
+ try:
+ options = Options({
+ S3Options.S3_ACCESS_KEY_ID.key(): "ak",
+ S3Options.S3_ACCESS_KEY_SECRET.key(): "sk",
+ S3Options.S3_ENDPOINT.key():
+ "http://127.0.0.1:{}".format(server.server_port),
+ S3Options.S3_REGION.key(): "us-east-1",
+ "fs.s3.path.style.access": "true",
+ })
+ with mock.patch.object(
+ PyArrowFileIO, "_initialize_s3_fs",
return_value=mock.Mock()), \
+ mock.patch.dict(os.environ, {
+ "NO_PROXY": "127.0.0.1,localhost",
+ "no_proxy": "127.0.0.1,localhost",
+ }):
+ file_io = PyArrowFileIO("s3://test-bucket/parent/table",
options)
+ file_io.filesystem = mock.Mock(spec=pafs.S3FileSystem)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("test-bucket/parent/table",
+ pafs.FileType.Directory)]
+ with self.assertRaisesRegex(OSError, "AccessDenied") as error:
+ file_io.delete(
+ "s3://test-bucket/parent/table", recursive=True)
+ self.assertIn(failed, str(error.exception))
+ self.assertEqual({"parent/", failed}, server.objects)
+ self.assertTrue(file_io.delete(
+ "s3://test-bucket/parent/table", recursive=True))
+ file_io._s3_delete_client.close()
+
+ self.assertEqual({"parent/"}, server.objects)
+ self.assertEqual(2, sum(
+ method == "POST" for method, _ in server.requests))
+ finally:
+ server.shutdown()
+ server.server_close()
+ server_thread.join()
+
+ @unittest.skipUnless(
+ parse(pyarrow.__version__) >= parse("22.0.0"),
+ "requires PyArrow 22+ and boto3",
+ )
+ def test_recursive_delete_reports_late_objects_during_races(self):
+ server = _ThreadingHTTPServer(
+ ("127.0.0.1", 0), _DeleteRequestHandler)
+ server.requests = []
+ server.prefix = "ta/ble/"
+ decoy = "ta/ble-other/keep.parquet"
+ server.objects = {
+ "ta/", server.prefix, server.prefix + "first.parquet", decoy}
+ server.inject_late_after_delete = server.prefix + "first.parquet"
+ server.late_object_added = False
+ server.missing_object_removed = False
+ server_thread = threading.Thread(target=server.serve_forever)
+ server_thread.start()
+ try:
+ options = Options({
+ S3Options.S3_ACCESS_KEY_ID.key(): "ak",
+ S3Options.S3_ACCESS_KEY_SECRET.key(): "sk",
+ S3Options.S3_ENDPOINT.key():
+ "http://127.0.0.1:{}".format(server.server_port),
+ S3Options.S3_REGION.key(): "us-east-1",
+ "fs.s3.path.style.access": "true",
+ })
+ with mock.patch.object(
+ PyArrowFileIO, "_initialize_s3_fs",
return_value=mock.Mock()), \
+ mock.patch.dict(os.environ, {
+ "NO_PROXY": "127.0.0.1,localhost",
+ "no_proxy": "127.0.0.1,localhost",
+ }):
+ file_io = PyArrowFileIO(
+ "s3://source-bucket/warehouse", options)
+ file_io.filesystem = mock.Mock(spec=pafs.S3FileSystem)
+ file_io.filesystem.get_file_info.return_value = [
+ _file_info("/ta/ble", pafs.FileType.Directory)]
+
+ with self.assertRaisesRegex(OSError, "changed during
deletion"):
+ file_io.delete("s3://target-bucket/ta//ble",
recursive=True)
+ file_io._s3_delete_client.close()
+
+ self.assertTrue(server.missing_object_removed)
+ self.assertEqual({decoy, "ta/", server.prefix + "late.parquet"},
+ server.objects)
+ self.assertEqual(
+ {"GET", "POST"},
+ {method for method, _ in server.requests})
+ self.assertEqual(
+ ["/target-bucket"],
+ [urlsplit(path).path for method, path in server.requests
+ if method == "POST"])
+ finally:
+ server.shutdown()
+ server.server_close()
+ server_thread.join()
diff --git a/paimon-python/pyproject.toml b/paimon-python/pyproject.toml
index d39be4f839..c4a4ecb893 100644
--- a/paimon-python/pyproject.toml
+++ b/paimon-python/pyproject.toml
@@ -169,6 +169,7 @@ hdfs = [
# Tools are separated from optional integrations so old-Python core tests and
# targeted compatibility jobs do not install unrelated native dependencies.
test = [
+ "boto3>=1.34,<2; python_version >= '3.10'",
"pytest>=6.2,<7; python_version < '3.7'",
"pytest>=7,<8; python_version >= '3.7'",
"parameterized>=0.8,<0.9; python_version < '3.7'",
