Skip to content

Commit 923e054

Browse files
wjhuang2016wjhuang2016
authored
Merge branch 'master' into fix_reverse
2 parents 93ef8ff + e3c56b7 commit 923e054

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

47 files changed

+9940
-9256
lines changed

‎br/pkg/restore/pipeline_items.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -360,6 +360,7 @@ func (b *tikvSender) restoreWorker(ctx context.Context, ranges <-chan drainResul
360360
eg.Go(func() error {
361361
e := b.client.RestoreFiles(ectx, files, r.result.RewriteRules, b.updateCh)
362362
if e != nil {
363+
r.done()
363364
return e
364365
}
365366
log.Info("restore batch done", rtree.ZapRanges(r.result.Ranges))

‎br/pkg/storage/local.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import (
1212
)
1313

1414
const (
15-
localDirPerm os.FileMode = 0o755
15+
localDirPerm os.FileMode = 0o777
1616
localFilePerm os.FileMode = 0o644
1717
// LocalURIPrefix represents the local storage prefix.
1818
LocalURIPrefix = "file://"

‎br/pkg/task/backup_test.go

Lines changed: 23 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -6,56 +6,53 @@ import (
66
"testing"
77
"time"
88

9-
. "github.com/pingcap/check"
10-
backuppb "github.com/pingcap/kvproto/pkg/brpb"
9+
backup "github.com/pingcap/kvproto/pkg/brpb"
10+
"github.com/stretchr/testify/require"
1111
)
1212

13-
var _ = Suite(&testBackupSuite{})
13+
func TestParseTSString(t *testing.T) {
14+
t.Parallel()
1415

15-
func TestT(t *testing.T) {
16-
TestingT(t)
17-
}
18-
19-
type testBackupSuite struct{}
20-
21-
func (s *testBackupSuite) TestParseTSString(c *C) {
2216
var (
2317
ts uint64
2418
err error
2519
)
2620

2721
ts, err = parseTSString("")
28-
c.Assert(err, IsNil)
29-
c.Assert(int(ts), Equals, 0)
22+
require.NoError(t, err)
23+
require.Zero(t, ts)
3024

3125
ts, err = parseTSString("400036290571534337")
32-
c.Assert(err, IsNil)
33-
c.Assert(int(ts), Equals, 400036290571534337)
26+
require.NoError(t, err)
27+
require.Equal(t, uint64(400036290571534337), ts)
3428

3529
_, offset := time.Now().Local().Zone()
3630
ts, err = parseTSString("2018-05-11 01:42:23")
37-
c.Assert(err, IsNil)
38-
c.Assert(int(ts), Equals, 400032515489792000-(offset*1000)<<18)
31+
require.NoError(t, err)
32+
require.Equal(t, uint64(400032515489792000-(offset*1000)<<18), ts)
3933
}
4034

41-
func (s *testBackupSuite) TestParseCompressionType(c *C) {
35+
func TestParseCompressionType(t *testing.T) {
36+
t.Parallel()
37+
4238
var (
43-
ct backuppb.CompressionType
39+
ct backup.CompressionType
4440
err error
4541
)
4642
ct, err = parseCompressionType("lz4")
47-
c.Assert(err, IsNil)
48-
c.Assert(int(ct), Equals, 1)
43+
require.NoError(t, err)
44+
require.Equal(t, 1, int(ct))
4945

5046
ct, err = parseCompressionType("snappy")
51-
c.Assert(err, IsNil)
52-
c.Assert(int(ct), Equals, 2)
47+
require.NoError(t, err)
48+
require.Equal(t, 2, int(ct))
5349

5450
ct, err = parseCompressionType("zstd")
55-
c.Assert(err, IsNil)
56-
c.Assert(int(ct), Equals, 3)
51+
require.NoError(t, err)
52+
require.Equal(t, 3, int(ct))
5753

5854
ct, err = parseCompressionType("Other Compression (strings)")
59-
c.Assert(err, ErrorMatches, "invalid compression.*")
60-
c.Assert(int(ct), Equals, 0)
55+
require.Error(t, err)
56+
require.Regexp(t, "invalid compression.*", err.Error())
57+
require.Zero(t, ct)
6158
}

‎br/pkg/task/common_test.go

Lines changed: 39 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -5,18 +5,15 @@ package task
55
import (
66
"encoding/hex"
77
"fmt"
8+
"testing"
89

9-
. "github.com/pingcap/check"
10-
backuppb "github.com/pingcap/kvproto/pkg/brpb"
10+
backup "github.com/pingcap/kvproto/pkg/brpb"
1111
"github.com/pingcap/kvproto/pkg/encryptionpb"
1212
"github.com/pingcap/tidb/config"
1313
"github.com/spf13/pflag"
14+
"github.com/stretchr/testify/require"
1415
)
1516

16-
var _ = Suite(&testCommonSuite{})
17-
18-
type testCommonSuite struct{}
19-
2017
type fakeValue string
2118

2219
func (f fakeValue) String() string {
@@ -31,43 +28,48 @@ func (f fakeValue) Type() string {
3128
panic("implement me")
3229
}
3330

34-
func (*testCommonSuite) TestUrlNoQuery(c *C) {
31+
func TestUrlNoQuery(t *testing.T) {
32+
t.Parallel()
3533
flag := &pflag.Flag{
3634
Name: flagStorage,
3735
Value: fakeValue("s3://some/what?secret=a123456789&key=987654321"),
3836
}
39-
4037
field := flagToZapField(flag)
41-
c.Assert(field.Key, Equals, flagStorage)
42-
c.Assert(field.Interface.(fmt.Stringer).String(), Equals, "s3://some/what")
38+
require.Equal(t, flagStorage, field.Key)
39+
require.Equal(t, "s3://some/what", field.Interface.(fmt.Stringer).String())
4340
}
4441

45-
func (s *testCommonSuite) TestTiDBConfigUnchanged(c *C) {
42+
func TestTiDBConfigUnchanged(t *testing.T) {
43+
t.Parallel()
4644
cfg := config.GetGlobalConfig()
4745
restoreConfig := enableTiDBConfig()
48-
c.Assert(cfg, Not(DeepEquals), config.GetGlobalConfig())
46+
require.NotEqual(t, config.GetGlobalConfig(), cfg)
4947
restoreConfig()
50-
c.Assert(cfg, DeepEquals, config.GetGlobalConfig())
48+
require.Equal(t, config.GetGlobalConfig(), cfg)
5149
}
5250

53-
func (s *testCommonSuite) TestStripingPDURL(c *C) {
51+
func TestStripingPDURL(t *testing.T) {
52+
t.Parallel()
5453
nor1, err := normalizePDURL("https://pd:5432", true)
55-
c.Assert(err, IsNil)
56-
c.Assert(nor1, Equals, "pd:5432")
54+
require.NoError(t, err)
55+
require.Equal(t, "pd:5432", nor1)
5756
_, err = normalizePDURL("https://pd.pingcap.com", false)
58-
c.Assert(err, ErrorMatches, ".*pd url starts with https while TLS disabled.*")
57+
require.Error(t, err)
58+
require.Regexp(t, ".*pd url starts with https while TLS disabled.*", err.Error())
5959
_, err = normalizePDURL("http://127.0.0.1:2379", true)
60-
c.Assert(err, ErrorMatches, ".*pd url starts with http while TLS enabled.*")
60+
require.Error(t, err)
61+
require.Regexp(t, ".*pd url starts with http while TLS enabled.*", err.Error())
6162
nor, err := normalizePDURL("http://127.0.0.1", false)
62-
c.Assert(nor, Equals, "127.0.0.1")
63-
c.Assert(err, IsNil)
63+
require.NoError(t, err)
64+
require.Equal(t, "127.0.0.1", nor)
6465
noChange, err := normalizePDURL("127.0.0.1:2379", false)
65-
c.Assert(err, IsNil)
66-
c.Assert(noChange, Equals, "127.0.0.1:2379")
66+
require.NoError(t, err)
67+
require.Equal(t, "127.0.0.1:2379", noChange)
6768
}
6869

69-
func (s *testCommonSuite) TestCheckCipherKeyMatch(c *C) {
70-
testCases := []struct {
70+
func TestCheckCipherKeyMatch(t *testing.T) {
71+
t.Parallel()
72+
cases := []struct {
7173
CipherType encryptionpb.EncryptionMethod
7274
CipherKey string
7375
ok bool
@@ -112,19 +114,18 @@ func (s *testCommonSuite) TestCheckCipherKeyMatch(c *C) {
112114
},
113115
}
114116

115-
for _, t := range testCases {
116-
cipherKey, err := hex.DecodeString(t.CipherKey)
117-
c.Assert(err, IsNil)
118-
119-
r := checkCipherKeyMatch(&backuppb.CipherInfo{
120-
CipherType: t.CipherType,
117+
for _, c := range cases {
118+
cipherKey, err := hex.DecodeString(c.CipherKey)
119+
require.NoError(t, err)
120+
require.Equal(t, c.ok, checkCipherKeyMatch(&backup.CipherInfo{
121+
CipherType: c.CipherType,
121122
CipherKey: cipherKey,
122-
})
123-
c.Assert(r, Equals, t.ok)
123+
}))
124124
}
125125
}
126126

127-
func (s *testCommonSuite) TestCheckCipherKey(c *C) {
127+
func TestCheckCipherKey(t *testing.T) {
128+
t.Parallel()
128129
cases := []struct {
129130
cipherKey string
130131
keyFile string
@@ -152,12 +153,12 @@ func (s *testCommonSuite) TestCheckCipherKey(c *C) {
152153
},
153154
}
154155

155-
for _, t := range cases {
156-
err := checkCipherKey(t.cipherKey, t.keyFile)
157-
if t.ok {
158-
c.Assert(err, IsNil)
156+
for _, c := range cases {
157+
err := checkCipherKey(c.cipherKey, c.keyFile)
158+
if c.ok {
159+
require.NoError(t, err)
159160
} else {
160-
c.Assert(err, NotNil)
161+
require.Error(t, err)
161162
}
162163
}
163164
}

‎br/pkg/task/restore_test.go

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,20 +3,20 @@
33
package task
44

55
import (
6-
. "github.com/pingcap/check"
6+
"testing"
7+
78
"github.com/pingcap/tidb/br/pkg/restore"
9+
"github.com/stretchr/testify/require"
810
)
911

10-
type testRestoreSuite struct{}
11-
12-
var _ = Suite(&testRestoreSuite{})
12+
func TestRestoreConfigAdjust(t *testing.T) {
13+
t.Parallel()
1314

14-
func (s *testRestoreSuite) TestRestoreConfigAdjust(c *C) {
1515
cfg := &RestoreConfig{}
1616
cfg.adjustRestoreConfig()
1717

18-
c.Assert(cfg.Config.Concurrency, Equals, uint32(defaultRestoreConcurrency))
19-
c.Assert(cfg.Config.SwitchModeInterval, Equals, defaultSwitchInterval)
20-
c.Assert(cfg.MergeSmallRegionKeyCount, Equals, restore.DefaultMergeRegionKeyCount)
21-
c.Assert(cfg.MergeSmallRegionSizeBytes, Equals, restore.DefaultMergeRegionSizeBytes)
18+
require.Equal(t, uint32(defaultRestoreConcurrency), cfg.Config.Concurrency)
19+
require.Equal(t, defaultSwitchInterval, cfg.Config.SwitchModeInterval)
20+
require.Equal(t, restore.DefaultMergeRegionKeyCount, cfg.MergeSmallRegionKeyCount)
21+
require.Equal(t, restore.DefaultMergeRegionSizeBytes, cfg.MergeSmallRegionSizeBytes)
2222
}

‎ddl/column.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1427,7 +1427,7 @@ func (w *updateColumnWorker) cleanRowMap() {
14271427
}
14281428
}
14291429

1430-
// BackfillDataInTxn will backfill the table record in a transaction.
1430+
// BackfillDataInTxn will backfill the table record in a transaction. A lock corresponds to a rowKey if the value of rowKey is changed.
14311431
func (w *updateColumnWorker) BackfillDataInTxn(handleRange reorgBackfillTask) (taskCtx backfillTaskContext, errInTxn error) {
14321432
oprStartTime := time.Now()
14331433
errInTxn = kv.RunInNewTxn(context.Background(), w.sessCtx.GetStore(), true, func(ctx context.Context, txn kv.Transaction) error {

‎ddl/ddl.go

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ import (
4747
"github.com/pingcap/tidb/table"
4848
goutil "github.com/pingcap/tidb/util"
4949
"github.com/pingcap/tidb/util/admin"
50+
"github.com/pingcap/tidb/util/gcutil"
5051
"github.com/pingcap/tidb/util/logutil"
5152
"go.etcd.io/etcd/clientv3"
5253
"go.uber.org/zap"
@@ -731,3 +732,39 @@ func init() {
731732
RunInGoTest = true
732733
}
733734
}
735+
736+
// GetDropOrTruncateTableInfoFromJobsByStore implements GetDropOrTruncateTableInfoFromJobs
737+
func GetDropOrTruncateTableInfoFromJobsByStore(jobs []*model.Job, gcSafePoint uint64, getTable func(uint64, int64, int64) (*model.TableInfo, error), fn func(*model.Job, *model.TableInfo) (bool, error)) (bool, error) {
738+
for _, job := range jobs {
739+
// Check GC safe point for getting snapshot infoSchema.
740+
err := gcutil.ValidateSnapshotWithGCSafePoint(job.StartTS, gcSafePoint)
741+
if err != nil {
742+
return false, err
743+
}
744+
if job.Type != model.ActionDropTable && job.Type != model.ActionTruncateTable {
745+
continue
746+
}
747+
748+
tbl, err := getTable(job.StartTS, job.SchemaID, job.TableID)
749+
if err != nil {
750+
if meta.ErrDBNotExists.Equal(err) {
751+
// The dropped/truncated DDL maybe execute failed that caused by the parallel DDL execution,
752+
// then can't find the table from the snapshot info-schema. Should just ignore error here,
753+
// see more in TestParallelDropSchemaAndDropTable.
754+
continue
755+
}
756+
return false, err
757+
}
758+
if tbl == nil {
759+
// The dropped/truncated DDL maybe execute failed that caused by the parallel DDL execution,
760+
// then can't find the table from the snapshot info-schema. Should just ignore error here,
761+
// see more in TestParallelDropSchemaAndDropTable.
762+
continue
763+
}
764+
finish, err := fn(job, tbl)
765+
if err != nil || finish {
766+
return finish, err
767+
}
768+
}
769+
return false, nil
770+
}

‎ddl/index.go

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1293,8 +1293,8 @@ func (w *addIndexWorker) batchCheckUniqueKey(txn kv.Transaction, idxRecords []*i
12931293
return nil
12941294
}
12951295

1296-
// BackfillDataInTxn will backfill table index in a transaction. If the value of rowKey is changed, there must be some other transactions
1297-
// update the row, result in write conflict, so the txn will rollback and retry.
1296+
// BackfillDataInTxn will backfill table index in a transaction. A lock corresponds to a rowKey if the value of rowKey is changed,
1297+
// Note that index columns values may change, and an index is not allowed to be added, so the txn will rollback and retry.
12981298
// BackfillDataInTxn will add w.batchCnt indices once, default value of w.batchCnt is 128.
12991299
func (w *addIndexWorker) BackfillDataInTxn(handleRange reorgBackfillTask) (taskCtx backfillTaskContext, errInTxn error) {
13001300
failpoint.Inject("errorMockPanic", func(val failpoint.Value) {
@@ -1329,6 +1329,13 @@ func (w *addIndexWorker) BackfillDataInTxn(handleRange reorgBackfillTask) (taskC
13291329
continue
13301330
}
13311331

1332+
// We need to add this lock to make sure pessimistic transaction can realize this operation.
1333+
// For the normal pessimistic transaction, it's ok. But if async commmit is used, it may lead to inconsistent data and index.
1334+
err := txn.LockKeys(context.Background(), new(kv.LockCtx), idxRecord.key)
1335+
if err != nil {
1336+
return errors.Trace(err)
1337+
}
1338+
13321339
// Create the index.
13331340
handle, err := w.index.Create(w.sessCtx, txn, idxRecord.vals, idxRecord.handle, idxRecord.rsData)
13341341
if err != nil {

‎ddl/placement/bundle.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,10 @@ func newBundleFromOptions(options *model.PlacementSettings) (bundle *Bundle, err
207207
return nil, fmt.Errorf("%w: options can not be nil", ErrInvalidPlacementOptions)
208208
}
209209

210+
if options.Followers > uint64(8) {
211+
return nil, fmt.Errorf("%w: followers should be less than or equal to 8: %d", ErrInvalidPlacementOptions, options.Followers)
212+
}
213+
210214
// always prefer the sugar syntax, which gives better schedule results most of the time
211215
isSyntaxSugar := true
212216
if len(options.LeaderConstraints) > 0 || len(options.LearnerConstraints) > 0 || len(options.FollowerConstraints) > 0 || len(options.Constraints) > 0 || options.Learners > 0 {

0 commit comments

Comments
 (0)