XiaoHongbo-Hope commented on code in PR #9821:
URL: https://github.com/apache/paimon/pull/9821#discussion_r4034658662
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -465,6 +554,198 @@ 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:
+ for name in self._list_s3_keys(client, bucket, prefix, deadline,
path_str):
+ if name == prefix:
+ continue
+ target = schemas if "/schema/schema-" in name else listed
+ target.write(json.dumps(name) + "\n")
+
+ listed.seek(0)
+ self._delete_s3_objects(
+ client, bucket, (json.loads(line) for line in listed),
+ deadline, path_str)
+
+ known_schemas = self._staged_keys(schemas)
+ expected = next(known_schemas, None)
+ for name in self._list_s3_keys(client, bucket, prefix, deadline,
path_str):
+ if name == prefix:
+ continue
+ while expected is not None and expected < name:
+ expected = next(known_schemas, None)
+ if name != expected:
+ raise OSError(f"S3 directory {path_str} changed during
deletion")
+ expected = next(known_schemas, None)
+
+ self._check_s3_delete_deadline(deadline, path_str)
+ self._ensure_s3_parent_exists(client, bucket, key)
+ if prefix:
+ self._check_s3_delete_deadline(deadline, path_str)
+ client.delete_object(Bucket=bucket, Key=prefix)
+ self._delete_s3_objects(
+ client, bucket, self._staged_schema_keys(schemas, zero=False),
+ deadline, path_str)
+ self._delete_s3_objects(
+ client, bucket, self._staged_schema_keys(schemas, zero=True),
+ deadline, path_str)
+ return True
+
+ @staticmethod
+ def _staged_keys(staged):
+ staged.seek(0)
+ for line in staged:
+ yield json.loads(line)
+
+ @staticmethod
+ def _staged_schema_keys(staged, zero: bool):
+ for name in PyArrowFileIO._staged_keys(staged):
+ if name.endswith("/schema/schema-0") == zero:
+ yield name
+
+ 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 _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 _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)
+ response = client.delete_objects(
Review Comment:
> I reproduced a regression on real OSS: keys containing \r are deleted, but
XML normalizes CR to LF, causing the returned-key comparison to fail. For
partition values containing \x01, append and PK tables write/read successfully,
but drop_table exhausts HTTP 500 retries after deleting the data, leaving
schema-0. The baseline succeeds in both cases. Could we investigate the
request/response handling and add regression tests, including partial failures?
Thanks, fixed and added case?
--
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]
