add concurrent methods to write manifests

Author: JanKaul

iceberg-rust/src/table/manifest.rs

@@ -17,6 +17,7 @@
 
 use std::{
     collections::HashSet,
+    future::Future,
     io::Read,
     iter::{repeat, Map, Repeat, Zip},
     sync::Arc,
@@ -26,6 +27,7 @@ use apache_avro::{
     to_value, types::Value as AvroValue, Reader as AvroReader, Schema as AvroSchema,
     Writer as AvroWriter,
 };
+use futures::TryFutureExt;
 use iceberg_rust_spec::{
     manifest::{Content, ManifestEntry, ManifestEntryV1, ManifestEntryV2, Status},
     manifest_list::{self, FieldSummary, ManifestListEntry},
@@ -35,7 +37,7 @@ use iceberg_rust_spec::{
     util::strip_prefix,
     values::{Struct, Value},
 };
-use object_store::ObjectStore;
+use object_store::{ObjectStore, PutResult};
 
 use crate::error::Error;
 
@@ -636,6 +638,54 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
             .await?;
         Ok(self.manifest)
     }
+
+    /// Finishes writing the manifest file concurrently.
+    ///
+    /// This method completes the manifest writing process by finalizing the writer
+    /// and returning both the manifest list entry and a future for the actual file upload.
+    /// The upload operation can be awaited separately, allowing for concurrent processing
+    /// of multiple manifest writes.
+    ///
+    /// # Arguments
+    ///
+    /// * `object_store` - The object store implementation used to persist the manifest file
+    ///
+    /// # Returns
+    ///
+    /// Returns a tuple containing:
+    /// - `ManifestListEntry`: The completed manifest entry with updated metadata
+    /// - `impl Future<Output = Result<PutResult, Error>>`: A future that performs the actual file upload
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - The writer cannot be finalized
+    /// - There are issues preparing the upload operation
+    pub(crate) fn finish_concurrently(
+        mut self,
+        object_store: Arc<dyn ObjectStore>,
+    ) -> Result<
+        (
+            ManifestListEntry,
+            impl Future<Output = Result<PutResult, Error>>,
+        ),
+        Error,
+    > {
+        let manifest_bytes = self.writer.into_inner()?;
+
+        let manifest_length: i64 = manifest_bytes.len() as i64;
+
+        self.manifest.manifest_length += manifest_length;
+
+        let path = strip_prefix(&self.manifest.manifest_path).as_str().into();
+        let future = async move {
+            object_store
+                .put(&path, manifest_bytes.into())
+                .map_err(Error::from)
+                .await
+        };
+        Ok((self.manifest, future))
+    }
 }
 
 #[allow(clippy::type_complexity)]

iceberg-rust/src/table/manifest_list.rs

