diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index bb4d350e5a..451fe6bcb0 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -3309,6 +3309,7 @@ pub fn iceberg::transaction::FastAppendAction::add_data_files(self, data_files: pub fn iceberg::transaction::FastAppendAction::set_commit_uuid(self, commit_uuid: uuid::Uuid) -> Self pub fn iceberg::transaction::FastAppendAction::set_snapshot_properties(self, snapshot_properties: std::collections::hash::map::HashMap) -> Self pub fn iceberg::transaction::FastAppendAction::with_check_duplicate(self, v: bool) -> Self +pub fn iceberg::transaction::FastAppendAction::with_idempotency_key(self, key: impl core::convert::Into) -> Self pub struct iceberg::transaction::ReplaceSortOrderAction impl iceberg::transaction::ReplaceSortOrderAction pub fn iceberg::transaction::ReplaceSortOrderAction::asc(self, name: &str, null_order: iceberg::spec::NullOrder) -> Self @@ -3347,6 +3348,7 @@ pub fn iceberg::transaction::UpdateStatisticsAction::set_statistics(self, statis pub struct iceberg::transaction::UpgradeFormatVersionAction impl iceberg::transaction::UpgradeFormatVersionAction pub fn iceberg::transaction::UpgradeFormatVersionAction::set_format_version(self, format_version: iceberg::spec::FormatVersion) -> Self +pub const iceberg::transaction::IDEMPOTENCY_KEY_SUMMARY_PROPERTY: &str pub trait iceberg::transaction::ApplyTransactionAction pub fn iceberg::transaction::ApplyTransactionAction::apply(self, tx: iceberg::transaction::Transaction) -> iceberg::Result impl iceberg::transaction::ApplyTransactionAction for T diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 3260f9ffa2..a75c97c574 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -22,18 +22,37 @@ use async_trait::async_trait; use uuid::Uuid; use crate::error::Result; -use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation, SnapshotRef}; use crate::table::Table; use crate::transaction::snapshot::{ DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, }; use crate::transaction::{ActionCommit, TransactionAction}; +use crate::util::snapshot::ancestors_of; + +/// Snapshot summary property that records the idempotency key of a fast append. +/// See [`FastAppendAction::with_idempotency_key`]. +pub const IDEMPOTENCY_KEY_SUMMARY_PROPERTY: &str = "iceberg-rust.idempotency-key"; + +/// The snapshot on the `main` branch of `table` whose summary records +/// idempotency key `key`, if any. +fn snapshot_with_idempotency_key(table: &Table, key: &str) -> Option { + let head = table.metadata().current_snapshot_id()?; + ancestors_of(&table.metadata_ref(), head).find(|snapshot| { + snapshot + .summary() + .additional_properties + .get(IDEMPOTENCY_KEY_SUMMARY_PROPERTY) + .is_some_and(|v| v == key) + }) +} /// FastAppendAction is a transaction action for fast append data files to the table. pub struct FastAppendAction { check_duplicate: bool, // below are properties used to create SnapshotProducer when commit commit_uuid: Option, + idempotency_key: Option, snapshot_properties: HashMap, added_data_files: Vec, } @@ -43,6 +62,7 @@ impl FastAppendAction { Self { check_duplicate: true, commit_uuid: None, + idempotency_key: None, snapshot_properties: HashMap::default(), added_data_files: vec![], } @@ -66,6 +86,24 @@ impl FastAppendAction { self } + /// Makes this append idempotent: if a snapshot on `main` already records + /// `key`, the append commits nothing. + /// + /// Use one key per logical write: an append whose key is already committed + /// is dropped whatever its files, and the commit returns `Ok`. A rerun + /// appending the files the first attempt committed is skipped, not + /// rejected as a duplicate, so after a skip delete only files the table + /// does not reference. + /// + /// The key is checked on every commit attempt, so of two racing attempts + /// only one commits. An append with no data files and no snapshot + /// properties commits nothing. Keys on snapshots expired or rolled back off + /// `main` are not found. + pub fn with_idempotency_key(mut self, key: impl Into) -> Self { + self.idempotency_key = Some(key.into()); + self + } + /// Set snapshot summary properties. pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap) -> Self { self.snapshot_properties = snapshot_properties; @@ -83,15 +121,42 @@ impl FastAppendAction { .cloned() .collect() } + + /// The snapshot summary properties set by the caller, plus the idempotency + /// key. + fn summary_properties(&self) -> HashMap { + let mut properties = self.snapshot_properties.clone(); + if let Some(key) = &self.idempotency_key { + properties.insert(IDEMPOTENCY_KEY_SUMMARY_PROPERTY.to_string(), key.clone()); + } + properties + } } #[async_trait] impl TransactionAction for FastAppendAction { async fn commit(self: Arc, table: &Table) -> Result { + if self.idempotency_key.is_some() + && self.added_data_files.is_empty() + && self.snapshot_properties.is_empty() + { + return Ok(ActionCommit::new(vec![], vec![])); + } + + if let Some(key) = self.idempotency_key.as_deref() + && let Some(snapshot) = snapshot_with_idempotency_key(table, key) + { + tracing::info!( + "Skipping fast append: idempotency key '{key}' was already committed in snapshot {}", + snapshot.snapshot_id() + ); + return Ok(ActionCommit::new(vec![], vec![])); + } + let snapshot_producer = SnapshotProducer::new( table, self.commit_uuid.unwrap_or_else(Uuid::now_v7), - self.snapshot_properties.clone(), + self.summary_properties(), self.dedupe_added_files(), ); @@ -952,3 +1017,214 @@ mod tests { assert_eq!(data_file, *manifest.entries()[0].data_file()); } } + +#[cfg(test)] +mod idempotency_key_tests { + use std::collections::HashMap; + + use mockall::Sequence; + + use super::{IDEMPOTENCY_KEY_SUMMARY_PROPERTY, snapshot_with_idempotency_key}; + use crate::Catalog; + use crate::catalog::MockCatalog; + use crate::memory::tests::new_memory_catalog; + use crate::spec::{ + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, Struct, TableMetadata, + }; + use crate::table::Table; + use crate::transaction::tests::make_v3_minimal_table_in_catalog; + use crate::transaction::{ApplyTransactionAction, Transaction}; + + fn data_file(path: &str) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(10) + .partition(Struct::from_iter([Some(Literal::long(1))])) + .partition_spec_id(0) + .build() + .unwrap() + } + + /// Fast-appends a data file at each of `paths` to `table` under + /// idempotency key `key`, and returns the table the commit leaves. + async fn append(catalog: &dyn Catalog, table: &Table, paths: &[&str], key: &str) -> Table { + let tx = Transaction::new(table); + tx.fast_append() + .add_data_files(paths.iter().map(|path| data_file(path))) + .with_idempotency_key(key) + .apply(tx) + .unwrap() + .commit(catalog) + .await + .unwrap() + } + + /// What identifies a version of a table: its metadata and where that is + /// stored. A commit that writes nothing leaves both unchanged. + fn version(table: &Table) -> (&TableMetadata, Option<&str>) { + (table.metadata(), table.metadata_location()) + } + + #[tokio::test] + async fn test_rerun_from_a_stale_table_commits_nothing() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + let committed = append(&catalog, &base, &["a.parquet"], "job-1").await; + + // The rerun starts from `base`, which predates the first commit. + let rerun = append(&catalog, &base, &["a-rerun.parquet"], "job-1").await; + + assert_eq!(version(&rerun), version(&committed)); + } + + /// Rerunning only the commit appends the files the first run committed. + #[tokio::test] + async fn test_rerun_of_the_same_files_commits_nothing() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + let committed = append(&catalog, &base, &["a.parquet"], "job-1").await; + + let rerun = append(&catalog, &committed, &["a.parquet"], "job-1").await; + + assert_eq!(version(&rerun), version(&committed)); + } + + #[tokio::test] + async fn test_append_without_files_commits_nothing() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + + let appended = append(&catalog, &base, &[], "job-1").await; + + assert_eq!(version(&appended), version(&base)); + } + + #[tokio::test] + async fn test_append_without_files_but_with_properties_commits() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + + let tx = Transaction::new(&base); + let table = tx + .fast_append() + .set_snapshot_properties(HashMap::from([("p".to_string(), "v".to_string())])) + .with_idempotency_key("job-1") + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + + let properties = &table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .additional_properties; + assert_eq!( + ( + properties.get("p").map(String::as_str), + properties + .get(IDEMPOTENCY_KEY_SUMMARY_PROPERTY) + .map(String::as_str) + ), + (Some("v"), Some("job-1")) + ); + } + + #[tokio::test] + async fn test_skipped_append_keeps_the_other_updates() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + let committed = append(&catalog, &base, &["a.parquet"], "job-1").await; + + let tx = Transaction::new(&committed); + let tx = tx + .update_table_properties() + .set("k".to_string(), "v".to_string()) + .apply(tx) + .unwrap(); + let tx = tx + .fast_append() + .add_data_files([data_file("a-rerun.parquet")]) + .with_idempotency_key("job-1") + .apply(tx) + .unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let snapshot_ids = |table: &Table| { + let mut ids: Vec<_> = table + .metadata() + .snapshots() + .map(|s| s.snapshot_id()) + .collect(); + ids.sort(); + ids + }; + assert_eq!( + table.metadata().properties(), + &HashMap::from([("k".to_string(), "v".to_string())]) + ); + assert_eq!(snapshot_ids(&table), snapshot_ids(&committed)); + } + + /// Two runs race: the loser loads the table before the winner commits and + /// commits after it. + #[tokio::test] + async fn test_loser_of_a_race_commits_nothing() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + let winner = append(&catalog, &base, &["winner.parquet"], "job-1").await; + + // What the loser sees of the catalog, in order. + let mut loser_view = MockCatalog::new(); + let mut seq = Sequence::new(); + let (before, current, after) = (base.clone(), winner.clone(), winner.clone()); + loser_view + .expect_load_table() + .times(1) + .in_sequence(&mut seq) + .return_once_st(|_| Box::pin(async move { Ok(before) })); + loser_view + .expect_update_table() + .times(1) + .in_sequence(&mut seq) + .return_once_st(|commit| Box::pin(async move { commit.apply(current) })); + loser_view + .expect_load_table() + .times(1) + .in_sequence(&mut seq) + .return_once_st(|_| Box::pin(async move { Ok(after) })); + + let loser = append(&loser_view, &base, &["loser.parquet"], "job-1").await; + + assert_eq!(version(&loser), version(&winner)); + } + + #[tokio::test] + async fn test_snapshot_with_idempotency_key_finds_each_key() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + let first = append(&catalog, &base, &["a.parquet"], "job-1").await; + let second = append(&catalog, &first, &["b.parquet"], "job-2").await; + + let snapshot_id = + |key| snapshot_with_idempotency_key(&second, key).map(|s| s.snapshot_id()); + assert_eq!( + [ + snapshot_id("job-1"), + snapshot_id("job-2"), + snapshot_id("job-3") + ], + [ + first.metadata().current_snapshot_id(), + second.metadata().current_snapshot_id(), + None, + ] + ); + assert_eq!(snapshot_with_idempotency_key(&base, "job-1"), None); + } +} diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 3622471277..2b3e2d24ce 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -73,7 +73,7 @@ use crate::error::Result; use crate::spec::TableProperties; use crate::table::Table; use crate::transaction::action::BoxedTransactionAction; -pub use crate::transaction::append::FastAppendAction; +pub use crate::transaction::append::{FastAppendAction, IDEMPOTENCY_KEY_SUMMARY_PROPERTY}; pub use crate::transaction::expire_snapshots::ExpireSnapshotsAction; pub use crate::transaction::sort_order::ReplaceSortOrderAction; pub use crate::transaction::update_location::UpdateLocationAction; @@ -232,6 +232,11 @@ impl Transaction { )?; } + if existing_updates.is_empty() && existing_requirements.is_empty() { + // every action was a no-op (e.g. an idempotent append already committed) + return Ok(self.table.clone()); + } + let table_commit = TableCommit::builder() .ident(self.table.identifier().to_owned()) .updates(existing_updates) @@ -568,6 +573,27 @@ mod tests { assert_eq!(summary.get("total-files-size").unwrap(), "300"); } + #[tokio::test] + async fn test_commit_without_updates_writes_nothing() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + // The table has no snapshots, so there is nothing to expire. + let tx = Transaction::new(&table); + let committed = tx + .expire_snapshots() + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + + assert_eq!( + (committed.metadata(), committed.metadata_location()), + (table.metadata(), table.metadata_location()) + ); + } + #[tokio::test] async fn test_commit_to_encrypted_table() { let table = make_encrypted_table().await.with_metadata_location(