package executor

import (
	"context"
	"fmt"
	"sort"
	"strings"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

const (
	defaultIndexScanThreshold = 1000
	defaultRangeEstimate      = 10000
	maxOrBranches             = 16
)

type condKind uint8

const (
	condEq condKind = iota
	condRange
)

type condInfo struct {
	eq          bool
	rng         bool
	inCount     int
	unsupported bool
}

type indexMeta struct {
	name    string
	columns []string
	unique  bool
	primary bool
}

type orderBySpec struct {
	column    string
	direction string
}

type queryIndexInfo struct {
	requiredColumns map[string]struct{}
	hasWildcard     bool
	orderBy         []orderBySpec
	orderByActive   bool
}

type indexCandidate struct {
	meta          indexMeta
	accessCols    int
	eqPrefix      int
	hasRange      bool
	fullMatch     bool
	pointRanges   int
	covering      bool
	needBackTable bool
	orderByMatch  bool
	filterCols    int
}

type orPatternInfo struct {
	columns []string
}

func (e *selectPlanner) selectIndexHint(ctx context.Context, user, password string, stmt *parser.SelectStatement) (string, error) {
	if stmt == nil || stmt.From == nil || stmt.From.Schema == "" || stmt.From.Name == "" {
		return "", nil
	}
	if stmt.Options.AllowFiltering {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "allow_filtering"})
		return "", nil
	}
	if e.schemaManager == nil {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "error", Rule: "no_schema_manager"})
		return "", fmt.Errorf("schema manager is not configured")
	}
	indexes, err := e.schemaManager.IndexesForTable(ctx, user, password, stmt.From.Schema, stmt.From.Name)
	if err != nil {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "error", Rule: "index_lookup"})
		return "", err
	}
	if len(indexes) == 0 {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "no_index"})
		return "", fmt.Errorf("no indexes available; use ALLOW FILTERING")
	}

	conds, disqualify, orPattern, err := extractIndexConditions(stmt.Where)
	if err != nil {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "error", Rule: "condition_extract"})
		return "", err
	}
	if disqualify {
		if best, rule, ok, err := selectBestIndexForOR(stmt, indexes); err != nil {
			metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "error", Rule: "or_expand"})
			return "", err
		} else if ok {
			metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{
				Outcome:   "selected_or",
				Rule:      rule,
				IndexType: indexTypeLabel(best.meta),
				Covering:  best.covering,
				OrderBy:   best.orderByMatch,
				BackTable: best.needBackTable,
			})
			return best.meta.name, nil
		}
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "or_disallowed"})
		return "", fmt.Errorf("OR conditions require ALLOW FILTERING")
	}
	if hasUnsupportedConditions(conds) {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "unsupported_where"})
		return "", fmt.Errorf("unsupported WHERE conditions require ALLOW FILTERING")
	}

	queryInfo := buildQueryIndexInfo(stmt)
	eqCols, rangeCols := extractConditionColumns(conds)
	groupByCols := groupByColumns(stmt)

	metas := buildIndexMetas(indexes)
	if orPattern != nil && len(orPattern.columns) > 0 {
		metas = filterIndexMetasForOrPattern(metas, orPattern.columns)
	}
	eligible := filterIndexMetas(metas, eqCols, rangeCols, groupByCols)
	if len(eligible) == 0 {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "no_match"})
		return "", fmt.Errorf("query requires ALLOW FILTERING or an indexed predicate")
	}

	candidates := buildIndexCandidatesFromMetas(eligible, conds, queryInfo)
	if len(candidates) == 0 {
		metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{Outcome: "no_match"})
		return "", fmt.Errorf("query requires ALLOW FILTERING or an indexed predicate")
	}
	best, rule := chooseBestIndex(candidates, queryInfo)
	metrics.RecordIndexSelection(ctx, metrics.IndexSelectionInfo{
		Outcome:   "selected",
		Rule:      rule,
		IndexType: indexTypeLabel(best.meta),
		Covering:  best.covering,
		OrderBy:   best.orderByMatch,
		BackTable: best.needBackTable,
	})
	return best.meta.name, nil
}

