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

79.71

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

use crate::pipeline::aggregation::avg::AvgAggregator;
use crate::pipeline::aggregation::count::CountAggregator;
use crate::pipeline::aggregation::max::MaxAggregator;
use crate::pipeline::aggregation::min::MinAggregator;
use crate::pipeline::aggregation::sum::SumAggregator;
use crate::pipeline::errors::PipelineError;

use dozer_core::storage::common::Database;
use dozer_core::storage::prefix_transaction::PrefixTransaction;
use dozer_types::types::{Field, FieldType};
use std::fmt::{Display, Formatter};

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
pub enum Aggregator {
    Avg,
    Count,
    Max,
    Min,
    Sum,
}

pub(crate) struct AggregationResult {
    pub value: Field,
    pub state: Option<Vec<u8>>,
}

impl AggregationResult {
    pub fn new(value: Field, state: Option<Vec<u8>>) -> Self {
        Self { value, state }
    }
}

impl Display for Aggregator {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{self:?}")
    }
}

impl Aggregator {
    pub(crate) fn get_return_type(&self, from: FieldType) -> FieldType {
        match (&self, from) {
            (Aggregator::Avg, _) => AvgAggregator::get_return_type(from),
            (Aggregator::Count, _) => CountAggregator::get_return_type(),
            (Aggregator::Max, from) => MaxAggregator::get_return_type(from),
            (Aggregator::Min, from) => MinAggregator::get_return_type(from),
            (Aggregator::Sum, from) => SumAggregator::get_return_type(from),
        }
    }

    pub(crate) fn _get_type(&self) -> u32 {
        match &self {
            Aggregator::Avg => AvgAggregator::_get_type(),
            Aggregator::Count => CountAggregator::_get_type(),
            Aggregator::Max => MaxAggregator::_get_type(),
            Aggregator::Min => MinAggregator::_get_type(),
            Aggregator::Sum => SumAggregator::_get_type(),
        }
    }

    pub(crate) fn insert(
        &self,
        cur_state: Option<&[u8]>,
        new: &Field,
        return_type: FieldType,
        txn: &mut PrefixTransaction,
        agg_db: Database,
    ) -> Result<AggregationResult, PipelineError> {
        match &self {
            Aggregator::Avg => AvgAggregator::insert(cur_state, new, return_type, txn, agg_db),
            Aggregator::Count => CountAggregator::insert(cur_state, new, return_type, txn),
            Aggregator::Max => MaxAggregator::insert(cur_state, new, return_type, txn, agg_db),
            Aggregator::Min => MinAggregator::insert(cur_state, new, return_type, txn, agg_db),
            Aggregator::Sum => SumAggregator::insert(cur_state, new, return_type, txn),
        }
    }

    pub(crate) fn update(
        &self,
        cur_state: Option<&[u8]>,
        old: &Field,
        new: &Field,
        return_type: FieldType,
        txn: &mut PrefixTransaction,
        agg_db: Database,
    ) -> Result<AggregationResult, PipelineError> {
        match &self {
            Aggregator::Avg => AvgAggregator::update(cur_state, old, new, return_type, txn, agg_db),
            Aggregator::Count => CountAggregator::update(cur_state, old, new, return_type, txn),
            Aggregator::Max => MaxAggregator::update(cur_state, old, new, return_type, txn, agg_db),
            Aggregator::Min => MinAggregator::update(cur_state, old, new, return_type, txn, agg_db),
            Aggregator::Sum => SumAggregator::update(cur_state, old, new, return_type, txn),
        }
    }

    pub(crate) fn delete(
        &self,
        cur_state: Option<&[u8]>,
        old: &Field,
        return_type: FieldType,
        txn: &mut PrefixTransaction,
        agg_db: Database,
    ) -> Result<AggregationResult, PipelineError> {
        match &self {
            Aggregator::Avg => AvgAggregator::delete(cur_state, old, return_type, txn, agg_db),
            Aggregator::Count => CountAggregator::delete(cur_state, old, return_type, txn),
            Aggregator::Max => MaxAggregator::delete(cur_state, old, return_type, txn, agg_db),
            Aggregator::Min => MinAggregator::delete(cur_state, old, return_type, txn, agg_db),
            Aggregator::Sum => SumAggregator::delete(cur_state, old, return_type, txn),
        }
    }
}

#[macro_export]
macro_rules! deserialize {
    ($stmt:expr) => {
        $stmt.try_into().unwrap()
    };
}

#[macro_export]
macro_rules! deserialize_f64 {
    ($stmt:expr) => {
        match $stmt {
            Some(v) => f64::from_be_bytes(deserialize!(v)),
            None => 0_f64,
        }
    };
}

