feat: enable Comet's in-memory cache by default

.github/workflows/pr_build_linux.yml

@@ -516,6 +516,8 @@ jobs:
               org.apache.comet.exec.CometInMemoryCacheSuite
               org.apache.comet.exec.CometInMemoryCachePruningSuite
               org.apache.comet.exec.CometInMemoryCacheKryoSuite
+              org.apache.comet.exec.CometInMemoryCacheKryoUnregisteredSuite
+              org.apache.comet.exec.CometInMemoryCacheKryoClassesToRegisterSuite
               org.apache.comet.exec.CometGenerateExecSuite
               org.apache.comet.exec.CometWindowExecSuite
               org.apache.comet.exec.CometJoinSuite

.github/workflows/pr_build_macos.yml

@@ -220,6 +220,8 @@ jobs:
               org.apache.comet.exec.CometInMemoryCacheSuite
               org.apache.comet.exec.CometInMemoryCachePruningSuite
               org.apache.comet.exec.CometInMemoryCacheKryoSuite
+              org.apache.comet.exec.CometInMemoryCacheKryoUnregisteredSuite
+              org.apache.comet.exec.CometInMemoryCacheKryoClassesToRegisterSuite
               org.apache.comet.exec.CometGenerateExecSuite
               org.apache.comet.exec.CometWindowExecSuite
               org.apache.comet.exec.CometJoinSuite

docs/source/contributor-guide/plugin_overview.md

@@ -48,6 +48,8 @@ and skips the remaining steps. Otherwise it:
 - Appends `CometSparkSessionExtensions` to `spark.sql.extensions`, unless it is already listed.
 - Sets `spark.sql.cache.serializer` to Comet's `ArrowCachedBatchSerializer` when
   `spark.comet.exec.inMemoryCache.enabled=true`, unless the application has chosen a different serializer.
+  It leaves Spark's serializer in place where Comet could not scan its format natively or Kryo would
+  reject it; see [In-Memory Cache](../user-guide/latest/in-memory-cache.md).
 - Registers `CometSource` with Spark's metrics system and adds `CometMetricsListener` to
   `spark.sql.queryExecutionListeners` when `spark.comet.metrics.enabled=true`.
 - Logs a warning for settings that are likely to cause problems, such as an unset `spark.executor.memoryOverhead`.

docs/source/user-guide/latest/compatibility/operators.md

@@ -35,8 +35,8 @@ readable empty output files and their schema metadata.
 ## In-Memory Cache
 
 Comet can store cached relations (`df.cache()`, `CACHE TABLE`) in Arrow format and scan them
-natively. This is experimental and disabled by default; see [In-Memory Cache](../in-memory-cache.md)
-for how to enable it. Comet does not replace a `spark.sql.cache.serializer` that the application
+natively. This is experimental and enabled by default; see [In-Memory Cache](../in-memory-cache.md)
+for how to turn it off. Comet does not replace a `spark.sql.cache.serializer` that the application
 has already set. Relations whose schema Comet's Arrow writer does not support are cached in
 Spark's default format, and their scans fall back to Spark. Reads that feed Spark operators rather
 than Comet operators can be slower than Spark's cache.

docs/source/user-guide/latest/in-memory-cache.md

@@ -21,24 +21,27 @@
 
 Comet can store Spark's in-memory cache (`CACHE TABLE`, `df.cache()`, `df.persist()`) in an Arrow
 format that Comet operators read directly. Without it, a cached table is stored in Spark's own
-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.
+format, which Comet operators cannot read. Under Comet's default settings the operators above the
+cache scan then run on Spark. With `spark.comet.sparkToColumnar.enabled`, a
+`CometSparkColumnarToColumnar` above the scan converts each batch for Comet operators instead.
 
-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:
 
 ```shell
 $SPARK_HOME/bin/spark-shell \
     ... \
-    --conf spark.comet.exec.inMemoryCache.enabled=true
+    --conf spark.comet.exec.inMemoryCache.enabled=false
 ```
 
 It has to be set before the `SparkContext` starts. Comet's driver plugin chooses
