TsDB2/DB/core/table_page.go

package core

import (
	"encoding/binary"
	"fmt"
	"unsafe"

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

type TablePage struct {
	hdr    page.PageHdr
	db     *DB // 反指所属 DB, 用于建表时给 Table 注入
	tables []*Table

	// schemaExt 全库共享的 schema 扩展页分配器 (链头记在表页第 1 页 PageHdr 之后的 4 字节)
	schemaExt *SchemaExtAllocator

	handle        page.HandleDb
	hdrSize       uint32 // hdr 的大小
	tableInfoSize uint32 // 表头大小
	lastPageID    uint32 // 最后一页
}

// bindDB 给自己与【已经从盘上加载出来的表】注入 db 指针。
//
// 为什么需要回填: 表的加载发生在 Open 的中途 (tablePage.Init), 那时 DB 结构体还没构造
// 出来, 加载出来的 *Table.db 只能是 nil。而 BlobRead / BlobFree / 事务外的 blob 落链
// (blobCommit) 都要靠它拿 file / 分配页 —— 不回填的话, 库关掉再打开之后读 blob 会直接
// 报 "table is nil" (新表看不出来, 因为它们走 Create, 那时 db 已经在了)。
func (t *TablePage) bindDB(db *DB) {
	t.db = db
	for _, tbl := range t.tables {
		if tbl.db == nil {
			tbl.db = db
		}
	}
}

// NewTablePage 创建表页管理器
func NewTablePage(handle page.HandleDb) *TablePage {
	return &TablePage{
		handle:        handle,
		hdrSize:       uint32(unsafe.Sizeof(page.PageHdr{})),
		tableInfoSize: uint32(unsafe.Sizeof(TableInfo{})),
	}
}

// ensureSchemaExt 惰性创建共享扩展页分配器
func (t *TablePage) ensureSchemaExt() *SchemaExtAllocator {
	if t.schemaExt == nil {
		t.schemaExt = newSchemaExtAllocator(t.handle, 0,
			func() []*Table { return t.tables },
			t.saveSchemaExtHead)
	}
	return t.schemaExt
}

// saveSchemaExtHead 把共享扩展页链头写回表页第 1 页 PageHdr 之后的 4 字节
func (t *TablePage) saveSchemaExtHead(pageID uint32) error {
	if t.hdr.PageID == 0 {
		return nil
	}
	buf := make([]byte, PageSize)
	if err := t.handle.ReadPage(t.hdr.PageID, buf); err != nil {
		return err
	}
	binary.BigEndian.PutUint32(buf[t.hdrSize:t.hdrSize+schemaExtHeadSize], pageID)
	return t.handle.SavePage(t.hdr.PageID, buf)
}

// Tables 获取所有表
func (t *TablePage) Tables() []*Table {
	return t.tables
}

// LastPageID 获取最后页ID
func (t *TablePage) LastPageID() uint32 {
	return t.lastPageID
}

// Init 初始化表页：从磁盘读取所有表 + 所有字段（支持跨页存储）
func (t *TablePage) Init(pageID uint32) {
	t.tables = make([]*Table, 0)
	t.hdrSize = uint32(unsafe.Sizeof(page.PageHdr{}))
	t.tableInfoSize = uint32(unsafe.Sizeof(TableInfo{}))
	t.hdr.PageID = pageID
	t.hdr.Type = page.TableIndex
	t.hdr.NextPage = 0
	t.lastPageID = 0

	if pageID == 0 {
		return
	}

	// 表页第 1 页 PageHdr 之后的 4 字节 = 共享 schema 扩展页链头
	first := make([]byte, PageSize)
	extPID := uint32(0)
	if err := t.handle.ReadPage(pageID, first); err == nil {
		extPID = binary.BigEndian.Uint32(first[t.hdrSize : t.hdrSize+schemaExtHeadSize])
	}
	t.ensureSchemaExt()
	t.schemaExt.firstPID = extPID

	for pageID != 0 {
		t.lastPageID = pageID
		pageBuf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pageID, pageBuf); err != nil {
			break
		}
		pageHdr := page.HdrOf(pageBuf)
		// 页类型 / 自述页号校验: 表页链同样可能走到"尚未初始化的新页" (或被回收复用的页),
		// 对不上就说明这不是表页, 必须立刻停止解析 —— 否则页里的旧内容会被当成 TableInfo,
		// 解析出的垃圾 FirstIndexID/FirstData 会一路传到页回收路径上。
		if pageHdr.Type != page.TableIndex || pageHdr.PageID != pageID {
			break
		}
		offset := t.hdrSize + schemaExtHeadSize

		for offset+t.tableInfoSize <= pageHdr.UseSize {
			info := (*TableInfo)(unsafe.Pointer(&pageBuf[offset]))
			if util.IsHighBitSet(info.Flag) {
				offset += t.tableInfoSize + uint32(info.FieldAllocateSize)
				continue
			}

			activeVer, masterFields, versions := readSchema(
				t.handle, pageID, uint16(offset+t.tableInfoSize), info.FieldAllocateSize,
				info.ExtPage, info.ExtOff)
			indexMate := &IndexMetaPage{
				handle: t.handle,
			}
			indexMate.Init(info.FirstIndexID)
			t.tables = append(t.tables, &Table{
				info:             info,
				Field:            masterFields,
				db:               t.db, // 见 bindDB: 这里通常还是 nil, Open 结束前会回填
				handle:           t.handle,
				indexMeta:        indexMate,
				tablePageID:      pageID,
				tableStartOffset: offset,
				schemaExt:        t.schemaExt,
				activeVersion:    activeVer,
				versions:         versions,
			})

			offset += t.tableInfoSize + uint32(info.FieldAllocateSize)
		}
		pageID = pageHdr.NextPage
	}
}

