package translator

import (
	"encoding/hex"
	"fmt"
	"strconv"
	"strings"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/keyspace"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
)

const (
	mysqlDefaultCharset     = "utf8mb4"
	mysqlDefaultCollation   = "utf8mb4_bin"
	mysqlDefaultEngine      = "InnoDB"
	maxBytesPerChar         = 4
	defaultMaxIndexBytes    = 3072
	defaultMinPrefixChars   = 8
	defaultMinPrefixBytes   = 8
	defaultDecimalPrecision = 65
	defaultDecimalScale     = 30
)

// Translator converts CQL statements to SQL queries
type Translator struct {
	// Configurable options for translation
	QuoteIdentifiers            bool
	indexMaxBytes               int
	minPrefixChars              int
	minPrefixBytes              int
	partitionDefaultSize        int
	enableClusteredIndex        bool
	disableTextBlobPartitionKey bool
}

// NewTranslator creates a new translator
func NewTranslator() *Translator {
	cfg := config.GetConfig()
	maxBytes := cfg.MySQL.MaxIndexBytes
	minChars := cfg.MySQL.MinIndexPrefixChars
	minBytes := cfg.MySQL.MinIndexPrefixBytes
	partitionSize := cfg.MySQL.PartitionDefaultSize
	if maxBytes <= 0 {
		maxBytes = defaultMaxIndexBytes
	}
	if minChars <= 0 {
		minChars = defaultMinPrefixChars
	}
	if minBytes <= 0 {
		minBytes = defaultMinPrefixBytes
	}
	if partitionSize <= 0 {
		partitionSize = 10
	}
	return &Translator{
		QuoteIdentifiers:            true,
		indexMaxBytes:               maxBytes,
		minPrefixChars:              minChars,
		minPrefixBytes:              minBytes,
		partitionDefaultSize:        partitionSize,
		enableClusteredIndex:        cfg.MySQL.EnableClusteredIndex,
		disableTextBlobPartitionKey: cfg.MySQL.DisableTextBlobPartitionKey,
	}
}

// Translate translates a CQL statement to SQL
func (t *Translator) Translate(stmt parser.Statement) (string, error) {
	switch s := stmt.(type) {
	case *parser.SelectStatement:
		return t.translateSelect(s)
	case *parser.InsertStatement:
		return t.translateInsert(s)
	case *parser.UpdateStatement:
		return t.translateUpdate(s)
	case *parser.DeleteStatement:
		return t.translateDelete(s)
	case *parser.TruncateStatement:
		return t.translateTruncate(s)
	case *parser.CreateTableStatement:
		return t.translateCreateTable(s)
	case *parser.DropTableStatement:
		return t.translateDropTable(s)
	case *parser.CreateKeyspaceStatement:
		return t.translateCreateKeyspace(s)
	case *parser.DropKeyspaceStatement:
		return t.translateDropKeyspace(s)
	case *parser.AlterTableStatement:
		return t.translateAlterTable(s)
	case *parser.CreateIndexStatement:
		return t.translateCreateIndex(s)
	case *parser.CreateMaterializedViewStatement:
		return t.translateCreateView(s)
	case *parser.DropMaterializedViewStatement:
		return t.translateDropView(s)
	default:
		return "", fmt.Errorf("unsupported statement type: %T", stmt)
	}
}

func (t *Translator) translateTruncate(stmt *parser.TruncateStatement) (string, error) {
	if stmt == nil {
		return "", fmt.Errorf("nil TRUNCATE statement")
	}
	if err := requireSchema(stmt.Table); err != nil {
		return "", err
	}
	return "TRUNCATE TABLE " + t.quoteIdentifier(stmt.Table.Schema) + "." + t.quoteIdentifier(stmt.Table.Name), nil
}

// translateSelect translates a SELECT statement to SQL
func (t *Translator) translateSelect(stmt *parser.SelectStatement) (string, error) {
	var sql strings.Builder

	if err := requireSchema(stmt.From); err != nil {
		return "", err
	}
	if stmt.Having != nil {
		return "", fmt.Errorf("HAVING is not supported")
	}

	// SELECT clause
	sql.WriteString("SELECT ")
	if stmt.Distinct {
		sql.WriteString("DISTINCT ")
	}

	// Column list
	for i, col := range stmt.Columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(t.translateExpression(col))
	}

	// FROM clause
	sql.WriteString(" FROM ")
	if stmt.From.Schema != "" {
		sql.WriteString(t.quoteIdentifier(stmt.From.Schema))
		sql.WriteString(".")
	}
	sql.WriteString(t.quoteIdentifier(stmt.From.Name))
	if stmt.IndexHint != "" {
		sql.WriteString(" FORCE INDEX (")
		sql.WriteString(t.quoteIdentifier(stmt.IndexHint))
		sql.WriteString(")")
	}

	// WHERE clause
	if stmt.Where != nil {
		sql.WriteString(" WHERE ")
		sql.WriteString(t.translateExpression(stmt.Where))
	}

	// GROUP BY clause
	if len(stmt.GroupBy) > 0 {
		sql.WriteString(" GROUP BY ")
		for i, expr := range stmt.GroupBy {
			if i > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString(t.translateExpression(expr))
		}
	}

	// ORDER BY clause (if present)
	if len(stmt.Options.OrderBy) > 0 {
		sql.WriteString(" ORDER BY ")
		for i, ob := range stmt.Options.OrderBy {
			if i > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString(t.quoteIdentifier(ob.Column))
			if ob.Direction == "DESC" {
				sql.WriteString(" DESC")
			} else {
				sql.WriteString(" ASC")
			}
		}
	}

	// LIMIT / PER PARTITION LIMIT clause
	limitExpr := stmt.Options.Limit
	ppExpr := stmt.Options.PerPartitionLimit
	hasLimitSQL := false
	if limitExpr != nil || ppExpr != nil {
		sql.WriteString(" LIMIT ")
		if limitExpr != nil && ppExpr != nil {
			limitValue, ok := limitLiteralValue(limitExpr)
			if !ok {
				return "", fmt.Errorf("LIMIT and PER PARTITION LIMIT require integer literals")
			}
			ppValue, ok := limitLiteralValue(ppExpr)
			if !ok {
				return "", fmt.Errorf("LIMIT and PER PARTITION LIMIT require integer literals")
			}
			if ppValue < limitValue {
				limitValue = ppValue
			}
			sql.WriteString(strconv.FormatInt(limitValue, 10))
		} else if limitExpr != nil {
			sql.WriteString(t.translateExpression(limitExpr))
		} else {
			sql.WriteString(t.translateExpression(ppExpr))
		}
		hasLimitSQL = true
	}
	if stmt.Options.Offset != nil {
		if !hasLimitSQL {
			return "", fmt.Errorf("OFFSET requires LIMIT")
		}
		sql.WriteString(" OFFSET ")
		sql.WriteString(t.translateExpression(stmt.Options.Offset))
	}

	return sql.String(), nil
}

