TiCDC supports a snapshot-map in replication

cdc/changefeed.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"sync"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -78,14 +79,20 @@ type changeFeed struct {
 	info   *model.ChangeFeedInfo
 	status *model.ChangeFeedStatus
 
-	schema        *entry.SingleSchemaSnapshot
-	ddlState      model.ChangeFeedDDLState
-	targetTs      uint64
-	taskStatus    model.ProcessorsInfos
-	taskPositions map[model.CaptureID]*model.TaskPosition
-	filter        *filter.Filter
-	sink          sink.Sink
-	scheduler     scheduler.Scheduler
+	schema           *entry.SingleSchemaSnapshot
+	ddlState         model.ChangeFeedDDLState
+	targetTs         uint64
+	ddlTs            uint64
+	syncpointMutex   sync.Mutex
+	updateResolvedTs bool
+	startTimer       chan bool
+	syncpointSink    sink.SyncpointSink
+	syncCancel       context.CancelFunc
+	taskStatus       model.ProcessorsInfos
+	taskPositions    map[model.CaptureID]*model.TaskPosition
+	filter           *filter.Filter
+	sink             sink.Sink
+	scheduler        scheduler.Scheduler
 
 	cyclicEnabled bool
 
@@ -685,6 +692,43 @@ func (c *changeFeed) handleDDL(ctx context.Context, captures map[string]*model.C
 	return nil
 }
 
+// handleSyncPoint record every syncpoint to downstream if the syncpoint feature is enable
+func (c *changeFeed) handleSyncPoint(ctx context.Context) error {
+	//sync-point on
+	if c.info.SyncPointEnabled {
+		c.syncpointMutex.Lock()
+		defer c.syncpointMutex.Unlock()
+		// ResolvedTs == CheckpointTs means a syncpoint reached;
+		// !c.updateResolvedTs means the syncpoint is setted by ticker;
+		// c.ddlTs == 0 means no DDL wait to exec and we can sink the syncpoint record securely ( c.ddlTs != 0 means some DDL should be sink to downstream and this syncpoint is fake ).
+		if c.status.ResolvedTs == c.status.CheckpointTs && !c.updateResolvedTs {
+			log.Info("sync point reached by ticker", zap.Uint64("ResolvedTs", c.status.ResolvedTs), zap.Uint64("CheckpointTs", c.status.CheckpointTs), zap.Bool("updateResolvedTs", c.updateResolvedTs), zap.Uint64("ddlResolvedTs", c.ddlResolvedTs), zap.Uint64("ddlTs", c.ddlTs), zap.Uint64("ddlExecutedTs", c.ddlExecutedTs))
+			c.updateResolvedTs = true
+			err := c.syncpointSink.SinkSyncpoint(ctx, c.id, c.status.CheckpointTs)
+			if err != nil {
+				log.Error("syncpoint sink fail", zap.Uint64("ResolvedTs", c.status.ResolvedTs), zap.Uint64("CheckpointTs", c.status.CheckpointTs), zap.Error(err))
+			}
+		}
+
+		if c.status.ResolvedTs == 0 {
+			c.updateResolvedTs = true
+		}
+
+		// ResolvedTs == CheckpointTs means a syncpoint reached;
+		// ResolvedTs == ddlTs means the syncpoint is setted by DDL;
+		// ddlTs <= ddlExecutedTs means the DDL has been execed.
+		if c.status.ResolvedTs == c.status.CheckpointTs && c.status.ResolvedTs == c.ddlTs && c.ddlTs <= c.ddlExecutedTs {
+			log.Info("sync point reached by ddl", zap.Uint64("ResolvedTs", c.status.ResolvedTs), zap.Uint64("CheckpointTs", c.status.CheckpointTs), zap.Bool("updateResolvedTs", c.updateResolvedTs), zap.Uint64("ddlResolvedTs", c.ddlResolvedTs), zap.Uint64("ddlTs", c.ddlTs), zap.Uint64("ddlExecutedTs", c.ddlExecutedTs))
+			err := c.syncpointSink.SinkSyncpoint(ctx, c.id, c.status.CheckpointTs)
+			if err != nil {
+				log.Error("syncpoint sink fail", zap.Uint64("ResolvedTs", c.status.ResolvedTs), zap.Uint64("CheckpointTs", c.status.CheckpointTs), zap.Error(err))
+			}
+			c.ddlTs = 0
+		}
+	}
+	return nil
+}
+
 // calcResolvedTs update every changefeed's resolve ts and checkpoint ts.
 func (c *changeFeed) calcResolvedTs(ctx context.Context) error {
 	if c.ddlState != model.ChangeFeedSyncDML && c.ddlState != model.ChangeFeedWaitToExecDDL {
@@ -796,6 +840,7 @@ func (c *changeFeed) calcResolvedTs(ctx context.Context) error {
 	if len(c.ddlJobHistory) > 0 && minResolvedTs >= c.ddlJobHistory[0].BinlogInfo.FinishedTS {
 		minResolvedTs = c.ddlJobHistory[0].BinlogInfo.FinishedTS
 		c.ddlState = model.ChangeFeedWaitToExecDDL
+		c.ddlTs = minResolvedTs
 	}
 
 	// if downstream sink is the MQ sink, the MQ sink do not promise that checkpoint is less than globalResolvedTs
@@ -806,7 +851,15 @@ func (c *changeFeed) calcResolvedTs(ctx context.Context) error {
 
 	var tsUpdated bool
 
-	if minResolvedTs > c.status.ResolvedTs {
+	//syncpoint on
+	if c.info.SyncPointEnabled {
+		c.syncpointMutex.Lock()
+		if c.updateResolvedTs && minResolvedTs > c.status.ResolvedTs {
+			c.status.ResolvedTs = minResolvedTs
+			tsUpdated = true
+		}
+		c.syncpointMutex.Unlock()
+	} else if minResolvedTs > c.status.ResolvedTs {
 		c.status.ResolvedTs = minResolvedTs
 		tsUpdated = true
 	}
@@ -850,6 +903,37 @@ func (c *changeFeed) pullDDLJob() error {
 	return nil
 }
 
+// startSyncPeriod start a timer for every changefeed to create sync point by time
+func (c *changeFeed) startSyncPeriod(ctx context.Context, interval time.Duration) {
+	log.Debug("sync ticker start", zap.Duration("sync-interval", interval))
+	go func(ctx context.Context) {
+		ticker := time.NewTicker(interval)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				c.syncpointMutex.Lock()
+				c.updateResolvedTs = false
+				c.syncpointMutex.Unlock()
+			}
+		}
+	}(ctx)
+}
+
+func (c *changeFeed) stopSyncPointTicker() {
+	if c.syncCancel != nil {
+		c.syncCancel()
+		c.syncCancel = nil
+	}
+}
+
+func (c *changeFeed) startSyncPointTicker(ctx context.Context, interval time.Duration) {
+	var syncCtx context.Context
+	syncCtx, c.syncCancel = context.WithCancel(ctx)
+	c.startSyncPeriod(syncCtx, interval)
+}
+
 func (c *changeFeed) Close() {
 	err := c.ddlHandler.Close()
 	if err != nil {
@@ -859,5 +943,11 @@ func (c *changeFeed) Close() {
 	if err != nil {
 		log.Warn("failed to close owner sink", zap.Error(err))
 	}
+	if c.syncpointSink != nil {
+		err = c.syncpointSink.Close()
+		if err != nil {
+			log.Warn("failed to close owner sink", zap.Error(err))
+		}
+	}
 	log.Info("changefeed closed", zap.String("id", c.id))
 }

cdc/model/changefeed.go

@@ -81,6 +81,9 @@ type ChangeFeedInfo struct {
 	State    FeedState             `json:"state"`
 	ErrorHis []int64               `json:"history"`
 	Error    *RunningError         `json:"error"`
+
+	SyncPointEnabled  bool          `json:"sync-point-enabled"`
+	SyncPointInterval time.Duration `json:"sync-point-interval"`
 }
 
 var changeFeedIDRe *regexp.Regexp = regexp.MustCompile(`^[a-zA-Z0-9]+(\-[a-zA-Z0-9]+)*$`)

cdc/owner.go

@@ -333,13 +333,13 @@ func (o *Owner) newChangeFeed(
 	}
 	errCh := make(chan error, 1)
 
-	sink, err := sink.NewSink(ctx, id, info.SinkURI, filter, info.Config, info.Opts, errCh)
+	primarySink, err := sink.NewSink(ctx, id, info.SinkURI, filter, info.Config, info.Opts, errCh)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 	defer func() {
-		if resultErr != nil && sink != nil {
-			sink.Close()
+		if resultErr != nil && primarySink != nil {
+			primarySink.Close()
 		}
 	}()
 	go func() {
@@ -352,11 +352,19 @@ func (o *Owner) newChangeFeed(
 		cancel()
 	}()
 
-	err = sink.Initialize(ctx, sinkTableInfo)
+	err = primarySink.Initialize(ctx, sinkTableInfo)
 	if err != nil {
 		log.Error("error on running owner", zap.Error(err))
 	}
 
+	var syncpointSink sink.SyncpointSink
+	if info.SyncPointEnabled {
+		syncpointSink, err = sink.NewSyncpointSink(ctx, id, info.SinkURI)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+	}
+
 	cf = &changeFeed{
 		info:          info,
 		id:            id,
@@ -375,11 +383,16 @@ func (o *Owner) newChangeFeed(
 		ddlState:          model.ChangeFeedSyncDML,
 		ddlExecutedTs:     checkpointTs,
 		targetTs:          info.GetTargetTs(),
+		ddlTs:             0,
+		updateResolvedTs:  true,
+		startTimer:        make(chan bool),
+		syncpointSink:     syncpointSink,
+		syncCancel:        nil,
 		taskStatus:        processorsInfos,
 		taskPositions:     taskPositions,
 		etcdCli:           o.etcdClient,
 		filter:            filter,
-		sink:              sink,
+		sink:              primarySink,
 		cyclicEnabled:     info.Config.Cyclic.IsEnabled(),
 		lastRebalanceTime: time.Now(),
 	}
@@ -523,6 +536,19 @@ func (o *Owner) loadChangeFeeds(ctx context.Context) error {
 				zap.String("changefeed", changeFeedID), zap.Error(err))
 			continue
 		}
+
+		if newCf.info.SyncPointEnabled {
+			log.Info("syncpoint is on", zap.Bool("syncpoint", newCf.info.SyncPointEnabled))
+			//create the sync table
+			err := newCf.syncpointSink.CreateSynctable(ctx)
+			if err != nil {
+				return err
+			}
+			newCf.startSyncPointTicker(ctx, newCf.info.SyncPointInterval)
+		} else {
+			log.Info("syncpoint is off", zap.Bool("syncpoint", newCf.info.SyncPointEnabled))
+		}
+
 		o.changeFeeds[changeFeedID] = newCf
 		delete(o.stoppedFeeds, changeFeedID)
 	}
@@ -647,6 +673,16 @@ func (o *Owner) handleDDL(ctx context.Context) error {
 	return nil
 }
 
+// handleSyncPoint call handleSyncPoint of every changefeeds
+func (o *Owner) handleSyncPoint(ctx context.Context) error {
+	for _, cf := range o.changeFeeds {
+		if err := cf.handleSyncPoint(ctx); err != nil {
+			return errors.Trace(err)
+		}
+	}
+	return nil
+}
+
 // dispatchJob dispatches job to processors
 func (o *Owner) dispatchJob(ctx context.Context, job model.AdminJob) error {
 	cf, ok := o.changeFeeds[job.CfID]
@@ -804,8 +840,10 @@ func (o *Owner) handleAdminJob(ctx context.Context) error {
 			if err != nil {
 				return errors.Trace(err)
 			}
+			cf.stopSyncPointTicker()
 		case model.AdminRemove, model.AdminFinish:
 			if cf != nil {
+				cf.stopSyncPointTicker()
 				err := o.dispatchJob(ctx, job)
 				if err != nil {
 					return errors.Trace(err)
@@ -1029,6 +1067,11 @@ func (o *Owner) run(ctx context.Context) error {
 		return errors.Trace(err)
 	}
 
+	err = o.handleSyncPoint(ctx)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
 	err = o.handleAdminJob(ctx)
 	if err != nil {
 		return errors.Trace(err)