4370408272

Build Type

push

github

Committed by GitHub

Commit Message

fix: Fix compilation error introduced in #1158 (#1183)

Run Details

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

28163 of 39541 relevant lines covered (71.22%)

73625.66 hits per line

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

82.61

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

use crate::calculate_err_field;
use crate::pipeline::aggregation::aggregator::Aggregator;
use crate::pipeline::errors::PipelineError;
use crate::pipeline::expression::aggregate::AggregateFunctionType::Sum;
use dozer_core::errors::ExecutionError::InvalidType;
use dozer_types::ordered_float::OrderedFloat;
use dozer_types::rust_decimal::Decimal;
use dozer_types::types::{Field, FieldType};
use num_traits::FromPrimitive;

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

#[derive(Debug)]
struct SumAggregatorState {
    int_state: i64,
    uint_state: u64,
    float_state: f64,
    decimal_state: Decimal,
}

impl SumAggregator {
    pub fn new() -> Self {
        Self {
            current_state: SumAggregatorState {
                int_state: 0_i64,
                uint_state: 0_u64,
                float_state: 0_f64,
                decimal_state: Decimal::from_f64(0_f64).unwrap(),
            },
            return_type: None,
        }
    }
}

impl Aggregator for SumAggregator {
    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> {
        get_sum(old, &mut self.current_state, self.return_type, true)
    }

    fn insert(&mut self, new: &[Field]) -> Result<Field, PipelineError> {
        get_sum(new, &mut self.current_state, self.return_type, false)
    }
}

fn get_sum(
    fields: &[Field],
    current_state: &mut SumAggregatorState,
    return_type: Option<FieldType>,
    decr: bool,
) -> Result<Field, PipelineError> {
    match return_type {
        Some(FieldType::UInt) => {
            if decr {
                for field in fields {
                    let val = calculate_err_field!(field.to_uint(), Sum, field);
                    current_state.uint_state -= val;
                }
            } else {
                for field in fields {
                    let val = calculate_err_field!(field.to_uint(), Sum, field);
                    current_state.uint_state += val;
                }
            }
            Ok(Field::UInt(current_state.uint_state))
        }
        Some(FieldType::Int) => {
            if decr {
                for field in fields {
                    let val = calculate_err_field!(field.to_int(), Sum, field);
                    current_state.int_state -= val;
                }
            } else {
                for field in fields {
                    let val = calculate_err_field!(field.to_int(), Sum, field);
                    current_state.int_state += val;
                }
            }
            Ok(Field::Int(current_state.int_state))
        }
        Some(FieldType::Float) => {
            if decr {
                for field in fields {
                    let val = calculate_err_field!(field.to_float(), Sum, field);
                    current_state.float_state -= val;
                }
            } else {
                for field in fields {
                    let val = calculate_err_field!(field.to_float(), Sum, field);
                    current_state.float_state += val;
                }
            }
            Ok(Field::Float(OrderedFloat::from(current_state.float_state)))
        }
        Some(FieldType::Decimal) => {
            if decr {
                for field in fields {
                    let val = calculate_err_field!(field.to_decimal(), Sum, field);
                    current_state.decimal_state -= val;
                }
            } else {
                for field in fields {
                    let val = calculate_err_field!(field.to_decimal(), Sum, field);
                    current_state.decimal_state += val;
                }
            }
            Ok(Field::Decimal(current_state.decimal_state))
        }
        Some(not_supported_return_type) => Err(PipelineError::InternalExecutionError(InvalidType(
            format!("Not supported return type {not_supported_return_type} for {Sum}"),
        ))),
        None => Err(PipelineError::InternalExecutionError(InvalidType(format!(
            "Not supported None return type for {Sum}"
        )))),
    }
}

