TsDB2/DB/driver/exec.go

package vrtxsql

// 语句执行: AST → 引擎查询体系 (core.Query / Condition / Table)。
// 调用约定: 进入本文件任一函数时 shared.mu 已由调用方持有。

import (
	"context"
	"database/sql/driver"
	"fmt"
	"math"
	"reflect"
	"strconv"
	"strings"
	"time"

	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)

// execStatement 执行非查询语句 (INSERT/UPDATE/DELETE/DDL)。
func execStatement(db *core.DB, st *statement, args []driver.NamedValue) (driver.Result, error) {
	switch st.kind {
	case kInsert:
		return execInsert(db, st.ins, args)
	case kUpdate:
		return execUpdate(db, st.upd, args)
	case kDelete:
		return execDelete(db, st.del, args)
	case kCreate:
		return execCreate(db, st.crt)
	case kCreateIndex:
		return execCreateIndex(db, st.cidx)
	case kDrop:
		return execDrop(db, st.drop)
	case kAlter:
		return execAlter(db, st.alt)
	case kTruncate:
		return execTruncate(db, st.trunc)
	default:
		return nil, fmt.Errorf("vrtx: 该语句不能用 Exec 执行")
	}
}

func execInsert(db *core.DB, ins *insertStmt, args []driver.NamedValue) (driver.Result, error) {
	tbl, err := findTable(db, ins.table)
	if err != nil {
		return nil, err
	}
	cols, _, err := resolveInsertColumns(tbl, ins.columns)
	if err != nil {
		return nil, err
	}
	// SQL INSERT 语义: 未提供的列 = NULL (可空列置 NULL 位, 非空列报错)。
	// 引擎原生约定"未提供的列 = 零值"——驱动层把缺失列显式补成 nil,
	// 由 serializeRow 按 FlagNullable 裁决。
	provided := make(map[string]int, len(cols))
	for i, c := range cols {
		provided[strings.ToLower(c)] = i
	}
	allCols := activeColumnNames(tbl)
	// 显式主键: 提供了主键列的整数值就走 InsertRowWithID (SQL 语义: 给定 id 就是行号);
	// 未提供则走引擎自增 (InsertRow 分配的号会回填进主键镜像列)
	pkCol := pkColumnName(tbl)
	var pkFi *types.FieldInfo
	if pkCol != "" {
		pkFi = findField(tbl, pkCol)
	}
	var lastID int64
	for _, row := range ins.rows {
		if len(row) != len(cols) {
			return nil, fmt.Errorf("INSERT 列数 (%d) 与值数 (%d) 不一致", len(cols), len(row))
		}
		fvs := make([]types.FieldValue, 0, len(allCols))
		for _, cn := range allCols {
			fi := findField(tbl, cn)
			if pi, ok := provided[strings.ToLower(cn)]; ok {
				fv, err := buildFieldValue(fi, row[pi], args)
				if err != nil {
					return nil, fmt.Errorf("列 %q: %w", cn, err)
				}
				if fv != nil {
					fvs = append(fvs, *fv)
					continue
				}
			}
			// 未提供 / 显式 NULL: 显式写 nil (引擎置 NULL 位或按非空拒绝)
			fvs = append(fvs, types.FieldValue{FieldName: fi.FieldName, Value: nil})
		}
		// 主键判定
		var explicitPK *uint64
		if pkFi != nil {
			if pi, ok := provided[strings.ToLower(pkCol)]; ok {
				if _, isNull := row[pi].(nullValue); isNull {
					return nil, fmt.Errorf("主键列 %q 不能为 NULL", pkCol)
				}
				v, err := resolveValue(pkFi, row[pi], args)
				if err != nil {
					return nil, fmt.Errorf("主键列 %q: %w", pkCol, err)
				}
				n, ok := v.(int64)
				if !ok {
					return nil, fmt.Errorf("主键列 %q 需要整数值, 实际 %T", pkCol, v)
				}
				if n < 0 {
					return nil, fmt.Errorf("主键列 %q 不能为负: %d", pkCol, n)
				}
				u := uint64(n)
				explicitPK = &u
			}
		}
		var id uint64
		if explicitPK != nil {
			id, err = tbl.InsertRowWithID(*explicitPK, fvs)
		} else {
			id, err = tbl.InsertRow(fvs)
		}
		if err != nil {
			return nil, err
		}
		if id > math.MaxInt64 {
			return nil, fmt.Errorf("pkID %d 超出 int64 表示范围", id)
		}
		lastID = int64(id)
	}
	return &result{lastID: lastID, affected: int64(len(ins.rows))}, nil
}

// pkColumnName 返回行存储索引 (主键/时序) 的第一个分量列名; 无行存储返回 ""。
func pkColumnName(tbl *core.Table) string {
	meta, ok := tbl.IndexMeta().GetRowStoreIndex()
	if !ok {
		return ""
	}
	fields := meta.Fields()
	if len(fields) == 0 {
		return ""
	}
	idx := int(fields[0])
	if idx < 0 || idx >= len(tbl.Field) {
		return ""
	}
	return string(tbl.Field[idx].FieldName)
}

