[tkaymak]

Part of #40949, part of #36841.

The streaming translator now accepts a stateless ParDo with additional outputs. It reuses the batch translation, the DoFn writes wide rows with one column per tag and each output filters its column. The wide rows are persisted only in batch, since persist on a streaming Dataset throws, the same guard as PipelineTranslator. A stateful ParDo with additional outputs is still rejected, BeamStatefulProcessor would merge every tag into the main output, so #40949 stays open for that case.

Combine.perKey().withHotKeyFanout() is rejected in streaming for now. With this change it would translate, but CombineTest$BasicTests.testHotKeyCombining then computes a wrong mean (2.0645 instead of 2.0), tracked in #41199. In streaming each consumed tag reruns the DoFn, as any shared upstream does today, tracked in #41198 with measured counts.

Of the 17 tests excluded from the streaming ValidatesRunner suite under #40949, which failed with "ParDo with additional outputs", 14 pass now, among them all 6 MetricsTest$AttemptedMetricTests. The two hot key tests stay excluded under #41199 and testValueStateTaggedOutput under #40949. Locally the streaming suite runs 212 tests with 0 failures, the batch Structured Streaming suites on Spark 4 and 3 and the full :runners:spark:4:test and :runners:spark:3:test pass. The trigger files run both ValidatesRunner PostCommits here. The streaming suite has a known flake on master, addressed by #41137.

[tkaymak]

R: @Abacn

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers