package executor

import (
	"context"
	"fmt"
	"regexp"
	"sort"
	"strconv"
	"strings"

	"go.uber.org/zap"

	cqlkeyspace "github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/keyspace"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/types"
)

var (
	describeShowCreateTTLStartRE = regexp.MustCompile(`(?is)\bTTL\s*=`)
	describeShowCreateTTLNextRE  = regexp.MustCompile(`(?is)\b(TTL_ENABLE|TTL_JOB_INTERVAL|PARTITION\s+BY)\b`)
	describeShowCreateTTLEnable  = regexp.MustCompile(`(?is)\bTTL_ENABLE\s*=\s*('(?:''|[^'])*'|[A-Za-z]+)`)
	describeShowCreateTTLJobInt  = regexp.MustCompile(`(?is)\bTTL_JOB_INTERVAL\s*=\s*('(?:''|[^'])*'|[^\s;]+)`)
	describeShowCreateHintOpenRE = regexp.MustCompile(`(?is)/\*T!\[[^\]]+\]\s*`)
	describeShowCreateHintClose  = regexp.MustCompile(`(?is)\s*\*/`)
)

func (e *Executor) executeDescribe(ctx context.Context, user, password string, stmt *parser.DescribeStatement) (*QueryResult, error) {
	if stmt == nil {
		return nil, fmt.Errorf("describe statement is nil")
	}
	if keyspaceName := describeKeyspaceName(stmt); keyspaceName != "" {
		if err := validateBlockedKeyspace(keyspaceName); err != nil {
			return nil, err
		}
	}
	switch stmt.Target {
	case parser.DescribeTargetKeyspaces:
		return e.describeKeyspaces(ctx, stmt, user, password)
	case parser.DescribeTargetKeyspace:
		return e.describeKeyspace(ctx, stmt, user, password)
	case parser.DescribeTargetTables:
		return e.describeTables(ctx, stmt, user, password)
	case parser.DescribeTargetTable:
		return e.describeTable(ctx, stmt, user, password)
	}

	selectStmt, err := describeToSelect(stmt)
	if err != nil {
		return nil, err
	}
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	return convertQueryResult(result), nil
}

func describeKeyspaceName(stmt *parser.DescribeStatement) string {
	if stmt == nil {
		return ""
	}
	switch stmt.Target {
	case parser.DescribeTargetKeyspace:
		return stmt.Keyspace
	case parser.DescribeTargetTables:
		return stmt.Keyspace
	case parser.DescribeTargetTable:
		if stmt.Table != nil {
			return stmt.Table.Schema
		}
	}
	return ""
}

func describeToSelect(stmt *parser.DescribeStatement) (*parser.SelectStatement, error) {
	if stmt == nil {
		return nil, fmt.Errorf("describe statement is nil")
	}

	switch stmt.Target {
	case parser.DescribeTargetKeyspaces:
		return describeSelectColumns(stmt.Token, "keyspaces", []string{"keyspace_name"}, nil), nil
	case parser.DescribeTargetKeyspace:
		if stmt.Keyspace == "" {
			return nil, fmt.Errorf("keyspace name is required for DESCRIBE KEYSPACE")
		}
		where := describeEquals("keyspace_name", stmt.Keyspace)
		return describeSelectAll(stmt.Token, "keyspaces", where), nil
	case parser.DescribeTargetTables:
		var where parser.Expression
		if stmt.Keyspace != "" {
			where = describeEquals("keyspace_name", stmt.Keyspace)
		}
		return describeSelectColumns(stmt.Token, "tables", []string{"keyspace_name", "table_name"}, where), nil
	case parser.DescribeTargetTable:
		if stmt.Table == nil {
			return nil, fmt.Errorf("table name is required for DESCRIBE TABLE")
		}
		if stmt.Table.Schema == "" {
			return nil, fmt.Errorf("keyspace is required for DESCRIBE TABLE %s", stmt.Table.Name)
		}
		where := describeAnd(
			describeEquals("keyspace_name", stmt.Table.Schema),
			describeEquals("table_name", stmt.Table.Name),
		)
		return describeSelectAll(stmt.Token, "columns", where), nil
	default:
		return nil, fmt.Errorf("unsupported DESCRIBE target %q", stmt.Target)
	}
}

func (e *Executor) describeKeyspaces(ctx context.Context, stmt *parser.DescribeStatement, user, password string) (*QueryResult, error) {
	selectStmt := describeSelectColumns(stmt.Token, "keyspaces", []string{"keyspace_name"}, nil)
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	return formatDescribeKeyspaces(result)
}

func (e *Executor) describeTables(ctx context.Context, stmt *parser.DescribeStatement, user, password string) (*QueryResult, error) {
	selectStmt, err := describeToSelect(stmt)
	if err != nil {
		return nil, err
	}
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	return formatDescribeTables(result)
}

