Support batch delete in Storage

[CTTY]

Is your feature request related to a problem or challenge?

As discussed here, we currently don't have a way to do batch deletes leveraging APIs like S3's DeleteObjects or Azure's BlobBatch

Describe the solution you'd like

Took a look at OpenDal and ObjectStore and noticed that they already have delete_iter and delete_stream to support this. From the iceberg side we can add delete_iter as well. Iterator can be easily converted to stream for ObjectStore's Storage implementation.

plus the paths to delete should be less than 1000(S3) and 256(Azure). pipelining the paths to backend storage won't be really necessary since the bottleneck is obviously the actual delete operation

Willingness to contribute

I can contribute to this feature independently

[CTTY]

Found some issues when designing the API:

1. delete_iter would be tricky to put in a trait due to object safety. fn delete_stream( &self, locations: BoxStream<'static, Result<String>>, ) seems like a better choice

2. How should we handle error?
   In object_store, it returns a stream of results:

    fn delete_stream(
        &self,
        locations: BoxStream<'static, Result<Path>>,
    ) -> BoxStream<'static, Result<Path>>;

While in opendal, it returns one result:

pub async fn delete_stream<S, D>(&mut self, stream: S) -> Result<()>

Looking at S3 sdk documentation, the semantics of DeleteObjects is closer to opendal

Currently in my PR I'm going with async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()>;

[blackmwk]

I think it's reasonable to implement delete_stream only, delete_iter is a sync version of delete_sttream, which I think is not practical in most cases.

[blackmwk]

fn delete_stream(
&self,
locations: BoxStream<'static, Result>,
) -> BoxStream<'static, Result>;

What's the return value here? It's quite unclear, it could be

1. Not successfully deleted objects.
2. All successfully deleted objects.

[CTTY]

fn delete_stream(
&self,
locations: BoxStream<'static, Result>,
) -> BoxStream<'static, Result>;

What's the return value here?

This is refering to object_store::delete_stream. quoting the doc:

The returned stream yields the results of the delete operations in the same order as the input locations. However, some errors will be from an overall call to a bulk delete operation, and not from a specific location.

If the object did not exist, the result may be an error or a success, depending on the behavior of the underlying store.

I think it makes sense to return results for each objects we attempted to delete.

But when looking at Opendal, it does not return results for each object when delete_stream, it just returns Result<()>. Maybe there are existing ways in Opendal to better handle the results? cc @Xuanwo

[CTTY]

Punting this issue to phase 2 of storage trait work

[blackmwk]

I prefer the opendal approach, e.g.

pub async fn delete_stream<S, D>(&mut self, stream: S) -> Result<()>

I think it will add extra requirements for implementations to return result of each path, and I'm not sure all implementations could implement it.