Repository navigation
feat(sql): execute system tables with native queries - #166
FrankChen021 wants to merge 16 commits into
Conversation
| "/druid/indexer/v1/isLeader" | ||
| "/druid/indexer/v1/isLeader", | ||
| "/druid/v2", | ||
| "/druid/v2/" |
There was a problem hiding this comment.
why do you add /druid/v2 and /druid/v2?
There was a problem hiding this comment.
Both entries are intentional exact URI variants: /druid/v2 handles a request without a trailing slash, while /druid/v2/ handles the canonical QueryResource path. OverlordRedirectInfo compares exact strings here, so retaining both prevents either valid form from being redirected to the leader.
| return (requestURI != null && LOCAL_PATHS.contains(requestURI)) || overlord.isLeader(); | ||
| return (requestURI != null | ||
| && (LOCAL_PATHS.contains(requestURI) | ||
| || requestURI.startsWith("/druid/v2/" + SystemTableDataSource.COMPONENT_QUERY_ID_PREFIX))) |
There was a problem hiding this comment.
what's the COMPONENT_QUERY_ID_PREFIX used for?
There was a problem hiding this comment.
This prefix identifies Broker-created component-local query IDs. POST /druid/v2 executes locally, while DELETE /druid/v2/{id} is cancellation; the prefix lets a non-leader Coordinator or Overlord keep cancellation local instead of redirecting it to the leader, where that component-local query does not exist.
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| /** Distributes a native system-table query to its owning components through their standard {@code /druid/v2} API. */ | ||
| public class NativeSystemQueryClient implements DataSourceQueryHandler |
There was a problem hiding this comment.
why do we need this? can't current broker to historical distribution handle this? or can we extend it to not only query historical nodes, but also all components?
There was a problem hiding this comment.
The existing Broker distribution path is segment-based: it uses the segment timeline/server view and can only select servers that hold datasource segments. System tables such as tasks and server_properties are owned by arbitrary node roles and have no segments. This client adds role-based discovery, but deliberately reuses DirectDruidClient plus the normal QueryRunnerFactory/toolchest merge pipeline rather than creating a second transport or result format.
| modules.add(new SegmentSchemaCacheModule()); | ||
| modules.add(new QueryableModule()); | ||
| } else { | ||
| modules.add(new RouterProcessingModule()); |
There was a problem hiding this comment.
why are these modules added in the else branch?
There was a problem hiding this comment.
Good catch. The original branch assumed SegmentSchemaCacheModule supplied the complete query stack, but it only supplies the segment-metadata subset. Fixed in 63a1756: shared processing, wrangler, joinable, and response-context dependencies are now installed for both configurations, and the standard query factories are combined with the schema-cache overrides. The all-components GROUP BY test now runs with centralized datasource schema enabled.
| new SegmentWranglerModule(), | ||
| new JoinableFactoryModule(), | ||
| new QueryablePeonModule(), | ||
| new NativeSystemQueryModule(), |
There was a problem hiding this comment.
I'm wondering can we encapsulate these modules inside NativeSystemQueryModule?
There was a problem hiding this comment.
I kept these dependencies outside NativeSystemQueryModule. That module is in indexing-service and owns only system datasource routing, descriptors, and suppliers. QueryResource and QueryablePeonModule live in services, so pulling them into it would create the wrong dependency direction; processing also differs by role and would duplicate bindings on Broker, Historical, and combined Coordinator-Overlord nodes. Commit 63a1756 does remove duplicate installation for the combined role while preserving these module boundaries.
| binder.bindConstant().annotatedWith(Names.named("serviceName")).to("druid/router"); | ||
| binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8888); | ||
| binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(9088); | ||
| binder.bind(ResponseContextConfig.class).toInstance(ResponseContextConfig.newConfig(true)); |
There was a problem hiding this comment.
ResponseContextConfig is required by QueryResourceQueryResultPusherFactory, which the Router now hosts for component-local /druid/v2 requests. The true value permits response-context truncation to the configured header limit instead of failing the query when the header is too large. Ordinary external Router queries are still forwarded; only authenticated component-local system queries enter QueryResource.
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.extension.RegisterExtension; | ||
|
|
||
| @SqlTestFrameworkConfig.ComponentSupplier(CalciteSysQueryTest.BindableComponentSupplier.class) |
There was a problem hiding this comment.
what's the annotation for?
There was a problem hiding this comment.
This annotation selects the custom BindableComponentSupplier for this test class. That supplier exposes a test engine whose name is not native, so the test verifies that sys tables still take the Bindable fallback when the selected engine is not native. Without the annotation, the test would use the default native engine and would not cover the fallback split.
4e35fa2 to
398dfd3
Compare
3eba472 to
4b10dca
Compare
ac1a496 to
cf64a28
Compare
cf64a28 to
dbb3c7c
Compare
dbb3c7c to
a0ce486
Compare
…asks-mvp # Conflicts: # server/src/test/java/org/apache/druid/client/DirectDruidClientTest.java
…asks-mvp # Conflicts: # processing/src/main/java/org/apache/druid/query/rowsandcols/LazilyDecoratedRowsAndColumns.java # services/src/main/java/org/apache/druid/cli/CliPeon.java
Merge tests that exercise the same code path into parameterized tests and drop tests fully covered by stronger equivalents.
Derive native system table names from the logical plan instead of building the native query, and collapse identical branches in RecoveringNodeIterator.
A virtual column shadows the physical column of the same name, so a filter on it must be evaluated by the native scan instead of being compared against the provider's physical column.
…ries WindowOperatorQuery stores null rather than empty virtual columns for a leaf scan without any, which caused an NPE when building the node query for a filtered window query.
FilteredSegment did not provide a CloseableShapeshifter, so window queries over a FilteredDataSource failed with 'cannot shapeshift'. DECOUPLED planning produces this shape for filtered window queries on native system tables.
…ask groups are complete (apache#20505) The changes in apache#19091 have shown that tasks often need to wait for their priors to finish publishing as we often see logs like: ``` Task[xyz] needs to wait before publishing as other taskGroups[abc, def] are currently publishing to partition[123]. ``` Before this patch, if publish from streaming task A is slow, a later generation task B keeps trying to publish repeatedly until eventually failing once the number of retries is exhausted. It is difficult to fix the number of retries since publish times may vary across supervisors and over time. It is also pointless to retry if a prior task is yet to finish publishing. This patch addresses these problems by blocking the publish of later task B until A is complete. Changes --------- - Add `TaskActionClient.submitAsync` which returns a `ListenableFuture` - Update API `/druid/indexer/v1/action` in `OverlordResource` to be async by employing the new `submitAsync()` method - Update task actions `segmentTransactionalAppend` and `segmentTransactionalReplace` to implement `performAsync()` - Update `LocalTaskActionClient.submitAsync()` to use `performAsync()` when supported - Attempt segment publish only when the task is ready to publish - Complete async API request only when the publish attempt is done - For non-streaming tasks, attempt to publish immediately - Add method `isTaskReadyToPublish()` in `SeekableStreamSupervisor`. - Return a `ListenableFuture` that completes when all prior task groups have completed - Update `SegmentAllocateAction` and `SegmentAllocationQueue` to use `ListenableFuture`
Description
This PR adds opt-in native-query execution for supported
systables while retaining the existing Bindable implementation for compatibility.Native system-table planning is selected only when:
native."useNativeQueryForSystemTables": true.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
ScannableTableorProjectableFilterableTable, 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:
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:
sys.tasksTaskStoragesupports them.sys.server_propertiesserverandservice_namefilters can avoid reading properties from components that don't match.Execution path
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: localstill passes through the receiving process'snormal 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
SystemTableDataSourceand explicitly requestthe
localroute. Other/druid/v2requests continue through normal Broker forwarding. This prevents recursiveforwarding while allowing the Router to contribute its own rows to
sys.server_properties.The implementation uses the standard
/druid/v2endpoint and standardScanResultValueresponse format. It doesnot 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,
DruidTableScanRulenormally sees the original CalciteTableScanand replaces a native-capable system table with itsDruidTableScanrepresentation.In DECOUPLED mode, Calcite may first convert a
ProjectableFilterableTableinto aBindables.BindableTableScan. This is particularly relevant tosys.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.DruidBindableTableScanRulehandles this DECOUPLED-only plan shape afterQueryHandlerhas selected native system-table planning. It converts: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_propertiesplanning 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.
Define a
SystemTableDescriptorcontaining:RowSignature.Register the descriptor in the
MapBinder<String, SystemTableDescriptor>installed bySystemTableModule. The descriptor must be available to the Broker for discovery, authorization, and row decoding, as well as to components serving the table.Implement
SystemTableDataProvider:The provider may override
getPushdownFilters()to advertise columns that support safe provider or storage pushdown.Register the provider by table name on the owning component:
The provider should be installed only on components represented by the descriptor's node roles. For example,
SystemTasksTableDataProvideris bound byCliOverlord, whileServerPropertiesTableDataProvideris installed on all components bySystemTableModule.Expose a native SQL representation. The Calcite system-table definition implements
NativeSystemTableand returns its nativeDruidTablefromasNativeTable(). The native table usesSystemTableDataSourceand the descriptor's shared row signature.The
DataSourceQueryHandlerregistration 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:
The following example shows how a filter moves from SQL to task metadata storage:
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 nativeDimFilterexpressions.MetadataTaskStoragesupports task metadata SQL pushdown.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.NativeQueryEngineModuleinstalls:QueryableModule.QueryRunnerFactoryModule.SegmentWranglerModule.JoinableFactoryModule.SystemTableModule.Its builder supports role-specific profiles:
scanOnly()installs only the Scan query factory and toolchest, plus a defaultNoopQuerySegmentWalker. 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 forSegmentSchemaCacheModule.withQueryResourceModule(...)replaces the standardQueryResourceModule. The Broker installsBrokerQueryResourceModule, and the Router installsRouterQueryResourceModule.Full native-query components use the normal query-runner factories. Scan-only management components reuse
QueryRunnerFactoryModulewith 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-Routeheader 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
AuthenticationResultand callsthe 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
AuthenticationResultand applies the table descriptor's row authorizerto 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.
useNativeQueryForSystemTablesdefaults tofalseto support rolling upgrades. Operators should leave it disableduntil the Broker and components serving native system tables have been upgraded. It can then be enabled per query or
globally on Brokers with:
druid.query.default.context.useNativeQueryForSystemTables=trueCurrent 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:
currently executes as:
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
QuerySchedulerlane and capacity limits.LIMIT,OFFSET, and column projection are not pushed into task metadata storage.ScanResultValuetransport rather than frame-based exchange.sys.tasksandsys.server_properties.Release note
SQL queries against supported system tables can now opt into Druid's native query engine with the
useNativeQueryForSystemTablesquery context parameter. The initial implementation supportssys.tasksandsys.server_properties, uses the standard/druid/v2endpoint for component fanout, supports safe storage filter pushdown, and retains the existing execution path by default for rolling-upgrade compatibility.Validation
Coverage includes:
COUNT(DISTINCT ...).IN, range,LIKE, negation, status, and nullable migration columns.sys.tasksandsys.server_properties.This PR has: