feat: let a transaction carry a schema metadata update

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";

python/src/transaction.rs

@@ -824,6 +824,8 @@ impl FromPyObject<'_, '_> for PyLance<Transaction> {
             operation,
             tag: None,
             transaction_properties,
+            // Not surfaced to Python: no binding writes one today.
+            schema_metadata_updates: None,
         }))
     }
 }

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 {

rust/lance/src/dataset/transaction.rs

@@ -263,6 +263,10 @@ pub struct Transaction {
     pub operation: Operation,
     pub tag: Option<String>,
     pub transaction_properties: Option<Arc<HashMap<String, String>>>,
+    /// 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<UpdateMap>,
 }
 
 #[derive(Debug, Clone, DeepSizeOf, PartialEq)]
@@ -1760,6 +1764,7 @@ pub struct TransactionBuilder {
     operation: Operation,
     tag: Option<String>,
     transaction_properties: Option<Arc<HashMap<String, String>>>,
+    schema_metadata_updates: Option<UpdateMap>,
 }
 
 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<pb::Transaction> 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(),

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![],

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;