func (e *Executor) describeKeyspace(ctx context.Context, stmt *parser.DescribeStatement, user, password string) (*QueryResult, error) {
	if stmt == nil || stmt.Keyspace == "" {
		return nil, fmt.Errorf("keyspace name is required for DESCRIBE KEYSPACE")
	}

	columns := describeTableResultColumns()
	definition, err := e.loadDescribeKeyspaceDefinition(ctx, stmt.Token, stmt.Keyspace, user, password)
	if err != nil {
		return nil, err
	}
	if definition == nil {
		return &QueryResult{Columns: columns, UseGlobalSpec: true}, nil
	}

	createStatement, err := buildKeyspaceCreateStatement(definition)
	if err != nil {
		return nil, err
	}

	rows := []Row{{
		Values: []interface{}{definition.Name, "keyspace", definition.Name, createStatement},
	}}

	tableRows, err := e.describeKeyspaceTables(ctx, stmt.Token, definition.Name, user, password)
	if err != nil {
		return nil, err
	}
	rows = append(rows, tableRows...)

	return &QueryResult{
		Columns:       columns,
		Rows:          rows,
		UseGlobalSpec: true,
	}, nil
}

func (e *Executor) describeTable(ctx context.Context, stmt *parser.DescribeStatement, user, password string) (*QueryResult, error) {
	if stmt == nil || stmt.Table == nil {
		return nil, fmt.Errorf("table name is required for DESCRIBE TABLE")
	}
	if stmt.Table.Schema == "" {
		return nil, fmt.Errorf("keyspace is required for DESCRIBE TABLE %s", stmt.Table.Name)
	}

	columns := describeTableResultColumns()
	def, err := e.loadDescribeTableDefinition(ctx, stmt.Token, stmt.Table.Schema, stmt.Table.Name, user, password)
	if err != nil {
		return nil, err
	}
	if def == nil {
		return &QueryResult{Columns: columns, UseGlobalSpec: true}, nil
	}

	createStatement := ""
	if len(def.PartitionKeys) > 0 {
		var err error
		createStatement, err = buildCreateStatement(def)
		if err != nil {
			return nil, err
		}
	}

	rows := []Row{{
		Values: []interface{}{def.Keyspace, "table", def.Table, createStatement},
	}}

	return &QueryResult{
		Columns:       columns,
		Rows:          rows,
		UseGlobalSpec: true,
	}, nil
}

type describeKeyspaceDefinition struct {
	Name          string
	DurableWrites bool
	Replication   map[string]string
}

func (e *Executor) loadDescribeKeyspaceDefinition(ctx context.Context, token parser.Position, keyspace, user, password string) (*describeKeyspaceDefinition, error) {
	selectStmt := describeSelectColumns(token, "keyspaces", []string{"keyspace_name", "durable_writes", "replication"}, describeEquals("keyspace_name", keyspace))
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	if result == nil || len(result.Rows) == 0 {
		return nil, nil
	}

	nameIndex := findColumnIndex(result.Columns, "keyspace_name")
	durableIndex := findColumnIndex(result.Columns, "durable_writes")
	replicationIndex := findColumnIndex(result.Columns, "replication")
	if nameIndex == -1 {
		return nil, fmt.Errorf("keyspace_name column not found in system_schema.keyspaces")
	}

	row := result.Rows[0]
	name := keyspace
	if nameIndex < len(row.Values) && row.Values[nameIndex] != nil {
		if value := fmt.Sprint(row.Values[nameIndex]); value != "" {
			name = value
		}
	}

	durableWrites := true
	if durableIndex != -1 && durableIndex < len(row.Values) && row.Values[durableIndex] != nil {
		durableWrites = parseDescribeBool(row.Values[durableIndex], true)
	}

	replication := map[string]string{}
	if replicationIndex != -1 && replicationIndex < len(row.Values) && row.Values[replicationIndex] != nil {
		replication = normalizeReplicationMap(row.Values[replicationIndex])
	}

	replication = normalizeKeyspaceReplication(name, replication)

	return &describeKeyspaceDefinition{
		Name:          name,
		DurableWrites: durableWrites,
		Replication:   replication,
	}, nil
}

