package core import ( "encoding/binary" "errors" "io" "github.com/golang/snappy" "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 包类型 (本文件独立可读). func pageHdrNextPage(buf []byte) uint32 { const off = 12 // PageHdr.NextPage 在头内偏移 return uint32(buf[off])<<24 | uint32(buf[off+1])<<16 | uint32(buf[off+2])<<8 | uint32(buf[off+3]) } // deferFreeBlobChain 走事务的 pendingFreeChain 路径 (Commit 后真正还链). func (db *DB) deferFreeBlobChain(head, tail, count uint32) { if head == 0 || count == 0 { return } if db.currentTxn != nil { db.currentTxn.pendingFreeChain = append(db.currentTxn.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 }