Repository navigation
Open
Labels
area:ffiArrow FFI / JNI boundaryArrow FFI / JNI boundaryenhancementNew feature or requestNew feature or requestperformance
Description
What is the problem the feature request solves?
Summary
When a native operator reads from a broadcast exchange, the data currently passes through Arrow FFI via ScanExec. This is the same path that shuffle reads used before ShuffleScanExec was introduced in #3731.
Current Path
CometBroadcastExchangeExecserializes data as coalesced, compressed Arrow IPC on the driver- On each executor,
CometBatchRDD.compute()deserializes the IPC stream →ColumnarBatchobjects on the JVM - A
CometBatchIteratorexposes those batches to native code - Native
ScanExecreads via Arrow FFI (CometBatchIterator.next()exports via Arrow C Data Interface)
Proposed Path
Introduce a BroadcastScanExec native operator analogous to ShuffleScanExec:
- Pass raw compressed Arrow IPC bytes from the JVM to native code (skip JVM-side deserialization)
- Decompress and decode natively using
read_ipc_compressed() - This avoids the JVM-side decompression +
ArrowStreamReaderdeserialization + Arrow FFI export overhead
Work Required
- New JVM iterator class (e.g.
CometBroadcastBlockIterator) that yields raw compressed IPC bytes, similar toCometShuffleBlockIterator - New protobuf op
BroadcastScanalongside existingShuffleScan - New Rust operator
BroadcastScanExecinnative/core/src/execution/operators/ - Planner changes in
CometSink.scalato emitBroadcastScaninstead ofScanfor broadcast exchange inputs (behind a config flag likespark.comet.exec.broadcast.directRead.enabled)
Notes
- Broadcast data is already marked
arrow_ffi_safe=truesince it comes fromArrowStreamReaderwith no mutable buffers, so the current FFI path is at least safe - Broadcast volumes are typically smaller than shuffle, so the impact may be less dramatic than perf: stop using FFI in native shuffle read path #3731, but the overhead of JVM-side decompression and FFI export is still real
Describe the potential solution
No response
Additional context
No response
Metadata
Metadata
Assignees
Labels
area:ffiArrow FFI / JNI boundaryArrow FFI / JNI boundaryenhancementNew feature or requestNew feature or requestperformance
Type
Projects
- StatusShow more project fieldsIn progress
Activity
andygrove commentedon May 6, 2026
We had good results doing this same work for shuffle in #3731
goutamadwant commentedon Aug 29, 2026
@andygrove Implemented this enhancement in #5538. Please let me know if you have any feedback. Thanks!
andygrove commentedon Sep 27, 2026
The current-path description here predates #4572.
CometBatchIteratorandarrow_ffi_safeare gone. Today the broadcast is decoded on the JVM byUtils.decodeBatchesand re-exported to native once per partition through the Arrow C Stream, viaCometArrowStream.wrapColumnarBatchRDD,ColumnarBatchArrowReaderandScanExec. The per-column export no longer exists, but the JVM-side decompression and decode, and the stream hop, still do, so the proposal still stands. #5538 implements it.Merge remote-tracking branch 'upstream/main' into fix/apachegh-3762-b…