TsDB2/DB/blob_stream_orm_test.go

package DB

import (
	"bytes"
	"hash/crc32"
	"io"
	"path/filepath"
	"testing"

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

// blobStreamRow 流式载体: 字段本身就是"持有流的对象" (types.Blob),
// 不写 type:blob tag 也会被 goTypeToFieldType 自动推断为 blob 列。
type blobStreamRow struct {
	BaseModel
	Name string     `vrtx:"maxlen:32"`
	Data types.Blob `vrtx:"type:blob"`
}

// TestBlobStreamCarrierRoundTrip 流式载体 (types.Blob) 的完整生命周期:
//
//  1. 写: Data.Source 是 io.Reader, 引擎流式落链 (不整块进内存);
//  2. 回填: Insert 成功后 ref (FirstPageID/TotalLen) 已回填进对象, Source 清空;
//  3. 读: Find 只填 ref, Open(table) 按需拿 io.ReadCloser 流式消费;
//  4. 保链: 只改普通字段的 Update 不抹掉 blob (读回镜像分支);
//  5. 重开: 关库重开照样流式读得回。
func TestBlobStreamCarrierRoundTrip(t *testing.T) {
	dbPath := filepath.Join(t.TempDir(), "blob_stream_orm.db")
	db, err := Open(dbPath)
	if err != nil {
		t.Fatalf("打开: %v", err)
	}
	defer func() { _ = db.Close() }()
	db.SetAutoCommitDelay(0)

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

	content := make([]byte, core.MaxBlobSize*3+7) // 跨 3+ 页, 单页链测不出链指针问题
	for i := range content {
		content[i] = byte(i % 251)
	}
	wantCRC := crc32.ChecksumIEEE(content)

	row := &blobStreamRow{Name: "stream", Data: types.Blob{Source: bytes.NewReader(content)}}
	if err := tbl.Insert(row); err != nil {
		t.Fatalf("Insert: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}

	// 2. ref 已回填, Source 已清 (再 Update 不会把死流灌一遍)
	if row.Data.FirstPageID == 0 || row.Data.TotalLen != uint32(len(content)) {
		t.Fatalf("Insert 后 ref 未回填: %+v", row.Data)
	}
	if row.Data.Source != nil {
		t.Fatalf("Insert 后 Source 应被清掉")
	}

	// 3. 读回: ref + Open 流式消费
	got, err := tbl.Find(row.ID)
	if err != nil {
		t.Fatalf("Find: %v", err)
	}
	rc, err := got.Data.Open(tbl.BlobReader())
	if err != nil {
		t.Fatalf("Open: %v", err)
	}
	data, err := io.ReadAll(rc)
	if err != nil {
		t.Fatalf("流式读: %v", err)
	}
	_ = rc.Close()
	if crc32.ChecksumIEEE(data) != wantCRC {
		t.Fatalf("流式读内容不一致")
	}

	// 4. 只改 Name 的 Update: blob 链必须原样保留 (曾因 blobFieldToRef 判空顺序
	//    把镜像当空 blob, Update 一发整条链被抹掉)
	freeBefore := db.CoreDB().DebugFreePageCount()
	got.Name = "renamed"
	if err := tbl.Update(got); err != nil {
		t.Fatalf("Update: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	renamed, err := tbl.Find(got.ID)
	if err != nil {
		t.Fatalf("Find: %v", err)
	}
	if renamed.Data.TotalLen != uint32(len(content)) {
		t.Fatalf("改名 Update 后 blob 丢失: TotalLen=%d", renamed.Data.TotalLen)
	}
	rc2, err := renamed.Data.Open(tbl.BlobReader())
	if err != nil {
		t.Fatalf("Open: %v", err)
	}
	again, err := io.ReadAll(rc2)
	if err != nil {
		t.Fatalf("流式读: %v", err)
	}
	_ = rc2.Close()
	if crc32.ChecksumIEEE(again) != wantCRC {
		t.Fatalf("改名 Update 后内容不一致")
	}
	// 链页没被回收也没新开 (Update 未触碰 blob)
	if n := db.CoreDB().DebugFreePageCount(); n != freeBefore {
		t.Logf("注意: Update 前后空闲页 %d -> %d", freeBefore, n)
	}

	// 5. 重开库
	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() }()
	tbl2, err := UseTable[blobStreamRow](db2, "stream")
	if err != nil {
		t.Fatalf("重开建表: %v", err)
	}
	reloaded, err := tbl2.Find(row.ID)
	if err != nil {
		t.Fatalf("重开 Find: %v", err)
	}
	rc3, err := reloaded.Data.Open(tbl2.BlobReader())
	if err != nil {
		t.Fatalf("重开 Open: %v", err)
	}
	persist, err := io.ReadAll(rc3)
	if err != nil {
		t.Fatalf("重开流式读: %v", err)
	}
	_ = rc3.Close()
	if crc32.ChecksumIEEE(persist) != wantCRC {
		t.Fatalf("重开后内容不一致")
	}
}