[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

[github-actions]

📋 Review Summary

This PR cleanly enables stateless ParDo with additional outputs in the Spark 4 Structured Streaming runner by reusing the existing batch multi-output translation and guarding Dataset.persist() with !allTagsDS.isStreaming(). It also explicitly rejects Stateful ParDo with additional outputs and Combine.perKey().withHotKeyFanout() at translation time, while enabling 14 additional ValidatesRunner streaming tests.

🔍 General Feedback

• The isStreaming() guard in ParDoTranslatorBatch is consistent with the existing caching and lineage-breaking guards in PipelineTranslator.putDataset.
• Rejecting stateful ParDo with additional outputs and Combine.PerKeyWithHotKeyFanout early in PipelineTranslatorStreaming prevents silent data corruption or incorrect output merging, with clear unit test coverage in PipelineTranslatorStreamingTest.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.