[MSE] Adopt Apache Arrow as the native columnar block format and wire protocol
- Dominant language
- Java
- Stars
- 6.1k
- Forks
- 1.5k
- Avg merge
- 2d 55m
- Merged PRs (30d)
- 182
Description
### Summary
This issue proposes adopting Apache Arrow as the columnar in-memory format for the Multi-Stage Query Engine (MSE), replacing the current row-oriented Object[] representation used inside operators, and replacing the gRPC+protobuf wire protocol for inter-node data transfer with Arrow Flight.
Doc : https://docs.google.com/document/d/1koNPiW8L0IbijQi1JarFscH-4Ct7IF8Jv5HrP3fZw3k/edit?tab=t.o3y1fgpa2hkj
---
### Background
The MSE today operates on a fundamentally row-oriented data model. When a server executes a sub-plan and needs to send results to the next stage (e.g. a HashJoin or aggregation on the broker), the data goes through a chain of conversions that are both expensive and GC-heavy:
ColumnDataBlock (columnar, off-heap)
→ deserialize to Object[][] rows (on-heap, boxed)
→ operator logic (e.g. HashJoin build/probe on Object[])
→ re-serialize to SerializedDataBlock (off-heap)
→ gRPC / protobuf wire
→ deserialize again on the receiver
Every step in this chain has costs:
- Boxing overhead: primitives like long must be wrapped in Long for storage in Object[]. For a HashJoin over 10M rows this creates tens of millions of short-lived objects.
- Serde cost: serializing and deserializing to/from gRPC ByteString involves memory copies even for data that is already in a contiguous buffer.
---
### Proposal
Adopt Apache Arrow as the native in-memory and wire format for the MSE.
ArrowDataBlock / ArrowBlock — a new MseBlock.Data backed by Arrow's VectorSchemaRoot. Primitives live in typed vectors (IntVector, BigIntVector, etc.) with no boxing. Reference-counted so blocks can
be shared across threads without copying.
Arrow-native HashJoin — an ArrowLookupTable that merges right-side batches into a single contiguous block and builds a linear-probe hash table over column vectors directly, and an ArrowJoinProbe that
probes it with zero object allocation.
Arrow Flight transport — replaces gRPC+protobuf for inter-node data transfer. Since the data is already in Arrow format, the send path becomes zero-copy. The gRPC channel is kept for EOS/error
signalling to allow incremental rollout.
Layer | Current | Arrow proposal
-- | -- | --
Broker → client result encoding | Arrow IPC (already done) | No change
Operator in-memory format | Object[][] rows (boxed, on-heap) | VectorSchemaRoot (unboxed, off-heap)
Inter-node data transfer | gRPC + custom serde | Arrow Flight (zero-copy)
HashJoin build/probe | Per-row HashMap | Linear-probe table over columnar vectors
---
Compatibility and rollout
The change is designed to be backward-compatible:
- ArrowBlock implements MseBlock.Data and provides asRowHeap() / asSerialized() fallbacks so operators that have not yet been Arrow-ified continue to work unchanged.
- Flight and gRPC mailbox services run side-by-side; the sender selects the transport per connection.
- The DataBlock.Type.ARROW ordinal is added to the existing enum without changing existing ordinals.
---
Expected impact
Based on prototype benchmarks :
- Significant reduction in GC pause time during large joins (eliminating tens of millions of boxed object allocations per query).
- Lower serialization overhead on the inter-node path due to Flight's zero-copy transfer.
- Foundation for future vectorised operator implementations (SIMD-friendly columnar loops, vectorised aggregation, etc.).
---
Reference
- Related: Apache DataFusion (Arrow-native query engine used as reference)
- Arrow Java docs: https://arrow.apache.org/docs/java/
---
Contributor guide
Research direction
Start with the linked design document and the MSE data path described in this issue; the payload does not name implementation files, tests, or entry points. Done means adopting Arrow-backed blocks and Arrow Flight alongside the compatibility and rollout requirements described here.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- grpc, java
- Domain
- data-engineering, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100