leave clues behind
@@ -313,8 +313,8 @@ impl<G, K, V, R1, T1> JoinCore<G, K, V, R1> for Arranged<G,K,V,R1,T1>
let mut trace2 = Some(other.trace.clone());
// acknowledged frontier for each input.
- let mut acknowledged1: Option<Vec<G::Timestamp>> = None;//vec![G::Timestamp::minimum()];
- let mut acknowledged2: Option<Vec<G::Timestamp>> = None;//vec![G::Timestamp::minimum()];
+ let mut acknowledged1: Option<Vec<G::Timestamp>> = None;
+ let mut acknowledged2: Option<Vec<G::Timestamp>> = None;
// deferred work of batches from each input.
let mut todo1 = Vec::new();
@@ -345,6 +345,7 @@ impl<G, K, V, R1, T1> JoinCore<G, K, V, R1> for Arranged<G,K,V,R1,T1>
data.swap(&mut input1_buffer);
for batch1 in input1_buffer.drain(..) {
if let Some(acknowledged2) = &acknowledged2 {
+ // TODO : cursor_through may be problematic for pre-merged traces.
let (trace2_cursor, trace2_storage) = trace2.cursor_through(&acknowledged2[..]).unwrap();
let batch1_cursor = batch1.cursor();
todo1.push(Deferred::new(trace2_cursor, trace2_storage, batch1_cursor, batch1.clone(), capability.clone(), |r2,r1| (r1.clone()) * (r2.clone())));
@@ -362,6 +363,7 @@ impl<G, K, V, R1, T1> JoinCore<G, K, V, R1> for Arranged<G,K,V,R1,T1>
data.swap(&mut input2_buffer);
for batch2 in input2_buffer.drain(..) {
if let Some(acknowledged1) = &acknowledged1 {
+ // TODO : cursor_through may be problematic for pre-merged traces.
let (trace1_cursor, trace1_storage) = trace1.cursor_through(&acknowledged1[..]).unwrap();
let batch2_cursor = batch2.cursor();
todo2.push(Deferred::new(trace1_cursor, trace1_storage, batch2_cursor, batch2.clone(), capability.clone(), |r1,r2| (r1.clone()) * (r2.clone())));
