feat: Add support for kernel default expression evaluation (delta-io#979)

## What changes are proposed in this pull request?

Pathfinding on delta-io#686
exposed a gap in kernel's default predicate evaluator API: It cannot
evaluate expressions, even tho expressions can legitimately be input to
predicates, e.g.
```sql
WHERE x < y + z
```

Add that missing support, with arithmetic on integer types as the
exemplar. Support for floating point types is left as future work,
because it requires extra care to decide how to handle Inf and NaN
values: Should they propagate as-is? Or one/both be detected and
converted to `Scalar::Null`?

NOTE: This PR is _not_ adding arbitrary expression evaluation support to
the `KernelPredicateEvaluator` API. That would require a much bigger
design, and it's not clear that it's even desirable -- normal expression
evaluation should be handled by the engine, potentially using an
`ExpressionEvaluationHandler` if the connector relies on the default
engine for expression evaluation.

## How was this change tested?

New unit tests

Author: scovich

kernel/src/expressions/scalars.rs

@@ -299,6 +299,58 @@ impl Scalar {
         };
         Ok(Self::Timestamp(timestamp.timestamp_micros()))
     }
+
+    /// Attempts to add two scalars, returning None if they were incompatible.
+    pub fn try_add(&self, other: &Scalar) -> Option<Scalar> {
+        use Scalar::*;
+        let result = match (self, other) {
+            (Integer(a), Integer(b)) => Integer(a.checked_add(*b)?),
+            (Long(a), Long(b)) => Long(a.checked_add(*b)?),
+            (Short(a), Short(b)) => Short(a.checked_add(*b)?),
+            (Byte(a), Byte(b)) => Byte(a.checked_add(*b)?),
+            _ => return None,
+        };
+        Some(result)
+    }
+
+    /// Attempts to subtract two scalars, returning None if they were incompatible.
+    pub fn try_sub(&self, other: &Scalar) -> Option<Scalar> {
+        use Scalar::*;
+        let result = match (self, other) {
+            (Integer(a), Integer(b)) => Integer(a.checked_sub(*b)?),
+            (Long(a), Long(b)) => Long(a.checked_sub(*b)?),
+            (Short(a), Short(b)) => Short(a.checked_sub(*b)?),
+            (Byte(a), Byte(b)) => Byte(a.checked_sub(*b)?),
+            _ => return None,
+        };
+        Some(result)
+    }
+
+    /// Attempts to multiply two scalars, returning None if they were incompatible.
+    pub fn try_mul(&self, other: &Scalar) -> Option<Scalar> {
+        use Scalar::*;
+        let result = match (self, other) {
+            (Integer(a), Integer(b)) => Integer(a.checked_mul(*b)?),
+            (Long(a), Long(b)) => Long(a.checked_mul(*b)?),
+            (Short(a), Short(b)) => Short(a.checked_mul(*b)?),
+            (Byte(a), Byte(b)) => Byte(a.checked_mul(*b)?),
+            _ => return None,
+        };
+        Some(result)
+    }
+
+    /// Attempts to divide two scalars, returning None if they were incompatible.
+    pub fn try_div(&self, other: &Scalar) -> Option<Scalar> {
+        use Scalar::*;
+        let result = match (self, other) {
+            (Integer(a), Integer(b)) => Integer(a.checked_div(*b)?),
+            (Long(a), Long(b)) => Long(a.checked_div(*b)?),
+            (Short(a), Short(b)) => Short(a.checked_div(*b)?),
+            (Byte(a), Byte(b)) => Byte(a.checked_div(*b)?),
+            _ => return None,
+        };
+        Some(result)
+    }
 }
 
 impl Display for Scalar {

kernel/src/kernel_predicates/mod.rs

@@ -4,8 +4,9 @@
 //! but data skipping "evaluation" actually produces a transformed predicate that replaces column
 //! references with stats column references, which log replay will instruct the engine to evaluate.
 use crate::expressions::{
-    BinaryPredicate, BinaryPredicateOp, ColumnName, Expression as Expr, JunctionPredicate,
-    JunctionPredicateOp, Predicate as Pred, Scalar, UnaryPredicate, UnaryPredicateOp,
+    BinaryExpression, BinaryExpressionOp, BinaryPredicate, BinaryPredicateOp, ColumnName,
+    Expression as Expr, JunctionPredicate, JunctionPredicateOp, Predicate as Pred, Scalar,
+    UnaryPredicate, UnaryPredicateOp,
 };
 use crate::schema::DataType;
 
@@ -571,6 +572,25 @@ impl<R: ResolveColumnAsScalar> DefaultKernelPredicateEvaluator<R> {
     fn resolve_column(&self, col: &ColumnName) -> Option<Scalar> {
         self.resolver.resolve_column(col)
     }
+
+    #[allow(unused)] // TODO: remove this annotation once opaque predicates start using it
+    pub(crate) fn eval_expr(&self, expr: &Expr) -> Option<Scalar> {
+        match expr {
+            Expr::Literal(value) => Some(value.clone()),
+            Expr::Column(name) => self.resolve_column(name),
+            Expr::Predicate(pred) => self.eval_pred(pred, false).map(Scalar::from),
+            Expr::Struct(_) => None, // TODO
+            Expr::Binary(BinaryExpression { op, left, right }) => {
+                let op_fn = match op {
+                    BinaryExpressionOp::Plus => Scalar::try_add,
+                    BinaryExpressionOp::Minus => Scalar::try_sub,
+                    BinaryExpressionOp::Multiply => Scalar::try_mul,
+                    BinaryExpressionOp::Divide => Scalar::try_div,
+                };
+                op_fn(&self.eval_expr(left)?, &self.eval_expr(right)?)
+            }
+        }
+    }
 }
 
 impl<R: ResolveColumnAsScalar + 'static> From<R> for DefaultKernelPredicateEvaluator<R> {

kernel/src/kernel_predicates/tests.rs

@@ -181,6 +181,135 @@ fn test_default_partial_cmp_scalars() {
     }
 }
 
