Add a property to limit user instance num.

docs/en/administrator-guide/config/fe_config.md

@@ -1997,3 +1997,10 @@ Load label cleaner will run every *label_clean_interval_second* to clean the out
 Default：30
 
 the transaction will be cleaned after transaction_clean_interval_second seconds if the transaction is visible or aborted  we should make this interval as short as possible and each clean cycle as soon as possible
+
+
+### `default_max_query_instances`
+
+The default value when user property max_query_instances is equal or less than 0. This config is used to limit the max number of instances for a user. This parameter is less than or equal to 0 means unlimited.
+
+The default value is -1。

docs/en/administrator-guide/config/user_property.md

@@ -60,6 +60,10 @@ User-level configuration items will only take effect for the specified users, an
 
     The maximum number of user connections, the default value is 100 In general, this parameter does not need to be changed unless the number of concurrent queries exceeds the default value.
 
+### max_query_instances
+
+    The maximum number of instances that the user can use at a certain point in time, The default value is -1, negative number means use default_max_query_instances config.
+
 ### resource
 
 ### quota

docs/en/sql-reference/sql-statements/Account Management/SET PROPERTY.md

@@ -38,7 +38,8 @@ Importing cluster is only applicable to Baidu internal users.
 key:
 
 Super user rights:
-Max_user_connections: Maximum number of connections.
+max_user_connections: Maximum number of connections.
+max_query_instances: Maximum number of query instance user can use when query.
 resource.cpu_share: cpu resource assignment.
 Load_cluster. {cluster_name}. priority: assigns priority to a specified cluster, which can be HIGH or NORMAL
 
@@ -77,6 +78,9 @@ SET PROPERTY FOR 'jack' 'default_load_cluster' = '{cluster_name}';
 7. Modify the cluster priority of user Jack to HIGH
 SET PROPERTY FOR 'jack' 'load_cluster.{cluster_name}.priority' = 'HIGH';
 
+8. Modify the maximum number of query instance for jack to 3000
+SET PROPERTY FOR 'jack' 'max_query_instances' = '3000';
+
 ## keyword
 SET, PROPERTY
 

docs/zh-CN/administrator-guide/config/fe_config.md

@@ -2003,3 +2003,10 @@ load 标签清理器将每隔 `label_clean_interval_second` 运行一次以清
 默认值：30
 
 如果事务 visible 或者 aborted 状态，事务将在 `transaction_clean_interval_second` 秒后被清除 ，我们应该让这个间隔尽可能短，每个清洁周期都尽快
+
+
+### `default_max_query_instances`
+
+默认值：-1
+
+用户属性max_query_instances小于等于0时，使用该配置，用来限制单个用户同一时刻可使用的查询instance个数。该参数小于等于0表示无限制。

docs/zh-CN/administrator-guide/config/user_property.md

@@ -60,6 +60,10 @@ User 级别的配置项只会对指定用户生效，并不会影响其他用户
 
     用户最大的连接数，默认值为100。一般情况不需要更改该参数，除非查询的并发数超过了默认值。
 
+### max_query_instances
+
+    用户同一时间点可使用的instance个数, 默认是-1，小于等于0将会使用配置default_max_query_instances.
+
 ### resource
 
 ### quota

docs/zh-CN/sql-reference/sql-statements/Account Management/SET PROPERTY.md

@@ -39,6 +39,7 @@ under the License.
 
     超级用户权限:
         max_user_connections: 最大连接数。
+        max_query_instances: 用户同一时间点执行查询可以使用的instance个数。
         resource.cpu_share: cpu资源分配。
         load_cluster.{cluster_name}.priority: 为指定的cluster分配优先级，可以为 HIGH 或 NORMAL
 
@@ -77,6 +78,9 @@ under the License.
     7. 修改用户 jack 的集群优先级为 HIGH
     SET PROPERTY FOR 'jack' 'load_cluster.{cluster_name}.priority' = 'HIGH';
 
+    8. 修改用户jack的查询可用instance个数为3000
+    SET PROPERTY FOR 'jack' 'max_query_instances' = '3000';
+
 ## keyword
     SET, PROPERTY
 

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

