TsDB2/DB/blob_test.go

package DB

import (
	"bytes"
	"path/filepath"
	"testing"

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

// blobOrmRow blob 列在 ORM 层的载体: []byte + `type:blob` 标签。
//
// 载体只能是 []byte (见 types.go 的 blob 载体校验: string 载体按定长串语义会被第一个
// 0x00 截断), 所以这里也不拿 string 试。
type blobOrmRow struct {
	BaseModel
	Name string `vrtx:"maxlen:32"`
	Body []byte `vrtx:"type:blob"`
}

// TestBlobColumnORMRoundTrip blob 列在 ORM 层的完整往返。
//
// 钉三件事:
//  1. 写: []byte 直接 Insert 即可, 调用方不用手动开事务 / 调 BlobCommit —— ORM 自动接管
//     这个边界 (Table.Insert 里那个 tx), 内容落链页;
//  2. 读: []byte 载体拿到的是【内容】而不是 8 字节引用 —— 引用只活在行内, 读回侧不把链
//     解开的话字段永远是 nil (写进去什么也读不回来);
//  3. 持久: 关库重开之后照样读得回来 (表是从盘上加载的, 读链要靠 db 反向注入)。
func TestBlobColumnORMRoundTrip(t *testing.T) {
	dbPath := filepath.Join(t.TempDir(), "blob_orm.db")
	db, err := Open(dbPath)
	if err != nil {
		t.Fatalf("打开: %v", err)
	}
	// 不 defer 关这个句柄: 本用例中途要显式关掉再重开 (见下), 重复 Close 会走一遍
	// 已经关掉的 file。fail-fast 时留下的句柄由进程退出回收。
	db.SetAutoCommitDelay(0)

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

	// big: 特意超过 Dynamicbyte 的单值上限 (内容整块落一个堆页), 只有 blob 链装得下。
	big := make([]byte, core.MaxBlobSize*2+5)
	for i := range big {
		big[i] = byte(i * 13 % 251)
	}
	// zeros: 内容含 0x00 (二进制里合法, 按字符串语义读会被静默截断)。
	zeros := []byte{0x00, 0x01, 0x00, 0xFF, 0x00, 0x7F}

	rows := []*blobOrmRow{
		{Name: "small", Body: []byte("hello blob")},
		{Name: "big", Body: big},
		{Name: "zeros", Body: zeros},
		{Name: "empty", Body: []byte{}},
	}
	for _, r := range rows {
		if err := tbl.Insert(r); err != nil {
			t.Fatalf("Insert(%s): %v", r.Name, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("提交: %v", err)
	}

	check := func(stage string) {
		for _, want := range rows {
			got, err := tbl.Find(want.ID)
			if err != nil {
				t.Fatalf("%s: Find(%d, %s): %v", stage, want.ID, want.Name, err)
			}
			if !bytes.Equal(got.Body, want.Body) {
				t.Fatalf("%s: %s 行内容不一致: 读回 %d 字节, 写入 %d 字节",
					stage, want.Name, len(got.Body), len(want.Body))
			}
		}
	}
	check("提交后")

	// 行内只放 8 字节引用: 内容不跟着行走 —— 这是"内容可以跨页"的前提。
	coreTbl, ok := db.CoreDB().FindTable("bloborm")
	if !ok {
		t.Fatal("core 找不到表 bloborm")
	}
	fvs, err := coreTbl.FindByPK(rows[1].ID)
	if err != nil {
		t.Fatalf("FindByPK: %v", err)
	}
	var ref types.BlobRef
	var seen bool
	for _, fv := range fvs {
		if r, isRef := fv.Value.(types.BlobRef); isRef {
			ref, seen = r, true
		}
	}
	if !seen {
		t.Fatalf("行内 blob 列的值不是 types.BlobRef: %+v (内容被内联进行里了?)", fvs)
	}
	if int(ref.TotalLen) != len(big) {
		t.Errorf("ref.TotalLen = %d, 期望 %d", ref.TotalLen, len(big))
	}

	// 关库重开: 盘上加载出来的表读 blob 依赖 db 反向注入 (bindDB)。
	if err := db.Close(); err != nil {
		t.Fatalf("关库: %v", err)
	}
	db2, err := Open(dbPath)
	if err != nil {
		t.Fatalf("重开: %v", err)
	}
	defer func() { _ = db2.Close() }()
	db2.SetAutoCommitDelay(0)

	tbl2, err := UseTable[blobOrmRow](db2, "bloborm")
	if err != nil {
		t.Fatalf("重开建表: %v", err)
	}
	for _, want := range rows {
		got, err := tbl2.Find(want.ID)
		if err != nil {
			t.Fatalf("重开: Find(%d, %s): %v", want.ID, want.Name, err)
		}
		if !bytes.Equal(got.Body, want.Body) {
			t.Fatalf("重开: %s 行内容不一致: 读回 %d 字节, 写入 %d 字节",
				want.Name, len(got.Body), len(want.Body))
		}
	}
}

// TestBlobColumnORMUpdateReplacesChain blob 列更新到新的内容后, 读回的是新内容。
//
// 更新走的是另一条路径 (serializeRowInternal 的 oldBlobRefs 释放 + 新链落地), 与插入
// 不是同一段代码 —— 只在插入上验一遍, 更新时"旧链没释放 / 新值没落链"都不会被发现。
func TestBlobColumnORMUpdateReplacesChain(t *testing.T) {
	db, err := Open(filepath.Join(t.TempDir(), "blob_update.db"))
	if err != nil {
		t.Fatalf("打开: %v", err)
	}
	defer func() { _ = db.Close() }()
	db.SetAutoCommitDelay(0)

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

	obj := &blobOrmRow{Name: "n", Body: []byte("first")}
	if err := tbl.Insert(obj); err != nil {
		t.Fatalf("Insert: %v", err)
	}

	next := make([]byte, core.MaxBlobSize+3)
	for i := range next {
		next[i] = byte(i % 97)
	}
	obj.Body = next
	if err := tbl.Update(obj); err != nil {
		t.Fatalf("Update: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("提交: %v", err)
	}

	got, err := tbl.Find(obj.ID)
	if err != nil {
		t.Fatalf("Find: %v", err)
	}
	if !bytes.Equal(got.Body, next) {
		t.Fatalf("更新后内容不一致: 读回 %d 字节, 写入 %d 字节", len(got.Body), len(next))
	}

	// 再读一次仍是同一份内容 (旧链若被误释放, 这里会读出损坏或短读)。
	again, err := tbl.Find(obj.ID)
	if err != nil {
		t.Fatalf("二次 Find: %v", err)
	}
	if !bytes.Equal(again.Body, next) {
		t.Fatalf("二次读内容不一致: %d 字节", len(again.Body))
	}
}