seal empty batches in reduce, as in arrange
@@ -436,6 +436,9 @@ where
// Pull in any subsequent empty batches we believe to exist.
source_trace.advance_upper(&mut upper_limit);
+ // Only if our upper limit has advanced should we do work.
+ if upper_limit != lower_limit {
+
// If we have no capabilities, then we (i) should not produce any outputs and (ii) could not send
// any produced outputs even if they were (incorrectly) produced. We cannot even send empty batches
// to indicate forward progress, and must hope that downstream operators look at progress frontiers
@@ -600,6 +603,9 @@ where
// ensure that observed progres is reflected in the output.
output_writer.seal(upper_limit.elements());
}
+ else {
+ output_writer.seal(upper_limit.elements());
+ }
// We only anticipate future times in advance of `upper_limit`.
source_trace.advance_by(upper_limit.elements());
@@ -610,6 +616,7 @@ where
output_reader.distinguish_since(upper_limit.elements());
}
}
+ }
)
};
