Merge remote-tracking branch 'upstream/main' into feat/metrics

Author: gotocoding-DB

acceptance/Cargo.toml

@@ -36,6 +36,7 @@ tar = "0.4"
 [dev-dependencies]
 datatest-stable = "0.3"
 test-log = { version = "0.2", default-features = false, features = ["trace"] }
+test_utils = { path = "../test-utils" }
 tempfile = "3"
 test-case = { version = "3.3.1" }
 tokio = { version = "1.47" }

acceptance/tests/dat_reader.rs

@@ -1,9 +1,6 @@
 use std::path::Path;
-use std::sync::Arc;
 
 use acceptance::read_dat_case;
-use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
-use delta_kernel::engine::default::DefaultEngine;
 
 // TODO(zach): skip iceberg_compat_v1 test until DAT is fixed
 static SKIPPED_TESTS: &[&str; 1] = &["iceberg_compat_v1"];
@@ -27,14 +24,7 @@ fn reader_test(path: &Path) -> datatest_stable::Result<()> {
         .block_on(async {
             let case = read_dat_case(root_dir).unwrap();
             let table_root = case.table_root().unwrap();
-            let engine = Arc::new(
-                DefaultEngine::try_new(
-                    &table_root,
-                    std::iter::empty::<(&str, &str)>(),
-                    Arc::new(TokioBackgroundExecutor::new()),
-                )
-                .unwrap(),
-            );
+            let engine = test_utils::create_default_engine(&table_root).unwrap();
 
             case.assert_metadata(engine.clone()).await.unwrap();
             acceptance::data::assert_scan_metadata(engine.clone(), &case)

acceptance/tests/other.rs

@@ -38,9 +38,7 @@ async fn test_read_table_with_checkpoint() {
     ))
     .unwrap();
     let location = url::Url::from_directory_path(path).unwrap();
-    let engine = Arc::new(
-        DefaultEngine::try_new(&location, HashMap::<String, String>::new()).unwrap(),
-    );
+    let engine = test_utils::create_default_engine(&location).unwrap();
     let snapshot = Snapshot::try_new(location, engine, None)
         .await
         .unwrap();

ffi/Cargo.toml

@@ -46,6 +46,7 @@ object_store = "0.12.3"
 default = ["default-engine-rustls"]
 default-engine-native-tls = ["delta_kernel/default-engine-native-tls", "default-engine-base"]
 default-engine-rustls = ["delta_kernel/default-engine-rustls", "default-engine-base"]
+catalog-managed = ["delta_kernel/catalog-managed"]
 
 # This is an 'internal' feature flag which has all the shared bits from default-engine-native-tls and
 # default-engine-rustls. There is a check in kernel/lib.rs to ensure you have enabled one of

ffi/src/domain_metadata.rs

@@ -93,7 +93,6 @@ mod tests {
         recover_string,
     };
     use crate::{engine_to_handle, free_engine, free_snapshot, kernel_string_slice, snapshot};
-    use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
     use delta_kernel::engine::default::DefaultEngine;
     use delta_kernel::DeltaResult;
     use object_store::memory::InMemory;
@@ -107,7 +106,7 @@ mod tests {
     async fn test_domain_metadata() -> DeltaResult<()> {
         let storage = Arc::new(InMemory::new());
 
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage.clone());
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
         let path = "memory:///";
 

ffi/src/lib.rs

@@ -13,8 +13,7 @@ use url::Url;
 
 use delta_kernel::schema::Schema;
 use delta_kernel::snapshot::Snapshot;
-use delta_kernel::Version;
-use delta_kernel::{DeltaResult, Engine, EngineData};
+use delta_kernel::{DeltaResult, Engine, EngineData, LogPath, Version};
 use delta_kernel_ffi_macros::handle_descriptor;
 
 // cbindgen doesn't understand our use of feature flags here, and by default it parses `mod handle`
@@ -43,6 +42,8 @@ use error::{AllocateError, AllocateErrorFn, ExternResult, IntoExternResult};
 pub mod expressions;
 #[cfg(feature = "tracing")]
 pub mod ffi_tracing;
+#[cfg(feature = "catalog-managed")]
+pub mod log_path;
 pub mod scan;
 pub mod schema;
 
@@ -560,13 +561,11 @@ fn get_default_engine_impl(
     allocate_error: AllocateErrorFn,
 ) -> DeltaResult<Handle<SharedExternEngine>> {
     use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
+    use delta_kernel::engine::default::storage::store_from_url_opts;
     use delta_kernel::engine::default::DefaultEngine;
-    let engine = DefaultEngine::<TokioBackgroundExecutor>::try_new(
-        &url,
-        options,
-        Arc::new(TokioBackgroundExecutor::new()),
-    );
-    Ok(engine_to_handle(Arc::new(engine?), allocate_error))
+    let store = store_from_url_opts(&url, options)?;
+    let engine = DefaultEngine::<TokioBackgroundExecutor>::new(store);
+    Ok(engine_to_handle(Arc::new(engine), allocate_error))
 }
 
 /// # Safety
@@ -596,10 +595,36 @@ pub unsafe extern "C" fn snapshot(
 ) -> ExternResult<Handle<SharedSnapshot>> {
     let url = unsafe { unwrap_and_parse_path_as_url(path) };
     let engine = unsafe { engine.as_ref() };
-    snapshot_impl(url, engine, None).into_extern_result(&engine)
+    snapshot_impl(url, engine, None, Vec::new()).into_extern_result(&engine)
+}
+
+/// Get the latest snapshot from the specified table with optional log tail
+///
+/// # Safety
+///
+/// Caller is responsible for passing valid handles and path pointer.
+/// The log_paths array and its contents must remain valid for the duration of this call.
+#[cfg(feature = "catalog-managed")]
+#[no_mangle]
+pub unsafe extern "C" fn snapshot_with_log_tail(
+    path: KernelStringSlice,
+    engine: Handle<SharedExternEngine>,
+    log_paths: log_path::LogPathArray,
+) -> ExternResult<Handle<SharedSnapshot>> {
+    let url = unsafe { unwrap_and_parse_path_as_url(path) };
+    let engine_ref = unsafe { engine.as_ref() };
+
+    // Convert LogPathArray to Vec<LogPath>
+    let log_tail = match unsafe { log_paths.log_paths() } {
+        Ok(paths) => paths,
+        Err(err) => return Err(err).into_extern_result(&engine_ref),
+    };
+
+    snapshot_impl(url, engine_ref, None, log_tail).into_extern_result(&engine_ref)
 }
 
-/// Get the snapshot from the specified table at a specific version
+/// Get the snapshot from the specified table at a specific version. Note this is only safe for
+/// non-catalog-managed tables.
 ///
 /// # Safety
 ///
@@ -612,21 +637,52 @@ pub unsafe extern "C" fn snapshot_at_version(
 ) -> ExternResult<Handle<SharedSnapshot>> {
     let url = unsafe { unwrap_and_parse_path_as_url(path) };
     let engine = unsafe { engine.as_ref() };
-    snapshot_impl(url, engine, version.into()).into_extern_result(&engine)
+    snapshot_impl(url, engine, version.into(), Vec::new()).into_extern_result(&engine)
+}
+
+/// Get the snapshot from the specified table at a specific version with log tail.
+///
+/// # Safety
+///
+/// Caller is responsible for passing valid handles and path pointer.
+/// The log_tail array and its contents must remain valid for the duration of this call.
+#[cfg(feature = "catalog-managed")]
+#[no_mangle]
+pub unsafe extern "C" fn snapshot_at_version_with_log_tail(
+    path: KernelStringSlice,
+    engine: Handle<SharedExternEngine>,
+    version: Version,
+    log_tail: log_path::LogPathArray,
+) -> ExternResult<Handle<SharedSnapshot>> {
+    let url = unsafe { unwrap_and_parse_path_as_url(path) };
+    let engine_ref = unsafe { engine.as_ref() };
+
+    // Convert LogPathArray to Vec<LogPath>
+    let log_tail = match unsafe { log_tail.log_paths() } {
+        Ok(paths) => paths,
+        Err(err) => return Err(err).into_extern_result(&engine_ref),
+    };
+
+    snapshot_impl(url, engine_ref, version.into(), log_tail).into_extern_result(&engine_ref)
 }
 
 fn snapshot_impl(
     url: DeltaResult<Url>,
     extern_engine: &dyn ExternEngine,
     version: Option<Version>,
+    #[allow(unused_variables)] log_tail: Vec<LogPath>,
 ) -> DeltaResult<Handle<SharedSnapshot>> {
-    let builder = Snapshot::builder_for(url?);
-    let builder = if let Some(v) = version {
-        // TODO: should we include a `with_version_opt` method for the builder?
-        builder.at_version(v)
-    } else {
-        builder
-    };
+    let mut builder = Snapshot::builder_for(url?);
+
+    if let Some(v) = version {
+        builder = builder.at_version(v);
+    }
+
+    #[cfg(feature = "catalog-managed")]
+    if !log_tail.is_empty() {
+        builder = builder.with_log_tail(log_tail);
+    }
+
     let snapshot = builder.build(extern_engine.engine().as_ref())?;
     Ok(snapshot.into())
 }
@@ -823,7 +879,7 @@ mod tests {
         allocate_err, allocate_str, assert_extern_result_error_with_message, ok_or_panic,
         recover_string,
     };
-    use delta_kernel::engine::default::{executor::tokio::TokioBackgroundExecutor, DefaultEngine};
+    use delta_kernel::engine::default::DefaultEngine;
     use object_store::memory::InMemory;
     use test_utils::{actions_to_string, actions_to_string_partitioned, add_commit, TestAction};
 
@@ -870,7 +926,7 @@ mod tests {
             actions_to_string(vec![TestAction::Metadata]),
         )
         .await?;
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage.clone());
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
         let path = "memory:///";
 
@@ -916,7 +972,7 @@ mod tests {
             actions_to_string_partitioned(vec![TestAction::Metadata]),
         )
         .await?;
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage.clone());
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
         let path = "memory:///";
 
@@ -952,7 +1008,7 @@ mod tests {
             actions_to_string(vec![TestAction::Metadata]),
         )
         .await?;
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage.clone());
         let engine = engine_to_handle(Arc::new(engine), allocate_null_err);
         let path = "memory:///";
 
