TsDB2/demo/verify_stress/main.go

package main

import (
	"git.ycl.pub/yl32940/TsDB2/DB"
	"fmt"
	"os"
	"runtime"
	"strings"
	"time"
)

// ========== 数据模型 ==========
type Employee struct {
	DB.BaseModel
	Name  string `vrtx:"maxlen:32"`
	Age   int64  `vrtx:"index:idx_age"`
	Score int64  `vrtx:"index:idx_score"`
	Dept  string `vrtx:"maxlen:16,index:idx_dept"`
}

type Order struct {
	DB.BaseModel
	UserID int64 `vrtx:"index:idx_userid"`
	Amount int64
	Status int64 `vrtx:"index:idx_status"`
}

const (
	dbPath     = "stress_test.db"
	EMPLOYEE_N = 500_000
	ORDER_N    = 100_000
)

var db *DB.DB
var passed, failed int

func memStats() string {
	var m runtime.MemStats
	runtime.ReadMemStats(&m)
	return fmt.Sprintf("Go堆=%.1fMB Sys=%.1fMB", float64(m.HeapInuse)/1024/1024, float64(m.Sys)/1024/1024)
}

func fileSize() string {
	fi, err := os.Stat(dbPath)
	if err != nil {
		return "N/A"
	}
	return fmt.Sprintf("%.1fMB", float64(fi.Size())/1024/1024)
}

func check(msg string, ok bool) {
	if ok {
		passed++
		fmt.Printf("  ✅ %s\n", msg)
	} else {
		failed++
		fmt.Printf("  ❌ %s\n", msg)
	}
}

func section(title string) {
	fmt.Printf("\n%s %s\n", strings.Repeat("=", 50), title)
}

