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 <[email protected]>
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 <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
(cherry picked from commit 2f4102c8117a087aba1324d52b268f7966670136)
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
