| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516 |
- package core
- import (
- "TsDB2/DB/core/index"
- "TsDB2/DB/core/index/btree"
- "TsDB2/DB/core/index/datapage"
- "TsDB2/DB/core/page"
- "TsDB2/DB/core/types"
- "bytes"
- "encoding/binary"
- "fmt"
- "runtime/debug"
- "sort"
- "sync"
- )
- // Deleter 删除操作对象
- type Deleter struct {
- table *Table
- where Condition
- }
- // Delete 返回删除操作对象
- func (table *Table) Delete() *Deleter {
- return &Deleter{table: table}
- }
- // Where 设置删除条件
- func (d *Deleter) Where(cond Condition) *Deleter {
- d.where = cond
- return d
- }
- // Exec 执行删除
- // SQLite 风格优化: 先软删除 DataPage 行 (极快), 再重建索引 (流式, 低内存)
- // 避免逐行级联删除 B+ 树的高昂开销
- func (d *Deleter) Exec() (int64, error) {
- // 找到主键索引
- var pkMeta *IndexMeta
- for i := range d.table.indexMeta.metas {
- m := &d.table.indexMeta.metas[i]
- if m.Flag&0x80 != 0 {
- continue
- }
- if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
- pkMeta = &d.table.indexMeta.metas[i]
- break
- }
- }
- if pkMeta == nil {
- return 0, nil
- }
- // 本流程靠"重建索引"落地, 前提是每种索引都支持从零批量构建。
- // 必须在 Phase 1 之前校验: 那里已经把行标记成删除, 事后再报错会留下
- // "行没了、索引还是旧的" 的半途状态, 比直接拒绝更糟。
- if err := d.checkRebuildSupported(pkMeta); err != nil {
- return 0, err
- }
- // 用 Scan 流式收集匹配行的 pkID (避免 Find 反序列化全行)。
- // 必须用 scanWithPKID 拿真实 pkID: 行数据里没有 pkID 字段 (它是行的 8 字节前缀),
- // 旧实现退化成"取第一个 uint64 字段值当 pkID", 主键列不是行内第一个 8 字节列时
- // 会删掉另一个主键指向的行, 而真正匹配的行一条都没删 —— 静默删错数据。
- pkIDs := make(map[uint64]bool)
- err := d.table.Query().Where(d.where).scanWithPKID(func(pkID uint64, _ []types.FieldValue) bool {
- pkIDs[pkID] = true
- return true
- })
- if err != nil {
- return 0, err
- }
- if len(pkIDs) == 0 {
- return 0, nil
- }
- // Phase 1: 软删除 DataPage 中的行 (只标记 DeletedFlag, 极快!)
- pkHandler, _ := index.Compose.GetHandler(pkMeta.IndexType)
- dpIDs := pkHandler.GetDataPageIDs(d.table.handle, pkMeta.EntryPage)
- var count int64
- pkKeyBE := make([]byte, 8) // 复用缓冲区
- for _, dpID := range dpIDs {
- dp, err := datapage.Load(d.table.handle, dpID)
- if err != nil {
- continue
- }
- changed := false
- rows := dp.GetAllRows()
- for _, storeRow := range rows {
- if len(storeRow) < 8 {
- continue
- }
- pkID := binary.BigEndian.Uint64(storeRow[:8])
- if pkIDs[pkID] {
- binary.BigEndian.PutUint64(pkKeyBE, pkID)
- if dp.DeleteByPK(pkKeyBE) {
- changed = true
- count++
- // 行已作废 → 释放它占用的动态字段堆块引用。
- // storeRow 是 GetAllRows 的独立拷贝, storeRow[8:] 即该行
- // 完整行数据(首字节是 schemaVer, 决定按哪个版本定位引用)。
- // 不释放的话堆页 RefCount 只增不减, 永不归零 → 堆页永久泄漏。
- d.table.heapReleaseRow(storeRow[8:])
- }
- }
- }
- if changed {
- dp.Save()
- }
- }
- // 释放 pkIDs 内存 (Phase 2 不需要)
- pkIDs = nil
- // Phase 2: 流式重建所有索引 (逐个索引从 DataPage 读取, 不缓存全部行)
- if err := d.rebuildIndexes(pkMeta, dpIDs); err != nil {
- return count, err
- }
- // Phase 3: 存活行已全部重建到全新的 DataPage, 旧 dpIDs 整个作废
- // (pkMeta.EntryPage 已指向新树, 二级索引只存 PKID, 都不再引用旧页)。
- // 必须在这里回收, 否则每执行一次批量删除就漏掉整批数据页。
- // 走延迟回收: 事务中途直写盘会让"旧页被清空"先于 WAL 落盘, 回滚即丢数据。
- // 顺序关键 —— 必须等上面两组读取循环都跑完, 否则会读到已还链的页。
- if d.table.handle.DeferFreePage != nil {
- for _, dpID := range dpIDs {
- d.table.handle.DeferFreePage(dpID)
- }
- }
- // 更新行数
- d.table.info.TotalRows -= uint64(count)
- d.table.saveTableInfo()
- // 失效缓存
- d.table.indexMeta.InvalidateRowCache()
- return count, nil
- }
- // rebuildIndexes 批量重建所有索引
- // 主键索引: 顺序创建 DataPage + 批量构建 B+ 树 (避免逐行搜索)
- // 二级索引: 收集条目 → 排序 → 批量构建 B+ 树
- func (d *Deleter) rebuildIndexes(pkMeta *IndexMeta, dpIDs []uint32) error {
- fields := d.table.Field
- // 行缓冲区按"所有 schema 版本中最宽的行" + 8B pkID 前缀分配。
- // 不能用 info.RowSize: AddField/DropField 之后它小于实际行长,
- // 沿用会让 GetAllRowsPooled 里的 copy(buf[:rowLen]) 越界 panic。
- rowSize := 8 + d.table.maxRowDataSize()
- // 内存池: 复用行缓冲区
- rowPool := &sync.Pool{
- New: func() interface{} {
- b := make([]byte, rowSize)
- return &b
- },
- }
- // ===== 重建主键索引: 顺序扫描 + 批量构建 =====
- var leafEntries [][]byte
- var curDP *datapage.DataPage
- var curDPID uint32
- for _, dpID := range dpIDs {
- dp, err := datapage.Load(d.table.handle, dpID)
- if err != nil {
- continue
- }
- rows := dp.GetAllRowsPooled(rowPool)
- for _, storeRow := range rows {
- if len(storeRow) < 8 {
- continue
- }
- rowData := storeRow[8:]
- indexKey := extractIndexKey(rowData, pkMeta, fields, d.table.activeVersion, d.table.versions)
- if indexKey == nil {
- continue
- }
- // 确保有可用的 DataPage
- if curDP == nil {
- curDP = datapage.New(d.table.handle)
- curDPID = curDP.PageID()
- }
- // 当前 DataPage 已满则创建新的
- if curDP.IsFull(uint32(len(storeRow))) {
- curDP = datapage.New(d.table.handle)
- curDPID = curDP.PageID()
- }
- curDP.AppendRow(storeRow)
- curDP.Save()
- // 收集叶条目: [pkID(8B BE) + dpID(4B BE)]
- pkID := binary.BigEndian.Uint64(storeRow[:8])
- pkKeyBE := make([]byte, 8)
- binary.BigEndian.PutUint64(pkKeyBE, pkID)
- entry := make([]byte, 12)
- copy(entry[:8], pkKeyBE)
- binary.BigEndian.PutUint32(entry[8:], curDPID)
- leafEntries = append(leafEntries, entry)
- }
- // 归还缓冲区到池
- for i := range rows {
- rowPool.Put(&rows[i])
- }
- }
- // 按 pkID 升序排序后再批量构建。
- // BuildFromSorted 假设输入有序, 不能依赖 DataPage 扫描顺序:
- // 更新导致的行搬迁会把新 DataPage 排到扫描序列末尾,
- // 从而产生 1,2,3,4,6,...,100,5 这样的失序序列, 破坏 B+ 树不变式。
- sort.Slice(leafEntries, func(i, j int) bool {
- return binary.BigEndian.Uint64(leafEntries[i][:8]) <
- binary.BigEndian.Uint64(leafEntries[j][:8])
- })
- // 批量构建主键 B+ 树 (一次创建所有节点, 避免逐行搜索)
- oldPKRoot := pkMeta.EntryPage
- pkMeta.EntryPage = btree.BuildFromSorted(leafEntries, d.table.handle)
- d.table.indexMeta.saveMeta(pkMeta)
- // 新树已建好、元数据已落盘, 旧主键树至此彻底不可达 → 归还其全部页。
- // 必须晚于建树: 否则这些页会先被 NewPage 拿去分给新树, 而这里还要按旧树结构遍历。
- d.releaseOldTree(pkMeta, oldPKRoot)
- leafEntries = nil // 释放内存
- debug.FreeOSMemory()
- // ===== 重建每个二级索引: 收集 → 排序 → 批量构建 =====
- for i := range d.table.indexMeta.metas {
- m := &d.table.indexMeta.metas[i]
- if m.Flag&0x80 != 0 {
- continue
- }
- if indexStoresRows(m) {
- // 行存储索引: 主键索引已在上一步重建; 时序索引在 Exec 开头的
- // checkRebuildSupported 里已被拒绝, 走不到这里。
- continue
- }
- // 收集所有二级索引条目
- var secEntries [][]byte
- for _, dpID := range dpIDs {
- dp, err := datapage.Load(d.table.handle, dpID)
- if err != nil {
- continue
- }
- rows := dp.GetAllRowsPooled(rowPool)
- for _, storeRow := range rows {
- if len(storeRow) < 8 {
- continue
- }
- pkID := binary.BigEndian.Uint64(storeRow[:8])
- rowData := storeRow[8:]
- secKey := extractIndexKey(rowData, m, fields, d.table.activeVersion, d.table.versions)
- if secKey == nil {
- continue
- }
- // 构建二级条目: [secKey + pkID(8B BE)]
- secEntry := make([]byte, len(secKey)+8)
- copy(secEntry[:len(secKey)], secKey)
- binary.BigEndian.PutUint64(secEntry[len(secKey):], pkID)
- secEntries = append(secEntries, secEntry)
- }
- for i := range rows {
- rowPool.Put(&rows[i])
- }
- }
- // 按 secKey 排序 (稳定排序保持同 key 的 pkID 顺序)
- sort.SliceStable(secEntries, func(i, j int) bool {
- ki := secEntries[i][:len(secEntries[i])-8]
- kj := secEntries[j][:len(secEntries[j])-8]
- return compareBytes(ki, kj) < 0
- })
- oldSecRoot := m.EntryPage
- if err := buildSecondaryFromEntries(d.table.handle, m, secEntries); err != nil {
- return err
- }
- d.table.indexMeta.saveMeta(m)
- // 同上: 新结构就位后归还旧结构的全部页。
- d.releaseOldTree(m, oldSecRoot)
- secEntries = nil
- debug.FreeOSMemory()
- }
- return nil
- }
- // buildSecondaryFromEntries 用"已按 secKey 排好序"的条目 [secKey + pkID(8B BE)] 构建一个二级索引。
- //
- // 页布局决定做法, 不能一律批量建树:
- // - B+ 树二级索引: 批量建树;
- // - 复合 / 哈希二级索引: 页不是 B+ 节点, 必须走各自的 handler.Insert
- // (详见 rebuildSecondaryViaHandler 注释)。
- //
- // 调用方负责 m.EntryPage 的旧根归还 (整表重建走 releaseOldTree); 新建索引回填时
- // 旧根本就是 0, 无需归还。
- func buildSecondaryFromEntries(handle page.HandleDb, m *IndexMeta, entries [][]byte) error {
- if isBTreePageIndex(m) {
- m.EntryPage = btree.SBTBuildFromSorted(entries, handle)
- return nil
- }
- return rebuildSecondaryViaHandler(handle, m, entries)
- }
- // releaseOldTree 归还一棵已被换根、彻底不可达的旧索引结构 (内部节点 + 叶子节点)。
- //
- // 调用前提:
- // - EntryPage 已改指新结构且已 saveMeta;
- // - 新结构已构建完成 —— 否则归还的页会被 NewPage 立刻分给新结构, 而这里还要
- // 按旧结构遍历收集页 ID;
- // - 此后不再有任何代码读这棵旧结构。
- //
- // 走 table.deferFreePage: 事务中途直写盘会让"页已清零"先于 WAL 落盘, 回滚/崩溃
- // 后旧根指针还在, 却指向一张空页。与 Phase 3 回收旧 DataPage 是同一套时序约定。
- //
- // 遍历方式按页布局分流: B+ 树节点结构统一, CollectAllPageIDs 按内部条目下降即可;
- // 哈希 / 复合索引没有共同形状, 走处理器自己实现的 index.PageCollector。
- //
- // 不回收的后果: 每执行一次批量删除就漏掉整个旧索引结构 (实测 2 万行两级索引
- // 约 40 页), 文件只涨不缩。
- func (d *Deleter) releaseOldTree(m *IndexMeta, rootPID uint32) {
- if m == nil {
- return
- }
- releaseIndexPages(d.table, m.IndexType, rootPID)
- }
- // releaseIndexPages 归还一棵已彻底不可达的索引结构 (内部节点 + 叶子 / 桶链) 的全部页。
- //
- // 这是 releaseOldTree 与 DROP INDEX 的共用实现: 两者都换掉/摘掉了根, 旧结构从此没人读,
- // 必须把它的页还给空闲链, 否则文件只涨不缩。遍历方式按索引类型的页布局分流:
- // - B+ 树 (主键 / 二级 / 旧 Str): CollectAllPageIDs 按内部条目 BFS 下降;
- // - 哈希: 桶页 next 已统一到 PageHdr.NextPage, 一条链只登记一次, 提交后一次
- // FreeChain 还链 (O(1));
- // - 复合 / 时序: 扁平条目链, 走处理器自己的 PageCollector。
- //
- // 一定要在"新结构就位 (已 saveMeta)"之后调用, 否则归还的页会被 NewPage 立刻分给新结构,
- // 而这里还要按旧结构遍历 —— 会取到半截结构。
- func releaseIndexPages(table *Table, itype page.IndexType, rootPID uint32) {
- if rootPID == 0 {
- return
- }
- if isBTreePageType(itype) {
- for _, pid := range btree.CollectAllPageIDs(table.handle, rootPID) {
- table.deferFreePage(pid)
- }
- return
- }
- h, ok := index.Compose.GetHandler(itype)
- if !ok {
- return
- }
- // 整链延迟回收优先: 哈希索引的桶页 next 已统一到 PageHdr.NextPage, 且每条桶链的
- // (head, tail, count) 就持久化在目录槽里 —— 一条链只登记一次, 提交后一次 FreeChain
- // 还链 (O(1)), 不必逐页查询、逐页改写空闲链。
- if cc, ok := h.(index.ChainCollector); ok && table.handle.DeferFreeChain != nil {
- for _, ch := range cc.CollectChains(table.handle, rootPID) {
- if ch.Head == 0 || ch.Count == 0 {
- continue
- }
- if ch.Head == ch.Tail {
- table.deferFreePage(ch.Head) // 单页链 (例如元数据页)
- continue
- }
- table.handle.DeferFreeChain(ch.Head, ch.Tail, ch.Count)
- }
- return
- }
- pc, ok := h.(index.PageCollector)
- if !ok {
- return
- }
- for _, pid := range pc.CollectPageIDs(table.handle, rootPID) {
- table.deferFreePage(pid)
- }
- }
- // isBTreePageIndex 判断该索引的页是不是 B+ 节点页 (page.IndexHead + 内部条目结构)。
- //
- // 只有 B+ 树页能用 CollectAllPageIDs 安全遍历, 也只有它们能用
- // BuildFromSorted / SBTBuildFromSorted 批量构建。其余类型的页布局都不同:
- // - HashHandler: 元数据页 + 目录页链 + 桶页链。按 IndexHead 解析会把
- // BucketID / NextPage 当成 EntryLen / IsLeaf, 遍历出一串垃圾页 ID, 进而把
- // 不属于本索引的页归还掉 —— 那是数据损坏, 不是泄漏;
- // - ComposeChainHandler / TimeIndexHandler: 扁平条目链, 没有内部节点。
- //
- // 注意时序索引 (IndexTypeTime): 它同样是行存储索引, 但页是扁平链而非 B+ 树。
- // 因此这里**不能**用 indexStoresRows 来判断 —— 那样会把时序索引当成 B+ 树, 既在
- // 回收时漏收页链上的其余页, 又让重建拿 B+ 树页去覆盖它 (见 bulkRebuildSupported)。
- func isBTreePageIndex(m *IndexMeta) bool {
- if m == nil {
- return false
- }
- return isBTreePageType(m.IndexType)
- }
- // isBTreePageType 仅按类型判断页布局是否为 B+ 节点页 (与 isBTreePageIndex 同源)。
- func isBTreePageType(t page.IndexType) bool {
- switch t {
- case page.IndexTypePrimary, page.IndexTypeBTree:
- return true
- case page.IndexTypeStr:
- // IndexTypeStr 是 IndexTypeBTree 的旧值, 页格式相同。
- return true
- default:
- return false
- }
- }
- // bulkRebuildSupported 校验一个索引能否参与批量删除的"重建"流程。
- //
- // 批量删除统一采用「存活行写进全新 DataPage + 索引从零重建」, 因此要求每种索引
- // 都具备从零批量构建的能力。页布局决定做法:
- // - 主键 B+ 树 / B+ 树二级索引: BuildFromSorted 批量建树
- // - 复合 / 哈希二级索引: 清空根后走各自 handler.Insert 逐条重建
- //
- // 时序索引 (IndexTypeTime) 是"扁平链 + TimeDataPage": 行本身就住在按时间分段的
- // TimeDataPage 里, 重建必须重新分段, 与本流程的模型完全不同 —— 明确拒绝。
- // 旧实现没有这个校验, 会默默用 btree.BuildFromSorted 把时序索引整体改写成 B+ 树
- // (元数据仍写着 IndexTypeTime, 页却成了 B+ 节点), 之后 Search / RangeScan 全部错乱。
- func bulkRebuildSupported(m *IndexMeta) error {
- if m == nil {
- return nil
- }
- name := string(bytes.TrimRight(m.IndexName[:], "\x00"))
- if indexStoresRows(m) && m.IndexType != page.IndexTypePrimary {
- return fmt.Errorf(
- "批量删除不支持索引 %q (行存储索引类型 %d): 时序索引的行数据按时间分段存放, "+
- "批量重建需要重新分段, 请改用逐行删除", name, m.IndexType)
- }
- switch m.IndexType {
- case page.IndexTypePrimary, page.IndexTypeBTree, page.IndexTypeStr,
- page.IndexTypeCompose, page.IndexTypeHash:
- return nil
- default:
- return fmt.Errorf("批量删除不支持索引 %q 的类型 %d", name, m.IndexType)
- }
- }
- // checkRebuildSupported 在动任何数据之前确认全部活跃索引都能被重建。
- func (d *Deleter) checkRebuildSupported(pkMeta *IndexMeta) error {
- if err := bulkRebuildSupported(pkMeta); err != nil {
- return err
- }
- for i := range d.table.indexMeta.metas {
- m := &d.table.indexMeta.metas[i]
- if m.Flag&0x80 != 0 {
- continue
- }
- if err := bulkRebuildSupported(m); err != nil {
- return err
- }
- }
- return nil
- }
- // rebuildSecondaryViaHandler 用索引自己的 Insert 从零重建一个非 B+ 树二级索引
- // (复合索引的扁平页链 / 哈希索引的目录+桶链)。
- //
- // 为什么不能沿用 SBTBuildFromSorted: 它产出的页是 B+ 节点布局, 而这两种索引的页
- // 布局完全不同 —— 元数据仍写着 IndexTypeCompose / IndexTypeHash, 页却成了 B+ 树。
- // 之后 readHashMeta 在 page.IndexHeadSize 处的魔数校验直接失败
- // (errHashFormatIncompatible: 等值查找恒为 not found, 插入直接报错), 而复合索引
- // 的扁平链遍历会把 B+ 内部节点当条目读, 逐步把整条链写乱。这不只是"格式不统一",
- // 等于把索引写坏。
- //
- // 走 handler.Insert 的好处是正确性由正常写入路径背书: 哈希的目录扩容 / 桶分裂、
- // 复合链的页追加与链接全部自动走通。代价是这条路径为 O(n log n), 但只影响这两种
- // 索引 —— B+ 树二级索引仍是批量建树, 性能没有回退。
- //
- // entries 已按 key 排序: 复合链页内用二分查找, 必须有序; 哈希不要求有序, 不受影响。
- func rebuildSecondaryViaHandler(handle page.HandleDb, m *IndexMeta, entries [][]byte) error {
- name := string(bytes.TrimRight(m.IndexName[:], "\x00"))
- h, ok := index.Compose.GetHandler(m.IndexType)
- if !ok {
- return fmt.Errorf("重建索引 %q: 类型 %d 没有注册处理器", name, m.IndexType)
- }
- // 从空根开始重建; 旧结构由 releaseOldTree 按 PageCollector 单独归还。
- m.EntryPage = 0
- for _, e := range entries {
- if len(e) < 8 {
- continue
- }
- keyLen := len(e) - 8
- root, err := h.Insert(handle, &m.EntryPage, e[:keyLen],
- binary.BigEndian.Uint64(e[keyLen:]), nil, keyLen)
- if err != nil {
- return fmt.Errorf("重建索引 %q: %w", name, err)
- }
- m.EntryPage = root
- }
- return nil
- }
- // compareBytes 按字典序比较两个字节切片
- func compareBytes(a, b []byte) int {
- ml := len(a)
- if len(b) < ml {
- ml = len(b)
- }
- for i := 0; i < ml; i++ {
- if a[i] < b[i] {
- return -1
- }
- if a[i] > b[i] {
- return 1
- }
- }
- if len(a) < len(b) {
- return -1
- }
- if len(a) > len(b) {
- return 1
- }
- return 0
- }
|