Skip to content

Commit

Permalink
checker(dm): handle expression UNIQUE index (#9353) (#9363)
Browse files Browse the repository at this point in the history 
close #9247
  • Loading branch information
@ti-chi-bot
ti-chi-bot authored Jul 13, 2023
1 parent 1225170 commit f8bcbba
Show file tree
Hide file tree
Showing 2 changed files with 120 additions and 1 deletion.
104 changes: 104 additions & 0 deletions dm/pkg/checker/table_structure_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -888,3 +888,107 @@ PRIMARY KEY ("c")
mock.ExpectQuery("SHOW CREATE TABLE `test-db`.`test-table-1`").WillReturnRows(createTableRow)
return mock
}

func TestExpressionUK(t *testing.T) {
db, mock, err := sqlmock.New()
require.NoError(t, err)
downDB, downMock, err := sqlmock.New()
require.NoError(t, err)
ctx := context.Background()

// test same table structure

maxConnectionsRow := sqlmock.NewRows([]string{"Variable_name", "Value"}).
AddRow("max_connections", "2")
mock.ExpectQuery("SHOW VARIABLES LIKE 'max_connections'").WillReturnRows(maxConnectionsRow)
sqlModeRow := sqlmock.NewRows([]string{"Variable_name", "Value"}).
AddRow("sql_mode", "ANSI_QUOTES")
mock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(sqlModeRow)
createTableRow := sqlmock.NewRows([]string{"Table", "Create Table"}).
AddRow("test-table-1", `CREATE TABLE "test-table-1" (
"c" int(11) NOT NULL,
"c2" int(11) NOT NULL,
PRIMARY KEY ("c"),
UNIQUE KEY "uk" (("c2"+1), "c")
) ENGINE=InnoDB`)
mock.ExpectQuery("SHOW CREATE TABLE `test-db`.`test-table-1`").WillReturnRows(createTableRow)
sqlModeRow2 := sqlmock.NewRows([]string{"Variable_name", "Value"}).
AddRow("sql_mode", "ANSI_QUOTES")
downMock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(sqlModeRow2)
createTableRow2 := sqlmock.NewRows([]string{"Table", "Create Table"}).
AddRow("test-table-1", `CREATE TABLE "test-table-1" (
"c" int(11) NOT NULL,
"c2" int(11) NOT NULL,
PRIMARY KEY ("c"),
UNIQUE KEY "uk" (("c2"+1), "c")
) ENGINE=InnoDB`)
downMock.ExpectQuery("SHOW CREATE TABLE `test-db`.`test-table-1`").WillReturnRows(createTableRow2)

checker := NewTablesChecker(
map[string]*conn.BaseDB{"test-source": conn.NewBaseDBForTest(db)},
conn.NewBaseDBForTest(downDB),
map[string]map[filter.Table][]filter.Table{
"test-source": {
{Schema: "test-db", Name: "test-table-1"}: {
{Schema: "test-db", Name: "test-table-1"},
},
},
},
nil,
1)
result := checker.Check(ctx)
require.Equal(t, StateSuccess, result.State)
require.NoError(t, mock.ExpectationsWereMet())
require.NoError(t, downMock.ExpectationsWereMet())

// test different table structure

maxConnectionsRow = sqlmock.NewRows([]string{"Variable_name", "Value"}).
AddRow("max_connections", "2")
mock.ExpectQuery("SHOW VARIABLES LIKE 'max_connections'").WillReturnRows(maxConnectionsRow)
sqlModeRow = sqlmock.NewRows([]string{"Variable_name", "Value"}).
AddRow("sql_mode", "ANSI_QUOTES")
mock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(sqlModeRow)
createTableRow = sqlmock.NewRows([]string{"Table", "Create Table"}).
AddRow("test-table-1", `CREATE TABLE "test-table-1" (
"c" int(11) NOT NULL,
"c2" int(11) NOT NULL,
PRIMARY KEY ("c"),
UNIQUE KEY "uk" (("c2"+1), "c")
) ENGINE=InnoDB`)
mock.ExpectQuery("SHOW CREATE TABLE `test-db`.`test-table-1`").WillReturnRows(createTableRow)
sqlModeRow2 = sqlmock.NewRows([]string{"Variable_name", "Value"}).
AddRow("sql_mode", "ANSI_QUOTES")
downMock.ExpectQuery("SHOW VARIABLES LIKE 'sql_mode'").WillReturnRows(sqlModeRow2)
createTableRow2 = sqlmock.NewRows([]string{"Table", "Create Table"}).
AddRow("test-table-1", `CREATE TABLE "test-table-1" (
"c" int(11) NOT NULL,
"c2" int(11) NOT NULL,
PRIMARY KEY ("c"),
UNIQUE KEY "uk" (("c2"+3), "c")
) ENGINE=InnoDB`)
downMock.ExpectQuery("SHOW CREATE TABLE `test-db`.`test-table-1`").WillReturnRows(createTableRow2)

checker = NewTablesChecker(
map[string]*conn.BaseDB{"test-source": conn.NewBaseDBForTest(db)},
conn.NewBaseDBForTest(downDB),
map[string]map[filter.Table][]filter.Table{
"test-source": {
{Schema: "test-db", Name: "test-table-1"}: {
{Schema: "test-db", Name: "test-table-1"},
},
},
},
nil,
1)
result = checker.Check(ctx)
require.Equal(t, StateWarning, result.State)
require.Len(t, result.Errors, 2)
// maybe [`c2`+1 c] or [c `c2`+1]
require.Contains(t, result.Errors[0].ShortErr, "upstream has more PK or NOT NULL UK than downstream")
require.Contains(t, result.Errors[0].ShortErr, "`c2`+1")
require.Contains(t, result.Errors[1].ShortErr, "downstream has more PK or NOT NULL UK than upstream")
require.Contains(t, result.Errors[1].ShortErr, "`c2`+3")
require.NoError(t, mock.ExpectationsWereMet())
require.NoError(t, downMock.ExpectationsWereMet())
}
17 changes: 16 additions & 1 deletion dm/pkg/checker/utils.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,10 @@ import (
"github.com/pingcap/tidb-tools/pkg/utils"
"github.com/pingcap/tidb/parser"
"github.com/pingcap/tidb/parser/ast"
"github.com/pingcap/tidb/parser/format"
"github.com/pingcap/tiflow/dm/pkg/log"
"github.com/pingcap/tiflow/dm/pkg/terror"
"go.uber.org/zap"
)

