ddl: Support exchange partition

ddl/db_partition_test.go

@@ -761,6 +761,134 @@ func (s *testIntegrationSuite5) TestAlterTableDropPartition(c *C) {
 	tk.MustGetErrCode("alter table t1 drop partition p2", tmysql.ErrOnlyOnRangeListPartition)
 }
 
+func (s *testIntegrationSuite5) TestAlterTableExchangePartition(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists e")
+	tk.MustExec("drop table if exists e2")
+	tk.MustExec(`CREATE TABLE e (
+		id INT NOT NULL
+	)
+    PARTITION BY RANGE (id) (
+        PARTITION p0 VALUES LESS THAN (50),
+        PARTITION p1 VALUES LESS THAN (100),
+        PARTITION p2 VALUES LESS THAN (150),
+        PARTITION p3 VALUES LESS THAN (MAXVALUE)
+	);`)
+	tk.MustExec(`CREATE TABLE e2 (
+		id INT NOT NULL
+	);`)
+	tk.MustExec(`INSERT INTO e VALUES (1669),(337),(16),(2005)`)
+	tk.MustExec("ALTER TABLE e EXCHANGE PARTITION p0 WITH TABLE e2")
+	tk.MustQuery("select * from e2").Check(testkit.Rows("16"))
+	tk.MustQuery("select * from e").Check(testkit.Rows("1669", "337", "2005"))
+	tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p1 WITH TABLE e2", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p2 WITH TABLE e2", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p3 WITH TABLE e2", tmysql.ErrRowDoesNotMatchPartition)
+
+	tk.MustExec("drop table if exists e3")
+
+	tk.MustExec(`CREATE TABLE e3 (
+		id int not null
+	) PARTITION BY HASH (id)
+	PARTITIONS 4;`)
+	tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p1 WITH TABLE e3;", tmysql.ErrPartitionExchangePartTable)
+	tk.MustExec("truncate table e2")
+	tk.MustExec(`INSERT INTO e3 VALUES (1),(5)`)
+
+	tk.MustExec("ALTER TABLE e3 EXCHANGE PARTITION p1 WITH TABLE e2;")
+	tk.MustGetErrCode("ALTER TABLE e3 EXCHANGE PARTITION p0 WITH TABLE e2", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e3 EXCHANGE PARTITION p2 WITH TABLE e2", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e3 EXCHANGE PARTITION p3 WITH TABLE e2", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustExec("ALTER TABLE e3 EXCHANGE PARTITION p0 with TABLE e2 WITHOUT VALIDATION")
+
+	tk.MustQuery("select * from e3 partition(p0)").Check(testkit.Rows("1", "5"))
+	tk.MustQuery("select * from e2").Check(testkit.Rows())
+
+}
+
+func (s *testIntegrationSuite4) TestExchangePartitionTableCompatiable(c *C) {
+
+	type testCase struct {
+		ptSQL       string
+		ntSQL       string
+		exchangeSQL string
+		err         bool
+	}
+	cases := []testCase{
+		{
+			"create table pt (id int not null) partition by hash (id) partitions 4;",
+			"create table nt (id int(1) not null);",
+			"alter table pt exchange partition p0 with table nt;",
+			false,
+		},
+		{
+			"create table pt1 (id int not null, fname varchar(3)) partition by hash (id) partitions 4;",
+			"create table nt1 (id int not null, fname varchar(4));",
+			"alter table pt1 exchange partition p0 with table nt1;",
+			true,
+		},
+		{
+			"create table pt2 (id int not null, salary decimal) partition by hash(id) partitions 4;",
+			"create table nt2 (id int not null, salary decimal(3,2));",
+			"alter table pt2 exchange partition p0 with table nt2;",
+			true,
+		},
+		{
+			"create table pt3 (id int not null, salary decimal) partition by hash(id) partitions 1;",
+			"create table nt3 (id int not null, salary decimal(10, 1));",
+			"alter table pt3 exchange partition p0 with table nt3",
+			true,
+		},
+		{
+			"create table pt4 (id int not null) partition by hash(id) partitions 1;",
+			"create table nt4 (id1 int not null);",
+			"alter table pt4 exchange partition p0 with table nt4;",
+			true,
+		},
+		{
+			"create table pt5 (id int not null, primary key (id)) partition by hash(id) partitions 1;",
+			"create table nt5 (id int not null);",
+			"alter table pt5 exchange partition p0 with table nt5;",
+			true,
+		},
+		{
+			"create table pt6 (id int not null, salary decimal, index idx (id, salary)) partition by hash(id) partitions 1;",
+			"create table nt6 (id int not null, salary decimal, index idx (salary, id));",
+			"alter table pt6 exchange partition p0 with table nt6;",
+			true,
+		},
+		{
+			"create table pt7 (id int not null, index idx (id) invisible) partition by hash(id) partitions 1;",
+			"create table nt7 (id int not null, index idx (id));",
+			"alter table pt7 exchange partition p0 with table nt7;",
+			false,
+		},
+		{
+			"create table pt8 (id int not null, index idx (id)) partition by hash(id) partitions 1;",
+			"create table nt8 (id int not null, index id_idx (id));",
+			"alter table pt8 exchange partition p0 with table nt8;",
+			true,
+		},
+	}
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	for i, t := range cases {
+		tk.MustExec(t.ptSQL)
+		tk.MustExec(t.ntSQL)
+		if t.err {
+			_, err := tk.Exec(t.exchangeSQL)
+			c.Assert(ddl.ErrTablesDifferentMetadata.Equal(err), IsTrue, Commentf(
+				"case %d fail, sql = `%s`\nexpected error = `%v`\n  actual error = `%v`",
+				i, t.exchangeSQL, ddl.ErrTablesDifferentMetadata, err,
+			))
+		} else {
+			tk.MustExec(t.exchangeSQL)
+		}
+	}
+
+}
+
 func (s *testIntegrationSuite4) TestAddPartitionTooManyPartitions(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")

ddl/ddl_algorithm_test.go

@@ -60,6 +60,7 @@ func (s *testDDLAlgorithmSuite) TestFindAlterAlgorithm(c *C) {
 		{ast.AlterTableSpec{Tp: ast.AlterTableAddPartitions}, instantAlgorithm, ast.AlgorithmTypeInstant},
 		{ast.AlterTableSpec{Tp: ast.AlterTableDropPartition}, instantAlgorithm, ast.AlgorithmTypeInstant},
 		{ast.AlterTableSpec{Tp: ast.AlterTableTruncatePartition}, instantAlgorithm, ast.AlgorithmTypeInstant},
+		{ast.AlterTableSpec{Tp: ast.AlterTableExchangePartition}, instantAlgorithm, ast.AlgorithmTypeInstant},
 
 		// TODO: after we support lock a table, change the below case.
 		{ast.AlterTableSpec{Tp: ast.AlterTableLock}, instantAlgorithm, ast.AlgorithmTypeInstant},

ddl/ddl_api.go

@@ -2168,6 +2168,8 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 			err = d.DropTablePartition(ctx, ident, spec)
 		case ast.AlterTableTruncatePartition:
 			err = d.TruncateTablePartition(ctx, ident, spec)
+		case ast.AlterTableExchangePartition:
+			err = d.ExchangeTableParition(ctx, ident, spec)
 		case ast.AlterTableAddConstraint:
 			constr := spec.Constraint
 			switch spec.Constraint.Tp {
@@ -2738,6 +2740,142 @@ func (d *ddl) DropTablePartition(ctx sessionctx.Context, ident ast.Ident, spec *
 	return errors.Trace(err)
 }
 
+func checkFielTypeCompatible(ft *types.FieldType, other *types.FieldType) bool {
+	// int(1) could match the type with int(8)
+	partialEqual := ft.Tp == other.Tp &&
+		ft.Decimal == other.Decimal &&
+		ft.Charset == other.Charset &&
+		ft.Collate == other.Collate &&
+		(ft.Flen == other.Flen || ft.StorageLength() != types.VarStorageLen) &&
+		mysql.HasUnsignedFlag(ft.Flag) == mysql.HasUnsignedFlag(other.Flag)
+	if !partialEqual || len(ft.Elems) != len(other.Elems) {
+		return false
+	}
+	for i := range ft.Elems {
+		if ft.Elems[i] != other.Elems[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func checkTableDefCompatible(source *model.TableInfo, target *model.TableInfo) error {
+	err := ErrTablesDifferentMetadata
+	if len(source.Cols()) != len(target.Cols()) {
+		return err
+	}
+	// Col compatible check
+	for i, sourceCol := range source.Cols() {
+		targetCol := target.Cols()[i]
+		if sourceCol.IsGenerated() != targetCol.IsGenerated() ||
+			sourceCol.Name.L != targetCol.Name.L ||
+			!checkFielTypeCompatible(&sourceCol.FieldType, &targetCol.FieldType) {
+			return err
+		}
+	}
+	if len(source.Indices) != len(target.Indices) {
+		return err
+	}
+	for _, sourceIdx := range source.Indices {
+		var compatIdx *model.IndexInfo
+		for _, targetIdx := range target.Indices {
+			if strings.EqualFold(sourceIdx.Name.O, targetIdx.Name.O) {
+				compatIdx = targetIdx
+			}
+		}
+		// No match index
+		if compatIdx == nil {
+			return err
+		}
+		// Index type is not compatiable
+		if sourceIdx.Tp != compatIdx.Tp ||
+			sourceIdx.Unique != compatIdx.Unique ||
+			sourceIdx.Primary != compatIdx.Primary {
+			return err
+		}
+		// The index column
+		if len(sourceIdx.Columns) != len(compatIdx.Columns) {
+			return err
+		}
+		for i, sourceIdxCol := range sourceIdx.Columns {
+			compatIdxCol := compatIdx.Columns[i]
+			if sourceIdxCol.Name.L != compatIdxCol.Name.L ||
+				sourceIdxCol.Length != compatIdxCol.Length {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (d *ddl) ExchangeTableParition(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
+	is := d.infoHandle.Get()
+	ptSchema, ok := is.SchemaByName(ident.Schema)
+	if !ok {
+		return errors.Trace(infoschema.ErrDatabaseNotExists.GenWithStackByArgs(ptSchema))
+	}
+	pt, err := is.TableByName(ident.Schema, ident.Name)
+	if err != nil {
+		return errors.Trace(infoschema.ErrTableNotExists.GenWithStackByArgs(ident.Schema, ident.Name))
+	}
+
+	ptMeta := pt.Meta()
+
+	ntIdent := ast.Ident{Schema: spec.NewTable.Schema, Name: spec.NewTable.Name}
+
+	ntSchema, ok := is.SchemaByName(ntIdent.Schema)
+	if !ok {
+		return errors.Trace(infoschema.ErrDatabaseNotExists.GenWithStackByArgs(ntSchema))
+	}
+
+	nt, err := is.TableByName(ntIdent.Schema, ntIdent.Name)
+	if err != nil {
+		return errors.Trace(infoschema.ErrTableNotExists.GenWithStackByArgs(ntIdent.Schema, ntIdent.Name))
+	}
+
+	ntMeta := nt.Meta()
+
+	if ptMeta.GetPartitionInfo() == nil {
+		return errors.Trace(ErrPartitionMgmtOnNonpartitioned)
+	}
+	if ntMeta.GetPartitionInfo() != nil {
+		return errors.Trace(ErrPartitionExchangePartTable.GenWithStackByArgs(ntIdent.Name))
+	}
+
+	// NOTE: if nt is temporary table, it should be checked
+
+	partName := spec.PartitionNames[0].L
+
+	_, err = tables.FindPartitionByName(ptMeta, partName)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	// NOTE: if pt is subPartitioned, it should be checked
+
+	err = checkTableDefCompatible(ptMeta, ntMeta)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	job := &model.Job{
+		SchemaID:   ntSchema.ID,
+		TableID:    ntMeta.ID,
+		SchemaName: ntSchema.Name.L,
+		Type:       model.ActionExchangeTablePartition,
+		BinlogInfo: &model.HistoryInfo{},
+		Args:       []interface{}{ptSchema.ID, ptMeta.ID, partName, spec.WithValidation},
+	}
+
+	err = d.doDDLJob(ctx, job)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	err = d.callHookOnChanged(err)
+	return errors.Trace(err)
+}
+
 // DropColumn will drop a column from the table, now we don't support drop the column with index covered.
 func (d *ddl) DropColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTableSpec) error {
 	schema, t, err := d.getSchemaAndTableByIdent(ctx, ti)

ddl/ddl_worker.go

@@ -601,6 +601,8 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 		ver, err = onDropTablePartition(t, job)
 	case model.ActionTruncateTablePartition:
 		ver, err = onTruncateTablePartition(d, t, job)
+	case model.ActionExchangeTablePartition:
+		ver, err = w.onExchangeTablePartition(d, t, job)
 	case model.ActionAddColumn:
 		ver, err = onAddColumn(d, t, job)
 	case model.ActionAddColumns:

ddl/ddl_worker_test.go

@@ -488,6 +488,9 @@ func buildCancelJobTests(firstID int64) []testCancelJob {
 
 		{act: model.ActionAlterIndexVisibility, jobIDs: []int64{firstID + 46}, cancelRetErrs: noErrs, cancelState: model.StateNone},
 		{act: model.ActionAlterIndexVisibility, jobIDs: []int64{firstID + 47}, cancelRetErrs: []error{admin.ErrCancelFinishedDDLJob.GenWithStackByArgs(firstID + 47)}, cancelState: model.StatePublic},
+
+		{act: model.ActionExchangeTablePartition, jobIDs: []int64{firstID + 53}, cancelRetErrs: noErrs, cancelState: model.StateNone},
+		{act: model.ActionExchangeTablePartition, jobIDs: []int64{firstID + 54}, cancelRetErrs: []error{admin.ErrCancelFinishedDDLJob.GenWithStackByArgs(firstID + 54)}, cancelState: model.StatePublic},
 	}
 
 	return tests
@@ -986,6 +989,31 @@ func (s *testDDLSuite) TestCancelJob(c *C) {
 	c.Check(checkErr, IsNil)
 	changedTable = testGetTable(c, d, dbInfo.ID, tblInfo.ID)
 	c.Assert(checkIdxVisibility(changedTable, indexName, true), IsTrue)
+
+	// test exchange partition failed  caused by canceled
+	pt := testTableInfoWithPartition(c, d, "pt", 5)
+	nt := testTableInfo(c, d, "nt", 5)
+	testCreateTable(c, ctx, d, dbInfo, pt)
+	testCreateTable(c, ctx, d, dbInfo, nt)
+
+	updateTest(&tests[43])
+	exchangeTablePartition := []interface{}{dbInfo.ID, pt.ID, "p0", true}
+	doDDLJobErrWithSchemaState(ctx, d, c, dbInfo.ID, nt.ID, test.act, exchangeTablePartition, &test.cancelState)
+	c.Check(checkErr, IsNil)
+	changedNtTable := testGetTable(c, d, dbInfo.ID, nt.ID)
+	changedPtTable := testGetTable(c, d, dbInfo.ID, pt.ID)
+	c.Assert(changedNtTable.Meta().ID == nt.ID, IsTrue)
+	c.Assert(changedPtTable.Meta().Partition.Definitions[0].ID == pt.Partition.Definitions[0].ID, IsTrue)
+
+	// canel exchange partition successfully
+	updateTest(&tests[44])
+	doDDLJobSuccess(ctx, d, c, dbInfo.ID, nt.ID, test.act, exchangeTablePartition)
+	c.Check(checkErr, IsNil)
+	changedNtTable = testGetTable(c, d, dbInfo.ID, pt.Partition.Definitions[0].ID)
+	changedPtTable = testGetTable(c, d, dbInfo.ID, pt.ID)
+	c.Assert(changedNtTable.Meta().ID == nt.ID, IsFalse)
+	c.Assert(changedPtTable.Meta().Partition.Definitions[0].ID == nt.ID, IsTrue)
+
 }
 
 func (s *testDDLSuite) TestIgnorableSpec(c *C) {

ddl/error.go

@@ -202,4 +202,10 @@ var (
 	ErrColumnTypeUnsupportedNextValue = terror.ClassDDL.New(mysql.ErrColumnTypeUnsupportedNextValue, mysql.MySQLErrName[mysql.ErrColumnTypeUnsupportedNextValue])
 	// ErrUnsupportedExpressionIndex is returned when create an expression index without allow-expression-index.
 	ErrUnsupportedExpressionIndex = terror.ClassDDL.New(mysql.ErrUnsupportedDDLOperation, fmt.Sprintf(mysql.MySQLErrName[mysql.ErrUnsupportedDDLOperation], "creating expression index without allow-expression-index in config"))
+	// ErrPartitionExchangePartTable is returned when exchange table partition with another table is partitioned
+	ErrPartitionExchangePartTable = terror.ClassDDL.New(mysql.ErrPartitionExchangePartTable, mysql.MySQLErrName[mysql.ErrPartitionExchangePartTable])
+	// ErrTableDifferentMetadata is returned when exchanges tables is not compatible
+	ErrTablesDifferentMetadata = terror.ClassDDL.New(mysql.ErrTablesDifferentMetadata, mysql.MySQLErrName[mysql.ErrTablesDifferentMetadata])
+	// ErrRowDoesNotMatchPartition is return when the row record of exchange table does not match the partition rule
+	ErrRowDoesNotMatchPartition = terror.ClassDDL.New(mysql.ErrRowDoesNotMatchPartition, mysql.MySQLErrName[mysql.ErrRowDoesNotMatchPartition])
 )