TsDB2/demo/hash_bench/main.go

// Hash 索引增删改查性能基准
//
// 目标: 在不同数据量 (1 万 ~ 20 万行) 下, 测量 Hash 二级索引的
//   1. 插入吞吐 (行/秒) 与提交耗时
//   2. 哈希索引等值查询延迟 (单条命中, Limit=1)
//   3. 哈希索引全匹配查询延迟 (统计同一 key 的全部重复行)
//   4. 主键回表查询延迟 (B+ 树)
//   5. 更新吞吐 (改 Age → 触发哈希索引 删旧键 + 插新键)
//   6. 删除吞吐 (触发哈希索引级联删除 + 桶合并)
//
// 说明: Age 取值空间 keySpace=1000, 平均每个 key 有 N/1000 行重复,
//       用于模拟真实的重复键场景 (哈希溢出链)。
package main

import (
	"fmt"
	"math/rand"
	"os"
	"slices"
	"strconv"
	"time"

	"git.ycl.pub/yl32940/TsDB2/DB"
)

type HB struct {
	DB.BaseModel
	Name  string `vrtx:"maxlen:32"`
	Age   int64  `vrtx:"index:idx_age,type:hash"`
	Score int64
	Dept  string `vrtx:"maxlen:16"`
}

const keySpace = 1000 // Age 取值空间 → 平均 N/keySpace 行重复

// --pure 模式: 关闭脏页阈值自动刷盘, 用于隔离索引自身开销
var noAutoFlush bool

type result struct {
	rows       int
	insRate    float64 // 行/秒
	commitMs   float64
	hashAvgUs  float64
	hashP50Us  float64
	hashP99Us  float64
	countAvgMs float64
	countP99Ms float64
	pkAvgUs    float64
	pkP50Us    float64
	pkP99Us    float64
	updRate    float64
	delRate    float64
	sizeMB     float64
	remain     int64
}

func pct(durs []time.Duration, p int) time.Duration {
	if len(durs) == 0 {
		return 0
	}
	idx := len(durs) * p / 100
	if idx >= len(durs) {
		idx = len(durs) - 1
	}
	return durs[idx]
}

func summarize(durs []time.Duration) (avg, p50, p99 time.Duration) {
	if len(durs) == 0 {
		return
	}
	slices.Sort(durs)
	var sum time.Duration
	for _, d := range durs {
		sum += d
	}
	avg = sum / time.Duration(len(durs))
	p50 = pct(durs, 50)
	p99 = pct(durs, 99)
	return
}

// runBatched 把 total 次操作切成 batch 大小的批次分别计时,
// 返回每批的"单次均摊"耗时。用于规避 Windows time.Now() 约 1ms 的
// 时钟粒度, 使微秒级延迟测量稳定可靠。
func runBatched(total, batch int, op func()) []time.Duration {
	n := total / batch
	if n == 0 {
		n = 1
	}
	per := make([]time.Duration, 0, n)
	for b := 0; b < n; b++ {
		s := time.Now()
		for k := 0; k < batch; k++ {
			op()
		}
		per = append(per, time.Since(s)/time.Duration(batch))
	}
	return per
}