// ===========================================================================
// 表页槽位管理
// ===========================================================================
//
// 表页链上的记录布局 (所有页统一):
//
//	[PageHdr 20B][SchemaExtPageID 4B(仅第 1 页用)][TableInfo 104B + 内联 schema 块]...
//
// 每条记录占 tableInfoSize + FieldAllocateSize 字节, 长度由记录自身描述。于是
// Drop 只要把 Flag 置高位, 这条记录就变成"可复用空洞", 遍历方按长度自然跳过它。
//
// Create 的落点优先级:
//  1. 尾页尾部追加            最省事, 也不打乱页内记录顺序
//  2. 复用某个空洞 (first-fit) 空洞够大时按需求切分, 余量补一条墓碑记录
//  3. 压缩整条表页链后重试     碎片太多时把空洞合并回尾部, 顺带回收空页
//  4. 追加新表页               真放不下才让链变长
//
// 有了这一层, "批量建表 → 删表 → 再建表" 的循环不会让表页链无限膨胀。

// tableSlot 一段可写入表记录的槽位
type tableSlot struct {
	pageID   uint32 // 所在页
	offset   uint32 // TableInfo 起始偏移
	capacity uint32 // 槽位容量 (>= 需求); 复用空洞时可能大于需求
	bump     bool   // 位于尾页尾部 → 需要推进 UseSize
}

// allocSlot 为 required 字节的表记录挑选落点
func (t *TablePage) allocSlot(required uint32) (tableSlot, error) {
	// 全库第一张表: 先把链头页建出来
	if t.hdr.PageID == 0 {
		pid := t.handle.NewPage()
		t.InitHeader(pid)
		t.hdr.PageID = pid
		t.lastPageID = pid
	}

	// 1) 尾页尾部直接追加
	if s, ok, err := t.tailSlot(required); err != nil {
		return tableSlot{}, err
	} else if ok {
		return s, nil
	}

	// 2) 复用 Drop 留下的空洞
	if s, ok, err := t.findFreeSlot(required); err != nil {
		return tableSlot{}, err
	} else if ok {
		return s, nil
	}

	// 3) 单块空洞都装不下 → 压缩整条链, 把碎片合并回尾部再试
	if t.hasHoles() {
		if err := t.Compact(); err != nil {
			return tableSlot{}, err
		}
		if s, ok, err := t.tailSlot(required); err != nil {
			return tableSlot{}, err
		} else if ok {
			return s, nil
		}
	}

	// 4) 真放不下: 追加新表页
	newPID, err := t.appendPage()
	if err != nil {
		return tableSlot{}, err
	}
	return tableSlot{
		pageID:   newPID,
		offset:   t.hdrSize + schemaExtHeadSize,
		capacity: required,
		bump:     true,
	}, nil
}

// tailSlot 尾页尾部是否还剩 required 字节
func (t *TablePage) tailSlot(required uint32) (tableSlot, bool, error) {
	if t.lastPageID == 0 {
		return tableSlot{}, false, nil
	}
	buf := make([]byte, PageSize)
	if err := t.handle.ReadPage(t.lastPageID, buf); err != nil {
		return tableSlot{}, false, err
	}
	off := page.HdrOf(buf).UseSize
	if off < t.hdrSize+schemaExtHeadSize {
		off = t.hdrSize + schemaExtHeadSize
	}
	if off+required > PageSize {
		return tableSlot{}, false, nil
	}
	return tableSlot{pageID: t.lastPageID, offset: off, capacity: required, bump: true}, true, nil
}

// findFreeSlot 在整条表页链上找第一个容量足够的空洞 (first-fit)
func (t *TablePage) findFreeSlot(required uint32) (tableSlot, bool, error) {
	for pid := t.hdr.PageID; pid != 0; {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pid, buf); err != nil {
			return tableSlot{}, false, err
		}
		hdr := page.HdrOf(buf)
		for off := t.hdrSize + schemaExtHeadSize; off+t.tableInfoSize <= hdr.UseSize; {
			info := (*TableInfo)(unsafe.Pointer(&buf[off]))
			size := t.tableInfoSize + uint32(info.FieldAllocateSize)
			if off+size > hdr.UseSize {
				break
			}
			if util.IsHighBitSet(info.Flag) && size >= required {
				return tableSlot{pageID: pid, offset: off, capacity: size}, true, nil
			}
			off += size
		}
		pid = hdr.NextPage
	}
	return tableSlot{}, false, nil
}

// hasHoles 表页链上是否还存在 Drop 留下的空洞
func (t *TablePage) hasHoles() bool {
	for pid := t.hdr.PageID; pid != 0; {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pid, buf); err != nil {
			return false
		}
		hdr := page.HdrOf(buf)
		for off := t.hdrSize + schemaExtHeadSize; off+t.tableInfoSize <= hdr.UseSize; {
			info := (*TableInfo)(unsafe.Pointer(&buf[off]))
			if util.IsHighBitSet(info.Flag) {
				return true
			}
			off += t.tableInfoSize + uint32(info.FieldAllocateSize)
		}
		pid = hdr.NextPage
	}
	return false
}