func execUpdate(db *core.DB, upd *updateStmt, args []driver.NamedValue) (driver.Result, error) {
	tbl, err := findTable(db, upd.table)
	if err != nil {
		return nil, err
	}
	cond, err := translateWhere(tbl, upd.where, args)
	if err != nil {
		return nil, err
	}
	sets := make([]types.FieldValue, 0, len(upd.sets))
	for _, set := range upd.sets {
		fi := findField(tbl, set.col)
		if fi == nil {
			return nil, fmt.Errorf("%w: %q", core.ErrFieldNotFound, set.col)
		}
		fv, err := buildFieldValue(fi, set.val, args)
		if err != nil {
			return nil, fmt.Errorf("SET 列 %q: %w", set.col, err)
		}
		if fv != nil {
			sets = append(sets, *fv)
		}
	}
	// WHERE → pkID 集合 → 逐行更新 (引擎的 UpdateRowByPK 未提供的列沿用旧值)
	q := tbl.Query()
	if cond != nil {
		q = q.Where(cond)
	}
	var pkIDs []uint64
	err = q.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
		pkIDs = append(pkIDs, pkID)
		return true
	})
	if err != nil {
		return nil, err
	}
	for _, pkID := range pkIDs {
		if err := tbl.UpdateRowByPK(pkID, sets); err != nil {
			return nil, fmt.Errorf("更新 pkID=%d: %w", pkID, err)
		}
	}
	return &result{affected: int64(len(pkIDs))}, nil
}

func execDelete(db *core.DB, del *deleteStmt, args []driver.NamedValue) (driver.Result, error) {
	tbl, err := findTable(db, del.table)
	if err != nil {
		return nil, err
	}
	cond, err := translateWhere(tbl, del.where, args)
	if err != nil {
		return nil, err
	}
	q := tbl.Query()
	if cond != nil {
		q = q.Where(cond)
	}
	var pkIDs []uint64
	err = q.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
		pkIDs = append(pkIDs, pkID)
		return true
	})
	if err != nil {
		return nil, err
	}
	for _, pkID := range pkIDs {
		if err := tbl.DeleteRowByPK(pkID); err != nil {
			return nil, fmt.Errorf("删除 pkID=%d: %w", pkID, err)
		}
	}
	return &result{affected: int64(len(pkIDs))}, nil
}

// execCreate CREATE TABLE → 引擎建表 + 主键/唯一/普通索引。
func execCreate(db *core.DB, crt *createStmt) (driver.Result, error) {
	if _, ok := db.FindTable(crt.table); ok {
		if crt.ifNotExists {
			return &result{}, nil
		}
		return nil, fmt.Errorf("表 %q 已存在", crt.table)
	}
	infos := make([]types.FieldInfo, 0, len(crt.columns))
	for _, c := range crt.columns {
		fi, err := sqlTypeToFieldInfo(c)
		if err != nil {
			return nil, fmt.Errorf("列 %q: %w", c.name, err)
		}
		infos = append(infos, fi)
	}
	tbl, err := db.CreateTable(crt.table, infos)
	if err != nil {
		return nil, err
	}
	// 主键索引 (整型列, 引擎行存储要求)
	if crt.pkCol != "" {
		fi := findField(tbl, crt.pkCol)
		if fi == nil {
			return nil, fmt.Errorf("%w (PRIMARY KEY): %q", core.ErrFieldNotFound, crt.pkCol)
		}
		if !isIntType(fi.FieldType) {
			return nil, fmt.Errorf("PRIMARY KEY 列 %q 必须是整数类型 (引擎行存储定位键)", crt.pkCol)
		}
		if _, err := tbl.CreateIndex("pk_"+crt.table, page.IndexTypePrimary, []byte{byte(fieldIndex(tbl, crt.pkCol))}); err != nil {
			return nil, err
		}
	}
	// 列级 UNIQUE → 唯一二级索引
	for _, c := range crt.columns {
		if !c.unique {
			continue
		}
		idx := inlineIndex{name: "ux_" + crt.table + "_" + c.name, columns: []string{c.name}, unique: true}
		crt.indexes = append(crt.indexes, idx)
	}
	// 内联索引 (KEY/UNIQUE KEY)
	for _, idx := range crt.indexes {
		fi := findField(tbl, idx.columns[0])
		if fi == nil {
			return nil, fmt.Errorf("%w (索引 %q): %q", core.ErrFieldNotFound, idx.name, idx.columns[0])
		}
		if fi.IsDropped() || types.IsDynamic(fi.FieldType) || types.IsBlob(fi.FieldType) {
			return nil, fmt.Errorf("索引 %q 的列 %q 不能建索引 (动态/二进制列)", idx.name, idx.columns[0])
		}
		name := idx.name
		if name == "" {
			name = "idx_" + crt.table + "_" + idx.columns[0]
		}
		if _, err := tbl.CreateIndex(name, page.IndexTypeBTree, []byte{byte(fieldIndex(tbl, idx.columns[0]))}, idx.unique); err != nil {
			return nil, err
		}
	}
	return &result{}, nil
}

