[fix](schema-change) Fix job replay failure when partitions added to table after job finish and before log edition

Author: TangSiyang2001

fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java

@@ -663,21 +663,21 @@ protected void runRunningJob() throws AlterCancelException {
             commitShadowIndex();
             // all partitions are good
             onFinished(tbl);
-        } finally {
-            tbl.writeUnlock();
-        }
-
-        pruneMeta();
-
-        LOG.info("schema change job finished: {}", jobId);
+            pruneMeta();
 
-        changeTableState(dbId, tableId, OlapTableState.NORMAL);
-        LOG.info("set table's state to NORMAL, table id: {}, job id: {}", tableId, jobId);
+            LOG.info("schema change job finished: {}", jobId);
 
-        this.jobState = JobState.FINISHED;
-        this.finishedTimeMs = System.currentTimeMillis();
-        Env.getCurrentEnv().getEditLog().logAlterJob(this);
+            changeTableState(dbId, tableId, OlapTableState.NORMAL);
+            LOG.info("set table's state to NORMAL, table id: {}, job id: {}", tableId, jobId);
 
+            this.jobState = JobState.FINISHED;
+            this.finishedTimeMs = System.currentTimeMillis();
+            // Write edit log with table's write lock held, to avoid adding partitions before writing edit log,
+            // else it will try to transform index in newly added partition while replaying and result in failure.
+            Env.getCurrentEnv().getEditLog().logAlterJob(this);
+        } finally {
+            tbl.writeUnlock();
+        }
         postProcessOriginIndex();
         // Drop table column stats after schema change finished.
         Env.getCurrentEnv().getAnalysisManager().dropStats(tbl, null);

regression-test/suites/schema_change_p0/test_sc_replay_after_add_partition.groovy

@@ -0,0 +1,35 @@
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.NodeType
+
+suite("test_sc_replay_after_add_partition", "docker") {
+    def options = new ClusterOptions()
+    options.enableDebugPoints()
+    options.setFeNum(1)
+    options.connectToFollower = true
+
+    docker(options) {
+        def tbl = "test_forward_query"
+        sql """ DROP TABLE IF EXISTS ${tbl} """
+        sql """
+            CREATE TABLE ${tbl}
+            (
+                k1 int,
+                v1 int
+            )
+            DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+            PROPERTIES (
+               "replication_num" = "1"
+            );
+        """
+        sql """
+            INSERT INTO ${tbl} VALUES(1, 1);
+        """
+        sql """
+            INSERT INTO ${tbl} VALUES(2, 2);
+        """
+        cluster.injectDebugPoints(NodeType.FE, ['StmtExecutor.forward_all_queries': [forwardAllQueries:true, execute:1]])
+        sql """
+            
+        """
+    }
+}