// Copyright (c) 2026 Charles KWON OhJun (charleskwonohjun@gmail.com)
// All rights reserved.

// Go-native SQL scan loop for FiveSql2 hot path.
//
// Motivation: FiveSql2 is a PRG-based SQL interpreter. For simple
// "SELECT cols FROM table WHERE cond" queries, the per-row cost is
// dominated by PRG interpreter overhead (AST tree walk, field name
// lookup, workarea switching). Moving just the inner scan loop to Go
// bypasses all that overhead and gets us ~15x speedup for the common
// case while keeping the rest of FiveSql2 untouched.
//
// The SQL engine remains responsible for:
//   - Parsing SQL and building AST
//   - Resolving field names to positions (column binding)
//   - Compiling WHERE expression to pcode (via PcCompile)
//   - GROUP BY, ORDER BY, aggregates (not per-row)
//
// This helper only handles the hot loop:
//   - Full table scan (workarea already positioned)
//   - Per-row WHERE evaluation via ExecPcode
//   - Column extraction via cached field positions
//   - Result array construction

package hbrtl

import (
	"five/hbrdd"
	"five/hbrdd/dbf"
	"five/hbrt"
	"sort"
	"strconv"
	"strings"
)

// SqlScan(aFieldPositions, pcWhere, nLimitHint) → aRows
//
// Scans the current workarea top-to-bottom, evaluates pcWhere per row
// (nil = no filter), collects selected column values into rows.
//
// aFieldPositions: array of 1-based field positions to extract per row.
//                  Resolve once before calling (FieldPos cache is O(1)
//                  but still has PRG → Go call overhead).
// pcWhere:         pcode function pointer from PcCompile, or NIL.
// nLimitHint:      optional early-termination cap. Zero / NIL means
//                  scan the whole table. The caller is responsible for
//                  verifying that the scan order matches the requested
//                  result order (either no ORDER BY, or an index tag
//                  that was already focused by OrdSetFocus).
//
// Returns:
//   Array of rows, each row = Array of field values.
//
// Notes on CHAR trimming: DBF character fields are space-padded. The
// caller decides whether to trim (via a SELECT-list AllTrim wrapper).
// We don't trim here — that's a semantic choice, and callers who need
// raw bytes shouldn't pay for a strings.TrimSpace().
func SqlScan(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	// Parse arguments
	fieldsVal := t.Local(1)
	if !fieldsVal.IsArray() {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}
	fieldsArr := fieldsVal.AsArray().Items
	nFields := len(fieldsArr)

	whereVal := t.Local(2)
	var whereFn *hbrt.PcodeFunc
	if !whereVal.IsNil() {
		if p := whereVal.AsPointer(); p != nil {
			whereFn, _ = p.(*hbrt.PcodeFunc)
		}
	}

	limitHint := 0
	if limitVal := t.Local(3); !limitVal.IsNil() {
		if n := int(limitVal.AsNumInt()); n > 0 {
			limitHint = n
		}
	}

	// Pre-convert field positions to []int (avoid Value->int per row)
	fieldPos := make([]int, nFields)
	for i := 0; i < nFields; i++ {
		fieldPos[i] = int(fieldsArr[i].AsNumInt())
		if fieldPos[i] < 1 {
			fieldPos[i] = 1
		}
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}
	area := wam.Current()
	if area == nil {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	// Type-assert to concrete DBFArea once so the hot loop calls
	// GoTop/EOF/Skip/GetValue directly on *dbf.DBFArea without paying
	// the interface dispatch on every row. Falls back to the generic
	// Area path for non-DBF drivers (rare in FiveSql2 context).
	dbfArea, _ := area.(*dbf.DBFArea)

	// SQLite-inspired: instead of one slice allocation per row, maintain
	// a single flat backing buffer and hand each row a sub-slice into it.
	// This halves allocations (row header + backing → just row header)
	// and keeps row data contiguous in memory for better cache locality.
	//
	// Safety: we cap each sub-slice to exactly nFields via the 3-index
	// slice form (flat[off:end:end]). Any later `append` on an individual
	// row will then trigger a reallocation of that row's backing, so we
	// don't clobber neighboring rows if PRG code mutates via AAdd.
	// Size the initial backing based on the workarea's record count —
	// even if WHERE filters most rows out, over-allocating beats five
	// regrowths of a 200 KB buffer mid-scan.
	estRows := 1024
	if rc, err := area.RecCount(); err == nil && rc > 0 {
		estRows = int(rc)
		if estRows > 1<<20 {
			estRows = 1 << 20
		}
	}
	// LIMIT pushdown: cap the initial backing allocation when the
	// caller guarantees we'll stop after at most `limitHint` rows.
	// Avoids allocating RecCount-sized buffers for `LIMIT 10` queries
	// on million-row tables.
	if limitHint > 0 && limitHint < estRows {
		estRows = limitHint
	}
	rows := make([]hbrt.Value, 0, estRows)
	flat := make([]hbrt.Value, 0, estRows*nFields)
	slab := hbrt.NewArraySlab(estRows)

	// Install the hot-path field getter so PcOpFieldGet in the compiled
	// WHERE predicate bypasses PushSymbol + Function dispatch + the
	// FieldGet RTL's own Frame. The closure captures the concrete
	// DBFArea directly so there's no interface dispatch per access.
	prevFG := t.FastFieldGetter
	if dbfArea != nil {
		t.FastFieldGetter = func(idx int) hbrt.Value {
			v, _ := dbfArea.GetValue(idx - 1)
			return v
		}
	} else {
		t.FastFieldGetter = func(idx int) hbrt.Value {
			v, _ := area.GetValue(idx - 1)
			return v
		}
	}
	defer func() { t.FastFieldGetter = prevFG }()

	// Scan — four specialized loops. Two axes of specialization:
	//
	//   DBF vs generic Area: devirtualization — Go inlines method calls
	//                        on the concrete type but pays an interface
	//                        dispatch on every call of the generic one.
	//
	//   WHERE vs no-WHERE  : branch hoisting — the no-WHERE case is a
	//                        hot full-scan path (SELECT * or similar),
	//                        where even the predictable `whereFn != nil`
	//                        check and the `keep` shadow variable show
	//                        up in pprof.
	//
	// Four combinations = four loop copies. Painful but each row save
	// counts when we're reaching for raw RDD parity.
	// LIMIT pushdown: when limitHint > 0 each loop bails out as soon
	// as we've collected enough rows. The caller guarantees scan order
	// matches result order (no ORDER BY, or matched index tag focused
	// before the call), so clipping early preserves correctness.
	switch {
	case dbfArea != nil && whereFn != nil:
		dbfArea.GoTop()
		for !dbfArea.EOF() {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			if t.GetRetValue().AsBool() {
				off := len(flat)
				end := off + nFields
				if end > cap(flat) {
					flat = append(flat, make([]hbrt.Value, nFields)...)
				} else {
					flat = flat[:end]
				}
				row := flat[off:end:end]
				for i := 0; i < nFields; i++ {
					v, _ := dbfArea.GetValue(fieldPos[i] - 1)
					row[i] = v
				}
				rows = append(rows, slab.WrapNext(row))
				if limitHint > 0 && len(rows) >= limitHint {
					break
				}
			}
			dbfArea.Skip(1)
		}
	case dbfArea != nil:
		// DBF + no WHERE — tightest inner loop
		dbfArea.GoTop()
		for !dbfArea.EOF() {
			off := len(flat)
			end := off + nFields
			if end > cap(flat) {
				flat = append(flat, make([]hbrt.Value, nFields)...)
			} else {
				flat = flat[:end]
			}
			row := flat[off:end:end]
			for i := 0; i < nFields; i++ {
				v, _ := dbfArea.GetValue(fieldPos[i] - 1)
				row[i] = v
			}
			rows = append(rows, slab.WrapNext(row))
			if limitHint > 0 && len(rows) >= limitHint {
				break
			}
			dbfArea.Skip(1)
		}
	case whereFn != nil:
		area.GoTop()
		for !area.EOF() {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			if t.GetRetValue().AsBool() {
				off := len(flat)
				end := off + nFields
				if end > cap(flat) {
					flat = append(flat, make([]hbrt.Value, nFields)...)
				} else {
					flat = flat[:end]
				}
				row := flat[off:end:end]
				for i := 0; i < nFields; i++ {
					v, _ := area.GetValue(fieldPos[i] - 1)
					row[i] = v
				}
				rows = append(rows, slab.WrapNext(row))
				if limitHint > 0 && len(rows) >= limitHint {
					break
				}
			}
			area.Skip(1)
		}
	default:
		area.GoTop()
		for !area.EOF() {
			off := len(flat)
			end := off + nFields
			if end > cap(flat) {
				flat = append(flat, make([]hbrt.Value, nFields)...)
			} else {
				flat = flat[:end]
			}
			row := flat[off:end:end]
			for i := 0; i < nFields; i++ {
				v, _ := area.GetValue(fieldPos[i] - 1)
				row[i] = v
			}
			rows = append(rows, slab.WrapNext(row))
			if limitHint > 0 && len(rows) >= limitHint {
				break
			}
			area.Skip(1)
		}
	}

	t.PushValue(hbrt.MakeArrayFrom(rows))
	t.RetValue()
}

// SqlHashBuild(nFieldPos) → hHash
//
// Scans the current workarea and returns a hash mapping each field
// value (as a string key) to an array of RecNos that have that value.
// Used by FiveSql2's HashJoin: FiveSql2 currently builds this in PRG,
// paying ~40μs per row from class dispatch + hb_HHasKey + AAdd growth.
// 50k rows × 40μs = 2 seconds wasted on what should be a sub-50ms op.
//
// Go-native build goes through *dbf.DBFArea directly and uses a native
// Go `map[string][]int64` which GC's as one unit. Final conversion to
// a Five hash is done once at the end.
func SqlHashBuild(t *hbrt.Thread) {
	t.Frame(1, 0)
	defer t.EndProc()

	nFieldPos := int(t.Local(1).AsNumInt()) - 1
	if nFieldPos < 0 {
		t.PushValue(hbrt.MakeHash())
		t.RetValue()
		return
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok {
		t.PushValue(hbrt.MakeHash())
		t.RetValue()
		return
	}
	area := wam.Current()
	if area == nil {
		t.PushValue(hbrt.MakeHash())
		t.RetValue()
		return
	}

	// Type-assert once so the per-row field reads inline.
	dbfArea, _ := area.(*dbf.DBFArea)

	goMap := make(map[string][]int64, 4096)

	if dbfArea != nil {
		dbfArea.GoTop()
		for !dbfArea.EOF() {
			v, _ := dbfArea.GetValue(nFieldPos)
			key := valueHashKey(v)
			goMap[key] = append(goMap[key], int64(dbfArea.RecNo()))
			dbfArea.Skip(1)
		}
	} else {
		area.GoTop()
		for !area.EOF() {
			v, _ := area.GetValue(nFieldPos)
			key := valueHashKey(v)
			// Generic RecNo via interface
			var rn int64
			if rmgr, ok := area.(interface{ RecNo() uint32 }); ok {
				rn = int64(rmgr.RecNo())
			}
			goMap[key] = append(goMap[key], rn)
			area.Skip(1)
		}
	}

	// Materialize as a Five hash — build Keys/Values slices directly on
	// the HbHash struct, skipping the per-key map-lookup path that PRG
	// hb_HSet would take.
	nKeys := len(goMap)
	keys := make([]hbrt.Value, 0, nKeys)
	vals := make([]hbrt.Value, 0, nKeys)
	order := make([]int, 0, nKeys)
	idx := 0
	for k, recs := range goMap {
		items := make([]hbrt.Value, len(recs))
		for i, r := range recs {
			items[i] = hbrt.MakeNumInt(r)
		}
		keys = append(keys, hbrt.MakeString(k))
		vals = append(vals, hbrt.MakeArrayFrom(items))
		order = append(order, idx)
		idx++
	}
	result := hbrt.MakeHash()
	hh := result.AsHash()
	hh.Keys = keys
	hh.Values = vals
	hh.Order = order

	t.PushValue(result)
	t.RetValue()
}

// valueHashKey converts a Value to a stable string key for Go map use.
// Matches what SqlValToStr does in PRG, but without allocation detours.
func valueHashKey(v hbrt.Value) string {
	switch {
	case v.IsNil():
		return "\x00NIL"
	case v.IsString():
		// Match PRG SqlValToStr: trim trailing spaces so CHAR hash probes
		// compare the same as the equivalent SqlCmpEq call.
		s := v.AsString()
		end := len(s)
		for end > 0 && s[end-1] == ' ' {
			end--
		}
		return s[:end]
	case v.IsNumeric():
		if v.IsNumInt() {
			return strconvItoa(v.AsNumInt())
		}
		return strconvFtoa(v.AsNumDouble())
	case v.IsLogical():
		if v.AsBool() {
			return "T"
		}
		return "F"
	case v.IsDate():
		return strconvItoa(v.AsJulian())
	}
	return ""
}

func strconvItoa(n int64) string {
	// strconv.Itoa is heavy on allocation for small ints — this is the
	// hot path for hash keys so use a tight formatter.
	if n == 0 {
		return "0"
	}
	neg := n < 0
	if neg {
		n = -n
	}
	var buf [20]byte
	i := len(buf)
	for n > 0 {
		i--
		buf[i] = byte('0' + n%10)
		n /= 10
	}
	if neg {
		i--
		buf[i] = '-'
	}
	return string(buf[i:])
}

func strconvFtoa(f float64) string {
	// Only used for non-integer numeric field values (rare in join keys);
	// OK to call into strconv.
	return strconv.FormatFloat(f, 'g', -1, 64)
}

// SqlHashJoin(aOuterFields, aJoinSpecs, aSelectFields) → aRows
//
// Go-native multi-table hash join. Replaces the per-row PRG overhead
// of JoinRecurse → FetchRow → dbSelectArea × N when the query has
// only equi-join conditions and all SELECT columns are plain field refs.
//
// Arguments (all PRG arrays):
//   aJoinSpecs:    array of {nInnerWA, nInnerKeyField, nOuterKeyField}
//                  Each entry describes one join level (1-based field positions).
//                  nOuterKeyField refers to a field in the PREVIOUS level's
//                  table (or the outer for the first entry).
//   aSelectFields: array of {nWA, nFieldPos} — columns to extract per
//                  matched row combination. 1-based field positions.
//   nOuterWA:      workarea number of the outermost (driving) table
//
// Returns: array of rows, each row = array of field values.
//
// The function builds hash tables for each inner level, then walks
// the outer table and probes each level recursively. All field access
// goes through *dbf.DBFArea.GetValue directly — no PRG frame overhead.
func SqlHashJoin(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	joinSpecsVal := t.Local(1)
	selectFieldsVal := t.Local(2)
	nOuterWA := int(t.Local(3).AsNumInt())

	if !joinSpecsVal.IsArray() || !selectFieldsVal.IsArray() {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	// Parse join specs
	jsArr := joinSpecsVal.AsArray().Items
	type joinLevel struct {
		area       *dbf.DBFArea
		innerKey   int // 0-based field index for hash key
		outerKey   int // 0-based field index on parent level
		hashTable  map[string][]uint32 // key → list of RecNos
		parentArea *dbf.DBFArea
	}
	levels := make([]joinLevel, len(jsArr))

	for i, js := range jsArr {
		row := js.AsArray()
		if row == nil || len(row.Items) < 3 {
			t.PushValue(hbrt.MakeArray(0))
			t.RetValue()
			return
		}
		innerWA := int(row.Items[0].AsNumInt())
		innerKeyF := int(row.Items[1].AsNumInt()) - 1
		outerKeyF := int(row.Items[2].AsNumInt()) - 1

		innerArea, _ := wam.AreaAt(uint16(innerWA)).(*dbf.DBFArea)
		if innerArea == nil {
			t.PushValue(hbrt.MakeArray(0))
			t.RetValue()
			return
		}

		// Build hash table for this level
		ht := make(map[string][]uint32, 4096)
		innerArea.GoTop()
		for !innerArea.EOF() {
			v, _ := innerArea.GetValue(innerKeyF)
			key := valueHashKey(v)
			ht[key] = append(ht[key], innerArea.RecNo())
			innerArea.Skip(1)
		}

		levels[i] = joinLevel{
			area:      innerArea,
			innerKey:  innerKeyF,
			outerKey:  outerKeyF,
			hashTable: ht,
		}
	}

	// Set parent area references
	outerArea, _ := wam.AreaAt(uint16(nOuterWA)).(*dbf.DBFArea)
	if outerArea == nil {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}
	for i := range levels {
		if i == 0 {
			levels[i].parentArea = outerArea
		} else {
			levels[i].parentArea = levels[i-1].area
		}
	}

	// Parse select fields
	sfArr := selectFieldsVal.AsArray().Items
	type selectCol struct {
		area     *dbf.DBFArea
		fieldIdx int // 0-based
	}
	selCols := make([]selectCol, len(sfArr))
	for i, sf := range sfArr {
		row := sf.AsArray()
		if row == nil || len(row.Items) < 2 {
			continue
		}
		waNum := int(row.Items[0].AsNumInt())
		fIdx := int(row.Items[1].AsNumInt()) - 1
		if waNum == 0 {
			// Aggregate placeholder — leave area nil, emit 0 per row
			selCols[i] = selectCol{area: nil, fieldIdx: -1}
			continue
		}
		a, _ := wam.AreaAt(uint16(waNum)).(*dbf.DBFArea)
		selCols[i] = selectCol{area: a, fieldIdx: fIdx}
	}

	nFields := len(selCols)
	estRows := 1024
	rows := make([]hbrt.Value, 0, estRows)
	flat := make([]hbrt.Value, 0, estRows*nFields)
	slab := hbrt.NewArraySlab(estRows)

	// Recursive join traversal — iterative via explicit stack
	type frame struct {
		level    int
		matches  []uint32
		matchIdx int
	}

	outerArea.GoTop()
	for !outerArea.EOF() {
		// Start the join chain from the outer row
		stack := []frame{{level: 0, matches: nil, matchIdx: 0}}

		// Get outer key for first level
		outerVal, _ := outerArea.GetValue(levels[0].outerKey)
		outerKey := valueHashKey(outerVal)
		matches, found := levels[0].hashTable[outerKey]
		if !found {
			outerArea.Skip(1)
			continue
		}
		stack[0].matches = matches

		for len(stack) > 0 {
			top := &stack[len(stack)-1]

			if top.matchIdx >= len(top.matches) {
				// Exhausted this level — pop
				stack = stack[:len(stack)-1]
				continue
			}

			// Position the inner area at the current match
			recNo := top.matches[top.matchIdx]
			top.matchIdx++
			levels[top.level].area.GoTo(recNo)

			if top.level == len(levels)-1 {
				// Last level — emit result row
				off := len(flat)
				end := off + nFields
				if end > cap(flat) {
					flat = append(flat, make([]hbrt.Value, nFields)...)
				} else {
					flat = flat[:end]
				}
				row := flat[off:end:end]
				for c := 0; c < nFields; c++ {
					if selCols[c].area != nil {
						v, _ := selCols[c].area.GetValue(selCols[c].fieldIdx)
						row[c] = v
					} else {
						// Aggregate placeholder — 0 for numeric aggregation
						row[c] = hbrt.MakeInt(0)
					}
				}
				rows = append(rows, slab.WrapNext(row))
			} else {
				// Probe next level
				nextLevel := top.level + 1
				probeVal, _ := levels[top.level].area.GetValue(levels[nextLevel].outerKey)
				probeKey := valueHashKey(probeVal)
				nextMatches, found := levels[nextLevel].hashTable[probeKey]
				if found {
					stack = append(stack, frame{
						level:   nextLevel,
						matches: nextMatches,
					})
				}
			}
		}

		outerArea.Skip(1)
	}

	t.PushValue(hbrt.MakeArrayFrom(rows))
	t.RetValue()
}

// SqlOrderBy(aRows, aSortSpec) → aRows (sorted in place)
//
// Go-native sort for SQL ORDER BY. Each aSortSpec element is
// {nColIdx, lDesc} where nColIdx is 1-based and lDesc is .T. for DESC.
// Uses Go's sort.Slice which is ~10-50x faster than PRG ASort with
// block callback for large result sets.
func SqlOrderBy(t *hbrt.Thread) {
	t.Frame(2, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	specVal := t.Local(2)

	if !rowsVal.IsArray() || !specVal.IsArray() {
		t.PushValue(rowsVal)
		t.RetValue()
		return
	}

	rows := rowsVal.AsArray().Items
	specs := specVal.AsArray().Items

	// Per-column sort spec. nullsFirst is derived once from direction
	// and explicit NULLS clause so the hot path is just a bool test.
	// Default (cNulls == ""): NIL is the largest value — NULLs LAST in
	// ASC, NULLs FIRST in DESC. Matches the pre-Go PRG SqlRowCompare.
	// Explicit NULLS FIRST/LAST (SQL:2003) overrides the direction.
	type sortCol struct {
		idx        int
		desc       bool
		nullsFirst bool
	}
	cols := make([]sortCol, len(specs))
	for i, s := range specs {
		arr := s.AsArray()
		if arr == nil || len(arr.Items) < 2 {
			continue
		}
		c := sortCol{
			idx:  int(arr.Items[0].AsNumInt()) - 1,
			desc: arr.Items[1].AsBool(),
		}
		c.nullsFirst = c.desc
		if len(arr.Items) >= 3 {
			switch arr.Items[2].AsString() {
			case "FIRST":
				c.nullsFirst = true
			case "LAST":
				c.nullsFirst = false
			}
		}
		cols[i] = c
	}

	sort.SliceStable(rows, func(a, b int) bool {
		ra := rows[a].AsArray()
		rb := rows[b].AsArray()
		if ra == nil || rb == nil {
			return false
		}
		for _, c := range cols {
			if c.idx < 0 || c.idx >= len(ra.Items) || c.idx >= len(rb.Items) {
				continue
			}
			va := ra.Items[c.idx]
			vb := rb.Items[c.idx]

			// NULL handling follows nullsFirst independent of direction.
			aNil, bNil := va.IsNil(), vb.IsNil()
			if aNil || bNil {
				if aNil && bNil {
					continue
				}
				// exactly one is NIL
				if c.nullsFirst {
					return aNil // NIL side comes first
				}
				return !aNil // non-NIL side comes first
			}

			cmp := compareValuesNonNil(va, vb)
			if cmp == 0 {
				continue
			}
			if c.desc {
				return cmp > 0
			}
			return cmp < 0
		}
		return false
	})

	t.PushValue(rowsVal)
	t.RetValue()
}

// compareValues returns -1, 0, or 1 for two Five Values.
//
// Historical NIL handling (NIL sorts as smallest) is retained here for
// existing callers that are fine with that. New sort paths should treat
// NIL specially based on NULLS FIRST/LAST instead — see compareValuesNonNil
// plus the sortCol.nullsFirst flag in SqlOrderBy.
func compareValues(a, b hbrt.Value) int {
	if a.IsNil() && b.IsNil() {
		return 0
	}
	if a.IsNil() {
		return -1
	}
	if b.IsNil() {
		return 1
	}
	return compareValuesNonNil(a, b)
}

// compareValuesNonNil compares two non-NIL Values. Callers must check
// IsNil() first and apply their own NULL-ordering policy.
func compareValuesNonNil(a, b hbrt.Value) int {
	// Numeric
	if a.IsNumeric() && b.IsNumeric() {
		fa := a.AsNumDouble()
		fb := b.AsNumDouble()
		if fa < fb {
			return -1
		}
		if fa > fb {
			return 1
		}
		return 0
	}

	// String
	if a.IsString() && b.IsString() {
		sa := a.AsString()
		sb := b.AsString()
		if sa < sb {
			return -1
		}
		if sa > sb {
			return 1
		}
		return 0
	}

	// Date
	if a.IsDate() && b.IsDate() {
		ja := a.AsJulian()
		jb := b.AsJulian()
		if ja < jb {
			return -1
		}
		if ja > jb {
			return 1
		}
		return 0
	}

	// Logical
	if a.IsLogical() && b.IsLogical() {
		ba := a.AsBool()
		bb := b.AsBool()
		if ba == bb {
			return 0
		}
		if !ba {
			return -1
		}
		return 1
	}

	// Mixed numeric/string: attempt Harbour-style coercion by reading
	// the string as a numeric. Mirrors the PRG SqlRowCompare branches
	// at TSqlSort.prg:145-148 for legacy DBFs that stored numbers in
	// CHAR columns.
	if a.IsNumeric() && b.IsString() {
		fb := parseLeadingNumeric(b.AsString())
		fa := a.AsNumDouble()
		if fa < fb {
			return -1
		}
		if fa > fb {
			return 1
		}
		return 0
	}
	if a.IsString() && b.IsNumeric() {
		fa := parseLeadingNumeric(a.AsString())
		fb := b.AsNumDouble()
		if fa < fb {
			return -1
		}
		if fa > fb {
			return 1
		}
		return 0
	}

	return 0
}

// parseLeadingNumeric mimics Harbour Val(AllTrim(s)): strips leading /
// trailing spaces, then parses the longest prefix that looks like a
// number. Anything non-numeric yields 0.
func parseLeadingNumeric(s string) float64 {
	i := 0
	for i < len(s) && (s[i] == ' ' || s[i] == '\t') {
		i++
	}
	start := i
	if i < len(s) && (s[i] == '+' || s[i] == '-') {
		i++
	}
	seenDigit, seenDot := false, false
	for i < len(s) {
		c := s[i]
		if c >= '0' && c <= '9' {
			seenDigit = true
			i++
			continue
		}
		if c == '.' && !seenDot {
			seenDot = true
			i++
			continue
		}
		break
	}
	if !seenDigit {
		return 0
	}
	f, err := strconv.ParseFloat(s[start:i], 64)
	if err != nil {
		return 0
	}
	return f
}

// SqlGroupBy(aRows, aGroupColIdx, aAggSpecs) → aResult
//
// Go-native GROUP BY. Builds groups by hashing group-key columns,
// then computes aggregates (SUM/AVG/COUNT/MIN/MAX) per group.
//
// aGroupColIdx: array of 1-based column indices for group key
// aAggSpecs:    array of {nColIdx, cFunc, nArgColIdx}
//               nColIdx: 1-based output position
//               cFunc: "SUM"/"AVG"/"COUNT"/"MIN"/"MAX"
//               nArgColIdx: 1-based column index of the argument (0 for COUNT(*))
func SqlGroupBy(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	groupColsVal := t.Local(2)
	aggSpecsVal := t.Local(3)

	if !rowsVal.IsArray() || !groupColsVal.IsArray() || !aggSpecsVal.IsArray() {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	rows := rowsVal.AsArray().Items
	nRows := len(rows)

	// Parse group column indices
	gcArr := groupColsVal.AsArray().Items
	groupCols := make([]int, len(gcArr))
	for i, v := range gcArr {
		groupCols[i] = int(v.AsNumInt()) - 1
	}

	// Parse aggregate specs
	type aggSpec struct {
		outCol int
		fn     string
		argCol int
	}
	asArr := aggSpecsVal.AsArray().Items
	aggs := make([]aggSpec, len(asArr))
	for i, s := range asArr {
		arr := s.AsArray()
		if arr == nil || len(arr.Items) < 3 {
			continue
		}
		aggs[i] = aggSpec{
			outCol: int(arr.Items[0].AsNumInt()) - 1,
			fn:     arr.Items[1].AsString(),
			argCol: int(arr.Items[2].AsNumInt()) - 1,
		}
	}

	// Build groups
	type groupData struct {
		firstRow int
		count    int
		sums     []float64
		mins     []hbrt.Value
		maxs     []hbrt.Value
		counts   []int
	}

	groups := make(map[string]*groupData)
	groupOrder := make([]string, 0, 256)

	nAggs := len(aggs)

	for i := 0; i < nRows; i++ {
		ra := rows[i].AsArray()
		if ra == nil {
			continue
		}

		// Build group key
		key := ""
		for _, gc := range groupCols {
			if gc >= 0 && gc < len(ra.Items) {
				key += valueHashKey(ra.Items[gc]) + "|"
			}
		}

		gd, exists := groups[key]
		if !exists {
			gd = &groupData{
				firstRow: i,
				sums:     make([]float64, nAggs),
				mins:     make([]hbrt.Value, nAggs),
				maxs:     make([]hbrt.Value, nAggs),
				counts:   make([]int, nAggs),
			}
			groups[key] = gd
			groupOrder = append(groupOrder, key)
		}
		gd.count++

		// Accumulate aggregates
		for ai, ag := range aggs {
			if ag.fn == "COUNT" && ag.argCol < 0 {
				gd.counts[ai]++
				continue
			}
			if ag.argCol >= 0 && ag.argCol < len(ra.Items) {
				v := ra.Items[ag.argCol]
				if !v.IsNil() {
					gd.counts[ai]++
					if v.IsNumeric() {
						gd.sums[ai] += v.AsNumDouble()
					}
					if gd.mins[ai].IsNil() || compareValues(v, gd.mins[ai]) < 0 {
						gd.mins[ai] = v
					}
					if gd.maxs[ai].IsNil() || compareValues(v, gd.maxs[ai]) > 0 {
						gd.maxs[ai] = v
					}
				}
			}
		}
	}

	// Build result rows
	nCols := 0
	if nRows > 0 && rows[0].AsArray() != nil {
		nCols = len(rows[0].AsArray().Items)
	}

	result := make([]hbrt.Value, 0, len(groups))
	for _, key := range groupOrder {
		gd := groups[key]
		srcRow := rows[gd.firstRow].AsArray()
		if srcRow == nil {
			continue
		}

		// Copy group columns from first row
		outItems := make([]hbrt.Value, nCols)
		copy(outItems, srcRow.Items)

		// Fill aggregate values
		for ai, ag := range aggs {
			if ag.outCol >= 0 && ag.outCol < nCols {
				switch ag.fn {
				case "COUNT":
					outItems[ag.outCol] = hbrt.MakeNumInt(int64(gd.counts[ai]))
				case "SUM":
					if gd.counts[ai] > 0 {
						outItems[ag.outCol] = hbrt.MakeDoubleAuto(gd.sums[ai])
					} else {
						outItems[ag.outCol] = hbrt.MakeNil()
					}
				case "AVG":
					if gd.counts[ai] > 0 {
						outItems[ag.outCol] = hbrt.MakeDoubleAuto(gd.sums[ai] / float64(gd.counts[ai]))
					} else {
						outItems[ag.outCol] = hbrt.MakeNil()
					}
				case "MIN":
					outItems[ag.outCol] = gd.mins[ai]
				case "MAX":
					outItems[ag.outCol] = gd.maxs[ai]
				}
			}
		}

		result = append(result, hbrt.MakeArrayFrom(outItems))
	}

	t.PushValue(hbrt.MakeArrayFrom(result))
	t.RetValue()
}

// appendValueHashKey writes the canonical key form of v into sb.
// Same mapping as valueHashKey but without the intermediate string
// allocation; used in tight row-key construction loops.
func appendValueHashKey(sb *strings.Builder, v hbrt.Value) {
	switch {
	case v.IsNil():
		sb.WriteString("\x00NIL")
	case v.IsString():
		s := v.AsString()
		end := len(s)
		for end > 0 && s[end-1] == ' ' {
			end--
		}
		sb.WriteString(s[:end])
	case v.IsNumeric():
		if v.IsNumInt() {
			sb.WriteString(strconvItoa(v.AsNumInt()))
		} else {
			sb.WriteString(strconvFtoa(v.AsNumDouble()))
		}
	case v.IsLogical():
		if v.AsBool() {
			sb.WriteByte('T')
		} else {
			sb.WriteByte('F')
		}
	case v.IsDate():
		sb.WriteString(strconvItoa(v.AsJulian()))
	}
}

// SqlDistinct(aRows) → aRows
//
// Go-native replacement for the PRG TSqlSort:Distinct method. Walks
// aRows once, builds a composite key per row by joining each column's
// SqlValToStr form with '|', and keeps only the first occurrence of
// each key. Output preserves input order (SQL DISTINCT semantic).
//
// Key construction matches PRG SqlValToStr via appendValueHashKey,
// so the dedup decision is byte-for-byte identical to the prior PRG
// hb_HHasKey check — same trailing-space trim on CHAR, same numeric
// formatting, same NIL marker.
//
// Empty / single-row inputs return the input array unchanged.
func SqlDistinct(t *hbrt.Thread) {
	t.Frame(1, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	if !rowsVal.IsArray() {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	rows := rowsVal.AsArray().Items
	nRows := len(rows)
	if nRows < 2 {
		t.PushValue(rowsVal)
		t.RetValue()
		return
	}

	seen := make(map[string]struct{}, nRows)
	result := make([]hbrt.Value, 0, nRows)
	var sb strings.Builder

	for i := 0; i < nRows; i++ {
		ra := rows[i].AsArray()
		if ra == nil {
			continue
		}
		sb.Reset()
		for _, item := range ra.Items {
			appendValueHashKey(&sb, item)
			sb.WriteByte('|')
		}
		if _, dup := seen[sb.String()]; dup {
			continue
		}
		seen[sb.String()] = struct{}{}
		result = append(result, rows[i])
	}

	t.PushValue(hbrt.MakeArrayFrom(result))
	t.RetValue()
}

// SqlUnionDistinct(aLeft, aRight) → aMerged
//
// Streaming DISTINCT for the SQL UNION operator. Builds a hash set
// keyed on each row's canonical composite key (same format used by
// SqlDistinct) over aLeft, then walks aRight once pushing only rows
// whose key isn't already seen. Replaces the PRG idiom of appending
// both arrays in full then calling SqlDistinct, which materialised
// the intermediate merged array and walked every row twice — once
// to append, once to rebuild the dedup hash.
//
// Output matches `aLeft ++ filter(aRight, unseen)`: left rows stay
// first and in their original order, right rows are appended in
// their original order after dedup against left + each other.
// Same byte-for-byte dedup decision as SqlDistinct.
func SqlUnionDistinct(t *hbrt.Thread) {
	t.Frame(2, 0)
	defer t.EndProc()

	leftVal := t.Local(1)
	rightVal := t.Local(2)
	if !leftVal.IsArray() {
		if rightVal.IsArray() {
			t.PushValue(rightVal)
		} else {
			t.PushValue(hbrt.MakeArray(0))
		}
		t.RetValue()
		return
	}
	leftRows := leftVal.AsArray().Items
	var rightRows []hbrt.Value
	if rightVal.IsArray() {
		rightRows = rightVal.AsArray().Items
	}

	nL := len(leftRows)
	nR := len(rightRows)
	seen := make(map[string]struct{}, nL+nR)
	out := make([]hbrt.Value, 0, nL+nR)
	var sb strings.Builder

	keyOf := func(v hbrt.Value) string {
		sb.Reset()
		if ra := v.AsArray(); ra != nil {
			for _, item := range ra.Items {
				appendValueHashKey(&sb, item)
				sb.WriteByte('|')
			}
		}
		return sb.String()
	}

	for i := 0; i < nL; i++ {
		if leftRows[i].AsArray() == nil {
			continue
		}
		k := keyOf(leftRows[i])
		if _, dup := seen[k]; dup {
			continue
		}
		seen[k] = struct{}{}
		out = append(out, leftRows[i])
	}
	for i := 0; i < nR; i++ {
		if rightRows[i].AsArray() == nil {
			continue
		}
		k := keyOf(rightRows[i])
		if _, dup := seen[k]; dup {
			continue
		}
		seen[k] = struct{}{}
		out = append(out, rightRows[i])
	}

	t.PushValue(hbrt.MakeArrayFrom(out))
	t.RetValue()
}

// SqlBuildSubCacheKey(nId, aValues) → cKey
//
// Builds the composite cache key for a correlated subquery:
//   "<nId>@<key(v1)>|<key(v2)>|..."
// where key(v) uses the canonical appendValueHashKey encoding (same
// as SqlDistinct / SqlWindow hash keys). Replaces the per-outer-row
// PRG loop of `hb_ntos(nId) + "@" + SqlValToStr(v1) + "|" + ...` which
// allocated a fresh string on every concatenation and paid the PRG
// dispatch on every SqlValToStr / ValType probe. For correlated
// subqueries over large outer tables this was the dominant cost on
// cache hits — where the point of the cache is to be cheap.
func SqlBuildSubCacheKey(t *hbrt.Thread) {
	t.Frame(2, 0)
	defer t.EndProc()

	nId := t.Local(1).AsNumInt()
	valsArg := t.Local(2)

	var sb strings.Builder
	sb.WriteString(strconvItoa(nId))
	sb.WriteByte('@')
	if valsArg.IsArray() {
		for _, v := range valsArg.AsArray().Items {
			appendValueHashKey(&sb, v)
			sb.WriteByte('|')
		}
	}
	t.RetString(sb.String())
}

// SqlComputeAggSimple(aGR, nCol, cFunc) → xResult
//
// Fast path for COUNT / SUM / AVG / MIN / MAX when the argument is a
// plain column reference (already resolved to a 1-based index by the
// caller). Replaces the PRG inner loop of TSqlAgg:ComputeAgg, which
// walks aGR rows and performs type-aware accumulation per iteration.
//
// The caller gates on: cFunc ∈ {COUNT,SUM,AVG,MIN,MAX}, argument is
// ND_COL, column index resolved > 0. Complex-argument aggregates
// (CASE / BIN / UDF) and GROUP_CONCAT/STRING_AGG stay in PRG.
//
// Arguments:
//   aGR   : array of group rows (each row is an array)
//   nCol  : 1-based column index (0 → COUNT treats every row; others → 0)
//   cFunc : uppercase function name
//
// Returns:
//   COUNT: non-NIL count, or Len(aGR) when nCol<=0 (COUNT(*))
//   SUM  : double sum, NIL if no non-NIL values (SQL NULL-safe)
//   AVG  : double sum/count, NIL if empty
//   MIN  : smallest value via type-aware compare, NIL if empty
//   MAX  : largest value via type-aware compare, NIL if empty
//   other: NIL (caller falls back to PRG)
func SqlComputeAggSimple(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	grVal := t.Local(1)
	if !grVal.IsArray() {
		t.RetNil()
		return
	}
	gr := grVal.AsArray().Items
	nCol := int(t.Local(2).AsNumInt()) - 1 // 0-based
	fn := t.Local(3).AsString()

	if fn == "COUNT" && nCol < 0 {
		t.RetInt(int64(len(gr)))
		return
	}
	if nCol < 0 {
		t.RetNil()
		return
	}

	count := 0
	sum := 0.0
	var minV, maxV hbrt.Value
	haveMin := false

	for i := 0; i < len(gr); i++ {
		ra := gr[i].AsArray()
		if ra == nil || nCol >= len(ra.Items) {
			continue
		}
		v := ra.Items[nCol]
		if v.IsNil() {
			continue
		}
		count++
		if v.IsNumeric() {
			sum += v.AsNumDouble()
		}
		if !haveMin {
			minV = v
			maxV = v
			haveMin = true
			continue
		}
		if compareValuesNonNil(v, minV) < 0 {
			minV = v
		}
		if compareValuesNonNil(v, maxV) > 0 {
			maxV = v
		}
	}

	switch fn {
	case "COUNT":
		t.RetInt(int64(count))
	case "SUM":
		if count == 0 {
			t.RetNil()
		} else {
			t.RetVal(hbrt.MakeDoubleAuto(sum))
		}
	case "AVG":
		if count == 0 {
			t.RetNil()
		} else {
			t.RetVal(hbrt.MakeDoubleAuto(sum / float64(count)))
		}
	case "MIN":
		if !haveMin {
			t.RetNil()
		} else {
			t.RetVal(minV)
		}
	case "MAX":
		if !haveMin {
			t.RetNil()
		} else {
			t.RetVal(maxV)
		}
	default:
		t.RetNil()
	}
}

// SqlGroupRows(aRows, aGroupColIdx) → aGroupedRows
//
// Groups rows (values, not indices) by their GROUP BY column values,
// preserving first-seen order. Replaces the PRG hot loop in
// TSqlAgg:GroupBy:
//
//    FOR i := 1 TO Len( aRows )
//       cKey := ""
//       FOR j := 1 TO Len( aGroupBy )
//          cKey += SqlValToStr( aRows[ i ][ aGroupIdx[ j ] ] ) + "|"
//       NEXT
//       IF ! hb_HHasKey( hGroups, cKey )
//          hGroups[ cKey ] := {}
//       ENDIF
//       AAdd( hGroups[ cKey ], aRows[ i ] )
//    NEXT
//
// Aggregate computation + HAVING evaluation stay in PRG (too many
// expression kinds to port cleanly); this RTL only collapses the
// grouping step — the dominant per-row boundary-crossing cost.
//
// Returns: array of groups, each group is an array of original rows
// (by reference — no copy). First-seen group key order.
func SqlGroupRows(t *hbrt.Thread) {
	t.Frame(2, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	colsVal := t.Local(2)
	if !rowsVal.IsArray() {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	rows := rowsVal.AsArray().Items
	nRows := len(rows)

	var groupCols []int
	if colsVal.IsArray() {
		colsArr := colsVal.AsArray().Items
		groupCols = make([]int, len(colsArr))
		for i, v := range colsArr {
			groupCols[i] = int(v.AsNumInt()) - 1
		}
	}

	// No GROUP BY columns → single group containing all rows. Matches
	// PRG semantic where HAVING or aggregate query with no GROUP BY
	// still aggregates over the whole result.
	if len(groupCols) == 0 {
		all := make([]hbrt.Value, nRows)
		copy(all, rows)
		t.PushValue(hbrt.MakeArrayFrom([]hbrt.Value{
			hbrt.MakeArrayFrom(all),
		}))
		t.RetValue()
		return
	}

	order := make([]string, 0, 16)
	groups := make(map[string][]hbrt.Value, 16)
	var sb strings.Builder

	for i := 0; i < nRows; i++ {
		ra := rows[i].AsArray()
		if ra == nil {
			continue
		}
		sb.Reset()
		for _, c := range groupCols {
			if c >= 0 && c < len(ra.Items) {
				appendValueHashKey(&sb, ra.Items[c])
			}
			sb.WriteByte('|')
		}
		key := sb.String()
		if _, ok := groups[key]; !ok {
			groups[key] = make([]hbrt.Value, 0, 8)
			order = append(order, key)
		}
		groups[key] = append(groups[key], rows[i])
	}

	out := make([]hbrt.Value, len(order))
	for oi, key := range order {
		out[oi] = hbrt.MakeArrayFrom(groups[key])
	}
	t.PushValue(hbrt.MakeArrayFrom(out))
	t.RetValue()
}

// SqlEvalHaving(xHaving, aNewRow, aCols, aGR, aFN, aParams) → {lOk, lPass}
//
// Go-native tree walker for HAVING clause evaluation, mirroring
// PRG TSqlAgg:EvalHavingExpr. Returns a 2-element array:
//   [1] lOk:   .T. if fully handled in Go, .F. to fall back to PRG
//   [2] lPass: truthiness when handled
//
// Supported nodes: ND_LIT, ND_NIL, ND_COL (lookup in aCols / aFN),
// ND_FN (COUNT/SUM/AVG/MIN/MAX with plain column args), ND_BIN
// (AND/OR/comparison), ND_UNI (NOT/-). Anything unsupported → returns
// {.F., .F.} so PRG takes over.
//
// Aggregates inside HAVING are recomputed per group using the same
// sqlComputeAggSimple path as the SELECT list. Redundant vs SELECT-
// list aggregate compute, but simple and bounded (HAVING is usually
// a single comparison).
func SqlEvalHaving(t *hbrt.Thread) {
	t.Frame(6, 0)
	defer t.EndProc()

	xE := t.Local(1)
	aNewRow := t.Local(2)
	aCols := t.Local(3)
	aGR := t.Local(4)
	aFN := t.Local(5)

	ctx := &havingCtx{
		aNewRow: aNewRow,
		aCols:   aCols,
		aGR:     aGR,
		aFN:     aFN,
	}
	ok, v := ctx.eval(xE)
	result := hbrt.MakeArray(2)
	arr := result.AsArray()
	arr.Items[0] = hbrt.MakeBool(ok)
	if ok {
		arr.Items[1] = hbrt.MakeBool(havingIsTrue(v))
	} else {
		arr.Items[1] = hbrt.MakeBool(false)
	}
	t.PushValue(result)
	t.RetValue()
}

type havingCtx struct {
	aNewRow hbrt.Value
	aCols   hbrt.Value
	aGR     hbrt.Value
	aFN     hbrt.Value
}

// eval walks the HAVING AST. Returns (ok, value). ok=false means
// "encountered unsupported node, caller must fall back to PRG."
func (c *havingCtx) eval(xE hbrt.Value) (bool, hbrt.Value) {
	if xE.IsNil() {
		return true, hbrt.MakeNil()
	}
	arr := xE.AsArray()
	if arr == nil || len(arr.Items) < 2 {
		return false, hbrt.MakeNil()
	}
	kind := int(arr.Items[0].AsNumInt())

	switch kind {
	case ndLit:
		return true, arr.Items[1]
	case ndNil:
		return true, hbrt.MakeNil()

	case ndCol:
		// Look up in aCols by upper-cased name, return aNewRow[i]
		name := arr.Items[1].AsString()
		if idx := strings.Index(name, "."); idx >= 0 {
			name = name[idx+1:]
		}
		name = strings.ToUpper(name)
		colsArr := c.aCols.AsArray()
		rowArr := c.aNewRow.AsArray()
		if colsArr != nil && rowArr != nil {
			for i, col := range colsArr.Items {
				ca := col.AsArray()
				if ca == nil || len(ca.Items) < 2 {
					continue
				}
				if strings.EqualFold(ca.Items[1].AsString(), name) && i < len(rowArr.Items) {
					return true, rowArr.Items[i]
				}
			}
		}
		// Fallback: lookup in aFN → aGR[0]
		fnArr := c.aFN.AsArray()
		if fnArr != nil {
			for i, n := range fnArr.Items {
				if strings.EqualFold(n.AsString(), name) {
					grArr := c.aGR.AsArray()
					if grArr != nil && len(grArr.Items) > 0 {
						firstRow := grArr.Items[0].AsArray()
						if firstRow != nil && i < len(firstRow.Items) {
							return true, firstRow.Items[i]
						}
					}
				}
			}
		}
		return true, hbrt.MakeNil()

	case ndFn:
		return c.evalAgg(arr)

	case ndBin:
		if len(arr.Items) < 4 {
			return false, hbrt.MakeNil()
		}
		op := arr.Items[1].AsString()
		// Short-circuit for AND/OR
		if op == "AND" {
			okL, vL := c.eval(arr.Items[2])
			if !okL {
				return false, hbrt.MakeNil()
			}
			if !havingIsTrue(vL) {
				return true, hbrt.MakeBool(false)
			}
			okR, vR := c.eval(arr.Items[3])
			if !okR {
				return false, hbrt.MakeNil()
			}
			return true, hbrt.MakeBool(havingIsTrue(vR))
		}
		if op == "OR" {
			okL, vL := c.eval(arr.Items[2])
			if !okL {
				return false, hbrt.MakeNil()
			}
			if havingIsTrue(vL) {
				return true, hbrt.MakeBool(true)
			}
			okR, vR := c.eval(arr.Items[3])
			if !okR {
				return false, hbrt.MakeNil()
			}
			return true, hbrt.MakeBool(havingIsTrue(vR))
		}
		okL, vL := c.eval(arr.Items[2])
		if !okL {
			return false, hbrt.MakeNil()
		}
		okR, vR := c.eval(arr.Items[3])
		if !okR {
			return false, hbrt.MakeNil()
		}
		switch op {
		case "=", "==":
			return true, hbrt.MakeBool(sqlCmpEq(vL, vR))
		case "<>", "!=":
			return true, hbrt.MakeBool(!sqlCmpEq(vL, vR))
		case "<":
			return true, hbrt.MakeBool(sqlCmpLt(vL, vR))
		case ">":
			return true, hbrt.MakeBool(sqlCmpLt(vR, vL))
		case "<=":
			return true, hbrt.MakeBool(sqlCmpEq(vL, vR) || sqlCmpLt(vL, vR))
		case ">=":
			return true, hbrt.MakeBool(sqlCmpEq(vL, vR) || sqlCmpLt(vR, vL))
		}
		return false, hbrt.MakeNil()

	case ndUni:
		if len(arr.Items) < 3 {
			return false, hbrt.MakeNil()
		}
		op := arr.Items[1].AsString()
		okX, vX := c.eval(arr.Items[2])
		if !okX {
			return false, hbrt.MakeNil()
		}
		if op == "NOT" {
			return true, hbrt.MakeBool(!havingIsTrue(vX))
		}
		return false, hbrt.MakeNil()
	}

	return false, hbrt.MakeNil()
}

// evalAgg runs a simple aggregate (COUNT/SUM/AVG/MIN/MAX) on aGR when
// the argument is a plain column (or "*" for COUNT). Anything else
// triggers a PRG fallback.
func (c *havingCtx) evalAgg(arr *hbrt.HbArray) (bool, hbrt.Value) {
	if len(arr.Items) < 3 {
		return false, hbrt.MakeNil()
	}
	name := strings.ToUpper(arr.Items[1].AsString())
	switch name {
	case "COUNT", "SUM", "AVG", "MIN", "MAX":
	default:
		return false, hbrt.MakeNil()
	}

	// Parse first arg to find column index (0 → COUNT(*))
	argsArr := arr.Items[2].AsArray()
	if argsArr == nil || len(argsArr.Items) == 0 {
		if name == "COUNT" {
			grArr := c.aGR.AsArray()
			if grArr == nil {
				return true, hbrt.MakeNumInt(0)
			}
			return true, hbrt.MakeNumInt(int64(len(grArr.Items)))
		}
		return false, hbrt.MakeNil()
	}
	firstArg := argsArr.Items[0].AsArray()
	if firstArg == nil || len(firstArg.Items) < 2 {
		return false, hbrt.MakeNil()
	}
	argKind := int(firstArg.Items[0].AsNumInt())
	if argKind != ndCol {
		return false, hbrt.MakeNil()
	}
	colName := firstArg.Items[1].AsString()
	if colName == "*" {
		if name == "COUNT" {
			grArr := c.aGR.AsArray()
			if grArr == nil {
				return true, hbrt.MakeNumInt(0)
			}
			return true, hbrt.MakeNumInt(int64(len(grArr.Items)))
		}
		return false, hbrt.MakeNil()
	}

	// Resolve column name → index in aFN
	if idx := strings.Index(colName, "."); idx >= 0 {
		colName = colName[idx+1:]
	}
	colName = strings.ToUpper(colName)
	fnArr := c.aFN.AsArray()
	nCol := -1
	if fnArr != nil {
		for i, n := range fnArr.Items {
			if strings.EqualFold(n.AsString(), colName) {
				nCol = i
				break
			}
		}
	}
	if nCol < 0 {
		return false, hbrt.MakeNil()
	}

	grArr := c.aGR.AsArray()
	if grArr == nil {
		return true, hbrt.MakeNumInt(0)
	}

	// Run the simple aggregate loop (mirrors SqlComputeAggSimple).
	count := 0
	sum := 0.0
	var minV, maxV hbrt.Value
	haveAny := false
	for _, rowVal := range grArr.Items {
		ra := rowVal.AsArray()
		if ra == nil || nCol >= len(ra.Items) {
			continue
		}
		v := ra.Items[nCol]
		if v.IsNil() {
			continue
		}
		count++
		if v.IsNumeric() {
			sum += v.AsNumDouble()
		}
		if !haveAny {
			minV = v
			maxV = v
			haveAny = true
			continue
		}
		if compareValuesNonNil(v, minV) < 0 {
			minV = v
		}
		if compareValuesNonNil(v, maxV) > 0 {
			maxV = v
		}
	}

	switch name {
	case "COUNT":
		return true, hbrt.MakeNumInt(int64(count))
	case "SUM":
		if count == 0 {
			return true, hbrt.MakeNil()
		}
		return true, hbrt.MakeDoubleAuto(sum)
	case "AVG":
		if count == 0 {
			return true, hbrt.MakeNil()
		}
		return true, hbrt.MakeDoubleAuto(sum / float64(count))
	case "MIN":
		if !haveAny {
			return true, hbrt.MakeNil()
		}
		return true, minV
	case "MAX":
		if !haveAny {
			return true, hbrt.MakeNil()
		}
		return true, maxV
	}
	return false, hbrt.MakeNil()
}

// havingIsTrue mirrors PRG SqlIsTrue — NIL/0/empty-string all false.
func havingIsTrue(v hbrt.Value) bool {
	return sqlIsTrue(v)
}

// SqlWindowPartitions(aRows, aPartColIdx) → aPartitions
//
// Groups row indices by their PARTITION BY column values, preserving
// first-seen order. Replaces the PRG hot loop in
// TSqlExecutor:ApplyWindowFunctions that per row does:
//
//    cPartKey := ""
//    FOR j := 1 TO Len( aPartBy )
//       cPartKey += SqlValToStr( aRows[ i ][ aPartCol[ j ] ] ) + "|"
//    NEXT
//    IF ! hb_HHasKey( hPartitions, cPartKey )
//       hPartitions[ cPartKey ] := {}
//    ENDIF
//    AAdd( hPartitions[ cPartKey ], i )
//
// Key construction reuses the shared valueHashKey → matches the PRG
// SqlValToStr equivalence classes byte-for-byte so partition
// identity is unchanged.
//
// Arguments:
//   aRows:        result rows (array of arrays)
//   aPartColIdx:  1-based column indices for partition key (empty
//                 array → single "all rows" partition)
//
// Returns:
//   Array of partitions. Each partition is an array of 1-based
//   row indices into aRows, in first-seen order inside the partition.
//   Partitions themselves are also in first-seen order of their key.
//
// Called at most once per window column per query — amortizes the
// Go↔PRG boundary cost across N·M operations.
func SqlWindowPartitions(t *hbrt.Thread) {
	t.Frame(2, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	colsVal := t.Local(2)
	if !rowsVal.IsArray() {
		t.PushValue(hbrt.MakeArray(0))
		t.RetValue()
		return
	}

	rows := rowsVal.AsArray().Items
	nRows := len(rows)

	var partCols []int
	if colsVal.IsArray() {
		colsArr := colsVal.AsArray().Items
		partCols = make([]int, len(colsArr))
		for i, v := range colsArr {
			partCols[i] = int(v.AsNumInt()) - 1
		}
	}

	// Fast path: no PARTITION BY → one partition holding all row indices.
	if len(partCols) == 0 {
		idxs := make([]hbrt.Value, nRows)
		for i := 0; i < nRows; i++ {
			idxs[i] = hbrt.MakeInt(i + 1)
		}
		t.PushValue(hbrt.MakeArrayFrom([]hbrt.Value{
			hbrt.MakeArrayFrom(idxs),
		}))
		t.RetValue()
		return
	}

	// Preserve first-seen order via parallel slice + map.
	order := make([]string, 0, 16)
	groups := make(map[string][]int, 16)
	var sb strings.Builder

	for i := 0; i < nRows; i++ {
		ra := rows[i].AsArray()
		if ra == nil {
			continue
		}
		sb.Reset()
		for _, c := range partCols {
			if c >= 0 && c < len(ra.Items) {
				appendValueHashKey(&sb, ra.Items[c])
			}
			sb.WriteByte('|')
		}
		key := sb.String()
		if _, ok := groups[key]; !ok {
			groups[key] = make([]int, 0, 8)
			order = append(order, key)
		}
		groups[key] = append(groups[key], i+1) // 1-based for PRG
	}

	out := make([]hbrt.Value, len(order))
	for oi, key := range order {
		g := groups[key]
		idxs := make([]hbrt.Value, len(g))
		for j, n := range g {
			idxs[j] = hbrt.MakeInt(n)
		}
		out[oi] = hbrt.MakeArrayFrom(idxs)
	}
	t.PushValue(hbrt.MakeArrayFrom(out))
	t.RetValue()
}

// SqlWindowSortPartition(aRows, aPartIdx, aSortSpec) → aPartIdx
//
// Sorts a partition (array of 1-based row indices into aRows) by the
// ORDER BY spec.  aSortSpec is an array of {nColIdx, lDesc} pairs
// with 1-based column indices. Mutates aPartIdx in place and returns
// it for chainability.
//
// Matches PRG SqlWinRowCmp semantics byte-for-byte:
//   - NIL sorts as the largest value (NULLs last in ASC, NULLs first
//     in DESC) — consistent with the #3 migration for ORDER BY.
//   - Mixed-type comparison: same ValType only; otherwise treated
//     equal on that column (moves to next sort key).
//   - Stable sort so the first-seen partition order (from SqlWindow-
//     Partitions) carries through equal-value ties.
//
// Replaces ASort(aPartIdx,,, {|a,b| SqlWinRowCmp(...) < 0}) — the PRG
// block is invoked O(N log N) times per partition; Go sort skips that
// bridge and uses pre-resolved column indices.
func SqlWindowSortPartition(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	idxVal := t.Local(2)
	specVal := t.Local(3)

	if !rowsVal.IsArray() || !idxVal.IsArray() || !specVal.IsArray() {
		t.PushValue(idxVal)
		t.RetValue()
		return
	}

	rows := rowsVal.AsArray().Items
	idxs := idxVal.AsArray().Items
	specs := specVal.AsArray().Items

	type sortCol struct {
		idx  int
		desc bool
	}
	cols := make([]sortCol, 0, len(specs))
	for _, s := range specs {
		arr := s.AsArray()
		if arr == nil || len(arr.Items) < 2 {
			continue
		}
		cols = append(cols, sortCol{
			idx:  int(arr.Items[0].AsNumInt()) - 1,
			desc: arr.Items[1].AsBool(),
		})
	}
	if len(cols) == 0 || len(idxs) < 2 {
		t.PushValue(idxVal)
		t.RetValue()
		return
	}

	sort.SliceStable(idxs, func(ai, bi int) bool {
		ra := rows[int(idxs[ai].AsNumInt())-1].AsArray()
		rb := rows[int(idxs[bi].AsNumInt())-1].AsArray()
		if ra == nil || rb == nil {
			return false
		}
		for _, c := range cols {
			if c.idx < 0 || c.idx >= len(ra.Items) || c.idx >= len(rb.Items) {
				continue
			}
			va := ra.Items[c.idx]
			vb := rb.Items[c.idx]

			// NIL handling: NIL is the largest value.
			aNil, bNil := va.IsNil(), vb.IsNil()
			if aNil && bNil {
				continue
			}
			if aNil {
				// a > b — in DESC, a comes first (less-than = true)
				return c.desc
			}
			if bNil {
				// b > a — in ASC, a comes first (less-than = true)
				return !c.desc
			}

			// Only compare if same type, otherwise skip (PRG semantic).
			if va.Type() != vb.Type() {
				continue
			}
			cmp := compareValuesNonNil(va, vb)
			if cmp == 0 {
				continue
			}
			if c.desc {
				return cmp > 0
			}
			return cmp < 0
		}
		return false
	})

	t.PushValue(idxVal)
	t.RetValue()
}

// SqlWindowAssignRank(aRows, aPartIdx, aSortSpec, nColIdx, cFunc) → NIL
//
// Assigns ROW_NUMBER / RANK / DENSE_RANK values to each row in a
// sorted partition. Replaces the PRG loop in ApplyWindowFunctions:
//
//    FOR k := 1 TO Len( aPartIdx )
//       IF ! SqlWinRowsEqual( aRows, aPartIdx[k], aPartIdx[k-1], ... )
//          nRank := k  (or nRank++)
//       ENDIF
//       aRows[ aPartIdx[k] ][ nColIdx ] := nRank
//    NEXT
//
// Collapses the per-row SqlWinRowsEqual + PRG indexing cost. aSortSpec
// is the same sort spec (array of {nCol, lDesc}) that
// SqlWindowSortPartition already consumes — caller reuses it without
// re-resolving column indices.
//
// Arguments:
//   aRows      : full result row set
//   aPartIdx   : partition (array of 1-based row indices, sorted)
//   aSortSpec  : ORDER BY spec; only column indices matter for
//                equality check (direction unused). Empty spec means
//                no ORDER BY → ROW_NUMBER semantic for RANK/DENSE too.
//   nColIdx    : 1-based output column index to receive the rank value
//   cFunc      : "ROW_NUMBER" | "RANK" | "DENSE_RANK"
//
// Mutates aRows in place. Returns NIL.
func SqlWindowAssignRank(t *hbrt.Thread) {
	t.Frame(5, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	idxVal := t.Local(2)
	specVal := t.Local(3)
	nColIdx := int(t.Local(4).AsNumInt()) - 1
	fn := t.Local(5).AsString()

	if !rowsVal.IsArray() || !idxVal.IsArray() {
		t.RetNil()
		return
	}
	rows := rowsVal.AsArray().Items
	idxs := idxVal.AsArray().Items
	if len(idxs) == 0 || nColIdx < 0 {
		t.RetNil()
		return
	}

	// Unpack sort spec — we only need column indices for equality check.
	var sortCols []int
	if specVal.IsArray() {
		specs := specVal.AsArray().Items
		sortCols = make([]int, 0, len(specs))
		for _, s := range specs {
			arr := s.AsArray()
			if arr == nil || len(arr.Items) < 2 {
				continue
			}
			sortCols = append(sortCols, int(arr.Items[0].AsNumInt())-1)
		}
	}

	// Helper: does row i equal row j on all sort columns? Reuses the
	// compareValuesNonNil path; NIL matches NIL, NIL ≠ non-NIL.
	rowsEqual := func(ri, rj int) bool {
		ra := rows[ri].AsArray()
		rb := rows[rj].AsArray()
		if ra == nil || rb == nil {
			return false
		}
		for _, c := range sortCols {
			if c < 0 || c >= len(ra.Items) || c >= len(rb.Items) {
				continue
			}
			va := ra.Items[c]
			vb := rb.Items[c]
			aNil, bNil := va.IsNil(), vb.IsNil()
			if aNil != bNil {
				return false
			}
			if aNil && bNil {
				continue
			}
			if compareValuesNonNil(va, vb) != 0 {
				return false
			}
		}
		return true
	}

	// Compute rank per row and write to aRows[ idx ][ nColIdx ].
	writeRank := func(rowIdx, rank int) {
		if rowIdx < 0 || rowIdx >= len(rows) {
			return
		}
		ra := rows[rowIdx].AsArray()
		if ra == nil || nColIdx >= len(ra.Items) {
			return
		}
		ra.Items[nColIdx] = hbrt.MakeNumInt(int64(rank))
	}

	switch fn {
	case "ROW_NUMBER":
		for k, ri := range idxs {
			writeRank(int(ri.AsNumInt())-1, k+1)
		}
	case "RANK":
		// Same value group → same rank, then jump to k+1.
		rank := 1
		prevRowIdx := int(idxs[0].AsNumInt()) - 1
		writeRank(prevRowIdx, rank)
		for k := 1; k < len(idxs); k++ {
			curIdx := int(idxs[k].AsNumInt()) - 1
			if len(sortCols) == 0 || !rowsEqual(curIdx, prevRowIdx) {
				rank = k + 1
			}
			writeRank(curIdx, rank)
			prevRowIdx = curIdx
		}
	case "DENSE_RANK":
		rank := 1
		prevRowIdx := int(idxs[0].AsNumInt()) - 1
		writeRank(prevRowIdx, rank)
		for k := 1; k < len(idxs); k++ {
			curIdx := int(idxs[k].AsNumInt()) - 1
			if len(sortCols) == 0 || !rowsEqual(curIdx, prevRowIdx) {
				rank++
			}
			writeRank(curIdx, rank)
			prevRowIdx = curIdx
		}
	}

	t.RetNil()
}

// SqlBulkUpdate(aFieldPositions, pcWhere, aValuePcodes) → nAffected
//
// Go-native replacement for the PRG UPDATE scan loop:
//
//    dbGoTop()
//    WHILE ! Eof()
//       IF xWhere == NIL .OR. SqlIsTrue( ::EvalExpr( xWhere ) )
//          IF dbRLock( RecNo() )
//             FOR i := 1 TO Len( aSet )
//                FieldPut( nFPos[i], ::EvalExpr( aSet[i][2] ) )
//             NEXT
//             dbRUnlock( RecNo() )
//             nAffected++
//          ENDIF
//       ENDIF
//       dbSkip()
//    ENDDO
//    dbCommit()
//
// Collapses: Eof + Skip + RLock + FieldPut×N + RUnlock cross the Go
// boundary once per record via the PRG VM. Moving to one RTL call
// keeps the scan inside Go and uses compiled pcode for both WHERE
// and every SET value expression.
//
// Arguments:
//   aFieldPositions: 1-based field positions to write (aligned with aValuePcodes)
//   pcWhere:         compiled WHERE pcode (NIL = unconditional update)
//   aValuePcodes:    compiled pcode per SET value expression
//
// Caller must ensure every SET value expression compiled successfully;
// any nil slot in aValuePcodes is silently skipped (leaves field unchanged).
//
// Txn caveat: does not call ::oTxn:LogRecord per row — caller is
// responsible for ensuring no active transaction when invoking this
// fast path, else undo semantics break.
func SqlBulkUpdate(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	fieldsVal := t.Local(1)
	whereVal := t.Local(2)
	pcodesVal := t.Local(3)

	if !fieldsVal.IsArray() || !pcodesVal.IsArray() {
		t.RetInt(0)
		return
	}
	fieldsArr := fieldsVal.AsArray().Items
	pcodesArr := pcodesVal.AsArray().Items
	nSets := len(fieldsArr)
	if nSets != len(pcodesArr) || nSets == 0 {
		t.RetInt(0)
		return
	}

	fieldPos := make([]int, nSets)
	for i := 0; i < nSets; i++ {
		fieldPos[i] = int(fieldsArr[i].AsNumInt()) - 1
	}
	valuePcodes := make([]*hbrt.PcodeFunc, nSets)
	for i, pv := range pcodesArr {
		if p := pv.AsPointer(); p != nil {
			if pc, ok := p.(*hbrt.PcodeFunc); ok {
				valuePcodes[i] = pc
			}
		}
	}

	var whereFn *hbrt.PcodeFunc
	if !whereVal.IsNil() {
		if p := whereVal.AsPointer(); p != nil {
			whereFn, _ = p.(*hbrt.PcodeFunc)
		}
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok {
		t.RetInt(0)
		return
	}
	area := wam.Current()
	if area == nil {
		t.RetInt(0)
		return
	}
	dbfArea, _ := area.(*dbf.DBFArea)
	if dbfArea == nil {
		// Non-DBF area falls back to the generic Area interface — use
		// the interface path; still a win over PRG boundary crossings.
		t.RetInt(sqlBulkUpdateGeneric(t, area, whereFn, fieldPos, valuePcodes))
		return
	}

	// Fast field getter — compiled pcode's PcOpFieldGet hits this
	// closure instead of the generic FieldGet RTL dispatch.
	prevFG := t.FastFieldGetter
	t.FastFieldGetter = func(idx int) hbrt.Value {
		v, _ := dbfArea.GetValue(idx - 1)
		return v
	}
	defer func() { t.FastFieldGetter = prevFG }()

	nAffected := 0
	shared := dbfArea.IsShared()

	dbfArea.GoTop()
	for !dbfArea.EOF() {
		match := true
		if whereFn != nil {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			match = t.GetRetValue().AsBool()
		}
		if match {
			recNo := dbfArea.RecNo()
			locked := true
			if shared {
				lockOk, _ := dbfArea.LockRecord(recNo)
				locked = lockOk
			}
			if locked {
				for i := 0; i < nSets; i++ {
					pc := valuePcodes[i]
					if pc == nil {
						continue
					}
					hbrt.ExecPcodeFast(t, pc, nil)
					dbfArea.PutValue(fieldPos[i], t.GetRetValue())
				}
				if shared {
					dbfArea.UnlockRecord(recNo)
				}
				nAffected++
			}
		}
		dbfArea.Skip(1)
	}
	/* Skip fsync when the WA cache is active — caller batches flush
	 * at SqlWACacheDisable / dbCloseAll. Per-call Flush on macOS APFS
	 * is ~1-2 ms (fsync), dominating the 100-row scan cost. */
	if !waCacheEnabledSafe() {
		dbfArea.Flush()
	}

	// Index maintenance. DBFArea.PutValue patches record bytes but does
	// not delete + re-add index keys, so any index whose expression
	// references one of the updated fields goes stale. We rebuild those
	// indexes on the spot rather than leaving divergent state behind.
	//
	// Triggering condition: an index is open AND at least one updated
	// field name appears in any index's key expression. We over-match
	// by substring (so "ID" matches a compound expression like
	// "DEPT+ID"), which is conservative — spurious rebuilds of indexes
	// that happened to share a substring but don't really reference
	// the field, never the reverse. Tables with no open indexes or
	// with indexes that don't cover the updated columns skip the
	// rebuild entirely, preserving the B13 UPDATE hot-path timing.
	if nAffected > 0 && sqlBulkUpdateNeedsIndexRebuild(dbfArea, fieldPos) {
		_ = dbfArea.OrderListRebuild()
	}

	t.RetInt(int64(nAffected))
}

// sqlBulkUpdateNeedsIndexRebuild reports whether any open index on the
// workarea references any of the just-written columns. Called once at
// the end of SqlBulkUpdate, so the hot path stays per-record-free.
func sqlBulkUpdateNeedsIndexRebuild(a *dbf.DBFArea, fieldPos []int) bool {
	nOrd := a.IndexCount()
	if nOrd == 0 {
		return false
	}
	// Collect upper-cased names of the updated fields.
	fieldNames := make([]string, 0, len(fieldPos))
	for _, idx := range fieldPos {
		if idx < 0 || idx >= a.FieldCount() {
			continue
		}
		name := strings.ToUpper(strings.TrimRight(a.GetFieldInfo(idx).Name, "\x00 "))
		if name != "" {
			fieldNames = append(fieldNames, name)
		}
	}
	if len(fieldNames) == 0 {
		return false
	}
	for i := 1; i <= nOrd; i++ {
		expr := strings.ToUpper(a.OrderKeyExpr(i))
		if expr == "" {
			// Index opened without a KeyExpr (legacy OrderListAdd path
			// prior to the NTX header read). Conservatively rebuild —
			// we can't prove the index doesn't cover these fields.
			return true
		}
		for _, name := range fieldNames {
			if strings.Contains(expr, name) {
				return true
			}
		}
	}
	return false
}

// waCacheEnabledSafe reads the cache flag under its lock — fast enough
// to call on every Bulk path, avoids the PRG→Go round-trip.
func waCacheEnabledSafe() bool {
	waCacheMu.Lock()
	on := waCacheEnabled
	waCacheMu.Unlock()
	return on
}

// sqlBulkUpdateGeneric handles non-DBF workareas via the Area interface.
func sqlBulkUpdateGeneric(t *hbrt.Thread, area hbrdd.Area, whereFn *hbrt.PcodeFunc, fieldPos []int, valuePcodes []*hbrt.PcodeFunc) int64 {
	prevFG := t.FastFieldGetter
	t.FastFieldGetter = func(idx int) hbrt.Value {
		v, _ := area.GetValue(idx - 1)
		return v
	}
	defer func() { t.FastFieldGetter = prevFG }()

	nAffected := int64(0)
	area.GoTop()
	for !area.EOF() {
		match := true
		if whereFn != nil {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			match = t.GetRetValue().AsBool()
		}
		if match {
			for i := 0; i < len(fieldPos); i++ {
				pc := valuePcodes[i]
				if pc == nil {
					continue
				}
				hbrt.ExecPcodeFast(t, pc, nil)
				area.PutValue(fieldPos[i], t.GetRetValue())
			}
			nAffected++
		}
		area.Skip(1)
	}
	return nAffected
}

// SqlBulkDelete(pcWhere) → nAffected
//
// Go-native DELETE scan loop — counterpart to SqlBulkUpdate for pure
// DELETE FROM t WHERE ... statements. Replaces the PRG pattern:
//
//    dbGoTop()
//    WHILE ! Eof()
//       IF xWhere == NIL .OR. SqlIsTrue( ::EvalExpr( xWhere ) )
//          dbRLock( RecNo() )
//          dbDelete()
//          dbRUnlock( RecNo() )
//          nAffected++
//       ENDIF
//       dbSkip()
//    ENDDO
//
// Same caveats as SqlBulkUpdate: caller must guarantee no active
// transaction (LogRecord is omitted) and SET DELETED handling stays
// with the PRG wrapper if it needs it.
//
// NIL whereFn ⇒ delete every row (caller should usually route that
// through TRUNCATE instead, but the behaviour is preserved for
// compat).
func SqlBulkDelete(t *hbrt.Thread) {
	t.Frame(1, 0)
	defer t.EndProc()

	whereVal := t.Local(1)
	var whereFn *hbrt.PcodeFunc
	if !whereVal.IsNil() {
		if p := whereVal.AsPointer(); p != nil {
			whereFn, _ = p.(*hbrt.PcodeFunc)
		}
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok {
		t.RetInt(0)
		return
	}
	area := wam.Current()
	if area == nil {
		t.RetInt(0)
		return
	}
	dbfArea, _ := area.(*dbf.DBFArea)
	if dbfArea == nil {
		t.RetInt(sqlBulkDeleteGeneric(t, area, whereFn))
		return
	}

	prevFG := t.FastFieldGetter
	t.FastFieldGetter = func(idx int) hbrt.Value {
		v, _ := dbfArea.GetValue(idx - 1)
		return v
	}
	defer func() { t.FastFieldGetter = prevFG }()

	nAffected := 0
	shared := dbfArea.IsShared()
	dbfArea.GoTop()
	for !dbfArea.EOF() {
		match := true
		if whereFn != nil {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			match = t.GetRetValue().AsBool()
		}
		if match {
			recNo := dbfArea.RecNo()
			locked := true
			if shared {
				lockOk, _ := dbfArea.LockRecord(recNo)
				locked = lockOk
			}
			if locked {
				dbfArea.Delete()
				if shared {
					dbfArea.UnlockRecord(recNo)
				}
				nAffected++
			}
		}
		dbfArea.Skip(1)
	}
	if !waCacheEnabledSafe() {
		dbfArea.Flush()
	}
	t.RetInt(int64(nAffected))
}

// sqlBulkDeleteGeneric handles non-DBF workareas via the Area interface.
func sqlBulkDeleteGeneric(t *hbrt.Thread, area hbrdd.Area, whereFn *hbrt.PcodeFunc) int64 {
	prevFG := t.FastFieldGetter
	t.FastFieldGetter = func(idx int) hbrt.Value {
		v, _ := area.GetValue(idx - 1)
		return v
	}
	defer func() { t.FastFieldGetter = prevFG }()

	nAffected := int64(0)
	area.GoTop()
	for !area.EOF() {
		match := true
		if whereFn != nil {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			match = t.GetRetValue().AsBool()
		}
		if match {
			area.Delete()
			nAffected++
		}
		area.Skip(1)
	}
	return nAffected
}

// Frame-offset sentinels for SqlWindowSlideAgg. PRG encodes the SQL
// frame bounds "UNBOUNDED PRECEDING / FOLLOWING" into these values;
// any other offset is a relative row count (-N preceding, +N
// following, 0 current row).
const (
	frameUnboundedPreceding = -(1 << 30)
	frameUnboundedFollowing = (1 << 30)
)

// SqlWindowSlideAgg(aRows, aPartIdx, nArgCol, nColIdx, cFunc, leftOff, rightOff) → lHandled
//
// O(N) replacement for the ApplyWindowFunctions general-frame inner
// loop. Two algorithms share one entry point:
//
//   SUM / AVG / COUNT — prefix-sum sweep. O(N) build, O(1) query per
//       row. Two subtractions per frame instead of the O(N·W) inner
//       loop that dominates wide-frame workloads like `ROWS BETWEEN
//       50 PRECEDING AND 50 FOLLOWING`.
//
//   MIN / MAX — monotonic deque. SQL frame bounds are linear in the
//       row index for every standard frame spec (UNBOUNDED PRECEDING,
//       fixed N PRECEDING, CURRENT ROW, fixed N FOLLOWING, UNBOUNDED
//       FOLLOWING), so L and R are both non-decreasing in k and the
//       classic sliding-window deque applies in one sweep. Amortized
//       O(1) per row.
//
// Returns .T. on success, .F. if the aggregate / value types aren't
// supported by the fast path — PRG falls back to the O(N·W) loop.
// Currently the MIN/MAX path only accepts numeric partition values;
// a non-numeric, non-NIL value in the scan column sends the RTL back
// to PRG so string / date comparisons still work correctly via the
// existing SqlCmpLt dispatch.
//
// Semantics match the PRG fallback:
//   - COUNT(*) counts every row in frame (nArgCol == 0, i.e. <=0 here).
//   - COUNT(expr), SUM, AVG, MIN, MAX skip NIL values.
//   - SUM / AVG / MIN / MAX with an empty or all-NIL frame return NIL.
//   - COUNT over empty frame returns 0.
//   - Frame clamped to [1..partLen] just like SqlFrameOffset did.
func SqlWindowSlideAgg(t *hbrt.Thread) {
	t.Frame(7, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	partVal := t.Local(2)
	nArgCol := int(t.Local(3).AsNumInt()) - 1 // 0-based; -1 = COUNT(*)
	nColIdx := int(t.Local(4).AsNumInt()) - 1
	cFunc := strings.ToUpper(t.Local(5).AsString())
	leftOff := int(t.Local(6).AsNumInt())
	rightOff := int(t.Local(7).AsNumInt())

	if !rowsVal.IsArray() || !partVal.IsArray() {
		t.RetBool(false)
		return
	}
	rowsArr := rowsVal.AsArray().Items
	partArr := partVal.AsArray().Items
	N := len(partArr)
	if N == 0 {
		t.RetBool(true)
		return
	}

	// Snapshot partition indices as 0-based int once.
	part := make([]int, N)
	for i, v := range partArr {
		part[i] = int(v.AsNumInt()) - 1
	}

	switch cFunc {
	case "SUM", "AVG", "COUNT":
		sqlWindowPrefixAgg(rowsArr, part, nArgCol, nColIdx, cFunc, leftOff, rightOff)
		t.RetBool(true)
	case "MIN", "MAX":
		if nArgCol < 0 {
			// MIN/MAX(*) has no meaning — matches PRG which treats it
			// as "always NIL" via the no-argcol branch.
			t.RetBool(false)
			return
		}
		ok := sqlWindowMonotonicMinMax(rowsArr, part, nArgCol, nColIdx, cFunc, leftOff, rightOff)
		t.RetBool(ok)
	default:
		t.RetBool(false)
	}
}

// sqlWindowPrefixAgg runs the O(N) prefix-sum sweep for SUM / AVG /
// COUNT. Extracted from the SqlWindowSlideAgg body so the MIN/MAX
// path can share the setup without duplicating it.
func sqlWindowPrefixAgg(
	rowsArr []hbrt.Value, part []int, nArgCol, nColIdx int,
	cFunc string, leftOff, rightOff int,
) {
	N := len(part)
	// Build prefix arrays: prefSum[i] = sum of values[0..i-1],
	// prefCnt[i] = count of non-NIL values[0..i-1].
	prefSum := make([]float64, N+1)
	prefCnt := make([]int, N+1)
	for i := 0; i < N; i++ {
		prefSum[i+1] = prefSum[i]
		prefCnt[i+1] = prefCnt[i]
		if nArgCol >= 0 {
			rowIdx := part[i]
			if rowIdx >= 0 && rowIdx < len(rowsArr) {
				rowArr := rowsArr[rowIdx].AsArray()
				if rowArr != nil && nArgCol < len(rowArr.Items) {
					v := rowArr.Items[nArgCol]
					if !v.IsNil() && v.IsNumeric() {
						prefSum[i+1] += v.AsNumDouble()
						prefCnt[i+1]++
					}
				}
			}
		}
	}
	for k := 0; k < N; k++ {
		L, R := resolveFrameBounds(k, N, leftOff, rightOff)
		rowIdx := part[k]
		if rowIdx < 0 || rowIdx >= len(rowsArr) {
			continue
		}
		rowArr := rowsArr[rowIdx].AsArray()
		if rowArr == nil || nColIdx < 0 || nColIdx >= len(rowArr.Items) {
			continue
		}
		var result hbrt.Value
		if L > R {
			switch cFunc {
			case "COUNT":
				result = hbrt.MakeInt(0)
			default:
				result = hbrt.MakeNil()
			}
		} else if cFunc == "COUNT" && nArgCol < 0 {
			result = hbrt.MakeInt(R - L + 1)
		} else {
			winSum := prefSum[R+1] - prefSum[L]
			winCnt := prefCnt[R+1] - prefCnt[L]
			switch cFunc {
			case "SUM":
				if winCnt == 0 {
					result = hbrt.MakeNil()
				} else {
					result = hbrt.MakeDouble(winSum, 0, 0)
				}
			case "AVG":
				if winCnt == 0 {
					result = hbrt.MakeNil()
				} else {
					result = hbrt.MakeDouble(winSum/float64(winCnt), 0, 0)
				}
			case "COUNT":
				result = hbrt.MakeInt(winCnt)
			default:
				result = hbrt.MakeNil()
			}
		}
		rowArr.Items[nColIdx] = result
	}
}

// sqlWindowMonotonicMinMax answers each row's MIN / MAX over its
// window frame in amortized O(1) using a monotonic deque of partition
// indices. Returns false (and writes nothing) if a non-numeric,
// non-NIL value is encountered — the PRG loop handles string / date
// comparisons via SqlCmpLt.
//
// The deque holds indices `i` into part[]; values stored at those
// indices form a monotonically non-increasing sequence (for MIN) or
// non-decreasing (for MAX), so the front is always the extremum of
// the currently valid window.
func sqlWindowMonotonicMinMax(
	rowsArr []hbrt.Value, part []int, nArgCol, nColIdx int,
	cFunc string, leftOff, rightOff int,
) bool {
	N := len(part)
	// Extract numeric values + NIL flags up front. If any non-NIL,
	// non-numeric value appears, bail so the PRG loop can handle it.
	vals := make([]float64, N)
	hasVal := make([]bool, N)
	origVal := make([]hbrt.Value, N) // preserve original Value for result
	for i := 0; i < N; i++ {
		rowIdx := part[i]
		if rowIdx < 0 || rowIdx >= len(rowsArr) {
			continue
		}
		rowArr := rowsArr[rowIdx].AsArray()
		if rowArr == nil || nArgCol >= len(rowArr.Items) {
			continue
		}
		v := rowArr.Items[nArgCol]
		if v.IsNil() {
			continue
		}
		if !v.IsNumeric() {
			return false
		}
		vals[i] = v.AsNumDouble()
		hasVal[i] = true
		origVal[i] = v
	}

	isMin := cFunc == "MIN"
	// Ring-buffer deque keyed by partition index. The index is also
	// its position in the monotonic sequence; values at those indices
	// are the comparison key. Capacity N is an upper bound.
	deque := make([]int, 0, N)
	nextToPush := 0

	for k := 0; k < N; k++ {
		L, R := resolveFrameBounds(k, N, leftOff, rightOff)

		// Ingest all partition indices up to R that haven't been
		// pushed yet. NIL values never enter the deque, matching
		// PRG's MIN/MAX which skip NILs.
		for nextToPush <= R && nextToPush < N {
			if hasVal[nextToPush] {
				x := vals[nextToPush]
				for len(deque) > 0 {
					back := deque[len(deque)-1]
					if (isMin && vals[back] >= x) || (!isMin && vals[back] <= x) {
						deque = deque[:len(deque)-1]
						continue
					}
					break
				}
				deque = append(deque, nextToPush)
			}
			nextToPush++
		}
		// Retire deque entries that fell outside the window's left edge.
		for len(deque) > 0 && deque[0] < L {
			deque = deque[1:]
		}

		rowIdx := part[k]
		if rowIdx < 0 || rowIdx >= len(rowsArr) {
			continue
		}
		rowArr := rowsArr[rowIdx].AsArray()
		if rowArr == nil || nColIdx < 0 || nColIdx >= len(rowArr.Items) {
			continue
		}

		if L > R || len(deque) == 0 {
			rowArr.Items[nColIdx] = hbrt.MakeNil()
		} else {
			rowArr.Items[nColIdx] = origVal[deque[0]]
		}
	}
	return true
}

// resolveFrameBounds turns the encoded relative offsets into 0-based
// inclusive [L, R] bounds clamped to the partition. The sentinel
// values map to absolute boundaries; everything else is k + offset.
func resolveFrameBounds(k, N, leftOff, rightOff int) (int, int) {
	var L, R int
	switch leftOff {
	case frameUnboundedPreceding:
		L = 0
	case frameUnboundedFollowing:
		L = N
	default:
		L = k + leftOff
	}
	switch rightOff {
	case frameUnboundedPreceding:
		R = -1
	case frameUnboundedFollowing:
		R = N - 1
	default:
		R = k + rightOff
	}
	if L < 0 {
		L = 0
	}
	if R >= N {
		R = N - 1
	}
	return L, R
}

// SqlBulkInsert(aRows) → nInserted
//
// Go-native bulk INSERT into the current workarea. Replaces the
// PRG pattern used by CTE materialization, CREATE TABLE AS SELECT,
// and subquery-driven temp tables:
//
//    FOR j := 1 TO Len( aRows )
//       dbAppend()
//       FOR k := 1 TO Min( Len(aStruct), Len(aRows[j]) )
//          IF aRows[j][k] != NIL
//             FieldPut( k, aRows[j][k] )
//          ENDIF
//       NEXT
//    NEXT
//    dbCommit()
//
// Collapses ~N·M Go RTL boundary crossings to a single call plus
// native Append/PutValue/Flush on *DBFArea. Semantics preserved:
//   - NIL element → field left at its default value
//   - Row length > field count → extra columns ignored
//   - Row length < field count → trailing fields left at default
//   - Flushes once at end (matches PRG dbCommit() after the loop)
//
// Returns the number of rows appended (excluding rows where aRows[i]
// is not an array — those are skipped silently, matching the PRG
// loop which would panic on non-array access).
func SqlBulkInsert(t *hbrt.Thread) {
	t.Frame(1, 0)
	defer t.EndProc()

	rowsVal := t.Local(1)
	if !rowsVal.IsArray() {
		t.RetInt(0)
		return
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok || wam == nil {
		t.RetInt(0)
		return
	}
	area := wam.Current()
	if area == nil {
		t.RetInt(0)
		return
	}

	nFields := area.FieldCount()
	rows := rowsVal.AsArray().Items
	inserted := 0

	// Type-assert the concrete DBF type once so the inner loop avoids
	// interface-dispatch per call. Non-DBF backends (MEMRDD) take the
	// generic hbrdd.Area path.
	// NIL values must still be routed through PutValue so the DBF
	// driver sets the _NullFlags bit for nullable columns. Skipping
	// the call leaves the raw bytes at their dbAppend() defaults
	// (spaces / zeros), which reads back as empty string / 0 rather
	// than SQL NULL. Pre-nullable code skipped NIL purely as an
	// optimization (no-op write); with the nullable bitmap that
	// "optimization" silently discards NULL markers on multi-row
	// INSERT VALUES (...), (...), ...
	if dbfArea, isDbf := area.(*dbf.DBFArea); isDbf {
		for _, rowVal := range rows {
			ra := rowVal.AsArray()
			if ra == nil {
				continue
			}
			if err := dbfArea.Append(); err != nil {
				break
			}
			limit := len(ra.Items)
			if limit > nFields {
				limit = nFields
			}
			for k := 0; k < limit; k++ {
				dbfArea.PutValue(k, ra.Items[k])
			}
			inserted++
		}
		dbfArea.Flush()
	} else {
		for _, rowVal := range rows {
			ra := rowVal.AsArray()
			if ra == nil {
				continue
			}
			if err := area.Append(); err != nil {
				break
			}
			limit := len(ra.Items)
			if limit > nFields {
				limit = nFields
			}
			for k := 0; k < limit; k++ {
				area.PutValue(k, ra.Items[k])
			}
			inserted++
		}
		if flusher, ok := area.(interface{ Flush() error }); ok {
			flusher.Flush()
		}
	}

	t.RetInt(int64(inserted))
}

// SqlEach(aFieldPositions, pcWhere, bBlock) → NIL
//
// Streaming variant of SqlScan — instead of materializing all matching
// rows into a result array (which costs N HbArray allocations plus a
// second pass when the PRG caller iterates it), we invoke a user-provided
// code block once per matching row, passing the selected field values as
// block parameters.
//
// This is the Harbour block-iteration idiom (`AEval`, `AScan`) applied
// to SQL. Total heap traffic collapses to ~0 — no result rows, no slab,
// no flat value buffer. Per-row overhead becomes just (field reads +
// WHERE eval + block invoke).
//
// Expected to hit raw-RDD parity on end-to-end "SQL → user code" timing.
//
// Arguments:
//   aFieldPositions: 1-based field positions to pass as block params
//   pcWhere:         compiled WHERE predicate, or NIL
//   bBlock:          code block receiving nFields positional params
func SqlEach(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	fieldsVal := t.Local(1)
	if !fieldsVal.IsArray() {
		t.RetNil()
		return
	}
	fieldsArr := fieldsVal.AsArray().Items
	nFields := len(fieldsArr)

	whereVal := t.Local(2)
	var whereFn *hbrt.PcodeFunc
	if !whereVal.IsNil() {
		if p := whereVal.AsPointer(); p != nil {
			whereFn, _ = p.(*hbrt.PcodeFunc)
		}
	}

	blockVal := t.Local(3)
	if !blockVal.IsBlock() {
		t.RetNil()
		return
	}
	blk := blockVal.AsBlock()

	fieldPos := make([]int, nFields)
	for i := 0; i < nFields; i++ {
		fieldPos[i] = int(fieldsArr[i].AsNumInt())
		if fieldPos[i] < 1 {
			fieldPos[i] = 1
		}
	}

	wam, ok := t.WA.(*hbrdd.WorkAreaManager)
	if !ok {
		t.RetNil()
		return
	}
	area := wam.Current()
	if area == nil {
		t.RetNil()
		return
	}
	dbfArea, _ := area.(*dbf.DBFArea)

	// Install FastFieldGetter for the WHERE predicate's PcOpFieldGet ops
	prevFG := t.FastFieldGetter
	if dbfArea != nil {
		t.FastFieldGetter = func(idx int) hbrt.Value {
			v, _ := dbfArea.GetValue(idx - 1)
			return v
		}
	} else {
		t.FastFieldGetter = func(idx int) hbrt.Value {
			v, _ := area.GetValue(idx - 1)
			return v
		}
	}
	defer func() { t.FastFieldGetter = prevFG }()

	// Block eval protocol: push N args on the stack, set pendingParams,
	// call blk.Fn(t). Matches what EvalBlock does inline, skipping the
	// per-call `make([]Value, nArgs)` temp slice.
	//
	// Four specialized loops on {DBF, generic}×{WHERE, none}, same
	// reasoning as SqlScan's loop split.
	switch {
	case dbfArea != nil && whereFn != nil:
		dbfArea.GoTop()
		for !dbfArea.EOF() {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			if t.GetRetValue().AsBool() {
				for i := 0; i < nFields; i++ {
					v, _ := dbfArea.GetValue(fieldPos[i] - 1)
					t.PushValue(v)
				}
				t.PendingParams2(nFields)
				blk.Fn(t)
			}
			dbfArea.Skip(1)
		}
	case dbfArea != nil:
		dbfArea.GoTop()
		for !dbfArea.EOF() {
			for i := 0; i < nFields; i++ {
				v, _ := dbfArea.GetValue(fieldPos[i] - 1)
				t.PushValue(v)
			}
			t.PendingParams2(nFields)
			blk.Fn(t)
			dbfArea.Skip(1)
		}
	case whereFn != nil:
		area.GoTop()
		for !area.EOF() {
			hbrt.ExecPcodeFast(t, whereFn, nil)
			if t.GetRetValue().AsBool() {
				for i := 0; i < nFields; i++ {
					v, _ := area.GetValue(fieldPos[i] - 1)
					t.PushValue(v)
				}
				t.PendingParams2(nFields)
				blk.Fn(t)
			}
			area.Skip(1)
		}
	default:
		area.GoTop()
		for !area.EOF() {
			for i := 0; i < nFields; i++ {
				v, _ := area.GetValue(fieldPos[i] - 1)
				t.PushValue(v)
			}
			t.PendingParams2(nFields)
			blk.Fn(t)
			area.Skip(1)
		}
	}

	t.RetNil()
}

// SqlFetchRowFast(oSelf, aExprs, aFetchCache) → aRow
//
// Go-native replacement for TSqlExecutor:FetchRow. Profile showed
// FetchRow at ~30% of B4 GROUP+HAVING CPU — 100 rows × 1000 iters of
// PRG method dispatch per column per row, even with the aFetchCache
// fast path. This collapses the per-row loop into one Go call: bound
// cache entries (`{nWA, nFPos}`) do a direct SelectByNum+GetValue;
// unbound entries fall back to `self:EvalExpr(exprs[i][1])` via Send.
// Character values get trimmed inline (mirrors PRG AllTrim, which is
// really TrimSpace in practice since DBF pads with ASCII space).
func SqlFetchRowFast(t *hbrt.Thread) {
	t.Frame(3, 0)
	defer t.EndProc()

	self := t.Local(1)
	exprsVal := t.Local(2)
	cacheVal := t.Local(3)

	if !exprsVal.IsArray() {
		t.PushValue(hbrt.MakeArrayFrom(nil))
		t.RetValue()
		return
	}
	exprs := exprsVal.AsArray().Items
	n := len(exprs)

	var cache []hbrt.Value
	useCache := false
	if cacheVal.IsArray() {
		cache = cacheVal.AsArray().Items
		useCache = len(cache) == n
	}

	wa := getWA(t)
	out := make([]hbrt.Value, 0, n)

	for i := 0; i < n; i++ {
		var val hbrt.Value
		hit := false

		if useCache {
			entry := cache[i]
			if !entry.IsNil() && entry.IsArray() {
				items := entry.AsArray().Items
				if len(items) >= 2 && wa != nil {
					nWA := uint16(items[0].AsNumInt())
					nFPos := int(items[1].AsNumInt())
					wa.SelectByNum(nWA)
					if area := wa.Current(); area != nil {
						if v, err := area.GetValue(nFPos - 1); err == nil {
							val = v
							hit = true
						}
					}
				}
			}
		}

		if !hit {
			// Fallback: self:EvalExpr(exprs[i][1])
			var exprNode hbrt.Value
			if exprs[i].IsArray() {
				items := exprs[i].AsArray().Items
				if len(items) > 0 {
					exprNode = items[0]
				}
			}
			t.PushValue(self)
			t.PushValue(exprNode)
			t.Send("EVALEXPR", 1)
			val = t.Pop2()
		}

		if val.IsString() {
			s := val.AsString()
			trimmed := strings.TrimSpace(s)
			if len(trimmed) != len(s) {
				val = hbrt.MakeString(trimmed)
			}
		}
		out = append(out, val)
	}

	t.PushValue(hbrt.MakeArrayFrom(out))
	t.RetValue()
}