This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 a7e43c6dbdb2 [SPARK-58659][SS] Set Real-Time Mode config defaults
a7e43c6dbdb2 is described below
commit a7e43c6dbdb255304c1525e534a85eaabfc41967
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Wed Aug 12 08:03:19 2026 +0800
[SPARK-58659][SS] Set Real-Time Mode config defaults
### What changes were proposed in this pull request?
This PR sets the configuration defaults a Real-Time Mode (RTM) streaming
query needs, resolves the commit log format version from the checkpoint rather
than blindly from the session config, and rejects incompatible session configs
up front. All three are ported from the Databricks Runtime implementation of
RTM.
**1. RTM configuration defaults.** A single method
`setSparkSessionConfigsForRealTimeMode`, run once at query start on the query's
isolated session, defaults the settings RTM wants but that are not the right
engine-wide default. Each is applied only when the user has not set the key, so
an explicit choice is preserved:
- `spark.sql.streaming.stateStore.checkpointFormatVersion=2` and the
RocksDB state store provider that v2 requires, guarded independently (pinning
the provider does not suppress the version default).
-
`spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled=true`
(RocksDB only), which shortens the state-commit step on the critical path
between batches.
- `spark.sql.execution.sortBeforeRepartition=false`, consolidated here from
`MicroBatchExecution` (originally added for RTM in SPARK-58508). The local sort
before a round-robin repartition never drains an unbounded stream, so it would
hang an RTM query; RTM does not retry tasks, so the determinism it provides is
not needed.
**2. Commit log format version resolution.** Previously the commit log
version was read from the session config at `CommitLog` construction
(`CommitLog.defaultVersion`), which was forced early during `StreamExecution`
initialization and ignored the format an existing checkpoint was created with.
This PR removes that field and resolves the write version explicitly, mirroring
the runtime:
- `CommitLog.createMetadata` requires callers to pass the resolved version
explicitly.
- `CheckpointVersionManager` gains a `CommitLogType` and
`resolveCommitLogVersion`: ordinary writes to an existing checkpoint keep its
recorded version, while a fresh checkpoint uses the version implied by the
state store checkpoint format. The sink-evolution path may independently
upgrade writes to commit log v3.
- The resolved version travels on the execution context
(`commitLogFormatVersionOpt`) and is written on both the synchronous and the
async-progress-tracking commit paths.
**3. RTM requires commit log v2.** RTM re-executes a failed batch, and with
checkpoint format v1 the re-execution can reuse the state file names of the
partially-written failed batch and lose data; format v2 avoids this with
per-batch state store checkpoint ids, which only a commit log at v2 or above
can persist. Two guards enforce this, covering disjoint cases:
- A pre-flight in `StreamingQueryManager`
(`throwIfConfsAreRealTimeModeIncompatible`) rejects explicit RTM-incompatible
session configs before the query is built (checkpoint format below v2, a
non-RocksDB provider, or `sortBeforeRepartition=true`), with
`STREAMING_REAL_TIME_MODE.SQL_CONFIGURATION_NOT_SUPPORTED`.
- A fail-fast in `MicroBatchExecution.initializeExecution` rejects a
resolved commit log version below v2 — the case of an existing v1 checkpoint,
which carries no explicit config for the pre-flight to see — with
`STREAMING_REAL_TIME_MODE.CHECKPOINT_FORMAT_V1_NOT_SUPPORTED`.
Both are bypassed by the escape hatch
`spark.sql.streaming.realTimeMode.dangerouslyAllowCheckpointV1.enabled`
(default false), which lets a query start on a v1 checkpoint while accepting
the data-loss risk on failure.
One new internal config is added:
`spark.sql.streaming.realTimeMode.dangerouslyAllowCheckpointV1.enabled`
(default false).
### Why are the changes needed?
RTM commits state at every batch boundary, and that commit sits on the
critical path between batches, so RTM needs a low-latency, recovery-correct
state store configuration (RocksDB + checkpoint format v2 + changelog
checkpointing) that is not the right default for the engine as a whole. Without
these defaults an RTM query either runs with a slower configuration or, in the
`sortBeforeRepartition` case, does not make progress at all.
Format v2 also matters for correctness: RTM re-executes a failed batch, and
at v1 the re-execution can reuse the failed batch's state file names, so a
stale file can be loaded (the checksum hazard documented on `StateStoreConf`).
Resolving the commit log version from the checkpoint (rather than the session
config) keeps ordinary writes at the format the checkpoint was created with,
and the two guards prevent an RTM query from silently running on a v1
checkpoint.
This aligns OSS RTM with the Databricks Runtime, so the two share the same
defaulting, version-resolution, and rejection behavior.
### Does this PR introduce _any_ user-facing change?
Yes, for Real-Time Mode queries (`Trigger.RealTime(...)`) only. Behavior
for every non-RTM query is unchanged.
- An RTM query now defaults to the RocksDB state store, checkpoint format
v2, and changelog checkpointing, when the user has not configured them. An
explicit user setting is preserved.
- Starting an RTM query with an explicit RTM-incompatible session config
(checkpoint format below v2, a non-RocksDB provider, or
`sortBeforeRepartition=true`) now fails at query start with
`STREAMING_REAL_TIME_MODE.SQL_CONFIGURATION_NOT_SUPPORTED`.
- Starting an RTM query on a checkpoint whose commit log is at version 1
now fails at start with
`STREAMING_REAL_TIME_MODE.CHECKPOINT_FORMAT_V1_NOT_SUPPORTED`.
- The escape hatch
`spark.sql.streaming.realTimeMode.dangerouslyAllowCheckpointV1.enabled=true`
allows a v1 checkpoint to proceed, accepting the possibility of data loss on a
failure.
The new config is `internal()`. These are new-in-unreleased changes; there
is no behavior change for released Spark versions, which have no Real-Time Mode.
### How was this patch tested?
New and updated unit tests:
- `StreamRealTimeModeDefaultConfsSuite` (new) covers: the v2 + RocksDB
defaults are applied; the version and provider defaults are independent;
explicit settings are preserved; an explicit incompatible config is rejected up
front; the escape hatch bypasses the pre-flight; an existing v1 checkpoint is
rejected; and the escape hatch allows it.
- `CommitLogSuite` gains tests for fresh commit-log versions deriving from
the state-store format, an existing checkpoint recorded version winning over
the session setting, and commit metadata restoring the implied state-store
format.
- `StreamRealTimeModeSuite`'s `sortBeforeRepartition` test is updated to
expect the up-front rejection.
Verified against the existing RTM suites (`StreamRealTimeModeSuite`,
`StreamRealTimeModeWithManualClockSuite`, the async-progress and streamline RTM
suites) and, because the commit-log write path is shared, against
`StreamingAggregationSuite` and `StreamingDeduplicationSuite` for non-RTM
regressions.
### Was this patch authored or co-authored using generative AI tooling?
co-authored with Claude Code
Closes #57842 from jerrypeng/rtm-default-configs.
Lead-authored-by: Boyang Jerry Peng <[email protected]>
Co-authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit a2d312ff9a19d66d6218459d49d1c1f805119dde)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 11 +
.../org/apache/spark/sql/internal/SQLConf.scala | 23 ++
.../spark/sql/classic/StreamingQueryManager.scala | 51 ++++-
.../checkpointing/CheckpointVersionManager.scala | 60 +++++-
.../streaming/checkpointing/CommitLog.scala | 13 +-
.../AsyncProgressTrackingMicroBatchExecution.scala | 21 +-
.../streaming/runtime/MicroBatchExecution.scala | 70 +++---
.../streaming/runtime/StreamExecution.scala | 89 +++++++-
.../streaming/runtime/StreamExecutionContext.scala | 2 +
.../execution/streaming/state/StateRewriter.scala | 3 +-
.../spark/sql/streaming/CommitLogSuite.scala | 68 +++++-
.../StreamRealTimeModeDefaultConfsSuite.scala | 238 +++++++++++++++++++++
.../sql/streaming/StreamRealTimeModeSuite.scala | 22 +-
.../test/StreamingSinkEvolutionSuite.scala | 37 +++-
14 files changed, 654 insertions(+), 54 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 099bffcf142e..db1bfd43ad23 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -7757,6 +7757,12 @@
"The checkpointing interval for async progress tracking must be set
to 0, which means each progress update is checkpointed. Set option
asyncProgressTrackingCheckpointIntervalMs to 0 in DataStreamWriter options and
retry your query."
]
},
+ "CHECKPOINT_FORMAT_V1_NOT_SUPPORTED" : {
+ "message" : [
+ "Real-time mode does not support state store checkpoint format v1,
because a re-executed batch can reuse the state file names of a
partially-written failed batch and lose data. If you encountered this while
switching an existing query to real-time mode, use a fresh checkpoint
location.",
+ "If you must continue with your existing checkpoint and accept that
doing so exposes you to the possibility of data loss on failure, set <config>
to true."
+ ]
+ },
"IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED" : {
"message" : [
"Real-time mode does not support union on two or more identical
streaming data sources in a single query. This includes scenarios such as
referencing the same source DataFrame more than once, or using two data sources
with identical configurations for some sources. For Kafka, avoid reusing the
same DataFrame and create different ones. Sources provided in the query:
<sources>"
@@ -7781,6 +7787,11 @@
"message" : [
"The <className> sink is currently not supported. See the Real-Time
Mode User Guide for a list of supported sinks."
]
+ },
+ "SQL_CONFIGURATION_NOT_SUPPORTED" : {
+ "message" : [
+ "The following session configuration(s) are incompatible with
Real-Time Mode: <invalidReasons>. Update or remove them and restart the query."
+ ]
}
},
"sqlState" : "0A000"
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 38bf897d9ba9..8803546e53f8 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3624,6 +3624,21 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(1)
+ val STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1 =
+
buildConf("spark.sql.streaming.realTimeMode.dangerouslyAllowCheckpointV1.enabled")
+ .internal()
+ .doc("Whether to allow a Real-Time Mode query to start on a checkpoint
whose commit log " +
+ "is at version 1. Real-Time Mode re-executes a failed batch, and with
checkpoint format " +
+ "version 1 the re-execution can reuse the state file names of the
partially-written " +
+ "failed batch, so starting on a version 1 checkpoint exposes the query
to data loss on " +
+ "failure. Format version 2 avoids this with per-batch state store
checkpoint ids, which " +
+ "only a commit log at version 2 or above can persist. Escape hatch
only; prefer a fresh " +
+ "checkpoint location.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
val STREAMING_MAX_NUM_STATE_SCHEMA_FILES =
buildConf("spark.sql.streaming.stateStore.maxNumStateSchemaFiles")
.internal()
@@ -8787,6 +8802,14 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def streamingOffsetLogFormatVersion: Int =
getConf(STREAMING_OFFSET_LOG_FORMAT_VERSION)
+ // The commit log format version implied by the session config for a fresh
checkpoint. A state
+ // store checkpoint format of v2 makes each batch write stateUniqueIds,
which only a commit log at
+ // VERSION_2 or above can persist, so the commit log version tracks the
state store format. It is
+ // capped at VERSION_2 here: VERSION_3 exists only to carry sink-evolution
metadata and is written
+ // exclusively by the sink-evolution path in MicroBatchExecution, never
derived from a config.
+ def streamingCommitLogFormatVersion: Int =
+ if (getConf(STATE_STORE_CHECKPOINT_FORMAT_VERSION) >= 2) 2 else 1
+
def stateStoreEncodingFormat: String =
getConf(STREAMING_STATE_STORE_ENCODING_FORMAT)
def streamingValueStateSchemaEvolutionThreshold: Int =
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
index c934be69bc69..4755cfa9960a 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
@@ -35,7 +35,7 @@ import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.execution.streaming.continuous.ContinuousExecution
import
org.apache.spark.sql.execution.streaming.runtime.{AsyncProgressTrackingMicroBatchExecution,
MicroBatchExecution, StreamingQueryListenerBus, StreamingQueryWrapper}
-import org.apache.spark.sql.execution.streaming.state.StateStoreCoordinatorRef
+import
org.apache.spark.sql.execution.streaming.state.{RocksDBStateStoreProvider,
StateStoreCoordinatorRef}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.StaticSQLConf.STREAMING_QUERY_LISTENERS
import org.apache.spark.sql.streaming
@@ -186,6 +186,54 @@ class StreamingQueryManager private[sql] (
SQLConf.STREAMING_ASYNC_PROGRESS_TRACKING_REAL_TIME_MODE_ENABLED_BY_DEFAULT))
}
+ /**
+ * Rejects, before the query is built, session configurations that are
incompatible with
+ * Real-Time Mode. Real-Time Mode defaults these when the user has not set
them
+ * (see StreamExecution.setSparkSessionConfigsForRealTimeMode), but an
explicit incompatible
+ * value is a mistake worth surfacing up front rather than silently
overriding or failing later
+ * mid-query. Mirrors the Databricks runtime's
throwIfConfsAreRTMIncompatible, limited to the
+ * checks that apply in OSS.
+ *
+ * - state store checkpoint format below v2 (unless the v1 escape hatch is
set): Real-Time Mode
+ * requires v2 (see the fail-fast in
MicroBatchExecution.initializeExecution);
+ * - a non-RocksDB state store provider: v2 requires RocksDB;
+ * - sortBeforeRepartition left true: the blocking sort never completes on
an unbounded stream.
+ */
+ private def throwIfConfsAreRealTimeModeIncompatible(sparkSession:
SparkSession): Unit = {
+ val conf = sparkSession.sessionState.conf
+ // Preserve declaration order for a stable message.
+ val invalidReasons = new mutable.LinkedHashMap[String, String]
+
+ val allowCheckpointV1 =
+
conf.getConf(SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1)
+ if (!allowCheckpointV1 &&
+ conf.contains(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key) &&
+ conf.getConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION) < 2) {
+ invalidReasons += (SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
"2 or above")
+ }
+
+ if (conf.contains(SQLConf.STATE_STORE_PROVIDER_CLASS.key) &&
+ conf.getConf(SQLConf.STATE_STORE_PROVIDER_CLASS) !=
+ classOf[RocksDBStateStoreProvider].getName) {
+ invalidReasons +=
+ (SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
classOf[RocksDBStateStoreProvider].getName)
+ }
+
+ if (conf.contains(SQLConf.SORT_BEFORE_REPARTITION.key) &&
+ conf.getConf(SQLConf.SORT_BEFORE_REPARTITION)) {
+ invalidReasons += (SQLConf.SORT_BEFORE_REPARTITION.key -> "false")
+ }
+
+ if (invalidReasons.nonEmpty) {
+ throw new SparkIllegalArgumentException(
+ errorClass =
"STREAMING_REAL_TIME_MODE.SQL_CONFIGURATION_NOT_SUPPORTED",
+ messageParameters = Map(
+ "invalidReasons" -> invalidReasons.zipWithIndex.map {
+ case ((confName, req), index) => s"${index + 1}. $confName must be
$req"
+ }.mkString("; ")))
+ }
+ }
+
// scalastyle:off argcount
private def createQuery(
userSpecifiedName: Option[String],
@@ -217,6 +265,7 @@ class StreamingQueryManager private[sql] (
)
)
}
+ throwIfConfsAreRealTimeModeIncompatible(sparkSession)
}
val dataStreamWritePlan = WriteToStreamStatement(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CheckpointVersionManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CheckpointVersionManager.scala
index 4f41f7cc46cf..7f7a614e9d33 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CheckpointVersionManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CheckpointVersionManager.scala
@@ -35,6 +35,7 @@ case class StreamingCheckpointVersion(offsetLogVersion: Int) {
sealed trait CheckpointLogType
case object OffsetLogType extends CheckpointLogType
+case object CommitLogType extends CheckpointLogType
/**
* The `CheckpointVersionManager` is responsible for managing the versioning
of the streaming
@@ -109,6 +110,35 @@ object CheckpointVersionManager extends Logging {
logType: CheckpointLogType): Int = {
logType match {
case OffsetLogType => getOffsetLogVersion(sparkSessionForStream)
+ case CommitLogType => getCommitLogVersion(sparkSessionForStream)
+ }
+ }
+
+ /**
+ * The commit log format version requested by the session config.
+ * `streamingCommitLogFormatVersion` tracks
[[SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION]]: a
+ * state store checkpoint format of v2 makes each batch write
`stateUniqueIds`, which only a
+ * commit log at [[CommitLog.VERSION_2]] or above can persist, so a v2 state
store format raises
+ * the commit log version to v2.
+ */
+ private def getCommitLogVersion(sparkSessionForStream: SparkSession): Int = {
+ val result =
sparkSessionForStream.sessionState.conf.streamingCommitLogFormatVersion
+ logInfo(s"Retrieved commit log writer version=$result")
+ result
+ }
+
+ /**
+ * Determines the commit log format version for ordinary writes in this
query run. An existing
+ * checkpoint wins, so a session config change cannot start writing a format
the checkpoint was
+ * not created with. Only a fresh checkpoint takes the version from the
session config. Sink
+ * evolution may independently upgrade writes to [[CommitLog.VERSION_3]].
+ */
+ def resolveCommitLogVersion(
+ sparkSessionForStream: SparkSession,
+ latestCommittedBatch: Option[(Long, CommitMetadataBase)]): Int = {
+ latestCommittedBatch match {
+ case Some((_, commitMetadata)) => commitMetadata.version
+ case None => getFormatVersionFromSession(sparkSessionForStream,
CommitLogType)
}
}
@@ -142,10 +172,38 @@ object CheckpointVersionManager extends Logging {
def setFormatVersion(
sparkSessionForStream: SparkSession,
logType: CheckpointLogType,
- version: Int): Unit = {
+ version: Int,
+ commitMetadata: Option[CommitMetadataBase] = None): Unit = {
logType match {
case OffsetLogType =>
setSparkSessionConfigsForOffsetLog(sparkSessionForStream, version)
+ case CommitLogType =>
+ setSparkSessionConfigsForCommitLog(sparkSessionForStream, version,
commitMetadata)
+ }
+ }
+
+ /**
+ * Records the state store checkpoint format used by the existing commit
log. VERSION_1 uses
+ * state store format v1 and VERSION_2 uses v2. VERSION_3 can represent
either format because it
+ * was introduced independently for sink metadata; the presence of state
store checkpoint ids
+ * distinguishes v2 from v1.
+ *
+ * The state store format is clamped to 2 rather than set to the commit log
version, because the
+ * two version spaces are separate: the commit log has a VERSION_3 (sink
metadata) with no state
+ * store counterpart, while the state store config accepts only versions 1
and 2. Writing 3 here
+ * would report a state store format that does not exist.
+ */
+ private def setSparkSessionConfigsForCommitLog(
+ sparkSessionForStream: SparkSession,
+ commitLogFormatVersion: Int,
+ commitMetadata: Option[CommitMetadataBase]): Unit = {
+ val stateStoreVersion = commitMetadata match {
+ case Some(metadata) if metadata.version == CommitLog.VERSION_3 =>
+ if (metadata.stateUniqueIds.isDefined) 2 else 1
+ case _ =>
+ if (commitLogFormatVersion >= CommitLog.VERSION_2) 2 else 1
}
+ sparkSessionForStream.conf
+ .set(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key,
stateStoreVersion.toString)
}
}
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 b5271f664cd7..59ad6b986b0a 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
@@ -28,7 +28,6 @@ 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
/**
* Used to write log files that represent batch commit points in structured
streaming.
@@ -56,11 +55,6 @@ class CommitLog(
import CommitLog._
- // 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): CommitMetadataBase
= {
CommitLog.readCommitMetadata(in)
}
@@ -76,7 +70,10 @@ 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]].
+ *
+ * [[commitLogFormatVersion]] is a required parameter rather than a field
read from the session
+ * config, so that a caller always supplies the version it resolved for this
query run. Reading
+ * the config here would ignore the format an existing checkpoint was
created with.
*
* For [[VERSION_3]], [[sinkMetadataMap]] must be non-empty and contain
exactly one active
* sink; [[CommitMetadataV3]] enforces this invariant.
@@ -85,7 +82,7 @@ class CommitLog(
nextBatchWatermarkMs: Long = 0,
stateUniqueIds: Option[Map[Long, Array[Array[String]]]] = None,
sinkMetadataMap: Map[String, SinkMetadataInfo] = Map.empty,
- commitLogFormatVersion: Int = defaultVersion): CommitMetadataBase = {
+ commitLogFormatVersion: Int): CommitMetadataBase = {
commitLogFormatVersion match {
case VERSION_3 =>
CommitMetadataV3(nextBatchWatermarkMs, stateUniqueIds, sinkMetadataMap)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
index d779f48f5e6e..a87c4c3574d1 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.streaming.WriteToStream
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.streaming.{AvailableNowTrigger,
OneTimeTrigger, ProcessingTimeTrigger, RealTimeTrigger, StreamingErrors}
-import org.apache.spark.sql.execution.streaming.checkpointing.{AsyncCommitLog,
AsyncOffsetSeqLog, CommitMetadata, OffsetSeqBase, OffsetSeqLog}
+import org.apache.spark.sql.execution.streaming.checkpointing.{AsyncCommitLog,
AsyncOffsetSeqLog, CommitMetadataBase, OffsetSeqBase, OffsetSeqLog}
import
org.apache.spark.sql.execution.streaming.operators.stateful.StateStoreWriter
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.util.{Clock, ThreadUtils}
@@ -261,7 +261,7 @@ class AsyncProgressTrackingMicroBatchExecution(
|| isFirstBatch) {
isFirstBatch = false
commitLog
- .addAsync(execCtx.batchId,
CommitMetadata(watermarkTracker.currentWatermark))
+ .addAsync(execCtx.batchId, createAsyncCommitMetadata(execCtx))
.thenAccept((batchId: Long) => {
logInfo(log"Committed async commit log to disk for batch
${MDC(BATCH_ID, batchId)}.")
})
@@ -273,7 +273,7 @@ class AsyncProgressTrackingMicroBatchExecution(
})
} else {
if (!commitLog.addInMemory(
- execCtx.batchId, CommitMetadata(watermarkTracker.currentWatermark)))
{
+ execCtx.batchId, createAsyncCommitMetadata(execCtx))) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
logInfo(
@@ -285,6 +285,21 @@ class AsyncProgressTrackingMicroBatchExecution(
committedOffsets ++= execCtx.endOffsets
}
+ /**
+ * Builds the commit log entry for an async batch at the version resolved
for this run, so that a
+ * Real-Time Mode query writes the commit log format its state store
checkpoint format implies
+ * (v2 for the RTM default) rather than always VERSION_1. Async progress
tracking does not support
+ * stateful queries, so there are no state store checkpoint ids to persist.
+ */
+ private def createAsyncCommitMetadata(
+ execCtx: MicroBatchExecutionContext): CommitMetadataBase = {
+ commitLog.createMetadata(
+ nextBatchWatermarkMs = watermarkTracker.currentWatermark,
+ stateUniqueIds = None,
+ commitLogFormatVersion = execCtx.commitLogFormatVersionOpt.getOrElse(
+
sparkSessionForStream.sessionState.conf.streamingCommitLogFormatVersion))
+ }
+
/**
* Categorize a raw IO failure surfaced via an async log-write future, then
route it through
* the standard async error handling path. CompletableFuture wraps the
underlying cause in
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 8621271768f4..f0201ccf302a 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,
CheckpointVersionManager, CommitLog, CommitMetadataV3, OffsetLogType,
OffsetSeqBase, OffsetSeqLog, OffsetSeqMetadata, OffsetSeqMetadataV2,
SinkMetadataInfo}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
CheckpointVersionManager, CommitLog, CommitLogType, CommitMetadataV3,
OffsetLogType, OffsetSeqBase, OffsetSeqLog, OffsetSeqMetadata,
OffsetSeqMetadataV2, SinkMetadataInfo}
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}
@@ -82,6 +82,7 @@ class MicroBatchExecution(
-1,
sparkSession,
offsetLogFormatVersionOpt = None,
+ commitLogFormatVersionOpt = None,
previousContext = None)
override def getLatestExecutionContext(): StreamExecutionContext =
latestExecutionContext
@@ -371,31 +372,6 @@ class MicroBatchExecution(
messageParameters = Map("className" -> s.getClass.getName)
)
}
-
- // `repartition(n)` uses RoundRobinPartitioning, and by default
(SPARK-23207) Spark inserts a
- // local sort before a round-robin shuffle so the row-to-partition
assignment is deterministic
- // -- otherwise a retried task could emit rows in a different order and
lose data. That sort
- // is fully blocking: it drains its whole input before emitting a row. A
Real-Time Mode task
- // reads an unbounded stream, so the input never ends, the producer
never emits, and the query
- // makes no progress at all. Note this sort is not a SortExec node (it
lives inside
- // ShuffleExchangeExec's RDD), so the Real-Time Mode operator allowlist
cannot catch it.
- //
- // Determinism is not needed here: Real-Time Mode does not retry tasks
(TaskSetManager caps a
- // pipelined task set at one attempt, and its group is aborted as a unit
rather than
- // recomputed), so the hazard the sort guards against does not arise.
- //
- // This overrides an explicit `true` as well. Honouring it would
reintroduce the hang above,
- // and a query that never produces a row is worse than silently ignoring
a config that only
- // guards against a retry that cannot happen here. Log at warning level
when overriding an
- // explicit setting so the operator can see their config is not the one
in force.
- if
(sparkSessionForStream.conf.contains(SQLConf.SORT_BEFORE_REPARTITION.key) &&
- sparkSessionForStream.conf.get(SQLConf.SORT_BEFORE_REPARTITION)) {
- logWarning(log"Ignoring ${MDC(LogKeys.CONFIG,
SQLConf.SORT_BEFORE_REPARTITION.key)}=true " +
- log"for this Real-Time Mode query: the local sort it inserts before
a round-robin " +
- log"repartition never completes on an unbounded stream. It is not
needed because a " +
- log"Real-Time Mode task is not retried.")
- }
- sparkSessionForStream.conf.set(SQLConf.SORT_BEFORE_REPARTITION.key,
"false")
}
// Initializing TriggerExecutor relies on `sources`, hence calling this
after initializing
@@ -579,9 +555,47 @@ class MicroBatchExecution(
CheckpointVersionManager.setFormatVersion(
sparkSessionForStream, OffsetLogType, offsetLogFormatVersion)
+ // Resolve the commit log format version the same way: an existing
checkpoint keeps the version
+ // it was created with, and only a fresh one takes the version from the
session config. Persist
+ // the implied state store checkpoint format so it agrees with what is
actually being written.
+ val commitLogFormatVersion =
CheckpointVersionManager.resolveCommitLogVersion(
+ sparkSessionForStream, latestCommittedBatch)
+ CheckpointVersionManager.setFormatVersion(
+ sparkSessionForStream,
+ CommitLogType,
+ commitLogFormatVersion,
+ latestCommittedBatch.map(_._2))
+
+ // Real-Time Mode requires commit log v2. Real-Time Mode writes the offset
log at batch end
+ // (markMicroBatchStart is a no-op for it), so a mid-batch failure can
leave durable state at a
+ // version that was never logged; the re-execution then rewrites that same
state version. With
+ // checkpoint format v1 the rewritten files reuse the same names as the
orphaned ones, so a load
+ // can pick up a stale file (see the checksum hazard documented on
+ // StateStoreConf.skipChecksumOnFileMissingChecksum). Format v2 avoids
this because each batch
+ // run generates unique state store checkpoint ids, and only a commit log
at v2 or above can
+ // persist them. Resolution above keeps an existing checkpoint at the
version it was created
+ // with, so a v1 checkpoint stays v1. Reject any Real-Time Mode query
whose resolved commit log
+ // version is below v2, with an escape hatch. This is unconditional,
matching the Databricks
+ // runtime -- a fresh checkpoint reaches v1 here only when the user
explicitly pinned it, which
+ // is exactly the case worth rejecting.
+ if (trigger.isInstanceOf[RealTimeTrigger] && commitLogFormatVersion <
CommitLog.VERSION_2) {
+ if (!sparkSessionForStream.sessionState.conf
+
.getConf(SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1)) {
+ throw new SparkIllegalArgumentException(
+ errorClass =
"STREAMING_REAL_TIME_MODE.CHECKPOINT_FORMAT_V1_NOT_SUPPORTED",
+ messageParameters = Map(
+ "config" ->
SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1.key))
+ }
+ logWarning(log"Starting a Real-Time Mode query on a commit log at
version " +
+ log"${MDC(LogKeys.FILE_VERSION, commitLogFormatVersion)} because " +
+ log"${MDC(LogKeys.CONFIG,
SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1
+ .key)} is set. A failed batch may lose data on rerun.")
+ }
+
val execCtx = new MicroBatchExecutionContext(id, runId, name,
triggerClock, sources, sink,
progressReporter, -1, sparkSession,
offsetLogFormatVersionOpt = Some(offsetLogFormatVersion),
+ commitLogFormatVersionOpt = Some(commitLogFormatVersion),
previousContext = None)
execCtx.offsetSeqMetadata = offsetLogFormatVersion match {
@@ -1524,7 +1538,9 @@ class MicroBatchExecution(
} else {
commitLog.createMetadata(
nextBatchWatermarkMs = watermarkTracker.currentWatermark,
- stateUniqueIds = stateStoreCkptId)
+ stateUniqueIds = stateStoreCkptId,
+ commitLogFormatVersion = execCtx.commitLogFormatVersionOpt.getOrElse(
+
sparkSessionForStream.sessionState.conf.streamingCommitLogFormatVersion))
}
if (!commitLog.add(execCtx.batchId, metadata)) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala
index 38b4a58385eb..cef636cc58b1 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala
@@ -34,7 +34,7 @@ import org.apache.logging.log4j.CloseableThreadContext
import org.apache.spark.{JobArtifactSet, SparkContext, SparkException,
SparkThrowable}
import org.apache.spark.internal.Logging
-import org.apache.spark.internal.LogKeys.{CHECKPOINT_PATH, CHECKPOINT_ROOT,
LOGICAL_PLAN, PATH, PRETTY_ID_STRING, QUERY_ID, RUN_ID, SPARK_DATA_STREAM}
+import org.apache.spark.internal.LogKeys.{CHECKPOINT_PATH, CHECKPOINT_ROOT,
CONFIG, LOGICAL_PLAN, PATH, PRETTY_ID_STRING, QUERY_ID, RUN_ID,
SPARK_DATA_STREAM}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.streaming.InternalOutputModes._
import org.apache.spark.sql.classic.{SparkSession, StreamingQuery}
@@ -43,10 +43,11 @@ import
org.apache.spark.sql.connector.read.streaming.{Offset => OffsetV2, ReadLi
import org.apache.spark.sql.connector.write.{LogicalWriteInfoImpl,
SupportsTruncate, Write}
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command.StreamingExplainCommand
-import org.apache.spark.sql.execution.streaming.ContinuousTrigger
+import org.apache.spark.sql.execution.streaming.{ContinuousTrigger,
RealTimeTrigger}
import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointFileManager,
CommitLog, OffsetSeqLog, OffsetSeqMetadata}
import
org.apache.spark.sql.execution.streaming.operators.stateful.{StatefulOperator,
StateStoreWriter}
import
org.apache.spark.sql.execution.streaming.sources.{ForeachBatchUserFuncException,
ForeachUserFuncException}
+import org.apache.spark.sql.execution.streaming.state.{RocksDBConf,
RocksDBStateStoreProvider}
import
org.apache.spark.sql.execution.streaming.state.OperatorStateMetadataV2FileManager
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.connector.SupportsStreamingUpdateAsAppend
@@ -325,6 +326,10 @@ abstract class StreamExecution(
sparkSessionForStream.conf.set(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, "false")
}
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ setSparkSessionConfigsForRealTimeMode(sparkSessionForStream)
+ }
+
sparkSessionForStream.conf.get(SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL)
match {
case Some(_) => // no-op
case None =>
@@ -458,6 +463,86 @@ abstract class StreamExecution(
}
}
+ /**
+ * Applies the configuration a Real-Time Mode query needs but that is not
the engine-wide default,
+ * because it is only the right choice for a low-latency, long-running
batch. Runs once at query
+ * start, before the logical plan is forced, so a config read during
planning sees the final
+ * value.
+ *
+ * Every setting here is a SOFT DEFAULT: applied only when the user has not
set the key, so an
+ * explicit choice always wins. The state store settings,
`changelogCheckpointing`, and
+ * `sortBeforeRepartition` are these. Except for checkpoint format v1 when
its escape hatch is
+ * enabled, an explicit value that is incompatible with Real-Time Mode is
rejected up front by the
+ * preflight in StreamingQueryManager
(throwIfConfsAreRealTimeModeIncompatible). Therefore, an
+ * explicit value that reaches this method is safe to keep or was explicitly
allowed.
+ *
+ * Every default applied here is logged, so those changes are recoverable
from the driver log.
+ *
+ * Deliberately not set, though Databricks Runtime does default them for
Real-Time Mode:
+ * - The incremental state-cleanup factor: that mechanism does not exist in
OSS yet.
+ * - The Python/Pandas UDF latency knobs: those configs do not exist in OSS.
+ */
+ private def setSparkSessionConfigsForRealTimeMode(sparkSessionForStream:
SparkSession): Unit = {
+ val conf = sparkSessionForStream.conf
+
+ // SOFT DEFAULTS. Real-Time Mode benefits from state store checkpoint
format v2: it gives each
+ // batch run its own state store checkpoint ids, which prevents a
re-executed batch from reusing
+ // the state file names of a partially-written failed batch (see the v1
hazard described at the
+ // fail-fast in MicroBatchExecution.initializeExecution). v2 requires the
RocksDB state store
+ // provider. These are defaulted with two INDEPENDENT guards, matching the
Databricks runtime:
+ // each key is set only if the user has not set that key. An explicit
non-RocksDB provider has
+ // already been rejected by the pre-flight in StreamingQueryManager
+ // (throwIfConfsAreRealTimeModeIncompatible), so by the time this runs the
provider is either
+ // unset (defaulted to RocksDB just below) or already RocksDB; the version
default is applied
+ // independently of the provider default.
+ val checkpointVersionKey =
SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key
+ if (!conf.contains(checkpointVersionKey)) {
+ logInfo(log"Real-Time Mode: defaulting ${MDC(CONFIG,
checkpointVersionKey)}=2")
+ conf.set(checkpointVersionKey, CommitLog.VERSION_2.toString)
+ }
+ val providerKey = SQLConf.STATE_STORE_PROVIDER_CLASS.key
+ if (!conf.contains(providerKey)) {
+ logInfo(log"Real-Time Mode: defaulting " +
+ log"${MDC(CONFIG, providerKey)}=RocksDBStateStoreProvider")
+ conf.set(providerKey, classOf[RocksDBStateStoreProvider].getName)
+ }
+
+ // SOFT DEFAULT. Changelog checkpointing writes a changelog rather than a
full snapshot on each
+ // commit, which shortens the state-commit step at a batch boundary. That
step is on the
+ // critical path between batches in Real-Time Mode, so this is a latency
optimization. It is
+ // only meaningful with RocksDB, hence the check against the provider
actually in force.
+ val changelogKey =
s"${RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX}.changelogCheckpointing.enabled"
+ val usingRocksDb =
+ conf.get(SQLConf.STATE_STORE_PROVIDER_CLASS.key,
+ SQLConf.STATE_STORE_PROVIDER_CLASS.defaultValueString) ==
+ classOf[RocksDBStateStoreProvider].getName
+ if (usingRocksDb && !conf.contains(changelogKey)) {
+ logInfo(log"Real-Time Mode: defaulting ${MDC(CONFIG,
changelogKey)}=true")
+ conf.set(changelogKey, "true")
+ }
+
+ // SOFT DEFAULT. `repartition(n)` uses RoundRobinPartitioning, and by
default (SPARK-23207)
+ // Spark inserts a local sort before a round-robin shuffle so the
row-to-partition assignment is
+ // deterministic -- otherwise a retried task could emit rows in a
different order and lose data.
+ // That sort is fully blocking: it drains its whole input before emitting
a row. A Real-Time
+ // Mode task reads an unbounded stream, so the input never ends, the
producer never emits, and
+ // the query makes no progress at all. Note this sort is not a SortExec
node (it lives inside
+ // ShuffleExchangeExec's RDD), so the Real-Time Mode operator allowlist
cannot catch it.
+ //
+ // Determinism is not needed here: Real-Time Mode does not retry tasks
(TaskSetManager caps a
+ // pipelined task set at one attempt, and its group is aborted as a unit
rather than
+ // recomputed), so the hazard the sort guards against does not arise.
+ //
+ // This is a soft default rather than an override: an explicit `true` is
rejected up front by
+ // the pre-flight in StreamingQueryManager
(throwIfConfsAreRealTimeModeIncompatible), so by the
+ // time we get here an explicit value can only be `false`. Matches the
Databricks runtime.
+ if (!conf.contains(SQLConf.SORT_BEFORE_REPARTITION.key)) {
+ logInfo(log"Real-Time Mode: defaulting " +
+ log"${MDC(CONFIG, SQLConf.SORT_BEFORE_REPARTITION.key)}=false")
+ conf.set(SQLConf.SORT_BEFORE_REPARTITION.key, "false")
+ }
+ }
+
private def isInterruptedByStop(e: Throwable, sc: SparkContext): Boolean = {
if (state.get == TERMINATED) {
StreamExecution.isInterruptionException(e, sc)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecutionContext.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecutionContext.scala
index a7061d14d334..1c2fa064ca88 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecutionContext.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecutionContext.scala
@@ -128,6 +128,7 @@ class MicroBatchExecutionContext(
var _batchId: Long,
sparkSession: SparkSession,
val offsetLogFormatVersionOpt: Option[Int],
+ val commitLogFormatVersionOpt: Option[Int],
var previousContext: Option[MicroBatchExecutionContext])
extends StreamExecutionContext(
id,
@@ -192,6 +193,7 @@ class MicroBatchExecutionContext(
batchId + 1,
sparkSession,
offsetLogFormatVersionOpt,
+ commitLogFormatVersionOpt,
Some(this))
}
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 546a9a601964..3280671851e3 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
@@ -381,7 +381,8 @@ class StateRewriter(
// 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
+ val configuredVersion =
+
sparkSession.sessionState.conf.getConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION)
if (metadata.version != configuredVersion) {
throw StateRewriterErrors.stateCheckpointFormatVersionMismatchError(
checkpointLocationForRead,
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 ba785aa830ab..a733bd6d7571 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, CommitMetadataV3,
OffsetSeqLog, SinkMetadataInfo}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{CheckpointVersionManager,
CommitLog, CommitLogType, CommitMetadata, CommitMetadataBase,
CommitMetadataV2, CommitMetadataV3, OffsetSeqLog, SinkMetadataInfo}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -250,4 +250,70 @@ class CommitLogSuite extends SharedSparkSession {
commitLogFormatVersion = CommitLog.VERSION_1).version ===
CommitLog.VERSION_1)
}
}
+
+ /** The commit log version the session config asks for, via the public
resolution entry point. */
+ private def sessionCommitLogVersion(): Int = {
+ CheckpointVersionManager.resolveCommitLogVersion(spark,
latestCommittedBatch = None)
+ }
+
+ test("commit log version derives from the state store checkpoint format") {
+ // Nothing set: defaults to VERSION_1.
+ assert(sessionCommitLogVersion() === CommitLog.VERSION_1)
+
+ // State store checkpoint format v2 makes each batch write stateUniqueIds,
which only a commit
+ // log at VERSION_2 or above can persist, so it raises the commit log
version to v2.
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
+ assert(sessionCommitLogVersion() === CommitLog.VERSION_2,
+ "state store v2 must raise the commit log to v2")
+ }
+
+ // The resolved version is capped at VERSION_2: VERSION_3 exists only to
carry sink-evolution
+ // metadata and is written exclusively by the sink-evolution path, never
derived from a config.
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "3") {
+ assert(sessionCommitLogVersion() === CommitLog.VERSION_2,
+ "a config-derived commit log version must never resolve to v3")
+ }
+ }
+
+ test("an existing checkpoint's commit log version wins over the session
config") {
+ // The whole point of resolution: a commit log created at one version
keeps being written at
+ // that version, so a higher state store checkpoint format cannot start
writing a format the
+ // checkpoint lacks.
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
+ val existing: CommitMetadataBase = CommitMetadata(nextBatchWatermarkMs =
0)
+ assert(existing.version === CommitLog.VERSION_1)
+ val resolved =
+ CheckpointVersionManager.resolveCommitLogVersion(spark, Some((7L,
existing)))
+ assert(resolved === CommitLog.VERSION_1,
+ s"an existing V1 commit log must stay V1 even with the state store at
v2, got $resolved")
+ }
+ }
+
+ test("recording a commit log version sets the implied state store format") {
+ val sinkMetadataMap = Map("sink" -> SinkMetadataInfo(
+ sinkName = "sink",
+ commitOffset = OffsetSeqLog.SERIALIZED_VOID_OFFSET,
+ providerName = "provider",
+ apiVersion = "DSv2"))
+ val session = spark.cloneSession()
+ val v3WithStateIds = CommitMetadataV3(0, Some(Map.empty), sinkMetadataMap)
+ CheckpointVersionManager.setFormatVersion(
+ session, CommitLogType, CommitLog.VERSION_3, Some(v3WithStateIds))
+ assert(session.conf.get(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key)
=== "2",
+ "state checkpoint ids in a V3 commit imply state store format v2")
+
+ val v3WithoutStateIdsSession = spark.cloneSession()
+ val v3WithoutStateIds = CommitMetadataV3(0, None, sinkMetadataMap)
+ CheckpointVersionManager.setFormatVersion(
+ v3WithoutStateIdsSession, CommitLogType, CommitLog.VERSION_3,
Some(v3WithoutStateIds))
+ assert(v3WithoutStateIdsSession.conf.get(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key) === "1",
+ "a V3 commit without state checkpoint ids must preserve state store
format v1")
+
+ val v1Session = spark.cloneSession()
+ CheckpointVersionManager.setFormatVersion(v1Session, CommitLogType,
CommitLog.VERSION_1)
+
assert(v1Session.conf.get(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key)
=== "1",
+ "a V1 commit log cannot carry state store checkpoint ids, so the state
store must be v1")
+ }
+
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeDefaultConfsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeDefaultConfsSuite.scala
new file mode 100644
index 000000000000..5b84f4cfa03d
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeDefaultConfsSuite.scala
@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.{SparkIllegalArgumentException, SparkThrowable}
+import org.apache.spark.sql.execution.streaming.sources.{ContinuousMemorySink,
LowLatencyMemoryStream}
+import
org.apache.spark.sql.execution.streaming.state.{HDFSBackedStateStoreProvider,
RocksDBConf,
+ RocksDBStateStoreProvider}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Tests the configuration defaults a Real-Time Mode query applies at query
start.
+ *
+ * These are not engine-wide defaults because they are only the right choice
for a low-latency,
+ * long-running batch. Each is applied only when the user has not set it, so
an explicit choice
+ * always wins -- both directions are asserted here, since a defaulting block
that silently
+ * overrode a user's setting would be worse than having no default at all.
+ */
+class StreamRealTimeModeDefaultConfsSuite extends StreamRealTimeModeSuiteBase {
+
+ import testImplicits._
+
+ private val changelogKey =
+
s"${RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX}.changelogCheckpointing.enabled"
+
+ /** Runs a trivial RTM query to completion and returns the effective conf
values afterwards. */
+ private def runRealTimeQueryAndReadConfs(keys: Seq[String]): Map[String,
Option[String]] = {
+ val inputData = LowLatencyMemoryStream[Int]
+ val mapped = inputData.toDS().map(_ + 1)
+ var observed: Map[String, Option[String]] = Map.empty
+
+ testStream(mapped, OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ AddData(inputData, 1, 2, 3),
+ StartStream(),
+ CheckAnswerWithTimeout(60000, 2, 3, 4),
+ Execute { q =>
+ // Read from sparkSessionForStream, the isolated CLONE the batches
actually run with
+ // (StreamExecution.sparkSessionForStream). The defaults are applied
there, not to the
+ // caller's session, so reading `q.sparkSession` would observe the
unmodified original.
+ val conf = q.sparkSessionForStream.conf
+ observed = keys.map(k => k -> conf.getOption(k)).toMap
+ },
+ StopStream
+ )
+ observed
+ }
+
+ test("a Real-Time Mode query defaults to checkpoint format v2 with the
RocksDB state store") {
+ val keys = Seq(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key,
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key)
+ val observed = runRealTimeQueryAndReadConfs(keys)
+
+
assert(observed(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).contains("2"),
+ "Real-Time Mode should default the state-store checkpoint format to v2,
got " +
+ observed(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key))
+ assert(observed(SQLConf.STATE_STORE_PROVIDER_CLASS.key)
+ .contains(classOf[RocksDBStateStoreProvider].getName),
+ "Real-Time Mode should default the state-store provider to RocksDB, got
" +
+ observed(SQLConf.STATE_STORE_PROVIDER_CLASS.key))
+ }
+
+ test("the checkpoint-format and provider defaults are applied
independently") {
+ // The two are defaulted by independent guards (matching the runtime):
pinning one does not
+ // suppress the other. Pin only the version (to a compatible v2, since an
incompatible explicit
+ // config is rejected up front by the pre-flight) and confirm the provider
is still defaulted.
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
+ val observed = runRealTimeQueryAndReadConfs(
+ Seq(SQLConf.STATE_STORE_PROVIDER_CLASS.key,
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key))
+
assert(observed(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).contains("2"),
+ "the pinned version must be left alone, got " +
+ observed(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key))
+ assert(observed(SQLConf.STATE_STORE_PROVIDER_CLASS.key)
+ .contains(classOf[RocksDBStateStoreProvider].getName),
+ "the provider default is independent of the version, so it must still
be RocksDB, got " +
+ observed(SQLConf.STATE_STORE_PROVIDER_CLASS.key))
+ }
+ }
+
+ test("an explicit checkpoint format version is not overridden by Real-Time
Mode") {
+ // The defaulting guard respects an explicit version rather than forcing
2. v2 is used here
+ // (not v1) because an explicit v1 is rejected up front by the pre-flight
(covered below).
+ withSQLConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "2") {
+ val observed = runRealTimeQueryAndReadConfs(
+ Seq(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key))
+
assert(observed(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key).contains("2"),
+ "an explicitly configured checkpoint format version must be left
alone, got " +
+ observed(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key))
+ }
+ }
+
+ test("an explicit incompatible config is rejected up front for a Real-Time
Mode query") {
+ // The pre-flight in StreamingQueryManager rejects explicit
RTM-incompatible session configs
+ // before the query is built: checkpoint format below v2, a non-RocksDB
provider, and
+ // sortBeforeRepartition=true. Each is reported in one
SQL_CONFIGURATION_NOT_SUPPORTED error.
+ def assertRejected(confs: (String, String)*): Unit = {
+ withSQLConf(confs: _*) {
+ val inputData = LowLatencyMemoryStream[Int]
+ val e = intercept[SparkIllegalArgumentException] {
+ testStream(inputData.toDS(), OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ StartStream())
+ }
+ checkError(e, condition =
"STREAMING_REAL_TIME_MODE.SQL_CONFIGURATION_NOT_SUPPORTED",
+ parameters = e.getMessageParameters.asScala.toMap)
+ }
+ }
+ assertRejected(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "1")
+ assertRejected(
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
classOf[HDFSBackedStateStoreProvider].getName)
+ assertRejected(SQLConf.SORT_BEFORE_REPARTITION.key -> "true")
+ }
+
+ test("the checkpoint-v1 escape hatch bypasses the pre-flight version check")
{
+ // With the escape hatch on, an explicit v1 config is permitted through
the pre-flight (the
+ // existing-v1-checkpoint fail-fast in initializeExecution is covered
separately below).
+ withSQLConf(
+ SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1.key ->
"true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "1") {
+ val inputData = LowLatencyMemoryStream[Int]
+ testStream(inputData.toDS(), OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ AddData(inputData, 1),
+ StartStream(),
+ CheckAnswerWithTimeout(60000, 1),
+ StopStream
+ )
+ }
+ }
+
+ test("a Real-Time Mode query defaults changelog checkpointing on with
RocksDB") {
+ withSQLConf(
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
classOf[RocksDBStateStoreProvider].getName) {
+ val observed = runRealTimeQueryAndReadConfs(Seq(changelogKey))
+ assert(observed(changelogKey).contains("true"),
+ s"Real-Time Mode should default $changelogKey to true, got
${observed(changelogKey)}")
+ }
+ }
+
+ test("an explicit changelog checkpointing setting is not overridden by
Real-Time Mode") {
+ // The RocksDB provider must be in force, otherwise the `usingRocksDb`
gate skips the block and
+ // the guard under test is never reached -- the assertion would hold
vacuously.
+ withSQLConf(
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
classOf[RocksDBStateStoreProvider].getName,
+ changelogKey -> "false") {
+ val observed = runRealTimeQueryAndReadConfs(Seq(changelogKey))
+ assert(observed(changelogKey).contains("false"),
+ s"an explicitly configured $changelogKey must be left alone, got
${observed(changelogKey)}")
+ }
+ }
+
+ test("switching an existing v1 checkpoint to Real-Time Mode fails fast") {
+ // An existing v1 checkpoint carries no explicit config, so the pre-flight
cannot see it; the
+ // initializeExecution fail-fast catches it instead, from the resolved
commit log version.
+ // Resolution keeps an existing checkpoint at the version it was created
with, so rather than
+ // silently running at v1 (where a re-executed batch can reuse the failed
batch's state file
+ // names and lose data) the query is rejected at start. The rejection is
unconditional -- a
+ // plain (stateless) query is rejected too.
+ withTempDir { checkpointDir =>
+ val inputData = LowLatencyMemoryStream[Int]
+ val mapped = inputData.toDS().map(_ + 1)
+ testStream(mapped, OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ AddData(inputData, 1),
+ // Phase 1: a micro-batch trigger, which writes a v1 commit log.
+ StartStream(
+ trigger = Trigger.ProcessingTime("1 second"),
+ checkpointLocation = checkpointDir.getAbsolutePath),
+ WaitUntilCurrentBatchProcessed,
+ StopStream,
+ AddData(inputData, 2),
+ // Phase 2: the same checkpoint under the Real-Time trigger must be
rejected.
+ StartStream(checkpointLocation = checkpointDir.getAbsolutePath),
+ ExpectFailure[SparkIllegalArgumentException] { e =>
+ checkError(
+ e.asInstanceOf[SparkThrowable],
+ condition =
"STREAMING_REAL_TIME_MODE.CHECKPOINT_FORMAT_V1_NOT_SUPPORTED",
+ parameters = Map(
+ "config" ->
SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1.key))
+ }
+ )
+ }
+ }
+
+ test("the escape hatch allows Real-Time Mode on an existing v1 checkpoint") {
+ withSQLConf(
+ SQLConf.STREAMING_REAL_TIME_MODE_DANGEROUSLY_ALLOW_CHECKPOINT_V1.key ->
"true") {
+ withTempDir { checkpointDir =>
+ val inputData = LowLatencyMemoryStream[Int]
+ val mapped = inputData.toDS().map(_ + 1)
+ testStream(mapped, OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ AddData(inputData, 1),
+ StartStream(
+ trigger = Trigger.ProcessingTime("1 second"),
+ checkpointLocation = checkpointDir.getAbsolutePath),
+ WaitUntilCurrentBatchProcessed,
+ StopStream,
+ AddData(inputData, 2),
+ StartStream(checkpointLocation = checkpointDir.getAbsolutePath),
+ // The sink accumulates across both phases, so batch 1's row is
still present.
+ CheckAnswerWithTimeout(60000, 2, 3),
+ StopStream
+ )
+ }
+ }
+ }
+
+ test("a fresh Real-Time Mode checkpoint is not rejected") {
+ // A fresh checkpoint takes v2 from the Real-Time Mode defaults, so its
resolved commit log is
+ // v2 and the rejection does not apply. Only a v1 commit log -- an
existing v1 checkpoint, or an
+ // explicit v1 config without the escape hatch -- is rejected.
+ withTempDir { checkpointDir =>
+ val inputData = LowLatencyMemoryStream[Int]
+ testStream(inputData.toDS(), OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ AddData(inputData, 1, 2),
+ StartStream(checkpointLocation = checkpointDir.getAbsolutePath),
+ CheckAnswerWithTimeout(60000, 1, 2),
+ StopStream
+ )
+ }
+ }
+
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
index 689d89e19e00..52b05b6f4fb5 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
@@ -22,6 +22,7 @@ import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
import scala.concurrent.duration.Duration
+import scala.jdk.CollectionConverters._
import org.scalatest.concurrent.PatienceConfiguration.Timeout
@@ -695,10 +696,11 @@ class StreamRealTimeModeWithManualClockSuite extends
StreamRealTimeModeManualClo
}
}
- test("pipelined shuffle: an explicit sortBeforeRepartition=true does not
stall the query") {
+ test("pipelined shuffle: an explicit sortBeforeRepartition=true is
rejected") {
// The deterministic local sort before a round-robin repartition never
drains an unbounded
- // stream, so Real-Time Mode overrides this config even when it is set
explicitly -- honouring
- // it would hang the query forever. See MicroBatchExecution.
+ // stream, so honouring sortBeforeRepartition=true would hang a Real-Time
Mode query forever.
+ // Rather than silently override it, the pre-flight in
StreamingQueryManager rejects the
+ // explicit value up front. See
StreamingQueryManager.throwIfConfsAreRealTimeModeIncompatible.
withSQLConf(
SQLConf.SHUFFLE_PARTITIONS.key -> "2",
SQLConf.SORT_BEFORE_REPARTITION.key -> "true") {
@@ -707,12 +709,14 @@ class StreamRealTimeModeWithManualClockSuite extends
StreamRealTimeModeManualClo
.repartition(4)
.dropDuplicates("key")
.select($"key")
- testStream(result, OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
- AddData(inputData, ("a", 1), ("b", 1), ("c", 1), ("a", 2)),
- StartStream(),
- CheckAnswerWithTimeout(60000, "a", "b", "c"),
- StopStream
- )
+ val e = intercept[SparkIllegalArgumentException] {
+ testStream(result, OutputMode.Update, Map.empty, new
ContinuousMemorySink())(
+ AddData(inputData, ("a", 1), ("b", 1), ("c", 1), ("a", 2)),
+ StartStream()
+ )
+ }
+ checkError(e, condition =
"STREAMING_REAL_TIME_MODE.SQL_CONFIGURATION_NOT_SUPPORTED",
+ parameters = e.getMessageParameters.asScala.toMap)
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
index 46de58fad333..42a2edd016b9 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/StreamingSinkEvolutionSuite.scala
@@ -23,8 +23,9 @@ import org.apache.spark.{SparkException,
SparkIllegalArgumentException}
import org.apache.spark.sql._
import org.apache.spark.sql.execution.streaming.checkpointing.{CommitLog,
CommitMetadataV3}
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import
org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.streaming.{StreamTest, Trigger}
+import org.apache.spark.sql.streaming.{StreamingQuery, StreamTest, Trigger}
import org.apache.spark.util.Utils
/**
@@ -313,6 +314,40 @@ class StreamingSinkEvolutionSuite extends StreamTest with
BeforeAndAfterEach {
assert(v3.sinkMetadataMap.size === 1)
}
+ testWithSinkEvolution("restart V3 commit log with state checkpoint format
V1") {
+ val checkpointDir = newMetadataDir
+ val input = MemoryStream[Int]
+
+ withSQLConf(
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key -> "1",
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
+ classOf[HDFSBackedStateStoreProvider].getName) {
+ def startQuery(): StreamingQuery = input.toDF()
+ .groupBy("value")
+ .count()
+ .writeStream
+ .format("noop")
+ .outputMode("update")
+ .name("stateful_sink")
+ .option("checkpointLocation", checkpointDir)
+ .start()
+
+ input.addData(1)
+ val firstRun = startQuery()
+ firstRun.processAllAvailable()
+ firstRun.stop()
+
+ input.addData(2)
+ val restarted = startQuery()
+ restarted.processAllAvailable()
+ restarted.stop()
+ }
+
+ val commitLog = new CommitLog(spark, s"$checkpointDir/commits", readOnly =
true)
+ val v3 = commitLog.getLatest().get._2.asInstanceOf[CommitMetadataV3]
+ assert(v3.stateUniqueIds.isEmpty)
+ }
+
// ==============
// Helper Methods
// ==============
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