// execCreateIndex CREATE [UNIQUE] INDEX name ON t (col)
func execCreateIndex(db *core.DB, ci *createIndexStmt) (driver.Result, error) {
	tbl, err := findTable(db, ci.table)
	if err != nil {
		return nil, err
	}
	fi := findField(tbl, ci.col)
	if fi == nil {
		return nil, fmt.Errorf("%w: %q", core.ErrFieldNotFound, ci.col)
	}
	if _, err := tbl.CreateIndex(ci.name, page.IndexTypeBTree, []byte{byte(fieldIndex(tbl, ci.col))}, ci.unique); err != nil {
		return nil, err
	}
	return &result{}, nil
}

func execDrop(db *core.DB, d *dropStmt) (driver.Result, error) {
	if _, ok := db.FindTable(d.table); !ok {
		if d.ifExists {
			return &result{}, nil
		}
		return nil, fmt.Errorf("表 %q 不存在", d.table)
	}
	if err := db.DropTable(d.table); err != nil {
		return nil, err
	}
	return &result{}, nil
}

// execAlter ALTER TABLE t ADD/DROP COLUMN ... —— 引擎 AddField/DropField 已支持 schema 演进。
func execAlter(db *core.DB, alt *alterStmt) (driver.Result, error) {
	tbl, err := findTable(db, alt.table)
	if err != nil {
		return nil, err
	}
	for _, op := range alt.ops {
		if op.drop {
			if err := tbl.DropField(op.col); err != nil {
				return nil, err
			}
			continue
		}
		fi, err := sqlTypeToFieldInfo(op.def)
		if err != nil {
			return nil, fmt.Errorf("列 %q: %w", op.def.name, err)
		}
		if err := tbl.AddField(fi); err != nil {
			return nil, err
		}
		// ADD 的 UNIQUE 列: AddField 只加列与 schema, 这里补建唯一索引。
		if op.def.unique {
			idx := fieldIndex(tbl, op.def.name)
			if idx < 0 {
				return nil, fmt.Errorf("ADD COLUMN 后找不到新列 %q", op.def.name)
			}
			if _, err := tbl.CreateIndex("ux_"+alt.table+"_"+op.def.name,
				page.IndexTypeBTree, []byte{byte(idx)}, true); err != nil {
				return nil, err
			}
		}
	}
	return &result{}, nil
}

// execTruncate TRUNCATE [TABLE] t —— 清空数据、保留表结构与索引定义。
func execTruncate(db *core.DB, tr *truncateStmt) (driver.Result, error) {
	if _, err := findTable(db, tr.table); err != nil {
		return nil, err
	}
	if err := db.TruncateTable(tr.table); err != nil {
		return nil, err
	}
	return &result{}, nil
}

// execShow 以查询方式执行 SHOW / DESCRIBE, 返回物化结果集。算完即 release 锁。
func execShow(db *core.DB, sh *showStmt, release func()) (driver.Rows, error) {
	if release != nil {
		defer release()
	}
	switch sh.what {
	case "TABLES":
		list := db.GetAllTables()
		data := make([][]driver.Value, 0, len(list))
		for _, t := range list {
			data = append(data, []driver.Value{[]byte(t.Name)})
		}
		cols := []string{"Tables_in_vrtx"}
		metas := []colMeta{{dbName: "VARCHAR", nullable: false, scanType: reflect.TypeOf("")}}
		return materialRows(cols, metas, data), nil
	case "COLUMNS", "DESCRIBE":
		tbl, err := findTable(db, sh.table)
		if err != nil {
			return nil, err
		}
		// MySQL 六列: Field / Type / Null / Key / Default / Extra
		cols := []string{"Field", "Type", "Null", "Key", "Default", "Extra"}
		vmeta := colMeta{dbName: "VARCHAR", nullable: false, scanType: reflect.TypeOf("")}
		metas := []colMeta{vmeta, vmeta, vmeta, vmeta, vmeta, vmeta}
		metas[4].nullable = true // Default 列可空
		var data [][]driver.Value
		for i := range tbl.Field {
			f := &tbl.Field[i]
			if f.IsDropped() {
				continue
			}
			null := "NO"
			if f.IsNullable() {
				null = "YES"
			}
			data = append(data, []driver.Value{
				[]byte(string(f.FieldName)),
				[]byte(columnTypeDisplay(f)),
				[]byte(null),
				[]byte(columnKeyIndicator(tbl, i)),
				nil, // Default: 引擎不持久化列默认值
				[]byte(""),
			})
		}
		return materialRows(cols, metas, data), nil
	}
	return nil, fmt.Errorf("不支持的 SHOW 类型 %q", sh.what)
}

// columnTypeDisplay 生成 SHOW COLUMNS 的 Type 列文本, 如 varchar(255) / int / datetime。
func columnTypeDisplay(f *types.FieldInfo) string {
	base := strings.ToLower(fieldTypeToSQLName(f.FieldType))
	switch f.FieldType {
	case types.FieldType_String:
		return fmt.Sprintf("%s(%d)", base, f.Width())
	case types.FieldType_Decimal:
		if f.Scale() > 0 { // FieldInfo 只持久化 scale, 无精度
			return fmt.Sprintf("%s(%d)", base, f.Scale())
		}
	}
	return base
}

