package dialect

import (
	"bytes"
	"fmt"
	"regexp"
	"strings"

	"github.com/pingcap/tidb/pkg/parser"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/format"
	"github.com/pingcap/tidb/pkg/parser/opcode"
)

// funcMap 逻辑函数名 -> 目标方言函数名
type funcMap map[string]string

// RewriteOptions 表名和数据库名替换选项，可选
type RewriteOptions struct {
	// TableNameMap 源表名 -> 目标表名，支持多表（如 slow_query_logs -> slow_query_logs_xxx, topsql -> topsql_xxx）
	TableNameMap map[string]string
	// DatabaseNameMap 源数据库名 -> 目标数据库名（如 clinic_data_proxy -> clinic_data_proxy_dev）
	DatabaseNameMap map[string]string
	// PartitionColumns 分区列，非空时在 rewrite 后基于 AST 校验 WHERE 中是否包含过滤条件
	PartitionColumns []string
}

// RewriteForEngine 根据 engine 使用对应规则重写 SQL，返回重写后的 SQL
// engine: athena | bigquery | maxcompute | synapse
// 若 opts 非空，会基于 AST 应用表名和数据库名替换
func RewriteForEngine(sql, engine string, opts *RewriteOptions) (rewritten string, err error) {
	cfg := configForEngine(engine)
	return rewriteSQL(sql, cfg, opts)
}

// ValidatePartitionFiltersForAllTableScans requires every SELECT that reads a
// physical table to carry its own partition filter. This stricter validation
// is used by the canonical public facade so a filtered UNION branch or outer
// CTE query cannot hide an unbounded table scan in another branch.
func ValidatePartitionFiltersForAllTableScans(sql string, partitionColumns []string) error {
	if len(partitionColumns) == 0 {
		return nil
	}

	p := parser.New()
	stmt, err := p.ParseOneStmt(sql, "", "")
	if err != nil {
		return fmt.Errorf("parse statement: %w", err)
	}
	if !isSupportedSelectLike(stmt) {
		return fmt.Errorf("only SELECT and WITH (CTE) queries are supported")
	}

	validator := &partitionTableScanVisitor{partitionColumns: partitionColumns}
	stmt.Accept(validator)
	return validator.err
}

func configForEngine(engine string) *dialectConfig {
	switch strings.ToLower(engine) {
	case "athena":
		return PrestoConfig // Athena 使用 Presto/Trino
	case "bigquery":
		return BigQueryConfig
	case "maxcompute":
		return MaxComputeConfig
	case "synapse":
		return SynapseConfig
	default:
		return nil // duckdb 等：不做转换
	}
}

func restoreFlags(cfg *dialectConfig) format.RestoreFlags {
	// 与历史 xsqlparser.ToSQLString 一致，关键字小写，便于与下游/测试对齐
	// RestoreStringWithoutCharset：输出普通 '...'，不带 _utf8mb4 前缀（见 test_driver ValueExpr.Restore）
	base := format.RestoreStringSingleQuotes | format.RestoreKeyWordLowercase | format.RestoreStringWithoutCharset
	if cfg == nil {
		return base | format.RestoreNameBackQuotes
	}
	switch cfg.identQuote {
	case `"`:
		return base | format.RestoreNameDoubleQuotes
	default:
		return base | format.RestoreNameBackQuotes
	}
}

func restoreStmt(stmt ast.StmtNode, cfg *dialectConfig) (string, error) {
	var buf bytes.Buffer
	ctx := format.NewRestoreCtx(restoreFlags(cfg), &buf)
	if err := stmt.Restore(ctx); err != nil {
		return "", fmt.Errorf("restore SQL: %w", err)
	}
	return buf.String(), nil
}