-`spark.sql.cache.serializer` once, while the context is initializing, so a session that started
-with the default goes on using Spark's cache format however the config is set afterwards. The
-plugin installs Comet's serializer only if `spark.comet.enabled` and `spark.comet.exec.enabled`
-are enabled at that point too, because an application that starts without native execution could
-not scan Comet's format natively.
+`spark.sql.cache.serializer` once, while the context is initializing, so an application keeps the
+cache format it started with however the config is set afterwards. The plugin installs Comet's
+serializer only if `spark.comet.enabled` and `spark.comet.exec.enabled` are enabled at that point
+too, because an application that starts without native execution could not scan Comet's format
+natively. It also keeps Spark's format when Comet shuffle is enabled but `spark.shuffle.manager` is
+not one of Comet's shuffle managers, since Comet then disables itself, and when Kryo would reject
+Comet's format; see [Kryo](#kryo).
 
 ## What changes when it is enabled
 
@@ -109,7 +112,7 @@ nowhere to record either that a column is dictionary encoded or the dictionary i
 
 | Config                                                  | Default | Description                                                                                                                                    |
 | ------------------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------- |
-| `spark.comet.exec.inMemoryCache.enabled`                | `false` | Whether to store and scan Spark's in-memory cache in Comet's format. Read at startup.                                                          |
+| `spark.comet.exec.inMemoryCache.enabled`                | `true`  | Whether to store and scan Spark's in-memory cache in Comet's format. Read at startup.                                                          |
 | `spark.comet.exec.inMemoryCache.compression.codec`      | `zstd`  | Arrow IPC compression codec for cached data: `zstd` or `none`. Affects newly cached data only — a batch records the codec it was written with. |
 | `spark.comet.exec.inMemoryCache.compression.zstd.level` | `1`     | Compression level when the codec is `zstd`. Ignored otherwise.                                                                                 |
 
@@ -158,10 +161,43 @@ back to Spark row execution above the scan and the two columns stop measuring th
 Read what this compares carefully. Comet execution is on in both columns, so the aggregation runs
 on Comet either way and only the cache-scan boundary moves: on the left, Spark's
 `InMemoryTableScanExec` feeds those same Comet operators through a `CometSparkColumnarToColumnar`
-bridge; on the right, `CometInMemoryTableScan` feeds them directly. Both columns read the same
+bridge, which the benchmark turns on with `spark.comet.sparkToColumnar.enabled`; on the right,
+`CometInMemoryTableScan` feeds them directly. Both columns read the same
 Comet-written `CometCachedBatch`. These numbers are therefore "keep the cached scan native" against
 "fall back to a Spark cache scan and convert", not Comet against Spark execution, and not a
-comparison with Spark's own cache format. That comparison is under [Limitations](#limitations).
+comparison with Spark's own cache format, which follows.
+
+### Against Spark's cache format
+
+What turning the feature on changes for a query that Comet runs is measured against Spark's own
+cache format by the benchmark's adaptive cases. Comet and AQE are on, Comet's other settings are at
+their defaults, and the same 5M-row relation is cached in each format. The defaults leave
+`spark.comet.sparkToColumnar.enabled` off, so Comet operators cannot read Spark's cache scan, and
+with Spark's format the operators directly above the scan run on Spark. Measured on an AMD Ryzen 9
+7950X3D (JDK 17, Spark 4.1, release build):
+
+| Query shape                | Spark's cache format | Comet's cache format | Relative |
+| -------------------------- | -------------------: | -------------------: | -------: |
+| Row count only (0 of 6)    |                29 ms |                24 ms |     1.2x |
+| Narrow projection (1 of 6) |                52 ms |                34 ms |     1.5x |
+| 3 of 6 columns             |               102 ms |               112 ms |     0.9x |
+| Full projection (6 of 6)   |               299 ms |               224 ms |     1.3x |
+
+A Spark operator above the cache scan, standing in for any operator Comet does not support, is
+measured the same way, with Comet's aggregate turned off. With Comet's format, the native scan feeds
+that operator through a columnar-to-row transition:
+
+| Query shape                | Spark's cache format | Comet's cache format | Relative |
+| -------------------------- | -------------------: | -------------------: | -------: |
+| Row count only (0 of 6)    |                39 ms |                16 ms |     2.4x |
+| Narrow projection (1 of 6) |                50 ms |                27 ms |     1.8x |
+| 3 of 6 columns             |                97 ms |               112 ms |     0.9x |
+| Full projection (6 of 6)   |               303 ms |               299 ms |     1.0x |
+
+Comet's format is as fast or faster in every shape but one: the read of three of the six columns,
+all of them longs, is about 10% slower under either kind of operator. That cost is `zstd`
+decompression. With the `none` codec, the same read is 2.7x faster than Spark's format with Comet
+operators above the scan, and 1.6x faster with a Spark operator above it.
 
 ## Kryo
 
