package executor

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math"
	"math/big"
	"net"
	"strings"
	"sync"
	"time"

	"go.uber.org/zap"

	"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/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/systemtables"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/translator"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/typemap"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/types"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/virtualtables"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

// Executor executes CQL queries against MySQL backend
type Executor struct {
	poolManager                 *mysqlbackend.PoolManager
	translator                  *translator.Translator
	logger                      *zap.Logger
	prepared                    *preparedStore
	ddl                         *ddlExecutor
	dml                         *dmlExecutor
	selectPlanner               *selectPlanner
	initOnce                    sync.Once
	schemaManager               *mysqlbackend.SchemaManager
	virtualExecutor             *virtualtables.VirtualTableExecutor
	systemExecutor              *systemtables.SystemTableExecutor
	systemState                 *systemtables.SystemTableState
	schemaSync                  *schemaSync
	preparedCache               preparedCacheConfig
	schemaSyncBackground        bool
	cqlMetaPrecreate            bool
	protocolMode                string
	insertDuplicateIncrements   map[string]struct{}
	unloggedBatchParallel       bool
	unloggedMergeBeforeParallel bool
	unloggedBatchWorkers        int
	unloggedBatchMySQLConns     int
	unloggedBatchConnLimiter    *connLimiter
	batchInsertMergeRows        int
	batchInsertMergeBytes       int
	batchGroupByTiDBHashPart    bool
	batchPartitionHintTTL       time.Duration
	batchPartitionHintMu        sync.RWMutex
	batchPartitionHintCache     map[batchTableKey]batchHashPartitionCacheEntry
	backendInfoMu               sync.RWMutex
	backendIsTiDBKnown          bool
	backendIsTiDB               bool
	schemaSyncMu                sync.Mutex
	schemaSyncStop              chan struct{}
	schemaSyncWG                sync.WaitGroup
	credMu                      sync.RWMutex
	credUser                    string
	credPassword                string
	credSet                     bool
	cqlMetaPrecreateDone        int32
	cqlMetaPrecreateIn          int32
	ensureCQLMeta               func(ctx context.Context, user, password string) error
	deadlockRetry               deadlockRetryConfig
	preparedStoreMax            int
	preparedStoreIdleTTL        time.Duration
}

type preparedCacheConfig struct {
	enabled bool
	prewarm bool
}

// PreparedStatement represents a prepared statement
type PreparedStatement struct {
	ID         string
	Query      string
	SQL        string
	ParamCount int
	CreatedAt  time.Time
	IsSelect   bool
	// IsSpecialQuery indicates if this is a system/virtual table query
	// that should not be translated to SQL
	IsSpecialQuery bool
	IsSetKeyspace  bool
	Statement      parser.Statement
	// Parameters contains the bind marker parameter metadata for prepared metadata
	// (used for binding values when executing the prepared statement)
	Parameters []ColumnInfo
	// Columns contains the result column metadata for result metadata
	// (used for describing the columns returned after execution)
	Columns []ColumnInfo
	// MetadataID is the v5+ result metadata identifier.
	MetadataID []byte
	// SelectTransform holds JSON/to_json transforms for SELECT results.
	SelectTransform *selectTransform
	// SelectLimit holds runtime limit evaluation for LIMIT + PER PARTITION LIMIT.
	SelectLimit *selectLimitPlan
	// DMLPlan stores SQL and parameter ordering for DML execution.
	DMLPlan *dmlPlan
	// NeedsParamInference indicates param types must be inferred at execution time.
	NeedsParamInference bool
}

// NewExecutor creates a new query executor
func NewExecutor(poolManager *mysqlbackend.PoolManager, logger *zap.Logger) *Executor {
	cfg := config.GetConfig()
	protocolMode := cfg.Executor.ProtocolModeNormalized()
	schemaManager := mysqlbackend.NewSchemaManager(poolManager, cfg.SchemaCacheTTL, cfg.CQLMetaCacheTTL)
	systemState, err := systemtables.NewSystemTableState(cfg, schemaManager)
	if err != nil {
		logger.Fatal("invalid system table configuration", zap.Error(err))
	}
	systemRegistry := systemtables.NewRegistryWithState(systemState)
	systemExecutor := systemtables.NewSystemTableExecutor(systemRegistry)
	systemExecutor.SetSchemaProvider(systemtables.NewMySQLSchemaProvider(schemaManager))
	exec := &Executor{
		poolManager:     poolManager,
		translator:      translator.NewTranslator(),
		logger:          logger,
		schemaManager:   schemaManager,
		virtualExecutor: virtualtables.NewVirtualTableExecutor(virtualtables.NewRegistry()),
		systemExecutor:  systemExecutor,
		systemState:     systemState,
		preparedCache: preparedCacheConfig{
			enabled: cfg.MySQL.Prepared.Enabled,
			prewarm: cfg.MySQL.Prepared.PrewarmOnPrepare,
		},
		schemaSyncBackground:        cfg.Executor.SchemaSyncBackground,
		cqlMetaPrecreate:            cfg.Executor.CQLMetaPrecreate,
		protocolMode:                protocolMode,
		insertDuplicateIncrements:   parseInsertDuplicateIncrementColumns(cfg.Executor.InsertDuplicateIncrementColumns),
		unloggedBatchParallel:       cfg.Executor.UnloggedBatchParallel,
		unloggedMergeBeforeParallel: cfg.Executor.UnloggedMergeBeforeParallelEnabled(),
		unloggedBatchWorkers:        cfg.Executor.UnloggedBatchParallelism,
		unloggedBatchMySQLConns:     cfg.Executor.UnloggedBatchMaxMySQLConnections,
		unloggedBatchConnLimiter:    newConnLimiter(cfg.Executor.UnloggedBatchMaxMySQLConnections),
		batchInsertMergeRows:        cfg.Executor.BatchInsertMergeMaxRows,
		batchInsertMergeBytes:       cfg.Executor.BatchInsertMergeMaxBytes,
		batchGroupByTiDBHashPart:    cfg.Executor.BatchGroupByTiDBHashPartition,
		batchPartitionHintTTL:       cfg.SchemaCacheTTL,
		batchPartitionHintCache:     make(map[batchTableKey]batchHashPartitionCacheEntry),
		deadlockRetry: newDeadlockRetryConfig(
			cfg.MySQL.Retry.Deadlock.Enabled,
			cfg.MySQL.Retry.Deadlock.MaxAttempts,
			cfg.MySQL.Retry.Deadlock.BaseBackoff,
			cfg.MySQL.Retry.Deadlock.MaxBackoff,
		),
		preparedStoreMax:     cfg.Executor.PreparedCacheMaxStatements,
		preparedStoreIdleTTL: cfg.Executor.PreparedCacheIdleTTL,
	}
	exec.ensureInternals()
	exec.schemaSync = newSchemaSync(cfg.SchemaSync, cfg.SystemTables.SchemaVersion, schemaManager, systemState, exec.prepared, poolManager, logger)
	exec.ensureCQLMeta = schemaManager.EnsureCQLMeta
	return exec
}

// Execute executes a CQL query
func (e *Executor) Execute(ctx context.Context, user, password, cql string) (out *QueryResult, err error) {
	e.ensureInternals()
	ctx = metrics.WithUsername(ctx, user)
	start := time.Now()
	queryType := "unknown"
	defer func() {
		metrics.ObserveCQLQuery(ctx, queryType, err == nil, time.Since(start))
		if err == nil && out != nil {
			metrics.ObserveCQLWriteRows(ctx, queryType, out.CQLWriteRows)
		}
	}()
	// Parse CQL
	stmt, err := parser.NewParser(cql).Parse()
	if err != nil {
		return nil, fmt.Errorf("failed to parse CQL: %w", err)
	}
	queryType = cqlStatementType(stmt)

	e.maybeSyncSchema(ctx, user, password)

	if useStmt, ok := stmt.(*parser.UseStatement); ok {
		if err := e.validateUseKeyspace(ctx, user, password, useStmt.Keyspace); err != nil {
			return nil, err
		}
		if state := keyspaceStateFromContext(ctx); state != nil {
			state.SetKeyspace(useStmt.Keyspace)
		}
		return &QueryResult{SetKeyspace: useStmt.Keyspace}, nil
	}

	if state := keyspaceStateFromContext(ctx); state != nil {
		applyDefaultKeyspace(stmt, state.GetKeyspace())
	}

	if selectStmt, ok := stmt.(*parser.SelectStatement); ok {
		if err := validateOffsetRequiresLimit(selectStmt); err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
	}

	if describeStmt, ok := stmt.(*parser.DescribeStatement); ok {
		e.logger.Debug("Executing DESCRIBE query",
			zap.String("user", user),
			zap.String("cql", cql))
		return e.executeDescribe(ctx, user, password, describeStmt)
	}

	// Check if this is a virtual table query
	selectStmt, isSelect := stmt.(*parser.SelectStatement)
	if isSelect && shouldValidateBlockedSelect(selectStmt) {
		if err := validateBlockedKeyspace(selectStmt.From.Schema); err != nil {
			return nil, err
		}
	}
	if isSelect && virtualtables.IsVirtualTableQuery(selectStmt) {
		e.logger.Debug("Executing virtual table query",
			zap.String("user", user),
			zap.String("cql", cql))
		result, err := e.virtualExecutor.Execute(ctx, selectStmt, user)
		if err != nil {
			return nil, err
		}
		converted := convertQueryResult(result)
		transform := buildSelectTransformFromResult(selectStmt, converted.Columns)
		return applySelectTransform(converted, transform)
	}

	// Check if this is a system table query
	if isSelect && systemtables.IsSystemTableQuery(selectStmt) {
		e.logger.Debug("Executing system table query",
			zap.String("user", user),
			zap.String("cql", cql))
		result, err := e.systemExecutor.Execute(ctx, selectStmt, user, password)
		if err != nil {
			return nil, err
		}
		converted := convertQueryResult(result)
		transform := buildSelectTransformFromResult(selectStmt, converted.Columns)
		return applySelectTransform(converted, transform)
	}

	if err := e.resolveStatementTables(ctx, user, password, stmt); err != nil {
		return nil, err
	}

	if isSelect {
		if err := e.validateDML(ctx, user, password, stmt); err != nil {
			return nil, err
		}

		plan, filterExpr, _, err := e.buildSelectPlanWithTimeFilters(ctx, user, password, selectStmt, nil, false)
		if err != nil {
			return nil, cqlerrors.Invalid(err.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()
		}

		sql := plan.sql
		args := []interface{}(nil)
		var limitValue *int64
		var offsetValue *int64
		var limitSQL *int64
		var offsetSQL *int64
		if plan.limitPlan != nil {
			if plan.limitPlan.hasLimit || plan.limitPlan.hasPartition {
				limit, err := plan.limitPlan.effectiveLimit(nil)
				if err != nil {
					return nil, err
				}
				if plan.limitPlan.applyInMemory {
					limitValue = &limit
				} else {
					limitSQL = &limit
				}
			}
			if plan.limitPlan.hasOffset {
				offset, err := plan.limitPlan.effectiveOffset(nil)
				if err != nil {
					return nil, err
				}
				if plan.limitPlan.applyInMemory {
					offsetValue = &offset
				} else {
					offsetSQL = &offset
				}
			}
			if !plan.limitPlan.applyInMemory && (limitSQL != nil || offsetSQL != nil) {
				sql = appendLimitOffsetSQL(sql, limitSQL, offsetSQL)
			}
		}
		e.logger.Debug("Executing query",
			zap.String("user", user),
			zap.String("cql", cql),
			zap.String("sql", sql))
		result, err := e.executeSelectWithMeta(ctx, conn, sql, args, plan.meta, mysqlOpRead)
		if err != nil {
			return nil, err
		}
		if filterExpr != nil {
			filtered, err := applySelectFilter(result, filterExpr)
			if err != nil {
				return nil, err
			}
			result = filtered
		}
		transformed, err := applySelectTransform(result, plan.transform)
		if err != nil {
			return nil, err
		}
		if plan.limitPlan != nil && plan.limitPlan.applyInMemory {
			if offsetValue != nil {
				applySelectOffset(transformed, *offsetValue)
			}
			if limitValue != nil {
				applySelectLimit(transformed, *limitValue)
			}
		}
		return transformed, nil
	}

	switch s := stmt.(type) {
	case *parser.CreateKeyspaceStatement:
		return e.executeCreateKeyspace(ctx, user, password, s)
	case *parser.DropKeyspaceStatement:
		return e.executeDropKeyspace(ctx, user, password, s)
	case *parser.CreateTableStatement:
		return e.executeCreateTable(ctx, user, password, s)
	case *parser.DropTableStatement:
		return e.executeDropTable(ctx, user, password, s)
	case *parser.TruncateStatement:
		return e.executeTruncate(ctx, user, password, s)
	case *parser.AlterTableStatement:
		return e.executeAlterTable(ctx, user, password, s)
	case *parser.CreateTypeStatement:
		return e.executeCreateType(ctx, user, password, s)
	case *parser.AlterTypeStatement:
		return e.executeAlterType(ctx, user, password, s)
	case *parser.DropTypeStatement:
		return e.executeDropType(ctx, user, password, s)
	case *parser.CreateIndexStatement:
		return e.executeCreateIndex(ctx, user, password, s)
	case *parser.DropIndexStatement:
		return e.executeDropIndex(ctx, user, password, s)
	case *parser.CreateMaterializedViewStatement:
		return e.executeCreateView(ctx, user, password, s)
	case *parser.DropMaterializedViewStatement:
		return e.executeDropView(ctx, user, password, s)
	case *parser.CreateTriggerStatement:
		return e.executeCreateTrigger(ctx, user, password, s)
	case *parser.DropTriggerStatement:
		return e.executeDropTrigger(ctx, user, password, s)
	case *parser.CreateUserStatement:
		return e.ddl.executeCreateUser(ctx, user, password, s)
	case *parser.AlterUserStatement:
		return e.ddl.executeAlterUser(ctx, user, password, s)
	case *parser.DropUserStatement:
		return e.ddl.executeDropUser(ctx, user, password, s)
	case *parser.ListUsersStatement:
		return e.ddl.executeListUsers(ctx, user, password, s)
	case *parser.GrantStatement:
		return e.ddl.executeGrant(ctx, user, password, s)
	case *parser.BatchStatement:
		return e.executeBatchStatement(ctx, user, password, s, nil)
	}

	if err := e.validateDML(ctx, user, password, stmt); err != nil {
		return nil, err
	}

	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()
	}

	// Execute DML statement
	return e.executeDMLStatement(ctx, conn, user, password, stmt, nil, nil)
}

