4061292542

Build Type

github

Committed by GitHub

Commit Message

Merge 069171d20 into de98caa91

Pull Request Pull Request #729: feat: Implement multi-way JOIN

Run Details

1356 of 1356 new or added lines in 10 files covered. (100.0%)

24817 of 38526 relevant lines covered (64.42%)

39509.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

43.95

/dozer-sql/src/pipeline/expression/builder.rs

use dozer_types::{
    ordered_float::OrderedFloat,
    types::{Field, FieldDefinition, Schema, SourceDefinition},
};

use sqlparser::ast::{
    BinaryOperator as SqlBinaryOperator, DataType, Expr as SqlExpr, Expr, Function, FunctionArg,
    FunctionArgExpr, Ident, TrimWhereField, UnaryOperator as SqlUnaryOperator, Value as SqlValue,
};

use crate::pipeline::errors::PipelineError;
use crate::pipeline::errors::PipelineError::{
    AmbiguousFieldIdentifier, IllegalFieldIdentifier, UnknownFieldIdentifier,
};
use crate::pipeline::expression::aggregate::AggregateFunctionType;
use crate::pipeline::expression::builder::PipelineError::InvalidArgument;
use crate::pipeline::expression::builder::PipelineError::InvalidExpression;
use crate::pipeline::expression::builder::PipelineError::InvalidOperator;
use crate::pipeline::expression::builder::PipelineError::InvalidValue;
use crate::pipeline::expression::execution::Expression;
use crate::pipeline::expression::execution::Expression::ScalarFunction;
use crate::pipeline::expression::operator::{BinaryOperatorType, UnaryOperatorType};
use crate::pipeline::expression::scalar::common::ScalarFunctionType;
use crate::pipeline::expression::scalar::string::TrimType;

use super::cast::CastOperatorType;

pub type Bypass = bool;

pub enum BuilderExpressionType {
    PreAggregation,
    Aggregation,
    // PostAggregation,
    FullExpression,
}
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub struct NameOrAlias(pub String, pub Option<String>);

pub struct ExpressionBuilder;

impl ExpressionBuilder {
    pub fn build(
        &self,
        expression_type: &BuilderExpressionType,
        sql_expression: &SqlExpr,
        schema: &Schema,
    ) -> Result<Box<Expression>, PipelineError> {
        let (expression, _bypass) =
            self.parse_sql_expression(expression_type, sql_expression, schema)?;
        Ok(expression)
    }

    pub fn parse_sql_expression(
        &self,
        expression_type: &BuilderExpressionType,
        expression: &SqlExpr,
        schema: &Schema,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        match expression {
            SqlExpr::Trim {
                expr,
                trim_where,
                trim_what,
            } => self.parse_sql_trim_function(expression_type, expr, trim_where, trim_what, schema),
            SqlExpr::Identifier(ident) => Ok((parse_sql_column(&[ident.clone()], schema)?, false)),
            SqlExpr::CompoundIdentifier(ident) => Ok((parse_sql_column(ident, schema)?, false)),
            SqlExpr::Value(SqlValue::Number(n, _)) => self.parse_sql_number(n),
            SqlExpr::Value(SqlValue::Null) => {
                Ok((Box::new(Expression::Literal(Field::Null)), false))
            }
            SqlExpr::Value(SqlValue::SingleQuotedString(s) | SqlValue::DoubleQuotedString(s)) => {
                parse_sql_string(s)
            }
            SqlExpr::UnaryOp { expr, op } => {
                self.parse_sql_unary_op(expression_type, op, expr, schema)
            }
            SqlExpr::BinaryOp { left, op, right } => {
                self.parse_sql_binary_op(expression_type, left, op, right, schema)
            }
            SqlExpr::Nested(expr) => self.parse_sql_expression(expression_type, expr, schema),
            SqlExpr::Function(sql_function) => match expression_type {
                BuilderExpressionType::PreAggregation => self.parse_sql_function_pre_aggregation(
                    expression_type,
                    sql_function,
                    schema,
                    expression,
                ),
                BuilderExpressionType::Aggregation => self.parse_sql_function_aggregation(
                    expression_type,
                    sql_function,
                    schema,
                    expression,
                ),
                // ExpressionType::PostAggregation => todo!(),
                BuilderExpressionType::FullExpression => {
                    self.parse_sql_function(expression_type, sql_function, schema, expression)
                }
            },
            SqlExpr::Like {
                negated,
                expr,
                pattern,
                escape_char,
            } => self.parse_sql_like_operator(
                expression_type,
                negated,
                expr,
                pattern,
                escape_char,
                schema,
            ),
            SqlExpr::Cast { expr, data_type } => {
                self.parse_sql_cast_operator(expression_type, expr, data_type, schema)
            }
            _ => Err(InvalidExpression(format!("{expression:?}"))),
        }
    }

    fn parse_sql_trim_function(
        &self,
        expression_type: &BuilderExpressionType,
        expr: &Expr,
        trim_where: &Option<TrimWhereField>,
        trim_what: &Option<Box<Expr>>,
        schema: &Schema,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let arg = self.parse_sql_expression(expression_type, expr, schema)?.0;
        let what = match trim_what {
            Some(e) => Some(self.parse_sql_expression(expression_type, e, schema)?.0),
            _ => None,
        };
        let typ = trim_where.as_ref().map(|e| match e {
            TrimWhereField::Both => TrimType::Both,
            TrimWhereField::Leading => TrimType::Leading,
            TrimWhereField::Trailing => TrimType::Trailing,
        });
        Ok((Box::new(Expression::Trim { arg, what, typ }), false))
    }

