correct upper frontier placement
@@ -73,16 +73,17 @@ where
let mut trace = self.trace.clone();
let mut buffer = Vec::new();
- self.stream.unary_frontier(Pipeline, "CountTotal", move |_,_| move |input, output| {
+ self.stream.unary_frontier(Pipeline, "CountTotal", move |_,_| {
// tracks the upper limit of known-complete timestamps.
let mut upper_limit = timely::progress::frontier::Antichain::from_elem(<G::Timestamp as timely::progress::Timestamp>::minimum());
+ move |input, output| {
+
input.for_each(|capability, batches| {
batches.swap(&mut buffer);
let mut session = output.session(&capability);
for batch in buffer.drain(..) {
-
let mut batch_cursor = batch.cursor();
let (mut trace_cursor, trace_storage) = trace.cursor_through(batch.lower().borrow()).unwrap();
upper_limit.clone_from(batch.upper());
@@ -125,6 +126,7 @@ where
trace.advance_upper(&mut upper_limit);
trace.advance_by(upper_limit.borrow());
trace.distinguish_since(upper_limit.borrow());
+ }
})
.as_collection()
}
