TsDB2/DB/core/schema_ext.go

package core

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

// =============================================================================
// SchemaExtAllocator —— 全库共享的 schema 扩展页分配器
// =============================================================================
//
// 为什么共享: schema 一般只有几十~几千字节。若每张表独占一张 16KB 页, 表一多
// 就极其浪费; 全库共用一条链, 大家按实际长度挤在同一张页里。
//
// 页链与定位:
//
//	表页第 1 页 : [PageHdr][SchemaExtPageID u32][TableInfo][内联块]...
//	                        └─ 链头 (0 = 还没有扩展页)
//	扩展页      : [PageHdr][frag][frag]...[空闲] ─NextPage─> ...
//	frag        : [SchemaFragHdr 8B][payload]
//
// 每张表自己的首个续块位置记在它的 TableInfo.ExtPage/ExtOff 里 (不在这里)。
// 页头的 UseSize 复用为「尾页 bump 偏移」, 重启后据此恢复 lastOff。
//
// 回收策略: 只往尾页追加(bump), 续块不单独释放 —— 每次 writeSchema 都是整条重写,
// 旧续块立刻成为垃圾。当尾页放不下时:
//   - 若压缩重排后能减少页数 → Compact(): 把所有活续块紧排进新页, 原子换头, 再回收旧链
//   - 否则直接追加新页
//
// 这样做无需页内 free list 与碎片管理, 代价是压缩时要把所有表重写一遍 (只在 DDL 触发)。
type SchemaExtAllocator struct {
	handle   page.HandleDb
	tables   func() []*Table           // 取全量表, 供压缩重排使用
	saveHead func(pageID uint32) error // 把链头写回表页第 1 页

	firstPID uint32 // 链首页 (0 = 空)
	lastPID  uint32 // 链尾页, 追加位置
	lastOff  uint32 // 尾页已 bump 到的物理偏移
	pages    int    // 链长度

	inited     bool
	compacting bool // 正在压缩重排 → 期间禁止再压缩
}

// schemaExtHeadSize 表页第 1 页在 PageHdr 之后为 SchemaExtPageID 预留的字节数
// (所有表页统一预留, 使表偏移遍历逻辑无需区分第几页)
const schemaExtHeadSize = 4

// schemaFragMaxPayload 单个续块能承载的最大 payload
func schemaFragMaxPayload() int {
	return int(PageSize) - int(page.PageHdrSize) - schemaFragHdrSize
}

// schemaFrag 一次分配得到的续块位置与容量
type schemaFrag struct {
	PageID uint32
	Off    uint32
	Len    uint16 // payload 字节数
}

// newSchemaExtAllocator 创建分配器 (firstPID 为已存在的链头, 0 表示空)
func newSchemaExtAllocator(handle page.HandleDb, firstPID uint32,
	tables func() []*Table, saveHead func(uint32) error) *SchemaExtAllocator {
	return &SchemaExtAllocator{
		handle:   handle,
		tables:   tables,
		saveHead: saveHead,
		firstPID: firstPID,
	}
}

// init 定位链尾并恢复 bump 偏移 (懒加载, 首次分配时才读盘)
func (a *SchemaExtAllocator) init() error {
	if a.inited {
		return nil
	}
	a.inited = true
	a.pages = 0
	a.lastPID = 0
	a.lastOff = 0
	for pid := a.firstPID; pid != 0; {
		buf := make([]byte, PageSize)
		if err := a.handle.ReadPage(pid, buf); err != nil {
			return err
		}
		hdr := page.HdrOf(buf)
		a.pages++
		if hdr.NextPage == 0 {
			a.lastPID = pid
			if hdr.UseSize >= page.PageHdrSize && hdr.UseSize <= PageSize {
				a.lastOff = hdr.UseSize
			} else {
				a.lastOff = page.PageHdrSize
			}
			return nil
		}
		pid = hdr.NextPage
	}
	return nil
}

// flushTail 把尾页的 bump 偏移写回页头 UseSize
func (a *SchemaExtAllocator) flushTail() error {
	if a.lastPID == 0 {
		return nil
	}
	buf := make([]byte, PageSize)
	if err := a.handle.ReadPage(a.lastPID, buf); err != nil {
		return err
	}
	page.HdrOf(buf).UseSize = a.lastOff
	return a.handle.SavePage(a.lastPID, buf)
}

// appendPage 在链尾追加一张新的扩展页, 并把它挂到前驱的 NextPage 上
func (a *SchemaExtAllocator) appendPage() error {
	prev := a.lastPID
	if prev != 0 {
		// 换页前先把旧尾页的 bump 偏移落盘
		if err := a.flushTail(); err != nil {
			return err
		}
	}

	pid := a.handle.NewPage()
	buf := make([]byte, PageSize)
	hdr := page.HdrOf(buf)
	hdr.Type = page.SchemaExtIndex
	hdr.PageID = pid
	hdr.UseSize = page.PageHdrSize
	hdr.PrePage = prev
	if err := a.handle.SavePage(pid, buf); err != nil {
		return err
	}

	if prev != 0 {
		pb := make([]byte, PageSize)
		if err := a.handle.ReadPage(prev, pb); err != nil {
			return err
		}
		page.HdrOf(pb).NextPage = pid
		if err := a.handle.SavePage(prev, pb); err != nil {
			return err
		}
	} else {
		// 空链 → 新页即链头, 需要回写表页第 1 页
		a.firstPID = pid
		if a.saveHead != nil {
			if err := a.saveHead(pid); err != nil {
				return err
			}
		}
	}

	a.lastPID = pid
	a.lastOff = page.PageHdrSize
	a.pages++
	return nil
}

