// ============================================================================
// TsDB2 千万级功能 + 性能测试
//
// 覆盖:
//   P2  千万级写入 (分档进度 / 分档 Commit / 磁盘与内存曲线)
//   P3  查询性能 (主键点查 / 二级索引等值 / 范围 / 分页 / 排序 / 投影 / 聚合)
//   P4  多表      (部门维度表 + 订单表 + 索引关联计数校验)
//   P5  更新      (单行全列重写 / core 批量更新 / 索引列更新后索引同步)
//   P6  删除      (逐行删 10 万行 / 批量删除全表重建) + 全量一致性校验
//   P7  Schema 变更 (ADD COLUMN / DROP COLUMN / 改列名) + 数据正确性校验
//   P8  引擎语义探针 (独立库, 不污染主库)
//   P9  关库重开后的持久化与 schema 校验
//   P10 汇总 (性能总表 / 校验总表 / 数据问题清单)
//
// 规模可用环境变量调整 (默认千万级):
//   TSDB_BENCH_N       主表行数    默认 10000000 (向下取整到 1000 的倍数)
//   TSDB_BENCH_ORDERS  订单表行数  默认 1000000
//   TSDB_BENCH_DB      主库文件名  默认 bench_10m.db (相对当前工作目录)
//
// 运行: cd demo/bench_10m; go run .
// 注意: 启动时会删除同前缀旧库文件; 运行结束后库文件保留, 可交给 analyzer 审计。
// ============================================================================
package main

