TsDB2/DB/core/index_meta_page.go

package core

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"slices"
	"unsafe"

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

// 🔥 最终对接版：只存索引定义 + 指向你的 IndexPage
type IndexMeta struct {
	IndexName  [IndexNameSize]byte
	IndexType  page.IndexType
	FieldCnt   byte     // 列的数量
	FieldIDs   [16]byte // 列
	EntryPage  uint32   // ✅ 你的 IndexPage 首页ID
	PrevMetaID uint32   // 上一个
	NextMetaID uint32   // 下一个
	Flag       byte     // 标志位 (见 IndexFlag* 常量)
}

// ===== IndexMeta.Flag 位布局 =====
//
//	bit7            | bit0
//	已退位(软删除)     | 唯一约束
//
// 位 7 的语义由 util.IsHighBitSet 统一判定 (Deleted); 位 0 是本轮新增的唯一约束标记。
const (
	// IndexFlagUnique 该索引是唯一索引: 同一索引键只允许存在一行。
	//
	// 【NULL 免除】复合唯一索引只要有一个分量是 NULL, extractIndexKey 就返回 nil,
	// 该行根本不进这个索引 —— 索引里没有条目, 唯一性自然无从违反。因此
	// "唯一但可空"的列可以存任意多行 NULL, 与 SQL 的 UNIQUE 语义一致。
	IndexFlagUnique byte = 1 << 0
)

// resolveChunk 批量回表的单批行数: 每批"游标走查 + 物化"完成后才触发上层回调,
// 回调里对行存储索引的 Update/Delete 不会踩坏正在推进的游标。
const resolveChunk = 512

// 编译期断言: 钉死 IndexMeta 的盘上大小与字段偏移。
// analyzer/main.go 里那套硬编码偏移 (META_SZ / im[...]) 必须与之一致,
// 因此改布局时这里会先编译失败, 而不是让分析工具静默读错。
//
// 布局 (IndexNameSize=64) = 100 字节:
//
//	IndexName[64] + IndexType(1) + FieldCnt(1) + FieldIDs[16]
//	+ pad(2) + EntryPage(4) + PrevMetaID(4) + NextMetaID(4) + Flag(1) + pad(3)
var (
	_ [unsafe.Sizeof(IndexMeta{}) - 100]struct{}
	_ [100 - unsafe.Sizeof(IndexMeta{})]struct{}
	_ [64]byte = [unsafe.Offsetof(IndexMeta{}.IndexType)]byte{}
	_ [65]byte = [unsafe.Offsetof(IndexMeta{}.FieldCnt)]byte{}
	_ [66]byte = [unsafe.Offsetof(IndexMeta{}.FieldIDs)]byte{}
	_ [84]byte = [unsafe.Offsetof(IndexMeta{}.EntryPage)]byte{}
	_ [96]byte = [unsafe.Offsetof(IndexMeta{}.Flag)]byte{}
)

type IndexMetaPage struct {
	hdr   page.PageHdr
	metas []IndexMeta

	handle     page.HandleDb
	hdrSize    uint32
	metaSize   uint32
	lastPageID uint32
	rowCache   []index.DPRange // 行存储索引回表定位缓存 (由处理器复用, 变更时失效)

	// active 活跃索引缓存 (元信息 + 已查表的处理器), 惰性构建。
	// 热路径 (InsertRow/DeleteByKey/UpdateByKey/ResolveByPK) 每行都要遍历索引,
	// 缓存后省去每行每索引的 Flag 判断与 handler 查表; 索引增删/重载时失效。
	active         []resolvedIndex
	activeResolved bool
	rowIdx         int // active 中行存储索引的下标, -1 表示无
}

// resolvedIndex 已解析的活跃索引: 元信息 + 已查表的处理器
type resolvedIndex struct {
	meta     *IndexMeta
	handler  index.IndexHandler
	rowStore index.RowStoreHandler // 非 nil = 承载行数据 (主键/时序), 断言结果随缓存一份
}

// 初始化
func (im *IndexMetaPage) Init(pageID uint32) {
	im.hdrSize = uint32(unsafe.Sizeof(page.PageHdr{}))
	im.metaSize = uint32(unsafe.Sizeof(IndexMeta{}))
	im.hdr.PageID = pageID
	im.hdr.Type = page.IndexMetaIndex
	im.hdr.NextPage = 0
	im.lastPageID = pageID
	im.metas = im.loadMetas(pageID)
	im.invalidateIndexCache()
}

// InitEmpty 把本页初始化成"空的索引元数据页", 专供「刚 NewPage 出来的新页」使用。
//
// 为什么不能直接用 Init —— 调用顺序错了就是 DROP TABLE 假死 52s 的根源:
//   - Init 是"打开一张已有页": 它用 loadMetas 从盘上回读本页内容来重建 metas。
//     而新页是从空闲链摘出来的, 回收链 (WAL 回收 / Drop 回收) 都只做 O(1) 拼接、
//     不擦除页内容, 盘上还留着它上一世的东西 —— 典型是 WAL 页 (Type=7、
//     UseSize=16384、NextPage 还指向旧 WAL 链)。此时 Init 会把整页旧内容当作
//     IndexMeta 数组解析, 还顺着旧链多解析好几页: 实测建 G 表时解出 91 条垃圾元数据
//     (IndexType=66/71/76…, EntryPage=16777216)。随后即便用 InitHeader 覆盖成干净页,
//     内存里的垃圾 meta 也不会消失, 最终被 DropTable 拿去枚举页 → FreePage 落到从未
//     分配过的页号上 → 写到几十 GB 偏移, DROP 假死。
//   - InitEmpty 不读盘: 先写一张只含页头的干净页覆盖旧内容, 再把内存状态置空。
func (im *IndexMetaPage) InitEmpty(pageID uint32) {
	im.hdrSize = uint32(unsafe.Sizeof(page.PageHdr{}))
	im.metaSize = uint32(unsafe.Sizeof(IndexMeta{}))
	im.hdr.Type = page.IndexMetaIndex
	im.hdr.PageID = pageID
	im.hdr.NextPage = 0
	im.hdr.UseSize = im.hdrSize
	im.lastPageID = pageID
	im.InitHeader(pageID) // 干净页覆盖回收页里的旧内容
	im.metas = nil
	im.invalidateIndexCache()
}