func limitLiteralValue(expr parser.Expression) (int64, bool) {
	if expr == nil {
		return 0, false
	}
	switch e := expr.(type) {
	case *parser.IntegerLiteral:
		if e.Value < 0 {
			return 0, false
		}
		return e.Value, true
	case *parser.UnaryExpr:
		switch e.Op {
		case "+":
			return limitLiteralValue(e.Expr)
		case "-":
			return 0, false
		default:
			return 0, false
		}
	case *parser.TypeHintExpr:
		return limitLiteralValue(e.Expr)
	default:
		return 0, false
	}
}

// translateInsert translates an INSERT statement to SQL
func (t *Translator) translateInsert(stmt *parser.InsertStatement) (string, error) {
	var sql strings.Builder

	if err := requireSchema(stmt.Table); err != nil {
		return "", err
	}
	if stmt.JSON {
		return "", fmt.Errorf("INSERT JSON is not supported")
	}
	if stmt.IfNotExists {
		return "", fmt.Errorf("INSERT IF NOT EXISTS is not supported")
	}
	if stmt.TTL != nil || stmt.Timestamp != nil {
		return "", fmt.Errorf("USING TTL/TIMESTAMP is not supported")
	}

	sql.WriteString("INSERT INTO ")
	if stmt.Table.Schema != "" {
		sql.WriteString(t.quoteIdentifier(stmt.Table.Schema))
		sql.WriteString(".")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Table.Name))

	// Column names
	if len(stmt.Columns) > 0 {
		sql.WriteString(" (")
		for i, col := range stmt.Columns {
			if i > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString(t.quoteIdentifier(col))
		}
		sql.WriteString(")")
	}

	if len(stmt.Values) == 0 {
		return "", fmt.Errorf("VALUES clause is required")
	}

	// VALUES - CQL supports multiple value sets
	sql.WriteString(" VALUES ")
	for i, values := range stmt.Values {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString("(")
		for j, val := range values {
			if j > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString(t.translateExpression(val))
		}
		sql.WriteString(")")
	}

	return sql.String(), nil
}

// translateUpdate translates an UPDATE statement to SQL
func (t *Translator) translateUpdate(stmt *parser.UpdateStatement) (string, error) {
	var sql strings.Builder

	if err := requireSchema(stmt.Table); err != nil {
		return "", err
	}
	if stmt.IfExists || stmt.IfCondition != nil {
		return "", fmt.Errorf("IF conditions are not supported")
	}
	if stmt.TTL != nil || stmt.Timestamp != nil {
		return "", fmt.Errorf("USING TTL/TIMESTAMP is not supported")
	}

	sql.WriteString("UPDATE ")
	if stmt.Table.Schema != "" {
		sql.WriteString(t.quoteIdentifier(stmt.Table.Schema))
		sql.WriteString(".")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Table.Name))
	sql.WriteString(" SET ")

	// Set clauses
	for i, assignment := range stmt.Assigments {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(t.quoteIdentifier(assignment.Column))
		sql.WriteString(" = ")
		sql.WriteString(t.translateExpression(assignment.Value))
	}

	// WHERE clause (required in CQL)
	if stmt.Where != nil {
		sql.WriteString(" WHERE ")
		sql.WriteString(t.translateExpression(stmt.Where))
	}

	return sql.String(), nil
}

// translateDelete translates a DELETE statement to SQL
func (t *Translator) translateDelete(stmt *parser.DeleteStatement) (string, error) {
	var sql strings.Builder

	if err := requireSchema(stmt.From); err != nil {
		return "", err
	}
	if stmt.Timestamp != nil {
		return "", fmt.Errorf("USING TIMESTAMP is not supported")
	}
	if stmt.IfExists || stmt.IfCondition != nil {
		return "", fmt.Errorf("IF conditions are not supported")
	}
	if len(stmt.Targets) > 0 || len(stmt.Columns) > 0 {
		return "", fmt.Errorf("column-level DELETE is not supported")
	}

	sql.WriteString("DELETE FROM ")
	if stmt.From.Schema != "" {
		sql.WriteString(t.quoteIdentifier(stmt.From.Schema))
		sql.WriteString(".")
	}
	sql.WriteString(t.quoteIdentifier(stmt.From.Name))

	// WHERE clause (required in CQL)
	if stmt.Where != nil {
		sql.WriteString(" WHERE ")
		sql.WriteString(t.translateExpression(stmt.Where))
	}

	return sql.String(), nil
}

