HeartSaVioR commented on code in PR #59048:
URL: https://github.com/apache/spark/pull/59048#discussion_r4162810172
##########
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 spot of 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.
--
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]
