This is an automated email from the ASF dual-hosted git repository.
anishshri-db 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 4d2626205413 [SPARK-56971][SS] Add CommitMetadataV3 and
SinkMetadataInfo for sink evolution
4d2626205413 is described below
commit 4d262620541385ddd7e70dfa431265be0173f285
Author: ericm-db <[email protected]>
AuthorDate: Mon Jun 15 20:57:58 2026 -0700
[SPARK-56971][SS] Add CommitMetadataV3 and SinkMetadataInfo for sink
evolution
### What changes were proposed in this pull request?
Add the commit log data structures for streaming sink evolution:
- `CommitMetadataV3` (`VERSION_3` of the commit log wire format) carries a
`sinkMetadataMap: Map[String, SinkMetadataInfo]` keyed by sink name, in
addition to the V2 fields (`nextBatchWatermarkMs`, `stateUniqueIds`).
- `SinkMetadataInfo` records per-sink metadata: `sinkName`, `commitOffset`
(serialized via `OffsetV2.json()`), `providerName`, and an `isActive` flag used
to distinguish the current sink from historical sinks that were used in earlier
batches but are no longer in use.
- `CommitMetadataV3.activeSinkMetadataInfoOpt` returns the entry with
`isActive = true`, if any.
- `CommitLog.createMetadata` learns to produce a `CommitMetadataV3` when
`commitLogFormatVersion = VERSION_3`, requiring a non-empty `sinkMetadataMap`.
- `CommitLog.readCommitMetadata` dispatches `v3` files to the new class.
The V3 metadata is dormant in this PR: no caller produces it yet. Wiring
through `MicroBatchExecution` (so each batch persists its sink name + offset,
and so restarts read the map back) is the SPARK-56972 follow-up.
This PR is built on top of #56018 (SPARK-56970). It currently shows the
SPARK-56970 commits in its diff; that will resolve once SPARK-56970 merges.
### Why are the changes needed?
SPARK-56719 added `DataStreamWriter.name()` as the API surface for sink
evolution. Without a place in the commit log to durably record the sink name
and offset alongside the rest of a committed batch's metadata, sink names
cannot be observed on restart and the evolution feature cannot be completed.
This PR introduces that storage in a separate, narrowly scoped change.
### Does this PR introduce _any_ user-facing change?
No. `CommitMetadataV3` is in the internal
`org.apache.spark.sql.execution.streaming.checkpointing` package and is not
produced by any code path yet.
### How was this patch tested?
Added unit tests in `CommitLogSuite`:
- V3 SerDe with a single active sink (round-trips through commit log).
- V3 retains historical sinks alongside the active one and
`activeSinkMetadataInfoOpt` resolves correctly.
- `createMetadata(version = V3, sinkMetadataMap = Map.empty)` fails fast
with `IllegalArgumentException`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-4-7)
Closes #56019 from ericm-db/sink-evolution-sink-metadata-info.
Authored-by: ericm-db <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>
---
.../streaming/checkpointing/CommitLog.scala | 86 +++++++++++++++++++-
.../spark/sql/streaming/CommitLogSuite.scala | 91 +++++++++++++++++++++-
2 files changed, 175 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CommitLog.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CommitLog.scala
index 820aecf70d0e..b5271f664cd7 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CommitLog.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CommitLog.scala
@@ -26,6 +26,7 @@ import org.json4s.{Formats, NoTypeHints}
import org.json4s.jackson.Serialization
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.read.streaming.{Offset => OffsetV2}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
@@ -76,12 +77,18 @@ class CommitLog(
/**
* Factory for creating a [[CommitMetadataBase]] for the requested wire
format version.
* Defaults to the version configured via
[[SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION]].
+ *
+ * For [[VERSION_3]], [[sinkMetadataMap]] must be non-empty and contain
exactly one active
+ * sink; [[CommitMetadataV3]] enforces this invariant.
*/
def createMetadata(
nextBatchWatermarkMs: Long = 0,
stateUniqueIds: Option[Map[Long, Array[Array[String]]]] = None,
+ sinkMetadataMap: Map[String, SinkMetadataInfo] = Map.empty,
commitLogFormatVersion: Int = defaultVersion): CommitMetadataBase = {
commitLogFormatVersion match {
+ case VERSION_3 =>
+ CommitMetadataV3(nextBatchWatermarkMs, stateUniqueIds, sinkMetadataMap)
case VERSION_2 =>
CommitMetadataV2(nextBatchWatermarkMs, stateUniqueIds)
case VERSION_1 =>
@@ -98,7 +105,8 @@ object CommitLog {
private val EMPTY_JSON = "{}"
val VERSION_1 = 1
val VERSION_2 = 2
- val MAX_VERSION: Int = VERSION_2
+ val VERSION_3 = 3
+ val MAX_VERSION: Int = VERSION_3
/**
* Reads a single commit log entry and dispatches to the matching
@@ -112,6 +120,7 @@ object CommitLog {
val version = MetadataVersionUtil.validateVersion(lines.next().trim,
MAX_VERSION)
val metadataJson = if (lines.hasNext) lines.next() else EMPTY_JSON
version match {
+ case VERSION_3 => CommitMetadataV3(metadataJson)
case VERSION_2 => CommitMetadataV2(metadataJson)
case VERSION_1 => CommitMetadata(metadataJson)
case v => throw QueryExecutionErrors.logVersionGreaterThanSupported(v,
MAX_VERSION)
@@ -205,3 +214,78 @@ object CommitMetadataV2 {
def apply(json: String): CommitMetadataV2 =
Serialization.read[CommitMetadataV2](json)
}
+
+/**
+ * Commit log metadata for [[CommitLog.VERSION_3]]. Extends V2 with a map of
per-sink metadata
+ * keyed by sink name. This enables streaming sink evolution: each batch
records the active sink
+ * along with any historical sinks that were used in earlier batches but are
no longer active.
+ *
+ * @param nextBatchWatermarkMs The watermark of the next batch.
+ * @param stateUniqueIds Per-operator state store unique ids (see
[[CommitMetadataV2]]).
+ * @param sinkMetadataMap Non-empty map keyed by sink name with exactly one
active entry per
+ * commit; deactivated sinks are retained to detect
reuse of a sink name.
+ */
+case class CommitMetadataV3(
+ nextBatchWatermarkMs: Long = 0,
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]] = None,
+ sinkMetadataMap: Map[String, SinkMetadataInfo]) extends CommitMetadataBase
{
+ require(sinkMetadataMap.nonEmpty,
+ "VERSION_3 commit log requires a non-empty sinkMetadataMap")
+ require(sinkMetadataMap.values.count(_.isActive) == 1,
+ "VERSION_3 commit log requires exactly one active sink, but found " +
+ s"${sinkMetadataMap.values.count(_.isActive)} in sinkMetadataMap")
+
+ override def version: Int = CommitLog.VERSION_3
+
+ override def withStateUniqueIds(
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]]):
CommitMetadataV3 =
+ copy(stateUniqueIds = stateUniqueIds)
+
+ /** Returns the currently active sink's metadata; exactly one always exists
(see require). */
+ def activeSinkMetadataInfo: SinkMetadataInfo =
sinkMetadataMap.values.find(_.isActive).get
+}
+
+object CommitMetadataV3 {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ def apply(json: String): CommitMetadataV3 =
Serialization.read[CommitMetadataV3](json)
+}
+
+/**
+ * Per-sink metadata recorded in a [[CommitMetadataV3]] entry.
+ *
+ * @param sinkName Sink name as supplied via `DataStreamWriter.name()`, or
+ * `MicroBatchExecution.DEFAULT_SINK_NAME` when sink evolution
is disabled.
+ * @param commitOffset The latest offset committed to the sink as a JSON string
+ * (i.e. [[OffsetV2.json()]]), or
[[OffsetSeqLog.SERIALIZED_VOID_OFFSET]] if
+ * no offset is available.
+ * @param providerName Identifies the sink implementation (e.g.
fully-qualified class name).
+ * @param apiVersion The API version for the sink - whether it is DSv1 or DSv2.
+ * @param isActive Whether this sink is the active sink for the current batch.
Historical sinks
+ * are retained with `isActive = false`.
+ */
+case class SinkMetadataInfo(
+ sinkName: String,
+ commitOffset: String,
+ providerName: String,
+ apiVersion: String,
+ isActive: Boolean = true) {
+ def json: String = Serialization.write(this)(SinkMetadataInfo.format)
+}
+
+object SinkMetadataInfo {
+ private implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ def apply(
+ sinkName: String,
+ commitOffset: Option[OffsetV2],
+ providerName: String,
+ apiVersion: String,
+ isActive: Boolean): SinkMetadataInfo = {
+ val offsetString = commitOffset match {
+ case Some(off) => off.json
+ case None => OffsetSeqLog.SERIALIZED_VOID_OFFSET
+ }
+ new SinkMetadataInfo(sinkName, offsetString, providerName, apiVersion,
isActive)
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
index e4becd057168..ba785aa830ab 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.streaming
import java.io.{ByteArrayInputStream, FileInputStream, FileOutputStream}
import java.nio.file.Path
-import org.apache.spark.sql.execution.streaming.checkpointing.{CommitLog,
CommitMetadata, CommitMetadataBase, CommitMetadataV2}
+import org.apache.spark.sql.execution.streaming.checkpointing.{CommitLog,
CommitMetadata, CommitMetadataBase, CommitMetadataV2, CommitMetadataV3,
OffsetSeqLog, SinkMetadataInfo}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -114,6 +114,95 @@ class CommitLogSuite extends SharedSparkSession {
}
}
+ test("Basic Commit Log V3 SerDe - single active sink") {
+ withTempDir { tempDir =>
+ val commitLog = new CommitLog(spark, tempDir.getAbsolutePath)
+ val sinkInfo = SinkMetadataInfo(
+ sinkName = "sink-0",
+ commitOffset = OffsetSeqLog.SERIALIZED_VOID_OFFSET,
+ providerName = "memory",
+ apiVersion = "v2",
+ isActive = true)
+ val metadata = commitLog.createMetadata(
+ nextBatchWatermarkMs = 42,
+ sinkMetadataMap = Map("sink-0" -> sinkInfo),
+ commitLogFormatVersion = CommitLog.VERSION_3)
+ assert(commitLog.add(0, metadata))
+
+ val read = commitLog.get(0).get
+ assert(read.version === CommitLog.VERSION_3)
+ assert(read.nextBatchWatermarkMs === 42)
+ val readV3 = read.asInstanceOf[CommitMetadataV3]
+ assert(readV3.sinkMetadataMap === Map("sink-0" -> sinkInfo))
+ assert(readV3.activeSinkMetadataInfo === sinkInfo)
+ }
+ }
+
+ test("Commit Log V3 - retains historical sinks alongside active") {
+ withTempDir { tempDir =>
+ val commitLog = new CommitLog(spark, tempDir.getAbsolutePath)
+ val historical = SinkMetadataInfo(
+ sinkName = "sink-0",
+ commitOffset = """{"offset":3}""",
+ providerName = "memory",
+ apiVersion = "v2",
+ isActive = false)
+ val active = SinkMetadataInfo(
+ sinkName = "sink-1",
+ commitOffset = """{"offset":7}""",
+ providerName = "memory",
+ apiVersion = "v2",
+ isActive = true)
+ val metadata = commitLog.createMetadata(
+ nextBatchWatermarkMs = 100,
+ sinkMetadataMap = Map("sink-0" -> historical, "sink-1" -> active),
+ commitLogFormatVersion = CommitLog.VERSION_3)
+ assert(commitLog.add(0, metadata))
+
+ val readV3 = commitLog.get(0).get.asInstanceOf[CommitMetadataV3]
+ assert(readV3.activeSinkMetadataInfo === active)
+ assert(readV3.sinkMetadataMap("sink-0") === historical)
+ assert(readV3.sinkMetadataMap("sink-1") === active)
+ }
+ }
+
+ test("createMetadata for V3 requires non-empty sinkMetadataMap") {
+ withTempDir { tempDir =>
+ val commitLog = new CommitLog(spark, tempDir.getAbsolutePath)
+ intercept[IllegalArgumentException] {
+ commitLog.createMetadata(
+ nextBatchWatermarkMs = 0,
+ sinkMetadataMap = Map.empty,
+ commitLogFormatVersion = CommitLog.VERSION_3)
+ }
+ }
+ }
+
+ test("CommitMetadataV3 requires exactly one active sink") {
+ val historical = SinkMetadataInfo(
+ sinkName = "sink-0",
+ commitOffset = OffsetSeqLog.SERIALIZED_VOID_OFFSET,
+ providerName = "memory",
+ apiVersion = "v2",
+ isActive = false)
+ val active = SinkMetadataInfo(
+ sinkName = "sink-1",
+ commitOffset = OffsetSeqLog.SERIALIZED_VOID_OFFSET,
+ providerName = "memory",
+ apiVersion = "v2",
+ isActive = true)
+
+ // No active sink.
+ intercept[IllegalArgumentException] {
+ CommitMetadataV3(sinkMetadataMap = Map("sink-0" -> historical))
+ }
+ // More than one active sink.
+ intercept[IllegalArgumentException] {
+ CommitMetadataV3(sinkMetadataMap =
+ Map("sink-0" -> active.copy(sinkName = "sink-0"), "sink-1" -> active))
+ }
+ }
+
// SPARK-50653: When the configured commit log version is V2, a V1 file on
disk should still
// deserialize successfully into a V1 [[CommitMetadata]] because the wire
format version is now
// discovered from the file header rather than enforced to match the conf.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
