tracker(dm): respect CHARSET when build TableInfo (#4638) (#4646)

close #4637

dm/pkg/schema/tracker.go

@@ -34,7 +34,6 @@ import (
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/pingcap/tidb/types"
-	"github.com/pingcap/tidb/util/mock"
 	"go.uber.org/zap"
 
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
@@ -47,8 +46,6 @@ import (
 const (
 	// TiDBClusteredIndex is the variable name for clustered index.
 	TiDBClusteredIndex = "tidb_enable_clustered_index"
-	// downstream mock table id, consists of serial numbers of letters.
-	mockTableID = 121402101900011104
 )
 
 var (
@@ -461,10 +458,11 @@ func (tr *Tracker) getTableInfoByCreateStmt(tctx *tcontext.Context, tableID stri
 		return nil, dmterror.ErrSchemaTrackerInvalidCreateTableStmt.Delegate(err, createStr)
 	}
 
-	ti, err := ddl.MockTableInfo(mock.NewContext(), stmtNode.(*ast.CreateTableStmt), mockTableID)
+	ti, err := ddl.BuildTableInfoFromAST(stmtNode.(*ast.CreateTableStmt))
 	if err != nil {
 		return nil, dmterror.ErrSchemaTrackerCannotMockDownstreamTable.Delegate(err, createStr)
 	}
+	ti.State = model.StatePublic
 	return ti, nil
 }
 

dm/pkg/schema/tracker_test.go

@@ -740,7 +740,7 @@ func (s *trackerSuite) TestGetDownStreamIndexInfo(c *C) {
 	delete(tracker.dsTracker.tableInfos, tableID)
 }
 
-func (s *trackerSuite) TestGetAvailableDownStreanUKIndexInfo(c *C) {
+func (s *trackerSuite) TestGetAvailableDownStreamUKIIndexInfo(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
 
 	// origin table info
@@ -928,3 +928,39 @@ func (s *trackerSuite) TestReTrackDownStreamIndex(c *C) {
 	_, ok = tracker.dsTracker.tableInfos[tableID]
 	c.Assert(ok, IsTrue)
 }
+
+func (s *trackerSuite) TestVarchar20000(c *C) {
+	log.SetLevel(zapcore.ErrorLevel)
+
+	// origin table info
+	p := parser.New()
+	node, err := p.ParseOneStmt("create table t(c varchar(20000)) charset=utf8", "", "")
+	c.Assert(err, IsNil)
+	oriTi, err := ddl.BuildTableInfoFromAST(node.(*ast.CreateTableStmt))
+	c.Assert(err, IsNil)
+
+	// tracker and sqlmock
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
+	con, err := db.Conn(context.Background())
+	c.Assert(err, IsNil)
+	baseConn := conn.NewBaseConn(con, nil)
+	dbConn := &dbconn.DBConn{Cfg: s.cfg, BaseConn: baseConn}
+	tracker, err := NewTracker(context.Background(), "test-tracker", defaultTestSessionCfg, dbConn)
+	c.Assert(err, IsNil)
+
+	mock.ExpectBegin()
+	mock.ExpectExec(fmt.Sprintf("SET SESSION SQL_MODE = '%s'", mysql.DefaultSQLMode)).WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectCommit()
+
+	tableID := "`test`.`test`"
+
+	mock.ExpectQuery("SHOW CREATE TABLE " + tableID).WillReturnRows(
+		sqlmock.NewRows([]string{"Table", "Create Table"}).
+			AddRow("test", "create table t(c varchar(20000)) charset=utf8"))
+	_, err = tracker.GetDownStreamTableInfo(tcontext.Background(), tableID, oriTi)
+	c.Assert(err, IsNil)
+	_, ok := tracker.dsTracker.tableInfos[tableID]
+	c.Assert(ok, IsTrue)
+}

dm/tests/downstream_diff_index/data/db1.increment.sql

@@ -1,3 +1,4 @@
 use downstream_diff_index1;
 update t1 set c3 = '111' where c1 = 1;
 delete from t1 where c1 = 2;
+insert into varchar20000 values (1, '1');

dm/tests/downstream_diff_index/data/db1.prepare.sql

@@ -5,3 +5,4 @@ create table t1 (c1 int, c2 int, c3 varchar(10), primary key(c1));
 insert into t1 values(1, 1, '1');
 insert into t1 values(2, 2, '2');
 insert into t1 values(3, 3, '3');
+create table varchar20000 (c int primary key, c2 varchar(20000)) charset=utf8;

dm/tests/downstream_diff_index/run.sh

@@ -66,6 +66,8 @@ function run() {
 	# check delete data
 	run_sql_tidb_with_retry "select count(1) from ${db}.${tb} where c1=1;" "count(1): 1"
 	check_log_contain_with_retry '\[DeleteWhereColumnsCheck\] \[Columns="\[c3\]"\]' $WORK_DIR/worker2/log/dm-worker.log
+
+	run_sql_tidb_with_retry "select count(1) from ${db}.varchar20000 where c=1;" "count(1): 1"
 }
 
 cleanup_data downstream_diff_index