JingsongLi commented on code in PR #9821:
URL: https://github.com/apache/paimon/pull/9821#discussion_r4023484275
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -465,6 +516,107 @@ 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 += "/"
+ while True:
+ response = client.list_objects_v2(
+ Bucket=bucket, Prefix=prefix, MaxKeys=1000)
+ keys = (
+ item["Key"] for item in response.get("Contents", ())
+ if item["Key"] != prefix
+ )
+ if self._delete_s3_objects(client, bucket, keys):
+ continue
+ if prefix:
+ client.delete_object(Bucket=bucket, Key=prefix)
Review Comment:
**[P2] Preserve implicit parent directory markers**
Both compatibility paths delete the target directory marker and return
without recreating its immediate parent. PyArrow's `S3FileSystem::DeleteDir`
explicitly calls `EnsureParentExists` after deleting a directory because an
implicit parent can otherwise disappear. This is reachable in the filesystem
catalog: `list_databases()` and `get_database()` accept an implicit S3 prefix,
but dropping its last table removes the prefix's only child and makes the
database become `NotFound`. The request-level cross-bucket test already models
`parent/child/` without a `parent/` marker, but currently expects the bucket to
become empty. Please recreate the immediate parent marker after successful
recursive and non-recursive directory deletion, except when the parent is the
bucket root.
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -98,11 +107,21 @@ def __getstate__(self):
state = self.__dict__.copy()
# threading.Lock cannot be pickled; recreated in __setstate__.
state.pop("_legacy_bucket_lock", 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)
return state
def __setstate__(self, state):
self.__dict__.update(state)
self._legacy_bucket_lock = threading.Lock()
+ self._s3_delete_client = None
+ if self._uses_s3_compatibility():
Review Comment:
**[P2] Recompute compatibility state in the worker**
This rebuilds a worker-local filesystem while retaining the producer's
serialized PyArrow gates (`_pyarrow_gte_8`, `_pyarrow_gte_16`,
`_pyarrow_gte_22`, and `_oss_bucket_in_endpoint`). With a supported PyArrow 21
driver and PyArrow 23 worker, `_pyarrow_gte_22` remains false, so recursive
deletion takes the incompatible native batch path that this PR is intended to
avoid. A pre-PR non-OSS pickle also has no `_s3_endpoint`, causing
`__setstate__` to fail with `AttributeError`. Please migrate missing fields and
recompute all version-dependent state from the worker's installed PyArrow
before rebuilding the client, or reconstruct from stable path/options state.
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -465,6 +516,107 @@ 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 += "/"
+ while True:
Review Comment:
**[P2] Bound the relist loop under concurrent writers**
This loop exits only after a verification listing contains no non-marker
object. A writer that continues adding objects under the prefix—or an
eventually consistent compatible endpoint that repeatedly returns a recently
deleted key—can therefore keep `drop_table`/`drop_database` issuing GET and
DELETE requests forever. The regression test injects exactly one late object
and does not cover sustained churn. Please add a pass, deadline, or request
budget and raise a clear concurrent-modification error when the prefix does not
quiesce.
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -163,6 +182,21 @@ def _get_s3_boolean_property(self, name: str) -> bool:
return value
return OptionsUtils.convert_to_boolean(value)
+ 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
+ and not (self._is_s3 and self._get_s3_boolean_property(
+ "delete.batch-enabled")))
+
+ @staticmethod
+ def _configure_s3_compatibility():
+ os.environ.setdefault(
Review Comment:
**[P2] Do not implement endpoint-local behavior through conditional
process-global state**
`setdefault` has two conflicting failure modes. If the host already sets the
valid value `AWS_REQUEST_CHECKSUM_CALCULATION=WHEN_SUPPORTED`, the
OSS/custom-endpoint workaround is silently skipped and uploads can still use
the incompatible optional checksum trailers. If the variable is initially
absent, this permanently changes the checksum policy of native AWS S3 clients
created later in the same process. The tests clear `os.environ` before each
construction, so they cover neither case. Please scope and restore this setting
around compatible client construction where possible, or detect an incompatible
pre-existing value and fail with an actionable message rather than silently
leaving the workaround disabled.
--
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]
