resolve comments

cdc/entry/mounter_test.go

@@ -1199,13 +1199,11 @@ func TestDecodeEventIgnoreRow(t *testing.T) {
 
 func TestBuildTableInfo(t *testing.T) {
 	cases := []struct {
-		caseNumber          int
 		origin              string
 		recovered           string
 		recoveredWithNilCol string
 	}{
 		{
-			1,
 			"CREATE TABLE t1 (c INT PRIMARY KEY)",
 			"CREATE TABLE `BuildTiDBTableInfo` (\n" +
 				"  `c` int(0) NOT NULL,\n" +
@@ -1217,7 +1215,6 @@ func TestBuildTableInfo(t *testing.T) {
 				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
 		},
 		{
-			2,
 			"CREATE TABLE t1 (" +
 				" c INT UNSIGNED," +
 				" c2 VARCHAR(10) NOT NULL," +
@@ -1239,7 +1236,6 @@ func TestBuildTableInfo(t *testing.T) {
 				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
 		},
 		{
-			3,
 			"CREATE TABLE t1 (" +
 				" c INT UNSIGNED," +
 				" gen INT AS (c+1) VIRTUAL," +
@@ -1265,7 +1261,6 @@ func TestBuildTableInfo(t *testing.T) {
 				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
 		},
 		{
-			4,
 			"CREATE TABLE `t1` (" +
 				"  `a` int(11) NOT NULL," +
 				"  `b` int(11) DEFAULT NULL," +
@@ -1287,8 +1282,7 @@ func TestBuildTableInfo(t *testing.T) {
 				"  PRIMARY KEY (`a`(0)) /*T![clustered_index] CLUSTERED */\n" +
 				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
 		},
-		{
-			5,
+		{ // This case is to check the primary key is correctly identified by BuildTiDBTableInfo
 			"CREATE TABLE your_table (" +
 				" id INT NOT NULL," +
 				" name VARCHAR(50) NOT NULL," +
@@ -1335,7 +1329,7 @@ func TestBuildTableInfo(t *testing.T) {
 		require.NotNil(t, handle.UniqueNotNullIdx)
 		require.Equal(t, c.recovered, showCreateTable(t, recoveredTI))
 		// make sure BuildTiDBTableInfo indentify the correct primary key
-		if c.caseNumber == 5 {
+		if i == 5 {
 			inexes := recoveredTI.Indices
 			primaryCount := 0
 			for i := range inexes {
@@ -1387,7 +1381,7 @@ func TestNewDMRowChange(t *testing.T) {
 		},
 	}
 	p := parser.New()
-	for _, c := range cases {
+	for i, c := range cases {
 		stmt, err := p.ParseOneStmt(c.origin, "", "")
 		require.NoError(t, err)
 		originTI, err := ddl.BuildTableInfoFromAST(stmt.(*ast.CreateTableStmt))

cdc/model/sink.go

@@ -584,23 +584,26 @@ func BuildTiDBTableInfo(columns []*Column, indexColumns [][]int) *model.TableInf
 			continue
 		}
 		if firstCol.Flag.IsPrimaryKey() {
-			// If all columns in the index are primary key, then the index is primary key.
-			if CheckIfIndexesPrimaryKey(colOffsets, columns) {
-				indexInfo.Primary = true
-			}
 			indexInfo.Unique = true
 		}
 		if firstCol.Flag.IsUniqueKey() {
 			indexInfo.Unique = true
 		}
 
+		isPrimary := true
 		for _, offset := range colOffsets {
-			col := ret.Columns[offset]
+			col := columns[offset]
+			// When only all columns in the index are primary key, then the index is primary key.
+			if col == nil || !col.Flag.IsPrimaryKey() {
+				isPrimary = false
+			}
 
+			tiCol := ret.Columns[offset]
 			indexCol := &model.IndexColumn{}
-			indexCol.Name = col.Name
+			indexCol.Name = tiCol.Name
 			indexCol.Offset = offset
 			indexInfo.Columns = append(indexInfo.Columns, indexCol)
+			indexInfo.Primary = isPrimary
 		}
 
 		// TODO: revert the "all column set index related flag" to "only the
@@ -611,17 +614,6 @@ func BuildTiDBTableInfo(columns []*Column, indexColumns [][]int) *model.TableInf
 	return ret
 }
 
-// CheckIfIndexesPrimaryKey checks if all columns in a index is primary key
-func CheckIfIndexesPrimaryKey(columnOffsets []int, columns []*Column) bool {
-	for _, offset := range columnOffsets {
-		column := columns[offset]
-		if column == nil || !column.Flag.IsPrimaryKey() {
-			return false
-		}
-	}
-	return true
-}
-
 // ColumnValueString returns the string representation of the column value
 func ColumnValueString(c interface{}) string {
 	var data string