4020902227

Build # 4020902227

Build Type

github

Committed by GitHub

Commit Message

Merge 57279c6b6 into a12da35a5

Pull Request Pull Request #743: Chore clippy fix

Run Details

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

23638 of 35485 relevant lines covered (66.61%)

38417.79 hits per line

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

16.07

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

use crate::argv;
use crate::pipeline::errors::PipelineError;

use crate::pipeline::expression::operator::{BinaryOperatorType, UnaryOperatorType};
use crate::pipeline::expression::scalar::common::{get_scalar_function_type, ScalarFunctionType};
use crate::pipeline::expression::scalar::string::{evaluate_trim, validate_trim, TrimType};
use dozer_types::types::{Field, FieldType, Record, Schema, SourceDefinition};

use super::aggregate::AggregateFunctionType;
use super::cast::CastOperatorType;
use super::scalar::string::{evaluate_like, get_like_operator_type};

#[derive(Clone, Debug, PartialEq)]
pub enum Expression {
    Column {
        index: usize,
    },
    Literal(Field),
    UnaryOperator {
        operator: UnaryOperatorType,
        arg: Box<Expression>,
    },
    BinaryOperator {
        left: Box<Expression>,
        operator: BinaryOperatorType,
        right: Box<Expression>,
    },
    ScalarFunction {
        fun: ScalarFunctionType,
        args: Vec<Expression>,
    },
    AggregateFunction {
        fun: AggregateFunctionType,
        args: Vec<Expression>,
    },
    Cast {
        arg: Box<Expression>,
        typ: CastOperatorType,
    },
    Trim {
        arg: Box<Expression>,
        what: Option<Box<Expression>>,
        typ: Option<TrimType>,
    },
    Like {
        arg: Box<Expression>,
        pattern: Box<Expression>,
        escape: Option<char>,
    },
}

pub struct ExpressionType {
    pub return_type: FieldType,
    pub nullable: bool,
    pub source: SourceDefinition,
}

impl ExpressionType {
    pub fn new(return_type: FieldType, nullable: bool, source: SourceDefinition) -> Self {
        Self {
            return_type,
            nullable,
            source,
        }
    }
}

impl Expression {}

pub trait ExpressionExecutor: Send + Sync {
    fn evaluate(&self, record: &Record, schema: &Schema) -> Result<Field, PipelineError>;
    fn get_type(&self, schema: &Schema) -> Result<ExpressionType, PipelineError>;
}

impl ExpressionExecutor for Expression {
    fn evaluate(&self, record: &Record, schema: &Schema) -> Result<Field, PipelineError> {
        match self {
            Expression::Literal(field) => Ok(field.clone()),
            Expression::Column { index } => Ok(record
                .get_value(*index)
                .map_err(|_e| {
                    PipelineError::InvalidInputType(format!("{} is an invalid field index", *index))
                })?
                .clone()),
            Expression::BinaryOperator {
                left,
                operator,
                right,
            } => operator.evaluate(schema, left, right, record),
            Expression::ScalarFunction { fun, args } => fun.evaluate(schema, args, record),
            Expression::UnaryOperator { operator, arg } => operator.evaluate(schema, arg, record),
            Expression::AggregateFunction { fun, args: _ } => {
                Err(PipelineError::InvalidExpression(format!(
                    "Aggregate Function {fun:?} should not be executed at this point"
                )))
            }
            Expression::Trim { typ, what, arg } => evaluate_trim(schema, arg, what, typ, record),
            Expression::Like {
                arg,
                pattern,
                escape,
            } => evaluate_like(schema, arg, pattern, *escape, record),
            Expression::Cast { arg, typ } => typ.evaluate(schema, arg, record),
        }
    }

    fn get_type(&self, schema: &Schema) -> Result<ExpressionType, PipelineError> {
        match self {
            Expression::Literal(field) => {
                let r = get_field_type(field).ok_or_else(|| {
                    PipelineError::InvalidExpression(
                        "literal expression cannot be null".to_string(),
                    )
                })?;
                Ok(ExpressionType::new(r, false, SourceDefinition::Dynamic))
            }
            Expression::Column { index } => {
                let t = schema.fields.get(*index).unwrap();

                Ok(ExpressionType::new(t.typ, t.nullable, t.source.clone()))
            }
            Expression::UnaryOperator { operator, arg } => {
                get_unary_operator_type(operator, arg, schema)
            }
            Expression::BinaryOperator {
                left,
                operator,
                right,
            } => get_binary_operator_type(left, operator, right, schema),
            Expression::ScalarFunction { fun, args } => get_scalar_function_type(fun, args, schema),
            Expression::AggregateFunction { fun, args } => {
                get_aggregate_function_type(fun, args, schema)
            }
            Expression::Trim {
                what: _,
                typ: _,
                arg,
            } => validate_trim(arg, schema),
            Expression::Like {
                arg,
                pattern,
                escape: _,
            } => get_like_operator_type(arg, pattern, schema),
            Expression::Cast { arg, typ } => typ.get_return_type(schema, arg),
        }
    }
}

