feat(sql): add native query support for system tables

docs/querying/sql-metadata-tables.md

@@ -135,10 +135,57 @@ 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 server) FROM sys.server_properties",
+  "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 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 `useNativeQueryForSystemTables` 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.useNativeQueryForSystemTables=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
+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.
+
 ### SEGMENTS table
 
 Segments table provides details on all Druid segments, whether they are published yet or not.

docs/querying/sql-query-context.md

@@ -45,6 +45,7 @@ The table below lists the query context parameters you can use with Druid SQL.
 |`useLexicographicTopN`|If `true`, Druid can use [TopN queries](topnquery.md) with lexicographic dimension ordering. If `false`, Druid uses [GroupBy queries](groupbyquery.md) instead for lexicographic ordering. When both `useLexicographicTopN` and `useApproximateTopN` are `false`, TopN queries are never used.|`false`|
 |`enableTimeBoundaryPlanning`|If `true`, Druid converts SQL queries to [time boundary queries](timeboundaryquery.md) wherever possible. Time boundary queries are very efficient for min-max calculation on the `__time` column in a datasource. |`false`|
 |`useNativeQueryExplain`|If `true`, `EXPLAIN PLAN FOR` returns the explain plan as a JSON representation of equivalent native query, else it returns the original version of explain plan generated by Calcite.<br /><br />This property is provided for backwards compatibility. We don't recommend setting this parameter unless your application depends on the older behavior.|`true`|
