Consolidate produces batch of data #31155
Conversation
antiguru
force-pushed
the
consolidate_rc
branch
from
39597c4 to
ef350fd
Compare
src/timely-util/src/operator.rs
Outdated
| @@ -777,7 +794,7 @@ pub fn consolidate_pact<Ba, P, G>( | |||
| stream: &StreamCore<G, Ba::Input>, | |||
| pact: P, | |||
| name: &str, | |||
| ) -> StreamCore<G, Ba::Output> | |||
| ) -> StreamCore<G, Rc<Vec<Ba::Output>>> | |||
There was a problem hiding this comment.
The purpose of the Rc isn't clear to me. Is it a safe-guard to avoid accidental expensive cloning of Vecs when the output stream is later forked?
There was a problem hiding this comment.
It is so that a chain of updates is inseparable, no matter what channel it traverses.
There was a problem hiding this comment.
The concern being that Timely might otherwise try to pull the Vec apart while passing it through channels? Would be great to have a comment explaining this!
There was a problem hiding this comment.
I changed the PR such that consolidate_pact produces a Stream<G, Vec<Ba::Output>>, which means that the elements we're transferring are Vec<_>. This gives us the certainty that Timely will not look into the chains.
antiguru
force-pushed
the
consolidate_rc
branch
from
ef350fd to
0b4a97c
Compare
Instead of flattening the outputs of consolidate, provide the output as a whole batch, encoded as an `Vec<C>`. This allows downstream operators to act on a whole batch at once, without relying on undocumented properties of how Timely channels behave. For operators that need to flatten the output need slightly different logic. The copy-to-s3 operator sees some changes, which are roughly: * Accept the chain formed by `consolidate_pact`, and rely on it partitioning the data according to the exchange function. * Skip encoding the batch, and recompute it when writing the data. Signed-off-by: Moritz Hoffmann <[email protected]>
antiguru
force-pushed
the
consolidate_rc
branch
from
0b4a97c to
c55fbdd
Compare
Instead of flattening the outputs of consolidate, provide the output as a
whole batch, encoded as an
Rc<Vec<C>>. This allows downstream operatorsto act on a whole batch at once, without relying on undocumented properties
of how Timely channels behave. For operators that need to flatten the
output need slightly different logic.
An alternative approach would be to wrap the chain in a
Box, but at thistime, Timely doesn't recognize
Boxas a valid container type.Checklist
$T ⇔ Proto$Tmapping (possibly in a backwards-incompatible way), then it is tagged with aT-protolabel.