Skip to content

Commit 710b3c4

Browse files
committed
[SPARK-56972][SS][FOLLOWUP] Reject sink evolution combined with async progress tracking
### What changes were proposed in this pull request? Follow-up to #56020 (SPARK-56972), which added V3 commit-log persistence of sink metadata inside `MicroBatchExecution.markMicroBatchEnd`. Two issues: 1. **Async progress tracking silently drops sink metadata.** `AsyncProgressTrackingMicroBatchExecution` overrides `markMicroBatchEnd` and writes only V1 commit metadata through its async path; it never goes through the V3 write the parent added. So when both `spark.sql.streaming.queryEvolution.enableSinkEvolution` and `asyncProgressTrackingEnabled` are on, the sink metadata is silently never persisted. This PR rejects the combination explicitly at query start (mirroring the existing async validations for `Once`/`AvailableNow` triggers and unsupported sinks), so the durability gap fails loudly instead of silently. 2. **`sinkMetadataMap` doc comment was inaccurate.** It claimed insertion order is preserved "so that we can re-emit deactivated sinks in the same order they originally appeared", but the order is not upheld end-to-end: the commit-log write rebuilds the map via `.toMap` and the field round-trips through an unordered serialized `Map` on restart. The active sink is found by its `isActive` flag, not by position, so order is never consumed. The comment also referenced `runBatch` as the mutation site when the mutation actually lives in `markMicroBatchEnd`. Fixed the comment and switched the field to a plain `mutable.HashMap`. ### Why are the changes needed? The first change is a correctness fix: a user enabling sink evolution together with async progress tracking would get no error and no persisted sink metadata, defeating the feature's durability guarantee with no signal. The second keeps the in-code documentation honest so a future maintainer does not rely on an ordering guarantee that does not hold. ### Does this PR introduce _any_ user-facing change? Yes, but only for the unreleased sink-evolution feature (off by default). A streaming query that sets both `spark.sql.streaming.queryEvolution.enableSinkEvolution=true` and `asyncProgressTrackingEnabled=true` now fails at start with `IllegalArgumentException("Async progress tracking cannot be used with streaming sink evolution (spark.sql.streaming.queryEvolution.enableSinkEvolution)")` instead of running while silently not persisting the sink metadata. ### How was this patch tested? Added `AsyncProgressTrackingMicroBatchExecutionSuite."Fail with streaming sink evolution enabled"`, which asserts the new validation error. Existing `AsyncProgressTrackingMicroBatchExecutionSuite` and `StreamingSinkEvolutionSuite` (12 tests) pass with the `HashMap` change. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-8) This pull request and its description were written by Isaac. Closes #56692 from cloud-fan/SPARK-56972-followup. Authored-by: Wenchen Fan <wenchen@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
1 parent b6f2fa0 commit 710b3c4

4 files changed

Lines changed: 67 additions & 5 deletions

File tree

