package executor

import (
	"context"
	"database/sql"
	"fmt"
	"math"
	"runtime"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"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/metrics"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
	"go.uber.org/zap"
)

// BatchEntry groups a DML statement with its decoded parameters for batch execution.
type BatchEntry struct {
	Statement     parser.Statement
	Params        []interface{}
	originals     []BatchEntry
	mergeKind     batchMergeKind
	folded        bool
	incrementCols map[string]struct{}
}

type batchBestEffortWorkItem struct {
	entry    BatchEntry
	fallback []BatchEntry
}

type batchPlanningEnvironment struct {
	resolvePK        batchUpdatePKResolver
	resolveSplitKeys batchSplitKeyResolver
	foldEnv          batchFoldEnvironment
}

type batchBestEffortPlanningDebug struct {
	globalFoldPlannerEnabled bool
	globalFoldPlannerApplied bool
	globalFoldFallbackReason string
}

type batchGlobalFoldPlannerDecision struct {
	Applied        bool
	FallbackReason string
}

type batchGroupDebugInfo struct {
	templateKey string
	breakReason string
}

type coalescedBatchGroup struct {
	merged           BatchEntry
	originals        []BatchEntry
	groupingStrategy batchGroupingStrategy
	partitionHits    []string
	debug            batchGroupDebugInfo
}

type batchExecutionMode uint8

type batchGroupingStrategy string

const (
	batchExecutionTransactional batchExecutionMode = iota
	batchExecutionBestEffortParallel
)

const (
	batchGroupingStrategyPrimaryKey     batchGroupingStrategy = "primary_key"
	batchGroupingStrategyPartitionGroup batchGroupingStrategy = "partition_group"
)

const (
	batchGlobalFoldFallbackDisabled                = "disabled"
	batchGlobalFoldFallbackBarrierStatement        = "barrier_statement"
	batchGlobalFoldFallbackNonFoldableStatement    = "non_foldable_statement"
	batchGlobalFoldFallbackMixedStatementKinds     = "mixed_statement_kinds"
	batchGlobalFoldFallbackFoldConflict            = "fold_conflict"
	batchGlobalFoldFallbackFoldedEntryNotMergeable = "folded_entry_not_mergeable"
	batchGlobalFoldFallbackFoldedKindMismatch      = "folded_kind_mismatch"
)

const (
	batchGroupBreakReasonSingleStatement                  = "single_statement"
	batchGroupBreakReasonPlannerDisabled                  = "planner_disabled"
	batchGroupBreakReasonEndOfEntries                     = "end_of_entries"
	batchGroupBreakReasonNonMergeableStatement            = "non_mergeable_statement"
	batchGroupBreakReasonCandidateBuildError              = "candidate_build_error"
	batchGroupBreakReasonMergeKindChange                  = "merge_kind_change"
	batchGroupBreakReasonTemplateChange                   = "template_change"
	batchGroupBreakReasonPartitionChange                  = "partition_change"
	batchGroupBreakReasonLimitRows                        = "limit_rows"
	batchGroupBreakReasonLimitBytes                       = "limit_bytes"
	batchGroupBreakReasonLimitRowsAndBytes                = "limit_rows_and_bytes"
	batchGroupBreakReasonSafeBoundaryAfterLimitRows       = "safe_boundary_after_limit_rows"
	batchGroupBreakReasonSafeBoundaryAfterLimitBytes      = "safe_boundary_after_limit_bytes"
	batchGroupBreakReasonSafeBoundaryAfterLimitRowsBytes  = "safe_boundary_after_limit_rows_and_bytes"
	batchGroupBreakReasonDuplicateFoldSingletonUnabsorbed = "duplicate_fold_singleton_unabsorbed"
)

const (
	// Cap merged INSERT row count to avoid oversized single SQL statements.
	defaultBatchInsertMergeMaxRows = 256
	// Cap merged INSERT estimated SQL size (best-effort) to keep statement size bounded.
	defaultBatchInsertMergeMaxBytes = 256 * 1024
)

func (e *Executor) UnloggedBatchParallelEnabled() bool {
	return e != nil && e.unloggedBatchParallel
}

func (m batchExecutionMode) String() string {
	switch m {
	case batchExecutionBestEffortParallel:
		return "best_effort_parallel"
	default:
		return "transactional"
	}
}

func (e *Executor) logBatchExecutionPlan(batchType string, mode batchExecutionMode, statementCount, executePlanCount, mergedGroupCount, mergedStmtCount int) {
	if e == nil || e.logger == nil {
		return
	}

	fields := []zap.Field{
		zap.String("batch_type", batchType),
		zap.String("batch_execution_mode", mode.String()),
		zap.Int("batch_input_statement_count", statementCount),
		zap.Int("batch_execute_plan_count", executePlanCount),
		zap.Int("batch_merged_group_count", mergedGroupCount),
		zap.Int("batch_merged_statement_count", mergedStmtCount),
		zap.Bool("batch_merge_applied", mergedGroupCount > 0),
	}
	if mergedGroupCount > 0 || statementCount != executePlanCount {
		e.logger.Info("batch merge planned", fields...)
		return
	}
	e.logger.Debug("batch execution planned", fields...)
}

// ExecuteBatch executes a batch of DML statements.
// UNLOGGED batches may run in best-effort parallel mode when enabled by config.
func (e *Executor) ExecuteBatch(ctx context.Context, user, password, batchType string, entries []BatchEntry) (*QueryResult, error) {
	if len(entries) == 0 {
		return &QueryResult{}, nil
	}
	// Batch requests are not deadlock-retried; flag the context so statements
	// executed under it skip retry.
	ctx = markBatchContext(ctx)
	metrics.SetRequestBatchInfo(ctx, batchType, len(entries), len(entries), 0, 0)
	requestWriteRows := cqlWriteRowsForBatchEntries(entries)

	if state := keyspaceStateFromContext(ctx); state != nil {
		keyspace := state.GetKeyspace()
		if keyspace != "" {
			for i := range entries {
				applyDefaultKeyspace(entries[i].Statement, keyspace)
			}
		}
	}

	for _, entry := range entries {
		if err := validateBatchEntry(entry.Statement); err != nil {
			return nil, err
		}
		if err := e.validateDML(ctx, user, password, entry.Statement); err != nil {
			return nil, err
		}
	}

	mode, workers := e.batchExecutionPlan(batchType)
	allowUpdateMerge := e.batchUpdateMergeEnabled()
	if mode == batchExecutionBestEffortParallel {
		mergePlanStart := time.Now()
		workItems := make([]batchBestEffortWorkItem, 0, len(entries))
		plannedWorkers := workers
		var planningDebug *batchBestEffortPlanningDebug
		if e.unloggedMergeBeforeParallel {
			mergeRows, mergeBytes := e.batchInsertMergeLimits()
			groups, serialFallback, debug, err := e.coalesceBatchGroupsForBestEffortExecution(ctx, user, password, entries, mergeRows, mergeBytes, allowUpdateMerge)
			if err != nil {
				metrics.ObserveStageFromContext(ctx, metrics.StageBatchMergePlan, time.Since(mergePlanStart))
				return nil, err
			}
			planningDebug = &debug
			if serialFallback && plannedWorkers > 1 {
				plannedWorkers = 1
			}
			workItems = buildBestEffortBatchWorkItemsFromGroups(groups)
			e.logBatchPrimaryKeys(ctx, user, password, batchType, groups, planningDebug)
		} else {
			planningDebug = &batchBestEffortPlanningDebug{
				globalFoldPlannerEnabled: false,
				globalFoldFallbackReason: batchGlobalFoldFallbackDisabled,
			}
			if plannedWorkers > 1 {
				plannedWorkers = 1
			}
			if e.logger != nil && e.logger.Core().Enabled(zap.DebugLevel) {
				groups := make([]coalescedBatchGroup, 0, len(entries))
				for _, entry := range entries {
					groups = append(groups, newPassthroughBatchGroup(entry, batchGroupBreakReasonPlannerDisabled))
				}
				e.logBatchPrimaryKeys(ctx, user, password, batchType, groups, planningDebug)
			}
			for _, entry := range entries {
				workItems = append(workItems, batchBestEffortWorkItem{entry: entry})
			}
		}
		executeItemCount, mergedGroupCount, mergedStmtCount := summarizeBatchWorkItems(workItems)
		metrics.SetRequestBatchInfo(ctx, batchType, len(entries), executeItemCount, mergedGroupCount, mergedStmtCount)
		e.logBatchExecutionPlan(batchType, mode, len(entries), executeItemCount, mergedGroupCount, mergedStmtCount)
		metrics.ObserveStageFromContext(ctx, metrics.StageBatchMergePlan, time.Since(mergePlanStart))
		executeStart := time.Now()
		result, err := e.executeBatchBestEffortParallel(ctx, user, password, workItems, plannedWorkers)
		metrics.ObserveStageFromContext(ctx, metrics.StageBatchExecuteWait, time.Since(executeStart))
		if err == nil && result != nil {
			result.CQLWriteRows = requestWriteRows
		}
		return result, err
	}

	mergePlanStart := time.Now()
	mergeRows, mergeBytes := e.batchInsertMergeLimits()
	groups, err := e.coalesceBatchGroupsForExecution(ctx, user, password, entries, mergeRows, mergeBytes, allowUpdateMerge)
	if err != nil {
		metrics.ObserveStageFromContext(ctx, metrics.StageBatchMergePlan, time.Since(mergePlanStart))
		return nil, err
	}
	executeItemCount, mergedGroupCount, mergedStmtCount := summarizeBatchGroups(groups)
	metrics.SetRequestBatchInfo(ctx, batchType, len(entries), executeItemCount, mergedGroupCount, mergedStmtCount)
	e.logBatchPrimaryKeys(ctx, user, password, batchType, groups, nil)
	e.logBatchExecutionPlan(batchType, mode, len(entries), executeItemCount, mergedGroupCount, mergedStmtCount)
	entries = batchEntriesFromGroups(groups)
	metrics.ObserveStageFromContext(ctx, metrics.StageBatchMergePlan, time.Since(mergePlanStart))
	executeStart := time.Now()
	result, err := e.executeBatchTransactional(ctx, user, password, entries)
	metrics.ObserveStageFromContext(ctx, metrics.StageBatchExecuteWait, time.Since(executeStart))
	if err == nil && result != nil {
		result.CQLWriteRows = requestWriteRows
	}
	return result, err
}

func cqlWriteRowsForBatchEntries(entries []BatchEntry) int64 {
	var rows int64
	for _, entry := range entries {
		rows += cqlWriteRowsForBatchEntry(entry)
	}
	return rows
}