func buildQueryIndexInfo(stmt *parser.SelectStatement) queryIndexInfo {
	info := queryIndexInfo{
		requiredColumns: make(map[string]struct{}),
	}
	if stmt == nil {
		return info
	}
	for _, expr := range stmt.Columns {
		if _, ok := expr.(*parser.StarExpr); ok {
			info.hasWildcard = true
			continue
		}
		collectExprColumns(expr, info.requiredColumns)
	}
	if stmt.Where != nil {
		collectExprColumns(stmt.Where, info.requiredColumns)
	}
	for _, expr := range stmt.GroupBy {
		collectExprColumns(expr, info.requiredColumns)
	}
	for _, ob := range stmt.Options.OrderBy {
		col := strings.ToLower(ob.Column)
		info.requiredColumns[col] = struct{}{}
		info.orderBy = append(info.orderBy, orderBySpec{
			column:    col,
			direction: strings.ToUpper(strings.TrimSpace(ob.Direction)),
		})
	}
	info.orderByActive = len(info.orderBy) > 0
	return info
}

func collectExprColumns(expr parser.Expression, out map[string]struct{}) {
	if expr == nil {
		return
	}
	refs := make(map[string]columnRef)
	collectColumnRefsInto(expr, refs)
	for _, ref := range refs {
		name := strings.ToLower(ref.name)
		if name != "" {
			out[name] = struct{}{}
		}
	}
}

func extractIndexConditions(expr parser.Expression) (map[string]*condInfo, bool, *orPatternInfo, error) {
	conds := make(map[string]*condInfo)
	disqualify := false
	var orPattern *orPatternInfo
	var walk func(parser.Expression) error
	walk = func(e parser.Expression) error {
		if e == nil {
			return nil
		}
		switch node := e.(type) {
		case *parser.BinaryExpr:
			op := strings.ToUpper(strings.TrimSpace(node.Op))
			switch op {
			case "AND":
				if err := walk(node.Left); err != nil {
					return err
				}
				if err := walk(node.Right); err != nil {
					return err
				}
				return nil
			case "OR":
				col, count, ok, err := extractOrEqPredicate(node)
				if err != nil {
					return err
				}
				if !ok {
					cols, ok, err := extractPagingOrColumns(node)
					if err != nil {
						return err
					}
					if !ok {
						disqualify = true
						return nil
					}
					if orPattern != nil && !sameOrPattern(orPattern.columns, cols) {
						disqualify = true
						return nil
					}
					if orPattern == nil {
						orPattern = &orPatternInfo{columns: cols}
					}
					addPagingOrConditions(conds, cols)
					return nil
				}
				addCond(conds, col, condEq, count)
				return nil
			default:
				col, kind, count, ok, err := extractSimpleColumnPredicate(node)
				if err != nil {
					return err
				}
				if ok {
					addCond(conds, col, kind, count)
				} else {
					markUnsupportedCondition(conds, node)
				}
				return nil
			}
		case *parser.IsNullExpr:
			col, kind, ok := extractIsNullPredicate(node)
			if ok {
				addCond(conds, col, kind, 1)
				return nil
			}
			markUnsupportedCondition(conds, node)
			return nil
		default:
			markUnsupportedCondition(conds, node)
			return nil
		}
	}
	if err := walk(expr); err != nil {
		return nil, false, nil, err
	}
	return conds, disqualify, orPattern, nil
}

func addPagingOrConditions(conds map[string]*condInfo, columns []string) {
	if len(columns) == 0 {
		return
	}
	for _, col := range columns {
		addCond(conds, col, condRange, 1)
	}
	if len(columns) > 1 {
		for i := 0; i < len(columns)-1; i++ {
			addCond(conds, columns[i], condEq, 1)
		}
	}
}

func sameOrPattern(left, right []string) bool {
	if len(left) != len(right) {
		return false
	}
	for i := range left {
		if left[i] != right[i] {
			return false
		}
	}
	return true
}

func extractPagingOrColumns(expr parser.Expression) ([]string, bool, error) {
	if expr == nil {
		return nil, false, nil
	}
	branches := flattenOrExpr(expr)
	if len(branches) < 2 {
		return nil, false, nil
	}
	type branchInfo struct {
		cols []string
		ops  []string
	}
	infos := make([]branchInfo, 0, len(branches))
	var maxCols []string
	lengths := make(map[int]struct{})
	rangeOp := ""
	for _, branch := range branches {
		cols, ops, ok, err := extractPagingBranch(branch)
		if err != nil {
			return nil, false, err
		}
		if !ok {
			return nil, false, nil
		}
		lengths[len(cols)] = struct{}{}
		if len(cols) > len(maxCols) {
			maxCols = cols
		}
		lastOp := ops[len(ops)-1]
		if rangeOp == "" {
			rangeOp = lastOp
		} else if rangeOp != lastOp {
			return nil, false, nil
		}
		infos = append(infos, branchInfo{cols: cols, ops: ops})
	}
	if len(maxCols) == 0 {
		return nil, false, nil
	}
	for i := 1; i <= len(maxCols); i++ {
		if _, ok := lengths[i]; !ok {
			return nil, false, nil
		}
	}
	for _, info := range infos {
		if len(info.cols) > len(maxCols) {
			return nil, false, nil
		}
		for i, col := range info.cols {
			if col != maxCols[i] {
				return nil, false, nil
			}
		}
	}
	return maxCols, true, nil
}

