graphrecords_query/element/
transition.rs

1use crate::{
2    BareValueDomain, ExpandedChild, ExpandedIndex, ExpandedIndexReference, Failure, IndexDomain,
3    QueryResult, ValueDomain,
4    element::{
5        Arity, Bare, ElementEmission, ElementShape, Expanding, Indexed, OrderState, Ordered,
6        Retention, Unordered,
7    },
8    error::index::DuplicateExpandedChildIndex,
9};
10use graphrecords_utils::aliases::GrHashSet;
11use std::marker::PhantomData;
12
13type ExpandedElement<'a, P, C, V> = (
14    ExpandedIndexReference<'a, P, C>,
15    QueryResult<<V as ValueDomain>::Value<'a>>,
16);
17
18pub struct Pipeline<'a, X: 'a, Y: 'a, E: ElementEmission> {
19    run: Box<dyn Fn(X) -> Y + 'a>,
20    emission: PhantomData<fn() -> E>,
21}
22
23impl<'a, X: 'a, Y: 'a, E: ElementEmission> Pipeline<'a, X, Y, E> {
24    #[must_use]
25    pub fn new(run: impl Fn(X) -> Y + 'a) -> Self {
26        Self {
27            run: Box::new(run),
28            emission: PhantomData,
29        }
30    }
31
32    pub(crate) fn run(&self, input: X) -> Y {
33        (self.run)(input)
34    }
35}
36
37impl<'a, A: 'a, B: 'a, Y: 'a, E: ElementEmission> Pipeline<'a, (A, B), Y, E> {
38    #[must_use]
39    pub fn keyed(run: impl Fn(A, B) -> Y + 'a) -> Self {
40        Self::new(move |(first, second)| run(first, second))
41    }
42
43    #[must_use]
44    pub fn unkeyed(run: impl Fn(B) -> Y + 'a) -> Self {
45        Self::new(move |(_, second)| run(second))
46    }
47}
48
49pub type IndexedValuePipeline<'a, I, V, W, E> = Pipeline<
50    'a,
51    (
52        <I as IndexDomain>::Index<'a>,
53        QueryResult<<V as ValueDomain>::Value<'a>>,
54    ),
55    <E as ElementEmission>::Step<QueryResult<<W as ValueDomain>::Value<'a>>>,
56    E,
57>;
58
59pub type BarePipeline<'a, V, W, E> = Pipeline<
60    'a,
61    QueryResult<<V as ValueDomain>::Value<'a>>,
62    <E as ElementEmission>::Step<QueryResult<<W as ValueDomain>::Value<'a>>>,
63    E,
64>;
65
66pub type IndexedExpansionPipeline<'a, P, C, V, W, O> = Pipeline<
67    'a,
68    (<P as IndexDomain>::Index<'a>, <V as ValueDomain>::Value<'a>),
69    QueryResult<Vec<ExpandedChild<'a, C, W>>>,
70    Expanding<O>,
71>;
72
73pub trait ElementTransition<T: ElementShape, E: ElementEmission>: ElementShape {
74    type Pipeline<'a>: 'a
75    where
76        Self: 'a,
77        T: 'a;
78
79    fn apply<'a, C: Arity>(
80        values: C::Container<'a, Self::Element<'a>>,
81        pipeline: Self::Pipeline<'a>,
82    ) -> <E::OutArity<C> as Arity>::Container<'a, T::Element<'a>>;
83}
84
85impl<I: IndexDomain, V: ValueDomain, W: ValueDomain, E: Retention>
86    ElementTransition<Indexed<I, W>, E> for Indexed<I, V>
87{
88    type Pipeline<'a> = IndexedValuePipeline<'a, I, V, W, E>;
89
90    fn apply<'a, C: Arity>(
91        values: C::Container<'a, Self::Element<'a>>,
92        pipeline: Self::Pipeline<'a>,
93    ) -> <E::OutArity<C> as Arity>::Container<'a, <Indexed<I, W> as ElementShape>::Element<'a>>
94    {
95        E::apply(values, move |element| {
96            let (index, value): (I::Index<'a>, _) = element;
97            let step = pipeline.run((index.clone(), value));
98
99            <E as Retention>::map_step(step, |value| (index, value))
100        })
101    }
102}
103
104impl<I: IndexDomain, V: ValueDomain, W: BareValueDomain, E: ElementEmission>
105    ElementTransition<Bare<W>, E> for Indexed<I, V>
106{
107    type Pipeline<'a> = IndexedValuePipeline<'a, I, V, W, E>;
108
109    fn apply<'a, C: Arity>(
110        values: C::Container<'a, Self::Element<'a>>,
111        pipeline: Self::Pipeline<'a>,
112    ) -> <E::OutArity<C> as Arity>::Container<'a, <Bare<W> as ElementShape>::Element<'a>> {
113        E::apply(values, move |value| pipeline.run(value))
114    }
115}
116
117impl<V: BareValueDomain, W: BareValueDomain, E: ElementEmission> ElementTransition<Bare<W>, E>
118    for Bare<V>
119{
120    type Pipeline<'a> = BarePipeline<'a, V, W, E>;
121
122    fn apply<'a, C: Arity>(
123        values: C::Container<'a, Self::Element<'a>>,
124        pipeline: Self::Pipeline<'a>,
125    ) -> <E::OutArity<C> as Arity>::Container<'a, <Bare<W> as ElementShape>::Element<'a>> {
126        E::apply(values, move |value| pipeline.run(value))
127    }
128}
129
130fn expand_indexed_source<'a, P, C, V, W, O>(
131    parent: P::Index<'a>,
132    source: QueryResult<V::Value<'a>>,
133    pipeline: &IndexedExpansionPipeline<'a, P, C, V, W, O>,
134) -> Vec<ExpandedElement<'a, P, C, W>>
135where
136    P: IndexDomain,
137    C: IndexDomain,
138    V: ValueDomain,
139    W: ValueDomain,
140    O: OrderState,
141    Expanding<O>: ElementEmission,
142{
143    let source_value = match source {
144        Ok(value) => value,
145        Err(failure) => {
146            return vec![(ExpandedIndexReference::source(parent), Err(failure))];
147        }
148    };
149
150    let children = match pipeline.run((parent.clone(), source_value)) {
151        Ok(children) => children,
152        Err(failure) => {
153            return vec![(ExpandedIndexReference::source(parent), Err(failure))];
154        }
155    };
156
157    let mut seen_children = GrHashSet::default();
158    let mut fragment = Vec::with_capacity(children.len());
159
160    for child in children {
161        let (child_index, outcome) = child.into_parts();
162
163        if !seen_children.insert(C::to_owned(&child_index)) {
164            let source_address = ExpandedIndexReference::source(parent.clone());
165            let failure = Failure::new_at::<ExpandedIndex<_, _>, _>(
166                "indexed expansion",
167                DuplicateExpandedChildIndex::<C>::new(C::to_owned(&child_index)),
168                &source_address,
169            );
170
171            return vec![(source_address, Err(failure))];
172        }
173
174        fragment.push((
175            ExpandedIndexReference::child(parent.clone(), child_index),
176            outcome,
177        ));
178    }
179
180    fragment
181}
182
183impl<P: IndexDomain, C: IndexDomain, V: ValueDomain, W: ValueDomain>
184    ElementTransition<Indexed<ExpandedIndex<P, C>, W>, Expanding<Ordered>> for Indexed<P, V>
185{
186    type Pipeline<'a>
187        = IndexedExpansionPipeline<'a, P, C, V, W, Ordered>
188    where
189        Self: 'a,
190        Indexed<ExpandedIndex<P, C>, W>: 'a;
191
192    fn apply<'a, A: Arity>(
193        values: A::Container<'a, Self::Element<'a>>,
194        pipeline: Self::Pipeline<'a>,
195    ) -> <<Expanding<Ordered> as ElementEmission>::OutArity<A> as Arity>::Container<
196        'a,
197        <Indexed<ExpandedIndex<P, C>, W> as ElementShape>::Element<'a>,
198    > {
199        Expanding::<Ordered>::apply::<A, _, _>(values, move |(parent, source)| {
200            expand_indexed_source::<_, _, V, _, _>(parent, source, &pipeline)
201        })
202    }
203}
204
205impl<P: IndexDomain, C: IndexDomain, V: ValueDomain, W: ValueDomain>
206    ElementTransition<Indexed<ExpandedIndex<P, C>, W>, Expanding<Unordered>> for Indexed<P, V>
207{
208    type Pipeline<'a>
209        = IndexedExpansionPipeline<'a, P, C, V, W, Unordered>
210    where
211        Self: 'a,
212        Indexed<ExpandedIndex<P, C>, W>: 'a;
213
214    fn apply<'a, A: Arity>(
215        values: A::Container<'a, Self::Element<'a>>,
216        pipeline: Self::Pipeline<'a>,
217    ) -> <<Expanding<Unordered> as ElementEmission>::OutArity<A> as Arity>::Container<
218        'a,
219        <Indexed<ExpandedIndex<P, C>, W> as ElementShape>::Element<'a>,
220    > {
221        Expanding::<Unordered>::apply::<A, _, _>(values, move |(parent, source)| {
222            expand_indexed_source::<_, _, V, _, _>(parent, source, &pipeline)
223        })
224    }
225}
graphrecords_query/element/transition.rs

graphrecords_query/element/
transition.rs
