loader, syncer: Update sql_mode to more compatible

loader/loader.go

@@ -529,7 +529,44 @@ func (l *Loader) Init(ctx context.Context) (err error) {
 		}
 	}
 	if !hasSQLMode {
-		lcfg.To.Session["sql_mode"] = l.cfg.LoaderConfig.SQLMode
+		// When upstream's datatime is 2020-00-00, 2020-00-01, 2020-06-00
+		// and so on, downstream will be 2019-11-30, 2019-12-01, 2020-05-31,
+		// as if set the 'NO_ZERO_IN_DATE', 'NO_ZERO_DATE'.
+		// This is because the implementation of go-mysql, that you can see
+		// https://github.com/go-mysql-org/go-mysql/blob/master/replication/row_event.go#L1063-L1087
+		sqlModes := strings.Split(l.cfg.LoaderConfig.SQLMode, ",")
+		needDisable := []string{
+			"NO_ZERO_IN_DATE",
+			"NO_ZERO_DATE",
+			"ERROR_FOR_DIVISION_BY_ZERO",
+			"NO_AUTO_CREATE_USER",
+			"STRICT_TRANS_TABLES",
+			"STRICT_ALL_TABLES",
+		}
+		for _, disable := range needDisable {
+			for i, sqlMode := range sqlModes {
+				if disable == sqlMode {
+					sqlModes = append(sqlModes[:i], sqlModes[i+1:]...)
+					break
+				}
+			}
+		}
+		needEnable := []string{
+			"IGNORE_SPACE",
+			"NO_AUTO_VALUE_ON_ZERO",
+			"ALLOW_INVALID_DATES",
+		}
+		waitEnable := sqlModes
+		for _, enable := range needEnable {
+			for _, sqlMode := range sqlModes {
+				if enable != sqlMode {
+					waitEnable = append(waitEnable, enable)
+					break
+				}
+			}
+		}
+		lcfg.To.Session["sql_mode"] = strings.Join(waitEnable, ",")
+		l.logger.Info("loader's sql_mode is ", zap.String("sqlmode", lcfg.To.Session["sql_mode"]))
 	}
 
 	l.toDB, l.toDBConns, err = createConns(tctx, lcfg, l.cfg.PoolSize)

syncer/syncer.go

@@ -2742,8 +2742,43 @@ func (s *Syncer) createDBs(ctx context.Context) error {
 		if err2 != nil {
 			s.tctx.L().Warn("cannot get sql_mode from upstream database", log.ShortError(err2))
 		} else {
-			s.cfg.To.Session["sql_mode"] = sqlMode
-		}
+			// When upstream's datatime is 2020-00-00, 2020-00-01, 2020-06-00
+			// and so on, downstream will be 2019-11-30, 2019-12-01, 2020-05-31,
+			// as if set the 'NO_ZERO_IN_DATE', 'NO_ZERO_DATE'.
+			// This is because the implementation of go-mysql, that you can see
+			// https://github.com/go-mysql-org/go-mysql/blob/master/replication/row_event.go#L1063-L1087
+			sqlModes := strings.Split(sqlMode, ",")
+			needDisable := []string{
+				"NO_ZERO_IN_DATE",
+				"NO_ZERO_DATE",
+				"ERROR_FOR_DIVISION_BY_ZERO",
+				"NO_AUTO_CREATE_USER",
+				"STRICT_TRANS_TABLES",
+				"STRICT_ALL_TABLES",
+			}
+			for _, disable := range needDisable {
+				for i, sqlMode := range sqlModes {
+					if disable == sqlMode {
+						sqlModes = append(sqlModes[:i], sqlModes[i+1:]...)
+						break
+					}
+				}
+			}
+			needEnable := []string{
+				"IGNORE_SPACE",
+				"NO_AUTO_VALUE_ON_ZERO",
+				"ALLOW_INVALID_DATES",
+			}
+			waitEnable := sqlModes
+			for _, enable := range needEnable {
+				for _, sqlMode := range sqlModes {
+					if enable != sqlMode {
+						waitEnable = append(waitEnable, enable)
+						break
+					}
+				}
+			}
+			s.cfg.To.Session["sql_mode"] = strings.Join(waitEnable, ",")
 	}
 
 	dbCfg = s.cfg.To

tests/others_integration.txt

@@ -8,4 +8,5 @@ only_dml
 adjust_gtid
 load_task
 downstream_more_column
-expression_filter
+expression_filter
+sql_mode

tests/sql_mode/conf/diff_config.toml