func extractPagingBranch(expr parser.Expression) ([]string, []string, bool, error) {
	conds := flattenAndExpr(expr)
	if len(conds) == 0 {
		return nil, nil, false, nil
	}
	cols := make([]string, 0, len(conds))
	ops := make([]string, 0, len(conds))
	seen := make(map[string]struct{}, len(conds))
	for _, cond := range conds {
		bin, ok := cond.(*parser.BinaryExpr)
		if !ok {
			return nil, nil, false, nil
		}
		op := strings.ToUpper(strings.TrimSpace(bin.Op))
		if op != "=" && op != ">" && op != "<" && op != ">=" && op != "<=" {
			return nil, nil, false, nil
		}
		col, value, ok := columnAndValue(bin.Left, bin.Right)
		if !ok {
			col, value, ok = columnAndValue(bin.Right, bin.Left)
		}
		if !ok || col == "" {
			return nil, nil, false, nil
		}
		if exprHasColumnRef(value) {
			return nil, nil, false, nil
		}
		name := strings.ToLower(col)
		if _, ok := seen[name]; ok {
			return nil, nil, false, nil
		}
		seen[name] = struct{}{}
		cols = append(cols, name)
		ops = append(ops, op)
	}
	for i := 0; i < len(ops)-1; i++ {
		if ops[i] != "=" {
			return nil, nil, false, nil
		}
	}
	if ops[len(ops)-1] == "=" {
		return nil, nil, false, nil
	}
	return cols, ops, true, nil
}

func flattenOrExpr(expr parser.Expression) []parser.Expression {
	if expr == nil {
		return nil
	}
	if bin, ok := expr.(*parser.BinaryExpr); ok && strings.EqualFold(bin.Op, "OR") {
		left := flattenOrExpr(bin.Left)
		right := flattenOrExpr(bin.Right)
		return append(left, right...)
	}
	return []parser.Expression{expr}
}

func expandOrBranches(expr parser.Expression, maxBranches int) ([]parser.Expression, bool) {
	if expr == nil {
		return nil, false
	}
	if maxBranches <= 0 {
		return nil, true
	}
	bin, ok := expr.(*parser.BinaryExpr)
	if !ok {
		return []parser.Expression{expr}, false
	}
	op := strings.ToUpper(strings.TrimSpace(bin.Op))
	switch op {
	case "OR":
		left, overflow := expandOrBranches(bin.Left, maxBranches)
		if overflow {
			return nil, true
		}
		right, overflow := expandOrBranches(bin.Right, maxBranches)
		if overflow {
			return nil, true
		}
		combined := append(left, right...)
		if len(combined) > maxBranches {
			return nil, true
		}
		return combined, false
	case "AND":
		left, overflow := expandOrBranches(bin.Left, maxBranches)
		if overflow {
			return nil, true
		}
		right, overflow := expandOrBranches(bin.Right, maxBranches)
		if overflow {
			return nil, true
		}
		if len(left) == 0 || len(right) == 0 {
			return nil, false
		}
		out := make([]parser.Expression, 0, len(left)*len(right))
		for _, l := range left {
			for _, r := range right {
				out = append(out, &parser.BinaryExpr{Op: "AND", Left: l, Right: r})
				if len(out) > maxBranches {
					return nil, true
				}
			}
		}
		return out, false
	default:
		return []parser.Expression{expr}, false
	}
}

func flattenAndExpr(expr parser.Expression) []parser.Expression {
	if expr == nil {
		return nil
	}
	if bin, ok := expr.(*parser.BinaryExpr); ok && strings.EqualFold(bin.Op, "AND") {
		left := flattenAndExpr(bin.Left)
		right := flattenAndExpr(bin.Right)
		return append(left, right...)
	}
	return []parser.Expression{expr}
}