    fn parse_sql_function(
        &self,
        expression_type: &BuilderExpressionType,
        sql_function: &Function,
        schema: &Schema,
        expression: &SqlExpr,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let name = sql_function.name.to_string().to_lowercase();
        if let Ok(function) = ScalarFunctionType::new(&name) {
            let mut arg_exprs = vec![];
            for arg in &sql_function.args {
                let r = self.parse_sql_function_arg(expression_type, arg, schema);
                match r {
                    Ok(result) => {
                        if result.1 {
                            return Ok(result);
                        } else {
                            arg_exprs.push(*result.0);
                        }
                    }
                    Err(error) => {
                        return Err(error);
                    }
                }
            }

            return Ok((
                Box::new(ScalarFunction {
                    fun: function,
                    args: arg_exprs,
                }),
                false,
            ));
        };
        if AggregateFunctionType::new(&name).is_ok() {
            let arg = sql_function.args.first().unwrap();
            let r = self.parse_sql_function_arg(expression_type, arg, schema)?;
            return Ok((r.0, false)); // switch bypass to true, since the argument of this Aggregation must be the final result
        };
        Err(InvalidExpression(format!("{expression:?}")))
    }

    fn parse_sql_function_pre_aggregation(
        &self,
        expression_type: &BuilderExpressionType,
        sql_function: &Function,
        schema: &Schema,
        expression: &SqlExpr,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let name = sql_function.name.to_string().to_lowercase();

        if let Ok(function) = ScalarFunctionType::new(&name) {
            let mut arg_exprs = vec![];
            for arg in &sql_function.args {
                let r = self.parse_sql_function_arg(expression_type, arg, schema);
                match r {
                    Ok(result) => {
                        if result.1 {
                            return Ok(result);
                        } else {
                            arg_exprs.push(*result.0);
                        }
                    }
                    Err(error) => {
                        return Err(error);
                    }
                }
            }

            return Ok((
                Box::new(ScalarFunction {
                    fun: function,
                    args: arg_exprs,
                }),
                false,
            ));
        };
        if AggregateFunctionType::new(&name).is_ok() {
            let arg = sql_function.args.first().unwrap();
            let r = self.parse_sql_function_arg(expression_type, arg, schema)?;
            return Ok((r.0, true)); // switch bypass to true, since the argument of this Aggregation must be the final result
        };
        Err(InvalidExpression(format!("{expression:?}")))
    }

    fn parse_sql_function_aggregation(
        &self,
        expression_type: &BuilderExpressionType,
        sql_function: &Function,
        schema: &Schema,
        expression: &SqlExpr,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let name = sql_function.name.to_string().to_lowercase();

        if let Ok(function) = ScalarFunctionType::new(&name) {
            let mut arg_exprs = vec![];
            for arg in &sql_function.args {
                let r = self.parse_sql_function_arg(expression_type, arg, schema);
                match r {
                    Ok(result) => {
                        if result.1 {
                            return Ok(result);
                        } else {
                            arg_exprs.push(*result.0);
                        }
                    }
                    Err(error) => {
                        return Err(error);
                    }
                }
            }

            return Ok((
                Box::new(ScalarFunction {
                    fun: function,
                    args: arg_exprs,
                }),
                false,
            ));
        };

        if let Ok(function) = AggregateFunctionType::new(&name) {
            let mut arg_exprs = vec![];
            for arg in &sql_function.args {
                let r = self.parse_sql_function_arg(expression_type, arg, schema);
                match r {
                    Ok(result) => {
                        if result.1 {
                            return Ok(result);
                        } else {
                            arg_exprs.push(*result.0);
                        }
                    }
                    Err(error) => {
                        return Err(error);
                    }
                }
            }

            return Ok((
                Box::new(Expression::AggregateFunction {
                    fun: function,
                    args: arg_exprs,
                }),
                true, // switch bypass to true, since this Aggregation must be the final result
            ));
        };

        Err(InvalidExpression(format!(
            "Unsupported Expression: {expression:?}"
        )))
    }

    fn parse_sql_function_arg(
        &self,
        expression_type: &BuilderExpressionType,
        argument: &FunctionArg,
        schema: &Schema,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        match argument {
            FunctionArg::Named {
                name: _,
                arg: FunctionArgExpr::Expr(arg),
            } => self.parse_sql_expression(expression_type, arg, schema),
            FunctionArg::Named {
                name: _,
                arg: FunctionArgExpr::Wildcard,
            } => Err(InvalidArgument(format!("{argument:?}"))),
            FunctionArg::Unnamed(FunctionArgExpr::Expr(arg)) => {
                self.parse_sql_expression(expression_type, arg, schema)
            }
            FunctionArg::Unnamed(FunctionArgExpr::Wildcard) => {
                Err(InvalidArgument(format!("{argument:?}")))
            }
            _ => Err(InvalidArgument(format!("{argument:?}"))),
        }
    }

    fn parse_sql_unary_op(
        &self,
        expression_type: &BuilderExpressionType,
        op: &SqlUnaryOperator,
        expr: &SqlExpr,
        schema: &Schema,
    ) -> Result<(Box<Expression>, Bypass), PipelineError> {
        let (arg, bypass) = self.parse_sql_expression(expression_type, expr, schema)?;
        if bypass {
            return Ok((arg, bypass));
        }

        let operator = match op {
            SqlUnaryOperator::Not => UnaryOperatorType::Not,
            SqlUnaryOperator::Plus => UnaryOperatorType::Plus,
            SqlUnaryOperator::Minus => UnaryOperatorType::Minus,
            _ => return Err(InvalidOperator(format!("{op:?}"))),
        };

        Ok((Box::new(Expression::UnaryOperator { operator, arg }), false))
    }

