4370408272

Build Type

push

github

Committed by GitHub

Commit Message

fix: Fix compilation error introduced in #1158 (#1183)

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

28163 of 39541 relevant lines covered (71.22%)

73625.66 hits per line

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

60.33

/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::{
    InvalidArgument, InvalidExpression, InvalidNestedAggregationFunction, InvalidOperator,
    InvalidValue,
};
use crate::pipeline::errors::{PipelineError, SqlError};
use crate::pipeline::expression::aggregate::AggregateFunctionType;
use crate::pipeline::expression::datetime::DateTimeFunctionType;

use crate::pipeline::expression::execution::Expression;
use crate::pipeline::expression::execution::Expression::{GeoFunction, ScalarFunction};
use crate::pipeline::expression::geo::common::GeoFunctionType;
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;

#[derive(Clone, PartialEq, Debug)]
pub struct ExpressionBuilder {
    // Must be an aggregation function
    pub aggregations: Vec<Expression>,
    pub offset: usize,
}

impl ExpressionBuilder {
    pub fn new(offset: usize) -> Self {
        Self {
            aggregations: Vec::new(),
            offset,
        }
    }

    pub fn from(offset: usize, aggregations: Vec<Expression>) -> Self {
        Self {
            aggregations,
            offset,
        }
    }

    pub fn build(
        &mut self,
        parse_aggregations: bool,
        sql_expression: &SqlExpr,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        self.parse_sql_expression(parse_aggregations, sql_expression, schema)
    }

    pub(crate) fn parse_sql_expression(
        &mut self,
        parse_aggregations: bool,
        expression: &SqlExpr,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        match expression {
            SqlExpr::Trim {
                expr,
                trim_where,
                trim_what,
            } => self.parse_sql_trim_function(
                parse_aggregations,
                expr,
                trim_where,
                trim_what,
                schema,
            ),
            SqlExpr::Identifier(ident) => Self::parse_sql_column(&[ident.clone()], schema),
            SqlExpr::CompoundIdentifier(ident) => Self::parse_sql_column(ident, schema),
            SqlExpr::Value(SqlValue::Number(n, _)) => Self::parse_sql_number(n),
            SqlExpr::Value(SqlValue::Null) => Ok(Expression::Literal(Field::Null)),
            SqlExpr::Value(SqlValue::SingleQuotedString(s) | SqlValue::DoubleQuotedString(s)) => {
                Self::parse_sql_string(s)
            }
            SqlExpr::UnaryOp { expr, op } => {
                self.parse_sql_unary_op(parse_aggregations, op, expr, schema)
            }
            SqlExpr::BinaryOp { left, op, right } => {
                self.parse_sql_binary_op(parse_aggregations, left, op, right, schema)
            }
            SqlExpr::Nested(expr) => self.parse_sql_expression(parse_aggregations, expr, schema),
            SqlExpr::Function(sql_function) => {
                self.parse_sql_function(parse_aggregations, sql_function, schema)
            }
            SqlExpr::Like {
                negated,
                expr,
                pattern,
                escape_char,
            } => self.parse_sql_like_operator(
                parse_aggregations,
                negated,
                expr,
                pattern,
                escape_char,
                schema,
            ),
            SqlExpr::Cast { expr, data_type } => {
                self.parse_sql_cast_operator(parse_aggregations, expr, data_type, schema)
            }
            SqlExpr::Extract { field, expr } => {
                self.parse_sql_extract_operator(parse_aggregations, field, expr, schema)
            }
            _ => Err(InvalidExpression(format!("{expression:?}"))),
        }
    }

