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 087fd11379d0 [SPARK-56972][4.X][SS] Persist sink name in V3 commit log
via MicroBatchExecution
087fd11379d0 is described below
commit 087fd11379d06138c8ebff6cfc7d2e79f45098f1
Author: Eric Marnadi <[email protected]>
AuthorDate: Wed Jun 24 09:53:11 2026 -0700
[SPARK-56972][4.X][SS] Persist sink name in V3 commit log via
MicroBatchExecution
### What changes were proposed in this pull request?
Backport of [SPARK-56972]
([apache/spark#56020](https://github.com/apache/spark/pull/56020)) to
`branch-4.x`.
Wire the sink name through `MicroBatchExecution` so that, when sink
evolution is enabled, each committed batch writes a `CommitMetadataV3` whose
`sinkMetadataMap` records the current sink as the active entry alongside any
sinks that were active in earlier batches:
- Add a per-execution `sinkMetadataMap` that is hydrated from the latest
`CommitMetadataV3` in `populateStartOffsets`.
- When `spark.sql.streaming.queryEvolution.enableSinkEvolution` is true,
the commit-log write in `runBatch` produces `CommitMetadataV3` with every prior
entry marked `isActive = false` and the current `(sinkName,
sink.getClass.getName)` entered as `isActive = true`.
- When sink evolution is disabled, the existing V1/V2 commit-log path is
preserved unchanged.
This is the minimal write-then-read parity for the sink evolution feature
added in SPARK-56719. Provider-mismatch and sink-reuse validation are
intentionally deferred.
**Prerequisites.** The predecessors SPARK-56970
([apache/spark#56018](https://github.com/apache/spark/pull/56018)), SPARK-56971
([apache/spark#56019](https://github.com/apache/spark/pull/56019)), and
SPARK-56719 (the `DataStreamWriter.name()` API) are already present in
`branch-4.x`, so this is a standalone cherry-pick of `cfa759af5b6`. The only
conflict was an import-line collision in `MicroBatchExecution.scala`
(`branch-4.x` does not carry the master-only `CheckpointVersionManager`/ [...]
### Why are the changes needed?
SPARK-56719 introduced the `DataStreamWriter.name()` API and the in-memory
`sinkName` plumbing inside `MicroBatchExecution`, but the sink name was not yet
persisted to the checkpoint. Without persistence, restarts cannot observe
historical sink identity and the feature is not durable.
### Does this PR introduce _any_ user-facing change?
Behavior change only when `enableSinkEvolution` is true (off by default):
the commit log directory now contains V3 commit log files instead of V1/V2
files. Wire format compatibility is preserved when the flag is left off.
### How was this patch tested?
- Cherry-picked `cfa759af5b6` from master; resolved the single import-line
conflict in `MicroBatchExecution.scala`.
- `StreamingSinkEvolutionSuite` passes on `branch-4.x` (12 tests, including
the four new V3 commit-log cases: named-sink active entry, historical-sink
retention across rename, V1/V2 preserved when disabled, and mid-checkpoint
upgrade to V3).
- `sql/core` main and test sources compile cleanly on `branch-4.x`
(`build/sbt sql/Test/compile`).
### 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 #56706 from ericm-db/SPARK-56972-branch-4.x.
Authored-by: Eric Marnadi <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../streaming/runtime/MicroBatchExecution.scala | 46 ++++++++-
.../test/StreamingSinkEvolutionSuite.scala | 113 +++++++++++++++++++++
2 files changed, 156 insertions(+), 3 deletions(-)
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 27c4967988ec..3acabd2f5070 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
@@ -46,7 +46,7 @@ import org.apache.spark.sql.execution.{SparkPlan,
SQLExecution}
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
RealTimeStreamScanExec, StreamingDataSourceV2Relation,
StreamingDataSourceV2ScanRelation, StreamWriterCommitProgress,
WriteToDataSourceV2Exec}
import org.apache.spark.sql.execution.streaming.{AvailableNowTrigger, Offset,
OneTimeTrigger, ProcessingTimeTrigger, RealTimeModeAllowlist, RealTimeTrigger,
Sink, Source, StreamingQueryPlanTraverseHelper}
-import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
OffsetSeqBase, OffsetSeqLog, OffsetSeqMetadata, OffsetSeqMetadataV2}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
CommitLog, CommitMetadataV3, OffsetSeqBase, OffsetSeqLog, OffsetSeqMetadata,
OffsetSeqMetadataV2, SinkMetadataInfo}
import
org.apache.spark.sql.execution.streaming.operators.stateful.{StatefulOperatorStateInfo,
StatefulOpStateStoreCheckpointInfo, StateStoreWriter}
import
org.apache.spark.sql.execution.streaming.runtime.StreamingCheckpointConstants.{DIR_NAME_COMMITS,
DIR_NAME_OFFSETS, DIR_NAME_STATE}
import org.apache.spark.sql.execution.streaming.sources.{ForeachBatchSink,
WriteToMicroBatchDataSource, WriteToMicroBatchDataSourceV1}
@@ -129,6 +129,15 @@ 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]
+
+ /** True when the current query should persist V3 sink metadata in the
commit log. */
+ private def commitLogV3Enabled: Boolean =
+ sparkSession.sessionState.conf.enableStreamingSinkEvolution
+
@volatile protected[sql] var triggerExecutor: TriggerExecutor = _
protected def getTrigger(): TriggerExecutor = {
@@ -788,6 +797,11 @@ class MicroBatchExecution(
commitMetadata.stateUniqueIds.foreach {
stateUniqueIds => currentStateStoreCkptId ++= stateUniqueIds
}
+ commitMetadata match {
+ case v3: CommitMetadataV3 =>
+ sinkMetadataMap ++= v3.sinkMetadataMap
+ case _ =>
+ }
if (latestBatchId == latestCommittedBatchId) {
/* The last batch was successfully committed, so we can safely
process a
* new next batch but first:
@@ -1486,10 +1500,36 @@ class MicroBatchExecution(
} else {
None
}
- if (!commitLog.add(execCtx.batchId,
+ val metadata = if (commitLogV3Enabled) {
+ val sinkApiVersion = sink match {
+ case _: SupportsWrite => "DSv2"
+ case _ => "DSv1"
+ }
+ val currentSinkInfo = SinkMetadataInfo(
+ sinkName = sinkName,
+ commitOffset = OffsetSeqLog.SERIALIZED_VOID_OFFSET,
+ providerName = sink.getClass.getName,
+ apiVersion = sinkApiVersion,
+ isActive = true)
+ // Mark every previously-seen sink as inactive, then overlay the
current sink as active.
+ // The previous entry for [[sinkName]], if any, is overwritten here.
+ val deactivated = sinkMetadataMap.iterator
+ .map { case (name, info) => name -> info.copy(isActive = false) }
+ .toMap
+ val updatedSinkMap = deactivated + (sinkName -> currentSinkInfo)
+ sinkMetadataMap.clear()
+ sinkMetadataMap ++= updatedSinkMap
commitLog.createMetadata(
nextBatchWatermarkMs = watermarkTracker.currentWatermark,
- stateUniqueIds = stateStoreCkptId))) {
+ stateUniqueIds = stateStoreCkptId,
+ sinkMetadataMap = updatedSinkMap,
+ commitLogFormatVersion = CommitLog.VERSION_3)
+ } else {
+ commitLog.createMetadata(
+ nextBatchWatermarkMs = watermarkTracker.currentWatermark,
+ stateUniqueIds = stateStoreCkptId)
+ }
+ if (!commitLog.add(execCtx.batchId, metadata)) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
}
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 a242faabaf92..ba25d19a9110 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
@@ -21,6 +21,7 @@ import org.scalatest.{BeforeAndAfterEach, Tag}
import org.apache.spark.SparkException
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
@@ -183,6 +184,118 @@ class StreamingSinkEvolutionSuite extends StreamTest with
BeforeAndAfterEach {
q2.stop()
}
+ // ===========================
+ // Commit log V3 persistence
+ // ===========================
+
+ testWithSinkEvolution("commit log records V3 metadata with named sink") {
+ val checkpointDir = newMetadataDir
+ val input = MemoryStream[Int]
+ input.addData(1, 2, 3)
+ val q = input.toDF().writeStream
+ .format("noop")
+ .name("my_sink")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+ q.processAllAvailable()
+ q.stop()
+
+ val commitLog = new CommitLog(spark, s"$checkpointDir/commits", readOnly =
true)
+ val latest = commitLog.getLatest().getOrElse(fail("No commit recorded"))
+ val v3 = latest._2 match {
+ case v: CommitMetadataV3 => v
+ case other => fail(s"Expected CommitMetadataV3, got $other")
+ }
+ val active = v3.activeSinkMetadataInfo
+ assert(active.sinkName === "my_sink")
+ assert(active.isActive)
+ assert(v3.sinkMetadataMap.size === 1)
+ }
+
+ testWithSinkEvolution("commit log V3 retains historical sink after rename") {
+ val checkpointDir = newMetadataDir
+ val input = MemoryStream[Int]
+
+ // First batch under sink name "old_sink".
+ input.addData(1)
+ val q1 = input.toDF().writeStream
+ .format("noop")
+ .name("old_sink")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+ q1.processAllAvailable()
+ q1.stop()
+
+ // Restart with a new sink name "new_sink" against the same checkpoint.
+ input.addData(2)
+ val q2 = input.toDF().writeStream
+ .format("noop")
+ .name("new_sink")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+ q2.processAllAvailable()
+ q2.stop()
+
+ val commitLog = new CommitLog(spark, s"$checkpointDir/commits", readOnly =
true)
+ val v3 = commitLog.getLatest().get._2.asInstanceOf[CommitMetadataV3]
+ assert(v3.sinkMetadataMap.keySet === Set("old_sink", "new_sink"))
+ assert(v3.activeSinkMetadataInfo.sinkName === "new_sink")
+ assert(v3.sinkMetadataMap("old_sink").isActive === false)
+ assert(v3.sinkMetadataMap("new_sink").isActive === true)
+ }
+
+ test("commit log stays V1/V2 when sink evolution is disabled") {
+ val checkpointDir = newMetadataDir
+ withSQLConf(SQLConf.ENABLE_STREAMING_SINK_EVOLUTION.key -> "false") {
+ val input = MemoryStream[Int]
+ input.addData(1, 2)
+ val q = input.toDF().writeStream
+ .format("noop")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+ q.processAllAvailable()
+ q.stop()
+ }
+
+ val commitLog = new CommitLog(spark, s"$checkpointDir/commits", readOnly =
true)
+ val latest = commitLog.getLatest().get._2
+ assert(latest.version === CommitLog.VERSION_1 || latest.version ===
CommitLog.VERSION_2,
+ s"Expected V1 or V2 commit log, got v${latest.version}")
+ assert(!latest.isInstanceOf[CommitMetadataV3])
+ }
+
+ testWithSinkEvolution("enabling sink evolution mid-checkpoint upgrades
commit log to V3") {
+ val checkpointDir = newMetadataDir
+ val input = MemoryStream[Int]
+
+ // First run with sink evolution disabled writes V1/V2, no sink metadata.
+ withSQLConf(SQLConf.ENABLE_STREAMING_SINK_EVOLUTION.key -> "false") {
+ input.addData(1)
+ val q = input.toDF().writeStream
+ .format("noop")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+ q.processAllAvailable()
+ q.stop()
+ }
+
+ // Restart with sink evolution enabled, supplying a name. V3 should now be
written; the
+ // previous V1/V2 batches contribute no historical sinks.
+ input.addData(2)
+ val q = input.toDF().writeStream
+ .format("noop")
+ .name("upgraded_sink")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+ q.processAllAvailable()
+ q.stop()
+
+ val commitLog = new CommitLog(spark, s"$checkpointDir/commits", readOnly =
true)
+ val v3 = commitLog.getLatest().get._2.asInstanceOf[CommitMetadataV3]
+ assert(v3.activeSinkMetadataInfo.sinkName === "upgraded_sink")
+ assert(v3.sinkMetadataMap.size === 1)
+ }
+
// ==============
// Helper Methods
// ==============
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