// ExecuteWithParams executes a parsed CQL statement with positional parameters.
func (e *Executor) ExecuteWithParams(ctx context.Context, user, password string, stmt parser.Statement, params []interface{}, paging *PagingOptions) (out *QueryResult, err error) {
	e.ensureInternals()
	ctx = metrics.WithUsername(ctx, user)
	start := time.Now()
	queryType := cqlStatementType(stmt)
	defer func() {
		metrics.ObserveCQLQuery(ctx, queryType, err == nil, time.Since(start))
		if err == nil && out != nil {
			metrics.ObserveCQLWriteRows(ctx, queryType, out.CQLWriteRows)
		}
	}()
	if stmt == nil {
		return nil, fmt.Errorf("statement is required")
	}

	e.maybeSyncSchema(ctx, user, password)

	if useStmt, ok := stmt.(*parser.UseStatement); ok {
		if err := e.validateUseKeyspace(ctx, user, password, useStmt.Keyspace); err != nil {
			return nil, err
		}
		if state := keyspaceStateFromContext(ctx); state != nil {
			state.SetKeyspace(useStmt.Keyspace)
		}
		return &QueryResult{SetKeyspace: useStmt.Keyspace}, nil
	}

	if state := keyspaceStateFromContext(ctx); state != nil {
		applyDefaultKeyspace(stmt, state.GetKeyspace())
	}

	if selectStmt, ok := stmt.(*parser.SelectStatement); ok {
		if err := validateOffsetRequiresLimit(selectStmt); err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
		if shouldValidateBlockedSelect(selectStmt) {
			if err := validateBlockedKeyspace(selectStmt.From.Schema); err != nil {
				return nil, err
			}
		}
	}

	if describeStmt, ok := stmt.(*parser.DescribeStatement); ok {
		e.logger.Debug("Executing DESCRIBE query with params", zap.String("user", user))
		return e.executeDescribe(ctx, user, password, describeStmt)
	}

	expectedParams := countBindMarkersInStatement(stmt)
	if expectedParams != len(params) {
		if expectedParams != 0 || len(params) != 0 {
			return nil, fmt.Errorf("parameter count mismatch: expected %d, got %d", expectedParams, len(params))
		}
	}

	var decodedParams []interface{}
	if len(params) > 0 {
		paramTypes, err := e.inferParamTypes(ctx, user, password, stmt)
		if err != nil {
			return nil, err
		}
		decodedParams, err = e.DecodeParams(ctx, user, password, params, paramTypes)
		if err != nil {
			return nil, err
		}
	}

	if selectStmt, ok := stmt.(*parser.SelectStatement); ok {
		if virtualtables.IsVirtualTableQuery(selectStmt) {
			boundStmt, err := applyBindMarkersToSelect(selectStmt, decodedParams)
			if err != nil {
				return nil, err
			}
			result, err := e.virtualExecutor.Execute(ctx, boundStmt, user)
			if err != nil {
				return nil, err
			}
			converted := convertQueryResult(result)
			transform := buildSelectTransformFromResult(boundStmt, converted.Columns)
			return applySelectTransform(converted, transform)
		}
		if systemtables.IsSystemTableQuery(selectStmt) {
			boundStmt, err := applyBindMarkersToSelect(selectStmt, decodedParams)
			if err != nil {
				return nil, err
			}
			result, err := e.systemExecutor.Execute(ctx, boundStmt, user, password)
			if err != nil {
				return nil, err
			}
			converted := convertQueryResult(result)
			transform := buildSelectTransformFromResult(boundStmt, converted.Columns)
			return applySelectTransform(converted, transform)
		}
	}

	if err := e.resolveStatementTables(ctx, user, password, stmt); err != nil {
		return nil, err
	}

	switch s := stmt.(type) {
	case *parser.CreateKeyspaceStatement:
		return e.executeCreateKeyspace(ctx, user, password, s)
	case *parser.DropKeyspaceStatement:
		return e.executeDropKeyspace(ctx, user, password, s)
	case *parser.CreateTableStatement:
		return e.executeCreateTable(ctx, user, password, s)
	case *parser.DropTableStatement:
		return e.executeDropTable(ctx, user, password, s)
	case *parser.TruncateStatement:
		return e.executeTruncate(ctx, user, password, s)
	case *parser.AlterTableStatement:
		return e.executeAlterTable(ctx, user, password, s)
	case *parser.CreateTypeStatement:
		return e.executeCreateType(ctx, user, password, s)
	case *parser.AlterTypeStatement:
		return e.executeAlterType(ctx, user, password, s)
	case *parser.DropTypeStatement:
		return e.executeDropType(ctx, user, password, s)
	case *parser.CreateIndexStatement:
		return e.executeCreateIndex(ctx, user, password, s)
	case *parser.DropIndexStatement:
		return e.executeDropIndex(ctx, user, password, s)
	case *parser.CreateMaterializedViewStatement:
		return e.executeCreateView(ctx, user, password, s)
	case *parser.DropMaterializedViewStatement:
		return e.executeDropView(ctx, user, password, s)
	case *parser.CreateTriggerStatement:
		return e.executeCreateTrigger(ctx, user, password, s)
	case *parser.DropTriggerStatement:
		return e.executeDropTrigger(ctx, user, password, s)
	case *parser.CreateUserStatement:
		return e.ddl.executeCreateUser(ctx, user, password, s)
	case *parser.AlterUserStatement:
		return e.ddl.executeAlterUser(ctx, user, password, s)
	case *parser.DropUserStatement:
		return e.ddl.executeDropUser(ctx, user, password, s)
	case *parser.ListUsersStatement:
		return e.ddl.executeListUsers(ctx, user, password, s)
	case *parser.GrantStatement:
		return e.ddl.executeGrant(ctx, user, password, s)
	case *parser.BatchStatement:
		return e.executeBatchStatement(ctx, user, password, s, decodedParams)
	case *parser.SelectStatement, *parser.InsertStatement, *parser.UpdateStatement, *parser.DeleteStatement:
		if err := e.validateDML(ctx, user, password, stmt); err != nil {
			return nil, err
		}
	default:
		return nil, fmt.Errorf("unsupported statement type: %T", stmt)
	}

	conn, release, err := e.getConn(ctx, user, password)
	if err != nil {
		return nil, fmt.Errorf("failed to get connection: %w", err)
	}
	if release != nil {
		defer release()
	}

	if selectStmt, ok := stmt.(*parser.SelectStatement); ok {
		if paging != nil && (paging.PageSize > 0 || len(paging.PagingState) > 0) {
			result, err := e.executePagedSelect(ctx, conn, user, password, selectStmt, decodedParams, paging)
			if err == nil {
				return result, nil
			}
			if !isPagingUnsupported(err) {
				return nil, err
			}
		}
		plan, filterExpr, updatedParams, err := e.buildSelectPlanWithTimeFilters(ctx, user, password, selectStmt, decodedParams, true)
		if err != nil {
			return nil, err
		}
		decodedParams = updatedParams
		sql := plan.sql
		args := decodedParams
		var limitValue *int64
		var offsetValue *int64
		var limitSQL *int64
		var offsetSQL *int64
		if plan.limitPlan != nil {
			if plan.limitPlan.hasLimit || plan.limitPlan.hasPartition {
				limit, err := plan.limitPlan.effectiveLimit(decodedParams)
				if err != nil {
					return nil, err
				}
				if plan.limitPlan.applyInMemory {
					limitValue = &limit
				} else {
					limitSQL = &limit
				}
			}
			if plan.limitPlan.hasOffset {
				offset, err := plan.limitPlan.effectiveOffset(decodedParams)
				if err != nil {
					return nil, err
				}
				if plan.limitPlan.applyInMemory {
					offsetValue = &offset
				} else {
					offsetSQL = &offset
				}
			}
			if !plan.limitPlan.applyInMemory && (limitSQL != nil || offsetSQL != nil) {
				sql = appendLimitOffsetSQL(sql, limitSQL, offsetSQL)
			}
			args, err = filterLimitParams(decodedParams, plan.limitPlan.dropIndexes)
			if err != nil {
				return nil, err
			}
		}
		e.logger.Debug("Executing query with params",
			zap.String("user", user),
			zap.Stringer("cql", stmt),
			zap.String("sql", sql))
		result, err := e.executeSelectWithMeta(ctx, conn, sql, args, plan.meta, mysqlOpRead)
		if err != nil {
			return nil, err
		}
		if filterExpr != nil {
			filtered, err := applySelectFilter(result, filterExpr)
			if err != nil {
				return nil, err
			}
			result = filtered
		}
		if plan.transform.json || len(plan.transform.toJSON) > 0 || len(plan.transform.casts) > 0 || len(plan.transform.dropIndexes) > 0 {
			transformed, err := applySelectTransform(result, plan.transform)
			if err != nil {
				return nil, err
			}
			result = transformed
		}
		if plan.limitPlan != nil && plan.limitPlan.applyInMemory {
			if offsetValue != nil {
				applySelectOffset(result, *offsetValue)
			}
			if limitValue != nil {
				applySelectLimit(result, *limitValue)
			}
		}
		return result, nil
	}

	return e.executeDMLStatement(ctx, conn, user, password, stmt, decodedParams, nil)
}

// executeSelect executes a SELECT query
func (e *Executor) executeSelect(ctx context.Context, exec sqlExecutor, sql string, params []interface{}, op mysqlOperation) (*QueryResult, error) {
	return e.executeSelectWithMeta(ctx, exec, sql, params, nil, op)
}

func (e *Executor) executeSelectWithMeta(ctx context.Context, exec sqlExecutor, sql string, params []interface{}, meta *selectMeta, op mysqlOperation) (*QueryResult, error) {
	metrics.RecordRequestSQL(ctx, sql, params...)
	start := time.Now()
	rows, err := exec.QueryContext(ctx, sql, params...)
	if err != nil {
		d := time.Since(start)
		metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
		metrics.ObserveMySQLOperation(ctx, operationLabelForQuery(op), false, d)
		return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to execute query: %w", err), op, sql)
	}
	defer rows.Close()

	// Get column information
	columns, err := rows.ColumnTypes()
	if err != nil {
		return nil, fmt.Errorf("failed to get column types: %w", err)
	}

	colInfos := make([]ColumnInfo, len(columns))
	mysqlTypes := make([]string, len(columns))
	colNames, _ := rows.Columns()
	for i, col := range columns {
		mysqlType := columnTypeName(col)
		mysqlTypes[i] = mysqlType
		colInfos[i] = ColumnInfo{
			Name:     colNames[i],
			Type:     mysqlType,
			Nullable: true,
		}
	}
	applySelectMeta(colInfos, meta)

	// Read all rows
	result := &QueryResult{
		Columns: colInfos,
	}

	for rows.Next() {
		// Create slice to hold values
		values := make([]interface{}, len(columns))
		valuePtrs := make([]interface{}, len(columns))
		for i := range values {
			valuePtrs[i] = &values[i]
		}

		if err := rows.Scan(valuePtrs...); err != nil {
			return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to scan row: %w", err), op, sql)
		}

		// Convert nil to NULL and handle type conversions
		convertedValues := make([]interface{}, len(values))
		for i, v := range values {
			converted, err := e.convertValue(v, mysqlTypes[i], colInfos[i].Type, colInfos[i].TypeInfo)
			if err != nil {
				return nil, fmt.Errorf("failed to convert column %q: %w", colInfos[i].Name, err)
			}
			convertedValues[i] = converted
		}

		result.Rows = append(result.Rows, Row{Values: convertedValues})
	}

	if err := rows.Err(); err != nil {
		d := time.Since(start)
		metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
		metrics.ObserveMySQLOperation(ctx, operationLabelForQuery(op), false, d)
		return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("error iterating rows: %w", err), op, sql)
	}

	d := time.Since(start)
	metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
	metrics.ObserveMySQLOperation(ctx, operationLabelForQuery(op), true, d)
	return result, nil
}

// executeUpdate executes an INSERT, UPDATE, or DELETE query
func (e *Executor) executeUpdate(ctx context.Context, conn *sql.Conn, sql string) (*QueryResult, error) {
	metrics.RecordRequestSQL(ctx, sql)
	start := time.Now()
	result, err := conn.ExecContext(ctx, sql)
	if err != nil {
		d := time.Since(start)
		metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
		metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpExecWrite, false, d)
		return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to execute update: %w", err), mysqlOpWrite, sql)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return nil, fmt.Errorf("failed to get rows affected: %w", err)
	}

	lastInsertID, err := result.LastInsertId()
	if err != nil {
		// LastInsertId may not be supported for all queries
		lastInsertID = 0
	}

	d := time.Since(start)
	metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
	metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpExecWrite, true, d)
	return &QueryResult{
		RowsAffected: rowsAffected,
		LastInsertID: lastInsertID,
	}, nil
}

