graphrecords_query/operations/aggregation/
maximum.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::Maximum,
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 = "Max")]
24#[plan(optimizer_hints(empty = if_any))]
25pub struct MaximumOperation;
26
27impl Prepare for MaximumOperation {
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 MaximumOperation
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 maximum = values.try_fold(None, |maximum, (index, value)| {
54            let value = value?;
55
56            let Some((maximum_index, maximum_value)) = maximum else {
57                return Ok(Some((index, value)));
58            };
59
60            match V::ordering(&value, &maximum_value) {
61                Some(Ordering::Greater) => Ok(Some((index, value))),
62                Some(Ordering::Less | Ordering::Equal) => Ok(Some((maximum_index, maximum_value))),
63                None => Err(Failure::new_at::<I, _>(
64                    Self::LABEL,
65                    IncomparableValuesAt::new(
66                        V::into_owned(value),
67                        V::into_owned(maximum_value),
68                        I::to_owned(&index),
69                        I::to_owned(&maximum_index),
70                    ),
71                    &index,
72                )),
73            }
74        });
75
76        Ok(match maximum {
77            Ok(maximum) => maximum.map(|(_, value)| Ok(value)),
78            Err(failure) => Some(Err(failure)),
79        })
80    }
81
82    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
83        input.zero_or_one()
84    }
85}
86
87impl<V, O> LaneKernel<Bare<V>, Multiple<O>> for MaximumOperation
88where
89    V: ValueOrdering + BareValueDomain,
90    O: OrderState,
91    V::Owned: Debug + Display + Send + Sync,
92{
93    type Output = OperandHandle<Bare<V>, Single>;
94
95    fn execute<'a>(
96        _graphrecord: &'a GraphRecord,
97        mut values: BareStream<'a, V, Multiple<O>>,
98        _prepared: Self::Prepared<'a>,
99    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
100        let maximum = values.try_fold(None, |maximum, value| {
101            let value = value?;
102
103            let Some(maximum) = maximum else {
104                return Ok(Some(value));
105            };
106
107            match V::ordering(&value, &maximum) {
108                Some(Ordering::Greater) => Ok(Some(value)),
109                Some(Ordering::Less | Ordering::Equal) => Ok(Some(maximum)),
110                None => Err(Failure::new(
111                    Self::LABEL,
112                    IncomparableValues::new(V::into_owned(value), V::into_owned(maximum)),
113                )),
114            }
115        });
116
117        Ok(maximum.transpose())
118    }
119
120    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
121        input.zero_or_one()
122    }
123}
124
125impl<O: Apply<MaximumOperation>> Maximum for O {
126    type ReturnOperand = O::Output;
127
128    fn max(&self) -> Self::ReturnOperand {
129        Self::ReturnOperand::new(OperationContext::new(self.clone(), MaximumOperation))
130    }
131}
132
133operation_manifest! {
134    MaximumOperation {
135        method: Maximum::max;
136        scope: lane;
137
138        kernel {
139            parameters: <
140                I: IndexDomain,
141                V: ValueOrdering + BareValueDomain,
142                O: OrderState,
143            >;
144            input: (Indexed<I, V>, Multiple<O>);
145            output: OperandHandle<Bare<V>, Single>;
146            where V::Owned: Debug + Display + Send + Sync;
147        }
148
149        kernel {
150            parameters: <
151                V: ValueOrdering + BareValueDomain,
152                O: OrderState,
153            >;
154            input: (Bare<V>, Multiple<O>);
155            output: OperandHandle<Bare<V>, Single>;
156            where V::Owned: Debug + Display + Send + Sync;
157        }
158    }
159}
graphrecords_query/operations/aggregation/maximum.rs

graphrecords_query/operations/aggregation/
maximum.rs
