[tkaymak]

Part of #36841.

This adds windowed GroupByKey to the Spark 4 Structured Streaming runner and lets Combine.perKey and Combine.globally().withoutDefaults() expand onto it. Combine.globally() with a default value, custom triggers, accumulating fired panes and merging windows are rejected at translation with a message naming the issue. Accumulating panes are deferred scope: with the default trigger they only matter for late panes inside allowed lateness, they follow with trigger support.

BeamStatefulProcessor becomes an abstract base that keeps the state, timer and per key watermark lag code from #40447 unchanged, decodes the encoded key once and calls two hooks. ParDoStatefulProcessor is the stateful ParDo code moved out of it. GroupByKeyStatefulProcessor hosts a GroupAlsoByWindowViaWindowSetNewDoFn with SystemReduceFn.buffering behind DoFnRunners.lateDataDroppingRunner, fed one KeyedWorkItem per visit for elements and one for due timers. Due timers fire in one pass, with the default trigger and non merging windows ReduceFnRunner.onTimers sets no timer that is already due.

GroupByKeyStreamingTranslator groups by encoded key bytes with Encoders.BINARY() like the stateful ParDo translator since #40516, so the null key of the Combine.globally expansion works.

One behavior change for stateful ParDo: due event time timers now also fire at the end of handleInputRows instead of only in handleExpiredTimer. The wake up reconciliation removes the Spark timer, so nothing fires twice.

WindowedGroupByKeyStreamingTest runs six live cases (fixed windows, sliding windows with Count.perKey, Combine.globally without defaults, late data dropped with pane timing asserted, a stateful ParDo timer output chained into GroupByKey, and an expired key that leaves no rows in beamState, beamTimers and beamWatermark). The late data and chained cases fail when the per key watermark lag is disabled, the expired key case fails without the cleanup from #40550, which the base class now applies to GroupByKey keys too. The stateRows helper from #40550 moves to StreamingTestUtils. PipelineTranslatorStreamingTest gains a table row per new guard. Full :runners:spark:4:test and :runners:spark:3:test pass locally, and the SparkStructuredStreaming ValidatesRunner trigger file is touched.

The CHANGES.md entry for Spark 4 streaming follows with the Impulse and PAssert slice that enables the streaming ValidatesRunner suite.

[tkaymak]

R: @Abacn

[github-actions]

📋 Review Summary

This PR cleanly implements windowed GroupByKey, Combine.perKey, and Combine.globally().withoutDefaults() for the Spark 4 Structured Streaming runner using transformWithState. The refactoring of BeamStatefulProcessor into a shared abstract base class for ParDoStatefulProcessor and GroupByKeyStatefulProcessor is well-structured, preserves state/timer/watermark cleanup guarantees, and is backed by thorough end-to-end streaming and translation tests.

🔍 General Feedback

• The separation of BeamStatefulProcessor, ParDoStatefulProcessor, and GroupByKeyStatefulProcessor effectively shares state, timer, and per-key watermark management without duplicating code.
• Firing due event-time timers at the end of handleInputRows after advancing the watermark is a sound improvement, with reconcileWakeupTimer properly cleaning up Spark timers to prevent duplicate firing.
• Unsupported windowing and trigger configurations (merging windows, custom triggers, accumulating panes, and Combine.globally() with defaults) are guarded with clear error messages and tested in PipelineTranslatorStreamingTest.

---

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

[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

[Abacn]

Both processElement and processTimers invoke stepContext.stateInternals(key), in addition to GroupAlsoByWindowViaWindowSetNewDoFn.create() in GroupByKeyStatefulProcessor.java:L107. Is this duplicated SparkStateInternals creation per key?

[tkaymak]

Yes, it built several views over the same MapState per visit. Done in 4a7beef: each handler call builds one SparkStateInternals and passes it to the hooks, and the GroupAlsoByWindow factory reads it from the step context (k -> ctx.stateInternals()).