import (
	"encoding/binary"
	"fmt"
	"os"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"time"

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

// ============================== 规模与派生参数 ==============================
const (
	deptCount  = 100
	scoreMod   = 1000
	ageBase    = 20
	ageMod     = 50
	bulkAgeAt  = 500
	bulkAgeTo  = 99
	bulkDeptAt = 321
)

const bulkDeptTo = "Dept099"

var (
	nEmp      int64
	nOrder    int64
	nJoinMod  int64
	delCount  int64
	rowUpdCnt int64
	rowUpdAt  int64
	delFrom   int64
	delTo     int64
	sampleNo  int64 // P7/P9 抽样行号, 随规模派生, 保证存在且不落在更新/删除区间
	extraRows int64 = 3

	dbPath    string
	probePath string
)

var (
	perfs    []perfRow
	checks   []checkRow
	problems []problemRow
)

// ============================== 数据公式 ==============================
func empName(no int64) string { return fmt.Sprintf("Emp%08d", no) }

func expName(no int64) string {
	if no >= rowUpdAt && no < rowUpdAt+rowUpdCnt {
		return "UP" + empName(no)
	}
	return empName(no)
}

func expAge(no int64) int64 {
	if no%scoreMod == bulkAgeAt {
		return bulkAgeTo
	}
	return ageBase + no%ageMod
}

func expScore(no int64) int64 { return no % scoreMod }

func expDept(no int64) string {
	if no%scoreMod == bulkDeptAt {
		return bulkDeptTo
	}
	return fmt.Sprintf("Dept%03d", no%deptCount)
}

func expSalary(no int64) float64 { return float64(no % 100000) }

func expActive(no int64) bool { return no%2 == 0 }

func expActiveByte(no int64) byte {
	if expActive(no) {
		return 1
	}
	return 0
}

func pkIDOf(no int64) uint64 {
	if no <= nEmp {
		return uint64(no)
	}
	return uint64(nEmp-delCount + (no - nEmp))
}

func isDeletedNo(no int64) bool { return no >= delFrom && no <= delTo }

func expectRows() int64 { return nEmp - delCount + extraRows }

func expectSalarySum(n int64) float64 {
	full := n / 100000
	rem := n % 100000
	return float64(full)*4999950000.0 + float64(rem)*float64(rem+1)/2
}

// ============================== 通用小工具 ==============================
func must(err error) {
	if err != nil {
		fmt.Printf("\n[FATAL] %v\n", err)
		os.Exit(1)
	}
}

// softMust 供探针使用: 出错时记录为数据问题并返回 true, 不终止整个进程。
// 探针的目的是"发现并报告引擎语义/缺陷", 单步失败不应让整场测试无结果。
func softMust(stage string, err error) bool {
	if err == nil {
		return false
	}
	fmt.Printf("    [探针错误] %s → %v\n", stage, err)
	problem("探针", stage, fmt.Sprintf("%v", err))
	return true
}

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

func sub(title string) { fmt.Printf("\n--- %s ---\n", title) }

func envInt64(key string, def int64) int64 {
	if s := os.Getenv(key); s != "" {
		if v, err := strconv.ParseInt(s, 10, 64); err == nil && v > 0 {
			return v
		}
	}
	return def
}

func human(v float64) string {
	switch {
	case v >= 1e6:
		return fmt.Sprintf("%.2fM", v/1e6)
	case v >= 1e3:
		return fmt.Sprintf("%.2fk", v/1e3)
	default:
		return fmt.Sprintf("%.0f", v)
	}
}

func mbOf(v uint64) float64 { return float64(v) / (1 << 20) }

func memLine() string {
	var m runtime.MemStats
	runtime.ReadMemStats(&m)
	return fmt.Sprintf("GoAlloc=%.0fMB Sys=%.0fMB HeapInuse=%.0fMB NumGC=%d",
		mbOf(m.Alloc), mbOf(m.Sys), mbOf(m.HeapInuse), m.NumGC)
}

func dbSize(path string) int64 {
	dir := filepath.Dir(path)
	stem := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
	entries, err := os.ReadDir(dir)
	if err != nil {
		return 0
	}
	var total int64
	for _, e := range entries {
		if e.IsDir() || !strings.HasPrefix(e.Name(), stem) {
			continue
		}
		if fi, err := e.Info(); err == nil {
			total += fi.Size()
		}
	}
	return total
}

func cleanOld(path string) {
	dir := filepath.Dir(path)
	stem := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
	entries, err := os.ReadDir(dir)
	if err != nil {
		return
	}
	for _, e := range entries {
		if !e.IsDir() && strings.HasPrefix(e.Name(), stem) {
			p := filepath.Join(dir, e.Name())
			if err := os.Remove(p); err == nil {
				fmt.Printf("  [清理旧库] %s\n", p)
			}
		}
	}
}

func pkKey(id uint64) []byte {
	b := make([]byte, 8)
	binary.BigEndian.PutUint64(b, id)
	return b
}

func contains(xs []string, s string) bool {
	for _, x := range xs {
		if x == s {
			return true
		}
	}
	return false
}

func dumpVal(v interface{}) string {
	if b, ok := v.([]byte); ok {
		return strings.TrimRight(string(b), "\x00")
	}
	return fmt.Sprint(v)
}

func dumpRow(row []types.FieldValue) string {
	parts := make([]string, 0, len(row))
	for _, fv := range row {
		parts = append(parts, fmt.Sprintf("%s=%s", string(fv.FieldName), dumpVal(fv.Value)))
	}
	return strings.Join(parts, " ")
}

func rowMap(row []types.FieldValue) map[string]interface{} {
	m := make(map[string]interface{}, len(row))
	for _, fv := range row {
		m[string(fv.FieldName)] = fv.Value
	}
	return m
}

func rawEqual(got, want map[string]interface{}) (bool, string) {
	for k, wv := range want {
		gv, ok := got[k]
		if !ok {
			return false, "缺少列 " + k
		}
		if dumpVal(gv) != dumpVal(wv) {
			return false, fmt.Sprintf("%s 期望=%s 实际=%s", k, dumpVal(wv), dumpVal(gv))
		}
	}
	return true, ""
}

func fieldIndexByName(t *core.Table, name string) int {
	for i, f := range t.GetFields() {
		if string(f.FieldName) == name {
			return i
		}
	}
	return -1
}

func coreRow(t *core.Table, pkIdx string, pk uint64) map[string]interface{} {
	row, err := t.SearchByIndex(pkIdx, pkKey(pk))
	if err != nil {
		return nil
	}
	return rowMap(row)
}

func sampleIDs(count, max int64) []uint64 {
	ids := make([]uint64, 0, count)
	x := uint64(0x9E3779B97F4A7C15)
	for int64(len(ids)) < count {
		x ^= x << 13
		x ^= x >> 7
		x ^= x << 17
		ids = append(ids, uint64(x%uint64(max))+1)
	}
	return ids
}

// ============================== 报告设施 ==============================
type perfRow struct {
	Phase, Op string
	Rows      int64
	Cost      time.Duration
	Note      string
}

type checkRow struct {
	Name, Expect, Actual string
	Pass                 bool
}

type problemRow struct {
	Level string
	Title string
	Detail string
}

func perf(ph, op string, rows int64, d time.Duration) {
	perfs = append(perfs, perfRow{ph, op, rows, d, ""})
	rate := "-"
	if d > 0 && rows > 0 {
		rate = human(float64(rows)/d.Seconds()) + "/s"
	}
	fmt.Printf("  %-52s 行数=%-9d 耗时=%-11s %s\n", op, rows, d.Round(time.Millisecond), rate)
}

func chk(name string, pass bool, expect, actual string) bool {
	checks = append(checks, checkRow{name, expect, actual, pass})
	if !pass {
		fmt.Printf("  [FAIL] %s | 期望=%s 实际=%s\n", name, expect, actual)
	}
	return pass
}

func problem(level, title, detail string) {
	problems = append(problems, problemRow{level, title, detail})
	fmt.Printf("  [%s] %s: %s\n", level, title, detail)
}

// ============================== 表结构 ==============================
type Emp struct {
	DB.BaseModel
	No     int64   `vrtx:"index:idx_emp_no"`
	Name   string  `vrtx:"maxlen:24"`
	Age    int64   `vrtx:"index:idx_emp_age"`
	Score  int64   // 故意不建索引: 用于对比"无索引条件 = 全表扫描"
	Dept   string  `vrtx:"maxlen:12,index:idx_emp_dept"`
	Salary float64 // float64 → 专用浮点列 (保序编码), 小数不会被截断
	Active bool
}

type DeptDim struct {
	DB.BaseModel
	Code  string `vrtx:"maxlen:12,index:idx_dept_code"`
	Title string `vrtx:"maxlen:32"`
	Level int64
}

type Order struct {
	DB.BaseModel
	OrderNo int64 `vrtx:"index:idx_order_no"`
	EmpNo   int64 `vrtx:"index:idx_order_empno"`
	Amount  int64
	Status  int64
}

func deptCode(i int64) string { return fmt.Sprintf("Dept%03d", i) }

// ============================== P0 环境 ==============================
func phase0() {
	phase("P0 环境与参数")
	fmt.Printf("  时间        : %s\n", time.Now().Format("2006-01-02 15:04:05"))
	fmt.Printf("  Go          : %s | CPU %d 核 (GOMAXPROCS=%d) | %s/%s\n",
		runtime.Version(), runtime.NumCPU(), runtime.GOMAXPROCS(0), runtime.GOOS, runtime.GOARCH)
	fmt.Printf("  规模        : 主表 %d 行 | 订单 %d 行 | 部门 %d 行 | 单行更新 %d 次 | 逐行删除 %d 行\n",
		nEmp, nOrder, deptCount, rowUpdCnt, delCount)
	fmt.Printf("  参数        : Age=20+No%%50 | Score=No%%1000 | Dept=Dept(No%%100) | Salary=No%%100000\n")
	fmt.Printf("  索引        : pk_emp + idx_emp_no/BTree + idx_emp_age/BTree + idx_emp_dept/BTree\n")
	fmt.Printf("  提交策略    : preset=batch(脏页阈值 500 页), 自动提交防抖禁用, 每 100 万行显式 Commit\n")
	fmt.Printf("  主库        : %s\n", dbPath)
	fmt.Printf("  探针库      : %s\n", probePath)
	fmt.Printf("  %s\n", memLine())
}

// ============================== P1 建表 ==============================
func phase1(db *DB.DB) (*DB.Table[Emp], *DB.Table[DeptDim], *DB.Table[Order], *core.Table) {
	phase("P1 建表 / 打开表 + 索引")
	t := time.Now()
	empT, err := DB.UseTable[Emp](db, "emp")
	must(err)
	deptT, err := DB.UseTable[DeptDim](db, "dept_dim")
	must(err)
	orderT, err := DB.UseTable[Order](db, "orders")
	must(err)
	empCore, ok := db.CoreDB().FindTable("emp")
	if !ok {
		must(fmt.Errorf("core 层找不到 emp 表"))
	}
	perf("P1 建表", "UseTable × 3 (含建表/建索引)", 3, time.Since(t))

	fmt.Printf("  emp 活跃列 : %v\n", empCore.GetActiveFieldNames())
	fmt.Printf("  emp 字段数 : 总 %d / 活跃 %d | TotalRows=%d\n",
		len(empCore.GetFields()), len(empCore.GetActiveFieldNames()), empCore.TotalRows())
	for _, nm := range []string{"pk_emp", "idx_emp_no", "idx_emp_age", "idx_emp_dept"} {
		if m, ok := empCore.IndexMeta().FindIndex(nm); ok {
			ids := make([]string, 0, m.FieldCnt)
			for k := 0; k < int(m.FieldCnt); k++ {
				ids = append(ids, strconv.Itoa(int(m.FieldIDs[k])))
			}
			fmt.Printf("  索引 %-14s type=%-2d 分量=%d fieldIDs=[%s]\n",
				nm, m.IndexType, m.FieldCnt, strings.Join(ids, ","))
		} else {
			fmt.Printf("  索引 %-14s 缺失!\n", nm)
		}
	}
	return empT, deptT, orderT, empCore
}

// ============================== P2 千万级写入 ==============================
func phase2InsertEmp(db *DB.DB, empT *DB.Table[Emp], empCore *core.Table) {
	phase(fmt.Sprintf("P2 千万级写入: 单行 Insert × %d (含进度 / 分档 Commit)", nEmp))
	t0 := time.Now()
	last, lastN := t0, int64(0)
	e := &Emp{}
	for i := int64(1); i <= nEmp; i++ {
		*e = Emp{
			BaseModel: DB.BaseModel{ID: 0},
			No:        i,
			Name:      empName(i),
			Age:       int64(ageBase + i%ageMod),
			Score:     expScore(i),
			Dept:      fmt.Sprintf("Dept%03d", i%deptCount),
			Salary:    expSalary(i),
			Active:    expActive(i),
		}
		if err := empT.Insert(e); err != nil {
			must(fmt.Errorf("插入第 %d 行失败: %w", i, err))
		}
		if i%200000 == 0 {
			now := time.Now()
			rate := float64(i-lastN) / now.Sub(last).Seconds()
			eta := time.Duration(float64(nEmp-i) / rate * float64(time.Second))
			fmt.Printf("  [%9d 行] 累计%7.1fs | 瞬时 %8s/s | 磁盘 %7.1fMB | ETA %6s | %s\n",
				i, now.Sub(t0).Seconds(), human(rate), float64(dbSize(dbPath))/(1<<20),
				eta.Round(time.Second), memLine())
			last, lastN = now, i
		}
		if i%1000000 == 0 {
			ct := time.Now()
			must(db.Commit())
			fmt.Printf("  [%9d 行] 显式 Commit %s | 磁盘 %.1fMB\n",
				i, time.Since(ct).Round(time.Millisecond), float64(dbSize(dbPath))/(1<<20))
		}
	}
	ct := time.Now()
	must(db.Commit())
	total := time.Since(t0)
	perf("P2 写入", fmt.Sprintf("单行 Insert × %d + 分档 Commit", nEmp), nEmp, total)
	fmt.Printf("  最终 Commit %s | 磁盘 %.1fMB | 文件 %.2fGB | %s\n",
		time.Since(ct).Round(time.Millisecond), float64(dbSize(dbPath))/(1<<20),
		float64(dbSize(dbPath))/(1<<30), memLine())

	chk("2.1 TotalRows == N", empCore.TotalRows() == uint64(nEmp),
		strconv.FormatInt(nEmp, 10), strconv.FormatUint(empCore.TotalRows(), 10))
	scanned, bad := verifyEmpInserted(empT, "写入后全量校验(插入态)", nEmp)
	chk("2.2 全表扫描行数 == N", scanned == nEmp, strconv.FormatInt(nEmp, 10), strconv.FormatInt(scanned, 10))
	chk("2.3 全表逐行公式校验 0 错", bad == 0, "0", strconv.FormatInt(bad, 10))
}

// verifyEmpORM 用 ORM 映射全表扫描并按公式逐行校验 (跳过已删除区间)
func verifyEmpORM(empT *DB.Table[Emp], label string, want int64) (int64, int64) {
	t := time.Now()
	var scanned, bad int64
	var first string
	err := empT.Select(func(e *Emp) bool {
		scanned++
		no := e.No
		if isDeletedNo(no) {
			bad++
			if first == "" {
				first = fmt.Sprintf("已删除行仍存活 No=%d", no)
			}
			return true
		}
		badCols := make([]string, 0, 4)
		if e.Name != expName(no) {
			badCols = append(badCols, "Name")
		}
		if e.Age != expAge(no) {
			badCols = append(badCols, "Age")
		}
		if e.Score != expScore(no) {
			badCols = append(badCols, "Score")
		}
		if e.Dept != expDept(no) {
			badCols = append(badCols, "Dept")
		}
		if e.Salary != expSalary(no) {
			badCols = append(badCols, "Salary")
		}
		if e.Active != expActive(no) {
			badCols = append(badCols, "Active")
		}
		if e.ID != pkIDOf(no) {
			badCols = append(badCols, fmt.Sprintf("_id(%d!=%d)", e.ID, pkIDOf(no)))
		}
		if len(badCols) > 0 {
			bad++
			if first == "" {
				first = fmt.Sprintf("No=%d 列不符: %s", no, strings.Join(badCols, ","))
			}
		}
		return true
	})
	must(err)
	perf("校验", "全表扫描 "+label, scanned, time.Since(t))
	if first != "" {
		fmt.Printf("    首个错行: %s\n", first)
	}
	if want >= 0 {
		chk("全表扫描 "+label+" 行数", scanned == want,
			strconv.FormatInt(want, 10), strconv.FormatInt(scanned, 10))
	}
	return scanned, bad
}

// verifyEmpInserted P2 专用: 只校验"刚插入时应有的原始值", 不含后续 P5 更新 / P6 删除的最终态。
// (verifyEmpORM 的公式里带有更新覆盖与删除区间, 在 P2 阶段尚不成立)
func verifyEmpInserted(empT *DB.Table[Emp], label string, want int64) (int64, int64) {
	t := time.Now()
	var scanned, bad int64
	var first string
	err := empT.Select(func(e *Emp) bool {
		scanned++
		no := e.No
		badCols := make([]string, 0, 4)
		if e.Name != empName(no) {
			badCols = append(badCols, "Name")
		}
		if e.Age != ageBase+no%ageMod {
			badCols = append(badCols, "Age")
		}
		if e.Score != no%scoreMod {
			badCols = append(badCols, "Score")
		}
		if e.Dept != fmt.Sprintf("Dept%03d", no%deptCount) {
			badCols = append(badCols, "Dept")
		}
		if e.Salary != float64(no%100000) {
			badCols = append(badCols, "Salary")
		}
		if e.Active != (no%2 == 0) {
			badCols = append(badCols, "Active")
		}
		if e.ID != uint64(no) {
			badCols = append(badCols, fmt.Sprintf("_id(%d!=%d)", e.ID, no))
		}
		if len(badCols) > 0 {
			bad++
			if first == "" {
				first = fmt.Sprintf("No=%d 列不符: %s", no, strings.Join(badCols, ","))
			}
		}
		return true
	})
	must(err)
	perf("校验", "全表扫描 "+label, scanned, time.Since(t))
	if first != "" {
		fmt.Printf("    首个错行: %s\n", first)
	}
	if want >= 0 {
		chk("全表扫描 "+label+" 行数", scanned == want,
			strconv.FormatInt(want, 10), strconv.FormatInt(scanned, 10))
	}
	return scanned, bad
}

// ============================== P3 查询 ==============================
func phase3Queries(empT *DB.Table[Emp], empCore *core.Table) {
	phase("P3 查询性能 (主键点查 / 二级索引等值 / 范围 / 分页 / 排序 / 投影 / 聚合)")
	const qn = 20000

	// 3.1 主键点查
	{
		ids := sampleIDs(qn, nEmp)
		bad := int64(0)
		t := time.Now()
		for _, id := range ids {
			e, err := empT.Find(id)
			if err != nil || e.ID != id || e.No != int64(id) || e.Name != empName(int64(id)) {
				bad++
			}
		}
		perf("P3 查询", fmt.Sprintf("主键点查 Find(pkID) × %d", qn), qn, time.Since(t))
		chk("3.1 主键点查结果全对", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))
	}
	// 3.2 二级索引等值
	{
		ids := sampleIDs(qn, nEmp)
		bad := int64(0)
		t := time.Now()
		for _, id := range ids {
			no := int64(id)
			e, err := empT.Where("No", "=", no).First()
			if err != nil || e.No != no || e.ID != uint64(no) {
				bad++
			}
		}
		perf("P3 查询", fmt.Sprintf("二级索引等值 Where(No=..) × %d", qn), qn, time.Since(t))
		chk("3.2 索引等值结果全对", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))
	}
	// 3.3 索引等值 Count (Age)
	{
		const times = 20
		want := nEmp / ageMod
		bad := int64(0)
		t := time.Now()
		for k := int64(0); k < times; k++ {
			c, err := empT.Where("Age", "=", ageBase+k).Count()
			if err != nil || c != want {
				bad++
			}
		}
		perf("P3 查询", fmt.Sprintf("索引等值 Count(Age=..) × %d (每次 %d 行)", times, want), want*times, time.Since(t))
		chk("3.3 索引等值计数正确", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))
	}
	// 3.4 索引范围 Count
	{
		want := nEmp * 5 / ageMod
		t := time.Now()
		c, err := empT.Where("Age", ">=", 30).Where("Age", "<=", 34).Count()
		perf("P3 查询", fmt.Sprintf("索引范围 Count(Age 30..34, %d 行)", want), want, time.Since(t))
		chk("3.4 索引范围计数正确", err == nil && c == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", c, err))
	}
	// 3.5 字符串索引等值 Count
	{
		d := deptCode(7)
		want := nEmp / deptCount
		t := time.Now()
		c, err := empT.Where("Dept", "=", d).Count()
		perf("P3 查询", fmt.Sprintf("字符串索引 Count(Dept=%s, %d 行)", d, want), want, time.Since(t))
		chk("3.5 字符串索引计数正确", err == nil && c == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", c, err))
	}
	// 3.6 无索引列条件 → 全表扫描
	{
		want := nEmp / scoreMod
		t := time.Now()
		c, err := empT.Where("Score", "=", 123).Count()
		perf("P3 查询", fmt.Sprintf("无索引条件 Count(Score=123, %d 行) 全表扫描", want), nEmp, time.Since(t))
		chk("3.6 无索引条件计数正确", err == nil && c == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", c, err))
	}
	// 3.7 分页
	{
		d := deptCode(7)
		rows, err := empT.Where("Dept", "=", d).Limit(20).Find()
		bad := 0
		for _, e := range rows {
			if e.Dept != d {
				bad++
			}
		}
		chk("3.7 Limit 分页", err == nil && len(rows) == 20 && bad == 0,
			"20 行且全为 "+d, fmt.Sprintf("%d 行 %d 不符 err=%v", len(rows), bad, err))

		// Offset 分页: 偏移量随规模派生, 避免小规模下 dept 总行数 < 偏移量导致取空
		deptSize := nEmp / deptCount
		off := int64(500)
		if deptSize > 0 && off > deptSize/2 {
			off = deptSize / 4
		}
		if off < 0 {
			off = 0
		}
		wantFull := int(off) + 10
		t := time.Now()
		rows2, err2 := empT.Where("Dept", "=", d).Limit(10).Offset(int(off)).Find()
		perf("P3 查询", fmt.Sprintf("Limit(10).Offset(%d)", off), 10, time.Since(t))
		full, _ := empT.Where("Dept", "=", d).Limit(wantFull).Find()
		match := len(rows2) == 10 && len(full) == wantFull
		if match {
			for k := 0; k < 10; k++ {
				if full[int(off)+k].No != rows2[k].No {
					match = false
					break
				}
			}
		}
		chk("3.7 Offset 分页与全量前缀一致", match && err2 == nil, "一致",
			fmt.Sprintf("一致=%v off=%d full=%d rows2=%d err=%v", match, off, len(full), len(rows2), err2))
	}
	// 3.8 排序
	{
		t := time.Now()
		rows, err := empT.Where("Age", ">=", 30).Where("Age", "<=", 34).OrderBy("Age").Limit(10).Find()
		perf("P3 查询", "OrderBy(索引列 Age) + Limit(10) 流式取前 N", 10, time.Since(t))
		ordered := true
		for i := 1; i < len(rows); i++ {
			if rows[i].Age < rows[i-1].Age {
				ordered = false
			}
		}
		chk("3.8 索引序排序正确", err == nil && len(rows) == 10 && ordered,
			"10 行且 Age 非降", fmt.Sprintf("%d 行 ordered=%v err=%v", len(rows), ordered, err))

		part := nEmp / deptCount
		t = time.Now()
		rows2, err2 := empT.Where("Dept", "=", deptCode(7)).OrderBy("Score").Limit(10).Find()
		perf("P3 查询", fmt.Sprintf("OrderBy(非索引列 Score) %d 行参与内存排序", part), part, time.Since(t))
		ordered2 := true
		for i := 1; i < len(rows2); i++ {
			if rows2[i].Score < rows2[i-1].Score {
				ordered2 = false
			}
		}
		chk("3.8 非索引列排序正确", err2 == nil && len(rows2) == 10 && ordered2,
			"10 行且 Score 非降", fmt.Sprintf("%d 行 ordered=%v err=%v", len(rows2), ordered2, err2))
	}
	// 3.9 投影与 FindOne
	{
		rows, err := empCore.Query().Select("No", "Age").Limit(5).Find()
		ok := err == nil && len(rows) == 5
		for _, r := range rows {
			if len(r) < 2 {
				ok = false
			}
		}
		if len(rows) > 0 {
			fmt.Printf("  Select(No,Age) 首行返回: %s\n", dumpRow(rows[0]))
		}
		chk("3.9 Select 投影返回指定列", ok, "5 行且含 No/Age",
			fmt.Sprintf("%d 行 err=%v", len(rows), err))

		one, err2 := empCore.Query().Where(core.Eq("No", int64(1), types.FieldType(0))).FindOne()
		chk("3.9 FindOne 命中", err2 == nil && len(one) > 0 && dumpVal(rowMap(one)["No"]) == "1",
			"No=1", fmt.Sprintf("%v err=%v", rowMap(one)["No"], err2))
	}
	// 3.10 聚合
	{
		t := time.Now()
		sum, err := empCore.Query().Sum("Salary")
		perf("P3 聚合", "Sum(Salary) 全表", nEmp, time.Since(t))
		want := expectSalarySum(nEmp)
		chk("3.10 Sum 正确", err == nil && sum == want,
			fmt.Sprintf("%.0f", want), fmt.Sprintf("%.0f err=%v", sum, err))

		t = time.Now()
		mx, err2 := empCore.Query().Max("No")
		perf("P3 聚合", "Max(No)", nEmp, time.Since(t))
		chk("3.10 Max 正确", err2 == nil && dumpVal(mx) == strconv.FormatInt(nEmp, 10),
			strconv.FormatInt(nEmp, 10), fmt.Sprintf("%v err=%v", mx, err2))

		mn, err3 := empCore.Query().Min("No")
		chk("3.10 Min 正确", err3 == nil && dumpVal(mn) == "1", "1", fmt.Sprintf("%v err=%v", mn, err3))
	}
	_ = coreRow
}

// ============================== P4 多表 ==============================
func phase4Multi(db *DB.DB, empT *DB.Table[Emp], deptT *DB.Table[DeptDim], orderT *DB.Table[Order],
	empCore *core.Table) (map[int64]int64, map[int64]int64) {
	phase("P4 多表: 部门维度表 + 订单表 + 索引关联计数校验")

	// 4.1 维度表
	t := time.Now()
	for i := int64(0); i < deptCount; i++ {
		d := &DeptDim{Code: deptCode(i), Title: "部门-" + deptCode(i), Level: i % 5}
		must(deptT.Insert(d))
	}
	must(db.Commit())
	perf("P4 多表", fmt.Sprintf("维度表 dept_dim 写入 %d 行", deptCount), deptCount, time.Since(t))
	c, err := deptT.Count()
	chk("4.1 维度表行数", err == nil && c == deptCount, strconv.Itoa(deptCount), fmt.Sprintf("%d err=%v", c, err))
	d7, err := deptT.Where("Code", "=", deptCode(7)).First()
	chk("4.1 维度表按 Code 索引查询", err == nil && d7 != nil && d7.Title == "部门-"+deptCode(7),
		"部门-"+deptCode(7), fmt.Sprintf("%v err=%v", d7, err))

	// 4.2 订单表
	orderCore, _ := db.CoreDB().FindTable("orders")
	t0 := time.Now()
	o := &Order{}
	for i := int64(1); i <= nOrder; i++ {
		*o = Order{OrderNo: i, EmpNo: 1 + i%nJoinMod, Amount: 100 + i%9900, Status: i % 5}
		if err := orderT.Insert(o); err != nil {
			must(fmt.Errorf("插入第 %d 条订单失败: %w", i, err))
		}
		if i%(nOrder/5) == 0 {
			rate := float64(i) / time.Since(t0).Seconds()
			fmt.Printf("  [订单 %8d 行] 累计%6.1fs | %8s/s | 磁盘 %.1fMB\n",
				i, time.Since(t0).Seconds(), human(rate), float64(dbSize(dbPath))/(1<<20))
		}
	}
	must(db.Commit())
	perf("P4 多表", fmt.Sprintf("订单表 orders 写入 %d 行", nOrder), nOrder, time.Since(t0))
	chk("4.2 订单表行数", orderCore.TotalRows() == uint64(nOrder),
		strconv.FormatInt(nOrder, 10), strconv.FormatUint(orderCore.TotalRows(), 10))

	// 4.3 关联期望 (公式推导: EmpNo = 1 + OrderNo%nJoinMod)
	orderExpect := map[int64]int64{}
	orderAfterDel := map[int64]int64{}
	for i := int64(1); i <= nOrder; i++ {
		e := 1 + i%nJoinMod
		orderExpect[e]++
		if i%5 != 0 { // P6 会删掉 Status==0 (i%5==0) 的订单
			orderAfterDel[e]++
		}
	}

	// 4.4 员工 → 订单 (走 idx_order_empno)
	sample := make([]int64, 0, 20)
	for k := int64(0); k < 20; k++ {
		if k%2 == 0 {
			sample = append(sample, 1+k*nJoinMod/40) // 落在 [1, nJoinMod] 内
		} else {
			sample = append(sample, nJoinMod+1+k) // 落在 (nJoinMod, nEmp] 内
		}
	}
	bad := int64(0)
	t = time.Now()
	for _, e := range sample {
		c, err := orderT.Where("EmpNo", "=", e).Count()
		if err != nil || c != orderExpect[e] {
			bad++
		}
	}
	perf("P4 多表", "关联计数: 订单表按 EmpNo 索引 × 20", 20, time.Since(t))
	chk("4.4 员工→订单 关联计数正确", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))

	// 4.5 部门 → 员工 (走 idx_emp_dept, 100 个部门全查)
	want := nEmp / deptCount
	bad = 0
	t = time.Now()
	for i := int64(0); i < deptCount; i++ {
		c, err := empT.Where("Dept", "=", deptCode(i)).Count()
		if err != nil || c != want {
			bad++
			fmt.Printf("    [部门计数异常] %s 期望=%d 实际=%d err=%v\n", deptCode(i), want, c, err)
		}
	}
	perf("P4 多表", fmt.Sprintf("关联计数: 员工表按 Dept 索引 × %d (每次 %d 行)", deptCount, want),
		want*deptCount, time.Since(t))
	chk("4.5 部门→员工 关联计数全对", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))
	chk("4.5 部门计数之和 == 全表行数", want*deptCount == nEmp,
		strconv.FormatInt(nEmp, 10), strconv.FormatInt(want*deptCount, 10))
	return orderExpect, orderAfterDel
}