// appendPage 在链尾追加一张空表页, 并挂到前驱的 NextPage 上
func (t *TablePage) appendPage() (uint32, error) {
	pid := t.handle.NewPage()
	t.InitHeader(pid)
	if t.lastPageID != 0 {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(t.lastPageID, buf); err != nil {
			return 0, err
		}
		page.HdrOf(buf).NextPage = pid
		if err := t.handle.SavePage(t.lastPageID, buf); err != nil {
			return 0, err
		}
	}
	t.lastPageID = pid
	return pid, nil
}

// checkFieldCount 校验列数能否落盘 (建表与加列共用同一条闸门)。
//
// 列数上限见 MaxFieldsPerTable: FieldCount 与字段索引都是 1 字节, 超了会回绕 / 截断。
// 错误里带上"本次多少列、上限多少列", 只说"字段数量非法"用户不知道要砍到几列。
func checkFieldCount(n int) error {
	if n <= 0 {
		return fmt.Errorf("%w: 一张表至少要有 1 列", ErrInvalidFieldCount)
	}
	if n > MaxFieldsPerTable {
		return fmt.Errorf("%w: 一张表最多 %d 列, 本次 %d 列 "+
			"(列数与字段索引都是 1 字节, 超过会截断成错位 schema)",
			ErrInvalidFieldCount, MaxFieldsPerTable, n)
	}
	return nil
}

// ===========================================================================
// ✅ Create 完整正确（字段数量全是 byte，255上限）
// ===========================================================================
func (t *TablePage) Create(name string, fields []types.FieldInfo) (*Table, error) {
	if len(name) > TableNameSize {
		return nil, fmt.Errorf("%w: 表名 %q 有 %d 字节, 上限 %d",
			ErrTableNameTooLong, name, len(name), TableNameSize)
	}
	for _, tb := range t.tables {
		if util.BytesToString(tb.info.Name[:]) == name && !util.IsHighBitSet(tb.info.Flag) {
			return nil, ErrTableExists
		}
	}
	if err := checkFieldCount(len(fields)); err != nil {
		return nil, err
	}
	// 建表即校验列级参数与行宽。
	//
	// 行宽是 schema 属性而不是行数据属性: 一张建出来就超上限的表, 之后每一行都会写入
	// 失败, 而且是在插数据时才暴露。必须在建表这一步拒绝, 并把"哪几列占了多少"报出来。
	// 校验放在 core 而不是只留在 ORM/migrate: schema 也能从 SQL 或低层 API 直接建。
	if err := validateFieldsForSchema(fields); err != nil {
		return nil, err
	}

	// 过完上面两道闸门后 len(fields) ≤ MaxFieldsPerTable(254), 转 byte 不会回绕。
	fieldCount := byte(len(fields))

	// 内联块按 schema 实际序列化长度精确分配 (不再 ×2 预分配, 也不加 256B 下限)。
	// 超出 MaxInlineSchemaSize 的部分截断, 由 writeSchema 写进共享扩展页。
	masterIndices := make([]byte, fieldCount)
	for i := range masterIndices {
		masterIndices[i] = byte(i)
	}
	initialVersions := []SchemaVersion{{Version: 1, FieldIndices: masterIndices}}
	probe := &Table{Field: fields, activeVersion: 1, versions: initialVersions}
	allocateSize := len(probe.serializeSchema())
	if allocateSize > MaxInlineSchemaSize {
		allocateSize = MaxInlineSchemaSize
	}
	if allocateSize <= 0 || uint32(allocateSize) > PageSize {
		return nil, ErrFieldAllocateTooLarge
	}

	// 这个表总共需要多少空间
	required := t.tableInfoSize + uint32(allocateSize)

	// 挑选槽位: 尾页尾部 → 复用 Drop 空洞 → 压缩后重试 → 新页
	slot, err := t.allocSlot(required)
	if err != nil {
		return nil, err
	}
	pageBuf := make([]byte, PageSize)
	if err := t.handle.ReadPage(slot.pageID, pageBuf); err != nil {
		return nil, err
	}
	pageHdr := page.HdrOf(pageBuf)
	offset := slot.offset

	info := (*TableInfo)(unsafe.Pointer(&pageBuf[offset]))
	*info = TableInfo{} // 复用空洞时必须整体清零, 否则会残留被删表的字段
	copy(info.Name[:], name)
	info.Flag = 0
	info.FieldCount = fieldCount
	info.ActiveFieldCount = fieldCount // 初始时实际字段数等于总字段数
	info.RowSize = types.CalcRowSize(fields)
	info.FieldAllocateSize = uint16(allocateSize)
	info.TotalRows = 0
	info.ExtPage = 0
	info.ExtOff = 0

	// 创建数据页
	dataPID := t.handle.NewPage()
	info.FirstData = dataPID
	info.LastData = dataPID
	dataBuf := make([]byte, PageSize)
	dHdr := page.HdrOf(dataBuf)
	dHdr.Type = page.DataIndex
	dHdr.PageID = dataPID
	dHdr.UseSize = t.hdrSize
	t.handle.SavePage(dataPID, dataBuf)

	// 创建堆页
	heapPID := t.handle.NewPage()
	info.FirstHeap = heapPID
	info.LastHeap = heapPID
	info.HeapPageCount = 1 // 建表即开一张堆页, 堆链长度从 1 起算
	heapBuf := make([]byte, PageSize)
	hHdr := page.HdrOf(heapBuf)
	hHdr.Type = page.HeapIndex
	hHdr.PageID = heapPID
	hHdr.UseSize = t.hdrSize
	t.handle.SavePage(heapPID, heapBuf)

	indexMate := &IndexMetaPage{
		handle: t.handle,
	}
	// 创建索引元数据。
	// 注意这里是 InitEmpty 而不是 Init: indexPID 刚从空闲链摘出来, 盘上还留着它上一世
	// 的内容 (常见是 WAL 页残留)。Init 会从盘上回读本页内容来重建 metas, 于是把整页旧
	// 内容当成 IndexMeta 数组解析出几十上百条垃圾元数据 (实测 91 条, EntryPage=16777216);
	// 之后即便再用 InitHeader 把页写干净, 也清不掉内存里那批垃圾 meta —— 最终由
	// DropTable 拿着垃圾 EntryPage 去枚举页, FreePage 落到从未分配过的页号上,
	// 写到几十 GB 偏移 (DROP 假死 52s、稀疏文件暴涨到 40GB)。
	indexPID := t.handle.NewPage()
	indexMate.InitEmpty(indexPID)
	info.FirstIndexID = indexPID
	info.LastIndexID = indexPID

	// 槽位大于需求(复用空洞): 余量按 TableInfo 拆成墓碑记录留给下次建表;
	// 连一个 TableInfo 都放不下时并入本表内联块 —— 由 FieldAllocateSize 描述的容量
	// 把遍历步进撑到槽位末尾 (writeSchema 会清零整块, 不会留下脏字节)。
	if rem := slot.capacity - required; rem >= t.tableInfoSize {
		free := (*TableInfo)(unsafe.Pointer(&pageBuf[offset+required]))
		*free = TableInfo{}
		free.Flag = 0x80
		free.FieldAllocateSize = uint16(rem - t.tableInfoSize)
	} else if rem > 0 {
		info.FieldAllocateSize = uint16(slot.capacity - t.tableInfoSize)
	}
	if slot.bump {
		pageHdr.UseSize = offset + slot.capacity
	}
	if err := t.handle.SavePage(slot.pageID, pageBuf); err != nil {
		return nil, err
	}

	// schema 定义统一由 writeSchema 落盘 (格式: [activeVer][masterCount][master fields][verCount][ver defs...])
	// 初始 schema 一般能完整放进内联块; 超过 MaxInlineSchemaSize 时余下部分自动进共享扩展页
	newTable := Table{
		info:             info,
		Field:            fields,
		db:               t.db, // 注入, BlobRead/BlobFree 需要
		handle:           t.handle,
		indexMeta:        indexMate,
		tablePageID:      slot.pageID,
		tableStartOffset: offset,
		schemaExt:        t.ensureSchemaExt(),
		activeVersion:    1,
		versions:         initialVersions,
	}
	if err := newTable.writeSchema(); err != nil {
		return nil, err
	}
	t.tables = append(t.tables, &newTable)
	return &newTable, nil
}

