package executor

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

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

type selectTransform struct {
	json        bool
	toJSON      []toJSONSpec
	casts       []castSpec
	dropIndexes []int
}

type toJSONSpec struct {
	index      int
	name       string
	sqlHandled bool
	typ        jsonTypeInfo
}

type selectPlan struct {
	sql       string
	transform selectTransform
	limitPlan *selectLimitPlan
	meta      *selectMeta
}

type selectMeta struct {
	columnInfo map[string]columnTypeInfo
	aliasToCol map[string]string
	keyspace   string
	table      string
}

type columnTypeInfo struct {
	cqlType   string
	mysqlType string
	typeInfo  types.TypeInfo
}

type limitTermKind uint8

const (
	limitTermLiteral limitTermKind = iota
	limitTermBind
)

type limitTerm struct {
	kind  limitTermKind
	value int64
	index int
}

type selectLimitPlan struct {
	limit         limitTerm
	perPartition  limitTerm
	offset        limitTerm
	dropIndexes   []int
	hasLimit      bool
	hasPartition  bool
	hasOffset     bool
	applyInMemory bool
}

type selectLimitDecision struct {
	limitOverride *int64
	omitLimit     bool
	omitOffset    bool
	plan          *selectLimitPlan
}

func (e *selectPlanner) buildSelectPlan(ctx context.Context, user, password string, stmt *parser.SelectStatement, allowLimitBind bool, proxyFilter bool) (*selectPlan, error) {
	if stmt == nil {
		return nil, fmt.Errorf("select statement is required")
	}
	if err := e.resolveTableName(ctx, user, password, stmt.From); err != nil {
		return nil, err
	}
	if err := e.validateSelectGroupBy(ctx, user, password, stmt); err != nil {
		return nil, err
	}
	if stmt.IndexHint == "" && e.schemaManager != nil {
		hint, err := e.selectIndexHint(ctx, user, password, stmt)
		if err != nil {
			return nil, err
		}
		stmt.IndexHint = hint
	}

	limitDecision, err := e.resolveSelectLimit(stmt, allowLimitBind, proxyFilter)
	if err != nil {
		return nil, err
	}

	hasToJSON := selectHasToJSON(stmt)
	hasCast := selectHasCast(stmt)
	hasTimeFunction := selectHasTimeFunction(stmt)
	transform := selectTransform{json: stmt.JSON}

	var typeInfo map[string]columnTypeInfo
	if stmt.From != nil && stmt.From.Schema != "" {
		typeInfo, err = e.columnTypesForTable(ctx, user, password, stmt.From)
		if err != nil {
			return nil, err
		}
	}

	if !hasToJSON && !hasCast && !hasTimeFunction && limitDecision == nil {
		sql, err := e.translator.Translate(stmt)
		if err != nil {
			return nil, err
		}
		return &selectPlan{sql: sql, transform: transform, meta: buildSelectMeta(stmt, typeInfo)}, nil
	}

	if hasToJSON && typeInfo == nil {
		return nil, fmt.Errorf("keyspace is required for to_json type lookup")
	}

	sql, planTransform, err := e.buildSelectSQLWithToJSON(stmt, typeInfo, limitDecision)
	if err != nil {
		return nil, err
	}
	planTransform.json = stmt.JSON
	var limitPlan *selectLimitPlan
	if limitDecision != nil {
		limitPlan = limitDecision.plan
	}
	return &selectPlan{sql: sql, transform: planTransform, limitPlan: limitPlan, meta: buildSelectMeta(stmt, typeInfo)}, nil
}

func buildSelectMeta(stmt *parser.SelectStatement, typeInfo map[string]columnTypeInfo) *selectMeta {
	if stmt == nil || stmt.From == nil || stmt.From.Schema == "" || len(typeInfo) == 0 {
		return nil
	}
	aliasToCol := invertAliasMap(extractColumnAliases(stmt))
	return &selectMeta{
		columnInfo: typeInfo,
		aliasToCol: aliasToCol,
		keyspace:   stmt.From.Schema,
		table:      stmt.From.Name,
	}
}

func invertAliasMap(aliases map[string]string) map[string]string {
	if len(aliases) == 0 {
		return nil
	}
	out := make(map[string]string, len(aliases))
	for base, alias := range aliases {
		out[alias] = base
	}
	return out
}

