package executor

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strings"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/cqlerrors"
	"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/translator"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

const (
	indexPrefixBytesPerChar       = 4
	defaultSecondaryIndexMaxBytes = 3072
	defaultSecondaryIndexMinChars = 8
	defaultSecondaryIndexMinBytes = 8
)

func (e *ddlExecutor) executeCreateKeyspace(ctx context.Context, user, password string, stmt *parser.CreateKeyspaceStatement) (*QueryResult, error) {
	if err := e.validateKeyspaceName(stmt.Name); err != nil {
		return nil, err
	}
	if keyspace.IsBuiltIn(stmt.Name) || keyspace.IsInternalSchema(stmt.Name) {
		return nil, cqlerrors.Invalid(fmt.Sprintf("keyspace %q is reserved", stmt.Name))
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	ddlSQL, err := e.translator.Translate(stmt)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.writeKeyspaceMeta(ctx, conn, stmt)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "CREATED", "KEYSPACE", stmt.Name, "")
}

func (e *ddlExecutor) executeDropKeyspace(ctx context.Context, user, password string, stmt *parser.DropKeyspaceStatement) (*QueryResult, error) {
	if err := e.validateKeyspaceName(stmt.Name); err != nil {
		return nil, err
	}
	if keyspace.IsBuiltIn(stmt.Name) || keyspace.IsInternalSchema(stmt.Name) {
		return nil, cqlerrors.Invalid(fmt.Sprintf("keyspace %q is reserved", stmt.Name))
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	if stmt.IfExists && e.schemaManager != nil {
		exists, err := e.schemaManager.SchemaExists(ctx, user, password, stmt.Name)
		if err != nil {
			return nil, err
		}
		if !exists {
			return &QueryResult{}, nil
		}
	}

	ddlSQL, err := e.translator.Translate(stmt)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.deleteKeyspaceMeta(ctx, conn, stmt.Name)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "DROPPED", "KEYSPACE", stmt.Name, "")
}

func (e *ddlExecutor) executeCreateTable(ctx context.Context, user, password string, stmt *parser.CreateTableStatement) (*QueryResult, error) {
	if err := e.validateTableName(stmt.Name); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Name.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Name.Schema); err != nil {
		return nil, err
	}
	for _, col := range stmt.Columns {
		if err := e.validateColumnName(col.Name); err != nil {
			return nil, err
		}
	}
	if _, hasRaw, err := translator.TTLRawClauseFromOptions(stmt.Options, false); err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	} else if hasRaw && !e.extendedProtocolModeEnabled() {
		return nil, cqlerrors.Invalid(fmt.Sprintf("%s is only supported in executor.protocol_mode=%q", translator.TTLRawOptionName, config.ExecutorProtocolModeExtendedSQL))
	}
	if err := e.validateCreateTableExtensions(stmt); err != nil {
		return nil, err
	}
	if err := validateCounterTableDefinition(stmt); err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	ddlSQL, err := e.translator.Translate(stmt)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.writeTableMeta(ctx, conn, stmt)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	result, err = e.withSchemaChange(ctx, user, password, result, "CREATED", "TABLE", stmt.Name.Schema, stmt.Name.Name)
	if err != nil {
		return nil, err
	}
	e.launchPostCreateTableTasks(user, password, stmt.Name.Schema, stmt.Name.Name)
	return result, nil
}

func (e *ddlExecutor) executeDropTable(ctx context.Context, user, password string, stmt *parser.DropTableStatement) (*QueryResult, error) {
	if err := e.validateTableName(stmt.Name); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Name.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Name.Schema); err != nil {
		return nil, err
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	if stmt.IfExists {
		exists, err := e.schemaManager.TableExists(ctx, user, password, stmt.Name.Schema, stmt.Name.Name)
		if err != nil {
			return nil, err
		}
		if !exists {
			return &QueryResult{}, nil
		}
	}

	ddlSQL, err := e.translator.Translate(stmt)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.deleteTableMeta(ctx, conn, stmt.Name.Schema, stmt.Name.Name)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "DROPPED", "TABLE", stmt.Name.Schema, stmt.Name.Name)
}

