TsDB2/DB/core/Table.go

package core

import (
	"TsDB2/DB/core/page"
	"TsDB2/DB/core/types"
	"TsDB2/DB/core/util"
	"bytes"
	"encoding/binary"
	"fmt"
	"sort"
	"unsafe"
)

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

// CreateIndex 建二级索引。
//
// 这里先校验分量列能不能做键, 再交给 IndexMetaPage 建 —— 拆成两层是因为
// IndexMetaPage 只拿得到索引自己的元信息, 看不到表的字段类型:
//   - 分量列不存在 → ErrFieldNotFound;
//   - 分量是动态列 → ErrIndexDynamicField。动态列在行内只存 6B 堆引用, 值 != 引用,
//     按值查找永远不可能命中; 建出来既查不到也不会被查询侧选中 (buildKeyLayout 会跳过),
//     只会白占空间、白写。拦在建的这一步, 而不是等用到才发现;
//   - 复合索引 (IndexTypeCompose) 的分量宽度不是 8 字节 → ErrIndexComposeWidth。
//     ComposeEntry 把每个分量写死成 8 字节 (EntryLength = 8*n+8, Compare 里
//     offset := i*8 后按 Uint64 读), 而拼键侧 (extractIndexKey) 是按分量的真实宽度拼的。
//     两者对不上时条目长度与键长度不一致, Compare 会读到相邻分量的字节 —— 排序和查找
//     会稳定地给出错误结果, 且不报错。这不是新引入的问题 (16byte 字段今天就对不上),
//     但新类型会让它更容易被踩到, 所以在建立时就拒绝, 而不是让它静默错位。
//     需要非 8 字节分量请用 B+ 树/哈希索引 (按真实宽度拼键, 无此限制)。
//
// 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)
		}
		if typ == page.IndexTypeCompose {
			if w := table.Field[fi].Width(); w != 8 {
				return nil, fmt.Errorf("%w (索引 %q 的第 %d 个分量是 %s, 宽度 %d)",
					ErrIndexComposeWidth, idxName, i+1, types.TypeNameOf(ft), w)
			}
		}
	}
	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
	}
	sort.SliceStable(entries, func(i, j int) bool {
		return compareBytes(entries[i][:len(entries[i])-8], entries[j][:len(entries[j])-8]) < 0
	})
	// 唯一索引进来了要顺手验一遍存量数据: 库里本来就有重复值时, 这个约束是【建不出来】的。
	// 硬建下去只会得到一个"违反自身约束"的索引 (同一键挂着两行), 之后点查可能返回其中
	// 任意一行, 而且从索引上完全看不出它已经坏了。宁可在建索引这一步报错, 由调用方决定
	// 是清理数据还是改用非唯一索引。排序后重复键必然相邻, 一次线性扫描即可。
	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)
		}
		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 中的索引
func (table *Table) findFieldIndex(name string) int {
	for i, field := range table.Field {
		if string(field.FieldName) == name {
			return i
		}
	}
	return -1
}

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

// getVersionedOffset 计算某行中指定 master 字段索引在行数据中的偏移。
//
// 行布局: [schemaVer 1B] [字段区...] [NULL 位图 ⌈n/8⌉ B]
// 位图长度按【行自身的 schemaVer】算 (n = 该版本 FieldIndices 长度), 不能取当前字段数。
//
// 位图【放在行尾】而不是 schemaVer 之后, 是为了让字段区起点恒为偏移 1:
// 时序索引按固定偏移 9 (== [pkID 8B][schemaVer 1B]) 读首列当时间键
// (见 DB/types.go 的 modelInfo 与 core/index 的 timeRowInRange / FindRowByIndexKey),
// 位图若前插会把首列整体后移 ⌈n/8⌉ 字节, 时序键读取全错 —— "行写得进却 Load 不到"。
// 放行尾则字段偏移与旧格式完全一致, 这些固定偏移读取器无需感知位图。
//
// 字段为 NULL 时返回 false。把 NULL 判定下沉到这里, 是因为"定位偏移"是所有读取路径
// (7 个条件类 + 更新时摘旧堆块) 的唯一入口, 各调用方再各判一次必然漏:
//   - 条件匹配: NULL 不匹配任何比较 (SQL 三值逻辑)。若按全零字节比, `x = 0` 会命中
//     NULL 行、`x < 1` 也会命中 —— 结果看着"有数据"却全是错的。
//   - 更新路径: 旧值为 NULL 时引用槽是全零, 本就没有堆块要摘。
func (table *Table) getVersionedOffset(rowData []byte, masterFieldIdx int) (uint16, bool) {
	if len(rowData) < 1 {
		return 0, false
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return 0, false
	}
	bitmap, ok := table.versionedNullBitmap(rowData, ver)
	if !ok {
		return 0, false
	}
	offset := uint16(1) // 跳过 schemaVer; 位图在行尾, 不影响字段起点
	for i, idx := range sv.FieldIndices {
		if int(idx) == masterFieldIdx {
			if types.NullBitSet(bitmap, i) {
				return 0, false // NULL: 不参与比较, 也没有堆块
			}
			return offset, true
		}
		offset += table.Field[idx].Width()
	}
	return 0, false
}

