TsDB2/DB/core/vacuum.go

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 + 全部行 → 在临时文件 <path>.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
}