1use crate::{
2 Arity, Bare, ElementShape, EvaluateOperand, IndexDomain, Indexed, Operand, QueryResult,
3 element::{ElementEmission, ElementTransition},
4 index::GroupKey,
5 operands::{GroupOperand, OperandHandle, Partition},
6 operations::{Apply, Element, Group, Lane, Operation, OperationScope},
7 optimizer::{Estimate, Stats},
8};
9use graphrecords_core::GraphRecord;
10
11pub type KeyedStream<'a, I, V, C> =
12 <OperandHandle<Indexed<I, V>, C> as EvaluateOperand>::ReturnValue<'a>;
13
14pub type BareStream<'a, V, C> = <OperandHandle<Bare<V>, C> as EvaluateOperand>::ReturnValue<'a>;
15
16pub type ElementPipeline<'a, S, P> = <S as ElementTransition<
17 <P as ElementKernel<S>>::OutShape,
18 <P as ElementKernel<S>>::Emission,
19>>::Pipeline<'a>;
20
21pub trait ElementKernel<S: ElementShape + ElementTransition<Self::OutShape, Self::Emission>>:
22 Operation<Scope = Element>
23{
24 type OutShape: ElementShape;
25 type Emission: ElementEmission;
26
27 fn pipeline<'a>(
28 graphrecord: &'a GraphRecord,
29 prepared: Self::Prepared<'a>,
30 ) -> QueryResult<ElementPipeline<'a, S, Self>>;
31
32 #[allow(unused_variables)]
33 fn estimate(&self, input: Estimate, stats: &Stats) -> Estimate {
34 Self::Emission::default_estimate(input)
35 }
36}
37
38pub trait LaneKernel<S: ElementShape, C: Arity>: Operation<Scope = Lane> {
39 type Output: Operand;
40
41 fn execute<'a>(
42 graphrecord: &'a GraphRecord,
43 values: <OperandHandle<S, C> as EvaluateOperand>::ReturnValue<'a>,
44 prepared: Self::Prepared<'a>,
45 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>;
46
47 #[allow(unused_variables)]
48 fn estimate(&self, input: Estimate, stats: &Stats) -> Estimate {
49 Estimate::UNKNOWN
50 }
51}
52
53pub trait GroupKernel<M: IndexDomain, K: GroupKey, O: Operand>: Operation<Scope = Group> {
54 type Output: Operand;
55
56 fn execute<'a>(
57 graphrecord: &'a GraphRecord,
58 partition: Partition<'a, M, K, O>,
59 prepared: Self::Prepared<'a>,
60 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>;
61
62 #[allow(unused_variables)]
63 fn estimate(&self, input: Estimate, stats: &Stats) -> Estimate {
64 Estimate::UNKNOWN
65 }
66}
67
68impl<S, C, P> Apply<P, Element> for OperandHandle<S, C>
69where
70 S: ElementShape + ElementTransition<P::OutShape, P::Emission>,
71 C: Arity,
72 P: ElementKernel<S>,
73{
74 type Output = OperandHandle<P::OutShape, <P::Emission as ElementEmission>::OutArity<C>>;
75
76 fn apply<'a>(
77 graphrecord: &'a GraphRecord,
78 values: Self::ReturnValue<'a>,
79 prepared: P::Prepared<'a>,
80 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>
81 where
82 Self: 'a,
83 {
84 let pipeline = P::pipeline(graphrecord, prepared)?;
85
86 Ok(S::apply(values, pipeline))
87 }
88
89 fn estimate(operation: &P, input: Estimate, stats: &Stats) -> Estimate {
90 operation.estimate(input, stats)
91 }
92}
93
94impl<S: ElementShape, C: Arity, P: LaneKernel<S, C>> Apply<P, Lane> for OperandHandle<S, C> {
95 type Output = P::Output;
96
97 fn apply<'a>(
98 graphrecord: &'a GraphRecord,
99 values: Self::ReturnValue<'a>,
100 prepared: P::Prepared<'a>,
101 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>
102 where
103 Self: 'a,
104 {
105 P::execute(graphrecord, values, prepared)
106 }
107
108 fn estimate(operation: &P, input: Estimate, stats: &Stats) -> Estimate {
109 operation.estimate(input, stats)
110 }
111}
112
113impl<M, K, S, C, P> Apply<P, Element> for GroupOperand<M, K, OperandHandle<S, C>>
114where
115 M: IndexDomain,
116 K: GroupKey,
117 S: ElementShape + ElementTransition<P::OutShape, P::Emission>,
118 C: Arity,
119 P: ElementKernel<S>,
120{
121 type Output = GroupOperand<M, K, <OperandHandle<S, C> as Apply<P, Element>>::Output>;
122
123 fn apply<'a>(
124 graphrecord: &'a GraphRecord,
125 values: Self::ReturnValue<'a>,
126 prepared: P::Prepared<'a>,
127 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>
128 where
129 Self: 'a,
130 {
131 Ok(values.map_payloads(|_, _, payload| {
132 payload.and_then(|values| {
133 <OperandHandle<S, C> as Apply<P, Element>>::apply(
134 graphrecord,
135 values,
136 prepared.clone(),
137 )
138 })
139 }))
140 }
141
142 fn estimate(operation: &P, mut input: Estimate, stats: &Stats) -> Estimate {
143 input.per_group = input.per_group.map(|estimate| {
144 Box::new(<OperandHandle<S, C> as Apply<P, Element>>::estimate(
145 operation, *estimate, stats,
146 ))
147 });
148 input
149 }
150}
151
152impl<M: IndexDomain, K: GroupKey, S: ElementShape, C: Arity, P: LaneKernel<S, C>> Apply<P, Lane>
153 for GroupOperand<M, K, OperandHandle<S, C>>
154{
155 type Output = GroupOperand<M, K, <OperandHandle<S, C> as Apply<P, Lane>>::Output>;
156
157 fn apply<'a>(
158 graphrecord: &'a GraphRecord,
159 values: Self::ReturnValue<'a>,
160 prepared: P::Prepared<'a>,
161 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>
162 where
163 Self: 'a,
164 {
165 Ok(values.map_payloads(|_, _, payload| {
166 payload.and_then(|values| {
167 <OperandHandle<S, C> as Apply<P, Lane>>::apply(
168 graphrecord,
169 values,
170 prepared.clone(),
171 )
172 })
173 }))
174 }
175
176 fn estimate(operation: &P, mut input: Estimate, stats: &Stats) -> Estimate {
177 input.per_group = input.per_group.map(|estimate| {
178 Box::new(<OperandHandle<S, C> as Apply<P, Lane>>::estimate(
179 operation, *estimate, stats,
180 ))
181 });
182 input
183 }
184}
185
186impl<M, K, S, C, P> Apply<P, Group> for GroupOperand<M, K, OperandHandle<S, C>>
187where
188 M: IndexDomain,
189 K: GroupKey,
190 S: ElementShape,
191 C: Arity,
192 P: GroupKernel<M, K, OperandHandle<S, C>>,
193{
194 type Output = P::Output;
195
196 fn apply<'a>(
197 graphrecord: &'a GraphRecord,
198 values: Self::ReturnValue<'a>,
199 prepared: P::Prepared<'a>,
200 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>
201 where
202 Self: 'a,
203 {
204 P::execute(graphrecord, values, prepared)
205 }
206
207 fn estimate(operation: &P, input: Estimate, stats: &Stats) -> Estimate {
208 operation.estimate(input, stats)
209 }
210}
211
212impl<M, K, N, L, O, P, S> Apply<P, S> for GroupOperand<M, K, GroupOperand<N, L, O>>
213where
214 M: IndexDomain,
215 K: GroupKey,
216 N: IndexDomain,
217 L: GroupKey,
218 O: Operand,
219 P: Operation<Scope = S>,
220 S: OperationScope,
221 GroupOperand<N, L, O>: Apply<P, S>,
222{
223 type Output = GroupOperand<M, K, <GroupOperand<N, L, O> as Apply<P, S>>::Output>;
224
225 fn apply<'a>(
226 graphrecord: &'a GraphRecord,
227 values: Self::ReturnValue<'a>,
228 prepared: P::Prepared<'a>,
229 ) -> QueryResult<<Self::Output as EvaluateOperand>::ReturnValue<'a>>
230 where
231 Self: 'a,
232 {
233 Ok(values.map_payloads(|_, _, payload| {
234 payload.and_then(|values| {
235 <GroupOperand<N, L, O> as Apply<P, S>>::apply(graphrecord, values, prepared.clone())
236 })
237 }))
238 }
239
240 fn estimate(operation: &P, mut input: Estimate, stats: &Stats) -> Estimate {
241 input.per_group = input.per_group.map(|estimate| {
242 Box::new(<GroupOperand<N, L, O> as Apply<P, S>>::estimate(
243 operation, *estimate, stats,
244 ))
245 });
246 input
247 }
248}