Merge branch 'JanKaul:main' into main

Author: gruuya

iceberg-rust/src/table/transaction/append.rs

@@ -1,7 +1,8 @@
 use std::cmp::Ordering;
-
-use iceberg_rust_spec::{manifest::ManifestEntry, manifest_list::ManifestListEntry};
+use std::collections::HashMap;
 use smallvec::SmallVec;
+use iceberg_rust_spec::{
+    manifest::ManifestEntry, manifest_list::ManifestListEntry, manifest::Content, manifest::DataFile};
 
 use crate::{
     error::Error,
@@ -189,3 +190,24 @@ pub(crate) fn select_manifest_unpartitioned(
         })
         .ok_or(Error::NotFound("Manifest for insert".to_owned()))
 }
+
+
+pub(crate) fn append_summary(files: &[DataFile]) -> Option<HashMap<String, String>> {
+    if files.is_empty() {
+        return None;
+    }
+
+    let (mut added_data_files, mut added_records, mut added_files_size) = (0usize, 0i64, 0i64);
+
+    for file in files.iter().filter(|f| *f.content() == Content::Data) {
+        added_data_files += 1;
+        added_records += file.record_count();
+        added_files_size += file.file_size_in_bytes();
+    }
+
+    Some(HashMap::from([
+        ("added-files-size".into(), added_files_size.to_string()),
+        ("added-records".into(), added_records.to_string()),
+        ("added-data-files".into(), added_data_files.to_string()),
+    ]))
+}

iceberg-rust/src/table/transaction/mod.rs

@@ -20,6 +20,7 @@ use std::collections::HashMap;
 use iceberg_rust_spec::spec::{manifest::DataFile, schema::Schema, snapshot::SnapshotReference};
 
 use crate::{catalog::commit::CommitTable, error::Error, table::Table};
+use crate::table::transaction::append::append_summary;
 
 use self::operation::Operation;
 
@@ -116,15 +117,12 @@ impl<'table> TableTransaction<'table> {
     ///     .await?;
     /// ```
     pub fn append_data(mut self, files: Vec<DataFile>) -> Self {
+        let summary = append_summary(&files);
+
         self.operations
             .entry(APPEND_KEY.to_owned())
             .and_modify(|mut x| {
-                if let Operation::Append {
-                    branch: _,
-                    data_files: old,
-                    delete_files: _,
-                    additional_summary: None,
-                } = &mut x
+                if let Operation::Append { data_files: old, ..} = &mut x
                 {
                     old.extend_from_slice(&files)
                 }
@@ -133,7 +131,7 @@ impl<'table> TableTransaction<'table> {
                 branch: self.branch.clone(),
                 data_files: files,
                 delete_files: Vec::new(),
-                additional_summary: None,
+                additional_summary: summary,
             });
         self
     }
@@ -346,6 +344,10 @@ impl<'table> TableTransaction<'table> {
             updates.extend(update);
         }
 
+        if updates.is_empty() {
+            return Ok(());
+        }
+
         let new_table = catalog
             .clone()
             .update_table(CommitTable {

iceberg-rust/src/table/transaction/operation.rs

@@ -106,10 +106,8 @@ impl Operation {
                 let old_snapshot = table_metadata.current_snapshot(branch.as_deref())?;
 
                 let snapshot_operation = match (data_files.len(), delete_files.len()) {
-                    (0, 0) => Err(Error::InvalidFormat(
-                        "Empty data and delete files".to_string(),
-                    )),
-                    (_, 0) => Ok(SnapshotOperation::Append),
+                    (0, 0) => return Ok((None, Vec::new())),
+                    (_, 0) => Ok::<_, Error>(SnapshotOperation::Append),
                     (0, _) => Ok(SnapshotOperation::Delete),
                     (_, _) => Ok(SnapshotOperation::Overwrite),
                 }?;
@@ -279,18 +277,17 @@ impl Operation {
                         (selected_manifest_opt, selected_manifest_bytes_opt)
                     {
                         let manifest_bytes = manifest_bytes.await??;
-                        let manifest_reader = ManifestReader::new(&*manifest_bytes)?
-                            .map(|entry| {
-                                let mut entry = entry?;
-                                *entry.status_mut() = Status::Existing;
-                                if entry.sequence_number().is_none() {
-                                    *entry.sequence_number_mut() = Some(manifest.sequence_number);
-                                }
-                                if entry.snapshot_id().is_none() {
-                                    *entry.snapshot_id_mut() = Some(manifest.added_snapshot_id);
-                                }
-                                Ok(entry)
-                            });
+                        let manifest_reader = ManifestReader::new(&*manifest_bytes)?.map(|entry| {
+                            let mut entry = entry?;
+                            *entry.status_mut() = Status::Existing;
+                            if entry.sequence_number().is_none() {
+                                *entry.sequence_number_mut() = Some(manifest.sequence_number);
+                            }
+                            if entry.snapshot_id().is_none() {
+                                *entry.snapshot_id_mut() = Some(manifest.added_snapshot_id);
+                            }
+                            Ok(entry)
+                        });
 
                         split_datafiles(
                             new_datafile_iter.chain(manifest_reader),
@@ -459,7 +456,8 @@ impl Operation {
                     let bounding_partition_values = files
                         .iter()
                         .try_fold(None, |acc, x| {
-                            let node = partition_struct_to_vec(x.partition(), &partition_column_names)?;
+                            let node =
+                                partition_struct_to_vec(x.partition(), &partition_column_names)?;
                             let Some(mut acc) = acc else {
                                 return Ok::<_, Error>(Some(Rectangle::new(node.clone(), node)));
                             };