TsDB2/DB/core/datapage_compact_test.go

package core

// 本文件回答一个具体的运维问题: 单页内"删得多 / 改得多"时, 引擎会在什么时候把这一页
// 就地整理一遍? 四个事实必须能被这些用例单独复现:
//
//  1. 删除本身【不整理】: 只打墓碑 + 累加 deadBytes, 页内物理行数 (Count, 含墓碑) 不变。
//     墓碑占用的行区在此期间只增不减, 也不会被复用 (allocSlot 只认空槽, 墓碑行的槽位
//     仍非 0), 于是这一页会提前判满。
//  2. 整理发生在"要为新行开新页"的那一刻: 主键/时序插入路径在末页装不下时的下一步不是
//     直接开页, 而是先 datapage.CompactForInsert —— 墓碑占比达到 CompactDeadRatio
//     (默认 0.4) 且"整理后确实装得下"时就地整理并重试写入。配置 <=0 即关闭自动整理,
//     行为退回"装不下就开新页"。
//  3. 整理是【就地】且【槽位号不变】的: 页 ID 不变、活行槽位号不变 —— 主键叶条目存的是
//     (dpID, slot), 一变就全指错行, 所以槽位稳定是硬约束。Compact 自身【不落盘】,
//     落盘由调用方 (CompactForInsert / 手工调用方) 承担。
//  4. 更新分两条路: 等长更新走 UpdateBySlot 原地覆盖, 既不产生墓碑也没有"整理"需求;
//     变长 (v9 字符串列长出来) 装不下时走 RebuildWithSlot 页内重建 —— 它会顺带把墓碑
//     一并回收, 但触发条件是"装不下", 而不是阈值。

import (
	"encoding/binary"
	"strings"
	"testing"

	tsindex "git.ycl.pub/yl32940/TsDB2/DB/core/index"
	"git.ycl.pub/yl32940/TsDB2/DB/core/index/btree"
	"git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)

// ======================== 测试脚手架 ========================

// cmpBE 把 pkID 编成 B+ 树叶键的 8 字节 BigEndian 形式 (与 PrimaryHandler 同一编码)。
func cmpBE(pk uint64) []byte {
	b := make([]byte, 8)
	binary.BigEndian.PutUint64(b, pk)
	return b
}

// cmpNewTable 建一张"小行、多行"的主键表: 8B 主键 + 1B + 8B ⇒ 单行约 37 字节,
// 一页能放几百行 —— 这样"单页内删掉一半"才容易构造出来。
func cmpNewTable(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_Uint8, FieldName: []byte("flag")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("建表 %s: %v", name, err)
	}
	// 主键要落在 8 字节定长列上 (列 0): 时序/主键索引都把首列当键。
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	return tbl
}

// cmpNewVarTable 建一张带变长 (字符串) 列的表, 用来区分"等长更新"与"变长更新"。
func cmpNewVarTable(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_String, Flag: types.WidthCodeFor(64), FieldName: []byte("s")},
	})
	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
}

// cmpInsert 插入 n 行, 返回它们的 pkID。
// 用 AllocPKID + InsertRowWithID (而不是让 InsertRow 自己发号) 是为了让行内的 id 列
// 与真实 pkID 逐字节一致 —— 否则叶条目校验行首 8B 时会判成"陈旧槽位"而读不到行。
func cmpInsert(t *testing.T, tbl *Table, n int) []uint64 {
	t.Helper()
	pks := make([]uint64, 0, n)
	for i := 0; i < n; i++ {
		pk := tbl.AllocPKID()
		if _, err := tbl.InsertRowWithID(pk, []types.FieldValue{
			{FieldName: []byte("id"), Value: pk},
			{FieldName: []byte("flag"), Value: uint8(i)},
			{FieldName: []byte("v"), Value: uint64(i)},
		}); err != nil {
			t.Fatalf("插入第 %d 行: %v", i, err)
		}
		pks = append(pks, pk)
	}
	return pks
}

// cmpPrimaryMeta 取表的主键索引元信息 (这些用例只建一个 Primary 索引)。
func cmpPrimaryMeta(t *testing.T, tbl *Table) *IndexMeta {
	t.Helper()
	return cmpMetaOf(t, tbl, page.IndexTypePrimary)
}

// cmpMetaOf 按索引类型取元信息 (返回的是切片元素的地址, 所以元信息里的
// EntryPage 被更新 —— 例如 B+ 树分裂改根 —— 时这里读到的也是最新值)。
func cmpMetaOf(t *testing.T, tbl *Table, indexType page.IndexType) *IndexMeta {
	t.Helper()
	metas := tbl.IndexMeta().metas
	for i := range metas {
		if metas[i].IndexType == indexType {
			return &metas[i]
		}
	}
	t.Fatalf("表上找不到类型为 %d 的索引", indexType)
	return nil
}