    fn parse_sql_binary_op(
        &self,
        expression_type: &BuilderExpressionType,
        left: &SqlExpr,
        op: &SqlBinaryOperator,
        right: &SqlExpr,
        schema: &Schema,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let (left_op, bypass_left) = self.parse_sql_expression(expression_type, left, schema)?;
        if bypass_left {
            return Ok((left_op, bypass_left));
        }
        let (right_op, bypass_right) = self.parse_sql_expression(expression_type, right, schema)?;
        if bypass_right {
            return Ok((right_op, bypass_right));
        }

        let operator = match op {
            SqlBinaryOperator::Gt => BinaryOperatorType::Gt,
            SqlBinaryOperator::GtEq => BinaryOperatorType::Gte,
            SqlBinaryOperator::Lt => BinaryOperatorType::Lt,
            SqlBinaryOperator::LtEq => BinaryOperatorType::Lte,
            SqlBinaryOperator::Eq => BinaryOperatorType::Eq,
            SqlBinaryOperator::NotEq => BinaryOperatorType::Ne,

            SqlBinaryOperator::Plus => BinaryOperatorType::Add,
            SqlBinaryOperator::Minus => BinaryOperatorType::Sub,
            SqlBinaryOperator::Multiply => BinaryOperatorType::Mul,
            SqlBinaryOperator::Divide => BinaryOperatorType::Div,
            SqlBinaryOperator::Modulo => BinaryOperatorType::Mod,

            SqlBinaryOperator::And => BinaryOperatorType::And,
            SqlBinaryOperator::Or => BinaryOperatorType::Or,

            // BinaryOperator::BitwiseAnd => ...
            // BinaryOperator::BitwiseOr => ...
            // BinaryOperator::StringConcat => ...
            _ => return Err(InvalidOperator(format!("{op:?}"))),
        };

        Ok((
            Box::new(Expression::BinaryOperator {
                left: left_op,
                operator,
                right: right_op,
            }),
            false,
        ))
    }

    fn parse_sql_number(&self, n: &str) -> Result<(Box<Expression>, Bypass), PipelineError> {
        match n.parse::<i64>() {
            Ok(n) => Ok((Box::new(Expression::Literal(Field::Int(n))), false)),
            Err(_) => match n.parse::<f64>() {
                Ok(f) => Ok((
                    Box::new(Expression::Literal(Field::Float(OrderedFloat(f)))),
                    false,
                )),
                Err(_) => Err(InvalidValue(n.to_string())),
            },
        }
    }

    fn parse_sql_like_operator(
        &self,
        expression_type: &BuilderExpressionType,
        negated: &bool,
        expr: &Expr,
        pattern: &Expr,
        escape_char: &Option<char>,
        schema: &Schema,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let arg = self.parse_sql_expression(expression_type, expr, schema)?;
        let pattern = self.parse_sql_expression(expression_type, pattern, schema)?;
        let like_expression = Box::new(Expression::Like {
            arg: arg.0,
            pattern: pattern.0,
            escape: *escape_char,
        });
        if *negated {
            Ok((
                Box::new(Expression::UnaryOperator {
                    operator: UnaryOperatorType::Not,
                    arg: like_expression,
                }),
                arg.1,
            ))
        } else {
            Ok((like_expression, arg.1))
        }
    }

    fn parse_sql_cast_operator(
        &self,
        expression_type: &BuilderExpressionType,
        expr: &Expr,
        data_type: &DataType,
        schema: &Schema,
    ) -> Result<(Box<Expression>, bool), PipelineError> {
        let expression = self.parse_sql_expression(expression_type, expr, schema)?;
        let cast_to = match data_type {
            DataType::Decimal(_) => CastOperatorType::Decimal,
            DataType::Binary(_) => CastOperatorType::Binary,
            DataType::Float(_) => CastOperatorType::Float,
            DataType::Int(_) => CastOperatorType::Int,
            DataType::Integer(_) => CastOperatorType::Int,
            DataType::UnsignedInt(_) => CastOperatorType::UInt,
            DataType::UnsignedInteger(_) => CastOperatorType::UInt,
            DataType::Boolean => CastOperatorType::Boolean,
            DataType::Date => CastOperatorType::Date,
            DataType::Timestamp(..) => CastOperatorType::Timestamp,
            DataType::Text => CastOperatorType::Text,
            DataType::String => CastOperatorType::String,
            DataType::Custom(name, ..) => {
                if name.to_string().to_lowercase() == "bson" {
                    CastOperatorType::Bson
                } else {
                    Err(PipelineError::InvalidFunction(format!(
                        "Unsupported Cast type {name}"
                    )))?
                }
            }
            _ => Err(PipelineError::InvalidFunction(format!(
                "Unsupported Cast type {data_type}"
            )))?,
        };
        Ok((
            Box::new(Expression::Cast {
                arg: expression.0,
                typ: cast_to,
            }),
            expression.1,
        ))
    }
}

pub fn fullname_from_ident(ident: &[Ident]) -> String {
    let mut ident_tokens = vec![];
    for token in ident.iter() {
        ident_tokens.push(token.value.clone());
    }
    ident_tokens.join(".")
}

fn parse_sql_string(s: &str) -> Result<(Box<Expression>, bool), PipelineError> {
    Ok((
        Box::new(Expression::Literal(Field::String(s.to_owned()))),
        false,
    ))
}

pub(crate) fn normalize_ident(id: &Ident) -> String {
    match id.quote_style {
        Some(_) => id.value.clone(),
        None => id.value.clone(),
    }
}

pub fn extend_schema_source_def(schema: &Schema, name: &NameOrAlias) -> Schema {
    let mut output_schema = schema.clone();
    let mut fields = vec![];
    for mut field in schema.clone().fields.into_iter() {
        if let Some(alias) = &name.1 {
            field.source = SourceDefinition::Alias {
                name: alias.to_string(),
            };
        }

        fields.push(field);
    }
    output_schema.fields = fields;

    output_schema
}

