Druid Modernization Ideas

[jtuglu1]

Motivation

I'm making a running list of Druid modernization ideas that I see other engines in the space have adopted/are already adopting that Druid could potentially implement as well. Please feel free to comment/add/edit the list of ideas below.

1. Query Latency/Throughput

Without hw-native implementation of physical operators, Druid lags behind other engines that implement query processing codepaths using SIMD/pipelining instructions and other native code speed-ups. Another factor to this conversation is avoiding garbage collection in high allocation/spilling scenarios. Given JDK 22's support for FFI, I wonder if it makes sense to consider adding support for these accelerations in something like Rust and plugging them into the existing Druid query processing path. The realtime streaming path could also potentially benefit from these changes where things like GC spikes can sink your p99 ingest throughput/increase query latencies.

This kind of split between parsing/planning and processing/execution is already being adopted by initiatives like https://github.com/facebookincubator/velox and https://github.com/StarRocks/starrocks.

2. Data ETL in/out of Druid

The Druid segment format, while hyper-optimized for workloads within Druid, serves little value to external ETL services (Spark, etc.) and data manipulation libraries that practitioners are familiar with (pandas, arrow, etc.). I think it would be a good idea to add Apache Arrow reader/writer support for Druid segments (that would allow any 3p system that speaks Arrow to integrate with Druid). It would also open up a path for switching the internal data transfer path (peon/historical -> broker -> router -> client) to use Arrow (instead of json) as well which could potentially speed up queries significantly.

3. CBO for MSQe

Currently the querying setup is split between native engine and MSQE. As MSQE matures, I believe the plan is to deprecate the native engine. To be competitive with other engines like Starrocks, etc. who have CBO/statistics-based planning for queries, I think it would be a good idea to add this to MSQ (this would involve tracking things like query/datasource-level statistics, etc. and exposing them in the planner).

4. Separation of storage/compute

This work has already begun with @clintropolis work on demand-based segment fetching. Not sure if there's anything else to add here.

5. SQL As First-Class Citizen

Engines like Clickhouse, etc. have most/all of their data/cluster configuration through a SQL DDL. This unifies the place where users need to go to make changes and exposes a configuration interface that's familiar to users. I think creating SQL DDLs for configs adjacent to "data" is a good place to start: retention rules, supervisor specs, etc.

6. Internal Metrics Ingestion

Druid should have an option to ingest its own emitted metrics and provide a queryable interface to them through the sys table. This will help Druid be more self-contained; I will note there is merit to decoupling of metrics from the system the metrics are observing, but I can see value for things like smarter janitoring, CBO for queried columns, auto-scaling, etc.

[maytasm]

@jtuglu1 Thanks for these great ideas. I think each one deserves its own issue, since they’re each fairly substantial on their own. This will keep the conversations clearly separated and make it easier to track progress (e.g., linking PRs to issues).

[gianm]

My 2¢ on each of these,

