br: Add pre-check of duplicate table in the downstream (#55044) (#59587)

close #55087

Author: ti-chi-bot

br/pkg/errors/errors.go

@@ -68,6 +68,7 @@ var (
 	ErrRestoreIncompatibleSys  = errors.Normalize("incompatible system table", errors.RFCCodeText("BR:Restore:ErrRestoreIncompatibleSys"))
 	ErrUnsupportedSystemTable  = errors.Normalize("the system table isn't supported for restoring yet", errors.RFCCodeText("BR:Restore:ErrUnsupportedSysTable"))
 	ErrDatabasesAlreadyExisted = errors.Normalize("databases already existed in restored cluster", errors.RFCCodeText("BR:Restore:ErrDatabasesAlreadyExisted"))
+	ErrTablesAlreadyExisted    = errors.Normalize("tables already existed in restored cluster", errors.RFCCodeText("BR:Restore:ErrTablesAlreadyExisted"))
 
 	// ErrStreamLogTaskExist is the error when stream log task already exists, because of supporting single task currently.
 	ErrStreamLogTaskExist = errors.Normalize("stream task already exists", errors.RFCCodeText("BR:Stream:ErrStreamLogTaskExist"))

br/pkg/glue/glue.go

@@ -13,6 +13,13 @@ import (
 	pd "github.com/tikv/pd/client"
 )
 
+type GlueClient int
+
+const (
+	ClientCLP GlueClient = iota
+	ClientSql
+)
+
 // Glue is an abstraction of TiDB function calls used in BR.
 type Glue interface {
 	GetDomain(store kv.Storage) (*domain.Domain, error)
@@ -36,6 +43,9 @@ type Glue interface {
 	// we can close domain as soon as possible.
 	// and we must reuse the exists session and don't close it in SQL backup job.
 	UseOneShotSession(store kv.Storage, closeDomain bool, fn func(se Session) error) error
+
+	// GetClient returns the client type of the glue
+	GetClient() GlueClient
 }
 
 // Session is an abstraction of the session.Session interface.

br/pkg/gluetidb/glue.go

@@ -157,6 +157,10 @@ func (g Glue) UseOneShotSession(store kv.Storage, closeDomain bool, fn func(glue
 	return nil
 }
 
+func (Glue) GetClient() glue.GlueClient {
+	return glue.ClientCLP
+}
+
 // GetSessionCtx implements glue.Glue
 func (gs *tidbSession) GetSessionCtx() sessionctx.Context {
 	return gs.se
@@ -366,3 +370,7 @@ func (m *MockGlue) UseOneShotSession(store kv.Storage, closeDomain bool, fn func
 	}
 	return fn(glueSession)
 }
+
+func (m *MockGlue) GetClient() glue.GlueClient {
+	return glue.ClientCLP
+}

br/pkg/gluetikv/glue.go

@@ -73,3 +73,7 @@ func (Glue) GetVersion() string {
 func (g Glue) UseOneShotSession(store kv.Storage, closeDomain bool, fn func(glue.Session) error) error {
 	return nil
 }
+
+func (Glue) GetClient() glue.GlueClient {
+	return glue.ClientCLP
+}

br/pkg/task/restore.go

@@ -147,6 +147,7 @@ func DefineRestoreCommonFlags(flags *pflag.FlagSet) {
 	_ = flags.MarkHidden(FlagPDConcurrency)
 	_ = flags.MarkHidden(FlagBatchFlushInterval)
 	_ = flags.MarkHidden(FlagDdlBatchSize)
+	_ = flags.MarkHidden(flagUseFSR)
 }
 
 // ParseFromFlags parses the config from the flag set.
@@ -1089,7 +1090,7 @@ func checkTableExistence(ctx context.Context, mgr *conn.Mgr, tables []*metautil.
 		}
 	}
 	if !allUnique {
-		return errors.Errorf("Target tables already existed: %s", message)
+		return errors.Annotate(berrors.ErrTablesAlreadyExisted, message)
 	}
 	return nil
 }

errors.toml

@@ -281,6 +281,11 @@ error = '''
 failed to write and ingest
 '''
 
+["BR:Restore:ErrTablesAlreadyExisted"]
+error = '''
+tables already existed in restored cluster
+'''
+
 ["BR:Restore:ErrUnsupportedSysTable"]
 error = '''
 the system table isn't supported for restoring yet

pkg/executor/brie.go

@@ -710,6 +710,10 @@ func (gs *tidbGlue) UseOneShotSession(_ kv.Storage, _ bool, fn func(se glue.Sess
 	return fn(glueSession)
 }
 
+func (*tidbGlue) GetClient() glue.GlueClient {
+	return glue.ClientSql
+}
+
 type tidbGlueSession struct {
 	// the session context of the brie task's subtask, such as `CREATE TABLE`.
 	se sessionctx.Context

tests/realtikvtest/brietest/BUILD.bazel

@@ -20,6 +20,7 @@ go_test(
         "//pkg/config",
         "//pkg/executor",
         "//pkg/parser/mysql",
+        "//pkg/session",
         "//pkg/sessionctx/binloginfo",
         "//pkg/store/mockstore/mockcopr",
         "//pkg/testkit",

tests/realtikvtest/brietest/brie_test.go

@@ -15,6 +15,7 @@
 package brietest
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"strings"
@@ -24,6 +25,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/executor"
+	"github.com/pingcap/tidb/pkg/session"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zapcore"
@@ -60,6 +62,7 @@ func TestShowBackupQuery(t *testing.T) {
 	restoreQuery := fmt.Sprintf("RESTORE TABLE `test`.`foo` FROM 'local://%s'", sqlTmp)
 	tk.MustQuery(restoreQuery)
 	res = tk.MustQuery("show br job query 2;")
+	tk.MustExec("drop table foo;")
 	res.CheckContain(restoreQuery)
 }
 
@@ -68,6 +71,7 @@ func TestShowBackupQueryRedact(t *testing.T) {
 
 	executor.ResetGlobalBRIEQueueForTest()
 	failpoint.Enable("github.com/pingcap/tidb/pkg/executor/block-on-brie", "return")
+	defer failpoint.Disable("github.com/pingcap/tidb/pkg/executor/block-on-brie")
 	ch := make(chan any)
 	go func() {
 		tk := testkit.NewTestKit(t, tk.Session().GetStore())
@@ -102,6 +106,7 @@ func TestCancel(t *testing.T) {
 	executor.ResetGlobalBRIEQueueForTest()
 	tk.MustExec("use test;")
 	failpoint.Enable("github.com/pingcap/tidb/pkg/executor/block-on-brie", "return")
+	defer failpoint.Disable("github.com/pingcap/tidb/pkg/executor/block-on-brie")
 
 	req := require.New(t)
 	ch := make(chan struct{})
@@ -126,3 +131,49 @@ func TestCancel(t *testing.T) {
 		req.FailNow("the backup job doesn't be canceled")
 	}
 }
+
+func TestExistedTables(t *testing.T) {
+	tk := initTestKit(t)
+	tmp := makeTempDirForBackup(t)
+	sqlTmp := strings.ReplaceAll(tmp, "'", "''")
+	executor.ResetGlobalBRIEQueueForTest()
+	tk.MustExec("use test;")
+	for i := 0; i < 10; i++ {
+		tableName := fmt.Sprintf("foo%d", i)
+		tk.MustExec(fmt.Sprintf("create table %s(pk int primary key auto_increment, v varchar(255));", tableName))
+		tk.MustExec(fmt.Sprintf("insert into %s(v) values %s;", tableName, strings.TrimSuffix(strings.Repeat("('hello, world'),", 100), ",")))
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		backupQuery := fmt.Sprintf("BACKUP DATABASE * TO 'local://%s'", sqlTmp)
+		_ = tk.MustQuery(backupQuery)
+	}()
+	select {
+	case <-time.After(20 * time.Second):
+		t.Fatal("Backup operation exceeded")
+	case <-done:
+	}
+
+	done = make(chan struct{})
+	go func() {
+		defer close(done)
+		restoreQuery := fmt.Sprintf("RESTORE DATABASE * FROM 'local://%s'", sqlTmp)
+		res, err := tk.Exec(restoreQuery)
+		require.NoError(t, err)
+
+		_, err = session.ResultSetToStringSlice(context.Background(), tk.Session(), res)
+		require.ErrorContains(t, err, "table already exists")
+	}()
+	select {
+	case <-time.After(20 * time.Second):
+		t.Fatal("Restore operation exceeded")
+	case <-done:
+	}
+
+	for i := 0; i < 10; i++ {
+		tableName := fmt.Sprintf("foo%d", i)
+		tk.MustExec(fmt.Sprintf("drop table %s;", tableName))
+	}
+}