// columnKeyIndicator 生成 SHOW COLUMNS 的 Key 列: PRI(主键/时序) / UNI(唯一) / MUL(普通) / ""。
func columnKeyIndicator(tbl *core.Table, target int) string {
	pri, uni, mul := "", "", ""
	for _, m := range tbl.IndexMeta().GetIndexes() {
		hit := false
		for _, fid := range m.Fields() {
			if int(fid) == target {
				hit = true
			}
		}
		if !hit {
			continue
		}
		switch {
		case m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime:
			pri = "PRI"
		case m.Unique():
			uni = "UNI"
		default:
			mul = "MUL"
		}
	}
	switch {
	case pri != "":
		return pri
	case uni != "":
		return uni
	default:
		return mul
	}
}

// queryStatement 执行 SELECT。普通单表 SELECT 流式返回 (持锁至 rows.Close, 首行
// 不必等全扫完, 内存 O(缓冲)); 聚合一次性跑完、立即放锁并返回物化单行;
// JOIN 查询 (join.go) 物化嵌套循环执行, 算完即放锁。
// release 非 nil 表示锁由本次查询 (非事务) 获取, 任何提前返回错误都必须 release。
func queryStatement(ctx context.Context, db *core.DB, sel *selectStmt, args []driver.NamedValue, release func()) (driver.Rows, error) {
	// SELECT 值清单 (无 FROM, 如探活语句 SELECT 1): 不触引擎, 物化单行立即返回。
	if sel.noFrom {
		return queryLiteralRow(sel, args, release)
	}
	if len(sel.joins) > 0 {
		return queryJoin(db, sel, args, release)
	}
	fail := func(err error) (driver.Rows, error) {
		if release != nil {
			release()
		}
		return nil, err
	}
	tbl, err := findTable(db, sel.from.name)
	if err != nil {
		return fail(err)
	}
	// 列引用统一允许限定名 (别名.列 / 表名.列): 单表路径校验限定词是本表后剥掉,
	// 剥好的 where/orderBy/agg.col 放语句副本, 不动 AST (Prepare 的语句要复用)。
	quals := []string{sel.from.alias, sel.from.name}
	sel2 := *sel
	if sel2.where, err = dequalifyExpr(sel.where, quals...); err != nil {
		return fail(err)
	}
	sel2.orderBy = make([]orderSpec, len(sel.orderBy))
	for i, o := range sel.orderBy {
		if sel2.orderBy[i].col, err = dequalifyColRef(o.col, quals...); err != nil {
			return fail(err)
		}
		sel2.orderBy[i].desc = o.desc
	}
	// 聚合: SELECT COUNT(*)/SUM/... FROM t [WHERE ...] —— 一次性算完, 立即放锁
	if sel.agg != nil {
		agg := *sel.agg
		if agg.col, err = dequalifyColRef(agg.col, quals...); err != nil {
			return fail(err)
		}
		sel2.agg = &agg
		r, err := queryAggregate(tbl, &sel2, args)
		if release != nil {
			release()
		}
		return r, err
	}
	// 列清单: * / t.* = 全部活跃列; 否则按请求列
	colNames, err := selectColsSingle(tbl, &sel2, quals...)
	if err != nil {
		return fail(err)
	}
	metas := make([]colMeta, len(colNames))
	for i, cn := range colNames {
		metas[i] = colMetaFor(findField(tbl, cn))
	}

	// 不在此 Select(): q.Scan 回调给的是未投影完整行, worker 按 colNames 自投影。
	q := tbl.Query()
	cond, err := translateWhere(tbl, sel2.where, args)
	if err != nil {
		return fail(err)
	}
	if cond != nil {
		q = q.Where(cond)
	}
	for _, o := range sel2.orderBy {
		// 引擎排序键按精确字节匹配列名: findField 校验存在后改用引擎规范名
		// (用户写 "ORDER BY Name" 也要落到 "name" 列)。
		fi := findField(tbl, o.col)
		if fi == nil {
			return fail(fmt.Errorf("%w: %q", core.ErrFieldNotFound, o.col))
		}
		q = q.OrderBy(string(fi.FieldName), o.desc)
	}
	if sel.limit != nil {
		lim, err := resolveLimitValue("LIMIT", sel.limit, args)
		if err != nil {
			return fail(err)
		}
		q = q.Limit(int(lim))
	}
	if sel.offset != nil {
		off, err := resolveLimitValue("OFFSET", sel.offset, args)
		if err != nil {
			return fail(err)
		}
		q = q.Offset(int(off))
	}

	r := newStreamRows(colNames, metas, blobReaderOf(tbl), release)
	r.start(ctx, colNames, sel.distinct, q)
	return r, nil
}

// queryLiteralRow 执行无 FROM 的 SELECT 值清单 ("SELECT 1" / "SELECT 'a', NULL"):
// 每个值物化成 driver.Value, 拼成单行结果。不触引擎, 立即放锁。
func queryLiteralRow(sel *selectStmt, args []driver.NamedValue, release func()) (driver.Rows, error) {
	if release != nil {
		defer release()
	}
	cols := make([]string, len(sel.literalVals))
	metas := make([]colMeta, len(sel.literalVals))
	row := make([]driver.Value, len(sel.literalVals))
	for i, v := range sel.literalVals {
		cols[i] = "?column?"
		dv, m, err := resolveLiteralValue(v, args)
		if err != nil {
			return nil, fmt.Errorf("SELECT 值清单第 %d 项: %w", i+1, err)
		}
		metas[i] = m
		row[i] = dv
	}
	return materialRows(cols, metas, [][]driver.Value{row}), nil
}

