4007818786

Build Type

github

Committed by GitHub

Commit Message

Merge baf5c38aa into 6c0ac2b2c

Pull Request Pull Request #733: Bump diesel from 2.0.2 to 2.0.3

Run Details

23389 of 34432 relevant lines covered (67.93%)

40326.78 hits per line

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

53.24

/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,490✔
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,590✔
78	let idx = get_field_index(&ConstraintIdentifier::Single(ident.clone()), schema);	2,590✔
79		×
80	let idx = idx?.map_or(	2,593✔
81	Err(PipelineError::InvalidExpression(ident.value.to_string())),	2,590✔
82	Ok,	2,590✔
83	)?;	2,590✔
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();	165✔
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	};	1✔
249	Err(InvalidExpression(format!("{:?}", expression)))	1✔
250	}	167✔
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: {:?}",	×
317	expression	×
318	)))	×
319	}	166✔
320		×
321	fn parse_sql_function_arg(	×
322	&self,
323	expression_type: &BuilderExpressionType,
324	argument: &FunctionArg,	×
325	schema: &Schema,
326	) -> Result<(Box<Expression>, bool), PipelineError> {	×
327	match argument {	354✔
328	FunctionArg::Named {	×
329	name: _,
330	arg: FunctionArgExpr::Expr(arg),	×
331	} => self.parse_sql_expression(expression_type, arg, schema),	×
332	FunctionArg::Named {	×
333	name: _,	×
334	arg: FunctionArgExpr::Wildcard,	×
335	} => Err(InvalidArgument(format!("{:?}", argument))),	×
336	FunctionArg::Unnamed(FunctionArgExpr::Expr(arg)) => {	354✔
337	self.parse_sql_expression(expression_type, arg, schema)	354✔
338	}	×
339	FunctionArg::Unnamed(FunctionArgExpr::Wildcard) => {	×
340	Err(InvalidArgument(format!("{:?}", argument)))	×
341	}
342	_ => Err(InvalidArgument(format!("{:?}", argument))),	×
343	}	×
344	}	354✔
345		×
346	fn parse_sql_unary_op(	×
347	&self,	×
348	expression_type: &BuilderExpressionType,	×
349	op: &SqlUnaryOperator,	×
350	expr: &SqlExpr,	×
351	schema: &Schema,	×
352	) -> Result<(Box<Expression>, Bypass), PipelineError> {	×
353	let (arg, bypass) = self.parse_sql_expression(expression_type, expr, schema)?;	×
354	if bypass {	×
355	return Ok((arg, bypass));	×
356	}	×
357		×
358	let operator = match op {	×
359	SqlUnaryOperator::Not => UnaryOperatorType::Not,	×
360	SqlUnaryOperator::Plus => UnaryOperatorType::Plus,	×
361	SqlUnaryOperator::Minus => UnaryOperatorType::Minus,	×
362	_ => return Err(InvalidOperator(format!("{:?}", op))),	×
363	};	×
364		×
365	Ok((Box::new(Expression::UnaryOperator { operator, arg }), false))	×
366	}	×
367		×
368	fn parse_sql_binary_op(	79✔
369	&self,	79✔
370	expression_type: &BuilderExpressionType,	79✔
371	left: &SqlExpr,	79✔
372	op: &SqlBinaryOperator,	79✔
373	right: &SqlExpr,	79✔
374	schema: &Schema,	79✔
375	) -> Result<(Box<Expression>, bool), PipelineError> {	79✔
376	let (left_op, bypass_left) = self.parse_sql_expression(expression_type, left, schema)?;	79✔
377	if bypass_left {	79✔
378	return Ok((left_op, bypass_left));	×
379	}	79✔
380	let (right_op, bypass_right) = self.parse_sql_expression(expression_type, right, schema)?;	79✔
381	if bypass_right {	79✔
382	return Ok((right_op, bypass_right));	×
383	}	79✔
384		×
385	let operator = match op {	79✔
386	SqlBinaryOperator::Gt => BinaryOperatorType::Gt,	18✔
387	SqlBinaryOperator::GtEq => BinaryOperatorType::Gte,	1✔
388	SqlBinaryOperator::Lt => BinaryOperatorType::Lt,	12✔
389	SqlBinaryOperator::LtEq => BinaryOperatorType::Lte,	12✔
390	SqlBinaryOperator::Eq => BinaryOperatorType::Eq,	18✔
391	SqlBinaryOperator::NotEq => BinaryOperatorType::Ne,	×
392		×
393	SqlBinaryOperator::Plus => BinaryOperatorType::Add,	×
394	SqlBinaryOperator::Minus => BinaryOperatorType::Sub,	×
395	SqlBinaryOperator::Multiply => BinaryOperatorType::Mul,	×
396	SqlBinaryOperator::Divide => BinaryOperatorType::Div,	×
397	SqlBinaryOperator::Modulo => BinaryOperatorType::Mod,	×
398		×
399	SqlBinaryOperator::And => BinaryOperatorType::And,	12✔
400	SqlBinaryOperator::Or => BinaryOperatorType::Or,	6✔
401
402	// BinaryOperator::BitwiseAnd => ...	×
403	// BinaryOperator::BitwiseOr => ...	×
404	// BinaryOperator::StringConcat => ...	×
405	_ => return Err(InvalidOperator(format!("{:?}", op))),	×
406	};	×
407		×
408	Ok((	79✔
409	Box::new(Expression::BinaryOperator {	79✔
410	left: left_op,	79✔
411	operator,	79✔
412	right: right_op,	79✔
413	}),	79✔
414	false,	79✔
415	))	79✔
416	}	79✔
417		×
418	fn parse_sql_number(&self, n: &str) -> Result<(Box<Expression>, Bypass), PipelineError> {	43✔
419	match n.parse::<i64>() {	43✔
420	Ok(n) => Ok((Box::new(Expression::Literal(Field::Int(n))), false)),	43✔
421	Err(_) => match n.parse::<f64>() {	×
422	Ok(f) => Ok((	×
423	Box::new(Expression::Literal(Field::Float(OrderedFloat(f)))),	×
424	false,	×
425	)),	×
426	Err(_) => Err(InvalidValue(n.to_string())),	×
427	},	×
428	}	×
429	}	43✔
430		×
431	fn parse_sql_like_operator(	×
432	&self,	×
433	expression_type: &BuilderExpressionType,	×
434	negated: &bool,	×
435	expr: &Expr,	×
436	pattern: &Expr,	×
437	escape_char: &Option<char>,	×
438	schema: &Schema,	×
439	) -> Result<(Box<Expression>, bool), PipelineError> {	×
440	let arg = self.parse_sql_expression(expression_type, expr, schema)?;	×
441	let pattern = self.parse_sql_expression(expression_type, pattern, schema)?;	×
442	let like_expression = Box::new(Expression::Like {	×
443	arg: arg.0,	×
444	pattern: pattern.0,	×
445	escape: *escape_char,	×
446	});	×
447	if *negated {	×
448	Ok((	×
449	Box::new(Expression::UnaryOperator {	×
450	operator: UnaryOperatorType::Not,	×
451	arg: like_expression,	×
452	}),	×
453	arg.1,	×
454	))	×
455	} else {	×
456	Ok((like_expression, arg.1))	×
457	}	×
458	}	×
459		×
460	fn parse_sql_cast_operator(	64✔
461	&self,	64✔
462	expression_type: &BuilderExpressionType,	64✔
463	expr: &Expr,	64✔
464	data_type: &DataType,	64✔
465	schema: &Schema,	64✔
466	) -> Result<(Box<Expression>, bool), PipelineError> {	64✔
467	let expression = self.parse_sql_expression(expression_type, expr, schema)?;	64✔
468	let cast_to = match data_type {	64✔
469	DataType::Decimal(_) => CastOperatorType::Decimal,	×
470	DataType::Binary(_) => CastOperatorType::Binary,	×
471	DataType::Float(_) => CastOperatorType::Float,	10✔
472	DataType::Int(_) => CastOperatorType::Int,	6✔
473	DataType::Integer(_) => CastOperatorType::Int,	×
474	DataType::UnsignedInt(_) => CastOperatorType::UInt,	×
475	DataType::UnsignedInteger(_) => CastOperatorType::UInt,	×
476	DataType::Boolean => CastOperatorType::Boolean,	12✔
477	DataType::Date => CastOperatorType::Date,	×
478	DataType::Timestamp(..) => CastOperatorType::Timestamp,	×
479	DataType::Text => CastOperatorType::Text,	18✔
480	DataType::String => CastOperatorType::String,	18✔
481	DataType::Custom(name, ..) => {	×
482	if name.to_string().to_lowercase() == "bson" {	×
483	CastOperatorType::Bson	×
484	} else {	×
485	Err(PipelineError::InvalidFunction(format!(	×
486	"Unsupported Cast type {}",	×
487	name	×
488	)))?	×
489	}
490	}
491	_ => Err(PipelineError::InvalidFunction(format!(	×
492	"Unsupported Cast type {}",	×
493	data_type	×
494	)))?,	×
495	};
496	Ok((	64✔
497	Box::new(Expression::Cast {	64✔
498	arg: expression.0,	64✔
499	typ: cast_to,	64✔
500	}),	64✔
501	expression.1,	64✔
502	))	64✔
503	}	64✔
504	}	×
505		×
506	pub fn fullname_from_ident(ident: &[Ident]) -> String {	28✔
507	let mut ident_tokens = vec![];	28✔
508	for token in ident.iter() {	28✔
509	ident_tokens.push(token.value.clone());	28✔
510	}	28✔
511	ident_tokens.join(".")	28✔
512	}	28✔
513		×
514	fn parse_sql_string(s: &str) -> Result<(Box<Expression>, bool), PipelineError> {	26✔
515	Ok((	26✔
516	Box::new(Expression::Literal(Field::String(s.to_owned()))),	26✔
517	false,	26✔
518	))	26✔
519	}	26✔
520		×
521	pub(crate) fn normalize_ident(id: &Ident) -> String {	146✔
522	match id.quote_style {	146✔
523	Some(_) => id.value.clone(),	×
524	None => id.value.clone(),	146✔
525	}	×
526	}	146✔
527		×
528	pub fn extend_schema_source_def(schema: &Schema, name: &NameOrAlias) -> Schema {	269✔
529	let mut output_schema = schema.clone();	269✔
530	let mut fields = vec![];	269✔
531	for mut field in schema.clone().fields.into_iter() {	1,104✔
532	if let Some(alias) = &name.1 {	1,104✔
533	field.source = SourceDefinition::Alias {	120✔
534	name: alias.to_string(),	120✔
535	};	120✔
536	}	984✔
537
538	fields.push(field);	1,104✔
539	}
540	output_schema.fields = fields;	269✔
541		269✔
542	output_schema	269✔
543	}	269✔
544
545	pub fn get_field_index(	2,683✔
546	ident: &ConstraintIdentifier,	2,683✔
547	schema: &Schema,	2,683✔
548	) -> Result<Option<usize>, PipelineError> {	2,683✔
549	let get_field_idx = \|ident: &Ident, schema: &Schema\| -> Option<(usize, FieldDefinition)> {	2,683✔
550	schema	2,683✔
551	.fields	2,683✔
552	.iter()	2,683✔
553	.enumerate()	2,683✔
554	.find(\|(_, f)\| f.name == ident.value)	5,365✔
555	.map(\|(idx, fd)\| (idx, fd.clone()))	2,683✔
556	};	2,683✔
557
558	let tables_matches = \|table_ident: &Ident, fd: FieldDefinition\| -> bool {	2,683✔
559	match fd.source {	92✔
560	dozer_types::types::SourceDefinition::Table {
561	connection: _,
562	name,	30✔
563	} => name == table_ident.value,	30✔
564	dozer_types::types::SourceDefinition::Alias { name } => name == table_ident.value,	60✔
565	dozer_types::types::SourceDefinition::Dynamic => false,	2✔
566	}
567	};	92✔
568
569	match ident {	2,683✔
570	ConstraintIdentifier::Single(ident) => {	2,591✔
571	let field_idx = get_field_idx(ident, schema);	2,591✔
572	field_idx.map_or(	2,591✔
573	Err(PipelineError::InvalidExpression(ident.value.to_string())),	2,591✔
574	\|t\| Ok(Some(t.0)),	2,592✔
575	)	2,591✔
576	}
577	ConstraintIdentifier::Compound(comp_ident) => {	92✔
578	if comp_ident.len() > 2 {	92✔
579	return Err(PipelineError::NameSpaceTooLong(	×
580	comp_ident	×
581	.iter()	×
582	.map(\|a\| a.value.clone())	×
583	.collect::<Vec<String>>()	×
584	.join("."),	×
585	));	×
586	}	92✔
587	let table_name = comp_ident.first().expect("table_name is expected");	92✔
588	let field_name = comp_ident.last().expect("field_name is expected");	92✔
589		92✔
590	let field_idx = get_field_idx(field_name, schema);	92✔
591	if let Some((idx, fd)) = field_idx {	92✔
592	if tables_matches(table_name, fd) {	92✔
593	return Ok(Some(idx));	72✔
594	}	20✔
595	}	×
596	Ok(None)	20✔
597	}
598	}
599	}	2,683✔

getdozer / dozer / 4007818786

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