package pushdown

import (
	"context"
	"database/sql"
	"fmt"
	"math/rand"
	"strconv"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s" // register k8s impl

	"github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/testcase/tiflash/pkg"
)

var _ = Describe("Pushdown function test", func() {
	var (
		tc          resource.TiDBCluster
		tidbClient  *tidb.Client
		db          *sql.DB
		duration    time.Duration
		randomTimes int
	)
	Describe("Tiflash", func() {
		Context("test pushdown function", func() {
			BeforeEach(func() {
				tc = Try(suiteTestCtx.Resource("tc")).(resource.TiDBCluster)
				duration = Try(time.ParseDuration(suiteTestCtx.Param("duration", "1h"))).(time.Duration)
				db = Try(pkg.OpenDB(tc, "")).(*sql.DB)
				randomTimes = Try(strconv.Atoi(suiteTestCtx.Param("randomTimes", "1000"))).(int)
				log.Info("create database and table")
				Try(db.Exec("drop database if exists pushdown"))
				Try(db.Exec("create database pushdown"))
				db = Try(pkg.OpenDB(tc, "pushdown")).(*sql.DB)
				Try(db.Exec("create table pushdown_0 (id int primary key auto_increment, i1 int, i2 int, real1 real, real2 real,i3 int, real3 real, bool boolean DEFAULT true, y year DEFAULT '2022', t time DEFAULT '-838:59:59', d date DEFAULT '1000-01-01', dt datetime DEFAULT '1000-01-01 00:00:00', ts timestamp DEFAULT '1980-01-01 00:00:01', reg varchar(100) DEFAULT '^[A-Z0-9._%-]+@[A-Z0-9.-]+.[A-Z]{2,4}$你我他')"))
				tidbClient = tidb.NewClient(db)
				log.Info("alter tiflash replica")
				Try(tidbClient.AlterDatabaseSetTiFlashReplica("pushdown", 1, ""))
				Eventually(func() bool {
					return Try(tidbClient.IsDBTiFlashReplicaAvailable("pushdown")).(bool)
				}, 10*time.Minute, 3*time.Second).Should(BeTrue())
				Try(tidbClient.ShowClusterVersionInformation())
			})

			It("sql will success #001#", func() {
				ctx, _ := context.WithTimeout(suiteTestCtx, duration)
				for i := 0; i < randomTimes; i++ {
					Try(prepareDada(db))
					Try(mathPushdownFunctionTest(ctx, db))
					Try(operatorPushdownFunctionTest(ctx, db))
					Try(dateTimeFunctionTest(ctx, db))
					Try(regexpFunctionTest(ctx, db))
				}
			})
		})
	})
})

// TODO : 在 zto 负载中补充正则sql
// https://docs.google.com/document/d/1c0i4vemwT3hkME9cp-UGFC6InW_J3jOrP9yA_W2i9Eg/edit?n=Sprint_5_Enrich_TiFlash_push-down_functions_and_operators#heading=h.mwwlwg1mv1d5
// 对所有列进行运算符判断
func operatorPushdownFunctionTest(ctx context.Context, db *sql.DB) error {
	booleans := []string{"true", "false", "null"}
	columnNames := []string{"id", "i1", "i2", "i3", "real1", "real2", "real3", "bool", "y", "t", "d", "dt", "ts", "reg"}
	for _, boolean := range booleans {
		for _, columnName := range columnNames {
			s := fmt.Sprintf("select id from pushdown_0 where %s is %s", columnName, boolean)
			if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
				return err
			}
			s = fmt.Sprintf("select id from pushdown_0 where %s is not %s", columnName, boolean)
			if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
				return err
			}
		}
	}
	return nil
}

// 对类型「int, double」进行 greatest 和 least 计算
func mathPushdownFunctionTest(ctx context.Context, db *sql.DB) error {
	log.Info("Math Pushdown Function")
	if err := greatest(ctx, db); err != nil {
		return err
	}
	if err := least(ctx, db); err != nil {
		return err
	}
	return nil
}

