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