package executor

import (
	"context"
	"fmt"
	"strings"
	"time"

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

type dmlConditionKind uint8

const (
	dmlConditionNone dmlConditionKind = iota
	dmlConditionInsertIfNotExists
	dmlConditionIfExists
	dmlConditionIf
)

type dmlConditionPlan struct {
	kind             dmlConditionKind
	existsSQL        string
	existsParamOrder []int
	fetchSQL         string
	fetchParamOrder  []int
	fetchColumns     []ColumnInfo
}

type dmlPlan struct {
	sql              string
	paramOrder       []int
	condition        dmlConditionPlan
	logicalWriteRows int64
}

type equalityCondition struct {
	column  string
	value   parser.Expression
	op      string
	isNull  bool
	notNull bool
}

func (e *dmlExecutor) buildDMLPlan(ctx context.Context, user, password string, stmt parser.Statement) (*dmlPlan, error) {
	if err := e.resolveStatementTables(ctx, user, password, stmt); err != nil {
		return nil, err
	}
	switch s := stmt.(type) {
	case *parser.InsertStatement:
		if s.JSON {
			return nil, fmt.Errorf("JSON insert requires runtime planning")
		}
		return e.buildInsertPlan(ctx, user, password, s)
	case *parser.UpdateStatement:
		return e.buildUpdatePlan(ctx, user, password, s)
	case *parser.DeleteStatement:
		return e.buildDeletePlan(ctx, user, password, s)
	default:
		return nil, fmt.Errorf("unsupported DML statement: %T", stmt)
	}
}

func (e *dmlExecutor) executeDMLPlan(ctx context.Context, exec sqlExecutor, plan *dmlPlan, params []interface{}) (*QueryResult, error) {
	if plan == nil {
		return nil, fmt.Errorf("DML plan is required")
	}
	args, err := mapParams(params, plan.paramOrder)
	if err != nil {
		return nil, err
	}
	metrics.RecordRequestSQL(ctx, plan.sql, args...)
	start := time.Now()
	result, err := exec.ExecContext(ctx, plan.sql, args...)
	d := time.Since(start)
	metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
	metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpExecWrite, err == nil, d)
	if err != nil {
		return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to execute update: %w", err), mysqlOpWrite, plan.sql)
	}
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return nil, fmt.Errorf("failed to get rows affected: %w", err)
	}

	switch plan.condition.kind {
	case dmlConditionNone:
		return &QueryResult{RowsAffected: rowsAffected, CQLWriteRows: cqlWriteRowsForDMLStatement(plan, rowsAffected)}, nil
	case dmlConditionInsertIfNotExists:
		if rowsAffected > 0 {
			result := appliedResult(true, nil)
			result.RowsAffected = rowsAffected
			result.CQLWriteRows = cqlWriteRowsForDMLStatement(plan, rowsAffected)
			return result, nil
		}
		return e.fetchConditionalResult(ctx, exec, plan.condition, params)
	case dmlConditionIfExists:
		if rowsAffected > 0 {
			result := appliedResult(true, nil)
			result.RowsAffected = rowsAffected
			result.CQLWriteRows = cqlWriteRowsForDMLStatement(plan, rowsAffected)
			return result, nil
		}
		exists, err := e.checkConditionalExists(ctx, exec, plan.condition, params)
		if err != nil {
			return nil, err
		}
		return appliedResult(exists, nil), nil
	case dmlConditionIf:
		if rowsAffected > 0 {
			result := appliedResult(true, nil)
			result.RowsAffected = rowsAffected
			result.CQLWriteRows = cqlWriteRowsForDMLStatement(plan, rowsAffected)
			return result, nil
		}
		exists, err := e.checkConditionalExists(ctx, exec, plan.condition, params)
		if err != nil {
			return nil, err
		}
		if exists {
			return appliedResult(true, nil), nil
		}
		return e.fetchConditionalResult(ctx, exec, plan.condition, params)
	default:
		return nil, fmt.Errorf("unknown conditional plan")
	}
}

func cqlWriteRowsForDMLStatement(plan *dmlPlan, rowsAffected int64) int64 {
	if plan == nil {
		return 0
	}
	if plan.logicalWriteRows > 0 {
		return plan.logicalWriteRows
	}
	if rowsAffected > 0 {
		return rowsAffected
	}
	return 0
}

func (e *dmlExecutor) checkConditionalExists(ctx context.Context, exec sqlExecutor, cond dmlConditionPlan, params []interface{}) (bool, error) {
	if cond.existsSQL == "" {
		return false, nil
	}
	args, err := mapParams(params, cond.existsParamOrder)
	if err != nil {
		return false, err
	}
	result, err := e.executeSelect(ctx, exec, cond.existsSQL, args, mysqlOpWrite)
	if err != nil {
		return false, err
	}
	return len(result.Rows) > 0, nil
}

func (e *dmlExecutor) fetchConditionalResult(ctx context.Context, exec sqlExecutor, cond dmlConditionPlan, params []interface{}) (*QueryResult, error) {
	if cond.fetchSQL == "" {
		return appliedResult(false, nil), nil
	}
	args, err := mapParams(params, cond.fetchParamOrder)
	if err != nil {
		return nil, err
	}
	result, err := e.executeSelect(ctx, exec, cond.fetchSQL, args, mysqlOpWrite)
	if err != nil {
		return nil, err
	}
	if len(result.Rows) == 0 {
		values := make([]interface{}, len(cond.fetchColumns))
		return appliedResult(false, &Row{Values: values}, cond.fetchColumns), nil
	}
	return appliedResult(false, &result.Rows[0], cond.fetchColumns), nil
}

func appliedResult(applied bool, row *Row, columns ...[]ColumnInfo) *QueryResult {
	resultCols := []ColumnInfo{
		{Name: "[applied]", Type: "boolean", Nullable: true},
	}
	if len(columns) > 0 {
		resultCols = append(resultCols, columns[0]...)
	}
	values := []interface{}{applied}
	if row != nil {
		values = append(values, row.Values...)
	}
	return &QueryResult{
		Columns: resultCols,
		Rows:    []Row{{Values: values}},
	}
}

func mapParams(params []interface{}, order []int) ([]interface{}, error) {
	if len(order) == 0 {
		return nil, nil
	}
	if len(params) == 0 {
		return nil, fmt.Errorf("bind markers require prepared parameters")
	}
	args := make([]interface{}, len(order))
	for i, idx := range order {
		if idx < 0 || idx >= len(params) {
			return nil, fmt.Errorf("parameter index %d out of range", idx)
		}
		args[i] = params[idx]
	}
	return args, nil
}

