[Spark 4] Filter late data against the key's previous watermark in the streaming stateful ParDo

[tkaymak]

Spark 4 Structured Streaming gives every stateful operator in a query the same event time watermark, transformWithState in EventTime mode reports its output watermark equal to its input watermark. Timer output of a stateful ParDo emitted behind that watermark reached a downstream stateful ParDo in the same batch and was dropped as late.

This follows the legacy runner rule in SparkGroupAlsoByWindowViaWindowSet: late data is filtered against the watermark of the key's previous visit, timers fire against the current one, through a new SparkTimerInternals.forWatermarks and a third state variable beamWatermark. Spark 4.0.2 accepts it as a new column family on restart from a #40281 checkpoint, existing keys take the first visit path once.

The bound is per key, not per batch, so a key seen for the first time or idle for several batches accepts data a per batch bound would drop. Spark keeps the per batch value in eventTimeWatermarkForLateEvents but does not expose it through TimerValues, #40446 tracks the tighter bound.

The new test testChainedStatefulParDoTimerOutputSurvives fails on master and passes with this change.

Part of #36841.

[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

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.74%. Comparing base (69bdf5f) to head (9b64737).
⚠️ Report is 30 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40447      +/-   ##
============================================
- Coverage     59.04%   58.74%   -0.30%     
+ Complexity    15624    13910    -1714     
============================================
  Files          2797     2584     -213     
  Lines        280750   271444    -9306     
  Branches      12488    11229    -1259     
============================================
- Hits         165764   159467    -6297     
+ Misses       108540   105985    -2555     
+ Partials       6446     5992     -454     

Flag	Coverage Δ
java	65.25% <100.00%> (-0.19%)	⬇️

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.

[Abacn]

Thanks!