graphrecords_query/operations/aggregation/
mode.rs

1use crate::{
2    Bare, BareValueDomain, EvaluateOperand, Explain, IndexDomain, Indexed, Multiple, Operand,
3    OrderState, QueryResult,
4    capabilities::ValueMode,
5    execution::EvaluationCache,
6    operands::OperandHandle,
7    operations::{
8        Apply, BareStream, KeyedStream, LaneKernel, Operation, OperationContext, Prepare,
9    },
10    optimizer::{OperationInputs, OptimizerHints, PlanIdentity, PlanInputs},
11    registry::operation_manifest,
12    traits::Mode,
13};
14use graphrecords_core::GraphRecord;
15use graphrecords_utils::aliases::GrHashMap;
16
17#[derive(Clone, Explain, Operation, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs)]
18#[operation(scope = Lane)]
19#[explain(label = "Mode")]
20#[plan(optimizer_hints(empty = if_any))]
21pub struct ModeOperation;
22
23impl Prepare for ModeOperation {
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 modal_values<'a, V: ValueMode>(
36    values: impl Iterator<Item = QueryResult<V::Value<'a>>>,
37) -> Vec<QueryResult<V::Value<'a>>> {
38    let mut counts: GrHashMap<_, usize> = GrHashMap::default();
39    let mut encountered = Vec::new();
40
41    for outcome in values {
42        let value = match outcome {
43            Ok(value) => value,
44            Err(failure) => return vec![Err(failure)],
45        };
46        let key = V::equivalence_key(&value);
47
48        if let Some(count) = counts.get_mut(&key) {
49            *count += 1;
50        } else {
51            counts.insert(key, 1);
52            encountered.push(value);
53        }
54    }
55
56    let Some(maximum_count) = counts.values().copied().max() else {
57        return Vec::new();
58    };
59
60    encountered
61        .into_iter()
62        .filter_map(|value| {
63            (counts.get(&V::equivalence_key(&value)) == Some(&maximum_count)).then_some(Ok(value))
64        })
65        .collect()
66}
67
68impl<I: IndexDomain, V: ValueMode + BareValueDomain, O: OrderState>
69    LaneKernel<Indexed<I, V>, Multiple<O>> for ModeOperation
70{
71    type Output = OperandHandle<Bare<V>, Multiple<O>>;
72
73    fn execute<'a>(
74        _graphrecord: &'a GraphRecord,
75        values: KeyedStream<'a, I, V, Multiple<O>>,
76        _prepared: Self::Prepared<'a>,
77    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
78        Ok(Box::new(
79            modal_values::<V>(values.map(|(_, value)| value)).into_iter(),
80        ))
81    }
82}
83
84impl<V: ValueMode + BareValueDomain, O: OrderState> LaneKernel<Bare<V>, Multiple<O>>
85    for ModeOperation
86{
87    type Output = OperandHandle<Bare<V>, Multiple<O>>;
88
89    fn execute<'a>(
90        _graphrecord: &'a GraphRecord,
91        values: BareStream<'a, V, Multiple<O>>,
92        _prepared: Self::Prepared<'a>,
93    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
94        Ok(Box::new(modal_values::<V>(values).into_iter()))
95    }
96}
97
98impl<O: Apply<ModeOperation>> Mode for O {
99    type ReturnOperand = O::Output;
100
101    fn mode(&self) -> Self::ReturnOperand {
102        Self::ReturnOperand::new(OperationContext::new(self.clone(), ModeOperation))
103    }
104}
105
106operation_manifest! {
107    ModeOperation {
108        method: Mode::mode;
109        scope: lane;
110
111        kernel {
112            parameters: <
113                I: IndexDomain,
114                V: ValueMode + BareValueDomain,
115                O: OrderState,
116            >;
117            input: (Indexed<I, V>, Multiple<O>);
118            output: OperandHandle<Bare<V>, Multiple<O>>;
119        }
120
121        kernel {
122            parameters: <
123                V: ValueMode + BareValueDomain,
124                O: OrderState,
125            >;
126            input: (Bare<V>, Multiple<O>);
127            output: OperandHandle<Bare<V>, Multiple<O>>;
128        }
129    }
130}
graphrecords_query/operations/aggregation/mode.rs

graphrecords_query/operations/aggregation/
mode.rs