// versionedNullBitmap 取出行尾的 NULL 位图切片 (长度按该行自身的 schemaVer 算)。
//
// 行布局: [schemaVer 1B] [字段区...] [NULL 位图 ⌈n/8⌉ B]
// 位图位置 = 该版本行总长 - ⌈n/8⌉。用【版本算出的总长】而不是 len(rowData) 定位:
// 行存储的原地更新允许把新行写进旧行的槽位 (剩余字节清零), 万一读到带尾随零填充的
// 行, 按 len(rowData) 取尾会取到零填充而不是真正的位图; 按版本总长取则始终精确。
func (table *Table) versionedNullBitmap(rowData []byte, ver byte) ([]byte, bool) {
	sv := table.getVersion(ver)
	if sv == nil {
		return nil, false
	}
	bitmapLen := types.NullBitmapLen(len(sv.FieldIndices))
	rowSize := table.rowDataSizeForVersion(ver)
	if len(rowData) < rowSize {
		return nil, false
	}
	return rowData[rowSize-bitmapLen : rowSize], true
}

// 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 计算当前活跃版本的行数据大小 (含 schemaVer 1B)
func (table *Table) calcRowDataSize() int {
	return table.rowDataSizeForVersion(table.activeVersion)
}

// rowDataSizeForVersion 计算指定 schema 版本的行数据大小
// (含 schemaVer 1B 与该版本字段数对应的 NULL 位图; 位图在行尾)
func (table *Table) rowDataSizeForVersion(ver byte) int {
	size := 1 // schemaVer
	sv := table.getVersion(ver)
	if sv != nil {
		size += types.NullBitmapLen(len(sv.FieldIndices)) // NULL 位图 (行尾)
		for _, idx := range sv.FieldIndices {
			size += int(table.Field[idx].Width())
		}
	}
	return size
}

