buoyant_kernel/expressions/
mod.rs

1//! Definitions and functions to create and manipulate kernel expressions
2
3use std::borrow::Cow;
4use std::collections::{HashMap, HashSet};
5use std::fmt::{Display, Formatter};
6use std::sync::Arc;
7
8use itertools::Itertools;
9use serde::{de, ser, Deserialize, Deserializer, Serialize, Serializer};
10
11pub use self::column_names::{
12    column_expr, column_expr_ref, column_name, column_pred, joined_column_expr, joined_column_name,
13    ColumnName,
14};
15pub use self::scalars::{ArrayData, DecimalData, MapData, Scalar, StructData};
16use crate::kernel_predicates::{
17    DirectDataSkippingPredicateEvaluator, DirectPredicateEvaluator,
18    IndirectDataSkippingPredicateEvaluator,
19};
20use crate::schema::SchemaRef;
21use crate::transforms::ExpressionTransform;
22use crate::{DataType, DeltaResult, DynPartialEq};
23
24mod column_names;
25pub(crate) mod literal_expression_transform;
26pub(crate) use literal_expression_transform::literal_expression_transform;
27mod scalars;
28
29pub type ExpressionRef = std::sync::Arc<Expression>;
30pub type PredicateRef = std::sync::Arc<Predicate>;
31
32////////////////////////////////////////////////////////////////////////
33// Operators
34////////////////////////////////////////////////////////////////////////
35
36/// A unary predicate operator.
37#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
38pub enum UnaryPredicateOp {
39    /// Unary Is Null
40    IsNull,
41}
42
43/// A binary predicate operator.
44#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
45pub enum BinaryPredicateOp {
46    /// Comparison Less Than
47    LessThan,
48    /// Comparison Greater Than
49    GreaterThan,
50    /// Comparison Equal
51    Equal,
52    /// Distinct
53    Distinct,
54    /// IN
55    In,
56}
57
58/// A unary expression operator.
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
60pub enum UnaryExpressionOp {
61    /// Convert struct data to JSON-encoded strings
62    ToJson,
63}
64
65/// A binary expression operator.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
67pub enum BinaryExpressionOp {
68    /// Arithmetic Plus
69    Plus,
70    /// Arithmetic Minus
71    Minus,
72    /// Arithmetic Multiply
73    Multiply,
74    /// Arithmetic Divide
75    Divide,
76}
77
78/// A variadic expression operator.
79#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
80pub enum VariadicExpressionOp {
81    /// Collapse multiple values into one by taking the first non-null value
82    Coalesce,
83}
84
85/// A junction (AND/OR) predicate operator.
86#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
87pub enum JunctionPredicateOp {
88    /// Conjunction
89    And,
90    /// Disjunction
91    Or,
92}
93
94/// A kernel-supplied scalar expression evaluator which in particular can convert column references
95/// (i.e. [`Expression::Column`]) to [`Scalar`] values. [`OpaqueExpressionOp::eval_expr_scalar`] and
96/// [`OpaquePredicateOp::eval_pred_scalar`] rely on this evaluator.
97///
98/// If the evaluator produces `None`, it means kernel was unable to evaluate
99/// the input expression. Otherwise, `Some(Scalar)` is the result of that evaluation (possibly
100/// `Scalar::Null` if the output was NULL).
101pub type ScalarExpressionEvaluator<'a> = dyn Fn(&Expression) -> Option<Scalar> + 'a;
102
103/// An opaque expression operation (ie defined and implemented by the engine).
104pub trait OpaqueExpressionOp: DynPartialEq + std::fmt::Debug {
105    /// Succinctly identifies this op
106    fn name(&self) -> &str;
107
108    /// Attempts scalar evaluation of this opaque expression, e.g. for partition pruning.
109    ///
110    /// Implementations can evaluate the child expressions however they see fit, possibly by
111    /// calling back to the provided [`ScalarExpressionEvaluator`],
112    ///
113    /// An output of `Err` indicates that this operation does not support scalar evaluation, or was
114    /// invoked incorrectly (e.g. with the wrong number and/or types of arguments, None input,
115    /// etc); the operation is disqualified from participating in partition pruning.
116    ///
117    /// `Ok(Scalar::Null)` means the operation actually produced a legitimately NULL result.
118    fn eval_expr_scalar(
119        &self,
120        eval_expr: &ScalarExpressionEvaluator<'_>,
121        exprs: &[Expression],
122    ) -> DeltaResult<Scalar>;
123}
124
125/// An opaque predicate operation (ie defined and implemented by the engine).
126pub trait OpaquePredicateOp: DynPartialEq + std::fmt::Debug {
127    /// Succinctly identifies this op
128    fn name(&self) -> &str;
129
130    /// Attempts scalar evaluation of this (possibly inverted) opaque predicate on behalf of a
131    /// [`DirectPredicateEvaluator`], e.g. for partition pruning or to evaluate an opaque data
132    /// skipping predicate produced previously by an [`IndirectDataSkippingPredicateEvaluator`].
133    ///
134    /// Implementations can evaluate the child expressions however they see fit, possibly by calling
135    /// back to the provided [`ScalarExpressionEvaluator`] and/or [`DirectPredicateEvaluator`].
136    ///
137    /// An output of `Err` indicates that this operation does not support scalar evaluation, or was
138    /// invoked incorrectly (e.g. wrong number and/or types of arguments, None input, etc); the
139    /// operation is disqualified from participating in partition pruning and/or data skipping.
140    ///
141    /// `Ok(None)` means the operation actually produced a legitimately NULL output.
142    fn eval_pred_scalar(
143        &self,
144        eval_expr: &ScalarExpressionEvaluator<'_>,
145        eval_pred: &DirectPredicateEvaluator<'_>,
146        exprs: &[Expression],
147        inverted: bool,
148    ) -> DeltaResult<Option<bool>>;
149
150    /// Evaluates this (possibly inverted) opaque predicate for data skipping on behalf of a
151    /// [`DirectDataSkippingPredicateEvaluator`], e.g. for parquet row group skipping.
152    ///
153    /// Implementations can evaluate the child expressions however they see fit, possibly by
154    /// calling back to the provided [`DirectDataSkippingPredicateEvaluator`].
155    ///
156    /// An output of `None` indicates that this operation does not support evaluation as a data
157    /// skipping predicate, or was invoked incorrectly (e.g. wrong number and/or types of arguments,
158    /// None input, etc.); the operation is disqualified from participating in row group skipping.
159    fn eval_as_data_skipping_predicate(
160        &self,
161        evaluator: &DirectDataSkippingPredicateEvaluator<'_>,
162        exprs: &[Expression],
163        inverted: bool,
164    ) -> Option<bool>;
165
166    /// Converts this (possibly inverted) opaque predicate to a data skipping predicate on behalf of
167    /// an [`IndirectDataSkippingPredicateEvaluator`], e.g. for stats-based file pruning.
168    ///
169    /// Implementations can transform the predicate and its child expressions however they see fit,
170    /// possibly by calling back to the owning [`IndirectDataSkippingPredicateEvaluator`].
171    ///
172    /// An output of `None` indicates that this operation does not support conversion to a data
173    /// skipping predicate, or was invoked incorrectly (e.g. wrong number and/or types of arguments,
174    /// None input, etc.); the operation is disqualified from participating in file pruning.
175    //
176    // NOTE: It would be nicer if this method could accept an `Arc<Self>`, in case the data skipping
177    // predicate rewrite can reuse the same operation. But sadly, that would not be dyn-compatible.
178    fn as_data_skipping_predicate(
179        &self,
180        evaluator: &IndirectDataSkippingPredicateEvaluator<'_>,
181        exprs: &[Expression],
182        inverted: bool,
183    ) -> Option<Predicate>;
184}
185
186/// A shared reference to an [`OpaqueExpressionOp`] instance.
187pub type OpaqueExpressionOpRef = Arc<dyn OpaqueExpressionOp>;
188
189/// A shared reference to an [`OpaquePredicateOp`] instance.
190pub type OpaquePredicateOpRef = Arc<dyn OpaquePredicateOp>;
191
192////////////////////////////////////////////////////////////////////////
193// Expressions and predicates
194////////////////////////////////////////////////////////////////////////
195
196#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
197pub struct UnaryPredicate {
198    /// The operator.
199    pub op: UnaryPredicateOp,
200    /// The input expression.
201    pub expr: Box<Expression>,
202}
203
204#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
205pub struct BinaryPredicate {
206    /// The operator.
207    pub op: BinaryPredicateOp,
208    /// The left-hand side of the operation.
209    pub left: Box<Expression>,
210    /// The right-hand side of the operation.
211    pub right: Box<Expression>,
212}
213
214#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
215pub struct UnaryExpression {
216    /// The operator.
217    pub op: UnaryExpressionOp,
218    /// The input expression.
219    pub expr: Box<Expression>,
220}
221
222#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
223pub struct BinaryExpression {
224    /// The operator.
225    pub op: BinaryExpressionOp,
226    /// The left-hand side of the operation.
227    pub left: Box<Expression>,
228    /// The right-hand side of the operation.
229    pub right: Box<Expression>,
230}
231
232#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
233pub struct VariadicExpression {
234    /// The operator.
235    pub op: VariadicExpressionOp,
236    /// The input expressions.
237    pub exprs: Vec<Expression>,
238}
239
240/// An expression that parses a JSON string into a struct with the given schema.
241/// This is the inverse of `ToJson` - it converts a JSON-encoded string column into a
242/// struct column.
243#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
244pub struct ParseJsonExpression {
245    /// The expression that evaluates to a STRING column containing JSON objects.
246    pub json_expr: Box<Expression>,
247    /// The schema defining the structure to parse the JSON into.
248    pub output_schema: SchemaRef,
249}
250
251#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
252pub struct JunctionPredicate {
253    /// The operator.
254    pub op: JunctionPredicateOp,
255    /// The input predicates.
256    pub preds: Vec<Predicate>,
257}
258
259// NOTE: We have to use `Arc<dyn OpaquePredicateOp>` instead of `Box<dyn OpaquePredicateOp>` because
260// we cannot require `OpaquePredicateOp: Clone` (not a dyn-compatible trait). Instead, we must rely
261// on cheap `Arc` clone, which does not duplicate the inner object.
262//
263// TODO(#1564): OpaquePredicate currently does not support serialization or deserialization. In the
264// future, the [`OpaquePredicateOp`] trait can be extended to support ser/de.
265#[derive(Clone, Debug)]
266pub struct OpaquePredicate {
267    pub op: OpaquePredicateOpRef,
268    pub exprs: Vec<Expression>,
269}
270fn fail_serialize_opaque_predicate<S>(
271    _value: &OpaquePredicate,
272    _serializer: S,
273) -> Result<S::Ok, S::Error>
274where
275    S: Serializer,
276{
277    Err(ser::Error::custom("Cannot serialize an Opaque Predicate"))
278}
279
280fn fail_deserialize_opaque_predicate<'de, D>(_deserializer: D) -> Result<OpaquePredicate, D::Error>
281where
282    D: Deserializer<'de>,
283{
284    Err(de::Error::custom("Cannot deserialize an Opaque Predicate"))
285}
286
287impl OpaquePredicate {
288    pub(crate) fn new(
289        op: OpaquePredicateOpRef,
290        exprs: impl IntoIterator<Item = Expression>,
291    ) -> Self {
292        let exprs = exprs.into_iter().collect();
293        Self { op, exprs }
294    }
295}
296
297// NOTE: We have to use `Arc<dyn OpaqueExpressionOp>` instead of `Box<dyn OpaqueExpressionOp>`
298// because we cannot require `OpaqueExpressionOp: Clone` (not a dyn-compatible trait). Instead, we
299// must rely on cheap `Arc` clone, which does not duplicate the inner object.
300//
301// TODO(#1564): OpaqueExpression currently does not support serialization or deserialization. In the
302// future, the [`OpaqueExpressionOp`] trait can be extended to support ser/de.
303#[derive(Clone, Debug)]
304pub struct OpaqueExpression {
305    pub op: OpaqueExpressionOpRef,
306    pub exprs: Vec<Expression>,
307}
308
309impl OpaqueExpression {
310    pub(crate) fn new(
311        op: OpaqueExpressionOpRef,
312        exprs: impl IntoIterator<Item = Expression>,
313    ) -> Self {
314        let exprs = exprs.into_iter().collect();
315        Self { op, exprs }
316    }
317}
318
319fn fail_serialize_opaque_expression<S>(
320    _value: &OpaqueExpression,
321    _serializer: S,
322) -> Result<S::Ok, S::Error>
323where
324    S: Serializer,
325{
326    Err(ser::Error::custom("Cannot serialize an Opaque Expression"))
327}
328
329fn fail_deserialize_opaque_expression<'de, D>(
330    _deserializer: D,
331) -> Result<OpaqueExpression, D::Error>
332where
333    D: Deserializer<'de>,
334{
335    Err(de::Error::custom("Cannot deserialize an Opaque Expression"))
336}
337
338/// A transformation affecting a single field (one pieces of a [`Transform`]). The transformation
339/// could insert 0+ new fields after the target, or could replace the target with 0+ a new fields).
340#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
341pub struct FieldTransform {
342    /// The list of expressions this field transform emits at the target location.
343    pub exprs: Vec<ExpressionRef>,
344    /// If true, the output expressions replace the input field instead of following after it.
345    pub is_replace: bool,
346    /// If true, this transform is silently ignored when the target field does not exist in the
347    /// input. Otherwise, a missing target field produces an error.
348    pub optional: bool,
349}
350
351/// A transformation that efficiently represents sparse modifications to struct schemas.
352///
353/// `Transform` achieves `O(changes)` space complexity instead of `O(schema_width)` by only
354/// specifying those fields that actually change (inserted, replaced, or deleted). Any input field
355/// not specifically mentioned by the transform is passed through, unmodified and with the same
356/// relative field ordering. This is particularly useful for wide schemas where only a few columns
357/// need to be modified and/or dropped, or where a small number of columns need to be injected.
358#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
359pub struct Transform {
360    /// The path to the nested input struct this transform operates on (if any). If no path is
361    /// given, the transform operates directly on top-level columns.
362    pub input_path: Option<ColumnName>,
363    /// A mapping from named input fields to the transform to be performed on each field.
364    pub field_transforms: HashMap<String, FieldTransform>,
365    /// A list of new fields to emit before processing the first input field.
366    pub prepended_fields: Vec<ExpressionRef>,
367}
368
369impl Transform {
370    /// Creates a new top-level identity transform. The various `with_xxx` helper methods can be
371    /// used to add specific field transforms.
372    pub fn new_top_level() -> Self {
373        Self::default()
374    }
375
376    /// Creates a new identity transform that operates on fields of a nested struct identified by
377    /// `path`. The various `with_xxx` helper methods can be used to add specific field transforms.
378    pub fn new_nested<A>(path: impl IntoIterator<Item = A>) -> Self
379    where
380        ColumnName: FromIterator<A>,
381    {
382        Self {
383            input_path: Some(ColumnName::new(path)),
384            ..Default::default()
385        }
386    }
387
388    /// Specifies a field to drop.
389    pub fn with_dropped_field(mut self, name: impl Into<String>) -> Self {
390        let field_transform = self.field_transform(name);
391        field_transform.is_replace = true;
392        self
393    }
394
395    /// Like [`Self::with_dropped_field`], but silently ignored if the field does not exist.
396    pub fn with_dropped_field_if_exists(mut self, name: impl Into<String>) -> Self {
397        let field_transform = self.field_transform(name);
398        field_transform.is_replace = true;
399        field_transform.optional = true;
400        self
401    }
402
403    /// Specifies an expression to replace a field with.
404    pub fn with_replaced_field(mut self, name: impl Into<String>, expr: ExpressionRef) -> Self {
405        let field_transform = self.field_transform(name);
406        field_transform.exprs.push(expr);
407        field_transform.is_replace = true;
408        self
409    }
410
411    /// Specifies an expression to insert after an optional predecessor (None = prepend, emit the
412    /// expression before the first input field). Multiple fields can be inserted after the same
413    /// predecessor, and they will be emitted in the same order they were registered.
414    pub fn with_inserted_field(
415        mut self,
416        after: Option<impl Into<String>>,
417        expr: ExpressionRef,
418    ) -> Self {
419        match after {
420            Some(field_name) => self.field_transform(field_name).exprs.push(expr),
421            None => self.prepended_fields.push(expr),
422        }
423        self
424    }
425
426    /// True if this is the identity transform (all input fields pass through unchanged, with no new
427    /// fields inserted).
428    pub fn is_identity(&self) -> bool {
429        self.prepended_fields.is_empty() && self.field_transforms.is_empty()
430    }
431
432    /// None, if this is a top-level transform. Otherwise, the path of this nested transform.
433    pub fn input_path(&self) -> Option<&ColumnName> {
434        self.input_path.as_ref()
435    }
436
437    // Gets or creates the field transform for a named input field
438    fn field_transform(&mut self, field_name: impl Into<String>) -> &mut FieldTransform {
439        self.field_transforms.entry(field_name.into()).or_default()
440    }
441}
442
443/// A SQL expression.
444///
445/// These expressions do not track or validate data types, other than the type
446/// of literals. It is up to the expression evaluator to validate the
447/// expression against a schema and add appropriate casts as required.
448#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
449pub enum Expression {
450    /// A literal value.
451    Literal(Scalar),
452    /// A column reference by name.
453    Column(ColumnName),
454    /// A predicate treated as a boolean expression
455    Predicate(Box<Predicate>), // should this be Arc?
456    /// A struct computed from a Vec of expressions.
457    /// The optional nullability predicate, if provided and evaluates to false/null, makes the entire struct null.
458    Struct(Vec<ExpressionRef>, Option<ExpressionRef>),
459    /// A sparse transformation of a struct schema. More efficient than `Struct` for wide schemas
460    /// where only a few fields change, achieving O(changes) instead of O(schema_width) complexity.
461    Transform(Transform),
462    /// An expression that takes one expression as input.
463    Unary(UnaryExpression),
464    /// An expression that takes two expressions as input.
465    Binary(BinaryExpression),
466    /// An expression that takes a variable number of expressions as input.
467    Variadic(VariadicExpression),
468    /// An expression that the engine defines and implements. Kernel interacts with the expression
469    /// only through methods provided by the [`OpaqueExpressionOp`] trait.
470    #[serde(serialize_with = "fail_serialize_opaque_expression")]
471    #[serde(deserialize_with = "fail_deserialize_opaque_expression")]
472    Opaque(OpaqueExpression),
473    /// An unknown expression (i.e. one that neither kernel nor engine attempts to evaluate). For
474    /// data skipping purposes, kernel treats unknown expressions as if they were literal NULL
475    /// values (which may disable skipping if it "poisons" the predicate), but engines MUST NOT
476    /// attempt to interpret them as NULL when evaluating query filters because it could produce
477    /// incorrect results. For example, converting `WHERE <fancy-udf-invocation> IS NULL` to `WHERE
478    /// <unknown> IS NULL` to `WHERE NULL IS NULL` is equivalent to `WHERE TRUE` and would include
479    /// all rows -- almost certainly NOT what the query author intended. Use `Expression::Opaque`
480    /// for expressions kernel doesn't understand but which engine can still evaluate.
481    Unknown(String),
482    /// Parse a JSON string expression into a struct with the given schema.
483    ParseJson(ParseJsonExpression),
484    /// Extract keys from a `Map<String, String>` and parse values into a typed struct using
485    /// Delta's partition value serialization rules.
486    MapToStruct(MapToStructExpression),
487}
488
489/// A SQL predicate.
490///
491/// These predicates do not track or validate data types, other than the type
492/// of literals. It is up to the predicate evaluator to validate the
493/// predicate against a schema and add appropriate casts as required.
494#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
495pub enum Predicate {
496    /// A boolean-valued expression, useful for e.g. `AND(<boolean_col1>, <boolean_col2>)`.
497    BooleanExpression(Expression),
498    /// Boolean inversion (true <-> false)
499    ///
500    /// NOTE: NOT is not a normal unary predicate, because it requires a predicate as input (not an
501    /// expression), and is never directly evaluated. Instead, observing that all predicates are
502    /// invertible, NOT is always pushed down into its child predicate, inverting it. For example,
503    /// `NOT (a < b)` pushes down and inverts `<` to `>=`, producing `a >= b`.
504    Not(Box<Predicate>),
505    /// A unary operation.
506    Unary(UnaryPredicate),
507    /// A binary operation.
508    Binary(BinaryPredicate),
509    /// A junction operation (AND/OR).
510    Junction(JunctionPredicate),
511    /// A predicate that the engine defines and implements. Kernel interacts with the predicate
512    /// only through methods provided by the [`OpaquePredicateOp`] trait.
513    #[serde(serialize_with = "fail_serialize_opaque_predicate")]
514    #[serde(deserialize_with = "fail_deserialize_opaque_predicate")]
515    Opaque(OpaquePredicate),
516    /// An unknown predicate (i.e. one that neither kernel nor engine attempts to evaluate). For
517    /// data skipping purposes, kernel treats unknown predicates as if they were literal NULL values
518    /// (which may disable skipping if it "poisons" the predicate), but engines MUST NOT attempt to
519    /// interpret them as NULL when evaluating query filters because it could produce incorrect
520    /// results. For example, converting `WHERE <fancy-udf-invocation>` to `WHERE NULL` is
521    /// equivalent to `WHERE FALSE` and would filter out all rows -- almost certainly NOT what the
522    /// query author intended. Use `Predicate::Opaque` for predicates kernel doesn't understand
523    /// but which engine can still evaluate.
524    Unknown(String),
525}
526
527////////////////////////////////////////////////////////////////////////
528// Struct/Enum impls
529////////////////////////////////////////////////////////////////////////
530
531impl BinaryPredicateOp {
532    /// True if this is a comparison for which NULL input always produces NULL output
533    pub(crate) fn is_null_intolerant(&self) -> bool {
534        use BinaryPredicateOp::*;
535        match self {
536            LessThan | GreaterThan | Equal => true,
537            Distinct | In => false, // tolerates NULL input
538        }
539    }
540}
541
542impl JunctionPredicateOp {
543    pub(crate) fn invert(&self) -> JunctionPredicateOp {
544        use JunctionPredicateOp::*;
545        match self {
546            And => Or,
547            Or => And,
548        }
549    }
550}
551
552impl UnaryExpression {
553    pub(crate) fn new(op: UnaryExpressionOp, expr: impl Into<Expression>) -> Self {
554        let expr = Box::new(expr.into());
555        Self { op, expr }
556    }
557}
558
559impl UnaryPredicate {
560    pub(crate) fn new(op: UnaryPredicateOp, expr: impl Into<Expression>) -> Self {
561        let expr = Box::new(expr.into());
562        Self { op, expr }
563    }
564}
565
566impl BinaryExpression {
567    pub(crate) fn new(
568        op: BinaryExpressionOp,
569        left: impl Into<Expression>,
570        right: impl Into<Expression>,
571    ) -> Self {
572        let left = Box::new(left.into());
573        let right = Box::new(right.into());
574        Self { op, left, right }
575    }
576}
577
578impl BinaryPredicate {
579    pub(crate) fn new(
580        op: BinaryPredicateOp,
581        left: impl Into<Expression>,
582        right: impl Into<Expression>,
583    ) -> Self {
584        let left = Box::new(left.into());
585        let right = Box::new(right.into());
586        Self { op, left, right }
587    }
588}
589
590impl VariadicExpression {
591    pub(crate) fn new(
592        op: VariadicExpressionOp,
593        exprs: impl IntoIterator<Item = impl Into<Expression>>,
594    ) -> Self {
595        let exprs = exprs.into_iter().map(Into::into).collect();
596        Self { op, exprs }
597    }
598}
599
600impl ParseJsonExpression {
601    pub(crate) fn new(json_expr: impl Into<Expression>, output_schema: SchemaRef) -> Self {
602        Self {
603            json_expr: Box::new(json_expr.into()),
604            output_schema,
605        }
606    }
607}
608
609/// Transforms a `Map<String, String>` column into a struct whose schema is provided by the
610/// evaluator's output type (via `result_type`). Each row in the map column becomes one row in
611/// the output struct column: a `key` -> `value` mapping in the map means the struct field named
612/// `key` receives `value`, parsed into the field's target type using Delta's partition value
613/// serialization rules ([`PrimitiveType::parse_scalar`]).
614///
615/// - Missing keys produce null values
616/// - Parse errors are propagated (indicating a broken table)
617/// - Duplicate map keys are resolved by taking the rightmost entry
618///
619/// [`PrimitiveType::parse_scalar`]: crate::schema::PrimitiveType::parse_scalar
620#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
621pub struct MapToStructExpression {
622    /// The expression that evaluates to a `Map<String, String>` column.
623    pub map_expr: Box<Expression>,
624}
625
626impl MapToStructExpression {
627    pub(crate) fn new(map_expr: impl Into<Expression>) -> Self {
628        Self {
629            map_expr: Box::new(map_expr.into()),
630        }
631    }
632}
633
634impl JunctionPredicate {
635    pub(crate) fn new(op: JunctionPredicateOp, preds: Vec<Predicate>) -> Self {
636        Self { op, preds }
637    }
638}
639
640impl Expression {
641    /// Returns a set of columns referenced by this expression.
642    pub fn references(&self) -> HashSet<&ColumnName> {
643        let mut references = GetColumnReferences::default();
644        let _ = references.transform_expr(self);
645        references.0
646    }
647
648    /// Create a new column name expression from input satisfying `FromIterator for ColumnName`.
649    pub fn column<A>(field_names: impl IntoIterator<Item = A>) -> Expression
650    where
651        ColumnName: FromIterator<A>,
652    {
653        ColumnName::new(field_names).into()
654    }
655
656    /// Create a new expression for a literal value
657    pub fn literal(value: impl Into<Scalar>) -> Self {
658        Self::Literal(value.into())
659    }
660
661    /// Creates a NULL literal expression
662    pub const fn null_literal(data_type: DataType) -> Self {
663        Self::Literal(Scalar::Null(data_type))
664    }
665
666    /// Wraps a predicate as a boolean-valued expression
667    pub fn from_pred(value: Predicate) -> Self {
668        match value {
669            Predicate::BooleanExpression(expr) => expr,
670            _ => Self::Predicate(Box::new(value)),
671        }
672    }
673
674    /// Create a new struct expression.
675    ///
676    /// The field names and types are supplied by the caller at evaluation time via the
677    /// `result_type` parameter of the expression evaluator. Use this when the schema is
678    /// always available from external context (e.g. the expression is the top-level output
679    /// of [`crate::ExpressionEvaluator`]).
680    pub fn struct_from(exprs: impl IntoIterator<Item = impl Into<Arc<Self>>>) -> Self {
681        Self::Struct(exprs.into_iter().map(Into::into).collect(), None)
682    }
683
684    /// Create a new struct expression with a nullability predicate.
685    ///
686    /// When the predicate evaluates to false or null for a row, the entire struct is null
687    /// for that row.
688    pub fn struct_with_nullability_from(
689        exprs: impl IntoIterator<Item = impl Into<Arc<Self>>>,
690        nullability_predicate: impl Into<Arc<Self>>,
691    ) -> Self {
692        Self::Struct(
693            exprs.into_iter().map(Into::into).collect(),
694            Some(nullability_predicate.into()),
695        )
696    }
697
698    /// Create a new transform expression
699    pub fn transform(transform: Transform) -> Self {
700        Self::Transform(transform)
701    }
702
703    /// Create a new predicate `self IS NULL`
704    pub fn is_null(self) -> Predicate {
705        Predicate::is_null(self)
706    }
707
708    /// Create a new predicate `self IS NOT NULL`
709    pub fn is_not_null(self) -> Predicate {
710        Predicate::is_not_null(self)
711    }
712
713    /// Create a new predicate `self == other`
714    pub fn eq(self, other: impl Into<Self>) -> Predicate {
715        Predicate::eq(self, other)
716    }
717
718    /// Create a new predicate `self != other`
719    pub fn ne(self, other: impl Into<Self>) -> Predicate {
720        Predicate::ne(self, other)
721    }
722
723    /// Create a new predicate `self <= other`
724    pub fn le(self, other: impl Into<Self>) -> Predicate {
725        Predicate::le(self, other)
726    }
727
728    /// Create a new predicate `self < other`
729    pub fn lt(self, other: impl Into<Self>) -> Predicate {
730        Predicate::lt(self, other)
731    }
732
733    /// Create a new predicate `self >= other`
734    pub fn ge(self, other: impl Into<Self>) -> Predicate {
735        Predicate::ge(self, other)
736    }
737
738    /// Create a new predicate `self > other`
739    pub fn gt(self, other: impl Into<Self>) -> Predicate {
740        Predicate::gt(self, other)
741    }
742
743    /// Create a new predicate `DISTINCT(self, other)`
744    pub fn distinct(self, other: impl Into<Self>) -> Predicate {
745        Predicate::distinct(self, other)
746    }
747
748    /// Creates a new unary expression
749    pub fn unary(op: UnaryExpressionOp, expr: impl Into<Expression>) -> Self {
750        Self::Unary(UnaryExpression::new(op, expr))
751    }
752
753    /// Creates a new binary expression lhs OP rhs
754    pub fn binary(
755        op: BinaryExpressionOp,
756        lhs: impl Into<Expression>,
757        rhs: impl Into<Expression>,
758    ) -> Self {
759        Self::Binary(BinaryExpression::new(op, lhs, rhs))
760    }
761
762    /// Creates a new variadic expression
763    pub fn variadic(
764        op: VariadicExpressionOp,
765        exprs: impl IntoIterator<Item = impl Into<Expression>>,
766    ) -> Self {
767        Self::Variadic(VariadicExpression::new(op, exprs))
768    }
769
770    /// Creates a new COALESCE expression that returns the first non-null value.
771    ///
772    /// COALESCE evaluates expressions in order and returns the first non-null result.
773    /// If all expressions evaluate to null, the result is null.
774    pub fn coalesce(exprs: impl IntoIterator<Item = impl Into<Expression>>) -> Self {
775        Self::variadic(VariadicExpressionOp::Coalesce, exprs)
776    }
777
778    /// Creates a new opaque expression
779    pub fn opaque(
780        op: impl OpaqueExpressionOp,
781        exprs: impl IntoIterator<Item = Expression>,
782    ) -> Self {
783        Self::Opaque(OpaqueExpression::new(Arc::new(op), exprs))
784    }
785
786    /// Creates a new unknown expression
787    pub fn unknown(name: impl Into<String>) -> Self {
788        Self::Unknown(name.into())
789    }
790
791    /// Creates a new ParseJson expression that parses a JSON string column into a struct.
792    /// This is the inverse of `ToJson` - it converts a JSON-encoded string into a struct.
793    pub fn parse_json(json_expr: impl Into<Expression>, output_schema: SchemaRef) -> Self {
794        Self::ParseJson(ParseJsonExpression::new(json_expr, output_schema))
795    }
796
797    /// Extracts keys from a `Map<String, String>` and parses values into a typed struct using
798    /// Delta's partition value serialization rules. The output struct schema is determined by the
799    /// evaluator's `result_type`.
800    pub fn map_to_struct(map_expr: impl Into<Expression>) -> Self {
801        Self::MapToStruct(MapToStructExpression::new(map_expr))
802    }
803}
804
805impl Predicate {
806    /// Returns a set of columns referenced by this predicate.
807    pub fn references(&self) -> HashSet<&ColumnName> {
808        let mut references = GetColumnReferences::default();
809        let _ = references.transform_pred(self);
810        references.0
811    }
812
813    /// Creates a new boolean column reference. See also [`Expression::column`].
814    pub fn column<A>(field_names: impl IntoIterator<Item = A>) -> Predicate
815    where
816        ColumnName: FromIterator<A>,
817    {
818        Self::from_expr(ColumnName::new(field_names))
819    }
820
821    /// Create a new literal boolean value
822    pub const fn literal(value: bool) -> Self {
823        Self::BooleanExpression(Expression::Literal(Scalar::Boolean(value)))
824    }
825
826    /// Creates a NULL literal boolean value
827    pub const fn null_literal() -> Self {
828        Self::BooleanExpression(Expression::Literal(Scalar::Null(DataType::BOOLEAN)))
829    }
830
831    /// Converts a boolean-valued expression into a predicate
832    pub fn from_expr(expr: impl Into<Expression>) -> Self {
833        match expr.into() {
834            Expression::Predicate(p) => *p,
835            expr => Predicate::BooleanExpression(expr),
836        }
837    }
838
839    /// Logical NOT (boolean inversion)
840    pub fn not(pred: impl Into<Self>) -> Self {
841        Self::Not(Box::new(pred.into()))
842    }
843
844    /// Create a new predicate `self IS NULL`
845    pub fn is_null(expr: impl Into<Expression>) -> Predicate {
846        Self::unary(UnaryPredicateOp::IsNull, expr)
847    }
848
849    /// Create a new predicate `self IS NOT NULL`
850    pub fn is_not_null(expr: impl Into<Expression>) -> Predicate {
851        Self::not(Self::is_null(expr))
852    }
853
854    /// Create a new predicate `self == other`
855    pub fn eq(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
856        Self::binary(BinaryPredicateOp::Equal, a, b)
857    }
858
859    /// Create a new predicate `self != other`
860    pub fn ne(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
861        Self::not(Self::binary(BinaryPredicateOp::Equal, a, b))
862    }
863
864    /// Create a new predicate `self <= other`
865    pub fn le(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
866        Self::not(Self::binary(BinaryPredicateOp::GreaterThan, a, b))
867    }
868
869    /// Create a new predicate `self < other`
870    pub fn lt(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
871        Self::binary(BinaryPredicateOp::LessThan, a, b)
872    }
873
874    /// Create a new predicate `self >= other`
875    pub fn ge(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
876        Self::not(Self::binary(BinaryPredicateOp::LessThan, a, b))
877    }
878
879    /// Create a new predicate `self > other`
880    pub fn gt(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
881        Self::binary(BinaryPredicateOp::GreaterThan, a, b)
882    }
883
884    /// Create a new predicate `DISTINCT(self, other)`
885    pub fn distinct(a: impl Into<Expression>, b: impl Into<Expression>) -> Self {
886        Self::binary(BinaryPredicateOp::Distinct, a, b)
887    }
888
889    /// Create a new predicate `self AND other`
890    pub fn and(a: impl Into<Self>, b: impl Into<Self>) -> Self {
891        Self::and_from([a.into(), b.into()])
892    }
893
894    /// Create a new predicate `self OR other`
895    pub fn or(a: impl Into<Self>, b: impl Into<Self>) -> Self {
896        Self::or_from([a.into(), b.into()])
897    }
898
899    /// Creates a new predicate AND(preds...). See [`Self::junction`] for normalization of
900    /// empty and single-element inputs.
901    pub fn and_from(preds: impl IntoIterator<Item = Self>) -> Self {
902        Self::junction(JunctionPredicateOp::And, preds)
903    }
904
905    /// Creates a new predicate OR(preds...). See [`Self::junction`] for normalization of
906    /// empty and single-element inputs.
907    pub fn or_from(preds: impl IntoIterator<Item = Self>) -> Self {
908        Self::junction(JunctionPredicateOp::Or, preds)
909    }
910
911    /// Creates a new unary predicate OP expr
912    pub fn unary(op: UnaryPredicateOp, expr: impl Into<Expression>) -> Self {
913        let expr = Box::new(expr.into());
914        Self::Unary(UnaryPredicate { op, expr })
915    }
916
917    /// Creates a new binary predicate lhs OP rhs
918    pub fn binary(
919        op: BinaryPredicateOp,
920        lhs: impl Into<Expression>,
921        rhs: impl Into<Expression>,
922    ) -> Self {
923        Self::Binary(BinaryPredicate {
924            op,
925            left: Box::new(lhs.into()),
926            right: Box::new(rhs.into()),
927        })
928    }
929
930    /// Creates a new junction predicate OP(preds...). Normalizes degenerate cases:
931    ///
932    /// - Empty junction returns the identity element (the value that has no effect when
933    ///   combined with other predicates under the same operator):
934    ///   - `AND()` -> `true`, because `true AND p` == `p` for any predicate `p`.
935    ///   - `OR()` -> `false`, because `false OR p` == `p` for any predicate `p`.
936    /// - Single-element junction unwraps the element: `AND(p)` / `OR(p)` -> `p`.
937    pub fn junction(op: JunctionPredicateOp, preds: impl IntoIterator<Item = Self>) -> Self {
938        let mut preds: Vec<_> = preds.into_iter().collect();
939        match preds.len() {
940            0 => match op {
941                JunctionPredicateOp::And => Self::literal(true),
942                JunctionPredicateOp::Or => Self::literal(false),
943            },
944            // A junction of one predicate is just that predicate.
945            1 => preds.remove(0),
946            _ => Self::Junction(JunctionPredicate { op, preds }),
947        }
948    }
949
950    /// Creates a new opaque predicate
951    pub fn opaque(op: impl OpaquePredicateOp, exprs: impl IntoIterator<Item = Expression>) -> Self {
952        Self::Opaque(OpaquePredicate::new(Arc::new(op), exprs))
953    }
954
955    /// Creates a new unknown predicate
956    pub fn unknown(name: impl Into<String>) -> Self {
957        Self::Unknown(name.into())
958    }
959}
960
961////////////////////////////////////////////////////////////////////////
962// Trait impls
963////////////////////////////////////////////////////////////////////////
964
965impl PartialEq for OpaquePredicate {
966    fn eq(&self, other: &Self) -> bool {
967        self.op.dyn_eq(other.op.any_ref()) && self.exprs == other.exprs
968    }
969}
970
971impl PartialEq for OpaqueExpression {
972    fn eq(&self, other: &Self) -> bool {
973        self.op.dyn_eq(other.op.any_ref()) && self.exprs == other.exprs
974    }
975}
976
977impl Display for UnaryExpressionOp {
978    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
979        use UnaryExpressionOp::*;
980        match self {
981            ToJson => write!(f, "TO_JSON"),
982        }
983    }
984}
985
986impl Display for BinaryExpressionOp {
987    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
988        use BinaryExpressionOp::*;
989        match self {
990            Plus => write!(f, "+"),
991            Minus => write!(f, "-"),
992            Multiply => write!(f, "*"),
993            Divide => write!(f, "/"),
994        }
995    }
996}
997
998impl Display for VariadicExpressionOp {
999    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1000        use VariadicExpressionOp::*;
1001        match self {
1002            Coalesce => write!(f, "COALESCE"),
1003        }
1004    }
1005}
1006
1007impl Display for BinaryPredicateOp {
1008    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1009        use BinaryPredicateOp::*;
1010        match self {
1011            LessThan => write!(f, "<"),
1012            GreaterThan => write!(f, ">"),
1013            Equal => write!(f, "="),
1014            // TODO(roeap): AFAIK DISTINCT does not have a commonly used operator symbol
1015            // so ideally this would not be used as we use Display for rendering expressions
1016            // in our code we take care of this, but theirs might not ...
1017            Distinct => write!(f, "DISTINCT"),
1018            In => write!(f, "IN"),
1019        }
1020    }
1021}
1022
1023// Helper for displaying the children of variadic expressions and predicates
1024fn format_child_list<T: Display>(children: &[T]) -> String {
1025    children.iter().map(|c| format!("{c}")).join(", ")
1026}
1027
1028impl Display for Expression {
1029    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1030        use Expression::*;
1031        match self {
1032            Literal(l) => write!(f, "{l}"),
1033            Column(name) => write!(f, "Column({name})"),
1034            Predicate(p) => write!(f, "{p}"),
1035            Struct(exprs, _) => write!(f, "Struct({})", format_child_list(exprs)),
1036            Transform(transform) => {
1037                write!(f, "Transform(")?;
1038                let mut sep = "";
1039                if !transform.prepended_fields.is_empty() {
1040                    let prepended_fields = format_child_list(&transform.prepended_fields);
1041                    write!(f, "prepend [{prepended_fields}]")?;
1042                    sep = ", ";
1043                }
1044                for (field_name, field_transform) in &transform.field_transforms {
1045                    let insertions = &field_transform.exprs;
1046                    if insertions.is_empty() {
1047                        if field_transform.is_replace {
1048                            write!(f, "{sep}drop {field_name}")?;
1049                        } else {
1050                            continue; // no-op; ignore it and don't change `sep` below
1051                        }
1052                    } else {
1053                        let insertions = format_child_list(insertions);
1054                        if field_transform.is_replace {
1055                            write!(f, "{sep}replace {field_name} with [{insertions}]")?;
1056                        } else {
1057                            write!(f, "{sep}after {field_name} insert [{insertions}]")?;
1058                        }
1059                    }
1060                    sep = ", ";
1061                }
1062                write!(f, ")")
1063            }
1064            Unary(UnaryExpression { op, expr }) => write!(f, "{op}({expr})"),
1065            Binary(BinaryExpression { op, left, right }) => write!(f, "{left} {op} {right}"),
1066            Variadic(VariadicExpression { op, exprs }) => {
1067                write!(f, "{op}({})", format_child_list(exprs))
1068            }
1069            Opaque(OpaqueExpression { op, exprs }) => {
1070                write!(f, "{op:?}({})", format_child_list(exprs))
1071            }
1072            Unknown(name) => write!(f, "<unknown: {name}>"),
1073            ParseJson(p) => {
1074                write!(
1075                    f,
1076                    "PARSE_JSON({}, <schema:{} fields>)",
1077                    p.json_expr,
1078                    p.output_schema.fields().len()
1079                )
1080            }
1081            MapToStruct(m) => write!(f, "MAP_TO_STRUCT({})", m.map_expr),
1082        }
1083    }
1084}
1085
1086impl Display for Predicate {
1087    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1088        use Predicate::*;
1089        match self {
1090            BooleanExpression(expr) => write!(f, "{expr}"),
1091            Not(pred) => write!(f, "NOT({pred})"),
1092            Binary(BinaryPredicate {
1093                op: BinaryPredicateOp::Distinct,
1094                left,
1095                right,
1096            }) => write!(f, "DISTINCT({left}, {right})"),
1097            Binary(BinaryPredicate { op, left, right }) => write!(f, "{left} {op} {right}"),
1098            Unary(UnaryPredicate { op, expr }) => match op {
1099                UnaryPredicateOp::IsNull => write!(f, "{expr} IS NULL"),
1100            },
1101            Junction(JunctionPredicate { op, preds }) => {
1102                let op = match op {
1103                    JunctionPredicateOp::And => "AND",
1104                    JunctionPredicateOp::Or => "OR",
1105                };
1106                write!(f, "{op}({})", format_child_list(preds))
1107            }
1108            Opaque(OpaquePredicate { op, exprs }) => {
1109                write!(f, "{op:?}({})", format_child_list(exprs))
1110            }
1111            Unknown(name) => write!(f, "<unknown: {name}>"),
1112        }
1113    }
1114}
1115
1116impl From<Scalar> for Expression {
1117    fn from(value: Scalar) -> Self {
1118        Self::literal(value)
1119    }
1120}
1121
1122impl From<ColumnName> for Expression {
1123    fn from(value: ColumnName) -> Self {
1124        Self::Column(value)
1125    }
1126}
1127
1128impl From<Predicate> for Expression {
1129    fn from(value: Predicate) -> Self {
1130        Self::from_pred(value)
1131    }
1132}
1133
1134impl From<ColumnName> for Predicate {
1135    fn from(value: ColumnName) -> Self {
1136        Self::from_expr(value)
1137    }
1138}
1139
1140impl<R: Into<Expression>> std::ops::Add<R> for Expression {
1141    type Output = Self;
1142
1143    fn add(self, rhs: R) -> Self::Output {
1144        Self::binary(BinaryExpressionOp::Plus, self, rhs)
1145    }
1146}
1147
1148impl<R: Into<Expression>> std::ops::Sub<R> for Expression {
1149    type Output = Self;
1150
1151    fn sub(self, rhs: R) -> Self {
1152        Self::binary(BinaryExpressionOp::Minus, self, rhs)
1153    }
1154}
1155
1156impl<R: Into<Expression>> std::ops::Mul<R> for Expression {
1157    type Output = Self;
1158
1159    fn mul(self, rhs: R) -> Self {
1160        Self::binary(BinaryExpressionOp::Multiply, self, rhs)
1161    }
1162}
1163
1164impl<R: Into<Expression>> std::ops::Div<R> for Expression {
1165    type Output = Self;
1166
1167    fn div(self, rhs: R) -> Self {
1168        Self::binary(BinaryExpressionOp::Divide, self, rhs)
1169    }
1170}
1171
1172/// Retrieves the set of column names referenced by an expression.
1173#[derive(Default)]
1174struct GetColumnReferences<'a>(HashSet<&'a ColumnName>);
1175
1176impl<'a> ExpressionTransform<'a> for GetColumnReferences<'a> {
1177    fn transform_expr_column(&mut self, name: &'a ColumnName) -> Option<Cow<'a, ColumnName>> {
1178        self.0.insert(name);
1179        Some(Cow::Borrowed(name))
1180    }
1181}
1182
1183#[cfg(test)]
1184mod tests {
1185    use std::fmt::Debug;
1186
1187    use serde::de::DeserializeOwned;
1188    use serde::Serialize;
1189
1190    use super::{column_expr, column_pred, Expression as Expr, Predicate as Pred};
1191
1192    /// Helper function to verify roundtrip serialization/deserialization
1193    fn assert_roundtrip<T: Serialize + DeserializeOwned + PartialEq + Debug>(value: &T) {
1194        let json = serde_json::to_string(value).expect("serialization should succeed");
1195        let deserialized: T = serde_json::from_str(&json).expect("deserialization should succeed");
1196        assert_eq!(value, &deserialized, "roundtrip should preserve value");
1197    }
1198
1199    #[test]
1200    fn test_expression_format() {
1201        let cases = [
1202            (column_expr!("x"), "Column(x)"),
1203            (
1204                (column_expr!("x") + Expr::literal(4)) / Expr::literal(10) * Expr::literal(42),
1205                "Column(x) + 4 / 10 * 42",
1206            ),
1207            (
1208                Expr::struct_from([column_expr!("x"), Expr::literal(2), Expr::literal(10)]),
1209                "Struct(Column(x), 2, 10)",
1210            ),
1211        ];
1212
1213        for (expr, expected) in cases {
1214            let result = format!("{expr}");
1215            assert_eq!(result, expected);
1216        }
1217    }
1218
1219    #[test]
1220    fn test_predicate_format() {
1221        let cases = [
1222            (column_pred!("x"), "Column(x)"),
1223            (column_expr!("x").eq(Expr::literal(2)), "Column(x) = 2"),
1224            (
1225                (column_expr!("x") - Expr::literal(4)).lt(Expr::literal(10)),
1226                "Column(x) - 4 < 10",
1227            ),
1228            (
1229                Pred::and(
1230                    column_expr!("x").ge(Expr::literal(2)),
1231                    column_expr!("x").le(Expr::literal(10)),
1232                ),
1233                "AND(NOT(Column(x) < 2), NOT(Column(x) > 10))",
1234            ),
1235            (
1236                Pred::and_from([
1237                    column_expr!("x").ge(Expr::literal(2)),
1238                    column_expr!("x").le(Expr::literal(10)),
1239                    column_expr!("x").le(Expr::literal(100)),
1240                ]),
1241                "AND(NOT(Column(x) < 2), NOT(Column(x) > 10), NOT(Column(x) > 100))",
1242            ),
1243            (
1244                Pred::or(
1245                    column_expr!("x").gt(Expr::literal(2)),
1246                    column_expr!("x").lt(Expr::literal(10)),
1247                ),
1248                "OR(Column(x) > 2, Column(x) < 10)",
1249            ),
1250            (
1251                column_expr!("x").eq(Expr::literal("foo")),
1252                "Column(x) = 'foo'",
1253            ),
1254        ];
1255
1256        for (pred, expected) in cases {
1257            let result = format!("{pred}");
1258            assert_eq!(result, expected);
1259        }
1260    }
1261
1262    // ==================== Serde Roundtrip Tests ====================
1263
1264    mod serde_tests {
1265        use std::sync::Arc;
1266
1267        use crate::expressions::scalars::{ArrayData, DecimalData, MapData, StructData};
1268        use crate::expressions::{
1269            column_expr, column_name, BinaryExpressionOp, BinaryPredicateOp, ColumnName,
1270            Expression, Predicate, Scalar, Transform, UnaryExpressionOp,
1271        };
1272        use crate::schema::{ArrayType, DataType, DecimalType, MapType, StructField};
1273        use crate::utils::test_utils::assert_result_error_with_message;
1274
1275        use super::assert_roundtrip;
1276
1277        // ==================== Expression::Literal Tests ====================
1278
1279        #[test]
1280        fn test_literal_scalars_roundtrip() {
1281            // Test all primitive scalar types that have proper PartialEq
1282            let cases: Vec<Expression> = vec![
1283                // Numeric types
1284                Expression::literal(42i32),         // Integer
1285                Expression::literal(9999999999i64), // Long
1286                Expression::literal(123i16),        // Short
1287                Expression::literal(42i8),          // Byte
1288                Expression::literal(1.12345677_32), // Float
1289                Expression::literal(1.12345667_64), // Double
1290                // String and Boolean
1291                Expression::literal("hello world"),
1292                Expression::literal(true),
1293                Expression::literal(false),
1294                // Temporal types
1295                Expression::Literal(Scalar::Timestamp(1234567890000000)),
1296                Expression::Literal(Scalar::TimestampNtz(1234567890000000)),
1297                Expression::Literal(Scalar::Date(19000)),
1298                // Binary
1299                Expression::Literal(Scalar::Binary(vec![1, 2, 3, 4, 5])),
1300                // Decimal
1301                Expression::Literal(Scalar::Decimal(
1302                    DecimalData::try_new(12345i128, DecimalType::try_new(10, 2).unwrap()).unwrap(),
1303                )),
1304            ];
1305
1306            for expr in &cases {
1307                assert_roundtrip(expr);
1308            }
1309        }
1310
1311        #[test]
1312        fn test_literal_complex_scalars_roundtrip() {
1313            // Test complex scalar types that need JSON comparison (partial_cmp returns None)
1314            let cases: Vec<Expression> = vec![
1315                // Null with different types
1316                Expression::null_literal(DataType::INTEGER),
1317                Expression::null_literal(DataType::STRING),
1318                Expression::null_literal(DataType::BOOLEAN),
1319                // Array
1320                Expression::Literal(Scalar::Array(
1321                    ArrayData::try_new(
1322                        ArrayType::new(DataType::INTEGER, false),
1323                        vec![Scalar::Integer(1), Scalar::Integer(2), Scalar::Integer(3)],
1324                    )
1325                    .unwrap(),
1326                )),
1327                // Map
1328                Expression::Literal(Scalar::Map(
1329                    MapData::try_new(
1330                        MapType::new(DataType::STRING, DataType::INTEGER, false),
1331                        vec![
1332                            (Scalar::String("a".to_string()), Scalar::Integer(1)),
1333                            (Scalar::String("b".to_string()), Scalar::Integer(2)),
1334                        ],
1335                    )
1336                    .unwrap(),
1337                )),
1338                // Struct
1339                Expression::Literal(Scalar::Struct(
1340                    StructData::try_new(
1341                        vec![
1342                            StructField::nullable("x", DataType::INTEGER),
1343                            StructField::nullable("y", DataType::STRING),
1344                        ],
1345                        vec![Scalar::Integer(42), Scalar::String("hello".to_string())],
1346                    )
1347                    .unwrap(),
1348                )),
1349            ];
1350
1351            for expr in &cases {
1352                assert_roundtrip(expr);
1353            }
1354        }
1355
1356        // ==================== Expression::Column Tests ====================
1357
1358        #[test]
1359        fn test_column_expressions_roundtrip() {
1360            let cases: Vec<Expression> = vec![
1361                column_expr!("my_column"),
1362                Expression::column(["parent", "child"]),
1363                Expression::column(["a", "b", "c", "d"]),
1364            ];
1365
1366            for expr in &cases {
1367                assert_roundtrip(expr);
1368            }
1369        }
1370
1371        #[test]
1372        fn test_column_names_roundtrip() {
1373            let cases: Vec<ColumnName> = vec![
1374                column_name!("simple"),
1375                ColumnName::new(["a", "b", "c"]),
1376                ColumnName::new::<&str>([]),
1377            ];
1378
1379            for col in &cases {
1380                assert_roundtrip(col);
1381            }
1382        }
1383
1384        // ==================== Expression Operations Tests ====================
1385
1386        #[test]
1387        fn test_unary_expression_roundtrip() {
1388            let expr = Expression::unary(UnaryExpressionOp::ToJson, column_expr!("data"));
1389            assert_roundtrip(&expr);
1390        }
1391
1392        #[test]
1393        fn test_binary_expressions_roundtrip() {
1394            let ops = [
1395                BinaryExpressionOp::Plus,
1396                BinaryExpressionOp::Minus,
1397                BinaryExpressionOp::Multiply,
1398                BinaryExpressionOp::Divide,
1399            ];
1400
1401            for op in ops {
1402                let expr = Expression::binary(op, column_expr!("a"), Expression::literal(10));
1403                assert_roundtrip(&expr);
1404            }
1405        }
1406
1407        #[test]
1408        fn test_variadic_expression_roundtrip() {
1409            let expr = Expression::coalesce([
1410                column_expr!("a"),
1411                column_expr!("b"),
1412                Expression::literal("default"),
1413            ]);
1414            assert_roundtrip(&expr);
1415        }
1416
1417        #[test]
1418        fn test_nested_arithmetic_expression_roundtrip() {
1419            // (a + b) * (c - d) / 2
1420            let left = Expression::binary(
1421                BinaryExpressionOp::Plus,
1422                column_expr!("a"),
1423                column_expr!("b"),
1424            );
1425            let right = Expression::binary(
1426                BinaryExpressionOp::Minus,
1427                column_expr!("c"),
1428                column_expr!("d"),
1429            );
1430            let mul = Expression::binary(BinaryExpressionOp::Multiply, left, right);
1431            let expr = Expression::binary(BinaryExpressionOp::Divide, mul, Expression::literal(2));
1432            assert_roundtrip(&expr);
1433        }
1434
1435        // ==================== Expression::Struct/Transform/Other Tests ====================
1436
1437        #[test]
1438        fn test_struct_expression_roundtrip() {
1439            let expr = Expression::struct_from([
1440                Arc::new(column_expr!("x")),
1441                Arc::new(Expression::literal(42)),
1442                Arc::new(Expression::literal("hello")),
1443            ]);
1444            assert_roundtrip(&expr);
1445        }
1446
1447        #[test]
1448        fn test_transform_expressions_roundtrip() {
1449            let cases: Vec<Expression> = vec![
1450                // Identity transform
1451                Expression::transform(Transform::new_top_level()),
1452                // Drop field
1453                Expression::transform(Transform::new_top_level().with_dropped_field("old_column")),
1454                // Replace field
1455                Expression::transform(
1456                    Transform::new_top_level()
1457                        .with_replaced_field("original", Arc::new(Expression::literal(0))),
1458                ),
1459                // Insert fields
1460                Expression::transform(
1461                    Transform::new_top_level()
1462                        .with_inserted_field(Some("after_col"), Arc::new(column_expr!("new_col")))
1463                        .with_inserted_field(
1464                            None::<String>,
1465                            Arc::new(Expression::literal("prepended")),
1466                        ),
1467                ),
1468                // Nested transform
1469                Expression::transform(
1470                    Transform::new_nested(["parent", "child"]).with_dropped_field("to_drop"),
1471                ),
1472            ];
1473
1474            for expr in &cases {
1475                assert_roundtrip(expr);
1476            }
1477        }
1478
1479        #[test]
1480        fn test_expression_wrapping_predicate_roundtrip() {
1481            let pred = Predicate::eq(column_expr!("x"), Expression::literal(10));
1482            let expr = Expression::from_pred(pred);
1483            assert_roundtrip(&expr);
1484        }
1485
1486        #[test]
1487        fn test_expression_unknown_roundtrip() {
1488            let expr = Expression::unknown("some_unknown_function()");
1489            assert_roundtrip(&expr);
1490        }
1491
1492        #[test]
1493        fn test_map_to_struct_expression_roundtrip() {
1494            let cases: Vec<Expression> = vec![
1495                Expression::map_to_struct(column_expr!("pv")),
1496                Expression::map_to_struct(Expression::literal("ignored")),
1497            ];
1498
1499            for expr in &cases {
1500                assert_roundtrip(expr);
1501            }
1502        }
1503
1504        // ==================== Predicate Tests ====================
1505
1506        #[test]
1507        fn test_predicate_basics_roundtrip() {
1508            let cases: Vec<Predicate> = vec![
1509                // Boolean expression
1510                Predicate::from_expr(column_expr!("is_active")),
1511                // Literals
1512                Predicate::literal(true),
1513                Predicate::literal(false),
1514                // NOT
1515                Predicate::not(Predicate::from_expr(column_expr!("x"))),
1516                // Nested NOT
1517                Predicate::not(Predicate::not(Predicate::gt(
1518                    column_expr!("x"),
1519                    Expression::literal(5),
1520                ))),
1521                // Unknown
1522                Predicate::unknown("some_unknown_predicate()"),
1523                // Unary predicates
1524                Predicate::is_null(column_expr!("nullable_col")),
1525                Predicate::is_not_null(column_expr!("nullable_col")),
1526            ];
1527
1528            for pred in &cases {
1529                assert_roundtrip(pred);
1530            }
1531        }
1532
1533        #[test]
1534        fn test_predicate_null_literal_roundtrip() {
1535            let pred = Predicate::null_literal();
1536            assert_roundtrip(&pred);
1537        }
1538
1539        #[test]
1540        fn test_predicate_comparisons_roundtrip() {
1541            let cases: Vec<Predicate> = vec![
1542                Predicate::eq(column_expr!("x"), Expression::literal(42)),
1543                Predicate::ne(column_expr!("status"), Expression::literal("active")),
1544                Predicate::lt(column_expr!("age"), Expression::literal(18)),
1545                Predicate::le(column_expr!("price"), Expression::literal(100)),
1546                Predicate::gt(column_expr!("score"), Expression::literal(90)),
1547                Predicate::ge(column_expr!("quantity"), Expression::literal(1)),
1548                Predicate::distinct(column_expr!("a"), column_expr!("b")),
1549            ];
1550
1551            for pred in &cases {
1552                assert_roundtrip(pred);
1553            }
1554        }
1555
1556        #[test]
1557        fn test_predicate_in_roundtrip() {
1558            let array_data = ArrayData::try_new(
1559                ArrayType::new(DataType::INTEGER, false),
1560                vec![Scalar::Integer(1), Scalar::Integer(2), Scalar::Integer(3)],
1561            )
1562            .unwrap();
1563            let pred = Predicate::binary(
1564                BinaryPredicateOp::In,
1565                column_expr!("x"),
1566                Expression::Literal(Scalar::Array(array_data)),
1567            );
1568            assert_roundtrip(&pred);
1569        }
1570
1571        #[test]
1572        fn test_predicate_junctions_roundtrip() {
1573            let cases: Vec<Predicate> = vec![
1574                // Simple AND
1575                Predicate::and(
1576                    Predicate::gt(column_expr!("x"), Expression::literal(0)),
1577                    Predicate::lt(column_expr!("x"), Expression::literal(100)),
1578                ),
1579                // Simple OR
1580                Predicate::or(
1581                    Predicate::eq(column_expr!("status"), Expression::literal("active")),
1582                    Predicate::eq(column_expr!("status"), Expression::literal("pending")),
1583                ),
1584                // Multiple AND
1585                Predicate::and_from([
1586                    Predicate::gt(column_expr!("x"), Expression::literal(0)),
1587                    Predicate::lt(column_expr!("x"), Expression::literal(100)),
1588                    Predicate::is_not_null(column_expr!("x")),
1589                ]),
1590                // Multiple OR
1591                Predicate::or_from([
1592                    Predicate::eq(column_expr!("type"), Expression::literal("A")),
1593                    Predicate::eq(column_expr!("type"), Expression::literal("B")),
1594                    Predicate::eq(column_expr!("type"), Expression::literal("C")),
1595                ]),
1596                // Nested: (a > 0 AND b < 100) OR (c = 'special')
1597                Predicate::or(
1598                    Predicate::and(
1599                        Predicate::gt(column_expr!("a"), Expression::literal(0)),
1600                        Predicate::lt(column_expr!("b"), Expression::literal(100)),
1601                    ),
1602                    Predicate::eq(column_expr!("c"), Expression::literal("special")),
1603                ),
1604            ];
1605
1606            for pred in &cases {
1607                assert_roundtrip(pred);
1608            }
1609        }
1610
1611        // ==================== Complex Nested Structures ====================
1612
1613        #[test]
1614        fn test_deeply_nested_structures_roundtrip() {
1615            // COALESCE(a + b, c * d, 0) > 100
1616            let add = Expression::binary(
1617                BinaryExpressionOp::Plus,
1618                column_expr!("a"),
1619                column_expr!("b"),
1620            );
1621            let mul = Expression::binary(
1622                BinaryExpressionOp::Multiply,
1623                column_expr!("c"),
1624                column_expr!("d"),
1625            );
1626            let coalesce = Expression::coalesce([add, mul, Expression::literal(0)]);
1627            let pred = Predicate::gt(coalesce, Expression::literal(100));
1628            assert_roundtrip(&pred);
1629
1630            // Expression wrapping a predicate that references expressions
1631            let inner_pred = Predicate::and(
1632                Predicate::eq(column_expr!("x"), Expression::literal(1)),
1633                Predicate::gt(
1634                    Expression::binary(
1635                        BinaryExpressionOp::Plus,
1636                        column_expr!("y"),
1637                        column_expr!("z"),
1638                    ),
1639                    Expression::literal(10),
1640                ),
1641            );
1642            let expr = Expression::from_pred(inner_pred);
1643            assert_roundtrip(&expr);
1644        }
1645
1646        // ==================== Opaque Variant Failure Tests ====================
1647
1648        #[test]
1649        fn test_opaque_expression_serialize_fails() {
1650            use crate::expressions::{OpaqueExpressionOp, ScalarExpressionEvaluator};
1651            use crate::DeltaResult;
1652
1653            #[derive(Debug, PartialEq)]
1654            struct TestOpaqueExprOp;
1655
1656            impl OpaqueExpressionOp for TestOpaqueExprOp {
1657                fn name(&self) -> &str {
1658                    "test_opaque"
1659                }
1660                fn eval_expr_scalar(
1661                    &self,
1662                    _eval_expr: &ScalarExpressionEvaluator<'_>,
1663                    _exprs: &[Expression],
1664                ) -> DeltaResult<Scalar> {
1665                    Ok(Scalar::Integer(0))
1666                }
1667            }
1668
1669            let expr = Expression::opaque(TestOpaqueExprOp, [Expression::literal(1)]);
1670            let result = serde_json::to_string(&expr);
1671            assert_result_error_with_message(result, "Cannot serialize an Opaque Expression");
1672        }
1673
1674        #[test]
1675        fn test_opaque_predicate_serialize_fails() {
1676            use crate::expressions::{OpaquePredicateOp, ScalarExpressionEvaluator};
1677            use crate::kernel_predicates::{
1678                DirectDataSkippingPredicateEvaluator, DirectPredicateEvaluator,
1679                IndirectDataSkippingPredicateEvaluator,
1680            };
1681            use crate::DeltaResult;
1682
1683            #[derive(Debug, PartialEq)]
1684            struct TestOpaquePredOp;
1685
1686            impl OpaquePredicateOp for TestOpaquePredOp {
1687                fn name(&self) -> &str {
1688                    "test_opaque_pred"
1689                }
1690                fn eval_pred_scalar(
1691                    &self,
1692                    _eval_expr: &ScalarExpressionEvaluator<'_>,
1693                    _eval_pred: &DirectPredicateEvaluator<'_>,
1694                    _exprs: &[Expression],
1695                    _inverted: bool,
1696                ) -> DeltaResult<Option<bool>> {
1697                    Ok(Some(true))
1698                }
1699                fn eval_as_data_skipping_predicate(
1700                    &self,
1701                    _evaluator: &DirectDataSkippingPredicateEvaluator<'_>,
1702                    _exprs: &[Expression],
1703                    _inverted: bool,
1704                ) -> Option<bool> {
1705                    Some(true)
1706                }
1707                fn as_data_skipping_predicate(
1708                    &self,
1709                    _evaluator: &IndirectDataSkippingPredicateEvaluator<'_>,
1710                    _exprs: &[Expression],
1711                    _inverted: bool,
1712                ) -> Option<Predicate> {
1713                    None
1714                }
1715            }
1716
1717            let pred = Predicate::opaque(TestOpaquePredOp, [Expression::literal(1)]);
1718            let result = serde_json::to_string(&pred);
1719            assert_result_error_with_message(result, "Cannot serialize an Opaque Predicate");
1720        }
1721    }
1722
1723    #[test]
1724    fn single_element_and_from_returns_unwrapped_predicate() {
1725        let inner = Pred::gt(column_expr!("x"), Expr::literal(0));
1726        let result = Pred::and_from([inner.clone()]);
1727        assert_eq!(result, inner);
1728    }
1729
1730    #[test]
1731    fn single_element_or_from_returns_unwrapped_predicate() {
1732        let inner = Pred::gt(column_expr!("x"), Expr::literal(0));
1733        let result = Pred::or_from([inner.clone()]);
1734        assert_eq!(result, inner);
1735    }
1736
1737    #[test]
1738    fn multi_element_and_from_returns_junction() {
1739        let p1 = Pred::gt(column_expr!("x"), Expr::literal(0));
1740        let p2 = Pred::lt(column_expr!("x"), Expr::literal(100));
1741        let result = Pred::and_from([p1.clone(), p2.clone()]);
1742        assert!(matches!(result, Pred::Junction(ref j) if j.preds.len() == 2));
1743        assert_eq!(result, Pred::and(p1, p2));
1744    }
1745
1746    #[test]
1747    fn empty_and_from_returns_identity_literal() {
1748        let result = Pred::and_from(std::iter::empty());
1749        assert_eq!(result, Pred::literal(true));
1750    }
1751
1752    #[test]
1753    fn empty_or_from_returns_identity_literal() {
1754        let result = Pred::or_from(std::iter::empty());
1755        assert_eq!(result, Pred::literal(false));
1756    }
1757}
buoyant_kernel/expressions/mod.rs

buoyant_kernel/expressions/
mod.rs
