Performance optimizations for native in-memory cache (follow-on to #4591)

[andygrove]

Is your feature request related to a problem or challenge?

PR #4591 adds native support for Spark's in-memory cache via ArrowCachedBatchSerializer and CometInMemoryTableScanExec. The goal is to land the functional version first. This issue tracks performance optimizations identified during review that can follow in later PRs.

Benchmark from #4591 (CometInMemoryCacheBenchmark, 5M-row cached table, Apple M3 Ultra, JDK 17, Spark 3.5) showed the native path is about 1.5x faster on a full repeated scan but only about 1.1x on a selective filter, which suggests there is headroom, especially on the read path.

Describe the potential solution

Read path (runs on every cached scan, highest impact)

1. Hoist UnsafeProjection.create out of the per-batch loop. In ArrowCachedBatchSerializer.convertCachedBatchToInternalRow, UnsafeProjection.create(...) is currently constructed inside the flatMap over decoded batches, so it recompiles Janino codegen once per cached batch. It should be built once per partition (inside mapPartitions, outside flatMap).

2. Reduce the per-scan deep copy. CometInMemoryTableScanExec sets arrow_ffi_safe = false, so scan.rs deep-copies every column of every surviving batch on each scan. This is correct today because ArrowReaderIterator owns and closes each decoded batch, but it is the main structural cost on the read path and is the likely reason the selective-filter benchmark only reached ~1.1x. Removing it requires reworking the decode path to transfer ownership to native rather than closing the batch on the JVM side. This is a larger, riskier change and should be scoped carefully.

3. Optional uncompressed cache format. Utils.decodeBatches decompresses (LZ4 by default) on every scan. A knob to store the Arrow bytes uncompressed would trade memory for faster repeated reads.

Write path (runs once per cache(), but slow for large caches)

4. Specialize computeStats per column and stop boxing. readValue returns Any, so every primitive read boxes, and both compare(dt, ...) and tracksBounds(dt) re-dispatch on DataType per row even though the type is invariant per column. Dispatch on the type once per column, then run a specialized primitive loop that keeps min/max in primitive locals.

5. Copy-on-update for string stats. readValue does col.getUTF8String(rowId).copy() for every non-null string, but only the min and max are kept. Compare against the current bound using the borrowed UTF8String and copy only when it becomes the new lower or upper bound.

Suggested priority

1. Hoist UnsafeProjection.create (trivial, clear win).
2. Specialize computeStats and add copy-on-update for strings (biggest build-time win).
3. Investigate reducing the per-scan deep copy, with a scan-isolating benchmark variant to confirm the pruning benefit.

Additional context

Follow-on to #4591. See the review discussion there for full detail.

[pchintar]

Hi @andygrove two of the optimization points mentioned above no longer seem relevant:

1. For UnsafeProjection.create, that's already addressed in feat: Add Native Support for In-Memory Cache #4591 . It's already created once per partition outside the per-batch loop in convertCachedBatchToInternalRow, so the projection is reused across all cached batches in the partition instead of being rebuilt repeatedly.

2. I also investigated the per-scan deep-copy point. After the Arrow C Stream migration in perf: replace CometBatchIterator FFI input path with the Arrow C Stream Interface #4572, the implementation has changed significantly. The old arrow_ffi_safe/setArrowFfiSafe path has been removed, and ScanExec now uses copy_or_unpack_array(..., CopyMode::UnpackOrClone). For non-dictionary arrays this just performs an Arc::clone (shared ownership of the existing Arrow buffers), not a deep copy. Dictionary arrays are still unpacked and copied because the Arrow dictionary-unpack path may reuse the input null buffer, and the code explicitly documents that safety requirement.

So the previous concern about every surviving batch being deep-copied on each scan no longer applies on the current codebase. The remaining copy is specific to dictionary arrays rather than all scanned columns. Given that, my next focus after PR #4591 is merged is to work on the remaining optimization areas you raised such as Specialize computeStats, remove boxing, and add copy-on-update for strings. Wdyt @andygrove ?