    fn parse_sql_column(ident: &[Ident], schema: &Schema) -> Result<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(PipelineError::SqlError(SqlError::InvalidColumn(
                    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() {
            1 => Ok(Expression::Column {
                index: matching_by_field[0].0,
            }),
            _ => match src_table_or_alias {
                None => Err(PipelineError::SqlError(SqlError::InvalidColumn(
                    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() {
                        1 => Ok(Expression::Column {
                            index: matching_by_table_or_alias[0].0,
                        }),
                        _ => match src_connection {
                            None => Err(PipelineError::SqlError(SqlError::InvalidColumn(
                                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() {
                                    1 => Ok(Expression::Column {
                                        index: matching_by_connection[0].0,
                                    }),
                                    _ => Err(PipelineError::SqlError(SqlError::InvalidColumn(
                                        ident
                                            .iter()
                                            .fold(String::new(), |a, b| a + "." + b.value.as_str()),
                                    ))),
                                }
                            }
                        },
                    }
                }
            },
        }
    }

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

    fn parse_sql_function(
        &mut self,
        parse_aggregations: bool,
        sql_function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let function_name = sql_function.name.to_string().to_lowercase();

        #[cfg(feature = "python")]
        if function_name.starts_with("py_") {
            // The function is from python udf.
            let udf_name = function_name.strip_prefix("py_").unwrap();
            return self.parse_python_udf(udf_name, sql_function, schema);
        }

        match (
            AggregateFunctionType::new(function_name.as_str()),
            parse_aggregations,
        ) {
            (Ok(aggr), true) => {
                let mut arg_expr: Vec<Expression> = Vec::new();
                for arg in &sql_function.args {
                    let aggregation = self.parse_sql_function_arg(true, arg, schema)?;
                    arg_expr.push(aggregation);
                }
                let measure = Expression::AggregateFunction {
                    fun: aggr,
                    args: arg_expr,
                };
                let index = match self
                    .aggregations
                    .iter()
                    .enumerate()
                    .find(|e| e.1 == &measure)
                {
                    Some((index, _existing)) => index,
                    _ => {
                        self.aggregations.push(measure);
                        self.aggregations.len() - 1
                    }
                };
                Ok(Expression::Column {
                    index: self.offset + index,
                })
            }
            (Ok(_agg), false) => Err(InvalidNestedAggregationFunction(function_name)),
            (Err(_), _) => {
                let mut function_args: Vec<Expression> = Vec::new();
                for arg in &sql_function.args {
                    function_args.push(self.parse_sql_function_arg(
                        parse_aggregations,
                        arg,
                        schema,
                    )?);
                }

                match ScalarFunctionType::new(function_name.as_str()) {
                    Ok(sft) => Ok(ScalarFunction {
                        fun: sft,
                        args: function_args.clone(),
                    }),
                    Err(_d) => match GeoFunctionType::new(function_name.as_str()) {
                        Ok(gft) => Ok(GeoFunction {
                            fun: gft,
                            args: function_args.clone(),
                        }),
                        Err(_err) => Err(InvalidNestedAggregationFunction(function_name)),
                    },
                }
            }
        }
    }

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

    fn parse_sql_unary_op(
        &mut self,
        parse_aggregations: bool,
        op: &SqlUnaryOperator,
        expr: &SqlExpr,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let arg = Box::new(self.parse_sql_expression(parse_aggregations, expr, schema)?);
        let operator = match op {
            SqlUnaryOperator::Not => UnaryOperatorType::Not,
            SqlUnaryOperator::Plus => UnaryOperatorType::Plus,
            SqlUnaryOperator::Minus => UnaryOperatorType::Minus,
            _ => return Err(InvalidOperator(format!("{op:?}"))),
        };

        Ok(Expression::UnaryOperator { operator, arg })
    }

    fn parse_sql_binary_op(
        &mut self,
        parse_aggregations: bool,
        left: &SqlExpr,
        op: &SqlBinaryOperator,
        right: &SqlExpr,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let left_op = self.parse_sql_expression(parse_aggregations, left, schema)?;
        let right_op = self.parse_sql_expression(parse_aggregations, right, schema)?;

        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,
            _ => return Err(InvalidOperator(format!("{op:?}"))),
        };

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

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

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

    fn parse_sql_extract_operator(
        &mut self,
        parse_aggregations: bool,
        field: &sqlparser::ast::DateTimeField,
        expr: &Expr,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let right = self.parse_sql_expression(parse_aggregations, expr, schema)?;
        Ok(Expression::DateTimeFunction {
            fun: DateTimeFunctionType::Extract { field: *field },
            arg: Box::new(right),
        })
    }

    fn parse_sql_cast_operator(
        &mut self,
        parse_aggregations: bool,
        expr: &Expr,
        data_type: &DataType,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let expression = self.parse_sql_expression(parse_aggregations, 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(Expression::Cast {
            arg: Box::new(expression),
            typ: cast_to,
        })
    }

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

    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(".")
    }

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

    #[cfg(feature = "python")]
    fn parse_python_udf(
        &mut self,
        name: &str,
        function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        // First, get python function define by name.
        // Then, transfer python function to Expression::PythonUDF

        use dozer_types::types::FieldType;
        use PipelineError::InvalidQuery;

        let args = function
            .args
            .iter()
            .map(|argument| self.parse_sql_function_arg(false, argument, schema))
            .collect::<Result<Vec<_>, PipelineError>>()?;

        let last_arg = args
            .last()
            .ok_or_else(|| InvalidQuery("Can't get python udf return type".to_string()))?;

        let return_type = match last_arg {
            Expression::Literal(Field::String(s)) => {
                FieldType::try_from(s.as_str()).map_err(|e| InvalidQuery(format!("Failed to parse Python UDF return type: {e}")))?
            }
            _ => return Err(InvalidArgument("The last arg for python udf should be a string literal, which represents return type".to_string())),
        };

        Ok(Expression::PythonUDF {
            name: name.to_string(),
            args,
            return_type,
        })
    }
}

#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub struct NameOrAlias(pub String, pub Option<String>);

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
}

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

getdozer / dozer / 4370408272

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