feat: new #[derive(IntoEngineData)] proc macro (delta-io#830)

## What changes are proposed in this pull request?
With the introduction of `EvaluationHandler::create_one`, we can create
`EngineData` now. This PR adds a handy new procedural macro to implement
a new `IntoEngineData` trait which allows for
`self.into_engine_data(schema, &dyn Engine)` as long as all fields
implement `Into<Scalar>` (for now just support the simple case of
all-primitive fields)

This would typically be used in conjunction with `ToSchema` to provide
the schema for the output `EngineData`.

### Future work
Need to implement tests (make doctest on `IntoEngineData` trait actually
build): lump that in to delta-io#514
1. need to rename `crate::` to `delta_kernel::` in macros
2. need to make all the types we need conditionally pub (that is, if
`cfg(doctest)` then make stuff `pub`)
3. need to do an `extern crate self as delta_kernel`
4. and finally, make the doctest a `no_run` (building) doctest

## How was this change tested?
new UT

Author: zachschuermann

derive-macros/src/lib.rs

@@ -138,6 +138,58 @@ fn gen_schema_fields(data: &Data) -> TokenStream {
     quote! { #(#schema_fields),* }
 }
 
+/// Derive an IntoEngineData trait for a struct that has all fields implement `Into<Scalar>`.
+///
+/// This is a relatively simple macro to produce the boilerplate for converting a struct into
+/// EngineData using the `create_one` method. TODO: (doc)tests included in the delta_kernel crate:
+/// `IntoEngineData` trait.
+#[proc_macro_derive(IntoEngineData)]
+pub fn into_engine_data_derive(input: proc_macro::TokenStream) -> proc_macro::TokenStream {
+    let input = parse_macro_input!(input as DeriveInput);
+    let struct_name = &input.ident;
+
+    let Data::Struct(DataStruct {
+        fields: Fields::Named(fields),
+        ..
+    }) = &input.data
+    else {
+        return Error::new(
+            struct_name.span(),
+            "IntoEngineData can only be derived for structs with named fields",
+        )
+        .to_compile_error()
+        .into();
+    };
+
+    let fields = &fields.named;
+    let field_idents = fields.iter().map(|f| &f.ident);
+    let field_types = fields.iter().map(|f| &f.ty);
+
+    let expanded = quote! {
+        #[automatically_derived]
+        impl crate::IntoEngineData for #struct_name
+        where
+            #(#field_types: Into<crate::expressions::Scalar>),*
+        {
+            fn into_engine_data(
+                self,
+                schema: crate::schema::SchemaRef,
+                engine: &dyn crate::Engine)
+            -> crate::DeltaResult<Box<dyn crate::EngineData>> {
+                // NB: we `use` here to avoid polluting the caller's namespace
+                use crate::EvaluationHandlerExtension as _;
+                let values = [
+                    #(self.#field_idents.into()),*
+                ];
+                let evaluator = engine.evaluation_handler();
+                evaluator.create_one(schema, &values)
+            }
+        }
+    };
+
+    proc_macro::TokenStream::from(expanded)
+}
+
 /// Mark items as `internal_api` to make them public iff the `internal-api` feature is enabled.
 /// Note this doesn't work for inline module definitions (see `internal_mod!` macro in delta_kernel
 /// crate - can't export macro_rules! from proc macro crate).

kernel/src/actions/mod.rs

@@ -14,13 +14,12 @@ use crate::table_features::{
 };
 use crate::table_properties::TableProperties;
 use crate::utils::require;
-use crate::EvaluationHandlerExtension;
-use crate::{DeltaResult, Engine, EngineData, Error, FileMeta, RowVisitor as _};
+use crate::{DeltaResult, EngineData, Error, FileMeta, RowVisitor as _};
 
 use url::Url;
 use visitors::{MetadataVisitor, ProtocolVisitor};
 
-use delta_kernel_derive::{internal_api, ToSchema};
+use delta_kernel_derive::{internal_api, IntoEngineData, ToSchema};
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 
@@ -597,7 +596,7 @@ pub(crate) struct Cdc {
     pub tags: Option<HashMap<String, String>>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, ToSchema)]
+#[derive(Debug, Clone, PartialEq, Eq, ToSchema, IntoEngineData)]
 #[internal_api]
 pub(crate) struct SetTransaction {
     /// A unique identifier for the application performing the transaction.
@@ -618,16 +617,6 @@ impl SetTransaction {
             last_updated,
         }
     }
-
-    pub(crate) fn into_engine_data(self, engine: &dyn Engine) -> DeltaResult<Box<dyn EngineData>> {
-        let values = [
-            self.app_id.into(),
-            self.version.into(),
-            self.last_updated.into(),
-        ];
-        let evaluator = engine.evaluation_handler();
-        evaluator.create_one(get_log_txn_schema().clone(), &values)
-    }
 }
 
 /// The sidecar action references a sidecar file which provides some of the checkpoint's
@@ -1125,4 +1114,79 @@ mod tests {
         ]);
         assert_eq!(parse_features::<ReaderFeature>(features), expected);
     }