#[macro_export]
macro_rules! deserialize_i64 {
    ($stmt:expr) => {
        match $stmt {
            Some(v) => i64::from_be_bytes(deserialize!(v)),
            None => 0_i64,
        }
    };
}

#[macro_export]
macro_rules! deserialize_u64 {
    ($stmt:expr) => {
        match $stmt {
            Some(v) => u64::from_be_bytes(deserialize!(v)),
            None => 0_u64,
        }
    };
}

#[macro_export]
macro_rules! deserialize_decimal {
    ($stmt:expr) => {
        match $stmt {
            Some(v) => Decimal::deserialize(deserialize!(v)),
            None => Decimal::from(0),
        }
    };
}

#[macro_export]
macro_rules! deserialize_u8 {
    ($stmt:expr) => {
        match $stmt {
            Some(v) => u8::from_be_bytes(deserialize!(v)),
            None => 0_u8,
        }
    };
}

#[macro_export]
macro_rules! check_nan_f64 {
    ($stmt:expr) => {
        if $stmt.is_nan() {
            0_f64
        } else {
            $stmt
        }
    };
}

#[macro_export]
macro_rules! check_nan_decimal {
    ($stmt:expr) => {
        if $stmt.is_nan() {
            dozer_types::rust_decimal::Decimal::zero()
        } else {
            $stmt
        }
    };
}

#[macro_export]
macro_rules! try_unwrap {
    ($stmt:expr) => {
        $stmt.unwrap_or_else(|e| panic!("{}", e.to_string()))
    };
}

#[macro_export]
macro_rules! to_bytes {
    ($stmt:expr) => {
        $stmt.to_be_bytes().as_slice()
    };
}

