[WIP] Add parition ddls in random test

.github/workflows/integration_test_mysql.yaml

@@ -555,6 +555,51 @@ jobs:
           make integration_test_build
           ls -l bin/ && ls -l tools/bin/
 
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
+      - name: Test fail_over_ddl_mix
+        run: |
+          pwd && ls -l bin/ && ls -l tools/bin/
+          export TICDC_NEWARCH=true && make integration_test CASE=fail_over_ddl_mix 
+
       - name: Test fail_over
         run: |
           pwd && ls -l bin/ && ls -l tools/bin/

logservice/schemastore/persist_storage_ddl_handlers.go

@@ -511,17 +511,19 @@ func buildPersistedDDLEventCommon(args buildPersistedDDLEventFuncArgs) Persisted
 	// Note: if a ddl involve multiple tables, job.TableID is different with job.BinlogInfo.TableInfo.ID
 	// and usually job.BinlogInfo.TableInfo.ID will be the newly created IDs.
 	event := PersistedDDLEvent{
-		ID:             job.ID,
-		Type:           byte(job.Type),
-		SchemaID:       job.SchemaID,
-		TableID:        job.TableID,
-		Query:          query,
-		SchemaVersion:  job.BinlogInfo.SchemaVersion,
-		DBInfo:         job.BinlogInfo.DBInfo,
-		TableInfo:      job.BinlogInfo.TableInfo,
-		FinishedTs:     job.BinlogInfo.FinishedTS,
-		BDRRole:        job.BDRRole,
-		CDCWriteSource: job.CDCWriteSource,
+		ID:                job.ID,
+		Type:              byte(job.Type),
+		TableNameInDDLJob: job.TableName,
+		DBNameInDDLJob:    job.SchemaName,
+		SchemaID:          job.SchemaID,
+		TableID:           job.TableID,
+		Query:             query,
+		SchemaVersion:     job.BinlogInfo.SchemaVersion,
+		DBInfo:            job.BinlogInfo.DBInfo,
+		TableInfo:         job.BinlogInfo.TableInfo,
+		FinishedTs:        job.BinlogInfo.FinishedTS,
+		BDRRole:           job.BDRRole,
+		CDCWriteSource:    job.CDCWriteSource,
 	}
 	return event
 }
@@ -1515,6 +1517,9 @@ func buildDDLEventCommon(rawEvent *PersistedDDLEvent, tableFilter filter.Filter,
 		TableInfo:  wrapTableInfo,
 		FinishedTs: rawEvent.FinishedTs,
 		TiDBOnly:   tiDBOnly,
+
+		TableNameInDDLJob: rawEvent.TableNameInDDLJob,
+		DBNameInDDLJob:    rawEvent.DBNameInDDLJob,
 	}, !filtered
 }
 

logservice/schemastore/types.go

@@ -25,6 +25,11 @@ type PersistedDDLEvent struct {
 	ID   int64 `msg:"id"`
 	Type byte  `msg:"type"`
 
+	// the table name for the ddl job in the information_schema.ddl_jobs table(just ddl job.TableName)
+	TableNameInDDLJob string `msg:"table_name_in_ddl_job"`
+	// the database name for the ddl job in the information_schema.ddl_jobs table(just ddl job.dbName)
+	DBNameInDDLJob string `msg:"db_name_in_ddl_job"`
+
 	// SchemaID is from upstream Job.SchemaID, it corresponds to TableID
 	// it is the DB id of the table after the ddl
 	SchemaID int64 `msg:"schema_id"`

pkg/common/event/ddl_event.go

@@ -74,6 +74,11 @@ type DDLEvent struct {
 	//   Recover Table
 	TableNameChange *TableNameChange `json:"table_name_change"`
 
+	// the table name for the ddl job in the information_schema.ddl_jobs table(just ddl job.TableName)
+	TableNameInDDLJob string `msg:"table_name_in_ddl_job"`
+	// the database name for the ddl job in the information_schema.ddl_jobs table(just ddl job.dbName)
+	DBNameInDDLJob string `msg:"db_name_in_ddl_job"`
+
 	TiDBOnly bool `json:"tidb_only"`
 	// Call when event flush is completed
 	PostTxnFlushed []func() `json:"-"`
@@ -125,6 +130,14 @@ func (d *DDLEvent) GetExtraTableName() string {
 	return d.ExtraTableName
 }
 
+func (d *DDLEvent) GetTableNameInDDLJob() string {
+	return d.TableNameInDDLJob
+}
+
+func (d *DDLEvent) GetDBNameInDDLJob() string {
+	return d.DBNameInDDLJob
+}
+
 func (d *DDLEvent) GetEvents() []*DDLEvent {
 	// Some ddl event may be multi-events, we need to split it into multiple messages.
 	// Such as rename table test.table1 to test.table10, test.table2 to test.table20