syncer(dm): init schemaTracker when syncer run

dm/pkg/schema/tracker.go

@@ -99,16 +99,25 @@ type DownstreamTableInfo struct {
 	WhereHandle *sqlmodel.WhereHandle
 }
 
-// NewTracker creates a new tracker. `sessionCfg` will be set as tracker's session variables if specified, or retrieve
+// NewDumpTracker simply returns an empty Tracker,
+// which should be followed by a subsequent initialization.
+func NewDumpTracker() *Tracker {
+	return &Tracker{}
+}
+
+// Init initializes the Tracker. `sessionCfg` will be set as tracker's session variables if specified, or retrieve
 // some variable from downstream using `downstreamConn`.
 // NOTE **sessionCfg is a reference to caller**.
-func NewTracker(
+func (tr *Tracker) Init(
 	ctx context.Context,
 	task string,
 	sessionCfg map[string]string,
 	downstreamConn *dbconn.DBConn,
 	logger log.Logger,
-) (*Tracker, error) {
+) error {
+	if tr == nil {
+		return nil
+	}
 	var (
 		err       error
 		storePath string
@@ -148,28 +157,28 @@ func NewTracker(
 			var ignoredColumn interface{}
 			rows, err2 := downstreamConn.QuerySQL(tctx, nil, fmt.Sprintf("SHOW VARIABLES LIKE '%s'", k))
 			if err2 != nil {
-				return nil, err2
+				return err2
 			}
 			if rows.Next() {
 				var value string
 				if err3 := rows.Scan(&ignoredColumn, &value); err3 != nil {
-					return nil, err3
+					return err3
 				}
 				sessionCfg[k] = value
 			}
 			// nolint:sqlclosecheck
 			if err2 = rows.Close(); err2 != nil {
-				return nil, err2
+				return err2
 			}
 			if err2 = rows.Err(); err2 != nil {
-				return nil, err2
+				return err2
 			}
 		}
 	}
 
 	storePath, err = newTmpFolderForTracker(task)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	rollbackHolder.Add(fr.FuncRollback{Name: "DeleteStorePath", Fn: func() {
 		_ = os.RemoveAll(storePath)
@@ -179,7 +188,7 @@ func NewTracker(
 		mockstore.WithStoreType(mockstore.EmbedUnistore),
 		mockstore.WithPath(storePath))
 	if err != nil {
-		return nil, err
+		return err
 	}
 	rollbackHolder.Add(fr.FuncRollback{Name: "CloseStore", Fn: func() {
 		_ = store.Close()
@@ -190,13 +199,13 @@ func NewTracker(
 
 	dom, err = session.BootstrapSession(store)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	rollbackHolder.Add(fr.FuncRollback{Name: "CloseDomain", Fn: dom.Close})
 
 	se, err = session.CreateSession(store)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	rollbackHolder.Add(fr.FuncRollback{Name: "CloseSession", Fn: se.Close})
 
@@ -216,13 +225,13 @@ func NewTracker(
 				log.L().Warn("can not set this variable", zap.Error(err))
 				continue
 			}
-			return nil, err
+			return err
 		}
 	}
 	for k, v := range globalVarsToSet {
 		err = se.GetSessionVars().SetSystemVarWithRelaxedValidation(k, v)
 		if err != nil {
-			return nil, err
+			return err
 		}
 	}
 	// skip DDL test https://github.com/pingcap/tidb/pull/33079
@@ -233,22 +242,39 @@ func NewTracker(
 	// exist by default. So we need to drop it first.
 	err = dom.DDL().DropSchema(se, model.NewCIStr("test"))
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	// init downstreamTracker
 	dsTracker := &downstreamTracker{
 		downstreamConn: downstreamConn,
 		tableInfos:     make(map[string]*DownstreamTableInfo),
 	}
+	tr.Lock()
+	defer tr.Unlock()
+	tr.storePath = storePath
+	tr.store = store
+	tr.dom = dom
+	tr.se = se
+	tr.dsTracker = dsTracker
+	tr.closed.Store(false)
+	return nil
+}
 
-	return &Tracker{
-		storePath: storePath,
-		store:     store,
-		dom:       dom,
-		se:        se,
-		dsTracker: dsTracker,
-	}, nil
+// NewTracker creates a new tracker. It's preserved for test.
+func NewTracker(
+	ctx context.Context,
+	task string,
+	sessionCfg map[string]string,
+	downstreamConn *dbconn.DBConn,
+	logger log.Logger,
+) (*Tracker, error) {
+	tr := NewDumpTracker()
+	err := tr.Init(ctx, task, sessionCfg, downstreamConn, logger)
+	if err != nil {
+		return nil, err
+	}
+	return tr, nil
 }
 
 func newTmpFolderForTracker(task string) (string, error) {
@@ -392,10 +418,18 @@ func (tr *Tracker) Close() error {
 	if !tr.closed.CAS(false, true) {
 		return nil
 	}
-	tr.se.Close()
-	tr.dom.Close()
-	if err := tr.store.Close(); err != nil {
-		return err
+	// Build of the Tracker and the initialization is divided.
+	// these fields can possibly be nil if the Tracker is closed before the initialization.
+	if tr.se != nil {
+		tr.se.Close()
+	}
+	if tr.dom != nil {
+		tr.dom.Close()
+	}
+	if tr.store != nil {
+		if err := tr.store.Close(); err != nil {
+			return err
+		}
 	}
 	return os.RemoveAll(tr.storePath)
 }

dm/syncer/syncer.go

@@ -391,6 +391,8 @@ func (s *Syncer) Init(ctx context.Context) (err error) {
 	}
 	s.sessCtx = utils.NewSessionCtx(vars)
 	s.exprFilterGroup = NewExprFilterGroup(s.tctx, s.sessCtx, s.cfg.ExprFilter)
+	// create an empty Tracker and will be initialized in `Run`
+	s.schemaTracker = schema.NewDumpTracker()
 
 	if len(s.cfg.ColumnMappingRules) > 0 {
 		s.columnMapping, err = cm.NewMapping(s.cfg.CaseSensitive, s.cfg.ColumnMappingRules)
@@ -1696,7 +1698,13 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		}
 	}
 
-	s.schemaTracker, err = schema.NewTracker(ctx, s.cfg.Name, s.cfg.To.Session, s.downstreamTrackConn, s.tctx.L())
+	if s.schemaTracker == nil {
+		s.schemaTracker, err = schema.NewTracker(ctx, s.cfg.Name, s.cfg.To.Session, s.downstreamTrackConn, s.tctx.L())
+	} else {
+		// prevent creating new Tracker on `Run` in order to avoid
+		// two different Trackers are invoked in the validator and the syncer.
+		err = s.schemaTracker.Init(ctx, s.cfg.Name, s.cfg.To.Session, s.downstreamTrackConn, s.tctx.L())
+	}
 	if err != nil {
 		return terror.ErrSchemaTrackerInit.Delegate(err)
 	}