JingsongLi commented on code in PR #9821:
URL: https://github.com/apache/paimon/pull/9821#discussion_r4025651319
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -465,6 +553,153 @@ 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
+ with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as listed, \
+ tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as
schemas, \
+ tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as
schema_zero:
+ token = None
+ while True:
+ if time.monotonic() >= deadline:
+ raise TimeoutError(f"Timed out listing S3 directory
{path_str}")
+ params = {"Bucket": bucket, "Prefix": prefix, "MaxKeys": 1000}
+ if token is not None:
+ params["ContinuationToken"] = token
+ response = client.list_objects_v2(**params)
+ for name in self._listed_s3_keys(response, bucket, prefix):
+ if name != prefix:
+ # Keep schema-0 until all other table objects are gone.
+ target = (schema_zero if
name.endswith("/schema/schema-0")
+ else schemas if "/schema/schema-" in name
+ else listed)
+ target.write(json.dumps(name) + "\n")
+ if not response.get("IsTruncated"):
+ break
+ next_token = response.get("NextContinuationToken")
+ if not next_token or next_token == token:
+ raise OSError("S3 listing did not advance")
+ token = next_token
+
+ for staged in (listed, schemas, schema_zero):
Review Comment:
**[P2] Validate concurrent changes before deleting `schema-0`**
This loop deletes `schema_zero` before the late-object check below. If a
writer adds a key after the initial snapshot, the method raises `changed during
deletion`, but `schema/schema-0` has already been removed, so the failed drop
leaves the table undiscoverable and cannot be retried as an existing table. I
reproduced this by combining the existing late-object scenario with a staged
`schema-0`: the exception is raised after both the data key and `schema-0`
appear in `delete_object` calls. Please delete ordinary objects first, relist
while allowing only the marker and known schema keys, and delete `schema-0`
only after that validation succeeds.
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -465,6 +553,153 @@ 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:
Review Comment:
**Simplify this fallback around the boto3 client**
This method now reimplements pagination, disk spooling, concurrency,
deadlines, schema ordering, marker cleanup, and scope validation inside
`PyArrowFileIO`. Since `_get_s3_delete_client` already sets
`request_checksum_calculation="when_required"`, could we first validate boto3
`delete_objects` against the real OSS/DLF endpoint and delete batches of up to
1,000 keys? OSS's S3 compatibility explicitly supports `DeleteObjects`:
https://www.alibabacloud.com/help/en/oss/developer-reference/compatibility-with-amazon-s3.
A smaller flow would snapshot/spool ordinary keys once, batch-delete them,
relist while retaining `schema-0`, then delete `schema-0` and the marker. That
removes the thread pool, per-object requests, three staging streams, and much
of this branching. If a specific endpoint truly cannot batch-delete, please
isolate the individual-object strategy in a small helper rather than embedding
the entire state machine here.
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -790,6 +1024,13 @@ def to_filesystem_path(self, path: str) -> str:
parsed = urlparse(path)
normalized_path = re.sub(r'/+', '/', parsed.path) if parsed.path else
''
+ if (self._is_oss and (self._use_jindo or self._oss_bucket_in_endpoint)
+ and (parsed.scheme or parsed.netloc)):
+ if (not parsed.netloc or "@" in parsed.netloc
Review Comment:
**[P2] Preserve credential-bearing OSS URIs**
Rejecting every authority containing `@` breaks the existing supported form
`oss://access_id:secret_key@Endpoint/bucket/...`. `_extract_oss_bucket`
explicitly parses this form, and `ao_simple_test.py` verifies its construction,
but in legacy bucket-in-endpoint and Jindo modes every subsequent
read/write/delete now fails here with `OSS path is outside current bucket`. I
reproduced the failure in both modes through `to_filesystem_path`. Please
validate the bucket returned by `_extract_oss_bucket` instead of rejecting `@`
unconditionally, and add an I/O-path regression test for this URI form.
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -465,6 +553,153 @@ 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
+ with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as listed, \
+ tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as
schemas, \
+ tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as
schema_zero:
+ token = None
+ while True:
+ if time.monotonic() >= deadline:
+ raise TimeoutError(f"Timed out listing S3 directory
{path_str}")
+ params = {"Bucket": bucket, "Prefix": prefix, "MaxKeys": 1000}
+ if token is not None:
+ params["ContinuationToken"] = token
+ response = client.list_objects_v2(**params)
+ for name in self._listed_s3_keys(response, bucket, prefix):
+ if name != prefix:
+ # Keep schema-0 until all other table objects are gone.
+ target = (schema_zero if
name.endswith("/schema/schema-0")
+ else schemas if "/schema/schema-" in name
+ else listed)
+ target.write(json.dumps(name) + "\n")
+ if not response.get("IsTruncated"):
+ break
+ next_token = response.get("NextContinuationToken")
+ if not next_token or next_token == token:
+ raise OSError("S3 listing did not advance")
+ token = next_token
+
+ for staged in (listed, schemas, schema_zero):
+ staged.seek(0)
+ self._delete_s3_objects(
+ client, bucket, (json.loads(line) for line in staged))
+
+ if time.monotonic() >= deadline:
+ raise TimeoutError(f"Timed out deleting S3 directory {path_str}")
+ response = client.list_objects_v2(
+ Bucket=bucket, Prefix=prefix, MaxKeys=2)
+ if any(name != prefix for name in
+ self._listed_s3_keys(response, bucket, prefix)):
+ raise OSError(f"S3 directory {path_str} changed during deletion")
+ if prefix:
+ client.delete_object(Bucket=bucket, Key=prefix)
+ self._ensure_s3_parent_exists(client, bucket, key)
+ return True
+
+ @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 _ensure_s3_parent_exists(client, bucket: str, key: str):
+ parent, _, _ = key.rstrip("/").rpartition("/")
+ if parent:
+ client.put_object(
+ Bucket=bucket, Key=parent + "/", Body=b"",
+ ContentType="application/x-directory")
+
+ @staticmethod
+ def _delete_s3_objects(
+ client, bucket: str, keys: Iterable[str]) -> int:
+ keys = iter(keys)
+ batch = list(islice(keys, 16))
+ if not batch:
+ return 0
+ deleted = 0
+ with ThreadPoolExecutor(max_workers=16) as executor:
Review Comment:
**[P2] Enforce the deadline inside the deletion loop**
`_delete_s3_objects` does not receive or check the deadline, so a slow
endpoint can continue submitting every later 16-key batch long after the
one-hour budget has expired. With 60-second connect/read timeouts and ten retry
attempts per request, the final check may run much later, potentially only
after schemas have already been deleted. Pass the deadline into this helper,
check it before every new batch/stage, and avoid entering schema deletion once
the budget is exhausted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
