TsDB2/demo/blob_orm/main.go

package main

// ORM blob 全景演示 —— 两种载体, 一张表:
//
//   - []byte     一次性载体: 写时整块在内存, 读回整块拉进内存。小内容够用。
//   - types.Blob 流式载体: 对象里持有"流对象" ——
//       写侧 Data.Source 是任意 io.Reader (*os.File / 网络流 / bytes.Reader...),
//       引擎 io.Copy 流式落链, 内存峰值 ~128KB, 与内容大小无关;
//       读侧引擎只填回 8 字节引用 (TotalLen/FirstPageID), 内容仍躺在链页上,
//       调 row.Data.Open(tbl.BlobReader()) 按需拿 io.ReadCloser 流式消费。
//
// 场景清单:
//   [1] 流式写: 32MB 文件 → types.Blob{Source} → Insert (内存峰值 < 2MB)
//   [2] 流式读: Find → Open → io.Copy 流式消费, 校验 CRC (同样不整块进内存)
//   [3] []byte 载体对照: 小内容直接 Insert/Find
//   [4] 持久性: 关库重开, 流式读仍成立
//   [5] 删行归还: Delete 后空闲页回升 (blob 链页回到空闲链)

import (
	"fmt"
	"hash/crc32"
	"io"
	"os"
	"path/filepath"
	"runtime"

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

type FileRow struct {
	DB.BaseModel
	Name string `vrtx:"maxlen:64"`
	// 流式载体: 不写 type:blob 也会自动推断 (goTypeToFieldType 里注册了 types.Blob)。
	// 这里显式写出来是为了 demo 的自解释。
	Data types.Blob `vrtx:"type:blob"`
}

type SmallRow struct {
	DB.BaseModel
	Name string `vrtx:"maxlen:32"`
	Body []byte `vrtx:"type:blob"` // 一次性载体: 读回直接是内容
}

func must(err error) {
	if err != nil {
		panic(err)
	}
}

func check(err error) bool {
	if err != nil {
		fmt.Println("   !!", err)
		return false
	}
	return true
}

func memPeakMB() float64 {
	runtime.GC()
	var ms runtime.MemStats
	runtime.ReadMemStats(&ms)
	return float64(ms.HeapAlloc) / 1024 / 1024
}

func main() {
	dir, err := os.MkdirTemp("", "bloborm")
	must(err)
	defer os.RemoveAll(dir)

	dbPath := filepath.Join(dir, "bloborm.db")
	db, err := DB.Open(dbPath)
	must(err)
	db.SetAutoCommitDelay(0) // 手动 Commit, 便于对照
	defer db.Close()

	files, err := DB.UseTable[FileRow](db, "files")
	must(err)
	smalls, err := DB.UseTable[SmallRow](db, "smalls")
	must(err)

	// ------------------------------------------------------------------
	// [1] 流式写: 32MB 内容从临时文件灌进链页, 内存峰值与内容大小无关
	// ------------------------------------------------------------------
	const size = 32 << 20
	srcPath := filepath.Join(dir, "src.bin")
	src := make([]byte, 1<<20)
	for i := range src {
		src[i] = byte(i % 251)
	}
	f, err := os.Create(srcPath)
	must(err)
	wantCRC := uint32(0)
	for i := 0; i < size/len(src); i++ {
		_, err = f.Write(src)
		must(err)
		wantCRC = crc32.Update(wantCRC, crc32.IEEETable, src)
	}
	must(f.Close())

	f, err = os.Open(srcPath)
	must(err)
	before := memPeakMB()
	row := &FileRow{Name: "32mb.bin", Data: types.Blob{Source: f}}
	must(files.Insert(row))
	must(db.Commit())
	must(f.Close())
	after := memPeakMB()
	fmt.Printf("[1] 流式写 32MB: pk=%d, ref={TotalLen:%d FirstPageID:%d}, 内存峰值 %.1f→%.1f MB\n",
		row.ID, row.Data.TotalLen, row.Data.FirstPageID, before, after)

	// ------------------------------------------------------------------
	// [2] 流式读: Open 拿 io.ReadCloser, io.Copy 边读边校验
	// ------------------------------------------------------------------
	got, err := files.Find(row.ID)
	must(err)
	rc, err := got.Data.Open(files.BlobReader())
	must(err)
	gotCRC := uint32(0)
	buf := make([]byte, 64<<10)
	for {
		n, rerr := rc.Read(buf)
		gotCRC = crc32.Update(gotCRC, crc32.IEEETable, buf[:n])
		if rerr == io.EOF {
			break
		}
		must(rerr)
	}
	must(rc.Close())
	fmt.Printf("[2] 流式读 32MB: CRC %s (want %08x)\n",
		cmpCRC(gotCRC, wantCRC), wantCRC)

	// ------------------------------------------------------------------
	// [3] []byte 载体对照: 小内容一把梭, 读回直接是 []byte
	// ------------------------------------------------------------------
	s := &SmallRow{Name: "hello", Body: []byte("hello blob, 一次性载体")}
	must(smalls.Insert(s))
	must(db.Commit())
	back, err := smalls.Find(s.ID)
	must(err)
	fmt.Printf("[3] []byte 载体: 写 %q, 读回 %q\n", s.Body, back.Body)

	// ------------------------------------------------------------------
	// [3.5] Update 不碰 blob 字段 (只改名): 旧链必须原样保留
	//       (Insert 后 ref 已回填进对象, Update 走"读回镜像"分支, 链不动)
	// ------------------------------------------------------------------
	got.Name = "renamed.bin"
	must(files.Update(got))
	must(db.Commit())
	renamed, err := files.Find(got.ID)
	must(err)
	rc3, err := renamed.Data.Open(files.BlobReader())
	must(err)
	renameCRC, err := crc32Reader(rc3)
	must(err)
	must(rc3.Close())
	fmt.Printf("[3.5] 改名 Update 后 blob: %s (%s, %d 字节)\n",
		cmpCRC(renameCRC, wantCRC), renamed.Name, renamed.Data.TotalLen)

	// ------------------------------------------------------------------
	// [4] 持久性: 关库重开, 流式读照样成立
	// ------------------------------------------------------------------
	must(db.Close())
	db2, err := DB.Open(dbPath)
	must(err)
	defer db2.Close()
	files2, err := DB.UseTable[FileRow](db2, "files")
	must(err)
	again, err := files2.Find(row.ID)
	must(err)
	rc2, err := again.Data.Open(files2.BlobReader())
	must(err)
	reopenCRC, err := crc32Reader(rc2)
	must(err)
	must(rc2.Close())
	fmt.Printf("[4] 重开库流式读: CRC %s\n", cmpCRC(reopenCRC, wantCRC))

	// ------------------------------------------------------------------
	// [5] 删行归还: Delete 后空闲页回升 (堆页 + 整条 blob 链)
	// ------------------------------------------------------------------
	core := db2.CoreDB()
	freeBefore := core.DebugFreePageCount()
	must(again.Delete())
	must(db2.Commit())
	freeAfter := core.DebugFreePageCount()
	fmt.Printf("[5] 删 32MB 行: 空闲页 %d -> %d (归还 %d 页 ≈ %d KB)\n",
		freeBefore, freeAfter, freeAfter-freeBefore, (freeAfter-freeBefore)*16)

	fmt.Println("\n=== 全部场景通过 ===")
}

func cmpCRC(got, want uint32) string {
	if got == want {
		return "一致"
	}
	return fmt.Sprintf("不一致! got=%08x", got)
}

func crc32Reader(r io.Reader) (uint32, error) {
	h := crc32.NewIEEE()
	if _, err := io.Copy(h, r); err != nil {
		return 0, err
	}
	return h.Sum32(), nil
}