// rewriteSQL parses sql, applies dialect-specific transformations and optionally table/database replacement via AST.
func rewriteSQL(sql string, cfg *dialectConfig, opts *RewriteOptions) (string, error) {
	needParse := cfg != nil || opts != nil
	if !needParse {
		return sql, nil
	}

	p := parser.New()
	stmt, err := p.ParseOneStmt(sql, "", "")
	if err != nil {
		return "", fmt.Errorf("parse statement: %w", err)
	}

	if !isSupportedSelectLike(stmt) {
		return "", fmt.Errorf("only SELECT and WITH (CTE) queries are supported")
	}

	if opts != nil {
		tv := &tableRewriteVisitor{opts: opts}
		stmt.Accept(tv)
		if len(opts.PartitionColumns) > 0 {
			if err := validatePartitionInAST(stmt, opts.PartitionColumns); err != nil {
				return "", err
			}
		}
	}
	if cfg != nil {
		dv := &dialectVisitor{cfg: cfg}
		stmt.Accept(dv)
	}

	rewritten, err := restoreStmt(stmt, cfg)
	if err != nil {
		return "", err
	}

	// Apply CAST type replacements (e.g. "as signed)" → "as INT64)" for BigQuery)
	if cfg != nil {
		rewritten = applyCastTypeReplaces(rewritten, cfg.castTypeReplaces)
	}
	if cfg == SynapseConfig {
		rewritten, err = rewriteLimitToTop(rewritten)
		if err != nil {
			return "", err
		}
	}

	return rewritten, nil
}

func isSupportedSelectLike(stmt ast.StmtNode) bool {
	switch stmt.(type) {
	case *ast.SelectStmt, *ast.SetOprStmt:
		return true
	default:
		return false
	}
}

// --- table rewrite visitor ---

type tableRewriteVisitor struct {
	opts *RewriteOptions
}

func (v *tableRewriteVisitor) Enter(n ast.Node) (ast.Node, bool) {
	switch x := n.(type) {
	case *ast.TableName:
		rewriteTableName(x, v.opts)
	case *ast.ColumnName:
		rewriteColumnNameForTable(x, v.opts)
	case *ast.WildCardField:
		rewriteWildCardField(x, v.opts)
	}
	return n, false
}

func (v *tableRewriteVisitor) Leave(n ast.Node) (ast.Node, bool) {
	return n, true
}

func rewriteTableName(tn *ast.TableName, opts *RewriteOptions) {
	if tn == nil || tn.IsAlias {
		return
	}
	if len(opts.DatabaseNameMap) > 0 && tn.Schema.String() != "" {
		v := trimQuotes(tn.Schema.String())
		if to, ok := opts.DatabaseNameMap[v]; ok {
			tn.Schema = ast.NewCIStr(to)
		}
	}
	if len(opts.TableNameMap) > 0 && tn.Name.String() != "" {
		v := trimQuotes(strings.ToLower(tn.Name.String()))
		if to, ok := opts.TableNameMap[v]; ok {
			tn.Name = ast.NewCIStr(to)
		}
	}
}

func rewriteColumnNameForTable(cn *ast.ColumnName, opts *RewriteOptions) {
	if cn == nil {
		return
	}
	// db.table.col
	if cn.Schema.String() != "" && cn.Table.String() != "" && cn.Name.String() != "" {
		if len(opts.DatabaseNameMap) > 0 {
			v := trimQuotes(cn.Schema.String())
			if to, ok := opts.DatabaseNameMap[v]; ok {
				cn.Schema = ast.NewCIStr(to)
			}
		}
		if len(opts.TableNameMap) > 0 {
			v := strings.ToLower(trimQuotes(cn.Table.String()))
			if to, ok := opts.TableNameMap[v]; ok {
				cn.Table = ast.NewCIStr(to)
			}
		}
		return
	}
	// table.col
	if cn.Schema.String() == "" && cn.Table.String() != "" && cn.Name.String() != "" {
		if len(opts.TableNameMap) > 0 {
			v := strings.ToLower(trimQuotes(cn.Table.String()))
			if to, ok := opts.TableNameMap[v]; ok {
				cn.Table = ast.NewCIStr(to)
			}
		}
	}
}