1. Query Latency/Throughput: I've had the exact same thought, of using FFM to load up some optimized processing code for certain operations. I'm not aware of anyone actively working on it, but it would definitely be an interesting experiment. It's something I've spoken with @clintropolis about in the past, so, tagging him. Probably the place to start would be reading the actual data out of the segments. Btw, now that we're close to dropping support for Hadoop, the pathway is open to raising the minimum Java version all the way to Java 25. I think we'd need changes to the Parquet reader (it doesn't work on Java 25), maybe some other stuff. Even before we do that, an experiment could be done. The core code of Druid works fine on Java 25. We run it there in production in environments that don't need Parquet support.

2. Data ETL in/out of Druid: Are you suggesting having an option to do a segment format that is internally Arrow, or are you suggesting keeping the current segment format, but writing some bridge code so it can be read into Arrow objects? Fwiw, as to the internal data transfer path, in MSQ we are already not using JSON. We're using WireTransferable internally, generally with an embedded Frame, which is an efficient binary format that can be either row- or column- oriented. So I wouldn't worry too much about that side of things.

3. CBO for MSQe: Yes, we're working towards being able to deprecate the native engine. As to CBO/statistics based planning, of course there is support in Calcite there for that. CBO tends to be most useful for queries involving multi-table joins, because it can do things like select join algorithm, reorder tables, etc. In other situations there just isn't as much for a CBO to do. This case (optimizing the planning of multi-table joins) hasn't been a priority for us so far, but if it becomes a priority for anyone, then sure, getting statistics into CBO via Calcite would be the way to go.

4. Separation of storage/compute: Yes, as you point out, active area of work. The initial story is pretty much there now: you can switch on virtual storage on the Historicals and then they'll start loading/evicting data as needed. Next step would be partial downloads. I think we'll start with doing partial downloads at the level of projections, because we have some internal use cases for that flavor of partial downloads.

[jtuglu1]

Btw, now that we're close to dropping support for Hadoop, the pathway is open to raising the minimum Java version all the way to Java 25.

I think we've dropped support already (in mainline) 😅. Clint has an issue here: #18980

Probably the place to start would be reading the actual data out of the segments.

Since this is I/O-bound (from object store) and (maybe?) CPU-bound on decompression when paging in from disk, I'd imagine we'd only get speed-ups here if we can "split" up the Druid segment format to be able to read a single range of rows/column at a time. I was thinking moreso on implementing actual physical operators (e.g. aggregation/filtering logic in a more CPU-friendly language). Something that is happening in this space is https://github.com/apache/datafusion. Basically a next-gen version of Calcite but a backend (physical operator implementations) are also implemented.

WRT to I/O, I think there's merit to exploring recent async IO capabilities in the segment readers. Anything that can speed up throughput would be helpful. Related blog: https://lancedb.com/blog/one-million-iops/

[jtuglu1]

Are you suggesting having an option to do a segment format that is internally Arrow, or are you suggesting keeping the current segment format, but writing some bridge code so it can be read into Arrow objects?

The latter. I suspect doing the former would be a lot more work and also cause a split between data format which might reduce adoption.

We're using WireTransferable internally, generally with an embedded Frame, which is an efficient binary format that can be either row- or column- oriented.

Have we benchmarked this against Arrow buffers?

[jtuglu1]

Next step would be partial downloads.

Do you mean being able to read a range (row range/column) out of a segment w/o needing to read the whole file?

[gianm]

Are you suggesting having an option to do a segment format that is internally Arrow, or are you suggesting keeping the current segment format, but writing some bridge code so it can be read into Arrow objects?

The latter. I suspect doing the former would be a lot more work and also cause a split between data format which might reduce adoption.

Got it, that makes sense.

We're using WireTransferable internally, generally with an embedded Frame, which is an efficient binary format that can be either row- or column- oriented.

Have we benchmarked this against Arrow buffers?

I have not.

[gianm]

Next step would be partial downloads.

Do you mean being able to read a range (row range/column) out of a segment w/o needing to read the whole file?

There's three kinds of partial downloads that come to mind:

• Read one projection
• Read one column
• Read part of one column (like Parquet "row groups")

I believe the first one we're likely to need (and therefore implement) is "read one projection".

[gianm]

Probably the place to start would be reading the actual data out of the segments.

Since this is I/O-bound (from object store) and (maybe?) CPU-bound on decompression when paging in from disk, I'd imagine we'd only get speed-ups here if we can "split" up the Druid segment format to be able to read a single range of rows/column at a time. I was thinking moreso on implementing actual physical operators (e.g. aggregation/filtering logic in a more CPU-friendly language). Something that is happening in this space is https://github.com/apache/datafusion. Basically a next-gen version of Calcite but a backend (physical operator implementations) are also implemented.

We can already read a single range of rows within a column at a time. Each column is block-compressed and we generally read one block at a time. Unless you meant something else?

Could be. If we start there then the initial version would need to read data out of Java objects, since that's what the current segment-reading code generates. It would mostly be primitive arrays for the data, ByteBuffers for string dictionaries, and ByteBuffer-backed ImmutableRoaringBitmap for bitmap indexes. I was thinking of starting with segment-reading code because we could potentially bypass that pit-stop in Java object land, by having e.g. native filtering code use native segment-reading code. It can always be changed later, though, so it probably isn't that important where we start exactly. (Native filtering code that initially starts out reading Java objects could later be changed to call native segment-reading code.)

[jtuglu1]

I was thinking of starting with segment-reading code because we could potentially bypass that pit-stop in Java object land, by having e.g. native filtering code use native segment-reading code.

This makes sense, yeah.

[FrankChen021]

After upgrading to JDK21, one thing I think we can solve is to replace ByteBuffer by MemorySegment. The ByteBuffer has 2GiB size limit which imposes lots of restrictions in Druid, from the ingestion to querying.

[jtuglu1]

After upgrading to JDK21, one thing I think we can solve is to replace ByteBuffer by MemorySegment. The ByteBuffer has 2GiB size limit which imposes lots of restrictions in Druid, from the ingestion to querying.

Yes. See my comment in the description here: #19025

[FrankChen021]

about the Modernization, there're two things I always think that we should do:

1. SQL centered. We should prioritize the SQL to be the first class citizen in Druid, we can use SQL to create task, kill task, and do anything, instead of using APIs.
2. Druid itself should be able to use its own engine to keep/query its own metrics, instead of exporting metrics to a 3rd system. This can be done by providing different system tables under the sys schema. By doing so, it can reduce the deployment complexity, cost and make the system more observable