tkaymak opened a new pull request, #40703: URL: https://github.com/apache/beam/pull/40703
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. -- 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]