func cqlWriteRowsForBatchEntry(entry BatchEntry) int64 {
	switch stmt := entry.Statement.(type) {
	case *parser.InsertStatement:
		if stmt == nil {
			return 0
		}
		if stmt.JSON {
			return 1
		}
		if len(stmt.Values) > 0 {
			return int64(len(stmt.Values))
		}
		return 1
	case *parser.UpdateStatement, *parser.DeleteStatement:
		return 1
	default:
		return 0
	}
}

func summarizeBatchGroups(groups []coalescedBatchGroup) (executeItemCount, mergedGroupCount, mergedStmtCount int) {
	executeItemCount = len(groups)
	for _, group := range groups {
		if len(group.originals) > 1 {
			mergedGroupCount++
			mergedStmtCount += len(group.originals)
		}
	}
	return executeItemCount, mergedGroupCount, mergedStmtCount
}

func summarizeBatchWorkItems(items []batchBestEffortWorkItem) (executeItemCount, mergedGroupCount, mergedStmtCount int) {
	executeItemCount = len(items)
	for _, item := range items {
		if len(item.fallback) > 1 {
			mergedGroupCount++
			mergedStmtCount += len(item.fallback)
		}
	}
	return executeItemCount, mergedGroupCount, mergedStmtCount
}

func coalesceBatchInsertEntries(entries []BatchEntry, maxRows, maxBytes int) []BatchEntry {
	groups := coalesceBatchInsertGroups(entries, maxRows, maxBytes)
	if len(groups) == 0 {
		return nil
	}
	out := make([]BatchEntry, 0, len(groups))
	for _, group := range groups {
		out = append(out, group.merged)
	}
	return out
}

func buildBestEffortBatchWorkItems(entries []BatchEntry, maxRows, maxBytes int) []batchBestEffortWorkItem {
	groups := coalesceBatchInsertGroups(entries, maxRows, maxBytes)
	if len(groups) == 0 {
		return nil
	}
	items := make([]batchBestEffortWorkItem, 0, len(groups))
	for _, group := range groups {
		item := batchBestEffortWorkItem{entry: group.merged}
		if len(group.originals) > 1 {
			item.fallback = append([]BatchEntry(nil), group.originals...)
		}
		items = append(items, item)
	}
	return items
}

func batchEntriesFromGroups(groups []coalescedBatchGroup) []BatchEntry {
	if len(groups) == 0 {
		return nil
	}
	out := make([]BatchEntry, 0, len(groups))
	for _, group := range groups {
		out = append(out, group.merged)
	}
	return out
}

func shallowBatchEntry(entry BatchEntry) BatchEntry {
	return BatchEntry{
		Statement:     entry.Statement,
		Params:        append([]interface{}(nil), entry.Params...),
		mergeKind:     entry.mergeKind,
		folded:        entry.folded,
		incrementCols: cloneBatchColumnSet(entry.incrementCols),
	}
}

func batchEntryOriginals(entry BatchEntry) []BatchEntry {
	if len(entry.originals) > 0 {
		return append([]BatchEntry(nil), entry.originals...)
	}
	return []BatchEntry{shallowBatchEntry(entry)}
}

func buildBestEffortBatchWorkItemsFromGroups(groups []coalescedBatchGroup) []batchBestEffortWorkItem {
	if len(groups) == 0 {
		return nil
	}
	items := make([]batchBestEffortWorkItem, 0, len(groups))
	for _, group := range groups {
		item := batchBestEffortWorkItem{entry: group.merged}
		if len(group.originals) > 1 {
			item.fallback = append([]BatchEntry(nil), group.originals...)
		}
		items = append(items, item)
	}
	return items
}

type batchUpdatePKResolver func(table *parser.TableName) ([]string, error)
type batchSplitKeyResolver func(entry BatchEntry, insert *parser.InsertStatement, params []interface{}) []string

type batchMergeKind uint8

const (
	batchMergeKindInsert batchMergeKind = iota
	batchMergeKindUpdate
)

type batchMergeCandidate struct {
	kind          batchMergeKind
	merged        BatchEntry
	insert        *parser.InsertStatement
	originals     []BatchEntry
	rows          int
	bytes         int
	pkKeys        []string
	splitKeys     []string
	incrementCols map[string]struct{}
}

type batchMergeSegment struct {
	candidates       []batchMergeCandidate
	groupingStrategy batchGroupingStrategy
	partitionHits    []string
	debug            batchGroupDebugInfo
}

func (e *Executor) newBatchPlanningEnvironment(ctx context.Context, user, password string) batchPlanningEnvironment {
	pkCache := make(map[batchTableKey][]string)
	columnOrderCache := make(map[batchTableKey][]string)
	columnTypesCache := make(map[batchTableKey]map[string]string)

	resolvePK := func(table *parser.TableName) ([]string, error) {
		if table == nil {
			return nil, fmt.Errorf("table is required")
		}
		key := batchTableKey{schema: table.Schema, table: table.Name}
		if cols, ok := pkCache[key]; ok {
			return cols, nil
		}
		cols, err := e.primaryKeyColumns(ctx, user, password, table)
		if err != nil {
			return nil, err
		}
		pkCache[key] = cols
		return cols, nil
	}
	resolveColumnOrder := func(table *parser.TableName) ([]string, error) {
		if table == nil {
			return nil, fmt.Errorf("table is required")
		}
		key := batchTableKey{schema: table.Schema, table: table.Name}
		if cols, ok := columnOrderCache[key]; ok {
			return cols, nil
		}
		if e == nil || e.schemaManager == nil {
			return nil, nil
		}
		columnInfos, err := e.schemaManager.ColumnsForTable(ctx, user, password, table.Schema, table.Name)
		if err != nil {
			return nil, err
		}
		cols := make([]string, 0, len(columnInfos))
		for _, info := range columnInfos {
			cols = append(cols, info.Name)
		}
		columnOrderCache[key] = cols
		return cols, nil
	}
	lookupColumnTypes := func(table *parser.TableName) (map[string]string, error) {
		if table == nil {
			return nil, fmt.Errorf("table is required")
		}
		key := batchTableKey{schema: table.Schema, table: table.Name}
		if types, ok := columnTypesCache[key]; ok {
			return types, nil
		}
		if e != nil {
			e.ensureInternals()
		}
		var (
			types map[string]string
			err   error
		)
		if e != nil && e.dml != nil {
			types, err = e.dml.lookupColumnTypesIfAvailable(ctx, user, password, table)
			if err != nil {
				return nil, err
			}
		}
		columnTypesCache[key] = types
		return types, nil
	}
	resolveSplitKeys := func(entry BatchEntry, insert *parser.InsertStatement, params []interface{}) []string {
		if !e.batchGroupByTiDBHashPartitionEnabled() {
			return nil
		}
		return e.batchHashPartitionSplitKeys(ctx, user, password, insert, params)
	}
	return batchPlanningEnvironment{
		resolvePK:        resolvePK,
		resolveSplitKeys: resolveSplitKeys,
		foldEnv: batchFoldEnvironment{
			resolvePK:          resolvePK,
			resolveColumnOrder: resolveColumnOrder,
			lookupColumnTypes:  lookupColumnTypes,
			shouldInsertIncrement: func(table *parser.TableName, column string) bool {
				if e == nil {
					return false
				}
				return e.shouldInsertDuplicateIncrement(table, column)
			},
		},
	}
}

func (e *Executor) coalesceBatchGroupsForExecution(ctx context.Context, user, password string, entries []BatchEntry, maxRows, maxBytes int, allowUpdateMerge bool) ([]coalescedBatchGroup, error) {
	if len(entries) == 0 {
		return nil, nil
	}
	planning := e.newBatchPlanningEnvironment(ctx, user, password)
	return coalesceBatchGroupsForExecutionWithPlanningEnvironment(entries, maxRows, maxBytes, allowUpdateMerge, planning)
}

func coalesceBatchGroupsForExecutionWithPlanningEnvironment(entries []BatchEntry, maxRows, maxBytes int, allowUpdateMerge bool, planning batchPlanningEnvironment) ([]coalescedBatchGroup, error) {
	foldedEntries, err := foldBatchEntriesWithEnv(entries, allowUpdateMerge, planning.foldEnv)
	if err != nil {
		return nil, err
	}
	if !allowUpdateMerge {
		return coalesceBatchInsertGroupsWithResolver(foldedEntries, maxRows, maxBytes, planning.resolvePK, planning.resolveSplitKeys), nil
	}
	return coalesceBatchGroupsWithUpdateMerge(foldedEntries, maxRows, maxBytes, planning.resolvePK, planning.resolveSplitKeys)
}

func (e *Executor) coalesceBatchGroupsForBestEffortExecution(ctx context.Context, user, password string, entries []BatchEntry, maxRows, maxBytes int, allowUpdateMerge bool) ([]coalescedBatchGroup, bool, batchBestEffortPlanningDebug, error) {
	debug := batchBestEffortPlanningDebug{globalFoldPlannerEnabled: true}
	if len(entries) == 0 {
		return nil, false, debug, nil
	}
	planning := e.newBatchPlanningEnvironment(ctx, user, password)
	groups, decision, err := tryCoalesceHomogeneousUnloggedBatchGroupsWithDebug(entries, maxRows, maxBytes, allowUpdateMerge, planning)
	if err != nil {
		return nil, false, debug, err
	}
	if decision.Applied {
		debug.globalFoldPlannerApplied = true
		return groups, false, debug, nil
	}
	debug.globalFoldFallbackReason = decision.FallbackReason
	groups, err = coalesceBatchGroupsForExecutionWithPlanningEnvironment(entries, maxRows, maxBytes, allowUpdateMerge, planning)
	if err != nil {
		return nil, false, debug, err
	}
	return groups, true, debug, nil
}

func tryCoalesceHomogeneousUnloggedBatchGroupsWithPlanningEnvironment(entries []BatchEntry, maxRows, maxBytes int, allowUpdateMerge bool, planning batchPlanningEnvironment) ([]coalescedBatchGroup, bool, error) {
	groups, decision, err := tryCoalesceHomogeneousUnloggedBatchGroupsWithDebug(entries, maxRows, maxBytes, allowUpdateMerge, planning)
	if err != nil {
		return nil, false, err
	}
	return groups, decision.Applied, nil
}

func tryCoalesceHomogeneousUnloggedBatchGroupsWithDebug(entries []BatchEntry, maxRows, maxBytes int, allowUpdateMerge bool, planning batchPlanningEnvironment) ([]coalescedBatchGroup, batchGlobalFoldPlannerDecision, error) {
	foldedEntries, expectedKind, ok, reason, err := foldBatchEntriesGloballyWithEnv(entries, allowUpdateMerge, planning.foldEnv)
	if err != nil {
		return nil, batchGlobalFoldPlannerDecision{}, err
	}
	if !ok {
		return nil, batchGlobalFoldPlannerDecision{FallbackReason: reason}, nil
	}
	groups, ok, reason, err := coalesceGloballyFoldedBatchGroupsWithResolver(foldedEntries, expectedKind, maxRows, maxBytes, planning.resolvePK, planning.resolveSplitKeys)
	if err != nil {
		return nil, batchGlobalFoldPlannerDecision{}, err
	}
	if !ok {
		return nil, batchGlobalFoldPlannerDecision{FallbackReason: reason}, nil
	}
	return groups, batchGlobalFoldPlannerDecision{Applied: true}, nil
}

