Add names to joins

differential-dataflow/examples/columnar/main.rs

@@ -146,6 +146,7 @@ mod reachability {
                 join_traces::<_, _, _, _, ValColBuilder<(Node, (), IterTime, Diff)>>(
                     edges_arr,
                     reach_arr,
+                    "Join",
                     |_src, dst, (), time, d1, d2, session| {
                         use differential_dataflow::difference::Multiply;
                         let dst: Node = *dst;

differential-dataflow/src/operators/arrange/arrangement.rs

@@ -205,7 +205,7 @@
                     while let Some(key) = cursor.get_key(batch) {
                         while let Some(val) = cursor.get_val(batch) {
                             for datum in logic(key, val) {
                                 cursor.map_times(batch, |time, diff| {
                                     session.give((datum.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
                                 });
                             }
@@ -251,6 +251,7 @@
         join_traces::<_, _, _, _, crate::consolidation::ConsolidatingContainerBuilder<_>>(
             self,
             other,
+            "Join",
             move |k, v1, v2, t, d1, d2, c| {
                 for datum in result(k, v1, v2, t, d1, d2) {
                     c.push_into(datum);

differential-dataflow/src/operators/join.rs

@@ -60,7 +60,7 @@ pub enum Fresh {
 /// The "correctness" of this method depends heavily on the behavior of the supplied `result` function.
 ///
 /// [`AsCollection`]: crate::collection::AsCollection
-pub fn join_traces<'scope, Tr1, Tr2, KC, L, CB>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, result: L) -> Stream<'scope, Tr1::Time, CB::Container>
+pub fn join_traces<'scope, Tr1, Tr2, KC, L, CB>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, name: &str, result: L) -> Stream<'scope, Tr1::Time, CB::Container>
 where
     Tr1: TraceReader<Batch: Navigable>+'static,
     Tr2: TraceReader<Batch: Navigable, Time = Tr1::Time>+'static,
@@ -71,7 +71,7 @@ where
     L: FnMut(KC::ReadItem<'_>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut CB)+'static,
     CB: ContainerBuilder<Container: Default> + 'static,
 {
-    join_with_tactic(arranged1, arranged2, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, _, CB>::new(result))
+    join_with_tactic(arranged1, arranged2, name, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, _, CB>::new(result))
 }
 
 /// Drives an equijoin of two traces using a supplied [`JoinTactic`].
@@ -80,7 +80,7 @@ where
 /// compaction) and routes the per-batch work through the tactic. It requires only `TraceReader` of its
 /// inputs, never `Navigable`: it extracts trace batches via `batches_through`, and building cursors over
 /// them (if that is how the join proceeds) is the tactic's concern.
-pub fn join_with_tactic<'scope, Tr1, Tr2, T, C>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, mut tactic: T) -> Stream<'scope, Tr1::Time, C>
+pub fn join_with_tactic<'scope, Tr1, Tr2, T, C>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, name: &str, mut tactic: T) -> Stream<'scope, Tr1::Time, C>
 where
     Tr1: TraceReader+'static,
     Tr2: TraceReader<Time = Tr1::Time>+'static,
@@ -92,7 +92,7 @@ where
     let mut trace2 = arranged2.trace;
 
     let scope = arranged1.stream.scope();
-    arranged1.stream.binary_frontier(arranged2.stream, Pipeline, Pipeline, "Join", move |capability, info| {
+    arranged1.stream.binary_frontier(arranged2.stream, Pipeline, Pipeline, name, move |capability, info| {
 
         // Acquire an activator to reschedule the operator when it has unfinished work.
         use timely::scheduling::Activator;

interactive/src/backend/corgi.rs

@@ -271,7 +271,7 @@ impl Backend for CorgiBackend {
         // columns directly as `CorgiContainer`s — column-native, no row round-trip.
         if compilable(&projection.key) && compilable(&projection.val) {
             let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone()));
-            join_with_tactic::<_, _, _, CC>(l, r, tactic).as_collection()
+            join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic).as_collection()
         } else {
             // Projections the lowering can't compile take the same shape as `linear`'s gate:
             // join with the identity projection (compilable by construction), then apply the
@@ -282,7 +282,7 @@ impl Backend for CorgiBackend {
             let key = Term::Var(0);
             let val = Term::Tuple(vec![Term::Var(1), Term::Var(2)]);
             let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(key, val));
-            let joined = join_with_tactic::<_, _, _, CC>(l, r, tactic).as_collection();
+            let joined = join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic).as_collection();
             let rebased = Projection { key: rebase_join_term(&projection.key), val: rebase_join_term(&projection.val) };
             Self::linear(joined, vec![LinearOp::Project(rebased)], 0)
         }