// ===========================================================================
// ✅ Compact / Stats —— 表页链压缩与布局统计
// ===========================================================================

// Compact 压缩整条表页链: 丢弃 Drop 留下的空洞, 把活表记录按原顺序紧排到链首的
// 若干页, 多余页回收。
//
// 安全性: 记录以"自描述长度"顺序排列, 页内偏移不对外暴露 —— 外部一律通过
// Table{tablePageID, tableStartOffset} 定位, 所以搬移记录后只要同步更新内存里的
// Table 引用即可, 不会破坏任何索引或 schema。
func (t *TablePage) Compact() error {
	if t.hdr.PageID == 0 {
		return nil
	}

	// 1. 顺序收集链上所有页 ID 与所有活记录的原始字节
	var pageIDs []uint32
	var live [][]byte
	for pid := t.hdr.PageID; pid != 0; {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pid, buf); err != nil {
			return err
		}
		hdr := page.HdrOf(buf)
		pageIDs = append(pageIDs, pid)
		for off := t.hdrSize + schemaExtHeadSize; off+t.tableInfoSize <= hdr.UseSize; {
			info := (*TableInfo)(unsafe.Pointer(&buf[off]))
			size := t.tableInfoSize + uint32(info.FieldAllocateSize)
			if off+size > hdr.UseSize {
				break
			}
			if !util.IsHighBitSet(info.Flag) {
				data := make([]byte, size)
				copy(data, buf[off:off+size])
				live = append(live, data)
			}
			off += size
		}
		pid = hdr.NextPage
	}
	if len(pageIDs) == 0 {
		return nil
	}

	// 2. 紧排。活记录总字节 <= 原占用, 所以一定放得进现有的页数
	first := t.hdrSize + schemaExtHeadSize
	next := 0
	keep := 0
	for pageIdx := 0; pageIdx < len(pageIDs); pageIdx++ {
		if pageIdx > 0 && next >= len(live) {
			break // 后面的页已经没有活记录, 直接回收
		}
		pid := pageIDs[pageIdx]
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pid, buf); err != nil {
			return err
		}
		hdr := page.HdrOf(buf)

		// 只清记录区: [hdrSize, hdrSize+schemaExtHeadSize) 保留共享扩展页链头
		for i := first; i < PageSize; i++ {
			buf[i] = 0
		}
		off := first
		for next < len(live) && off+uint32(len(live[next])) <= PageSize {
			copy(buf[off:], live[next])
			off += uint32(len(live[next]))
			next++
		}
		hdr.UseSize = off

		// 3. 同步内存 Table 的落点 (活表名唯一)
		for o := first; o+t.tableInfoSize <= hdr.UseSize; {
			info := (*TableInfo)(unsafe.Pointer(&buf[o]))
			if tb := t.findTableByName(util.BytesToString(info.Name[:])); tb != nil {
				tb.info = info
				tb.tablePageID = pid
				tb.tableStartOffset = o
			}
			o += t.tableInfoSize + uint32(info.FieldAllocateSize)
		}
		if err := t.handle.SavePage(pid, buf); err != nil {
			return err
		}
		keep++
	}

	// 4. 重连链尾, 回收多余页
	for i := 0; i < keep; i++ {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pageIDs[i], buf); err != nil {
			return err
		}
		nx := uint32(0)
		if i+1 < keep {
			nx = pageIDs[i+1]
		}
		page.HdrOf(buf).NextPage = nx
		if err := t.handle.SavePage(pageIDs[i], buf); err != nil {
			return err
		}
	}
	t.lastPageID = pageIDs[keep-1]
	if t.handle.FreePage != nil {
		for _, pid := range pageIDs[keep:] {
			t.handle.FreePage(pid)
		}
	}
	return nil
}