func (e *Executor) describeKeyspaceTables(ctx context.Context, token parser.Position, keyspace, user, password string) ([]Row, error) {
	selectStmt := describeSelectColumns(token, "tables", []string{"keyspace_name", "table_name"}, describeEquals("keyspace_name", keyspace))
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	if result == nil {
		return nil, nil
	}

	keyspaceIndex := findColumnIndex(result.Columns, "keyspace_name")
	if keyspaceIndex == -1 {
		return nil, fmt.Errorf("keyspace_name column not found in system_schema.tables")
	}
	tableIndex := findColumnIndex(result.Columns, "table_name")
	if tableIndex == -1 {
		return nil, fmt.Errorf("table_name column not found in system_schema.tables")
	}

	rows := make([]Row, 0, len(result.Rows))
	for _, row := range result.Rows {
		tableName := ""
		if tableIndex < len(row.Values) && row.Values[tableIndex] != nil {
			tableName = fmt.Sprint(row.Values[tableIndex])
		}
		if tableName == "" {
			continue
		}

		ksName := keyspace
		if keyspaceIndex < len(row.Values) && row.Values[keyspaceIndex] != nil {
			if value := fmt.Sprint(row.Values[keyspaceIndex]); value != "" {
				ksName = value
			}
		}

		def, err := e.loadDescribeTableDefinition(ctx, token, ksName, tableName, user, password)
		if err != nil {
			return nil, err
		}
		if def == nil {
			rows = append(rows, Row{
				Values: []interface{}{ksName, "table", tableName, ""},
			})
			continue
		}

		createStatement := ""
		if len(def.PartitionKeys) > 0 {
			createStatement, err = buildCreateStatement(def)
			if err != nil {
				return nil, err
			}
		}
		rows = append(rows, Row{
			Values: []interface{}{ksName, "table", tableName, createStatement},
		})
	}

	return rows, nil
}

func describeSelectAll(token parser.Position, table string, where parser.Expression) *parser.SelectStatement {
	return &parser.SelectStatement{
		Token:   token,
		Columns: []parser.Expression{&parser.StarExpr{Token: token}},
		From: &parser.TableName{
			Schema: "system_schema",
			Name:   table,
			Token:  token,
		},
		Where: where,
	}
}

func describeResultColumns() []ColumnInfo {
	return []ColumnInfo{
		{Keyspace: "system", Table: "describe", Name: "keyspace_name", Type: "text"},
		{Keyspace: "system", Table: "describe", Name: "type", Type: "text"},
		{Keyspace: "system", Table: "describe", Name: "name", Type: "text"},
	}
}

func describeTableResultColumns() []ColumnInfo {
	return []ColumnInfo{
		{Keyspace: "system", Table: "describe", Name: "keyspace_name", Type: "text"},
		{Keyspace: "system", Table: "describe", Name: "type", Type: "text"},
		{Keyspace: "system", Table: "describe", Name: "name", Type: "text"},
		{Keyspace: "system", Table: "describe", Name: "create_statement", Type: "text"},
	}
}

func formatDescribeKeyspaces(result *types.QueryResult) (*QueryResult, error) {
	columns := describeResultColumns()
	if result == nil {
		return &QueryResult{Columns: columns, UseGlobalSpec: true}, nil
	}

	nameIndex := findColumnIndex(result.Columns, "keyspace_name")
	if nameIndex == -1 {
		return nil, fmt.Errorf("keyspace_name column not found in system_schema.keyspaces")
	}

	rows := make([]Row, 0, len(result.Rows))
	for _, row := range result.Rows {
		var name string
		if nameIndex < len(row.Values) && row.Values[nameIndex] != nil {
			name = fmt.Sprint(row.Values[nameIndex])
		}
		rows = append(rows, Row{
			Values: []interface{}{name, "keyspace", name},
		})
	}

	return &QueryResult{
		Columns:       columns,
		Rows:          rows,
		UseGlobalSpec: true,
	}, nil
}

func formatDescribeTables(result *types.QueryResult) (*QueryResult, error) {
	columns := describeResultColumns()
	if result == nil {
		return &QueryResult{Columns: columns, UseGlobalSpec: true}, nil
	}

	keyspaceIndex := findColumnIndex(result.Columns, "keyspace_name")
	if keyspaceIndex == -1 {
		return nil, fmt.Errorf("keyspace_name column not found in system_schema.tables")
	}
	tableIndex := findColumnIndex(result.Columns, "table_name")
	if tableIndex == -1 {
		return nil, fmt.Errorf("table_name column not found in system_schema.tables")
	}

	rows := make([]Row, 0, len(result.Rows))
	for _, row := range result.Rows {
		var keyspace string
		if keyspaceIndex < len(row.Values) && row.Values[keyspaceIndex] != nil {
			keyspace = fmt.Sprint(row.Values[keyspaceIndex])
		}
		var table string
		if tableIndex < len(row.Values) && row.Values[tableIndex] != nil {
			table = fmt.Sprint(row.Values[tableIndex])
		}
		rows = append(rows, Row{
			Values: []interface{}{keyspace, "table", table},
		})
	}

	return &QueryResult{
		Columns:       columns,
		Rows:          rows,
		UseGlobalSpec: true,
	}, nil
}

func describeSelectColumns(token parser.Position, table string, columns []string, where parser.Expression) *parser.SelectStatement {
	exprs := make([]parser.Expression, 0, len(columns))
	for _, name := range columns {
		exprs = append(exprs, &parser.ColumnExpr{
			Token: token,
			Name:  name,
		})
	}
	return &parser.SelectStatement{
		Token:   token,
		Columns: exprs,
		From: &parser.TableName{
			Schema: "system_schema",
			Name:   table,
			Token:  token,
		},
		Where: where,
	}
}