func (e *dmlExecutor) lookupColumnTypesIfAvailable(ctx context.Context, user, password string, table *parser.TableName) (map[string]string, error) {
	if e.schemaManager == nil && e.poolManager == nil {
		return nil, nil
	}
	return e.lookupColumnTypes(ctx, user, password, table)
}

func (e *dmlExecutor) translateExpressionWithType(expr parser.Expression, cqlType string) string {
	if expr == nil {
		return ""
	}
	if hint, ok := expr.(*parser.TypeHintExpr); ok {
		hinted := typeStringForParam(hint.Type)
		if hinted != "" {
			return e.translateExpressionWithType(hint.Expr, hinted)
		}
		return e.translateExpressionWithType(hint.Expr, cqlType)
	}
	if normalizeCQLTypeString(cqlType) == "timestamp" {
		if lit, ok := expr.(*parser.StringLiteral); ok {
			if ts, ok := parseTimestampString(lit.Value); ok {
				return fmt.Sprintf("'%s'", formatTimestampValue(ts))
			}
		}
	}
	return e.translator.TranslateExpression(expr)
}

func (e *dmlExecutor) buildInsertPlan(ctx context.Context, user, password string, stmt *parser.InsertStatement) (*dmlPlan, error) {
	if err := requireInsertColumns(stmt); err != nil {
		return nil, err
	}
	if stmt.TTL != nil {
		return nil, fmt.Errorf("USING TTL is not supported")
	}

	columnTypes, err := e.lookupColumnTypesIfAvailable(ctx, user, password, stmt.Table)
	if err != nil {
		return nil, err
	}
	plan := &dmlPlan{}
	markerIndex := buildBindMarkerIndex(stmt)

	var sql strings.Builder
	if stmt.IfNotExists {
		if len(stmt.Values) != 1 {
			return nil, fmt.Errorf("INSERT IF NOT EXISTS does not support multiple value sets")
		}
		plan.condition.kind = dmlConditionInsertIfNotExists
		sql.WriteString("INSERT IGNORE INTO ")
	} else {
		sql.WriteString("INSERT INTO ")
	}

	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Name))

	sql.WriteString(" (")
	for i, col := range stmt.Columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(col))
	}
	sql.WriteString(") VALUES ")

	for i, values := range stmt.Values {
		if len(values) != len(stmt.Columns) {
			return nil, fmt.Errorf("VALUES count does not match column count")
		}
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString("(")
		for j, expr := range values {
			if j > 0 {
				sql.WriteString(", ")
			}
			colType := lookupColumnType(columnTypes, stmt.Columns[j])
			sql.WriteString(e.translateExpressionWithType(expr, colType))
			appendBindMarkerIndices(expr, markerIndex, &plan.paramOrder)
		}
		sql.WriteString(")")
	}

	if !stmt.IfNotExists {
		sql.WriteString(" ON DUPLICATE KEY UPDATE ")
		for i, col := range stmt.Columns {
			if i > 0 {
				sql.WriteString(", ")
			}
			quoted := e.translator.QuoteIdentifier(col)
			sql.WriteString(quoted)
			if shouldInsertDuplicateIncrementWithCoalesce(stmt, col) {
				sql.WriteString(" = COALESCE(")
				sql.WriteString(quoted)
				sql.WriteString(", 0) + VALUES(")
				sql.WriteString(quoted)
				sql.WriteString(")")
			} else if e.shouldInsertDuplicateIncrement(stmt.Table, col) {
				sql.WriteString(" = ")
				sql.WriteString(quoted)
				sql.WriteString(" + VALUES(")
				sql.WriteString(quoted)
				sql.WriteString(")")
			} else {
				sql.WriteString(" = VALUES(")
				sql.WriteString(quoted)
				sql.WriteString(")")
			}
		}
	}

	plan.sql = sql.String()
	plan.logicalWriteRows = int64(len(stmt.Values))

	if stmt.IfNotExists {
		pkCols, err := e.primaryKeyColumns(ctx, user, password, stmt.Table)
		if err != nil {
			return nil, err
		}
		fetchSQL, fetchOrder, err := e.buildFetchByPK(stmt.Table, stmt.Columns, stmt.Values[0], pkCols, markerIndex, columnTypes)
		if err != nil {
			return nil, err
		}
		fetchCols, err := e.columnInfosForTableColumns(ctx, user, password, stmt.Table, stmt.Columns)
		if err != nil {
			return nil, err
		}
		plan.condition.fetchSQL = fetchSQL
		plan.condition.fetchParamOrder = fetchOrder
		plan.condition.fetchColumns = fetchCols
	}

	return plan, nil
}

func shouldInsertDuplicateIncrementWithCoalesce(stmt *parser.InsertStatement, column string) bool {
	if stmt == nil || len(stmt.OnDuplicateIncrementColumns) == 0 {
		return false
	}
	key := strings.ToLower(strings.TrimSpace(column))
	if key == "" {
		return false
	}
	_, ok := stmt.OnDuplicateIncrementColumns[key]
	return ok
}

