From aa34612a622ec3283ec1a3db0989fa595810d187 Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Sat, 15 Aug 2026 07:43:00 -0700 Subject: [PATCH] feat: let a transaction carry a schema metadata update A transaction may now carry a schema metadata update alongside its operation, applied to the same manifest, and two transactions that both carry one conflict whatever their operations, so the loser is rejected before anything lands. This matches how two UpdateConfig transactions already treat schema metadata. An update on a single side is not a conflict: such an append rebases over an unrelated index build. The proto field is additive: an older reader applies such a transaction's data without its metadata. Co-Authored-By: Claude Fable 5 --- protos/transaction.proto | 4 ++ python/src/transaction.rs | 2 + rust/lance/src/dataset/blob.rs | 2 + rust/lance/src/dataset/transaction.rs | 30 +++++++++++ rust/lance/src/dataset/write/commit.rs | 6 +++ rust/lance/src/io/commit/conflict_resolver.rs | 50 +++++++++++++++++++ 6 files changed, 94 insertions(+) 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;