+
+    #[test]
+    fn test_into_engine_data() {
+        use crate::arrow::array::{Int64Array, StringArray};
+        use crate::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
+        use crate::arrow::record_batch::RecordBatch;
+
+        use crate::engine::arrow_data::ArrowEngineData;
+        use crate::engine::arrow_expression::ArrowEvaluationHandler;
+        use crate::IntoEngineData;
+        use crate::{Engine, EvaluationHandler, JsonHandler, ParquetHandler, StorageHandler};
+
+        // duplicated
+        struct ExprEngine(Arc<dyn EvaluationHandler>);
+
+        impl ExprEngine {
+            fn new() -> Self {
+                ExprEngine(Arc::new(ArrowEvaluationHandler))
+            }
+        }
+
+        impl Engine for ExprEngine {
+            fn evaluation_handler(&self) -> Arc<dyn EvaluationHandler> {
+                self.0.clone()
+            }
+
+            fn json_handler(&self) -> Arc<dyn JsonHandler> {
+                unimplemented!()
+            }
+
+            fn parquet_handler(&self) -> Arc<dyn ParquetHandler> {
+                unimplemented!()
+            }
+
+            fn storage_handler(&self) -> Arc<dyn StorageHandler> {
+                unimplemented!()
+            }
+        }
+
+        let engine = ExprEngine::new();
+
+        let set_transaction = SetTransaction {
+            app_id: "app_id".to_string(),
+            version: 0,
+            last_updated: None,
+        };
+
+        let engine_data =
+            set_transaction.into_engine_data(SetTransaction::to_schema().into(), &engine);
+
+        let record_batch: crate::arrow::array::RecordBatch = engine_data
+            .unwrap()
+            .into_any()
+            .downcast::<ArrowEngineData>()
+            .unwrap()
+            .into();
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("appId", ArrowDataType::Utf8, false),
+            Field::new("version", ArrowDataType::Int64, false),
+            Field::new("lastUpdated", ArrowDataType::Int64, true),
+        ]));
+
+        let expected = RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(StringArray::from(vec!["app_id"])),
+                Arc::new(Int64Array::from(vec![0_i64])),
+                Arc::new(Int64Array::from(vec![None::<i64>])),
+            ],
+        )
+        .unwrap();
+
+        assert_eq!(record_batch, expected);
+    }
 }

kernel/src/lib.rs

@@ -432,6 +432,39 @@ trait EvaluationHandlerExtension: EvaluationHandler {
 // Auto-implement the extension trait for all EvaluationHandlers
 impl<T: EvaluationHandler + ?Sized> EvaluationHandlerExtension for T {}
 
+/// A trait that allows converting a type into (single-row) EngineData
+///
+/// This is typically used with the `#[derive(IntoEngineData)]` macro
+/// which leverages the traits `ToDataType` and `Into<Scalar>` for struct fields
+/// to convert a struct into EngineData.
+///
+/// # Example
+/// ```ignore
+/// # use std::sync::Arc;
+/// # use delta_kernel_derive::{Schema, IntoEngineData};
+///
+/// #[derive(Schema, IntoEngineData)]
+/// struct MyStruct {
+///    a: i32,
+///    b: String,
+/// }
+///
+/// let my_struct = MyStruct { a: 42, b: "Hello".to_string() };
+/// // typically used with ToSchema
+/// let schema = Arc::new(MyStruct::to_schema());
+/// // single-row EngineData
+/// let engine = todo!(); // create an engine
+/// let engine_data = my_struct.into_engine_data(schema, engine);
+/// ```
+pub(crate) trait IntoEngineData {
+    /// Consume this type to produce a single-row EngineData using the provided schema.
+    fn into_engine_data(
+        self,
+        schema: SchemaRef,
+        engine: &dyn Engine,
+    ) -> DeltaResult<Box<dyn EngineData>>;
+}
+
 /// Provides file system related functionalities to Delta Kernel.
 ///
 /// Delta Kernel uses this handler whenever it needs to access the underlying

kernel/src/transaction.rs

@@ -5,13 +5,13 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use crate::actions::SetTransaction;
 use crate::actions::COMMIT_INFO_NAME;
-use crate::actions::{get_log_add_schema, get_log_commit_info_schema};
+use crate::actions::{get_log_add_schema, get_log_commit_info_schema, get_log_txn_schema};
 use crate::error::Error;
 use crate::expressions::{column_expr, Scalar, StructData};
 use crate::path::ParsedLogPath;
 use crate::schema::{MapType, SchemaRef, StructField, StructType};
 use crate::snapshot::Snapshot;
-use crate::{DataType, DeltaResult, Engine, EngineData, Expression, Version};
+use crate::{DataType, DeltaResult, Engine, EngineData, Expression, IntoEngineData, Version};
 
 use url::Url;
 
@@ -132,7 +132,7 @@ impl Transaction {
             .set_transactions
             .clone()
             .into_iter()
-            .map(|txn| txn.into_engine_data(engine));
+            .map(|txn| txn.into_engine_data(get_log_txn_schema().clone(), engine));
 
         // step one: construct the iterator of commit info + file actions we want to commit
         let engine_commit_info = self

kernel/tests/write.rs

@@ -741,7 +741,7 @@ async fn test_write_txn_actions() -> Result<(), Box<dyn std::error::Error>> {
         // commit!
         txn.commit(&engine)?;
 
-        let snapshot = Arc::new(table.snapshot(&engine, None)?);
+        let snapshot = Arc::new(table.snapshot(&engine, 1.into())?);
         assert_eq!(
             snapshot.clone().get_app_id_version("app_id1", &engine)?,
             Some(1)