[Enhancement](group commit)Optimize be select for group commit (apache#35558)

1. Streamload and insert into, if batched and sent to the master FE,
should use a consistent BE strategy (previously, insert into reused the
first selected BE, while streamload used round robin). First, a map
<table id, be id> records a fixed be id for a certain table. The first
time a table is imported, a BE is randomly selected, and this table id
and be id are recorded in the map permanently. Subsequently, all data
imported into this table will select the BE corresponding to the table
id recorded in the map. This ensures that batching is maximized to a
single BE.
To address the issue of excessive load on a single BE, a variable
similar to a bvar window is used to monitor the total data volume sent
to a specific BE for a specific table during the batch interval (default
10 seconds). A second map <be id, window variable> is used to track
this. If a new import finds that its corresponding BE's window variable
is less than a certain value (e.g., 1G), the new import continues to be
sent to the corresponding BE according to map1. If it exceeds this
value, the new import is sent to another BE with the smallest window
variable value, and map1 is updated. If every BE exceeds this value, the
one with the smallest value is still chosen. This helps to alleviate
excessive pressure on a single BE.

2. For streamload, if batched and sent to a BE, it will batch directly
on this BE and will commit the transaction at the end of the import. At
this point, a request is sent to the FE, which records the size of this
import and adds it to the window variable.

3. Streamload sent to observer FE, as well as insert into sent to
observer FE, follow the logic in 1 by RPC, passing the table id to the
master FE to obtain the selected be id.

Author: Yukang-Lian

be/src/runtime/group_commit_mgr.cpp

@@ -397,6 +397,13 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
         request.__set_db_id(db_id);
         request.__set_table_id(table_id);
         request.__set_txnId(txn_id);
+        request.__set_groupCommit(true);
+        request.__set_receiveBytes(state->num_bytes_load_total());
+        if (_exec_env->master_info()->__isset.backend_id) {
+            request.__set_backendId(_exec_env->master_info()->backend_id);
+        } else {
+            LOG(WARNING) << "_exec_env->master_info not set backend_id";
+        }
         if (state) {
             request.__set_commitInfos(state->tablet_commit_infos());
         }

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

@@ -0,0 +1,73 @@
+// 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.common.util;
+
+import java.util.concurrent.atomic.AtomicLongArray;
+
+public class SlidingWindowCounter {
+    private final int windowSizeInSeconds;
+    private final int numberOfBuckets;
+    private final AtomicLongArray buckets;
+    private final AtomicLongArray bucketTimestamps;
+
+    public SlidingWindowCounter(int windowSizeInSeconds) {
+        this.windowSizeInSeconds = windowSizeInSeconds;
+        this.numberOfBuckets = windowSizeInSeconds; // Each bucket represents 1 second
+        this.buckets = new AtomicLongArray(numberOfBuckets);
+        this.bucketTimestamps = new AtomicLongArray(numberOfBuckets);
+    }
+
+    private int getCurrentBucketIndex() {
+        long currentTime = System.currentTimeMillis() / 1000; // Current time in seconds
+        return (int) (currentTime % numberOfBuckets);
+    }
+
+    private void updateCurrentBucket() {
+        long currentTime = System.currentTimeMillis() / 1000; // Current time in seconds
+        int currentBucketIndex = getCurrentBucketIndex();
+        long bucketTimestamp = bucketTimestamps.get(currentBucketIndex);
+
+        if (currentTime - bucketTimestamp >= 1) {
+            buckets.set(currentBucketIndex, 0);
+            bucketTimestamps.set(currentBucketIndex, currentTime);
+        }
+    }
+
+    public void add(long value) {
+        updateCurrentBucket();
+        int bucketIndex = getCurrentBucketIndex();
+        buckets.addAndGet(bucketIndex, value);
+    }
+
+    public long get() {
+        updateCurrentBucket();
+        long currentTime = System.currentTimeMillis() / 1000; // Current time in seconds
+        long count = 0;
+
+        for (int i = 0; i < numberOfBuckets; i++) {
+            if (currentTime - bucketTimestamps.get(i) < windowSizeInSeconds) {
+                count += buckets.get(i);
+            }
+        }
+        return count;
+    }
+
+    public String toString() {
+        return String.valueOf(get());
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
@@ -128,11 +129,16 @@ public Object streamLoadWithSql(HttpServletRequest request, HttpServletResponse
         String sql = request.getHeader("sql");
         LOG.info("streaming load sql={}", sql);
         boolean groupCommit = false;
+        long tableId = -1;
         String groupCommitStr = request.getHeader("group_commit");
         if (groupCommitStr != null && groupCommitStr.equalsIgnoreCase("async_mode")) {
             groupCommit = true;
             try {
                 String[] pair = parseDbAndTb(sql);
+                Database db = Env.getCurrentInternalCatalog()
+                        .getDbOrException(pair[0], s -> new TException("database is invalid for dbName: " + s));
+                Table tbl = db.getTableOrException(pair[1], s -> new TException("table is invalid: " + s));
+                tableId = tbl.getId();
                 if (isGroupCommitBlock(pair[0], pair[1])) {
                     String msg = "insert table " + pair[1] + " is blocked on schema change";
                     return new RestBaseResult(msg);
@@ -150,8 +156,7 @@ public Object streamLoadWithSql(HttpServletRequest request, HttpServletResponse
             }
 
             String label = request.getHeader(LABEL_KEY);
-            TNetworkAddress redirectAddr;
-            redirectAddr = selectRedirectBackend(groupCommit);
+            TNetworkAddress redirectAddr = selectRedirectBackend(request, groupCommit, tableId);
 
             LOG.info("redirect load action to destination={}, label: {}",
                     redirectAddr.toString(), label);
@@ -274,7 +279,9 @@ private Object executeWithoutPassword(HttpServletRequest request,
                     return new RestBaseResult(e.getMessage());
                 }
             } else {
-                redirectAddr = selectRedirectBackend(groupCommit);
+                long tableId = ((OlapTable) ((Database) Env.getCurrentEnv().getCurrentCatalog().getDb(dbName)
+                        .get()).getTable(tableName).get()).getId();
+                redirectAddr = selectRedirectBackend(request, groupCommit, tableId);
             }
 
             LOG.info("redirect load action to destination={}, stream: {}, db: {}, tbl: {}, label: {}",
@@ -305,7 +312,7 @@ private Object executeStreamLoad2PC(HttpServletRequest request, String db) {
                 return new RestBaseResult("No transaction operation(\'commit\' or \'abort\') selected.");
             }
 
-            TNetworkAddress redirectAddr = selectRedirectBackend(false);
+            TNetworkAddress redirectAddr = selectRedirectBackend(request, false, -1);
             LOG.info("redirect stream load 2PC action to destination={}, db: {}, txn: {}, operation: {}",
                     redirectAddr.toString(), dbName, request.getHeader(TXN_ID_KEY), txnOperation);
 
@@ -323,12 +330,40 @@ private final synchronized int getLastSelectedBackendIndexAndUpdate() {
         return index;
     }
 
-    private TNetworkAddress selectRedirectBackend(boolean groupCommit) throws LoadException {
+    private String getCloudClusterName(HttpServletRequest request) {
+        String cloudClusterName = request.getHeader(SessionVariable.CLOUD_CLUSTER);
+        if (!Strings.isNullOrEmpty(cloudClusterName)) {
+            return cloudClusterName;
+        }
+
+        cloudClusterName = ConnectContext.get().getCloudCluster();
+        if (!Strings.isNullOrEmpty(cloudClusterName)) {
+            return cloudClusterName;
+        }
+
+        return "";
+    }
+
+    private TNetworkAddress selectRedirectBackend(HttpServletRequest request, boolean groupCommit, long tableId)
+            throws LoadException {
         long debugBackendId = DebugPointUtil.getDebugParamOrDefault("LoadAction.selectRedirectBackend.backendId", -1L);
         if (debugBackendId != -1L) {
             Backend backend = Env.getCurrentSystemInfo().getBackend(debugBackendId);
             return new TNetworkAddress(backend.getHost(), backend.getHttpPort());
         }
+        if (Config.isCloudMode()) {
+            String cloudClusterName = getCloudClusterName(request);
+            if (Strings.isNullOrEmpty(cloudClusterName)) {
+                throw new LoadException("No cloud cluster name selected.");
+            }
+            return selectCloudRedirectBackend(cloudClusterName, request, groupCommit);
+        } else {
+            return selectLocalRedirectBackend(groupCommit, request, tableId);
+        }
+    }
+
+    private TNetworkAddress selectLocalRedirectBackend(boolean groupCommit, HttpServletRequest request, long tableId)
+            throws LoadException {
         Backend backend = null;
         BeSelectionPolicy policy = null;
         String qualifiedUser = ConnectContext.get().getQualifiedUser();
@@ -348,12 +383,17 @@ private TNetworkAddress selectRedirectBackend(boolean groupCommit) throws LoadEx
             throw new LoadException(SystemInfoService.NO_BACKEND_LOAD_AVAILABLE_MSG + ", policy: " + policy);
         }
         if (groupCommit) {
-            for (Long backendId : backendIds) {
-                Backend candidateBe = Env.getCurrentSystemInfo().getBackend(backendId);
-                if (!candidateBe.isDecommissioned()) {
-                    backend = candidateBe;
-                    break;
-                }
+            ConnectContext ctx = new ConnectContext();
+            ctx.setEnv(Env.getCurrentEnv());
+            ctx.setThreadLocalInfo();
+            ctx.setRemoteIP(request.getRemoteAddr());
+            ctx.setThreadLocalInfo();
+
+            try {
+                backend = Env.getCurrentEnv().getGroupCommitManager()
+                        .selectBackendForGroupCommit(tableId, ctx, false);
+            } catch (DdlException e) {
+                throw new RuntimeException(e);
             }
         } else {
             backend = Env.getCurrentSystemInfo().getBackend(backendIds.get(0));
@@ -416,10 +456,10 @@ private Object executeWithClusterToken(HttpServletRequest request, String db,
                 return new RestBaseResult("No label selected.");
             }
 
-            TNetworkAddress redirectAddr = selectRedirectBackend(false);
+            TNetworkAddress redirectAddr = selectRedirectBackend(request, false, -1);
 
             LOG.info("Redirect load action with auth token to destination={},"
-                        + "stream: {}, db: {}, tbl: {}, label: {}",
+                            + "stream: {}, db: {}, tbl: {}, label: {}",
                     redirectAddr.toString(), isStreamLoad, dbName, tableName, label);
 
             URI urlObj = null;