func extractOrEqPredicate(expr parser.Expression) (string, int, bool, error) {
	if expr == nil {
		return "", 0, false, nil
	}
	switch node := expr.(type) {
	case *parser.BinaryExpr:
		if strings.EqualFold(node.Op, "OR") {
			leftCol, leftCount, ok, err := extractOrEqPredicate(node.Left)
			if err != nil {
				return "", 0, false, err
			}
			if !ok {
				return "", 0, false, nil
			}
			rightCol, rightCount, ok, err := extractOrEqPredicate(node.Right)
			if err != nil {
				return "", 0, false, err
			}
			if !ok || !strings.EqualFold(leftCol, rightCol) {
				return "", 0, false, nil
			}
			if leftCount <= 0 {
				leftCount = 1
			}
			if rightCount <= 0 {
				rightCount = 1
			}
			return leftCol, leftCount + rightCount, true, nil
		}
		col, kind, count, ok, err := extractSimpleColumnPredicate(node)
		if err != nil || !ok || kind != condEq {
			return "", 0, false, err
		}
		if count <= 0 {
			count = 1
		}
		return col, count, true, nil
	default:
		col, kind, count, ok, err := extractSimpleColumnPredicate(node)
		if err != nil || !ok || kind != condEq {
			return "", 0, false, err
		}
		if count <= 0 {
			count = 1
		}
		return col, count, true, nil
	}
}

func extractSimpleColumnPredicate(expr parser.Expression) (string, condKind, int, bool, error) {
	if expr == nil {
		return "", condEq, 0, false, nil
	}
	switch node := expr.(type) {
	case *parser.BinaryExpr:
		op := strings.ToUpper(strings.TrimSpace(node.Op))
		switch op {
		case "=", "<", "<=", ">", ">=", "IN", "BETWEEN":
			return extractBinaryPredicate(node, op)
		default:
			return "", condEq, 0, false, nil
		}
	case *parser.IsNullExpr:
		col, kind, ok := extractIsNullPredicate(node)
		if !ok {
			return "", condEq, 0, false, nil
		}
		return col, kind, 1, true, nil
	default:
		return "", condEq, 0, false, nil
	}
}

func selectBestIndexForOR(stmt *parser.SelectStatement, indexes []mysqlbackend.IndexInfo) (indexCandidate, string, bool, error) {
	if stmt == nil || stmt.Where == nil {
		return indexCandidate{}, "", false, nil
	}
	branches, overflow := expandOrBranches(stmt.Where, maxOrBranches)
	if overflow || len(branches) < 2 {
		return indexCandidate{}, "", false, nil
	}
	metas := buildIndexMetas(indexes)
	if len(metas) == 0 {
		return indexCandidate{}, "", false, nil
	}
	queryInfo := buildQueryIndexInfo(stmt)
	groupByCols := groupByColumns(stmt)

	branchCandidates := make([]map[string]indexCandidate, 0, len(branches))
	intersection := make(map[string]struct{})
	for i, branch := range branches {
		conds, disqualify, orPattern, err := extractIndexConditions(branch)
		if err != nil {
			return indexCandidate{}, "", false, err
		}
		if disqualify || hasUnsupportedConditions(conds) {
			return indexCandidate{}, "", false, nil
		}
		eqCols, rangeCols := extractConditionColumns(conds)
		eligible := metas
		if orPattern != nil && len(orPattern.columns) > 0 {
			eligible = filterIndexMetasForOrPattern(eligible, orPattern.columns)
		}
		eligible = filterIndexMetas(eligible, eqCols, rangeCols, groupByCols)
		if len(eligible) == 0 {
			return indexCandidate{}, "", false, nil
		}
		candidateMap := make(map[string]indexCandidate, len(eligible))
		for _, meta := range eligible {
			candidateMap[meta.name] = buildIndexCandidate(meta, conds, queryInfo)
		}
		if i == 0 {
			for name := range candidateMap {
				intersection[name] = struct{}{}
			}
		} else {
			for name := range intersection {
				if _, ok := candidateMap[name]; !ok {
					delete(intersection, name)
				}
			}
		}
		if len(intersection) == 0 {
			return indexCandidate{}, "", false, nil
		}
		branchCandidates = append(branchCandidates, candidateMap)
	}

	aggregated := make([]indexCandidate, 0, len(intersection))
	for name := range intersection {
		base := branchCandidates[0][name]
		agg := base
		for i := 1; i < len(branchCandidates); i++ {
			cand := branchCandidates[i][name]
			if cand.accessCols < agg.accessCols {
				agg.accessCols = cand.accessCols
			}
			if cand.eqPrefix < agg.eqPrefix {
				agg.eqPrefix = cand.eqPrefix
			}
			if cand.pointRanges > agg.pointRanges {
				agg.pointRanges = cand.pointRanges
			}
			if cand.filterCols < agg.filterCols {
				agg.filterCols = cand.filterCols
			}
			agg.hasRange = agg.hasRange || cand.hasRange
			agg.fullMatch = agg.fullMatch && cand.fullMatch
			agg.orderByMatch = agg.orderByMatch && cand.orderByMatch
		}
		aggregated = append(aggregated, agg)
	}

	if len(aggregated) == 0 {
		return indexCandidate{}, "", false, nil
	}
	best, rule := chooseBestIndex(aggregated, queryInfo)
	return best, rule, true, nil
}