// resolveLiteralValue 把无 FROM 值清单里的字面量/占位符物化成 driver.Value,
// 并按值的种类给出列元数据 (供 ColumnType* 反射)。
func resolveLiteralValue(v value, args []driver.NamedValue) (driver.Value, colMeta, error) {
	intMeta := colMeta{dbName: "BIGINT", nullable: false, scanType: reflect.TypeOf(int64(0))}
	floatMeta := colMeta{dbName: "DOUBLE", nullable: false, scanType: reflect.TypeOf(float64(0))}
	strMeta := colMeta{dbName: "VARCHAR", nullable: false, scanType: reflect.TypeOf("")}
	nullMeta := colMeta{dbName: "NULL", nullable: true, scanType: reflect.TypeOf("")}
	switch t := v.(type) {
	case nullValue:
		return nil, nullMeta, nil
	case strValue:
		return []byte(t.s), strMeta, nil
	case litValue:
		raw := strings.TrimSpace(t.raw)
		if strings.Contains(raw, ".") {
			f, err := strconv.ParseFloat(raw, 64)
			if err != nil {
				return nil, colMeta{}, fmt.Errorf("浮点字面量 %q: %w", t.raw, err)
			}
			return f, floatMeta, nil
		}
		n, err := strconv.ParseInt(raw, 10, 64)
		if err != nil {
			return nil, colMeta{}, fmt.Errorf("整数字面量 %q: %w", t.raw, err)
		}
		return n, intMeta, nil
	case argValue:
		if t.idx >= len(args) {
			return nil, colMeta{}, fmt.Errorf("占位符 ?%d 没有对应参数 (共 %d 个)", t.idx+1, len(args))
		}
		switch x := args[t.idx].Value.(type) {
		case nil:
			return nil, nullMeta, nil
		case int64:
			return x, intMeta, nil
		case float64:
			return x, floatMeta, nil
		case bool:
			return x, colMeta{dbName: "BOOLEAN", nullable: false, scanType: reflect.TypeOf(false)}, nil
		case string:
			return []byte(x), strMeta, nil
		case []byte:
			return x, strMeta, nil
		case time.Time:
			return x, colMeta{dbName: "DATETIME", nullable: false, scanType: reflect.TypeOf(time.Time{})}, nil
		}
		return nil, colMeta{}, fmt.Errorf("参数类型 %T 不能出现在 SELECT 值清单", args[t.idx].Value)
	}
	return nil, colMeta{}, fmt.Errorf("无法解析的值")
}

// resolveLimitValue 解析 LIMIT/OFFSET 的量: 字面量直解整数; 占位符从 args
// 取整数 (int64/uint64/数字字符串/整值浮点均可, 非整数报错)。
func resolveLimitValue(keyword string, v value, args []driver.NamedValue) (int64, error) {
	switch t := v.(type) {
	case litValue:
		n, err := strconv.ParseInt(strings.TrimSpace(t.raw), 10, 64)
		if err != nil {
			return 0, fmt.Errorf("%s 期望整数, 实际是 %q", keyword, t.raw)
		}
		return n, nil
	case argValue:
		if t.idx >= len(args) {
			return 0, fmt.Errorf("占位符 ?%d 没有对应参数 (共 %d 个)", t.idx+1, len(args))
		}
		n, err := limitArgToInt64(args[t.idx].Value)
		if err != nil {
			return 0, fmt.Errorf("%s: %w", keyword, err)
		}
		return n, nil
	}
	return 0, fmt.Errorf("%s 只支持整数或 ? 占位符", keyword)
}

// limitArgToInt64 把 LIMIT/OFFSET 的参数值规整成 int64。
func limitArgToInt64(v driver.Value) (int64, error) {
	switch x := v.(type) {
	case int64:
		return x, nil
	case uint64:
		if x > math.MaxInt64 {
			return 0, fmt.Errorf("超出 int64 范围: %d", x)
		}
		return int64(x), nil
	case float64:
		if x != math.Trunc(x) {
			return 0, fmt.Errorf("参数必须是非负整数, 实际 %v", x)
		}
		return int64(x), nil
	case string:
		n, err := strconv.ParseInt(strings.TrimSpace(x), 10, 64)
		if err != nil {
			return 0, fmt.Errorf("参数 %q 不是整数", x)
		}
		return n, nil
	case []byte:
		return limitArgToInt64(string(x))
	}
	return 0, fmt.Errorf("参数必须是整数, 实际 %T", v)
}

