refactor: simplify creation of default engine in tests

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/src/domain_metadata.rs

@@ -46,7 +46,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;
@@ -58,7 +57,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

@@ -560,13 +560,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
@@ -823,7 +821,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 +868,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 +914,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 +950,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:///";
 

ffi/src/table_changes.rs

@@ -351,7 +351,7 @@ mod tests {
     use delta_kernel::arrow::record_batch::RecordBatch;
     use delta_kernel::arrow::util::pretty::pretty_format_batches;
     use delta_kernel::engine::arrow_conversion::TryIntoArrow as _;
-    use delta_kernel::engine::default::{executor::tokio::TokioBackgroundExecutor, DefaultEngine};
+    use delta_kernel::engine::default::DefaultEngine;
     use delta_kernel::schema::{DataType, StructField, StructType};
     use delta_kernel::Engine;
     use delta_kernel_ffi::engine_data::get_engine_data;
@@ -545,7 +545,7 @@ mod tests {
         put_file(storage.as_ref(), PARQUET_FILE2.to_string(), &batch).await?;
 
         let path = "memory:///";
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage);
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
 
         let table_changes = ok_or_panic(unsafe {
@@ -632,7 +632,7 @@ mod tests {
         put_file(storage.as_ref(), PARQUET_FILE2.to_string(), &batch).await?;
 
         let path = "memory:///";
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage);
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
 
         let table_changes = ok_or_panic(unsafe {
@@ -688,7 +688,7 @@ mod tests {
         put_file(storage.as_ref(), PARQUET_FILE2.to_string(), &batch).await?;
 
         let path = "memory:///";
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage);
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
 
         let table_changes = ok_or_panic(unsafe {
@@ -768,7 +768,7 @@ mod tests {
         put_file(storage.as_ref(), PARQUET_FILE2.to_string(), &batch).await?;
 
         let path = "memory:///";
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage);
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
 
         let table_changes = ok_or_panic(unsafe {

kernel/benches/metadata_bench.rs

@@ -17,7 +17,6 @@
 //!
 //! Follow-ups: <https://github.com/delta-io/delta-kernel-rs/issues/1185>
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
@@ -41,9 +40,9 @@ fn setup() -> (TempDir, Url, Arc<DefaultEngine<TokioBackgroundExecutor>>) {
     let table_path = tempdir.path().join(table);
     let url = try_parse_uri(table_path.to_str().unwrap()).expect("Failed to parse table path");
     // TODO: use multi-threaded executor
-    let executor = Arc::new(TokioBackgroundExecutor::new());
-    let engine = DefaultEngine::try_new(&url, HashMap::<String, String>::new(), executor)
-        .expect("Failed to create engine");
+    use delta_kernel::engine::default::storage::store_from_url;
+    let store = store_from_url(&url).expect("Failed to create store");
+    let engine = DefaultEngine::new(store);
 
     (tempdir, url, Arc::new(engine))
 }

kernel/examples/common/src/lib.rs

@@ -4,11 +4,9 @@ use std::{collections::HashMap, sync::Arc};
 
 use clap::{Args, CommandFactory, FromArgMatches};
 use delta_kernel::{
-    arrow::array::RecordBatch,
-    engine::default::{executor::tokio::TokioBackgroundExecutor, DefaultEngine},
-    scan::Scan,
-    schema::MetadataColumnSpec,
-    DeltaResult, SnapshotRef,
+    arrow::array::RecordBatch, engine::default::executor::tokio::TokioBackgroundExecutor,
+    engine::default::storage::store_from_url_opts, engine::default::DefaultEngine, scan::Scan,
+    schema::MetadataColumnSpec, DeltaResult, SnapshotRef,
 };
 
 use object_store::{
@@ -158,16 +156,13 @@ pub fn get_engine(
                 )));
             }
         };
-        Ok(DefaultEngine::new(
-            store,
-            Arc::new(TokioBackgroundExecutor::new()),
-        ))
+        Ok(DefaultEngine::new(Arc::new(store)))
     } else if !args.option.is_empty() {
         let opts = args.option.iter().map(|option| {
             let parts: Vec<&str> = option.split("=").collect();
             (parts[0].to_ascii_lowercase(), parts[1])
         });
-        DefaultEngine::try_new(url, opts, Arc::new(TokioBackgroundExecutor::new()))
+        Ok(DefaultEngine::new(store_from_url_opts(url, opts)?))
     } else {
         let mut options = if let Some(ref region) = args.region {
             HashMap::from([("region", region.clone())])
@@ -177,7 +172,7 @@ pub fn get_engine(
         if args.public {
             options.insert("skip_signature", "true".to_string());
         }
-        DefaultEngine::try_new(url, options, Arc::new(TokioBackgroundExecutor::new()))
+        Ok(DefaultEngine::new(store_from_url_opts(url, options)?))
     }
 }
 

kernel/examples/write-table/src/main.rs

@@ -76,11 +76,8 @@ async fn try_main() -> DeltaResult<()> {
     println!("Using Delta table at: {url}");
 
     // Get the engine for local filesystem
-    let engine = DefaultEngine::try_new(
-        &url,
-        HashMap::<String, String>::new(),
-        Arc::new(TokioBackgroundExecutor::new()),
-    )?;
+    use delta_kernel::engine::default::storage::store_from_url;
+    let engine = DefaultEngine::new(store_from_url(&url)?);
 
     // Create or get the table
     let snapshot = create_or_get_base_snapshot(&url, &engine, &cli.schema).await?;

kernel/src/checkpoint/tests.rs

@@ -8,7 +8,7 @@ use crate::arrow::array::{ArrayRef, StructArray};
 use crate::arrow::datatypes::{DataType, Schema};
 use crate::checkpoint::create_last_checkpoint_data;
 use crate::engine::arrow_data::ArrowEngineData;
-use crate::engine::default::{executor::tokio::TokioBackgroundExecutor, DefaultEngine};
+use crate::engine::default::DefaultEngine;
 use crate::log_replay::HasSelectionVector;
 use crate::schema::{DataType as KernelDataType, StructField, StructType};
 use crate::utils::test_utils::Action;
@@ -60,7 +60,7 @@ fn test_deleted_file_retention_timestamp() -> DeltaResult<()> {
 #[test]
 fn test_create_checkpoint_metadata_batch() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // 1st commit (version 0) - metadata and protocol actions
     // Protocol action does not include the v2Checkpoint reader/writer feature.
@@ -116,7 +116,7 @@ fn test_create_last_checkpoint_data() -> DeltaResult<()> {
     let add_actions_counter = 75;
     let size_in_bytes: i64 = 1024 * 1024; // 1MB
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // Create last checkpoint metadata
     let last_checkpoint_batch = create_last_checkpoint_data(
@@ -277,7 +277,7 @@ fn read_last_checkpoint_file(store: &Arc<InMemory>) -> DeltaResult<Value> {
 #[test]
 fn test_v1_checkpoint_latest_version_by_default() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // 1st commit: adds `fake_path_1`
     write_commit_to_store(&store, vec![create_add_action("fake_path_1")], 0)?;
@@ -347,7 +347,7 @@ fn test_v1_checkpoint_latest_version_by_default() -> DeltaResult<()> {
 #[test]
 fn test_v1_checkpoint_specific_version() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // 1st commit (version 0) - metadata and protocol actions
     // Protocol action does not include the v2Checkpoint reader/writer feature.
@@ -409,7 +409,7 @@ fn test_v1_checkpoint_specific_version() -> DeltaResult<()> {
 #[test]
 fn test_finalize_errors_if_checkpoint_data_iterator_is_not_exhausted() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // 1st commit (version 0) - metadata and protocol actions
     write_commit_to_store(
@@ -451,7 +451,7 @@ fn test_finalize_errors_if_checkpoint_data_iterator_is_not_exhausted() -> DeltaR
 #[test]
 fn test_v2_checkpoint_supported_table() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // 1st commit: adds `fake_path_2` & removes `fake_path_1`
     write_commit_to_store(
@@ -523,7 +523,7 @@ fn test_v2_checkpoint_supported_table() -> DeltaResult<()> {
 #[test]
 fn test_no_checkpoint_staged_commits() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
+    let engine = DefaultEngine::new(store.clone());
 
     // normal commit
     write_commit_to_store(

kernel/src/committer.rs

@@ -159,7 +159,6 @@ mod tests {
 
     use std::sync::Arc;
 
-    use crate::engine::default::executor::tokio::TokioBackgroundExecutor;
     use crate::engine::default::DefaultEngine;
     use crate::path::LogRoot;
 
@@ -212,7 +211,7 @@ mod tests {
     async fn catalog_managed_tables_block_transactions() {
         let storage = Arc::new(InMemory::new());
         let table_root = Url::parse("memory:///").unwrap();
-        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(storage.clone());
 
         let actions = [
             r#"{"commitInfo":{"timestamp":12345678900,"inCommitTimestamp":12345678900}}"#,

kernel/src/engine/default/filesystem.rs

@@ -274,7 +274,7 @@ mod tests {
         store.put(&name, data.clone().into()).await.unwrap();
 
         let table_root = Url::parse("memory:///").expect("valid url");
-        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(store);
         let files: Vec<_> = engine
             .storage_handler()
             .list_from(&table_root.join("_delta_log").unwrap().join("0").unwrap())
@@ -304,7 +304,7 @@ mod tests {
 
         let url = Url::from_directory_path(tmp.path()).unwrap();
         let store = Arc::new(LocalFileSystem::new());
-        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(store);
         let files = engine
             .storage_handler()
             .list_from(&url.join("_delta_log").unwrap().join("0").unwrap())