[feature](expr) Support kill query by query_id

docs/en/docs/sql-manual/sql-reference/Database-Administration-Statements/KILL.md

@@ -39,9 +39,18 @@ The thread process list identifier can be determined from the ID column of the I
 grammar:
 
 ```sql
-KILL [CONNECTION | QUERY] processlist_id
+KILL [CONNECTION] processlist_id
 ````
 
+In addition, you can also use processlist_id or query_id terminates the executing query command
+
+grammar:
+
+```sql
+KILL QUERY processlist_id | query_id
+````
+
+
 ### Example
 
 ### Keywords

docs/zh-CN/docs/sql-manual/sql-reference/Database-Administration-Statements/KILL.md

@@ -39,9 +39,19 @@ KILL
 语法：
 
 ```sql
-KILL [CONNECTION | QUERY] processlist_id
+KILL [CONNECTION] processlist_id
 ```
 
+除此之外，您还可以使用 processlist_id 或者 query_id 终止正在执行的查询命令
+
+语法：
+
+```sql
+KILL QUERY processlist_id | query_id
+```
+
+
+
 ### Example
 
 ### Keywords

fe/fe-core/src/main/cup/sql_parser.cup

@@ -4906,6 +4906,10 @@ kill_stmt ::=
     {:
         RESULT = new KillStmt(false, value.intValue());
     :}
+    | KW_KILL KW_QUERY STRING_LITERAL:value
+    {:
+        RESULT = new KillStmt(value);
+    :}
     ;
 
 kill_analysis_job_stmt ::=

fe/fe-core/src/main/java/org/apache/doris/analysis/KillStmt.java

@@ -25,10 +25,18 @@
 public class KillStmt extends StatementBase {
     private final boolean isConnectionKill;
     private final int connectionId;
+    private final String queryId;
 
     public KillStmt(boolean isConnectionKill, int connectionId) {
         this.isConnectionKill = isConnectionKill;
         this.connectionId = connectionId;
+        this.queryId = "";
+    }
+
+    public KillStmt(String queryId) {
+        this.isConnectionKill = false;
+        this.connectionId = -1;
+        this.queryId = queryId;
     }
 
     public boolean isConnectionKill() {
@@ -39,6 +47,10 @@ public int getConnectionId() {
         return connectionId;
     }
 
+    public String getQueryId() {
+        return queryId;
+    }
+
     @Override
     public void analyze(Analyzer analyzer) {
         // No operation.

fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java

@@ -43,6 +43,7 @@ public enum ErrorCode {
     ERR_NO_SUCH_THREAD(1094, new byte[]{'H', 'Y', '0', '0', '0'}, "Unknown thread id: %d"),
     ERR_KILL_DENIED_ERROR(1095, new byte[]{'H', 'Y', '0', '0', '0'}, "You are not owner of thread %d"),
     ERR_NO_TABLES_USED(1096, new byte[]{'H', 'Y', '0', '0', '0'}, "No tables used"),
+    ERR_NO_SUCH_QUERY(1097, new byte[]{'H', 'Y', '0', '0', '0'}, "Unknown query id: %s"),
     ERR_WRONG_DB_NAME(1102, new byte[]{'4', '2', '0', '0', '0'}, "Incorrect database name '%s'"),
     ERR_WRONG_TABLE_NAME(1103, new byte[]{'4', '2', '0', '0', '0'}, "Incorrect table name '%s'. Table name regex is '%s'"),
     ERR_TOO_BIG_SELECT(1104, new byte[]{'4', '2', '0', '0', '0'}, "The SELECT would examine more than MAX_JOIN_SIZE "

fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java

@@ -126,6 +126,15 @@ public ConnectContext getContext(int connectionId) {
         return connectionMap.get(connectionId);
     }
 
+    public ConnectContext getContextWithQueryId(String queryId) {
+        for (ConnectContext context : connectionMap.values()) {
+            if (queryId.equals(DebugUtil.printId(context.queryId))) {
+                return context;
+            }
+        }
+        return null;
+    }
+
     public ConnectContext getContext(String flightToken) {
         if (flightToken2ConnectionId.containsKey(flightToken)) {
             int connectionId = flightToken2ConnectionId.get(flightToken);

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -1242,11 +1242,21 @@ public void cancel() {
     // Handle kill statement.
     private void handleKill() throws DdlException {
         KillStmt killStmt = (KillStmt) parsedStmt;
+        ConnectContext killCtx = null;
         int id = killStmt.getConnectionId();
-        ConnectContext killCtx = context.getConnectScheduler().getContext(id);
-        if (killCtx == null) {
-            ErrorReport.reportDdlException(ErrorCode.ERR_NO_SUCH_THREAD, id);
+        String queryId = killStmt.getQueryId();
+        if (id == -1) {
+            killCtx = context.getConnectScheduler().getContextWithQueryId(queryId);
+            if (killCtx == null) {
+                ErrorReport.reportDdlException(ErrorCode.ERR_NO_SUCH_QUERY, queryId);
+            }
+        } else {
+            killCtx = context.getConnectScheduler().getContext(id);
+            if (killCtx == null) {
+                ErrorReport.reportDdlException(ErrorCode.ERR_NO_SUCH_THREAD, id);
+            }
         }
+
         if (context == killCtx) {
             // Suicide
             context.setKilled();