package attributes_test

import (
	"context"
	"fmt"
	"net/url"
	"sort"
	"strings"
	"time"

	"github.com/jmoiron/sqlx"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/test-infra/sdk/resource"

	"github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"

	_ "github.com/go-sql-driver/mysql"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
)

type Attributes struct {
	ID         string `db:"ID"`
	Type       string `db:"TYPE"`
	Attributes string `db:"ATTRIBUTES"`
	Ranges     string `db:"RANGES"`
}

var _ = Describe("DDL functionality testing", func() {
	var (
		tc     resource.TiDBCluster
		dbDSN  string
		db     *sqlx.DB
		client *pd.Client
	)

	ctx := suiteTestCtx
	BeforeEach(func() {
		r := ctx.Resource("tc")
		tc = r.(resource.TiDBCluster)
		dbURL := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
		dbDSN = fmt.Sprintf("root@tcp(%s)/test", dbURL.Host)
		// dbDSN = "root@tcp(127.0.0.1:4000)/test"
		// prometheusUrl = "http://127.0.0.1:9090"
		db = Try(sqlx.Open("mysql", dbDSN)).(*sqlx.DB)
		client = Try(pd.NewPDClient(tc)).(*pd.Client)
	})

	AfterEach(func() {
		db.Close()
	})

	Context("Alter Clause", func() {
		TestCase("alter table attritbutes", "DB_ATTRIBUTES_FUN_000", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Expect(attrs[0].ID).To(Equal("schema/test/t1"))
			Expect(attrs[0].Type).To(Equal("key-range"))
			Expect(attrs[0].Attributes).To(Equal(`"key=value"`))
		})
		TestCase("alter table partition attritbutes", "DB_ATTRIBUTES_FUN_001", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t2/p0")
				client.DeleteLabelRule("schema/test/t2")
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
			}()
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t2 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='key1=value1';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(attrs, func(i, j int) bool { return attrs[i].ID > attrs[j].ID })
			Expect(attrs).Should(HaveLen(2))
			Expect(attrs[0].Type).To(Equal("key-range"))
			Expect(attrs[0].Attributes).To(Equal(`"key1=value1"`))
			Expect(attrs[1].Type).To(Equal("key-range"))
			Expect(attrs[1].Attributes).To(Equal(`"key=value"`))
			Expect(strings.Split(attrs[1].Ranges, ", [")).Should(HaveLen(3))
		})
	})
	Context("Default Keyword", func() {
		TestCase("use default to reset", "DB_ATTRIBUTES_FUN_011", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(2))
			Try(db.ExecContext(ctx2, "alter table t1 attributes=default;"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes=default;"))
			attrs = []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(0))
		})
		TestCase("use empty attributes to reset", "DB_ATTRIBUTES_FUN_012", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(2))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='';"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='';"))
			attrs = []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(0))
		})
	})
	Context("Truncate Clause", func() {
		TestCase("truncate table with no partition attribute", "DB_ATTRIBUTES_FUN_002", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Try(db.ExecContext(ctx2, "truncate table t1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules1).Should(HaveLen(1))
			Expect(rules1[0].ID).To(Equal("schema/test/t1"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key=value"`))
			// TODO: We'd better assert the ID matches the key range.
			Expect(attrs[0].Ranges).NotTo(Equal(rules1[0].Ranges))
		})
		TestCase("truncate table with partition attribute", "DB_ATTRIBUTES_FUN_003", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Try(db.ExecContext(ctx2, "truncate table t1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules1).Should(HaveLen(1))
			Expect(rules1[0].ID).To(Equal("schema/test/t1/p0"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).NotTo(Equal(rules1[0].Ranges))
		})
		TestCase("truncate partition with partition attribute", "DB_ATTRIBUTES_FUN_016", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(20000), partition p2 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Try(db.ExecContext(ctx2, "alter table t1 truncate partition p1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules1).Should(HaveLen(1))
			Expect(attrs[0].Ranges).To(Equal(rules1[0].Ranges))
			Try(db.ExecContext(ctx2, "alter table t1 truncate partition p0;"))
			rules2 := []Attributes{}
			Expect(db.Select(&rules2, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules2).Should(HaveLen(1))
			Expect(rules2[0].ID).To(Equal("schema/test/t1/p0"))
			Expect(rules2[0].Type).To(Equal("key-range"))
			Expect(rules2[0].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).NotTo(Equal(rules2[0].Ranges))
		})
	})
	Context("Rename Clause", func() {
		TestCase("rename table with partition attribute", "DB_ATTRIBUTES_FUN_008", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t1/p0")
				client.DeleteLabelRule("schema/test1/t2")
				client.DeleteLabelRule("schema/test1/t2/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists test1.t2;"))
				Try(db.ExecContext(ctx2, "drop database if exists test1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "create database test1;"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='key1=value1';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(attrs, func(i, j int) bool { return attrs[i].ID > attrs[j].ID })
			Try(db.ExecContext(ctx2, "rename table test.t1 to test1.t2;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(rules1, func(i, j int) bool { return rules1[i].ID > rules1[j].ID })
			Expect(rules1).Should(HaveLen(2))
			Expect(rules1[0].ID).To(Equal("schema/test1/t2/p0"))
			Expect(rules1[1].ID).To(Equal("schema/test1/t2"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[1].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key1=value1"`))
			Expect(rules1[1].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).To(Equal(rules1[0].Ranges))
			Expect(attrs[1].Ranges).To(Equal(rules1[1].Ranges))
		})
	})
	Context("Flashback Clause", func() {
		TestCase("flashback table with both table and partition attrs", "DB_ATTRIBUTES_FUN_013", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='key1=value1';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(attrs, func(i, j int) bool { return attrs[i].ID > attrs[j].ID })

			// drop table
			Try(db.ExecContext(ctx2, "drop table t1;"))
			Try(db.ExecContext(ctx2, "flashback table t1 to t2;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(rules1, func(i, j int) bool { return rules1[i].ID > rules1[j].ID })
			Expect(rules1).Should(HaveLen(2))
			Expect(rules1[0].ID).To(Equal("schema/test/t2/p0"))
			Expect(rules1[1].ID).To(Equal("schema/test/t2"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[1].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key1=value1"`))
			Expect(rules1[1].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).To(Equal(rules1[0].Ranges))
			Expect(attrs[1].Ranges).To(Equal(rules1[1].Ranges))

			// truncate table
			Try(db.ExecContext(ctx2, "truncate table t2;"))
			Try(db.ExecContext(ctx2, "flashback table t2 to t1;"))
			rules2 := []Attributes{}
			Expect(db.Select(&rules2, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(rules2, func(i, j int) bool { return rules2[i].ID > rules2[j].ID })
			Expect(rules2).Should(HaveLen(2))
			Expect(rules2[0].ID).To(Equal("schema/test/t1/p0"))
			Expect(rules2[1].ID).To(Equal("schema/test/t1"))
			Expect(rules2[0].Type).To(Equal("key-range"))
			Expect(rules2[1].Type).To(Equal("key-range"))
			Expect(rules2[0].Attributes).To(Equal(`"key1=value1"`))
			Expect(rules2[1].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).To(Equal(rules2[0].Ranges))
			Expect(attrs[1].Ranges).To(Equal(rules2[1].Ranges))
		})
	})
	Context("Recover Clause", func() {
		TestCase("recover table with both table and partition attrs", "DB_ATTRIBUTES_FUN_014", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='key1=value1';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(attrs, func(i, j int) bool { return attrs[i].ID > attrs[j].ID })

			Try(db.ExecContext(ctx2, "drop table t1;"))
			Try(db.ExecContext(ctx2, "recover table t1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			sort.Slice(rules1, func(i, j int) bool { return rules1[i].ID > rules1[j].ID })
			Expect(rules1).Should(HaveLen(2))
			Expect(rules1[0].ID).To(Equal("schema/test/t1/p0"))
			Expect(rules1[1].ID).To(Equal("schema/test/t1"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[1].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key1=value1"`))
			Expect(rules1[1].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).To(Equal(rules1[0].Ranges))
			Expect(attrs[1].Ranges).To(Equal(rules1[1].Ranges))
		})
	})
	Context("Drop Clause", func() {
		TestCase("drop table partition", "DB_ATTRIBUTES_FUN_015", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(20000), partition p2 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='key=value';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))

			Try(db.ExecContext(ctx2, "alter table t1 drop partition p1;"))
			attrs = []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))

			Try(db.ExecContext(ctx2, "alter table t1 drop partition p0;"))
			attrs = []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(0))
		})
	})
	Context("Exchange Clause", func() {
		TestCase("exchange partition only with partition attribute", "DB_ATTRIBUTES_FUN_017", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				Try(db.ExecContext(ctx2, "set @@tidb_enable_exchange_partition=0;"))
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
				client.DeleteLabelRule("schema/test/t1")
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(20000), partition p2 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "set @@tidb_enable_exchange_partition=1;"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Try(db.ExecContext(ctx2, "alter table t2 exchange partition p0 with table t1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules1).Should(HaveLen(1))
			Expect(rules1[0].ID).To(Equal("schema/test/t1"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).To(Equal(rules1[0].Ranges))
		})
		TestCase("exchange partition only with table attribute", "DB_ATTRIBUTES_FUN_018", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				Try(db.ExecContext(ctx2, "set @@tidb_enable_exchange_partition=0;"))
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
				client.DeleteLabelRule("schema/test/t2/p0")
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(20000), partition p2 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "set @@tidb_enable_exchange_partition=1;"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Try(db.ExecContext(ctx2, "alter table t2 exchange partition p0 with table t1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules1).Should(HaveLen(1))
			Expect(rules1[0].ID).To(Equal("schema/test/t2/p0"))
			Expect(rules1[0].Type).To(Equal("key-range"))
			Expect(rules1[0].Attributes).To(Equal(`"key=value"`))
			Expect(attrs[0].Ranges).To(Equal(rules1[0].Ranges))
		})
		TestCase("exchange partition with both table and partition attrs", "DB_ATTRIBUTES_FUN_019", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				Try(db.ExecContext(ctx2, "set @@tidb_enable_exchange_partition=0;"))
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t2/p0")
			}()
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(20000), partition p2 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='key=value';"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='key1=value1';"))
			Try(db.ExecContext(ctx2, "set @@tidb_enable_exchange_partition=1;"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(2))
			Try(db.ExecContext(ctx2, "alter table t2 exchange partition p0 with table t1;"))
			rules1 := []Attributes{}
			Expect(db.Select(&rules1, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(rules1).Should(HaveLen(2))
			sort.Slice(attrs, func(i, j int) bool { return attrs[i].ID > attrs[j].ID })
			sort.Slice(rules1, func(i, j int) bool { return rules1[i].ID > rules1[j].ID })
			Expect(attrs[0].Attributes).NotTo(Equal(rules1[0].Attributes))
			Expect(attrs[1].Attributes).NotTo(Equal(rules1[1].Attributes))
			Expect(attrs[0].Ranges).NotTo(Equal(rules1[0].Ranges))
			Expect(attrs[1].Ranges).NotTo(Equal(rules1[1].Ranges))
		})
	})
})

var _ = Describe("Merge functionality testing during DDL", func() {
	var (
		tc     resource.TiDBCluster
		dbDSN  string
		db     *sqlx.DB
		client *pd.Client
	)

	ctx := suiteTestCtx
	BeforeEach(func() {
		r := ctx.Resource("tc")
		tc = r.(resource.TiDBCluster)
		dbURL := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
		dbDSN = fmt.Sprintf("root@tcp(%s)/test", dbURL.Host)
		db = Try(sqlx.Open("mysql", dbDSN)).(*sqlx.DB)
		client = Try(pd.NewPDClient(tc)).(*pd.Client)
	})

	AfterEach(func() {
		db.Close()
	})

	Context("Table Merge", func() {
		TestCase("not merge table regions", "DB_ATTRIBUTES_FUN_004", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(client.SetConfig("split-merge-interval", "1s"))
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=deny';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Expect(attrs[0].ID).To(Equal("schema/test/t1"))
			Expect(attrs[0].Attributes).To(Equal(`"merge_option=deny"`))
			Try(db.ExecContext(ctx2, "split table t1 between (0) and (1000000000) regions 16;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(16))
			// test default keyword
			Try(db.ExecContext(ctx2, "alter table t1 attributes=default;"))
			time.Sleep(3 * time.Second)
			rows1, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows1.Close()
			count = 0
			for rows1.Next() {
				count += 1
			}
			Expect(count).To(Equal(1))
		})
		TestCase("not merge index regions", "DB_ATTRIBUTES_FUN_022", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(client.SetConfig("split-merge-interval", "1s"))
			Try(db.ExecContext(ctx2, "create table t1(col1 char(10), col2 char(10), index(col1));"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=deny';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Expect(attrs[0].ID).To(Equal("schema/test/t1"))
			Expect(attrs[0].Attributes).To(Equal(`"merge_option=deny"`))
			Try(db.ExecContext(ctx2, "split table t1 index col1 between ('a') and ('z') regions 25;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(26))
			Try(db.ExecContext(ctx2, "alter table t1 attributes=default;"))
			time.Sleep(10 * time.Second)
			rows1, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows1.Close()
			count = 0
			for rows1.Next() {
				count += 1
			}
			Expect(count).To(Equal(1))
		})
		TestCase("no merge table after truncate table", "DB_ATTRIBUTES_FUN_005", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(Try(client.SetConfig("split-merge-interval", "1s")))
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=deny';"))
			// table ID will change
			Try(db.ExecContext(ctx2, "truncate table t1;"))
			Try(db.ExecContext(ctx2, "split table t1 between (0) and (1000000000) regions 16;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(16))
		})
		TestCase("no merge table after rename table", "DB_ATTRIBUTES_FUN_009", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t2")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
				Try(Try(client.SetConfig("split-merge-interval", "1h")))
			}()
			Try(Try(client.SetConfig("split-merge-interval", "1s")))
			Try(db.ExecContext(ctx2, "create table t1 (c int);"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "split table t1 between (0) and (1000000000) regions 16;"))
			// table ID will not change
			Try(db.ExecContext(ctx2, "rename table t1 to t2;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t2 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(16))
		})
	})
	Context("Partition Merge", func() {
		TestCase("not merge partition", "DB_ATTRIBUTES_FUN_006", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(Try(client.SetConfig("split-merge-interval", "1s")))
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='merge_option=deny';"))
			attrs := []Attributes{}
			Expect(db.Select(&attrs, "select * from information_schema.attributes;")).Should(BeNil())
			Expect(attrs).Should(HaveLen(1))
			Expect(attrs[0].ID).To(Equal("schema/test/t1/p0"))
			Expect(attrs[0].Attributes).To(Equal(`"merge_option=deny"`))
			Try(db.ExecContext(ctx2, "split partition table t1 partition (p0,p1) between (0) and (100000) regions 4;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(5))
			// test default keyword
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes=default;"))
			time.Sleep(3 * time.Second)
			rows1, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows1.Close()
			count = 0
			for rows1.Next() {
				count += 1
			}
			Expect(count).To(Equal(1))
		})
		TestCase("not merge partition after truncate table", "DB_ATTRIBUTES_FUN_007", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t2/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(client.SetConfig("split-merge-interval", "1s"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "truncate table t2;"))
			Try(db.ExecContext(ctx2, "split partition table t2 partition (p0,p1) between (0) and (10000) regions 4;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t2 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(5))
		})
		TestCase("not merge partition after rename table", "DB_ATTRIBUTES_FUN_010", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t2/p0")
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t2;"))
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(client.SetConfig("split-merge-interval", "1s"))
			Try(db.ExecContext(ctx2, "create table t2 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t2 partition p0 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "split partition table t2 partition (p0,p1) between (0) and (10000) regions 4;"))
			Try(db.ExecContext(ctx2, "rename table t2 to t1;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(5))
		})
	})
	Context("Override", func() {
		TestCase("override the table attribute", "DB_ATTRIBUTES_FUN_020", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t1/p2")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(client.SetConfig("split-merge-interval", "1s"))
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than (20000), partition p2 values less than(MAXVALUE) );"))
			// table deny but partition allow
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p2 attributes='merge_option=allow';"))
			Try(db.ExecContext(ctx2, "split partition table t1 partition (p0,p1,p2) between (0) and (100000) regions 4;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(9))

			// table allow but partition deny
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=allow';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p2 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "split partition table t1 partition (p0,p1,p2) between (0) and (100000) regions 4;"))
			time.Sleep(3 * time.Second)
			rows, err = db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count = 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(6))
		})

		TestCase("reset the partition attribute", "DB_ATTRIBUTES_FUN_021", func() {
			ctx2, cancel := context.WithCancel(context.TODO())
			defer cancel()
			defer func() {
				client.DeleteLabelRule("schema/test/t1")
				client.DeleteLabelRule("schema/test/t1/p0")
				Try(db.ExecContext(ctx2, "drop table if exists t1;"))
				Try(client.SetConfig("split-merge-interval", "1h"))
			}()
			Try(client.SetConfig("split-merge-interval", "1s"))
			Try(db.ExecContext(ctx2, "create table t1 (c int) partition by range (c) ( partition p0 values less than (10000), partition p1 values less than(MAXVALUE) );"))
			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=allow';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes=default;"))
			Try(db.ExecContext(ctx2, "split partition table t1 partition (p0,p1) between (0) and (100000) regions 4;"))
			time.Sleep(3 * time.Second)
			rows, err := db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count := 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(2))

			Try(db.ExecContext(ctx2, "alter table t1 attributes='merge_option=deny';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes='merge_option=allow';"))
			Try(db.ExecContext(ctx2, "alter table t1 partition p0 attributes=default;"))
			Try(db.ExecContext(ctx2, "split partition table t1 partition (p0,p1) between (0) and (100000) regions 4;"))
			time.Sleep(3 * time.Second)
			rows, err = db.QueryContext(ctx2, "show table t1 regions;")
			Expect(err).Should(BeNil())
			defer rows.Close()
			count = 0
			for rows.Next() {
				count += 1
			}
			Expect(count).To(Equal(8))
		})
	})
})