nahtonaj commented on code in PR #59048:
URL: https://github.com/apache/spark/pull/59048#discussion_r4168466602
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala:
##########
@@ -462,6 +463,13 @@ object AggUtils {
val aggregateAttributes = aggregateExpressions.map(_.resultAttribute)
ProjectAggregationBufferExec(
+ // A global aggregation must initialize its buffer exactly once per
empty batch. Planning
Review Comment:
Agreed, thanks. I moved the empty-batch initialization into
`StatefulStreamlineAggregateExec`. For a global aggregation that operator
already requires `AllTuples`, so it runs on a single partition; when that
partition has no input it merges the empty grouping key against state
(`processEmptyGroupingKey`), producing the stored value or the initialized
buffer. `ProjectAggregationBufferExec` and `AggUtils` go back to initializing
buffers on every partition, so there is no extra single-partition step, which
matches how `HashAggregateExec` handles it.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamlineStreamingAggregationRealTimeSuite.scala:
##########
@@ -131,4 +134,46 @@ class StreamlineStreamingAggregationRealTimeSuite extends
StreamRealTimeModeSuit
}
)
}
+
+ testWithAllStateVersions("global aggregation over empty input emits the
initialized row") {
+ val inputData = LowLatencyMemoryStream[Int]
+ val sink = new ContinuousMemorySink()
+
+ val agg = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ // Update mode may re-emit the current row for no-data batches in
real-time mode, so the
+ // assertions pin the value of every emitted row instead of counting rows.
+ testStream(agg, OutputMode.Update, sink = sink)(
+ StartStream(),
+ AddData(inputData, 1, 2),
+ Execute { _ =>
+ // Every row is filtered out, so the aggregation input is empty for
this batch: only the
+ // initialized value may be emitted, as the ordinary plan does.
+ eventually(timeout(60.seconds)) {
+ assert(sink.allData.nonEmpty, "expected the initialized row for the
empty batch")
+ }
+ assert(sink.allData.forall(_ == Row(0L, null)),
+ s"unexpected rows for the empty batch: ${sink.allData}")
+ sink.clear()
Review Comment:
Good catch. I removed the `clear()`. The sink does not order rows across
batches, so the test no longer relies on arrival order: it waits until the
accumulated row `(1, -5)` has been emitted at least twice. Only the batch with
`-5` and later batches can produce it, so the second one proves a later empty
batch kept the accumulated value; a reset would emit the initialized row
instead. It also checks that every emitted row is either the initialized or the
accumulated value.
--
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]
