cloud-fan commented on code in PR #57945:
URL: https://github.com/apache/spark/pull/57945#discussion_r3776345022
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala:
##########
@@ -684,6 +684,110 @@ class StreamingDeduplicationSuite extends
StateStoreMetricsTest
sqlConf = spark.sessionState.conf
)
}
+
+ // Total incremental removals reported across all batches, read from the
operator's
+ // numRowsIncrementallyRemoved custom metric. Where a stateful operator ran
(stateOperators is
+ // non-empty) the metric must be present -- assert rather than default it to
0, so a regression
+ // that stops emitting the metric fails the test instead of silently reading
as zero.
+ private def numRowsIncrementallyRemoved(q: StreamingQuery): Long = {
+ q.recentProgress.flatMap(_.stateOperators.headOption).map { op =>
+ assert(op.customMetrics.containsKey("numRowsIncrementallyRemoved"),
+ s"numRowsIncrementallyRemoved custom metric missing; got
${op.customMetrics}")
+ op.customMetrics.get("numRowsIncrementallyRemoved").toLong
+ }.sum
+ }
+
+ // Total state rows removed across all batches, read from the operator's
first-class
+ // numRowsRemoved metric (the incremental removals are a subset of these).
This asserts that
+ // eviction actually removed state, which numRowsIncrementallyRemoved alone
does not: a batch-end
+ // drain removes rows without incrementing the incremental counter.
+ private def totalStateRowsRemoved(q: StreamingQuery): Long =
+
q.recentProgress.flatMap(_.stateOperators.headOption.map(_.numRowsRemoved)).sum
+
+ test("deduplicate with watermark - incremental cleanup preserves dedup
output") {
+ // With a non-zero incremental cleanup factor, watermark-expired state is
removed spread across
+ // input-record processing rather than all at once at batch end. The
deduplicated OUTPUT must be
+ // unchanged. (State is evicted against the late-events watermark under
incremental cleanup, so
+ // the timing of state removal lags by a batch compared to the factor-0
default -- this test
+ // asserts on output and on the late-event safety property, not on
per-batch state counts.)
+ withSQLConf(SQLConf.STREAMING_STATE_INCREMENTAL_CLEANUP_FACTOR.key ->
"10") {
+ val inputData = MemoryStream[Int]
+ val result = inputData.toDS()
+ .withColumn("eventTime", timestamp_seconds($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .dropDuplicates()
+ .select($"eventTime".cast("long").as[Long])
+
+ testStream(result, Append)(
+ // Duplicates within the batch are dropped; each distinct event time
is emitted once.
+ AddData(inputData, (1 to 5).flatMap(_ => (10 to 15)): _*),
+ CheckAnswer(10 to 15: _*),
+
+ AddData(inputData, 25), // Advances watermark; 25 is new and emitted.
+ CheckNewAnswer(25),
+
+ // A record at 10 is now below the watermark: it must be dropped, and
crucially it must not
+ // be re-emitted even though incremental cleanup may not yet have
removed its key. This is
+ // the safety property behind evicting against the late-events (not
eviction) watermark.
+ AddData(inputData, 10),
+ CheckNewAnswer(),
+
+ AddData(inputData, 45),
+ CheckNewAnswer(45),
+
+ // A duplicate of a surviving recent key (45) is still deduplicated.
+ AddData(inputData, 45),
+ CheckNewAnswer()
+ )
+ }
+ }
+
+ test("deduplicate with watermark - incremental cleanup evicts during record
processing") {
+ // A batch that carries input records AND has state eligible for eviction
under the late-events
+ // watermark should evict incrementally as those records are processed, so
+ // numRowsIncrementallyRemoved is non-zero. (Eviction uses the late-events
watermark under
+ // incremental cleanup, which lags the eviction watermark by one batch.)
+ withSQLConf(SQLConf.STREAMING_STATE_INCREMENTAL_CLEANUP_FACTOR.key ->
"10") {
+ val inputData = MemoryStream[Int]
+ val result = inputData.toDS()
+ .withColumn("eventTime", timestamp_seconds($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .dropDuplicates()
+ .select($"eventTime".cast("long").as[Long])
+
+ testStream(result, Append)(
+ // Batch 0: three distinct keys at 10, 11, 12. Watermark is 0, nothing
evictable yet.
+ AddData(inputData, 10, 11, 12),
+ CheckAnswer(10, 11, 12),
+ assertNumStateRows(total = 3, updated = 3),
+ AssertOnQuery(q => numRowsIncrementallyRemoved(q) == 0,
+ "no incremental removal before any state is evictable"),
+
+ // Batch 1: a new key at 100 advances the eviction watermark to 90,
but the late-events
+ // watermark used by incremental cleanup becomes 0 -> then this
batch's own late-events
+ // watermark lags, so removal of the [10,12] keys happens as batch 1's
record is processed
Review Comment:
**Nit:**
The late-events watermark still lags in batch 1, and the assertion is
intentionally deferred until batch 2 processes `101`. Please say removal does
not happen until batch 2, after the keys fall below the late-events watermark.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala:
##########
@@ -684,6 +684,110 @@ class StreamingDeduplicationSuite extends
StateStoreMetricsTest
sqlConf = spark.sessionState.conf
)
}
+
+ // Total incremental removals reported across all batches, read from the
operator's
+ // numRowsIncrementallyRemoved custom metric. Where a stateful operator ran
(stateOperators is
+ // non-empty) the metric must be present -- assert rather than default it to
0, so a regression
+ // that stops emitting the metric fails the test instead of silently reading
as zero.
+ private def numRowsIncrementallyRemoved(q: StreamingQuery): Long = {
+ q.recentProgress.flatMap(_.stateOperators.headOption).map { op =>
+ assert(op.customMetrics.containsKey("numRowsIncrementallyRemoved"),
+ s"numRowsIncrementallyRemoved custom metric missing; got
${op.customMetrics}")
+ op.customMetrics.get("numRowsIncrementallyRemoved").toLong
+ }.sum
+ }
+
+ // Total state rows removed across all batches, read from the operator's
first-class
+ // numRowsRemoved metric (the incremental removals are a subset of these).
This asserts that
+ // eviction actually removed state, which numRowsIncrementallyRemoved alone
does not: a batch-end
+ // drain removes rows without incrementing the incremental counter.
+ private def totalStateRowsRemoved(q: StreamingQuery): Long =
+
q.recentProgress.flatMap(_.stateOperators.headOption.map(_.numRowsRemoved)).sum
+
+ test("deduplicate with watermark - incremental cleanup preserves dedup
output") {
+ // With a non-zero incremental cleanup factor, watermark-expired state is
removed spread across
Review Comment:
**Nit:**
`state is removed spread` is grammatically broken. Please rewrite this as
`removal of watermark-expired state is spread across input-record processing
rather than occurring all at once at batch end`.
--
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]
