You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2026/10/03 13:31:39 UTC
(spark) branch branch-4.x updated: [SPARK-59788][SS] Initialize the global aggregation once per empty batch in the streamline plan
This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 0e2a66bd52ad [SPARK-59788][SS] Initialize the global aggregation once per empty batch in the streamline plan
0e2a66bd52ad is described below
commit 0e2a66bd52ad6388a8377e48dada2ab062ded5fb
Author: Jonathan Gao <jo...@hotmail.com>
AuthorDate: Sat Oct 3 22:31:08 2026 +0900
[SPARK-59788][SS] Initialize the global aggregation once per empty batch in the streamline plan
### What changes were proposed in this pull request?
The streamline aggregation plan initializes the global aggregation buffer exactly once per executed
empty batch: StatefulStreamlineAggregateExec, which already runs on a single partition (AllTuples)
for a global aggregation, merges the empty grouping key against state when that partition has no
input, producing the stored value or the initialized buffer. Buffer initialization still runs on
every input partition, and nothing is seeded 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 without relying on
cross-batch sink arrival order, and checks that a later empty batch keeps the accumulated value.
Local runs: StreamingAggregationSuite, StreamlineStreamingAggregationRealTimeSuite and
RocksDBStateStoreStreamingAggregationSuite: 243 passed, 0 failed; scalastyle 0 errors/warnings.
Closes #59048 from nahtonaj/sc-246233-streamline-empty-global.
Authored-by: Jonathan Gao <jo...@hotmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit 2f4102c8117a087aba1324d52b268f7966670136)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 11 +-
.../spark/sql/execution/SparkStrategies.scala | 3 +-
.../spark/sql/execution/aggregate/AggUtils.scala | 3 +-
.../stateful/ProjectAggregationBufferExec.scala | 10 +-
.../stateful/StatefulStreamlineAggregateExec.scala | 56 +++++--
.../streaming/runtime/RealTimeModeAllowlist.scala | 7 +-
.../sql/streaming/StreamingAggregationSuite.scala | 177 +++++++++++++++++++++
...reamlineStreamingAggregationRealTimeSuite.scala | 48 ++++++
8 files changed, 290 insertions(+), 25 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 1c50810684ff..fcc3bbc2cfbc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3864,10 +3864,13 @@ object SQLConf {
.internal()
.doc("Test/development only, not intended for production use. When true, plan a streaming " +
"aggregation with the streamline aggregation operator, which merges each input row " +
- "against state and emits immediately, instead of the microbatch operators that only emit " +
- "once the batch ends. Real-Time Mode queries use the streamline operator regardless of " +
- "this config; this flag exists only so the operator can be exercised under an ordinary " +
- "microbatch trigger in tests, and changes an aggregation's output timing when set.")
+ "against state and emits an intermediate result per input in Update mode, instead of the " +
+ "microbatch operators that only emit once the batch ends. Append and Complete drain the " +
+ "input before producing their mode-specific output, and only non-session aggregations " +
+ "are planned this way (session windows use a separate planning path). Real-Time Mode " +
+ "queries use the streamline operator regardless of this config; this flag exists only so " +
+ "the operator can be exercised under an ordinary microbatch trigger in tests, and " +
+ "changes an aggregation's output timing when set.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.booleanConf
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index c25414a009c0..4d48bed992fd 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -619,7 +619,8 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
// A Real-Time Mode batch runs until its duration elapses rather than until its input
// is exhausted, so an aggregation that only emits once the batch ends would hold every
// result back for the whole batch. Plan the streamline operator instead, which merges
- // each input row against state and emits immediately.
+ // each input row against state; Update mode can emit an intermediate result per
+ // input.
if (isRealTimeMode(child) ||
conf.getConf(SQLConf.STREAMING_USE_STREAMLINE_AGGREGATOR)) {
AggUtils.planStreamlineStreamingAggregation(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
index 1bb5b0e916b3..9b0a656b4ece 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
@@ -441,7 +441,8 @@ object AggUtils {
* - For each input, do the following
* - Read the previous value for grouping key in state store
* - Merge the input and previous value (if any)
- * - Store the new value to the state store
+ * - Hold the new value in a bounded dirty-write cache, which is flushed to the state
+ * store on cache eviction or when the input iterator completes
* - Complete (output the current result of the aggregation)
*
* The concept is to aggregate only between input and the state store, enabling an output to
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/ProjectAggregationBufferExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/ProjectAggregationBufferExec.scala
index 7046e4236aa0..c8ba2a4b284d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/ProjectAggregationBufferExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/ProjectAggregationBufferExec.scala
@@ -28,8 +28,10 @@ import org.apache.spark.sql.execution.metric.SQLMetrics
* This class handles the part of aggregation functions in the input rows, based on the function's
* mode. This class intends to either initialize the aggregation buffer or complete the aggregate
* buffer and produce the result, so it is expected to be used for two aggregate modes:
- * 1) partial merge 2) final. This class is pass-through and does not perform the actual
- * aggregation.
+ * 1) partial, which initializes the buffer for each input row, and 2) final, which completes the
+ * merged buffer. The partial merge stage is the stateful streamline aggregate, not this class;
+ * this class only initializes or completes a single row's buffer and never combines rows with
+ * each other.
*/
case class ProjectAggregationBufferExec(
requiredChildDistributionExpressions: Option[Seq[Expression]] = None,
@@ -82,8 +84,8 @@ case class ProjectAggregationBufferExec(
/**
* This class is an implementation of GenericBufferAggregationIterator which only handles the
- * aggregation buffer of input, depending on the aggregate mode. This class is pass-through
- * and does not perform the actual aggregation.
+ * aggregation buffer of input, depending on the aggregate mode. It initializes or completes the
+ * buffer of one row at a time and does not combine rows with each other.
*/
class ProjectAggregationBufferProcessor(
partIndex: Int,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/StatefulStreamlineAggregateExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/StatefulStreamlineAggregateExec.scala
index 18d7d1a68b5f..a93499295789 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/StatefulStreamlineAggregateExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/StatefulStreamlineAggregateExec.scala
@@ -174,20 +174,34 @@ case class StatefulStreamlineAggregateExec(
// one is produced.
//
// flushDirtyWrites is aggIter's completion action, so it runs when aggIter is exhausted and
- // BEFORE store.commit() in every mode below: Complete drains aggIter (line ~191) ahead of the
- // commit iterator; Append drains it (~222) ahead of its own; Update chains it inside the
- // outer CompletionIterator whose completion commits, and draining the outer drains aggIter
- // first. This ordering is what lets flushDirtyWrites surface a state-write failure (see its
- // rethrow) in time to fail the task before the batch commits -- a partial write can never be
- // committed.
+ // BEFORE store.commit() in every mode below: the Complete branch drains aggIter ahead of
+ // the commit iterator; the Append branch drains it ahead of its own; Update chains it
+ // inside the outer CompletionIterator whose completion commits, and draining the outer
+ // drains aggIter first. This ordering is what lets flushDirtyWrites surface a state-write
+ // failure (see its rethrow) in time to fail the task before the batch commits -- a partial
+ // write can never be committed.
var tmpRow: UnsafeRow = null
- val aggIter = CompletionIterator[UnsafeRow, Iterator[UnsafeRow]](
- baseIterator.map { row =>
- allUpdatesTimeMs += timeTakenMs {
- tmpRow = aggProcessor.process(row)
+ val processedIter: Iterator[UnsafeRow] =
+ if (groupingExpressions.isEmpty && !baseIterator.hasNext) {
+ // A global aggregation still produces its result for a batch without input, as
+ // HashAggregateExec does for the ordinary plan. The required AllTuples distribution
+ // runs this operator on a single partition, so this fires exactly once per empty batch.
+ Iterator.single(()).map { _ =>
+ allUpdatesTimeMs += timeTakenMs {
+ tmpRow = aggProcessor.processEmptyGroupingKey()
+ }
+ tmpRow
}
- tmpRow
- },
+ } else {
+ baseIterator.map { row =>
+ allUpdatesTimeMs += timeTakenMs {
+ tmpRow = aggProcessor.process(row)
+ }
+ tmpRow
+ }
+ }
+ val aggIter = CompletionIterator[UnsafeRow, Iterator[UnsafeRow]](
+ processedIter,
aggProcessor.flushDirtyWrites() // Lazily evaluated
)
@@ -481,6 +495,24 @@ class StatefulStreamlineAggregationProcessor(
existingValueRef.putRow(output.copy())
output
}
+
+ /**
+ * Merges no input into the empty grouping key of a global aggregation: the result is the stored
+ * value if one exists, otherwise the initialized buffer, which is also written to state.
+ */
+ def processEmptyGroupingKey(): UnsafeRow = {
+ val groupingKey = groupingProjection.apply(InternalRow.empty).copy()
+ val buffer = newAggregationBuffer()
+
+ val existingValueRef = dirtyWrites.get(groupingKey)
+ if (existingValueRef.getRow != null) {
+ processRow(buffer, existingValueRef.getRow)
+ }
+
+ val output = generateOutput(groupingKey, buffer)
+ existingValueRef.putRow(output.copy())
+ output
+ }
}
class UnsafeRowReference {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/RealTimeModeAllowlist.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/RealTimeModeAllowlist.scala
index 1b9ba9861f80..e8fd87274751 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/RealTimeModeAllowlist.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/RealTimeModeAllowlist.scala
@@ -66,9 +66,10 @@ object RealTimeModeAllowlist extends Logging {
"org.apache.spark.sql.execution.joins.BroadcastHashJoinExec",
// Streaming aggregation. A Real-Time Mode batch does not end when its input is exhausted, so
// an aggregation is planned as the streamline operator, which merges each input row against
- // state and emits immediately, wrapped by the two buffer-projection stages that initialize
- // the aggregation buffer and produce the result columns (see
- // AggUtils.planStreamlineStreamingAggregation).
+ // state and emits an intermediate result per input in Update mode (Append and Complete drain
+ // the input before producing their mode-specific output). It is wrapped by the two
+ // buffer-projection stages that initialize the aggregation buffer and produce the result
+ // columns (see AggUtils.planStreamlineStreamingAggregation).
"org.apache.spark.sql.execution.streaming.ProjectAggregationBufferExec",
"org.apache.spark.sql.execution.streaming.StatefulStreamlineAggregateExec",
// Streaming deduplication and the state-store access operators it plans into. These run in the
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
index 0055b1f91a7c..aab5cb6887ee 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
@@ -1095,6 +1095,183 @@ class StreamingAggregationSuite extends StateStoreMetricsTest with Assertions {
)
}
+ // A global aggregation (no grouping key) must return its initialized row for an executed batch
+ // whose rows are all dropped upstream, so the aggregation's input is empty. The micro-batch plan
+ // seeds that row for the empty grouping key in HashAggregateExec; the streamline plan must match.
+ testWithAllStateVersions("streamline aggregation: global aggregation over empty input emits " +
+ "the initialized result", streamlineEnabled) {
+ val inputData = MemoryStream[Int]
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Complete)(
+ AddData(inputData, 1, 2),
+ // Every row is filtered out: count 0, sum null.
+ CheckLastBatch((0L, null)),
+ AddData(inputData, -5),
+ CheckLastBatch((1L, -5L))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: global aggregation over empty input emits " +
+ "the initialized result in update mode", streamlineEnabled) {
+ val inputData = MemoryStream[Int]
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Update)(
+ AddData(inputData, 1, 2),
+ CheckLastBatch((0L, null)),
+ // Update mode emits an intermediate result per input row.
+ AddData(inputData, -3, -4),
+ CheckLastBatch((1L, -3L), (2L, -7L))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: complete mode retains the accumulated global " +
+ "result when a later batch has no input", streamlineEnabled) {
+ val inputData = MemoryStream[Int]
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Complete)(
+ AddData(inputData, -1, -2),
+ CheckLastBatch((2L, -3L)),
+ AddData(inputData, 3, 4),
+ // The batch has no surviving rows; the accumulated result must be re-emitted unchanged.
+ CheckLastBatch((2L, -3L))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: update mode retains the accumulated global " +
+ "result when a later batch has no input", streamlineEnabled) {
+ val inputData = MemoryStream[Int]
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Update)(
+ AddData(inputData, -1, -2),
+ CheckLastBatch((1L, -1L), (2L, -3L)),
+ AddData(inputData, 3, 4),
+ CheckLastBatch((2L, -3L))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: global aggregation with multiple partitions " +
+ "over empty input emits a single initialized row", streamlineEnabled) {
+ val inputData = MemoryStream[Int](2)
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Complete)(
+ AddData(inputData, 1, 2, 3, 4),
+ // Two partitions with no surviving rows must still produce one initialized global row.
+ CheckLastBatch((0L, null))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: multi-partition empty batch emits exactly " +
+ "one initialized row in update mode", streamlineEnabled) {
+ val inputData = MemoryStream[Int](2)
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Update)(
+ AddData(inputData, 1, 2, 3, 4),
+ // Both partitions have no surviving rows, and the batch must seed the global key exactly
+ // once, not once per empty partition.
+ CheckLastBatch((0L, null))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: an empty sibling partition does not seed " +
+ "a non-empty batch", streamlineEnabled) {
+ val inputData = MemoryStream[Int](2)
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .agg(count("*").as("count"), sum("value").as("sum"))
+
+ testStream(aggregated, Update)(
+ // Only one partition carries the surviving row; the empty sibling must not add a row.
+ AddData(inputData, -5),
+ CheckLastBatch((1L, -5L))
+ )
+ }
+
+ testWithAllStateVersions("streamline aggregation: coalesce(1) with a 0 partition RDD still " +
+ "initializes the global aggregate", streamlineEnabled) {
+ val inputSource = new BlockRDDBackedSource(spark)
+ MockSourceProvider.withMockSources(inputSource) {
+ val aggregated: Dataset[Long] =
+ spark.readStream.format((new MockSourceProvider).getClass.getCanonicalName)
+ .load().coalesce(1).groupBy().count().as[Long]
+
+ testStream(aggregated, Complete())(
+ // The very first batch is an empty trigger with a 0-partition RDD: without an
+ // initialization path for it, no state is written and Complete emits nothing.
+ AddBlockData(inputSource),
+ CheckLastBatch(0),
+ AddBlockData(inputSource, Seq(1)),
+ CheckLastBatch(1),
+ AddBlockData(inputSource), // another empty trigger with a 0-partition RDD
+ CheckLastBatch(1),
+ AddBlockData(inputSource, Seq(2, 3)),
+ CheckLastBatch(3),
+ StopStream
+ )
+ }
+ }
+
+ testWithAllStateVersions("streamline aggregation: a 0 partition source batch still initializes " +
+ "the global aggregate", streamlineEnabled) {
+ val inputSource = new BlockRDDBackedSource(spark)
+ MockSourceProvider.withMockSources(inputSource) {
+ val aggregated: Dataset[Long] =
+ spark.readStream.format((new MockSourceProvider).getClass.getCanonicalName)
+ .load().groupBy().count().as[Long]
+
+ testStream(aggregated, Complete())(
+ // The first batch has no blocks at all: the source RDD has zero partitions.
+ AddBlockData(inputSource),
+ CheckLastBatch(0),
+ AddBlockData(inputSource, Seq(1)),
+ CheckLastBatch(1),
+ StopStream
+ )
+ }
+ }
+
+ testWithAllStateVersions("streamline aggregation: grouped aggregation over empty input emits " +
+ "no rows", streamlineEnabled) {
+ val inputData = MemoryStream[Int]
+
+ val aggregated = inputData.toDF()
+ .filter($"value" < 0)
+ .groupBy($"value")
+ .agg(count("*").as("count"))
+
+ testStream(aggregated, Update)(
+ AddData(inputData, 1, 2),
+ // No group is created for an empty input.
+ CheckLastBatch(),
+ AddData(inputData, -5),
+ CheckLastBatch((-5, 1L))
+ )
+ }
+
// Append mode is the one mode that drives the operator's eviction path: a windowed grouping key
// is emitted only once the watermark passes it, via EvictionIterator (which removes in hasNext).
// Mirrors the stateStoreSave Append test above so the streamline operator is held to the same
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamlineStreamingAggregationRealTimeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamlineStreamingAggregationRealTimeSuite.scala
index 5c5409f7a7d5..d6510229e3b3 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamlineStreamingAggregationRealTimeSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamlineStreamingAggregationRealTimeSuite.scala
@@ -17,6 +17,9 @@
package org.apache.spark.sql.streaming
+import org.scalatest.time.SpanSugar._
+
+import org.apache.spark.sql.Row
import org.apache.spark.sql.execution.streaming.StatefulStreamlineAggregateExec
import org.apache.spark.sql.execution.streaming.operators.stateful.StreamingAggregationStateManager
import org.apache.spark.sql.execution.streaming.sources.{ContinuousMemorySink, LowLatencyMemoryStream}
@@ -131,4 +134,49 @@ 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}")
+ },
+ AddData(inputData, -5),
+ Execute { q =>
+ // The sink does not order rows across batches, so the checks do not rely on arrival
+ // order. Only the batch with -5 and later batches can emit the accumulated row, so a
+ // second one proves a later empty batch kept the accumulated value; a reset would emit
+ // the initialized row instead and the count would never reach two.
+ eventually(timeout(60.seconds)) {
+ assert(sink.allData.count(_ == Row(1L, -5L)) >= 2,
+ s"expected the accumulated row from a later empty batch: ${sink.allData}")
+ }
+ val emitted = sink.allData
+ assert(emitted.forall(r => r == Row(0L, null) || r == Row(1L, -5L)),
+ s"unexpected rows: $emitted")
+ val aggregates = q.lastExecution.executedPlan.collect {
+ case a: StatefulStreamlineAggregateExec => a
+ }
+ assert(aggregates.size == 1,
+ s"expected the streamline aggregate operator, got:\n${q.lastExecution.executedPlan}")
+ },
+ StopStream
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org