maintainer: add redo scheduler

Makefile

@@ -121,8 +121,8 @@ P=3
 # The following packages are used in unit tests.
 # Add new packages here if you want to include them in unit tests.
 UT_PACKAGES_DISPATCHER := ./pkg/sink/cloudstorage/... ./pkg/sink/mysql/... ./pkg/sink/util/... ./downstreamadapter/sink/... ./downstreamadapter/dispatcher/... ./downstreamadapter/dispatchermanager/... ./downstreamadapter/eventcollector/... ./pkg/sink/...
-UT_PACKAGES_MAINTAINER := ./maintainer/...
-UT_PACKAGES_COORDINATOR := ./coordinator/...
+UT_PACKAGES_MAINTAINER := ./maintainer/... ./pkg/scheduler/...
+UT_PACKAGES_COORDINATOR := ./coordinator/... 
 UT_PACKAGES_LOGSERVICE := ./logservice/...
 UT_PACKAGES_OTHERS := ./pkg/eventservice/... ./pkg/version/... ./utils/dynstream/... ./pkg/common/event/... ./pkg/common/...
 

api/v2/changefeed.go

@@ -785,7 +785,8 @@ func (h *OpenAPIV2) MoveTable(c *gin.Context) {
 	}
 
 	targetNodeID := c.Query("targetNodeID")