@@ -0,0 +1,56 @@
+# diff Configuration.
+
+log-level = "info"
+
+chunk-size = 20
+
+check-thread-count = 4
+
+sample-percent = 100
+
+use-checksum = true
+
+fix-sql-file = "fix.sql"
+
+# tables need to check.
+[[check-tables]]
+schema = "sql_mode"
+tables = ["~.*"]
+
+[[source-db]]
+host = "127.0.0.1"
+port = 3306
+user = "root"
+password = "123456"
+instance-id = "source-1"
+
+[[table-config]]
+schema = "sql_mode"
+table = "t_1"
+
+[[table-config.source-tables]]
+instance-id = "source-1"
+schema = "sql_mode"
+table  = "t_1"
+
+[[source-db]]
+host = "127.0.0.1"
+port = 3307
+user = "root"
+password = "123456"
+instance-id = "source-2"
+
+[[table-config]]
+schema = "sql_mode"
+table = "t_2"
+
+[[table-config.source-tables]]
+instance-id = "source-2"
+schema = "sql_mode"
+table  = "t_2"
+
+[target-db]
+host = "127.0.0.1"
+port = 4000
+user = "test"
+password = "123456"

tests/sql_mode/conf/dm-master.toml

@@ -0,0 +1,4 @@
+# Master Configuration.
+master-addr = ":8261"
+advertise-addr = "127.0.0.1:8261"
+auto-compaction-retention = "3s"

tests/sql_mode/conf/dm-task.yaml

@@ -0,0 +1,46 @@
+---
+name: sql_mode
+task-mode: all
+is-sharding: false
+meta-schema: "dm_meta"
+enable-heartbeat: false
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+    block-allow-list:  "instance"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+
+  - source-id: "mysql-replica-02"
+    block-allow-list:  "instance"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+
+block-allow-list:
+  instance:
+    do-dbs: ["sql_mode"]
+
+mydumpers:
+  global:
+    threads: 4
+    chunk-filesize: 0
+    skip-tz-utc: true
+    extra-args: "--statement-size=4000"
+
+loaders:
+  global:
+    pool-size: 16
+    dir: "./dumped_data"
+
+syncers:
+  global:
+    worker-count: 16
+    batch: 100

tests/sql_mode/conf/dm-worker1.toml

@@ -0,0 +1,2 @@
+name = "worker1"
+join = "127.0.0.1:8261"

tests/sql_mode/conf/dm-worker2.toml

@@ -0,0 +1,2 @@
+name = "worker2"
+join = "127.0.0.1:8261"

tests/sql_mode/conf/source1.yaml

@@ -0,0 +1,11 @@
+source-id: mysql-replica-01
+flavor: ''
+enable-gtid: false
+enable-relay: true
+relay-binlog-name: ''
+relay-binlog-gtid: ''
+from:
+  host: 127.0.0.1
+  user: root
+  password: /Q7B9DizNLLTTfiZHv9WoEAKamfpIUs=
+  port: 3306

tests/sql_mode/conf/source2.yaml

@@ -0,0 +1,11 @@
+source-id: mysql-replica-02
+flavor: ''
+enable-gtid: false
+enable-relay: true
+relay-binlog-name: ''
+relay-binlog-gtid: ''
+from:
+  host: 127.0.0.1
+  user: root
+  password: /Q7B9DizNLLTTfiZHv9WoEAKamfpIUs=
+  port: 3307

tests/sql_mode/data/db1.increment.sql

@@ -0,0 +1,32 @@
+set @@session.SQL_MODE='';
+use sql_mode;
+
+-- test sql_mode PIPES_AS_CONCAT
+insert into t_1(num) values('pipes'||'as'||'concat');
+
+-- test sql_mode ANSI_QUOTES
+insert into t_1(name) values("a");
+
+-- test sql_mode IGNORE_SPACE
+create table count (id int not null, primary key(id));
+
+-- test sql_mode NO_AUTO_VALUE_ON_ZERO
+insert into t_1(id, name) values (10, 'a');
+insert into t_1(id, name) values (0, 'b');
+insert into t_1(id, name) values (0, 'c');
+
+-- test sql_mode NO_BACKSLASH_ESCAPES
+insert into t_1(name) values ('\\a');
+
+-- test sql_mode STRICT_TRANS_TABLES && STRICT_ALL_TABLES && NO_ZERO_IN_DATE && NO_ZERO_DATE && ALLOW_INVALID_DATES
+insert into t_1(dt) values('0000-06-00');
+insert into t_1(dt) values('0000-00-01');
+insert into t_1(dt) values('0000-06-01');
+insert into t_1(dt) values('0000-00-00');
+
+-- test sql_mode ERROR_FOR_DIVISION_BY_ZERO
+insert into t_1(num) values(4/0);
+
+-- test sql_mode NO_AUTO_CREATE_USER
+drop user if exists 'no_auto_create_user';
+grant select on *.* to 'no_auto_create_user';

tests/sql_mode/data/db1.prepare.sql