// empSample 员工抽样: 一半落在 [1, nJoinMod] (有订单), 一半落在其后 (无订单)
func empSample() []int64 {
	s := make([]int64, 0, 20)
	for k := int64(0); k < 20; k++ {
		if k%2 == 0 {
			s = append(s, 1+k*nJoinMod/40)
		} else {
			s = append(s, nJoinMod+1+k)
		}
	}
	return s
}

// ============================== P5 更新 ==============================
func phase5Update(db *DB.DB, empT *DB.Table[Emp], empCore *core.Table) {
	phase("P5 更新 (单行整行重写 / core 批量更新 / 索引列更新后索引是否同步)")

	// 5.1 单行更新
	{
		bad := int64(0)
		t := time.Now()
		for k := int64(0); k < rowUpdCnt; k++ {
			no := rowUpdAt + k
			e, err := empT.Find(uint64(no))
			if err != nil {
				bad++
				continue
			}
			e.Name = "UP" + empName(no)
			if err := empT.Update(e); err != nil {
				bad++
			}
		}
		must(db.Commit())
		perf("P5 更新", fmt.Sprintf("单行更新 Update(整行重写) × %d", rowUpdCnt), rowUpdCnt, time.Since(t))
		chk("5.1 单行更新全部成功", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))

		e, err := empT.Find(uint64(rowUpdAt))
		chk("5.1 更新后读回正确", err == nil && e.Name == "UP"+empName(rowUpdAt) && e.No == rowUpdAt &&
			e.Age == expAge(rowUpdAt) && e.Dept == expDept(rowUpdAt) && e.Score == expScore(rowUpdAt),
			"UP"+empName(rowUpdAt),
			fmt.Sprintf("Name=%s No=%d Age=%d Dept=%s Score=%d err=%v", e.Name, e.No, e.Age, e.Dept, e.Score, err))
	}
	// 5.2 core 批量更新 (整数索引列 Age)
	{
		want := nEmp / scoreMod
		t := time.Now()
		cnt, err := empCore.Update().Set("Age", int64(bulkAgeTo)).
			Where(core.Eq("Score", int64(bulkAgeAt), types.FieldType(0))).Exec()
		perf("P5 更新", fmt.Sprintf("core 批量更新 Updater.Exec (Score=%d → Age=%d)", bulkAgeAt, bulkAgeTo),
			cnt, time.Since(t))
		chk("5.2 批量更新影响行数", err == nil && cnt == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", cnt, err))

		t = time.Now()
		c99, err2 := empT.Where("Age", "=", int64(bulkAgeTo)).Count()
		perf("P5 校验", "索引列更新后 Count(Age=99)", want, time.Since(t))
		chk("5.2 索引列更新后索引已同步", err2 == nil && c99 == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", c99, err2))

		want20 := nEmp/ageMod - nEmp/scoreMod
		c20, err3 := empT.Where("Age", "=", int64(ageBase)).Count()
		chk("5.2 原 Age 桶计数同步减少", err3 == nil && c20 == want20,
			strconv.FormatInt(want20, 10), fmt.Sprintf("%d err=%v", c20, err3))
	}
	// 5.3 core 批量更新 (字符串索引列 Dept)
	{
		want := nEmp / scoreMod
		t := time.Now()
		// 直接写原始字节: v9 起字符串列按【实际长度】存储且超宽报错 (不再静默截断),
		// 不能再按 types.GetSize(ft) (类型默认档 256B) 造 256B 值 —— 那会把 maxlen:12 的
		// 列 (16B) 喂成超宽, 整批更新失败。列宽由引擎按列定义把关。
		cnt, err := empCore.Update().Set("Dept", []byte(bulkDeptTo)).
			Where(core.Eq("Score", int64(bulkDeptAt), types.FieldType(0))).Exec()
		perf("P5 更新", fmt.Sprintf("core 批量更新字符串索引列 (Score=%d → Dept=%s)", bulkDeptAt, bulkDeptTo),
			cnt, time.Since(t))
		chk("5.3 字符串列批量更新影响行数", err == nil && cnt == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", cnt, err))

		want99 := nEmp/deptCount + nEmp/scoreMod
		c99, err2 := empT.Where("Dept", "=", bulkDeptTo).Count()
		chk("5.3 字符串索引同步 (并入 "+bulkDeptTo+")", err2 == nil && c99 == want99,
			strconv.FormatInt(want99, 10), fmt.Sprintf("%d err=%v", c99, err2))

		want21 := nEmp/deptCount - nEmp/scoreMod
		c21, err3 := empT.Where("Dept", "=", deptCode(21)).Count()
		chk("5.3 原字符串桶计数同步减少", err3 == nil && c21 == want21,
			strconv.FormatInt(want21, 10), fmt.Sprintf("%d err=%v", c21, err3))
	}
	// 5.4 负数测试: 更新不存在的主键
	{
		ghost := &Emp{BaseModel: DB.BaseModel{ID: 1 << 60}, No: 1, Name: "ghost"}
		err := empT.Update(ghost)
		chk("5.4 更新不存在的主键应报错", err != nil, "error", fmt.Sprintf("%v", err))
	}
}

// ============================== P6 删除 + 一致性校验 ==============================
func phase6Delete(db *DB.DB, empT *DB.Table[Emp], orderT *DB.Table[Order], empCore *core.Table,
	orderAfterDel map[int64]int64) {
	phase(fmt.Sprintf("P6 删除 (逐行删 %d 行 / 批量删除全表重建) + 全量一致性校验", delCount))

	// 6.1 逐行删除
	{
		t := time.Now()
		ok := int64(0)
		for id := uint64(delFrom); id <= uint64(delTo); id++ {
			if err := empT.Delete(id); err != nil {
				fmt.Printf("    [删除失败] pkID=%d err=%v\n", id, err)
			} else {
				ok++
			}
		}
		perf("P6 删除", fmt.Sprintf("逐行 Delete(pkID) × %d", delCount), ok, time.Since(t))
		chk("6.1 逐行删除全部成功", ok == delCount,
			strconv.FormatInt(delCount, 10), strconv.FormatInt(ok, 10))
		chk("6.1 TotalRows 已减少", empCore.TotalRows() == uint64(nEmp-delCount),
			strconv.FormatInt(nEmp-delCount, 10), strconv.FormatUint(empCore.TotalRows(), 10))

		_, err := empT.Find(uint64(delFrom))
		chk("6.1 已删除行 Find 应 not found", err != nil, "error", fmt.Sprintf("%v", err))

		alive := delFrom - 1
		e, err2 := empT.Find(uint64(alive))
		chk("6.1 边界外一行仍存活", err2 == nil && e.No == alive,
			fmt.Sprintf("No=%d", alive), fmt.Sprintf("%v err=%v", e, err2))
	}
	// 6.2 批量删除 orders (Status==0 无索引列 → 收集后全表重建)
	{
		orderCore, _ := db.CoreDB().FindTable("orders")
		want := nOrder / 5
		t := time.Now()
		cnt, err := orderCore.Delete().Where(core.Eq("Status", int64(0), types.FieldType(0))).Exec()
		perf("P6 删除", fmt.Sprintf("批量删除 orders (Status=0, %d 行) 触发全表重建", want), want, time.Since(t))
		chk("6.2 批量删除影响行数", err == nil && cnt == want,
			strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", cnt, err))
		chk("6.2 订单表行数已减少", orderCore.TotalRows() == uint64(nOrder-want),
			strconv.FormatInt(nOrder-want, 10), strconv.FormatUint(orderCore.TotalRows(), 10))

		bad := 0
		t2 := time.Now()
		for _, e := range empSample() {
			c, err := orderT.Where("EmpNo", "=", e).Count()
			if err != nil || c != orderAfterDel[e] {
				bad++
			}
		}
		perf("P6 校验", "批量删除后 员工→订单 关联计数 × 20", 20, time.Since(t2))
		chk("6.2 批量删除后关联计数正确", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))
	}
	// 6.3 全量一致性校验 (期望值全部由公式推导)
	{
		sub("6.3 一致性校验 (更新 + 删除之后, 期望值由公式推导)")
		scanned, bad := verifyEmpORM(empT, "删除后全量校验(公式)", nEmp-delCount)
		chk("6.3 全表逐行公式校验 0 错", bad == 0, "0", strconv.FormatInt(bad, 10))

		deptWant := map[string]int64{}
		ageWant := map[int64]int64{}
		for i := int64(1); i <= nEmp; i++ {
			if isDeletedNo(i) {
				continue
			}
			deptWant[expDept(i)]++
			ageWant[expAge(i)]++
		}
		badD := 0
		t := time.Now()
		for i := int64(0); i < deptCount; i++ {
			c, err := empT.Where("Dept", "=", deptCode(i)).Count()
			if err != nil || c != deptWant[deptCode(i)] {
				badD++
				fmt.Printf("    [部门计数异常] %s 期望=%d 实际=%d err=%v\n", deptCode(i), deptWant[deptCode(i)], c, err)
			}
		}
		perf("P6 校验", fmt.Sprintf("索引计数复核 Dept × %d", deptCount), nEmp-delCount, time.Since(t))
		chk("6.3 全部部门索引计数正确", badD == 0, "0 错", fmt.Sprintf("%d 错", badD))

		badA := 0
		for _, a := range []int64{int64(bulkAgeTo), int64(ageBase), int64(ageBase) + 1, 40, 55} {
			c, err := empT.Where("Age", "=", a).Count()
			if err != nil || c != ageWant[a] {
				badA++
				fmt.Printf("    [Age 计数异常] %d 期望=%d 实际=%d err=%v\n", a, ageWant[a], c, err)
			}
		}
		chk("6.3 抽样 Age 索引计数正确", badA == 0, "0 错", fmt.Sprintf("%d 错", badA))

		sum := int64(0)
		for _, v := range deptWant {
			sum += v
		}
		chk("6.3 部门计数之和 == 存活行数", sum == nEmp-delCount,
			strconv.FormatInt(nEmp-delCount, 10), strconv.FormatInt(sum, 10))
		_ = scanned
	}
}

// ============================== P7 Schema 变更 ==============================
func expectEmpRaw(no int64) map[string]interface{} {
	return map[string]interface{}{
		"_id":    pkIDOf(no),
		"No":     uint64(no),
		"Name":   []byte(expName(no)),
		"Age":    uint64(expAge(no)),
		"Score":  uint64(expScore(no)),
		"Dept":   []byte(expDept(no)),
		"Salary": uint64(expSalary(no)),
		"Active": expActive(no),
	}
}

func expectEmpVals(no int64, pk, level uint64) []types.FieldValue {
	return []types.FieldValue{
		{FieldName: []byte("_id"), Value: pk},
		{FieldName: []byte("No"), Value: uint64(no)},
		{FieldName: []byte("Name"), Value: []byte(expName(no))},
		{FieldName: []byte("Age"), Value: uint64(expAge(no))},
		{FieldName: []byte("Score"), Value: uint64(expScore(no))},
		{FieldName: []byte("Dept"), Value: []byte(expDept(no))},
		{FieldName: []byte("Salary"), Value: uint64(expSalary(no))},
		{FieldName: []byte("Active"), Value: expActiveByte(no)},
		{FieldName: []byte("Level"), Value: level},
	}
}

func calcDeptWant(nExtra int64) map[string]int64 {
	m := map[string]int64{}
	for i := int64(1); i <= nEmp; i++ {
		if isDeletedNo(i) {
			continue
		}
		m[expDept(i)]++
	}
	for k := int64(1); k <= nExtra; k++ {
		m[expDept(nEmp+k)]++
	}
	return m
}

// verifyEmpCoreScan 纯 core 视角全表扫描校验 (不经过 ORM 结构体映射)
func verifyEmpCoreScan(empCore *core.Table, label string, wantRows int64, wantDept bool) (int64, int64) {
	t := time.Now()
	var scanned, bad int64
	var first string
	must(empCore.Query().Scan(func(r []types.FieldValue) bool {
		scanned++
		m := rowMap(r)
		// No 列现在是 int64 (保序 Int64 编码), 旧实现按 uint64 断言会拿到零值,
		// 导致整表每行都被判错。兼容两种整数形态。
		var ino int64
		switch v := m["No"].(type) {
		case int64:
			ino = v
		case uint64:
			ino = int64(v)
		}
		if isDeletedNo(ino) {
			bad++
			if first == "" {
				first = fmt.Sprintf("已删除行仍存活 No=%d", ino)
			}
			return true
		}
		cols := make([]string, 0, 7)
		if dumpVal(m["Name"]) != expName(ino) {
			cols = append(cols, "Name")
		}
		if dumpVal(m["Age"]) != strconv.FormatInt(expAge(ino), 10) {
			cols = append(cols, "Age")
		}
		if dumpVal(m["Score"]) != strconv.FormatInt(expScore(ino), 10) {
			cols = append(cols, "Score")
		}
		if wantDept && dumpVal(m["Dept"]) != expDept(ino) {
			cols = append(cols, "Dept")
		}
		if dumpVal(m["Salary"]) != strconv.FormatUint(uint64(expSalary(ino)), 10) {
			cols = append(cols, "Salary")
		}
		if dumpVal(m["Active"]) != dumpVal(expActive(ino)) {
			cols = append(cols, "Active")
		}
		if dumpVal(m["_id"]) != strconv.FormatUint(pkIDOf(ino), 10) {
			cols = append(cols, "_id")
		}
		if len(cols) > 0 {
			bad++
			if first == "" {
				first = fmt.Sprintf("No=%d 列不符: %s", ino, strings.Join(cols, ","))
			}
		}
		return true
	}))
	perf("校验", "core 全表扫描 "+label, scanned, time.Since(t))
	if first != "" {
		fmt.Printf("    首个错行: %s\n", first)
	}
	chk("core 全表扫描 "+label+" 行数", scanned == wantRows,
		strconv.FormatInt(wantRows, 10), strconv.FormatInt(scanned, 10))
	return scanned, bad
}

func phase7Schema(db *DB.DB, empT *DB.Table[Emp], empCore *core.Table) {
	phase("P7 Schema 变更: 增列 / 删列 / 改列名 (含数据正确性校验)")

	rowsBefore := empCore.TotalRows()
	sub("7.0 变更前基线")
	fmt.Printf("  活跃列: %v\n", empCore.GetActiveFieldNames())
	fmt.Printf("  TotalRows=%d | 字段数 总%d/活跃%d\n",
		rowsBefore, len(empCore.GetFields()), len(empCore.GetActiveFieldNames()))
	row0 := coreRow(empCore, "pk_emp", uint64(sampleNo))
	fmt.Printf("  基线行 No=%d → %s\n", sampleNo, dumpRow2(row0))

	// ---- 7.1 ADD COLUMN ----
	sub("7.1 ADD COLUMN Level (8 字节新列)")
	must(empCore.AddField(types.FieldInfo{FieldType: types.FieldType_Uint64, FieldName: []byte("Level")}))
	cols := empCore.GetActiveFieldNames()
	chk("7.1 活跃列含 Level", contains(cols, "Level"), "含 Level", fmt.Sprintf("%v", cols))

	row := coreRow(empCore, "pk_emp", uint64(sampleNo))
	_, hasLevel := row["Level"]
	chk("7.1 老行不返回新增列(缺失而非零值)", !hasLevel, "无 Level", fmt.Sprintf("有=%v", hasLevel))
	eq, diff := rawEqual(row, expectEmpRaw(sampleNo))
	chk("7.1 老行其它列不受加列影响", eq, "全等", diff)

	sampled, missLevel := int64(0), int64(0)
	must(empCore.Query().Limit(2000).Scan(func(r []types.FieldValue) bool {
		sampled++
		if _, ok := rowMap(r)["Level"]; !ok {
			missLevel++
		}
		return true
	}))
	chk("7.1 抽样老行均不含 Level", missLevel == sampled,
		fmt.Sprintf("%d/%d", sampled, sampled), fmt.Sprintf("%d/%d", missLevel, sampled))

	bad := int64(0)
	newPKs := make([]uint64, 0, extraRows)
	for k := int64(1); k <= extraRows; k++ {
		no := nEmp + k
		// 这里要的是"插入一条指定 pkID 的新行", 用显式主键 API。
		// 不能再用 TotalRows()+1 去凑: 自增编号只增不减, 删除后不会回退 (见 T4 探针),
		// 靠行数现算出来的号会与它期望的 pkID 对不上。
		pk := pkIDOf(no)
		got, err := empCore.InsertRowWithID(pk, expectEmpVals(no, pk, 777))
		if err != nil || got != pk {
			bad++
			fmt.Printf("    [新行插入异常] No=%d pk=%d err=%v\n", no, pk, err)
			continue
		}
		newPKs = append(newPKs, pk)
	}
	chk("7.1 加列后新行插入成功", bad == 0, "0 错", fmt.Sprintf("%d 错", bad))
	r1 := coreRow(empCore, "pk_emp", newPKs[0])
	lv, okLv := r1["Level"]
	chk("7.1 新行返回 Level=777", okLv && dumpVal(lv) == "777", "777", dumpVal(lv))
	eq1, diff1 := rawEqual(r1, expectEmpRaw(nEmp+1))
	chk("7.1 新行其它列值正确", eq1, "全等", diff1)

	// ---- 7.2 在新增列上建索引 ----
	sub("7.2 在新增列上建索引")
	lvIdx := fieldIndexByName(empCore, "Level")
	if lvIdx < 0 {
		must(fmt.Errorf("找不到 Level 字段的 master 索引"))
	}
	_, err := empCore.CreateIndex("idx_emp_level", page.IndexTypeBTree, []byte{byte(lvIdx)})
	must(err)
	c7, err7 := empCore.Query().Where(core.Eq("Level", uint64(777), types.FieldType(0))).Count()
	perf("P7 校验", "新增列索引 Count(Level=777)", extraRows, time.Millisecond)
	chk("7.2 新增列索引能查到新行", err7 == nil && c7 == extraRows,
		strconv.FormatInt(extraRows, 10), fmt.Sprintf("%d err=%v", c7, err7))
	if err7 == nil && c7 == 0 {
		problem("重要限制", "CreateIndex 不回填存量行",
			"在已有数据的表上新建二级索引, 索引中不含任何存量行: 本例新行(3 行 Level=777)在 CreateIndex 之前已插入, 建好后 Count(Level=777) 仍为 0。只有建索引之后新插入的行才进索引; 引擎未提供公开的『重建全部二级索引』接口。")
	}
	c0, err0 := empCore.Query().Where(core.Eq("Level", uint64(0), types.FieldType(0))).Count()
	chk("7.2 老行在新增列上视为无值(不落索引)", err0 == nil && c0 == 0,
		"0", fmt.Sprintf("%d err=%v", c0, err0))

	// ---- 7.3 DROP COLUMN ----
	sub("7.3 DROP COLUMN Level (先 DropIndex 再 DropField)")
	must(empCore.DropIndex("idx_emp_level"))
	must(empCore.DropField("Level"))
	cols = empCore.GetActiveFieldNames()
	chk("7.3 活跃列不含 Level", !contains(cols, "Level"), "不含 Level", fmt.Sprintf("%v", cols))
	_, stillThere := empCore.IndexMeta().FindIndex("idx_emp_level")
	chk("7.3 索引已删除", !stillThere, "不存在", fmt.Sprintf("仍存在=%v", stillThere))

	row = coreRow(empCore, "pk_emp", uint64(sampleNo))
	eq2, diff2 := rawEqual(row, expectEmpRaw(sampleNo))
	chk("7.3 删列后老行其它列完好", eq2, "全等", diff2)
	_, hasAfterDrop := row["Level"]
	chk("7.3 删列后老行不再有 Level", !hasAfterDrop, "无 Level", fmt.Sprintf("有=%v", hasAfterDrop))

	r1 = coreRow(empCore, "pk_emp", newPKs[0])
	_, has := r1["Level"]
	eq3, diff3 := rawEqual(r1, expectEmpRaw(nEmp+1))
	chk("7.3 新行的 Level 列已不可见", !has, "无 Level", fmt.Sprintf("有=%v", has))
	chk("7.3 新行其它列完好", eq3, "全等", diff3)
	chk("7.3 删列不影响行数", empCore.TotalRows() == uint64(int64(rowsBefore)+extraRows),
		strconv.FormatInt(int64(rowsBefore)+extraRows, 10), strconv.FormatUint(empCore.TotalRows(), 10))

	// ---- 7.4 改列名 ----
	sub("7.4 改列名 (引擎无 RENAME API, 只能 DROP 旧列 + ADD 新列)")
	oldDept := dumpVal(row0["Dept"])
	// 复活软删除的列必须沿用【原类型 + 原宽度码】: master registry 的类型/参数是所有历史
	// 版本共用的行偏移依据 (见 getVersionedOffset / rowDataSizeForVersion), 换成别的类型/
	// 宽度会让已落盘的旧行整段错位, 引擎会直接拒。所以先把原定义抄下来, 7.5 用它复活。
	deptDef := empCore.Field[fieldIndexByName(empCore, "Dept")]
	must(empCore.DropIndex("idx_emp_dept"))
	must(empCore.DropField("Dept"))
	must(empCore.AddField(types.FieldInfo{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(32), FieldName: []byte("DeptName")}))
	r := coreRow(empCore, "pk_emp", uint64(sampleNo))
	_, hasOld := r["Dept"]
	_, hasNew := r["DeptName"]
	chk("7.4 改名后老列名不可读", !hasOld, "无 Dept", fmt.Sprintf("有=%v", hasOld))
	chk("7.4 改名后新列名在老行上无值", !hasNew, "无 DeptName", fmt.Sprintf("有=%v", hasNew))
	fmt.Printf("  证据: 老行 Dept 原值=%q, 改名后既读不到 Dept 也读不到 DeptName\n", oldDept)
	problem("设计约束", "引擎没有 RENAME COLUMN / RENAME TABLE API",
		"改列名只能 DROP 旧列 + ADD 新列; 老行原值不会被迁移到新列名(数据仍在行内, 只是列名不可达), 需要业务侧自行回填")

	// ---- 7.5 复活同名列 ----
	// 用 7.4 抄下来的原定义 (类型 + 宽度码) 复活; 只有"同名同类型"才允许复活,
	// 想换宽度必须改用新列名 (见 Table.AddField 的复活分支)。
	sub("7.5 重新加回同名 Dept 列 (DropField 实为软删除)")
	must(empCore.AddField(deptDef))
	r = coreRow(empCore, "pk_emp", uint64(sampleNo))
	got, okGot := r["Dept"]
	chk("7.5 同名列复活后老行数据可见", okGot, "可见", fmt.Sprintf("可见=%v", okGot))
	chk("7.5 复活值与删列前完全一致", dumpVal(got) == oldDept, oldDept, dumpVal(got))
	eq4, diff4 := rawEqual(r, expectEmpRaw(sampleNo))
	chk("7.5 复活后老行全列正确", eq4, "全等", diff4)
	problem("重要语义", "DROP COLUMN 是软删除, 数据仍留在行内; 重新 ADD 同名列会\"复活\"老行数据",
		"这解释了为什么改列名(删+加新名)后数据看似丢失, 但加回旧列名后又能读到 → 敏感数据无法靠删列来清除, 需要重建表/覆写")

	// ---- 7.6 重建索引 + 全表校验 ----
	sub("7.6 重建 Dept 索引 + core 层全表校验")
	dIdx := fieldIndexByName(empCore, "Dept")
	_, ciErr := empCore.CreateIndex("idx_emp_dept", page.IndexTypeBTree, []byte{byte(dIdx)})
	must(ciErr)
	want := calcDeptWant(extraRows)
	badD := 0
	t := time.Now()
	for i := int64(0); i < deptCount; i++ {
		c, err := empT.Where("Dept", "=", deptCode(i)).Count()
		if err != nil || c != want[deptCode(i)] {
			badD++
			fmt.Printf("    [部门计数异常] %s 期望=%d 实际=%d err=%v\n", deptCode(i), want[deptCode(i)], c, err)
		}
	}
	perf("P7 校验", fmt.Sprintf("重建索引后 索引计数复核 Dept × %d", deptCount), expectRows(), time.Since(t))
	chk("7.6 重建索引后部门计数全对", badD == 0, "0 错", fmt.Sprintf("%d 错", badD))
	if badD == int(deptCount) {
		problem("重要限制", "DROP 后重建的二级索引同样不回填存量行",
			"idx_emp_dept 删除后重新 CreateIndex, 对全部存量行按索引计数均为 0; 而紧随其后的 core 逐行全表扫描校验 0 错 → 数据本身完好, 只是索引不含存量行。")
	}

	_, badRows := verifyEmpCoreScan(empCore, "P7 变更后全量校验", expectRows(), true)
	chk("7.6 P7 变更后逐行公式校验 0 错", badRows == 0, "0", strconv.FormatInt(badRows, 10))
	_ = db
}

func dumpRow2(m map[string]interface{}) string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	// 固定顺序输出, 便于对比
	order := []string{"_id", "No", "Name", "Age", "Score", "Dept", "Salary", "Active", "Level", "DeptName"}
	parts := make([]string, 0, len(keys))
	for _, k := range order {
		if v, ok := m[k]; ok {
			parts = append(parts, fmt.Sprintf("%s=%s", k, dumpVal(v)))
		}
	}
	return strings.Join(parts, " ")
}

// ============================== P8 引擎语义探针 ==============================
func coreRows(pdb *DB.DB, name string) uint64 {
	if t, ok := pdb.CoreDB().FindTable(name); ok {
		return t.TotalRows()
	}
	return 0
}

func phase8Probes() {
	phase("P8 引擎语义探针 (独立库 " + filepath.Base(probePath) + ", 不影响主库)")
	pdb, err := DB.Open(probePath)
	must(err)
	defer func() { must(pdb.Close()) }()
	probeLayoutAssumption(pdb)
	probeDanglingIndex(pdb)
	probeExplicitPK(pdb)
	probePKReuse(pdb)
	probeFloatTrunc(pdb)
}

func probeLayoutAssumption(pdb *DB.DB) {
	sub("T1 core 建表(无 _id 列) 时, Delete/Update 如何确定 pkID")
	ct, err := pdb.CoreDB().CreateTable("probe_layout", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("No")},
		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(32), FieldName: []byte("Label")},
	})
	if softMust("CreateTable(probe_layout)", err) {
		return
	}
	_, err = ct.CreateIndex("pk_probe_layout", page.IndexTypePrimary, []byte{0})
	if softMust("CreateIndex(pk_probe_layout)", err) {
		return
	}
	for i, no := range []uint64{5, 5, 5, 6, 7} {
		_, err := ct.InsertRow([]types.FieldValue{
			{FieldName: []byte("No"), Value: no},
			{FieldName: []byte("Label"), Value: []byte(fmt.Sprintf("L%d", i+1))},
		})
		if softMust(fmt.Sprintf("InsertRow(No=%d)", no), err) {
			return
		}
	}
	one, err := ct.SearchByIndex("pk_probe_layout", pkKey(1))
	if softMust("SearchByIndex(pkID=1)", err) {
		one = nil
	}
	fmt.Printf("  表 = [No(8B), Label(32B)], 无 _id 列; 引擎反序列化出的行: %s\n", dumpRow(one))
	fmt.Printf("  数据: pkID 1→No5, 2→No5, 3→No5, 4→No6, 5→No7\n")

	cnt, errD := ct.Delete().Where(core.Eq("No", uint64(5), types.FieldType(0))).Exec()
	alive := map[uint64]uint64{}
	for pk := uint64(1); pk <= 5; pk++ {
		if r, err := ct.SearchByIndex("pk_probe_layout", pkKey(pk)); err == nil {
			if v, ok := rowMap(r)["No"].(uint64); ok {
				alive[pk] = v
			}
		}
	}
	fmt.Printf("  Delete(No==5) → cnt=%d err=%v | 剩余: ", cnt, errD)
	for pk := uint64(1); pk <= 5; pk++ {
		if v, ok := alive[pk]; ok {
			fmt.Printf("pkID%d(No=%d) ", pk, v)
		}
	}
	fmt.Printf("| TotalRows=%d\n", ct.TotalRows())

	if len(alive) == 0 && ct.TotalRows() > 0 {
		problem("严重", "core 主键索引建在非 _id 列上时, 行按 pkID 不可达",
			"表 [No(8B), Label(32B)] 建 IndexTypePrimary(field 0=No) 并插入 5 行: TotalRows=5、Query().Count()=5 数据可见, 但 SearchByIndex(pk_..., pkKey(1..5)) 全部 key not found, 只有 1 个键能命中 → 主键 B+ 树未按 pkID 建立完整条目。纯 core 表应把 _id 作为首列主键。")
	}

	aliveNo5 := int64(0)
	for _, v := range alive {
		if v == 5 {
			aliveNo5++
		}
	}
	chk("T1 批量删除影响行数应等于匹配行数 3", cnt == 3, "3", strconv.FormatInt(cnt, 10))
	chk("T1 被匹配的行(No=5)应全部删除", aliveNo5 == 0, "0 行", fmt.Sprintf("%d 行", aliveNo5))
	_, keep5 := alive[5]
	chk("T1 未被匹配的行(pkID=5,No=7)不应被删", keep5, "存活", fmt.Sprintf("存活=%v", keep5))
	if cnt != 3 || aliveNo5 != 0 || !keep5 {
		problem("严重", "core 层 Deleter.Exec 用\"行内第一个 uint64 列的值\"当 pkID",
			fmt.Sprintf("条件 No==5 匹配 pkID 1/2/3 三行, 但 Deleter 把匹配行第一个 uint64 列的值 5 当成 pkID, 于是删掉了 pkID=5 的行(其 No=7, 未匹配), 真正匹配的 3 行一行未删 (cnt=%d)", cnt))
	}

	cnt2, errU := ct.Update().Set("Label", []byte("X")).
		Where(core.Eq("No", uint64(6), types.FieldType(0))).Exec()
	fmt.Printf("  Update(No==6 → Label=X) → cnt=%d err=%v\n", cnt2, errU)
	if errU != nil {
		problem("严重", "同一假设影响 core 层 Updater.Exec",
			fmt.Sprintf("匹配行的第一个 uint64 值(6)被当成 pkID 去 UpdateRow → err=%v", errU))
	} else if cnt2 != 1 {
		problem("严重", "同一假设影响 core 层 Updater.Exec",
			fmt.Sprintf("影响行数 %d != 匹配数 1", cnt2))
	}
	fmt.Printf("  对照: ORM 表的行首列是 _id(=pkID), 因此 ORM 路径不受影响; 纯 core 表需自行保证首列语义\n")
}

