cloud-fan commented on code in PR #57842:
URL: https://github.com/apache/spark/pull/57842#discussion_r3754708007
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CommitLog.scala:
##########
@@ -76,7 +70,13 @@ 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]].
+ *
+ * The version is a 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
Review Comment:
This says callers always provide the resolved version, but the parameter
still defaults to `VERSION_1`. Please either remove the default so the contract
is enforced, or describe when intentionally omitting the argument is valid.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala:
##########
@@ -458,6 +463,84 @@ 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. An explicit value that is incompatible
with Real-Time Mode
+ * is not overridden here -- it is rejected up front by the preflight in
StreamingQueryManager
+ * (throwIfConfsAreRealTimeModeIncompatible), so by the time this runs an
explicit value is always
+ * a safe one to keep.
+ *
+ * Every change is logged, so a run's effective configuration is 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. If the user pins
the provider to a
+ // non-RocksDB store but leaves the version unset, the version is still
raised to 2 and the
+ // incompatible combination throws later (HDFSBackedStateStoreProvider
rejects
Review Comment:
This combination no longer reaches a later HDFS-backed provider failure: the
new `StreamingQueryManager` preflight rejects an explicit non-RocksDB provider
before this method runs. Please update this paragraph to describe that startup
rejection.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3624,6 +3624,34 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(1)
+ val STREAMING_COMMIT_LOG_FORMAT_VERSION =
+ buildConf("spark.sql.streaming.commitLog.formatVersion")
+ .internal()
+ .doc("The version of the commit log format. The default value is 1. Will
use the max of " +
+ "this and STATE_STORE_CHECKPOINT_FORMAT_VERSION as the commit log
version, since a " +
+ "state store checkpoint format above 1 writes state store checkpoint
ids that only a " +
+ "commit log at version 2 or above can persist.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .intConf
+ .checkValue(v => Set(1, 2, 3).contains(v), "Valid versions are 1, 2 and
3")
Review Comment:
Version 3 is accepted here even when sink evolution is disabled, but the
resolved-version write paths call `createMetadata` without a `sinkMetadataMap`;
`CommitMetadataV3` then throws. Please either restrict this config to v1/v2
unless sink evolution owns the write, or make every v3 path supply the required
sink metadata.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