@@ -1403,9 +1403,15 @@ public class Config extends ConfigBase {
     @ConfField(mutable = true, masterOnly = true)
     public static int max_dynamic_partition_num = 500;
 
-    /*
+    /**
      * Control the max num of backup/restore job per db
      */
     @ConfField(mutable = true, masterOnly = true)
     public static int max_backup_restore_job_num_per_db = 10;
+
+    /**
+     * Control the default max num of the instance for a user.
+     */
+    @ConfField(mutable = true)
+    public static int default_max_query_instances = -1;
 }

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

@@ -210,6 +210,8 @@ public final class FeMetaVersion {
     public static final int VERSION_98 = 98;
     // add audit steam load and change the serialization backend method to json
     public static final int VERSION_99 = 99;
+    // for max query instance
+    public static final int VERSION_100 = 100;
     // note: when increment meta version, should assign the latest version to VERSION_CURRENT
-    public static final int VERSION_CURRENT = VERSION_99;
+    public static final int VERSION_CURRENT = VERSION_100;
 }

fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java

@@ -32,10 +32,12 @@
 import org.apache.doris.monitor.jvm.JvmService;
 import org.apache.doris.monitor.jvm.JvmStats;
 import org.apache.doris.persist.EditLog;
+import org.apache.doris.qe.QeProcessorImpl;
 import org.apache.doris.service.ExecuteEnv;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.SystemInfoService;
 
+import com.codahale.metrics.Gauge;
 import com.codahale.metrics.Histogram;
 import com.codahale.metrics.MetricRegistry;
 import com.google.common.collect.Sets;
@@ -49,6 +51,7 @@
 import java.util.SortedMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BinaryOperator;
 
 public final class MetricRepo {
     private static final Logger LOG = LogManager.getLogger(MetricRepo.class);
@@ -302,6 +305,16 @@ public Long getValue() {
         HISTO_QUERY_LATENCY = METRIC_REGISTER.histogram(MetricRegistry.name("query", "latency", "ms"));
         HISTO_EDIT_LOG_WRITE_LATENCY = METRIC_REGISTER.histogram(MetricRegistry.name("editlog", "write", "latency", "ms"));
 
+        METRIC_REGISTER.register(MetricRegistry.name("palo", "fe", "query", "max_instances_num_per_user"), (Gauge<Integer>) () -> {
+            try{
+                return ((QeProcessorImpl)QeProcessorImpl.INSTANCE).getInstancesNumPerUser().values().stream()
+                        .reduce(-1, BinaryOperator.maxBy(Integer::compareTo));
+            } catch (Throwable ex) {
+                LOG.warn("Get max_instances_num_per_user error", ex);
+                return -2;
+            }
+        });
+
         // init system metrics
         initSystemMetrics();
 

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/CommonUserProperties.java

@@ -0,0 +1,65 @@
+// 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.doris.mysql.privilege;
+
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Used in
+ */
+public class CommonUserProperties implements Writable {
+    @SerializedName("maxConn")
+    private long maxConn = 100;
+    @SerializedName("maxQueryInstances")
+    private long maxQueryInstances = -1;
+
+    long getMaxConn() {
+        return maxConn;
+    }
+
+    long getMaxQueryInstances() {
+        return maxQueryInstances;
+    }
+
+    void setMaxConn(long maxConn) {
+        this.maxConn = maxConn;
+    }
+
+    void setMaxQueryInstances(long maxQueryInstances) {
+        this.maxQueryInstances = maxQueryInstances;
+    }
+
+    public static CommonUserProperties read(DataInput in) throws IOException {
+        String json = Text.readString(in);
+        return GsonUtils.GSON.fromJson(json, CommonUserProperties.class);
+    }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+        String json = GsonUtils.GSON.toJson(this);
+        Text.writeString(out, json);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/PaloAuth.java

@@ -1048,6 +1048,15 @@ public long getMaxConn(String qualifiedUser) {
         }
     }
 
+    public long getMaxQueryInstances(String qualifiedUser) {
+        readLock();
+        try {
+            return propertyMgr.getMaxQueryInstances(qualifiedUser);
+        } finally {
+            readUnlock();
+        }
+    }
+
     public void getAllDomains(Set<String> allDomains) {
         readLock();
         try {