graphrecords_query/operations/structure/
attribute.rs

1use crate::{
2    EntityReference, Explain, Failure, IndexDomain, Indexed, Labeled, Operand, QueryResult, Scalar,
3    Unit,
4    element::{Pipeline, Preserving},
5    error::structure::{MissingAttribute, MissingTraversedAttribute},
6    execution::EvaluationCache,
7    index::EntityAttributes,
8    operations::{Apply, ElementKernel, ElementPipeline, Operation, OperationContext, Prepare},
9    optimizer::{Estimate, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs, Stats},
10    registry::operation_manifest,
11    traits::Attribute,
12};
13use graphrecords_core::{GraphRecord, graphrecord::GraphRecordAttribute};
14
15#[derive(Clone, Explain, Operation, OperationInputs, OptimizerHints, PlanIdentity, PlanInputs)]
16#[operation(scope = Element)]
17#[explain(label = "Attribute")]
18#[plan(optimizer_hints(allows_limit_pushdown, empty = if_any))]
19pub struct AttributeOperation {
20    #[explain(label)]
21    attribute: GraphRecordAttribute,
22}
23
24impl Prepare for AttributeOperation {
25    type Prepared<'a> = &'a GraphRecordAttribute;
26
27    fn prepare<'a>(
28        &'a self,
29        _graphrecord: &'a GraphRecord,
30        _cache: &'a EvaluationCache<'a>,
31    ) -> QueryResult<Self::Prepared<'a>> {
32        Ok(&self.attribute)
33    }
34}
35
36impl<I: EntityAttributes> ElementKernel<Indexed<I, Unit>> for AttributeOperation {
37    type Emission = Preserving;
38    type OutShape = Indexed<I, Scalar>;
39
40    fn pipeline<'a>(
41        graphrecord: &'a GraphRecord,
42        prepared: Self::Prepared<'a>,
43    ) -> QueryResult<ElementPipeline<'a, Indexed<I, Unit>, Self>> {
44        Ok(Pipeline::keyed(move |index, membership| {
45            membership?;
46            let attributes = I::attributes(graphrecord, &index).expect("Entity must exist");
47
48            if let Some(value) = attributes.get(prepared) {
49                return Ok(value.clone());
50            }
51
52            let failure = Failure::new_at::<I, _>(
53                Self::LABEL,
54                MissingAttribute::new(prepared.clone()),
55                &index,
56            );
57
58            Err(failure)
59        }))
60    }
61
62    fn estimate(&self, input: Estimate, stats: &Stats) -> Estimate {
63        let mut distinct = I::attribute_cardinality(stats, &self.attribute);
64        if let Some(elements) = input.elements {
65            distinct = distinct.min(elements);
66        }
67
68        Estimate {
69            distinct: Some(distinct),
70            selectivity: None,
71            ..input
72        }
73    }
74}
75
76impl<E: EntityAttributes, I: IndexDomain> ElementKernel<Indexed<I, EntityReference<E>>>
77    for AttributeOperation
78{
79    type Emission = Preserving;
80    type OutShape = Indexed<I, Scalar>;
81
82    fn pipeline<'a>(
83        graphrecord: &'a GraphRecord,
84        prepared: Self::Prepared<'a>,
85    ) -> QueryResult<ElementPipeline<'a, Indexed<I, EntityReference<E>>, Self>> {
86        Ok(Pipeline::keyed(move |key, reference: QueryResult<_>| {
87            reference.and_then(|entity| {
88                let attributes = E::attributes(graphrecord, &entity).expect("Entity must exist");
89
90                if let Some(value) = attributes.get(prepared) {
91                    return Ok(value.clone());
92                }
93
94                Err(Failure::new_at::<I, _>(
95                    Self::LABEL,
96                    MissingTraversedAttribute::new(prepared.clone(), E::to_owned(&entity)),
97                    &key,
98                ))
99            })
100        }))
101    }
102
103    fn estimate(&self, input: Estimate, stats: &Stats) -> Estimate {
104        let mut distinct = E::attribute_cardinality(stats, &self.attribute);
105        if let Some(input_distinct) = input.distinct {
106            distinct = distinct.min(input_distinct);
107        }
108        if let Some(elements) = input.elements {
109            distinct = distinct.min(elements);
110        }
111
112        Estimate {
113            distinct: Some(distinct),
114            selectivity: None,
115            ..input
116        }
117    }
118}
119
120impl<O: Apply<AttributeOperation>> Attribute for O {
121    type ReturnOperand = O::Output;
122
123    fn attribute(&self, attribute: GraphRecordAttribute) -> Self::ReturnOperand {
124        Self::ReturnOperand::new(OperationContext::new(
125            self.clone(),
126            AttributeOperation { attribute },
127        ))
128    }
129}
130
131operation_manifest! {
132    AttributeOperation {
133        method: Attribute::attribute;
134        scope: element;
135
136        kernel {
137            parameters: <I: EntityAttributes>;
138            field: attribute: GraphRecordAttribute;
139            input: Indexed<I, Unit>;
140            output: Indexed<I, Scalar>;
141            emission: Preserving;
142        }
143
144        kernel {
145            parameters: <E: EntityAttributes, I: IndexDomain>;
146            field: attribute: GraphRecordAttribute;
147            input: Indexed<I, EntityReference<E>>;
148            output: Indexed<I, Scalar>;
149            emission: Preserving;
150        }
151    }
152}
graphrecords_query/operations/structure/attribute.rs

graphrecords_query/operations/structure/
attribute.rs
