package core

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"sort"
	"strings"
	"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

	// 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 切片 (身份比对用)
}

// CreateIndex 建二级索引。
//
// 这里先校验分量列能不能做键, 再交给 IndexMetaPage 建 —— 拆成两层是因为
// IndexMetaPage 只拿得到索引自己的元信息, 看不到表的字段类型:
//   - 分量列不存在 → ErrFieldNotFound;
//   - 分量是动态列 → ErrIndexDynamicField。动态列在行内只存 6B 堆引用, 值 != 引用,
//     按值查找永远不可能命中; 建出来既查不到也不会被查询侧选中 (buildKeyLayout 会跳过),
//     只会白占空间、白写。拦在建的这一步, 而不是等用到才发现;
//   - 分量是 blob 列 → ErrIndexBlobField。blob 列行内只存 8B BlobRef (页号 + 长度),
//     拿它当索引键等于按页号排序/查找 —— 索引里存的不是内容, 命中与否全看内容落在哪一页;
//   - 复合索引 (IndexTypeCompose) 的分量列宽度不再强制 8 字节 ——
//     底层已升级为 B+ 树, 按分量真实宽度拼定长键, 溢出闸门 (checkIndexSpillFloor)
//     负责兜住"一页两行"的行宽上限。
//
// unique 为可选 (只取第一个值), 见 IndexMetaPage.CreateIndex: 传 true 建的是唯一索引,
// 同一索引键只允许一行; 任一分量为 NULL 的行不进该索引, 因此多个 NULL 天然共存。
// 唯一约束在写入路径 (InsertRow / UpdateByKey) 上校验, 违反时返回 ErrDuplicatedKey。
func (table *Table) CreateIndex(idxName string, typ page.IndexType, fields []byte, unique ...bool) (*IndexMeta, error) {
	for i, fi := range fields {
		if int(fi) >= len(table.Field) {
			return nil, fmt.Errorf("%w: 索引 %q 的第 %d 个分量", ErrFieldNotFound, idxName, i+1)
		}
		ft := table.Field[fi].FieldType
		if types.IsDynamic(ft) {
			return nil, fmt.Errorf("%w (索引 %q 的第 %d 个分量)", ErrIndexDynamicField, idxName, i+1)
		}
		// v10: blob 列行内只有 8B BlobRef (页号 + 长度), 拿它当索引键等于按页号排序/查找 ——
		// 索引里存的不是内容, 命中与否全看内容落在哪一页. 与动态列同等拦下.
		if types.IsBlob(ft) {
			return nil, fmt.Errorf("%w (索引 %q 的第 %d 个分量)", ErrIndexBlobField, idxName, i+1)
		}
	}
	// 溢出闸门: 分量列一旦建索引就不能再值级溢出, 行内下界随之抬升。
	// 下界超过"一页两行"的配额时必须在这里拒绝 —— 否则这张表从建索引成功起就再也压不
	// 进两行 (单条插入仍会"成功"地落在 8159..16332 之间, 页内顺序/有序插入/页内重建
	// 一起退化, 而且从现象上完全看不出是索引造成的)。
	if err := table.checkIndexSpillFloor(fields); err != nil {
		return nil, fmt.Errorf("%w: 索引 %q: %v", ErrIndexRowTooWide, idxName, err)
	}
	m, err := table.indexMeta.CreateIndex(idxName, typ, fields, unique...)
	if err != nil {
		return nil, err
	}
	// 回填存量行。
	// 旧实现只建元数据, 索引里一条记录都没有: 建索引之前就存在的行永远查不到,
	// 而引擎又没有公开的"重建全部二级索引"接口 —— 数据在、索引空, 是最难排查的
	// 一类不一致。建索引这一步必须自己把存量行灌进去。
	if err := table.backfillIndex(m); err != nil {
		// 回填失败不能留下"半个索引": 元数据还在但内容不全, 比没有索引更危险
		// (查询会选中它, 于是稳定地查不到行)。摘掉元数据, 让错误显式冒出来。
		// 走 table.DropIndex 而不是 indexMeta.DropIndex: 后者只软删元数据, 会把
		// 回填到一半的树页留成孤儿页。
		_ = table.DropIndex(idxName)
		return nil, err
	}
	return m, nil
}

// backfillIndex 把表里已有的行灌进一个刚建好的二级索引。
//
// 与写入侧共用 extractIndexKey + 处理器 Insert, 因此:
//   - 键编码与后续单条插入完全一致, 不会出现"回填进去了却查不到"的错位;
//   - 老行 (该列在其自身的 schema 版本里还不存在, 例如 AddField 之前插入的行)
//     自动跳过, 与"单条插入时缺失列不落索引"的语义保持一致;
//   - 动态列在 CreateIndex 已被拒绝, 这里不必再判。
//
// 行存储索引 (主键/时序) 不在此处理: 它们的键就是 pkID, 行数据本就由写入路径维护。
func (table *Table) backfillIndex(m *IndexMeta) error {
	if m == nil || indexStoresRows(m) {
		return nil
	}
	primMeta, primHandler, ok := table.indexMeta.rowStoreMeta()
	if !ok || primMeta == nil || primHandler == nil {
		return nil // 表里没有承载行的索引, 也就没有存量行
	}

	// 收集 → 排序 → 批量建树。
	// 不能逐条 handler.Insert: 那是 n 次 O(log n) 的树写放大, 千万行级别慢到
	// 不可接受; 索引重建本来就有 O(n) 的批量路径, 这里直接复用。
	var entries [][]byte
	for _, dpID := range primHandler.GetDataPageIDs(table.handle, primMeta.EntryPage) {
		for _, storeRow := range primHandler.ReadDataPageRows(table.handle, dpID) {
			if len(storeRow) < 8 {
				continue
			}
			secKey := extractIndexKey(storeRow[8:], m, table.Field, table.activeVersion, table.versions)
			if secKey == nil {
				continue // 该行没有这个键 (列不在它的 schema 版本里)
			}
			entry := make([]byte, len(secKey)+8)
			copy(entry, secKey)
			// 条目尾部是 pkID (8B BE), 与 rebuildIndexes 的二级条目同一布局
			binary.BigEndian.PutUint64(entry[len(secKey):], binary.BigEndian.Uint64(storeRow[:8]))
			entries = append(entries, entry)
		}
	}
	if len(entries) == 0 {
		// 没有任何存量行进这个索引 → 保持空树 (EntryPage==0), 后续插入照常开根
		return nil
	}
	// 按复合键 (secKey, pkID) 升序排序: 二级树的复合分隔键与叶内复合二分要求
	// 全树按复合键全局有序, 同 key 的条目必须再按 pkID 排
	sort.SliceStable(entries, func(i, j int) bool {
		a, b := entries[i], entries[j]
		if c := compareBytes(a[:len(a)-8], b[:len(b)-8]); c != 0 {
			return c < 0
		}
		return binary.BigEndian.Uint64(a[len(a)-8:]) < binary.BigEndian.Uint64(b[len(b)-8:])
	})
	// 唯一索引进来了要顺手验一遍存量数据: 库里本来就有重复值时, 这个约束是【建不出来】的。
	// 硬建下去只会得到一个"违反自身约束"的索引 (同一键挂着两行), 之后点查可能返回其中
	// 任意一行, 而且从索引上完全看不出它已经坏了。宁可在建索引这一步报错, 由调用方决定
	// 是清理数据还是改用非唯一索引。排序后重复键必然相邻, 一次线性扫描即可。
	if m.Unique() {
		for i := 1; i < len(entries); i++ {
			if compareBytes(entries[i-1][:len(entries[i-1])-8], entries[i][:len(entries[i])-8]) == 0 {
				return fmt.Errorf("%w: 索引 %q 的存量数据中已存在重复键 (pkID %d 与 %d), 无法建成唯一索引",
					ErrDuplicatedKey, util.BytesToString(m.IndexName[:]),
					binary.BigEndian.Uint64(entries[i-1][len(entries[i-1])-8:]),
					binary.BigEndian.Uint64(entries[i][len(entries[i])-8:]))
			}
		}
	}
	if err := buildSecondaryFromEntries(table.handle, m, entries); err != nil {
		return err
	}
	table.indexMeta.saveMeta(m)
	table.indexMeta.InvalidateRowCache()
	return nil
}
func (table *Table) DropIndex(idx string) error {
	// DROP INDEX 必须把整棵索引树还给空闲链。
	// 旧实现只把元数据打上软删除标记 (Flag|=0x80) 并从 metas 摘掉: 根指针一丢, 整棵树
	// (内部节点 + 叶子) 就既不可达、又不在空闲链上 —— 变成永久孤儿页, 文件只涨不缩。
	// 实测 20 万行的一棵二级索引约 197 页; bench_10m 里 9.9M 行的 Dept 索引是数万页,
	// 正是那批孤儿页的主因。
	// 注意: indexMeta.DropIndex 会用 append 挪动 metas 元素, 之后 &metas[i] 就失效了,
	// 所以必须在摘除之前把 IndexType / EntryPage 快照出来。
	var itype page.IndexType
	var rootPID uint32
	if m := table.indexMeta.findIndexMetaByName(idx); m != nil {
		itype, rootPID = m.IndexType, m.EntryPage
	}
	if err := table.indexMeta.DropIndex(idx); err != nil {
		return err
	}
	// 旧结构此刻已不可达 (元数据已摘), 归还它的全部页。
	releaseIndexPages(table, itype, rootPID)
	return nil
}