// Stats 表页链布局统计: 页数 / 活表数 / 空洞数 / 可回收字节数 (诊断与测试用)
func (t *TablePage) Stats() (pages, live, dead int, reclaimable uint32) {
	for pid := t.hdr.PageID; pid != 0; {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pid, buf); err != nil {
			break
		}
		hdr := page.HdrOf(buf)
		pages++
		for off := t.hdrSize + schemaExtHeadSize; off+t.tableInfoSize <= hdr.UseSize; {
			info := (*TableInfo)(unsafe.Pointer(&buf[off]))
			size := t.tableInfoSize + uint32(info.FieldAllocateSize)
			if util.IsHighBitSet(info.Flag) {
				dead++
				reclaimable += size
			} else {
				live++
			}
			off += size
		}
		pid = hdr.NextPage
	}
	return
}

// findTableByName 按表名找内存中的 Table (活表名唯一)
func (t *TablePage) findTableByName(name string) *Table {
	for _, tb := range t.tables {
		if tb != nil && util.BytesToString(tb.info.Name[:]) == name {
			return tb
		}
	}
	return nil
}

// ===========================================================================
// ✅ RenameTable 重命名表 (只改元数据字符串, 不涉及数据页/索引页迁移)
// ===========================================================================
func (t *TablePage) RenameTable(oldName, newName string) error {
	if len(newName) > TableNameSize {
		return fmt.Errorf("%w: 新表名 %q 有 %d 字节, 上限 %d",
			ErrTableNameTooLong, newName, len(newName), TableNameSize)
	}
	// 1. 内存里新表名不能已存在
	for _, tb := range t.tables {
		if util.BytesToString(tb.info.Name[:]) == newName && !util.IsHighBitSet(tb.info.Flag) {
			return ErrTableExists
		}
	}
	tb := t.findTableByName(oldName)
	if tb == nil || util.IsHighBitSet(tb.info.Flag) {
		return ErrTableNotFound
	}

	// 2. 改内存中的 TableInfo.Name
	copy(tb.info.Name[:], newName)
	// 清零剩余字节 (旧名可能更长, 尾巴残留会让 BytesToString 读到垃圾)
	for i := len(newName); i < TableNameSize; i++ {
		tb.info.Name[i] = 0
	}

	// 3. 持久化盘上 TableInfo: 遍历表页链找到这条 TableInfo, 原地改 Name 字段
	for pageID := t.hdr.PageID; pageID != 0; {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pageID, buf); err != nil {
			return err
		}
		hdr := page.HdrOf(buf)
		offset := t.hdrSize + schemaExtHeadSize
		for offset+t.tableInfoSize <= hdr.UseSize {
			info := (*TableInfo)(unsafe.Pointer(&buf[offset]))
			if util.BytesToString(info.Name[:]) == oldName && !util.IsHighBitSet(info.Flag) {
				copy(info.Name[:], newName)
				for i := len(newName); i < TableNameSize; i++ {
					info.Name[i] = 0
				}
				if err := t.handle.SavePage(pageID, buf); err != nil {
					return err
				}
				return nil
			}
			size := t.tableInfoSize + uint32(info.FieldAllocateSize)
			offset += size
		}
		pageID = hdr.NextPage
	}
	return fmt.Errorf("rename: 磁盘上找不到表 %q (内存有但盘上无, 可能未 Commit)", oldName)
}

// ===========================================================================
// ✅ Drop 删除表并回收所有页
// ===========================================================================
func (t *TablePage) Drop(name string) error {
	pageID := t.hdr.PageID

	for pageID != 0 {
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(pageID, buf); err != nil {
			return err
		}

		hdr := page.HdrOf(buf)
		offset := t.hdrSize + schemaExtHeadSize
		for offset+t.tableInfoSize <= hdr.UseSize {
			info := (*TableInfo)(unsafe.Pointer(&buf[offset]))
			if util.BytesToString(info.Name[:]) == name && !util.IsHighBitSet(info.Flag) {
				// 找到该表在内存中的 Table 对象，获取其索引元数据
				var tableEntry *Table
				for _, tb := range t.tables {
					if util.BytesToString(tb.info.Name[:]) == name && !util.IsHighBitSet(tb.info.Flag) {
						tableEntry = tb
						break
					}
				}
				// 回收该表所有页 (用该表自己的 indexMeta.metas，不遍历共享的 IndexMeta 页链)
				t.recycleTablePages(info, tableEntry)
				// 软删除标记
				info.Flag = util.SetHighBit(info.Flag)
				t.handle.SavePage(pageID, buf)
				for i := range t.tables {
					if util.BytesToString(t.tables[i].info.Name[:]) == name {
						t.tables = append(t.tables[:i], t.tables[i+1:]...)
						break
					}
				}
				return nil
			}
			offset += t.tableInfoSize + uint32(info.FieldAllocateSize)
		}
		pageID = hdr.NextPage
	}
	return ErrTableNotFound
}