// executeDDL executes a DDL query (CREATE, DROP, ALTER)
func (e *Executor) executeDDL(ctx context.Context, conn *sql.Conn, sql string) (*QueryResult, error) {
	metrics.RecordRequestSQL(ctx, sql)
	start := time.Now()
	_, err := conn.ExecContext(ctx, sql)
	if err != nil {
		d := time.Since(start)
		metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
		metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpExecWrite, false, d)
		return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to execute DDL: %w", err), mysqlOpWrite, sql)
	}

	d := time.Since(start)
	metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
	metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpExecWrite, true, d)
	return &QueryResult{}, nil
}

// Prepare prepares a CQL statement for execution
func (e *Executor) Prepare(ctx context.Context, user, password, cql string) (*PreparedStatement, error) {
	e.ensureInternals()
	// Parse CQL
	stmt, err := parser.NewParser(cql).Parse()
	if err != nil {
		return nil, fmt.Errorf("failed to parse CQL: %w", err)
	}

	e.maybeSyncSchema(ctx, user, password)

	_, isUseStatement := stmt.(*parser.UseStatement)
	describeStmt, isDescribe := stmt.(*parser.DescribeStatement)

	if state := keyspaceStateFromContext(ctx); state != nil {
		applyDefaultKeyspace(stmt, state.GetKeyspace())
	}

	// Check if this is a system/virtual table query (SELECT only)
	selectStmt, isSelect := stmt.(*parser.SelectStatement)
	if isDescribe {
		var err error
		selectStmt, err = describeToSelect(describeStmt)
		if err != nil {
			return nil, err
		}
		isSelect = true
	}
	if isDescribe {
		if keyspaceName := describeKeyspaceName(describeStmt); keyspaceName != "" {
			if err := validateBlockedKeyspace(keyspaceName); err != nil {
				return nil, err
			}
		}
	}
	if isSelect && shouldValidateBlockedSelect(selectStmt) {
		if err := validateBlockedKeyspace(selectStmt.From.Schema); err != nil {
			return nil, err
		}
	}
	if isSelect && selectStmt != nil {
		if err := validateOffsetRequiresLimit(selectStmt); err != nil {
			return nil, err
		}
	}
	isSpecialQuery := isSelect && (virtualtables.IsVirtualTableQuery(selectStmt) || systemtables.IsSystemTableQuery(selectStmt))

	allowMissingKeyspace := allowMissingKeyspaceFromContext(ctx)
	deferKeyspace := allowMissingKeyspace && statementMissingKeyspace(stmt)

	if !isSpecialQuery {
		if err := e.resolveStatementTables(ctx, user, password, stmt); err != nil {
			return nil, err
		}
	}

	if !isSpecialQuery && !deferKeyspace {
		if err := e.validatePreparedStatement(ctx, user, password, stmt); err != nil {
			return nil, err
		}
	}

	var sql string
	var paramCount int
	var columns []ColumnInfo
	var paramTypes []string
	var selectTransform *selectTransform
	var selectLimit *selectLimitPlan
	var dmlPlan *dmlPlan
	var needsParamInference bool
	var selectMeta *selectMeta
	prepareOp := mysqlOpWrite

	if isUseStatement {
		sql = ""
		paramCount = 0
	} else if isSpecialQuery {
		// For system/virtual table queries, don't translate to SQL
		// Store the original CQL and count parameters directly
		sql = cql
		paramCount = countBindMarkersInSelect(selectStmt)

		// Get column information from the registry
		tableName := selectStmt.From.String()

		// Extract which columns are selected
		selectedCols, isWildcard := extractSelectedColumns(selectStmt)
		aliasMap := extractColumnAliases(selectStmt)

		if virtualtables.IsVirtualTableQuery(selectStmt) {
			// Get from virtual table registry
			if vt, ok := e.virtualExecutor.LookupTable(tableName); ok {
				if isWildcard {
					// SELECT *: include all columns
					columns = make([]ColumnInfo, len(vt.Columns))
					for i, col := range vt.Columns {
						columns[i] = ColumnInfo{
							Keyspace: col.Keyspace,
							Table:    col.Table,
							Name:     col.Name,
							Type:     col.Type,
							Nullable: col.Nullable,
						}
					}
				} else {
					// Filter to only selected columns
					for _, colName := range selectedCols {
						for _, vtCol := range vt.Columns {
							if vtCol.Name == colName {
								columns = append(columns, ColumnInfo{
									Keyspace: vtCol.Keyspace,
									Table:    vtCol.Table,
									Name:     vtCol.Name,
									Type:     vtCol.Type,
									Nullable: vtCol.Nullable,
								})
								break
							}
						}
					}
				}
			}
			if len(aliasMap) > 0 {
				applyAliasMap(columns, aliasMap)
			}
		} else if systemtables.IsSystemTableQuery(selectStmt) {
			// Get from system table registry
			if st, ok := e.systemExecutor.LookupTable(tableName); ok {
				if isWildcard {
					// SELECT *: include all columns
					columns = make([]ColumnInfo, len(st.Columns))
					for i, col := range st.Columns {
						columns[i] = ColumnInfo{
							Keyspace: col.Keyspace,
							Table:    col.Table,
							Name:     col.Name,
							Type:     col.Type,
							Nullable: col.Nullable,
						}
					}
				} else {
					// Filter to only selected columns
					for _, colName := range selectedCols {
						for _, stCol := range st.Columns {
							if stCol.Name == colName {
								columns = append(columns, ColumnInfo{
									Keyspace: stCol.Keyspace,
									Table:    stCol.Table,
									Name:     stCol.Name,
									Type:     stCol.Type,
									Nullable: stCol.Nullable,
								})
								break
							}
						}
					}
				}
			}
			if len(aliasMap) > 0 {
				applyAliasMap(columns, aliasMap)
			}
		}
		if isDescribe {
			switch describeStmt.Target {
			case parser.DescribeTargetKeyspaces, parser.DescribeTargetTables:
				columns = describeResultColumns()
			case parser.DescribeTargetKeyspace, parser.DescribeTargetTable:
				columns = describeTableResultColumns()
			}
		}
		if selectStmt != nil {
			transform := buildSelectTransformFromResult(selectStmt, columns)
			columns = applySelectTransformToColumns(columns, transform)
			if transform.json || len(transform.toJSON) > 0 || len(transform.casts) > 0 || len(transform.dropIndexes) > 0 {
				selectTransform = &transform
			}
		}
	} else {
		if selectStmt != nil {
			prepareOp = mysqlOpRead
		}
		if deferKeyspace {
			sql = ""
		} else {
			// Regular query: translate to SQL
			if selectStmt != nil {
				plan, err := e.buildSelectPlan(ctx, user, password, selectStmt, true, false)
				if err != nil {
					return nil, clientVisibleRequestError(err)
				}
				sql = plan.sql
				selectMeta = plan.meta
				if plan.transform.json || len(plan.transform.toJSON) > 0 || len(plan.transform.casts) > 0 || len(plan.transform.dropIndexes) > 0 {
					selectTransform = &plan.transform
				}
				if plan.limitPlan != nil {
					selectLimit = plan.limitPlan
				}
			} else if _, ok := stmt.(*parser.InsertStatement); ok || isUpdateOrDelete(stmt) {
				if insertStmt, ok := stmt.(*parser.InsertStatement); ok && insertStmt.JSON {
					sql = ""
				} else if updateStmt, ok := stmt.(*parser.UpdateStatement); ok {
					_, hasMutation, err := e.dml.collectAssignmentInfoForUpdate(ctx, user, password, updateStmt)
					if err != nil {
						return nil, clientVisibleRequestError(err)
					}
					if hasMutation {
						sql = ""
					} else {
						plan, err := e.buildDMLPlan(ctx, user, password, stmt)
						if err != nil {
							return nil, clientVisibleRequestError(err)
						}
						sql = plan.sql
						dmlPlan = plan
					}
				} else {
					plan, err := e.buildDMLPlan(ctx, user, password, stmt)
					if err != nil {
						return nil, clientVisibleRequestError(err)
					}
					sql = plan.sql
					dmlPlan = plan
				}
			} else if _, ok := stmt.(*parser.BatchStatement); ok {
				sql = ""
			} else if _, ok := stmt.(*parser.TruncateStatement); ok {
				// TRUNCATE is executed as DDL; skip MySQL PREPARE validation.
				sql = ""
			} else {
				sql, err = e.translator.Translate(stmt)
				if err != nil {
					return nil, clientVisibleRequestError(err)
				}
			}

			conn, release, err := e.getConn(ctx, user, password)
			if err != nil {
				return nil, fmt.Errorf("failed to get connection: %w", err)
			}
			if release != nil {
				defer release()
			}

			if sql != "" {
				// Prepare statement on MySQL to validate
				prepStmt, err := conn.PrepareContext(ctx, sql)
				if err != nil {
					return nil, e.mapMySQLErrorWithSQL(ctx, err, prepareOp, sql)
				}
				defer prepStmt.Close()
			}

			// For SELECT queries, we need to execute to get column metadata
			// For non-SELECT queries, we can't get column info, so leave empty
			if selectStmt != nil {
				// Execute a zero-row wrapper to get metadata only. Keep bind
				// markers bound so TiDB receives this through prepared execution,
				// not as a text query containing raw '?' markers.
				metaSQL, metaArgs := selectMetadataQuery(sql)

				rows, err := conn.QueryContext(ctx, metaSQL, metaArgs...)
				if err != nil {
					// If query fails, just leave columns empty
					e.logger.Warn("failed to get column metadata", zap.Error(err))
				} else {
					defer rows.Close()

					// Get column information
					colTypes, err := rows.ColumnTypes()
					if err != nil {
						e.logger.Warn("failed to get column types", zap.Error(err))
					} else {
						colNames, _ := rows.Columns()
						// Build column info
						columns = make([]ColumnInfo, len(colTypes))
						for i, colType := range colTypes {
							colName := ""
							if i < len(colNames) {
								colName = colNames[i]
							}
							columns[i] = ColumnInfo{
								Name:     colName,
								Type:     columnTypeName(colType),
								Nullable: true, // Default to true
							}
						}
						applySelectMeta(columns, selectMeta)
					}
				}
			}
			if selectStmt != nil && len(columns) == 0 {
				fallback, ok, err := e.fallbackSelectColumns(ctx, user, password, selectStmt)
				if err != nil {
					e.logger.Warn("failed to build select metadata without query", zap.Error(err))
				} else if ok {
					columns = fallback
					applySelectMeta(columns, selectMeta)
				}
			}
			if selectTransform != nil {
				columns = applySelectTransformToColumns(columns, *selectTransform)
			}
			if dmlPlan != nil && dmlPlan.condition.kind != dmlConditionNone {
				columns = conditionalResultColumns(dmlPlan.condition)
			}
			if insertStmt, ok := stmt.(*parser.InsertStatement); ok && insertStmt.JSON && insertStmt.IfNotExists {
				allColumns, err := e.schemaManager.ColumnsForTable(ctx, user, password, insertStmt.Table.Schema, insertStmt.Table.Name)
				if err != nil {
					return nil, err
				}
				fetchCols, err := e.columnInfosForTableColumns(ctx, user, password, insertStmt.Table, columnNames(allColumns))
				if err != nil {
					return nil, err
				}
				columns = conditionalResultColumns(dmlConditionPlan{
					kind:         dmlConditionInsertIfNotExists,
					fetchColumns: fetchCols,
				})
			}
		}

		// Count parameters (simple heuristic - count ? placeholders)
		paramCount = countBindMarkersInStatement(stmt)
		if paramCount > 0 {
			paramTypes = make([]string, paramCount)
			for i := range paramTypes {
				paramTypes[i] = "text"
			}
			if !isSpecialQuery && !deferKeyspace {
				inferred, err := e.inferParamTypes(ctx, user, password, stmt)
				if err != nil {
					e.logger.Warn("failed to infer parameter types", zap.Error(err))
				} else if len(inferred) == paramCount {
					paramTypes = inferred
				}
			} else if deferKeyspace {
				needsParamInference = true
			}
		}
	}

	// Generate prepared statement ID (keyspace-aware)
	stmtKeyspace := statementKeyspace(stmt)
	stmtID := generateStatementID(stmtKeyspace, cql)

	// Build parameter metadata for prepared metadata
	// The prepared metadata describes bind markers (parameters) that need values
	// For now, we use generic "unknown" type for parameters since CQL doesn't
	// always specify parameter types in the query itself
	var parameters []ColumnInfo
	if paramCount > 0 {
		paramKeyspace := statementKeyspace(stmt)
		paramNames, hasNamedMarkers := PreparedBindMarkerNames(stmt)
		parameters = make([]ColumnInfo, paramCount)
		for i := 0; i < paramCount; i++ {
			paramType := "text"
			if i < len(paramTypes) && paramTypes[i] != "" {
				paramType = paramTypes[i]
			}
			typeInfo, err := e.resolveTypeInfo(ctx, user, password, paramType, paramKeyspace)
			if err != nil {
				e.logger.Warn("failed to resolve parameter type info", zap.Error(err))
			}
			parameters[i] = ColumnInfo{
				Keyspace: "", // Empty for parameters in prepared metadata
				Table:    "", // Empty for parameters in prepared metadata
				Name:     fmt.Sprintf("param_%d", i+1),
				Type:     paramType,
				TypeInfo: typeInfo,
				Nullable: true,
			}
			if hasNamedMarkers && i < len(paramNames) {
				parameters[i].Name = paramNames[i]
			}
		}
	}

	metadataID := computeResultMetadataID(columns)
	prepared := &PreparedStatement{
		ID:                  stmtID,
		Query:               cql,
		SQL:                 sql,
		ParamCount:          paramCount,
		CreatedAt:           time.Now(),
		IsSelect:            isSelect,
		IsSpecialQuery:      isSpecialQuery,
		IsSetKeyspace:       isUseStatement,
		Statement:           stmt,
		Parameters:          parameters,
		Columns:             columns,
		MetadataID:          metadataID,
		SelectTransform:     selectTransform,
		SelectLimit:         selectLimit,
		DMLPlan:             dmlPlan,
		NeedsParamInference: needsParamInference,
	}

	existing, stored := e.prepared.putIfAbsent(prepared)
	if !stored {
		return existing, nil
	}

	e.maybePrewarmPrepared(ctx, user, password, prepared)

	e.logger.Debug("Prepared statement",
		zap.String("id", stmtID),
		zap.String("cql", cql),
		zap.String("sql", sql),
		zap.Int("param_count", paramCount),
		zap.Bool("is_special_query", isSpecialQuery),
		zap.Int("column_count", len(columns)))

	return prepared, nil
}