func extractBinaryPredicate(node *parser.BinaryExpr, op string) (string, condKind, int, bool, error) {
	if node == nil {
		return "", condEq, 0, false, nil
	}
	col, value, ok := columnAndValue(node.Left, node.Right)
	if !ok {
		col, value, ok = columnAndValue(node.Right, node.Left)
	}
	if !ok || col == "" {
		return "", condEq, 0, false, nil
	}
	switch op {
	case "=":
		if exprHasColumnRef(value) {
			return "", condEq, 0, false, nil
		}
		return col, condEq, 1, true, nil
	case "<", "<=", ">", ">=":
		if exprHasColumnRef(value) {
			return "", condEq, 0, false, nil
		}
		return col, condRange, 1, true, nil
	case "BETWEEN":
		tuple, ok := value.(*parser.TupleLiteral)
		if !ok || len(tuple.Elements) != 2 {
			return "", condEq, 0, false, nil
		}
		if exprHasColumnRef(tuple.Elements[0]) || exprHasColumnRef(tuple.Elements[1]) {
			return "", condEq, 0, false, nil
		}
		return col, condRange, 1, true, nil
	case "IN":
		list, ok := value.(*parser.ListLiteral)
		if !ok {
			return "", condEq, 0, false, nil
		}
		for _, elem := range list.Elements {
			if exprHasColumnRef(elem) {
				return "", condEq, 0, false, nil
			}
		}
		count := len(list.Elements)
		if count == 0 {
			count = 1
		}
		return col, condEq, count, true, nil
	default:
		return "", condEq, 0, false, nil
	}
}

func extractIsNullPredicate(node *parser.IsNullExpr) (string, condKind, bool) {
	if node == nil {
		return "", condEq, false
	}
	colExpr, ok := node.Expr.(*parser.ColumnExpr)
	if !ok || colExpr.Name == "" {
		return "", condEq, false
	}
	if node.Not {
		return colExpr.Name, condRange, true
	}
	return colExpr.Name, condEq, true
}

func columnAndValue(left parser.Expression, right parser.Expression) (string, parser.Expression, bool) {
	col, ok := left.(*parser.ColumnExpr)
	if !ok || col.Name == "" {
		return "", nil, false
	}
	return col.Name, right, true
}

func addCond(conds map[string]*condInfo, column string, kind condKind, count int) {
	if column == "" {
		return
	}
	key := strings.ToLower(column)
	entry := conds[key]
	if entry == nil {
		entry = &condInfo{}
		conds[key] = entry
	}
	switch kind {
	case condEq:
		entry.eq = true
		if count <= 0 {
			count = 1
		}
		if entry.inCount == 0 {
			entry.inCount = count
		} else {
			entry.inCount += count
		}
	case condRange:
		entry.rng = true
	}
}

func markUnsupportedCondition(conds map[string]*condInfo, expr parser.Expression) {
	if expr == nil {
		return
	}
	refs := make(map[string]columnRef)
	collectColumnRefsInto(expr, refs)
	for _, ref := range refs {
		key := strings.ToLower(ref.name)
		if key == "" {
			continue
		}
		entry := conds[key]
		if entry == nil {
			entry = &condInfo{}
			conds[key] = entry
		}
		entry.unsupported = true
	}
}

func hasUnsupportedConditions(conds map[string]*condInfo) bool {
	for _, cond := range conds {
		if cond != nil && cond.unsupported {
			return true
		}
	}
	return false
}

func extractConditionColumns(conds map[string]*condInfo) (map[string]struct{}, map[string]struct{}) {
	eqCols := make(map[string]struct{})
	rangeCols := make(map[string]struct{})
	for col, cond := range conds {
		if cond == nil {
			continue
		}
		if cond.eq {
			eqCols[col] = struct{}{}
		}
		if cond.rng {
			rangeCols[col] = struct{}{}
		}
	}
	return eqCols, rangeCols
}