func runVolume(rows int) result {
	rng := rand.New(rand.NewSource(int64(rows)))
	path := fmt.Sprintf("hash_bench_%d.db", rows)
	_ = os.Remove(path)

	db, err := DB.Open(path)
	if err != nil {
		panic(err)
	}
	db.SetAutoCommitDelay(0)
	db.SetBufferPoolCapacity(16384)
	if noAutoFlush {
		// 关闭脏页阈值自动 Commit, 隔离 "索引自身开销" 与 "刷盘策略开销"
		db.SetDirtyThreshold(1 << 30)
	}
	tbl, err := DB.UseTable[HB](db, "hb")
	if err != nil {
		panic(err)
	}

	ages := make([]int64, rows+1)

	// ---------------- 插入 ----------------
	t0 := time.Now()
	for i := 1; i <= rows; i++ {
		a := int64(rng.Intn(keySpace))
		ages[i] = a
		if err := tbl.Insert(&HB{
			Name:  fmt.Sprintf("N%07d", i),
			Age:   a,
			Score: int64(i % 500),
			Dept:  fmt.Sprintf("D%d", i%20),
		}); err != nil {
			panic(err)
		}
	}
	insDur := time.Since(t0)
	tc := time.Now()
	if err := db.Commit(); err != nil {
		panic(err)
	}
	commitDur := time.Since(tc)

	// ---------------- 哈希索引: 等值单条查询 (批量计时) ----------------
	hashDurs := runBatched(20000, 1000, func() {
		a := int64(rng.Intn(keySpace))
		_, _ = tbl.Where("Age", "=", a).First()
	})

	// ---------------- 哈希索引: 全匹配统计 ----------------
	countDurs := runBatched(500, 20, func() {
		a := int64(rng.Intn(keySpace))
		_, _ = tbl.Where("Age", "=", a).Count()
	})

	// ---------------- 主键回表 ----------------
	pkDurs := runBatched(20000, 1000, func() {
		id := uint64(rng.Intn(rows) + 1)
		_, _ = tbl.Find(id)
	})

	// ---------------- 更新 (改 Age, 触发哈希索引重排) ----------------
	updN := rows
	if updN > 20000 {
		updN = 20000
	}
	tu := time.Now()
	for j := 1; j <= updN; j++ {
		newAge := (ages[j] + 1) % keySpace
		ages[j] = newAge
		if err := tbl.Update(&HB{
			BaseModel: DB.BaseModel{ID: uint64(j)},
			Name:      fmt.Sprintf("U%07d", j),
			Age:       newAge,
			Score:     int64(j % 1000),
			Dept:      fmt.Sprintf("D%d", j%20),
		}); err != nil {
			panic(err)
		}
	}
	updDur := time.Since(tu)
	if err := db.Commit(); err != nil {
		panic(err)
	}

	// ---------------- 删除 (级联删哈希索引) ----------------
	delN := rows
	if delN > 20000 {
		delN = 20000
	}
	td := time.Now()
	for j := 1; j <= delN; j++ {
		if err := tbl.Delete(uint64(j)); err != nil {
			panic(err)
		}
	}
	delDur := time.Since(td)
	if err := db.Commit(); err != nil {
		panic(err)
	}

	remain, _ := tbl.Count()

	if err := db.Close(); err != nil {
		panic(err)
	}
	fi, _ := os.Stat(path)
	sizeMB := 0.0
	if fi != nil {
		sizeMB = float64(fi.Size()) / 1024 / 1024
	}
	// [KEEP-DB] 保留数据库文件供 Analyzer 分析, 屏蔽运行结束后的清理删除
	// _ = os.Remove(path)

	hAvg, hP50, hP99 := summarize(hashDurs)
	cAvg, _, cP99 := summarize(countDurs)
	pAvg, pP50, pP99 := summarize(pkDurs)

	return result{
		rows:       rows,
		insRate:    float64(rows) / insDur.Seconds(),
		commitMs:   float64(commitDur.Microseconds()) / 1000,
		hashAvgUs:  float64(hAvg.Microseconds()),
		hashP50Us:  float64(hP50.Microseconds()),
		hashP99Us:  float64(hP99.Microseconds()),
		countAvgMs: float64(cAvg.Microseconds()) / 1000,
		countP99Ms: float64(cP99.Microseconds()) / 1000,
		pkAvgUs:    float64(pAvg.Microseconds()),
		pkP50Us:    float64(pP50.Microseconds()),
		pkP99Us:    float64(pP99.Microseconds()),
		updRate:    float64(updN) / updDur.Seconds(),
		delRate:    float64(delN) / delDur.Seconds(),
		sizeMB:     sizeMB,
		remain:     remain,
	}
}