func (e *dmlExecutor) buildUpdatePlan(ctx context.Context, user, password string, stmt *parser.UpdateStatement) (*dmlPlan, error) {
	if stmt.TTL != nil {
		return nil, fmt.Errorf("USING TTL is not supported")
	}
	if len(stmt.Assigments) == 0 {
		return nil, fmt.Errorf("UPDATE requires assignments")
	}

	markerIndex := buildBindMarkerIndex(stmt)
	pkCols, err := e.primaryKeyColumns(ctx, user, password, stmt.Table)
	if err != nil {
		return nil, err
	}
	columnTypes, err := e.lookupColumnTypesIfAvailable(ctx, user, password, stmt.Table)
	if err != nil {
		return nil, err
	}
	counterAssignments, hasCounter, err := collectCounterAssignments(stmt.Assigments, columnTypes)
	if err != nil {
		return nil, err
	}

	whereConditions, err := extractEqualityConditions(stmt.Where)
	if err != nil {
		return nil, err
	}
	hasAllPKEq := hasAllPrimaryKeyEquality(pkCols, whereConditions)
	hasAllPKExact := hasAllPrimaryKeyExact(pkCols, whereConditions)
	hasNonPK := hasNonPrimaryKeyConditions(pkCols, whereConditions)
	hasColumnRefAssign := hasColumnRefAssignments(stmt.Assigments)

	if hasCounter {
		if stmt.IfExists || stmt.IfCondition != nil {
			return nil, fmt.Errorf("IF conditions are not supported for counter updates")
		}
		for _, assignment := range counterAssignments {
			if isColumnInList(assignment.column, pkCols) {
				return nil, fmt.Errorf("UPDATE cannot modify primary key column %q", assignment.column)
			}
		}
		return e.buildCounterUpdatePlan(stmt.Table, counterAssignments, whereConditions, markerIndex, columnTypes)
	}

	if stmt.IfExists || stmt.IfCondition != nil {
		if !hasAllPKEq || hasNonPK {
			return nil, fmt.Errorf("conditional updates require full primary key equality")
		}
		return e.buildConditionalUpdatePlan(ctx, user, password, stmt, whereConditions, markerIndex, pkCols, columnTypes)
	}

	for _, assignment := range stmt.Assigments {
		if isColumnInList(assignment.Column, pkCols) {
			return nil, fmt.Errorf("UPDATE cannot modify primary key column %q", assignment.Column)
		}
		if err := validateUpdateAssignment(assignment, columnTypes); err != nil {
			return nil, err
		}
	}

	if hasRangeConditions(whereConditions) || !hasAllPKExact || hasNonPK || (hasColumnRefAssign && !hasAllPKEq) {
		return e.buildPlainUpdatePlan(ctx, user, password, stmt, whereConditions, markerIndex, pkCols, columnTypes)
	}

	if hasInConditions(whereConditions) {
		return e.buildUpdatePlanWithIn(stmt, whereConditions, markerIndex, columnTypes)
	}

	var sql strings.Builder
	sql.WriteString("INSERT INTO ")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Name))

	columns := make([]string, 0, len(stmt.Assigments)+len(pkCols))
	for _, assignment := range stmt.Assigments {
		columns = append(columns, assignment.Column)
	}
	for _, cond := range whereConditions {
		columns = append(columns, cond.column)
	}

	sql.WriteString(" (")
	for i, col := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(col))
	}
	sql.WriteString(") VALUES (")

	plan := &dmlPlan{}
	insertExprs := make([]parser.Expression, len(stmt.Assigments))
	incrementOnDuplicate := make([]bool, len(stmt.Assigments))
	for i, assignment := range stmt.Assigments {
		insertExpr := assignment.Value
		if exprHasColumnRef(assignment.Value) {
			delta, err := arithmeticAssignmentDelta(assignment)
			if err != nil {
				return nil, err
			}
			insertExpr = delta
			incrementOnDuplicate[i] = true
		}
		insertExprs[i] = insertExpr
	}

	for i, assignment := range stmt.Assigments {
		if i > 0 {
			sql.WriteString(", ")
		}
		colType := lookupColumnType(columnTypes, assignment.Column)
		sql.WriteString(e.translateExpressionWithType(insertExprs[i], colType))
		appendBindMarkerIndices(insertExprs[i], markerIndex, &plan.paramOrder)
	}
	for i, cond := range whereConditions {
		if i > 0 || len(stmt.Assigments) > 0 {
			sql.WriteString(", ")
		}
		condType := lookupColumnType(columnTypes, cond.column)
		sql.WriteString(e.translateExpressionWithType(cond.value, condType))
		appendBindMarkerIndices(cond.value, markerIndex, &plan.paramOrder)
	}
	sql.WriteString(") ON DUPLICATE KEY UPDATE ")

	for i, assignment := range stmt.Assigments {
		if i > 0 {
			sql.WriteString(", ")
		}
		quoted := e.translator.QuoteIdentifier(assignment.Column)
		if incrementOnDuplicate[i] {
			sql.WriteString(quoted)
			sql.WriteString(" = COALESCE(")
			sql.WriteString(quoted)
			sql.WriteString(", 0) + VALUES(")
			sql.WriteString(quoted)
			sql.WriteString(")")
		} else {
			sql.WriteString(quoted)
			sql.WriteString(" = VALUES(")
			sql.WriteString(quoted)
			sql.WriteString(")")
		}
	}

	plan.sql = sql.String()
	plan.logicalWriteRows = 1
	return plan, nil
}

func hasInConditions(conditions []equalityCondition) bool {
	for _, cond := range conditions {
		if strings.EqualFold(cond.op, "IN") {
			return true
		}
	}
	return false
}

func hasRangeConditions(conditions []equalityCondition) bool {
	for _, cond := range conditions {
		if isRangeOp(cond.op) {
			return true
		}
	}
	return false
}

func hasColumnRefAssignments(assignments []*parser.Assignment) bool {
	for _, assignment := range assignments {
		if exprHasColumnRef(assignment.Value) {
			return true
		}
	}
	return false
}

func isRangeOp(op string) bool {
	switch strings.ToUpper(strings.TrimSpace(op)) {
	case "<", "<=", ">", ">=", "BETWEEN":
		return true
	default:
		return false
	}
}

func validateUpdateAssignment(assignment *parser.Assignment, columnTypes map[string]string) error {
	if assignment == nil {
		return nil
	}
	return validateArithmeticAssignment(assignment.Column, lookupColumnType(columnTypes, assignment.Column), assignment.Value)
}

func expandInConditions(conditions []equalityCondition) ([]map[string]parser.Expression, error) {
	rows := []map[string]parser.Expression{{}}
	for _, cond := range conditions {
		if strings.EqualFold(cond.op, "IN") {
			list, ok := cond.value.(*parser.ListLiteral)
			if !ok {
				return nil, fmt.Errorf("IN conditions require a list literal")
			}
			if len(list.Elements) == 0 {
				return nil, fmt.Errorf("IN conditions cannot use an empty list")
			}
			next := make([]map[string]parser.Expression, 0, len(rows)*len(list.Elements))
			for _, row := range rows {
				for _, elem := range list.Elements {
					newRow := make(map[string]parser.Expression, len(row)+1)
					for k, v := range row {
						newRow[k] = v
					}
					newRow[cond.column] = elem
					next = append(next, newRow)
				}
			}
			rows = next
			continue
		}
		for _, row := range rows {
			row[cond.column] = cond.value
		}
	}
	return rows, nil
}

