HeartSaVioR commented on code in PR #59048:
URL: https://github.com/apache/spark/pull/59048#discussion_r4162784560
##########
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:
My hunch is that we don't really need to do the projection (initializing
buffer) in a single partition; instead, we can differentiate the process of
global aggregation in StatefulStreamlineAggregateExec. With this, we still use
all the partitions to initialize buffer for the data, and perform grouping
operation in a single partition.
In the other approach, the code changes of AggUtils and
ProjectAggregationBufferExec aren't necessary. Instead, we make a code change
in StatefulStreamlineAggregateExec to indicate the global aggregation and
handle the data as such.
This also fits with the approach of HashAggregateExec.
##########
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:
It looks like a potential race. Maybe we could logically deal with this via
checking the result of the input for 1 and 2 (so there will be no further
output), or some other way for synchronization.
##########
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:
For sure, if this approach gets better performance we can go with it. I
suspect this approach would/could be slower.
--
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]