// 对类型「year, time, date, datetime, timestamp」进行日期相关函数计算
func dateTimeFunctionTest(ctx context.Context, db *sql.DB) error {
	log.Info("Date Time Function")
	functionType := []string{"DAYNAME", "DAYOFMONTH", "DAYOFWEEK", "DAYOFYEAR", "LAST_DAY", "MONTHNAME"}
	colType := []string{"y", "t", "d", "dt", "ts"}
	for i := range functionType {
		for j := range colType {
			s := fmt.Sprintf("select %s(%s) from pushdown_0", functionType[i], colType[j])
			if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
				return err
			}
		}
	}
	return nil
}

// 对所有列进行正则匹配/不匹配，类型包括 [int, real, year, time, date, datetime, timestamp, varchar(100)]
func regexpFunctionTest(ctx context.Context, db *sql.DB) error {
	log.Info("Regexp Function")
	regexps := getRegexp()
	rows := []string{"i1", "i2", "real1", "real2", "i3", "real3", "y", "t", "d", "dt", "ts", "reg"}
	for i := range regexps {
		for _, row := range rows {
			s := fmt.Sprintf("select id, %s from pushdown_0 where %s REGEXP '%s'", row, row, regexps[i])
			if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
				return err
			}
			s = fmt.Sprintf("select id, %s from pushdown_0 where %s not REGEXP '%s'", row, row, regexps[i])
			if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
				return err
			}
		}
	}
	return nil
}

func greatest(ctx context.Context, db *sql.DB) error {
	rowsCombination := []string{"i1, i2", "real1, real2", "i3, real3"}
	// 列与列之间比较
	for _, row := range rowsCombination {
		s := fmt.Sprintf("select greatest(%s) from pushdown_0", row)
		if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
			return err
		}
	}
	// 列与常量之间比较
	var constNums []string
	constNums = append(constNums, getIntRange()...)
	constNums = append(constNums, getDoubleRange()...)
	for _, constNum := range constNums {
		s := fmt.Sprintf("select greatest(i1, %s) from pushdown_0", constNum)
		if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
			return err
		}
		s = fmt.Sprintf("select greatest(real1, %s) from pushdown_0", constNum)
		if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
			return err
		}
	}
	return nil
}

func least(ctx context.Context, db *sql.DB) error {
	rowsCombination := []string{"i1, i2", "real1, real2", "i3, real3"}
	// 列与列之间比较
	for _, row := range rowsCombination {
		s := fmt.Sprintf("select least(%s) from pushdown_0", row)
		if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
			return err
		}
	}
	// 列与常量之间比较
	var constNums []string
	constNums = append(constNums, getIntRange()...)
	constNums = append(constNums, getDoubleRange()...)
	for _, constNum := range constNums {
		s := fmt.Sprintf("select least(i1, %s) from pushdown_0", constNum)
		if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
			return err
		}
		s = fmt.Sprintf("select least(real1, %s) from pushdown_0", constNum)
		if err := pkg.DiffTiFlashTiKV(ctx, db, s, false, false); err != nil {
			return err
		}
	}
	return nil
}

