diff --git a/ddl/column.go b/ddl/column.go
index 1cad2362c1187..a38c3afab881a 100644
--- a/ddl/column.go
+++ b/ddl/column.go
@@ -931,3 +931,11 @@ func getColumnInfoByName(tbInfo *model.TableInfo, column string) *model.ColumnIn
 	}
 	return nil
 }
+
+// isVirtualGeneratedColumn checks the column if it is virtual.
+func isVirtualGeneratedColumn(col *model.ColumnInfo) bool {
+	if col.IsGenerated() && !col.GeneratedStored {
+		return true
+	}
+	return false
+}
diff --git a/ddl/db_partition_test.go b/ddl/db_partition_test.go
index 58ead7849c3ad..252d7516d2565 100644
--- a/ddl/db_partition_test.go
+++ b/ddl/db_partition_test.go
@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/terror"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/ddl/testutil"
 	"github.com/pingcap/tidb/domain"
@@ -789,6 +790,385 @@ func (s *testIntegrationSuite5) TestAlterTableDropPartition(c *C) {
 	tk.MustGetErrCode("alter table t1 drop partition p2", tmysql.ErrOnlyOnRangeListPartition)
 }
 
+func (s *testIntegrationSuite4) 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"))
+	// validation test for range partition
+	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.MustQuery("select * from e3 partition(p0)").Check(testkit.Rows())
+	tk.MustQuery("select * from e2").Check(testkit.Rows("1", "5"))
+
+	// validation test for hash partition
+	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)
+
+	// without validation test
+	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())
+
+	// more boundary test of range partition
+	// for partition p0
+	tk.MustExec(`create table e4 (a int) partition by range(a) (
+		partition p0 values less than (3),
+		partition p1 values less than (6),
+        PARTITION p2 VALUES LESS THAN (9),
+        PARTITION p3 VALUES LESS THAN (MAXVALUE)
+		);`)
+	tk.MustExec(`create table e5(a int);`)
+
+	tk.MustExec("insert into e5 values (1)")
+
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p1 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p2 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p3 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustExec("ALTER TABLE e4 EXCHANGE PARTITION p0 with TABLE e5")
+	tk.MustQuery("select * from e4 partition(p0)").Check(testkit.Rows("1"))
+
+	// for partition p1
+	tk.MustExec("insert into e5 values (3)")
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p0 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p2 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p3 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustExec("ALTER TABLE e4 EXCHANGE PARTITION p1 with TABLE e5")
+	tk.MustQuery("select * from e4 partition(p1)").Check(testkit.Rows("3"))
+
+	// for partition p2
+	tk.MustExec("insert into e5 values (6)")
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p0 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p1 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p3 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustExec("ALTER TABLE e4 EXCHANGE PARTITION p2 with TABLE e5")
+	tk.MustQuery("select * from e4 partition(p2)").Check(testkit.Rows("6"))
+
+	// for partition p3
+	tk.MustExec("insert into e5 values (9)")
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p0 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("ALTER TABLE e4 EXCHANGE PARTITION p1 WITH TABLE e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustGetErrCode("alter table e4 exchange partition p2 with table e5", tmysql.ErrRowDoesNotMatchPartition)
+	tk.MustExec("ALTER TABLE e4 EXCHANGE PARTITION p3 with TABLE e5")
+	tk.MustQuery("select * from e4 partition(p3)").Check(testkit.Rows("9"))
+
+	// for columns range partition
+	tk.MustExec(`create table e6 (a varchar(3)) partition by range columns (a) (
+		partition p0 values less than ('3'),
+		partition p1 values less than ('6')
+	);`)
+	tk.MustExec(`create table e7 (a varchar(3));`)
+	tk.MustExec(`insert into e6 values ('1');`)
+	tk.MustExec(`insert into e7 values ('2');`)
+	tk.MustExec("alter table e6 exchange partition p0 with table e7")
+
+	tk.MustQuery("select * from e6 partition(p0)").Check(testkit.Rows("2"))
+	tk.MustQuery("select * from e7").Check(testkit.Rows("1"))
+	tk.MustGetErrCode("alter table e6 exchange partition p1 with table e7", tmysql.ErrRowDoesNotMatchPartition)
+
+	// test exchange partition from different databases
+	tk.MustExec("create table e8 (a int) partition by hash(a) partitions 2;")
+	tk.MustExec("create database if not exists exchange_partition")
+	tk.MustExec("insert into e8 values (1), (3), (5)")
+	tk.MustExec("use exchange_partition;")
+	tk.MustExec("create table e9 (a int);")
+	tk.MustExec("insert into e9 values (7), (9)")
+	tk.MustExec("alter table test.e8 exchange partition p1 with table e9")
+
+	tk.MustExec("insert into e9 values (11)")
+	tk.MustQuery("select * from e9").Check(testkit.Rows("1", "3", "5", "11"))
+	tk.MustExec("insert into test.e8 values (11)")
+	tk.MustQuery("select * from test.e8").Check(testkit.Rows("7", "9", "11"))
+
+	tk.MustExec("use test")
+	tk.MustExec("create table e10 (a int) partition by hash(a) partitions 2")
+	tk.MustExec("insert into e10 values (0), (2), (4)")
+	tk.MustExec("create table e11 (a int)")
+	tk.MustExec("insert into e11 values (1), (3)")
+	tk.MustExec("alter table e10 exchange partition p1 with table e11")
+	tk.MustExec("insert into e11 values (5)")
+	tk.MustQuery("select * from e11").Check(testkit.Rows("5"))
+	tk.MustExec("insert into e10 values (5), (6)")
+	tk.MustQuery("select * from e10 partition(p0)").Check(testkit.Rows("0", "2", "4", "6"))
+	tk.MustQuery("select * from e10 partition(p1)").Check(testkit.Rows("1", "3", "5"))
+
+	// test for column id
+	tk.MustExec("create table e12 (a int(1), b int, index (a)) partition by hash(a) partitions 3")
+	tk.MustExec("create table e13 (a int(8), b int, index (a));")
+	tk.MustExec("alter table e13 drop column b")
+	tk.MustExec("alter table e13 add column b int")
+	tk.MustGetErrCode("alter table e12 exchange partition p0 with table e13", tmysql.ErrPartitionExchangeDifferentOption)
+	// test for index id
+	tk.MustExec("create table e14 (a int, b int, index(a));")
+	tk.MustExec("alter table e12 drop index a")
+	tk.MustExec("alter table e12 add index (a);")
+	tk.MustGetErrCode("alter table e12 exchange partition p0 with table e14", tmysql.ErrPartitionExchangeDifferentOption)
+
+}
+
+func (s *testIntegrationSuite4) TestExchangePartitionTableCompatiable(c *C) {
+	type testCase struct {
+		ptSQL       string
+		ntSQL       string
+		exchangeSQL string
+		err         *terror.Error
+	}
+	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;",
+			nil,
+		},
+		{
+			"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;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"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;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"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",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"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;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"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;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"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;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"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;",
+			nil,
+		},
+		{
+			"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;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			// foreign key test
+			// Partition table doesn't support to add foreign keys in mysql
+			"create table pt9 (id int not null primary key auto_increment,t_id int not null) partition by hash(id) partitions 1;",
+			"create table nt9 (id int not null primary key auto_increment, t_id int not null,foreign key fk_id (t_id) references pt5(id));",
+			"alter table pt9 exchange partition p0 with table nt9;",
+			ddl.ErrPartitionExchangeForeignKey,
+		},
+		{
+			// Generated column (virtual)
+			"create table pt10 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname,' ')) virtual) partition by hash(id) partitions 1;",
+			"create table nt10 (id int not null, lname varchar(30), fname varchar(100));",
+			"alter table pt10 exchange partition p0 with table nt10;",
+			ddl.ErrUnsupportedOnGeneratedColumn,
+		},
+		{
+			"create table pt11 (id int not null, lname varchar(30), fname varchar(100)) partition by hash(id) partitions 1;",
+			"create table nt11 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) virtual);",
+			"alter table pt11 exchange partition p0 with table nt11;",
+			ddl.ErrUnsupportedOnGeneratedColumn,
+		},
+		{
+
+			"create table pt12 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname,' ')) stored) partition by hash(id) partitions 1;",
+			"create table nt12 (id int not null, lname varchar(30), fname varchar(100));",
+			"alter table pt12 exchange partition p0 with table nt12;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"create table pt13 (id int not null, lname varchar(30), fname varchar(100)) partition by hash(id) partitions 1;",
+			"create table nt13 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) stored);",
+			"alter table pt13 exchange partition p0 with table nt13;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"create table pt14 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) virtual) partition by hash(id) partitions 1;",
+			"create table nt14 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) virtual);",
+			"alter table pt14 exchange partition p0 with table nt14;",
+			nil,
+		},
+		{
+			// unique index
+			"create table pt15 (id int not null, unique index uk_id (id)) partition by hash(id) partitions 1;",
+			"create table nt15 (id int not null, index uk_id (id));",
+			"alter table pt15 exchange partition p0 with table nt15",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			// auto_increment
+			"create table pt16 (id int not null primary key auto_increment) partition by hash(id) partitions 1;",
+			"create table nt16 (id int not null primary key);",
+			"alter table pt16 exchange partition p0 with table nt16;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			// default
+			"create table pt17 (id int not null default 1) partition by hash(id) partitions 1;",
+			"create table nt17 (id int not null);",
+			"alter table pt17 exchange partition p0 with table nt17;",
+			nil,
+		},
+		{
+			// view test
+			"create table pt18 (id int not null) partition by hash(id) partitions 1;",
+			"create view nt18 as select id from nt17;",
+			"alter table pt18 exchange partition p0 with table nt18",
+			ddl.ErrCheckNoSuchTable,
+		},
+		{
+			"create table pt19 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) stored) partition by hash(id) partitions 1;",
+			"create table nt19 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) virtual);",
+			"alter table pt19 exchange partition p0 with table nt19;",
+			ddl.ErrUnsupportedOnGeneratedColumn,
+		},
+		{
+			"create table pt20 (id int not null) partition by hash(id) partitions 1;",
+			"create table nt20 (id int default null);",
+			"alter table pt20 exchange partition p0 with table nt20;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			// unsigned
+			"create table pt21 (id int unsigned) partition by hash(id) partitions 1;",
+			"create table nt21 (id int);",
+			"alter table pt21 exchange partition p0 with table nt21;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			// zerofill
+			"create table pt22 (id int) partition by hash(id) partitions 1;",
+			"create table nt22 (id int zerofill);",
+			"alter table pt22 exchange partition p0 with table nt22;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"create table pt23 (id int, lname varchar(10) charset binary) partition by hash(id) partitions 1;",
+			"create table nt23 (id int, lname varchar(10));",
+			"alter table pt23 exchange partition p0 with table nt23;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"create table pt25 (id int, a datetime on update current_timestamp) partition by hash(id) partitions 1;",
+			"create table nt25 (id int, a datetime);",
+			"alter table pt25 exchange partition p0 with table nt25;",
+			nil,
+		},
+		{
+			"create table pt26 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(lname, ' ')) virtual) partition by hash(id) partitions 1;",
+			"create table nt26 (id int not null, lname varchar(30), fname varchar(100) generated always as (concat(id, ' ')) virtual);",
+			"alter table pt26 exchange partition p0 with table nt26;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+		{
+			"create table pt27 (a int key, b int, index(a)) partition by hash(a) partitions 1;",
+			"create table nt27 (a int not null, b int, index(a));",
+			"alter table pt27 exchange partition p0 with table nt27;",
+			ddl.ErrTablesDifferentMetadata,
+		},
+	}
+
+	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 != nil {
+			_, err := tk.Exec(t.exchangeSQL)
+			c.Assert(terror.ErrorEqual(err, t.err), IsTrue, Commentf(
+				"case %d fail, sql = `%s`\nexpected error = `%v`\n  actual error = `%v`",
+				i, t.exchangeSQL, t.err, err,
+			))
+		} else {
+			tk.MustExec(t.exchangeSQL)
+		}
+	}
+
+}
+
+func (s *testIntegrationSuite7) TestExchangePartitionExpressIndex(c *C) {
+	config.GetGlobalConfig().Experimental.AllowsExpressionIndex = true
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists pt1;")
+	tk.MustExec("create table pt1(a int, b int, c int) PARTITION BY hash (a) partitions 1;")
+	tk.MustExec("alter table pt1 add index idx((a+c));")
+
+	tk.MustExec("drop table if exists nt1;")
+	tk.MustExec("create table nt1(a int, b int, c int);")
+	tk.MustGetErrCode("alter table pt1 exchange partition p0 with table nt1;", tmysql.ErrTablesDifferentMetadata)
+
+	tk.MustExec("alter table nt1 add column (`_V$_idx_0` bigint(20) generated always as (a+b) virtual);")
+	tk.MustGetErrCode("alter table pt1 exchange partition p0 with table nt1;", tmysql.ErrTablesDifferentMetadata)
+
+	// test different expression index when expression returns same field type
+	tk.MustExec("alter table nt1 drop column `_V$_idx_0`;")
+	tk.MustExec("alter table nt1 add index idx((b-c));")
+	tk.MustGetErrCode("alter table pt1 exchange partition p0 with table nt1;", tmysql.ErrTablesDifferentMetadata)
+
+	// test different expression index when expression returns different field type
+	tk.MustExec("alter table nt1 drop index idx;")
+	tk.MustExec("alter table nt1 add index idx((concat(a, b)));")
+	tk.MustGetErrCode("alter table pt1 exchange partition p0 with table nt1;", tmysql.ErrTablesDifferentMetadata)
+
+	tk.MustExec("drop table if exists nt2;")
+	tk.MustExec("create table nt2 (a int, b int, c int)")
+	tk.MustExec("alter table nt2 add index idx((a+c))")
+	tk.MustExec("alter table pt1 exchange partition p0 with table nt2")
+
+}
+
 func (s *testIntegrationSuite4) TestAddPartitionTooManyPartitions(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")
@@ -1724,8 +2104,6 @@ func (s *testIntegrationSuite3) TestPartitionErrorCode(c *C) {
 	tk.MustGetErrCode("alter table t_part optimize partition p0,p1;", tmysql.ErrUnsupportedDDLOperation)
 	tk.MustGetErrCode("alter table t_part rebuild partition p0,p1;", tmysql.ErrUnsupportedDDLOperation)
 	tk.MustGetErrCode("alter table t_part remove partitioning;", tmysql.ErrUnsupportedDDLOperation)
-	tk.MustExec("create table t_part2 like t_part")
-	tk.MustGetErrCode("alter table t_part exchange partition p0 with table t_part2", tmysql.ErrUnsupportedDDLOperation)
 }
 
 func (s *testIntegrationSuite5) TestConstAndTimezoneDepent(c *C) {
@@ -1898,4 +2276,38 @@ func (s *testIntegrationSuite3) TestCommitWhenSchemaChange(c *C) {
 	// That bug will cause data and index inconsistency!
 	tk.MustExec("admin check table schema_change")
 	tk.MustQuery("select * from schema_change").Check(testkit.Rows())
+
+	// Check inconsistency when exchanging partition
+	tk.MustExec(`drop table if exists pt, nt;`)
+	tk.MustExec(`create table pt (a int) partition by hash(a) partitions 2;`)
+	tk.MustExec(`create table nt (a int);`)
+
+	tk.MustExec("begin")
+	tk.MustExec("insert into nt values (1), (3), (5);")
+	tk2.MustExec("alter table pt exchange partition p1 with table nt;")
+	tk.MustExec("insert into nt values (7), (9);")
+
+	atomic.StoreUint32(&session.SchemaChangedWithoutRetry, 1)
+	_, err = tk.Se.Execute(context.Background(), "commit")
+	atomic.StoreUint32(&session.SchemaChangedWithoutRetry, 0)
+	c.Assert(domain.ErrInfoSchemaChanged.Equal(err), IsTrue)
+
+	tk.MustExec("admin check table pt")
+	tk.MustQuery("select * from pt").Check(testkit.Rows())
+	tk.MustExec("admin check table nt")
+	tk.MustQuery("select * from nt").Check(testkit.Rows())
+
+	tk.MustExec("begin")
+	tk.MustExec("insert into pt values (1), (3), (5);")
+	tk2.MustExec("alter table pt exchange partition p1 with table nt;")
+	tk.MustExec("insert into pt values (7), (9);")
+	atomic.StoreUint32(&session.SchemaChangedWithoutRetry, 1)
+	_, err = tk.Se.Execute(context.Background(), "commit")
+	atomic.StoreUint32(&session.SchemaChangedWithoutRetry, 0)
+	c.Assert(domain.ErrInfoSchemaChanged.Equal(err), IsTrue)
+
+	tk.MustExec("admin check table pt")
+	tk.MustQuery("select * from pt").Check(testkit.Rows())
+	tk.MustExec("admin check table nt")
+	tk.MustQuery("select * from nt").Check(testkit.Rows())
 }
diff --git a/ddl/ddl_algorithm_test.go b/ddl/ddl_algorithm_test.go
index 0457c64ae8606..e0cc7fba286b8 100644
--- a/ddl/ddl_algorithm_test.go
+++ b/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},
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index c9e9dfe499535..7bc77d8fd7220 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -2212,8 +2212,6 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 			err = errors.Trace(errUnsupportedOptimizePartition)
 		case ast.AlterTableRemovePartitioning:
 			err = errors.Trace(errUnsupportedRemovePartition)
-		case ast.AlterTableExchangePartition:
-			err = errors.Trace(errUnsupportedExchangePartition)
 		case ast.AlterTableDropColumn:
 			err = d.DropColumn(ctx, ident, spec)
 		case ast.AlterTableDropIndex:
@@ -2226,6 +2224,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.ExchangeTablePartition(ctx, ident, spec)
 		case ast.AlterTableAddConstraint:
 			constr := spec.Constraint
 			switch spec.Constraint.Tp {
@@ -2454,7 +2454,7 @@ func checkAndCreateNewColumn(ctx sessionctx.Context, ti ast.Ident, schema *model
 			}
 
 			if option.Stored {
-				return nil, errUnsupportedOnGeneratedColumn.GenWithStackByArgs("Adding generated stored column through ALTER TABLE")
+				return nil, ErrUnsupportedOnGeneratedColumn.GenWithStackByArgs("Adding generated stored column through ALTER TABLE")
 			}
 
 			_, dependColNames := findDependedColumnNames(specNewColumn)
@@ -2816,6 +2816,174 @@ func (d *ddl) DropTablePartition(ctx sessionctx.Context, ident ast.Ident, spec *
 	return errors.Trace(err)
 }
 
+func checkFieldTypeCompatible(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) &&
+		mysql.HasAutoIncrementFlag(ft.Flag) == mysql.HasAutoIncrementFlag(other.Flag) &&
+		mysql.HasNotNullFlag(ft.Flag) == mysql.HasNotNullFlag(other.Flag) &&
+		mysql.HasZerofillFlag(ft.Flag) == mysql.HasZerofillFlag(other.Flag) &&
+		mysql.HasBinaryFlag(ft.Flag) == mysql.HasBinaryFlag(other.Flag) &&
+		mysql.HasPriKeyFlag(ft.Flag) == mysql.HasPriKeyFlag(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 {
+	// check auto_random
+	if source.AutoRandomBits != target.AutoRandomBits ||
+		source.Charset != target.Charset ||
+		source.Collate != target.Collate ||
+		source.ShardRowIDBits != target.ShardRowIDBits ||
+		source.MaxShardRowIDBits != target.MaxShardRowIDBits ||
+		source.TiFlashReplica != target.TiFlashReplica {
+		return errors.Trace(ErrTablesDifferentMetadata)
+	}
+	if len(source.Cols()) != len(target.Cols()) {
+		return errors.Trace(ErrTablesDifferentMetadata)
+	}
+	// Col compatible check
+	for i, sourceCol := range source.Cols() {
+		targetCol := target.Cols()[i]
+		if isVirtualGeneratedColumn(sourceCol) != isVirtualGeneratedColumn(targetCol) {
+			return ErrUnsupportedOnGeneratedColumn.GenWithStackByArgs("Exchanging partitions for non-generated columns")
+		}
+		// It should strictyle compare expressions for generated columns
+		if sourceCol.Name.L != targetCol.Name.L ||
+			sourceCol.Hidden != targetCol.Hidden ||
+			!checkFieldTypeCompatible(&sourceCol.FieldType, &targetCol.FieldType) ||
+			sourceCol.GeneratedExprString != targetCol.GeneratedExprString {
+			return errors.Trace(ErrTablesDifferentMetadata)
+		}
+		if sourceCol.State != model.StatePublic ||
+			targetCol.State != model.StatePublic {
+			return errors.Trace(ErrTablesDifferentMetadata)
+		}
+		if sourceCol.ID != targetCol.ID {
+			return ErrPartitionExchangeDifferentOption.GenWithStackByArgs(fmt.Sprintf("column: %s", sourceCol.Name))
+		}
+	}
+	if len(source.Indices) != len(target.Indices) {
+		return errors.Trace(ErrTablesDifferentMetadata)
+	}
+	for _, sourceIdx := range source.Indices {
+		var compatIdx *model.IndexInfo
+		for _, targetIdx := range target.Indices {
+			if strings.EqualFold(sourceIdx.Name.L, targetIdx.Name.L) {
+				compatIdx = targetIdx
+			}
+		}
+		// No match index
+		if compatIdx == nil {
+			return errors.Trace(ErrTablesDifferentMetadata)
+		}
+		// Index type is not compatible
+		if sourceIdx.Tp != compatIdx.Tp ||
+			sourceIdx.Unique != compatIdx.Unique ||
+			sourceIdx.Primary != compatIdx.Primary {
+			return errors.Trace(ErrTablesDifferentMetadata)
+		}
+		// The index column
+		if len(sourceIdx.Columns) != len(compatIdx.Columns) {
+			return errors.Trace(ErrTablesDifferentMetadata)
+		}
+		for i, sourceIdxCol := range sourceIdx.Columns {
+			compatIdxCol := compatIdx.Columns[i]
+			if sourceIdxCol.Length != compatIdxCol.Length ||
+				sourceIdxCol.Name.L != compatIdxCol.Name.L {
+				return errors.Trace(ErrTablesDifferentMetadata)
+			}
+		}
+		if sourceIdx.ID != compatIdx.ID {
+			return ErrPartitionExchangeDifferentOption.GenWithStackByArgs(fmt.Sprintf("index: %s", sourceIdx.Name))
+		}
+	}
+
+	return nil
+}
+
+func checkExchangePartition(pt *model.TableInfo, nt *model.TableInfo) error {
+	if nt.IsView() || nt.IsSequence() {
+		return errors.Trace(ErrCheckNoSuchTable)
+	}
+	if pt.GetPartitionInfo() == nil {
+		return errors.Trace(ErrPartitionMgmtOnNonpartitioned)
+	}
+	if nt.GetPartitionInfo() != nil {
+		return errors.Trace(ErrPartitionExchangePartTable.GenWithStackByArgs(nt.Name))
+	}
+
+	if nt.ForeignKeys != nil {
+		return errors.Trace(ErrPartitionExchangeForeignKey.GenWithStackByArgs(nt.Name))
+	}
+
+	// NOTE: if nt is temporary table, it should be checked
+	return nil
+}
+
+func (d *ddl) ExchangeTablePartition(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
+	ptSchema, pt, err := d.getSchemaAndTableByIdent(ctx, ident)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	ptMeta := pt.Meta()
+
+	ntIdent := ast.Ident{Schema: spec.NewTable.Schema, Name: spec.NewTable.Name}
+	ntSchema, nt, err := d.getSchemaAndTableByIdent(ctx, ntIdent)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	ntMeta := nt.Meta()
+
+	err = checkExchangePartition(ptMeta, ntMeta)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	partName := spec.PartitionNames[0].L
+
+	// NOTE: if pt is subPartitioned, it should be checked
+
+	defID, err := tables.FindPartitionByName(ptMeta, partName)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	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{}{defID, 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)
diff --git a/ddl/ddl_worker.go b/ddl/ddl_worker.go
index 6a26a2295f1a8..218eba5beb23f 100644
--- a/ddl/ddl_worker.go
+++ b/ddl/ddl_worker.go
@@ -617,6 +617,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:
@@ -842,6 +844,23 @@ func updateSchemaVersion(t *meta.Meta, job *model.Job) (int64, error) {
 			return 0, errors.Trace(err)
 		}
 		diff.TableID = job.TableID
+	case model.ActionExchangeTablePartition:
+		var (
+			ptSchemaID int64
+			ptTableID  int64
+		)
+		err = job.DecodeArgs(&diff.TableID, &ptSchemaID, &ptTableID)
+		if err != nil {
+			return 0, errors.Trace(err)
+		}
+		diff.OldTableID = job.TableID
+		affects := make([]*model.AffectedOption, 1)
+		affects[0] = &model.AffectedOption{
+			SchemaID:   ptSchemaID,
+			TableID:    ptTableID,
+			OldTableID: ptTableID,
+		}
+		diff.AffectedOpts = affects
 	default:
 		diff.TableID = job.TableID
 	}
diff --git a/ddl/ddl_worker_test.go b/ddl/ddl_worker_test.go
index 6f1b13280e568..cb2ddef475437 100644
--- a/ddl/ddl_worker_test.go
+++ b/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])
+	defID := pt.Partition.Definitions[0].ID
+	exchangeTablePartition := []interface{}{defID, 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)
+
+	// cancel 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) {
diff --git a/ddl/error.go b/ddl/error.go
index b73b4188f02c4..780debc1c1207 100644
--- a/ddl/error.go
+++ b/ddl/error.go
@@ -64,8 +64,8 @@ var (
 	errWrongKeyColumn = terror.ClassDDL.New(mysql.ErrWrongKeyColumn, mysql.MySQLErrName[mysql.ErrWrongKeyColumn])
 	// errWrongFKOptionForGeneratedColumn is for wrong foreign key reference option on generated columns.
 	errWrongFKOptionForGeneratedColumn = terror.ClassDDL.New(mysql.ErrWrongFKOptionForGeneratedColumn, mysql.MySQLErrName[mysql.ErrWrongFKOptionForGeneratedColumn])
-	// errUnsupportedOnGeneratedColumn is for unsupported actions on generated columns.
-	errUnsupportedOnGeneratedColumn = terror.ClassDDL.New(mysql.ErrUnsupportedOnGeneratedColumn, mysql.MySQLErrName[mysql.ErrUnsupportedOnGeneratedColumn])
+	// ErrUnsupportedOnGeneratedColumn is for unsupported actions on generated columns.
+	ErrUnsupportedOnGeneratedColumn = terror.ClassDDL.New(mysql.ErrUnsupportedOnGeneratedColumn, mysql.MySQLErrName[mysql.ErrUnsupportedOnGeneratedColumn])
 	// errGeneratedColumnNonPrior forbids to refer generated column non prior to it.
 	errGeneratedColumnNonPrior = terror.ClassDDL.New(mysql.ErrGeneratedColumnNonPrior, mysql.MySQLErrName[mysql.ErrGeneratedColumnNonPrior])
 	// errDependentByGeneratedColumn forbids to delete columns which are dependent by generated columns.
@@ -210,4 +210,17 @@ var (
 	ErrAddColumnWithSequenceAsDefault = terror.ClassDDL.New(mysql.ErrAddColumnWithSequenceAsDefault, mysql.MySQLErrName[mysql.ErrAddColumnWithSequenceAsDefault])
 	// 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])
+	// ErrTablesDifferentMetadata is returned when exchanges tables is not compatible.
+	ErrTablesDifferentMetadata = terror.ClassDDL.New(mysql.ErrTablesDifferentMetadata, mysql.MySQLErrName[mysql.ErrTablesDifferentMetadata])
+	// ErrRowDoesNotMatchPartition is returned when the row record of exchange table does not match the partition rule.
+	ErrRowDoesNotMatchPartition = terror.ClassDDL.New(mysql.ErrRowDoesNotMatchPartition, mysql.MySQLErrName[mysql.ErrRowDoesNotMatchPartition])
+	// ErrPartitionExchangeForeignKey is returned when exchanged normal table has foreign keys.
+	ErrPartitionExchangeForeignKey = terror.ClassDDL.New(mysql.ErrPartitionExchangeForeignKey, mysql.MySQLErrName[mysql.ErrPartitionExchangeForeignKey])
+	// ErrCheckNoSuchTable is returned when exchaned normal table is view or sequence.
+	ErrCheckNoSuchTable         = terror.ClassDDL.New(mysql.ErrCheckNoSuchTable, mysql.MySQLErrName[mysql.ErrCheckNoSuchTable])
+	errUnsupportedPartitionType = terror.ClassDDL.New(mysql.ErrUnsupportedDDLOperation, fmt.Sprintf(mysql.MySQLErrName[mysql.ErrUnsupportedDDLOperation], "partition type of table %s when exchanging partition"))
+	// ErrPartitionExchangeDifferentOption is returned when attribute doesnot match between partition table and normal table.
+	ErrPartitionExchangeDifferentOption = terror.ClassDDL.New(mysql.ErrPartitionExchangeDifferentOption, mysql.MySQLErrName[mysql.ErrPartitionExchangeDifferentOption])
 )
diff --git a/ddl/failtest/fail_db_test.go b/ddl/failtest/fail_db_test.go
index 83154f20fd8cf..6ae1f72aaa25f 100644
--- a/ddl/failtest/fail_db_test.go
+++ b/ddl/failtest/fail_db_test.go
@@ -94,80 +94,72 @@ func (s *testFailDBSuite) TestHalfwayCancelOperations(c *C) {
 	defer func() {
 		c.Assert(failpoint.Disable("github.com/pingcap/tidb/ddl/truncateTableErr"), IsNil)
 	}()
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("create database cancel_job_db")
+	tk.MustExec("use cancel_job_db")
+
 	// test for truncating table
-	_, err := s.se.Execute(context.Background(), "create database cancel_job_db")
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "use cancel_job_db")
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "create table t(a int)")
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "insert into t values(1)")
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "truncate table t")
+	tk.MustExec("create table t(a int)")
+	tk.MustExec("insert into t values(1)")
+	_, err := tk.Exec("truncate table t")
 	c.Assert(err, NotNil)
