feat!: Make expression and predicate evaluator constructors fallible

This PR refactors the `new_expression_evaluator` and `new_predicate_evaluator`
methods in the `EvaluationHandler` trait to return `DeltaResult` instead of
being infallible.

This is a breaking API change that enables:
- Better error reporting at evaluator construction time rather than evaluation
- Early validation of expression/predicate compatibility with input schemas
- More idiomatic Rust error handling patterns

**Changes:**
- Updated `EvaluationHandler` trait signatures to return
	`DeltaResult<Arc<dyn Evaluator>>`
- Updated `ArrowEvaluationHandler` implementation to wrap returns in `Ok(...)`
- Made `ScanLogReplayProcessor::new` fallible
- Updated `scan_action_iter` to return `DeltaResult<impl Iterator<...>>`
- Updated 12 callsites across the codebase to propagate errors with `?` operator
- Updated FFI layer to handle errors (`.unwrap()` for backward compatibility)

- All 630 existing kernel tests pass
- All 18 FFI tests pass
- Verified compilation with `cargo check --all-features`
- No functional changes - purely refactoring error handling patterns

Author: sanujbasu

ffi/src/engine_funcs.rs

@@ -171,10 +171,10 @@ fn new_expression_evaluator_impl(
     output_type: DataType,
 ) -> Handle<SharedExpressionEvaluator> {
     let engine = extern_engine.engine();
-    let evaluator =
-        engine
-            .evaluation_handler()
-            .new_expression_evaluator(input_schema, expression, output_type);
+    let evaluator = engine
+        .evaluation_handler()
+        .new_expression_evaluator(input_schema, expression, output_type)
+        .unwrap();
     evaluator.into()
 }
 

kernel/src/actions/visitors.rs

@@ -1091,6 +1091,7 @@ mod tests {
                 expression.into(),
                 InCommitTimestampVisitor::schema().into(),
             )
+            .unwrap()
             .evaluate(batch.as_ref())
             .unwrap()
     }

kernel/src/engine/arrow_expression/mod.rs

@@ -238,23 +238,23 @@ impl EvaluationHandler for ArrowEvaluationHandler {
         schema: SchemaRef,
         expression: ExpressionRef,
         output_type: DataType,
-    ) -> Arc<dyn ExpressionEvaluator> {
-        Arc::new(DefaultExpressionEvaluator {
+    ) -> DeltaResult<Arc<dyn ExpressionEvaluator>> {
+        Ok(Arc::new(DefaultExpressionEvaluator {
             input_schema: schema,
             expression,
             output_type,
-        })
+        }))
     }
 
     fn new_predicate_evaluator(
         &self,
         schema: SchemaRef,
         predicate: PredicateRef,
-    ) -> Arc<dyn PredicateEvaluator> {
-        Arc::new(DefaultPredicateEvaluator {
+    ) -> DeltaResult<Arc<dyn PredicateEvaluator>> {
+        Ok(Arc::new(DefaultPredicateEvaluator {
             input_schema: schema,
             predicate,
-        })
+        }))
     }
 
     /// Create a single-row array with all-null leaf values. Note that if a nested struct is

kernel/src/engine/default/mod.rs

@@ -106,7 +106,7 @@ impl<E: TaskExecutor> DefaultEngine<E> {
             input_schema.into(),
             transform.clone(),
             output_schema.clone().into(),
-        );
+        )?;
         let physical_data = logical_to_physical_expr.evaluate(data)?;
         self.parquet
             .write_parquet_file(write_context.target_dir(), physical_data, partition_values)

kernel/src/lib.rs

@@ -426,7 +426,7 @@ pub trait EvaluationHandler: AsAny {
         input_schema: SchemaRef,
         expression: ExpressionRef,
         output_type: DataType,
-    ) -> Arc<dyn ExpressionEvaluator>;
+    ) -> DeltaResult<Arc<dyn ExpressionEvaluator>>;
 
     /// Create a [`PredicateEvaluator`] that can evaluate the given [`Predicate`] on columnar
     /// batches with the given [`Schema`] to produce a column of boolean results.
@@ -443,7 +443,7 @@ pub trait EvaluationHandler: AsAny {
         &self,
         input_schema: SchemaRef,
         predicate: PredicateRef,
-    ) -> Arc<dyn PredicateEvaluator>;
+    ) -> DeltaResult<Arc<dyn PredicateEvaluator>>;
 
     /// Create a single-row all-null-value [`EngineData`] with the schema specified by
     /// `output_schema`.
@@ -474,7 +474,8 @@ trait EvaluationHandlerExtension: EvaluationHandler {
         schema_transform.transform_struct(schema.as_ref());
         let row_expr = schema_transform.try_into_expr()?;
 
-        let eval = self.new_expression_evaluator(null_row_schema, row_expr.into(), schema.into());
+        let eval =
+            self.new_expression_evaluator(null_row_schema, row_expr.into(), schema.into())?;
         eval.evaluate(null_row.as_ref())
     }
 }

kernel/src/scan/data_skipping.rs