func shouldValidateBlockedSelect(stmt *parser.SelectStatement) bool {
	if stmt == nil || stmt.From == nil || stmt.From.Schema == "" {
		return false
	}
	if systemtables.IsSystemTableQuery(stmt) && strings.EqualFold(stmt.From.Schema, "system_auth") {
		return false
	}
	return true
}

// ExecutePrepared executes a prepared statement with given parameters.
func (e *Executor) ExecutePrepared(ctx context.Context, user, password, stmtID string, params []interface{}) (*QueryResult, error) {
	return e.executePrepared(ctx, user, password, stmtID, params, nil)
}

// ExecutePreparedWithPaging executes a prepared statement with paging options.
func (e *Executor) ExecutePreparedWithPaging(ctx context.Context, user, password, stmtID string, params []interface{}, paging *PagingOptions) (*QueryResult, error) {
	return e.executePrepared(ctx, user, password, stmtID, params, paging)
}

func (e *Executor) executePrepared(ctx context.Context, user, password, stmtID string, params []interface{}, paging *PagingOptions) (out *QueryResult, err error) {
	e.ensureInternals()
	ctx = metrics.WithUsername(ctx, user)
	start := time.Now()
	queryType := "unknown"
	defer func() {
		metrics.ObserveCQLQuery(ctx, queryType, err == nil, time.Since(start))
		if err == nil && out != nil {
			metrics.ObserveCQLWriteRows(ctx, queryType, out.CQLWriteRows)
		}
	}()
	prepared, ok := e.prepared.get(stmtID)
	if !ok {
		return nil, cqlerrors.Unprepared(fmt.Sprintf("Prepared statement not found: %s", stmtID))
	}

	stmt := prepared.Statement
	if stmt == nil {
		parsed, err := parser.NewParser(prepared.Query).Parse()
		if err != nil {
			return nil, fmt.Errorf("failed to parse CQL: %w", err)
		}
		stmt = parsed
	}
	queryType = cqlStatementType(stmt)

	e.maybeSyncSchema(ctx, user, password)

	if useStmt, ok := stmt.(*parser.UseStatement); ok {
		if err := e.validateUseKeyspace(ctx, user, password, useStmt.Keyspace); err != nil {
			return nil, err
		}
		if state := keyspaceStateFromContext(ctx); state != nil {
			state.SetKeyspace(useStmt.Keyspace)
		}
		return &QueryResult{SetKeyspace: useStmt.Keyspace}, nil
	}

	if state := keyspaceStateFromContext(ctx); state != nil {
		applyDefaultKeyspace(stmt, state.GetKeyspace())
	}

	if selectStmt, ok := stmt.(*parser.SelectStatement); ok {
		if shouldValidateBlockedSelect(selectStmt) {
			if err := validateBlockedKeyspace(selectStmt.From.Schema); err != nil {
				return nil, err
			}
		}
	}
	if describeStmt, ok := stmt.(*parser.DescribeStatement); ok {
		if keyspaceName := describeKeyspaceName(describeStmt); keyspaceName != "" {
			if err := validateBlockedKeyspace(keyspaceName); err != nil {
				return nil, err
			}
		}
	}

	if prepared.ParamCount > 0 && len(params) != prepared.ParamCount {
		return nil, fmt.Errorf("prepared statement parameter count mismatch: expected %d, got %d", prepared.ParamCount, len(params))
	}

	var releaseConn func()
	if prepared.ParamCount > 0 && prepared.NeedsParamInference {
		var err error
		ctx, releaseConn, err = e.ensureConnForInference(ctx, user, password)
		if err != nil {
			return nil, err
		}
		if releaseConn != nil {
			defer releaseConn()
		}
	}

	var decodedParams []interface{}
	if prepared.ParamCount > 0 {
		if prepared.NeedsParamInference {
			inferred, err := e.inferParamTypes(ctx, user, password, stmt)
			if err != nil {
				return nil, err
			}
			decodedParams, err = e.DecodeParams(ctx, user, password, params, inferred)
		} else {
			decodedParams, err = e.decodePreparedParams(prepared, params)
		}
		if err != nil {
			return nil, err
		}
	}

	// Handle system/virtual table queries
	if prepared.IsSpecialQuery {
		switch s := stmt.(type) {
		case *parser.DescribeStatement:
			return e.executeDescribe(ctx, user, password, s)
		case *parser.SelectStatement:
			boundStmt, err := applyBindMarkersToSelect(s, decodedParams)
			if err != nil {
				return nil, err
			}
			if virtualtables.IsVirtualTableQuery(boundStmt) {
				e.logger.Debug("Executing prepared virtual table query",
					zap.String("user", user),
					zap.String("stmtID", stmtID))
				result, err := e.virtualExecutor.Execute(ctx, boundStmt, user)
				if err != nil {
					return nil, err
				}
				converted := convertQueryResult(result)
				if prepared.SelectTransform != nil {
					return applySelectTransform(converted, *prepared.SelectTransform)
				}
				return converted, nil
			}
			if systemtables.IsSystemTableQuery(boundStmt) {
				e.logger.Debug("Executing prepared system table query",
					zap.String("user", user),
					zap.String("stmtID", stmtID))
				result, err := e.systemExecutor.Execute(ctx, boundStmt, user, password)
				if err != nil {
					return nil, err
				}
				converted := convertQueryResult(result)
				if prepared.SelectTransform != nil {
					return applySelectTransform(converted, *prepared.SelectTransform)
				}
				return converted, nil
			}
			return nil, fmt.Errorf("special query type not recognized")
		default:
			return nil, fmt.Errorf("special query must be a SELECT or DESCRIBE statement")
		}
	} else if selectStmt, ok := stmt.(*parser.SelectStatement); ok {
		boundStmt, err := applyBindMarkersToSelect(selectStmt, decodedParams)
		if err != nil {
			return nil, err
		}
		if virtualtables.IsVirtualTableQuery(boundStmt) {
			e.logger.Debug("Executing prepared virtual table query",
				zap.String("user", user),
				zap.String("stmtID", stmtID))
			result, err := e.virtualExecutor.Execute(ctx, boundStmt, user)
			if err != nil {
				return nil, err
			}
			converted := convertQueryResult(result)
			if prepared.SelectTransform != nil {
				return applySelectTransform(converted, *prepared.SelectTransform)
			}
			return converted, nil
		}
		if systemtables.IsSystemTableQuery(boundStmt) {
			e.logger.Debug("Executing prepared system table query",
				zap.String("user", user),
				zap.String("stmtID", stmtID))
			result, err := e.systemExecutor.Execute(ctx, boundStmt, user, password)
			if err != nil {
				return nil, err
			}
			converted := convertQueryResult(result)
			if prepared.SelectTransform != nil {
				return applySelectTransform(converted, *prepared.SelectTransform)
			}
			return converted, nil
		}
	}

	switch s := stmt.(type) {
	case *parser.CreateKeyspaceStatement:
		return e.executeCreateKeyspace(ctx, user, password, s)
	case *parser.DropKeyspaceStatement:
		return e.executeDropKeyspace(ctx, user, password, s)
	case *parser.CreateTableStatement:
		return e.executeCreateTable(ctx, user, password, s)
	case *parser.DropTableStatement:
		return e.executeDropTable(ctx, user, password, s)
	case *parser.TruncateStatement:
		return e.executeTruncate(ctx, user, password, s)
	case *parser.AlterTableStatement:
		return e.executeAlterTable(ctx, user, password, s)
	case *parser.CreateTypeStatement:
		return e.executeCreateType(ctx, user, password, s)
	case *parser.AlterTypeStatement:
		return e.executeAlterType(ctx, user, password, s)
	case *parser.DropTypeStatement:
		return e.executeDropType(ctx, user, password, s)
	case *parser.CreateIndexStatement:
		return e.executeCreateIndex(ctx, user, password, s)
	case *parser.DropIndexStatement:
		return e.executeDropIndex(ctx, user, password, s)
	case *parser.CreateMaterializedViewStatement:
		return e.executeCreateView(ctx, user, password, s)
	case *parser.DropMaterializedViewStatement:
		return e.executeDropView(ctx, user, password, s)
	case *parser.CreateTriggerStatement:
		return e.executeCreateTrigger(ctx, user, password, s)
	case *parser.DropTriggerStatement:
		return e.executeDropTrigger(ctx, user, password, s)
	case *parser.CreateUserStatement:
		return e.ddl.executeCreateUser(ctx, user, password, s)
	case *parser.AlterUserStatement:
		return e.ddl.executeAlterUser(ctx, user, password, s)
	case *parser.DropUserStatement:
		return e.ddl.executeDropUser(ctx, user, password, s)
	case *parser.ListUsersStatement:
		return e.ddl.executeListUsers(ctx, user, password, s)
	case *parser.GrantStatement:
		return e.ddl.executeGrant(ctx, user, password, s)
	case *parser.BatchStatement:
		// handled after parameter decoding
	case *parser.SelectStatement, *parser.InsertStatement, *parser.UpdateStatement, *parser.DeleteStatement:
		if err := e.validateDML(ctx, user, password, stmt); err != nil {
			return nil, err
		}
	default:
		return nil, fmt.Errorf("unsupported prepared statement type: %T", stmt)
	}

	if batchStmt, ok := stmt.(*parser.BatchStatement); ok {
		return e.executeBatchStatement(ctx, user, password, batchStmt, decodedParams)
	}

	useCache := e.preparedCache.enabled && e.poolManager != nil && e.preparedCacheEligible(prepared, stmt, paging)
	var exec sqlExecutor
	var conn *sql.Conn
	var release func()
	if useCache {
		pool, err := e.poolManager.GetPool(user, password)
		if err != nil {
			useCache = false
		} else {
			exec = newCachedStmtExecutor(pool, e.logger)
		}
	}
	if !useCache {
		var err error
		conn, release, err = e.getConn(ctx, user, password)
		if err != nil {
			return nil, fmt.Errorf("failed to get connection: %w", err)
		}
		if release != nil {
			defer release()
			ctx = mysqlbackend.WithConn(ctx, conn)
		} else if _, ok := mysqlbackend.ConnFromContext(ctx); !ok {
			ctx = mysqlbackend.WithConn(ctx, conn)
		}
		exec = conn
	}

	if prepared.IsSelect {
		return e.executePreparedSelect(ctx, exec, user, password, prepared, stmt, decodedParams, paging)
	}
	return e.executeDMLStatement(ctx, exec, user, password, stmt, decodedParams, prepared.DMLPlan)
}