// 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 落盘。
//
// 惰性演进语义:
//   - 本次要写的字段在旧行自身的版本里都存在 → 保持旧版本。行长不变, 可原地更新,
//     不会因为"只改了一个老字段"而无谓地把行升级并搬迁。
//   - 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), 行级过滤按不匹配处理, 至少不会污染数据。
func (table *Table) compareFieldBytes(row []byte, offset uint16, ct colType, value interface{}) (int, bool) {
	if types.IsDynamic(ct.typ) {
		if int(offset)+types.DynamicRefSize > len(row) {
			return 0, false
		}
		pageID := binary.BigEndian.Uint32(row[offset : offset+4])
		offInPage := binary.BigEndian.Uint16(row[offset+4 : offset+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 || int(offset)+size > len(row) {
		return 0, false
	}
	stored := row[offset : int(offset)+size]

	// 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 比较字段值是否相等
func (table *Table) compareValue(row []byte, offset uint16, ct colType, value interface{}) bool {
	c, ok := table.compareFieldBytes(row, offset, ct, value)
	return ok && c == 0
}

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

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

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

// compareLessOrEqual 比较字段值是否小于等于给定值
func (table *Table) compareLessOrEqual(row []byte, offset uint16, ct colType, value interface{}) bool {
	return table.compareLess(row, offset, ct, value) || table.compareValue(row, offset, 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) error {
	sv := table.getVersion(table.activeVersion)
	probe := make([]types.FieldInfo, 0, len(table.Field)+1)
	if sv != nil {
		for _, idx := range sv.FieldIndices {
			if int(idx) < len(table.Field) {
				probe = append(probe, table.Field[idx])
			}
		}
	}
	probe = append(probe, newField)
	if err := types.CheckRowSize(probe); err != nil {
		return fmt.Errorf("新增列 %q 后行宽超限: %w", newField.FieldName, err)
	}
	return nil
}

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())
		}
		// 复活同样会让行变宽, 先做行宽预演
		if err := table.checkProspectiveRowSize(newField); err != nil {
			return err
		}
		// 只恢复"存活 + 可空/唯一"这些标志位, 低 5 位的列级参数保持原值
		existing.Flag = (existing.Flag & types.FlagScaleMask) | (newField.Flag &^ types.FlagScaleMask)
		table.info.ActiveFieldCount++

		// 找到该字段在 master registry 中的索引
		revivedIdx := -1
		for i := range table.Field {
			if &table.Field[i] == existing {
				revivedIdx = i
				break
			}
		}

		// 构建新版本: 旧版本字段 + 复活的字段索引
		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 {
			return err
		}
		table.syncRowSize()
		table.saveTableInfo()
		return nil
	}

	// 全新字段: 同样先做行宽预演 (加列只会让行更宽)
	if err := table.checkProspectiveRowSize(newField); 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 {
		return err
	}
	table.syncRowSize()
	table.saveTableInfo()
	return nil
}

func (table *Table) saveTableInfo() {
	buf := make([]byte, PageSize)
	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
	}

	// 版本号闸门必须在改动之前 (理由同 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 {
		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_Blob 值的大小上限 (字节)。
//
// 它不是拍出来的数, 而是从存储约束直接推出: 一个堆块必须整体落进一个堆页, 可用
// 载荷 = 页大小 - 页头 - 块长度前缀 (与 heapAllocBlock 的 maxPayload 同源)。
//
// 为什么单独列出来: Blob 是"存文件/大对象"的类型, 使用者必须能在写入前就知道
// 这条线, 而不是写进去才撞上 heapAllocBlock 里那句与"值太大"看不出因果的
// ErrHeapPageTooLarge。写入侧的检查见 writeFieldValueAt。
//
// 它同时是【声明侧】的容量上限 (声明超限的字符串列也落动态列, 容量同样是这个数):
// 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++
	}

	_, off, err := tail.Alloc(total)
	if err != nil {
		return nil, 0, err
	}
	if off > 0xFFFF {
		return nil, 0, ErrHeapPageTooLarge
	}
	table.saveTableInfo()
	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)
}

// releaseOldDynamic 释放旧行在 masterIdx 处持有的堆块 (若有)。
//
// 专供"更新成 NULL"这条路径: 新行不再调用 writeDynamicField, 旧块不会有人摘链,
// 不在这里释放就是每更新一次漏一个堆块。旧值本就是 NULL 时, getVersionedOffset 会因
// 位图置位直接返回 false, 不会误摘。
func (table *Table) releaseOldDynamic(oldRowData []byte, masterIdx int) {
	if len(oldRowData) == 0 {
		return
	}
	oldOff, ok := table.getVersionedOffset(oldRowData, masterIdx)
	if !ok || int(oldOff)+types.DynamicRefSize > len(oldRowData) {
		return
	}
	if pid := binary.BigEndian.Uint32(oldRowData[oldOff : oldOff+4]); pid != 0 {
		_ = table.heapUnref(pid)
	}
}

