XiaoHongbo-Hope commented on code in PR #9821:
URL: https://github.com/apache/paimon/pull/9821#discussion_r4034736030
##########
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.
Thanks, fixed
--
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]
