Merge pull request #170 from Embucket/aosipov/append_operation_summary

Append operation summary

Author: JanKaul

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
     }