// MySQLVersion represents MySQL version number.
Expand Down Expand Up @@ -201,6 +204,8 @@ func getCollation(stmt *ast.CreateTableStmt) string {
// getPKAndUK returns a map of INDEX_NAME -> set of COLUMN_NAMEs.
func getPKAndUK(stmt *ast.CreateTableStmt) map[string]map[string]struct{} {
ret := make(map[string]map[string]struct{})
var sb strings.Builder
restoreCtx := format.NewRestoreCtx(format.DefaultRestoreFlags, &sb)

for _, constraint := range stmt.Constraints {
switch constraint.Tp {
Expand All @@ -212,7 +217,17 @@ func getPKAndUK(stmt *ast.CreateTableStmt) map[string]map[string]struct{} {
case ast.ConstraintUniq, ast.ConstraintUniqKey, ast.ConstraintUniqIndex:
ret[constraint.Name] = make(map[string]struct{})
for _, key := range constraint.Keys {
ret[constraint.Name][key.Column.Name.L] = struct{}{}
if key.Column != nil {
ret[constraint.Name][key.Column.Name.L] = struct{}{}
} else {
sb.Reset()
err := key.Expr.Restore(restoreCtx)
if err != nil {
log.L().Warn("failed to restore expression", zap.Error(err))
continue
}
ret[constraint.Name][sb.String()] = struct{}{}
}
}
}
}
Expand Down

0 comments on commit f8bcbba

Please sign in to comment.