Merge remote-tracking branch 'pingcap/master' into feat-reorg-part-me…

…rge-master

DEPS.bzl

@@ -1163,8 +1163,8 @@ def go_deps():
         name = "com_github_go_sql_driver_mysql",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/go-sql-driver/mysql",
-        sum = "h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=",
-        version = "v1.6.0",
+        sum = "h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc=",
+        version = "v1.7.0",
     )
     go_repository(
         name = "com_github_go_stack_stack",
@@ -3519,8 +3519,8 @@ def go_deps():
         name = "com_github_tikv_client_go_v2",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/tikv/client-go/v2",
-        sum = "h1:vlgZedcfExiTzB3BB4nt5CpaghDfm9La/0Ofn7weIUA=",
-        version = "v2.0.3-0.20221129032117-857772dd0907",
+        sum = "h1:Nr2EhvqkOE9xFyU7LV9c9EbsgN3OzVALdbfobK7Fmn4=",
+        version = "v2.0.3-0.20221205084317-ad59ca833a78",
     )
     go_repository(
         name = "com_github_tikv_pd_client",
@@ -4132,10 +4132,6 @@ def go_deps():
         name = "io_etcd_go_etcd_raft_v3",
         build_file_proto_mode = "disable_global",
         importpath = "go.etcd.io/etcd/raft/v3",
-        patch_args = ["-p1"],
-        patches = [
-            "//build/patches:io_etcd_go_etcd_raft_v3.patch",
-        ],
         sum = "h1:uCC37qOXqBvKqTGHGyhASsaCsnTuJugl1GvneJNwHWo=",
         version = "v3.5.2",
     )

br/pkg/checkpoint/checkpoint.go