func probeDanglingIndex(pdb *DB.DB) {
	sub("T2 只删列不删索引: 残留索引的行为")
	ct, err := pdb.CoreDB().CreateTable("probe_dangling", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("No")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("Tag")},
	})
	if softMust("CreateTable(probe_dangling)", err) {
		return
	}
	_, err = ct.CreateIndex("pk_probe_dangling", page.IndexTypePrimary, []byte{0})
	if softMust("CreateIndex(pk_probe_dangling)", err) {
		return
	}
	_, err = ct.CreateIndex("idx_probe_tag", page.IndexTypeBTree, []byte{1})
	if softMust("CreateIndex(idx_probe_tag)", err) {
		return
	}
	for i := uint64(1); i <= 5; i++ {
		_, err := ct.InsertRow([]types.FieldValue{
			{FieldName: []byte("No"), Value: i},
			{FieldName: []byte("Tag"), Value: i * 10},
		})
		if softMust(fmt.Sprintf("InsertRow(No=%d)", i), err) {
			return
		}
	}
	c0, err0 := ct.Query().Where(core.Eq("Tag", uint64(20), types.FieldType(0))).Count()
	fmt.Printf("  删列前 Count(Tag==20)=%d err=%v\n", c0, err0)
	if softMust("DropField(Tag)", ct.DropField("Tag")) {
		return
	}
	fmt.Printf("  删列后活跃列=%v\n", ct.GetActiveFieldNames())
	_, dangling := ct.IndexMeta().FindIndex("idx_probe_tag")
	fmt.Printf("  残留索引 idx_probe_tag 仍存在=%v\n", dangling)
	_, errQ := ct.Query().Where(core.Eq("Tag", uint64(20), types.FieldType(0))).Count()
	fmt.Printf("  删列后按残留索引查询 Tag==20 → err=%v\n", errQ)
	_, errI := ct.InsertRow([]types.FieldValue{{FieldName: []byte("No"), Value: uint64(6)}})
	fmt.Printf("  删列后插入不含 Tag 的新行 → err=%v TotalRows=%d\n", errI, ct.TotalRows())
	r6, err6 := ct.SearchByIndex("pk_probe_dangling", pkKey(6))
	fmt.Printf("  读回新行: %s err=%v\n", dumpRow(r6), err6)
	chk("T2 残留索引不影响插入(缺失列应被跳过)", errI == nil, "nil", fmt.Sprintf("%v", errI))
	if errQ != nil {
		problem("使用约束", "DropField 不会自动删除引用该列的索引",
			fmt.Sprintf("残留索引会导致按该列查询直接报错(err=%v); 正确顺序是先 DropIndex 再 DropField", errQ))
	}
}

