TsDB2/DB/core/table_schema.go

package core

import (
	"fmt"
	"sync"
	"unsafe"

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

// validateFieldsForSchema 校验一组列定义能否构成合法 schema:
// 列名长度可落盘 + 列级参数自洽 + 行宽不超建列闸门。
//
// 校验落在 core 而不是只留在 ORM / migrate: schema 也能从 SQL 或低层 API 直接建出来,
// 只有拦在真正写 schema 的地方才拦得住。
func validateFieldsForSchema(fields []types.FieldInfo) error {
	for _, f := range fields {
		// 列名长度: schema 里是 [nameLen(1B)][name...] (见 serializeSchema / readSchema)。
		// 长度为 0 或 >255 都不能落盘 —— 0 会让解析侧把下一个字段的类型字节当名字,
		// >255 会让 nameLen 截断并让其后所有字段整体错位。都不是"名字有瑕疵", 是整表错乱。
		if len(f.FieldName) == 0 {
			return fmt.Errorf("%w: 列名长度为 0 (schema 里列名长度是 1 字节, 空名字无法落盘)",
				types.ErrValueRange)
		}
		if len(f.FieldName) > FieldNameSize {
			return fmt.Errorf("%w: 列名 %q 有 %d 字节, 上限 %d 字节 "+
				"(schema 里列名长度是 1 字节, 超长会截断并让后续列整体错位)",
				types.ErrValueRange, f.FieldName, len(f.FieldName), FieldNameSize)
		}
		if err := types.ValidateColumnParam(f.FieldType, f.Param()); err != nil {
			return fmt.Errorf("列 %q: %w", f.FieldName, err)
		}
	}
	return types.CheckRowSize(fields)
}

// nextSchemaVersion 返回下一个可用的 schema 版本号。
//
// schemaVer 是每行行首的 1 字节 (getVersionedOffset / rowDataSizeForVersion 都按它取版本),
// 版本条数也是 1 字节落盘 (serializeSchema 的 verCount)。到 0xFF 再自增就会回绕成 0:
// 0 不是任何已知版本, getVersion 返回 nil → 该行按"未知版本"解析, NULL 位图取不到,
// 所有列静默读成 0 / NULL。宁可在演进这一步拒绝, 让用户重建表。
//
// 每次 AddField / DropField 都会新增一版, 因此这道闸门也限定了"一张表的 schema
// 最多能演进多少次"。
func (table *Table) nextSchemaVersion() (byte, error) {
	if table.activeVersion >= 0xFF || len(table.versions) >= 0xFF {
		return 0, fmt.Errorf("schema 版本数已达上限 %d (activeVer=%d, 已存 %d 版): "+
			"版本号是行首 1 字节, 再演进会回绕并让旧行解析错位; 请重建该表",
			0xFF, table.activeVersion, len(table.versions))
	}
	return table.activeVersion + 1, nil
}

// findVersion 在版本列表中线性查找指定版本的 schema 定义。
// getVersion (Table 方法) 与 extractIndexKey (自由函数, 手里没有 *Table) 共用这条扫描。
func findVersion(versions []SchemaVersion, ver byte) *SchemaVersion {
	for i := range versions {
		if versions[i].Version == ver {
			return &versions[i]
		}
	}
	return nil
}

// getVersion 获取指定版本的 schema 定义
func (table *Table) getVersion(ver byte) *SchemaVersion {
	return findVersion(table.versions, ver)
}

// svCol0WidthOf svCol0Width 的自由函数形态: fields 由调用方传入 (extractIndexKey 无
// *Table 接收者, 手里只有调用点传进来的 table.Field)。
func svCol0WidthOf(fields []types.FieldInfo, sv *SchemaVersion) int {
	if sv == nil || len(sv.FieldIndices) == 0 {
		return 0
	}
	idx := int(sv.FieldIndices[0])
	if idx < 0 || idx >= len(fields) {
		return 0
	}
	return int(fields[idx].Width())
}

// svCol0Width 某版本行的【列 0 档位宽度】W0 (由 schema 唯一确定)。
//
// v9 行布局把列 0 固定按档位宽度存储 (见 types/rowvar.go), 偏移表基址 = 1 + W0。
// 解析端必须能静态算出 W0 —— 这正是"列 0 不参与变长存储"的原因: 若列 0 也按实际
// 长度存储, 偏移表的位置就取决于内容, 解析将不得不先猜列 0 有多长 (自举问题)。
func (table *Table) svCol0Width(sv *SchemaVersion) int {
	return svCol0WidthOf(table.Field, sv)
}

// rowLayout 行布局解析前导的公共产物: 版本定义 + 列 0 档位宽度 + 行内偏移表。
// NULL 位图不在结构体内 —— 按调用方需要再调 types.RowNullBitmap(rowData, ends, n),
// 因为并非所有调用方都要位图 (如 getVersionedRange 走 types.RowIsNull 逐列判定)。
type rowLayout struct {
	sv        *SchemaVersion
	col0Width int
	ends      []uint16
}

// parseRowLayoutOf 行布局解析前导的唯一实现, 四步: ver → findVersion → svCol0WidthOf →
// RowParseOffsets。任一步失败 ok=false, 调用方按各自的失败语义返回。
// 调用方必须保证 len(rowData) ≥ 1 —— 各调用点原有的长度检查保持不动。
func parseRowLayoutOf(fields []types.FieldInfo, versions []SchemaVersion, rowData []byte, ver byte) (rowLayout, bool) {
	sv := findVersion(versions, ver)
	if sv == nil {
		return rowLayout{}, false
	}
	n := len(sv.FieldIndices)
	col0Width := svCol0WidthOf(fields, sv)
	ends, ok := types.RowParseOffsets(rowData, n, col0Width)
	if !ok {
		return rowLayout{}, false
	}
	return rowLayout{sv: sv, col0Width: col0Width, ends: ends}, true
}

// parseRowLayout 从行首字节读出版本号, 解析行布局前导 (行解码 / 更新 / 删行路径的入口)。
func (table *Table) parseRowLayout(rowData []byte) (rowLayout, bool) {
	return parseRowLayoutOf(table.Field, table.versions, rowData, rowData[0])
}

// parseRowLayoutVer 与 parseRowLayout 相同, 但版本号由调用方显式给出
// (版本号不取自 rowData[0] 时, 如 versionedNullBitmap 的 ver 参数)。
func (table *Table) parseRowLayoutVer(rowData []byte, ver byte) (rowLayout, bool) {
	return parseRowLayoutOf(table.Field, table.versions, rowData, ver)
}

// rowCol0Width 指定 schemaVer 对应行的列 0 档位宽度。
func (table *Table) rowCol0Width(ver byte) int {
	return table.svCol0Width(table.getVersion(ver))
}

// getVersionedOffset 计算某行中指定 master 字段索引在行数据中的偏移。
//
// 行布局 v9: [schemaVer 1B][列0 W0 B][列结束偏移表 2N B][列1..N-1 内容][NULL 位图 ⌈n/8⌉ B]
// 列位置一律由【行内偏移表】给出 (见 getVersionedRange), 位图长度按【行自身的
// schemaVer】算 (n = 该版本 FieldIndices 长度), 不能取当前字段数。
//
// 时序索引按【绝对偏移 9】(== [pkID 8B][schemaVer 1B]) 读首列当时间键
// (见 DB/types.go 的 modelInfo 与 core/index 的 timeRowInRange / FindRowByIndexKey)。
// 这条能成立, 靠的是"列 0 恒占其档位宽度 W0 且时间键列恒为 8 字节" —— 不是靠"位图放行尾":
// 位图若前插会把首列整体后移 ⌈n/8⌉ 字节, 时序键读取全错, 所以位图必须落在字段内容之后。
//
// 字段为 NULL 时返回 false。把 NULL 判定下沉到这里, 是因为"定位偏移"是所有读取路径
// (7 个条件类 + 更新时摘旧堆块) 的唯一入口, 各调用方再各判一次必然漏:
//   - 条件匹配: NULL 不匹配任何比较 (SQL 三值逻辑)。若按全零字节比, `x = 0` 会命中
//     NULL 行、`x < 1` 也会命中 —— 结果看着"有数据"却全是错的。
//   - 更新路径: 旧值为 NULL 时引用槽是全零, 本就没有堆块要摘。
func (table *Table) getVersionedOffset(rowData []byte, masterFieldIdx int) (uint16, bool) {
	start, _, ok := table.getVersionedRange(rowData, masterFieldIdx)
	return start, ok
}

// getVersionedRange 定位某行中 masterFieldIdx 列内容的 [起点, 终点)。
//
// 【v9 起改走行内偏移表】旧实现从偏移 1 开始"累加各列 Width"推算列位置, 那是建立在
// "字段区定长、每列实付档位宽度"的前提上。v9 让字符串列按实际长度存储 (见
// types/rowvar.go), 累加 Width 会越算越偏 —— 必须由行内偏移表给出精确位置。
//
// 终点是行内【实际】字节的结尾, 不再等于 起点+Width: 调用方若需要"档位宽度"的定长
// 视角 (比较 / 索引键), 必须自己用 types.PadToWidth 补齐。
//
// 字段为 NULL 时返回 false。
func (table *Table) getVersionedRange(rowData []byte, masterFieldIdx int) (uint16, uint16, bool) {
	if len(rowData) < 1 {
		return 0, 0, false
	}
	layout, ok := table.parseRowLayout(rowData)
	if !ok {
		return 0, 0, false
	}
	sv, n, col0Width, ends := layout.sv, len(layout.sv.FieldIndices), layout.col0Width, layout.ends
	for i, idx := range sv.FieldIndices {
		if int(idx) != masterFieldIdx {
			continue
		}
		if types.RowIsNull(rowData, ends, n, i) {
			return 0, 0, false // NULL: 不参与比较, 也没有堆块
		}
		return types.RowFieldStart(ends, i), types.RowFieldEnd(ends, i, col0Width), true
	}
	return 0, 0, false
}

// versionedNullBitmap 取出行尾的 NULL 位图切片 (长度按该行自身的 schemaVer 算)。
//
// 行布局 v9: [schemaVer 1B][列0 W0 B][列结束偏移表 2N B][列1..N-1 内容][NULL 位图 ⌈n/8⌉ B]
//
// 位图起点由偏移表末项给出, 不再靠"版本算出的总长" —— v9 行按内容变长, 不存在
// "版本对应的总长"。而【为什么仍然不能用 len(rowData)】: 行存储的原地更新允许把新行
// 写进旧行的槽位 (剩余字节清零), 按 len 取尾会取到零填充而不是真正的位图。偏移表落在
// 只取决于字段数的固定位置上, 不受零填充影响, 因此始终精确。
func (table *Table) versionedNullBitmap(rowData []byte, ver byte) ([]byte, bool) {
	layout, ok := table.parseRowLayoutVer(rowData, ver)
	if !ok {
		return nil, false
	}
	n := len(layout.sv.FieldIndices)
	return types.RowNullBitmap(rowData, layout.ends, n)
}

// getActiveFieldTypes 返回当前活跃版本的字段类型列表 (不含已删除)
func (table *Table) getActiveFieldTypes() []types.FieldInfo {
	sv := table.getVersion(table.activeVersion)
	if sv == nil {
		return nil
	}
	fields := make([]types.FieldInfo, len(sv.FieldIndices))
	for i, idx := range sv.FieldIndices {
		fields[i] = table.Field[idx]
	}
	return fields
}

// calcRowDataSize 计算当前活跃版本一行的【最大】字节数。
func (table *Table) calcRowDataSize() int {
	return table.rowDataSizeForVersion(table.activeVersion)
}

// rowDataSizeForVersion 计算指定 schema 版本一行的【最大】字节数。
//
// 【v9 起语义变化】行按内容变长, 这个值只是上界 (每列都取满档位宽度时的长度), 不再是
// "该版本行的实际长度"。它只用于"按行分配缓冲"的场景 (见 maxRowDataSize): 拿它当实际
// 行长去做闸门判断, 会把本来装得下的行误判为超长。
//
// 上界 = schemaVer(1) + Σ 各列档位宽度 + 列结束偏移表(2N) + NULL 位图(⌈N/8⌉)。
// 列 0 的 8 字节本身就在 Σ 里 (它是真实列, 不是额外前缀)。
func (table *Table) rowDataSizeForVersion(ver byte) int {
	size := 1 // schemaVer
	sv := table.getVersion(ver)
	if sv != nil {
		n := len(sv.FieldIndices)
		size += types.NullBitmapLen(n)
		size += types.RowOffTableLen(n)
		for _, idx := range sv.FieldIndices {
			size += int(table.Field[idx].Width())
		}
	}
	return size
}

// minRowDataSizeForVersion 该版本一行的【最小】字节数 (列 0 之外全部为 NULL 时)。
//
// 列 0 恒占其档位宽度 W0 (偏移表基址恒为 1+W0, 这是解析与"列 0 定长"的前提), 所以最小值
// 就是表头 (schemaVer + 列 0 + 偏移表) 加位图。
func (table *Table) minRowDataSizeForVersion(ver byte) int {
	sv := table.getVersion(ver)
	if sv == nil {
		return 1
	}
	n := len(sv.FieldIndices)
	return types.RowHeaderSize(n, table.svCol0Width(sv)) + types.NullBitmapLen(n)
}

// maxRowDataSize 返回所有已知 schema 版本中最大的行数据大小。
//
// 给"按行分配缓冲"的调用方使用 (如删除路径的行复用池): info.RowSize 只反映最近
// 一次落盘时的 schema, AddField/DropField 之后会偏小, 沿用它会 slice 越界 panic。
func (table *Table) maxRowDataSize() int {
	maxSize := table.rowDataSizeForVersion(table.activeVersion)
	for i := range table.versions {
		if n := table.rowDataSizeForVersion(table.versions[i].Version); n > maxSize {
			maxSize = n
		}
	}
	return maxSize
}

// syncRowSize 让 info.RowSize 反映所有 schema 版本中最宽的一行
// (语义同 types.CalcRowSize: Σ 字段字节数, 不含 schemaVer、NULL 位图与 pkID 前缀)。
// 仅在 schema 变更时调用。
func (table *Table) syncRowSize() {
	// 前缀 = schemaVer(1B) + 该版本的 NULL 位图。位图长度按活跃版本算 —— RowSize 本就
	// 只是"字段宽度之和"的估计值 (删除/分页路径都明确不用它, 见 deleter.go 里的说明),
	// 不必为跨版本差一格去逐版本精确。
	prefix := 1
	if sv := table.getVersion(table.activeVersion); sv != nil {
		prefix += types.NullBitmapLen(len(sv.FieldIndices))
	}
	if size := table.maxRowDataSize() - prefix; size > 0 && size <= 0xFFFF {
		table.info.RowSize = uint16(size)
	}
}

// pickVersionForUpdate 决定更新后按哪一版 schema 落盘。
//
// 惰性演进语义:
//   - 本次要写的字段在旧行自身的版本里都存在 → 保持旧版本。列序与偏移表大小都不变,
//     不会因为"只改了一个老字段"而无谓地把行升级并搬迁。
//     注意这不保证【行长】不变: v9 字符串列按实际长度存, 换一个更长的值就会让行变长,
//     由索引层负责"原地 → 页内重建 → 换页"三级兜底。
//   - values 触及了旧版本没有的字段 (AddField 之后新增的) → 升级到当前版本;
//     此时列数变化 (偏移表也变), 由索引层负责"页内重建 / 换页"。
func (table *Table) pickVersionForUpdate(oldRowData []byte, values []types.FieldValue) byte {
	if len(oldRowData) < 1 {
		return table.activeVersion
	}
	oldVer := oldRowData[0]
	if oldVer == table.activeVersion {
		return oldVer
	}
	sv := table.getVersion(oldVer)
	if sv == nil {
		// 旧版本元信息缺失, 只能按当前版本重写
		return table.activeVersion
	}
	for _, fv := range values {
		idx := table.findFieldIndex(string(fv.FieldName))
		if idx < 0 {
			return table.activeVersion
		}
		inOldVer := false
		for _, i := range sv.FieldIndices {
			if int(i) == idx {
				inOldVer = true
				break
			}
		}
		if !inOldVer {
			return table.activeVersion
		}
	}
	return oldVer
}

// serializeSchema 将当前 schema 序列化为字节流
//
// master field 每项: [type(1B)][flag(1B)][nameLen(1B)][name...]
// flag 落盘是必须的: decimal 列的 scale 一旦丢失, 读取侧就只能信任行内的 scale,
// 于是"同一列出现两种 scale"这种破坏保序的情况再也无法被发现。删除/可空两个状态位
// 同样必须随 schema 落盘, 否则重开库后被 DropField 的列会悄悄"复活"。
func (table *Table) serializeSchema() []byte {
	var data []byte
	data = append(data, table.activeVersion)
	data = append(data, byte(len(table.Field)))
	for _, f := range table.Field {
		data = append(data, byte(f.FieldType))
		data = append(data, f.Flag)
		data = append(data, byte(len(f.FieldName)))
		data = append(data, f.FieldName...)
	}
	data = append(data, byte(len(table.versions)))
	for _, sv := range table.versions {
		data = append(data, sv.Version)
		data = append(data, byte(len(sv.FieldIndices)))
		data = append(data, sv.FieldIndices...)
	}
	return data
}

// writeSchema 整条重写本表的 schema 定义
//
// 布局:
//  1. 内联块 —— 表页内, 紧跟 TableInfo, 固定 FieldAllocateSize 字节, **无跳转槽**。
//     建表时按 len(serializeSchema()) 精确分配 (上限 MaxInlineSchemaSize)。
//  2. 续块   —— schema 超出内联块的部分, 写入「全库共享」的 schema 扩展页,
//     每块格式 [SchemaFragHdr 8B][payload], 多块由 hdr.NextPage/NextOff 串成链。
//     本表首个续块的位置记录在 TableInfo.ExtPage/ExtOff。
//
// 续块只追加、不单独释放: 每次都是整条重写, 旧续块立刻成为垃圾, 由
// SchemaExtAllocator 在尾页放不下时做「满页压缩重排」统一回收。因此这里
// 不再需要任何引用计数 —— 这也正是旧实现最容易出错的地方。
func (table *Table) writeSchema() error {
	data := table.serializeSchema()
	alloc := int(table.info.FieldAllocateSize) // 内联块容量 (可能大于 schema 实际长度)
	inline := len(data)
	if inline > alloc {
		inline = alloc // 容错: schema 超出内联块, 余下部分走续块
	}
	rest := data[inline:]

	// 1. 先分配续块。可能触发压缩重排 —— 那会改写(可能同页的)其他表的 TableInfo,
	//    所以内联块的写入必须放到分配之后再做, 否则会拿旧页缓冲覆盖掉这些更新。
	var frags []schemaFrag
	if len(rest) > 0 {
		if table.schemaExt == nil {
			return ErrFieldSpaceNotEnough
		}
		var err error
		frags, err = table.schemaExt.AllocChain(len(rest))
		if err != nil {
			return err
		}
	}

	// 2. 写续块
	if len(frags) > 0 {
		if err := table.writeFrags(frags, rest); err != nil {
			return err
		}
	}

	// 3. 写内联块 (重新读页, 见上面第 1 步的说明)
	buf := make([]byte, PageSize)
	if err := table.handle.ReadPage(table.tablePageID, buf); err != nil {
		return err
	}
	start := int(table.tableStartOffset + uint32(unsafe.Sizeof(*table.info)))
	end := start + alloc
	if end > len(buf) {
		return ErrFieldSpaceNotEnough
	}
	// 清零整块(alloc 字节): 槽位可能大于 schema 实际长度(复用空洞/预分配),
	// 多出来的部分是填充, 必须清零, 防止旧表的残留字节被后续误读
	for i := start; i < end; i++ {
		buf[i] = 0
	}
	copy(buf[start:end], data[:inline])
	if err := table.handle.SavePage(table.tablePageID, buf); err != nil {
		return err
	}

	// 4. 记录本表首个续块的位置
	if len(frags) > 0 {
		table.info.ExtPage = frags[0].PageID
		table.info.ExtOff = uint16(frags[0].Off)
	} else {
		table.info.ExtPage = 0
		table.info.ExtOff = 0
	}
	return nil
}

// writeSchemaExt 把本表 schema 的续块部分写进指定的扩展页分配器, 并落盘 TableInfo
// 专供 SchemaExtAllocator.Compact 压缩重排时调用。
func (table *Table) writeSchemaExt(alloc *SchemaExtAllocator) error {
	data := table.serializeSchema()
	inline := int(table.info.FieldAllocateSize)
	if inline > len(data) {
		inline = len(data)
	}
	rest := data[inline:]

	if len(rest) == 0 {
		table.info.ExtPage = 0
		table.info.ExtOff = 0
		table.saveTableInfo()
		return nil
	}

	frags, err := alloc.AllocChain(len(rest))
	if err != nil {
		return err
	}
	if err := table.writeFrags(frags, rest); err != nil {
		return err
	}
	table.info.ExtPage = frags[0].PageID
	table.info.ExtOff = uint16(frags[0].Off)
	table.saveTableInfo()
	return nil
}

// writeFrags 把 rest 按 frags 描述的容量逐块写入扩展页, 并串好续块链
func (table *Table) writeFrags(frags []schemaFrag, rest []byte) error {
	written := 0
	for i := range frags {
		f := frags[i]
		buf := make([]byte, PageSize)
		if err := table.handle.ReadPage(f.PageID, buf); err != nil {
			return err
		}
		h := SchemaFragHdr{DataLen: f.Len}
		if i+1 < len(frags) {
			h.NextPage = frags[i+1].PageID
			h.NextOff = uint16(frags[i+1].Off)
		}
		putSchemaFragHdr(buf, int(f.Off), h)
		copy(buf[int(f.Off)+schemaFragHdrSize:], rest[written:written+int(f.Len)])
		if err := table.handle.SavePage(f.PageID, buf); err != nil {
			return err
		}
		written += int(f.Len)
	}
	return nil
}

// checkProspectiveRowSize 预演"把 newField 加进当前活跃字段集合之后"的行宽。
//
// 加列只会让行变宽, 而超宽的行写不进 DataPage (见 serializeRowInternal 的写行闸门),
// 所以必须在建列这一步就拒绝: 否则这张表从加列成功的那一刻起, 一条也写不进去 ——
// 而且是插数据时才暴露 (而且加列本身是"成功"的, 没有任何回滚线索)。
func (table *Table) checkProspectiveRowSize(newField types.FieldInfo, newMasterIdx int) error {
	sv := table.getVersion(table.activeVersion)
	ineligible := table.spillIneligibleFields()
	probe := make([]types.FieldInfo, 0, len(table.Field)+1)
	unspillable := make([]bool, 0, len(table.Field)+1)
	if sv != nil {
		for _, idx := range sv.FieldIndices {
			if int(idx) >= len(table.Field) {
				continue
			}
			probe = append(probe, table.Field[idx])
			unspillable = append(unspillable, ineligible[idx])
		}
	}
	// 新增/复活的这一列此刻通常还没有索引 (建索引是独立一步), 但复活的情形下它可能
	// 已被某个仍活跃的索引引用 —— 那就同样不能再溢出, 按真实集合算。
	probe = append(probe, newField)
	if newMasterIdx >= 0 && newMasterIdx < len(ineligible) {
		unspillable = append(unspillable, ineligible[newMasterIdx])
	} else {
		unspillable = append(unspillable, false)
	}
	if err := types.CheckRowSpillFloor(probe, unspillable); err != nil {
		return fmt.Errorf("新增列 %q 后行宽超限: %w", newField.FieldName, err)
	}
	return nil
}

// spillIneligibleFields 返回"不能值级溢出的列"集合 (按 master registry 下标)。
//
// 建了索引的列必须留在行内: 索引键按行内字节 (补齐到档位) 拼, 一旦该列被溢出成 6B
// 堆引用, 拼出的键就是页号/偏移而不是值 —— 索引里挂着一个永远查不到的键。所以凡是
// 任何活跃索引的分量列, 都不参与值级溢出。
//
// 列 0 不在这里标记: 它本就恒占档位宽度 (偏移表基址依赖它, 见 types.CalcSpillFloor
// 的 live==0 分支), 任何情况下都不参与溢出。
func (table *Table) spillIneligibleFields() []bool {
	n := len(table.Field)
	if n == 0 {
		return nil
	}
	out := make([]bool, n)
	metas := table.indexMeta.metas
	for i := range metas {
		if metas[i].Deleted() {
			continue
		}
		for _, fid := range metas[i].Fields() {
			if int(fid) < n {
				out[fid] = true
			}
		}
	}
	return out
}

// rowSpillFloorForVersion 计算某 schema 版本一行经【值级溢出】之后的最小可能长度。
//
// 它就是写入闸门 (serializeRowInternal 开头) 的判据: 下界超物理墙说明"把所有可溢出的
// 字符串列都溢出去也装不下", 不动堆永远救不回来 —— 且这个判断发生在动堆之前 (见那里
// 的说明)。下界按【不可溢出列: 建了索引的列 + 列 0 + 定长列全宽】+ 可溢出字符串 6B 计。
func (table *Table) rowSpillFloorForVersion(ver byte) int {
	sv := table.getVersion(ver)
	if sv == nil || len(sv.FieldIndices) == 0 {
		return 1 // 只有 schemaVer 1 字节
	}
	ineligible := table.spillIneligibleFields()
	fields := make([]types.FieldInfo, 0, len(sv.FieldIndices))
	unspillable := make([]bool, 0, len(sv.FieldIndices))
	for _, idx := range sv.FieldIndices {
		if int(idx) >= len(table.Field) {
			continue
		}
		fields = append(fields, table.Field[idx])
		unspillable = append(unspillable, ineligible[idx])
	}
	return types.CalcSpillFloor(fields, unspillable)
}

// checkIndexSpillFloor 预演"把 fieldIDs 作为索引分量"之后, 活跃版本行的溢出下界。
//
// 分量列一旦建索引就不能再溢出, 下界随之抬升; 若抬过"一页两行"的配额 (MaxInlineRowData),
// 这张表从建索引成功起就再也压不进两行 —— 必须在建索引这一步拒绝 (ErrIndexRowTooWide),
// 而不是等某次插入悄悄把行撑到 8159..16332 之间、整页行为退化时才被发现。
func (table *Table) checkIndexSpillFloor(fieldIDs []byte) error {
	sv := table.getVersion(table.activeVersion)
	if sv == nil {
		return nil
	}
	ineligible := table.spillIneligibleFields()
	for _, fid := range fieldIDs {
		if int(fid) < len(ineligible) {
			ineligible[fid] = true
		}
	}
	fields := make([]types.FieldInfo, 0, len(sv.FieldIndices))
	unspillable := make([]bool, 0, len(sv.FieldIndices))
	for _, idx := range sv.FieldIndices {
		if int(idx) >= len(table.Field) {
			continue
		}
		fields = append(fields, table.Field[idx])
		unspillable = append(unspillable, ineligible[idx])
	}
	return types.CheckRowSpillFloor(fields, unspillable)
}

func (table *Table) AddField(newField types.FieldInfo) error {
	if table.db != nil && table.db.HasActiveTransaction() {
		return errDDLInTransaction("AddField")
	}
	defer table.db.lockOp()()
	// 列级参数必须先自洽: 宽度码/scale 越界会让该列宽度算成 0, 而 0 宽度会少算字节,
	// 让其后每一列都从错位偏移解码 (见 types.FieldSize)。
	if err := validateFieldsForSchema([]types.FieldInfo{newField}); err != nil {
		return err
	}

	// 列数闸门与建表同一条 (见 checkFieldCount): master registry 只增不减
	// (DropField 是软删除, 列仍留在 registry 里), 所以按"加完之后"的列数判断。
	if err := checkFieldCount(len(table.Field) + 1); err != nil {
		return err
	}

	// 版本号闸门必须在【任何改动之前】: 两个分支都会在最后新增一版, 版本耗尽时
	// 若已经把 in-memory 状态改了一半再返回错误, 表就停在"改了一半"的不一致状态。
	nextVer, err := table.nextSchemaVersion()
	if err != nil {
		return err
	}

	existing := table.findFieldByName(newField.FieldName)
	if existing != nil {
		// 字段存在且活跃 → 报错
		if !existing.IsDropped() {
			return ErrFieldExists
		}
		// 复活必须沿用原类型与列级参数。
		//
		// master registry 的类型与列级参数是【所有历史版本共用】的行偏移依据
		// (见 getVersionedOffset / rowDataSizeForVersion)。旧实现是整字节覆盖
		// (existing.FieldType = newField.FieldType), 看着像"恢复", 实际是"改掉":
		// varchar(50) 的列复活成 varchar(500) 之后, 已经落盘的旧行仍按 224 字节排布,
		// 却会被按 2048 字节解析 —— 该列之后的每一个字段都静默错位。
		// 想改宽度只能换列名新建, 或者重建表 (DropField 本就只标记删除、保留原类型)。
		if existing.FieldType != newField.FieldType || existing.Param() != newField.Param() {
			return fmt.Errorf("%w: 列 %q 已被删除, 复活时不允许改变类型/宽度 "+
				"(原 %s/参数 %d, 新 %s/参数 %d); 请改用新列名, 或重建该表",
				ErrFieldExists, newField.FieldName,
				types.TypeNameOf(existing.FieldType), existing.Param(),
				types.TypeNameOf(newField.FieldType), newField.Param())
		}
		// 找到该字段在 master registry 中的索引 (复活后的 masterIdx): 行宽预演要靠它
		// 判断这一列是否已被某个活跃索引引用 —— 已建索引的列不能再参与值级溢出。
		revivedIdx := -1
		for i := range table.Field {
			if &table.Field[i] == existing {
				revivedIdx = i
				break
			}
		}
		// 复活同样会让行变宽, 先做行宽预演
		if err := table.checkProspectiveRowSize(newField, revivedIdx); err != nil {
			return err
		}
		// 只恢复"存活 + 可空/唯一"这些标志位, 低 5 位的列级参数保持原值
		existing.Flag = (existing.Flag & types.FlagScaleMask) | (newField.Flag &^ types.FlagScaleMask)
		table.info.ActiveFieldCount++

		// 构建新版本: 旧版本字段 + 复活的字段索引
		sv := table.getVersion(table.activeVersion)
		newIndices := make([]byte, 0, len(sv.FieldIndices)+1)
		if sv != nil {
			newIndices = append(newIndices, sv.FieldIndices...)
		}
		newIndices = append(newIndices, byte(revivedIdx))

		table.activeVersion = nextVer
		table.versions = append(table.versions, SchemaVersion{
			Version:      nextVer,
			FieldIndices: newIndices,
		})

		if err := table.writeSchema(); err != nil {
			// 回滚内存态: 写盘失败不能让内存 schema 与磁盘分叉 —— 后续写入会按
			// 新 schemaVer 落盘, 重开后该版本不可解析 (整行静默丢失)。
			existing.Flag |= types.FlagDeleted
			table.info.ActiveFieldCount--
			table.activeVersion = sv.Version
			table.versions = table.versions[:len(table.versions)-1]
			return err
		}
		table.syncRowSize()
		table.saveTableInfo()
		return nil
	}

	// 全新字段: 同样先做行宽预演 (加列只会让行更宽)
	if err := table.checkProspectiveRowSize(newField, -1); err != nil {
		return err
	}

	// 追加到 master registry
	table.Field = append(table.Field, newField)
	table.info.FieldCount++
	table.info.ActiveFieldCount++

	// 构建新版本: 旧版本字段 + 新字段索引
	sv := table.getVersion(table.activeVersion)
	newIndices := make([]byte, 0, len(sv.FieldIndices)+1)
	if sv != nil {
		newIndices = append(newIndices, sv.FieldIndices...)
	}
	newIndices = append(newIndices, byte(len(table.Field)-1))

	table.activeVersion = nextVer
	table.versions = append(table.versions, SchemaVersion{
		Version:      nextVer,
		FieldIndices: newIndices,
	})

	if err := table.writeSchema(); err != nil {
		// 回滚内存态 (理由同复活分支): Field/registry/版本全部撤回到改动前,
		// 否则内存 schema 与磁盘分叉, 后续写入按新 schemaVer 落盘的行重开后不可解析。
		table.Field = table.Field[:len(table.Field)-1]
		table.info.FieldCount--
		table.info.ActiveFieldCount--
		table.activeVersion = sv.Version
		table.versions = table.versions[:len(table.versions)-1]
		return err
	}
	table.syncRowSize()
	table.saveTableInfo()
	if table.db != nil {
		if err := table.db.ddlBoundary("AddField"); err != nil {
			return err
		}
	}
	return nil
}

// tableInfoBufPool 复用 saveTableInfo 的整页读缓冲。
// saveTableInfo 在【每次行写入】后都会被调用来同步 TotalRows/RowSize 等元信息,
// 旧实现每行 make 一个 16KB —— 灌数时它一个函数就贡献 ~5% 的总分配量。
// 缓冲只在函数内使用 (ReadPage 拷入 / SavePage 拷出), 不会被任何路径持有。
var tableInfoBufPool = sync.Pool{New: func() any { b := make([]byte, PageSize); return &b }}

func (table *Table) saveTableInfo() {
	bp := tableInfoBufPool.Get().(*[]byte)
	buf := *bp
	defer tableInfoBufPool.Put(bp)
	if err := table.handle.ReadPage(table.tablePageID, buf); err != nil {
		return
	}
	info := (*TableInfo)(unsafe.Pointer(&buf[table.tableStartOffset]))
	info.ExtPage = table.info.ExtPage
	info.ExtOff = table.info.ExtOff
	info.FieldCount = table.info.FieldCount
	info.ActiveFieldCount = table.info.ActiveFieldCount
	info.TotalRows = table.info.TotalRows
	// 行宽随 schema 演进变化, 必须落盘同步 (否则删除路径按旧 RowSize 分配缓冲会越界)
	info.RowSize = table.info.RowSize
	// 动态字段堆页链 (分配新堆页时才会变化)
	info.FirstHeap = table.info.FirstHeap
	info.LastHeap = table.info.LastHeap
	info.HeapPageCount = table.info.HeapPageCount
	table.handle.SavePage(table.tablePageID, buf)
}

// =============================================================================
// 删除字段（软删除 + 新版本）
// =============================================================================
func (table *Table) DropField(fieldName string) error {
	if table.db != nil && table.db.HasActiveTransaction() {
		return errDDLInTransaction("DropField")
	}
	defer table.db.lockOp()()
	// 在 master registry 中查找
	masterIdx := -1
	for i, f := range table.Field {
		if f.IsDropped() {
			continue // 已删除
		}
		if string(f.FieldName) == fieldName {
			masterIdx = i
			break
		}
	}
	if masterIdx < 0 {
		return ErrFieldNotFound
	}

	// 守卫: 行存储索引 (主键/时序) 的分量列不允许删除 —— 删了之后
	// extractIndexKey 提不出键, 新行不进任何索引 (插入"成功"但行静默丢失,
	// Count 虚高)。ORM 层的 syncSchema 本来就不动主键列, core 层补上同一约束。
	if im := table.indexMeta; im != nil {
		if rm, _, ok := im.rowStoreMeta(); ok && rm != nil {
			for _, id := range rm.Fields() {
				if int(id) == masterIdx {
					return fmt.Errorf("%w: 列 %q 是行存储索引 %q 的分量, 不能删除 (需重建表)",
						ErrFieldNotFound, fieldName, rm.Name())
				}
			}
		}
	}

	// 版本号闸门必须在改动之前 (理由同 AddField): 删列也要新增一版 schema。
	nextVer, err := table.nextSchemaVersion()
	if err != nil {
		return err
	}

	// 标记删除 (FieldType 本身不动: 它是"这一列是什么", 与存活状态无关)
	table.Field[masterIdx].Flag |= types.FlagDeleted
	table.info.ActiveFieldCount--

	// 构建新版本: 跳过被删字段
	sv := table.getVersion(table.activeVersion)
	newIndices := make([]byte, 0, len(sv.FieldIndices))
	if sv != nil {
		for _, idx := range sv.FieldIndices {
			if int(idx) != masterIdx {
				newIndices = append(newIndices, idx)
			}
		}
	}

	table.activeVersion = nextVer
	table.versions = append(table.versions, SchemaVersion{
		Version:      nextVer,
		FieldIndices: newIndices,
	})

	if err := table.writeSchema(); err != nil {
		// 回滚内存态 (理由同 AddField): 撤销删除标记与版本推进。
		table.Field[masterIdx].Flag &^= types.FlagDeleted
		table.info.ActiveFieldCount++
		table.activeVersion = sv.Version
		table.versions = table.versions[:len(table.versions)-1]
		return err
	}
	table.syncRowSize()
	table.saveTableInfo()
	if table.db != nil {
		if err := table.db.ddlBoundary("DropField"); err != nil {
			return err
		}
	}
	return nil
}

// =============================================================================
// 重命名字段 (只改 FieldInfo.FieldName 字节 + writeSchema, 不改任何列偏移/索引)
// =============================================================================
func (table *Table) RenameColumn(oldName, newName string) error {
	if table.db != nil && table.db.HasActiveTransaction() {
		return errDDLInTransaction("RenameColumn")
	}
	if len(newName) == 0 || len(newName) > FieldNameSize {
		return fmt.Errorf("列名长度 %d 非法 (1..%d)", len(newName), FieldNameSize)
	}
	defer table.db.lockOp()()

	// 找目标列 (只能是活跃列, 已删列不能改名)
	var target *types.FieldInfo
	for i := range table.Field {
		f := &table.Field[i]
		if f.IsDropped() {
			continue
		}
		if util.BytesToString(f.FieldName) == oldName {
			target = f
			break
		}
	}
	if target == nil {
		return fmt.Errorf("列 %q 不存在", oldName)
	}

	// 新名不能和别的活跃列重名
	for i := range table.Field {
		f := &table.Field[i]
		if f.IsDropped() {
			continue
		}
		if f != target && util.BytesToString(f.FieldName) == newName {
			return fmt.Errorf("列 %q 已存在", newName)
		}
	}

	// 改内存态 FieldName (新建切片, 不原地改引用外部可能持有)
	target.FieldName = make([]byte, len(newName))
	copy(target.FieldName, newName)

	// schema 落盘 (writeSchema 会重新 serializeSchema 把新名写进去)
	if err := table.writeSchema(); err != nil {
		return err
	}
	if table.db != nil {
		if err := table.db.ddlBoundary("RenameColumn"); err != nil {
			return err
		}
	}
	return nil
}