func rewriteWildCardField(w *ast.WildCardField, opts *RewriteOptions) {
	if w == nil {
		return
	}
	if len(opts.DatabaseNameMap) > 0 && w.Schema.String() != "" {
		v := trimQuotes(w.Schema.String())
		if to, ok := opts.DatabaseNameMap[v]; ok {
			w.Schema = ast.NewCIStr(to)
		}
	}
	if len(opts.TableNameMap) > 0 && w.Table.String() != "" {
		v := strings.ToLower(trimQuotes(w.Table.String()))
		if to, ok := opts.TableNameMap[v]; ok {
			w.Table = ast.NewCIStr(to)
		}
	}
}

// --- dialect visitor ---

type dialectVisitor struct {
	cfg *dialectConfig
}

func (v *dialectVisitor) Enter(n ast.Node) (ast.Node, bool) {
	switch x := n.(type) {
	case *ast.FuncCallExpr:
		applyFuncDialect(x, v.cfg)
	case *ast.AggregateFuncExpr:
		applyAggregateFuncDialect(x, v.cfg)
	case *ast.ColumnName:
		applyColumnNameDialect(x, v.cfg)
	case *ast.TableName:
		applyTableNameDialect(x, v.cfg)
	case *ast.WildCardField:
		applyWildCardDialect(x, v.cfg)
	}
	return n, false
}

func (v *dialectVisitor) Leave(n ast.Node) (ast.Node, bool) {
	return n, true
}

func applyAggregateFuncDialect(fn *ast.AggregateFuncExpr, cfg *dialectConfig) {
	if fn == nil || fn.F == "" {
		return
	}
	lower := strings.ToLower(fn.F)
	if target, ok := cfg.funcMap[lower]; ok {
		fn.F = target
	}
	targetLower := strings.ToLower(fn.F)
	if perm, ok := cfg.paramReorder[targetLower]; ok && len(perm) >= len(fn.Args) && len(fn.Args) > 0 {
		reordered := make([]ast.ExprNode, len(fn.Args))
		for i := range fn.Args {
			if i < len(perm) {
				p := perm[i]
				if p < len(fn.Args) {
					reordered[i] = fn.Args[p]
				} else {
					reordered[i] = fn.Args[i]
				}
			} else {
				reordered[i] = fn.Args[i]
			}
		}
		fn.Args = reordered
	}
}

func applyFuncDialect(fn *ast.FuncCallExpr, cfg *dialectConfig) {
	if fn == nil || fn.FnName.L == "" {
		return
	}
	lower := strings.ToLower(fn.FnName.L)
	if target, ok := cfg.funcMap[lower]; ok {
		fn.FnName = ast.NewCIStr(target)
	}
	targetLower := strings.ToLower(fn.FnName.L)
	if perm, ok := cfg.paramReorder[targetLower]; ok && len(perm) >= len(fn.Args) && len(fn.Args) > 0 {
		reordered := make([]ast.ExprNode, len(fn.Args))
		for i := range fn.Args {
			if i < len(perm) {
				p := perm[i]
				if p < len(fn.Args) {
					reordered[i] = fn.Args[p]
				} else {
					reordered[i] = fn.Args[i]
				}
			} else {
				reordered[i] = fn.Args[i]
			}
		}
		fn.Args = reordered
	}
}

func cistrFromQuoteIdent(s string, cfg *dialectConfig) ast.CIStr {
	q := cfg.quoteIdent(s)
	inner := trimQuotes(q)
	return ast.NewCIStr(inner)
}

func applyColumnNameDialect(cn *ast.ColumnName, cfg *dialectConfig) {
	if cn == nil {
		return
	}
	if cn.Schema.String() != "" {
		cn.Schema = cistrFromQuoteIdent(cn.Schema.String(), cfg)
	}
	if cn.Table.String() != "" {
		cn.Table = cistrFromQuoteIdent(cn.Table.String(), cfg)
	}
	if cn.Name.String() != "" {
		cn.Name = cistrFromQuoteIdent(cn.Name.String(), cfg)
	}
}