func queryAggregate(tbl *core.Table, sel *selectStmt, args []driver.NamedValue) (*rows, error) {
	cond, err := translateWhere(tbl, sel.where, args)
	if err != nil {
		return nil, err
	}
	q := tbl.Query()
	if cond != nil {
		q = q.Where(cond)
	}
	col := sel.agg.col
	// 聚合目标列在引擎侧同样按精确字节匹配: findField 校验存在后改用引擎
	// 规范名 (COUNT(*) 无目标列, 跳过校验)。
	if col != "" {
		fi := findField(tbl, col)
		if fi == nil {
			return nil, fmt.Errorf("%w: %q", core.ErrFieldNotFound, col)
		}
		col = string(fi.FieldName)
	}
	var dv driver.Value
	var m colMeta
	switch sel.agg.fn {
	case "COUNT":
		n, err := q.Count()
		if err != nil {
			return nil, err
		}
		dv = n
		m = colMeta{dbName: "BIGINT", nullable: false, scanType: reflect.TypeOf(int64(0))}
	case "SUM":
		v, err := q.Sum(col)
		if err != nil {
			return nil, err
		}
		dv = v
		m = colMeta{dbName: "DOUBLE", nullable: true, scanType: reflect.TypeOf(float64(0))}
	case "MIN":
		fi := findField(tbl, col)
		v, err := q.Min(col)
		if err != nil {
			return nil, err
		}
		dv, err = fieldValueToDriverTyped(fi.FieldType, v)
		if err != nil {
			return nil, err
		}
		m = colMetaFor(fi)
		m.nullable = true
	case "MAX":
		fi := findField(tbl, col)
		v, err := q.Max(col)
		if err != nil {
			return nil, err
		}
		dv, err = fieldValueToDriverTyped(fi.FieldType, v)
		if err != nil {
			return nil, err
		}
		m = colMetaFor(fi)
		m.nullable = true
	case "AVG":
		s, err := q.Sum(col)
		if err != nil {
			return nil, err
		}
		// AVG 的分母是"该列非 NULL 的行数", 不是全表行数。
		// 注意引擎 Query.Where 是覆盖语义, 分母查询必须把 WHERE 与
		// IS NOT NULL 合成一个条件, 否则后写覆盖先写、过滤被丢掉。
		var denomCond core.Condition = core.IsNotNull(col)
		if cond != nil {
			denomCond = core.And(cond, denomCond)
		}
		n, err := tbl.Query().Where(denomCond).Count()
		if err != nil {
			return nil, err
		}
		if n == 0 {
			dv = nil // SQL: 无输入行 → NULL
		} else {
			dv = s / float64(n)
		}
		m = colMeta{dbName: "DOUBLE", nullable: true, scanType: reflect.TypeOf(float64(0))}
	default:
		return nil, fmt.Errorf("不支持的聚合函数 %q", sel.agg.fn)
	}
	name := strings.ToLower(sel.agg.fn)
	if col != "" {
		name += "_" + col
	}
	return materialRows([]string{name}, []colMeta{m}, [][]driver.Value{{dv}}), nil
}

// =============================================================================
// 结果集列元数据 (driver.RowsColumnType* 系列)
// =============================================================================

// colMeta 描述一列在结果集中的类型信息。
type colMeta struct {
	ft       types.FieldType // 引擎列类型 (typed 值转换用)
	dbName   string          // 数据库类型名 (BIGINT/VARCHAR/...)
	nullable bool            // 是否可空
	scanType reflect.Type    // Scan 的目标 Go 类型 (sqlx/GORM 反射用)
	length   int64           // 定长字符串宽度
	hasLen   bool
}

func colMetaFor(fi *types.FieldInfo) colMeta {
	m := colMeta{
		ft:       fi.FieldType,
		dbName:   fieldTypeToSQLName(fi.FieldType),
		nullable: fi.Flag&types.FlagNullable != 0,
		scanType: fieldTypeScanType(fi.FieldType),
	}
	if fi.FieldType == types.FieldType_String {
		if w := fi.Width(); w > 0 {
			m.length = int64(w)
			m.hasLen = true
		}
	}
	return m
}

func fieldTypeScanType(ft types.FieldType) reflect.Type {
	switch ft {
	case types.FieldType_DateTime:
		return reflect.TypeOf(time.Time{})
	case types.FieldType_Float64, types.FieldType_Float32:
		return reflect.TypeOf(float64(0))
	case types.FieldType_Bool:
		return reflect.TypeOf(false)
	case types.FieldType_Blob, types.FieldType_Dynamicbyte:
		return reflect.TypeOf([]byte(nil))
	default:
		if isIntType(ft) {
			return reflect.TypeOf(int64(0))
		}
		return reflect.TypeOf("") // String / Decimal / UUID
	}
}

func fieldTypeToSQLName(ft types.FieldType) string {
	switch ft {
	case types.FieldType_DateTime:
		return "DATETIME"
	case types.FieldType_Int64:
		return "BIGINT"
	case types.FieldType_Int32:
		return "INT"
	case types.FieldType_Int16:
		return "SMALLINT"
	case types.FieldType_Int8:
		return "TINYINT"
	case types.FieldType_Uint64:
		return "BIGINT UNSIGNED"
	case types.FieldType_Uint32:
		return "INT UNSIGNED"
	case types.FieldType_Uint16:
		return "SMALLINT UNSIGNED"
	case types.FieldType_Uint8:
		return "TINYINT UNSIGNED"
	case types.FieldType_Float64:
		return "DOUBLE"
	case types.FieldType_Float32:
		return "FLOAT"
	case types.FieldType_Bool:
		return "BOOLEAN"
	case types.FieldType_String:
		return "VARCHAR"
	case types.FieldType_Dynamicbyte:
		return "TEXT"
	case types.FieldType_Blob:
		return "BLOB"
	case types.FieldType_Decimal:
		return "DECIMAL"
	case types.FieldType_UUID:
		return "UUID"
	case types.FieldType_Year:
		return "YEAR"
	}
	return ""
}