// cmpDataPages 返回主键索引当前引用到的全部 DataPage。
func cmpDataPages(t *testing.T, tbl *Table, meta *IndexMeta) []uint32 {
	t.Helper()
	return btree.GetDataPageIDs(tbl.handle, meta.EntryPage)
}

// cmpLoad 从盘 (缓冲池) 重新载入一页 —— 断言必须看【落盘后】的状态, 而不是内存里的对象。
func cmpLoad(t *testing.T, tbl *Table, dpID uint32) *datapage.DataPage {
	t.Helper()
	dp, err := datapage.Load(tbl.handle, dpID)
	if err != nil {
		t.Fatalf("载入数据页 %d: %v", dpID, err)
	}
	return dp
}

// cmpOnlyPage 断言当前只有一页并返回它 —— "单页内操作"是本文件所有用例的前提。
func cmpOnlyPage(t *testing.T, tbl *Table, meta *IndexMeta) uint32 {
	t.Helper()
	dpIDs := cmpDataPages(t, tbl, meta)
	if len(dpIDs) != 1 {
		t.Fatalf("前置条件: 数据应当全落在一页, 实际 %d 页", len(dpIDs))
	}
	return dpIDs[0]
}

// cmpFreezeThreshold 冻结整理阈值, 并在用例结束后还原 —— CompactDeadRatio 是包级可变量,
// 不还原会污染同包其它用例。
func cmpFreezeThreshold(t *testing.T, ratio float64) {
	t.Helper()
	saved := datapage.CompactDeadRatio
	t.Cleanup(func() { datapage.SetCompactDeadRatio(saved) })
	datapage.SetCompactDeadRatio(ratio)
}

// cmpU64 把读回的 8 字节列取值归一成 uint64 (DecodeValue 的具体类型可能变,
// 但"读出来的数就是写进去的数"这一条不能变)。
func cmpU64(t *testing.T, fv types.FieldValue) uint64 {
	t.Helper()
	switch v := fv.Value.(type) {
	case uint64:
		return v
	case int64:
		return uint64(v)
	case uint32:
		return uint64(v)
	case int32:
		return uint64(v)
	default:
		t.Fatalf("列 %q 的值类型是 %T, 期望整数类型", fv.FieldName, fv.Value)
		return 0
	}
}

// cmpStr 把读回的字符串列取值归一成 string (DecodeValue 给出的是 []byte)。
func cmpStr(t *testing.T, fv types.FieldValue) string {
	t.Helper()
	switch v := fv.Value.(type) {
	case string:
		return v
	case []byte:
		return string(v)
	default:
		t.Fatalf("列 %q 的值类型是 %T, 期望字符串家族", fv.FieldName, fv.Value)
		return ""
	}
}

// ======================== 用例 ========================

// TestDataPageDeleteLeavesTombstonesOnly 钉死事实 1:
// 删除只打墓碑 —— 即使墓碑占比已远超整理阈值, 删除本身也不会重排这一页
// (整理被推迟到"要开新页"的那一刻, 见 TestDataPageCompactsBeforeOpeningNewPage)。
func TestDataPageDeleteLeavesTombstonesOnly(t *testing.T) {
	cmpFreezeThreshold(t, 0.4)

	db := fixOpenDB(t)
	tbl := cmpNewTable(t, db, "CMPDEL")
	meta := cmpPrimaryMeta(t, tbl)

	const total, live = 200, 80
	pks := cmpInsert(t, tbl, total)

	dpID := cmpOnlyPage(t, tbl, meta)
	if got := cmpLoad(t, tbl, dpID).Count(); got != uint32(total) {
		t.Fatalf("前置条件: 页内应有 %d 行, 实际 %d", total, got)
	}

	for _, pk := range pks[live:] {
		if err := tbl.DeleteRowByPK(pk); err != nil {
			t.Fatalf("删除 pkID=%d: %v", pk, err)
		}
	}

	dp := cmpLoad(t, tbl, dpID)
	if dp.DeadBytes() == 0 {
		t.Fatal("删除后页内应有失效 (墓碑) 字节, 实际 0 —— 删除没走到软删除")
	}
	if r := dp.DeadRatio(); r < datapage.CompactDeadRatio {
		t.Fatalf("前置条件: 墓碑占比 %.2f 应已越过整理阈值 %.2f", r, datapage.CompactDeadRatio)
	}
	// 键断言: 墓碑占比都过阈值了, 页内物理行数却还是删除前的数字 —— 删除没有重排页。
	// (整理被推迟到"要开新页"的那一刻: 删除时整理代价确定而收益未知, 开页前整理才划算。)
	if dp.Count() != uint32(total) {
		t.Fatalf("墓碑占比已过阈值 %.2f, 但页内物理行数从 %d 变成 %d —— 删除路径在重排页",
			datapage.CompactDeadRatio, total, dp.Count())
	}
	// 删除不应换页, 更不应把页回收 (还有活行)
	if dpIDs := cmpDataPages(t, tbl, meta); len(dpIDs) != 1 || dpIDs[0] != dpID {
		t.Fatalf("删除不应换页: 期望 [%d], 实际 %v", dpID, dpIDs)
	}

	// 活行照旧可读: 墓碑只影响被删掉的那些行
	for _, pk := range pks[:live] {
		if _, err := tbl.FindByPK(pk); err != nil {
			t.Fatalf("活行 pkID=%d 应当仍可读: %v", pk, err)
		}
	}
	// 被删的行则查不到
	for _, pk := range pks[live:] {
		if _, err := tbl.FindByPK(pk); err == nil {
			t.Errorf("已删除的 pkID=%d 不应还能读到", pk)
		}
	}
}