// loadMetas 解析索引元数据链 (Init / Reload 共用的读取部分)
func (im *IndexMetaPage) loadMetas(pageID uint32) []IndexMeta {
	metas := make([]IndexMeta, 0)
	for pageID != 0 {
		im.lastPageID = pageID
		buf := make([]byte, PageSize)
		if err := im.handle.ReadPage(pageID, buf); err != nil {
			break
		}
		hdr := page.HdrOf(buf)
		// 页类型 / 自述页号校验: 读索引元数据链绝不能"盘上有多少就解析多少"。
		// 新页可能刚从空闲链摘出来, 盘上留着上一世的内容 (典型: WAL 页 Type=7、
		// UseSize=16384、NextPage 指向旧 WAL 链)。不校验就会把整页旧内容按 IndexMeta
		// 数组解析, 还顺着旧链一路解析下去 —— 实测建表时解出 91 条垃圾元数据
		// (EntryPage=16777216), 最终让 DropTable 去 FreePage 从未分配过的页号:
		// 写到几十 GB 偏移、DROP 假死 52s、稀疏文件暴涨到 40GB。
		if hdr.Type != page.IndexMetaIndex || hdr.PageID != pageID {
			break
		}
		offset := im.hdrSize

		for offset+im.metaSize <= hdr.UseSize {
			m := (*IndexMeta)(unsafe.Pointer(&buf[offset]))
			if m.Flag&0x80 == 0 {
				metas = append(metas, *m)
			}
			offset += im.metaSize
		}
		pageID = hdr.NextPage
	}
	return metas
}

// Reload 从磁盘重新读回索引元数据, 原地覆盖 im.metas —— 供事务 Rollback 使用。
//
// metas 是 Init 时从页里**拷贝**到 Go 堆的独立副本, 丢弃页缓存并不会刷新它。
// 不重载的话 EntryPage 仍停留在事务中的值 (指向事务里新建的 B+ 树根, 删空时还会被
// 写成 0), 而那个根页已随缓存一起被丢弃 → 回滚后所有索引查找都走空。
//
// firstPageID 必须由调用方从"刚从磁盘读回的 TableInfo.FirstIndexID"传入:
// im.hdr.PageID 同样是内存副本, 事务内首次建索引时可能已被改写。
func (im *IndexMetaPage) Reload(firstPageID uint32) {
	im.hdr.PageID = firstPageID
	im.hdr.NextPage = 0
	im.lastPageID = firstPageID
	im.rowCache = nil // 回表定位缓存指向事务中的 DataPage, 必须失效

	metas := im.loadMetas(firstPageID)
	if len(metas) == len(im.metas) {
		copy(im.metas, metas) // 长度一致时原地覆盖, 外部持有的 &metas[i] 继续有效
	} else {
		im.metas = metas
	}
	im.invalidateIndexCache() // 元信息已变, 缓存里的 &metas[i] 与 handler 必须重算
}

// PageID 返回索引元数据页 ID (调试用)
func (im *IndexMetaPage) PageID() uint32 { return im.hdr.PageID }

// HdrSize 返回页头大小 (调试用)
func (im *IndexMetaPage) HdrSize() uint32 { return im.hdrSize }

// 创建索引（纯元数据，不碰IndexPage）
//
// unique 是变参 (只取第一个值): 这样 `CreateIndex(name, typ, fields)` 的既有调用
// 一处都不用改, 需要唯一约束的调用方写成 `CreateIndex(name, typ, fields, true)` 即可。
func (im *IndexMetaPage) CreateIndex(indexName string, itype page.IndexType, fieldIDs []byte, unique ...bool) (*IndexMeta, error) {
	uniqueIdx := len(unique) > 0 && unique[0]

	// 名字超长必须显式报错。下面用 copy 写入定长数组, 超长会被静默截断: 存进盘里的
	// 名字与调用方用来查询的名字不一致, 索引建了却永远查不到 —— 必须拦在建的这一步。
	// (自动命名 pk_<表名> / idx_<表名>_<列名> 会额外多出几个字节, 所以名字接近上限时
	//  这里就会报错, 而不是默默建出一个查不到的索引。
	//  上限取 IndexNameSize 而非 TableNameSize —— 索引名与表名是两套独立额度。)
	if len(indexName) > IndexNameSize {
		return nil, fmt.Errorf("%w: 索引名 %q 有 %d 字节, 上限 %d",
			ErrIndexNameTooLong, indexName, len(indexName), IndexNameSize)
	}
	// 分量个数闸门。先按 int 判再转 byte: 直接 `byte(len(fieldIDs))` 会在超过 255 时回绕
	// (例如 257 个份量截成 1), 于是越界的调用反而"通过"了, 之后按错误的分量数建条目。
	if len(fieldIDs) == 0 || len(fieldIDs) > MaxIndexFields {
		return nil, fmt.Errorf("%w (本次 %d 个分量, 索引 %q)",
			ErrIndexFieldCount, len(fieldIDs), indexName)
	}
	fc := byte(len(fieldIDs))

	// 复合键要求各分量互不相同。重复列会让 extractIndexKey 永远取不齐分量
	// (某一分量无人填充), 索引静默变成空索引 —— 写入成功却查不到, 因此直接拒绝。
	var seen [256]bool
	for _, fid := range fieldIDs {
		if seen[fid] {
			return nil, ErrIndexFieldDuplicated
		}
		seen[fid] = true
	}

	// 行存储索引 (主键 B+ 树 / 时序) 的键同时就是"行在哪里"的定位键,
	// 复合键会让回表语义失效 (ResolveByPK 用 pkID, 与键是两回事), 明确拒绝。
	if fc > 1 {
		if h, ok := index.Compose.GetHandler(itype); ok && handlerStoresRows(h) {
			return nil, ErrIndexCompositeRowStore
		}
	}

	// 行存储索引不支持唯一约束, 明确拒绝而不是静默忽略:
	//   - 主键索引本就唯一 (InsertRowWithID 用 ErrDuplicatePK 兜着), 唯一标记是废话;
	//   - 时序索引的多行同刻是【刻意允许】的 (同一时刻的多个测点), 给它加唯一约束
	//     等于把时序表写死 —— 而"忽略这个标记"会让调用方以为约束生效了,
	//     之后插入同刻第二行成功时才发觉不对, 那时数据已经写进去了。
	if uniqueIdx {
		if h, ok := index.Compose.GetHandler(itype); ok && handlerStoresRows(h) {
			return nil, fmt.Errorf("%w (索引 %q)", ErrUniqueRowStore, indexName)
		}
	}

	// 一张表只能有一个行存储索引。行数据 ([pkID + rowData]) 只可能落在其中一个里, 而
	// resolveIndexes 取的是"第一个实现 RowStoreHandler 的索引"; 再建一个不会立刻出错, 只会得到一个
	// "元数据说自己是行存储、却一行都没有"的索引 —— 之后所有按行存储解读它的路径
	// (回表、删除、重建) 都会错乱, 而且错得没有来路。所以拦在建的这一步。
	//
	// 主键 (B+ 树) 与时序都属于行存储, 因此"建了主键表再补时序索引"同样被拒 —— 这不是
	// 额外限制, 而是引擎的真实约束: 一份行数据不可能同时按 pkID 和按时间分段存放。
	if h, ok := index.Compose.GetHandler(itype); ok && handlerStoresRows(h) {
		for i := range im.metas {
			m := &im.metas[i]
			if m.Flag&0x80 != 0 { // 已退位的元数据不占位
				continue
			}
			if indexStoresRows(m) {
				return nil, fmt.Errorf("%w: 表上已有行存储索引 %q (类型 %d), 不能再建 %q (类型 %d)",
					ErrRowStoreExists, util.BytesToString(m.IndexName[:]), m.IndexType, indexName, itype)
			}
		}
	}

	for _, m := range im.metas {
		if util.BytesToString(m.IndexName[:]) == indexName && m.Flag&0x80 == 0 {
			return nil, ErrIndexExists
		}
	}
	var m IndexMeta
	copy(m.IndexName[:], indexName)
	m.IndexType = itype
	m.FieldCnt = fc
	copy(m.FieldIDs[:fc], fieldIDs)
	m.Flag = 0
	m.SetUnique(uniqueIdx) // bit7 是软删除 (Deleted), 新索引一律不带; 唯一位按参数置
	m.EntryPage = 0

	var buf []byte
	var hdr *page.PageHdr

	if im.hdr.PageID == 0 {
		pid := im.handle.NewPage()
		buf, hdr = im.InitHeader(pid)
		im.hdr.PageID = pid
		im.lastPageID = pid
	} else {
		buf = make([]byte, PageSize)
		im.handle.ReadPage(im.lastPageID, buf)
		hdr = page.HdrOf(buf)
	}

	offset := hdr.UseSize
	if offset+im.metaSize > PageSize {
		newID := im.handle.NewPage()
		newBuf, newHdr := im.InitHeader(newID)
		hdr.NextPage = newID
		im.handle.SavePage(im.lastPageID, buf) // 保存下当前页 不然就被新开的页替换了
		im.lastPageID = newID
		buf = newBuf
		hdr = newHdr
		offset = im.hdrSize
	}

	dst := (*IndexMeta)(unsafe.Pointer(&buf[offset]))
	*dst = m
	hdr.UseSize = offset + im.metaSize
	im.handle.SavePage(im.lastPageID, buf)
	im.metas = append(im.metas, m)
	im.invalidateIndexCache() // metas 扩容后旧缓存里的 &metas[i] 会失效
	// 必须返回 im.metas 里的那一份, 而不是局部变量 m 的地址:
	// 调用方 (Table.CreateIndex → backfillIndex) 会在建完树后回写 EntryPage。
	// 若返回 &m, 回写只落在局部副本 + 磁盘页上, 而查询走的是 im.metas ——
	// 于是"回填了存量行却查不到", 直到关库重开 (metas 从盘上重载) 才恢复正常。
	// 回填过程中不会 append 新 meta, 故该指针在本次调用内保持有效。
	return &im.metas[len(im.metas)-1], nil
}