func findColumnIndex(columns []types.ColumnInfo, name string) int {
	for i, col := range columns {
		if col.Name == name {
			return i
		}
	}
	return -1
}

func buildKeyspaceCreateStatement(def *describeKeyspaceDefinition) (string, error) {
	if def == nil {
		return "", fmt.Errorf("describe keyspace definition is nil")
	}

	var builder strings.Builder
	builder.WriteString("CREATE KEYSPACE ")
	builder.WriteString(quoteCQLIdentifier(def.Name))
	builder.WriteString(" WITH replication = ")
	builder.WriteString(formatReplicationMap(def.Replication))
	builder.WriteString(" AND durable_writes = ")
	builder.WriteString(strconv.FormatBool(def.DurableWrites))
	builder.WriteString(";")

	return builder.String(), nil
}

func normalizeReplicationMap(value interface{}) map[string]string {
	normalized := make(map[string]string)
	switch v := value.(type) {
	case map[string]string:
		for key, val := range v {
			normalized[key] = val
		}
	case map[string]interface{}:
		for key, val := range v {
			normalized[key] = fmt.Sprint(val)
		}
	case map[interface{}]interface{}:
		for key, val := range v {
			normalized[fmt.Sprint(key)] = fmt.Sprint(val)
		}
	}
	return normalized
}

func normalizeKeyspaceReplication(name string, replication map[string]string) map[string]string {
	if isLocalDescribeKeyspace(name) {
		return map[string]string{"class": "LocalStrategy"}
	}

	normalized := make(map[string]string, len(replication))
	for key, val := range replication {
		normalized[key] = val
	}
	if len(normalized) == 0 {
		normalized["class"] = "SimpleStrategy"
		normalized["replication_factor"] = "3"
	}
	if _, ok := normalized["class"]; !ok {
		normalized["class"] = "SimpleStrategy"
	}
	if strings.EqualFold(normalized["class"], "SimpleStrategy") {
		if _, ok := normalized["replication_factor"]; !ok {
			normalized["replication_factor"] = "3"
		}
	}
	if strings.EqualFold(normalized["class"], "LocalStrategy") {
		delete(normalized, "replication_factor")
	}
	return normalized
}

func formatReplicationMap(replication map[string]string) string {
	if len(replication) == 0 {
		return "{}"
	}

	parts := make([]string, 0, len(replication))
	if value, ok := replication["class"]; ok {
		parts = append(parts, formatCQLMapEntry("class", value))
	}
	keys := make([]string, 0, len(replication))
	for key := range replication {
		if key == "class" {
			continue
		}
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		parts = append(parts, formatCQLMapEntry(key, replication[key]))
	}
	return "{" + strings.Join(parts, ", ") + "}"
}

func formatCQLMapEntry(key, value string) string {
	return fmt.Sprintf("'%s': '%s'", escapeCQLString(key), escapeCQLString(value))
}

func escapeCQLString(value string) string {
	return strings.ReplaceAll(value, "'", "''")
}

func parseDescribeBool(value interface{}, fallback bool) bool {
	switch v := value.(type) {
	case bool:
		return v
	case string:
		if parsed, err := strconv.ParseBool(v); err == nil {
			return parsed
		}
	default:
		if parsed, err := strconv.ParseBool(fmt.Sprint(v)); err == nil {
			return parsed
		}
	}
	return fallback
}

func parseDescribeInt(value interface{}) int {
	switch v := value.(type) {
	case int:
		return v
	case int8:
		return int(v)
	case int16:
		return int(v)
	case int32:
		return int(v)
	case int64:
		return int(v)
	case uint:
		return int(v)
	case uint8:
		return int(v)
	case uint16:
		return int(v)
	case uint32:
		return int(v)
	case uint64:
		return int(v)
	case string:
		if parsed, err := strconv.Atoi(v); err == nil {
			return parsed
		}
	case []byte:
		if parsed, err := strconv.Atoi(string(v)); err == nil {
			return parsed
		}
	}
	return 0
}

func isLocalDescribeKeyspace(name string) bool {
	switch name {
	case "system", "system_schema", "system_accord":
		return true
	default:
		return false
	}
}

func describeEquals(column, value string) parser.Expression {
	return &parser.BinaryExpr{
		Op:   "=",
		Left: &parser.ColumnExpr{Name: column},
		Right: &parser.StringLiteral{
			Value: value,
		},
	}
}

func describeAnd(left, right parser.Expression) parser.Expression {
	if left == nil {
		return right
	}
	if right == nil {
		return left
	}
	return &parser.BinaryExpr{
		Op:    "AND",
		Left:  left,
		Right: right,
	}
}