1	use crate::pipeline::aggregation::avg::AvgAggregator;
2	use crate::pipeline::aggregation::count::CountAggregator;
3	use crate::pipeline::aggregation::max::MaxAggregator;
4	use crate::pipeline::aggregation::min::MinAggregator;
5	use crate::pipeline::aggregation::sum::SumAggregator;
6	use crate::pipeline::errors::PipelineError;
7
8	use dozer_core::storage::common::Database;
9	use dozer_core::storage::prefix_transaction::PrefixTransaction;
10	use dozer_types::types::{Field, FieldType};
11	use std::fmt::{Display, Formatter};
12
13	#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]	86✔
14	pub enum Aggregator {
15	Avg,
16	Count,
17	Max,
18	Min,
19	Sum,
20	}
21
22	pub(crate) struct AggregationResult {
23	pub value: Field,
24	pub state: Option<Vec<u8>>,
25	}
26
27	impl AggregationResult {
28	pub fn new(value: Field, state: Option<Vec<u8>>) -> Self {	16,970✔
29	Self { value, state }	16,970✔
30	}	16,970✔
31	}
32
33	impl Display for Aggregator {
34	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {	×
35	write!(f, "{self:?}")	×
36	}	×
37	}
38
39	impl Aggregator {
40	pub(crate) fn get_return_type(&self, from: FieldType) -> FieldType {	80✔
41	match (&self, from) {	80✔
42	(Aggregator::Avg, _) => AvgAggregator::get_return_type(from),	×
43	(Aggregator::Count, _) => CountAggregator::get_return_type(),	79✔
44	(Aggregator::Max, from) => MaxAggregator::get_return_type(from),	×
45	(Aggregator::Min, from) => MinAggregator::get_return_type(from),	×
46	(Aggregator::Sum, from) => SumAggregator::get_return_type(from),	1✔
47	}
48	}	80✔
49
50	pub(crate) fn _get_type(&self) -> u32 {	×
51	match &self {	×
52	Aggregator::Avg => AvgAggregator::_get_type(),	×
53	Aggregator::Count => CountAggregator::_get_type(),	×
54	Aggregator::Max => MaxAggregator::_get_type(),	×
55	Aggregator::Min => MinAggregator::_get_type(),	×
56	Aggregator::Sum => SumAggregator::_get_type(),	×
57	}
58	}	×
59
60	pub(crate) fn insert(	16,773✔
61	&self,	16,773✔
62	cur_state: Option<&[u8]>,	16,773✔
63	new: &Field,	16,773✔
64	return_type: FieldType,	16,773✔
65	txn: &mut PrefixTransaction,	16,773✔
66	agg_db: Database,	16,773✔
67	) -> Result<AggregationResult, PipelineError> {	16,773✔
68	match &self {	16,773✔
69	Aggregator::Avg => AvgAggregator::insert(cur_state, new, return_type, txn, agg_db),	22✔
70	Aggregator::Count => CountAggregator::insert(cur_state, new, return_type, txn),	16,660✔
71	Aggregator::Max => MaxAggregator::insert(cur_state, new, return_type, txn, agg_db),	34✔
72	Aggregator::Min => MinAggregator::insert(cur_state, new, return_type, txn, agg_db),	34✔
73	Aggregator::Sum => SumAggregator::insert(cur_state, new, return_type, txn),	23✔
74	}
75	}	16,773✔
76
77	pub(crate) fn update(	44✔
78	&self,	44✔
79	cur_state: Option<&[u8]>,	44✔
80	old: &Field,	44✔
81	new: &Field,	44✔
82	return_type: FieldType,	44✔
83	txn: &mut PrefixTransaction,	44✔
84	agg_db: Database,	44✔
85	) -> Result<AggregationResult, PipelineError> {	44✔
86	match &self {	44✔
87	Aggregator::Avg => AvgAggregator::update(cur_state, old, new, return_type, txn, agg_db),	7✔
88	Aggregator::Count => CountAggregator::update(cur_state, old, new, return_type, txn),	8✔
89	Aggregator::Max => MaxAggregator::update(cur_state, old, new, return_type, txn, agg_db),	11✔
90	Aggregator::Min => MinAggregator::update(cur_state, old, new, return_type, txn, agg_db),	11✔
91	Aggregator::Sum => SumAggregator::update(cur_state, old, new, return_type, txn),	7✔
92	}
93	}	44✔
94
95	pub(crate) fn delete(	153✔
96	&self,	153✔
97	cur_state: Option<&[u8]>,	153✔
98	old: &Field,	153✔
99	return_type: FieldType,	153✔
100	txn: &mut PrefixTransaction,	153✔
101	agg_db: Database,	153✔
102	) -> Result<AggregationResult, PipelineError> {	153✔
103	match &self {	153✔
104	Aggregator::Avg => AvgAggregator::delete(cur_state, old, return_type, txn, agg_db),	22✔
105	Aggregator::Count => CountAggregator::delete(cur_state, old, return_type, txn),	41✔
106	Aggregator::Max => MaxAggregator::delete(cur_state, old, return_type, txn, agg_db),	34✔
107	Aggregator::Min => MinAggregator::delete(cur_state, old, return_type, txn, agg_db),	34✔
108	Aggregator::Sum => SumAggregator::delete(cur_state, old, return_type, txn),	22✔
109	}
110	}	153✔
111	}
112
113	#[macro_export]
114	macro_rules! deserialize {
115	($stmt:expr) => {
116	$stmt.try_into().unwrap()
117	};
118	}
119
120	#[macro_export]
121	macro_rules! deserialize_f64 {
122	($stmt:expr) => {
123	match $stmt {
124	Some(v) => f64::from_be_bytes(deserialize!(v)),
125	None => 0_f64,
126	}
127	};
128	}
129
130	#[macro_export]
131	macro_rules! deserialize_i64 {
132	($stmt:expr) => {
133	match $stmt {
134	Some(v) => i64::from_be_bytes(deserialize!(v)),
135	None => 0_i64,
136	}
137	};
138	}
139
140	#[macro_export]
141	macro_rules! deserialize_u64 {
142	($stmt:expr) => {
143	match $stmt {
144	Some(v) => u64::from_be_bytes(deserialize!(v)),
145	None => 0_u64,
146	}
147	};
148	}
149
150	#[macro_export]
151	macro_rules! deserialize_decimal {
152	($stmt:expr) => {
153	match $stmt {
154	Some(v) => Decimal::deserialize(deserialize!(v)),
155	None => Decimal::from(0),
156	}
157	};
158	}
159
160	#[macro_export]
161	macro_rules! deserialize_u8 {
162	($stmt:expr) => {
163	match $stmt {
164	Some(v) => u8::from_be_bytes(deserialize!(v)),
165	None => 0_u8,
166	}
167	};
168	}
169
170	#[macro_export]
171	macro_rules! check_nan_f64 {
172	($stmt:expr) => {
173	if $stmt.is_nan() {
174	0_f64
175	} else {
176	$stmt
177	}
178	};
179	}
180
181	#[macro_export]
182	macro_rules! check_nan_decimal {
183	($stmt:expr) => {
184	if $stmt.is_nan() {
185	dozer_types::rust_decimal::Decimal::zero()
186	} else {
187	$stmt
188	}
189	};
190	}
191
192	#[macro_export]
193	macro_rules! try_unwrap {
194	($stmt:expr) => {
195	$stmt.unwrap_or_else(\|e\| panic!("{}", e.to_string()))
196	};
197	}
198
199	#[macro_export]
200	macro_rules! to_bytes {
201	($stmt:expr) => {
202	$stmt.to_be_bytes().as_slice()
203	};
204	}

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