feat(msq): execute system tables with Dart

docs/querying/dart.md

@@ -47,6 +47,12 @@ When processing these kinds of queries, Dart can parallelize through the entire
 
 By default, Dart queries include results from published segments and realtime tasks.
 
+Dart can also execute queries against `sys.server_properties`. The Broker plans this table as a distributed MSQ
+input and uses an embedded worker for system-table processing, so these queries do not require a Historical. The
+worker reads the local server's properties directly and fetches properties from discovered remote Druid services;
+later stages perform aggregation, joins, and ordering using the standard Dart execution path. Other system tables
+continue to use their existing SQL execution paths.
+
 ## Enable Dart
 
 To enable Dart, add the following line to your `_common/common.runtime.properties` files:
@@ -127,7 +133,7 @@ You can use any SQL query context parameters to control Dart's behavior unless o
 | `finalizeAggregations` | Determines the type of aggregation to return. If true, Druid finalizes the results of complex aggregations that directly appear in query results. If false, Druid returns the aggregation's intermediate type rather than finalized type. This parameter is useful during ingestion, where it enables storing sketches directly in Druid tables. For more information about aggregations, see [SQL aggregation functions](../querying/sql-aggregations.md). | `true` |
 | `includeSegmentSource` |  Controls the sources Druid queries for results in addition to the segments present on deep storage. Can be `NONE` or `REALTIME`. If set to `NONE`, only non-realtime (published and used) segments are downloaded from deep storage. If set to `REALTIME`, results are also included from realtime tasks.|  `REALTIME` |
 | `removeNullBytes` |The MSQ engine can't process null bytes in strings and throws `InvalidNullByteFault` if it encounters them in the source data. If the parameter is set to true, the MSQ engine removes the null bytes in string fields when reading the data. | `false` |
-|`maxConcurrentStages`|Number of stages that can run concurrently for a query. A higher number can potentially improve pipelining but results in less memory available for each stage.|2|
+|`maxConcurrentStages`|Number of stages that can run concurrently for a query. A higher number can potentially improve pipelining but results in less memory available for each stage. The nano and micro quickstart profiles configure this to `1`. An explicit query-context value overrides the configured default.|2|
 |`maxNonLeafWorkers`|Number of workers to use for stages beyond the leaf stage.| 1 (Scatter-gather style)|
 | `sqlJoinAlgorithm` | Algorithm to use for JOIN. Use `broadcast` (the default) for broadcast hash join or `sortMerge` for sort-merge join. Affects all JOIN operations in the query. This is a hint to the MSQ engine and the actual joins in the query may proceed in a different way than specified. See [Joins](../multi-stage-query/reference.md#joins) for more details. | `broadcast` |
 |`targetPartitionsPerWorker`|Number of partitions Druid generates for each worker. This number controls how much parallelism can be maintained throughout a query.|1|
@@ -144,3 +150,4 @@ You can use any SQL query context parameters to control Dart's behavior unless o
 - Realtime scans from the MSQ engine can't reliably read complex types. This can happen in situations such as if your data includes HLL Sketches for realtime data. Dart returns a `NullPointerException`. For more information, see [#18340](https://github.com/apache/druid/issues/18340).
 - The `NilStageOutputReader` can sometimes lead to a `NoClassDefFoundError`. For more information, see [#18336](https://github.com/apache/druid/pull/18336).
 - Broadcast joins with realtime data aren't supported. If the left table of a join has realtime data and you're doing a broadcast join, you must set `sqlJoinAlgorithm` to `sortMerge`.
+- Of the `sys` tables, only `sys.server_properties` supports distributed Dart execution.

docs/querying/sql-metadata-tables.md

@@ -339,6 +339,17 @@ For example, to retrieve properties for a specific server, use the query
 SELECT * FROM sys.server_properties WHERE server='192.168.1.1:8081'
 ```
 
+You can use the [Dart query engine](dart.md) to run native operators such as exact distinct aggregation across
+server properties:
+
+```sql
+SET engine = 'msq-dart';
+SELECT COUNT(DISTINCT server) FROM sys.server_properties;
+```
+
+For this table, Dart distributes collection across system-table sources and performs later query stages through
+the regular MSQ execution path.
+
 ### QUERIES table
 
 :::info

embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedDartReportApiTest.java

@@ -537,6 +537,19 @@ public void test_sysQueries_authorization() throws IOException
     Assertions.assertTrue(regularUserVisibleSqlQueryIds.contains(regularUserQueryId));
   }
 
+  /** Dart requires STATE READ before planning or executing {@code sys.server_properties}. */
+  @Test
+  @Timeout(60)
+  public void test_serverProperties_authorization()
+  {
+    final String sql = "SET engine = 'msq-dart'; SELECT COUNT(*) FROM sys.server_properties";
+
+    Assertions.assertFalse(runSqlWithClient(sql, adminClient).isBlank());
+    ThrowableMatcher.of(RuntimeException.class)
+                    .expectMessageContains("403 Forbidden")
+                    .assertThrowsAndMatches(() -> runSqlWithClient(sql, regularUserClient));
+  }
+
   /**
    * Extracts SQL query IDs from a {@link GetQueriesResponse} for Dart queries only.
    */

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

@@ -0,0 +1,83 @@
+/*
+ * 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.msq.dart.controller.sql.DartSqlEngine;
+import org.apache.druid.query.QueryContexts;
+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.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 DartSysServerPropertiesNoHistoricalQueryTest extends EmbeddedClusterTestBase
+{
+  private static final String BROKER_PROPERTY = "dart.e2e.noHistorical.broker";
+  private static final String COORDINATOR_PROPERTY = "dart.e2e.noHistorical.coordinator";
+
+  private final EmbeddedBroker broker = new EmbeddedBroker()
+      .setServerMemory(1_000_000_000L)
+      .setServerDirectMemory(1_000_000_000L)
+      .addProperty("druid.msq.dart.controller.heapFraction", "0.5")
+      .addProperty(BROKER_PROPERTY, "broker");
+  private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator()
+      .addProperty(COORDINATOR_PROPERTY, "coordinator");
+
+  @Override
+  protected EmbeddedDruidCluster createCluster()
+  {
+    return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+                               .addCommonProperty("druid.msq.dart.enabled", "true")
+                               .addServer(coordinator)
+                               .addServer(broker);
+  }
+
+  /**
+   * {@code SELECT COUNT(*), COUNT(DISTINCT service_name) FROM sys.server_properties WHERE property IN (...)}
+   *
+   * <p>A Dart system-table query remains executable through the embedded Broker worker when the cluster has no
+   * Historical workers.</p>
+   */
+  @ParameterizedTest(name = "plannerStrategy = {0}")
+  @ValueSource(strings = {
+      QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+      QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+  })
+  public void testBrokerWorkerExecutesWithoutHistorical(final String plannerStrategy)
+  {
+    final String result = cluster.runSql(
+        "SELECT COUNT(*), COUNT(DISTINCT service_name) "
+        + "FROM sys.server_properties "
+        + "WHERE property IN ('" + BROKER_PROPERTY + "', '" + COORDINATOR_PROPERTY + "')",
+        Map.of(
+            QueryContexts.ENGINE,
+            DartSqlEngine.NAME,
+            QueryContexts.CTX_NATIVE_QUERY_SQL_PLANNING_MODE,
+            plannerStrategy
+        )
+    );
+
+    Assertions.assertEquals("2,2", result);
+  }
+}