andygrove opened a new issue, #7305: URL: https://github.com/apache/datafusion-comet/issues/7305
### Describe the bug An executor JVM aborted with a native-memory corruption error while a job was writing to and reading from Comet's in-memory cache (`spark.comet.exec.inMemoryCache.enabled=true`). The same job completes when the in-memory cache is disabled. I could not reproduce it locally, so this issue records what the crash shows and what has been ruled out. The JVM reported: ``` Internal Error (mallocTracker.cpp:169) fatal error: NMT corruption: Block at 0x...: footer canary broken at 0x... (buffer overflow?) Problematic frame: V [libjvm.so+0x...] MallocTracker::record_free(void*)+0xf0 ``` Native Memory Tracking (NMT) puts a canary after every block the JVM allocates. The check fires when the block is freed, so something wrote past the end of the block earlier. The abort site says nothing about who wrote it. ### Environment - Spark 3.4.3, Kubernetes, executors with 16 cores, a 128 GB heap and 8 GB of off-heap memory - JDK 17.0.18, Linux aarch64, G1 - A recent Comet build; the exact version still has to be confirmed - NMT was enabled on the executors - `spark.sql.cache.serializer` was set by `CometDriverPlugin` to `ArrowCachedBatchSerializer` (`CometDriverPlugin: Auto-set spark.sql.cache.serializer=...`), because `spark.comet.exec.inMemoryCache.enabled=true` - The cache compression codec and the storage level are not known yet. The executor log shows cache blocks stored as values (`rdd_<n>_<p>` blocks of about 70 MiB, "stored as values in memory"), so the storage level is a deserialized one. ### What the job does It reads a wide Iceberg table that has complex types (arrays, structs, maps) and binary columns holding serialized protobuf messages. It persists the result, then runs several queries over the cache, including a scalar UDF and an `explode`. 16 tasks run per executor. The crash came while tasks of the stage that materializes the cache were storing blocks. Several executors were lost within about a minute. ### What the crash shows - The corrupted block is 512 bytes. It holds a zstd frame (it starts with the zstd magic number `28 b5 2f fd`) that decompresses to 437 bytes. - That matches the output of Arrow's `ZstdCompressionCodec`: `compressBound(437) + 8`, rounded up to the allocator's power-of-two size, is 512. The cache compresses each Arrow buffer with that codec. - The first 8 bytes of the block hold no valid length prefix. It could be a block that was freed or reused, or one that was overwritten. I can't tell which from the dump. - With the in-memory cache disabled the job completes, so the cache is the prime suspect. The cache-off run uses neither the cache serializer nor the cache writer. ### What I tried locally Seven shapes, all on Spark 3.4, `local[16]`, a debug libcomet, and NMT detail on, so an overrun would trip the same canary check. None crashed or produced a wrong result. - About 120k rows and 210 columns of strings, longs, doubles, `decimal(18,2)`, `decimal(38,10)`, `array<string>`, `array<struct>`, `map<string,string>`, nested structs and binary. The binary columns held both 200 KB values and incompressible 64 to 512 byte values. Nulls and empty collections were included. - `persist(MEMORY_AND_DISK)`, then 8 queries run concurrently on 6 threads, for 3 to 6 rounds. They covered group-by, filter, `explode`, nested-field access, a full-width noop write, a 150-column projection and per-column checksums. - Codecs zstd (five shapes), lz4 and none. Cache batch sizes from 300 to 10000, with various `spark.comet.batchSize` values, with AQE on and off, and with the native scan and a Spark scan converted to columnar. Not covered: a real Iceberg table (Parquet only), Linux aarch64 with a release libcomet and large heaps, eviction and disk spill of cached blocks under memory pressure, and the UDF from the job. ### Leads (none confirmed) 1. **Arrow's zstd codec passes zstd a destination capacity 8 bytes too large.** In arrow-compression 18.3.0, `ZstdCompressionCodec.doCompress` allocates `8 + compressBound(n)` bytes and calls `Zstd.compressUnsafe(dst + 8, 8 + compressBound(n), ...)`. The extra 8 bytes cannot overflow while zstd stays within `compressBound`, and I could not make it overflow, so this is a latent defect rather than a cause. 2. **Arrow's bounds checks are off.** Comet sets `arrow.enable_unsafe_memory_access=true` in `NativeBase.setArrowProperties`. A Java-side overrun in the cache write or read path then corrupts memory silently instead of throwing, and shows up only as the canary failure when the block is freed. 3. **The Spark 3.4 cache build uses `ArrowWriters.scala`.** It writes with raw `Platform.copyMemory` and `putLong` into Arrow buffers and does its own capacity handling for validity, offsets and nested children. The array, map and struct writers are the first place I would read for a missed capacity check. ### What would help - The `hs_err_pid*.log` from the crashed executor. Its Java stack for the crashing thread and its Native Memory Tracking section would show which allocation was corrupted. - Whether the job still crashes with `spark.comet.exec.inMemoryCache.compression.codec=none` (and with `lz4`). A clean `none` run would point at the compression step. A crash with `none` would point at the writers or the read path. - The cache storage level and whether any cached blocks were evicted or spilled to disk before the crash. ### Related EPIC #6834 (in-memory cache audit). Nothing in it describes a crash. -- 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]