func main() {
	os.Remove(dbPath)
	os.Remove(dbPath + ".wal")

	fmt.Println(strings.Repeat("=", 60))
	fmt.Printf("  TsDB2 大数据压力测试 (%d万行 Employee + %d万行 Order)\n", EMPLOYEE_N/10000, ORDER_N/10000)
	fmt.Println("  覆盖: 批量插入/多索引/事务/WAL/条件查询/排序/聚合/持久化")
	fmt.Println(strings.Repeat("=", 60))

	// ============================================================
	// 1. 打开数据库
	// ============================================================
	section("1. 打开数据库")
	var err error
	db, err = DB.Open(dbPath)
	check("OpenDB", err == nil)
	db.SetAutoCommitDelay(0) // 禁用自动提交, 手动控制
	db.SetBufferPoolCapacity(2000)
	db.SetDirtyThreshold(500)
	fmt.Printf("  配置: BufferPool=2000页(32MB), DirtyThreshold=500页\n")
	fmt.Printf("  [初始状态] 磁盘: %s | %s\n", fileSize(), memStats())

	// ============================================================
	// 2. 建表 + 多索引
	// ============================================================
	section("2. 建表 + 多索引")

	empTbl, err := DB.UseTable[Employee](db, "employee")
	check("UseTable[Employee]", err == nil)
	// 索引已通过 vrtx 标签自动创建: idx_age, idx_dept, idx_score
	fmt.Println("  ✅ 索引: pk_employee + idx_age + idx_dept + idx_score (标签自动创建)")
	passed++

	orderTbl, err := DB.UseTable[Order](db, "order")
	check("UseTable[Order]", err == nil)
	// 索引已通过 vrtx 标签自动创建: idx_status, idx_userid
	fmt.Println("  ✅ 索引: pk_order + idx_status + idx_userid (标签自动创建)")
	passed++

	// ============================================================
	// 3. 批量插入 Employee (50万行)
	// ============================================================
	section("3. 批量插入 Employee (50万行)")

	depts := []string{"Engineering", "Sales", "HR", "Finance", "Operations", "Marketing", "Support", "Legal"}
	t0 := time.Now()
	for i := 1; i <= EMPLOYEE_N; i++ {
		err := empTbl.Insert(&Employee{
			Name:  fmt.Sprintf("Emp%07d", i),
			Age:   int64(20 + (i % 50)),
			Score: int64(i % 1000),
			Dept:  depts[i%len(depts)],
		})
		if err != nil {
			fmt.Printf("  ❌ Insert(%d): %v\n", i, err)
			failed++
			break
		}
		if i%100000 == 0 {
			fmt.Printf("  [%d行] 耗时: %.1fs | 磁盘: %s | %s\n", i, time.Since(t0).Seconds(), fileSize(), memStats())
		}
	}
	insertTime := time.Since(t0)
	fmt.Printf("  ✅ 插入 %d 行耗时: %.2fs (%.0f rows/s)\n", EMPLOYEE_N, insertTime.Seconds(), float64(EMPLOYEE_N)/insertTime.Seconds())
	passed++

	// ============================================================
	// 4. Commit (WAL 持久化)
	// ============================================================
	section("4. Commit (WAL 持久化)")

	t1 := time.Now()
	err = db.Commit()
	commitTime := time.Since(t1)
	check("DB.Commit()", err == nil)
	fmt.Printf("  Commit 耗时: %.3fs | 磁盘: %s | %s\n", commitTime.Seconds(), fileSize(), memStats())

	// ============================================================
	// 5. 批量插入 Order (10万行)
	// ============================================================
	section("5. 批量插入 Order (10万行)")

	t2 := time.Now()
	for i := 1; i <= ORDER_N; i++ {
		orderTbl.Insert(&Order{
			UserID: int64(1 + i%EMPLOYEE_N),
			Amount: int64(100 + i*10),
			Status: int64(i % 5),
		})
	}
	fmt.Printf("  ✅ 插入 %d 行 Order 耗时: %.2fs (%.0f rows/s)\n", ORDER_N, time.Since(t2).Seconds(), float64(ORDER_N)/time.Since(t2).Seconds())
	passed++

	err = db.Commit()
	check("Order Commit", err == nil)
	fmt.Printf("  磁盘: %s | %s\n", fileSize(), memStats())

	// ============================================================
	// 6. 主键精确查找 (性能)
	// ============================================================
	section("6. 主键精确查找 (性能)")

	samples := []uint64{1, 1000, 50000, 100000, 250000, 499999}
	t3 := time.Now()
	for _, id := range samples {
		u, err := empTbl.Find(id)
		if err != nil || u == nil {
			fmt.Printf("  ❌ Find(%d) 失败\n", id)
			failed++
		}
	}
	findTime := time.Since(t3)
	fmt.Printf("  ✅ %d 次主键查找 耗时: %.4fs (%.0f finds/s)\n", len(samples), findTime.Seconds(), float64(len(samples))/findTime.Seconds())
	passed++

	// ============================================================
	// 7. 条件查询 (二级索引)
	// ============================================================
	section("7. 条件查询 (二级索引)")

	// Age = 25
	t4 := time.Now()
	users, err := empTbl.Where("Age", "=", int64(25)).Find()
	q1Time := time.Since(t4)
	if err == nil {
		fmt.Printf("  ✅ WHERE Age=25: %d 行, 耗时: %.3fs\n", len(users), q1Time.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ WHERE Age=25: %v\n", err)
		failed++
	}

	// Dept = "Engineering"
	t5 := time.Now()
	users, err = empTbl.Where("Dept", "=", "Engineering").Find()
	q2Time := time.Since(t5)
	if err == nil {
		fmt.Printf("  ✅ WHERE Dept='Engineering': %d 行, 耗时: %.3fs\n", len(users), q2Time.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ WHERE Dept: %v\n", err)
		failed++
	}

	// Score > 900
	t6 := time.Now()
	users, err = empTbl.Where("Score", ">", int64(900)).Find()
	q3Time := time.Since(t6)
	if err == nil {
		fmt.Printf("  ✅ WHERE Score>900: %d 行, 耗时: %.3fs\n", len(users), q3Time.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ WHERE Score>900: %v\n", err)
		failed++
	}

	// Age >= 30 AND Age <= 40
	t7 := time.Now()
	users, err = empTbl.Where("Age", ">=", int64(30)).Where("Age", "<=", int64(40)).Find()
	q4Time := time.Since(t7)
	if err == nil {
		fmt.Printf("  ✅ WHERE Age>=30 AND Age<=40: %d 行, 耗时: %.3fs\n", len(users), q4Time.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ WHERE Age range: %v\n", err)
		failed++
	}

	// ============================================================
	// 8. 排序 + 分页 (大数据量)
	// ============================================================
	section("8. 排序 + 分页")

	t8 := time.Now()
	top10, err := empTbl.OrderBy("Score", true).Limit(10).Find()
	sortTime := time.Since(t8)
	if err == nil && len(top10) == 10 {
		fmt.Printf("  ✅ ORDER BY Score DESC LIMIT 10, 耗时: %.3fs\n", sortTime.Seconds())
		fmt.Printf("    Top3: Score=%d, %d, %d\n", top10[0].Score, top10[1].Score, top10[2].Score)
		passed++
	} else {
		fmt.Printf("  ❌ 排序: err=%v rows=%d\n", err, len(top10))
		failed++
	}

	// 分页: OFFSET 10000 LIMIT 100
	t9 := time.Now()
	page, err := empTbl.OrderBy("Age", false).Offset(10000).Limit(100).Find()
	pageTime := time.Since(t9)
	if err == nil && len(page) == 100 {
		fmt.Printf("  ✅ ORDER BY Age ASC OFFSET 10000 LIMIT 100, 耗时: %.3fs\n", pageTime.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ 分页: err=%v rows=%d\n", err, len(page))
		failed++
	}

	// ============================================================
	// 9. 聚合查询
	// ============================================================
	section("9. 聚合查询")

	t10 := time.Now()
	cnt, err := empTbl.Query().Count()
	countTime := time.Since(t10)
	if err == nil {
		fmt.Printf("  ✅ COUNT(*): %d, 耗时: %.3fs\n", cnt, countTime.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ COUNT: %v\n", err)
		failed++
	}

	// ============================================================
	// 10. 事务: 大批量操作
	// ============================================================
	section("10. 事务: 批量插入 + 提交")

	txn := db.BeginTransaction()
	txnN := 10000
	t11 := time.Now()
	for i := EMPLOYEE_N + 1; i <= EMPLOYEE_N+txnN; i++ {
		empTbl.Insert(&Employee{
			Name:  fmt.Sprintf("Txn%07d", i),
			Age:   int64(25 + i%20),
			Score: int64(500 + i%500),
			Dept:  depts[i%len(depts)],
		})
	}
	err = txn.Commit()
	txnTime := time.Since(t11)
	check(fmt.Sprintf("事务插入 %d 行 + Commit, 耗时: %.2fs", txnN, txnTime.Seconds()), err == nil)

	// 验证 (查找事务范围内的行, 验证事务数据持久化)
	u, _ := empTbl.Find(uint64(EMPLOYEE_N + 1))
	if u != nil {
		fmt.Printf("  ✅ 验证: ID=%d 存在 (Name=%s)\n", EMPLOYEE_N+1, u.Name)
		passed++
	} else {
		fmt.Println("  ❌ 验证: 事务数据不存在")
		failed++
	}

	// ============================================================
	// 11. 事务: 回滚
	// ============================================================
	section("11. 事务: 回滚")

	txn2 := db.BeginTransaction()
	rbN := 5000
	for i := EMPLOYEE_N + txnN + 1; i <= EMPLOYEE_N+txnN+rbN; i++ {
		empTbl.Insert(&Employee{
			Name:  fmt.Sprintf("Rb%07d", i),
			Age:   int64(30),
			Score: int64(0),
			Dept:  "ROLLBACK",
		})
	}
	fmt.Printf("  事务内插入 %d 行\n", rbN)
	err = txn2.Rollback()
	check("Transaction.Rollback()", err == nil)

	// ============================================================
	// 12. 更新 + 删除 (大批量 Scan + Update/Delete)
	// ============================================================
	section("12. 更新 + 删除 (大批量)")

	// 更新: Score > 950 的设为 9999
	t12 := time.Now()
	updateCount := 0
	var updatedIDs []uint64
	empTbl.Where("Score", ">", int64(950)).Scan(func(u *Employee) bool {
		updatedIDs = append(updatedIDs, u.ID)
		u.Score = 9999
		empTbl.Update(u)
		updateCount++
		return true
	})
	updateTime := time.Since(t12)
	fmt.Printf("  ✅ 更新 Score>950 → 9999: %d 行, 耗时: %.3fs\n", updateCount, updateTime.Seconds())
	passed++

	// 删除: Dept = "Legal" 的前100行
	deleteCount := 0
	var firstDeletedID uint64
	deletedSet := make(map[uint64]bool)
	t13 := time.Now()
	empTbl.Where("Dept", "=", "Legal").Limit(100).Scan(func(u *Employee) bool {
		deletedSet[u.ID] = true
		if firstDeletedID == 0 {
			firstDeletedID = u.ID
		}
		empTbl.Delete(u.ID)
		deleteCount++
		return true
	})
	deleteTime := time.Since(t13)
	fmt.Printf("  ✅ 删除 Dept='Legal' 前100行: %d 行, 耗时: %.3fs\n", deleteCount, deleteTime.Seconds())
	passed++

	// 选取一个"确实被更新且未被删除"的行, 用于验证更新持久化
	var verifyUpdatedID uint64
	for _, id := range updatedIDs {
		if !deletedSet[id] {
			verifyUpdatedID = id
			break
		}
	}

	// 验证删除 (检查确实被删除的那一行已不存在)
	_, err = empTbl.Find(firstDeletedID)
	check(fmt.Sprintf("验证 Delete(%d) 成功", firstDeletedID), err != nil)

	err = db.Commit()
	check("更新+删除后 Commit", err == nil)

	// ============================================================
	// 13. 关闭 + 重开验证持久化
	// ============================================================
	section("13. 关闭 + 重开验证持久化")

	fmt.Printf("  关闭前磁盘: %s\n", fileSize())
	err = db.Close()
	check("DB.Close()", err == nil)

	db, err = DB.Open(dbPath)
	check("重新打开数据库", err == nil)
	empTbl, _ = DB.UseTable[Employee](db, "employee")

	// 验证数据完整性:
	//   基础 50万行 + 事务提交 1万行 - 删除 deleteCount 行
	//   (回滚的 5000 行不应存在, 事务行必须持久化)
	expected := int64(EMPLOYEE_N + txnN - deleteCount)
	cnt, _ = empTbl.Query().Count()
	if cnt == expected {
		fmt.Printf("  ✅ 重开后 COUNT: %d (期望 %d, 事务已持久化)\n", cnt, expected)
		passed++
	} else {
		fmt.Printf("  ❌ 重开后 COUNT: %d (期望 %d)\n", cnt, expected)
		failed++
	}

	// 验证更新持久化 (使用确实被更新且未被删除的行)
	u, _ = empTbl.Find(verifyUpdatedID)
	if verifyUpdatedID != 0 && u != nil && u.Score == 9999 {
		fmt.Printf("  ✅ 重开后 Find(%d): Score=%d (更新持久化)\n", verifyUpdatedID, u.Score)
		passed++
	} else if u == nil {
		fmt.Printf("  ❌ 重开后 Find(%d) 不存在\n", verifyUpdatedID)
		failed++
	} else {
		fmt.Printf("  ❌ 重开后 Find(%d): Score=%d (期望 9999)\n", verifyUpdatedID, u.Score)
		failed++
	}

	// 验证回滚数据不存在
	_, err = empTbl.Find(uint64(EMPLOYEE_N + txnN + 1))
	check("重开后回滚数据不存在", err != nil)

	fmt.Printf("  重开后磁盘: %s | %s\n", fileSize(), memStats())

	// ============================================================
	// 14. 全量扫描验证
	// ============================================================
	section("14. 全量扫描验证")

	scanCount := 0
	t14 := time.Now()
	err = empTbl.Query().Scan(func(u *Employee) bool {
		scanCount++
		return true
	})
	scanTime := time.Since(t14)
	if err == nil {
		fmt.Printf("  ✅ 全量扫描: %d 行, 耗时: %.2fs (%.0f rows/s)\n",
			scanCount, scanTime.Seconds(), float64(scanCount)/scanTime.Seconds())
		passed++
	} else {
		fmt.Printf("  ❌ 全量扫描: %v\n", err)
		failed++
	}

	// ============================================================
	// 15. Order 表验证
	// ============================================================
	section("15. Order 表验证")

	orderTbl, _ = DB.UseTable[Order](db, "order")
	cnt, _ = orderTbl.Query().Count()
	if cnt == int64(ORDER_N) {
		fmt.Printf("  ✅ Order COUNT: %d\n", cnt)
		passed++
	} else {
		fmt.Printf("  ❌ Order COUNT: %d (期望 %d)\n", cnt, ORDER_N)
		failed++
	}

	orders, _ := orderTbl.Where("Status", "=", int64(0)).Find()
	fmt.Printf("  ✅ Order WHERE Status=0: %d 行\n", len(orders))
	passed++

	// ============================================================
	// 最终汇总
	// ============================================================
	fmt.Println()
	fmt.Println(strings.Repeat("=", 60))
	fmt.Printf("  测试结果: %d 通过, %d 失败\n", passed, failed)
	fmt.Printf("  数据量: Employee=%d + Order=%d\n", EMPLOYEE_N, ORDER_N)
	fmt.Printf("  磁盘大小: %s\n", fileSize())
	fmt.Printf("  %s\n", memStats())
	if failed == 0 {
		fmt.Println("  🎉 全部通过!")
	} else {
		fmt.Printf("  ⚠️  %d 项失败\n", failed)
	}
	fmt.Println(strings.Repeat("=", 60))

	db.Close()
	// [KEEP-DB] 保留数据库文件供 Analyzer 分析, 屏蔽运行结束后的清理删除
	// os.Remove(dbPath)
	// os.Remove(dbPath + ".wal")
}