type describeTableDefinition struct {
	Keyspace        string
	Table           string
	Columns         []types.ColumnInfo
	Defaults        map[string]string
	OnUpdateExprs   map[string]string
	PartitionKeys   []string
	ClusteringKeys  []string
	ClusteringOrder map[string]string
	TTLClause       string
	Indexes         []describeIndexDefinition
}

type keyPosition struct {
	name     string
	position int
}

func (e *Executor) loadDescribeTableDefinition(ctx context.Context, token parser.Position, keyspaceName, tableName, user, password string) (*describeTableDefinition, error) {
	if e == nil {
		return nil, nil
	}
	if def, ok := e.lookupDescribeTableDefinition(keyspaceName, tableName); ok {
		return def, nil
	}
	if e.systemExecutor == nil {
		return nil, nil
	}

	where := describeAnd(
		describeEquals("keyspace_name", keyspaceName),
		describeEquals("table_name", tableName),
	)
	selectStmt := describeSelectAll(token, "columns", where)
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	if result == nil || len(result.Rows) == 0 {
		return nil, nil
	}

	nameIndex := findColumnIndex(result.Columns, "column_name")
	typeIndex := findColumnIndex(result.Columns, "type")
	kindIndex := findColumnIndex(result.Columns, "kind")
	positionIndex := findColumnIndex(result.Columns, "position")
	orderIndex := findColumnIndex(result.Columns, "clustering_order")
	defaultExprIndex := findColumnIndex(result.Columns, "default_expr")
	onUpdateExprIndex := findColumnIndex(result.Columns, "on_update_expr")
	if nameIndex == -1 || typeIndex == -1 || kindIndex == -1 {
		return nil, fmt.Errorf("system_schema.columns missing required columns for DESCRIBE")
	}

	columns := make([]types.ColumnInfo, 0, len(result.Rows))
	defaults := make(map[string]string)
	onUpdates := make(map[string]string)
	partitionKeys := make([]keyPosition, 0)
	clusteringKeys := make([]keyPosition, 0)
	var clusteringOrder map[string]string

	for _, row := range result.Rows {
		if nameIndex >= len(row.Values) || row.Values[nameIndex] == nil {
			continue
		}
		name := fmt.Sprint(row.Values[nameIndex])
		if name == "" {
			continue
		}
		colType := ""
		if typeIndex < len(row.Values) && row.Values[typeIndex] != nil {
			colType = fmt.Sprint(row.Values[typeIndex])
		}
		columns = append(columns, types.ColumnInfo{
			Keyspace: keyspaceName,
			Table:    tableName,
			Name:     name,
			Type:     colType,
		})
		if defaultExprIndex != -1 && defaultExprIndex < len(row.Values) && row.Values[defaultExprIndex] != nil {
			expr := strings.TrimSpace(fmt.Sprint(row.Values[defaultExprIndex]))
			if expr != "" {
				defaults[name] = expr
			}
		}
		if onUpdateExprIndex != -1 && onUpdateExprIndex < len(row.Values) && row.Values[onUpdateExprIndex] != nil {
			expr := strings.TrimSpace(fmt.Sprint(row.Values[onUpdateExprIndex]))
			if expr != "" {
				onUpdates[name] = expr
			}
		}

		kind := ""
		if kindIndex < len(row.Values) && row.Values[kindIndex] != nil {
			kind = strings.ToLower(fmt.Sprint(row.Values[kindIndex]))
		}
		position := 0
		if positionIndex != -1 && positionIndex < len(row.Values) && row.Values[positionIndex] != nil {
			position = parseDescribeInt(row.Values[positionIndex])
		}

		switch kind {
		case "partition_key":
			partitionKeys = append(partitionKeys, keyPosition{name: name, position: position})
		case "clustering":
			clusteringKeys = append(clusteringKeys, keyPosition{name: name, position: position})
			if orderIndex != -1 && orderIndex < len(row.Values) && row.Values[orderIndex] != nil {
				order := strings.ToUpper(fmt.Sprint(row.Values[orderIndex]))
				if order != "" && order != "NONE" {
					if clusteringOrder == nil {
						clusteringOrder = make(map[string]string)
					}
					clusteringOrder[name] = order
				}
			}
		}
	}

	sort.Slice(partitionKeys, func(i, j int) bool {
		if partitionKeys[i].position == partitionKeys[j].position {
			return partitionKeys[i].name < partitionKeys[j].name
		}
		return partitionKeys[i].position < partitionKeys[j].position
	})
	sort.Slice(clusteringKeys, func(i, j int) bool {
		if clusteringKeys[i].position == clusteringKeys[j].position {
			return clusteringKeys[i].name < clusteringKeys[j].name
		}
		return clusteringKeys[i].position < clusteringKeys[j].position
	})

	partitionNames := make([]string, 0, len(partitionKeys))
	for _, entry := range partitionKeys {
		partitionNames = append(partitionNames, entry.name)
	}
	clusteringNames := make([]string, 0, len(clusteringKeys))
	for _, entry := range clusteringKeys {
		clusteringNames = append(clusteringNames, entry.name)
	}

	indexes, err := e.loadDescribeIndexDefinitions(ctx, token, keyspaceName, tableName, user, password)
	if err != nil {
		return nil, err
	}
	ttlClause, err := e.loadDescribeTTLClause(ctx, keyspaceName, tableName, user, password)
	if err != nil {
		return nil, err
	}

	return &describeTableDefinition{
		Keyspace:        keyspaceName,
		Table:           tableName,
		Columns:         columns,
		Defaults:        defaults,
		OnUpdateExprs:   onUpdates,
		PartitionKeys:   partitionNames,
		ClusteringKeys:  clusteringNames,
		ClusteringOrder: clusteringOrder,
		TTLClause:       ttlClause,
		Indexes:         indexes,
	}, nil
}