func (e *dmlExecutor) buildPlainUpdatePlan(ctx context.Context, user, password string, stmt *parser.UpdateStatement, whereConditions []equalityCondition, markerIndex map[parser.Expression]int, pkCols []string, columnTypes map[string]string) (*dmlPlan, error) {
	if stmt.IfExists || stmt.IfCondition != nil {
		return e.buildConditionalUpdatePlan(ctx, user, password, stmt, whereConditions, markerIndex, pkCols, columnTypes)
	}
	var sql strings.Builder
	sql.WriteString("UPDATE ")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Name))
	sql.WriteString(" SET ")

	plan := &dmlPlan{}
	for i, assignment := range stmt.Assigments {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(assignment.Column))
		sql.WriteString(" = ")
		colType := lookupColumnType(columnTypes, assignment.Column)
		sql.WriteString(e.translateExpressionWithType(assignment.Value, colType))
		appendBindMarkerIndices(assignment.Value, markerIndex, &plan.paramOrder)
	}

	whereSQL, whereOrder := e.buildWhereClause(whereConditions, markerIndex, columnTypes)
	sql.WriteString(" WHERE ")
	sql.WriteString(whereSQL)
	plan.paramOrder = append(plan.paramOrder, whereOrder...)

	plan.sql = sql.String()
	plan.logicalWriteRows = 1
	return plan, nil
}

func (e *dmlExecutor) buildUpdatePlanWithIn(stmt *parser.UpdateStatement, whereConditions []equalityCondition, markerIndex map[parser.Expression]int, columnTypes map[string]string) (*dmlPlan, error) {
	rows, err := expandInConditions(whereConditions)
	if err != nil {
		return nil, err
	}
	if len(rows) == 0 {
		return nil, fmt.Errorf("IN conditions produced no rows")
	}

	var sql strings.Builder
	sql.WriteString("INSERT INTO ")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Name))

	columns := make([]string, 0, len(stmt.Assigments)+len(whereConditions))
	for _, assignment := range stmt.Assigments {
		columns = append(columns, assignment.Column)
	}
	for _, cond := range whereConditions {
		columns = append(columns, cond.column)
	}

	sql.WriteString(" (")
	for i, col := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(col))
	}
	sql.WriteString(") VALUES ")

	plan := &dmlPlan{}
	for rowIdx, row := range rows {
		if rowIdx > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString("(")
		for i, assignment := range stmt.Assigments {
			if i > 0 {
				sql.WriteString(", ")
			}
			colType := lookupColumnType(columnTypes, assignment.Column)
			sql.WriteString(e.translateExpressionWithType(assignment.Value, colType))
			appendBindMarkerIndices(assignment.Value, markerIndex, &plan.paramOrder)
		}
		for i, cond := range whereConditions {
			if i > 0 || len(stmt.Assigments) > 0 {
				sql.WriteString(", ")
			}
			value := cond.value
			if rowValue, ok := row[cond.column]; ok {
				value = rowValue
			}
			colType := lookupColumnType(columnTypes, cond.column)
			sql.WriteString(e.translateExpressionWithType(value, colType))
			appendBindMarkerIndices(value, markerIndex, &plan.paramOrder)
		}
		sql.WriteString(")")
	}

	sql.WriteString(" ON DUPLICATE KEY UPDATE ")
	for i, assignment := range stmt.Assigments {
		if i > 0 {
			sql.WriteString(", ")
		}
		quoted := e.translator.QuoteIdentifier(assignment.Column)
		sql.WriteString(quoted)
		sql.WriteString(" = VALUES(")
		sql.WriteString(quoted)
		sql.WriteString(")")
	}

	plan.sql = sql.String()
	plan.logicalWriteRows = int64(len(rows))
	return plan, nil
}

func (e *dmlExecutor) buildConditionalUpdatePlan(ctx context.Context, user, password string, stmt *parser.UpdateStatement, whereConditions []equalityCondition, markerIndex map[parser.Expression]int, pkCols []string, columnTypes map[string]string) (*dmlPlan, error) {
	for _, assignment := range stmt.Assigments {
		if isColumnInList(assignment.Column, pkCols) {
			return nil, fmt.Errorf("UPDATE cannot modify primary key column %q", assignment.Column)
		}
		if err := validateUpdateAssignment(assignment, columnTypes); err != nil {
			return nil, err
		}
	}

	var sql strings.Builder
	sql.WriteString("UPDATE ")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.Table.Name))
	sql.WriteString(" SET ")

	plan := &dmlPlan{}
	for i, assignment := range stmt.Assigments {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(assignment.Column))
		sql.WriteString(" = ")
		colType := lookupColumnType(columnTypes, assignment.Column)
		sql.WriteString(e.translateExpressionWithType(assignment.Value, colType))
		appendBindMarkerIndices(assignment.Value, markerIndex, &plan.paramOrder)
	}

	whereSQL, whereOrder := e.buildWhereClause(whereConditions, markerIndex, columnTypes)
	sql.WriteString(" WHERE ")
	sql.WriteString(whereSQL)
	plan.paramOrder = append(plan.paramOrder, whereOrder...)

	if stmt.IfCondition != nil {
		ifConditions, err := extractConditionalConditions(stmt.IfCondition)
		if err != nil {
			return nil, err
		}
		ifSQL, ifOrder := e.buildWhereClause(ifConditions, markerIndex, columnTypes)
		sql.WriteString(" AND ")
		sql.WriteString(ifSQL)
		plan.paramOrder = append(plan.paramOrder, ifOrder...)

		plan.condition.kind = dmlConditionIf
		plan.condition.existsSQL = e.buildExistsSQL(stmt.Table, whereConditions, ifConditions, markerIndex, &plan.condition.existsParamOrder, columnTypes)
		plan.condition.fetchSQL = e.buildFetchSQL(stmt.Table, conditionColumns(ifConditions), whereConditions, markerIndex, &plan.condition.fetchParamOrder, columnTypes)
		fetchCols, err := e.columnInfosForTableColumns(ctx, user, password, stmt.Table, conditionColumns(ifConditions))
		if err != nil {
			return nil, err
		}
		plan.condition.fetchColumns = fetchCols
	} else if stmt.IfExists {
		plan.condition.kind = dmlConditionIfExists
		plan.condition.existsSQL = e.buildExistsSQL(stmt.Table, whereConditions, nil, markerIndex, &plan.condition.existsParamOrder, columnTypes)
	}

	plan.sql = sql.String()
	plan.logicalWriteRows = 1
	return plan, nil
}

