Merge branch 'master' into feature/dropping-multi-indexes

.github/workflows/bug-closed.yml

@@ -0,0 +1,29 @@
+name: Bug Closed
+
+on:
+  issues:
+    types:
+      - closed
+
+jobs:
+  label_issues:
+    if: |
+      contains(github.event.issue.labels.*.name, 'type/bug') &&
+      !(contains(join(github.event.issue.labels.*.name, ', '), 'affects-') &&
+        contains(join(github.event.issue.labels.*.name, ', '), 'backport-'))
+    runs-on: ubuntu-latest
+    permissions:
+      issues: write
+    steps:
+      - name: Label issues
+        uses: andymckay/[email protected]
+        with:
+          add-labels: "needs-more-info"
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - name: Add comment
+        uses: peter-evans/[email protected]
+        with:
+          issue-number: ${{ github.event.issue.number }}
+          body: |
+            Please check whether the issue should be labeled with 'affects-x.y' or 'backport-x.y.z', 
+            and then remove 'needs-more-info' label.

br/pkg/backup/client.go

@@ -268,10 +268,6 @@ func BuildBackupRangeAndSchema(
 			continue
 		}
 
-		idAlloc := autoid.NewAllocator(storage, dbInfo.ID, false, autoid.RowIDAllocType)
-		seqAlloc := autoid.NewAllocator(storage, dbInfo.ID, false, autoid.SequenceType)
-		randAlloc := autoid.NewAllocator(storage, dbInfo.ID, false, autoid.AutoRandomType)
-
 		tables, err := m.ListTables(dbInfo.ID)
 		if err != nil {
 			return nil, nil, errors.Trace(err)
@@ -294,14 +290,19 @@ func BuildBackupRangeAndSchema(
 				zap.String("table", tableInfo.Name.O),
 			)
 
+			tblVer := autoid.AllocOptionTableInfoVersion(tableInfo.Version)
+			idAlloc := autoid.NewAllocator(storage, dbInfo.ID, tableInfo.ID, false, autoid.RowIDAllocType, tblVer)
+			seqAlloc := autoid.NewAllocator(storage, dbInfo.ID, tableInfo.ID, false, autoid.SequenceType, tblVer)
+			randAlloc := autoid.NewAllocator(storage, dbInfo.ID, tableInfo.ID, false, autoid.AutoRandomType, tblVer)
+
 			var globalAutoID int64
 			switch {
 			case tableInfo.IsSequence():
-				globalAutoID, err = seqAlloc.NextGlobalAutoID(tableInfo.ID)
+				globalAutoID, err = seqAlloc.NextGlobalAutoID()
 			case tableInfo.IsView() || !utils.NeedAutoID(tableInfo):
 				// no auto ID for views or table without either rowID nor auto_increment ID.
 			default:
-				globalAutoID, err = idAlloc.NextGlobalAutoID(tableInfo.ID)
+				globalAutoID, err = idAlloc.NextGlobalAutoID()
 			}
 			if err != nil {
 				return nil, nil, errors.Trace(err)
@@ -311,7 +312,7 @@ func BuildBackupRangeAndSchema(
 			if tableInfo.PKIsHandle && tableInfo.ContainsAutoRandomBits() {
 				// this table has auto_random id, we need backup and rebase in restoration
 				var globalAutoRandID int64
-				globalAutoRandID, err = randAlloc.NextGlobalAutoID(tableInfo.ID)
+				globalAutoRandID, err = randAlloc.NextGlobalAutoID()
 				if err != nil {
 					return nil, nil, errors.Trace(err)
 				}

br/pkg/conn/conn.go

@@ -5,6 +5,7 @@ package conn
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"os"
 	"sync"
 	"time"
@@ -162,6 +163,27 @@ func GetAllTiKVStores(
 	return stores[:j], nil
 }
 
+func checkStoresAlive(ctx context.Context,
+	pdclient pd.Client,
+	storeBehavior StoreBehavior) error {
+	// Check live tikv.
+	stores, err := GetAllTiKVStores(ctx, pdclient, storeBehavior)
+	if err != nil {
+		log.Error("fail to get store", zap.Error(err))
+		return errors.Trace(err)
+	}
+
+	liveStoreCount := 0
+	for _, s := range stores {
+		if s.GetState() != metapb.StoreState_Up {
+			continue
+		}
+		liveStoreCount++
+	}
+	log.Info("checked alive KV stores", zap.Int("aliveStores", liveStoreCount), zap.Int("totalStores", len(stores)))
+	return nil
+}
+
 // NewMgr creates a new Mgr.
 //
 // Domain is optional for Backup, set `needDomain` to false to disable
@@ -170,7 +192,6 @@ func NewMgr(
 	ctx context.Context,
 	g glue.Glue,
 	pdAddrs string,
-	storage kv.Storage,
 	tlsConf *tls.Config,
 	securityOption pd.SecurityOption,
 	keepalive keepalive.ClientParameters,
@@ -184,10 +205,7 @@ func NewMgr(
 		ctx = opentracing.ContextWithSpan(ctx, span1)
 	}
 
-	tikvStorage, ok := storage.(tikv.Storage)
-	if !ok {
-		return nil, berrors.ErrKVNotTiKV
-	}
+	log.Info("new mgr", zap.String("pdAddrs", pdAddrs))
 
 	controller, err := pdutil.NewPdController(ctx, pdAddrs, tlsConf, securityOption)
 	if err != nil {
@@ -201,20 +219,21 @@ func NewMgr(
 				"if you believe it's OK, use --check-requirements=false to skip.")
 		}
 	}
-	log.Info("new mgr", zap.String("pdAddrs", pdAddrs))
 
-	// Check live tikv.
-	stores, err := GetAllTiKVStores(ctx, controller.GetPDClient(), storeBehavior)
+	err = checkStoresAlive(ctx, controller.GetPDClient(), storeBehavior)
 	if err != nil {
-		log.Error("fail to get store", zap.Error(err))
 		return nil, errors.Trace(err)
 	}
-	liveStoreCount := 0
-	for _, s := range stores {
-		if s.GetState() != metapb.StoreState_Up {
-			continue
-		}
-		liveStoreCount++
+
+	// Disable GC because TiDB enables GC already.
+	storage, err := g.Open(fmt.Sprintf("tikv://%s?disableGC=true", pdAddrs), securityOption)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	tikvStorage, ok := storage.(tikv.Storage)
+	if !ok {
+		return nil, berrors.ErrKVNotTiKV
 	}
 
 	var dom *domain.Domain
@@ -232,9 +251,12 @@ func NewMgr(
 		dom:          dom,
 		tlsConf:      tlsConf,
 		ownsStorage:  g.OwnsStorage(),
+		grpcClis: struct {
+			mu   sync.Mutex
+			clis map[uint64]*grpc.ClientConn
+		}{clis: make(map[uint64]*grpc.ClientConn)},
+		keepalive: keepalive,
 	}
-	mgr.grpcClis.clis = make(map[uint64]*grpc.ClientConn)
-	mgr.keepalive = keepalive
 	return mgr, nil
 }
 

br/pkg/conn/conn_test.go

@@ -43,6 +43,63 @@ func (fpdc fakePDClient) GetAllStores(context.Context, ...pd.GetStoreOption) ([]
 	return append([]*metapb.Store{}, fpdc.stores...), nil
 }
 
+func (s *testClientSuite) TestCheckStoresAlive(c *C) {
+	stores := []*metapb.Store{
+		{
+			Id:    1,
+			State: metapb.StoreState_Up,
+			Labels: []*metapb.StoreLabel{
+				{
+					Key:   "engine",
+					Value: "tiflash",
+				},
+			},
+		},
+		{
+			Id:    2,
+			State: metapb.StoreState_Offline,
+			Labels: []*metapb.StoreLabel{
+				{
+					Key:   "engine",
+					Value: "tiflash",
+				},
+			},
+		},
+		{
+			Id:    3,
+			State: metapb.StoreState_Up,
+			Labels: []*metapb.StoreLabel{
+				{
+					Key:   "engine",
+					Value: "tikv",
+				},
+			},
+		},
+		{
+			Id:    4,
+			State: metapb.StoreState_Offline,
+			Labels: []*metapb.StoreLabel{
+				{
+					Key:   "engine",
+					Value: "tikv",
+				},
+			},
+		},
+	}
+
+	fpdc := fakePDClient{
+		stores: stores,
+	}
+
+	kvStores, err := GetAllTiKVStores(s.ctx, fpdc, SkipTiFlash)
+	c.Assert(err, IsNil)
+	c.Assert(len(kvStores), Equals, 2)
+	c.Assert(kvStores, DeepEquals, stores[2:])
+
+	err = checkStoresAlive(s.ctx, fpdc, SkipTiFlash)
+	c.Assert(err, IsNil)
+}
+
 func (s *testClientSuite) TestGetAllTiKVStores(c *C) {
 	testCases := []struct {
 		stores         []*metapb.Store

br/pkg/lightning/backend/kv/allocator.go

@@ -40,7 +40,7 @@ func NewPanickingAllocators(base int64) autoid.Allocators {
 }
 
 // Rebase implements the autoid.Allocator interface
-func (alloc *panickingAllocator) Rebase(tableID, newBase int64, allocIDs bool) error {
+func (alloc *panickingAllocator) Rebase(newBase int64, allocIDs bool) error {
 	// CAS
 	for {
 		oldBase := atomic.LoadInt64(alloc.base)

br/pkg/lightning/backend/local/local.go

@@ -1272,9 +1272,7 @@ func (local *local) allocateTSIfNotExists(ctx context.Context, engine *File) err
 	return engine.saveEngineMeta()
 }
 
-// CloseEngine closes backend engine by uuid
-// NOTE: we will return nil if engine is not exist. This will happen if engine import&cleanup successfully
-// but exit before update checkpoint. Thus after restart, we will try to import this engine again.
+// CloseEngine closes backend engine by uuid.
 func (local *local) CloseEngine(ctx context.Context, cfg *backend.EngineConfig, engineUUID uuid.UUID) error {
 	// flush mem table to storage, to free memory,
 	// ask others' advise, looks like unnecessary, but with this we can control memory precisely.
@@ -1283,10 +1281,6 @@ func (local *local) CloseEngine(ctx context.Context, cfg *backend.EngineConfig,
 		// recovery mode, we should reopen this engine file
 		db, err := local.openEngineDB(engineUUID, true)
 		if err != nil {
-			// if engine db does not exist, just skip
-			if os.IsNotExist(errors.Cause(err)) {
-				return nil
-			}
 			return err
 		}
 		engineFile := &File{

br/pkg/lightning/checkpoints/checkpoints.go

@@ -490,7 +490,7 @@ type DB interface {
 	// It assumes the entire table has not been imported before and will fill in
 	// default values for the column permutations and checksums.
 	InsertEngineCheckpoints(ctx context.Context, tableName string, checkpoints map[int32]*EngineCheckpoint) error
-	Update(checkpointDiffs map[string]*TableCheckpointDiff)
+	Update(checkpointDiffs map[string]*TableCheckpointDiff) error
 
 	RemoveCheckpoint(ctx context.Context, tableName string) error
 	// MoveCheckpoints renames the checkpoint schema to include a suffix
@@ -599,7 +599,9 @@ func (*NullCheckpointsDB) InsertEngineCheckpoints(_ context.Context, _ string, _
 	return nil
 }
 
-func (*NullCheckpointsDB) Update(map[string]*TableCheckpointDiff) {}
+func (*NullCheckpointsDB) Update(map[string]*TableCheckpointDiff) error {
+	return nil
+}
 
 type MySQLCheckpointsDB struct {
 	db     *sql.DB
@@ -859,15 +861,15 @@ func (cpdb *MySQLCheckpointsDB) InsertEngineCheckpoints(ctx context.Context, tab
 	return nil
 }
 
-func (cpdb *MySQLCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDiff) {
+func (cpdb *MySQLCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDiff) error {
 	chunkQuery := fmt.Sprintf(UpdateChunkTemplate, cpdb.schema, CheckpointTableNameChunk)
 	rebaseQuery := fmt.Sprintf(UpdateTableRebaseTemplate, cpdb.schema, CheckpointTableNameTable)
 	tableStatusQuery := fmt.Sprintf(UpdateTableStatusTemplate, cpdb.schema, CheckpointTableNameTable)
 	tableChecksumQuery := fmt.Sprintf(UpdateTableChecksumTemplate, cpdb.schema, CheckpointTableNameTable)
 	engineStatusQuery := fmt.Sprintf(UpdateEngineTemplate, cpdb.schema, CheckpointTableNameEngine)
 
 	s := common.SQLWithRetry{DB: cpdb.db, Logger: log.L()}
-	err := s.Transact(context.Background(), "update checkpoints", func(c context.Context, tx *sql.Tx) error {
+	return s.Transact(context.Background(), "update checkpoints", func(c context.Context, tx *sql.Tx) error {
 		chunkStmt, e := tx.PrepareContext(c, chunkQuery)
 		if e != nil {
 			return errors.Trace(e)
@@ -933,9 +935,6 @@ func (cpdb *MySQLCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpoi
 
 		return nil
 	})
-	if err != nil {
-		log.L().Error("save checkpoint failed", zap.Error(err))
-	}
 }
 
 type FileCheckpointsDB struct {
@@ -1165,7 +1164,7 @@ func (cpdb *FileCheckpointsDB) InsertEngineCheckpoints(_ context.Context, tableN
 	return errors.Trace(cpdb.save())
 }
 
-func (cpdb *FileCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDiff) {
+func (cpdb *FileCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDiff) error {
 	cpdb.lock.Lock()
 	defer cpdb.lock.Unlock()
 
@@ -1200,9 +1199,7 @@ func (cpdb *FileCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpoin
 		}
 	}
 
-	if err := cpdb.save(); err != nil {
-		log.L().Error("save checkpoint failed", zap.Error(err))
-	}
+	return cpdb.save()
 }
 
 // Management functions ----------------------------------------------------------------------------

br/pkg/lightning/checkpoints/glue_checkpoint.go

@@ -411,20 +411,20 @@ func (g GlueCheckpointsDB) InsertEngineCheckpoints(ctx context.Context, tableNam
 	return errors.Trace(err)
 }
 
-func (g GlueCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDiff) {
+func (g GlueCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDiff) error {
 	logger := log.L()
 	se, err := g.getSessionFunc()
 	if err != nil {
 		log.L().Error("can't get a session to update GlueCheckpointsDB", zap.Error(errors.Trace(err)))
-		return
+		return err
 	}
 	defer se.Close()
 
 	chunkQuery := fmt.Sprintf(UpdateChunkTemplate, g.schema, CheckpointTableNameChunk)
 	rebaseQuery := fmt.Sprintf(UpdateTableRebaseTemplate, g.schema, CheckpointTableNameTable)
 	tableStatusQuery := fmt.Sprintf(UpdateTableStatusTemplate, g.schema, CheckpointTableNameTable)
 	engineStatusQuery := fmt.Sprintf(UpdateEngineTemplate, g.schema, CheckpointTableNameEngine)
-	err = Transact(context.Background(), "update checkpoints", se, logger, func(c context.Context, s Session) error {
+	return Transact(context.Background(), "update checkpoints", se, logger, func(c context.Context, s Session) error {
 		chunkStmt, _, _, err := s.PrepareStmt(chunkQuery)
 		if err != nil {
 			return errors.Trace(err)
@@ -501,9 +501,6 @@ func (g GlueCheckpointsDB) Update(checkpointDiffs map[string]*TableCheckpointDif
 		}
 		return nil
 	})
-	if err != nil {
-		log.L().Error("save checkpoint failed", zap.Error(err))
-	}
 }
 
 func (g GlueCheckpointsDB) RemoveCheckpoint(ctx context.Context, tableName string) error {