Switch to new data passing API

acceptance/src/lib.rs

@@ -1,4 +1,4 @@
-//! Helpers to validate implementaions of TableClients
+//! Helpers to validate implementaions of EngineClients
 
 pub mod meta;
 pub use meta::*;

acceptance/src/meta.rs

@@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize};
 use url::Url;
 
 use deltakernel::snapshot::Snapshot;
-use deltakernel::{Error, Table, TableClient, Version};
+use deltakernel::{EngineClient, Error, Table, Version};
 
 #[derive(Debug, thiserror::Error)]
 pub enum AssertionError {
@@ -96,17 +96,17 @@ impl TestCaseInfo {
         Ok(())
     }
 
-    pub async fn assert_metadata(&self, table_client: Arc<dyn TableClient>) -> TestResult<()> {
-        let table_client = table_client.as_ref();
+    pub async fn assert_metadata(&self, engine_client: Arc<dyn EngineClient>) -> TestResult<()> {
+        let engine_client = engine_client.as_ref();
         let table = Table::new(self.table_root()?);
 
         let (latest, versions) = self.versions().await?;
 
-        let snapshot = table.snapshot(table_client, None)?;
+        let snapshot = table.snapshot(engine_client, None)?;
         self.assert_snapshot_meta(&latest, &snapshot)?;
 
         for table_version in versions {
-            let snapshot = table.snapshot(table_client, Some(table_version.version))?;
+            let snapshot = table.snapshot(engine_client, Some(table_version.version))?;
             self.assert_snapshot_meta(&table_version, &snapshot)?;
         }
 

acceptance/tests/dat_reader.rs

@@ -18,7 +18,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 table_client = Arc::new(
+            let engine_client = Arc::new(
                 DefaultTableClient::try_new(
                     &table_root,
                     std::iter::empty::<(&str, &str)>(),
@@ -27,7 +27,7 @@ fn reader_test(path: &Path) -> datatest_stable::Result<()> {
                 .unwrap(),
             );
 
-            case.assert_metadata(table_client.clone()).await.unwrap();
+            case.assert_metadata(engine_client.clone()).await.unwrap();
         });
     Ok(())
 }

acceptance/tests/other.rs

@@ -38,10 +38,10 @@ async fn test_read_table_with_checkpoint() {
     ))
     .unwrap();
     let location = url::Url::from_directory_path(path).unwrap();
-    let table_client = Arc::new(
+    let engine_client = Arc::new(
         DefaultTableClient::try_new(&location, HashMap::<String, String>::new()).unwrap(),
     );
-    let snapshot = Snapshot::try_new(location, table_client, None)
+    let snapshot = Snapshot::try_new(location, engine_client, None)
         .await
         .unwrap();
 

kernel/Cargo.toml

@@ -39,26 +39,25 @@ visibility = "0.1.0"
 # Used in default client
 futures = { version = "0.3", optional = true }
 object_store = { version = "^0.8.0", optional = true }
-parquet = { version = "^49.0", optional = true, features = [
-  "async",
-  "object_store",
-] }
+# Used in default and simple client
+parquet = { version = "^49.0", optional = true }
 
 # optionally used with default client (though not required)
 tokio = { version = "1", optional = true, features = ["rt-multi-thread"] }
 
 [features]
-default = ["default-client"]
-default-client = ["chrono", "futures", "object_store", "parquet"]
+arrow-conversion = []
+default = ["simple-client"]
+default-client = ["arrow-conversion", "chrono", "futures", "object_store", "parquet/async", "parquet/object_store"]
 developer-visibility = []
+simple-client = ["arrow-conversion", "parquet"]
 
 [dev-dependencies]
 arrow = { version = "^49.0", features = ["json", "prettyprint"] }
-deltakernel = { path = ".", features = ["tokio"] }
+deltakernel = { path = ".", features = ["tokio", "default-client"] }
 test-log = { version = "0.2", default-features = false, features = ["trace"] }
 tempfile = "3"
 test-case = { version = "3.1.0" }
-tokio = { version = "1" }
 tracing-subscriber = { version = "0.3", default-features = false, features = [
   "env-filter",
   "fmt",

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

@@ -86,21 +86,21 @@ fn main() {
         println!("Invalid url");
         return;
     };
-    let table_client = DefaultTableClient::try_new(
+    let engine_client = DefaultTableClient::try_new(
         &url,
         HashMap::<String, String>::new(),
         Arc::new(TokioBackgroundExecutor::new()),
     );
-    let Ok(table_client) = table_client else {
+    let Ok(engine_client) = engine_client else {
         println!(
             "Failed to construct table client: {}",
-            table_client.err().unwrap()
+            engine_client.err().unwrap()
         );
         return;
     };
 
     let table = Table::new(url);
-    let snapshot = table.snapshot(&table_client, None);
+    let snapshot = table.snapshot(&engine_client, None);
     let Ok(snapshot) = snapshot else {
         println!(
             "Failed to construct latest snapshot: {}",
@@ -127,7 +127,7 @@ fn main() {
     }
     table.set_header(header_names);
 
-    for batch in scan.execute(&table_client).unwrap() {
+    for batch in scan.execute(&engine_client).unwrap() {
         for row in 0..batch.num_rows() {
             let table_row =
                 (0..batch.num_columns()).map(|col| extract_value(batch.column(col), row));

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

@@ -54,21 +54,21 @@ fn main() {
         println!("Invalid url");
         return;
     };
-    let table_client = DefaultTableClient::try_new(
+    let engine_client = DefaultTableClient::try_new(
         &url,
         HashMap::<String, String>::new(),
         Arc::new(TokioBackgroundExecutor::new()),
     );
-    let Ok(table_client) = table_client else {
+    let Ok(engine_client) = engine_client else {
         println!(
             "Failed to construct table client: {}",
-            table_client.err().unwrap()
+            engine_client.err().unwrap()
         );
         return;
     };
 
     let table = Table::new(url);
-    let snapshot = table.snapshot(&table_client, None);
+    let snapshot = table.snapshot(&engine_client, None);
     let Ok(snapshot) = snapshot else {
         println!(
             "Failed to construct latest snapshot: {}",
@@ -91,7 +91,7 @@ fn main() {
             use deltakernel::Add;
             let scan = ScanBuilder::new(snapshot).build();
             let files: Vec<Add> = scan
-                .files(&table_client)
+                .files(&engine_client)
                 .unwrap()
                 .map(|r| r.unwrap())
                 .collect();
@@ -116,7 +116,7 @@ fn main() {
 
             let batches = snapshot
                 ._log_segment()
-                .replay(&table_client, read_schema, None);
+                .replay(&engine_client, read_schema, None);
 
             let batch_vec = batches
                 .unwrap()