MCPcopy Create free account
hub / github.com/comnik/declarative-dataflow / implement

Method implement

src/plan/union.rs:43–85  ·  view source on GitHub ↗
(
        &self,
        nested: &mut Iterative<'b, S, u64>,
        local_arrangements: &VariableMap<Iterative<'b, S, u64>>,
        context: &mut I,
    )

Source from the content-addressed store, hash-verified

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}

Callers

nothing calls this directly

Calls 2

merge_withMethod · 0.80
projectedMethod · 0.80

Tested by

no test coverage detected