feat: allow visiting entire domain metadata (#1384)

## What changes are proposed in this pull request?
Adding the ability to scan the entire domain metadata. This PR is in a
way a follow up to
#1342. I wanted to use
the `visit_string_map` function from that PR, but then we would run into
some ownership/lifetime issues: we would need to return a pointer to a
kernel-allocated map

I thought the nicest solution would be to just create a separate visitor
function which avoids any lifetime issues by scoping things to the
visitor callback.

I also considered:

```rust
pub unsafe extern "C" fn visit_domain_metadata(
    snapshot: Handle<SharedSnapshot>,
    engine: Handle<SharedExternEngine>,
    engine_context: NullableCvoid,
    visitor: extern "C" fn(
        engine_context: NullableCvoid,
        map: &CStringMap,
    ),
)
```

Which would allow reusing the `visit_string_map` function and make this
more in line with how engine is expected to consume other string maps,
but I don't think the extra level of indirection really adds much and
the visitor is very simple anyway.

### This PR affects the following public APIs
- Added new `Snapshot::get_all_domain_metadata` function
- Added new ffi `visit_domain_metadata` to go over all domain metadata

## How was this change tested?
- Testing code added to existing
`kernel::snapshot::tests::test_domain_metadata` test
- Testing code added to existing
`ffi::domain_metadata::tests::test_domain_metadata` test

---------

Co-authored-by: Nick Lanham <nicklan@users.noreply.github.com>

Author: samansmink

ffi/src/domain_metadata.rs

@@ -37,6 +37,53 @@ fn get_domain_metadata_impl(
         .and_then(|config: String| allocate_fn(kernel_string_slice!(config))))
 }
 
