5610743655

pending completion

Build # 5610743655

Build Type

push

github

Committed by

web-flow

Commit Message

feat: Register errors from error manager in jaeger (#1753)

Run Details

11 of 11 new or added lines in 2 files covered. (100.0%)

42948 of 56422 relevant lines covered (76.12%)

41547.83 hits per line

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

67.92

/dozer-sql/src/pipeline/aggregation/count.rs

use crate::calculate_err_type;
use crate::pipeline::aggregation::aggregator::Aggregator;
use crate::pipeline::errors::PipelineError;
use crate::pipeline::expression::aggregate::AggregateFunctionType::Count;
use crate::pipeline::expression::execution::{Expression, ExpressionType};
use dozer_types::ordered_float::OrderedFloat;
use dozer_types::rust_decimal::Decimal;
use dozer_types::types::{Field, FieldType, Schema, SourceDefinition};
use num_traits::FromPrimitive;

pub fn validate_count(
    _args: &[Expression],
    _schema: &Schema,
) -> Result<ExpressionType, PipelineError> {
    Ok(ExpressionType::new(
        FieldType::Int,
        false,
        SourceDefinition::Dynamic,
        false,
    ))
}

#[derive(Debug)]
pub struct CountAggregator {
    current_state: u64,
    return_type: Option<FieldType>,
}

impl CountAggregator {
    pub fn new() -> Self {
        Self {
            current_state: 0_u64,
            return_type: None,
        }
    }
}

impl Aggregator for CountAggregator {
    fn init(&mut self, return_type: FieldType) {
        self.return_type = Some(return_type);
    }

    fn update(&mut self, old: &[Field], new: &[Field]) -> Result<Field, PipelineError> {
        self.delete(old)?;
        self.insert(new)
    }

    fn delete(&mut self, old: &[Field]) -> Result<Field, PipelineError> {
        self.current_state -= old.len() as u64;
        get_count(self.current_state, self.return_type)
    }

    fn insert(&mut self, new: &[Field]) -> Result<Field, PipelineError> {
        self.current_state += new.len() as u64;
        get_count(self.current_state, self.return_type)
    }
}

fn get_count(count: u64, return_type: Option<FieldType>) -> Result<Field, PipelineError> {
    match return_type {
        Some(typ) => match typ {
            FieldType::UInt => Ok(Field::UInt(count)),
            FieldType::U128 => Ok(Field::U128(count as u128)),
            FieldType::Int => Ok(Field::Int(count as i64)),
            FieldType::I128 => Ok(Field::I128(count as i128)),
            FieldType::Float => Ok(Field::Float(OrderedFloat::from(count as f64))),
            FieldType::Decimal => Ok(Field::Decimal(calculate_err_type!(
                Decimal::from_f64(count as f64),
                Count,
                FieldType::Decimal
            ))),
            FieldType::Duration => Ok(Field::Int(count as i64)),
            FieldType::Boolean
            | FieldType::String
            | FieldType::Text
            | FieldType::Date
            | FieldType::Timestamp
            | FieldType::Binary
            | FieldType::Json
            | FieldType::Point => Err(PipelineError::InvalidReturnType(format!(
                "Not supported return type {typ} for {Count}"
            ))),
        },
        None => Err(PipelineError::InvalidReturnType(format!(
            "Not supported None return type for {Count}"
        ))),
    }
}

1	use crate::calculate_err_type;
2	use crate::pipeline::aggregation::aggregator::Aggregator;
3	use crate::pipeline::errors::PipelineError;
4	use crate::pipeline::expression::aggregate::AggregateFunctionType::Count;
5	use crate::pipeline::expression::execution::{Expression, ExpressionType};
6	use dozer_types::ordered_float::OrderedFloat;
7	use dozer_types::rust_decimal::Decimal;
8	use dozer_types::types::{Field, FieldType, Schema, SourceDefinition};
9	use num_traits::FromPrimitive;
10
11	pub fn validate_count(	297✔
12	_args: &[Expression],	297✔
13	_schema: &Schema,	297✔
14	) -> Result<ExpressionType, PipelineError> {	297✔
15	Ok(ExpressionType::new(	297✔
16	FieldType::Int,	297✔
17	false,	297✔
18	SourceDefinition::Dynamic,	297✔
19	false,	297✔
20	))	297✔
21	}	297✔
22
23	#[derive(Debug)]	×
24	pub struct CountAggregator {
25	current_state: u64,
26	return_type: Option<FieldType>,
27	}
28
29	impl CountAggregator {
30	pub fn new() -> Self {	10,960✔
31	Self {	10,960✔
32	current_state: 0_u64,	10,960✔
33	return_type: None,	10,960✔
34	}	10,960✔
35	}	10,960✔
36	}
37
38	impl Aggregator for CountAggregator {
39	fn init(&mut self, return_type: FieldType) {	10,960✔
40	self.return_type = Some(return_type);	10,960✔
41	}	10,960✔
42
43	fn update(&mut self, old: &[Field], new: &[Field]) -> Result<Field, PipelineError> {
44	self.delete(old)?;	54✔
45	self.insert(new)	54✔
46	}	54✔
47
48	fn delete(&mut self, old: &[Field]) -> Result<Field, PipelineError> {	231✔
49	self.current_state -= old.len() as u64;	231✔
50	get_count(self.current_state, self.return_type)	231✔
51	}	231✔
52
53	fn insert(&mut self, new: &[Field]) -> Result<Field, PipelineError> {	11,014✔
54	self.current_state += new.len() as u64;	11,014✔
55	get_count(self.current_state, self.return_type)	11,014✔
56	}	11,014✔
57	}
58
59	fn get_count(count: u64, return_type: Option<FieldType>) -> Result<Field, PipelineError> {	11,245✔
60	match return_type {	11,245✔
61	Some(typ) => match typ {	11,245✔
62	FieldType::UInt => Ok(Field::UInt(count)),	×
63	FieldType::U128 => Ok(Field::U128(count as u128)),	×
64	FieldType::Int => Ok(Field::Int(count as i64)),	11,245✔
65	FieldType::I128 => Ok(Field::I128(count as i128)),	×
66	FieldType::Float => Ok(Field::Float(OrderedFloat::from(count as f64))),	×
67	FieldType::Decimal => Ok(Field::Decimal(calculate_err_type!(	×
68	Decimal::from_f64(count as f64),	×
69	Count,	×
70	FieldType::Decimal	×
71	))),	×
72	FieldType::Duration => Ok(Field::Int(count as i64)),	×
73	FieldType::Boolean
74	\| FieldType::String
75	\| FieldType::Text
76	\| FieldType::Date
77	\| FieldType::Timestamp
78	\| FieldType::Binary
79	\| FieldType::Json
80	\| FieldType::Point => Err(PipelineError::InvalidReturnType(format!(	×
81	"Not supported return type {typ} for {Count}"	×
82	))),	×
83	},
84	None => Err(PipelineError::InvalidReturnType(format!(	×
85	"Not supported None return type for {Count}"	×
86	))),	×
87	}
88	}	11,245✔

getdozer / dozer / 5610743655

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