[feature](fe) Add resource group affinity extension framework

bin/start_fe.sh

@@ -31,6 +31,7 @@ OPTS="$(getopt \
     -l 'daemon' \
     -l 'helper:' \
     -l 'image:' \
+    -l 'local_resource_group:' \
     -l 'version' \
     -l 'metadata_failure_recovery' \
     -l 'recovery_journal_id:' \
@@ -47,6 +48,7 @@ HELPER=''
 IMAGE_PATH=''
 IMAGE_TOOL=''
 OPT_VERSION=''
+declare -a LOCAL_RESOURCE_GROUP_ARGS=()
 declare -a HELPER_ARGS=()
 declare -a METADATA_FAILURE_RECOVERY_ARGS=()
 declare -a RECOVERY_JOURNAL_ID_ARGS=()
@@ -83,6 +85,10 @@ while true; do
         IMAGE_PATH="$2"
         shift 2
         ;;
+    --local_resource_group)
+        LOCAL_RESOURCE_GROUP_ARGS=("--local_resource_group" "$2")
+        shift 2
+        ;;
     --cluster_snapshot)
         CLUSTER_SNAPSHOT_ARGS=("--cluster_snapshot" "$2")
         shift 2
@@ -428,23 +434,23 @@ fi
 
 if [[ "${OPT_VERSION}" != "" ]]; then
     export DORIS_LOG_TO_STDERR=1
-    ${LIMIT:+${LIMIT}} "${JAVA}" org.apache.doris.DorisFE --version
+    ${LIMIT:+${LIMIT}} "${JAVA}" org.apache.doris.DorisFE "${LOCAL_RESOURCE_GROUP_ARGS[@]}" --version
     exit 0
 fi
 
 if [[ "${IMAGE_TOOL}" -eq 1 ]]; then
     if [[ -n "${IMAGE_PATH}" ]]; then
-        ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE -i "${IMAGE_PATH}"
+        ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${LOCAL_RESOURCE_GROUP_ARGS[@]}" -i "${IMAGE_PATH}"
     else
         echo "Internal error, USE IMAGE_TOOL like: ./start_fe.sh --image image_path"
     fi
 elif [[ "${RUN_DAEMON}" -eq 1 ]]; then
-    nohup ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null &
+    nohup ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${LOCAL_RESOURCE_GROUP_ARGS[@]}" "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null &
 elif [[ "${RUN_CONSOLE}" -eq 1 ]]; then
     export DORIS_LOG_TO_STDERR=1