@@ -0,0 +1,11 @@
+drop database if exists `sql_mode`;
+create database `sql_mode`;
+use `sql_mode`;
+CREATE TABLE `t_1` (
+  `id` int NOT NULL AUTO_INCREMENT,
+  `name` varchar(60),
+  `num` int,
+  `dt` datetime,
+  -- `ts` timestamp,
+  PRIMARY KEY (id)
+);

tests/sql_mode/data/db2.increment.sql

@@ -0,0 +1,19 @@
+set @@session.SQL_MODE='PIPES_AS_CONCAT,IGNORE_SPACE,ONLY_FULL_GROUP_BY,NO_UNSIGNED_SUBTRACTION,NO_DIR_IN_CREATE,NO_AUTO_VALUE_ON_ZERO,NO_BACKSLASH_ESCAPES,STRICT_TRANS_TABLES,STRICT_ALL_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ALLOW_INVALID_DATES,ERROR_FOR_DIVISION_BY_ZERO,HIGH_NOT_PRECEDENCE,NO_ENGINE_SUBSTITUTION,REAL_AS_FLOAT';
+-- NO_AUTO_CREATE_USER set failed in mysql8.0
+use sql_mode;
+
+-- test sql_mode PIPES_AS_CONCAT
+insert into t_2(name) values('pipes'||'as'||'concat');
+
+-- test sql_mode ANSI_QUOTES
+insert into t_2(name) values("a");
+
+-- test sql_mode IGNORE_SPACE
+select count (*) as c from t_2;
+
+-- test sql_mode NO_AUTO_VALUE_ON_ZERO
+insert into t_2(id, name) values (10, 'a');
+insert into t_2(id, name) values (0, 'b');
+
+-- test sql_mode NO_BACKSLASH_ESCAPES
+insert into t_2(name) values ('\\a');

tests/sql_mode/data/db2.prepare.sql

@@ -0,0 +1,10 @@
+drop database if exists `sql_mode`;
+create database `sql_mode`;
+use `sql_mode`;
+CREATE TABLE `t_2` (
+  `id` int NOT NULL AUTO_INCREMENT,
+  `name` varchar(60),
+  `num` int,
+  `dt` datetime,
+  PRIMARY KEY (id)
+);

tests/sql_mode/run.sh

@@ -0,0 +1,50 @@
+#!/bin/bash
+
+set -eu
+
+cur=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $cur/../_utils/test_prepare
+WORK_DIR=$TEST_DIR/$TEST_NAME
+
+function run() {
+	run_sql_source1 "SET @@GLOBAL.SQL_MODE='PIPES_AS_CONCAT,IGNORE_SPACE,ONLY_FULL_GROUP_BY,NO_UNSIGNED_SUBTRACTION,NO_DIR_IN_CREATE,NO_AUTO_VALUE_ON_ZERO,NO_BACKSLASH_ESCAPES,STRICT_TRANS_TABLES,STRICT_ALL_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ALLOW_INVALID_DATES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,HIGH_NOT_PRECEDENCE,NO_ENGINE_SUBSTITUTION,REAL_AS_FLOAT'"
+	# run_sql_source1 "SET @@GLOBAL.SQL_MODE='ANSI_QUOTES'"
+	run_sql_source2 "SET @@GLOBAL.SQL_MODE=''"
+
+	run_sql_file $cur/data/db1.prepare.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/db2.prepare.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+
+	run_dm_master $WORK_DIR/master $MASTER_PORT $cur/conf/dm-master.toml
+	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT
+	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $cur/conf/dm-worker1.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER1_PORT
+	run_dm_worker $WORK_DIR/worker2 $WORKER2_PORT $cur/conf/dm-worker2.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER2_PORT
+
+	# operate mysql config to worker
+	cp $cur/conf/source1.yaml $WORK_DIR/source1.yaml
+	sed -i "/relay-binlog-name/i\relay-dir: $WORK_DIR/worker1/relay_log" $WORK_DIR/source1.yaml
+	dmctl_operate_source create $WORK_DIR/source1.yaml $SOURCE_ID1
+	cp $cur/conf/source2.yaml $WORK_DIR/source2.yaml
+	sed -i "/relay-binlog-name/i\relay-dir: $WORK_DIR/worker2/relay_log" $WORK_DIR/source2.yaml
+	dmctl_operate_source create $WORK_DIR/source2.yaml $SOURCE_ID2
+
+	# start DM task only
+	dmctl_start_task $cur/conf/dm-task.yaml
+
+	# use sync_diff_inspector to check full dump loader
+	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+
+	run_sql_file $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+	run_sql_file $cur/data/db2.increment.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+}
+
+cleanup_data $TEST_NAME
+# also cleanup dm processes in case of last run failed
+cleanup_process $*
+run $*
+cleanup_process $*
+
+echo "[$(date)] <<<<<< test case $TEST_NAME success! >>>>>>"