graphrecords_query/operations/aggregation/
minimum.rs

1use crate::{
2    Bare, BareValueDomain, EvaluateOperand, Explain, Failure, IndexDomain, Indexed, Labeled,
3    Multiple, Operand, OrderState, QueryResult, Single,
4    capabilities::ValueOrdering,
5    error::comparison::{IncomparableValues, IncomparableValuesAt},
6    execution::EvaluationCache,
7    operands::OperandHandle,
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::Minimum,
14};
15use graphrecords_core::GraphRecord;
16use std::{
17    cmp::Ordering,
18    fmt::{Debug, Display},
19};
20
21#[derive(Clone, Explain, Operation, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs)]
22#[operation(scope = Lane)]
23#[explain(label = "Min")]
24#[plan(optimizer_hints(empty = if_any))]
25pub struct MinimumOperation;
26
27impl Prepare for MinimumOperation {
28    type Prepared<'a> = ();
29
30    fn prepare<'a>(
31        &'a self,
32        _graphrecord: &'a GraphRecord,
33        _cache: &'a EvaluationCache<'a>,
34    ) -> QueryResult<Self::Prepared<'a>> {
35        Ok(())
36    }
37}
38
39impl<I, V, O> LaneKernel<Indexed<I, V>, Multiple<O>> for MinimumOperation
40where
41    I: IndexDomain,
42    V: ValueOrdering + BareValueDomain,
43    O: OrderState,
44    V::Owned: Debug + Display + Send + Sync,
45{
46    type Output = OperandHandle<Bare<V>, Single>;
47
48    fn execute<'a>(
49        _graphrecord: &'a GraphRecord,
50        mut values: KeyedStream<'a, I, V, Multiple<O>>,
51        _prepared: Self::Prepared<'a>,
52    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
53        let minimum = values.try_fold(None, |minimum, (index, value)| {
54            let value = value?;
55
56            let Some((minimum_index, minimum_value)) = minimum else {
57                return Ok(Some((index, value)));
58            };
59
60            match V::ordering(&value, &minimum_value) {
61                Some(Ordering::Less) => Ok(Some((index, value))),
62                Some(Ordering::Equal | Ordering::Greater) => {
63                    Ok(Some((minimum_index, minimum_value)))
64                }
65                None => Err(Failure::new_at::<I, _>(
66                    Self::LABEL,
67                    IncomparableValuesAt::new(
68                        V::into_owned(value),
69                        V::into_owned(minimum_value),
70                        I::to_owned(&index),
71                        I::to_owned(&minimum_index),
72                    ),
73                    &index,
74                )),
75            }
76        });
77
78        Ok(match minimum {
79            Ok(minimum) => minimum.map(|(_, value)| Ok(value)),
80            Err(failure) => Some(Err(failure)),
81        })
82    }
83
84    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
85        input.zero_or_one()
86    }
87}
88
89impl<V, O> LaneKernel<Bare<V>, Multiple<O>> for MinimumOperation
90where
91    V: ValueOrdering + BareValueDomain,
92    O: OrderState,
93    V::Owned: Debug + Display + Send + Sync,
94{
95    type Output = OperandHandle<Bare<V>, Single>;
96
97    fn execute<'a>(
98        _graphrecord: &'a GraphRecord,
99        mut values: BareStream<'a, V, Multiple<O>>,
100        _prepared: Self::Prepared<'a>,
101    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
102        let minimum = values.try_fold(None, |minimum, value| {
103            let value = value?;
104
105            let Some(minimum) = minimum else {
106                return Ok(Some(value));
107            };
108
109            match V::ordering(&value, &minimum) {
110                Some(Ordering::Less) => Ok(Some(value)),
111                Some(Ordering::Equal | Ordering::Greater) => Ok(Some(minimum)),
112                None => Err(Failure::new(
113                    Self::LABEL,
114                    IncomparableValues::new(V::into_owned(value), V::into_owned(minimum)),
115                )),
116            }
117        });
118
119        Ok(minimum.transpose())
120    }
121
122    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
123        input.zero_or_one()
124    }
125}
126
127impl<O: Apply<MinimumOperation>> Minimum for O {
128    type ReturnOperand = O::Output;
129
130    fn min(&self) -> Self::ReturnOperand {
131        Self::ReturnOperand::new(OperationContext::new(self.clone(), MinimumOperation))
132    }
133}
134
135operation_manifest! {
136    MinimumOperation {
137        method: Minimum::min;
138        scope: lane;
139
140        kernel {
141            parameters: <
142                I: IndexDomain,
143                V: ValueOrdering + BareValueDomain,
144                O: OrderState,
145            >;
146            input: (Indexed<I, V>, Multiple<O>);
147            output: OperandHandle<Bare<V>, Single>;
148            where V::Owned: Debug + Display + Send + Sync;
149        }
150
151        kernel {
152            parameters: <
153                V: ValueOrdering + BareValueDomain,
154                O: OrderState,
155            >;
156            input: (Bare<V>, Multiple<O>);
157            output: OperandHandle<Bare<V>, Single>;
158            where V::Owned: Debug + Display + Send + Sync;
159        }
160    }
161}
graphrecords_query/operations/aggregation/minimum.rs

graphrecords_query/operations/aggregation/
minimum.rs
