package vrtxsql

// JOIN 查询执行: 多表嵌套循环 + 右表索引点查, 结果物化后立即放锁。
// 仅 SELECT ... FROM t1 JOIN t2 ON ... 走本路径 (单表查询仍走流式路径,
// 见 exec.go queryStatement)。
//
// 执行策略:
//   - 左表全扫描 (本表 WHERE 条件下推), 每行存 别名(小写) → 列(小写) → driver.Value;
//   - 逐 JOIN: 取组合行里 ON 左列的值, 对右表做索引点查
//     (tbl.Query().Where(core.Eq(...)) + 右表本表条件), 命中行做笛卡尔组合;
//     INNER 无匹配丢行, LEFT 无匹配时右表列全 NULL 保留行;
//   - WHERE 按列归属拆分: 叶子条件 (AND 结构) 下推到所属表, 跨表 OR/NOT/列比较
//     作为残余谓词在组合行上用 Go 值比较过滤 (= != < <= > >=);
//   - 投影 (*/t.*/t.col/col 混用, 重名列用 别名_列名 去重) → ORDER BY 内存
//     稳定排序 (NULL 视为最小) → LIMIT/OFFSET, 返回物化 *rows。

import (
	"database/sql/driver"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

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

// =============================================================================
// 限定名列引用的解析辅助 (单表路径解限定 + JOIN 路径归属判定共用)
// =============================================================================

// splitQualCol 把列引用拆成 (限定词, 列名); 无 '.' 时 qual 为空。
func splitQualCol(ref string) (qual, col string) {
	if i := strings.IndexByte(ref, '.'); i >= 0 {
		return ref[:i], ref[i+1:]
	}
	return "", ref
}

// dequalifyColRef 校验并剥掉列引用的限定词 ("t.col" → "col"); 非限定名原样返回。
func dequalifyColRef(ref string, quals ...string) (string, error) {
	qual, col := splitQualCol(ref)
	if qual == "" {
		return ref, nil
	}
	for _, q := range quals {
		if strings.EqualFold(q, qual) {
			return col, nil
		}
	}
	return "", fmt.Errorf("列限定 %q 与 FROM 表 (%s) 不匹配", qual, strings.Join(quals, "/"))
}

// dequalifyExpr 剥掉表达式里所有列引用的限定词 (单表 SELECT 用; 值原样保留)。
func dequalifyExpr(e expr, quals ...string) (expr, error) {
	if e == nil {
		return nil, nil
	}
	switch t := e.(type) {
	case *andExpr:
		l, err := dequalifyExpr(t.l, quals...)
		if err != nil {
			return nil, err
		}
		r, err := dequalifyExpr(t.r, quals...)
		if err != nil {
			return nil, err
		}
		return &andExpr{l: l, r: r}, nil
	case *orExpr:
		l, err := dequalifyExpr(t.l, quals...)
		if err != nil {
			return nil, err
		}
		r, err := dequalifyExpr(t.r, quals...)
		if err != nil {
			return nil, err
		}
		return &orExpr{l: l, r: r}, nil
	case *notExpr:
		inner, err := dequalifyExpr(t.e, quals...)
		if err != nil {
			return nil, err
		}
		return &notExpr{e: inner}, nil
	case *cmpExpr:
		col, err := dequalifyColRef(t.col, quals...)
		if err != nil {
			return nil, err
		}
		return &cmpExpr{col: col, op: t.op, val: t.val}, nil
	case *colCmpExpr:
		l, err := dequalifyColRef(t.left, quals...)
		if err != nil {
			return nil, err
		}
		r, err := dequalifyColRef(t.right, quals...)
		if err != nil {
			return nil, err
		}
		return &colCmpExpr{left: l, op: t.op, right: r}, nil
	case *inExpr:
		col, err := dequalifyColRef(t.col, quals...)
		if err != nil {
			return nil, err
		}
		return &inExpr{col: col, vals: t.vals, neg: t.neg}, nil
	case *isNullExpr:
		col, err := dequalifyColRef(t.col, quals...)
		if err != nil {
			return nil, err
		}
		return &isNullExpr{col: col, neg: t.neg}, nil
	case *likeExpr:
		col, err := dequalifyColRef(t.col, quals...)
		if err != nil {
			return nil, err
		}
		return &likeExpr{col: col, pattern: t.pattern, neg: t.neg}, nil
	}
	return nil, fmt.Errorf("无法处理的 WHERE 表达式")
}

// selectColsSingle 单表 SELECT 的列清单展开: 空 / "*" / "t.*" → 全部活跃列;
// "t.col" → 校验限定词后取列; 列存在性校验与旧行为一致 (ErrFieldNotFound)。
func selectColsSingle(tbl *core.Table, sel *selectStmt, quals ...string) ([]string, error) {
	if len(sel.columns) == 0 {
		return activeColumnNames(tbl), nil
	}
	out := make([]string, 0, len(sel.columns))
	for _, c := range sel.columns {
		if c == "*" {
			out = append(out, activeColumnNames(tbl)...)
			continue
		}
		if strings.HasSuffix(c, ".*") {
			if _, err := dequalifyColRef(c, quals...); err != nil {
				return nil, err
			}
			out = append(out, activeColumnNames(tbl)...)
			continue
		}
		col, err := dequalifyColRef(c, quals...)
		if err != nil {
			return nil, err
		}
		if findField(tbl, col) == nil {
			return nil, fmt.Errorf("%w: %q", core.ErrFieldNotFound, c)
		}
		out = append(out, col)
	}
	return out, nil
}

// =============================================================================
// JOIN 执行
// =============================================================================

// joinTable 参与 JOIN 的一张表: 引擎表 + 归一化别名 + 活跃列。
type joinTable struct {
	idx   int
	name  string
	alias string // 限定名归一化结果 (alias 缺省 = 表名), 比较时大小写不敏感
	tbl   *core.Table
	cols  []string                   // 活跃列名 (原始大小写, 投影展示用)
	ft    map[string]types.FieldType // 小写列名 → 列类型 (typed 值转换用)
}

func buildJoinTable(idx int, ref tableRef, tbl *core.Table) *joinTable {
	alias := ref.alias
	if alias == "" {
		alias = ref.name
	}
	jt := &joinTable{idx: idx, name: ref.name, alias: alias, tbl: tbl, ft: map[string]types.FieldType{}}
	for i := range tbl.Field {
		f := &tbl.Field[i]
		if f.IsDropped() {
			continue
		}
		cn := string(f.FieldName)
		jt.cols = append(jt.cols, cn)
		jt.ft[strings.ToLower(cn)] = f.FieldType
	}
	return jt
}

// joinRow 组合行: 表别名(小写) → 列名(小写) → driver.Value (未匹配侧为 NULL)。
type joinRow map[string]map[string]driver.Value

func (r joinRow) get(jt *joinTable, col string) driver.Value {
	return r[strings.ToLower(jt.alias)][strings.ToLower(col)]
}

func cloneJoinRow(r joinRow) joinRow {
	nr := make(joinRow, len(r)+1)
	for k, v := range r {
		nr[k] = v
	}
	return nr
}

// nullJoinRow 右表无匹配时 (LEFT JOIN) 的全 NULL 行。
func nullJoinRow(jt *joinTable) map[string]driver.Value {
	m := make(map[string]driver.Value, len(jt.cols))
	for _, cn := range jt.cols {
		m[strings.ToLower(cn)] = nil
	}
	return m
}

// findQual 按限定词找表 (别名或表名, 大小写不敏感)。
func findQual(ts []*joinTable, qual string) *joinTable {
	for _, jt := range ts {
		if strings.EqualFold(jt.alias, qual) || strings.EqualFold(jt.name, qual) {
			return jt
		}
	}
	return nil
}

// ownerOfCol 解析列引用归属的表: 限定名直查; 非限定名要求在恰好一张表上存在
// (存在于多张报"列不明确", 没有报列不存在)。返回归属表与剥掉限定后的列名。
func ownerOfCol(ts []*joinTable, ref string) (*joinTable, string, error) {
	qual, col := splitQualCol(ref)
	if qual != "" {
		jt := findQual(ts, qual)
		if jt == nil {
			return nil, "", fmt.Errorf("未知的表限定 %q (列 %q)", qual, ref)
		}
		if findField(jt.tbl, col) == nil {
			return nil, "", fmt.Errorf("%w: %q", core.ErrFieldNotFound, ref)
		}
		return jt, col, nil
	}
	var hit *joinTable
	for _, jt := range ts {
		if findField(jt.tbl, col) != nil {
			if hit != nil {
				return nil, "", fmt.Errorf("列 %q 不明确 (同时存在于 %q 和 %q)", col, hit.alias, jt.alias)
			}
			hit = jt
		}
	}
	if hit == nil {
		return nil, "", fmt.Errorf("%w: %q", core.ErrFieldNotFound, ref)
	}
	return hit, col, nil
}

// scanJoinTable 扫一张表 (应用下推条件), 每行 → 小写列名 → driver.Value。
func scanJoinTable(jt *joinTable, cond core.Condition) ([]map[string]driver.Value, error) {
	q := jt.tbl.Query()
	if cond != nil {
		q = q.Where(cond)
	}
	var out []map[string]driver.Value
	var convErr error
	err := q.Scan(func(full []types.FieldValue) bool {
		m := make(map[string]driver.Value, len(full))
		for _, fv := range full {
			key := strings.ToLower(string(fv.FieldName))
			dv, err := fieldValueToDriverTyped(jt.ft[key], fv.Value)
			if err != nil {
				convErr = err
				return false
			}
			m[key] = joinTypedValue(jt.ft[key], dv)
		}
		out = append(out, m)
		return true
	})
	if convErr != nil {
		return nil, convErr
	}
	if err != nil {
		return nil, err
	}
	return out, nil
}

// joinTypedValue 把整型/浮点列的文本化值归一回数值: fieldValueToDriver 对
// int32 等小整型走 fmt.Sprint 兜底成 []byte, 残余谓词/排序会因此变成字符串
// 比较 ("100" < "25"), 这里按列类型归一, 与 BIGINT 的输出形态 (int64) 对齐。
func joinTypedValue(ft types.FieldType, v driver.Value) driver.Value {
	if v == nil {
		return nil
	}
	var s string
	switch t := v.(type) {
	case []byte:
		s = string(t)
	case string:
		s = t
	default:
		return v
	}
	if isIntType(ft) {
		if n, err := strconv.ParseInt(strings.TrimSpace(s), 10, 64); err == nil {
			return n
		}
	}
	switch ft {
	case types.FieldType_Float64, types.FieldType_Float32:
		if f, err := strconv.ParseFloat(strings.TrimSpace(s), 64); err == nil {
			return f
		}
	}
	return v
}

// =============================================================================
// WHERE 按归属拆分: 下推条件 + 残余谓词
// =============================================================================

// joinWherePlan WHERE 拆分结果: perTable[i] 下推到第 i 张表扫描的 AND 条件,
// residual 为不可下推的表达式 (跨表 OR/NOT、列与列比较), 组合行上过滤。
type joinWherePlan struct {
	perTable []core.Condition
	residual expr
}

func planJoinWhere(ts []*joinTable, where expr, args []driver.NamedValue) (*joinWherePlan, error) {
	plan := &joinWherePlan{perTable: make([]core.Condition, len(ts))}
	if where == nil {
		return plan, nil
	}
	if err := splitJoinWhere(ts, where, args, plan); err != nil {
		return nil, err
	}
	return plan, nil
}

// splitJoinWhere 递归拆分 AND 结构; 叶子条件按列归属下推, 其余整体留作残余。
func splitJoinWhere(ts []*joinTable, e expr, args []driver.NamedValue, plan *joinWherePlan) error {
	switch t := e.(type) {
	case *andExpr:
		if err := splitJoinWhere(ts, t.l, args, plan); err != nil {
			return err
		}
		return splitJoinWhere(ts, t.r, args, plan)
	case *orExpr, *notExpr:
		// 结构条件: 整棵子树只涉及一张表仍可下推该表, 跨表则留作残余谓词
		owners := map[*joinTable]bool{}
		if err := exprOwners(ts, e, owners); err != nil {
			return err
		}
		if len(owners) == 1 {
			for jt := range owners {
				return pushJoinWhere(jt, e, args, plan)
			}
		}
		plan.residual = andResidual(plan.residual, e)
		return nil
	case *colCmpExpr:
		// 列与列比较: 引擎 Condition 只支持列与值, 一律留作残余 (列归属校验在此做)
		if _, _, err := ownerOfCol(ts, t.left); err != nil {
			return err
		}
		if _, _, err := ownerOfCol(ts, t.right); err != nil {
			return err
		}
		plan.residual = andResidual(plan.residual, e)
		return nil
	case *cmpExpr:
		return pushJoinLeaf(ts, t.col, e, args, plan)
	case *inExpr:
		return pushJoinLeaf(ts, t.col, e, args, plan)
	case *isNullExpr:
		return pushJoinLeaf(ts, t.col, e, args, plan)
	case *likeExpr:
		return pushJoinLeaf(ts, t.col, e, args, plan)
	}
	return fmt.Errorf("无法拆分的 WHERE 表达式")
}

// pushJoinLeaf 叶子条件按列归属下推到对应表。
func pushJoinLeaf(ts []*joinTable, ref string, e expr, args []driver.NamedValue, plan *joinWherePlan) error {
	jt, _, err := ownerOfCol(ts, ref)
	if err != nil {
		return err
	}
	return pushJoinWhere(jt, e, args, plan)
}

// pushJoinWhere 把表达式翻译后 AND 进对应表的下推条件 (先剥限定词再走标准翻译)。
func pushJoinWhere(jt *joinTable, e expr, args []driver.NamedValue, plan *joinWherePlan) error {
	dq, err := dequalifyExpr(e, jt.alias, jt.name)
	if err != nil {
		return err
	}
	cond, err := translateWhere(jt.tbl, dq, args)
	if err != nil {
		return err
	}
	if cond == nil {
		return nil
	}
	if plan.perTable[jt.idx] == nil {
		plan.perTable[jt.idx] = cond
	} else {
		plan.perTable[jt.idx] = core.And(plan.perTable[jt.idx], cond)
	}
	return nil
}

// andResidual 把不可下推的条件用 AND 串起来。
func andResidual(a, b expr) expr {
	if a == nil {
		return b
	}
	return &andExpr{l: a, r: b}
}

// exprOwners 收集表达式引用的全部表 (按指针去重); 列归属解析错误直接透出。
func exprOwners(ts []*joinTable, e expr, owners map[*joinTable]bool) error {
	mark := func(ref string) error {
		jt, _, err := ownerOfCol(ts, ref)
		if err != nil {
			return err
		}
		owners[jt] = true
		return nil
	}
	switch t := e.(type) {
	case *andExpr:
		if err := exprOwners(ts, t.l, owners); err != nil {
			return err
		}
		return exprOwners(ts, t.r, owners)
	case *orExpr:
		if err := exprOwners(ts, t.l, owners); err != nil {
			return err
		}
		return exprOwners(ts, t.r, owners)
	case *notExpr:
		return exprOwners(ts, t.e, owners)
	case *cmpExpr:
		return mark(t.col)
	case *colCmpExpr:
		if err := mark(t.left); err != nil {
			return err
		}
		return mark(t.right)
	case *inExpr:
		return mark(t.col)
	case *isNullExpr:
		return mark(t.col)
	case *likeExpr:
		return mark(t.col)
	}
	return nil
}

// =============================================================================
// 残余谓词求值: 组合行上的 Go 值比较
// =============================================================================

// evalJoinExpr 在组合行上求值残余谓词 (跨表 WHERE 条件)。
func evalJoinExpr(e expr, ts []*joinTable, r joinRow, args []driver.NamedValue) (bool, error) {
	switch t := e.(type) {
	case *andExpr:
		l, err := evalJoinExpr(t.l, ts, r, args)
		if err != nil || !l {
			return l, err
		}
		return evalJoinExpr(t.r, ts, r, args)
	case *orExpr:
		l, err := evalJoinExpr(t.l, ts, r, args)
		if err != nil || l {
			return l, err
		}
		return evalJoinExpr(t.r, ts, r, args)
	case *notExpr:
		v, err := evalJoinExpr(t.e, ts, r, args)
		if err != nil {
			return false, err
		}
		return !v, nil
	case *cmpExpr:
		jt, col, err := ownerOfCol(ts, t.col)
		if err != nil {
			return false, err
		}
		fi := findField(jt.tbl, col)
		raw, err := resolveValue(fi, t.val, args)
		if err != nil {
			return false, err
		}
		rv, err := fieldValueToDriverTyped(fi.FieldType, raw)
		if err != nil {
			return false, err
		}
		return compareOp(r.get(jt, col), rv, t.op), nil
	case *colCmpExpr:
		ljt, lcol, err := ownerOfCol(ts, t.left)
		if err != nil {
			return false, err
		}
		rjt, rcol, err := ownerOfCol(ts, t.right)
		if err != nil {
			return false, err
		}
		return compareOp(r.get(ljt, lcol), r.get(rjt, rcol), t.op), nil
	case *inExpr:
		jt, col, err := ownerOfCol(ts, t.col)
		if err != nil {
			return false, err
		}
		fi := findField(jt.tbl, col)
		lv := r.get(jt, col)
		hit := false
		for _, tv := range t.vals {
			raw, err := resolveValue(fi, tv, args)
			if err != nil {
				return false, err
			}
			ev, err := fieldValueToDriverTyped(fi.FieldType, raw)
			if err != nil {
				return false, err
			}
			if c, ok := compareDriverValues(lv, ev); ok && c == 0 {
				hit = true
				break
			}
		}
		return hit != t.neg, nil
	case *isNullExpr:
		jt, col, err := ownerOfCol(ts, t.col)
		if err != nil {
			return false, err
		}
		return (r.get(jt, col) == nil) != t.neg, nil
	case *likeExpr:
		jt, col, err := ownerOfCol(ts, t.col)
		if err != nil {
			return false, err
		}
		pat, err := resolveLikePattern(t.pattern, args)
		if err != nil {
			return false, err
		}
		s, ok := toCmpString(r.get(jt, col))
		if !ok {
			// NULL / 非文本: LIKE 不匹配 (NOT LIKE 相应为真)
			return t.neg, nil
		}
		return likeMatch(s, pat) != t.neg, nil
	}
	return false, fmt.Errorf("无法求值的 WHERE 表达式")
}

// compareOp 按比较符求值; 类型不可比视为不匹配。
func compareOp(l, r driver.Value, op string) bool {
	c, ok := compareDriverValues(l, r)
	if !ok {
		return false
	}
	switch op {
	case "=":
		return c == 0
	case "!=":
		return c != 0
	case "<":
		return c < 0
	case "<=":
		return c <= 0
	case ">":
		return c > 0
	case ">=":
		return c >= 0
	}
	return false
}

// compareDriverValues 归一化比较两个 driver.Value, 返回 -1/0/1; 类型不可比返回
// ok=false。NULL 视为最小 (两 NULL 相等); 数值按数值比, 文本按字典序比。
func compareDriverValues(a, b driver.Value) (int, bool) {
	if a == nil || b == nil {
		switch {
		case a == nil && b == nil:
			return 0, true
		case a == nil:
			return -1, true
		default:
			return 1, true
		}
	}
	if av, ok := toCmpFloat(a); ok {
		bv, ok := toCmpFloat(b)
		if !ok {
			return 0, false
		}
		switch {
		case av < bv:
			return -1, true
		case av > bv:
			return 1, true
		default:
			return 0, true
		}
	}
	if av, ok := toCmpString(a); ok {
		bv, ok := toCmpString(b)
		if !ok {
			return 0, false
		}
		return strings.Compare(av, bv), true
	}
	if av, ok := a.(time.Time); ok {
		bv, ok := b.(time.Time)
		if !ok {
			return 0, false
		}
		switch {
		case av.Before(bv):
			return -1, true
		case av.After(bv):
			return 1, true
		default:
			return 0, true
		}
	}
	return 0, false
}

// toCmpFloat 数值型 driver.Value → float64 (bool 按 0/1)。
func toCmpFloat(v driver.Value) (float64, bool) {
	switch t := v.(type) {
	case int64:
		return float64(t), true
	case float64:
		return t, true
	case bool:
		if t {
			return 1, true
		}
		return 0, true
	}
	return 0, false
}

// toCmpString 文本型 driver.Value → string。
func toCmpString(v driver.Value) (string, bool) {
	switch t := v.(type) {
	case string:
		return t, true
	case []byte:
		return string(t), true
	}
	return "", false
}

// likeMatch SQL LIKE 匹配: % 任意长度任意字符, _ 单字符, 其余原样。
func likeMatch(s, pat string) bool {
	sr, pr := []rune(s), []rune(pat)
	si, pi := 0, 0
	starP, starS := -1, 0
	for si < len(sr) {
		switch {
		case pi < len(pr) && (pr[pi] == '_' || pr[pi] == sr[si]):
			si++
			pi++
		case pi < len(pr) && pr[pi] == '%':
			starP, starS = pi, si
			pi++
		case starP != -1:
			pi = starP + 1
			starS++
			si = starS
		default:
			return false
		}
	}
	for pi < len(pr) && pr[pi] == '%' {
		pi++
	}
	return pi == len(pr)
}

// =============================================================================
// ON 解析 / 投影 / 排序
// =============================================================================

// resolvedJoinOn 定好左右的 ON 条件: right 指向本 JOIN 的新表, left 指向其左侧
// 已连接的表 (允许 "b.y = a.x" 写法, 解析期交换回来)。
type resolvedJoinOn struct {
	leftIdx  int
	leftCol  string
	rightIdx int
	rightCol string
}

func resolveJoinOn(ts []*joinTable, newIdx int, jc *joinClause) (resolvedJoinOn, error) {
	li, lcol, err := qualIndexCol(ts, jc.onL)
	if err != nil {
		return resolvedJoinOn{}, err
	}
	ri, rcol, err := qualIndexCol(ts, jc.onR)
	if err != nil {
		return resolvedJoinOn{}, err
	}
	if li == newIdx && ri < newIdx {
		li, lcol, ri, rcol = ri, rcol, li, lcol
	}
	if ri != newIdx || li >= newIdx {
		return resolvedJoinOn{}, fmt.Errorf("JOIN ON 条件 %q = %q 必须连接新表 %q 与其左侧的表", jc.onL, jc.onR, ts[newIdx].alias)
	}
	return resolvedJoinOn{leftIdx: li, leftCol: lcol, rightIdx: ri, rightCol: rcol}, nil
}

// qualIndexCol 解析 ON 里的限定列名为 (表下标, 列名); 必须是限定名且列存在。
func qualIndexCol(ts []*joinTable, ref string) (int, string, error) {
	qual, col := splitQualCol(ref)
	if qual == "" {
		return -1, "", fmt.Errorf("JOIN ON 列 %q 必须是限定名 (别名.列)", ref)
	}
	for i, jt := range ts {
		if strings.EqualFold(jt.alias, qual) || strings.EqualFold(jt.name, qual) {
			if findField(jt.tbl, col) == nil {
				return -1, "", fmt.Errorf("%w: %q", core.ErrFieldNotFound, ref)
			}
			return i, col, nil
		}
	}
	return -1, "", fmt.Errorf("未知的表限定 %q (列 %q)", qual, ref)
}

// joinProjCol 投影列: 归属表 + 小写列名 (行内取值) + 列元信息 + 展示名 (已去重)。
type joinProjCol struct {
	jt   *joinTable
	col  string
	fi   *types.FieldInfo
	name string
}

// buildJoinProjection 展开 SELECT 清单 (*/t.*/t.col/col 混用); 展示名为列名,
// 重名列用 "别名_列名" 去重。
func buildJoinProjection(ts []*joinTable, columns []string) ([]joinProjCol, error) {
	var out []joinProjCol
	seen := map[string]bool{}
	add := func(jt *joinTable, col string) error {
		fi := findField(jt.tbl, col)
		if fi == nil {
			return fmt.Errorf("%w: %q", core.ErrFieldNotFound, col)
		}
		name := string(fi.FieldName)
		if seen[strings.ToLower(name)] {
			name = jt.alias + "_" + name
			for seen[strings.ToLower(name)] {
				name += "_"
			}
		}
		seen[strings.ToLower(name)] = true
		out = append(out, joinProjCol{jt: jt, col: strings.ToLower(col), fi: fi, name: name})
		return nil
	}
	expand := func(jt *joinTable) error {
		for _, cn := range jt.cols {
			if err := add(jt, cn); err != nil {
				return err
			}
		}
		return nil
	}
	expandAll := func() error {
		for _, jt := range ts {
			if err := expand(jt); err != nil {
				return err
			}
		}
		return nil
	}
	if len(columns) == 0 {
		return out, expandAll()
	}
	for _, c := range columns {
		switch {
		case c == "*":
			if err := expandAll(); err != nil {
				return nil, err
			}
		case strings.HasSuffix(c, ".*"):
			jt := findQual(ts, c[:len(c)-2])
			if jt == nil {
				return nil, fmt.Errorf("未知的表限定 %q (列 %q)", c[:len(c)-2], c)
			}
			if err := expand(jt); err != nil {
				return nil, err
			}
		default:
			jt, col, err := ownerOfCol(ts, c) // 限定名/非限定名统一走归属解析
			if err != nil {
				return nil, err
			}
			if err := add(jt, col); err != nil {
				return nil, err
			}
		}
	}
	return out, nil
}

// joinOrderKey 排序键: 归属表 + 列 + 方向。
type joinOrderKey struct {
	jt   *joinTable
	col  string
	desc bool
}

// resolveJoinOrderKeys 解析 ORDER BY 键 (限定名直查, 非限定名按唯一归属)。
func resolveJoinOrderKeys(ts []*joinTable, specs []orderSpec) ([]joinOrderKey, error) {
	keys := make([]joinOrderKey, len(specs))
	for i, o := range specs {
		jt, col, err := ownerOfCol(ts, o.col)
		if err != nil {
			return nil, err
		}
		keys[i] = joinOrderKey{jt: jt, col: strings.ToLower(col), desc: o.desc}
	}
	return keys, nil
}

// =============================================================================
// queryJoin: JOIN 查询主流程
// =============================================================================

func queryJoin(db *core.DB, sel *selectStmt, args []driver.NamedValue, release func()) (driver.Rows, error) {
	if release != nil {
		defer release() // 物化结果集, 算完即放锁
	}
	if sel.agg != nil {
		return nil, fmt.Errorf("JOIN 与聚合函数暂不支持同时使用")
	}

	// a. 解析所有表并归一化别名 (别名冲突提前报错)
	refs := make([]tableRef, 0, len(sel.joins)+1)
	refs = append(refs, sel.from)
	for _, j := range sel.joins {
		refs = append(refs, j.table)
	}
	ts := make([]*joinTable, len(refs))
	seenAlias := map[string]bool{}
	for i, ref := range refs {
		tbl, err := findTable(db, ref.name)
		if err != nil {
			return nil, err
		}
		ts[i] = buildJoinTable(i, ref, tbl)
		key := strings.ToLower(ts[i].alias)
		if seenAlias[key] {
			return nil, fmt.Errorf("JOIN 表别名 %q 重复", ts[i].alias)
		}
		seenAlias[key] = true
	}

	// b. WHERE 按归属拆分: 单表条件下推, 跨表条件留作残余谓词
	plan, err := planJoinWhere(ts, sel.where, args)
	if err != nil {
		return nil, err
	}

	// c. 左表扫描 (本表条件下推), 每行包一层别名空间
	leftRows, err := scanJoinTable(ts[0], plan.perTable[0])
	if err != nil {
		return nil, err
	}
	rows := make([]joinRow, 0, len(leftRows))
	for _, m := range leftRows {
		rows = append(rows, joinRow{strings.ToLower(ts[0].alias): m})
	}

	// d. 逐 JOIN: 左行 ON 列值 → 右表索引点查 (+右表条件下推), 组合笛卡尔行
	for i := range sel.joins {
		jc := &sel.joins[i]
		jt := ts[i+1]
		on, err := resolveJoinOn(ts, i+1, jc)
		if err != nil {
			return nil, err
		}
		onFi := findField(jt.tbl, on.rightCol)
		next := make([]joinRow, 0, len(rows))
		for _, lr := range rows {
			onVal := lr.get(ts[on.leftIdx], on.leftCol)
			var matched []map[string]driver.Value
			if onVal != nil {
				// NULL 与任何值都不相等, 右表点查直接跳过 (INNER 丢行 / LEFT 补 NULL)
				ev, err := driverToEngineValue(onFi, onVal)
				if err != nil {
					return nil, fmt.Errorf("JOIN ON 列 %q: %w", on.rightCol, err)
				}
				cond := core.Eq(on.rightCol, ev, onFi.FieldType)
				if plan.perTable[i+1] != nil {
					cond = core.And(cond, plan.perTable[i+1])
				}
				matched, err = scanJoinTable(jt, cond)
				if err != nil {
					return nil, err
				}
			}
			if len(matched) == 0 {
				if jc.left {
					nr := cloneJoinRow(lr)
					nr[strings.ToLower(jt.alias)] = nullJoinRow(jt)
					next = append(next, nr)
				}
				continue // INNER JOIN: 无匹配丢行
			}
			for _, rm := range matched {
				nr := cloneJoinRow(lr)
				nr[strings.ToLower(jt.alias)] = rm
				next = append(next, nr)
			}
		}
		rows = next
	}

	// e. 残余谓词: 组合行上过滤 (跨表比较)
	if plan.residual != nil {
		kept := make([]joinRow, 0, len(rows))
		for _, r := range rows {
			ok, err := evalJoinExpr(plan.residual, ts, r, args)
			if err != nil {
				return nil, err
			}
			if ok {
				kept = append(kept, r)
			}
		}
		rows = kept
	}

	// f. 投影列清单 (*/t.*/t.col/col 混用, 重名列 别名_列名 去重)
	proj, err := buildJoinProjection(ts, sel.columns)
	if err != nil {
		return nil, err
	}

	// g. ORDER BY: 投影前对组合行内存稳定排序 (NULL 视为最小), 再 LIMIT/OFFSET
	if len(sel.orderBy) > 0 {
		keys, err := resolveJoinOrderKeys(ts, sel.orderBy)
		if err != nil {
			return nil, err
		}
		sort.SliceStable(rows, func(x, y int) bool {
			for _, k := range keys {
				c, ok := compareDriverValues(rows[x].get(k.jt, k.col), rows[y].get(k.jt, k.col))
				if !ok || c == 0 {
					continue
				}
				if k.desc {
					return c > 0
				}
				return c < 0
			}
			return false
		})
	}
	if sel.offset != nil {
		o := int(*sel.offset)
		if o >= len(rows) {
			rows = nil
		} else {
			rows = rows[o:]
		}
	}
	if sel.limit != nil && int(*sel.limit) < len(rows) {
		rows = rows[:int(*sel.limit)]
	}

	// 物化投影行 (DISTINCT 按投影值去重)
	colNames := make([]string, len(proj))
	metas := make([]colMeta, len(proj))
	for i, pc := range proj {
		colNames[i] = pc.name
		metas[i] = colMetaFor(pc.fi)
	}
	data := make([][]driver.Value, 0, len(rows))
	var seen map[string]bool
	if sel.distinct {
		seen = map[string]bool{}
	}
	for _, r := range rows {
		vals := make([]driver.Value, len(proj))
		for i, pc := range proj {
			vals[i] = r.get(pc.jt, pc.col) // LEFT JOIN 未匹配侧 → NULL
		}
		if seen != nil {
			key := distinctKey(vals)
			if seen[key] {
				continue
			}
			seen[key] = true
		}
		data = append(data, vals)
	}
	return materialRows(colNames, metas, data), nil
}