// recycleTablePages 回收表的所有页 (B+树页、DataPage页、Heap页)
// 注意: IndexMeta 页是所有表共享的，不能释放！
func (t *TablePage) recycleTablePages(info *TableInfo, tableEntry *Table) {
	if t.handle.FreePage == nil {
		return
	}

	var pagesToFree []uint32

	// 1. 用该表自己的 indexMeta.metas 收集索引页和 DataPage
	//    不遍历共享的 IndexMeta 页链，避免误释放其他表的索引页
	//
	//    必须按页布局分流, 与 Deleter.releaseOldTree 同一套约定:
	//    B+ 树节点结构统一, 按内部条目下降即可枚举全部节点页; 哈希 / 复合 / 时序
	//    的页根本不是 B+ 节点 (元数据页+目录页+桶页链, 或扁平条目链), 拿 CollectAllPageIDs
	//    去遍历不会报错, 只会「少收」—— 实测哈希索引 20 页里只回收了元数据页 1 页,
	//    目录页与 18 个桶页全部泄漏; 复合索引 3 页只回收 1 页。文件只涨不缩。
	if tableEntry != nil && tableEntry.indexMeta != nil {
		for i := range tableEntry.indexMeta.metas {
			m := &tableEntry.indexMeta.metas[i]
			if m.Flag&0x80 != 0 || m.EntryPage == 0 {
				continue
			}
			if isBTreePageIndex(m) {
				// 收集 B+ 树所有节点页 (内部节点+叶子节点)
				btPages := btree.CollectAllPageIDs(t.handle, m.EntryPage)
				pagesToFree = append(pagesToFree, btPages...)
				// 只有承载行数据的索引 (主键/时序) 叶子节点存 DataPage ID
				// 二级索引叶子存的是 PKID(8B)，不是 DataPage ID！
				if indexStoresRows(m) {
					dpIDs := btree.GetDataPageIDs(t.handle, m.EntryPage)
					pagesToFree = append(pagesToFree, dpIDs...)
				}
				continue
			}
			h, ok := index.Compose.GetHandler(m.IndexType)
			if !ok {
				continue
			}
			// 优先整链回收: 哈希索引的桶页 next 早已统一到 PageHdr.NextPage, 且每条
			// 桶链的 (head, tail, count) 就持久化在目录槽里 —— 每条链一次 FreeChain,
			// O(1) 拼进空闲链, 完全不必读链内任何一页。
			if cc, ok := h.(index.ChainCollector); ok && t.handle.FreeChain != nil {
				for _, ch := range cc.CollectChains(t.handle, m.EntryPage) {
					if ch.Head == 0 || ch.Count == 0 {
						continue
					}
					if ch.Head == ch.Tail {
						t.handle.FreePage(ch.Head) // 单页链 (例如元数据页)
						continue
					}
					_ = t.handle.FreeChain(ch.Head, ch.Tail, ch.Count)
				}
				continue
			}
			if pc, ok := h.(index.PageCollector); ok {
				pagesToFree = append(pagesToFree, pc.CollectPageIDs(t.handle, m.EntryPage)...)
				continue
			}
			// 兜底: 该类型没有页枚举能力时, 至少把数据页收干净 ——
			// 数据页是行真正的落点, 漏掉它等于漏掉整张表的数据。
			// 承载行数据 ⇔ 实现 RowStoreHandler, 直接断言 (等价于 indexStoresRows, 省一次查表)。
			if rs, ok := h.(index.RowStoreHandler); ok {
				pagesToFree = append(pagesToFree, rs.GetDataPageIDs(t.handle, m.EntryPage)...)
			}
		}
	}

	// 1.5) 表自身的数据页链与索引元数据页链。
	//    CreateTable 每建一张表都会固定分配三类页: Data(FirstData/LastData)、
	//    Heap(FirstHeap/LastHeap)、IndexMeta(FirstIndexID/LastIndexID)。
	//    上面只收了"索引自己的树页/桶页", 这里还必须把 Data 链与 IndexMeta 链整条收掉,
	//    否则 DROP 之后这两条链的页既不被任何表引用、也不在空闲链上 —— 成为永久孤儿页
	//    (analyzer 报的 Data / IdxMeta 孤儿页正是它们)。
	//
	//    IndexMeta 链是「本表独有」的: 每张表的 TableInfo 各自持有 FirstIndexID..LastIndexID,
	//    建表时各自 NewPage, 并不共享。旧注释"IndexMeta 页是全库共享的不能释放"不成立。
	//    链内 next 用 PageHdr.NextPage (与 loadMetas 遍历一致), 逐页收集后统一去重释放。
	for _, head := range []uint32{info.FirstData, info.FirstIndexID} {
		// 成环保护: 链一旦自环 (某页被复用后又把 next 指回自己), 这个循环永远转不完,
		// DROP TABLE 会直接假死。宁可少收几页, 也不能挂在一条坏链上。
		walked := make(map[uint32]bool)
		for pid := head; pid != 0; {
			if walked[pid] {
				break
			}
			walked[pid] = true
			buf := make([]byte, PageSize)
			if err := t.handle.ReadPage(pid, buf); err != nil {
				break
			}
			next := page.HdrOf(buf).NextPage
			pagesToFree = append(pagesToFree, pid)
			pid = next
		}
	}

	// 2. 堆链 (溢出页): 首尾已知 (FirstHeap/LastHeap), 链内 next 与空闲链同用
	//    PageHdr.NextPage —— 整条一次性拼进空闲链, 不遍历、不逐页写, O(1)。
	//
	//    为什么以前必须逐页走: 那时的还链入口只有 FreePage(单页), 所以要先把
	//    每个页号读出来。现在 FreeChain 只要 (链首, 链尾, 链长): 前两个 TableInfo
	//    里就有, 链长由 heapAllocBlock/heapDropPage 一直维护在 HeapPageCount。
	//
	//    退化: 缺 FreeChain 能力 (老 demo 手搓的 HandleDb) 或链长未知时, 退回逐页遍历。
	heapSpliced := false
	if info.FirstHeap != 0 && info.LastHeap != 0 && info.HeapPageCount > 0 && t.handle.FreeChain != nil {
		if err := t.handle.FreeChain(info.FirstHeap, info.LastHeap, info.HeapPageCount); err == nil {
			heapSpliced = true
		}
	}
	if !heapSpliced {
		walked := make(map[uint32]bool)
		heapID := info.FirstHeap
		for heapID != 0 {
			// 同 1.5 的成环保护: 坏链不能把 DROP 拖成死循环。
			if walked[heapID] {
				break
			}
			walked[heapID] = true
			buf := make([]byte, PageSize)
			if err := t.handle.ReadPage(heapID, buf); err != nil {
				break
			}
			hdr := page.HdrOf(buf)
			nextHeap := hdr.NextPage
			pagesToFree = append(pagesToFree, heapID)
			heapID = nextHeap
		}
	}

	// 3. schema 续块不再单独回收: 它位于全库共享的扩展页里, 表被删除后
	//    那一段自然成为垃圾, 由 SchemaExtAllocator 在满页压缩重排时统一回收。

	// 2.5) v10: 释放表中所有行持有的 blob 链. 链页不在 pagesToFree 集合 (它们不是
	//    "表自身的"页, 是行字节携带的外部页). 路径: 遍历 data 页每行, 抽出非空 BlobRef
	//    调 tableEntry.BlobFree 归还空闲链.
	//
	//    为什么放在堆链回收之后: 链页与堆页分属两类存储空间, 顺序本身不重要; 放后面
	//    是为了和"先把本页相关的全部页回收, 再扫内容"的原则一致 (堆链回收完后, 本表
	//    相关的其它页也就剩 data 页与 blob 链, 而 data 页马上就要被 free, 行内容仍可读).
	//
	//    错误处理: 单行扫描失败 / 单链回收失败继续处理后续, 不让一条坏链把整批回收中断
	//    (与 releaseRowBlobRefs 一致的"放任不中断"语义).
	if tableEntry != nil && tableEntry.db != nil && tableEntry.Field != nil {
		hasBlob := false
		for _, f := range tableEntry.Field {
			if types.IsBlob(f.FieldType) {
				hasBlob = true
				break
			}
		}
		if hasBlob {
			dataWalked := make(map[uint32]bool)
			pid := info.FirstData
			for pid != 0 {
				if dataWalked[pid] {
					break
				}
				dataWalked[pid] = true
				buf := make([]byte, PageSize)
				if err := t.handle.ReadPage(pid, buf); err != nil {
					break
				}
				releaseBlobRefsInDataPage(tableEntry, buf)
				next := page.HdrOf(buf).NextPage
				pid = next
			}
		}
	}

	// 4. 统一释放所有页。去重并跳过 0: 多个枚举器拼在一起后, 同一个页 ID 出现两次
	//    会让空闲链形成自环 (NewPage 之后再也走不出去), 而 0 是保留页号, 直接归还
	//    会把空闲链头指向非法页。
	seen := make(map[uint32]bool, len(pagesToFree))
	for _, pid := range pagesToFree {
		if pid == 0 || seen[pid] {
			continue
		}
		// 合法性兜底 (最后一道闸): 页枚举 (CollectAllPageIDs / GetDataPageIDs / 链遍历)
		// 只按偏移解析页头, 不看页类型。meta.EntryPage 一旦指向被复用过的页, 行数据会被
		// 当成 B+ 条目解析, 截出「从未分配过」的垃圾页号 (实测 2463488)。
		// 这种页号绝不能还链: 空闲链头被指向文件里不存在的位置, 下一次 NewPage 拿到它,
		// 所有写入都跑到几十 GB 偏移处 (DROP TABLE 因此假死 52s、稀疏文件涨到 40GB)。
		// 上游已在 btLoadNode / GetDataPageIDs 里按页类型校验, 这里只处理漏网之鱼。
		if t.handle.ValidPage != nil && !t.handle.ValidPage(pid) {
			continue
		}
		seen[pid] = true
		t.handle.FreePage(pid)
	}

	if heapSpliced {
		// 堆链已整条交还空闲链, 表记录里的堆指针一并清空, 别留下悬垂引用。
		// info 指向 Drop 的 buf 内部, 随后的 SavePage 会把这三项一起落盘。
		info.FirstHeap = 0
		info.LastHeap = 0
		info.HeapPageCount = 0
	}

	// Data 链 / IndexMeta 链已回收, 同样清空表记录里的指针, 避免留下指向空闲页的悬垂引用。
	// (表随后被标记删除, 这四项只影响这条墓碑记录的持久化内容。)
	info.FirstData = 0
	info.LastData = 0
	info.FirstIndexID = 0
	info.LastIndexID = 0
}