func (e *Executor) executePreparedSelect(ctx context.Context, exec sqlExecutor, user, password string, prepared *PreparedStatement, stmt parser.Statement, decodedParams []interface{}, paging *PagingOptions) (*QueryResult, error) {
	if prepared == nil {
		return nil, fmt.Errorf("prepared select statement is required")
	}
	if paging != nil && (paging.PageSize > 0 || len(paging.PagingState) > 0) {
		if paging.Query == "" {
			paging.Query = prepared.Query
		}
		selectStmt, ok := stmt.(*parser.SelectStatement)
		if !ok {
			return nil, fmt.Errorf("prepared select statement is required")
		}
		result, err := e.executePagedSelect(ctx, exec, user, password, selectStmt, decodedParams, paging)
		if err == nil {
			return result, nil
		}
		if !isPagingUnsupported(err) {
			return nil, err
		}
	}
	var plan *selectPlan
	var filterExpr parser.Expression
	var updatedParams []interface{}
	var err error
	selectStmt, ok := stmt.(*parser.SelectStatement)
	if !ok {
		return nil, fmt.Errorf("prepared select statement is required")
	}
	if exprHasTimeFunction(selectStmt.Where) {
		plan, filterExpr, updatedParams, err = e.buildSelectPlanWithTimeFilters(ctx, user, password, selectStmt, decodedParams, true)
		if err != nil {
			return nil, err
		}
		decodedParams = updatedParams
	}

	sql := prepared.SQL
	limitPlan := prepared.SelectLimit
	transform := prepared.SelectTransform
	meta := selectMetaFromColumns(prepared.Columns)
	if plan != nil {
		sql = plan.sql
		if plan.transform.json || len(plan.transform.toJSON) > 0 || len(plan.transform.casts) > 0 || len(plan.transform.dropIndexes) > 0 {
			transform = &plan.transform
		} else {
			transform = nil
		}
		limitPlan = plan.limitPlan
		meta = plan.meta
	} else if sql == "" {
		plan, err = e.buildSelectPlan(ctx, user, password, selectStmt, true, false)
		if err != nil {
			return nil, err
		}
		sql = plan.sql
		if plan.transform.json || len(plan.transform.toJSON) > 0 || len(plan.transform.casts) > 0 || len(plan.transform.dropIndexes) > 0 {
			transform = &plan.transform
		} else {
			transform = nil
		}
		limitPlan = plan.limitPlan
		meta = plan.meta
	}

	args := decodedParams
	var limitValue *int64
	var offsetValue *int64
	var limitSQL *int64
	var offsetSQL *int64
	if limitPlan != nil {
		if limitPlan.hasLimit || limitPlan.hasPartition {
			limit, err := limitPlan.effectiveLimit(decodedParams)
			if err != nil {
				return nil, err
			}
			if limitPlan.applyInMemory {
				limitValue = &limit
			} else {
				limitSQL = &limit
			}
		}
		if limitPlan.hasOffset {
			offset, err := limitPlan.effectiveOffset(decodedParams)
			if err != nil {
				return nil, err
			}
			if limitPlan.applyInMemory {
				offsetValue = &offset
			} else {
				offsetSQL = &offset
			}
		}
		if !limitPlan.applyInMemory && (limitSQL != nil || offsetSQL != nil) {
			sql = appendLimitOffsetSQL(sql, limitSQL, offsetSQL)
		}
		args, err = filterLimitParams(decodedParams, limitPlan.dropIndexes)
		if err != nil {
			return nil, err
		}
	}
	e.logger.Debug("Executing prepared query",
		zap.String("user", user),
		zap.String("stmtID", prepared.ID),
		zap.String("cql", prepared.Query),
		zap.String("sql", sql))
	result, err := e.executeSelectWithMeta(ctx, exec, sql, args, meta, mysqlOpRead)
	if err != nil {
		return nil, err
	}
	if filterExpr != nil {
		filtered, err := applySelectFilter(result, filterExpr)
		if err != nil {
			return nil, err
		}
		result = filtered
	}
	if transform != nil {
		transformed, err := applySelectTransform(result, *transform)
		if err != nil {
			return nil, err
		}
		result = transformed
	}
	if limitPlan != nil && limitPlan.applyInMemory {
		if offsetValue != nil {
			applySelectOffset(result, *offsetValue)
		}
		if limitValue != nil {
			applySelectLimit(result, *limitValue)
		}
	}
	return result, nil
}

// convertValue converts a MySQL value to CQL-compatible format
func (e *Executor) convertValue(value interface{}, mysqlType, cqlType string, typeInfo types.TypeInfo) (interface{}, error) {
	if value == nil {
		return nil, nil
	}

	normalized := normalizeCQLTypeString(cqlType)
	if shouldJSONEncode(normalized, typeInfo) {
		switch v := value.(type) {
		case []byte:
			return decodeJSONValue(v)
		case string:
			return decodeJSONValue([]byte(v))
		default:
			return value, nil
		}
	}

	switch v := value.(type) {
	case []byte:
		if mysqlType == "BLOB" || mysqlType == "LONGBLOB" || mysqlType == "MEDIUMBLOB" || mysqlType == "TINYBLOB" {
			if v == nil {
				return []byte{}, nil
			}
			return v, nil
		}
		if mysqlType == "BINARY" || mysqlType == "VARBINARY" {
			if v == nil {
				return []byte{}, nil
			}
			return v, nil
		}
		return string(v), nil
	default:
		return v, nil
	}
}

func decodeJSONValue(raw []byte) (interface{}, error) {
	if len(raw) == 0 {
		return nil, nil
	}
	dec := json.NewDecoder(bytes.NewReader(raw))
	dec.UseNumber()
	var out interface{}
	if err := dec.Decode(&out); err != nil {
		return nil, err
	}
	return out, nil
}

func columnTypeName(col *sql.ColumnType) string {
	if col == nil {
		return ""
	}
	mysqlType := col.DatabaseTypeName()
	if mysqlType == "" {
		return ""
	}
	if strings.EqualFold(mysqlType, "decimal") || strings.EqualFold(mysqlType, "numeric") {
		precision, scale, ok := col.DecimalSize()
		if ok && precision == 40 && scale == 0 {
			return "varint"
		}
	}
	return mysqlType
}

// generateStatementID generates a unique ID for a prepared statement
func generateStatementID(keyspace, query string) string {
	hasher := sha256.New()
	if keyspace != "" {
		_, _ = hasher.Write([]byte(keyspace))
	}
	_, _ = hasher.Write([]byte{0})
	_, _ = hasher.Write([]byte(query))
	sum := hasher.Sum(nil)
	return hex.EncodeToString(sum)
}

// countParameters counts the number of parameter placeholders in SQL
func countParameters(sql string) int {
	count := 0
	for _, c := range sql {
		if c == '?' {
			count++
		}
	}
	return count
}

// GetPreparedStatement retrieves a prepared statement by ID
func (e *Executor) GetPreparedStatement(id string) (*PreparedStatement, bool) {
	e.ensureInternals()
	return e.prepared.get(id)
}

// RemovePreparedStatement removes a prepared statement
func (e *Executor) RemovePreparedStatement(id string) {
	e.ensureInternals()
	e.prepared.remove(id)
}

// Cleanup removes prepared statements older than the given duration
func (e *Executor) Cleanup(olderThan time.Duration) {
	e.ensureInternals()
	e.prepared.cleanup(olderThan)
}

// extractSelectedColumns extracts the column names from a SELECT statement's Columns field
// Handles cases like: SELECT col1, col2 FROM table, SELECT * FROM table, SELECT table.col FROM table
func extractSelectedColumns(selectStmt *parser.SelectStatement) ([]string, bool) {
	// Check for SELECT * (wildcard)
	if len(selectStmt.Columns) == 1 {
		if _, ok := selectStmt.Columns[0].(*parser.StarExpr); ok {
			return nil, true // Return nil to indicate all columns, true for wildcard
		}
	}

	// Extract column names from the SELECT clause
	var columns []string
	for _, col := range selectStmt.Columns {
		switch c := col.(type) {
		case *parser.ColumnExpr:
			// Use just the column name (ignore table alias if present)
			columns = append(columns, c.Name)
		case *parser.AliasedExpr:
			if colExpr, ok := c.Expr.(*parser.ColumnExpr); ok {
				columns = append(columns, colExpr.Name)
			} else {
				columns = append(columns, c.Expr.String())
			}
		case *parser.StarExpr:
			// If any column is a star, return nil to indicate all columns
			return nil, true
		case *parser.FunctionCall:
			// For functions like count(), sum(), etc., we need to handle them
			// For now, use the function name as the column name
			columns = append(columns, c.Name)
		default:
			// For other expressions, use the string representation
			columns = append(columns, col.String())
		}
	}
	return columns, false
}

func selectProjectionIsSimple(selectStmt *parser.SelectStatement) bool {
	if selectStmt == nil {
		return false
	}
	for _, col := range selectStmt.Columns {
		switch c := col.(type) {
		case *parser.ColumnExpr:
			continue
		case *parser.AliasedExpr:
			if _, ok := c.Expr.(*parser.ColumnExpr); ok {
				continue
			}
			return false
		case *parser.StarExpr:
			continue
		default:
			return false
		}
	}
	return true
}

func (e *Executor) fallbackSelectColumns(ctx context.Context, user, password string, selectStmt *parser.SelectStatement) ([]ColumnInfo, bool, error) {
	if selectStmt == nil || selectStmt.From == nil || selectStmt.From.Schema == "" {
		return nil, false, nil
	}
	if e.schemaManager == nil || !selectProjectionIsSimple(selectStmt) {
		return nil, false, nil
	}

	selectedCols, isWildcard := extractSelectedColumns(selectStmt)
	aliasMap := extractColumnAliases(selectStmt)
	if isWildcard {
		allColumns, err := e.schemaManager.ColumnsForTable(ctx, user, password, selectStmt.From.Schema, selectStmt.From.Name)
		if err != nil {
			return nil, false, err
		}
		selectedCols = columnNames(allColumns)
	}
	if len(selectedCols) == 0 {
		return nil, false, nil
	}

	cols, err := e.columnInfosForTableColumns(ctx, user, password, selectStmt.From, selectedCols)
	if err != nil {
		return nil, false, err
	}
	if len(aliasMap) > 0 {
		applyAliasMap(cols, aliasMap)
	}
	return cols, true, nil
}

func extractColumnAliases(selectStmt *parser.SelectStatement) map[string]string {
	aliases := make(map[string]string)
	for _, col := range selectStmt.Columns {
		aliasExpr, ok := col.(*parser.AliasedExpr)
		if !ok {
			continue
		}
		if base, ok := aliasExpr.Expr.(*parser.ColumnExpr); ok {
			aliases[base.Name] = aliasExpr.Alias
		}
	}
	return aliases
}

func applyAliasMap(columns []ColumnInfo, aliases map[string]string) {
	for i := range columns {
		if alias, ok := aliases[columns[i].Name]; ok {
			columns[i].Name = alias
		}
	}
}

func countBindMarkersInSelect(selectStmt *parser.SelectStatement) int {
	if selectStmt == nil {
		return 0
	}
	count := 0
	for _, col := range selectStmt.Columns {
		count += countBindMarkersInExpr(col)
	}
	for _, expr := range selectStmt.GroupBy {
		count += countBindMarkersInExpr(expr)
	}
	count += countBindMarkersInExpr(selectStmt.Where)
	count += countBindMarkersInExpr(selectStmt.Having)
	count += countBindMarkersInExpr(selectStmt.Options.Limit)
	count += countBindMarkersInExpr(selectStmt.Options.PerPartitionLimit)
	count += countBindMarkersInExpr(selectStmt.Options.Offset)
	return count
}

func countBindMarkersInExpr(expr parser.Expression) int {
	if expr == nil {
		return 0
	}
	switch e := expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		return 1
	case *parser.TypeHintExpr:
		return countBindMarkersInExpr(e.Expr)
	case *parser.IsNullExpr:
		return countBindMarkersInExpr(e.Expr)
	case *parser.AliasedExpr:
		return countBindMarkersInExpr(e.Expr)
	case *parser.BinaryExpr:
		return countBindMarkersInExpr(e.Left) + countBindMarkersInExpr(e.Right)
	case *parser.UnaryExpr:
		return countBindMarkersInExpr(e.Expr)
	case *parser.FunctionCall:
		count := 0
		for _, arg := range e.Args {
			count += countBindMarkersInExpr(arg)
		}
		return count
	case *parser.CollectionAccessExpr:
		return countBindMarkersInExpr(e.Target) + countBindMarkersInExpr(e.Start) + countBindMarkersInExpr(e.End)
	case *parser.ListLiteral:
		count := 0
		for _, elem := range e.Elements {
			count += countBindMarkersInExpr(elem)
		}
		return count
	case *parser.SetLiteral:
		count := 0
		for _, elem := range e.Elements {
			count += countBindMarkersInExpr(elem)
		}
		return count
	case *parser.TupleLiteral:
		count := 0
		for _, elem := range e.Elements {
			count += countBindMarkersInExpr(elem)
		}
		return count
	case *parser.VectorLiteral:
		count := 0
		for _, elem := range e.Elements {
			count += countBindMarkersInExpr(elem)
		}
		return count
	case *parser.MapLiteral:
		count := 0
		for _, entry := range e.Entries {
			count += countBindMarkersInExpr(entry.Key)
			count += countBindMarkersInExpr(entry.Value)
		}
		return count
	default:
		return 0
	}
}

