query.go 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333
  1. package DB
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core"
  4. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  5. "errors"
  6. "fmt"
  7. "reflect"
  8. )
  9. // Query 泛型查询构建器
  10. type Query[T any] struct {
  11. table *Table[T]
  12. wheres []whereClause
  13. orders []orderClause
  14. selectCols []string
  15. limit int
  16. offset int
  17. // buildErr 收集构建期的校验错误 (Where / OrderBy / Select 里出现未知字段时记录)。
  18. // 构建器方法按链式约定不返回 error, 所以错误先攒在这里, 执行入口
  19. // (Find / First / Count / Scan) 一开闸就检查并原样返回 —— 绝不带着
  20. // "未知字段"的条件去执行: Where("不存在的字段","!=",x) 会被翻成 Not(Eq(...)),
  21. // 未知字段的 Eq 恒 false、取反恒 true, 等于静默返回全表; OrderBy 错字段则
  22. // 静默不排序 —— 这两类都是"危险地静默错数据"。
  23. buildErr error
  24. }
  25. type whereClause struct {
  26. field string
  27. op string
  28. value interface{}
  29. }
  30. type orderClause struct {
  31. field string
  32. desc bool
  33. }
  34. // recordBuildErr 聚合构建期错误。多个错误用 errors.Join 串成一条链,
  35. // 执行入口返回的 err 里能逐条看到 (errors.Is / Unwrap 均可达)。
  36. func (q *Query[T]) recordBuildErr(err error) {
  37. q.buildErr = errors.Join(q.buildErr, err)
  38. }
  39. // fieldExists 判断字段名能否对上本表的列: 先查 DB schema 活跃列, 回退 struct model
  40. // (按列名或 Go 字段名)。与 findFieldType 同一套解析口径 —— 它认的名字这里必须认,
  41. // 否则校验会比实际执行更严, 把原本能跑的查询误杀。
  42. func (q *Query[T]) fieldExists(name string) bool {
  43. if _, ok := q.table.coreTable.GetActiveFieldType(name); ok {
  44. return true
  45. }
  46. for _, fd := range q.table.model.AllFields {
  47. if fd.ColName == name || fd.Name == name {
  48. return true
  49. }
  50. }
  51. return false
  52. }
  53. func (q *Query[T]) addWhere(field, op string, value interface{}) *Query[T] {
  54. if !q.fieldExists(field) {
  55. q.recordBuildErr(fmt.Errorf("tsdb: Where: 未知字段 %q (不是本表的列名或 Go 字段名)", field))
  56. }
  57. q.wheres = append(q.wheres, whereClause{field: field, op: op, value: value})
  58. return q
  59. }
  60. // Where 添加条件
  61. func (q *Query[T]) Where(field string, op string, value interface{}) *Query[T] {
  62. return q.addWhere(field, op, value)
  63. }
  64. // OrderBy 排序
  65. func (q *Query[T]) OrderBy(field string, desc ...bool) *Query[T] {
  66. if !q.fieldExists(field) {
  67. q.recordBuildErr(fmt.Errorf("tsdb: OrderBy: 未知字段 %q (不是本表的列名或 Go 字段名)", field))
  68. }
  69. d := false
  70. if len(desc) > 0 && desc[0] {
  71. d = true
  72. }
  73. q.orders = append(q.orders, orderClause{field: field, desc: d})
  74. return q
  75. }
  76. // Limit 限制行数
  77. func (q *Query[T]) Limit(n int) *Query[T] {
  78. q.limit = n
  79. return q
  80. }
  81. // Offset 跳过行数
  82. func (q *Query[T]) Offset(n int) *Query[T] {
  83. q.offset = n
  84. return q
  85. }
  86. // Select 只取指定列 (Go 字段名)。投影下推到引擎: 未选中的列不做反序列化与
  87. // 堆页解引用; 读出的对象里未选中的字段保持零值。
  88. func (q *Query[T]) Select(cols ...string) *Query[T] {
  89. for _, c := range cols {
  90. if !q.fieldExists(c) {
  91. q.recordBuildErr(fmt.Errorf("tsdb: Select: 未知字段 %q (不是本表的列名或 Go 字段名)", c))
  92. }
  93. }
  94. q.selectCols = cols
  95. return q
  96. }
  97. // Find 执行查询返回所有结果
  98. //
  99. // 用 ScanWithPKID 而不是 Find: 行号不在字段值里, 但它决定了返回的对象能不能被精确
  100. // 地改 / 删 —— 同一个时间戳下有多行时, 只有行号指得清"返回的是哪一行"。
  101. func (q *Query[T]) Find() ([]*T, error) {
  102. if q.buildErr != nil {
  103. return nil, q.buildErr
  104. }
  105. cq := q.buildCoreQuery()
  106. result := make([]*T, 0)
  107. var convErr error
  108. err := cq.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
  109. obj, err := q.table.convertRow(pkID, row)
  110. if err != nil {
  111. // 反序列化失败: 停止扫描并让 Find 返回错误, 不能静默跳过该行
  112. if convErr == nil {
  113. convErr = err
  114. }
  115. return false
  116. }
  117. result = append(result, obj)
  118. return true
  119. })
  120. if err != nil {
  121. return nil, err
  122. }
  123. if convErr != nil {
  124. return nil, convErr
  125. }
  126. return result, nil
  127. }
  128. // First 返回第一条结果
  129. func (q *Query[T]) First() (*T, error) {
  130. q.limit = 1
  131. results, err := q.Find()
  132. if err != nil {
  133. return nil, err
  134. }
  135. if len(results) == 0 {
  136. return nil, fmt.Errorf("tsdb: record not found")
  137. }
  138. return results[0], nil
  139. }
  140. // Count 返回匹配行数
  141. func (q *Query[T]) Count() (int64, error) {
  142. if q.buildErr != nil {
  143. return 0, q.buildErr
  144. }
  145. cq := q.buildCoreQuery()
  146. return cq.Count()
  147. }
  148. // Scan 流式遍历
  149. func (q *Query[T]) Scan(callback func(*T) bool) error {
  150. if q.buildErr != nil {
  151. return q.buildErr
  152. }
  153. cq := q.buildCoreQuery()
  154. var convErr error
  155. err := cq.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
  156. obj, err := q.table.convertRow(pkID, row)
  157. if err != nil {
  158. // 反序列化失败: 停止扫描并让 Scan 返回错误, 不能静默跳过该行
  159. if convErr == nil {
  160. convErr = err
  161. }
  162. return false
  163. }
  164. return callback(obj)
  165. })
  166. if err != nil {
  167. return err
  168. }
  169. return convErr
  170. }
  171. // buildCoreQuery 构建 core 层的 Query
  172. func (q *Query[T]) buildCoreQuery() *core.Query {
  173. cq := q.table.coreTable.Query()
  174. // 构建 WHERE 条件
  175. if len(q.wheres) > 0 {
  176. cond := q.buildCondition(q.wheres[0])
  177. for i := 1; i < len(q.wheres); i++ {
  178. cond = core.And(cond, q.buildCondition(q.wheres[i]))
  179. }
  180. cq = cq.Where(cond)
  181. }
  182. // ORDER BY
  183. for _, o := range q.orders {
  184. cq = cq.OrderBy(o.field, o.desc)
  185. }
  186. // SELECT 投影下推
  187. if len(q.selectCols) > 0 {
  188. cq = cq.Select(q.selectCols...)
  189. }
  190. // LIMIT / OFFSET
  191. if q.limit > 0 {
  192. cq = cq.Limit(q.limit)
  193. }
  194. if q.offset > 0 {
  195. cq = cq.Offset(q.offset)
  196. }
  197. return cq
  198. }
  199. // buildCondition 将 whereClause 转换为 core.Condition
  200. func (q *Query[T]) buildCondition(w whereClause) core.Condition {
  201. // ft 只用来把条件值归一化成列期望的 Go 类型 (normalizeCondValue);
  202. // 条件构造函数的第三个参数已废弃 —— 比较一律按列的实际存储类型进行
  203. // (见 core.Table.compareTypeFor), 故统一传 types.FieldType(0)。
  204. ft := q.findFieldType(w.field)
  205. val := normalizeCondValue(w.value, ft)
  206. switch w.op {
  207. case "=", "==", "eq":
  208. return core.Eq(w.field, val, types.FieldType(0))
  209. case "!=", "<>":
  210. return core.Not(core.Eq(w.field, val, types.FieldType(0)))
  211. case ">", "gt":
  212. return core.Gt(w.field, val, types.FieldType(0))
  213. case ">=", "ge", "gte":
  214. return core.Ge(w.field, val, types.FieldType(0))
  215. case "<", "lt":
  216. return core.Lt(w.field, val, types.FieldType(0))
  217. case "<=", "le", "lte":
  218. return core.Le(w.field, val, types.FieldType(0))
  219. case "like":
  220. if s, ok := w.value.(string); ok {
  221. return core.Like(w.field, s)
  222. }
  223. return core.Eq(w.field, val, types.FieldType(0))
  224. case "in", "In", "IN":
  225. // 值须是切片/数组; 每个分量先经 normalizeCondValue 归一成列期望的
  226. // Go 类型再交给 core.In (分量类型不一会按列类型逐个编码比较)。
  227. var vals []interface{}
  228. rv := reflect.ValueOf(w.value)
  229. switch rv.Kind() {
  230. case reflect.Slice, reflect.Array:
  231. vals = make([]interface{}, rv.Len())
  232. for i := 0; i < rv.Len(); i++ {
  233. vals[i] = normalizeCondValue(rv.Index(i).Interface(), ft)
  234. }
  235. default:
  236. // 单值容忍: Where("Age","in",25) ≡ Where("Age","=",25)
  237. return core.Eq(w.field, val, types.FieldType(0))
  238. }
  239. if len(vals) == 0 {
  240. // 空集合恒假: InCondition.Match 对每行返回 false (无索引特化路径,
  241. // 不会被 nil 条件的"跳过过滤"语义坑成查全表)。
  242. return core.In(w.field, nil, types.FieldType(0))
  243. }
  244. return core.In(w.field, vals, types.FieldType(0))
  245. default:
  246. return core.Eq(w.field, val, types.FieldType(0))
  247. }
  248. }
  249. // normalizeCondValue 将条件值转换为 core 期望的类型
  250. func normalizeCondValue(v interface{}, ft types.FieldType) interface{} {
  251. switch ft {
  252. case types.FieldType_Uint64:
  253. return toUint64(v)
  254. case types.FieldType_Uint32:
  255. return toInt32(v)
  256. case types.FieldType_Uint16:
  257. return int16(toUint64(v))
  258. case types.FieldType_Uint8:
  259. return uint8(toUint64(v))
  260. case types.FieldType_Int8:
  261. return toInt8(v)
  262. case types.FieldType_Int16:
  263. return toInt16(v)
  264. case types.FieldType_Int32:
  265. return toInt32(v)
  266. case types.FieldType_Int64:
  267. return toInt64(v)
  268. case types.FieldType_Float64:
  269. // 不能走 toUint64: 那会把条件值的小数先砍掉, 变成在浮点列上按整数匹配
  270. return toFloatVal(v)
  271. // ===== 扩展类型 =====
  272. //
  273. // 全部原样下传, 交给 core 的 types.EncodeValue / DecimalFromValue 解析与校验。
  274. // 尤其是 Decimal / UUID: 条件值常常是字符串 ("1.50" / "550e8400-..."), 一旦落进
  275. // 下面的 default 分支会被当定长串补零成定长字节, 比较和索引键就全错了。
  276. case types.FieldType_Bool,
  277. types.FieldType_Float32,
  278. types.FieldType_Decimal,
  279. types.FieldType_UUID,
  280. types.FieldType_Year,
  281. types.FieldType_Int128,
  282. types.FieldType_Blob:
  283. return v
  284. default:
  285. // 定长字符串类型: 原样下传, 由 core 按【列的实际宽度】编码
  286. // (types.EncodeValue 接受 string, 并顺带成为宽度闸门: 超宽报错而不是截断)。
  287. //
  288. // 不能在这里按 types.GetSize(ft) 自己补零: 那只是类型默认宽度, 对用宽度码声明的
  289. // 列 (maxlen:1000 → 1024 字节档) 会编出 256 字节——与行内字节不同源, 比较与索引键
  290. // 全错。而且这里只有 ft 没有列级参数, 本就拼不出正确的宽度。
  291. return v
  292. }
  293. }
  294. // findFieldType 查找字段的 FieldType
  295. // 优先从 DB schema 查找,回退到 struct model
  296. func (q *Query[T]) findFieldType(name string) types.FieldType {
  297. // 优先从 DB schema 查找
  298. if ft, ok := q.table.coreTable.GetActiveFieldType(name); ok {
  299. return ft
  300. }
  301. // 回退到 struct model
  302. for _, fd := range q.table.model.AllFields {
  303. if fd.ColName == name || fd.Name == name {
  304. return fd.FieldType
  305. }
  306. }
  307. return types.FieldType_Uint64 // 默认
  308. }