4020902227

Build # 4020902227

Build Type

github

Committed by GitHub

Commit Message

Merge 57279c6b6 into a12da35a5

Pull Request Pull Request #743: Chore clippy fix

Run Details

165 of 165 new or added lines in 60 files covered. (100.0%)

23638 of 35485 relevant lines covered (66.61%)

38417.79 hits per line

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

51.58

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

use std::fmt::Display;

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::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 enum ConstraintIdentifier {
    Single(Ident),
    Compound(Vec<Ident>),
}

impl Display for ConstraintIdentifier {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ConstraintIdentifier::Single(ident) => f.write_fmt(format_args!("{ident}")),
            ConstraintIdentifier::Compound(ident) => f.write_fmt(format_args!("{ident:?}")),
        }
    }
}
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) => {
                let idx = get_field_index(&ConstraintIdentifier::Single(ident.clone()), schema);

                let idx = idx?.map_or(
                    Err(PipelineError::InvalidExpression(ident.value.to_string())),
                    Ok,
                )?;
                Ok((Box::new(Expression::Column { index: idx }), false))
            }
            SqlExpr::CompoundIdentifier(ident) => {
                let idx = get_field_index(&ConstraintIdentifier::Compound(ident.clone()), schema)?
                    .map_or(
                        Err(PipelineError::InvalidExpression(format!("{ident:?}"))),
                        Ok,
                    )?;
                Ok((Box::new(Expression::Column { index: idx }), 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
}

pub fn get_field_index(
    ident: &ConstraintIdentifier,
    schema: &Schema,
) -> Result<Option<usize>, PipelineError> {
    let get_field_idx = |ident: &Ident, schema: &Schema| -> Option<(usize, FieldDefinition)> {
        schema
            .fields
            .iter()
            .enumerate()
            .find(|(_, f)| f.name == ident.value)
            .map(|(idx, fd)| (idx, fd.clone()))
    };

    let tables_matches = |table_ident: &Ident, fd: FieldDefinition| -> bool {
        match fd.source {
            dozer_types::types::SourceDefinition::Table {
                connection: _,
                name,
            } => name == table_ident.value,
            dozer_types::types::SourceDefinition::Alias { name } => name == table_ident.value,
            dozer_types::types::SourceDefinition::Dynamic => false,
        }
    };

    match ident {
        ConstraintIdentifier::Single(ident) => {
            let field_idx = get_field_idx(ident, schema);
            field_idx.map_or(
                Err(PipelineError::InvalidExpression(ident.value.to_string())),
                |t| Ok(Some(t.0)),
            )
        }
        ConstraintIdentifier::Compound(comp_ident) => {
            if comp_ident.len() > 2 {
                return Err(PipelineError::NameSpaceTooLong(
                    comp_ident
                        .iter()
                        .map(|a| a.value.clone())
                        .collect::<Vec<String>>()
                        .join("."),
                ));
            }
            let table_name = comp_ident.first().expect("table_name is expected");
            let field_name = comp_ident.last().expect("field_name is expected");

            let field_idx = get_field_idx(field_name, schema);
            if let Some((idx, fd)) = field_idx {
                if tables_matches(table_name, fd) {
                    return Ok(Some(idx));
                }
            }
            Ok(None)
        }
    }
}

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

getdozer / dozer / 4020902227

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