// 查找单个索引
func (im *IndexMetaPage) FindIndex(idxName string) (*IndexMeta, bool) {
	for i := range im.metas {
		m := &im.metas[i]
		if util.BytesToString(m.IndexName[:]) == idxName && m.Flag&0x80 == 0 {
			return m, true
		}
	}
	return nil, false
}

// 删除索引
func (im *IndexMetaPage) DropIndex(idxName string) error {
	cur := im.hdr.PageID
	for cur != 0 {
		buf := make([]byte, PageSize)
		im.handle.ReadPage(cur, buf)
		hdr := page.HdrOf(buf)
		// 同 loadMetas: 页类型/自述页号对不上就不是索引元数据页, 不能在这上面按名字改 Flag
		if hdr.Type != page.IndexMetaIndex || hdr.PageID != cur {
			break
		}
		offset := im.hdrSize

		for offset+im.metaSize <= hdr.UseSize {
			m := (*IndexMeta)(unsafe.Pointer(&buf[offset]))
			if util.BytesToString(m.IndexName[:]) == idxName && m.Flag&0x80 == 0 {
				m.Flag |= 0x80
				im.handle.SavePage(cur, buf)
				for i := range im.metas {
					if util.BytesToString(im.metas[i].IndexName[:]) == idxName {
						im.metas = append(im.metas[:i], im.metas[i+1:]...)
						im.invalidateIndexCache() // metas 已删除元素, 旧缓存 (含下标) 失效
						break
					}
				}
				return nil
			}
			offset += im.metaSize
		}
		cur = hdr.NextPage
	}
	return ErrIndexNotFound
}

// RenameIndex 重命名索引 (只改 IndexMeta.IndexName 字符串, 零索引页迁移)。
func (im *IndexMetaPage) RenameIndex(oldName, newName string) error {
	if len(newName) > IndexNameSize {
		return fmt.Errorf("新索引名 %q 太长: %d > %d", newName, len(newName), IndexNameSize)
	}
	// 新名不能和别的活索引重名
	for _, m := range im.metas {
		if util.BytesToString(m.IndexName[:]) == newName && m.Flag&0x80 == 0 {
			return fmt.Errorf("索引 %q 已存在", newName)
		}
	}

	// 盘上改
	cur := im.hdr.PageID
	for cur != 0 {
		buf := make([]byte, PageSize)
		im.handle.ReadPage(cur, buf)
		hdr := page.HdrOf(buf)
		if hdr.Type != page.IndexMetaIndex || hdr.PageID != cur {
			break
		}
		offset := im.hdrSize
		for offset+im.metaSize <= hdr.UseSize {
			m := (*IndexMeta)(unsafe.Pointer(&buf[offset]))
			if util.BytesToString(m.IndexName[:]) == oldName && m.Flag&0x80 == 0 {
				for i := range m.IndexName {
					m.IndexName[i] = 0
				}
				copy(m.IndexName[:], newName)
				if err := im.handle.SavePage(cur, buf); err != nil {
					return err
				}
				// 内存态同步
				for i := range im.metas {
					if util.BytesToString(im.metas[i].IndexName[:]) == oldName {
						for j := range im.metas[i].IndexName {
							im.metas[i].IndexName[j] = 0
						}
						copy(im.metas[i].IndexName[:], newName)
						break
					}
				}
				im.invalidateIndexCache()
				return nil
			}
			offset += im.metaSize
		}
		cur = hdr.NextPage
	}
	return ErrIndexNotFound
}

func (im *IndexMetaPage) InitHeader(pageID uint32) ([]byte, *page.PageHdr) {
	buf := make([]byte, PageSize)
	hdr := page.HdrOf(buf)
	hdr.Type = page.IndexMetaIndex
	hdr.PageID = pageID
	hdr.UseSize = uint32(unsafe.Sizeof(*hdr))
	im.handle.SavePage(pageID, buf)
	return buf, hdr
}

// ======================== 高层索引操作 ========================

