HeartSaVioR opened a new pull request, #59621: URL: https://github.com/apache/spark/pull/59621
### What changes were proposed in this pull request? For [SPARK-60373](https://issues.apache.org/jira/browse/SPARK-60373), count state rows removed by `removeAndReturnOldState` in the streaming join's `numRowsRemoved` metric. Increment the counter as each removed row is consumed, before filtering matched rows from outer-join output. ### Why are the changes needed? Left and right outer joins currently count removals only from the side cleaned up after output generation. Full outer joins remove state from both sides while generating output, so their removal metric stays zero even when state is deleted. The windowed full-outer regression stores five rows on each side, including three matching pairs. After restarting from the same checkpoint and advancing the watermark, all ten old state rows are removed, while four unmatched rows produce null-extended output. Query progress should report ten removed rows; the original implementation reports zero. The left/right range-join regressions similarly remove six state rows, but the original metric omits removals from the preserved side. The omission dates to Spark 3.2.0, when `numRowsRemoved` was introduced. ### Does this PR introduce _any_ user-facing change? Yes. Streaming query progress correctly counts all evicted outer-join state rows in `numRowsRemoved`, including matched rows that do not produce null-extended output. This changes metric accounting only; query results, eviction decisions, state layout, and checkpoint compatibility remain unchanged. ### How was this patch tested? Added query-progress assertions to the existing end-to-end left/right range-join and full-outer checkpoint-restart tests, retaining their output and state-row assertions. ```text build/sbt 'sql/testOnly *StreamingOuterJoin*Suite -- -z "with watermark range condition"' build/sbt 'sql/testOnly *StreamingFullOuterJoin*Suite -- -z "windowed full outer join"' ``` Before the fix, the full-outer regression failed with both HDFS and RocksDB state store providers (`0 did not equal 10`). With the fix, all 18 cases passed: 12 left/right range-join cases and 6 full-outer checkpoint-restart cases. Coverage includes state format V2 with HDFS/RocksDB and V3/V4 with RocksDB changelog checkpointing disabled/enabled. SQL Compile/Test scalastyle reported zero errors and warnings. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6) -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
