Implement parsing for the new domainMetadata actions in the commit log

The Delta Lake protocol which will be released in conjunction with "3.0.0"
(currently at RC1) introduces `domainMetadata` actions to the commit log to
enable system or user-provided metadata about the commits to be added to the
log. With DBR 13.3 in the Databricks ecosystem, tables are already being written
with this action via the "liquid clustering" feature.

This change enables the clean reading of these tables, but at present nothing
novel is done with this information.

[Read more here](https://www.databricks.com/blog/announcing-delta-lake-30-new-universal-format-and-liquid-clustering)

Fixes #1626

Sponsored-by: Databricks Inc

rust/src/action/mod.rs

@@ -681,6 +681,18 @@ pub struct CommitInfo {
     pub info: Map<String, serde_json::Value>,
 }
 
+/// The domain metadata action contains a configuration (string) for a named metadata domain
+#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct DomainMetaData {
+    /// Identifier for this domain (system or user-provided)
+    pub domain: String,
+    /// String containing configuration for the metadata domain
+    pub configuration: String,
+    /// When `true` the action serves as a tombstone
+    pub removed: bool,
+}
+
 /// Represents an action in the Delta log. The Delta log is an aggregate of all actions performed
 /// on the table, so the full list of actions is required to properly read a table.
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -701,6 +713,8 @@ pub enum Action {
     protocol(Protocol),
     /// Describes commit provenance information for the table.
     commitInfo(CommitInfo),
+    /// Describe s the configuration for a named metadata domain
+    domainMetadata(DomainMetaData),
 }
 
 impl Action {
@@ -1163,6 +1177,13 @@ mod tests {
         assert!(info.info.contains_key("additionalStruct"));
     }
 
+    #[test]
+    fn test_read_domain_metadata() {
+        let buf = r#"{"domainMetadata":{"domain":"delta.liquid","configuration":"{\"clusteringColumns\":[{\"physicalName\":[\"id\"]}],\"domainName\":\"delta.liquid\"}","removed":false}}"#;
+        let _action: Action =
+            serde_json::from_str(buf).expect("Expected to be able to deserialize");
+    }
+
     #[cfg(feature = "arrow")]
     mod arrow_tests {
         use arrow::array::{self, ArrayRef, StructArray};

rust/src/table_state.rs

@@ -35,6 +35,8 @@ pub struct DeltaTableState {
     files: Vec<action::Add>,
     // Information added to individual commits
     commit_infos: Vec<action::CommitInfo>,
+    // Domain metadatas provided by the system or user
+    domain_metadatas: Vec<action::DomainMetaData>,
     app_transaction_version: HashMap<String, i64>,
     min_reader_version: i32,
     min_writer_version: i32,
@@ -357,6 +359,9 @@ impl DeltaTableState {
             action::Action::commitInfo(v) => {
                 self.commit_infos.push(v);
             }
+            action::Action::domainMetadata(v) => {
+                self.domain_metadatas.push(v);
+            }
         }
 
         Ok(())
@@ -412,6 +417,7 @@ mod tests {
             tombstones: Default::default(),
             files: vec![],
             commit_infos: vec![],
+            domain_metadatas: vec![],
             app_transaction_version: Default::default(),
             min_reader_version: 0,
             min_writer_version: 0,
@@ -435,6 +441,7 @@ mod tests {
             version: -1,
             files: vec![],
             commit_infos: vec![],
+            domain_metadatas: vec![],
             tombstones: HashSet::new(),
             current_metadata: None,
             min_reader_version: 1,

rust/tests/data/table_with_liquid_clustering/_delta_log/00000000000000000000.crc

@@ -0,0 +1 @@
+{"txnId":"29ebf587-9705-4bb4-ac40-2f02324065c8","tableSizeBytes":5259,"numFiles":10,"numDeletedRecordsOpt":0,"numDeletionVectorsOpt":0,"numMetadata":1,"numProtocol":1,"setTransactions":[],"domainMetadata":[{"domain":"delta.liquid","configuration":"{\"clusteringColumns\":[{\"physicalName\":[\"id\"]}],\"domainName\":\"delta.liquid\"}","removed":false},{"domain":"delta.rowTracking","configuration":"{\"rowIdHighWaterMark\":9,\"domainName\":\"delta.rowTracking\"}","removed":false}],"metadata":{"id":"6ca3020b-3cd9-4048-82e3-1417a0abb98f","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.liquid.clusteringColumns":"id","delta.enableDeletionVectors":"true","delta.checkpoint.writeStatsAsStruct":"true","delta.enableRowTracking":"true","delta.checkpoint.writeStatsAsJson":"false","delta.rowTracking.materializedRowCommitVersionColumnName":"_row-commit-version-col-2f60dcc1-9e36-4424-95e7-799b707e4ddb","delta.rowTracking.materializedRowIdColumnName":"_row-id-col-4cbc7924-f662-4db1-aa59-22c23f59eb5d"},"createdTime":1694758256009},"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["domainMetadata","deletionVectors","rowTracking","liquid"]},"histogramOpt":{"sortedBinBoundaries":[0,8192,16384,32768,65536,131072,262144,524288,1048576,2097152,4194304,8388608,12582912,16777216,20971520,25165824,29360128,33554432,37748736,41943040,50331648,58720256,67108864,75497472,83886080,92274688,100663296,109051904,117440512,125829120,130023424,134217728,138412032,142606336,146800640,150994944,167772160,184549376,201326592,218103808,234881024,251658240,268435456,285212672,301989888,318767104,335544320,352321536,369098752,385875968,402653184,419430400,436207616,452984832,469762048,486539264,503316480,520093696,536870912,553648128,570425344,587202560,603979776,671088640,738197504,805306368,872415232,939524096,1006632960,1073741824,1140850688,1207959552,1275068416,1342177280,1409286144,1476395008,1610612736,1744830464,1879048192,2013265920,2147483648,2415919104,2684354560,2952790016,3221225472,3489660928,3758096384,4026531840,4294967296,8589934592,17179869184,34359738368,68719476736,137438953472,274877906944],"fileCounts":[10,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],"totalBytes":[5259,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]},"deletedRecordCountsHistogramOpt":{"deletedRecordCounts":[10,0,0,0,0,0,0,0,0,0]},"allFiles":[{"path":"part-00223-24d8cffb-245d-4027-87d6-940fcf593a60.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":4},\"maxValues\":{\"id\":4},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000004","MIN_INSERTION_TIME":"1694758257000004","MAX_INSERTION_TIME":"1694758257000004","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":4,"defaultRowCommitVersion":0},{"path":"part-00268-365db28b-f856-49e6-a25f-b0211cf95d20.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":5},\"maxValues\":{\"id\":5},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000005","MIN_INSERTION_TIME":"1694758257000005","MAX_INSERTION_TIME":"1694758257000005","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":5,"defaultRowCommitVersion":0},{"path":"part-00089-b466c656-9b4a-41d6-ab41-f02007d1658c.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":1},\"maxValues\":{\"id\":1},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000001","MIN_INSERTION_TIME":"1694758257000001","MAX_INSERTION_TIME":"1694758257000001","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":1,"defaultRowCommitVersion":0},{"path":"part-00044-22c23f7f-2411-4d88-b78c-cebe430cdd47.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":0},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000000","MIN_INSERTION_TIME":"1694758257000000","MAX_INSERTION_TIME":"1694758257000000","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":0,"defaultRowCommitVersion":0},{"path":"part-00447-1755ad02-9b47-4287-8333-92cb01a5124b.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":9},\"maxValues\":{\"id\":9},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000009","MIN_INSERTION_TIME":"1694758257000009","MAX_INSERTION_TIME":"1694758257000009","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":9,"defaultRowCommitVersion":0},{"path":"part-00134-34f9b771-c60a-4bd4-bdc0-cd25fcc951c6.c000.snappy.parquet","partitionValues":{},"size":525,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":2},\"maxValues\":{\"id\":2},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000002","MIN_INSERTION_TIME":"1694758257000002","MAX_INSERTION_TIME":"1694758257000002","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":2,"defaultRowCommitVersion":0},{"path":"part-00358-5937ec73-64a5-44dd-a793-922e30c1b9df.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":7},\"maxValues\":{\"id\":7},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000007","MIN_INSERTION_TIME":"1694758257000007","MAX_INSERTION_TIME":"1694758257000007","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":7,"defaultRowCommitVersion":0},{"path":"part-00313-c528546e-c8ab-425d-b49a-5afe731aaac8.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":6},\"maxValues\":{\"id\":6},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000006","MIN_INSERTION_TIME":"1694758257000006","MAX_INSERTION_TIME":"1694758257000006","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":6,"defaultRowCommitVersion":0},{"path":"part-00179-76f56874-b389-409b-8a2d-18462928840e.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":3},\"maxValues\":{\"id\":3},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000003","MIN_INSERTION_TIME":"1694758257000003","MAX_INSERTION_TIME":"1694758257000003","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":3,"defaultRowCommitVersion":0},{"path":"part-00403-6af19469-0fc5-4809-b02a-ddebda3966e8.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":false,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":8},\"maxValues\":{\"id\":8},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000008","MIN_INSERTION_TIME":"1694758257000008","MAX_INSERTION_TIME":"1694758257000008","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":8,"defaultRowCommitVersion":0}]}

