This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6b719b39daf8 [SPARK-56972][SS][FOLLOWUP] Reject sink evolution
combined with continuous processing
6b719b39daf8 is described below
commit 6b719b39daf82a9e5cdd579992b6e3577b0b519b
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
### What changes were proposed in this pull request?
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.
### Why are the changes needed?
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.
### Does this PR introduce _any_ user-facing change?
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.
### How was this patch tested?
Added `StreamingSinkEvolutionSuite."fail with continuous processing
enabled"`, which asserts the new validation error. Existing
`StreamingSinkEvolutionSuite` tests pass (13 tests total).
### 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 #57293 from cloud-fan/SPARK-56972-followup-continuous.
Authored-by: Wenchen Fan <[email protected]>
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 6b38eb1c23e3..a72e279f25fb 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 @@
"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."
]
},
+ "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]