// translateCreateTable translates a CREATE TABLE statement to SQL
func (t *Translator) translateCreateTable(stmt *parser.CreateTableStatement) (string, error) {
	var sql strings.Builder

	if err := requireSchema(stmt.Name); err != nil {
		return "", err
	}
	if err := validateIdentifier("keyspace", stmt.Name.Schema); err != nil {
		return "", err
	}
	if err := validateIdentifier("table", stmt.Name.Name); err != nil {
		return "", err
	}

	sql.WriteString("CREATE TABLE ")
	if stmt.IfNotExists {
		sql.WriteString("IF NOT EXISTS ")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Name.Schema))
	sql.WriteString(".")
	sql.WriteString(t.quoteIdentifier(stmt.Name.Name))
	sql.WriteString(" (")

	// Column definitions
	var columnDefs []string
	ttlRawClause, hasTTLRaw, err := TTLRawClauseFromOptions(stmt.Options, false)
	if err != nil {
		return "", err
	}

	for _, colDef := range stmt.Columns {
		if err := validateIdentifier("column", colDef.Name); err != nil {
			return "", err
		}
		colSql := t.quoteIdentifier(colDef.Name)
		colSql += " " + t.translateType(colDef.Type)
		if colDef.HasDefault {
			defaultSQL, err := t.translateColumnDefaultExpression(colDef.DefaultExpr)
			if err != nil {
				return "", err
			}
			colSql += " DEFAULT " + defaultSQL
		}
		if colDef.HasOnUpdate {
			onUpdateSQL, err := t.translateColumnOnUpdateExpression(colDef.OnUpdateExpr)
			if err != nil {
				return "", err
			}
			colSql += " ON UPDATE " + onUpdateSQL
		}
		if colDef.Static {
			colSql += " /* STATIC */"
		}
		columnDefs = append(columnDefs, colSql)
	}

	pkClause, err := t.buildPrimaryKeyClause(stmt)
	if err != nil {
		return "", fmt.Errorf("primary key for %s: %w", stmt.Name, err)
	}
	if pkClause != "" {
		columnDefs = append(columnDefs, pkClause)
	}

	sql.WriteString(strings.Join(columnDefs, ", "))
	sql.WriteString(")")
	sql.WriteString(" ENGINE=")
	sql.WriteString(mysqlDefaultEngine)
	sql.WriteString(" DEFAULT CHARSET=")
	sql.WriteString(mysqlDefaultCharset)
	sql.WriteString(" COLLATE ")
	sql.WriteString(mysqlDefaultCollation)
	if hasTTLRaw {
		sql.WriteString(" ")
		sql.WriteString(ttlRawClause)
	}

	partitionClause, err := t.buildPartitionClause(stmt)
	if err != nil {
		return "", err
	}
	if partitionClause != "" {
		sql.WriteString(" ")
		sql.WriteString(partitionClause)
	}

	return sql.String(), nil
}

// translateDropTable translates a DROP TABLE statement to SQL
func (t *Translator) translateDropTable(stmt *parser.DropTableStatement) (string, error) {
	var sql strings.Builder
	if err := requireSchema(stmt.Name); err != nil {
		return "", err
	}
	sql.WriteString("DROP TABLE ")
	if stmt.IfExists {
		sql.WriteString("IF EXISTS ")
	}
	if stmt.Name.Schema != "" {
		sql.WriteString(t.quoteIdentifier(stmt.Name.Schema))
		sql.WriteString(".")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Name.Name))
	return sql.String(), nil
}

// translateCreateView translates a CREATE MATERIALIZED VIEW statement to SQL.
func (t *Translator) translateCreateView(stmt *parser.CreateMaterializedViewStatement) (string, error) {
	if stmt == nil || stmt.Name == nil {
		return "", fmt.Errorf("view name is required")
	}
	if err := requireSchema(stmt.Name); err != nil {
		return "", err
	}
	if err := validateIdentifier("keyspace", stmt.Name.Schema); err != nil {
		return "", err
	}
	if err := validateIdentifier("view", stmt.Name.Name); err != nil {
		return "", err
	}
	if stmt.Select == nil {
		return "", fmt.Errorf("view select statement is required")
	}

	selectSQL, err := t.translateSelect(stmt.Select)
	if err != nil {
		return "", err
	}

	var sql strings.Builder
	sql.WriteString("CREATE VIEW ")
	sql.WriteString(t.quoteIdentifier(stmt.Name.Schema))
	sql.WriteString(".")
	sql.WriteString(t.quoteIdentifier(stmt.Name.Name))
	sql.WriteString(" AS ")
	sql.WriteString(selectSQL)
	return sql.String(), nil
}

// translateDropView translates a DROP MATERIALIZED VIEW statement to SQL.
func (t *Translator) translateDropView(stmt *parser.DropMaterializedViewStatement) (string, error) {
	var sql strings.Builder
	if err := requireSchema(stmt.Name); err != nil {
		return "", err
	}
	sql.WriteString("DROP VIEW ")
	if stmt.IfExists {
		sql.WriteString("IF EXISTS ")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Name.Schema))
	sql.WriteString(".")
	sql.WriteString(t.quoteIdentifier(stmt.Name.Name))
	return sql.String(), nil
}

// translateCreateKeyspace translates a CREATE KEYSPACE statement to SQL
func (t *Translator) translateCreateKeyspace(stmt *parser.CreateKeyspaceStatement) (string, error) {
	// Translate keyspace to database
	var sql strings.Builder

	if err := validateIdentifier("keyspace", stmt.Name); err != nil {
		return "", err
	}
	sql.WriteString("CREATE DATABASE ")
	if stmt.IfNotExists {
		sql.WriteString("IF NOT EXISTS ")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Name))
	sql.WriteString(" DEFAULT CHARACTER SET ")
	sql.WriteString(mysqlDefaultCharset)
	sql.WriteString(" COLLATE ")
	sql.WriteString(mysqlDefaultCollation)

	return sql.String(), nil
}

// translateDropKeyspace translates a DROP KEYSPACE statement to SQL.
func (t *Translator) translateDropKeyspace(stmt *parser.DropKeyspaceStatement) (string, error) {
	var sql strings.Builder
	if err := validateIdentifier("keyspace", stmt.Name); err != nil {
		return "", err
	}
	sql.WriteString("DROP DATABASE ")
	if stmt.IfExists {
		sql.WriteString("IF EXISTS ")
	}
	sql.WriteString(t.quoteIdentifier(stmt.Name))
	return sql.String(), nil
}

