deleter.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338
  1. package core
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core/index"
  4. "git.ycl.pub/yl32940/TsDB2/DB/core/index/btree"
  5. "git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage"
  6. "git.ycl.pub/yl32940/TsDB2/DB/core/page"
  7. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  8. "cmp"
  9. "encoding/binary"
  10. "runtime/debug"
  11. "slices"
  12. "sync"
  13. )
  14. // Deleter 删除操作对象
  15. type Deleter struct {
  16. table *Table
  17. where Condition
  18. }
  19. // Delete 返回删除操作对象
  20. func (table *Table) Delete() *Deleter {
  21. return &Deleter{table: table}
  22. }
  23. // Where 设置删除条件
  24. func (d *Deleter) Where(cond Condition) *Deleter {
  25. d.where = cond
  26. return d
  27. }
  28. // Exec 执行删除
  29. // SQLite 风格优化: 先软删除 DataPage 行 (极快), 再重建索引 (流式, 低内存)
  30. // 避免逐行级联删除 B+ 树的高昂开销
  31. func (d *Deleter) Exec() (int64, error) {
  32. defer d.table.db.lockOp()() // 批量删除全程串行; 内部 scanWithPKID/重建索引同 goroutine 重入
  33. // 找到主键索引
  34. var pkMeta *IndexMeta
  35. for i := range d.table.indexMeta.metas {
  36. m := &d.table.indexMeta.metas[i]
  37. if m.Flag&0x80 != 0 {
  38. continue
  39. }
  40. if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
  41. pkMeta = &d.table.indexMeta.metas[i]
  42. break
  43. }
  44. }
  45. if pkMeta == nil {
  46. return 0, nil
  47. }
  48. // 本流程靠"重建索引"落地, 前提是每种索引都支持从零批量构建。
  49. // 必须在 Phase 1 之前校验: 那里已经把行标记成删除, 事后再报错会留下
  50. // "行没了、索引还是旧的" 的半途状态, 比直接拒绝更糟。
  51. if err := d.checkRebuildSupported(pkMeta); err != nil {
  52. return 0, err
  53. }
  54. // 用 Scan 流式收集匹配行的 pkID (避免 Find 反序列化全行)。
  55. // 必须用 scanWithPKID 拿真实 pkID: 行数据里没有 pkID 字段 (它是行的 8 字节前缀),
  56. // 旧实现退化成"取第一个 uint64 字段值当 pkID", 主键列不是行内第一个 8 字节列时
  57. // 会删掉另一个主键指向的行, 而真正匹配的行一条都没删 —— 静默删错数据。
  58. pkIDs := make(map[uint64]bool)
  59. err := d.table.Query().Where(d.where).scanWithPKID(func(pkID uint64, _ []types.FieldValue) bool {
  60. pkIDs[pkID] = true
  61. return true
  62. })
  63. if err != nil {
  64. return 0, err
  65. }
  66. if len(pkIDs) == 0 {
  67. return 0, nil
  68. }
  69. // Phase 1: 软删除 DataPage 中的行 (只标记 DeletedFlag, 极快!)
  70. pkHandler, _ := index.Compose.GetHandler(pkMeta.IndexType)
  71. pkStore, _ := pkHandler.(index.RowStoreHandler)
  72. if pkStore == nil {
  73. return 0, nil // 主键索引不承载行数据, 软删除无从谈起 (正常不会发生)
  74. }
  75. dpIDs := pkStore.GetDataPageIDs(d.table.handle, pkMeta.EntryPage)
  76. var count int64
  77. pkKeyBE := make([]byte, 8) // 复用缓冲区
  78. for _, dpID := range dpIDs {
  79. dp, err := datapage.Load(d.table.handle, dpID)
  80. if err != nil {
  81. continue
  82. }
  83. changed := false
  84. rows := dp.GetAllRows()
  85. for _, storeRow := range rows {
  86. if len(storeRow) < 8 {
  87. continue
  88. }
  89. pkID := binary.BigEndian.Uint64(storeRow[:8])
  90. if pkIDs[pkID] {
  91. binary.BigEndian.PutUint64(pkKeyBE, pkID)
  92. if dp.DeleteByPK(pkKeyBE) {
  93. changed = true
  94. count++
  95. // 行已作废 → 释放它占用的动态字段堆块引用。
  96. // storeRow 是 GetAllRows 的独立拷贝, storeRow[8:] 即该行
  97. // 完整行数据(首字节是 schemaVer, 决定按哪个版本定位引用)。
  98. // 不释放的话堆页 RefCount 只增不减, 永不归零 → 堆页永久泄漏。
  99. d.table.heapReleaseRow(storeRow[8:])
  100. }
  101. }
  102. }
  103. if changed {
  104. dp.Save()
  105. }
  106. }
  107. // 释放 pkIDs 内存 (Phase 2 不需要)
  108. pkIDs = nil
  109. // Phase 2: 流式重建所有索引 (逐个索引从 DataPage 读取, 不缓存全部行)
  110. if err := d.rebuildIndexes(pkMeta, dpIDs); err != nil {
  111. return count, err
  112. }
  113. // Phase 3: 存活行已全部重建到全新的 DataPage, 旧 dpIDs 整个作废
  114. // (pkMeta.EntryPage 已指向新树, 二级索引只存 PKID, 都不再引用旧页)。
  115. // 必须在这里回收, 否则每执行一次批量删除就漏掉整批数据页。
  116. // 走延迟回收: 事务中途直写盘会让"旧页被清空"先于 WAL 落盘, 回滚即丢数据。
  117. // 顺序关键 —— 必须等上面两组读取循环都跑完, 否则会读到已还链的页。
  118. if d.table.handle.DeferFreePage != nil {
  119. for _, dpID := range dpIDs {
  120. d.table.handle.DeferFreePage(dpID)
  121. }
  122. }
  123. // 更新行数
  124. d.table.info.TotalRows -= uint64(count)
  125. d.table.saveTableInfo()
  126. // 失效缓存
  127. d.table.indexMeta.InvalidateRowCache()
  128. return count, nil
  129. }
  130. // rebuildIndexes 批量重建所有索引
  131. // 主键索引: 顺序创建 DataPage + 批量构建 B+ 树 (避免逐行搜索)
  132. // 二级索引: 收集条目 → 排序 → 批量构建 B+ 树
  133. func (d *Deleter) rebuildIndexes(pkMeta *IndexMeta, dpIDs []uint32) error {
  134. fields := d.table.Field
  135. // 行缓冲区按"所有 schema 版本中最宽的行" + 8B pkID 前缀分配。
  136. // 不能用 info.RowSize: AddField/DropField 之后它小于实际行长,
  137. // 沿用会让 GetAllRowsPooled 里的 copy(buf[:rowLen]) 越界 panic。
  138. rowSize := 8 + d.table.maxRowDataSize()
  139. // 内存池: 复用行缓冲区
  140. rowPool := &sync.Pool{
  141. New: func() interface{} {
  142. b := make([]byte, rowSize)
  143. return &b
  144. },
  145. }
  146. // ===== 重建主键索引: 顺序扫描 + 批量构建 =====
  147. var leafEntries [][]byte
  148. var curDP *datapage.DataPage
  149. var curDPID uint32
  150. for _, dpID := range dpIDs {
  151. dp, err := datapage.Load(d.table.handle, dpID)
  152. if err != nil {
  153. continue
  154. }
  155. rows := dp.GetAllRowsPooled(rowPool)
  156. for _, storeRow := range rows {
  157. if len(storeRow) < 8 {
  158. continue
  159. }
  160. rowData := storeRow[8:]
  161. indexKey := extractIndexKey(rowData, pkMeta, fields, d.table.versions)
  162. if indexKey == nil {
  163. continue
  164. }
  165. // 确保有可用的 DataPage
  166. if curDP == nil {
  167. curDP = datapage.New(d.table.handle)
  168. curDPID = curDP.PageID()
  169. }
  170. // 当前 DataPage 已满则创建新的
  171. if curDP.IsFull(uint32(len(storeRow))) {
  172. curDP = datapage.New(d.table.handle)
  173. curDPID = curDP.PageID()
  174. }
  175. slot := curDP.AppendRowSlot(storeRow)
  176. curDP.Save()
  177. if slot == 0 {
  178. // IsFull 预判与实际容量出现出入时的兜底: 宁可跳过, 也不要写出 slot=0 的坏叶条目
  179. continue
  180. }
  181. // 收集叶条目: [pkID(8B BE) + dpID(4B BE) + slot(2B BE)] —— 带 slot 才能 O(1) 取行
  182. pkID := binary.BigEndian.Uint64(storeRow[:8])
  183. pkKeyBE := make([]byte, 8)
  184. binary.BigEndian.PutUint64(pkKeyBE, pkID)
  185. entry := make([]byte, 14)
  186. copy(entry[:8], pkKeyBE)
  187. binary.BigEndian.PutUint32(entry[8:12], curDPID)
  188. binary.BigEndian.PutUint16(entry[12:14], uint16(slot))
  189. leafEntries = append(leafEntries, entry)
  190. }
  191. // 归还缓冲区到池
  192. for i := range rows {
  193. rowPool.Put(&rows[i])
  194. }
  195. }
  196. // 按 pkID 升序排序后再批量构建。
  197. // BuildFromSorted 假设输入有序, 不能依赖 DataPage 扫描顺序:
  198. // 更新导致的行搬迁会把新 DataPage 排到扫描序列末尾,
  199. // 从而产生 1,2,3,4,6,...,100,5 这样的失序序列, 破坏 B+ 树不变式。
  200. slices.SortFunc(leafEntries, func(a, b []byte) int {
  201. return cmp.Compare(binary.BigEndian.Uint64(a[:8]), binary.BigEndian.Uint64(b[:8]))
  202. })
  203. // 批量构建主键 B+ 树 (一次创建所有节点, 避免逐行搜索)
  204. oldPKRoot := pkMeta.EntryPage
  205. pkMeta.EntryPage = btree.BuildFromSorted(leafEntries, d.table.handle)
  206. d.table.indexMeta.saveMeta(pkMeta)
  207. // 新树已建好、元数据已落盘, 旧主键树至此彻底不可达 → 归还其全部页。
  208. // 必须晚于建树: 否则这些页会先被 NewPage 拿去分给新树, 而这里还要按旧树结构遍历。
  209. d.releaseOldTree(pkMeta, oldPKRoot)
  210. leafEntries = nil // 释放内存
  211. debug.FreeOSMemory()
  212. // ===== 重建每个二级索引: 收集 → 排序 → 批量构建 =====
  213. for i := range d.table.indexMeta.metas {
  214. m := &d.table.indexMeta.metas[i]
  215. if m.Flag&0x80 != 0 {
  216. continue
  217. }
  218. if indexStoresRows(m) {
  219. // 行存储索引: 主键索引已在上一步重建; 时序索引在 Exec 开头的
  220. // checkRebuildSupported 里已被拒绝, 走不到这里。
  221. continue
  222. }
  223. // 收集所有二级索引条目
  224. var secEntries [][]byte
  225. for _, dpID := range dpIDs {
  226. dp, err := datapage.Load(d.table.handle, dpID)
  227. if err != nil {
  228. continue
  229. }
  230. rows := dp.GetAllRowsPooled(rowPool)
  231. for _, storeRow := range rows {
  232. if len(storeRow) < 8 {
  233. continue
  234. }
  235. pkID := binary.BigEndian.Uint64(storeRow[:8])
  236. rowData := storeRow[8:]
  237. secKey := extractIndexKey(rowData, m, fields, d.table.versions)
  238. if secKey == nil {
  239. continue
  240. }
  241. // 构建二级条目: [secKey + pkID(8B BE)]
  242. secEntry := make([]byte, len(secKey)+8)
  243. copy(secEntry[:len(secKey)], secKey)
  244. binary.BigEndian.PutUint64(secEntry[len(secKey):], pkID)
  245. secEntries = append(secEntries, secEntry)
  246. }
  247. for i := range rows {
  248. rowPool.Put(&rows[i])
  249. }
  250. }
  251. // 按复合键 (secKey, pkID) 排序: 二级树的复合分隔键与叶内复合二分都要求
  252. // 全树按复合键全局有序 —— 只按 secKey 排序不够 (同 key 条目的 pkID 顺序
  253. // 无法靠"稳定排序保持物理序"保证, 数据页按主键序而非 pkID 序)。
  254. slices.SortStableFunc(secEntries, func(a, b []byte) int {
  255. ka, kb := a[:len(a)-8], b[:len(b)-8]
  256. if c := compareBytes(ka, kb); c != 0 {
  257. return c
  258. }
  259. return cmp.Compare(binary.BigEndian.Uint64(a[len(a)-8:]), binary.BigEndian.Uint64(b[len(b)-8:]))
  260. })
  261. oldSecRoot := m.EntryPage
  262. if err := buildSecondaryFromEntries(d.table.handle, m, secEntries); err != nil {
  263. return err
  264. }
  265. d.table.indexMeta.saveMeta(m)
  266. // 同上: 新结构就位后归还旧结构的全部页。
  267. d.releaseOldTree(m, oldSecRoot)
  268. secEntries = nil
  269. debug.FreeOSMemory()
  270. }
  271. return nil
  272. }
  273. // releaseOldTree 归还一棵已被换根、彻底不可达的旧索引结构 (内部节点 + 叶子节点)。
  274. //
  275. // 调用前提:
  276. // - EntryPage 已改指新结构且已 saveMeta;
  277. // - 新结构已构建完成 —— 否则归还的页会被 NewPage 立刻分给新结构, 而这里还要
  278. // 按旧结构遍历收集页 ID;
  279. // - 此后不再有任何代码读这棵旧结构。
  280. //
  281. // 走 table.deferFreePage: 事务中途直写盘会让"页已清零"先于 WAL 落盘, 回滚/崩溃
  282. // 后旧根指针还在, 却指向一张空页。与 Phase 3 回收旧 DataPage 是同一套时序约定。
  283. //
  284. // 遍历方式按页布局分流: B+ 树节点结构统一, CollectAllPageIDs 按内部条目下降即可;
  285. // 哈希 / 复合索引没有共同形状, 走处理器自己实现的 index.PageCollector。
  286. //
  287. // 不回收的后果: 每执行一次批量删除就漏掉整个旧索引结构 (实测 2 万行两级索引
  288. // 约 40 页), 文件只涨不缩。
  289. func (d *Deleter) releaseOldTree(m *IndexMeta, rootPID uint32) {
  290. if m == nil {
  291. return
  292. }
  293. releaseIndexPages(d.table, m.IndexType, rootPID)
  294. }
  295. // checkRebuildSupported 在动任何数据之前确认全部活跃索引都能被重建。
  296. func (d *Deleter) checkRebuildSupported(pkMeta *IndexMeta) error {
  297. if err := bulkRebuildSupported(pkMeta); err != nil {
  298. return err
  299. }
  300. for i := range d.table.indexMeta.metas {
  301. m := &d.table.indexMeta.metas[i]
  302. if m.Flag&0x80 != 0 {
  303. continue
  304. }
  305. if err := bulkRebuildSupported(m); err != nil {
  306. return err
  307. }
  308. }
  309. return nil
  310. }