@@ -4,6 +4,7 @@
 
 use std::{
     collections::{HashMap, HashSet},
+    future::Future,
     io::{Cursor, Read},
     iter::{repeat, Map, Repeat, Zip},
     sync::Arc,
@@ -23,7 +24,7 @@ use iceberg_rust_spec::{
     table_metadata::{FormatVersion, TableMetadata},
     util::strip_prefix,
 };
-use object_store::ObjectStore;
+use object_store::{ObjectStore, PutResult};
 use smallvec::SmallVec;
 
 use crate::{
@@ -700,6 +701,20 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
         .await
     }
 
+    pub(crate) async fn append_filtered(
+        &mut self,
+        data_files: impl Iterator<Item = Result<ManifestEntry, Error>>,
+        snapshot_id: i64,
+        filter: Option<impl Fn(&Result<ManifestEntry, Error>) -> bool>,
+        object_store: Arc<dyn ObjectStore>,
+        content: Content,
+    ) -> Result<(), Error> {
+        self.append_filtered_concurrently(data_files, snapshot_id, filter, object_store, content)
+            .await?
+            .await?;
+        Ok(())
+    }
+
     /// Appends data files to a single manifest with optional filtering and finalizes the manifest list.
     ///
     /// This method extends the basic `append` functionality by providing the ability to
@@ -754,14 +769,14 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
     ///     object_store,
     /// ).await?;
     /// ```
-    pub(crate) async fn append_filtered(
+    pub(crate) async fn append_filtered_concurrently(
         &mut self,
         data_files: impl Iterator<Item = Result<ManifestEntry, Error>>,
         snapshot_id: i64,
         filter: Option<impl Fn(&Result<ManifestEntry, Error>) -> bool>,
         object_store: Arc<dyn ObjectStore>,
         content: Content,
-    ) -> Result<(), Error> {
+    ) -> Result<impl Future<Output = Result<PutResult, Error>>, Error> {
         let selected_manifest = match content {
             Content::Data => self.selected_data_manifest.take(),
             Content::Deletes => self.selected_delete_manifest.take(),
@@ -820,11 +835,11 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
             manifest_writer.append(manifest_entry?)?;
         }
 
-        let manifest = manifest_writer.finish(object_store.clone()).await?;
+        let (manifest, future) = manifest_writer.finish_concurrently(object_store.clone())?;
 
         self.writer.append_ser(manifest)?;
 
-        Ok(())
+        Ok(future)
     }
 
     /// Appends data files by splitting them across multiple manifests and finalizes the manifest list.
@@ -894,6 +909,27 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
         .await
     }
 
+    pub(crate) async fn append_multiple_filtered(
+        &mut self,
+        data_files: impl Iterator<Item = Result<ManifestEntry, Error>>,
+        snapshot_id: i64,
+        n_splits: u32,
+        filter: Option<impl Fn(&Result<ManifestEntry, Error>) -> bool>,
+        object_store: Arc<dyn ObjectStore>,
+        content: Content,
+    ) -> Result<(), Error> {
+        self.append_multiple_filtered_concurrently(
+            data_files,
+            snapshot_id,
+            n_splits,
+            filter,
+            object_store,
+            content,
+        )
+        .await?
+        .await?;
+        Ok(())
+    }
     /// Appends data files across multiple manifests with optional filtering and finalizes the manifest list.
     ///
     /// This method extends the `append_multiple` functionality by providing the ability to
@@ -955,15 +991,15 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
     ///     object_store,
     /// ).await?;
     /// ```
-    pub(crate) async fn append_multiple_filtered(
+    pub(crate) async fn append_multiple_filtered_concurrently(
         &mut self,
         data_files: impl Iterator<Item = Result<ManifestEntry, Error>>,
         snapshot_id: i64,
         n_splits: u32,
         filter: Option<impl Fn(&Result<ManifestEntry, Error>) -> bool>,
         object_store: Arc<dyn ObjectStore>,
         content: Content,
-    ) -> Result<(), Error> {
+    ) -> Result<impl Future<Output = Result<Vec<PutResult>, Error>>, Error> {
         let partition_fields = self
             .table_metadata
             .current_partition_fields(self.branch.as_deref())?;
@@ -1032,7 +1068,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
             split_datafiles(data_files, bounds, &partition_column_names, n_splits)?
         };
 
-        let manifest_futures = splits
+        let (manifests, manifest_futures) = splits
             .into_iter()
             .map(|entries| {
                 let manifest_location = self.next_manifest_location();
@@ -1050,17 +1086,17 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
                     manifest_writer.append(manifest_entry)?;
                 }
 
-                Ok::<_, Error>(manifest_writer.finish(object_store.clone()))
+                manifest_writer.finish_concurrently(object_store.clone())
             })
-            .collect::<Result<Vec<_>, _>>()?;
-
-        let manifests = futures::future::try_join_all(manifest_futures).await?;
+            .collect::<Result<(Vec<_>, Vec<_>), _>>()?;
 
         for manifest in manifests {
             self.writer.append_ser(manifest)?;
         }
 
-        Ok(())
+        let future = futures::future::try_join_all(manifest_futures);
+
+        Ok(future)
     }
 
     pub(crate) async fn finish(