[runners-spark] Support stateful ParDo in the Structured Streaming batch runner

[f-loris]

Adds support for stateful ParDo — state, timers and @RequiresTimeSortedInput — to the Spark Structured Streaming (Dataset-based) runner in batch mode, including additional (tagged) outputs.

Fixes #39779

Approach

Stateful, timer-using and @RequiresTimeSortedInput DoFns are dispatched to a new StatefulParDoTranslatorBatch, which groups by key via KeyValueGroupedDataset#flatMapSortedGroups and orders each group by event time. StatefulDoFnGroupFunction then runs the DoFn per key with InMemoryStateInternals/InMemoryTimerInternals from runners-core.

Because batch has no state store to bridge onto — state is a heap object scoped to a single key — every Beam state type is already implemented by InMemoryStateInternals, so the state categories are enabled together rather than incrementally. Memory is bounded by the state one key holds rather than by that key's element count: groups stream off Spark's spillable sort instead of being materialized, unlike the classic runner's RDD groupByKey.

Sorting uses the timestamp column of the WindowedValue encoder, which is LongType epoch millis, with asc_nulls_last: a null timestamp encodes END_OF_WINDOW, which no concrete timestamp of the same window can exceed.

The DoFn is set up once per task and torn down from a TaskCompletionListener, with state, timers and a bundle per key.

ValidatesRunner

Enabled for validatesStructuredStreamingRunnerBatch: UsesStatefulParDo, UsesKeyInParDo, UsesTimersInParDo, UsesMapState, UsesMultimapState, UsesSetState, UsesOrderedListState, UsesTimerMap.

Results on this branch:

Suite	Result
:runners:spark:3:validatesStructuredStreamingRunnerBatch	304 tests, 0 failures
:runners:spark:4:validatesStructuredStreamingRunnerBatch	303 tests, 0 failures

Also added: unit tests for translator dispatch and preconditions, and execution tests covering per-key state, state isolation across many keys in one partition, @RequiresTimeSortedInput ordering, event-time timers, looping timers, @FinishBundle ordering relative to timers, and tagged outputs.

Notes for reviewers

• No CI covers this suite on Spark 4. beam_PostCommit_Java_ValidatesRunner_Spark4 runs :runners:spark:4:validatesRunner (the classic runner), and beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming is pinned to :runners:spark:3:.... This is pre-existing — the suite has never run on Spark 4 — so the 303/0 above was verified locally. Happy to add a beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming4 workflow here or as a follow-up if you'd like that coverage; it would be a near-copy of the existing one with java-version: '17'.

• Spark 3.4+ requirement. flatMapSortedGroups was added in Spark 3.4, so stateful pipelines are rejected at translation time on 3.1–3.3 with a clear message rather than failing with NoSuchMethodError mid-job. The stateful tests skip on those versions so sparkVersion31Test/sparkVersion33Test stay green.

• PerKeyOrderingTest#testMultipleStatefulOrdering* are excluded by name, not by category. They build on PeriodicImpulse and so are unbounded, but are not categorized as UsesUnboundedPCollections; adding that category upstream would also stop Dataflow and Flink running them, so the exclusion is kept local to this runner.

• The ParDoTest OrderedListState range tests gain UsesOnWindowExpiration, which their DoFns declare but which they were not categorized for. This is the one change outside runners/spark. It is a no-op for other runners: every runner excluding UsesOnWindowExpiration also excludes UsesOrderedListState, so those tests did not run there.

• Deliberate mirroring of the stateless path. createEncoders, createSideInputReader, the output receivers and the persist/split-by-tag block intentionally follow ParDoTranslatorBatch/DoFnPartitionIteratorFactory. Happy to consolidate into shared helpers if preferred.

• @OnWindowExpiration remains unsupported ([Feature Request]: Support onWindowExpiration in Spark runner so that GroupIntoBatches can be used #22524) and is rejected explicitly; UsesOnWindowExpiration stays excluded. Processing-time timers re-armed from @OnTimer do not re-fire in batch, matching the classic runner; this is documented on fireNextTimer.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

Thanks, could you please trigger the following postcommit tests:

https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.yml

https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_ValidatesRunner_Spark4.yml

by making any changes to these trigger files

• .github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json

• .github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json

[f-loris]

@Abacn Thanks for your feedback. Updated the PR based an you comments and triggered the PostCommit runs you mentioned.