+#[test]
+fn test_default_scalar_arithmetic() {
+    use Scalar::*;
+    let left = &[Byte(2), Short(200), Integer(20000), Long(2000000)];
+    let right = &[Byte(3), Short(30), Integer(3000), Long(300000)];
+    let expected = [
+        (Byte(5), Byte(-1), Byte(6), Byte(0)),
+        (Short(230), Short(170), Short(6000), Short(6)),
+        (
+            Integer(23000),
+            Integer(17000),
+            Integer(60000000),
+            Integer(6),
+        ),
+        (Long(2300000), Long(1700000), Long(600000000000), Long(6)),
+    ];
+
+    let filter = DefaultKernelPredicateEvaluator::from(Scalar::from(1));
+    for ((l, r), (add, sub, mul, div)) in left.iter().zip(right).zip(expected) {
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) + Expr::literal(r.clone()))),
+            Some(add),
+            "add({l:?}, {r:?})"
+        );
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) - Expr::literal(r.clone()))),
+            Some(sub),
+            "sub({l:?}, {r:?})"
+        );
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) * Expr::literal(r.clone()))),
+            Some(mul),
+            "mul({l:?}, {r:?})"
+        );
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) / Expr::literal(r.clone()))),
+            Some(div),
+            "div({l:?}, {r:?})"
+        );
+    }
+
+    // Invalid type combinations
+    expect_eq!(
+        filter.eval_expr(&(Expr::literal("hi") + Expr::literal("ho"))),
+        None,
+        "add(string, string)"
+    );
+    expect_eq!(
+        filter.eval_expr(&(Expr::literal(1i8) + Expr::literal(1i64))),
+        None,
+        "add(byte, long)"
+    );
+    expect_eq!(
+        filter.eval_expr(&(Expr::literal(1i8) - Expr::literal(1i64))),
+        None,
+        "sub(byte, long)"
+    );
+    expect_eq!(
+        filter.eval_expr(&(Expr::literal(1i8) * Expr::literal(1i64))),
+        None,
+        "mul(byte, long)"
+    );
+    expect_eq!(
+        filter.eval_expr(&(Expr::literal(1i8) / Expr::literal(1i64))),
+        None,
+        "div(byte, long)"
+    );
+
+    // Addition overflow
+    let args = [
+        (Byte(i8::MAX), Byte(1)),
+        (Short(i16::MAX), Short(1)),
+        (Integer(i32::MAX), Integer(1)),
+        (Long(i64::MAX), Long(1)),
+    ];
+    for (l, r) in args {
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) + Expr::literal(r.clone()))),
+            None,
+            "add({l:?}, {r:?})"
+        );
+    }
+
+    // Subtraction overflow
+    let args = [
+        (Byte(i8::MIN), Byte(1)),
+        (Short(i16::MIN), Short(1)),
+        (Integer(i32::MIN), Integer(1)),
+        (Long(i64::MIN), Long(1)),
+    ];
+    for (l, r) in args {
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) - Expr::literal(r.clone()))),
+            None,
+            "sub({l:?}, {r:?})"
+        );
+    }
+
+    // Multiplication overflow
+    let args = [
+        Byte(i8::MAX),
+        Short(i16::MAX),
+        Integer(i32::MAX),
+        Long(i64::MAX),
+    ];
+    for arg in args {
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(arg.clone()) * Expr::literal(arg.clone()))),
+            None,
+            "mul({arg:?}, {arg:?})"
+        );
+    }
+
+    // Division overflow
+    let args = [
+        (Byte(i8::MAX), Byte(0)),
+        (Short(i16::MAX), Short(0)),
+        (Integer(i32::MAX), Integer(0)),
+        (Long(i64::MAX), Long(0)),
+    ];
+    for (l, r) in args {
+        expect_eq!(
+            filter.eval_expr(&(Expr::literal(l.clone()) / Expr::literal(r.clone()))),
+            None,
+            "div({l:?}, {r:?})"
+        );
+    }
+}
+
 // Verifies that eval_binary_scalars uses partial_cmp_scalars correctly
 #[test]
 fn test_eval_binary_scalars() {