feat: enable Comet's in-memory cache by default #5634
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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(true) | ||
|
Contributor
There was a problem hiding this 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?
Member
Author
There was a problem hiding this comment. The audit turned up part of the answer. Under AQE the first case is the normal outcome, not an edge case. Once the table-cache stage materializes, the re-plan leaves the operators above it on Spark, so a plain aggregate or join over a cached table reads Comet's format through a Yes to option 1 from #5485 landing with the default flip. It will need to cover the #6202 path too, since nothing records a fallback reason there today.
Member
Author
There was a problem hiding this comment. #6202 is fixed by #6208, which is now merged into this branch in 299d381. Under AQE, an aggregate or join over a cached table now stays native once the table-cache stage materializes, so the first case is no longer the normal outcome: it takes an operator Comet does not support above the cached scan. Option 1 no longer has a #6202 path to cover either, since the operators above the stage now convert, or record their own fallback reason, like any other operator. The benchmark with AQE on against Spark's own format is next. |
||
|
|
||
| val COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_CODEC: ConfigEntry[String] = | ||
| conf("spark.comet.exec.inMemoryCache.compression.codec") | ||
|
|
||
| Original file line number | Diff line number | Diff line change | ||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -106,7 +106,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, | ||||||||||||||
| CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.defaultValue.get)) { | ||||||||||||||
|
Comment on lines
+109
to
+111
Contributor
There was a problem hiding this comment. With the default flipped, this installs Comet's serializer for every application that loads #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
Suggested change
A session that starts with execution off and turns it on later would then keep Spark's format, and The driver-plugin test sets the key to
Member
Author
There was a problem hiding this comment. Agreed. With the default on, an application that can never plan the native scan shouldn't get Comet's format. I'll gate the install on
Member
Author
There was a problem hiding this comment. |
||||||||||||||
| val serializerKey = StaticSQLConf.SPARK_CACHE_SERIALIZER.key | ||||||||||||||
| val serializerValue = | ||||||||||||||
| "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer" | ||||||||||||||
|
|
||||||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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=trueand noCometKryoRegistrator, adf.cache()that spills to disk now fails with "Class is not registered" where it did not before. The plugin only logs a warning for that.The versioning policy counts a new error under the same explicit configuration as a behavior change. Could you add an entry to the upgrade guide under the next release that covers the format change and the Kryo requirement? The policy asks for a
spark.comet.legacy.*key, butspark.comet.exec.inMemoryCache.enabled=falsealready 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.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Agreed on the upgrade guide entry, covering both the format change and the Kryo requirement. Moving the flip past 1.1.0 changes one premise, though. 1.1.0 ships this key with a default of
false, so turning it on in the next release is a change to an existing key's default, which is the first case the policy lists. Let's settle the legacy-key question when this comes out of draft.