+|`useNativeQueryForSystemTables`|If `true` and the resolved SQL engine is `native`, Druid uses native query planning for [system tables](sql-metadata-tables.md#native-query-execution) that advertise native query support. System tables without native query support continue to use their existing execution path.<br /><br />During a rolling upgrade, leave this parameter disabled until the Broker and the components that serve the native system tables have been upgraded. You can set a cluster-wide default on Brokers with `druid.query.default.context.useNativeQueryForSystemTables`.|`false`|
 |`sqlFinalizeOuterSketches`|If `false` (default behavior in Druid 25.0.0 and later), `DS_HLL`, `DS_THETA`, and `DS_QUANTILES_SKETCH` return sketches in query results. If `true` (default behavior in Druid 24.0.1 and earlier), Druid finalizes sketches from these functions when they appear in query results.<br /><br />This property is provided for backwards compatibility with behavior in Druid 24.0.1 and earlier. We don't recommend setting this parameter unless your application uses Druid 24.0.1 or earlier. Instead, use a function that doesn't return a sketch, such as `APPROX_COUNT_DISTINCT_DS_HLL`, `APPROX_COUNT_DISTINCT_DS_THETA`, `APPROX_QUANTILE_DS`, `DS_THETA_ESTIMATE`, or `DS_GET_QUANTILE`.|`false`|
 |`sqlUseBoundAndSelectors`|If `false` (default behavior in Druid 27.0.0 and later), the SQL planner uses [equality](./filters.md#equality-filter), [null](./filters.md#null-filter), and [range](./filters.md#range-filter) filters instead of [selector](./filters.md#selector-filter) and [bounds](./filters.md#bound-filter). For filtering `ARRAY` typed values, `sqlUseBoundAndSelectors` must be `false`. | `false`.|
 |`sqlUseExtractionFns`|If false, the SQL planner avoids using [`extractionFn`](dimensionspecs.md#extraction-functions) in favor of using other constructs such as [virtual columns](virtual-columns.md). This parameter is provided for compatibility with prior behavior, and may be removed in a future release.|false|

docs/querying/sql-translation.md

@@ -817,14 +817,20 @@ This query context  parameter is a temporary solution to avoid the known issue.
 
 Druid does not support all SQL features. In particular, the following features are not supported.
 
-- JOIN between native datasources (table, lookup, subquery) and [system tables](sql-metadata-tables.md).
+- JOIN between native datasources (table, lookup, subquery) and [system tables](sql-metadata-tables.md) that use the
+  traditional SQL-layer execution path. System tables that support native execution can participate in these joins
+  when `useNativeQueryForSystemTables` is enabled.
 - JOIN conditions that are not an equality between expressions from the left- and right-hand sides.
 - JOIN conditions containing a constant value inside the condition.
 - JOIN conditions on a column which contains a multi-value dimension.
-- ORDER BY for a non-aggregating query, except for `ORDER BY __time` or `ORDER BY __time DESC`, which are supported.
-  This restriction only applies to non-aggregating queries; you can ORDER BY any column in an aggregating query.
+- ORDER BY for a non-aggregating query, except for `ORDER BY __time` or `ORDER BY __time DESC`, and native-enabled
+  system-table queries that use the native window-query path. This restriction only applies to non-aggregating queries;
+  you can ORDER BY any column in an aggregating query.
 - DDL and DML.
-- Using Druid-specific functions like `TIME_PARSE` and `APPROX_QUANTILE_DS` on [system tables](sql-metadata-tables.md).
+- Using Druid-specific functions like `TIME_PARSE` and `APPROX_QUANTILE_DS` on system tables that use the traditional
+  SQL-layer execution path. [System tables with native query support](sql-metadata-tables.md#native-query-execution) can
+  use expressions and aggregations that the native SQL engine can translate when `useNativeQueryForSystemTables` is
+  enabled.
 
 Additionally, some Druid native query features are not supported by the SQL language. Some unsupported Druid features
 include:
@@ -834,5 +840,3 @@ include:
 - [Multi-value dimensions](sql-data-types.md#multi-value-strings) are only partially implemented in Druid SQL. There are known
 inconsistencies between their behavior in SQL queries and in native queries due to how they are currently treated by
 the SQL planner.
-
-

docs/querying/sql.md

@@ -79,11 +79,15 @@ datasources can be referenced as either `druid.dataSourceName` or simply `dataSo
 - [Lookups](datasource.md#lookup) from the `lookup` schema, for example `lookup.countries`. Note that lookups can
 also be queried using the [`LOOKUP` function](sql-scalar.md#string-functions).
 - [Subqueries](datasource.md#query).
-- [Joins](datasource.md#join) between anything in this list, except between native datasources (table, lookup,
-query) and system tables. The join condition must be an equality between expressions from the left- and right-hand side
-of the join.
-- [Metadata tables](sql-metadata-tables.md) from the `INFORMATION_SCHEMA` or `sys` schemas. Unlike the other options for the
-FROM clause, metadata tables are not considered datasources. They exist only in the SQL layer.
+- [Joins](datasource.md#join) between anything in this list. The traditional SQL-layer system-table path does not
+  support joins between native datasources (table, lookup, query) and system tables. When
+  [`useNativeQueryForSystemTables`](sql-metadata-tables.md#native-query-execution) is enabled, system tables that support
+  native execution can participate in native datasource joins. The join condition must be an equality between
+  expressions from the left- and right-hand side of the join.
+- [Metadata tables](sql-metadata-tables.md) from the `INFORMATION_SCHEMA` or `sys` schemas. By default, metadata tables
+  exist only in the SQL layer and are not considered datasources. When
+  [`useNativeQueryForSystemTables`](sql-metadata-tables.md#native-query-execution) is enabled, supported `sys` tables are
+  represented as native system-table datasources.
 
 For more information about table, lookup, query, and join datasources, refer to the [Datasources](datasource.md)
 documentation.
@@ -295,7 +299,8 @@ grouping expressions or aggregated values. It can only be used together with GRO
 The ORDER BY clause refers to columns that are present after execution of GROUP BY. It can be used to order the results
 based on either grouping expressions or aggregated values. ORDER BY can refer to an expression or a select clause
 ordinal position (like `ORDER BY 2` to order by the second selected column). For non-aggregation queries, ORDER BY
-can only order by the `__time` column. For aggregation queries, ORDER BY can order by any column.
+can normally only order by the `__time` column. Native-enabled system tables can also use the native window-query path
+to order by other columns. For aggregation queries, ORDER BY can order by any column.
 
 ## LIMIT
 

embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServerPropertiesQueryTest.java

@@ -0,0 +1,182 @@
+/*
+ * 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.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.EmbeddedHistorical;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.EmbeddedRouter;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Map;
+
+public class NativeSysServerPropertiesQueryTest extends EmbeddedClusterTestBase
+{
+  private static final String SERVICE_NAME = "native/mvp/broker";
+  private static final String COORDINATOR_PROPERTY = "native.sys.server.properties.coordinator";
+  private static final String OVERLORD_PROPERTY = "native.sys.server.properties.overlord";
+  private static final String BROKER_PROPERTY = "native.sys.server.properties.broker";
+  private static final String HISTORICAL_PROPERTY = "native.sys.server.properties.historical";
+  private static final String INDEXER_PROPERTY = "native.sys.server.properties.indexer";
+  private static final String ROUTER_PROPERTY = "native.sys.server.properties.router";
+
+  private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator()
+      .addProperty(COORDINATOR_PROPERTY, "enabled");
+
+  private final EmbeddedOverlord overlord = new EmbeddedOverlord()
+      .addProperty(OVERLORD_PROPERTY, "enabled");
+
+  private final EmbeddedBroker broker = new EmbeddedBroker()
+      .addProperty("druid.service", SERVICE_NAME)
+      .addProperty(BROKER_PROPERTY, "enabled");
+
+  private final EmbeddedHistorical historical = new EmbeddedHistorical()
+      .addProperty(HISTORICAL_PROPERTY, "enabled");
+
+  private final EmbeddedIndexer indexer = new EmbeddedIndexer()
+      .addProperty(INDEXER_PROPERTY, "enabled");
+
+  private final EmbeddedRouter router = new EmbeddedRouter()
+      .addProperty(ROUTER_PROPERTY, "enabled");
+
+  @Override
+  protected EmbeddedDruidCluster createCluster()
+  {
+    return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+                               .useLatchableEmitter()
+                               .addCommonProperty("druid.centralizedDatasourceSchema.enabled", "true")
+                               .addServer(coordinator)
+                               .addServer(overlord)
+                               .addServer(broker)
+                               .addServer(historical)
+                               .addServer(indexer)
+                               .addServer(router);
+  }
+
+  /**
+   * Verifies that a native query for {@code sys.server_properties} reaches every persistent node type in the
+   * embedded cluster. In particular, the Router row proves that the Broker's native request included
+   * {@code X-Druid-Native-Query-Route: local}: without that header the Router would forward the request back to the
+   * Broker instead of reading its local system-table provider.
+   */
+  @ParameterizedTest(name = "plannerStrategy = {0}")
+  @ValueSource(strings = {
+      QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+      QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+  })
+  public void testServerPropertiesFansOutToAllNodes(final String plannerStrategy)
+  {
+    final String result = cluster.runSql(
+        "SELECT service_name, COUNT(*) "
+        + "FROM sys.server_properties "
+        + "WHERE property IN ('" + COORDINATOR_PROPERTY + "', '" + OVERLORD_PROPERTY + "', '" + BROKER_PROPERTY
+        + "', '" + HISTORICAL_PROPERTY + "', '" + INDEXER_PROPERTY + "', '" + ROUTER_PROPERTY + "') "
+        + "GROUP BY service_name ORDER BY service_name",
+        nativeQueryContext(plannerStrategy)
+    );
+
+    Assertions.assertEquals(
+        String.join(
+            "\n",
+            "druid/coordinator,1",
+            "druid/historical,1",
+            "druid/indexer,1",
+            "druid/overlord,1",
+            "druid/router,1",
+            SERVICE_NAME + ",1"
+        ),
+        result
+    );
+  }
+
+  @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 service_name), COUNT(DISTINCT server), "
+        + "COUNT(DISTINCT property), SUM(1) "
+        + "FROM sys.server_properties "
+        + "WHERE property IN ('" + COORDINATOR_PROPERTY + "', '" + OVERLORD_PROPERTY + "', '" + BROKER_PROPERTY
+        + "', '" + HISTORICAL_PROPERTY + "', '" + INDEXER_PROPERTY + "', '" + ROUTER_PROPERTY + "')",
+        nativeQueryContext(plannerStrategy)
+    );
+
+    Assertions.assertEquals("6,6,6,6,6", result);
+  }
+
+  /**
+   * A filtered window-function query is planned as a {@code WindowOperatorQuery} over a system-table Scan, whose
+   * filter is pushed into the node scans.
+   */
+  @ParameterizedTest(name = "plannerStrategy = {0}")
+  @ValueSource(strings = {
+      QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+      QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+  })
+  public void testNativeWindowFunction(final String plannerStrategy)
+  {
+    final String result = cluster.runSql(
+        "SELECT service_name, ROW_NUMBER() OVER (ORDER BY service_name) "
+        + "FROM sys.server_properties "
+        + "WHERE property IN ('" + COORDINATOR_PROPERTY + "', '" + OVERLORD_PROPERTY + "', '" + BROKER_PROPERTY
+        + "', '" + HISTORICAL_PROPERTY + "', '" + INDEXER_PROPERTY + "', '" + ROUTER_PROPERTY + "') "
+        + "ORDER BY service_name",
+        nativeQueryContext(plannerStrategy)
+    );
+
+    Assertions.assertEquals(
+        String.join(
+            "\n",
+            "druid/coordinator,1",
+            "druid/historical,2",
+            "druid/indexer,3",
+            "druid/overlord,4",
+            "druid/router,5",
+            SERVICE_NAME + ",6"
+        ),
+        result
+    );
+  }
+
+  private static Map<String, Object> nativeQueryContext(final String plannerStrategy)
+  {
+    return Map.of(
+        QueryContexts.ENGINE,
+        NativeSqlEngine.NAME,
+        PlannerContext.CTX_USE_NATIVE_QUERY_FOR_SYSTEM_TABLES,
+        true,
+        QueryContexts.CTX_NATIVE_QUERY_SQL_PLANNING_MODE,
+        plannerStrategy
+    );
+  }
+}

indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordRedirectInfo.java

@@ -24,6 +24,7 @@
 import com.google.inject.Inject;
 import org.apache.druid.indexing.overlord.DruidOverlord;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.SystemTableDataSource;
 import org.apache.druid.server.http.RedirectInfo;
 
 import java.net.URL;
@@ -35,7 +36,9 @@ public class OverlordRedirectInfo implements RedirectInfo
 {
   private static final Set<String> LOCAL_PATHS = ImmutableSet.of(
       "/druid/indexer/v1/leader",
-      "/druid/indexer/v1/isLeader"
+      "/druid/indexer/v1/isLeader",
+      "/druid/v2",
+      "/druid/v2/"
   );
 
   private final DruidOverlord overlord;
@@ -49,7 +52,10 @@ public OverlordRedirectInfo(DruidOverlord overlord)
   @Override
   public boolean doLocal(String requestURI)
   {
-    return (requestURI != null && LOCAL_PATHS.contains(requestURI)) || overlord.isLeader();
+    return (requestURI != null
+            && (LOCAL_PATHS.contains(requestURI)
+                || requestURI.startsWith("/druid/v2/" + SystemTableDataSource.NODE_QUERY_ID_PREFIX)))
+           || overlord.isLeader();
   }
 
   @Override