fn get_field_type(field: &Field) -> Option<FieldType> {
    match field {
        Field::Int(_) => Some(FieldType::Int),
        Field::Float(_) => Some(FieldType::Float),
        Field::Boolean(_) => Some(FieldType::Boolean),
        Field::String(_) => Some(FieldType::String),
        Field::Binary(_) => Some(FieldType::Binary),
        Field::Decimal(_) => Some(FieldType::Decimal),
        Field::Timestamp(_) => Some(FieldType::Timestamp),
        Field::Bson(_) => Some(FieldType::Bson),
        Field::Null => None,
        Field::UInt(_) => Some(FieldType::UInt),
        Field::Text(_) => Some(FieldType::Text),
        Field::Date(_) => Some(FieldType::Date),
    }
}

fn get_unary_operator_type(
    operator: &UnaryOperatorType,
    expression: &Expression,
    schema: &Schema,
) -> Result<ExpressionType, PipelineError> {
    let field_type = expression.get_type(schema)?;
    match operator {
        UnaryOperatorType::Not => match field_type.return_type {
            FieldType::Boolean => Ok(field_type),
            field_type => Err(PipelineError::InvalidExpression(format!(
                "cannot apply NOT to {field_type:?}"
            ))),
        },
        UnaryOperatorType::Plus => Ok(field_type),
        UnaryOperatorType::Minus => Ok(field_type),
    }
}

fn get_binary_operator_type(
    left: &Expression,
    operator: &BinaryOperatorType,
    right: &Expression,
    schema: &Schema,
) -> Result<ExpressionType, PipelineError> {
    let left_field_type = left.get_type(schema)?;
    let right_field_type = right.get_type(schema)?;
    match operator {
        BinaryOperatorType::Eq
        | BinaryOperatorType::Ne
        | BinaryOperatorType::Gt
        | BinaryOperatorType::Gte
        | BinaryOperatorType::Lt
        | BinaryOperatorType::Lte => Ok(ExpressionType::new(
            FieldType::Boolean,
            false,
            SourceDefinition::Dynamic,
        )),

        BinaryOperatorType::And | BinaryOperatorType::Or => {
            match (left_field_type.return_type, right_field_type.return_type) {
                (FieldType::Boolean, FieldType::Boolean) => Ok(ExpressionType::new(
                    FieldType::Boolean,
                    false,
                    SourceDefinition::Dynamic,
                )),
                (left_field_type, right_field_type) => {
                    Err(PipelineError::InvalidExpression(format!(
                        "cannot apply {operator:?} to {left_field_type:?} and {right_field_type:?}"
                    )))
                }
            }
        }

        BinaryOperatorType::Add | BinaryOperatorType::Sub | BinaryOperatorType::Mul => {
            match (left_field_type.return_type, right_field_type.return_type) {
                (FieldType::Int, FieldType::Int) => Ok(ExpressionType::new(
                    FieldType::Int,
                    false,
                    SourceDefinition::Dynamic,
                )),
                (FieldType::Int, FieldType::Float)
                | (FieldType::Float, FieldType::Int)
                | (FieldType::Float, FieldType::Float) => Ok(ExpressionType::new(
                    FieldType::Float,
                    false,
                    SourceDefinition::Dynamic,
                )),
                (left_field_type, right_field_type) => {
                    Err(PipelineError::InvalidExpression(format!(
                        "cannot apply {operator:?} to {left_field_type:?} and {right_field_type:?}"
                    )))
                }
            }
        }
        BinaryOperatorType::Div | BinaryOperatorType::Mod => {
            match (left_field_type.return_type, right_field_type.return_type) {
                (FieldType::Int, FieldType::Float)
                | (FieldType::Float, FieldType::Int)
                | (FieldType::Float, FieldType::Float) => Ok(ExpressionType::new(
                    FieldType::Float,
                    false,
                    SourceDefinition::Dynamic,
                )),
                (left_field_type, right_field_type) => {
                    Err(PipelineError::InvalidExpression(format!(
                        "cannot apply {operator:?} to {left_field_type:?} and {right_field_type:?}"
                    )))
                }
            }
        }
    }
}