+
 	// Make sure that the table's data has not been deleted.
-	rs, err := s.se.Execute(context.Background(), "select count(*) from t")
-	c.Assert(err, IsNil)
-	req := rs[0].NewChunk()
-	err = rs[0].Next(context.Background(), req)
-	c.Assert(err, IsNil)
-	c.Assert(req.NumRows() == 0, IsFalse)
-	row := req.GetRow(0)
-	c.Assert(row.Len(), Equals, 1)
-	c.Assert(row.GetInt64(0), DeepEquals, int64(1))
-	c.Assert(rs[0].Close(), IsNil)
-	// Execute ddl statement reload schema.
-	_, err = s.se.Execute(context.Background(), "alter table t comment 'test1'")
-	c.Assert(err, IsNil)
+	tk.MustQuery("select * from t").Check(testkit.Rows("1"))
+	// Execute ddl statement reload schema
+	tk.MustExec("alter table t comment 'test1'")
 	err = s.dom.DDL().GetHook().OnChanged(nil)
 	c.Assert(err, IsNil)
-	s.se, err = session.CreateSession4Test(s.store)
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "use cancel_job_db")
-	c.Assert(err, IsNil)
-	// Test schema is correct.
-	_, err = s.se.Execute(context.Background(), "select * from t")
-	c.Assert(err, IsNil)
 