func applyTableNameDialect(tn *ast.TableName, cfg *dialectConfig) {
	if tn == nil {
		return
	}
	if tn.Schema.String() != "" {
		tn.Schema = cistrFromQuoteIdent(tn.Schema.String(), cfg)
	}
	if tn.Name.String() != "" {
		tn.Name = cistrFromQuoteIdent(tn.Name.String(), cfg)
	}
}

func applyWildCardDialect(w *ast.WildCardField, cfg *dialectConfig) {
	if w == nil {
		return
	}
	if w.Schema.String() != "" {
		w.Schema = cistrFromQuoteIdent(w.Schema.String(), cfg)
	}
	if w.Table.String() != "" {
		w.Table = cistrFromQuoteIdent(w.Table.String(), cfg)
	}
}

// applyCastTypeReplaces 替换 SQL 中 CAST 表达式的类型关键字
// TiDB parser 将 CAST(x AS SIGNED) 序列化为 "cast(x as signed)"（RestoreKeyWordLowercase），
// 目标引擎可能需要不同的类型名（如 BigQuery 的 INT64、Presto 的 BIGINT）
func applyCastTypeReplaces(sql string, replaces []castTypeReplace) string {
	for _, r := range replaces {
		sql = strings.ReplaceAll(sql, " as "+r.from+")", " as "+r.to+")")
	}
	return sql
}

var simpleLimitSuffixRE = regexp.MustCompile(`(?is)\s+limit\s+([0-9]+)\s*$`)
var lastSelectBeforeLimitRE = regexp.MustCompile(`(?is)^(.*\bselect\s+)(distinct\s+)?(.*)$`)

func rewriteLimitToTop(sql string) (string, error) {
	matches := simpleLimitSuffixRE.FindStringSubmatchIndex(sql)
	if matches == nil {
		if regexp.MustCompile(`(?is)\blimit\s+`).MatchString(sql) {
			return "", fmt.Errorf("synapse dialect only supports a single trailing LIMIT n")
		}
		return sql, nil
	}
	limitValue := sql[matches[2]:matches[3]]
	withoutLimit := strings.TrimSpace(sql[:matches[0]])
	selectMatch := lastSelectBeforeLimitRE.FindStringSubmatchIndex(withoutLimit)
	if selectMatch == nil {
		return "", fmt.Errorf("synapse dialect failed to rewrite LIMIT because no SELECT was found")
	}
	distinctStart := selectMatch[2]
	distinctEnd := selectMatch[3]
	if distinctStart >= 0 && distinctEnd >= 0 {
		return withoutLimit[:distinctEnd] + "top " + limitValue + " " + withoutLimit[distinctEnd:], nil
	}
	insertAt := selectMatch[3]
	return withoutLimit[:insertAt] + "top " + limitValue + " " + withoutLimit[insertAt:], nil
}

// ensureTrailingSemicolon 确保 SQL 以分号结尾（MaxCompute 要求）
func ensureTrailingSemicolon(sql string) string {
	sql = strings.TrimSpace(sql)
	if sql == "" {
		return sql
	}
	if !strings.HasSuffix(sql, ";") {
		return sql + ";"
	}
	return sql
}

// trimQuotes 去除标识符两侧的引号（"x" 或 `x`）
func trimQuotes(s string) string {
	if len(s) < 2 {
		return s
	}
	if (s[0] == '"' && s[len(s)-1] == '"') || (s[0] == '`' && s[len(s)-1] == '`') {
		return s[1 : len(s)-1]
	}
	return s
}

