feat: make ExpressionHandler::get_evaluator fallible

kernel/src/engine/arrow_expression.rs

@@ -519,12 +519,12 @@ impl ExpressionHandler for ArrowExpressionHandler {
         schema: SchemaRef,
         expression: Expression,
         output_type: DataType,
-    ) -> Arc<dyn ExpressionEvaluator> {
-        Arc::new(DefaultExpressionEvaluator {
+    ) -> DeltaResult<Arc<dyn ExpressionEvaluator>> {
+        Ok(Arc::new(DefaultExpressionEvaluator {
             input_schema: schema,
             expression: Box::new(expression),
             output_type,
-        })
+        }))
     }
 }
 

kernel/src/engine/default/mod.rs

@@ -128,7 +128,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(

kernel/src/lib.rs

@@ -332,7 +332,7 @@ pub trait ExpressionHandler: AsAny {
         schema: SchemaRef,
         expression: Expression,
         output_type: DataType,
-    ) -> Arc<dyn ExpressionEvaluator>;
+    ) -> DeltaResult<Arc<dyn ExpressionEvaluator>>;
 }
 
 /// Provides file system related functionalities to Delta Kernel.

kernel/src/scan/data_skipping.rs

@@ -2,7 +2,7 @@ use std::borrow::Cow;
 use std::cmp::Ordering;
 use std::sync::{Arc, LazyLock};
 
-use tracing::debug;
+use tracing::{debug, warn};
 
 use crate::actions::get_log_add_schema;
 use crate::actions::visitors::SelectionVectorVisitor;
@@ -99,24 +99,32 @@ 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.get_expression_handler().get_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 select_stats_evaluator = engine
+            .get_expression_handler()
+            .get_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,
+            )
+            .inspect_err(|e| warn!("Failed to create stats selector evaluator: {}", e))
+            .ok()?;
 
-        let skipping_evaluator = engine.get_expression_handler().get_evaluator(
-            stats_schema.clone(),
-            Expr::struct_from([as_data_skipping_predicate(&predicate, false)?]),
-            PREDICATE_SCHEMA.clone(),
-        );
+        let skipping_evaluator = engine
+            .get_expression_handler()
+            .get_evaluator(
+                stats_schema.clone(),
+                Expr::struct_from([as_data_skipping_predicate(&predicate, false)?]),
+                PREDICATE_SCHEMA.clone(),
+            )
+            .inspect_err(|e| warn!("Failed to create skipping evaluator: {}", e))
+            .ok()?;
 
-        let filter_evaluator = engine.get_expression_handler().get_evaluator(
-            stats_schema.clone(),
-            FILTER_EXPR.clone(),
-            DataType::BOOLEAN,
-        );
+        let filter_evaluator = engine
+            .get_expression_handler()
+            .get_evaluator(stats_schema.clone(), FILTER_EXPR.clone(), DataType::BOOLEAN)
+            .inspect_err(|e| warn!("Failed to create filter evaluator: {}", e))
+            .ok()?;
 
         Some(Self {
             stats_schema,

kernel/src/scan/log_replay.rs

@@ -237,21 +237,24 @@ impl LogReplayScanner {
 /// indicates whether the record batch is a log or checkpoint batch.
 pub fn scan_action_iter(
     engine: &dyn Engine,
-    action_iter: impl Iterator<Item = DeltaResult<(Box<dyn EngineData>, bool)>>,
+    action_iter: impl Iterator<Item = DeltaResult<(Box<dyn EngineData>, bool)>> + 'static,
     physical_predicate: Option<(ExpressionRef, SchemaRef)>,
-) -> impl Iterator<Item = DeltaResult<ScanData>> {
+) -> DeltaResult<impl Iterator<Item = DeltaResult<ScanData>>> {
     let mut log_scanner = LogReplayScanner::new(engine, physical_predicate);
     let add_transform = engine.get_expression_handler().get_evaluator(
         get_log_add_schema().clone(),
         get_add_transform_expr(),
         SCAN_ROW_DATATYPE.clone(),
-    );
-    action_iter
+    )?;
+
+    let actions = action_iter
         .map(move |action_res| {
             let (batch, is_log_batch) = action_res?;
             log_scanner.process_scan_batch(add_transform.as_ref(), batch.as_ref(), is_log_batch)
         })
-        .filter(|res| res.as_ref().map_or(true, |(_, sv)| sv.contains(&true)))
+        .filter(|res| res.as_ref().map_or(true, |(_, sv)| sv.contains(&true)));
+
+    Ok(actions)
 }
 
 #[cfg(test)]
@@ -291,7 +294,8 @@ mod tests {
             &[true, false],
             (),
             validate_simple,
-        );
+        )
+        .unwrap();
     }
 
     #[test]