// BuildIndexKey 按索引定义, 把各分量值编码成索引键 (与写入侧 extractIndexKey 同一编码)。
//
// 复合索引的键是各分量定长拼接, 外部无法凭空构造 —— 因此点查/删除/更新复合索引
// (SearchByIndex / DeleteRow / UpdateRow) 都要先经这里把值编成键。
// values 的顺序与建索引时给的 fieldIDs 顺序一致, 个数必须相等。
func (table *Table) BuildIndexKey(idxName string, values ...interface{}) ([]byte, error) {
	m := table.indexMeta.findIndexMetaByName(idxName)
	if m == nil {
		return nil, ErrIndexNotFound
	}
	n := int(m.FieldCnt)
	if n != len(values) {
		return nil, fmt.Errorf("索引 %q 需要 %d 个分量值, 实际给了 %d 个", idxName, n, len(values))
	}
	key := make([]byte, 0, n*8)
	for k := 0; k < n; k++ {
		fi := int(m.FieldIDs[k])
		if fi >= len(table.Field) {
			return nil, ErrFieldNotFound
		}
		ft := table.Field[fi].FieldType
		if table.Field[fi].IsDropped() {
			return nil, fmt.Errorf("索引 %q 的第 %d 个分量列已被删除", idxName, k+1)
		}
		if types.IsDynamic(ft) {
			return nil, fmt.Errorf("索引 %q 的第 %d 个分量是动态列, 不能作为索引键", idxName, k+1)
		}
		if types.IsBlob(ft) {
			return nil, fmt.Errorf("索引 %q 的第 %d 个分量是 blob 列, 不能作为索引键", idxName, k+1)
		}
		enc, err := encodeIndexKeySafeWithParam(values[k], ft, table.fieldParam(fi))
		if err != nil {
			return nil, fmt.Errorf("索引 %q 的第 %d 个分量: %w", idxName, k+1, err)
		}
		if want := int(table.Field[fi].Width()); len(enc) != want {
			return nil, fmt.Errorf("索引 %q 的第 %d 个分量: 编码 %d 字节, 列宽 %d 字节",
				idxName, k+1, len(enc), want)
		}
		key = append(key, enc...)
	}
	return key, nil
}

// 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()
}

// 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

// 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
}

// getVersion 获取指定版本的 schema 定义
func (table *Table) getVersion(ver byte) *SchemaVersion {
	for i := range table.versions {
		if table.versions[i].Version == ver {
			return &table.versions[i]
		}
	}
	return nil
}

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

// 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
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return 0, 0, false
	}
	n := len(sv.FieldIndices)
	col0Width := table.svCol0Width(sv)
	ends, ok := types.RowParseOffsets(rowData, n, col0Width)
	if !ok {
		return 0, 0, false
	}
	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) {
	sv := table.getVersion(ver)
	if sv == nil {
		return nil, false
	}
	n := len(sv.FieldIndices)
	ends, ok := types.RowParseOffsets(rowData, n, table.svCol0Width(sv))
	if !ok {
		return nil, false
	}
	return types.RowNullBitmap(rowData, 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
}

// maxCompareInlineSize 字段比较时使用的栈缓冲上界 == 最宽的行内定长列。
//
// 取自 types.MaxInlineColumnWidth 而不是写死 256: 字符串列可以用宽度码声明到 2048 字节,
// 缓冲若还停在 256, 宽字符串列的每次比较都会走堆分配。
const maxCompareInlineSize = types.MaxInlineColumnWidth

// ======================== 行内字段比较 ========================
//
// 三个比较函数都经由 compareFieldBytes 实现, 也就是"把条件值编码成与行内同源的字节,
// 再按字节比较"。这样做的理由:
//   - 行内字节与索引键字节本就同源 (都出自 types.EncodeValue), 于是行级过滤与索引
//     范围扫描必然给出同一个结论, 不会出现"索引选出 5 行、过滤后剩 0 行"的自相矛盾;
//   - 新增类型不必再回来改这三个函数 (以前每个类型要在这里各写 3 遍, 漏改一处就是静默漏行)。
//
// 与旧的"解码成 Go 值再比"相比, 唯一的行为差异是 NaN: 按字节比时 NaN == NaN 成立,
// 而 float 的 == 不成立。索引键本来就是按字节比的, 统一到字节序反而消除了
// "全表扫和索引给出不同答案"的隐患。

// compareFieldBytes 比较行内字段与给定值, 返回 (cmp, ok); ok=false 表示无法比较。
//
// ct 必须是【列的实际存储类型 + 列级参数】(调用方经 compareColTypeFor 取, 与索引路径
// 编键同源), 否则编码出来的字节与行内不同源, 比较结果没有意义。
// ok=false 的来源: 字段越界、动态列解引用失败、值无法按该类型编码。
// 前两种按"不匹配"处理是正确的; 第三种是调用方 bug —— 索引路径会显式报错
// (见 encodeIndexKeySafe), 行级过滤按不匹配处理, 至少不会污染数据。
// rowFieldForCompare 定位列并返回【可直接比较】的字节。
//
// 【为什么必须补齐到档位宽度】行内在 v9 之后存的是实际长度, 而条件值与索引键走的仍是
// 定长档位编码 (EncodeValue 的 buf 长度恒为 FieldSize)。两侧若不补齐就不是同一份编码:
// B+ 树按定长拼键比较、行级过滤按变长比较, 同一条件会得出不同结论 (表现为"索引选出
// 5 行, 过滤后剩 0 行")。补齐用零字节, 与 v8 的定长存储逐字节相同, 因此比较结果与 v8
// 完全一致; 索引键那条路径也按同一规则补齐, 两条路径始终同源。
//
// 动态列【不补齐】: 它的内容是 6B 堆引用, 补齐毫无意义, 交由 compareFieldBytes 解引用。
//
// 【值级溢出同理】字符串列一旦在【这一行】被溢出, 行内就是 6B 堆引用而不是内容。
// 是否溢出必须看行内 tag (RowFieldRefTagged), 不能看列类型 —— 值级溢出是逐行的,
// 同一列可能"这行内联、那行溢出"。漏了这步就会把页号/偏移当成字符串内容去比,
// 症状是"溢出发生后, 对未建索引的宽字符串列过滤开始静默漏行"。
func (table *Table) rowFieldForCompare(row []byte, masterFieldIdx int, ct colType) ([]byte, bool) {
	start, end, ok := table.getVersionedRange(row, masterFieldIdx)
	if !ok {
		return nil, false
	}
	seg := row[start:end]
	if table.columnRefTagged(row, masterFieldIdx) {
		if types.IsDynamic(ct.typ) {
			return seg, true // 动态列: 保持既有契约, 交 compareFieldBytes 解引用
		}
		// 被值级溢出的字符串列: 解引用拿回内容, 再按档位补齐 → 与索引键同源。
		data, ok := table.derefHeapSeg(seg)
		if !ok {
			data = nil // pageID == 0 或读取失败 → 空内容
		}
		return types.PadToWidth(data, ct.width()), true
	}
	if types.IsDynamic(ct.typ) {
		return seg, true
	}
	return types.PadToWidth(seg, ct.width()), true
}

// columnRefTagged 判断【这一行】的 masterFieldIdx 列在行内是不是 6B 堆引用:
// 动态列恒是; 字符串列看 RowFieldRefTagged 的 tag 位 —— 该实例被值级溢出了。
func (table *Table) columnRefTagged(rowData []byte, masterFieldIdx int) bool {
	if len(rowData) < 1 {
		return false
	}
	sv := table.getVersion(rowData[0])
	if sv == nil {
		return false
	}
	col0Width := table.svCol0Width(sv)
	offTableBase := types.RowOffTableBase(col0Width)
	for i, idx := range sv.FieldIndices {
		if int(idx) != masterFieldIdx || int(idx) >= len(table.Field) {
			continue
		}
		if types.IsDynamic(table.Field[idx].FieldType) {
			return true
		}
		if i == 0 {
			return false // 列 0 恒内联 (偏移表基址依赖它, 不参与溢出)
		}
		if offTableBase+2*i >= len(rowData) {
			return false
		}
		return types.RowFieldRefTagged(rowData, i, col0Width)
	}
	return false
}

// derefHeapSeg 把行内 6B 堆引用 [PageID 4B][Offset 2B] 解引用成内容。
// pageID == 0 (空引用) 或读取失败返回 ok=false, 调用方按"空内容"处理。
func (table *Table) derefHeapSeg(seg []byte) ([]byte, bool) {
	if len(seg) < types.DynamicRefSize {
		return nil, false
	}
	pageID := binary.BigEndian.Uint32(seg[:4])
	if pageID == 0 {
		return nil, false
	}
	data, err := table.heapRead(pageID, binary.BigEndian.Uint16(seg[4:6]))
	if err != nil {
		return nil, false
	}
	return data, true
}

func (table *Table) compareFieldBytes(seg []byte, ct colType, value interface{}) (int, bool) {
	if types.IsDynamic(ct.typ) {
		if len(seg) < types.DynamicRefSize {
			return 0, false
		}
		pageID := binary.BigEndian.Uint32(seg[:4])
		offInPage := binary.BigEndian.Uint16(seg[4:6])
		stored, err := table.heapRead(pageID, offInPage)
		if err != nil {
			return 0, false
		}
		switch v := value.(type) {
		case []byte:
			return bytes.Compare(stored, v), true
		case string:
			return bytes.Compare(stored, []byte(v)), true
		}
		return 0, false
	}

	size := int(ct.width())
	if size == 0 || len(seg) != size {
		// 长度必须正好等于档位宽度: 调用方 (rowFieldForCompare) 已按档位补齐, 不齐说明列
		// 参数非法或行已损坏 —— 这时按"不匹配"处理, 绝不拿错位字节去比。
		return 0, false
	}
	stored := seg

	// decimal 的 scale 存在行内, 条件值的 scale 可能与行内不同 (1.5 与 1.50 的系数
	// 分别是 15 和 150), 系数直接 memcmp 会得出错误结论 → 走数值比较。
	if ct.typ == types.FieldType_Decimal {
		return table.compareDecimalBytes(stored, value)
	}

	// 编码到栈缓冲再比较 (最宽列 2048 字节; 更宽的类型退化为堆分配, 不 panic)
	var stack [maxCompareInlineSize]byte
	var buf []byte
	if size <= len(stack) {
		buf = stack[:size]
	} else {
		buf = make([]byte, size)
	}
	// param 必须一起传: 字符串列的宽度码决定 EncodeValue 期望的 buf 长度,
	// 传 0 会让"按类型默认 256 编"与"按声明 2048 存"对不上 → 该列静默全不匹配。
	if err := types.EncodeValue(ct.typ, ct.param, value, buf); err != nil {
		return 0, false
	}
	return bytes.Compare(stored, buf), true
}

// compareDecimalBytes 比较行内 decimal 字节与给定值。
//
// 同 scale (同一列的常态) → 直接字节比较, 与索引键行为一致;
// 不同 scale (条件值写成 1.5 而列是 DECIMAL(18,2)) → 退回数值比较, 保证语义正确。
func (table *Table) compareDecimalBytes(stored []byte, value interface{}) (int, bool) {
	d, err := types.DecimalFromValue(value)
	if err != nil {
		return 0, false
	}
	if d.Scale() == types.DecimalScaleOfBytes(stored) {
		var buf [types.DecimalSize]byte
		if err := types.EncodeValue(types.FieldType_Decimal, byte(d.Scale()), d, buf[:]); err != nil {
			return 0, false
		}
		return bytes.Compare(stored, buf[:]), true
	}
	return types.DecimalFromBytes(stored).Cmp(d), true
}

// colType 一列"编码/比较所需的全部类型信息": 类型值 + 列级参数 (scale / 宽度码)。
//
// 打包成一个值传递, 而不是把 typ 与 param 拆成两个形参: 两者必须一起走 —— 只传类型
// 会出现"按 256 字节编、按 2048 字节存"这类不一致, 而它表现为整列静默不匹配或整行
// 错位, 都不是编译错误。打包后"忘了带参数"这件事在签名上就写不出来。
type colType struct {
	typ   types.FieldType
	param byte
}

// width 该列的行内宽度 (字节); 0 表示参数非法。
func (c colType) width() uint16 { return types.FieldSize(c.typ, c.param) }

// compareColTypeFor 取出行级过滤比较要用的列信息 —— 以【列的实际存储类型与参数】为准。
//
// 为什么不能沿用条件里声明的 Type:
//   - 索引路径编键用的是列定义 (见 query.go buildKeyLayout / planIndex:
//     q.table.Field[..].FieldType 与 fieldParam), 行级过滤若用声明类型, 两条路径就会
//     分叉: 同一列带索引时"声明错类型"仍能命中 (索引键按列定义编), 不带索引时却按
//     声明类型编出另一种字节 → 静默返回 0 行, 看起来像数据不存在。
//   - 声明类型因此降级为"值怎么从 interface{} 取出来"的提示: EncodeValue 按 Go
//     值的 kind 分派, 本身并不依赖它。
//
// 列级参数 (字符串宽度码) 同理必须一起取: 宽度决定了"行内那几字节从哪切、切多长",
// 与是否走索引无关。
//
// 删除标记已迁到 FieldInfo.Flag, FieldType 本身不再混入状态位, 因此这里直接返回
// 即可 —— 旧实现在此清 0x80, 留着标记位会让 IsDynamic 查不到注册项: 动态列会被
// 当成定长字段, 拿 6 字节堆引用去和值做字节比较。
//
// 拿不到字段时返回零值 (未注册类型 + 参数 0), compareFieldBytes 判 width()==0 后按
// 不匹配处理 —— 调用方已先判过 fieldIdx < 0, 这里只是兜底。
func (table *Table) compareColTypeFor(fieldIdx int) colType {
	if fieldIdx < 0 || fieldIdx >= len(table.Field) {
		return colType{}
	}
	return colType{table.Field[fieldIdx].FieldType, table.Field[fieldIdx].Param()}
}

// compareValue 比较字段值是否相等 (seg 由 rowFieldForCompare 定位并补齐)
func (table *Table) compareValue(seg []byte, ct colType, value interface{}) bool {
	c, ok := table.compareFieldBytes(seg, ct, value)
	return ok && c == 0
}

// compareGreater 比较字段值是否大于给定值
func (table *Table) compareGreater(seg []byte, ct colType, value interface{}) bool {
	c, ok := table.compareFieldBytes(seg, ct, value)
	return ok && c > 0
}

// compareLess 比较字段值是否小于给定值
func (table *Table) compareLess(seg []byte, ct colType, value interface{}) bool {
	c, ok := table.compareFieldBytes(seg, ct, value)
	return ok && c < 0
}

// compareGreaterOrEqual 比较字段值是否大于等于给定值
func (table *Table) compareGreaterOrEqual(seg []byte, ct colType, value interface{}) bool {
	return table.compareGreater(seg, ct, value) || table.compareValue(seg, ct, value)
}

// compareLessOrEqual 比较字段值是否小于等于给定值
func (table *Table) compareLessOrEqual(seg []byte, ct colType, value interface{}) bool {
	return table.compareLess(seg, ct, value) || table.compareValue(seg, ct, value)
}

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
}

