TsDB2/DB/core/vacuum_test.go

package core

import (
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)

// vacNewTable 建一个普通主键表 (uint64 id + uint64 v)
func vacNewTable(t *testing.T, db *DB, name string) *Table {
	t.Helper()
	tbl, err := db.CreateTable(name, []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("建表 %s: %v", name, err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	return tbl
}

// vacInsert 插入 n 行 (id=i+1, v=i*10)
func vacInsert(t *testing.T, tbl *Table, n int) {
	t.Helper()
	for i := 0; i < n; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i+1), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i + 1)},
			{FieldName: []byte("v"), Value: uint64(i * 10)},
		}); err != nil {
			t.Fatalf("插入第 %d 行: %v", i+1, err)
		}
	}
}

// vacCount 拿表的当前行数
func vacCount(t *testing.T, tbl *Table) int {
	t.Helper()
	c, err := tbl.Query().Count()
	if err != nil {
		t.Fatalf("Count: %v", err)
	}
	return int(c)
}

// ============================================================================
// 测试用例
// ============================================================================

// TestVACUUMEmpty 空库跑 VACUUM 不应崩溃
func TestVACUUMEmpty(t *testing.T) {
	db := fixOpenDB(t)
	res, err := db.VACUUM()
	if err != nil {
		t.Fatalf("空库 VACUUM 失败: %v", err)
	}
	t.Logf("空库 VACUUM 结果: tables=%d rows=%d dur=%v", res.Tables, res.RowsCopied, res.Duration)
	if res.Tables != 0 {
		t.Errorf("空库应 tables=0, 得到 %d", res.Tables)
	}
}

// TestVACUUMKeepData 插 1000 行, 跑 VACUUM, 行数与数据应一致
func TestVACUUMKeepData(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	db.SetAutoCommitDelay(0)
	t.Cleanup(func() { _ = db.Close() })

	tbl := vacNewTable(t, db, "t")
	vacInsert(t, tbl, 1000)
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	oldSize, _ := os.Stat(path)

	res, err := db.VACUUM(VACUUMOptions{
		Progress: func(phase string, copied, total int64) {
			if copied%100 == 0 || phase == "done" || phase == "scan" || phase == "swap" {
				t.Logf("  progress: phase=%s copied=%d total=%d", phase, copied, total)
			}
		},
	})
	if err != nil {
		t.Fatalf("VACUUM 失败: %v", err)
	}

	t.Logf("整理结果: 旧=%d 新=%d 节省=%d (%.1f%%) tables=%d rows=%d 耗时=%v",
		res.OldFileSize, res.NewFileSize, res.SavedBytes, res.SavedPercent,
		res.Tables, res.RowsCopied, res.Duration)

	if res.RowsCopied != 1000 {
		t.Errorf("应复制 1000 行, 得到 %d", res.RowsCopied)
	}
	if res.Tables != 1 {
		t.Errorf("应有 1 张表, 得到 %d", res.Tables)
	}
	if res.NewFileSize == 0 {
		t.Errorf("新文件大小不应为 0")
	}
	if oldSize != nil && res.OldFileSize != oldSize.Size() {
		t.Errorf("旧文件大小不匹配: 期望 %d, 得到 %d", oldSize.Size(), res.OldFileSize)
	}

	c := vacCount(t, tbl)
	if c != 1000 {
		t.Errorf("VACUUM 后行数应仍为 1000, 得到 %d", c)
	}

	// 抽查几行的值
	for _, id := range []uint64{1, 100, 500, 999, 1000} {
		row, err := tbl.FindByPK(id)
		if err != nil {
			t.Errorf("查 id=%d 失败: %v", id, err)
			continue
		}
		expectedV := uint64(id-1) * 10
		if len(row) >= 2 {
			if v, ok := row[1].Value.(uint64); !ok || v != expectedV {
				t.Errorf("id=%d 的 v 应为 %d, 得到 %v", id, expectedV, row[1])
			}
		}
	}
}

