Skip to content

Commit

Permalink
ddl: support table placement option (#27799)
Browse files Browse the repository at this point in the history
  • Loading branch information
@AilinKid
AilinKid authored Sep 7, 2021
1 parent 2ae0bb7 commit 6aec001
Show file tree
Hide file tree
Showing 14 changed files with 257 additions and 213 deletions.
86 changes: 81 additions & 5 deletions ddl/ddl_api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,6 @@ import (
"github.com/pingcap/tidb/util/domainutil"
"github.com/pingcap/tidb/util/logutil"
"github.com/pingcap/tidb/util/mock"
"github.com/pingcap/tidb/util/placementpolicy"
"github.com/pingcap/tidb/util/set"
"go.uber.org/zap"
)
Expand Down Expand Up @@ -1609,6 +1608,24 @@ func checkTableInfoValidWithStmt(ctx sessionctx.Context, tbInfo *model.TableInfo
}
}
}
// Can not use both a placement policy and direct assignment. If you alter specify both in a CREATE TABLE or ALTER TABLE an error will be returned.
if tbInfo.DirectPlacementOpts != nil && tbInfo.PlacementPolicyRef != nil {
return errors.Trace(ErrPlacementPolicyWithDirectOption.GenWithStackByArgs(tbInfo.PlacementPolicyRef.Name))
}
if tbInfo.DirectPlacementOpts != nil {
// check the direct placement option compatibility.
if err := checkPolicyValidation(tbInfo.DirectPlacementOpts); err != nil {
return errors.Trace(err)
}
}
if tbInfo.PlacementPolicyRef != nil {
// placement policy reference will override the direct placement options.
policy, ok := ctx.GetInfoSchema().(infoschema.InfoSchema).PolicyByName(tbInfo.PlacementPolicyRef.Name)
if !ok {
return errors.Trace(infoschema.ErrPlacementPolicyNotExists.GenWithStackByArgs(tbInfo.PlacementPolicyRef.Name))
}
tbInfo.PlacementPolicyRef.ID = policy.ID
}
return nil
}

Expand Down Expand Up @@ -2275,6 +2292,65 @@ func handleTableOptions(options []*ast.TableOption, tbInfo *model.TableInfo) err
return errors.Trace(errUnsupportedEngineTemporary)
}
}
case ast.TableOptionPlacementPolicy:
tbInfo.PlacementPolicyRef = &model.PolicyRefInfo{
Name: model.NewCIStr(op.StrValue),
}
case ast.TableOptionPlacementPrimaryRegion:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.PrimaryRegion = op.StrValue
case ast.TableOptionPlacementRegions:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.Regions = op.StrValue
case ast.TableOptionPlacementFollowerCount:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.Followers = op.UintValue
case ast.TableOptionPlacementVoterCount:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.Voters = op.UintValue
case ast.TableOptionPlacementLearnerCount:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.Learners = op.UintValue
case ast.TableOptionPlacementSchedule:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.Schedule = op.StrValue
case ast.TableOptionPlacementConstraints:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.Constraints = op.StrValue
case ast.TableOptionPlacementLeaderConstraints:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.LeaderConstraints = op.StrValue
case ast.TableOptionPlacementLearnerConstraints:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.LearnerConstraints = op.StrValue
case ast.TableOptionPlacementFollowerConstraints:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.FollowerConstraints = op.StrValue
case ast.TableOptionPlacementVoterConstraints:
if tbInfo.DirectPlacementOpts == nil {
tbInfo.DirectPlacementOpts = &model.PlacementSettings{}
}
tbInfo.DirectPlacementOpts.VoterConstraints = op.StrValue
}
}
shardingBits := shardingBits(tbInfo)
Expand Down Expand Up @@ -6222,8 +6298,8 @@ func (d *ddl) AlterTablePartitionAttributes(ctx sessionctx.Context, ident ast.Id
return errors.Trace(err)
}

