andygrove opened a new pull request, #6537: URL: https://github.com/apache/datafusion-comet/pull/6537
## Which issue does this PR close? No issue. Raised in the review of #5634, which turns the in-memory cache on by default: the [strict-Kryo thread](https://github.com/apache/datafusion-comet/pull/5634#discussion_r4095410684). Follows #6360. ## Rationale for this change #6360 made `CometDriverPlugin` install Comet's cache serializer only when Comet and its native execution are enabled at startup. `spark.sql.cache.serializer` is static, so an application that can never use Comet's format should keep Spark's. Two more startup settings decide whether Comet's format can work, and turning the cache on by default in #5634 would expose both: - **Kryo with registration required.** Under `spark.kryo.registrationRequired=true`, Kryo rejects any class it was not told about. Comet's cached batch is registered only through `CometKryoRegistrator`, and `spark.kryo.registrator` is read before any plugin runs, so Comet cannot add it. Without it, the first cached block Spark serializes (the disk half of the default `MEMORY_AND_DISK`, the `_SER` levels, replication) fails with `Class is not registered: org.apache.spark.sql.comet.execution.arrow.CometCachedBatch`, where Spark's own format works. The plugin only warned. Once the cache is on by default, that is a new error under an existing configuration. - **Comet shuffle without Comet's shuffle manager.** Comet disables itself when `spark.comet.shuffle.enabled`, which defaults to `true`, is on and `spark.shuffle.manager` is not one of Comet's managers. Such an application would cache everything in Comet's format with only Spark operators to read it, which is the case #6360 set out to exclude. ## What changes are included in this PR? - `CometDriverPlugin.maybeSetCacheSerializer` also requires that Kryo can store Comet's cached batch, and, while Comet shuffle is enabled, that `spark.shuffle.manager` names `CometShuffleManager` or `CometCelebornShuffleManager`. Kryo can store it unless it requires registration and `spark.kryo.registrator` does not list `CometKryoRegistrator`. Otherwise caches keep Spark's format, as they do with the cache config off. - The plugin's startup warning for the Kryo case now says that Comet keeps Spark's cache format. Native broadcast still needs the registrator, as before. - The plugin's boolean config reads also check a deprecated alternative key, as a session's reads do, so `spark.comet.exec.shuffle.enabled=false` counts. This applies to the existing executor memory overhead warning too. - The in-memory cache guide, the installation guide's Kryo section, the plugin overview and the config's doc string describe the new conditions. The cache guide also notes that Spark registers its own cached batch with Kryo only from 4.1, and that `CometKryoRegistrator` registers it too. ## How are these changes tested? - `CometInMemoryCacheSuite`: a new test covers the shuffle manager conditions (unset, `sort`, both of Comet's managers, Comet shuffle off under either key) and the Kryo conditions (no registrator, another registrator only, Comet's listed beside another, Kryo without registration required). The existing plugin tests now set Comet's shuffle manager, which they had relied on implicitly. - The new `CometInMemoryCacheKryoUnregisteredSuite` runs the setup from the review end to end: the plugin, Kryo with `registrationRequired=true`, and a registrator that registers Spark's cached batch but nothing of Comet's, as an application whose caches already work under registration on Spark 3.4 to 4.0 has to. A `DISK_ONLY` cache is stored as `DefaultCachedBatch` and reads back correctly. Without the change, the same test fails with `Class is not registered: org.apache.spark.sql.comet.execution.arrow.CometCachedBatch`. The suite is added to both PR workflows. - Ran locally on Spark 4.1: the plugin tests in `CometInMemoryCacheSuite`, `CometInMemoryCacheKryoSuite`, `CometInMemoryCacheKryoUnregisteredSuite`, the `CometPlugins*` suites and `CometConfSuite`. On Spark 3.5: the same cache suites and `CometPluginsMemoryOverheadWarningSuite`, plus the scalafix check. -- 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]