+	tk = testkit.NewTestKit(c, s.store)
+	tk.MustExec("use cancel_job_db")
+	// Test schema is correct.
+	tk.MustExec("select * from t")
 	// test for renaming table
 	c.Assert(failpoint.Enable("github.com/pingcap/tidb/ddl/renameTableErr", `return(true)`), IsNil)
 	defer func() {
 		c.Assert(failpoint.Disable("github.com/pingcap/tidb/ddl/renameTableErr"), IsNil)
 	}()
-
-	_, err = s.se.Execute(context.Background(), "create table tx(a int)")
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "insert into tx values(1)")
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "rename table tx to ty")
+	tk.MustExec("create table tx(a int)")
+	tk.MustExec("insert into tx values(1)")
+	_, err = tk.Exec("rename table tx to ty")
 	c.Assert(err, NotNil)
 	// Make sure that the table's data has not been deleted.
-	rs, err = s.se.Execute(context.Background(), "select count(*) from tx")
-	c.Assert(err, IsNil)
-	req = rs[0].NewChunk()
-	err = rs[0].Next(context.Background(), req)
-	c.Assert(err, IsNil)
-	c.Assert(req.NumRows() == 0, IsFalse)
-	row = req.GetRow(0)
-	c.Assert(row.Len(), Equals, 1)
-	c.Assert(row.GetInt64(0), DeepEquals, int64(1))
-	c.Assert(rs[0].Close(), IsNil)
+	tk.MustQuery("select * from tx").Check(testkit.Rows("1"))
 	// Execute ddl statement reload schema.
