package core import ( "cmp" "fmt" "os" "path/filepath" "slices" "time" "git.ycl.pub/yl32940/TsDB2/DB/core/page" "git.ycl.pub/yl32940/TsDB2/DB/core/types" ) // ============================================================================ // VACUUM 整库重写 —— 回收整库碎片, 让文件真正缩小 // // 工作原理: // 扫描旧库 schema + 全部行 → 在临时文件 .vacuum.tmp 上重建 → // 原子 rename 替换旧文件 → 重新打开。 // // 与自动 compact (40% 失效字节触发, 详见 datapage.go) 的区别: // - 自动 compact: 页内墓碑回收, 文件大小几乎不变 // - VACUUM: 整库重写, 文件真正缩小, 归还所有空闲页给 OS // // 失败回滚: 任何阶段失败 → 删除临时文件, 旧库保持不变。 // ============================================================================ // VACUUMMode VACUUM 工作模式 type VACUUMMode int const ( // VACUUMModeSync 同步模式: 锁写后单步完成, 阻塞所有读写。 VACUUMModeSync VACUUMMode = iota // VACUUMModeOnline 在线模式: 扫描期允许读 (但不允许写), // 切换期短暂阻塞。当前 MVP 实现退化为 Sync (Close-Open), // 保留 API 形状供后续接入增量捕获。 VACUUMModeOnline ) // VACUUMProgress 进度回调类型。 // phase: "scan" | "build" | "index" | "swap" | "done" | "fail" // copied: 已复制行数 (scan/build/index 阶段有效) // total: 总行数 (估计, 0 表示未知) type VACUUMProgress func(phase string, copied, total int64) // VACUUMOptions VACUUM 选项 type VACUUMOptions struct { // Progress 进度回调 (nil = 不调用) Progress VACUUMProgress // Mode 工作模式, 默认 VACUUMModeSync Mode VACUUMMode // KeepTempOnError 失败时保留临时文件供分析, 默认 false KeepTempOnError bool } // VACUUMResult VACUUM 结果 type VACUUMResult struct { OldFileSize int64 NewFileSize int64 SavedBytes int64 SavedPercent float64 Tables int RowsCopied int64 Duration time.Duration Mode VACUUMMode } // vacuumSrcRow 源行快照 (deserializeRow 的产物, 跨 Copy 安全) type vacuumSrcRow struct { pkID uint64 vals []types.FieldValue } // vacuumSrcIndex 源索引定义 type vacuumSrcIndex struct { name string typ page.IndexType fields []byte unique bool } // vacuumSrcTable 源表信息 (schema + 索引 + 行快照) type vacuumSrcTable struct { name string fields []types.FieldInfo indexes []vacuumSrcIndex // rowStore 行存储索引 (主键/时序) 的原始定义; nil = 源表没有 // (core API 允许不建行存储索引, 此时重建也不能凭空建一个)。 rowStore *vacuumSrcIndex // fieldRemap 旧 master 下标 → 新活跃下标。新表剔除了 DROP 列, // 列下标整体前移; 不映射就拿旧 FieldIDs 建新索引 → 越界/错位。 fieldRemap map[byte]byte rows []vacuumSrcRow // tsKeyName 时序表的时间键列名 (rowStore.typ == Time 时非空), // 重建前按它排序行, 避免乱序时间戳行触发 ErrOutOfOrder 被静默跳过。 tsKeyName string } // vacuumRowTS 取行快照里时间键列的 uint64 值 (时序表重建排序用); // 缺列或类型不符返回 0。 func vacuumRowTS(vals []types.FieldValue, keyName string) uint64 { for _, fv := range vals { if string(fv.FieldName) != keyName { continue } switch v := fv.Value.(type) { case uint64: return v case int64: return uint64(v) } return 0 } return 0 } // remapIndexFields 把索引分量的旧 master 下标换算到新表的活跃下标; // 映射缺失的分量原样保留 (不该发生, 保住可见错误而非静默错位)。 func remapIndexFields(fields []byte, remap map[byte]byte) []byte { out := make([]byte, len(fields)) for i, f := range fields { if nf, ok := remap[f]; ok { out[i] = nf } else { out[i] = f } } return out } // ============================================================================ // 主入口 // ============================================================================ // VACUUM 整库重写, 让数据库文件真正缩小。 // // 拒绝条件: // - 当前有活跃事务 (db.currentTxn != nil) // - db 路径不可用 // // 调用后 db 句柄会被关闭并重新打开, 调用者继续使用即可。 func (db *DB) VACUUM(opts ...VACUUMOptions) (*VACUUMResult, error) { opt := VACUUMOptions{Mode: VACUUMModeSync} if len(opts) > 0 { opt = opts[0] } if db.currentTxn.Load() != nil { return nil, fmt.Errorf("%w: VACUUM", ErrTxnActive) } if db.path == "" || db.path == os.DevNull { return nil, fmt.Errorf("%w (%q)", ErrVACUUMNoPath, db.path) } // 手动与自动调度可能并发触发: 双开会对同一文件 Close/Open 交错, 必须串行。 db.vacuumMu.Lock() defer db.vacuumMu.Unlock() // 操作级串行锁: VACUUM 全程 Close/Open/rename, 必须持有到 adoptEngineState 之后, // 防止其他 goroutine 在切换间隙操作句柄。内部 db.Close()/db.Commit() 同 goroutine 重入, // 可重入锁保证它们不会误交这把锁。 db.accMu.Lock() defer db.accMu.Unlock() progress := opt.Progress if progress == nil { progress = func(string, int64, int64) {} } start := time.Now() // 1. 把 WAL / 脏页全部落盘, 拿到一致快照 if err := db.Commit(); err != nil { return nil, fmt.Errorf("VACUUM: 提交源库失败: %w", err) } srcStat, _ := os.Stat(db.path) var oldSize int64 if srcStat != nil { oldSize = srcStat.Size() } // 2. 扫描旧库 progress("scan", 0, 0) sources, totalRows, err := db.vacuumScan(progress) if err != nil { return nil, fmt.Errorf("VACUUM: 扫描源库失败: %w", err) } // 3. 关闭旧 db (确保所有页都落盘) if err := db.Close(); err != nil { _ = os.Remove(db.path + ".vacuum.tmp") return nil, fmt.Errorf("VACUUM: 关闭源库失败: %w", err) } // 4. 在临时路径上重建 tmpPath := db.path + ".vacuum.tmp" _ = os.Remove(tmpPath) // 清理残留 progress("build", 0, totalRows) newDB, err := Open(tmpPath) if err != nil { return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 打开临时库失败: %w", err)) } // 5. 重建表 + 索引 + 插入行 var rowsCopied int64 for _, src := range sources { // 5a. 建表 newTbl, err := newDB.CreateTable(src.name, src.fields) if err != nil { _ = newDB.Close() _ = os.Remove(tmpPath) return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 重建表 %q 失败: %w", src.name, err)) } // 行存储索引按【原始名字/原始类型】重建: 旧实现无条件建成 // ("pk", Primary) —— ORM 按 pk_<表名> 寻址, 重建后 Find/Delete/Update // 全部 ErrFieldNotFound; 时序表则被静默改成主键表, 分段语义丢失。 if src.rowStore != nil { rsFields := remapIndexFields(src.rowStore.fields, src.fieldRemap) if _, err := newTbl.CreateIndex(src.rowStore.name, src.rowStore.typ, rsFields); err != nil { _ = newDB.Close() _ = os.Remove(tmpPath) return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 重建行存储索引 %q 失败: %w", src.rowStore.name, err)) } } // 5b. 插入全部行 (走主键索引) for ri, row := range src.rows { if _, err := newTbl.InsertRowWithID(row.pkID, row.vals); err != nil { // 单行失败: 跳过 (deserializeRow 失败/罕见约束冲突) continue } rowsCopied++ if ri%1000 == 0 { progress("build", rowsCopied, totalRows) } } // 周期 commit, 减少 WAL 大小 if err := newDB.Commit(); err != nil { _ = newDB.Close() _ = os.Remove(tmpPath) return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 提交 %q 失败: %w", src.name, err)) } progress("build", rowsCopied, totalRows) // 5c. 重建二级索引 (自动批量回填已存在的行)。行存储索引已按 // 原始定义建好, 跳过; FieldIDs 经 fieldRemap 换算到新表下标。 progress("index", rowsCopied, totalRows) for _, idx := range src.indexes { if isRowStoreType(idx.typ) { continue // 行存储索引已在 CreateTable 后手动建好 } if _, err := newTbl.CreateIndex(idx.name, idx.typ, remapIndexFields(idx.fields, src.fieldRemap), idx.unique); err != nil { _ = newDB.Close() _ = os.Remove(tmpPath) return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 重建索引 %q 失败: %w", idx.name, err)) } } progress("index", rowsCopied, totalRows) } // 6. 关闭新 db if err := newDB.Commit(); err != nil { _ = newDB.Close() _ = os.Remove(tmpPath) return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 最终提交失败: %w", err)) } if err := newDB.Close(); err != nil { _ = os.Remove(tmpPath) return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 关闭新库失败: %w", err)) } // 7. fsync 父目录 (确保 rename 的目录项落盘) if dir := filepath.Dir(db.path); dir != "" { if d, derr := os.Open(dir); derr == nil { _ = d.Sync() _ = d.Close() } } // 8. 原子替换 progress("swap", 0, 0) if err := os.Rename(tmpPath, db.path); err != nil { if !opt.KeepTempOnError { _ = os.Remove(tmpPath) } return db.reopenAfterVacuum(nil, fmt.Errorf("VACUUM: 替换文件失败: %w", err)) } // 9. 重新打开 progress("done", 0, 0) newDB2, err := Open(db.path) if err != nil { return nil, fmt.Errorf("VACUUM: 重新打开失败: %w", err) } db.adoptEngineState(newDB2) // 10. 统计结果 newStat, _ := os.Stat(db.path) var newSize int64 if newStat != nil { newSize = newStat.Size() } result := &VACUUMResult{ OldFileSize: oldSize, NewFileSize: newSize, SavedBytes: oldSize - newSize, Tables: len(sources), RowsCopied: rowsCopied, Duration: time.Since(start), Mode: opt.Mode, } if oldSize > 0 { result.SavedPercent = float64(oldSize-newSize) * 100 / float64(oldSize) } return result, nil } // ============================================================================ // 扫描源库 // ============================================================================ // vacuumScan 扫描源库, 拿到 schema + 索引定义 + 行快照 func (db *DB) vacuumScan(progress VACUUMProgress) ([]vacuumSrcTable, int64, error) { tables := db.GetAllTables() sources := make([]vacuumSrcTable, 0, len(tables)) // 先统计总行数 (用于进度回调的 total) totalRows := int64(0) for _, meta := range tables { tbl, ok := db.FindTable(meta.Name) if !ok { continue } totalRows += int64(tbl.TotalRows()) } progress("scan", 0, totalRows) var scanned int64 for _, meta := range tables { tbl, ok := db.FindTable(meta.Name) if !ok { continue } // 字段定义 (剔除已 DROP 的列)。同时建立 旧 master 下标 → 新活跃下标 // 的映射: 新表 registry 只含活跃列, 下标整体前移, 索引的 FieldIDs // 必须按映射换算, 否则"删过列"的表上重建索引必然越界/错位。 fields := tbl.GetFields() activeFields := make([]types.FieldInfo, 0, len(fields)) fieldRemap := make(map[byte]byte, len(fields)) for i, f := range fields { if f.IsDropped() { continue } fieldRemap[byte(i)] = byte(len(activeFields)) activeFields = append(activeFields, f) } // 索引定义 var srcIndexes []vacuumSrcIndex var rowStore *vacuumSrcIndex for _, im := range tbl.IndexMeta().GetIndexes() { idx := vacuumSrcIndex{ name: im.Name(), typ: im.IndexType, fields: append([]byte(nil), im.Fields()...), unique: im.Unique(), } srcIndexes = append(srcIndexes, idx) if isRowStoreType(im.IndexType) { captured := idx rowStore = &captured } } // 行快照 (保留 NULL: deserializeRow 会跳过 NULL 列, VACUUM 会把 // NULL 物化成类型零值 —— 语义漂移且不可逆)。 // ⭐ Blob 列特判: deserialize 产出的是 8B BlobRef (指向【本文件】的链页), // 直接搬过去新库后链页并不存在 —— 读回 EOF/空 (DockerManager 实测: // 迁移+VACUUM 后 param 列全崩)。必须在扫描时把链内容物化进 vals, // 让新库的序列化走 blobFieldRef 落【新链】。 blobCols := map[string]bool{} for _, f := range fields { if types.IsBlob(f.FieldType) { blobCols[string(f.FieldName)] = true } } rows := make([]vacuumSrcRow, 0, tbl.TotalRows()) q := tbl.Query() var blobErr error // 闭包内物化 Blob 的失败载体 (闭包不能引用后声明的 scanErr) scanErr := q.fullTableScanRaw(func(pkID uint64, rowData []byte) bool { vals := tbl.deserializeRowKeepNull(rowData) if vals == nil { return true // 解析失败, 跳过 } if len(blobCols) > 0 { for i, fv := range vals { if !blobCols[string(fv.FieldName)] { continue } ref, ok := fv.Value.(types.BlobRef) if !ok || ref.Empty() { continue } data, err := tbl.BlobRead(ref, types.CompressNone) // 写侧约定 CompressNone if err != nil { blobErr = fmt.Errorf("物化 Blob 列 %q (pkID=%d): %w", fv.FieldName, pkID, err) return false } vals[i].Value = data } if blobErr != nil { return false } } rows = append(rows, vacuumSrcRow{pkID: pkID, vals: vals}) scanned++ if scanned%10000 == 0 { progress("scan", scanned, totalRows) } return true }) if blobErr != nil { return nil, 0, blobErr } if scanErr != nil { return nil, 0, fmt.Errorf("扫描表 %q: %w", meta.Name, scanErr) } // 时序表: 重建前按时间键排序。时序索引断言时间有序, 按 pkID 序 // 插入乱序时间戳行会 ErrOutOfOrder 被静默跳过 (数据丢失)。 tsKeyName := "" if rowStore != nil && rowStore.typ == page.IndexTypeTime && len(rowStore.fields) > 0 { fi := int(rowStore.fields[0]) if fi < len(fields) { tsKeyName = string(fields[fi].FieldName) } slices.SortStableFunc(rows, func(a, b vacuumSrcRow) int { return cmp.Compare(vacuumRowTS(a.vals, tsKeyName), vacuumRowTS(b.vals, tsKeyName)) }) } sources = append(sources, vacuumSrcTable{ name: meta.Name, fields: activeFields, indexes: srcIndexes, rowStore: rowStore, fieldRemap: fieldRemap, tsKeyName: tsKeyName, rows: rows, }) } progress("scan", totalRows, totalRows) return sources, totalRows, nil } // ============================================================================ // 自动调度 // ============================================================================ // SetAutoVACUUM 开启自动 VACUUM, 周期触发。 // // interval: 两次 VACUUM 的最小间隔 (例如 1*time.Hour) // options: 每次 VACUUM 使用的选项 // // 关闭: SetAutoVACUUM(0) func (db *DB) SetAutoVACUUM(interval time.Duration, opts ...VACUUMOptions) { db.autoVacuumMu.Lock() defer db.autoVacuumMu.Unlock() if interval <= 0 { if db.autoVacuumTimer != nil { db.autoVacuumTimer.Stop() db.autoVacuumTimer = nil } db.autoVacuumEnabled = false return } opt := VACUUMOptions{Mode: VACUUMModeSync} if len(opts) > 0 { opt = opts[0] } db.autoVacuumOpts = opt db.autoVacuumEnabled = true db.lastVACUUM = time.Now() run := func() { db.autoVacuumMu.Lock() enabled := db.autoVacuumEnabled curOpt := db.autoVacuumOpts db.autoVacuumMu.Unlock() if !enabled { return } // VACUUM 本体拿 vacuumMu 串行: 与并发手动 VACUUM 排队执行, // 不会双开对同一文件 Close/Open 交错。 // 自动调度: 失败也吞掉 (用户可通过 Progress 回调处理) _, _ = db.VACUUM(curOpt) db.autoVacuumMu.Lock() db.lastVACUUM = time.Now() stillEnabled := db.autoVacuumEnabled // Reset 必须在锁内: SetAutoVACUUM(0) 会并发把 timer 置 nil, // 锁外读/重置是 data race (race detector 实测抓到)。 if stillEnabled && db.autoVacuumTimer != nil { db.autoVacuumTimer.Reset(interval) } db.autoVacuumMu.Unlock() } db.autoVacuumTimer = time.AfterFunc(interval, run) } // LastVACUUM 返回上次自动 VACUUM 的时间 func (db *DB) LastVACUUM() time.Time { db.autoVacuumMu.Lock() defer db.autoVacuumMu.Unlock() return db.lastVACUUM } // ============================================================================ // 辅助 // ============================================================================ // reopenAfterVacuum VACUUM 失败时尝试重新打开 db.path, 让 db 句柄继续可用 // // 返回 (*VACUUMResult, error): 即使失败也返回 nil result + 错误, // 调用方处理逻辑一致。 func (db *DB) reopenAfterVacuum(_ *VACUUMResult, origErr error) (*VACUUMResult, error) { newDB, err := Open(db.path) if err != nil { return nil, fmt.Errorf("%w; 且重新打开源库也失败: %v", origErr, err) } db.adoptEngineState(newDB) return nil, origErr } // adoptEngineState 用 other (刚重新打开的句柄) 的引擎状态替换 db 的, // 保留 db 上用户配置的运行时参数 (autoCommitDelay / autoVacuum 调度 / lastVACUUM)。 // // VACUUM 完成后 db.path 已被替换成重建的文件, 必须换引擎状态才能继续使用句柄; // 但不能整体 `*db = *other`: DB 含多把互斥锁, 结构体整体赋值会把锁的内部状态 // 一并按值拷贝 (go vet copylocks 报错), 更糟的是会把用户配置覆盖成零值 —— // autoVacuumEnabled 变回 false 导致自动 VACUUM 首轮之后不再重排, // autoCommitDelay 被重置成 0 导致防抖自动提交静默失效。 func (db *DB) adoptEngineState(other *DB) { db.file = other.file db.header = other.header db.tables = other.tables db.currentTxn.Store(nil) db.pendingFree = nil db.pendingFreeChain = nil db.freePage = other.freePage db.freeChain = other.freeChain // other 的表树在 Open 时把内部 db 指针绑到了 other (bindDB 只补 nil, 不会改绑), // 必须强制改绑到本句柄, 否则后续表操作会读到 other 的幽灵状态 // (独立的锁、永远为 nil 的事务指针)。 db.tables.db = db for _, tbl := range db.tables.tables { tbl.db = db } // 运行期回调与句柄闭包重绑到本句柄: other 的 onPageWrite / autoCommit / // handle 闭包捕获的是 other —— 不重绑的话, VACUUM 之后开事务, // markTouched 查的是 other.currentTxn (恒 nil), Rollback 静默失效; // 新页登记到 other.allocated (回滚后成孤儿页); 延迟回收进 other.pendingFree // (永不 drain, 页永久泄漏)。 db.wireRuntime() } // isRowStoreType 索引是否为行存储索引 (主键 / 时序) // // 行存储索引在 step 5a 已由 VACUUM 自己手动建好, 重建二级索引时跳过, 避免重复。 func isRowStoreType(typ page.IndexType) bool { return typ == page.IndexTypePrimary || typ == page.IndexTypeTime }