common/utils/src/main/resources/error/error-conditions.json

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7322,6 +7322,11 @@
73227322
"Streaming query evolution error:"
73237323
],
73247324
"subClass" : {
7325+
"ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED" : {
7326+
"message" : [
7327+
"Streaming sink evolution cannot be used together with async progress tracking. Async progress tracking persists only V1 commit metadata and would silently drop the per-sink metadata that sink evolution requires. Disable one of the two: set spark.sql.streaming.queryEvolution.enableSinkEvolution to false, or set the option asyncProgressTrackingEnabled to false."
7328+
]
7329+
},
73257330
"CANNOT_ENABLE_ON_EXISTING_CHECKPOINT" : {
73267331
"message" : [
73277332
"Cannot enable streaming source evolution on a checkpoint that was created without it. The existing checkpoint uses offset log format version <existingVersion>, which does not support the named source tracking required by streaming source evolution. To use source evolution, start the query with a fresh checkpoint."

sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,17 @@ class StreamingQueryManager private[sql] (
239239
errorClass = "STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED"
240240
)
241241
}
242+
// Sink evolution persists per-sink metadata via the V3 commit log written in
243+
// MicroBatchExecution.markMicroBatchEnd, which AsyncProgressTrackingMicroBatchExecution
244+
// overrides with an async write that only emits V1 commit metadata. The sink metadata
245+
// would therefore never be persisted, so reject the combination explicitly instead of
246+
// silently dropping it. This is checked here, before constructing the execution, so the
247+
// error is raised consistently regardless of whether the sink is named.
248+
if (sparkSession.sessionState.conf.enableStreamingSinkEvolution) {
249+
throw new SparkIllegalArgumentException(
250+
errorClass = "STREAMING_QUERY_EVOLUTION_ERROR.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED"
251+
)
252+
}
242253
new AsyncProgressTrackingMicroBatchExecution(
243254
sparkSession,
244255
trigger,

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -129,10 +129,11 @@ class MicroBatchExecution(
129129
}
130130
}
131131

132-
// Historical sink metadata read from the commit log on restart. Insertion order is preserved so
133-
// that we can re-emit deactivated sinks in the same order they originally appeared. Mutated by
134-
// [[populateStartOffsets]] (reads) and by the commit-log write in [[runBatch]] (updates).
135-
private val sinkMetadataMap = mutable.LinkedHashMap.empty[String, SinkMetadataInfo]
132+
// Historical sink metadata keyed by sink name, read from the commit log on restart. Hydrated by
133+
// [[populateStartOffsets]] from the latest CommitMetadataV3 and rewritten by the commit-log write
134+
// in [[markMicroBatchEnd]]. The active sink is identified by its isActive flag, not by position,
135+
// so the iteration order is not significant.
136+
private val sinkMetadataMap = mutable.HashMap.empty[String, SinkMetadataInfo]
136137

137138
/** True when the current query should persist V3 sink metadata in the commit log. */
138139
private def commitLogV3Enabled: Boolean =

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingMicroBatchExecutionSuite.scala

Lines changed: 46 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import org.scalatest.BeforeAndAfter
2626
import org.scalatest.matchers.should.Matchers
2727
import org.scalatest.time.{Seconds, Span}
2828

29-
import org.apache.spark.TestUtils
29+
import org.apache.spark.{SparkIllegalArgumentException, TestUtils}
3030
import org.apache.spark.sql._
3131
import org.apache.spark.sql.connector.read.streaming
3232
import org.apache.spark.sql.execution.streaming.checkpointing.{AsyncCommitLog, AsyncOffsetSeqLog, CommitMetadata, OffsetSeq}
@@ -336,6 +336,51 @@ class AsyncProgressTrackingMicroBatchExecutionSuite
336336
e.getMessage should equal("Async progress tracking cannot be used with AvailableNow trigger")
337337
}
338338

339+
test("Fail with streaming sink evolution enabled") {
340+
val inputData = new MemoryStream[Int](id = 0, spark)
341+
val ds = inputData.toDF()
342+
343+
// The sink is intentionally left unnamed: the combination is rejected before the execution is
344+
// constructed, so the error does not depend on whether name() is set.
345+
withSQLConf(SQLConf.ENABLE_STREAMING_SINK_EVOLUTION.key -> "true") {
346+
val e = intercept[SparkIllegalArgumentException] {
347+
ds.writeStream
348+
.format("noop")
349+
.option(ASYNC_PROGRESS_TRACKING_ENABLED, true)
350+
.start()
351+
}
352+
checkError(
353+
e,
354+
condition = "STREAMING_QUERY_EVOLUTION_ERROR.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED",
355+
parameters = Map.empty)
356+
}
357+
}
358+
359+
test("Succeed when only one of sink evolution / async progress tracking is enabled") {
360+
val inputData = new MemoryStream[Int](id = 0, spark)
361+
val ds = inputData.toDF()
362+
363+
// The guard must be narrow: it fires only when both configs are on. Verify that each config
364+
// on its own still starts a query, so a future broadening of the condition is caught here.
365+
366+
// Async progress tracking on, sink evolution off (default).
367+
val asyncOnly = ds.writeStream
368+
.format("noop")
369+
.option(ASYNC_PROGRESS_TRACKING_ENABLED, true)
370+
.start()
371+
try assert(asyncOnly.isActive) finally asyncOnly.stop()
372+
373+
// Sink evolution on, async progress tracking off. The sink must be named when sink evolution
374+
// is enabled, otherwise the query is rejected before execution regardless of async tracking.
375+
withSQLConf(SQLConf.ENABLE_STREAMING_SINK_EVOLUTION.key -> "true") {
376+
val evolutionOnly = ds.writeStream
377+
.format("noop")
378+
.name("evolution_only_sink")
379+
.start()
380+
try assert(evolutionOnly.isActive) finally evolutionOnly.stop()
381+
}
382+
}
383+
339384
test("switching between async wal commit enabled and trigger once") {
340385
val checkpointLocation = Utils.createTempDir(namePrefix = "streaming.metadata").getCanonicalPath
341386

0 commit comments

Comments
 (0)