[FE](debug) Add selected BE logics

Author: Gabriel39

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

@@ -63,6 +63,7 @@
 import java.util.Arrays;
 import java.util.BitSet;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -277,6 +278,8 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String BLOCK_ENCRYPTION_MODE = "block_encryption_mode";
 
+    public static final String SELECTED_BACKEND_IDS = "selected_backend_ids";
+
     public static final String AUTO_BROADCAST_JOIN_THRESHOLD = "auto_broadcast_join_threshold";
 
     public static final String ENABLE_PROJECTION = "enable_projection";
@@ -1241,6 +1244,9 @@ public void setEnableLeftZigZag(boolean enableLeftZigZag) {
     @VariableMgr.VarAttr(name = ENABLE_NEREIDS_RULES, needForward = true)
     public String enableNereidsRules = "";
 
+    @VariableMgr.VarAttr(name = SELECTED_BACKEND_IDS, needForward = true)
+    public String selectedBackendIds = "";
+
     @VariableMgr.VarAttr(name = ENABLE_NEW_COST_MODEL, needForward = true)
     private boolean enableNewCostModel = false;
 
@@ -3182,6 +3188,18 @@ public boolean getEnableNewCostModel() {
         return this.enableNewCostModel;
     }
 
+    public Set<Long> getSelectedBackendIds() {
+        Set<Long> ids = new HashSet<>();
+        if (selectedBackendIds.isEmpty()) {
+            return ids;
+        }
+        String[] selectedBes = selectedBackendIds.split(",");
+        for (String be : selectedBes) {
+            ids.add(Long.valueOf(be));
+        }
+        return ids;
+    }
+
     public void setDisableNereidsRules(String disableNereidsRules) {
         this.disableNereidsRules = disableNereidsRules;
     }

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

@@ -37,6 +37,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
@@ -131,6 +132,21 @@ public static TNetworkAddress getHost(ImmutableMap<Long, Backend> backends,
         if (backends.isEmpty()) {
             throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG);
         }
+        if (ConnectContext.get() != null
+                && !ConnectContext.get().getSessionVariable().getSelectedBackendIds().isEmpty()) {
+            Set<Long> beIds = ConnectContext.get().getSessionVariable().getSelectedBackendIds();
+            long id = nextId.getAndIncrement() % beIds.size();
+            Long selectedId = beIds.stream().skip(id).filter(e -> backends.containsKey(e)
+                            && isAvailable(backends.get(e))).findFirst().orElse(null);
+            Backend backend = backends.getOrDefault(selectedId, null);
+            if (isAvailable(backend)) {
+                backendIdRef.setRef(selectedId);
+                return new TNetworkAddress(backend.getHost(), backend.getBePort());
+            }
+            // no backend returned
+            throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG
+                    + " BEID: " + selectedId + " host: " + backend.getHost());
+        }
         long id = nextId.getAndIncrement() % backends.size();
         Map.Entry<Long, Backend> backendEntry = backends.entrySet().stream().skip(id).filter(
                 e -> isAvailable(e.getValue())).findFirst().orElse(null);
@@ -148,6 +164,22 @@ public static TNetworkAddress getHost(ImmutableMap<Long, Backend> backends,
                 + getBackendErrorMsg(Lists.newArrayList(backends.keySet()), backends, 3));
     }
 
+    public static TNetworkAddress getHost(ImmutableMap<Long, Backend> backends,
+            Reference<Long> backendIdRef, Long beId)
+            throws UserException {
+        if (backends.isEmpty()) {
+            throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG);
+        }
+        Backend backend = backends.getOrDefault(beId, null);
+        if (backend != null && isAvailable(backend)) {
+            backendIdRef.setRef(beId);
+            return new TNetworkAddress(backend.getHost(), backend.getBePort());
+        }
+        // no backend returned
+        throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG
+                + " BEID: " + beId + " host: " + backend.getHost());
+    }
+
     // get the reason why backends can not be chosen.
     private static String getBackendErrorMsg(List<Long> backendIds, ImmutableMap<Long, Backend> backends, int limit) {
         List<String> res = Lists.newArrayList();