1	use crate::calculate_err_field;
2	use crate::pipeline::aggregation::aggregator::Aggregator;
3	use crate::pipeline::errors::PipelineError;
4	use crate::pipeline::expression::aggregate::AggregateFunctionType::Sum;
5	use dozer_core::errors::ExecutionError::InvalidType;
6	use dozer_types::ordered_float::OrderedFloat;
7	use dozer_types::rust_decimal::Decimal;
8	use dozer_types::types::{Field, FieldType};
9	use num_traits::FromPrimitive;
10
11	#[derive(Debug)]	×
12	pub struct SumAggregator {
13	current_state: SumAggregatorState,
14	return_type: Option<FieldType>,
15	}
16
17	#[derive(Debug)]	×
18	struct SumAggregatorState {
19	int_state: i64,
20	uint_state: u64,
21	float_state: f64,
22	decimal_state: Decimal,
23	}
24
25	impl SumAggregator {
26	pub fn new() -> Self {	24✔
27	Self {	24✔
28	current_state: SumAggregatorState {	24✔
29	int_state: 0_i64,	24✔
30	uint_state: 0_u64,	24✔
31	float_state: 0_f64,	24✔
32	decimal_state: Decimal::from_f64(0_f64).unwrap(),	24✔
33	},	24✔
34	return_type: None,	24✔
35	}	24✔
36	}	24✔
37	}
38
39	impl Aggregator for SumAggregator {
40	fn init(&mut self, return_type: FieldType) {	24✔
41	self.return_type = Some(return_type);	24✔
42	}	24✔
43
44	fn update(&mut self, old: &[Field], new: &[Field]) -> Result<Field, PipelineError> {
45	self.delete(old)?;	7✔
46	self.insert(new)	7✔
47	}	7✔
48
49	fn delete(&mut self, old: &[Field]) -> Result<Field, PipelineError> {	29✔
50	get_sum(old, &mut self.current_state, self.return_type, true)	29✔
51	}	29✔
52
53	fn insert(&mut self, new: &[Field]) -> Result<Field, PipelineError> {	31✔
54	get_sum(new, &mut self.current_state, self.return_type, false)	31✔
55	}	31✔
56	}
57
58	fn get_sum(
59	fields: &[Field],
60	current_state: &mut SumAggregatorState,
61	return_type: Option<FieldType>,
62	decr: bool,
63	) -> Result<Field, PipelineError> {
64	match return_type {	61✔
65	Some(FieldType::UInt) => {
66	if decr {	10✔
67	for field in fields {	10✔
68	let val = calculate_err_field!(field.to_uint(), Sum, field);	5✔
69	current_state.uint_state -= val;	5✔
70	}
71	} else {
72	for field in fields {	10✔
73	let val = calculate_err_field!(field.to_uint(), Sum, field);	5✔
74	current_state.uint_state += val;	5✔
75	}
76	}
77	Ok(Field::UInt(current_state.uint_state))	10✔
78	}
79	Some(FieldType::Int) => {
80	if decr {	19✔
81	for field in fields {	16✔
82	let val = calculate_err_field!(field.to_int(), Sum, field);	8✔
83	current_state.int_state -= val;	8✔
84	}
85	} else {
86	for field in fields {	22✔
87	let val = calculate_err_field!(field.to_int(), Sum, field);	11✔
88	current_state.int_state += val;	11✔
89	}
90	}
91	Ok(Field::Int(current_state.int_state))	19✔
92	}
93	Some(FieldType::Float) => {
94	if decr {	16✔
95	for field in fields {	16✔
96	let val = calculate_err_field!(field.to_float(), Sum, field);	8✔
97	current_state.float_state -= val;	8✔
98	}
99	} else {
100	for field in fields {	16✔
101	let val = calculate_err_field!(field.to_float(), Sum, field);	8✔
102	current_state.float_state += val;	8✔
103	}
104	}
105	Ok(Field::Float(OrderedFloat::from(current_state.float_state)))	16✔
106	}
107	Some(FieldType::Decimal) => {
108	if decr {	16✔
109	for field in fields {	16✔
110	let val = calculate_err_field!(field.to_decimal(), Sum, field);	8✔
111	current_state.decimal_state -= val;	8✔
112	}
113	} else {
114	for field in fields {	16✔
115	let val = calculate_err_field!(field.to_decimal(), Sum, field);	8✔
116	current_state.decimal_state += val;	8✔
117	}
118	}
119	Ok(Field::Decimal(current_state.decimal_state))	16✔
120	}
121	Some(not_supported_return_type) => Err(PipelineError::InternalExecutionError(InvalidType(	×
122	format!("Not supported return type {not_supported_return_type} for {Sum}"),	×
123	))),	×
124	None => Err(PipelineError::InternalExecutionError(InvalidType(format!(	×
125	"Not supported None return type for {Sum}"	×
126	)))),	×
127	}	×
128	}	61✔

getdozer / dozer / 4370408272

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