// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package functionality

import (
	"errors"
	"fmt"
	"math"
	"strings"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/workload"
	. "github.com/pingcap/endless/testcase/geopartition/utils"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

const (
	directOpt1 string = `FOLLOWERS=1 LEADER_CONSTRAINTS="[+region=us-east-1]" FOLLOWER_CONSTRAINTS="[+region=us-east-2]"`
	directOpt2 string = `PRIMARY_REGION="us-east-2" REGIONS="us-east-1,us-east-2" FOLLOWERS=1`
	directOpt3 string = `FOLLOWERS=1 LEADER_CONSTRAINTS="[+region=us-west-1]"`
)

type partitionedTable struct {
	partitionType  string
	createSQL      string
	tablePlacement string
	hidePartitions bool
}

func pickPartitionedTable(g *GenerateContext) *partitionedTable {
	return Try(g.Pick(NewGenerationForker(func(ctx *GenerateContext) (interface{}, error) {
		createSQL := "CREATE TABLE `t1` (\n  `a` int(11) DEFAULT NULL\n)" +
			" ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
		tablePlacement := Try(ctx.PickEnum("", "PLACEMENT POLICY=`p1`")).(string)
		if tablePlacement != "" {
			createSQL += fmt.Sprintf(" /*T![placement] %s */", tablePlacement)
		}

		partitionType := Try(ctx.PickEnum("RANGE", "LIST", "HASH")).(string)
		createSQL += fmt.Sprintf("\nPARTITION BY %s (`a`)", partitionType)

		hidePartitions := partitionType == "HASH" && tablePlacement != "" && Try(ctx.PickBool()).(bool)
		if !hidePartitions {
			partitions := make([]string, 3)
			for i := range partitions {
				partition := fmt.Sprintf("PARTITION `p%d`", i)
				switch partitionType {
				case "RANGE":
					partition += fmt.Sprintf(" VALUES LESS THAN (1%0*d)", i+1, 0)
				case "LIST":
					base := 0
					if i > 0 {
						base = int(math.Pow(10.0, float64(i)))
					}
					partition += fmt.Sprintf(" VALUES IN (%d,%d)", base+1, base+2)
				case "HASH":
				default:
					return nil, errors.New("Invalid partition type: " + partitionType)
				}
				partitions[i] = partition
				if i == 1 {
					partitions[i] += " /*T![placement] PLACEMENT POLICY=`p2` */"
				}
			}
			createSQL += fmt.Sprintf("\n(%s)", strings.Join(partitions, ",\n "))
		} else {
			createSQL += " PARTITIONS 3"
		}

		return &partitionedTable{
			partitionType:  partitionType,
			createSQL:      createSQL,
			tablePlacement: tablePlacement,
			hidePartitions: hidePartitions,
		}, nil
	}))).(*partitionedTable)
}

func preparePartitionTestMeta(ctx *SuiteContext) func() {
	MustExec(ctx.DB, "set @@session.tidb_enable_list_partition = ON")
	MustExec(ctx.DB, "DROP TABLE IF EXISTS t1,t2")
	MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p1")
	MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p2")
	MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p3")
	MustExec(ctx.DB, "CREATE PLACEMENT POLICY p1 "+directOpt1)
	MustExec(ctx.DB, "CREATE PLACEMENT POLICY p2 "+directOpt2)
	MustExec(ctx.DB, "CREATE PLACEMENT POLICY p3 "+directOpt3)
	return func() {
		MustExec(ctx.DB, "DROP TABLE IF EXISTS t1,t2")
		MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p1")
		MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p2")
		MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p3")
	}
}

func partitionTableTest(ctx *SuiteContext) {
	It("Created partition table should be scheduled", func() {
		clear := preparePartitionTestMeta(ctx)
		defer clear()

		ForkRun(func(g *GenerateContext) {
			MustExec(ctx.DB, "drop table if exists t1")
			table := pickPartitionedTable(g)
			log.Info("Test create partitioned table with placement", zap.String("SQL", table.createSQL))
			MustExec(ctx.DB, table.createSQL)

			MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101)")
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}}))

			truncate := Try(g.PickEnum("", "table", "partition")).(string)
			if truncate == "table" {
				log.Info("truncate table t1")
				MustExec(ctx.DB, "TRUNCATE TABLE t1")
				Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{}))
				MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101)")
			}

			if truncate == "partition" {
				log.Info("truncate table partition t1.p1")
				MustExec(ctx.DB, "ALTER TABLE t1 TRUNCATE PARTITION p1")
				if table.partitionType == "HASH" {
					Expect(ReadAll(MustQuery(ctx.DB, "SELECT COUNT(1) FROM t1 ORDER BY a"))).Should(Equal([][]string{{"2"}}))
					MustExec(ctx.DB, "DELETE FROM t1 WHERE 1")
					MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101)")
				} else {
					Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"101"}}))
					MustExec(ctx.DB, "INSERT INTO t1 VALUES (11)")
				}
			}

			// Check show create table
			showCreateSQL := ReadAll(MustQuery(ctx.DB, "show create table t1"))[0][1]
			Expect(showCreateSQL).Should(
				Equal(table.createSQL),
				fmt.Sprintf("%s\n[!=]\n%s", showCreateSQL, table.createSQL),
			)

			Try(WaitScheduled(ctx.DB, "t1", time.Minute))
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}}))

			// Check p0 schedule
			if table.tablePlacement != "" {
				placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p0"))[0][1]
				expected := `LEADER_CONSTRAINTS="[+region=us-east-1]" FOLLOWERS=1 FOLLOWER_CONSTRAINTS="[+region=us-east-2]"`
				Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))

				CheckRegions(ctx, "t1", "p0", func(r Regions) {
					r.CheckLeaderPlacement("+region=us-east-1,-engine=tiflash").
						CheckFollowerPlacement("+region=us-east-2,-engine=tiflash", 1).
						CheckNoPeersLeft()
				})
			} else {
				CheckRegions(ctx, "t1", "p0", func(r Regions) {
					r.CheckLeaderPlacement("-engine=tiflash").
						CheckFollowerPlacement("-engine=tiflash", 2).
						CheckNoPeersLeft()
				})
			}

			// Check p1 schedule
			if !table.hidePartitions {
				placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p1"))[0][1]
				expected := `PRIMARY_REGION="us-east-2" REGIONS="us-east-1,us-east-2" FOLLOWERS=1`
				Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
				CheckRegions(ctx, "t1", "p1", func(r Regions) {
					r.CheckLeaderPlacement("+region=us-east-2,-engine=tiflash").
						CheckFollowerPlacement("+region=us-east-1,-engine=tiflash", 1).
						CheckNoPeersLeft()
				})
			}

			if table.partitionType != "HASH" {
				MustExec(ctx.DB, "ALTER TABLE t1 DROP PARTITION p1")
				Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"101"}}))
				Expect(ReadAll(MustQuery(ctx.DB, "SHOW PLACEMENT LIKE '%t1 PARTITION p1'"))).Should(Equal([][]string{}))
			}
		})
	})

	It("Newly added partition should be scheduled", func() {
		clear := preparePartitionTestMeta(ctx)
		defer clear()

		ForkRun(func(g *GenerateContext) {
			MustExec(ctx.DB, "drop table if exists t1")
			table := pickPartitionedTable(g)
			if table.partitionType == "HASH" {
				// Hash partition do not support add partition
				return
			}

			if table.tablePlacement == "" {
				// We only need to test cases when table has placement settings
				return
			}

			log.Info("Test add partition with placement", zap.String("SQL", table.createSQL))
			MustExec(ctx.DB, table.createSQL)

			switch table.partitionType {
			case "RANGE":
				MustExec(ctx.DB, "ALTER TABLE t1 ADD PARTITION ("+
					"PARTITION p3 VALUES LESS THAN (10000),"+
					"PARTITION p4 VALUES LESS THAN (100000) PLACEMENT POLICY p2"+
					")")
			case "LIST":
				MustExec(ctx.DB, "ALTER TABLE t1 ADD PARTITION ("+
					"PARTITION p3 VALUES IN (1001,1002),"+
					"PARTITION p4 VALUES IN (10001,10002) PLACEMENT POLICY p2"+
					")")
			default:
				Fail(fmt.Sprintf("Unsupported partition type: %s", table.partitionType))
			}

			MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101), (1001), (10001)")
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}, {"1001"}, {"10001"}}))
			Try(WaitScheduled(ctx.DB, "t1", time.Minute))
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}, {"1001"}, {"10001"}}))

			// Partition without placement settings should use the table's settings
			placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p3"))[0][1]
			expected := `LEADER_CONSTRAINTS="[+region=us-east-1]" FOLLOWERS=1 FOLLOWER_CONSTRAINTS="[+region=us-east-2]"`
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
			CheckRegions(ctx, "t1", "p3", func(r Regions) {
				r.CheckLeaderPlacement("+region=us-east-1,-engine=tiflash").
					CheckFollowerPlacement("+region=us-east-2,-engine=tiflash", 1).
					CheckNoPeersLeft()
			})

			// Partition with its own placement settings should use its own
			placement = ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p4"))[0][1]
			expected = `PRIMARY_REGION="us-east-2" REGIONS="us-east-1,us-east-2" FOLLOWERS=1`
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
			CheckRegions(ctx, "t1", "p4", func(r Regions) {
				r.CheckLeaderPlacement("+region=us-east-2,-engine=tiflash").
					CheckFollowerPlacement("+region=us-east-1,-engine=tiflash", 1).
					CheckNoPeersLeft()
			})
		})
	})

	It("Alter table placement should affect partition placement", func() {
		clear := preparePartitionTestMeta(ctx)
		defer clear()

		ForkRun(func(g *GenerateContext) {
			MustExec(ctx.DB, "drop table if exists t1")
			table := pickPartitionedTable(g)
			log.Info("Test alter table placement for partition table", zap.String("SQL", table.createSQL))
			MustExec(ctx.DB, table.createSQL)

			resetToDefault := Try(g.PickBool()).(bool)
			expectedShowCreate := ""
			if resetToDefault {
				MustExec(ctx.DB, "ALTER TABLE t1 PLACEMENT POLICY `default`")
				expectedShowCreate = strings.ReplaceAll(table.createSQL, " /*T![placement] "+table.tablePlacement+" */", "")
			} else if table.tablePlacement == "" {
				MustExec(ctx.DB, "ALTER TABLE t1 PLACEMENT POLICY p3")
				expectedShowCreate = strings.ReplaceAll(table.createSQL,
					"COLLATE=utf8mb4_bin", "COLLATE=utf8mb4_bin /*T![placement] PLACEMENT POLICY=`p3` */")
			} else {
				MustExec(ctx.DB, "ALTER TABLE t1 PLACEMENT POLICY p3")
				expectedShowCreate = strings.ReplaceAll(table.createSQL, "/*T![placement] "+table.tablePlacement+" */", "/*T![placement] PLACEMENT POLICY=`p3` */")
			}

			showCreate := ReadAll(MustQuery(ctx.DB, "SHOW CREATE TABLE t1"))[0][1]
			Expect(showCreate).Should(Equal(expectedShowCreate), fmt.Sprintf("%s\n!=\n%s", showCreate, expectedShowCreate))

			MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101)")
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}}))

			Try(WaitScheduled(ctx.DB, "t1", time.Minute))
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}}))

			// p0 will be affected by table placement
			if resetToDefault {
				CheckRegions(ctx, "t1", "p0", func(r Regions) {
					r.CheckLeaderPlacement("-engine=tiflash").
						CheckFollowerPlacement("-engine=tiflash", 2).
						CheckNoPeersLeft()
				})
			} else {
				CheckRegions(ctx, "t1", "p0", func(r Regions) {
					r.CheckLeaderPlacement("+region=us-west-1,-engine=tiflash").
						CheckFollowerPlacement("-engine=tiflash", 1).
						CheckNoPeersLeft()
				})
			}

			// p1 will not be affected by table placement because it has its own configuration
			if !table.hidePartitions {
				CheckRegions(ctx, "t1", "p1", func(r Regions) {
					r.CheckLeaderPlacement("+region=us-east-2,-engine=tiflash").
						CheckFollowerPlacement("+region=us-east-1,-engine=tiflash", 1).
						CheckNoPeersLeft()
				})
			}
		})
	})

	It("Alter partition placement should take effect", func() {
		clear := preparePartitionTestMeta(ctx)
		defer clear()

		ForkRun(func(g *GenerateContext) {
			MustExec(ctx.DB, "drop table if exists t1")
			table := pickPartitionedTable(g)
			log.Info("Test alter table placement for partition table", zap.String("SQL", table.createSQL))
			MustExec(ctx.DB, table.createSQL)

			checkP1WithConstraints3 := func() {
				placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p1"))[0][1]
				expected := `LEADER_CONSTRAINTS="[+region=us-west-1]" FOLLOWERS=1`
				Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))

				Try(WaitScheduled(ctx.DB, "t1", time.Minute))
				CheckRegions(ctx, "t1", "p1", func(r Regions) {
					r.CheckLeaderPlacement("+region=us-west-1,-engine=tiflash").
						CheckFollowerPlacement("-engine=tiflash", 1).
						CheckNoPeersLeft()
				})
			}

			checkP1WithTableConstraints := func() {
				if table.tablePlacement == "" {
					Expect(ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p1"))).
						Should(Equal([][]string{}))

					Try(WaitScheduled(ctx.DB, "t1", time.Minute))
					CheckRegions(ctx, "t1", "p1", func(r Regions) {
						r.CheckLeaderPlacement("-engine=tiflash").
							CheckFollowerPlacement("-engine=tiflash", 2).
							CheckNoPeersLeft()
					})
				} else {
					placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p1"))[0][1]
					expected := `LEADER_CONSTRAINTS="[+region=us-east-1]" FOLLOWERS=1 FOLLOWER_CONSTRAINTS="[+region=us-east-2]"`
					Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
					CheckRegions(ctx, "t1", "p1", func(r Regions) {
						r.CheckLeaderPlacement("+region=us-east-1,-engine=tiflash").
							CheckFollowerPlacement("+region=us-east-2,-engine=tiflash", 1).
							CheckNoPeersLeft()
					})
				}
			}

			var checkP1Func func()
			var alterP1 string
			if Try(g.PickBool()).(bool) {
				alterP1 = "PLACEMENT POLICY `default`"
				checkP1Func = checkP1WithTableConstraints
			} else {
				alterP1 = "PLACEMENT POLICY p3"
				checkP1Func = checkP1WithConstraints3
			}

			MustExec(ctx.DB, "ALTER TABLE t1 PARTITION p0 PLACEMENT POLICY p2")
			MustExec(ctx.DB, "ALTER TABLE t1 PARTITION p1 "+alterP1)

			Try(WaitScheduled(ctx.DB, "t1", time.Minute))

			placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p0"))[0][1]
			expected := `PRIMARY_REGION="us-east-2" REGIONS="us-east-1,us-east-2" FOLLOWERS=1`
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
			CheckRegions(ctx, "t1", "p0", func(r Regions) {
				r.CheckLeaderPlacement("+region=us-east-2,-engine=tiflash").
					CheckFollowerPlacement("+region=us-east-1,-engine=tiflash", 1).
					CheckNoPeersLeft()
			})

			checkP1Func()

			rows := ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p2"))
			if table.tablePlacement != "" {
				Expect(rows[0][1]).Should(Equal(ReadAll(MustQuery(ctx.DB, "show placement for table t1"))[0][1]))
			} else {
				Expect(rows).Should(Equal([][]string{}))
			}

			MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101)")
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}}))
		})
	})

	It("Alter placement policy should effect all table and partitions", func() {
		clear := preparePartitionTestMeta(ctx)
		defer clear()

		ForkRun(func(g *GenerateContext) {
			MustExec(ctx.DB, "drop table if exists t1,t2")
			table := pickPartitionedTable(g)
			if table.tablePlacement == "" {
				return
			}

			log.Info("Test alter policy for partition table", zap.String("SQL", table.createSQL))
			MustExec(ctx.DB, table.createSQL)
			MustExec(ctx.DB, "CREATE TABLE t2(a int) PLACEMENT POLICY p2")

			MustExec(ctx.DB, "ALTER PLACEMENT POLICY p1 "+directOpt3)
			Try(WaitScheduled(ctx.DB, "t1", time.Minute))

			expected := `LEADER_CONSTRAINTS="[+region=us-west-1]" FOLLOWERS=1`
			placement := ReadAll(MustQuery(ctx.DB, "show placement for table t1"))[0][1]
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
			placement = ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p0"))[0][1]
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))

			CheckRegions(ctx, "t1", "p0", func(r Regions) {
				r.CheckLeaderPlacement("+region=us-west-1,-engine=tiflash").
					CheckFollowerPlacement("-engine=tiflash", 1).
					CheckNoPeersLeft()
			})

			MustExec(ctx.DB, "ALTER PLACEMENT POLICY p2 "+directOpt3)
			Try(WaitScheduled(ctx.DB, "t1", time.Minute))
			Try(WaitScheduled(ctx.DB, "t2", time.Minute))

			placement = ReadAll(MustQuery(ctx.DB, "show placement for table t2"))[0][1]
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))
			placement = ReadAll(MustQuery(ctx.DB, "show placement for table t1 partition p1"))[0][1]
			Expect(placement).Should(Equal(expected), fmt.Sprintf("%s\n[!=]\n%s", placement, expected))

			CheckRegions(ctx, "t1", "p1", func(r Regions) {
				r.CheckLeaderPlacement("+region=us-west-1,-engine=tiflash").
					CheckFollowerPlacement("-engine=tiflash", 1).
					CheckNoPeersLeft()
			})

			CheckRegions(ctx, "t2", "", func(r Regions) {
				r.CheckLeaderPlacement("+region=us-west-1,-engine=tiflash").
					CheckFollowerPlacement("-engine=tiflash", 1).
					CheckNoPeersLeft()
			})

			MustExec(ctx.DB, "INSERT INTO t1 VALUES(1), (11), (101)")
			Expect(ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 ORDER BY a"))).Should(Equal([][]string{{"1"}, {"11"}, {"101"}}))
		})
	})

	It("Partition exchange on table with placement options should work", func() {
		defer func() {
			MustExec(ctx.DB, "set tidb_enable_exchange_partition=0")
			MustExec(ctx.DB, "DROP TABLE IF EXISTS t1,t2")
			MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p1")
			MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p2")
		}()

		MustExec(ctx.DB, "set tidb_enable_exchange_partition=1")
		MustExec(ctx.DB, "DROP TABLE IF EXISTS t1,t2")
		MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p1")
		MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p2")
		MustExec(ctx.DB, "CREATE PLACEMENT POLICY p1 FOLLOWERS=4")
		MustExec(ctx.DB, "CREATE PLACEMENT POLICY p2 FOLLOWERS=1")
		MustExec(ctx.DB, "CREATE TABLE t1 ("+
			"firstname VARCHAR(25) NOT NULL,"+
			"lastname VARCHAR(25) NOT NULL,"+
			"username VARCHAR(16) NOT NULL,"+
			"email VARCHAR(35),"+
			"joined DATE NOT NULL)"+
			"PARTITION BY RANGE(YEAR(joined)) ("+
			"PARTITION p0 VALUES LESS THAN (1960) PLACEMENT POLICY=p1,"+
			"PARTITION p1 VALUES LESS THAN (1970),"+
			"PARTITION p2 VALUES LESS THAN MAXVALUE"+
			")")
		MustExec(ctx.DB, "INSERT INTO t1 values('f10', 'l10', 'u10', 'a10@g.cn', '1951-01-01')")
		MustExec(ctx.DB, "INSERT INTO t1 values('f11', 'l11', 'u11', 'a11@g.cn', '1961-01-01')")
		MustExec(ctx.DB, "CREATE TABLE t2 ("+
			"firstname VARCHAR(25) NOT NULL,"+
			"lastname VARCHAR(25) NOT NULL,"+
			"username VARCHAR(16) NOT NULL,"+
			"email VARCHAR(35),"+
			"joined DATE NOT NULL"+
			") PLACEMENT POLICY=p1")
		MustExec(ctx.DB, "INSERT INTO t2 values('f2', 'l2', 'u2', 'a2@g.cn', '1951-02-02')")

		checkTablePlacement := func() {
			t1 := ReadAll(MustQuery(ctx.DB, "SHOW CREATE TABLE t1"))[0][1]
			Expect(t1).Should(Equal("CREATE TABLE `t1` (\n  " +
				"`firstname` varchar(25) NOT NULL,\n  " +
				"`lastname` varchar(25) NOT NULL,\n  " +
				"`username` varchar(16) NOT NULL,\n  " +
				"`email` varchar(35) DEFAULT NULL,\n  " +
				"`joined` date NOT NULL\n) " +
				"ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin\n" +
				"PARTITION BY RANGE (YEAR(`joined`))\n" +
				"(PARTITION `p0` VALUES LESS THAN (1960) /*T![placement] PLACEMENT POLICY=`p1` */,\n " +
				"PARTITION `p1` VALUES LESS THAN (1970),\n " +
				"PARTITION `p2` VALUES LESS THAN (MAXVALUE))"))

			t2 := ReadAll(MustQuery(ctx.DB, "SHOW CREATE TABLE t2"))[0][1]
			Expect(t2).Should(Equal("CREATE TABLE `t2` (\n  " +
				"`firstname` varchar(25) NOT NULL,\n  " +
				"`lastname` varchar(25) NOT NULL,\n  " +
				"`username` varchar(16) NOT NULL,\n  " +
				"`email` varchar(35) DEFAULT NULL,\n  " +
				"`joined` date NOT NULL\n) " +
				"ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin /*T![placement] PLACEMENT POLICY=`p1` */"))

			Try(WaitScheduled(ctx.DB, "t1", time.Minute))
			Try(WaitScheduled(ctx.DB, "t2", time.Minute))
			// wait local cache updated
			CheckRegions(ctx, "t1", "p0", func(r Regions) {
				r.CheckLeaderPlacement("").
					CheckFollowerPlacement("", 4).
					CheckNoPeersLeft()
			})
			CheckRegions(ctx, "t1", "p1", func(r Regions) {
				r.CheckLeaderPlacement("").
					CheckFollowerPlacement("", 2).
					CheckNoPeersLeft()
			})
			CheckRegions(ctx, "t1", "p2", func(r Regions) {
				r.CheckLeaderPlacement("").
					CheckFollowerPlacement("", 2).
					CheckNoPeersLeft()
			})
			CheckRegions(ctx, "t2", "", func(r Regions) {
				r.CheckLeaderPlacement("").
					CheckFollowerPlacement("", 4).
					CheckNoPeersLeft()
			})
		}
		checkTablePlacement()

		MustExec(ctx.DB, "ALTER TABLE t1 EXCHANGE PARTITION p0 WITH TABLE t2")
		rows := ReadAll(MustQuery(ctx.DB, "select * from t1 order by joined"))
		Expect(rows).Should(Equal([][]string{
			{"f2", "l2", "u2", "a2@g.cn", "1951-02-02"},
			{"f11", "l11", "u11", "a11@g.cn", "1961-01-01"},
		}))
		rows = ReadAll(MustQuery(ctx.DB, "select * from t2"))
		Expect(rows).Should(Equal([][]string{{"f10", "l10", "u10", "a10@g.cn", "1951-01-01"}}))
		checkTablePlacement()
	})
}