func coalesceBatchGroupsWithUpdateMerge(entries []BatchEntry, maxRows, maxBytes int, resolvePK batchUpdatePKResolver, splitResolvers ...batchSplitKeyResolver) ([]coalescedBatchGroup, error) {
	if len(entries) == 0 {
		return nil, nil
	}
	if maxRows <= 0 {
		maxRows = 1
	}
	var resolveSplitKeys batchSplitKeyResolver
	if len(splitResolvers) > 0 {
		resolveSplitKeys = splitResolvers[0]
	}
	if maxBytes <= 0 {
		maxBytes = 1
	}

	out := make([]coalescedBatchGroup, 0, len(entries))
	pending := make([]batchMergeCandidate, 0, len(entries))

	flushPending := func(finalReason string) {
		if len(pending) == 0 {
			return
		}
		out = append(out, coalesceBatchMergeCandidateRun(pending, maxRows, maxBytes, finalReason)...)
		pending = pending[:0]
	}

	for _, entry := range entries {
		candidate, ok, err := buildBatchMergeCandidate(entry, resolvePK, resolveSplitKeys)
		if err != nil {
			return nil, err
		}
		if !ok {
			flushPending(batchGroupBreakReasonNonMergeableStatement)
			out = append(out, newPassthroughBatchGroup(entry, batchGroupBreakReasonNonMergeableStatement))
			continue
		}

		if len(pending) == 0 {
			pending = append(pending, candidate)
			continue
		}
		if pending[0].kind != candidate.kind || !sameMergeableInsertTemplate(pending[len(pending)-1].insert, candidate.insert) {
			finalReason := batchGroupBreakReasonTemplateChange
			if pending[0].kind != candidate.kind {
				finalReason = batchGroupBreakReasonMergeKindChange
			}
			flushPending(finalReason)
			pending = append(pending, candidate)
			continue
		}

		pending = append(pending, candidate)
	}

	flushPending(batchGroupBreakReasonEndOfEntries)
	return out, nil
}

func coalesceGloballyFoldedBatchGroupsWithResolver(entries []BatchEntry, expectedKind batchMergeKind, maxRows, maxBytes int, resolvePK batchUpdatePKResolver, splitResolvers ...batchSplitKeyResolver) ([]coalescedBatchGroup, bool, string, error) {
	if len(entries) == 0 {
		return nil, true, "", nil
	}
	if maxRows <= 0 {
		maxRows = 1
	}
	if maxBytes <= 0 {
		maxBytes = 1
	}
	var resolveSplitKeys batchSplitKeyResolver
	if len(splitResolvers) > 0 {
		resolveSplitKeys = splitResolvers[0]
	}

	candidates := make([]batchMergeCandidate, 0, len(entries))
	for _, entry := range entries {
		candidate, ok, err := buildBatchMergeCandidate(entry, resolvePK, resolveSplitKeys)
		if err != nil {
			return nil, false, "", err
		}
		if !ok {
			return nil, false, batchGlobalFoldFallbackFoldedEntryNotMergeable, nil
		}
		if candidate.kind != expectedKind {
			return nil, false, batchGlobalFoldFallbackFoldedKindMismatch, nil
		}
		candidates = append(candidates, candidate)
	}
	return coalesceTemplateBucketedStrictBatchMergeCandidates(candidates, maxRows, maxBytes), true, "", nil
}

func coalesceStrictBatchMergeCandidates(candidates []batchMergeCandidate, maxRows, maxBytes int) []coalescedBatchGroup {
	if len(candidates) == 0 {
		return nil
	}
	if maxRows <= 0 {
		maxRows = 1
	}
	if maxBytes <= 0 {
		maxBytes = 1
	}

	useSplitKeys := true
	for i := range candidates {
		if len(candidates[i].splitKeys) == 0 {
			useSplitKeys = false
			break
		}
	}
	groupingStrategy := batchGroupingStrategyPrimaryKey
	if useSplitKeys {
		groupingStrategy = batchGroupingStrategyPartitionGroup
	}

	out := make([]coalescedBatchGroup, 0, len(candidates))
	pending := make([]batchMergeCandidate, 0, len(candidates))
	rows := 0
	bytes := 0
	splitKeySignature := ""

	flushPending := func(breakReason string) {
		if len(pending) == 0 {
			return
		}
		segment := batchMergeSegment{
			candidates:       append([]batchMergeCandidate(nil), pending...),
			groupingStrategy: groupingStrategy,
			debug: batchGroupDebugInfo{
				templateKey: batchMergeCandidatesTemplateKey(pending),
				breakReason: breakReason,
			},
		}
		if useSplitKeys {
			segment.partitionHits = batchMergeSegmentPartitionHits(pending)
		}
		out = append(out, buildCoalescedBatchGroup(segment))
		pending = pending[:0]
		rows = 0
		bytes = 0
		splitKeySignature = ""
	}

	for _, candidate := range candidates {
		candidateSplitKeySignature := ""
		if useSplitKeys {
			candidateSplitKeySignature = batchMergeCandidateSplitKeySignature(candidate)
		}
		if len(pending) > 0 {
			rowsExceeded := rows+candidate.rows > maxRows
			bytesExceeded := bytes+candidate.bytes > maxBytes
			limitExceeded := rowsExceeded || bytesExceeded
			templateChanged := !sameMergeableInsertTemplate(pending[len(pending)-1].insert, candidate.insert)
			partitionChanged := useSplitKeys && splitKeySignature != candidateSplitKeySignature
			if limitExceeded || templateChanged || partitionChanged {
				flushPending(batchChunkBreakReason(rowsExceeded, bytesExceeded, templateChanged, partitionChanged, batchGroupBreakReasonEndOfEntries))
			}
		}
		if len(pending) == 0 && useSplitKeys {
			splitKeySignature = candidateSplitKeySignature
		}
		pending = append(pending, candidate)
		rows += candidate.rows
		bytes += candidate.bytes
	}

	flushPending(batchGroupBreakReasonEndOfEntries)
	return out
}

func coalesceTemplateBucketedStrictBatchMergeCandidates(candidates []batchMergeCandidate, maxRows, maxBytes int) []coalescedBatchGroup {
	if len(candidates) == 0 {
		return nil
	}

	useSplitKeys := true
	for i := range candidates {
		if len(candidates[i].splitKeys) == 0 {
			useSplitKeys = false
			break
		}
	}

	type bucketKey struct {
		templateKey       string
		splitKeySignature string
	}

	order := make([]bucketKey, 0, len(candidates))
	buckets := make(map[bucketKey][]batchMergeCandidate, len(candidates))
	for _, candidate := range candidates {
		key := bucketKey{
			templateKey: batchEntryPlanTemplateKey(candidate.merged),
		}
		if useSplitKeys {
			key.splitKeySignature = batchMergeCandidateSplitKeySignature(candidate)
		}
		if _, ok := buckets[key]; !ok {
			order = append(order, key)
		}
		buckets[key] = append(buckets[key], candidate)
	}

	out := make([]coalescedBatchGroup, 0, len(candidates))
	for _, key := range order {
		out = append(out, coalesceStrictBatchMergeCandidates(buckets[key], maxRows, maxBytes)...)
	}
	return out
}

func batchMergeCandidateSplitKeySignature(candidate batchMergeCandidate) string {
	if len(candidate.splitKeys) == 0 {
		return ""
	}
	return strings.Join(candidate.splitKeys, "\x00")
}

func buildBatchMergeCandidate(entry BatchEntry, resolvePK batchUpdatePKResolver, splitResolvers ...batchSplitKeyResolver) (batchMergeCandidate, bool, error) {
	var resolveSplitKeys batchSplitKeyResolver
	if len(splitResolvers) > 0 {
		resolveSplitKeys = splitResolvers[0]
	}
	if insert, ok := mergeableBatchInsert(entry.Statement); ok {
		cloned := cloneInsertStatementForMerge(insert)
		pkKeys := batchEntryPrimaryKeyStrings(entry, resolvePK)
		splitKeys := resolvedBatchSplitKeys(entry, cloned, entry.Params, resolveSplitKeys)
		kind := batchMergeKindInsert
		if entry.folded {
			kind = entry.mergeKind
		}
		incrementCols := batchEntryIncrementColumns(entry, cloned)
		return batchMergeCandidate{
			kind: kind,
			merged: BatchEntry{
				Statement:     cloned,
				Params:        append([]interface{}(nil), entry.Params...),
				mergeKind:     kind,
				folded:        entry.folded,
				incrementCols: cloneBatchColumnSet(incrementCols),
			},
			insert:        cloned,
			originals:     batchEntryOriginals(entry),
			rows:          len(cloned.Values),
			bytes:         estimateInsertBatchBytes(cloned),
			pkKeys:        append([]string(nil), pkKeys...),
			splitKeys:     splitKeys,
			incrementCols: cloneBatchColumnSet(incrementCols),
		}, true, nil
	}
	if resolvePK == nil {
		return batchMergeCandidate{}, false, nil
	}

	converted, ok, err := mergeableBatchUpdateAsInsert(entry.Statement, resolvePK)
	if err != nil {
		return batchMergeCandidate{}, false, err
	}
	if !ok {
		return batchMergeCandidate{}, false, nil
	}
	pkKeys := batchEntryPrimaryKeyStrings(entry, resolvePK)
	splitKeys := resolvedBatchSplitKeys(entry, converted, entry.Params, resolveSplitKeys)
	return batchMergeCandidate{
		kind: batchMergeKindUpdate,
		merged: BatchEntry{
			Statement:     converted,
			Params:        append([]interface{}(nil), entry.Params...),
			mergeKind:     batchMergeKindUpdate,
			folded:        entry.folded,
			incrementCols: cloneBatchColumnSet(converted.OnDuplicateIncrementColumns),
		},
		insert:        converted,
		originals:     batchEntryOriginals(entry),
		rows:          len(converted.Values),
		bytes:         estimateInsertBatchBytes(converted),
		pkKeys:        append([]string(nil), pkKeys...),
		splitKeys:     splitKeys,
		incrementCols: cloneBatchColumnSet(converted.OnDuplicateIncrementColumns),
	}, true, nil
}