func (e *ddlExecutor) executeAlterTable(ctx context.Context, user, password string, stmt *parser.AlterTableStatement) (*QueryResult, error) {
	if err := e.validateTableName(stmt.Table); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Table.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Table.Schema); err != nil {
		return nil, err
	}

	tableExists, err := e.schemaManager.TableExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name)
	if err != nil {
		return nil, err
	}
	if !tableExists {
		if stmt.IfExists {
			return &QueryResult{}, nil
		}
		return nil, cqlerrors.Invalid(fmt.Sprintf("table %q.%q does not exist", stmt.Table.Schema, stmt.Table.Name))
	}

	switch stmt.Operation {
	case "ADD":
		if err := e.validateColumnName(stmt.Column); err != nil {
			return nil, err
		}
		if stmt.IfNotExists {
			exists, err := e.ensureColumnExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name, stmt.Column)
			if err != nil {
				return nil, err
			}
			if exists {
				return &QueryResult{}, nil
			}
		}
		columnType := cqlTypeString(stmt.Type, stmt.Table.Schema)
		columnTypes, err := e.lookupColumnTypes(ctx, user, password, stmt.Table)
		if err != nil {
			return nil, err
		}
		pkCols, err := e.primaryKeyColumns(ctx, user, password, stmt.Table)
		if err != nil {
			return nil, err
		}
		if err := validateCounterTableAlterAdd(columnType, columnTypes, pkCols); err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
	case "DROP":
		if err := e.validateColumnName(stmt.Column); err != nil {
			return nil, err
		}
		if stmt.ColumnIfExists {
			exists, err := e.ensureColumnExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name, stmt.Column)
			if err != nil {
				return nil, err
			}
			if !exists {
				return &QueryResult{}, nil
			}
		}
	case "RENAME":
		if len(stmt.Renames) == 0 {
			return nil, cqlerrors.Invalid("RENAME requires at least one column")
		}
		if stmt.ColumnIfExists {
			renames := make([]parser.RenamePair, 0, len(stmt.Renames))
			for _, pair := range stmt.Renames {
				exists, err := e.ensureColumnExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name, pair.From)
				if err != nil {
					return nil, err
				}
				if exists {
					renames = append(renames, pair)
				}
			}
			if len(renames) == 0 {
				return &QueryResult{}, nil
			}
			stmt.Renames = renames
		}
	case "WITH":
		if len(stmt.Options) != 1 {
			return nil, cqlerrors.Invalid(fmt.Sprintf("%s cannot be combined with other ALTER TABLE WITH options", translator.TTLRawOptionName))
		}
		_, hasTTLRaw, err := translator.TTLRawClauseFromOptions(stmt.Options, true)
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
		if !hasTTLRaw {
			return nil, cqlerrors.Invalid(fmt.Sprintf("only %s is supported in ALTER TABLE WITH", translator.TTLRawOptionName))
		}
		if !e.extendedProtocolModeEnabled() {
			return nil, cqlerrors.Invalid(fmt.Sprintf("%s is only supported in executor.protocol_mode=%q", translator.TTLRawOptionName, config.ExecutorProtocolModeExtendedSQL))
		}
	default:
		return nil, cqlerrors.Invalid(fmt.Sprintf("unsupported ALTER TABLE operation %q", stmt.Operation))
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	ddlSQL, err := e.translator.Translate(stmt)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.alterTableMeta(ctx, conn, stmt)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "UPDATED", "TABLE", stmt.Table.Schema, stmt.Table.Name)
}