// TestDataPageManualCompactKeepsPageAndSlots 钉死事实 2:
// 手动整理是【就地】的 —— 页 ID 不变、活行槽位号不变 (所以主键叶条目一个字都不用改),
// 墓碑空间全部回收; 但 Compact 自己不落盘, 必须由调用方 Save。
func TestDataPageManualCompactKeepsPageAndSlots(t *testing.T) {
	cmpFreezeThreshold(t, 0.4)

	db := fixOpenDB(t)
	tbl := cmpNewTable(t, db, "CMPCMP")
	meta := cmpPrimaryMeta(t, tbl)

	const total, live = 200, 60
	pks := cmpInsert(t, tbl, total)
	dpID := cmpOnlyPage(t, tbl, meta)

	// 记下每个活行在主键叶里的定位 (dpID, slot): 整理前后必须逐字不变。
	type loc struct {
		dp   uint32
		slot uint16
	}
	before := make(map[uint64]loc, live)
	for _, pk := range pks[:live] {
		gotDP, slot, ok := btree.BTreePKSearch(meta.EntryPage, cmpBE(pk), tbl.handle)
		if !ok || gotDP != dpID {
			t.Fatalf("前置条件: pkID=%d 的叶条目应指向数据页 %d, 实际 (%d, ok=%v)", pk, dpID, gotDP, ok)
		}
		before[pk] = loc{gotDP, slot}
	}

	for _, pk := range pks[live:] {
		if err := tbl.DeleteRowByPK(pk); err != nil {
			t.Fatalf("删除 pkID=%d: %v", pk, err)
		}
	}
	if r := cmpLoad(t, tbl, dpID).DeadRatio(); r < datapage.CompactDeadRatio {
		t.Fatalf("前置条件: 墓碑占比 %.2f 未越过阈值 %.2f", r, datapage.CompactDeadRatio)
	}

	dp := cmpLoad(t, tbl, dpID)
	if !dp.MaybeCompact() {
		t.Fatal("墓碑占比已越过阈值, MaybeCompact 应当动手整理")
	}
	dp.Save() // Compact 只改内存缓冲, 落盘是调用方的责任 (见 datapage.Compact)

	re := cmpLoad(t, tbl, dpID)
	if re.DeadBytes() != 0 {
		t.Errorf("整理后失效字节应归零, 实际 %d", re.DeadBytes())
	}
	if re.Count() != uint32(live) {
		t.Errorf("整理后页内物理行数应等于活行数 %d, 实际 %d", live, re.Count())
	}
	if dpIDs := cmpDataPages(t, tbl, meta); len(dpIDs) != 1 || dpIDs[0] != dpID {
		t.Errorf("整理应当就地完成: 期望页集合 [%d], 实际 %v", dpID, dpIDs)
	}

	for _, pk := range pks[:live] {
		gotDP, slot, ok := btree.BTreePKSearch(meta.EntryPage, cmpBE(pk), tbl.handle)
		if !ok || gotDP != before[pk].dp || slot != before[pk].slot {
			t.Errorf("pkID=%d 的定位从 (%d,%d) 变成 (%d,%d, ok=%v) —— 叶条目与槽位对不上了",
				pk, before[pk].dp, before[pk].slot, gotDP, slot, ok)
			continue
		}
		// 行的内容也要对: 整理是搬字节, 搬错行会在这里露出来 (v 列写的就是 i)
		row, err := tbl.FindByPK(pk)
		if err != nil {
			t.Errorf("整理后 pkID=%d 读不回: %v", pk, err)
			continue
		}
		if len(row) == 0 || cmpU64(t, row[len(row)-1]) != uint64(pk-1) {
			t.Errorf("整理后 pkID=%d 的 v 列不对: %v", pk, row)
		}
	}
	// 墓碑行的叶条目已被摘除, 不应再查得到
	for _, pk := range pks[live:] {
		if _, _, ok := btree.BTreePKSearch(meta.EntryPage, cmpBE(pk), tbl.handle); ok {
			t.Errorf("已删除的 pkID=%d 不应还能在叶里查到", pk)
		}
	}
}