func (e *dmlExecutor) buildDeletePlan(ctx context.Context, user, password string, stmt *parser.DeleteStatement) (*dmlPlan, error) {
	assignments := deleteAssignments(stmt)
	if len(assignments) > 0 {
		updateStmt := &parser.UpdateStatement{
			Token:       stmt.Token,
			Table:       stmt.From,
			Assigments:  assignments,
			Where:       stmt.Where,
			IfExists:    stmt.IfExists,
			IfCondition: stmt.IfCondition,
		}
		return e.buildUpdatePlan(ctx, user, password, updateStmt)
	}

	markerIndex := buildBindMarkerIndex(stmt)
	columnTypes, err := e.lookupColumnTypesIfAvailable(ctx, user, password, stmt.From)
	if err != nil {
		return nil, err
	}

	whereConditions, err := extractEqualityConditions(stmt.Where)
	if err != nil {
		return nil, err
	}

	var sql strings.Builder
	sql.WriteString("DELETE FROM ")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.From.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(stmt.From.Name))

	whereSQL, whereOrder := e.buildWhereClause(whereConditions, markerIndex, columnTypes)
	sql.WriteString(" WHERE ")
	sql.WriteString(whereSQL)

	plan := &dmlPlan{
		paramOrder: whereOrder,
	}

	if stmt.IfCondition != nil {
		ifConditions, err := extractConditionalConditions(stmt.IfCondition)
		if err != nil {
			return nil, err
		}
		ifSQL, ifOrder := e.buildWhereClause(ifConditions, markerIndex, columnTypes)
		sql.WriteString(" AND ")
		sql.WriteString(ifSQL)
		plan.paramOrder = append(plan.paramOrder, ifOrder...)

		plan.condition.kind = dmlConditionIf
		plan.condition.existsSQL = e.buildExistsSQL(stmt.From, whereConditions, ifConditions, markerIndex, &plan.condition.existsParamOrder, columnTypes)
		plan.condition.fetchSQL = e.buildFetchSQL(stmt.From, conditionColumns(ifConditions), whereConditions, markerIndex, &plan.condition.fetchParamOrder, columnTypes)
		fetchCols, err := e.columnInfosForTableColumns(ctx, user, password, stmt.From, conditionColumns(ifConditions))
		if err != nil {
			return nil, err
		}
		plan.condition.fetchColumns = fetchCols
	} else if stmt.IfExists {
		plan.condition.kind = dmlConditionIfExists
		plan.condition.existsSQL = e.buildExistsSQL(stmt.From, whereConditions, nil, markerIndex, &plan.condition.existsParamOrder, columnTypes)
	}

	plan.sql = sql.String()
	return plan, nil
}

func requireInsertColumns(stmt *parser.InsertStatement) error {
	if len(stmt.Values) == 0 {
		return fmt.Errorf("VALUES clause is required")
	}
	if len(stmt.Columns) == 0 {
		return fmt.Errorf("column list is required")
	}
	return nil
}

func extractEqualityConditions(expr parser.Expression) ([]equalityCondition, error) {
	if expr == nil {
		return nil, fmt.Errorf("WHERE clause is required")
	}
	if exprContainsNullLiteral(expr) {
		return nil, fmt.Errorf("NULL literal is not allowed in WHERE clause")
	}
	switch e := expr.(type) {
	case *parser.BinaryExpr:
		if strings.EqualFold(e.Op, "AND") {
			left, err := extractEqualityConditions(e.Left)
			if err != nil {
				return nil, err
			}
			right, err := extractEqualityConditions(e.Right)
			if err != nil {
				return nil, err
			}
			return append(left, right...), nil
		}
		op := strings.ToUpper(strings.TrimSpace(e.Op))
		if op != "=" && op != "IN" && !isRangeOp(op) {
			return nil, fmt.Errorf("only '=', 'IN', or range conditions are supported")
		}
		if op == "IN" {
			col, ok := e.Left.(*parser.ColumnExpr)
			if !ok {
				return nil, fmt.Errorf("IN conditions require a column on the left")
			}
			list, ok := e.Right.(*parser.ListLiteral)
			if !ok {
				return nil, fmt.Errorf("IN conditions require a list literal")
			}
			if len(list.Elements) == 1 {
				elem := list.Elements[0]
				if exprHasColumnRef(elem) {
					return nil, fmt.Errorf("condition value cannot reference columns")
				}
				return []equalityCondition{{column: col.Name, value: elem, op: "="}}, nil
			}
			for _, elem := range list.Elements {
				if exprHasColumnRef(elem) {
					return nil, fmt.Errorf("condition value cannot reference columns")
				}
			}
			return []equalityCondition{{column: col.Name, value: e.Right, op: "IN"}}, nil
		}
		if op == "BETWEEN" {
			col, ok := e.Left.(*parser.ColumnExpr)
			if !ok {
				return nil, fmt.Errorf("BETWEEN conditions require a column on the left")
			}
			tuple, ok := e.Right.(*parser.TupleLiteral)
			if !ok || len(tuple.Elements) != 2 {
				return nil, fmt.Errorf("BETWEEN conditions require a two-element tuple")
			}
			if exprHasColumnRef(tuple.Elements[0]) || exprHasColumnRef(tuple.Elements[1]) {
				return nil, fmt.Errorf("condition value cannot reference columns")
			}
			return []equalityCondition{{column: col.Name, value: e.Right, op: "BETWEEN"}}, nil
		}
		if op != "=" {
			col, ok := e.Left.(*parser.ColumnExpr)
			if !ok {
				return nil, fmt.Errorf("range conditions require a column on the left")
			}
			if exprHasColumnRef(e.Right) {
				return nil, fmt.Errorf("condition value cannot reference columns")
			}
			return []equalityCondition{{column: col.Name, value: e.Right, op: op}}, nil
		}
		if col, ok := e.Left.(*parser.ColumnExpr); ok {
			if exprHasColumnRef(e.Right) {
				return nil, fmt.Errorf("condition value cannot reference columns")
			}
			return []equalityCondition{{column: col.Name, value: e.Right, op: "="}}, nil
		}
		if col, ok := e.Right.(*parser.ColumnExpr); ok {
			if exprHasColumnRef(e.Left) {
				return nil, fmt.Errorf("condition value cannot reference columns")
			}
			return []equalityCondition{{column: col.Name, value: e.Left, op: "="}}, nil
		}
		return nil, fmt.Errorf("condition must compare a column to a term")
	default:
		return nil, fmt.Errorf("unsupported condition expression")
	}
}