func (e *selectPlanner) resolveSelectLimit(stmt *parser.SelectStatement, allowBind bool, suppressSQL bool) (*selectLimitDecision, error) {
	if stmt == nil {
		return nil, nil
	}
	limitExpr := stmt.Options.Limit
	ppExpr := stmt.Options.PerPartitionLimit
	offsetExpr := stmt.Options.Offset
	if offsetExpr != nil && limitExpr == nil && ppExpr == nil {
		return nil, fmt.Errorf("OFFSET requires LIMIT")
	}
	if limitExpr == nil && ppExpr == nil && offsetExpr == nil {
		return nil, nil
	}

	markerIndex := buildBindMarkerIndex(stmt)
	var limitTermVal limitTerm
	var ppTermVal limitTerm
	var err error

	if limitExpr != nil {
		limitTermVal, err = parseLimitTerm(limitExpr, markerIndex)
		if err != nil {
			return nil, fmt.Errorf("invalid LIMIT term: %w", err)
		}
		if limitTermVal.kind == limitTermBind && !allowBind {
			return nil, fmt.Errorf("LIMIT requires an integer literal")
		}
	}
	if ppExpr != nil {
		ppTermVal, err = parseLimitTerm(ppExpr, markerIndex)
		if err != nil {
			return nil, fmt.Errorf("invalid PER PARTITION LIMIT term: %w", err)
		}
		if ppTermVal.kind == limitTermBind && !allowBind {
			return nil, fmt.Errorf("PER PARTITION LIMIT requires an integer literal")
		}
	}

	if suppressSQL {
		var offsetTermVal limitTerm
		if offsetExpr != nil {
			offsetTermVal, err = parseOffsetTerm(offsetExpr, markerIndex)
			if err != nil {
				return nil, err
			}
			if offsetTermVal.kind == limitTermBind && !allowBind {
				return nil, fmt.Errorf("OFFSET requires an integer literal")
			}
		}
		plan := newSelectLimitPlan(limitTermVal, limitExpr != nil, ppTermVal, ppExpr != nil, offsetTermVal, offsetExpr != nil, true)
		plan.applyInMemory = true
		return &selectLimitDecision{omitLimit: true, omitOffset: offsetExpr != nil, plan: plan}, nil
	}

	if limitExpr == nil || ppExpr == nil {
		return nil, nil
	}

	if limitTermVal.kind == limitTermLiteral && ppTermVal.kind == limitTermLiteral {
		effective := limitTermVal.value
		if ppTermVal.value < effective {
			effective = ppTermVal.value
		}
		return &selectLimitDecision{limitOverride: &effective, omitOffset: false}, nil
	}
	if !allowBind {
		return nil, fmt.Errorf("LIMIT and PER PARTITION LIMIT require integer literals")
	}

	var offsetTermVal limitTerm
	if offsetExpr != nil {
		offsetTermVal, err = parseOffsetTerm(offsetExpr, markerIndex)
		if err != nil {
			return nil, err
		}
		if offsetTermVal.kind == limitTermBind && !allowBind {
			return nil, fmt.Errorf("OFFSET requires an integer literal")
		}
	}
	plan := newSelectLimitPlan(limitTermVal, true, ppTermVal, true, offsetTermVal, offsetExpr != nil, offsetExpr != nil)
	return &selectLimitDecision{omitLimit: true, omitOffset: offsetExpr != nil, plan: plan}, nil
}

func parseOffsetTerm(expr parser.Expression, markerIndex map[parser.Expression]int) (limitTerm, error) {
	term, err := parseLimitTerm(expr, markerIndex)
	if err != nil {
		return limitTerm{}, fmt.Errorf("invalid OFFSET term: %w", err)
	}
	return term, nil
}

func parseLimitTerm(expr parser.Expression, markerIndex map[parser.Expression]int) (limitTerm, error) {
	if expr == nil {
		return limitTerm{}, fmt.Errorf("limit term is required")
	}
	switch e := expr.(type) {
	case *parser.TypeHintExpr:
		return parseLimitTerm(e.Expr, markerIndex)
	case *parser.UnaryExpr:
		switch e.Op {
		case "+":
			return parseLimitTerm(e.Expr, markerIndex)
		case "-":
			return limitTerm{}, fmt.Errorf("limit must be non-negative")
		default:
			return limitTerm{}, fmt.Errorf("unsupported unary operator %q", e.Op)
		}
	case *parser.IntegerLiteral:
		if e.Value < 0 {
			return limitTerm{}, fmt.Errorf("limit must be non-negative")
		}
		return limitTerm{kind: limitTermLiteral, value: e.Value, index: -1}, nil
	case *parser.BindMarker, *parser.NamedBindMarker:
		if markerIndex == nil {
			return limitTerm{}, fmt.Errorf("bind marker index is not available")
		}
		idx, ok := markerIndex[expr]
		if !ok {
			return limitTerm{}, fmt.Errorf("bind marker index not found")
		}
		return limitTerm{kind: limitTermBind, index: idx}, nil
	default:
		return limitTerm{}, fmt.Errorf("limit must be an integer literal or bind marker")
	}
}

