updater.go 2.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  1. package core
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  4. "git.ycl.pub/yl32940/TsDB2/DB/core/util"
  5. "encoding/binary"
  6. )
  7. // Updater 更新操作对象
  8. type Updater struct {
  9. table *Table
  10. updates map[string]interface{}
  11. where Condition
  12. }
  13. // Update 返回更新操作对象
  14. func (table *Table) Update() *Updater {
  15. return &Updater{
  16. table: table,
  17. updates: make(map[string]interface{}),
  18. }
  19. }
  20. // Set 设置更新字段
  21. func (u *Updater) Set(field string, value interface{}) *Updater {
  22. u.updates[field] = value
  23. return u
  24. }
  25. // Where 设置更新条件
  26. func (u *Updater) Where(cond Condition) *Updater {
  27. u.where = cond
  28. return u
  29. }
  30. // Exec 执行更新
  31. func (u *Updater) Exec() (int64, error) {
  32. defer u.table.db.lockOp()() // 批量更新全程串行; 内部 scanWithPKID/UpdateRow 同 goroutine 重入
  33. // 找承载行数据的一级索引名 (实现 RowStoreHandler 者, 本层不认识索引类型)
  34. pkIdxName := ""
  35. if pkMeta, _, ok := u.table.indexMeta.rowStoreMeta(); ok {
  36. pkIdxName = util.BytesToString(pkMeta.IndexName[:])
  37. }
  38. if pkIdxName == "" {
  39. return 0, nil
  40. }
  41. // 先收集匹配行的真实 pkID, 再逐行改。
  42. // 两处都不能沿用旧实现:
  43. // - 用 Find() 收集会把每一行都反序列化成 []FieldValue, 万行级的更新白占内存,
  44. // 这里只物化 pkID;
  45. // - pkID 必须取行前缀里的真实值。行数据里没有 pkID 字段, 旧实现只能"取第一个
  46. // uint64 字段值当 pkID", 主键列不是行内第一个 8 字节列时会改到另一行上。
  47. var pkIDs []uint64
  48. if err := u.table.Query().Where(u.where).scanWithPKID(func(pkID uint64, _ []types.FieldValue) bool {
  49. pkIDs = append(pkIDs, pkID)
  50. return true
  51. }); err != nil {
  52. return 0, err
  53. }
  54. var count int64
  55. for _, pkID := range pkIDs {
  56. key := make([]byte, 8)
  57. binary.BigEndian.PutUint64(key, pkID)
  58. // 只把本次要改的字段交给 UpdateRow, 其余列由 serializeRow 沿用旧值
  59. var values []types.FieldValue
  60. for field, val := range u.updates {
  61. values = append(values, types.FieldValue{
  62. FieldName: []byte(field),
  63. Value: val,
  64. })
  65. }
  66. // 过去这里写成 `if err == nil { count++ }`, 把失败行悄悄跳过:
  67. // 调用方会拿到成功的行数, 却不知道有行没更新。必须向上抛出。
  68. if err := u.table.UpdateRow(pkIdxName, key, values); err != nil {
  69. return count, err
  70. }
  71. count++
  72. }
  73. return count, nil
  74. }