[EPIC] Track arrow-java bugs and limitations that Comet works around #7362

What is the problem the feature request solves?

Comet uses arrow-java (pinned at 18.3.0 in pom.xml) for every batch that crosses the JVM/native boundary, for the in-memory cache's IPC format, for broadcast serialization, and for the Arrow Python runner. A number of arrow-java bugs and design limits show up on these paths, and Comet works around each of them in its own code. The workarounds are spread across the native FFI layer, the importer, the cache, shuffle and codegen, and several of them are rules that every new code path has to remember to follow.

This epic collects them in one place, so that each one can be tracked to an upstream fix (or a decision to keep the workaround) and the workaround removed once Comet moves to an arrow-java release that has the fix. The FFI-related ones are also described in the FFI contributor guide.

Describe the potential solution

arrow-java bugs that Comet works around

• C Data import ignores ArrowArray.offset at every level and reads each buffer from its start (upstream: [Java][C] sliced RecordBatch offset info is lost when imported from c-data arrow-java#88, open). Native runs zero_offsets (native/common/src/ffi_offsets.rs) on every array it exports to the JVM, re-slicing boolean bitmaps to start at bit 0 (Sliced booleans nested in structs, and sliced inputs to the JVM UDF bridge, reach the JVM misaligned and return wrong results #6288). Every new native-to-JVM export path must call it. An upstream fix would remove the workaround and the rule.
• VectorSchemaRootAppender and VectorUnloader assume offsets start at zero. An imported slice of a string, binary, list or map array has offsets that start past zero, so the appender folds the earlier rows into the first value it appends and IPC writes them into the stream. Utils.withRebasedOffsets re-slices with VectorSchemaRoot.slice before broadcast coalescing (Broadcast hash join produces incorrect results with sliced Arrow string/binary batches #6779), cache writes (A cached aggregate stores each output slice with the bytes of the slices before it, inflating its size statistics and memory #6995) and the Arrow Python runner. No upstream issue yet.
• Data.importField creates a new SchemaImporter per field, and each numbers dictionaries from 0, so two dictionary-encoded columns collide in the shared CDataDictionaryProvider and corrupt data. ArrowImporter shares one SchemaImporter for all columns, and lives in spark/src/main/java/org/apache/arrow/c/ to reach arrow-java's package-private SchemaImporter and ArrayImporter. A public API that shares dictionary-id state across fields would let Comet drop the class. No upstream issue yet.
• VectorLoader.loadBuffers leaks on a decompression failure. It holds a field's decompressed buffers in a local list until the whole field loads, so if one buffer fails to decompress, the buffers before it are never freed, and a single corrupt cached batch leaks off-heap memory for the life of the executor. CachedBatchIpc.decompressed does the decompression itself to keep every allocation on its own error path. No upstream issue yet.
• ZstdCompressionCodec passes a destination capacity 8 bytes too large (upstream: [IPC][ZSTD] Compressed buffer prefix can be written as 0 while ZSTD frame has non-zero content size arrow-java#1116, fix GH-1116: [Java] Fix compressed buffer prefix write and ZSTD dstCapacity arrow-java#1119 merged for 20.0.0, unreleased). Comet side: Executor JVM abort (NMT footer canary broken) when the in-memory cache is enabled on Spark 3.4 #7305, PR fix: give zstd the real destination capacity when compressing cached batches #7307.
• C Data import leaks native memory when the allocator runs out mid-import (upstream: C Data import leaks native memory when the allocator runs out of space mid-import arrow-java#1239, fixed for 20.0.0). ArrowImporter.importVector has its own failure cleanup of the vector, its dictionaries and the array. Revisit after the upgrade.

arrow-java design limits and sharp edges

• Struct children are keyed by name, so a struct with duplicate field names collapses on import. Comet rejects or falls back for such types in cast, native shuffle, the in-memory cache, local table scan and DataTypeSupport (Comet named_struct fails on duplicate field names #1015, Native shuffle accepts structs with duplicate field names, then fails importing the batch back to the JVM #5605). Open gap: arrays_zip with two same-named inputs fails with "ArrowArray struct has 2 children (expected 1)" #6251 (arrays_zip over two same-named inputs). A positional child API upstream would ease this, but it is a deeper change to arrow-java.
• An exported stream hands native only the text of the exception the reader throws in get_next. CometArrowStream keeps the throwable until the task completes and CometExecIterator rethrows it, so the task fails with the exception Spark would have thrown. An upstream way to retrieve the original exception would simplify this.
• ArrowArrayStream.close frees only the C struct. A stream that native never took leaves its reader open and pinned by the JNI global ref arrow-java holds on the private data. CometNativeArrowSource.stream registers a task-completion listener that releases untaken streams itself.
• ListVector.getField rebuilds the child label as "$data$" from the runtime data vector, and MapVector and StructVector pick it up through recursion. Codegen output uses RenamedListVector, RenamedMapVector and RenamedStructVector (CometBatchKernelCodegenOutput.scala) to pin the field names Comet expects on FFI export.
• The JVM allocator only guarantees 8-byte alignment, while arrow-rs needs Decimal128 buffers 16-byte aligned. arrow-rs 59 realigns under-aligned buffers on import (Call align_buffers() in from_ffi, remove redundant call from arrow-pyarrow arrow-rs#10030), which is a copy, and realigns_under_aligned_decimal128 in scan.rs guards against a downgrade. An arrow-java allocator option for 16- or 64-byte alignment would avoid the copy.
• Imported FFI buffers are charged to the allocator that wraps them (BaseAllocator.wrapForeignAllocation), which makes them indistinguishable from JVM allocations. Comet imports through a separate CometArrowImportAllocator child to tell them apart, and keeps a single root allocator that is never closed, since reference-counted buffers outlive the tasks that imported them.
• arrow-java's LZ4_FRAME codec is commons-compress's pure-Java LZ4, which is far slower than zstd and compresses worse. Comet ships Lz4FrameCompressionCodec on lz4-java, writing the same format. This could be offered upstream.
• getBufferSizeFor counts the validity bitmap even when every value is non-null. arrow_python_udf.rs mirrors this so its batch sizing matches the JVM's. Minor.

Upgrade

• Move from arrow-java 18.3.0 to 20.0.0 once it is released, which brings in GH-1116: [Java] Fix compressed buffer prefix write and ZSTD dstCapacity arrow-java#1119 (zstd capacity), C Data import leaks native memory when the allocator runs out of space mid-import arrow-java#1239 (import leak) and Lz4CompressionCodec trusts the uncompressed-length prefix when decompressing arrow-java#1206 (LZ4 decompression trusting the length prefix). Then remove or simplify the workarounds above that the release makes unnecessary.

Comet shades arrow-java (spark/pom.xml relocation) so that it is independent of the Arrow version Spark bundles, which is what makes picking its own version possible. That is a packaging choice, not an arrow-java issue, and is not tracked here.

Additional context

Items marked "No upstream issue yet" have no arrow-java issue that I could find. When one is filed, link it from the item here.