func extractConditionalConditions(expr parser.Expression) ([]equalityCondition, error) {
	if expr == nil {
		return nil, fmt.Errorf("IF clause is required")
	}
	switch e := expr.(type) {
	case *parser.BinaryExpr:
		if strings.EqualFold(e.Op, "AND") {
			left, err := extractConditionalConditions(e.Left)
			if err != nil {
				return nil, err
			}
			right, err := extractConditionalConditions(e.Right)
			if err != nil {
				return nil, err
			}
			return append(left, right...), nil
		}
		if e.Op != "=" {
			return nil, fmt.Errorf("only '=' conditions are supported")
		}
		if col, ok := e.Left.(*parser.ColumnExpr); ok {
			if isNullLiteral(e.Right) {
				return []equalityCondition{{column: col.Name, isNull: true}}, nil
			}
			if exprHasColumnRef(e.Right) {
				return nil, fmt.Errorf("condition value cannot reference columns")
			}
			return []equalityCondition{{column: col.Name, value: e.Right}}, nil
		}
		if col, ok := e.Right.(*parser.ColumnExpr); ok {
			if isNullLiteral(e.Left) {
				return []equalityCondition{{column: col.Name, isNull: true}}, nil
			}
			if exprHasColumnRef(e.Left) {
				return nil, fmt.Errorf("condition value cannot reference columns")
			}
			return []equalityCondition{{column: col.Name, value: e.Left}}, nil
		}
		return nil, fmt.Errorf("condition must compare a column to a term")
	case *parser.IsNullExpr:
		col, ok := e.Expr.(*parser.ColumnExpr)
		if !ok {
			return nil, fmt.Errorf("condition must apply to a column")
		}
		if e.Not {
			return []equalityCondition{{column: col.Name, notNull: true}}, nil
		}
		return []equalityCondition{{column: col.Name, isNull: true}}, nil
	default:
		return nil, fmt.Errorf("unsupported condition expression")
	}
}

func isNullLiteral(expr parser.Expression) bool {
	switch e := expr.(type) {
	case *parser.NullLiteral:
		return true
	case *parser.TypeHintExpr:
		return isNullLiteral(e.Expr)
	default:
		return false
	}
}

func ensurePrimaryKeyConditions(pkCols []string, conditions []equalityCondition) error {
	if len(pkCols) == 0 {
		return fmt.Errorf("primary key is required")
	}
	condMap := make(map[string]struct{}, len(conditions))
	for _, cond := range conditions {
		condMap[strings.ToLower(cond.column)] = struct{}{}
	}
	for _, pk := range pkCols {
		if _, ok := condMap[strings.ToLower(pk)]; !ok {
			return fmt.Errorf("missing primary key column %q in WHERE clause", pk)
		}
	}
	for _, cond := range conditions {
		if !isColumnInList(cond.column, pkCols) {
			return fmt.Errorf("WHERE clause must restrict only primary key columns")
		}
	}
	return nil
}

func hasAllPrimaryKeyEquality(pkCols []string, conditions []equalityCondition) bool {
	if len(pkCols) == 0 {
		return false
	}
	conds := make(map[string]equalityCondition, len(conditions))
	for _, cond := range conditions {
		conds[strings.ToLower(cond.column)] = cond
	}
	for _, pk := range pkCols {
		cond, ok := conds[strings.ToLower(pk)]
		if !ok {
			return false
		}
		if !strings.EqualFold(cond.op, "=") {
			return false
		}
	}
	return true
}

func hasAllPrimaryKeyExact(pkCols []string, conditions []equalityCondition) bool {
	if len(pkCols) == 0 {
		return false
	}
	conds := make(map[string]equalityCondition, len(conditions))
	for _, cond := range conditions {
		conds[strings.ToLower(cond.column)] = cond
	}
	for _, pk := range pkCols {
		cond, ok := conds[strings.ToLower(pk)]
		if !ok {
			return false
		}
		if !(strings.EqualFold(cond.op, "=") || strings.EqualFold(cond.op, "IN")) {
			return false
		}
	}
	return true
}

func hasNonPrimaryKeyConditions(pkCols []string, conditions []equalityCondition) bool {
	for _, cond := range conditions {
		if !isColumnInList(cond.column, pkCols) {
			return true
		}
	}
	return false
}

func isColumnInList(column string, list []string) bool {
	for _, item := range list {
		if strings.EqualFold(item, column) {
			return true
		}
	}
	return false
}

func exprHasColumnRefInAssignments(assignments []*parser.Assignment) bool {
	for _, assignment := range assignments {
		if exprHasColumnRef(assignment.Value) {
			return true
		}
	}
	return false
}

func exprHasColumnRef(expr parser.Expression) bool {
	if expr == nil {
		return false
	}
	switch e := expr.(type) {
	case *parser.ColumnExpr, *parser.CollectionAccessExpr:
		return true
	case *parser.TypeHintExpr:
		return exprHasColumnRef(e.Expr)
	case *parser.IsNullExpr:
		return exprHasColumnRef(e.Expr)
	case *parser.AliasedExpr:
		return exprHasColumnRef(e.Expr)
	case *parser.BinaryExpr:
		return exprHasColumnRef(e.Left) || exprHasColumnRef(e.Right)
	case *parser.UnaryExpr:
		return exprHasColumnRef(e.Expr)
	case *parser.FunctionCall:
		for _, arg := range e.Args {
			if exprHasColumnRef(arg) {
				return true
			}
		}
		return false
	case *parser.ListLiteral:
		for _, elem := range e.Elements {
			if exprHasColumnRef(elem) {
				return true
			}
		}
		return false
	case *parser.SetLiteral:
		for _, elem := range e.Elements {
			if exprHasColumnRef(elem) {
				return true
			}
		}
		return false
	case *parser.TupleLiteral:
		for _, elem := range e.Elements {
			if exprHasColumnRef(elem) {
				return true
			}
		}
		return false
	case *parser.VectorLiteral:
		for _, elem := range e.Elements {
			if exprHasColumnRef(elem) {
				return true
			}
		}
		return false
	case *parser.MapLiteral:
		for _, entry := range e.Entries {
			if exprHasColumnRef(entry.Key) || exprHasColumnRef(entry.Value) {
				return true
			}
		}
		return false
	default:
		return false
	}
}

func (e *dmlExecutor) primaryKeyColumns(ctx context.Context, user, password string, table *parser.TableName) ([]string, error) {
	if table == nil {
		return nil, fmt.Errorf("table is required")
	}
	if e.schemaManager == nil {
		return nil, fmt.Errorf("schema manager is not configured")
	}
	pkInfo, err := e.schemaManager.PrimaryKeyColumns(ctx, user, password, table.Schema, table.Name)
	if err != nil {
		return nil, err
	}
	columns := make([]string, 0, len(pkInfo))
	for _, pk := range pkInfo {
		columns = append(columns, pk.Column)
	}
	return columns, nil
}

