TsDB2/DB/core/table_heap.go

package core

import (
	"encoding/binary"

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

// MaxPhysicalRowData 单行 rowData 的物理上限 (字节) —— 一行必须能装进一个 DataPage。
//
//	16384 (页) - 40 (DataPage 头) - 4 (行长度前缀) - 8 (storeRow 里的 pkID) = 16332
//
// 它与建列闸门 types.MaxRowDataSize (8171, 一页两行) 是两件事:
//   - MaxRowDataSize 是设计政策, 在 CreateTable / AddField 处执行, 保证一页至少装两行;
//     它卡的是【整行占用】= 4 + 8 + 这里的 rowData, 所以比这道物理墙紧得多 (8159 / 16332);
//   - MaxPhysicalRowData 是物理墙, 在写入时兜底 —— 超过它的行字节流会写坏 DataPage,
//     必须有第二道网拦住"建列闸门之前的旧表"与"schema 被外部改坏"这两种情况。
const MaxPhysicalRowData = int(PageSize) - int(page.DataPageHdrSize) - 4 - 8

// ======================== 动态字段堆页存储 ========================
//
// 动态字段 (FieldType_Dynamicbyte: 声明宽度超过 MaxInlineColumnWidth(2048) 字节的
// 字符串 / []byte) 的行内只保存 6 字节引用: [PageID uint32 BE(4B)][Offset uint16 BE(2B)]
//   - PageID == 0 → 空值 (NULL / 空串 / 空二进制)
//   - 否则指向堆页 PageID 的物理偏移 Offset 处的数据块
// 堆页数据块格式: [Len uint32 BE][Payload...]
// 单块上限 = PageSize - PageHdrSize - dynamicBlockHdr (约 16KB)
// 堆页链: TableInfo.FirstHeap → ... → TableInfo.LastHeap
// 注意: 该链与 schema 续块所在的「共享扩展页」相互独立。
//
// 由于行内引用固定 6 字节, 动态字段变长/变短都不会改变行长,
// 因此更新走 DataPage 原地覆盖即可, 不会触发 errRowTooLarge。

const dynamicBlockHdr = 4 // 堆数据块的长度前缀大小

// MaxBlobSize 单个【动态列】(FieldType_Dynamicbyte) 值的大小上限 (字节)。
//
// 它不是拍出来的数, 而是从存储约束直接推出: 一个堆块必须整体落进一个堆页, 可用
// 载荷 = 页大小 - 页头 - 块长度前缀 (与 heapAllocBlock 的 maxPayload 同源)。
//
// 为什么单独列出来: 动态列是"变长文本/二进制"的落点 (源库里超宽的 varchar/text
// 都落这里), 使用者必须能在写入前就知道这条线, 而不是写进去才撞上 heapAllocBlock
// 里那句与"值太大"看不出因果的 ErrHeapPageTooLarge。写入侧的检查见 writeFieldValueAt。
//
// 【v10 起不适用于 FieldType_Blob】blob 列的行内是 8B BlobRef, 内容落在可跨页的链上,
// 单值上限是 BlobChainMax (~4GB), 不受这里约束 —— 名字里的 Blob 是历史遗留
// (v10 之前 Blob 列与动态列同布局, 所以共用这一条线)。
//
// 它同时是【声明侧】的容量上限 (声明超限的字符串列也落动态列, 容量同样是这个数):
// types.MaxStringSize 就是它, 由下面那行编译期断言钉住。
const MaxBlobSize = PageSize - page.PageHdrSize - dynamicBlockHdr

// 编译期断言: types 里那组【页面布局镜像】推出的两个上限, 必须与这里的物理布局一致
// (行宽那条因为要整除, 只钉到 ±1 字节 —— "它恰好是一页能放下的最大值"这个精确性质由
// core 的 TestMaxRowDataSizeFitsTwoRows 用真实的 PageSize / DataPageHdrSize 复核)。
// 两处一旦漂移, 后果不是报错而是"承诺了做不到的事":
//   - MaxStringSize 对不上 MaxBlobSize → 允许声明一个写不进堆页的容量, 建表时通过、
//     某天写入大值时才撞 ErrHeapPageTooLarge;
//   - MaxRowDataSize 对不上"一页两行" → 放进来一张两行装不进一页的表, 页内顺序扫描、
//     按 pkID 的有序插入、更新时的页内重建全部退化, 正是那条政策要避免的情形。
//
// 用"下标必须为 0"表达相等: 常量下标越界直接编译失败, 且表达式不占任何空间。改页大小 /
// 页头 / 块长度前缀时这里会先报错, 提醒同步 types 里那组镜像。
var (
	_ = [1]struct{}{}[int(MaxBlobSize)-types.MaxStringSize]
	_ = [1]struct{}{}[int(types.MaxRowDataSize)-int((PageSize-page.DataPageHdrSize-1)/2)]
)

// maxHeapRefCount 单个堆页的引用数防御上限 (RefCount 为 uint16, 容器上限 65535)。
//
// 注意: 这是纯防御性常量, 实际永远不会触发 —— 堆页块数的物理上限只有 3272
// (数据区 16364 字节 ÷ 最小块 5 字节), 远在 65535 之前空间就先耗尽, 所以
// heapAllocBlock 里的 `RefCount < maxHeapRefCount` 恒为真, 从不决定任何事。
// 真正保证计数不溢出的是同一行的空间检查。
//
// 它唯一的价值: 若将来放开"1 引用 = 1 块, 且块不共享"的前提 (块级去重、多行共用
// 一个堆块), 上限会变成"全库行数", 这道闸才真正开始工作。
const maxHeapRefCount = 1<<16 - 1

// heapAllocBlock 在堆页链中分配 need 字节数据空间, 返回承载页与页内物理偏移
func (table *Table) heapAllocBlock(need uint32) (*HeapPage, uint16, error) {
	maxPayload := uint32(MaxBlobSize)
	if need > maxPayload {
		return nil, 0, ErrHeapPageTooLarge
	}
	total := need + dynamicBlockHdr

	// 1. 尾页是当前唯一可写页: 堆链只往尾部追加, 新块一律落在链尾。
	//
	// 为什么不再从 FirstHeap 逐个扫"第一个有空间的页": 均匀定长插入下, 前面每一
	// 页都会因剩余空间不足而被扫到且都放不下, 单次插入退化成 O(堆页数), 整体
	// O(n²) —— 实测插 10 万行要十几分钟且内存顶不住。尾插把单次插入降到 O(1)。
	//
	// 代价: 旧尾页里不足一个块的零头不再被后续小块回填, 每页最多浪费「放不下的
	// 那个块大小」以内。若尾页恰好是 heapUnref 清空复用的空页, 则它必然放得下
	// 任何块(空页可用空间 >= 单块上限), 不会因此多开一张页。
	var tail *HeapPage
	appended := false // 本次是否新开了一张堆页 —— 决定 HeapPageCount 是否 +1
	if table.info.LastHeap != 0 {
		hp, err := LoadHeapPage(table.handle, table.info.LastHeap)
		if err != nil {
			return nil, 0, err
		}
		// 左半边恒为真 (见 maxHeapRefCount 的注释): 空间总是先于计数耗尽。
		if hp.Info.Hdr.RefCount < maxHeapRefCount && PageSize-hp.Info.Hdr.UseSize >= total {
			tail = hp
		} else {
			// 尾页放不下 → 在它后面接一张新页作为新尾
			nhp := NewHeapPage(table.handle, page.HeapIndex)
			hp.Info.Hdr.NextPage = nhp.Info.Hdr.PageID
			nhp.Info.Hdr.PrePage = hp.Info.Hdr.PageID
			if err := hp.Save(); err != nil {
				return nil, 0, err
			}
			tail = nhp
			appended = true
		}
	} else {
		// 链为空 (旧库或已被清空): 新建首页
		tail = NewHeapPage(table.handle, page.HeapIndex)
		table.info.FirstHeap = tail.Info.Hdr.PageID
		appended = true
	}
	table.info.LastHeap = tail.Info.Hdr.PageID
	if appended {
		// 堆链长度自增 —— DropTable 就是靠它不遍历也拿到回收页数
		table.info.HeapPageCount++
		// 只有新开堆页时 FirstHeap/LastHeap/HeapPageCount 才会变 —— 尾页直接分配
		// 的常见路径什么都不用落盘, 省一次"读 16KB 页 + 写 16KB 页"的往返。
		// (旧实现无条件 saveTableInfo, 每行动态列都白付这两趟拷贝。)
		table.saveTableInfo()
	}

	_, off, err := tail.Alloc(total)
	if err != nil {
		return nil, 0, err
	}
	if off > 0xFFFF {
		return nil, 0, ErrHeapPageTooLarge
	}
	return tail, uint16(off), nil
}

// heapWrite 分配堆空间并写入数据块, 返回行内引用 (pageID, offset)
func (table *Table) heapWrite(data []byte) (uint32, uint16, error) {
	if len(data) == 0 {
		return 0, 0, nil // 空值: PageID = 0
	}
	hp, off, err := table.heapAllocBlock(uint32(len(data)))
	if err != nil {
		return 0, 0, err
	}
	di := int(off) - int(page.PageHdrSize)
	if di < 0 || di+dynamicBlockHdr+len(data) > len(hp.Info.Data) {
		return 0, 0, ErrPageRead
	}
	binary.BigEndian.PutUint32(hp.Info.Data[di:], uint32(len(data)))
	copy(hp.Info.Data[di+dynamicBlockHdr:], data)
	if err := hp.Save(); err != nil {
		return 0, 0, err
	}
	return hp.Info.Hdr.PageID, off, nil
}

// heapRead 读取行内引用指向的堆数据; PageID == 0 表示空值
func (table *Table) heapRead(pageID uint32, off uint16) ([]byte, error) {
	if pageID == 0 {
		return nil, nil
	}
	hp, err := LoadHeapPage(table.handle, pageID)
	if err != nil {
		return nil, err
	}
	di := int(off) - int(page.PageHdrSize)
	if di < 0 || di+dynamicBlockHdr > len(hp.Info.Data) {
		return nil, ErrPageRead
	}
	n := int(binary.BigEndian.Uint32(hp.Info.Data[di:]))
	if n < 0 || di+dynamicBlockHdr+n > len(hp.Info.Data) {
		return nil, ErrPageRead
	}
	out := make([]byte, n)
	copy(out, hp.Info.Data[di+dynamicBlockHdr:di+dynamicBlockHdr+n])
	return out, nil
}

// heapRewrite 尝试原地覆写已有堆块 (新数据不超过旧块容量时), 成功返回 true
// 用于更新场景: 复用旧块可避免每次变长更新都产生堆垃圾
func (table *Table) heapRewrite(pageID uint32, off uint16, data []byte) bool {
	if pageID == 0 {
		return false
	}
	hp, err := LoadHeapPage(table.handle, pageID)
	if err != nil {
		return false
	}
	di := int(off) - int(page.PageHdrSize)
	if di < 0 || di+dynamicBlockHdr > len(hp.Info.Data) {
		return false
	}
	oldLen := int(binary.BigEndian.Uint32(hp.Info.Data[di:]))
	if oldLen < len(data) || di+dynamicBlockHdr+oldLen > len(hp.Info.Data) {
		return false
	}
	binary.BigEndian.PutUint32(hp.Info.Data[di:], uint32(len(data)))
	copy(hp.Info.Data[di+dynamicBlockHdr:], data)
	return hp.Save() == nil
}

// heapUnref 释放一个堆页引用。计数归零后按"该页还有没有后继"分流:
//
//   - 有后继 → 摘链, 整页交还空闲链表
//   - 是链尾 → 作为堆尾保留, 但把页内空间清零复用
//
// 链尾为什么不能只是"原样保留": DecRef 是引用计数下降的唯一入口, 判定只在这一刻做。
// 若链尾归零后原样保留, 它的 UseSize 仍是满的; 之后新块放不下时新页会接在它后面,
// 它就变成"计数已归零的中间页" —— 而再没有任何人会来 DecRef 它, 于是 16KB
// 永久泄漏且永不复用。所以链尾归零时就把 UseSize 重置回页头之后
// (计数归零 ⟺ 页内没有活块, 这些空间全是垃圾), 让它退化成一张可用的空页。
func (table *Table) heapUnref(pageID uint32) error {
	if pageID == 0 {
		return nil
	}
	hp, err := LoadHeapPage(table.handle, pageID)
	if err != nil {
		return err
	}
	if !hp.Info.Hdr.DecRef() {
		// 仍有活引用: 只把下降后的计数落盘
		return hp.Save()
	}
	if hp.Info.Hdr.NextPage == 0 {
		// 链尾: 保留作为堆尾, 但清空可用空间以便复用
		hp.Info.Hdr.UseSize = page.PageHdrSize
		return hp.Save()
	}
	return table.heapDropPage(hp)
}

// heapDropPage 把已无引用的堆页从双向链上摘除, 并登记回收
// 前置条件: 该页引用计数已归零
func (table *Table) heapDropPage(hp *HeapPage) error {
	next := hp.Info.Hdr.NextPage
	prev := hp.Info.Hdr.PrePage

	// 上级指向下级
	if prev != 0 {
		p, err := LoadHeapPage(table.handle, prev)
		if err != nil {
			return err
		}
		p.Info.Hdr.NextPage = next
		if err := p.Save(); err != nil {
			return err
		}
	} else {
		table.info.FirstHeap = next
	}
	// 下级指向上级
	if next != 0 {
		n, err := LoadHeapPage(table.handle, next)
		if err != nil {
			return err
		}
		n.Info.Hdr.PrePage = prev
		if err := n.Save(); err != nil {
			return err
		}
	} else {
		table.info.LastHeap = prev
	}

	// 交还空闲链表 (页内容会被链表指针覆盖)。
	// 走延迟回收: 事务中途直写盘会让"页已清零"先于 WAL 落盘, 回滚/崩溃后
	// 堆链还指向这张空页 → 数据丢失。
	table.deferFreePage(hp.Info.Hdr.PageID)
	// 堆链长度自减 (与 heapAllocBlock 的 +1 配对, 保证 DropTable 时链长可信)
	if table.info.HeapPageCount > 0 {
		table.info.HeapPageCount--
	}
	table.saveTableInfo()
	return nil
}

// deferFreePage 登记一个可回收页。有延迟回收回调时走"等提交后再还链",
// 否则退回立即回收 (例如 demo 里手搓但未提供该回调的 HandleDb)。
func (table *Table) deferFreePage(pageID uint32) {
	if table.handle.DeferFreePage != nil {
		table.handle.DeferFreePage(pageID)
		return
	}
	table.handle.FreePage(pageID)
}

// oldHeapRefInfo 旧行在某列上持有的堆引用 (动态列或值级溢出列)。
type oldHeapRefInfo struct {
	pid uint32
	off uint16
}

// oldRowHeapRefs 解析旧行, 收集各列持有的堆引用, 返回 masterIdx → 引用;
// 旧行为空 / 版本未知 / 偏移表不可信时返回 nil。
//
// 【只认"动态列或带 RowVarTag 的列"是安全边界】值级溢出让同一字符串列
// "这行内联、那行溢出": 旧行该列若是内联内容, 位置上那 6 字节就是字符串本身 ——
// 当引用解会拿到垃圾页号, 后续 heapRewrite / heapUnref 会读写不相干的页 (静默破坏)。
// 所以绝不能只按偏移取 6 字节就当引用, 必须先看类型或偏移表项的 tag 位。
//
// 解析一次、整行共用: 更新路径每列都要查旧引用 (复用 / 释放), 逐列各解析一遍
// 是 O(n²), n 上限 254 时每行白付几万次偏移表扫描。
func (table *Table) oldRowHeapRefs(oldRowData []byte) map[int]oldHeapRefInfo {
	if len(oldRowData) < 1 {
		return nil
	}
	layout, ok := table.parseRowLayout(oldRowData)
	if !ok {
		return nil
	}
	sv, n, col0Width, ends := layout.sv, len(layout.sv.FieldIndices), layout.col0Width, layout.ends
	if len(sv.FieldIndices) == 0 {
		return nil
	}
	bitmap, ok := types.RowNullBitmap(oldRowData, ends, n)
	if !ok {
		return nil
	}
	var refs map[int]oldHeapRefInfo
	for i, idx := range sv.FieldIndices {
		f := table.Field[idx]
		// NULL 列不持有堆块; 引用槽即便残留非零也不是本行的。
		if f.IsDropped() || types.NullBitSet(bitmap, i) {
			continue
		}
		if !types.IsDynamic(f.FieldType) && !types.RowFieldRefTagged(oldRowData, i, col0Width) {
			continue // 内联内容, 不是引用
		}
		start := int(types.RowFieldStart(ends, i))
		if start+types.DynamicRefSize > len(oldRowData) {
			continue
		}
		pid := binary.BigEndian.Uint32(oldRowData[start : start+4])
		if pid == 0 {
			continue // 空引用 = 空内容
		}
		if refs == nil {
			refs = make(map[int]oldHeapRefInfo)
		}
		refs[int(idx)] = oldHeapRefInfo{
			pid: pid,
			off: binary.BigEndian.Uint16(oldRowData[start+4 : start+6]),
		}
	}
	return refs
}

// heapReleaseRow 释放一行占用的全部堆页引用 (删除行时调用)。
// 按行自身的 schema 版本 + 行内偏移表定位各动态字段 / 值级溢出字段的 6 字节引用。
//
// v10 起: 同时回收 blob 列的内联 8B BlobRef 引用的链页. 路径与堆块不同 ——
// heap 页走 heapUnref (RefCount), blob 链页走 BlobFree (整链挂回空闲链表).
// 两者必须分别处理, 不能合并.
func (table *Table) heapReleaseRow(rowData []byte) {
	// 1. 堆块 (动态列 ∪ 值级溢出列): 删除只释放, 不复用 — 直接用 oldRowHeapRefs 判定.
	if refs := table.oldRowHeapRefs(rowData); refs != nil {
		for _, r := range refs {
			_ = table.heapUnref(r.pid)
		}
	}
	// 2. blob 链页: 按列扫描非空 BlobRef, 一一归还空闲链表. releaseRowBlobRefs 内部
	// 错误"放任不中断"语义, 让一条 blob 释放失败不会把整批后续 blob 漏掉.
	_ = table.releaseRowBlobRefs(rowData)
}

// writeHeapRefField 把 data 落堆, 并在行内 offset 处写入 6 字节引用。
//
// 动态列与【值级溢出的字符串列】共用这条路径 (盘上布局相同)。old 为旧行同列的堆引用
// (来自 oldRowHeapRefs): 能原地复用就复用, 换块后释放旧块; nil 表示旧行该列没有引用。
// checkOnly 为真时只占位 (行内引用保持全零), 不碰堆 —— 试算行可能被整丢弃,
// 绝不能有分配 / 覆写 / 释放堆块的副作用 (见 serializeRowForCheck)。
func (table *Table) writeHeapRefField(rowData []byte, offset uint16, data []byte, old *oldHeapRefInfo, checkOnly bool) error {
	if checkOnly {
		return nil
	}
	if len(data) == 0 {
		// 新值为空: 行内引用保持全 0, 旧块引用需要释放
		if old != nil {
			_ = table.heapUnref(old.pid)
		}
		return nil
	}

	// 1. 尽量复用旧堆块 (原地覆写), 此时引用不变
	if old != nil && table.heapRewrite(old.pid, old.off, data) {
		binary.BigEndian.PutUint32(rowData[offset:], old.pid)
		binary.BigEndian.PutUint16(rowData[offset+4:], old.off)
		return nil
	}

	// 2. 分配新块 (新引用 +1)
	pageID, off, err := table.heapWrite(data)
	if err != nil {
		return err
	}
	binary.BigEndian.PutUint32(rowData[offset:], pageID)
	binary.BigEndian.PutUint16(rowData[offset+4:], off)

	// 3. 旧块不再被本行引用 → 释放
	// 放在分配之后, 避免刚摘链回收的页立刻又被分配造成抖动
	if old != nil && !(old.pid == pageID && old.off == off) {
		_ = table.heapUnref(old.pid)
	}
	return nil
}