test fixes part1: use sync engine in kernel, otherwise just use default. misc fixes

Author: zachschuermann

delta-kernel-engine/Cargo.toml

@@ -72,4 +72,5 @@ arrow-55 = [ "dep:arrow_55", "dep:parquet_55", "dep:object_store_55" ]
 rustls = [ "reqwest/rustls-tls-native-roots", "reqwest/http2" ]
 
 [dev-dependencies]
+test_utils = { path = "../test-utils" }
 async-trait = "0.1" # only used for our custom SlowGetStore ObjectStore implementation

delta-kernel-engine/src/arrow_expression/tests.rs

@@ -8,9 +8,9 @@ use crate::arrow::buffer::{OffsetBuffer, ScalarBuffer};
 use crate::arrow::datatypes::{DataType, Field, Fields, Schema};
 
 use super::*;
-use crate::expressions::*;
+use delta_kernel::expressions::*;
+use delta_kernel::schema::DataType as DeltaDataTypes;
 use delta_kernel::schema::{ArrayType, MapType, StructField, StructType};
-use delta_kernel::DataType as DeltaDataTypes;
 use delta_kernel::EvaluationHandlerExtension as _;
 
 use Expression as Expr;
@@ -139,7 +139,7 @@ fn test_literal_complex_type_array() {
     ];
     let struct_type = StructType::new(struct_fields.clone());
     let struct_value = Scalar::Struct(
-        crate::expressions::StructData::try_new(
+        StructData::try_new(
             struct_fields.clone(),
             vec![
                 Scalar::Integer(42),
@@ -457,11 +457,11 @@ fn test_null_row() {
         StructField::nullable(
             "x",
             StructType::new([
-                StructField::nullable("a", crate::schema::DataType::INTEGER),
-                StructField::not_null("b", crate::schema::DataType::STRING),
+                StructField::nullable("a", DeltaDataTypes::INTEGER),
+                StructField::not_null("b", DeltaDataTypes::STRING),
             ]),
         ),
-        StructField::nullable("c", crate::schema::DataType::STRING),
+        StructField::nullable("c", DeltaDataTypes::STRING),
     ]));
     let handler = ArrowEvaluationHandler;
     let result = handler.null_row(schema.clone()).unwrap();
@@ -492,7 +492,7 @@ fn test_null_row() {
 fn test_null_row_err() {
     let not_null_schema = Arc::new(StructType::new(vec![StructField::not_null(
         "a",
-        crate::schema::DataType::STRING,
+        DeltaDataTypes::STRING,
     )]));
     let handler = ArrowEvaluationHandler;
     assert!(handler.null_row(not_null_schema).is_err());

delta-kernel-engine/src/default/parquet.rs

@@ -405,7 +405,8 @@ mod tests {
 
     use crate::arrow_data::ArrowEngineData;
     use crate::default::executor::tokio::TokioBackgroundExecutor;
-    use crate::EngineData;
+
+    use delta_kernel::{transaction, EngineData};
 
     use itertools::Itertools;
 
@@ -472,7 +473,7 @@ mod tests {
         let actual = ArrowEngineData::try_from_engine_data(actual).unwrap();
 
         let schema = Arc::new(
-            crate::transaction::get_write_metadata_schema()
+            transaction::get_write_metadata_schema()
                 .as_ref()
                 .try_into()
                 .unwrap(),

delta-kernel-engine/src/lib.rs

@@ -10,7 +10,8 @@ pub mod parquet_row_group_skipping;
 mod arrow_compat;
 pub use arrow_compat::*;
 
-pub(crate) mod arrow_conversion;
+pub mod arrow_conversion;
+
 pub(crate) mod arrow_get_data;
 pub(crate) mod arrow_utils;
 pub(crate) mod ensure_data_types;
@@ -36,7 +37,7 @@ mod tests {
     use crate::arrow::array::{RecordBatch, StringArray};
     use crate::arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
     use crate::arrow_data::ArrowEngineData;
-    use crate::{Engine, EngineData};
+    use delta_kernel::{Engine, EngineData};
 
     use test_utils::delta_path_for_version;
 

kernel/Cargo.toml

@@ -63,7 +63,7 @@ rustc_version = "0.4.1"
 delta-kernel-engine = { path = "../delta-kernel-engine", features = ["arrow-55"] }
 test_utils = { path = "../test-utils" }
 # Used for testing parse_url_opts extensibility
-hdfs-native-object-store = { version = "0.14.0" }
+hdfs-native-object-store = { version = "0.14.0", features = ["integration-test"] }
 hdfs-native = "0.11.1"
 walkdir = { version = "2.5.0" }
 paste = "1.0"
@@ -75,3 +75,6 @@ tracing-subscriber = { version = "0.3", default-features = false, features = [
   "env-filter",
   "fmt",
 ] }
+# TODO: remove these - only used for write.rs example
+tokio = { version = "1", features = ["full"] }
+futures = "0.3"

kernel/src/engine/mod.rs

@@ -1,13 +1,14 @@
 //! A simple, single threaded, [`Engine`] that can only read from the local filesystem
 
-use super::arrow_expression::ArrowEvaluationHandler;
-use crate::engine::arrow_data::ArrowEngineData;
 use crate::{
     DeltaResult, Engine, Error, EvaluationHandler, FileDataReadResultIterator, FileMeta,
     JsonHandler, ParquetHandler, PredicateRef, SchemaRef, StorageHandler,
 };
+use delta_kernel_engine::arrow_conversion::something;
+use delta_kernel_engine::arrow_data::ArrowEngineData;
+use delta_kernel_engine::arrow_expression::ArrowEvaluationHandler;
 
-use crate::arrow::datatypes::{Schema as ArrowSchema, SchemaRef as ArrowSchemaRef};
+use delta_kernel_engine::arrow::datatypes::{Schema as ArrowSchema, SchemaRef as ArrowSchemaRef};
 use itertools::Itertools;
 use std::fs::File;
 use std::sync::Arc;
@@ -19,7 +20,7 @@ mod storage;
 
 /// This is a simple implementation of [`Engine`]. It only supports reading data from the local
 /// filesystem, and internally represents data using `Arrow`.
-pub struct SyncEngine {
+pub(crate) struct SyncEngine {
     storage_handler: Arc<storage::SyncStorageHandler>,
     json_handler: Arc<json::SyncJsonHandler>,
     parquet_handler: Arc<parquet::SyncParquetHandler>,
@@ -101,7 +102,7 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::engine::tests::test_arrow_engine;
+    use delta_kernel_engine::tests::test_arrow_engine;
 
     #[test]
     fn test_sync_engine() {

kernel/src/lib.rs

@@ -405,7 +405,7 @@ pub trait EvaluationHandler: AsAny {
 /// EvaluationHandlers.
 // For some reason rustc doesn't detect it's usage so we allow(dead_code) here...
 #[allow(dead_code)]
-trait EvaluationHandlerExtension: EvaluationHandler {
+pub trait EvaluationHandlerExtension: EvaluationHandler {
     /// Create a single-row [`EngineData`] by applying the given schema to the leaf-values given in
     /// `values`.
     // Note: we will stick with a Schema instead of DataType (more constrained can expand in

kernel/src/table.rs

@@ -103,7 +103,7 @@ impl Table {
     /// Creates a [`CheckpointWriter`] for generating checkpoints at the specified table version.
     ///
     /// See the [`crate::checkpoint`] module documentation for more details on checkpoint types
-    /// and the overall checkpoint process.    
+    /// and the overall checkpoint process.
     ///
     /// # Parameters
     /// - `engine`: Implementation of [`Engine`] apis.
@@ -165,7 +165,7 @@ mod tests {
     use std::path::PathBuf;
 
     use super::*;
-    use crate::engine::sync::SyncEngine;
+    use crate::engine::SyncEngine;
 
     #[test]
     fn test_table() {

kernel/tests/common/mod.rs

@@ -1,11 +1,12 @@
-use delta_kernel::arrow::compute::filter_record_batch;
-use delta_kernel::arrow::record_batch::RecordBatch;
-use delta_kernel::arrow::util::pretty::pretty_format_batches;
+use delta_kernel_engine::arrow::compute::filter_record_batch;
+use delta_kernel_engine::arrow::record_batch::RecordBatch;
+use delta_kernel_engine::arrow::util::pretty::pretty_format_batches;
+
 use itertools::Itertools;
 
-use crate::ArrowEngineData;
 use delta_kernel::scan::Scan;
 use delta_kernel::{DeltaResult, Engine, EngineData, Table};
+use delta_kernel_engine::arrow_data::ArrowEngineData;
 
 use std::sync::Arc;
 
@@ -24,7 +25,7 @@ macro_rules! sort_lines {
 #[macro_export]
 macro_rules! assert_batches_sorted_eq {
     ($expected_lines_sorted: expr, $CHUNKS: expr) => {
-        let formatted = delta_kernel::arrow::util::pretty::pretty_format_batches($CHUNKS)
+        let formatted = delta_kernel_engine::arrow::util::pretty::pretty_format_batches($CHUNKS)
             .unwrap()
             .to_string();
         // fix for windows: \r\n -->
@@ -95,7 +96,11 @@ pub(crate) fn read_scan(scan: &Scan, engine: Arc<dyn Engine>) -> DeltaResult<Vec
             let data = scan_result.raw_data?;
             let record_batch = to_arrow(data)?;
             if let Some(mask) = mask {
-                Ok(filter_record_batch(&record_batch, &mask.into())?)
+                Ok(
+                    filter_record_batch(&record_batch, &mask.into()).map_err(|e| {
+                        delta_kernel::Error::Generic(format!("Failed to filter record batch: {e}"))
+                    })?,
+                )
             } else {
                 Ok(record_batch)
             }

kernel/tests/dv.rs

@@ -4,7 +4,7 @@ use std::ops::Add;
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use delta_kernel::engine::sync::SyncEngine;
+use delta_kernel::engine::SyncEngine;
 use delta_kernel::scan::ScanResult;
 use delta_kernel::{DeltaResult, Table};
 

kernel/tests/hdfs.rs

@@ -5,11 +5,11 @@
 //
 // Run these integration tests with:
 //   cargo test --features integration-test --test hdfs
-#![cfg(all(feature = "integration-test", not(target_os = "windows")))]
+#![cfg(not(target_os = "windows"))]
 
-use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
-use delta_kernel::engine::default::DefaultEngine;
 use delta_kernel::Table;
+use delta_kernel_engine::default::executor::tokio::TokioBackgroundExecutor;
+use delta_kernel_engine::default::DefaultEngine;
 use hdfs_native::{Client, WriteOptions};
 use hdfs_native_object_store::minidfs::MiniDfs;
 use std::collections::HashSet;
@@ -19,6 +19,7 @@ use std::sync::Arc;
 extern crate walkdir;
 use walkdir::WalkDir;
 
+#[allow(unused)]
 async fn write_local_path_to_hdfs(
     local_path: &Path,
     remote_path: &Path,
@@ -51,7 +52,8 @@ async fn write_local_path_to_hdfs(
     Ok(())
 }
 
-#[tokio::test]
+#[ignore]
+#[allow(unused)]
 async fn read_table_version_hdfs() -> Result<(), Box<dyn std::error::Error>> {
     let minidfs = MiniDfs::with_features(&HashSet::new());
     let hdfs_client = Client::default();

kernel/tests/read.rs

@@ -2,22 +2,24 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use delta_kernel_engine::arrow::compute::{concat_batches, filter_record_batch};
+use delta_kernel_engine::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use delta_kernel_engine::object_store::{memory::InMemory, path::Path, ObjectStore};
+use delta_kernel_engine::parquet::file::properties::{EnabledStatistics, WriterProperties};
+
 use delta_kernel::actions::deletion_vector::split_vector;
-use delta_kernel::arrow::compute::{concat_batches, filter_record_batch};
-use delta_kernel::arrow::datatypes::SchemaRef as ArrowSchemaRef;
-use delta_kernel::engine::arrow_data::ArrowEngineData;
-use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
-use delta_kernel::engine::default::DefaultEngine;
 use delta_kernel::expressions::{
     column_expr, column_pred, BinaryPredicateOp, Expression as Expr, ExpressionRef,
     Predicate as Pred,
 };
-use delta_kernel::object_store::{memory::InMemory, path::Path, ObjectStore};
-use delta_kernel::parquet::file::properties::{EnabledStatistics, WriterProperties};
 use delta_kernel::scan::state::{transform_to_logical, DvInfo, Stats};
 use delta_kernel::scan::Scan;
 use delta_kernel::schema::{DataType, Schema};
 use delta_kernel::{Engine, FileMeta, Table};
+use delta_kernel_engine::arrow_conversion::arrow_schema_from_struct_type;
+use delta_kernel_engine::default::executor::tokio::TokioBackgroundExecutor;
+use delta_kernel_engine::default::DefaultEngine;
+
 use itertools::Itertools;
 use test_utils::{
     actions_to_string, add_commit, generate_batch, generate_simple_batch, into_record_batch,
@@ -316,7 +318,8 @@ fn read_with_execute(
     scan: &Scan,
     expected: &[String],
 ) -> Result<(), Box<dyn std::error::Error>> {
-    let result_schema: ArrowSchemaRef = Arc::new(scan.schema().as_ref().try_into()?);
+    let result_schema: ArrowSchemaRef =
+        Arc::new(arrow_schema_from_struct_type(scan.schema().as_ref())?);
     let batches = read_scan(scan, engine)?;
 
     if expected.is_empty() {
@@ -359,7 +362,8 @@ fn read_with_scan_metadata(
     expected: &[String],
 ) -> Result<(), Box<dyn std::error::Error>> {
     let global_state = scan.global_scan_state();
-    let result_schema: ArrowSchemaRef = Arc::new(scan.schema().as_ref().try_into()?);
+    let result_schema: ArrowSchemaRef =
+        Arc::new(arrow_schema_from_struct_type(scan.schema().as_ref())?);
     let scan_metadata = scan.scan_metadata(engine)?;
     let mut scan_files = vec![];
     for res in scan_metadata {
@@ -431,36 +435,32 @@ fn read_table_data(
     let path = std::fs::canonicalize(PathBuf::from(path))?;
     let predicate = predicate.map(Arc::new);
     let url = url::Url::from_directory_path(path).unwrap();
-    let default_engine = DefaultEngine::try_new(
+    let default_engine = Arc::new(DefaultEngine::try_new(
         &url,
         std::iter::empty::<(&str, &str)>(),
         Arc::new(TokioBackgroundExecutor::new()),
-    )?;
-    let sync_engine = delta_kernel::engine::sync::SyncEngine::new();
-
-    let engines: Vec<Arc<dyn Engine>> = vec![Arc::new(sync_engine), Arc::new(default_engine)];
-    for engine in engines {
-        let table = Table::new(url.clone());
-        let snapshot = table.snapshot(engine.as_ref(), None)?;
-
-        let read_schema = select_cols.map(|select_cols| {
-            let table_schema = snapshot.schema();
-            let selected_fields = select_cols
-                .iter()
-                .map(|col| table_schema.field(col).cloned().unwrap());
-            Arc::new(Schema::new(selected_fields))
-        });
-        println!("Read {url:?} with schema {read_schema:#?} and predicate {predicate:#?}");
-        let scan = snapshot
-            .into_scan_builder()
-            .with_schema_opt(read_schema)
-            .with_predicate(predicate.clone())
-            .build()?;
+    )?);
 
-        sort_lines!(expected);
-        read_with_scan_metadata(table.location(), engine.as_ref(), &scan, &expected)?;
-        read_with_execute(engine, &scan, &expected)?;
-    }
+    let table = Table::new(url.clone());
+    let snapshot = table.snapshot(default_engine.as_ref(), None)?;
+
+    let read_schema = select_cols.map(|select_cols| {
+        let table_schema = snapshot.schema();
+        let selected_fields = select_cols
+            .iter()
+            .map(|col| table_schema.field(col).cloned().unwrap());
+        Arc::new(Schema::new(selected_fields))
+    });
+    println!("Read {url:?} with schema {read_schema:#?} and predicate {predicate:#?}");
+    let scan = snapshot
+        .into_scan_builder()
+        .with_schema_opt(read_schema)
+        .with_predicate(predicate.clone())
+        .build()?;
+
+    sort_lines!(expected);
+    read_with_scan_metadata(table.location(), default_engine.as_ref(), &scan, &expected)?;
+    read_with_execute(default_engine, &scan, &expected)?;
     Ok(())
 }
 

kernel/tests/v2_checkpoints.rs

@@ -1,9 +1,10 @@
 use std::sync::Arc;
 
-use delta_kernel::arrow::array::RecordBatch;
-use delta_kernel::engine::sync::SyncEngine;
+use delta_kernel_engine::arrow::array::RecordBatch;
+use delta_kernel_engine::default::executor::tokio::TokioBackgroundExecutor;
+use delta_kernel_engine::default::DefaultEngine;
+use delta_kernel_engine::object_store::local::LocalFileSystem;
 
-use delta_kernel::engine::arrow_data::ArrowEngineData;
 use delta_kernel::{DeltaResult, Table};
 
 mod common;
@@ -15,7 +16,11 @@ fn read_v2_checkpoint_table(test_name: impl AsRef<str>) -> DeltaResult<Vec<Recor
     let test_path = test_dir.path().join(test_name.as_ref());
 
     let table = Table::try_from_uri(test_path.to_str().expect("table path to string")).unwrap();
-    let engine = Arc::new(SyncEngine::new());
+    let object_store = Arc::new(LocalFileSystem::new());
+    let engine = Arc::new(DefaultEngine::new(
+        object_store,
+        TokioBackgroundExecutor::new().into(),
+    ));
     let snapshot = table.snapshot(engine.as_ref(), None)?;
     let scan = snapshot.into_scan_builder().build()?;
     let batches = read_scan(&scan, engine)?;