[SPARK-56971][SS] Add CommitMetadataV3 and SinkMetadataInfo for sink evolution

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.

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.

No. `CommitMetadataV3` is in the internal `org.apache.spark.sql.execution.streaming.checkpointing` package and is not produced by any code path yet.

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`.

Generated-by: Claude Code (claude-opus-4-7)

Closes #56019 from ericm-db/sink-evolution-sink-metadata-info.

Authored-by: ericm-db <eric.marnadi@databricks.com>
Signed-off-by: Anish Shrigondekar <anish.shrigondekar@databricks.com>
(cherry picked from commit 4d26262)
Signed-off-by: Anish Shrigondekar <anish.shrigondekar@databricks.com>

Author: ericm-db

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)
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala

@@ -21,7 +21,7 @@ import java.io.{ByteArrayInputStream, FileInputStream, FileOutputStream}
 import java.nio.file.Path
 
 import org.apache.spark.SparkFunSuite
-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
 
@@ -115,6 +115,95 @@ class CommitLogSuite extends SparkFunSuite with 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.