In-Memory Cache#

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, which Comet operators cannot read. Under Comet’s default settings the operators above the cache scan then run on Spark. With spark.comet.convert.inMemoryCache.enabled, a CometSparkColumnarToColumnar above the scan converts each batch for Comet operators instead.

This feature is experimental and disabled by default; see Limitations. Turn it on at startup, alongside the rest of Comet’s configuration:

$SPARK_HOME/bin/spark-shell \
    ... \
    --conf spark.comet.exec.inMemoryCache.enabled=true

On Spark 3.4, a union of a shuffle and a relation cached in Comet’s format keeps every shuffle partition. Spark 3.4 has no hook for the rule through which Comet lets AQE coalesce the shuffle partitions of a union that Comet runs.

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 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.

What changes when it is enabled#

With Comet’s serializer installed as spark.sql.cache.serializer:

  • Cached data is stored as CometCachedBatch rather than Spark’s DefaultCachedBatch.

  • Cached tables are scanned by CometInMemoryTableScan, which feeds Comet operators directly.

  • Per-batch column statistics are recorded in the layout Spark’s SimpleMetricsCachedBatchSerializer expects, so Spark can prune whole cached batches on a predicate before any of them is decoded.

  • The size Spark’s planner sees for a cached relation is its decoded Arrow size, not the compressed size it occupies in memory, as with Spark’s own cache formats. Compression therefore does not change how queries over a cached relation are planned, such as whether a join broadcasts it.

Relations whose schema Comet’s Arrow writer cannot store — interval types, most notably — are delegated in full to Spark’s default cache format, per relation. Which format a relation uses does not depend on a runtime config, because spark.sql.cache.serializer is a static setting and a relation whose format could change mid-session could not be read back reliably. The compression codec is a runtime config, but each batch records the codec it was written with, so data cached under one setting stays readable after the setting changes. Turning spark.comet.exec.inMemoryCache.enabled off at runtime only sends cached scans back to Spark’s execution path, where Spark operators read them through the row reader described under Limitations; the cached data stays readable either way.

A relation whose cached plan records observed metrics, from Dataset.observe, is still stored in Comet’s format but is scanned by Spark’s InMemoryTableScanExec, because Spark collects those metrics only through that scan.

Storage format#

Each cached batch is stored as a single Arrow IPC record batch message and its body.

The message carries no Arrow schema. The reader already has one: InMemoryRelation knows the cached relation’s attributes, and Comet maps them to exactly the Arrow fields the writer produced. Storing a schema in every batch would repeat the same bytes once per cached batch — for a wide relation cached in many batches, a large share of a payload that is not data.

Compression is applied by Arrow to each buffer separately, rather than by wrapping the whole payload in a Spark compression codec. That is what makes a projected read cheap: the message metadata records every buffer’s offset and length within the body, so a scan copies out only the byte ranges belonging to the columns it selected, and only those are decompressed. A read of one column out of six does roughly a sixth of the decompression work, and a SELECT count(*), which selects no columns at all, answers from the row count stored beside the payload without touching it.

Three codecs are offered: zstd, the default, lz4 and none. Measured by the same benchmark over the same 5M-row, six-column relation the tables under Performance use (Apple M3 Ultra, JDK 17, Spark 4.1, release build):

Codec

Materialize

Footprint

Read 1 of 6

Read 6 of 6

zstd

1232 ms

51 MiB

36 ms

248 ms

lz4

1115 ms

145 MiB

33 ms

110 ms

none

857 ms

315 MiB

31 ms

54 ms

zstd is the default, for footprint rather than for speed. It holds the data in a sixth of the memory none takes and pays for that on both sides: about 45% longer to materialize, and, on a read wide enough to inflate everything, more than four times longer. A narrow projection pays far less, because it only inflates the columns it asked for.

lz4 decompresses several times faster than zstd, so a read of every column takes less than half as long, at nearly three times the footprint. It is written with lz4-java, the JNI-accelerated library behind spark.io.compression.codec, in Arrow’s standard LZ4_FRAME format. Arrow’s own LZ4 codec is not used: it is commons-compress’s pure-Java implementation, and is orders of magnitude slower to write than zstd while also producing larger output.

