mbutrovich commented on code in PR #6195:
URL: https://github.com/apache/datafusion-comet/pull/6195#discussion_r4095349957
##########
spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala:
##########
@@ -292,6 +299,12 @@ object CometSparkSessionExtensions extends Logging {
sparkConf.getBoolean("spark.memory.offHeap.enabled", false)
}
+ // Spark copies the SparkConf into each session's SQLConf and, by default,
refuses to set core
+ // configs such as this one at session level, so this is the application's
memory mode.
+ private def isOffHeapEnabled(conf: SQLConf): Boolean = {
+ conf.getConfString("spark.memory.offHeap.enabled", "false").toBoolean
+ }
Review Comment:
Can the session's `SQLConf` disagree with the memory mode the executors run
in? Spark rejects core configs in `SET` and `spark.conf.set`, but
`SparkSession.builder().config(...).getOrCreate()` against an existing session
goes through `applyModifiableSettings`, which calls `conf.setConfString` for
every non-static key, core configs included ([Spark 3.5
`SparkSession.scala#L1209-L1220`](https://github.com/apache/spark/blob/303c18c74664f161b9b969ac343784c088b47593/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala#L1209-L1220)).
Spark 4.0 has the same code in `classic/SparkSession.scala`. It only logs
"Using an existing Spark session; only runtime SQL configurations will take
effect."
So in a notebook where the SparkContext already started with off-heap
disabled, `.config("spark.memory.offHeap.enabled", "true")` on the builder puts
`true` in the session's `SQLConf`. `isCometLoaded` then passes, while the
executors still take the on-heap path, because `getMemoryConfig` reads
`SparkEnv.get.conf`
([`CometExecIterator.scala#L103`](https://github.com/apache/datafusion-comet/blob/b4e48b7d670640ac8d42af2535dfff377640cb1a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala#L103)).
That is the case this PR sets out to prevent.
How about reading the memory mode from the same place the executors do? On
the driver, `SparkEnv.get.conf` is the SparkContext's conf, and the existing
public `isOffHeapEnabled(SparkConf)` already handles it, so the new overload
can go. `spark.comet.exec.onHeap.enabled` can stay on the `SQLConf` so
`ENABLE_COMET_ONHEAP` still supplies its default. A test in
`CometPluginsExtensionOnlySuite` that sets `spark.memory.offHeap.enabled=true`
on the session and checks that no Comet operators are planned would cover it.
```suggestion
// Read the SparkContext's conf rather than the session's SQLConf. Reusing
a session through
// SparkSession.Builder copies core configs into its SQLConf, but
executors keep the
// SparkContext's memory mode.
private def isOffHeapEnabled(conf: SQLConf): Boolean = {
Option(SparkEnv.get).exists(env => isOffHeapEnabled(env.conf))
}
```
--
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]
