nahtonaj opened a new pull request, #59048: URL: https://github.com/apache/spark/pull/59048
### What changes were proposed in this pull request? The streamline aggregation plan initializes the global aggregation buffer exactly once per executed empty batch: the non-final ProjectAggregationBufferExec is planned with a single-partition (AllTuples) requirement for a global aggregation, so its empty-input guard fires once for the batch instead of once per empty upstream partition, and it never fires for a batch that has input. A batch whose source RDD has no partitions is also covered. The change corrects comments that misdescribed the projection stages, the state-store write timing, the output-mode timing, and the aggregation scope. ### Why are the changes needed? A global streaming aggregation on the streamline plan produced either no row (a 0-partition source batch) or duplicate seeded rows (one per empty partition, and an extra seeded row inside a batch that had input in another partition). The ordinary plan initializes once and emits the initialized result (count 0, sum NULL) for an executed empty batch, so the streamline plan diverged. ### Does this PR introduce _any_ user-facing change? Yes, a fix: global aggregations under the streamline plan (Trigger.RealTime, and the internal useStreamlineAggregator test config) now emit the initialized/accumulated result exactly once per executed empty batch and never seed a batch that has input, matching the ordinary micro-batch plan. ### How was this patch tested? New and updated tests in StreamingAggregationSuite (streamline-enabled): complete/update empty-first-batch initialization, state continuity across empty and non-empty batches, exact Update emissions for a multi-partition empty batch, no seeded row for a non-empty batch with an empty sibling partition, initialization from a 0-partition source batch, and the coalesce(1) SPARK-21977 shape as a regression guard. The real-time-mode suite pins emitted values for the empty-input case. Local runs: StreamingAggregationSuite 79/79, StreamlineStreamingAggregationRealTimeSuite 6/6, RocksDBStateStoreStreamingAggregationSuite streamline subset 70/70, scalastyle 0 errors/warnings. -- 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]