-    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" </dev/null
+    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${LOCAL_RESOURCE_GROUP_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" </dev/null
 else
-    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null
+    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${LOCAL_RESOURCE_GROUP_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null
 fi
 
 if [[ "${OPT_VERSION}" != "" ]]; then

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

@@ -135,6 +135,19 @@ public class Config extends ConfigBase {
             description = {"Whether to check for table lock leaks"})
     public static boolean check_table_lock_leaky = false;
 
+    @ConfField(mutable = false, description = {"当前 FE 节点所属的 Resource Group。可通过命令行参数 "
+            + "`--local_resource_group` 或环境变量 `DORIS_LOCAL_RESOURCE_GROUP` 覆盖。空字符串表示未设置。",
+            "The Resource Group that the current FE node belongs to. It can be overridden by the "
+                    + "`--local_resource_group` command line option or the "
+                    + "`DORIS_LOCAL_RESOURCE_GROUP` environment variable. An empty string means unset."})
+    public static String local_resource_group = "";
+
+    @ConfField(mutable = false, description = {"是否启用基于 location resource tag 的权限检查。关闭后，不再根据"
+            + "用户的 location resource tag 过滤可访问的副本。",
+            "Whether to enable permission checks based on location resource tags. If disabled, "
+                    + "accessible replicas will no longer be filtered by the user's location resource tags."})
+    public static boolean resource_tag_location_check = true;
+
     @ConfField(mutable = true, masterOnly = false,
             description = {"PreparedStatement stmtId starting position, used for testing only"})
     public static long prepared_stmt_start_id = -1;
@@ -1001,6 +1014,14 @@ public class Config extends ConfigBase {
     @ConfField(mutable = true, masterOnly = true)
     public static long tablet_schedule_high_priority_second = 30 * 60;
 
+    @ConfField(mutable = true, masterOnly = true,
+            description = {"是否允许可选 Backend 选择策略参与副本修复 clone 选源。"
+                    + "默认策略为 no-op，不改变修复行为。",
+                    "Whether optional backend selection policies may participate in repair clone source selection. "
+                    + "The default policy is a no-op and does not change repair "
+                    + "behavior."})
+    public static boolean enable_repair_source_backend_selection = true;
+
     /**
      * publish version queue's size in be, report it to fe,
      * if publish task in be exceed direct_publish_limit_number,

fe/fe-core/src/main/java/org/apache/doris/DorisFE.java

@@ -42,6 +42,7 @@
 import org.apache.doris.qe.QeProcessorImpl;
 import org.apache.doris.qe.QeService;
 import org.apache.doris.qe.SimpleScheduler;
+import org.apache.doris.resource.Tag;
 import org.apache.doris.service.ExecuteEnv;
 import org.apache.doris.service.FrontendOptions;
 import org.apache.doris.tls.server.FeServerStarterFactory;
@@ -89,6 +90,7 @@ public class DorisFE {
     private static String LOCK_FILE_PATH;
 
     private static final String LOCK_FILE_NAME = "process.lock";
+    private static final String LOCAL_RESOURCE_GROUP_ENV = "DORIS_LOCAL_RESOURCE_GROUP";
     private static FileChannel processLockFileChannel;
     private static FileLock processFileLock;
 
@@ -136,8 +138,6 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             return;
         }
 
-        CommandLineOptions cmdLineOpts = parseArgs(args);
-
         try {
             // init config
             Config config = new Config();
@@ -146,6 +146,8 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             // Because the path of custom config file is defined in fe.conf
             config.initCustom(Config.custom_config_dir + "/fe_custom.conf");
 
+            CommandLineOptions cmdLineOpts = parseArgs(args);
+
             LdapConfig ldapConfig = new LdapConfig();
             if (new File(dorisHomeDir + "/conf/ldap.conf").exists()) {
                 ldapConfig.init(dorisHomeDir + "/conf/ldap.conf");
@@ -382,6 +384,8 @@ private static CommandLineOptions parseArgs(String[] args) {
         options.addOption("c", "cluster_snapshot", true, "Specify the cluster snapshot json file");
         options.addOption(Option.builder().longOpt(FeConstants.DROP_BACKENDS_KEY)
                 .desc("When this FE becomes MASTER, drop all backends from cluster metadata (destructive)").build());
+        options.addOption(null, "local_resource_group", true,
+                "Specify the local resource group for the current FE");
 
         CommandLine cmd = null;
         try {
@@ -392,6 +396,8 @@ private static CommandLineOptions parseArgs(String[] args) {
             System.exit(-1);
         }
 
+        resolveLocalResourceGroup(cmd);
+
         // version
         if (cmd.hasOption('v') || cmd.hasOption("version")) {
             return new CommandLineOptions(true, "", null, "");
@@ -494,6 +500,32 @@ private static CommandLineOptions parseArgs(String[] args) {
         return new CommandLineOptions(false, null, null, "");
     }
 
+    private static void resolveLocalResourceGroup(CommandLine cmd) {
+        String localResourceGroup = Strings.nullToEmpty(Config.local_resource_group);
+        String source = localResourceGroup.isEmpty() ? "DEFAULT" : "FE_CONF";
+        if (System.getenv().containsKey(LOCAL_RESOURCE_GROUP_ENV)) {
+            localResourceGroup = Strings.nullToEmpty(System.getenv(LOCAL_RESOURCE_GROUP_ENV));
+            source = "ENV";
+        }
+        if (cmd.hasOption("local_resource_group")) {
+            localResourceGroup = Strings.nullToEmpty(cmd.getOptionValue("local_resource_group"));
+            source = "CMDLINE";
+        }
+
+        LOG.info("effective local_resource_group={}, source={}", localResourceGroup, source);
+        Config.local_resource_group = localResourceGroup;
+        if (!localResourceGroup.isEmpty()) {
+            try {
+                Tag.create(Tag.TYPE_LOCATION, localResourceGroup);
+            } catch (Exception e) {
+                String errMsg = "Invalid local_resource_group: " + localResourceGroup;
+                LOG.error(errMsg, e);
+                System.err.println(errMsg);
+                System.exit(-1);
+            }
+        }
+    }
+
     private static void printVersion() {
         LogUtils.stdout("Build version: " + Version.DORIS_BUILD_VERSION);
         LogUtils.stdout("Build time: " + Version.DORIS_BUILD_TIME);

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -1241,6 +1241,7 @@ public void initialize(String[] args) throws Exception {
         // 3. Load image first and replay edits
         this.editLog = new EditLog(nodeName);
         loadImage(this.imageDir); // load image file
+        seedSelfLocalResourceGroup();
         migrateConstraintsFromTables(); // migrate old table-based constraints
         editLog.open(); // open bdb env
         this.globalTransactionMgr.setEditLog(editLog);
@@ -1271,6 +1272,23 @@ public void initialize(String[] args) throws Exception {
         StmtExecutor.initBlockSqlAstNames();
     }
 
+    private static void seedSelfLocalResourceGroup() {
+        Env env = getCurrentEnv();
+        String selfNodeName = env.getNodeName();
+        if (Strings.isNullOrEmpty(selfNodeName)) {
+            LOG.debug("skip seeding local resource group because self node name is not initialized");
+            return;
+        }
+
+        Frontend selfFrontend = env.frontends.get(selfNodeName);
+        if (selfFrontend == null) {
+            LOG.debug("skip seeding local resource group because self frontend {} is not found", selfNodeName);
+            return;
+        }
+
+        selfFrontend.setLocalResourceGroup(Config.local_resource_group);
+    }
+
     // wait until FE is ready.
     public void waitForReady() throws InterruptedException {
         long counter = 0;
@@ -1391,6 +1409,7 @@ protected void getClusterIdAndRole() throws IOException {
                 isFirstTimeStartUp = true;
                 Frontend self = new Frontend(role, nodeName, selfNode.getHost(),
                         selfNode.getPort());
+                self.setLocalResourceGroup(Config.local_resource_group);
                 // Set self alive to true, the BDBEnvironment.getReplicationGroupAdmin() will rely on this to get
                 // helper node, before the heartbeat thread is started.
                 self.setIsAlive(true);
@@ -1873,6 +1892,7 @@ private void transferToMaster() {
 
             MetricRepo.init();
 
+            seedSelfLocalResourceGroup();
             toMasterProgress = "finished";
             canRead.set(true);
             isReady.set(true);

fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java

@@ -230,6 +230,10 @@ public class InternalSchema {
                 ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA.add(new ColumnDef("compute_group",
                 ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA.add(new ColumnDef("backend_selection_preferred_key",
+                ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA.add(new ColumnDef("backend_selection_mode",
+                ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
         // Keep stmt as last column. So that in fe.audit.log, it will be easier to get sql string
         AUDIT_SCHEMA.add(new ColumnDef("stmt",
                 ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));

fe/fe-core/src/main/java/org/apache/doris/clone/TabletSchedCtx.java

@@ -40,6 +40,8 @@
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.persist.ReplicaPersistInfo;
+import org.apache.doris.resource.BackendSelectionPolicy;
+import org.apache.doris.resource.BackendSelectionPolicyFactory;
 import org.apache.doris.resource.Tag;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.SystemInfoService;
@@ -194,6 +196,9 @@ public enum State {
     // tag is only set for BALANCE task, used to identify which workload group this Balance job is in
     private Tag tag;
 
+    private BackendSelectionPolicy.RepairSourceSelectionResult repairSourceSelectionResult =
+            BackendSelectionPolicy.RepairSourceSelectionResult.DISABLED;
+
     private SubCode schedFailedCode;
 
     private boolean isUniqKeyMergeOnWrite = false;
@@ -446,6 +451,10 @@ public long getCopyTimeMs() {
         return copyTimeMs;
     }
 
+    public BackendSelectionPolicy.RepairSourceSelectionResult getRepairSourceSelectionResult() {
+        return repairSourceSelectionResult;
+    }
+
     public long getSrcBackendId() {
         if (srcReplica != null) {
             return srcReplica.getBackendIdWithoutException();
@@ -653,7 +662,13 @@ public void chooseSrcReplica(Map<Long, PathSlot> backendsWorkingSlots, long exce
         // choose a replica which slot is available from candidates.
         // sort replica by version count asc and isUserDrop, so that we prefer to choose replicas with fewer versions
         Collections.sort(candidates, CLONE_SRC_COMPARATOR);
-        for (Replica srcReplica : candidates) {
+
+        BackendSelectionPolicy selectionPolicy = BackendSelectionPolicyFactory.get();
+        boolean selectionEnabled = isRepairSourceSelectionEnabled(selectionPolicy, destBackendId);
+        List<Replica> orderedCandidates = selectionEnabled
+                ? selectionPolicy.orderRepairSourceCandidates(candidates, destBackendId)
+                : candidates;
+        for (Replica srcReplica : orderedCandidates) {
             long replicaBeId = srcReplica.getBackendIdWithoutException();
             PathSlot slot = backendsWorkingSlots.get(replicaBeId);
             if (slot == null) {
@@ -673,12 +688,22 @@ public void chooseSrcReplica(Map<Long, PathSlot> backendsWorkingSlots, long exce
                 continue;
             }
             setSrc(srcReplica);
+            repairSourceSelectionResult = selectionEnabled
+                    ? selectionPolicy.classifyRepairSource(replicaBeId, destBackendId,
+                            tablet.getReplicas(), candidates)
+                    : BackendSelectionPolicy.RepairSourceSelectionResult.DISABLED;
             return;
         }
         throw new SchedException(Status.SCHEDULE_FAILED, SubCode.WAITING_SLOT,
                 "waiting for source replica's slot");
     }
 
+    static boolean isRepairSourceSelectionEnabled(BackendSelectionPolicy selectionPolicy, long destBackendId) {
+        return Config.enable_repair_source_backend_selection
+                && selectionPolicy.isRepairSourceSelectionEnabled()
+                && destBackendId != -1;
+    }
+
     /*
      * Same rules as choosing source replica for supplement.
      * But we need to check that we can not choose the same replica as dest replica,

fe/fe-core/src/main/java/org/apache/doris/clone/TabletScheduler.java

@@ -47,7 +47,9 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.DebugPointUtil;
 import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.metric.MetricRepo;
 import org.apache.doris.persist.ReplicaPersistInfo;
+import org.apache.doris.resource.BackendSelectionPolicy.RepairSourceSelectionResult;
 import org.apache.doris.resource.Tag;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.SystemInfoService;
@@ -1917,6 +1919,8 @@ public boolean finishCloneTask(CloneTask cloneTask, TFinishTaskRequest request)
      * This should only be called when the tablet is down with state FINISHED.
      */
     private void gatherStatistics(TabletSchedCtx tabletCtx) {
+        gatherCloneAzBytes(tabletCtx);
+
         if (tabletCtx.getCopySize() > 0 && tabletCtx.getCopyTimeMs() > 0) {
             if (tabletCtx.getSrcBackendId() != -1 && tabletCtx.getSrcPathHash() != -1) {
                 PathSlot pathSlot = backendsWorkingSlots.get(tabletCtx.getSrcBackendId());
@@ -1945,6 +1949,39 @@ private void gatherStatistics(TabletSchedCtx tabletCtx) {
         lastSlotAdjustTime = System.currentTimeMillis();
     }
 
+    private void gatherCloneAzBytes(TabletSchedCtx tabletCtx) {
+        long bytes = tabletCtx.getCopySize();
+        if (bytes <= 0) {
+            return;
+        }
+        Backend src = infoService.getBackend(tabletCtx.getSrcBackendId());
+        Backend dest = infoService.getBackend(tabletCtx.getDestBackendId());
+        if (src == null || dest == null) {
+            return;
+        }
+        if (src.getLocationTag().equals(dest.getLocationTag())) {
+            if (MetricRepo.COUNTER_CLONE_BYTES_LOCAL_AZ != null) {
+                MetricRepo.COUNTER_CLONE_BYTES_LOCAL_AZ.increase(bytes);
+            }
+            return;
+        }
+
+        if (MetricRepo.COUNTER_CLONE_BYTES_CROSS_AZ != null) {
+            MetricRepo.COUNTER_CLONE_BYTES_CROSS_AZ.increase(bytes);
+        }
+        RepairSourceSelectionResult result = tabletCtx.getRepairSourceSelectionResult();
+        if (result == RepairSourceSelectionResult.FALLBACK_SLOT_FULL) {
+            if (MetricRepo.COUNTER_CLONE_CROSS_AZ_SLOT_FULL != null) {
+                MetricRepo.COUNTER_CLONE_CROSS_AZ_SLOT_FULL.increase(bytes);
+            }
+        } else if (result == RepairSourceSelectionResult.FALLBACK_NO_PREFERRED
+                || result == RepairSourceSelectionResult.FALLBACK_PREFERRED_UNAVAILABLE) {
+            if (MetricRepo.COUNTER_CLONE_CROSS_AZ_NO_LOCAL != null) {
+                MetricRepo.COUNTER_CLONE_CROSS_AZ_NO_LOCAL.increase(bytes);
+            }
+        }
+    }
+
     /**
      * handle tablets which are running.
      * We should finished the task if

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

@@ -163,6 +163,7 @@ public static void getFrontendsInfo(Env env, List<List<String>> infos, String cu
             info.add(fe.getVersion());
             // To indicate which FE we currently connected
             info.add(fe.getHost().equals(selfNode) ? "Yes" : "No");
+            info.add(fe.getLocalResourceGroup());
             info.add(TimeUtils.longToTimeString(fe.getLiveSince()));
 
             infos.add(info);