Repository: pingcap/tidb-lightning
Pull request #243 (2019-09-26 06:09:31)

backend/tidb: use REPLACE INTO or INSERT IGNORE INTO to provide idempotent insertion
<!--
Thank you for contributing to TiDB-Lightning! Please read the [CONTRIBUTING](https://github.com/pingcap/tidb-lightning/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
1 modifications from #242 : 
- use `REPLACE INTO` to provide idempotent insertion while PK or UNIQUE exsits

### What is changed and how it works?
replace according SQL to `REPLACE INTO` or `INSERT IGNORE INTO` statement in backend/tidb's WriteRows. 
add a config field to decide behaviour

```
[tikv-importer]
backend = "tidb"
on-duplicate = "replace" # or "ignore", "error". which represents REPLACE INTO, INSERT IGNORE INTO, INSERT INTO, respectively.
```

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

Side effects

 - Possible performance regression: not sure `REPLACE INTO` is slower

Related changes


lance6716: /run-all-tests

PTAL @kennytm @3pointer 
lance6716: /run-all-tests


kennytm: Could we make it configurable to choose whether to use `INSERT INTO` / `REPLACE INTO` or `INSERT IGNORE INTO`?
lance6716: Did a simple test using `mysql < text.sql`, `insert ignore into` is slightly faster(50% duplicated), so how about switch to `insert ignore into`?

ignore is reasonable since dumped files won't change.
kennytm: This would mean that, during incremental import, if the new data does conflict with old data the default behavior is the new data being discarded, which is not very intuitive.

I think the default should still be REPLACE but we could configure to INSERT or IGNORE if needed
lance6716: Oh I forget we should keep `INSERT INTO` as an option
GregoryIan: LGTM
GregoryIan: @kennytm PTAL
lance6716: > @kennytm PTAL

Kenny says we might provide 3 types of behavior: replace, insert ignore, insert. I'm agree that we might provide all 3 types, and if a further benchmark shows there's little performance difference, we might reduce to 2 or 1 type at that time.
GregoryIan: would lightning compute checksum many times for the rows which are executed more than twice? Rest LGTM
kennytm: @GregoryIan Lightning currently won't do checksum with TiDB backend (since the ADMIN CHECKSUM is based on KV pairs).
kennytm: @lance6716 I'll start review after you've changed the config from a bool to a string option.
lance6716: /run-all-tests

PTAL @kennytm
lance6716: /run-all-tests
lance6716: /run-all-tests

PTAL @kennytm @GregoryIan 

Patch:
diff --git a/lightning/backend/tidb.go b/lightning/backend/tidb.go
index 1ab858c..40f9181 100644
--- a/lightning/backend/tidb.go
+++ b/lightning/backend/tidb.go
@@ -21,6 +21,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
@@ -28,6 +29,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tidb-lightning/lightning/common"
+	"github.com/pingcap/tidb-lightning/lightning/config"
 	"github.com/pingcap/tidb-lightning/lightning/log"
 	"github.com/pingcap/tidb-lightning/lightning/verification"
 )
@@ -41,15 +43,22 @@ type tidbEncoder struct {
 }
 
 type tidbBackend struct {
-	db *sql.DB
+	db          *sql.DB
+	onDuplicate string
 }
 
 // NewTiDBBackend creates a new TiDB backend using the given database.
 //
 // The backend does not take ownership of `db`. Caller should close `db`
 // manually after the backend expired.
