diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs index dc4daf24f..dad44cf86 100644 --- a/differential-dataflow/examples/columnar/main.rs +++ b/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; diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 6bbce6419..3aab9716d 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -251,6 +251,7 @@ impl<'scope, Tr1: TraceReader+'static> Arranged<'scope, Tr1> { 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); diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs index 413e61d0e..d33f52b63 100644 --- a/differential-dataflow/src/operators/join.rs +++ b/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+'static, Tr2: TraceReader+'static, @@ -71,7 +71,7 @@ where L: FnMut(KC::ReadItem<'_>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff,&BatchDiff,&mut CB)+'static, CB: ContainerBuilder + 'static, { - join_with_tactic(arranged1, arranged2, cursors::CursorTactic::::new(result)) + join_with_tactic(arranged1, arranged2, name, cursors::CursorTactic::::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