Repository: pingcap/dumpling
Pull request #215 (2020-12-07 12:24:47)

consistency: refine consistency lock logic
<!--
Thank you for contributing to Dumpling! Please read the [CONTRIBUTING](https://github.com/pingcap/dumpling/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.

If you want to open the **Challenge Program** pull request, please use the following template:
https://raw.githubusercontent.com/pingcap/.github/master/.github/PULL_REQUEST_TEMPLATE/challenge-program.md

You can use it with query parameters: https://github.com/pingcap/dumpling/compare/master...${you branch}?template=challenge-program.md
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
close https://github.com/pingcap/dumpling/issues/210
We lock tables one by one now so that we can't assure the snapshot of tables.

### What is changed and how it works?
Lock all tables in one SQL and record metadata info after all tables are locked for consistency lock.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

Related changes

 - Need to cherry-pick to the release branch
 
### Release note

<!-- bugfixes or new feature need a release note, must in the form of a list, such as

- support -T/--tables-list argument

or if no need to be included in the release note, just add the following line

- No release note
-->
- Make consistency lock support snapshot


Patch:
diff --git a/v4/export/consistency.go b/v4/export/consistency.go
index d3504bc..bfcde5a 100644
--- a/v4/export/consistency.go
+++ b/v4/export/consistency.go
@@ -6,6 +6,7 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/pingcap/br/pkg/utils"
 	"github.com/pingcap/errors"
 )
 
@@ -113,15 +114,12 @@ type ConsistencyLockDumpingTables struct {
 
 // Setup implements ConsistencyController.Setup
 func (c *ConsistencyLockDumpingTables) Setup(ctx context.Context) error {
-	for dbName, tables := range c.allTables {
-		for _, table := range tables {
-			err := LockTables(ctx, c.conn, dbName, table.Name)
-			if err != nil {
-				return err
-			}
-		}
-	}
-	return nil
+	blockList := make(map[string]map[string]interface{})
+	return utils.WithRetry(ctx, func() error {
+		lockTablesSQL := buildLockTablesSQL(c.allTables, blockList)
+		_, err := c.conn.ExecContext(ctx, lockTablesSQL)
+		return errors.Trace(err)
+	}, newLockTablesBackoffer(blockList))
 }
 
 // TearDown implements ConsistencyController.TearDown
diff --git a/v4/export/dump.go b/v4/export/dump.go
index 8ce03e7..e789131 100755
--- a/v4/export/dump.go
+++ b/v4/export/dump.go
@@ -84,18 +84,12 @@ func (d *Dumper) Dump() (dumpErr error) {
 		}
 	}()
 
