sink(ticdc): adjust sql mode compatibility for mysql sink

cdc/sink/mysql.go

@@ -29,6 +29,7 @@ import (
  "github.com/pingcap/log"
  "github.com/pingcap/ticdc/cdc/model"
  "github.com/pingcap/ticdc/cdc/sink/common"
+ dmutils "github.com/pingcap/ticdc/dm/pkg/utils"
  "github.com/pingcap/ticdc/pkg/config"
  "github.com/pingcap/ticdc/pkg/cyclic"
  "github.com/pingcap/ticdc/pkg/cyclic/mark"
@@ -166,6 +167,10 @@ func newMySQLSink(
  cancel: cancel,
  }
 
+ if err = sink.adjustSQLModeCompatible(ctx); err != nil {
+ return nil, errors.Trace(err)
+ }
+
  if val, ok := opts[mark.OptCyclicConfig]; ok {
  cfg := new(config.CyclicConfig)
  err := cfg.Unmarshal([]byte(val))
@@ -174,7 +179,7 @@ func newMySQLSink(
  }
  sink.cyclic = cyclic.NewCyclic(cfg)
 
- err = sink.adjustSQLMode(ctx)
+ err = sink.adjustSQLModeCyclic(ctx)
  if err != nil {
  return nil, errors.Trace(err)
  }
@@ -348,8 +353,26 @@ func needSwitchDB(ddl *model.DDLEvent) bool {
  return true
 }
 
+func (s *mysqlSink) adjustSQLModeCompatible(ctx context.Context) error {
+ var oldMode, newMode string
+ row := s.db.QueryRowContext(ctx, "SELECT @@SESSION.sql_mode;")
+ err := row.Scan(&oldMode)
+ if err != nil {
+ return cerror.WrapError(cerror.ErrMySQLQueryError, err)
+ }
+ newMode, err = dmutils.AdjustSQLModeCompatible(oldMode)
+ if err != nil {
+ return cerror.WrapError(cerror.ErrMySQLQueryError, err)
+ }
+ _, err = s.db.ExecContext(ctx, fmt.Sprintf("SET sql_mode = '%s';", newMode))
+ if err != nil {
+ return cerror.WrapError(cerror.ErrMySQLQueryError, err)
+ }
+ return nil
+}
+
 // adjustSQLMode adjust sql mode according to sink config.
-func (s *mysqlSink) adjustSQLMode(ctx context.Context) error {
+func (s *mysqlSink) adjustSQLModeCyclic(ctx context.Context) error {
  // Must relax sql mode to support cyclic replication, as downstream may have
  // extra columns (not null and no default value).
  if s.cyclic == nil || !s.cyclic.Enabled() {

cdc/sink/mysql_test.go

@@ -444,6 +444,20 @@ func mockTestDB() (*sql.DB, error) {
  return db, nil
 }
 
+func mockDBWithAdjustedSQLMode() (*sql.DB, sqlmock.Sqlmock, error) {
+ db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ if err != nil {
+ return db, mock, err
+ }
+ // sql mode is adjust for compatibility.
+ mock.ExpectQuery("SELECT @@SESSION.sql_mode;").
+ WillReturnRows(sqlmock.NewRows([]string{"@@SESSION.sql_mode"}).
+ AddRow("STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE"))
+ mock.ExpectExec("SET sql_mode = 'ALLOW_INVALID_DATES,IGNORE_SPACE,NO_AUTO_VALUE_ON_ZERO';").
+ WillReturnResult(sqlmock.NewResult(0, 0))
+ return db, mock, err
+}
+
 func TestAdjustSQLMode(t *testing.T) {
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()
@@ -460,12 +474,14 @@ func TestAdjustSQLMode(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
  require.Nil(t, err)
+
+ // sql mode is adjust for cyclic.
  mock.ExpectQuery("SELECT @@SESSION.sql_mode;").
  WillReturnRows(sqlmock.NewRows([]string{"@@SESSION.sql_mode"}).
- AddRow("ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE"))
- mock.ExpectExec("SET sql_mode = 'ONLY_FULL_GROUP_BY,NO_ZERO_IN_DATE,NO_ZERO_DATE';").
+ AddRow("ALLOW_INVALID_DATES,IGNORE_SPACE,NO_AUTO_VALUE_ON_ZERO"))
+ mock.ExpectExec("SET sql_mode = 'ALLOW_INVALID_DATES,IGNORE_SPACE,NO_AUTO_VALUE_ON_ZERO';").
  WillReturnResult(sqlmock.NewResult(0, 0))
  mock.ExpectClose()
  return db, nil
@@ -577,7 +593,7 @@ func TestNewMySQLSinkExecDML(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
  require.Nil(t, err)
  mock.ExpectBegin()
  mock.ExpectExec("REPLACE INTO `s1`.`t1`(`a`,`b`) VALUES (?,?),(?,?)").
@@ -721,7 +737,7 @@ func TestExecDMLRollbackErrDatabaseNotExists(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
  require.Nil(t, err)
  mock.ExpectBegin()
  mock.ExpectExec("REPLACE INTO `s1`.`t1`(`a`) VALUES (?),(?)").
@@ -787,7 +803,7 @@ func TestExecDMLRollbackErrTableNotExists(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
  require.Nil(t, err)
  mock.ExpectBegin()
  mock.ExpectExec("REPLACE INTO `s1`.`t1`(`a`) VALUES (?),(?)").
@@ -853,7 +869,7 @@ func TestExecDMLRollbackErrRetryable(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
  require.Nil(t, err)
  for i := 0; i < defaultDMLMaxRetryTime; i++ {
  mock.ExpectBegin()
@@ -902,7 +918,7 @@ func TestNewMySQLSinkExecDDL(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
  require.Nil(t, err)
  mock.ExpectBegin()
  mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(1, 1))
@@ -1032,7 +1048,8 @@ func TestNewMySQLSink(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
+ require.Nil(t, err)
  mock.ExpectClose()
  require.Nil(t, err)
  return db, nil
@@ -1071,7 +1088,8 @@ func TestMySQLSinkClose(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
+ require.Nil(t, err)
  mock.ExpectClose()
  require.Nil(t, err)
  return db, nil
@@ -1111,7 +1129,8 @@ func TestMySQLSinkFlushResovledTs(t *testing.T) {
  return db, nil
  }
  // normal db
- db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ db, mock, err := mockDBWithAdjustedSQLMode()
+ require.Nil(t, err)
  mock.ExpectBegin()
  mock.ExpectExec("REPLACE INTO `s1`.`t1`(`a`) VALUES (?)").
  WithArgs(1).

dm/pkg/utils/db.go

@@ -19,6 +19,7 @@ import (
  "fmt"
  "math"
  "math/rand"
+ "sort"
  "strconv"
  "strings"
  "time"
@@ -618,6 +619,7 @@ func GetSQLModeStrBySQLMode(sqlMode tmysql.SQLMode) string {
  sqlModeStr = append(sqlModeStr, str)
  }
  }
+ sort.Strings(sqlModeStr)
  return strings.Join(sqlModeStr, ",")
 }