junaiddshaukat opened a new pull request, #39410:
URL: https://github.com/apache/beam/pull/39410
A fused executable stage can have more than one output — a DoFn with side
outputs, or a Read whose SDF wrapper produces several. The runner rejected
multi-output stages; this routes each output to its own downstream.
How it works:
- Each output PCollection of a multi-output stage gets a relay node
(`StageOutputProcessor`) that stands in as that output's producer.
Downstream
transforms are wired to a producer node by PCollection id, and that node
must
exist when the stage is translated (before its consumers are), so the relay
gives each output a stable producer to wire to.
- The stage tags each harness output with its PCollection id and forwards it
to
the matching relay with `ctx.forward(record, childName)` — in-process
routing
to a named processor node, no extra topic. The relay forwards data
unchanged
and re-stamps the stage's watermark with its own id so a downstream
watermark
aggregator sees a consistent source.
- A single-output stage is unchanged: it forwards to its one downstream
directly
and registers itself as that output's producer.
An output that feeds a GroupByKey is stored in that GroupByKey's own
repartition
topic (the shuffle creates it for the key change); an output feeding a
stateless
ParDo flows in-process. Discussed the storage with @je-ik on Slack.
Result on Beam's suite: `GroupByKeyTest` basic tests now pass (their PAssert
path produces multi-output stages), and the multi-consumer `FlattenTest` is
un-sickbayed (14/14). `CreateTest` stays 5/5. Adds `MultiOutputStageTest`: a
DoFn with a side output whose two outputs each feed a separate GroupByKey.
The remaining `GroupByKeyTest` failures are non-global windowing (a
`GlobalWindow` cast) — sickbayed until windowing lands.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
