[improvement](resource-tag) Add Backend tag location check (#22670)

Add Backend tag location check.
Avoid user set a bad backend tag, cause create table and dynamic partitions failed.
For example, the default value for all backends tag is default, When setting the replication_allocation of a table, user use the following command: ALTER TABLE example_db.mysql_table SET ("replication_allocation" = "tag.location.tag1: 1");, it can set success, but tag1 is not exist, cause dynamic partition can't create.

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

@@ -31,10 +31,12 @@
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
 import org.apache.doris.datasource.CatalogMgr;
 import org.apache.doris.policy.Policy;
 import org.apache.doris.policy.StoragePolicy;
 import org.apache.doris.resource.Tag;
+import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.thrift.TCompressionType;
 import org.apache.doris.thrift.TSortType;
 import org.apache.doris.thrift.TStorageFormat;
@@ -981,6 +983,16 @@ public static ReplicaAllocation analyzeReplicaAllocation(Map<String, String> pro
             Short replicationNum = Short.valueOf(parts[1]);
             replicaAlloc.put(Tag.create(Tag.TYPE_LOCATION, locationVal), replicationNum);
             totalReplicaNum += replicationNum;
+
+            // Check if the current backends satisfy the ReplicaAllocation condition,
+            // to avoid user set it success but failed to create table or dynamic partitions
+            try {
+                SystemInfoService systemInfoService = Env.getCurrentSystemInfo();
+                systemInfoService.selectBackendIdsForReplicaCreation(
+                        replicaAlloc, null, false, true);
+            } catch (DdlException ddlException) {
+                throw new AnalysisException(ddlException.getMessage());
+            }
         }
         if (totalReplicaNum < Config.min_replication_num_per_tablet
                 || totalReplicaNum > Config.max_replication_num_per_tablet) {

fe/fe-core/src/test/java/org/apache/doris/catalog/ModifyBackendTest.java

@@ -21,6 +21,7 @@
 import org.apache.doris.analysis.AlterTableStmt;
 import org.apache.doris.analysis.CreateDbStmt;
 import org.apache.doris.analysis.CreateTableStmt;
+import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ExceptionChecker;
 import org.apache.doris.qe.ConnectContext;
@@ -134,23 +135,27 @@ public void testModifyBackendTag() throws Exception {
         Assert.assertEquals("tag.location.default: 3", tblProperties.get("default.replication_allocation"));
 
         // modify default replica
-        String alterStr = "alter table test.tbl4 set ('default.replication_allocation' = 'tag.location.zonex:1')";
+        String alterStr = "alter table test.tbl4 set ('default.replication_allocation' = 'tag.location.zone1:1')";
         AlterTableStmt alterStmt = (AlterTableStmt) UtFrameUtils.parseAndAnalyzeStmt(alterStr, connectContext);
         ExceptionChecker.expectThrowsNoException(() -> DdlExecutor.execute(Env.getCurrentEnv(), alterStmt));
         defaultAlloc = tbl.getDefaultReplicaAllocation();
         ReplicaAllocation expectedAlloc = new ReplicaAllocation();
-        expectedAlloc.put(Tag.create(Tag.TYPE_LOCATION, "zonex"), (short) 1);
+        expectedAlloc.put(Tag.create(Tag.TYPE_LOCATION, "zone1"), (short) 1);
         Assert.assertEquals(expectedAlloc, defaultAlloc);
         tblProperties = tableProperty.getProperties();
         Assert.assertTrue(tblProperties.containsKey("default.replication_allocation"));
 
         // modify partition replica with wrong zone
+        // it will fail because of we check tag location during the analysis process, so we check AnalysisException
         String partName = tbl.getPartitionNames().stream().findFirst().get();
-        alterStr = "alter table test.tbl4 modify partition " + partName
+        String wrongAlterStr = "alter table test.tbl4 modify partition " + partName
                 + " set ('replication_allocation' = 'tag.location.zonex:1')";
-        AlterTableStmt alterStmt2 = (AlterTableStmt) UtFrameUtils.parseAndAnalyzeStmt(alterStr, connectContext);
-        ExceptionChecker.expectThrowsWithMsg(DdlException.class, "Failed to find enough host with tag",
-                () -> DdlExecutor.execute(Env.getCurrentEnv(), alterStmt2));
+        ExceptionChecker.expectThrowsWithMsg(AnalysisException.class, "errCode = 2, detailMessage = "
+                + "errCode = 2, detailMessage = Failed to find enough backend, "
+                + "please check the replication num,replication tag and storage medium.\n"
+                + "Create failed replications:\n"
+                + "replication tag: {\"location\" : \"zonex\"}, replication num: 1, storage medium: null",
+                () -> UtFrameUtils.parseAndAnalyzeStmt(wrongAlterStr, connectContext));
         tblProperties = tableProperty.getProperties();
         Assert.assertTrue(tblProperties.containsKey("default.replication_allocation"));
 

fe/fe-core/src/test/java/org/apache/doris/catalog/ReplicaAllocationTest.java

@@ -18,14 +18,21 @@
 package org.apache.doris.catalog;
 
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ExceptionChecker;
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.util.PropertyAnalyzer;
 import org.apache.doris.meta.MetaContext;
 import org.apache.doris.resource.Tag;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TStorageMedium;
 
 import com.google.common.collect.Maps;
+import mockit.Delegate;
+import mockit.Expectations;
+import mockit.Mocked;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 
 import java.io.DataInputStream;
@@ -34,9 +41,27 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.List;
 import java.util.Map;
 
 public class ReplicaAllocationTest {
+    @Mocked
+    SystemInfoService systemInfoService;
+
+    @Before
+    public void setUp() throws DdlException {
+        new Expectations() {
+            {
+                systemInfoService.selectBackendIdsForReplicaCreation((ReplicaAllocation) any, (TStorageMedium) any, false, true);
+                minTimes = 0;
+                result = new Delegate() {
+                    Map<Tag, List<Long>> selectBackendIdsForReplicaCreation() {
+                        return Maps.newHashMap();
+                    }
+                };
+            }
+        };
+    }
 
     @Test
     public void testNormal() throws AnalysisException {

fe/fe-core/src/test/java/org/apache/doris/clone/TabletRepairAndBalanceTest.java

@@ -255,6 +255,7 @@ public void test() throws Exception {
         ExceptionChecker.expectThrows(DdlException.class, () -> createTable(createStr));
 
         // nodes of zone2 not enough, create will fail
+        // it will fail because of we check tag location during the analysis process, so we check AnalysisException
         String createStr2 = "create table test.tbl1\n"
                 + "(k1 date, k2 int)\n"
                 + "partition by range(k1)\n"
@@ -268,7 +269,7 @@ public void test() throws Exception {
                 + "(\n"
                 + "    \"replication_allocation\" = \"tag.location.zone1: 2, tag.location.zone2: 3\"\n"
                 + ")";
-        ExceptionChecker.expectThrows(DdlException.class, () -> createTable(createStr2));
+        ExceptionChecker.expectThrows(AnalysisException.class, () -> createTable(createStr2));
 
         // normal, create success
         String createStr3 = "create table test.tbl1\n"
@@ -291,7 +292,7 @@ public void test() throws Exception {
         // alter table's replica allocation failed, tag not enough
         String alterStr = "alter table test.tbl1"
                 + " set (\"replication_allocation\" = \"tag.location.zone1: 2, tag.location.zone2: 3\");";
-        ExceptionChecker.expectThrows(DdlException.class, () -> alterTable(alterStr));
+        ExceptionChecker.expectThrows(AnalysisException.class, () -> alterTable(alterStr));
         ReplicaAllocation tblReplicaAlloc = tbl.getDefaultReplicaAllocation();
         Assert.assertEquals(3, tblReplicaAlloc.getTotalReplicaNum());
         Assert.assertEquals(Short.valueOf((short) 2), tblReplicaAlloc.getReplicaNumByTag(tag1));
@@ -417,17 +418,17 @@ public void test() throws Exception {
         //      p1: zone1:1, zone2:2
         //      p2,p2: zone1:2, zone2:1
 
-        // change tbl1's default replica allocation to zone1:4, which is allowed
-        String alterStr3 = "alter table test.tbl1 set ('default.replication_allocation' = 'tag.location.zone1:4')";
+        // change tbl1's default replica allocation to zone1:3, which is allowed
+        String alterStr3 = "alter table test.tbl1 set ('default.replication_allocation' = 'tag.location.zone1:3')";
         ExceptionChecker.expectThrowsNoException(() -> alterTable(alterStr3));
 
         // change tbl1's p1's replica allocation to zone1:4, which is forbidden
         String alterStr4 = "alter table test.tbl1 modify partition p1"
                 + " set ('replication_allocation' = 'tag.location.zone1:4')";
-        ExceptionChecker.expectThrows(DdlException.class, () -> alterTable(alterStr4));
+        ExceptionChecker.expectThrows(AnalysisException.class, () -> alterTable(alterStr4));
 
-        // change col_tbl1's default replica allocation to zone2:4, which is allowed
-        String alterStr5 = "alter table test.col_tbl1 set ('default.replication_allocation' = 'tag.location.zone2:4')";
+        // change col_tbl1's default replica allocation to zone2:2, which is allowed
+        String alterStr5 = "alter table test.col_tbl1 set ('default.replication_allocation' = 'tag.location.zone2:2')";
         ExceptionChecker.expectThrowsNoException(() -> alterTable(alterStr5));
 
         // Drop all tables