func (e *Executor) inferParamTypes(ctx context.Context, user, password string, stmt parser.Statement) ([]string, error) {
	switch s := stmt.(type) {
	case *parser.InsertStatement:
		columnTypes, err := e.lookupColumnTypes(ctx, user, password, s.Table)
		if err != nil {
			return nil, err
		}
		var types []string
		if s.JSON {
			appendTypesFromExpr(s.JSONValue, columnTypes, &types, "")
		} else {
			for _, row := range s.Values {
				for i, expr := range row {
					col := ""
					if i < len(s.Columns) {
						col = s.Columns[i]
					}
					appendTypesFromExpr(expr, columnTypes, &types, col)
				}
			}
		}
		if s.Timestamp != nil {
			appendTimestampParamTypes(s.Timestamp, &types)
		}
		if s.TTL != nil {
			appendTypesFromExpr(s.TTL, columnTypes, &types, "")
		}
		return types, nil
	case *parser.UpdateStatement:
		columnTypes, err := e.lookupColumnTypes(ctx, user, password, s.Table)
		if err != nil {
			return nil, err
		}
		var types []string
		if s.TTL != nil {
			appendTypesFromExpr(s.TTL, columnTypes, &types, "")
		}
		if s.Timestamp != nil {
			appendTimestampParamTypes(s.Timestamp, &types)
		}
		for _, assignment := range s.Assigments {
			appendAssignmentParamTypes(assignment, columnTypes, &types)
		}
		appendTypesFromExpr(s.Where, columnTypes, &types, "")
		if s.IfCondition != nil {
			appendTypesFromExpr(s.IfCondition, columnTypes, &types, "")
		}
		return types, nil
	case *parser.DeleteStatement:
		columnTypes, err := e.lookupColumnTypes(ctx, user, password, s.From)
		if err != nil {
			return nil, err
		}
		var types []string
		if s.Timestamp != nil {
			appendTimestampParamTypes(s.Timestamp, &types)
		}
		for _, assignment := range deleteAssignments(s) {
			appendAssignmentParamTypes(assignment, columnTypes, &types)
		}
		appendTypesFromExpr(s.Where, columnTypes, &types, "")
		if s.IfCondition != nil {
			appendTypesFromExpr(s.IfCondition, columnTypes, &types, "")
		}
		return types, nil
	case *parser.BatchStatement:
		var types []string
		if s.Using != nil {
			appendTimestampParamTypes(s.Using, &types)
		}
		for _, entry := range s.Entries {
			entryTypes, err := e.inferParamTypes(ctx, user, password, entry)
			if err != nil {
				return nil, err
			}
			types = append(types, entryTypes...)
		}
		return types, nil
	case *parser.SelectStatement:
		columnTypes, err := e.lookupColumnTypes(ctx, user, password, s.From)
		if err != nil {
			return nil, err
		}
		var types []string
		for _, col := range s.Columns {
			appendTypesFromExpr(col, columnTypes, &types, "")
		}
		appendTypesFromExpr(s.Where, columnTypes, &types, "")
		appendTypesFromExpr(s.Having, columnTypes, &types, "")
		for _, expr := range s.GroupBy {
			appendTypesFromExpr(expr, columnTypes, &types, "")
		}
		appendLimitTypes(s.Options.Limit, &types)
		appendLimitTypes(s.Options.PerPartitionLimit, &types)
		return types, nil
	default:
		return nil, nil
	}
}

// InferParamTypes infers parameter types for a CQL statement using MySQL metadata.
func (e *Executor) InferParamTypes(ctx context.Context, user, password string, stmt parser.Statement) ([]string, error) {
	return e.inferParamTypes(ctx, user, password, stmt)
}

func (e *Executor) lookupColumnTypes(ctx context.Context, user, password string, table *parser.TableName) (map[string]string, error) {
	if table == nil {
		return nil, fmt.Errorf("table is required for parameter type inference")
	}

	if table.Schema == "" {
		return nil, cqlerrors.Invalid(fmt.Sprintf("keyspace is required for %s", table.Name))
	}

	if e.schemaManager != nil {
		metaCols, metaAvailable, err := e.schemaManager.CQLColumnsForTable(ctx, user, password, table.Schema, table.Name)
		if err != nil {
			return nil, err
		}
		if metaAvailable && len(metaCols) > 0 {
			columnTypes := make(map[string]string, len(metaCols))
			for _, col := range metaCols {
				columnTypes[strings.ToLower(col.Name)] = col.CQLType
			}
			return columnTypes, nil
		}
		cols, err := e.schemaManager.ColumnsForTable(ctx, user, password, table.Schema, table.Name)
		if err != nil {
			return nil, err
		}
		if len(cols) > 0 {
			columnTypes := make(map[string]string, len(cols))
			for _, col := range cols {
				columnTypes[strings.ToLower(col.Name)] = typemap.MySQLTypeToCQLTypeStringWithPrecisionScale(col.DataType, col.NumericPrecision, col.NumericScale)
			}
			return columnTypes, nil
		}
	}

	conn, release, err := e.getConn(ctx, user, password)
	if err != nil {
		return nil, fmt.Errorf("failed to get connection: %w", err)
	}
	if release != nil {
		defer release()
	}

	rows, err := conn.QueryContext(ctx, `
		SELECT column_name, data_type, numeric_precision, numeric_scale
		FROM information_schema.columns
		WHERE table_schema = ? AND table_name = ?
	`, table.Schema, table.Name)
	if err != nil {
		return nil, fmt.Errorf("failed to lookup column types: %w", err)
	}
	defer rows.Close()

	columnTypes := make(map[string]string)
	for rows.Next() {
		var name string
		var dataType string
		var precision sql.NullInt64
		var scale sql.NullInt64
		if err := rows.Scan(&name, &dataType, &precision, &scale); err != nil {
			return nil, fmt.Errorf("failed to scan column types: %w", err)
		}
		precVal := 0
		scaleVal := 0
		if precision.Valid {
			precVal = int(precision.Int64)
		}
		if scale.Valid {
			scaleVal = int(scale.Int64)
		}
		columnTypes[strings.ToLower(name)] = typemap.MySQLTypeToCQLTypeStringWithPrecisionScale(dataType, precVal, scaleVal)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("failed to read column types: %w", err)
	}

	return columnTypes, nil
}

func lookupColumnType(columnTypes map[string]string, column string) string {
	if columnTypes == nil {
		return "text"
	}
	if cqlType, ok := columnTypes[strings.ToLower(column)]; ok && cqlType != "" {
		return cqlType
	}
	return "text"
}

func appendTypesFromExpr(expr parser.Expression, columnTypes map[string]string, types *[]string, defaultColumn string) {
	if expr == nil {
		return
	}

	switch e := expr.(type) {
	case *parser.BindMarker:
		if defaultColumn != "" {
			*types = append(*types, lookupColumnType(columnTypes, defaultColumn))
		} else {
			*types = append(*types, "text")
		}
	case *parser.NamedBindMarker:
		if defaultColumn != "" {
			*types = append(*types, lookupColumnType(columnTypes, defaultColumn))
		} else {
			*types = append(*types, "text")
		}
	case *parser.TypeHintExpr:
		if isDirectBindMarker(e.Expr) {
			*types = append(*types, typeStringForParam(e.Type))
			return
		}
		appendTypesFromExpr(e.Expr, columnTypes, types, "")
	case *parser.IsNullExpr:
		appendTypesFromExpr(e.Expr, columnTypes, types, "")
	case *parser.AliasedExpr:
		appendTypesFromExpr(e.Expr, columnTypes, types, "")
	case *parser.BinaryExpr:
		leftHint, leftIsBind := bindMarkerHint(e.Left)
		rightHint, rightIsBind := bindMarkerHint(e.Right)
		switch {
		case leftIsBind && rightIsBind:
			*types = append(*types, markerTypeOrDefault(leftHint), markerTypeOrDefault(rightHint))
		case leftIsBind:
			if leftHint != "" {
				*types = append(*types, leftHint)
			} else {
				*types = append(*types, inferTypeFromExpr(e.Right, columnTypes))
			}
			appendTypesFromExpr(e.Right, columnTypes, types, "")
		case rightIsBind:
			if rightHint != "" {
				*types = append(*types, rightHint)
			} else {
				*types = append(*types, inferTypeFromExpr(e.Left, columnTypes))
			}
			appendTypesFromExpr(e.Left, columnTypes, types, "")
		default:
			appendTypesFromExpr(e.Left, columnTypes, types, "")
			appendTypesFromExpr(e.Right, columnTypes, types, "")
		}
	case *parser.UnaryExpr:
		appendTypesFromExpr(e.Expr, columnTypes, types, "")
	case *parser.FunctionCall:
		for _, arg := range e.Args {
			appendTypesFromExpr(arg, columnTypes, types, "")
		}
	case *parser.CollectionAccessExpr:
		appendTypesFromExpr(e.Target, columnTypes, types, "")
		appendTypesFromExpr(e.Start, columnTypes, types, "")
		appendTypesFromExpr(e.End, columnTypes, types, "")
	case *parser.ListLiteral:
		for _, elem := range e.Elements {
			appendTypesFromExpr(elem, columnTypes, types, "")
		}
	case *parser.SetLiteral:
		for _, elem := range e.Elements {
			appendTypesFromExpr(elem, columnTypes, types, "")
		}
	case *parser.TupleLiteral:
		for _, elem := range e.Elements {
			appendTypesFromExpr(elem, columnTypes, types, "")
		}
	case *parser.VectorLiteral:
		for _, elem := range e.Elements {
			appendTypesFromExpr(elem, columnTypes, types, "")
		}
	case *parser.MapLiteral:
		for _, entry := range e.Entries {
			appendTypesFromExpr(entry.Key, columnTypes, types, "")
			appendTypesFromExpr(entry.Value, columnTypes, types, "")
		}
	}
}

func appendAssignmentParamTypes(assignment *parser.Assignment, columnTypes map[string]string, types *[]string) {
	if assignment == nil {
		return
	}
	appendAssignmentTargetTypes(assignment, columnTypes, types)
	appendAssignmentValueTypes(assignment, columnTypes, types)
}

func appendAssignmentTargetTypes(assignment *parser.Assignment, columnTypes map[string]string, types *[]string) {
	access, ok := assignment.Target.(*parser.CollectionAccessExpr)
	if !ok || access == nil {
		return
	}
	columnType := normalizeCQLTypeString(lookupColumnType(columnTypes, assignment.Column))
	switch {
	case strings.HasPrefix(columnType, "list<"):
		if !appendDirectBindMarkerType(access.Start, "int", types) {
			appendTypesFromExpr(access.Start, columnTypes, types, "")
		}
	case strings.HasPrefix(columnType, "map<"):
		keyType := mapKeyTypeFromString(columnType)
		if !appendDirectBindMarkerType(access.Start, keyType, types) {
			appendTypesFromExpr(access.Start, columnTypes, types, "")
		}
	default:
		appendTypesFromExpr(access.Start, columnTypes, types, "")
	}
}

func appendAssignmentValueTypes(assignment *parser.Assignment, columnTypes map[string]string, types *[]string) {
	if assignment == nil || assignment.Value == nil {
		return
	}
	columnType := normalizeCQLTypeString(lookupColumnType(columnTypes, assignment.Column))
	if bin, ok := assignment.Value.(*parser.BinaryExpr); ok && strings.EqualFold(bin.Op, "-") && strings.HasPrefix(columnType, "map<") {
		if _, ok := columnExprMatch(bin.Left, assignment.Column); ok {
			keyType := mapKeyTypeFromString(columnType)
			if appendDirectBindMarkerType(bin.Right, setTypeString(keyType), types) {
				return
			}
		}
		if _, ok := columnExprMatch(bin.Right, assignment.Column); ok {
			keyType := mapKeyTypeFromString(columnType)
			if appendDirectBindMarkerType(bin.Left, setTypeString(keyType), types) {
				return
			}
		}
	}
	appendTypesFromExpr(assignment.Value, columnTypes, types, assignment.Column)
}

func appendDirectBindMarkerType(expr parser.Expression, defaultType string, types *[]string) bool {
	switch e := expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		*types = append(*types, markerTypeOrDefault(defaultType))
		return true
	case *parser.TypeHintExpr:
		if isDirectBindMarker(e.Expr) {
			*types = append(*types, typeStringForParam(e.Type))
			return true
		}
		return appendDirectBindMarkerType(e.Expr, defaultType, types)
	default:
		return false
	}
}

func mapKeyTypeFromString(cqlType string) string {
	parts := splitTopLevelTypes(innerType(cqlType))
	if len(parts) < 2 {
		return "text"
	}
	return strings.TrimSpace(parts[0])
}

func setTypeString(elemType string) string {
	elemType = strings.TrimSpace(elemType)
	if elemType == "" {
		elemType = "text"
	}
	return "set<" + elemType + ">"
}

func appendTimestampParamTypes(expr parser.Expression, types *[]string) {
	if expr == nil {
		return
	}
	switch e := expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		*types = append(*types, "bigint")
	case *parser.TypeHintExpr:
		appendTimestampParamTypes(e.Expr, types)
	case *parser.IsNullExpr:
		appendTimestampParamTypes(e.Expr, types)
	case *parser.AliasedExpr:
		appendTimestampParamTypes(e.Expr, types)
	case *parser.BinaryExpr:
		appendTimestampParamTypes(e.Left, types)
		appendTimestampParamTypes(e.Right, types)
	case *parser.UnaryExpr:
		appendTimestampParamTypes(e.Expr, types)
	case *parser.FunctionCall:
		for _, arg := range e.Args {
			appendTimestampParamTypes(arg, types)
		}
	case *parser.CollectionAccessExpr:
		appendTimestampParamTypes(e.Target, types)
		appendTimestampParamTypes(e.Start, types)
		appendTimestampParamTypes(e.End, types)
	case *parser.ListLiteral:
		for _, elem := range e.Elements {
			appendTimestampParamTypes(elem, types)
		}
	case *parser.SetLiteral:
		for _, elem := range e.Elements {
			appendTimestampParamTypes(elem, types)
		}
	case *parser.TupleLiteral:
		for _, elem := range e.Elements {
			appendTimestampParamTypes(elem, types)
		}
	case *parser.VectorLiteral:
		for _, elem := range e.Elements {
			appendTimestampParamTypes(elem, types)
		}
	case *parser.MapLiteral:
		for _, entry := range e.Entries {
			appendTimestampParamTypes(entry.Key, types)
			appendTimestampParamTypes(entry.Value, types)
		}
	}
}

func appendLimitTypes(expr parser.Expression, types *[]string) {
	if expr == nil {
		return
	}
	switch e := expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		*types = append(*types, "int")
	case *parser.TypeHintExpr:
		appendLimitTypes(e.Expr, types)
	case *parser.UnaryExpr:
		appendLimitTypes(e.Expr, types)
	case *parser.BinaryExpr:
		appendLimitTypes(e.Left, types)
		appendLimitTypes(e.Right, types)
	}
}