func newSelectLimitPlan(limitTermVal limitTerm, hasLimit bool, ppTermVal limitTerm, hasPartition bool, offsetTermVal limitTerm, hasOffset bool, dropOffset bool) *selectLimitPlan {
	drop := make([]int, 0, 3)
	if hasLimit && limitTermVal.kind == limitTermBind {
		drop = append(drop, limitTermVal.index)
	}
	if hasPartition && ppTermVal.kind == limitTermBind {
		drop = append(drop, ppTermVal.index)
	}
	if dropOffset && hasOffset && offsetTermVal.kind == limitTermBind {
		drop = append(drop, offsetTermVal.index)
	}
	if len(drop) > 1 {
		sort.Ints(drop)
		drop = uniqueInts(drop)
	}
	return &selectLimitPlan{
		limit:        limitTermVal,
		perPartition: ppTermVal,
		offset:       offsetTermVal,
		dropIndexes:  drop,
		hasLimit:     hasLimit,
		hasPartition: hasPartition,
		hasOffset:    hasOffset,
	}
}

func (p *selectLimitPlan) effectiveLimit(params []interface{}) (int64, error) {
	if p == nil {
		return 0, fmt.Errorf("select limit plan is nil")
	}
	if !p.hasLimit && !p.hasPartition {
		return 0, fmt.Errorf("select limit plan has no limit terms")
	}
	var limitVal int64
	var ppVal int64
	if p.hasLimit {
		val, err := limitTermValue(p.limit, params)
		if err != nil {
			return 0, err
		}
		limitVal = val
	}
	if p.hasPartition {
		val, err := limitTermValue(p.perPartition, params)
		if err != nil {
			return 0, err
		}
		ppVal = val
	}
	switch {
	case p.hasLimit && p.hasPartition:
		if ppVal < limitVal {
			return ppVal, nil
		}
		return limitVal, nil
	case p.hasLimit:
		return limitVal, nil
	default:
		return ppVal, nil
	}
}

func (p *selectLimitPlan) effectiveOffset(params []interface{}) (int64, error) {
	if p == nil {
		return 0, fmt.Errorf("select limit plan is nil")
	}
	if !p.hasOffset {
		return 0, fmt.Errorf("select limit plan has no offset term")
	}
	return limitTermValue(p.offset, params)
}

func limitTermValue(term limitTerm, params []interface{}) (int64, error) {
	switch term.kind {
	case limitTermLiteral:
		return term.value, nil
	case limitTermBind:
		if term.index < 0 || term.index >= len(params) {
			return 0, fmt.Errorf("limit parameter index %d out of range", term.index)
		}
		return limitValueFromParam(params[term.index])
	default:
		return 0, fmt.Errorf("unknown limit term kind")
	}
}

func limitValueFromParam(value interface{}) (int64, error) {
	if value == nil {
		return 0, fmt.Errorf("limit value is required")
	}
	switch v := value.(type) {
	case int:
		return clampLimit(int64(v))
	case int8:
		return clampLimit(int64(v))
	case int16:
		return clampLimit(int64(v))
	case int32:
		return clampLimit(int64(v))
	case int64:
		return clampLimit(v)
	case uint:
		return clampLimit(int64(v))
	case uint8:
		return clampLimit(int64(v))
	case uint16:
		return clampLimit(int64(v))
	case uint32:
		return clampLimit(int64(v))
	case uint64:
		const maxInt64 = int64(^uint64(0) >> 1)
		if v > uint64(maxInt64) {
			return 0, fmt.Errorf("limit value is too large")
		}
		return clampLimit(int64(v))
	case string:
		parsed, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64)
		if err != nil {
			return 0, fmt.Errorf("invalid limit value %q", v)
		}
		return clampLimit(parsed)
	default:
		return 0, fmt.Errorf("invalid limit value type %T", value)
	}
}

