package core import ( "cmp" "encoding/binary" "slices" "sync" "git.ycl.pub/yl32940/TsDB2/DB/core/index" "git.ycl.pub/yl32940/TsDB2/DB/core/index/btree" "git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage" "git.ycl.pub/yl32940/TsDB2/DB/core/page" "git.ycl.pub/yl32940/TsDB2/DB/core/types" ) // asyncFreeMemory 把 FreeOSMemory() 丢到后台 goroutine 执行, 不阻塞 VACUUM 主流程。 // // 之前每 rebuild 完一个索引就同步 asyncFreeMemory() —— 大表 (百万行 + 多索引) // VACUUM 时 GC 峰值可达数秒, 用户侧直接超时。后台化后 VACUUM 主线程立即返回, // GC 由 Go scheduler 在空闲时调度; 若未来需要可加参数开关禁用。 func asyncFreeMemory() { go func() { defer func() { _ = recover() }() asyncFreeMemory() }() } // 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) { defer d.table.db.lockOp()() // 批量删除全程串行; 内部 scanWithPKID/重建索引同 goroutine 重入 defer d.table.db.beginOp()() // 软删 + 全索引重建是一个操作, 中途阈值提交会切成半成品 // 找到主键索引 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) pkStore, _ := pkHandler.(index.RowStoreHandler) if pkStore == nil { return 0, nil // 主键索引不承载行数据, 软删除无从谈起 (正常不会发生) } dpIDs := pkStore.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.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() } slot := curDP.AppendRowSlot(storeRow) curDP.Save() if slot == 0 { // IsFull 预判与实际容量出现出入时的兜底: 宁可跳过, 也不要写出 slot=0 的坏叶条目 continue } // 收集叶条目: [pkID(8B BE) + dpID(4B BE) + slot(2B BE)] —— 带 slot 才能 O(1) 取行 pkID := binary.BigEndian.Uint64(storeRow[:8]) pkKeyBE := make([]byte, 8) binary.BigEndian.PutUint64(pkKeyBE, pkID) entry := make([]byte, 14) copy(entry[:8], pkKeyBE) binary.BigEndian.PutUint32(entry[8:12], curDPID) binary.BigEndian.PutUint16(entry[12:14], uint16(slot)) leafEntries = append(leafEntries, entry) } // 归还缓冲区到池 for i := range rows { rowPool.Put(&rows[i]) } } // 按 pkID 升序排序后再批量构建。 // BuildFromSorted 假设输入有序, 不能依赖 DataPage 扫描顺序: // 更新导致的行搬迁会把新 DataPage 排到扫描序列末尾, // 从而产生 1,2,3,4,6,...,100,5 这样的失序序列, 破坏 B+ 树不变式。 slices.SortFunc(leafEntries, func(a, b []byte) int { return cmp.Compare(binary.BigEndian.Uint64(a[:8]), binary.BigEndian.Uint64(b[: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 // 释放内存 asyncFreeMemory() // ===== 重建每个二级索引: 收集 → 排序 → 批量构建 ===== 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.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, pkID) 排序: 二级树的复合分隔键与叶内复合二分都要求 // 全树按复合键全局有序 —— 只按 secKey 排序不够 (同 key 条目的 pkID 顺序 // 无法靠"稳定排序保持物理序"保证, 数据页按主键序而非 pkID 序)。 slices.SortStableFunc(secEntries, func(a, b []byte) int { ka, kb := a[:len(a)-8], b[:len(b)-8] if c := compareBytes(ka, kb); c != 0 { return c } return cmp.Compare(binary.BigEndian.Uint64(a[len(a)-8:]), binary.BigEndian.Uint64(b[len(b)-8:])) }) 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 asyncFreeMemory() } return nil } // 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) } // 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 }