[Spark 4] Support stateful ParDo in Structured Streaming via transformWithState

[tkaymak]

Part of #36841, slice 5b.

Stateful ParDo, with state and event time timers, now runs on the Spark 4 Structured Streaming runner through Spark 4 transformWithState. Before this change, the streaming translator rejected it.

How it works

• User state reuses the legacy SparkStateInternals. A new StateCells interface lets it run on a Spark MapState. The existing callers keep the Table backed path.
• Timers reuse the legacy SparkTimerInternals. They are persisted per key in a Spark ValueState. Each key gets one Spark timer as a wake up, at the earliest pending Beam timer plus 1 ms. Spark fires when the expiry is less than or equal to the watermark, and Beam fires only when the watermark is strictly after the timestamp, so the extra 1 ms matches the two rules. Timers fire one at a time, so a callback can still set or delete other timers.
• The per task runner lifecycle is shared with the batch stateful ParDo through StatefulTaskRunner. That lifecycle covers setup on the executor, metrics, and teardown on task completion or failure.
• These are still rejected at translation, with a message naming [Feature Request]: Support Apache Spark 4.x in Beam Spark Runner #36841: processing time and synchronized processing time timers, @OnWindowExpiration, @RequiresTimeSortedInput, and merging windows.

Watermark semantics
The runner's input watermark is Spark's event time watermark, as set up in the earlier slices: max event time minus --watermarkDelayMillis. Lateness and timer firing both follow that watermark. For out of order sources, raise the delay. Spark has no watermark hold for timer output timestamps. That is harmless while only stateless transforms follow, and it is tracked as a blocker for GroupByKey in 5c (#36841 comment).

Tests

• StatefulParDoStreamingTest runs a live query. It covers state across micro batches, timer firing, a callback that deletes a later due timer, and a checkpoint restart.
• PipelineTranslatorStreamingTest has one table driven test for the rejected shapes.
• The existing batch stateful ParDo tests and the legacy SparkStateInternalsTest still pass on Spark 3 and Spark 4.

CHANGES.md follows with GroupByKey in 5c. A ValidatesRunner suite follows once Impulse and PAssert are supported.

[codecov]

Codecov Report

❌ Patch coverage is 77.77778% with 20 lines in your changes missing coverage. Please review.
✅ Project coverage is 58.74%. Comparing base (69bdf5f) to head (e13c345).
⚠️ Report is 25 commits behind head on master.

Files with missing lines	Patch %	Lines
...treaming/translation/batch/StatefulTaskRunner.java	73.58%	8 Missing and 6 partials ⚠️
...am/runners/spark/stateful/SparkStateInternals.java	88.00%	2 Missing and 1 partial ⚠️
...g/translation/batch/StatefulDoFnGroupFunction.java	80.00%	2 Missing ⚠️
...am/runners/spark/stateful/SparkTimerInternals.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40281      +/-   ##
============================================
- Coverage     59.04%   58.74%   -0.31%     
+ Complexity    15624    13909    -1715     
============================================
  Files          2797     2584     -213     
  Lines        280750   271444    -9306     
  Branches      12488    11229    -1259     
============================================
- Hits         165764   159454    -6310     
+ Misses       108540   105996    -2544     
+ Partials       6446     5994     -452     

Flag	Coverage Δ
java	65.24% <77.77%> (-0.20%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @damccorm added as fallback since no labels match configuration

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).

[tkaymak]

R: @Abacn

Slice 5b of #36841, follow up to #40090. CI is green. The watermark semantics are covered in the PR body, and the one known gap for 5c is noted on the issue.

[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]

Thanks, again, could you please trigger PostCommit ValidateRunner Spark StructuredStreaming tests?

[tkaymak]

Thanks @Abacn! All five threads are addressed in 841f15c, one new commit on top so the change should be easy to review and I triggered PostCommit ValidatesRunner Spark StructuredStreaming

[tkaymak]

The PostCommit run is red, but master is red the same way since #40324 (scheduled runs from 2026-10-01 22:46, about 112 failures, "Expected 1 successful assertions, but found N"). The failing set differs randomly between runs and none of the stateful tests fail only here. It looks like the parallel test threads from #40324 make PAssert counters from concurrent pipelines add up on the Spark runner.
I have opened issue #40427 to track this and PR #40428 to fix it.

[tkaymak]

Rebased on master with the Spark fix from #40428, PostCommit ValidatesRunner Spark StructuredStreaming is green now: https://github.com/apache/beam/actions/runs/37522104261