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 2026/10/13 04:45:07 UTC
(spark) branch branch-4.x updated: [SPARK-60181][SS] Guard stream-stream join state format and RocksDB merge operator compatibility
This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR 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 bd6f74144e56 [SPARK-60181][SS] Guard stream-stream join state format and RocksDB merge operator compatibility
bd6f74144e56 is described below
commit bd6f74144e56f9aa901637a99e65c34ad5441711
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Tue Oct 13 13:44:19 2026 +0900
[SPARK-60181][SS] Guard stream-stream join state format and RocksDB merge operator compatibility
### What changes were proposed in this pull request?
For a new stream-stream join checkpoint, align the configured state format and merge operator with the state store:
- Fall back from join state format V3 or V4 to V2 with a non-RocksDB-compatible provider. RocksDB subclasses remain compatible.
- When V4 is retained with RocksDB, promote merge operator V1 to V2 before recording the configuration in the offset log.
Existing checkpoints keep their pinned settings. If a checkpoint pins an incompatible V4/merge-V1 combination, physical planning fails with a structured error; changing the merge version requires a new checkpoint.
### Why are the changes needed?
V3 requires virtual column families. V4 also depends on RocksDB-supported StateStore APIs (including `rangeScanWithMultiValues` for bounded timestamp scans) and requires merge operator V2. A query that explicitly requests V3/V4 with an incompatible provider, or V4 with merge operator V1, cannot run as configured. Spark's default HDFS provider and join state format V2 are unchanged.
### Does this PR introduce _any_ user-facing change?
Yes. New stream-stream joins configured with V3/V4 and an incompatible provider use V2. New V4 joins with RocksDB and merge operator V1 use merge operator V2 instead. Existing checkpoints retain their recorded settings; an incompatible pinned V4/merge-V1 combination fails early with a structured error rather than changing stored state.
### How was this patch tested?
`build/sbt 'sql/testOnly org.apache.spark.sql.streaming.StreamingInnerJoinWithVCFSuite org.apache.spark.sql.streaming.StreamingInnerJoinV4Suite -- -z "non-RocksDB provider uses compatible format" -z "promotes RocksDB merge operator version 1" -z "state format version 4 rejects an incompatible merge operator"'` (6 passed).
`build/sbt sql/scalastyle sql/Test/scalastyle` (0 errors).
The tests cover the effective state format and merge operator in the plan and offset log, the unchanged caller session, a restart from the same checkpoint, and fail-fast validation of an incompatible physical plan.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5); GPT-5.2 (Codex CLI).
Closes #59389 from HeartSaVioR/guard-join-state-format-non-rocksdb.
Authored-by: Jungtaek Lim <ka...@gmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit 0974a9f5970f47de5b320ce663c130bf0bef2b97)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../src/main/resources/error/error-conditions.json | 6 ++
.../join/StreamingSymmetricHashJoinExec.scala | 16 +++++
.../streaming/runtime/MicroBatchExecution.scala | 42 +++++++++++-
.../spark/sql/streaming/StreamingJoinSuite.scala | 43 ++++++++----
.../spark/sql/streaming/StreamingJoinV4Suite.scala | 77 ++++++++++++++++++++++
5 files changed, 170 insertions(+), 14 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 04df7b966afb..47694155ddb1 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -8126,6 +8126,12 @@
],
"sqlState" : "XXKST"
},
+ "STREAM_STREAM_JOIN_INCOMPATIBLE_STATE_STORE_CONFIGS" : {
+ "message" : [
+ "The configuration values `<stateFormatVersionConfig>=<stateFormatVersion>` and `<mergeOperatorVersionConfig>=<mergeOperatorVersion>` are incompatible. Stream-stream join state format version <stateFormatVersion> requires RocksDB merge operator version <requiredMergeOperatorVersion>. Start a new query with `<mergeOperatorVersionConfig>` set to <requiredMergeOperatorVersion> and a new checkpoint."
+ ],
+ "sqlState" : "42K06"
+ },
"STREAM_STREAM_JOIN_INCONSISTENT_STATE" : {
"message" : [
"Detected inconsistency in stream-stream join state."
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala
index 8f90a603c7ef..853117c49c9f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala
@@ -21,6 +21,7 @@ import java.util.concurrent.TimeUnit.NANOSECONDS
import org.apache.hadoop.conf.Configuration
+import org.apache.spark.SparkException
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.StreamingJoinHelper
@@ -173,6 +174,21 @@ case class StreamingSymmetricHashJoinExec(
" the checkpoint and rerun the query. See SPARK-26154 for more details.")
}
+ if (stateFormatVersion == 4) {
+ val mergeOperatorVersion =
+ conf.getConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION)
+ SparkException.require(
+ mergeOperatorVersion == 2,
+ errorClass = "STREAM_STREAM_JOIN_INCOMPATIBLE_STATE_STORE_CONFIGS",
+ messageParameters = Map(
+ "stateFormatVersionConfig" -> SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key,
+ "stateFormatVersion" -> stateFormatVersion.toString,
+ "mergeOperatorVersionConfig" ->
+ SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key,
+ "mergeOperatorVersion" -> mergeOperatorVersion.toString,
+ "requiredMergeOperatorVersion" -> "2"))
+ }
+
private lazy val errorMessageForJoinType =
s"${getClass.getSimpleName} should not take $joinType as the JoinType"
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 1bda5691f80b..c92bb4f12ad4 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
@@ -51,7 +51,7 @@ import org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileMan
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}
-import org.apache.spark.sql.execution.streaming.state.{OfflineStateRepartitionUtils, StateSchemaBroadcast, StateStoreErrors}
+import org.apache.spark.sql.execution.streaming.state.{OfflineStateRepartitionUtils, RocksDBStateStoreProvider, StateSchemaBroadcast, StateStoreErrors}
import org.apache.spark.sql.execution.streaming.utils.StreamingUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.connector.PartitionOffsetWithIndex
@@ -241,6 +241,7 @@ class MicroBatchExecution(
// the order pinned in the offset log (mirroring how `enforceNamed` is read above). For a new
// query the offset log has no entry yet, so the session value is used and pinned at batch 0.
val streamConf = sparkSessionForStream.sessionState.conf
+ applyJoinStateConfigGuardrails(streamConf)
val orderDeterministically = initialLatestOffsetSeq.flatMap { case (_, offsetSeq) =>
offsetSeq.metadataOpt.flatMap { metadata =>
OffsetSeqMetadata.readValueOpt(metadata, SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER)
@@ -466,6 +467,45 @@ class MicroBatchExecution(
}
}
+ /**
+ * Align new stream-stream joins with the selected state store before checkpoint configuration
+ * is pinned. On restart the checkpoint's configuration must take precedence.
+ */
+ private def applyJoinStateConfigGuardrails(streamConf: SQLConf): Unit = {
+ def hasStreamStreamJoin: Boolean = analyzedPlan.exists {
+ case join: Join => join.left.isStreaming && join.right.isStreaming
+ case _ => false
+ }
+
+ def isRocksDBProvider(providerClassName: String): Boolean =
+ classOf[RocksDBStateStoreProvider].isAssignableFrom(Utils.classForName(providerClassName))
+
+ val configuredVersion = streamConf.getConf(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION)
+ if (initialLatestOffsetSeq.isEmpty && configuredVersion >= 3 && hasStreamStreamJoin) {
+ val stateStoreProvider = streamConf.getConf(SQLConf.STATE_STORE_PROVIDER_CLASS)
+ if (!isRocksDBProvider(stateStoreProvider)) {
+ val fallbackVersion = 2
+ logInfo(log"Overriding " +
+ log"${MDC(CONFIG, SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key)} from " +
+ log"${MDC(OLD_VALUE, configuredVersion)} to ${MDC(NEW_VALUE, fallbackVersion)} " +
+ log"because ${MDC(CONFIG, SQLConf.STATE_STORE_PROVIDER_CLASS.key)} is set to " +
+ log"${MDC(CLASS_NAME, stateStoreProvider)}, which does not support the virtual column " +
+ log"families required by this join state format.")
+ streamConf.setConf(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION, fallbackVersion)
+ } else if (configuredVersion == 4) {
+ val mergeOperatorVersion =
+ streamConf.getConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION)
+ if (mergeOperatorVersion != 2) {
+ logInfo(log"Overriding " +
+ log"${MDC(CONFIG, SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key)} from " +
+ log"${MDC(OLD_VALUE, mergeOperatorVersion)} to ${MDC(NEW_VALUE, 2)} " +
+ log"because join state format version 4 requires RocksDB merge operator version 2.")
+ streamConf.setConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION, 2)
+ }
+ }
+ }
+ }
+
/**
* Signals to the thread executing micro-batches that it should stop running after the next
* batch. This method blocks until the thread stops running.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala
index c0983f338abe..9b91dea19606 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
import org.apache.spark.sql.execution.streaming.checkpointing.CheckpointFileManager
import org.apache.spark.sql.execution.streaming.operators.stateful.StatefulOperatorStateInfo
import org.apache.spark.sql.execution.streaming.operators.stateful.join.{StreamingSymmetricHashJoinExec, StreamingSymmetricHashJoinHelper}
-import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.execution.streaming.runtime.{MemoryStream, StreamExecution}
import org.apache.spark.sql.execution.streaming.state._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
@@ -977,7 +977,7 @@ abstract class StreamingInnerJoinBase extends StreamingJoinSuite {
}
testWithVirtualColumnFamilyJoins(
- "SPARK-51779 Join query using virtual column families with HDFS should fail") {
+ "Join query with a non-RocksDB provider uses compatible format") {
withTempDir { checkpointDir =>
withSQLConf(
SQLConf.STATE_STORE_PROVIDER_CLASS.key -> classOf[HDFSBackedStateStoreProvider].getName
@@ -988,20 +988,37 @@ abstract class StreamingInnerJoinBase extends StreamingJoinSuite {
val df1 = input1.toDF().select($"value" as "key", ($"value" * 2) as "leftValue")
val df2 = input2.toDF().select($"value" as "key", ($"value" * 3) as "rightValue")
val joined = df1.join(df2, "key")
+ val configuredVersion =
+ SQLConf.get.getConf(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION)
+
+ def checkJoinStateFormat(query: StreamExecution): Unit = {
+ val join = query.lastExecution.executedPlan.collectFirst {
+ case join: StreamingSymmetricHashJoinExec => join
+ }.getOrElse(fail("StreamingSymmetricHashJoinExec not found"))
+ assert(join.stateFormatVersion == 2)
+ assert(SQLConf.get.getConf(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION) ==
+ configuredVersion, "the guardrail must not mutate the caller's session")
+ }
+ assert(Set(3, 4).contains(configuredVersion))
testStream(joined)(
StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
- AddData(input1, 1, 2),
- ExpectFailure[StateStoreMultipleColumnFamiliesNotSupportedException] {
- (t: Throwable) => {
- checkError(
- exception = t.asInstanceOf[StateStoreMultipleColumnFamiliesNotSupportedException],
- condition = "UNSUPPORTED_FEATURE.STATE_STORE_MULTIPLE_COLUMN_FAMILIES",
- sqlState = Some("0A000"),
- parameters = Map("stateStoreProvider" -> "HDFSBackedStateStoreProvider")
- )
- }
- }
+ AddData(input1, 1),
+ AddData(input2, 1),
+ CheckAnswer((1, 2, 3)),
+ Execute { query =>
+ checkJoinStateFormat(query)
+ assert(query.offsetLog.getLatest().flatMap(_._2.metadataOpt)
+ .flatMap(_.conf.get(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key))
+ .contains("2"))
+ },
+ StopStream,
+ StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
+ AddData(input1, 2),
+ AddData(input2, 2),
+ CheckNewAnswer((2, 4, 6)),
+ Execute(checkJoinStateFormat),
+ StopStream
)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala
index 6d4a97861efe..6436c0c8d6ba 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinV4Suite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.streaming
import org.apache.hadoop.fs.Path
import org.scalatest.Tag
+import org.apache.spark.SparkIllegalArgumentException
import org.apache.spark.sql.execution.datasources.v2.state.StateSourceOptions
import org.apache.spark.sql.execution.streaming.checkpointing.CheckpointFileManager
import org.apache.spark.sql.execution.streaming.operators.stateful.join.StreamingSymmetricHashJoinExec
@@ -88,6 +89,82 @@ class StreamingInnerJoinV4Suite
)
}
+ test("state format version 4 promotes RocksDB merge operator version 1 for new joins") {
+ withTempDir { checkpointDir =>
+ withSQLConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key -> "1") {
+ val input1 = MemoryStream[Int]
+ val input2 = MemoryStream[Int]
+ val df1 = input1.toDF().select($"value" as "key", ($"value" * 2) as "leftValue")
+ val df2 = input2.toDF().select($"value" as "key", ($"value" * 3) as "rightValue")
+ val joined = df1.join(df2, "key")
+
+ def checkMergeOperatorVersion(query: StreamExecution): Unit = {
+ val join = query.lastExecution.executedPlan.collectFirst {
+ case join: StreamingSymmetricHashJoinExec => join
+ }.getOrElse(fail("StreamingSymmetricHashJoinExec not found"))
+ assert(join.stateFormatVersion == 4)
+ assert(SQLConf.get.getConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION) == 1,
+ "the guardrail must not mutate the caller's session")
+ }
+
+ testStream(joined)(
+ StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
+ AddData(input1, 1),
+ AddData(input2, 1),
+ CheckAnswer((1, 2, 3)),
+ Execute { query =>
+ checkMergeOperatorVersion(query)
+ assert(query.offsetLog.getLatest().flatMap(_._2.metadataOpt)
+ .flatMap(_.conf.get(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key))
+ .contains("2"))
+ },
+ StopStream,
+ StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
+ AddData(input1, 2),
+ AddData(input2, 2),
+ CheckNewAnswer((2, 4, 6)),
+ Execute(checkMergeOperatorVersion),
+ StopStream
+ )
+ }
+ }
+ }
+
+ test("state format version 4 rejects an incompatible merge operator in a physical plan") {
+ val input1 = MemoryStream[Int]
+ val input2 = MemoryStream[Int]
+ val joined = input1.toDF().join(input2.toDF(), "value")
+
+ testStream(joined)(
+ AddData(input1, 1),
+ CheckAnswer(),
+ Execute { query =>
+ val join = query.lastExecution.executedPlan.collectFirst {
+ case join: StreamingSymmetricHashJoinExec => join
+ }.getOrElse(fail("StreamingSymmetricHashJoinExec not found"))
+ withSQLConf(SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key -> "1") {
+ // New streams promote v1 to v2 before planning. copy() rebuilds this join with the
+ // caller's v1 setting so the incompatible-plan validation remains covered.
+ val error = intercept[SparkIllegalArgumentException] {
+ join.copy()
+ }
+ checkError(
+ exception = error,
+ condition = "STREAM_STREAM_JOIN_INCOMPATIBLE_STATE_STORE_CONFIGS",
+ sqlState = Some("42K06"),
+ parameters = Map(
+ "stateFormatVersionConfig" -> SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key,
+ "stateFormatVersion" -> "4",
+ "mergeOperatorVersionConfig" ->
+ SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key,
+ "mergeOperatorVersion" -> "1",
+ "requiredMergeOperatorVersion" -> "2"))
+ }
+ },
+ StopStream
+ )
+ }
+
// V4 uses different column families (keyWithTsToValues, tsWithKey)
// with timestamp-based key encoder specs instead of V3's
// keyToNumValues/keyWithIndexToValue.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org