TiCDC supports a snapshot-map in replication

README.md

@@ -68,3 +68,5 @@ for details on submitting patches and the contribution workflow.
 ## License
 
 TiCDC is under the Apache 2.0 license. See the [LICENSE](./LICENSE) file for details.
+
+## Branch For dev

cdc/changefeed.go

@@ -15,6 +15,7 @@ package cdc
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"math"
 	"time"
@@ -35,6 +36,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const defaultSyncInterval = time.Minute * 10 //default Interval for record syncpoint
+
 type tableIDMap = map[model.TableID]struct{}
 
 // OwnerDDLHandler defines the ddl handler for Owner
@@ -78,14 +81,19 @@ 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
+	updateResolvedTs bool
+	startTimer       chan bool
+	syncDB           *sql.DB
+	syncCancel       context.CancelFunc
+	taskStatus       model.ProcessorsInfos
+	taskPositions    map[model.CaptureID]*model.TaskPosition
+	filter           *filter.Filter
+	sink             sink.Sink
+	scheduler        scheduler.Scheduler
 
 	cyclicEnabled bool
 
@@ -712,6 +720,26 @@ func (c *changeFeed) calcResolvedTs(ctx context.Context) error {
 		}
 	}
 
+	//sync-point on
+	if c.info.SyncPoint {
+		//if (c.status.ResolvedTs == c.status.CheckpointTs && (c.updateResolvedTs == false || c.status.ResolvedTs == c.ddlResolvedTs)) || c.status.ResolvedTs == 0 {
+		if c.status.ResolvedTs == c.status.CheckpointTs && !c.updateResolvedTs {
+			log.Info("achive the sync point with timer", 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))
+			c.updateResolvedTs = true
+			c.sinkSyncpoint(ctx)
+		}
+
+		if c.status.ResolvedTs == 0 {
+			c.updateResolvedTs = true
+		}
+
+		if c.status.ResolvedTs == c.status.CheckpointTs && c.status.ResolvedTs == c.ddlTs {
+			log.Info("achive the sync point with 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))
+			c.sinkSyncpoint(ctx)
+			c.ddlTs = 0
+		}
+	}
+
 	if len(c.taskPositions) < len(c.taskStatus) {
 		log.Debug("skip update resolved ts",
 			zap.Int("taskPositions", len(c.taskPositions)),
@@ -782,6 +810,7 @@ func (c *changeFeed) calcResolvedTs(ctx context.Context) error {
 
 		if minResolvedTs > c.ddlResolvedTs {
 			minResolvedTs = c.ddlResolvedTs
+			//c.ddlTs = c.ddlResolvedTs
 		}
 	}
 
@@ -793,6 +822,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
@@ -803,7 +833,19 @@ func (c *changeFeed) calcResolvedTs(ctx context.Context) error {
 
 	var tsUpdated bool
 
-	if minResolvedTs > c.status.ResolvedTs {
+	//syncpoint on
+	if c.info.SyncPoint {
+		if c.updateResolvedTs && minResolvedTs > c.status.ResolvedTs {
+			//prevResolvedTs := c.status.ResolvedTs
+			c.status.ResolvedTs = minResolvedTs
+			tsUpdated = true
+			/*if prevResolvedTs == c.status.CheckpointTs || prevResolvedTs == 0 {
+				//到达sync point
+				//todo 需要重新开始启动计时
+				c.startTimer <- true
+			}*/
+		}
+	} else if minResolvedTs > c.status.ResolvedTs {
 		c.status.ResolvedTs = minResolvedTs
 		tsUpdated = true
 	}
@@ -847,6 +889,84 @@ 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.updateResolvedTs = false
+			}
+		}
+	}(ctx)
+}
+
+//createSynctable create a sync table to record the
+func (c *changeFeed) createSynctable(ctx context.Context) error {
+	database := "TiCDC"
+	tx, err := c.syncDB.BeginTx(ctx, nil)
+	if err != nil {
+		log.Info("create sync table: begin Tx fail")
+		return err
+	}
+	_, err = tx.Exec("CREATE DATABASE IF NOT EXISTS " + database)
+	//_,err := c.syncDB.Exec("CREATE DATABASE IF NOT EXISTS " + database)
+	if err != nil {
+		tx.Rollback()
+		return err
+	}
+	_, err = tx.Exec("USE " + database)
+	if err != nil {
+		tx.Rollback()
+		return err
+	}
+	_, err = tx.Exec("CREATE TABLE  IF NOT EXISTS syncpoint (cf varchar(255),primary_ts varchar(18),secondary_ts varchar(18),PRIMARY KEY ( `primary_ts` ) )")
+	if err != nil {
+		tx.Rollback()
+		return err
+	}
+	err = tx.Commit()
+	return err
+}
+
+//sinkSyncpoint record the syncpoint(a map with ts) in downstream db
+func (c *changeFeed) sinkSyncpoint(ctx context.Context) error {
+	tx, err := c.syncDB.BeginTx(ctx, nil)
+	if err != nil {
+		log.Info("sync table: begin Tx fail")
+		return err
+	}
+	row := tx.QueryRow("select @@tidb_current_ts")
+	var slaveTs string
+	err = row.Scan(&slaveTs)
+	if err != nil {
+		log.Info("sync table: get tidb_current_ts err")
+		tx.Rollback()
+		return err
+	}
+	tx.Exec("insert into TiCDC.syncpoint(cf, primary_ts, secondary_ts) VALUES (?,?,?)", c.id, c.status.CheckpointTs, slaveTs)
+	//tx.Exec("insert into TiCDC.syncpoint( master_ts, slave_ts) VALUES (?,?)", c.status.CheckpointTs, 0)
+	tx.Commit() //TODO deal with error
+	return nil
+}
+
+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 {

cdc/model/changefeed.go

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

cdc/owner.go

@@ -333,7 +333,7 @@ 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)
 	}
