Support max queued_time limit (prestodb#25589)

Summary:

Support Max queued time limit. Queries queued for more than this threshold 
should fail with EXCEEDED_TIME_LIMIT error code and proper error msg

Differential Revision: D78709808

Author: pgupta2

presto-docs/src/main/sphinx/admin/properties-session.rst

@@ -484,4 +484,14 @@ The corresponding configuration property is :ref:`admin/properties:\`\`query.cli
 * **Default value:** ``1``
 
 This property defines the priority of queries for execution and plays an important role in query admission.
-Queries with higher priority are scheduled first than the ones with lower priority. Higher number indicates higher priority.
+Queries with higher priority are scheduled first than the ones with lower priority. Higher number indicates higher priority.
+
+``query_max_queued_time``
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``Duration``
+* **Default value:** ``100d``
+
+This property can be used to configure how long a query can be queued before its terminated.
+
+The corresponding configuration property is :ref:`admin/properties:\`\`query.max-queued-time\`\``.

presto-docs/src/main/sphinx/admin/properties.rst

@@ -1136,4 +1136,14 @@ Query Manager Properties
 This property can be used to configure how long a query runs without contact
 from the client application, such as the CLI, before it's abandoned.
 
-The corresponding session property is :ref:`admin/properties-session:\`\`query_client_timeout\`\``.
+The corresponding session property is :ref:`admin/properties-session:\`\`query_client_timeout\`\``.
+
+``query.max-queued-time``
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``Duration``
+* **Default value:** ``100d``
+
+This property can be used to configure how long a query can be queued before its terminated.
+
+The corresponding session property is :ref:`admin/properties-session:\`\`query_max_queued_time\`\``.

presto-main-base/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -115,6 +115,7 @@ public final class SystemSessionProperties
     public static final String QUERY_MAX_BROADCAST_MEMORY = "query_max_broadcast_memory";
     public static final String QUERY_MAX_TOTAL_MEMORY = "query_max_total_memory";
     public static final String QUERY_MAX_TOTAL_MEMORY_PER_NODE = "query_max_total_memory_per_node";
+    public static final String QUERY_MAX_QUEUED_TIME = "query_max_queued_time";
     public static final String QUERY_MAX_EXECUTION_TIME = "query_max_execution_time";
     public static final String QUERY_MAX_RUN_TIME = "query_max_run_time";
     public static final String RESOURCE_OVERCOMMIT = "resource_overcommit";
@@ -569,6 +570,15 @@ public SystemSessionProperties(
                         false,
                         value -> Duration.valueOf((String) value),
                         Duration::toString),
+                new PropertyMetadata<>(
+                        QUERY_MAX_QUEUED_TIME,
+                        "Maximum Queued time of a query",
+                        VARCHAR,
+                        Duration.class,
+                        queryManagerConfig.getQueryMaxQueuedTime(),
+                        false,
+                        value -> Duration.valueOf((String) value),
+                        Duration::toString),
                 new PropertyMetadata<>(
                         QUERY_MAX_EXECUTION_TIME,
                         "Maximum execution time of a query",
@@ -2188,6 +2198,11 @@ public static Duration getQueryMaxRunTime(Session session)
         return session.getSystemProperty(QUERY_MAX_RUN_TIME, Duration.class);
     }
 
+    public static Duration getQueryMaxQueuedTime(Session session)
+    {
+        return session.getSystemProperty(QUERY_MAX_QUEUED_TIME, Duration.class);
+    }
+
     public static Duration getQueryMaxExecutionTime(Session session)
     {
         return session.getSystemProperty(QUERY_MAX_EXECUTION_TIME, Duration.class);

presto-main-base/src/main/java/com/facebook/presto/execution/QueryManagerConfig.java

@@ -74,6 +74,7 @@ public class QueryManagerConfig
 
     private String queryExecutionPolicy = "all-at-once";
     private Duration queryMaxRunTime = new Duration(100, TimeUnit.DAYS);
+    private Duration queryMaxQueuedTime = new Duration(100, TimeUnit.DAYS);
     private Duration queryMaxExecutionTime = new Duration(100, TimeUnit.DAYS);
     private Duration queryMaxCpuTime = new Duration(1_000_000_000, TimeUnit.DAYS);
 
@@ -431,6 +432,19 @@ public QueryManagerConfig setQueryMaxRunTime(Duration queryMaxRunTime)
         return this;
     }
 
+    @NotNull
+    public Duration getQueryMaxQueuedTime()
+    {
+        return queryMaxQueuedTime;
+    }
+
+    @Config("query.max-queued-time")
+    public QueryManagerConfig setQueryMaxQueuedTime(Duration queryMaxQueuedTime)
+    {
+        this.queryMaxQueuedTime = queryMaxQueuedTime;
+        return this;
+    }
+
     @NotNull
     public Duration getQueryMaxExecutionTime()
     {

presto-main-base/src/main/java/com/facebook/presto/execution/QueryTracker.java

@@ -43,6 +43,7 @@
 
 import static com.facebook.presto.SystemSessionProperties.getQueryClientTimeout;
 import static com.facebook.presto.SystemSessionProperties.getQueryMaxExecutionTime;
+import static com.facebook.presto.SystemSessionProperties.getQueryMaxQueuedTime;
 import static com.facebook.presto.SystemSessionProperties.getQueryMaxRunTime;
 import static com.facebook.presto.execution.QueryLimit.Source.QUERY;
 import static com.facebook.presto.execution.QueryLimit.Source.RESOURCE_GROUP;
@@ -211,7 +212,7 @@ public long getQueriesKilledDueToTooManyTask()
     }
 
     /**
-     * Enforce query max runtime/execution time limits
+     * Enforce query max runtime/queued/execution time limits
      */
     private void enforceTimeLimits()
     {
@@ -220,13 +221,24 @@ private void enforceTimeLimits()
                 continue;
             }
             Duration queryMaxRunTime = getQueryMaxRunTime(query.getSession());
+            Duration queryMaxQueuedTime = getQueryMaxQueuedTime(query.getSession());
             QueryLimit<Duration> queryMaxExecutionTime = getMinimum(
                     createDurationLimit(getQueryMaxExecutionTime(query.getSession()), QUERY),
                     query.getResourceGroupQueryLimits()
                             .flatMap(ResourceGroupQueryLimits::getExecutionTimeLimit)
                             .map(rgLimit -> createDurationLimit(rgLimit, RESOURCE_GROUP)).orElse(null));
             long executionStartTime = query.getExecutionStartTimeInMillis();
             long createTimeInMillis = query.getCreateTimeInMillis();
+            long queuedTimeInMillis;
+            if (executionStartTime > 0) {
+                queuedTimeInMillis = executionStartTime - createTimeInMillis;
+            }
+            else {
+                queuedTimeInMillis = currentTimeMillis() - createTimeInMillis;
+            }
+            if (queuedTimeInMillis > queryMaxQueuedTime.toMillis()) {
+                query.fail(new PrestoException(EXCEEDED_TIME_LIMIT, "Query exceeded maximum queued time limit of " + queryMaxQueuedTime));
+            }
             if (executionStartTime > 0 && (executionStartTime + queryMaxExecutionTime.getLimit().toMillis()) < currentTimeMillis()) {
                 query.fail(
                         new PrestoException(EXCEEDED_TIME_LIMIT,

presto-main-base/src/test/java/com/facebook/presto/execution/TestQueryManagerConfig.java

@@ -62,6 +62,7 @@ public void testDefaults()
                 .setRemoteTaskMaxCallbackThreads(Runtime.getRuntime().availableProcessors())
                 .setQueryExecutionPolicy("all-at-once")
                 .setQueryMaxRunTime(new Duration(100, TimeUnit.DAYS))
+                .setQueryMaxQueuedTime(new Duration(100, TimeUnit.DAYS))
                 .setQueryMaxExecutionTime(new Duration(100, TimeUnit.DAYS))
                 .setQueryMaxCpuTime(new Duration(1_000_000_000, TimeUnit.DAYS))
                 .setQueryMaxScanRawInputBytes(new DataSize(1000, PETABYTE))
@@ -115,6 +116,7 @@ public void testExplicitPropertyMappings()
                 .put("query.remote-task.max-callback-threads", "11")
                 .put("query.execution-policy", "phased")
                 .put("query.max-run-time", "2h")
+                .put("query.max-queued-time", "1h")
                 .put("query.max-execution-time", "3h")
                 .put("query.max-cpu-time", "2d")
                 .put("query.max-scan-raw-input-bytes", "1MB")
@@ -167,6 +169,7 @@ public void testExplicitPropertyMappings()
                 .setRemoteTaskMaxCallbackThreads(11)
                 .setQueryExecutionPolicy("phased")
                 .setQueryMaxRunTime(new Duration(2, TimeUnit.HOURS))
+                .setQueryMaxQueuedTime(new Duration(1, TimeUnit.HOURS))
                 .setQueryMaxExecutionTime(new Duration(3, TimeUnit.HOURS))
                 .setQueryMaxCpuTime(new Duration(2, TimeUnit.DAYS))
                 .setQueryMaxScanRawInputBytes(new DataSize(1, MEGABYTE))