TsDB2/DB/batch_insert_test.go

package DB

import (
	"fmt"
	"path/filepath"
	"testing"
)

// biRow 批量插入用例的行: 二级索引 (Age) + 动态列 (Name/Dept 字符串) + 数值列,
// 覆盖批量路径会碰到的全部结构 (主键树 / 二级树 / 数据页 / 动态堆 / 表信息页)。
type biRow struct {
	BaseModel
	Name  string `vrtx:"maxlen:32"`
	Age   int    `vrtx:"index"`
	Dept  string `vrtx:"maxlen:16"`
	Score int
}

// biUnique 唯一索引用例的行
type biUnique struct {
	BaseModel
	Email string `vrtx:"index,unique"`
}

func biMake(i int) *biRow {
	return &biRow{
		Name:  fmt.Sprintf("name_%06d", i),
		Age:   18 + i%60,
		Dept:  fmt.Sprintf("dept_%02d", i%7),
		Score: i % 100,
	}
}

// TestBatchInsertRoundtrip 批量 + 逐行混插后全量校验:
//   - 行数一致 (TotalRows)
//   - 每行按 _id 点查命中且字段完整
//   - 二级索引按键反查能取到行
//   - 关库重开后一切照旧 (批量写入经缓冲池/提交链路的持久化)
func TestBatchInsertRoundtrip(t *testing.T) {
	dbPath := filepath.Join(t.TempDir(), "batch_insert.db")
	db, err := Open(dbPath)
	if err != nil {
		t.Fatalf("打开: %v", err)
	}
	db.SetCommitPreset("batch")
	db.SetAutoCommitDelay(0)

	tbl, err := UseTable[biRow](db, "bi")
	if err != nil {
		t.Fatalf("建表: %v", err)
	}

	const (
		batch1 = 2000
		single = 500
		batch2 = 1500
		total  = batch1 + single + batch2
	)

	// 批 1
	b1 := make([]*biRow, batch1)
	for i := range b1 {
		b1[i] = biMake(i)
	}
	if err := tbl.BatchInsert(b1); err != nil {
		t.Fatalf("批 1: %v", err)
	}
	// 逐行 500 条 (混插: 批量与单行路径交替构造同一棵树)
	for i := batch1; i < batch1+single; i++ {
		if err := tbl.Insert(biMake(i)); err != nil {
			t.Fatalf("单行 %d: %v", i, err)
		}
	}
	// 批 2
	b2 := make([]*biRow, batch2)
	for j := range b2 {
		b2[j] = biMake(batch1 + single + j)
	}
	if err := tbl.BatchInsert(b2); err != nil {
		t.Fatalf("批 2: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("提交: %v", err)
	}

	core, ok := db.CoreDB().FindTable("bi")
	if !ok {
		t.Fatal("core 找不到表 bi")
	}
	if got := core.TotalRows(); got != total {
		t.Fatalf("行数: 期望 %d, 实际 %d", total, got)
	}

	// 每行按 _id 点查
	checkAll := func(stage string) {
		for i := 0; i < total; i++ {
			obj, err := tbl.Find(uint64(i + 1))
			if err != nil {
				t.Fatalf("%s: Find(%d): %v", stage, i+1, err)
			}
			want := biMake(i)
			if obj.Name != want.Name || obj.Age != want.Age ||
				obj.Dept != want.Dept || obj.Score != want.Score {
				t.Fatalf("%s: 第 %d 行字段错: got %+v", stage, i+1, obj)
			}
		}
	}
	checkAll("提交后")

	// 二级索引反查: Age 索引等值 COUNT, 与期望行数比对
	// Age = 18 + i%60 → 键 age 对应 i ≡ (age-18) mod 60
	age := 18 + 123%60 // 任取一个存在的键
	expect := 0
	for i := 0; i < total; i++ {
		if 18+i%60 == age {
			expect++
		}
	}
	got, err := tbl.Where("Age", "=", age).Count()
	if err != nil {
		t.Fatalf("Age 索引 COUNT: %v", err)
	}
	if int(got) != expect {
		t.Fatalf("Age=%d 索引计数: 期望 %d, 实际 %d", age, expect, got)
	}

	// 关库重开: 持久化 + 重开后的可查性
	if err := db.Close(); err != nil {
		t.Fatalf("关库: %v", err)
	}
	db2, err := Open(dbPath)
	if err != nil {
		t.Fatalf("重开: %v", err)
	}
	defer db2.Close()
	tbl2, err := UseTable[biRow](db2, "bi")
	if err != nil {
		t.Fatalf("重开建表: %v", err)
	}
	_ = tbl2
	// 用 core 逐行点查 (Find 走 ORM, 重开后同一路径)
	for i := 0; i < total; i += 97 {
		obj, err := tbl2.Find(uint64(i + 1))
		if err != nil {
			t.Fatalf("重开: Find(%d): %v", i+1, err)
		}
		want := biMake(i)
		if obj.Name != want.Name || obj.Age != want.Age {
			t.Fatalf("重开: 第 %d 行字段错", i+1)
		}
	}
}

// TestBatchInsertExplicitPK 显式主键混批: 退回逐行路径, 语义不变
func TestBatchInsertExplicitPK(t *testing.T) {
	db, err := Open(filepath.Join(t.TempDir(), "batch_explicit.db"))
	if err != nil {
		t.Fatalf("打开: %v", err)
	}
	defer db.Close()
	db.SetAutoCommitDelay(0)

	tbl, err := UseTable[biRow](db, "bix")
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	// 一半显式主键, 一半自增
	objs := make([]*biRow, 100)
	for i := range objs {
		objs[i] = biMake(i)
		if i%2 == 0 {
			objs[i].ID = uint64(1000 + i)
		}
	}
	if err := tbl.BatchInsert(objs); err != nil {
		t.Fatalf("混批: %v", err)
	}
	// 自增号在显式号之后继续, 不撞车
	next := &biRow{Name: "next", Age: 20, Dept: "d", Score: 1}
	if err := tbl.Insert(next); err != nil {
		t.Fatalf("后续单行: %v", err)
	}
	core, _ := db.CoreDB().FindTable("bix")
	if got := core.TotalRows(); got != 101 {
		t.Fatalf("行数: 期望 101, 实际 %d", got)
	}
	// 显式号 (偶数行, id=1000+i) 能查到
	if _, err := tbl.Find(1000); err != nil {
		t.Fatalf("显式主键 1000 查不到: %v", err)
	}
	if _, err := tbl.Find(1098); err != nil {
		t.Fatalf("显式主键 1098 查不到: %v", err)
	}
	// 自增号 (奇数行, 显式号顶起游标后交错续号, id=1000+i) 也能查到
	if _, err := tbl.Find(1001); err != nil {
		t.Fatalf("自增主键 1001 查不到: %v", err)
	}
	if _, err := tbl.Find(1099); err != nil {
		t.Fatalf("自增主键 1099 查不到: %v", err)
	}
}

// TestBatchInsertUniqueDup 唯一索引批内重复: 整批拒绝, 一行都不落
func TestBatchInsertUniqueDup(t *testing.T) {
	db, err := Open(filepath.Join(t.TempDir(), "batch_unique.db"))
	if err != nil {
		t.Fatalf("打开: %v", err)
	}
	defer db.Close()
	db.SetAutoCommitDelay(0)

	tbl, err := UseTable[biUnique](db, "biu")
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	objs := make([]*biUnique, 50)
	for i := range objs {
		objs[i] = &biUnique{Email: fmt.Sprintf("u%03d@x.com", i)}
	}
	objs[30].Email = objs[7].Email // 批内重复

	if err := tbl.BatchInsert(objs); err == nil {
		t.Fatal("批内重复键应当报错")
	}
	core, _ := db.CoreDB().FindTable("biu")
	if got := core.TotalRows(); got != 0 {
		t.Fatalf("唯一冲突后行数: 期望 0, 实际 %d", got)
	}
	// 去掉重复后整批可写入
	objs[30].Email = "u030@x.com"
	if err := tbl.BatchInsert(objs); err != nil {
		t.Fatalf("去重后批插: %v", err)
	}
	if got := core.TotalRows(); got != 50 {
		t.Fatalf("去重后行数: 期望 50, 实际 %d", got)
	}
}