@@ -964,4 +1020,65 @@ mod tests {
         unsafe { free_engine(engine) }
         Ok(())
     }
+
+    #[cfg(feature = "catalog-managed")]
+    #[tokio::test]
+    async fn test_snapshot_log_tail() -> Result<(), Box<dyn std::error::Error>> {
+        use test_utils::add_staged_commit;
+        let storage = Arc::new(InMemory::new());
+        add_commit(
+            storage.as_ref(),
+            0,
+            actions_to_string(vec![TestAction::Metadata]),
+        )
+        .await?;
+        let commit1 = add_staged_commit(
+            storage.as_ref(),
+            1,
+            actions_to_string(vec![TestAction::Add("path1".into())]),
+        )
+        .await?;
+        let engine = DefaultEngine::new(storage.clone());
+        let engine = engine_to_handle(Arc::new(engine), allocate_err);
+        let path = "memory:///";
+
+        let commit1_path = format!(
+            "{}_delta_log/_staged_commits/{}",
+            path,
+            commit1.filename().unwrap()
+        );
+        let log_path =
+            log_path::FfiLogPath::new(kernel_string_slice!(commit1_path), 123456789, 100);
+        let log_tail = [log_path];
+        let log_tail = log_path::LogPathArray {
+            ptr: log_tail.as_ptr(),
+            len: log_tail.len(),
+        };
+        let snapshot = unsafe {
+            ok_or_panic(snapshot_with_log_tail(
+                kernel_string_slice!(path),
+                engine.shallow_copy(),
+                log_tail.clone(),
+            ))
+        };
+        let snapshot_version = unsafe { version(snapshot.shallow_copy()) };
+        assert_eq!(snapshot_version, 1);
+
+        // Test getting snapshot at version
+        let snapshot2 = unsafe {
+            ok_or_panic(snapshot_at_version_with_log_tail(
+                kernel_string_slice!(path),
+                engine.shallow_copy(),
+                1,
+                log_tail,
+            ))
+        };
+        let snapshot_version = unsafe { version(snapshot.shallow_copy()) };
+        assert_eq!(snapshot_version, 1);
+
+        unsafe { free_snapshot(snapshot) }
+        unsafe { free_snapshot(snapshot2) }
+        unsafe { free_engine(engine) }
+        Ok(())
+    }
 }