func resolvedBatchSplitKeys(entry BatchEntry, insert *parser.InsertStatement, params []interface{}, resolveSplitKeys batchSplitKeyResolver) []string {
	if resolveSplitKeys == nil {
		return nil
	}
	keys := resolveSplitKeys(entry, insert, params)
	if len(keys) == 0 {
		return nil
	}
	out := make([]string, 0, len(keys))
	seen := make(map[string]struct{}, len(keys))
	for _, key := range keys {
		if key == "" {
			continue
		}
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		out = append(out, key)
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func batchEntryIncrementColumns(entry BatchEntry, insert *parser.InsertStatement) map[string]struct{} {
	if len(entry.incrementCols) > 0 {
		return cloneBatchColumnSet(entry.incrementCols)
	}
	if insert == nil {
		return nil
	}
	return cloneBatchColumnSet(insert.OnDuplicateIncrementColumns)
}

func mergeableBatchUpdateAsInsert(stmt parser.Statement, resolvePK batchUpdatePKResolver) (*parser.InsertStatement, bool, error) {
	update, ok := stmt.(*parser.UpdateStatement)
	if !ok || update == nil {
		return nil, false, nil
	}
	if update.IfExists || update.IfCondition != nil || update.TTL != nil || update.Timestamp != nil {
		return nil, false, nil
	}
	if update.Table == nil || len(update.Assigments) == 0 {
		return nil, false, nil
	}
	if hasDuplicateAssignmentColumns(update.Assigments) {
		return nil, false, nil
	}
	for _, assignment := range update.Assigments {
		if assignment == nil || assignment.Op != "=" {
			return nil, false, nil
		}
		if _, ok := assignment.Target.(*parser.ColumnExpr); !ok {
			return nil, false, nil
		}
	}

	whereConditions, err := extractEqualityConditions(update.Where)
	if err != nil {
		return nil, false, nil
	}
	if len(whereConditions) == 0 || hasRangeConditions(whereConditions) || hasInConditions(whereConditions) {
		return nil, false, nil
	}

	pkCols, err := resolvePK(update.Table)
	if err != nil {
		return nil, false, err
	}
	if !hasAllPrimaryKeyEquality(pkCols, whereConditions) || hasNonPrimaryKeyConditions(pkCols, whereConditions) {
		return nil, false, nil
	}
	for _, assignment := range update.Assigments {
		if isColumnInList(assignment.Column, pkCols) {
			return nil, false, nil
		}
	}

	converted, err := convertBatchUpdateToInsert(update, whereConditions)
	if err != nil {
		// Keep merge conversion best-effort; unsupported arithmetic falls back to plain UPDATE execution.
		return nil, false, nil
	}
	return converted, true, nil
}

func convertBatchUpdateToInsert(update *parser.UpdateStatement, whereConditions []equalityCondition) (*parser.InsertStatement, error) {
	if update == nil {
		return nil, nil
	}
	columns := make([]string, 0, len(update.Assigments)+len(whereConditions))
	values := make([]parser.Expression, 0, len(update.Assigments)+len(whereConditions))
	incrementCols := make(map[string]struct{}, len(update.Assigments))
	for _, assignment := range update.Assigments {
		insertValue := assignment.Value
		if exprHasColumnRef(assignment.Value) {
			delta, err := arithmeticAssignmentDelta(assignment)
			if err != nil {
				return nil, err
			}
			insertValue = delta
			if normalized := normalizeBatchColumnName(assignment.Column); normalized != "" {
				incrementCols[normalized] = struct{}{}
			}
		}
		columns = append(columns, assignment.Column)
		values = append(values, cloneExpressionForBatchMerge(insertValue))
	}
	for _, cond := range whereConditions {
		columns = append(columns, cond.column)
		values = append(values, cloneExpressionForBatchMerge(cond.value))
	}

	var table *parser.TableName
	if update.Table != nil {
		copied := *update.Table
		table = &copied
	}

	stmt := &parser.InsertStatement{
		Token:   update.Token,
		Table:   table,
		Columns: columns,
		Values:  [][]parser.Expression{values},
	}
	if len(incrementCols) > 0 {
		stmt.OnDuplicateIncrementColumns = incrementCols
	}
	return stmt, nil
}

func hasDuplicateAssignmentColumns(assignments []*parser.Assignment) bool {
	seen := make(map[string]struct{}, len(assignments))
	for _, assignment := range assignments {
		if assignment == nil {
			continue
		}
		key := strings.ToLower(strings.TrimSpace(assignment.Column))
		if key == "" {
			continue
		}
		if _, ok := seen[key]; ok {
			return true
		}
		seen[key] = struct{}{}
	}
	return false
}

func coalesceBatchInsertGroups(entries []BatchEntry, maxRows, maxBytes int) []coalescedBatchGroup {
	return coalesceBatchInsertGroupsWithResolver(entries, maxRows, maxBytes, nil, nil)
}

func coalesceBatchInsertGroupsWithResolver(entries []BatchEntry, maxRows, maxBytes int, resolvePK batchUpdatePKResolver, splitResolvers ...batchSplitKeyResolver) []coalescedBatchGroup {
	if len(entries) < 2 {
		if len(entries) == 0 {
			return nil
		}
		out := make([]coalescedBatchGroup, 0, len(entries))
		for _, entry := range entries {
			out = append(out, newPassthroughBatchGroup(entry, batchGroupBreakReasonSingleStatement))
		}
		return out
	}
	if maxRows <= 0 {
		maxRows = 1
	}
	if maxBytes <= 0 {
		maxBytes = 1
	}
	var resolveSplitKeys batchSplitKeyResolver
	if len(splitResolvers) > 0 {
		resolveSplitKeys = splitResolvers[0]
	}

	out := make([]coalescedBatchGroup, 0, len(entries))
	pending := make([]batchMergeCandidate, 0, len(entries))

	flushPending := func(finalReason string) {
		if len(pending) == 0 {
			return
		}
		out = append(out, coalesceBatchMergeCandidateRun(pending, maxRows, maxBytes, finalReason)...)
		pending = pending[:0]
	}

	for _, entry := range entries {
		candidate, ok, err := buildBatchMergeCandidate(entry, resolvePK, resolveSplitKeys)
		if err != nil {
			flushPending(batchGroupBreakReasonCandidateBuildError)
			out = append(out, newPassthroughBatchGroup(entry, batchGroupBreakReasonCandidateBuildError))
			continue
		}
		if !ok || candidate.kind != batchMergeKindInsert {
			flushPending(batchGroupBreakReasonNonMergeableStatement)
			out = append(out, newPassthroughBatchGroup(entry, batchGroupBreakReasonNonMergeableStatement))
			continue
		}

		if len(pending) == 0 {
			pending = append(pending, candidate)
			continue
		}
		if !sameMergeableInsertTemplate(pending[len(pending)-1].insert, candidate.insert) {
			flushPending(batchGroupBreakReasonTemplateChange)
			pending = append(pending, candidate)
			continue
		}

		pending = append(pending, candidate)
	}

	flushPending(batchGroupBreakReasonEndOfEntries)
	return out
}

func coalesceBatchMergeCandidateRun(candidates []batchMergeCandidate, maxRows, maxBytes int, finalReason string) []coalescedBatchGroup {
	if len(candidates) == 0 {
		return nil
	}
	if maxRows <= 0 {
		maxRows = 1
	}
	if maxBytes <= 0 {
		maxBytes = 1
	}

	candidates = collapseBatchMergeCandidatesByPrimaryKey(candidates)
	segments := splitBatchMergeCandidateRun(candidates, maxRows, maxBytes, finalReason)
	segments = absorbDuplicateFoldSingletonSegments(segments, maxRows, maxBytes)
	for i := range segments {
		if isDuplicateFoldSingletonSegment(segments[i]) {
			segments[i].debug.breakReason = batchGroupBreakReasonDuplicateFoldSingletonUnabsorbed
		}
	}
	out := make([]coalescedBatchGroup, 0, len(segments))
	for _, segment := range segments {
		if len(segment.candidates) == 0 {
			continue
		}
		out = append(out, buildCoalescedBatchGroup(segment))
	}
	return out
}

func splitBatchMergeCandidateRun(candidates []batchMergeCandidate, maxRows, maxBytes int, finalReason string) []batchMergeSegment {
	if len(candidates) == 0 {
		return nil
	}
	if finalReason == "" {
		finalReason = batchGroupBreakReasonEndOfEntries
	}

	useSplitKeys := true
	for i := range candidates {
		if len(candidates[i].splitKeys) == 0 {
			useSplitKeys = false
			break
		}
	}
	keySetForCandidate := func(candidate batchMergeCandidate) []string {
		if useSplitKeys {
			return candidate.splitKeys
		}
		return candidate.pkKeys
	}

	lastByKey := make(map[string]int)
	for i := range candidates {
		keys := keySetForCandidate(candidates[i])
		seen := make(map[string]struct{}, len(keys))
		for _, key := range keys {
			if key == "" {
				continue
			}
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			lastByKey[key] = i
		}
	}

	groupingStrategy := batchGroupingStrategyPrimaryKey
	if useSplitKeys {
		groupingStrategy = batchGroupingStrategyPartitionGroup
	}
	makeSegment := func(segmentCandidates []batchMergeCandidate, breakReason string) batchMergeSegment {
		segment := batchMergeSegment{
			candidates:       segmentCandidates,
			groupingStrategy: groupingStrategy,
			debug: batchGroupDebugInfo{
				templateKey: batchMergeCandidatesTemplateKey(segmentCandidates),
				breakReason: breakReason,
			},
		}
		if useSplitKeys {
			segment.partitionHits = batchMergeSegmentPartitionHits(segmentCandidates)
		}
		return segment
	}

	segments := make([]batchMergeSegment, 0, len(candidates))
	for start := 0; start < len(candidates); {
		rows := 0
		bytes := 0
		openUntil := start - 1
		safeEnd := -1
		end := start
		rowsExceeded := false
		bytesExceeded := false

		for end < len(candidates) {
			candidate := candidates[end]
			rowsExceeded = rows+candidate.rows > maxRows
			bytesExceeded = bytes+candidate.bytes > maxBytes
			if rows > 0 && (rowsExceeded || bytesExceeded) && safeEnd >= start {
				break
			}

			rows += candidate.rows
			bytes += candidate.bytes

			candidateUntil := end
			keys := keySetForCandidate(candidate)
			seen := make(map[string]struct{}, len(keys))
			for _, key := range keys {
				if key == "" {
					continue
				}
				if _, ok := seen[key]; ok {
					continue
				}
				seen[key] = struct{}{}
				if last, ok := lastByKey[key]; ok && last > candidateUntil {
					candidateUntil = last
				}
			}
			if candidateUntil > openUntil {
				openUntil = candidateUntil
			}
			if end == openUntil {
				safeEnd = end
			}
			end++
		}

		if end == len(candidates) {
			segments = append(segments, makeSegment(candidates[start:end], finalReason))
			break
		}
		if safeEnd >= start {
			segments = append(segments, makeSegment(candidates[start:safeEnd+1], batchLimitBreakReason(rowsExceeded, bytesExceeded)))
			start = safeEnd + 1
			continue
		}

		for end < len(candidates) {
			candidate := candidates[end]
			rows += candidate.rows
			bytes += candidate.bytes

			candidateUntil := end
			keys := keySetForCandidate(candidate)
			seen := make(map[string]struct{}, len(keys))
			for _, key := range keys {
				if key == "" {
					continue
				}
				if _, ok := seen[key]; ok {
					continue
				}
				seen[key] = struct{}{}
				if last, ok := lastByKey[key]; ok && last > candidateUntil {
					candidateUntil = last
				}
			}
			if candidateUntil > openUntil {
				openUntil = candidateUntil
			}
			if end == openUntil {
				safeEnd = end
				end++
				break
			}
			end++
		}

		if safeEnd >= start {
			segments = append(segments, makeSegment(candidates[start:safeEnd+1], batchSafeBoundaryAfterLimitBreakReason(rowsExceeded, bytesExceeded)))
			start = safeEnd + 1
			continue
		}

		segments = append(segments, makeSegment(candidates[start:len(candidates)], finalReason))
		break
	}

	return segments
}

func absorbDuplicateFoldSingletonSegments(segments []batchMergeSegment, maxRows, maxBytes int) []batchMergeSegment {
	if len(segments) < 2 {
		return segments
	}

	out := make([]batchMergeSegment, 0, len(segments))
	for i := 0; i < len(segments); i++ {
		segment := segments[i]
		if !isDuplicateFoldSingletonSegment(segment) {
			out = append(out, segment)
			continue
		}
		if len(out) > 0 && canMergeBatchMergeSegments(out[len(out)-1], segment, maxRows, maxBytes) {
			merged := mergeBatchMergeSegments(out[len(out)-1], segment)
			merged.debug = out[len(out)-1].debug
			out[len(out)-1] = merged
			continue
		}
		if i+1 < len(segments) && canMergeBatchMergeSegments(segment, segments[i+1], maxRows, maxBytes) {
			merged := mergeBatchMergeSegments(segment, segments[i+1])
			merged.debug = segments[i+1].debug
			segments[i+1] = merged
			continue
		}
		out = append(out, segment)
	}
	return out
}

func isDuplicateFoldSingletonSegment(segment batchMergeSegment) bool {
	if len(segment.candidates) != 1 {
		return false
	}
	candidate := segment.candidates[0]
	return candidate.rows == 1 && len(candidate.originals) > 1
}

func canMergeBatchMergeSegments(left, right batchMergeSegment, maxRows, maxBytes int) bool {
	if len(left.candidates) == 0 || len(right.candidates) == 0 {
		return false
	}
	if left.groupingStrategy != "" && right.groupingStrategy != "" && left.groupingStrategy != right.groupingStrategy {
		return false
	}
	leftRows, leftBytes := batchMergeSegmentSize(left)
	rightRows, rightBytes := batchMergeSegmentSize(right)
	return leftRows+rightRows <= maxRows && leftBytes+rightBytes <= maxBytes
}

func batchMergeSegmentSize(segment batchMergeSegment) (rows, bytes int) {
	for _, candidate := range segment.candidates {
		rows += candidate.rows
		bytes += candidate.bytes
	}
	return rows, bytes
}

func mergeBatchMergeSegments(left, right batchMergeSegment) batchMergeSegment {
	groupingStrategy := left.groupingStrategy
	if groupingStrategy == "" {
		groupingStrategy = right.groupingStrategy
	}
	merged := batchMergeSegment{
		candidates:       make([]batchMergeCandidate, 0, len(left.candidates)+len(right.candidates)),
		groupingStrategy: groupingStrategy,
	}
	merged.candidates = append(merged.candidates, left.candidates...)
	merged.candidates = append(merged.candidates, right.candidates...)
	if groupingStrategy == batchGroupingStrategyPartitionGroup {
		merged.partitionHits = batchMergeSegmentPartitionHits(merged.candidates)
	}
	return merged
}

func batchMergeSegmentPartitionHits(candidates []batchMergeCandidate) []string {
	if len(candidates) == 0 {
		return nil
	}
	out := make([]string, 0)
	seen := make(map[string]struct{})
	for _, candidate := range candidates {
		for _, key := range candidate.splitKeys {
			if key == "" {
				continue
			}
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			out = append(out, key)
		}
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func buildCoalescedBatchGroup(segment batchMergeSegment) coalescedBatchGroup {
	candidates := segment.candidates
	if len(candidates) == 0 {
		return coalescedBatchGroup{}
	}
	if len(candidates) == 1 {
		candidate := candidates[0]
		originals := append([]BatchEntry(nil), candidate.originals...)
		if len(originals) == 0 {
			return coalescedBatchGroup{}
		}
		group := coalescedBatchGroup{
			merged:           candidate.merged,
			originals:        originals,
			groupingStrategy: segment.groupingStrategy,
			partitionHits:    append([]string(nil), segment.partitionHits...),
			debug:            segment.debug,
		}
		if len(originals) == 1 && !candidate.merged.folded {
			group.merged = originals[0]
		}
		group.debug.templateKey = batchEntryPlanTemplateKey(group.merged)
		return group
	}

	mergedInsert := cloneInsertStatementForMerge(candidates[0].insert)
	merged := BatchEntry{
		Statement:     mergedInsert,
		Params:        append([]interface{}(nil), candidates[0].merged.Params...),
		mergeKind:     candidates[0].kind,
		folded:        candidates[0].merged.folded,
		incrementCols: cloneBatchColumnSet(candidates[0].incrementCols),
	}
	originals := append([]BatchEntry(nil), candidates[0].originals...)
	for i := 1; i < len(candidates); i++ {
		mergedInsert.Values = append(mergedInsert.Values, cloneInsertValues(candidates[i].insert.Values)...)
		merged.Params = append(merged.Params, candidates[i].merged.Params...)
		originals = append(originals, candidates[i].originals...)
	}

	return coalescedBatchGroup{
		merged:           merged,
		originals:        originals,
		groupingStrategy: segment.groupingStrategy,
		partitionHits:    append([]string(nil), segment.partitionHits...),
		debug: batchGroupDebugInfo{
			templateKey: batchEntryPlanTemplateKey(merged),
			breakReason: segment.debug.breakReason,
		},
	}
}

func newPassthroughBatchGroup(entry BatchEntry, breakReason string) coalescedBatchGroup {
	return coalescedBatchGroup{
		merged:           entry,
		originals:        batchEntryOriginals(entry),
		groupingStrategy: batchGroupingStrategyPrimaryKey,
		debug: batchGroupDebugInfo{
			templateKey: batchEntryPlanTemplateKey(entry),
			breakReason: breakReason,
		},
	}
}

func batchMergeCandidatesTemplateKey(candidates []batchMergeCandidate) string {
	if len(candidates) == 0 {
		return ""
	}
	return batchEntryPlanTemplateKey(candidates[0].merged)
}

func batchEntryPlanTemplateKey(entry BatchEntry) string {
	switch stmt := entry.Statement.(type) {
	case *parser.InsertStatement:
		return batchInsertPlanTemplateKey(stmt, entry.mergeKind)
	case *parser.UpdateStatement:
		cols := make([]string, 0, len(stmt.Assigments))
		for _, assignment := range stmt.Assigments {
			if assignment == nil {
				continue
			}
			cols = append(cols, normalizeBatchColumnName(assignment.Column))
		}
		return fmt.Sprintf("update:%s|set=%s", batchTableLogName(stmt.Table), strings.Join(cols, ","))
	case *parser.DeleteStatement:
		return fmt.Sprintf("delete:%s", batchTableLogName(stmt.From))
	default:
		return fmt.Sprintf("%T", entry.Statement)
	}
}

func batchInsertPlanTemplateKey(stmt *parser.InsertStatement, kind batchMergeKind) string {
	if stmt == nil {
		return "<nil_insert_template>"
	}
	columns := make([]string, 0, len(stmt.Columns))
	for _, col := range stmt.Columns {
		columns = append(columns, normalizeBatchColumnName(col))
	}
	incrementCols := batchSortedColumnSet(stmt.OnDuplicateIncrementColumns)
	flags := make([]string, 0, 4)
	if stmt.JSON {
		flags = append(flags, "json")
	}
	if stmt.IfNotExists {
		flags = append(flags, "if_not_exists")
	}
	if stmt.TTL != nil {
		flags = append(flags, "ttl")
	}
	if stmt.Timestamp != nil {
		flags = append(flags, "timestamp")
	}
	kindName := "insert"
	if kind == batchMergeKindUpdate {
		kindName = "update"
	}
	parts := []string{
		fmt.Sprintf("%s:%s", kindName, batchTableLogName(stmt.Table)),
		fmt.Sprintf("cols=%s", strings.Join(columns, ",")),
	}
	if len(incrementCols) > 0 {
		parts = append(parts, fmt.Sprintf("increments=%s", strings.Join(incrementCols, ",")))
	}
	if len(flags) > 0 {
		parts = append(parts, fmt.Sprintf("flags=%s", strings.Join(flags, ",")))
	}
	return strings.Join(parts, "|")
}

func batchSortedColumnSet(columns map[string]struct{}) []string {
	if len(columns) == 0 {
		return nil
	}
	out := make([]string, 0, len(columns))
	for column := range columns {
		if column == "" {
			continue
		}
		out = append(out, column)
	}
	if len(out) == 0 {
		return nil
	}
	sort.Strings(out)
	return out
}

func batchChunkBreakReason(rowsExceeded, bytesExceeded, templateChanged, partitionChanged bool, fallback string) string {
	reasons := make([]string, 0, 4)
	if rowsExceeded && bytesExceeded {
		reasons = append(reasons, batchGroupBreakReasonLimitRowsAndBytes)
	} else {
		if rowsExceeded {
			reasons = append(reasons, batchGroupBreakReasonLimitRows)
		}
		if bytesExceeded {
			reasons = append(reasons, batchGroupBreakReasonLimitBytes)
		}
	}
	if templateChanged {
		reasons = append(reasons, batchGroupBreakReasonTemplateChange)
	}
	if partitionChanged {
		reasons = append(reasons, batchGroupBreakReasonPartitionChange)
	}
	if len(reasons) == 0 {
		return fallback
	}
	return strings.Join(reasons, "+")
}

func batchLimitBreakReason(rowsExceeded, bytesExceeded bool) string {
	if rowsExceeded && bytesExceeded {
		return batchGroupBreakReasonLimitRowsAndBytes
	}
	if rowsExceeded {
		return batchGroupBreakReasonLimitRows
	}
	if bytesExceeded {
		return batchGroupBreakReasonLimitBytes
	}
	return batchGroupBreakReasonEndOfEntries
}

func batchSafeBoundaryAfterLimitBreakReason(rowsExceeded, bytesExceeded bool) string {
	if rowsExceeded && bytesExceeded {
		return batchGroupBreakReasonSafeBoundaryAfterLimitRowsBytes
	}
	if rowsExceeded {
		return batchGroupBreakReasonSafeBoundaryAfterLimitRows
	}
	if bytesExceeded {
		return batchGroupBreakReasonSafeBoundaryAfterLimitBytes
	}
	return batchGroupBreakReasonEndOfEntries
}

func collapseBatchMergeCandidatesByPrimaryKey(candidates []batchMergeCandidate) []batchMergeCandidate {
	if len(candidates) < 2 {
		return candidates
	}

	out := make([]batchMergeCandidate, 0, len(candidates))
	indexByPK := make(map[string]int)
	for _, candidate := range candidates {
		pkKey, ok := batchMergeCandidatePrimaryKey(candidate)
		if !ok {
			out = append(out, candidate)
			continue
		}
		if idx, exists := indexByPK[pkKey]; exists {
			merged, err := mergeBatchCandidatesSamePrimaryKey(out[idx], candidate)
			if err == nil {
				out[idx] = merged
				continue
			}
		}
		indexByPK[pkKey] = len(out)
		out = append(out, candidate)
	}
	return out
}

func batchMergeCandidatePrimaryKey(candidate batchMergeCandidate) (string, bool) {
	if len(candidate.pkKeys) != 1 || candidate.pkKeys[0] == "" {
		return "", false
	}
	return candidate.pkKeys[0], true
}

func mergeBatchCandidatesSamePrimaryKey(base, next batchMergeCandidate) (batchMergeCandidate, error) {
	if base.kind != next.kind || !sameMergeableInsertTemplate(base.insert, next.insert) {
		return batchMergeCandidate{}, fmt.Errorf("candidate templates differ")
	}

	baseValues, err := batchMergeCandidateRowValues(base)
	if err != nil {
		return batchMergeCandidate{}, err
	}
	nextValues, err := batchMergeCandidateRowValues(next)
	if err != nil {
		return batchMergeCandidate{}, err
	}
	if len(baseValues) != len(nextValues) || len(baseValues) != len(base.insert.Columns) {
		return batchMergeCandidate{}, fmt.Errorf("candidate values do not match columns")
	}

	incrementCols := cloneBatchColumnSet(base.incrementCols)
	if !sameBatchColumnSet(incrementCols, next.incrementCols) {
		return batchMergeCandidate{}, fmt.Errorf("candidate increment columns differ")
	}
	mergedValues := make([]interface{}, len(baseValues))
	for i, col := range base.insert.Columns {
		if _, ok := incrementCols[normalizeBatchColumnName(col)]; ok {
			value, err := accumulateBatchIncrementValue(baseValues[i], nextValues[i])
			if err != nil {
				return batchMergeCandidate{}, err
			}
			mergedValues[i] = value
			continue
		}
		mergedValues[i] = nextValues[i]
	}

	mergedCandidate := buildBatchMergeCandidateFromResolvedValues(base, mergedValues)
	mergedCandidate.originals = append(append([]BatchEntry(nil), base.originals...), next.originals...)
	if pkKey, ok := batchMergeCandidatePrimaryKey(base); ok {
		mergedCandidate.pkKeys = []string{pkKey}
	}
	mergedCandidate.splitKeys = append([]string(nil), base.splitKeys...)
	return mergedCandidate, nil
}

func batchMergeCandidateRowValues(candidate batchMergeCandidate) ([]interface{}, error) {
	if candidate.insert == nil || len(candidate.insert.Values) != 1 {
		return nil, fmt.Errorf("candidate must contain exactly one row")
	}
	row := candidate.insert.Values[0]
	if len(row) != len(candidate.insert.Columns) {
		return nil, fmt.Errorf("candidate row does not match columns")
	}

	markerIndex := buildBindMarkerIndex(candidate.insert)
	values := make([]interface{}, len(row))
	for i, expr := range row {
		value, err := resolveScalarExpr(expr, candidate.merged.Params, markerIndex)
		if err != nil {
			return nil, err
		}
		values[i] = value
	}
	return values, nil
}

func buildBatchMergeCandidateFromResolvedValues(proto batchMergeCandidate, values []interface{}) batchMergeCandidate {
	insert := cloneInsertStatementForMerge(proto.insert)
	insert.Values = make([][]parser.Expression, 1)
	insert.Values[0] = make([]parser.Expression, len(values))
	params := make([]interface{}, len(values))
	for i, value := range values {
		insert.Values[0][i] = &parser.BindMarker{}
		params[i] = value
	}
	return batchMergeCandidate{
		kind:          proto.kind,
		merged:        BatchEntry{Statement: insert, Params: params, mergeKind: proto.kind, folded: proto.merged.folded, incrementCols: cloneBatchColumnSet(proto.incrementCols)},
		insert:        insert,
		originals:     append([]BatchEntry(nil), proto.originals...),
		rows:          1,
		bytes:         estimateInsertBatchBytes(insert),
		pkKeys:        append([]string(nil), proto.pkKeys...),
		splitKeys:     append([]string(nil), proto.splitKeys...),
		incrementCols: cloneBatchColumnSet(proto.incrementCols),
	}
}

func accumulateBatchIncrementValue(current, next interface{}) (interface{}, error) {
	if current == nil {
		current = int64(0)
	}
	if next == nil {
		next = int64(0)
	}

	if isFloatLike(current) || isFloatLike(next) {
		a, err := batchNumericAsFloat64(current)
		if err != nil {
			return nil, err
		}
		b, err := batchNumericAsFloat64(next)
		if err != nil {
			return nil, err
		}
		return a + b, nil
	}

	a, err := batchNumericAsInt64(current)
	if err != nil {
		return nil, err
	}
	b, err := batchNumericAsInt64(next)
	if err != nil {
		return nil, err
	}
	return a + b, nil
}

func isFloatLike(value interface{}) bool {
	switch value.(type) {
	case float32, float64:
		return true
	default:
		return false
	}
}

func batchNumericAsFloat64(value interface{}) (float64, error) {
	switch v := value.(type) {
	case int:
		return float64(v), nil
	case int8:
		return float64(v), nil
	case int16:
		return float64(v), nil
	case int32:
		return float64(v), nil
	case int64:
		return float64(v), nil
	case uint:
		return float64(v), nil
	case uint8:
		return float64(v), nil
	case uint16:
		return float64(v), nil
	case uint32:
		return float64(v), nil
	case uint64:
		return float64(v), nil
	case float32:
		return float64(v), nil
	case float64:
		return v, nil
	default:
		return 0, fmt.Errorf("unsupported increment value type %T", value)
	}
}

func batchNumericAsInt64(value interface{}) (int64, error) {
	switch v := value.(type) {
	case int:
		return int64(v), nil
	case int8:
		return int64(v), nil
	case int16:
		return int64(v), nil
	case int32:
		return int64(v), nil
	case int64:
		return v, nil
	case uint:
		if uint64(v) > math.MaxInt64 {
			return 0, fmt.Errorf("unsigned increment overflows int64")
		}
		return int64(v), nil
	case uint8:
		return int64(v), nil
	case uint16:
		return int64(v), nil
	case uint32:
		return int64(v), nil
	case uint64:
		if v > math.MaxInt64 {
			return 0, fmt.Errorf("unsigned increment overflows int64")
		}
		return int64(v), nil
	default:
		return 0, fmt.Errorf("unsupported increment value type %T", value)
	}
}

func (e *Executor) batchInsertMergeLimits() (int, int) {
	if e == nil {
		return defaultBatchInsertMergeMaxRows, defaultBatchInsertMergeMaxBytes
	}
	maxRows := e.batchInsertMergeRows
	if maxRows <= 0 {
		maxRows = defaultBatchInsertMergeMaxRows
	}
	maxBytes := e.batchInsertMergeBytes
	if maxBytes <= 0 {
		maxBytes = defaultBatchInsertMergeMaxBytes
	}
	return maxRows, maxBytes
}

func mergeableBatchInsert(stmt parser.Statement) (*parser.InsertStatement, bool) {
	insert, ok := stmt.(*parser.InsertStatement)
	if !ok || insert == nil {
		return nil, false
	}
	if insert.JSON || insert.IfNotExists || insert.TTL != nil || insert.Timestamp != nil {
		return nil, false
	}
	if insert.Table == nil || len(insert.Columns) == 0 || len(insert.Values) == 0 {
		return nil, false
	}
	for _, row := range insert.Values {
		if len(row) != len(insert.Columns) {
			return nil, false
		}
	}
	return insert, true
}

func sameMergeableInsertTemplate(a, b *parser.InsertStatement) bool {
	if a == nil || b == nil || a.Table == nil || b.Table == nil {
		return false
	}
	if !strings.EqualFold(a.Table.Schema, b.Table.Schema) || !strings.EqualFold(a.Table.Name, b.Table.Name) {
		return false
	}
	if len(a.Columns) != len(b.Columns) {
		return false
	}
	for i := range a.Columns {
		if !strings.EqualFold(a.Columns[i], b.Columns[i]) {
			return false
		}
	}
	return sameBatchColumnSet(a.OnDuplicateIncrementColumns, b.OnDuplicateIncrementColumns)
}

func cloneInsertStatementForMerge(src *parser.InsertStatement) *parser.InsertStatement {
	if src == nil {
		return nil
	}
	cloned := *src
	cloned.Columns = append([]string(nil), src.Columns...)
	cloned.Values = cloneInsertValues(src.Values)
	cloned.OnDuplicateIncrementColumns = cloneBatchColumnSet(src.OnDuplicateIncrementColumns)
	return &cloned
}

func cloneBatchColumnSet(src map[string]struct{}) map[string]struct{} {
	if len(src) == 0 {
		return nil
	}
	cloned := make(map[string]struct{}, len(src))
	for col := range src {
		if normalized := normalizeBatchColumnName(col); normalized != "" {
			cloned[normalized] = struct{}{}
		}
	}
	if len(cloned) == 0 {
		return nil
	}
	return cloned
}

func sameBatchColumnSet(a, b map[string]struct{}) bool {
	a = cloneBatchColumnSet(a)
	b = cloneBatchColumnSet(b)
	if len(a) != len(b) {
		return false
	}
	for col := range a {
		if _, ok := b[col]; !ok {
			return false
		}
	}
	return true
}

func normalizeBatchColumnName(column string) string {
	return strings.ToLower(strings.TrimSpace(column))
}

func cloneInsertValues(rows [][]parser.Expression) [][]parser.Expression {
	if len(rows) == 0 {
		return nil
	}
	cloned := make([][]parser.Expression, len(rows))
	for i, row := range rows {
		cloned[i] = make([]parser.Expression, len(row))
		for j, expr := range row {
			cloned[i][j] = cloneExpressionForBatchMerge(expr)
		}
	}
	return cloned
}

func cloneExpressionForBatchMerge(expr parser.Expression) parser.Expression {
	if expr == nil {
		return nil
	}
	switch e := expr.(type) {
	case *parser.BindMarker:
		cloned := *e
		return &cloned
	case *parser.NamedBindMarker:
		cloned := *e
		return &cloned
	case *parser.NullLiteral:
		cloned := *e
		return &cloned
	case *parser.StringLiteral:
		cloned := *e
		return &cloned
	case *parser.IntegerLiteral:
		cloned := *e
		return &cloned
	case *parser.FloatLiteral:
		cloned := *e
		return &cloned
	case *parser.BooleanLiteral:
		cloned := *e
		return &cloned
	case *parser.UUIDLiteral:
		cloned := *e
		return &cloned
	case *parser.BlobLiteral:
		cloned := *e
		if e.Value != nil {
			cloned.Value = append([]byte(nil), e.Value...)
		}
		return &cloned
	case *parser.ListLiteral:
		cloned := &parser.ListLiteral{Token: e.Token}
		if len(e.Elements) > 0 {
			cloned.Elements = make([]parser.Expression, len(e.Elements))
			for i, elem := range e.Elements {
				cloned.Elements[i] = cloneExpressionForBatchMerge(elem)
			}
		}
		return cloned
	case *parser.VectorLiteral:
		cloned := &parser.VectorLiteral{Token: e.Token}
		if len(e.Elements) > 0 {
			cloned.Elements = make([]parser.Expression, len(e.Elements))
			for i, elem := range e.Elements {
				cloned.Elements[i] = cloneExpressionForBatchMerge(elem)
			}
		}
		return cloned
	case *parser.SetLiteral:
		cloned := &parser.SetLiteral{Token: e.Token}
		if len(e.Elements) > 0 {
			cloned.Elements = make([]parser.Expression, len(e.Elements))
			for i, elem := range e.Elements {
				cloned.Elements[i] = cloneExpressionForBatchMerge(elem)
			}
		}
		return cloned
	case *parser.MapLiteral:
		cloned := &parser.MapLiteral{Token: e.Token}
		if len(e.Entries) > 0 {
			cloned.Entries = make([]parser.MapEntry, len(e.Entries))
			for i, entry := range e.Entries {
				cloned.Entries[i] = parser.MapEntry{
					Key:   cloneExpressionForBatchMerge(entry.Key),
					Value: cloneExpressionForBatchMerge(entry.Value),
				}
			}
		}
		return cloned
	case *parser.TupleLiteral:
		cloned := &parser.TupleLiteral{Token: e.Token}
		if len(e.Elements) > 0 {
			cloned.Elements = make([]parser.Expression, len(e.Elements))
			for i, elem := range e.Elements {
				cloned.Elements[i] = cloneExpressionForBatchMerge(elem)
			}
		}
		return cloned
	case *parser.BinaryExpr:
		return &parser.BinaryExpr{
			Token: e.Token,
			Op:    e.Op,
			Left:  cloneExpressionForBatchMerge(e.Left),
			Right: cloneExpressionForBatchMerge(e.Right),
		}
	case *parser.UnaryExpr:
		return &parser.UnaryExpr{
			Token: e.Token,
			Op:    e.Op,
			Expr:  cloneExpressionForBatchMerge(e.Expr),
		}
	case *parser.IsNullExpr:
		return &parser.IsNullExpr{
			Token: e.Token,
			Expr:  cloneExpressionForBatchMerge(e.Expr),
			Not:   e.Not,
		}
	case *parser.CollectionAccessExpr:
		return &parser.CollectionAccessExpr{
			Token:  e.Token,
			Target: cloneExpressionForBatchMerge(e.Target),
			Start:  cloneExpressionForBatchMerge(e.Start),
			End:    cloneExpressionForBatchMerge(e.End),
			Slice:  e.Slice,
		}
	case *parser.AliasedExpr:
		return &parser.AliasedExpr{
			Token: e.Token,
			Expr:  cloneExpressionForBatchMerge(e.Expr),
			Alias: e.Alias,
		}
	case *parser.CastExpr:
		return &parser.CastExpr{
			Token: e.Token,
			Expr:  cloneExpressionForBatchMerge(e.Expr),
			Type:  e.Type,
		}
	case *parser.TypeHintExpr:
		return &parser.TypeHintExpr{
			Token: e.Token,
			Expr:  cloneExpressionForBatchMerge(e.Expr),
			Type:  e.Type,
		}
	case *parser.FunctionCall:
		cloned := &parser.FunctionCall{
			Token:    e.Token,
			Name:     e.Name,
			Distinct: e.Distinct,
		}
		if len(e.Args) > 0 {
			cloned.Args = make([]parser.Expression, len(e.Args))
			for i, arg := range e.Args {
				cloned.Args[i] = cloneExpressionForBatchMerge(arg)
			}
		}
		return cloned
	default:
		// Preserve unknown expression nodes as-is; bind markers/literals and common containers are deep-cloned above.
		return expr
	}
}

func estimateInsertBatchBytes(stmt *parser.InsertStatement) int {
	if stmt == nil {
		return 0
	}
	total := len(stmt.Columns) * 6
	if stmt.Table != nil {
		total += len(stmt.Table.Schema) + len(stmt.Table.Name) + 32
	}
	for _, row := range stmt.Values {
		total += 4
		for _, expr := range row {
			total += estimateExpressionSQLBytes(expr) + 2
		}
	}
	return total
}

func estimateExpressionSQLBytes(expr parser.Expression) int {
	switch e := expr.(type) {
	case *parser.StringLiteral:
		return len(e.Value) + 2
	case *parser.BlobLiteral:
		return len(e.Value)*2 + 2
	default:
		if expr == nil {
			return 4
		}
		return len(expr.String()) + 2
	}
}

func (e *Executor) batchExecutionPlan(batchType string) (batchExecutionMode, int) {
	if e == nil {
		return batchExecutionTransactional, 1
	}
	if !e.unloggedBatchParallel || !strings.EqualFold(batchType, "UNLOGGED") {
		return batchExecutionTransactional, 1
	}
	workers := e.unloggedBatchWorkers
	if workers <= 0 {
		workers = runtime.NumCPU() * 4
	}
	if workers < 1 {
		workers = 1
	}
	return batchExecutionBestEffortParallel, workers
}

func (e *Executor) executeBatchTransactional(ctx context.Context, user, password string, entries []BatchEntry) (*QueryResult, 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()
	}

	beginStart := time.Now()
	tx, err := conn.BeginTx(ctx, nil)
	metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpBeginTx, err == nil, time.Since(beginStart))
	if err != nil {
		return nil, fmt.Errorf("failed to begin transaction: %w", err)
	}

	var rowsAffected int64
	var cqlWriteRows int64
	for _, entry := range entries {
		result, err := e.executeDMLStatement(ctx, tx, user, password, entry.Statement, entry.Params, nil)
		if err != nil {
			_ = tx.Rollback()
			return nil, err
		}
		if result != nil {
			rowsAffected += result.RowsAffected
			cqlWriteRows += result.CQLWriteRows
		}
	}

	if err := tx.Commit(); err != nil {
		_ = tx.Rollback()
		return nil, fmt.Errorf("failed to commit transaction: %w", err)
	}

	return &QueryResult{RowsAffected: rowsAffected, CQLWriteRows: cqlWriteRows}, nil
}

func (e *Executor) executeBatchBestEffortParallel(ctx context.Context, user, password string, items []batchBestEffortWorkItem, workers int) (*QueryResult, error) {
	if len(items) == 0 {
		return &QueryResult{}, nil
	}
	if workers <= 0 {
		workers = 1
	}
	if _, ok := mysqlbackend.ConnFromContext(ctx); ok && workers > 1 {
		workers = 1
	}
	if e.poolManager == nil && workers > 1 {
		workers = 1
	}
	if workers > len(items) {
		workers = len(items)
	}
	if workers == 1 {
		return e.executeBatchBestEffortSerial(ctx, user, password, items)
	}

	sem := make(chan struct{}, workers)
	errCh := make(chan error, len(items))
	var wg sync.WaitGroup
	var rowsAffected int64
	var cqlWriteRows int64

	for _, item := range items {
		item := item
		wg.Add(1)
		go func() {
			defer wg.Done()
			select {
			case sem <- struct{}{}:
			case <-ctx.Done():
				errCh <- ctx.Err()
				return
			}
			defer func() { <-sem }()

			conn, release, err := e.getParallelConn(ctx, user, password)
			if err != nil {
				errCh <- fmt.Errorf("failed to get connection: %w", err)
				return
			}
			if release != nil {
				defer release()
			}
			itemRowsAffected, itemCQLWriteRows, err := e.executeBatchBestEffortItem(ctx, conn, user, password, item)
			if err != nil {
				errCh <- err
				return
			}
			atomic.AddInt64(&rowsAffected, itemRowsAffected)
			atomic.AddInt64(&cqlWriteRows, itemCQLWriteRows)
		}()
	}

	wg.Wait()
	close(errCh)

	var firstErr error
	for err := range errCh {
		if err != nil && firstErr == nil {
			firstErr = err
		}
	}
	if firstErr != nil {
		return nil, firstErr
	}
	return &QueryResult{RowsAffected: rowsAffected, CQLWriteRows: cqlWriteRows}, nil
}

func (e *Executor) executeBatchBestEffortSerial(ctx context.Context, user, password string, items []batchBestEffortWorkItem) (*QueryResult, error) {
	var rowsAffected int64
	var cqlWriteRows int64
	for _, item := range items {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		conn, release, err := e.getParallelConn(ctx, user, password)
		if err != nil {
			return nil, fmt.Errorf("failed to get connection: %w", err)
		}
		itemRowsAffected, itemCQLWriteRows, err := e.executeBatchBestEffortItem(ctx, conn, user, password, item)
		if release != nil {
			release()
		}
		if err != nil {
			return nil, err
		}
		rowsAffected += itemRowsAffected
		cqlWriteRows += itemCQLWriteRows
	}
	return &QueryResult{RowsAffected: rowsAffected, CQLWriteRows: cqlWriteRows}, nil
}

func (e *Executor) executeBatchBestEffortItem(ctx context.Context, conn *sql.Conn, user, password string, item batchBestEffortWorkItem) (int64, int64, error) {
	result, err := e.executeDMLStatement(ctx, conn, user, password, item.entry.Statement, item.entry.Params, nil)
	if err != nil {
		if len(item.fallback) > 0 {
			var fallbackErr error
			var fallbackRowsAffected int64
			var fallbackCQLWriteRows int64
			for _, fallback := range item.fallback {
				fallbackResult, splitErr := e.executeDMLStatement(ctx, conn, user, password, fallback.Statement, fallback.Params, nil)
				if splitErr != nil && fallbackErr == nil {
					fallbackErr = splitErr
				}
				if splitErr == nil && fallbackResult != nil {
					fallbackRowsAffected += fallbackResult.RowsAffected
					fallbackCQLWriteRows += fallbackResult.CQLWriteRows
				}
			}
			return fallbackRowsAffected, fallbackCQLWriteRows, fallbackErr
		}
		return 0, 0, err
	}
	if result == nil {
		return 0, 0, nil
	}
	return result.RowsAffected, result.CQLWriteRows, nil
}

func (e *Executor) getParallelConn(ctx context.Context, user, password string) (conn *sql.Conn, release func(), err error) {
	bulkRelease, err := e.acquireUnloggedBatchConnLease(ctx)
	if err != nil {
		return nil, nil, err
	}
	if e == nil {
		if bulkRelease != nil {
			bulkRelease()
		}
		return nil, nil, fmt.Errorf("executor is not initialized")
	}
	if e.poolManager == nil {
		c, connRelease, cErr := e.getConn(ctx, user, password)
		if cErr != nil {
			if bulkRelease != nil {
				bulkRelease()
			}
			return nil, nil, cErr
		}
		return c, joinConnReleases(nil, connRelease, bulkRelease), nil
	}
	leaseRelease, err := mysqlbackend.AcquireConnLeaseFromContext(ctx, mysqlbackend.ConnLeaseKindBulk)
	if err != nil {
		if bulkRelease != nil {
			bulkRelease()
		}
		return nil, nil, err
	}
	pool, err := e.poolManager.GetPool(user, password)
	if err != nil {
		if leaseRelease != nil {
			leaseRelease()
		}
		if bulkRelease != nil {
			bulkRelease()
		}
		return nil, nil, fmt.Errorf("failed to get connection pool: %w", err)
	}
	c, err := pool.Conn(ctx)
	if err != nil {
		if leaseRelease != nil {
			leaseRelease()
		}
		if bulkRelease != nil {
			bulkRelease()
		}
		return nil, nil, fmt.Errorf("failed to get connection: %w", err)
	}
	return c, joinConnReleases(c, leaseRelease, bulkRelease), nil
}

func (e *Executor) executeBatchStatement(ctx context.Context, user, password string, stmt *parser.BatchStatement, params []interface{}) (*QueryResult, error) {
	if stmt == nil {
		return &QueryResult{}, nil
	}
	if err := validateBatchStatement(stmt); err != nil {
		return nil, err
	}
	if err := e.ValidateBatchCounterUsage(ctx, user, password, isCounterBatchType(stmt.BatchType), stmt.Entries); err != nil {
		return nil, err
	}
	entries, err := splitBatchParams(stmt, params)
	if err != nil {
		return nil, err
	}
	return e.ExecuteBatch(ctx, user, password, stmt.BatchType, entries)
}

func validateBatchStatement(stmt *parser.BatchStatement) error {
	if stmt == nil {
		return nil
	}
	for _, entry := range stmt.Entries {
		if err := validateBatchEntry(entry); err != nil {
			return err
		}
	}
	return nil
}

func validateBatchEntry(stmt parser.Statement) error {
	switch stmt.(type) {
	case *parser.InsertStatement, *parser.UpdateStatement, *parser.DeleteStatement:
	default:
		return cqlerrors.Invalid("batch supports only DML statements")
	}
	if isConditionalDML(stmt) {
		return cqlerrors.Invalid("conditional statements are not supported in batch")
	}
	return nil
}

type columnTypesLookup func(table *parser.TableName) (map[string]string, error)

type batchTableKey struct {
	schema string
	table  string
}

func isCounterBatchType(batchType string) bool {
	return strings.EqualFold(batchType, "COUNTER")
}

// ValidateBatchCounterUsage enforces counter batch rules for the provided entries.
func (e *Executor) ValidateBatchCounterUsage(ctx context.Context, user, password string, isCounterBatch bool, entries []parser.Statement) error {
	if len(entries) == 0 {
		return nil
	}
	cache := make(map[batchTableKey]map[string]string)
	lookup := func(table *parser.TableName) (map[string]string, error) {
		if table == nil {
			return nil, fmt.Errorf("table is required")
		}
		key := batchTableKey{schema: table.Schema, table: table.Name}
		if cached, ok := cache[key]; ok {
			return cached, nil
		}
		types, err := e.lookupColumnTypes(ctx, user, password, table)
		if err != nil {
			return nil, err
		}
		cache[key] = types
		return types, nil
	}
	return validateBatchCounterUsage(entries, isCounterBatch, lookup)
}

// ValidateBatchCounterUsageEntries enforces counter batch rules for parsed batch entries.
func (e *Executor) ValidateBatchCounterUsageEntries(ctx context.Context, user, password string, isCounterBatch bool, entries []BatchEntry) error {
	if len(entries) == 0 {
		return nil
	}
	cache := make(map[batchTableKey]map[string]string)
	lookup := func(table *parser.TableName) (map[string]string, error) {
		if table == nil {
			return nil, fmt.Errorf("table is required")
		}
		key := batchTableKey{schema: table.Schema, table: table.Name}
		if cached, ok := cache[key]; ok {
			return cached, nil
		}
		types, err := e.lookupColumnTypes(ctx, user, password, table)
		if err != nil {
			return nil, err
		}
		cache[key] = types
		return types, nil
	}
	return validateBatchCounterUsageEntries(entries, isCounterBatch, lookup)
}

func validateBatchCounterUsage(entries []parser.Statement, isCounterBatch bool, lookup columnTypesLookup) error {
	if len(entries) == 0 {
		return nil
	}
	hasCounterMutation := false
	hasNonCounterEntry := false
	for _, entry := range entries {
		isCounter, isNonCounter, err := evaluateBatchCounterEntry(entry, lookup)
		if err != nil {
			return err
		}
		if isCounter {
			hasCounterMutation = true
		}
		if isNonCounter {
			hasNonCounterEntry = true
		}
	}
	return finalizeBatchCounterUsage(isCounterBatch, hasCounterMutation, hasNonCounterEntry)
}

func validateBatchCounterUsageEntries(entries []BatchEntry, isCounterBatch bool, lookup columnTypesLookup) error {
	if len(entries) == 0 {
		return nil
	}
	hasCounterMutation := false
	hasNonCounterEntry := false
	for _, entry := range entries {
		isCounter, isNonCounter, err := evaluateBatchCounterEntry(entry.Statement, lookup)
		if err != nil {
			return err
		}
		if isCounter {
			hasCounterMutation = true
		}
		if isNonCounter {
			hasNonCounterEntry = true
		}
	}
	return finalizeBatchCounterUsage(isCounterBatch, hasCounterMutation, hasNonCounterEntry)
}

func evaluateBatchCounterEntry(entry parser.Statement, lookup columnTypesLookup) (bool, bool, error) {
	switch stmt := entry.(type) {
	case *parser.UpdateStatement:
		columnTypes, err := lookup(stmt.Table)
		if err != nil {
			return false, false, err
		}
		_, hasCounter, err := collectCounterAssignments(stmt.Assigments, columnTypes)
		if err != nil {
			return false, false, cqlerrors.Invalid(err.Error())
		}
		if hasCounter {
			return true, false, nil
		}
		return false, true, nil
	case *parser.DeleteStatement:
		columnTypes, err := lookup(stmt.From)
		if err != nil {
			return false, false, err
		}
		if tableHasCounterColumns(columnTypes) {
			return true, false, nil
		}
		return false, true, nil
	default:
		return false, true, nil
	}
}

func finalizeBatchCounterUsage(isCounterBatch bool, hasCounterMutation bool, hasNonCounterEntry bool) error {
	if isCounterBatch {
		if !hasCounterMutation || hasNonCounterEntry {
			return cqlerrors.Invalid("counter batches may only contain counter mutations")
		}
		return nil
	}
	if hasCounterMutation {
		return cqlerrors.Invalid("counter mutations require BEGIN COUNTER BATCH")
	}
	return nil
}

func isConditionalDML(stmt parser.Statement) bool {
	switch s := stmt.(type) {
	case *parser.InsertStatement:
		return s.IfNotExists
	case *parser.UpdateStatement:
		return s.IfExists || s.IfCondition != nil
	case *parser.DeleteStatement:
		return s.IfExists || s.IfCondition != nil
	default:
		return false
	}
}

func splitBatchParams(stmt *parser.BatchStatement, params []interface{}) ([]BatchEntry, error) {
	if stmt == nil {
		return nil, nil
	}
	markerIndex := buildBindMarkerIndex(stmt)
	entries := make([]BatchEntry, len(stmt.Entries))
	for i, entry := range stmt.Entries {
		var order []int
		appendBindMarkerIndicesStatement(entry, markerIndex, &order)
		entryParams, err := mapParams(params, order)
		if err != nil {
			return nil, err
		}
		entries[i] = BatchEntry{
			Statement: entry,
			Params:    entryParams,
		}
	}
	return entries, nil
}

func appendBindMarkerIndicesStatement(stmt parser.Statement, markerIndex map[parser.Expression]int, order *[]int) {
	if stmt == nil {
		return
	}
	switch s := stmt.(type) {
	case *parser.InsertStatement:
		if s.JSON {
			appendBindMarkerIndices(s.JSONValue, markerIndex, order)
		} else {
			for _, row := range s.Values {
				for _, expr := range row {
					appendBindMarkerIndices(expr, markerIndex, order)
				}
			}
		}
		if s.Timestamp != nil {
			appendBindMarkerIndices(s.Timestamp, markerIndex, order)
		}
		if s.TTL != nil {
			appendBindMarkerIndices(s.TTL, markerIndex, order)
		}
	case *parser.UpdateStatement:
		if s.TTL != nil {
			appendBindMarkerIndices(s.TTL, markerIndex, order)
		}
		if s.Timestamp != nil {
			appendBindMarkerIndices(s.Timestamp, markerIndex, order)
		}
		for _, assignment := range s.Assigments {
			appendBindMarkerIndices(assignment.Value, markerIndex, order)
		}
		appendBindMarkerIndices(s.Where, markerIndex, order)
		if s.IfCondition != nil {
			appendBindMarkerIndices(s.IfCondition, markerIndex, order)
		}
	case *parser.DeleteStatement:
		if s.Timestamp != nil {
			appendBindMarkerIndices(s.Timestamp, markerIndex, order)
		}
		appendBindMarkerIndices(s.Where, markerIndex, order)
		if s.IfCondition != nil {
			appendBindMarkerIndices(s.IfCondition, markerIndex, order)
		}
	case *parser.BatchStatement:
		for _, entry := range s.Entries {
			appendBindMarkerIndicesStatement(entry, markerIndex, order)
		}
	}
}