chore: cleanup expression handler

Author: roeap

crates/deltalake-core/src/kernel/client/expressions.rs

@@ -6,18 +6,15 @@ use std::sync::Arc;
 
 use arrow_arith::boolean::{and, is_null, not, or};
 use arrow_arith::numeric::{add, div, mul, sub};
-use arrow_array::RecordBatch as ColumnarBatch;
 use arrow_array::{
     Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
     Int32Array, RecordBatch, StringArray, TimestampMicrosecondArray,
 };
 use arrow_ord::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
 
-use super::ExpressionEvaluator;
 use crate::kernel::error::{DeltaResult, Error};
 use crate::kernel::expressions::{scalars::Scalar, Expression};
 use crate::kernel::expressions::{BinaryOperator, UnaryOperator};
-use crate::kernel::schema::SchemaRef;
 
 // TODO leverage scalars / Datum
 
@@ -43,7 +40,10 @@ impl Scalar {
     }
 }
 
-fn evaluate_expression(expression: &Expression, batch: &RecordBatch) -> DeltaResult<ArrayRef> {
+pub(crate) fn evaluate_expression(
+    expression: &Expression,
+    batch: &RecordBatch,
+) -> DeltaResult<ArrayRef> {
     match expression {
         Expression::Literal(scalar) => Ok(scalar.to_array(batch.num_rows())),
         Expression::Column(name) => batch
@@ -166,20 +166,6 @@ fn evaluate_expression(expression: &Expression, batch: &RecordBatch) -> DeltaRes
     }
 }
 
-#[derive(Debug)]
-/// Expression evaluator based on arrow compute kernels.
-pub struct ArrowExpressionEvaluator {
-    _input_schema: SchemaRef,
-    expression: Box<Expression>,
-}
-
-impl ExpressionEvaluator for ArrowExpressionEvaluator {
-    fn evaluate(&self, batch: &ColumnarBatch) -> DeltaResult<ColumnarBatch> {
-        let _result = evaluate_expression(&self.expression, batch)?;
-        todo!()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/deltalake-core/src/kernel/client/mod.rs

@@ -1,8 +1,12 @@
 //! Delta kernel client implementation.
+use std::sync::Arc;
 
 use arrow_array::RecordBatch;
 
-use super::error::DeltaResult;
+use self::expressions::evaluate_expression;
+use crate::kernel::error::DeltaResult;
+use crate::kernel::expressions::Expression;
+use crate::kernel::schema::SchemaRef;
 
 pub mod expressions;
 
@@ -16,5 +20,21 @@ pub trait ExpressionEvaluator {
     ///
     /// Contains one value for each row of the input.
     /// The data type of the output is same as the type output of the expression this evaluator is using.
-    fn evaluate(&self, batch: &RecordBatch) -> DeltaResult<RecordBatch>;
+    fn evaluate(&self, batch: &RecordBatch, output_schema: SchemaRef) -> DeltaResult<RecordBatch>;
+}
+
+#[derive(Debug)]
+/// Expression evaluator based on arrow compute kernels.
+pub struct ArrowExpressionEvaluator {
+    _input_schema: SchemaRef,
+    expression: Box<Expression>,
+}
+
+impl ExpressionEvaluator for ArrowExpressionEvaluator {
+    fn evaluate(&self, batch: &RecordBatch, output_schema: SchemaRef) -> DeltaResult<RecordBatch> {
+        Ok(RecordBatch::try_new(
+            Arc::new(output_schema.as_ref().try_into()?),
+            vec![evaluate_expression(&self.expression, batch)?],
+        )?)
+    }
 }

crates/deltalake-core/src/kernel/mod.rs

@@ -1,6 +1,7 @@
 //! Kernel module
 
 pub mod actions;
+#[cfg(feature = "arrow")]
 pub mod client;
 pub mod error;
 pub mod expressions;