mbutrovich commented on code in PR #5634:
URL: https://github.com/apache/datafusion-comet/pull/5634#discussion_r4095403630
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -278,7 +278,7 @@ object CometConf extends ShimCometConf {
"SparkContext, otherwise caching fails as soon as a block is
serialized, including " +
"the disk half of the default MEMORY_AND_DISK storage level.")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
Review Comment:
#5485 treats the slower Spark-operator reads as acceptable because "the
cache path is off by default ... rather than a regression in a shipped path."
This PR makes it a shipped path, and the format is fixed when the relation
materializes, so a user can't avoid it for one query.
With the plugin gated as suggested above, the remaining exposure is a query
where the cached scan runs natively and a Spark operator above it reads through
a columnar-to-row transition, and a session that turns Comet off at runtime
after caching. Is there a benchmark number for the first case against Spark's
own cache format? The published numbers compare against Comet off entirely. For
the second case, option 1 in #5485 (a fallback reason when Spark operators read
a relation stored in Comet's format) is what would tell a user to turn the
feature off. Could that land before or with this PR?
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -104,7 +104,9 @@ object CometDriverPlugin extends Logging {
private[apache] def maybeSetCacheSerializer(
conf: SparkConf,
extraConfs: ju.HashMap[String, String]): Unit = {
- if (conf.getBoolean(CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key,
false)) {
+ if (conf.getBoolean(
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key,
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.defaultValue.get)) {
Review Comment:
With the default flipped, this installs Comet's serializer for every
application that loads `CometPlugin`, including ones that start with
`spark.comet.enabled=false` or `spark.comet.exec.enabled=false`. Those
applications can never plan `CometInMemoryTableScan`, so every cached read goes
through Spark operators on top of Comet's format. That is the 1.5x to 5.2x
slower case in the [Limitations
table](https://github.com/apache/datafusion-comet/blob/d5983c2d64784a34a5a750c851b1d0a18615251f/docs/source/user-guide/latest/in-memory-cache.md#L176-L190).
Keeping the plugin in `spark.plugins` cluster-wide and switching Comet off
with `spark.comet.enabled=false` is a common setup, and before this PR it left
the cache format alone.
#5485 already lists this check as sound, and only calls it narrow because
anyone who opted in would have execution enabled. That premise no longer holds
once the feature is on by default. Could the plugin also require both configs
at startup? This object already has a
[`getBooleanConf`](https://github.com/apache/datafusion-comet/blob/d5983c2d64784a34a5a750c851b1d0a18615251f/spark/src/main/scala/org/apache/spark/Plugins.scala#L200-L201)
helper that falls back to the entry's default, so the new read can use it too:
```suggestion
if (getBooleanConf(conf, CometConf.COMET_ENABLED) &&
getBooleanConf(conf, CometConf.COMET_EXEC_ENABLED) &&
getBooleanConf(conf, CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED)) {
```
A session that starts with execution off and turns it on later would then
keep Spark's format, and `CometExecRule` already records a fallback reason for
that case. That seems like the right trade, since the serializer is fixed for
the application.
The [driver-plugin
test](https://github.com/apache/datafusion-comet/blob/d5983c2d64784a34a5a750c851b1d0a18615251f/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala#L910-L937)
sets the key to `true` explicitly, so nothing exercises the fallback to the
default, which is the code change in this file. Could you add a case where the
key is unset (serializer installed), plus cases for `spark.comet.enabled=false`
and `spark.comet.exec.enabled=false` (serializer not installed)?
##########
docs/source/user-guide/latest/in-memory-cache.md:
##########
@@ -24,13 +24,13 @@ format that Comet operators read directly. Without it, a
cached table is stored
format and every scan of it has to convert each batch before Comet can
continue, which shows up in
the plan as a `CometSparkColumnarToColumnar` above the cache scan.
-This feature is **experimental and disabled by default**. Turn it on at
startup, alongside the rest
-of Comet's configuration:
+This feature is **experimental and enabled by default**. To turn it off, set
the config at startup,
+alongside the rest of Comet's configuration:
Review Comment:
This key did not exist in 1.0.0, so a user upgrading from 1.0.0 goes from
Spark's cache format to Comet's without setting anything. With
`spark.kryo.registrationRequired=true` and no `CometKryoRegistrator`, a
`df.cache()` that spills to disk now fails with "Class is not registered" where
it did not before. The plugin only [logs a
warning](https://github.com/apache/datafusion-comet/blob/d5983c2d64784a34a5a750c851b1d0a18615251f/spark/src/main/scala/org/apache/spark/Plugins.scala#L134-L152)
for that.
The versioning policy counts a new error under the same explicit
configuration as a [behavior
change](https://github.com/apache/datafusion-comet/blob/d5983c2d64784a34a5a750c851b1d0a18615251f/docs/source/about/versioning_policy.md#L166-L185).
Could you add an entry to the [upgrade
guide](https://github.com/apache/datafusion-comet/blob/d5983c2d64784a34a5a750c851b1d0a18615251f/docs/source/user-guide/latest/migration-guide.md)
under the next release that covers the format change and the Kryo requirement?
The policy asks for a `spark.comet.legacy.*` key, but
`spark.comet.exec.inMemoryCache.enabled=false` already restores the old
behavior, so naming that key in the entry seems enough. If you read the policy
differently, it would be good to settle that here, since this is one of the
first behavior changes since 1.0.0.
--
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]
