TsDB2/DB/core/overflow_repro_db_test.go

package core

import (
	"fmt"
	"testing"

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

// ============================================================================
// DB 层 "节点满员先写再裂" 越界回归 (生产 panic: DockerManager 持续插入
// 宽二级索引时 sbtEntry 越界 ← sbtPromote 右移循环)。
//
// 触发链: 批量删除会走"软删 + 全索引重建" (Deleter.rebuildIndexes), 二级索引由
// SBTBuildFromSorted 批量构建 —— 内部节点被装到物理满员 (128B 键 iel=140,
// 容量 (16384-32)/140 = 116)。之后任何一次让叶分裂的插入, 提升项都会打进这个
// 满员内部节点: 修复前右移循环越过 sbtEntry 切片上界 panic; 修复后 pre-split
// 先裂内部节点再插。本测试复刻该几何, 修复前红 (panic)、修复后绿。
// ============================================================================

// wideKey 生成 128 字节定宽、字节序 == 数值序的字符串键 (零填充十进制),
// 落在 string(128) 列的档位宽度内, SBT 键 = 定长 128B 存储字节。
func wideKey(i uint64) string { return fmt.Sprintf("%0128d", i) }

// wideTable id(uint64 主键) + s(string(128)) + 主键索引 + s 上的 SBT 二级索引。
func wideTable(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, FieldName: []byte("s"), Flag: types.WidthCodeFor(128)},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	if _, err := tbl.CreateIndex("idx_s", page.IndexTypeBTree, []byte{1}); err != nil {
		t.Fatalf("建二级索引: %v", err)
	}
	return tbl
}

func countScan(t *testing.T, tbl *Table) int64 {
	t.Helper()
	var n int64
	if err := tbl.Query().Scan(func(row []types.FieldValue) bool { n++; return true }); err != nil {
		t.Fatalf("全表扫描: %v", err)
	}
	return n
}

// TestWideKeySBTAfterMassDeleteRebuild 生产形态复刻:
// 批量灌 13921 行 → 批量删 1 行 (全索引重建, SBT 根内部节点恰好物理满员 116 条)
// → 持续单行插入 → 修复前在第一个分裂的叶处 panic。
func TestWideKeySBTAfterMassDeleteRebuild(t *testing.T) {
	db := fixOpenDB(t)
	tbl := wideTable(t, db, "t")

	// 1) 批量灌 13921 行, s 单调递增
	const total = 13921
	pkIDs := make([]uint64, total)
	vals := make([][]types.FieldValue, total)
	for i := 0; i < total; i++ {
		id := uint64(i + 1)
		pkIDs[i] = id
		vals[i] = []types.FieldValue{
			{FieldName: []byte("id"), Value: id},
			{FieldName: []byte("s"), Value: wideKey(id)},
		}
	}
	if _, err := tbl.InsertRowsWithIDs(pkIDs, vals); err != nil {
		t.Fatalf("批量插入: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 2) 批量删 1 行 → 全索引重建: 存活 13920 = 116×120 条 → 116 张满叶 →
	//    SBT 根内部节点恰好 116 条 = 物理满员 (iel=140 容量 116)
	if _, err := tbl.Delete().Where(Eq("id", uint64(1), types.FieldType(0))).Exec(); err != nil {
		t.Fatalf("批量删除: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 3) 持续单行插入 (生产形态): 每张满叶的分裂都把提升项打进满员内部节点。
	//    修复前: sbtPromote 右移循环越界 panic。
	const more = 1000
	for i := uint64(total + 1); i <= total+more; i++ {
		if _, err := tbl.InsertRowWithID(i, []types.FieldValue{
			{FieldName: []byte("id"), Value: i},
			{FieldName: []byte("s"), Value: wideKey(i)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 4) 校验: 行数 / 全扫 / 主键回读 / 二级索引等值查
	const want = int64(total - 1 + more)
	if got := int64(tbl.TotalRows()); got != want {
		t.Fatalf("TotalRows=%d, 想要 %d", got, want)
	}
	if got := countScan(t, tbl); got != want {
		t.Fatalf("全表扫描 %d 行, 想要 %d", got, want)
	}
	if _, err := tbl.FindByPK(1); err == nil {
		t.Fatal("id=1 已删除却仍可读")
	}
	if _, err := tbl.FindByPK(total + 777); err != nil {
		t.Fatalf("id=%d 读不到: %v", total+777, err)
	}
	// 二级索引等值查: 单调键, 每个 s 值恰一行
	hit := 0
	if err := tbl.Query().Where(Eq("s", wideKey(uint64(total+500)), types.FieldType(0))).
		Scan(func(row []types.FieldValue) bool { hit++; return true }); err != nil {
		t.Fatalf("二级索引等值查: %v", err)
	}
	if hit != 1 {
		t.Fatalf("s 等值查应命中 1 行, 实际 %d", hit)
	}
}

// TestBatchInsertMonotonicAndRangeScan 综合用例: 单调递增键 + 二级索引 +
// 持续批量 (InsertRows 2000 行 × 10 批), 之后范围扫描 / 等值查 / 全扫校验行数。
func TestBatchInsertMonotonicAndRangeScan(t *testing.T) {
	db := fixOpenDB(t)
	tbl := wideTable(t, db, "t2")

	// 10 批 × 2000 行, 主键自增 (pkID=0 由引擎分配), s 与 id 同步单调
	const batches = 10
	const per = 2000
	for b := 0; b < batches; b++ {
		pkIDs := make([]uint64, per)
		vals := make([][]types.FieldValue, per)
		for i := 0; i < per; i++ {
			id := uint64(b*per + i + 1) // 仅作 s 的取值来源, 主键走自增
			pkIDs[i] = 0
			vals[i] = []types.FieldValue{
				{FieldName: []byte("id"), Value: id},
				{FieldName: []byte("s"), Value: wideKey(id)},
			}
		}
		if _, err := tbl.InsertRowsWithIDs(pkIDs, vals); err != nil {
			t.Fatalf("第 %d 批插入: %v", b, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	const want = int64(batches * per)
	if got := int64(tbl.TotalRows()); got != want {
		t.Fatalf("TotalRows=%d, 想要 %d", got, want)
	}
	if got := countScan(t, tbl); got != want {
		t.Fatalf("全表扫描 %d 行, 想要 %d", got, want)
	}

	// 范围扫描: id ∈ [5001, 15001) → 10000 行 (走主键索引 RangeScan)
	rng := int64(0)
	if err := tbl.Query().Where(And(
		Ge("id", uint64(5001), types.FieldType(0)),
		Lt("id", uint64(15001), types.FieldType(0)),
	)).Scan(func(row []types.FieldValue) bool { rng++; return true }); err != nil {
		t.Fatalf("范围扫描: %v", err)
	}
	if rng != 10000 {
		t.Fatalf("范围扫描 %d 行, 想要 10000", rng)
	}

	// 二级索引等值 + 边界行回读
	hit := 0
	if err := tbl.Query().Where(Eq("s", wideKey(12345), types.FieldType(0))).
		Scan(func(row []types.FieldValue) bool { hit++; return true }); err != nil {
		t.Fatalf("二级索引等值查: %v", err)
	}
	if hit != 1 {
		t.Fatalf("s=wideKey(12345) 应命中 1 行, 实际 %d", hit)
	}
	for _, id := range []uint64{1, 5000, 5001, 15000, uint64(want)} {
		if _, err := tbl.FindByPK(id); err != nil {
			t.Fatalf("id=%d 读不到: %v", id, err)
		}
	}
}