This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 da117f667ed6 [SPARK-56972][SS][FOLLOWUP] Reject sink evolution
combined with async progress tracking
da117f667ed6 is described below
commit da117f667ed61667902cbd1d5e2686588fae40ce
Author: Wenchen Fan <[email protected]>
AuthorDate: Sat Jul 11 17:23:18 2026 +0800
[SPARK-56972][SS][FOLLOWUP] Reject sink evolution combined with async
progress tracking
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 a [...]
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 mu [...]
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.
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.
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.
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 <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 710b3c45aab88bd14e51d49f400e2f31e3b65772)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 5 +++
.../spark/sql/classic/StreamingQueryManager.scala | 11 +++++
.../streaming/runtime/MicroBatchExecution.scala | 9 +++--
...cProgressTrackingMicroBatchExecutionSuite.scala | 47 +++++++++++++++++++++-
4 files changed, 67 insertions(+), 5 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index c70620306840..079507803bd6 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -7327,6 +7327,11 @@
"Streaming query evolution error:"
],
"subClass" : {
+ "ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED" : {
+ "message" : [
+ "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."
+ ]
+ },
"DUPLICATE_SOURCE_NAMES" : {
"message" : [
"Duplicate streaming source names detected: <names>. Each streaming
source must have a unique name."
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
index fff8d32a0709..fc21815dbf72 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
@@ -239,6 +239,17 @@ class StreamingQueryManager private[sql] (
errorClass =
"STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED"
)
}
+ // Sink evolution persists per-sink metadata via the V3 commit log
written in
+ // MicroBatchExecution.markMicroBatchEnd, which
AsyncProgressTrackingMicroBatchExecution
+ // overrides with an async write that only emits V1 commit metadata.
The sink metadata
+ // would therefore never be persisted, so reject the combination
explicitly instead of
+ // silently dropping it. This is checked here, before constructing
the execution, so the
+ // error is raised consistently regardless of whether the sink is
named.
+ if (sparkSession.sessionState.conf.enableStreamingSinkEvolution) {
+ throw new SparkIllegalArgumentException(
+ errorClass =
"STREAMING_QUERY_EVOLUTION_ERROR.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED"
+ )
+ }
new AsyncProgressTrackingMicroBatchExecution(
sparkSession,
trigger,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
index 8da51cadf97a..f32a48c6f1c8 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
@@ -129,10 +129,11 @@ class MicroBatchExecution(
}
}
- // Historical sink metadata read from the commit log on restart. Insertion
order is preserved so
- // that we can re-emit deactivated sinks in the same order they originally
appeared. Mutated by
- // [[populateStartOffsets]] (reads) and by the commit-log write in
[[runBatch]] (updates).
- private val sinkMetadataMap = mutable.LinkedHashMap.empty[String,
SinkMetadataInfo]
+ // Historical sink metadata keyed by sink name, read from the commit log on
restart. Hydrated by
+ // [[populateStartOffsets]] from the latest CommitMetadataV3 and rewritten
by the commit-log write
+ // in [[markMicroBatchEnd]]. The active sink is identified by its isActive
flag, not by position,
+ // so the iteration order is not significant.
+ private val sinkMetadataMap = mutable.HashMap.empty[String, SinkMetadataInfo]
/** True when the current query should persist V3 sink metadata in the
commit log. */
private def commitLogV3Enabled: Boolean =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingMicroBatchExecutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingMicroBatchExecutionSuite.scala
index ba7af98b3766..9d5b5b1bfce4 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingMicroBatchExecutionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingMicroBatchExecutionSuite.scala
@@ -26,7 +26,7 @@ import org.scalatest.BeforeAndAfter
import org.scalatest.matchers.should.Matchers
import org.scalatest.time.{Seconds, Span}
-import org.apache.spark.TestUtils
+import org.apache.spark.{SparkIllegalArgumentException, TestUtils}
import org.apache.spark.sql._
import org.apache.spark.sql.connector.read.streaming
import org.apache.spark.sql.execution.streaming.checkpointing.{AsyncCommitLog,
AsyncOffsetSeqLog, CommitMetadata, OffsetSeq}
@@ -336,6 +336,51 @@ class AsyncProgressTrackingMicroBatchExecutionSuite
e.getMessage should equal("Async progress tracking cannot be used with
AvailableNow trigger")
}
+ test("Fail with streaming sink evolution enabled") {
+ val inputData = new MemoryStream[Int](id = 0, spark)
+ val ds = inputData.toDF()
+
+ // The sink is intentionally left unnamed: the combination is rejected
before the execution is
+ // constructed, so the error does not depend on whether name() is set.
+ withSQLConf(SQLConf.ENABLE_STREAMING_SINK_EVOLUTION.key -> "true") {
+ val e = intercept[SparkIllegalArgumentException] {
+ ds.writeStream
+ .format("noop")
+ .option(ASYNC_PROGRESS_TRACKING_ENABLED, true)
+ .start()
+ }
+ checkError(
+ e,
+ condition =
"STREAMING_QUERY_EVOLUTION_ERROR.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED",
+ parameters = Map.empty)
+ }
+ }
+
+ test("Succeed when only one of sink evolution / async progress tracking is
enabled") {
+ val inputData = new MemoryStream[Int](id = 0, spark)
+ val ds = inputData.toDF()
+
+ // The guard must be narrow: it fires only when both configs are on.
Verify that each config
+ // on its own still starts a query, so a future broadening of the
condition is caught here.
+
+ // Async progress tracking on, sink evolution off (default).
+ val asyncOnly = ds.writeStream
+ .format("noop")
+ .option(ASYNC_PROGRESS_TRACKING_ENABLED, true)
+ .start()
+ try assert(asyncOnly.isActive) finally asyncOnly.stop()
+
+ // Sink evolution on, async progress tracking off. The sink must be named
when sink evolution
+ // is enabled, otherwise the query is rejected before execution regardless
of async tracking.
+ withSQLConf(SQLConf.ENABLE_STREAMING_SINK_EVOLUTION.key -> "true") {
+ val evolutionOnly = ds.writeStream
+ .format("noop")
+ .name("evolution_only_sink")
+ .start()
+ try assert(evolutionOnly.isActive) finally evolutionOnly.stop()
+ }
+ }
+
test("switching between async wal commit enabled and trigger once") {
val checkpointLocation = Utils.createTempDir(namePrefix =
"streaming.metadata").getCanonicalPath
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
