[fix](meta) do not check replica allocation when replay

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

@@ -905,7 +905,7 @@ public void alterColocateGroup(AlterColocateGroupStmt stmt) throws UserException
     }
 
     private void modifyColocateGroupReplicaAllocation(GroupId groupId, ReplicaAllocation replicaAlloc,
-            Map<Tag, List<List<Long>>> backendsPerBucketSeq, boolean needEditLog) throws UserException {
+            Map<Tag, List<List<Long>>> backendsPerBucketSeq, boolean isReplay) throws UserException {
         ColocateGroupSchema groupSchema = getGroupSchema(groupId);
         if (groupSchema == null) {
             LOG.warn("not found group {}", groupId);
@@ -939,7 +939,7 @@ private void modifyColocateGroupReplicaAllocation(GroupId groupId, ReplicaAlloca
                     origDynamicProperties.put(DynamicPartitionProperty.REPLICATION_ALLOCATION,
                             replicaAlloc.toCreateStmt());
                     Map<String, String> analyzedDynamicPartition = DynamicPartitionUtil.analyzeDynamicPartition(
-                            origDynamicProperties, table, db);
+                            origDynamicProperties, table, db, isReplay);
                     tableProperty.modifyTableProperties(analyzedDynamicPartition);
                     tableProperty.buildDynamicProperty();
                 }
@@ -959,11 +959,11 @@ private void modifyColocateGroupReplicaAllocation(GroupId groupId, ReplicaAlloca
         groupSchema.setReplicaAlloc(replicaAlloc);
         setBackendsPerBucketSeq(groupId, backendsPerBucketSeq);
 
-        if (needEditLog) {
+        if (!isReplay) {
             ColocatePersistInfo info = ColocatePersistInfo.createForModifyReplicaAlloc(groupId,
                     replicaAlloc, backendsPerBucketSeq);
             Env.getCurrentEnv().getEditLog().logColocateModifyRepliaAlloc(info);
         }
-        LOG.info("modify group {} replication allocation to {}, is replay {}", groupId, replicaAlloc, !needEditLog);
+        LOG.info("modify group {} replication allocation to {}, is replay {}", groupId, replicaAlloc, isReplay);
     }
 }

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

@@ -5487,7 +5487,7 @@ public void modifyTableDynamicPartition(Database db, OlapTable table, Map<String
             Map<String, String> origDynamicProperties = tableProperty.getOriginDynamicPartitionProperty();
             origDynamicProperties.putAll(properties);
             Map<String, String> analyzedDynamicPartition = DynamicPartitionUtil.analyzeDynamicPartition(
-                    origDynamicProperties, table, db);
+                    origDynamicProperties, table, db, false);
             tableProperty.modifyTableProperties(analyzedDynamicPartition);
             tableProperty.buildDynamicProperty();
         }

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

@@ -240,8 +240,8 @@ private static void checkReplicationNum(String val, Database db) throws DdlExcep
                 null, false, true);
     }
 
-    private static void checkReplicaAllocation(ReplicaAllocation replicaAlloc, int hotPartitionNum,
-            Database db) throws DdlException {
+    private static void checkReplicaAllocation(ReplicaAllocation replicaAlloc, int hotPartitionNum)
+            throws DdlException {
         if (replicaAlloc.getTotalReplicaNum() <= 0) {
             ErrorReport.reportDdlException(ErrorCode.ERROR_DYNAMIC_PARTITION_REPLICATION_NUM_ZERO);
         }
@@ -547,8 +547,9 @@ public static void partitionIntervalCompatible(String dynamicUnit, ArrayList<Exp
     }
 
     // Analyze all properties to check their validation
+    // ATTN, should not throw any exception when isReplay is true.
     public static Map<String, String> analyzeDynamicPartition(Map<String, String> properties,
-            OlapTable olapTable, Database db) throws UserException {
+            OlapTable olapTable, Database db, boolean isReplay) throws UserException {
         // properties should not be empty, check properties before call this function
         Map<String, String> analyzedProperties = new HashMap<>();
         if (properties.containsKey(DynamicPartitionProperty.TIME_UNIT)) {
@@ -624,7 +625,6 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
         // If create_history_partition is false, history partition is not considered.
         // If create_history_partition is true, will pre-create history partition according the valid value from
         // start and history_partition_num.
-        //
         long expectCreatePartitionNum = 0;
         if (!createHistoryPartition) {
             start = 0;
@@ -640,7 +640,7 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
         }
         expectCreatePartitionNum = (long) end - start;
 
-        if (hasEnd && (expectCreatePartitionNum > Config.max_dynamic_partition_num)
+        if (!isReplay && hasEnd && (expectCreatePartitionNum > Config.max_dynamic_partition_num)
                 && Boolean.parseBoolean(analyzedProperties.getOrDefault(DynamicPartitionProperty.ENABLE, "true"))) {
             throw new DdlException("Too many dynamic partitions: "
                     + expectCreatePartitionNum + ". Limit: " + Config.max_dynamic_partition_num);
@@ -693,11 +693,14 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
         } else {
             replicaAlloc = olapTable.getDefaultReplicaAllocation();
         }
-        if (olapTable.getMinLoadReplicaNum() > replicaAlloc.getTotalReplicaNum()) {
+        if (!isReplay && olapTable.getMinLoadReplicaNum() > replicaAlloc.getTotalReplicaNum()) {
             throw new DdlException("Failed to check min load replica num [" + olapTable.getMinLoadReplicaNum()
                     + "]  <= dynamic partition replica num [" + replicaAlloc.getTotalReplicaNum() + "]");
         }
-        checkReplicaAllocation(replicaAlloc, hotPartitionNum, db);
+
+        if (!isReplay) {
+            checkReplicaAllocation(replicaAlloc, hotPartitionNum);
+        }
 
         if (properties.containsKey(DynamicPartitionProperty.RESERVED_HISTORY_PERIODS)) {
             String reservedHistoryPeriods = properties.get(DynamicPartitionProperty.RESERVED_HISTORY_PERIODS);
@@ -767,7 +770,7 @@ public static void checkAndSetDynamicPartitionProperty(OlapTable olapTable, Map<
             Database db) throws UserException {
         if (DynamicPartitionUtil.checkInputDynamicPartitionProperties(properties, olapTable)) {
             Map<String, String> dynamicPartitionProperties =
-                    DynamicPartitionUtil.analyzeDynamicPartition(properties, olapTable, db);
+                    DynamicPartitionUtil.analyzeDynamicPartition(properties, olapTable, db, false);
             TableProperty tableProperty = olapTable.getTableProperty();
             if (tableProperty != null) {
                 tableProperty.modifyTableProperties(dynamicPartitionProperties);