func (e *Executor) lookupDescribeTableDefinition(keyspace, table string) (*describeTableDefinition, bool) {
	if e == nil {
		return nil, false
	}
	fullName := fmt.Sprintf("%s.%s", keyspace, table)
	if e.systemExecutor != nil {
		if st, ok := e.systemExecutor.LookupTable(fullName); ok {
			return &describeTableDefinition{
				Keyspace:        st.Keyspace,
				Table:           st.Table,
				Columns:         st.Columns,
				Defaults:        map[string]string{},
				OnUpdateExprs:   map[string]string{},
				PartitionKeys:   st.PartitionKeys,
				ClusteringKeys:  st.ClusteringKeys,
				ClusteringOrder: st.ClusteringOrder,
				TTLClause:       "",
			}, true
		}
	}
	if e.virtualExecutor != nil {
		if vt, ok := e.virtualExecutor.LookupTable(fullName); ok {
			return &describeTableDefinition{
				Keyspace:        vt.Keyspace,
				Table:           vt.Table,
				Columns:         vt.Columns,
				Defaults:        map[string]string{},
				OnUpdateExprs:   map[string]string{},
				PartitionKeys:   vt.PartitionKeys,
				ClusteringKeys:  vt.ClusteringKeys,
				ClusteringOrder: vt.ClusteringOrder,
				TTLClause:       "",
			}, true
		}
	}
	return nil, false
}

func buildCreateStatement(def *describeTableDefinition) (string, error) {
	if def == nil {
		return "", fmt.Errorf("describe table definition is nil")
	}
	if len(def.PartitionKeys) == 0 {
		return "", fmt.Errorf("table %s.%s missing partition key", def.Keyspace, def.Table)
	}

	var builder strings.Builder
	builder.WriteString("CREATE TABLE ")
	builder.WriteString(quoteCQLIdentifier(def.Keyspace))
	builder.WriteString(".")
	builder.WriteString(quoteCQLIdentifier(def.Table))
	builder.WriteString(" (\n")
	for _, col := range def.Columns {
		builder.WriteString("  ")
		builder.WriteString(quoteCQLIdentifier(col.Name))
		builder.WriteString(" ")
		builder.WriteString(col.Type)
		if def.Defaults != nil {
			if expr, ok := def.Defaults[col.Name]; ok && strings.TrimSpace(expr) != "" {
				builder.WriteString(" DEFAULT ")
				builder.WriteString(expr)
			}
		}
		if def.OnUpdateExprs != nil {
			if expr, ok := def.OnUpdateExprs[col.Name]; ok && strings.TrimSpace(expr) != "" {
				builder.WriteString(" ON UPDATE ")
				builder.WriteString(expr)
			}
		}
		builder.WriteString(",\n")
	}
	builder.WriteString("  ")
	builder.WriteString(formatPrimaryKey(def.PartitionKeys, def.ClusteringKeys))
	builder.WriteString("\n")
	builder.WriteString(")")
	if hasNonDefaultClusteringOrder(def.ClusteringKeys, def.ClusteringOrder) {
		builder.WriteString(" WITH CLUSTERING ORDER BY (")
		builder.WriteString(formatClusteringOrder(def.ClusteringKeys, def.ClusteringOrder))
		builder.WriteString(")")
	}
	if strings.TrimSpace(def.TTLClause) != "" {
		builder.WriteString(" ")
		builder.WriteString(def.TTLClause)
	}
	builder.WriteString(";")
	if len(def.Indexes) > 0 {
		builder.WriteString("\n\n")
		builder.WriteString(formatIndexStatements(def))
	}

	return builder.String(), nil
}

type describeIndexDefinition struct {
	Name          string
	TargetColumns []string
}

