executor, ddl: support database placement option and partition placement option

ddl/ddl_api.go

@@ -2723,6 +2723,8 @@ func (d *ddl) AlterTable(ctx context.Context, sctx sessionctx.Context, ident ast
 			err = d.AlterTableAttributes(sctx, ident, spec)
 		case ast.AlterTablePartitionAttributes:
 			err = d.AlterTablePartitionAttributes(sctx, ident, spec)
+		case ast.AlterTablePartitionOptions:
+			err = d.AlterTablePartitionOptions(sctx, ident, spec)
 		default:
 			// Nothing to do now.
 		}
@@ -6318,6 +6320,123 @@ func (d *ddl) AlterTablePartitionAttributes(ctx sessionctx.Context, ident ast.Id
 	return errors.Trace(err)
 }
 
+func (d *ddl) AlterTablePartitionOptions(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) (err error) {
+	schema, tb, err := d.getSchemaAndTableByIdent(ctx, ident)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	meta := tb.Meta()
+	if meta.Partition == nil {
+		return errors.Trace(ErrPartitionMgmtOnNonpartitioned)
+	}
+
+	partitionID, err := tables.FindPartitionByName(meta, spec.PartitionNames[0].L)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	var policyRefInfo *model.PolicyRefInfo
+	var placementSettings *model.PlacementSettings
+	if spec.Options != nil {
+		for _, op := range spec.Options {
+			switch op.Tp {
+			case ast.TableOptionPlacementPolicy:
+				policyRefInfo = &model.PolicyRefInfo{
+					Name: model.NewCIStr(op.StrValue),
+				}
+			case ast.TableOptionPlacementPrimaryRegion:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.PrimaryRegion = op.StrValue
+			case ast.TableOptionPlacementRegions:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.Regions = op.StrValue
+			case ast.TableOptionPlacementFollowerCount:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.Followers = op.UintValue
+			case ast.TableOptionPlacementVoterCount:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.Voters = op.UintValue
+			case ast.TableOptionPlacementLearnerCount:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.Learners = op.UintValue
+			case ast.TableOptionPlacementSchedule:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.Schedule = op.StrValue
+			case ast.TableOptionPlacementConstraints:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.Constraints = op.StrValue
+			case ast.TableOptionPlacementLeaderConstraints:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.LeaderConstraints = op.StrValue
+			case ast.TableOptionPlacementLearnerConstraints:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.LearnerConstraints = op.StrValue
+			case ast.TableOptionPlacementFollowerConstraints:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.FollowerConstraints = op.StrValue
+			case ast.TableOptionPlacementVoterConstraints:
+				if placementSettings == nil {
+					placementSettings = &model.PlacementSettings{}
+				}
+				placementSettings.VoterConstraints = op.StrValue
+			default:
+				return errors.Trace(errors.New("unknown partition option"))
+			}
+		}
+	}
+
+	// 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 placementSettings != nil && policyRefInfo != nil {
+		return errors.Trace(ErrPlacementPolicyWithDirectOption.GenWithStackByArgs(policyRefInfo.Name))
+	}
+	if placementSettings != nil {
+		// check the direct placement option compatibility.
+		if err := checkPolicyValidation(placementSettings); err != nil {
+			return errors.Trace(err)
+		}
+	}
+	if policyRefInfo != nil {
+		policy, ok := ctx.GetInfoSchema().(infoschema.InfoSchema).PolicyByName(policyRefInfo.Name)
+		if !ok {
+			return errors.Trace(infoschema.ErrPlacementPolicyNotExists.GenWithStackByArgs(policyRefInfo.Name))
+		}
+		policyRefInfo.ID = policy.ID
+	}
+
+	job := &model.Job{
+		SchemaID:   schema.ID,
+		TableID:    meta.ID,
+		SchemaName: schema.Name.L,
+		Type:       model.ActionAlterTablePartitionPolicy,
+		BinlogInfo: &model.HistoryInfo{},
+		Args:       []interface{}{partitionID, policyRefInfo, placementSettings},
+	}
+
+	err = d.doDDLJob(ctx, job)
+	err = d.callHookOnChanged(err)
+	return errors.Trace(err)
+}
+
 func buildPolicyInfo(name model.CIStr, options []*ast.PlacementOption) (*model.PolicyInfo, error) {
 	policyInfo := &model.PolicyInfo{PlacementSettings: &model.PlacementSettings{}}
 	policyInfo.Name = name

ddl/ddl_worker.go

@@ -832,6 +832,8 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 		ver, err = onDropPlacementPolicy(d, t, job)
 	case model.ActionAlterPlacementPolicy:
 		ver, err = onAlterPlacementPolicy(d, t, job)
+	case model.ActionAlterTablePartitionPolicy:
+		ver, err = onAlterTablePartitionOptions(t, job)
 	default:
 		// Invalid job, cancel it.
 		job.State = model.JobStateCancelled

ddl/placement_policy_test.go

@@ -534,3 +534,124 @@ func (s *testDBSuite6) TestPolicyCacheAndPolicyDependencyCache(c *C) {
 	c.Assert(dependencies, NotNil)
 	c.Assert(len(dependencies), Equals, 0)
 }
+
+func (s *testDBSuite6) TestAlterTablePartitionWithPlacementPolicy(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	defer func() {
+		tk.MustExec("drop table if exists t1")
+		tk.MustExec("drop placement policy if exists x")
+	}()
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1")
+	// Direct placement option: special constraints may be incompatible with common constraint.
+	tk.MustExec("create table t1 (c int) PARTITION BY RANGE (c) " +
+		"(PARTITION p0 VALUES LESS THAN (6)," +
+		"PARTITION p1 VALUES LESS THAN (11)," +
+		"PARTITION p2 VALUES LESS THAN (16)," +
+		"PARTITION p3 VALUES LESS THAN (21));")
+
+	tk.MustExec("alter table t1 partition p0 " +
+		"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", "t1")
+	c.Assert(tbl, NotNil)
+	ptDef := testGetPartitionDefinitionsByName(c, tk.Se, "test", "t1", "p0")
+	c.Assert(ptDef.PlacementPolicyRef.Name.L, Equals, "")
+	c.Assert(ptDef.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(ptDef.DirectPlacementOpts)
+
+	//Direct placement option and placement policy can't co-exist.
+	_, err := tk.Exec("alter table t1 partition p0 " +
+		"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("alter table t1 partition p0 "+
+		"PLACEMENT POLICY=\"x\"", mysql.ErrPlacementPolicyNotExists)
+	tk.MustExec("create placement policy x " +
+		"FOLLOWERS=2 ")
+	tk.MustExec("alter table t1 partition p0 " +
+		"PLACEMENT POLICY=\"x\"")
+
+	ptDef = testGetPartitionDefinitionsByName(c, tk.Se, "test", "t1", "p0")
+	c.Assert(ptDef, NotNil)
+	c.Assert(ptDef.PlacementPolicyRef, NotNil)
+	c.Assert(ptDef.PlacementPolicyRef.Name.L, Equals, "x")
+	c.Assert(ptDef.PlacementPolicyRef.ID != 0, Equals, true)
+
+	// Only direct placement options should check the compatibility itself.
+	_, err = tk.Exec("alter table t1 partition p0 " +
+		"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("alter table t1 partition p0 " +
+		"PRIMARY_REGION=\"cn-east-1\" " +
+		"REGIONS=\"cn-east-1, cn-east-2\" " +
+		"FOLLOWERS=2 " +
+		"FOLLOWER_CONSTRAINTS=\"[+zone=cn-east-1]\" " +
+		"CONSTRAINTS=\"[+disk=ssd]\" ")
+
+	ptDef = testGetPartitionDefinitionsByName(c, tk.Se, "test", "t1", "p0")
+	c.Assert(ptDef, NotNil)
+	c.Assert(ptDef.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(ptDef.DirectPlacementOpts)
+}
+
+func testGetPartitionDefinitionsByName(c *C, ctx sessionctx.Context, db string, table string, ptName string) model.PartitionDefinition {
+	dom := domain.GetDomain(ctx)
+	// Make sure the table schema is the new schema.
+	err := dom.Reload()
+	c.Assert(err, IsNil)
+	tbl, err := dom.InfoSchema().TableByName(model.NewCIStr(db), model.NewCIStr(table))
+	c.Assert(err, IsNil)
+	c.Assert(tbl, NotNil)
+	var ptDef model.PartitionDefinition
+	for _, def := range tbl.Meta().Partition.Definitions {
+		if ptName == def.Name.L {
+			ptDef = def
+			break
+		}
+	}
+	return ptDef
+}

ddl/table.go

@@ -1292,6 +1292,43 @@ func onAlterTablePartitionAttributes(t *meta.Meta, job *model.Job) (ver int64, e
 	return ver, nil
 }
 
+func onAlterTablePartitionOptions(t *meta.Meta, job *model.Job) (ver int64, err error) {
+	var partitionID int64
+	policyRefInfo := &model.PolicyRefInfo{}
+	placementSettings := &model.PlacementSettings{}
+	err = job.DecodeArgs(&partitionID, policyRefInfo, placementSettings)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return 0, errors.Trace(err)
+	}
+	tblInfo, err := getTableInfoAndCancelFaultJob(t, job, job.SchemaID)
+	if err != nil {
+		return 0, err
+	}
+
+	ptInfo := tblInfo.GetPartitionInfo()
+	if ptInfo.GetNameByID(partitionID) == "" {
+		job.State = model.JobStateCancelled
+		return 0, errors.Trace(table.ErrUnknownPartition.GenWithStackByArgs("drop?", tblInfo.Name.O))
+	}
+	for idx, ptDef := range ptInfo.Definitions {
+		if ptDef.ID == partitionID {
+			ptDef.DirectPlacementOpts = placementSettings
+			ptDef.PlacementPolicyRef = policyRefInfo
+			ptInfo.Definitions[idx] = ptDef
+			break
+		}
+	}
+
+	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+	job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tblInfo)
+
+	return ver, nil
+}
+
 func getOldLabelRules(tblInfo *model.TableInfo, oldSchemaName, oldTableName string) (string, []string, []string, map[string]*label.Rule, error) {
 	tableRuleID := fmt.Sprintf(label.TableIDFormat, label.IDPrefix, oldSchemaName, oldTableName)
 	oldRuleIDs := []string{tableRuleID}

go.mod

@@ -49,7 +49,7 @@ require (
 	github.com/pingcap/fn v0.0.0-20200306044125-d5540d389059
 	github.com/pingcap/kvproto v0.0.0-20210806074406-317f69fb54b4
 	github.com/pingcap/log v0.0.0-20210906054005-afc726e70354
-	github.com/pingcap/parser v0.0.0-20210917114242-ac711116bdff
+	github.com/pingcap/parser v0.0.0-20210923132047-19e7f91ed500
 	github.com/pingcap/sysutil v0.0.0-20210730114356-fcd8a63f68c5
 	github.com/pingcap/tidb-tools v5.0.3+incompatible
 	github.com/pingcap/tipb v0.0.0-20210802080519-94b831c6db55

go.sum

@@ -601,8 +601,8 @@ github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuR
 github.com/pingcap/log v0.0.0-20210906054005-afc726e70354 h1:SvWCbCPh1YeHd9yQLksvJYAgft6wLTY1aNG81tpyscQ=
 github.com/pingcap/log v0.0.0-20210906054005-afc726e70354/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/parser v0.0.0-20210525032559-c37778aff307/go.mod h1:xZC8I7bug4GJ5KtHhgAikjTfU4kBv1Sbo3Pf1MZ6lVw=
-github.com/pingcap/parser v0.0.0-20210917114242-ac711116bdff h1:LiwvvutmyeSkFkdVM09mH6KK+OeDVJzX7WKy9Lf0ri0=
-github.com/pingcap/parser v0.0.0-20210917114242-ac711116bdff/go.mod h1:+xcMiiZzdIktT/Nqdfm81dkECJ2EPuoAYywd57py4Pk=
+github.com/pingcap/parser v0.0.0-20210923132047-19e7f91ed500 h1:xe2A/2r15+ahyAVrlindm08M3guhQdv++Mmu29nNXoY=
+github.com/pingcap/parser v0.0.0-20210923132047-19e7f91ed500/go.mod h1:+xcMiiZzdIktT/Nqdfm81dkECJ2EPuoAYywd57py4Pk=
 github.com/pingcap/sysutil v0.0.0-20200206130906-2bfa6dc40bcd/go.mod h1:EB/852NMQ+aRKioCpToQ94Wl7fktV+FNnxf3CX/TTXI=
 github.com/pingcap/sysutil v0.0.0-20210315073920-cc0985d983a3/go.mod h1:tckvA041UWP+NqYzrJ3fMgC/Hw9wnmQ/tUkp/JaHly8=
 github.com/pingcap/sysutil v0.0.0-20210730114356-fcd8a63f68c5 h1:7rvAtZe/ZUzOKzgriNPQoBNvleJXBk4z7L3Z47+tS98=