[fix](cloud)Fix create dynamic table race with insert overwrite (apache#59489)

1. Thread 1 executed a DROP TABLE operation, followed by a CREATE TABLE
operation, initializing a dynamic partition (Partition A) in memory.
2. Due to an optimization in the cloud, batch partition edit logs were
attempted; however, the edit log for Partition A was not yet
synchronized to the followers.
3. Thread 2 began an INSERT INTO ... OVERWRITE PARTITION (*) operation,
identifying Partition A in memory and writing a replace partition edit
log.
4. In follower, the replay thread attempted to replay the replace
partition edit log but found a dependency on Partition A, which was
absent from memory, resulting in an exception and subsequent
termination.

Fix
```
2025-12-30 16:10:42,774 ERROR (replayer|123) [EditLog.loadJournal():1445] replay Operation Type 210, log id: 1910
java.lang.NullPointerException
        at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:906)
        at org.apache.doris.catalog.OlapTable.checkPartition(OlapTable.java:2835)
        at org.apache.doris.catalog.OlapTable.replaceTempPartitions(OlapTable.java:2799)
        at org.apache.doris.catalog.Env.replayReplaceTempPartition(Env.java:6750)
        at org.apache.doris.persist.EditLog.loadJournal(EditLog.java:986)
        at org.apache.doris.catalog.Env.replayJournal(Env.java:3100)
        at org.apache.doris.catalog.Env$4.runOneCycle(Env.java:2862)
        at org.apache.doris.common.util.Daemon.run(Daemon.java:119)
2025-12-30 16:10:42,775 INFO (Thread-0|32) [DorisFE.lambda$start$0():159] Received shutdown signal, starting graceful shutdown...
2025-12-30 16:10:42,776 INFO (Thread-0|32) [DorisFE.gracefulShutdown():639] graceful shutdown finished
```

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

None

- Test <!-- At least one of them must be included. -->
    - [x] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: deardeng

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

@@ -3489,14 +3489,17 @@ public boolean createTable(CreateTableInfo createTableInfo) throws UserException
      * @param isCreateTable this call is for creating table
      * @param generatedPartitionId the preset partition id for the partition to add
      * @param writeEditLog whether to write an edit log for this addition
-     * @return PartitionPersistInfo to be written to editlog. It may be null if no partitions added.
+     * @batchPartitions output parameter, used to batch write edit log outside this function, can be null.
+     * first is editlog PartitionPersistInfo, second is the added Partition
      * @throws DdlException
      */
-    public PartitionPersistInfo addPartition(Database db, String tableName, AddPartitionClause addPartitionClause,
+    public void addPartition(Database db, String tableName, AddPartitionClause addPartitionClause,
                                              boolean isCreateTable, long generatedPartitionId,
-                                             boolean writeEditLog) throws DdlException {
-        return getInternalCatalog().addPartition(db, tableName, addPartitionClause,
-            isCreateTable, generatedPartitionId, writeEditLog);
+                                             boolean writeEditLog,
+                                             List<Pair<PartitionPersistInfo, Partition>> batchPartitions)
+            throws DdlException {
+        getInternalCatalog().addPartition(db, tableName, addPartitionClause,
+                isCreateTable, generatedPartitionId, writeEditLog, batchPartitions);
     }
 
     public void addMultiPartitions(Database db, String tableName, AlterMultiPartitionClause multiPartitionClause)

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

@@ -776,7 +776,7 @@ public void executeDynamicPartition(Collection<Pair<Long, Long>> dynamicPartitio
                 cloudBatchBeforeCreatePartitions(executeFirstTime, addPartitionClauses, olapTable, indexIds,
                         db, tableName, generatedPartitionIds);
 
-                List<PartitionPersistInfo> partsInfo = new ArrayList<>();
+                List<Pair<PartitionPersistInfo, Partition>> batchPartsInfo = new ArrayList<>();
                 for (int i = 0; i < addPartitionClauses.size(); i++) {
                     try {
                         boolean needWriteEditLog = true;
@@ -785,15 +785,10 @@ public void executeDynamicPartition(Collection<Pair<Long, Long>> dynamicPartitio
                         if (Config.isCloudMode()) {
                             needWriteEditLog = !executeFirstTime;
                         }
-                        PartitionPersistInfo info =
-                                Env.getCurrentEnv().addPartition(db, tableName, addPartitionClauses.get(i),
+                        Env.getCurrentEnv().addPartition(db, tableName, addPartitionClauses.get(i),
                                     executeFirstTime,
                                     executeFirstTime && Config.isCloudMode() ? generatedPartitionIds.get(i) : 0,
-                                    needWriteEditLog);
-                        if (info == null) {
-                            throw new Exception("null persisted partition returned");
-                        }
-                        partsInfo.add(info);
+                                    needWriteEditLog, batchPartsInfo);
                         clearCreatePartitionFailedMsg(olapTable.getId());
                     } catch (Exception e) {
                         recordCreatePartitionFailedMsg(db.getFullName(), tableName, e.getMessage(), olapTable.getId());
@@ -804,7 +799,7 @@ public void executeDynamicPartition(Collection<Pair<Long, Long>> dynamicPartitio
                         }
                     }
                 }
-                cloudBatchAfterCreatePartitions(executeFirstTime, partsInfo,
+                cloudBatchAfterCreatePartitions(executeFirstTime, batchPartsInfo,
                         addPartitionClauses, db, olapTable, indexIds, tableName);
 
                 // ATTN: Breaking up dynamic partition table scheduling, consuming peak CPU consumption
@@ -824,15 +819,16 @@ public void executeDynamicPartition(Collection<Pair<Long, Long>> dynamicPartitio
         }
     }
 
-    private void cloudBatchAfterCreatePartitions(boolean executeFirstTime, List<PartitionPersistInfo> partsInfo,
-                                                       ArrayList<AddPartitionClause> addPartitionClauses, Database db,
-                                                       OlapTable olapTable, List<Long> indexIds,
-                                                       String tableName) throws DdlException {
+    private void cloudBatchAfterCreatePartitions(boolean executeFirstTime,
+                                                 List<Pair<PartitionPersistInfo, Partition>> batchPartsInfo,
+                                                 ArrayList<AddPartitionClause> addPartitionClauses, Database db,
+                                                 OlapTable olapTable, List<Long> indexIds,
+                                                 String tableName) throws DdlException {
         if (Config.isNotCloudMode()) {
             return;
         }
-        List<Long> succeedPartitionIds = partsInfo.stream().map(partitionPersistInfo
-                -> partitionPersistInfo.getPartition().getId()).collect(Collectors.toList());
+        List<Long> succeedPartitionIds = batchPartsInfo.stream().map(partitionInfo
+                -> partitionInfo.first.getPartition().getId()).collect(Collectors.toList());
         if (!executeFirstTime || addPartitionClauses.isEmpty()) {
             LOG.info("cloud commit rpc in batch, {}-{}", !executeFirstTime, addPartitionClauses.size());
             return;
@@ -849,28 +845,56 @@ private void cloudBatchAfterCreatePartitions(boolean executeFirstTime, List<Part
                     succeedPartitionIds, indexIds, true /* isCreateTable */, false /* isBatchCommit */);
             LOG.info("begin write edit log to add partitions in batch, "
                     + "numPartitions: {}, db: {}, table: {}, tableId: {}",
-                    partsInfo.size(), db.getFullName(), tableName, olapTable.getId());
+                    batchPartsInfo.size(), db.getFullName(), tableName, olapTable.getId());
             // ATTN: here, edit log must after commit cloud partition,
             // prevent commit RPC failure from causing data loss
             if (DebugPointUtil.isEnable("FE.DynamicPartitionScheduler.before.logEditPartitions")) {
                 LOG.info("debug point FE.DynamicPartitionScheduler.before.logEditPartitions, throw e");
                 // committed, but not log edit
                 throw new Exception("debug point FE.DynamicPartitionScheduler.before.commitCloudPartition");
             }
-            for (int i = 0; i < partsInfo.size(); i++) {
-                Env.getCurrentEnv().getEditLog().logAddPartition(partsInfo.get(i));
-                if (DebugPointUtil.isEnable("FE.DynamicPartitionScheduler.in.logEditPartitions")) {
-                    if (i == partsInfo.size() / 2) {
-                        LOG.info("debug point FE.DynamicPartitionScheduler.in.logEditPartitions, throw e");
-                        // committed, but log some edit, others failed
-                        throw new Exception("debug point FE.DynamicPartitionScheduler"
-                            + ".in.commitCloudPartition");
+
+            for (int i = 0; i < batchPartsInfo.size(); i++) {
+                // get table write lock to add partition, edit log and modify table state must be atomic
+                olapTable.writeLockOrDdlException();
+                try {
+                    boolean isTempPartition = addPartitionOps.get(i).isTempPartition();
+                    Partition toAddPartition = batchPartsInfo.get(i).second;
+                    String partitionName = toAddPartition.getName();
+                    // ATTN: Check here to see if the newly created dynamic
+                    // partition has already been added by another process.
+                    // If it has, do not add this dynamic partition again,
+                    // and call `onErasePartition` to clean up any remaining information.
+                    Partition checkIsAdded = olapTable.getPartition(partitionName, isTempPartition);
+                    if (checkIsAdded != null) {
+                        LOG.warn("dynamic partition has been added, skip it. "
+                                + "db: {}, table: {}, partition: {}, tableId: {}",
+                                db.getFullName(), tableName, partitionName, olapTable.getId());
+                        Env.getCurrentEnv().onErasePartition(toAddPartition);
+                        continue;
+                    }
+                    if (isTempPartition) {
+                        olapTable.addTempPartition(toAddPartition);
+                    } else {
+                        olapTable.addPartition(toAddPartition);
                     }
+
+                    Env.getCurrentEnv().getEditLog().logAddPartition(batchPartsInfo.get(i).first);
+                    if (DebugPointUtil.isEnable("FE.DynamicPartitionScheduler.in.logEditPartitions")) {
+                        if (i == batchPartsInfo.size() / 2) {
+                            LOG.info("debug point FE.DynamicPartitionScheduler.in.logEditPartitions, throw e");
+                            // committed, but log some edit, others failed
+                            throw new Exception("debug point FE.DynamicPartitionScheduler"
+                                + ".in.commitCloudPartition");
+                        }
+                    }
+                } finally {
+                    olapTable.writeUnlock();
                 }
             }
             LOG.info("finish write edit log to add partitions in batch, "
                     + "numPartitions: {}, db: {}, table: {}, tableId: {}",
-                    partsInfo.size(), db.getFullName(), tableName, olapTable.getId());
+                    batchPartsInfo.size(), db.getFullName(), tableName, olapTable.getId());
         } catch (Exception e) {
             LOG.warn("cloud in commit step, dbName {}, tableName {}, tableId {} exception {}",
                     db.getFullName(), tableName, olapTable.getId(), e.getMessage());

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -1358,7 +1358,7 @@ public void addPartitionLike(Database db, String tableName, AddPartitionLikeClau
             } finally {
                 table.readUnlock();
             }
-            addPartition(db, tableName, clause, false, 0, true);
+            addPartition(db, tableName, clause, false, 0, true, null);
 
         } catch (UserException e) {
             throw new DdlException("Failed to ADD PARTITION " + addPartitionLikeClause.getPartitionName()
@@ -1447,9 +1447,11 @@ private ResultOr<CompletableFuture<Void>, DdlException> getCurrentPartitionFutur
         }
     }
 
-    public PartitionPersistInfo addPartition(Database db, String tableName, AddPartitionClause addPartitionClause,
+    public void addPartition(Database db, String tableName, AddPartitionClause addPartitionClause,
                                              boolean isCreateTable, long generatedPartitionId,
-                                             boolean writeEditLog) throws DdlException {
+                                             boolean writeEditLog,
+                                             List<Pair<PartitionPersistInfo, Partition>> batchPartitions)
+            throws DdlException {
         // in cloud mode, isCreateTable == true, create dynamic partition use, so partitionId must have been generated.
         // isCreateTable == false, other case, partitionId generate in below, must be set 0
         if (!FeConstants.runningUnitTest && Config.isCloudMode()
@@ -1478,7 +1480,7 @@ public PartitionPersistInfo addPartition(Database db, String tableName, AddParti
                 if (singlePartitionDesc.isSetIfNotExists()) {
                     LOG.info("table[{}] add partition[{}] which already exists", olapTable.getName(), partitionName);
                     if (!DebugPointUtil.isEnable("InternalCatalog.addPartition.noCheckExists")) {
-                        return null;
+                        return;
                     }
                 } else {
                     ErrorReport.reportDdlException(ErrorCode.ERR_SAME_NAME_PARTITION, partitionName);
@@ -1645,7 +1647,7 @@ public PartitionPersistInfo addPartition(Database db, String tableName, AddParti
                 db, tableName, olapTable, partitionName, singlePartitionDesc);
         if (ownerFutureOr.isErr()) {
             if (ownerFutureOr.unwrapErr() == null) {
-                return null;
+                return;
             } else {
                 throw ownerFutureOr.unwrapErr();
             }
@@ -1701,7 +1703,7 @@ public PartitionPersistInfo addPartition(Database db, String tableName, AddParti
                     LOG.info("table[{}] add partition[{}] which already exists", olapTable.getName(), partitionName);
                     if (singlePartitionDesc.isSetIfNotExists()) {
                         failedCleanCallback.run();
-                        return null;
+                        return;
                     } else {
                         ErrorReport.reportDdlException(ErrorCode.ERR_SAME_NAME_PARTITION, partitionName);
                     }
@@ -1759,12 +1761,6 @@ public PartitionPersistInfo addPartition(Database db, String tableName, AddParti
                 // update partition info
                 partitionInfo.handleNewSinglePartitionDesc(singlePartitionDesc, partitionId, isTempPartition);
 
-                if (isTempPartition) {
-                    olapTable.addTempPartition(partition);
-                } else {
-                    olapTable.addPartition(partition);
-                }
-
                 // log
                 PartitionPersistInfo info = null;
                 if (partitionInfo.getType() == PartitionType.RANGE) {
@@ -1790,11 +1786,16 @@ public PartitionPersistInfo addPartition(Database db, String tableName, AddParti
                 }
                 if (writeEditLog) {
                     Env.getCurrentEnv().getEditLog().logAddPartition(info);
+                    if (isTempPartition) {
+                        olapTable.addTempPartition(partition);
+                    } else {
+                        olapTable.addPartition(partition);
+                    }
                     LOG.info("succeed in creating partition[{}], temp: {}", partitionId, isTempPartition);
                 } else {
+                    batchPartitions.add(Pair.of(info, partition));
                     LOG.info("postpone creating partition[{}], temp: {}", partitionId, isTempPartition);
                 }
-                return info;
             } finally {
                 olapTable.writeUnlock();
             }
@@ -1840,7 +1841,7 @@ public void addMultiPartitions(Database db, String tableName, AlterMultiPartitio
         for (SinglePartitionDesc singlePartitionDesc : singlePartitionDescs) {
             AddPartitionClause addPartitionClause = new AddPartitionClause(singlePartitionDesc, null,
                     multiPartitionClause.getProperties(), false);
-            addPartition(db, tableName, addPartitionClause, false, 0, true);
+            addPartition(db, tableName, addPartitionClause, false, 0, true, null);
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java

@@ -404,7 +404,7 @@ public static void addPartition(MTMV mtmv, PartitionKeyDesc oldPartitionKeyDesc)
         AddPartitionClause addPartitionClause = new AddPartitionClause(singlePartitionDesc,
                 mtmv.getDefaultDistributionInfo().toDistributionDesc(), partitionProperties, false);
         Env.getCurrentEnv().addPartition((Database) mtmv.getDatabase(), mtmv.getName(), addPartitionClause,
-                false, 0, true);
+                false, 0, true, null);
     }
 
     /**

fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java

@@ -1679,14 +1679,29 @@ public void logRefreshExternalTableSchema(RefreshExternalTableInfo info) {
     }
 
     public long logAddPartition(PartitionPersistInfo info) {
+        if (DebugPointUtil.isEnable("FE.logAddPartition.slow")) {
+            DebugPointUtil.DebugPoint debugPoint = DebugPointUtil.getDebugPoint("FE.logAddPartition.slow");
+            String pName = debugPoint.param("pName", "");
+            if (info.getPartition().getName().equals(pName)) {
+                int sleepMs = debugPoint.param("sleep", 1000);
+                LOG.info("logAddPartition debug point hit, pName {}, sleep {} s", pName, sleepMs);
+                try {
+                    Thread.sleep(sleepMs);
+                } catch (InterruptedException e) {
+                    LOG.warn("sleep interrupted", e);
+                }
+            }
+        }
         long logId = logEdit(OperationType.OP_ADD_PARTITION, info);
+        LOG.info("log add partition, logId:{}, info: {}", logId, info.toJson());
         AddPartitionRecord record = new AddPartitionRecord(logId, info);
         Env.getCurrentEnv().getBinlogManager().addAddPartitionRecord(record);
         return logId;
     }
 
     public long logDropPartition(DropPartitionInfo info) {
         long logId = logEdit(OperationType.OP_DROP_PARTITION, info);
+        LOG.info("log drop partition, logId:{}, info: {}", logId, info.toJson());
         Env.getCurrentEnv().getBinlogManager().addDropPartitionRecord(info, logId);
         return logId;
     }
@@ -1697,6 +1712,7 @@ public void logErasePartition(long partitionId) {
 
     public void logRecoverPartition(RecoverInfo info) {
         long logId = logEdit(OperationType.OP_RECOVER_PARTITION, info);
+        LOG.info("log recover partition, logId:{}, info: {}", logId, info.toJson());
         Env.getCurrentEnv().getBinlogManager().addRecoverTableRecord(info, logId);
     }
 
@@ -1715,6 +1731,7 @@ public void logBatchModifyPartition(BatchModifyPartitionsInfo info) {
 
     public void logDropTable(DropInfo info) {
         long logId = logEdit(OperationType.OP_DROP_TABLE, info);
+        LOG.info("log drop table, logId : {}, infos: {}", logId, info);
         if (Strings.isNullOrEmpty(info.getCtl()) || info.getCtl().equals(InternalCatalog.INTERNAL_CATALOG_NAME)) {
             DropTableRecord record = new DropTableRecord(logId, info);
             Env.getCurrentEnv().getBinlogManager().addDropTableRecord(record);
@@ -1727,11 +1744,13 @@ public void logEraseTable(long tableId) {
 
     public void logRecoverTable(RecoverInfo info) {
         long logId = logEdit(OperationType.OP_RECOVER_TABLE, info);
+        LOG.info("log recover table, logId : {}, infos: {}", logId, info);
         Env.getCurrentEnv().getBinlogManager().addRecoverTableRecord(info, logId);
     }
 
     public void logDropRollup(DropInfo info) {
         long logId = logEdit(OperationType.OP_DROP_ROLLUP, info);
+        LOG.info("log drop rollup, logId : {}, infos: {}", logId, info);
         Env.getCurrentEnv().getBinlogManager().addDropRollup(info, logId);
     }
 
@@ -1848,7 +1867,8 @@ public void logDropRole(PrivInfo info) {
     }
 
     public void logDatabaseRename(DatabaseInfo databaseInfo) {
-        logEdit(OperationType.OP_RENAME_DB, databaseInfo);
+        long logId = logEdit(OperationType.OP_RENAME_DB, databaseInfo);
+        LOG.info("log database rename, logId : {}, infos: {}", logId, databaseInfo);
     }
 
     public void logTableRename(TableInfo tableInfo) {

fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java

@@ -3785,7 +3785,7 @@ public TCreatePartitionResult createPartition(TCreatePartitionRequest request) t
         for (AddPartitionClause addPartitionClause : addPartitionClauseMap.values()) {
             try {
                 // here maybe check and limit created partitions num
-                Env.getCurrentEnv().addPartition(db, olapTable.getName(), addPartitionClause, false, 0, true);
+                Env.getCurrentEnv().addPartition(db, olapTable.getName(), addPartitionClause, false, 0, true, null);
             } catch (DdlException e) {
                 LOG.warn(e);
                 errorStatus.setErrorMsgs(

…rent_serialize/diffrent_serialize.groovy …ent_serialize/different_serialize.groovyregression-test/suites/cloud_p0/diffrent_serialize/diffrent_serialize.groovy renamed to regression-test/suites/cloud_p0/different_serialize/different_serialize.groovy regression-test/suites/cloud_p0/diffrent_serialize/diffrent_serialize.groovy renamed to regression-test/suites/cloud_p0/different_serialize/different_serialize.groovy

@@ -17,7 +17,7 @@
 
 import org.codehaus.groovy.runtime.IOGroovyMethods
 
-suite ("diffrent_serialize_cloud") {
+suite ("different_serialize_cloud") {
 
     sql """ DROP TABLE IF EXISTS d_table; """