// 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 {
	// 列级参数必须先自洽: 宽度码/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()
	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 {
	// 在 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()
	return nil
}

// ======================== 动态字段堆页存储 ========================
//
// 动态字段 (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
	}
	sv := table.getVersion(oldRowData[0])
	if sv == nil || len(sv.FieldIndices) == 0 {
		return nil
	}
	n := len(sv.FieldIndices)
	col0Width := table.svCol0Width(sv)
	ends, ok := types.RowParseOffsets(oldRowData, n, col0Width)
	if !ok {
		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
}

// ======================== 行数据序列化/反序列化 ========================

// lookupFieldValue 在 FieldValue 切片中按字段名取值
func lookupFieldValue(values []types.FieldValue, name []byte) (interface{}, bool) {
	for _, fv := range values {
		// bytes.Equal 直接比较字节, 避免 string(x)==string(y) 触发两次 string 分配
		if bytes.Equal(fv.FieldName, name) {
			return fv.Value, true
		}
	}
	return nil, false
}

// encodeFieldValue 把一列的值编码成行内字节, 不写缓冲: 字符串返回【实际长度】,
// 定长类型返回档位宽度的字节。布局阶段统一切片拷贝 (见 serializeRowInternal)。
//
// 形参是 FieldInfo 而不是 FieldType: 宽度与列级参数都必须取自【同一个列定义】——
// 拆成两个参数传 (类型来自这里、宽度码来自那里) 就会出现"按 256 字节编、按 2048 字节存"
// 这类不一致, 而它表现为整行错位而不是报错。
func (table *Table) encodeFieldValue(masterIdx int, f types.FieldInfo, value interface{}) ([]byte, error) {
	ft := f.FieldType
	// 字符串家族: 按【实际长度】编码 —— 这是 v9 变长省空间的主体。超宽仍在编码时直接
	// 报错 (声明宽度是硬闸门), 但不再补零到档位。
	if types.IsVarWidth(ft) {
		enc, err := types.EncodeValueVar(ft, f.Param(), value)
		if err != nil {
			return nil, fmt.Errorf("字段 %q 写入失败: %w", table.Field[masterIdx].FieldName, err)
		}
		return enc, nil
	}
	size := f.Width()
	if size == 0 {
		return nil, fmt.Errorf("%w: 字段 %q (%s) 的列级参数 %d 非法, 宽度不可计算",
			types.ErrValueRange, f.FieldName, types.TypeNameOf(ft), f.Param())
	}
	buf := make([]byte, size)
	// 编码统一走 types.EncodeValue —— 与索引键字节同源 (见 types/codec.go)。
	// param 取列级类型参数: decimal 必须在这里校验"值的 scale == 列声明的 scale",
	// 否则同一列出现两种 scale 会让"字节序 == 数值序"失效, 索引范围扫描静默漏行;
	// 字符串列则由它决定列宽闸门 (超宽在 EncodeValue 里直接报错, 不再截断)。
	if err := types.EncodeValue(ft, f.Param(), value, buf); err != nil {
		return nil, fmt.Errorf("字段 %q 写入失败: %w", table.Field[masterIdx].FieldName, err)
	}
	return buf, nil
}

// dynamicFieldData 解析动态列的值 ([]byte; nil 表示空值), 并卡单块尺寸闸门。
//
// 动态列的值只认 []byte。旧写法是 `data, _ := value.([]byte)`: 一旦传进来
// 的是别的类型 (string / 数值), 断言失败 → data 为 nil → 整个值被静默写成空。
// 这与"不静默丢内容"直接冲突, 所以这里改为显式报错 (nil 仍表示空值)。
func (table *Table) dynamicFieldData(masterIdx int, f types.FieldInfo, value interface{}) ([]byte, error) {
	if value == nil {
		return nil, nil
	}
	data, ok := value.([]byte)
	if !ok {
		return nil, fmt.Errorf("字段 %q (%s): 动态列只接受 []byte, 收到 %T",
			f.FieldName, types.TypeNameOf(f.FieldType), value)
	}
	// 动态列的单块尺寸闸门: 超限在这里就报出列名与上限, 而不是等 heapAllocBlock
	// 抛出与"值太大"因果关系不明的 ErrHeapPageTooLarge。
	//
	// 只覆盖 Dynamicbyte: 它的行内是 6B 堆引用, 内容整块落进一个堆页, 装不下更大的值。
	// Blob 列不走这里 —— v10 起它的行内是 8B BlobRef, 内容落在可跨页的链上 (上限 ~4GB,
	// 见 blobCommit). 往 Dynamicbyte 写超限值若不在这一层拦, 只会得到不带列名、不带
	// 字节数的"申请空间超过单页最大容量", 用户看不出是哪一列、超了多少。
	if uint32(len(data)) > MaxBlobSize {
		return nil, fmt.Errorf("%w: 字段 %q (%s) 的值 %d 字节超过单块上限 %d 字节 —— "+
			"动态列的内容整块落进一个堆页, 更大的内容引擎目前装不下: 多页分块存储尚未实现, "+
			"请改小这个值, 或改用 blob 列 (链式存储, 可跨页)。",
			types.ErrValueRange, f.FieldName,
			types.TypeNameOf(f.FieldType), len(data), MaxBlobSize)
	}
	return data, nil
}