func (e *dmlExecutor) buildFetchByPK(table *parser.TableName, columns []string, values []parser.Expression, pkCols []string, markerIndex map[parser.Expression]int, columnTypes map[string]string) (string, []int, error) {
	indexByColumn := make(map[string]int, len(columns))
	for i, col := range columns {
		indexByColumn[strings.ToLower(col)] = i
	}

	pkValues := make([]parser.Expression, 0, len(pkCols))
	for _, pk := range pkCols {
		idx, ok := indexByColumn[strings.ToLower(pk)]
		if !ok {
			return "", nil, fmt.Errorf("primary key column %q is missing from INSERT", pk)
		}
		pkValues = append(pkValues, values[idx])
	}

	var sql strings.Builder
	sql.WriteString("SELECT ")
	for i, col := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(col))
	}
	sql.WriteString(" FROM ")
	sql.WriteString(e.translator.QuoteIdentifier(table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(table.Name))

	whereConds := make([]equalityCondition, 0, len(pkCols))
	for i, pk := range pkCols {
		whereConds = append(whereConds, equalityCondition{column: pk, value: pkValues[i]})
	}
	whereSQL, whereOrder := e.buildWhereClause(whereConds, markerIndex, columnTypes)
	sql.WriteString(" WHERE ")
	sql.WriteString(whereSQL)

	return sql.String(), whereOrder, nil
}