func isDirectBindMarker(expr parser.Expression) bool {
	switch expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		return true
	default:
		return false
	}
}

func bindMarkerHint(expr parser.Expression) (string, bool) {
	switch e := expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		return "", true
	case *parser.TypeHintExpr:
		if isDirectBindMarker(e.Expr) {
			return typeStringForParam(e.Type), true
		}
		return "", false
	default:
		return "", false
	}
}

func markerTypeOrDefault(typ string) string {
	if typ != "" {
		return typ
	}
	return "text"
}

func inferTypeFromExpr(expr parser.Expression, columnTypes map[string]string) string {
	switch e := expr.(type) {
	case *parser.ColumnExpr:
		return lookupColumnType(columnTypes, e.Name)
	case *parser.IntegerLiteral:
		return "int"
	case *parser.FloatLiteral:
		return "double"
	case *parser.BooleanLiteral:
		return "boolean"
	case *parser.StringLiteral:
		return "text"
	case *parser.BlobLiteral:
		return "blob"
	case *parser.TypeHintExpr:
		return typeStringForParam(e.Type)
	default:
		return "text"
	}
}

func columnExprMatch(expr parser.Expression, column string) (*parser.ColumnExpr, bool) {
	col, ok := expr.(*parser.ColumnExpr)
	if !ok {
		return nil, false
	}
	if column == "" {
		return col, true
	}
	if strings.EqualFold(col.Name, column) {
		return col, true
	}
	if col.Table != "" {
		full := col.Table + "." + col.Name
		return col, strings.EqualFold(full, column)
	}
	return col, false
}

func typeStringForParam(typ parser.Type) string {
	if typ == nil {
		return "text"
	}
	return strings.ToLower(typ.TypeString())
}

func (e *Executor) decodePreparedParams(prepared *PreparedStatement, params []interface{}) ([]interface{}, error) {
	if len(params) == 0 {
		return nil, nil
	}

	decoded := make([]interface{}, len(params))
	for i, param := range params {
		if isUnsetParam(param) {
			decoded[i] = UnsetValue
			continue
		}
		if param == nil {
			decoded[i] = nil
			continue
		}
		raw, ok := param.([]byte)
		if !ok {
			decoded[i] = param
			continue
		}
		cqlType := "text"
		var typeInfo types.TypeInfo
		if i < len(prepared.Parameters) && prepared.Parameters[i].Type != "" {
			cqlType = prepared.Parameters[i].Type
			typeInfo = prepared.Parameters[i].TypeInfo
		}
		value, err := decodeCQLValue(raw, cqlType, typeInfo)
		if err != nil {
			return nil, fmt.Errorf("failed to decode param %d: %w", i, err)
		}
		decoded[i] = value
	}
	return decoded, nil
}

// DecodeParams decodes raw CQL parameter bytes using the provided type list.
func (e *Executor) DecodeParams(ctx context.Context, user, password string, params []interface{}, paramTypes []string) ([]interface{}, error) {
	if len(params) == 0 {
		return nil, nil
	}
	paramKeyspace := ""
	if state := keyspaceStateFromContext(ctx); state != nil {
		paramKeyspace = state.GetKeyspace()
	}
	decoded := make([]interface{}, len(params))
	for i, param := range params {
		if isUnsetParam(param) {
			decoded[i] = UnsetValue
			continue
		}
		if param == nil {
			decoded[i] = nil
			continue
		}
		raw, ok := param.([]byte)
		if !ok {
			decoded[i] = param
			continue
		}
		cqlType := "text"
		if i < len(paramTypes) && paramTypes[i] != "" {
			cqlType = paramTypes[i]
		}
		typeInfo, err := e.resolveTypeInfo(ctx, user, password, cqlType, paramKeyspace)
		if err != nil {
			return nil, fmt.Errorf("failed to resolve param type info for %q: %w", cqlType, err)
		}
		value, err := decodeCQLValue(raw, cqlType, typeInfo)
		if err != nil {
			return nil, fmt.Errorf("failed to decode param %d: %w", i, err)
		}
		decoded[i] = value
	}
	return decoded, nil
}

func decodeCQLValue(value []byte, cqlType string, typeInfo types.TypeInfo) (interface{}, error) {
	if value == nil {
		return nil, nil
	}
	normalized := normalizeCQLTypeString(cqlType)
	raw, err := decodeCQLValueRaw(value, normalized, typeInfo)
	if err != nil {
		return nil, err
	}
	if shouldJSONEncode(normalized, typeInfo) {
		jsonVal, err := cqlValueToJSON(raw, normalized, typeInfo)
		if err != nil {
			return nil, err
		}
		encoded, err := json.Marshal(jsonVal)
		if err != nil {
			return nil, err
		}
		return encoded, nil
	}
	switch normalized {
	case "time":
		if d, ok := raw.(time.Duration); ok {
			return formatDurationAsTime(d), nil
		}
	case "duration":
		if d, ok := raw.(cqlDuration); ok {
			return formatDurationValue(d), nil
		}
	}
	return raw, nil
}

func normalizeCQLTypeString(cqlType string) string {
	trimmed := strings.TrimSpace(cqlType)
	if trimmed == "" {
		return trimmed
	}
	lower := strings.ToLower(trimmed)
	if strings.HasPrefix(lower, "frozen<") && strings.HasSuffix(lower, ">") {
		inner := strings.TrimSuffix(strings.TrimPrefix(lower, "frozen<"), ">")
		return strings.TrimSpace(inner)
	}
	return lower
}

func shouldJSONEncode(cqlType string, typeInfo types.TypeInfo) bool {
	if _, ok := typeInfo.(*types.UDTTypeInfo); ok {
		return true
	}
	return strings.HasPrefix(cqlType, "list<") ||
		strings.HasPrefix(cqlType, "set<") ||
		strings.HasPrefix(cqlType, "map<") ||
		strings.HasPrefix(cqlType, "tuple<")
}

func decodeCQLValueRaw(value []byte, cqlType string, typeInfo types.TypeInfo) (interface{}, error) {
	if info, ok := typeInfo.(*types.UDTTypeInfo); ok {
		return decodeUDTValue(value, info)
	}
	if info, ok := typeInfo.(*types.VectorTypeInfo); ok {
		return decodeVectorValue(value, cqlType, info)
	}
	if strings.HasPrefix(cqlType, "vector<") {
		return decodeVectorValue(value, cqlType, nil)
	}
	if strings.HasPrefix(cqlType, "list<") || strings.HasPrefix(cqlType, "set<") {
		elemType := innerType(cqlType)
		var elemInfo types.TypeInfo
		switch info := typeInfo.(type) {
		case *types.ListTypeInfo:
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		case *types.SetTypeInfo:
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		}
		return decodeListValue(value, elemType, elemInfo)
	}
	if strings.HasPrefix(cqlType, "map<") {
		parts := splitTopLevelTypes(innerType(cqlType))
		if len(parts) < 2 {
			return nil, fmt.Errorf("invalid map type %q", cqlType)
		}
		keyType := parts[0]
		valType := parts[1]
		var keyInfo types.TypeInfo
		var valInfo types.TypeInfo
		if info, ok := typeInfo.(*types.MapTypeInfo); ok {
			if info.KeyType != "" {
				keyType = info.KeyType
			}
			if info.ValueType != "" {
				valType = info.ValueType
			}
			keyInfo = info.KeyInfo
			valInfo = info.ValueInfo
		}
		return decodeMapValue(value, keyType, keyInfo, valType, valInfo)
	}
	if strings.HasPrefix(cqlType, "tuple<") {
		elemTypes := splitTopLevelTypes(innerType(cqlType))
		var elemInfos []types.TypeInfo
		if info, ok := typeInfo.(*types.TupleTypeInfo); ok {
			if len(info.ElemTypes) > 0 {
				elemTypes = info.ElemTypes
			}
			elemInfos = info.ElemInfos
		}
		return decodeTupleValue(value, elemTypes, elemInfos)
	}
	if strings.HasPrefix(cqlType, "'") && strings.HasSuffix(cqlType, "'") {
		return string(value), nil
	}

	switch cqlType {
	case "int":
		if len(value) != 4 {
			return nil, fmt.Errorf("invalid int length %d", len(value))
		}
		return int32(binary.BigEndian.Uint32(value)), nil
	case "bigint", "counter":
		if len(value) != 8 {
			return nil, fmt.Errorf("invalid bigint length %d", len(value))
		}
		return int64(binary.BigEndian.Uint64(value)), nil
	case "smallint":
		if len(value) != 2 {
			return nil, fmt.Errorf("invalid smallint length %d", len(value))
		}
		return int16(binary.BigEndian.Uint16(value)), nil
	case "tinyint":
		if len(value) != 1 {
			return nil, fmt.Errorf("invalid tinyint length %d", len(value))
		}
		return int8(value[0]), nil
	case "boolean":
		if len(value) != 1 {
			return nil, fmt.Errorf("invalid boolean length %d", len(value))
		}
		return value[0] != 0, nil
	case "float":
		if len(value) != 4 {
			return nil, fmt.Errorf("invalid float length %d", len(value))
		}
		return math.Float32frombits(binary.BigEndian.Uint32(value)), nil
	case "double":
		if len(value) != 8 {
			return nil, fmt.Errorf("invalid double length %d", len(value))
		}
		return math.Float64frombits(binary.BigEndian.Uint64(value)), nil
	case "uuid", "timeuuid":
		uuid, err := formatUUID(value)
		if err != nil {
			return nil, err
		}
		return uuid, nil
	case "timestamp":
		if len(value) != 8 {
			return nil, fmt.Errorf("invalid timestamp length %d", len(value))
		}
		millis := int64(binary.BigEndian.Uint64(value))
		return time.Unix(0, millis*int64(time.Millisecond)), nil
	case "date":
		if len(value) != 4 {
			return nil, fmt.Errorf("invalid date length %d", len(value))
		}
		days := int64(binary.BigEndian.Uint32(value)) - (1 << 31)
		return time.Unix(0, 0).UTC().AddDate(0, 0, int(days)), nil
	case "time":
		if len(value) != 8 {
			return nil, fmt.Errorf("invalid time length %d", len(value))
		}
		nanos := int64(binary.BigEndian.Uint64(value))
		return time.Duration(nanos), nil
	case "duration":
		dur, err := decodeDurationBytes(value)
		if err != nil {
			return nil, err
		}
		return dur, nil
	case "inet":
		if len(value) != net.IPv4len && len(value) != net.IPv6len {
			return nil, fmt.Errorf("invalid inet length %d", len(value))
		}
		return net.IP(value).String(), nil
	case "blob":
		return value, nil
	case "decimal":
		return decodeDecimal(value)
	case "varint":
		return decodeVarint(value), nil
	case "text", "varchar", "ascii":
		return string(value), nil
	default:
		return string(value), nil
	}
}

func decodeListValue(value []byte, elemType string, elemInfo types.TypeInfo) ([]interface{}, error) {
	if len(value) < 4 {
		return nil, fmt.Errorf("invalid collection length %d", len(value))
	}
	count := int(int32(binary.BigEndian.Uint32(value[:4])))
	if count < 0 {
		return nil, fmt.Errorf("invalid collection count %d", count)
	}
	out := make([]interface{}, count)
	offset := 4
	for i := 0; i < count; i++ {
		if len(value[offset:]) < 4 {
			return nil, fmt.Errorf("invalid collection element length")
		}
		elemLen := int(int32(binary.BigEndian.Uint32(value[offset:])))
		offset += 4
		if elemLen < 0 {
			out[i] = nil
			continue
		}
		if len(value[offset:]) < elemLen {
			return nil, fmt.Errorf("invalid collection element size")
		}
		elemBytes := value[offset : offset+elemLen]
		offset += elemLen
		elemVal, err := decodeCQLValueRaw(elemBytes, elemType, elemInfo)
		if err != nil {
			return nil, err
		}
		out[i] = elemVal
	}
	return out, nil
}

func decodeMapValue(value []byte, keyType string, keyInfo types.TypeInfo, valType string, valInfo types.TypeInfo) (map[string]interface{}, error) {
	if len(value) < 4 {
		return nil, fmt.Errorf("invalid map length %d", len(value))
	}
	count := int(int32(binary.BigEndian.Uint32(value[:4])))
	if count < 0 {
		return nil, fmt.Errorf("invalid map count %d", count)
	}
	out := make(map[string]interface{}, count)
	offset := 4
	for i := 0; i < count; i++ {
		if len(value[offset:]) < 4 {
			return nil, fmt.Errorf("invalid map key length")
		}
		keyLen := int(int32(binary.BigEndian.Uint32(value[offset:])))
		offset += 4
		if keyLen < 0 {
			return nil, fmt.Errorf("invalid map key length %d", keyLen)
		}
		if len(value[offset:]) < keyLen {
			return nil, fmt.Errorf("invalid map key size")
		}
		keyBytes := value[offset : offset+keyLen]
		offset += keyLen
		keyVal, err := decodeCQLValueRaw(keyBytes, keyType, keyInfo)
		if err != nil {
			return nil, err
		}
		keyStr, err := cqlValueToString(keyVal, keyType)
		if err != nil {
			return nil, err
		}

		if len(value[offset:]) < 4 {
			return nil, fmt.Errorf("invalid map value length")
		}
		valLen := int(int32(binary.BigEndian.Uint32(value[offset:])))
		offset += 4
		if valLen < 0 {
			out[keyStr] = nil
			continue
		}
		if len(value[offset:]) < valLen {
			return nil, fmt.Errorf("invalid map value size")
		}
		valBytes := value[offset : offset+valLen]
		offset += valLen
		valVal, err := decodeCQLValueRaw(valBytes, valType, valInfo)
		if err != nil {
			return nil, err
		}
		out[keyStr] = valVal
	}
	return out, nil
}