// blobFieldRef 把 Blob 列的载体值统一成行内 8B 的 types.BlobRef.
//
// Blob 列行内只放引用, 而用户写这一列时手上多半是【内容本身】。这里就是
// types/blob.go 里"把 io.Reader / []byte 写进 FieldType_Blob 列, 引擎自动走链页"
// 的那个落点 —— 调用方不必自己开事务、自己调 BlobCommit。
//
// 接受的值:
//
//	nil             -> 空引用 (NULL 已在上游 Pass 1 拦掉, 这里只兜底非空列的空值)
//	types.BlobRef   -> 原样 (已经是引用: 用户自己 tx.BlobCommit 过, 或读回侧镜像)
//	types.Blob      -> Source 优先 (可流式), 否则 Bytes; 已带 ref 的直接用
//	[]byte / string -> 走链页 (空内容不分配页, 直接给空引用)
//	io.Reader       -> 走链页 (流式, 内存峰值与内容大小无关)
//	其它类型         -> 报错 (与"不静默丢内容"一致)
//
// checkOnly 是唯一性试算行 (serializeRowForCheck): 绝不能落盘, 返回零引用即可 ——
// Blob 列不能当索引键 (见 CreateIndex), 试算行的键字节与它无关。
func (table *Table) blobFieldRef(f types.FieldInfo, value interface{}, checkOnly bool) (BlobRef, error) {
	switch v := value.(type) {
	case BlobRef: // == types.BlobRef (类型别名), 引用原样透传
		return v, nil
	case types.Blob:
		return table.blobRefFromValue(f, v, checkOnly)
	case *types.Blob:
		if v == nil {
			return BlobRef{}, nil
		}
		return table.blobRefFromValue(f, *v, checkOnly)
	}

	if checkOnly {
		return BlobRef{}, nil
	}

	var src io.Reader
	switch v := value.(type) {
	case []byte:
		if len(v) == 0 {
			return BlobRef{}, nil
		}
		src = bytes.NewReader(v)
	case string:
		if v == "" {
			return BlobRef{}, nil
		}
		src = strings.NewReader(v)
	case io.Reader:
		src = v
	default:
		return BlobRef{}, fmt.Errorf("字段 %q (%s): blob 列只接受 []byte / string / io.Reader / types.Blob / types.BlobRef, 收到 %T",
			f.FieldName, types.TypeNameOf(f.FieldType), value)
	}
	ref, err := table.db.blobCommit(src, types.CompressNone)
	if err != nil {
		return BlobRef{}, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
			f.FieldName, types.TypeNameOf(f.FieldType), err)
	}
	return ref, nil
}

// blobRefFromValue 处理 types.Blob 载体: 既有"内容"形态 (Source / Bytes),
// 也有"读回镜像"形态 (FirstPageID / TotalLen 已由引擎填好).
func (table *Table) blobRefFromValue(f types.FieldInfo, b types.Blob, checkOnly bool) (BlobRef, error) {
	hasRef := b.FirstPageID != 0 || b.TotalLen != 0
	hasContent := b.Source != nil || len(b.Bytes) > 0
	if hasRef {
		if hasContent {
			return BlobRef{}, fmt.Errorf("字段 %q: types.Blob 同时带了引用与内容 (Source/Bytes), 无法判定该用哪个", f.FieldName)
		}
		return b.Ref(), nil // 读回镜像: 原样透传
	}
	if !hasContent {
		return BlobRef{}, nil // 空 blob: 不分配链页
	}
	if checkOnly {
		return BlobRef{}, nil
	}
	if b.Source != nil && len(b.Bytes) > 0 {
		return BlobRef{}, fmt.Errorf("字段 %q: types.Blob 的 Source 与 Bytes 都非空, 引擎无法选择", f.FieldName)
	}
	if !b.Compress.IsValid() {
		return BlobRef{}, fmt.Errorf("字段 %q: types.Blob.Compress 非法", f.FieldName)
	}
	var src io.Reader
	if b.Source != nil {
		src = b.Source
	} else {
		src = bytes.NewReader(b.Bytes)
	}
	ref, err := table.db.blobCommit(src, b.Compress)
	if err != nil {
		return BlobRef{}, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
			f.FieldName, types.TypeNameOf(f.FieldType), err)
	}
	return ref, 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()
}

// serializeRow 将 FieldValue 切片序列化为行数据字节 (不含 pkID 前缀)
// 行格式 (v9): [schemaVer 1B][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图]
// 动态字段的行内数据为 6 字节堆引用; oldRowData 用于更新时复用堆块。
//
// 两点关键语义 (否则 schema 演进后更新旧行会静默丢数据):
//   - 版本选择走 pickVersionForUpdate: 未触及新字段时保持旧版本, 偏移表与列序都不变,
//     不会因为"只改了个老字段"就无谓升版; 但 v9 的字符串列按【实际长度】存, 所以
//     "版本不变"不等于"行长不变", 变长仍可能发生 (由索引层的原地→页内重建→换页兜住)。
//   - values 中未提供的字段用旧行原值补齐, 而不是写零
//     (否则 "只改一列" 会把该行其它列全部清空)
func (table *Table) serializeRow(values []types.FieldValue, oldRowData []byte) ([]byte, error) {
	return table.serializeRowInternal(values, oldRowData, false)
}

// serializeRowForCheck 试算一遍行字节: 除动态列外与 serializeRow 逐字节一致。
//
// 动态列在这里【不碰堆】—— 不分配、不复用、不释放, 行内 6B 引用留全零。存在的理由
// 是更新前的唯一性预检必须早于 serializeRow: writeDynamicField 会就地覆写旧行的堆块
// (heapRewrite), 换块时还会 heapUnref 旧块。等 UpdateByKey 报出键冲突再返回错误,
// 旧行的动态列已被改成新值 (极端情况下堆块已释放而旧行仍指着它), 而固定列与索引条目
// 还是旧的 —— 行被改了一半, 且不留任何痕迹。多一次序列化换"要么全改要么全不动"。
//
// 索引分量一定不是动态列 (CreateIndex 已拒绝), NULL 位图也只取决于"值是否为 nil 且
// 列可空", 与堆无关 —— 所以试算行算出的索引键与真身完全一致。
func (table *Table) serializeRowForCheck(values []types.FieldValue, oldRowData []byte) ([]byte, error) {
	return table.serializeRowInternal(values, oldRowData, true)
}

// nullFieldError 非空列被写入 NULL 时的统一报错 (两处循环共用, 文案必须一致)。
func nullFieldError(name string, ft types.FieldType) error {
	return fmt.Errorf("%w: 字段 %q (%s) 未声明可空 (nullable), 不接受 NULL; "+
		"要允许写入 NULL, 请在列声明里加 `nullable` / `null`",
		types.ErrValueRange, name, types.TypeNameOf(ft))
}

