Skip to content

Commit

Permalink
Merge remote-tracking branch 'pingcap/release-6.5' into cherry-pick-3…
Browse files Browse the repository at this point in the history
…8738-6.5
  • Loading branch information
mjonss committed Apr 10, 2023
2 parents 2ce98d3 + 3ab3be0 commit 76f02d8
Show file tree
Hide file tree
Showing 109 changed files with 2,179 additions and 970 deletions.
2 changes: 1 addition & 1 deletion WORKSPACE
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ go_download_sdk(
"https://mirrors.aliyun.com/golang/{}",
"https://dl.google.com/go/{}",
],
version = "1.19.5",
version = "1.19.8",
)

go_register_toolchains(
Expand Down
8 changes: 5 additions & 3 deletions br/pkg/conn/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -281,7 +281,8 @@ func (mgr *Mgr) GetTS(ctx context.Context) (uint64, error) {
}

// GetMergeRegionSizeAndCount returns the tikv config `coprocessor.region-split-size` and `coprocessor.region-split-key`.
func (mgr *Mgr) GetMergeRegionSizeAndCount(ctx context.Context, client *http.Client) (uint64, uint64, error) {
// returns the default config when failed.
func (mgr *Mgr) GetMergeRegionSizeAndCount(ctx context.Context, client *http.Client) (uint64, uint64) {
regionSplitSize := DefaultMergeRegionSizeBytes
regionSplitKeys := DefaultMergeRegionKeyCount
type coprocessor struct {
Expand Down Expand Up @@ -310,9 +311,10 @@ func (mgr *Mgr) GetMergeRegionSizeAndCount(ctx context.Context, client *http.Cli
return nil
})
if err != nil {
return 0, 0, errors.Trace(err)
log.Warn("meet error when getting config from TiKV; using default", logutil.ShortError(err))
return DefaultMergeRegionSizeBytes, DefaultMergeRegionKeyCount
}
return regionSplitSize, regionSplitKeys, nil
return regionSplitSize, regionSplitKeys
}

// GetConfigFromTiKV get configs from all alive tikv stores.
Expand Down
35 changes: 33 additions & 2 deletions br/pkg/conn/conn_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,38 @@ func TestGetMergeRegionSizeAndCount(t *testing.T) {
regionSplitSize: DefaultMergeRegionSizeBytes,
regionSplitKeys: DefaultMergeRegionKeyCount,
},
{
stores: []*metapb.Store{
{
Id: 1,
State: metapb.StoreState_Up,
Labels: []*metapb.StoreLabel{
{
Key: "engine",
Value: "tiflash",
},
},
},
{
Id: 2,
State: metapb.StoreState_Up,
Labels: []*metapb.StoreLabel{
{
Key: "engine",
Value: "tikv",
},
},
},
},
content: []string{
"",
// Assuming the TiKV has failed due to some reason.
"",
},
// no tikv detected in this case
regionSplitSize: DefaultMergeRegionSizeBytes,
regionSplitKeys: DefaultMergeRegionKeyCount,
},
{
stores: []*metapb.Store{
{
Expand Down Expand Up @@ -388,8 +420,7 @@ func TestGetMergeRegionSizeAndCount(t *testing.T) {
httpCli := mockServer.Client()
mgr := &Mgr{PdController: &pdutil.PdController{}}
mgr.PdController.SetPDClient(pdCli)
rs, rk, err := mgr.GetMergeRegionSizeAndCount(ctx, httpCli)
require.NoError(t, err)
rs, rk := mgr.GetMergeRegionSizeAndCount(ctx, httpCli)
require.Equal(t, ca.regionSplitSize, rs)
require.Equal(t, ca.regionSplitKeys, rk)
mockServer.Close()
Expand Down
20 changes: 19 additions & 1 deletion br/pkg/gluetidb/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
load("@io_bazel_rules_go//go:def.bzl", "go_library")
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")

go_library(
name = "gluetidb",
Expand All @@ -25,3 +25,21 @@ go_library(
"@org_uber_go_zap//:zap",
],
)

go_test(
name = "gluetidb_test",
timeout = "short",
srcs = ["glue_test.go"],
embed = [":gluetidb"],
flaky = True,
deps = [
"//ddl",
"//kv",
"//meta",
"//parser/model",
"//sessionctx",
"//testkit",
"@com_github_pingcap_failpoint//:failpoint",
"@com_github_stretchr_testify//require",
],
)
31 changes: 28 additions & 3 deletions br/pkg/gluetidb/glue.go
Original file line number Diff line number Diff line change
Expand Up @@ -207,9 +207,34 @@ func (gs *tidbSession) CreatePlacementPolicy(ctx context.Context, policy *model.
return d.CreatePlacementPolicyWithInfo(gs.se, policy, ddl.OnExistIgnore)
}

// SplitBatchCreateTable provide a way to split batch into small batch when batch size is large than 6 MB.
// The raft entry has limit size of 6 MB, a batch of CreateTables may hit this limitation
// TODO: shall query string be set for each split batch create, it looks does not matter if we set once for all.
func (gs *tidbSession) SplitBatchCreateTable(schema model.CIStr, infos []*model.TableInfo, cs ...ddl.CreateTableWithInfoConfigurier) error {
var err error
d := domain.GetDomain(gs.se).DDL()

if err = d.BatchCreateTableWithInfo(gs.se, schema, infos, append(cs, ddl.OnExistIgnore)...); kv.ErrEntryTooLarge.Equal(err) {
log.Info("entry too large, split batch create table", zap.Int("num table", len(infos)))
if len(infos) == 1 {
return err
}
mid := len(infos) / 2
err = gs.SplitBatchCreateTable(schema, infos[:mid], cs...)
if err != nil {
return err
}
err = gs.SplitBatchCreateTable(schema, infos[mid:], cs...)
if err != nil {
return err
}
return nil
}
return err
}

// CreateTables implements glue.BatchCreateTableSession.
func (gs *tidbSession) CreateTables(ctx context.Context, tables map[string][]*model.TableInfo, cs ...ddl.CreateTableWithInfoConfigurier) error {
d := domain.GetDomain(gs.se).DDL()
var dbName model.CIStr

// Disable foreign key check when batch create tables.
Expand Down Expand Up @@ -237,8 +262,8 @@ func (gs *tidbSession) CreateTables(ctx context.Context, tables map[string][]*mo
cloneTables = append(cloneTables, table)
}
gs.se.SetValue(sessionctx.QueryString, queryBuilder.String())
err := d.BatchCreateTableWithInfo(gs.se, dbName, cloneTables, append(cs, ddl.OnExistIgnore)...)
if err != nil {

if err := gs.SplitBatchCreateTable(dbName, cloneTables, cs...); err != nil {
//It is possible to failure when TiDB does not support model.ActionCreateTables.
//In this circumstance, BatchCreateTableWithInfo returns errno.ErrInvalidDDLJob,
//we fall back to old way that creating table one by one
Expand Down
208 changes: 208 additions & 0 deletions br/pkg/gluetidb/glue_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,208 @@
// Copyright 2023 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.

package gluetidb

import (
"context"
"strconv"
"testing"

"github.com/pingcap/failpoint"
"github.com/pingcap/tidb/ddl"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/meta"
"github.com/pingcap/tidb/parser/model"
"github.com/pingcap/tidb/sessionctx"
"github.com/pingcap/tidb/testkit"
"github.com/stretchr/testify/require"
)

// batch create table with table id reused
func TestSplitBatchCreateTableWithTableId(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("drop table if exists table_id_resued1")
tk.MustExec("drop table if exists table_id_resued2")
tk.MustExec("drop table if exists table_id_new")

d := dom.DDL()
require.NotNil(t, d)

infos1 := []*model.TableInfo{}
infos1 = append(infos1, &model.TableInfo{
ID: 124,
Name: model.NewCIStr("table_id_resued1"),
})
infos1 = append(infos1, &model.TableInfo{
ID: 125,
Name: model.NewCIStr("table_id_resued2"),
})

se := &tidbSession{se: tk.Session()}

// keep/reused table id verification
tk.Session().SetValue(sessionctx.QueryString, "skip")
err := se.SplitBatchCreateTable(model.NewCIStr("test"), infos1, ddl.AllocTableIDIf(func(ti *model.TableInfo) bool {
return false
}))
require.NoError(t, err)

tk.MustQuery("select tidb_table_id from information_schema.tables where table_name = 'table_id_resued1'").Check(testkit.Rows("124"))
tk.MustQuery("select tidb_table_id from information_schema.tables where table_name = 'table_id_resued2'").Check(testkit.Rows("125"))
ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnOthers)

// allocate new table id verification
// query the global id
var id int64
err = kv.RunInNewTxn(ctx, store, true, func(_ context.Context, txn kv.Transaction) error {
m := meta.NewMeta(txn)
var err error
id, err = m.GenGlobalID()
return err
})

require.NoError(t, err)

infos2 := []*model.TableInfo{}
infos2 = append(infos2, &model.TableInfo{
ID: 124,
Name: model.NewCIStr("table_id_new"),
})

tk.Session().SetValue(sessionctx.QueryString, "skip")
err = se.SplitBatchCreateTable(model.NewCIStr("test"), infos2, ddl.AllocTableIDIf(func(ti *model.TableInfo) bool {
return true
}))
require.NoError(t, err)

idGen, ok := tk.MustQuery("select tidb_table_id from information_schema.tables where table_name = 'table_id_new'").Rows()[0][0].(string)
require.True(t, ok)
idGenNum, err := strconv.ParseInt(idGen, 10, 64)
require.NoError(t, err)
require.Greater(t, idGenNum, id)

// a empty table info with len(info3) = 0
infos3 := []*model.TableInfo{}

err = se.SplitBatchCreateTable(model.NewCIStr("test"), infos3, ddl.AllocTableIDIf(func(ti *model.TableInfo) bool {
return false
}))
require.NoError(t, err)
}

// batch create table with table id reused
func TestSplitBatchCreateTable(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("drop table if exists table_1")
tk.MustExec("drop table if exists table_2")
tk.MustExec("drop table if exists table_3")

d := dom.DDL()
require.NotNil(t, d)

infos := []*model.TableInfo{}
infos = append(infos, &model.TableInfo{
ID: 1234,
Name: model.NewCIStr("tables_1"),
})
infos = append(infos, &model.TableInfo{
ID: 1235,
Name: model.NewCIStr("tables_2"),
})
infos = append(infos, &model.TableInfo{
ID: 1236,
Name: model.NewCIStr("tables_3"),
})

se := &tidbSession{se: tk.Session()}

// keep/reused table id verification
tk.Session().SetValue(sessionctx.QueryString, "skip")
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/RestoreBatchCreateTableEntryTooLarge", "return(1)"))
err := se.SplitBatchCreateTable(model.NewCIStr("test"), infos, ddl.AllocTableIDIf(func(ti *model.TableInfo) bool {
return false
}))

require.NoError(t, err)
tk.MustQuery("show tables like '%tables_%'").Check(testkit.Rows("tables_1", "tables_2", "tables_3"))
jobs := tk.MustQuery("admin show ddl jobs").Rows()
require.Greater(t, len(jobs), 3)
// check table_1
job1 := jobs[0]
require.Equal(t, "test", job1[1])
require.Equal(t, "tables_3", job1[2])
require.Equal(t, "create tables", job1[3])
require.Equal(t, "public", job1[4])

// check table_2
job2 := jobs[1]
require.Equal(t, "test", job2[1])
require.Equal(t, "tables_2", job2[2])
require.Equal(t, "create tables", job2[3])
require.Equal(t, "public", job2[4])

// check table_3
job3 := jobs[2]
require.Equal(t, "test", job3[1])
require.Equal(t, "tables_1", job3[2])
require.Equal(t, "create tables", job3[3])
require.Equal(t, "public", job3[4])

// check reused table id
tk.MustQuery("select tidb_table_id from information_schema.tables where table_name = 'tables_1'").Check(testkit.Rows("1234"))
tk.MustQuery("select tidb_table_id from information_schema.tables where table_name = 'tables_2'").Check(testkit.Rows("1235"))
tk.MustQuery("select tidb_table_id from information_schema.tables where table_name = 'tables_3'").Check(testkit.Rows("1236"))

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/RestoreBatchCreateTableEntryTooLarge"))
}

// batch create table with table id reused
func TestSplitBatchCreateTableFailWithEntryTooLarge(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("drop table if exists table_1")
tk.MustExec("drop table if exists table_2")
tk.MustExec("drop table if exists table_3")

d := dom.DDL()
require.NotNil(t, d)

infos := []*model.TableInfo{}
infos = append(infos, &model.TableInfo{
Name: model.NewCIStr("tables_1"),
})
infos = append(infos, &model.TableInfo{
Name: model.NewCIStr("tables_2"),
})
infos = append(infos, &model.TableInfo{
Name: model.NewCIStr("tables_3"),
})

se := &tidbSession{se: tk.Session()}

tk.Session().SetValue(sessionctx.QueryString, "skip")
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/RestoreBatchCreateTableEntryTooLarge", "return(0)"))
err := se.SplitBatchCreateTable(model.NewCIStr("test"), infos, ddl.AllocTableIDIf(func(ti *model.TableInfo) bool {
return true
}))

require.True(t, kv.ErrEntryTooLarge.Equal(err))

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/RestoreBatchCreateTableEntryTooLarge"))
}
1 change: 1 addition & 0 deletions br/pkg/lightning/backend/local/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@ go_test(
"//br/pkg/lightning/glue",
"//br/pkg/lightning/log",
"//br/pkg/lightning/mydump",
"//br/pkg/lightning/worker",
"//br/pkg/membuf",
"//br/pkg/mock",
"//br/pkg/pdutil",
Expand Down
Loading

0 comments on commit 76f02d8

Please sign in to comment.