What lz4 saves depends more on the data than what zstd saves does. It holds this relation in less memory than Spark’s own cache format does (217 MiB), but not a relation of sequential bigint columns, which Spark’s format delta-encodes: each of the wide relations under Limitations takes about 74 MiB with lz4, against 26 to 27 MiB in Spark’s format and with zstd.

none is the better setting for a relation that fits in memory uncompressed and is read at close to full width. The default is zstd because a cache that does not fit costs more than one that is slower to read, and Spark’s own cache format compresses by default too.

Dictionary-encoded columns are decoded before they are stored. A payload with no schema message has nowhere to record either that a column is dictionary encoded or the dictionary itself.

Configuration#

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.compression.codec

zstd

Arrow IPC compression codec for cached data: zstd, lz4 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.

Performance#

Measured with CometInMemoryCacheBenchmark (Apple M3 Max, JDK 17, Spark 4.1, release build). Regenerate with:

SPARK_GENERATE_BENCHMARK_FILES=1 \
  make benchmark-org.apache.spark.sql.benchmark.CometInMemoryCacheBenchmark

On a 5M-row relation of six flat columns:

Query shape

Spark cache scan + convert

CometInMemoryTableScan

Relative

Repeated scan (3 of 6 columns)

209 ms

172 ms

1.2x

Selective filter

72 ms

61 ms

1.2x

Row count only (0 of 6)

46 ms

38 ms

1.2x

Narrow projection (1 of 6)

70 ms

58 ms

1.2x

Full projection (6 of 6)

566 ms

324 ms

1.7x

And on a 1M-row relation of six columns whose middle three are structs, one of them nested two levels deep:

Query shape

Spark cache scan + convert

CometInMemoryTableScan

Relative

Row count only (0 of 6)

38 ms

32 ms

1.2x

Narrow projection (1 of 6)

109 ms

58 ms

1.9x

Full projection (6 of 6)

275 ms

130 ms

2.1x

Both columns read the cache at the default codec, zstd. The codec table above shows what none changes, and it is the full projection that moves most: nothing has to be inflated, so it runs several times faster, at six times the memory.

The two relations are not comparable to each other — different row counts, and a struct column carries several values per row.

Array and map columns are deliberately absent from the benchmark, not from the format — the cache stores and projects them, and CometInMemoryCacheSuite covers them. They cannot be measured here because the left column would not exist: it needs Spark’s cache scan to bridge into Comet operators, and CometSparkToColumnarExec declines ArrayType and MapType, so a query projecting one falls back to Spark row execution above the scan and the two columns stop measuring the same boundary.

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, which the benchmark turns on with spark.comet.convert.inMemoryCache.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, 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.convert.inMemoryCache.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.

Measured on an Apple M3 Ultra, Comet’s format with lz4 is faster than Spark’s in every shape under both kinds of operator. The three-column read is 1.4x faster than Spark’s format with either kind of operator above the scan, where zstd is about 15% slower on the same machine, and the full projection is 2.5x faster with Comet operators above the scan and 1.7x faster with a Spark operator above it.

Building the cache is measured the same way, from the same source, with its rows produced either by Comet operators or by Spark operators. Comet’s format is written straight from the Arrow batches of Comet operators, while Spark’s is always built from rows, so above Comet operators it first converts their batches. Measured on an Apple M3 Max (JDK 17, Spark 4.1, release build):

Rows produced by

Spark’s cache format

Comet’s cache format, zstd

Comet’s cache format, none

Comet operators

3285 ms

944 ms

601 ms

Spark operators

3680 ms

2079 ms

1710 ms

Spark’s format holds the relation in 217 MiB, and Comet’s in 51 to 55 MiB with zstd and 315 MiB with none. So with the default codec, Comet’s format builds 1.8x to 3.5x faster than Spark’s and takes about a quarter of the memory.

Kryo#

