TsDB2/DB/core/blob_read.go

package core

import (
	"encoding/binary"
	"errors"
	"io"

	"github.com/golang/snappy"

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

// ReadBlob types.BlobReader 接口实现 — 给 ORM 在读回 Blob 后拿 io.ReadCloser.
//
// 行为: 与 BlobRead 内部走同一份链页 / 解码路径, 区别只在"按流 / 按块":
//   - BlobRead 返 []byte, 适合一次性读完的小内容;
//   - ReadBlob 返 io.ReadCloser, 适合"调方按需读、不全拉到内存"的大内容与流式对接 (用户
//     可在 io.Copy 中途断流)。
//
// chainReader 已经在跨页时按 PageSize 缓冲, 不需要再包一层 bufio。
// 关链逻辑: chainReader 不持有 fd / 内存分配, 直接 NopCloser 即可。
func (t *Table) ReadBlob(ref types.BlobRef, compress types.FieldCompress) (io.ReadCloser, error) {
	if t == nil || t.db == nil {
		return nil, errors.New("ReadBlob: table is nil")
	}
	if !compress.IsValid() {
		return nil, errors.New("ReadBlob: unknown compress option")
	}
	if ref.Empty() {
		return nil, nil
	}
	chainR, err := newBlobChainReader(t.db, ref.FirstPageID)
	if err != nil {
		return nil, err
	}
	if compress == types.CompressSnappy {
		return io.NopCloser(snappy.NewReader(chainR)), nil
	}
	return io.NopCloser(chainR), nil
}

// 静态断言: *Table 实现了 types.BlobReader 接口, 防止接口签名漂移。
var _ types.BlobReader = (*Table)(nil)

// BlobRead 把 BlobRef 引用的整条链读回, 自动按 compress 解压.
//
// 返回的是原始字节 (解压后), 长度 == ref.TotalLen, 任何不一致都返 ErrBlobCorrupted.
//
// 内存: 全读到内存 (io.ReadAll). Snappy 是块解压, 需要看到全部数据才能解, 无法流式;
// 对极大 blob (>1GB) 请改用 Seek+部分读 API (v2 TODO).
func (t *Table) BlobRead(ref BlobRef, compress types.FieldCompress) ([]byte, error) {
	if t == nil || t.db == nil {
		return nil, errors.New("BlobRead: table is nil")
	}
	if !compress.IsValid() {
		return nil, errors.New("BlobRead: unknown compress option")
	}
	if ref.Empty() {
		return nil, nil
	}

	chainR, err := newBlobChainReader(t.db, ref.FirstPageID)
	if err != nil {
		return nil, err
	}

	var data []byte
	if compress == types.CompressSnappy {
		// snappy.NewReader 是流式 API: 内部缓冲, 但 Read 返 EOF 后才能确认解压完整.
		// io.ReadAll 一次性读完, 与解码要求一致 (Snappy 帧必须见尾才知道解压完).
		sr := snappy.NewReader(chainR)
		data, err = io.ReadAll(sr)
	} else {
		data, err = io.ReadAll(chainR)
	}
	if err != nil {
		return nil, err
	}
	if uint32(len(data)) != ref.TotalLen {
		return nil, ErrBlobCorrupted
	}
	return data, nil
}

// BlobFree 把整条 blob 链归还空闲链表.
//
// 行删除/更新/DropTable 都需要释放旧 blob; 否则链页成为永久泄漏.
//
// 时序: 在事务里调用, 走 DeferFreeChain 路径, Commit 成功才真正还链, Rollback 丢弃.
// 无事务上下文时直写空闲链 (与 DeferFreePage 在 db==nil 时的行为一致).
func (t *Table) BlobFree(ref BlobRef) error {
	if ref.Empty() {
		return nil
	}
	if t == nil || t.db == nil {
		return errors.New("BlobFree: table is nil")
	}

	// 1. 走链, 找到尾页 + 计链长.
	cur := ref.FirstPageID
	tail := cur
	count := uint32(0)
	buf := make([]byte, PageSize)
	for cur != 0 {
		if err := t.db.file.ReadPage(cur, buf); err != nil {
			return err
		}
		tail = cur
		count++
		if count > BlobChainMax {
			return ErrBlobCorrupted // 链长不可能这么长, 拒绝继续走
		}
		cur = pageHdrNextPage(buf)
	}

	// 2. 整链回收 (与现有 FreeChain 同一接口).
	t.db.deferFreeBlobChain(ref.FirstPageID, tail, count)
	return nil
}

// pageHdrNextPage 读 PageHdr.NextPage, 避免依赖 page 包类型 (本文件独立可读).
//
// ★ 必须按小端解: PageHdr 是 unsafe 结构体直映射落盘 (page/hdr.go), NextPage 字段
// 按本机字节序 (小端) 序列化; 写侧 blob_writer.go 与读侧 blob_reader.go 也都走
// page.HdrOf 直读写。这里曾按大端解 → 多页链的第二页起 NextPage 被读成 0x0N000000,
// BlobFree 走链找尾页时 ReadPage 越界报错, 而错误被 releaseRowBlobRefs 的
// "放任不中断"语义吞掉 —— 行删了、链页却永不归还 (单页链 NextPage=0 不受影响,
// 所以小 blob 测试全绿, 唯独跨页大 blob 泄漏)。
func pageHdrNextPage(buf []byte) uint32 {
	const off = 12 // PageHdr.NextPage 在头内偏移
	return uint32(buf[off]) | uint32(buf[off+1])<<8 |
		uint32(buf[off+2])<<16 | uint32(buf[off+3])<<24
}

// deferFreeBlobChain 走事务的 pendingFreeChain 路径 (Commit 后真正还链).
func (db *DB) deferFreeBlobChain(head, tail, count uint32) {
	if head == 0 || count == 0 {
		return
	}
	if txn := db.currentTxn.Load(); txn != nil {
		txn.pendingFreeChain = append(txn.pendingFreeChain,
			pendingChain{head: head, tail: tail, count: count})
		return
	}
	db.pendingMu.Lock()
	db.pendingFreeChain = append(db.pendingFreeChain,
		pendingChain{head: head, tail: tail, count: count})
	db.pendingMu.Unlock()
}

// rowBlobRefs 扫描行内 blob 列, 收集非空 BlobRef.
//
// 引擎层在 DeleteRow / UpdateRow / DropTable 时需要把旧行的 blob 链还掉.
// 行内 blob 列的 8B BlobRef 没有 RowVarTag (内联内容), 必须按列类型定位, 不像堆引
// 用可以走 RowFieldRefTagged.
//
// 返回 []BlobRef (空切片表示"无 blob 列"或"全 NULL"); 行布局解析失败时返 nil.
//
// 列级顺序与 FieldIndices 一致, 调用方按位置 / master 索引映射.
func (table *Table) rowBlobRefs(rowData []byte) []BlobRef {
	if len(rowData) < 1 {
		return nil
	}
	sv := table.getVersion(rowData[0])
	if sv == nil || len(sv.FieldIndices) == 0 {
		return nil
	}
	n := len(sv.FieldIndices)
	col0Width := table.svCol0Width(sv)
	ends, ok := types.RowParseOffsets(rowData, n, col0Width)
	if !ok {
		return nil
	}
	bitmap, ok := types.RowNullBitmap(rowData, ends, n)
	if !ok {
		return nil
	}

	var refs []BlobRef
	for i, idx := range sv.FieldIndices {
		f := table.Field[idx]
		if f.IsDropped() || !types.IsBlob(f.FieldType) {
			continue
		}
		if types.NullBitSet(bitmap, i) {
			continue
		}
		start := int(types.RowFieldStart(ends, i))
		end := int(types.RowFieldEnd(ends, i, col0Width))
		if end-start < types.BlobRefSize || start+types.BlobRefSize > len(rowData) {
			continue
		}
		seg := rowData[start : start+types.BlobRefSize]
		ref := BlobRef{
			TotalLen:    binary.BigEndian.Uint32(seg[:4]),
			FirstPageID: binary.BigEndian.Uint32(seg[4:8]),
		}
		if ref.Empty() {
			continue
		}
		refs = append(refs, ref)
	}
	return refs
}

// releaseRowBlobRefs 把 row 中所有 blob 链归还空闲链表 (DeleteRow / DropTable /
// Update 替换旧值时调用).
//
// 单独走这条路径而不是合并进 heapReleaseRow: blob 链在文件上是"链页"序列, 释放走
// deferFreeBlobChain, 与 heap 页 (PageID + RefCount) 不是同一条释放路径.
//
// 错误采用"放任单列失败不影响后续列"语义: DropTable 要把所有 blob 都还掉, 一列
// 失败而中断会把后面的全漏, 与"全部回收"的承诺背离. 调用方最后会再走 defrag/
// 复检; 单条失败可被日志/计数器观察, 但不会留下半回收状态.
//
// 行布局不可信 / 表结构已变时返错误但仍继续归还本轮已扫到的 blob.
func (table *Table) releaseRowBlobRefs(rowData []byte) error {
	refs := table.rowBlobRefs(rowData)
	if refs == nil {
		if len(rowData) == 0 || rowData[0] == 0 {
			return nil // 空行 / 零版本 → 跳过
		}
		// 布局解析失败, 但仍可能是合法的"无 blob 列" — 行/表不匹配时 rowBlobRefs 返 nil
		// (其实有合法返回) 还是错乱都让这条路径容错过去, 不让一个坏行毁掉整批回收.
	}
	var firstErr error
	for _, ref := range refs {
		if err := table.BlobFree(ref); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// oldBlobRefsByMaster 与 rowBlobRefs 类似, 但按 masterIdx (字段 master 注册表下标)
// 索引 ——
//
//	Update 路径 (serializeRowInternal) 按 masterIdx 决定是否释放旧 ref, 这是按列访问,
//	与 rowBlobRefs 返回的"按 FieldIndices 下标"的扁平切片不匹配, 单独提供这条.
//
// 行内 NULL / 空 blob 列不出现在返回 map 中 ——
//
//	"old == empty ref"在行里语义上等价于"该列没有旧 ref", 不需要释放.
func (table *Table) oldBlobRefsByMaster(rowData []byte) map[int]BlobRef {
	if len(rowData) < 1 {
		return nil
	}
	sv := table.getVersion(rowData[0])
	if sv == nil || len(sv.FieldIndices) == 0 {
		return nil
	}
	n := len(sv.FieldIndices)
	col0Width := table.svCol0Width(sv)
	ends, ok := types.RowParseOffsets(rowData, n, col0Width)
	if !ok {
		return nil
	}
	bitmap, ok := types.RowNullBitmap(rowData, ends, n)
	if !ok {
		return nil
	}

	var refs map[int]BlobRef
	for i, idx := range sv.FieldIndices {
		f := table.Field[idx]
		if f.IsDropped() || !types.IsBlob(f.FieldType) {
			continue
		}
		if types.NullBitSet(bitmap, i) {
			continue
		}
		start := int(types.RowFieldStart(ends, i))
		end := int(types.RowFieldEnd(ends, i, col0Width))
		if end-start < types.BlobRefSize || start+types.BlobRefSize > len(rowData) {
			continue
		}
		seg := rowData[start : start+types.BlobRefSize]
		ref := BlobRef{
			TotalLen:    binary.BigEndian.Uint32(seg[:4]),
			FirstPageID: binary.BigEndian.Uint32(seg[4:8]),
		}
		if ref.Empty() {
			continue
		}
		if refs == nil {
			refs = make(map[int]BlobRef, n)
		}
		refs[int(idx)] = ref
	}
	return refs
}