@@ -353,6 +353,7 @@ func (r *CheckpointRunner) startCheckpointLoop(ctx context.Context, tickDuration
 		var wg sync.WaitGroup
 		errCh := r.startCheckpointRunner(cctx, &wg)
 		ticker := time.NewTicker(tickDuration)
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ctx.Done():

br/pkg/lightning/restore/check_info_test.go

@@ -493,11 +493,11 @@ func TestCheckTableEmpty(t *testing.T) {
 	require.NoError(t, err)
 	mock.MatchExpectationsInOrder(false)
 	targetInfoGetter.targetDBGlue = glue.NewExternalTiDBGlue(db, mysql.ModeNone)
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
-	mock.ExpectQuery("SELECT 1 FROM `test2`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test2`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
 	rc.checkTemplate = NewSimpleTemplate()
 	err = rc.checkTableEmpty(ctx)
@@ -510,13 +510,13 @@ func TestCheckTableEmpty(t *testing.T) {
 	targetInfoGetter.targetDBGlue = glue.NewExternalTiDBGlue(db, mysql.ModeNone)
 	mock.MatchExpectationsInOrder(false)
 	// test auto retry retryable error
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnError(&gmysql.MySQLError{Number: errno.ErrPDServerTimeout})
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
-	mock.ExpectQuery("SELECT 1 FROM `test2`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test2`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).AddRow(1))
 	rc.checkTemplate = NewSimpleTemplate()
 	err = rc.checkTableEmpty(ctx)
@@ -532,11 +532,11 @@ func TestCheckTableEmpty(t *testing.T) {
 	require.NoError(t, err)
 	targetInfoGetter.targetDBGlue = glue.NewExternalTiDBGlue(db, mysql.ModeNone)
 	mock.MatchExpectationsInOrder(false)
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).AddRow(1))
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
-	mock.ExpectQuery("SELECT 1 FROM `test2`.`tbl1` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test2`.`tbl1` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).AddRow(1))
 	rc.checkTemplate = NewSimpleTemplate()
 	err = rc.checkTableEmpty(ctx)
@@ -576,7 +576,7 @@ func TestCheckTableEmpty(t *testing.T) {
 	require.NoError(t, err)
 	targetInfoGetter.targetDBGlue = glue.NewExternalTiDBGlue(db, mysql.ModeNone)
 	// only need to check the one that is not in checkpoint
-	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test1`.`tbl2` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(sqlmock.NewRows([]string{""}).RowError(0, sql.ErrNoRows))
 	err = rc.checkTableEmpty(ctx)
 	require.NoError(t, err)

br/pkg/lightning/restore/get_pre_info.go

@@ -189,7 +189,12 @@ func (g *TargetInfoGetterImpl) IsTableEmpty(ctx context.Context, schemaName stri
 	}
 	var dump int
 	err = exec.QueryRow(ctx, "check table empty",
-		fmt.Sprintf("SELECT 1 FROM %s LIMIT 1", common.UniqueTable(schemaName, tableName)),
+		// Here we use the `USE INDEX()` hint to skip fetch the record from index.
+		// In Lightning, if previous importing is halted half-way, it is possible that
+		// the data is partially imported, but the index data has not been imported.
+		// In this situation, if no hint is added, the SQL executor might fetch the record from index,
+		// which is empty.  This will result in missing check.
+		fmt.Sprintf("SELECT 1 FROM %s USE INDEX() LIMIT 1", common.UniqueTable(schemaName, tableName)),
 		&dump,
 	)
 

br/pkg/lightning/restore/get_pre_info_test.go

@@ -762,7 +762,7 @@ func TestGetPreInfoIsTableEmpty(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, lnConfig, targetGetter.cfg)
 
-	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` USE INDEX\\(\\) LIMIT 1").
 		WillReturnError(&mysql_sql_driver.MySQLError{
 			Number:  errno.ErrNoSuchTable,
 			Message: "Table 'test_db.test_tbl' doesn't exist",
@@ -772,7 +772,7 @@ func TestGetPreInfoIsTableEmpty(t *testing.T) {
 	require.NotNil(t, pIsEmpty)
 	require.Equal(t, true, *pIsEmpty)
 
-	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(
 			sqlmock.NewRows([]string{"1"}).
 				RowError(0, sql.ErrNoRows),
@@ -782,7 +782,7 @@ func TestGetPreInfoIsTableEmpty(t *testing.T) {
 	require.NotNil(t, pIsEmpty)
 	require.Equal(t, true, *pIsEmpty)
 
-	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` USE INDEX\\(\\) LIMIT 1").
 		WillReturnRows(
 			sqlmock.NewRows([]string{"1"}).AddRow(1),
 		)
@@ -791,7 +791,7 @@ func TestGetPreInfoIsTableEmpty(t *testing.T) {
 	require.NotNil(t, pIsEmpty)
 	require.Equal(t, false, *pIsEmpty)
 
-	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` LIMIT 1").
+	mock.ExpectQuery("SELECT 1 FROM `test_db`.`test_tbl` USE INDEX\\(\\) LIMIT 1").
 		WillReturnError(errors.New("some dummy error"))
 	_, err = targetGetter.IsTableEmpty(ctx, "test_db", "test_tbl")
 	require.Error(t, err)

br/pkg/lightning/restore/table_restore.go

@@ -384,6 +384,11 @@ func (tr *TableRestore) restoreEngines(pCtx context.Context, rc *Controller, cp
 	if cp.Status < checkpoints.CheckpointStatusIndexImported {
 		var err error
 		if indexEngineCp.Status < checkpoints.CheckpointStatusImported {
+			failpoint.Inject("FailBeforeStartImportingIndexEngine", func() {
+				errMsg := "fail before importing index KV data"
+				tr.logger.Warn(errMsg)
+				failpoint.Return(errors.New(errMsg))
+			})
 			err = tr.importKV(ctx, closedIndexEngine, rc, indexEngineID)
 			failpoint.Inject("FailBeforeIndexEngineImported", func() {
 				finished := rc.status.FinishedFileSize.Load()

br/tests/br_foreign_key/run.sh

@@ -50,6 +50,5 @@ check_contains 'id: 3'
 check_contains 'a: NULL'
 check_contains 'b: 11'
 check_contains 'b: 13'
-check_contains 'b: 14'
 
 run_sql "drop schema $DB"

br/tests/lightning_check_partial_imported/config.toml

@@ -0,0 +1,5 @@
+[tikv-importer]
+backend = "local"
+
+[mydumper.csv]
+header = true

br/tests/lightning_check_partial_imported/data/db01.tbl01-schema.sql

@@ -0,0 +1,12 @@
+CREATE TABLE tbl01 (
+  `id` INTEGER,
+  `val` VARCHAR(64),
+  `aaa` CHAR(66) DEFAULT NULL,
+  `bbb` CHAR(10) NOT NULL,
+  `ccc` CHAR(42) DEFAULT NULL,
+  `ddd` CHAR(42) DEFAULT NULL,
+  `eee` CHAR(66) DEFAULT NULL,
+  `fff` VARCHAR(128) DEFAULT NULL,
+  KEY `aaa` (`aaa`),
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

br/tests/lightning_check_partial_imported/data/db01.tbl01.csv

@@ -0,0 +1,6 @@
+id,val,aaa,bbb,ccc,ddd,eee,fff
+1,"v01","a01","b01","c01","d01","e01","f01"
+2,"v02","a02","b02","c02","d02","e02","f02"
+3,"v03","a03","b03","c03","d03","e03","f03"
+4,"v04","a04","b04","c04","d04","e04","f04"
+5,"v05","a05","b05","c05","d05","e05","f05"

br/tests/lightning_check_partial_imported/run.sh

@@ -0,0 +1,47 @@
+#!/bin/bash
+#
+# Copyright 2022 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,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+MYDIR=$(dirname "${BASH_SOURCE[0]}")
+set -eux
+
+check_cluster_version 4 0 0 'local backend' || exit 0
+
+export GO_FAILPOINTS="github.com/pingcap/tidb/br/pkg/lightning/restore/FailBeforeStartImportingIndexEngine=return"
+set +e
+if run_lightning; then
+    echo "The first import doesn't fail as expected" >&2
+    exit 1
+fi
+set -e
+
+data_records=$(tail -n +2 "${MYDIR}/data/db01.tbl01.csv" | wc -l | xargs echo )
+run_sql "SELECT COUNT(*) FROM db01.tbl01 USE INDEX();"
+check_contains "${data_records}"
+
+export GO_FAILPOINTS=""
+set +e
+if run_lightning --check-requirements=1; then
+    echo "The pre-check doesn't find out the non-empty table problem"
+    exit 2
+fi
+set -e
+
+run_sql "TRUNCATE TABLE db01.tbl01;"
+run_lightning --check-requirements=1
+run_sql "SELECT COUNT(*) FROM db01.tbl01;"
+check_contains "${data_records}"
+run_sql "SELECT COUNT(*) FROM db01.tbl01 USE INDEX();"
+check_contains "${data_records}"

cmd/explaintest/r/collation_agg_func_disabled.result

@@ -133,13 +133,13 @@ approx_count_distinct(value collate utf8mb4_bin, value1)
 create table tt(a char(10), b enum('a', 'B', 'c'), c set('a', 'B', 'c'), d json) collate utf8mb4_general_ci;
 insert into tt values ("a", "a", "a", JSON_OBJECT("a", "a"));
 insert into tt values ("A", "A", "A", JSON_OBJECT("A", "A"));
-Error 1265: Data truncated for column 'b' at row 1
+Error 1265 (01000): Data truncated for column 'b' at row 1
 insert into tt values ("b", "b", "b", JSON_OBJECT("b", "b"));
-Error 1265: Data truncated for column 'b' at row 1
+Error 1265 (01000): Data truncated for column 'b' at row 1
 insert into tt values ("B", "B", "B", JSON_OBJECT("B", "B"));
 insert into tt values ("c", "c", "c", JSON_OBJECT("c", "c"));
 insert into tt values ("C", "C", "C", JSON_OBJECT("C", "C"));
-Error 1265: Data truncated for column 'b' at row 1
+Error 1265 (01000): Data truncated for column 'b' at row 1
 split table tt by (0), (1), (2), (3), (4), (5);
 desc format='brief' select min(a) from tt;
 id	estRows	task	access object	operator info
@@ -210,9 +210,9 @@ select min(b) from tt;
 min(b)
 B
 desc format='brief' select min(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select min(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select max(b) from tt;
 id	estRows	task	access object	operator info
 StreamAgg	1.00	root		funcs:max(Column#8)->Column#6
@@ -223,9 +223,9 @@ select max(b) from tt;
 max(b)
 c
 desc format='brief' select max(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select max(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select min(c) from tt;
 id	estRows	task	access object	operator info
 HashAgg	1.00	root		funcs:min(collation_agg_func.tt.c)->Column#6
@@ -235,9 +235,9 @@ select min(c) from tt;
 min(c)
 B
 desc format='brief' select min(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select min(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select max(c) from tt;
 id	estRows	task	access object	operator info
 HashAgg	1.00	root		funcs:max(collation_agg_func.tt.c)->Column#6
@@ -247,9 +247,9 @@ select max(c) from tt;
 max(c)
 c
 desc format='brief' select max(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select max(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select min(d) from tt;
 id	estRows	task	access object	operator info
 StreamAgg	1.00	root		funcs:min(collation_agg_func.tt.d)->Column#6

cmd/explaintest/r/collation_agg_func_enabled.result

@@ -207,9 +207,9 @@ select min(b) from tt;
 min(b)
 a
 desc format='brief' select min(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select min(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select max(b) from tt;
 id	estRows	task	access object	operator info
 StreamAgg	1.00	root		funcs:max(Column#8)->Column#6
@@ -220,9 +220,9 @@ select max(b) from tt;
 max(b)
 c
 desc format='brief' select max(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select max(b collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select min(c) from tt;
 id	estRows	task	access object	operator info
 HashAgg	1.00	root		funcs:min(collation_agg_func.tt.c)->Column#6
@@ -232,9 +232,9 @@ select min(c) from tt;
 min(c)
 a
 desc format='brief' select min(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select min(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select max(c) from tt;
 id	estRows	task	access object	operator info
 HashAgg	1.00	root		funcs:max(collation_agg_func.tt.c)->Column#6
@@ -244,9 +244,9 @@ select max(c) from tt;
 max(c)
 c
 desc format='brief' select max(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 select max(c collate utf8mb4_bin) from tt;
-Error 1235: This version of TiDB doesn't yet support 'use collate clause for enum or set'
+Error 1235 (42000): This version of TiDB doesn't yet support 'use collate clause for enum or set'
 desc format='brief' select min(d) from tt;
 id	estRows	task	access object	operator info
 StreamAgg	1.00	root		funcs:min(collation_agg_func.tt.d)->Column#6

cmd/explaintest/r/collation_check_use_collation_disabled.result

@@ -32,7 +32,7 @@ drop table if exists t;
 create table t(a enum('a', 'b') charset utf8mb4 collate utf8mb4_general_ci, b varchar(20));
 insert into t values ("b", "c");
 insert into t values ("B", "b");
-Error 1265: Data truncated for column 'a' at row 1
+Error 1265 (01000): Data truncated for column 'a' at row 1
 select * from t where 'B' collate utf8mb4_general_ci in (a);
 a	b
 select * from t where 'B' collate utf8mb4_bin in (a);
@@ -82,7 +82,7 @@ drop table if exists t;
 create table t(a set('a', 'b') charset utf8mb4 collate utf8mb4_general_ci, b varchar(20));
 insert into t values ("b", "c");
 insert into t values ("B", "b");
-Error 1265: Data truncated for column 'a' at row 1
+Error 1265 (01000): Data truncated for column 'a' at row 1
 select * from t where 'B' collate utf8mb4_general_ci in (a);
 a	b
 select * from t where 'B' collate utf8mb4_bin in (a);