// =============================================================================
// WHERE 翻译: SQL 表达式 → core.Condition
// =============================================================================

func translateWhere(tbl *core.Table, e expr, args []driver.NamedValue) (core.Condition, error) {
	if e == nil {
		return nil, nil
	}
	switch t := e.(type) {
	case *andExpr:
		l, err := translateWhere(tbl, t.l, args)
		if err != nil {
			return nil, err
		}
		r, err := translateWhere(tbl, t.r, args)
		if err != nil {
			return nil, err
		}
		return core.And(l, r), nil
	case *orExpr:
		l, err := translateWhere(tbl, t.l, args)
		if err != nil {
			return nil, err
		}
		r, err := translateWhere(tbl, t.r, args)
		if err != nil {
			return nil, err
		}
		return core.Or(l, r), nil
	case *notExpr:
		inner, err := translateWhere(tbl, t.e, args)
		if err != nil {
			return nil, err
		}
		return core.Not(inner), nil
	case *cmpExpr:
		col := resolveColRef(t.col)
		fi := findField(tbl, col)
		if fi == nil {
			return nil, fmt.Errorf("%w (WHERE): %q", core.ErrFieldNotFound, t.col)
		}
		v, err := resolveValue(fi, t.val, args)
		if err != nil {
			return nil, err
		}
		// 引擎条件按精确字节匹配列名, 统一改用引擎规范名 (findField 已做
		// 大小写不敏感的存在性校验, 用户写 "Title" 也能落到 "title" 列)。
		canon := string(fi.FieldName)
		switch t.op {
		case "=":
			return core.Eq(canon, v, fi.FieldType), nil
		case "!=":
			return core.Not(core.Eq(canon, v, fi.FieldType)), nil
		case ">":
			return core.Gt(canon, v, fi.FieldType), nil
		case ">=":
			return core.Ge(canon, v, fi.FieldType), nil
		case "<":
			return core.Lt(canon, v, fi.FieldType), nil
		case "<=":
			return core.Le(canon, v, fi.FieldType), nil
		}
		return nil, fmt.Errorf("不支持的比较符 %q", t.op)
	case *colCmpExpr:
		// 列与列比较: 引擎 Condition 只支持列与值, 仅 JOIN 查询把它当残余谓词在
		// 组合行上求值; 单表/UPDATE/DELETE 走到这里说明用法越界, 明确报错。
		return nil, fmt.Errorf("列与列的比较 (%s %s %s) 仅在多表 JOIN 查询中支持", t.left, t.op, t.right)
	case *inExpr:
		col := resolveColRef(t.col)
		fi := findField(tbl, col)
		if fi == nil {
			return nil, fmt.Errorf("%w (WHERE): %q", core.ErrFieldNotFound, t.col)
		}
		vals := make([]interface{}, 0, len(t.vals))
		for _, tv := range t.vals {
			v, err := resolveValue(fi, tv, args)
			if err != nil {
				return nil, err
			}
			vals = append(vals, v)
		}
		c := core.In(string(fi.FieldName), vals, fi.FieldType)
		if t.neg {
			return core.Not(c), nil
		}
		return c, nil
	case *isNullExpr:
		// IS [NOT] NULL → 引擎的 NullCondition: 直接读行内 NULL 位图
		// (与反序列化/索引提取同一口径; Eq 对 NULL 行一律不匹配, 表达不了 IS NULL)
		// 列名同样要剥限定前缀 — GORM 生成 `t`.`deleted_at` IS NULL,
		// 不剥的话 NullCondition 按全名找不到列、静默恒假 (UPDATE/DELETE 0 行)。
		// NullCondition.Field 也是按精确字节匹配, 必须改用引擎规范名。
		col := resolveColRef(t.col)
		fi := findField(tbl, col)
		if fi == nil {
			return nil, fmt.Errorf("%w (WHERE): %q", core.ErrFieldNotFound, t.col)
		}
		if t.neg {
			return core.IsNotNull(string(fi.FieldName)), nil
		}
		return core.IsNull(string(fi.FieldName)), nil
	case *likeExpr:
		col := resolveColRef(t.col)
		fi := findField(tbl, col)
		if fi == nil {
			return nil, fmt.Errorf("%w (WHERE): %q", core.ErrFieldNotFound, t.col)
		}
		pat, err := resolveLikePattern(t.pattern, args)
		if err != nil {
			return nil, err
		}
		c := core.Like(string(fi.FieldName), pat)
		if t.neg {
			return core.Not(c), nil
		}
		return c, nil
	}
	return nil, fmt.Errorf("无法翻译的 WHERE 表达式")
}