-	_, err = s.se.Execute(context.Background(), "alter table tx comment 'tx'")
-	c.Assert(err, IsNil)
+	tk.MustExec("alter table tx comment 'tx'")
 	err = s.dom.DDL().GetHook().OnChanged(nil)
 	c.Assert(err, IsNil)
-	s.se, err = session.CreateSession4Test(s.store)
-	c.Assert(err, IsNil)
-	_, err = s.se.Execute(context.Background(), "use cancel_job_db")
+
+	tk = testkit.NewTestKit(c, s.store)
+	tk.MustExec("use cancel_job_db")
+	tk.MustExec("select * from tx")
+	// test for exchanging partition
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/ddl/exchangePartitionErr", `return(true)`), IsNil)
+	defer func() {
+		c.Assert(failpoint.Disable("github.com/pingcap/tidb/ddl/exchangePartitionErr"), IsNil)
+	}()
+	tk.MustExec("create table pt(a int) partition by hash (a) partitions 2")
+	tk.MustExec("insert into pt values(1), (3), (5)")
+	tk.MustExec("create table nt(a int)")
+	tk.MustExec("insert into nt values(7)")
+	_, err = tk.Exec("alter table pt exchange partition p1 with table nt")
+	c.Assert(err, NotNil)
+
+	tk.MustQuery("select * from pt").Check(testkit.Rows("1", "3", "5"))
+	tk.MustQuery("select * from nt").Check(testkit.Rows("7"))
+	// Execute ddl statement reload schema.
+	tk.MustExec("alter table pt comment 'pt'")
+	err = s.dom.DDL().GetHook().OnChanged(nil)
 	c.Assert(err, IsNil)