// rowStoreMeta 返回承载行数据的一级索引 (主键/时序)。
// 本层不认识索引类型枚举: 谁是"行存储"由处理器是否实现 RowStoreHandler 回答。
func (im *IndexMetaPage) rowStoreMeta() (*IndexMeta, index.RowStoreHandler, bool) {
	im.resolveIndexes()
	if im.rowIdx < 0 {
		return nil, nil, false
	}
	ri := im.active[im.rowIdx]
	return ri.meta, ri.rowStore, true
}

// Name 索引名 (去掉定长补零)。
func (m *IndexMeta) Name() string {
	if m == nil {
		return ""
	}
	return util.BytesToString(m.IndexName[:])
}

// Fields 索引分量在表 master registry 中的字段下标 (按建索引时给的顺序)。
// 返回拷贝: 调用方不该拿到会随元数据增删而失效的内存。
func (m *IndexMeta) Fields() []byte {
	if m == nil || m.FieldCnt == 0 {
		return nil
	}
	n := int(m.FieldCnt)
	if n > len(m.FieldIDs) {
		n = len(m.FieldIDs)
	}
	out := make([]byte, n)
	copy(out, m.FieldIDs[:n])
	return out
}

// Deleted 索引是否已被软删除 (Flag 最高位)。
func (m *IndexMeta) Deleted() bool {
	return m != nil && util.IsHighBitSet(m.Flag)
}

// Unique 该索引是否为唯一索引 (同一索引键只允许一行; NULL 分量不进索引, 不参与约束)。
//
// 这是唯一约束的【权威标记】: 列上的 types.FlagUnique 只是声明, 约束由索引在
// 写入路径上执行 (插入/更新前先按键探一次), 所以判定必须以索引元信息为准。
func (m *IndexMeta) Unique() bool {
	return m != nil && m.Flag&IndexFlagUnique != 0
}

// SetUnique 就地设置/取消唯一标记 (仅改内存元信息, 落盘由 saveMeta 负责)。
func (m *IndexMeta) SetUnique(v bool) {
	if m == nil {
		return
	}
	if v {
		m.Flag |= IndexFlagUnique
	} else {
		m.Flag &^= IndexFlagUnique
	}
}

// SetName 就地设置索引名 (仅改内存元信息, 落盘由 saveMeta 负责)。
func (m *IndexMeta) SetName(name string) error {
	if m == nil {
		return fmt.Errorf("nil IndexMeta")
	}
	if len(name) > IndexNameSize {
		return fmt.Errorf("索引名 %q 太长: %d > %d", name, len(name), IndexNameSize)
	}
	for i := range m.IndexName {
		m.IndexName[i] = 0
	}
	copy(m.IndexName[:], name)
	return nil
}

// GetIndexes 返回本表全部可用索引的元数据拷贝 (按创建顺序, 已删除的跳过)。
//
// 元数据是迁移/导出类工具的入口: 在此之前 core 包外没有任何办法枚举一张表的索引
// (metas 未导出, 只能逐名 FindIndex —— 那要求调用方事先知道所有索引名)。
// 返回拷贝而不是内部指针: metas 会随 CreateIndex/DropIndex 重新分配,
// 外部持有 &metas[i] 只会在下一次建/删索引时变成野指针。
func (im *IndexMetaPage) GetIndexes() []IndexMeta {
	im.resolveIndexes()
	out := make([]IndexMeta, 0, len(im.active))
	for _, ri := range im.active {
		out = append(out, *ri.meta)
	}
	return out
}

// GetRowStoreIndex 返回承载行数据的一级索引 (VRTX 的主键索引 / 时序索引) 的元数据拷贝。
// 第二个返回值为 false 表示这张表没有行存储索引 —— 此时行数据无处安放,
// 调用方应当报错而不是"照常建表".
func (im *IndexMetaPage) GetRowStoreIndex() (IndexMeta, bool) {
	m, _, ok := im.rowStoreMeta()
	if !ok {
		return IndexMeta{}, false
	}
	return *m, true
}

// resolveIndexes 惰性构建并返回活跃索引缓存 (跳过已删除/无处理器的索引)。
//
// 缓存的 &metas[i] 指针在 metas 发生增删或整体替换后会失效, 因此 CreateIndex /
// DropIndex / Reload / Init 都必须调用 invalidateIndexCache。
func (im *IndexMetaPage) resolveIndexes() []resolvedIndex {
	if im.activeResolved {
		return im.active
	}
	im.active = im.active[:0]
	im.rowIdx = -1
	for i := range im.metas {
		m := &im.metas[i]
		if m.Flag&0x80 != 0 {
			continue
		}
		h, ok := index.Compose.GetHandler(m.IndexType)
		if !ok {
			continue
		}
		rs, _ := h.(index.RowStoreHandler)
		if im.rowIdx < 0 && rs != nil {
			im.rowIdx = len(im.active)
		}
		im.active = append(im.active, resolvedIndex{meta: m, handler: h, rowStore: rs})
	}
	im.activeResolved = true
	return im.active
}

// invalidateIndexCache 使活跃索引缓存失效 (元信息增删/重载后必须调用)。
func (im *IndexMetaPage) invalidateIndexCache() {
	im.activeResolved = false
	im.active = im.active[:0]
	im.rowIdx = -1
}

// indexStoresRows 询问索引处理器: 该索引是否承载行数据。
// 实现 index.RowStoreHandler 即承载; 调用方据此判断"行存在哪里",
// 不需要认识具体索引类型 (主键/时序)。
func indexStoresRows(m *IndexMeta) bool {
	if m == nil {
		return false
	}
	h, ok := index.Compose.GetHandler(m.IndexType)
	if !ok {
		return false
	}
	return handlerStoresRows(h)
}

// handlerStoresRows 该处理器是否承载行数据 (实现 index.RowStoreHandler)。
func handlerStoresRows(h index.IndexHandler) bool {
	_, ok := h.(index.RowStoreHandler)
	return ok
}

// resolver 提供给索引处理器的回表函数: 二级索引内部据此"去找一级索引"取整行,
// 上层无需参与, 也无需知道哪个索引才是主键。
func (im *IndexMetaPage) resolver() index.RowResolver {
	return func(pkID uint64) ([]byte, bool) { return im.ResolveByPK(pkID) }
}

// findIndexMetaByName 按索引名查找活跃索引的元信息 (跳过已删除)。
func (im *IndexMetaPage) findIndexMetaByName(name string) *IndexMeta {
	for i := range im.metas {
		if util.BytesToString(im.metas[i].IndexName[:]) == name && im.metas[i].Flag&0x80 == 0 {
			return &im.metas[i]
		}
	}
	return nil
}

