feat: CommitInfo sets a txnId (#1262)

MannDP · web-flow · commit 67aad5c4a5fd · 2025-09-08T16:51:21.000-07:00
Generate a txnId stored inside the commitInfo metadata of a Delta commit. - resolves #1149
diff --git a/ffi/src/transaction/mod.rs b/ffi/src/transaction/mod.rs
@@ -157,10 +157,18 @@ mod tests {
 
     use std::sync::Arc;
 
+    const ZERO_UUID: &str = "00000000-0000-0000-0000-000000000000";
+
     use super::*;
 
     use tempfile::tempdir;
 
+    fn check_txn_id_exists(commit_info: &serde_json::Value) {
+        commit_info["txnId"]
+            .as_str()
+            .expect("txnId should be present in commitInfo");
+    }
+
     fn create_arrow_ffi_from_json(
         schema: ArrowSchema,
         json_string: &str,
@@ -341,9 +349,12 @@ mod tests {
                 .into_iter::<serde_json::Value>()
                 .try_collect()?;
 
-            // set timestamps to 0 and paths to known string values for comparison
-            // (otherwise timestamps are non-deterministic and paths are random UUIDs)
+            check_txn_id_exists(&parsed_commits[0]["commitInfo"]);
+
+            // set timestamps to 0, paths and txn_id to known string values for comparison
+            // (otherwise timestamps are non-deterministic, paths and txn_id are random UUIDs)
             set_json_value(&mut parsed_commits[0], "commitInfo.timestamp", json!(0))?;
+            set_json_value(&mut parsed_commits[0], "commitInfo.txnId", json!(ZERO_UUID))?;
             set_json_value(&mut parsed_commits[1], "add.modificationTime", json!(0))?;
             set_json_value(&mut parsed_commits[1], "add.size", json!(0))?;
 
@@ -355,6 +366,7 @@ mod tests {
                         "operation": "UNKNOWN",
                         "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                         "operationParameters": {},
+                        "txnId": ZERO_UUID
                     }
                 }),
                 json!({
diff --git a/kernel/src/actions/mod.rs b/kernel/src/actions/mod.rs
@@ -584,8 +584,7 @@ pub(crate) struct CommitInfo {
     pub(crate) kernel_version: Option<String>,
     /// A place for the engine to store additional metadata associated with this commit
     pub(crate) engine_info: Option<String>,
-    /// A unique transaction identified for this commit. When the `catalogManaged` table feature is
-    /// enabled (not yet implemented), this field will be required. Otherwise, it is optional.
+    /// A unique transaction identifier for this commit.
     pub(crate) txn_id: Option<String>,
 }
 
@@ -602,7 +601,7 @@ impl CommitInfo {
             operation_parameters: None,
             kernel_version: Some(format!("v{KERNEL_VERSION}")),
             engine_info,
-            txn_id: None,
+            txn_id: Some(uuid::Uuid::new_v4().to_string()),
         }
     }
 }
@@ -1393,6 +1392,7 @@ mod tests {
         let engine = ExprEngine::new();
 
         let commit_info = CommitInfo::new(0, None, None);
+        let commit_info_txn_id = commit_info.txn_id.clone();
 
         let engine_data = commit_info.into_engine_data(CommitInfo::to_schema().into(), &engine);
 
@@ -1416,7 +1416,7 @@ mod tests {
                 operation_parameters,
                 Arc::new(StringArray::from(vec![Some(format!("v{KERNEL_VERSION}"))])),
                 Arc::new(StringArray::from(vec![None::<String>])),
-                Arc::new(StringArray::from(vec![None::<String>])),
+                Arc::new(StringArray::from(vec![commit_info_txn_id])),
             ],
         )
         .unwrap();
diff --git a/kernel/tests/write.rs b/kernel/tests/write.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use delta_kernel::Error as KernelError;
 use delta_kernel::{DeltaResult, Engine, Snapshot, Version};
+use uuid::Uuid;
 
 use delta_kernel::arrow::array::{ArrayRef, BinaryArray, StructArray};
 use delta_kernel::arrow::array::{Int32Array, StringArray, TimestampMicrosecondArray};
@@ -35,6 +36,15 @@ use test_utils::{create_table, engine_store_setup, setup_test_tables, test_read}
 mod common;
 use url::Url;
 
+fn validate_txn_id(commit_info: &serde_json::Value) {
+    let txn_id = commit_info["txnId"]
+        .as_str()
+        .expect("txnId should be present in commitInfo");
+    Uuid::parse_str(txn_id).expect("txnId should be valid UUID format");
+}
+
+const ZERO_UUID: &str = "00000000-0000-0000-0000-000000000000";
+
 #[tokio::test]
 async fn test_commit_info() -> Result<(), Box<dyn std::error::Error>> {
     // setup tracing
@@ -63,11 +73,11 @@ async fn test_commit_info() -> Result<(), Box<dyn std::error::Error>> {
             .await?;
 
         let mut parsed_commit: serde_json::Value = serde_json::from_slice(&commit1.bytes().await?)?;
-        *parsed_commit
-            .get_mut("commitInfo")
-            .unwrap()
-            .get_mut("timestamp")
-            .unwrap() = serde_json::Value::Number(0.into());
+
+        validate_txn_id(&parsed_commit["commitInfo"]);
+
+        set_json_value(&mut parsed_commit, "commitInfo.timestamp", json!(0))?;
+        set_json_value(&mut parsed_commit, "commitInfo.txnId", json!(ZERO_UUID))?;
 
         let expected_commit = json!({
             "commitInfo": {
@@ -76,6 +86,7 @@ async fn test_commit_info() -> Result<(), Box<dyn std::error::Error>> {
                 "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                 "operationParameters": {},
                 "engineInfo": "default engine",
+                "txnId": ZERO_UUID,
             }
         });
 
@@ -217,9 +228,12 @@ async fn test_commit_info_action() -> Result<(), Box<dyn std::error::Error>> {
             .into_iter::<serde_json::Value>()
             .try_collect()?;
 
-        // set timestamps to 0 and paths to known string values for comparison
-        // (otherwise timestamps are non-deterministic and paths are random UUIDs)
+        validate_txn_id(&parsed_commits[0]["commitInfo"]);
+
+        // set timestamps to 0, paths and txn_id to known string values for comparison
+        // (otherwise timestamps are non-deterministic, paths and txn_id are random UUIDs)
         set_json_value(&mut parsed_commits[0], "commitInfo.timestamp", json!(0))?;
+        set_json_value(&mut parsed_commits[0], "commitInfo.txnId", json!(ZERO_UUID))?;
 
         let expected_commit = vec![json!({
             "commitInfo": {
@@ -228,6 +242,7 @@ async fn test_commit_info_action() -> Result<(), Box<dyn std::error::Error>> {
                 "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                 "operationParameters": {},
                 "engineInfo": "default engine",
+                "txnId": ZERO_UUID
             }
         })];
 
@@ -270,10 +285,13 @@ async fn test_append() -> Result<(), Box<dyn std::error::Error>> {
         // check that the timestamps in commit_info and add actions are within 10s of SystemTime::now()
         // before we clear them for comparison
         check_action_timestamps(parsed_commits.iter())?;
+        // check that the txn_id is valid before we clear it for comparison
+        validate_txn_id(&parsed_commits[0]["commitInfo"]);
 
-        // set timestamps to 0 and paths to known string values for comparison
-        // (otherwise timestamps are non-deterministic and paths are random UUIDs)
+        // set timestamps to 0, paths and txn_id to known string values for comparison
+        // (otherwise timestamps are non-deterministic, paths and txn_id are random UUIDs)
         set_json_value(&mut parsed_commits[0], "commitInfo.timestamp", json!(0))?;
+        set_json_value(&mut parsed_commits[0], "commitInfo.txnId", json!(ZERO_UUID))?;
         set_json_value(&mut parsed_commits[1], "add.modificationTime", json!(0))?;
         set_json_value(&mut parsed_commits[1], "add.path", json!("first.parquet"))?;
         set_json_value(&mut parsed_commits[2], "add.modificationTime", json!(0))?;
@@ -286,6 +304,7 @@ async fn test_append() -> Result<(), Box<dyn std::error::Error>> {
                     "operation": "UNKNOWN",
                     "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                     "operationParameters": {},
+                    "txnId": ZERO_UUID
                 }
             }),
             json!({
@@ -437,10 +456,13 @@ async fn test_append_partitioned() -> Result<(), Box<dyn std::error::Error>> {
         // check that the timestamps in commit_info and add actions are within 10s of SystemTime::now()
         // before we clear them for comparison
         check_action_timestamps(parsed_commits.iter())?;
+        // check that the txn_id is valid before we clear it for comparison
+        validate_txn_id(&parsed_commits[0]["commitInfo"]);
 
-        // set timestamps to 0 and paths to known string values for comparison
-        // (otherwise timestamps are non-deterministic and paths are random UUIDs)
+        // set timestamps to 0, paths and txn_id to known string values for comparison
+        // (otherwise timestamps are non-deterministic, paths and txn_id are random UUIDs)
         set_json_value(&mut parsed_commits[0], "commitInfo.timestamp", json!(0))?;
+        set_json_value(&mut parsed_commits[0], "commitInfo.txnId", json!(ZERO_UUID))?;
         set_json_value(&mut parsed_commits[1], "add.modificationTime", json!(0))?;
         set_json_value(&mut parsed_commits[1], "add.path", json!("first.parquet"))?;
         set_json_value(&mut parsed_commits[2], "add.modificationTime", json!(0))?;
@@ -454,6 +476,7 @@ async fn test_append_partitioned() -> Result<(), Box<dyn std::error::Error>> {
                     "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                     "operationParameters": {},
                     "engineInfo": "default engine",
+                    "txnId": ZERO_UUID
                 }
             }),
             json!({
@@ -620,11 +643,7 @@ async fn test_write_txn_actions() -> Result<(), Box<dyn std::error::Error>> {
             .into_iter::<serde_json::Value>()
             .try_collect()?;
 
-        *parsed_commits[0]
-            .get_mut("commitInfo")
-            .unwrap()
-            .get_mut("timestamp")
-            .unwrap() = serde_json::Value::Number(0.into());
+        set_json_value(&mut parsed_commits[0], "commitInfo.timestamp", json!(0)).unwrap();
 
         let time_ms: i64 = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)?
@@ -663,6 +682,10 @@ async fn test_write_txn_actions() -> Result<(), Box<dyn std::error::Error>> {
         assert!((last_updated.as_i64().unwrap() - time_ms).abs() < 10_000);
         *last_updated = serde_json::Value::Number(2.into());
 
+        validate_txn_id(&parsed_commits[0]["commitInfo"]);
+
+        set_json_value(&mut parsed_commits[0], "commitInfo.txnId", json!(ZERO_UUID))?;
+
         let expected_commit = vec![
             json!({
                 "commitInfo": {
@@ -671,6 +694,7 @@ async fn test_write_txn_actions() -> Result<(), Box<dyn std::error::Error>> {
                     "kernelVersion": format!("v{}", env!("CARGO_PKG_VERSION")),
                     "operationParameters": {},
                     "engineInfo": "default engine",
+                    "txnId": ZERO_UUID
                 }
             }),
             json!({

Original file line number	Diff line number	Diff line change
`@@ -584,8 +584,7 @@ pub(crate) struct CommitInfo {`
`584`	`584`	`pub(crate) kernel_version: Option<String>,`
`585`	`585`	`/// A place for the engine to store additional metadata associated with this commit`
`586`	`586`	`pub(crate) engine_info: Option<String>,`
`587`		- /// A unique transaction identified for this commit. When the `catalogManaged` table feature is
`588`		`- /// enabled (not yet implemented), this field will be required. Otherwise, it is optional.`
	`587`	`+ /// A unique transaction identifier for this commit.`
`589`	`588`	`pub(crate) txn_id: Option<String>,`
`590`	`589`	`}`
`591`	`590`
`@@ -602,7 +601,7 @@ impl CommitInfo {`
`602`	`601`	`operation_parameters: None,`
`603`	`602`	`kernel_version: Some(format!("v{KERNEL_VERSION}")),`
`604`	`603`	`engine_info,`
`605`		`- txn_id: None,`
	`604`	`+ txn_id: Some(uuid::Uuid::new_v4().to_string()),`
`606`	`605`	`}`
`607`	`606`	`}`
`608`	`607`	`}`
`@@ -1393,6 +1392,7 @@ mod tests {`
`1393`	`1392`	`let engine = ExprEngine::new();`
`1394`	`1393`
`1395`	`1394`	`let commit_info = CommitInfo::new(0, None, None);`
	`1395`	`+ let commit_info_txn_id = commit_info.txn_id.clone();`
`1396`	`1396`
`1397`	`1397`	`let engine_data = commit_info.into_engine_data(CommitInfo::to_schema().into(), &engine);`
`1398`	`1398`
`@@ -1416,7 +1416,7 @@ mod tests {`
`1416`	`1416`	`operation_parameters,`
`1417`	`1417`	`Arc::new(StringArray::from(vec![Some(format!("v{KERNEL_VERSION}"))])),`
`1418`	`1418`	`Arc::new(StringArray::from(vec![None::<String>])),`
`1419`		`- Arc::new(StringArray::from(vec![None::<String>])),`
	`1419`	`+ Arc::new(StringArray::from(vec![commit_info_txn_id])),`
`1420`	`1420`	`],`
`1421`	`1421`	`)`
`1422`	`1422`	`.unwrap();`