func (table *Table) serializeRowInternal(values []types.FieldValue, oldRowData []byte, checkOnly bool) ([]byte, error) {
	ver := table.pickVersionForUpdate(oldRowData, values)

	// 写行闸门 (第三处行宽校验)。
	//
	// 前两处在建列侧 (TablePage.Create / Table.AddField / CreateIndex), 管的是"新表 /
	// 新索引不许超宽"; 这一处管的是"已经存在的老表": 表是在限制生效前建的 (或 schema
	// 被外部改坏) 时, 行字节数会超过一个 DataPage 能装下的量。放任它走下去的后果不是
	// "写失败"这么轻 —— 行长度会在 DataPage 里溢出到相邻区域, 把整页的行解析全部错位
	// (页内行是顺序排布、靠每行 4B 长度前缀自描述的, 一行错位等于整页报废)。
	//
	// 【v9 起判据是"溢出下界"】行按内容变长, 实际长度要等内容全编完才知道, 而那时堆块
	// 可能已经动过 (writeHeapRefField 会覆写/释放旧块), 再报错就留下了半改的行。这里用
	// types.CalcSpillFloor 的下界 —— "把所有可溢出的字符串列都按 6B 引用计"之后仍超
	// 物理墙, 说明不动堆永远救不回来, 且这个判断发生在动堆之前。
	floor := table.rowSpillFloorForVersion(ver)
	if floor > MaxPhysicalRowData {
		return nil, fmt.Errorf("%w: 表 %q 的 schema 版本 %d 行宽溢出下界 %d 字节超过单页可容纳上限 %d 字节, 拒绝写入; "+
			"请重建该表并把大列改为动态列 (FieldType_Dynamicbyte / Blob)",
			types.ErrValueRange, util.BytesToString(table.info.Name[:]), ver, floor, MaxPhysicalRowData)
	}

	sv := table.getVersion(ver)
	if sv == nil {
		return []byte{ver}, nil
	}
	n := len(sv.FieldIndices)
	if n == 0 {
		return []byte{ver}, nil
	}

	// 基底: 旧行已存在的字段值 (版本感知解析, 旧行不会凭空多出新字段)
	var base []types.FieldValue
	if len(oldRowData) > 0 {
		base = table.deserializeRow(oldRowData)
	}

	// 行布局 (v9, 详见 types/rowvar.go):
	//   [schemaVer 1B][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图]
	//
	//   - 列 0 恒占 [1, 1+W0) 且 W0 由其【档位宽度】决定 (不是实际内容长度): 偏移表基址
	//     因此可静态算出, 解析端不必先猜列 0 有多长。时序表的列 0 就是 8 字节时间键
	//     (W0 == 8), 时间键于是落在【绝对偏移 9】, 与 index 侧写死的读取偏移一致。
	//     列 0 不足 W0 时补零 —— 与 v8 的定长存储逐字节相同。
	//   - 偏移表落在"只取决于字段数与列 0 宽度"的固定位置, 因此原地更新 (新行写进旧槽位、
	//     尾部清零) 冲不掉它, 解析永远精确, 不依赖 len(rowData)。
	//   - 表项 0 = 变长内容区起点 (= RowHeaderSize); 列 i (i ≥ 1) 的起点 = 表项 i-1。
	//   - 表项的 bit15 (RowVarTag) 为真表示该位置是 6B 堆引用: 动态列恒置位, 值级溢出
	//     的字符串列溢出时置位 —— 读侧一律以 tag 为准, 不再看列类型。
	//
	// 序列化分三步: ① 解析并编码各列 (不写缓冲); ② 算全内联长度, 决定哪些字符串列要
	// 【值级溢出】(行内装不下时把未建索引的宽字符串列降级为 6B 堆引用, 见下方 Pass 2);
	// ③ 按最终布局一次写出行字节。分步的代价是每列多一次切片拷贝, 换来溢出决策发生在
	// 任何堆写入之前 —— 布局对溢出集合是闭式的, 不需要"先写再改"。
	col0Width := table.svCol0Width(sv)
	if col0Width <= 0 {
		return nil, fmt.Errorf("%w: 表 %q 的首列 %q 宽度不可计算 (列级参数非法); "+
			"行内偏移表的位置依赖首列宽度, 无法写行",
			types.ErrValueRange, util.BytesToString(table.info.Name[:]),
			util.BytesToString(table.Field[sv.FieldIndices[0]].FieldName))
	}

	// 旧行持有的堆引用 (动态列 ∪ 值级溢出列), 布局阶段按列取用: 复用 / 释放。
	// oldRowHeapRefs 只认"动态列或带 tag 的列", 内联内容绝不会被误当引用。
	oldRefs := table.oldRowHeapRefs(oldRowData)
	releaseOld := func(masterIdx int) {
		if !checkOnly {
			if r, ok := oldRefs[masterIdx]; ok {
				_ = table.heapUnref(r.pid)
			}
		}
	}
	// v10 起 blob 列同样需要在更新路径回收旧链 ——
	//   - 改写该列 (NULL / 新 ref) 时 Pass 1 释放旧 ref;
	//   - 同值"复用"时 Pass 3 不释放, 否则会与新 ref 重叠 (同一链页释放两次 = 误回收).
	// 同样按 masterIdx 索引, 解析失败/无 blob 列时为 nil, 调用安全.
	oldBlobRefs := table.oldBlobRefsByMaster(oldRowData)
	releaseOldBlob := func(masterIdx int) {
		if !checkOnly {
			if r, ok := oldBlobRefs[masterIdx]; ok {
				_ = table.BlobFree(r)
			}
		}
	}

	// ---- Pass 1: 解析并编码各列 ----
	//
	// colBytes[i]: 动态列 = 原始内容 (布局时落堆); 其余非 NULL 列 = 行内字节
	// (字符串按实际长度, 定长按档位宽度)。未提供且旧行也没有的列保持 nil:
	// 字符串零值就是"空" (0 字节), 定长零值就是全零 (布局时按宽度零填充)。
	colBytes := make([][]byte, n)
	colNull := make([]bool, n)
	bitmapLen := types.NullBitmapLen(n)
	bitmap := make([]byte, bitmapLen)

	// 列名→值映射: 列数较多时避免 Pass 1 整体 O(列数²) 的线性扫。
	// 小表 (≤8 列) 直接线性扫更快 —— 免一次 map 分配; base (旧行) 只在缺列时
	// 兜底, 常见路径 (全列提交) 不会碰到, 保持线性扫即可。
	var valByName map[string]interface{}
	if n > 8 {
		valByName = make(map[string]interface{}, len(values))
		for _, fv := range values {
			valByName[string(fv.FieldName)] = fv.Value
		}
	}
	lookupVal := func(name []byte) (interface{}, bool) {
		if valByName != nil {
			v, ok := valByName[string(name)]
			return v, ok
		}
		return lookupFieldValue(values, name)
	}

	for i, masterIdx := range sv.FieldIndices {
		f := table.Field[masterIdx]
		name := f.FieldName
		// 取值优先级: 本次传入 > 旧行原值 (未提供的字段沿用旧值, 否则"只改一列"会清空其它列)
		value, provided := lookupVal(name)
		if !provided {
			value, provided = lookupFieldValue(base, name)
		}

		switch {
		case provided && value == nil:
			if !f.IsNullable() {
				// 非空列写 NULL: 拒绝并点名列。
				//
				// 这里不是"收紧", 而是把已有的不一致统一掉: 定长列写 nil 本来就走不通
				// (EncodeValue 对 nil 报 ErrValueType), 唯独动态列曾把 nil 静默当成
				// "空内容"存下去 (行内引用全 0, NULL 位图也不置位) —— 同一句"写 nil"
				// 在定长列报错、在动态列静默成功。既然声明了非空, 就该两边都拒。
				//
				// 注意 base (旧行原值) 不会贡献 nil: deserializeRow 只 append 非 nil 的值
				// (见那里的 `if val != nil`), 所以这条只拦"本次显式传入的 NULL",
				// 不会让"只改其它列"的更新半路失败。
				return nil, nullFieldError(util.BytesToString(name), f.FieldType)
			}
			// NULL: 位图置位, 不占内容字节 (语义完全由位图承载; 动态列/溢出列因此不写堆引用)。
			// 试算模式也要置位 —— extractIndexKey 正是靠位图判定"该行不进本索引"。
			types.SetNullBit(bitmap, i)
			// 旧行若在该列持有堆块, 新行不再引用它 —— 不在这里释放就是每更新一次漏一块。
			releaseOld(int(masterIdx))
			// v10: blob 列同样要释放旧 ref, 走 BlobFree 归还链页.
			releaseOldBlob(int(masterIdx))
		case provided && types.IsDynamic(f.FieldType):
			data, err := table.dynamicFieldData(int(masterIdx), f, value)
			if err != nil {
				return nil, err
			}
			colBytes[i] = data
		case provided && types.IsBlob(f.FieldType):
			// Blob 列行内只有 8B BlobRef, 但用户手上多半是"内容本身" ([]byte / io.Reader),
			// 由引擎在这里把内容落进链页、换成引用 (见 blobFieldRef). 已经是引用的原样用。
			// 试算行 (checkOnly) 不落盘: Blob 列不能当索引键 (见 CreateIndex), 键字节与它无关。
			ref, err := table.blobFieldRef(f, value, checkOnly)
			if err != nil {
				return nil, err
			}
			enc, err := table.encodeFieldValue(int(masterIdx), f, ref)
			if err != nil {
				return nil, err
			}
			colBytes[i] = enc
		case provided:
			enc, err := table.encodeFieldValue(int(masterIdx), f, value)
			if err != nil {
				return nil, err
			}
			if i == 0 && len(enc) > col0Width {
				return nil, fmt.Errorf("%w: 字段 %q 编码后 %d 字节超过首列档位宽度 %d",
					types.ErrEncodeBufSize, util.BytesToString(name), len(enc), col0Width)
			}
			colBytes[i] = enc
		default:
			// 未提供: 列 0 与定长列在布局阶段按档位宽度零填充; 字符串零值 = 空 (0 字节)。
			colNull[i] = false // 占位, 语义见上
		}
	}

	// ---- Pass 2: 值级溢出决策 ----
	//
	// 全内联长度超过 MaxInlineRowData ("一页两行"政策的行内配额) 时, 把【未建索引的
	// 宽字符串列】按"内容长度降序"逐个降级为 6B 堆引用, 直到退回限内。列 0 不能溢出
	// (偏移表基址依赖它的档位宽度); 建了索引的列不能溢出 (索引键按行内字节拼,
	// 见 CreateIndex 的溢出闸门); NULL 列与内容 ≤ 6B 的列没有可省的字节。
	//
	// 下界闸门 (开头的 rowSpillFloorForVersion) 保证了"把能溢出的全溢出去"一定装得下,
	// 所以这个循环必然终止于限内 (除非老表超限 —— 那已在上界处拒绝)。
	// checkOnly (试算行) 走同样的决策: 溢出的列不可能是索引分量, 键不受影响; 但堆引用
	// 只写零占位, 不碰堆。
	contentStart := types.RowHeaderSize(n, col0Width)
	inlineLen := int(contentStart) + bitmapLen
	for i := 1; i < n; i++ {
		f := table.Field[sv.FieldIndices[i]]
		switch {
		case types.IsDynamic(f.FieldType):
			// 动态列恒占 6B 堆引用 (即便这一行是 NULL: 布局仍写零引用参差对齐)。
			inlineLen += types.DynamicRefSize
		case types.IsBlob(f.FieldType):
			// blob 列同理恒占 8B 内联引用: Pass 3 对 blob 列【无条件】推进 8B
			// (NULL 只是不拷字节, 槽位仍在), 这里若按 colNull 记 0, 末尾那句
			// "行布局长度失配" 就会在"写一行带 NULL blob 列"时直接报错。
			inlineLen += types.BlobRefSize
		case colNull[i]:
			// NULL 列不占内容字节 (仅位图置位)。
		default:
			// 必须与 Pass 3 用同一个占用公式, 否则"预算"与"实际布局"会对不上:
			// 字符串列在这里也要按实际长度计, 而不是档位宽度。
			inlineLen += fieldInlineWidth(f, colBytes[i])
		}
	}
	spilled := make([]bool, n)
	if inlineLen > types.MaxInlineRowData {
		ineligible := table.spillIneligibleFields()
		type spillCand struct {
			i, gain int
		}
		cands := make([]spillCand, 0, n)
		for i := 1; i < n; i++ {
			if spilled[i] || colNull[i] {
				continue
			}
			masterIdx := int(sv.FieldIndices[i])
			f := table.Field[masterIdx]
			if !types.IsVarWidth(f.FieldType) || ineligible[masterIdx] {
				continue
			}
			if gain := len(colBytes[i]) - types.DynamicRefSize; gain > 0 {
				cands = append(cands, spillCand{i, gain})
			}
		}
		sort.Slice(cands, func(a, b int) bool { return cands[a].gain > cands[b].gain })
		cur := inlineLen
		for _, c := range cands {
			if cur <= types.MaxInlineRowData {
				break
			}
			spilled[c.i] = true
			cur -= c.gain
		}
	}

	// ---- Pass 3: 布局 (一次成形, 长度精确) ----
	// 预算必须扣掉"值级溢出省下的字节": Pass 3 对溢出列只写 6B 堆引用, 而 inlineLen
	// 仍按溢出前的实际长度计那些列。少扣这一笔, 末尾的 length 断言必然失败 —— 而且
	// 只在真正触发溢出时才失败 (窄行永远测不出来)。
	finalLen := inlineLen - curAdjust(spilled, colBytes)
	rowData := make([]byte, finalLen)
	rowData[0] = ver
	ends := make([]uint16, n)
	refs := make([]bool, n) // 偏移表项要置 RowVarTag 的列 (动态列 ∪ 溢出列)
	ends[0] = uint16(contentStart)

	// 列 0: 档位宽度定长存储 —— 编码字节拷进 [1, 1+W0), 余下保持全零。
	switch f0 := table.Field[sv.FieldIndices[0]]; {
	case colNull[0]:
		// NULL: 位图已置位, 不占内容字节 (旧引用已在 Pass 1 释放)。
	case types.IsDynamic(f0.FieldType):
		// 动态首列 (理论情形): 行内就是 6B 堆引用, 与其它动态列走同一条路。
		// 绝不能把内容原样拷进 6B 槽位 —— 那会盖住后面的偏移表, 整行错位。
		if err := table.writeHeapRefField(rowData, 1, colBytes[0],
			oldRefOrNil(oldRefs, int(sv.FieldIndices[0])), checkOnly); err != nil {
			return nil, err
		}
	case types.IsBlob(f0.FieldType):
		// Blob 首列: 行内 8B BlobRef 直引链页, 与其它 blob 列一致. 不走堆.
		if c0 := colBytes[0]; len(c0) >= types.BlobRefSize {
			copy(rowData[1:1+types.BlobRefSize], c0[:types.BlobRefSize])
		}
	default:
		if c0 := colBytes[0]; len(c0) > 0 {
			copy(rowData[1:], c0)
		}
	}

	offset := uint16(contentStart)
	for i := 1; i < n; i++ {
		f := table.Field[sv.FieldIndices[i]]
		isDynamic := types.IsDynamic(f.FieldType)
		isBlob := types.IsBlob(f.FieldType)
		switch {
		case isDynamic || spilled[i]:
			refs[i] = true
			old := oldRefOrNil(oldRefs, int(sv.FieldIndices[i]))
			if err := table.writeHeapRefField(rowData, offset, colBytes[i], old, checkOnly); err != nil {
				return nil, err
			}
			offset += types.DynamicRefSize
		case isBlob:
			// v10 起 Blob 列行内 8B BlobRef (直引链页), 不再走堆块。
			// colBytes[i] = 8B [TotalLen BE 4][FirstPageID BE 4] (由 encodeFieldValue 编码);
			// 空 / NULL 时为 0 字节 — 不拷任何字节, 但 offset 必须推进 8B 保持布局.
			if !colNull[i] && len(colBytes[i]) >= types.BlobRefSize {
				copy(rowData[offset:offset+types.BlobRefSize], colBytes[i][:types.BlobRefSize])
				// 改值情形 (新 ref 与旧 ref 不同): 旧链必须归还空闲链.
				// 同值"复用" (FirstPageID 与 TotalLen 都一致) 不释放 ——
				// 释放后再写新行会立刻挂空链. NULL 情形在 Pass 1 已释放.
				if !checkOnly {
					if oldRef, ok := oldBlobRefs[int(sv.FieldIndices[i])]; ok {
						newRef := BlobRef{
							TotalLen:    binary.BigEndian.Uint32(colBytes[i][:4]),
							FirstPageID: binary.BigEndian.Uint32(colBytes[i][4:8]),
						}
						if oldRef != newRef {
							_ = table.BlobFree(oldRef)
						}
					}
				}
			}
			offset += types.BlobRefSize
			// 不置 RowVarTag — BlobRef 是内联 8B 字节, 读侧按类型识别.
		case colNull[i]:
			// 不占内容字节
		default:
			if len(colBytes[i]) > 0 {
				copy(rowData[offset:], colBytes[i])
			}
			// 定长列未提供时按档位宽度零填充 (make 已清零, 只推进偏移)。
			offset += uint16(fieldInlineWidth(f, colBytes[i]))
			// 旧行该列若曾溢出到堆 (值级溢出), 新行收回内联 → 旧块不再被引用, 释放。
			releaseOld(int(sv.FieldIndices[i]))
		}
		ends[i] = offset
	}

	// NULL 位图放在偏移表末项之后。
	copy(rowData[offset:], bitmap)
	if int(offset)+bitmapLen != finalLen {
		return nil, fmt.Errorf("%w: 行布局长度失配 (内容末尾 %d + 位图 %d != 预算 %d)",
			types.ErrEncodeBufSize, offset, bitmapLen, finalLen)
	}
	types.PutRowOffsets(rowData, ends, refs, col0Width)
	return rowData, nil
}