fn parse_sql_column(ident: &[Ident], schema: &Schema) -> Result<Box<Expression>, PipelineError> {
    let (src_field, src_table_or_alias, src_connection) = match ident.len() {
        1 => (&ident[0].value, None, None),
        2 => (&ident[1].value, Some(&ident[0].value), None),
        3 => (
            &ident[2].value,
            Some(&ident[1].value),
            Some(&ident[0].value),
        ),
        _ => {
            return Err(IllegalFieldIdentifier(
                ident
                    .iter()
                    .fold(String::new(), |a, b| a + "." + b.value.as_str()),
            ))
        }
    };

    let matching_by_field: Vec<(usize, &FieldDefinition)> = schema
        .fields
        .iter()
        .enumerate()
        .filter(|(_idx, f)| &f.name == src_field)
        .collect();

    match matching_by_field.len() {
        0 => Err(UnknownFieldIdentifier(
            ident
                .iter()
                .fold(String::new(), |a, b| a + "." + b.value.as_str()),
        )),
        1 => Ok(Box::new(Expression::Column {
            index: matching_by_field[0].0,
        })),
        _ => match src_table_or_alias {
            None => Err(AmbiguousFieldIdentifier(
                ident
                    .iter()
                    .fold(String::new(), |a, b| a + "." + b.value.as_str()),
            )),
            Some(src_table_or_alias) => {
                let matching_by_table_or_alias: Vec<(usize, &FieldDefinition)> = matching_by_field
                    .into_iter()
                    .filter(|(_idx, field)| match &field.source {
                        SourceDefinition::Alias { name } => name == src_table_or_alias,
                        SourceDefinition::Table {
                            name,
                            connection: _,
                        } => name == src_table_or_alias,
                        _ => false,
                    })
                    .collect();

                match matching_by_table_or_alias.len() {
                    0 => Err(UnknownFieldIdentifier(
                        ident
                            .iter()
                            .fold(String::new(), |a, b| a + "." + b.value.as_str()),
                    )),
                    1 => Ok(Box::new(Expression::Column {
                        index: matching_by_table_or_alias[0].0,
                    })),
                    _ => match src_connection {
                        None => Err(InvalidExpression(
                            ident
                                .iter()
                                .fold(String::new(), |a, b| a + "." + b.value.as_str()),
                        )),
                        Some(src_connection) => {
                            let matching_by_connection: Vec<(usize, &FieldDefinition)> =
                                matching_by_table_or_alias
                                    .into_iter()
                                    .filter(|(_idx, field)| match &field.source {
                                        SourceDefinition::Table {
                                            name: _,
                                            connection,
                                        } => connection == src_connection,
                                        _ => false,
                                    })
                                    .collect();

                            match matching_by_connection.len() {
                                0 => Err(UnknownFieldIdentifier(
                                    ident
                                        .iter()
                                        .fold(String::new(), |a, b| a + "." + b.value.as_str()),
                                )),
                                1 => Ok(Box::new(Expression::Column {
                                    index: matching_by_connection[0].0,
                                })),
                                _ => Err(InvalidExpression(
                                    ident
                                        .iter()
                                        .fold(String::new(), |a, b| a + "." + b.value.as_str()),
                                )),
                            }
                        }
                    },
                }
            }
        },
    }
}