func (e *ddlExecutor) executeCreateIndex(ctx context.Context, user, password string, stmt *parser.CreateIndexStatement) (*QueryResult, error) {
	if err := e.validateTableName(stmt.Table); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Table.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Table.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureTableExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name); err != nil {
		return nil, err
	}
	if err := e.validateIndexColumns(stmt.Columns); err != nil {
		return nil, err
	}
	if stmt.Name == "" {
		stmt.Name = defaultIndexName(stmt.Table.Name, stmt.Columns)
	}
	if err := e.validateIndexName(stmt.Name); err != nil {
		return nil, err
	}

	indexes, err := e.schemaManager.IndexesForTable(ctx, user, password, stmt.Table.Schema, stmt.Table.Name)
	if err != nil {
		return nil, err
	}
	if dup := findDuplicateIndexName(indexes, stmt.Columns); dup != "" && !strings.EqualFold(dup, stmt.Name) {
		return nil, cqlerrors.Invalid(fmt.Sprintf("Index %s is a duplicate of existing index %s", stmt.Name, dup))
	}

	if stmt.IfNotExists {
		exists, err := e.schemaManager.IndexExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name, stmt.Name)
		if err != nil {
			return nil, err
		}
		if exists {
			return &QueryResult{}, nil
		}
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	ddlSQL, err := e.buildCreateIndexDDL(ctx, user, password, stmt)
	if err != nil {
		return nil, err
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.writeIndexMeta(ctx, conn, stmt)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "UPDATED", "TABLE", stmt.Table.Schema, stmt.Table.Name)
}

func findDuplicateIndexName(indexes []mysqlbackend.IndexInfo, columns []string) string {
	if len(columns) == 0 {
		return ""
	}
	target := make([]string, len(columns))
	for i, col := range columns {
		target[i] = strings.ToLower(col)
	}
	byName := make(map[string][]mysqlbackend.IndexInfo)
	for _, idx := range indexes {
		if idx.IsPrimary || idx.Column == "" {
			continue
		}
		byName[idx.Name] = append(byName[idx.Name], idx)
	}
	dup := ""
	dupLower := ""
	for name, entries := range byName {
		sorted := append([]mysqlbackend.IndexInfo(nil), entries...)
		sort.Slice(sorted, func(i, j int) bool { return sorted[i].Seq < sorted[j].Seq })
		if len(sorted) != len(target) {
			continue
		}
		match := true
		for i, idx := range sorted {
			if strings.ToLower(idx.Column) != target[i] {
				match = false
				break
			}
		}
		if !match {
			continue
		}
		lower := strings.ToLower(name)
		if dup == "" || lower < dupLower || (lower == dupLower && name < dup) {
			dup = name
			dupLower = lower
		}
	}
	return dup
}

func (e *ddlExecutor) buildCreateIndexDDL(ctx context.Context, user, password string, stmt *parser.CreateIndexStatement) (string, error) {
	if e.schemaManager == nil {
		return "", cqlerrors.Invalid("schema manager not initialized")
	}
	columnInfos, err := e.schemaManager.ColumnsForTable(ctx, user, password, stmt.Table.Schema, stmt.Table.Name)
	if err != nil {
		return "", err
	}
	columnTypes := make(map[string]mysqlbackend.ColumnInfo, len(columnInfos))
	for _, info := range columnInfos {
		columnTypes[strings.ToLower(info.Name)] = info
	}
	prefixes, err := computeSecondaryIndexPrefixes(stmt.Columns, columnTypes, config.GetConfig().MySQL)
	if err != nil {
		return "", err
	}
	return buildCreateIndexSQL(stmt.Table.Schema, stmt.Table.Name, stmt.Name, stmt.Columns, prefixes), nil
}

type secondaryIndexPrefixColumn struct {
	nameLower string
	isBinary  bool
	minBytes  int
	maxBytes  int
}

