use std::collections::HashSet;
use std::fmt::{self, Display, Formatter, Write};
use std::hash::{Hash, Hasher};
use std::mem;
use std::str::FromStr;
use std::sync::Arc;
use crate::expr_fn::binary_expr;
use crate::logical_plan::Subquery;
use crate::utils::expr_to_columns;
use crate::{
aggregate_function, built_in_window_function, udaf, ExprSchemable, Operator,
Signature,
};
use crate::{window_frame, Volatility};
use arrow::datatypes::{DataType, FieldRef};
use datafusion_common::tree_node::{
Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
};
use datafusion_common::{
internal_err, plan_err, Column, DFSchema, Result, ScalarValue, TableReference,
};
use sqlparser::ast::NullTreatment;
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub enum Expr {
Alias(Alias),
Column(Column),
ScalarVariable(DataType, Vec<String>),
Literal(ScalarValue),
BinaryExpr(BinaryExpr),
Like(Like),
SimilarTo(Like),
Not(Box<Expr>),
IsNotNull(Box<Expr>),
IsNull(Box<Expr>),
IsTrue(Box<Expr>),
IsFalse(Box<Expr>),
IsUnknown(Box<Expr>),
IsNotTrue(Box<Expr>),
IsNotFalse(Box<Expr>),
IsNotUnknown(Box<Expr>),
Negative(Box<Expr>),
Between(Between),
Case(Case),
Cast(Cast),
TryCast(TryCast),
Sort(Sort),
ScalarFunction(ScalarFunction),
AggregateFunction(AggregateFunction),
WindowFunction(WindowFunction),
InList(InList),
Exists(Exists),
InSubquery(InSubquery),
ScalarSubquery(Subquery),
Wildcard { qualifier: Option<TableReference> },
GroupingSet(GroupingSet),
Placeholder(Placeholder),
OuterReferenceColumn(DataType, Column),
Unnest(Unnest),
}
impl Default for Expr {
fn default() -> Self {
Expr::Literal(ScalarValue::Null)
}
}
impl From<Column> for Expr {
fn from(value: Column) -> Self {
Expr::Column(value)
}
}
impl<'a> From<(Option<&'a TableReference>, &'a FieldRef)> for Expr {
fn from(value: (Option<&'a TableReference>, &'a FieldRef)) -> Self {
Expr::from(Column::from(value))
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Unnest {
pub expr: Box<Expr>,
}
impl Unnest {
pub fn new(expr: Expr) -> Self {
Self {
expr: Box::new(expr),
}
}
pub fn new_boxed(boxed: Box<Expr>) -> Self {
Self { expr: boxed }
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Alias {
pub expr: Box<Expr>,
pub relation: Option<TableReference>,
pub name: String,
}
impl Alias {
pub fn new(
expr: Expr,
relation: Option<impl Into<TableReference>>,
name: impl Into<String>,
) -> Self {
Self {
expr: Box::new(expr),
relation: relation.map(|r| r.into()),
name: name.into(),
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct BinaryExpr {
pub left: Box<Expr>,
pub op: Operator,
pub right: Box<Expr>,
}
impl BinaryExpr {
pub fn new(left: Box<Expr>, op: Operator, right: Box<Expr>) -> Self {
Self { left, op, right }
}
}
impl Display for BinaryExpr {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
fn write_child(
f: &mut Formatter<'_>,
expr: &Expr,
precedence: u8,
) -> fmt::Result {
match expr {
Expr::BinaryExpr(child) => {
let p = child.op.precedence();
if p == 0 || p < precedence {
write!(f, "({child})")?;
} else {
write!(f, "{child}")?;
}
}
_ => write!(f, "{expr}")?,
}
Ok(())
}
let precedence = self.op.precedence();
write_child(f, self.left.as_ref(), precedence)?;
write!(f, " {} ", self.op)?;
write_child(f, self.right.as_ref(), precedence)
}
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct Case {
pub expr: Option<Box<Expr>>,
pub when_then_expr: Vec<(Box<Expr>, Box<Expr>)>,
pub else_expr: Option<Box<Expr>>,
}
impl Case {
pub fn new(
expr: Option<Box<Expr>>,
when_then_expr: Vec<(Box<Expr>, Box<Expr>)>,
else_expr: Option<Box<Expr>>,
) -> Self {
Self {
expr,
when_then_expr,
else_expr,
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Like {
pub negated: bool,
pub expr: Box<Expr>,
pub pattern: Box<Expr>,
pub escape_char: Option<char>,
pub case_insensitive: bool,
}
impl Like {
pub fn new(
negated: bool,
expr: Box<Expr>,
pattern: Box<Expr>,
escape_char: Option<char>,
case_insensitive: bool,
) -> Self {
Self {
negated,
expr,
pattern,
escape_char,
case_insensitive,
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Between {
pub expr: Box<Expr>,
pub negated: bool,
pub low: Box<Expr>,
pub high: Box<Expr>,
}
impl Between {
pub fn new(expr: Box<Expr>, negated: bool, low: Box<Expr>, high: Box<Expr>) -> Self {
Self {
expr,
negated,
low,
high,
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct ScalarFunction {
pub func: Arc<crate::ScalarUDF>,
pub args: Vec<Expr>,
}
impl ScalarFunction {
pub fn name(&self) -> &str {
self.func.name()
}
}
impl ScalarFunction {
pub fn new_udf(udf: Arc<crate::ScalarUDF>, args: Vec<Expr>) -> Self {
Self { func: udf, args }
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub enum GetFieldAccess {
NamedStructField { name: ScalarValue },
ListIndex { key: Box<Expr> },
ListRange {
start: Box<Expr>,
stop: Box<Expr>,
stride: Box<Expr>,
},
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Cast {
pub expr: Box<Expr>,
pub data_type: DataType,
}
impl Cast {
pub fn new(expr: Box<Expr>, data_type: DataType) -> Self {
Self { expr, data_type }
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct TryCast {
pub expr: Box<Expr>,
pub data_type: DataType,
}
impl TryCast {
pub fn new(expr: Box<Expr>, data_type: DataType) -> Self {
Self { expr, data_type }
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Sort {
pub expr: Box<Expr>,
pub asc: bool,
pub nulls_first: bool,
}
impl Sort {
pub fn new(expr: Box<Expr>, asc: bool, nulls_first: bool) -> Self {
Self {
expr,
asc,
nulls_first,
}
}
pub fn reverse(&self) -> Self {
Self {
expr: self.expr.clone(),
asc: !self.asc,
nulls_first: !self.nulls_first,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum AggregateFunctionDefinition {
BuiltIn(aggregate_function::AggregateFunction),
UDF(Arc<crate::AggregateUDF>),
}
impl AggregateFunctionDefinition {
pub fn name(&self) -> &str {
match self {
AggregateFunctionDefinition::BuiltIn(fun) => fun.name(),
AggregateFunctionDefinition::UDF(udf) => udf.name(),
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct AggregateFunction {
pub func_def: AggregateFunctionDefinition,
pub args: Vec<Expr>,
pub distinct: bool,
pub filter: Option<Box<Expr>>,
pub order_by: Option<Vec<Expr>>,
pub null_treatment: Option<NullTreatment>,
}
impl AggregateFunction {
pub fn new(
fun: aggregate_function::AggregateFunction,
args: Vec<Expr>,
distinct: bool,
filter: Option<Box<Expr>>,
order_by: Option<Vec<Expr>>,
null_treatment: Option<NullTreatment>,
) -> Self {
Self {
func_def: AggregateFunctionDefinition::BuiltIn(fun),
args,
distinct,
filter,
order_by,
null_treatment,
}
}
pub fn new_udf(
udf: Arc<crate::AggregateUDF>,
args: Vec<Expr>,
distinct: bool,
filter: Option<Box<Expr>>,
order_by: Option<Vec<Expr>>,
null_treatment: Option<NullTreatment>,
) -> Self {
Self {
func_def: AggregateFunctionDefinition::UDF(udf),
args,
distinct,
filter,
order_by,
null_treatment,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum WindowFunctionDefinition {
AggregateFunction(aggregate_function::AggregateFunction),
BuiltInWindowFunction(built_in_window_function::BuiltInWindowFunction),
AggregateUDF(Arc<crate::AggregateUDF>),
WindowUDF(Arc<crate::WindowUDF>),
}
impl WindowFunctionDefinition {
pub fn return_type(
&self,
input_expr_types: &[DataType],
input_expr_nullable: &[bool],
) -> Result<DataType> {
match self {
WindowFunctionDefinition::AggregateFunction(fun) => {
fun.return_type(input_expr_types, input_expr_nullable)
}
WindowFunctionDefinition::BuiltInWindowFunction(fun) => {
fun.return_type(input_expr_types)
}
WindowFunctionDefinition::AggregateUDF(fun) => {
fun.return_type(input_expr_types)
}
WindowFunctionDefinition::WindowUDF(fun) => fun.return_type(input_expr_types),
}
}
pub fn signature(&self) -> Signature {
match self {
WindowFunctionDefinition::AggregateFunction(fun) => fun.signature(),
WindowFunctionDefinition::BuiltInWindowFunction(fun) => fun.signature(),
WindowFunctionDefinition::AggregateUDF(fun) => fun.signature().clone(),
WindowFunctionDefinition::WindowUDF(fun) => fun.signature().clone(),
}
}
pub fn name(&self) -> &str {
match self {
WindowFunctionDefinition::BuiltInWindowFunction(fun) => fun.name(),
WindowFunctionDefinition::WindowUDF(fun) => fun.name(),
WindowFunctionDefinition::AggregateFunction(fun) => fun.name(),
WindowFunctionDefinition::AggregateUDF(fun) => fun.name(),
}
}
}
impl fmt::Display for WindowFunctionDefinition {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
WindowFunctionDefinition::AggregateFunction(fun) => {
std::fmt::Display::fmt(fun, f)
}
WindowFunctionDefinition::BuiltInWindowFunction(fun) => {
std::fmt::Display::fmt(fun, f)
}
WindowFunctionDefinition::AggregateUDF(fun) => std::fmt::Display::fmt(fun, f),
WindowFunctionDefinition::WindowUDF(fun) => std::fmt::Display::fmt(fun, f),
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct WindowFunction {
pub fun: WindowFunctionDefinition,
pub args: Vec<Expr>,
pub partition_by: Vec<Expr>,
pub order_by: Vec<Expr>,
pub window_frame: window_frame::WindowFrame,
pub null_treatment: Option<NullTreatment>,
}
impl WindowFunction {
pub fn new(
fun: WindowFunctionDefinition,
args: Vec<Expr>,
partition_by: Vec<Expr>,
order_by: Vec<Expr>,
window_frame: window_frame::WindowFrame,
null_treatment: Option<NullTreatment>,
) -> Self {
Self {
fun,
args,
partition_by,
order_by,
window_frame,
null_treatment,
}
}
}
pub fn find_df_window_func(name: &str) -> Option<WindowFunctionDefinition> {
let name = name.to_lowercase();
if let Ok(built_in_function) =
built_in_window_function::BuiltInWindowFunction::from_str(name.as_str())
{
Some(WindowFunctionDefinition::BuiltInWindowFunction(
built_in_function,
))
} else if let Ok(aggregate) =
aggregate_function::AggregateFunction::from_str(name.as_str())
{
Some(WindowFunctionDefinition::AggregateFunction(aggregate))
} else {
None
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Exists {
pub subquery: Subquery,
pub negated: bool,
}
impl Exists {
pub fn new(subquery: Subquery, negated: bool) -> Self {
Self { subquery, negated }
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct AggregateUDF {
pub fun: Arc<udaf::AggregateUDF>,
pub args: Vec<Expr>,
pub filter: Option<Box<Expr>>,
pub order_by: Option<Vec<Expr>>,
}
impl AggregateUDF {
pub fn new(
fun: Arc<udaf::AggregateUDF>,
args: Vec<Expr>,
filter: Option<Box<Expr>>,
order_by: Option<Vec<Expr>>,
) -> Self {
Self {
fun,
args,
filter,
order_by,
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct InList {
pub expr: Box<Expr>,
pub list: Vec<Expr>,
pub negated: bool,
}
impl InList {
pub fn new(expr: Box<Expr>, list: Vec<Expr>, negated: bool) -> Self {
Self {
expr,
list,
negated,
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct InSubquery {
pub expr: Box<Expr>,
pub subquery: Subquery,
pub negated: bool,
}
impl InSubquery {
pub fn new(expr: Box<Expr>, subquery: Subquery, negated: bool) -> Self {
Self {
expr,
subquery,
negated,
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Placeholder {
pub id: String,
pub data_type: Option<DataType>,
}
impl Placeholder {
pub fn new(id: String, data_type: Option<DataType>) -> Self {
Self { id, data_type }
}
}
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub enum GroupingSet {
Rollup(Vec<Expr>),
Cube(Vec<Expr>),
GroupingSets(Vec<Vec<Expr>>),
}
impl GroupingSet {
pub fn distinct_expr(&self) -> Vec<&Expr> {
match self {
GroupingSet::Rollup(exprs) | GroupingSet::Cube(exprs) => {
exprs.iter().collect()
}
GroupingSet::GroupingSets(groups) => {
let mut exprs: Vec<&Expr> = vec![];
for exp in groups.iter().flatten() {
if !exprs.contains(&exp) {
exprs.push(exp);
}
}
exprs
}
}
}
}
const SEED: ahash::RandomState = ahash::RandomState::with_seeds(0, 0, 0, 0);
impl PartialOrd for Expr {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
let s = SEED.hash_one(self);
let o = SEED.hash_one(other);
Some(s.cmp(&o))
}
}
impl Expr {
pub fn display_name(&self) -> Result<String> {
create_name(self)
}
pub fn canonical_name(&self) -> String {
format!("{self}")
}
pub fn variant_name(&self) -> &str {
match self {
Expr::AggregateFunction { .. } => "AggregateFunction",
Expr::Alias(..) => "Alias",
Expr::Between { .. } => "Between",
Expr::BinaryExpr { .. } => "BinaryExpr",
Expr::Case { .. } => "Case",
Expr::Cast { .. } => "Cast",
Expr::Column(..) => "Column",
Expr::OuterReferenceColumn(_, _) => "Outer",
Expr::Exists { .. } => "Exists",
Expr::GroupingSet(..) => "GroupingSet",
Expr::InList { .. } => "InList",
Expr::InSubquery(..) => "InSubquery",
Expr::IsNotNull(..) => "IsNotNull",
Expr::IsNull(..) => "IsNull",
Expr::Like { .. } => "Like",
Expr::SimilarTo { .. } => "RLike",
Expr::IsTrue(..) => "IsTrue",
Expr::IsFalse(..) => "IsFalse",
Expr::IsUnknown(..) => "IsUnknown",
Expr::IsNotTrue(..) => "IsNotTrue",
Expr::IsNotFalse(..) => "IsNotFalse",
Expr::IsNotUnknown(..) => "IsNotUnknown",
Expr::Literal(..) => "Literal",
Expr::Negative(..) => "Negative",
Expr::Not(..) => "Not",
Expr::Placeholder(_) => "Placeholder",
Expr::ScalarFunction(..) => "ScalarFunction",
Expr::ScalarSubquery { .. } => "ScalarSubquery",
Expr::ScalarVariable(..) => "ScalarVariable",
Expr::Sort { .. } => "Sort",
Expr::TryCast { .. } => "TryCast",
Expr::WindowFunction { .. } => "WindowFunction",
Expr::Wildcard { .. } => "Wildcard",
Expr::Unnest { .. } => "Unnest",
}
}
pub fn eq(self, other: Expr) -> Expr {
binary_expr(self, Operator::Eq, other)
}
pub fn not_eq(self, other: Expr) -> Expr {
binary_expr(self, Operator::NotEq, other)
}
pub fn gt(self, other: Expr) -> Expr {
binary_expr(self, Operator::Gt, other)
}
pub fn gt_eq(self, other: Expr) -> Expr {
binary_expr(self, Operator::GtEq, other)
}
pub fn lt(self, other: Expr) -> Expr {
binary_expr(self, Operator::Lt, other)
}
pub fn lt_eq(self, other: Expr) -> Expr {
binary_expr(self, Operator::LtEq, other)
}
pub fn and(self, other: Expr) -> Expr {
binary_expr(self, Operator::And, other)
}
pub fn or(self, other: Expr) -> Expr {
binary_expr(self, Operator::Or, other)
}
pub fn like(self, other: Expr) -> Expr {
Expr::Like(Like::new(
false,
Box::new(self),
Box::new(other),
None,
false,
))
}
pub fn not_like(self, other: Expr) -> Expr {
Expr::Like(Like::new(
true,
Box::new(self),
Box::new(other),
None,
false,
))
}
pub fn ilike(self, other: Expr) -> Expr {
Expr::Like(Like::new(
false,
Box::new(self),
Box::new(other),
None,
true,
))
}
pub fn not_ilike(self, other: Expr) -> Expr {
Expr::Like(Like::new(true, Box::new(self), Box::new(other), None, true))
}
pub fn name_for_alias(&self) -> Result<String> {
match self {
Expr::Sort(Sort { expr, .. }) => expr.name_for_alias(),
expr => expr.display_name(),
}
}
pub fn alias_if_changed(self, original_name: String) -> Result<Expr> {
let new_name = self.name_for_alias()?;
if new_name == original_name {
return Ok(self);
}
Ok(self.alias(original_name))
}
pub fn alias(self, name: impl Into<String>) -> Expr {
match self {
Expr::Sort(Sort {
expr,
asc,
nulls_first,
}) => Expr::Sort(Sort::new(Box::new(expr.alias(name)), asc, nulls_first)),
_ => Expr::Alias(Alias::new(self, None::<&str>, name.into())),
}
}
pub fn alias_qualified(
self,
relation: Option<impl Into<TableReference>>,
name: impl Into<String>,
) -> Expr {
match self {
Expr::Sort(Sort {
expr,
asc,
nulls_first,
}) => Expr::Sort(Sort::new(
Box::new(expr.alias_qualified(relation, name)),
asc,
nulls_first,
)),
_ => Expr::Alias(Alias::new(self, relation, name.into())),
}
}
pub fn unalias(self) -> Expr {
match self {
Expr::Alias(alias) => *alias.expr,
_ => self,
}
}
pub fn unalias_nested(self) -> Transformed<Expr> {
self.transform_down_up(
|expr| {
let recursion = if matches!(
expr,
Expr::Exists { .. } | Expr::ScalarSubquery(_) | Expr::InSubquery(_)
) {
TreeNodeRecursion::Jump
} else {
TreeNodeRecursion::Continue
};
Ok(Transformed::new(expr, false, recursion))
},
|expr| {
if let Expr::Alias(Alias { expr, .. }) = expr {
Ok(Transformed::yes(*expr))
} else {
Ok(Transformed::no(expr))
}
},
)
.unwrap()
}
pub fn in_list(self, list: Vec<Expr>, negated: bool) -> Expr {
Expr::InList(InList::new(Box::new(self), list, negated))
}
pub fn is_null(self) -> Expr {
Expr::IsNull(Box::new(self))
}
pub fn is_not_null(self) -> Expr {
Expr::IsNotNull(Box::new(self))
}
pub fn sort(self, asc: bool, nulls_first: bool) -> Expr {
Expr::Sort(Sort::new(Box::new(self), asc, nulls_first))
}
pub fn is_true(self) -> Expr {
Expr::IsTrue(Box::new(self))
}
pub fn is_not_true(self) -> Expr {
Expr::IsNotTrue(Box::new(self))
}
pub fn is_false(self) -> Expr {
Expr::IsFalse(Box::new(self))
}
pub fn is_not_false(self) -> Expr {
Expr::IsNotFalse(Box::new(self))
}
pub fn is_unknown(self) -> Expr {
Expr::IsUnknown(Box::new(self))
}
pub fn is_not_unknown(self) -> Expr {
Expr::IsNotUnknown(Box::new(self))
}
pub fn between(self, low: Expr, high: Expr) -> Expr {
Expr::Between(Between::new(
Box::new(self),
false,
Box::new(low),
Box::new(high),
))
}
pub fn not_between(self, low: Expr, high: Expr) -> Expr {
Expr::Between(Between::new(
Box::new(self),
true,
Box::new(low),
Box::new(high),
))
}
#[deprecated(since = "39.0.0", note = "use try_as_col instead")]
pub fn try_into_col(&self) -> Result<Column> {
match self {
Expr::Column(it) => Ok(it.clone()),
_ => plan_err!("Could not coerce '{self}' into Column!"),
}
}
pub fn try_as_col(&self) -> Option<&Column> {
if let Expr::Column(it) = self {
Some(it)
} else {
None
}
}
#[deprecated(since = "40.0.0", note = "use Expr::column_refs instead")]
pub fn to_columns(&self) -> Result<HashSet<Column>> {
let mut using_columns = HashSet::new();
expr_to_columns(self, &mut using_columns)?;
Ok(using_columns)
}
pub fn column_refs(&self) -> HashSet<&Column> {
let mut using_columns = HashSet::new();
self.add_column_refs(&mut using_columns);
using_columns
}
pub fn add_column_refs<'a>(&'a self, set: &mut HashSet<&'a Column>) {
self.apply(|expr| {
if let Expr::Column(col) = expr {
set.insert(col);
}
Ok(TreeNodeRecursion::Continue)
})
.expect("traversal is infallable");
}
pub fn any_column_refs(&self) -> bool {
self.exists(|expr| Ok(matches!(expr, Expr::Column(_))))
.unwrap()
}
pub fn contains_outer(&self) -> bool {
self.exists(|expr| Ok(matches!(expr, Expr::OuterReferenceColumn { .. })))
.unwrap()
}
pub fn is_volatile_node(&self) -> bool {
matches!(self, Expr::ScalarFunction(func) if func.func.signature().volatility == Volatility::Volatile)
}
pub fn is_volatile(&self) -> Result<bool> {
self.exists(|expr| Ok(expr.is_volatile_node()))
}
pub fn infer_placeholder_types(self, schema: &DFSchema) -> Result<Expr> {
self.transform(|mut expr| {
if let Expr::BinaryExpr(BinaryExpr { left, op: _, right }) = &mut expr {
rewrite_placeholder(left.as_mut(), right.as_ref(), schema)?;
rewrite_placeholder(right.as_mut(), left.as_ref(), schema)?;
};
if let Expr::Between(Between {
expr,
negated: _,
low,
high,
}) = &mut expr
{
rewrite_placeholder(low.as_mut(), expr.as_ref(), schema)?;
rewrite_placeholder(high.as_mut(), expr.as_ref(), schema)?;
}
Ok(Transformed::yes(expr))
})
.data()
}
pub fn short_circuits(&self) -> bool {
match self {
Expr::ScalarFunction(ScalarFunction { func, .. }) => func.short_circuits(),
Expr::BinaryExpr(BinaryExpr { op, .. }) => {
matches!(op, Operator::And | Operator::Or)
}
Expr::Case { .. } => true,
Expr::AggregateFunction(..)
| Expr::Alias(..)
| Expr::Between(..)
| Expr::Cast(..)
| Expr::Column(..)
| Expr::Exists(..)
| Expr::GroupingSet(..)
| Expr::InList(..)
| Expr::InSubquery(..)
| Expr::IsFalse(..)
| Expr::IsNotFalse(..)
| Expr::IsNotNull(..)
| Expr::IsNotTrue(..)
| Expr::IsNotUnknown(..)
| Expr::IsNull(..)
| Expr::IsTrue(..)
| Expr::IsUnknown(..)
| Expr::Like(..)
| Expr::ScalarSubquery(..)
| Expr::ScalarVariable(_, _)
| Expr::SimilarTo(..)
| Expr::Not(..)
| Expr::Negative(..)
| Expr::OuterReferenceColumn(_, _)
| Expr::TryCast(..)
| Expr::Unnest(..)
| Expr::Wildcard { .. }
| Expr::WindowFunction(..)
| Expr::Literal(..)
| Expr::Sort(..)
| Expr::Placeholder(..) => false,
}
}
pub fn hash_node<H: Hasher>(&self, hasher: &mut H) {
mem::discriminant(self).hash(hasher);
match self {
Expr::Alias(Alias {
expr: _expr,
relation,
name,
}) => {
relation.hash(hasher);
name.hash(hasher);
}
Expr::Column(column) => {
column.hash(hasher);
}
Expr::ScalarVariable(data_type, name) => {
data_type.hash(hasher);
name.hash(hasher);
}
Expr::Literal(scalar_value) => {
scalar_value.hash(hasher);
}
Expr::BinaryExpr(BinaryExpr {
left: _left,
op,
right: _right,
}) => {
op.hash(hasher);
}
Expr::Like(Like {
negated,
expr: _expr,
pattern: _pattern,
escape_char,
case_insensitive,
})
| Expr::SimilarTo(Like {
negated,
expr: _expr,
pattern: _pattern,
escape_char,
case_insensitive,
}) => {
negated.hash(hasher);
escape_char.hash(hasher);
case_insensitive.hash(hasher);
}
Expr::Not(_expr)
| Expr::IsNotNull(_expr)
| Expr::IsNull(_expr)
| Expr::IsTrue(_expr)
| Expr::IsFalse(_expr)
| Expr::IsUnknown(_expr)
| Expr::IsNotTrue(_expr)
| Expr::IsNotFalse(_expr)
| Expr::IsNotUnknown(_expr)
| Expr::Negative(_expr) => {}
Expr::Between(Between {
expr: _expr,
negated,
low: _low,
high: _high,
}) => {
negated.hash(hasher);
}
Expr::Case(Case {
expr: _expr,
when_then_expr: _when_then_expr,
else_expr: _else_expr,
}) => {}
Expr::Cast(Cast {
expr: _expr,
data_type,
})
| Expr::TryCast(TryCast {
expr: _expr,
data_type,
}) => {
data_type.hash(hasher);
}
Expr::Sort(Sort {
expr: _expr,
asc,
nulls_first,
}) => {
asc.hash(hasher);
nulls_first.hash(hasher);
}
Expr::ScalarFunction(ScalarFunction { func, args: _args }) => {
func.hash(hasher);
}
Expr::AggregateFunction(AggregateFunction {
func_def,
args: _args,
distinct,
filter: _filter,
order_by: _order_by,
null_treatment,
}) => {
func_def.hash(hasher);
distinct.hash(hasher);
null_treatment.hash(hasher);
}
Expr::WindowFunction(WindowFunction {
fun,
args: _args,
partition_by: _partition_by,
order_by: _order_by,
window_frame,
null_treatment,
}) => {
fun.hash(hasher);
window_frame.hash(hasher);
null_treatment.hash(hasher);
}
Expr::InList(InList {
expr: _expr,
list: _list,
negated,
}) => {
negated.hash(hasher);
}
Expr::Exists(Exists { subquery, negated }) => {
subquery.hash(hasher);
negated.hash(hasher);
}
Expr::InSubquery(InSubquery {
expr: _expr,
subquery,
negated,
}) => {
subquery.hash(hasher);
negated.hash(hasher);
}
Expr::ScalarSubquery(subquery) => {
subquery.hash(hasher);
}
Expr::Wildcard { qualifier } => {
qualifier.hash(hasher);
}
Expr::GroupingSet(grouping_set) => {
mem::discriminant(grouping_set).hash(hasher);
match grouping_set {
GroupingSet::Rollup(_exprs) | GroupingSet::Cube(_exprs) => {}
GroupingSet::GroupingSets(_exprs) => {}
}
}
Expr::Placeholder(place_holder) => {
place_holder.hash(hasher);
}
Expr::OuterReferenceColumn(data_type, column) => {
data_type.hash(hasher);
column.hash(hasher);
}
Expr::Unnest(Unnest { expr: _expr }) => {}
};
}
}
fn rewrite_placeholder(expr: &mut Expr, other: &Expr, schema: &DFSchema) -> Result<()> {
if let Expr::Placeholder(Placeholder { id: _, data_type }) = expr {
if data_type.is_none() {
let other_dt = other.get_type(schema);
match other_dt {
Err(e) => {
Err(e.context(format!(
"Can not find type of {other} needed to infer type of {expr}"
)))?;
}
Ok(dt) => {
*data_type = Some(dt);
}
}
};
}
Ok(())
}
#[macro_export]
macro_rules! expr_vec_fmt {
( $ARRAY:expr ) => {{
$ARRAY
.iter()
.map(|e| format!("{e}"))
.collect::<Vec<String>>()
.join(", ")
}};
}
impl fmt::Display for Expr {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Expr::Alias(Alias { expr, name, .. }) => write!(f, "{expr} AS {name}"),
Expr::Column(c) => write!(f, "{c}"),
Expr::OuterReferenceColumn(_, c) => write!(f, "outer_ref({c})"),
Expr::ScalarVariable(_, var_names) => write!(f, "{}", var_names.join(".")),
Expr::Literal(v) => write!(f, "{v:?}"),
Expr::Case(case) => {
write!(f, "CASE ")?;
if let Some(e) = &case.expr {
write!(f, "{e} ")?;
}
for (w, t) in &case.when_then_expr {
write!(f, "WHEN {w} THEN {t} ")?;
}
if let Some(e) = &case.else_expr {
write!(f, "ELSE {e} ")?;
}
write!(f, "END")
}
Expr::Cast(Cast { expr, data_type }) => {
write!(f, "CAST({expr} AS {data_type:?})")
}
Expr::TryCast(TryCast { expr, data_type }) => {
write!(f, "TRY_CAST({expr} AS {data_type:?})")
}
Expr::Not(expr) => write!(f, "NOT {expr}"),
Expr::Negative(expr) => write!(f, "(- {expr})"),
Expr::IsNull(expr) => write!(f, "{expr} IS NULL"),
Expr::IsNotNull(expr) => write!(f, "{expr} IS NOT NULL"),
Expr::IsTrue(expr) => write!(f, "{expr} IS TRUE"),
Expr::IsFalse(expr) => write!(f, "{expr} IS FALSE"),
Expr::IsUnknown(expr) => write!(f, "{expr} IS UNKNOWN"),
Expr::IsNotTrue(expr) => write!(f, "{expr} IS NOT TRUE"),
Expr::IsNotFalse(expr) => write!(f, "{expr} IS NOT FALSE"),
Expr::IsNotUnknown(expr) => write!(f, "{expr} IS NOT UNKNOWN"),
Expr::Exists(Exists {
subquery,
negated: true,
}) => write!(f, "NOT EXISTS ({subquery:?})"),
Expr::Exists(Exists {
subquery,
negated: false,
}) => write!(f, "EXISTS ({subquery:?})"),
Expr::InSubquery(InSubquery {
expr,
subquery,
negated: true,
}) => write!(f, "{expr} NOT IN ({subquery:?})"),
Expr::InSubquery(InSubquery {
expr,
subquery,
negated: false,
}) => write!(f, "{expr} IN ({subquery:?})"),
Expr::ScalarSubquery(subquery) => write!(f, "({subquery:?})"),
Expr::BinaryExpr(expr) => write!(f, "{expr}"),
Expr::Sort(Sort {
expr,
asc,
nulls_first,
}) => {
if *asc {
write!(f, "{expr} ASC")?;
} else {
write!(f, "{expr} DESC")?;
}
if *nulls_first {
write!(f, " NULLS FIRST")
} else {
write!(f, " NULLS LAST")
}
}
Expr::ScalarFunction(fun) => {
fmt_function(f, fun.name(), false, &fun.args, true)
}
Expr::WindowFunction(WindowFunction {
fun,
args,
partition_by,
order_by,
window_frame,
null_treatment,
}) => {
fmt_function(f, &fun.to_string(), false, args, true)?;
if let Some(nt) = null_treatment {
write!(f, "{}", nt)?;
}
if !partition_by.is_empty() {
write!(f, " PARTITION BY [{}]", expr_vec_fmt!(partition_by))?;
}
if !order_by.is_empty() {
write!(f, " ORDER BY [{}]", expr_vec_fmt!(order_by))?;
}
write!(
f,
" {} BETWEEN {} AND {}",
window_frame.units, window_frame.start_bound, window_frame.end_bound
)?;
Ok(())
}
Expr::AggregateFunction(AggregateFunction {
func_def,
distinct,
ref args,
filter,
order_by,
null_treatment,
..
}) => {
fmt_function(f, func_def.name(), *distinct, args, true)?;
if let Some(nt) = null_treatment {
write!(f, " {}", nt)?;
}
if let Some(fe) = filter {
write!(f, " FILTER (WHERE {fe})")?;
}
if let Some(ob) = order_by {
write!(f, " ORDER BY [{}]", expr_vec_fmt!(ob))?;
}
Ok(())
}
Expr::Between(Between {
expr,
negated,
low,
high,
}) => {
if *negated {
write!(f, "{expr} NOT BETWEEN {low} AND {high}")
} else {
write!(f, "{expr} BETWEEN {low} AND {high}")
}
}
Expr::Like(Like {
negated,
expr,
pattern,
escape_char,
case_insensitive,
}) => {
write!(f, "{expr}")?;
let op_name = if *case_insensitive { "ILIKE" } else { "LIKE" };
if *negated {
write!(f, " NOT")?;
}
if let Some(char) = escape_char {
write!(f, " {op_name} {pattern} ESCAPE '{char}'")
} else {
write!(f, " {op_name} {pattern}")
}
}
Expr::SimilarTo(Like {
negated,
expr,
pattern,
escape_char,
case_insensitive: _,
}) => {
write!(f, "{expr}")?;
if *negated {
write!(f, " NOT")?;
}
if let Some(char) = escape_char {
write!(f, " SIMILAR TO {pattern} ESCAPE '{char}'")
} else {
write!(f, " SIMILAR TO {pattern}")
}
}
Expr::InList(InList {
expr,
list,
negated,
}) => {
if *negated {
write!(f, "{expr} NOT IN ([{}])", expr_vec_fmt!(list))
} else {
write!(f, "{expr} IN ([{}])", expr_vec_fmt!(list))
}
}
Expr::Wildcard { qualifier } => match qualifier {
Some(qualifier) => write!(f, "{qualifier}.*"),
None => write!(f, "*"),
},
Expr::GroupingSet(grouping_sets) => match grouping_sets {
GroupingSet::Rollup(exprs) => {
write!(f, "ROLLUP ({})", expr_vec_fmt!(exprs))
}
GroupingSet::Cube(exprs) => {
write!(f, "CUBE ({})", expr_vec_fmt!(exprs))
}
GroupingSet::GroupingSets(lists_of_exprs) => {
write!(
f,
"GROUPING SETS ({})",
lists_of_exprs
.iter()
.map(|exprs| format!("({})", expr_vec_fmt!(exprs)))
.collect::<Vec<String>>()
.join(", ")
)
}
},
Expr::Placeholder(Placeholder { id, .. }) => write!(f, "{id}"),
Expr::Unnest(Unnest { expr }) => {
write!(f, "UNNEST({expr:?})")
}
}
}
}
fn fmt_function(
f: &mut fmt::Formatter,
fun: &str,
distinct: bool,
args: &[Expr],
display: bool,
) -> fmt::Result {
let args: Vec<String> = match display {
true => args.iter().map(|arg| format!("{arg}")).collect(),
false => args.iter().map(|arg| format!("{arg:?}")).collect(),
};
let distinct_str = match distinct {
true => "DISTINCT ",
false => "",
};
write!(f, "{}({}{})", fun, distinct_str, args.join(", "))
}
fn write_function_name<W: Write>(
w: &mut W,
fun: &str,
distinct: bool,
args: &[Expr],
) -> Result<()> {
write!(w, "{}(", fun)?;
if distinct {
w.write_str("DISTINCT ")?;
}
write_names_join(w, args, ",")?;
w.write_str(")")?;
Ok(())
}
pub(crate) fn create_name(e: &Expr) -> Result<String> {
let mut s = String::new();
write_name(&mut s, e)?;
Ok(s)
}
fn write_name<W: Write>(w: &mut W, e: &Expr) -> Result<()> {
match e {
Expr::Alias(Alias { name, .. }) => write!(w, "{}", name)?,
Expr::Column(c) => write!(w, "{}", c.flat_name())?,
Expr::OuterReferenceColumn(_, c) => write!(w, "outer_ref({})", c.flat_name())?,
Expr::ScalarVariable(_, variable_names) => {
write!(w, "{}", variable_names.join("."))?
}
Expr::Literal(value) => write!(w, "{value:?}")?,
Expr::BinaryExpr(binary_expr) => {
write_name(w, binary_expr.left.as_ref())?;
write!(w, " {} ", binary_expr.op)?;
write_name(w, binary_expr.right.as_ref())?;
}
Expr::Like(Like {
negated,
expr,
pattern,
escape_char,
case_insensitive,
}) => {
write!(
w,
"{} {}{} {}",
expr,
if *negated { "NOT " } else { "" },
if *case_insensitive { "ILIKE" } else { "LIKE" },
pattern,
)?;
if let Some(char) = escape_char {
write!(w, " CHAR '{char}'")?;
}
}
Expr::SimilarTo(Like {
negated,
expr,
pattern,
escape_char,
case_insensitive: _,
}) => {
write!(
w,
"{} {} {}",
expr,
if *negated {
"NOT SIMILAR TO"
} else {
"SIMILAR TO"
},
pattern,
)?;
if let Some(char) = escape_char {
write!(w, " CHAR '{char}'")?;
}
}
Expr::Case(case) => {
write!(w, "CASE ")?;
if let Some(e) = &case.expr {
write_name(w, e)?;
w.write_str(" ")?;
}
for (when, then) in &case.when_then_expr {
w.write_str("WHEN ")?;
write_name(w, when)?;
w.write_str(" THEN ")?;
write_name(w, then)?;
w.write_str(" ")?;
}
if let Some(e) = &case.else_expr {
w.write_str("ELSE ")?;
write_name(w, e)?;
w.write_str(" ")?;
}
w.write_str("END")?;
}
Expr::Cast(Cast { expr, .. }) => {
write_name(w, expr)?;
}
Expr::TryCast(TryCast { expr, .. }) => {
write_name(w, expr)?;
}
Expr::Not(expr) => {
w.write_str("NOT ")?;
write_name(w, expr)?;
}
Expr::Negative(expr) => {
w.write_str("(- ")?;
write_name(w, expr)?;
w.write_str(")")?;
}
Expr::IsNull(expr) => {
write_name(w, expr)?;
w.write_str(" IS NULL")?;
}
Expr::IsNotNull(expr) => {
write_name(w, expr)?;
w.write_str(" IS NOT NULL")?;
}
Expr::IsTrue(expr) => {
write_name(w, expr)?;
w.write_str(" IS TRUE")?;
}
Expr::IsFalse(expr) => {
write_name(w, expr)?;
w.write_str(" IS FALSE")?;
}
Expr::IsUnknown(expr) => {
write_name(w, expr)?;
w.write_str(" IS UNKNOWN")?;
}
Expr::IsNotTrue(expr) => {
write_name(w, expr)?;
w.write_str(" IS NOT TRUE")?;
}
Expr::IsNotFalse(expr) => {
write_name(w, expr)?;
w.write_str(" IS NOT FALSE")?;
}
Expr::IsNotUnknown(expr) => {
write_name(w, expr)?;
w.write_str(" IS NOT UNKNOWN")?;
}
Expr::Exists(Exists { negated: true, .. }) => w.write_str("NOT EXISTS")?,
Expr::Exists(Exists { negated: false, .. }) => w.write_str("EXISTS")?,
Expr::InSubquery(InSubquery { negated: true, .. }) => w.write_str("NOT IN")?,
Expr::InSubquery(InSubquery { negated: false, .. }) => w.write_str("IN")?,
Expr::ScalarSubquery(subquery) => {
w.write_str(subquery.subquery.schema().field(0).name().as_str())?;
}
Expr::Unnest(Unnest { expr }) => {
w.write_str("unnest(")?;
write_name(w, expr)?;
w.write_str(")")?;
}
Expr::ScalarFunction(fun) => {
w.write_str(fun.func.display_name(&fun.args)?.as_str())?;
}
Expr::WindowFunction(WindowFunction {
fun,
args,
window_frame,
partition_by,
order_by,
null_treatment,
}) => {
write_function_name(w, &fun.to_string(), false, args)?;
if let Some(nt) = null_treatment {
w.write_str(" ")?;
write!(w, "{}", nt)?;
}
if !partition_by.is_empty() {
w.write_str(" ")?;
write!(w, "PARTITION BY [{}]", expr_vec_fmt!(partition_by))?;
}
if !order_by.is_empty() {
w.write_str(" ")?;
write!(w, "ORDER BY [{}]", expr_vec_fmt!(order_by))?;
}
w.write_str(" ")?;
write!(w, "{window_frame}")?;
}
Expr::AggregateFunction(AggregateFunction {
func_def,
distinct,
args,
filter,
order_by,
null_treatment,
}) => {
write_function_name(w, func_def.name(), *distinct, args)?;
if let Some(fe) = filter {
write!(w, " FILTER (WHERE {fe})")?;
};
if let Some(order_by) = order_by {
write!(w, " ORDER BY [{}]", expr_vec_fmt!(order_by))?;
};
if let Some(nt) = null_treatment {
write!(w, " {}", nt)?;
}
}
Expr::GroupingSet(grouping_set) => match grouping_set {
GroupingSet::Rollup(exprs) => {
write!(w, "ROLLUP (")?;
write_names(w, exprs.as_slice())?;
write!(w, ")")?;
}
GroupingSet::Cube(exprs) => {
write!(w, "CUBE (")?;
write_names(w, exprs.as_slice())?;
write!(w, ")")?;
}
GroupingSet::GroupingSets(lists_of_exprs) => {
write!(w, "GROUPING SETS (")?;
for (i, exprs) in lists_of_exprs.iter().enumerate() {
if i != 0 {
write!(w, ", ")?;
}
write!(w, "(")?;
write_names(w, exprs.as_slice())?;
write!(w, ")")?;
}
write!(w, ")")?;
}
},
Expr::InList(InList {
expr,
list,
negated,
}) => {
write_name(w, expr)?;
let list = list.iter().map(create_name);
if *negated {
write!(w, " NOT IN ({list:?})")?;
} else {
write!(w, " IN ({list:?})")?;
}
}
Expr::Between(Between {
expr,
negated,
low,
high,
}) => {
write_name(w, expr)?;
if *negated {
write!(w, " NOT BETWEEN ")?;
} else {
write!(w, " BETWEEN ")?;
}
write_name(w, low)?;
write!(w, " AND ")?;
write_name(w, high)?;
}
Expr::Sort { .. } => {
return internal_err!("Create name does not support sort expression")
}
Expr::Wildcard { qualifier } => match qualifier {
Some(qualifier) => {
return internal_err!(
"Create name does not support qualified wildcard, got {qualifier}"
)
}
None => write!(w, "*")?,
},
Expr::Placeholder(Placeholder { id, .. }) => write!(w, "{}", id)?,
};
Ok(())
}
fn write_names<W: Write>(w: &mut W, exprs: &[Expr]) -> Result<()> {
exprs.iter().try_for_each(|e| write_name(w, e))
}
fn write_names_join<W: Write>(w: &mut W, exprs: &[Expr], sep: &str) -> Result<()> {
let mut iter = exprs.iter();
if let Some(first_arg) = iter.next() {
write_name(w, first_arg)?;
}
for a in iter {
w.write_str(sep)?;
write_name(w, a)?;
}
Ok(())
}
#[cfg(test)]
mod test {
use crate::expr_fn::col;
use crate::{case, lit, ColumnarValue, ScalarUDF, ScalarUDFImpl, Volatility};
use std::any::Any;
#[test]
fn format_case_when() -> Result<()> {
let expr = case(col("a"))
.when(lit(1), lit(true))
.when(lit(0), lit(false))
.otherwise(lit(ScalarValue::Null))?;
let expected = "CASE a WHEN Int32(1) THEN Boolean(true) WHEN Int32(0) THEN Boolean(false) ELSE NULL END";
assert_eq!(expected, expr.canonical_name());
assert_eq!(expected, format!("{expr}"));
assert_eq!(expected, expr.display_name()?);
Ok(())
}
#[test]
fn format_cast() -> Result<()> {
let expr = Expr::Cast(Cast {
expr: Box::new(Expr::Literal(ScalarValue::Float32(Some(1.23)))),
data_type: DataType::Utf8,
});
let expected_canonical = "CAST(Float32(1.23) AS Utf8)";
assert_eq!(expected_canonical, expr.canonical_name());
assert_eq!(expected_canonical, format!("{expr}"));
assert_eq!("Float32(1.23)", expr.display_name()?);
Ok(())
}
#[test]
fn test_partial_ord() {
let exp1 = col("a") + lit(1);
let exp2 = col("a") + lit(2);
let exp3 = !(col("a") + lit(2));
let greater_or_equal = exp1 >= exp2;
assert_eq!(exp1 < exp2, !greater_or_equal);
let greater_or_equal = exp3 >= exp2;
assert_eq!(exp3 < exp2, !greater_or_equal);
}
#[test]
fn test_collect_expr() -> Result<()> {
{
let expr = &Expr::Cast(Cast::new(Box::new(col("a")), DataType::Float64));
let columns = expr.column_refs();
assert_eq!(1, columns.len());
assert!(columns.contains(&Column::from_name("a")));
}
{
let expr = col("a") + col("b") + lit(1);
let columns = expr.column_refs();
assert_eq!(2, columns.len());
assert!(columns.contains(&Column::from_name("a")));
assert!(columns.contains(&Column::from_name("b")));
}
Ok(())
}
#[test]
fn test_logical_ops() {
assert_eq!(
format!("{}", lit(1u32).eq(lit(2u32))),
"UInt32(1) = UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).not_eq(lit(2u32))),
"UInt32(1) != UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).gt(lit(2u32))),
"UInt32(1) > UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).gt_eq(lit(2u32))),
"UInt32(1) >= UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).lt(lit(2u32))),
"UInt32(1) < UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).lt_eq(lit(2u32))),
"UInt32(1) <= UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).and(lit(2u32))),
"UInt32(1) AND UInt32(2)"
);
assert_eq!(
format!("{}", lit(1u32).or(lit(2u32))),
"UInt32(1) OR UInt32(2)"
);
}
#[test]
fn test_is_volatile_scalar_func() {
#[derive(Debug)]
struct TestScalarUDF {
signature: Signature,
}
impl ScalarUDFImpl for TestScalarUDF {
fn as_any(&self) -> &dyn Any {
self
}
fn name(&self) -> &str {
"TestScalarUDF"
}
fn signature(&self) -> &Signature {
&self.signature
}
fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
Ok(DataType::Utf8)
}
fn invoke(&self, _args: &[ColumnarValue]) -> Result<ColumnarValue> {
Ok(ColumnarValue::Scalar(ScalarValue::from("a")))
}
}
let udf = Arc::new(ScalarUDF::from(TestScalarUDF {
signature: Signature::uniform(1, vec![DataType::Float32], Volatility::Stable),
}));
assert_ne!(udf.signature().volatility, Volatility::Volatile);
let udf = Arc::new(ScalarUDF::from(TestScalarUDF {
signature: Signature::uniform(
1,
vec![DataType::Float32],
Volatility::Volatile,
),
}));
assert_eq!(udf.signature().volatility, Volatility::Volatile);
}
use super::*;
#[test]
fn test_first_value_return_type() -> Result<()> {
let fun = find_df_window_func("first_value").unwrap();
let observed = fun.return_type(&[DataType::Utf8], &[true])?;
assert_eq!(DataType::Utf8, observed);
let observed = fun.return_type(&[DataType::UInt64], &[true])?;
assert_eq!(DataType::UInt64, observed);
Ok(())
}
#[test]
fn test_last_value_return_type() -> Result<()> {
let fun = find_df_window_func("last_value").unwrap();
let observed = fun.return_type(&[DataType::Utf8], &[true])?;
assert_eq!(DataType::Utf8, observed);
let observed = fun.return_type(&[DataType::Float64], &[true])?;
assert_eq!(DataType::Float64, observed);
Ok(())
}
#[test]
fn test_lead_return_type() -> Result<()> {
let fun = find_df_window_func("lead").unwrap();
let observed = fun.return_type(&[DataType::Utf8], &[true])?;
assert_eq!(DataType::Utf8, observed);
let observed = fun.return_type(&[DataType::Float64], &[true])?;
assert_eq!(DataType::Float64, observed);
Ok(())
}
#[test]
fn test_lag_return_type() -> Result<()> {
let fun = find_df_window_func("lag").unwrap();
let observed = fun.return_type(&[DataType::Utf8], &[true])?;
assert_eq!(DataType::Utf8, observed);
let observed = fun.return_type(&[DataType::Float64], &[true])?;
assert_eq!(DataType::Float64, observed);
Ok(())
}
#[test]
fn test_nth_value_return_type() -> Result<()> {
let fun = find_df_window_func("nth_value").unwrap();
let observed =
fun.return_type(&[DataType::Utf8, DataType::UInt64], &[true, true])?;
assert_eq!(DataType::Utf8, observed);
let observed =
fun.return_type(&[DataType::Float64, DataType::UInt64], &[true, true])?;
assert_eq!(DataType::Float64, observed);
Ok(())
}
#[test]
fn test_percent_rank_return_type() -> Result<()> {
let fun = find_df_window_func("percent_rank").unwrap();
let observed = fun.return_type(&[], &[])?;
assert_eq!(DataType::Float64, observed);
Ok(())
}
#[test]
fn test_cume_dist_return_type() -> Result<()> {
let fun = find_df_window_func("cume_dist").unwrap();
let observed = fun.return_type(&[], &[])?;
assert_eq!(DataType::Float64, observed);
Ok(())
}
#[test]
fn test_ntile_return_type() -> Result<()> {
let fun = find_df_window_func("ntile").unwrap();
let observed = fun.return_type(&[DataType::Int16], &[true])?;
assert_eq!(DataType::UInt64, observed);
Ok(())
}
#[test]
fn test_window_function_case_insensitive() -> Result<()> {
let names = vec![
"row_number",
"rank",
"dense_rank",
"percent_rank",
"cume_dist",
"ntile",
"lag",
"lead",
"first_value",
"last_value",
"nth_value",
"min",
"max",
];
for name in names {
let fun = find_df_window_func(name).unwrap();
let fun2 = find_df_window_func(name.to_uppercase().as_str()).unwrap();
assert_eq!(fun, fun2);
if fun.to_string() == "first_value" || fun.to_string() == "last_value" {
assert_eq!(fun.to_string(), name);
} else {
assert_eq!(fun.to_string(), name.to_uppercase());
}
}
Ok(())
}
#[test]
fn test_find_df_window_function() {
assert_eq!(
find_df_window_func("max"),
Some(WindowFunctionDefinition::AggregateFunction(
aggregate_function::AggregateFunction::Max
))
);
assert_eq!(
find_df_window_func("min"),
Some(WindowFunctionDefinition::AggregateFunction(
aggregate_function::AggregateFunction::Min
))
);
assert_eq!(
find_df_window_func("cume_dist"),
Some(WindowFunctionDefinition::BuiltInWindowFunction(
built_in_window_function::BuiltInWindowFunction::CumeDist
))
);
assert_eq!(
find_df_window_func("first_value"),
Some(WindowFunctionDefinition::BuiltInWindowFunction(
built_in_window_function::BuiltInWindowFunction::FirstValue
))
);
assert_eq!(
find_df_window_func("LAST_value"),
Some(WindowFunctionDefinition::BuiltInWindowFunction(
built_in_window_function::BuiltInWindowFunction::LastValue
))
);
assert_eq!(
find_df_window_func("LAG"),
Some(WindowFunctionDefinition::BuiltInWindowFunction(
built_in_window_function::BuiltInWindowFunction::Lag
))
);
assert_eq!(
find_df_window_func("LEAD"),
Some(WindowFunctionDefinition::BuiltInWindowFunction(
built_in_window_function::BuiltInWindowFunction::Lead
))
);
assert_eq!(find_df_window_func("not_exist"), None)
}
}