4763034755

Build Type

github

Committed by GitHub

Commit Message

Merge 2e63b376c into c58df4a0b

Pull Request Pull Request #1460: Update init.rs

Run Details

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

34417 of 43846 relevant lines covered (78.5%)

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

getdozer / dozer / 4763034755

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