func main() {
	volumes := []int{10000, 50000, 100000, 200000, 400000}
	for _, a := range os.Args[1:] {
		if a == "--pure" {
			noAutoFlush = true
			continue
		}
		if n, err := strconv.Atoi(a); err == nil && n > 0 {
			volumes = []int{n} // 指定单一数据量
		}
	}

	fmt.Println("============================================================")
	fmt.Println("  Hash 索引 (线性哈希) 增删改查性能基准")
	fmt.Printf("  Age 取值空间=%d (平均每 key 重复 N/%d 行)\n", keySpace, keySpace)
	if noAutoFlush {
		fmt.Println("  模式: --pure (关闭脏页阈值自动刷盘, 仅手动 Commit)")
	} else {
		fmt.Println("  模式: 默认策略 (脏页阈值 500 页, 超限自动 Commit)")
	}
	fmt.Println("============================================================")

	results := make([]result, 0, len(volumes))
	for _, n := range volumes {
		fmt.Printf("\n>>> 运行数据量 N=%d ...\n", n)
		results = append(results, runVolume(n))
	}

	// ---------------- 吞吐总览 ----------------
	fmt.Println("\n============================================================")
	fmt.Println("  吞吐总览")
	fmt.Println("============================================================")
	fmt.Printf("%-10s %14s %12s %14s %14s %10s\n",
		"数据量", "插入(行/秒)", "提交(ms)", "更新(行/秒)", "删除(行/秒)", "文件(MB)")
	for _, r := range results {
		fmt.Printf("%-10d %14.0f %12.1f %14.0f %14.0f %10.2f\n",
			r.rows, r.insRate, r.commitMs, r.updRate, r.delRate, r.sizeMB)
	}

	// ---------------- 查询延迟总览 ----------------
	fmt.Println("\n============================================================")
	fmt.Println("  查询延迟 (单位: 微秒 µs / 毫秒 ms)")
	fmt.Println("============================================================")
	fmt.Printf("%-10s %26s %26s %18s\n",
		"数据量", "哈希等值查询(Limit=1)", "主键回表(Find)", "哈希全匹配(Count)")
	fmt.Printf("%-10s %10s %8s %8s %10s %8s %8s %10s %8s\n",
		"", "avg", "p50", "p99", "avg", "p50", "p99", "avg(ms)", "p99(ms)")
	for _, r := range results {
		fmt.Printf("%-10d %10.1f %8.1f %8.1f %10.1f %8.1f %8.1f %10.2f %8.2f\n",
			r.rows,
			r.hashAvgUs, r.hashP50Us, r.hashP99Us,
			r.pkAvgUs, r.pkP50Us, r.pkP99Us,
			r.countAvgMs, r.countP99Ms)
	}

	// ---------------- 查询吞吐 (QPS) ----------------
	fmt.Println("\n============================================================")
	fmt.Println("  查询吞吐 (QPS = 1000/avg µs, 按均摊延迟换算)")
	fmt.Println("============================================================")
	fmt.Printf("%-10s %18s %18s %18s\n", "数据量", "哈希等值查询", "主键回表", "哈希全匹配")
	for _, r := range results {
		hQPS, pQPS, cQPS := 0.0, 0.0, 0.0
		if r.hashAvgUs > 0 {
			hQPS = 1e6 / r.hashAvgUs
		}
		if r.pkAvgUs > 0 {
			pQPS = 1e6 / r.pkAvgUs
		}
		if r.countAvgMs > 0 {
			cQPS = 1000 / r.countAvgMs
		}
		fmt.Printf("%-10d %18.0f %18.0f %18.0f\n", r.rows, hQPS, pQPS, cQPS)
	}

	fmt.Println("\n注: 每档数据量均执行 插入 N 行 / 2万次哈希查询 / 500次全匹配 /")
	fmt.Println("    2万次主键查询 / 2万次更新 / 2万次删除。删除后剩余行数校验一致。")
	fmt.Println("    点查询延迟为批量均摊值 (每批 1000 次), 已规避系统时钟粒度影响。")
}