func prepareDada(db *sql.DB) error {
	log.Info("insert data")
	intNums := getIntRange()
	doubleNums := getDoubleRange()
	booleanList := []string{"true", "false", "null"}
	var s string
	// 覆盖[int, int], [int, double], [double, double]，并且「bool」列随机插入 true\false\null
	// 所有 [int, double] 会 insert 到 varchar(100) 中
	for i := range intNums {
		boolean := booleanList[rand.Intn(len(booleanList))]
		for j := range intNums {
			s = fmt.Sprintf("insert into pushdown_0(i1, i2, i3, real1, real2, real3, reg, bool) VALUES (%s, %s, %s, %s, %s, %s, '%s', %s)", intNums[i], intNums[j], intNums[i], intNums[i], intNums[j], intNums[i], intNums[i], boolean)
			if _, err := db.Exec(s); err != nil {
				return err
			}
		}
	}
	for i := range doubleNums {
		boolean := booleanList[rand.Intn(len(booleanList))]
		for j := range doubleNums {
			s = fmt.Sprintf("insert into pushdown_0(real1, real2, real3, reg, bool) VALUES (%s, %s, %s, '%s', %s)", doubleNums[i], doubleNums[i], doubleNums[j], doubleNums[i], boolean)
			if _, err := db.Exec(s); err != nil {
				return err
			}
		}
	}
	for i := range intNums {
		boolean := booleanList[rand.Intn(len(booleanList))]
		for j := range doubleNums {
			s = fmt.Sprintf("insert into pushdown_0(i1, real1, i2, real2, i3, real3, reg, bool) VALUES (%s, %s, %s, %s, %s, %s, '%s', %s)", intNums[i], doubleNums[j], intNums[i], doubleNums[j], intNums[i], doubleNums[j], doubleNums[i], boolean)
			if _, err := db.Exec(s); err != nil {
				return err
			}
		}
	}
	// 所有 时间变量 会 insert 到 varchar(100) 中
	years := getYearRange()
	for _, year := range years {
		s = fmt.Sprintf("insert into pushdown_0(y, reg) value(%s, '%s')", year, year)
		if _, err := db.Exec(s); err != nil {
			return err
		}
	}
	times := getTimeRange()
	for _, t := range times {
		if t == "NULL" {
			s = fmt.Sprintf("insert into pushdown_0(t, reg) value(%s, '%s')", t, t)
		} else {
			s = fmt.Sprintf("insert into pushdown_0(t, reg) value('%s', '%s')", t, t)
		}
		if _, err := db.Exec(s); err != nil {
			return err
		}
	}
	dates := getDateRange()
	for _, date := range dates {
		if date == "NULL" {
			s = fmt.Sprintf("insert into pushdown_0(d, reg) value(%s,'%s')", date, date)
		} else {
			s = fmt.Sprintf("insert into pushdown_0(d, reg) value('%s','%s')", date, date)
		}

		if _, err := db.Exec(s); err != nil {
			return err
		}
	}
	datetimes := getDatetimeRange()
	for _, datetime := range datetimes {
		if datetime == "NULL" {
			s = fmt.Sprintf("insert into pushdown_0(dt, reg) value(%s,'%s')", datetime, datetime)
		} else {
			s = fmt.Sprintf("insert into pushdown_0(dt, reg) value('%s','%s')", datetime, datetime)
		}
		if _, err := db.Exec(s); err != nil {
			return err
		}
	}
	timestamps := getTimestampRange()
	for _, timestamp := range timestamps {
		if timestamp == "NULL" {
			s = fmt.Sprintf("insert into pushdown_0(ts, reg) value(%s,'%s')", timestamp, timestamp)
		} else {
			s = fmt.Sprintf("insert into pushdown_0(ts, reg) value('%s','%s')", timestamp, timestamp)
		}
		if _, err := db.Exec(s); err != nil {
			return err
		}
	}
	regexps := getRegRange()
	regexps = append(regexps, getRegexp()...)
	for _, regexp := range regexps {
		s := fmt.Sprintf("insert into pushdown_0(reg) value('%s')", regexp)
		if _, err := db.Exec(s); err != nil {
			return err
		}
	}
	return nil
}