func probeExplicitPK(pdb *DB.DB) {
	sub("T3 ORM 显式指定 _id ≠ 引擎 pkID")
	pt, err := DB.UseTable[Emp](pdb, "probe_emp")
	if softMust("UseTable(probe_emp)", err) {
		return
	}
	e := &Emp{BaseModel: DB.BaseModel{ID: 999999}, No: 1, Name: "P1", Age: 30, Score: 1, Dept: "Dept000", Salary: 1, Active: true}
	errI := pt.Insert(e)
	fmt.Printf("  Insert(ID=999999) → err=%v, 对象 ID=%d\n", errI, e.ID)
	_, errF := pt.Find(999999)
	fmt.Printf("  Find(999999) → err=%v\n", errF)
	real, errR := pt.Find(1)
	if errR == nil {
		fmt.Printf("  Find(1) → ID=%d No=%d (说明引擎分配的 pkID 是 1, 与 _id 列 999999 不一致)\n", real.ID, real.No)
	} else {
		fmt.Printf("  Find(1) → err=%v\n", errR)
	}
	chk("T3 按指定的 _id 应能查到该行", errF == nil, "查得到", fmt.Sprintf("err=%v", errF))
	if errF != nil {
		problem("风险", "ORM 显式指定的主键值会被静默忽略, 产生\"插入成功但按主键查不到\"的行",
			fmt.Sprintf("引擎 pkID 恒为 TotalRows+1; 写 ID=999999 后 Find(999999) 失败, 只有 Find(1) 能查到, 且该行 _id 列值仍是 999999 → _id 与真实 pkID 失配, 之后 Delete/批量删除会按 _id 操作到错误的行 (err=%v)", errF))
	}
}

