viirya commented on PR #5051:
URL:
https://github.com/apache/datafusion-comet/pull/5051#issuecomment-5441403778
Reviewed the current HEAD (`70f046abf`). This is in good shape — the design
boundary is clean (the serializer owns only the cached payload; Spark's cache
manager keeps lifecycle, storage, and eviction), the fallback path is careful
(`enabled && cometCacheFormat`, with distinct fallback reasons, and the
guarantee that turning it on is never worse than off), and the comments
explaining the non-obvious decisions — the empty-output/`count(*)` handling,
`doCanonicalize`, deferring `cachedColumnBuffers` to execution, the fixed
`NATIVE_TIMEZONE` label, the pruning-safety restriction in `buildFilter` — are
genuinely helpful and clearly the residue of the earlier review rounds. I
re-checked the fixes for @sunchao's findings (zero-column join output,
per-column dictionary providers across the cache-write / broadcast / shuffle
paths, reader-leak-on-init, canonicalization) and they hold; I didn't find new
sibling instances of those bug-classes.
One thing I'd like to see addressed before merge, and one nit.
**`spark.kryo.registrationRequired=true` — the blast radius is wider than
"DISK_ONLY".**
This is the Kryo-registration gap you already recorded in #5487 ("caching
fails outright under `spark.kryo.registrationRequired=true`"), so the finding
itself isn't in dispute — but I think it's under-scoped there, and that changes
whether it should be a pre-merge item.
#5487 (and the natural reading of it) frames this as only affecting
explicitly serialized storage levels, with DISK_ONLY as the path that exercises
it. But a `CachedBatch` goes through the configured serializer in more cases
than that:
- the `*_SER` levels (`MEMORY_ONLY_SER`, `MEMORY_AND_DISK_SER`),
- replicated levels (`_2`) and any cross-executor block fetch,
- and — the one that matters — **the disk-spill portion of the default
`MEMORY_AND_DISK`** that a plain `df.cache()` uses. The in-memory copy is
deserialized, but once a partition spills, it's serialized.
So under `spark.kryo.registrationRequired=true`, an ordinary `.cache()` that
spills is enough to hit this; the user doesn't have to reach for DISK_ONLY. And
because `CometCachedBatch` is unregistered, the failure surfaces as Kryo's
"Class is not registered" rather than anything pointing back at this feature,
so it's hard to attribute.
The existing DISK_ONLY test doesn't cover it — the suite never sets a Kryo
serializer, so that test runs on the default Java serializer.
Suggestion, in order of preference:
1. Register `CometCachedBatch` (and the classes it carries) in a Comet Kryo
registrator — Spark does exactly this for its own `ArrowCachedBatch`, so
there's a direct precedent — plus a `KryoSerializer` +
`registrationRequired=true` DISK_ONLY regression test. Small, and it closes a
hard-failure hole.
2. If you'd rather keep it a follow-up: at minimum document it on
`spark.comet.exec.inMemoryCache.enabled`. That doc already lists several
limitations but not this one, and the near-zero-cost win is turning a silent
hard failure into a documented one.
I'd lean against leaving it purely as #5487 given that plain `.cache()` +
spill can trip it, not just DISK_ONLY.
**Nit:** the `enabled` config conflates two things — whether
`CometDriverPlugin` installs the serializer (decided at startup, since
`spark.sql.cache.serializer` is static) and whether scans run natively. The doc
explains this honestly, so this is just a "confirm it's intended": once the
feature has been on at startup, every supported relation is cached in Arrow
format for the life of the application even if the flag is later flipped off
(that only reroutes scans back to Spark). That's a reasonable consequence of
the static conf, not a bug — flagging it only because a reader turning the flag
off at runtime might expect the cache footprint to revert too.
Overall I think this is mergeable as an experimental, default-off feature
along the "merge and iterate" lines you proposed, with the Kryo item as the one
I'd want handled (fix or document) first. The AQE gap (#5245) — especially the
SPARK-37742 stats-divergence you called out, where Comet's own size accounting
feeds AQE's join-strategy choice — is the follow-up I'd weight most after this
lands.
--
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]