// resolveColRef 列名归一: 剥掉 "表名." / "别名." 限定前缀。GORM 等 ORM 生成的
// SQL 普遍带反引号限定 (`t`.`col`, 词法层拼成 "t.col"), 而引擎条件按裸列名
// 匹配 — 不剥的话: 等值/IN/LIKE 报"列不存在", IS NULL 按全名找不到列、
// Match 静默恒假 (UPDATE/DELETE 作用 0 行, 无任何报错)。
func resolveColRef(col string) string {
	if i := strings.LastIndexByte(col, '.'); i >= 0 {
		return col[i+1:]
	}
	return col
}

// =============================================================================
// 表/列/值解析辅助
// =============================================================================

func findTable(db *core.DB, name string) (*core.Table, error) {
	tbl, ok := db.FindTable(name)
	if !ok {
		return nil, fmt.Errorf("%w: %q", core.ErrTableNotFound, name)
	}
	return tbl, nil
}

// findField 按列名查活跃字段 (大小写不敏感)。
func findField(tbl *core.Table, name string) *types.FieldInfo {
	for i := range tbl.Field {
		f := &tbl.Field[i]
		if f.IsDropped() {
			continue
		}
		if strings.EqualFold(string(f.FieldName), name) {
			return f
		}
	}
	return nil
}

func fieldIndex(tbl *core.Table, name string) int {
	for i := range tbl.Field {
		if !tbl.Field[i].IsDropped() && strings.EqualFold(string(tbl.Field[i].FieldName), name) {
			return i
		}
	}
	return -1
}

func activeColumnNames(tbl *core.Table) []string {
	out := make([]string, 0, len(tbl.Field))
	for i := range tbl.Field {
		if !tbl.Field[i].IsDropped() {
			out = append(out, string(tbl.Field[i].FieldName))
		}
	}
	return out
}

func resolveInsertColumns(tbl *core.Table, columns []string) ([]string, []*types.FieldInfo, error) {
	if len(columns) == 0 {
		names := activeColumnNames(tbl)
		infos := make([]*types.FieldInfo, 0, len(names))
		for _, n := range names {
			infos = append(infos, findField(tbl, n))
		}
		return names, infos, nil
	}
	infos := make([]*types.FieldInfo, 0, len(columns))
	for _, c := range columns {
		fi := findField(tbl, c)
		if fi == nil {
			return nil, nil, fmt.Errorf("%w: %q", core.ErrFieldNotFound, c)
		}
		infos = append(infos, fi)
	}
	return columns, infos, nil
}

// resolveValue 把 AST 值 (占位/字面量/字符串/NULL) 解析成引擎值。
// 字面量按列类型解释 (数字 → int64/float64; 字符串按列类型编码)。
func resolveValue(fi *types.FieldInfo, v value, args []driver.NamedValue) (interface{}, error) {
	switch t := v.(type) {
	case argValue:
		if t.idx >= len(args) {
			return nil, fmt.Errorf("占位符 ?%d 没有对应参数 (共 %d 个)", t.idx+1, len(args))
		}
		return driverToEngineValue(fi, args[t.idx].Value)
	case strValue:
		return parseLiteralAsType(fi, t.s)
	case litValue:
		return parseLiteralAsType(fi, t.raw)
	case nullValue:
		return nil, nil
	}
	return nil, fmt.Errorf("无法解析的值")
}

// buildFieldValue 构造 FieldValue; NULL 返回 nil (调用方跳过, 由引擎按可空/非空裁决)。
func buildFieldValue(fi *types.FieldInfo, v value, args []driver.NamedValue) (*types.FieldValue, error) {
	if _, isNull := v.(nullValue); isNull {
		return nil, nil
	}
	ev, err := resolveValue(fi, v, args)
	if err != nil {
		return nil, err
	}
	return &types.FieldValue{FieldName: fi.FieldName, Value: ev}, nil
}

func resolveLikePattern(v value, args []driver.NamedValue) (string, error) {
	switch t := v.(type) {
	case strValue:
		return t.s, nil
	case argValue:
		if t.idx >= len(args) {
			return "", fmt.Errorf("LIKE 的占位符没有对应参数")
		}
		s, ok := args[t.idx].Value.(string)
		if !ok {
			if b, ok2 := args[t.idx].Value.([]byte); ok2 {
				return string(b), nil
			}
			return "", fmt.Errorf("LIKE 参数必须是字符串")
		}
		return s, nil
	}
	return "", fmt.Errorf("LIKE 模式必须是字符串字面量或占位符")
}

func isIntType(ft types.FieldType) bool {
	switch ft {
	case types.FieldType_Int64, types.FieldType_Uint64,
		types.FieldType_Int32, types.FieldType_Uint32,
		types.FieldType_Int16, types.FieldType_Uint16,
		types.FieldType_Int8, types.FieldType_Uint8:
		return true
	}
	return false
}

// 时间戳约定: driver 的 time.Time 与引擎 uint64 毫秒互转。
var unixEpoch = time.Unix(0, 0)

func timeToMillis(t time.Time) uint64 {
	return uint64(t.Sub(unixEpoch).Milliseconds())
}

func millisToTime(ms uint64) time.Time {
	return unixEpoch.Add(time.Duration(ms) * time.Millisecond)
}