feat(datafusion): DataFusion metrics query layer

[alexanderbianchi]

Why this PR exists

Quickwit is growing a second query/storage shape alongside Tantivy:

• logs and traces: document-oriented workloads on top of Tantivy
• metrics: columnar, time-series workloads on top of Parquet

#6231 made Parquet metrics a real storage path. #6285 isolated that pipeline so it can evolve without bleeding across the rest of Quickwit. This PR adds the query-side half of that story: a query layer built on Apache DataFusion.

The immediate payoff: Quickwit can execute SQL and Substrait queries over OSS Parquet metrics.

The larger payoff is architectural: Quickwit gets a reusable query substrate with a real optimizer, Arrow-native execution, Substrait interop, and distributed planning hooks.

That bigger direction is deliberate. The same extension points introduced here are also the shape we want for future SQL-over-logs work, as explored in:

• alexanderbianchi/tantivy-datafusion
• alexanderbianchi/quickwit#1

In other words: this PR is not just “metrics can be queried with DataFusion.” It is the first in-tree implementation of a more general idea:

Quickwit-specific storage engines stay specialized, while query planning and execution becomes shared.

Why DataFusion

Apache DataFusion is an extensible Rust query engine built on Arrow. It already provides the parts we do not want to reinvent:

• SQL planning
• logical and physical optimization
• vectorized, streaming execution
• Arrow-native interoperability
• extensibility around catalogs, table providers, UDFs, optimizer rules, codecs, and Substrait

That makes it a strong fit for Quickwit.

Quickwit still owns the parts that are actually Quickwit-shaped:

• index discovery and split enumeration
• storage resolution
• storage-engine-specific pushdown
• worker selection and distributed execution across searchers

The boundary is intentional: DataFusion handles relational query execution; Quickwit handles how that execution maps onto Quickwit storage and cluster topology.

Core APIs introduced

QuickwitDataSource

This is the main extension point for plugging a Quickwit-native storage engine into DataFusion.

A source can contribute:

• additive query/runtime state via contributions()
• default table resolution via create_default_table_provider()
• schema enumeration via list_index_names()
• worker-side setup via register_for_worker()
• optional Substrait ReadRel handling via try_consume_read_rel()

The important design point is that storage-specific logic stays behind this trait. The generic session, catalog, and worker code does not need to know whether it is serving Parquet metrics, Tantivy logs, or something else later.

DataSourceContributions

This is the additive container for source-provided DataFusion state.

Right now that includes:

• optimizer rules
• UDFs
• codec appliers

The builder merges contributions from all registered sources and checks invariants up front so two sources cannot silently stomp on each other.

QuickwitSchemaProvider

This is the schema bridge between DataFusion and Quickwit.

Resolution order is:

1. session-local DDL tables
2. source-backed Quickwit indexes

That gives us normal DataFusion ergonomics for explicit DDL while still allowing Quickwit-native indexes to appear as tables without pre-registering each one manually.

DataFusionSessionBuilder

This is the coordinator-side assembly point.

It owns the shared runtime state and builds DataFusion sessions with:

• merged source contributions
• Quickwit schema/catalog wiring
• distributed planning hooks when a worker resolver is present
• shared object store/runtime state reused across sessions

QuickwitWorkerSessionBuilder and build_quickwit_worker

These are the worker-side mirrors of the coordinator builder.

They rebuild the same source/runtime shape on worker nodes so distributed physical plans can execute without inventing a separate query runtime just for workers.

DataFusionService

This is the execution entry point.

It exposes:

• execute_sql
• execute_substrait
• execute_substrait_json

and streams results as Arrow RecordBatches. In OSS, this is wired over gRPC as Arrow IPC batches, which is a good low-level transport for both internal execution and future integrations.

QuickwitSubstraitConsumer

This is the glue for Substrait plans that need Quickwit-aware table resolution.

A source can claim a ReadRel, materialize a provider, and still fall back onto normal DataFusion planning behavior for filter and projection handling.

What is wired up in this PR today

This PR wires the first in-tree source: MetricsDataSource.

That includes:

• MetricsDataSource implementing QuickwitDataSource
• MetricsTableProvider backed by ParquetSource
• metric index resolution through the metastore and storage resolver
• DDL support via CREATE EXTERNAL TABLE ... STORED AS metrics
• SQL execution
• Substrait execution
• distributed execution across workers and searchers
• gRPC streaming of Arrow IPC record batches

On the Parquet side, the query path is not a toy wrapper. It is already doing real storage-aware work, including:

• filter pushdown
• bloom filter and page-index aware pruning
• split-aware planning
• metrics-specific predicate extraction for time-range pruning

Example shape

This is the sort of flow this PR enables for metrics today:

CREATE OR REPLACE EXTERNAL TABLE "cpu-prod" (
  metric_name VARCHAR NOT NULL,
  metric_type TINYINT,
  timestamp_secs BIGINT NOT NULL,
  value DOUBLE NOT NULL,
  service VARCHAR
) STORED AS metrics LOCATION 'cpu-prod';

SELECT service, AVG(value) AS avg_cpu
FROM "cpu-prod"
WHERE metric_name = 'cpu.usage'
  AND timestamp_secs BETWEEN 1710000000 AND 1710003600
GROUP BY service
ORDER BY service;

Scope of this PR

This PR is intentionally the first slice, not the final shape.

Current scope:

• query layer only
• metrics is the first in-tree source
• disabled by default unless a datafusion_session_builder is configured
• gRPC serving path for Arrow IPC batch streaming
• no claim that all Quickwit queries should immediately move to DataFusion

The point is to land the reusable query substrate and prove it on the storage path that is naturally columnar first: Parquet metrics.

Test coverage

This PR includes coverage for the important end-to-end query behaviors already implemented, including:

• SQL pruning
• aggregation
• time-range filtering
• GROUP BY
• distributed task planning without shuffles
• NULL fill for missing Parquet columns
• Substrait named-table queries
• rollup from JSON Substrait plan

Some Quirks

• metrics source ownership is name-based for now
• DDL registration is session-local / per-query, not persistent
  catalog state
• the endpoint is experimental and opt-in only

[alexanderbianchi]

@mattmkim left some comments on this draft PR that I closed #6270 , addressed in 07f8548

[mattmkim]

disclaimer only really reviewed the metrics specific logic, not a DF expert. made comments here,

what would give us more confidence in merging this? should we feature flag the datafusion code/path, so you have to "opt-in" to compiling an image with it? we could also add an env var to actually enable the endpoint?

[alexanderbianchi]

@mattmkim the session is behind a env var already FYI. Open to other rec's for isolating.

[mattmkim]

ah, missed that 👍

[alexanderbianchi]

@guilload would be good to have a quickwit expert review this part

[evance-br]

Hi @alexanderbianchi, Please can you give more context on this effort. Are you trying to query parquet files in DataFusion via Quickwit? How is this supposed to work for a user? How Quickwit end up with the parquet files being queried.