-	// for consistency lock, we should lock tables at first to get the tables we want to lock & dump
-	// for consistency lock, record meta pos before lock tables because other tables may still be modified while locking tables
+	// for consistency lock, we should get table list at first to generate the lock tables SQL
 	if conf.Consistency == consistencyTypeLock {
 		conn, err = createConnWithConsistency(ctx, pool)
 		if err != nil {
 			return errors.Trace(err)
 		}
-		m.recordStartTime(time.Now())
-		err = m.recordGlobalMetaData(conn, conf.ServerInfo.ServerType, false)
-		if err != nil {
-			log.Info("get global metadata failed", zap.Error(err))
-		}
 		if err = prepareTableListToDump(conf, conn); err != nil {
 			conn.Close()
 			return err
@@ -118,41 +112,29 @@ func (d *Dumper) Dump() (dumpErr error) {
 		}
 	}()
 
-	// for other consistencies, we should get table list after consistency is set up and GlobalMetaData is cached
-	// for other consistencies, record snapshot after whole tables are locked. The recorded meta info is exactly the locked snapshot.
-	if conf.Consistency != consistencyTypeLock {
-		conn, err = pool.Conn(ctx)
-		if err != nil {
-			return errors.Trace(err)
-		}
-		m.recordStartTime(time.Now())
-		err = m.recordGlobalMetaData(conn, conf.ServerInfo.ServerType, false)
-		if err != nil {
-			log.Info("get global metadata failed", zap.Error(err))
-		}
-		conn.Close()
-	}
-
 	metaConn, err := createConnWithConsistency(ctx, pool)
 	if err != nil {
 		return err
 	}
-
-	if conf.PosAfterConnect {
-		// record again, to provide a location to exit safe mode for DM
-		err = m.recordGlobalMetaData(metaConn, conf.ServerInfo.ServerType, true)
-		if err != nil {
-			log.Info("get global metadata (after connection pool established) failed", zap.Error(err))
-		}
+	m.recordStartTime(time.Now())
+	// for consistency lock, we can write snapshot info after all tables are locked.
+	// the binlog pos may changed because there is still possible write between we lock tables and write master status.
+	// but for the locked tables doing replication that starts from metadata is safe.
+	// for consistency flush, record snapshot after whole tables are locked. The recorded meta info is exactly the locked snapshot.
+	// for consistency snapshot, we should use the snapshot that we get/set at first in metadata. TiDB will assure the snapshot of TSO.
+	// for consistency none, the binlog pos in metadata might be earlier than dumped data. We need to enable safe-mode to assure data safety.
+	err = m.recordGlobalMetaData(metaConn, conf.ServerInfo.ServerType, false)
+	if err != nil {
+		log.Info("get global metadata failed", zap.Error(err))
 	}
 
+	// for other consistencies, we should get table list after consistency is set up and GlobalMetaData is cached
 	if conf.Consistency != consistencyTypeLock {
 		if err = prepareTableListToDump(conf, metaConn); err != nil {
 			return err
 		}
 	}
 
-	failpoint.Inject("ConsistencyCheck", nil)
 	rebuildConn := func(conn *sql.Conn) (*sql.Conn, error) {
 		// make sure that the lock connection is still alive
 		err1 := conCtrl.PingContext(ctx)
@@ -184,6 +166,7 @@ func (d *Dumper) Dump() (dumpErr error) {
 		return err
 	}
 	defer tearDownWriters()
+
 	if conf.TransactionalConsistency {
 		if conf.Consistency == consistencyTypeFlush || conf.Consistency == consistencyTypeLock {
 			log.Info("All the dumping transactions have started. Start to unlock tables")
@@ -192,6 +175,16 @@ func (d *Dumper) Dump() (dumpErr error) {
 			return errors.Trace(err)
 		}
 	}
+	// Inject consistency failpoint test after we release the table lock
+	failpoint.Inject("ConsistencyCheck", nil)
+
+	if conf.PosAfterConnect {
+		// record again, to provide a location to exit safe mode for DM
+		err = m.recordGlobalMetaData(metaConn, conf.ServerInfo.ServerType, true)
+		if err != nil {
+			log.Info("get global metadata (after connection pool established) failed", zap.Error(err))
+		}
+	}
 
 	summary.SetLogCollector(summary.NewLogCollector(log.Info))
 	summary.SetUnit(summary.BackupUnit)
@@ -513,19 +506,19 @@ func buildTiDBTableSampleQuery(pkFields []string, dbName, tblName string) string
 	return fmt.Sprintf(template, pks, escapeString(dbName), escapeString(tblName), pks)
 }
 
-func prepareTableListToDump(conf *Config, pool *sql.Conn) error {
-	databases, err := prepareDumpingDatabases(conf, pool)
+func prepareTableListToDump(conf *Config, db *sql.Conn) error {
+	databases, err := prepareDumpingDatabases(conf, db)
 	if err != nil {
 		return err
 	}
 
-	conf.Tables, err = listAllTables(pool, databases)
+	conf.Tables, err = listAllTables(db, databases)
 	if err != nil {
 		return err
 	}
 
 	if !conf.NoViews {
-		views, err := listAllViews(pool, databases)
+		views, err := listAllViews(db, databases)
 		if err != nil {
 			return err
 		}
diff --git a/v4/export/retry.go b/v4/export/retry.go
index f4a75ff..089d572 100644
--- a/v4/export/retry.go
+++ b/v4/export/retry.go
@@ -3,17 +3,24 @@
 package export
 
 import (
+	"strings"
 	"time"
 
 	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
+	"go.uber.org/zap"
+
+	"github.com/pingcap/dumpling/v4/log"
 )
 
 const (
 	dumpChunkRetryTime       = 3
+	lockTablesRetryTime      = 5
 	dumpChunkWaitInterval    = 50 * time.Millisecond
 	dumpChunkMaxWaitInterval = 200 * time.Millisecond
+	// ErrNoSuchTable is the error code no such table in MySQL/TiDB
+	ErrNoSuchTable uint16 = 1146
 )
 
 func newDumpChunkBackoffer(shouldRetry bool) *dumpChunkBackoffer { // revive:disable-line:flag-parameter
@@ -56,3 +63,53 @@ func (b *dumpChunkBackoffer) NextBackoff(err error) time.Duration {
 func (b *dumpChunkBackoffer) Attempt() int {
 	return b.attempt
 }
+
+func newLockTablesBackoffer(blockList map[string]map[string]interface{}) *lockTablesBackoffer {
+	return &lockTablesBackoffer{
+		attempt:   lockTablesRetryTime,
+		blockList: blockList,
+	}
+}
+
+type lockTablesBackoffer struct {
+	attempt   int
+	blockList map[string]map[string]interface{}
+}
+
+func (b *lockTablesBackoffer) NextBackoff(err error) time.Duration {
+	err = errors.Cause(err)
+	if mysqlErr, ok := err.(*mysql.MySQLError); ok && mysqlErr.Number == ErrNoSuchTable {
+		b.attempt--
+		db, table, err := getTableFromMySQLError(mysqlErr.Message)
+		if err != nil {
+			log.Error("retry lock tables meet error", zap.Error(err))
+			b.attempt = 0
+			return 0
+		}
+		if _, ok := b.blockList[db]; !ok {
+			b.blockList[db] = make(map[string]interface{})
+		}
+		b.blockList[db][table] = struct{}{}
+		return 0
+	}
+	b.attempt = 0
+	return 0
+}
+
+func (b *lockTablesBackoffer) Attempt() int {
+	return b.attempt
+}
+
+func getTableFromMySQLError(msg string) (db, table string, err error) {
+	// examples of the error msg:
+	// Error 1146: Table 'pingcap.t1' doesn't exist
+	// Error 1146: Table 'quo`te/database.quo.te/database-1.quo.te/table-1' doesn't exist /* doesn't support */
+	msg = strings.TrimPrefix(msg, "Table '")
+	msg = strings.TrimSuffix(msg, "' doesn't exist")
+	failPart := strings.Split(msg, ".")
+	if len(failPart) != 2 {
+		err = errors.Errorf("doesn't support retry lock table %s", msg)
+		return
+	}
+	return failPart[0], failPart[1], nil
+}
diff --git a/v4/export/sql.go b/v4/export/sql.go
index ca5a235..d2f7d1f 100644
--- a/v4/export/sql.go
+++ b/v4/export/sql.go
@@ -3,6 +3,7 @@
 package export
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"fmt"
@@ -635,6 +636,35 @@ func buildOrderByClauseString(handleColNames []string) string {
 	return fmt.Sprintf("ORDER BY %s", strings.Join(quotaCols, separator))
 }
 
+func buildLockTablesSQL(allTables DatabaseTables, blockList map[string]map[string]interface{}) string {
+	// ,``.`` READ has 11 bytes, "LOCK TABLE" has 10 bytes
+	estimatedCap := len(allTables)*11 + 10
+	s := bytes.NewBuffer(make([]byte, 0, estimatedCap))
+	n := false
+	for dbName, tables := range allTables {
+		escapedDBName := escapeString(dbName)
+		for _, table := range tables {
+			// Lock views will lock related tables. However, we won't dump data only the create sql of view, so we needn't lock view here.
+			// Besides, mydumper also only lock base table here. https://github.com/maxbube/mydumper/blob/1fabdf87e3007e5934227b504ad673ba3697946c/mydumper.c#L1568
+			if table.Type != TableTypeBase {
+				continue
+			}
+			if blockTable, ok := blockList[dbName]; ok {
+				if _, ok := blockTable[table.Name]; ok {
+					continue
+				}
+			}
+			if !n {
+				fmt.Fprintf(s, "LOCK TABLES `%s`.`%s` READ", escapedDBName, escapeString(table.Name))
+				n = true
+			} else {
+				fmt.Fprintf(s, ",`%s`.`%s` READ", escapedDBName, escapeString(table.Name))
+			}
+		}
+	}
+	return s.String()
+}
+
 type oneStrColumnTable struct {
 	data []string
 }


Test patch:
diff --git a/v4/export/consistency_test.go b/v4/export/consistency_test.go
index 5ad1b5a..8c135bf 100644
--- a/v4/export/consistency_test.go
+++ b/v4/export/consistency_test.go
@@ -8,6 +8,7 @@ import (
 	"strings"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
 )
 
@@ -17,7 +18,7 @@ type testConsistencySuite struct{}
 
 func (s *testConsistencySuite) assertNil(err error, c *C) {
 	if err != nil {
-		c.Fatalf(err.Error())
+		c.Fatal(err.Error())
 	}
 }
 
@@ -49,7 +50,7 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	c.Assert(ok, IsTrue)
 	s.assertLifetimeErrNil(ctx, ctrl, c)
 	if err = mock.ExpectationsWereMet(); err != nil {
-		c.Fatalf(err.Error())
+		c.Fatal(err.Error())
 	}
 
 	conf.Consistency = consistencyTypeSnapshot
@@ -63,16 +64,40 @@ func (s *testConsistencySuite) TestConsistencyController(c *C) {
 	conf.Tables = NewDatabaseTables().
 		AppendTables("db1", "t1", "t2", "t3").
 		AppendViews("db2", "t4")
-	for i := 0; i < 4; i++ {
-		mock.ExpectExec("LOCK TABLES").WillReturnResult(resultOk)
-	}
+	mock.ExpectExec("LOCK TABLES `db1`.`t1` READ,`db1`.`t2` READ,`db1`.`t3` READ").WillReturnResult(resultOk)
 	mock.ExpectExec("UNLOCK TABLES").WillReturnResult(resultOk)
 	ctrl, _ = NewConsistencyController(ctx, conf, db)
 	_, ok = ctrl.(*ConsistencyLockDumpingTables)
 	c.Assert(ok, IsTrue)
 	s.assertLifetimeErrNil(ctx, ctrl, c)
 	if err = mock.ExpectationsWereMet(); err != nil {
-		c.Fatalf(err.Error())
+		c.Fatal(err.Error())
+	}
+}
+
+func (s *testConsistencySuite) TestConsistencyLockControllerRetry(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	conf := DefaultConfig()
+	resultOk := sqlmock.NewResult(0, 1)
+
+	conf.Consistency = consistencyTypeLock
+	conf.Tables = NewDatabaseTables().
+		AppendTables("db1", "t1", "t2", "t3").
+		AppendViews("db2", "t4")
+	mock.ExpectExec("LOCK TABLES `db1`.`t1` READ,`db1`.`t2` READ,`db1`.`t3` READ").
+		WillReturnError(&mysql.MySQLError{Number: ErrNoSuchTable, Message: "Table 'db1.t3' doesn't exist"})
+	mock.ExpectExec("LOCK TABLES `db1`.`t1` READ,`db1`.`t2` READ").WillReturnResult(resultOk)
+	mock.ExpectExec("UNLOCK TABLES").WillReturnResult(resultOk)
+	ctrl, _ := NewConsistencyController(ctx, conf, db)
+	_, ok := ctrl.(*ConsistencyLockDumpingTables)
+	c.Assert(ok, IsTrue)
+	s.assertLifetimeErrNil(ctx, ctrl, c)
+	if err = mock.ExpectationsWereMet(); err != nil {
+		c.Fatal(err.Error())
 	}
 }