FrankChen021 opened a new pull request, #20183:
URL: https://github.com/apache/druid/pull/20183
### Description
This PR adds opt-in native-query execution for supported `sys` tables while
retaining the existing Bindable implementation for compatibility.
Native system-table planning is selected only when:
- The resolved SQL engine is `native`.
- The query context contains `"useNativeQueryForSystemTables": true`.
- The system table implements `NativeSystemTable`.
The context parameter defaults to `false`. Unsupported system tables
continue to use their existing Bindable path, and other SQL engines are
unaffected.
### Background and motivation
Druid system tables traditionally use Calcite's Bindable execution path. A
system table exposes rows through `ScannableTable` or
`ProjectableFilterableTable`, and Calcite evaluates the SQL query on the Broker.
This path provides basic SQL access to system metadata, but system tables
are not represented as native Druid datasources. As a result, many expressions,
aggregations, and operators implemented by Druid's native query engine are
unavailable. A simple example is:
```sql
SELECT COUNT(DISTINCT task_id)
FROM sys.tasks
```
The traditional Bindable path cannot translate this aggregation into Druid's
native distinct-count implementation. Similar limitations apply to other
Druid-specific aggregations, expressions, native filters, sorting strategies,
and query operators.
System tables can also retrieve substantially more data than the final query
needs. Although some traditional table implementations perform limited
filtering, they cannot generally reuse the native query filter model or push
supported predicates through component boundaries into the underlying storage
system.
This PR allows supported system tables to advertise a native representation.
When explicitly enabled, Calcite plans the table as a `SystemTableDataSource`,
component-owned rows are retrieved through native Scan queries, and the Broker
executes the original query using the normal native query engine.
The initial native-capable tables are:
| Table | Component providing rows | Storage pushdown |
| --- | --- | --- |
| `sys.tasks` | Overlord | Supported task filters are pushed into metadata
task storage when the configured `TaskStorage` supports them. |
| `sys.server_properties` | All discovered Druid server components |
`server` and `service_name` filters can avoid reading properties from
components that don't match. |
### Execution path
```text
Client
|
| POST /druid/v2/sql
v
Router AsyncQueryForwardingServlet
|
| SQL requests follow normal routing; the Router forwards to a Broker.
v
Broker SQL planner
|
| Native engine
| useNativeQueryForSystemTables = true
| table implements NativeSystemTable
v
Native query over SystemTableDataSource
|
v
QueryLifecycle -> SystemTableBrokerQueryHandler -> SystemTableQueryClient
|
| Use SystemTableDescriptor to discover contributing components
| Build a scan-only query for each component
|
+-- Selected component is this Broker
| |
| | Invoke the raw SystemTableQueryHandler in-process
| | No HTTP request and no recursive Broker fanout
| v
| Component-local Scan
|
`-- Selected component is remote
|
| POST /druid/v2
| X-Druid-Native-Query-Route: local
| Escalated Druid-to-Druid credentials
v
Receiving component
|
| Router: dispatch to its local QueryResource instead of forwarding
| Broker: SystemTableBrokerQueryHandler executes locally instead of
fanning out
| Other components: their QueryResource executes locally as usual
v
Component-local Scan
|
v
SystemTableQueryHandler
|
| Apply the descriptor's authorization rules for the authenticated identity
| Extract provider-supported pushdown filters
| Call SystemTableDataProvider.getRows(...)
| Wrap supplied rows in an InlineDataSource
| Execute the Scan query
v
Stream of ScanResultValue rows
|
v
Broker
|
| Apply the original user's row authorization
| Wrap authorized rows in an InlineDataSource
| Execute the original native query
v
SQL result
```
The route header is a public routing instruction, not an authentication
credential or proof that the request came
from a Broker. A request carrying `X-Druid-Native-Query-Route: local` still
passes through the receiving process's
normal authenticator chain and system-table authorization. This also permits
an authenticated client to send a
native system-table Scan directly to a component that serves the table.
For remote fanout, the Broker uses the same escalated HTTP client mechanism
as ordinary Broker-to-Historical native
queries. When discovery selects the Broker itself, the Broker calls the raw
local handler with the equivalent
in-process escalated authentication result. This avoids an HTTP loopback
while preserving the authorization behavior
of remote internal requests. The raw handler is used deliberately so the
request cannot re-enter Broker fanout.
The Router only changes routing for native queries that contain a
`SystemTableDataSource` and explicitly request
the `local` route. Other `/druid/v2` requests continue through normal Broker
forwarding. This prevents recursive
forwarding while allowing the Router to contribute its own rows to
`sys.server_properties`.
The implementation uses the standard `/druid/v2` endpoint and standard
`ScanResultValue` response format. It does
not introduce a system-table-specific HTTP endpoint or response protocol.
### COUPLED and DECOUPLED planner support
The native system-table planning path supports both SQL planner strategies.
In COUPLED mode, `DruidTableScanRule` normally sees the original Calcite
`TableScan` and replaces a native-capable system table with its
`DruidTableScan` representation.
In DECOUPLED mode, Calcite may first convert a `ProjectableFilterableTable`
into a `Bindables.BindableTableScan`. This is particularly relevant to
`sys.server_properties`, whose traditional implementation supports filter and
projection pushdown. The resulting Bindable scan can contain its filters and
projected column indexes inside the scan node before Druid's native logical
conversion runs.
`DruidBindableTableScanRule` handles this DECOUPLED-only plan shape after
`QueryHandler` has selected native system-table planning. It converts:
```text
BindableTableScan(filters, projects)
-> DruidProject
`- DruidFilter
`- DruidTableScan
```
The filter or project node is omitted when it is not needed. Reconstructing
these logical nodes preserves the filters and projections that Calcite embedded
in the Bindable scan while allowing the table itself to become a native
`SystemTableDataSource`.
The rule applies only after the native path has been selected and only when
the table advertises a native representation. Queries that use the traditional
Bindable path are unchanged. Dedicated tests cover native
`sys.server_properties` planning in both COUPLED and DECOUPLED modes, including
embedded filter and projection conversion.
### Adding another native system table
A new table uses the generic system-table infrastructure and does not
require a new HTTP resource, RPC client, or Broker query branch.
1. Define a `SystemTableDescriptor` containing:
- The table name.
- The node roles that contribute rows.
- The shared `RowSignature`.
- The Broker-side row authorizer for the original user.
2. Register the descriptor in the `MapBinder<String, SystemTableDescriptor>`
installed by `SystemTableModule`. The descriptor must be available to the
Broker for discovery, authorization, and row decoding, as well as to components
serving the table.
3. Implement `SystemTableDataProvider`:
```java
Iterable<Object[]> getRows(
List<DimFilter> pushdownFilters,
AuthenticationResult requestAuthenticationResult
);
```
The provider may override `getPushdownFilters()` to advertise columns
that support safe provider or storage pushdown.
4. Register the provider by table name on the owning component:
```java
dataProviderBinder.addBinding(TABLE_NAME)
.to(NewSystemTableDataProvider.class)
.in(LazySingleton.class);
```
The provider should be installed only on components represented by the
descriptor's node roles. For example, `SystemTasksTableDataProvider` is bound
by `CliOverlord`, while `ServerPropertiesTableDataProvider` is installed on all
components by `SystemTableModule`.
5. Expose a native SQL representation. The Calcite system-table definition
implements `NativeSystemTable` and returns its native `DruidTable` from
`asNativeTable()`. The native table uses `SystemTableDataSource` and the
descriptor's shared row signature.
The `DataSourceQueryHandler` registration then routes the new datasource
through the existing Broker fanout and component-local execution path.
### Filter pushdown
System-table providers advertise which columns they can push down and, when
necessary, map the system-table column name to its storage column:
```java
new SystemTablePushdownFilter("task_id", "id")
```
The following example shows how a filter moves from SQL to task metadata
storage:
```sql
SELECT task_id, datasource
FROM sys.tasks
WHERE task_id = 'task-1'
AND datasource = 'wikipedia'
AND status = 'SUCCESS'
```
```text
SQL WHERE clause
|
v
Calcite translates SQL expressions into native DimFilter objects
|
| AND
|-- EqualityFilter(column = "task_id", value = "task-1")
|-- EqualityFilter(column = "datasource", value = "wikipedia")
`-- EqualityFilter(column = "status", value = "SUCCESS")
v
Broker SystemTableQueryClient
|
| Copies the native filters into the component Scan query
v
Overlord SystemTableQueryHandler
|
| SystemTablePushdownFilter.extract(...)
|-- checks that each filter uses a provider-advertised column
|-- checks that the filter expression is supported for pushdown
`-- rewrites provider column names
|
| task_id -> id
| datasource -> datasource
| status -> status
v
SystemTasksTableDataProvider
|
| Wraps the extracted List<DimFilter>
v
TaskStorageQueryFilter
|
| Filters:
|-- EqualityFilter(column = "id", value = "task-1")
|-- EqualityFilter(column = "datasource", value = "wikipedia")
`-- EqualityFilter(column = "status", value = "SUCCESS")
|
| status = SUCCESS
|-- excludes the ACTIVE task lookup
`-- includes the COMPLETE task lookup
v
TaskQueryTool
|
v
MetadataTaskStorage
|
| Status is represented by ACTIVE and COMPLETE task lookups,
| not by a physical metadata column.
|
| metadataStorageFilter = filter.withoutStatusFilters()
|
| Remaining filters:
|-- EqualityFilter(column = "id", value = "task-1")
`-- EqualityFilter(column = "datasource", value = "wikipedia")
v
TaskStorageSqlPredicateBuilder
|
| Generated SQL predicate:
| AND id = :task_filter_0_0
| AND datasource = :task_filter_1_0
|
| Generated bindings:
| task_filter_0_0 = "task-1"
| task_filter_1_0 = "wikipedia"
v
SELECT ...
FROM druid_tasks
WHERE active = FALSE
AND id = :task_filter_0_0
AND datasource = :task_filter_1_0
...
```
The metadata query returns only candidate rows. The original native filter
remains on the Broker query and is evaluated again after rows return. Pushdown
is therefore an optimization; the Broker remains responsible for final filter
correctness.
Pushdown support is provider- and storage-dependent:
- `SystemTableDataProvider.getPushdownFilters()` declares available columns
and mappings.
- `SystemTablePushdownFilter.extract()` selects supported native `DimFilter`
expressions.
- `MetadataTaskStorage` supports task metadata SQL pushdown.
- Storage implementations that do not support metadata filtering advertise
only filters they can safely handle.
- Filters that cannot be pushed down remain Broker-side residual filters.
### Native query engine module refactoring
This PR replaces the former peon-oriented native-query module arrangement
with `NativeQueryEngineModule`, a shared facade installed by Druid server
components.
`NativeQueryEngineModule` installs:
- `QueryableModule`.
- `QueryRunnerFactoryModule`.
- `SegmentWranglerModule`.
- `JoinableFactoryModule`.
- `SystemTableModule`.
- A native query HTTP resource module.
Its builder supports role-specific profiles:
- `scanOnly()` installs only the Scan query factory and toolchest, plus a
default `NoopQuerySegmentWalker`. This lets management components serve
component-local system-table scans without requiring merge buffers or the
complete native aggregation stack.
- `withOverrideModule(...)` lets a role replace execution bindings. The
Coordinator uses this for `SegmentSchemaCacheModule`.
- `withQueryResourceModule(...)` replaces the standard
`QueryResourceModule`. The Broker installs `BrokerQueryResourceModule`, and the
Router installs `RouterQueryResourceModule`.
Full native-query components use the normal query-runner factories.
Scan-only management components reuse `QueryRunnerFactoryModule` with a
restricted query-type set instead of maintaining a separate scan execution
module.
This gives Broker, Router, Coordinator, Overlord, Historical, Indexer,
MiddleManager, and Peon processes a consistent native-query module facade while
allowing each role to customize its execution and HTTP-resource bindings.
### Authorization and compatibility
Remote component requests use Druid's existing escalated HTTP client,
matching ordinary internal native-query
traffic. The receiving component authenticates the request through its
normal authenticator chain. The
`X-Druid-Native-Query-Route` header only selects local routing and does not
bypass authentication or authorization.
No system-table-specific authenticator name or trusted-header configuration
is required.
When the Broker is itself a selected component, it uses the escalator's
in-process `AuthenticationResult` and calls
the local system-table handler directly. This is semantically equivalent to
the remote internal request, without
serializing the query through HTTP.
The Broker retains the original request's `AuthenticationResult` and applies
the table descriptor's row authorizer
to the combined component rows before residual filtering, expressions,
sorting, or aggregation. A client that sends
a local native system-table Scan directly to a component is authorized using
that client's authenticated identity.
`useNativeQueryForSystemTables` defaults to `false` to support rolling
upgrades. Operators should leave it disabled
until the Broker and components serving native system tables have been
upgraded. It can then be enabled per query or
globally on Brokers with:
```properties
druid.query.default.context.useNativeQueryForSystemTables=true
```
### Current scope and limitations
#### Component execution is scan-only
The Broker always generates a Scan query when requesting rows from a
component. Components do not currently execute the original GroupBy,
Timeseries, TopN, or window query.
For example:
```sql
SELECT datasource, COUNT(*)
FROM sys.tasks
GROUP BY datasource
```
currently executes as:
```text
Broker
|
| Scan SystemTableDataSource("tasks")
v
Overlord
|
| Read and prefilter task rows
| Execute component-local Scan
v
Broker
|
| Execute the original GroupBy
v
Result
```
The GroupBy is deliberately executed on the Broker rather than pushed into
the Overlord. Native GroupBy execution requires merge buffers and the
associated processing configuration. Components such as Coordinator, Overlord,
Router, and MiddleManager did not previously need the complete native
aggregation stack. Requiring every component to configure and reserve
merge-buffer memory solely for system-table queries would introduce significant
configuration, memory-management, and operational complexity.
`NativeQueryEngineModule.scanOnly()` therefore installs only the
infrastructure required for native Scan execution, with no GroupBy merge-buffer
requirement. The Broker already has the complete native query infrastructure
and merge buffers, so it executes the original aggregation after receiving
component rows.
Supporting component-side GroupBy in the future requires more than enabling
its query factory. The distributed result contract must preserve intermediate
aggregation state so the Broker can correctly merge partial results from
multiple components, especially for `COUNT(DISTINCT ...)`, sketches,
non-additive aggregations, post-aggregators, and group ordering or limits.
The current scan-only transport keeps that future change isolated behind
`SystemTableQueryClient`: it can later accept another component result type or
frame-based exchange without changing table registration, discovery,
authorization, or SQL planning.
#### Other limitations
- Component cancellation stops Broker result consumption and closes the
component HTTP request, but does not guarantee interruption of provider or
metadata-storage work that has already begun.
- Component-local system-table scans invoke the scan engine directly and are
not governed by the normal `QueryScheduler` lane and capacity limits.
- Aggregations, expressions, residual filters, sorting, limits, and window
processing execute on the Broker.
- `LIMIT`, `OFFSET`, and column projection are not pushed into task metadata
storage.
- Component results use row-oriented `ScanResultValue` transport rather than
frame-based exchange.
- Most component results stream lazily, but window queries currently require
Broker-side materialization.
- Native support currently covers only `sys.tasks` and
`sys.server_properties`.
- Tables without a native representation retain their existing Bindable
execution path.
- Filter pushdown is provider-dependent. Unsupported filters remain on the
original Broker query and are evaluated there for correctness.
#### Release note
SQL queries against supported system tables can now opt into Druid's native
query engine with the `useNativeQueryForSystemTables` query context parameter.
The initial implementation supports `sys.tasks` and `sys.server_properties`,
uses the standard `/druid/v2` endpoint for component fanout, supports safe
storage filter pushdown, and retains the existing execution path by default for
rolling-upgrade compatibility.
### Validation
Coverage includes:
- Native planning in both COUPLED and DECOUPLED planner strategies.
- Bindable fallback when native system-table execution is disabled or
unsupported.
- Native aggregations such as `COUNT(DISTINCT ...)`.
- The web-console tasks query with computed status and non-time ordering.
- Component-local Scan execution and Broker-side result processing.
- Filter extraction and column rewriting.
- Derby-backed task metadata predicates for equality, `IN`, range, `LIKE`,
negation, status, and nullable migration columns.
- Provider registration and role-based discovery.
- Router header routing, Broker in-process self execution, and
component-local execution.
- Authorization behavior.
- Embedded end-to-end tests for `sys.tasks` and `sys.server_properties`.
This PR has:
- [x] been self-reviewed.
- [x] added documentation for the new query context and native system-table
behavior.
- [x] included a release note in the PR description.
- [x] added Javadocs for the principal new classes and interfaces.
- [x] added unit and embedded end-to-end tests.
- [x] been tested in a local Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