func groupByColumns(stmt *parser.SelectStatement) map[string]struct{} {
	out := make(map[string]struct{})
	if stmt == nil || len(stmt.GroupBy) == 0 {
		return out
	}
	for _, expr := range stmt.GroupBy {
		col, ok := expr.(*parser.ColumnExpr)
		if !ok || col == nil || col.Name == "" {
			continue
		}
		out[strings.ToLower(col.Name)] = struct{}{}
	}
	return out
}

func hasAnyIndexableCondition(eqCols, rangeCols, groupByCols map[string]struct{}) bool {
	return len(eqCols) > 0 || len(rangeCols) > 0 || len(groupByCols) > 0
}

func buildIndexCandidatesFromMetas(metas []indexMeta, conds map[string]*condInfo, query queryIndexInfo) []indexCandidate {
	candidates := make([]indexCandidate, 0, len(metas))
	for _, meta := range metas {
		candidate := buildIndexCandidate(meta, conds, query)
		candidates = append(candidates, candidate)
	}
	return candidates
}

func buildIndexCandidate(meta indexMeta, conds map[string]*condInfo, query queryIndexInfo) indexCandidate {
	eqPrefix, hasRange := indexAccessStats(meta, conds)
	accessCols := eqPrefix
	if hasRange {
		accessCols++
	}
	fullMatch := eqPrefix == len(meta.columns) && !hasRange
	pointRanges := 1
	if eqPrefix == 0 {
		pointRanges = 0
	} else {
		for i := 0; i < eqPrefix; i++ {
			key := strings.ToLower(meta.columns[i])
			count := 1
			if cond := conds[key]; cond != nil && cond.inCount > 0 {
				count = cond.inCount
			}
			pointRanges = safeMul(pointRanges, count)
		}
	}
	covering := coversAllColumns(meta, query)
	filterCols := countFilterColumns(meta, conds, accessCols)
	orderByMatch := orderByMatches(meta, query, eqPrefix)
	return indexCandidate{
		meta:          meta,
		accessCols:    accessCols,
		eqPrefix:      eqPrefix,
		hasRange:      hasRange,
		fullMatch:     fullMatch,
		pointRanges:   pointRanges,
		covering:      covering,
		needBackTable: !covering,
		orderByMatch:  orderByMatch,
		filterCols:    filterCols,
	}
}

func buildIndexMetas(indexes []mysqlbackend.IndexInfo) []indexMeta {
	if len(indexes) == 0 {
		return nil
	}
	sort.Slice(indexes, func(i, j int) bool {
		if indexes[i].Name == indexes[j].Name {
			return indexes[i].Seq < indexes[j].Seq
		}
		return indexes[i].Name < indexes[j].Name
	})
	var metas []indexMeta
	var current *indexMeta
	for _, idx := range indexes {
		if current == nil || current.name != idx.Name {
			metas = append(metas, indexMeta{
				name:    idx.Name,
				unique:  !idx.NonUnique,
				primary: idx.IsPrimary,
			})
			current = &metas[len(metas)-1]
		}
		current.columns = append(current.columns, idx.Column)
		if idx.NonUnique {
			current.unique = false
		}
		if idx.IsPrimary {
			current.primary = true
		}
	}
	return metas
}

func filterIndexMetas(metas []indexMeta, eqCols, rangeCols, groupByCols map[string]struct{}) []indexMeta {
	out := make([]indexMeta, 0, len(metas))
	for _, meta := range metas {
		if !indexMeetsMatchRequirements(meta, eqCols, rangeCols, groupByCols) {
			continue
		}
		out = append(out, meta)
	}
	return out
}

func indexMeetsMatchRequirements(meta indexMeta, eqCols, rangeCols, groupByCols map[string]struct{}) bool {
	if len(eqCols) > 0 {
		for col := range eqCols {
			if !indexHasColumn(meta, col) {
				return false
			}
		}
	}
	if len(rangeCols) > 0 {
		ok := false
		for col := range rangeCols {
			if indexHasColumn(meta, col) {
				ok = true
				break
			}
		}
		if !ok {
			return false
		}
	}
	if len(groupByCols) > 0 {
		ok := false
		for col := range groupByCols {
			if indexHasColumn(meta, col) {
				ok = true
				break
			}
		}
		if !ok {
			return false
		}
	}
	return true
}

