jerrypeng commented on code in PR #57842:
URL: https://github.com/apache/spark/pull/57842#discussion_r3755487106
##########
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:
updated
--
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]
