diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs index 966bb7105..dc4daf24f 100644 --- a/differential-dataflow/examples/columnar/main.rs +++ b/differential-dataflow/examples/columnar/main.rs @@ -143,7 +143,7 @@ mod reachability { // join_traces with ValColBuilder: produces Stream<_, RecordedUpdates<...>>. let proposed = - join_traces::<_, _, _, ValColBuilder<(Node, (), IterTime, Diff)>>( + join_traces::<_, _, _, _, ValColBuilder<(Node, (), IterTime, Diff)>>( edges_arr, reach_arr, |_src, dst, (), time, d1, d2, session| { diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index b5261f2a7..6bbce6419 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -227,26 +227,28 @@ impl<'scope, Tr1: TraceReader+'static> Arranged<'scope, Tr1> { /// A convenience method to join and produce `VecCollection` output. /// /// Avoid this method, as it is likely to evolve into one without the `VecCollection` opinion. - pub fn join_core(self, other: Arranged<'scope, Tr2>, mut result: L) -> VecCollection<'scope, Tr1::Time,I::Item,>::Output> + pub fn join_core(self, other: Arranged<'scope, Tr2>, mut result: L) -> VecCollection<'scope, Tr1::Time,I::Item,>::Output> where Tr2: TraceReader+Clone+'static, // Pin the cursor diffs to named params `R1`/`R2`: a `Multiply` bound on a projection // does not connect to its use-site (the solver normalizes the use but not the bound's // subject), so we constrain plain params instead. - BatchCursor: Cursor, + BatchCursor: Cursor, BatchCursor: Cursor, - for<'a> BatchCursor: Cursor = BatchKey<'a, Tr1>>, + KC: BatchContainer, + for<'a> BatchCursor: Cursor = KC::ReadItem<'a>>, + for<'a> BatchCursor: Cursor = KC::ReadItem<'a>>, R1: Multiply + Clone, I: IntoIterator, - L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>)->I+'static + L: FnMut(KC::ReadItem<'_>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>)->I+'static { - let mut result = move |k: BatchKey<'_, Tr1>, v1: BatchVal<'_, Tr1>, v2: BatchVal<'_, Tr2>, t: Tr1::Time, r1: &R1, r2: &R2| { + let mut result = move |k: KC::ReadItem<'_>, v1: BatchVal<'_, Tr1>, v2: BatchVal<'_, Tr2>, t: Tr1::Time, r1: &R1, r2: &R2| { let r = (r1.clone()).multiply(r2); result(k, v1, v2).into_iter().map(move |d| (d, t.clone(), r.clone())) }; use crate::operators::join::join_traces; - join_traces::<_, _, _, crate::consolidation::ConsolidatingContainerBuilder<_>>( + join_traces::<_, _, _, _, crate::consolidation::ConsolidatingContainerBuilder<_>>( self, other, move |k, v1, v2, t, d1, d2, c| { diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs index 58b42e6f3..023c8648d 100644 --- a/differential-dataflow/src/operators/join.rs +++ b/differential-dataflow/src/operators/join.rs @@ -17,8 +17,9 @@ use timely::dataflow::operators::Capability; use crate::lattice::Lattice; use crate::operators::arrange::Arranged; -use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Cursor, Navigable, TraceReader}; +use crate::trace::{BatchCursor, BatchDiff, BatchReader, BatchVal, Cursor, Navigable, TraceReader}; use crate::trace::cursor::cursor_list; +use crate::trace::implementations::containers::BatchContainer; use crate::operators::ValueHistory; /// A type that can manage the joining of lists of batches. @@ -59,13 +60,15 @@ 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, 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>, result: L) -> Stream<'scope, Tr1::Time, CB::Container> where Tr1: TraceReader+'static, Tr2: TraceReader+'static, - BatchCursor: Cursor