func indexHasColumn(meta indexMeta, column string) bool {
	needle := strings.ToLower(column)
	for _, col := range meta.columns {
		if strings.ToLower(col) == needle {
			return true
		}
	}
	return false
}

func indexAccessStats(meta indexMeta, conds map[string]*condInfo) (int, bool) {
	eqPrefix := 0
	for _, col := range meta.columns {
		cond := conds[strings.ToLower(col)]
		if cond != nil && cond.rng {
			return eqPrefix, true
		}
		if cond != nil && cond.eq {
			eqPrefix++
			continue
		}
		break
	}
	return eqPrefix, false
}

func filterIndexMetasForOrPattern(metas []indexMeta, columns []string) []indexMeta {
	if len(columns) == 0 {
		return metas
	}
	out := make([]indexMeta, 0, len(metas))
	for _, meta := range metas {
		if len(meta.columns) < len(columns) {
			continue
		}
		match := true
		for i, col := range columns {
			if strings.ToLower(meta.columns[i]) != col {
				match = false
				break
			}
		}
		if match {
			out = append(out, meta)
		}
	}
	return out
}

func coversAllColumns(meta indexMeta, query queryIndexInfo) bool {
	if query.hasWildcard {
		return false
	}
	if len(query.requiredColumns) == 0 {
		return true
	}
	indexCols := make(map[string]struct{}, len(meta.columns))
	for _, col := range meta.columns {
		indexCols[strings.ToLower(col)] = struct{}{}
	}
	for col := range query.requiredColumns {
		if _, ok := indexCols[col]; !ok {
			return false
		}
	}
	return true
}

func orderByMatches(meta indexMeta, query queryIndexInfo, eqPrefix int) bool {
	if !query.orderByActive {
		return true
	}
	if len(query.orderBy) == 0 {
		return false
	}
	if eqPrefix > len(meta.columns) {
		return false
	}
	start := eqPrefix
	if len(meta.columns)-start < len(query.orderBy) {
		return false
	}
	direction := ""
	for i, ob := range query.orderBy {
		col := strings.ToLower(meta.columns[start+i])
		if col != ob.column {
			return false
		}
		if ob.direction == "" {
			continue
		}
		if direction == "" {
			direction = ob.direction
			continue
		}
		if direction != ob.direction {
			return false
		}
	}
	return true
}

func countFilterColumns(meta indexMeta, conds map[string]*condInfo, accessCols int) int {
	if len(conds) == 0 {
		return 0
	}
	count := 0
	for i, col := range meta.columns {
		cond := conds[strings.ToLower(col)]
		if cond == nil {
			continue
		}
		if i < accessCols {
			continue
		}
		if cond.eq || cond.rng || cond.unsupported {
			count++
		}
	}
	return count
}

func chooseBestIndex(candidates []indexCandidate, query queryIndexInfo) (indexCandidate, string) {
	if len(candidates) == 1 {
		return candidates[0], "single"
	}
	if best, rule, ok := applyPreRules(candidates); ok {
		return best, rule
	}
	skyline := pruneSkyline(candidates, query)
	return pickByHeuristic(skyline, query), "skyline"
}

func applyPreRules(candidates []indexCandidate) (indexCandidate, string, bool) {
	if len(candidates) == 0 {
		return indexCandidate{}, "", false
	}
	uniqueCovering := filterCandidates(candidates, func(c indexCandidate) bool {
		return c.meta.unique && c.fullMatch && !c.needBackTable
	})
	if len(uniqueCovering) > 0 {
		return pickByPointRanges(uniqueCovering), "rule1", true
	}

	rule2Candidates := filterCandidates(candidates, func(c indexCandidate) bool {
		return c.meta.unique && c.fullMatch && c.needBackTable
	})
	var rule2Best *indexCandidate
	if len(rule2Candidates) > 0 {
		best := pickByPointRanges(rule2Candidates)
		rule2Best = &best
	}

	rule3Candidates := filterCandidates(candidates, func(c indexCandidate) bool {
		if c.meta.unique || c.needBackTable {
			return false
		}
		estimate := estimateRows(c)
		return estimate > 0 && estimate <= defaultIndexScanThreshold
	})
	var rule3Best *indexCandidate
	if len(rule3Candidates) > 0 {
		best := pickByEstimatedRows(rule3Candidates)
		rule3Best = &best
	}

	if rule2Best != nil && rule3Best != nil {
		if totalRowsEstimate(*rule2Best) <= totalRowsEstimate(*rule3Best) {
			return *rule2Best, "rule4", true
		}
		return *rule3Best, "rule4", true
	}
	if rule2Best != nil {
		return *rule2Best, "rule2", true
	}
	if rule3Best != nil {
		return *rule3Best, "rule3", true
	}
	return indexCandidate{}, "", false
}

