diff --git a/ffi/src/domain_metadata.rs b/ffi/src/domain_metadata.rs
index d2ca0fc40c..0553e6461d 100644
--- a/ffi/src/domain_metadata.rs
+++ b/ffi/src/domain_metadata.rs
@@ -37,6 +37,39 @@ fn get_domain_metadata_impl(
         .and_then(|config: String| allocate_fn(kernel_string_slice!(config))))
 }
 
+/// Get the domain metadata including system-controlled (`delta.*`) domains.
+/// Unlike `get_domain_metadata`, this does not reject `delta.*` domains.
+/// This is needed to read metadata such as `delta.clustering`.
+///
+/// # Safety
+///
+/// Caller is responsible for passing in a valid handle
+#[no_mangle]
+pub unsafe extern "C" fn get_domain_metadata_internal(
+    snapshot: Handle<SharedSnapshot>,
+    domain: KernelStringSlice,
+    engine: Handle<SharedExternEngine>,
+    allocate_fn: AllocateStringFn,
+) -> ExternResult<NullableCvoid> {
+    let snapshot = unsafe { snapshot.as_ref() };
+    let engine = unsafe { engine.as_ref() };
+    let domain = unsafe { String::try_from_slice(&domain) };
+
+    get_domain_metadata_internal_impl(snapshot, domain, engine, allocate_fn)
+        .into_extern_result(&engine)
+}
+
+fn get_domain_metadata_internal_impl(
+    snapshot: &Snapshot,
+    domain: DeltaResult<String>,
+    extern_engine: &dyn ExternEngine,
+    allocate_fn: AllocateStringFn,
+) -> DeltaResult<NullableCvoid> {
+    Ok(snapshot
+        .get_domain_metadata_internal(&domain?, extern_engine.engine().as_ref())?
+        .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
@@ -253,6 +286,28 @@ mod tests {
             "domain2_commit1"
         );
 
+        // Test get_domain_metadata_internal (allows delta.* domains)
+        let get_internal_helper = |domain: &str| unsafe {
+            get_domain_metadata_internal(
+                snapshot.shallow_copy(),
+                kernel_string_slice!(domain),
+                engine.shallow_copy(),
+                allocate_str,
+            )
+        };
+
+        // delta.* domains succeed via the internal API
+        let res = ok_or_panic(get_internal_helper("delta.domain3"));
+        assert_eq!(recover_string(res.unwrap()), "domain3_commit1");
+
+        // Regular domains also work
+        let res = ok_or_panic(get_internal_helper("domain2"));
+        assert_eq!(recover_string(res.unwrap()), "domain2_commit1");
+
+        // Removed domain returns None
+        let res = ok_or_panic(get_internal_helper("domain1"));
+        assert!(res.is_none());
+
         unsafe { free_snapshot(snapshot) }
         unsafe { free_engine(engine) }
 
diff --git a/kernel/src/snapshot.rs b/kernel/src/snapshot.rs
index 2145c76b14..1c85838dab 100644
--- a/kernel/src/snapshot.rs
+++ b/kernel/src/snapshot.rs
@@ -475,6 +475,21 @@ impl Snapshot {
         domain_metadata_configuration(self.log_segment(), domain, engine)
     }
 
+    /// Fetch the domainMetadata for a specific domain, including system-controlled (`delta.*`)
+    /// domains. Unlike [`get_domain_metadata`], internal domains are allowed.
+    ///
+    /// This is needed to read metadata such as `delta.clustering` which stores the clustering
+    /// columns for liquid-clustered tables.
+    ///
+    /// Note that this method performs log replay (fetches and processes metadata from storage).
+    pub fn get_domain_metadata_internal(
+        &self,
+        domain: &str,
+        engine: &dyn Engine,
+    ) -> DeltaResult<Option<String>> {
+        domain_metadata_configuration(self.log_segment(), domain, engine)
+    }
+
     #[allow(unused)]
     #[internal_api]
     pub(crate) fn get_all_domain_metadata(
@@ -1276,6 +1291,22 @@ mod tests {
         assert!(matches!(err, Error::Generic(msg) if
                 msg == "User DomainMetadata are not allowed to use system-controlled 'delta.*' domain"));
 
+        // Test get_domain_metadata_internal (allows delta.* domains)
+        assert_eq!(
+            snapshot.get_domain_metadata_internal("delta.domain3", &engine)?,
+            Some("domain3_commit1".to_string())
+        );
+        // Also works for regular domains
+        assert_eq!(
+            snapshot.get_domain_metadata_internal("domain2", &engine)?,
+            Some("domain2_commit1".to_string())
+        );
+        // Removed domain returns None
+        assert_eq!(
+            snapshot.get_domain_metadata_internal("domain1", &engine)?,
+            None
+        );
+
         // Test get_all_domain_metadata
         let mut metadata = snapshot.get_all_domain_metadata(&engine)?;
         metadata.sort_by(|a, b| a.domain().cmp(b.domain()));