|
|
@@ -5,6 +5,7 @@ import (
|
|
|
"git.ycl.pub/yl32940/TsDB2/DB/core/page"
|
|
|
"git.ycl.pub/yl32940/TsDB2/DB/core/types"
|
|
|
"bytes"
|
|
|
+ "container/heap"
|
|
|
"encoding/binary"
|
|
|
"fmt"
|
|
|
"math/big"
|
|
|
@@ -21,6 +22,11 @@ type Query struct {
|
|
|
offset int // OFFSET
|
|
|
groupBys []string // GROUP BY
|
|
|
havingCond Condition // HAVING
|
|
|
+
|
|
|
+ // wantMask 投影掩码 (master 下标 → 是否解码), selectCols 与 orderBys 的并集;
|
|
|
+ // 排序键从解码后的行里取, ORDER BY 列必须保留在掩码中。nil = 不过滤 (全列)。
|
|
|
+ wantMask []bool
|
|
|
+ wantBuilt bool
|
|
|
}
|
|
|
|
|
|
type SortField struct {
|
|
|
@@ -28,14 +34,21 @@ type SortField struct {
|
|
|
Descending bool
|
|
|
}
|
|
|
|
|
|
-// scanPlan 索引扫描计划
|
|
|
-type scanPlan struct {
|
|
|
- meta *IndexMeta
|
|
|
- handler index.IndexHandler
|
|
|
+// keyRange 索引扫描的单个键区间。
|
|
|
+type keyRange struct {
|
|
|
startKey []byte
|
|
|
endKey []byte
|
|
|
startInclusive bool
|
|
|
endInclusive bool
|
|
|
+}
|
|
|
+
|
|
|
+// scanPlan 索引扫描计划。
|
|
|
+// ranges 通常为单区间; WHERE ... IN (...) 落在可用分量上时展开成多区间
|
|
|
+// (每个值一个精确区间), 空切片表示"结果必为空" (如 IN 空集合)。
|
|
|
+type scanPlan struct {
|
|
|
+ meta *IndexMeta
|
|
|
+ handler index.IndexHandler
|
|
|
+ ranges []keyRange
|
|
|
rowStore bool // 该索引是否承载行数据 (主键/时序=true)
|
|
|
remainingCond Condition // 需要行级过滤的剩余条件
|
|
|
}
|
|
|
@@ -45,7 +58,11 @@ func (table *Table) Query() *Query {
|
|
|
return &Query{table: table}
|
|
|
}
|
|
|
|
|
|
-func (q *Query) Select(cols ...string) *Query { q.selectCols = cols; return q }
|
|
|
+func (q *Query) Select(cols ...string) *Query {
|
|
|
+ q.selectCols = cols
|
|
|
+ q.wantBuilt = false // 投影掩码随 selectCols 失效
|
|
|
+ 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 }
|
|
|
@@ -58,6 +75,7 @@ func (q *Query) OrderBy(field string, descending ...bool) *Query {
|
|
|
desc = descending[0]
|
|
|
}
|
|
|
q.orderBys = append(q.orderBys, SortField{Field: field, Descending: desc})
|
|
|
+ q.wantBuilt = false // 投影掩码随 orderBys 失效
|
|
|
return q
|
|
|
}
|
|
|
|
|
|
@@ -433,10 +451,6 @@ func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Con
|
|
|
compCond[k] = append(compCond[k], c)
|
|
|
}
|
|
|
|
|
|
- if cons[0].count == 0 {
|
|
|
- return nil // 首分量没有边界 → 索引给不出连续区间
|
|
|
- }
|
|
|
-
|
|
|
// 从左往右消费"等值分量"; 之后最多再消费一个范围分量, 更后面的分量帮不上忙。
|
|
|
eqLen := 0
|
|
|
for eqLen < n && cons[eqLen].isEquality() {
|
|
|
@@ -447,6 +461,35 @@ func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Con
|
|
|
rangeIdx = eqLen
|
|
|
}
|
|
|
|
|
|
+ // IN 展开: 只看"等值前缀的下一个分量", 且其上必须还没有任何边界条件
|
|
|
+ // (有范围/等值条件时让位给常规区间规划, IN 由行级过滤兜底)。
|
|
|
+ // 每个 IN 值变成该分量取值的精确区间 [前缀||v||00.., 前缀||v||FF..]。
|
|
|
+ // 多个 IN 只消费第一个适用的, 其余留在行级过滤 (正确性不受影响)。
|
|
|
+ // 注意本段在 cons[0].count 判定【之前】: 只有 IN 条件时 cons 全空,
|
|
|
+ // 但它照样能给出一组精确区间。
|
|
|
+ for _, c := range leaves {
|
|
|
+ in, ok := c.(InCondition)
|
|
|
+ if !ok {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if eqLen >= n || cons[eqLen].count != 0 {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if q.locateComponent(in.Field, m) != eqLen {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if plan := q.planInRanges(m, handler, lay, cons, eqLen, rest, compCond, in); plan != nil {
|
|
|
+ return plan
|
|
|
+ }
|
|
|
+ // 返回了 nil = 有值编不出键: 本索引放弃 IN, 让行级过滤兜住正确性,
|
|
|
+ // 继续走下面的常规区间规划 (等值前缀等条件仍可用)。
|
|
|
+ break
|
|
|
+ }
|
|
|
+
|
|
|
+ if cons[0].count == 0 {
|
|
|
+ return nil // 首分量没有边界 → 索引给不出连续区间
|
|
|
+ }
|
|
|
+
|
|
|
startKey := make([]byte, lay.total)
|
|
|
endKey := make([]byte, lay.total)
|
|
|
prefix := 0
|
|
|
@@ -458,8 +501,8 @@ func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Con
|
|
|
|
|
|
plan := &scanPlan{
|
|
|
meta: m, handler: handler,
|
|
|
- startKey: startKey, endKey: endKey,
|
|
|
- startInclusive: true, endInclusive: true,
|
|
|
+ ranges: []keyRange{{startKey: startKey, endKey: endKey,
|
|
|
+ startInclusive: true, endInclusive: true}},
|
|
|
remainingCond: q.remainCond(rest, compCond, eqLen, rangeIdx),
|
|
|
}
|
|
|
|
|
|
@@ -485,6 +528,7 @@ func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Con
|
|
|
}
|
|
|
|
|
|
// 上界: [前缀 || 分量值 || FF..]
|
|
|
+ kr := &plan.ranges[0]
|
|
|
if rc.hi == nil {
|
|
|
fillFF(endKey, prefix)
|
|
|
} else if rc.hiInc {
|
|
|
@@ -493,11 +537,66 @@ func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Con
|
|
|
} else {
|
|
|
// 严格上界: 分量 == hi 的键全部排除
|
|
|
copy(endKey[prefix:], rc.hi)
|
|
|
- plan.endInclusive = false
|
|
|
+ kr.endInclusive = false
|
|
|
}
|
|
|
return plan
|
|
|
}
|
|
|
|
|
|
+// planInRanges 把 IN 条件展开成多区间扫描计划。
|
|
|
+//
|
|
|
+// 返回 nil 表示 IN 无法索引化 (有值编不出键) —— 调用方退回常规规划/全表扫描,
|
|
|
+// IN 本身仍在 whereCond 里, 行级过滤保证正确性。
|
|
|
+// IN 空集合 → 返回 ranges 为空的计划: 任何区间都不扫, 结果必空 (IN 恒假),
|
|
|
+// 与 InCondition.Match 的空集合语义一致。
|
|
|
+func (q *Query) planInRanges(m *IndexMeta, handler index.IndexHandler, lay *indexKeyLayout,
|
|
|
+ cons []keyConstraint, eqLen int, rest []Condition, compCond [][]Condition, in InCondition) *scanPlan {
|
|
|
+ k := eqLen
|
|
|
+ size := lay.sizes[k]
|
|
|
+ fi := int(m.FieldIDs[k])
|
|
|
+ ft := q.table.Field[fi].FieldType
|
|
|
+ param := q.table.fieldParam(fi)
|
|
|
+
|
|
|
+ // 等值前缀字节 (与常规规划同一编码: cons[j].lo, isEquality 保证两端闭且相等)
|
|
|
+ prefix := make([]byte, lay.offsets[k])
|
|
|
+ for j := 0; j < eqLen; j++ {
|
|
|
+ copy(prefix[lay.offsets[j]:], cons[j].lo)
|
|
|
+ }
|
|
|
+
|
|
|
+ if len(in.Values) == 0 {
|
|
|
+ // 恒空: 不扫任何区间。remainingCond 保留全部条件 (无行可过滤, 仅为语义完整)。
|
|
|
+ return &scanPlan{meta: m, handler: handler,
|
|
|
+ ranges: nil,
|
|
|
+ remainingCond: q.remainCond(rest, compCond, 0, -1)}
|
|
|
+ }
|
|
|
+
|
|
|
+ ranges := make([]keyRange, 0, len(in.Values))
|
|
|
+ seen := make(map[string]struct{}, len(in.Values)) // 编码后同键的值只留一个区间
|
|
|
+ for _, v := range in.Values {
|
|
|
+ enc, err := encodeIndexKeySafeWithParam(v, ft, param)
|
|
|
+ if err != nil || len(enc) != size {
|
|
|
+ return nil // 任一值编不出正确宽度的键 → 整个 IN 退回行级过滤
|
|
|
+ }
|
|
|
+ // 该分量取值 v 的全部键: [前缀||v||00.., 前缀||v||FF..] (单列索引即 [v, v])
|
|
|
+ start := make([]byte, lay.total)
|
|
|
+ end := make([]byte, lay.total)
|
|
|
+ copy(start, prefix)
|
|
|
+ copy(end, prefix)
|
|
|
+ copy(start[lay.offsets[k]:], enc)
|
|
|
+ copy(end[lay.offsets[k]:], enc)
|
|
|
+ fillFF(end, lay.offsets[k]+size)
|
|
|
+ // 重复值/编码碰撞值去重: 不去重的话, raw 聚合路径 (Count/Sum 的
|
|
|
+ // scanRaw) 会对同一行重复计数 —— 行交付路径有去重, 聚合没有。
|
|
|
+ if _, dup := seen[string(start)]; dup {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ seen[string(start)] = struct{}{}
|
|
|
+ ranges = append(ranges, keyRange{startKey: start, endKey: end,
|
|
|
+ startInclusive: true, endInclusive: true})
|
|
|
+ }
|
|
|
+ return &scanPlan{meta: m, handler: handler, ranges: ranges,
|
|
|
+ remainingCond: q.remainCond(rest, compCond, eqLen, -1)}
|
|
|
+}
|
|
|
+
|
|
|
// remainCond 汇总索引区间没能表达、仍需逐行过滤的条件。
|
|
|
// 已被区间精确覆盖的分量 (等值前缀 + 那个范围分量) 上的条件可以安全丢掉。
|
|
|
func (q *Query) remainCond(rest []Condition, compCond [][]Condition, eqLen, rangeIdx int) Condition {
|
|
|
@@ -525,24 +624,52 @@ type rowWithPK struct {
|
|
|
row []types.FieldValue
|
|
|
}
|
|
|
|
|
|
-// compareRowsForOrderBy 按 OrderBy 规则比较两行, 返回 -1/0/1。
|
|
|
-// 0 表示"所有排序列都无法区分"(字段缺失), 由调用方用稳定排序保持原顺序。
|
|
|
-func (q *Query) compareRowsForOrderBy(a, b []types.FieldValue) int {
|
|
|
+// orderKeySpec 预解析的排序键: 排序列不存在于表中时在构建阶段就被丢掉,
|
|
|
+// 比较阶段不再做任何字段查找。
|
|
|
+type orderKeySpec struct {
|
|
|
+ field string
|
|
|
+ ft types.FieldType
|
|
|
+ desc bool
|
|
|
+}
|
|
|
+
|
|
|
+// buildOrderSpecs 把 OrderBy 解析成排序键规格 (字段名 + 列类型 + 方向)。
|
|
|
+func (q *Query) buildOrderSpecs() []orderKeySpec {
|
|
|
+ specs := make([]orderKeySpec, 0, len(q.orderBys))
|
|
|
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)
|
|
|
+ specs = append(specs, orderKeySpec{
|
|
|
+ field: ob.Field,
|
|
|
+ ft: q.table.Field[idx].FieldType,
|
|
|
+ desc: ob.Descending,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ return specs
|
|
|
+}
|
|
|
+
|
|
|
+// compareRowsForOrderBy 按 OrderBy 规则比较两行, 返回 -1/0/1。
|
|
|
+// 0 表示"所有排序列都无法区分"(字段缺失), 由调用方用稳定排序保持原顺序。
|
|
|
+//
|
|
|
+// 保留给外部调用方; 查询内部排序走 sortResults (decoration 每行只提取一次键)。
|
|
|
+func (q *Query) compareRowsForOrderBy(a, b []types.FieldValue) int {
|
|
|
+ return compareRowsBySpecs(a, b, q.buildOrderSpecs())
|
|
|
+}
|
|
|
+
|
|
|
+// compareRowsBySpecs 按预解析规格比较两行 (每行先经 findFieldValue 取值)。
|
|
|
+func compareRowsBySpecs(a, b []types.FieldValue, specs []orderKeySpec) int {
|
|
|
+ for _, s := range specs {
|
|
|
+ vi := findFieldValue(a, s.field)
|
|
|
+ vj := findFieldValue(b, s.field)
|
|
|
if vi == nil || vj == nil {
|
|
|
continue
|
|
|
}
|
|
|
- cmp := compareFieldValues(vi, vj, q.table.Field[idx].FieldType)
|
|
|
+ cmp := compareFieldValues(vi, vj, s.ft)
|
|
|
if cmp == 0 {
|
|
|
continue
|
|
|
}
|
|
|
- if ob.Descending {
|
|
|
+ if s.desc {
|
|
|
return -cmp
|
|
|
}
|
|
|
return cmp
|
|
|
@@ -608,7 +735,9 @@ func (q *Query) scanWithPKID(callback func(pkID uint64, row []types.FieldValue)
|
|
|
needSort = true
|
|
|
} else {
|
|
|
idxField := q.fieldNameFromMeta(plan.meta)
|
|
|
- if !(len(q.orderBys) == 1 && q.orderBys[0].Field == idxField) {
|
|
|
+ // 免排序只适用于单区间: 多区间 (IN 展开) 的输出是"各区间有序、
|
|
|
+ // 全局无序"的拼接, 必须交给下面的排序管线。
|
|
|
+ if !(len(q.orderBys) == 1 && q.orderBys[0].Field == idxField && len(plan.ranges) == 1) {
|
|
|
needSort = true
|
|
|
}
|
|
|
}
|
|
|
@@ -616,18 +745,47 @@ func (q *Query) scanWithPKID(callback func(pkID uint64, row []types.FieldValue)
|
|
|
|
|
|
// 需要排序时, 必须先收集全部再排序, 无法流式 LIMIT
|
|
|
if needSort {
|
|
|
+ specs := q.buildOrderSpecs()
|
|
|
+ // ORDER BY + LIMIT: 流式 top-k, 只保留 limit+offset 行的小顶堆,
|
|
|
+ // 内存 O(k) 且免全量排序 (旧实现物化全部结果再整体排序)。
|
|
|
+ if q.limit > 0 && len(specs) > 0 {
|
|
|
+ return q.topKScan(callback, plan, specs)
|
|
|
+ }
|
|
|
var all []rowWithPK
|
|
|
err := q.executeScan(func(pkID uint64, row []types.FieldValue) bool {
|
|
|
all = append(all, rowWithPK{pkID, row})
|
|
|
return true
|
|
|
- }, plan)
|
|
|
+ }, plan, 0)
|
|
|
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
|
|
|
- })
|
|
|
+ // 排序必须连 pkID 一起搬: 只排 []FieldValue 会让回调拿到错位的 pkID。
|
|
|
+ // decoration: 每行只按名提取一次排序键, 键必须随行进数组 —— 排序会重排
|
|
|
+ // 切片元素, 平行键数组会因此错位 (索引指向交换后的别的行)。
|
|
|
+ // 各行列集合可能因 NULL 跳过而不同, 键按名取、缺失用 nil 占位。
|
|
|
+ if len(specs) > 0 {
|
|
|
+ dec := make([]struct {
|
|
|
+ item rowWithPK
|
|
|
+ keys []interface{}
|
|
|
+ }, len(all))
|
|
|
+ for i := range all {
|
|
|
+ k := make([]interface{}, len(specs))
|
|
|
+ for j, s := range specs {
|
|
|
+ k[j] = findFieldValue(all[i].row, s.field)
|
|
|
+ }
|
|
|
+ dec[i] = struct {
|
|
|
+ item rowWithPK
|
|
|
+ keys []interface{}
|
|
|
+ }{all[i], k}
|
|
|
+ }
|
|
|
+ sort.SliceStable(dec, func(i, j int) bool {
|
|
|
+ return compareKeys(dec[i].keys, dec[j].keys, specs) < 0
|
|
|
+ })
|
|
|
+ all = all[:0]
|
|
|
+ for i := range dec {
|
|
|
+ all = append(all, dec[i].item)
|
|
|
+ }
|
|
|
+ }
|
|
|
// OFFSET
|
|
|
if q.offset > 0 {
|
|
|
if q.offset >= len(all) {
|
|
|
@@ -647,7 +805,12 @@ func (q *Query) scanWithPKID(callback func(pkID uint64, row []types.FieldValue)
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
- // 不需要排序: 直接流式传递 callback, 支持 LIMIT 提前终止
|
|
|
+ // 不需要排序: 直接流式传递 callback, 支持 LIMIT 提前终止。
|
|
|
+ // 回表预算 = LIMIT+OFFSET: 达到即收工 (OFFSET 行也要交付给下面的跳行逻辑)。
|
|
|
+ budget := 0
|
|
|
+ if q.limit > 0 {
|
|
|
+ budget = q.limit + q.offset
|
|
|
+ }
|
|
|
seen := 0
|
|
|
return q.executeScan(func(pkID uint64, row []types.FieldValue) bool {
|
|
|
if seen < q.offset {
|
|
|
@@ -663,17 +826,45 @@ func (q *Query) scanWithPKID(callback func(pkID uint64, row []types.FieldValue)
|
|
|
return false // 已满足, 提前终止扫描
|
|
|
}
|
|
|
return true
|
|
|
- }, plan)
|
|
|
+ }, plan, budget)
|
|
|
}
|
|
|
|
|
|
-// executeScan 执行索引扫描或全表扫描, 将反序列化后的行传递给回调
|
|
|
-func (q *Query) executeScan(callback func(pkID uint64, row []types.FieldValue) bool, plan *scanPlan) error {
|
|
|
+// executeScan 执行索引扫描或全表扫描, 将反序列化后的行传递给回调。
|
|
|
+// budget > 0 时最多交付 budget 行 (LIMIT+OFFSET 场景): 二级索引回表据此
|
|
|
+// 提前收工, 而不是把命中的上千行全部解出再丢掉 —— Where(hash=重).First()
|
|
|
+// 这类点查只允许解 1 行。
|
|
|
+func (q *Query) executeScan(callback func(pkID uint64, row []types.FieldValue) bool, plan *scanPlan, budget int) error {
|
|
|
if plan != nil {
|
|
|
- return q.indexScan(plan, callback)
|
|
|
+ return q.indexScan(plan, callback, budget)
|
|
|
}
|
|
|
return q.fullTableScan(callback)
|
|
|
}
|
|
|
|
|
|
+// deserialize 行解码入口: 无投影走全列解码; 有投影只解码选中列 (含 ORDER BY 列,
|
|
|
+// 排序键要从解码后的行里取)。掩码按 master 下标, 随查询构建器惰性构建一次。
|
|
|
+func (q *Query) deserialize(rowData []byte) []types.FieldValue {
|
|
|
+ if len(q.selectCols) == 0 {
|
|
|
+ return q.table.deserializeRow(rowData)
|
|
|
+ }
|
|
|
+ if !q.wantBuilt {
|
|
|
+ mask := make([]bool, len(q.table.Field))
|
|
|
+ mark := func(name string) {
|
|
|
+ if i := q.table.findFieldIndex(name); i >= 0 {
|
|
|
+ mask[i] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for _, c := range q.selectCols {
|
|
|
+ mark(c)
|
|
|
+ }
|
|
|
+ for _, ob := range q.orderBys {
|
|
|
+ mark(ob.Field)
|
|
|
+ }
|
|
|
+ q.wantMask = mask
|
|
|
+ q.wantBuilt = true
|
|
|
+ }
|
|
|
+ return q.table.deserializeRowSelect(rowData, q.wantMask)
|
|
|
+}
|
|
|
+
|
|
|
// indexScan 使用索引扫描并按"先物化、后回表"两阶段执行。
|
|
|
//
|
|
|
// 为什么不能像旧实现那样在 RangeScan 回调里直接回表 + 触发用户 callback:
|
|
|
@@ -686,10 +877,10 @@ func (q *Query) executeScan(callback func(pkID uint64, row []types.FieldValue) b
|
|
|
// 物化把"遍历"与"修改"彻底分开: 遍历阶段树保持只读, 游标始终有效; 修改发生在
|
|
|
// 遍历结束之后, 不再影响游标。二级索引只物化 pkID(8B/行), 内存开销与结果集成正比
|
|
|
// 且远小于整行物化; rowStore 索引物化的是行的快照 (必须拷贝, 因为它是页内缓冲视图)。
|
|
|
-func (q *Query) indexScan(plan *scanPlan, callback func(pkID uint64, row []types.FieldValue) bool) error {
|
|
|
- // 判断 OrderBy 是否需要反向
|
|
|
+func (q *Query) indexScan(plan *scanPlan, callback func(pkID uint64, row []types.FieldValue) bool, budget int) error {
|
|
|
+ // 判断 OrderBy 是否需要反向 (只对单区间有定义; 多区间已强制走排序管线)
|
|
|
reverse := false
|
|
|
- if len(q.orderBys) == 1 {
|
|
|
+ if len(plan.ranges) == 1 && len(q.orderBys) == 1 {
|
|
|
idxField := q.fieldNameFromMeta(plan.meta)
|
|
|
if q.orderBys[0].Field == idxField && q.orderBys[0].Descending {
|
|
|
reverse = true
|
|
|
@@ -699,19 +890,33 @@ func (q *Query) indexScan(plan *scanPlan, callback func(pkID uint64, row []types
|
|
|
if plan.rowStore {
|
|
|
// 承载行数据的索引: RangeScan 返回 storeRow = [pkID(8B) + rowData]。
|
|
|
// storeRow 指向页缓冲, 遍历结束后页可能被回收改写, 必须拷贝一份快照。
|
|
|
+ // 多区间 (IN 展开) 时不同值可能编码出重叠区间 → 按 pkID 去重。
|
|
|
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 {
|
|
|
+ var seen map[uint64]struct{}
|
|
|
+ if len(plan.ranges) > 1 {
|
|
|
+ seen = make(map[uint64]struct{}, len(plan.ranges)*4)
|
|
|
+ }
|
|
|
+ for _, kr := range plan.ranges {
|
|
|
+ if err := plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
+ kr.startKey, kr.endKey, kr.startInclusive, kr.endInclusive, reverse,
|
|
|
+ func(row []byte) bool {
|
|
|
+ if len(row) < 8 {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ if seen != nil {
|
|
|
+ pkID := binary.BigEndian.Uint64(row[:8])
|
|
|
+ if _, dup := seen[pkID]; dup {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ seen[pkID] = struct{}{}
|
|
|
+ }
|
|
|
+ snapshot := make([]byte, len(row))
|
|
|
+ copy(snapshot, row)
|
|
|
+ rows = append(rows, snapshot)
|
|
|
return true
|
|
|
- }
|
|
|
- snapshot := make([]byte, len(row))
|
|
|
- copy(snapshot, row)
|
|
|
- rows = append(rows, snapshot)
|
|
|
- return true
|
|
|
- }); err != nil {
|
|
|
- return err
|
|
|
+ }); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
}
|
|
|
for _, row := range rows {
|
|
|
rowData := row[8:]
|
|
|
@@ -720,42 +925,51 @@ func (q *Query) indexScan(plan *scanPlan, callback func(pkID uint64, row []types
|
|
|
continue
|
|
|
}
|
|
|
}
|
|
|
- if !callback(binary.BigEndian.Uint64(row[:8]), q.table.deserializeRow(rowData)) {
|
|
|
+ if !callback(binary.BigEndian.Uint64(row[:8]), q.deserialize(rowData)) {
|
|
|
return nil
|
|
|
}
|
|
|
}
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
- // 二级索引: RangeScan 返回 pkID(8B BE)。只按顺序物化 pkID, 回表放到遍历之后。
|
|
|
+ // 二级索引: RangeScan 的输出序即索引键序 (B+ 叶序 / Hash 内部已按键排序),
|
|
|
+ // 免排序 ORDER BY 路径依赖它 —— 单区间时原样保留; 只有 IN 多区间才排序去重
|
|
|
+ // (此时顺序本来就交给排序管线, 升序输入也是批量回表游标的要求)。
|
|
|
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 {
|
|
|
+ for _, kr := range plan.ranges {
|
|
|
+ if err := plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
+ kr.startKey, kr.endKey, kr.startInclusive, kr.endInclusive, reverse,
|
|
|
+ func(row []byte) bool {
|
|
|
+ if len(row) < 8 {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ pkIDs = append(pkIDs, binary.BigEndian.Uint64(row[:8]))
|
|
|
return true
|
|
|
- }
|
|
|
- pkIDs = append(pkIDs, binary.BigEndian.Uint64(row[:8]))
|
|
|
- return true
|
|
|
- }); err != nil {
|
|
|
- return err
|
|
|
+ }); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
}
|
|
|
- for _, pkID := range pkIDs {
|
|
|
- storeRow, found := q.table.indexMeta.ResolveByPK(pkID)
|
|
|
- if !found {
|
|
|
- // 行已被之前的回调删除, 跳过
|
|
|
- continue
|
|
|
+ if len(plan.ranges) > 1 && len(pkIDs) > 1 {
|
|
|
+ sort.Slice(pkIDs, func(i, j int) bool { return pkIDs[i] < pkIDs[j] })
|
|
|
+ // 多区间才可能重复 (同 pkID 命中多个重叠区间); 单区间去重徒增一次扫描
|
|
|
+ w := 1
|
|
|
+ for i := 1; i < len(pkIDs); i++ {
|
|
|
+ if pkIDs[i] != pkIDs[w-1] {
|
|
|
+ pkIDs[w] = pkIDs[i]
|
|
|
+ w++
|
|
|
+ }
|
|
|
}
|
|
|
+ pkIDs = pkIDs[:w]
|
|
|
+ }
|
|
|
+ q.table.indexMeta.BatchResolveByPK(pkIDs, budget, func(pkID uint64, storeRow []byte) bool {
|
|
|
rowData := storeRow[8:]
|
|
|
if plan.remainingCond != nil {
|
|
|
if !plan.remainingCond.Match(rowData, q.table) {
|
|
|
- continue
|
|
|
+ return true
|
|
|
}
|
|
|
}
|
|
|
- if !callback(pkID, q.table.deserializeRow(rowData)) {
|
|
|
- return nil
|
|
|
- }
|
|
|
- }
|
|
|
+ return callback(pkID, q.deserialize(rowData))
|
|
|
+ })
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
@@ -780,7 +994,7 @@ func (q *Query) fullTableScan(callback func(pkID uint64, row []types.FieldValue)
|
|
|
continue
|
|
|
}
|
|
|
}
|
|
|
- if !callback(binary.BigEndian.Uint64(storeRow[:8]), q.table.deserializeRow(rowData)) {
|
|
|
+ if !callback(binary.BigEndian.Uint64(storeRow[:8]), q.deserialize(rowData)) {
|
|
|
return nil // 回调要求提前终止
|
|
|
}
|
|
|
}
|
|
|
@@ -832,38 +1046,46 @@ func (q *Query) executeScanRaw(callback func(pkID uint64, rowData []byte) bool,
|
|
|
// 合并会让裸游标读到被改写的页。本路径的消费者全是只读聚合, 遍历中不存在任何
|
|
|
// 写操作, 游标必然有效; 于是物化 (含 rowStore 的逐行快照拷贝) 一并省掉。
|
|
|
func (q *Query) indexScanRaw(plan *scanPlan, callback func(pkID uint64, rowData []byte) bool) error {
|
|
|
- if plan.rowStore {
|
|
|
- // 承载行数据的索引: 条目即 storeRow = [pkID(8B) + rowData]
|
|
|
+ scanRange := func(kr keyRange) error {
|
|
|
+ if plan.rowStore {
|
|
|
+ // 承载行数据的索引: 条目即 storeRow = [pkID(8B) + rowData]
|
|
|
+ return plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
+ kr.startKey, kr.endKey, kr.startInclusive, kr.endInclusive, false,
|
|
|
+ func(row []byte) bool {
|
|
|
+ if len(row) < 8 {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ rowData := row[8:]
|
|
|
+ if plan.remainingCond != nil && !plan.remainingCond.Match(rowData, q.table) {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ return callback(binary.BigEndian.Uint64(row[:8]), rowData)
|
|
|
+ })
|
|
|
+ }
|
|
|
+ // 二级索引: 条目 = pkID(8B BE), 遍历中同步回表 (只读, 不触碰被遍历的索引结构)
|
|
|
return plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
- plan.startKey, plan.endKey, plan.startInclusive, plan.endInclusive, false,
|
|
|
+ kr.startKey, kr.endKey, kr.startInclusive, kr.endInclusive, false,
|
|
|
func(row []byte) bool {
|
|
|
if len(row) < 8 {
|
|
|
return true
|
|
|
}
|
|
|
- rowData := row[8:]
|
|
|
+ storeRow, found := q.table.indexMeta.ResolveByPK(binary.BigEndian.Uint64(row[:8]))
|
|
|
+ if !found {
|
|
|
+ return true // 行已被删除, 跳过
|
|
|
+ }
|
|
|
+ rowData := storeRow[8:]
|
|
|
if plan.remainingCond != nil && !plan.remainingCond.Match(rowData, q.table) {
|
|
|
return true
|
|
|
}
|
|
|
return callback(binary.BigEndian.Uint64(row[:8]), rowData)
|
|
|
})
|
|
|
}
|
|
|
- // 二级索引: 条目 = pkID(8B BE), 遍历中同步回表 (只读, 不触碰被遍历的索引结构)
|
|
|
- return plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
- plan.startKey, plan.endKey, plan.startInclusive, plan.endInclusive, false,
|
|
|
- func(row []byte) bool {
|
|
|
- if len(row) < 8 {
|
|
|
- return true
|
|
|
- }
|
|
|
- storeRow, found := q.table.indexMeta.ResolveByPK(binary.BigEndian.Uint64(row[:8]))
|
|
|
- if !found {
|
|
|
- return true // 行已被删除, 跳过
|
|
|
- }
|
|
|
- rowData := storeRow[8:]
|
|
|
- if plan.remainingCond != nil && !plan.remainingCond.Match(rowData, q.table) {
|
|
|
- return true
|
|
|
- }
|
|
|
- return callback(binary.BigEndian.Uint64(row[:8]), rowData)
|
|
|
- })
|
|
|
+ for _, kr := range plan.ranges {
|
|
|
+ if err := scanRange(kr); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return nil
|
|
|
}
|
|
|
|
|
|
// fullTableScanRaw 与 fullTableScan 同构, 回调拿裸行字节。
|
|
|
@@ -895,10 +1117,145 @@ func (q *Query) fullTableScanRaw(callback func(pkID uint64, rowData []byte) bool
|
|
|
|
|
|
// ======================== 排序 ========================
|
|
|
|
|
|
+// sortResults 内存排序: decoration 阶段每行按名提取一次排序键, 键随行进数组
|
|
|
+// (排序会重排元素, 平行键数组会错位), 比较阶段零查找。
|
|
|
+// 行集合可能因 NULL 跳列而宽窄不一, 键按名进行、缺失用 nil 占位,
|
|
|
+// 语义与旧比较器完全一致。
|
|
|
func (q *Query) sortResults(results [][]types.FieldValue) {
|
|
|
- sort.SliceStable(results, func(i, j int) bool {
|
|
|
- return q.compareRowsForOrderBy(results[i], results[j]) < 0
|
|
|
+ specs := q.buildOrderSpecs()
|
|
|
+ if len(specs) == 0 {
|
|
|
+ return
|
|
|
+ }
|
|
|
+ dec := make([]struct {
|
|
|
+ row []types.FieldValue
|
|
|
+ keys []interface{}
|
|
|
+ }, len(results))
|
|
|
+ for i := range results {
|
|
|
+ k := make([]interface{}, len(specs))
|
|
|
+ for j, s := range specs {
|
|
|
+ k[j] = findFieldValue(results[i], s.field)
|
|
|
+ }
|
|
|
+ dec[i] = struct {
|
|
|
+ row []types.FieldValue
|
|
|
+ keys []interface{}
|
|
|
+ }{results[i], k}
|
|
|
+ }
|
|
|
+ sort.SliceStable(dec, func(i, j int) bool {
|
|
|
+ return compareKeys(dec[i].keys, dec[j].keys, specs) < 0
|
|
|
})
|
|
|
+ for i := range dec {
|
|
|
+ results[i] = dec[i].row
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// compareKeys 比较两行预提取的排序键。
|
|
|
+func compareKeys(ka, kb []interface{}, specs []orderKeySpec) int {
|
|
|
+ for j, s := range specs {
|
|
|
+ vi, vj := ka[j], kb[j]
|
|
|
+ if vi == nil || vj == nil {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ cmp := compareFieldValues(vi, vj, s.ft)
|
|
|
+ if cmp == 0 {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if s.desc {
|
|
|
+ return -cmp
|
|
|
+ }
|
|
|
+ return cmp
|
|
|
+ }
|
|
|
+ return 0
|
|
|
+}
|
|
|
+
|
|
|
+// ======================== 流式 top-k (ORDER BY + LIMIT) ========================
|
|
|
+
|
|
|
+// topkItem 堆内元素: 行 + 预提取排序键 + 输入序号。
|
|
|
+// 序号做相等键的决胜 —— 弹出顺序与 sort.SliceStable 完全一致 (稳定)。
|
|
|
+type topkItem struct {
|
|
|
+ pkID uint64
|
|
|
+ row []types.FieldValue
|
|
|
+ keys []interface{}
|
|
|
+ idx int
|
|
|
+}
|
|
|
+
|
|
|
+// topkHeap 容量上限 limit+offset 的小顶堆: 超过即弹出当前最小, 堆里始终是
|
|
|
+// 已见行中的前 k 名。heap.Interface。
|
|
|
+type topkHeap struct {
|
|
|
+ specs []orderKeySpec
|
|
|
+ items []topkItem
|
|
|
+}
|
|
|
+
|
|
|
+func (h *topkHeap) Len() int { return len(h.items) }
|
|
|
+
|
|
|
+// Less 定义"输出序的逆序" (大顶堆): 堆顶是【当前集合里最后输出】的元素 ——
|
|
|
+// 容量超限时的淘汰对象。相等键时输入序号大者更靠后、先被淘汰 (稳定)。
|
|
|
+func (h *topkHeap) Less(i, j int) bool {
|
|
|
+ c := compareKeys(h.items[i].keys, h.items[j].keys, h.specs)
|
|
|
+ if c != 0 {
|
|
|
+ return c > 0
|
|
|
+ }
|
|
|
+ return h.items[i].idx > h.items[j].idx
|
|
|
+}
|
|
|
+
|
|
|
+func (h *topkHeap) Swap(i, j int) { h.items[i], h.items[j] = h.items[j], h.items[i] }
|
|
|
+
|
|
|
+func (h *topkHeap) Push(x interface{}) { h.items = append(h.items, x.(topkItem)) }
|
|
|
+
|
|
|
+func (h *topkHeap) Pop() interface{} {
|
|
|
+ old := h.items
|
|
|
+ n := len(old)
|
|
|
+ it := old[n-1]
|
|
|
+ old[n-1] = topkItem{} // 防内存泄漏
|
|
|
+ h.items = old[:n-1]
|
|
|
+ return it
|
|
|
+}
|
|
|
+
|
|
|
+// topKScan ORDER BY + LIMIT 的流式执行: 扫描全程只保留 limit+offset 行的小顶堆,
|
|
|
+// 内存 O(k)、免全量排序 (旧实现先物化全部结果再整体排序, 百万行翻页也要全排)。
|
|
|
+// 稳定性: 相等键按输入序号决胜, 输出与 sort.SliceStable 后切片逐字节一致。
|
|
|
+func (q *Query) topKScan(callback func(pkID uint64, row []types.FieldValue) bool,
|
|
|
+ plan *scanPlan, specs []orderKeySpec) error {
|
|
|
+ cap := q.limit + q.offset
|
|
|
+ if cap < q.limit {
|
|
|
+ cap = q.limit // 防 offset 溢出 (理论值, int 实际不会)
|
|
|
+ }
|
|
|
+ h := &topkHeap{specs: specs}
|
|
|
+ idx := 0
|
|
|
+ err := q.executeScan(func(pkID uint64, row []types.FieldValue) bool {
|
|
|
+ keys := make([]interface{}, len(specs))
|
|
|
+ for j, s := range specs {
|
|
|
+ keys[j] = findFieldValue(row, s.field)
|
|
|
+ }
|
|
|
+ heap.Push(h, topkItem{pkID: pkID, row: row, keys: keys, idx: idx})
|
|
|
+ idx++
|
|
|
+ if h.Len() > cap {
|
|
|
+ heap.Pop(h)
|
|
|
+ }
|
|
|
+ return true
|
|
|
+ }, plan, 0)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ // 全部弹出: 大顶堆弹出的是"输出序最后"的元素 → items 是输出序的逆序。
|
|
|
+ items := make([]topkItem, 0, h.Len())
|
|
|
+ for h.Len() > 0 {
|
|
|
+ items = append(items, heap.Pop(h).(topkItem))
|
|
|
+ }
|
|
|
+ if q.offset > 0 {
|
|
|
+ if q.offset >= len(items) {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ items = items[:len(items)-q.offset] // 弹出序的尾部 = 输出序的头部 → 丢弃
|
|
|
+ }
|
|
|
+ if q.limit > 0 && len(items) > q.limit {
|
|
|
+ items = items[len(items)-q.limit:] // 保留弹出序尾部 = 输出序的前 limit 名
|
|
|
+ }
|
|
|
+ for i := len(items) - 1; i >= 0; i-- {
|
|
|
+ if !callback(items[i].pkID, items[i].row) {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return nil
|
|
|
}
|
|
|
|
|
|
func findFieldValue(row []types.FieldValue, field string) interface{} {
|
|
|
@@ -1085,15 +1442,20 @@ func (q *Query) Count() (int64, error) {
|
|
|
if !paged {
|
|
|
if plan := q.buildScanPlan(); plan != nil && plan.remainingCond == nil {
|
|
|
var count int64
|
|
|
- err := plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
- plan.startKey, plan.endKey, plan.startInclusive, plan.endInclusive, false,
|
|
|
- func(row []byte) bool {
|
|
|
- if len(row) >= 8 {
|
|
|
- count++
|
|
|
- }
|
|
|
- return true
|
|
|
- })
|
|
|
- return count, err
|
|
|
+ for _, kr := range plan.ranges {
|
|
|
+ err := plan.handler.RangeScan(q.table.handle, plan.meta.EntryPage,
|
|
|
+ kr.startKey, kr.endKey, kr.startInclusive, kr.endInclusive, false,
|
|
|
+ func(row []byte) bool {
|
|
|
+ if len(row) >= 8 {
|
|
|
+ count++
|
|
|
+ }
|
|
|
+ return true
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return 0, err
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return count, nil
|
|
|
}
|
|
|
var count int64
|
|
|
err := q.scanRaw(func(_ uint64, _ []byte) bool {
|