// validatePartitionInAST 校验查询中所有 WHERE 子句是否包含对每个分区列的过滤（col = literal 或 col IN (literals)）
func validatePartitionInAST(stmt ast.StmtNode, partitionColumns []string) error {
	var whereExprs []ast.ExprNode
	collectWhereExprs(stmt, &whereExprs)
	if len(whereExprs) == 0 {
		return fmt.Errorf("SQL validation failed: missing partition filter for %q. Please add WHERE condition like: WHERE %q = '...' or WHERE %q IN (...)", partitionColumns[0], partitionColumns[0], partitionColumns[0])
	}
	for _, col := range partitionColumns {
		found := false
		for _, we := range whereExprs {
			if hasPartitionFilter(we, col) {
				found = true
				break
			}
		}
		if !found {
			return fmt.Errorf("SQL validation failed: missing partition filter for %q. Please add WHERE condition like: WHERE %q = '...' or WHERE %q IN (...)", col, col, col)
		}
	}
	return nil
}

type partitionTableScanVisitor struct {
	partitionColumns []string
	cteScopes        []map[string]struct{}
	err              error
}

func (v *partitionTableScanVisitor) Enter(node ast.Node) (ast.Node, bool) {
	switch n := node.(type) {
	case *ast.SelectStmt:
		v.pushCTEScope()
	case *ast.SetOprStmt:
		v.pushCTEScope()
	case *ast.CommonTableExpression:
		if n.IsRecursive {
			v.addCurrentCTE(n.Name.L)
		}
	}
	return node, false
}

func (v *partitionTableScanVisitor) Leave(node ast.Node) (ast.Node, bool) {
	switch n := node.(type) {
	case *ast.CommonTableExpression:
		if !n.IsRecursive {
			v.addCurrentCTE(n.Name.L)
		}
	case *ast.SelectStmt:
		if v.err == nil {
			v.validateSelect(n, v.cteScopes[len(v.cteScopes)-1])
		}
		v.cteScopes = v.cteScopes[:len(v.cteScopes)-1]
	case *ast.SetOprStmt:
		v.cteScopes = v.cteScopes[:len(v.cteScopes)-1]
	}
	return node, true
}

func (v *partitionTableScanVisitor) pushCTEScope() {
	scope := make(map[string]struct{})
	if len(v.cteScopes) > 0 {
		for name := range v.cteScopes[len(v.cteScopes)-1] {
			scope[name] = struct{}{}
		}
	}
	v.cteScopes = append(v.cteScopes, scope)
}

func (v *partitionTableScanVisitor) addCurrentCTE(name string) {
	if len(v.cteScopes) > 0 {
		v.cteScopes[len(v.cteScopes)-1][name] = struct{}{}
	}
}

func (v *partitionTableScanVisitor) validateSelect(stmt *ast.SelectStmt, cteNames map[string]struct{}) {
	if stmt == nil || stmt.From == nil {
		return
	}
	scans := collectPhysicalTableScans(stmt.From.TableRefs, cteNames, nil)
	allowUnqualified := len(scans) == 1
	for _, scan := range scans {
		for _, column := range v.partitionColumns {
			if !hasPartitionFilterForScan(stmt.Where, column, scan.qualifier, allowUnqualified) {
				v.err = fmt.Errorf("SQL validation failed: every table scan must filter partition column %q", column)
				return
			}
		}
	}
}

type physicalTableScan struct {
	qualifier string
}

func collectPhysicalTableScans(node ast.ResultSetNode, cteNames map[string]struct{}, scans []physicalTableScan) []physicalTableScan {
	switch n := node.(type) {
	case *ast.Join:
		scans = collectPhysicalTableScans(n.Left, cteNames, scans)
		return collectPhysicalTableScans(n.Right, cteNames, scans)
	case *ast.TableSource:
		if table, ok := n.Source.(*ast.TableName); ok {
			if !isCTETableName(table, cteNames) {
				qualifier := table.Name.L
				if n.AsName.L != "" {
					qualifier = n.AsName.L
				}
				scans = append(scans, physicalTableScan{qualifier: qualifier})
			}
			return scans
		}
		return collectPhysicalTableScans(n.Source, cteNames, scans)
	case *ast.TableName:
		if !isCTETableName(n, cteNames) {
			scans = append(scans, physicalTableScan{qualifier: n.Name.L})
		}
		return scans
	case *ast.SelectStmt, *ast.SetOprStmt:
		// Nested queries are visited and validated independently.
		return scans
	default:
		return scans
	}
}

