dump: retrieve ANSI_QUOTES from upstream (pingcap#909)

dm/config/task.go

@@ -592,8 +592,6 @@ func (c *TaskConfig) SubTaskConfigs(sources map[string]DBConfig) ([]*SubTaskConf
 
 		cfg.CleanDumpFile = c.CleanDumpFile
 
-		cfg.EnableANSIQuotes = c.EnableANSIQuotes
-
 		err := cfg.Adjust(true)
 		if err != nil {
 			return nil, terror.Annotatef(err, "source %s", inst.SourceID)

dm/worker/worker.go

@@ -402,12 +402,15 @@ func (w *Worker) observeSubtaskStage(ctx context.Context, etcdCli *clientv3.Clie
 }
 
 func (w *Worker) handleSubTaskStage(ctx context.Context, stageCh chan ha.Stage, errCh chan error) error {
+	closed := false
 	for {
 		select {
 		case <-ctx.Done():
-			log.L().Info("worker is closed, handleSubTaskStage will quit now")
-			return nil
-		case stage := <-stageCh:
+			closed = true
+		case stage, ok := <-stageCh:
+			if !ok {
+				closed = true
+			}
 			opType, err := w.operateSubTaskStageWithoutConfig(stage)
 			if err != nil {
 				opErrCounter.WithLabelValues(w.name, opType).Inc()
@@ -416,13 +419,20 @@ func (w *Worker) handleSubTaskStage(ctx context.Context, stageCh chan ha.Stage,
 					return err
 				}
 			}
-		case err := <-errCh:
+		case err, ok := <-errCh:
+			if !ok {
+				closed = true
+			}
 			// TODO: deal with err
 			log.L().Error("WatchSubTaskStage received an error", zap.Error(err))
 			if etcdutil.IsRetryableError(err) {
 				return err
 			}
 		}
+		if closed {
+			log.L().Info("worker is closed, handleSubTaskStage will quit now")
+			return nil
+		}
 	}
 }
 

dumpling/dumpling.go

@@ -15,6 +15,7 @@ package dumpling
 
 import (
 	"context"
+	"database/sql"
 	"os"
 	"strings"
 	"time"
@@ -57,6 +58,7 @@ func NewDumpling(cfg *config.SubTaskConfig) *Dumpling {
 func (m *Dumpling) Init(ctx context.Context) error {
 	var err error
 	m.dumpConfig, err = m.constructArgs()
+	m.detectAnsiQuotes()
 	return err
 }
 
@@ -266,3 +268,22 @@ func (m *Dumpling) constructArgs() (*export.Config, error) {
 
 	return dumpConfig, nil
 }
+
+// detectAnsiQuotes tries to detect ANSI_QUOTES from upstream. If success, change EnableANSIQuotes in subtask config
+func (m *Dumpling) detectAnsiQuotes() {
+	db, err := sql.Open("mysql", m.dumpConfig.GetDSN(""))
+	if err != nil {
+		return
+	}
+	defer db.Close()
+	enable, err := utils.HasAnsiQuotesMode(db)
+	if err != nil {
+		return
+	}
+	if enable != m.cfg.EnableANSIQuotes {
+		m.logger.Warn("found mismatched ANSI_QUOTES setting, going to overwrite it to DB specified",
+			zap.Bool("DB specified", enable),
+			zap.Bool("config file specified", m.cfg.EnableANSIQuotes))
+	}
+	m.cfg.EnableANSIQuotes = enable
+}

tests/drop_column_with_index/conf/dm-task.yaml

@@ -13,11 +13,6 @@ target-database:
   port: 4000
   user: "root"
   password: ""
-  session:
-    sql_mode: "NO_AUTO_VALUE_ON_ZERO,ANSI_QUOTES"
-    tidb_skip_utf8_check: 1
-    tidb_disable_txn_auto_retry: off
-    tidb_retry_limit: "10"
 
 mysql-instances:
   - source-id: "mysql-replica-01"

tests/drop_column_with_index/run.sh

@@ -35,7 +35,7 @@ function run() {
     run_sql_file $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
 
     # use sync_diff_inspector to check data now!
-    check_sync_diff $WORK_DIR $cur/conf/diff_config.toml 30
+    check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
 
     # check column covered by multi-column indices won't drop, and its indices won't drop
     run_sql "alter table drop_column_with_index.t1 drop column c2;" $MYSQL_PORT1 $MYSQL_PASSWORD1