@@ -179,16 +215,27 @@ spark.kryo.registrator=org.apache.comet.CometKryoRegistrator
 
 Comet cannot set `spark.kryo.registrator` for you the way it sets `spark.sql.cache.serializer`:
 `KryoSerializer` reads it when `SparkEnv` builds the serializer, which happens before any plugin
-runs. Without it, caching fails with a "Class is not registered" error that does not name this
-feature. Comet's driver plugin warns at startup when it sees Kryo, `registrationRequired`, and no
-registrator. Native broadcast needs the same registrator even when the cache is disabled; see
+runs. Without it, Kryo would reject Comet's cached batch with a "Class is not registered" error
+that does not name this feature. So when Kryo requires registration and has not registered
+Comet's cached batch, Comet's driver plugin does not install Comet's serializer, and caches stay in
+Spark's format. Registrations made another way, through a registrator of the application's own or
+`spark.kryo.classesToRegister`, count as well. The plugin warns at startup when Kryo requires
+registration and has not registered every class `CometKryoRegistrator` registers. An application
+that sets `spark.sql.cache.serializer` to Comet's serializer itself gets the error instead. Native
+broadcast needs the same registrator even when the cache is disabled; see
 [Kryo serialization](installation.md#kryo-serialization).
 
+Spark registers its own cached batch with Kryo only from Spark 4.1, so on earlier versions caching
+in either format under `registrationRequired` needs a registrator. `CometKryoRegistrator` registers
+Spark's cached batch too.
+
 ## Limitations
 
-Reads that feed **Spark** operators rather than Comet ones are slower than Spark's own cache
-format, and the narrower the read, the wider the gap. Measured by the same benchmark over the same
-5M-row relation, with Comet off so that Spark operators consume the cached data:
+Spark's own cache scan, `InMemoryTableScanExec`, reads Comet's format more slowly than Spark's, and
+the narrower the read, the wider the gap. Spark's scan reads a cached relation when a session turns
+Comet or its native execution off after caching, and when the relation's cached plan records
+`Dataset.observe` metrics, and Comet records a fallback reason on the scan in either case. Measured
+by the same benchmark over the same 5M-row relation, with Comet off:
 
 | Read shape              | Spark's cache format | Comet's cache format | Slowdown |
 | ----------------------- | -------------------: | -------------------: | -------: |
@@ -197,8 +244,9 @@ format, and the narrower the read, the wider the gap. Measured by the same bench
 | 3 of 6 columns          |                98 ms |               331 ms |     3.4x |
 | 6 of 6 columns          |               410 ms |               623 ms |     1.5x |
 
-This is why the feature is off by default. The cause is not yet established;
-[#5485](https://github.com/apache/datafusion-comet/issues/5485) tracks it.
+A Spark operator above Comet's native cache scan does not pay this; see [Performance](#performance).
+This gap is the main reason the feature is still described as experimental. The cause is not yet
+established; [#5485](https://github.com/apache/datafusion-comet/issues/5485) tracks it.
 
 Comet's serializer exists because Spark's own Arrow cache format
 ([SPARK-57268](https://issues.apache.org/jira/browse/SPARK-57268)) is only available from Spark

docs/source/user-guide/latest/installation.md

@@ -269,8 +269,9 @@ If the application uses Kryo (`spark.serializer=org.apache.spark.serializer.Kryo
 
 Without it, any query that uses Comet's native broadcast exchange, which is enabled by default,
 fails with Kryo's "Class is not registered" error, for example on the first broadcast hash join.
-The [in-memory cache](in-memory-cache.md#kryo) needs the same registrator. Set it before the
-`SparkContext` is created: `KryoSerializer` reads it before Comet's plugin runs, so Comet cannot
-add it for you. `spark.kryo.registrator` accepts a comma-separated list, so an application with
-its own registrator can list both. Comet logs a warning at startup when Kryo requires registration
-and this registrator is missing.
+Comet's [in-memory cache](in-memory-cache.md#kryo) format needs the same registrations, and while
+Kryo has not registered Comet's cached batch, Comet's plugin keeps caches in Spark's format. Set it
+before the `SparkContext` is created: `KryoSerializer` reads it before Comet's plugin runs, so
+Comet cannot add it for you. `spark.kryo.registrator` accepts a comma-separated list, so an
+application with its own registrator can list both. Comet logs a warning at startup when Kryo
+requires registration and has not registered the classes this registrator covers.

docs/source/user-guide/latest/migration-guide.md

@@ -57,6 +57,35 @@ Treat setting one of these keys as a temporary measure. If you find you cannot s
 legacy behavior, please open an issue describing your use case so it can be considered before the
 key is removed.
 
+## Upgrading to Comet 1.2.0
+
+Comet `1.2.0` makes no behavior changes that need a `spark.comet.legacy.*` key. The changes below
+need none either, but check whether any of them applies to your deployment.
+
+### In-Memory Cache Enabled by Default
+
+`spark.comet.exec.inMemoryCache.enabled` now defaults to `true`. An application that loads
+`CometPlugin` now stores what it caches with `CACHE TABLE`, `df.cache()` or `df.persist()` in
+Comet's Arrow format instead of Spark's, and Comet scans it natively. The format does not change
+query results, but it can change performance. Spark's own cache scan reads Comet's format more
+slowly than Spark's, which matters when a session turns Comet or its native execution off after
+caching, and Comet records a fallback reason on such a scan. See
+[In-Memory Cache](in-memory-cache.md#limitations).
+
+The format is chosen once, when the application starts. To keep Spark's format, set
+`spark.comet.exec.inMemoryCache.enabled=false` then. Comet also keeps Spark's format without that
+setting when the application:
+
+- starts with `spark.comet.enabled` or `spark.comet.exec.enabled` set to `false`.
+- leaves Comet shuffle enabled without one of Comet's shuffle managers, so that Comet disables
+  itself.
+- uses Kryo with `spark.kryo.registrationRequired=true` and has not registered Comet's cached
+  batch, because Kryo would reject it. To use Comet's format, register Comet's classes with
+  `spark.kryo.registrator=org.apache.comet.CometKryoRegistrator`; see
+  [Kryo](in-memory-cache.md#kryo).
+
+An application that sets `spark.sql.cache.serializer` itself keeps the serializer it chose.
+
 ## Upgrading to Comet 1.1.0
 
 Comet `1.1.0` makes no behavior changes that need a `spark.comet.legacy.*` key. The changes below

docs/source/user-guide/latest/operators.md

@@ -65,7 +65,7 @@ omitted from the tables below and may be reconsidered based on demand:
 | `LocalTableScanExec`    | ⚠️     | Disabled by default; there is no acceleration advantage and this operator is typically only used in test code. Can be opted into via config ([#4393](https://github.com/apache/datafusion-comet/pull/4393)).                                                                                                                                                                                                       |
 | `EmptyRelationExec`     | ✅     | Spark 4.0 and later. See [Empty Relations](compatibility/operators.md#empty-relations) for native-input support and writer fallback.                                                                                                                                                                                                                                                                               |
 | `RangeExec`             | ⚠️     | Disabled by default. Set `spark.comet.exec.range.enabled=true` to generate the rows of `spark.range` and SQL `range()` in native code, so the operators above them run natively. It can be slower than Spark when those operators are only cheap expressions, such as a filter, which Spark compiles together with the range into one loop. Ranges whose arithmetic overflows the `Long` range fall back to Spark. |
-| `InMemoryTableScanExec` | ⚠️     | Experimental, disabled by default. Set `spark.comet.exec.inMemoryCache.enabled=true` before the application starts so Comet installs its Arrow cache serializer. Relations with unsupported column types stay in Spark's cache format and fall back. See [In-Memory Cache](in-memory-cache.md).                                                                                                                    |
+| `InMemoryTableScanExec` | ⚠️     | Experimental, enabled by default. Comet installs its Arrow cache serializer as the application starts, unless `spark.comet.exec.inMemoryCache.enabled` is false. Relations with unsupported column types stay in Spark's cache format and fall back. See [In-Memory Cache](in-memory-cache.md).                                                                                                                    |
 
 ## Projection and filtering