package core import ( "TsDB2/DB/core/index" "TsDB2/DB/core/page" "TsDB2/DB/core/types" "bytes" "encoding/binary" "fmt" "math/big" "sort" ) // Query 查询操作对象 type Query struct { table *Table selectCols []string // SELECT 字段列表 whereCond Condition // WHERE 条件 orderBys []SortField // ORDER BY limit int // LIMIT offset int // OFFSET groupBys []string // GROUP BY havingCond Condition // HAVING } type SortField struct { Field string Descending bool } // scanPlan 索引扫描计划 type scanPlan struct { meta *IndexMeta handler index.IndexHandler startKey []byte endKey []byte startInclusive bool endInclusive bool rowStore bool // 该索引是否承载行数据 (主键/时序=true) remainingCond Condition // 需要行级过滤的剩余条件 } // Query 返回查询操作对象 func (table *Table) Query() *Query { return &Query{table: table} } func (q *Query) Select(cols ...string) *Query { q.selectCols = cols; return q } func (q *Query) Where(cond Condition) *Query { q.whereCond = cond; return q } func (q *Query) Limit(n int) *Query { q.limit = n; return q } func (q *Query) Offset(n int) *Query { q.offset = n; return q } func (q *Query) GroupBy(cols ...string) *Query { q.groupBys = cols; return q } func (q *Query) Having(cond Condition) *Query { q.havingCond = cond; return q } func (q *Query) OrderBy(field string, descending ...bool) *Query { desc := false if len(descending) > 0 { desc = descending[0] } q.orderBys = append(q.orderBys, SortField{Field: field, Descending: desc}) return q } // ======================== 索引键编码 ======================== // encodeIndexKey 将条件值编码为索引键字节。 // // 编码与行内存储【同源】(都走 types.EncodeValue): 索引的 B+ 树直接按字节序比较键, // 因此"索引键字节 == 行内字节"是范围扫描正确的前提。这里不再按类型手写分支 —— // 以前每加一个类型都要在这里补一份, 漏补的后果是索引选出来的行与过滤结果不一致。 // // param 是列级类型参数: decimal 是小数位数, 字符串家族是行内宽度码。少了它就没法把 // 条件值编成与行内完全一致的字节 —— decimal 会因 scale 不同而长度/系数不一致, // 字符串则会按错误的宽度截断或越界, 两种都让"索引选出 5 行、过滤后剩 0 行"。 // // 键长必须等于 FieldSize(ft, param): B+ 树按【定长拼键】比较, 键长与行内宽度不一致 // 就等于拿两种编码去比大小。 func encodeIndexKey(value interface{}, ft types.FieldType, param byte) []byte { if types.IsDynamic(ft) { // 动态列不能作为索引分量 (CreateIndex 已拒绝), 这里仅作兜底 if b, ok := value.([]byte); ok { return b } return nil } size := types.FieldSize(ft, param) if size == 0 { // 未注册类型 / 非法列参数: 交给下面 EncodeValue 报出具体原因 (buf 长度为 0) panic(fmt.Errorf("%w: 类型 %s 的宽度不可计算 (列级参数 %d)", types.ErrValueType, types.TypeNameOf(ft), param)) } buf := make([]byte, size) if err := types.EncodeValue(ft, param, value, buf); err != nil { // 沿用既有约定: panic 由 encodeIndexKeySafe* 转成 error。 panic(err) } return buf } // encodeIndexKeySafe 在 encodeIndexKey 外面包一层 panic 恢复。 // 值类型与列不符时 encodeIndexKey 会 panic, 而公开 API 传进来的值不可信, // 必须转成 error, 不能把用户输入直接变成崩溃。 func encodeIndexKeySafe(value interface{}, ft types.FieldType) (enc []byte, err error) { return encodeIndexKeySafeWithParam(value, ft, 0) } // encodeIndexKeySafeWithParam 同 encodeIndexKeySafe, 但带上列级类型参数。 // 查询侧拿得到字段元信息时必须走这个版本 (decimal 需要 param 才能编出正确的字节)。 func encodeIndexKeySafeWithParam(value interface{}, ft types.FieldType, param byte) (enc []byte, err error) { defer func() { if r := recover(); r != nil { enc, err = nil, fmt.Errorf("值 %v 与存储类型 %s 不匹配: %v", value, types.TypeNameOf(ft), r) } }() return encodeIndexKey(value, ft, param), nil } // ======================== 索引匹配 ======================== // fieldNameFromMeta 获取索引第一个字段的名称 func (q *Query) fieldNameFromMeta(meta *IndexMeta) string { fieldIdx := int(meta.FieldIDs[0]) if fieldIdx < len(q.table.Field) { return string(q.table.Field[fieldIdx].FieldName) } return "" } // ======================== 复合索引键规划 ======================== // // 索引键 = 各分量字段定长拼接 (编码与 extractIndexKey 完全一致), 因此: // - 已知前 k 个分量等值 → 键上一段连续区间; // - 第 k+1 个分量上的范围条件 → 再收窄这段区间; // - 更靠后的分量索引帮不上忙, 原样留作行级后置过滤。 // 下面的代码把 WHERE 按这个模型折叠成 (startKey, endKey, 残余条件)。 // // 单列索引是 n == 1 的特例, 行为与旧实现一致。 // indexKeyLayout 索引键各分量的偏移与长度 (定长拼接) type indexKeyLayout struct { offsets []int sizes []int total int } // cmpKind 叶子条件能为键分量提供的边界方向 type cmpKind int const ( cmpEq cmpKind = iota cmpGe cmpGt cmpLe cmpLt ) // keyConstraint 某个键分量上的合并边界 (多个 AND 子条件取交集) type keyConstraint struct { lo, hi []byte // 编码后的分量值; nil 表示该侧无约束 loInc, hiInc bool count int // 落在该分量上的条件个数, 0 = 无约束 } // add 把一个边界并入本分量; 返回 false 表示条件自相矛盾。 func (kc *keyConstraint) add(kind cmpKind, enc []byte) bool { switch kind { case cmpEq: if !kc.addLower(enc, true) || !kc.addUpper(enc, true) { return false } case cmpGe: if !kc.addLower(enc, true) { return false } case cmpGt: if !kc.addLower(enc, false) { return false } case cmpLe: if !kc.addUpper(enc, true) { return false } case cmpLt: if !kc.addUpper(enc, false) { return false } } kc.count++ return true } func (kc *keyConstraint) addLower(enc []byte, inclusive bool) bool { if kc.lo == nil { kc.lo, kc.loInc = enc, inclusive return true } switch c := bytes.Compare(enc, kc.lo); { case c > 0: // 取更高的下界 kc.lo, kc.loInc = enc, inclusive case c == 0: // 同值: 两端都闭才算"含" kc.loInc = kc.loInc && inclusive } return kc.notContradictory() } func (kc *keyConstraint) addUpper(enc []byte, inclusive bool) bool { if kc.hi == nil { kc.hi, kc.hiInc = enc, inclusive return true } switch c := bytes.Compare(enc, kc.hi); { case c < 0: // 取更低的上界 kc.hi, kc.hiInc = enc, inclusive case c == 0: kc.hiInc = kc.hiInc && inclusive } return kc.notContradictory() } // notContradictory 判断 [lo, hi] 是否还可能有值。 // // 注意: 这里按编码字节比较, 而定长整数用的是补码 —— 负数的字节序与数值序相反。 // 所以它只在"字节序下已确定矛盾"时才返回 false, 此时真值也必然为空, 反向不成立。 // 调用方因此只把它当作"放弃使用索引"的依据, 绝不据此断言区间非空。 func (kc *keyConstraint) notContradictory() bool { if kc.lo == nil || kc.hi == nil { return true } switch c := bytes.Compare(kc.lo, kc.hi); { case c > 0: return false case c == 0: return kc.loInc && kc.hiInc } return true } // isEquality 该分量已被钉死为单值 (lo == hi 且两端都闭) func (kc *keyConstraint) isEquality() bool { return kc.count > 0 && kc.lo != nil && kc.hi != nil && kc.loInc && kc.hiInc && bytes.Equal(kc.lo, kc.hi) } // buildKeyLayout 依据索引元信息与表字段类型算出键布局。 // 返回 false 表示该索引当前不可用于查询: // - 分量列不存在或已被 Drop (新行提不出该分量, 索引里根本没有这批键); // - 分量是动态列 (行内只存 6B 堆引用, 值 != 引用, 无法按键定位)。 func (q *Query) buildKeyLayout(meta *IndexMeta) (*indexKeyLayout, bool) { n := int(meta.FieldCnt) if n <= 0 || n > len(meta.FieldIDs) || n > len(q.table.Field) { return nil, false } lay := &indexKeyLayout{offsets: make([]int, n), sizes: make([]int, n)} off := 0 for k := 0; k < n; k++ { fi := int(meta.FieldIDs[k]) if fi >= len(q.table.Field) { return nil, false } if q.table.Field[fi].IsDropped() { return nil, false } ft := q.table.Field[fi].FieldType if types.IsDynamic(ft) { return nil, false } // 键分量宽度必须与行内宽度同源 (含字符串列的宽度码), 否则索引键的偏移切不出该列 size := int(q.table.Field[fi].Width()) if size <= 0 { return nil, false } lay.offsets[k] = off lay.sizes[k] = size off += size } lay.total = off return lay, true } // locateComponent 返回字段 master 下标在索引键中的分量位置; -1 表示不属于该索引 func (q *Query) locateComponent(fieldName string, meta *IndexMeta) int { fi := q.table.findFieldIndex(fieldName) if fi < 0 { return -1 } n := int(meta.FieldCnt) if n > len(meta.FieldIDs) { n = len(meta.FieldIDs) } for k := 0; k < n; k++ { if int(meta.FieldIDs[k]) == fi { return k } } return -1 } // classifyLeaf 判断一个叶子条件能否成为某个键分量的边界。 // comp < 0 表示与该索引无关 (字段不在索引里, 或该条件类型给不出可比边界)。 func (q *Query) classifyLeaf(c Condition, meta *IndexMeta) (comp int, value interface{}, kind cmpKind) { switch t := c.(type) { case EqCondition: return q.locateComponent(t.Field, meta), t.Value, cmpEq case GeCondition: return q.locateComponent(t.Field, meta), t.Value, cmpGe case GtCondition: return q.locateComponent(t.Field, meta), t.Value, cmpGt case LeCondition: return q.locateComponent(t.Field, meta), t.Value, cmpLe case LtCondition: return q.locateComponent(t.Field, meta), t.Value, cmpLt default: return -1, nil, cmpEq } } // flattenAnd 把 AND 树展平成叶子条件。 // OR / NOT / LIKE / IN 不是合取项, 整体当叶子返回 —— 它们给不出索引边界, // 但必须原样保留下来做行级过滤。 func flattenAnd(cond Condition, out *[]Condition) { if and, ok := cond.(AndCondition); ok { for _, sub := range and.Conditions { flattenAnd(sub, out) } return } *out = append(*out, cond) } // incBE 把 buf 当作大端无符号整数加一 (原地)。返回 false 表示溢出 —— // 此时 buf 之后不再存在任何更大的键。 func incBE(buf []byte) bool { for i := len(buf) - 1; i >= 0; i-- { buf[i]++ if buf[i] != 0 { return true } } return false } // fillFF 把 buf[from:] 全置 0xFF (该分量及其后分量的最大值)。 // from 可以是 0 —— 首分量就是范围分量时前缀为空, 此时整个键都无上界。 func fillFF(buf []byte, from int) { for i := from; i < len(buf); i++ { buf[i] = 0xFF } } // buildScanPlan 分析 WHERE 条件, 选择最优索引 func (q *Query) buildScanPlan() *scanPlan { if q.whereCond == nil { return nil } if q.table.indexMeta == nil { return nil } // 先展平成合取项: 索引区间是按"分量"折叠出来的, 只有 AND 语义才可加。 var leaves []Condition flattenAnd(q.whereCond, &leaves) var bestPlan *scanPlan for i := range q.table.indexMeta.metas { m := &q.table.indexMeta.metas[i] if m.Flag&0x80 != 0 { continue } // 主键索引直接承载全部行, 但它的 B+ 树叶键是 pkID, 而不是"建在哪一列上"。 // planIndex 会用 meta.FieldIDs 编码出的【字段值】当扫描区间 (见 buildKeyLayout), // 拿它去比对 pkID 叶键, 语义对不上 —— 只有当该列的值恒等于 pkID 时才偶然正确, // 否则会静默地只命中/漏掉若干行 (例如 Delete/Update 的条件扫描)。 // 因此主键索引不参与列条件规划, 这类查询退回全表扫描 + 行级过滤 (fullTableScan // 会施加 whereCond), 结果正确; 按 pkID 的点查/回表仍走 SearchByIndex/ResolveByPK, // 不经过这里, 不受影响。 if m.IndexType == page.IndexTypePrimary { continue } handler, ok := index.Compose.GetHandler(m.IndexType) if !ok { continue } plan := q.planIndex(m, handler, leaves) if plan == nil { continue // 此索引不匹配 } // 该索引是否承载行数据: 由处理器自报, 本层不认识索引类型 plan.rowStore = handler.IsRowStore() if plan.rowStore { return plan // 行存储索引最优 (免回表), 不再找 } if bestPlan == nil { bestPlan = plan } } return bestPlan } // planIndex 为一个索引推导扫描区间; 该索引用不上时返回 nil。 func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Condition) *scanPlan { lay, ok := q.buildKeyLayout(m) if !ok { return nil } n := len(lay.sizes) cons := make([]keyConstraint, n) // 各分量合并后的边界 compCond := make([][]Condition, n) // 各分量上的原始条件 (用于决定哪些能丢) var rest []Condition // 与索引无关、必须保留的条件 for _, c := range leaves { k, value, kind := q.classifyLeaf(c, m) if k < 0 { rest = append(rest, c) continue } ft := q.table.Field[m.FieldIDs[k]].FieldType // 用 Safe 版本: 条件的值类型与列存储类型不符时 encodeIndexKey 会 panic, // 规划阶段不该把用户输入直接变成崩溃 —— 退回行级过滤由 Match 去处理。 // 带上列级 param (decimal 的 scale / 字符串的宽度码): 索引键必须与行内字节完全一致。 enc, err := encodeIndexKeySafeWithParam(value, ft, q.table.fieldParam(int(m.FieldIDs[k]))) if err != nil || len(enc) != lay.sizes[k] { // 编码失败, 或长度与分量长度不符 (拼进定长键会错位) → 交给行级过滤 rest = append(rest, c) continue } if !cons[k].add(kind, enc) { // 该分量上的条件互相矛盾 → 结果必为空。 // 这里保守地"放弃索引"而不是返回空区间: 定长补码编码下负数字节序与 // 数值序相反, 用字节比较断定"空"对负数不成立, 退回全表扫描更稳。 return nil } compCond[k] = append(compCond[k], c) } if cons[0].count == 0 { return nil // 首分量没有边界 → 索引给不出连续区间 } // 从左往右消费"等值分量"; 之后最多再消费一个范围分量, 更后面的分量帮不上忙。 eqLen := 0 for eqLen < n && cons[eqLen].isEquality() { eqLen++ } rangeIdx := -1 if eqLen < n && cons[eqLen].count > 0 { rangeIdx = eqLen } startKey := make([]byte, lay.total) endKey := make([]byte, lay.total) prefix := 0 for j := 0; j < eqLen; j++ { copy(startKey[prefix:], cons[j].lo) copy(endKey[prefix:], cons[j].lo) prefix += len(cons[j].lo) } plan := &scanPlan{ meta: m, handler: handler, startKey: startKey, endKey: endKey, startInclusive: true, endInclusive: true, remainingCond: q.remainCond(rest, compCond, eqLen, rangeIdx), } // 全部等值 → 精确键 (此时 prefix == total, fillFF 无事可做); // 只消费了等值前缀 → 该前缀下的全部键: [前缀||00.., 前缀||FF..]。 if eqLen == n || rangeIdx < 0 { fillFF(endKey, prefix) return plan } rc := &cons[rangeIdx] size := lay.sizes[rangeIdx] // 下界: [前缀 || 分量值 || 00..] if rc.lo != nil { copy(startKey[prefix:], rc.lo) if !rc.loInc { // 严格下界: 要跳过分量 == lo 的全部键, 即把 (前缀||lo) 整体加一 if !incBE(startKey[:prefix+size]) { return nil // 已经是最大键, 该条件不可能有匹配行 } } } // 上界: [前缀 || 分量值 || FF..] if rc.hi == nil { fillFF(endKey, prefix) } else if rc.hiInc { copy(endKey[prefix:], rc.hi) fillFF(endKey, prefix+size) } else { // 严格上界: 分量 == hi 的键全部排除 copy(endKey[prefix:], rc.hi) plan.endInclusive = false } return plan } // remainCond 汇总索引区间没能表达、仍需逐行过滤的条件。 // 已被区间精确覆盖的分量 (等值前缀 + 那个范围分量) 上的条件可以安全丢掉。 func (q *Query) remainCond(rest []Condition, compCond [][]Condition, eqLen, rangeIdx int) Condition { remaining := rest for j := range compCond { if j < eqLen || j == rangeIdx { continue } remaining = append(remaining, compCond[j]...) } switch len(remaining) { case 0: return nil case 1: return remaining[0] default: return And(remaining...) } } // rowWithPK 把结果行与它的 pkID 绑在一起 —— 排序/翻页会重排结果集, pkID 必须 // 跟着行一起移动, 否则回调会把 A 行的 pkID 配到 B 行的数据上。 type rowWithPK struct { pkID uint64 row []types.FieldValue } // compareRowsForOrderBy 按 OrderBy 规则比较两行, 返回 -1/0/1。 // 0 表示"所有排序列都无法区分"(字段缺失), 由调用方用稳定排序保持原顺序。 func (q *Query) compareRowsForOrderBy(a, b []types.FieldValue) int { for _, ob := range q.orderBys { idx := q.table.findFieldIndex(ob.Field) if idx < 0 { continue } vi := findFieldValue(a, ob.Field) vj := findFieldValue(b, ob.Field) if vi == nil || vj == nil { continue } cmp := compareFieldValues(vi, vj, q.table.Field[idx].FieldType) if cmp == 0 { continue } if ob.Descending { return -cmp } return cmp } return 0 } // ======================== 核心查询 ======================== // Find 执行查询,返回多行 func (q *Query) Find() ([][]types.FieldValue, error) { var results [][]types.FieldValue err := q.scan(func(row []types.FieldValue) bool { results = append(results, row) return true }) if err != nil { return nil, err } // 投影 if len(q.selectCols) > 0 { results = q.project(results) } return results, nil } // Scan 流式扫描查询, 回调函数逐行处理数据, 返回 false 提前终止 // 适用于不需要全量收集结果的场景, 内存友好 func (q *Query) Scan(callback func(row []types.FieldValue) bool) error { return q.scan(callback) } // scan 内部流式扫描核心方法 // 自动处理: 索引选择 → 排序 → OFFSET → LIMIT → 回调 func (q *Query) scan(callback func(row []types.FieldValue) bool) error { return q.scanWithPKID(func(_ uint64, row []types.FieldValue) bool { return callback(row) }) } // ScanWithPKID 流式扫描, 回调同时拿到行主键号 pkID 与列值。 // // 迁移/导出工具必须用这个而不是 Scan: 行数据里不含 pkID (它是行的 8 字节前缀, // 不是字段), 只拿字段值就没法还原"这一行的主键号是多少" —— 导出到 SQL 时要么 // 丢掉真实主键, 要么被迫用某个列值冒充 (列值可能是 0、可能被改过)。 func (q *Query) ScanWithPKID(callback func(pkID uint64, row []types.FieldValue) bool) error { return q.scanWithPKID(callback) } // scanWithPKID 与 scan 完全一致, 但额外把每行的真实 pkID 交给回调。 // // 为什么需要: 行数据里不含 pkID (它是行的 8 字节前缀, 不是字段), 普通 Scan 回调 // 只拿得到字段值。删除/更新必须按真实 pkID 定位行, 旧实现只能退而求其次地 // "取第一个 uint64 类型的字段值当 pkID" —— 主键列不是行内第一个 8 字节列时, // 它删掉的是另一个主键指向的行, 匹配上的行反而一条都没动, 而且不报错。 func (q *Query) scanWithPKID(callback func(pkID uint64, row []types.FieldValue) bool) error { plan := q.buildScanPlan() // 判断是否需要内存排序 (OrderBy 字段与索引不一致时) needSort := false if len(q.orderBys) > 0 { if plan == nil { needSort = true } else { idxField := q.fieldNameFromMeta(plan.meta) if !(len(q.orderBys) == 1 && q.orderBys[0].Field == idxField) { needSort = true } } } // 需要排序时, 必须先收集全部再排序, 无法流式 LIMIT if needSort { var all []rowWithPK err := q.executeScan(func(pkID uint64, row []types.FieldValue) bool { all = append(all, rowWithPK{pkID, row}) return true }, plan) if err != nil { return err } // 排序必须连 pkID 一起搬: 只排 []FieldValue 会让回调拿到错位的 pkID sort.SliceStable(all, func(i, j int) bool { return q.compareRowsForOrderBy(all[i].row, all[j].row) < 0 }) // OFFSET if q.offset > 0 { if q.offset >= len(all) { return nil } all = all[q.offset:] } // LIMIT if q.limit > 0 && q.limit < len(all) { all = all[:q.limit] } for _, item := range all { if !callback(item.pkID, item.row) { return nil } } return nil } // 不需要排序: 直接流式传递 callback, 支持 LIMIT 提前终止 seen := 0 return q.executeScan(func(pkID uint64, row []types.FieldValue) bool { if seen < q.offset { seen++ return true // 跳过 OFFSET } seen++ if !callback(pkID, row) { return false // 用户要求停止 } // LIMIT 检查 (0 表示无限制) if q.limit > 0 && seen-q.offset >= q.limit { return false // 已满足, 提前终止扫描 } return true }, plan) } // executeScan 执行索引扫描或全表扫描, 将反序列化后的行传递给回调 func (q *Query) executeScan(callback func(pkID uint64, row []types.FieldValue) bool, plan *scanPlan) error { if plan != nil { return q.indexScan(plan, callback) } return q.fullTableScan(callback) } // indexScan 使用索引扫描并按"先物化、后回表"两阶段执行。 // // 为什么不能像旧实现那样在 RangeScan 回调里直接回表 + 触发用户 callback: // RangeScan 是沿叶子双向链 (PageHdr.NextPage) 推进的裸游标, 而用户的 Scan 回调 // 常常会对【同一棵索引】做 Update/Delete —— 删旧条目 + 插新条目会触发叶分裂/合并, // 被合并掉的叶子页会被登记回收, 下一个 Commit/drain 时该页被清零并改写成空闲链节点 // (空闲链同样用 PageHdr.NextPage 串接)。此时游标再读该页的 NextPage, 拿到的就是 // 空闲链指针, 于是跳到一张 type=0 的空页上 (实测 panic: bad page type=0)。 // // 物化把"遍历"与"修改"彻底分开: 遍历阶段树保持只读, 游标始终有效; 修改发生在 // 遍历结束之后, 不再影响游标。二级索引只物化 pkID(8B/行), 内存开销与结果集成正比 // 且远小于整行物化; rowStore 索引物化的是行的快照 (必须拷贝, 因为它是页内缓冲视图)。 func (q *Query) indexScan(plan *scanPlan, callback func(pkID uint64, row []types.FieldValue) bool) error { // 判断 OrderBy 是否需要反向 reverse := false if len(q.orderBys) == 1 { idxField := q.fieldNameFromMeta(plan.meta) if q.orderBys[0].Field == idxField && q.orderBys[0].Descending { reverse = true } } if plan.rowStore { // 承载行数据的索引: RangeScan 返回 storeRow = [pkID(8B) + rowData]。 // storeRow 指向页缓冲, 遍历结束后页可能被回收改写, 必须拷贝一份快照。 var rows [][]byte if err := plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage, plan.startKey, plan.endKey, plan.startInclusive, plan.endInclusive, reverse, func(row []byte) bool { if len(row) < 8 { return true } snapshot := make([]byte, len(row)) copy(snapshot, row) rows = append(rows, snapshot) return true }); err != nil { return err } for _, row := range rows { rowData := row[8:] if plan.remainingCond != nil { if !plan.remainingCond.Match(rowData, q.table) { continue } } if !callback(binary.BigEndian.Uint64(row[:8]), q.table.deserializeRow(rowData)) { return nil } } return nil } // 二级索引: RangeScan 返回 pkID(8B BE)。只按顺序物化 pkID, 回表放到遍历之后。 var pkIDs []uint64 if err := plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage, plan.startKey, plan.endKey, plan.startInclusive, plan.endInclusive, reverse, func(row []byte) bool { if len(row) < 8 { return true } pkIDs = append(pkIDs, binary.BigEndian.Uint64(row[:8])) return true }); err != nil { return err } for _, pkID := range pkIDs { storeRow, found := q.table.indexMeta.ResolveByPK(pkID) if !found { // 行已被之前的回调删除, 跳过 continue } rowData := storeRow[8:] if plan.remainingCond != nil { if !plan.remainingCond.Match(rowData, q.table) { continue } } if !callback(pkID, q.table.deserializeRow(rowData)) { return nil } } return nil } // fullTableScan 全表扫描 (流式回调模式) func (q *Query) fullTableScan(callback func(pkID uint64, row []types.FieldValue) bool) error { // 找承载行数据的索引 (由处理器自报, 本层不认识索引类型) primMeta, primHandler, ok := q.table.indexMeta.rowStoreMeta() if !ok || primMeta == nil || primHandler == nil { return nil } dpIDs := primHandler.GetDataPageIDs(q.table.handle, primMeta.EntryPage) for _, dpID := range dpIDs { rows := primHandler.ReadDataPageRows(q.table.handle, dpID) for _, storeRow := range rows { if len(storeRow) < 8 { continue } rowData := storeRow[8:] if q.whereCond != nil { if !q.whereCond.Match(rowData, q.table) { continue } } if !callback(binary.BigEndian.Uint64(storeRow[:8]), q.table.deserializeRow(rowData)) { return nil // 回调要求提前终止 } } } return nil } // ======================== 排序 ======================== func (q *Query) sortResults(results [][]types.FieldValue) { sort.SliceStable(results, func(i, j int) bool { return q.compareRowsForOrderBy(results[i], results[j]) < 0 }) } func findFieldValue(row []types.FieldValue, field string) interface{} { for _, fv := range row { if string(fv.FieldName) == field { return fv.Value } } return nil } // cmpInt64 / cmpUint64 把三态比较收敛到一处, 避免每个分支重复写一遍 if 链。 func cmpInt64(a, b int64) int { switch { case a < b: return -1 case a > b: return 1 } return 0 } func cmpUint64(a, b uint64) int { switch { case a < b: return -1 case a > b: return 1 } return 0 } // compareFieldValues 按【字段类型】比较同一列的两个值, 返回 -1/0/1。 // // 类型必须由调用方传入, 不能只看 Go 值类型: 同一列都可能读成 int16, 但它既可能来自 // 有符号的 FieldType_Int16 (int16 语义), 也可能来自无符号的 FieldType_2byte —— // 后者的值是 uint16 的位模式被重解释成 int16, 两者序不同 (位模式 0xFFFF: int16 是 -1, // uint16 是 65535)。旧实现只按 Go 值类型分派, 既分不清这两种, 又整个漏掉了 int8/int64。 // // 有符号类型比的是【解码后的值】而不是盘上字节: 盘上是保序编码 (符号位已翻转), // 直接比字节也对, 但这里拿到的是已解码的值, 比数值更直白。 func compareFieldValues(a, b interface{}, ft types.FieldType) int { switch ft { case types.FieldType_Int8: return cmpInt64(int64(a.(int8)), int64(b.(int8))) case types.FieldType_Int16: return cmpInt64(int64(a.(int16)), int64(b.(int16))) case types.FieldType_Int32: return cmpInt64(int64(a.(int32)), int64(b.(int32))) case types.FieldType_Int64: return cmpInt64(a.(int64), b.(int64)) case types.FieldType_1byte: return cmpUint64(uint64(a.(byte)), uint64(b.(byte))) case types.FieldType_2byte: // 无符号: 先把 int16 位模式还原回 uint16 再比, 否则 > 32767 的值会被当成负数 return cmpUint64(uint64(uint16(a.(int16))), uint64(uint16(b.(int16)))) case types.FieldType_4byte: return cmpUint64(uint64(uint32(a.(int32))), uint64(uint32(b.(int32)))) case types.FieldType_8byte: return cmpUint64(a.(uint64), b.(uint64)) case types.FieldType_Float64: // 同一列读出的一定是同一类型, 故 b 也是 float64; 容错地走 toFloat64 fa, fb := toFloat64(a), toFloat64(b) switch { case fa < fb: return -1 case fa > fb: return 1 } return 0 case types.FieldType_Float32: fa, fb := a.(float32), b.(float32) switch { case fa < fb: return -1 case fa > fb: return 1 } return 0 case types.FieldType_Bool: // false < true (与盘上 0x00 < 0x01 的字节序一致) ba, bb := a.(bool), b.(bool) switch { case ba == bb: return 0 case !ba: return -1 } return 1 case types.FieldType_Year: // 同 2byte: 载体是 int16, 但语义是无符号年份 return cmpUint64(uint64(uint16(a.(int16))), uint64(uint16(b.(int16)))) case types.FieldType_Decimal: // 走数值比较 (自动对齐 scale): 同一列的 scale 必然相同, 但比较函数不该依赖这一点 return a.(types.Decimal).Cmp(b.(types.Decimal)) case types.FieldType_Int128: na, oka := a.(*big.Int) nb, okb := b.(*big.Int) if !oka || !okb { return 0 } return na.Cmp(nb) case types.FieldType_UUID: ua, oka := a.(types.UUID) ub, okb := b.(types.UUID) if !oka || !okb { return 0 } return bytes.Compare(ua[:], ub[:]) } // 动态字段与所有定长字符串: 按字节序 ba, ok := a.([]byte) if !ok { return 0 } bb, ok := b.([]byte) if !ok { return 0 } return bytes.Compare(ba, bb) } // fieldTypeOf 取某列的存储类型, 供按类型比较/聚合使用; 列不存在时返回 0。 func (q *Query) fieldTypeOf(name string) types.FieldType { if i := q.table.findFieldIndex(name); i >= 0 { return q.table.Field[i].FieldType } return 0 } // ======================== 投影 ======================== func (q *Query) project(results [][]types.FieldValue) [][]types.FieldValue { colSet := make(map[string]bool) for _, c := range q.selectCols { colSet[c] = true } for i, row := range results { var filtered []types.FieldValue for _, fv := range row { if colSet[string(fv.FieldName)] { filtered = append(filtered, fv) } } results[i] = filtered } return results } // ======================== 流式聚合 ======================== // FindOne 流式查询,返回单行 (命中即停,不遍历全表) func (q *Query) FindOne() ([]types.FieldValue, error) { var result []types.FieldValue found := false err := q.scan(func(row []types.FieldValue) bool { result = row found = true return false // 找到第一行即停止 }) if err != nil { return nil, err } if !found { return nil, page.ErrNotFound } return result, nil } // Count 流式计数 (不收集行,扫描过程中直接累加) func (q *Query) Count() (int64, error) { var count int64 err := q.scan(func(row []types.FieldValue) bool { count++ return true }) return count, err } // Sum 流式求和 (不收集行,扫描过程中直接累加) func (q *Query) Sum(col string) (float64, error) { var sum float64 err := q.scan(func(row []types.FieldValue) bool { for _, fv := range row { if string(fv.FieldName) == col { sum += toFloat64(fv.Value) break } } return true }) return sum, err } // Max 流式求最大值 (不收集行,扫描过程中维护当前最大值) func (q *Query) Max(col string) (interface{}, error) { ft := q.fieldTypeOf(col) var maxVal interface{} err := q.scan(func(row []types.FieldValue) bool { for _, fv := range row { if string(fv.FieldName) == col { if maxVal == nil || compareFieldValues(fv.Value, maxVal, ft) > 0 { maxVal = fv.Value } break } } return true }) return maxVal, err } // Min 流式求最小值 (不收集行,扫描过程中维护当前最小值) func (q *Query) Min(col string) (interface{}, error) { ft := q.fieldTypeOf(col) var minVal interface{} err := q.scan(func(row []types.FieldValue) bool { for _, fv := range row { if string(fv.FieldName) == col { if minVal == nil || compareFieldValues(fv.Value, minVal, ft) < 0 { minVal = fv.Value } break } } return true }) return minVal, err } func toFloat64(v interface{}) float64 { switch val := v.(type) { case float64: return val case float32: return float64(val) case uint64: return float64(val) case int64: return float64(val) case int: return float64(val) case int8: return float64(val) case int16: return float64(val) case int32: return float64(val) case byte: return float64(val) case bool: if val { return 1 } return 0 case types.Decimal: // 近似: decimal → float64 会失真, 但 Sum/Avg 的返回值类型就是 float64。 // 需要精确求和的金额场景请用 Decimal.Add 自行累加。 return val.Float64() case *big.Int: if val == nil { return 0 } f, _ := new(big.Float).SetInt(val).Float64() return f } return 0 }