apache / apache/pinot

[MSE] Adopt Apache Arrow as the native columnar block format and wire protocol

Open
#18,205 2 comments 0 reactions 0 assignees View on GitHub
multi-stage PEP-Request
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.