diff --git a/protos/transaction.proto b/protos/transaction.proto index bbde96fed5d..61acc31765c 100644 --- a/protos/transaction.proto +++ b/protos/transaction.proto @@ -396,6 +396,10 @@ message Transaction { DataOverlay data_overlay = 115; } + // A schema metadata update to write in the same commit as the operation + // above. Two transactions that both carry one conflict. + optional UpdateMap schema_metadata_updates = 5; + // Fields 200/202 (`blob_append` / `blob_overwrite`) previously represented blob dataset ops. reserved 200, 202; reserved "blob_append", "blob_overwrite"; diff --git a/python/src/transaction.rs b/python/src/transaction.rs index bf70cf3dd6d..1d9a88b06c1 100644 --- a/python/src/transaction.rs +++ b/python/src/transaction.rs @@ -824,6 +824,8 @@ impl FromPyObject<'_, '_> for PyLance { operation, tag: None, transaction_properties, + // Not surfaced to Python: no binding writes one today. + schema_metadata_updates: None, })) } } diff --git a/rust/lance/src/dataset/blob.rs b/rust/lance/src/dataset/blob.rs index 381c993747c..bd68163715c 100644 --- a/rust/lance/src/dataset/blob.rs +++ b/rust/lance/src/dataset/blob.rs @@ -5308,6 +5308,7 @@ mod tests { .await .unwrap(); let transaction = Transaction { + schema_metadata_updates: None, read_version: dataset.manifest.version, uuid: Uuid::new_v4().hyphenated().to_string(), operation: Operation::DataReplacement { @@ -5491,6 +5492,7 @@ mod tests { .await .unwrap(); let transaction = Transaction { + schema_metadata_updates: None, read_version: dataset.manifest.version, uuid: Uuid::new_v4().hyphenated().to_string(), operation: Operation::DataReplacement { diff --git a/rust/lance/src/dataset/transaction.rs b/rust/lance/src/dataset/transaction.rs index 1fd5ed15405..12bd33d5f10 100644 --- a/rust/lance/src/dataset/transaction.rs +++ b/rust/lance/src/dataset/transaction.rs @@ -263,6 +263,10 @@ pub struct Transaction { pub operation: Operation, pub tag: Option, pub transaction_properties: Option>>, + /// A schema metadata update to write in the same commit as + /// `operation`. Two transactions that both carry one conflict, so the + /// loser commits nothing. + pub schema_metadata_updates: Option, } #[derive(Debug, Clone, DeepSizeOf, PartialEq)] @@ -1760,6 +1764,7 @@ pub struct TransactionBuilder { operation: Operation, tag: Option, transaction_properties: Option>>, + schema_metadata_updates: Option, } impl TransactionBuilder { @@ -1770,9 +1775,16 @@ impl TransactionBuilder { operation, tag: None, transaction_properties: None, + schema_metadata_updates: None, } } + /// See [`Transaction::schema_metadata_updates`]. + pub fn schema_metadata_updates(mut self, updates: UpdateMap) -> Self { + self.schema_metadata_updates = Some(updates); + self + } + pub fn uuid(mut self, uuid: String) -> Self { self.uuid = Some(uuid); self @@ -1801,6 +1813,7 @@ impl TransactionBuilder { operation: self.operation, tag: self.tag, transaction_properties: self.transaction_properties, + schema_metadata_updates: self.schema_metadata_updates, } } } @@ -3337,6 +3350,13 @@ impl Transaction { manifest.transaction_file = Some(transaction_file_path.to_string()); + // Same manifest as the operation: the two land together or not at all. + if let Some(schema_metadata_updates) = &self.schema_metadata_updates { + let mut schema_metadata = manifest.schema.metadata.clone(); + apply_update_map(&mut schema_metadata, schema_metadata_updates); + manifest.schema.metadata = schema_metadata; + } + if let Some(next_row_id) = next_row_id { manifest.next_row_id = next_row_id; } @@ -4300,6 +4320,10 @@ impl TryFrom for Transaction { read_version: message.read_version, uuid: message.uuid.clone(), operation, + schema_metadata_updates: message + .schema_metadata_updates + .as_ref() + .map(UpdateMap::from), tag: if message.tag.is_empty() { None } else { @@ -4612,6 +4636,10 @@ impl From<&Transaction> for pb::Transaction { read_version: value.read_version, uuid: value.uuid.clone(), operation: Some(operation), + schema_metadata_updates: value + .schema_metadata_updates + .as_ref() + .map(pb::transaction::UpdateMap::from), tag: value.tag.clone().unwrap_or("".to_string()), transaction_properties, } @@ -6351,6 +6379,7 @@ mod tests { fn test_proto_legacy_field_9_read() { // Simulate a manifest written by old Lance: only field 9, no field 10. let pb_tx = pb::Transaction { + schema_metadata_updates: None, read_version: 1, uuid: "test".to_string(), tag: String::new(), @@ -6400,6 +6429,7 @@ mod tests { .unwrap(); let pb_tx = pb::Transaction { + schema_metadata_updates: None, read_version: 1, uuid: "test".to_string(), tag: String::new(), diff --git a/rust/lance/src/dataset/write/commit.rs b/rust/lance/src/dataset/write/commit.rs index 3af12bc4953..c3b46e90960 100644 --- a/rust/lance/src/dataset/write/commit.rs +++ b/rust/lance/src/dataset/write/commit.rs @@ -549,6 +549,10 @@ impl<'a> CommitBuilder<'a> { let merged = Transaction { uuid: uuid::Uuid::new_v4().hyphenated().to_string(), + // Merging is for plain appends; a transaction carrying schema + // metadata is rejected before it reaches here (see + // `can_merge_transactions`). + schema_metadata_updates: None, operation: Operation::Append { fragments: transactions .iter() @@ -623,6 +627,7 @@ mod tests { fn sample_transaction(read_version: u64) -> Transaction { Transaction { + schema_metadata_updates: None, uuid: uuid::Uuid::new_v4().hyphenated().to_string(), operation: Operation::Append { fragments: vec![sample_fragment()], @@ -1087,6 +1092,7 @@ mod tests { // Attempting to commit update gives error let update_transaction = Transaction { + schema_metadata_updates: None, uuid: uuid::Uuid::new_v4().hyphenated().to_string(), operation: Operation::Update { updated_fragments: vec![], diff --git a/rust/lance/src/io/commit/conflict_resolver.rs b/rust/lance/src/io/commit/conflict_resolver.rs index 99b5188f91b..9ad1e00a828 100644 --- a/rust/lance/src/io/commit/conflict_resolver.rs +++ b/rust/lance/src/io/commit/conflict_resolver.rs @@ -280,6 +280,14 @@ impl<'a> TransactionRebase<'a> { return Err(self.retryable_conflict_err(other_transaction, other_version)); } + // Both planned from state the other replaced; rejecting here keeps + // the loser's data from landing. + if self.transaction.schema_metadata_updates.is_some() + && other_transaction.schema_metadata_updates.is_some() + { + return Err(self.incompatible_conflict_err(other_transaction, other_version)); + } + let op = &self.transaction.operation; match op { Operation::Delete { .. } => self.check_delete_txn(other_transaction, other_version), @@ -4281,6 +4289,48 @@ mod tests { ); } + // Two transactions carrying a schema metadata update conflict whatever + // their operations; one on a single side does not. + #[tokio::test] + async fn test_schema_metadata_stamp_conflicts_with_another_stamp() { + let dataset = test_dataset(10, 2).await; + + use crate::dataset::transaction::{TransactionBuilder, UpdateMap, UpdateMapEntry}; + + let stamp = |value: &str| UpdateMap { + update_entries: vec![UpdateMapEntry::from(("mv.source_version", value))], + replace: false, + }; + let stamped_append = |value: &str| { + TransactionBuilder::new(1, Operation::Append { fragments: vec![] }) + .schema_metadata_updates(stamp(value)) + .build() + }; + + // Two stamped appends: one is rejected. + let mut resolver = TransactionRebase::try_new(&dataset, stamped_append("7"), None) + .await + .unwrap(); + let err = resolver + .check_txn(&stamped_append("8"), 2) + .expect_err("two stamped transactions must conflict"); + assert!( + matches!(err, Error::IncompatibleTransaction { .. }), + "{err:?}" + ); + + // One stamp only: still compatible. + let mut resolver = TransactionRebase::try_new(&dataset, stamped_append("7"), None) + .await + .unwrap(); + resolver + .check_txn( + &Transaction::new_from_version(1, Operation::Append { fragments: vec![] }), + 2, + ) + .expect("an unstamped append must stay compatible"); + } + #[tokio::test] async fn test_add_bases_no_conflict_with_data_operations() { let dataset = test_dataset(10, 2).await;