@@ -352,7 +352,12 @@ 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))
+	}
+
+	syncDB, err := sink.NewSyncpointSinklink(ctx, info, id)
 	if err != nil {
 		log.Error("error on running owner", zap.Error(err))
 	}
@@ -375,11 +380,16 @@ func (o *Owner) newChangeFeed(
 		ddlState:          model.ChangeFeedSyncDML,
 		ddlExecutedTs:     checkpointTs,
 		targetTs:          info.GetTargetTs(),
+		ddlTs:             0,
+		updateResolvedTs:  true,
+		startTimer:        make(chan bool),
+		syncDB:            syncDB,
+		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 +533,25 @@ func (o *Owner) loadChangeFeeds(ctx context.Context) error {
 				zap.String("changefeed", changeFeedID), zap.Error(err))
 			continue
 		}
+
+		if newCf.info.SyncPoint {
+			log.Info("syncpoint is on", zap.Bool("syncpoint", newCf.info.SyncPoint))
+			//create the sync table
+			err := newCf.createSynctable(ctx)
+			if err != nil {
+				return err
+			}
+
+			//start SyncPeriod for create the sync point
+			syncInterval, err := time.ParseDuration(newCf.info.SyncInterval)
+			if err != nil {
+				syncInterval = defaultSyncInterval
+			}
+			newCf.startSyncPointTicker(ctx, syncInterval)
+		} else {
+			log.Info("syncpoint is off", zap.Bool("syncpoint", newCf.info.SyncPoint))
+		}
+
 		o.changeFeeds[changeFeedID] = newCf
 		delete(o.stoppedFeeds, changeFeedID)
 	}
@@ -804,6 +833,7 @@ 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 {
 				err := o.dispatchJob(ctx, job)
@@ -853,6 +883,7 @@ func (o *Owner) handleAdminJob(ctx context.Context) error {
 					return errors.Trace(err)
 				}
 			}
+			cf.stopSyncPointTicker()
 		case model.AdminResume:
 			// resume changefeed must read checkpoint from ChangeFeedStatus
 			if cerror.ErrChangeFeedNotExists.Equal(err) {
@@ -1295,3 +1326,48 @@ func (o *Owner) startCaptureWatcher(ctx context.Context) {
 		}
 	}()
 }
+
+/*func (o *Owner) configureSinkURI(ctx context.Context, dsnCfg *dmysql.Config, tz *time.Location) (string, error) {
+if dsnCfg.Params == nil {
+	dsnCfg.Params = make(map[string]string, 1)
+}
+dsnCfg.DBName = ""
+dsnCfg.InterpolateParams = true
+dsnCfg.MultiStatements = true
+dsnCfg.Params["time_zone"] = fmt.Sprintf(`"%s"`, tz.String())
+
+testDB, err := sql.Open("mysql", dsnCfg.FormatDSN())
+if err != nil {
+	return "", errors.Annotate(err, "fail to open MySQL connection when configuring sink")
+}
+defer testDB.Close()
+log.Debug("Opened connection to configure some tidb special parameters")
+
+var variableName string
+var autoRandomInsertEnabled string
+queryStr := "show session variables like 'allow_auto_random_explicit_insert';"
+err = testDB.QueryRowContext(ctx, queryStr).Scan(&variableName, &autoRandomInsertEnabled)
+if err != nil && err != sql.ErrNoRows {
+	return "", errors.Annotate(err, "fail to query sink for support of auto-random")
+}
+if err == nil && (autoRandomInsertEnabled == "off" || autoRandomInsertEnabled == "0") {
+	dsnCfg.Params["allow_auto_random_explicit_insert"] = "1"
+	log.Debug("Set allow_auto_random_explicit_insert to 1")
+}
+
+var txnMode string
+queryStr = "show session variables like 'tidb_txn_mode';"
+err = testDB.QueryRowContext(ctx, queryStr).Scan(&variableName, &txnMode)
+if err != nil && err != sql.ErrNoRows {
+	return "", errors.Annotate(err, "fail to query sink for txn mode")
+}
+/*if err == nil {
+	dsnCfg.Params["tidb_txn_mode"] = params.tidbTxnMode
+}*/
+
+/*	dsnClone := dsnCfg.Clone()
+	dsnClone.Passwd = "******"
+	log.Info("sink uri is configured", zap.String("format dsn", dsnClone.FormatDSN()))
+
+	return dsnCfg.FormatDSN(), nil
+}*/