func (e *dmlExecutor) buildWhereClause(conditions []equalityCondition, markerIndex map[parser.Expression]int, columnTypes map[string]string) (string, []int) {
	var sql strings.Builder
	var order []int
	for i, cond := range conditions {
		if i > 0 {
			sql.WriteString(" AND ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(cond.column))
		if cond.notNull {
			sql.WriteString(" IS NOT NULL")
			continue
		}
		if cond.isNull || isNullLiteral(cond.value) {
			sql.WriteString(" IS NULL")
			continue
		}
		colType := lookupColumnType(columnTypes, cond.column)
		if strings.EqualFold(cond.op, "IN") {
			list, ok := cond.value.(*parser.ListLiteral)
			if !ok {
				sql.WriteString(" IN ")
				sql.WriteString(e.translateExpressionWithType(cond.value, colType))
				appendBindMarkerIndices(cond.value, markerIndex, &order)
				continue
			}
			if len(list.Elements) == 0 {
				sql.WriteString(" IN (NULL)")
				continue
			}
			sql.WriteString(" IN (")
			for i, elem := range list.Elements {
				if i > 0 {
					sql.WriteString(", ")
				}
				sql.WriteString(e.translateExpressionWithType(elem, colType))
				appendBindMarkerIndices(elem, markerIndex, &order)
			}
			sql.WriteString(")")
			continue
		}
		if strings.EqualFold(cond.op, "BETWEEN") {
			tuple, ok := cond.value.(*parser.TupleLiteral)
			if !ok || len(tuple.Elements) != 2 {
				sql.WriteString(" BETWEEN ")
				sql.WriteString(e.translateExpressionWithType(cond.value, colType))
				appendBindMarkerIndices(cond.value, markerIndex, &order)
				continue
			}
			sql.WriteString(" BETWEEN ")
			sql.WriteString(e.translateExpressionWithType(tuple.Elements[0], colType))
			sql.WriteString(" AND ")
			sql.WriteString(e.translateExpressionWithType(tuple.Elements[1], colType))
			appendBindMarkerIndices(tuple.Elements[0], markerIndex, &order)
			appendBindMarkerIndices(tuple.Elements[1], markerIndex, &order)
			continue
		}
		if isRangeOp(cond.op) {
			sql.WriteString(" ")
			sql.WriteString(cond.op)
			sql.WriteString(" ")
			sql.WriteString(e.translateExpressionWithType(cond.value, colType))
			appendBindMarkerIndices(cond.value, markerIndex, &order)
			continue
		}
		sql.WriteString(" = ")
		sql.WriteString(e.translateExpressionWithType(cond.value, colType))
		appendBindMarkerIndices(cond.value, markerIndex, &order)
	}
	return sql.String(), order
}

func (e *dmlExecutor) buildExistsSQL(table *parser.TableName, pkConditions, extra []equalityCondition, markerIndex map[parser.Expression]int, order *[]int, columnTypes map[string]string) string {
	if table == nil {
		return ""
	}
	var sql strings.Builder
	sql.WriteString("SELECT 1 FROM ")
	sql.WriteString(e.translator.QuoteIdentifier(table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(table.Name))

	whereConditions := make([]equalityCondition, 0, len(pkConditions)+len(extra))
	whereConditions = append(whereConditions, pkConditions...)
	whereConditions = append(whereConditions, extra...)
	whereSQL, whereOrder := e.buildWhereClause(whereConditions, markerIndex, columnTypes)
	sql.WriteString(" WHERE ")
	sql.WriteString(whereSQL)
	*order = whereOrder
	return sql.String()
}

func (e *dmlExecutor) buildFetchSQL(table *parser.TableName, columns []string, whereConditions []equalityCondition, markerIndex map[parser.Expression]int, order *[]int, columnTypes map[string]string) string {
	if table == nil {
		return ""
	}
	var sql strings.Builder
	sql.WriteString("SELECT ")
	for i, col := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(e.translator.QuoteIdentifier(col))
	}
	sql.WriteString(" FROM ")
	sql.WriteString(e.translator.QuoteIdentifier(table.Schema))
	sql.WriteString(".")
	sql.WriteString(e.translator.QuoteIdentifier(table.Name))

	whereSQL, whereOrder := e.buildWhereClause(whereConditions, markerIndex, columnTypes)
	sql.WriteString(" WHERE ")
	sql.WriteString(whereSQL)
	*order = whereOrder
	return sql.String()
}

func conditionColumns(conditions []equalityCondition) []string {
	seen := make(map[string]struct{}, len(conditions))
	cols := make([]string, 0, len(conditions))
	for _, cond := range conditions {
		name := strings.ToLower(cond.column)
		if _, ok := seen[name]; ok {
			continue
		}
		seen[name] = struct{}{}
		cols = append(cols, cond.column)
	}
	return cols
}

func (e *dmlExecutor) columnInfosForTableColumns(ctx context.Context, user, password string, table *parser.TableName, columns []string) ([]ColumnInfo, error) {
	if len(columns) == 0 {
		return nil, nil
	}
	columnTypes, err := e.lookupColumnTypes(ctx, user, password, table)
	if err != nil {
		return nil, err
	}
	infos := make([]ColumnInfo, 0, len(columns))
	for _, name := range columns {
		typ := lookupColumnType(columnTypes, name)
		typeInfo, err := e.resolveTypeInfo(ctx, user, password, typ, table.Schema)
		if err != nil {
			return nil, err
		}
		infos = append(infos, ColumnInfo{
			Keyspace: table.Schema,
			Table:    table.Name,
			Name:     name,
			Type:     typ,
			TypeInfo: typeInfo,
			Nullable: true,
		})
	}
	return infos, nil
}

func buildBindMarkerIndex(stmt parser.Statement) map[parser.Expression]int {
	markerIndex := make(map[parser.Expression]int)
	index := 0
	assignBindMarkers(stmt, markerIndex, &index)
	return markerIndex
}

func assignBindMarkers(stmt parser.Statement, markerIndex map[parser.Expression]int, index *int) {
	if stmt == nil {
		return
	}
	switch s := stmt.(type) {
	case *parser.InsertStatement:
		if s.JSON {
			assignBindMarkersExpr(s.JSONValue, markerIndex, index)
		} else {
			for _, row := range s.Values {
				for _, expr := range row {
					assignBindMarkersExpr(expr, markerIndex, index)
				}
			}
		}
		if s.Timestamp != nil {
			assignBindMarkersExpr(s.Timestamp, markerIndex, index)
		}
		if s.TTL != nil {
			assignBindMarkersExpr(s.TTL, markerIndex, index)
		}
	case *parser.UpdateStatement:
		if s.TTL != nil {
			assignBindMarkersExpr(s.TTL, markerIndex, index)
		}
		if s.Timestamp != nil {
			assignBindMarkersExpr(s.Timestamp, markerIndex, index)
		}
		for _, assignment := range s.Assigments {
			assignBindMarkersExpr(assignment.Target, markerIndex, index)
			assignBindMarkersExpr(assignment.Value, markerIndex, index)
		}
		assignBindMarkersExpr(s.Where, markerIndex, index)
		if s.IfCondition != nil {
			assignBindMarkersExpr(s.IfCondition, markerIndex, index)
		}
	case *parser.DeleteStatement:
		if s.Timestamp != nil {
			assignBindMarkersExpr(s.Timestamp, markerIndex, index)
		}
		for _, target := range s.Targets {
			assignBindMarkersExpr(target, markerIndex, index)
		}
		assignBindMarkersExpr(s.Where, markerIndex, index)
		if s.IfCondition != nil {
			assignBindMarkersExpr(s.IfCondition, markerIndex, index)
		}
	case *parser.BatchStatement:
		if s.Using != nil {
			assignBindMarkersExpr(s.Using, markerIndex, index)
		}
		for _, entry := range s.Entries {
			assignBindMarkers(entry, markerIndex, index)
		}
	case *parser.SelectStatement:
		for _, col := range s.Columns {
			assignBindMarkersExpr(col, markerIndex, index)
		}
		assignBindMarkersExpr(s.Where, markerIndex, index)
		assignBindMarkersExpr(s.Having, markerIndex, index)
		for _, expr := range s.GroupBy {
			assignBindMarkersExpr(expr, markerIndex, index)
		}
		assignBindMarkersExpr(s.Options.Limit, markerIndex, index)
		assignBindMarkersExpr(s.Options.PerPartitionLimit, markerIndex, index)
	}
}

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

func appendBindMarkerIndices(expr parser.Expression, markerIndex map[parser.Expression]int, order *[]int) {
	if expr == nil {
		return
	}
	switch e := expr.(type) {
	case *parser.BindMarker, *parser.NamedBindMarker:
		if idx, ok := markerIndex[expr]; ok {
			*order = append(*order, idx)
		}
	case *parser.TypeHintExpr:
		appendBindMarkerIndices(e.Expr, markerIndex, order)
	case *parser.CastExpr:
		appendBindMarkerIndices(e.Expr, markerIndex, order)
	case *parser.IsNullExpr:
		appendBindMarkerIndices(e.Expr, markerIndex, order)
	case *parser.AliasedExpr:
		appendBindMarkerIndices(e.Expr, markerIndex, order)
	case *parser.BinaryExpr:
		appendBindMarkerIndices(e.Left, markerIndex, order)
		appendBindMarkerIndices(e.Right, markerIndex, order)
	case *parser.UnaryExpr:
		appendBindMarkerIndices(e.Expr, markerIndex, order)
	case *parser.FunctionCall:
		for _, arg := range e.Args {
			appendBindMarkerIndices(arg, markerIndex, order)
		}
	case *parser.CollectionAccessExpr:
		appendBindMarkerIndices(e.Target, markerIndex, order)
		appendBindMarkerIndices(e.Start, markerIndex, order)
		appendBindMarkerIndices(e.End, markerIndex, order)
	case *parser.ListLiteral:
		for _, elem := range e.Elements {
			appendBindMarkerIndices(elem, markerIndex, order)
		}
	case *parser.SetLiteral:
		for _, elem := range e.Elements {
			appendBindMarkerIndices(elem, markerIndex, order)
		}
	case *parser.TupleLiteral:
		for _, elem := range e.Elements {
			appendBindMarkerIndices(elem, markerIndex, order)
		}
	case *parser.VectorLiteral:
		for _, elem := range e.Elements {
			appendBindMarkerIndices(elem, markerIndex, order)
		}
	case *parser.MapLiteral:
		for _, entry := range e.Entries {
			appendBindMarkerIndices(entry.Key, markerIndex, order)
			appendBindMarkerIndices(entry.Value, markerIndex, order)
		}
	}
}

func countBindMarkersInStatement(stmt parser.Statement) int {
	return len(buildBindMarkerIndex(stmt))
}

func isDMLStatement(stmt parser.Statement) bool {
	switch stmt.(type) {
	case *parser.InsertStatement, *parser.UpdateStatement, *parser.DeleteStatement, *parser.BatchStatement:
		return true
	default:
		return false
	}
}

func isUpdateOrDelete(stmt parser.Statement) bool {
	switch stmt.(type) {
	case *parser.UpdateStatement, *parser.DeleteStatement:
		return true
	default:
		return false
	}
}

func conditionalResultColumns(cond dmlConditionPlan) []ColumnInfo {
	if cond.kind == dmlConditionNone {
		return nil
	}
	cols := []ColumnInfo{{Name: "[applied]", Type: "boolean", Nullable: true}}
	if len(cond.fetchColumns) > 0 {
		cols = append(cols, cond.fetchColumns...)
	}
	return cols
}