rust/tests/data/table_with_liquid_clustering/_delta_log/00000000000000000000.json

@@ -0,0 +1,15 @@
+{"commitInfo":{"timestamp":1694758257261,"operation":"CREATE OR REPLACE TABLE AS SELECT","operationParameters":{"partitionBy":"[]","clusterBy":"[\"id\"]","description":null,"isManaged":"true","properties":"{\"delta.liquid.clusteringColumns\":\"id\",\"delta.enableDeletionVectors\":\"true\",\"delta.checkpoint.writeStatsAsStruct\":\"true\",\"delta.enableRowTracking\":\"true\",\"delta.checkpoint.writeStatsAsJson\":\"false\",\"delta.rowTracking.materializedRowCommitVersionColumnName\":\"_row-commit-version-col-2f60dcc1-9e36-4424-95e7-799b707e4ddb\",\"delta.rowTracking.materializedRowIdColumnName\":\"_row-id-col-4cbc7924-f662-4db1-aa59-22c23f59eb5d\"}","statsOnLoad":false},"isolationLevel":"WriteSerializable","isBlindAppend":false,"operationMetrics":{"numFiles":"10","numOutputRows":"10","numOutputBytes":"5259"},"engineInfo":"Databricks-Runtime/13.2.x-photon-scala2.12","txnId":"29ebf587-9705-4bb4-ac40-2f02324065c8"}}
+{"metaData":{"id":"6ca3020b-3cd9-4048-82e3-1417a0abb98f","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.liquid.clusteringColumns":"id","delta.enableDeletionVectors":"true","delta.checkpoint.writeStatsAsStruct":"true","delta.enableRowTracking":"true","delta.checkpoint.writeStatsAsJson":"false","delta.rowTracking.materializedRowCommitVersionColumnName":"_row-commit-version-col-2f60dcc1-9e36-4424-95e7-799b707e4ddb","delta.rowTracking.materializedRowIdColumnName":"_row-id-col-4cbc7924-f662-4db1-aa59-22c23f59eb5d"},"createdTime":1694758256009}}
+{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["domainMetadata","deletionVectors","rowTracking","liquid"]}}
+{"add":{"path":"part-00044-22c23f7f-2411-4d88-b78c-cebe430cdd47.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":0},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000000","MIN_INSERTION_TIME":"1694758257000000","MAX_INSERTION_TIME":"1694758257000000","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":0,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00089-b466c656-9b4a-41d6-ab41-f02007d1658c.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":1},\"maxValues\":{\"id\":1},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000001","MIN_INSERTION_TIME":"1694758257000001","MAX_INSERTION_TIME":"1694758257000001","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":1,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00134-34f9b771-c60a-4bd4-bdc0-cd25fcc951c6.c000.snappy.parquet","partitionValues":{},"size":525,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":2},\"maxValues\":{\"id\":2},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000002","MIN_INSERTION_TIME":"1694758257000002","MAX_INSERTION_TIME":"1694758257000002","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":2,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00179-76f56874-b389-409b-8a2d-18462928840e.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":3},\"maxValues\":{\"id\":3},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000003","MIN_INSERTION_TIME":"1694758257000003","MAX_INSERTION_TIME":"1694758257000003","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":3,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00223-24d8cffb-245d-4027-87d6-940fcf593a60.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":4},\"maxValues\":{\"id\":4},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000004","MIN_INSERTION_TIME":"1694758257000004","MAX_INSERTION_TIME":"1694758257000004","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":4,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00268-365db28b-f856-49e6-a25f-b0211cf95d20.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":5},\"maxValues\":{\"id\":5},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000005","MIN_INSERTION_TIME":"1694758257000005","MAX_INSERTION_TIME":"1694758257000005","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":5,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00313-c528546e-c8ab-425d-b49a-5afe731aaac8.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":6},\"maxValues\":{\"id\":6},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000006","MIN_INSERTION_TIME":"1694758257000006","MAX_INSERTION_TIME":"1694758257000006","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":6,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00358-5937ec73-64a5-44dd-a793-922e30c1b9df.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":7},\"maxValues\":{\"id\":7},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000007","MIN_INSERTION_TIME":"1694758257000007","MAX_INSERTION_TIME":"1694758257000007","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":7,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00403-6af19469-0fc5-4809-b02a-ddebda3966e8.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":8},\"maxValues\":{\"id\":8},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000008","MIN_INSERTION_TIME":"1694758257000008","MAX_INSERTION_TIME":"1694758257000008","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":8,"defaultRowCommitVersion":0}}
+{"add":{"path":"part-00447-1755ad02-9b47-4287-8333-92cb01a5124b.c000.snappy.parquet","partitionValues":{},"size":526,"modificationTime":1694758257000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":9},\"maxValues\":{\"id\":9},\"nullCount\":{\"id\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1694758257000009","MIN_INSERTION_TIME":"1694758257000009","MAX_INSERTION_TIME":"1694758257000009","OPTIMIZE_TARGET_SIZE":"268435456"},"baseRowId":9,"defaultRowCommitVersion":0}}
+{"domainMetadata":{"domain":"delta.liquid","configuration":"{\"clusteringColumns\":[{\"physicalName\":[\"id\"]}],\"domainName\":\"delta.liquid\"}","removed":false}}
+{"domainMetadata":{"domain":"delta.rowTracking","configuration":"{\"rowIdHighWaterMark\":9,\"domainName\":\"delta.rowTracking\"}","removed":false}}

rust/tests/read_delta_log_test.rs

@@ -1,4 +1,4 @@
-use deltalake::DeltaTableBuilder;
+use deltalake::{DeltaResult, DeltaTableBuilder};
 use pretty_assertions::assert_eq;
 use std::collections::HashMap;
 use std::time::SystemTime;
@@ -142,3 +142,10 @@ async fn test_log_buffering_fail() {
         .is_err();
     assert!(table_err);
 }
+
+#[tokio::test]
+async fn test_read_liquid_table() -> DeltaResult<()> {
+    let path = "./tests/data/table_with_liquid_clustering";
+    let _table = deltalake::open_table(&path).await?;
+    Ok(())
+}