Skip to main content

datafusion_expr/
expr_fn.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Functions for creating logical expressions
19
20use crate::expr::{
21    AggregateFunction, BinaryExpr, Cast, Exists, GroupingSet, InList, InSubquery, Lambda,
22    LambdaVariable, NullTreatment, Placeholder, TryCast, Unnest, WildcardOptions,
23    WindowFunction,
24};
25use crate::function::{
26    AccumulatorArgs, AccumulatorFactoryFunction, PartitionEvaluatorFactory,
27    StateFieldsArgs,
28};
29use crate::ptr_eq::PtrEq;
30use crate::select_expr::SelectExpr;
31use crate::{
32    AggregateUDF, Expr, LimitEffect, LogicalPlan, Operator, PartitionEvaluator,
33    ScalarFunctionArgs, ScalarFunctionImplementation, ScalarUDF, Signature, Volatility,
34    conditional_expressions::CaseBuilder, expr::Sort, logical_plan::Subquery,
35};
36use crate::{
37    AggregateUDFImpl, ColumnarValue, ScalarUDFImpl, WindowFrame, WindowUDF, WindowUDFImpl,
38};
39use arrow::compute::kernels::cast_utils::{
40    parse_interval_day_time, parse_interval_month_day_nano, parse_interval_year_month,
41};
42use arrow::datatypes::{DataType, Field, FieldRef};
43use datafusion_common::{Column, Result, ScalarValue, Spans, TableReference, plan_err};
44use datafusion_functions_window_common::field::WindowUDFFieldArgs;
45use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
46use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
47use std::collections::HashMap;
48use std::fmt::Debug;
49use std::hash::Hash;
50use std::ops::Not;
51use std::sync::Arc;
52
53/// Create a column expression based on a qualified or unqualified column name. Will
54/// normalize unquoted identifiers according to SQL rules (identifiers will become lowercase).
55///
56/// For example:
57///
58/// ```rust
59/// # use datafusion_expr::col;
60/// let c1 = col("a");
61/// let c2 = col("A");
62/// assert_eq!(c1, c2);
63///
64/// // note how quoting with double quotes preserves the case
65/// let c3 = col(r#""A""#);
66/// assert_ne!(c1, c3);
67/// ```
68pub fn col(ident: impl Into<Column>) -> Expr {
69    Expr::Column(ident.into())
70}
71
72/// Create an out reference column which hold a reference that has been resolved to a field
73/// outside of the current plan.
74/// The expression created by this function does not preserve the metadata of the outer column.
75/// Please use `out_ref_col_with_metadata` if you want to preserve the metadata.
76pub fn out_ref_col(dt: DataType, ident: impl Into<Column>) -> Expr {
77    out_ref_col_with_metadata(dt, HashMap::new(), ident)
78}
79
80/// Create an out reference column from an existing field (preserving metadata)
81pub fn out_ref_col_with_metadata(
82    dt: DataType,
83    metadata: HashMap<String, String>,
84    ident: impl Into<Column>,
85) -> Expr {
86    let column = ident.into();
87    let field: FieldRef =
88        Arc::new(Field::new(column.name(), dt, true).with_metadata(metadata));
89    Expr::OuterReferenceColumn(field, column)
90}
91
92/// Create an unqualified column expression from the provided name, without normalizing
93/// the column.
94///
95/// For example:
96///
97/// ```rust
98/// # use datafusion_expr::{col, ident};
99/// let c1 = ident("A"); // not normalized staying as column 'A'
100/// let c2 = col("A"); // normalized via SQL rules becoming column 'a'
101/// assert_ne!(c1, c2);
102///
103/// let c3 = col(r#""A""#);
104/// assert_eq!(c1, c3);
105///
106/// let c4 = col("t1.a"); // parses as relation 't1' column 'a'
107/// let c5 = ident("t1.a"); // parses as column 't1.a'
108/// assert_ne!(c4, c5);
109/// ```
110pub fn ident(name: impl Into<String>) -> Expr {
111    Expr::Column(Column::from_name(name))
112}
113
114/// Create placeholder value that will be filled in (such as `$1`)
115///
116/// Note the parameter type can be inferred using [`Expr::infer_placeholder_types`]
117///
118/// # Example
119///
120/// ```rust
121/// # use datafusion_expr::{placeholder};
122/// let p = placeholder("$1"); // $1, refers to parameter 1
123/// assert_eq!(p.to_string(), "$1")
124/// ```
125pub fn placeholder(id: impl Into<String>) -> Expr {
126    Expr::Placeholder(Placeholder {
127        id: id.into(),
128        field: None,
129    })
130}
131
132/// Create an '*' [`Expr::Wildcard`] expression that matches all columns
133///
134/// # Example
135///
136/// ```rust
137/// # use datafusion_expr::{wildcard};
138/// let p = wildcard();
139/// assert_eq!(p.to_string(), "*")
140/// ```
141pub fn wildcard() -> SelectExpr {
142    SelectExpr::Wildcard(WildcardOptions::default())
143}
144
145/// Create an '*' [`Expr::Wildcard`] expression with the wildcard options
146pub fn wildcard_with_options(options: WildcardOptions) -> SelectExpr {
147    SelectExpr::Wildcard(options)
148}
149
150/// Create an 't.*' [`Expr::Wildcard`] expression that matches all columns from a specific table
151///
152/// # Example
153///
154/// ```rust
155/// # use datafusion_common::TableReference;
156/// # use datafusion_expr::{qualified_wildcard};
157/// let p = qualified_wildcard(TableReference::bare("t"));
158/// assert_eq!(p.to_string(), "t.*")
159/// ```
160pub fn qualified_wildcard(qualifier: impl Into<TableReference>) -> SelectExpr {
161    SelectExpr::QualifiedWildcard(qualifier.into(), WildcardOptions::default())
162}
163
164/// Create an 't.*' [`Expr::Wildcard`] expression with the wildcard options
165pub fn qualified_wildcard_with_options(
166    qualifier: impl Into<TableReference>,
167    options: WildcardOptions,
168) -> SelectExpr {
169    SelectExpr::QualifiedWildcard(qualifier.into(), options)
170}
171
172/// Return a new expression `left <op> right`
173pub fn binary_expr(left: Expr, op: Operator, right: Expr) -> Expr {
174    Expr::BinaryExpr(BinaryExpr::new(Box::new(left), op, Box::new(right)))
175}
176
177/// Return a new expression with a logical AND
178pub fn and(left: Expr, right: Expr) -> Expr {
179    Expr::BinaryExpr(BinaryExpr::new(
180        Box::new(left),
181        Operator::And,
182        Box::new(right),
183    ))
184}
185
186/// Return a new expression with a logical OR
187pub fn or(left: Expr, right: Expr) -> Expr {
188    Expr::BinaryExpr(BinaryExpr::new(
189        Box::new(left),
190        Operator::Or,
191        Box::new(right),
192    ))
193}
194
195/// Return a new expression with a logical NOT
196pub fn not(expr: Expr) -> Expr {
197    expr.not()
198}
199
200/// Return a new expression with bitwise AND
201pub fn bitwise_and(left: Expr, right: Expr) -> Expr {
202    Expr::BinaryExpr(BinaryExpr::new(
203        Box::new(left),
204        Operator::BitwiseAnd,
205        Box::new(right),
206    ))
207}
208
209/// Return a new expression with bitwise OR
210pub fn bitwise_or(left: Expr, right: Expr) -> Expr {
211    Expr::BinaryExpr(BinaryExpr::new(
212        Box::new(left),
213        Operator::BitwiseOr,
214        Box::new(right),
215    ))
216}
217
218/// Return a new expression with bitwise XOR
219pub fn bitwise_xor(left: Expr, right: Expr) -> Expr {
220    Expr::BinaryExpr(BinaryExpr::new(
221        Box::new(left),
222        Operator::BitwiseXor,
223        Box::new(right),
224    ))
225}
226
227/// Return a new expression with bitwise SHIFT RIGHT
228pub fn bitwise_shift_right(left: Expr, right: Expr) -> Expr {
229    Expr::BinaryExpr(BinaryExpr::new(
230        Box::new(left),
231        Operator::BitwiseShiftRight,
232        Box::new(right),
233    ))
234}
235
236/// Return a new expression with bitwise SHIFT LEFT
237pub fn bitwise_shift_left(left: Expr, right: Expr) -> Expr {
238    Expr::BinaryExpr(BinaryExpr::new(
239        Box::new(left),
240        Operator::BitwiseShiftLeft,
241        Box::new(right),
242    ))
243}
244
245/// Create an in_list expression
246pub fn in_list(expr: Expr, list: Vec<Expr>, negated: bool) -> Expr {
247    Expr::InList(InList::new(Box::new(expr), list, negated))
248}
249
250/// Create an EXISTS subquery expression
251pub fn exists(subquery: Arc<LogicalPlan>) -> Expr {
252    let outer_ref_columns = subquery.all_out_ref_exprs();
253    Expr::Exists(Exists {
254        subquery: Subquery {
255            subquery,
256            outer_ref_columns,
257            spans: Spans::new(),
258        },
259        negated: false,
260    })
261}
262
263/// Create a NOT EXISTS subquery expression
264pub fn not_exists(subquery: Arc<LogicalPlan>) -> Expr {
265    let outer_ref_columns = subquery.all_out_ref_exprs();
266    Expr::Exists(Exists {
267        subquery: Subquery {
268            subquery,
269            outer_ref_columns,
270            spans: Spans::new(),
271        },
272        negated: true,
273    })
274}
275
276/// Create an IN subquery expression
277pub fn in_subquery(expr: Expr, subquery: Arc<LogicalPlan>) -> Expr {
278    let outer_ref_columns = subquery.all_out_ref_exprs();
279    Expr::InSubquery(InSubquery::new(
280        Box::new(expr),
281        Subquery {
282            subquery,
283            outer_ref_columns,
284            spans: Spans::new(),
285        },
286        false,
287    ))
288}
289
290/// Create a NOT IN subquery expression
291pub fn not_in_subquery(expr: Expr, subquery: Arc<LogicalPlan>) -> Expr {
292    let outer_ref_columns = subquery.all_out_ref_exprs();
293    Expr::InSubquery(InSubquery::new(
294        Box::new(expr),
295        Subquery {
296            subquery,
297            outer_ref_columns,
298            spans: Spans::new(),
299        },
300        true,
301    ))
302}
303
304/// Create a scalar subquery expression
305pub fn scalar_subquery(subquery: Arc<LogicalPlan>) -> Expr {
306    let outer_ref_columns = subquery.all_out_ref_exprs();
307    Expr::ScalarSubquery(Subquery {
308        subquery,
309        outer_ref_columns,
310        spans: Spans::new(),
311    })
312}
313
314/// Create a grouping set
315pub fn grouping_set(exprs: Vec<Vec<Expr>>) -> Expr {
316    Expr::GroupingSet(GroupingSet::GroupingSets(exprs))
317}
318
319/// Create a grouping set for all combination of `exprs`
320pub fn cube(exprs: Vec<Expr>) -> Expr {
321    Expr::GroupingSet(GroupingSet::Cube(exprs))
322}
323
324/// Create a grouping set for rollup
325pub fn rollup(exprs: Vec<Expr>) -> Expr {
326    Expr::GroupingSet(GroupingSet::Rollup(exprs))
327}
328
329/// Create a cast expression
330pub fn cast(expr: Expr, data_type: DataType) -> Expr {
331    Expr::Cast(Cast::new(Box::new(expr), data_type))
332}
333
334/// Create a try cast expression
335pub fn try_cast(expr: Expr, data_type: DataType) -> Expr {
336    Expr::TryCast(TryCast::new(Box::new(expr), data_type))
337}
338
339/// Create is null expression
340pub fn is_null(expr: Expr) -> Expr {
341    Expr::IsNull(Box::new(expr))
342}
343
344/// Create is not null expression
345pub fn is_not_null(expr: Expr) -> Expr {
346    Expr::IsNotNull(Box::new(expr))
347}
348
349/// Create is true expression
350pub fn is_true(expr: Expr) -> Expr {
351    Expr::IsTrue(Box::new(expr))
352}
353
354/// Create is not true expression
355pub fn is_not_true(expr: Expr) -> Expr {
356    Expr::IsNotTrue(Box::new(expr))
357}
358
359/// Create is false expression
360pub fn is_false(expr: Expr) -> Expr {
361    Expr::IsFalse(Box::new(expr))
362}
363
364/// Create is not false expression
365pub fn is_not_false(expr: Expr) -> Expr {
366    Expr::IsNotFalse(Box::new(expr))
367}
368
369/// Create is unknown expression
370pub fn is_unknown(expr: Expr) -> Expr {
371    Expr::IsUnknown(Box::new(expr))
372}
373
374/// Create is not unknown expression
375pub fn is_not_unknown(expr: Expr) -> Expr {
376    Expr::IsNotUnknown(Box::new(expr))
377}
378
379/// Create a CASE WHEN statement with literal WHEN expressions for comparison to the base expression.
380pub fn case(expr: Expr) -> CaseBuilder {
381    CaseBuilder::new(Some(Box::new(expr)), vec![], vec![], None)
382}
383
384/// Create a CASE WHEN statement with boolean WHEN expressions and no base expression.
385pub fn when(when: Expr, then: Expr) -> CaseBuilder {
386    CaseBuilder::new(None, vec![when], vec![then], None)
387}
388
389/// Create a Unnest expression with default (non-outer) semantics.
390pub fn unnest(expr: Expr) -> Expr {
391    Expr::Unnest(Unnest {
392        expr: Box::new(expr),
393        outer: false,
394    })
395}
396
397/// Convenience method to create a new user defined scalar function (UDF) with a
398/// specific signature and specific return type.
399///
400/// Note this function does not expose all available features of [`ScalarUDF`],
401/// such as
402///
403/// * computing return types based on input types
404/// * multiple [`Signature`]s
405/// * aliases
406///
407/// See [`ScalarUDF`] for details and examples on how to use the full
408/// functionality.
409pub fn create_udf(
410    name: &str,
411    input_types: Vec<DataType>,
412    return_type: DataType,
413    volatility: Volatility,
414    fun: ScalarFunctionImplementation,
415) -> ScalarUDF {
416    ScalarUDF::from(SimpleScalarUDF::new(
417        name,
418        input_types,
419        return_type,
420        volatility,
421        fun,
422    ))
423}
424
425/// Implements [`ScalarUDFImpl`] for functions that have a single signature and
426/// return type.
427#[derive(PartialEq, Eq, Hash)]
428pub struct SimpleScalarUDF {
429    name: String,
430    signature: Signature,
431    return_type: DataType,
432    fun: PtrEq<ScalarFunctionImplementation>,
433}
434
435impl Debug for SimpleScalarUDF {
436    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
437        f.debug_struct("SimpleScalarUDF")
438            .field("name", &self.name)
439            .field("signature", &self.signature)
440            .field("return_type", &self.return_type)
441            .field("fun", &"<FUNC>")
442            .finish()
443    }
444}
445
446impl SimpleScalarUDF {
447    /// Create a new `SimpleScalarUDF` from a name, input types, return type and
448    /// implementation. Implementing [`ScalarUDFImpl`] allows more flexibility
449    pub fn new(
450        name: impl Into<String>,
451        input_types: Vec<DataType>,
452        return_type: DataType,
453        volatility: Volatility,
454        fun: ScalarFunctionImplementation,
455    ) -> Self {
456        Self::new_with_signature(
457            name,
458            Signature::exact(input_types, volatility),
459            return_type,
460            fun,
461        )
462    }
463
464    /// Create a new `SimpleScalarUDF` from a name, signature, return type and
465    /// implementation. Implementing [`ScalarUDFImpl`] allows more flexibility
466    pub fn new_with_signature(
467        name: impl Into<String>,
468        signature: Signature,
469        return_type: DataType,
470        fun: ScalarFunctionImplementation,
471    ) -> Self {
472        Self {
473            name: name.into(),
474            signature,
475            return_type,
476            fun: fun.into(),
477        }
478    }
479}
480
481impl ScalarUDFImpl for SimpleScalarUDF {
482    fn name(&self) -> &str {
483        &self.name
484    }
485
486    fn signature(&self) -> &Signature {
487        &self.signature
488    }
489
490    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
491        Ok(self.return_type.clone())
492    }
493
494    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
495        (self.fun)(&args.args)
496    }
497}
498
499/// Creates a new UDAF with a specific signature, state type and return type.
500/// The signature and state type must match the `Accumulator's implementation`.
501pub fn create_udaf(
502    name: &str,
503    input_type: Vec<DataType>,
504    return_type: Arc<DataType>,
505    volatility: Volatility,
506    accumulator: AccumulatorFactoryFunction,
507    state_type: Arc<Vec<DataType>>,
508) -> AggregateUDF {
509    let return_type = Arc::unwrap_or_clone(return_type);
510    let state_type = Arc::unwrap_or_clone(state_type);
511    let state_fields = state_type
512        .into_iter()
513        .enumerate()
514        .map(|(i, t)| Field::new(format!("{i}"), t, true))
515        .map(Arc::new)
516        .collect::<Vec<_>>();
517    AggregateUDF::from(SimpleAggregateUDF::new(
518        name,
519        input_type,
520        return_type,
521        volatility,
522        accumulator,
523        state_fields,
524    ))
525}
526
527/// Implements [`AggregateUDFImpl`] for functions that have a single signature and
528/// return type.
529#[derive(PartialEq, Eq, Hash)]
530pub struct SimpleAggregateUDF {
531    name: String,
532    signature: Signature,
533    return_type: DataType,
534    accumulator: PtrEq<AccumulatorFactoryFunction>,
535    state_fields: Vec<FieldRef>,
536}
537
538impl Debug for SimpleAggregateUDF {
539    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
540        f.debug_struct("SimpleAggregateUDF")
541            .field("name", &self.name)
542            .field("signature", &self.signature)
543            .field("return_type", &self.return_type)
544            .field("fun", &"<FUNC>")
545            .finish()
546    }
547}
548
549impl SimpleAggregateUDF {
550    /// Create a new `SimpleAggregateUDF` from a name, input types, return type, state type and
551    /// implementation. Implementing [`AggregateUDFImpl`] allows more flexibility
552    pub fn new(
553        name: impl Into<String>,
554        input_type: Vec<DataType>,
555        return_type: DataType,
556        volatility: Volatility,
557        accumulator: AccumulatorFactoryFunction,
558        state_fields: Vec<FieldRef>,
559    ) -> Self {
560        let name = name.into();
561        let signature = Signature::exact(input_type, volatility);
562        Self {
563            name,
564            signature,
565            return_type,
566            accumulator: accumulator.into(),
567            state_fields,
568        }
569    }
570
571    /// Create a new `SimpleAggregateUDF` from a name, signature, return type, state type and
572    /// implementation. Implementing [`AggregateUDFImpl`] allows more flexibility
573    pub fn new_with_signature(
574        name: impl Into<String>,
575        signature: Signature,
576        return_type: DataType,
577        accumulator: AccumulatorFactoryFunction,
578        state_fields: Vec<FieldRef>,
579    ) -> Self {
580        let name = name.into();
581        Self {
582            name,
583            signature,
584            return_type,
585            accumulator: accumulator.into(),
586            state_fields,
587        }
588    }
589}
590
591impl AggregateUDFImpl for SimpleAggregateUDF {
592    fn name(&self) -> &str {
593        &self.name
594    }
595
596    fn signature(&self) -> &Signature {
597        &self.signature
598    }
599
600    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
601        Ok(self.return_type.clone())
602    }
603
604    fn accumulator(
605        &self,
606        acc_args: AccumulatorArgs,
607    ) -> Result<Box<dyn crate::Accumulator>> {
608        (self.accumulator)(acc_args)
609    }
610
611    fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
612        Ok(self.state_fields.clone())
613    }
614}
615
616/// Creates a new UDWF with a specific signature, state type and return type.
617///
618/// The signature and state type must match the [`PartitionEvaluator`]'s implementation`.
619///
620/// [`PartitionEvaluator`]: crate::PartitionEvaluator
621pub fn create_udwf(
622    name: &str,
623    input_type: DataType,
624    return_type: Arc<DataType>,
625    volatility: Volatility,
626    partition_evaluator_factory: PartitionEvaluatorFactory,
627) -> WindowUDF {
628    let return_type = Arc::unwrap_or_clone(return_type);
629    WindowUDF::from(SimpleWindowUDF::new(
630        name,
631        input_type,
632        return_type,
633        volatility,
634        partition_evaluator_factory,
635    ))
636}
637
638/// Implements [`WindowUDFImpl`] for functions that have a single signature and
639/// return type.
640#[derive(PartialEq, Eq, Hash)]
641pub struct SimpleWindowUDF {
642    name: String,
643    signature: Signature,
644    return_type: DataType,
645    partition_evaluator_factory: PtrEq<PartitionEvaluatorFactory>,
646}
647
648impl Debug for SimpleWindowUDF {
649    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
650        f.debug_struct("WindowUDF")
651            .field("name", &self.name)
652            .field("signature", &self.signature)
653            .field("return_type", &"<func>")
654            .field("partition_evaluator_factory", &"<FUNC>")
655            .finish()
656    }
657}
658
659impl SimpleWindowUDF {
660    /// Create a new `SimpleWindowUDF` from a name, input types, return type and
661    /// implementation. Implementing [`WindowUDFImpl`] allows more flexibility
662    pub fn new(
663        name: impl Into<String>,
664        input_type: DataType,
665        return_type: DataType,
666        volatility: Volatility,
667        partition_evaluator_factory: PartitionEvaluatorFactory,
668    ) -> Self {
669        let name = name.into();
670        let signature = Signature::exact([input_type].to_vec(), volatility);
671        Self {
672            name,
673            signature,
674            return_type,
675            partition_evaluator_factory: partition_evaluator_factory.into(),
676        }
677    }
678}
679
680impl WindowUDFImpl for SimpleWindowUDF {
681    fn name(&self) -> &str {
682        &self.name
683    }
684
685    fn signature(&self) -> &Signature {
686        &self.signature
687    }
688
689    fn partition_evaluator(
690        &self,
691        _partition_evaluator_args: PartitionEvaluatorArgs,
692    ) -> Result<Box<dyn PartitionEvaluator>> {
693        (self.partition_evaluator_factory)()
694    }
695
696    fn field(&self, field_args: WindowUDFFieldArgs) -> Result<FieldRef> {
697        Ok(Arc::new(Field::new(
698            field_args.name(),
699            self.return_type.clone(),
700            true,
701        )))
702    }
703
704    fn limit_effect(&self, _args: &[Arc<dyn PhysicalExpr>]) -> LimitEffect {
705        LimitEffect::Unknown
706    }
707}
708
709pub fn interval_year_month_lit(value: &str) -> Expr {
710    let interval = parse_interval_year_month(value).ok();
711    Expr::Literal(ScalarValue::IntervalYearMonth(interval), None)
712}
713
714pub fn interval_datetime_lit(value: &str) -> Expr {
715    let interval = parse_interval_day_time(value).ok();
716    Expr::Literal(ScalarValue::IntervalDayTime(interval), None)
717}
718
719pub fn interval_month_day_nano_lit(value: &str) -> Expr {
720    let interval = parse_interval_month_day_nano(value).ok();
721    Expr::Literal(ScalarValue::IntervalMonthDayNano(interval), None)
722}
723
724/// Create a lambda expression
725pub fn lambda(params: impl IntoIterator<Item = impl Into<String>>, body: Expr) -> Expr {
726    Expr::Lambda(Lambda::new(
727        params.into_iter().map(Into::into).collect(),
728        body,
729    ))
730}
731
732/// Create an unresolved lambda variable expression
733///
734/// The expression tree or [`LogicalPlan`] which
735/// owns this variable must be resolved before usage with either
736/// [`Expr::resolve_lambda_variables`] or [`LogicalPlan::resolve_lambda_variables`].
737///
738/// [LogicalPlan::resolve_lambda_variables]: crate::LogicalPlan::resolve_lambda_variables
739pub fn lambda_var(name: impl Into<String>) -> Expr {
740    Expr::LambdaVariable(LambdaVariable::new(name.into(), None))
741}
742
743/// Extensions for configuring [`Expr::AggregateFunction`] or [`Expr::WindowFunction`]
744///
745/// Adds methods to [`Expr`] that make it easy to set optional options
746/// such as `ORDER BY`, `FILTER` and `DISTINCT`
747///
748/// # Example
749/// ```no_run
750/// # use datafusion_common::Result;
751/// # use datafusion_expr::expr::NullTreatment;
752/// # use datafusion_expr::test::function_stub::count;
753/// # use datafusion_expr::{ExprFunctionExt, lit, Expr, col};
754/// # // first_value is an aggregate function in another crate
755/// # fn first_value(_arg: Expr) -> Expr {
756/// unimplemented!() }
757/// # fn main() -> Result<()> {
758/// // Create an aggregate count, filtering on column y > 5
759/// let agg = count(col("x")).filter(col("y").gt(lit(5))).build()?;
760///
761/// // Find the first value in an aggregate sorted by column y
762/// // equivalent to:
763/// // `FIRST_VALUE(x ORDER BY y ASC IGNORE NULLS)`
764/// let sort_expr = col("y").sort(true, true);
765/// let agg = first_value(col("x"))
766///     .order_by(vec![sort_expr])
767///     .null_treatment(NullTreatment::IgnoreNulls)
768///     .build()?;
769///
770/// // Create a window expression for percent rank partitioned on column a
771/// // equivalent to:
772/// // `PERCENT_RANK() OVER (PARTITION BY a ORDER BY b ASC NULLS LAST IGNORE NULLS)`
773/// // percent_rank is an udwf function in another crate
774/// # fn percent_rank() -> Expr {
775/// unimplemented!() }
776/// let window = percent_rank()
777///     .partition_by(vec![col("a")])
778///     .order_by(vec![col("b").sort(true, true)])
779///     .null_treatment(NullTreatment::IgnoreNulls)
780///     .build()?;
781/// #     Ok(())
782/// # }
783/// ```
784pub trait ExprFunctionExt {
785    /// Add `ORDER BY <order_by>`
786    fn order_by(self, order_by: Vec<Sort>) -> ExprFuncBuilder;
787    /// Add `FILTER <filter>`
788    fn filter(self, filter: Expr) -> ExprFuncBuilder;
789    /// Add `DISTINCT`
790    fn distinct(self) -> ExprFuncBuilder;
791    /// Add `RESPECT NULLS` or `IGNORE NULLS`
792    fn null_treatment(
793        self,
794        null_treatment: impl Into<Option<NullTreatment>>,
795    ) -> ExprFuncBuilder;
796    /// Add `PARTITION BY`
797    fn partition_by(self, partition_by: Vec<Expr>) -> ExprFuncBuilder;
798    /// Add appropriate window frame conditions
799    fn window_frame(self, window_frame: WindowFrame) -> ExprFuncBuilder;
800}
801
802#[derive(Debug, Clone)]
803pub enum ExprFuncKind {
804    Aggregate(AggregateFunction),
805    Window(Box<WindowFunction>),
806}
807
808/// Implementation of [`ExprFunctionExt`].
809///
810/// See [`ExprFunctionExt`] for usage and examples
811#[derive(Debug, Clone)]
812pub struct ExprFuncBuilder {
813    fun: Option<ExprFuncKind>,
814    order_by: Option<Vec<Sort>>,
815    filter: Option<Expr>,
816    distinct: bool,
817    null_treatment: Option<NullTreatment>,
818    partition_by: Option<Vec<Expr>>,
819    window_frame: Option<WindowFrame>,
820}
821
822impl ExprFuncBuilder {
823    /// Create a new `ExprFuncBuilder`, see [`ExprFunctionExt`]
824    fn new(fun: Option<ExprFuncKind>) -> Self {
825        Self {
826            fun,
827            order_by: None,
828            filter: None,
829            distinct: false,
830            null_treatment: None,
831            partition_by: None,
832            window_frame: None,
833        }
834    }
835
836    /// Updates and returns the in progress [`Expr::AggregateFunction`] or [`Expr::WindowFunction`]
837    ///
838    /// # Errors:
839    ///
840    /// Returns an error if this builder  [`ExprFunctionExt`] was used with an
841    /// `Expr` variant other than [`Expr::AggregateFunction`] or [`Expr::WindowFunction`]
842    pub fn build(self) -> Result<Expr> {
843        let Self {
844            fun,
845            order_by,
846            filter,
847            distinct,
848            null_treatment,
849            partition_by,
850            window_frame,
851        } = self;
852
853        let Some(fun) = fun else {
854            return plan_err!(
855                "ExprFunctionExt can only be used with Expr::AggregateFunction or Expr::WindowFunction"
856            );
857        };
858
859        let fun_expr = match fun {
860            ExprFuncKind::Aggregate(mut udaf) => {
861                udaf.params.order_by = order_by.unwrap_or_default();
862                udaf.params.filter = filter.map(Box::new);
863                udaf.params.distinct = distinct;
864                udaf.params.null_treatment = null_treatment;
865                Expr::AggregateFunction(udaf)
866            }
867            ExprFuncKind::Window(mut udwf) => {
868                let has_order_by = order_by.as_ref().map(|o| !o.is_empty());
869                udwf.params.partition_by = partition_by.unwrap_or_default();
870                udwf.params.order_by = order_by.unwrap_or_default();
871                udwf.params.window_frame =
872                    window_frame.unwrap_or_else(|| WindowFrame::new(has_order_by));
873                udwf.params.filter = filter.map(Box::new);
874                udwf.params.null_treatment = null_treatment;
875                udwf.params.distinct = distinct;
876                Expr::WindowFunction(udwf)
877            }
878        };
879
880        Ok(fun_expr)
881    }
882}
883
884impl ExprFunctionExt for ExprFuncBuilder {
885    /// Add `ORDER BY <order_by>`
886    fn order_by(mut self, order_by: Vec<Sort>) -> ExprFuncBuilder {
887        self.order_by = Some(order_by);
888        self
889    }
890
891    /// Add `FILTER <filter>`
892    fn filter(mut self, filter: Expr) -> ExprFuncBuilder {
893        self.filter = Some(filter);
894        self
895    }
896
897    /// Add `DISTINCT`
898    fn distinct(mut self) -> ExprFuncBuilder {
899        self.distinct = true;
900        self
901    }
902
903    /// Add `RESPECT NULLS` or `IGNORE NULLS`
904    fn null_treatment(
905        mut self,
906        null_treatment: impl Into<Option<NullTreatment>>,
907    ) -> ExprFuncBuilder {
908        self.null_treatment = null_treatment.into();
909        self
910    }
911
912    fn partition_by(mut self, partition_by: Vec<Expr>) -> ExprFuncBuilder {
913        self.partition_by = Some(partition_by);
914        self
915    }
916
917    fn window_frame(mut self, window_frame: WindowFrame) -> ExprFuncBuilder {
918        self.window_frame = Some(window_frame);
919        self
920    }
921}
922
923impl ExprFunctionExt for Expr {
924    fn order_by(self, order_by: Vec<Sort>) -> ExprFuncBuilder {
925        let mut builder = match self {
926            Expr::AggregateFunction(udaf) => {
927                ExprFuncBuilder::new(Some(ExprFuncKind::Aggregate(udaf)))
928            }
929            Expr::WindowFunction(udwf) => {
930                ExprFuncBuilder::new(Some(ExprFuncKind::Window(udwf)))
931            }
932            _ => ExprFuncBuilder::new(None),
933        };
934        if builder.fun.is_some() {
935            builder.order_by = Some(order_by);
936        }
937        builder
938    }
939    fn filter(self, filter: Expr) -> ExprFuncBuilder {
940        match self {
941            Expr::AggregateFunction(udaf) => {
942                let mut builder =
943                    ExprFuncBuilder::new(Some(ExprFuncKind::Aggregate(udaf)));
944                builder.filter = Some(filter);
945                builder
946            }
947            _ => ExprFuncBuilder::new(None),
948        }
949    }
950    fn distinct(self) -> ExprFuncBuilder {
951        match self {
952            Expr::AggregateFunction(udaf) => {
953                let mut builder =
954                    ExprFuncBuilder::new(Some(ExprFuncKind::Aggregate(udaf)));
955                builder.distinct = true;
956                builder
957            }
958            _ => ExprFuncBuilder::new(None),
959        }
960    }
961    fn null_treatment(
962        self,
963        null_treatment: impl Into<Option<NullTreatment>>,
964    ) -> ExprFuncBuilder {
965        let mut builder = match self {
966            Expr::AggregateFunction(udaf) => {
967                ExprFuncBuilder::new(Some(ExprFuncKind::Aggregate(udaf)))
968            }
969            Expr::WindowFunction(udwf) => {
970                ExprFuncBuilder::new(Some(ExprFuncKind::Window(udwf)))
971            }
972            _ => ExprFuncBuilder::new(None),
973        };
974        if builder.fun.is_some() {
975            builder.null_treatment = null_treatment.into();
976        }
977        builder
978    }
979
980    fn partition_by(self, partition_by: Vec<Expr>) -> ExprFuncBuilder {
981        match self {
982            Expr::WindowFunction(udwf) => {
983                let mut builder = ExprFuncBuilder::new(Some(ExprFuncKind::Window(udwf)));
984                builder.partition_by = Some(partition_by);
985                builder
986            }
987            _ => ExprFuncBuilder::new(None),
988        }
989    }
990
991    fn window_frame(self, window_frame: WindowFrame) -> ExprFuncBuilder {
992        match self {
993            Expr::WindowFunction(udwf) => {
994                let mut builder = ExprFuncBuilder::new(Some(ExprFuncKind::Window(udwf)));
995                builder.window_frame = Some(window_frame);
996                builder
997            }
998            _ => ExprFuncBuilder::new(None),
999        }
1000    }
1001}
1002
1003#[cfg(test)]
1004mod test {
1005    use super::*;
1006
1007    #[test]
1008    fn filter_is_null_and_is_not_null() {
1009        let col_null = col("col1");
1010        let col_not_null = ident("col2");
1011        assert_eq!(format!("{}", col_null.is_null()), "col1 IS NULL");
1012        assert_eq!(
1013            format!("{}", col_not_null.is_not_null()),
1014            "col2 IS NOT NULL"
1015        );
1016    }
1017}