TsDB2/DB/core/Table.go

package core

import (
	"encoding/binary"
	"sync"
	"unsafe"

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

// SchemaFragHdr 共享扩展页里每个 schema 续块的头 (定长 8 字节, 紧随 payload)
// 警告: 该结构体仅用于计算大小/表达语义, 写盘一律用 putSchemaFragHdr,
// 不要做 `*(*SchemaFragHdr)(p) = v` 的结构体赋值。
type SchemaFragHdr struct {
	NextPage uint32 // 下一个续块所在页 (0 = 本表续块链结束)
	NextOff  uint16 // 下一个续块在页内的物理偏移
	DataLen  uint16 // 本续块 payload 字节数
}

// schemaFragHdrSize schema 续块头的物理大小 (字节)
const schemaFragHdrSize = 8

// putSchemaFragHdr 写入续块头 (大端: NextPage 4B + NextOff 2B + DataLen 2B)
func putSchemaFragHdr(buf []byte, off int, h SchemaFragHdr) {
	binary.BigEndian.PutUint32(buf[off:off+4], h.NextPage)
	binary.BigEndian.PutUint16(buf[off+4:off+6], h.NextOff)
	binary.BigEndian.PutUint16(buf[off+6:off+8], h.DataLen)
}

// readSchemaFragHdr 读取续块头
func readSchemaFragHdr(buf []byte, off int) SchemaFragHdr {
	return SchemaFragHdr{
		NextPage: binary.BigEndian.Uint32(buf[off : off+4]),
		NextOff:  binary.BigEndian.Uint16(buf[off+4 : off+6]),
		DataLen:  binary.BigEndian.Uint16(buf[off+6 : off+8]),
	}
}

// SchemaVersion 表的一个 schema 版本定义
type SchemaVersion struct {
	Version      byte
	FieldIndices []byte // 每个字段在 masterFields 中的索引
}
type TableInfo struct {
	Name             [TableNameSize]byte
	Flag             byte // 表的状态是否被删除
	FieldCount       byte // 当前字段的总数量（包含已删除，永不减少）byte=255上限
	ActiveFieldCount byte // 实际活跃字段数量（不包含已删除）
	_                byte // 1 显式对齐填充 (TotalRows 需要 8 字节对齐)

	// HeapPageCount 本表堆链 (FirstHeap→LastHeap) 上的页数。
	//
	// 塞进 TotalRows 前的对齐填充里 (那 4 字节本来就是编译器要插入的空洞), 不额外
	// 增加 TableInfo 的体积。
	// 用途: DropTable 把整条堆链 O(1) 拼进空闲链时, 不遍历也有链长可累加空闲页计数。
	HeapPageCount uint32 // 4 堆链页数
	TotalRows     uint64 // 当前表行的总数量

	FirstHeap uint32 // 堆链首页
	LastHeap  uint32 // 堆链尾页

	FirstData uint32 // 数据链首页
	LastData  uint32 // 数据链尾

	RowSize           uint16 // 单条记录大小（字节）
	FieldAllocateSize uint16 // 当前列总分配大小

	IndexCount   byte   // 这张表有几个索引
	FirstIndexID uint32 // 本表第一个索引的元数据位置
	LastIndexID  uint32 // 本表最后一个索引的元数据位置

	// ExtRef schema 续块在「全库共享扩展页」中的位置
	// (复用原 FirstOverflow/LastOverflow 占用的 8 字节: ExtPage 落在 8 字节对齐处,
	//  尾部 2 字节显式填充, 使 TableInfo 整体大小为 8 的倍数)
	ExtPage uint32 // 4  首个续块所在扩展页 ID (0 = schema 全部内联, 无续块)
	ExtOff  uint16 // 2  首个续块在页内的物理偏移
	_       uint16 // 2  显式填充, 使整体大小为 8 的倍数
}

// 编译期断言: 钉死 TableInfo 的盘上大小与字段布局。
//
// TableNameSize 变化会整体平移 Name 之后所有字段的偏移 —— analyzer 那套硬编码偏移
// (analyzer/main.go 的 TI_*) 必须同步修改, 改漏了会静默读错数据。这里把大小钉死,
// 让「改了布局却忘了改断言」直接编译失败。
//
// 布局 (TableNameSize=48) = 104 字节:
//
//	Name[48] + Flag/FieldCount/ActiveFieldCount/pad(4) + HeapPageCount(4) + TotalRows(8)
//	+ FirstHeap/LastHeap/FirstData/LastData(16) + RowSize/FieldAllocateSize(4)
//	+ IndexCount/pad(4) + FirstIndexID/LastIndexID/ExtPage(12) + ExtOff/pad(4)
var (
	_ [unsafe.Sizeof(TableInfo{}) - 104]struct{}
	_ [104 - unsafe.Sizeof(TableInfo{})]struct{}
)

// readSchema 读取一张表完整的 schema 定义
// 数据来源: 表页内的固定内联块 (allocSize 字节) + 共享扩展页里的续块链
// 新格式: [activeVer(1B)] [masterCount(1B)] [type+len+name]×masterCount [verCount(1B)] [verDefs...]
func readSchema(handle page.HandleDb, pageID uint32, startOff uint16, allocSize uint16,
	extPage uint32, extOff uint16) (byte, []types.FieldInfo, []SchemaVersion) {

	// 1. 内联块 (表页内, 固定 allocSize 字节, 无跳转槽)
	var data []byte
	buf := make([]byte, PageSize)
	if err := handle.ReadPage(pageID, buf); err == nil {
		end := int(startOff) + int(allocSize)
		if end > len(buf) {
			end = len(buf)
		}
		if int(startOff) < end {
			data = append(data, buf[startOff:end]...)
		}
	}

	// 2. 续块链 (共享扩展页): [SchemaFragHdr 8B][payload]
	for pid, off := extPage, extOff; pid != 0; {
		fb := make([]byte, PageSize)
		if err := handle.ReadPage(pid, fb); err != nil {
			break
		}
		if int(off)+schemaFragHdrSize > len(fb) {
			break
		}
		frag := readSchemaFragHdr(fb, int(off))
		start := int(off) + schemaFragHdrSize
		end := start + int(frag.DataLen)
		if end > len(fb) {
			break
		}
		data = append(data, fb[start:end]...)
		pid, off = frag.NextPage, frag.NextOff
	}

	if len(data) < 2 {
		return 0, nil, nil
	}

	pos := 0
	activeVer := data[pos]
	pos++
	masterCount := int(data[pos])
	pos++

	// 读取 master fields: [type(1B)][flag(1B)][nameLen(1B)][name...]
	// flag 是列级标志 (删除/可空/唯一 + decimal 的 scale), 见 types.FieldInfo.Flag。
	// 语义变了但字节位置与顺序不变: 必须与 serializeSchema 严格对照。
	var masterFields []types.FieldInfo
	for i := 0; i < masterCount && pos+3 <= len(data); i++ {
		typ := types.FieldType(data[pos])
		flag := data[pos+1]
		nameLen := int(data[pos+2])
		pos += 3
		if pos+nameLen > len(data) {
			break
		}
		name := make([]byte, nameLen)
		copy(name, data[pos:pos+nameLen])
		pos += nameLen
		masterFields = append(masterFields, types.FieldInfo{FieldType: typ, FieldName: name, Flag: flag})
	}

	// 读取版本定义
	var versions []SchemaVersion
	if pos < len(data) {
		verCount := int(data[pos])
		pos++
		for i := 0; i < verCount && pos < len(data); i++ {
			verNo := data[pos]
			pos++
			if pos >= len(data) {
				break
			}
			fieldCount := int(data[pos])
			pos++
			indices := make([]byte, fieldCount)
			if pos+fieldCount <= len(data) {
				copy(indices, data[pos:pos+fieldCount])
				pos += fieldCount
			}
			versions = append(versions, SchemaVersion{Version: verNo, FieldIndices: indices})
		}
	}

	return activeVer, masterFields, versions
}

type Table struct {
	info  *TableInfo
	Field []types.FieldInfo // master registry (所有字段,含已删除,删除标记见 types.FlagDeleted)

	// db 所属 DB (BlobRead/BlobFree 需要它取 file / pendingFreeChain).
	// 创建时由 db.tables.Create 注入, 运行时不变.
	db               *DB
	handle           page.HandleDb
	indexMeta        *IndexMetaPage
	tablePageID      uint32 // 表ID
	tableStartOffset uint32 // 表所在页的偏移

	// schema 续块所在的「全库共享扩展页」分配器 (由 TablePage 注入, 各表同一实例)
	schemaExt *SchemaExtAllocator

	// schema 版本管理
	activeVersion byte
	versions      []SchemaVersion

	// 写操作通知回调 (防抖自动提交)
	onWrite func()

	// 自增主键游标 (内存态, 详见 AllocPKID/ensureNextPK):
	// 不能再现算 TotalRows+1 —— 删除会让行数回退, 编号随之回退并撞上存活的 pkID。
	nextPK       uint64
	nextPKInit   bool
	pendingPK    uint64 // AllocPKID 刚预留、等待 InsertRowWithID 消费的号
	hasPendingPK bool
	// pkMu 保护自增主键游标的读-改-写 (ensureNextPK / nextPK++ / pendingPK 预留):
	// 多写者并发 AllocPKID 时, 无锁的 nextPK++ 会给两行发出同一个号 (重复主键)。
	pkMu sync.Mutex

	// fieldIdxCache: 列名 → table.Field 下标的惰性缓存。条件求值
	// (Eq/In/Gt... 的 Match) 每行每条件都查一次字段位置, 线性扫是 O(列数)/行;
	// 表字段在建模后基本不变, 缓存后 O(1)。Field 切片身份变化 (AddField 追加 /
	// ReloadInfos 整体换片) 时按首元素指针 + 长度自动重建, 无需手工失效。
	// fieldIdxMu 保护首次构建: 引擎允许并发读, 两个读 goroutine 可能同时建缓存。
	fieldIdxMu    sync.Mutex
	fieldIdxCache map[string]int
	fieldIdxSrc   []types.FieldInfo // 建缓存时的 Field 切片 (身份比对用)
}

// TotalRows 返回表的总行数
func (table *Table) TotalRows() uint64 { return table.info.TotalRows }

// TablePageID 返回表页 ID (调试用)
func (table *Table) TablePageID() uint32 { return table.tablePageID }

// TableStartOffset 返回表信息在页中的偏移 (调试用)
func (table *Table) TableStartOffset() uint32 { return table.tableStartOffset }

// IndexMeta 返回索引元数据页 (调试用)
func (table *Table) IndexMeta() *IndexMetaPage { return table.indexMeta }

// SetOnWrite 设置写操作通知回调 (防抖自动提交)
func (table *Table) SetOnWrite(fn func()) { table.onWrite = fn }

// findFieldIndex 查找字段在 master registry 中的索引
//
// 走 fieldIndexMap 的惰性缓存: Match 类条件是"每行每条件一次"的热路径, 线性扫
// O(列数) 在宽表 + 全表扫描下会累积成可见开销。缓存按 Field 切片身份
// (首元素指针 + 长度) 自动失效, AddField / ReloadInfos 后下一次调用即重建。
func (table *Table) findFieldIndex(name string) int {
	if idx, ok := table.fieldIndexMap()[name]; ok {
		return idx
	}
	return -1
}

// fieldIndexMap 返回 (必要时重建) 列名 → Field 下标缓存。
func (table *Table) fieldIndexMap() map[string]int {
	src := table.Field
	if table.fieldIdxCache != nil && len(table.fieldIdxSrc) == len(src) &&
		(len(src) == 0 || &table.fieldIdxSrc[0] == &src[0]) {
		return table.fieldIdxCache
	}
	table.fieldIdxMu.Lock()
	defer table.fieldIdxMu.Unlock()
	// 双检: 等待锁期间另一读线程可能已建好
	if table.fieldIdxCache != nil && len(table.fieldIdxSrc) == len(src) &&
		(len(src) == 0 || &table.fieldIdxSrc[0] == &src[0]) {
		return table.fieldIdxCache
	}
	m := make(map[string]int, len(src))
	for i := range src {
		m[string(src[i].FieldName)] = i
	}
	table.fieldIdxCache = m
	table.fieldIdxSrc = src
	return m
}

// fieldWidth 取某列的行内宽度 (字节) —— 引擎内所有行偏移计算的唯一入口。
//
// 为什么不能直接 types.GetSize(f.FieldType): 字符串列的行内宽度由【列自己声明】
// (Flag 低 5 位的宽度码, 上限 types.MaxInlineColumnWidth), 只看类型会一律算成 256,
// 于是该列之后的所有列都从错位偏移解码。这里统一经 FieldInfo.Width()。
//
// 返回 0 表示索引越界或列级参数非法 —— 调用方必须当错误处理: 0 宽度会让整行错位。
func (table *Table) fieldWidth(masterIdx int) uint16 {
	if masterIdx < 0 || masterIdx >= len(table.Field) {
		return 0
	}
	return table.Field[masterIdx].Width()
}

func (table *Table) GetFieldCount() byte       { return table.info.FieldCount }
func (table *Table) GetActiveFieldCount() byte { return table.info.ActiveFieldCount }

// GetActiveFieldNames 返回当前活跃版本中包含的所有字段名 (不含已删除)
func (table *Table) GetActiveFieldNames() []string {
	sv := table.getVersion(table.activeVersion)
	if sv == nil {
		return nil
	}
	names := make([]string, 0, len(sv.FieldIndices))
	for _, idx := range sv.FieldIndices {
		if table.Field[idx].IsDropped() {
			continue // 跳过已删除
		}
		names = append(names, string(table.Field[idx].FieldName))
	}
	return names
}

// GetActiveFieldType 返回指定字段的 FieldType (基于 master registry)
func (table *Table) GetActiveFieldType(name string) (types.FieldType, bool) {
	for _, f := range table.Field {
		if string(f.FieldName) == name && !f.IsDropped() {
			return f.FieldType, true
		}
	}
	return 0, false
}

// GetActiveFieldFlag 返回指定字段的列级标志 (含 Decimal 的 scale 与可空位)。
//
// 与 GetActiveFieldType 配对: 类型一致但 scale 不一致是"同列不同 scale"这种会被
// 静默破坏保序前提的情况, 调用方 (ORM 重开表时) 必须能读到它来做一致性校验。
func (table *Table) GetActiveFieldFlag(name string) (byte, bool) {
	for i := range table.Field {
		f := &table.Field[i]
		if string(f.FieldName) == name && !f.IsDropped() {
			return f.Flag, true
		}
	}
	return 0, false
}
func (table *Table) GetFieldAllocateSize() uint16 { return table.info.FieldAllocateSize }
func (table *Table) GetFirstData() uint32         { return table.info.FirstData }

// GetSchemaExtRef 返回本表首个 schema 续块的位置 (pageID, off); pageID == 0 表示无续块
func (table *Table) GetSchemaExtRef() (uint32, uint16) {
	return table.info.ExtPage, table.info.ExtOff
}

func (table *Table) GetFields() []types.FieldInfo {
	return table.Field
}

func (table *Table) findFieldByName(name []byte) *types.FieldInfo {
	target := util.BytesToString(name)
	for i := range table.Field {
		if util.BytesToString(table.Field[i].FieldName) == target {
			return &table.Field[i]
		}
	}
	return nil
}

// fieldParam 取列级类型参数 (Flag 低 5 位): decimal 是 scale, 字符串家族是行内宽度码,
// 其余类型为 0 (由 types.ValidateColumnParam 保证)。
//
// 以前这里叫 fieldScale 且只服务 decimal; 字符串列也用上同一格之后, "取参数"与
// "算宽度"必须成对 —— 只传 scale 会漏掉宽度码, 编出来的字节宽度与行内不一致。
func (table *Table) fieldParam(masterIdx int) byte {
	if masterIdx < 0 || masterIdx >= len(table.Field) {
		return 0
	}
	return table.Field[masterIdx].Param()
}

// AllocPKID 分配并预留下一个自增 pkID。
//
// 供"必须先把号写进对象、再序列化整行"的上层使用 (典型是 ORM:
// 只有先拿到 pkID 填进 _id 列, 存进行里的 _id 才与真实 pkID 一致)。
// 分配即推进游标并预留该号, 紧随其后的 InsertRowWithID 会跳过查重。
// 若最终未插入, 该号只是被跳过 —— 游标只增不减, 不会与后续插入冲突。
func (table *Table) AllocPKID() uint64 {
	// 整个"惰性初始化 → 取号 → 游标推进 → 预留登记"在锁内完成:
	// 并发写者各拿各的号, 号段互不重叠。
	table.pkMu.Lock()
	defer table.pkMu.Unlock()
	table.ensureNextPK()
	pk := table.nextPK
	table.nextPK++
	table.pendingPK = pk
	table.hasPendingPK = true
	return pk
}

// ensureNextPK 惰性初始化自增游标: 只增不减, 从行存储里当前最大的存活 pkID 之后继续。
//
// 旧实现 (TotalRows+1) 删除后会回退编号, 与仍存活的 pkID 撞车 (探针 T4)。
// 首次使用时取一次当前最大 pkID —— 每次开表只算一次, 之后 O(1)。
func (table *Table) ensureNextPK() {
	if table.nextPKInit {
		return
	}
	table.nextPK = table.maxRowStorePK() + 1
	if table.nextPK == 0 {
		table.nextPK = 1
	}
	table.nextPKInit = true
}

// maxRowStorePK 返回行存储索引中当前最大的存活 pkID (无行或没有行存储索引时为 0)。
// DataPage 沿 B+ 树叶链按 pkID 升序排列, 故从最后一页往回找第一个非空页即可;
// 每页内取最大 pkID 以兼容页内乱序的极端情况。
func (table *Table) maxRowStorePK() uint64 {
	primMeta, primHandler, ok := table.indexMeta.rowStoreMeta()
	if !ok {
		return 0
	}
	ids := primHandler.GetDataPageIDs(table.handle, primMeta.EntryPage)
	for i := len(ids) - 1; i >= 0; i-- {
		rows := primHandler.ReadDataPageRows(table.handle, ids[i])
		var maxPK uint64
		for _, r := range rows {
			if len(r) < 8 {
				continue
			}
			if pk := binary.BigEndian.Uint64(r[:8]); pk > maxPK {
				maxPK = pk
			}
		}
		if maxPK != 0 {
			return maxPK
		}
	}
	return 0
}