Support ALTER TABLE t REORGANIZE PARTITION

cdc/entry/schema/snapshot.go

@@ -496,7 +496,8 @@ func (s *Snapshot) DoHandleDDL(job *timodel.Job) error {
 		}
 	case timodel.ActionTruncateTablePartition,
 		timodel.ActionAddTablePartition,
-		timodel.ActionDropTablePartition:
+		timodel.ActionDropTablePartition,
+		timodel.ActionReorganizePartition:
 		err := s.inner.updatePartition(getWrapTableInfo(job), job.BinlogInfo.FinishedTS)
 		if err != nil {
 			return errors.Trace(err)

cdc/entry/schema_storage_test.go

@@ -822,6 +822,8 @@ func TestSchemaStorage(t *testing.T) {
 		"alter table test_ddl2.simple_test1 add primary key pk(id)",                          // ActionAddPrimaryKey
 		"ALTER TABLE test_ddl2.simple_test1 ALTER id SET DEFAULT 18",                         // ActionSetDefaultValue
 		"ALTER TABLE test_ddl2.simple_test1 CHARACTER SET = utf8mb4",                         // ActionModifyTableCharsetAndCollate
+
+		"ALTER TABLE test_ddl2.employees REORGANIZE PARTITION p3 INTO (PARTITION p2 VALUES LESS THAN (15), PARTITION p3 VALUES LESS THAN (20))", // ActionReorganizePartition
 		// "recover table test_ddl2.employees",                                                  // ActionRecoverTable this ddl can't work on mock tikv
 
 		"DROP TABLE test_ddl2.employees",

pkg/filter/filter.go

@@ -54,6 +54,7 @@ var allowDDLList = []timodel.ActionType{
 	timodel.ActionAlterIndexVisibility,
 	timodel.ActionMultiSchemaChange,
 	timodel.ActionExchangeTablePartition,
+	timodel.ActionReorganizePartition,
 	timodel.ActionAlterTTLInfo,
 	timodel.ActionAlterTTLRemove,
 }

pkg/filter/filter_test.go

@@ -177,6 +177,7 @@ func TestShouldDiscardDDL(t *testing.T) {
 	require.False(t, filter.ShouldDiscardDDL(timodel.ActionAlterIndexVisibility, "", ""))
 	require.False(t, filter.ShouldDiscardDDL(timodel.ActionMultiSchemaChange, "", ""))
 	require.False(t, filter.ShouldDiscardDDL(timodel.ActionExchangeTablePartition, "", ""))
+	require.False(t, filter.ShouldDiscardDDL(timodel.ActionReorganizePartition, "", ""))
 
 	// Discard sequence DDL.
 	require.True(t, filter.ShouldDiscardDDL(timodel.ActionAddForeignKey, "", ""))

pkg/filter/utils_test.go

@@ -98,6 +98,7 @@ func TestDDLToEventType(t *testing.T) {
 		{"Alter table test.t1 drop partition t11", timodel.ActionDropTablePartition, bf.DropTablePartition, nil},
 		{"Alter table test.t1 add partition (partition p3 values less than (2002))", timodel.ActionDropTablePartition, bf.DropTablePartition, nil},
 		{"Alter table test.t1 truncate partition t11", timodel.ActionDropTablePartition, bf.DropTablePartition, nil},
+		{"Alter table test.t1 reorganize partition p11 into (partition p1 values less than (10), partition p2 values less than (20))", timodel.ActionReorganizePartition, bf.AlterTable, nil},
 		{"alter table add i", timodel.ActionAddIndex, bf.NullEvent, cerror.ErrConvertDDLToEventTypeFailed},
 	}
 	p := parser.New()

pkg/sink/codec/canal/canal_entry.go

@@ -265,7 +265,7 @@ func convertDdlEventType(e *model.DDLEvent) canal.EventType {
 	case mm.ActionAddColumn, mm.ActionDropColumn, mm.ActionModifyColumn, mm.ActionRebaseAutoID,
 		mm.ActionSetDefaultValue, mm.ActionModifyTableComment, mm.ActionRenameIndex, mm.ActionAddTablePartition,
 		mm.ActionDropTablePartition, mm.ActionModifyTableCharsetAndCollate, mm.ActionTruncateTablePartition,
-		mm.ActionAlterIndexVisibility, mm.ActionMultiSchemaChange,
+		mm.ActionAlterIndexVisibility, mm.ActionMultiSchemaChange, mm.ActionReorganizePartition,
 		// AddColumns and DropColumns are removed in TiDB v6.2.0, see https://github.com/pingcap/tidb/pull/35862.
 		mm.ActionAddColumns, mm.ActionDropColumns:
 		return canal.EventType_ALTER

tests/integration_tests/partition_table/data/prepare.sql

@@ -26,4 +26,7 @@ ALTER TABLE t1 EXCHANGE PARTITION p3 WITH TABLE t2;
 insert into t2 values (100),(101),(102),(103),(104),(105); /*these values will be replicated to in downstream t2*/
 insert into t1 values (25),(29); /*these values will be replicated to in downstream t1.p3*/
 
+ALTER TABLE t1 REORGANIZE PARTITION p0,p2 INTO (PARTITION p0 VALUES LESS THAN (5), PARTITION p1 VALUES LESS THAN (10), PARTITION p2 VALUES LESS THAN (21));
+ALTER TABLE t1 REORGANIZE PARTITION p2,p3,p4 INTO (PARTITION p2 VALUES LESS THAN (20), PARTITION p3 VALUES LESS THAN (26), PARTITION p4 VALUES LESS THAN (35), PARTITION pMax VALUES LESS THAN (MAXVALUE));
+
 create table finish_mark (a int primary key);