func filterCandidates(candidates []indexCandidate, keep func(indexCandidate) bool) []indexCandidate {
	out := make([]indexCandidate, 0, len(candidates))
	for _, c := range candidates {
		if keep(c) {
			out = append(out, c)
		}
	}
	return out
}

func pickByPointRanges(candidates []indexCandidate) indexCandidate {
	sort.Slice(candidates, func(i, j int) bool {
		if candidates[i].pointRanges != candidates[j].pointRanges {
			return candidates[i].pointRanges < candidates[j].pointRanges
		}
		return candidates[i].meta.name < candidates[j].meta.name
	})
	return candidates[0]
}

func pickByEstimatedRows(candidates []indexCandidate) indexCandidate {
	sort.Slice(candidates, func(i, j int) bool {
		left := estimateRows(candidates[i])
		right := estimateRows(candidates[j])
		if left != right {
			return left < right
		}
		return candidates[i].meta.name < candidates[j].meta.name
	})
	return candidates[0]
}

func estimateRows(c indexCandidate) int {
	if c.fullMatch {
		if c.pointRanges > 0 {
			return c.pointRanges
		}
		return 1
	}
	if c.accessCols == 0 {
		return defaultRangeEstimate
	}
	return defaultRangeEstimate
}

func totalRowsEstimate(c indexCandidate) int {
	indexRows := estimateRows(c)
	if c.needBackTable {
		return safeMul(indexRows, 2)
	}
	return indexRows
}

func pruneSkyline(candidates []indexCandidate, query queryIndexInfo) []indexCandidate {
	if len(candidates) <= 1 {
		return candidates
	}
	out := make([]indexCandidate, 0, len(candidates))
	for i := range candidates {
		dominated := false
		for j := range candidates {
			if i == j {
				continue
			}
			if dominates(candidates[j], candidates[i], query) {
				dominated = true
				break
			}
		}
		if !dominated {
			out = append(out, candidates[i])
		}
	}
	if len(out) == 0 {
		return candidates
	}
	return out
}

func dominates(a, b indexCandidate, query queryIndexInfo) bool {
	if a.accessCols < b.accessCols {
		return false
	}
	if a.needBackTable && !b.needBackTable {
		return false
	}
	if query.orderByActive {
		if !a.orderByMatch && b.orderByMatch {
			return false
		}
	}
	if a.needBackTable && b.needBackTable {
		if a.filterCols < b.filterCols {
			return false
		}
	}
	better := false
	if a.accessCols > b.accessCols {
		better = true
	}
	if a.needBackTable != b.needBackTable && !a.needBackTable {
		better = true
	}
	if query.orderByActive && a.orderByMatch != b.orderByMatch && a.orderByMatch {
		better = true
	}
	if a.needBackTable && b.needBackTable && a.filterCols > b.filterCols {
		better = true
	}
	return better
}

func pickByHeuristic(candidates []indexCandidate, query queryIndexInfo) indexCandidate {
	sort.Slice(candidates, func(i, j int) bool {
		left := candidates[i]
		right := candidates[j]
		if left.accessCols != right.accessCols {
			return left.accessCols > right.accessCols
		}
		if query.orderByActive && left.orderByMatch != right.orderByMatch {
			return left.orderByMatch
		}
		if left.needBackTable != right.needBackTable {
			return !left.needBackTable
		}
		if left.needBackTable && right.needBackTable && left.filterCols != right.filterCols {
			return left.filterCols > right.filterCols
		}
		if left.pointRanges != right.pointRanges {
			return left.pointRanges < right.pointRanges
		}
		if left.meta.unique != right.meta.unique {
			return left.meta.unique
		}
		if len(left.meta.columns) != len(right.meta.columns) {
			return len(left.meta.columns) < len(right.meta.columns)
		}
		return left.meta.name < right.meta.name
	})
	return candidates[0]
}

func safeMul(a, b int) int {
	if a == 0 || b == 0 {
		return 0
	}
	if a > (int(^uint(0)>>1) / b) {
		return int(^uint(0) >> 1)
	}
	return a * b
}

func indexTypeLabel(meta indexMeta) string {
	if meta.primary {
		return "primary"
	}
	if meta.unique {
		return "unique"
	}
	return "secondary"
}