func clampLimit(value int64) (int64, error) {
	if value < 0 {
		return 0, fmt.Errorf("limit must be non-negative")
	}
	return value, nil
}

func filterLimitParams(params []interface{}, drop []int) ([]interface{}, error) {
	if len(drop) == 0 {
		return params, nil
	}
	out := make([]interface{}, 0, len(params)-len(drop))
	dropIdx := 0
	for i, param := range params {
		if dropIdx < len(drop) && i == drop[dropIdx] {
			dropIdx++
			continue
		}
		out = append(out, param)
	}
	if dropIdx != len(drop) {
		return nil, fmt.Errorf("limit parameter index out of range")
	}
	return out, nil
}

func applySelectLimit(result *QueryResult, limit int64) {
	if result == nil || limit < 0 {
		return
	}
	if limit == 0 {
		result.Rows = nil
		return
	}
	if int64(len(result.Rows)) > limit {
		result.Rows = result.Rows[:limit]
	}
}

func applySelectOffset(result *QueryResult, offset int64) {
	if result == nil || offset <= 0 {
		return
	}
	if int64(len(result.Rows)) <= offset {
		result.Rows = nil
		return
	}
	result.Rows = result.Rows[offset:]
}

func appendLimitOffsetSQL(sql string, limit *int64, offset *int64) string {
	if limit == nil && offset == nil {
		return sql
	}

	var builder strings.Builder
	builder.WriteString(sql)
	builder.WriteString(" LIMIT ")
	if limit != nil {
		builder.WriteString(strconv.FormatInt(*limit, 10))
	}
	if offset != nil {
		builder.WriteString(" OFFSET ")
		builder.WriteString(strconv.FormatInt(*offset, 10))
	}
	return builder.String()
}

func selectMetadataQuery(sql string) (string, []interface{}) {
	metaSQL := "SELECT * FROM (" + sql + ") AS cql_metadata LIMIT 0"
	paramCount := countSQLBindMarkers(sql)
	if paramCount <= 0 {
		return metaSQL, nil
	}
	args := make([]interface{}, paramCount)
	for i := range args {
		args[i] = 0
	}
	return metaSQL, args
}

func countSQLBindMarkers(sql string) int {
	// This scans SQL emitted by our translator, which never includes SQL
	// comments. It intentionally only skips quoted strings and identifiers.
	count := 0
	inSingle := false
	inDouble := false
	inBacktick := false
	escaped := false

	for i := 0; i < len(sql); i++ {
		ch := sql[i]
		if escaped {
			escaped = false
			continue
		}
		if inSingle {
			if ch == '\\' {
				escaped = true
			} else if ch == '\'' {
				if i+1 < len(sql) && sql[i+1] == '\'' {
					i++
				} else {
					inSingle = false
				}
			}
			continue
		}
		if inDouble {
			if ch == '\\' {
				escaped = true
			} else if ch == '"' {
				if i+1 < len(sql) && sql[i+1] == '"' {
					i++
				} else {
					inDouble = false
				}
			}
			continue
		}
		if inBacktick {
			if ch == '`' {
				if i+1 < len(sql) && sql[i+1] == '`' {
					i++
				} else {
					inBacktick = false
				}
			}
			continue
		}

		switch ch {
		case '\'':
			inSingle = true
		case '"':
			inDouble = true
		case '`':
			inBacktick = true
		case '?':
			count++
		}
	}
	return count
}

func uniqueInts(values []int) []int {
	if len(values) < 2 {
		return values
	}
	out := values[:1]
	for _, v := range values[1:] {
		if v != out[len(out)-1] {
			out = append(out, v)
		}
	}
	return out
}