func isCTETableName(table *ast.TableName, cteNames map[string]struct{}) bool {
	if table.Schema.L != "" {
		return false
	}
	_, ok := cteNames[table.Name.L]
	return ok
}

func collectWhereExprs(stmt ast.StmtNode, out *[]ast.ExprNode) {
	switch s := stmt.(type) {
	case *ast.SelectStmt:
		collectSelectWhere(s, out)
	case *ast.SetOprStmt:
		collectSetOprWhere(s, out)
	}
}

func collectSelectWhere(s *ast.SelectStmt, out *[]ast.ExprNode) {
	if s.With != nil {
		for _, cte := range s.With.CTEs {
			if cte.Query != nil && cte.Query.Query != nil {
				collectFromResultSet(cte.Query.Query, out)
			}
		}
	}
	if s.Where != nil {
		*out = append(*out, s.Where)
	}
	if s.From != nil {
		collectFromJoin(s.From.TableRefs, out)
	}
}

func collectSetOprWhere(s *ast.SetOprStmt, out *[]ast.ExprNode) {
	if s.With != nil {
		for _, cte := range s.With.CTEs {
			if cte.Query != nil && cte.Query.Query != nil {
				collectFromResultSet(cte.Query.Query, out)
			}
		}
	}
	if s.SelectList != nil {
		collectFromSetOprList(s.SelectList, out)
	}
}

func collectFromSetOprList(s *ast.SetOprSelectList, out *[]ast.ExprNode) {
	if s.With != nil {
		for _, cte := range s.With.CTEs {
			if cte.Query != nil && cte.Query.Query != nil {
				collectFromResultSet(cte.Query.Query, out)
			}
		}
	}
	for _, node := range s.Selects {
		switch n := node.(type) {
		case *ast.SelectStmt:
			collectSelectWhere(n, out)
		case *ast.SetOprSelectList:
			collectFromSetOprList(n, out)
		}
	}
}

func collectFromJoin(j *ast.Join, out *[]ast.ExprNode) {
	if j == nil {
		return
	}
	collectFromResultSet(j.Left, out)
	if j.Right != nil {
		collectFromResultSet(j.Right, out)
	}
}

func collectFromResultSet(rs ast.ResultSetNode, out *[]ast.ExprNode) {
	switch r := rs.(type) {
	case *ast.SelectStmt:
		collectSelectWhere(r, out)
	case *ast.SetOprStmt:
		collectSetOprWhere(r, out)
	case *ast.TableSource:
		switch src := r.Source.(type) {
		case *ast.SelectStmt:
			collectSelectWhere(src, out)
		case *ast.SetOprStmt:
			collectSetOprWhere(src, out)
		case *ast.SubqueryExpr:
			if src.Query != nil {
				collectFromResultSet(src.Query, out)
			}
		}
	case *ast.Join:
		collectFromJoin(r, out)
	}
}

func peelExpr(e ast.ExprNode) ast.ExprNode {
	for {
		if p, ok := e.(*ast.ParenthesesExpr); ok {
			e = p.Expr
			continue
		}
		return e
	}
}

