graphrecords_query/operations/uniqueness/
drop_duplicates.rs

1use crate::{
2    EvaluateOperand, Explain, IndexDomain, Indexed, Multiple, Operand, Ordered, QueryResult,
3    capabilities::ValueEquivalence,
4    execution::EvaluationCache,
5    operands::OperandHandle,
6    operations::{Apply, KeyedStream, LaneKernel, Operation, OperationContext, Prepare},
7    optimizer::{Estimate, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs, Stats},
8    registry::operation_manifest,
9    traits::DropDuplicates,
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 = "DropDuplicates")]
17#[plan(optimizer_hints(empty = if_any))]
18pub struct DropDuplicatesOperation;
19
20impl Prepare for DropDuplicatesOperation {
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<I: IndexDomain, V: ValueEquivalence> LaneKernel<Indexed<I, V>, Multiple<Ordered>>
33    for DropDuplicatesOperation
34{
35    type Output = OperandHandle<Indexed<I, V>, Multiple<Ordered>>;
36
37    fn execute<'a>(
38        _graphrecord: &'a GraphRecord,
39        values: KeyedStream<'a, I, V, Multiple<Ordered>>,
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(
45            move |(index, outcome)| match outcome {
46                Ok(value) if seen.insert(V::equivalence_key(&value)) => Some((index, Ok(value))),
47                Ok(_) => None,
48                Err(failure) => Some((index, Err(failure))),
49            },
50        )))
51    }
52
53    fn estimate(&self, input: Estimate, _stats: &Stats) -> Estimate {
54        Estimate {
55            elements: None,
56            distinct: input.distinct,
57            selectivity: None,
58            per_group: None,
59        }
60    }
61}
62
63impl<O: Apply<DropDuplicatesOperation>> DropDuplicates for O {
64    type ReturnOperand = O::Output;
65
66    fn drop_duplicates(&self) -> Self::ReturnOperand {
67        Self::ReturnOperand::new(OperationContext::new(self.clone(), DropDuplicatesOperation))
68    }
69}
70
71operation_manifest! {
72    DropDuplicatesOperation {
73        method: DropDuplicates::drop_duplicates;
74        scope: lane;
75
76        kernel {
77            parameters: <
78                I: IndexDomain,
79                V: ValueEquivalence,
80            >;
81            input: (Indexed<I, V>, Multiple<Ordered>);
82            output: OperandHandle<Indexed<I, V>, Multiple<Ordered>>;
83        }
84    }
85}
graphrecords_query/operations/uniqueness/drop_duplicates.rs

graphrecords_query/operations/uniqueness/
drop_duplicates.rs