func (im *IndexMetaPage) saveMeta(m *IndexMeta) {
	cur := im.hdr.PageID
	for cur != 0 {
		buf := make([]byte, PageSize)
		im.handle.ReadPage(cur, buf)
		hdr := page.HdrOf(buf)
		// 同 loadMetas: 页类型/自述页号对不上就不是索引元数据页, 不能在这上面按名字覆盖元数据
		if hdr.Type != page.IndexMetaIndex || hdr.PageID != cur {
			break
		}
		offset := im.hdrSize
		for offset+im.metaSize <= hdr.UseSize {
			dst := (*IndexMeta)(unsafe.Pointer(&buf[offset]))
			if util.BytesToString(dst.IndexName[:]) == util.BytesToString(m.IndexName[:]) {
				*dst = *m
				im.handle.SavePage(cur, buf)
				return
			}
			offset += im.metaSize
		}
		cur = hdr.NextPage
	}
}

// rowStoreInsertKey 行存储索引的定位键:
//
//   - 主键 B+ 树: 恒为 pkID(BE)。查询/删除/更新侧一律按 pkID 定位, 与镜像列
//     的存储编码无关 —— 旧实现统一走 extractIndexKey (取镜像列字节), 镜像列是
//     Int64 等带 sort-key 编码的类型时, 树键与 pkID-BE 不一致: 按 pkID 查找
//     退化成全叶扫描, 删除/更新按错键静默失效 (ErrNotFound 按索引逐个吞掉)。
//     Uint64 镜像因"存储字节==BE"而侥幸正确, 掩盖了这条错位。
//   - 时序索引: 键 = 时间分量 (extractIndexKey); 其顺序断言/分段路由都在同一
//     编码内自洽 (sort-key 保序), 维持原样。
//
// 插入/删除/更新三侧必须用同一键源, 否则写进树里的键和去树里找的键对不上。
func rowStoreInsertKey(meta *IndexMeta, pkID uint64, rowData []byte, fields []types.FieldInfo, versions []SchemaVersion) []byte {
	if meta.IndexType == page.IndexTypePrimary {
		key := make([]byte, 8)
		binary.BigEndian.PutUint64(key, pkID)
		return key
	}
	return extractIndexKey(rowData, meta, fields, versions)
}

// extractIndexKey 从行数据中提取索引键 (版本感知)
// rowData 格式 (v9): [schemaVer 1B][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图]
//
// 复合索引的键 = 按 FieldIDs 顺序, 把各分量字段的原始字节【补齐到档位宽度】后定长拼接。
//   - 顺序取 FieldIDs 而非行内物理顺序 —— 查询侧据此才能做"列前缀"范围推导:
//     (a, b) 上的索引既能服务 a=?, 也能服务 a=? AND b>?;
//   - 每个分量长度 = FieldSize(字段类型, 列级参数), 与查询侧编键完全一致, 因此拼接键是
//     定长的, 字节序比较等价于"先比 a 再比 b"的字典序。宽度必须带列级参数 (字符串列的
//     宽度码): 按类型默认档切键会把宽字符串列切成半截, 后面那个分量整体读错位;
//   - v9 起行内字符串列按【实际长度】存储, 取出的字节要 PadToWidth 补零到档位宽度 ——
//     索引键必须保持定长, 二级索引布局 [key][PKID8] 与 idxKeyLenOf 才不会变形;
//   - 任一分量在该行版本里取不到 (列被 Drop / schema 版本不匹配) → 返回 nil,
//     该索引跳过这一行, 与单列索引"取不到键就跳过"的既有语义一致。
func extractIndexKey(rowData []byte, meta *IndexMeta, fields []types.FieldInfo, versions []SchemaVersion) []byte {
	if meta == nil || meta.FieldCnt == 0 || len(rowData) < 1 {
		return nil
	}
	n := int(meta.FieldCnt)
	if n > len(meta.FieldIDs) {
		return nil
	}

	// 从 rowData 读取该行的 schema 版本; 列位置由行内偏移表给出 —— v9 字符串列变长,
	// 旧实现"从偏移 1 起累加档位宽度"必然错位。
	layout, ok := parseRowLayoutOf(fields, versions, rowData, rowData[0])
	if !ok {
		return nil
	}
	sv := layout.sv
	rowN := len(sv.FieldIndices)
	if rowN == 0 {
		return nil
	}
	col0Width := layout.col0Width
	if col0Width <= 0 {
		return nil
	}
	ends := layout.ends

	// seg[k] 指向拼键用的定长字节切片 (补齐后的副本或原位切片), 全部取齐后再一次性拼键。
	var (
		seg  [16][]byte
		need = n
	)
	for i, idx := range sv.FieldIndices {
		fi := int(idx)
		if fi >= len(fields) {
			return nil
		}
		size := int(fields[fi].Width())
		if size == 0 { // 未知类型/非法列参数: 切不出定长分量, 拼出的键会错位
			return nil
		}
		for k := 0; k < n; k++ {
			if int(meta.FieldIDs[k]) != fi {
				continue
			}
			// 分量是 NULL → 该行不进本索引 (键里放不下"缺失"这一档, 上层按 nil 键跳过)。
			if types.RowIsNull(rowData, ends, rowN, i) {
				return nil
			}
			start := int(types.RowFieldStart(ends, i))
			end := int(types.RowFieldEnd(ends, i, col0Width))
			if end < start || end > len(rowData) {
				return nil
			}
			// 行内存的是实际长度, 索引键要的是档位定长: 补零对齐, 与查询侧 EncodeValue 同源。
			seg[k] = types.PadToWidth(rowData[start:end], uint16(size))
			need--
			break
		}
		if need == 0 {
			break // 分量已取齐, 后面的字段与本索引无关
		}
	}
	if need != 0 {
		return nil // 有分量不在该版本中 (如该列已被 Drop), 无法构成完整键
	}

	total := 0
	for k := 0; k < n; k++ {
		total += len(seg[k])
	}
	key := make([]byte, 0, total)
	for k := 0; k < n; k++ {
		key = append(key, seg[k]...)
	}
	return key
}

// InsertRow 插入行到所有索引
func (im *IndexMetaPage) InsertRow(fields []types.FieldInfo, pkID uint64, rowData []byte, activeVer byte, versions []SchemaVersion) error {
	im.rowCache = nil // 插入可能创建新 DataPage, 失效缓存
	indexes := im.resolveIndexes()

	// 唯一约束预检必须先于任何写入: 逐个索引边查边写的话, 撞车时前面几个索引
	// (含承载行数据的行存储索引) 已经写进去了, 只剩一个错误返回 —— 库里多出一行
	// 半成品, 而且它的行还在、二级条目不全, 之后按任何二级索引都查不到它。
	if err := im.checkUniqueForKey(indexes, rowData, pkID, fields, activeVer, versions); err != nil {
		return err
	}

	for _, ri := range indexes {
		m := ri.meta
		indexKey := extractIndexKey(rowData, m, fields, versions)
		if indexStoresRows(m) {
			indexKey = rowStoreInsertKey(m, pkID, rowData, fields, versions)
		}
		if indexKey == nil {
			continue
		}
		entryPage := m.EntryPage
		_, err := ri.handler.Insert(im.handle, &entryPage, indexKey, pkID, rowData, len(indexKey))
		if err != nil {
			return err
		}
		if entryPage != m.EntryPage {
			m.EntryPage = entryPage
			im.saveMeta(m)
		}
	}
	return nil
}

