TsDB2/DB/core/table_rowcodec.go

package core

import (
	"bytes"
	"cmp"
	"encoding/binary"
	"fmt"
	"io"
	"slices"
	"strings"

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

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

// 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, bool, error) {
	switch v := value.(type) {
	case BlobRef: // == types.BlobRef (类型别名), 引用原样透传
		return v, false, nil
	case types.Blob:
		return table.blobRefFromValue(f, v, checkOnly)
	case *types.Blob:
		if v == nil {
			return BlobRef{}, false, nil
		}
		return table.blobRefFromValue(f, *v, checkOnly)
	}

	if checkOnly {
		return BlobRef{}, false, nil
	}

	var src io.Reader
	switch v := value.(type) {
	case []byte:
		if len(v) == 0 {
			return BlobRef{}, false, nil
		}
		src = bytes.NewReader(v)
	case string:
		if v == "" {
			return BlobRef{}, false, nil
		}
		src = strings.NewReader(v)
	case io.Reader:
		src = v
	default:
		return BlobRef{}, false, 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{}, false, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
			f.FieldName, types.TypeNameOf(f.FieldType), err)
	}
	return ref, true, nil
}

// blobRefFromValue 处理 types.Blob 载体: 既有"内容"形态 (Source / Bytes),
// 也有"读回镜像"形态 (FirstPageID / TotalLen 已由引擎填好).
//
// 返回值 owned 标注该引用是否本次新落的链: 调用方在【整行写入失败】时只能还
// 自己新落的链 —— 透传引用 (调用方持有的) 还掉会把别的行/别的表的链误回收。
func (table *Table) blobRefFromValue(f types.FieldInfo, b types.Blob, checkOnly bool) (BlobRef, bool, error) {
	hasRef := b.FirstPageID != 0 || b.TotalLen != 0
	hasContent := b.Source != nil || len(b.Bytes) > 0
	if hasRef {
		if hasContent {
			return BlobRef{}, false, fmt.Errorf("字段 %q: types.Blob 同时带了引用与内容 (Source/Bytes), 无法判定该用哪个", f.FieldName)
		}
		return b.Ref(), false, nil // 读回镜像: 原样透传
	}
	if !hasContent {
		return BlobRef{}, false, nil // 空 blob: 不分配链页
	}
	if checkOnly {
		return BlobRef{}, false, nil
	}
	if b.Source != nil && len(b.Bytes) > 0 {
		return BlobRef{}, false, fmt.Errorf("字段 %q: types.Blob 的 Source 与 Bytes 都非空, 引擎无法选择", f.FieldName)
	}
	if !b.Compress.IsValid() {
		return BlobRef{}, false, 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{}, false, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
			f.FieldName, types.TypeNameOf(f.FieldType), err)
	}
	return ref, true, nil
}

// 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, err error) {
	// 失败回收: blob 列在 Pass 1 就落链 (行内只有 8B 引用), 一旦后续列编码失败 /
	// 布局自检失败, 这批新链页还不掉就成为孤儿页 (无事务写入的既有泄漏路径,
	// 见 blobCommit 的注释)。只还"本次新落的链" (owned) —— 透传引用
	// (调用方持有的 types.BlobRef / 读回镜像) 不属于本行, 绝不能碰。
	// 这些链此刻尚未被任何行/索引引用, 无条件归还安全; 更新路径 Pass 3 已释放的
	// 旧链不会出现在这里 (owned 只记新链)。
	var newBlobRefs []BlobRef
	defer func() {
		if err != nil && !checkOnly {
			for _, r := range newBlobRefs {
				_ = table.BlobFree(r)
			}
		}
	}()

	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)
	}
	// 旧行的 NULL 列集合: deserializeRow 不产出 NULL 列 (跳过), 部分更新时
	// "未提供"的列查 base 会落空、被零值兜底 —— 旧行里的 NULL 必须随位图保留,
	// 否则一次只改 title 的 UPDATE 会把 deleted_at 这类 NULL 列静默写成零值,
	// 软删除作用域 (deleted_at IS NULL) 从此查不到该行 (实测报告 #1/#3 根因)。
	oldNulls := map[int]bool{} // masterIdx → 旧行该列为 NULL
	if len(oldRowData) > 0 {
		if layout, ok := table.parseRowLayout(oldRowData); ok {
			oldN := len(layout.sv.FieldIndices)
			for oi, idx := range layout.sv.FieldIndices {
				if types.RowIsNull(oldRowData, layout.ends, oldN, oi) {
					oldNulls[int(idx)] = true
				}
			}
		}
	}

	// 行布局 (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, owned, err := table.blobFieldRef(f, value, checkOnly)
			if err != nil {
				return nil, err
			}
			if owned {
				newBlobRefs = append(newBlobRefs, ref)
			}
			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 字节)。
			// 例外: 旧行该列本是 NULL — 位图置位、不占内容字节 (与显式 NULL 同布局;
			// 不做非空拒绝: 旧数据可能早于约束收紧, 保留原状比重写/报错安全)。
			if oldNulls[int(masterIdx)] {
				types.SetNullBit(bitmap, i)
				colNull[i] = true
			} else {
				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})
			}
		}
		slices.SortFunc(cands, func(a, b spillCand) int { return cmp.Compare(b.gain, a.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
	}
	layout, ok := table.parseRowLayout(rowData)
	if !ok {
		return nil // 未知版本 / 偏移表不可信, 无法解析
	}
	sv, n, col0Width, ends := layout.sv, len(layout.sv.FieldIndices), layout.col0Width, layout.ends
	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 data, ok := table.derefHeapSeg(seg); ok {
				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
	}
	layout, ok := table.parseRowLayout(rowData)
	if !ok {
		return nil, false
	}
	sv, n, col0Width, ends := layout.sv, len(layout.sv.FieldIndices), layout.col0Width, layout.ends
	// 找到字段在行内的位置 (与 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
	}
	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):
		return table.derefHeapSeg(seg)
	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):
// 解码与编码同源, 新增类型不必再回来改这里。