graphrecords_query/operations/aggregation/
any.rs

1use crate::{
2    Bare, EvaluateOperand, Explain, IndexDomain, Indexed, Mask, Multiple, Operand, OrderState,
3    QueryResult,
4    execution::EvaluationCache,
5    operands::DefiniteBareBoolOperand,
6    operations::{
7        Apply, BareStream, KeyedStream, LaneKernel, Operation, OperationContext, Prepare,
8    },
9    optimizer::{Estimate, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs, Stats},
10    registry::operation_manifest,
11    traits::Any,
12};
13use graphrecords_core::GraphRecord;
14
15#[derive(Clone, Explain, Operation, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs)]
16#[operation(scope = Lane)]
17#[explain(label = "Any")]
18pub struct AnyOperation;
19
20impl Prepare for AnyOperation {
21    type Prepared<'a> = ();
22
23    fn prepare<'a>(
24        &'a self,
25        _graphrecord: &'a GraphRecord,
26        _cache: &'a EvaluationCache<'a>,
27    ) -> QueryResult<Self::Prepared<'a>> {
28        Ok(())
29    }
30}
31
32fn any(values: impl Iterator<Item = QueryResult<bool>>) -> QueryResult<bool> {
33    for value in values {
34        if value? {
35            return Ok(true);
36        }
37    }
38
39    Ok(false)
40}
41
42impl<I: IndexDomain, O: OrderState> LaneKernel<Indexed<I, Mask>, Multiple<O>> for AnyOperation {
43    type Output = DefiniteBareBoolOperand;
44
45    fn execute<'a>(
46        _graphrecord: &'a GraphRecord,
47        values: KeyedStream<'a, I, Mask, Multiple<O>>,
48        _prepared: Self::Prepared<'a>,
49    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
50        Ok(any(values.map(|(_, value)| value)))
51    }
52
53    fn estimate(&self, _input: Estimate, _stats: &Stats) -> Estimate {
54        Estimate::singleton()
55    }
56}
57
58impl<O: OrderState> LaneKernel<Bare<Mask>, Multiple<O>> for AnyOperation {
59    type Output = DefiniteBareBoolOperand;
60
61    fn execute<'a>(
62        _graphrecord: &'a GraphRecord,
63        values: BareStream<'a, Mask, Multiple<O>>,
64        _prepared: Self::Prepared<'a>,
65    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
66        Ok(any(values))
67    }
68
69    fn estimate(&self, _input: Estimate, _stats: &Stats) -> Estimate {
70        Estimate::singleton()
71    }
72}
73
74impl<O: Apply<AnyOperation>> Any for O {
75    type ReturnOperand = O::Output;
76
77    fn any(&self) -> Self::ReturnOperand {
78        Self::ReturnOperand::new(OperationContext::new(self.clone(), AnyOperation))
79    }
80}
81
82operation_manifest! {
83    AnyOperation {
84        method: Any::any;
85        scope: lane;
86
87        kernel {
88            parameters: <
89                I: IndexDomain,
90                O: OrderState,
91            >;
92            input: (Indexed<I, Mask>, Multiple<O>);
93            output: DefiniteBareBoolOperand;
94        }
95
96        kernel {
97            parameters: <O: OrderState>;
98            input: (Bare<Mask>, Multiple<O>);
99            output: DefiniteBareBoolOperand;
100        }
101    }
102}
graphrecords_query/operations/aggregation/any.rs

graphrecords_query/operations/aggregation/
any.rs