// InsertRows 批量插入: pkIDs[i] 与 rowDatas[i] 一一对应。
//
// 与单行 InsertRow 的语义差异只在"摊销", 不在"语义":
//   - 唯一约束仍是"先全部预检、再动任何写入"—— 每行对既有树的查重照做,
//     另补"批内查重"(单行路径里批内冲突会在第 i+1 行的树查重中被发现, 批量
//     路径一次性排序比出来, 语义等价且更便宜)。
//   - 实现了 BatchInserter 的索引 (主键 B+ 树 / 二级 B+ 树 / 复合索引) 走批量路径;
//     其余 (时序 / hash) 退化为逐条 Insert, 行为与单行路径逐字节一致。
//
// 失败语义与单行路径相同: 中途报错时前面已写入的索引不回滚 (调用方靠事务回滚)。
func (im *IndexMetaPage) InsertRows(fields []types.FieldInfo, pkIDs []uint64, rowDatas [][]byte, activeVer byte, versions []SchemaVersion) error {
	if len(rowDatas) == 0 {
		return nil
	}
	im.rowCache = nil // 批量插入必然新建 DataPage, 一次性失效缓存
	indexes := im.resolveIndexes()

	// 唯一约束预检: 全部查完才动第一笔写入 (与单行路径同一铁律)
	hasUnique := false
	for _, ri := range indexes {
		if ri.meta.Unique() {
			hasUnique = true
			break
		}
	}
	if hasUnique {
		for i := range rowDatas {
			if err := im.checkUniqueForKey(indexes, rowDatas[i], pkIDs[i], fields, activeVer, versions); err != nil {
				return err
			}
		}
		// 批内查重: 同一唯一索引里两行同键
		for _, ri := range indexes {
			if !ri.meta.Unique() {
				continue
			}
			keys := make([][]byte, 0, len(rowDatas))
			for i := range rowDatas {
				if k := extractIndexKey(rowDatas[i], ri.meta, fields, versions); k != nil {
					keys = append(keys, k)
				}
			}
			slices.SortFunc(keys, bytes.Compare)
			for j := 1; j < len(keys); j++ {
				if bytes.Equal(keys[j-1], keys[j]) {
					return fmt.Errorf("%w: 索引 %q, 批内重复键",
						ErrDuplicatedKey, util.BytesToString(ri.meta.IndexName[:]))
				}
			}
		}
	}

	for _, ri := range indexes {
		m := ri.meta
		entryPage := m.EntryPage
		var err error
		if bi, ok := ri.handler.(index.BatchInserter); ok {
			rows := make([]index.BatchRow, len(rowDatas))
			for i := range rowDatas {
				var ik []byte
				if indexStoresRows(m) {
					ik = rowStoreInsertKey(m, pkIDs[i], rowDatas[i], fields, versions)
				} else {
					ik = extractIndexKey(rowDatas[i], m, fields, versions)
				}
				rows[i] = index.BatchRow{
					PKID:     pkIDs[i],
					IndexKey: ik,
					RowData:  rowDatas[i],
				}
			}
			err = bi.InsertBatch(im.handle, &entryPage, rows)
		} else {
			for i := range rowDatas {
				indexKey := extractIndexKey(rowDatas[i], m, fields, versions)
				if indexStoresRows(m) {
					indexKey = rowStoreInsertKey(m, pkIDs[i], rowDatas[i], fields, versions)
				}
				if indexKey == nil {
					continue
				}
				if _, err = ri.handler.Insert(im.handle, &entryPage, indexKey, pkIDs[i], rowDatas[i], len(indexKey)); err != nil {
					break
				}
			}
		}
		if err != nil {
			return err
		}
		if entryPage != m.EntryPage {
			m.EntryPage = entryPage
			im.saveMeta(m)
		}
	}
	return nil
}

// CheckUniqueForKey 校验行数据 rowData 是否会撞上某个唯一索引里【别的行】已有的键。
// 冲突返回 ErrDuplicatedKey; 无冲突 (含"该行压根不进这些索引") 返回 nil。
//
// 抽成独立方法是因为它有两个必须分开的调用时机:
//   - InsertRow / UpdateByKey 内部各调一次 —— 保证任何入口在写入前都拦得住;
//   - Table 的更新路径还要在 serializeRow 之前【再】调一次: serializeRow 会就地
//     覆写旧行的动态堆块, 冲突必须在它动手之前发现 (见 serializeRowForCheck)。
//
// 因此这里只读不写、可重复调用。
func (im *IndexMetaPage) CheckUniqueForKey(rowData []byte, pkID uint64, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) error {
	return im.checkUniqueForKey(im.resolveIndexes(), rowData, pkID, fields, activeVer, versions)
}

// HasUniqueIndex 是否存在活跃的唯一索引。
//
// 更新路径 (checkUniqueUpdate) 用它跳过 serializeRowForCheck: 预检的唯一目的是在
// serializeRow 动手覆写旧行动态堆块之前拦下唯一冲突 —— 没有唯一索引时整行试序列化
// 纯属浪费一次全行序列化。resolveIndexes 的结果有缓存, 本判断近乎零成本。
func (im *IndexMetaPage) HasUniqueIndex() bool {
	for _, ri := range im.resolveIndexes() {
		if ri.meta.Unique() {
			return true
		}
	}
	return false
}

// checkUniqueForKey 是 CheckUniqueForKey 的内核, 收【已解析】的索引集。
// InsertRow / UpdateByKey 手里本来就有一份 resolveIndexes 的结果, 再解析一遍等于
// 每次写入都把索引元信息链多走一趟 —— 插入/更新是热路径, 这趟省掉。
func (im *IndexMetaPage) checkUniqueForKey(indexes []resolvedIndex, rowData []byte, pkID uint64, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) error {
	for _, ri := range indexes {
		if !ri.meta.Unique() {
			continue
		}
		indexKey := extractIndexKey(rowData, ri.meta, fields, versions)
		if indexKey == nil {
			continue // 任一分量为 NULL → 该行不进本索引, 唯一性对 NULL 不设限
		}
		if im.uniqueConflict(ri, indexKey, pkID) {
			return fmt.Errorf("%w: 索引 %q, pkID=%d",
				ErrDuplicatedKey, util.BytesToString(ri.meta.IndexName[:]), pkID)
		}
	}
	return nil
}