func getIntRange() []string {
	return []string{"NULL", "0", "1", "-1", "1.1", "-1.1", "2147483647", "2147483646", "-2147483648", "-2147483647", "1234567890", "-1234567890", "123456.7890", "-123456.7890", "1.0000000000000002"}
}
func getDoubleRange() []string {
	return []string{"NULL", "0.0", "1.1", "-1.1", "2147483647", "2147483646", "-2147483648", "-2147483647", "1234567890", "-1234567890", "-1.7976931348623157e308", "1.7976931348623157e308", "2.2250738585072014e-308", "-2.2250738585072014e-308", "123456.7890", "-123456.7890", "1.0000000000000002", "4.9406564584124654e-324", "2.2250738585072009e-308", "-2.2250738585072009e-308"}
}
func getYearRange() []string {
	return []string{"NULL", "1901", "1902", "2155", "2154", "2000", "2100"}
}
func getTimeRange() []string {
	return []string{"NULL", "-838:59:59", "-838:59:58", "838:59:58", "838:59:59", "0:0:0", "59:0:0", "0:59:0", "0:0:59", "123:59:59", "123:0:0"}
}
func getDateRange() []string {
	return []string{"NULL", "1000-01-01", "1000-01-02", "9999-12-3", "9999-12-2", "1234-01-01", "2022-1-1", "2022-1-31", "2022-2-1", "2022-2-28", "2022-3-1", "2022-3-31", "2022-4-1", "2022-4-30", "2022-5-1", "2022-5-31", "2022-6-1", "2022-6-30", "2022-7-1", "2022-7-31", "2022-8-1", "2022-8-31", "2022-9-1", "2022-9-30", "2022-10-1", "2022-10-31", "2022-11-1", "2022-11-30", "2022-12-1", "2022-12-30", "2020-2-29"}
}
func getDatetimeRange() []string {
	return []string{"NULL", "1000-01-01 00:00:00", "1000-01-02 23:59:59", "9999-12-3 00:00:00", "9999-12-2 23:59:59", "1234-01-01 00:00:00", "2022-1-1 23:59:59", "2022-1-31 00:00:00", "2022-2-1 23:59:59", "2022-2-28 00:00:00", "2022-3-1 23:59:59", "2022-3-31 00:00:00", "2022-4-1 23:59:59", "2022-4-30 00:00:00", "2022-5-1 23:59:59", "2022-5-31 00:00:00", "2022-6-1 23:59:59", "2022-6-30 00:00:00", "2022-7-1 23:59:59", "2022-7-31 00:00:00", "2022-8-1 23:59:59", "2022-8-31 00:00:00", "2022-9-1 23:59:59", "2022-9-30 00:00:00", "2022-10-1 23:59:59", "2022-10-31 00:00:00", "2022-11-1 00:00:00", "2022-11-30 23:59:59", "2022-12-1 00:00:00", "2022-12-30 23:59:59", "2020-2-29 00:00:00"}
}
func getTimestampRange() []string {
	return []string{"NULL", "1980-01-01 00:00:01", "1980-01-02 00:00:01", "1970-01-01 00:00:01", "2038-01-19 03:14:07", "2038-01-19 03:14:06", "2022-1-1 23:59:59", "2022-1-31 00:00:00", "2022-2-1 23:59:59", "2022-2-28 00:00:00", "2022-3-1 23:59:59", "2022-3-31 00:00:00", "2022-4-1 23:59:59", "2022-4-30 00:00:00", "2022-5-1 23:59:59", "2022-5-31 00:00:00", "2022-6-1 23:59:59", "2022-6-30 00:00:00", "2022-7-1 23:59:59", "2022-7-31 00:00:00", "2022-8-1 23:59:59", "2022-8-31 00:00:00", "2022-9-1 23:59:59", "2022-9-30 00:00:00", "2022-10-1 23:59:59", "2022-10-31 00:00:00", "2022-11-1 00:00:00", "2022-11-30 23:59:59", "2022-12-1 00:00:00", "2022-12-30 23:59:59", "2020-2-29 00:00:00"}
}
func getRegRange() []string {
	return []string{"NULL", " ", "aaaaaa", "AAAAAA", "abcdef@pingcap.com", "123@pingcap.com", "a", "A", "abcdefghijklmnopqrstuvwxyz", "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abc123", "123abc", "ABC123", "123ABC", "a1b2c3", "A1B2C3", "a a", "1 1", "1234567891011121314151617181920212", "-123", "123.456", "-123.456", "0.123456789", "-0.123456789", "你我他。", "aeiou123", "aeiouok", "abcabcabc", "abcabcabcd", "0x00", "0xff"}
}

// 正则表达式 : email 地址、 以 a 开头、以 a 结尾，a >= 0，a > 0，123 = 0|1，a开头+k 结尾，出现 a，出现 a 或 b，任意字母
func getRegexp() []string {
	return []string{"^[A-Z0-9._%-]+@[A-Z0-9.-]+.[A-Z]{2,4}$", "^a", "a$", "a*", "a+", "123?", "a", "^[a]|k$", "a", "a|b", "(abc)*", "a{1,}", "a{0,}", "[a-zA-Z]", "[:alnum:]", "[:alpha:]", "[:blank:]", "[:cntrl:]", "[:digit:]", "[:graph:]", "[:lower:]", "[:print:]", "[:punct:]", "[:space:]", "[:upper:]", "[:xdigit:]"}
}