You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "FrankChen021 (via GitHub)" <gi...@apache.org> on 2026/08/28 15:28:16 UTC
[PR] feat(sql): add native query support for system tables (druid)
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4314863293
##########
server/src/main/java/org/apache/druid/server/QueryLifecycle.java:
##########
@@ -420,13 +465,58 @@ public <T> QueryResponse<T> execute()
final ResponseContext responseContext = DirectDruidClient.makeResponseContextForQuery();
@SuppressWarnings("unchecked")
- final Sequence<T> res = QueryPlus.wrap((Query<T>) baseQuery)
- .withIdentity(authenticationResult.getIdentity())
- .run(texasRanger, responseContext);
+ final Query<T> originalQuery = (Query<T>) baseQuery;
+ final DataSourceQueryHandler dataSourceQueryHandler = findDataSourceQueryHandler(originalQuery.getDataSource());
+ // Handlers can start fanout or provider work before reaching a segment walker, or bypass it entirely.
+ // Apply the same server guardrails first, and pass the normalized query to both runner creation and execution.
+ final Query<T> query = dataSourceQueryHandler == null
+ ? originalQuery
+ : SetAndVerifyContextQueryRunner.withTimeoutAndMaxScatterGatherBytes(
+ originalQuery,
+ serverConfig,
+ System.currentTimeMillis()
+ );
+ final QueryRunner<T> queryRunner = dataSourceQueryHandler == null
Review Comment:
Leave this in a follow up PR, reaons: the case that cancels a query on system table is not common
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4230034700
##########
server/src/main/java/org/apache/druid/server/system/table/SystemTablePushdownFilter.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.table;
+
+import org.apache.druid.query.Query;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.BoundDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.EqualityFilter;
+import org.apache.druid.query.filter.InDimFilter;
+import org.apache.druid.query.filter.LikeDimFilter;
+import org.apache.druid.query.filter.NotDimFilter;
+import org.apache.druid.query.filter.OrDimFilter;
+import org.apache.druid.query.filter.RangeFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.query.filter.TypedInFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.ordering.StringComparators;
+import org.apache.druid.segment.column.ColumnType;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Declares a pushdown-capable system-table column and its optional supplier-side name.
+ */
+public record SystemTablePushdownFilter(String key, @Nullable String value)
+{
+ public static List<DimFilter> extract(
+ final Query<?> query,
+ final List<SystemTablePushdownFilter> pushdownFilters
+ )
+ {
+ if (pushdownFilters.isEmpty()) {
+ return Collections.emptyList();
+ }
+ final Map<String, String> columnMappings = pushdownFilters.stream().collect(
+ Collectors.toMap(
+ SystemTablePushdownFilter::key,
+ filter -> filter.value() == null ? filter.key() : filter.value()
+ )
+ );
+
+ final List<DimFilter> extracted = new ArrayList<>();
+ extractConjuncts(query.getFilter(), columnMappings, extracted);
Review Comment:
[P2] Exclude virtual columns from provider filter pushdown
**Finding:** The extraction checks only the column name and ignores the query's virtual columns. A valid local native Scan of `server_properties` can define a constant STRING virtual column named `server` with value `alias` and filter `server = alias`. Druid's virtualized selectors give that column precedence over the physical `server`, so all rows should match, but this method pushes the selector to `ServerPropertiesTableDataProvider`, which compares `alias` to the actual host:port and returns no rows before the native filter runs. The same issue affects `service_name` and Broker fanout when it forwards these virtual columns. This silently changes query results.
**Suggestion:** Exclude columns shadowed by query or leaf-scan virtual columns from supplier-side pushdown, and add a regression case for a native Scan that shadows a pushdown column.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4359631207
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryClient.java:
##########
@@ -0,0 +1,917 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import io.netty.channel.ChannelException;
+import org.apache.druid.client.DirectDruidClientFactory;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.LazySequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.Yielders;
+import org.apache.druid.query.BaseQuery;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.QueryDataSource;
+import org.apache.druid.query.QueryException;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.QueryContextParameters;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.QueryScheduler;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.system.SystemTableNotLeaderException;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.utils.CloseableUtils;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+/**
+ * Recursively replaces native system-table datasources with authorized inline rows fetched from nodes, then
+ * executes the rebuilt query through the Broker's normal native-query walker.
+ */
+public class SystemTableQueryClient implements DataSourceQueryHandler
+{
+ private static final String SYSTEM_TABLE_TIER = "_system";
+
+ private final SystemTableNodeLocator nodeLocator;
+ private final DirectDruidClientFactory directDruidClientFactory;
+ private final QueryScheduler queryScheduler;
+ private final QuerySegmentWalker querySegmentWalker;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final AuthorizerMapper authorizerMapper;
+ private final SystemTableQueryHandler localQueryHandler;
+ private final AuthenticationResult escalatedAuthenticationResult;
+ private final DruidNode selfNode;
+
+ @Inject
+ public SystemTableQueryClient(
+ final SystemTableNodeLocator nodeLocator,
+ final DirectDruidClientFactory directDruidClientFactory,
+ final QueryScheduler queryScheduler,
+ final QuerySegmentWalker querySegmentWalker,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final AuthorizerMapper authorizerMapper,
+ final SystemTableQueryHandler localQueryHandler,
+ final Escalator escalator,
+ @Self final DruidNode selfNode
+ )
+ {
+ this.nodeLocator = nodeLocator;
+ this.directDruidClientFactory = directDruidClientFactory;
+ this.queryScheduler = queryScheduler;
+ this.querySegmentWalker = querySegmentWalker;
+ this.tableDescriptors = tableDescriptors;
+ this.authorizerMapper = authorizerMapper;
+ this.localQueryHandler = localQueryHandler;
+ this.escalatedAuthenticationResult = escalator.createEscalatedAuthenticationResult();
+ this.selfNode = selfNode;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult authenticationResult,
+ final boolean executeLocally
+ )
+ {
+ return (queryPlus, responseContext) -> {
+ final NodeSequencesCloser nodeSequencesCloser = new NodeSequencesCloser(queryScheduler);
+ try {
+ final DataSource resolvedDataSource = resolveDataSource(
+ query.getDataSource(),
+ query,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ final Query<T> resolvedQuery = query.withDataSource(resolvedDataSource).withOverriddenContext(
+ Map.of(QueryContexts.QUERY_RESOURCE_ID, UUID.randomUUID().toString())
+ );
+ final QueryRunner<T> queryRunner = querySegmentWalker.getQueryRunnerForIntervals(
+ resolvedQuery,
+ resolvedQuery.getIntervals()
+ );
+ return Sequences.withBaggage(
+ queryRunner.run(queryPlus.withQuery(resolvedQuery), responseContext),
+ nodeSequencesCloser
+ );
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, nodeSequencesCloser);
+ }
+ };
+ }
+
+ private DataSource resolveDataSource(
+ final DataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ if (dataSource instanceof SystemTableDataSource systemTableDataSource) {
+ return resolveSystemTableDataSource(
+ systemTableDataSource,
+ owningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ }
+
+ final Query<?> childOwningQuery;
+ final DataSource resolvedParent;
+ if (dataSource instanceof QueryDataSource queryDataSource) {
+ final Query<?> childQuery = queryDataSource.getQuery();
+ // An inner query may tighten the outer guardrails, but must not discard them before node fanout.
+ childOwningQuery = childQuery.withOverriddenContext(
+ Map.of(
+ QueryContextParameters.QUERY_FAIL_TIME.getName(),
+ Math.min(getNodeFailTime(owningQuery), getNodeFailTime(childQuery)),
+ QueryContextParameters.MAX_SCATTER_GATHER_BYTES.getName(),
+ Math.min(owningQuery.context().getMaxScatterGatherBytes(), childQuery.context().getMaxScatterGatherBytes())
+ )
+ );
+ resolvedParent = new QueryDataSource(childOwningQuery);
+ } else {
+ childOwningQuery = owningQuery;
+ resolvedParent = dataSource;
+ }
+ final List<DataSource> resolvedChildren = new ArrayList<>();
+ for (final DataSource child : dataSource.getChildren()) {
+ resolvedChildren.add(
+ resolveDataSource(
+ child,
+ childOwningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ )
+ );
+ }
+ return resolvedParent.withChildren(resolvedChildren);
+ }
+
+ private InlineDataSource resolveSystemTableDataSource(
+ final SystemTableDataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final SystemTableDescriptor descriptor = tableDescriptors.get(dataSource.getTable());
+ if (descriptor == null) {
+ throw new ISE("No routing descriptor is registered for system table[%s]", dataSource.getTable());
+ }
+
+ final ScanQuery nodeQuery = makeNodeQuery(dataSource, descriptor, owningQuery);
+ final List<QueryRunner<ScanResultValue>> nodeRunners = makeNodeRunners(
+ nodeQuery,
+ descriptor,
+ nodeSequencesCloser
+ );
+ if (nodeRunners.isEmpty() && !descriptor.isEmptyDiscoveryAllowed()) {
+ throw new ISE("No node is available to serve system table[%s]", dataSource.getTable());
+ }
+
+ /*
+ * Keep the scan result transport lazy. The node response is a Sequence, so this lets the local query
+ * runner consume and process rows while the HTTP response is still arriving instead of materializing every
+ * node row in two Broker-side lists. The scan transport is intentionally kept behind this adapter so a
+ * future node-side aggregation mode can use a different result adapter without changing routing or
+ * authorization orchestration here.
+ */
+ final NodeSequenceCloser nodeSequenceCloser = new NodeSequenceCloser();
+ nodeSequencesCloser.add(nodeSequenceCloser);
+ final Iterable<Object[]> nodeRows = scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ responseContext,
+ nodeSequenceCloser
+ );
+ final Iterable<Object[]> authorizedRows = descriptor.getRowAuthorizer().filterAuthorizedRows(
+ nodeRows,
+ authenticationResult,
+ authorizerMapper
+ );
+ final Iterable<Object[]> queryRows = requiresMaterializedRows(owningQuery)
+ ? materializeRows(authorizedRows)
+ : authorizedRows;
+ return InlineDataSource.fromIterable(queryRows, descriptor.getRowSignature());
+ }
+
+ private ScanQuery makeNodeQuery(
+ final SystemTableDataSource dataSource,
+ final SystemTableDescriptor descriptor,
+ final Query<?> owningQuery
+ )
+ {
+ final Map<String, Object> nodeContext = new LinkedHashMap<>(owningQuery.getContext());
+ final DimFilter nodeFilter = nodeFilter(dataSource, owningQuery);
+ nodeContext.put(QueryContextParameters.QUERY_FAIL_TIME.getName(), getNodeFailTime(owningQuery));
+ // The node endpoint always returns plain ScanResultValue objects. bySegment belongs to the user-facing query and
+ // would make DirectDruidClient deserialize this private transport using an incompatible result type.
+ nodeContext.put(QueryContexts.BY_SEGMENT_KEY, false);
+ return Druids.newScanQueryBuilder()
+ .dataSource(dataSource)
+ .eternityInterval()
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_COMPACTED_LIST)
+ .limit(Long.MAX_VALUE)
+ .filters(nodeFilter)
+ .virtualColumns(nodeVirtualColumns(dataSource, owningQuery, nodeFilter))
+ .columns(descriptor.getRowSignature())
Review Comment:
correct, there should be a PROJECTION push down, which is not included now
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-5597929654
> @clintropolis @gianm please help review the changes. I hope this can be merged as soon as possible so that we can move on migrating other system tables into native execution path, and push forward for #18087 and #19855
Very cool. I will try to have a look soon.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4361099644
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
Review Comment:
Reworded to "the Broker's built-in SQL interpreter".
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT server) FROM sys.server_properties",
+ "context": {
+ "allowNativeFunctionsForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET allowNativeFunctionsForSystemTables = 'true';
+SELECT COUNT(DISTINCT server) FROM sys.server_properties;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `allowNativeFunctionsForSystemTables` is `true`.
+
+The parameter defaults to `false`. During a rolling upgrade, leave it disabled until the Broker and all nodes
+that serve the native system tables have been upgraded. After the upgrade, you can enable it by query or set
+`druid.query.default.context.allowNativeFunctionsForSystemTables=true` on Brokers as the cluster-wide default. For more
+information, see [SQL query context](sql-query-context.md).
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
Review Comment:
Removed.
##########
server/src/main/java/org/apache/druid/client/DirectDruidClient.java:
##########
@@ -785,7 +785,8 @@ private void checkTotalBytesLimit(long bytes)
.setHeader(
HttpHeaders.Names.CONTENT_TYPE,
isSmile ? SmileMediaTypes.APPLICATION_JACKSON_SMILE : MediaType.APPLICATION_JSON
- ),
+ )
+ .setHeader(QueryResource.HEADER_NATIVE_QUERY_ROUTE, QueryResource.NATIVE_QUERY_ROUTE_LOCAL),
Review Comment:
Now set only for `SystemTableDataSource` queries (POST and cancel DELETE).
##########
processing/src/main/java/org/apache/druid/query/SystemTableDataSource.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Identifies a Druid system table whose rows are supplied by a service rather than by segments.
+ */
+public class SystemTableDataSource extends LeafDataSource
Review Comment:
Overridden: returns itself, and throws if a policy is ever returned for a system table.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3989732312
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryHandler.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.inject.Inject;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryEngine;
+import org.apache.druid.segment.InlineSegmentWrangler;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.system.table.SystemTableDataProvider;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTablePushdownFilter;
+
+import java.util.Map;
+
+/** Resolves one node-local system table and returns its rows through the standard native Scan stack. */
+public class SystemTableQueryHandler implements DataSourceQueryHandler
+{
+ private final Map<String, SystemTableDataProvider> dataSuppliers;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final ScanQueryEngine scanQueryEngine;
+ private final AuthorizerMapper authorizerMapper;
+
+ @Inject
+ public SystemTableQueryHandler(
+ final Map<String, SystemTableDataProvider> dataSuppliers,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final ScanQueryEngine scanQueryEngine,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.dataSuppliers = dataSuppliers;
+ this.tableDescriptors = tableDescriptors;
+ this.scanQueryEngine = scanQueryEngine;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult requestAuthenticationResult,
+ final boolean executeLocally
+ )
+ {
+ if (!(query instanceof ScanQuery)) {
+ throw new IAE("Local system table queries must be scan queries");
+ }
+
+ final SystemTableDataSource dataSource = (SystemTableDataSource) query.getDataSource();
Review Comment:
Fixed in `922c82dcdd`.
Local execution now requires a root `SystemTableDataSource`; a composite query carrying the local-route header is rejected with `BadQueryContextException` instead of reaching the unsafe cast. `SystemTableQueryHandler` also validates the datasource shape defensively. Composite queries without the local-route header continue through `SystemTableQueryClient`, which recursively resolves system-table leaves.
Added regression coverage in `SystemTableBrokerQueryHandlerTest` and `SystemTableQueryHandlerTest`. The focused server run passed all 5 tests.
<!-- mergelens:review -->
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4237678941
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryClient.java:
##########
@@ -0,0 +1,899 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import io.netty.channel.ChannelException;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.client.DirectDruidClientFactory;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.LazySequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.Yielders;
+import org.apache.druid.query.BaseQuery;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.QueryDataSource;
+import org.apache.druid.query.QueryException;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.QueryScheduler;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.system.SystemTableNotLeaderException;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.utils.CloseableUtils;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+/**
+ * Recursively replaces native system-table datasources with authorized inline rows fetched from nodes, then
+ * executes the rebuilt query through the Broker's normal native-query walker.
+ */
+public class SystemTableQueryClient implements DataSourceQueryHandler
+{
+ private static final String SYSTEM_TABLE_TIER = "_system";
+
+ private final SystemTableNodeLocator nodeLocator;
+ private final DirectDruidClientFactory directDruidClientFactory;
+ private final QueryScheduler queryScheduler;
+ private final QuerySegmentWalker querySegmentWalker;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final AuthorizerMapper authorizerMapper;
+ private final SystemTableQueryHandler localQueryHandler;
+ private final AuthenticationResult escalatedAuthenticationResult;
+ private final DruidNode selfNode;
+
+ @Inject
+ public SystemTableQueryClient(
+ final SystemTableNodeLocator nodeLocator,
+ final DirectDruidClientFactory directDruidClientFactory,
+ final QueryScheduler queryScheduler,
+ final QuerySegmentWalker querySegmentWalker,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final AuthorizerMapper authorizerMapper,
+ final SystemTableQueryHandler localQueryHandler,
+ final Escalator escalator,
+ @Self final DruidNode selfNode
+ )
+ {
+ this.nodeLocator = nodeLocator;
+ this.directDruidClientFactory = directDruidClientFactory;
+ this.queryScheduler = queryScheduler;
+ this.querySegmentWalker = querySegmentWalker;
+ this.tableDescriptors = tableDescriptors;
+ this.authorizerMapper = authorizerMapper;
+ this.localQueryHandler = localQueryHandler;
+ this.escalatedAuthenticationResult = escalator.createEscalatedAuthenticationResult();
+ this.selfNode = selfNode;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult authenticationResult,
+ final boolean executeLocally
+ )
+ {
+ return (queryPlus, responseContext) -> {
+ final NodeSequencesCloser nodeSequencesCloser = new NodeSequencesCloser(queryScheduler);
+ try {
+ final DataSource resolvedDataSource = resolveDataSource(
+ query.getDataSource(),
+ query,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ final Query<T> resolvedQuery = query.withDataSource(resolvedDataSource).withOverriddenContext(
+ Map.of(QueryContexts.QUERY_RESOURCE_ID, UUID.randomUUID().toString())
+ );
+ final QueryRunner<T> queryRunner = querySegmentWalker.getQueryRunnerForIntervals(
+ resolvedQuery,
+ resolvedQuery.getIntervals()
+ );
+ return Sequences.withBaggage(
+ queryRunner.run(queryPlus.withQuery(resolvedQuery), responseContext),
+ nodeSequencesCloser
+ );
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, nodeSequencesCloser);
+ }
+ };
+ }
+
+ private DataSource resolveDataSource(
+ final DataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ if (dataSource instanceof SystemTableDataSource systemTableDataSource) {
+ return resolveSystemTableDataSource(
+ systemTableDataSource,
+ owningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ }
+
+ final Query<?> childOwningQuery = dataSource instanceof QueryDataSource queryDataSource
+ ? queryDataSource.getQuery()
+ : owningQuery;
+ final List<DataSource> resolvedChildren = new ArrayList<>();
+ for (final DataSource child : dataSource.getChildren()) {
+ resolvedChildren.add(
+ resolveDataSource(
+ child,
+ childOwningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ )
+ );
+ }
+ return dataSource.withChildren(resolvedChildren);
+ }
+
+ private InlineDataSource resolveSystemTableDataSource(
+ final SystemTableDataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final SystemTableDescriptor descriptor = tableDescriptors.get(dataSource.getTable());
+ if (descriptor == null) {
+ throw new ISE("No routing descriptor is registered for system table[%s]", dataSource.getTable());
+ }
+
+ final ScanQuery nodeQuery = makeNodeQuery(dataSource, descriptor, owningQuery);
+ final List<QueryRunner<ScanResultValue>> nodeRunners = makeNodeRunners(
+ nodeQuery,
+ descriptor,
+ nodeSequencesCloser
+ );
+ if (nodeRunners.isEmpty() && !descriptor.isEmptyDiscoveryAllowed()) {
+ throw new ISE("No node is available to serve system table[%s]", dataSource.getTable());
+ }
+
+ /*
+ * Keep the scan result transport lazy. The node response is a Sequence, so this lets the local query
+ * runner consume and process rows while the HTTP response is still arriving instead of materializing every
+ * node row in two Broker-side lists. The scan transport is intentionally kept behind this adapter so a
+ * future node-side aggregation mode can use a different result adapter without changing routing or
+ * authorization orchestration here.
+ */
+ final NodeSequenceCloser nodeSequenceCloser = new NodeSequenceCloser();
+ nodeSequencesCloser.add(nodeSequenceCloser);
+ final Iterable<Object[]> nodeRows = scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ responseContext,
+ nodeSequenceCloser
+ );
+ final Iterable<Object[]> authorizedRows = descriptor.getRowAuthorizer().filterAuthorizedRows(
+ nodeRows,
+ authenticationResult,
+ authorizerMapper
+ );
+ final Iterable<Object[]> queryRows = requiresMaterializedRows(owningQuery)
+ ? materializeRows(authorizedRows)
+ : authorizedRows;
+ return InlineDataSource.fromIterable(queryRows, descriptor.getRowSignature());
+ }
+
+ private ScanQuery makeNodeQuery(
+ final SystemTableDataSource dataSource,
+ final SystemTableDescriptor descriptor,
+ final Query<?> owningQuery
+ )
+ {
+ final Map<String, Object> nodeContext = new LinkedHashMap<>(owningQuery.getContext());
+ final DimFilter nodeFilter = nodeFilter(dataSource, owningQuery);
+ nodeContext.put(DirectDruidClient.QUERY_FAIL_TIME, getNodeFailTime(owningQuery));
+ // The node endpoint always returns plain ScanResultValue objects. bySegment belongs to the user-facing query and
+ // would make DirectDruidClient deserialize this private transport using an incompatible result type.
+ nodeContext.put(QueryContexts.BY_SEGMENT_KEY, false);
+ return Druids.newScanQueryBuilder()
+ .dataSource(dataSource)
+ .eternityInterval()
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_COMPACTED_LIST)
+ .limit(Long.MAX_VALUE)
+ .filters(nodeFilter)
+ .virtualColumns(nodeVirtualColumns(dataSource, owningQuery, nodeFilter))
+ .columns(descriptor.getRowSignature())
+ .context(nodeContext)
+ .build();
+ }
+
+ private List<QueryRunner<ScanResultValue>> makeNodeRunners(
+ final ScanQuery nodeQuery,
+ final SystemTableDescriptor descriptor,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final List<QueryRunner<ScanResultValue>> nodeRunners = new ArrayList<>();
+ for (final SystemTableNode node : nodeLocator.locate(descriptor, nodeQuery)) {
+ nodeRunners.add(
+ (queryPlus, responseContext) -> recoverNodeFailure(
+ () -> runNodeQuery(nodeQuery, node, nodeSequencesCloser, queryPlus, responseContext),
+ node,
+ descriptor,
+ descriptor.getRoutingMode() == SystemTableRoutingMode.LEADER_ONLY
+ ? () -> {
+ final SystemTableNode currentLeader = Iterables.getOnlyElement(
+ nodeLocator.locate(descriptor, nodeQuery)
+ );
+ return new NodeQuerySequence(
+ currentLeader,
+ runNodeQuery(nodeQuery, currentLeader, nodeSequencesCloser, queryPlus, responseContext)
+ );
+ }
+ : null
+ )
+ );
+ }
+ return nodeRunners;
+ }
+
+ private Sequence<ScanResultValue> runNodeQuery(
+ final ScanQuery nodeQuery,
+ final SystemTableNode node,
+ final NodeSequencesCloser nodeSequencesCloser,
+ final QueryPlus<ScanResultValue> queryPlus,
+ final ResponseContext responseContext
+ )
+ {
+ final String nodeResourceId = UUID.randomUUID().toString();
+ final String nodeQueryId = SystemTableDataSource.NODE_QUERY_ID_PREFIX + UUID.randomUUID();
+ final ScanQuery subNativeQuery = nodeQuery.withOverriddenContext(
+ Map.of(
+ BaseQuery.QUERY_ID,
+ nodeQueryId,
+ QueryContexts.QUERY_RESOURCE_ID,
+ nodeResourceId
+ )
+ );
+ final QueryRunner<ScanResultValue> nodeRunner;
+ if (SystemTableNodeLocator.sameServer(
+ selfNode.getUriToUse(),
+ node.getDiscoveryNode().getDruidNode().getUriToUse()
+ )) {
+ // Invoke the raw local handler rather than the Broker handler. This avoids an HTTP request back to this
+ // Broker and, more importantly, prevents the local request from entering node fanout recursively.
+ nodeRunner = localQueryHandler.createRunner(subNativeQuery, escalatedAuthenticationResult, true);
+ } else {
+ nodeRunner = directDruidClientFactory.makeDirectClient(toDruidServer(node.getDiscoveryNode()));
+ nodeSequencesCloser.addQueryId(nodeQueryId);
+ }
+ return nodeRunner.run(queryPlus.withQuery(subNativeQuery), responseContext);
+ }
+
+ /**
+ * Window operators consume a {@code RowsAndColumns} view of the segment. The current row-based segment can only
+ * provide that view when its rows are list-backed, so preserve the existing materialized path for those queries.
+ */
+ private static boolean requiresMaterializedRows(final Query<?> query)
+ {
+ return query instanceof WindowOperatorQuery;
+ }
+
+ /**
+ * Eagerly consumes the lazy, single-pass node response into an {@link ArrayList} without changing its rows.
+ *
+ * <p>This representation change is required by {@link org.apache.druid.segment.InlineSegmentWrangler}: an inline
+ * datasource backed specifically by an {@code ArrayList} becomes an {@link org.apache.druid.segment.ArrayListSegment},
+ * which can provide the {@code RowsAndColumns} view required by window operators. Other query types retain the lazy
+ * iterable so the Broker can process rows while node responses are still arriving.</p>
+ */
+ private static Iterable<Object[]> materializeRows(final Iterable<Object[]> rows)
+ {
+ final List<Object[]> materializedRows = new ArrayList<>();
+ rows.forEach(materializedRows::add);
+ return materializedRows;
+ }
+
+ /**
+ * Creates a lazy, single-pass view of the rows returned by the node scan runners.
+ *
+ * <p>The view is single-pass because a node response is backed by a live HTTP stream. The normal native query
+ * path consumes an {@link InlineDataSource} once per query. Re-iterating would otherwise issue duplicate node
+ * requests or retain all rows to support replay.</p>
+ */
+ static Iterable<Object[]> scanNodeRows(
+ final List<QueryRunner<ScanResultValue>> nodeRunners,
+ final Query<ScanResultValue> nodeQuery,
+ final ResponseContext responseContext
+ )
+ {
+ return scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ responseContext,
+ new NodeSequenceCloser()
+ );
+ }
+
+ private static Iterable<Object[]> scanNodeRows(
+ final List<QueryRunner<ScanResultValue>> nodeRunners,
+ final Query<ScanResultValue> nodeQuery,
+ final ResponseContext responseContext,
+ final NodeSequenceCloser nodeSequenceCloser
+ )
+ {
+ final Sequence<Object[]> rows = new LazySequence<>(() -> {
+ final List<Sequence<ScanResultValue>> nodeSequences = new ArrayList<>();
+ for (final QueryRunner<ScanResultValue> nodeRunner : nodeRunners) {
+ // DirectDruidClient.run starts its asynchronous HTTP request. Start every request before waiting for any
+ // response so slow nodes do not serialize the entire system-table query.
+ nodeSequences.add(nodeRunner.run(QueryPlus.wrap(nodeQuery), responseContext));
+ }
+ // Node scans have no cross-node ordering. Consume them in discovery order so a ready node can yield rows without
+ // MergeSequence first initializing every other node sequence. The HTTP requests were all started above.
+ return Sequences.concat(nodeSequences).flatMap(SystemTableQueryClient::rowsFromScanResult);
+ });
+ return sequenceAsIterable(rows, nodeSequenceCloser);
+ }
+
+ private static Sequence<ScanResultValue> recoverNodeFailure(
+ final Supplier<Sequence<ScanResultValue>> sequenceSupplier,
+ final SystemTableNode node,
+ final SystemTableDescriptor descriptor,
+ @Nullable final Supplier<NodeQuerySequence> leaderRetrySupplier
+ )
+ {
+ final Sequence<ScanResultValue> sequence;
+ try {
+ sequence = sequenceSupplier.get();
+ }
+ catch (Exception failure) {
+ if (leaderRetrySupplier != null && isLeaderChangeFailure(failure)) {
+ try {
+ final NodeQuerySequence retry = leaderRetrySupplier.get();
+ return recoveringSequence(retry, descriptor, null);
+ }
+ catch (Exception retryFailure) {
+ return failureSequence(retryFailure, node, descriptor);
+ }
+ }
+ return failureSequence(failure, node, descriptor);
+ }
+ return recoveringSequence(new NodeQuerySequence(node, sequence), descriptor, leaderRetrySupplier);
+ }
+
+ private static Sequence<ScanResultValue> recoveringSequence(
+ final NodeQuerySequence nodeQuerySequence,
+ final SystemTableDescriptor descriptor,
+ @Nullable final Supplier<NodeQuerySequence> leaderRetrySupplier
+ )
+ {
+ return new BaseSequence<>(
+ new BaseSequence.IteratorMaker<ScanResultValue, RecoveringNodeIterator>()
+ {
+ @Override
+ public RecoveringNodeIterator make()
+ {
+ return new RecoveringNodeIterator(nodeQuerySequence, descriptor, leaderRetrySupplier);
+ }
+
+ @Override
+ public void cleanup(final RecoveringNodeIterator iterator)
+ {
+ CloseableUtils.closeAndWrapExceptions(iterator);
+ }
+ }
+ );
+ }
+
+ private static Sequence<ScanResultValue> failureSequence(
+ final Exception failure,
+ final SystemTableNode node,
+ final SystemTableDescriptor descriptor
+ )
+ {
+ if (!isNodeAvailabilityFailure(failure)) {
+ throw propagate(failure);
+ }
+ return descriptor.getNodeFailureRow(
+ node.getDiscoveryNode().getDruidNode(),
+ node.getNodeRoles(),
+ failure
+ ).map(
+ row -> Sequences.simple(
+ List.of(
+ new ScanResultValue(
+ null,
+ descriptor.getRowSignature().getColumnNames(),
+ List.of((Object) row)
+ )
+ )
+ )
+ ).orElseThrow(() -> QueryInterruptedException.wrapIfNeeded(failure));
+ }
+
+ private static boolean isNodeAvailabilityFailure(final Throwable failure)
+ {
+ Throwable cause = failure;
+ while (cause != null) {
+ if (cause instanceof SocketException
+ || cause instanceof UnknownHostException
+ || cause instanceof EOFException
+ || cause instanceof ClosedChannelException
+ || cause instanceof ChannelException) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
+ }
+
+ private static boolean isLeaderChangeFailure(final Throwable failure)
+ {
+ Throwable cause = failure;
+ while (cause != null) {
+ if (cause instanceof SystemTableNotLeaderException
+ || cause instanceof QueryException
+ && SystemTableNotLeaderException.class.getName().equals(((QueryException) cause).getErrorClass())) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
+ }
+
+ private static RuntimeException propagate(final Exception failure)
+ {
+ if (failure instanceof RuntimeException runtimeException) {
+ return runtimeException;
+ } else if (failure instanceof java.util.concurrent.TimeoutException) {
+ return new QueryTimeoutException(failure.getMessage());
+ } else {
+ if (failure instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ return QueryInterruptedException.wrapIfNeeded(failure);
+ }
+ }
+
+ private static Sequence<Object[]> rowsFromScanResult(final ScanResultValue scanResult)
+ {
+ return Sequences.simple((List<?>) scanResult.getEvents())
+ .map(event -> event instanceof Object[] ? (Object[]) event : ((List<?>) event).toArray());
+ }
+
+ private static long getNodeFailTime(final Query<?> query)
+ {
+ final long existingFailTime = query.context().getLong(DirectDruidClient.QUERY_FAIL_TIME, -1L);
+ if (existingFailTime > 0) {
+ return existingFailTime;
+ } else if (query.context().hasTimeout()) {
+ return System.currentTimeMillis() + query.context().getTimeout();
+ } else {
+ return JodaUtils.MAX_INSTANT;
+ }
+ }
+
+ /**
+ * Adapts a Druid {@link Sequence} to the lazy {@link Iterable} accepted by {@link InlineDataSource}.
+ *
+ * <p>The iterator owns the sequence yielder and closes it when the sequence is exhausted. This is important for
+ * direct node clients because closing the yielder closes the underlying response stream and allows an early
+ * terminating Broker query to cancel the node request.</p>
+ */
+ private static <T> Iterable<T> sequenceAsIterable(
+ final Sequence<T> sequence,
+ final NodeSequenceCloser nodeSequenceCloser
+ )
+ {
+ return new Iterable<>()
+ {
+ private boolean iterated;
+
+ @Override
+ public Iterator<T> iterator()
+ {
+ if (iterated) {
+ throw new ISE("A native system table result sequence can only be consumed once");
+ }
+ iterated = true;
+
+ return new Iterator<>()
+ {
+ @Nullable
+ private Yielder<T> yielder = Yielders.each(sequence);
+ private boolean closed;
+
+ {
+ nodeSequenceCloser.setYielder(yielder);
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (yielder == null) {
+ return false;
+ }
+ if (yielder.isDone()) {
+ close();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public T next()
+ {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final Yielder<T> currentYielder = Preconditions.checkNotNull(yielder);
+ final T value = currentYielder.get();
+ yielder = currentYielder.next(null);
+ nodeSequenceCloser.setYielder(yielder);
+ return value;
+ }
+
+ private void close()
+ {
+ if (!closed) {
+ closed = true;
+ yielder = null;
+ nodeSequenceCloser.close();
+ }
+ }
+ };
+ }
+ };
+ }
+
+ private static class NodeSequenceCloser implements Closeable
+ {
+ @Nullable
+ private Yielder<?> yielder;
+ private boolean closed;
+
+ synchronized void setYielder(final Yielder<?> newYielder)
+ {
+ if (closed) {
+ CloseableUtils.closeAndWrapExceptions(newYielder);
+ } else {
+ yielder = newYielder;
+ }
+ }
+
+ @Override
+ public synchronized void close()
+ {
+ if (!closed) {
+ closed = true;
+ final Yielder<?> currentYielder = yielder;
+ yielder = null;
+ CloseableUtils.closeAndWrapExceptions(currentYielder);
+ }
+ }
+ }
+
+ private static class NodeSequencesCloser implements Closeable
+ {
+ private final List<NodeSequenceCloser> nodeSequenceClosers = new ArrayList<>();
+ private final List<String> nodeQueryIds = new ArrayList<>();
+ private final QueryScheduler queryScheduler;
+ private boolean closed;
+
+ private NodeSequencesCloser(final QueryScheduler queryScheduler)
+ {
+ this.queryScheduler = queryScheduler;
+ }
+
+ synchronized void addQueryId(final String nodeQueryId)
+ {
+ if (closed) {
+ queryScheduler.cancelQuery(nodeQueryId);
+ } else {
+ nodeQueryIds.add(nodeQueryId);
+ }
+ }
+
+ synchronized void add(final NodeSequenceCloser nodeSequenceCloser)
+ {
+ if (closed) {
+ CloseableUtils.closeAndWrapExceptions(nodeSequenceCloser);
+ } else {
+ nodeSequenceClosers.add(nodeSequenceCloser);
+ }
+ }
+
+ @Override
+ public synchronized void close() throws IOException
+ {
+ if (!closed) {
+ closed = true;
+ try {
+ CloseableUtils.closeAll(nodeSequenceClosers);
+ }
+ finally {
+ nodeSequenceClosers.clear();
+ nodeQueryIds.forEach(queryScheduler::cancelQuery);
+ nodeQueryIds.clear();
+ }
+ }
+ }
+ }
+
+ private static class RecoveringNodeIterator implements Iterator<ScanResultValue>, Closeable
+ {
+ private final SystemTableDescriptor descriptor;
+ @Nullable
+ private final Supplier<NodeQuerySequence> leaderRetrySupplier;
+
+ private NodeQuerySequence nodeQuerySequence;
+
+ @Nullable
+ private Yielder<ScanResultValue> yielder;
+ @Nullable
+ private ScanResultValue failureResult;
+ private boolean started;
+ private boolean finished;
+ private boolean retriedLeader;
+ private boolean emittedResult;
+
+ private RecoveringNodeIterator(
+ final NodeQuerySequence nodeQuerySequence,
+ final SystemTableDescriptor descriptor,
+ @Nullable final Supplier<NodeQuerySequence> leaderRetrySupplier
+ )
+ {
+ this.nodeQuerySequence = nodeQuerySequence;
+ this.descriptor = descriptor;
+ this.leaderRetrySupplier = leaderRetrySupplier;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (failureResult != null) {
+ return true;
+ }
+ if (finished) {
+ return false;
+ }
+
+ try {
+ if (!started) {
+ started = true;
+ yielder = Yielders.each(nodeQuerySequence.sequence());
+ }
+ if (Preconditions.checkNotNull(yielder).isDone()) {
+ finished = true;
+ close();
+ return false;
+ }
+ return true;
+ }
+ catch (Exception e) {
+ return recoverOrThrow(e) ? hasNext() : true;
+ }
+ }
+
+ @Override
+ public ScanResultValue next()
+ {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ if (failureResult != null) {
+ final ScanResultValue result = failureResult;
+ failureResult = null;
+ finished = true;
+ return result;
+ }
+
+ final Yielder<ScanResultValue> currentYielder = Preconditions.checkNotNull(yielder);
+ final ScanResultValue result;
+ try {
+ result = currentYielder.get();
+ emittedResult = true;
+ }
+ catch (Exception e) {
+ // Either the leader retry restarted the sequence or a failure row is pending; next() serves both.
+ recoverOrThrow(e);
+ return next();
+ }
+ try {
+ yielder = currentYielder.next(null);
+ }
+ catch (Exception e) {
+ recoverOrThrow(e);
+ }
+ return result;
+ }
+
+ private boolean recoverOrThrow(final Exception originalFailure)
+ {
+ Exception failure = originalFailure;
+ if (!emittedResult && !retriedLeader && leaderRetrySupplier != null && isLeaderChangeFailure(failure)) {
+ retriedLeader = true;
+ try {
+ close();
+ nodeQuerySequence = leaderRetrySupplier.get();
+ started = false;
+ finished = false;
+ return true;
+ }
+ catch (Exception retryFailure) {
+ failure = retryFailure;
+ }
+ }
+ if (!isNodeAvailabilityFailure(failure)) {
+ throw CloseableUtils.closeAndWrapInCatch(failure, this);
+ }
+ final Optional<Object[]> failureRow = descriptor.getNodeFailureRow(
+ nodeQuerySequence.node().getDiscoveryNode().getDruidNode(),
+ nodeQuerySequence.node().getNodeRoles(),
+ failure
+ );
+ if (failureRow.isEmpty()) {
+ throw CloseableUtils.closeAndWrapInCatch(failure, this);
+ }
+ try {
+ close();
+ }
+ catch (RuntimeException closeException) {
+ failure.addSuppressed(closeException);
+ }
+ failureResult = new ScanResultValue(
+ null,
+ descriptor.getRowSignature().getColumnNames(),
+ List.of((Object) failureRow.get())
+ );
+ return false;
+ }
+
+ @Override
+ public void close()
+ {
+ final Yielder<ScanResultValue> currentYielder = yielder;
+ yielder = null;
+ CloseableUtils.closeAndWrapExceptions(currentYielder);
+ }
+ }
+
+ private record NodeQuerySequence(SystemTableNode node, Sequence<ScanResultValue> sequence)
+ {
+ }
+
+ @Nullable
+ private static DimFilter nodeFilter(final SystemTableDataSource dataSource, final Query<?> query)
+ {
+ // Push into the remote scan only when this query directly owns the system table. A join or other composite filter
+ // remains on the resolved Broker query, where the normal native query machinery can apply it with full datasource
+ // semantics; it cannot safely be attributed to an individual remote leaf here.
+ if (!dataSource.equals(query.getDataSource())) {
+ return null;
+ }
+
+ final List<DimFilter> filters = new ArrayList<>();
+ if (query.getFilter() != null) {
+ filters.add(query.getFilter());
+ }
+ if (query instanceof WindowOperatorQuery) {
+ for (final OperatorFactory operator : ((WindowOperatorQuery) query).getLeafOperators()) {
+ if (operator instanceof ScanOperatorFactory && ((ScanOperatorFactory) operator).getFilter() != null) {
+ filters.add(((ScanOperatorFactory) operator).getFilter());
+ }
+ }
+ }
+ if (filters.isEmpty()) {
+ return null;
+ } else if (filters.size() == 1) {
+ return filters.get(0);
+ } else {
+ return new AndDimFilter(filters);
+ }
+ }
+
+ private static VirtualColumns nodeVirtualColumns(
+ final SystemTableDataSource dataSource,
+ final Query<?> query,
+ @Nullable final DimFilter nodeFilter
+ )
+ {
+ if (!dataSource.equals(query.getDataSource()) || nodeFilter == null) {
+ return VirtualColumns.EMPTY;
+ }
+
+ final List<VirtualColumn> virtualColumns = new ArrayList<>(List.of(query.getVirtualColumns().getVirtualColumns()));
+ if (query instanceof WindowOperatorQuery) {
+ for (final OperatorFactory operator : ((WindowOperatorQuery) query).getLeafOperators()) {
+ if (operator instanceof ScanOperatorFactory) {
+ virtualColumns.addAll(List.of(((ScanOperatorFactory) operator).getVirtualColumns().getVirtualColumns()));
Review Comment:
[P1] Handle null window-leaf virtual columns
**Finding:** `WindowOperatorQuery` intentionally stores null in `ScanOperatorFactory` when a leaf has no virtual columns. When the owning window query has a pushed-down filter, `nodeFilter` is non-null, so this expression dereferences that null value for any empty leaf and fails with an NPE before the node query executes.
**Suggestion:** Treat null leaf virtual columns as empty before adding them, and add regression coverage for a filtered window query with a leaf that has no virtual columns.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3964261599
##########
services/src/main/java/org/apache/druid/cli/CliHistorical.java:
##########
@@ -92,30 +86,22 @@ protected List<? extends Module> getModules()
{
return ImmutableList.of(
new DruidProcessingModule(),
- new QueryableModule(),
- new QueryRunnerFactoryModule(),
- new SegmentWranglerModule(),
- new JoinableFactoryModule(),
Review Comment:
Not out of the scope. it's a tiny refactoring of how native query related module are installed. otherwise, for overlord, coordinator, middle manager, we will repeate these code. Now we only need the `NativeQueryEngineModule` installed, all dependencies are enclosed in this new module.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4361394056
##########
server/src/main/java/org/apache/druid/server/system/table/SystemTableDataProvider.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.table;
+
+import jakarta.validation.constraints.NotNull;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.server.security.AuthenticationResult;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Supplies storage-prefiltered rows of one native system table. Authorization isn't applied here: the table's
+ * {@link SystemTableDescriptor#getRowAuthorizer()} filters these rows wherever they are consumed.
+ * The implementation is deployed in related service.
+ * For example, the data provider of sys.tasks is deployed in overlord module
+ * */
+public interface SystemTableDataProvider
+{
+ default List<SystemTablePushdownFilter> getPushdownFilters()
+ {
+ return Collections.emptyList();
+ }
+
+ Iterable<Object[]> getRows(
+ @NotNull List<DimFilter> filters,
+ AuthenticationResult internalAuthenticationResult
Review Comment:
## CodeQL / Useless parameter
The parameter 'internalAuthenticationResult' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/6816)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4367256564
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryClient.java:
##########
@@ -0,0 +1,923 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import io.netty.channel.ChannelException;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.client.DirectDruidClientFactory;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.LazySequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.Yielders;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContext;
+import org.apache.druid.query.QueryDataSource;
+import org.apache.druid.query.QueryException;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.QueryContextParameters;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.QueryScheduler;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.system.SystemTableNotLeaderException;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.utils.CloseableUtils;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+/**
+ * Recursively replaces native system-table datasources with authorized inline rows fetched from nodes, then
+ * executes the rebuilt query through the Broker's normal native-query walker.
+ */
+public class SystemTableQueryClient implements DataSourceQueryHandler
+{
+ private static final String SYSTEM_TABLE_TIER = "_system";
+
+ private final SystemTableNodeLocator nodeLocator;
+ private final DirectDruidClientFactory directDruidClientFactory;
+ private final QueryScheduler queryScheduler;
+ private final QuerySegmentWalker querySegmentWalker;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final AuthorizerMapper authorizerMapper;
+ private final SystemTableQueryHandler localQueryHandler;
+ private final AuthenticationResult escalatedAuthenticationResult;
+ private final DruidNode selfNode;
+
+ @Inject
+ public SystemTableQueryClient(
+ final SystemTableNodeLocator nodeLocator,
+ final DirectDruidClientFactory directDruidClientFactory,
+ final QueryScheduler queryScheduler,
+ final QuerySegmentWalker querySegmentWalker,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final AuthorizerMapper authorizerMapper,
+ final SystemTableQueryHandler localQueryHandler,
+ final Escalator escalator,
+ @Self final DruidNode selfNode
+ )
+ {
+ this.nodeLocator = nodeLocator;
+ this.directDruidClientFactory = directDruidClientFactory;
+ this.queryScheduler = queryScheduler;
+ this.querySegmentWalker = querySegmentWalker;
+ this.tableDescriptors = tableDescriptors;
+ this.authorizerMapper = authorizerMapper;
+ this.localQueryHandler = localQueryHandler;
+ this.escalatedAuthenticationResult = escalator.createEscalatedAuthenticationResult();
+ this.selfNode = selfNode;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult authenticationResult,
+ final boolean executeLocally
+ )
+ {
+ return (queryPlus, responseContext) -> {
+ final NodeSequencesCloser nodeSequencesCloser = new NodeSequencesCloser(queryScheduler);
+ try {
+ final DataSource resolvedDataSource = resolveDataSource(
+ query.getDataSource(),
+ query,
+ authenticationResult,
+ nodeSequencesCloser
+ );
+ final Query<T> resolvedQuery = query.withDataSource(resolvedDataSource).withOverriddenContext(
+ QueryContextParameters.QUERY_RESOURCE_ID,
+ UUID.randomUUID().toString()
+ );
+ final QueryRunner<T> queryRunner = querySegmentWalker.getQueryRunnerForIntervals(
+ resolvedQuery,
+ resolvedQuery.getIntervals()
+ );
+ return Sequences.withBaggage(
+ queryRunner.run(queryPlus.withQuery(resolvedQuery), responseContext),
+ nodeSequencesCloser
+ );
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, nodeSequencesCloser);
+ }
+ };
+ }
+
+ private DataSource resolveDataSource(
+ final DataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ if (dataSource instanceof SystemTableDataSource systemTableDataSource) {
+ return resolveSystemTableDataSource(
+ systemTableDataSource,
+ owningQuery,
+ authenticationResult,
+ nodeSequencesCloser
+ );
+ }
+
+ final Query<?> childOwningQuery;
+ final DataSource resolvedParent;
+ if (dataSource instanceof QueryDataSource queryDataSource) {
+ final Query<?> childQuery = queryDataSource.getQuery();
+ // An inner query may tighten the outer guardrails, but must not discard them before node fanout.
+ childOwningQuery = childQuery.withOverriddenContext(
+ QueryContext.ofMap(
+ QueryContextParameters.QUERY_FAIL_TIME,
+ Math.min(getNodeFailTime(owningQuery), getNodeFailTime(childQuery)),
+ QueryContextParameters.MAX_SCATTER_GATHER_BYTES,
+ Math.min(owningQuery.context().getMaxScatterGatherBytes(), childQuery.context().getMaxScatterGatherBytes())
+ )
+ );
+ resolvedParent = new QueryDataSource(childOwningQuery);
+ } else {
+ childOwningQuery = owningQuery;
+ resolvedParent = dataSource;
+ }
+ final List<DataSource> resolvedChildren = new ArrayList<>();
+ for (final DataSource child : dataSource.getChildren()) {
+ resolvedChildren.add(
+ resolveDataSource(
+ child,
+ childOwningQuery,
+ authenticationResult,
+ nodeSequencesCloser
+ )
+ );
+ }
+ return resolvedParent.withChildren(resolvedChildren);
+ }
+
+ private InlineDataSource resolveSystemTableDataSource(
+ final SystemTableDataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final SystemTableDescriptor descriptor = tableDescriptors.get(dataSource.getTable());
+ if (descriptor == null) {
+ throw DruidException.defensive("No routing descriptor is registered for system table[%s]", dataSource.getTable());
+ }
+
+ final ScanQuery nodeQuery = makeNodeQuery(dataSource, descriptor, owningQuery);
+ final List<QueryRunner<ScanResultValue>> nodeRunners = makeNodeRunners(
+ nodeQuery,
+ descriptor,
+ nodeSequencesCloser
+ );
+ if (nodeRunners.isEmpty() && !descriptor.isEmptyDiscoveryAllowed()) {
+ throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.SERVICE_UNAVAILABLE)
+ .build("No node is available to serve system table[%s]", dataSource.getTable());
+ }
+
+ /*
+ * Keep the scan result transport lazy. The node response is a Sequence, so this lets the local query
+ * runner consume and process rows while the HTTP response is still arriving instead of materializing every
+ * node row in two Broker-side lists. The scan transport is intentionally kept behind this adapter so a
+ * future node-side aggregation mode can use a different result adapter without changing routing or
+ * authorization orchestration here.
+ */
+ final NodeSequenceCloser nodeSequenceCloser = new NodeSequenceCloser();
+ nodeSequencesCloser.add(nodeSequenceCloser);
+ // Node scans get their own response context. Sharing the caller's would add the node scans' row counts to the
+ // Broker query's, and a Scan with a limit returns no rows once that count reaches the limit. For example, the
+ // node scans of a join's right side would empty a limited Scan over the join before it read its left side.
+ final Iterable<Object[]> nodeRows = scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ DirectDruidClient.makeResponseContextForQuery(),
Review Comment:
[P2] Query-wide scatter-gather limit resets per system-table leaf
**Finding:** This creates a fresh ResponseContext for every resolved system-table leaf. DirectDruidClient stores and checks maxScatterGatherBytes in that context, so a query with multiple system-table leaves, such as a self-join or UNION of sys.server_properties, gets a separate full byte budget per leaf; the counters are not merged into the caller's context. The documented query-wide guardrail can therefore be exceeded, allowing more response data to be buffered on the Broker than configured.
**Suggestion:** Share an aggregate byte counter across the per-leaf response contexts while keeping row-scan counters isolated, then merge each leaf's byte total into the query-wide counter before subsequent leaf fan-out.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3969307620
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryHandler.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.inject.Inject;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryEngine;
+import org.apache.druid.segment.InlineSegmentWrangler;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.system.table.SystemTableDataProvider;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTablePushdownFilter;
+
+import java.util.Map;
+
+/** Resolves one node-local system table and returns its rows through the standard native Scan stack. */
+public class SystemTableQueryHandler implements DataSourceQueryHandler
+{
+ private final Map<String, SystemTableDataProvider> dataSuppliers;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final ScanQueryEngine scanQueryEngine;
+ private final AuthorizerMapper authorizerMapper;
+
+ @Inject
+ public SystemTableQueryHandler(
+ final Map<String, SystemTableDataProvider> dataSuppliers,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final ScanQueryEngine scanQueryEngine,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.dataSuppliers = dataSuppliers;
+ this.tableDescriptors = tableDescriptors;
+ this.scanQueryEngine = scanQueryEngine;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult requestAuthenticationResult,
+ final boolean executeLocally
+ )
+ {
+ if (!(query instanceof ScanQuery)) {
+ throw new IAE("Local system table queries must be scan queries");
+ }
+
+ final SystemTableDataSource dataSource = (SystemTableDataSource) query.getDataSource();
Review Comment:
[P2] [P2] Handle nested system-table datasources before casting the root
`QueryLifecycle.findDataSourceQueryHandler` deliberately selects this handler when a registered datasource appears below the root, and `DataSourceQueryHandler` documents that the handler receives the complete query. A native `ScanQuery` such as a join or `QueryDataSource` containing `SystemTableDataSource` therefore reaches this method with a composite root and throws `ClassCastException` at this cast. The Router also forwards a composite query carrying `X-Druid-Native-Query-Route: local` to the Broker, whose local handler takes this same path. Native system-table joins are advertised as supported; either recursively replace local system-table leaves here or reject/route composite local requests before selecting this handler.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4361140289
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableNodeLocator.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.inject.Inject;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.rpc.indexing.OverlordClient;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/** Resolves a descriptor's logical routing policy to concrete Druid processes. */
+public class SystemTableNodeLocator
+{
+ private final DruidNodeDiscoveryProvider discoveryProvider;
+ private final Map<NodeRole, SystemTableLeaderLocator> leaderLocators;
+
+ @Inject
+ public SystemTableNodeLocator(
+ final DruidNodeDiscoveryProvider discoveryProvider,
+ final CoordinatorClient coordinatorClient,
+ final OverlordClient overlordClient
+ )
+ {
+ this.discoveryProvider = discoveryProvider;
+ this.leaderLocators = Map.of(
+ NodeRole.COORDINATOR,
+ coordinatorClient::findCurrentLeader,
+ NodeRole.OVERLORD,
+ overlordClient::findCurrentLeader
+ );
+ }
+
+ List<SystemTableNode> locate(final SystemTableDescriptor descriptor, final Query<?> query)
Review Comment:
Correct, this is not included in this PR. This is because doing so we need to unify the definition of `server` and `service_name` across all system tables so that we can prune servers for any tables.
This can be done as a follow up after all other system tables have supported the native query execution path.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4365862477
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
Review Comment:
I asked Claude to propose a better name based on the context, it suggested me this name `useQueryEngineForSystemTables` which I think is better and satisfies our current document description
> By default, the Broker evaluates queries on "sys" tables with a built-in SQL interpreter, which supports only standard SQL functions. You can enable
experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
aggregations that the native SQL engine can translate.
The implementation has been updated.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4271802285
##########
server/src/main/java/org/apache/druid/server/QueryLifecycle.java:
##########
@@ -420,13 +460,49 @@ public <T> QueryResponse<T> execute()
final ResponseContext responseContext = DirectDruidClient.makeResponseContextForQuery();
@SuppressWarnings("unchecked")
- final Sequence<T> res = QueryPlus.wrap((Query<T>) baseQuery)
- .withIdentity(authenticationResult.getIdentity())
- .run(texasRanger, responseContext);
+ final Query<T> query = (Query<T>) baseQuery;
+ final DataSourceQueryHandler dataSourceQueryHandler = findDataSourceQueryHandler(query.getDataSource());
+ final QueryRunner<T> queryRunner = dataSourceQueryHandler == null
+ ? query.getRunner(texasRanger)
+ : dataSourceQueryHandler.createRunner(
Review Comment:
[P1] Preserve server timeout enforcement for handler-backed queries
**Finding:** Handler-backed queries skip the normal SetAndVerifyContextQueryRunner path, which applies ServerConfig default and maximum timeouts and writes DirectDruidClient.QUERY_FAIL_TIME. The system-table handlers instead derive deadlines from the raw query context: omitted timeouts use the query-context fallback rather than druid.server.http.defaultQueryTimeout, timeout: 0 becomes JodaUtils.MAX_INSTANT instead of the server maximum, and values above maxQueryTimeout are not rejected. Since node fanout begins before the later normal walker call, a stalled node request can outlive or bypass the configured query deadline.
**Suggestion:** Normalize handler-backed queries with the same server timeout/max-timeout policy before invoking the handler, and pass the normalized deadline into both local and fanout execution paths.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4325154670
##########
server/src/main/java/org/apache/druid/server/QueryLifecycle.java:
##########
@@ -420,13 +465,58 @@ public <T> QueryResponse<T> execute()
final ResponseContext responseContext = DirectDruidClient.makeResponseContextForQuery();
@SuppressWarnings("unchecked")
- final Sequence<T> res = QueryPlus.wrap((Query<T>) baseQuery)
- .withIdentity(authenticationResult.getIdentity())
- .run(texasRanger, responseContext);
+ final Query<T> originalQuery = (Query<T>) baseQuery;
+ final DataSourceQueryHandler dataSourceQueryHandler = findDataSourceQueryHandler(originalQuery.getDataSource());
+ // Handlers can start fanout or provider work before reaching a segment walker, or bypass it entirely.
+ // Apply the same server guardrails first, and pass the normalized query to both runner creation and execution.
+ final Query<T> query = dataSourceQueryHandler == null
+ ? originalQuery
+ : SetAndVerifyContextQueryRunner.withTimeoutAndMaxScatterGatherBytes(
+ originalQuery,
+ serverConfig,
+ System.currentTimeMillis()
+ );
+ final QueryRunner<T> queryRunner = dataSourceQueryHandler == null
+ ? query.getRunner(texasRanger)
+ : dataSourceQueryHandler.createRunner(
Review Comment:
[P2] Register handler queries for cancellation
**Finding:** Handler-backed native system-table queries bypass QueryScheduler registration. QueryResource cancellation therefore cannot resolve the user-facing query ID, while generated child requests and local provider work continue independently.
**Suggestion:** Register the handler's top-level execution or an equivalent parent cancellation handle with QueryScheduler using the user query ID, and ensure cancellation closes NodeSequencesCloser, local/provider work, and generated child futures.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4359001564
##########
server/src/main/java/org/apache/druid/client/DirectDruidClient.java:
##########
@@ -785,7 +785,8 @@ private void checkTotalBytesLimit(long bytes)
.setHeader(
HttpHeaders.Names.CONTENT_TYPE,
isSmile ? SmileMediaTypes.APPLICATION_JACKSON_SMILE : MediaType.APPLICATION_JSON
- ),
+ )
+ .setHeader(QueryResource.HEADER_NATIVE_QUERY_ROUTE, QueryResource.NATIVE_QUERY_ROUTE_LOCAL),
Review Comment:
This appears to be getting set unconditionally. It would be nice to only set it for the requests that need it, for a few reasons:
- to make it clear it's only needed for certain requests
- cleanliness
- to save the bytes
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT server) FROM sys.server_properties",
+ "context": {
+ "allowNativeFunctionsForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET allowNativeFunctionsForSystemTables = 'true';
+SELECT COUNT(DISTINCT server) FROM sys.server_properties;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `allowNativeFunctionsForSystemTables` is `true`.
+
+The parameter defaults to `false`. During a rolling upgrade, leave it disabled until the Broker and all nodes
+that serve the native system tables have been upgraded. After the upgrade, you can enable it by query or set
+`druid.query.default.context.allowNativeFunctionsForSystemTables=true` on Brokers as the cluster-wide default. For more
+information, see [SQL query context](sql-query-context.md).
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
+system-table Scan against only the contacted node, set the HTTP header
+`X-Druid-Native-Query-Route: local`. Local execution uses the authenticated request identity and applies the table's
Review Comment:
What authorization rules are used when the header is not present? The sentence as written implies that it's something other than "the table's authorization rules", but hopefully they _are_ still applied.
##########
processing/src/main/java/org/apache/druid/query/SystemTableDataSource.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Identifies a Druid system table whose rows are supplied by a service rather than by segments.
+ */
+public class SystemTableDataSource extends LeafDataSource
Review Comment:
This doesn't override `withPolicies` but it should, because it has table names (`getTableNames`). I raised https://github.com/apache/druid/pull/20705 which would throw from `withPolicies` for this class.
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableNodeLocator.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.inject.Inject;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.rpc.indexing.OverlordClient;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/** Resolves a descriptor's logical routing policy to concrete Druid processes. */
+public class SystemTableNodeLocator
+{
+ private final DruidNodeDiscoveryProvider discoveryProvider;
+ private final Map<NodeRole, SystemTableLeaderLocator> leaderLocators;
+
+ @Inject
+ public SystemTableNodeLocator(
+ final DruidNodeDiscoveryProvider discoveryProvider,
+ final CoordinatorClient coordinatorClient,
+ final OverlordClient overlordClient
+ )
+ {
+ this.discoveryProvider = discoveryProvider;
+ this.leaderLocators = Map.of(
+ NodeRole.COORDINATOR,
+ coordinatorClient::findCurrentLeader,
+ NodeRole.OVERLORD,
+ overlordClient::findCurrentLeader
+ );
+ }
+
+ List<SystemTableNode> locate(final SystemTableDescriptor descriptor, final Query<?> query)
Review Comment:
This method appears to be ignoring the query filter, which is a regression in efficiency. The Bindable implementation can prune the set of servers to query based on `server` and `service_name`.
##########
server/src/main/java/org/apache/druid/server/system/table/ServerPropertiesTableDataProvider.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.table;
+
+import com.google.common.collect.Maps;
+import com.google.inject.Inject;
+import org.apache.druid.client.DruidServerConfig;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.EqualityFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizationResult;
+import org.apache.druid.server.security.AuthorizationUtils;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.ForbiddenException;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+/** Native row supplier for {@code sys.server_properties}. */
+public class ServerPropertiesTableDataProvider implements SystemTableDataProvider
+{
+ private static final List<SystemTablePushdownFilter> PUSHDOWN_FILTERS = List.of(
+ new SystemTablePushdownFilter("server", null),
+ new SystemTablePushdownFilter("service_name", null)
+ );
+
+ private final DruidNode selfNode;
+ private final Set<NodeRole> selfNodeRoles;
+ private final AuthorizerMapper authorizerMapper;
+ private final Properties properties;
+ private final DruidServerConfig druidServerConfig;
+
+ @Inject
+ public ServerPropertiesTableDataProvider(
+ @Self final DruidNode selfNode,
+ @Self final Set<NodeRole> selfNodeRoles,
+ final AuthorizerMapper authorizerMapper,
+ final Properties properties,
+ final DruidServerConfig druidServerConfig
+ )
+ {
+ this.selfNode = selfNode;
+ this.selfNodeRoles = selfNodeRoles;
+ this.authorizerMapper = authorizerMapper;
+ this.properties = properties;
+ this.druidServerConfig = druidServerConfig;
+ }
+
+ @Override
+ public List<SystemTablePushdownFilter> getPushdownFilters()
+ {
+ return PUSHDOWN_FILTERS;
+ }
+
+ @Override
+ public Iterable<Object[]> getRows(
+ final List<DimFilter> filters,
+ final AuthenticationResult internalAuthenticationResult
+ )
+ {
+ authorizeServerRead(internalAuthenticationResult);
+
+ String serverFilter = null;
+ String serviceNameFilter = null;
+ for (final DimFilter filter : filters) {
+ if (isStringEqualityFilter(filter, "server")) {
+ serverFilter = getFilterValue(filter);
+ } else if (isStringEqualityFilter(filter, "service_name")) {
+ serviceNameFilter = getFilterValue(filter);
+ }
+ }
+
+ final String server = selfNode.getHostAndPortToUse();
+ if (serverFilter != null && !serverFilter.equals(server)) {
+ return Collections.emptyList();
+ }
+ if (serviceNameFilter != null && !serviceNameFilter.equals(selfNode.getServiceName())) {
+ return Collections.emptyList();
+ }
+
+ final ServerProperties serverProperties = new ServerProperties(selfNode.getServiceName(), server);
+ selfNodeRoles.stream()
+ .map(NodeRole::getJsonName)
+ .sorted()
+ .forEach(serverProperties.nodeRoles::add);
+
+ return buildRows(serverProperties);
+ }
+
+ private void authorizeServerRead(final AuthenticationResult authenticationResult)
Review Comment:
Seems like it could be a helper, all system tables will need this.
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
Review Comment:
IMO `useNativeEngineForSystemTables` is a better name for this parameter.
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryClient.java:
##########
@@ -0,0 +1,917 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import io.netty.channel.ChannelException;
+import org.apache.druid.client.DirectDruidClientFactory;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.LazySequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.Yielders;
+import org.apache.druid.query.BaseQuery;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.QueryDataSource;
+import org.apache.druid.query.QueryException;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.QueryContextParameters;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.QueryScheduler;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.system.SystemTableNotLeaderException;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.utils.CloseableUtils;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+/**
+ * Recursively replaces native system-table datasources with authorized inline rows fetched from nodes, then
+ * executes the rebuilt query through the Broker's normal native-query walker.
+ */
+public class SystemTableQueryClient implements DataSourceQueryHandler
+{
+ private static final String SYSTEM_TABLE_TIER = "_system";
+
+ private final SystemTableNodeLocator nodeLocator;
+ private final DirectDruidClientFactory directDruidClientFactory;
+ private final QueryScheduler queryScheduler;
+ private final QuerySegmentWalker querySegmentWalker;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final AuthorizerMapper authorizerMapper;
+ private final SystemTableQueryHandler localQueryHandler;
+ private final AuthenticationResult escalatedAuthenticationResult;
+ private final DruidNode selfNode;
+
+ @Inject
+ public SystemTableQueryClient(
+ final SystemTableNodeLocator nodeLocator,
+ final DirectDruidClientFactory directDruidClientFactory,
+ final QueryScheduler queryScheduler,
+ final QuerySegmentWalker querySegmentWalker,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final AuthorizerMapper authorizerMapper,
+ final SystemTableQueryHandler localQueryHandler,
+ final Escalator escalator,
+ @Self final DruidNode selfNode
+ )
+ {
+ this.nodeLocator = nodeLocator;
+ this.directDruidClientFactory = directDruidClientFactory;
+ this.queryScheduler = queryScheduler;
+ this.querySegmentWalker = querySegmentWalker;
+ this.tableDescriptors = tableDescriptors;
+ this.authorizerMapper = authorizerMapper;
+ this.localQueryHandler = localQueryHandler;
+ this.escalatedAuthenticationResult = escalator.createEscalatedAuthenticationResult();
+ this.selfNode = selfNode;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult authenticationResult,
+ final boolean executeLocally
+ )
+ {
+ return (queryPlus, responseContext) -> {
+ final NodeSequencesCloser nodeSequencesCloser = new NodeSequencesCloser(queryScheduler);
+ try {
+ final DataSource resolvedDataSource = resolveDataSource(
+ query.getDataSource(),
+ query,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ final Query<T> resolvedQuery = query.withDataSource(resolvedDataSource).withOverriddenContext(
+ Map.of(QueryContexts.QUERY_RESOURCE_ID, UUID.randomUUID().toString())
+ );
+ final QueryRunner<T> queryRunner = querySegmentWalker.getQueryRunnerForIntervals(
+ resolvedQuery,
+ resolvedQuery.getIntervals()
+ );
+ return Sequences.withBaggage(
+ queryRunner.run(queryPlus.withQuery(resolvedQuery), responseContext),
+ nodeSequencesCloser
+ );
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, nodeSequencesCloser);
+ }
+ };
+ }
+
+ private DataSource resolveDataSource(
+ final DataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ if (dataSource instanceof SystemTableDataSource systemTableDataSource) {
+ return resolveSystemTableDataSource(
+ systemTableDataSource,
+ owningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ }
+
+ final Query<?> childOwningQuery;
+ final DataSource resolvedParent;
+ if (dataSource instanceof QueryDataSource queryDataSource) {
+ final Query<?> childQuery = queryDataSource.getQuery();
+ // An inner query may tighten the outer guardrails, but must not discard them before node fanout.
+ childOwningQuery = childQuery.withOverriddenContext(
+ Map.of(
+ QueryContextParameters.QUERY_FAIL_TIME.getName(),
+ Math.min(getNodeFailTime(owningQuery), getNodeFailTime(childQuery)),
+ QueryContextParameters.MAX_SCATTER_GATHER_BYTES.getName(),
+ Math.min(owningQuery.context().getMaxScatterGatherBytes(), childQuery.context().getMaxScatterGatherBytes())
+ )
+ );
+ resolvedParent = new QueryDataSource(childOwningQuery);
+ } else {
+ childOwningQuery = owningQuery;
+ resolvedParent = dataSource;
+ }
+ final List<DataSource> resolvedChildren = new ArrayList<>();
+ for (final DataSource child : dataSource.getChildren()) {
+ resolvedChildren.add(
+ resolveDataSource(
+ child,
+ childOwningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ )
+ );
+ }
+ return resolvedParent.withChildren(resolvedChildren);
+ }
+
+ private InlineDataSource resolveSystemTableDataSource(
+ final SystemTableDataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final SystemTableDescriptor descriptor = tableDescriptors.get(dataSource.getTable());
+ if (descriptor == null) {
+ throw new ISE("No routing descriptor is registered for system table[%s]", dataSource.getTable());
+ }
+
+ final ScanQuery nodeQuery = makeNodeQuery(dataSource, descriptor, owningQuery);
+ final List<QueryRunner<ScanResultValue>> nodeRunners = makeNodeRunners(
+ nodeQuery,
+ descriptor,
+ nodeSequencesCloser
+ );
+ if (nodeRunners.isEmpty() && !descriptor.isEmptyDiscoveryAllowed()) {
+ throw new ISE("No node is available to serve system table[%s]", dataSource.getTable());
+ }
+
+ /*
+ * Keep the scan result transport lazy. The node response is a Sequence, so this lets the local query
+ * runner consume and process rows while the HTTP response is still arriving instead of materializing every
+ * node row in two Broker-side lists. The scan transport is intentionally kept behind this adapter so a
+ * future node-side aggregation mode can use a different result adapter without changing routing or
+ * authorization orchestration here.
+ */
+ final NodeSequenceCloser nodeSequenceCloser = new NodeSequenceCloser();
+ nodeSequencesCloser.add(nodeSequenceCloser);
+ final Iterable<Object[]> nodeRows = scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ responseContext,
+ nodeSequenceCloser
+ );
+ final Iterable<Object[]> authorizedRows = descriptor.getRowAuthorizer().filterAuthorizedRows(
+ nodeRows,
+ authenticationResult,
+ authorizerMapper
+ );
+ final Iterable<Object[]> queryRows = requiresMaterializedRows(owningQuery)
+ ? materializeRows(authorizedRows)
+ : authorizedRows;
+ return InlineDataSource.fromIterable(queryRows, descriptor.getRowSignature());
+ }
+
+ private ScanQuery makeNodeQuery(
+ final SystemTableDataSource dataSource,
+ final SystemTableDescriptor descriptor,
+ final Query<?> owningQuery
+ )
+ {
+ final Map<String, Object> nodeContext = new LinkedHashMap<>(owningQuery.getContext());
+ final DimFilter nodeFilter = nodeFilter(dataSource, owningQuery);
+ nodeContext.put(QueryContextParameters.QUERY_FAIL_TIME.getName(), getNodeFailTime(owningQuery));
+ // The node endpoint always returns plain ScanResultValue objects. bySegment belongs to the user-facing query and
+ // would make DirectDruidClient deserialize this private transport using an incompatible result type.
+ nodeContext.put(QueryContexts.BY_SEGMENT_KEY, false);
+ return Druids.newScanQueryBuilder()
+ .dataSource(dataSource)
+ .eternityInterval()
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_COMPACTED_LIST)
+ .limit(Long.MAX_VALUE)
+ .filters(nodeFilter)
+ .virtualColumns(nodeVirtualColumns(dataSource, owningQuery, nodeFilter))
+ .columns(descriptor.getRowSignature())
Review Comment:
This should include some facility for limiting what columns are returned, rather than taking all columns in the table. If someone does `select server from server_properties` then only `server` should be sent back from the other nodes. (It's ok to do this in a follow-up.)
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryHandler.java:
##########
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.inject.Inject;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.BadQueryContextException;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryEngine;
+import org.apache.druid.segment.InlineSegmentWrangler;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.system.table.SystemTableDataProvider;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTablePushdownFilter;
+
+import java.util.Map;
+
+/** Resolves one node-local system table and returns its rows through the standard native Scan stack. */
+public class SystemTableQueryHandler implements DataSourceQueryHandler
+{
+ private final Map<String, SystemTableDataProvider> dataSuppliers;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final ScanQueryEngine scanQueryEngine;
+ private final AuthorizerMapper authorizerMapper;
+
+ @Inject
+ public SystemTableQueryHandler(
+ final Map<String, SystemTableDataProvider> dataSuppliers,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final ScanQueryEngine scanQueryEngine,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.dataSuppliers = dataSuppliers;
+ this.tableDescriptors = tableDescriptors;
+ this.scanQueryEngine = scanQueryEngine;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult requestAuthenticationResult,
+ final boolean executeLocally
+ )
+ {
+ if (!(query instanceof ScanQuery)) {
+ throw new BadQueryContextException(
+ "Local system-table execution requires a Scan query with a SystemTableDataSource at the query root"
+ );
+ }
+ if (!(query.getDataSource() instanceof SystemTableDataSource dataSource)) {
+ throw new BadQueryContextException(
+ "Local system-table execution requires a Scan query with a SystemTableDataSource at the query root"
+ );
+ }
+ final SystemTableDataProvider dataSupplier = dataSuppliers.get(dataSource.getTable());
+ if (dataSupplier == null) {
+ throw new ISE("System table[%s] is not served by this node", dataSource.getTable());
Review Comment:
`DruidException` is preferred. I mention it only here but it applies in various other locations.
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
Review Comment:
Hmm, instead of "use the SQL-layer execution path" how about "use a SQL interpreter on the Broker"? I'm not sure if it's better, tbh. I just thought "SQL-layer execution path" wasn't clear.
##########
processing/src/main/java/org/apache/druid/query/SystemTableDataSource.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Identifies a Druid system table whose rows are supplied by a service rather than by segments.
+ */
+public class SystemTableDataSource extends LeafDataSource
+{
+ public static final String NODE_QUERY_ID_PREFIX = "native-system-node-";
+
+ private final String table;
+
+ @JsonCreator
+ public SystemTableDataSource(@JsonProperty("table") final String table)
+ {
+ this.table = Preconditions.checkNotNull(table, "table");
+ }
+
+ @JsonProperty
+ public String getTable()
+ {
+ return table;
+ }
+
+ @Override
+ public Set<String> getTableNames()
+ {
+ // QueryScheduler uses table names as cancellation authorization resources. Namespace the synthetic datasource
+ // name so a system table cannot collide with a regular Druid datasource that has the same unqualified name.
Review Comment:
This seems sketchy. Regular datasources can have periods in them, so even after adding `sys.`, it could still collide.
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT server) FROM sys.server_properties",
+ "context": {
+ "allowNativeFunctionsForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET allowNativeFunctionsForSystemTables = 'true';
+SELECT COUNT(DISTINCT server) FROM sys.server_properties;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `allowNativeFunctionsForSystemTables` is `true`.
+
+The parameter defaults to `false`. During a rolling upgrade, leave it disabled until the Broker and all nodes
+that serve the native system tables have been upgraded. After the upgrade, you can enable it by query or set
+`druid.query.default.context.allowNativeFunctionsForSystemTables=true` on Brokers as the cluster-wide default. For more
+information, see [SQL query context](sql-query-context.md).
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
Review Comment:
This paragraph probably isn't necessary, it seems like it's about an internal detail, not really about SQL.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4361445182
##########
server/src/main/java/org/apache/druid/server/system/table/SystemTableDataProvider.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.table;
+
+import jakarta.validation.constraints.NotNull;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.server.security.AuthenticationResult;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Supplies storage-prefiltered rows of one native system table. Authorization isn't applied here: the table's
+ * {@link SystemTableDescriptor#getRowAuthorizer()} filters these rows wherever they are consumed.
+ * The implementation is deployed in related service.
+ * For example, the data provider of sys.tasks is deployed in overlord module
+ * */
+public interface SystemTableDataProvider
+{
+ default List<SystemTablePushdownFilter> getPushdownFilters()
+ {
+ return Collections.emptyList();
+ }
+
+ Iterable<Object[]> getRows(
+ @NotNull List<DimFilter> filters,
+ AuthenticationResult internalAuthenticationResult
Review Comment:
## CodeQL / Useless parameter
The parameter 'internalAuthenticationResult' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/12114)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3964483517
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,58 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
:::
+### Native query execution
+
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`useNativeQueryForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT task_id) FROM sys.tasks",
+ "context": {
+ "useNativeQueryForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET useNativeQueryForSystemTables = 'true';
+SELECT COUNT(DISTINCT datasource) FROM sys.tasks;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.tasks`](#tasks-table)|The Overlord that owns task state. Supported filters are pushed into task storage when the configured task storage implementation supports filter pushdown.|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `useNativeQueryForSystemTables` is `true`.
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
+system-table Scan against only the contacted node, set the HTTP header
+`X-Druid-Native-Query-Route: local`. Local execution uses the authenticated request identity and applies the table's
+authorization rules. The Broker uses the same header for remote node fan-out requests. If the Broker itself is
+one of the selected nodes, it executes that node Scan in-process without an HTTP request. The header
+controls routing and doesn't grant additional permissions.
+
+The parameter defaults to `false`. During a rolling upgrade, leave it disabled until the Broker and all nodes
+that serve the native system tables have been upgraded. After the upgrade, you can enable it by query or set
+`druid.query.default.context.useNativeQueryForSystemTables=true` on Brokers as the cluster-wide default. For more
+information, see [SQL query context](sql-query-context.md).
Review Comment:
Addressed on this PR branch: the default and rolling-upgrade guidance now follows the native-planning/fallback explanation, before the route-header paragraph.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNestedTaskAggregation(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*) "
+ + "FROM ("
+ + " SELECT task_id, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY task_id"
+ + ") "
+ + "WHERE task_count > 0",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ /**
+ * Verifies that an outer aggregation can consume grouped rows from a native system-table subquery.
+ *
+ * <pre>{@code
+ * SELECT SUM(task_count)
+ * FROM
+ * (
+ * SELECT
+ * datasource,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY datasource
+ * )
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testOuterAggregationOverTaskSubquery(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT SUM(task_count) "
+ + "FROM ("
+ + " SELECT datasource, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY datasource"
+ + ")",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testInAndOrFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id IN ('native_sys_mvp_a_0', 'native_sys_mvp_b_2') "
+ + "OR task_id = 'native_sys_mvp_missing'",
+ Set.of("native_sys_mvp_a_0", "native_sys_mvp_b_2"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testRangeAndLikeFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id >= 'native_sys_mvp_b_0' "
+ + "AND task_id < 'native_sys_mvp_b_2' "
+ + "AND task_id LIKE 'native_sys_mvp_b_%'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ /**
+ * Verifies that expression virtual columns required by a pushed node filter are preserved.
+ *
+ * <pre>{@code
+ * SELECT task_id
+ * FROM sys.tasks
+ * WHERE UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testExpressionFilterPreservesVirtualColumn(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'",
+ Set.of("native_sys_mvp_a_0"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNegatedLikeFilter(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "datasource = 'native_sys_b' AND task_id NOT LIKE '%b_2'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testStatusLikeFilterRemainsForResidualEvaluation(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id LIKE 'native_sys_mvp_%' AND status LIKE '%'",
+ Set.of(
+ "native_sys_mvp_a_0",
+ "native_sys_mvp_a_1",
+ "native_sys_mvp_b_0",
+ "native_sys_mvp_b_1",
+ "native_sys_mvp_b_2"
+ ),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testWebConsoleTasksQueryUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "WITH tasks AS (SELECT\n"
+ + " \"task_id\", \"group_id\", \"type\", \"datasource\", \"created_time\", \"location\", "
+ + "\"duration\", \"error_msg\",\n"
+ + " CASE WHEN \"error_msg\" IN ('Shutdown request from user', "
+ + "'Canceled: Query canceled by user or by task shutdown.') THEN 'CANCELED' "
+ + "WHEN \"status\" = 'RUNNING' THEN \"runner_status\" ELSE \"status\" END AS \"status\"\n"
Review Comment:
Added a 100,000-task JMH benchmark in FrankChen021/druid#198. The full-list case uses the Web Console query as the representative workload, and the PR description highlights the measured Bindable-versus-native results.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNestedTaskAggregation(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*) "
+ + "FROM ("
+ + " SELECT task_id, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY task_id"
+ + ") "
+ + "WHERE task_count > 0",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ /**
+ * Verifies that an outer aggregation can consume grouped rows from a native system-table subquery.
+ *
+ * <pre>{@code
+ * SELECT SUM(task_count)
+ * FROM
+ * (
+ * SELECT
+ * datasource,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY datasource
+ * )
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testOuterAggregationOverTaskSubquery(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT SUM(task_count) "
+ + "FROM ("
+ + " SELECT datasource, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY datasource"
+ + ")",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testInAndOrFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id IN ('native_sys_mvp_a_0', 'native_sys_mvp_b_2') "
+ + "OR task_id = 'native_sys_mvp_missing'",
+ Set.of("native_sys_mvp_a_0", "native_sys_mvp_b_2"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testRangeAndLikeFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id >= 'native_sys_mvp_b_0' "
+ + "AND task_id < 'native_sys_mvp_b_2' "
+ + "AND task_id LIKE 'native_sys_mvp_b_%'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ /**
+ * Verifies that expression virtual columns required by a pushed node filter are preserved.
+ *
+ * <pre>{@code
+ * SELECT task_id
+ * FROM sys.tasks
+ * WHERE UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'
+ * }</pre>
+ */
Review Comment:
Addressed in the stacked PR: the repeated SQL Javadocs in this test were removed. See FrankChen021/druid#198.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
Review Comment:
Addressed in the stacked PR: the redundant test Javadocs were removed. See FrankChen021/druid#198.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3961974547
##########
services/src/main/java/org/apache/druid/cli/CliHistorical.java:
##########
@@ -92,30 +86,22 @@ protected List<? extends Module> getModules()
{
return ImmutableList.of(
new DruidProcessingModule(),
- new QueryableModule(),
- new QueryRunnerFactoryModule(),
- new SegmentWranglerModule(),
- new JoinableFactoryModule(),
Review Comment:
Are these not needed? Wondering if this is general cleanup, outside the scope of this PR
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
Review Comment:
These javadocs here seem like a repeat of the SQL in the test and the test name. Can be removed.
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,58 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
:::
+### Native query execution
+
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`useNativeQueryForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT task_id) FROM sys.tasks",
+ "context": {
+ "useNativeQueryForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET useNativeQueryForSystemTables = 'true';
+SELECT COUNT(DISTINCT datasource) FROM sys.tasks;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.tasks`](#tasks-table)|The Overlord that owns task state. Supported filters are pushed into task storage when the configured task storage implementation supports filter pushdown.|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `useNativeQueryForSystemTables` is `true`.
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
+system-table Scan against only the contacted node, set the HTTP header
+`X-Druid-Native-Query-Route: local`. Local execution uses the authenticated request identity and applies the table's
+authorization rules. The Broker uses the same header for remote node fan-out requests. If the Broker itself is
+one of the selected nodes, it executes that node Scan in-process without an HTTP request. The header
+controls routing and doesn't grant additional permissions.
Review Comment:
Hmm, is this header required only for the `sys.server_properties` table? I wonder if there's a better way to achieve this without passing in additional headers.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNestedTaskAggregation(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*) "
+ + "FROM ("
+ + " SELECT task_id, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY task_id"
+ + ") "
+ + "WHERE task_count > 0",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ /**
+ * Verifies that an outer aggregation can consume grouped rows from a native system-table subquery.
+ *
+ * <pre>{@code
+ * SELECT SUM(task_count)
+ * FROM
+ * (
+ * SELECT
+ * datasource,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY datasource
+ * )
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testOuterAggregationOverTaskSubquery(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT SUM(task_count) "
+ + "FROM ("
+ + " SELECT datasource, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY datasource"
+ + ")",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testInAndOrFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id IN ('native_sys_mvp_a_0', 'native_sys_mvp_b_2') "
+ + "OR task_id = 'native_sys_mvp_missing'",
+ Set.of("native_sys_mvp_a_0", "native_sys_mvp_b_2"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testRangeAndLikeFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id >= 'native_sys_mvp_b_0' "
+ + "AND task_id < 'native_sys_mvp_b_2' "
+ + "AND task_id LIKE 'native_sys_mvp_b_%'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ /**
+ * Verifies that expression virtual columns required by a pushed node filter are preserved.
+ *
+ * <pre>{@code
+ * SELECT task_id
+ * FROM sys.tasks
+ * WHERE UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'
+ * }</pre>
+ */
Review Comment:
Same - the javadocs seem redundant here and a few other tests below
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,58 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
:::
+### Native query execution
+
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`useNativeQueryForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT task_id) FROM sys.tasks",
+ "context": {
+ "useNativeQueryForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET useNativeQueryForSystemTables = 'true';
+SELECT COUNT(DISTINCT datasource) FROM sys.tasks;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.tasks`](#tasks-table)|The Overlord that owns task state. Supported filters are pushed into task storage when the configured task storage implementation supports filter pushdown.|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `useNativeQueryForSystemTables` is `true`.
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
+system-table Scan against only the contacted node, set the HTTP header
+`X-Druid-Native-Query-Route: local`. Local execution uses the authenticated request identity and applies the table's
+authorization rules. The Broker uses the same header for remote node fan-out requests. If the Broker itself is
+one of the selected nodes, it executes that node Scan in-process without an HTTP request. The header
+controls routing and doesn't grant additional permissions.
+
+The parameter defaults to `false`. During a rolling upgrade, leave it disabled until the Broker and all nodes
+that serve the native system tables have been upgraded. After the upgrade, you can enable it by query or set
+`druid.query.default.context.useNativeQueryForSystemTables=true` on Brokers as the cluster-wide default. For more
+information, see [SQL query context](sql-query-context.md).
Review Comment:
I believe this is a continuation of two paragraphs above:
>A system table that doesn't advertise native query support continues to
use its existing SQL-layer execution path, even when `useNativeQueryForSystemTables` is `true`.
Consider moving this above the paragraph about the additional header.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNestedTaskAggregation(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*) "
+ + "FROM ("
+ + " SELECT task_id, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY task_id"
+ + ") "
+ + "WHERE task_count > 0",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ /**
+ * Verifies that an outer aggregation can consume grouped rows from a native system-table subquery.
+ *
+ * <pre>{@code
+ * SELECT SUM(task_count)
+ * FROM
+ * (
+ * SELECT
+ * datasource,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY datasource
+ * )
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testOuterAggregationOverTaskSubquery(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT SUM(task_count) "
+ + "FROM ("
+ + " SELECT datasource, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY datasource"
+ + ")",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testInAndOrFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id IN ('native_sys_mvp_a_0', 'native_sys_mvp_b_2') "
+ + "OR task_id = 'native_sys_mvp_missing'",
+ Set.of("native_sys_mvp_a_0", "native_sys_mvp_b_2"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testRangeAndLikeFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id >= 'native_sys_mvp_b_0' "
+ + "AND task_id < 'native_sys_mvp_b_2' "
+ + "AND task_id LIKE 'native_sys_mvp_b_%'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ /**
+ * Verifies that expression virtual columns required by a pushed node filter are preserved.
+ *
+ * <pre>{@code
+ * SELECT task_id
+ * FROM sys.tasks
+ * WHERE UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testExpressionFilterPreservesVirtualColumn(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'",
+ Set.of("native_sys_mvp_a_0"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNegatedLikeFilter(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "datasource = 'native_sys_b' AND task_id NOT LIKE '%b_2'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testStatusLikeFilterRemainsForResidualEvaluation(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id LIKE 'native_sys_mvp_%' AND status LIKE '%'",
+ Set.of(
+ "native_sys_mvp_a_0",
+ "native_sys_mvp_a_1",
+ "native_sys_mvp_b_0",
+ "native_sys_mvp_b_1",
+ "native_sys_mvp_b_2"
+ ),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testWebConsoleTasksQueryUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "WITH tasks AS (SELECT\n"
+ + " \"task_id\", \"group_id\", \"type\", \"datasource\", \"created_time\", \"location\", "
+ + "\"duration\", \"error_msg\",\n"
+ + " CASE WHEN \"error_msg\" IN ('Shutdown request from user', "
+ + "'Canceled: Query canceled by user or by task shutdown.') THEN 'CANCELED' "
+ + "WHEN \"status\" = 'RUNNING' THEN \"runner_status\" ELSE \"status\" END AS \"status\"\n"
Review Comment:
I think this web-console query would be a good candidate to benchmark
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3964277056
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,58 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
:::
+### Native query execution
+
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`useNativeQueryForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT task_id) FROM sys.tasks",
+ "context": {
+ "useNativeQueryForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET useNativeQueryForSystemTables = 'true';
+SELECT COUNT(DISTINCT datasource) FROM sys.tasks;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.tasks`](#tasks-table)|The Overlord that owns task state. Supported filters are pushed into task storage when the configured task storage implementation supports filter pushdown.|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `useNativeQueryForSystemTables` is `true`.
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
+system-table Scan against only the contacted node, set the HTTP header
+`X-Druid-Native-Query-Route: local`. Local execution uses the authenticated request identity and applies the table's
+authorization rules. The Broker uses the same header for remote node fan-out requests. If the Broker itself is
+one of the selected nodes, it executes that node Scan in-process without an HTTP request. The header
+controls routing and doesn't grant additional permissions.
Review Comment:
it's needed as the broker will send native query to router or itself to retrieve data. for router, we need to tell whether it acts as a proxy or execute the native query.
another way is by using query context, but I don't think the query context is better way because this value here is just an internal flag, not user-facing context parameter.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4237747547
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryClient.java:
##########
@@ -0,0 +1,899 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import io.netty.channel.ChannelException;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.client.DirectDruidClientFactory;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.LazySequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.Yielders;
+import org.apache.druid.query.BaseQuery;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.QueryDataSource;
+import org.apache.druid.query.QueryException;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.QueryScheduler;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.system.SystemTableNotLeaderException;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.utils.CloseableUtils;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+/**
+ * Recursively replaces native system-table datasources with authorized inline rows fetched from nodes, then
+ * executes the rebuilt query through the Broker's normal native-query walker.
+ */
+public class SystemTableQueryClient implements DataSourceQueryHandler
+{
+ private static final String SYSTEM_TABLE_TIER = "_system";
+
+ private final SystemTableNodeLocator nodeLocator;
+ private final DirectDruidClientFactory directDruidClientFactory;
+ private final QueryScheduler queryScheduler;
+ private final QuerySegmentWalker querySegmentWalker;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final AuthorizerMapper authorizerMapper;
+ private final SystemTableQueryHandler localQueryHandler;
+ private final AuthenticationResult escalatedAuthenticationResult;
+ private final DruidNode selfNode;
+
+ @Inject
+ public SystemTableQueryClient(
+ final SystemTableNodeLocator nodeLocator,
+ final DirectDruidClientFactory directDruidClientFactory,
+ final QueryScheduler queryScheduler,
+ final QuerySegmentWalker querySegmentWalker,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final AuthorizerMapper authorizerMapper,
+ final SystemTableQueryHandler localQueryHandler,
+ final Escalator escalator,
+ @Self final DruidNode selfNode
+ )
+ {
+ this.nodeLocator = nodeLocator;
+ this.directDruidClientFactory = directDruidClientFactory;
+ this.queryScheduler = queryScheduler;
+ this.querySegmentWalker = querySegmentWalker;
+ this.tableDescriptors = tableDescriptors;
+ this.authorizerMapper = authorizerMapper;
+ this.localQueryHandler = localQueryHandler;
+ this.escalatedAuthenticationResult = escalator.createEscalatedAuthenticationResult();
+ this.selfNode = selfNode;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult authenticationResult,
+ final boolean executeLocally
+ )
+ {
+ return (queryPlus, responseContext) -> {
+ final NodeSequencesCloser nodeSequencesCloser = new NodeSequencesCloser(queryScheduler);
+ try {
+ final DataSource resolvedDataSource = resolveDataSource(
+ query.getDataSource(),
+ query,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ final Query<T> resolvedQuery = query.withDataSource(resolvedDataSource).withOverriddenContext(
+ Map.of(QueryContexts.QUERY_RESOURCE_ID, UUID.randomUUID().toString())
+ );
+ final QueryRunner<T> queryRunner = querySegmentWalker.getQueryRunnerForIntervals(
+ resolvedQuery,
+ resolvedQuery.getIntervals()
+ );
+ return Sequences.withBaggage(
+ queryRunner.run(queryPlus.withQuery(resolvedQuery), responseContext),
+ nodeSequencesCloser
+ );
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, nodeSequencesCloser);
+ }
+ };
+ }
+
+ private DataSource resolveDataSource(
+ final DataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ if (dataSource instanceof SystemTableDataSource systemTableDataSource) {
+ return resolveSystemTableDataSource(
+ systemTableDataSource,
+ owningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ );
+ }
+
+ final Query<?> childOwningQuery = dataSource instanceof QueryDataSource queryDataSource
+ ? queryDataSource.getQuery()
+ : owningQuery;
+ final List<DataSource> resolvedChildren = new ArrayList<>();
+ for (final DataSource child : dataSource.getChildren()) {
+ resolvedChildren.add(
+ resolveDataSource(
+ child,
+ childOwningQuery,
+ authenticationResult,
+ responseContext,
+ nodeSequencesCloser
+ )
+ );
+ }
+ return dataSource.withChildren(resolvedChildren);
+ }
+
+ private InlineDataSource resolveSystemTableDataSource(
+ final SystemTableDataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final ResponseContext responseContext,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final SystemTableDescriptor descriptor = tableDescriptors.get(dataSource.getTable());
+ if (descriptor == null) {
+ throw new ISE("No routing descriptor is registered for system table[%s]", dataSource.getTable());
+ }
+
+ final ScanQuery nodeQuery = makeNodeQuery(dataSource, descriptor, owningQuery);
+ final List<QueryRunner<ScanResultValue>> nodeRunners = makeNodeRunners(
+ nodeQuery,
+ descriptor,
+ nodeSequencesCloser
+ );
+ if (nodeRunners.isEmpty() && !descriptor.isEmptyDiscoveryAllowed()) {
+ throw new ISE("No node is available to serve system table[%s]", dataSource.getTable());
+ }
+
+ /*
+ * Keep the scan result transport lazy. The node response is a Sequence, so this lets the local query
+ * runner consume and process rows while the HTTP response is still arriving instead of materializing every
+ * node row in two Broker-side lists. The scan transport is intentionally kept behind this adapter so a
+ * future node-side aggregation mode can use a different result adapter without changing routing or
+ * authorization orchestration here.
+ */
+ final NodeSequenceCloser nodeSequenceCloser = new NodeSequenceCloser();
+ nodeSequencesCloser.add(nodeSequenceCloser);
+ final Iterable<Object[]> nodeRows = scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ responseContext,
+ nodeSequenceCloser
+ );
+ final Iterable<Object[]> authorizedRows = descriptor.getRowAuthorizer().filterAuthorizedRows(
+ nodeRows,
+ authenticationResult,
+ authorizerMapper
+ );
+ final Iterable<Object[]> queryRows = requiresMaterializedRows(owningQuery)
+ ? materializeRows(authorizedRows)
+ : authorizedRows;
+ return InlineDataSource.fromIterable(queryRows, descriptor.getRowSignature());
+ }
+
+ private ScanQuery makeNodeQuery(
+ final SystemTableDataSource dataSource,
+ final SystemTableDescriptor descriptor,
+ final Query<?> owningQuery
+ )
+ {
+ final Map<String, Object> nodeContext = new LinkedHashMap<>(owningQuery.getContext());
+ final DimFilter nodeFilter = nodeFilter(dataSource, owningQuery);
+ nodeContext.put(DirectDruidClient.QUERY_FAIL_TIME, getNodeFailTime(owningQuery));
+ // The node endpoint always returns plain ScanResultValue objects. bySegment belongs to the user-facing query and
+ // would make DirectDruidClient deserialize this private transport using an incompatible result type.
+ nodeContext.put(QueryContexts.BY_SEGMENT_KEY, false);
+ return Druids.newScanQueryBuilder()
+ .dataSource(dataSource)
+ .eternityInterval()
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_COMPACTED_LIST)
+ .limit(Long.MAX_VALUE)
+ .filters(nodeFilter)
+ .virtualColumns(nodeVirtualColumns(dataSource, owningQuery, nodeFilter))
+ .columns(descriptor.getRowSignature())
+ .context(nodeContext)
+ .build();
+ }
+
+ private List<QueryRunner<ScanResultValue>> makeNodeRunners(
+ final ScanQuery nodeQuery,
+ final SystemTableDescriptor descriptor,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final List<QueryRunner<ScanResultValue>> nodeRunners = new ArrayList<>();
+ for (final SystemTableNode node : nodeLocator.locate(descriptor, nodeQuery)) {
+ nodeRunners.add(
+ (queryPlus, responseContext) -> recoverNodeFailure(
+ () -> runNodeQuery(nodeQuery, node, nodeSequencesCloser, queryPlus, responseContext),
+ node,
+ descriptor,
+ descriptor.getRoutingMode() == SystemTableRoutingMode.LEADER_ONLY
+ ? () -> {
+ final SystemTableNode currentLeader = Iterables.getOnlyElement(
+ nodeLocator.locate(descriptor, nodeQuery)
+ );
+ return new NodeQuerySequence(
+ currentLeader,
+ runNodeQuery(nodeQuery, currentLeader, nodeSequencesCloser, queryPlus, responseContext)
+ );
+ }
+ : null
+ )
+ );
+ }
+ return nodeRunners;
+ }
+
+ private Sequence<ScanResultValue> runNodeQuery(
+ final ScanQuery nodeQuery,
+ final SystemTableNode node,
+ final NodeSequencesCloser nodeSequencesCloser,
+ final QueryPlus<ScanResultValue> queryPlus,
+ final ResponseContext responseContext
+ )
+ {
+ final String nodeResourceId = UUID.randomUUID().toString();
+ final String nodeQueryId = SystemTableDataSource.NODE_QUERY_ID_PREFIX + UUID.randomUUID();
+ final ScanQuery subNativeQuery = nodeQuery.withOverriddenContext(
+ Map.of(
+ BaseQuery.QUERY_ID,
+ nodeQueryId,
+ QueryContexts.QUERY_RESOURCE_ID,
+ nodeResourceId
+ )
+ );
+ final QueryRunner<ScanResultValue> nodeRunner;
+ if (SystemTableNodeLocator.sameServer(
+ selfNode.getUriToUse(),
+ node.getDiscoveryNode().getDruidNode().getUriToUse()
+ )) {
+ // Invoke the raw local handler rather than the Broker handler. This avoids an HTTP request back to this
+ // Broker and, more importantly, prevents the local request from entering node fanout recursively.
+ nodeRunner = localQueryHandler.createRunner(subNativeQuery, escalatedAuthenticationResult, true);
+ } else {
+ nodeRunner = directDruidClientFactory.makeDirectClient(toDruidServer(node.getDiscoveryNode()));
+ nodeSequencesCloser.addQueryId(nodeQueryId);
+ }
+ return nodeRunner.run(queryPlus.withQuery(subNativeQuery), responseContext);
+ }
+
+ /**
+ * Window operators consume a {@code RowsAndColumns} view of the segment. The current row-based segment can only
+ * provide that view when its rows are list-backed, so preserve the existing materialized path for those queries.
+ */
+ private static boolean requiresMaterializedRows(final Query<?> query)
+ {
+ return query instanceof WindowOperatorQuery;
+ }
+
+ /**
+ * Eagerly consumes the lazy, single-pass node response into an {@link ArrayList} without changing its rows.
+ *
+ * <p>This representation change is required by {@link org.apache.druid.segment.InlineSegmentWrangler}: an inline
+ * datasource backed specifically by an {@code ArrayList} becomes an {@link org.apache.druid.segment.ArrayListSegment},
+ * which can provide the {@code RowsAndColumns} view required by window operators. Other query types retain the lazy
+ * iterable so the Broker can process rows while node responses are still arriving.</p>
+ */
+ private static Iterable<Object[]> materializeRows(final Iterable<Object[]> rows)
+ {
+ final List<Object[]> materializedRows = new ArrayList<>();
+ rows.forEach(materializedRows::add);
+ return materializedRows;
+ }
+
+ /**
+ * Creates a lazy, single-pass view of the rows returned by the node scan runners.
+ *
+ * <p>The view is single-pass because a node response is backed by a live HTTP stream. The normal native query
+ * path consumes an {@link InlineDataSource} once per query. Re-iterating would otherwise issue duplicate node
+ * requests or retain all rows to support replay.</p>
+ */
+ static Iterable<Object[]> scanNodeRows(
+ final List<QueryRunner<ScanResultValue>> nodeRunners,
+ final Query<ScanResultValue> nodeQuery,
+ final ResponseContext responseContext
+ )
+ {
+ return scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ responseContext,
+ new NodeSequenceCloser()
+ );
+ }
+
+ private static Iterable<Object[]> scanNodeRows(
+ final List<QueryRunner<ScanResultValue>> nodeRunners,
+ final Query<ScanResultValue> nodeQuery,
+ final ResponseContext responseContext,
+ final NodeSequenceCloser nodeSequenceCloser
+ )
+ {
+ final Sequence<Object[]> rows = new LazySequence<>(() -> {
+ final List<Sequence<ScanResultValue>> nodeSequences = new ArrayList<>();
+ for (final QueryRunner<ScanResultValue> nodeRunner : nodeRunners) {
+ // DirectDruidClient.run starts its asynchronous HTTP request. Start every request before waiting for any
+ // response so slow nodes do not serialize the entire system-table query.
+ nodeSequences.add(nodeRunner.run(QueryPlus.wrap(nodeQuery), responseContext));
+ }
+ // Node scans have no cross-node ordering. Consume them in discovery order so a ready node can yield rows without
+ // MergeSequence first initializing every other node sequence. The HTTP requests were all started above.
+ return Sequences.concat(nodeSequences).flatMap(SystemTableQueryClient::rowsFromScanResult);
+ });
+ return sequenceAsIterable(rows, nodeSequenceCloser);
+ }
+
+ private static Sequence<ScanResultValue> recoverNodeFailure(
+ final Supplier<Sequence<ScanResultValue>> sequenceSupplier,
+ final SystemTableNode node,
+ final SystemTableDescriptor descriptor,
+ @Nullable final Supplier<NodeQuerySequence> leaderRetrySupplier
+ )
+ {
+ final Sequence<ScanResultValue> sequence;
+ try {
+ sequence = sequenceSupplier.get();
+ }
+ catch (Exception failure) {
+ if (leaderRetrySupplier != null && isLeaderChangeFailure(failure)) {
+ try {
+ final NodeQuerySequence retry = leaderRetrySupplier.get();
+ return recoveringSequence(retry, descriptor, null);
+ }
+ catch (Exception retryFailure) {
+ return failureSequence(retryFailure, node, descriptor);
+ }
+ }
+ return failureSequence(failure, node, descriptor);
+ }
+ return recoveringSequence(new NodeQuerySequence(node, sequence), descriptor, leaderRetrySupplier);
+ }
+
+ private static Sequence<ScanResultValue> recoveringSequence(
+ final NodeQuerySequence nodeQuerySequence,
+ final SystemTableDescriptor descriptor,
+ @Nullable final Supplier<NodeQuerySequence> leaderRetrySupplier
+ )
+ {
+ return new BaseSequence<>(
+ new BaseSequence.IteratorMaker<ScanResultValue, RecoveringNodeIterator>()
+ {
+ @Override
+ public RecoveringNodeIterator make()
+ {
+ return new RecoveringNodeIterator(nodeQuerySequence, descriptor, leaderRetrySupplier);
+ }
+
+ @Override
+ public void cleanup(final RecoveringNodeIterator iterator)
+ {
+ CloseableUtils.closeAndWrapExceptions(iterator);
+ }
+ }
+ );
+ }
+
+ private static Sequence<ScanResultValue> failureSequence(
+ final Exception failure,
+ final SystemTableNode node,
+ final SystemTableDescriptor descriptor
+ )
+ {
+ if (!isNodeAvailabilityFailure(failure)) {
+ throw propagate(failure);
+ }
+ return descriptor.getNodeFailureRow(
+ node.getDiscoveryNode().getDruidNode(),
+ node.getNodeRoles(),
+ failure
+ ).map(
+ row -> Sequences.simple(
+ List.of(
+ new ScanResultValue(
+ null,
+ descriptor.getRowSignature().getColumnNames(),
+ List.of((Object) row)
+ )
+ )
+ )
+ ).orElseThrow(() -> QueryInterruptedException.wrapIfNeeded(failure));
+ }
+
+ private static boolean isNodeAvailabilityFailure(final Throwable failure)
+ {
+ Throwable cause = failure;
+ while (cause != null) {
+ if (cause instanceof SocketException
+ || cause instanceof UnknownHostException
+ || cause instanceof EOFException
+ || cause instanceof ClosedChannelException
+ || cause instanceof ChannelException) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
+ }
+
+ private static boolean isLeaderChangeFailure(final Throwable failure)
+ {
+ Throwable cause = failure;
+ while (cause != null) {
+ if (cause instanceof SystemTableNotLeaderException
+ || cause instanceof QueryException
+ && SystemTableNotLeaderException.class.getName().equals(((QueryException) cause).getErrorClass())) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
+ }
+
+ private static RuntimeException propagate(final Exception failure)
+ {
+ if (failure instanceof RuntimeException runtimeException) {
+ return runtimeException;
+ } else if (failure instanceof java.util.concurrent.TimeoutException) {
+ return new QueryTimeoutException(failure.getMessage());
+ } else {
+ if (failure instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ return QueryInterruptedException.wrapIfNeeded(failure);
+ }
+ }
+
+ private static Sequence<Object[]> rowsFromScanResult(final ScanResultValue scanResult)
+ {
+ return Sequences.simple((List<?>) scanResult.getEvents())
+ .map(event -> event instanceof Object[] ? (Object[]) event : ((List<?>) event).toArray());
+ }
+
+ private static long getNodeFailTime(final Query<?> query)
+ {
+ final long existingFailTime = query.context().getLong(DirectDruidClient.QUERY_FAIL_TIME, -1L);
+ if (existingFailTime > 0) {
+ return existingFailTime;
+ } else if (query.context().hasTimeout()) {
+ return System.currentTimeMillis() + query.context().getTimeout();
+ } else {
+ return JodaUtils.MAX_INSTANT;
+ }
+ }
+
+ /**
+ * Adapts a Druid {@link Sequence} to the lazy {@link Iterable} accepted by {@link InlineDataSource}.
+ *
+ * <p>The iterator owns the sequence yielder and closes it when the sequence is exhausted. This is important for
+ * direct node clients because closing the yielder closes the underlying response stream and allows an early
+ * terminating Broker query to cancel the node request.</p>
+ */
+ private static <T> Iterable<T> sequenceAsIterable(
+ final Sequence<T> sequence,
+ final NodeSequenceCloser nodeSequenceCloser
+ )
+ {
+ return new Iterable<>()
+ {
+ private boolean iterated;
+
+ @Override
+ public Iterator<T> iterator()
+ {
+ if (iterated) {
+ throw new ISE("A native system table result sequence can only be consumed once");
+ }
+ iterated = true;
+
+ return new Iterator<>()
+ {
+ @Nullable
+ private Yielder<T> yielder = Yielders.each(sequence);
+ private boolean closed;
+
+ {
+ nodeSequenceCloser.setYielder(yielder);
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (yielder == null) {
+ return false;
+ }
+ if (yielder.isDone()) {
+ close();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public T next()
+ {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final Yielder<T> currentYielder = Preconditions.checkNotNull(yielder);
+ final T value = currentYielder.get();
+ yielder = currentYielder.next(null);
+ nodeSequenceCloser.setYielder(yielder);
+ return value;
+ }
+
+ private void close()
+ {
+ if (!closed) {
+ closed = true;
+ yielder = null;
+ nodeSequenceCloser.close();
+ }
+ }
+ };
+ }
+ };
+ }
+
+ private static class NodeSequenceCloser implements Closeable
+ {
+ @Nullable
+ private Yielder<?> yielder;
+ private boolean closed;
+
+ synchronized void setYielder(final Yielder<?> newYielder)
+ {
+ if (closed) {
+ CloseableUtils.closeAndWrapExceptions(newYielder);
+ } else {
+ yielder = newYielder;
+ }
+ }
+
+ @Override
+ public synchronized void close()
+ {
+ if (!closed) {
+ closed = true;
+ final Yielder<?> currentYielder = yielder;
+ yielder = null;
+ CloseableUtils.closeAndWrapExceptions(currentYielder);
+ }
+ }
+ }
+
+ private static class NodeSequencesCloser implements Closeable
+ {
+ private final List<NodeSequenceCloser> nodeSequenceClosers = new ArrayList<>();
+ private final List<String> nodeQueryIds = new ArrayList<>();
+ private final QueryScheduler queryScheduler;
+ private boolean closed;
+
+ private NodeSequencesCloser(final QueryScheduler queryScheduler)
+ {
+ this.queryScheduler = queryScheduler;
+ }
+
+ synchronized void addQueryId(final String nodeQueryId)
+ {
+ if (closed) {
+ queryScheduler.cancelQuery(nodeQueryId);
+ } else {
+ nodeQueryIds.add(nodeQueryId);
+ }
+ }
+
+ synchronized void add(final NodeSequenceCloser nodeSequenceCloser)
+ {
+ if (closed) {
+ CloseableUtils.closeAndWrapExceptions(nodeSequenceCloser);
+ } else {
+ nodeSequenceClosers.add(nodeSequenceCloser);
+ }
+ }
+
+ @Override
+ public synchronized void close() throws IOException
+ {
+ if (!closed) {
+ closed = true;
+ try {
+ CloseableUtils.closeAll(nodeSequenceClosers);
+ }
+ finally {
+ nodeSequenceClosers.clear();
+ nodeQueryIds.forEach(queryScheduler::cancelQuery);
+ nodeQueryIds.clear();
+ }
+ }
+ }
+ }
+
+ private static class RecoveringNodeIterator implements Iterator<ScanResultValue>, Closeable
+ {
+ private final SystemTableDescriptor descriptor;
+ @Nullable
+ private final Supplier<NodeQuerySequence> leaderRetrySupplier;
+
+ private NodeQuerySequence nodeQuerySequence;
+
+ @Nullable
+ private Yielder<ScanResultValue> yielder;
+ @Nullable
+ private ScanResultValue failureResult;
+ private boolean started;
+ private boolean finished;
+ private boolean retriedLeader;
+ private boolean emittedResult;
+
+ private RecoveringNodeIterator(
+ final NodeQuerySequence nodeQuerySequence,
+ final SystemTableDescriptor descriptor,
+ @Nullable final Supplier<NodeQuerySequence> leaderRetrySupplier
+ )
+ {
+ this.nodeQuerySequence = nodeQuerySequence;
+ this.descriptor = descriptor;
+ this.leaderRetrySupplier = leaderRetrySupplier;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (failureResult != null) {
+ return true;
+ }
+ if (finished) {
+ return false;
+ }
+
+ try {
+ if (!started) {
+ started = true;
+ yielder = Yielders.each(nodeQuerySequence.sequence());
+ }
+ if (Preconditions.checkNotNull(yielder).isDone()) {
+ finished = true;
+ close();
+ return false;
+ }
+ return true;
+ }
+ catch (Exception e) {
+ return recoverOrThrow(e) ? hasNext() : true;
+ }
+ }
+
+ @Override
+ public ScanResultValue next()
+ {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ if (failureResult != null) {
+ final ScanResultValue result = failureResult;
+ failureResult = null;
+ finished = true;
+ return result;
+ }
+
+ final Yielder<ScanResultValue> currentYielder = Preconditions.checkNotNull(yielder);
+ final ScanResultValue result;
+ try {
+ result = currentYielder.get();
+ emittedResult = true;
+ }
+ catch (Exception e) {
+ // Either the leader retry restarted the sequence or a failure row is pending; next() serves both.
+ recoverOrThrow(e);
+ return next();
+ }
+ try {
+ yielder = currentYielder.next(null);
+ }
+ catch (Exception e) {
+ recoverOrThrow(e);
+ }
+ return result;
+ }
+
+ private boolean recoverOrThrow(final Exception originalFailure)
+ {
+ Exception failure = originalFailure;
+ if (!emittedResult && !retriedLeader && leaderRetrySupplier != null && isLeaderChangeFailure(failure)) {
+ retriedLeader = true;
+ try {
+ close();
+ nodeQuerySequence = leaderRetrySupplier.get();
+ started = false;
+ finished = false;
+ return true;
+ }
+ catch (Exception retryFailure) {
+ failure = retryFailure;
+ }
+ }
+ if (!isNodeAvailabilityFailure(failure)) {
+ throw CloseableUtils.closeAndWrapInCatch(failure, this);
+ }
+ final Optional<Object[]> failureRow = descriptor.getNodeFailureRow(
+ nodeQuerySequence.node().getDiscoveryNode().getDruidNode(),
+ nodeQuerySequence.node().getNodeRoles(),
+ failure
+ );
+ if (failureRow.isEmpty()) {
+ throw CloseableUtils.closeAndWrapInCatch(failure, this);
+ }
+ try {
+ close();
+ }
+ catch (RuntimeException closeException) {
+ failure.addSuppressed(closeException);
+ }
+ failureResult = new ScanResultValue(
+ null,
+ descriptor.getRowSignature().getColumnNames(),
+ List.of((Object) failureRow.get())
+ );
+ return false;
+ }
+
+ @Override
+ public void close()
+ {
+ final Yielder<ScanResultValue> currentYielder = yielder;
+ yielder = null;
+ CloseableUtils.closeAndWrapExceptions(currentYielder);
+ }
+ }
+
+ private record NodeQuerySequence(SystemTableNode node, Sequence<ScanResultValue> sequence)
+ {
+ }
+
+ @Nullable
+ private static DimFilter nodeFilter(final SystemTableDataSource dataSource, final Query<?> query)
+ {
+ // Push into the remote scan only when this query directly owns the system table. A join or other composite filter
+ // remains on the resolved Broker query, where the normal native query machinery can apply it with full datasource
+ // semantics; it cannot safely be attributed to an individual remote leaf here.
+ if (!dataSource.equals(query.getDataSource())) {
+ return null;
+ }
+
+ final List<DimFilter> filters = new ArrayList<>();
+ if (query.getFilter() != null) {
+ filters.add(query.getFilter());
+ }
+ if (query instanceof WindowOperatorQuery) {
+ for (final OperatorFactory operator : ((WindowOperatorQuery) query).getLeafOperators()) {
+ if (operator instanceof ScanOperatorFactory && ((ScanOperatorFactory) operator).getFilter() != null) {
+ filters.add(((ScanOperatorFactory) operator).getFilter());
+ }
+ }
+ }
+ if (filters.isEmpty()) {
+ return null;
+ } else if (filters.size() == 1) {
+ return filters.get(0);
+ } else {
+ return new AndDimFilter(filters);
+ }
+ }
+
+ private static VirtualColumns nodeVirtualColumns(
+ final SystemTableDataSource dataSource,
+ final Query<?> query,
+ @Nullable final DimFilter nodeFilter
+ )
+ {
+ if (!dataSource.equals(query.getDataSource()) || nodeFilter == null) {
+ return VirtualColumns.EMPTY;
+ }
+
+ final List<VirtualColumn> virtualColumns = new ArrayList<>(List.of(query.getVirtualColumns().getVirtualColumns()));
+ if (query instanceof WindowOperatorQuery) {
+ for (final OperatorFactory operator : ((WindowOperatorQuery) query).getLeafOperators()) {
+ if (operator instanceof ScanOperatorFactory) {
+ virtualColumns.addAll(List.of(((ScanOperatorFactory) operator).getVirtualColumns().getVirtualColumns()));
Review Comment:
Fixed in `d4ddcf9e31`.
Reproduced first: the new `SystemTableQueryClientTest#testWindowLeafFilterWithoutVirtualColumns` builds the window query the way production does (a filtered Scan subquery with `leafOperators` left for the `WindowOperatorQuery` constructor, which stores null virtual columns) and failed with the reported NPE at `nodeVirtualColumns`. `nodeVirtualColumns` now skips leaves whose virtual columns are null; the test asserts the filter is still pushed to the node scan with no virtual columns.
This path is reachable through a native `windowOperator` query without `leafOperators`; SQL planning does not produce it today. While adding SQL-level coverage (`NativeSysServerPropertiesQueryTest#testNativeWindowFunction`, both planners), a separate DECOUPLED bug surfaced: the window query is planned directly over `filter(systemTable)` and failed with `Segment [filtered->ArrayListSegment] cannot shapeshift`, because `FilteredSegment` never provided a `CloseableShapeshifter`. Fixed generally in `24dee021a5`: `FilteredSegment` now returns `CursorFactoryRowsAndColumns` over its filtered cursor factory, the same way `HashJoinSegment` does, with a new `FilteredSegmentTest`. The window, DECOUPLED and Drill window SQL suites still pass.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4302743515
##########
processing/src/main/java/org/apache/druid/segment/FilteredSegment.java:
##########
@@ -44,6 +45,8 @@ public <T> T as(@Nonnull Class<T> clazz)
{
if (CursorFactory.class.equals(clazz)) {
return (T) new FilteredCursorFactory(delegate.as(CursorFactory.class), filter);
+ } else if (CloseableShapeshifter.class.equals(clazz)) {
+ return (T) new CursorFactoryRowsAndColumns(as(CursorFactory.class));
Review Comment:
[P2] Prevent silent truncation of large filtered windows
**Finding:** This new path routes FilteredDataSource window queries through CursorFactoryRowsAndColumns.materialize(). That loop ignores the boolean returned by FrameWriter.addSelection(); once its fixed 200 MB allocator is exhausted, it still advances the cursor and serializes only the rows already written, allowing an oversized filtered window query to succeed with incomplete results.
**Suggestion:** Check addSelection() before advancing and fail with ResourceLimitExceededException, or use a bounded multi-frame materialization path, with a regression test for an oversized filtered window result.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-5459703544
@clintropolis @gianm please help review the changes. I hope this can be merged as soon as possible so that we can move on migrating other system tables into native execution path, and push forward for #18087 and #19855
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-5595243757
> Thanks for these changes @FrankChen021! The `sys.*` tables have historically been slow and less performant, especially `sys.segments` and `sys.tasks`.
>
> I wonder if we can benchmark `sys.tasks` with thousands of tasks to see how these changes perform at scale with the native support + filter pushdown.
>
> Also, regarding the scope of these changes, I feel it would be helpful to break this into a few more patches for ease of review. Perhaps something like:
>
> * Separate PRs for `sys.tasks` and `sys.server_properties`, isolating the appropriate wirings, filter pushdown mechanisms and tests to make it work for each table.
> * Handle `DECOUPLED` planner support in a separate change, since `COUPLED` is the default planner strategy today (and `DECOUPLED` is currently undocumented).
Thanks for reviewing.
I deliberately chose the sys.tasks and sys.server_properties included in this PR to demonstrate how the native query for system tables are supported.
These two system tables has different fan-out path and push down policies:
1. for sys.server_properties, we need to fan out sub queries to all nodes, while for sys.tasks we need to fan out the query to leader overlord only. This is also the header `X-Druid-Native-Query-Route` is introduced to server the purpose
2. for sys.tasks, it also demonstrate how filters are pushed down, and demonstrate the the other most valueable gain we want to achieve.
If we move any of these system table out of this PR, we can't have a full picture of the change, and understand how the small framework for the system tables work.
As for DECOUPLED mode, only a few files are involved, I think it's better to included in this PR.
To address your concern, I can split the changes for sys.tasks out of this one. But We still need to have a full picture of the core change.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-6365479242
@gianm Thank you for the review. I replied some questions you raised. For others, I will address them in a new commit.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4359767021
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
+
+```json
+{
+ "query": "SELECT COUNT(DISTINCT server) FROM sys.server_properties",
+ "context": {
+ "allowNativeFunctionsForSystemTables": true
+ }
+}
+```
+
+For clients that support setting query context parameters with SQL statements, you can enable native system-table
+execution with `SET`:
+
+```sql
+SET allowNativeFunctionsForSystemTables = 'true';
+SELECT COUNT(DISTINCT server) FROM sys.server_properties;
+```
+
+Native system-table execution is available when the resolved SQL engine is `native`. You don't need to explicitly set
+the `engine` context parameter when `native` is already the default engine. The following tables support native query
+execution:
+
+|Table|Source of rows|
+|-----|--------------|
+|[`sys.server_properties`](#server_properties-table)|The Druid server processes discovered in the cluster. Filters on `server` and `service_name` can avoid reading properties from nodes that don't match.|
+
+After Druid retrieves the system-table rows, the native engine applies the remaining filters, expressions,
+aggregations, sorting, and result processing. A system table that doesn't advertise native query support continues to
+use its existing SQL-layer execution path, even when `allowNativeFunctionsForSystemTables` is `true`.
+
+The parameter defaults to `false`. During a rolling upgrade, leave it disabled until the Broker and all nodes
+that serve the native system tables have been upgraded. After the upgrade, you can enable it by query or set
+`druid.query.default.context.allowNativeFunctionsForSystemTables=true` on Brokers as the cluster-wide default. For more
+information, see [SQL query context](sql-query-context.md).
+
+Native system-table queries sent to the Router use distributed Broker execution by default. To execute a native
+system-table Scan against only the contacted node, set the HTTP header
+`X-Druid-Native-Query-Route: local`. Local execution uses the authenticated request identity and applies the table's
Review Comment:
For this header, it's set by broker, not by user. I think I should remove it from the documentation. Exposing this internal technical details to users does not make any sense.
User can also set this header, in this case, the query at router side is treated as a query execution instead of proxying by using user's identity.
Broker fan out subqueries with this header and escalcated identity to nodes, and each node uses the escalcated identity for authorization.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4361101749
##########
processing/src/main/java/org/apache/druid/query/SystemTableDataSource.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Identifies a Druid system table whose rows are supplied by a service rather than by segments.
+ */
+public class SystemTableDataSource extends LeafDataSource
+{
+ public static final String NODE_QUERY_ID_PREFIX = "native-system-node-";
+
+ private final String table;
+
+ @JsonCreator
+ public SystemTableDataSource(@JsonProperty("table") final String table)
+ {
+ this.table = Preconditions.checkNotNull(table, "table");
+ }
+
+ @JsonProperty
+ public String getTable()
+ {
+ return table;
+ }
+
+ @Override
+ public Set<String> getTableNames()
+ {
+ // QueryScheduler uses table names as cancellation authorization resources. Namespace the synthetic datasource
+ // name so a system table cannot collide with a regular Druid datasource that has the same unqualified name.
Review Comment:
The `sys.` name is no longer used for authorization. Reads and cancellation now use `SYSTEM_TABLE` resources named after the table.
##########
server/src/main/java/org/apache/druid/server/system/table/ServerPropertiesTableDataProvider.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.table;
+
+import com.google.common.collect.Maps;
+import com.google.inject.Inject;
+import org.apache.druid.client.DruidServerConfig;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.EqualityFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizationResult;
+import org.apache.druid.server.security.AuthorizationUtils;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.ForbiddenException;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+/** Native row supplier for {@code sys.server_properties}. */
+public class ServerPropertiesTableDataProvider implements SystemTableDataProvider
+{
+ private static final List<SystemTablePushdownFilter> PUSHDOWN_FILTERS = List.of(
+ new SystemTablePushdownFilter("server", null),
+ new SystemTablePushdownFilter("service_name", null)
+ );
+
+ private final DruidNode selfNode;
+ private final Set<NodeRole> selfNodeRoles;
+ private final AuthorizerMapper authorizerMapper;
+ private final Properties properties;
+ private final DruidServerConfig druidServerConfig;
+
+ @Inject
+ public ServerPropertiesTableDataProvider(
+ @Self final DruidNode selfNode,
+ @Self final Set<NodeRole> selfNodeRoles,
+ final AuthorizerMapper authorizerMapper,
+ final Properties properties,
+ final DruidServerConfig druidServerConfig
+ )
+ {
+ this.selfNode = selfNode;
+ this.selfNodeRoles = selfNodeRoles;
+ this.authorizerMapper = authorizerMapper;
+ this.properties = properties;
+ this.druidServerConfig = druidServerConfig;
+ }
+
+ @Override
+ public List<SystemTablePushdownFilter> getPushdownFilters()
+ {
+ return PUSHDOWN_FILTERS;
+ }
+
+ @Override
+ public Iterable<Object[]> getRows(
+ final List<DimFilter> filters,
+ final AuthenticationResult internalAuthenticationResult
+ )
+ {
+ authorizeServerRead(internalAuthenticationResult);
+
+ String serverFilter = null;
+ String serviceNameFilter = null;
+ for (final DimFilter filter : filters) {
+ if (isStringEqualityFilter(filter, "server")) {
+ serverFilter = getFilterValue(filter);
+ } else if (isStringEqualityFilter(filter, "service_name")) {
+ serviceNameFilter = getFilterValue(filter);
+ }
+ }
+
+ final String server = selfNode.getHostAndPortToUse();
+ if (serverFilter != null && !serverFilter.equals(server)) {
+ return Collections.emptyList();
+ }
+ if (serviceNameFilter != null && !serviceNameFilter.equals(selfNode.getServiceName())) {
+ return Collections.emptyList();
+ }
+
+ final ServerProperties serverProperties = new ServerProperties(selfNode.getServiceName(), server);
+ selfNodeRoles.stream()
+ .map(NodeRole::getJsonName)
+ .sorted()
+ .forEach(serverProperties.nodeRoles::add);
+
+ return buildRows(serverProperties);
+ }
+
+ private void authorizeServerRead(final AuthenticationResult authenticationResult)
Review Comment:
Moved to `SystemTableAuthorization`. The descriptor's row authorizer is now the only place it's applied.
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryHandler.java:
##########
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.inject.Inject;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.BadQueryContextException;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryEngine;
+import org.apache.druid.segment.InlineSegmentWrangler;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.system.table.SystemTableDataProvider;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTablePushdownFilter;
+
+import java.util.Map;
+
+/** Resolves one node-local system table and returns its rows through the standard native Scan stack. */
+public class SystemTableQueryHandler implements DataSourceQueryHandler
+{
+ private final Map<String, SystemTableDataProvider> dataSuppliers;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final ScanQueryEngine scanQueryEngine;
+ private final AuthorizerMapper authorizerMapper;
+
+ @Inject
+ public SystemTableQueryHandler(
+ final Map<String, SystemTableDataProvider> dataSuppliers,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final ScanQueryEngine scanQueryEngine,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.dataSuppliers = dataSuppliers;
+ this.tableDescriptors = tableDescriptors;
+ this.scanQueryEngine = scanQueryEngine;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult requestAuthenticationResult,
+ final boolean executeLocally
+ )
+ {
+ if (!(query instanceof ScanQuery)) {
+ throw new BadQueryContextException(
+ "Local system-table execution requires a Scan query with a SystemTableDataSource at the query root"
+ );
+ }
+ if (!(query.getDataSource() instanceof SystemTableDataSource dataSource)) {
+ throw new BadQueryContextException(
+ "Local system-table execution requires a Scan query with a SystemTableDataSource at the query root"
+ );
+ }
+ final SystemTableDataProvider dataSupplier = dataSuppliers.get(dataSource.getTable());
+ if (dataSupplier == null) {
+ throw new ISE("System table[%s] is not served by this node", dataSource.getTable());
Review Comment:
Converted, except `BadQueryContextException` in `readQuery` (only `QueryException` maps to 400 there) and `SystemTableNotLeaderException` (retry matches its class name).
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r3964484012
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
Review Comment:
Addressed in the stacked sys.tasks PR: the redundant test Javadocs were removed. See FrankChen021/druid#198.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysTasksQueryTest.java:
##########
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.embedded.query;
+
+import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class NativeSysTasksQueryTest extends EmbeddedClusterTestBase
+{
+ private static final String TASK_PREFIX = "native_sys_mvp_";
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedIndexer()
+ .addProperty("druid.worker.capacity", "5"))
+ .addServer(overlord)
+ .addServer(broker);
+ }
+
+ @BeforeAll
+ public void createTasks()
+ {
+ createTasks("a", "native_sys_a", 2);
+ createTasks("b", "native_sys_b", 3);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testGroupByUsesOverlordProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT datasource, COUNT(*) "
+ + "FROM sys.tasks "
+ + "WHERE task_id = 'native_sys_mvp_a_0' AND datasource = 'native_sys_a' "
+ + "GROUP BY datasource",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ final Set<String> rows = Arrays.stream(result.split("\\n")).collect(Collectors.toSet());
+ Assertions.assertEquals(Set.of("native_sys_a,1"), rows);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationsSupportDistinctCount(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT task_id), COUNT(DISTINCT datasource), SUM(1) "
+ + "FROM sys.tasks "
+ + "WHERE datasource IN ('native_sys_a', 'native_sys_b')",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5,5,2,5", result);
+ }
+
+ /**
+ * Verifies execution when native planning represents the inner aggregation as a query datasource.
+ *
+ * <pre>{@code
+ * SELECT COUNT(*)
+ * FROM
+ * (
+ * SELECT
+ * task_id,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY task_id
+ * )
+ * WHERE task_count > 0
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNestedTaskAggregation(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*) "
+ + "FROM ("
+ + " SELECT task_id, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY task_id"
+ + ") "
+ + "WHERE task_count > 0",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ /**
+ * Verifies that an outer aggregation can consume grouped rows from a native system-table subquery.
+ *
+ * <pre>{@code
+ * SELECT SUM(task_count)
+ * FROM
+ * (
+ * SELECT
+ * datasource,
+ * COUNT(*) AS task_count
+ * FROM sys.tasks
+ * WHERE datasource IN ('native_sys_a', 'native_sys_b')
+ * GROUP BY datasource
+ * )
+ * }</pre>
+ */
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testOuterAggregationOverTaskSubquery(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT SUM(task_count) "
+ + "FROM ("
+ + " SELECT datasource, COUNT(*) AS task_count "
+ + " FROM sys.tasks "
+ + " WHERE datasource IN ('native_sys_a', 'native_sys_b') "
+ + " GROUP BY datasource"
+ + ")",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("5", result);
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testInAndOrFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id IN ('native_sys_mvp_a_0', 'native_sys_mvp_b_2') "
+ + "OR task_id = 'native_sys_mvp_missing'",
+ Set.of("native_sys_mvp_a_0", "native_sys_mvp_b_2"),
+ plannerStrategy
+ );
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testRangeAndLikeFilters(final String plannerStrategy)
+ {
+ assertTaskFilterQuery(
+ "task_id >= 'native_sys_mvp_b_0' "
+ + "AND task_id < 'native_sys_mvp_b_2' "
+ + "AND task_id LIKE 'native_sys_mvp_b_%'",
+ Set.of("native_sys_mvp_b_0", "native_sys_mvp_b_1"),
+ plannerStrategy
+ );
+ }
+
+ /**
+ * Verifies that expression virtual columns required by a pushed node filter are preserved.
+ *
+ * <pre>{@code
+ * SELECT task_id
+ * FROM sys.tasks
+ * WHERE UPPER(task_id) = 'NATIVE_SYS_MVP_A_0'
+ * }</pre>
+ */
Review Comment:
Addressed in the stacked sys.tasks PR: the repeated SQL Javadocs in this test were removed. See FrankChen021/druid#198.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4236460168
##########
server/src/main/java/org/apache/druid/server/system/table/SystemTablePushdownFilter.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.table;
+
+import org.apache.druid.query.Query;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.BoundDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.EqualityFilter;
+import org.apache.druid.query.filter.InDimFilter;
+import org.apache.druid.query.filter.LikeDimFilter;
+import org.apache.druid.query.filter.NotDimFilter;
+import org.apache.druid.query.filter.OrDimFilter;
+import org.apache.druid.query.filter.RangeFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.query.filter.TypedInFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.ordering.StringComparators;
+import org.apache.druid.segment.column.ColumnType;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Declares a pushdown-capable system-table column and its optional supplier-side name.
+ */
+public record SystemTablePushdownFilter(String key, @Nullable String value)
+{
+ public static List<DimFilter> extract(
+ final Query<?> query,
+ final List<SystemTablePushdownFilter> pushdownFilters
+ )
+ {
+ if (pushdownFilters.isEmpty()) {
+ return Collections.emptyList();
+ }
+ final Map<String, String> columnMappings = pushdownFilters.stream().collect(
+ Collectors.toMap(
+ SystemTablePushdownFilter::key,
+ filter -> filter.value() == null ? filter.key() : filter.value()
+ )
+ );
+
+ final List<DimFilter> extracted = new ArrayList<>();
+ extractConjuncts(query.getFilter(), columnMappings, extracted);
Review Comment:
Fixed in `12d36e8233`.
Reproduced first: the new `SystemTableQueryHandlerTest#testFilterOnVirtualColumnShadowingPushdownColumnKeepsMatchingRows` runs a local `server_properties` Scan with a constant virtual column `server = 'alias'` and filter `server = 'alias'`. On the previous head it returned 0 rows (expected 1) because the provider compared `alias` against the physical host:port; with the fix it returns the expected row.
`SystemTablePushdownFilter.extract` now excludes pushdown columns shadowed by query or window leaf-scan virtual columns, so those filters stay in the native scan. Non-shadowed conjuncts are still pushed down. Broker fanout forwards the virtual columns together with the pushed filter, so the node-side extraction applies the same rule. Also added extraction-level regression tests in `SystemTablePushdownFilterTest` for the query-level and window leaf-scan cases.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4306123857
##########
server/src/main/java/org/apache/druid/server/QueryLifecycle.java:
##########
@@ -420,13 +465,58 @@ public <T> QueryResponse<T> execute()
final ResponseContext responseContext = DirectDruidClient.makeResponseContextForQuery();
@SuppressWarnings("unchecked")
- final Sequence<T> res = QueryPlus.wrap((Query<T>) baseQuery)
- .withIdentity(authenticationResult.getIdentity())
- .run(texasRanger, responseContext);
+ final Query<T> originalQuery = (Query<T>) baseQuery;
+ final DataSourceQueryHandler dataSourceQueryHandler = findDataSourceQueryHandler(originalQuery.getDataSource());
+ // Handlers can start fanout or provider work before reaching a segment walker, or bypass it entirely.
+ // Apply the same server guardrails first, and pass the normalized query to both runner creation and execution.
+ final Query<T> query = dataSourceQueryHandler == null
+ ? originalQuery
+ : SetAndVerifyContextQueryRunner.withTimeoutAndMaxScatterGatherBytes(
+ originalQuery,
+ serverConfig,
+ System.currentTimeMillis()
+ );
+ final QueryRunner<T> queryRunner = dataSourceQueryHandler == null
Review Comment:
[P2] System-table queries cannot be cancelled by query ID
**Finding:** The new DataSourceQueryHandler branch invokes createRunner directly instead of registering the outer query with the QueryScheduler/query-watcher path used by normal native execution. When DELETE targets the user-facing system-table query ID, QueryResource finds no registered future or datasource and returns without cancelling the generated native-system-node-* requests, so long-running or fanout queries continue until timeout or stream cleanup.
**Suggestion:** Add an explicit scheduler/cancellation bridge for handler execution that associates the outer query ID with the handler and all child requests, and cover DELETE cancellation for broker and node-local system-table queries.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-6307449119
@gianm @GWphua @abhishekrb19 any comments for this one?
While the change is wide, but it has little impact over existing logic as most of the newly added code is activated via the new context paramterer `allowNativeFunctionsForSystemTables` which is also mark as experimental.
As the AI has performed multiple rounds of review, I think the risk is low to us to include it into the master.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-6365220399
> Server-level opt-in: There should be a common.runtime.properties server property to enable this feature, off by default. When it's off, don't bind the new stuff or accept systemTable datasources. Also, the context key should do nothing if the property is off (it can't override the server property). This helps operators control the rollout.
Currently I use the query context parameter to control the new path. The query context parameter is not enabled by default. If a system property is introduced, will it be a little bit complex?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-6368449437
@gianm Fixed the join issue: node scans shared the caller's `ResponseContext`, so the right side's scanned-row counts made the outer limited Scan return no rows. Each system-table leaf now gets its own response context.
Tests:
- `NativeSysServerPropertiesQueryTest#testInSubqueryOnSameSystemTable`: your query (`IN` subquery on `sys.server_properties`, `LIMIT 5`), with both planners.
- `SystemTableQueryClientTest#testNodeScansDoNotShareCallerResponseContext`: node scans don't update the caller's response context.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4359604777
##########
docs/querying/sql-metadata-tables.md:
##########
@@ -135,10 +135,61 @@ WHERE "IS_AGGREGATOR" = 'YES'
The "sys" schema provides visibility into Druid segments, servers and tasks.
:::info
- Note: "sys" tables do not currently support Druid-specific functions like `TIME_PARSE` and
- `APPROX_QUANTILE_DS`. Only standard SQL functions can be used.
+ By default, "sys" tables use the SQL-layer execution path and support only standard SQL functions. You can enable
+ experimental [native query execution](#native-query-execution) for supported system tables, which enables expressions and
+ aggregations that the native SQL engine can translate.
+:::
+
+### Native query execution
+
+:::caution Experimental
+Native query execution for system tables is an experimental, opt-in feature. Its behavior and interfaces may change.
:::
+The native SQL engine can plan supported system tables as native datasources. To enable this behavior for a query, set
+`allowNativeFunctionsForSystemTables` to `true` in the SQL query context:
Review Comment:
Initially, I used the name `useNativeEngineForSystemTables`, however, for long term evolution, I found that the dart engine might be the better(it already has the exeuction path that fans out scan queries and aggregation sub queries), considering that, I change the name to current one.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #20183:
URL: https://github.com/apache/druid/pull/20183#issuecomment-6365071107
> At a high level the approach looks OK, although I wonder why you used `/druid/v2/` for fanout rather than the existing APIs? I
What do you mean by existing APIs here? do you mean the API for the server properties like `/status/properties` ?
The reason is that I would like the internal data fetching from different servers to be unified like how we execute sub queries on different nodes. By this way, in future, there's any new table(like the sys.stack_trace table) added, we don't add new APIs.
Currently SCAN query is fan out from broker to different nodes, and I was planning to push down aggregation queries in a later stage. So the /druid/v2 is the best place to serve the need.
However, for the native engine, I will not do the aggregation query push down, because I found that the the dart engine already does this, so the long term evolution here is to use the dart engine to fan out sub queries.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat(sql): add experimental native query support for system tables (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20183:
URL: https://github.com/apache/druid/pull/20183#discussion_r4369253772
##########
server/src/main/java/org/apache/druid/server/system/handler/SystemTableQueryClient.java:
##########
@@ -0,0 +1,923 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.system.handler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import io.netty.channel.ChannelException;
+import org.apache.druid.client.DirectDruidClient;
+import org.apache.druid.client.DirectDruidClientFactory;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.LazySequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.Yielders;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContext;
+import org.apache.druid.query.QueryDataSource;
+import org.apache.druid.query.QueryException;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.QueryTimeoutException;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.query.context.QueryContextParameters;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.operator.OperatorFactory;
+import org.apache.druid.query.operator.ScanOperatorFactory;
+import org.apache.druid.query.operator.WindowOperatorQuery;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.server.DataSourceQueryHandler;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.QueryScheduler;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.system.SystemTableNotLeaderException;
+import org.apache.druid.server.system.table.SystemTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.utils.CloseableUtils;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+/**
+ * Recursively replaces native system-table datasources with authorized inline rows fetched from nodes, then
+ * executes the rebuilt query through the Broker's normal native-query walker.
+ */
+public class SystemTableQueryClient implements DataSourceQueryHandler
+{
+ private static final String SYSTEM_TABLE_TIER = "_system";
+
+ private final SystemTableNodeLocator nodeLocator;
+ private final DirectDruidClientFactory directDruidClientFactory;
+ private final QueryScheduler queryScheduler;
+ private final QuerySegmentWalker querySegmentWalker;
+ private final Map<String, SystemTableDescriptor> tableDescriptors;
+ private final AuthorizerMapper authorizerMapper;
+ private final SystemTableQueryHandler localQueryHandler;
+ private final AuthenticationResult escalatedAuthenticationResult;
+ private final DruidNode selfNode;
+
+ @Inject
+ public SystemTableQueryClient(
+ final SystemTableNodeLocator nodeLocator,
+ final DirectDruidClientFactory directDruidClientFactory,
+ final QueryScheduler queryScheduler,
+ final QuerySegmentWalker querySegmentWalker,
+ final Map<String, SystemTableDescriptor> tableDescriptors,
+ final AuthorizerMapper authorizerMapper,
+ final SystemTableQueryHandler localQueryHandler,
+ final Escalator escalator,
+ @Self final DruidNode selfNode
+ )
+ {
+ this.nodeLocator = nodeLocator;
+ this.directDruidClientFactory = directDruidClientFactory;
+ this.queryScheduler = queryScheduler;
+ this.querySegmentWalker = querySegmentWalker;
+ this.tableDescriptors = tableDescriptors;
+ this.authorizerMapper = authorizerMapper;
+ this.localQueryHandler = localQueryHandler;
+ this.escalatedAuthenticationResult = escalator.createEscalatedAuthenticationResult();
+ this.selfNode = selfNode;
+ }
+
+ @Override
+ public <T> QueryRunner<T> createRunner(
+ final Query<T> query,
+ final AuthenticationResult authenticationResult,
+ final boolean executeLocally
+ )
+ {
+ return (queryPlus, responseContext) -> {
+ final NodeSequencesCloser nodeSequencesCloser = new NodeSequencesCloser(queryScheduler);
+ try {
+ final DataSource resolvedDataSource = resolveDataSource(
+ query.getDataSource(),
+ query,
+ authenticationResult,
+ nodeSequencesCloser
+ );
+ final Query<T> resolvedQuery = query.withDataSource(resolvedDataSource).withOverriddenContext(
+ QueryContextParameters.QUERY_RESOURCE_ID,
+ UUID.randomUUID().toString()
+ );
+ final QueryRunner<T> queryRunner = querySegmentWalker.getQueryRunnerForIntervals(
+ resolvedQuery,
+ resolvedQuery.getIntervals()
+ );
+ return Sequences.withBaggage(
+ queryRunner.run(queryPlus.withQuery(resolvedQuery), responseContext),
+ nodeSequencesCloser
+ );
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, nodeSequencesCloser);
+ }
+ };
+ }
+
+ private DataSource resolveDataSource(
+ final DataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ if (dataSource instanceof SystemTableDataSource systemTableDataSource) {
+ return resolveSystemTableDataSource(
+ systemTableDataSource,
+ owningQuery,
+ authenticationResult,
+ nodeSequencesCloser
+ );
+ }
+
+ final Query<?> childOwningQuery;
+ final DataSource resolvedParent;
+ if (dataSource instanceof QueryDataSource queryDataSource) {
+ final Query<?> childQuery = queryDataSource.getQuery();
+ // An inner query may tighten the outer guardrails, but must not discard them before node fanout.
+ childOwningQuery = childQuery.withOverriddenContext(
+ QueryContext.ofMap(
+ QueryContextParameters.QUERY_FAIL_TIME,
+ Math.min(getNodeFailTime(owningQuery), getNodeFailTime(childQuery)),
+ QueryContextParameters.MAX_SCATTER_GATHER_BYTES,
+ Math.min(owningQuery.context().getMaxScatterGatherBytes(), childQuery.context().getMaxScatterGatherBytes())
+ )
+ );
+ resolvedParent = new QueryDataSource(childOwningQuery);
+ } else {
+ childOwningQuery = owningQuery;
+ resolvedParent = dataSource;
+ }
+ final List<DataSource> resolvedChildren = new ArrayList<>();
+ for (final DataSource child : dataSource.getChildren()) {
+ resolvedChildren.add(
+ resolveDataSource(
+ child,
+ childOwningQuery,
+ authenticationResult,
+ nodeSequencesCloser
+ )
+ );
+ }
+ return resolvedParent.withChildren(resolvedChildren);
+ }
+
+ private InlineDataSource resolveSystemTableDataSource(
+ final SystemTableDataSource dataSource,
+ final Query<?> owningQuery,
+ final AuthenticationResult authenticationResult,
+ final NodeSequencesCloser nodeSequencesCloser
+ )
+ {
+ final SystemTableDescriptor descriptor = tableDescriptors.get(dataSource.getTable());
+ if (descriptor == null) {
+ throw DruidException.defensive("No routing descriptor is registered for system table[%s]", dataSource.getTable());
+ }
+
+ final ScanQuery nodeQuery = makeNodeQuery(dataSource, descriptor, owningQuery);
+ final List<QueryRunner<ScanResultValue>> nodeRunners = makeNodeRunners(
+ nodeQuery,
+ descriptor,
+ nodeSequencesCloser
+ );
+ if (nodeRunners.isEmpty() && !descriptor.isEmptyDiscoveryAllowed()) {
+ throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.SERVICE_UNAVAILABLE)
+ .build("No node is available to serve system table[%s]", dataSource.getTable());
+ }
+
+ /*
+ * Keep the scan result transport lazy. The node response is a Sequence, so this lets the local query
+ * runner consume and process rows while the HTTP response is still arriving instead of materializing every
+ * node row in two Broker-side lists. The scan transport is intentionally kept behind this adapter so a
+ * future node-side aggregation mode can use a different result adapter without changing routing or
+ * authorization orchestration here.
+ */
+ final NodeSequenceCloser nodeSequenceCloser = new NodeSequenceCloser();
+ nodeSequencesCloser.add(nodeSequenceCloser);
+ // Node scans get their own response context. Sharing the caller's would add the node scans' row counts to the
+ // Broker query's, and a Scan with a limit returns no rows once that count reaches the limit. For example, the
+ // node scans of a join's right side would empty a limited Scan over the join before it read its left side.
+ final Iterable<Object[]> nodeRows = scanNodeRows(
+ nodeRunners,
+ nodeQuery,
+ DirectDruidClient.makeResponseContextForQuery(),
Review Comment:
leave this in a follow up PR
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org