Spark serializes a cached batch with spark.serializer whenever the block leaves the heap: the _SER storage levels, replication, cross-executor fetches, and the disk half of the default MEMORY_AND_DISK. So an ordinary df.cache() that spills is enough to reach it.

If you run Kryo with spark.kryo.registrationRequired=true, register Comet’s classes:

spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrationRequired=true
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, 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.

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#

A relation cached after a sort, such as sortWithinPartitions, reports that order to the queries that read it, and Spark removes the sorts it makes redundant. Some native joins over such a relation do not keep the order, so partitions above them can come back unsorted, and an operator that relies on the order, such as a sort-merge join, can return wrong rows. These cases are tracked in #6787, #6898 and #6922, and are why the feature is disabled by default.

Reads that feed Spark operators rather than Comet ones can be slower than Spark’s own cache format, because every cached batch is decoded from Arrow before Spark reads it. How a Spark operator reads a relation cached in Comet’s format depends on the scan below it:

  • With native execution enabled (spark.comet.exec.enabled=true), the cache is scanned by CometInMemoryTableScan, and a Spark operator above it reads the scan’s batches through CometColumnarToRow, as it would above any other Comet operator.

  • With Comet enabled but native execution disabled, the cache is scanned by Spark’s InMemoryTableScanExec. When the operator directly above the scan takes part in whole-stage code generation, as filters, projections and aggregates do, Comet puts Spark’s ColumnarToRowExec between the two, and the generated code reads the cached Arrow vectors directly, with no intermediate row. The plan shows this as a ColumnarToRow above the InMemoryTableScan. It needs spark.sql.inMemoryColumnarStorage.enableVectorizedReader (on by default) and whole-stage code generation, and applies to relations of at most spark.sql.codegen.maxFields fields (100 by default, counting nested fields), beyond which Spark reads a cached relation only as rows. It is not applied in plan-only mode (spark.comet.explain.planOnly.enabled), where Spark executes its own plan unchanged.

  • Otherwise the scan’s row reader decodes each batch and writes its rows into one reused UnsafeRow. That covers Comet or spark.comet.exec.inMemoryCache.enabled turned off at runtime, and operators that do not take part in code generation, such as exchanges and limits, or a query that returns the cached rows as they are.

Whenever Spark’s scan reads Comet’s format, which also happens for a relation whose cached plan records Dataset.observe metrics, Comet records a fallback reason on the scan.

Measured by the same benchmark over the same 5M-row relation, with native execution off so that Spark operators consume the cached data, Comet disabled for the row reader and enabled for the fused reader (Apple M4, JDK 17, Spark 4.1; the average of two runs):

Read shape

Spark’s cache format

Comet’s format, row reader

Comet’s format, fused reader

Row count only (0 of 6)

63 ms

57 ms

35 ms

1 of 6 columns

63 ms

77 ms

54 ms

3 of 6 columns

113 ms

176 ms

133 ms

6 of 6 columns

306 ms

500 ms

334 ms

The fused reader is faster than Spark’s own format for the narrowest reads and within 20% of it for the others. The row reader takes up to 1.6 times as long, and it is the only reader for relations wider than spark.sql.codegen.maxFields: reading every column of relations of 100, 200 and 1500 nullable bigint columns took 2.2 to 2.5 times as long as from Spark’s format. A Spark operator above Comet’s native cache scan does not pay this; see Performance.

Most of what the row reader pays is decompression, so lz4 narrows these gaps. Measured on an Apple M3 Ultra, the row reader takes up to 1.25 times as long as Spark’s format with lz4, against up to 1.7 times with zstd, and the fused reader is faster than Spark’s format for every read. Reading every column of the wide relations takes 1.3 to 2.0 times as long as from Spark’s format with lz4, against 2.1 to 2.6 times with zstd. lz4 takes more memory than Spark’s format for those relations, though; see Storage format. #5485 tracks these gaps.

Comet’s serializer exists because Spark’s own Arrow cache format (SPARK-57268) is only available from Spark 4.3, which Comet does not yet support.