(
&self,
nested: &mut Iterative<'b, S, u64>,
local_arrangements: &VariableMap<Iterative<'b, S, u64>>,
context: &mut I,
)
| 41 | } |
| 42 | |
| 43 | fn implement<'b, T, I, S>( |
| 44 | &self, |
| 45 | nested: &mut Iterative<'b, S, u64>, |
| 46 | local_arrangements: &VariableMap<Iterative<'b, S, u64>>, |
| 47 | context: &mut I, |
| 48 | ) -> (Implemented<'b, S>, ShutdownHandle) |
| 49 | where |
| 50 | T: Timestamp + Lattice, |
| 51 | I: ImplContext<T>, |
| 52 | S: Scope<Timestamp = T>, |
| 53 | { |
| 54 | use differential_dataflow::AsCollection; |
| 55 | use timely::dataflow::operators::Concatenate; |
| 56 | |
| 57 | let mut scope = nested.clone(); |
| 58 | let mut shutdown_handle = ShutdownHandle::empty(); |
| 59 | |
| 60 | let streams = self.plans.iter().map(|plan| { |
| 61 | let relation = { |
| 62 | let (relation, shutdown) = plan.implement(&mut scope, local_arrangements, context); |
| 63 | shutdown_handle.merge_with(shutdown); |
| 64 | relation |
| 65 | }; |
| 66 | |
| 67 | let projected = { |
| 68 | let (projected, shutdown) = |
| 69 | relation.projected(&mut scope, context, &self.variables); |
| 70 | shutdown_handle.merge_with(shutdown); |
| 71 | projected |
| 72 | }; |
| 73 | |
| 74 | projected.inner |
| 75 | }); |
| 76 | |
| 77 | let concat = nested.concatenate(streams).as_collection(); |
| 78 | |
| 79 | let concatenated = CollectionRelation { |
| 80 | variables: self.variables.to_vec(), |
| 81 | tuples: concat.distinct(), |
| 82 | }; |
| 83 | |
| 84 | (Implemented::Collection(concatenated), shutdown_handle) |
| 85 | } |
| 86 | } |
nothing calls this directly
no test coverage detected