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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