// fieldInlineWidth 非动态、非 NULL 列的行内占用。
//
//   - 变长 (字符串家族): 按【实际长度】占位 —— 空串 / 未提供都是 0 字节, 这才是 v9 省
//     空间的本体; 若按档位宽度零填充, 一个未提供的 1024 宽串会白占 1KB。
//   - 定长: 有编码字节按实际长度, 未提供则必须按档位宽度零填充 —— 定长列的零值是全零
//     字节, 必须占住位置, 否则后面所有列错位。
func fieldInlineWidth(f types.FieldInfo, enc []byte) int {
	if types.IsVarWidth(f.FieldType) || len(enc) > 0 {
		return len(enc)
	}
	return int(f.Width())
}

// oldRefOrNil 从旧行引用表取某列的引用 (无则 nil)。
func oldRefOrNil(refs map[int]oldHeapRefInfo, masterIdx int) *oldHeapRefInfo {
	if r, ok := refs[masterIdx]; ok {
		return &r
	}
	return nil
}

// curAdjust 计算溢出集合实际省下的行内字节数 (仅用于布局预算自检)。
func curAdjust(spilled []bool, colBytes [][]byte) int {
	saved := 0
	for i, sp := range spilled {
		if sp {
			saved += len(colBytes[i]) - types.DynamicRefSize
		}
	}
	return saved
}

func writeFieldValue(buf []byte, ft types.FieldType, value interface{}) {
	switch ft {
	case types.FieldType_Uint8:
		buf[0] = value.(byte)
	case types.FieldType_Uint16:
		binary.BigEndian.PutUint16(buf, uint16(value.(int16)))
	case types.FieldType_Uint32:
		binary.BigEndian.PutUint32(buf, uint32(value.(int32)))
	case types.FieldType_Uint64:
		binary.BigEndian.PutUint64(buf, value.(uint64))
	case types.FieldType_Int8:
		buf[0] = types.Int8ToSortKey(value.(int8))
	case types.FieldType_Int16:
		binary.BigEndian.PutUint16(buf, types.Int16ToSortKey(value.(int16)))
	case types.FieldType_Int32:
		binary.BigEndian.PutUint32(buf, types.Int32ToSortKey(value.(int32)))
	case types.FieldType_Int64:
		// 同 Float64: 存保序编码 (Int64ToSortKey) 而不是补码本身。补码大端对负数不保序,
		// 会让跨 0 的范围扫描 / ORDER BY 静默漏行; 换成保序编码后行内字节序即数值序,
		// 索引键直接复用行内字节 (extractIndexKey 原样切片) 也就自动保序。
		binary.BigEndian.PutUint64(buf, types.Int64ToSortKey(value.(int64)))
	case types.FieldType_Float64:
		// 存保序编码 (Float64ToSortKey) 而不是原始位模式: 行内字节序即数值序,
		// 索引键直接复用行内字节 (extractIndexKey 原样切片) 也就自动保序。
		binary.BigEndian.PutUint64(buf, types.Float64ToSortKey(toFloat64(value)))
	default:
		// 所有字符串/定长类型 (动态类型不走此分支, 由 writeDynamicField 处理)
		if b, ok := value.([]byte); ok {
			copy(buf, b)
		}
	}
}

// deserializeRow 将行数据字节反序列化为 FieldValue 切片 (不含 pkID 前缀)。
//
// 行格式 (v9): [schemaVer 1B][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图]
//
// 列位置一律由行内偏移表给出 —— 旧实现"从偏移 1 开始累加各列 Width"的前提是字段区
// 定长, 而 v9 的字符串列按实际长度存储, 累加 Width 会越算越偏 (表现为整行错列)。
func (table *Table) deserializeRow(rowData []byte) []types.FieldValue {
	return table.deserializeRowMode(rowData, nil, false)
}

// deserializeRowKeepNull 与 deserializeRow 相同, 但 NULL 列也产出
// FieldValue{Value: nil} (而不是跳过)。供 VACUUM 快照使用: 重建插入时
// 显式传 nil 走 NULL 位图路径 —— deserializeRow 的"跳过 NULL"会让 VACUUM
// 把 NULL 物化成类型零值 (可空列语义被静默改写, 且不可逆)。
func (table *Table) deserializeRowKeepNull(rowData []byte) []types.FieldValue {
	return table.deserializeRowMode(rowData, nil, true)
}

// deserializeRowSelect 只解码掩码选中的列 (掩码按 master 下标)。
//
// 与 deserializeRow 同一套布局解析, 未选中列跳过解码与其堆解引用 —— 宽表的投影查询
// 不再为不看的列付解码/堆页 IO 成本。掩码必须包含调用方后续要读的列
// (Query.deserialize 负责把 ORDER BY 列并进来)。
func (table *Table) deserializeRowSelect(rowData []byte, want []bool) []types.FieldValue {
	return table.deserializeRowMode(rowData, want, false)
}

