syncer(dm): init schemaTracker when syncer run

dm/pkg/schema/tracker.go

@@ -73,9 +73,14 @@ func init() {
 
 // Tracker is used to track schema locally.
 type Tracker struct {
-	// we're using an embedded tidb, there's no need to sync operations on it, but we may recreate(drop and create)
-	// a table such as when checkpoint rollback, we need to make sure others(validator for now) can't see the table
-	// is deleted. so we add an extra layer of synchronization for GetTableInfo/RecreateTables for now.
+	// The Tracker is an embedded tidb in essence, where there was basically no parallel operation at the beginning.
+	// However, since the validator is introduced and heavily dependent on the Tracker, we need to make sure
+	// the synchronization between the reading from the validator and the modification from the syncer (e.g.
+	// when the checkpoint is being rolled back, we have to make sure the validator can still vision the original tables)
+	// From this point, we add an extra layer of the synchronization for the following operations:
+	// 1. GetTableInfo: the validator reads table infos.
+	// 2. Init: when the syncer restarts, it may re-initialize the Tracker while the validator may read the Tracker at the same time.
+	// 3. Close: Being similar as above, the validator can read the Tracker while the syncer is closing the Tracker.
 	sync.RWMutex
 	storePath string
 	store     kv.Storage
@@ -99,16 +104,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
+// NewTracker simply returns an empty Tracker,
+// which should be followed by an initialization before used.
+func NewTracker() *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 +162,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 +193,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 +204,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 +230,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 +247,40 @@ 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
+// NewTestTracker creates an empty Tracker and initializes it subsequently.
+// It is useful for test.
+func NewTestTracker(
+	ctx context.Context,
+	task string,
+	sessionCfg map[string]string,
+	downstreamConn *dbconn.DBConn,
+	logger log.Logger,
+) (*Tracker, error) {
+	tr := NewTracker()
+	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 +424,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/pkg/schema/tracker_test.go

@@ -92,28 +92,28 @@ func (s *trackerSuite) TestTiDBAndSessionCfg(c *C) {
 	dbConn := dbconn.NewDBConn(s.cfg, baseConn)
 	// user give correct session config
 
-	t, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
+	t, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	err = t.Close()
 	c.Assert(err, IsNil)
 
 	// user give wrong session session, will return error
 	sessionCfg := map[string]string{"sql_mode": "HaHa"}
-	_, err = NewTracker(context.Background(), "test-tracker", sessionCfg, dbConn, dlog.L())
+	_, err = NewTestTracker(context.Background(), "test-tracker", sessionCfg, dbConn, dlog.L())
 	c.Assert(err, NotNil)
 
 	// discover session config failed, will return error
 	mock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(
 		sqlmock.NewRows([]string{"Variable_name", "Value"}).
 			AddRow("sql_mode", "HaHa"))
-	_, err = NewTracker(context.Background(), "test-tracker", nil, dbConn, dlog.L())
+	_, err = NewTestTracker(context.Background(), "test-tracker", nil, dbConn, dlog.L())
 	c.Assert(err, NotNil)
 
 	// empty or default config in downstream
 	mock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(
 		sqlmock.NewRows([]string{"Variable_name", "Value"}).
 			AddRow("sql_mode", defaultTestSessionCfg["sql_mode"]))
-	tracker, err := NewTracker(context.Background(), "test-tracker", nil, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", nil, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 	err = tracker.Exec(context.Background(), "", "create database testdb;")
@@ -125,7 +125,7 @@ func (s *trackerSuite) TestTiDBAndSessionCfg(c *C) {
 	mock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(
 		sqlmock.NewRows([]string{"Variable_name", "Value"}).
 			AddRow("sql_mode", "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE"))
-	tracker, err = NewTracker(context.Background(), "test-tracker", nil, dbConn, dlog.L())
+	tracker, err = NewTestTracker(context.Background(), "test-tracker", nil, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 	c.Assert(tracker.se.GetSessionVars().SQLMode.HasOnlyFullGroupBy(), IsTrue)
@@ -144,7 +144,7 @@ func (s *trackerSuite) TestTiDBAndSessionCfg(c *C) {
 	// user set session config, get tracker config from downstream
 	// no `STRICT_TRANS_TABLES`, no error now
 	sessionCfg = map[string]string{"sql_mode": "NO_ZERO_DATE,NO_ZERO_IN_DATE,ANSI_QUOTES"}
-	tracker, err = NewTracker(context.Background(), "test-tracker", sessionCfg, dbConn, dlog.L())
+	tracker, err = NewTestTracker(context.Background(), "test-tracker", sessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 
@@ -179,7 +179,7 @@ func (s *trackerSuite) TestTiDBAndSessionCfg(c *C) {
 	err = tracker.Close()
 	c.Assert(err, IsNil)
 
-	tracker, err = NewTracker(context.Background(), "test-tracker", sessionCfg, dbConn, dlog.L())
+	tracker, err = NewTestTracker(context.Background(), "test-tracker", sessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -203,7 +203,7 @@ func (s *trackerSuite) TestDDL(c *C) {
 		Name:   "foo",
 	}
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -277,7 +277,7 @@ func (s *trackerSuite) TestDDL(c *C) {
 func (s *trackerSuite) TestGetSingleColumnIndices(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -320,7 +320,7 @@ func (s *trackerSuite) TestGetSingleColumnIndices(c *C) {
 func (s *trackerSuite) TestCreateSchemaIfNotExists(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -352,7 +352,7 @@ func (s *trackerSuite) TestCreateSchemaIfNotExists(c *C) {
 func (s *trackerSuite) TestMultiDrop(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -404,7 +404,7 @@ func (s *trackerSuite) TestCreateTableIfNotExists(c *C) {
 		Name:   "foo",
 	}
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -486,7 +486,7 @@ func (s *trackerSuite) TestCreateTableIfNotExists(c *C) {
 
 func (s *trackerSuite) TestBatchCreateTableIfNotExist(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -604,7 +604,7 @@ func (s *trackerSuite) TestAllSchemas(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
 	ctx := context.Background()
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, s.dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -700,7 +700,7 @@ func (s *trackerSuite) TestNotSupportedVariable(c *C) {
 	oldSessionVar := map[string]string{
 		"tidb_enable_change_column_type": "ON",
 	}
-	tracker, err := NewTracker(context.Background(), "test-tracker", oldSessionVar, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", oldSessionVar, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -720,7 +720,7 @@ func (s *trackerSuite) TestInitDownStreamSQLModeAndParser(c *C) {
 	baseConn := conn.NewBaseConn(con, nil)
 	dbConn := dbconn.NewDBConn(s.cfg, baseConn)
 
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -757,7 +757,7 @@ func (s *trackerSuite) TestGetDownStreamIndexInfo(c *C) {
 	c.Assert(err, IsNil)
 	baseConn := conn.NewBaseConn(con, nil)
 	dbConn := dbconn.NewDBConn(s.cfg, baseConn)
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -799,7 +799,7 @@ func (s *trackerSuite) TestReTrackDownStreamIndex(c *C) {
 	c.Assert(err, IsNil)
 	baseConn := conn.NewBaseConn(con, nil)
 	dbConn := dbconn.NewDBConn(s.cfg, baseConn)
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -891,7 +891,7 @@ func (s *trackerSuite) TestVarchar20000(c *C) {
 	c.Assert(err, IsNil)
 	baseConn := conn.NewBaseConn(con, nil)
 	dbConn := dbconn.NewDBConn(s.cfg, baseConn)
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()
@@ -931,7 +931,7 @@ func (s *trackerSuite) TestPlacementRule(c *C) {
 	c.Assert(err, IsNil)
 	baseConn := conn.NewBaseConn(con, nil)
 	dbConn := dbconn.NewDBConn(s.cfg, baseConn)
-	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
+	tracker, err := NewTestTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn, dlog.L())
 	c.Assert(err, IsNil)
 	defer func() {
 		err = tracker.Close()

dm/syncer/checkpoint_test.go

@@ -83,7 +83,7 @@ func (s *testCheckpointSuite) SetUpSuite(c *C) {
 		}
 	)
 
-	s.tracker, err = schema.NewTracker(context.Background(), s.cfg.Name, defaultTestSessionCfg, nil, dlog.L())
+	s.tracker, err = schema.NewTestTracker(context.Background(), s.cfg.Name, defaultTestSessionCfg, nil, dlog.L())
 	c.Assert(err, IsNil)
 }
 
@@ -499,7 +499,7 @@ func TestRemoteCheckPointLoadIntoSchemaTracker(t *testing.T) {
 	dbConn, err := db.Conn(ctx)
 	require.NoError(t, err)
 	downstreamTrackConn := dbconn.NewDBConn(cfg, conn.NewBaseConn(dbConn, &retry.FiniteRetryStrategy{}))
-	schemaTracker, err := schema.NewTracker(ctx, cfg.Name, defaultTestSessionCfg, downstreamTrackConn, dlog.L())
+	schemaTracker, err := schema.NewTestTracker(ctx, cfg.Name, defaultTestSessionCfg, downstreamTrackConn, dlog.L())
 	require.NoError(t, err)
 	defer schemaTracker.Close() //nolint
 

dm/syncer/data_validator_test.go

@@ -357,7 +357,7 @@ func TestValidatorDoValidate(t *testing.T) {
 	dbConn, err := db.Conn(context.Background())
 	require.NoError(t, err)
 	syncerObj.downstreamTrackConn = dbconn.NewDBConn(cfg, conn.NewBaseConn(dbConn, &retry.FiniteRetryStrategy{}))
-	syncerObj.schemaTracker, err = schema.NewTracker(context.Background(), cfg.Name, defaultTestSessionCfg, syncerObj.downstreamTrackConn, log.L())
+	syncerObj.schemaTracker, err = schema.NewTestTracker(context.Background(), cfg.Name, defaultTestSessionCfg, syncerObj.downstreamTrackConn, log.L())
 	defer syncerObj.schemaTracker.Close()
 	require.NoError(t, err)
 	require.NoError(t, syncerObj.schemaTracker.CreateSchemaIfNotExists(schemaName))