6015696039

Committed 29 Aug 2023 05:40PM UTC coverage: 75.532% (-0.8%) from 76.332%

Build # 6015696039

Build Type

push

github

Committed by

web-flow

Commit Message

Bump clap from 4.3.11 to 4.4.1 (#1941)

Bumps [clap](https://github.com/clap-rs/clap) from 4.3.11 to 4.4.1.
- [Release notes](https://github.com/clap-rs/clap/releases)
- [Changelog](https://github.com/clap-rs/clap/blob/master/CHANGELOG.md)
- [Commits](https://github.com/clap-rs/clap/compare/v4.3.11...v4.4.1)

---
updated-dependencies:
- dependency-name: clap
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

49076 of 64974 relevant lines covered (75.53%)

65447.31 hits per line

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

63.39

/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, DateTimeField, Expr as SqlExpr, Expr, Function,
    FunctionArg, FunctionArgExpr, Ident, Interval, TrimWhereField,
    UnaryOperator as SqlUnaryOperator, Value as SqlValue,
};

use crate::pipeline::errors::PipelineError::{
    InvalidArgument, InvalidExpression, InvalidFunction, InvalidNestedAggregationFunction,
    InvalidOperator, InvalidValue,
};
use crate::pipeline::errors::{PipelineError, SqlError};
use crate::pipeline::expression::aggregate::AggregateFunctionType;
use crate::pipeline::expression::conditional::ConditionalExpressionType;
use crate::pipeline::expression::datetime::DateTimeFunctionType;

use crate::pipeline::expression::execution::Expression;
use crate::pipeline::expression::execution::Expression::{
    ConditionalExpression, GeoFunction, Now, ScalarFunction,
};
use crate::pipeline::expression::geo::common::GeoFunctionType;
use crate::pipeline::expression::json_functions::JsonFunctionType;
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::InList {
                expr,
                list,
                negated,
            } => self.parse_sql_in_list_operator(parse_aggregations, expr, list, *negated, 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)
            }
            SqlExpr::Interval(Interval {
                value,
                leading_field,
                leading_precision: _,
                last_field: _,
                fractional_seconds_precision: _,
            }) => {
                self.parse_sql_interval_expression(parse_aggregations, value, leading_field, schema)
            }
            SqlExpr::Case {
                operand,
                conditions,
                results,
                else_result,
            } => self.parse_sql_case_expression(
                parse_aggregations,
                operand,
                conditions,
                results,
                else_result,
                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()
                        .map(|e| e.value.as_str())
                        .collect::<Vec<&str>>()
                        .join("."),
                )));
            }
        };

        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()
                        .map(|e| e.value.as_str())
                        .collect::<Vec<&str>>()
                        .join("."),
                ))),
                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()
                                            .map(|e| e.value.as_str())
                                            .collect::<Vec<&str>>()
                                            .join("."),
                                    ))),
                                }
                            }
                        },
                    }
                }
            },
        }
    }

    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 aggr_function_check(
        &mut self,
        function_name: String,
        parse_aggregations: bool,
        sql_function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        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(_), _) => Err(InvalidNestedAggregationFunction(function_name)),
        }
    }

    fn scalar_function_check(
        &mut self,
        function_name: String,
        parse_aggregations: bool,
        sql_function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        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) => Err(InvalidFunction(function_name)),
        }
    }

    fn geo_expr_check(
        &mut self,
        function_name: String,
        parse_aggregations: bool,
        sql_function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        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 GeoFunctionType::new(function_name.as_str()) {
            Ok(gft) => Ok(GeoFunction {
                fun: gft,
                args: function_args.clone(),
            }),
            Err(_e) => Err(InvalidFunction(function_name)),
        }
    }

    fn datetime_expr_check(&mut self, function_name: String) -> Result<Expression, PipelineError> {
        match DateTimeFunctionType::new(function_name.as_str()) {
            Ok(dtf) => Ok(Now { fun: dtf }),
            Err(_e) => Err(InvalidFunction(function_name)),
        }
    }

    fn json_func_check(
        &mut self,
        function_name: String,
        parse_aggregations: bool,
        sql_function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        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 JsonFunctionType::new(function_name.as_str()) {
            Ok(jft) => Ok(Expression::Json {
                fun: jft,
                args: function_args,
            }),
            Err(_e) => Err(InvalidFunction(function_name)),
        }
    }

    fn conditional_expr_check(
        &mut self,
        function_name: String,
        parse_aggregations: bool,
        sql_function: &Function,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        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 ConditionalExpressionType::new(function_name.as_str()) {
            Ok(cet) => Ok(ConditionalExpression {
                fun: cet,
                args: function_args.clone(),
            }),
            Err(_err) => Err(InvalidFunction(function_name)),
        }
    }

    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);
        }

        let aggr_check = self.aggr_function_check(
            function_name.clone(),
            parse_aggregations,
            sql_function,
            schema,
        );
        if aggr_check.is_ok() {
            return aggr_check;
        }

        let scalar_check = self.scalar_function_check(
            function_name.clone(),
            parse_aggregations,
            sql_function,
            schema,
        );
        if scalar_check.is_ok() {
            return scalar_check;
        }

        let geo_check = self.geo_expr_check(
            function_name.clone(),
            parse_aggregations,
            sql_function,
            schema,
        );
        if geo_check.is_ok() {
            return geo_check;
        }

        let conditional_check = self.conditional_expr_check(
            function_name.clone(),
            parse_aggregations,
            sql_function,
            schema,
        );
        if conditional_check.is_ok() {
            return conditional_check;
        }

        let datetime_check = self.datetime_expr_check(function_name.clone());
        if datetime_check.is_ok() {
            return datetime_check;
        }

        self.json_func_check(function_name, parse_aggregations, sql_function, schema)
    }

    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,
            } => Ok(Expression::Literal(Field::Null)),
            FunctionArg::Unnamed(FunctionArgExpr::Expr(arg)) => {
                self.parse_sql_expression(parse_aggregations, arg, schema)
            }
            FunctionArg::Unnamed(FunctionArgExpr::Wildcard) => Ok(Expression::Literal(Field::Null)),
            FunctionArg::Named {
                name: _,
                arg: FunctionArgExpr::QualifiedWildcard(_),
            } => Err(InvalidArgument(format!("{argument:?}"))),
            FunctionArg::Unnamed(FunctionArgExpr::QualifiedWildcard(_)) => {
                Err(InvalidArgument(format!("{argument:?}")))
            }
        }
    }

    fn parse_sql_case_expression(
        &mut self,
        parse_aggregations: bool,
        operand: &Option<Box<Expr>>,
        conditions: &[Expr],
        results: &[Expr],
        else_result: &Option<Box<Expr>>,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let op = match operand {
            Some(o) => Some(Box::new(self.parse_sql_expression(
                parse_aggregations,
                o,
                schema,
            )?)),
            None => None,
        };
        let conds = conditions
            .iter()
            .map(|cond| self.parse_sql_expression(parse_aggregations, cond, schema))
            .collect::<Result<Vec<_>, PipelineError>>()?;
        let res = results
            .iter()
            .map(|r| self.parse_sql_expression(parse_aggregations, r, schema))
            .collect::<Result<Vec<_>, PipelineError>>()?;
        let else_res = match else_result {
            Some(r) => Some(Box::new(self.parse_sql_expression(
                parse_aggregations,
                r,
                schema,
            )?)),
            None => None,
        };

        Ok(Expression::Case {
            operand: op,
            conditions: conds,
            results: res,
            else_result: else_res,
        })
    }

    fn parse_sql_interval_expression(
        &mut self,
        parse_aggregations: bool,
        value: &Expr,
        leading_field: &Option<DateTimeField>,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let right = self.parse_sql_expression(parse_aggregations, value, schema)?;
        if leading_field.is_some() {
            Ok(Expression::DateTimeFunction {
                fun: DateTimeFunctionType::Interval {
                    field: leading_field.unwrap(),
                },
                arg: Box::new(right),
            })
        } else {
            Err(InvalidExpression(format!("INTERVAL for {leading_field:?}")))
        }
    }

    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),
        })
    }

    #[cfg(not(feature = "bigdecimal"))]
    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())),
            },
        }
    }

    #[cfg(feature = "bigdecimal")]
    fn parse_sql_number(n: &bigdecimal::BigDecimal) -> Result<Expression, PipelineError> {
        use bigdecimal::ToPrimitive;
        if n.is_integer() {
            Ok(Expression::Literal(Field::Int(n.to_i64().unwrap())))
        } else {
            match n.to_f64() {
                Some(f) => Ok(Expression::Literal(Field::Float(OrderedFloat(f)))),
                None => 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::JSON => CastOperatorType::Json,
            DataType::Custom(name, ..) => {
                if name.to_string().to_lowercase() == "uint" {
                    CastOperatorType::UInt
                } else if name.to_string().to_lowercase() == "u128" {
                    CastOperatorType::U128
                } else if name.to_string().to_lowercase() == "i128" {
                    CastOperatorType::I128
                } 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 return_type = {
            let ident = function
                .return_type
                .as_ref()
                .ok_or_else(|| InvalidQuery("Python UDF must have a return type. The syntax is: function_name<return_type>(arguments)".to_string()))?;

            FieldType::try_from(ident.value.as_str())
                .map_err(|e| InvalidQuery(format!("Failed to parse Python UDF return type: {e}")))?
        };

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

    fn parse_sql_in_list_operator(
        &mut self,
        parse_aggregations: bool,
        expr: &Expr,
        list: &[Expr],
        negated: bool,
        schema: &Schema,
    ) -> Result<Expression, PipelineError> {
        let expr = self.parse_sql_expression(parse_aggregations, expr, schema)?;
        let list = list
            .iter()
            .map(|expr| self.parse_sql_expression(parse_aggregations, expr, schema))
            .collect::<Result<Vec<_>, PipelineError>>()?;
        let in_list_expression = Expression::InList {
            expr: Box::new(expr),
            list,
            negated,
        };

        Ok(in_list_expression)
    }
}

#[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
}

getdozer / dozer / 6015696039

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