config: support setting session variable

dm/config/subtask.go

@@ -81,11 +81,12 @@ func (c *RawDBConfig) SetMaxIdleConns(value int) *RawDBConfig {
 
 // DBConfig is the DB configuration.
 type DBConfig struct {
-	Host             string `toml:"host" json:"host" yaml:"host"`
-	Port             int    `toml:"port" json:"port" yaml:"port"`
-	User             string `toml:"user" json:"user" yaml:"user"`
-	Password         string `toml:"password" json:"-" yaml:"password"` // omit it for privacy
-	MaxAllowedPacket *int   `toml:"max-allowed-packet" json:"max-allowed-packet" yaml:"max-allowed-packet"`
+	Host             string            `toml:"host" json:"host" yaml:"host"`
+	Port             int               `toml:"port" json:"port" yaml:"port"`
+	User             string            `toml:"user" json:"user" yaml:"user"`
+	Password         string            `toml:"password" json:"-" yaml:"password"` // omit it for privacy
+	MaxAllowedPacket *int              `toml:"max-allowed-packet" json:"max-allowed-packet" yaml:"max-allowed-packet"`
+	Session          map[string]string `toml:"session" json:"session" yaml:"session"`
 
 	RawDBCfg *RawDBConfig `toml:"-" json:"-" yaml:"-"`
 }

pkg/conn/basedb.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"net/url"
 	"sync"
 
 	"github.com/pingcap/dm/dm/config"
@@ -56,10 +57,19 @@ func (d *DefaultDBProviderImpl) Apply(config config.DBConfig) (*BaseDB, error) {
 		}
 		maxIdleConns = rawCfg.MaxIdleConns
 	}
+
+	for key, val := range config.Session {
+		// for num such as 1/"1", format as key='1'
+		// for string, format as key='string'
+		// both are valid for mysql and tidb
+		dsn += fmt.Sprintf("&%s='%s'", key, url.QueryEscape(val))
+	}
+
 	db, err := sql.Open("mysql", dsn)
 	if err != nil {
 		return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
 	}
+
 	db.SetMaxIdleConns(maxIdleConns)
 
 	return NewBaseDB(db), nil

tests/_utils/check_sync_diff

@@ -2,6 +2,7 @@
 # parameter 1: work directory
 # parameter 2: config file for sync_diff_inspector
 # parameter 3: max check times
+# parameter 4: check diff should fail or not
 
 workdir=$1
 conf=$2
@@ -23,6 +24,9 @@ do
     ret=$?
     if [ "$ret" == 0 ]; then
         echo "check diff successfully"
+        if [[ $4 = "fail" ]]; then
+            exit 1
+        fi
         break
     fi
     ((i++))
@@ -32,8 +36,10 @@ done
 
 if [ $i -ge $check_time ]; then
     echo "check diff failed at last"
-    # show \n and other blanks
-    printf "$(cat $LOG)\n"
-    exit 1
+    if [[ $4 != "fail" ]]; then
+        # show \n and other blanks
+        printf "$(cat $LOG)\n"
+        exit 1
+    fi
 fi
 cd $PWD

tests/all_mode/conf/dm-task.yaml

@@ -13,6 +13,11 @@ target-database:
   port: 4000
   user: "root"
   password: ""
+  session:
+    sql_mode: "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+    tidb_skip_utf8_check: 1
+    tidb_disable_txn_auto_retry: off
+    tidb_retry_limit: "10"
 
 mysql-instances:
   - source-id: "mysql-replica-01"

tests/all_mode/run.sh

@@ -7,7 +7,61 @@ source $cur/../_utils/test_prepare
 WORK_DIR=$TEST_DIR/$TEST_NAME
 API_VERSION="v1alpha1"
 
+function test_session_config(){
+    run_sql_file $cur/data/db1.prepare.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+    check_contains 'Query OK, 2 rows affected'
+    run_sql_file $cur/data/db2.prepare.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+    check_contains 'Query OK, 3 rows affected'
+
+    # start DM worker and master
+    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.toml $WORK_DIR/source1.toml
+    cp $cur/conf/source2.toml $WORK_DIR/source2.toml
+    sed -i "/relay-binlog-name/i\relay-dir = \"$WORK_DIR/worker1/relay_log\"" $WORK_DIR/source1.toml
+    sed -i "/relay-binlog-name/i\relay-dir = \"$WORK_DIR/worker2/relay_log\"" $WORK_DIR/source2.toml
+    dmctl_operate_source create $WORK_DIR/source1.toml $SOURCE_ID1
+    dmctl_operate_source create $WORK_DIR/source2.toml $SOURCE_ID2
+
+    cp $cur/conf/dm-task.yaml $WORK_DIR/dm-task.yaml
+
+    # error config
+    sed -i 's/tidb_retry_limit: "10"/tidb_retry_limit: "fjs"/g'  $WORK_DIR/dm-task.yaml
+    run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "start-task $WORK_DIR/dm-task.yaml --remove-meta" \
+        "'tidb_retry_limit' can't be set to the value" 1
+
+    # sql_mode: "ANSI_QUOTES"
+    sed -i 's/tidb_retry_limit: "fjs"/tidb_retry_limit: "10"/g'  $WORK_DIR/dm-task.yaml
+    sed -i 's/sql_mode: ".*"/sql_mode: "ANSI_QUOTES"/g'  $WORK_DIR/dm-task.yaml
+    dmctl_start_task "$WORK_DIR/dm-task.yaml" "--remove-meta"
+
+    # dumpling works fine since it use single quote
+    check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+
+    run_sql_file $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+    run_sql_file $cur/data/db2.increment.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+
+    # but syncer should fail
+    echo "check diff should fail"
+    check_sync_diff $WORK_DIR $cur/conf/diff_config.toml 10 "fail"
+
+    run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+            "stop-task test"\
+            "\"result\": true" 3
+
+    cleanup_data all_mode
+    cleanup_process $*
+}
+
 function run() {
+    test_session_config
+
     export GO_FAILPOINTS="github.com/pingcap/dm/dm/worker/TaskCheckInterval=return(\"500ms\")"
 
     run_sql_file $cur/data/db1.prepare.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1