func (e *selectPlanner) columnTypesForTable(ctx context.Context, user, password string, table *parser.TableName) (map[string]columnTypeInfo, error) {
	if table == nil || table.Schema == "" {
		return nil, fmt.Errorf("keyspace is required for table lookup")
	}
	if e.schemaManager == nil {
		return nil, nil
	}
	cols, err := e.schemaManager.ColumnsForTable(ctx, user, password, table.Schema, table.Name)
	if err != nil {
		return nil, err
	}
	info := make(map[string]columnTypeInfo, len(cols))
	for _, col := range cols {
		info[strings.ToLower(col.Name)] = columnTypeInfo{
			cqlType:   typemap.MySQLTypeToCQLTypeStringWithPrecisionScale(col.DataType, col.NumericPrecision, col.NumericScale),
			mysqlType: col.DataType,
		}
	}

	metaCols, metaAvailable, err := e.schemaManager.CQLColumnsForTable(ctx, user, password, table.Schema, table.Name)
	if err != nil {
		return nil, err
	}
	if metaAvailable {
		for _, col := range metaCols {
			key := strings.ToLower(col.Name)
			entry := info[key]
			entry.cqlType = col.CQLType
			typeInfo, err := e.resolveTypeInfo(ctx, user, password, col.CQLType, table.Schema)
			if err != nil {
				return nil, err
			}
			entry.typeInfo = typeInfo
			info[key] = entry
		}
	}
	return info, nil
}

func selectHasToJSON(stmt *parser.SelectStatement) bool {
	for _, expr := range stmt.Columns {
		if isToJSONExpr(expr) {
			return true
		}
	}
	return false
}

func selectHasCast(stmt *parser.SelectStatement) bool {
	for _, expr := range stmt.Columns {
		if isCastExpr(expr) {
			return true
		}
	}
	return false
}

func selectHasTimeFunction(stmt *parser.SelectStatement) bool {
	for _, expr := range stmt.Columns {
		if isTimeFunctionExpr(expr) {
			return true
		}
	}
	return false
}

func isToJSONExpr(expr parser.Expression) bool {
	if expr == nil {
		return false
	}
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		expr = aliased.Expr
	}
	call, ok := expr.(*parser.FunctionCall)
	if !ok {
		return false
	}
	return isToJSONName(call.Name)
}

func isTimeFunctionExpr(expr parser.Expression) bool {
	if expr == nil {
		return false
	}
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		expr = aliased.Expr
	}
	call, ok := expr.(*parser.FunctionCall)
	if !ok {
		return false
	}
	_, ok = timeFunctionSpecForName(call.Name)
	return ok
}

func isCastExpr(expr parser.Expression) bool {
	if expr == nil {
		return false
	}
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		expr = aliased.Expr
	}
	_, ok := expr.(*parser.CastExpr)
	return ok
}

func isToJSONName(name string) bool {
	return strings.EqualFold(name, "to_json") || strings.EqualFold(name, "tojson")
}

func buildSelectTransformFromResult(stmt *parser.SelectStatement, columns []ColumnInfo) selectTransform {
	transform := selectTransform{json: stmt.JSON}
	if stmt == nil {
		return transform
	}
	for i, expr := range stmt.Columns {
		spec, ok := toJSONSpecForResult(expr, i, columns)
		if ok {
			transform.toJSON = append(transform.toJSON, spec)
		}
		if castSpec, ok := castSpecForResult(expr, i, columns); ok {
			transform.casts = append(transform.casts, *castSpec)
		}
		if timeSpec, ok := timeFunctionCastSpecForResult(expr, i, columns); ok {
			transform.casts = append(transform.casts, *timeSpec)
		}
	}
	return transform
}

func toJSONSpecForResult(expr parser.Expression, index int, columns []ColumnInfo) (toJSONSpec, bool) {
	alias := ""
	base := expr
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		alias = aliased.Alias
		base = aliased.Expr
	}
	call, ok := base.(*parser.FunctionCall)
	if !ok || !isToJSONName(call.Name) {
		return toJSONSpec{}, false
	}
	if len(call.Args) != 1 {
		return toJSONSpec{}, false
	}

	name := alias
	if name == "" {
		name = "system.tojson(" + call.Args[0].String() + ")"
	}
	spec := toJSONSpec{
		index: index,
		name:  name,
		typ:   jsonTypeInfo{kind: jsonKindUnknown},
	}
	if index < len(columns) {
		spec.typ = jsonTypeInfoFromColumn(columns[index])
	}
	return spec, true
}