// TestDataPageCompactsBeforeOpeningNewPage 覆盖事实 2 (A/B 对照):
// 先把第一页写满 (再写一行就要开新页), 再把它删到只剩 keep 行活行 —— 此刻第一页"行区已满"
// 且几乎全是墓碑, 下一次写入正是引擎做"整理还是开新页"抉择的时刻:
//   - 默认阈值 (0.4): 先就地整理, 探针行仍写进第一页, 墓碑清零, 这一页随后还能吃下大量行;
//   - 阈值配成 0 (关闭): 探针行落到别的页, 第一页永远拖着那批墓碑。
func TestDataPageCompactsBeforeOpeningNewPage(t *testing.T) {
	db := fixOpenDB(t)
	const keep = 10 // 第一页保留的活行数 (删完墓碑占比 ≈0.97)

	type outcome struct {
		firstDP   uint32 // 第一页 (被删到只剩 keep 行的那一页) 的页 ID
		probePage uint32 // 探针行落在哪一页
		pages     int    // 探针写入后的总页数
		firstRows uint32 // 第一页物理行数
		firstDead uint32 // 第一页墓碑字节
		extra     int    // 之后第一页还能吃下多少行 (换页前)
	}
	run := func(name string, ratio float64) outcome {
		cmpFreezeThreshold(t, ratio) // 每组独立设定阈值, 用例结束时自动还原
		tbl := cmpNewTable(t, db, name)
		meta := cmpPrimaryMeta(t, tbl)

		// 1) 一直写到出现第二页: 第一页至此装满 (pkID 1..fill-1 都在第一页里)
		fill := 0
		for {
			cmpInsert(t, tbl, 1)
			fill++
			if len(cmpDataPages(t, tbl, meta)) > 1 {
				break
			}
		}
		pageIDs := cmpDataPages(t, tbl, meta)
		firstDP := pageIDs[0]
		firstRows := cmpLoad(t, tbl, firstDP).Count()
		if firstRows == 0 || firstRows != uint32(fill-1) {
			t.Fatalf("%s: 前置条件不成立: 第一页行数 %d, 期望 %d", name, firstRows, fill-1)
		}

		// 2) 删掉溢出的那一行: 它独占的第二页随之被回收 (AllDeleted → FreePage),
		//    "当前追加页"又回到第一页 —— 而第一页此刻是满的, 于是下一次写入必然要面对
		//    "整理本页还是开新页"的抉择。
		if err := tbl.DeleteRowByPK(uint64(fill)); err != nil {
			t.Fatalf("%s: 删除溢出页里的 pkID=%d: %v", name, fill, err)
		}
		if got := cmpDataPages(t, tbl, meta); len(got) != 1 || got[0] != firstDP {
			t.Fatalf("%s: 前置条件: 溢出页应被回收, 实际页集合 %v (第一页 %d)", name, got, firstDP)
		}

		// 3) 第一页只留前 keep 行, 其余全删 (墓碑留在第一页)
		for pk := uint64(keep + 1); pk <= uint64(firstRows); pk++ {
			if err := tbl.DeleteRowByPK(pk); err != nil {
				t.Fatalf("%s: 删除 pkID=%d: %v", name, pk, err)
			}
		}
		if dp := cmpLoad(t, tbl, firstDP); dp.DeadBytes() == 0 {
			t.Fatalf("%s: 前置条件: 删除后第一页应留下墓碑", name)
		}

		// 4) 探针: 第一页已满且几乎全是墓碑, 引擎必须在这里抉择
		probePK := cmpInsert(t, tbl, 1)[0]
		probePage, _, ok := btree.BTreePKSearch(meta.EntryPage, cmpBE(probePK), tbl.handle)
		if !ok {
			t.Fatalf("%s: 探针行 pkID=%d 写入后查不到", name, probePK)
		}
		out := outcome{firstDP: firstDP, probePage: probePage, pages: len(cmpDataPages(t, tbl, meta))}
		mid := cmpLoad(t, tbl, firstDP)
		out.firstRows, out.firstDead = mid.Count(), mid.DeadBytes()

		// 5) 第一页还能吃下多少行 (探针留在第一页才有意义)
		if out.probePage == firstDP {
			base := out.pages
			for out.extra < 1000 {
				cmpInsert(t, tbl, 1)
				out.extra++
				if len(cmpDataPages(t, tbl, meta)) > base {
					break
				}
			}
		}
		// 活行必须始终读得到: 整理保持槽位号不变, 叶条目不该指错行
		for pk := uint64(1); pk <= keep; pk++ {
			if _, err := tbl.FindByPK(pk); err != nil {
				t.Errorf("%s: 整理后 pkID=%d 读不回: %v", name, pk, err)
			}
		}
		return out
	}

	on := run("CMPON", 0.4) // 默认配置: 开页前先按阈值整理
	off := run("CMPOFF", 0) // 阈值 <=0: 自动整理关闭

	if off.probePage == off.firstDP {
		t.Error("关掉自动整理时探针行不该写进那个满是墓碑的页")
	}
	if off.pages != 2 {
		t.Errorf("关掉自动整理时总页数应为 2, 实际 %d", off.pages)
	}
	if off.firstDead == 0 {
		t.Errorf("关掉自动整理后第一页应仍拖着墓碑, 实际 0 B")
	}
	if on.probePage != on.firstDP {
		t.Errorf("自动整理后探针行应写进第一页 %d, 实际落在 %d", on.firstDP, on.probePage)
	}
	if on.pages != 1 {
		t.Errorf("自动整理后不该多开页, 实际 %d 页", on.pages)
	}
	if on.firstRows != keep+1 {
		t.Errorf("自动整理后第一页应只剩 %d 行活行 + 探针行, 实际 %d 行", keep, on.firstRows)
	}
	if on.firstDead != 0 {
		t.Errorf("自动整理后第一页墓碑应清零, 实际 %d B", on.firstDead)
	}
	// 这才是整理的意义: 同一页随后还能装回大量行 (墓碑占的位置没有白占)
	if on.extra < 300 {
		t.Errorf("整理后第一页应还能追加 300+ 行, 实际 %d 行", on.extra)
	}
	t.Logf("写满一页后删剩 %d 行再写 1 行: 自动整理 → 第一页 %d 行/墓碑 %d B, 之后还能追加 %d 行;"+
		" 关闭整理 → 第一页 %d 行/墓碑 %d B",
		keep, on.firstRows, on.firstDead, on.extra, off.firstRows, off.firstDead)
}