1	use dozer_types::{
2	ordered_float::OrderedFloat,
3	types::{Field, FieldDefinition, Schema, SourceDefinition},
4	};
5
6	use sqlparser::ast::{
7	BinaryOperator as SqlBinaryOperator, DataType, Expr as SqlExpr, Expr, Function, FunctionArg,
8	FunctionArgExpr, Ident, TrimWhereField, UnaryOperator as SqlUnaryOperator, Value as SqlValue,
9	};
10
11	use crate::pipeline::errors::PipelineError;
12	use crate::pipeline::errors::PipelineError::{
13	AmbiguousFieldIdentifier, IllegalFieldIdentifier, UnknownFieldIdentifier,
14	};
15	use crate::pipeline::expression::aggregate::AggregateFunctionType;
16	use crate::pipeline::expression::builder::PipelineError::InvalidArgument;
17	use crate::pipeline::expression::builder::PipelineError::InvalidExpression;
18	use crate::pipeline::expression::builder::PipelineError::InvalidOperator;
19	use crate::pipeline::expression::builder::PipelineError::InvalidValue;
20	use crate::pipeline::expression::execution::Expression;
21	use crate::pipeline::expression::execution::Expression::ScalarFunction;
22	use crate::pipeline::expression::operator::{BinaryOperatorType, UnaryOperatorType};
23	use crate::pipeline::expression::scalar::common::ScalarFunctionType;
24	use crate::pipeline::expression::scalar::string::TrimType;
25
26	use super::cast::CastOperatorType;
27
28	pub type Bypass = bool;
29
30	pub enum BuilderExpressionType {
31	PreAggregation,
32	Aggregation,
33	// PostAggregation,
34	FullExpression,
35	}	×
36	#[derive(Debug, Clone, Hash, PartialEq, Eq)]	1,595✔
37	pub struct NameOrAlias(pub String, pub Option<String>);
38		×
39	pub struct ExpressionBuilder;	×
40		×
41	impl ExpressionBuilder {	×
42	pub fn build(	188✔
43	&self,	188✔
44	expression_type: &BuilderExpressionType,	188✔
45	sql_expression: &SqlExpr,	188✔
46	schema: &Schema,	188✔
47	) -> Result<Box<Expression>, PipelineError> {	188✔
48	let (expression, _bypass) =	188✔
49	self.parse_sql_expression(expression_type, sql_expression, schema)?;	188✔
50	Ok(expression)	188✔
51	}	188✔
52		×
53	pub fn parse_sql_expression(
54	&self,	×
55	expression_type: &BuilderExpressionType,	×
56	expression: &SqlExpr,	×
57	schema: &Schema,	×
58	) -> Result<(Box<Expression>, bool), PipelineError> {	×
59	match expression {	69✔
60	SqlExpr::Trim {	×
61	expr,	50✔
62	trim_where,	50✔
63	trim_what,	50✔
64	} => self.parse_sql_trim_function(expression_type, expr, trim_where, trim_what, schema),	50✔
65	SqlExpr::Identifier(ident) => Ok((parse_sql_column(&[ident.clone()], schema)?, false)),	2,555✔
66	SqlExpr::CompoundIdentifier(ident) => Ok((parse_sql_column(ident, schema)?, false)),	182✔
67	SqlExpr::Value(SqlValue::Number(n, _)) => self.parse_sql_number(n),	43✔
68	SqlExpr::Value(SqlValue::Null) => {	×
69	Ok((Box::new(Expression::Literal(Field::Null)), false))	×
70	}
71	SqlExpr::Value(SqlValue::SingleQuotedString(s) \| SqlValue::DoubleQuotedString(s)) => {	26✔
72	parse_sql_string(s)	26✔
73	}	×
74	SqlExpr::UnaryOp { expr, op } => {	×
75	self.parse_sql_unary_op(expression_type, op, expr, schema)	×
76	}	×
77	SqlExpr::BinaryOp { left, op, right } => {	79✔
78	self.parse_sql_binary_op(expression_type, left, op, right, schema)	79✔
79	}	×
80	SqlExpr::Nested(expr) => self.parse_sql_expression(expression_type, expr, schema),	12✔
81	SqlExpr::Function(sql_function) => match expression_type {	346✔
82	BuilderExpressionType::PreAggregation => self.parse_sql_function_pre_aggregation(	165✔
83	expression_type,	165✔
84	sql_function,	165✔
85	schema,	165✔
86	expression,	165✔
87	),	165✔
88	BuilderExpressionType::Aggregation => self.parse_sql_function_aggregation(	165✔
89	expression_type,	165✔
90	sql_function,	165✔
91	schema,	165✔
92	expression,	165✔
93	),	165✔
94	// ExpressionType::PostAggregation => todo!(),	×
95	BuilderExpressionType::FullExpression => {	×
96	self.parse_sql_function(expression_type, sql_function, schema, expression)	16✔
97	}	×
98	},	×
99	SqlExpr::Like {	×
100	negated,	×
101	expr,	×
102	pattern,	×
103	escape_char,	×
104	} => self.parse_sql_like_operator(	×
105	expression_type,	×
106	negated,	×
107	expr,	×
108	pattern,	×
109	escape_char,	×
110	schema,	×
111	),	×
112	SqlExpr::Cast { expr, data_type } => {	64✔
113	self.parse_sql_cast_operator(expression_type, expr, data_type, schema)	64✔
114	}	×
115	_ => Err(InvalidExpression(format!("{expression:?}"))),	×
116	}	×
117	}	3,357✔
118		×
119	fn parse_sql_trim_function(	50✔
120	&self,	50✔
121	expression_type: &BuilderExpressionType,	50✔
122	expr: &Expr,	50✔
123	trim_where: &Option<TrimWhereField>,	50✔
124	trim_what: &Option<Box<Expr>>,	50✔
125	schema: &Schema,	50✔
126	) -> Result<(Box<Expression>, bool), PipelineError> {	50✔
127	let arg = self.parse_sql_expression(expression_type, expr, schema)?.0;	50✔
128	let what = match trim_what {	50✔
129	Some(e) => Some(self.parse_sql_expression(expression_type, e, schema)?.0),	8✔
130	_ => None,	42✔
131	};	×
132	let typ = trim_where.as_ref().map(\|e\| match e {	50✔
133	TrimWhereField::Both => TrimType::Both,	2✔
134	TrimWhereField::Leading => TrimType::Leading,	2✔
135	TrimWhereField::Trailing => TrimType::Trailing,	2✔
136	});	50✔
137	Ok((Box::new(Expression::Trim { arg, what, typ }), false))	50✔
138	}	50✔
139		×
140	fn parse_sql_function(	16✔
141	&self,	16✔
142	expression_type: &BuilderExpressionType,	16✔
143	sql_function: &Function,	16✔
144	schema: &Schema,	16✔
145	expression: &SqlExpr,	16✔
146	) -> Result<(Box<Expression>, bool), PipelineError> {	16✔
147	let name = sql_function.name.to_string().to_lowercase();	16✔
148	if let Ok(function) = ScalarFunctionType::new(&name) {	16✔
149	let mut arg_exprs = vec![];	15✔
150	for arg in &sql_function.args {	37✔
151	let r = self.parse_sql_function_arg(expression_type, arg, schema);	22✔
152	match r {	22✔
153	Ok(result) => {	22✔
154	if result.1 {	22✔
155	return Ok(result);	×
156	} else {	22✔
157	arg_exprs.push(*result.0);	22✔
158	}	22✔
159	}	×
160	Err(error) => {	×
161	return Err(error);	×
162	}	×
163	}	×
164	}	×
165		×
166	return Ok((	15✔
167	Box::new(ScalarFunction {	15✔
168	fun: function,	15✔
169	args: arg_exprs,	15✔
170	}),	15✔
171	false,	15✔
172	));	15✔
173	};	1✔
174	if AggregateFunctionType::new(&name).is_ok() {	1✔
175	let arg = sql_function.args.first().unwrap();	1✔
176	let r = self.parse_sql_function_arg(expression_type, arg, schema)?;	1✔
177	return Ok((r.0, false)); // switch bypass to true, since the argument of this Aggregation must be the final result	1✔
178	};	×
179	Err(InvalidExpression(format!("{expression:?}")))	×
180	}	16✔
181		×
182	fn parse_sql_function_pre_aggregation(	165✔
183	&self,	165✔
184	expression_type: &BuilderExpressionType,	165✔
185	sql_function: &Function,	165✔
186	schema: &Schema,	165✔
187	expression: &SqlExpr,	165✔
188	) -> Result<(Box<Expression>, bool), PipelineError> {	165✔
189	let name = sql_function.name.to_string().to_lowercase();	165✔
190		×
191	if let Ok(function) = ScalarFunctionType::new(&name) {	165✔
192	let mut arg_exprs = vec![];	×
193	for arg in &sql_function.args {	×
194	let r = self.parse_sql_function_arg(expression_type, arg, schema);	×
195	match r {	×
196	Ok(result) => {	×
197	if result.1 {	×
198	return Ok(result);	×
199	} else {	×
200	arg_exprs.push(*result.0);	×
201	}	×
202	}	×
203	Err(error) => {	×
204	return Err(error);	×
205	}	×
206	}	×
207	}	×
208		×
209	return Ok((	×
210	Box::new(ScalarFunction {	×
211	fun: function,	×
212	args: arg_exprs,	×
213	}),	×
214	false,	×
215	));	×
216	};	165✔
217	if AggregateFunctionType::new(&name).is_ok() {	165✔
218	let arg = sql_function.args.first().unwrap();	165✔
219	let r = self.parse_sql_function_arg(expression_type, arg, schema)?;	165✔
220	return Ok((r.0, true)); // switch bypass to true, since the argument of this Aggregation must be the final result	165✔
221	};	×
222	Err(InvalidExpression(format!("{expression:?}")))	×
223	}	165✔
224		×
225	fn parse_sql_function_aggregation(	165✔
226	&self,	165✔
227	expression_type: &BuilderExpressionType,	165✔
228	sql_function: &Function,	165✔
229	schema: &Schema,	165✔
230	expression: &SqlExpr,	165✔
231	) -> Result<(Box<Expression>, bool), PipelineError> {	165✔
232	let name = sql_function.name.to_string().to_lowercase();	165✔
233		×
234	if let Ok(function) = ScalarFunctionType::new(&name) {	165✔
235	let mut arg_exprs = vec![];	×
236	for arg in &sql_function.args {	×
237	let r = self.parse_sql_function_arg(expression_type, arg, schema);	×
238	match r {	×
239	Ok(result) => {	×
240	if result.1 {	×
241	return Ok(result);	×
242	} else {	×
243	arg_exprs.push(*result.0);	×
244	}	×
245	}	×
246	Err(error) => {	×
247	return Err(error);	×
248	}	×
249	}	×
250	}	×
251		×
252	return Ok((	×
253	Box::new(ScalarFunction {	×
254	fun: function,	×
255	args: arg_exprs,	×
256	}),	×
257	false,	×
258	));	×
259	};	165✔
260
261	if let Ok(function) = AggregateFunctionType::new(&name) {	165✔
262	let mut arg_exprs = vec![];	165✔
263	for arg in &sql_function.args {	330✔
264	let r = self.parse_sql_function_arg(expression_type, arg, schema);	165✔
265	match r {	165✔
266	Ok(result) => {	165✔
267	if result.1 {	165✔
268	return Ok(result);	×
269	} else {	165✔
270	arg_exprs.push(*result.0);	165✔
271	}	165✔
272	}	×
273	Err(error) => {	×
274	return Err(error);	×
275	}	×
276	}	×
277	}	×
278		×
279	return Ok((	165✔
280	Box::new(Expression::AggregateFunction {	165✔
281	fun: function,	165✔
282	args: arg_exprs,	165✔
283	}),	165✔
284	true, // switch bypass to true, since this Aggregation must be the final result	165✔
285	));	165✔
286	};	×
287		×
288	Err(InvalidExpression(format!(	×
289	"Unsupported Expression: {expression:?}"	×
290	)))	×
291	}	165✔
292		×
293	fn parse_sql_function_arg(	×
294	&self,	×
295	expression_type: &BuilderExpressionType,	×
296	argument: &FunctionArg,	×
297	schema: &Schema,	×
298	) -> Result<(Box<Expression>, bool), PipelineError> {	×
299	match argument {	353✔
300	FunctionArg::Named {	×
301	name: _,	×
302	arg: FunctionArgExpr::Expr(arg),	×
303	} => self.parse_sql_expression(expression_type, arg, schema),	×
304	FunctionArg::Named {	×
305	name: _,	×
306	arg: FunctionArgExpr::Wildcard,	×
307	} => Err(InvalidArgument(format!("{argument:?}"))),	×
308	FunctionArg::Unnamed(FunctionArgExpr::Expr(arg)) => {	353✔
309	self.parse_sql_expression(expression_type, arg, schema)	353✔
310	}	×
311	FunctionArg::Unnamed(FunctionArgExpr::Wildcard) => {	×
312	Err(InvalidArgument(format!("{argument:?}")))	×
313	}	×
314	_ => Err(InvalidArgument(format!("{argument:?}"))),	×
315	}	×
316	}	353✔
317		×
318	fn parse_sql_unary_op(	×
319	&self,	×
320	expression_type: &BuilderExpressionType,	×
321	op: &SqlUnaryOperator,	×
322	expr: &SqlExpr,	×
323	schema: &Schema,	×
324	) -> Result<(Box<Expression>, Bypass), PipelineError> {	×
325	let (arg, bypass) = self.parse_sql_expression(expression_type, expr, schema)?;	×
326	if bypass {	×
327	return Ok((arg, bypass));	×
328	}	×
329		×
330	let operator = match op {	×
331	SqlUnaryOperator::Not => UnaryOperatorType::Not,	×
332	SqlUnaryOperator::Plus => UnaryOperatorType::Plus,	×
333	SqlUnaryOperator::Minus => UnaryOperatorType::Minus,	×
334	_ => return Err(InvalidOperator(format!("{op:?}"))),	×
335	};	×
336		×
337	Ok((Box::new(Expression::UnaryOperator { operator, arg }), false))	×
338	}	×
339		×
340	fn parse_sql_binary_op(	79✔
341	&self,	79✔
342	expression_type: &BuilderExpressionType,	79✔
343	left: &SqlExpr,	79✔
344	op: &SqlBinaryOperator,	79✔
345	right: &SqlExpr,	79✔
346	schema: &Schema,	79✔
347	) -> Result<(Box<Expression>, bool), PipelineError> {	79✔
348	let (left_op, bypass_left) = self.parse_sql_expression(expression_type, left, schema)?;	79✔
349	if bypass_left {	79✔
350	return Ok((left_op, bypass_left));	×
351	}	79✔
352	let (right_op, bypass_right) = self.parse_sql_expression(expression_type, right, schema)?;	79✔
353	if bypass_right {	79✔
354	return Ok((right_op, bypass_right));	×
355	}	79✔
356		×
357	let operator = match op {	79✔
358	SqlBinaryOperator::Gt => BinaryOperatorType::Gt,	18✔
359	SqlBinaryOperator::GtEq => BinaryOperatorType::Gte,	1✔
360	SqlBinaryOperator::Lt => BinaryOperatorType::Lt,	12✔
361	SqlBinaryOperator::LtEq => BinaryOperatorType::Lte,	12✔
362	SqlBinaryOperator::Eq => BinaryOperatorType::Eq,	18✔
363	SqlBinaryOperator::NotEq => BinaryOperatorType::Ne,	×
364		×
365	SqlBinaryOperator::Plus => BinaryOperatorType::Add,	×
366	SqlBinaryOperator::Minus => BinaryOperatorType::Sub,	×
367	SqlBinaryOperator::Multiply => BinaryOperatorType::Mul,	×
368	SqlBinaryOperator::Divide => BinaryOperatorType::Div,	×
369	SqlBinaryOperator::Modulo => BinaryOperatorType::Mod,	×
370		×
371	SqlBinaryOperator::And => BinaryOperatorType::And,	12✔
372	SqlBinaryOperator::Or => BinaryOperatorType::Or,	6✔
373		×
374	// BinaryOperator::BitwiseAnd => ...	×
375	// BinaryOperator::BitwiseOr => ...	×
376	// BinaryOperator::StringConcat => ...	×
377	_ => return Err(InvalidOperator(format!("{op:?}"))),	×
378	};	×
379		×
380	Ok((	79✔
381	Box::new(Expression::BinaryOperator {	79✔
382	left: left_op,	79✔
383	operator,	79✔
384	right: right_op,	79✔
385	}),	79✔
386	false,	79✔
387	))	79✔
388	}	79✔
389		×
390	fn parse_sql_number(&self, n: &str) -> Result<(Box<Expression>, Bypass), PipelineError> {	43✔
391	match n.parse::<i64>() {	43✔
392	Ok(n) => Ok((Box::new(Expression::Literal(Field::Int(n))), false)),	43✔
393	Err(_) => match n.parse::<f64>() {	×
394	Ok(f) => Ok((	×
395	Box::new(Expression::Literal(Field::Float(OrderedFloat(f)))),	×
396	false,	×
397	)),	×
398	Err(_) => Err(InvalidValue(n.to_string())),	×
399	},	×
400	}	×
401	}	43✔
402		×
403	fn parse_sql_like_operator(	×
404	&self,	×
405	expression_type: &BuilderExpressionType,	×
406	negated: &bool,	×
407	expr: &Expr,	×
408	pattern: &Expr,	×
409	escape_char: &Option<char>,	×
410	schema: &Schema,	×
411	) -> Result<(Box<Expression>, bool), PipelineError> {	×
412	let arg = self.parse_sql_expression(expression_type, expr, schema)?;	×
413	let pattern = self.parse_sql_expression(expression_type, pattern, schema)?;	×
414	let like_expression = Box::new(Expression::Like {	×
415	arg: arg.0,	×
416	pattern: pattern.0,	×
417	escape: *escape_char,	×
418	});	×
419	if *negated {	×
420	Ok((	×
421	Box::new(Expression::UnaryOperator {	×
422	operator: UnaryOperatorType::Not,	×
423	arg: like_expression,	×
424	}),	×
425	arg.1,	×
426	))	×
427	} else {	×
428	Ok((like_expression, arg.1))	×
429	}	×
430	}	×
431		×
432	fn parse_sql_cast_operator(	64✔
433	&self,	64✔
434	expression_type: &BuilderExpressionType,	64✔
435	expr: &Expr,	64✔
436	data_type: &DataType,	64✔
437	schema: &Schema,	64✔
438	) -> Result<(Box<Expression>, bool), PipelineError> {	64✔
439	let expression = self.parse_sql_expression(expression_type, expr, schema)?;	64✔
440	let cast_to = match data_type {	64✔
441	DataType::Decimal(_) => CastOperatorType::Decimal,	×
442	DataType::Binary(_) => CastOperatorType::Binary,	×
443	DataType::Float(_) => CastOperatorType::Float,	10✔
444	DataType::Int(_) => CastOperatorType::Int,	6✔
445	DataType::Integer(_) => CastOperatorType::Int,	×
446	DataType::UnsignedInt(_) => CastOperatorType::UInt,	×
447	DataType::UnsignedInteger(_) => CastOperatorType::UInt,	×
448	DataType::Boolean => CastOperatorType::Boolean,	12✔
449	DataType::Date => CastOperatorType::Date,	×
450	DataType::Timestamp(..) => CastOperatorType::Timestamp,	×
451	DataType::Text => CastOperatorType::Text,	18✔
452	DataType::String => CastOperatorType::String,	18✔
453	DataType::Custom(name, ..) => {	×
454	if name.to_string().to_lowercase() == "bson" {	×
455	CastOperatorType::Bson	×
456	} else {	×
457	Err(PipelineError::InvalidFunction(format!(	×
458	"Unsupported Cast type {name}"	×
459	)))?	×
460	}	×
461	}	×
462	_ => Err(PipelineError::InvalidFunction(format!(	×
463	"Unsupported Cast type {data_type}"	×
464	)))?,	×
465	};	×
466	Ok((	64✔
467	Box::new(Expression::Cast {	64✔
468	arg: expression.0,	64✔
469	typ: cast_to,	64✔
470	}),	64✔
471	expression.1,	64✔
472	))	64✔
473	}	64✔
474	}	×
475		×
476	pub fn fullname_from_ident(ident: &[Ident]) -> String {	46✔
477	let mut ident_tokens = vec![];	46✔
478	for token in ident.iter() {	46✔
479	ident_tokens.push(token.value.clone());	46✔
480	}	46✔
481	ident_tokens.join(".")	46✔
482	}	46✔
483		×
484	fn parse_sql_string(s: &str) -> Result<(Box<Expression>, bool), PipelineError> {	26✔
485	Ok((	26✔
486	Box::new(Expression::Literal(Field::String(s.to_owned()))),	26✔
487	false,	26✔
488	))	26✔
489	}	26✔
490		×
491	pub(crate) fn normalize_ident(id: &Ident) -> String {	170✔
492	match id.quote_style {	170✔
493	Some(_) => id.value.clone(),	×
494	None => id.value.clone(),	170✔
495	}	×
496	}	170✔
497		×
498	pub fn extend_schema_source_def(schema: &Schema, name: &NameOrAlias) -> Schema {	488✔
499	let mut output_schema = schema.clone();	488✔
500	let mut fields = vec![];	488✔
501	for mut field in schema.clone().fields.into_iter() {	2,076✔
502	if let Some(alias) = &name.1 {	2,076✔
503	field.source = SourceDefinition::Alias {	540✔
504	name: alias.to_string(),	540✔
505	};	540✔
506	}	1,536✔
507		×
508	fields.push(field);	2,076✔
509	}	×
510	output_schema.fields = fields;	488✔
511		488✔
512	output_schema	488✔
513	}	488✔
514		×
515	fn parse_sql_column(ident: &[Ident], schema: &Schema) -> Result<Box<Expression>, PipelineError> {	2,737✔
516	let (src_field, src_table_or_alias, src_connection) = match ident.len() {	2,737✔
517	1 => (&ident[0].value, None, None),	2,555✔
518	2 => (&ident[1].value, Some(&ident[0].value), None),	182✔
519	3 => (	×
520	&ident[2].value,	×
521	Some(&ident[1].value),	×
522	Some(&ident[0].value),	×
523	),	×
524	_ => {	×
525	return Err(IllegalFieldIdentifier(	×
526	ident	×
527	.iter()	×
528	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
529	))	×
530	}	×
531	};	×
532		×
533	let matching_by_field: Vec<(usize, &FieldDefinition)> = schema	2,737✔
534	.fields	2,737✔
535	.iter()	2,737✔
536	.enumerate()	2,737✔
537	.filter(\|(_idx, f)\| &f.name == src_field)	11,870✔
538	.collect();	2,737✔
539		2,737✔
540	match matching_by_field.len() {	2,737✔
541	0 => Err(UnknownFieldIdentifier(	×
542	ident	×
543	.iter()	×
544	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
545	)),	×
546	1 => Ok(Box::new(Expression::Column {	2,629✔
547	index: matching_by_field[0].0,	2,629✔
548	})),	2,629✔
549	_ => match src_table_or_alias {	108✔
550	None => Err(AmbiguousFieldIdentifier(	×
551	ident	×
552	.iter()	×
553	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
554	)),	×
555	Some(src_table_or_alias) => {	108✔
556	let matching_by_table_or_alias: Vec<(usize, &FieldDefinition)> = matching_by_field	108✔
557	.into_iter()	108✔
558	.filter(\|(_idx, field)\| match &field.source {	216✔
559	SourceDefinition::Alias { name } => name == src_table_or_alias,	144✔
560	SourceDefinition::Table {	×
561	name,	72✔
562	connection: _,	72✔
563	} => name == src_table_or_alias,	72✔
564	_ => false,	×
565	})	216✔
566	.collect();	108✔
567		108✔
568	match matching_by_table_or_alias.len() {	108✔
569	0 => Err(UnknownFieldIdentifier(	×
570	ident	×
571	.iter()	×
572	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
573	)),	×
574	1 => Ok(Box::new(Expression::Column {	108✔
575	index: matching_by_table_or_alias[0].0,	108✔
576	})),	108✔
577	_ => match src_connection {	×
578	None => Err(InvalidExpression(	×
579	ident	×
580	.iter()	×
581	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
582	)),	×
583	Some(src_connection) => {	×
584	let matching_by_connection: Vec<(usize, &FieldDefinition)> =	×
585	matching_by_table_or_alias	×
586	.into_iter()	×
587	.filter(\|(_idx, field)\| match &field.source {	×
588	SourceDefinition::Table {	×
589	name: _,	×
590	connection,	×
591	} => connection == src_connection,	×
592	_ => false,	×
593	})	×
594	.collect();	×
595		×
596	match matching_by_connection.len() {	×
597	0 => Err(UnknownFieldIdentifier(	×
598	ident	×
599	.iter()	×
600	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
601	)),	×
602	1 => Ok(Box::new(Expression::Column {	×
603	index: matching_by_connection[0].0,	×
604	})),	×
605	_ => Err(InvalidExpression(	×
606	ident	×
607	.iter()	×
608	.fold(String::new(), \|a, b\| a + "." + b.value.as_str()),	×
609	)),	×
610	}	×
611	}	×
612	},	×
613	}	×
614	}	×
615	},	×
616	}	×
617	}	2,737✔

getdozer / dozer / 4061292542

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous