This is an automated email from the ASF dual-hosted git repository.
anishshri-db pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new bfcc62bdb0fc [SPARK-56971][SS] Add CommitMetadataV3 and
SinkMetadataInfo for sink evolution
bfcc62bdb0fc is described below
commit bfcc62bdb0fc29634b19ab5951716bd1fb436d67
Author: Eric Marnadi <[email protected]>
AuthorDate: Tue Jun 30 15:50:44 2026 -0700
[SPARK-56971][SS] Add CommitMetadataV3 and SinkMetadataInfo for sink
evolution
### What changes were proposed in this pull request?
Backport of [SPARK-56971]
([apache/spark#56019](https://github.com/apache/spark/pull/56019)) to
`branch-4.2`.
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`, `apiVersion`, 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.activeSinkMetadataInfo` returns the entry with
`isActive = true`; `CommitMetadataV3` requires exactly one active sink.
- `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` is the SPARK-56972 follow-up.
**Prerequisite commit.** SPARK-56971 was built on top of [SPARK-56970]
([apache/spark#56018](https://github.com/apache/spark/pull/56018)), which
splits `CommitMetadata` into a `CommitMetadataBase` trait with concrete
`CommitMetadata` (V1) and `CommitMetadataV2` case classes. `branch-4.2` does
not yet have SPARK-56970, so this PR includes it as the first commit and adds
SPARK-56971 on top. Both commits are cherry-picked from the `branch-4.x`
backports (`5322ec30c02` and `706ce2f3743`). [...]
### 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 the 4.2 release line.
### 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. As part of the SPARK-56970 refactor, V1 commit
log files no longer serialize `stateUniqueIds: null`; old V1 files continue to
be read because the V1 deserializer ignores the (now-unknown) field.
### How was this patch tested?
- Cherry-picked the two `branch-4.x` commits; resolved import conflicts in
`CommitLogSuite.scala`.
- Existing and new `CommitLogSuite` cases (V1/V2/V3 SerDe, historical-sink
retention, `createMetadata` V3 empty-map failure, exactly-one-active-sink
invariant).
- `sql/core` main and test sources compile cleanly on `branch-4.2`
(`build/sbt sql/Test/compile`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-4-8)
Closes #56548 from ericm-db/SPARK-56971-branch-4.2.
Lead-authored-by: Eric Marnadi <[email protected]>
Co-authored-by: ericm-db <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>
---
.../streaming/checkpointing/AsyncCommitLog.scala | 4 +-
.../streaming/checkpointing/CommitLog.scala | 211 +++++++++++++++++++--
.../streaming/runtime/MicroBatchExecution.scala | 6 +-
.../state/OfflineStateRepartitionRunner.scala | 4 +-
.../execution/streaming/state/StateRewriter.scala | 34 ++--
.../state/StateDataSourceChangeDataReadSuite.scala | 6 +-
.../v2/state/StateDataSourceReadSuite.scala | 77 +++++---
...tatePartitionAllColumnFamiliesWriterSuite.scala | 2 +-
.../spark/sql/streaming/CommitLogSuite.scala | 142 ++++++++++++--
9 files changed, 393 insertions(+), 93 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/AsyncCommitLog.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/AsyncCommitLog.scala
index 116ea18326ef..0f031fcbb951 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/AsyncCommitLog.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/AsyncCommitLog.scala
@@ -48,7 +48,7 @@ class AsyncCommitLog(sparkSession: SparkSession, path:
String, executorService:
* the async write of the batch is completed. Future may also be
completed exceptionally
* to indicate some write error.
*/
- def addAsync(batchId: Long, metadata: CommitMetadata):
CompletableFuture[Long] = {
+ def addAsync(batchId: Long, metadata: CommitMetadataBase):
CompletableFuture[Long] = {
require(metadata != null, "'null' metadata cannot be written to a metadata
log")
val future: CompletableFuture[Long] = addNewBatchByStreamAsync(batchId) {
output =>
serialize(metadata, output)
@@ -72,7 +72,7 @@ class AsyncCommitLog(sparkSession: SparkSession, path:
String, executorService:
* @param metadata metadata of batch to write
* @return true if operation is successful otherwise false.
*/
- def addInMemory(batchId: Long, metadata: CommitMetadata): Boolean = {
+ def addInMemory(batchId: Long, metadata: CommitMetadataBase): Boolean = {
if (batchCache.containsKey(batchId)) {
false
} else {
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 b73020b6060c..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,8 @@ 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
/**
@@ -50,39 +52,127 @@ class CommitLog(
sparkSession: SparkSession,
path: String,
readOnly: Boolean = false)
- extends HDFSMetadataLog[CommitMetadata](sparkSession, path, readOnly) {
+ extends HDFSMetadataLog[CommitMetadataBase](sparkSession, path, readOnly) {
import CommitLog._
- private val VERSION: Int = sparkSession.conf.get(
+ // The configured commit log format version. Used as the default version
when callers
+ // construct metadata through [[createMetadata]].
+ private[sql] val defaultVersion: Int = sparkSession.conf.get(
SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).toInt
- override protected[sql] def deserialize(in: InputStream): CommitMetadata = {
- // called inside a try-finally where the underlying stream is closed in
the caller
- val lines = IOSource.fromInputStream(in, UTF_8.name()).getLines()
- if (!lines.hasNext) {
- throw new IllegalStateException("Incomplete log file in the offset
commit log")
- }
- // TODO [SPARK-49462] This validation should be relaxed for a stateless
query.
- // TODO [SPARK-50653] This validation should be relaxed to support reading
- // a V1 log file when VERSION is V2
- validateVersionExactMatch(lines.next().trim, VERSION)
- val metadataJson = if (lines.hasNext) lines.next() else EMPTY_JSON
- CommitMetadata(metadataJson)
+ override protected[sql] def deserialize(in: InputStream): CommitMetadataBase
= {
+ CommitLog.readCommitMetadata(in)
}
- override protected[sql] def serialize(metadata: CommitMetadata, out:
OutputStream): Unit = {
+ override protected[sql] def serialize(metadata: CommitMetadataBase, out:
OutputStream): Unit = {
// called inside a try-finally where the underlying stream is closed in
the caller
- out.write(s"v${VERSION}".getBytes(UTF_8))
+ out.write(s"v${metadata.version}".getBytes(UTF_8))
out.write('\n')
// write metadata
out.write(metadata.json.getBytes(UTF_8))
}
+
+ /**
+ * 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 =>
+ // VERSION_1 cannot persist stateUniqueIds; withStateUniqueIds
enforces this invariant
+ // (it throws if stateUniqueIds is non-empty).
+ CommitMetadata(nextBatchWatermarkMs).withStateUniqueIds(stateUniqueIds)
+ case v =>
+ throw QueryExecutionErrors.logVersionGreaterThanSupported(v,
CommitLog.MAX_VERSION)
+ }
+ }
}
object CommitLog {
private val EMPTY_JSON = "{}"
+ val VERSION_1 = 1
+ val VERSION_2 = 2
+ val VERSION_3 = 3
+ val MAX_VERSION: Int = VERSION_3
+
+ /**
+ * Reads a single commit log entry and dispatches to the matching
+ * [[CommitMetadataBase]] subclass based on the wire format version recorded
in the file.
+ */
+ private[spark] def readCommitMetadata(in: InputStream): CommitMetadataBase =
{
+ val lines = IOSource.fromInputStream(in, UTF_8.name()).getLines()
+ if (!lines.hasNext) {
+ throw new IllegalStateException("Incomplete log file in the offset
commit log")
+ }
+ 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)
+ }
+ }
+}
+
+/**
+ * Base trait for commit log metadata. Concrete subclasses correspond to wire
format versions
+ * and override [[version]] accordingly.
+ */
+trait CommitMetadataBase extends Serializable {
+ def version: Int
+ def nextBatchWatermarkMs: Long
+ def stateUniqueIds: Option[Map[Long, Array[Array[String]]]]
+
+ /**
+ * Returns a copy of this metadata with the given state store unique ids,
preserving the
+ * concrete subclass and all of its other fields. Deriving a new commit from
an existing one
+ * should go through this method (rather than reconstructing via
[[CommitLog.createMetadata]])
+ * so that version-specific fields are not silently dropped when new
metadata versions are
+ * introduced.
+ */
+ def withStateUniqueIds(
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]]):
CommitMetadataBase
+
+ def json: String = Serialization.write(this)(CommitMetadata.format)
+}
+
+/**
+ * Commit log metadata for [[CommitLog.VERSION_1]]. Records the watermark for
the next batch only.
+ *
+ * @param nextBatchWatermarkMs The watermark of the next batch.
+ */
+case class CommitMetadata(
+ nextBatchWatermarkMs: Long = 0) extends CommitMetadataBase {
+ override def version: Int = CommitLog.VERSION_1
+ override def stateUniqueIds: Option[Map[Long, Array[Array[String]]]] = None
+
+ override def withStateUniqueIds(
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]]): CommitMetadata
= {
+ require(stateUniqueIds.forall(_.isEmpty),
+ s"stateUniqueIds cannot be set for commit log format version
${CommitLog.VERSION_1}; " +
+ s"use version ${CommitLog.VERSION_2} to persist state store checkpoint
ids.")
+ this
+ }
+}
+
+object CommitMetadata {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ def apply(json: String): CommitMetadata =
Serialization.read[CommitMetadata](json)
}
/**
@@ -104,19 +194,98 @@ object CommitLog {
* +--- ......
* In the commit log, in addition to nextBatchWatermarkMs, we also store the
unique ids of the
* state store files.
+ *
* @param nextBatchWatermarkMs The watermark of the next batch.
* @param stateUniqueIds Map[Long, Array[Array[String]]] of map
* OperatorId -> (partitionID -> array of uniqueID)
*/
+case class CommitMetadataV2(
+ nextBatchWatermarkMs: Long = 0,
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]] = None) extends
CommitMetadataBase {
+ override def version: Int = CommitLog.VERSION_2
-case class CommitMetadata(
+ override def withStateUniqueIds(
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]]):
CommitMetadataV2 =
+ copy(stateUniqueIds = stateUniqueIds)
+}
+
+object CommitMetadataV2 {
+ import CommitMetadata.format
+
+ 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) {
- def json: String = Serialization.write(this)(CommitMetadata.format)
+ 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 CommitMetadata {
+object CommitMetadataV3 {
implicit val format: Formats = Serialization.formats(NoTypeHints)
- def apply(json: String): CommitMetadata =
Serialization.read[CommitMetadata](json)
+ 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/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 c06015a59264..84eb9bdc24dc 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, RealTimeTrigger, Sink, Source,
StreamingQueryPlanTraverseHelper}
-import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
CommitMetadata, OffsetSeqBase, OffsetSeqLog, OffsetSeqMetadata,
OffsetSeqMetadataV2}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
OffsetSeqBase, OffsetSeqLog, OffsetSeqMetadata, OffsetSeqMetadataV2}
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}
@@ -1464,7 +1464,9 @@ class MicroBatchExecution(
None
}
if (!commitLog.add(execCtx.batchId,
- CommitMetadata(watermarkTracker.currentWatermark, stateStoreCkptId))) {
+ commitLog.createMetadata(
+ nextBatchWatermarkMs = watermarkTracker.currentWatermark,
+ stateUniqueIds = stateStoreCkptId))) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala
index 1491d2698906..dc13fa1030a0 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/OfflineStateRepartitionRunner.scala
@@ -294,7 +294,9 @@ class OfflineStateRepartitionRunner(
lastCommittedBatchId: Long,
opIdToStateStoreCkptInfo: Option[Map[Long, Array[Array[String]]]]): Unit
= {
val latestCommit =
checkpointMetadata.commitLog.get(lastCommittedBatchId).get
- val commitMetadata = latestCommit.copy(stateUniqueIds =
opIdToStateStoreCkptInfo)
+ // Derive the new commit from the latest one so version-specific fields
are preserved and the
+ // wire format version stays consistent with the source checkpoint.
+ val commitMetadata =
latestCommit.withStateUniqueIds(opIdToStateStoreCkptInfo)
if (!checkpointMetadata.commitLog.add(newBatchId, commitMetadata)) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(newBatchId)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala
index fd890161caaf..546a9a601964 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateRewriter.scala
@@ -22,7 +22,7 @@ import java.util.UUID
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
-import org.apache.spark.{SparkIllegalStateException, SparkThrowable,
TaskContext}
+import org.apache.spark.{SparkIllegalStateException, TaskContext}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
@@ -376,27 +376,19 @@ class StateRewriter(
}
private def verifyCheckpointFormatVersion(): Unit = {
- // Verify checkpoint version in sqlConf based on commitLog for
readCheckpoint
- // in case user forgot to set STATE_STORE_CHECKPOINT_FORMAT_VERSION.
- // Using read batch commit since the latest commit could be a skipped
batch.
- // If SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION is wrong,
readCheckpoint.commitLog
- // will throw an exception, and we will propagate this exception upstream.
- // This prevents the StateRewriter from failing to write the correct state
files
- try {
- readCheckpoint.commitLog.get(readBatchId)
- } catch {
- case e: IllegalStateException if e.getCause != null &&
- e.getCause.isInstanceOf[SparkThrowable] =>
- val sparkThrowable = e.getCause.asInstanceOf[SparkThrowable]
- if (sparkThrowable.getCondition ==
"INVALID_LOG_VERSION.EXACT_MATCH_VERSION") {
- val params = sparkThrowable.getMessageParameters
- val expectedVersion = params.get("version")
- val actualVersion = params.get("matchVersion")
- throw
StateRewriterErrors.stateCheckpointFormatVersionMismatchError(
- checkpointLocationForRead, expectedVersion, actualVersion)
- }
- throw e
+ // Verify checkpoint version in sqlConf matches the version recorded in
the read commit log,
+ // in case the user forgot to set STATE_STORE_CHECKPOINT_FORMAT_VERSION.
This prevents the
+ // StateRewriter from writing state files in a format that disagrees with
the source
+ // checkpoint. Using the read batch commit since the latest commit could
be a skipped batch.
+ readCheckpoint.commitLog.get(readBatchId).foreach { metadata =>
+ val configuredVersion = readCheckpoint.commitLog.defaultVersion
+ if (metadata.version != configuredVersion) {
+ throw StateRewriterErrors.stateCheckpointFormatVersionMismatchError(
+ checkpointLocationForRead,
+ expectedVersion = metadata.version.toString,
+ actualVersion = configuredVersion.toString)
}
+ }
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
index bae78f0b4762..4e9f6cca2ffc 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceChangeDataReadSuite.scala
@@ -25,7 +25,7 @@ import org.apache.hadoop.conf.Configuration
import org.scalatest.Assertions
import org.apache.spark.sql.Row
-import org.apache.spark.sql.execution.streaming.checkpointing.{CommitLog,
CommitMetadata}
+import org.apache.spark.sql.execution.streaming.checkpointing.{CommitLog,
CommitMetadata, CommitMetadataV2}
import org.apache.spark.sql.execution.streaming.runtime.{MemoryStream,
StreamExecution}
import org.apache.spark.sql.execution.streaming.state._
import org.apache.spark.sql.functions.{col, window}
@@ -237,11 +237,11 @@ abstract class StateDataSourceChangeDataReaderSuite
extends StateDataSourceTestB
new File(tempDir.getAbsolutePath, "commits").getAbsolutePath)
// Start version: treated as v1 (no operator unique ids)
- val startMetadata = CommitMetadata(0, None)
+ val startMetadata = CommitMetadata(0)
assert(commitLog.add(0, startMetadata))
// End version: treated as v2 (operator 0 has unique ids)
- val endMetadata = CommitMetadata(0,
+ val endMetadata = CommitMetadataV2(0,
Some(Map[Long, Array[Array[String]]](0L -> Array(Array("uid")))))
assert(commitLog.add(1, endMetadata))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
index 2def79828fac..4a2a454077a7 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
@@ -23,7 +23,7 @@ import java.util.UUID
import org.apache.hadoop.conf.Configuration
import org.scalatest.Assertions
-import org.apache.spark.{SparkException, SparkThrowable,
SparkUnsupportedOperationException}
+import org.apache.spark.{SparkException, SparkUnsupportedOperationException}
import org.apache.spark.io.CompressionCodec
import org.apache.spark.sql.{AnalysisException, DataFrame, Encoders, Row}
import org.apache.spark.sql.catalyst.expressions.{BoundReference,
GenericInternalRow}
@@ -589,8 +589,6 @@ class RocksDBWithCheckpointV2StateDataSourceReaderSuite
extends StateDataSourceR
override protected def newStateStoreProvider(): RocksDBStateStoreProvider =
new RocksDBStateStoreProvider
- import testImplicits._
-
override def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION, 2)
@@ -600,34 +598,57 @@ class RocksDBWithCheckpointV2StateDataSourceReaderSuite
extends StateDataSourceR
"true")
}
- // TODO: Remove this test once we allow migrations from checkpoint v1 to v2
- test("reading checkpoint v2 store with version 1 should fail") {
- withTempDir { tmpDir =>
- val inputData = MemoryStream[(Int, Long)]
- val query = getStreamStreamJoinQuery(inputData)
- testStream(query)(
- StartStream(checkpointLocation = tmpDir.getCanonicalPath),
- AddData(inputData, (1, 1L), (2, 2L), (3, 3L), (4, 4L), (5, 5L)),
- ProcessAllAvailable(),
- Execute { _ => Thread.sleep(2000) },
- StopStream
- )
+ // Expected state after runLargeDataStreamingAggregationQuery, read from
batch 2 / operator 0.
+ private val expectedLargeAggregationState: Seq[Row] = Seq(
+ Row(0, 5, 60, 30, 0), Row(1, 5, 65, 31, 1), Row(2, 5, 70, 32, 2),
+ Row(3, 4, 72, 33, 3), Row(4, 4, 76, 34, 4), Row(5, 4, 80, 35, 5),
+ Row(6, 4, 84, 36, 6), Row(7, 4, 88, 37, 7), Row(8, 4, 92, 38, 8),
+ Row(9, 4, 96, 39, 9))
+
+ private def readLargeAggregationState(checkpointDir: String): DataFrame =
+ spark.read.format("statestore")
+ .option(StateSourceOptions.PATH, checkpointDir)
+ .option(StateSourceOptions.BATCH_ID, 2)
+ .option(StateSourceOptions.OPERATOR_ID, 0)
+ .load()
+ .selectExpr("key.groupKey AS key_groupKey", "value.count AS value_cnt",
+ "value.sum AS value_sum", "value.max AS value_max", "value.min AS
value_min")
+ // SPARK-56970: The commit log wire format version is now discovered from
the file header
+ // rather than required to match STATE_STORE_CHECKPOINT_FORMAT_VERSION. As a
result a V1 commit
+ // log can be read under a V2-configured session (and vice versa). Note this
only applies to the
+ // commit log layer; reading a V2 state store still requires version 2 to be
configured because
+ // the state store files are named with checkpoint unique ids.
+ test("SPARK-56970: reading a v1 checkpoint with commit log version 2
configured succeeds") {
+ withTempDir { tempDir =>
+ // Override the suite default to write a V1 checkpoint (no checkpoint
unique ids).
withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "1") {
- // Verify reading state throws error when reading checkpoint v2 with
version 1
- val exc = intercept[IllegalStateException] {
- val stateDf = spark.read.format("statestore")
- .option(StateSourceOptions.BATCH_ID, 0)
- .option(StateSourceOptions.OPERATOR_ID, 0)
- .load(tmpDir.getCanonicalPath)
- stateDf.collect()
- }
+ runLargeDataStreamingAggregationQuery(tempDir.getAbsolutePath)
+ }
+
+ // The suite default reads with version 2 configured; the V1 commit log
must still be read.
+ checkAnswer(
+ readLargeAggregationState(tempDir.getAbsolutePath),
expectedLargeAggregationState)
+ }
+ }
- checkError(exc.getCause.asInstanceOf[SparkThrowable],
- "INVALID_LOG_VERSION.EXACT_MATCH_VERSION", "KD002",
- Map(
- "version" -> "2",
- "matchVersion" -> "1"))
+ test("SPARK-56970: reading a v2 checkpoint with commit log version 1
configured fails on the " +
+ "state store, not the commit log") {
+ withTempDir { tempDir =>
+ // The suite configures commit log format version 2, so this writes a V2
checkpoint whose
+ // state store files are named with checkpoint unique ids.
+ runLargeDataStreamingAggregationQuery(tempDir.getAbsolutePath)
+
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "1") {
+ // The commit log now deserializes across versions, so this no longer
fails with
+ // INVALID_LOG_VERSION at the commit-log layer. Reading the V2 state
store itself still
+ // requires version 2 to be configured: with version 1 the reader
looks for non-unique
+ // state file names and cannot locate the unique-id-named files.
+ val ex = intercept[SparkException] {
+ readLargeAggregationState(tempDir.getAbsolutePath).collect()
+ }
+ assert(ex.getMessage.contains("CANNOT_LOAD_STATE_STORE") ||
+
Option(ex.getCause).map(_.getMessage).exists(_.contains("CANNOT_LOAD_STATE_STORE")))
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
index be7874e806cd..22d0af0a77fd 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatePartitionAllColumnFamiliesWriterSuite.scala
@@ -99,7 +99,7 @@ class StatePartitionAllColumnFamiliesWriterSuite extends
StateDataSourceTestBase
// Commit to commitLog with checkpoint IDs
val latestCommit = targetCheckpointMetadata.commitLog.get(lastBatch).get
- val commitMetadata = latestCommit.copy(stateUniqueIds = checkpointInfos)
+ val commitMetadata = latestCommit.withStateUniqueIds(checkpointInfos)
targetCheckpointMetadata.commitLog.add(writeBatchId, commitMetadata)
val versionToCheck = writeBatchId + 1
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 aa5826572240..5dcfb279fe7a 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
@@ -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}
+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
@@ -63,7 +63,7 @@ class CommitLogSuite extends SparkFunSuite with
SharedSparkSession {
)
}
- private def testSerde(commitMetadata: CommitMetadata, path: Path): Unit = {
+ private def testSerde(commitMetadata: CommitMetadataBase, path: Path): Unit
= {
if (regenerateGoldenFiles) {
val commitLog = new CommitLog(spark, path.toString)
val outputStream = new
FileOutputStream(path.resolve("testCommitLog").toFile)
@@ -103,19 +103,110 @@ class CommitLogSuite extends SparkFunSuite with
SharedSparkSession {
0L -> Array(Array("unique_id1", "unique_id2"), Array("unique_id3",
"unique_id4")),
1L -> Array(Array("unique_id5", "unique_id6"), Array("unique_id7",
"unique_id8"))
)
- val testMetadataV2 = CommitMetadata(0, Some(testStateUniqueIds))
+ val testMetadataV2 = CommitMetadataV2(0, Some(testStateUniqueIds))
testSerde(testMetadataV2, testCommitLogV2FilePath)
}
}
test("Basic Commit Log V2 SerDe - empty stateUniqueIds") {
withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
- val testMetadataV2 = CommitMetadata(0, Some(Map[Long,
Array[Array[String]]]()))
+ val testMetadataV2 = CommitMetadataV2(0, Some(Map[Long,
Array[Array[String]]]()))
testSerde(testMetadataV2, testCommitLogV2FilePathEmptyUniqueId)
}
}
- // Old metadata structure with no state unique ids should not affect the
deserialization
+ 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.
test("Cross-version V1 SerDe") {
withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
val commitlogV1 = """v1
@@ -123,18 +214,41 @@ class CommitLogSuite extends SparkFunSuite with
SharedSparkSession {
val inputStream: ByteArrayInputStream =
new ByteArrayInputStream(commitlogV1.getBytes("UTF-8"))
- // TODO [SPARK-50653]: Uncomment the below when v2 -> v1 backward
compatibility is added
- // val commitMetadata: CommitMetadata = new CommitLog(
- // spark, testCommitLogV1FilePath.toString).deserialize(inputStream)
- // assert(commitMetadata.nextBatchWatermarkMs === 233)
- // assert(commitMetadata.stateUniqueIds === Map.empty)
+ val commitMetadata = new CommitLog(
+ spark, testCommitLogV1FilePath.toString).deserialize(inputStream)
+ assert(commitMetadata.version === CommitLog.VERSION_1)
+ assert(commitMetadata.nextBatchWatermarkMs === 233)
+ assert(commitMetadata.stateUniqueIds.isEmpty)
+ }
+ }
+
+ test("SPARK-56970: creating a V1 commit with stateUniqueIds should fail") {
+ withTempDir { tmpDir =>
+ val commitLog = new CommitLog(spark, tmpDir.getCanonicalPath)
+ val stateUniqueIds: Map[Long, Array[Array[String]]] =
+ Map(0L -> Array(Array("unique_id1", "unique_id2")))
+
+ // Through the createMetadata factory with an explicit V1 format version.
+ val e1 = intercept[IllegalArgumentException] {
+ commitLog.createMetadata(
+ nextBatchWatermarkMs = 1,
+ stateUniqueIds = Some(stateUniqueIds),
+ commitLogFormatVersion = CommitLog.VERSION_1)
+ }
+ assert(e1.getMessage.contains("stateUniqueIds cannot be set"))
- // TODO [SPARK-50653]: remove the below when v2 -> v1 backward
compatibility is added
- val e = intercept[IllegalStateException] {
- new CommitLog(spark,
testCommitLogV1FilePath.toString).deserialize(inputStream)
+ // Directly through withStateUniqueIds on a V1 metadata.
+ val e2 = intercept[IllegalArgumentException] {
+ CommitMetadata(1).withStateUniqueIds(Some(stateUniqueIds))
}
+ assert(e2.getMessage.contains("stateUniqueIds cannot be set"))
- assert (e.getMessage.contains("only supported log version"))
+ // None and an empty map are allowed for V1 (no unique ids to persist).
+ assert(CommitMetadata(1).withStateUniqueIds(None).stateUniqueIds.isEmpty)
+ assert(commitLog.createMetadata(
+ nextBatchWatermarkMs = 1,
+ stateUniqueIds = Some(Map.empty[Long, Array[Array[String]]]),
+ commitLogFormatVersion = CommitLog.VERSION_1).version ===
CommitLog.VERSION_1)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
