feat: support writing domain metadata (2/2) (delta-io#1275)

## What changes are proposed in this pull request?
This PR is (2/2) to support writing domain metadata. It adds support for
removing metadata for user-specified domains. As per the Delta
[specification](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#reader-requirements-for-domain-metadata:~:text=When%20true%2C%20the%20action%20serves%20as%20a%20tombstone%20to%20logically%20delete%20a%20metadata%20domain.%20Writers%20should%20preserve%20an%20accurate%20pre%2Dimage%20of%20the%20configuration.),
for a removed domain metadata "writers should preserve an accurate
pre-image of the configuration". Thus for any removals, we need to
perform a log replay and restore the original configuration of the
domain with the `removed` flag set to `true`. Furthermore, for any
removal where the domain does not already exist in the log, we treat
this as a no-op and do not write any record to the Delta Log.

- resolves delta-io#1270

## How was this change tested?
Added new integration tests in `write.rs`.

---------

Co-authored-by: Zach Schuermann <zachary.zvs@gmail.com>

Author: MannDP

kernel/src/actions/domain_metadata.rs

@@ -38,7 +38,7 @@ pub(crate) fn domain_metadata_configuration(
 /// Scan the entire log for all domain metadata actions but terminate early if a specific domain
 /// is provided. Note that this returns the latest domain metadata for each domain, accounting for
 /// tombstones (removed=true) - that is, removed domain metadatas will _never_ be returned.
-fn scan_domain_metadatas(
+pub(crate) fn scan_domain_metadatas(
     log_segment: &LogSegment,
     domain: Option<&str>,
     engine: &dyn Engine,

kernel/src/actions/mod.rs

@@ -969,6 +969,15 @@ impl DomainMetadata {
         }
     }
 
+    // Create a new DomainMetadata action to remove a domain.
+    pub(crate) fn remove(domain: String, configuration: String) -> Self {
+        Self {
+            domain,
+            configuration,
+            removed: true,
+        }
+    }
+
     // returns true if the domain metadata is an system-controlled domain (all domains that start
     // with "delta.")
     #[allow(unused)]
@@ -979,6 +988,14 @@ impl DomainMetadata {
     pub(crate) fn domain(&self) -> &str {
         &self.domain
     }
+
+    pub(crate) fn configuration(&self) -> &str {
+        &self.configuration
+    }
+
+    pub(crate) fn is_removed(&self) -> bool {
+        self.removed
+    }
 }
 
 #[cfg(test)]

kernel/src/transaction/mod.rs

@@ -1,13 +1,13 @@
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::iter;
 use std::ops::Deref;
 use std::sync::{Arc, LazyLock};
 
 use url::Url;
 
 use crate::actions::{
-    as_log_add_schema, get_log_commit_info_schema, get_log_domain_metadata_schema,
-    get_log_txn_schema, CommitInfo, DomainMetadata, SetTransaction,
+    as_log_add_schema, domain_metadata::scan_domain_metadatas, get_log_commit_info_schema,
+    get_log_domain_metadata_schema, get_log_txn_schema, CommitInfo, DomainMetadata, SetTransaction,
 };
 #[cfg(feature = "catalog-managed")]
 use crate::committer::FileSystemCommitter;
@@ -352,7 +352,27 @@ impl Transaction {
         self
     }
 
+    /// Remove domain metadata from the Delta log.
+    /// If the domain exists in the Delta log, this creates a tombstone to logically delete
+    /// the domain. The tombstone preserves the previous configuration value.
+    /// If the domain does not exist in the Delta log, this is a no-op.
+    /// Note that each domain can only appear once per transaction. That is, multiple operations
+    /// on the same domain are disallowed in a single transaction, as well as setting and removing
+    /// the same domain in a single transaction. If a duplicate domain is included, the `commit` will
+    /// fail (that is, we don't eagerly check domain validity here).
+    /// Removing metadata for multiple distinct domains is allowed.
+    pub fn with_domain_metadata_removed(mut self, domain: String) -> Self {
+        // actual configuration value determined during commit
+        self.domain_metadatas
+            .push(DomainMetadata::remove(domain, String::new()));
+        self
+    }
+
     /// Generate domain metadata actions with validation. Handle both user and system domains.
+    ///
+    /// This function may perform an expensive log replay operation if there are any domain removals.
+    /// The log replay is required to fetch the previous configuration value for the domain to preserve
+    /// in removal tombstones as mandated by the Delta spec.
     fn generate_domain_metadata_actions<'a>(
         &'a self,
         engine: &'a dyn Engine,
@@ -370,32 +390,58 @@ impl Transaction {
             ));
         }
 
-        // validate domain metadata
-        let mut domains = HashSet::new();
-        for domain_metadata in &self.domain_metadatas {
-            if domain_metadata.is_internal() {
+        // validate user domain metadata and check if we have removals
+        let mut seen_domains = HashSet::new();
+        let mut has_removals = false;
+        for dm in &self.domain_metadatas {
+            if dm.is_internal() {
                 return Err(Error::Generic(
                     "Cannot modify domains that start with 'delta.' as those are system controlled"
                         .to_string(),
                 ));
             }
-            if !domains.insert(domain_metadata.domain()) {
+
+            if !seen_domains.insert(dm.domain()) {
                 return Err(Error::Generic(format!(
                     "Metadata for domain {} already specified in this transaction",
-                    domain_metadata.domain()
+                    dm.domain()
                 )));
             }
+
+            if dm.is_removed() {
+                has_removals = true;
+            }
         }
 
+        // fetch previous configuration values (requires log replay)
+        let existing_domains = if has_removals {
+            scan_domain_metadatas(self.read_snapshot.log_segment(), None, engine)?
+        } else {
+            HashMap::new()
+        };
+
+        let user_domains = self
+            .domain_metadatas
+            .iter()
+            .filter_map(move |dm: &DomainMetadata| {
+                if dm.is_removed() {
+                    existing_domains.get(dm.domain()).map(|existing| {
+                        DomainMetadata::remove(
+                            dm.domain().to_string(),
+                            existing.configuration().to_string(),
+                        )
+                    })
+                } else {
+                    Some(dm.clone())
+                }
+            });
+
         let system_domains = row_tracking_high_watermark
             .map(DomainMetadata::try_from)
             .transpose()?
             .into_iter();
 
-        Ok(self
-            .domain_metadatas
-            .iter()
-            .cloned()
+        Ok(user_domains
             .chain(system_domains)
             .map(|dm| dm.into_engine_data(get_log_domain_metadata_schema().clone(), engine)))
     }

kernel/tests/write.rs

@@ -1409,6 +1409,214 @@ async fn test_set_domain_metadata_unsupported_writer_feature(
     Ok(())
 }
 
+#[tokio::test]
+async fn test_remove_domain_metadata_non_existent_domain() -> Result<(), Box<dyn std::error::Error>>
+{
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let table_name = "test_domain_metadata_unsupported";
+
+    let (store, engine, table_location) = engine_store_setup(table_name, None);
+    let table_url = create_table(
+        store.clone(),
+        table_location,
+        schema.clone(),
+        &[],
+        true,
+        vec![],
+        vec!["domainMetadata"],
+    )
+    .await?;
+
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+    let txn = snapshot.transaction(Box::new(FileSystemCommitter::new()))?;
+
+    let domain = "app.deprecated";
+
+    // removing domain metadata that doesn't exist should NOT write a tombstone
+    let _ = txn
+        .with_domain_metadata_removed(domain.to_string())
+        .commit(&engine)?;
+
+    let commit_data = store
+        .get(&Path::from(format!(
+            "/{table_name}/_delta_log/00000000000000000001.json"
+        )))
+        .await?
+        .bytes()
+        .await?;
+    let actions: Vec<serde_json::Value> = Deserializer::from_slice(&commit_data)
+        .into_iter()
+        .try_collect()?;
+
+    let domain_action = actions.iter().find(|v| v.get("domainMetadata").is_some());
+    assert!(
+        domain_action.is_none(),
+        "No tombstone should be written for non-existent domain"
+    );
+
+    let final_snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+    let config = final_snapshot.get_domain_metadata(domain, &engine)?;
+    assert_eq!(config, None);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_domain_metadata_set_remove_conflicts() -> Result<(), Box<dyn std::error::Error>> {
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let table_name = "test_domain_metadata_unsupported";
+
+    let (store, engine, table_location) = engine_store_setup(table_name, None);
+    let table_url = create_table(
+        store.clone(),
+        table_location,
+        schema.clone(),
+        &[],
+        true,
+        vec![],
+        vec!["domainMetadata"],
+    )
+    .await?;
+
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+
+    // set then remove same domain
+    let txn = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?;
+    let err = txn
+        .with_domain_metadata("app.config".to_string(), "v1".to_string())
+        .with_domain_metadata_removed("app.config".to_string())
+        .commit(&engine)
+        .unwrap_err();
+    assert!(err
+        .to_string()
+        .contains("already specified in this transaction"));
+
+    // remove then set same domain
+    let txn2 = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?;
+    let err = txn2
+        .with_domain_metadata_removed("test.domain".to_string())
+        .with_domain_metadata("test.domain".to_string(), "v1".to_string())
+        .commit(&engine)
+        .unwrap_err();
+    assert!(err
+        .to_string()
+        .contains("already specified in this transaction"));
+
+    // remove same domain twice
+    let txn3 = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?;
+    let err = txn3
+        .with_domain_metadata_removed("another.domain".to_string())
+        .with_domain_metadata_removed("another.domain".to_string())
+        .commit(&engine)
+        .unwrap_err();
+    assert!(err
+        .to_string()
+        .contains("already specified in this transaction"));
+
+    // remove system domain
+    let txn4 = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?;
+    let err = txn4
+        .with_domain_metadata_removed("delta.system".to_string())
+        .commit(&engine)
+        .unwrap_err();
+    assert!(err
+        .to_string()
+        .contains("Cannot modify domains that start with 'delta.' as those are system controlled"));
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_domain_metadata_set_then_remove() -> Result<(), Box<dyn std::error::Error>> {
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let table_name = "test_domain_metadata_unsupported";
+
+    let (store, engine, table_location) = engine_store_setup(table_name, None);
+    let table_url = create_table(
+        store.clone(),
+        table_location,
+        schema.clone(),
+        &[],
+        true,
+        vec![],
+        vec!["domainMetadata"],
+    )
+    .await?;
+
+    let domain = "app.config";
+    let configuration = r#"{"version": 1}"#;
+
+    // txn 1: set domain metadata
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+    let txn = snapshot.transaction(Box::new(FileSystemCommitter::new()))?;
+    let _ = txn
+        .with_domain_metadata(domain.to_string(), configuration.to_string())
+        .commit(&engine)?;
+
+    // txn 2: remove the same domain metadata
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+    let txn = snapshot.transaction(Box::new(FileSystemCommitter::new()))?;
+    let _ = txn
+        .with_domain_metadata_removed(domain.to_string())
+        .commit(&engine)?;
+
+    // verify removal commit preserves the previous configuration
+    let commit_data = store
+        .get(&Path::from(format!(
+            "/{table_name}/_delta_log/00000000000000000002.json"
+        )))
+        .await?
+        .bytes()
+        .await?;
+    let actions: Vec<serde_json::Value> = Deserializer::from_slice(&commit_data)
+        .into_iter()
+        .try_collect()?;
+
+    let domain_action = actions
+        .iter()
+        .find(|v| v.get("domainMetadata").is_some())
+        .unwrap();
+    assert_eq!(domain_action["domainMetadata"]["domain"], domain);
+    assert_eq!(
+        domain_action["domainMetadata"]["configuration"],
+        configuration
+    );
+    assert_eq!(domain_action["domainMetadata"]["removed"], true);
+
+    // verify reads see the metadata removal
+    let final_snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+    let domain_config = final_snapshot.get_domain_metadata(domain, &engine)?;
+    assert_eq!(domain_config, None);
+
+    Ok(())
+}
+
 async fn get_ict_at_version(
     store: Arc<dyn ObjectStore>,
     table_url: &Url,