// TestCompactForInsertGuards 覆盖事实 2 的前置判断: 只有"墓碑占比达阈值"且"整理后确实
// 装得下"才值得搬一遍 (整理是 O(页内行数) 的排序 + 拷贝)。两个条件缺一, 都必须在改动
// 本页之前就返回 false —— 调用方据此开新页。
func TestCompactForInsertGuards(t *testing.T) {
	cmpFreezeThreshold(t, 0.4)

	db := fixOpenDB(t)
	tbl := cmpNewTable(t, db, "CMPGUARD")

	const rowLen = 24 // 每行 24B, 页内占 4+24=28B
	need := uint32(rowLen + 4)
	row := func(fill byte) []byte {
		r := make([]byte, rowLen)
		for i := range r {
			r[i] = fill
		}
		return r
	}

	// ① 没有墓碑可回收: 不整理
	fresh := datapage.New(tbl.handle)
	for i := 1; i <= 3; i++ {
		fresh.InsertRowSorted(row(byte(i)))
	}
	fresh.Save()
	if fresh.CompactForInsert(need) {
		t.Error("没有墓碑时不该整理")
	}

	// ② 有墓碑, 但"整理了也装不下": 宁可让调用方开新页, 也不白搬一趟
	guarded := datapage.New(tbl.handle)
	for i := 1; i <= 10; i++ {
		guarded.InsertRowSorted(row(byte(i)))
	}
	for i := 3; i <= 10; i++ { // 删掉 8 行 → 墓碑占比 0.8
		if !guarded.DeleteBySlot(uint32(i)) {
			t.Fatalf("删除槽位 %d 失败", i)
		}
	}
	guarded.Save()
	dead, rows := guarded.DeadBytes(), guarded.Count()
	if dead == 0 {
		t.Fatal("前置条件: 应留下墓碑")
	}
	if guarded.CompactForInsert(datapage.PageSize) {
		t.Error("整理后也装不下时不该整理")
	}
	if re := cmpLoad(t, tbl, guarded.PageID()); re.DeadBytes() != dead || re.Count() != rows {
		t.Errorf("判断不成立时不该改动本页: 墓碑 %d→%d, 物理行数 %d→%d",
			dead, re.DeadBytes(), rows, re.Count())
	}

	// ③ 墓碑占比没到阈值: 配置说了算
	datapage.SetCompactDeadRatio(0.9)
	if guarded.CompactForInsert(need) {
		t.Error("墓碑占比 0.8 未达阈值 0.9, 不该整理")
	}
	if re := cmpLoad(t, tbl, guarded.PageID()); re.DeadBytes() != dead {
		t.Errorf("未达阈值时不该改动本页: 墓碑 %d→%d", dead, re.DeadBytes())
	}

	// ④ 阈值达标且确实装得下: 就地整理并落盘, 活行槽位号不变 (1/2 仍在原槽)
	datapage.SetCompactDeadRatio(0.4)
	if !guarded.CompactForInsert(need) {
		t.Fatal("墓碑占比 0.8 已过阈值且整理后装得下, 应当整理")
	}
	re := cmpLoad(t, tbl, guarded.PageID())
	if re.DeadBytes() != 0 || re.Count() != 2 {
		t.Errorf("整理后应只剩 2 行活行且无墓碑, 实际 行数=%d 墓碑=%d B", re.Count(), re.DeadBytes())
	}
	for i := uint32(1); i <= 2; i++ {
		if _, ok := re.ReadRowBySlot(i); !ok {
			t.Errorf("活行应留在原槽位 %d (槽位号不变是叶条目有效的前提)", i)
		}
	}
	for i := uint32(3); i <= 10; i++ {
		if _, ok := re.ReadRowBySlot(i); ok {
			t.Errorf("槽位 %d 上的墓碑已被回收, 不该还能读到行", i)
		}
	}
}

