refactor!: make SyncEngine test-only, use DefaultEngine everywhere else (#957)

## What changes are proposed in this pull request?
Again prompted by the crate split exploration (#941). Notice that the
`SyncEngine` is not really useful to external users, instead just adding
to our public API and complicating our feature flags. We expect users to
reach for the default engine in the case they want a 'prebuilt' solution
with minimal deps and sane defaults (arrow, tokio). The `SyncEngine`
will continue to exist as a test-only engine (for now; and will maybe
evolve in the future).

This PR includes
1. removing `sync-engine` feature flag (in core kernel and FFI), and
making the existing `engine::sync` module `#[cfg(test)]`
2. removing any consumption of the sync engine in (1) examples and (2)
integration tests
3. adding a simple test_util trait `DefaultEngineExtension ` and a
simple method to return a new local default engine (to more easily mimic
SyncEngine API for easy replacement)

### This PR affects the following public APIs
- no more `SyncEngine`, no more `sync-engine` feature flag

## How was this change tested?
Existing

Author: zachschuermann

.github/workflows/build.yml

@@ -178,7 +178,7 @@ jobs:
           cargo build
           popd
           pushd ffi
-          cargo b --features default-engine,sync-engine,test-ffi,tracing
+          cargo b --features default-engine,test-ffi,tracing
           popd
       - name: build and run read-table test
         run: |
@@ -210,7 +210,7 @@ jobs:
       - name: Test with Miri
         run: |
           pushd ffi
-          MIRIFLAGS=-Zmiri-disable-isolation cargo miri test --features sync-engine,default-engine
+          MIRIFLAGS=-Zmiri-disable-isolation cargo miri test --features default-engine
 
   coverage:
     runs-on: ubuntu-latest

README.md

@@ -65,7 +65,6 @@ include:
 | Feature flag  | Description   |
 | ------------- | ------------- |
 | `default-engine`    | Turn on the 'default' engine: async, arrow-based `Engine` implementation  |
-| `sync-engine`       | Turn on the 'sync' engine: synchronous, arrow-based `Engine` implementation. Only supports local storage! |
 | `arrow-conversion`  | Conversion utilities for arrow/kernel schema interoperation |
 | `arrow-expression`  | Expression system implementation for arrow |
 
@@ -75,13 +74,13 @@ are still unstable. We therefore may break APIs within minor releases (that is,
 we will not break APIs in patch releases (`0.1.0` -> `0.1.1`).
 
 ## Arrow versioning
-If you enable the `default-engine` or `sync-engine` features, you get an implementation of the
-`Engine` trait that uses [Arrow] as its data format.
+If you enable the `default-engine` feature, you get an implementation of the `Engine` trait that
+uses [Arrow] as its data format.
 
 The [`arrow crate`](https://docs.rs/arrow/latest/arrow/) tends to release new major versions rather
-quickly. To enable engines that already integrate arrow to also integrate kernel and not force them
-to track a specific version of arrow that kernel depends on, we take as broad dependency on arrow
-versions as we can.
+frequently. To enable engines that already integrate arrow to also integrate kernel and not force
+them to track a specific version of arrow that kernel depends on, we take as broad dependency on
+arrow versions as we can.
 
 We allow selecting the version of arrow to use via feature flags. Currently we support the following
 flags:
@@ -97,10 +96,6 @@ Note that if more than one `arrow-x` feature is enabled, kernel will use the _hi
 specified flag. This also means that if you use `--all-features` you will get the latest version of
 arrow that kernel supports.
 
-If you enable at least one of `default-engine`, `sync-engine`, `arrow-conversion`, or
-`arrow-expression`, you must enable either `arrow` (latest arrow version) or `arrow-54` or
-`arrow-55`.
-
 ### Object Store
 You may also need to patch the `object_store` version used if the version of `parquet` you depend on
 depends on a different version of `object_store`. This can be done by including `object_store` in

ffi/Cargo.toml

@@ -40,6 +40,5 @@ trybuild = "1.0"
 default = ["default-engine"]
 default-engine = ["delta_kernel/default-engine", "delta_kernel/arrow"]
 tracing = [ "tracing-core", "tracing-subscriber" ]
-sync-engine = ["delta_kernel/sync-engine", "delta_kernel/arrow"]
 internal-api = []
 test-ffi = []

ffi/cbindgen.toml

@@ -11,7 +11,6 @@ namespace = "ffi"
 
 [defines]
 "feature = default-engine" = "DEFINE_DEFAULT_ENGINE"
-"feature = sync-engine" = "DEFINE_SYNC_ENGINE"
 
 [export.mangle]
 remove_underscores = true

ffi/src/error.rs

@@ -8,13 +8,13 @@ use crate::{kernel_string_slice, ExternEngine, KernelStringSlice};
 pub enum KernelError {
     UnknownError, // catch-all for unrecognized kernel Error types
     FFIError,     // errors encountered in the code layer that supports FFI
-    #[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+    #[cfg(feature = "default-engine")]
     ArrowError,
     EngineDataTypeError,
     ExtractError,
     GenericError,
     IOErrorError,
-    #[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+    #[cfg(feature = "default-engine")]
     ParquetError,
     #[cfg(feature = "default-engine")]
     ObjectStoreError,
@@ -62,15 +62,15 @@ impl From<Error> for KernelError {
     fn from(e: Error) -> Self {
         match e {
             // NOTE: By definition, no kernel Error maps to FFIError
-            #[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+            #[cfg(feature = "default-engine")]
             Error::Arrow(_) => KernelError::ArrowError,
             Error::CheckpointWrite(_) => KernelError::CheckpointWriteError,
             Error::EngineDataType(_) => KernelError::EngineDataTypeError,
             Error::Extract(..) => KernelError::ExtractError,
             Error::Generic(_) => KernelError::GenericError,
             Error::GenericError { .. } => KernelError::GenericError,
             Error::IOError(_) => KernelError::IOErrorError,
-            #[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+            #[cfg(feature = "default-engine")]
             Error::Parquet(_) => KernelError::ParquetError,
             #[cfg(feature = "default-engine")]
             Error::ObjectStore(_) => KernelError::ObjectStoreError,

ffi/src/lib.rs

@@ -355,14 +355,14 @@ pub trait ExternEngine: Send + Sync {
 #[handle_descriptor(target=dyn ExternEngine, mutable=false)]
 pub struct SharedExternEngine;
 
-#[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+#[cfg(feature = "default-engine")]
 struct ExternEngineVtable {
     // Actual engine instance to use
     engine: Arc<dyn Engine>,
     allocate_error: AllocateErrorFn,
 }
 
-#[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+#[cfg(feature = "default-engine")]
 impl Drop for ExternEngineVtable {
     fn drop(&mut self) {
         debug!("dropping engine interface");
@@ -373,7 +373,7 @@ impl Drop for ExternEngineVtable {
 ///
 /// Kernel doesn't use any threading or concurrency. If engine chooses to do so, engine is
 /// responsible for handling  any races that could result.
-#[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+#[cfg(feature = "default-engine")]
 unsafe impl Send for ExternEngineVtable {}
 
 /// # Safety
@@ -385,10 +385,10 @@ unsafe impl Send for ExternEngineVtable {}
 /// Basically, by failing to implement these traits, we forbid the engine from being able to declare
 /// its thread-safety (because rust assumes it is not threadsafe). By implementing them, we leave it
 /// up to the engine to enforce thread safety if engine chooses to use threads at all.
-#[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+#[cfg(feature = "default-engine")]
 unsafe impl Sync for ExternEngineVtable {}
 
-#[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+#[cfg(feature = "default-engine")]
 impl ExternEngine for ExternEngineVtable {
     fn engine(&self) -> Arc<dyn Engine> {
         self.engine.clone()
@@ -513,18 +513,7 @@ fn get_default_default_engine_impl(
     get_default_engine_impl(url?, Default::default(), allocate_error)
 }
 
-/// # Safety
-///
-/// Caller is responsible for passing a valid path pointer.
-#[cfg(feature = "sync-engine")]
-#[no_mangle]
-pub unsafe extern "C" fn get_sync_engine(
-    allocate_error: AllocateErrorFn,
-) -> ExternResult<Handle<SharedExternEngine>> {
-    get_sync_engine_impl(allocate_error).into_extern_result(&allocate_error)
-}
-
-#[cfg(any(feature = "default-engine", feature = "sync-engine"))]
+#[cfg(feature = "default-engine")]
 fn engine_to_handle(
     engine: Arc<dyn Engine>,
     allocate_error: AllocateErrorFn,
@@ -552,14 +541,6 @@ fn get_default_engine_impl(
     Ok(engine_to_handle(Arc::new(engine?), allocate_error))
 }
 
-#[cfg(feature = "sync-engine")]
-fn get_sync_engine_impl(
-    allocate_error: AllocateErrorFn,
-) -> DeltaResult<Handle<SharedExternEngine>> {
-    let engine = delta_kernel::engine::sync::SyncEngine::new();
-    Ok(engine_to_handle(Arc::new(engine), allocate_error))
-}
-
 /// # Safety
 ///
 /// Caller is responsible for passing a valid handle.
@@ -904,14 +885,4 @@ mod tests {
         unsafe { free_engine(engine) }
         Ok(())
     }
-
-    #[test]
-    #[cfg(feature = "sync-engine")]
-    fn sync_engine() {
-        let engine = unsafe { get_sync_engine(allocate_err) };
-        let engine = ok_or_panic(engine);
-        unsafe {
-            free_engine(engine);
-        }
-    }
 }

integration-tests/Cargo.toml

@@ -6,4 +6,4 @@ edition = "2021"
 [workspace]
 
 [dependencies]
-delta_kernel = { path = "../kernel", features = ["default-engine", "sync-engine"] }
+delta_kernel = { path = "../kernel", features = ["default-engine"] }

kernel/Cargo.toml

@@ -54,7 +54,6 @@ uuid = { version = "1.16.0", features = ["v4", "fast-rng"] }
 z85 = "3.0.6"
 
 # optional deps
-tempfile = { version = "3", optional = true }
 futures = { version = "0.3", optional = true }
 # Used for fetching direct urls (like pre-signed urls)
 reqwest = { version = "0.12.15", default-features = false, optional = true }
@@ -128,16 +127,12 @@ default-engine-rustls = [
   "reqwest/rustls-tls-native-roots",
   "reqwest/http2",
 ]
-sync-engine = [
-  "need-arrow",
-  "tempfile",
-]
 
 [build-dependencies]
 rustc_version = "0.4.1"
 
 [dev-dependencies]
-delta_kernel = { path = ".", features = ["arrow", "default-engine", "sync-engine"] }
+delta_kernel = { path = ".", features = ["arrow", "default-engine"] }
 test_utils = { path = "../test-utils" }
 # Used for testing parse_url_opts extensibility
 hdfs-native-object-store = { version = "0.14.0" }

kernel/examples/read-table-multi-threaded/Cargo.toml

@@ -10,7 +10,6 @@ clap = { version = "4.5", features = ["derive"] }
 delta_kernel = { path = "../../../kernel", features = [
   "arrow-55",
   "default-engine",
-  "sync-engine",
   "internal-api",
 ] }
 env_logger = "0.11.8"

kernel/examples/read-table-multi-threaded/src/main.rs

@@ -11,12 +11,11 @@ use delta_kernel::actions::deletion_vector::split_vector;
 use delta_kernel::engine::arrow_data::ArrowEngineData;
 use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
 use delta_kernel::engine::default::DefaultEngine;
-use delta_kernel::engine::sync::SyncEngine;
 use delta_kernel::scan::state::{transform_to_logical, DvInfo, GlobalScanState, Stats};
 use delta_kernel::schema::Schema;
 use delta_kernel::{DeltaResult, Engine, EngineData, ExpressionRef, FileMeta, Table};
 
-use clap::{Parser, ValueEnum};
+use clap::Parser;
 use url::Url;
 
 /// An example program that reads a table using multiple threads. This shows the use of the
@@ -33,10 +32,6 @@ struct Cli {
     #[arg(short, long, default_value_t = 2, value_parser = 1..=2048)]
     thread_count: i64,
 
-    /// Which Engine to use
-    #[arg(short, long, value_enum, default_value_t = EngineType::Default)]
-    engine: EngineType,
-
     /// Comma separated list of columns to select
     #[arg(long, value_delimiter=',', num_args(0..))]
     columns: Option<Vec<String>>,
@@ -56,14 +51,6 @@ struct Cli {
     limit: Option<usize>,
 }
 
-#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
-enum EngineType {
-    /// Use the default, async engine
-    Default,
-    /// Use the sync engine (local files only)
-    Sync,
-}
-
 fn main() -> ExitCode {
     env_logger::init();
     match try_main() {
@@ -129,27 +116,21 @@ fn try_main() -> DeltaResult<()> {
     let table = Table::try_from_uri(&cli.path)?;
     println!("Reading {}", table.location());
 
-    // create the requested engine
-    let engine: Arc<dyn Engine> = match cli.engine {
-        EngineType::Default => {
-            let mut options = if let Some(region) = cli.region {
-                HashMap::from([("region", region)])
-            } else {
-                HashMap::new()
-            };
-            if cli.public {
-                options.insert("skip_signature", "true".to_string());
-            }
-            Arc::new(DefaultEngine::try_new(
-                table.location(),
-                options,
-                Arc::new(TokioBackgroundExecutor::new()),
-            )?)
-        }
-        EngineType::Sync => Arc::new(SyncEngine::new()),
+    let mut options = if let Some(region) = cli.region {
+        HashMap::from([("region", region)])
+    } else {
+        HashMap::new()
     };
+    if cli.public {
+        options.insert("skip_signature", "true".to_string());
+    }
+    let engine = DefaultEngine::try_new(
+        table.location(),
+        options,
+        Arc::new(TokioBackgroundExecutor::new()),
+    )?;
 
-    let snapshot = table.snapshot(engine.as_ref(), None)?;
+    let snapshot = table.snapshot(&engine, None)?;
 
     // process the columns requested and build a schema from them
     let read_schema_opt = cli
@@ -179,7 +160,7 @@ fn try_main() -> DeltaResult<()> {
     // [`delta_kernel::scan::scan_row_schema`]. Generally engines will not need to interact with
     // this data directly, and can just call [`visit_scan_files`] to get pre-parsed data back from
     // the kernel.
-    let scan_metadata = scan.scan_metadata(engine.as_ref())?;
+    let scan_metadata = scan.scan_metadata(&engine)?;
 
     // get any global state associated with this scan
     let global_state = Arc::new(scan.global_scan_state());
@@ -192,18 +173,17 @@ fn try_main() -> DeltaResult<()> {
 
     // fire up each thread. we don't need the handles as we rely on the channels to indicate when
     // things are done
-    let _handles: Vec<_> = (0..cli.thread_count)
-        .map(|_| {
+    thread::scope(|s| {
+        (0..cli.thread_count).for_each(|_| {
             // items that we need to send to the other thread
             let scan_state = global_state.clone();
             let rb_tx = record_batch_tx.clone();
             let scan_file_rx = scan_file_rx.clone();
-            let engine = engine.clone();
-            thread::spawn(move || {
-                do_work(engine, scan_state, rb_tx, scan_file_rx);
-            })
-        })
-        .collect();
+            s.spawn(|| {
+                do_work(&engine, scan_state, rb_tx, scan_file_rx);
+            });
+        });
+    });
 
     // have handed out all copies needed, drop so record_batch_rx will exit when the last thread is
     // done sending
@@ -244,13 +224,11 @@ fn try_main() -> DeltaResult<()> {
 
 // this is the work each thread does
 fn do_work(
-    engine: Arc<dyn Engine>,
+    engine: &dyn Engine,
     scan_state: Arc<GlobalScanState>,
     record_batch_tx: Sender<RecordBatch>,
     scan_file_rx: spmc::Receiver<ScanFile>,
 ) {
-    // get the type for the function calls
-    let engine: &dyn Engine = engine.as_ref();
     // in a loop, try and get a ScanFile. Note that `recv` will return an `Err` when the other side
     // hangs up, which indicates there's no more data to process.
     while let Ok(scan_file) = scan_file_rx.recv() {

kernel/examples/read-table-single-threaded/Cargo.toml

@@ -10,7 +10,6 @@ clap = { version = "4.5", features = ["derive"] }
 delta_kernel = { path = "../../../kernel", features = [
   "arrow-55",
   "default-engine",
-  "sync-engine",
   "internal-api",
 ] }
 env_logger = "0.11.8"