You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "CTTY (via GitHub)" <gi...@apache.org> on 2026/03/06 00:21:23 UTC
[PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
CTTY opened a new pull request, #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Closes #2065
## What changes are included in this PR?
- Add `delete_stream` to `Storage` trait to support batch delete
<!--
Provide a summary of the modifications in this PR. List the main changes such as new features, bug fixes, refactoring, or any other updates.
-->
## Are these changes tested?
Added uts
<!--
Specify what test covers (unit test, integration test, etc.).
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "blackmwk (via GitHub)" <gi...@apache.org>.
blackmwk commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2915353718
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +498,40 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ let mut deleters: HashMap<String, opendal::Deleter> = HashMap::new();
+
+ while let Some(path) = paths.next().await {
+ let bucket = url::Url::parse(&path)
+ .ok()
+ .and_then(|u| u.host_str().map(|s| s.to_string()))
+ .unwrap_or_default();
+
+ let relative_path = if deleters.contains_key(&bucket) {
+ self.relativize_path(&path)?.to_string()
+ } else {
+ let (op, rel) = self.create_operator(&path)?;
+ let rel = rel.to_string();
+ let deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleters.insert(bucket.clone(), deleter);
+ rel
+ };
+
+ deleters
+ .get_mut(&bucket)
+ .unwrap()
Review Comment:
nit: We could use `deleter.entry` api to avoid unwrap here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "blackmwk (via GitHub)" <gi...@apache.org>.
blackmwk commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2902733920
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -203,4 +204,46 @@ mod tests {
}
}
}
+
+ #[tokio::test]
+ async fn test_file_io_s3_delete_stream() {
Review Comment:
Why only s3?
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +497,33 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ let mut deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleter
+ .delete(first_relative)
+ .await
+ .map_err(from_opendal_error)?;
+
+ // Use relativize_path for remaining paths to avoid rebuilding the operator each time.
+ while let Some(path) = paths.next().await {
+ let relative_path = self.relativize_path(&path)?;
Review Comment:
I see some problems with this approach, what if we are deleting things as following:
```
s3://bucket1/a.txt
s3://bucketb/b.txt
```
##########
crates/iceberg/src/io/file_io.rs:
##########
@@ -146,6 +147,15 @@ impl FileIO {
self.get_storage()?.delete_prefix(path.as_ref()).await
}
+ /// Delete multiple files from a stream of paths.
+ ///
+ /// # Arguments
+ ///
+ /// * paths: A stream of absolute paths starting with the scheme string used to construct [`FileIO`].
+ pub async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()> {
Review Comment:
nit: We could make this api more general as following:
```rust
fn delete_stream(&self, paths: impl Stream<Item=String>)
```
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -203,4 +204,46 @@ mod tests {
}
}
}
+
Review Comment:
Sorry, I don't get your point.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2908534173
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -203,4 +204,46 @@ mod tests {
}
}
}
+
Review Comment:
The same test would fail for gcs, because we use fake-gcs-server for testing, and fake-gcs-server doesn't support batch delete.
https://github.com/fsouza/fake-gcs-server/issues/1443
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2908534173
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -203,4 +204,46 @@ mod tests {
}
}
}
+
Review Comment:
This works for S3 but the same test would fail for gcs, because we use fake-gcs-server for testing, and fake-gcs-server doesn't support batch delete.
https://github.com/fsouza/fake-gcs-server/issues/1443
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2898387875
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -203,4 +204,46 @@ mod tests {
}
}
}
+
Review Comment:
I didn't add this for gcs because fake-gcs-server doesn't support batch delete
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2908567521
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +497,33 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ let mut deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleter
+ .delete(first_relative)
+ .await
+ .map_err(from_opendal_error)?;
+
+ // Use relativize_path for remaining paths to avoid rebuilding the operator each time.
+ while let Some(path) = paths.next().await {
+ let relative_path = self.relativize_path(&path)?;
Review Comment:
I thought about this as well, but I think it would be tricky to solve this. Because OpenDal's operator is tied to a bucket, and if user passes in locations from multiple buckets then we will have to create operator for each of the incoming locations --- it's gonna be slow
Also I'd say this is an acceptable limitation because most users will not have data spanned across several buckets
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2914632131
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +497,33 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ let mut deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleter
+ .delete(first_relative)
+ .await
+ .map_err(from_opendal_error)?;
+
+ // Use relativize_path for remaining paths to avoid rebuilding the operator each time.
+ while let Some(path) = paths.next().await {
+ let relative_path = self.relativize_path(&path)?;
Review Comment:
I've updated the implementation to include a HashMap to map from bucket name to deleter
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2898233322
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +402,30 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let storage = self.clone();
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ // Create a stream of relative paths, starting with the first one
+ let relative_paths = futures::stream::once(async move { first_relative.to_string() })
+ .chain(paths.map(move |path| {
+ let (_, relative_path) = storage
+ .create_operator(&path)
Review Comment:
This is a good catch, I think we just need to relativize the paths, there should be some way to refactor it. I'll take another look
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "blackmwk (via GitHub)" <gi...@apache.org>.
blackmwk commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2908645093
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +497,33 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ let mut deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleter
+ .delete(first_relative)
+ .await
+ .map_err(from_opendal_error)?;
+
+ // Use relativize_path for remaining paths to avoid rebuilding the operator each time.
+ while let Some(path) = paths.next().await {
+ let relative_path = self.relativize_path(&path)?;
Review Comment:
We don't need to create operator for each location, we just need to create operator for each bucket.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2908567521
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +497,33 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ let mut deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleter
+ .delete(first_relative)
+ .await
+ .map_err(from_opendal_error)?;
+
+ // Use relativize_path for remaining paths to avoid rebuilding the operator each time.
+ while let Some(path) = paths.next().await {
+ let relative_path = self.relativize_path(&path)?;
Review Comment:
I thought about this as well, but I think it would be tricky to solve this. Because OpenDal's operator is tied to a bucket, and if user passes in locations from multiple buckets then we will have to create operator for each of the incoming locations.
But I think this is an acceptable limitation because most users will not have data spanned across several buckets
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "blackmwk (via GitHub)" <gi...@apache.org>.
blackmwk merged PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "blackmwk (via GitHub)" <gi...@apache.org>.
blackmwk commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2893243901
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +402,30 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let storage = self.clone();
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ // Create a stream of relative paths, starting with the first one
+ let relative_paths = futures::stream::once(async move { first_relative.to_string() })
+ .chain(paths.map(move |path| {
+ let (_, relative_path) = storage
+ .create_operator(&path)
Review Comment:
This seems quite expensive, we create operator for each path?
##########
crates/storage/opendal/src/lib.rs:
##########
Review Comment:
We need to tests for opendal.
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +402,30 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let storage = self.clone();
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ // Create a stream of relative paths, starting with the first one
+ let relative_paths = futures::stream::once(async move { first_relative.to_string() })
+ .chain(paths.map(move |path| {
+ let (_, relative_path) = storage
+ .create_operator(&path)
+ .expect("Failed to create operator");
Review Comment:
We should not panic here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#discussion_r2908567521
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -400,6 +497,33 @@ impl Storage for OpenDalStorage {
Ok(op.remove_all(&path).await.map_err(from_opendal_error)?)
}
+ async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+ // Get the first path to create the operator
+ let Some(first_path) = paths.next().await else {
+ return Ok(());
+ };
+
+ let (op, first_relative) = self.create_operator(&first_path)?;
+
+ let mut deleter = op.deleter().await.map_err(from_opendal_error)?;
+ deleter
+ .delete(first_relative)
+ .await
+ .map_err(from_opendal_error)?;
+
+ // Use relativize_path for remaining paths to avoid rebuilding the operator each time.
+ while let Some(path) = paths.next().await {
+ let relative_path = self.relativize_path(&path)?;
Review Comment:
I thought about this as well, but I think it would be tricky to solve this. Because OpenDal's operator is tied to a bucket, and if user passes in locations from multiple buckets then we will have to create operator for each of the incoming locations.
Also I'd say this is an acceptable limitation because most users will not have data spanned across several buckets
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io): Add delete_stream to Storage trait [iceberg-rust]
Posted by "blackmwk (via GitHub)" <gi...@apache.org>.
blackmwk commented on PR #2216:
URL: https://github.com/apache/iceberg-rust/pull/2216#issuecomment-4035575510
Let's hold on a while for 0.9.0 release.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org