func computeSecondaryIndexPrefixes(columns []string, columnTypes map[string]mysqlbackend.ColumnInfo, cfg config.MySQLConfig) (map[string]int, error) {
	maxBytes := cfg.MaxIndexBytes
	if maxBytes <= 0 {
		maxBytes = defaultSecondaryIndexMaxBytes
	}
	minChars := cfg.MinIndexPrefixChars
	if minChars <= 0 {
		minChars = defaultSecondaryIndexMinChars
	}
	minBytes := cfg.MinIndexPrefixBytes
	if minBytes <= 0 {
		minBytes = defaultSecondaryIndexMinBytes
	}

	prefixCols := make([]secondaryIndexPrefixColumn, 0, len(columns))
	for _, name := range columns {
		lookup := strings.ToLower(name)
		info, ok := columnTypes[lookup]
		if !ok {
			return nil, cqlerrors.Invalid(fmt.Sprintf("column %q does not exist", name))
		}
		required, isBinary, reqMinBytes, reqMaxBytes := classifySecondaryIndexPrefix(info.DataType, maxBytes, minChars, minBytes)
		if !required {
			continue
		}
		if cfg.DisableTextBlobSecondaryIndex {
			return nil, cqlerrors.Invalid(fmt.Sprintf("Index on BLOB/TEXT column %q is not supported", name))
		}
		if reqMinBytes > reqMaxBytes {
			reqMinBytes = reqMaxBytes
		}
		prefixCols = append(prefixCols, secondaryIndexPrefixColumn{
			nameLower: lookup,
			isBinary:  isBinary,
			minBytes:  reqMinBytes,
			maxBytes:  reqMaxBytes,
		})
	}
	if len(prefixCols) == 0 {
		return nil, nil
	}

	minTotal := 0
	for _, col := range prefixCols {
		minTotal += col.minBytes
	}
	if minTotal > maxBytes {
		return nil, cqlerrors.Invalid(fmt.Sprintf("index minimum prefix length %d exceeds max index length %d", minTotal, maxBytes))
	}

	prefixes := make(map[string]int, len(prefixCols))
	remainingBudget := maxBytes
	remainingMin := minTotal
	for _, col := range prefixCols {
		remainingMin -= col.minBytes
		allocBytes := remainingBudget - remainingMin
		if allocBytes > col.maxBytes {
			allocBytes = col.maxBytes
		}
		if allocBytes < col.minBytes {
			allocBytes = col.minBytes
		}
		if col.isBinary {
			prefixes[col.nameLower] = allocBytes
			remainingBudget -= allocBytes
			continue
		}

		prefixChars := allocBytes / indexPrefixBytesPerChar
		if prefixChars <= 0 {
			prefixChars = 1
		}
		minCharsNeeded := (col.minBytes + indexPrefixBytesPerChar - 1) / indexPrefixBytesPerChar
		if prefixChars < minCharsNeeded {
			prefixChars = minCharsNeeded
		}
		maxChars := col.maxBytes / indexPrefixBytesPerChar
		if maxChars > 0 && prefixChars > maxChars {
			prefixChars = maxChars
		}
		prefixes[col.nameLower] = prefixChars
		remainingBudget -= prefixChars * indexPrefixBytesPerChar
	}
	return prefixes, nil
}

func classifySecondaryIndexPrefix(dataType string, maxBytes, minChars, minBytes int) (required bool, isBinary bool, reqMinBytes int, reqMaxBytes int) {
	switch strings.ToLower(strings.TrimSpace(dataType)) {
	case "text", "tinytext", "mediumtext", "longtext":
		return true, false, minChars * indexPrefixBytesPerChar, maxBytes
	case "blob", "tinyblob", "mediumblob", "longblob":
		return true, true, minBytes, maxBytes
	default:
		return false, false, 0, 0
	}
}

func buildCreateIndexSQL(schema, table, index string, columns []string, prefixes map[string]int) string {
	var builder strings.Builder
	builder.WriteString("CREATE INDEX ")
	builder.WriteString(quoteMySQLIdentifier(index))
	builder.WriteString(" ON ")
	builder.WriteString(quoteMySQLIdentifier(schema))
	builder.WriteString(".")
	builder.WriteString(quoteMySQLIdentifier(table))
	builder.WriteString(" (")
	for i, col := range columns {
		if i > 0 {
			builder.WriteString(", ")
		}
		colName := quoteMySQLIdentifier(col)
		if prefix, ok := prefixes[strings.ToLower(col)]; ok && prefix > 0 {
			colName = fmt.Sprintf("%s(%d)", colName, prefix)
		}
		builder.WriteString(colName)
	}
	builder.WriteString(")")
	return builder.String()
}