// TestVACUUMShrinkFile 删一半数据后 VACUUM, 文件应显著缩小
func TestVACUUMShrinkFile(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	db.SetAutoCommitDelay(0)
	t.Cleanup(func() { _ = db.Close() })

	tbl := vacNewTable(t, db, "t")
	vacInsert(t, tbl, 5000)
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	oldSize, _ := os.Stat(path)
	if oldSize == nil {
		t.Fatal("拿不到旧大小")
	}
	t.Logf("插完: %d 字节", oldSize.Size())

	for id := uint64(1); id <= 2500; id++ {
		if err := tbl.DeleteRowByPK(id); err != nil {
			t.Fatalf("删 id=%d 失败: %v", id, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	afterDelete, _ := os.Stat(path)
	t.Logf("删一半后: %d 字节 (只有墓碑, 不明显缩)", afterDelete.Size())

	res, err := db.VACUUM()
	if err != nil {
		t.Fatalf("VACUUM 失败: %v", err)
	}
	t.Logf("VACUUM 后: %d 字节 (节省 %d, %.1f%%) 耗时=%v",
		res.NewFileSize, res.SavedBytes, res.SavedPercent, res.Duration)

	if res.NewFileSize >= afterDelete.Size() {
		t.Logf("注意: VACUUM 后文件未明显缩小 (旧=%d 新=%d) — 小库可能因最小页数未减少",
			afterDelete.Size(), res.NewFileSize)
	}

	c := vacCount(t, tbl)
	if c != 2500 {
		t.Errorf("VACUUM 后应剩 2500 行, 得到 %d", c)
	}
}

// TestVACUUMMultiTable 多表多索引 VACUUM
func TestVACUUMMultiTable(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	db.SetAutoCommitDelay(0)
	t.Cleanup(func() { _ = db.Close() })

	expectedRows := 0
	for i, name := range []string{"a", "b", "c"} {
		tbl := vacNewTable(t, db, name)
		n := 100 * (i + 1)
		vacInsert(t, tbl, n)
		expectedRows += n
		if name == "a" {
			// 表 a 加二级 B+ 树索引 (v 列, 唯一)
			if _, err := tbl.CreateIndex("v_idx", page.IndexTypeBTree, []byte{1}, true); err != nil {
				t.Fatalf("建二级索引: %v", err)
			}
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	res, err := db.VACUUM()
	if err != nil {
		t.Fatalf("VACUUM 失败: %v", err)
	}
	t.Logf("MultiTable VACUUM: copied=%d tables=%d old=%d new=%d", res.RowsCopied, res.Tables, res.OldFileSize, res.NewFileSize)

	if res.Tables != 3 {
		t.Errorf("应有 3 张表, 得到 %d", res.Tables)
	}
	if res.RowsCopied != int64(expectedRows) {
		t.Errorf("应复制 %d 行, 得到 %d", expectedRows, res.RowsCopied)
	}

	for i, name := range []string{"a", "b", "c"} {
		tbl, ok := db.FindTable(name)
		if !ok {
			t.Errorf("表 %s 不见了", name)
			continue
		}
		c := vacCount(t, tbl)
		// 直接读主键 DataPage 的 LiveCount
		primMeta, ok := tbl.IndexMeta().GetRowStoreIndex()
		t.Logf("  表 %s: Count()=%d TotalRows=%d primMeta=%+v ok=%v",
			name, c, tbl.TotalRows(), primMeta, ok)
		if c != 100*(i+1) {
			t.Errorf("表 %s 应有 %d 行, 得到 %d", name, 100*(i+1), c)
		}
	}

	// 验证表 a 的二级索引: v=990 对应 id=100 (因为 v=(id-1)*10, 即 v=99*10=990)
	tblA, _ := db.FindTable("a")
	row, err := tblA.SearchByIndex("v_idx", encodeUint64BE(990))
	if err != nil {
		t.Errorf("通过二级索引查 v=990 失败: %v", err)
	}
	if len(row) >= 1 {
		if id, ok := row[0].Value.(uint64); !ok || id != 100 {
			t.Errorf("通过 v=990 期望 id=100, 得到 %v", row[0])
		}
	}
}

// TestVACUUMRejectTxn 活跃事务中调用 VACUUM 应被拒绝
func TestVACUUMRejectTxn(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	db.SetAutoCommitDelay(0)
	t.Cleanup(func() { _ = db.Close() })

	tbl := vacNewTable(t, db, "t")
	vacInsert(t, tbl, 10)

	_ = db.BeginTransaction()
	_, err = db.VACUUM()
	if err == nil {
		t.Error("活跃事务中 VACUUM 应报错, 但通过了")
	} else {
		t.Logf("正确拒绝: %v", err)
	}
}

// TestVACUUMRejectNoPath 对 /dev/null 库的 VACUUM 应被拒绝
func TestVACUUMRejectNoPath(t *testing.T) {
	// 这个测试用例跳过 — 我们没有合适的"无路径" db 创建方式
	// 实际生产中, Open 时会强制要求路径
	t.Skip("跳过: 无路径 db 创建路径")
}

// TestAutoVACUUM 自动调度: 间隔触发
func TestAutoVACUUM(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	db.SetAutoCommitDelay(0)
	t.Cleanup(func() {
		db.SetAutoVACUUM(0)
		_ = db.Close()
	})

	tbl := vacNewTable(t, db, "t")
	vacInsert(t, tbl, 50)
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	// 启用自动 VACUUM: 间隔 50ms (测试用短间隔)
	db.SetAutoVACUUM(50 * time.Millisecond)

	// 轮询等首次自动 VACUUM 完成 (固定 sleep 在 -race 下必然脆弱)
	before := db.LastVACUUM()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && !db.LastVACUUM().After(before) {
		time.Sleep(10 * time.Millisecond)
	}

	// 期间加点数据。自动 VACUUM 可能已换过引擎状态, 必须用【当前句柄】取表。
	tbl2, ok := db.FindTable("t")
	if !ok {
		t.Fatalf("自动 VACUUM 后表丢失")
	}
	for i := 100; i < 110; i++ {
		if _, err := tbl2.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("v"), Value: uint64(i * 10)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	_ = db.Commit()

	// 关闭自动 VACUUM
	db.SetAutoVACUUM(0)

	// 验证表还在 (初始 50 行 + 新插 10 行; 用当前句柄取表再数)
	tbl3, ok := db.FindTable("t")
	if !ok {
		t.Fatalf("取表失败")
	}
	cnt, _ := tbl3.Query().Count()
	if cnt < 50 {
		t.Errorf("行数应不少于 50, 得到 %d", cnt)
	}
	t.Logf("LastVACUUM: %v (当前: %v)", db.LastVACUUM(), time.Now())
}

// TestAutoVACUUMRepeats 回归测试: 自动 VACUUM 必须周期重排, 而不是首轮之后静默停摆。
//
// 旧实现 VACUUM 内部 `*db = *newDB2` 整体结构体拷贝, 把 autoVacuumEnabled 覆盖成
// false —— 首轮跑完后调度器读到 stillEnabled=false 不再重排定时器, 自动 VACUUM
// 实际只生效一次 (且该整体拷贝因复制互斥锁被 go vet copylocks 报警)。
func TestAutoVACUUMRepeats(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	db.SetAutoCommitDelay(0)
	t.Cleanup(func() {
		db.SetAutoVACUUM(0)
		_ = db.Close()
	})

	tbl := vacNewTable(t, db, "t")
	vacInsert(t, tbl, 20)
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	var mu sync.Mutex
	dones := 0
	db.SetAutoVACUUM(40*time.Millisecond, VACUUMOptions{
		Progress: func(phase string, _, _ int64) {
			if phase == "done" {
				mu.Lock()
				dones++
				mu.Unlock()
			}
		},
	})

	deadline := time.Now().Add(600 * time.Millisecond)
	for time.Now().Before(deadline) {
		time.Sleep(20 * time.Millisecond)
		mu.Lock()
		n := dones
		mu.Unlock()
		if n >= 2 {
			break
		}
	}
	db.SetAutoVACUUM(0)

	mu.Lock()
	defer mu.Unlock()
	if dones < 2 {
		t.Fatalf("自动 VACUUM 应周期重排, 600ms 内只触发了 %d 次", dones)
	}
	t.Logf("600ms 内自动 VACUUM 触发 %d 次", dones)
}

// TestVACUUMPreservesRuntimeConfig 回归测试: VACUUM 更换引擎状态后,
// 用户配置 (autoCommitDelay) 必须保留, 句柄必须继续可用。
//
// 旧实现整体结构体拷贝会把 autoCommitDelay 重置为默认值 —— 用户设的手动提交模式
// (0 = 禁用防抖) 被静默改回 1s 防抖。
func TestVACUUMPreservesRuntimeConfig(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "test.db")

	db, err := Open(path)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	defer db.Close()

	db.SetAutoCommitDelay(0) // 手动提交模式
	tbl := vacNewTable(t, db, "t")
	vacInsert(t, tbl, 10)
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	if _, err := db.VACUUM(); err != nil {
		t.Fatalf("VACUUM: %v", err)
	}

	// 1) 运行时配置保留: 手动提交模式不被静默改回防抖模式
	if db.autoCommitDelay != 0 {
		t.Fatalf("VACUUM 后 autoCommitDelay 应保留为 0, 得到 %v", db.autoCommitDelay)
	}

	// 2) 句柄可用: VACUUM 之后建新表、插入、查询
	tbl2 := vacNewTable(t, db, "t2")
	for i := 0; i < 5; i++ {
		if _, err := tbl2.InsertRowWithID(uint64(i+1), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i + 1)},
			{FieldName: []byte("v"), Value: uint64(i * 100)},
		}); err != nil {
			t.Fatalf("VACUUM 后插入第 %d 行: %v", i+1, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("VACUUM 后 commit: %v", err)
	}
	if got := vacCount(t, tbl2); got != 5 {
		t.Fatalf("t2 应有 5 行, 得到 %d", got)
	}
}

// ============================================================================
// 辅助
// ============================================================================

// encodeUint64BE 编码 uint64 为大端 8 字节 (二级索引查找用)
func encodeUint64BE(v uint64) []byte {
	b := make([]byte, 8)
	for i := 7; i >= 0; i-- {
		b[i] = byte(v)
		v >>= 8
	}
	return b
}

var _ = fmt.Sprintf