-	err = maintainer.MoveTable(int64(tableId), node.ID(targetNodeID))
+	mode, _ := strconv.ParseInt(c.Query("mode"), 10, 64)
+	err = maintainer.MoveTable(int64(tableId), node.ID(targetNodeID), mode)
 	if err != nil {
 		log.Error("failed to move table", zap.Error(err), zap.Int64("tableID", tableId), zap.String("targetNodeID", targetNodeID))
 		_ = c.Error(err)
@@ -859,7 +860,8 @@ func (h *OpenAPIV2) MoveSplitTable(c *gin.Context) {
 	}
 
 	targetNodeID := c.Query("targetNodeID")
-	err = maintainer.MoveSplitTable(int64(tableId), node.ID(targetNodeID))
+	mode, _ := strconv.ParseInt(c.Query("mode"), 10, 64)
+	err = maintainer.MoveSplitTable(int64(tableId), node.ID(targetNodeID), mode)
 	if err != nil {
 		log.Error("failed to move split table", zap.Error(err), zap.Int64("tableID", tableId), zap.String("targetNodeID", targetNodeID))
 		_ = c.Error(err)
@@ -930,8 +932,8 @@ func (h *OpenAPIV2) SplitTableByRegionCount(c *gin.Context) {
 		_ = c.Error(apperror.ErrMaintainerNotFounded)
 		return
 	}
-
-	err = maintainer.SplitTableByRegionCount(int64(tableId))
+	mode, _ := strconv.ParseInt(c.Query("mode"), 10, 64)
+	err = maintainer.SplitTableByRegionCount(int64(tableId), mode)
 	if err != nil {
 		log.Error("failed to split table by region count", zap.Error(err), zap.Int64("tableID", tableId))
 		_ = c.Error(err)
@@ -1002,7 +1004,8 @@ func (h *OpenAPIV2) MergeTable(c *gin.Context) {
 		return
 	}
 
-	err = maintainer.MergeTable(int64(tableId))
+	mode, _ := strconv.ParseInt(c.Query("mode"), 10, 64)
+	err = maintainer.MergeTable(int64(tableId), mode)
 	if err != nil {
 		log.Error("failed to merge table", zap.Error(err), zap.Int64("tableID", tableId))
 		_ = c.Error(err)
@@ -1061,7 +1064,8 @@ func (h *OpenAPIV2) ListTables(c *gin.Context) {
 		return
 	}
 
-	tables := maintainer.GetTables()
+	mode, _ := strconv.ParseInt(c.Query("mode"), 10, 64)
+	tables := maintainer.GetTables(mode)
 
 	nodeTableInfoMap := make(map[string]*NodeTableInfo)
 
@@ -1131,7 +1135,8 @@ func (h *OpenAPIV2) getDispatcherCount(c *gin.Context) {
 		return
 	}
 
-	number := maintainer.GetDispatcherCount()
+	mode, _ := strconv.ParseInt(c.Query("mode"), 10, 64)
+	number := maintainer.GetDispatcherCount(mode)
 	c.JSON(http.StatusOK, &DispatcherCount{Count: number})
 }
 

cmd/cdc/cli/cli_changefeed_merge_table.go

@@ -29,6 +29,7 @@ type mergeTableChangefeedOptions struct {
 	changefeedID string
 	namespace    string
 	tableId      int64
+	mode         int64
 }
 
 // newCreateChangefeedOptions creates new options for the `cli changefeed create` command.
@@ -42,6 +43,7 @@ func (o *mergeTableChangefeedOptions) addFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().StringVarP(&o.namespace, "namespace", "n", "default", "Replication task (changefeed) Namespace")
 	cmd.PersistentFlags().StringVarP(&o.changefeedID, "changefeed-id", "c", "", "Replication task (changefeed) ID")
 	cmd.PersistentFlags().Int64VarP(&o.tableId, "table-id", "t", 0, "the id of table to move")
+	cmd.PersistentFlags().Int64Var(&o.mode, "mode", 0, "enable redo when mode is 1")
 	_ = cmd.MarkPersistentFlagRequired("changefeed-id")
 	_ = cmd.MarkPersistentFlagRequired("table-id")
 }
@@ -61,7 +63,7 @@ func (o *mergeTableChangefeedOptions) complete(f factory.Factory) error {
 func (o *mergeTableChangefeedOptions) run(cmd *cobra.Command) error {
 	ctx := context.Background()
 
-	err := o.apiClientV2.Changefeeds().MergeTable(ctx, o.namespace, o.changefeedID, o.tableId)
+	err := o.apiClientV2.Changefeeds().MergeTable(ctx, o.namespace, o.changefeedID, o.tableId, o.mode)
 	var errStr string
 	if err != nil {
 		errStr = err.Error()

cmd/cdc/cli/cli_changefeed_move_split_table.go

@@ -30,6 +30,7 @@ type moveSplitTableChangefeedOptions struct {
 	namespace    string
 	tableId      int64
 	targetNodeID string
+	mode         int64
 }
 
 // newCreateChangefeedOptions creates new options for the `cli changefeed create` command.
@@ -44,6 +45,7 @@ func (o *moveSplitTableChangefeedOptions) addFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().StringVarP(&o.changefeedID, "changefeed-id", "c", "", "Replication task (changefeed) ID")
 	cmd.PersistentFlags().Int64VarP(&o.tableId, "table-id", "t", 0, "the id of table to move")
 	cmd.PersistentFlags().StringVarP(&o.targetNodeID, "target-node-id", "d", "", "the dest for the table to move")
+	cmd.PersistentFlags().Int64Var(&o.mode, "mode", 0, "enable redo when mode is 1")
 	_ = cmd.MarkPersistentFlagRequired("changefeed-id")
 	_ = cmd.MarkPersistentFlagRequired("table-id")
 	_ = cmd.MarkPersistentFlagRequired("target-node-id")
@@ -64,7 +66,7 @@ func (o *moveSplitTableChangefeedOptions) complete(f factory.Factory) error {
 func (o *moveSplitTableChangefeedOptions) run(cmd *cobra.Command) error {
 	ctx := context.Background()
 
-	err := o.apiClientV2.Changefeeds().MoveSplitTable(ctx, o.namespace, o.changefeedID, o.tableId, o.targetNodeID)
+	err := o.apiClientV2.Changefeeds().MoveSplitTable(ctx, o.namespace, o.changefeedID, o.tableId, o.targetNodeID, o.mode)
 	var errStr string
 	if err != nil {
 		errStr = err.Error()

cmd/cdc/cli/cli_changefeed_move_table.go

@@ -30,6 +30,7 @@ type moveTableChangefeedOptions struct {
 	namespace    string
 	tableId      int64
 	targetNodeID string
+	mode         int64
 }
 
 // newCreateChangefeedOptions creates new options for the `cli changefeed create` command.
@@ -44,6 +45,7 @@ func (o *moveTableChangefeedOptions) addFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().StringVarP(&o.changefeedID, "changefeed-id", "c", "", "Replication task (changefeed) ID")
 	cmd.PersistentFlags().Int64VarP(&o.tableId, "table-id", "t", 0, "the id of table to move")
 	cmd.PersistentFlags().StringVarP(&o.targetNodeID, "target-node-id", "d", "", "the dest for the table to move")
+	cmd.PersistentFlags().Int64Var(&o.mode, "mode", 0, "enable redo when mode is 1")
 	_ = cmd.MarkPersistentFlagRequired("changefeed-id")
 	_ = cmd.MarkPersistentFlagRequired("table-id")
 	_ = cmd.MarkPersistentFlagRequired("target-node-id")
@@ -69,7 +71,7 @@ type response struct {
 func (o *moveTableChangefeedOptions) run(cmd *cobra.Command) error {
 	ctx := context.Background()
 
-	err := o.apiClientV2.Changefeeds().MoveTable(ctx, o.namespace, o.changefeedID, o.tableId, o.targetNodeID)
+	err := o.apiClientV2.Changefeeds().MoveTable(ctx, o.namespace, o.changefeedID, o.tableId, o.targetNodeID, o.mode)
 	var errStr string
 	if err != nil {
 		errStr = err.Error()

cmd/cdc/cli/cli_changefeed_split_table_by_region_count.go

@@ -29,6 +29,7 @@ type splitTableByRegionCountChangefeedOptions struct {
 	changefeedID string
 	namespace    string
 	tableId      int64
+	mode         int64
 }
 
 // newCreateChangefeedOptions creates new options for the `cli changefeed create` command.
@@ -42,6 +43,7 @@ func (o *splitTableByRegionCountChangefeedOptions) addFlags(cmd *cobra.Command)
 	cmd.PersistentFlags().StringVarP(&o.namespace, "namespace", "n", "default", "Replication task (changefeed) Namespace")
 	cmd.PersistentFlags().StringVarP(&o.changefeedID, "changefeed-id", "c", "", "Replication task (changefeed) ID")
 	cmd.PersistentFlags().Int64VarP(&o.tableId, "table-id", "t", 0, "the id of table to move")
+	cmd.PersistentFlags().Int64Var(&o.mode, "mode", 0, "enable redo when mode is 1")
 	_ = cmd.MarkPersistentFlagRequired("changefeed-id")
 	_ = cmd.MarkPersistentFlagRequired("table-id")
 }
@@ -61,7 +63,7 @@ func (o *splitTableByRegionCountChangefeedOptions) complete(f factory.Factory) e
 func (o *splitTableByRegionCountChangefeedOptions) run(cmd *cobra.Command) error {
 	ctx := context.Background()
 
-	err := o.apiClientV2.Changefeeds().SplitTableByRegionCount(ctx, o.namespace, o.changefeedID, o.tableId)
+	err := o.apiClientV2.Changefeeds().SplitTableByRegionCount(ctx, o.namespace, o.changefeedID, o.tableId, o.mode)
 	var errStr string
 	if err != nil {
 		errStr = err.Error()

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -33,7 +33,7 @@ import (
 // DispatcherService defines the interface for providing dispatcher information and basic event handling.
 type DispatcherService interface {
 	GetId() common.DispatcherID
-	GetType() int
+	GetMode() int64
 	GetStartTs() uint64
 	GetBDRMode() bool
 	GetChangefeedID() common.ChangeFeedID
@@ -121,6 +121,9 @@ type BasicDispatcher struct {
 	// componentStatus is the status of the dispatcher, such as working, removing, stopped.
 	componentStatus *ComponentStateWithMutex
 
+	// schemaIDToDispatchers is shared in the DispatcherManager
+	schemaIDToDispatchers *SchemaIDToDispatchers
+
 	// Shared info containing all common configuration and resources
 	sharedInfo *SharedInfo
 
@@ -148,8 +151,8 @@ type BasicDispatcher struct {
 
 	isRemoving atomic.Bool
 
-	seq            uint64
-	dispatcherType int
+	seq  uint64
+	mode int64
 
 	BootstrapState bootstrapState
 }
@@ -159,29 +162,31 @@ func NewBasicDispatcher(
 	tableSpan *heartbeatpb.TableSpan,
 	startTs uint64,
 	schemaID int64,
+	schemaIDToDispatchers *SchemaIDToDispatchers,
 	startTsIsSyncpoint bool,
 	currentPDTs uint64,
-	dispatcherType int,
+	mode int64,
 	sink sink.Sink,
 	sharedInfo *SharedInfo,
 ) *BasicDispatcher {
 	dispatcher := &BasicDispatcher{
-		id:                 id,
-		tableSpan:          tableSpan,
-		startTs:            startTs,
-		startTsIsSyncpoint: startTsIsSyncpoint,
-		sharedInfo:         sharedInfo,
-		sink:               sink,
-		componentStatus:    newComponentStateWithMutex(heartbeatpb.ComponentState_Initializing),
-		resolvedTs:         startTs,
-		isRemoving:         atomic.Bool{},
-		blockEventStatus:   BlockEventStatus{blockPendingEvent: nil},
-		tableProgress:      NewTableProgress(),
-		schemaID:           schemaID,
-		resendTaskMap:      newResendTaskMap(),
-		creationPDTs:       currentPDTs,
-		dispatcherType:     dispatcherType,
-		BootstrapState:     BootstrapFinished,
+		id:                    id,
+		tableSpan:             tableSpan,
+		startTs:               startTs,
+		startTsIsSyncpoint:    startTsIsSyncpoint,
+		sharedInfo:            sharedInfo,
+		sink:                  sink,
+		componentStatus:       newComponentStateWithMutex(heartbeatpb.ComponentState_Initializing),
+		resolvedTs:            startTs,
+		isRemoving:            atomic.Bool{},
+		blockEventStatus:      BlockEventStatus{blockPendingEvent: nil},
+		tableProgress:         NewTableProgress(),
+		schemaID:              schemaID,
+		schemaIDToDispatchers: schemaIDToDispatchers,
+		resendTaskMap:         newResendTaskMap(),
+		creationPDTs:          currentPDTs,
+		mode:                  mode,
+		BootstrapState:        BootstrapFinished,
 	}
 
 	return dispatcher
@@ -275,7 +280,7 @@ func (d *BasicDispatcher) updateDispatcherStatusToWorking() {
 			ID:              d.id.ToPB(),
 			ComponentStatus: heartbeatpb.ComponentState_Working,
 			CheckpointTs:    d.GetCheckpointTs(),
-			IsRedo:          IsRedoDispatcher(d),
+			Mode:            d.GetMode(),
 		},
 		Seq: d.seq,
 	}
@@ -306,17 +311,17 @@ func (d *BasicDispatcher) HandleEvents(dispatcherEvents []DispatcherEvent, wakeC
 //
 // wakeCallback is used to wake the dynamic stream to handle the next batch events.
 // It will be called when all the events are flushed to downstream successfully.
-func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeCallback func()) (block bool) {
+func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeCallback func()) bool {
 	// Only return false when all events are resolvedTs Event.
-	block = false
+	block := false
 	dmlWakeOnce := &sync.Once{}
 	dmlEvents := make([]*commonEvent.DMLEvent, 0, len(dispatcherEvents))
 	latestResolvedTs := uint64(0)
-	isRedo := IsRedoDispatcher(d)
 	// Dispatcher is ready, handle the events
 	for _, dispatcherEvent := range dispatcherEvents {
 		log.Debug("dispatcher receive all event",
-			zap.Stringer("dispatcher", d.id), zap.Bool("isRedo", isRedo),
+			zap.Stringer("dispatcher", d.id), zap.Int64("mode", d.mode),
+			zap.String("eventType", commonEvent.TypeToString(dispatcherEvent.Event.GetType())),
 			zap.Any("event", dispatcherEvent.Event))
 		failpoint.Inject("HandleEventsSlowly", func() {
 			lag := time.Duration(rand.Intn(5000)) * time.Millisecond
@@ -376,7 +381,7 @@ func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeC
 			err := ddl.GetError()
 			if err != nil {
 				d.HandleError(err)
-				return
+				return block
 			}
 			log.Info("dispatcher receive ddl event",
 				zap.Stringer("dispatcher", d.id),
@@ -392,7 +397,7 @@ func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeC
 			})
 			d.dealWithBlockEvent(ddl)
 		case commonEvent.TypeSyncPointEvent:
-			if isRedo {
+			if common.IsRedoMode(d.GetMode()) {
 				continue
 			}
 			if len(dispatcherEvents) != 1 {
@@ -510,7 +515,7 @@ func (d *BasicDispatcher) HandleDispatcherStatus(dispatcherStatus *heartbeatpb.D
 				IsSyncPoint: dispatcherStatus.GetAction().IsSyncPoint,
 				Stage:       heartbeatpb.BlockStage_DONE,
 			},
-			IsRedo: IsRedoDispatcher(d),
+			Mode: d.GetMode(),
 		}
 	}
 }
@@ -568,7 +573,7 @@ func (d *BasicDispatcher) dealWithBlockEvent(event commonEvent.BlockEvent) {
 					IsSyncPoint:       false, // sync point event must should block
 					Stage:             heartbeatpb.BlockStage_NONE,
 				},
-				IsRedo: IsRedoDispatcher(d),
+				Mode: d.GetMode(),
 			}
 			identifier := BlockEventIdentifier{
 				CommitTs:    event.GetCommitTs(),
@@ -622,7 +627,7 @@ func (d *BasicDispatcher) dealWithBlockEvent(event commonEvent.BlockEvent) {
 						IsSyncPoint:       true,
 						Stage:             heartbeatpb.BlockStage_WAITING,
 					},
-					IsRedo: IsRedoDispatcher(d),
+					Mode: d.GetMode(),
 				}
 				identifier := BlockEventIdentifier{
 					CommitTs:    commitTs,
@@ -644,7 +649,7 @@ func (d *BasicDispatcher) dealWithBlockEvent(event commonEvent.BlockEvent) {
 					IsSyncPoint:       false,
 					Stage:             heartbeatpb.BlockStage_WAITING,
 				},
-				IsRedo: IsRedoDispatcher(d),
+				Mode: d.GetMode(),
 			}
 			identifier := BlockEventIdentifier{
 				CommitTs:    event.GetCommitTs(),
@@ -676,7 +681,7 @@ func (d *BasicDispatcher) dealWithBlockEvent(event commonEvent.BlockEvent) {
 					return
 				} else {
 					d.schemaID = schemaIDChange.NewSchemaID
-					d.sharedInfo.schemaIDToDispatchers.Update(schemaIDChange.OldSchemaID, schemaIDChange.NewSchemaID)
+					d.schemaIDToDispatchers.Update(schemaIDChange.OldSchemaID, schemaIDChange.NewSchemaID)
 					return
 				}
 			}
@@ -742,7 +747,7 @@ func (d *BasicDispatcher) TryClose() (w heartbeatpb.Watermark, ok bool) {
 		log.Info("dispatcher component has stopped and is ready for cleanup",
 			zap.Stringer("changefeedID", d.sharedInfo.changefeedID),
 			zap.Stringer("dispatcher", d.id),
-			zap.Bool("isRedo", IsRedoDispatcher(d)),
+			zap.Int64("mode", d.mode),
 			zap.String("table", common.FormatTableSpan(d.tableSpan)),
 			zap.Uint64("checkpointTs", d.GetCheckpointTs()),
 			zap.Uint64("resolvedTs", d.GetResolvedTs()),
@@ -751,7 +756,7 @@ func (d *BasicDispatcher) TryClose() (w heartbeatpb.Watermark, ok bool) {
 	}
 	log.Info("dispatcher is not ready to close",
 		zap.Stringer("dispatcher", d.id),
-		zap.Bool("isRedo", IsRedoDispatcher(d)),
+		zap.Int64("mode", d.mode),
 		zap.Bool("sinkIsNormal", d.sink.IsNormal()),
 		zap.Bool("tableProgressEmpty", d.tableProgress.Empty()),
 		zap.Int("tableProgressLen", d.tableProgress.Len()),
@@ -763,7 +768,7 @@ func (d *BasicDispatcher) TryClose() (w heartbeatpb.Watermark, ok bool) {
 func (d *BasicDispatcher) removeDispatcher() {
 	log.Info("remove dispatcher",
 		zap.Stringer("dispatcher", d.id),
-		zap.Bool("isRedo", IsRedoDispatcher(d)),
+		zap.Int64("mode", d.mode),
 		zap.Stringer("changefeedID", d.sharedInfo.changefeedID),
 		zap.String("table", common.FormatTableSpan(d.tableSpan)))
 	dispatcherStatusDS := GetDispatcherStatusDynamicStream()