// heapReleaseRow 释放一行占用的全部堆页引用 (删除行时调用)
// 按行自身的 schema 版本定位各动态字段的 6 字节引用
func (table *Table) heapReleaseRow(rowData []byte) {
	if len(rowData) < 1 {
		return
	}
	sv := table.getVersion(rowData[0])
	if sv == nil {
		return
	}
	bitmap, ok := table.versionedNullBitmap(rowData, rowData[0])
	if !ok {
		return
	}
	offset := uint16(1) // 位图在行尾, 字段区从偏移 1 开始
	for i, idx := range sv.FieldIndices {
		f := table.Field[idx]
		// NULL 字段的引用槽是全零, 下面 pageID != 0 本就能跳过; 显式查位图是为了不把
		// "零引用"误当成可解引用的堆块 (将来若改了零值约定, 这里仍然安全)。
		if !f.IsDropped() && types.IsDynamic(f.FieldType) && !types.NullBitSet(bitmap, i) {
			if int(offset)+types.DynamicRefSize <= len(rowData) {
				pageID := binary.BigEndian.Uint32(rowData[offset : offset+4])
				if pageID != 0 {
					_ = table.heapUnref(pageID)
				}
			}
		}
		offset += f.Width()
	}
}

// writeDynamicField 将变长数据落堆, 并在行内 offset 处写入 6 字节引用
// oldRowData 为旧行 (更新场景), 用于复用已存在的堆块; 插入时传 nil
func (table *Table) writeDynamicField(rowData []byte, offset uint16, masterIdx int, data []byte, oldRowData []byte) error {
	// 旧引用 (更新场景): 既用于原地复用, 也在换块后释放
	var (
		oldPID       uint32
		oldOffInPage uint16
		hasOldRef    bool
	)
	if oldRowData != nil {
		if oldOff, ok := table.getVersionedOffset(oldRowData, masterIdx); ok &&
			int(oldOff)+types.DynamicRefSize <= len(oldRowData) {
			oldPID = binary.BigEndian.Uint32(oldRowData[oldOff : oldOff+4])
			oldOffInPage = binary.BigEndian.Uint16(oldRowData[oldOff+4 : oldOff+6])
			hasOldRef = oldPID != 0
		}
	}

	if len(data) == 0 {
		// 新值为空: 行内引用保持全 0, 旧块引用需要释放
		if hasOldRef {
			_ = table.heapUnref(oldPID)
		}
		return nil
	}

	// 1. 尽量复用旧堆块 (原地覆写), 此时引用不变
	if hasOldRef && table.heapRewrite(oldPID, oldOffInPage, data) {
		binary.BigEndian.PutUint32(rowData[offset:], oldPID)
		binary.BigEndian.PutUint16(rowData[offset+4:], oldOffInPage)
		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 hasOldRef && !(oldPID == pageID && oldOffInPage == off) {
		_ = table.heapUnref(oldPID)
	}
	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
}

// writeFieldValueAt 写入单个字段; 动态字段写入行内 6B 堆引用
//
// 形参是 FieldInfo 而不是 FieldType: 宽度与列级参数都必须取自【同一个列定义】——
// 拆成两个参数传 (类型来自这里、宽度码来自那里) 就会出现"按 256 字节编、按 2048 字节存"
// 这类不一致, 而它表现为整行错位而不是报错。
//
// checkOnly 为真时动态列直接跳过 (行内引用保持全零), 见 serializeRowForCheck: 试算行
// 只用于算索引键, 绝不能有分配/覆写/释放堆块的副作用 —— 它的结果可能被整丢弃。
func (table *Table) writeFieldValueAt(rowData []byte, offset uint16, masterIdx int,
	f types.FieldInfo, value interface{}, oldRowData []byte, checkOnly bool) error {
	ft := f.FieldType
	if types.IsDynamic(ft) {
		if checkOnly {
			return nil
		}
		// 动态列的值只认 []byte。旧写法是 `data, _ := value.([]byte)`: 一旦传进来
		// 的是别的类型 (string / 数值), 断言失败 → data 为 nil → 整个值被静默写成空。
		// 这与"不静默丢内容"直接冲突, 所以这里改为显式报错 (nil 仍表示空值)。
		var data []byte
		if value != nil {
			b, ok := value.([]byte)
			if !ok {
				return fmt.Errorf("字段 %q (%s): 动态列只接受 []byte, 收到 %T",
					table.Field[masterIdx].FieldName, types.TypeNameOf(ft), value)
			}
			data = b
		}
		// 动态列的单块尺寸闸门: 超限在这里就报出列名与上限, 而不是等 heapAllocBlock
		// 抛出与"值太大"因果关系不明的 ErrHeapPageTooLarge。
		//
		// 必须覆盖【所有】动态列, 不能只拦 Blob: Dynamicbyte 与 Blob 的盘上布局相同
		// (行内 6B 引用, 内容整块落进一个堆页), 单值上限也一样。只拦 Blob 的话,
		// 往 Dynamicbyte 写一个超限值只会得到不带列名、不带字节数的
		// "申请空间超过单页最大容量", 用户看不出是哪一列、超了多少。
		if uint32(len(data)) > MaxBlobSize {
			return fmt.Errorf("%w: 字段 %q (%s) 的值 %d 字节超过单块上限 %d 字节, "+
				"更大的内容需要分块存储 (见 types.FieldType_Blob)",
				types.ErrValueRange, table.Field[masterIdx].FieldName,
				types.TypeNameOf(ft), len(data), MaxBlobSize)
		}
		return table.writeDynamicField(rowData, offset, masterIdx, data, oldRowData)
	}
	size := f.Width()
	if size == 0 {
		return fmt.Errorf("%w: 字段 %q (%s) 的列级参数 %d 非法, 宽度不可计算",
			types.ErrValueRange, f.FieldName, types.TypeNameOf(ft), f.Param())
	}
	if int(offset)+int(size) > len(rowData) {
		return fmt.Errorf("%w: 字段 %d 写入越界 (offset=%d, 列宽=%d, 行长=%d)",
			types.ErrEncodeBufSize, masterIdx, offset, size, len(rowData))
	}
	// 编码统一走 types.EncodeValue —— 与索引键字节同源 (见 types/codec.go)。
	// param 取列级类型参数: decimal 必须在这里校验"值的 scale == 列声明的 scale",
	// 否则同一列出现两种 scale 会让"字节序 == 数值序"失效, 索引范围扫描静默漏行;
	// 字符串列则由它决定列宽闸门 (超宽在 EncodeValue 里直接报错, 不再截断)。
	if err := types.EncodeValue(ft, f.Param(), value, rowData[offset:int(offset)+int(size)]); err != nil {
		return fmt.Errorf("字段 %q 写入失败: %w", table.Field[masterIdx].FieldName, err)
	}
	return nil
}

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

// serializeRow 将 FieldValue 切片序列化为行数据字节 (不含 pkID 前缀)
// 行格式: [schemaVer(1B)] [field0_data] [field1_data] ... [NULL 位图]
// 动态字段的行内数据为 6 字节堆引用; oldRowData 用于更新时复用堆块。
//
// 两点关键语义 (否则 schema 演进后更新旧行会静默丢数据):
//   - 版本选择走 pickVersionForUpdate: 未触及新字段时保持旧版本, 行长不变 → 可原地更新
//   - 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)
}

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

	// 写行闸门 (第三处行宽校验)。
	//
	// 前两处在建列侧 (TablePage.Create / Table.AddField), 管的是"新表不许超宽";
	// 这一处管的是"已经存在的老表": 表是在限制生效前建的 (或 schema 被外部改坏) 时,
	// 行字节数会超过一个 DataPage 能装下的量。放任它走下去的后果不是"写失败"这么轻——
	// 行长度会在 DataPage 里溢出到相邻区域, 把整页的行解析全部错位 (页内行是顺序排布、
	// 靠每行 4B 长度前缀自描述的, 一行错位等于整页报废)。
	//
	// 放在这里而不是"逐字段校验"是因为行总长只有这里算得出来 (行宽由 schema 版本决定)。
	// 代价是每行一次整数比较。
	rowSize := table.rowDataSizeForVersion(ver)
	if rowSize > MaxPhysicalRowData {
		return nil, fmt.Errorf("%w: 表 %q 的 schema 版本 %d 行宽 %d 字节超过单页可容纳上限 %d 字节, 拒绝写入; "+
			"请重建该表并把大列改为动态列 (FieldType_Dynamicbyte / Blob)",
			types.ErrValueRange, util.BytesToString(table.info.Name[:]), ver, rowSize, MaxPhysicalRowData)
	}
	rowData := make([]byte, rowSize)
	rowData[0] = ver

	sv := table.getVersion(ver)
	if sv == nil {
		return rowData, nil
	}

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

	// NULL 位图放在行尾 (见 getVersionedOffset 的说明): 字段区起点恒为偏移 1,
	// 时序索引的固定偏移读取器因此无需感知位图。rowData 由 make 分配, 天然全零 →
	// 默认"全非 NULL", 因此只有真正写 NULL 的字段需要置位。
	bitmapLen := types.NullBitmapLen(len(sv.FieldIndices))
	bitmap := rowData[len(rowData)-bitmapLen:]
	offset := uint16(1)
	for i, masterIdx := range sv.FieldIndices {
		f := table.Field[masterIdx]
		name := f.FieldName
		// 取值优先级: 本次传入 > 旧行原值 (未提供的字段沿用旧值, 否则"只改一列"会清空其它列)
		value, provided := lookupFieldValue(values, name)
		if !provided {
			value, provided = lookupFieldValue(base, name)
		}
		if 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, fmt.Errorf("%w: 字段 %q (%s) 未声明可空 (nullable), 不接受 NULL; "+
					"要允许写入 NULL, 请在列声明里加 `nullable` / `null`",
					types.ErrValueRange, name, types.TypeNameOf(f.FieldType))
			}
			// NULL: 位图置位, 字段区留零 (行长固定; 动态列因此不会写入堆引用)。
			// 试算模式也要置位 —— extractIndexKey 正是靠位图判定"该行不进本索引"。
			types.SetNullBit(bitmap, i)
			// 旧行若在该列持有堆块, 新行不再引用它 —— 不在这里释放就是每更新一次漏一块。
			if !checkOnly {
				table.releaseOldDynamic(oldRowData, int(masterIdx))
			}
		} else if provided {
			if err := table.writeFieldValueAt(rowData, offset, int(masterIdx), f, value, oldRowData, checkOnly); err != nil {
				return nil, err
			}
		}
		offset += f.Width()
	}
	return rowData, nil
}