// TestTimeDataPageCompactsBeforeOpeningNewSegment 覆盖事实 2 的时序侧:
// 时序表的"页"就是时间段分段 (扁平索引链 + 每段一个 DataPage), 所以"本段装不下时先按配置
// 整理本段, 整理了也装不下才新开分段"的行为与主键表一致 —— 少开的分段就是少一页碎片。
//
// 构造手法: 每行约 6.2KB (3 个 2048 字节的字符串列), 一页只放得下极少几行 —— 把末段填到
// "再写一行就装不下" (用 IsFull 判定, 不猜容量), 再删掉刚写进去的那一行, 就得到"末段已满
// (行区不因删除而缩小) 且拖着墓碑"的状态, 下一次写入必然要做抉择。
// 用小行构造不出来: 小行的末段永远是刚新开的那个空段, 轮不到整理。
func TestTimeDataPageCompactsBeforeOpeningNewSegment(t *testing.T) {
	db := fixOpenDB(t)
	const bigLen = 2048 // 3 × 2048 字节字符串 + 时间戳 ⇒ 整行约 6.2KB, 一页两行

	big := strings.Repeat("x", bigLen)
	newTimeTable := func(name string) (*Table, *IndexMeta, *tsindex.TimeIndexHandler) {
		t.Helper()
		tbl, err := db.CreateTable(name, []types.FieldInfo{
			{FieldType: types.FieldType_Uint64, FieldName: []byte("_ts")},
			{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(bigLen), FieldName: []byte("s1")},
			{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(bigLen), FieldName: []byte("s2")},
			{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(bigLen), FieldName: []byte("s3")},
		})
		if err != nil {
			t.Fatalf("建时序表 %s: %v", name, err)
		}
		if _, err := tbl.CreateIndex("ts", page.IndexTypeTime, []byte{0}); err != nil {
			t.Fatalf("%s: 建时序索引: %v", name, err)
		}
		return tbl, cmpMetaOf(t, tbl, page.IndexTypeTime), &tsindex.TimeIndexHandler{}
	}

	type outcome struct {
		segments        int    // 探针写入后的分段数
		liveBeforeProbe int    // 删除后、探针写入前末段里活着的行数
		lastRows        uint32 // 末段物理行数
		lastDead        uint32 // 末段墓碑字节
	}
	run := func(name string, ratio float64) outcome {
		cmpFreezeThreshold(t, ratio)
		tbl, meta, h := newTimeTable(name)
		// 宽行在页内的占用: storeRow = [pkID 8B][rowData], rowData = [版本(1) + 列偏移表(2×3) +
		// 时间戳(8) + 3×2048 字符串 + NULL 位图(1)]。
		// 注意 IsFull 的 rowSize 约定【不含】4B 长度前缀 (见 datapage.IsFull 注释), 别多算。
		rowSize := uint32(8 + (1 + 2*3 + 8 + 3*bigLen + 1))
		ts := uint64(1_700_000_000_000) // 毫秒时间戳, 必须非递减 (时序表只收顺序写入)
		insert := func() uint64 {
			t.Helper()
			pk := tbl.AllocPKID()
			if _, err := tbl.InsertRowWithID(pk, []types.FieldValue{
				{FieldName: []byte("_ts"), Value: ts},
				{FieldName: []byte("s1"), Value: big},
				{FieldName: []byte("s2"), Value: big},
				{FieldName: []byte("s3"), Value: big},
			}); err != nil {
				t.Fatalf("%s: 插入 ts=%d: %v", name, ts, err)
			}
			ts += 1000
			return pk
		}

		// 1) 先插到出现第 2 个分段: 第 1 段至此装满, 末段 (seg2) 里有且只有刚溢出的那一行
		pk1, pk2 := insert(), insert()
		lastPK := uint64(0)
		for len(h.GetDataPageIDs(tbl.handle, meta.EntryPage)) < 2 {
			lastPK = insert()
		}
		seg2 := h.GetDataPageIDs(tbl.handle, meta.EntryPage)[1]

		// 2) 把末段填到"再插一行同样的宽行就装不下"为止 —— 用 IsFull 判定, 不猜页容量
		for !cmpLoad(t, tbl, seg2).IsFull(rowSize) {
			lastPK = insert()
		}

		// 3) 删掉刚写进末段的那一行: 删除不缩小行区, 所以末段对本尺寸的行仍然"满",
		//    但多了一块墓碑 —— 这正是"该整理"的状态。
		//    (时序删除有意不回收页与分段条目: 条目还指向它, 只是行没了。)
		if err := tbl.DeleteRowByPK(lastPK); err != nil {
			t.Fatalf("%s: 删除 pkID=%d: %v", name, lastPK, err)
		}
		before := cmpLoad(t, tbl, seg2)
		if before.DeadBytes() == 0 {
			t.Fatalf("%s: 前置条件: 末段删除后应留下墓碑", name)
		}
		if !before.IsFull(rowSize) {
			t.Fatalf("%s: 前置条件: 删掉一行后末段对宽行仍应是满的", name)
		}
		if got := h.GetDataPageIDs(tbl.handle, meta.EntryPage); len(got) != 2 || got[1] != seg2 {
			t.Fatalf("%s: 前置条件: 分段应仍是 2 个 (末段 %d), 实际 %v", name, seg2, got)
		}
		// 末段里还活着的行数 (删掉 lastPK 之后的幸存者) —— 整理只回收墓碑, 这些行必须原地留下,
		// 所以"整理成功"之后的末段行数必然等于它 + 1 (探针那一行)。不猜页容量。
		liveBefore := 0
		for s := uint32(1); s <= before.SlotCount(); s++ {
			if _, ok := before.ReadRowBySlot(s); ok {
				liveBefore++
			}
		}

		// 4) 探针: 时间戳更大 → 路由到末段 (它满且拖着墓碑) → 引擎在这里抉择
		probePK := insert()
		out := outcome{segments: len(h.GetDataPageIDs(tbl.handle, meta.EntryPage)), liveBeforeProbe: liveBefore}
		last := cmpLoad(t, tbl, seg2)
		out.lastRows, out.lastDead = last.Count(), last.DeadBytes()

		// 分段里的活行必须一直读得到 (整理保持槽位号不变, 索引条目不该指错行)
		for _, pk := range []uint64{pk1, pk2, probePK} {
			if _, err := tbl.FindByPK(pk); err != nil {
				t.Errorf("%s: pkID=%d 读不回: %v", name, pk, err)
			}
		}
		return out
	}

	on := run("TSPON", 0.4) // 默认配置: 开新分段前先按阈值整理本段
	off := run("TSPOFF", 0) // 阈值 <=0: 自动整理关闭

	if off.segments != 3 {
		t.Errorf("关掉自动整理时探针行应开新分段 (共 3 段), 实际 %d 段", off.segments)
	}
	if off.lastDead == 0 {
		t.Errorf("关掉自动整理后末段应仍拖着墓碑, 实际 0 B")
	}
	if on.segments != 2 {
		t.Errorf("自动整理后探针行应写进本段 (分段数不变 = 2), 实际 %d 段", on.segments)
	}
	if want := uint32(on.liveBeforeProbe + 1); on.lastRows != want {
		t.Errorf("自动整理后末段应只剩 %d 行 (整理前 %d 活行 + 探针 1 行), 实际 %d 行",
			want, on.liveBeforeProbe, on.lastRows)
	}
	if on.lastDead != 0 {
		t.Errorf("自动整理后末段墓碑应清零, 实际 %d B", on.lastDead)
	}
	t.Logf("时序表末段写满后删一行再写 1 行: 自动整理 → 分段数 %d, 末段 %d 行/墓碑 %d B;"+
		" 关闭整理 → 分段数 %d, 末段 %d 行/墓碑 %d B",
		on.segments, on.lastRows, on.lastDead, off.segments, off.lastRows, off.lastDead)
}