// AllocChain 为 total 字节的续块数据分配若干片段, 返回每段的位置与容量
// 分配只发生在尾页, 一段放不下就跨页续下一段
//
// 注意: 压缩重排会回收整条旧链, 使**已经记下的片段地址全部失效**。所以压缩只允许
// 发生在开始记录地址之前 (下面这一次判断); 之后的循环一律只追加新页, 绝不压缩。
func (a *SchemaExtAllocator) AllocChain(total int) ([]schemaFrag, error) {
	if total <= 0 {
		return nil, nil
	}
	if err := a.init(); err != nil {
		return nil, err
	}
	// 尾页装不下这次要写的全部数据 → 马上要新增页, 这是值得压缩重排的时机。
	// 反过来, 只要尾页装得下就不压缩 —— 否则每次 DDL 都要重排所有表。
	if a.lastPID != 0 && PageSize-a.lastOff < uint32(total) {
		if a.shouldCompact(total) {
			if err := a.Compact(); err != nil {
				return nil, err
			}
		}
	}

	frags := make([]schemaFrag, 0, 1+total/schemaFragMaxPayload())
	remain := total
	for remain > 0 {
		if a.lastPID == 0 || PageSize-a.lastOff < uint32(schemaFragHdrSize)+1 {
			if err := a.appendPage(); err != nil {
				return nil, err
			}
		}
		n := int(PageSize-a.lastOff) - schemaFragHdrSize
		if n > remain {
			n = remain
		}
		if n > schemaFragMaxPayload() {
			n = schemaFragMaxPayload()
		}
		frags = append(frags, schemaFrag{PageID: a.lastPID, Off: a.lastOff, Len: uint16(n)})
		a.lastOff += uint32(schemaFragHdrSize + n)
		remain -= n
	}
	if err := a.flushTail(); err != nil {
		return nil, err
	}
	return frags, nil
}

// liveStats 统计当前所有表续块占用的净字节数与续块数
func (a *SchemaExtAllocator) liveStats() (int, int) {
	live, n := 0, 0
	if a.tables == nil {
		return 0, 0
	}
	for _, t := range a.tables() {
		if t == nil || t.info.ExtPage == 0 {
			continue
		}
		inline := int(t.info.FieldAllocateSize)
		data := t.serializeSchema()
		if len(data) <= inline {
			continue
		}
		live += len(data) - inline
		n++
	}
	return live, n
}

// shouldCompact 判断压缩重排 (外加即将写入的 extra 字节) 是否能减少页数 (不能减就不折腾)
func (a *SchemaExtAllocator) shouldCompact(extra int) bool {
	if a.pages <= 1 || a.compacting {
		return false
	}
	live, liveTables := a.liveStats()
	usable := uint64(PageSize - page.PageHdrSize)
	need := uint64(live) + uint64(liveTables)*uint64(schemaFragHdrSize) + uint64(extra)
	pagesAfter := int((need + usable - 1) / usable)
	if pagesAfter < 1 {
		pagesAfter = 1
	}
	return pagesAfter < a.pages
}

// Compact 满页压缩重排: 把所有活续块紧排进一条新链, 换头后再回收旧链
//
// 先在新链上把所有表重写完毕, 最后才回写表页里的链头 —— 中途失败只会泄漏几张
// 页, 不会丢失已有 schema。
func (a *SchemaExtAllocator) Compact() error {
	if err := a.init(); err != nil {
		return err
	}
	oldFirst := a.firstPID

	// 新链在独立的临时分配器上构建, saveHead 置空 → 先不暴露给外部
	// compacting=true → 构建过程中禁止再压缩 (否则会递归, 且新链地址会失效)
	tmp := &SchemaExtAllocator{
		handle:     a.handle,
		tables:     a.tables,
		saveHead:   func(uint32) error { return nil },
		inited:     true,
		compacting: true,
	}
	if err := tmp.appendPage(); err != nil {
		return err
	}

	for _, t := range a.tables() {
		if t == nil || t.info.ExtPage == 0 {
			continue
		}
		if err := t.writeSchemaExt(tmp); err != nil {
			return err
		}
	}

	// 原子换头
	a.firstPID, a.lastPID, a.lastOff, a.pages = tmp.firstPID, tmp.lastPID, tmp.lastOff, tmp.pages
	if a.saveHead != nil {
		if err := a.saveHead(a.firstPID); err != nil {
			return err
		}
	}

	a.freeChain(oldFirst)
	return nil
}

// freeChain 回收整条扩展页链
func (a *SchemaExtAllocator) freeChain(first uint32) {
	if a.handle.FreePage == nil {
		return
	}
	for pid := first; pid != 0; {
		buf := make([]byte, PageSize)
		if err := a.handle.ReadPage(pid, buf); err != nil {
			return
		}
		next := page.HdrOf(buf).NextPage
		a.handle.FreePage(pid)
		pid = next
	}
}