wangzhigang1999 commented on code in PR #9821:
URL: https://github.com/apache/paimon/pull/9821#discussion_r4032071222
##########
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(
+ Bucket=bucket,
+ Delete={"Objects": [{"Key": key} for key in batch], "Quiet":
False})
+ deleted = [item["Key"] for item in response.get("Deleted", ())]
+ if response.get("Errors") or set(deleted) != set(batch) \
+ or len(deleted) != len(batch):
+ raise OSError(f"S3 batch delete incomplete for {path_str}")
Review Comment:
Could we include the error codes and a few failed keys from Errors? The
current message makes it difficult to distinguish per-object failures from a
mismatch in the returned keys.
##########
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(
+ Bucket=bucket,
+ Delete={"Objects": [{"Key": key} for key in batch], "Quiet":
False})
+ deleted = [item["Key"] for item in response.get("Deleted", ())]
+ if response.get("Errors") or set(deleted) != set(batch) \
+ or len(deleted) != len(batch):
+ 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
+ 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._get_s3_boolean_property("path-style-access") or
+ self._get_s3_boolean_property("path.style.access"))
+ addressing_style = "path" if path_style else "virtual"
Review Comment:
Could we share the connection-option parsing with _initialize_s3_fs?
Credentials, region, and path-style settings are parsed in both places and
could drift as options change.
##########
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?
--
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]