func (e *Executor) loadDescribeIndexDefinitions(ctx context.Context, token parser.Position, keyspaceName, tableName, user, password string) ([]describeIndexDefinition, error) {
	if e == nil || e.systemExecutor == nil {
		return nil, nil
	}
	where := describeAnd(
		describeEquals("keyspace_name", keyspaceName),
		describeEquals("table_name", tableName),
	)
	selectStmt := describeSelectAll(token, "indexes", where)
	result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
	if err != nil {
		return nil, err
	}
	if result == nil || len(result.Rows) == 0 {
		return nil, nil
	}

	nameIndex := findColumnIndex(result.Columns, "index_name")
	optionsIndex := findColumnIndex(result.Columns, "options")
	if nameIndex == -1 || optionsIndex == -1 {
		return nil, fmt.Errorf("system_schema.indexes missing required columns for DESCRIBE")
	}

	indexes := make([]describeIndexDefinition, 0, len(result.Rows))
	for _, row := range result.Rows {
		if nameIndex >= len(row.Values) || row.Values[nameIndex] == nil {
			continue
		}
		name := fmt.Sprint(row.Values[nameIndex])
		if name == "" {
			continue
		}
		var target []string
		if optionsIndex < len(row.Values) && row.Values[optionsIndex] != nil {
			target = describeIndexTarget(row.Values[optionsIndex])
		}
		if len(target) == 0 {
			continue
		}
		indexes = append(indexes, describeIndexDefinition{Name: name, TargetColumns: target})
	}

	sort.Slice(indexes, func(i, j int) bool {
		left := strings.ToLower(indexes[i].Name)
		right := strings.ToLower(indexes[j].Name)
		if left == right {
			return indexes[i].Name < indexes[j].Name
		}
		return left < right
	})

	return indexes, nil
}

func describeIndexTarget(value interface{}) []string {
	options := normalizeDescribeOptions(value)
	target, ok := options["target"]
	if !ok {
		return nil
	}
	switch v := target.(type) {
	case []string:
		out := make([]string, 0, len(v))
		for _, item := range v {
			if item != "" {
				out = append(out, item)
			}
		}
		return out
	case []interface{}:
		out := make([]string, 0, len(v))
		for _, item := range v {
			val := fmt.Sprint(item)
			if val != "" {
				out = append(out, val)
			}
		}
		return out
	case string:
		if v == "" {
			return nil
		}
		return []string{v}
	default:
		val := fmt.Sprint(v)
		if val == "" {
			return nil
		}
		return []string{val}
	}
}

func normalizeDescribeOptions(value interface{}) map[string]interface{} {
	normalized := make(map[string]interface{})
	switch v := value.(type) {
	case map[string]string:
		for key, val := range v {
			normalized[key] = val
		}
	case map[string]interface{}:
		for key, val := range v {
			normalized[fmt.Sprint(key)] = val
		}
	case map[interface{}]interface{}:
		for key, val := range v {
			normalized[fmt.Sprint(key)] = val
		}
	}
	return normalized
}

func (e *Executor) loadDescribeTTLClause(ctx context.Context, keyspaceName, tableName, user, password string) (string, error) {
	if e == nil || e.poolManager == nil {
		return "", nil
	}
	if cqlkeyspace.IsInternalSchema(keyspaceName) || cqlkeyspace.IsBuiltIn(keyspaceName) {
		return "", nil
	}

	conn, release, err := e.getConn(ctx, user, password)
	if err != nil {
		return "", err
	}
	if release != nil {
		defer release()
	}

	showSQL := fmt.Sprintf("SHOW CREATE TABLE %s.%s", quoteMySQLIdentifier(keyspaceName), quoteMySQLIdentifier(tableName))
	rows, err := conn.QueryContext(ctx, showSQL)
	if err != nil {
		if e.logger != nil {
			e.logger.Debug("SHOW CREATE TABLE failed while building DESCRIBE TABLE output",
				zap.String("keyspace", keyspaceName),
				zap.String("table", tableName),
				zap.Error(err))
		}
		return "", nil
	}
	defer rows.Close()

	columns, err := rows.Columns()
	if err != nil || len(columns) == 0 {
		return "", nil
	}
	if !rows.Next() {
		return "", nil
	}

	values := make([]interface{}, len(columns))
	dest := make([]interface{}, len(columns))
	for i := range values {
		dest[i] = &values[i]
	}
	if err := rows.Scan(dest...); err != nil {
		return "", nil
	}
	if err := rows.Err(); err != nil {
		return "", nil
	}

	createSQL := ""
	for i, col := range columns {
		if strings.EqualFold(col, "Create Table") {
			createSQL = describeCellString(values[i])
			break
		}
	}
	if createSQL == "" {
		for i := len(values) - 1; i >= 0; i-- {
			if s := strings.TrimSpace(describeCellString(values[i])); s != "" {
				createSQL = s
				break
			}
		}
	}
	return extractTiDBTTLClauseFromShowCreate(createSQL), nil
}

