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 6c0a252193f8 [SPARK-56972][SS][FOLLOWUP] Reject sink evolution
combined with continuous processing
6c0a252193f8 is described below
commit 6c0a252193f8d81bd8790a67a2b0e96eee8efc9a
Author: Wenchen Fan <[email protected]>
AuthorDate: Mon Jul 20 22:43:19 2026 +0800
[SPARK-56972][SS][FOLLOWUP] Reject sink evolution combined with continuous
processing
Follow-up to #56692 (SPARK-56972), which rejected streaming sink evolution
combined with async progress tracking because the async path writes only V1
commit metadata and would silently drop the per-sink metadata that sink
evolution persists via the V3 commit log.
Continuous processing has the same gap. Sink evolution's V3 commit-log
write lives entirely in `MicroBatchExecution.markMicroBatchEnd`.
`ContinuousExecution` extends `StreamExecution` (not `MicroBatchExecution`),
and its `commit` writes a bare `CommitMetadata()` (V1). So enabling
`spark.sql.streaming.queryEvolution.enableSinkEvolution` together with
`Trigger.Continuous` runs the query while silently never persisting the sink
metadata, defeating the feature's durability guarantee with [...]
This PR rejects the combination explicitly at query start, in the
`ContinuousTrigger` match arm of `StreamingQueryManager.createQuery`, mirroring
the async guard. A new error subclass
`STREAMING_QUERY_EVOLUTION_ERROR.CONTINUOUS_PROCESSING_NOT_SUPPORTED` is added.
Correctness: a user enabling sink evolution together with continuous
processing would get no error and no persisted sink metadata. This makes the
durability gap fail loudly instead of silently, consistent with the async-path
fix.
Yes, but only for the unreleased sink-evolution feature (off by default,
and internal). A streaming query that sets
`spark.sql.streaming.queryEvolution.enableSinkEvolution=true` and uses
`Trigger.Continuous` now fails at start with a
`STREAMING_QUERY_EVOLUTION_ERROR.CONTINUOUS_PROCESSING_NOT_SUPPORTED` error
instead of running while silently not persisting the sink metadata.
Added `StreamingSinkEvolutionSuite."fail with continuous processing
enabled"`, which asserts the new validation error. Existing
`StreamingSinkEvolutionSuite` tests pass (13 tests total).
Generated-by: Claude Code (claude-opus-4-8)
This pull request and its description were written by Isaac.
Closes #57293 from cloud-fan/SPARK-56972-followup-continuous.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 6b719b39daf82a9e5cdd579992b6e3577b0b519b)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 5 +++++
.../spark/sql/classic/StreamingQueryManager.scala | 9 +++++++++
.../test/StreamingSinkEvolutionSuite.scala | 21 +++++++++++++++++++--
3 files changed, 33 insertions(+), 2 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 8a3b4969d003..f4a35bc3d527 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -7373,6 +7373,11 @@
"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."
]
},
+ "CONTINUOUS_PROCESSING_NOT_SUPPORTED" : {
+ "message" : [
+ "Streaming sink evolution cannot be used together with continuous
processing. Continuous processing persists only V1 commit metadata and would
silently drop the per-sink metadata that sink evolution requires. Disable sink
evolution by setting spark.sql.streaming.queryEvolution.enableSinkEvolution to
false, or use a micro-batch trigger instead of Trigger.Continuous."
+ ]
+ },
"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 fc21815dbf72..2a59b8f08f9e 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
@@ -226,6 +226,15 @@ class StreamingQueryManager private[sql] (
(sink, trigger) match {
case (_: SupportsWrite, trigger: ContinuousTrigger) =>
+ // Sink evolution persists per-sink metadata via the V3 commit log,
which is written only
+ // by MicroBatchExecution.markMicroBatchEnd.
ContinuousExecution.commit writes just V1
+ // commit metadata and never goes through that path, so the sink
metadata would be silently
+ // never persisted. Reject the combination explicitly instead of
silently dropping it.
+ if (sparkSession.sessionState.conf.enableStreamingSinkEvolution) {
+ throw new SparkIllegalArgumentException(
+ errorClass =
"STREAMING_QUERY_EVOLUTION_ERROR.CONTINUOUS_PROCESSING_NOT_SUPPORTED"
+ )
+ }
new StreamingQueryWrapper(new ContinuousExecution(
sparkSession,
trigger,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
index ba25d19a9110..46de58fad333 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
@@ -19,12 +19,12 @@ package org.apache.spark.sql.streaming.test
import org.scalatest.{BeforeAndAfterEach, Tag}
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, SparkIllegalArgumentException}
import org.apache.spark.sql._
import org.apache.spark.sql.execution.streaming.checkpointing.{CommitLog,
CommitMetadataV3}
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.streaming.StreamTest
+import org.apache.spark.sql.streaming.{StreamTest, Trigger}
import org.apache.spark.util.Utils
/**
@@ -147,6 +147,23 @@ class StreamingSinkEvolutionSuite extends StreamTest with
BeforeAndAfterEach {
}
}
+ testWithSinkEvolution("fail with continuous processing enabled") {
+ val input = MemoryStream[Int]
+ input.addData(1, 2, 3)
+ // 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.
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ input.toDF().writeStream
+ .format("noop")
+ .trigger(Trigger.Continuous(1000))
+ .option("checkpointLocation", newMetadataDir)
+ .start()
+ },
+ condition =
"STREAMING_QUERY_EVOLUTION_ERROR.CONTINUOUS_PROCESSING_NOT_SUPPORTED",
+ parameters = Map.empty)
+ }
+
testWithSinkEvolution("named sink succeeds with sink evolution enabled") {
val input = MemoryStream[Int]
input.addData(1, 2, 3)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
