diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index f8e69c4c86..eaca57fc16 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -48,6 +48,7 @@ const TOTAL_RECORDS: &str = "total-records"; const TOTAL_FILE_SIZE: &str = "total-files-size"; const CHANGED_PARTITION_COUNT_PROP: &str = "changed-partition-count"; const CHANGED_PARTITION_PREFIX: &str = "partitions."; +const ICEBERG_VERSION_PROP: &str = "iceberg-version"; /// `SnapshotSummaryCollector` collects and aggregates snapshot update metrics. /// It gathers metrics about added or removed data files and manifests, and tracks @@ -330,6 +331,14 @@ where T: PartialOrd + Default + ToString { } } +pub(crate) fn set_iceberg_version(properties: &mut HashMap) { + let version = env!("CARGO_PKG_VERSION"); + properties.insert( + ICEBERG_VERSION_PROP.to_string(), + format!("Apache Iceberg Rust {version}"), + ); +} + pub(crate) fn update_snapshot_summaries( summary: Summary, previous_summary: Option<&Summary>, diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 3260f9ffa2..7369183938 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -764,6 +764,37 @@ mod tests { ); } + #[tokio::test] + async fn test_snapshot_summary_records_writer_identity() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + + let snapshot_properties = HashMap::from([ + ("key".to_string(), "val".to_string()), + ("iceberg-version".to_string(), "caller".to_string()), + ]); + + let action = tx + .fast_append() + .set_snapshot_properties(snapshot_properties); + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + snapshot + } else { + unreachable!() + }; + let props = &new_snapshot.summary().additional_properties; + let version = env!("CARGO_PKG_VERSION"); + + assert_eq!( + props.get("iceberg-version").unwrap(), + &format!("Apache Iceberg Rust {version}") + ); + assert_eq!(props.get("key").unwrap(), "val"); + } + #[tokio::test] async fn test_append_snapshot_properties() { let table = make_v2_minimal_table(); diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index a0ce1fba14..656a580ce8 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -28,12 +28,11 @@ use crate::spec::{ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestEntry, ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, Snapshot, SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, StructType, Summary, - TableProperties, update_snapshot_summaries, + TableProperties, set_iceberg_version, update_snapshot_summaries, }; use crate::table::Table; use crate::transaction::ActionCommit; use crate::{Error, ErrorKind, TableRequirement, TableUpdate}; - /// A trait that defines how different table operations produce new snapshots. /// /// `SnapshotProduceOperation` is used by [`SnapshotProducer`] to customize snapshot creation @@ -411,11 +410,13 @@ impl<'a> SnapshotProducer<'a> { additional_properties, }; - update_snapshot_summaries( + let mut summary = update_snapshot_summaries( summary, previous_snapshot.map(|s| s.summary()), snapshot_produce_operation.operation() == Operation::Overwrite, - ) + )?; + set_iceberg_version(&mut summary.additional_properties); + Ok(summary) } fn generate_manifest_list_file_path(&self, attempt: i64) -> Result {