@@ -134,21 +134,28 @@ impl DataSkippingFilter {
         //
         // 3. The selection evaluator does DISTINCT(col(predicate), 'false') to produce true (= keep) when
         //    the predicate is true/null and false (= skip) when the predicate is false.
-        let select_stats_evaluator = engine.evaluation_handler().new_expression_evaluator(
-            // safety: kernel is very broken if we don't have the schema for Add actions
-            get_log_add_schema().clone(),
-            STATS_EXPR.clone(),
-            DataType::STRING,
-        );
-
-        let skipping_evaluator = engine.evaluation_handler().new_predicate_evaluator(
-            stats_schema.clone(),
-            Arc::new(as_sql_data_skipping_predicate(&predicate)?),
-        );
+        let select_stats_evaluator = engine
+            .evaluation_handler()
+            .new_expression_evaluator(
+                // safety: kernel is very broken if we don't have the schema for Add actions
+                get_log_add_schema().clone(),
+                STATS_EXPR.clone(),
+                DataType::STRING,
+            )
+            .ok()?;
+
+        let skipping_evaluator = engine
+            .evaluation_handler()
+            .new_predicate_evaluator(
+                stats_schema.clone(),
+                Arc::new(as_sql_data_skipping_predicate(&predicate)?),
+            )
+            .ok()?;
 
         let filter_evaluator = engine
             .evaluation_handler()
-            .new_predicate_evaluator(stats_schema.clone(), FILTER_PRED.clone());
+            .new_predicate_evaluator(stats_schema.clone(), FILTER_PRED.clone())
+            .ok()?;
 
         Some(Self {
             stats_schema,

kernel/src/scan/log_replay.rs

@@ -53,7 +53,7 @@ pub(crate) struct ScanLogReplayProcessor {
 
 impl ScanLogReplayProcessor {
     /// Create a new [`ScanLogReplayProcessor`] instance
-    fn new(engine: &dyn Engine, state_info: Arc<StateInfo>) -> Self {
+    fn new(engine: &dyn Engine, state_info: Arc<StateInfo>) -> DeltaResult<Self> {
         // Extract the physical predicate from StateInfo's PhysicalPredicate enum.
         // The DataSkippingFilter and partition_filter components expect the predicate
         // in the format Option<(PredicateRef, SchemaRef)>, so we need to convert from
@@ -72,17 +72,17 @@ impl ScanLogReplayProcessor {
                 None
             }
         };
-        Self {
+        Ok(Self {
             partition_filter: physical_predicate.as_ref().map(|(e, _)| e.clone()),
             data_skipping_filter: DataSkippingFilter::new(engine, physical_predicate),
             add_transform: engine.evaluation_handler().new_expression_evaluator(
                 get_log_add_schema().clone(),
                 get_add_transform_expr(),
                 SCAN_ROW_DATATYPE.clone(),
-            ),
+            )?,
             seen_file_keys: Default::default(),
             state_info,
-        }
+        })
     }
 }
 
@@ -385,8 +385,8 @@ pub(crate) fn scan_action_iter(
     engine: &dyn Engine,
     action_iter: impl Iterator<Item = DeltaResult<ActionsBatch>>,
     state_info: Arc<StateInfo>,
-) -> impl Iterator<Item = DeltaResult<ScanMetadata>> {
-    ScanLogReplayProcessor::new(engine, state_info).process_actions_iter(action_iter)
+) -> DeltaResult<impl Iterator<Item = DeltaResult<ScanMetadata>>> {
+    Ok(ScanLogReplayProcessor::new(engine, state_info)?.process_actions_iter(action_iter))
 }
 
 #[cfg(test)]
@@ -478,7 +478,8 @@ mod tests {
                 .into_iter()
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             state_info,
-        );
+        )
+        .unwrap();
         for res in iter {
             let scan_metadata = res.unwrap();
             assert!(
@@ -503,7 +504,8 @@ mod tests {
                 .into_iter()
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             Arc::new(state_info),
-        );
+        )
+        .unwrap();
 
         fn validate_transform(transform: Option<&ExpressionRef>, expected_date_offset: i32) {
             assert!(transform.is_some());
@@ -580,7 +582,8 @@ mod tests {
                 .into_iter()
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             Arc::new(state_info),
-        );
+        )
+        .unwrap();
 
         for res in iter {
             let scan_metadata = res.unwrap();

kernel/src/scan/mod.rs

@@ -478,7 +478,7 @@ impl Scan {
             scan_row_schema(),
             get_scan_metadata_transform_expr(),
             RESTORED_ADD_SCHEMA.clone(),
-        );
+        )?;
         let apply_transform = move |data: Box<dyn EngineData>| {
             Ok(ActionsBatch::new(transform.evaluate(data.as_ref())?, false))
         };
@@ -537,7 +537,7 @@ impl Scan {
         if let PhysicalPredicate::StaticSkipAll = self.state_info.physical_predicate {
             return Ok(None.into_iter().flatten());
         }
-        let it = scan_action_iter(engine, action_batch_iter, self.state_info.clone());
+        let it = scan_action_iter(engine, action_batch_iter, self.state_info.clone())?;
         Ok(Some(it).into_iter().flatten())
     }
 
@@ -850,7 +850,8 @@ pub(crate) mod test_utils {
                 .into_iter()
                 .map(|batch| Ok(ActionsBatch::new(batch as _, true))),
             state_info,
-        );
+        )
+        .unwrap();
         let mut batch_count = 0;
         for res in iter {
             let scan_metadata = res.unwrap();

kernel/src/scan/state.rs

@@ -107,7 +107,7 @@ pub fn transform_to_logical(
                 physical_schema.clone(),
                 transform,
                 logical_schema.clone().into(), // TODO: expensive deep clone!
-            )
+            )?
             .evaluate(physical_data.as_ref()),
         None => Ok(physical_data),
     }

kernel/src/table_changes/log_replay.rs

@@ -243,7 +243,7 @@ impl LogReplayScanner {
             get_log_add_schema().clone(),
             Arc::new(cdf_scan_row_expression(timestamp, commit_version)),
             cdf_scan_row_schema().into(),
-        );
+        )?;
 
         let result = action_iter.map(move |actions| -> DeltaResult<_> {
             let actions = actions?;