Merge branch 'master' into regexSupportCompress

br/pkg/lightning/checkpoints/checkpoints.go

@@ -517,7 +517,15 @@ func OpenCheckpointsDB(ctx context.Context, cfg *config.Config) (DB, error) {
 
 	switch cfg.Checkpoint.Driver {
 	case config.CheckpointDriverMySQL:
-		db, err := common.ConnectMySQL(cfg.Checkpoint.DSN)
+		var (
+			db  *sql.DB
+			err error
+		)
+		if cfg.Checkpoint.MySQLParam != nil {
+			db, err = cfg.Checkpoint.MySQLParam.Connect()
+		} else {
+			db, err = sql.Open("mysql", cfg.Checkpoint.DSN)
+		}
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
@@ -546,7 +554,15 @@ func IsCheckpointsDBExists(ctx context.Context, cfg *config.Config) (bool, error
 	}
 	switch cfg.Checkpoint.Driver {
 	case config.CheckpointDriverMySQL:
-		db, err := sql.Open("mysql", cfg.Checkpoint.DSN)
+		var (
+			db  *sql.DB
+			err error
+		)
+		if cfg.Checkpoint.MySQLParam != nil {
+			db, err = cfg.Checkpoint.MySQLParam.Connect()
+		} else {
+			db, err = sql.Open("mysql", cfg.Checkpoint.DSN)
+		}
 		if err != nil {
 			return false, errors.Trace(err)
 		}

br/pkg/lightning/common/util.go

@@ -23,7 +23,6 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"net/url"
 	"os"
 	"strconv"
 	"strings"
@@ -58,28 +57,38 @@ type MySQLConnectParam struct {
 	Vars             map[string]string
 }
 
-func (param *MySQLConnectParam) ToDSN() string {
-	hostPort := net.JoinHostPort(param.Host, strconv.Itoa(param.Port))
-	dsn := fmt.Sprintf("%s:%s@tcp(%s)/?charset=utf8mb4&sql_mode='%s'&maxAllowedPacket=%d&tls=%s",
-		param.User, param.Password, hostPort,
-		param.SQLMode, param.MaxAllowedPacket, param.TLS)
+func (param *MySQLConnectParam) ToDriverConfig() *mysql.Config {
+	cfg := mysql.NewConfig()
+	cfg.Params = make(map[string]string)
+
+	cfg.User = param.User
+	cfg.Passwd = param.Password
+	cfg.Net = "tcp"
+	cfg.Addr = net.JoinHostPort(param.Host, strconv.Itoa(param.Port))
+	cfg.Params["charset"] = "utf8mb4"
+	cfg.Params["sql_mode"] = fmt.Sprintf("'%s'", param.SQLMode)
+	cfg.MaxAllowedPacket = int(param.MaxAllowedPacket)
+	cfg.TLSConfig = param.TLS
 
 	for k, v := range param.Vars {
-		dsn += fmt.Sprintf("&%s='%s'", k, url.QueryEscape(v))
+		cfg.Params[k] = fmt.Sprintf("'%s'", v)
 	}
-
-	return dsn
+	return cfg
 }
 
-func tryConnectMySQL(dsn string) (*sql.DB, error) {
-	driverName := "mysql"
-	failpoint.Inject("MockMySQLDriver", func(val failpoint.Value) {
-		driverName = val.(string)
+func tryConnectMySQL(cfg *mysql.Config) (*sql.DB, error) {
+	failpoint.Inject("MustMySQLPassword", func(val failpoint.Value) {
+		pwd := val.(string)
+		if cfg.Passwd != pwd {
+			failpoint.Return(nil, &mysql.MySQLError{Number: tmysql.ErrAccessDenied, Message: "access denied"})
+		}
+		failpoint.Return(nil, nil)
 	})
-	db, err := sql.Open(driverName, dsn)
+	c, err := mysql.NewConnector(cfg)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	db := sql.OpenDB(c)
 	if err = db.Ping(); err != nil {
 		_ = db.Close()
 		return nil, errors.Trace(err)
@@ -89,13 +98,9 @@ func tryConnectMySQL(dsn string) (*sql.DB, error) {
 
 // ConnectMySQL connects MySQL with the dsn. If access is denied and the password is a valid base64 encoding,
 // we will try to connect MySQL with the base64 decoding of the password.
-func ConnectMySQL(dsn string) (*sql.DB, error) {
-	cfg, err := mysql.ParseDSN(dsn)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
+func ConnectMySQL(cfg *mysql.Config) (*sql.DB, error) {
 	// Try plain password first.
-	db, firstErr := tryConnectMySQL(dsn)
+	db, firstErr := tryConnectMySQL(cfg)
 	if firstErr == nil {
 		return db, nil
 	}
@@ -104,9 +109,9 @@ func ConnectMySQL(dsn string) (*sql.DB, error) {
 		// If password is encoded by base64, try the decoded string as well.
 		if password, decodeErr := base64.StdEncoding.DecodeString(cfg.Passwd); decodeErr == nil && string(password) != cfg.Passwd {
 			cfg.Passwd = string(password)
-			db, err = tryConnectMySQL(cfg.FormatDSN())
+			db2, err := tryConnectMySQL(cfg)
 			if err == nil {
-				return db, nil
+				return db2, nil
 			}
 		}
 	}
@@ -115,7 +120,7 @@ func ConnectMySQL(dsn string) (*sql.DB, error) {
 }
 
 func (param *MySQLConnectParam) Connect() (*sql.DB, error) {
-	db, err := ConnectMySQL(param.ToDSN())
+	db, err := ConnectMySQL(param.ToDriverConfig())
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

br/pkg/lightning/common/util_test.go

@@ -16,16 +16,12 @@ package common_test
 
 import (
 	"context"
-	"database/sql"
-	"database/sql/driver"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
-	"math/rand"
 	"net/http"
 	"net/http/httptest"
-	"strconv"
 	"testing"
 	"time"
 
@@ -35,7 +31,6 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/br/pkg/lightning/common"
 	"github.com/pingcap/tidb/br/pkg/lightning/log"
-	tmysql "github.com/pingcap/tidb/errno"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -85,66 +80,14 @@ func TestGetJSON(t *testing.T) {
 	require.Regexp(t, ".*http status code != 200.*", err.Error())
 }
 
-func TestToDSN(t *testing.T) {
-	param := common.MySQLConnectParam{
-		Host:             "127.0.0.1",
-		Port:             4000,
-		User:             "root",
-		Password:         "123456",
-		SQLMode:          "strict",
-		MaxAllowedPacket: 1234,
-		TLS:              "cluster",
-		Vars: map[string]string{
-			"tidb_distsql_scan_concurrency": "1",
-		},
-	}
-	require.Equal(t, "root:123456@tcp(127.0.0.1:4000)/?charset=utf8mb4&sql_mode='strict'&maxAllowedPacket=1234&tls=cluster&tidb_distsql_scan_concurrency='1'", param.ToDSN())
-
-	param.Host = "::1"
-	require.Equal(t, "root:123456@tcp([::1]:4000)/?charset=utf8mb4&sql_mode='strict'&maxAllowedPacket=1234&tls=cluster&tidb_distsql_scan_concurrency='1'", param.ToDSN())
-}
-
-type mockDriver struct {
-	driver.Driver
-	plainPsw string
-}
-
-func (m *mockDriver) Open(dsn string) (driver.Conn, error) {
-	cfg, err := mysql.ParseDSN(dsn)
-	if err != nil {
-		return nil, err
-	}
-	accessDenied := cfg.Passwd != m.plainPsw
-	return &mockConn{accessDenied: accessDenied}, nil
-}
-
-type mockConn struct {
-	driver.Conn
-	driver.Pinger
-	accessDenied bool
-}
-
-func (c *mockConn) Ping(ctx context.Context) error {
-	if c.accessDenied {
-		return &mysql.MySQLError{Number: tmysql.ErrAccessDenied, Message: "access denied"}
-	}
-	return nil
-}
-
-func (c *mockConn) Close() error {
-	return nil
-}
-
 func TestConnect(t *testing.T) {
 	plainPsw := "dQAUoDiyb1ucWZk7"
-	driverName := "mysql-mock-" + strconv.Itoa(rand.Int())
-	sql.Register(driverName, &mockDriver{plainPsw: plainPsw})
 
 	require.NoError(t, failpoint.Enable(
-		"github.com/pingcap/tidb/br/pkg/lightning/common/MockMySQLDriver",
-		fmt.Sprintf("return(\"%s\")", driverName)))
+		"github.com/pingcap/tidb/br/pkg/lightning/common/MustMySQLPassword",
+		fmt.Sprintf("return(\"%s\")", plainPsw)))
 	defer func() {
-		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/common/MockMySQLDriver"))
+		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/common/MustMySQLPassword"))
 	}()
 
 	param := common.MySQLConnectParam{
@@ -155,13 +98,11 @@ func TestConnect(t *testing.T) {
 		SQLMode:          "strict",
 		MaxAllowedPacket: 1234,
 	}
-	db, err := param.Connect()
+	_, err := param.Connect()
 	require.NoError(t, err)
-	require.NoError(t, db.Close())
 	param.Password = base64.StdEncoding.EncodeToString([]byte(plainPsw))
-	db, err = param.Connect()
+	_, err = param.Connect()
 	require.NoError(t, err)
-	require.NoError(t, db.Close())
 }
 
 func TestIsContextCanceledError(t *testing.T) {

br/pkg/lightning/config/config.go

@@ -553,11 +553,12 @@ type TikvImporter struct {
 }
 
 type Checkpoint struct {
-	Schema           string                 `toml:"schema" json:"schema"`
-	DSN              string                 `toml:"dsn" json:"-"` // DSN may contain password, don't expose this to JSON.
-	Driver           string                 `toml:"driver" json:"driver"`
-	Enable           bool                   `toml:"enable" json:"enable"`
-	KeepAfterSuccess CheckpointKeepStrategy `toml:"keep-after-success" json:"keep-after-success"`
+	Schema           string                    `toml:"schema" json:"schema"`
+	DSN              string                    `toml:"dsn" json:"-"` // DSN may contain password, don't expose this to JSON.
+	MySQLParam       *common.MySQLConnectParam `toml:"-" json:"-"`   // For some security reason, we use MySQLParam instead of DSN.
+	Driver           string                    `toml:"driver" json:"driver"`
+	Enable           bool                      `toml:"enable" json:"enable"`
+	KeepAfterSuccess CheckpointKeepStrategy    `toml:"keep-after-success" json:"keep-after-success"`
 }
 
 type Cron struct {
@@ -1142,7 +1143,7 @@ func (cfg *Config) AdjustCheckPoint() {
 				MaxAllowedPacket: defaultMaxAllowedPacket,
 				TLS:              cfg.TiDB.TLS,
 			}
-			cfg.Checkpoint.DSN = param.ToDSN()
+			cfg.Checkpoint.MySQLParam = &param
 		case CheckpointDriverFile:
 			cfg.Checkpoint.DSN = "/tmp/" + cfg.Checkpoint.Schema + ".pb"
 		}

br/pkg/lightning/config/config_test.go

@@ -32,7 +32,6 @@ import (
 	"github.com/BurntSushi/toml"
 	"github.com/pingcap/tidb/br/pkg/lightning/common"
 	"github.com/pingcap/tidb/br/pkg/lightning/config"
-	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/stretchr/testify/require"
 )
 
@@ -626,7 +625,9 @@ func TestLoadConfig(t *testing.T) {
 	taskCfg.TiDB.DistSQLScanConcurrency = 1
 	err = taskCfg.Adjust(context.Background())
 	require.NoError(t, err)
-	require.Equal(t, "guest:12345@tcp(172.16.30.11:4001)/?charset=utf8mb4&sql_mode='"+mysql.DefaultSQLMode+"'&maxAllowedPacket=67108864&tls=false", taskCfg.Checkpoint.DSN)
+	equivalentDSN := taskCfg.Checkpoint.MySQLParam.ToDriverConfig().FormatDSN()
+	expectedDSN := "guest:12345@tcp(172.16.30.11:4001)/?tls=false&maxAllowedPacket=67108864&charset=utf8mb4&sql_mode=%27ONLY_FULL_GROUP_BY%2CSTRICT_TRANS_TABLES%2CNO_ZERO_IN_DATE%2CNO_ZERO_DATE%2CERROR_FOR_DIVISION_BY_ZERO%2CNO_AUTO_CREATE_USER%2CNO_ENGINE_SUBSTITUTION%27"
+	require.Equal(t, expectedDSN, equivalentDSN)
 
 	result := taskCfg.String()
 	require.Regexp(t, `.*"pd-addr":"172.16.30.11:2379,172.16.30.12:2379".*`, result)

cmd/explaintest/r/imdbload.result

@@ -286,7 +286,7 @@ IndexLookUp_7	1005030.94	root
 └─TableRowIDScan_6(Probe)	1005030.94	cop[tikv]	table:char_name	keep order:false
 trace plan target = 'estimation' select * from char_name where ((imdb_index = 'I') and (surname_pcode < 'E436')) or ((imdb_index = 'L') and (surname_pcode < 'E436'));
 CE_trace
-[{"table_name":"char_name","type":"Column Stats-Point","expr":"((imdb_index = 'I'))","row_count":0},{"table_name":"char_name","type":"Column Stats-Point","expr":"((imdb_index = 'L'))","row_count":0},{"table_name":"char_name","type":"Column Stats-Range","expr":"((id >= -9223372036854775808 and id <= 9223372036854775807))","row_count":4314864},{"table_name":"char_name","type":"Index Stats-Range","expr":"((imdb_index = 'I') and (surname_pcode < 'E436')) or ((imdb_index = 'L') and (surname_pcode < 'E436'))","row_count":0},{"table_name":"char_name","type":"Index Stats-Range","expr":"((surname_pcode < 'E436'))","row_count":1005030},{"table_name":"char_name","type":"Table Stats-Expression-CNF","expr":"`or`(`and`(`eq`(imdbload.char_name.imdb_index, 'I'), `lt`(imdbload.char_name.surname_pcode, 'E436')), `and`(`eq`(imdbload.char_name.imdb_index, 'L'), `lt`(imdbload.char_name.surname_pcode, 'E436')))","row_count":804024}]
+[{"table_name":"char_name","type":"Column Stats-Point","expr":"((imdb_index = 'I'))","row_count":0},{"table_name":"char_name","type":"Column Stats-Point","expr":"((imdb_index = 'L'))","row_count":0},{"table_name":"char_name","type":"Column Stats-Range","expr":"((id >= -9223372036854775808 and id <= 9223372036854775807))","row_count":4314864},{"table_name":"char_name","type":"Column Stats-Range","expr":"((surname_pcode < 'E436'))","row_count":1005030},{"table_name":"char_name","type":"Index Stats-Range","expr":"((imdb_index = 'I') and (surname_pcode < 'E436')) or ((imdb_index = 'L') and (surname_pcode < 'E436'))","row_count":0},{"table_name":"char_name","type":"Index Stats-Range","expr":"((surname_pcode < 'E436'))","row_count":1005030},{"table_name":"char_name","type":"Table Stats-Expression-CNF","expr":"`or`(`and`(`eq`(imdbload.char_name.imdb_index, 'I'), `lt`(imdbload.char_name.surname_pcode, 'E436')), `and`(`eq`(imdbload.char_name.imdb_index, 'L'), `lt`(imdbload.char_name.surname_pcode, 'E436')))","row_count":804024}]
 
 explain select * from char_name where ((imdb_index = 'V') and (surname_pcode < 'L3416'));
 id	estRows	task	access object	operator info
@@ -356,7 +356,7 @@ IndexLookUp_11	901.00	root
   └─TableRowIDScan_9	901.00	cop[tikv]	table:keyword	keep order:false
 trace plan target = 'estimation' select * from keyword where ((phonetic_code = 'R1652') and (keyword > 'ecg-monitor' and keyword < 'killers'));
 CE_trace
-[{"table_name":"keyword","type":"Column Stats-Point","expr":"((phonetic_code = 'R1652'))","row_count":23480},{"table_name":"keyword","type":"Column Stats-Range","expr":"((id >= -9223372036854775808 and id <= 9223372036854775807))","row_count":236627},{"table_name":"keyword","type":"Column Stats-Range","expr":"((keyword > 'ecg-monitor' and keyword < 'killers'))","row_count":44075},{"table_name":"keyword","type":"Index Stats-Point","expr":"((phonetic_code = 'R1652'))","row_count":23480},{"table_name":"keyword","type":"Index Stats-Range","expr":"((keyword > 'ecg-monitor' and keyword < 'killers'))","row_count":44036},{"table_name":"keyword","type":"Index Stats-Range","expr":"((keyword >= 'ecg-m' and keyword <= 'kille'))","row_count":44036},{"table_name":"keyword","type":"Index Stats-Range","expr":"((phonetic_code = 'R1652') and (keyword > 'ecg-monitor' and keyword < 'killers'))","row_count":901},{"table_name":"keyword","type":"Table Stats-Expression-CNF","expr":"`and`(`eq`(imdbload.keyword.phonetic_code, 'R1652'), `and`(`gt`(imdbload.keyword.keyword, 'ecg-monitor'), `lt`(imdbload.keyword.keyword, 'killers')))","row_count":901}]
+[{"table_name":"keyword","type":"Column Stats-Point","expr":"((phonetic_code = 'R1652'))","row_count":23480},{"table_name":"keyword","type":"Column Stats-Range","expr":"((id >= -9223372036854775808 and id <= 9223372036854775807))","row_count":236627},{"table_name":"keyword","type":"Column Stats-Range","expr":"((keyword > 'ecg-monitor' and keyword < 'killers'))","row_count":44075},{"table_name":"keyword","type":"Index Stats-Point","expr":"((phonetic_code = 'R1652'))","row_count":23480},{"table_name":"keyword","type":"Index Stats-Range","expr":"((keyword >= 'ecg-m' and keyword <= 'kille'))","row_count":44036},{"table_name":"keyword","type":"Index Stats-Range","expr":"((phonetic_code = 'R1652') and (keyword > 'ecg-monitor' and keyword < 'killers'))","row_count":901},{"table_name":"keyword","type":"Table Stats-Expression-CNF","expr":"`and`(`eq`(imdbload.keyword.phonetic_code, 'R1652'), `and`(`gt`(imdbload.keyword.keyword, 'ecg-monitor'), `lt`(imdbload.keyword.keyword, 'killers')))","row_count":901}]
 
 explain select * from cast_info where (nr_order is null) and (person_role_id = 2) and (note >= '(key set pa: Florida');
 id	estRows	task	access object	operator info

cmd/importer/db.go

@@ -22,7 +22,7 @@ import (
 	"strconv"
 	"strings"
 
-	_ "github.com/go-sql-driver/mysql"
+	mysql2 "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/parser/mysql"
@@ -318,13 +318,18 @@ func execSQL(db *sql.DB, sql string) error {
 }
 
 func createDB(cfg DBConfig) (*sql.DB, error) {
-	dbDSN := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Name)
-	db, err := sql.Open("mysql", dbDSN)
+	driverCfg := mysql2.NewConfig()
+	driverCfg.User = cfg.User
+	driverCfg.Passwd = cfg.Password
+	driverCfg.Net = "tcp"
+	driverCfg.Addr = cfg.Host + ":" + strconv.Itoa(cfg.Port)
+	driverCfg.DBName = cfg.Name
+
+	c, err := mysql2.NewConnector(driverCfg)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-
-	return db, nil
+	return sql.OpenDB(c), nil
 }
 
 func closeDB(db *sql.DB) error {

dumpling/export/config.go

@@ -218,6 +218,31 @@ func (conf *Config) GetDSN(db string) string {
 	return dsn
 }
 
+// GetDriverConfig returns the MySQL driver config from Config.
+func (conf *Config) GetDriverConfig(db string) *mysql.Config {
+	driverCfg := mysql.NewConfig()
+	// maxAllowedPacket=0 can be used to automatically fetch the max_allowed_packet variable from server on every connection.
+	// https://github.com/go-sql-driver/mysql#maxallowedpacket
+	hostPort := net.JoinHostPort(conf.Host, strconv.Itoa(conf.Port))
+	driverCfg.User = conf.User
+	driverCfg.Passwd = conf.Password
+	driverCfg.Net = "tcp"
+	driverCfg.Addr = hostPort
+	driverCfg.DBName = db
+	driverCfg.Collation = "utf8mb4_general_ci"
+	driverCfg.ReadTimeout = conf.ReadTimeout
+	driverCfg.WriteTimeout = 30 * time.Second
+	driverCfg.InterpolateParams = true
+	driverCfg.MaxAllowedPacket = 0
+	if conf.Security.DriveTLSName != "" {
+		driverCfg.TLSConfig = conf.Security.DriveTLSName
+	}
+	if conf.AllowCleartextPasswords {
+		driverCfg.AllowCleartextPasswords = true
+	}
+	return driverCfg
+}
+
 func timestampDirName() string {
 	return fmt.Sprintf("./export-%s", time.Now().Format(time.RFC3339))
 }