func (t *TablePage) Find(name string) (*Table, bool) {
	for i := range t.tables {
		tb := t.tables[i]
		if util.BytesToString(tb.info.Name[:]) == name && !util.IsHighBitSet(tb.info.Flag) {
			return tb, true
		}
	}
	return nil, false
}

// ReloadInfos 从磁盘重新读回各表的 TableInfo, 原地覆盖 table.info 指向的内存结构。
//
// 供事务 Rollback 使用。事务回滚只是把被改的页从 BufferPool 里丢掉, 而 Table.info 是
// 独立于页缓存的内存副本(不随缓存回收而消失), 不重载的话 TotalRows / FirstData 等字段
// 仍停留在事务中的值 —— 例如回滚后 TotalRows 偏大, 下一次 InsertRow 会按错误的行号继续,
// 并把错误值随 saveTableInfo 写回页里, 造成真正的数据不一致。
//
// only 为 nil 表示全部重载; 否则只重载 tablePageID 落在其中的表。
// 原地赋值(而非替换指针), 保证外部持有的 *Table / *TableInfo 引用继续有效。
func (t *TablePage) ReloadInfos(only map[uint32]struct{}) {
	for _, tb := range t.tables {
		if tb == nil || tb.info == nil {
			continue
		}
		if only != nil {
			if _, ok := only[tb.tablePageID]; !ok {
				continue
			}
		}
		buf := make([]byte, PageSize)
		if err := t.handle.ReadPage(tb.tablePageID, buf); err != nil {
			continue
		}
		end := tb.tableStartOffset + t.tableInfoSize
		if end > uint32(len(buf)) {
			continue
		}
		disk := (*TableInfo)(unsafe.Pointer(&buf[tb.tableStartOffset]))
		*tb.info = *disk

		// 索引元数据 (IndexMetaPage.metas) 同样是独立于页缓存的内存副本, 必须一起重载:
		// 否则 EntryPage 仍指向事务中新建的 B+ 树根 (或被删空时写成的 0),
		// 而该根页已随缓存丢弃 → 回滚后 SearchByIndex / 范围扫描全部走空。
		if tb.indexMeta != nil {
			tb.indexMeta.Reload(tb.info.FirstIndexID)
		}
	}
}