func probePKReuse(pdb *DB.DB) {
	sub("T4 删除中间 pkID 后再插入: pkID 复用与碰撞")
	pr, err := DB.UseTable[Emp](pdb, "probe_reuse")
	if softMust("UseTable(probe_reuse)", err) {
		return
	}
	for i := int64(1); i <= 3; i++ {
		if softMust(fmt.Sprintf("Insert(No=%d)", i), pr.Insert(&Emp{No: i, Name: empName(i), Age: 20 + i, Score: i, Dept: "Dept000", Salary: float64(i), Active: true})) {
			return
		}
	}
	if softMust("Delete(pkID=2)", pr.Delete(2)) {
		return
	}
	cnt, _ := pr.Count()
	fmt.Printf("  插 3 行(pkID 1,2,3) → 删 pkID=2 → Count=%d TotalRows=%d\n", cnt, coreRows(pdb, "probe_reuse"))
	if softMust("Insert(No=99)", pr.Insert(&Emp{No: 99, Name: empName(99), Age: 99, Score: 9, Dept: "Dept000", Salary: 9, Active: true})) {
		return
	}
	type kv struct {
		ID uint64
		No int64
	}
	rows := make([]kv, 0, 4)
	if softMust("Select(probe_reuse)", pr.Select(func(e *Emp) bool { rows = append(rows, kv{e.ID, e.No}); return true })) {
		return
	}
	fmt.Printf("  再插 1 行(No=99) 后存活: ")
	for _, r := range rows {
		fmt.Printf("{_id=%d,No=%d} ", r.ID, r.No)
	}
	fmt.Printf("| Count=%d TotalRows=%d\n", len(rows), coreRows(pdb, "probe_reuse"))
	hasNo3 := false
	for _, r := range rows {
		if r.No == 3 {
			hasNo3 = true
		}
	}
	chk("T4 新插入不应覆盖既有存活行(No=3 应仍在, 共 3 行)", hasNo3 && len(rows) == 3,
		"3 行且含 No=3", fmt.Sprintf("%d 行 含No3=%v", len(rows), hasNo3))
	if !hasNo3 || len(rows) != 3 {
		problem("严重", "pkID = TotalRows+1 且删除会回退 TotalRows → 删中间行后再插入会复用已占用的 pkID",
			fmt.Sprintf("删 pkID=2 后 TotalRows 回退为 2, 下一条插入分到 pkID=3, 与仍存活的 pkID=3 冲突, 实测存活 %d 行(期望 3, 含 No=3)", len(rows)))
	}
}

