graphrecords_query/operations/uniqueness/
unique.rs

1use crate::{
2    Bare, BareValueDomain, EvaluateOperand, Explain, Multiple, Operand, OrderState, QueryResult,
3    capabilities::ValueEquivalence,
4    execution::EvaluationCache,
5    operands::OperandHandle,
6    operations::{Apply, BareStream, LaneKernel, Operation, OperationContext, Prepare},
7    optimizer::{Estimate, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs, Stats},
8    registry::operation_manifest,
9    traits::Unique,
10};
11use graphrecords_core::GraphRecord;
12use graphrecords_utils::aliases::GrHashSet;
13
14#[derive(Clone, Explain, Operation, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs)]
15#[operation(scope = Lane)]
16#[explain(label = "Unique")]
17#[plan(optimizer_hints(empty = if_any))]
18pub struct UniqueOperation;
19
20impl Prepare for UniqueOperation {
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
32impl<V: ValueEquivalence + BareValueDomain, O: OrderState> LaneKernel<Bare<V>, Multiple<O>>
33    for UniqueOperation
34{
35    type Output = OperandHandle<Bare<V>, Multiple<O>>;
36
37    fn execute<'a>(
38        _graphrecord: &'a GraphRecord,
39        values: BareStream<'a, V, Multiple<O>>,
40        _prepared: Self::Prepared<'a>,
41    ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>> {
42        let mut seen = GrHashSet::default();
43
44        Ok(Box::new(values.filter_map(move |outcome| match outcome {
45            Ok(value) if seen.insert(V::equivalence_key(&value)) => Some(Ok(value)),
46            Ok(_) => None,
47            Err(failure) => Some(Err(failure)),
48        })))
49    }
50
51    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
52        Estimate {
53            elements: None,
54            distinct: input.distinct,
55            selectivity: None,
56            per_group: None,
57        }
58    }
59}
60
61impl<O: Apply<UniqueOperation>> Unique for O {
62    type ReturnOperand = O::Output;
63
64    fn unique(&self) -> Self::ReturnOperand {
65        Self::ReturnOperand::new(OperationContext::new(self.clone(), UniqueOperation))
66    }
67}
68
69operation_manifest! {
70    UniqueOperation {
71        method: Unique::unique;
72        scope: lane;
73
74        kernel {
75            parameters: <
76                V: ValueEquivalence + BareValueDomain,
77                O: OrderState,
78            >;
79            input: (Bare<V>, Multiple<O>);
80            output: OperandHandle<Bare<V>, Multiple<O>>;
81        }
82    }
83}
graphrecords_query/operations/uniqueness/unique.rs

graphrecords_query/operations/uniqueness/
unique.rs