// translateAlterTable translates ALTER TABLE ADD/DROP/RENAME operations to SQL.
func (t *Translator) translateAlterTable(stmt *parser.AlterTableStatement) (string, error) {
	var sql strings.Builder
	if err := requireSchema(stmt.Table); err != nil {
		return "", err
	}
	if err := validateIdentifier("keyspace", stmt.Table.Schema); err != nil {
		return "", err
	}
	if err := validateIdentifier("table", stmt.Table.Name); err != nil {
		return "", err
	}

	sql.WriteString("ALTER TABLE ")
	sql.WriteString(t.quoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(t.quoteIdentifier(stmt.Table.Name))

	switch stmt.Operation {
	case "ADD":
		if err := validateIdentifier("column", stmt.Column); err != nil {
			return "", err
		}
		sql.WriteString(" ADD COLUMN ")
		sql.WriteString(t.quoteIdentifier(stmt.Column))
		sql.WriteString(" ")
		sql.WriteString(t.translateType(stmt.Type))
	case "DROP":
		if err := validateIdentifier("column", stmt.Column); err != nil {
			return "", err
		}
		sql.WriteString(" DROP COLUMN ")
		sql.WriteString(t.quoteIdentifier(stmt.Column))
	case "RENAME":
		if len(stmt.Renames) == 0 {
			return "", fmt.Errorf("RENAME requires at least one column")
		}
		sql.WriteString(" ")
		for i, pair := range stmt.Renames {
			if err := validateIdentifier("column", pair.From); err != nil {
				return "", err
			}
			if err := validateIdentifier("column", pair.To); err != nil {
				return "", err
			}
			if i > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString("RENAME COLUMN ")
			sql.WriteString(t.quoteIdentifier(pair.From))
			sql.WriteString(" TO ")
			sql.WriteString(t.quoteIdentifier(pair.To))
		}
	case "WITH":
		ttlRawClause, hasTTLRaw, err := TTLRawClauseFromOptions(stmt.Options, true)
		if err != nil {
			return "", err
		}
		if !hasTTLRaw {
			return "", fmt.Errorf("only %s is supported in ALTER TABLE WITH", TTLRawOptionName)
		}
		sql.WriteString(" ")
		sql.WriteString(ttlRawClause)
	default:
		return "", fmt.Errorf("unsupported ALTER TABLE operation %q", stmt.Operation)
	}

	return sql.String(), nil
}

// translateCreateIndex translates CREATE INDEX for one or more columns.
func (t *Translator) translateCreateIndex(stmt *parser.CreateIndexStatement) (string, error) {
	var sql strings.Builder
	if err := requireSchema(stmt.Table); err != nil {
		return "", err
	}
	if err := validateIdentifier("keyspace", stmt.Table.Schema); err != nil {
		return "", err
	}
	if err := validateIdentifier("table", stmt.Table.Name); err != nil {
		return "", err
	}
	if len(stmt.Columns) == 0 {
		return "", fmt.Errorf("index column is required")
	}
	for _, col := range stmt.Columns {
		if err := validateIdentifier("column", col); err != nil {
			return "", err
		}
	}
	if stmt.Name == "" {
		return "", fmt.Errorf("index name is required")
	}
	if err := validateIdentifier("index", stmt.Name); err != nil {
		return "", err
	}

	sql.WriteString("CREATE INDEX ")
	sql.WriteString(t.quoteIdentifier(stmt.Name))
	sql.WriteString(" ON ")
	sql.WriteString(t.quoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(t.quoteIdentifier(stmt.Table.Name))
	sql.WriteString(" (")
	for i, col := range stmt.Columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(t.quoteIdentifier(col))
	}
	sql.WriteString(")")

	return sql.String(), nil
}

// translateExpression translates a CQL expression to SQL
func (t *Translator) translateExpression(expr parser.Expression) string {
	switch e := expr.(type) {
	case *parser.ColumnExpr:
		if e.Table != "" {
			return t.quoteIdentifier(e.Table) + "." + t.quoteIdentifier(e.Name)
		}
		return t.quoteIdentifier(e.Name)
	case *parser.StarExpr:
		if e.Table != "" {
			return t.quoteIdentifier(e.Table) + ".*"
		}
		return "*"
	case *parser.BindMarker:
		return "?"
	case *parser.NamedBindMarker:
		return "?"
	case *parser.StringLiteral:
		return fmt.Sprintf("'%s'", escapeMySQLStringLiteral(e.Value))
	case *parser.BlobLiteral:
		if len(e.Value) == 0 {
			return "X''"
		}
		return fmt.Sprintf("0x%s", hex.EncodeToString(e.Value))
	case *parser.IntegerLiteral:
		return fmt.Sprintf("%d", e.Value)
	case *parser.FloatLiteral:
		if e.Raw != "" {
			return e.Raw
		}
		return fmt.Sprintf("%f", e.Value)
	case *parser.BooleanLiteral:
		return e.Value
	case *parser.UUIDLiteral:
		return fmt.Sprintf("'%s'", e.Value)
	case *parser.NullLiteral:
		return "NULL"
	case *parser.BinaryExpr:
		return t.translateBinaryExpr(e)
	case *parser.UnaryExpr:
		return fmt.Sprintf("%s%s", e.Op, t.translateExpression(e.Expr))
	case *parser.IsNullExpr:
		if e.Not {
			return fmt.Sprintf("(%s IS NOT NULL)", t.translateExpression(e.Expr))
		}
		return fmt.Sprintf("(%s IS NULL)", t.translateExpression(e.Expr))
	case *parser.AliasedExpr:
		return fmt.Sprintf("%s AS %s", t.translateExpression(e.Expr), t.quoteIdentifier(e.Alias))
	case *parser.CastExpr:
		return fmt.Sprintf("CAST(%s AS %s)", t.translateExpression(e.Expr), t.translateType(e.Type))
	case *parser.TypeHintExpr:
		return t.translateExpression(e.Expr)
	case *parser.FunctionCall:
		return t.translateFunctionCall(e)
	case *parser.ListLiteral:
		return t.translateJSONArray(e.Elements)
	case *parser.SetLiteral:
		return t.translateJSONArray(e.Elements)
	case *parser.TupleLiteral:
		return t.translateJSONArray(e.Elements)
	case *parser.MapLiteral:
		return t.translateJSONObject(e.Entries)
	default:
		return fmt.Sprintf("/* unsupported expression: %T */", expr)
	}
}

func (t *Translator) translateColumnDefaultExpression(expr parser.Expression) (string, error) {
	if expr == nil {
		return "NULL", nil
	}
	if tsExpr, ok := translateCurrentTimestampExpr(expr); ok {
		return tsExpr, nil
	}
	switch e := expr.(type) {
	case *parser.TypeHintExpr:
		return t.translateColumnDefaultExpression(e.Expr)
	case *parser.StringLiteral, *parser.BlobLiteral, *parser.IntegerLiteral, *parser.FloatLiteral,
		*parser.BooleanLiteral, *parser.UUIDLiteral, *parser.NullLiteral, *parser.UnaryExpr, *parser.FunctionCall:
		return t.translateExpression(expr), nil
	default:
		return "", fmt.Errorf("unsupported DEFAULT expression type: %T", expr)
	}
}

func (t *Translator) translateColumnOnUpdateExpression(expr parser.Expression) (string, error) {
	if expr == nil {
		return "", fmt.Errorf("ON UPDATE expression is required")
	}
	if tsExpr, ok := translateCurrentTimestampExpr(expr); ok {
		return tsExpr, nil
	}
	return "", fmt.Errorf("unsupported ON UPDATE expression type: %T", expr)
}

func translateCurrentTimestampExpr(expr parser.Expression) (string, bool) {
	switch e := expr.(type) {
	case *parser.TypeHintExpr:
		return translateCurrentTimestampExpr(e.Expr)
	case *parser.ColumnExpr:
		if e.Table == "" && (strings.EqualFold(e.Name, "current_timestamp") || strings.EqualFold(e.Name, "now")) {
			return "CURRENT_TIMESTAMP(6)", true
		}
	case *parser.FunctionCall:
		if len(e.Args) == 0 && (strings.EqualFold(e.Name, "current_timestamp") || strings.EqualFold(e.Name, "now")) {
			return "CURRENT_TIMESTAMP(6)", true
		}
	}
	return "", false
}

func escapeMySQLStringLiteral(value string) string {
	if value == "" {
		return value
	}
	if !strings.ContainsAny(value, "\\'") {
		return value
	}
	var builder strings.Builder
	builder.Grow(len(value))
	for i := 0; i < len(value); i++ {
		switch value[i] {
		case '\\':
			builder.WriteString("\\\\")
		case '\'':
			builder.WriteString("''")
		default:
			builder.WriteByte(value[i])
		}
	}
	return builder.String()
}

func (t *Translator) translateBinaryExpr(e *parser.BinaryExpr) string {
	op := strings.ToUpper(strings.TrimSpace(e.Op))
	switch op {
	case "IN":
		return t.translateInExpr(e)
	case "BETWEEN":
		return t.translateBetweenExpr(e)
	default:
		return fmt.Sprintf("(%s %s %s)",
			t.translateExpression(e.Left),
			e.Op,
			t.translateExpression(e.Right))
	}
}

func (t *Translator) translateInExpr(e *parser.BinaryExpr) string {
	left := t.translateExpression(e.Left)
	list, ok := e.Right.(*parser.ListLiteral)
	if !ok {
		return fmt.Sprintf("(%s IN %s)", left, t.translateExpression(e.Right))
	}
	if len(list.Elements) == 0 {
		return fmt.Sprintf("(%s IN (NULL))", left)
	}
	items := make([]string, len(list.Elements))
	for i, elem := range list.Elements {
		items[i] = t.translateExpression(elem)
	}
	return fmt.Sprintf("(%s IN (%s))", left, strings.Join(items, ", "))
}

func (t *Translator) translateBetweenExpr(e *parser.BinaryExpr) string {
	left := t.translateExpression(e.Left)
	tuple, ok := e.Right.(*parser.TupleLiteral)
	if !ok || len(tuple.Elements) != 2 {
		return fmt.Sprintf("(%s BETWEEN %s)", left, t.translateExpression(e.Right))
	}
	start := t.translateExpression(tuple.Elements[0])
	end := t.translateExpression(tuple.Elements[1])
	return fmt.Sprintf("(%s BETWEEN %s AND %s)", left, start, end)
}

func (t *Translator) translateJSONArray(elements []parser.Expression) string {
	if len(elements) == 0 {
		return "JSON_ARRAY()"
	}
	items := make([]string, len(elements))
	for i, elem := range elements {
		items[i] = t.translateExpression(elem)
	}
	return fmt.Sprintf("JSON_ARRAY(%s)", strings.Join(items, ", "))
}

func (t *Translator) translateJSONObject(entries []parser.MapEntry) string {
	if len(entries) == 0 {
		return "JSON_OBJECT()"
	}
	items := make([]string, 0, len(entries)*2)
	for _, entry := range entries {
		items = append(items, t.translateExpression(entry.Key))
		items = append(items, t.translateExpression(entry.Value))
	}
	return fmt.Sprintf("JSON_OBJECT(%s)", strings.Join(items, ", "))
}

// translateFunctionCall translates a CQL function call to SQL
func (t *Translator) translateFunctionCall(fc *parser.FunctionCall) string {
	funcName := strings.ToUpper(fc.Name)

	// Handle CQL-specific functions
	switch funcName {
	case "COUNT":
		if len(fc.Args) > 0 {
			if _, ok := fc.Args[0].(*parser.StarExpr); ok {
				return "COUNT(*)"
			}
		}
		return fmt.Sprintf("COUNT(%s)", t.translateExpression(fc.Args[0]))
	case "TOKEN":
		// TOKEN function is used for partition key routing in Cassandra
		// In MySQL, we can just use the column value directly
		if len(fc.Args) > 0 {
			return t.translateExpression(fc.Args[0])
		}
		return "TOKEN()"
	case "NOW":
		return "NOW()"
	case "MIN":
		if len(fc.Args) > 0 {
			return fmt.Sprintf("MIN(%s)", t.translateExpression(fc.Args[0]))
		}
		return "MIN()"
	case "MAX":
		if len(fc.Args) > 0 {
			return fmt.Sprintf("MAX(%s)", t.translateExpression(fc.Args[0]))
		}
		return "MAX()"
	case "SUM":
		if len(fc.Args) > 0 {
			return fmt.Sprintf("SUM(%s)", t.translateExpression(fc.Args[0]))
		}
		return "SUM()"
	case "AVG":
		if len(fc.Args) > 0 {
			return fmt.Sprintf("AVG(%s)", t.translateExpression(fc.Args[0]))
		}
		return "AVG()"
	default:
		args := make([]string, len(fc.Args))
		for i, arg := range fc.Args {
			args[i] = t.translateExpression(arg)
		}
		return fmt.Sprintf("%s(%s)", funcName, strings.Join(args, ", "))
	}
}

// translateType translates a CQL type to a MySQL type
func (t *Translator) translateType(typ parser.Type) string {
	switch tt := typ.(type) {
	case *parser.SimpleType:
		return t.translateSimpleType(tt.Name)
	case *parser.CollectionType:
		return t.translateCollectionType(tt)
	case *parser.TupleType:
		return "JSON" /* Tuple stored as JSON */
	case *parser.VectorType:
		return t.translateVectorType(tt)
	default:
		return "TEXT" /* Default fallback */
	}
}

func requireSchema(table *parser.TableName) error {
	if table == nil {
		return fmt.Errorf("table name is required")
	}
	if table.Schema == "" {
		if table.Name == "" {
			return fmt.Errorf("keyspace is required")
		}
		return fmt.Errorf("keyspace is required for %s", table.Name)
	}
	return nil
}

func validateIdentifier(kind, name string) error {
	if err := keyspace.ValidateIdentifier(name, kind); err != nil {
		return err
	}
	return nil
}

// translateBaseType translates a CQL base type to a MySQL type
func (t *Translator) translateBaseType(name string) string {
	switch strings.ToUpper(name) {
	case "ASCII":
		return "VARCHAR(256)"
	case "BIGINT":
		return "BIGINT"
	case "BLOB":
		return "LONGBLOB"
	case "BOOLEAN":
		return "BOOLEAN"
	case "COUNTER":
		return "BIGINT"
	case "DECIMAL":
		return fmt.Sprintf("DECIMAL(%d,%d)", defaultDecimalPrecision, defaultDecimalScale)
	case "DOUBLE":
		return "DOUBLE"
	case "FLOAT":
		return "FLOAT"
	case "INT":
		return "INT"
	case "TEXT":
		return "TEXT"
	case "TIMESTAMP":
		return "DATETIME(6)" /* Microsecond precision */
	case "UUID":
		return "CHAR(36)"
	case "TIMEUUID":
		return "CHAR(36)"
	case "VARCHAR":
		return "VARCHAR(65535)"
	case "VARINT":
		return "DECIMAL(40,0)"
	case "INET":
		return "VARCHAR(45)" /* IPv6 address */
	case "DATE":
		return "DATE"
	case "TIME":
		return "TIME(6)"
	case "SMALLINT":
		return "SMALLINT"
	case "TINYINT":
		return "TINYINT"
	case "DURATION":
		return "TEXT"
	default:
		return "TEXT"
	}
}

func (t *Translator) translateSimpleType(name string) string {
	upper := strings.ToUpper(strings.TrimSpace(name))
	if strings.HasPrefix(upper, "FROZEN<") {
		return "JSON"
	}
	if isBuiltinCQLType(upper) {
		return t.translateBaseType(upper)
	}
	// Treat user-defined types as JSON storage.
	return "JSON"
}

// translateCollectionType translates a CQL collection type to a MySQL type
func (t *Translator) translateCollectionType(ct *parser.CollectionType) string {
	switch strings.ToLower(ct.Kind) {
	case "list":
		return "JSON" /* Lists stored as JSON array */
	case "map":
		return "JSON" /* Maps stored as JSON object */
	case "set":
		return "JSON" /* Sets stored as JSON array */
	default:
		return "JSON"
	}
}

func (t *Translator) translateVectorType(vt *parser.VectorType) string {
	if vt == nil || vt.ElemType == nil || vt.Size <= 0 {
		return "VARBINARY(1)"
	}
	elemBytes := vectorElementBytes(vt.ElemType)
	if elemBytes == 0 {
		return "VARBINARY(1)"
	}
	return fmt.Sprintf("VARBINARY(%d)", vt.Size*elemBytes)
}

func vectorElementBytes(typ parser.Type) int {
	if typ == nil {
		return 0
	}
	switch t := typ.(type) {
	case *parser.SimpleType:
		switch strings.ToUpper(strings.TrimSpace(t.Name)) {
		case "FLOAT":
			return 4
		case "DOUBLE":
			return 8
		}
	}
	return 0
}

func isBuiltinCQLType(name string) bool {
	switch name {
	case "ASCII", "BIGINT", "BLOB", "BOOLEAN", "COUNTER", "DATE",
		"DECIMAL", "DOUBLE", "DURATION", "FLOAT", "INET", "INT",
		"SMALLINT", "TEXT", "TIME", "TIMESTAMP", "TIMEUUID",
		"TINYINT", "UUID", "VARCHAR", "VARINT":
		return true
	}
	return false
}

// quoteIdentifier quotes an identifier for MySQL
func (t *Translator) quoteIdentifier(name string) string {
	if !t.QuoteIdentifiers {
		return name
	}
	return fmt.Sprintf("`%s`", strings.ReplaceAll(name, "`", "``"))
}

// QuoteIdentifier returns a quoted MySQL identifier.
func (t *Translator) QuoteIdentifier(name string) string {
	return t.quoteIdentifier(name)
}

// TranslateExpression converts a CQL expression into a MySQL expression string.
func (t *Translator) TranslateExpression(expr parser.Expression) string {
	return t.translateExpression(expr)
}

type pkTypeInfo struct {
	prefixable     bool
	requiresPrefix bool
	isBinary       bool
	fixedBytes     int
	maxBytes       int
}

type pkColumnInfo struct {
	name           string
	prefixable     bool
	requiresPrefix bool
	isBinary       bool
	fixedBytes     int
	maxBytes       int
	minBytes       int
}

func (t *Translator) indexPrefixConfig() (int, int, int) {
	maxBytes := t.indexMaxBytes
	minChars := t.minPrefixChars
	minBytes := t.minPrefixBytes
	if maxBytes <= 0 {
		maxBytes = defaultMaxIndexBytes
	}
	if minChars <= 0 {
		minChars = defaultMinPrefixChars
	}
	if minBytes <= 0 {
		minBytes = defaultMinPrefixBytes
	}
	return maxBytes, minChars, minBytes
}

func (t *Translator) buildPrimaryKeyClause(stmt *parser.CreateTableStatement) (string, error) {
	pkCols, _, err := t.primaryKeyColumns(stmt)
	if err != nil {
		return "", err
	}
	if len(pkCols) == 0 {
		return "", nil
	}

	prefixes, err := t.computePrimaryKeyPrefixes(pkCols)
	if err != nil {
		return "", err
	}

	parts := make([]string, 0, len(pkCols))
	for _, col := range pkCols {
		name := t.quoteIdentifier(col.name)
		if prefix, ok := prefixes[col.name]; ok && prefix > 0 {
			name = fmt.Sprintf("%s(%d)", name, prefix)
		}
		parts = append(parts, name)
	}

	suffix := " NONCLUSTERED"
	if t.enableClusteredIndex {
		suffix = " CLUSTERED"
	}
	return "PRIMARY KEY (" + strings.Join(parts, ", ") + ")" + suffix, nil
}

func (t *Translator) primaryKeyColumns(stmt *parser.CreateTableStatement) ([]pkColumnInfo, map[string]*parser.ColumnDefinition, error) {
	if stmt.PrimaryKey == nil {
		return nil, nil, nil
	}

	pkNames := make([]string, 0, len(stmt.PrimaryKey.PartitionKeys)+len(stmt.PrimaryKey.ClusteringColumns))
	pkNames = append(pkNames, stmt.PrimaryKey.PartitionKeys...)
	pkNames = append(pkNames, stmt.PrimaryKey.ClusteringColumns...)
	if len(pkNames) == 0 {
		return nil, nil, nil
	}

	columns := make(map[string]*parser.ColumnDefinition, len(stmt.Columns))
	for _, col := range stmt.Columns {
		columns[col.Name] = col
	}

	pkCols := make([]pkColumnInfo, 0, len(pkNames))
	for _, name := range pkNames {
		col := columns[name]
		if col == nil {
			return nil, nil, fmt.Errorf("primary key column %q is not defined", name)
		}
		switch t := col.Type.(type) {
		case *parser.CollectionType, *parser.TupleType, *parser.VectorType:
			return nil, nil, fmt.Errorf("primary key column %q uses unsupported type %s", name, col.Type.TypeString())
		case *parser.SimpleType:
			upper := strings.ToUpper(strings.TrimSpace(t.Name))
			if strings.HasPrefix(upper, "FROZEN<") || !isBuiltinCQLType(upper) {
				return nil, nil, fmt.Errorf("primary key column %q uses unsupported type %s", name, col.Type.TypeString())
			}
		}
		info := t.classifyPrimaryKeyType(col.Type)
		pkCols = append(pkCols, pkColumnInfo{
			name:           name,
			prefixable:     info.prefixable,
			requiresPrefix: info.requiresPrefix,
			isBinary:       info.isBinary,
			fixedBytes:     info.fixedBytes,
			maxBytes:       info.maxBytes,
		})
	}

	return pkCols, columns, nil
}

func (t *Translator) buildPartitionClause(stmt *parser.CreateTableStatement) (string, error) {
	if stmt.PrimaryKey == nil || len(stmt.PrimaryKey.PartitionKeys) == 0 {
		return "", nil
	}

	pkCols, columns, err := t.primaryKeyColumns(stmt)
	if err != nil {
		return "", err
	}

	for _, name := range stmt.PrimaryKey.PartitionKeys {
		col := columns[name]
		if col == nil {
			return "", fmt.Errorf("partition key column %q is not defined", name)
		}
		typeName, ok := partitionKeyTypeName(col.Type)
		if !ok {
			return "", fmt.Errorf("partition key column %q uses unsupported type %s", name, col.Type.TypeString())
		}
		if isPartitionKeyTypeDisallowed(typeName, t.disableTextBlobPartitionKey) {
			return "", fmt.Errorf("partition key column %q uses disallowed type %s", name, typeName)
		}
	}

	// TEXT/BLOB partition keys are accepted for Cassandra compatibility; skip MySQL partitioning
	// because prefix-key constraints conflict with PARTITION BY rules.
	if !t.disableTextBlobPartitionKey && hasTextBlobPartitionKey(stmt.PrimaryKey.PartitionKeys, columns) {
		return "", nil
	}

	partitionSize := t.partitionDefaultSize
	if partitionSize <= 0 {
		partitionSize = 10
	}

	if len(stmt.PrimaryKey.PartitionKeys) == 1 {
		name := stmt.PrimaryKey.PartitionKeys[0]
		if isHashPartitionableType(columns[name].Type) {
			return fmt.Sprintf("PARTITION BY HASH(%s) PARTITIONS %d", t.quoteIdentifier(name), partitionSize), nil
		}
		// Fall back to KEY partitioning.
		if err := t.ensurePartitionKeyNoPrefix(stmt.PrimaryKey.PartitionKeys, pkCols); err != nil {
			return "", err
		}
		return fmt.Sprintf("PARTITION BY KEY(%s) PARTITIONS %d", t.quoteIdentifier(name), partitionSize), nil
	}

	if err := t.ensurePartitionKeyNoPrefix(stmt.PrimaryKey.PartitionKeys, pkCols); err != nil {
		return "", err
	}
	parts := make([]string, 0, len(stmt.PrimaryKey.PartitionKeys))
	for _, name := range stmt.PrimaryKey.PartitionKeys {
		parts = append(parts, t.quoteIdentifier(name))
	}
	return fmt.Sprintf("PARTITION BY KEY(%s) PARTITIONS %d", strings.Join(parts, ", "), partitionSize), nil
}

func (t *Translator) ensurePartitionKeyNoPrefix(partitionKeys []string, pkCols []pkColumnInfo) error {
	prefixes, err := t.computePrimaryKeyPrefixes(pkCols)
	if err != nil {
		return err
	}
	for _, name := range partitionKeys {
		if prefix, ok := prefixes[name]; ok && prefix > 0 {
			return fmt.Errorf("partition key column %q requires index prefix; partitioning is not supported", name)
		}
	}
	return nil
}

func partitionKeyTypeName(typ parser.Type) (string, bool) {
	if typ == nil {
		return "", false
	}
	switch t := typ.(type) {
	case *parser.SimpleType:
		name := strings.ToUpper(strings.TrimSpace(t.Name))
		if strings.HasPrefix(name, "FROZEN<") {
			return "", false
		}
		return name, true
	default:
		return "", false
	}
}

func isPartitionKeyTypeDisallowed(name string, disableTextBlobPartitionKey bool) bool {
	switch name {
	case "BOOLEAN", "DECIMAL", "FLOAT":
		return true
	case "BLOB", "TEXT":
		return disableTextBlobPartitionKey
	default:
		return false
	}
}

func hasTextBlobPartitionKey(partitionKeys []string, columns map[string]*parser.ColumnDefinition) bool {
	for _, name := range partitionKeys {
		col := columns[name]
		if col == nil {
			continue
		}
		typeName, ok := partitionKeyTypeName(col.Type)
		if !ok {
			continue
		}
		switch typeName {
		case "TEXT", "BLOB":
			return true
		}
	}
	return false
}

func isHashPartitionableType(typ parser.Type) bool {
	name, ok := partitionKeyTypeName(typ)
	if !ok {
		return false
	}
	switch name {
	case "TINYINT", "SMALLINT", "INT", "BIGINT", "COUNTER":
		return true
	default:
		return false
	}
}

func (t *Translator) classifyPrimaryKeyType(typ parser.Type) pkTypeInfo {
	switch tt := typ.(type) {
	case *parser.SimpleType:
		name := strings.ToUpper(tt.Name)
		switch name {
		case "ASCII":
			return pkTypeInfo{prefixable: true, maxBytes: 256 * maxBytesPerChar}
		case "VARCHAR":
			return pkTypeInfo{prefixable: true, maxBytes: 65535 * maxBytesPerChar}
		case "TEXT":
			return pkTypeInfo{prefixable: true, requiresPrefix: true}
		case "BLOB":
			return pkTypeInfo{prefixable: true, requiresPrefix: true, isBinary: true}
		case "UUID", "TIMEUUID":
			return pkTypeInfo{fixedBytes: 36 * maxBytesPerChar}
		case "INET":
			return pkTypeInfo{fixedBytes: 45 * maxBytesPerChar}
		case "BOOLEAN":
			return pkTypeInfo{fixedBytes: 1}
		case "TINYINT":
			return pkTypeInfo{fixedBytes: 1}
		case "SMALLINT":
			return pkTypeInfo{fixedBytes: 2}
		case "INT":
			return pkTypeInfo{fixedBytes: 4}
		case "BIGINT", "COUNTER":
			return pkTypeInfo{fixedBytes: 8}
		case "FLOAT":
			return pkTypeInfo{fixedBytes: 4}
		case "DOUBLE":
			return pkTypeInfo{fixedBytes: 8}
		case "DATE":
			return pkTypeInfo{fixedBytes: 3}
		case "TIME":
			return pkTypeInfo{fixedBytes: timeStorageBytes(3)}
		case "TIMESTAMP":
			return pkTypeInfo{fixedBytes: datetimeStorageBytes(3)}
		case "DECIMAL":
			return pkTypeInfo{fixedBytes: decimalStorageBytes(defaultDecimalPrecision, defaultDecimalScale)}
		case "VARINT":
			return pkTypeInfo{fixedBytes: decimalStorageBytes(40, 0)}
		default:
			return pkTypeInfo{prefixable: true, requiresPrefix: true}
		}
	case *parser.CollectionType, *parser.TupleType:
		return pkTypeInfo{prefixable: true, requiresPrefix: true}
	default:
		return pkTypeInfo{prefixable: true, requiresPrefix: true}
	}
}

func (t *Translator) computePrimaryKeyPrefixes(pkCols []pkColumnInfo) (map[string]int, error) {
	if len(pkCols) == 0 {
		return nil, nil
	}

	maxBytes, minChars, minBytes := t.indexPrefixConfig()
	fixedBytes := 0
	totalFullBytes := 0
	prefixableCols := make([]pkColumnInfo, 0, len(pkCols))
	requiresPrefix := false

	for _, col := range pkCols {
		if !col.prefixable {
			fixedBytes += col.fixedBytes
			continue
		}
		prefixableCols = append(prefixableCols, col)
		if col.requiresPrefix {
			requiresPrefix = true
		}
		if col.maxBytes > 0 {
			totalFullBytes += col.maxBytes
		} else {
			totalFullBytes += maxBytes + 1
		}
	}

	if fixedBytes > maxBytes {
		return nil, fmt.Errorf("primary key fixed length %d exceeds max index length %d", fixedBytes, maxBytes)
	}
	if len(prefixableCols) == 0 {
		return nil, nil
	}
	if !requiresPrefix && fixedBytes+totalFullBytes <= maxBytes {
		return nil, nil
	}

	minTotal := 0
	for i := range prefixableCols {
		min := minBytes
		if !prefixableCols[i].isBinary {
			min = minChars * maxBytesPerChar
		}
		if min <= 0 {
			if prefixableCols[i].isBinary {
				min = defaultMinPrefixBytes
			} else {
				min = defaultMinPrefixChars * maxBytesPerChar
			}
		}
		if prefixableCols[i].maxBytes > 0 && min > prefixableCols[i].maxBytes {
			min = prefixableCols[i].maxBytes
		}
		prefixableCols[i].minBytes = min
		minTotal += min
	}

	if fixedBytes+minTotal > maxBytes {
		return nil, fmt.Errorf("primary key minimum prefix length %d exceeds max index length %d (fixed=%d)", fixedBytes+minTotal, maxBytes, fixedBytes)
	}

	remainingBudget := maxBytes - fixedBytes
	remainingMin := minTotal
	prefixes := make(map[string]int, len(prefixableCols))

	for _, col := range prefixableCols {
		remainingMin -= col.minBytes
		allocBytes := remainingBudget - remainingMin
		if col.maxBytes > 0 && allocBytes > col.maxBytes {
			allocBytes = col.maxBytes
		}
		if allocBytes < col.minBytes {
			allocBytes = col.minBytes
		}

		if col.isBinary {
			if allocBytes <= 0 {
				return nil, fmt.Errorf("primary key prefix length for %s is invalid", col.name)
			}
			prefixes[col.name] = allocBytes
			remainingBudget -= allocBytes
			continue
		}

		prefixChars := allocBytes / maxBytesPerChar
		if prefixChars <= 0 {
			return nil, fmt.Errorf("primary key prefix length for %s is invalid", col.name)
		}
		if col.maxBytes > 0 {
			maxChars := col.maxBytes / maxBytesPerChar
			if prefixChars > maxChars {
				prefixChars = maxChars
			}
		}
		usedBytes := prefixChars * maxBytesPerChar
		prefixes[col.name] = prefixChars
		remainingBudget -= usedBytes
	}

	return prefixes, nil
}

func decimalStorageBytes(precision, scale int) int {
	if precision < 0 {
		precision = 0
	}
	if scale < 0 {
		scale = 0
	}
	if scale > precision {
		scale = precision
	}
	intDigits := precision - scale
	return decimalDigitsBytes(intDigits) + decimalDigitsBytes(scale)
}

func decimalDigitsBytes(digits int) int {
	if digits <= 0 {
		return 0
	}
	groups := digits / 9
	rem := digits % 9
	bytes := groups * 4
	switch {
	case rem == 0:
	case rem <= 2:
		bytes += 1
	case rem <= 4:
		bytes += 2
	case rem <= 6:
		bytes += 3
	default:
		bytes += 4
	}
	return bytes
}

func datetimeStorageBytes(fsp int) int {
	return 5 + fractionalSecondsBytes(fsp)
}

func timeStorageBytes(fsp int) int {
	return 3 + fractionalSecondsBytes(fsp)
}

func fractionalSecondsBytes(fsp int) int {
	if fsp <= 0 {
		return 0
	}
	if fsp <= 2 {
		return 1
	}
	if fsp <= 4 {
		return 2
	}
	return 3
}