func timeFunctionCastSpecForResult(expr parser.Expression, index int, columns []ColumnInfo) (*castSpec, bool) {
	if expr == nil {
		return nil, false
	}
	alias := ""
	base := expr
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		alias = aliased.Alias
		base = aliased.Expr
	}
	call, ok := base.(*parser.FunctionCall)
	if !ok {
		return nil, false
	}
	spec, ok := timeFunctionSpecForName(call.Name)
	if !ok {
		return nil, false
	}
	if len(call.Args) != 1 {
		return nil, false
	}
	lookup := func(name string) (string, bool) {
		for _, col := range columns {
			if strings.EqualFold(col.Name, name) {
				return normalizeCastType(col.Type), true
			}
		}
		return "", false
	}
	fromType, ok := inferCastSourceType(call.Args[0], lookup)
	if !ok {
		fromType = ""
	}
	fromType = normalizeTimeFunctionSourceType(fromType)
	name := alias
	if name == "" {
		name = call.String()
	}
	resultType := timeFunctionResultType(spec)
	specOut := &castSpec{
		index:      index,
		name:       name,
		fromType:   fromType,
		toType:     resultType,
		sqlHandled: false,
		timeFunc:   &spec,
	}
	return specOut, true
}

func (e *selectPlanner) buildSelectSQLWithToJSON(stmt *parser.SelectStatement, typeInfo map[string]columnTypeInfo, limitDecision *selectLimitDecision) (string, selectTransform, error) {
	var sql strings.Builder
	transform := selectTransform{json: stmt.JSON}

	if stmt.From == nil || stmt.From.Schema == "" {
		if stmt.From == nil || stmt.From.Name == "" {
			return "", transform, fmt.Errorf("keyspace is required")
		}
		return "", transform, fmt.Errorf("keyspace is required for %s", stmt.From.Name)
	}
	if stmt.Having != nil {
		return "", transform, fmt.Errorf("HAVING is not supported")
	}

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

	for i, col := range stmt.Columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		exprSQL, spec, castSpec, err := e.translateSelectExpr(col, typeInfo)
		if err != nil {
			return "", transform, err
		}
		if spec != nil {
			spec.index = i
			transform.toJSON = append(transform.toJSON, *spec)
		}
		if castSpec != nil {
			castSpec.index = i
			transform.casts = append(transform.casts, *castSpec)
		}
		sql.WriteString(exprSQL)
	}

	sql.WriteString(" FROM ")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.From.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.From.Name))
	if stmt.IndexHint != "" {
		sql.WriteString(" FORCE INDEX (")
		sql.WriteString(e.translator.QuoteIdentifier(stmt.IndexHint))
		sql.WriteString(")")
	}

	if stmt.Where != nil {
		sql.WriteString(" WHERE ")
		sql.WriteString(e.translator.TranslateExpression(stmt.Where))
	}

	if len(stmt.GroupBy) > 0 {
		sql.WriteString(" GROUP BY ")
		for i, expr := range stmt.GroupBy {
			if i > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString(e.translator.TranslateExpression(expr))
		}
	}

	if len(stmt.Options.OrderBy) > 0 {
		sql.WriteString(" ORDER BY ")
		for i, ob := range stmt.Options.OrderBy {
			if i > 0 {
				sql.WriteString(", ")
			}
			sql.WriteString(e.translator.QuoteIdentifier(ob.Column))
			if ob.Direction == "DESC" {
				sql.WriteString(" DESC")
			} else {
				sql.WriteString(" ASC")
			}
		}
	}

	hasLimitSQL := false
	if limitDecision == nil || !limitDecision.omitLimit {
		if limitDecision != nil && limitDecision.limitOverride != nil {
			sql.WriteString(" LIMIT ")
			sql.WriteString(strconv.FormatInt(*limitDecision.limitOverride, 10))
			hasLimitSQL = true
		} else {
			limitExpr := stmt.Options.Limit
			ppExpr := stmt.Options.PerPartitionLimit
			if limitExpr != nil || ppExpr != nil {
				if limitExpr != nil && ppExpr != nil {
					return "", transform, fmt.Errorf("LIMIT and PER PARTITION LIMIT require proxy evaluation")
				}
				sql.WriteString(" LIMIT ")
				if limitExpr != nil {
					sql.WriteString(e.translator.TranslateExpression(limitExpr))
				} else {
					sql.WriteString(e.translator.TranslateExpression(ppExpr))
				}
				hasLimitSQL = true
			}
		}
	}
	omitOffset := limitDecision != nil && limitDecision.omitOffset
	if stmt.Options.Offset != nil && !omitOffset {
		if !hasLimitSQL && stmt.Options.Limit == nil && stmt.Options.PerPartitionLimit == nil {
			return "", transform, fmt.Errorf("OFFSET requires LIMIT")
		}
		sql.WriteString(" OFFSET ")
		sql.WriteString(e.translator.TranslateExpression(stmt.Options.Offset))
	}

	return sql.String(), transform, nil
}