// deserializeRowMode 行解码的公共实现。want 非 nil 时是投影掩码 (master
// 下标 → 是否解码); keepNull 为 true 时 NULL 列产出显式 nil 值。
func (table *Table) deserializeRowMode(rowData []byte, want []bool, keepNull bool) []types.FieldValue {
	if len(rowData) < 1 {
		return nil
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return nil // 未知版本, 无法解析
	}
	n := len(sv.FieldIndices)
	col0Width := table.svCol0Width(sv)
	ends, ok := types.RowParseOffsets(rowData, n, col0Width)
	if !ok {
		return nil
	}
	bitmap, ok := types.RowNullBitmap(rowData, ends, n)
	if !ok {
		return nil
	}

	// 预分配容量: 字段数即结果条数的上界 (被删除 / NULL 字段会跳过), 避免 append 反复扩容
	result := make([]types.FieldValue, 0, n)
	for i, idx := range sv.FieldIndices {
		if want != nil && int(idx) < len(want) && !want[idx] {
			continue // 投影模式: 未选中的列不解码、不解堆引用
		}
		f := table.Field[idx]
		// 跳过已删除字段 (master registry 中标记 FlagDeleted) 与 NULL 字段。
		// NULL 不出现在结果里 —— 与"零值不 append"是同一约定: 读不到值就是没有值;
		// 更新路径靠它区分"沿用旧值"与"显式写 NULL"(见 serializeRow 的取值优先级)。
		// keepNull 模式 (VACUUM 快照) 例外: NULL 列产出显式 nil, 重建时走 NULL 位图。
		if f.IsDropped() || types.NullBitSet(bitmap, i) {
			if keepNull && !f.IsDropped() && types.NullBitSet(bitmap, i) {
				result = append(result, types.FieldValue{FieldName: f.FieldName, Value: nil})
			}
			continue
		}
		start := int(types.RowFieldStart(ends, i))
		end := int(types.RowFieldEnd(ends, i, col0Width))
		if end < start || end > len(rowData) {
			continue
		}
		seg := rowData[start:end]
		var val interface{}
		switch {
		case types.IsDynamic(f.FieldType) || types.RowFieldRefTagged(rowData, i, col0Width):
			// 解引用堆页。NULL / 未提供的动态列内容为 0 字节或全零引用 —— 读不到就是没有值。
			if len(seg) < types.DynamicRefSize {
				continue
			}
			pageID := binary.BigEndian.Uint32(seg[:4])
			offInPage := binary.BigEndian.Uint16(seg[4:6])
			if pageID == 0 {
				continue // 空引用 = 空内容, 不产出"空值"
			}
			if data, err := table.heapRead(pageID, offInPage); err == nil {
				val = data
			}
		case types.IsBlob(f.FieldType):
			// v10 起 Blob 列行内 8B BlobRef, 不走堆页. 直接还原 BlobRef.
			// 空 blob (TotalLen=0, FirstPageID=0) 视为"读不到", 与其它列的零值规则一致.
			if len(seg) < types.BlobRefSize {
				continue
			}
			ref := types.BlobRef{
				TotalLen:    binary.BigEndian.Uint32(seg[:4]),
				FirstPageID: binary.BigEndian.Uint32(seg[4:8]),
			}
			if ref.Empty() {
				continue
			}
			val = ref
		case types.IsVarWidth(f.FieldType):
			// 字符串家族: seg 就是实际字节 (无补零), DecodeValue 按 len 解码, 天然支持变长。
			val = types.DecodeValue(f.FieldType, seg)
		default:
			// 定长类型: 长度必须恰好等于档位宽度, 否则行已损坏 (DecodeValue 对定长类型会
			// 按下标取值)。这时宁可跳过该列, 也不要按错位字节解出一个"看似合理"的数值。
			if len(seg) != int(f.Width()) {
				continue
			}
			val = types.DecodeValue(f.FieldType, seg)
		}
		if val != nil {
			result = append(result, types.FieldValue{FieldName: f.FieldName, Value: val})
		}
	}
	return result
}

// deserializeField 解码单个字段, 不物化整行。
//
// 聚合 (Sum/Max/Min) 每行只需要一列; 用 deserializeRow 意味着每行都分配一个
// []FieldValue 并解码全部活跃列, 100 万行的 SUM 里这占了绝对大头。
// 解码逻辑与 deserializeRow 逐分支一致 (行布局解析、NULL 位图、堆解引用),
// 只是跳过"其余字段的解码与组装"。
// 返回 (值, true); 字段不存在、NULL、行损坏时返回 (nil, false)。
func (table *Table) deserializeField(rowData []byte, fieldName string) (interface{}, bool) {
	if len(rowData) < 1 {
		return nil, false
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return nil, false
	}
	n := len(sv.FieldIndices)
	// 找到字段在行内的位置 (与 deserializeRow 的循环同一编号空间)
	pos := -1
	for i, idx := range sv.FieldIndices {
		f := table.Field[idx]
		if f.IsDropped() {
			continue
		}
		if string(f.FieldName) == fieldName {
			pos = i
			break
		}
	}
	if pos < 0 {
		return nil, false
	}
	col0Width := table.svCol0Width(sv)
	ends, ok := types.RowParseOffsets(rowData, n, col0Width)
	if !ok {
		return nil, false
	}
	bitmap, ok := types.RowNullBitmap(rowData, ends, n)
	if !ok {
		return nil, false
	}
	if types.NullBitSet(bitmap, pos) {
		return nil, false // NULL: 聚合语义上"读不到值"
	}
	f := table.Field[sv.FieldIndices[pos]]
	start := int(types.RowFieldStart(ends, pos))
	end := int(types.RowFieldEnd(ends, pos, col0Width))
	if end < start || end > len(rowData) {
		return nil, false
	}
	seg := rowData[start:end]
	switch {
	case types.IsDynamic(f.FieldType) || types.RowFieldRefTagged(rowData, pos, col0Width):
		if len(seg) < types.DynamicRefSize {
			return nil, false
		}
		pageID := binary.BigEndian.Uint32(seg[:4])
		offInPage := binary.BigEndian.Uint16(seg[4:6])
		if pageID == 0 {
			return nil, false
		}
		if data, err := table.heapRead(pageID, offInPage); err == nil {
			return data, true
		}
		return nil, false
	case types.IsBlob(f.FieldType):
		if len(seg) < types.BlobRefSize {
			return nil, false
		}
		ref := types.BlobRef{
			TotalLen:    binary.BigEndian.Uint32(seg[:4]),
			FirstPageID: binary.BigEndian.Uint32(seg[4:8]),
		}
		if ref.Empty() {
			return nil, false
		}
		return ref, true
	case types.IsVarWidth(f.FieldType):
		return types.DecodeValue(f.FieldType, seg), true
	default:
		if len(seg) != int(f.Width()) {
			return nil, false // 行损坏: 宁可当没有值, 也不解错位字节
		}
		val := types.DecodeValue(f.FieldType, seg)
		return val, val != nil
	}
}

// readFieldValue / 行内字段解码已收敛到 types.DecodeValue (见 types/codec.go):
// 解码与编码同源, 新增类型不必再回来改这里。

// ======================== 高层 CRUD ========================

// InsertRow 插入一行数据, pkID 由引擎按自增游标分配 (只增不减, 见 AllocPKID)。
func (table *Table) InsertRow(values []types.FieldValue) (uint64, error) {
	return table.InsertRowWithID(0, values)
}

// InsertRowWithID 插入一行数据, pkID == 0 表示由引擎自增分配。
//
// 允许调用方指定主键: 上层 (ORM/导入) 往往已经拿到一个明确的行标识, 旧实现
// 无条件用 TotalRows+1 覆盖, 结果是"调用方以为按 _id 写进去了, 之后按 _id 却
// 查不到" —— 行确实存在, 只是不可达 (数据静默错位的一类)。
//
// 指定的主键若已存在则报 ErrDuplicatePK, 避免同一 pkID 在主键树里出现两条记录。
// 例外: 引擎刚分配并"预留"的号 (AllocPKID) 必然唯一, 跳过这次查重 ——
// 查重走的是回表 ResolveByPK, 会重建 O(页数) 的定位缓存, 而每次插入又使其失效;
// 逐行插入将因此退化成 O(N²) (千万级写入从分钟级变成不可完成)。
func (table *Table) InsertRowWithID(pkID uint64, values []types.FieldValue) (uint64, error) {
	if pkID == 0 {
		pkID = table.AllocPKID()
	}
	reserved := table.hasPendingPK && table.pendingPK == pkID
	table.hasPendingPK = false
	if !reserved {
		if table.indexMeta.rowStoreHasPK(pkID) {
			return 0, fmt.Errorf("%w: pkID=%d", ErrDuplicatePK, pkID)
		}
	}
	// 显式指定更大的主键时把游标顶到它之后, 避免后续自增再发一次同样的号
	if pkID+1 > table.nextPK {
		table.nextPK = pkID + 1
	}

	rowData, err := table.serializeRow(values, nil)
	if err != nil {
		return 0, err
	}
	if err := table.indexMeta.InsertRow(table.Field, pkID, rowData, table.activeVersion, table.versions); err != nil {
		return 0, err
	}
	table.info.TotalRows++
	table.saveTableInfo()
	if table.onWrite != nil {
		table.onWrite()
	}
	return pkID, nil
}

// InsertRowsWithIDs 批量插入。pkIDs[i]==0 的行由引擎自增分配 (号先拿到、写进
// _id 列后再序列化, 与 ORM 单行路径同一约定); pkIDs[i]!=0 为显式主键, 逐个走
// 查重。返回实际使用的 pkID 列表 (与入参等长)。
//
// 与逐行 InsertRowWithID 的语义差异只在"摊销", 不在"语义":
//   - 自增号来自只增不减的游标, 整批天然唯一, 跳过逐行查重 (单行路径靠
//     pendingPK 预留跳过同一件事; 批量下预留槽只有一格, 改为按"号段"整批豁免);
//   - 显式主键照旧 rowStoreHasPK 查重, 另补批内查重 (单行路径里批内冲突会
//     在第 i+1 行的查重中被发现, 这里一次性比出来);
//   - 序列化 (含动态堆写入) 仍逐行做 —— 堆页的写入粒度是页级的, 堆块追加
//     本身已经近似批量; saveTableInfo / onWrite 摊到批末一次。
//
// 失败语义: 序列化阶段报错时, 先前行写入的堆块成为无主垃圾 (泄漏而非错乱,
// 行数据并未落任何索引, 不会被读到); 索引阶段报错时部分索引已写入 —— 与
// 单行路径"写了一半报错"一致, 靠事务回滚兜底。
func (table *Table) InsertRowsWithIDs(pkIDs []uint64, values [][]types.FieldValue) ([]uint64, error) {
	return table.insertRows(pkIDs, values, false)
}

// InsertRowsReserved 批量插入"引擎预留号"的行: pkIDs 全部来自 AllocPKID
// (ORM 先要号填进对象、再连同行数据一起送回来), 唯一性由只增不减的游标保证,
// 逐行查重整个跳过 —— 语义同单行路径的 pendingPK 预留豁免, 只是预留号不止
// 一格, 按号段整批豁免。批内查重仍保留 (号段自身不可能重复, 这层是防御)。
//
// 注意: 走这个入口的前提是调用方【没有】夹带任何非预留号 —— 夹带了请用
// InsertRowsWithIDs (显式号走逐行查重)。
func (table *Table) InsertRowsReserved(pkIDs []uint64, values [][]types.FieldValue) ([]uint64, error) {
	return table.insertRows(pkIDs, values, true)
}