func (e *ddlExecutor) executeDropIndex(ctx context.Context, user, password string, stmt *parser.DropIndexStatement) (*QueryResult, error) {
	if stmt.Name == "" {
		return nil, cqlerrors.Invalid("index name is required")
	}
	schema, index := splitIndexName(stmt.Name)
	if schema == "" {
		return nil, cqlerrors.Invalid("keyspace is required for DROP INDEX")
	}
	if err := e.validateKeyspaceName(schema); err != nil {
		return nil, err
	}
	if err := e.validateIndexName(index); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, schema); err != nil {
		return nil, err
	}

	indexes, err := e.schemaManager.FindIndexesByName(ctx, user, password, schema, index)
	if err != nil {
		return nil, err
	}
	if len(indexes) == 0 {
		if stmt.IfExists {
			return &QueryResult{}, nil
		}
		return nil, cqlerrors.Invalid(fmt.Sprintf("index %q does not exist", stmt.Name))
	}

	tableName, err := resolveIndexTable(indexes)
	if err != nil {
		return nil, err
	}

	ddlSQL := buildDropIndexSQL(schema, tableName, index)
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.deleteIndexMeta(ctx, conn, schema, tableName, index)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "UPDATED", "TABLE", schema, tableName)
}

func (e *ddlExecutor) withSchemaChange(ctx context.Context, user, password string, result *QueryResult, changeType, target, keyspace, name string) (*QueryResult, error) {
	if result == nil {
		result = &QueryResult{}
	}
	if err := e.bumpSchemaVersion(ctx, user, password); err != nil {
		return nil, err
	}
	if e.prepared != nil {
		e.prepared.Clear()
	}
	if e.poolManager != nil {
		e.poolManager.ClearPrepared()
	}
	result.SchemaChange = &SchemaChange{
		ChangeType: changeType,
		Target:     target,
		Keyspace:   keyspace,
		Name:       name,
	}
	return result, nil
}

func (e *ddlExecutor) invalidateSchemaCache(user, password string) {
	if e.schemaManager == nil {
		e.invalidateBatchPartitionHintCache()
		return
	}
	e.schemaManager.Invalidate(user, password)
	e.invalidateBatchPartitionHintCache()
}

func defaultIndexName(table string, columns []string) string {
	if len(columns) == 0 {
		return fmt.Sprintf("%s_idx", table)
	}
	return fmt.Sprintf("%s_%s_idx", table, strings.Join(columns, "_"))
}

func splitIndexName(name string) (string, string) {
	parts := strings.SplitN(name, ".", 2)
	if len(parts) == 2 {
		return parts[0], parts[1]
	}
	return "", name
}

func resolveIndexTable(indexes []mysqlbackend.IndexInfo) (string, error) {
	table := ""
	for _, idx := range indexes {
		if table == "" {
			table = idx.Table
			continue
		}
		if table != idx.Table {
			return "", cqlerrors.Invalid("index name is ambiguous across tables")
		}
	}
	if table == "" {
		return "", cqlerrors.Invalid("index table not found")
	}
	return table, nil
}

func buildDropIndexSQL(schema, table, index string) string {
	var builder strings.Builder
	builder.WriteString("DROP INDEX ")
	builder.WriteString(quoteMySQLIdentifier(index))
	builder.WriteString(" ON ")
	builder.WriteString(quoteMySQLIdentifier(schema))
	builder.WriteString(".")
	builder.WriteString(quoteMySQLIdentifier(table))
	return builder.String()
}

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

func (e *ddlExecutor) executeDDLWithSQL(ctx context.Context, user, password, sql string) (*QueryResult, error) {
	conn, release, err := e.getConn(ctx, user, password)
	if err != nil {
		return nil, cqlerrors.Invalid(fmt.Sprintf("failed to get connection: %s", err.Error()))
	}
	if release != nil {
		defer release()
	}
	return e.executeDDL(ctx, conn, sql)
}

func (e *ddlExecutor) executeDDLWithMeta(ctx context.Context, user, password, sql string, meta func(context.Context, *sql.Conn) error) (*QueryResult, error) {
	conn, release, err := e.getConn(ctx, user, password)
	if err != nil {
		return nil, cqlerrors.Invalid(fmt.Sprintf("failed to get connection: %s", err.Error()))
	}
	if release != nil {
		defer release()
	}
	if sql != "" {
		if _, err := conn.ExecContext(ctx, sql); err != nil {
			return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to execute DDL: %w", err), mysqlOpWrite, sql)
		}
	}
	if meta != nil {
		if err := meta(ctx, conn); err != nil {
			return nil, e.mapMySQLError(ctx, err, mysqlOpWrite)
		}
	}
	return &QueryResult{}, nil
}