+
+	tk = testkit.NewTestKit(c, s.store)
+	tk.MustExec("use cancel_job_db")
 	// Test schema is correct.
-	_, err = s.se.Execute(context.Background(), "select * from tx")
-	c.Assert(err, IsNil)
+	tk.MustExec("select * from pt")
 
 	// clean up
-	_, err = s.se.Execute(context.Background(), "drop database cancel_job_db")
-	c.Assert(err, IsNil)
+	tk.MustExec("drop database cancel_job_db")
 }
 
 // TestInitializeOffsetAndState tests the case that the column's offset and state don't be initialized in the file of ddl_api.go when
diff --git a/ddl/generated_column.go b/ddl/generated_column.go
index 0ea41d8691c2e..1e6b4480068a5 100644
--- a/ddl/generated_column.go
+++ b/ddl/generated_column.go
@@ -165,7 +165,7 @@ func checkModifyGeneratedColumn(tbl table.Table, oldCol, newCol *table.Column, n
 	oldColIsStored := !oldCol.IsGenerated() || oldCol.GeneratedStored
 	newColIsStored := !newCol.IsGenerated() || newCol.GeneratedStored
 	if oldColIsStored != newColIsStored {
-		return errUnsupportedOnGeneratedColumn.GenWithStackByArgs("Changing the STORED status")
+		return ErrUnsupportedOnGeneratedColumn.GenWithStackByArgs("Changing the STORED status")
 	}
 
 	// rule 2.
@@ -285,13 +285,13 @@ func checkIndexOrStored(tbl table.Table, oldCol, newCol *table.Column) error {
 	}
 
 	if newCol.GeneratedStored {
-		return errUnsupportedOnGeneratedColumn.GenWithStackByArgs("modifying a stored column")
+		return ErrUnsupportedOnGeneratedColumn.GenWithStackByArgs("modifying a stored column")
 	}
 
 	for _, idx := range tbl.Indices() {
 		for _, col := range idx.Meta().Columns {
 			if col.Name.L == newCol.Name.L {
-				return errUnsupportedOnGeneratedColumn.GenWithStackByArgs("modifying an indexed column")
+				return ErrUnsupportedOnGeneratedColumn.GenWithStackByArgs("modifying an indexed column")
 			}
 		}
 	}
diff --git a/ddl/index.go b/ddl/index.go
index af6ecb3267dcf..23b7fb28d041e 100755
--- a/ddl/index.go
+++ b/ddl/index.go
@@ -101,7 +101,7 @@ func checkPKOnGeneratedColumn(tblInfo *model.TableInfo, indexPartSpecifications
 		}
 		// Virtual columns cannot be used in primary key.
 		if lastCol.IsGenerated() && !lastCol.GeneratedStored {
-			return nil, errUnsupportedOnGeneratedColumn.GenWithStackByArgs("Defining a virtual generated column as primary key")
+			return nil, ErrUnsupportedOnGeneratedColumn.GenWithStackByArgs("Defining a virtual generated column as primary key")
 		}
 	}
 