func (e *selectPlanner) translateSelectExpr(expr parser.Expression, typeInfo map[string]columnTypeInfo) (string, *toJSONSpec, *castSpec, error) {
	if expr == nil {
		return "", nil, nil, fmt.Errorf("select expression is required")
	}

	alias := ""
	base := expr
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		alias = aliased.Alias
		base = aliased.Expr
	}

	call, ok := base.(*parser.FunctionCall)
	if ok {
		if isToJSONName(call.Name) {
			if len(call.Args) != 1 {
				return "", nil, nil, fmt.Errorf("to_json expects a single argument")
			}

			arg := call.Args[0]
			aliasName := alias
			if aliasName == "" {
				aliasName = "system.tojson(" + arg.String() + ")"
			}

			spec := &toJSONSpec{
				name: aliasName,
				typ:  jsonTypeInfo{kind: jsonKindUnknown},
			}

			colExpr, ok := arg.(*parser.ColumnExpr)
			if ok {
				colInfo, found := typeInfo[strings.ToLower(colExpr.Name)]
				if found {
					cqlType := normalizeCQLType(colInfo.cqlType)
					spec.typ = jsonTypeInfoFromCQL(cqlType)
					sqlExpr, handled := e.toJSONSQLExpression(e.translator.TranslateExpression(arg), cqlType, colInfo.mysqlType)
					spec.sqlHandled = handled
					if handled {
						return sqlExpr + " AS " + e.translator.QuoteIdentifier(aliasName), spec, nil, nil
					}
				}
			}

			argSQL := e.translator.TranslateExpression(arg)
			return argSQL + " AS " + e.translator.QuoteIdentifier(aliasName), spec, nil, nil
		}

		if spec, ok := timeFunctionSpecForName(call.Name); ok {
			if len(call.Args) != 1 {
				return "", nil, nil, fmt.Errorf("%s expects a single argument", call.Name)
			}
			arg := call.Args[0]
			lookup := func(name string) (string, bool) {
				if len(typeInfo) == 0 {
					return "", false
				}
				info, ok := typeInfo[strings.ToLower(name)]
				if !ok {
					return "", false
				}
				return normalizeCastType(info.cqlType), true
			}
			fromType, ok := inferCastSourceType(arg, lookup)
			if !ok {
				fromType = ""
			}
			fromType = normalizeTimeFunctionSourceType(fromType)
			if fromType != "" && !isTimeFunctionSourceType(spec, fromType) {
				return "", nil, nil, fmt.Errorf("%s has unsupported argument type", call.Name)
			}
			name := alias
			if name == "" {
				name = call.String()
			}
			sqlExpr := e.translator.TranslateExpression(arg)
			if alias != "" {
				sqlExpr = sqlExpr + " AS " + e.translator.QuoteIdentifier(alias)
			}
			specOut := &castSpec{
				name:       name,
				fromType:   fromType,
				toType:     timeFunctionResultType(spec),
				sqlHandled: false,
				timeFunc:   &spec,
			}
			return sqlExpr, nil, specOut, nil
		}
	}

	if castExpr, ok := base.(*parser.CastExpr); ok {
		sql, spec, err := e.translateCastExpr(castExpr, alias, typeInfo)
		return sql, nil, spec, err
	}
	return e.translator.TranslateExpression(expr), nil, nil, nil
}

func (e *selectPlanner) toJSONSQLExpression(argSQL, cqlType, mysqlType string) (string, bool) {
	if strings.EqualFold(mysqlType, "json") {
		return "CAST(" + argSQL + " AS JSON)", true
	}

	switch cqlType {
	case "int", "bigint", "smallint", "tinyint", "float", "double", "decimal", "varint", "counter", "boolean":
		return "CAST(" + argSQL + " AS JSON)", true
	case "text", "varchar", "ascii", "uuid", "inet":
		return "JSON_QUOTE(" + argSQL + ")", true
	case "timestamp", "date", "time":
		return "", false
	default:
		return "", false
	}
}