func writeFieldValue(buf []byte, ft types.FieldType, value interface{}) {
	switch ft {
	case types.FieldType_1byte:
		buf[0] = value.(byte)
	case types.FieldType_2byte:
		binary.BigEndian.PutUint16(buf, uint16(value.(int16)))
	case types.FieldType_4byte:
		binary.BigEndian.PutUint32(buf, uint32(value.(int32)))
	case types.FieldType_8byte:
		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 前缀)
// 行格式: [schemaVer(1B)] [field_data...] [NULL 位图] — 按行自身的版本解析
func (table *Table) deserializeRow(rowData []byte) []types.FieldValue {
	if len(rowData) < 1 {
		return nil
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return nil // 未知版本, 无法解析
	}

	// NULL 位图在行尾, 长度按本行自己的版本算。
	bitmap, ok := table.versionedNullBitmap(rowData, ver)
	if !ok {
		return nil
	}

	// 预分配容量: 字段数即结果条数的上界 (被删除 / NULL 字段会跳过), 避免 append 反复扩容
	result := make([]types.FieldValue, 0, len(sv.FieldIndices))
	offset := uint16(1)
	for i, idx := range sv.FieldIndices {
		f := table.Field[idx]
		// 跳过已删除字段 (master registry 中标记 FlagDeleted) 与 NULL 字段。
		// NULL 不出现在结果里 —— 与"零值不 append"是同一约定: 读不到值就是没有值;
		// 更新路径靠它区分"沿用旧值"与"显式写 NULL"(见 serializeRow 的取值优先级)。
		if !f.IsDropped() && !types.NullBitSet(bitmap, i) {
			size := f.Width()
			if int(offset)+int(size) <= len(rowData) {
				var val interface{}
				if types.IsDynamic(f.FieldType) {
					// 解引用堆页
					pageID := binary.BigEndian.Uint32(rowData[offset : offset+4])
					offInPage := binary.BigEndian.Uint16(rowData[offset+4 : offset+6])
					if data, err := table.heapRead(pageID, offInPage); err == nil {
						val = data
					}
				} else {
					val = types.DecodeValue(f.FieldType, rowData[offset:offset+size])
				}
				if val != nil {
					result = append(result, types.FieldValue{FieldName: f.FieldName, Value: val})
				}
			}
		}
		offset += table.Field[idx].Width()
	}
	return result
}

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

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