@@ -301,6 +305,7 @@ mod tests {
             &[false, false, true, false],
             (),
             validate_simple,
-        );
+        )
+        .unwrap();
     }
 }

kernel/src/scan/mod.rs

@@ -381,11 +381,13 @@ impl Scan {
             PhysicalPredicate::Some(predicate, schema) => Some((predicate, schema)),
             PhysicalPredicate::None => None,
         };
+
         let it = scan_action_iter(
             engine,
             self.replay_for_scan_data(engine)?,
             physical_predicate,
-        );
+        )?;
+
         Ok(Some(it).into_iter().flatten())
     }
 
@@ -687,7 +689,7 @@ fn transform_to_logical_internal(
             physical_schema,
             read_expression,
             global_state.logical_schema.clone().into(),
-        )
+        )?
         .evaluate(data.as_ref())?;
     Ok(result)
 }
@@ -707,7 +709,7 @@ pub(crate) mod test_utils {
             sync::{json::SyncJsonHandler, SyncEngine},
         },
         scan::log_replay::scan_action_iter,
-        EngineData, JsonHandler,
+        DeltaResult, EngineData, JsonHandler,
     };
 
     use super::state::ScanCallback;
@@ -759,26 +761,26 @@ pub(crate) mod test_utils {
         expected_sel_vec: &[bool],
         context: T,
         validate_callback: ScanCallback<T>,
-    ) {
+    ) -> DeltaResult<()> {
         let iter = scan_action_iter(
             &SyncEngine::new(),
             batch.into_iter().map(|batch| Ok((batch as _, true))),
             None,
-        );
+        )?;
         let mut batch_count = 0;
         for res in iter {
-            let (batch, sel) = res.unwrap();
-            assert_eq!(sel, expected_sel_vec);
+            let (batch, sel) = res?;
+            assert_eq!(sel.as_slice(), expected_sel_vec);
             crate::scan::state::visit_scan_files(
                 batch.as_ref(),
                 &sel,
                 context.clone(),
                 validate_callback,
-            )
-            .unwrap();
+            )?;
             batch_count += 1;
         }
         assert_eq!(batch_count, 1);
+        Ok(())
     }
 }
 

kernel/src/scan/state.rs

@@ -253,6 +253,7 @@ mod tests {
             &[true, false],
             context,
             validate_visit,
-        );
+        )
+        .unwrap();
     }
 }

kernel/src/table_changes/log_replay.rs

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

kernel/src/table_changes/scan.rs

@@ -289,7 +289,8 @@ fn read_scan_file(
         physical_schema.clone(),
         physical_to_logical_expr,
         global_state.logical_schema.clone().into(),
-    );
+    )?;
+
     // Determine if the scan file was derived from a deletion vector pair
     let is_dv_resolved_pair = scan_file.remove_dv.is_some();
 

kernel/src/transaction.rs

@@ -201,7 +201,7 @@ fn generate_adds<'a>(
             write_metadata_schema.clone(),
             adds_expr,
             log_schema.clone().into(),
-        );
+        )?;
         adds_evaluator.evaluate(write_metadata_batch)
     })
 }
@@ -321,7 +321,7 @@ fn generate_commit_info(
         engine_commit_info_schema.into(),
         commit_info_expr,
         commit_info_empty_struct_schema.into(),
-    );
+    )?;
 
     commit_info_evaluator.evaluate(engine_commit_info)
 }