func applySelectTransform(result *QueryResult, transform selectTransform) (*QueryResult, error) {
	if result == nil {
		return result, nil
	}
	if len(transform.casts) > 0 {
		if err := applyCastTransform(result, transform.casts); err != nil {
			return nil, err
		}
	}
	if transform.json {
		if len(transform.dropIndexes) > 0 {
			applySelectDropIndexes(result, transform.dropIndexes)
		}
		return buildSelectJSONResult(result)
	}
	if len(transform.toJSON) == 0 {
		if len(transform.dropIndexes) > 0 {
			applySelectDropIndexes(result, transform.dropIndexes)
		}
		return result, nil
	}

	typeByIndex := make(map[int]jsonTypeInfo, len(transform.toJSON))
	for _, spec := range transform.toJSON {
		info := spec.typ
		if info.kind == jsonKindUnknown && spec.index < len(result.Columns) {
			info = jsonTypeInfoFromColumn(result.Columns[spec.index])
		}
		typeByIndex[spec.index] = info
	}

	for _, spec := range transform.toJSON {
		if spec.index >= len(result.Columns) {
			continue
		}
		result.Columns[spec.index].Name = spec.name
		result.Columns[spec.index].Type = "text"
		result.Columns[spec.index].TypeInfo = nil
	}

	for i := range result.Rows {
		for _, spec := range transform.toJSON {
			if spec.index >= len(result.Rows[i].Values) {
				continue
			}
			value := result.Rows[i].Values[spec.index]
			if spec.sqlHandled {
				if value == nil {
					result.Rows[i].Values[spec.index] = "null"
					continue
				}
				switch v := value.(type) {
				case []byte:
					result.Rows[i].Values[spec.index] = string(v)
				default:
					result.Rows[i].Values[spec.index] = fmt.Sprint(v)
				}
				continue
			}
			result.Rows[i].Values[spec.index] = formatJSONValue(value, typeByIndex[spec.index])
		}
	}

	if len(transform.dropIndexes) > 0 {
		applySelectDropIndexes(result, transform.dropIndexes)
	}
	return result, nil
}

func applySelectDropIndexes(result *QueryResult, dropIndexes []int) {
	if result == nil || len(dropIndexes) == 0 {
		return
	}
	drop := make(map[int]struct{}, len(dropIndexes))
	for _, idx := range dropIndexes {
		if idx >= 0 {
			drop[idx] = struct{}{}
		}
	}
	if len(drop) == 0 {
		return
	}
	filteredCols := result.Columns[:0]
	for i, col := range result.Columns {
		if _, ok := drop[i]; ok {
			continue
		}
		filteredCols = append(filteredCols, col)
	}
	result.Columns = filteredCols
	for i := range result.Rows {
		values := result.Rows[i].Values
		filtered := values[:0]
		for j, val := range values {
			if _, ok := drop[j]; ok {
				continue
			}
			filtered = append(filtered, val)
		}
		result.Rows[i].Values = filtered
	}
}

func applySelectTransformToColumns(columns []ColumnInfo, transform selectTransform) []ColumnInfo {
	if transform.json {
		return []ColumnInfo{
			{
				Name:     "json",
				Type:     "text",
				TypeInfo: nil,
				Nullable: true,
			},
		}
	}

	if len(transform.casts) == 0 && len(transform.toJSON) == 0 {
		return columns
	}

	updated := make([]ColumnInfo, len(columns))
	copy(updated, columns)
	for _, spec := range transform.casts {
		if spec.index >= len(updated) {
			continue
		}
		updated[spec.index].Name = spec.name
		updated[spec.index].Type = spec.toType
		updated[spec.index].TypeInfo = nil
	}
	for _, spec := range transform.toJSON {
		if spec.index >= len(updated) {
			continue
		}
		updated[spec.index].Name = spec.name
		updated[spec.index].Type = "text"
		updated[spec.index].TypeInfo = nil
	}
	if len(transform.dropIndexes) > 0 {
		updated = dropColumnIndexes(updated, transform.dropIndexes)
	}
	return updated
}

func dropColumnIndexes(columns []ColumnInfo, dropIndexes []int) []ColumnInfo {
	if len(columns) == 0 || len(dropIndexes) == 0 {
		return columns
	}
	drop := make(map[int]struct{}, len(dropIndexes))
	for _, idx := range dropIndexes {
		if idx >= 0 {
			drop[idx] = struct{}{}
		}
	}
	if len(drop) == 0 {
		return columns
	}
	filtered := columns[:0]
	for i, col := range columns {
		if _, ok := drop[i]; ok {
			continue
		}
		filtered = append(filtered, col)
	}
	return filtered
}