You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2025/01/16 02:54:48 UTC
(spark) branch branch-4.0 updated: [SPARK-50654][SS] CommitMetadata should set stateUniqueIds to None in V1
This is an automated email from the ASF dual-hosted git repository.
kabhwan pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 4ac6bcc25208 [SPARK-50654][SS] CommitMetadata should set stateUniqueIds to None in V1
4ac6bcc25208 is described below
commit 4ac6bcc252087ac11cb25a73937e169267a88ec9
Author: Wei Liu <we...@databricks.com>
AuthorDate: Thu Jan 16 11:54:19 2025 +0900
[SPARK-50654][SS] CommitMetadata should set stateUniqueIds to None in V1
### What changes were proposed in this pull request?
Per request from this comment, https://github.com/apache/spark/pull/48355/files#r1866546734, we will put `None` in the `stateUniqueIds` of `CommitMetadata` when in V1. So now the logic is as follows:
1. In V1, `CommitMetadata.stateUniqueIds = None` for all stateful and stateless queries
2. In V2, `CommitMetadata.stateUniqueIds = Some(Map.empty)` for stateless queries
3. In V2, `CommitMetadata.stateUniqueIds = Some(<non-empty-map>)` for stateful queries
Also removed the legacy test as a side effect
### Why are the changes needed?
Follow up
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Unit tests
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #49278 from WweiL/microbatchExecution-commitlog-none.
Authored-by: Wei Liu <we...@databricks.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit b6685e36354a9718b89fe5898d703b251ee608fe)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../spark/sql/execution/streaming/CommitLog.scala | 6 +-
.../execution/streaming/MicroBatchExecution.scala | 12 +-
.../testCommitLogV1/testCommitLog | 2 +-
.../testCommitLogV2-empty-unique-id/testCommitLog | 2 +
.../testCommitLogV2/testCommitLog | 2 +-
.../spark/sql/streaming/CommitLogSuite.scala | 128 ++++++++-------------
6 files changed, 66 insertions(+), 86 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CommitLog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CommitLog.scala
index 84e70d89fd8e..a7fcaa5ccbcb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CommitLog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CommitLog.scala
@@ -51,6 +51,9 @@ class CommitLog(sparkSession: SparkSession, path: String)
import CommitLog._
+ private val VERSION: 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()
@@ -74,7 +77,6 @@ class CommitLog(sparkSession: SparkSession, path: String)
}
object CommitLog {
- private val VERSION = SQLConf.get.stateStoreCheckpointFormatVersion
private val EMPTY_JSON = "{}"
}
@@ -104,7 +106,7 @@ object CommitLog {
case class CommitMetadata(
nextBatchWatermarkMs: Long = 0,
- stateUniqueIds: Map[Long, Array[Array[String]]] = Map.empty) {
+ stateUniqueIds: Option[Map[Long, Array[Array[String]]]] = None) {
def json: String = Serialization.write(this)(CommitMetadata.format)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
index 23e72fc4e3e2..18513a904e25 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
@@ -513,7 +513,9 @@ class MicroBatchExecution(
execCtx.startOffsets ++= execCtx.endOffsets
watermarkTracker.setWatermark(
math.max(watermarkTracker.currentWatermark, commitMetadata.nextBatchWatermarkMs))
- currentStateStoreCkptId ++= commitMetadata.stateUniqueIds
+ commitMetadata.stateUniqueIds.foreach {
+ stateUniqueIds => currentStateStoreCkptId ++= stateUniqueIds
+ }
} else if (latestCommittedBatchId == latestBatchId - 1) {
execCtx.endOffsets.foreach {
case (source: Source, end: Offset) =>
@@ -966,8 +968,14 @@ class MicroBatchExecution(
updateStateStoreCkptId(execCtx, latestExecPlan)
}
execCtx.reportTimeTaken("commitOffsets") {
+ val stateStoreCkptId = if (StatefulOperatorStateInfo.enableStateStoreCheckpointIds(
+ sparkSessionForStream.sessionState.conf)) {
+ Some(currentStateStoreCkptId.toMap)
+ } else {
+ None
+ }
if (!commitLog.add(execCtx.batchId,
- CommitMetadata(watermarkTracker.currentWatermark, currentStateStoreCkptId.toMap))) {
+ CommitMetadata(watermarkTracker.currentWatermark, stateStoreCkptId))) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
}
diff --git a/sql/core/src/test/resources/structured-streaming/testCommitLogV1/testCommitLog b/sql/core/src/test/resources/structured-streaming/testCommitLogV1/testCommitLog
index 47bde671130f..c60b098f4319 100644
--- a/sql/core/src/test/resources/structured-streaming/testCommitLogV1/testCommitLog
+++ b/sql/core/src/test/resources/structured-streaming/testCommitLogV1/testCommitLog
@@ -1,2 +1,2 @@
v1
-{"nextBatchWatermarkMs":1,"stateUniqueIds":{}}
\ No newline at end of file
+{"nextBatchWatermarkMs":1}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/testCommitLogV2-empty-unique-id/testCommitLog b/sql/core/src/test/resources/structured-streaming/testCommitLogV2-empty-unique-id/testCommitLog
new file mode 100644
index 000000000000..4773fd170aeb
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/testCommitLogV2-empty-unique-id/testCommitLog
@@ -0,0 +1,2 @@
+v2
+{"nextBatchWatermarkMs":0,"stateUniqueIds":{}}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/testCommitLogV2/testCommitLog b/sql/core/src/test/resources/structured-streaming/testCommitLogV2/testCommitLog
index 27f51eb053cf..8648c062939c 100644
--- a/sql/core/src/test/resources/structured-streaming/testCommitLogV2/testCommitLog
+++ b/sql/core/src/test/resources/structured-streaming/testCommitLogV2/testCommitLog
@@ -1,2 +1,2 @@
-v1
+v2
{"nextBatchWatermarkMs":0,"stateUniqueIds":{"0":[["unique_id1","unique_id2"],["unique_id3","unique_id4"]],"1":[["unique_id5","unique_id6"],["unique_id7","unique_id8"]]}}
\ No newline at end of file
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 068f56839e6e..ea2f4154170a 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
@@ -17,18 +17,12 @@
package org.apache.spark.sql.streaming
-import java.io.{ByteArrayInputStream, FileInputStream, FileOutputStream, InputStream, OutputStream}
-import java.nio.charset.StandardCharsets.UTF_8
+import java.io.{ByteArrayInputStream, FileInputStream, FileOutputStream}
import java.nio.file.Path
-import scala.io.{Source => IOSource}
-
-import org.json4s.{Formats, NoTypeHints}
-import org.json4s.jackson.Serialization
-
import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.execution.streaming.{CommitLog, CommitMetadata, HDFSMetadataLog}
+import org.apache.spark.sql.execution.streaming.{CommitLog, CommitMetadata}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class CommitLogSuite extends SparkFunSuite with SharedSparkSession {
@@ -45,6 +39,18 @@ class CommitLogSuite extends SparkFunSuite with SharedSparkSession {
)
}
+ private def testCommitLogV2FilePathEmptyUniqueId: Path = {
+ getWorkspaceFilePath(
+ "sql",
+ "core",
+ "src",
+ "test",
+ "resources",
+ "structured-streaming",
+ "testCommitLogV2-empty-unique-id"
+ )
+ }
+
private def testCommitLogV1FilePath: Path = {
getWorkspaceFilePath(
"sql",
@@ -68,30 +74,45 @@ class CommitLogSuite extends SparkFunSuite with SharedSparkSession {
val metadata = commitLog.deserialize(inputStream)
// Array comparison are reference based, so we need to compare the elements
assert(metadata.nextBatchWatermarkMs == commitMetadata.nextBatchWatermarkMs)
- assert(metadata.stateUniqueIds.size == commitMetadata.stateUniqueIds.size)
- commitMetadata.stateUniqueIds.foreach { case (operatorId, uniqueIds) =>
- assert(metadata.stateUniqueIds.contains(operatorId))
- assert(metadata.stateUniqueIds(operatorId).length == uniqueIds.length)
- assert(metadata.stateUniqueIds(operatorId).zip(uniqueIds).forall {
- case (a, b) => a.sameElements(b)
- })
+ if (metadata.stateUniqueIds.isEmpty) {
+ assert(commitMetadata.stateUniqueIds.isEmpty)
+ } else {
+ assert(metadata.stateUniqueIds.get.size == commitMetadata.stateUniqueIds.get.size)
+ commitMetadata.stateUniqueIds.get.foreach { case (operatorId, uniqueIds) =>
+ assert(metadata.stateUniqueIds.get.contains(operatorId))
+ assert(metadata.stateUniqueIds.get(operatorId).length == uniqueIds.length)
+ assert(metadata.stateUniqueIds.get(operatorId).zip(uniqueIds).forall {
+ case (a, b) => a.sameElements(b)
+ })
+ }
}
}
}
test("Basic Commit Log V1 SerDe") {
- val testMetadataV1 = CommitMetadata(1)
- testSerde(testMetadataV1, testCommitLogV1FilePath)
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "1") {
+ val testMetadataV1 = CommitMetadata(1)
+ testSerde(testMetadataV1, testCommitLogV1FilePath)
+ }
}
- test("Basic Commit Log V2 SerDe") {
- val testStateUniqueIds: Map[Long, Array[Array[String]]] =
- Map(
- 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, testStateUniqueIds)
- testSerde(testMetadataV2, testCommitLogV2FilePath)
+ test("Basic Commit Log V2 SerDe - nonempty stateUniqueIds") {
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
+ val testStateUniqueIds: Map[Long, Array[Array[String]]] =
+ Map(
+ 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))
+ 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]]]()))
+ testSerde(testMetadataV2, testCommitLogV2FilePathEmptyUniqueId)
+ }
}
// Old metadata structure with no state unique ids should not affect the deserialization
@@ -103,59 +124,6 @@ class CommitLogSuite extends SparkFunSuite with SharedSparkSession {
val commitMetadata: CommitMetadata = new CommitLog(
spark, testCommitLogV1FilePath.toString).deserialize(inputStream)
assert(commitMetadata.nextBatchWatermarkMs === 233)
- assert(commitMetadata.stateUniqueIds === Map.empty)
- }
-
- // Test an old version of Spark can ser-de the new version of commit log,
- // but running under V1 (i.e. no stateUniqueIds)
- test("v1 Serde backward compatibility") {
- // This is the json created by a V1 commit log
- val commitLogV1WithStateUniqueId = """v1
- |{"nextBatchWatermarkMs":1,"stateUniqueIds":{}}""".stripMargin
- val inputStream: ByteArrayInputStream =
- new ByteArrayInputStream(commitLogV1WithStateUniqueId.getBytes("UTF-8"))
- val commitMetadata: CommitMetadataLegacy = new CommitLogLegacy(
- spark, testCommitLogV1FilePath.toString).deserialize(inputStream)
- assert(commitMetadata.nextBatchWatermarkMs === 1)
- }
-}
-
-// DO-NOT-MODIFY-THE-CODE-BELOW
-// Below are the legacy commit log code carbon copied from Spark branch-3.5, except
-// adding a "Legacy" to the class names.
-case class CommitMetadataLegacy(nextBatchWatermarkMs: Long = 0) {
- def json: String = Serialization.write(this)(CommitMetadataLegacy.format)
-}
-
-object CommitMetadataLegacy {
- implicit val format: Formats = Serialization.formats(NoTypeHints)
-
- def apply(json: String): CommitMetadataLegacy = Serialization.read[CommitMetadataLegacy](json)
-}
-
-class CommitLogLegacy(sparkSession: SparkSession, path: String)
- extends HDFSMetadataLog[CommitMetadataLegacy](sparkSession, path) {
-
- private val VERSION = 1
- private val EMPTY_JSON = "{}"
-
- override def deserialize(in: InputStream): CommitMetadataLegacy = {
- // 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")
- }
- validateVersion(lines.next().trim, VERSION)
- val metadataJson = if (lines.hasNext) lines.next() else EMPTY_JSON
- CommitMetadataLegacy(metadataJson)
- }
-
- override def serialize(metadata: CommitMetadataLegacy, 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('\n')
-
- // write metadata
- out.write(metadata.json.getBytes(UTF_8))
+ assert(commitMetadata.stateUniqueIds === None)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org