Conversation
…trings Follow-up to apache#5051, applying items from apache#5487. Replace the per-column Arrow IPC stream layout of `CometCachedBatch` with a single encapsulated IPC record batch message per cached batch, carrying no Schema message and no end-of-stream marker. The reader rebuilds the schema from the cached relation's attributes, so a wide relation no longer repeats the same schema bytes once per cached batch. Compression moves from a whole-payload Spark codec to Arrow's per-buffer IPC compression. That is what makes projection cheap: the message metadata records every buffer's offset and length in the body, so `CachedBatchIpc.readProjected` copies out only the byte ranges of the columns a scan selected and decompresses just those. This subsumes the separate "drop the schema message" item, since there is no longer a per-column stream to frame. Dictionary-encoded columns are decoded before being stored: a payload with no schema message cannot describe a dictionary encoding. The codec defaults to zstd, and lz4 is deliberately not offered. Arrow's lz4 is commons-compress's pure-Java implementation, unrelated to the JNI-accelerated lz4-java behind `spark.io.compression.codec`. Over a 200k-row six-column relation it measured 205s to write against 347ms for zstd, while also producing larger output, so no workload prefers it. zstd also beats storing batches uncompressed on both axes (347ms and 2 MiB against 1743ms and 13 MiB), because the bytes it saves cost more to copy and store than compressing them costs. Decompression is done here rather than left to `VectorLoader`, which leaks: `VectorLoader.loadBuffers` collects a field's decompressed buffers into a local list and releases them only after the whole field loads, so a buffer that fails to decompress strands every buffer of that field decompressed before it. A string column reaches this, its offsets buffer decompressing before its data buffer throws. Also track statistics bounds for collated string columns, comparing with the collation's own ordering through a new `CometTypeShim.compareStrings`. Matching the bare `StringType` object excluded collated columns, which then got null bounds and no pruning. Benchmark over a 5M-row six-column relation, keeping the cached scan native against falling back to a Spark cache scan and converting: 1.3x on a repeated scan, 1.3x on a narrow projection and 2.3x on a full projection.
…on layout Cleanup pass over the cache format change. No behaviour change. Drop the `compareStrings` shim in favour of `TypeUtils.getInterpretedOrdering`. That method is public with the same signature on every supported Spark version, and on Spark 4 it resolves a `StringType` through `CollationFactory.fetchCollation(collationId).comparator` -- the comparison the shim was reaching for. So the collation awareness comes from Spark itself and the shim, its Spark 3.x stub and the hand-rolled per-type `compare` all go. The ordering is now resolved once per column per partition rather than being re-dispatched on the `DataType` twice per row. Build the projection's index layout once per partition instead of per batch. The node, buffer and variadic index arithmetic is a pure function of the cached schema and the selected columns, but it walks every field of the relation, so recomputing it per batch made the bookkeeping O(total columns) against O(selected columns) of useful work -- worst in the wide-relation, narrow-projection case the format exists for. `CachedBatchIpc.Projection` now holds that layout and the projected schema, and owns the whole decode; `ProjectedBatch` is left with ownership only. This also puts the projected schema next to the code that packs buffers in the same order, an invariant that previously spanned two files unstated. Smaller cleanups: use Arrow's `DataSizeRoundingUtil.roundUpTo8Multiple` rather than open-coding IPC body alignment; size the serialization buffer from the record batch's known body length instead of growing from 32 bytes; resolve decompressors once instead of per batch; share the dictionary lookup guard between `Utils.combineDictionaryProviders` and the cache writer; read the codec config through one helper carrying the driver-vs-executor rationale; and collapse the duplicated compressed-buffer predicate and scramble loop in the test helper. Corrects two `Utils` scaladocs that still described the per-column stream format this change replaced. Benchmark and codec figures in the docs re-measured against the current code.
arrow-compression ships META-INF/services/org.apache.arrow.vector.compression.CompressionCodec$Factory. The shade plugin copies it verbatim without a ServicesResourceTransformer, so the jar declared a provider for Spark's own unshaded Arrow interface while naming a class that exists here only under the relocated package. Every ServiceLoader lookup Spark's Arrow made then failed with a ServiceConfigurationError, which took CompressionCodec.Factory's static initializer down with it and broke unrelated Arrow IPC reads, including mapInArrow. Add ServicesResourceTransformer so the service file name and its contents are both relocated. arrow-compression is the only bundled artifact that ships one. Also drop an unused NonFatal import that scalafix flagged.
"releases its vectors when a column fails part way through" zeroed the last 16 bytes of a compressed buffer and required the read to fail. Whether that fails is a property of the zstd runtime, not of Comet: the cached payload is byte-identical across Spark versions, but Comet takes zstd-jni from Spark rather than from arrow-compression, and 1.5.5 (Spark 3.4, 3.5) decodes that frame while 1.5.7 (Spark 4.x) reports it corrupt. So the test passed on 4.x and failed on 3.4 and 3.5. The scenario it claimed to cover is also unreachable: CachedBatchIpc decompresses every selected buffer before VectorLoader runs, so no content corruption can fail part way through the load. The two remaining leak tests corrupt a frame from its header onwards, which every zstd release rejects, and already cover a failure at a column's first buffer and a failure after an earlier buffer of the same column decoded. Records the constraint on scramble so a future test does not reach for a tail-only corruption again, and drops the now unused truncateColumn helper and the dictionary fixture's payload argument.
Flip spark.comet.exec.inMemoryCache.enabled to true so cached tables are stored and scanned in Comet's Arrow format without an opt-in. CometDriverPlugin.maybeSetCacheSerializer read the config out of SparkConf with a hardcoded false default, so flipping the ConfigEntry alone would have left the serializer uninstalled unless the user set the key explicitly. It now falls back to the entry's own default, matching how the plugin reads spark.comet.metrics.enabled. Stacked on apache#5543.
andygrove
changed the title
…enchmark Addresses review feedback asking whether nested data should be tested and benchmarked. Nested columns were already round-tripped, but only under a full projection, which cannot see the part of the format that is nontrivial for them. A flat column always owns one field node and two or three buffers; a nested one owns a run as long as its subtree, and selecting every column covers the whole sequence however it is partitioned. So the buffer-span arithmetic was only exercised in the one shape where getting it wrong does not show. Adds two tests over a six-column relation whose middle four columns are a struct, an array, a map and a struct wrapping an array: - Each column takes its turn as the sole projection while the other five are corrupted, so a run computed short or long is caught by reaching into a corrupted neighbour. - Values are compared against the uncached query across single-column, paired and out-of-order projections. Row counts cannot catch a window that is misaligned but still decompresses, and out-of-order is the case a full projection cannot stand in for. The per-column statistics test now runs over the nested relation too, since a nested column's recorded size is the sum of its whole subtree. Both new tests fail if fieldNodeCount stops recursing into children. In the benchmark, adds the three projection widths over a relation of struct columns, and asserts the width each case claims. That assertion caught the existing "full projection (6 of 6 columns)" case reading three: count() over a non-nullable column is rewritten to count(1) by NullPropagation, which prunes the column out of the scan, and only k, s1 and s2 were nullable -- and those only incidentally, because Remainder can divide by zero. Every column of both relations is now nullable so count(c) genuinely reads c, and the documented numbers are regenerated. Array and map columns are left out of the benchmark deliberately: the baseline arm needs Spark's cache scan to bridge into Comet operators, and CometSparkToColumnarExec declines ArrayType and MapType, so for those the arm does not exist and the two cases stop measuring the same boundary. The docs say so rather than leaving it to be rediscovered.
…ection-projection
…ection-projection
Reader-side: a cached payload carries no schema, so `Projection` derived every node and buffer window from `Utils.toArrowSchema(cacheAttributes)` with nothing checking the writer had produced that layout. `load` now compares `nodesLength()`/`buffersLength()` against the totals `selectedRange` already computes, before any unchecked `batch.buffers(j)`. Writer-side: `isArrowBacked` accepts a `FixedSizeBinaryVector` for a `BinaryType` column, which is two buffers where the reader rebuilds three, and it answers for the top-level vector only -- so a struct of large strings passes it and is stored with 64-bit offsets. `matchesReaderLayout` compares the batch's Arrow types against the reader's recursively, and a batch that disagrees takes the conversion path instead. A dictionary column's field carries the index type, so the dictionary's field is what is compared. Also: an unrecognized body-compression byte is rejected rather than read as plain bytes, `fieldVariadicCount` and the variadic plumbing are gone (the length check covers view vectors, which the counts would not have), `columnSizes` no longer re-walks each column's subtree, the write codec is a case class rather than a bare tuple, the per-partition `Projection` is lazy so a row-count-only read never builds it, `hydrateDictionaries` is `decodeDictionaries`, `Projection` takes an `IndexedSeq`, and the stale `readProjected` links and some over-long comments are fixed. Tests: the two projection tests become one parameterized over both relations, caching once and restoring the payload between columns instead of re-caching; the two leak tests become one with two corruption points. New tests cover the reader's layout check and the writer declining a fixed-size-binary batch.
…ample Compressing through VectorUnloader leaks on the failure path: appendNodes retains each input buffer and accumulates the compressed ones into a list local to getRecordBatch, so a buffer that fails to compress strands that retain and leaves every buffer compressed before it reachable from nothing. Closing the input batch afterwards undoes neither. Unload plain and compress in CachedBatchIpc.compressed instead, mirroring what decompressed already does on the read side, so every allocation stays reachable from an error path that owns it. The docs enabled the cache with spark.conf.set, which cannot work: the driver plugin picks spark.sql.cache.serializer while the SparkContext is initializing. Show it as a startup --conf. Also drops a redundant s interpolator that the scalafix lint rejected.
…ection' into feat/cache-enabled-by-default # Conflicts: # docs/source/user-guide/latest/in-memory-cache.md
…-default # Conflicts: # spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
ArrowWriter.writeColumns drove its loop from the input ColumnarBatch's width while indexing the writer's fields, which come from the schema the batch is written under. That assumed every producer hands over a batch exactly as wide as the schema. Iceberg's vectorized reader does not. BatchDeleteFilter.filterBatch reads with the delete filter's requiredSchema, which carries _pos after the projected columns when a data file has position deletes, and trims the extras back only when the file also has equality deletes. A merge-on-read UPDATE writes position deletes and no equality deletes, so the extra column survives into the batch, and caching such a relation failed with ArrayIndexOutOfBoundsException inside the write loop. Drive the loop from the writer's fields instead, which writes exactly the columns the schema describes: the extras are trailing, the same prefix Iceberg keeps when it does trim. A batch narrower than the schema is a genuine contract violation and is now refused with a message naming both widths. Closes apache#6087.
andygrove
added
the
run-spark-4.1-tests
andygrove
changed the title
…-default apache#5543, which this branch carried at an earlier revision, landed on main as e8eedcb. Its files take main's version, with this PR's change re-applied on top: spark.comet.exec.inMemoryCache.enabled defaults to true, and the in-memory cache guide says so.
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
First pass, thanks @andygrove!
| if (conf.getBoolean( | ||
| CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key, | ||
| CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.defaultValue.get)) { |
There was a problem hiding this 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. 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 helper that falls back to the entry's default, so the new read can use it too:
| if (conf.getBoolean( | |
| CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key, | |
| CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.defaultValue.get)) { | |
| 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 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)?
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 spark.comet.enabled and spark.comet.exec.enabled through getBooleanConf, and add the three plugin cases, before this comes out of draft. It won't avoid the slow path on its own, though: with both on, AQE still puts Spark operators above the cache scan, which is #6202.
There was a problem hiding this comment.
| "the disk half of the default MEMORY_AND_DISK storage level.") | ||
| .booleanConf | ||
| .createWithDefault(false) | ||
| .createWithDefault(true) |
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?
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 CometColumnarToRow (#6202). CometInMemoryCacheBenchmark runs with AQE off, so the published numbers don't show it. I'll fix #6202 first and then benchmark with AQE on against Spark's own format, so the number measures the path users will actually get.
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.
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.
| This feature is **experimental and enabled by default**. To turn it off, set the config at startup, | ||
| alongside the rest of Comet's configuration: |
There was a problem hiding this 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 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, 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.
There was a problem hiding this comment.
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.
|
I audited the in-memory cache and filed what came out of it. #6202 is the one that bears most on this PR. With AQE on, once the table-cache stage materializes, the re-plan leaves the operators directly above it on Spark, so a plain #6203 is about the signal this PR collects. 23 of the 53 cache test definitions check a cached query with I also added a cached-relation repro to #3079. A relation cached while Comet is on keeps native-shuffled partitions on a wide-decimal key, and a later join with Comet off returns 57 of 10000 rows. That doesn't depend on the cache format or on this PR, but caching is what carries it into queries that have Comet off. |
|
Moving this back to draft. We'll ship the in-memory cache in 1.1.0 the way it is on |
2 participants
Which issue does this PR close?
Part of #5487.
Rationale for this change
spark.comet.exec.inMemoryCache.enabledhas been off by default since #5051, so the native cache path only ever runs underCometInMemoryCacheSuiteandCometInMemoryCacheKryoSuite, which exercise it deliberately. Nothing tells us how it behaves under the rest of the suite: the Spark SQL test diffs, the fuzz suites, the Iceberg and Delta jobs, and any test that callscache()/persist()incidentally.This PR flips the default so a full CI run exercises the cache format everywhere caching happens. It is opened as a draft to collect that signal, not as a proposal to ship the feature on by default. A clean run is evidence the format is ready for that conversation; a red one is the list of things to fix.
What changes are included in this PR?
spark.comet.exec.inMemoryCache.enableddefaults totrue.CometDriverPlugin.maybeSetCacheSerializerread the config out ofSparkConfwith a hardcodedfalsefallback, so flipping theConfigEntryalone would have left the cache serializer uninstalled for anyone who did not set the key explicitly. It now falls back to the entry's own default, the same way the plugin already readsspark.comet.metrics.enabled.The feature is still described as experimental.
How are these changes tested?
The point of the PR is the CI run itself. Every job now builds cached tables in Comet's Arrow format wherever a test caches anything, rather than only in the two suites that opt in.
The existing cache suites are unaffected: they set the config explicitly, including the two cases that set it to
false.CometInMemoryCacheSuite's driver-plugin test passes an explicittrue, so it still covers the install path rather than relying on the new default.