Support max queued_time limit (#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``
+
+Use to configure how long a query can be queued before it is 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``
+
+Use to configure how long a query can be queued before it is 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/dispatcher/FailedDispatchQuery.java

@@ -156,6 +156,12 @@ public long getCreateTimeInMillis()
         return basicQueryInfo.getQueryStats().getCreateTimeInMillis();
     }
 
+    @Override
+    public Duration getQueuedTime()
+    {
+        return basicQueryInfo.getQueryStats().getQueuedTime();
+    }
+
     @Override
     public long getExecutionStartTimeInMillis()
     {

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

@@ -145,6 +145,12 @@ public long getCreateTimeInMillis()
         return stateMachine.getCreateTimeInMillis();
     }
 
+    @Override
+    public Duration getQueuedTime()
+    {
+        return stateMachine.getQueuedTime();
+    }
+
     @Override
     public long getExecutionStartTimeInMillis()
     {

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

@@ -124,6 +124,12 @@ public long getCreateTimeInMillis()
         return stateMachine.getCreateTimeInMillis();
     }
 
+    @Override
+    public Duration getQueuedTime()
+    {
+        return stateMachine.getQueuedTime();
+    }
+
     @Override
     public long getExecutionStartTimeInMillis()
     {

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/QueryStateMachine.java

@@ -57,6 +57,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import io.airlift.units.Duration;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
@@ -1025,6 +1026,11 @@ public long getCreateTimeInMillis()
         return queryStateTimer.getCreateTimeInMillis();
     }
 
+    public Duration getQueuedTime()
+    {
+        return queryStateTimer.getQueuedTime();
+    }
+
     public long getExecutionStartTimeInMillis()
     {
         return queryStateTimer.getExecutionStartTimeInMillis();

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,22 +212,28 @@ public long getQueriesKilledDueToTooManyTask()
     }
 
     /**
-     * Enforce query max runtime/execution time limits
+     * Enforce query max runtime/queued/execution time limits
      */
-    private void enforceTimeLimits()
+    @VisibleForTesting
+    void enforceTimeLimits()
     {
         for (T query : queries.values()) {
             if (query.isDone()) {
                 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 = query.getQueuedTime().toMillis();
+            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,
@@ -399,6 +406,8 @@ public interface TrackedQuery
 
         long getCreateTimeInMillis();
 
+        Duration getQueuedTime();
+
         long getExecutionStartTimeInMillis();
 
         long getLastHeartbeatInMillis();

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

@@ -341,6 +341,12 @@ public long getCreateTimeInMillis()
         return stateMachine.getCreateTimeInMillis();
     }
 
+    @Override
+    public Duration getQueuedTime()
+    {
+        return stateMachine.getQueuedTime();
+    }
+
     /**
      * For a query that has started executing, returns the timestamp when this query started executing
      * Otherwise returns a {@link Optional#empty()}