4381907514

Build # 4381907514

Build Type

push

github

Committed by GitHub

Commit Message

feat: implement tracing using open telemetry (#1176)

Run Details

510 of 510 new or added lines in 31 files covered. (100.0%)

27878 of 39615 relevant lines covered (70.37%)

47752.4 hits per line

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

19.39

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

use crate::pipeline::errors::PipelineError;
use crate::pipeline::expression::execution::{Expression, ExpressionExecutor};
use dozer_types::rust_decimal::Decimal;
use dozer_types::types::Schema;
use dozer_types::{
    ordered_float::OrderedFloat,
    types::{Field, Record},
};
use num_traits::cast::*;
use std::ops::Neg;

macro_rules! define_math_operator {
    ($id:ident, $op:expr, $fct:expr, $t: expr) => {
        pub fn $id(
            schema: &Schema,
            left: &Expression,
            right: &Expression,
            record: &Record,
        ) -> Result<Field, PipelineError> {
            let left_p = left.evaluate(&record, schema)?;
            let right_p = right.evaluate(&record, schema)?;

            match left_p {
                Field::Timestamp(left_v) => match right_p {
                    Field::Timestamp(right_v) => match $op {
                        "-" => {
                            let duration = left_v - right_v;
                            duration
                                .num_nanoseconds()
                                .ok_or(PipelineError::UnableToCast(
                                    format!("{}", duration),
                                    "i64".to_string(),
                                ))
                                .map(Field::Int)
                        }
                        _ => Err(PipelineError::InvalidTypeComparison(
                            left_p,
                            right_p,
                            $op.to_string(),
                        )),
                    },
                    _ => Err(PipelineError::InvalidTypeComparison(
                        left_p,
                        right_p,
                        $op.to_string(),
                    )),
                },
                Field::Float(left_v) => match right_p {
                    // left: Float, right: Int
                    Field::Int(right_v) => Ok(Field::Float($fct(
                        left_v,
                        OrderedFloat::<f64>::from_i64(right_v).ok_or(
                            PipelineError::UnableToCast(format!("{}", right_v), "f64".to_string()),
                        )?,
                    ))),
                    // left: Float, right: UInt
                    Field::UInt(right_v) => Ok(Field::Float($fct(
                        left_v,
                        OrderedFloat::<f64>::from_u64(right_v).ok_or(
                            PipelineError::UnableToCast(format!("{}", right_v), "f64".to_string()),
                        )?,
                    ))),
                    // left: Float, right: Float
                    Field::Float(right_v) => Ok(Field::Float($fct(left_v, right_v))),
                    // left: Float, right: Decimal
                    Field::Decimal(right_v) => Ok(Field::Decimal($fct(
                        Decimal::from_f64(left_v.to_f64().ok_or(PipelineError::UnableToCast(
                            format!("{}", left_v),
                            "f64".to_string(),
                        ))?)
                        .ok_or(PipelineError::UnableToCast(
                            format!("{}", left_v),
                            "Decimal".to_string(),
                        ))?,
                        right_v,
                    ))),
                    _ => Err(PipelineError::InvalidTypeComparison(
                        left_p,
                        right_p,
                        $op.to_string(),
                    )),
                },
                Field::Int(left_v) => match right_p {
                    // left: Int, right: Int
                    Field::Int(right_v) => {
                        return match ($t) {
                            // When Int / Int division happens
                            1 => Ok(Field::Float($fct(
                                OrderedFloat::<f64>::from_i64(left_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", left_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                                OrderedFloat::<f64>::from_i64(right_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", right_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                            ))),
                            // When it's not division operation
                            _ => Ok(Field::Int($fct(left_v, right_v))),
                        };
                    }
                    // left: Int, right: UInt
                    Field::UInt(right_v) => {
                        return match ($t) {
                            // When Int / UInt division happens
                            1 => Ok(Field::Float($fct(
                                OrderedFloat::<f64>::from_i64(left_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", left_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                                OrderedFloat::<f64>::from_u64(right_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", right_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                            ))),
                            // When it's not division operation
                            _ => Ok(Field::Int($fct(
                                left_v,
                                right_v.to_i64().ok_or(PipelineError::UnableToCast(
                                    format!("{}", right_v),
                                    "i64".to_string(),
                                ))?,
                            ))),
                        };
                    }
                    // left: Int, right: Float
                    Field::Float(right_v) => Ok(Field::Float($fct(
                        OrderedFloat::<f64>::from_i64(left_v).ok_or(
                            PipelineError::UnableToCast(format!("{}", left_v), "f64".to_string()),
                        )?,
                        right_v,
                    ))),
                    // left: Int, right: Decimal
                    Field::Decimal(right_v) => Ok(Field::Decimal($fct(
                        Decimal::from_i64(left_v).ok_or(PipelineError::UnableToCast(
                            format!("{}", left_v),
                            "Decimal".to_string(),
                        ))?,
                        right_v,
                    ))),
                    _ => Err(PipelineError::InvalidTypeComparison(
                        left_p,
                        right_p,
                        $op.to_string(),
                    )),
                },
                Field::UInt(left_v) => match right_p {
                    // left: UInt, right: Int
                    Field::Int(right_v) => {
                        return match ($t) {
                            // When UInt / Int division happens
                            1 => Ok(Field::Float($fct(
                                OrderedFloat::<f64>::from_u64(left_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", left_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                                OrderedFloat::<f64>::from_i64(right_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", right_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                            ))),
                            // When it's not division operation
                            _ => Ok(Field::Int($fct(
                                left_v.to_i64().ok_or(PipelineError::UnableToCast(
                                    format!("{}", left_v),
                                    "i64".to_string(),
                                ))?,
                                right_v,
                            ))),
                        };
                    }
                    // left: UInt, right: UInt
                    Field::UInt(right_v) => {
                        return match ($t) {
                            // When UInt / UInt division happens
                            1 => Ok(Field::Float($fct(
                                OrderedFloat::<f64>::from_u64(left_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", left_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                                OrderedFloat::<f64>::from_u64(right_v).ok_or(
                                    PipelineError::UnableToCast(
                                        format!("{}", right_v),
                                        "f64".to_string(),
                                    ),
                                )?,
                            ))),
                            // When it's not division operation
                            _ => Ok(Field::UInt($fct(left_v, right_v))),
                        };
                    }
                    // left: UInt, right: Float
                    Field::Float(right_v) => Ok(Field::Float($fct(
                        OrderedFloat::<f64>::from_u64(left_v).ok_or(
                            PipelineError::UnableToCast(format!("{}", left_v), "f64".to_string()),
                        )?,
                        right_v,
                    ))),
                    // left: UInt, right: Decimal
                    Field::Decimal(right_v) => Ok(Field::Decimal($fct(
                        Decimal::from_i64(left_v.to_i64().ok_or(PipelineError::UnableToCast(
                            format!("{}", left_v),
                            "i64".to_string(),
                        ))?)
                        .ok_or(PipelineError::UnableToCast(
                            format!("{}", left_v),
                            "Decimal".to_string(),
                        ))?,
                        right_v,
                    ))),
                    _ => Err(PipelineError::InvalidTypeComparison(
                        left_p,
                        right_p,
                        $op.to_string(),
                    )),
                },
                Field::Decimal(left_v) => {
                    match right_p {
                        // left: Decimal, right: Int
                        Field::Int(right_v) => Ok(Field::Decimal($fct(
                            left_v,
                            Decimal::from_i64(right_v).ok_or(PipelineError::UnableToCast(
                                format!("{}", left_v),
                                "Decimal".to_string(),
                            ))?,
                        ))),
                        // left: Decimal, right: UInt
                        Field::UInt(right_v) => Ok(Field::Decimal($fct(
                            left_v,
                            Decimal::from_i64(right_v.to_i64().ok_or(
                                PipelineError::UnableToCast(
                                    format!("{}", right_v),
                                    "i64".to_string(),
                                ),
                            )?)
                            .ok_or(PipelineError::UnableToCast(
                                format!("{}", right_v),
                                "Decimal".to_string(),
                            ))?,
                        ))),
                        // left: Decimal, right: Float
                        Field::Float(right_v) => Ok(Field::Decimal($fct(
                            left_v,
                            Decimal::from_f64(right_v.to_f64().ok_or(
                                PipelineError::UnableToCast(
                                    format!("{}", right_v),
                                    "f64".to_string(),
                                ),
                            )?)
                            .ok_or(PipelineError::UnableToCast(
                                format!("{}", right_v),
                                "Decimal".to_string(),
                            ))?,
                        ))),
                        // left: Decimal, right: Decimal
                        Field::Decimal(right_v) => Ok(Field::Decimal($fct(left_v, right_v))),
                        _ => Err(PipelineError::InvalidTypeComparison(
                            left_p,
                            right_p,
                            $op.to_string(),
                        )),
                    }
                }
                _ => Err(PipelineError::InvalidTypeComparison(
                    left_p,
                    right_p,
                    $op.to_string(),
                )),
            }
        }
    };
}

define_math_operator!(evaluate_add, "+", |a, b| { a + b }, 0);
define_math_operator!(evaluate_sub, "-", |a, b| { a - b }, 0);
define_math_operator!(evaluate_mul, "*", |a, b| { a * b }, 0);
define_math_operator!(evaluate_div, "/", |a, b| { a / b }, 1);
define_math_operator!(evaluate_mod, "%", |a, b| { a % b }, 0);

pub fn evaluate_plus(
    schema: &Schema,
    expression: &Expression,
    record: &Record,
) -> Result<Field, PipelineError> {
    let expression_result = expression.evaluate(record, schema)?;
    match expression_result {
        Field::UInt(v) => Ok(Field::UInt(v)),
        Field::Int(v) => Ok(Field::Int(v)),
        Field::Float(v) => Ok(Field::Float(v)),
        Field::Decimal(v) => Ok(Field::Decimal(v)),
        not_supported_field => Err(PipelineError::InvalidType(
            not_supported_field,
            "+".to_string(),
        )),
    }
}

pub fn evaluate_minus(
    schema: &Schema,
    expression: &Expression,
    record: &Record,
) -> Result<Field, PipelineError> {
    let expression_result = expression.evaluate(record, schema)?;
    match expression_result {
        Field::UInt(v) => Ok(Field::UInt(v)),
        Field::Int(v) => Ok(Field::Int(-v)),
        Field::Float(v) => Ok(Field::Float(-v)),
        Field::Decimal(v) => Ok(Field::Decimal(v.neg())),
        not_supported_field => Err(PipelineError::InvalidType(
            not_supported_field,
            "-".to_string(),
        )),
    }
}

1	use crate::pipeline::errors::PipelineError;
2	use crate::pipeline::expression::execution::{Expression, ExpressionExecutor};
3	use dozer_types::rust_decimal::Decimal;
4	use dozer_types::types::Schema;
5	use dozer_types::{
6	ordered_float::OrderedFloat,
7	types::{Field, Record},
8	};
9	use num_traits::cast::*;
10	use std::ops::Neg;
11
12	macro_rules! define_math_operator {
13	($id:ident, $op:expr, $fct:expr, $t: expr) => {
14	pub fn $id(	20✔
15	schema: &Schema,	20✔
16	left: &Expression,	20✔
17	right: &Expression,	20✔
18	record: &Record,	20✔
19	) -> Result<Field, PipelineError> {	20✔
20	let left_p = left.evaluate(&record, schema)?;	20✔
21	let right_p = right.evaluate(&record, schema)?;	20✔
22
23	match left_p {	20✔
24	Field::Timestamp(left_v) => match right_p {	3✔
25	Field::Timestamp(right_v) => match $op {	3✔
26	"-" => {	3✔
27	let duration = left_v - right_v;	3✔
28	duration	3✔
29	.num_nanoseconds()	3✔
30	.ok_or(PipelineError::UnableToCast(	3✔
31	format!("{}", duration),	3✔
32	"i64".to_string(),	3✔
33	))	3✔
34	.map(Field::Int)	3✔
35	}
36	_ => Err(PipelineError::InvalidTypeComparison(	×
37	left_p,	×
38	right_p,	×
39	$op.to_string(),	×
40	)),	×
41	},
42	_ => Err(PipelineError::InvalidTypeComparison(	×
43	left_p,	×
44	right_p,	×
45	$op.to_string(),	×
46	)),	×
47	},	×
48	Field::Float(left_v) => match right_p {	5✔
49	// left: Float, right: Int
50	Field::Int(right_v) => Ok(Field::Float($fct(	×
51	left_v,	×
52	OrderedFloat::<f64>::from_i64(right_v).ok_or(	×
53	PipelineError::UnableToCast(format!("{}", right_v), "f64".to_string()),	×
54	)?,	×
55	))),	×
56	// left: Float, right: UInt
57	Field::UInt(right_v) => Ok(Field::Float($fct(	×
58	left_v,	×
59	OrderedFloat::<f64>::from_u64(right_v).ok_or(	×
60	PipelineError::UnableToCast(format!("{}", right_v), "f64".to_string()),	×
61	)?,	×
62	))),	×
63	// left: Float, right: Float	×
64	Field::Float(right_v) => Ok(Field::Float($fct(left_v, right_v))),	×
65	// left: Float, right: Decimal	×
66	Field::Decimal(right_v) => Ok(Field::Decimal($fct(	5✔
67	Decimal::from_f64(left_v.to_f64().ok_or(PipelineError::UnableToCast(	5✔
68	format!("{}", left_v),	5✔
69	"f64".to_string(),	5✔
70	))?)	5✔
71	.ok_or(PipelineError::UnableToCast(	5✔
72	format!("{}", left_v),	5✔
73	"Decimal".to_string(),	5✔
74	))?,	5✔
75	right_v,	5✔
76	))),
77	_ => Err(PipelineError::InvalidTypeComparison(	×
78	left_p,	×
79	right_p,	×
80	$op.to_string(),	×
81	)),	×
82	},
83	Field::Int(left_v) => match right_p {	7✔
84	// left: Int, right: Int	×
85	Field::Int(right_v) => {	2✔
86	return match ($t) {	×
87	// When Int / Int division happens	×
88	1 => Ok(Field::Float($fct(	×
89	OrderedFloat::<f64>::from_i64(left_v).ok_or(	×
90	PipelineError::UnableToCast(	×
91	format!("{}", left_v),	×
92	"f64".to_string(),	×
93	),	×
94	)?,	×
95	OrderedFloat::<f64>::from_i64(right_v).ok_or(	×
96	PipelineError::UnableToCast(	×
97	format!("{}", right_v),	×
98	"f64".to_string(),	×
99	),	×
100	)?,	×
101	))),	×
102	// When it's not division operation
103	_ => Ok(Field::Int($fct(left_v, right_v))),	2✔
104	};
105	}	×
106	// left: Int, right: UInt	×
107	Field::UInt(right_v) => {	×
108	return match ($t) {	×
109	// When Int / UInt division happens	×
110	1 => Ok(Field::Float($fct(	×
111	OrderedFloat::<f64>::from_i64(left_v).ok_or(	×
112	PipelineError::UnableToCast(	×
113	format!("{}", left_v),	×
114	"f64".to_string(),	×
115	),	×
116	)?,	×
117	OrderedFloat::<f64>::from_u64(right_v).ok_or(	×
118	PipelineError::UnableToCast(	×
119	format!("{}", right_v),	×
120	"f64".to_string(),	×
121	),	×
122	)?,	×
123	))),	×
124	// When it's not division operation	×
125	_ => Ok(Field::Int($fct(
126	left_v,	×
127	right_v.to_i64().ok_or(PipelineError::UnableToCast(	×
128	format!("{}", right_v),	×
129	"i64".to_string(),	×
130	))?,	×
131	))),	×
132	};	×
133	}	×
134	// left: Int, right: Float
135	Field::Float(right_v) => Ok(Field::Float($fct(	×
136	OrderedFloat::<f64>::from_i64(left_v).ok_or(	×
137	PipelineError::UnableToCast(format!("{}", left_v), "f64".to_string()),	×
138	)?,	×
139	right_v,	×
140	))),	×
141	// left: Int, right: Decimal	×
142	Field::Decimal(right_v) => Ok(Field::Decimal($fct(	5✔
143	Decimal::from_i64(left_v).ok_or(PipelineError::UnableToCast(	5✔
144	format!("{}", left_v),	5✔
145	"Decimal".to_string(),	5✔
146	))?,	5✔
147	right_v,	5✔
148	))),
149	_ => Err(PipelineError::InvalidTypeComparison(	×
150	left_p,	×
151	right_p,	×
152	$op.to_string(),	×
153	)),	×
154	},
155	Field::UInt(left_v) => match right_p {	5✔
156	// left: UInt, right: Int	×
157	Field::Int(right_v) => {	×
158	return match ($t) {	×
159	// When UInt / Int division happens	×
160	1 => Ok(Field::Float($fct(	×
161	OrderedFloat::<f64>::from_u64(left_v).ok_or(	×
162	PipelineError::UnableToCast(	×
163	format!("{}", left_v),	×
164	"f64".to_string(),	×
165	),	×
166	)?,	×
167	OrderedFloat::<f64>::from_i64(right_v).ok_or(	×
168	PipelineError::UnableToCast(	×
169	format!("{}", right_v),	×
170	"f64".to_string(),	×
171	),	×
172	)?,	×
173	))),	×
174	// When it's not division operation	×
175	_ => Ok(Field::Int($fct(
176	left_v.to_i64().ok_or(PipelineError::UnableToCast(	×
177	format!("{}", left_v),	×
178	"i64".to_string(),	×
179	))?,	×
180	right_v,	×
181	))),
182	};
183	}	×
184	// left: UInt, right: UInt	×
185	Field::UInt(right_v) => {	×
186	return match ($t) {	×
187	// When UInt / UInt division happens	×
188	1 => Ok(Field::Float($fct(	×
189	OrderedFloat::<f64>::from_u64(left_v).ok_or(	×
190	PipelineError::UnableToCast(	×
191	format!("{}", left_v),	×
192	"f64".to_string(),	×
193	),	×
194	)?,	×
195	OrderedFloat::<f64>::from_u64(right_v).ok_or(	×
196	PipelineError::UnableToCast(	×
197	format!("{}", right_v),	×
198	"f64".to_string(),	×
199	),	×
200	)?,	×
201	))),	×
202	// When it's not division operation	×
203	_ => Ok(Field::UInt($fct(left_v, right_v))),	×
204	};	×
205	}	×
206	// left: UInt, right: Float
207	Field::Float(right_v) => Ok(Field::Float($fct(	×
208	OrderedFloat::<f64>::from_u64(left_v).ok_or(	×
209	PipelineError::UnableToCast(format!("{}", left_v), "f64".to_string()),	×
210	)?,	×
211	right_v,	×
212	))),	×
213	// left: UInt, right: Decimal	×
214	Field::Decimal(right_v) => Ok(Field::Decimal($fct(	5✔
215	Decimal::from_i64(left_v.to_i64().ok_or(PipelineError::UnableToCast(	5✔
216	format!("{}", left_v),	5✔
217	"i64".to_string(),	5✔
218	))?)	5✔
219	.ok_or(PipelineError::UnableToCast(	5✔
220	format!("{}", left_v),	5✔
221	"Decimal".to_string(),	5✔
222	))?,	5✔
223	right_v,	5✔
224	))),
225	_ => Err(PipelineError::InvalidTypeComparison(	×
226	left_p,	×
227	right_p,	×
228	$op.to_string(),	×
229	)),	×
230	},	×
231	Field::Decimal(left_v) => {	×
232	match right_p {	×
233	// left: Decimal, right: Int	×
234	Field::Int(right_v) => Ok(Field::Decimal($fct(	×
235	left_v,	×
236	Decimal::from_i64(right_v).ok_or(PipelineError::UnableToCast(	×
237	format!("{}", left_v),	×
238	"Decimal".to_string(),	×
239	))?,	×
240	))),	×
241	// left: Decimal, right: UInt	×
242	Field::UInt(right_v) => Ok(Field::Decimal($fct(	×
243	left_v,	×
244	Decimal::from_i64(right_v.to_i64().ok_or(	×
245	PipelineError::UnableToCast(	×
246	format!("{}", right_v),	×
247	"i64".to_string(),	×
248	),	×
249	)?)	×
250	.ok_or(PipelineError::UnableToCast(	×
251	format!("{}", right_v),	×
252	"Decimal".to_string(),	×
253	))?,	×
254	))),	×
255	// left: Decimal, right: Float	×
256	Field::Float(right_v) => Ok(Field::Decimal($fct(	×
257	left_v,	×
258	Decimal::from_f64(right_v.to_f64().ok_or(	×
259	PipelineError::UnableToCast(	×
260	format!("{}", right_v),	×
261	"f64".to_string(),	×
262	),	×
263	)?)	×
264	.ok_or(PipelineError::UnableToCast(	×
265	format!("{}", right_v),	×
266	"Decimal".to_string(),	×
267	))?,	×
268	))),	×
269	// left: Decimal, right: Decimal	×
270	Field::Decimal(right_v) => Ok(Field::Decimal($fct(left_v, right_v))),	×
271	_ => Err(PipelineError::InvalidTypeComparison(	×
272	left_p,	×
273	right_p,	×
274	$op.to_string(),	×
275	)),	×
276	}	×
277	}
278	_ => Err(PipelineError::InvalidTypeComparison(	×
279	left_p,	×
280	right_p,	×
281	$op.to_string(),	×
282	)),	×
283	}	×
284	}	20✔
285	};	×
286	}	×
287
288	define_math_operator!(evaluate_add, "+", \|a, b\| { a + b }, 0);	5✔
289	define_math_operator!(evaluate_sub, "-", \|a, b\| { a - b }, 0);	3✔
290	define_math_operator!(evaluate_mul, "", \|a, b\| { a b }, 0);	3✔
291	define_math_operator!(evaluate_div, "/", \|a, b\| { a / b }, 1);	3✔
292	define_math_operator!(evaluate_mod, "%", \|a, b\| { a % b }, 0);	3✔
293		×
294	pub fn evaluate_plus(	×
295	schema: &Schema,	×
296	expression: &Expression,	×
297	record: &Record,	×
298	) -> Result<Field, PipelineError> {	×
299	let expression_result = expression.evaluate(record, schema)?;	×
300	match expression_result {	×
301	Field::UInt(v) => Ok(Field::UInt(v)),	×
302	Field::Int(v) => Ok(Field::Int(v)),	×
303	Field::Float(v) => Ok(Field::Float(v)),	×
304	Field::Decimal(v) => Ok(Field::Decimal(v)),	×
305	not_supported_field => Err(PipelineError::InvalidType(	×
306	not_supported_field,	×
307	"+".to_string(),	×
308	)),	×
309	}	×
310	}	×
311		×
312	pub fn evaluate_minus(	×
313	schema: &Schema,	×
314	expression: &Expression,	×
315	record: &Record,	×
316	) -> Result<Field, PipelineError> {	×
317	let expression_result = expression.evaluate(record, schema)?;	×
318	match expression_result {	×
319	Field::UInt(v) => Ok(Field::UInt(v)),	×
320	Field::Int(v) => Ok(Field::Int(-v)),	×
321	Field::Float(v) => Ok(Field::Float(-v)),	×
322	Field::Decimal(v) => Ok(Field::Decimal(v.neg())),	×
323	not_supported_field => Err(PipelineError::InvalidType(	×
324	not_supported_field,	×
325	"-".to_string(),	×
326	)),	×
327	}
328	}	×

getdozer / dozer / 4381907514

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