func describeCellString(value interface{}) string {
	switch v := value.(type) {
	case nil:
		return ""
	case string:
		return v
	case []byte:
		return string(v)
	default:
		return fmt.Sprint(v)
	}
}

func extractTiDBTTLClauseFromShowCreate(createSQL string) string {
	sqlText := strings.TrimSpace(stripTiDBShowCreateHints(createSQL))
	if sqlText == "" {
		return ""
	}

	ttlLoc := describeShowCreateTTLStartRE.FindStringIndex(sqlText)
	if ttlLoc == nil {
		return ""
	}
	ttlExpr := strings.TrimSpace(sqlText[ttlLoc[1]:])
	if nextLoc := describeShowCreateTTLNextRE.FindStringIndex(ttlExpr); nextLoc != nil {
		ttlExpr = ttlExpr[:nextLoc[0]]
	}
	ttlExpr = strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(ttlExpr), ";"))
	if ttlExpr == "" {
		return ""
	}

	var builder strings.Builder
	builder.WriteString("ttl = ")
	builder.WriteString(ttlExpr)

	if enableMatch := describeShowCreateTTLEnable.FindStringSubmatch(sqlText); len(enableMatch) >= 2 {
		builder.WriteString(" ttl_enable = ")
		builder.WriteString(strings.TrimSpace(enableMatch[1]))
	}
	if intervalMatch := describeShowCreateTTLJobInt.FindStringSubmatch(sqlText); len(intervalMatch) >= 2 {
		builder.WriteString(" ttl_job_interval = ")
		builder.WriteString(strings.TrimSpace(intervalMatch[1]))
	}

	return builder.String()
}

func stripTiDBShowCreateHints(sqlText string) string {
	normalized := describeShowCreateHintOpenRE.ReplaceAllString(sqlText, "")
	normalized = describeShowCreateHintClose.ReplaceAllString(normalized, " ")
	return strings.TrimSpace(normalized)
}

func formatIndexStatements(def *describeTableDefinition) string {
	if def == nil || len(def.Indexes) == 0 {
		return ""
	}
	var builder strings.Builder
	for i, idx := range def.Indexes {
		if i > 0 {
			builder.WriteString("\n")
		}
		builder.WriteString("CREATE INDEX ")
		builder.WriteString(quoteCQLIdentifier(idx.Name))
		builder.WriteString(" ON ")
		builder.WriteString(quoteCQLIdentifier(def.Keyspace))
		builder.WriteString(".")
		builder.WriteString(quoteCQLIdentifier(def.Table))
		builder.WriteString(" (")
		builder.WriteString(joinIdentifiers(idx.TargetColumns))
		builder.WriteString(");")
	}
	return builder.String()
}

func formatPrimaryKey(partitionKeys, clusteringKeys []string) string {
	partitionSpec := joinIdentifiers(partitionKeys)
	if len(partitionKeys) > 1 {
		partitionSpec = "(" + partitionSpec + ")"
	}
	if len(clusteringKeys) == 0 {
		return fmt.Sprintf("PRIMARY KEY (%s)", partitionSpec)
	}
	return fmt.Sprintf("PRIMARY KEY (%s, %s)", partitionSpec, joinIdentifiers(clusteringKeys))
}

func formatClusteringOrder(clusteringKeys []string, order map[string]string) string {
	orders := make([]string, 0, len(clusteringKeys))
	for _, col := range clusteringKeys {
		direction := "ASC"
		if order != nil {
			if value, ok := order[col]; ok && value != "" {
				direction = value
			}
		}
		orders = append(orders, fmt.Sprintf("%s %s", quoteCQLIdentifier(col), strings.ToUpper(direction)))
	}
	return strings.Join(orders, ", ")
}

func hasNonDefaultClusteringOrder(clusteringKeys []string, order map[string]string) bool {
	if len(clusteringKeys) == 0 || len(order) == 0 {
		return false
	}
	for _, col := range clusteringKeys {
		if value, ok := order[col]; ok && strings.ToUpper(value) != "ASC" {
			return true
		}
	}
	return false
}

func joinIdentifiers(names []string) string {
	quoted := make([]string, 0, len(names))
	for _, name := range names {
		quoted = append(quoted, quoteCQLIdentifier(name))
	}
	return strings.Join(quoted, ", ")
}

func quoteCQLIdentifier(name string) string {
	if isSimpleCQLIdentifier(name) {
		return name
	}
	return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
}

func isSimpleCQLIdentifier(name string) bool {
	if name == "" {
		return false
	}
	if name != strings.ToLower(name) {
		return false
	}
	for i := 0; i < len(name); i++ {
		ch := name[i]
		if ch >= 'a' && ch <= 'z' {
			continue
		}
		if ch == '_' {
			continue
		}
		if i > 0 && ch >= '0' && ch <= '9' {
			continue
		}
		return false
	}
	return true
}