fn get_aggregate_function_type(
    function: &AggregateFunctionType,
    args: &[Expression],
    schema: &Schema,
) -> Result<ExpressionType, PipelineError> {
    match function {
        AggregateFunctionType::Avg => Ok(ExpressionType::new(
            FieldType::Float,
            false,
            SourceDefinition::Dynamic,
        )),
        AggregateFunctionType::Count => Ok(ExpressionType::new(
            FieldType::Int,
            false,
            SourceDefinition::Dynamic,
        )),
        AggregateFunctionType::Max => argv!(args, 0, AggregateFunctionType::Max)?.get_type(schema),
        AggregateFunctionType::Median => {
            argv!(args, 0, AggregateFunctionType::Median)?.get_type(schema)
        }
        AggregateFunctionType::Min => argv!(args, 0, AggregateFunctionType::Min)?.get_type(schema),
        AggregateFunctionType::Sum => argv!(args, 0, AggregateFunctionType::Sum)?.get_type(schema),
        AggregateFunctionType::Stddev => Ok(ExpressionType::new(
            FieldType::Float,
            false,
            SourceDefinition::Dynamic,
        )),
        AggregateFunctionType::Variance => Ok(ExpressionType::new(
            FieldType::Float,
            false,
            SourceDefinition::Dynamic,
        )),
    }
}