// InitHeader 初始化头部信息
// 注意: UseSize 额外让出 schemaExtHeadSize 字节 —— 所有表页都在 PageHdr 之后统一预留
// 4 字节(仅第 1 页用来存共享扩展页链头), 这样表偏移遍历逻辑无需区分第几页。
func (t *TablePage) InitHeader(pageID uint32) ([]byte, *page.PageHdr) {
	newBuf := make([]byte, PageSize)
	newHdr := page.HdrOf(newBuf)
	newHdr.Type = page.TableIndex
	newHdr.PageID = pageID
	newHdr.UseSize = uint32(unsafe.Sizeof(*newHdr)) + schemaExtHeadSize
	t.handle.SavePage(pageID, newBuf)
	return newBuf, newHdr
}

// releaseBlobRefsInDataPage 扫描一张数据页, 把每行的 blob 链归还空闲链表.
//
// DropTable 调用: 数据页马上要被 FreePage 收回, 行内字节虽然还在缓冲里, 但任何后续
// 操作都不该再读它. 现在释放行持有的 blob 链是"唯一安全的时机"—— 错过这一秒, 链页
// 就永远泄露.
//
// 行级格式 (DataPage 内):
//
//	[lenRow 4B BE][pkID 8B BE][schemaVer 1B][row bytes lenRow-8B]
//
// 我们只关心 row bytes 段; 跳过前 12B 头部, 把后 lenRow-8 字节喂给 tableEntry 解析.
//
// 实现细节:
//   - DataPageHdr 含 deadBytes / liveCount 但拿不到逐 slot 信息, 这里走"扫描每一行"
//     而不是按 slot 目录 — slot 目录在页尾但 len 字段并不直接列到行起, 走扫描更稳;
//   - 表 schemaVer 必须已知, 否则 deserializeRow 返 nil — 行损坏或 schema 已变更时
//     静默跳过即可 (blob 释放失败比行解析失败更糟).
func releaseBlobRefsInDataPage(tableEntry *Table, buf []byte) {
	if tableEntry == nil {
		return
	}
	// DataPage 头部: PageHdr (固定 20B) + DataPageHdr (固定 16B) — 见 page 包.
	const dataHdrSize = 20 + 16
	useSize := uint32(buf[8])<<24 | uint32(buf[9])<<16 | uint32(buf[10])<<8 | uint32(buf[11])
	if useSize < uint32(dataHdrSize) || int(useSize) > len(buf) {
		return
	}
	off := uint32(dataHdrSize)
	for off+12 <= useSize {
		rowLen := uint32(binary.BigEndian.Uint32(buf[off : off+4]))
		if rowLen < 9 || off+4+rowLen > useSize {
			break // 剩余字节不足或行长非法, 终止扫描 (防御性)
		}
		// 行格式: [pkID 8B][schemaVer 1B][行字节 rowLen-8B]
		rowStart := off + 4 + 8
		rowBytes := buf[rowStart : off+4+rowLen]
		_ = tableEntry.releaseRowBlobRefs(rowBytes)
		off += 4 + rowLen
	}
}