func buildPolicyInfo(name model.CIStr, options []*ast.PlacementOption) (*placementpolicy.PolicyInfo, error) {
policyInfo := &placementpolicy.PolicyInfo{}
func buildPolicyInfo(name model.CIStr, options []*ast.PlacementOption) (*model.PolicyInfo, error) {
policyInfo := &model.PolicyInfo{PlacementSettings: &model.PlacementSettings{}}
policyInfo.Name = name
for _, opt := range options {
switch opt.Tp {
Expand Down Expand Up @@ -6275,7 +6351,7 @@ func (d *ddl) CreatePlacementPolicy(ctx sessionctx.Context, stmt *ast.CreatePlac
return errors.Trace(err)
}

err = checkPolicyValidation(policyInfo)
err = checkPolicyValidation(policyInfo.PlacementSettings)
if err != nil {
return err
}
Expand Down Expand Up @@ -6331,7 +6407,7 @@ func (d *ddl) AlterPlacementPolicy(ctx sessionctx.Context, stmt *ast.AlterPlacem
return errors.Trace(err)
}

err = checkPolicyValidation(newPolicyInfo)
err = checkPolicyValidation(newPolicyInfo.PlacementSettings)
if err != nil {
return err
}
Expand Down
3 changes: 3 additions & 0 deletions ddl/error.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -259,6 +259,9 @@ var (
// ErrInvalidPlacementPolicyCheck is returned when txn_scope and commit data changing do not meet the placement policy
ErrInvalidPlacementPolicyCheck = dbterror.ClassDDL.NewStd(mysql.ErrPlacementPolicyCheck)

// ErrPlacementPolicyWithDirectOption is returned when create/alter table with both placement policy and placement options existed.
ErrPlacementPolicyWithDirectOption = dbterror.ClassDDL.NewStd(mysql.ErrPlacementPolicyWithDirectOption)

// ErrMultipleDefConstInListPart returns multiple definition of same constant in list partitioning.
ErrMultipleDefConstInListPart = dbterror.ClassDDL.NewStd(mysql.ErrMultipleDefConstInListPart)

Expand Down
17 changes: 8 additions & 9 deletions ddl/placement_policy.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,10 @@ import (
"github.com/pingcap/tidb/ddl/placement"
"github.com/pingcap/tidb/infoschema"
"github.com/pingcap/tidb/meta"
"github.com/pingcap/tidb/util/placementpolicy"
)

func onCreatePlacementPolicy(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error) {
policyInfo := &placementpolicy.PolicyInfo{}
policyInfo := &model.PolicyInfo{}
if err := job.DecodeArgs(policyInfo); err != nil {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
Expand All @@ -37,7 +36,7 @@ func onCreatePlacementPolicy(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64
if err != nil {
return ver, errors.Trace(err)
}
err = checkPolicyValidation(policyInfo)
err = checkPolicyValidation(policyInfo.PlacementSettings)
if err != nil {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
Expand Down Expand Up @@ -65,7 +64,7 @@ func onCreatePlacementPolicy(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64
}
}

func checkPolicyValidation(info *placementpolicy.PolicyInfo) error {
func checkPolicyValidation(info *model.PlacementSettings) error {
checkMergeConstraint := func(replica uint64, constr1, constr2 string) error {
// Constr2 only make sense when replica is set (whether it is in the replica field or included in the constr1)
if replica == 0 && constr1 == "" {
Expand All @@ -92,7 +91,7 @@ func checkPolicyValidation(info *placementpolicy.PolicyInfo) error {
return nil
}

func getPolicyInfo(t *meta.Meta, policyID int64) (*placementpolicy.PolicyInfo, error) {
func getPolicyInfo(t *meta.Meta, policyID int64) (*model.PolicyInfo, error) {
policy, err := t.GetPolicy(policyID)
if err != nil {
if meta.ErrPolicyNotExists.Equal(err) {
Expand All @@ -105,7 +104,7 @@ func getPolicyInfo(t *meta.Meta, policyID int64) (*placementpolicy.PolicyInfo, e
return policy, nil
}

func checkPlacementPolicyNotExistAndCancelExistJob(d *ddlCtx, t *meta.Meta, job *model.Job, info *placementpolicy.PolicyInfo) error {
func checkPlacementPolicyNotExistAndCancelExistJob(d *ddlCtx, t *meta.Meta, job *model.Job, info *model.PolicyInfo) error {
currVer, err := t.GetSchemaVersion()
if err != nil {
return err
Expand Down Expand Up @@ -134,7 +133,7 @@ func checkPlacementPolicyNotExistAndCancelExistJob(d *ddlCtx, t *meta.Meta, job
return nil
}

func checkPlacementPolicyExistAndCancelNonExistJob(t *meta.Meta, job *model.Job, policyID int64) (*placementpolicy.PolicyInfo, error) {
func checkPlacementPolicyExistAndCancelNonExistJob(t *meta.Meta, job *model.Job, policyID int64) (*model.PolicyInfo, error) {
policy, err := getPolicyInfo(t, policyID)
if err == nil {
return policy, nil
Expand Down Expand Up @@ -199,7 +198,7 @@ func onDropPlacementPolicy(t *meta.Meta, job *model.Job) (ver int64, _ error) {
}

func onAlterPlacementPolicy(t *meta.Meta, job *model.Job) (ver int64, _ error) {
alterPolicy := &placementpolicy.PolicyInfo{}
alterPolicy := &model.PolicyInfo{}
if err := job.DecodeArgs(alterPolicy); err != nil {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
Expand All @@ -213,7 +212,7 @@ func onAlterPlacementPolicy(t *meta.Meta, job *model.Job) (ver int64, _ error) {
newPolicyInfo := *oldPolicy
newPolicyInfo.PlacementSettings = alterPolicy.PlacementSettings

err = checkPolicyValidation(&newPolicyInfo)
err = checkPolicyValidation(newPolicyInfo.PlacementSettings)
if err != nil {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
Expand Down
119 changes: 114 additions & 5 deletions ddl/placement_policy_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import (
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/meta"
"github.com/pingcap/tidb/sessionctx"
"github.com/pingcap/tidb/util/placementpolicy"
"github.com/pingcap/tidb/util/testkit"
)

Expand Down Expand Up @@ -60,7 +59,7 @@ func (s *testDBSuite6) TestPlacementPolicy(c *C) {
"VOTERS=3 " +
"VOTER_CONSTRAINTS=\"[+disk=ssd]\"")

checkFunc := func(policyInfo *placementpolicy.PolicyInfo) {
checkFunc := func(policyInfo *model.PolicyInfo) {
c.Assert(policyInfo.ID != 0, Equals, true)
c.Assert(policyInfo.Name.L, Equals, "x")
c.Assert(policyInfo.PrimaryRegion, Equals, "cn-east-1")
Expand Down Expand Up @@ -108,9 +107,9 @@ func (s *testDBSuite6) TestPlacementPolicy(c *C) {
// TODO: privilege check & constraint syntax check.
}

func testGetPolicyByIDFromMeta(c *C, store kv.Storage, policyID int64) *placementpolicy.PolicyInfo {
func testGetPolicyByIDFromMeta(c *C, store kv.Storage, policyID int64) *model.PolicyInfo {
var (
policyInfo *placementpolicy.PolicyInfo
policyInfo *model.PolicyInfo
err error
)
err1 := kv.RunInNewTxn(context.Background(), store, false, func(ctx context.Context, txn kv.Transaction) error {
Expand All @@ -126,7 +125,7 @@ func testGetPolicyByIDFromMeta(c *C, store kv.Storage, policyID int64) *placemen
return policyInfo
}

func testGetPolicyByNameFromIS(c *C, ctx sessionctx.Context, policy string) *placementpolicy.PolicyInfo {
func testGetPolicyByNameFromIS(c *C, ctx sessionctx.Context, policy string) *model.PolicyInfo {
dom := domain.GetDomain(ctx)
// Make sure the table schema is the new schema.
err := dom.Reload()
Expand Down Expand Up @@ -266,3 +265,113 @@ func (s *testDBSuite6) TestAlterPlacementPolicy(c *C) {
tk.MustGetErrCode("alter placement policy x REGIONS=\"bj,sh\"", mysql.ErrPlacementPolicyNotExists)
tk.MustGetErrCode("alter placement policy x2 REGIONS=\"bj,sh\"", mysql.ErrPlacementPolicyNotExists)
}

func (s *testDBSuite6) TestCreateTableWithPlacementPolicy(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t")

// Direct placement option: special constraints may be incompatible with common constraint.
_, err := tk.Exec("create table t(a int) " +
"PRIMARY_REGION=\"cn-east-1\" " +
"REGIONS=\"cn-east-1, cn-east-2\" " +
"FOLLOWERS=2 " +
"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
"CONSTRAINTS=\"[+disk=ssd,-zone=cn-east-1]\"")
c.Assert(err, NotNil)
c.Assert(err.Error(), Equals, "conflicting label constraints: '-zone=cn-east-1' and '+zone=cn-east-1'")

tk.MustExec("create table t(a int) " +
"PRIMARY_REGION=\"cn-east-1\" " +
"REGIONS=\"cn-east-1, cn-east-2\" " +
"FOLLOWERS=2 " +
"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
"CONSTRAINTS=\"[+disk=ssd]\"")

tbl := testGetTableByName(c, tk.Se, "test", "t")
c.Assert(tbl, NotNil)
c.Assert(tbl.Meta().PlacementPolicyRef, IsNil)
c.Assert(tbl.Meta().DirectPlacementOpts, NotNil)

checkFunc := func(policySetting *model.PlacementSettings) {
c.Assert(policySetting.PrimaryRegion, Equals, "cn-east-1")
c.Assert(policySetting.Regions, Equals, "cn-east-1, cn-east-2")
c.Assert(policySetting.Followers, Equals, uint64(2))
c.Assert(policySetting.FollowerConstraints, Equals, "[+zone=cn-east-1]")
c.Assert(policySetting.Voters, Equals, uint64(0))
c.Assert(policySetting.VoterConstraints, Equals, "")
c.Assert(policySetting.Learners, Equals, uint64(0))
c.Assert(policySetting.LearnerConstraints, Equals, "")
c.Assert(policySetting.Constraints, Equals, "[+disk=ssd]")
c.Assert(policySetting.Schedule, Equals, "")
}
checkFunc(tbl.Meta().DirectPlacementOpts)
tk.MustExec("drop table if exists t")

// Direct placement option and placement policy can't co-exist.
_, err = tk.Exec("create table t(a int) " +
"PRIMARY_REGION=\"cn-east-1\" " +
"REGIONS=\"cn-east-1, cn-east-2\" " +
"FOLLOWERS=2 " +
"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
"CONSTRAINTS=\"[+disk=ssd]\" " +
"PLACEMENT POLICY=\"x\"")
c.Assert(err, NotNil)
c.Assert(err.Error(), Equals, "[ddl:8240]Placement policy 'x' can't co-exist with direct placement options")

// Only placement policy should check the policy existence.
tk.MustGetErrCode("create table t(a int)"+
"PLACEMENT POLICY=\"x\"", mysql.ErrPlacementPolicyNotExists)
tk.MustExec("create placement policy x " +
"PRIMARY_REGION=\"cn-east-1\" " +
"REGIONS=\"cn-east-1, cn-east-2\" " +
"FOLLOWERS=2 " +
"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
"CONSTRAINTS=\"[+disk=ssd]\" ")
tk.MustExec("create table t(a int)" +
"PLACEMENT POLICY=\"x\"")

tbl = testGetTableByName(c, tk.Se, "test", "t")
c.Assert(tbl, NotNil)
c.Assert(tbl.Meta().PlacementPolicyRef, NotNil)
c.Assert(tbl.Meta().PlacementPolicyRef.Name.L, Equals, "x")
c.Assert(tbl.Meta().PlacementPolicyRef.ID != 0, Equals, true)
tk.MustExec("drop table if exists t")

// Only direct placement options should check the compatibility itself.
_, err = tk.Exec("create table t(a int)" +
"PRIMARY_REGION=\"cn-east-1\" " +
"REGIONS=\"cn-east-1, cn-east-2\" " +
"FOLLOWERS=2 " +
"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
"CONSTRAINTS=\"[+disk=ssd, -zone=cn-east-1]\" ")
c.Assert(err, NotNil)
c.Assert(err.Error(), Equals, "conflicting label constraints: '-zone=cn-east-1' and '+zone=cn-east-1'")

tk.MustExec("create table t(a int)" +
"PRIMARY_REGION=\"cn-east-1\" " +
"REGIONS=\"cn-east-1, cn-east-2\" " +
"FOLLOWERS=2 " +
"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
"CONSTRAINTS=\"[+disk=ssd]\" ")

tbl = testGetTableByName(c, tk.Se, "test", "t")
c.Assert(tbl, NotNil)
c.Assert(tbl.Meta().DirectPlacementOpts, NotNil)

checkFunc = func(policySetting *model.PlacementSettings) {
c.Assert(policySetting.PrimaryRegion, Equals, "cn-east-1")
c.Assert(policySetting.Regions, Equals, "cn-east-1, cn-east-2")
c.Assert(policySetting.Followers, Equals, uint64(2))
c.Assert(policySetting.FollowerConstraints, Equals, "[+zone=cn-east-1]")
c.Assert(policySetting.Voters, Equals, uint64(0))
c.Assert(policySetting.VoterConstraints, Equals, "")
c.Assert(policySetting.Learners, Equals, uint64(0))
c.Assert(policySetting.LearnerConstraints, Equals, "")
c.Assert(policySetting.Constraints, Equals, "[+disk=ssd]")
c.Assert(policySetting.Schedule, Equals, "")
}
checkFunc(tbl.Meta().DirectPlacementOpts)
tk.MustExec("drop table if exists t")
tk.MustExec("drop placement policy if exists x")
}
17 changes: 17 additions & 0 deletions ddl/table.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,23 @@ func onCreateTable(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
}
return ver, errors.Trace(err)
}
// Can not use both a placement policy and direct assignment. If you alter specify both in a CREATE TABLE or ALTER TABLE an error will be returned.
if tbInfo.DirectPlacementOpts != nil && tbInfo.PlacementPolicyRef != nil {
return ver, errors.Trace(ErrPlacementPolicyWithDirectOption.GenWithStackByArgs(tbInfo.PlacementPolicyRef.Name))
}
if tbInfo.DirectPlacementOpts != nil {
// check the direct placement option compatibility.
if err := checkPolicyValidation(tbInfo.DirectPlacementOpts); err != nil {
return ver, errors.Trace(err)
}
}
if tbInfo.PlacementPolicyRef != nil {
// placement policy reference will override the direct placement options.
_, err = checkPlacementPolicyExistAndCancelNonExistJob(t, job, tbInfo.PlacementPolicyRef.ID)
if err != nil {
return ver, errors.Trace(infoschema.ErrPlacementPolicyNotExists.GenWithStackByArgs(tbInfo.PlacementPolicyRef.Name))
}
}

ver, err = updateSchemaVersion(t, job)
if err != nil {
Expand Down
1 change: 1 addition & 0 deletions errno/errcode.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1050,6 +1050,7 @@ const (
ErrInvalidAttributesSpec = 8237
ErrPlacementPolicyExists = 8238
ErrPlacementPolicyNotExists = 8239
ErrPlacementPolicyWithDirectOption = 8240

// TiKV/PD/TiFlash errors.
ErrPDServerTimeout = 9001
Expand Down
Loading

0 comments on commit 6aec001

Please sign in to comment.