func probeFloatTrunc(pdb *DB.DB) {
	sub("T5 float64 列的小数精度")
	pt, err := DB.UseTable[Emp](pdb, "probe_float")
	if softMust("UseTable(probe_float)", err) {
		return
	}
	softMust("Insert(Salary=123.75)", pt.Insert(&Emp{No: 1, Name: "F", Age: 1, Score: 1, Dept: "Dept000", Salary: 123.75, Active: true}))
	got, err2 := pt.Find(1)
	if err2 == nil {
		fmt.Printf("  写入 Salary=123.75 → 读回 %v\n", got.Salary)
	}
	chk("T5 float64 的小数部分应保留", err2 == nil && got.Salary == 123.75,
		"123.75", fmt.Sprintf("%v err=%v", got, err2))
	if got != nil && got.Salary != 123.75 {
		problem("风险", "ORM 把 float64 映射成 8 字节整数列, 小数被静默截断",
			fmt.Sprintf("写入 123.75 读回 %v; 金额/比例等小数数据不能直接存 float64 列(需业务侧放大成整数)", got.Salary))
	}
}

// ============================== P9 关库重开 ==============================
func phase9Reopen(path string) {
	phase("P9 关库重开: 持久化 / schema / 索引 校验")
	db2, err := DB.Open(path)
	must(err)
	defer func() { must(db2.Close()) }()
	empCore, ok := db2.CoreDB().FindTable("emp")
	if !ok {
		must(fmt.Errorf("重开后找不到 emp 表"))
	}
	cols := empCore.GetActiveFieldNames()
	fmt.Printf("  重开后 emp 活跃列: %v\n", cols)
	fmt.Printf("  TotalRows=%d (期望 %d) | 磁盘 %.1fMB\n",
		empCore.TotalRows(), expectRows(), float64(dbSize(path))/(1<<20))
	chk("9.1 重开后 TotalRows 一致", empCore.TotalRows() == uint64(expectRows()),
		strconv.FormatInt(expectRows(), 10), strconv.FormatUint(empCore.TotalRows(), 10))
	chk("9.1 重开后 schema 保留 P7 结果", contains(cols, "Dept") && contains(cols, "DeptName"),
		"含 Dept 与 DeptName", fmt.Sprintf("%v", cols))

	_, bad := verifyEmpCoreScan(empCore, "重开后 core 全量校验", expectRows(), true)
	chk("9.1 重开后逐行公式校验 0 错", bad == 0, "0", strconv.FormatInt(bad, 10))

	idxOK := 0
	for _, nm := range []string{"pk_emp", "idx_emp_no", "idx_emp_age", "idx_emp_dept"} {
		if _, ok := empCore.IndexMeta().FindIndex(nm); ok {
			idxOK++
		}
	}
	chk("9.1 主键+三个二级索引均存在", idxOK == 4, "4", strconv.Itoa(idxOK))

	sub("9.2 ORM 重开表 (UseTable 会按 struct 同步 schema)")
	empT2, err := DB.UseTable[Emp](db2, "emp")
	must(err)
	cols2 := empCore.GetActiveFieldNames()
	fmt.Printf("  UseTable[Emp] 之后活跃列: %v\n", cols2)
	if contains(cols, "DeptName") && !contains(cols2, "DeptName") {
		problem("风险", "ORM 重开表会按 struct 强制同步 schema, 静默删除 struct 里没有的列",
			fmt.Sprintf("纯 core 视角活跃列=%v 含 DeptName, 调用 UseTable[Emp] 后活跃列=%v, DeptName 被自动 DropField; 手工加过列的表被 ORM 打开就会\"丢列\"(数据仍在行内, 再加回同名列可复活)", cols, cols2))
	}
	scanned, bad2 := verifyEmpORM(empT2, "重开后 ORM 全量校验", expectRows())
	chk("9.2 ORM 重开后可读全表", scanned == expectRows() && bad2 == 0,
		fmt.Sprintf("%d 行 0 错", expectRows()), fmt.Sprintf("%d 行 %d 错", scanned, bad2))

	e, errF := empT2.Find(1)
	chk("9.2 重开后主键点查可用", errF == nil && e.No == 1, "No=1", fmt.Sprintf("%v err=%v", e, errF))
	c, errC := empT2.Where("No", "=", int64(sampleNo)).Count()
	chk("9.2 重开后索引等值查询可用", errC == nil && c == 1, "1", fmt.Sprintf("%d err=%v", c, errC))

	want := int64(0)
	for i := int64(1); i <= nEmp; i++ {
		if !isDeletedNo(i) && expAge(i) == int64(bulkAgeTo) {
			want++
		}
	}
	for k := int64(1); k <= extraRows; k++ {
		if expAge(nEmp+k) == int64(bulkAgeTo) {
			want++
		}
	}
	c2, errC2 := empT2.Where("Age", "=", int64(bulkAgeTo)).Count()
	chk("9.2 重开后 Age 索引计数正确", errC2 == nil && c2 == want,
		strconv.FormatInt(want, 10), fmt.Sprintf("%d err=%v", c2, errC2))
}