func decodeTupleValue(value []byte, elemTypes []string, elemInfos []types.TypeInfo) ([]interface{}, error) {
	out := make([]interface{}, len(elemTypes))
	offset := 0
	for i, elemType := range elemTypes {
		if len(value[offset:]) < 4 {
			return nil, fmt.Errorf("invalid tuple element length")
		}
		elemLen := int(int32(binary.BigEndian.Uint32(value[offset:])))
		offset += 4
		if elemLen < 0 {
			out[i] = nil
			continue
		}
		if len(value[offset:]) < elemLen {
			return nil, fmt.Errorf("invalid tuple element size")
		}
		elemBytes := value[offset : offset+elemLen]
		offset += elemLen
		var elemInfo types.TypeInfo
		if i < len(elemInfos) {
			elemInfo = elemInfos[i]
		}
		elemVal, err := decodeCQLValueRaw(elemBytes, elemType, elemInfo)
		if err != nil {
			return nil, err
		}
		out[i] = elemVal
	}
	return out, nil
}

func decodeUDTValue(value []byte, info *types.UDTTypeInfo) (map[string]interface{}, error) {
	if info == nil {
		return nil, fmt.Errorf("udt type info is required")
	}
	out := make(map[string]interface{}, len(info.Fields))
	offset := 0
	for _, field := range info.Fields {
		if len(value[offset:]) < 4 {
			return nil, fmt.Errorf("invalid udt field length")
		}
		fieldLen := int(int32(binary.BigEndian.Uint32(value[offset:])))
		offset += 4
		if fieldLen < 0 {
			out[field.Name] = nil
			continue
		}
		if len(value[offset:]) < fieldLen {
			return nil, fmt.Errorf("invalid udt field size")
		}
		fieldBytes := value[offset : offset+fieldLen]
		offset += fieldLen
		fieldVal, err := decodeCQLValueRaw(fieldBytes, field.Type, field.TypeInfo)
		if err != nil {
			return nil, err
		}
		out[field.Name] = fieldVal
	}
	return out, nil
}

func vectorTypeSpec(cqlType string, info *types.VectorTypeInfo) (string, int, int, bool) {
	elemType := ""
	size := 0
	if info != nil {
		elemType = info.ElemType
		size = info.Size
	}
	if (elemType == "" || size <= 0) && strings.HasPrefix(cqlType, "vector<") {
		if parsedElem, parsedSize, ok := parseVectorType(cqlType); ok {
			elemType = parsedElem
			size = parsedSize
		}
	}
	elemBytes := vectorElementBytes(elemType)
	if elemBytes == 0 || size <= 0 {
		return "", 0, 0, false
	}
	return elemType, size, elemBytes, true
}

func decodeVectorValue(value []byte, cqlType string, info *types.VectorTypeInfo) ([]byte, error) {
	_, size, elemBytes, ok := vectorTypeSpec(cqlType, info)
	if !ok {
		return nil, fmt.Errorf("invalid vector type %q", cqlType)
	}
	expectedLen := size * elemBytes
	if len(value) != expectedLen {
		return nil, fmt.Errorf("invalid vector length %d (expected %d)", len(value), expectedLen)
	}
	return value, nil
}

func decodeVectorElements(value []byte, cqlType string, info *types.VectorTypeInfo) (interface{}, error) {
	_, size, elemBytes, ok := vectorTypeSpec(cqlType, info)
	if !ok {
		return nil, fmt.Errorf("invalid vector type %q", cqlType)
	}
	expectedLen := size * elemBytes
	if len(value) != expectedLen {
		return nil, fmt.Errorf("invalid vector length %d (expected %d)", len(value), expectedLen)
	}
	switch elemBytes {
	case 4:
		out := make([]float32, size)
		for i := 0; i < size; i++ {
			bits := binary.BigEndian.Uint32(value[i*4 : (i+1)*4])
			out[i] = math.Float32frombits(bits)
		}
		return out, nil
	case 8:
		out := make([]float64, size)
		for i := 0; i < size; i++ {
			bits := binary.BigEndian.Uint64(value[i*8 : (i+1)*8])
			out[i] = math.Float64frombits(bits)
		}
		return out, nil
	default:
		return nil, fmt.Errorf("unsupported vector element size %d", elemBytes)
	}
}

func cqlValueToJSON(value interface{}, cqlType string, typeInfo types.TypeInfo) (interface{}, error) {
	if value == nil {
		return nil, nil
	}
	if info, ok := typeInfo.(*types.UDTTypeInfo); ok {
		m, ok := value.(map[string]interface{})
		if !ok {
			return nil, fmt.Errorf("udt value must be map")
		}
		out := make(map[string]interface{}, len(info.Fields))
		for _, field := range info.Fields {
			fieldVal := m[field.Name]
			encoded, err := cqlValueToJSON(fieldVal, field.Type, field.TypeInfo)
			if err != nil {
				return nil, err
			}
			out[field.Name] = encoded
		}
		return out, nil
	}
	if strings.HasPrefix(cqlType, "list<") || strings.HasPrefix(cqlType, "set<") {
		elemType := innerType(cqlType)
		var elemInfo types.TypeInfo
		switch info := typeInfo.(type) {
		case *types.ListTypeInfo:
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		case *types.SetTypeInfo:
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		}
		arr, ok := value.([]interface{})
		if !ok {
			return nil, fmt.Errorf("collection value must be slice")
		}
		out := make([]interface{}, len(arr))
		for i, elem := range arr {
			encoded, err := cqlValueToJSON(elem, elemType, elemInfo)
			if err != nil {
				return nil, err
			}
			out[i] = encoded
		}
		return out, nil
	}
	if strings.HasPrefix(cqlType, "map<") {
		parts := splitTopLevelTypes(innerType(cqlType))
		if len(parts) < 2 {
			return nil, fmt.Errorf("invalid map type %q", cqlType)
		}
		valType := parts[1]
		var valInfo types.TypeInfo
		if info, ok := typeInfo.(*types.MapTypeInfo); ok {
			if info.ValueType != "" {
				valType = info.ValueType
			}
			valInfo = info.ValueInfo
		}
		m, ok := value.(map[string]interface{})
		if !ok {
			return nil, fmt.Errorf("map value must be map")
		}
		out := make(map[string]interface{}, len(m))
		for key, val := range m {
			encoded, err := cqlValueToJSON(val, valType, valInfo)
			if err != nil {
				return nil, err
			}
			out[key] = encoded
		}
		return out, nil
	}
	if strings.HasPrefix(cqlType, "tuple<") {
		elemTypes := splitTopLevelTypes(innerType(cqlType))
		var elemInfos []types.TypeInfo
		if info, ok := typeInfo.(*types.TupleTypeInfo); ok {
			if len(info.ElemTypes) > 0 {
				elemTypes = info.ElemTypes
			}
			elemInfos = info.ElemInfos
		}
		arr, ok := value.([]interface{})
		if !ok {
			return nil, fmt.Errorf("tuple value must be slice")
		}
		out := make([]interface{}, len(elemTypes))
		for i, elemType := range elemTypes {
			var elemInfo types.TypeInfo
			if i < len(elemInfos) {
				elemInfo = elemInfos[i]
			}
			var elem interface{}
			if i < len(arr) {
				elem = arr[i]
			}
			encoded, err := cqlValueToJSON(elem, elemType, elemInfo)
			if err != nil {
				return nil, err
			}
			out[i] = encoded
		}
		return out, nil
	}
	if strings.HasPrefix(cqlType, "vector<") {
		if bytes, ok := value.([]byte); ok {
			var vectorInfo *types.VectorTypeInfo
			if info, ok := typeInfo.(*types.VectorTypeInfo); ok {
				vectorInfo = info
			}
			decoded, err := decodeVectorElements(bytes, cqlType, vectorInfo)
			if err != nil {
				return nil, err
			}
			return decoded, nil
		}
		switch v := value.(type) {
		case []float32, []float64:
			return v, nil
		}
	}

	switch cqlType {
	case "blob":
		if bytes, ok := value.([]byte); ok {
			return hex.EncodeToString(bytes), nil
		}
		if str, ok := value.(string); ok {
			return str, nil
		}
		return fmt.Sprint(value), nil
	case "timestamp":
		return formatTimestampValue(value), nil
	case "date":
		return formatDateValue(value), nil
	case "time":
		return formatTimeValue(value), nil
	case "duration":
		if d, ok := value.(cqlDuration); ok {
			return formatDurationValue(d), nil
		}
		if str, ok := value.(string); ok {
			return str, nil
		}
		return fmt.Sprint(value), nil
	}
	return value, nil
}

func cqlValueToString(value interface{}, cqlType string) (string, error) {
	if value == nil {
		return "", nil
	}
	switch cqlType {
	case "blob":
		if bytes, ok := value.([]byte); ok {
			return hex.EncodeToString(bytes), nil
		}
		if str, ok := value.(string); ok {
			return str, nil
		}
		return fmt.Sprint(value), nil
	case "timestamp":
		return formatTimestampValue(value), nil
	case "date":
		return formatDateValue(value), nil
	case "time":
		return formatTimeValue(value), nil
	case "duration":
		if d, ok := value.(cqlDuration); ok {
			return formatDurationValue(d), nil
		}
		return fmt.Sprint(value), nil
	}
	if _, ok := value.(string); ok {
		return fmt.Sprint(value), nil
	}
	return fmt.Sprint(value), nil
}

func formatUUID(value []byte) (string, error) {
	if len(value) != 16 {
		return "", fmt.Errorf("invalid UUID length %d", len(value))
	}
	hexStr := hex.EncodeToString(value)
	return fmt.Sprintf("%s-%s-%s-%s-%s",
		hexStr[0:8],
		hexStr[8:12],
		hexStr[12:16],
		hexStr[16:20],
		hexStr[20:32],
	), nil
}

func decodeVarint(value []byte) string {
	if len(value) == 0 {
		return "0"
	}
	return decodeSignedBigInt(value).String()
}

func decodeDecimal(value []byte) (string, error) {
	if len(value) < 4 {
		return "", fmt.Errorf("invalid decimal length %d", len(value))
	}
	scale := int32(binary.BigEndian.Uint32(value[:4]))
	unscaled := decodeSignedBigInt(value[4:])
	return formatDecimal(unscaled, int(scale)), nil
}

func decodeSignedBigInt(value []byte) *big.Int {
	if len(value) == 0 {
		return big.NewInt(0)
	}
	negative := value[0]&0x80 != 0
	if !negative {
		return new(big.Int).SetBytes(value)
	}
	// Two's complement
	inverted := make([]byte, len(value))
	for i := range value {
		inverted[i] = ^value[i]
	}
	intVal := new(big.Int).SetBytes(inverted)
	intVal.Add(intVal, big.NewInt(1))
	intVal.Neg(intVal)
	return intVal
}

func formatDecimal(unscaled *big.Int, scale int) string {
	if scale == 0 {
		return unscaled.String()
	}

	sign := ""
	valueStr := unscaled.String()
	if strings.HasPrefix(valueStr, "-") {
		sign = "-"
		valueStr = strings.TrimPrefix(valueStr, "-")
	}
	if scale >= len(valueStr) {
		valueStr = strings.Repeat("0", scale-len(valueStr)+1) + valueStr
	}
	pos := len(valueStr) - scale
	return sign + valueStr[:pos] + "." + valueStr[pos:]
}

// ExecuteToFrame executes a CQL query and returns a CQL protocol frame (as bytes)
func (e *Executor) ExecuteToFrame(ctx context.Context, user, password, cql string, stream int16) ([]byte, int16, error) {
	// Execute query
	result, err := e.Execute(ctx, user, password, cql)
	if err != nil {
		return nil, 0, err
	}

	// Translate to CQL frame bytes
	resultTranslator := NewQueryResultTranslator()
	frame, err := resultTranslator.TranslateToResultFrame(result, stream)
	if err != nil {
		return nil, 0, err
	}

	// Return frame body and opcode for the handler to create the response
	return frame.Body, int16(frame.Opcode), nil
}

// convertQueryResult returns the input result or a non-nil empty result.
func convertQueryResult(result *types.QueryResult) *QueryResult {
	if result == nil {
		return &QueryResult{
			Columns: make([]ColumnInfo, 0),
			Rows:    make([]Row, 0),
		}
	}

	columns := make([]ColumnInfo, len(result.Columns))
	for i, col := range result.Columns {
		columns[i] = ColumnInfo{
			Keyspace: col.Keyspace,
			Table:    col.Table,
			Name:     col.Name,
			Type:     col.Type,
			TypeInfo: col.TypeInfo,
			Nullable: col.Nullable,
		}
	}

	rows := make([]Row, len(result.Rows))
	for i, r := range result.Rows {
		rows[i] = Row{Values: r.Values}
	}

	return &QueryResult{
		Columns:       columns,
		Rows:          rows,
		RowsAffected:  result.RowsAffected,
		CQLWriteRows:  result.CQLWriteRows,
		LastInsertID:  result.LastInsertID,
		SetKeyspace:   result.SetKeyspace,
		UseGlobalSpec: result.UseGlobalSpec,
		PagingState:   result.PagingState,
		HasMorePages:  result.HasMorePages,
		SchemaChange:  result.SchemaChange,
		Warnings:      result.Warnings,
	}
}