From 537f79113976e0d801cd0ae8f5f1da32b1e19911 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sun, 11 Oct 2026 02:43:07 -0700 Subject: [PATCH 1/2] [python] Fix OSS directory deletion with PyArrow 22+ --- .github/workflows/ci-python.yml | 26 + .../pypaimon/filesystem/pyarrow_file_io.py | 219 +++- .../tests/s3_delete_compatibility_test.py | 1068 +++++++++++++++++ paimon-python/pyproject.toml | 1 + 4 files changed, 1313 insertions(+), 1 deletion(-) create mode 100644 paimon-python/pypaimon/tests/s3_delete_compatibility_test.py diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 952a1173a559..95962575454a 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 446bb64ed5f2..1a4362aedb2b 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 @@ def __init__(self, path: str, catalog_options: Options): 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 @@ def __init__(self, path: str, catalog_options: Options): 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 @@ def _uses_s3_compatibility(self) -> bool: 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 @@ def __getstate__(self): # 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 @@ def __getstate__(self): 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 @@ def _initialize_s3_fs(self) -> FileSystem: 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 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: } 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 @@ def delete(self, path: str, recursive: bool = False) -> bool: 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 000000000000..1004196ae930 --- /dev/null +++ b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py @@ -0,0 +1,1068 @@ +# 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"AuthorizationHeaderMalformed") + 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 = ( + '' + '' + '{}{}{}' + '{}{}{}' + 'false{}{}'.format( + escape(bucket), escape(response_prefix), escape(response_delimiter), + len(contents) + len(common_prefixes), max_keys, + 'url' if encode_keys else '', + "".join( + "{}" + "2026-01-01T00:00:00Z" + "{}STANDARD" + "".format( + escape(key), 0 if key.endswith("/") else 1) + for key in response_contents), + "".join( + "{}".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"InvalidRequest") + content_md5 = base64.b64encode( + hashlib.md5(body, usedforsecurity=False).digest()).decode() + if self.headers.get("Content-MD5") != content_md5: + return self._respond( + 400, b"MissingArgument") + if b"\x01" in body: + return self._respond(400, b"MalformedXML") + 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 = "{}{}".format( + "".join("{}".format(escape(key)) + for key in deleted), + "{}AccessDenied".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_does_not_modify_parent_marker(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)] + _set_listed_keys(file_io, ["db-uuid.db/tbl-uuid/data.parquet"], []) + client = file_io._s3_delete_client + self.assertTrue(file_io.delete(TABLE_PATH, recursive=True)) + + client.head_object.assert_not_called() + client.put_object.assert_not_called() + + def test_pyarrow_22_recursive_delete_preserves_late_objects(self): + file_io = self._new_file_io(legacy=False) + file_io._pyarrow_gte_22 = True + directory = file_io.to_filesystem_path(TABLE_PATH) + first = directory.rstrip("/") + "/first.parquet" + late = directory.rstrip("/") + "/late.parquet" + file_io.filesystem.get_file_info.return_value = [ + _file_info(directory, pafs.FileType.Directory)] + _set_listed_keys(file_io, [first.split("/", 1)[1]], [late.split("/", 1)[1]], []) + + with self.assertRaisesRegex(OSError, "changed during deletion"): + file_io.delete(TABLE_PATH, recursive=True) + + self.assertEqual(["db-uuid.db/tbl-uuid/first.parquet"], + _all_deleted_keys(file_io._s3_delete_client)) + file_io._s3_delete_client.put_object.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_recursive_delete_reports_batch_error(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/" + data = prefix + "data/file.parquet" + _set_listed_keys(file_io, [data]) + client = file_io._s3_delete_client + client.delete_objects.side_effect = None + client.delete_objects.return_value = { + "Deleted": [], "Errors": [{"Key": data, "Code": "AccessDenied"}]} + + with self.assertRaisesRegex(OSError, "batch delete incomplete") as error: + file_io.delete(TABLE_PATH, recursive=True) + self.assertIn("AccessDenied", str(error.exception)) + self.assertIn(data, str(error.exception)) + + client.delete_object.assert_not_called() + client.put_object.assert_not_called() + self.assertEqual([data], [ + item["Key"] for item in + client.delete_objects.call_args[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_modern_non_recursive_delete_removes_empty_directory(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.side_effect = [ + [_file_info(directory, pafs.FileType.Directory)], + [], + ] + + self.assertTrue(file_io.delete(TABLE_PATH)) + + file_io.filesystem.delete_file.assert_not_called() + file_io.filesystem.delete_dir.assert_not_called() + file_io._s3_delete_client.delete_object.assert_called_once_with( + Bucket="test-bucket", Key="db-uuid.db/tbl-uuid/") + file_io._s3_delete_client.put_object.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_delete_rejects_bucket_root_with_fallback(self): + file_io = self._new_file_io() + 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("s3://test-bucket/", recursive) + + file_io.filesystem.get_file_info.assert_not_called() + file_io.filesystem.delete_dir_contents.assert_not_called() + file_io._s3_delete_client.delete_object.assert_not_called() + + def test_old_pyarrow_delete_keeps_native_s3_path(self): + file_io = self._new_file_io() + 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("s3://test-bucket/", recursive=True)) + file_io.filesystem.get_file_info.assert_called_once() + file_io._s3_delete_client.delete_object.assert_not_called() + + def test_recursive_delete_rejects_listed_keys_outside_prefix(self): + 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)] + _set_listed_keys(file_io, [ + "table/data.parquet", "table-other/keep.parquet"]) + + with self.assertRaisesRegex(OSError, "outside 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_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_recursive_delete_uses_bucket_from_target_uri(self): + file_io = self._new_file_io() + file_io._pyarrow_gte_22 = True + 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)] + _set_listed_keys(file_io, ["table/data.parquet"], []) + + self.assertTrue(file_io.delete( + "s3://target-bucket/table", recursive=True)) + + file_io._s3_delete_client.delete_objects.assert_called_once_with( + Bucket="target-bucket", Delete={ + "Objects": [{"Key": "table/data.parquet"}], "Quiet": False}) + file_io._s3_delete_client.delete_object.assert_not_called() + + def test_recursive_delete_uses_bucket_from_target_filesystem_path(self): + file_io = self._new_file_io() + file_io._pyarrow_gte_22 = True + file_io.filesystem.get_file_info.return_value = [ + _file_info("target-bucket/table", pafs.FileType.Directory)] + _set_listed_keys(file_io, ["table/data.parquet"], []) + + self.assertTrue(file_io.delete( + "target-bucket/table", recursive=True)) + + file_io._s3_delete_client.delete_objects.assert_called_once_with( + Bucket="target-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() + + def test_non_recursive_delete_uses_bucket_from_target_uri(self): + file_io = self._new_file_io() + file_io._pyarrow_gte_22 = True + file_io.to_filesystem_path = mock.Mock( + return_value="target-bucket/table") + file_io.filesystem.get_file_info.side_effect = [ + [_file_info("target-bucket/table", pafs.FileType.Directory)], + [], + ] + + self.assertTrue(file_io.delete("s3://target-bucket/table")) + + file_io._s3_delete_client.delete_object.assert_called_once_with( + Bucket="target-bucket", Key="table/") + + @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 d39be4f839df..c4a4ecb893bd 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'", From 6ade0dbcfd8342b129a07d9ce76ddc3c85068af1 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sun, 11 Oct 2026 05:15:49 -0700 Subject: [PATCH 2/2] [python] Trim overlapping S3 deletion tests --- .../tests/s3_delete_compatibility_test.py | 155 ------------------ 1 file changed, 155 deletions(-) diff --git a/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py index 1004196ae930..f5dd6d08ec97 100644 --- a/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py +++ b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py @@ -304,36 +304,6 @@ def test_pyarrow_22_recursive_delete_batches_objects(self): file_io.filesystem.delete_dir_contents.assert_not_called() file_io.filesystem.delete_dir.assert_not_called() - def test_recursive_delete_does_not_modify_parent_marker(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)] - _set_listed_keys(file_io, ["db-uuid.db/tbl-uuid/data.parquet"], []) - client = file_io._s3_delete_client - self.assertTrue(file_io.delete(TABLE_PATH, recursive=True)) - - client.head_object.assert_not_called() - client.put_object.assert_not_called() - - def test_pyarrow_22_recursive_delete_preserves_late_objects(self): - file_io = self._new_file_io(legacy=False) - file_io._pyarrow_gte_22 = True - directory = file_io.to_filesystem_path(TABLE_PATH) - first = directory.rstrip("/") + "/first.parquet" - late = directory.rstrip("/") + "/late.parquet" - file_io.filesystem.get_file_info.return_value = [ - _file_info(directory, pafs.FileType.Directory)] - _set_listed_keys(file_io, [first.split("/", 1)[1]], [late.split("/", 1)[1]], []) - - with self.assertRaisesRegex(OSError, "changed during deletion"): - file_io.delete(TABLE_PATH, recursive=True) - - self.assertEqual(["db-uuid.db/tbl-uuid/first.parquet"], - _all_deleted_keys(file_io._s3_delete_client)) - file_io._s3_delete_client.put_object.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 @@ -477,31 +447,6 @@ def test_recursive_delete_batches_at_most_1000_keys(self): item["Key"] for call in calls for item in call[1]["Delete"]["Objects"]]) - def test_recursive_delete_reports_batch_error(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/" - data = prefix + "data/file.parquet" - _set_listed_keys(file_io, [data]) - client = file_io._s3_delete_client - client.delete_objects.side_effect = None - client.delete_objects.return_value = { - "Deleted": [], "Errors": [{"Key": data, "Code": "AccessDenied"}]} - - with self.assertRaisesRegex(OSError, "batch delete incomplete") as error: - file_io.delete(TABLE_PATH, recursive=True) - self.assertIn("AccessDenied", str(error.exception)) - self.assertIn(data, str(error.exception)) - - client.delete_object.assert_not_called() - client.put_object.assert_not_called() - self.assertEqual([data], [ - item["Key"] for item in - client.delete_objects.call_args[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 @@ -516,23 +461,6 @@ def test_pre_pyarrow_22_recursive_delete_keeps_native_batch(self): file_io.filesystem.delete_dir.assert_called_once_with(directory) file_io.filesystem.delete_file.assert_not_called() - def test_modern_non_recursive_delete_removes_empty_directory(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.side_effect = [ - [_file_info(directory, pafs.FileType.Directory)], - [], - ] - - self.assertTrue(file_io.delete(TABLE_PATH)) - - file_io.filesystem.delete_file.assert_not_called() - file_io.filesystem.delete_dir.assert_not_called() - file_io._s3_delete_client.delete_object.assert_called_once_with( - Bucket="test-bucket", Key="db-uuid.db/tbl-uuid/") - file_io._s3_delete_client.put_object.assert_not_called() - def test_delete_rejects_bucket_root(self): file_io = self._new_file_io(legacy=False) file_io._pyarrow_gte_22 = True @@ -653,42 +581,6 @@ def test_jindo_and_native_s3_do_not_use_delete_fallback(self): file_io._s3_endpoint = None self.assertFalse(file_io._uses_s3_delete_fallback()) - def test_delete_rejects_bucket_root_with_fallback(self): - file_io = self._new_file_io() - 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("s3://test-bucket/", recursive) - - file_io.filesystem.get_file_info.assert_not_called() - file_io.filesystem.delete_dir_contents.assert_not_called() - file_io._s3_delete_client.delete_object.assert_not_called() - - def test_old_pyarrow_delete_keeps_native_s3_path(self): - file_io = self._new_file_io() - 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("s3://test-bucket/", recursive=True)) - file_io.filesystem.get_file_info.assert_called_once() - file_io._s3_delete_client.delete_object.assert_not_called() - - def test_recursive_delete_rejects_listed_keys_outside_prefix(self): - 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)] - _set_listed_keys(file_io, [ - "table/data.parquet", "table-other/keep.parquet"]) - - with self.assertRaisesRegex(OSError, "outside 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_recursive_delete_rejects_mismatched_list_scope(self): for scope in ({"Name": "other-bucket"}, {"Prefix": "table-other/"}): @@ -728,38 +620,6 @@ def test_pyarrow_22_recursive_delete_uses_batch(self): "Quiet": False}) file_io._s3_delete_client.delete_object.assert_not_called() - def test_recursive_delete_uses_bucket_from_target_uri(self): - file_io = self._new_file_io() - file_io._pyarrow_gte_22 = True - 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)] - _set_listed_keys(file_io, ["table/data.parquet"], []) - - self.assertTrue(file_io.delete( - "s3://target-bucket/table", recursive=True)) - - file_io._s3_delete_client.delete_objects.assert_called_once_with( - Bucket="target-bucket", Delete={ - "Objects": [{"Key": "table/data.parquet"}], "Quiet": False}) - file_io._s3_delete_client.delete_object.assert_not_called() - - def test_recursive_delete_uses_bucket_from_target_filesystem_path(self): - file_io = self._new_file_io() - file_io._pyarrow_gte_22 = True - file_io.filesystem.get_file_info.return_value = [ - _file_info("target-bucket/table", pafs.FileType.Directory)] - _set_listed_keys(file_io, ["table/data.parquet"], []) - - self.assertTrue(file_io.delete( - "target-bucket/table", recursive=True)) - - file_io._s3_delete_client.delete_objects.assert_called_once_with( - Bucket="target-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 @@ -777,21 +637,6 @@ def test_pre_pyarrow_22_cross_bucket_delete_keeps_native_path(self): "target-bucket/table") file_io._s3_delete_client.delete_object.assert_not_called() - def test_non_recursive_delete_uses_bucket_from_target_uri(self): - file_io = self._new_file_io() - file_io._pyarrow_gte_22 = True - file_io.to_filesystem_path = mock.Mock( - return_value="target-bucket/table") - file_io.filesystem.get_file_info.side_effect = [ - [_file_info("target-bucket/table", pafs.FileType.Directory)], - [], - ] - - self.assertTrue(file_io.delete("s3://target-bucket/table")) - - file_io._s3_delete_client.delete_object.assert_called_once_with( - Bucket="target-bucket", Key="table/") - @unittest.skipUnless( parse(pyarrow.__version__) >= parse("22.0.0"), "requires PyArrow 22+ and boto3",