// TestDataPageUpdateInPlaceVsRebuild 钉死事实 3:
// 等长更新原地覆盖 —— 不产生墓碑, 也就没有"整理"的需求;
// 变长更新装不下时走页内重建 —— 它会顺带回收墓碑, 但触发条件是"装不下"而非阈值。
func TestDataPageUpdateInPlaceVsRebuild(t *testing.T) {
	cmpFreezeThreshold(t, 0.4)

	db := fixOpenDB(t)
	tbl := cmpNewVarTable(t, db, "CMPUPD")
	meta := cmpPrimaryMeta(t, tbl)

	const total, live = 40, 25
	pks := make([]uint64, 0, total)
	for i := 0; i < total; i++ {
		pk := tbl.AllocPKID()
		if _, err := tbl.InsertRowWithID(pk, []types.FieldValue{
			{FieldName: []byte("id"), Value: pk},
			{FieldName: []byte("s"), Value: "abc"},
		}); err != nil {
			t.Fatalf("插入第 %d 行: %v", i, err)
		}
		pks = append(pks, pk)
	}
	dpID := cmpOnlyPage(t, tbl, meta)

	for _, pk := range pks[live:] {
		if err := tbl.DeleteRowByPK(pk); err != nil {
			t.Fatalf("删除 pkID=%d: %v", pk, err)
		}
	}
	// 记下活行槽位: 变长更新若换了槽位/换了页, 这里会立刻发现
	beforeSlot := make(map[uint64]uint16, live)
	for _, pk := range pks[:live] {
		_, slot, ok := btree.BTreePKSearch(meta.EntryPage, cmpBE(pk), tbl.handle)
		if !ok {
			t.Fatalf("前置条件: pkID=%d 应能定位", pk)
		}
		beforeSlot[pk] = slot
	}
	deletedBytes := cmpLoad(t, tbl, dpID).DeadBytes()
	if deletedBytes == 0 {
		t.Fatal("前置条件: 删除后应留下墓碑字节")
	}

	// (1) 等长更新: "abc" → "xyz", 行长不变 → UpdateBySlot 原地覆盖。
	// 原地覆盖既不产生新墓碑, 也不回收旧墓碑 —— 失效字节保持不动。
	if err := tbl.UpdateRowByPK(pks[0], []types.FieldValue{
		{FieldName: []byte("s"), Value: "xyz"},
	}); err != nil {
		t.Fatalf("等长更新应当成功: %v", err)
	}
	dp := cmpLoad(t, tbl, dpID)
	if dp.DeadBytes() != deletedBytes {
		t.Errorf("等长更新不该动墓碑: 更新前 %d, 更新后 %d", deletedBytes, dp.DeadBytes())
	}
	if dp.Count() != uint32(total) {
		t.Errorf("等长更新不该重排页: 物理行数应为 %d, 实际 %d", total, dp.Count())
	}
	if got, err := tbl.FindByPK(pks[0]); err != nil {
		t.Errorf("等长更新后读不回: %v", err)
	} else if cmpStr(t, got[len(got)-1]) != "xyz" {
		t.Errorf("等长更新没生效: %v", got)
	}

	// (2) 变长更新: 3 字节 → 40 字节, 原地装不下 → 页内重建 (RebuildWithSlot)。
	// 重建会紧凑重排, 于是墓碑被【顺带】回收 —— 但它不是被阈值触发的。
	long := ""
	for i := 0; i < 40; i++ {
		long += "x"
	}
	if err := tbl.UpdateRowByPK(pks[1], []types.FieldValue{
		{FieldName: []byte("s"), Value: long},
	}); err != nil {
		t.Fatalf("变长更新应当成功 (页内重建): %v", err)
	}
	dp = cmpLoad(t, tbl, dpID)
	if dp.DeadBytes() != 0 {
		t.Errorf("页内重建应顺带回收墓碑, 实际仍剩 %d 字节", dp.DeadBytes())
	}
	if dp.Count() != uint32(live) {
		t.Errorf("页内重建后物理行数应等于活行数 %d, 实际 %d", live, dp.Count())
	}
	if dpIDs := cmpDataPages(t, tbl, meta); len(dpIDs) != 1 || dpIDs[0] != dpID {
		t.Errorf("变长 40 字节应当仍在本页, 页集合变成 %v", dpIDs)
	}
	for _, pk := range pks[:live] {
		_, slot, ok := btree.BTreePKSearch(meta.EntryPage, cmpBE(pk), tbl.handle)
		if !ok || slot != beforeSlot[pk] {
			t.Errorf("pkID=%d 的槽位从 %d 变成 %d (ok=%v) —— 叶条目会指错行",
				pk, beforeSlot[pk], slot, ok)
		}
	}
	if got, err := tbl.FindByPK(pks[1]); err != nil {
		t.Errorf("变长更新后读不回: %v", err)
	} else if cmpStr(t, got[len(got)-1]) != long {
		t.Errorf("变长更新没生效: %v", got)
	}
}