// uniqueConflict 判断唯一索引里是否已有"别的行"占着这个键。
//
// 用 resolve=nil 直接拿条目里的 8B pkID: 唯一性只关心"是不是同一行", 不需要回表取整行
// (回表会顺着行存储索引再查一次, 每次插入都多一遍 O(页数) 的定位)。返回的 pkID 与
// 当前行相同说明命中的是本行自己的条目 (更新场景), 不算冲突。
func (im *IndexMetaPage) uniqueConflict(ri resolvedIndex, indexKey []byte, pkID uint64) bool {
	got, found := ri.handler.Search(im.handle, ri.meta.EntryPage, indexKey, len(indexKey), nil)
	if !found || len(got) < 8 {
		return false
	}
	return binary.BigEndian.Uint64(got[:8]) != pkID
}

// searchAt 在已定位的索引元信息上查一行 (统一契约: 返回整行, 带 8B pkID 前缀)。
//
// 本层不判断索引类型: 二级索引所缺的行数据由处理器内部用回表函数补齐。
// 独立抽出是为了让"先定位、再复用"的调用方 (Table.SearchByIndex/DeleteRow/UpdateRow)
// 只做一次树查找, 不必重复「findIndexMetaByName + handler.Search」。
func (im *IndexMetaPage) searchAt(meta *IndexMeta, indexKey []byte) ([]byte, bool) {
	if meta == nil {
		return nil, false
	}
	handler, ok := index.Compose.GetHandler(meta.IndexType)
	if !ok {
		return nil, false
	}
	return handler.Search(im.handle, meta.EntryPage, indexKey, len(indexKey), im.resolver())
}

// SearchByIndex 通过索引搜索 (统一契约: 返回整行, 带 8B pkID 前缀)
func (im *IndexMetaPage) SearchByIndex(idxName string, indexKey []byte) ([]byte, bool) {
	return im.searchAt(im.findIndexMetaByName(idxName), indexKey)
}

// ResolveByPK 按 pkID 查找整行数据 (带 8B pkID 前缀)。
//
// 本层不知道行数据"存在哪种索引里": 由实现 RowStoreHandler 的处理器自己读取
// (主键 B+ 树走 DataPage 二分缓存, 时序索引走条目链)。
func (im *IndexMetaPage) ResolveByPK(pkID uint64) ([]byte, bool) {
	m, h, ok := im.rowStoreMeta()
	if !ok {
		return nil, false
	}
	return h.ResolveRow(im.handle, m.EntryPage, pkID, &im.rowCache)
}

// BatchResolveByPK 批量回表, 结果按【传入顺序】交付 (sink 回调)。
//
// 传入顺序即二级索引 RangeScan 的输出序 —— B+ 树按索引键序输出、Hash 范围扫描
// 内部也按键排序; needSort=false 的免排序 ORDER BY 路径依赖这个顺序不被打乱,
// 所以内部排序只是为了游标走查, 交付时恢复传入顺序。
//
// budget > 0 时最多交付 budget 行 (LIMIT+OFFSET 场景): 批尺寸随剩余预算收缩,
// 达到预算立即收工 —— Where(重键=值).First() 这类点查只解 1 行, 而不是把
// 命中桶里的上千行全部解出再丢掉。剩余条件会过滤行的场景: 某批交付不足预算时
// 继续下一批, 直到预算耗尽或输入走完。
//
// 实现: 输入的升序拷贝走单调游标 (BatchRowResolver) 收进 map, 再按原序回调;
// 未实现批量接口的处理器退化为逐个 ResolveRow (按传入顺序, 与旧行为一致)。
// 行快照跨 sink 调用有效。sink 返回 false 时提前停止。
func (im *IndexMetaPage) BatchResolveByPK(pkIDs []uint64, budget int,
	sink func(pkID uint64, storeRow []byte) bool) {
	m, h, ok := im.rowStoreMeta()
	if !ok || len(pkIDs) == 0 {
		return
	}
	emitted := 0
	br, isBatch := h.(index.BatchRowResolver)
	if !isBatch {
		for _, pkID := range pkIDs {
			storeRow, found := h.ResolveRow(im.handle, m.EntryPage, pkID, &im.rowCache)
			if !found {
				continue // 行已被删除, 跳过
			}
			if !sink(pkID, storeRow) {
				return
			}
			emitted++
			if budget > 0 && emitted >= budget {
				return
			}
		}
		return
	}
	// 分批: 每批取输入的一个连续子段, 子段单独升序后走游标走查,
	// 交付严格按该子段的输入顺序。批与批之间游标重启, 回调里的索引改写安全。
	for start := 0; start < len(pkIDs); {
		if budget > 0 && budget-emitted <= 0 {
			return
		}
		chunk := resolveChunk
		if budget > 0 && budget-emitted < chunk {
			chunk = budget - emitted // 预算紧张时批随预算收缩 (点查退化为单行解)
		}
		end := start + chunk
		if end > len(pkIDs) {
			end = len(pkIDs)
		}
		inputChunk := pkIDs[start:end]
		if len(inputChunk) == 1 {
			// 点查快路径 (LIMIT=1/First 常见): 单行直解, 免排序拷贝与 map
			pkID := inputChunk[0]
			if storeRow, found := h.ResolveRow(im.handle, m.EntryPage, pkID, &im.rowCache); found {
				if !sink(pkID, storeRow) {
					return
				}
				emitted++
			}
			start = end
			continue
		}
		sortedChunk := make([]uint64, len(inputChunk))
		copy(sortedChunk, inputChunk)
		slices.Sort(sortedChunk)
		rows := make(map[uint64][]byte, len(inputChunk))
		br.ResolveMany(im.handle, m.EntryPage, sortedChunk,
			func(pkID uint64, storeRow []byte) bool {
				rows[pkID] = storeRow
				return true
			})
		for _, pkID := range inputChunk { // 本批输入子段, 原序交付
			storeRow, ok := rows[pkID]
			if !ok {
				continue // 行已被删除
			}
			if !sink(pkID, storeRow) {
				return
			}
			emitted++
			if budget > 0 && emitted >= budget {
				return
			}
		}
		start = end
	}
}

// InvalidateRowCache 使回表定位缓存失效 (外部直接改动行存储索引后调用)
func (im *IndexMetaPage) InvalidateRowCache() { im.rowCache = nil }

// rowStoreHasPK 判断行存储里是否已存在该 pkID —— 不建立/复用回表定位缓存。
//
// 插入查重专用: ResolveByPK 会按需构建 im.rowCache (O(页数)), 而任何一次插入
// 都会让它失效; 若每行插入都走 ResolveByPK, 逐行插入就退化成 O(N²)
// (千万级写入从分钟级变成不可完成)。这里传 cache=nil 走处理器自己的定位路径。
//
// 语义与 ResolveByPK 一致 (仅看行是否存在, 且过滤墓碑页), 只是不做缓存。
func (im *IndexMetaPage) rowStoreHasPK(pkID uint64) bool {
	m, h, ok := im.rowStoreMeta()
	if !ok || m == nil {
		return false
	}
	_, found := h.ResolveRow(im.handle, m.EntryPage, pkID, nil)
	return found
}