// ============================== P10 汇总 ==============================
func report(start time.Time) {
	phase("P10 汇总报告")
	fmt.Printf("\n【性能总表】共 %d 项\n", len(perfs))
	fmt.Printf("%-10s %-52s %12s %13s %14s\n", "阶段", "操作", "行数", "耗时", "速率")
	fmt.Printf("%s\n", strings.Repeat("-", 108))
	for _, p := range perfs {
		rate := "-"
		if p.Cost > 0 && p.Rows > 0 {
			rate = human(float64(p.Rows)/p.Cost.Seconds()) + "/s"
		}
		fmt.Printf("%-10s %-52s %12d %13s %14s\n", p.Phase, p.Op, p.Rows, p.Cost.Round(time.Millisecond), rate)
	}

	pass, fail := 0, 0
	for _, c := range checks {
		if c.Pass {
			pass++
		} else {
			fail++
		}
	}
	fmt.Printf("\n【校验总表】共 %d 项: 通过 %d, 失败 %d\n", len(checks), pass, fail)
	if fail > 0 {
		fmt.Printf("%-66s %-24s %s\n", "失败检查项", "期望", "实际")
		for _, c := range checks {
			if !c.Pass {
				fmt.Printf("%-66s %-24s %s\n", c.Name, c.Expect, c.Actual)
			}
		}
	}

	fmt.Printf("\n【数据问题清单】共 %d 项\n", len(problems))
	for i, p := range problems {
		fmt.Printf("  %d. [%s] %s\n     %s\n", i+1, p.Level, p.Title, p.Detail)
	}
	fmt.Printf("\n【结论】总耗时 %s | 校验 %d/%d 通过 | 数据问题 %d 项\n",
		time.Since(start).Round(time.Second), pass, len(checks), len(problems))
	fmt.Printf("  主库保留: %s (%.1fMB) | 探针库: %s (%.1fMB)\n",
		dbPath, float64(dbSize(dbPath))/(1<<20), probePath, float64(dbSize(probePath))/(1<<20))
	fmt.Printf("  可将上述 .db 交给 analyzer 审计\n")
}

// ============================== main ==============================
func deriveScale() {
	delCount = nEmp / 100
	if delCount > 100000 {
		delCount = 100000
	}
	if delCount < 100 {
		delCount = 100
	}
	delCount = delCount / 100 * 100
	delFrom, delTo = nEmp-delCount+1, nEmp

	rowUpdCnt = nEmp / 5000
	if rowUpdCnt > 2000 {
		rowUpdCnt = 2000
	}
	if rowUpdCnt < 10 {
		rowUpdCnt = 10
	}
	rowUpdAt = nEmp/4 + 1

	// 抽样行号: 取 nEmp/3, 保证存在且不落在 rowUpd 区间(仅 10 行)或尾部删除区间内
	sampleNo = nEmp / 3
	if sampleNo < 1 {
		sampleNo = 1
	}

	nJoinMod = nOrder / 5
	if nJoinMod > 200000 {
		nJoinMod = 200000
	}
	if nJoinMod < 5 {
		nJoinMod = 5
	}
}

func main() {
	start := time.Now()
	nEmp = envInt64("TSDB_BENCH_N", 10_000_000)
	nEmp = nEmp / 1000 * 1000
	if nEmp < 1000 {
		nEmp = 1000
	}
	nOrder = envInt64("TSDB_BENCH_ORDERS", 1_000_000)
	name := os.Getenv("TSDB_BENCH_DB")
	if name == "" {
		name = "bench_10m.db"
	}
	abs, err := filepath.Abs(name)
	must(err)
	dbPath = abs
	probePath = filepath.Join(filepath.Dir(abs), strings.TrimSuffix(filepath.Base(abs), ".db")+"_probe.db")
	deriveScale()

	phase0()
	fmt.Printf("\n  清理同前缀旧库文件...\n")
	cleanOld(dbPath)
	cleanOld(probePath)

	db, err := DB.Open(dbPath)
	must(err)
	db.SetCommitPreset("batch")
	db.SetAutoCommitDelay(0)

	empT, deptT, orderT, empCore := phase1(db)
	phase2InsertEmp(db, empT, empCore)
	phase3Queries(empT, empCore)
	_, orderAfterDel := phase4Multi(db, empT, deptT, orderT, empCore)
	phase5Update(db, empT, empCore)
	phase6Delete(db, empT, orderT, empCore, orderAfterDel)
	phase7Schema(db, empT, empCore)

	must(db.Commit())
	must(db.Close())
	fmt.Printf("\n  [已关库] 磁盘 %.1fMB | %s\n", float64(dbSize(dbPath))/(1<<20), memLine())

	phase8Probes()
	phase9Reopen(dbPath)
	report(start)
}