graphrecords_query/operations/aggregation/
standard_deviation.rs

1use crate::{
2    Bare, BareValueDomain, EvaluateOperand, Explain, Failure, IndexDomain, Indexed, Labeled,
3    Multiple, Operand, OrderState, QueryResult,
4    capabilities::ValueScalar,
5    error::aggregation::InvalidStandardDeviationValue,
6    execution::EvaluationCache,
7    operands::BareValueOperand,
8    operations::{
9        Apply, BareStream, KeyedStream, LaneKernel, Operation, OperationContext, Prepare,
10    },
11    optimizer::{Estimate, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs, Stats},
12    registry::operation_manifest,
13    traits::StandardDeviation,
14};
15use graphrecords_core::{GraphRecord, graphrecord::GraphRecordValue};
16
17#[derive(Clone, Explain, Operation, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs)]
18#[operation(scope = Lane)]
19#[explain(label = "Std")]
20#[plan(optimizer_hints(empty = if_any))]
21pub struct StandardDeviationOperation;
22
23impl Prepare for StandardDeviationOperation {
24    type Prepared<'a> = ();
25
26    fn prepare<'a>(
27        &'a self,
28        _graphrecord: &'a GraphRecord,
29        _cache: &'a EvaluationCache<'a>,
30    ) -> QueryResult<Self::Prepared<'a>> {
31        Ok(())
32    }
33}
34
35fn update_state(
36    (count, mean, squared_deviation): (usize, f64, f64),
37    value: f64,
38) -> (usize, f64, f64) {
39    let count = count + 1;
40    let difference = value - mean;
41    let mean = mean + difference / count as f64;
42    let updated_difference = value - mean;
43    let squared_deviation = difference.mul_add(updated_difference, squared_deviation);
44
45    (count, mean, squared_deviation)
46}
47
48impl<I, V, O> LaneKernel<Indexed<I, V>, Multiple<O>> for StandardDeviationOperation
49where
50    I: IndexDomain,
51    V: ValueScalar + BareValueDomain,
52    O: OrderState,
53{
54    type Output = BareValueOperand;
55
56    fn execute<'a>(
57        _graphrecord: &'a GraphRecord,
58        mut values: KeyedStream<'a, I, V, Multiple<O>>,
59        _prepared: Self::Prepared<'a>,
60    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
61        let standard_deviation = values
62            .try_fold((0_usize, 0.0, 0.0), |state, (index, value)| {
63                let value = V::into_scalar(Self::LABEL, value?)
64                    .map_err(|failure| failure.at::<I>(&index))?;
65                let value = match value {
66                    GraphRecordValue::Int(value) => value as f64,
67                    GraphRecordValue::Float(value) => value,
68                    value => {
69                        return Err(Failure::new_at::<I, _>(
70                            Self::LABEL,
71                            InvalidStandardDeviationValue::new(value),
72                            &index,
73                        ));
74                    }
75                };
76
77                Ok(update_state(state, value))
78            })
79            .map(|(count, _, squared_deviation)| {
80                (count > 1).then(|| {
81                    GraphRecordValue::Float((squared_deviation / (count - 1) as f64).sqrt())
82                })
83            });
84
85        Ok(standard_deviation.transpose())
86    }
87
88    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
89        input.zero_or_one()
90    }
91}
92
93impl<V, O> LaneKernel<Bare<V>, Multiple<O>> for StandardDeviationOperation
94where
95    V: ValueScalar + BareValueDomain,
96    O: OrderState,
97{
98    type Output = BareValueOperand;
99
100    fn execute<'a>(
101        _graphrecord: &'a GraphRecord,
102        mut values: BareStream<'a, V, Multiple<O>>,
103        _prepared: Self::Prepared<'a>,
104    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
105        let standard_deviation = values
106            .try_fold((0_usize, 0.0, 0.0), |state, value| {
107                let value = V::into_scalar(Self::LABEL, value?)?;
108                let value = match value {
109                    GraphRecordValue::Int(value) => value as f64,
110                    GraphRecordValue::Float(value) => value,
111                    value => {
112                        return Err(Failure::new(
113                            Self::LABEL,
114                            InvalidStandardDeviationValue::new(value),
115                        ));
116                    }
117                };
118
119                Ok(update_state(state, value))
120            })
121            .map(|(count, _, squared_deviation)| {
122                (count > 1).then(|| {
123                    GraphRecordValue::Float((squared_deviation / (count - 1) as f64).sqrt())
124                })
125            });
126
127        Ok(standard_deviation.transpose())
128    }
129
130    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
131        input.zero_or_one()
132    }
133}
134
135impl<O: Apply<StandardDeviationOperation>> StandardDeviation for O {
136    type ReturnOperand = O::Output;
137
138    fn std(&self) -> Self::ReturnOperand {
139        Self::ReturnOperand::new(OperationContext::new(
140            self.clone(),
141            StandardDeviationOperation,
142        ))
143    }
144}
145
146operation_manifest! {
147    StandardDeviationOperation {
148        method: StandardDeviation::std;
149        scope: lane;
150
151        kernel {
152            parameters: <
153                I: IndexDomain,
154                V: ValueScalar + BareValueDomain,
155                O: OrderState,
156            >;
157            input: (Indexed<I, V>, Multiple<O>);
158            output: BareValueOperand;
159        }
160
161        kernel {
162            parameters: <
163                V: ValueScalar + BareValueDomain,
164                O: OrderState,
165            >;
166            input: (Bare<V>, Multiple<O>);
167            output: BareValueOperand;
168        }
169    }
170}
graphrecords_query/operations/aggregation/standard_deviation.rs

graphrecords_query/operations/aggregation/
standard_deviation.rs