diff --git a/ddl/partition.go b/ddl/partition.go
index 5b09a0b4e0b09..34d5495aa37ea 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -19,7 +19,9 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/cznic/mathutil"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/parser"
 	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/format"
@@ -34,6 +36,7 @@ import (
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
+	"github.com/pingcap/tidb/util/sqlexec"
 )
 
 const (
@@ -605,6 +608,16 @@ func removePartitionInfo(tblInfo *model.TableInfo, partName string) int64 {
 	return pid
 }
 
+func getPartitionDef(tblInfo *model.TableInfo, partName string) (index int, def *model.PartitionDefinition, _ error) {
+	defs := tblInfo.Partition.Definitions
+	for i := 0; i < len(defs); i++ {
+		if strings.EqualFold(defs[i].Name.L, strings.ToLower(partName)) {
+			return i, &(defs[i]), nil
+		}
+	}
+	return index, nil, table.ErrUnknownPartition.GenWithStackByArgs(partName, tblInfo.Name.O)
+}
+
 // onDropTablePartition deletes old partition meta.
 func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	var partName string
@@ -696,6 +709,240 @@ func onTruncateTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (int64, e
 	return ver, nil
 }
 
+// onExchangeTablePartition exchange partition data
+func (w *worker) onExchangeTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error) {
+	var (
+		//defID only for updateSchemaVersion
+		defID          int64
+		ptSchemaID     int64
+		ptID           int64
+		partName       string
+		withValidation bool
+	)
+
+	if err := job.DecodeArgs(&defID, &ptSchemaID, &ptID, &partName, &withValidation); err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	ntDbInfo, err := checkSchemaExistAndCancelNotExistJob(t, job)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	nt, err := getTableInfoAndCancelFaultJob(t, job, job.SchemaID)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	pt, err := getTableInfo(t, ptID, ptSchemaID)
+	if err != nil {
+		if infoschema.ErrDatabaseNotExists.Equal(err) || infoschema.ErrTableNotExists.Equal(err) {
+			job.State = model.JobStateCancelled
+		}
+		return ver, errors.Trace(err)
+	}
+
+	if pt.State != model.StatePublic {
+		job.State = model.JobStateCancelled
+		return ver, ErrInvalidDDLState.GenWithStack("table %s is not in public, but %s", pt.Name, pt.State)
+	}
+
+	err = checkExchangePartition(pt, nt)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	err = checkTableDefCompatible(pt, nt)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	index, _, err := getPartitionDef(pt, partName)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	if withValidation {
+		err = checkExchangePartitionRecordValidation(w, pt, index, ntDbInfo.Name, nt.Name)
+		if err != nil {
+			job.State = model.JobStateCancelled
+			return ver, errors.Trace(err)
+		}
+	}
+
+	// partition table base auto id
+	ptBaseID, err := t.GetAutoTableID(ptSchemaID, pt.ID)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	ptRandID, err := t.GetAutoRandomID(ptSchemaID, pt.ID)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	// non-partition table base auto id
+	ntBaseID, err := t.GetAutoTableID(job.SchemaID, nt.ID)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	ntRandID, err := t.GetAutoRandomID(job.SchemaID, nt.ID)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	_, partDef, err := getPartitionDef(pt, partName)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	tempID := partDef.ID
+	// exchange table meta id
+	partDef.ID = nt.ID
+
+	err = t.UpdateTable(ptSchemaID, pt)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	failpoint.Inject("exchangePartitionErr", func(val failpoint.Value) {
+		if val.(bool) {
+			job.State = model.JobStateCancelled
+			failpoint.Return(ver, errors.New("occur an error after updating partition id"))
+		}
+	})
+
+	// recreate non-partition table meta info
+	err = t.DropTableOrView(job.SchemaID, nt.ID, true)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	nt.ID = tempID
+
+	err = t.CreateTableOrView(job.SchemaID, nt)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	// both pt and nt set the maximum auto_id between ntBaseID and ptBaseID
+	if ntBaseID > ptBaseID {
+		_, err = t.GenAutoTableID(ptSchemaID, pt.ID, ntBaseID-ptBaseID)
+		if err != nil {
+			job.State = model.JobStateCancelled
+			return ver, errors.Trace(err)
+		}
+	}
+
+	_, err = t.GenAutoTableID(job.SchemaID, nt.ID, mathutil.MaxInt64(ptBaseID, ntBaseID))
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	if ntRandID != 0 || ptRandID != 0 {
+		if ntRandID > ptRandID {
+			_, err = t.GenAutoRandomID(ptSchemaID, pt.ID, ntRandID-ptRandID)
+			if err != nil {
+				job.State = model.JobStateCancelled
+				return ver, errors.Trace(err)
+			}
+		}
+
+		_, err = t.GenAutoRandomID(job.SchemaID, nt.ID, mathutil.MaxInt64(ptRandID, ntRandID))
+		if err != nil {
+			job.State = model.JobStateCancelled
+			return ver, errors.Trace(err)
+		}
+	}
+
+	ver, err = updateSchemaVersion(t, job)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	job.FinishTableJob(model.JobStateDone, model.StateNone, ver, pt)
+	return ver, nil
+}
+
+func checkExchangePartitionRecordValidation(w *worker, pt *model.TableInfo, index int, schemaName, tableName model.CIStr) error {
+	var sql string
+
+	pi := pt.Partition
+
+	switch pi.Type {
+	case model.PartitionTypeHash:
+		if pi.Num == 1 {
+			return nil
+		}
+		sql = fmt.Sprintf("select 1 from `%s`.`%s` where mod(%s, %d) != %d limit 1", schemaName.L, tableName.L, pi.Expr, pi.Num, index)
+	case model.PartitionTypeRange:
+		// Table has only one partition and has the maximum value
+		if len(pi.Definitions) == 1 && strings.EqualFold(pi.Definitions[index].LessThan[0], partitionMaxValue) {
+			return nil
+		}
+		// For range expression and range columns
+		if len(pi.Columns) == 0 {
+			sql = buildCheckSQLForRangeExprPartition(pi, index, schemaName, tableName)
+		} else if len(pi.Columns) == 1 {
+			sql = buildCheckSQLForRangeColumnsPartition(pi, index, schemaName, tableName)
+		}
+	default:
+		return errUnsupportedPartitionType.GenWithStackByArgs(pt.Name.O)
+	}
+
+	var ctx sessionctx.Context
+	ctx, err := w.sessPool.get()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	defer w.sessPool.put(ctx)
+
+	rows, _, err := ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(sql)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	rowCount := len(rows)
+	if rowCount != 0 {
+		return errors.Trace(ErrRowDoesNotMatchPartition)
+	}
+	return nil
+}
+
+func buildCheckSQLForRangeExprPartition(pi *model.PartitionInfo, index int, schemaName, tableName model.CIStr) string {
+	if index == 0 {
+		return fmt.Sprintf("select 1 from `%s`.`%s` where %s >= %s limit 1", schemaName.L, tableName.L, pi.Expr, pi.Definitions[index].LessThan[0])
+	} else if index == len(pi.Definitions)-1 && strings.EqualFold(pi.Definitions[index].LessThan[0], partitionMaxValue) {
+		return fmt.Sprintf("select 1 from `%s`.`%s` where %s < %s limit 1", schemaName.L, tableName.L, pi.Expr, pi.Definitions[index-1].LessThan[0])
+	} else {
+		return fmt.Sprintf("select 1 from `%s`.`%s` where %s < %s or %s >= %s limit 1", schemaName.L, tableName.L, pi.Expr, pi.Definitions[index-1].LessThan[0], pi.Expr, pi.Definitions[index].LessThan[0])
+	}
+}
+
+func buildCheckSQLForRangeColumnsPartition(pi *model.PartitionInfo, index int, schemaName, tableName model.CIStr) string {
+	colName := pi.Columns[0].L
+	if index == 0 {
+		return fmt.Sprintf("select 1 from `%s`.`%s` where `%s` >= %s limit 1", schemaName.L, tableName.L, colName, pi.Definitions[index].LessThan[0])
+	} else if index == len(pi.Definitions)-1 && strings.EqualFold(pi.Definitions[index].LessThan[0], partitionMaxValue) {
+		return fmt.Sprintf("select 1 from `%s`.`%s` where `%s` < %s limit 1", schemaName.L, tableName.L, colName, pi.Definitions[index-1].LessThan[0])
+	} else {
+		return fmt.Sprintf("select 1 from `%s`.`%s` where `%s` < %s or `%s` >= %s limit 1", schemaName.L, tableName.L, colName, pi.Definitions[index-1].LessThan[0], colName, pi.Definitions[index].LessThan[0])
+	}
+}
+
 func checkAddPartitionTooManyPartitions(piDefs uint64) error {
 	if piDefs > uint64(PartitionCountLimit) {
 		return errors.Trace(ErrTooManyPartitions)
diff --git a/ddl/rollingback.go b/ddl/rollingback.go
index 207b9218d1197..bed7c9f310ced 100644
--- a/ddl/rollingback.go
+++ b/ddl/rollingback.go
@@ -350,7 +350,8 @@ func convertJob2RollbackJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job)
 		model.ActionDropForeignKey, model.ActionRenameTable,
 		model.ActionModifyTableCharsetAndCollate, model.ActionTruncateTablePartition,
 		model.ActionModifySchemaCharsetAndCollate, model.ActionRepairTable,
-		model.ActionModifyTableAutoIdCache, model.ActionAlterIndexVisibility:
+		model.ActionModifyTableAutoIdCache, model.ActionAlterIndexVisibility,
+		model.ActionExchangeTablePartition:
 		ver, err = cancelOnlyNotHandledJob(job)
 	default:
 		job.State = model.JobStateCancelled
diff --git a/ddl/serial_test.go b/ddl/serial_test.go
index 7e0513de1ae12..070080e59d347 100644
--- a/ddl/serial_test.go
+++ b/ddl/serial_test.go
@@ -1023,6 +1023,40 @@ func (s *testSerialSuite) TestAutoRandom(c *C) {
 	assertExperimentDisabled("create table auto_random_table (a int primary key auto_random(3))")
 }
 
+func (s *testSerialSuite) TestAutoRandomExchangePartition(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("create database if not exists auto_random_db")
+	defer tk.MustExec("drop database if exists auto_random_db")
+
+	testutil.ConfigTestUtils.SetupAutoRandomTestConfig()
+	defer testutil.ConfigTestUtils.RestoreAutoRandomTestConfig()
+
+	tk.MustExec("use auto_random_db")
+
+	tk.MustExec("drop table if exists e1, e2, e3, e4;")
+
+	tk.MustExec("create table e1 (a bigint primary key auto_random(3)) partition by hash(a) partitions 1;")
+
+	tk.MustExec("create table e2 (a bigint primary key);")
+	tk.MustGetErrCode("alter table e1 exchange partition p0 with table e2;", errno.ErrTablesDifferentMetadata)
+
+	tk.MustExec("create table e3 (a bigint primary key auto_random(2));")
+	tk.MustGetErrCode("alter table e1 exchange partition p0 with table e3;", errno.ErrTablesDifferentMetadata)
+	tk.MustExec("insert into e1 values (), (), ()")
+
+	tk.MustExec("create table e4 (a bigint primary key auto_random(3));")
+	tk.MustExec("insert into e4 values ()")
+	tk.MustExec("alter table e1 exchange partition p0 with table e4;")
+
+	tk.MustQuery("select count(*) from e1").Check(testkit.Rows("1"))
+	tk.MustExec("insert into e1 values ()")
+	tk.MustQuery("select count(*) from e1").Check(testkit.Rows("2"))
+
+	tk.MustQuery("select count(*) from e4").Check(testkit.Rows("3"))
+	tk.MustExec("insert into e4 values ()")
+	tk.MustQuery("select count(*) from e4").Check(testkit.Rows("4"))
+}
+
 func (s *testSerialSuite) TestAutoRandomIncBitsIncrementAndOffset(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("create database if not exists auto_random_db")
diff --git a/infoschema/builder.go b/infoschema/builder.go
index c58c8499c907f..01718c41cf2d3 100644
--- a/infoschema/builder.go
+++ b/infoschema/builder.go
@@ -60,7 +60,7 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
 		newTableID = diff.TableID
 	case model.ActionDropTable, model.ActionDropView, model.ActionDropSequence:
 		oldTableID = diff.TableID
-	case model.ActionTruncateTable, model.ActionCreateView:
+	case model.ActionTruncateTable, model.ActionCreateView, model.ActionExchangeTablePartition:
 		oldTableID = diff.OldTableID
 		newTableID = diff.TableID
 	default:
@@ -74,7 +74,8 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
 	// We try to reuse the old allocator, so the cached auto ID can be reused.
 	var allocs autoid.Allocators
 	if tableIDIsValid(oldTableID) {
-		if oldTableID == newTableID && diff.Type != model.ActionRenameTable {
+		if oldTableID == newTableID && diff.Type != model.ActionRenameTable &&
+			diff.Type != model.ActionExchangeTablePartition {
 			oldAllocs, _ := b.is.AllocByID(oldTableID)
 			allocs = filterAllocators(diff, oldAllocs)
 		}
@@ -106,6 +107,24 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
 			return nil, errors.Trace(err)
 		}
 	}
+	if diff.AffectedOpts != nil {
+		for _, opt := range diff.AffectedOpts {
+			var err error
+			affectedDiff := &model.SchemaDiff{
+				Version:     diff.Version,
+				Type:        diff.Type,
+				SchemaID:    opt.SchemaID,
+				TableID:     opt.TableID,
+				OldSchemaID: opt.OldSchemaID,
+				OldTableID:  opt.OldTableID,
+			}
+			affectedIDs, err := b.ApplyDiff(m, affectedDiff)
+			if err != nil {
+				return nil, errors.Trace(err)
+			}
+			tblIDs = append(tblIDs, affectedIDs...)
+		}
+	}
 	return tblIDs, nil
 }
 
diff --git a/planner/core/planbuilder.go b/planner/core/planbuilder.go
index ceb81d757d1dc..13d6e00578369 100644
--- a/planner/core/planbuilder.go
+++ b/planner/core/planbuilder.go
@@ -2633,7 +2633,7 @@ func (b *PlanBuilder) buildDDL(ctx context.Context, node ast.DDLNode) (Plan, err
 		b.visitInfo = appendVisitInfo(b.visitInfo, mysql.AlterPriv, v.Table.Schema.L,
 			v.Table.Name.L, "", authErr)
 		for _, spec := range v.Specs {
-			if spec.Tp == ast.AlterTableRenameTable {
+			if spec.Tp == ast.AlterTableRenameTable || spec.Tp == ast.AlterTableExchangePartition {
 				if b.ctx.GetSessionVars().User != nil {
 					authErr = ErrTableaccessDenied.GenWithStackByArgs("DROP", b.ctx.GetSessionVars().User.AuthUsername,
 						b.ctx.GetSessionVars().User.AuthHostname, v.Table.Name.L)