refactor!: add TryIntoKernel/Arrow traits (#946)

## What changes are proposed in this pull request?
As discovered in crate split exploration we implement
`TryFrom`/`TryInto` for converting between kernel types and arrow types.
This is problematic for two reasons: (1) in the future whenever we do
the split, the current implementation breaks the orphan rule and (2)
it's difficult to reason about the types you are converting to/from.

This PR solves both issues by introducing new traits: `TryFromKernel`
and `TryFromArrow` which can be implemented to give
`Self::try_from_kernel(...)` and `Self::try_from_arrow(...)`
respectively. Additionally (similar to std lib) we automatically derive
`TryIntoArrow` for types that implement `TryFromKernel` and
`TryIntoKernel` for types that implement `TryFromArrow`.

### This PR affects the following public APIs
Removes old From/Into implementations for kernel schema types, replaces
with `TryFromKernel`/`TryIntoKernel`/`TryFromArrow`/`TryIntoArrow`.
migration will usually be as simple as changing a `.try_into()` to a
`.try_into_kernel()` or `.try_into_arrow()`.

## How was this change tested?
Existing UT

Author: zachschuermann

integration-tests/src/main.rs

@@ -1,4 +1,5 @@
 use delta_kernel::arrow::datatypes::{DataType, Field, Schema};
+use delta_kernel::engine::arrow_conversion::TryFromArrow as _;
 
 fn create_arrow_schema() -> Schema {
     let field_a = Field::new("a", DataType::Int64, false);
@@ -17,7 +18,7 @@ fn main() {
     let arrow_schema = create_arrow_schema();
     let kernel_schema = create_kernel_schema();
     let converted: delta_kernel::schema::Schema =
-        delta_kernel::schema::Schema::try_from(&arrow_schema).expect("couldn't convert");
+        delta_kernel::schema::Schema::try_from_arrow(&arrow_schema).expect("couldn't convert");
     assert!(kernel_schema == converted);
     println!("Okay, made it");
 }

kernel/src/engine/arrow_conversion.rs

@@ -1,4 +1,4 @@
-//! Conversions from kernel types to arrow types
+//! Conversions from kernel schema types to arrow schema types.
 
 use std::sync::Arc;
 
@@ -19,19 +19,57 @@ pub(crate) const MAP_ROOT_DEFAULT: &str = "key_value";
 pub(crate) const MAP_KEY_DEFAULT: &str = "key";
 pub(crate) const MAP_VALUE_DEFAULT: &str = "value";
 
-impl TryFrom<&StructType> for ArrowSchema {
-    type Error = ArrowError;
+/// Convert a kernel type into an arrow type (automatically implemented for all types that
+/// implement [`TryFromKernel`])
+pub trait TryIntoArrow<ArrowType> {
+    fn try_into_arrow(self) -> Result<ArrowType, ArrowError>;
+}
 
-    fn try_from(s: &StructType) -> Result<Self, ArrowError> {
-        let fields: Vec<ArrowField> = s.fields().map(TryInto::try_into).try_collect()?;
-        Ok(ArrowSchema::new(fields))
+/// Convert an arrow type into a kernel type (a similar [`TryIntoKernel`] trait is automatically
+/// implemented for all types that implement [`TryFromArrow`])
+pub trait TryFromArrow<ArrowType>: Sized {
+    fn try_from_arrow(t: ArrowType) -> Result<Self, ArrowError>;
+}
+
+/// Convert an arrow type into a kernel type (automatically implemented for all types that
+/// implement [`TryFromArrow`])
+pub trait TryIntoKernel<KernelType> {
+    fn try_into_kernel(self) -> Result<KernelType, ArrowError>;
+}
+
+/// Convert a kernel type into an arrow type (a similar [`TryIntoArrow`] trait is automatically
+/// implemented for all types that implement [`TryFromKernel`])
+pub trait TryFromKernel<KernelType>: Sized {
+    fn try_from_kernel(t: KernelType) -> Result<Self, ArrowError>;
+}
+
+impl<KernelType, ArrowType> TryIntoArrow<ArrowType> for KernelType
+where
+    ArrowType: TryFromKernel<KernelType>,
+{
+    fn try_into_arrow(self) -> Result<ArrowType, ArrowError> {
+        ArrowType::try_from_kernel(self)
     }
 }
 
-impl TryFrom<&StructField> for ArrowField {
-    type Error = ArrowError;
+impl<KernelType, ArrowType> TryIntoKernel<KernelType> for ArrowType
+where
+    KernelType: TryFromArrow<ArrowType>,
+{
+    fn try_into_kernel(self) -> Result<KernelType, ArrowError> {
+        KernelType::try_from_arrow(self)
+    }
+}
+
+impl TryFromKernel<&StructType> for ArrowSchema {
+    fn try_from_kernel(s: &StructType) -> Result<Self, ArrowError> {
+        let fields: Vec<ArrowField> = s.fields().map(|f| f.try_into_arrow()).try_collect()?;
+        Ok(ArrowSchema::new(fields))
+    }
+}
 
-    fn try_from(f: &StructField) -> Result<Self, ArrowError> {
+impl TryFromKernel<&StructField> for ArrowField {
+    fn try_from_kernel(f: &StructField) -> Result<Self, ArrowError> {
         let metadata = f
             .metadata()
             .iter()
@@ -42,45 +80,33 @@ impl TryFrom<&StructField> for ArrowField {
             .collect::<Result<_, serde_json::Error>>()
             .map_err(|err| ArrowError::JsonError(err.to_string()))?;
 
-        let field = ArrowField::new(
-            f.name(),
-            ArrowDataType::try_from(f.data_type())?,
-            f.is_nullable(),
-        )
-        .with_metadata(metadata);
+        let field = ArrowField::new(f.name(), f.data_type().try_into_arrow()?, f.is_nullable())
+            .with_metadata(metadata);
 
         Ok(field)
     }
 }
 
-impl TryFrom<&ArrayType> for ArrowField {
-    type Error = ArrowError;
-
-    fn try_from(a: &ArrayType) -> Result<Self, ArrowError> {
+impl TryFromKernel<&ArrayType> for ArrowField {
+    fn try_from_kernel(a: &ArrayType) -> Result<Self, ArrowError> {
         Ok(ArrowField::new(
             LIST_ARRAY_ROOT,
-            ArrowDataType::try_from(a.element_type())?,
+            a.element_type().try_into_arrow()?,
             a.contains_null(),
         ))
     }
 }
 
-impl TryFrom<&MapType> for ArrowField {
-    type Error = ArrowError;
-
-    fn try_from(a: &MapType) -> Result<Self, ArrowError> {
+impl TryFromKernel<&MapType> for ArrowField {
+    fn try_from_kernel(a: &MapType) -> Result<Self, ArrowError> {
         Ok(ArrowField::new(
             MAP_ROOT_DEFAULT,
             ArrowDataType::Struct(
                 vec![
-                    ArrowField::new(
-                        MAP_KEY_DEFAULT,
-                        ArrowDataType::try_from(a.key_type())?,
-                        false,
-                    ),
+                    ArrowField::new(MAP_KEY_DEFAULT, a.key_type().try_into_arrow()?, false),
                     ArrowField::new(
                         MAP_VALUE_DEFAULT,
-                        ArrowDataType::try_from(a.value_type())?,
+                        a.value_type().try_into_arrow()?,
                         a.value_contains_null(),
                     ),
                 ]
@@ -91,10 +117,8 @@ impl TryFrom<&MapType> for ArrowField {
     }
 }
 
-impl TryFrom<&DataType> for ArrowDataType {
-    type Error = ArrowError;
-
-    fn try_from(t: &DataType) -> Result<Self, ArrowError> {
+impl TryFromKernel<&DataType> for ArrowDataType {
+    fn try_from_kernel(t: &DataType) -> Result<Self, ArrowError> {
         match t {
             DataType::Primitive(p) => {
                 match p {
@@ -128,54 +152,49 @@ impl TryFrom<&DataType> for ArrowDataType {
             }
             DataType::Struct(s) => Ok(ArrowDataType::Struct(
                 s.fields()
-                    .map(TryInto::try_into)
+                    .map(TryIntoArrow::try_into_arrow)
                     .collect::<Result<Vec<ArrowField>, ArrowError>>()?
                     .into(),
             )),
-            DataType::Array(a) => Ok(ArrowDataType::List(Arc::new(a.as_ref().try_into()?))),
-            DataType::Map(m) => Ok(ArrowDataType::Map(Arc::new(m.as_ref().try_into()?), false)),
+            DataType::Array(a) => Ok(ArrowDataType::List(Arc::new(a.as_ref().try_into_arrow()?))),
+            DataType::Map(m) => Ok(ArrowDataType::Map(
+                Arc::new(m.as_ref().try_into_arrow()?),
+                false,
+            )),
         }
     }
 }
 
-impl TryFrom<&ArrowSchema> for StructType {
-    type Error = ArrowError;
-
-    fn try_from(arrow_schema: &ArrowSchema) -> Result<Self, ArrowError> {
+impl TryFromArrow<&ArrowSchema> for StructType {
+    fn try_from_arrow(arrow_schema: &ArrowSchema) -> Result<Self, ArrowError> {
         StructType::try_new(
             arrow_schema
                 .fields()
                 .iter()
-                .map(|field| field.as_ref().try_into()),
+                .map(|field| field.as_ref().try_into_kernel()),
         )
     }
 }
 
-impl TryFrom<ArrowSchemaRef> for StructType {
-    type Error = ArrowError;
-
-    fn try_from(arrow_schema: ArrowSchemaRef) -> Result<Self, ArrowError> {
-        arrow_schema.as_ref().try_into()
+impl TryFromArrow<ArrowSchemaRef> for StructType {
+    fn try_from_arrow(arrow_schema: ArrowSchemaRef) -> Result<Self, ArrowError> {
+        arrow_schema.as_ref().try_into_kernel()
     }
 }
 
-impl TryFrom<&ArrowField> for StructField {
-    type Error = ArrowError;
-
-    fn try_from(arrow_field: &ArrowField) -> Result<Self, ArrowError> {
+impl TryFromArrow<&ArrowField> for StructField {
+    fn try_from_arrow(arrow_field: &ArrowField) -> Result<Self, ArrowError> {
         Ok(StructField::new(
             arrow_field.name().clone(),
-            DataType::try_from(arrow_field.data_type())?,
+            DataType::try_from_arrow(arrow_field.data_type())?,
             arrow_field.is_nullable(),
         )
         .with_metadata(arrow_field.metadata().iter().map(|(k, v)| (k.clone(), v))))
     }
 }
 
-impl TryFrom<&ArrowDataType> for DataType {
-    type Error = ArrowError;
-
-    fn try_from(arrow_datatype: &ArrowDataType) -> Result<Self, ArrowError> {
+impl TryFromArrow<&ArrowDataType> for DataType {
+    fn try_from_arrow(arrow_datatype: &ArrowDataType) -> Result<Self, ArrowError> {
         match arrow_datatype {
             ArrowDataType::Utf8 => Ok(DataType::STRING),
             ArrowDataType::LargeUtf8 => Ok(DataType::STRING),
@@ -212,28 +231,38 @@ impl TryFrom<&ArrowDataType> for DataType {
             {
                 Ok(DataType::TIMESTAMP)
             }
-            ArrowDataType::Struct(fields) => {
-                DataType::try_struct_type(fields.iter().map(|field| field.as_ref().try_into()))
-            }
-            ArrowDataType::List(field) => {
-                Ok(ArrayType::new((*field).data_type().try_into()?, (*field).is_nullable()).into())
-            }
-            ArrowDataType::ListView(field) => {
-                Ok(ArrayType::new((*field).data_type().try_into()?, (*field).is_nullable()).into())
-            }
-            ArrowDataType::LargeList(field) => {
-                Ok(ArrayType::new((*field).data_type().try_into()?, (*field).is_nullable()).into())
-            }
-            ArrowDataType::LargeListView(field) => {
-                Ok(ArrayType::new((*field).data_type().try_into()?, (*field).is_nullable()).into())
-            }
-            ArrowDataType::FixedSizeList(field, _) => {
-                Ok(ArrayType::new((*field).data_type().try_into()?, (*field).is_nullable()).into())
-            }
+            ArrowDataType::Struct(fields) => DataType::try_struct_type(
+                fields.iter().map(|field| field.as_ref().try_into_kernel()),
+            ),
+            ArrowDataType::List(field) => Ok(ArrayType::new(
+                (*field).data_type().try_into_kernel()?,
+                (*field).is_nullable(),
+            )
+            .into()),
+            ArrowDataType::ListView(field) => Ok(ArrayType::new(
+                (*field).data_type().try_into_kernel()?,
+                (*field).is_nullable(),
+            )
+            .into()),
+            ArrowDataType::LargeList(field) => Ok(ArrayType::new(
+                (*field).data_type().try_into_kernel()?,
+                (*field).is_nullable(),
+            )
+            .into()),
+            ArrowDataType::LargeListView(field) => Ok(ArrayType::new(
+                (*field).data_type().try_into_kernel()?,
+                (*field).is_nullable(),
+            )
+            .into()),
+            ArrowDataType::FixedSizeList(field, _) => Ok(ArrayType::new(
+                (*field).data_type().try_into_kernel()?,
+                (*field).is_nullable(),
+            )
+            .into()),
             ArrowDataType::Map(field, _) => {
                 if let ArrowDataType::Struct(struct_fields) = field.data_type() {
-                    let key_type = DataType::try_from(struct_fields[0].data_type())?;
-                    let value_type = DataType::try_from(struct_fields[1].data_type())?;
+                    let key_type = DataType::try_from_arrow(struct_fields[0].data_type())?;
+                    let value_type = DataType::try_from_arrow(struct_fields[1].data_type())?;
                     let value_type_nullable = struct_fields[1].is_nullable();
                     Ok(MapType::new(key_type, value_type, value_type_nullable).into())
                 } else {
@@ -242,7 +271,9 @@ impl TryFrom<&ArrowDataType> for DataType {
             }
             // Dictionary types are just an optimized in-memory representation of an array.
             // Schema-wise, they are the same as the value type.
-            ArrowDataType::Dictionary(_, value_type) => Ok(value_type.as_ref().try_into()?),
+            ArrowDataType::Dictionary(_, value_type) => {
+                Ok(value_type.as_ref().try_into_kernel()?)
+            }
             s => Err(ArrowError::SchemaError(format!(
                 "Invalid data type for Delta Lake: {s}"
             ))),
@@ -252,6 +283,7 @@ impl TryFrom<&ArrowDataType> for DataType {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use crate::engine::arrow_conversion::ArrowField;
     use crate::{
         schema::{DataType, StructField},
@@ -265,7 +297,7 @@ mod tests {
         metadata.insert("description", "hello world".to_owned());
         let struct_field = StructField::not_null("name", DataType::STRING).with_metadata(metadata);
 
-        let arrow_field = ArrowField::try_from(&struct_field)?;
+        let arrow_field = ArrowField::try_from_kernel(&struct_field)?;
         let new_metadata = arrow_field.metadata();
 
         assert_eq!(

kernel/src/engine/arrow_expression/mod.rs

@@ -6,6 +6,7 @@ use crate::arrow::datatypes::{
     DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema,
 };
 
+use super::arrow_conversion::{TryFromKernel as _, TryIntoArrow as _};
 use crate::engine::arrow_data::ArrowEngineData;
 use crate::error::{DeltaResult, Error};
 use crate::expressions::{Expression, Predicate, Scalar};
@@ -64,7 +65,7 @@ impl ArrayBuilderAs for StructFieldBuilder<'_> {
 impl Scalar {
     /// Convert scalar to arrow array.
     pub fn to_array(&self, num_rows: usize) -> DeltaResult<ArrayRef> {
-        let data_type = ArrowDataType::try_from(&self.data_type())?;
+        let data_type = ArrowDataType::try_from_kernel(&self.data_type())?;
         let mut builder = array::make_builder(&data_type, num_rows);
         self.append_to(&mut builder, num_rows)?;
         Ok(builder.finish())
@@ -301,7 +302,7 @@ impl EvaluationHandler for ArrowEvaluationHandler {
             .map(|field| Scalar::Null(field.data_type().clone()).to_array(1))
             .try_collect()?;
         let record_batch =
-            RecordBatch::try_new(Arc::new(output_schema.as_ref().try_into()?), arrays)?;
+            RecordBatch::try_new(Arc::new(output_schema.as_ref().try_into_arrow()?), arrays)?;
         Ok(Box::new(ArrowEngineData::new(record_batch)))
     }
 }
@@ -321,7 +322,7 @@ impl ExpressionEvaluator for DefaultExpressionEvaluator {
             .downcast_ref::<ArrowEngineData>()
             .ok_or_else(|| Error::engine_data_type("ArrowEngineData"))?
             .record_batch();
-        let _input_schema: ArrowSchema = self.input_schema.as_ref().try_into()?;
+        let _input_schema: ArrowSchema = self.input_schema.as_ref().try_into_arrow()?;
         // TODO: make sure we have matching schemas for validation
         // if batch.schema().as_ref() != &input_schema {
         //     return Err(Error::Generic(format!(
@@ -335,7 +336,7 @@ impl ExpressionEvaluator for DefaultExpressionEvaluator {
             apply_schema(&array_ref, &self.output_type)?
         } else {
             let array_ref = apply_schema_to(&array_ref, &self.output_type)?;
-            let arrow_type: ArrowDataType = ArrowDataType::try_from(&self.output_type)?;
+            let arrow_type = ArrowDataType::try_from_kernel(&self.output_type)?;
             let schema = ArrowSchema::new(vec![ArrowField::new("output", arrow_type, true)]);
             RecordBatch::try_new(Arc::new(schema), vec![array_ref])?
         };
@@ -357,7 +358,7 @@ impl PredicateEvaluator for DefaultPredicateEvaluator {
             .downcast_ref::<ArrowEngineData>()
             .ok_or_else(|| Error::engine_data_type("ArrowEngineData"))?
             .record_batch();
-        let _input_schema: ArrowSchema = self.input_schema.as_ref().try_into()?;
+        let _input_schema: ArrowSchema = self.input_schema.as_ref().try_into_arrow()?;
         // TODO: make sure we have matching schemas for validation
         // if batch.schema().as_ref() != &input_schema {
         //     return Err(Error::Generic(format!(

kernel/src/engine/arrow_expression/tests.rs

@@ -523,7 +523,7 @@ fn test_null_row() {
     let handler = ArrowEvaluationHandler;
     let result = handler.null_row(schema.clone()).unwrap();
     let expected = RecordBatch::try_new(
-        Arc::new(schema.as_ref().try_into().unwrap()),
+        Arc::new(schema.as_ref().try_into_arrow().unwrap()),
         vec![
             Arc::new(StructArray::new_null(
                 [