// hasPartitionFilter 判断表达式树中是否包含对 partitionCol 的有效过滤（= literal 或 IN (literals)）
func hasPartitionFilter(expr ast.ExprNode, partitionCol string) bool {
	if expr == nil {
		return false
	}
	expr = peelExpr(expr)
	colNorm := strings.ToLower(partitionCol)
	switch n := expr.(type) {
	case *ast.BinaryOperationExpr:
		switch n.Op {
		case opcode.EQ, opcode.GE, opcode.GT, opcode.LE, opcode.LT:
			return (isPartitionColRef(n.L, colNorm) && isLiteralExpr(n.R)) ||
				(isPartitionColRef(n.R, colNorm) && isLiteralExpr(n.L))
		case opcode.LogicAnd, opcode.LogicOr:
			return hasPartitionFilter(n.L, colNorm) || hasPartitionFilter(n.R, colNorm)
		}
		return false
	case *ast.PatternInExpr:
		if n.Not || n.Sel != nil {
			return false
		}
		return isPartitionColRef(n.Expr, colNorm) && allLiteralsExpr(n.List)
	case *ast.BetweenExpr:
		if n.Not {
			return false
		}
		return isPartitionColRef(n.Expr, colNorm) && isLiteralExpr(n.Left) && isLiteralExpr(n.Right)
	}
	return false
}

func hasPartitionFilterForScan(expr ast.ExprNode, partitionCol, qualifier string, allowUnqualified bool) bool {
	if expr == nil {
		return false
	}
	expr = peelExpr(expr)
	column := strings.ToLower(partitionCol)
	qualifier = strings.ToLower(qualifier)
	switch n := expr.(type) {
	case *ast.BinaryOperationExpr:
		switch n.Op {
		case opcode.EQ, opcode.GE, opcode.GT, opcode.LE, opcode.LT:
			return (isPartitionColRefForScan(n.L, column, qualifier, allowUnqualified) && isLiteralExpr(n.R)) ||
				(isPartitionColRefForScan(n.R, column, qualifier, allowUnqualified) && isLiteralExpr(n.L))
		case opcode.LogicAnd:
			return hasPartitionFilterForScan(n.L, column, qualifier, allowUnqualified) ||
				hasPartitionFilterForScan(n.R, column, qualifier, allowUnqualified)
		case opcode.LogicOr:
			return hasPartitionFilterForScan(n.L, column, qualifier, allowUnqualified) &&
				hasPartitionFilterForScan(n.R, column, qualifier, allowUnqualified)
		}
		return false
	case *ast.PatternInExpr:
		if n.Not || n.Sel != nil {
			return false
		}
		return isPartitionColRefForScan(n.Expr, column, qualifier, allowUnqualified) && allLiteralsExpr(n.List)
	case *ast.BetweenExpr:
		if n.Not {
			return false
		}
		return isPartitionColRefForScan(n.Expr, column, qualifier, allowUnqualified) &&
			isLiteralExpr(n.Left) && isLiteralExpr(n.Right)
	default:
		return false
	}
}

func isPartitionColRefForScan(expr ast.ExprNode, column, qualifier string, allowUnqualified bool) bool {
	if expr == nil {
		return false
	}
	expr = peelExpr(expr)
	ref, ok := expr.(*ast.ColumnNameExpr)
	if !ok || ref.Name == nil || ref.Name.Name.L != column {
		return false
	}
	if ref.Name.Table.L == "" {
		return allowUnqualified
	}
	return ref.Name.Table.L == qualifier
}

func isPartitionColRef(e ast.ExprNode, colNorm string) bool {
	if e == nil {
		return false
	}
	e = peelExpr(e)
	switch n := e.(type) {
	case *ast.ColumnNameExpr:
		if n.Name == nil {
			return false
		}
		return strings.ToLower(n.Name.Name.L) == colNorm
	}
	return false
}

func isLiteralExpr(e ast.ExprNode) bool {
	if e == nil {
		return false
	}
	_, ok := e.(ast.ValueExpr)
	return ok
}

func allLiteralsExpr(nodes []ast.ExprNode) bool {
	if len(nodes) == 0 {
		return false
	}
	for _, n := range nodes {
		if !isLiteralExpr(n) {
			return false
		}
	}
	return true
}