feat(sql): add native query support for sys.tasks

docs/querying/sql-metadata-tables.md

@@ -147,7 +147,7 @@ The native SQL engine can plan supported system tables as native datasources. To
 
 ```json
 {
-  "query": "SELECT COUNT(DISTINCT server) FROM sys.server_properties",
+  "query": "SELECT COUNT(DISTINCT task_id) FROM sys.tasks",
   "context": {
     "useNativeQueryForSystemTables": true
   }
@@ -159,7 +159,7 @@ execution with `SET`:
 
 ```sql
 SET useNativeQueryForSystemTables = 'true';
-SELECT COUNT(DISTINCT server) FROM sys.server_properties;
+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
@@ -168,6 +168,7 @@ 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,

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

@@ -0,0 +1,302 @@
+/*
+ * 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);
+  }
+
+  @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);
+  }
+
+  @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
+    );
+  }
+
+  @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"
+        + "  FROM sys.tasks\n"
+        + ")\n"
+        + "SELECT \"task_id\", \"group_id\", \"type\", \"datasource\", \"created_time\", \"location\", "
+        + "\"duration\", \"error_msg\", \"status\"\n"
+        + "FROM tasks\n"
+        + "ORDER BY\n"
+        + "  (CASE \"status\" WHEN 'RUNNING' THEN 4 WHEN 'PENDING' THEN 3 WHEN 'WAITING' THEN 2 ELSE 1 END) DESC,\n"
+        + "  \"created_time\" DESC",
+        nativeQueryContext(plannerStrategy)
+    );
+
+    final Set<String> taskIds = Arrays.stream(result.split("\\n"))
+                                      .map(row -> row.substring(0, row.indexOf(',')))
+                                      .collect(Collectors.toSet());
+    Assertions.assertEquals(
+        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"
+        ),
+        taskIds
+    );
+  }
+
+  private void createTasks(final String suffix, final String datasource, final int count)
+  {
+    for (int i = 0; i < count; i++) {
+      final String taskId = TASK_PREFIX + suffix + "_" + i;
+      cluster.callApi().runTask(new NoopTask(taskId, null, datasource, 1L, 0L, null), overlord);
+    }
+  }
+
+  private void assertTaskFilterQuery(
+      final String whereClause,
+      final Set<String> expectedTaskIds,
+      final String plannerStrategy
+  )
+  {
+    // EmbeddedServiceClient formats SQL with the selected broker address, so escape LIKE wildcards for that step.
+    final String escapedWhereClause = StringUtils.replace(whereClause, "%", "%%");
+    final String result = cluster.runSql(
+        "SELECT task_id FROM sys.tasks WHERE " + escapedWhereClause,
+        nativeQueryContext(plannerStrategy)
+    );
+
+    Assertions.assertEquals(expectedTaskIds, Arrays.stream(result.split("\\n")).collect(Collectors.toSet()));
+  }
+
+  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
+    );
+  }
+}

extensions-contrib/sqlserver-metadata-storage/src/main/java/org/apache/druid/metadata/SQLServerMetadataStorageActionHandler.java

@@ -38,4 +38,10 @@ protected String decorateSqlWithLimit(String sql)
   {
     return "SELECT TOP (:n)" + sql.substring("SELECT".length());
   }
+
+  @Override
+  protected SqlDialect getSqlDialect()
+  {
+    return SqlDialect.SQL_SERVER;
+  }
 }

extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/metadata/MySQLMetadataStorageActionHandler.java

@@ -38,4 +38,10 @@ protected String decorateSqlWithLimit(String sql)
   {
     return sql + " LIMIT :n";
   }
+
+  @Override
+  protected SqlDialect getSqlDialect()
+  {
+    return SqlDialect.MYSQL;
+  }
 }

extensions-core/postgresql-metadata-storage/src/main/java/org/apache/druid/metadata/PostgreSQLMetadataStorageActionHandler.java

@@ -39,4 +39,10 @@ protected String decorateSqlWithLimit(String sql)
     return sql + " LIMIT :n";
   }
 
+  @Override
+  protected SqlDialect getSqlDialect()
+  {
+    return SqlDialect.POSTGRESQL;
+  }
+
 }