1	use crate::argv;
2	use crate::pipeline::errors::PipelineError;
3
4	use crate::pipeline::expression::operator::{BinaryOperatorType, UnaryOperatorType};
5	use crate::pipeline::expression::scalar::common::{get_scalar_function_type, ScalarFunctionType};
6	use crate::pipeline::expression::scalar::string::{evaluate_trim, validate_trim, TrimType};
7	use dozer_types::types::{Field, FieldType, Record, Schema, SourceDefinition};
8
9	use super::aggregate::AggregateFunctionType;
10	use super::cast::CastOperatorType;
11	use super::scalar::string::{evaluate_like, get_like_operator_type};
12
13	#[derive(Clone, Debug, PartialEq)]	111,466✔
14	pub enum Expression {
15	Column {
16	index: usize,
17	},
18	Literal(Field),
19	UnaryOperator {
20	operator: UnaryOperatorType,
21	arg: Box<Expression>,
22	},
23	BinaryOperator {
24	left: Box<Expression>,
25	operator: BinaryOperatorType,
26	right: Box<Expression>,
27	},
28	ScalarFunction {
29	fun: ScalarFunctionType,
30	args: Vec<Expression>,
31	},
32	AggregateFunction {
33	fun: AggregateFunctionType,
34	args: Vec<Expression>,
35	},
36	Cast {
37	arg: Box<Expression>,
38	typ: CastOperatorType,
39	},
40	Trim {
41	arg: Box<Expression>,
42	what: Option<Box<Expression>>,
43	typ: Option<TrimType>,
44	},
45	Like {
46	arg: Box<Expression>,
47	pattern: Box<Expression>,
48	escape: Option<char>,
49	},
50	}
51
52	pub struct ExpressionType {
53	pub return_type: FieldType,
54	pub nullable: bool,
55	pub source: SourceDefinition,
56	}
57
58	impl ExpressionType {	×
59	pub fn new(return_type: FieldType, nullable: bool, source: SourceDefinition) -> Self {	17,996✔
60	Self {	17,996✔
61	return_type,	17,996✔
62	nullable,	17,996✔
63	source,	17,996✔
64	}	17,996✔
65	}	17,996✔
66	}
67
68	impl Expression {}
69
70	pub trait ExpressionExecutor: Send + Sync {
71	fn evaluate(&self, record: &Record, schema: &Schema) -> Result<Field, PipelineError>;
72	fn get_type(&self, schema: &Schema) -> Result<ExpressionType, PipelineError>;
73	}
74		×
75	impl ExpressionExecutor for Expression {	×
76	fn evaluate(&self, record: &Record, schema: &Schema) -> Result<Field, PipelineError> {	186,506✔
77	match self {	186,506✔
78	Expression::Literal(field) => Ok(field.clone()),	14,936✔
79	Expression::Column { index } => Ok(record	154,807✔
80	.get_value(*index)	154,807✔
81	.map_err(\|_e\| {	154,807✔
82	PipelineError::InvalidInputType(format!("{} is an invalid field index", *index))	×
83	})?	154,807✔
84	.clone()),	154,826✔
85	Expression::BinaryOperator {	×
86	left,	16,694✔
87	operator,	16,694✔
88	right,	16,694✔
89	} => operator.evaluate(schema, left, right, record),	16,694✔
90	Expression::ScalarFunction { fun, args } => fun.evaluate(schema, args, record),	8✔
91	Expression::UnaryOperator { operator, arg } => operator.evaluate(schema, arg, record),	1✔
92	Expression::AggregateFunction { fun, args: _ } => {	×
93	Err(PipelineError::InvalidExpression(format!(	×
94	"Aggregate Function {fun:?} should not be executed at this point"	×
95	)))	×
96	}	×
97	Expression::Trim { typ, what, arg } => evaluate_trim(schema, arg, what, typ, record),	28✔
98	Expression::Like {	×
99	arg,	×
100	pattern,	×
101	escape,	×
102	} => evaluate_like(schema, arg, pattern, *escape, record),	×
103	Expression::Cast { arg, typ } => typ.evaluate(schema, arg, record),	32✔
104	}	×
105	}	186,525✔
106		×
107	fn get_type(&self, schema: &Schema) -> Result<ExpressionType, PipelineError> {	18,051✔
108	match self {	18,051✔
109	Expression::Literal(field) => {	×
110	let r = get_field_type(field).ok_or_else(\|\| {	×
111	PipelineError::InvalidExpression(	×
112	"literal expression cannot be null".to_string(),	×
113	)	×
114	})?;	×
115	Ok(ExpressionType::new(r, false, SourceDefinition::Dynamic))	×
116	}	×
117	Expression::Column { index } => {	17,992✔
118	let t = schema.fields.get(*index).unwrap();	17,992✔
119		17,992✔
120	Ok(ExpressionType::new(t.typ, t.nullable, t.source.clone()))	17,992✔
121	}	×
122	Expression::UnaryOperator { operator, arg } => {	×
123	get_unary_operator_type(operator, arg, schema)	×
124	}	×
125	Expression::BinaryOperator {	×
126	left,	×
127	operator,	×
128	right,	×
129	} => get_binary_operator_type(left, operator, right, schema),	×
130	Expression::ScalarFunction { fun, args } => get_scalar_function_type(fun, args, schema),	8✔
131	Expression::AggregateFunction { fun, args } => {	×
132	get_aggregate_function_type(fun, args, schema)	×
133	}	×
134	Expression::Trim {
135	what: _,	×
136	typ: _,	×
137	arg,	19✔
138	} => validate_trim(arg, schema),	19✔
139	Expression::Like {	×
140	arg,	×
141	pattern,	×
142	escape: _,	×
143	} => get_like_operator_type(arg, pattern, schema),	×
144	Expression::Cast { arg, typ } => typ.get_return_type(schema, arg),	32✔
145	}	×
146	}	18,051✔
147	}	×
148		×
149	fn get_field_type(field: &Field) -> Option<FieldType> {	×
150	match field {	×
151	Field::Int(_) => Some(FieldType::Int),	×
152	Field::Float(_) => Some(FieldType::Float),	×
153	Field::Boolean(_) => Some(FieldType::Boolean),	×
154	Field::String(_) => Some(FieldType::String),	×
155	Field::Binary(_) => Some(FieldType::Binary),	×
156	Field::Decimal(_) => Some(FieldType::Decimal),	×
157	Field::Timestamp(_) => Some(FieldType::Timestamp),	×
158	Field::Bson(_) => Some(FieldType::Bson),	×
159	Field::Null => None,	×
160	Field::UInt(_) => Some(FieldType::UInt),	×
161	Field::Text(_) => Some(FieldType::Text),	×
162	Field::Date(_) => Some(FieldType::Date),	×
163	}	×
164	}	×
165		×
166	fn get_unary_operator_type(	×
167	operator: &UnaryOperatorType,	×
168	expression: &Expression,	×
169	schema: &Schema,	×
170	) -> Result<ExpressionType, PipelineError> {	×
171	let field_type = expression.get_type(schema)?;	×
172	match operator {	×
173	UnaryOperatorType::Not => match field_type.return_type {	×
174	FieldType::Boolean => Ok(field_type),	×
175	field_type => Err(PipelineError::InvalidExpression(format!(	×
176	"cannot apply NOT to {field_type:?}"	×
177	))),	×
178	},	×
179	UnaryOperatorType::Plus => Ok(field_type),	×
180	UnaryOperatorType::Minus => Ok(field_type),	×
181	}	×
182	}	×
183		×
184	fn get_binary_operator_type(	×
185	left: &Expression,	×
186	operator: &BinaryOperatorType,	×
187	right: &Expression,	×
188	schema: &Schema,	×
189	) -> Result<ExpressionType, PipelineError> {	×
190	let left_field_type = left.get_type(schema)?;	×
191	let right_field_type = right.get_type(schema)?;	×
192	match operator {	×
193	BinaryOperatorType::Eq	×
194	\| BinaryOperatorType::Ne	×
195	\| BinaryOperatorType::Gt
196	\| BinaryOperatorType::Gte
197	\| BinaryOperatorType::Lt
198	\| BinaryOperatorType::Lte => Ok(ExpressionType::new(	×
199	FieldType::Boolean,	×
200	false,	×
201	SourceDefinition::Dynamic,	×
202	)),	×
203		×
204	BinaryOperatorType::And \| BinaryOperatorType::Or => {	×
205	match (left_field_type.return_type, right_field_type.return_type) {	×
206	(FieldType::Boolean, FieldType::Boolean) => Ok(ExpressionType::new(	×
207	FieldType::Boolean,	×
208	false,	×
209	SourceDefinition::Dynamic,	×
210	)),	×
211	(left_field_type, right_field_type) => {	×
212	Err(PipelineError::InvalidExpression(format!(	×
213	"cannot apply {operator:?} to {left_field_type:?} and {right_field_type:?}"	×
214	)))	×
215	}	×
216	}	×
217	}	×
218
219	BinaryOperatorType::Add \| BinaryOperatorType::Sub \| BinaryOperatorType::Mul => {
220	match (left_field_type.return_type, right_field_type.return_type) {	×
221	(FieldType::Int, FieldType::Int) => Ok(ExpressionType::new(	×
222	FieldType::Int,	×
223	false,	×
224	SourceDefinition::Dynamic,	×
225	)),	×
226	(FieldType::Int, FieldType::Float)	×
227	\| (FieldType::Float, FieldType::Int)	×
228	\| (FieldType::Float, FieldType::Float) => Ok(ExpressionType::new(	×
229	FieldType::Float,	×
230	false,	×
231	SourceDefinition::Dynamic,	×
232	)),	×
233	(left_field_type, right_field_type) => {	×
234	Err(PipelineError::InvalidExpression(format!(	×
235	"cannot apply {operator:?} to {left_field_type:?} and {right_field_type:?}"	×
236	)))	×
237	}	×
238	}	×
239	}	×
240	BinaryOperatorType::Div \| BinaryOperatorType::Mod => {	×
241	match (left_field_type.return_type, right_field_type.return_type) {	×
242	(FieldType::Int, FieldType::Float)	×
243	\| (FieldType::Float, FieldType::Int)
244	\| (FieldType::Float, FieldType::Float) => Ok(ExpressionType::new(	×
245	FieldType::Float,	×
246	false,	×
247	SourceDefinition::Dynamic,	×
248	)),	×
249	(left_field_type, right_field_type) => {	×
250	Err(PipelineError::InvalidExpression(format!(	×
251	"cannot apply {operator:?} to {left_field_type:?} and {right_field_type:?}"	×
252	)))	×
253	}	×
254	}	×
255	}	×
256	}	×
257	}	×
258
259	fn get_aggregate_function_type(	×
260	function: &AggregateFunctionType,	×
261	args: &[Expression],	×
262	schema: &Schema,	×
263	) -> Result<ExpressionType, PipelineError> {	×
264	match function {	×
265	AggregateFunctionType::Avg => Ok(ExpressionType::new(	×
266	FieldType::Float,	×
267	false,	×
268	SourceDefinition::Dynamic,	×
269	)),	×
270	AggregateFunctionType::Count => Ok(ExpressionType::new(	×
271	FieldType::Int,	×
272	false,	×
273	SourceDefinition::Dynamic,	×
274	)),	×
275	AggregateFunctionType::Max => argv!(args, 0, AggregateFunctionType::Max)?.get_type(schema),	×
276	AggregateFunctionType::Median => {	×
277	argv!(args, 0, AggregateFunctionType::Median)?.get_type(schema)	×
278	}	×
279	AggregateFunctionType::Min => argv!(args, 0, AggregateFunctionType::Min)?.get_type(schema),	×
280	AggregateFunctionType::Sum => argv!(args, 0, AggregateFunctionType::Sum)?.get_type(schema),	×
281	AggregateFunctionType::Stddev => Ok(ExpressionType::new(	×
282	FieldType::Float,	×
283	false,	×
284	SourceDefinition::Dynamic,	×
285	)),	×
286	AggregateFunctionType::Variance => Ok(ExpressionType::new(	×
287	FieldType::Float,	×
288	false,	×
289	SourceDefinition::Dynamic,	×
290	)),	×
291	}	×
292	}	×

getdozer / dozer / 4020902227

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

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