-func NewTiDBBackend(db *sql.DB) Backend {
-	return MakeBackend(&tidbBackend{db: db})
+func NewTiDBBackend(db *sql.DB, onDuplicate string) Backend {
+	switch onDuplicate {
+	case config.ReplaceOnDup, config.IgnoreOnDup, config.ErrorOnDup:
+	default:
+		log.L().Warn("unsupported action on duplicate, overwrite with `replace`")
+		onDuplicate = config.ReplaceOnDup
+	}
+	return MakeBackend(&tidbBackend{db: db, onDuplicate: onDuplicate})
 }
 
 func (row tidbRow) ClassifyAndAppend(data *Rows, checksum *verification.KVChecksum, _ *Rows, _ *verification.KVChecksum) {
@@ -229,6 +238,9 @@ func (be *tidbBackend) RetryImportDelay() time.Duration {
 }
 
 func (be *tidbBackend) MaxChunkSize() int {
+	failpoint.Inject("FailIfImportedSomeRows", func() {
+		failpoint.Return(1)
+	})
 	return 1048576
 }
 
@@ -263,7 +275,15 @@ func (be *tidbBackend) WriteRows(ctx context.Context, _ uuid.UUID, tableName str
 	}
 
 	var insertStmt strings.Builder
-	insertStmt.WriteString("INSERT INTO ")
+	switch be.onDuplicate {
+	case config.ReplaceOnDup:
+		insertStmt.WriteString("REPLACE INTO ")
+	case config.IgnoreOnDup:
+		insertStmt.WriteString("INSERT IGNORE INTO ")
+	case config.ErrorOnDup:
+		insertStmt.WriteString("INSERT INTO ")
+	}
+
 	insertStmt.WriteString(tableName)
 	if len(columnNames) > 0 {
 		insertStmt.WriteByte('(')
@@ -289,5 +309,8 @@ func (be *tidbBackend) WriteRows(ctx context.Context, _ uuid.UUID, tableName str
 
 	// Retry will be done externally, so we're not going to retry here.
 	_, err := be.db.ExecContext(ctx, insertStmt.String())
+	failpoint.Inject("FailIfImportedSomeRows", func() {
+		panic("forcing failure due to FailIfImportedSomeRows, before saving checkpoint")
+	})
 	return err
 }
diff --git a/lightning/config/config.go b/lightning/config/config.go
index 0a782ef..47eb2cd 100644
--- a/lightning/config/config.go
+++ b/lightning/config/config.go
@@ -47,6 +47,13 @@ const (
 	CheckpointDriverMySQL = "mysql"
 	// CheckpointDriverFile is a constant for choosing the "File" checkpoint driver in the configuration.
 	CheckpointDriverFile = "file"
+
+	// ReplaceOnDup indicates using REPLACE INTO to insert data
+	ReplaceOnDup = "replace"
+	// IgnoreOnDup indicates using INSERT IGNORE INTO to insert data
+	IgnoreOnDup = "ignore"
+	// ErrorOnDup indicates using INSERT INTO to insert data, which would violate PK or UNIQUE constraint
+	ErrorOnDup = "error"
 )
 
 var defaultConfigPaths = []string{"tidb-lightning.toml", "conf/tidb-lightning.toml"}
@@ -129,8 +136,9 @@ type MydumperRuntime struct {
 }
 
 type TikvImporter struct {
-	Addr    string `toml:"addr" json:"addr"`
-	Backend string `toml:"backend" json:"backend"`
+	Addr         string `toml:"addr" json:"addr"`
+	Backend      string `toml:"backend" json:"backend"`
+	OnDuplicate  string `toml:"on-duplicate" json:"on-duplicate"`
 }
 
 type Checkpoint struct {
@@ -194,6 +202,7 @@ func NewConfig() *Config {
 		},
 		TikvImporter: TikvImporter{
 			Backend: BackendImporter,
+			OnDuplicate: ReplaceOnDup,
 		},
 		PostRestore: PostRestore{
 			Checksum: true,
@@ -326,6 +335,15 @@ func (cfg *Config) Adjust() error {
 		return errors.Errorf("invalid config: unsupported `tikv-importer.backend` (%s)", cfg.TikvImporter.Backend)
 	}
 
+	if cfg.TikvImporter.Backend == BackendTiDB {
+		cfg.TikvImporter.OnDuplicate = strings.ToLower(cfg.TikvImporter.OnDuplicate)
+		switch cfg.TikvImporter.OnDuplicate {
+		case ReplaceOnDup, IgnoreOnDup, ErrorOnDup:
+		default:
+			return errors.Errorf("invalid config: unsupported `tikv-importer.on-duplicate` (%s)", cfg.TikvImporter.OnDuplicate)
+		}
+	}
+
 	var err error
 	cfg.TiDB.SQLMode, err = mysql.GetSQLMode(cfg.TiDB.StrSQLMode)
 	if err != nil {
diff --git a/lightning/restore/restore.go b/lightning/restore/restore.go
index e49b775..9f4ffc3 100644
--- a/lightning/restore/restore.go
+++ b/lightning/restore/restore.go
@@ -172,7 +172,7 @@ func NewRestoreController(ctx context.Context, dbMetas []*mydump.MDDatabaseMeta,
 			return nil, err
 		}
 	case config.BackendTiDB:
-		backend = kv.NewTiDBBackend(tidbMgr.db)
+		backend = kv.NewTiDBBackend(tidbMgr.db, cfg.TikvImporter.OnDuplicate)
 	default:
 		return nil, errors.New("unknown backend: " + cfg.TikvImporter.Backend)
 	}


Test patch:
diff --git a/lightning/backend/tidb_test.go b/lightning/backend/tidb_test.go
index 77e8750..73f7329 100644
--- a/lightning/backend/tidb_test.go
+++ b/lightning/backend/tidb_test.go
@@ -15,12 +15,14 @@ package backend_test
 
 import (
 	"context"
+	"database/sql"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb/types"
 
 	kv "github.com/pingcap/tidb-lightning/lightning/backend"
+	"github.com/pingcap/tidb-lightning/lightning/config"
 	"github.com/pingcap/tidb-lightning/lightning/log"
 	"github.com/pingcap/tidb-lightning/lightning/verification"
 )
@@ -28,16 +30,18 @@ import (
 var _ = Suite(&mysqlSuite{})
 
 type mysqlSuite struct {
-	mockDB  sqlmock.Sqlmock
-	backend kv.Backend
+	dbHandle *sql.DB
+	mockDB   sqlmock.Sqlmock
+	backend  kv.Backend
 }
 
 func (s *mysqlSuite) SetUpTest(c *C) {
 	db, mock, err := sqlmock.New()
 	c.Assert(err, IsNil)
 
+	s.dbHandle = db
 	s.mockDB = mock
-	s.backend = kv.NewTiDBBackend(db)
+	s.backend = kv.NewTiDBBackend(db, config.ReplaceOnDup)
 }
 
 func (s *mysqlSuite) TearDownTest(c *C) {
@@ -45,9 +49,9 @@ func (s *mysqlSuite) TearDownTest(c *C) {
 	c.Assert(s.mockDB.ExpectationsWereMet(), IsNil)
 }
 
-func (s *mysqlSuite) TestWriteRows(c *C) {
+func (s *mysqlSuite) TestWriteRowsReplaceOnDup(c *C) {
 	s.mockDB.
-		ExpectExec("\\QINSERT INTO `foo`.`bar`(`a`,`b`,`c`,`d`,`e`,`f`,`g`,`h`,`i`,`j`,`k`,`l`,`m`,`n`,`o`) VALUES(18446744073709551615,-9223372036854775808,0,NULL,7.5,5e-324,1.7976931348623157e+308,0,'甲乙丙\\r\\n\\0\\Z''\"\\\\`',0x000000abcdef,2557891634,'12.5',51)\\E").
+		ExpectExec("\\QREPLACE INTO `foo`.`bar`(`a`,`b`,`c`,`d`,`e`,`f`,`g`,`h`,`i`,`j`,`k`,`l`,`m`,`n`,`o`) VALUES(18446744073709551615,-9223372036854775808,0,NULL,7.5,5e-324,1.7976931348623157e+308,0,'甲乙丙\\r\\n\\0\\Z''\"\\\\`',0x000000abcdef,2557891634,'12.5',51)\\E").
 		WillReturnResult(sqlmock.NewResult(1, 1))
 
 	ctx := context.Background()
@@ -83,3 +87,59 @@ func (s *mysqlSuite) TestWriteRows(c *C) {
 	err = engine.WriteRows(ctx, []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o"}, dataRows)
 	c.Assert(err, IsNil)
 }
+
+func (s *mysqlSuite) TestWriteRowsIgnoreOnDup(c *C) {
+	s.mockDB.
+		ExpectExec("\\QINSERT IGNORE INTO `foo`.`bar`(`a`) VALUES(1)\\E").
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	ctx := context.Background()
+	logger := log.L()
+
+	ignoreBackend := kv.NewTiDBBackend(s.dbHandle, config.IgnoreOnDup)
+	engine, err := ignoreBackend.OpenEngine(ctx, "`foo`.`bar`", 1)
+	c.Assert(err, IsNil)
+
+	dataRows := ignoreBackend.MakeEmptyRows()
+	dataChecksum := verification.MakeKVChecksum(0, 0, 0)
+	indexRows := ignoreBackend.MakeEmptyRows()
+	indexChecksum := verification.MakeKVChecksum(0, 0, 0)
+
+	encoder := ignoreBackend.NewEncoder(nil, 0)
+	row, err := encoder.Encode(logger, []types.Datum{
+		types.NewIntDatum(1),
+	}, 1, nil)
+	c.Assert(err, IsNil)
+	row.ClassifyAndAppend(&dataRows, &dataChecksum, &indexRows, &indexChecksum)
+
+	err = engine.WriteRows(ctx, []string{"a"}, dataRows)
+	c.Assert(err, IsNil)
+}
+
+func (s *mysqlSuite) TestWriteRowsErrorOnDup(c *C) {
+	s.mockDB.
+		ExpectExec("\\QINSERT INTO `foo`.`bar`(`a`) VALUES(1)\\E").
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	ctx := context.Background()
+	logger := log.L()
+
+	ignoreBackend := kv.NewTiDBBackend(s.dbHandle, config.ErrorOnDup)
+	engine, err := ignoreBackend.OpenEngine(ctx, "`foo`.`bar`", 1)
+	c.Assert(err, IsNil)
+
+	dataRows := ignoreBackend.MakeEmptyRows()
+	dataChecksum := verification.MakeKVChecksum(0, 0, 0)
+	indexRows := ignoreBackend.MakeEmptyRows()
+	indexChecksum := verification.MakeKVChecksum(0, 0, 0)
+
+	encoder := ignoreBackend.NewEncoder(nil, 0)
+	row, err := encoder.Encode(logger, []types.Datum{
+		types.NewIntDatum(1),
+	}, 1, nil)
+	c.Assert(err, IsNil)
+	row.ClassifyAndAppend(&dataRows, &dataChecksum, &indexRows, &indexChecksum)
+
+	err = engine.WriteRows(ctx, []string{"a"}, dataRows)
+	c.Assert(err, IsNil)
+}
diff --git a/tests/tidb_duplicate_data/data/dup-schema-create.sql b/tests/tidb_duplicate_data/data/dup-schema-create.sql
new file mode 100644
index 0000000..5895df5
--- /dev/null
+++ b/tests/tidb_duplicate_data/data/dup-schema-create.sql
@@ -0,0 +1,1 @@
+CREATE DATABASE dup;
diff --git a/tests/tidb_duplicate_data/data/dup.dup-schema.sql b/tests/tidb_duplicate_data/data/dup.dup-schema.sql
new file mode 100644
index 0000000..a113b9e
--- /dev/null
+++ b/tests/tidb_duplicate_data/data/dup.dup-schema.sql
@@ -0,0 +1,1 @@
+create table dup (pk int primary key, d varchar(3));
diff --git a/tests/tidb_duplicate_data/data/dup.dup.sql b/tests/tidb_duplicate_data/data/dup.dup.sql
new file mode 100644
index 0000000..ecbb057
--- /dev/null
+++ b/tests/tidb_duplicate_data/data/dup.dup.sql
@@ -0,0 +1,3 @@
+insert into dup values
+(1, 'old'),
+(2, 'old');
diff --git a/tests/tidb_duplicate_data/error.toml b/tests/tidb_duplicate_data/error.toml
new file mode 100644
index 0000000..f9b9abf
--- /dev/null
+++ b/tests/tidb_duplicate_data/error.toml
@@ -0,0 +1,24 @@
+[lightning]
+check-requirements = false
+file = "/tmp/lightning_test_result/lightning-error-on-dup.log"
+level = "warning"
+
+[tikv-importer]
+backend = "tidb"
+on-duplicate = "error"
+
+[mydumper]
+data-source-dir = "tests/tidb_duplicate_data/data"
+
+[tidb]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+status-port = 10080
+pd-addr = "127.0.0.1:2379"
+log-level = "error"
+
+[post-restore]
+checksum = true
+compact = true
+analyze = true
diff --git a/tests/tidb_duplicate_data/ignore.toml b/tests/tidb_duplicate_data/ignore.toml
new file mode 100644
index 0000000..d88096b
--- /dev/null
+++ b/tests/tidb_duplicate_data/ignore.toml
@@ -0,0 +1,24 @@
+[lightning]
+check-requirements = false
+file = "/tmp/lightning_test_result/lightning.log"
+level = "warning"
+
+[tikv-importer]
+backend = "tidb"
+on-duplicate = "ignore"
+
+[mydumper]
+data-source-dir = "tests/tidb_duplicate_data/data"
+
+[tidb]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+status-port = 10080
+pd-addr = "127.0.0.1:2379"
+log-level = "error"
+
+[post-restore]
+checksum = true
+compact = true
+analyze = true
diff --git a/tests/tidb_duplicate_data/replace.toml b/tests/tidb_duplicate_data/replace.toml
new file mode 100644
index 0000000..4dd7f82
--- /dev/null
+++ b/tests/tidb_duplicate_data/replace.toml
@@ -0,0 +1,23 @@
+[lightning]
+check-requirements = false
+file = "/tmp/lightning_test_result/lightning.log"
+level = "warning"
+
+[tikv-importer]
+backend = "tidb"
+
+[mydumper]
+data-source-dir = "tests/tidb_duplicate_data/data"
+
+[tidb]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+status-port = 10080
+pd-addr = "127.0.0.1:2379"
+log-level = "error"
+
+[post-restore]
+checksum = true
+compact = true
+analyze = true
diff --git a/tests/tidb_duplicate_data/run.sh b/tests/tidb_duplicate_data/run.sh
new file mode 100644
index 0000000..548008b
--- /dev/null
+++ b/tests/tidb_duplicate_data/run.sh
@@ -0,0 +1,64 @@
+#!/bin/sh
+#
+# Copyright 2019 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -eu
+
+# reset substitution if last time failed half way
+# on BSD/macOS sed -i must have a following string as backup filename extension
+sed -i.bak 's/new/old/g' "tests/tidb_duplicate_data/data/dup.dup.sql" && rm tests/tidb_duplicate_data/data/dup.dup.sql.bak
+
+for type in replace ignore error; do
+    run_sql 'DROP DATABASE IF EXISTS dup;'
+
+    export GO_FAILPOINTS="github.com/pingcap/tidb-lightning/lightning/backend/FailIfImportedSomeRows=return"
+    set +e
+    run_lightning $type 2> /dev/null
+    ERRORCODE=$?
+    set -e
+    [ "$ERRORCODE" -ne 0 ]
+
+    # backup original sql to dup.dup.sql.bak
+    sed -i.bak 's/old/new/g' "tests/tidb_duplicate_data/data/dup.dup.sql"
+
+    unset GO_FAILPOINTS
+
+    if [ $type = 'error' ]; then
+        set +e
+        run_lightning $type
+        ERRORCODE=$?
+        set -e
+        [ "$ERRORCODE" -ne 0 ]
+        tail -20 "$TEST_DIR/lightning-error-on-dup.log" > "$TEST_DIR/lightning-error-on-dup.tail"
+        grep -Fq 'Duplicate entry' "$TEST_DIR/lightning-error-on-dup.tail"
+    elif [ $type = 'replace' ]; then
+        run_lightning $type
+        run_sql 'SELECT count(*) FROM dup.dup'
+        check_contains 'count(*): 2'
+        run_sql 'SELECT d FROM dup.dup WHERE pk = 1'
+        check_contains 'd: new'
+        run_sql 'SELECT d FROM dup.dup WHERE pk = 2'
+        check_contains 'd: new'
+    elif [ $type = 'ignore' ]; then
+        run_lightning $type
+        run_sql 'SELECT count(*) FROM dup.dup'
+        check_contains 'count(*): 2'
+        run_sql 'SELECT d FROM dup.dup WHERE pk = 1'
+        check_contains 'd: old'
+        run_sql 'SELECT d FROM dup.dup WHERE pk = 2'
+        check_contains 'd: new'
+    fi
+
+    mv tests/tidb_duplicate_data/data/dup.dup.sql.bak tests/tidb_duplicate_data/data/dup.dup.sql
+done
\ No newline at end of file