andygrove commented on PR #5051:
URL:
https://github.com/apache/datafusion-comet/pull/5051#issuecomment-5443337945
@viirya thanks — you were right on both counts, and the Kryo one turned out
to be wider than either of us had it. Fixed in 7e19dff20.
**On the blast radius.** Agreed, and your framing is what made me go
looking. #5487 framed it as a DISK_ONLY concern and that was under-scoped: the
`_SER` levels, replication, cross-executor fetches and the disk half of the
default `MEMORY_AND_DISK` all serialize the `CachedBatch`, so a plain
`.cache()` that spills is enough. Took your option 1.
**But it is not only the cache.** While writing the test I checked whether
the registrator needed anything beyond the cached batch, and
`CometBroadcastExchangeExec` broadcasts an `Array[ChunkedByteBuffer]` from
`Utils.serializeBatches`. Spark registers `ChunkedByteBuffer` in
`KryoSerializer.toRegister` but not an array of them, so a native broadcast
join fails under `registrationRequired=true` on `main` today, with no cache
involved:
```
java.lang.IllegalArgumentException: Class is not registered:
org.apache.spark.util.io.ChunkedByteBuffer[]
at
org.apache.spark.sql.comet.CometBroadcastExchangeExec.doExecuteBroadcast(CometBroadcastExchangeExec.scala:232)
```
Same registration fixes it, since the cache write path hands back the same
type, so I fixed it here rather than filing it — but it is a pre-existing bug
rather than one this PR introduces, so say the word if you would rather it were
split out.
**One thing I could not do, and it changes the shape of the fix.** Comet
cannot install this for the user. `KryoSerializer` reads
`spark.kryo.registrator` into a `val` in its constructor, and `SparkEnv.create`
builds it at `SparkContext.scala:478`, whereas `PluginContainer` is constructed
at line 574. So unlike `spark.sql.cache.serializer` — a `StaticSQLConf` read
lazily from the session, which is exactly why `maybeSetCacheSerializer` can
inject it — this one is already captured by the time any plugin runs. Setting
it from the driver plugin would reach executors but not the driver's own
`SparkEnv`, so it would work in cluster mode and fail in local mode, which is
worse than a documented requirement.
So the fix is three parts rather than one:
* `org.apache.comet.CometKryoRegistrator`, with the class list split between
`Utils.arrowBytesKryoClasses` (the Arrow-bytes carriers, shared by broadcast
and cache) and `ArrowCachedBatchSerializer.kryoClasses` (the cached batch and
its statistics row).
* Documented on `spark.comet.exec.inMemoryCache.enabled`, as your option 2,
since option 1 does not remove the need to tell people about it.
* `CometDriverPlugin` warns at startup when `spark.serializer` is Kryo,
`registrationRequired` is true, and the registrator is absent, naming the conf
to add. That is the part that addresses "hard to attribute" — the warning fires
before anything is cached or broadcast.
**Two findings worth recording from building the list.**
`DefaultCachedBatch` had to be registered too. A schema Comet cannot store
is delegated to Spark's serializer, and Spark registers that class itself only
from 4.1 — 3.4, 3.5 and 4.0 do not, along with `GenericInternalRow`. So the
delegated path fails on those versions without Comet registering it, and some
entries that look redundant on 4.1 are load-bearing on 3.5.
`java.math.MathContext` cannot be registered at all on JDK 17 —
`kryo.register` builds a `FieldSerializer` eagerly and gets
`InaccessibleObjectException: module java.base does not "opens java.math"`.
Decimal bounds above `Long` precision survive anyway because Chill's
`scala.math.BigDecimal` serializer writes the `java.math.BigDecimal` inside as
a class-and-object rather than reflecting over the `MathContext`, which is why
the list registers the `java.math` types and not the Scala one. Worth knowing
before anyone tries to extend this list by intuition.
**Testing.** `CometInMemoryCacheKryoSuite`, four tests, all
reverted-and-rechecked rather than assumed: cache round trip at `DISK_ONLY` and
`MEMORY_AND_DISK_SER` over a relation carrying every type whose bounds the
statistics row records, read back through a predicate so the bounds go through
Kryo too; the native broadcast; and the delegated `DefaultCachedBatch` path.
Without the registrator they fail with exactly the `Class is not registered`
you predicted. Green on the 3.4, 3.5, 4.0 and 4.1 profiles, along with
`CometInMemoryCacheSuite`, `UtilsSuite`, `CometArrowStreamSuite`,
`NativeUtilSuite` and `CometPluginsSuite`.
I did not test the `MEMORY_AND_DISK` spill case itself. Both levels in the
suite serialize deterministically on put; forcing a spill is not cheap to
arrange, so the case that motivated your comment is covered by argument rather
than by a test.
**On the nit** — confirmed, that is intended and not a bug. Because
`spark.sql.cache.serializer` is static, the format is an application-lifetime
decision made at startup; flipping `spark.comet.exec.inMemoryCache.enabled` off
later only routes scans back to Spark's execution path and the footprint stays
Arrow. The config doc says that, and I would rather keep the honest explanation
than split the flag, since a relation whose cached format could change
mid-session could not be read back reliably.
Agreed on AQE (#5245) being the follow-up to weight most, and SPARK-37742
within it for the reason you and @peterxcli both landed on.
--
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]
