TsDB2/analyzer/blobwalk.go

package main

// ===== Blob 链走查 (v10 Type=9) =====
//
// v10 起 Blob 列行内是 8B BlobRef [TotalLen BE 4][FirstPageID BE 4], 链页由
// PageHdr.NextPage 串联 (DB/core/page/base_page.go BlobIndex)。行是链的唯一主人:
// 引擎删行 / DropTable 都靠行内引用还链 (table_page.go releaseBlobRefsInDataPage)。
//
// 早期 analyzer 的可达根不含 blob 链 —— 任何带 blob 数据的库, 全部活链页都会被
// 报成"既不属于任何表/索引, 也不在空闲链上"的孤儿 (DockerManager 实测 44 页的主因)。
// 这里把行内引用抽出来走链整链认领, 与引擎 DropTable 的回收口径同源。

import (
	"encoding/binary"

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

// blobRefsOfRow 从一行 (不含 pkID 前缀的 rowData) 抽出所有非空 blob 链首页。
// 行布局解析与引擎 rowBlobRefs (DB/core/blob_read.go) 逐条对照:
// 版本不符 / 偏移表不可信时返回 nil (与引擎"跳过该行"同义, 不让坏行毁掉整批)。
func blobRefsOfRow(t *TInfo, rowData []byte) []uint32 {
	if len(rowData) < 1 {
		return nil
	}
	var sv *VInfo
	for i := range t.Versions {
		if t.Versions[i].Ver == rowData[0] {
			sv = &t.Versions[i]
			break
		}
	}
	if sv == nil || len(sv.Indices) == 0 {
		return nil
	}
	n := len(sv.Indices)
	col0Idx := sv.Indices[0]
	if col0Idx >= len(t.MasterFields) || t.MasterFields[col0Idx].Sz <= 0 {
		return nil
	}
	col0Width := t.MasterFields[col0Idx].Sz
	// v9 行布局: [schemaVer 1B][列0 W0 B][列结束偏移表 2N][列内容...][NULL 位图]
	tblBase := 1 + col0Width
	header := tblBase + 2*n
	if header > len(rowData) {
		return nil
	}
	// 表项 0 必须等于变长内容区起点 (被零填充/冲掉都会在这里暴露)。
	if int(binary.BigEndian.Uint16(rowData[tblBase:tblBase+2])) != header {
		return nil
	}
	bitmapLen := (n + 7) / 8
	ends := make([]int, n)
	prev := header
	for i := 0; i < n; i++ {
		off := int(binary.BigEndian.Uint16(rowData[tblBase+2*i:tblBase+2*i+2])) & rowVarOffMask
		if off < prev || off > len(rowData) {
			return nil
		}
		ends[i] = off
		prev = off
	}
	if prev+bitmapLen > len(rowData) {
		return nil
	}
	bitmap := rowData[prev : prev+bitmapLen]

	var out []uint32
	for i, idx := range sv.Indices {
		if idx >= len(t.MasterFields) {
			return nil
		}
		f := t.MasterFields[idx]
		if f.Dropped || f.Type != byte(types.FieldType_Blob) {
			continue
		}
		if bitmap[i>>3]&(1<<uint(i&7)) != 0 {
			continue // NULL 列不占内容字节
		}
		start := 1
		if i > 0 {
			start = ends[i-1]
		}
		end := tblBase
		if i > 0 {
			end = ends[i]
		}
		if end-start < 8 || start+8 > len(rowData) {
			continue
		}
		if ref := binary.BigEndian.Uint32(rowData[start+4 : start+8]); ref != 0 {
			out = append(out, ref)
		}
	}
	return out
}

// hasBlobColumn 该表是否有未删除的 blob 列 (没有就跳过链走查)。
func hasBlobColumn(t *TInfo) bool {
	for i := range t.MasterFields {
		f := &t.MasterFields[i]
		if !f.Dropped && f.Type == byte(types.FieldType_Blob) {
			return true
		}
	}
	return false
}

// walkTableBlobChains 扫一张表数据页里每行的 blob 引用, 把链页 (Type=9) 整链认领。
// 返回 (链数, 链页数)。
//
// ⭐ 数据页集合来自【行存储 B+ 树叶条目】的 dpID 引用 —— btree 行存储的数据页
// 由叶条目 (页号+槽位) 引用, FirstData 链是空的 (不再维护); 走 FirstData 链
// 会扫不到任何行, 活链全被误报孤儿。
//
// 数据页迭代复用 eachRow (indexcheck.go): 它自带 PG_DATA 类型校验, 时序表的
// TimeDataPage (布局不同) 会被整体跳过 —— 时序表 blob 列的链页因此仍会被报孤儿,
// 与修复前一致, 不产生新的误报 (时序表是否允许 blob 列, 引擎未禁止, 属于已知缺口)。
func walkTableBlobChains(w *walker, t *TInfo) (chains, pages int) {
	if !hasBlobColumn(t) {
		return 0, 0
	}
	// 行存储 = 主键 B+ 树 (时序表的 TimeDataPage 布局不同, 维持已知缺口)。
	var rowStore *IInfo
	for i := range t.Idxs {
		if isRowStore(t.Idxs[i].Type) && isBTreeIdx(t.Idxs[i].Type) {
			rowStore = &t.Idxs[i]
			break
		}
	}
	if rowStore == nil {
		return 0, 0
	}
	dps := map[uint32]bool{}
	pb := newPageBuf()
	queue := []uint32{rowStore.EP}
	seen := map[uint32]bool{}
	for len(queue) > 0 {
		pid := queue[0]
		queue = queue[1:]
		if pid == 0 || seen[pid] {
			continue
		}
		seen[pid] = true
		if pid < 2 || int(pid) > w.total {
			continue
		}
		b, err := pb.page(pid)
		if err != nil || b[0] != PG_IDATA {
			continue
		}
		cn := int(le32(b[IX_CNT : IX_CNT+4]))
		el := int(le16(b[IX_ELEN : IX_ELEN+2]))
		if el <= 4 {
			el = 12
		}
		if b[IX_LEAF] == 1 {
			refOff := rowStoreRefOff(rowStore.Type, el)
			if refOff >= 0 && refOff+4 <= el {
				collect := func(nb []byte) {
					n := int(le32(nb[IX_CNT : IX_CNT+4]))
					for i := 0; i < n; i++ {
						o := IDX_HDR + i*el
						if o+el > len(nb) {
							break
						}
						dps[be32(nb[o+refOff:o+refOff+4])] = true
					}
				}
				collect(b)
				// 叶同层链: 与 indexPages 一致顺链扫完
				for nx := le32(b[PH_NEXT : PH_NEXT+4]); nx != 0; {
					if seen[nx] || nx < 2 || int(nx) > w.total {
						break
					}
					seen[nx] = true
					nb, err := pb.page(nx)
					if err != nil || nb[0] != PG_IDATA {
						break
					}
					collect(nb)
					nx = le32(nb[PH_NEXT : PH_NEXT+4])
				}
			}
			continue
		}
		for i := 0; i < cn; i++ {
			o := IDX_HDR + i*el
			if o+el > len(b) {
				break
			}
			child := be32(b[o+el-4 : o+el])
			if child != 0 {
				queue = append(queue, child)
			}
		}
	}
	for dp := range dps {
		if dp < 2 || int(dp) > w.total {
			continue
		}
		b, err := pb.page(dp)
		if err != nil || b[0] != PG_DATA {
			continue
		}
		eachRow(b, func(_ uint64, off, ln int) {
			for _, first := range blobRefsOfRow(t, b[off:off+ln]) {
				chains++
				pages += w.chain(first, "blob", t.Name)
			}
		})
	}
	return chains, pages
}