andygrove opened a new issue, #7362: URL: https://github.com/apache/datafusion-comet/issues/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](https://github.com/apache/datafusion-comet/blob/main/docs/source/contributor-guide/ffi.md). ### 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: apache/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 (#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 (#6779), cache writes (#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: apache/arrow-java#1116, fix apache/arrow-java#1119 merged for 20.0.0, unreleased). Comet side: #7305, PR #7307. - [ ] **C Data import leaks native memory when the allocator runs out mid-import** (upstream: apache/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` (#1015, #5605). Open gap: #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 (apache/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 apache/arrow-java#1119 (zstd capacity), apache/arrow-java#1239 (import leak) and apache/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. -- 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]
