[Proposal] Native, vectorised, zero-copy execution path for Druid

[Shekharrajak]

Motivation

Native, vectorised, zero-copy execution path for Druid via FFM and Arrow — staged from segment reader → filter/project → DataFusion (follow-up to #19039 item 1)

Druid today does the entire query hot path on the JVM. Each row passes through a ColumnSelector interface, materialises into Java objects, gets compared/aggregated by JIT-compiled byte code - means :

• No SIMD on filter / project / aggregate hot loops.
• GC pressure from per-block wrapper objects on cold-cache scans.
• 2 GiB ceiling on off-heap buffers (see Speed up node boot times by parallelizing buffer acquisition #19025).
• Bottleneck on MSQ joins / aggregates

Proposed changes

Phase 0 — Spike: native segment column reader via FFM
Phase 1 — Layer0 production-shape: native column reader for all common types
Phase 2 — L1: vectorised native filter + project over Arrow batches
Phase 3 — L2: DataFusion as candidate aggregate/join executor in MSQ stages

Rationale

Operational impact

Performance 2–3× speedup on

• speedup on cold-cache scan + filter for typical Druid datasources.
• speedup on MSQ fact-fact joins.
• speedup on external Parquet ingestion.
• lower p99 for high-cardinality GroupBy.
• faster execution under memory pressure (DataFusion graceful spill).

Operational

• 2 GiB ByteBuffer ceiling lifted on the native path (aligns with Speed up node boot times by parallelizing buffer acquisition #19025).
• Reduced GC pressure → fewer p99 latency spikes.

Test plan (optional)

Future work (optional)

[jtuglu1]

Thanks for raising this. I was waiting until Druid v38 was cut (so compiler can be switched to JDK 25). This way, we can use latest version of FFM.

[jtuglu1]

I think using Datafusion would be a good idea (much better than the current query planner/execution engine set up), but one thing to call out here is that Datafusion always materializes rows into ArrowBatch between operators, which is a bit different in some cases compared to Druid. This materialization cost can be expensive.

[jtuglu1]

To be clear, I would much rather rely on Datafusion to implement the core physical operators (since they do it much better, and we get future performance improvements for free), the only caveat being we will need to ensure we can get the same or better read amplification than the current native JDK Druid code does, e.g. where we can rely on column stats, row count, bitmap/dictionary indices to pushdown filters/projections and avoid decompressing/scanning more than we need to.

[Shekharrajak]

Thanks @jtuglu1 for checking.

Yes, I understand the complexity and work for this initiative due to JNI/FFM and prerequisite if we want vectorization support.

We can actually evaluate the Java bindings for Apache DataFusion: https://github.com/apache/datafusion-java and see if those work can be minimised.

[gianm]

Interesting stuff. Will you be working on this? One nudge, you might have luck doing things in the MSQ path (the FrameProcessors) by using a RowsAndColumns backed by Arrow rather than Druid's builtin Frame.

It may be easier to experiment with the nonleaf paths first. The nonleaf paths are the ones that read from ReadableFrameChannel. They today read Frame and could probably be swapped out for something that processes Arrow without too much fuss.

The leaf paths— the ones that read Segment— may require some additional care to avoid materializing things too early. Druid's leaf computation pipeline pushes down a lot of operations such that they run directly on a Cursor or VectorCursor. Even a fairly complex set of operations like filter -> project -> broadcast hash join -> aggregate can all get pushed down, with a Frame being created with the results of the aggregation.

[Shekharrajak]

@gianm Yes, this can be assigned to me (I am involved in datafusion community as well - and analysing datafusion-java : https://lists.apache.org/list.html?dev@datafusion.apache.org for this usage) but as know this will touch multiple components - other contributors are open to explore, help in design/architect and evaluate design decisions.

I will analyse your comment and come up with some draft.

[jtuglu1]

The leaf paths— the ones that read Segment— may require some additional care to avoid materializing things too early.

This is a pretty well-known issue in Datafusion where you need to implement a lot of custom pushdown (e.g. physical operators to read/filter on segment bitmaps/dictionaries) yourself to avoid expensive re-materializing between operators. This will be especially painful if we need to cross the JVM/Rust boundary frequently.

Another thing I want to call out here is starting to move towards deprecating and unifying the available engines in Druid. Currently I see a lot of work being done on various engines (dart, msq, native, etc.) and what I think we should be aiming towards is the native processing core to be built out on whatever the decided "next" gen engine is (and its operator interfaces). While I think the native segment readers are a bit more generic and can be shared, this kind of thing will help motivate a deprecation of older execution paths.

@gianm I think there's a lot of value in considering adoption of Datafusion's planner/optimizer/physical operator ecosystem in the long term; while I think there will need to be many Druid-specific operator overrides, to me it seems like a net-positive if we're able to get statistics support, advanced planning (better join algos, etc.), and optimized physical operators (spilling grouping, etc.) out-of-the-box without needing to worry about implementing these ourself. That being said, I think distributed Datafusion is still in a nascent phase (Ballista, etc.), so this is definitely a longer-term thing.

[gianm]

Another thing I want to call out here is starting to move towards deprecating and unifying the available engines in Druid. Currently I see a lot of work being done on various engines (dart, msq, native, etc.) and what I think we should be aiming towards is the native processing core to be built out on whatever the decided "next" gen engine is (and its operator interfaces).

This is another reason I mentioned a nudge towards MSQ. In addition to having a Frame concept that should be relatively straightforward to swap out for an Arrow batch, it is also a generally more capable engine and one that I expect we'll standardize on in time.

Fwiw, MSQ and Dart are the same underlying engine. Really MSQ is the engine and Dart is a "profile" of the engine. It is a mode where the Broker runs query controllers and the Historicals are query workers. The other MSQ profile is Tasks, where the controllers and workers are all launched as tasks.

Anyway, I think a lot of interesting experiments could be done without necessarily needing to get into the leaf processing stuff right away. For example, the MSQ sortMergeJoin implementation operates purely on Frames and could be swapped out for something else. The MSQ groupBy implementation has a per-segment partial-aggregation phase that emits Frames, followed by a phase that sorts those Frames, followed by a merge of sorted streams of Frames. Another experiment there could keep the per-segment partial-aggregation stuff as-is, and swap out the sort-to-sort-merge part with some other form of aggregation operation.

[Shekharrajak]

While deep diving into MSQ, Frame processor - we can see it would be hard to completely adopt Arrow as of now.
But we can start with something that already suppot Arrow - that get me idea about Iceberg Arrow reader : #19498

I think this will help us understanding of arrow dependency and how it can work along with current druid memory model.

[Shekharrajak]

Big win - we see improvements in arrow reader implementation for read :
#19510 (comment)

Working on refactoring and missing corner cases .