func (table *Table) insertRows(pkIDs []uint64, values [][]types.FieldValue, allReserved bool) ([]uint64, error) {
	n := len(values)
	if n == 0 {
		return nil, nil
	}
	if len(pkIDs) != n {
		return nil, fmt.Errorf("tsdb: 批量插入: pkIDs (%d) 与 values (%d) 数量不一致", len(pkIDs), n)
	}

	// 1) 补齐自增号
	auto := make([]bool, n)
	for i := 0; i < n; i++ {
		if pkIDs[i] == 0 {
			pkIDs[i] = table.AllocPKID()
			auto[i] = true
		}
	}
	table.hasPendingPK = false // 预留号已被本批整体消费

	// 2) 批内主键查重 (显式 vs 显式, 以及显式撞上自增号段)
	{
		seen := make(map[uint64]struct{}, n)
		for i := 0; i < n; i++ {
			if _, dup := seen[pkIDs[i]]; dup {
				return nil, fmt.Errorf("%w: pkID=%d (批内重复)", ErrDuplicatePK, pkIDs[i])
			}
			seen[pkIDs[i]] = struct{}{}
		}
	}
	// 3) 显式主键对既有行查重 (自增/预留号必然唯一, 跳过)
	if !allReserved {
		for i := 0; i < n; i++ {
			if auto[i] {
				continue
			}
			if table.indexMeta.rowStoreHasPK(pkIDs[i]) {
				return nil, fmt.Errorf("%w: pkID=%d", ErrDuplicatePK, pkIDs[i])
			}
		}
	}
	// 4) 显式大号顶起自增游标 (与单行路径一致; 预留号本就来自游标, 无需)
	if !allReserved {
		for i := 0; i < n; i++ {
			if !auto[i] && pkIDs[i]+1 > table.nextPK {
				table.nextPK = pkIDs[i] + 1
			}
		}
	}

	// 5) 逐行序列化 (动态堆写入在这一步发生)
	rowDatas := make([][]byte, n)
	for i := 0; i < n; i++ {
		rowData, err := table.serializeRow(values[i], nil)
		if err != nil {
			return nil, err
		}
		rowDatas[i] = rowData
	}

	// 6) 索引层批量写入
	if err := table.indexMeta.InsertRows(table.Field, pkIDs, rowDatas, table.activeVersion, table.versions); err != nil {
		return nil, err
	}

	table.info.TotalRows += uint64(n)
	table.saveTableInfo()
	if table.onWrite != nil {
		table.onWrite()
	}
	return pkIDs, nil
}

// AllocPKID 分配并预留下一个自增 pkID。
//
// 供"必须先把号写进对象、再序列化整行"的上层使用 (典型是 ORM:
// 只有先拿到 pkID 填进 _id 列, 存进行里的 _id 才与真实 pkID 一致)。
// 分配即推进游标并预留该号, 紧随其后的 InsertRowWithID 会跳过查重。
// 若最终未插入, 该号只是被跳过 —— 游标只增不减, 不会与后续插入冲突。
func (table *Table) AllocPKID() uint64 {
	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
}

// SearchByIndex 通过索引搜索一行
func (table *Table) SearchByIndex(idxName string, key []byte) ([]types.FieldValue, error) {
	// 元信息只查一次, 命中后复用给 searchAt, 避免重复扫描 metas
	meta := table.indexMeta.findIndexMetaByName(idxName)
	if meta == nil {
		return nil, ErrFieldNotFound
	}
	storeRow, found := table.indexMeta.searchAt(meta, key)
	// storeRow = [pkID(8B BE) + rowData]
	if !found || len(storeRow) < 8 {
		return nil, page.ErrNotFound
	}
	return table.deserializeRow(storeRow[8:]), nil
}

// DeleteRow 通过索引删除一行
func (table *Table) DeleteRow(idxName string, key []byte) error {
	// 一次定位: 既用于释放该行的堆页引用, 也下传给索引层删除 —— 避免同一行被查两次树
	storeRow, found := table.indexMeta.SearchByIndex(idxName, key)
	if !found || len(storeRow) < 8 {
		return page.ErrNotFound
	}
	if err := table.indexMeta.DeleteByKey(idxName, key, storeRow, table.Field, table.activeVersion, table.versions); err != nil {
		return err
	}
	// 行已删除 → 释放其动态字段占用的堆块; 计数归零的堆页会被摘链回收
	table.heapReleaseRow(storeRow[8:])
	table.info.TotalRows--
	table.saveTableInfo()
	if table.onWrite != nil {
		table.onWrite()
	}
	return nil
}

// checkUniqueUpdate 更新前的唯一性预检 —— 必须早于 serializeRow。
//
// 用 serializeRowForCheck 试算行字节 (全程不碰堆), 再交给索引层比对。冲突时旧行分毫
// 未动, 动态堆块也没有被覆写或释放。代价是更新路径多一次序列化, 但更新本来就要回表、
// 定位、重写行, 这笔开销可接受; 换来的是"要么全改, 要么全不动"。
//
// 为什么不能只靠 UpdateByKey 里那次预检: 那时 serializeRow 已经跑过了, 旧行的动态
// 列早就被改成新值, 而固定列与索引条目还是旧的 —— 行被改了一半。
func (table *Table) checkUniqueUpdate(values []types.FieldValue, oldRowData []byte, pkID uint64) error {
	// 没有唯一索引的表占绝大多数: 预检唯一的目的就是拦唯一冲突, 此时整行试序列化
	// (serializeRowForCheck) 纯属浪费 —— 后续真正的 serializeRow 仍会校验值宽度,
	// 非法值照样在写入前被拦下, 只是少了这次纯为唯一性服务的试算。
	if !table.indexMeta.HasUniqueIndex() {
		return nil
	}
	scratch, err := table.serializeRowForCheck(values, oldRowData)
	if err != nil {
		return err
	}
	return table.indexMeta.CheckUniqueForKey(scratch, pkID, table.Field, table.activeVersion, table.versions)
}

// UpdateRow 通过索引更新一行
func (table *Table) UpdateRow(idxName string, key []byte, values []types.FieldValue) error {
	// 先取旧行: 动态字段可复用旧行已占用的堆块, 避免每次更新都产生堆垃圾。
	// 该行同时下传给 UpdateByKey, 免去索引层再次回表。
	storeRow, found := table.indexMeta.SearchByIndex(idxName, key)
	if !found {
		return ErrFieldNotFound
	}
	if len(storeRow) < 8 {
		return ErrFieldNotFound
	}
	pkID := binary.BigEndian.Uint64(storeRow[:8])

	// 唯一性预检放在序列化之前: 见 checkUniqueUpdate。
	if err := table.checkUniqueUpdate(values, storeRow[8:], pkID); err != nil {
		return err
	}

	newRowData, err := table.serializeRow(values, storeRow[8:])
	if err != nil {
		return err
	}
	if err := table.indexMeta.UpdateByKey(key, newRowData, pkID, storeRow, table.Field, table.activeVersion, table.versions); err != nil {
		return err
	}
	if table.onWrite != nil {
		table.onWrite()
	}
	return nil
}

// ======================== 按行号 (pkID) 直接读写 ========================
//
// 索引键只决定"落在哪个分段/哪一批", 行号才唯一确定"是哪一行"。
// 时序表里同一个时间戳允许存在多行 (同一时刻的多个测点/多次事件), 而上面那组
// "按索引键取第一条"的入口只能碰到第一行 —— 其余行存得进、读不到、改不了、删不掉。
// 下面这组入口按行号直达, 供 ORM (与任何需要精确定位一行的上层) 使用。

// FindByPK 按行号读取一行的字段值。
func (table *Table) FindByPK(pkID uint64) ([]types.FieldValue, error) {
	storeRow, found := table.indexMeta.ResolveByPK(pkID)
	if !found || len(storeRow) < 8 {
		return nil, page.ErrNotFound
	}
	return table.deserializeRow(storeRow[8:]), nil
}

// SearchByIndexWithPK 同 SearchByIndex, 但额外把该行的真实行号一并返回。
//
// 行号是行的 8 字节前缀, 不在字段值里 —— 上层要把它带回对象, 之后才能按行号
// 精确 Update/Delete (见 ORM 的 modelCore.rowID)。
func (table *Table) SearchByIndexWithPK(idxName string, key []byte) (uint64, []types.FieldValue, error) {
	meta := table.indexMeta.findIndexMetaByName(idxName)
	if meta == nil {
		return 0, nil, ErrFieldNotFound
	}
	storeRow, found := table.indexMeta.searchAt(meta, key)
	// storeRow = [pkID(8B BE) + rowData]
	if !found || len(storeRow) < 8 {
		return 0, nil, page.ErrNotFound
	}
	return binary.BigEndian.Uint64(storeRow[:8]), table.deserializeRow(storeRow[8:]), nil
}

// DeleteRowByPK 按行号删除一行, 并级联清掉它在全部索引里的条目。
func (table *Table) DeleteRowByPK(pkID uint64) error {
	// 一次回表: 既拿到行数据用于释放动态字段的堆块, 也下传给索引层定位 pkID。
	storeRow, found := table.indexMeta.ResolveByPK(pkID)
	if !found || len(storeRow) < 8 {
		return page.ErrNotFound
	}
	name, _, ok := table.indexMeta.rowStoreMeta()
	if !ok {
		return ErrFieldNotFound
	}
	// indexKey 传 nil: 行数据在手, 每个索引都能自己从行里推出键 (时序键/主键列)。
	if err := table.indexMeta.DeleteByKey(name.Name(), nil, storeRow, table.Field, table.activeVersion, table.versions); err != nil {
		return err
	}
	table.heapReleaseRow(storeRow[8:])
	table.info.TotalRows--
	table.saveTableInfo()
	if table.onWrite != nil {
		table.onWrite()
	}
	return nil
}

// UpdateRowByPK 按行号更新一行, 并级联更新它在全部索引里的条目。
func (table *Table) UpdateRowByPK(pkID uint64, values []types.FieldValue) error {
	// 旧行既用于复用动态字段堆块, 也用于让索引层提取"旧键" (时序键变了要搬分段)。
	storeRow, found := table.indexMeta.ResolveByPK(pkID)
	if !found || len(storeRow) < 8 {
		return ErrFieldNotFound
	}

	// 唯一性预检放在序列化之前: 见 checkUniqueUpdate。
	if err := table.checkUniqueUpdate(values, storeRow[8:], pkID); err != nil {
		return err
	}

	newRowData, err := table.serializeRow(values, storeRow[8:])
	if err != nil {
		return err
	}
	if err := table.indexMeta.UpdateByKey(nil, newRowData, pkID, storeRow, table.Field, table.activeVersion, table.versions); err != nil {
		return err
	}
	if table.onWrite != nil {
		table.onWrite()
	}
	return nil
}