+/// Get the domain metadata as an optional string allocated by `AllocatedStringFn` for a specific domain in this snapshot
+///
+/// # Safety
+///
+/// Caller is responsible for passing in a valid handle
+#[no_mangle]
+pub unsafe extern "C" fn visit_domain_metadata(
+    snapshot: Handle<SharedSnapshot>,
+    engine: Handle<SharedExternEngine>,
+    engine_context: NullableCvoid,
+    visitor: extern "C" fn(
+        engine_context: NullableCvoid,
+        domain: KernelStringSlice,
+        configuration: KernelStringSlice,
+    ),
+) -> ExternResult<bool> {
+    let snapshot = unsafe { snapshot.as_ref() };
+    let engine = unsafe { engine.as_ref() };
+
+    visit_domain_metadata_impl(snapshot, engine, engine_context, visitor)
+        .into_extern_result(&engine)
+}
+
+fn visit_domain_metadata_impl(
+    snapshot: &Snapshot,
+    extern_engine: &dyn ExternEngine,
+    engine_context: NullableCvoid,
+    visitor: extern "C" fn(
+        engine_context: NullableCvoid,
+        key: KernelStringSlice,
+        value: KernelStringSlice,
+    ),
+) -> DeltaResult<bool> {
+    let res = snapshot.get_all_domain_metadata(extern_engine.engine().as_ref())?;
+    res.iter().for_each(|metadata| {
+        let domain = &metadata.domain();
+        let configuration = &metadata.configuration();
+        visitor(
+            engine_context,
+            kernel_string_slice!(domain),
+            kernel_string_slice!(configuration),
+        );
+    });
+
+    Ok(true)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -51,6 +98,8 @@ mod tests {
     use delta_kernel::DeltaResult;
     use object_store::memory::InMemory;
     use serde_json::json;
+    use std::collections::HashMap;
+    use std::ptr::NonNull;
     use std::sync::Arc;
     use test_utils::add_commit;
 
@@ -148,6 +197,8 @@ mod tests {
             )
         };
 
+        // First, we test fetching the domain metadata one-by-one
+
         let domain1 = "domain1";
         let res = ok_or_panic(get_domain_metadata_helper(domain1));
         assert!(res.is_none());
@@ -160,6 +211,49 @@ mod tests {
         let res = get_domain_metadata_helper(domain3);
         assert_extern_result_error_with_message(res, KernelError::GenericError, "Generic delta kernel error: User DomainMetadata are not allowed to use system-controlled 'delta.*' domain");
 
+        // Secondly, we visit the entire domain metadata
+
+        // Create visitor state
+        let visitor_state: Box<HashMap<String, String>> = Box::default();
+        let visitor_state_ptr = Box::into_raw(visitor_state);
+
+        // Test visitor function
+        extern "C" fn visitor(
+            state: NullableCvoid,
+            key: KernelStringSlice,
+            value: KernelStringSlice,
+        ) {
+            let mut collected_metadata = unsafe {
+                Box::from_raw(
+                    state.unwrap().as_ptr() as *mut std::collections::HashMap<String, String>
+                )
+            };
+            let key: DeltaResult<String> = unsafe { TryFromStringSlice::try_from_slice(&key) };
+            let value: DeltaResult<String> = unsafe { TryFromStringSlice::try_from_slice(&value) };
+            collected_metadata.insert(key.unwrap(), value.unwrap());
+            Box::leak(collected_metadata);
+        }
+
+        // Visit all (user) domain metadata
+        let res = unsafe {
+            ok_or_panic(visit_domain_metadata(
+                snapshot.shallow_copy(),
+                engine.shallow_copy(),
+                Some(NonNull::new_unchecked(visitor_state_ptr).cast()),
+                visitor,
+            ))
+        };
+
+        // Confirm visitor picked up all entries in map
+        let collected_metadata = unsafe { Box::from_raw(visitor_state_ptr) };
+        assert!(res);
+        assert!(collected_metadata.get("domain1").is_none());
+        assert!(collected_metadata.get("delta.domain3").is_none());
+        assert_eq!(
+            collected_metadata.get("domain2").unwrap(),
+            "domain2_commit1"
+        );
+
         unsafe { free_snapshot(snapshot) }
         unsafe { free_engine(engine) }
 

kernel/src/actions/domain_metadata.rs

@@ -4,15 +4,15 @@
 //! For now, this module only exposes the ability to read a single domain at once from the log. In
 //! the future this should allow for reading all domains from the log at once.
 
-use std::collections::HashMap;
-use std::sync::{Arc, LazyLock};
-
 use crate::actions::get_log_domain_metadata_schema;
 use crate::actions::visitors::DomainMetadataVisitor;
 use crate::actions::{DomainMetadata, DOMAIN_METADATA_NAME};
 use crate::log_replay::ActionsBatch;
 use crate::log_segment::LogSegment;
 use crate::{DeltaResult, Engine, Expression as Expr, PredicateRef, RowVisitor as _};
+use delta_kernel_derive::internal_api;
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
 
 const DOMAIN_METADATA_DOMAIN_FIELD: &str = "domain";
 
@@ -35,6 +35,16 @@ pub(crate) fn domain_metadata_configuration(
         .map(|domain_metadata| domain_metadata.configuration))
 }
 
+#[allow(unused)]
+#[internal_api]
+pub(crate) fn all_domain_metadata_configuration(
+    log_segment: &LogSegment,
+    engine: &dyn Engine,
+) -> DeltaResult<Vec<DomainMetadata>> {
+    let domain_metadatas = scan_domain_metadatas(log_segment, None, engine)?;
+    Ok(domain_metadatas.into_values().collect())
+}
+
 /// 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.

kernel/src/actions/mod.rs

@@ -1062,14 +1062,17 @@ impl DomainMetadata {
     // returns true if the domain metadata is an system-controlled domain (all domains that start
     // with "delta.")
     #[allow(unused)]
+    #[internal_api]
     pub(crate) fn is_internal(&self) -> bool {
         self.domain.starts_with(INTERNAL_DOMAIN_PREFIX)
     }
 
+    #[internal_api]
     pub(crate) fn domain(&self) -> &str {
         &self.domain
     }
 
+    #[internal_api]
     pub(crate) fn configuration(&self) -> &str {
         &self.configuration
     }

kernel/src/snapshot.rs

@@ -4,7 +4,9 @@
 use std::sync::Arc;
 
 use crate::action_reconciliation::calculate_transaction_expiration_timestamp;
-use crate::actions::domain_metadata::domain_metadata_configuration;
+use crate::actions::domain_metadata::{
+    all_domain_metadata_configuration, domain_metadata_configuration,
+};
 use crate::actions::set_transaction::SetTransactionScanner;
 use crate::actions::INTERNAL_DOMAIN_PREFIX;
 use crate::checkpoint::CheckpointWriter;
@@ -24,6 +26,7 @@ use delta_kernel_derive::internal_api;
 mod builder;
 pub use builder::SnapshotBuilder;
 
+use delta_kernel::actions::DomainMetadata;
 use tracing::debug;
 use url::Url;
 
@@ -371,6 +374,19 @@ impl Snapshot {
         domain_metadata_configuration(self.log_segment(), domain, engine)
     }
 
+    #[allow(unused)]
+    #[internal_api]
+    pub(crate) fn get_all_domain_metadata(
+        &self,
+        engine: &dyn Engine,
+    ) -> DeltaResult<Vec<DomainMetadata>> {
+        let all_metadata = all_domain_metadata_configuration(self.log_segment(), engine)?;
+        Ok(all_metadata
+            .into_iter()
+            .filter(|domain| !domain.is_internal())
+            .collect())
+    }
+
     /// Get the In-Commit Timestamp (ICT) for this snapshot.
     ///
     /// Returns the `inCommitTimestamp` from the CommitInfo action of the commit that created this snapshot.
@@ -443,6 +459,7 @@ mod tests {
     use crate::parquet::arrow::ArrowWriter;
     use crate::path::ParsedLogPath;
     use crate::utils::test_utils::{assert_result_error_with_message, string_array_to_engine_data};
+    use delta_kernel::actions::DomainMetadata;
 
     /// Helper function to create a commitInfo action with optional ICT
     fn create_commit_info(timestamp: i64, ict: Option<i64>) -> serde_json::Value {
@@ -1152,6 +1169,8 @@ mod tests {
 
         let snapshot = Snapshot::builder_for(url.clone()).build(&engine)?;
 
+        // Test get_domain_metadata
+
         assert_eq!(snapshot.get_domain_metadata("domain1", &engine)?, None);
         assert_eq!(
             snapshot.get_domain_metadata("domain2", &engine)?,
@@ -1166,6 +1185,19 @@ mod tests {
             .unwrap_err();
         assert!(matches!(err, Error::Generic(msg) if
                 msg == "User DomainMetadata are not allowed to use system-controlled 'delta.*' domain"));
+
+        // Test get_all_domain_metadata
+        let mut metadata = snapshot.get_all_domain_metadata(&engine)?;
+        metadata.sort_by(|a, b| a.domain().cmp(b.domain()));
+
+        let mut expected = vec![
+            DomainMetadata::new("domain2".to_string(), "domain2_commit1".to_string()),
+            DomainMetadata::new("domain3".to_string(), "domain3_commit0".to_string()),
+        ];
+        expected.sort_by(|a, b| a.domain().cmp(b.domain()));
+
+        assert_eq!(metadata, expected);
+
         Ok(())
     }