TsDB2/DB/core/blob_writer.go

package core

import (
	"errors"

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

// blobChainWriter 把 io.Writer.Write 调用直接转为"写 blob 链尾", 自动跨页 NewPage。
//
// 约束:
//   - 一页 = 一个 blob chunk (与 HeapPage 那种"一页多块"完全不同);
//   - 链由 PageHdr.NextPage 串, 最后一页 NextPage = 0;
//   - 不维护 RefCount (单 owner, 删除时整链 FreeChain 一次 O(1) 还链)。
//
// 所有页头统一用 PageHdr (共 20 字节, 与 heap/index/schema-ext 一致):
// PageHdr 占每页前 20 字节, 后续 Data 区就是 raw blob 字节, 无 chunk 头、无长度前缀。
//
// 错误:
//   - ErrBlobTooLarge: 单 blob 链长超过 BlobChainMax (uint32 上限);
//
//	调用方应中止 io.Copy 并 tx.Rollback(), 由事务模型自动还链。
type blobChainWriter struct {
	db         *DB
	firstPage  uint32 // 链首
	tailPage   uint32 // 链尾 (Write 时优先填这一页, 满了再 NewPage)
	tailUsed   uint16 // tailPage.Data 区已用字节
	chainLen   uint32 // 已分配的页数 (含第一页)
	totalBytes uint64 // 已写入字节数 (= TotalLen)
}

// newBlobChainWriter 预分配第一页并把 writer 初始化到该页。
//
// 之所以必须"先 NewPage 再 Write": snappy.Writer 在第一次 Write 之前会先写一个
// stream header (几字节), 如果第一页还没 NewPage, snappy 的 Write 会落到 nil tailPage
// 上 panic。预开第一页后, 写入路径与 snappy 不耦合。
//
// 参数是 *DB 而不是 *Transaction: 链页的分配 (NewPage) 与写盘 (file.WritePage) 都不需要
// 事务参与 —— 事务只是在有 currentTxn 时顺带把新页登记进 allocated (回滚时归还)。
// 于是"没有显式事务"的写入路径 (core 层 InsertRow 直接塞 []byte, 见 Table.blobFieldRef)
// 也能复用同一个 writer。
func newBlobChainWriter(db *DB) (*blobChainWriter, error) {
	pid := db.NewPage()
	buf := make([]byte, PageSize)
	hdr := page.HdrOf(buf)
	hdr.Init(page.BlobIndex, pid)
	hdr.NextPage = 0
	hdr.UseSize = page.PageHdrSize
	if err := db.file.WritePage(pid, buf); err != nil {
		return nil, err
	}
	return &blobChainWriter{
		db:         db,
		firstPage:  pid,
		tailPage:   pid,
		tailUsed:   0,
		chainLen:   1,
		totalBytes: 0,
	}, nil
}

// Write 实现 io.Writer. 一次 Write 可能跨多页, 由内部 NewPage 衔接.
//
// 返回值语义: 总是返回 (len(p), nil) 或 (已写入字节, err).
// 错误只可能是 ErrBlobTooLarge, 一旦发生, 链已被 NewPage 占用,
// 调用方应中止 io.Copy 并 tx.Rollback() 走事务模型回收整链.
func (w *blobChainWriter) Write(p []byte) (int, error) {
	const pageData = uint32(PageSize - page.PageHdrSize) // 16364
	written := 0
	for written < len(p) {
		// 1. 当前尾页已满? 开新页并挂链
		if uint32(w.tailUsed) >= pageData {
			if w.chainLen >= BlobChainMax {
				return written, ErrBlobTooLarge
			}
			newPID := w.db.NewPage()
			// 挂到老尾页的 NextPage
			tailBuf := make([]byte, PageSize)
			if err := w.db.file.ReadPage(w.tailPage, tailBuf); err != nil {
				return written, err
			}
			page.HdrOf(tailBuf).NextPage = newPID
			if err := w.db.file.WritePage(w.tailPage, tailBuf); err != nil {
				return written, err
			}
			// 初始化新页
			newBuf := make([]byte, PageSize)
			hdr := page.HdrOf(newBuf)
			hdr.Init(page.BlobIndex, newPID)
			hdr.NextPage = 0
			hdr.UseSize = page.PageHdrSize
			if err := w.db.file.WritePage(newPID, newBuf); err != nil {
				return written, err
			}
			// 推进尾指针
			w.tailPage = newPID
			w.tailUsed = 0
			w.chainLen++
		}

		// 2. 计算本批可写字节数 (min(本页剩余空间, p 剩余字节))
		space := uint32(pageData - uint32(w.tailUsed))
		remaining := uint32(len(p) - written)
		n := space
		if remaining < n {
			n = remaining
		}

		// 3. 读尾页 + 写本批
		tailBuf := make([]byte, PageSize)
		if err := w.db.file.ReadPage(w.tailPage, tailBuf); err != nil {
			return written, err
		}
		copy(tailBuf[page.PageHdrSize+uint32(w.tailUsed):], p[written:written+int(n)])
		page.HdrOf(tailBuf).UseSize = page.PageHdrSize + uint32(w.tailUsed) + n
		if err := w.db.file.WritePage(w.tailPage, tailBuf); err != nil {
			return written, err
		}

		w.tailUsed += uint16(n)
		written += int(n)
		w.totalBytes += uint64(n)
	}
	return written, nil
}

// ErrBlobTooLarge 单 blob 链长超过 BlobChainMax (256K 页 = ~4GB).
var ErrBlobTooLarge = errors.New("blob chain exceeds maximum length")

// BlobChainMax 单 blob 链页数上限 (uint32 计数).
//
// 上限由来: 链长用 uint32 计数 (避免 RowFieldRef 里的 2 字节截断隐患),
// 单页 16KB, 256K 页 ≈ 4GB. 这个量级对绝大多数 binary 大对象够用;
// 超大文件 (>4GB) 应改用对象存储而非单字段.
const BlobChainMax uint32 = 256 * 1024

// BlobRef 行内 8 字节引用. 类型权威定义见 types.BlobRef (本类型为 alias, 历史代码兼容)。
//
//	[TotalLen 4B BE][FirstPageID 4B BE]
//
//	TotalLen = 原始 (解压后) 字节数;
//	FirstPageID = 0 表示空 blob (与 TotalLen = 0 一同出现).
//
// 【为什么要 alias】types 包持有 BlobRef 是为了让 EncodeValue / DecodeValue 不必反向依赖
// core 包 (core 已经依赖 types, 反过来会循环)。core 内的历史 API (BlobCommit / BlobRead
// / BlobFree 等) 全部继续用这个名字, 直接 type 别名过去, 调用方代码零改动。
//
// types.BlobRef 已定义 Empty / SizeBytes 方法, alias 自动继承, 不在这里再写一次
// (否则是 duplicate method)。
type BlobRef = types.BlobRef