// DeleteByKey 通过索引删除一行 —— 统一入口, 不区分索引类型。
//
// storeRow 由调用方 SearchByIndex 预先定位后传入 (头 8 字节为 pkID), 本方法不再
// 重复树查找 —— 否则删除路径会对同一行做两次 handler.Search。
//
// 步骤:
//  1. 复用调用方已定位的整行 (二级索引所缺的行数据已由处理器内部回表补齐)。
//  2. 遍历全部索引, 让每个处理器各自删掉该行在本索引中的条目,
//     pkID 一并传入以便二级索引在重复 key 下精确定位。
//     主键 <-> 二级索引的级联因此天然成立, 无需在本层判断索引类型。
func (im *IndexMetaPage) DeleteByKey(idxName string, indexKey []byte, storeRow []byte, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) error {
	im.rowCache = nil // 删除可能释放 DataPage, 失效缓存
	meta := im.findIndexMetaByName(idxName)
	if meta == nil || len(storeRow) < 8 {
		return page.ErrNotFound
	}
	pkID := binary.BigEndian.Uint64(storeRow[:8])
	rowData := storeRow[8:]

	// 每个索引各自删除该行在本索引中的条目
	var firstErr error
	for _, ri := range im.resolveIndexes() {
		m := ri.meta
		key := extractIndexKey(rowData, m, fields, versions)
		if indexStoresRows(m) {
			key = rowStoreInsertKey(m, pkID, rowData, fields, versions)
		}
		if len(key) == 0 {
			if m != meta {
				continue // 取不到行数据, 无法推导该索引键, 跳过
			}
			key = indexKey // 目标索引: 调用方给的键就是它在索引里的键
		}
		entryPage := m.EntryPage
		if err := ri.handler.Delete(im.handle, &entryPage, key, pkID, len(key)); err != nil {
			if err != page.ErrNotFound && firstErr == nil {
				firstErr = err
			}
		}
		if entryPage != m.EntryPage {
			m.EntryPage = entryPage
			im.saveMeta(m)
		}
	}
	return firstErr
}

// UpdateByKey 通过索引更新一行 —— 统一入口, 不区分索引类型。
//
// oldStoreRow 由调用方 SearchByIndex 预先定位后传入 (头 8 字节为 pkID), 本方法不再
// 重复回表 —— 否则更新路径会先 ResolveByPK, 失败还会再 SearchByIndex。
//
// 每个索引各自处理自己条目的更新:
//   - 行存储索引(主键/时序): 就地改写行数据;
//   - 二级索引: 只有 [key + pkID], 键未变则无需动作, 键变了则按 (旧键, pkID) 删旧插新。
//
// 本层只负责"把新行数据 + 新旧键交给处理器", 不判断索引类型。
func (im *IndexMetaPage) UpdateByKey(indexKey []byte, newRowData []byte, pkID uint64, oldStoreRow []byte, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) error {
	// 取旧行数据: 二级索引需要用旧行提取"旧键", 才能精确删掉旧条目
	var oldRowData []byte
	if len(oldStoreRow) > 8 {
		oldRowData = oldStoreRow[8:]
	}

	indexes := im.resolveIndexes()

	// 唯一约束预检: 新键若已被"别的行"占着, 必须在任何写入之前拒绝。
	// 放在循环里逐个索引边查边更新是不行的 —— 撞车时前面几个索引已经改好了,
	// 旧键已删、新键已插, 而函数返回错误, 调用方以为没更新, 数据却变了一半。
	// (Table 的更新路径还会在 serializeRow 之前先调一次, 以免旧行的动态堆块被
	//  试改 —— 这里这次是给"直接调 UpdateByKey 的调用方"兜底。)
	if err := im.checkUniqueForKey(indexes, newRowData, pkID, fields, activeVer, versions); err != nil {
		return err
	}

	// 逐个索引更新本行的条目
	var firstErr error
	for _, ri := range indexes {
		m := ri.meta
		// 定位键: 能用旧行提取就用旧键; 提取不到时退回调用方给的查找键
		// (主键/时序索引的查找键即其在索引中的键, 与既有行为一致)
		oldKey := indexKey
		oldKeyFromRow := false
		if oldRowData != nil {
			if indexStoresRows(m) {
				if k := rowStoreInsertKey(m, pkID, oldRowData, fields, versions); len(k) > 0 {
					oldKey, oldKeyFromRow = k, true
				}
			} else if k := extractIndexKey(oldRowData, m, fields, versions); len(k) > 0 {
				oldKey, oldKeyFromRow = k, true
			}
		}
		newKey := extractIndexKey(newRowData, m, fields, versions)
		if indexStoresRows(m) {
			newKey = rowStoreInsertKey(m, pkID, newRowData, fields, versions)
		}

		entryPage := m.EntryPage
		var err error
		switch {
		case len(newKey) > 0:
			err = ri.handler.Update(im.handle, &entryPage, oldKey, newKey, newRowData, pkID, len(newKey))
		case oldRowData == nil:
			// 行存储索引 (主键/时序): 键取不到也必须改写行数据, 用旧键原地更新
			err = ri.handler.Update(im.handle, &entryPage, oldKey, oldKey, newRowData, pkID, len(oldKey))
		case oldKeyFromRow:
			// 二级索引且旧行有键、新行取不到键: 该列已变为 NULL (或其版本不再有此列),
			// 这一行已退出本索引 —— 必须删掉旧条目。留下旧条目会让该索引继续命中一个
			// 现在为 NULL 的行, 直接破坏"NULL 不匹配任何比较"的三值语义。
			err = ri.handler.Delete(im.handle, &entryPage, oldKey, pkID, len(oldKey))
		default:
			// 二级索引且新旧都取不到键: 该行本来就不在本索引里, 无需动作。
		}
		if err != nil && err != page.ErrNotFound {
			// 过去这里丢弃了返回值, 导致 schema 演进后更新旧行"静默失败":
			// 对外返回 nil 但数据实际没变。必须向上抛出。
			firstErr = fmt.Errorf("更新索引 %q (pkID=%d) 失败: %w",
				util.BytesToString(m.IndexName[:]), pkID, err)
		}
		if entryPage != m.EntryPage {
			m.EntryPage = entryPage
			im.saveMeta(m)
		}
		if firstErr != nil {
			break
		}
	}
	// 回表定位缓存 (rowCache) 只在"行可能换页"时才失效。
	// PrimaryHandler.Update 的三级策略里, 只有行变长且页内装不下才会换页;
	// 行长不变 / 变短时行仍留在原 DataPage, 缓存依旧有效。
	// 若在这里无条件失效, 逐行更新会把 O(页数) 的缓存全量重建摊到每一行上,
	// 退化成 O(N²) —— 50 万行的批量更新会卡死。
	if oldRowData == nil || len(newRowData) > len(oldRowData) {
		im.rowCache = nil
	}
	return firstErr
}