feat: make ExpressionHandler::get_evaluator fallible

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;
@@ -107,20 +107,32 @@ impl DataSkippingFilter {
                 STATS_EXPR.clone(),
                 DataType::STRING,
             )
+            .map_err(|e| {
+                warn!("Failed to create stats selector evaluator: {}", e);
+                e
+            })
             .ok()?;
 
         let skipping_evaluator = engine
             .get_expression_handler()
             .get_evaluator(
                 stats_schema.clone(),
-                Expr::struct_from([as_data_skipping_predicate(predicate, false)?]),
+                Expr::struct_from([as_data_skipping_predicate(&predicate, false)?]),
                 PREDICATE_SCHEMA.clone(),
             )
+            .map_err(|e| {
+                warn!("Failed to create skipping evaluator: {}", e);
+                e
+            })
             .ok()?;
 
         let filter_evaluator = engine
             .get_expression_handler()
             .get_evaluator(stats_schema.clone(), FILTER_EXPR.clone(), DataType::BOOLEAN)
+            .map_err(|e| {
+                warn!("Failed to create filter evaluator: {}", e);
+                e
+            })
             .ok()?;
 
         Some(Self {

kernel/src/scan/log_replay.rs

@@ -1,6 +1,5 @@
 use std::clone::Clone;
 use std::collections::HashSet;
-use std::iter;
 use std::sync::{Arc, LazyLock};
 
 use tracing::debug;
@@ -239,29 +238,21 @@ impl LogReplayScanner {
 pub fn scan_action_iter(
     engine: &dyn Engine,
     action_iter: impl Iterator<Item = DeltaResult<(Box<dyn EngineData>, bool)>> + 'static,
-    table_schema: &SchemaRef,
-    predicate: Option<ExpressionRef>,
-) -> Box<dyn Iterator<Item = DeltaResult<ScanData>>> {
-    let mut log_scanner = LogReplayScanner::new(engine, table_schema, predicate);
-    match engine.get_expression_handler().get_evaluator(
+    physical_predicate: Option<(ExpressionRef, SchemaRef)>,
+) -> 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(),
-    ) {
-        Ok(add_transform) => Box::new(
-            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))),
-        ),
-        Err(e) => Box::new(iter::once(Err(e))),
-    }
+    )?;
+
+    Ok(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))))
 }
 
 #[cfg(test)]
@@ -301,7 +292,8 @@ mod tests {
             &[true, false],
             (),
             validate_simple,
-        );
+        )
+        .unwrap();
     }
 
     #[test]
@@ -311,6 +303,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())
     }
 
@@ -701,7 +703,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;
@@ -753,26 +755,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

@@ -235,14 +235,11 @@ impl LogReplayScanner {
             .version
             .try_into()
             .map_err(|_| Error::generic("Failed to convert commit version to i64"))?;
-        let evaluator = engine
-            .get_expression_handler()
-            .get_evaluator(
-                get_log_add_schema().clone(),
-                cdf_scan_row_expression(timestamp, commit_version),
-                cdf_scan_row_schema().into(),
-            )?;
-
+        let evaluator = engine.get_expression_handler().get_evaluator(
+            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

@@ -281,7 +281,7 @@ fn read_scan_file(
         physical_schema.clone(),
         physical_to_logical_expr,
         global_state.logical_schema.clone().into(),
-    );
+    )?;
 
     let table_root = Url::parse(&global_state.table_root)?;
     let location = table_root.join(&scan_file.path)?;