TsDB2/DB/core/table_index.go

package core

import (
	"cmp"
	"encoding/binary"
	"fmt"
	"slices"

	"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"
)

// CreateIndex 建二级索引。
//
// 这里先校验分量列能不能做键, 再交给 IndexMetaPage 建 —— 拆成两层是因为
// IndexMetaPage 只拿得到索引自己的元信息, 看不到表的字段类型:
//   - 分量列不存在 → ErrFieldNotFound;
//   - 分量是动态列 → ErrIndexDynamicField。动态列在行内只存 6B 堆引用, 值 != 引用,
//     按值查找永远不可能命中; 建出来既查不到也不会被查询侧选中 (buildKeyLayout 会跳过),
//     只会白占空间、白写。拦在建的这一步, 而不是等用到才发现;
//   - 分量是 blob 列 → ErrIndexBlobField。blob 列行内只存 8B BlobRef (页号 + 长度),
//     拿它当索引键等于按页号排序/查找 —— 索引里存的不是内容, 命中与否全看内容落在哪一页;
//   - 复合索引 (IndexTypeCompose) 的分量列宽度不再强制 8 字节 ——
//     底层已升级为 B+ 树, 按分量真实宽度拼定长键, 溢出闸门 (checkIndexSpillFloor)
//     负责兜住"一页两行"的行宽上限。
//
// unique 为可选 (只取第一个值), 见 IndexMetaPage.CreateIndex: 传 true 建的是唯一索引,
// 同一索引键只允许一行; 任一分量为 NULL 的行不进该索引, 因此多个 NULL 天然共存。
// 唯一约束在写入路径 (InsertRow / UpdateByKey) 上校验, 违反时返回 ErrDuplicatedKey。
func (table *Table) CreateIndex(idxName string, typ page.IndexType, fields []byte, unique ...bool) (*IndexMeta, error) {
	if table.db != nil && table.db.HasActiveTransaction() {
		return nil, errDDLInTransaction("CreateIndex")
	}
	defer table.db.lockOp()()
	for i, fi := range fields {
		if int(fi) >= len(table.Field) {
			return nil, fmt.Errorf("%w: 索引 %q 的第 %d 个分量", ErrFieldNotFound, idxName, i+1)
		}
		ft := table.Field[fi].FieldType
		if types.IsDynamic(ft) {
			return nil, fmt.Errorf("%w (索引 %q 的第 %d 个分量)", ErrIndexDynamicField, idxName, i+1)
		}
		// v10: blob 列行内只有 8B BlobRef (页号 + 长度), 拿它当索引键等于按页号排序/查找 ——
		// 索引里存的不是内容, 命中与否全看内容落在哪一页. 与动态列同等拦下.
		if types.IsBlob(ft) {
			return nil, fmt.Errorf("%w (索引 %q 的第 %d 个分量)", ErrIndexBlobField, idxName, i+1)
		}
	}
	// 行存储索引 (主键/时序) 的分量列不允许可空: 它是行的定位键, 写入路径按它
	// 提取索引键 (extractIndexKey), NULL 分量不提取 —— 键提取不出来时删除/更新
	// 对目标索引静默失效 (ErrNotFound 按索引逐个吞掉, 不报错也不动数据)。
	// 建索引这一步就拦下, 比"数据写进去之后删不掉"好查得多。
	if h, ok := index.Compose.GetHandler(typ); ok && handlerStoresRows(h) {
		for _, fid := range fields {
			if int(fid) < len(table.Field) && table.Field[fid].IsNullable() {
				return nil, fmt.Errorf("行存储索引 %q 的分量列 %q 声明了可空 (定位键必须非空, 请去掉 nullable 标记)",
					idxName, table.Field[fid].FieldName)
			}
		}
	}
	// 溢出闸门: 分量列一旦建索引就不能再值级溢出, 行内下界随之抬升。
	// 下界超过"一页两行"的配额时必须在这里拒绝 —— 否则这张表从建索引成功起就再也压不
	// 进两行 (单条插入仍会"成功"地落在 8159..16332 之间, 页内顺序/有序插入/页内重建
	// 一起退化, 而且从现象上完全看不出是索引造成的)。
	if err := table.checkIndexSpillFloor(fields); err != nil {
		return nil, fmt.Errorf("%w: 索引 %q: %v", ErrIndexRowTooWide, idxName, err)
	}
	m, err := table.indexMeta.CreateIndex(idxName, typ, fields, unique...)
	if err != nil {
		return nil, err
	}
	// 回填存量行。
	// 旧实现只建元数据, 索引里一条记录都没有: 建索引之前就存在的行永远查不到,
	// 而引擎又没有公开的"重建全部二级索引"接口 —— 数据在、索引空, 是最难排查的
	// 一类不一致。建索引这一步必须自己把存量行灌进去。
	if err := table.backfillIndex(m); err != nil {
		// 回填失败不能留下"半个索引": 元数据还在但内容不全, 比没有索引更危险
		// (查询会选中它, 于是稳定地查不到行)。摘掉元数据, 让错误显式冒出来。
		// 走 table.DropIndex 而不是 indexMeta.DropIndex: 后者只软删元数据, 会把
		// 回填到一半的树页留成孤儿页。
		_ = table.DropIndex(idxName)
		return nil, err
	}
	if table.db != nil {
		if err := table.db.ddlBoundary("CreateIndex"); err != nil {
			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.versions)
			if secKey == nil {
				continue // 该行没有这个键 (列不在它的 schema 版本里)
			}
			entry := make([]byte, len(secKey)+8)
			copy(entry, secKey)
			// 条目尾部是 pkID (8B BE), 与 rebuildIndexes 的二级条目同一布局
			binary.BigEndian.PutUint64(entry[len(secKey):], binary.BigEndian.Uint64(storeRow[:8]))
			entries = append(entries, entry)
		}
	}
	if len(entries) == 0 {
		// 没有任何存量行进这个索引 → 保持空树 (EntryPage==0), 后续插入照常开根
		return nil
	}
	// 按复合键 (secKey, pkID) 升序排序: 二级树的复合分隔键与叶内复合二分要求
	// 全树按复合键全局有序, 同 key 的条目必须再按 pkID 排
	slices.SortStableFunc(entries, func(a, b []byte) int {
		if c := compareBytes(a[:len(a)-8], b[:len(b)-8]); c != 0 {
			return c
		}
		return cmp.Compare(binary.BigEndian.Uint64(a[len(a)-8:]), binary.BigEndian.Uint64(b[len(b)-8:]))
	})
	// 唯一索引进来了要顺手验一遍存量数据: 库里本来就有重复值时, 这个约束是【建不出来】的。
	// 硬建下去只会得到一个"违反自身约束"的索引 (同一键挂着两行), 之后点查可能返回其中
	// 任意一行, 而且从索引上完全看不出它已经坏了。宁可在建索引这一步报错, 由调用方决定
	// 是清理数据还是改用非唯一索引。排序后重复键必然相邻, 一次线性扫描即可。
	if m.Unique() {
		for i := 1; i < len(entries); i++ {
			if compareBytes(entries[i-1][:len(entries[i-1])-8], entries[i][:len(entries[i])-8]) == 0 {
				return fmt.Errorf("%w: 索引 %q 的存量数据中已存在重复键 (pkID %d 与 %d), 无法建成唯一索引",
					ErrDuplicatedKey, util.BytesToString(m.IndexName[:]),
					binary.BigEndian.Uint64(entries[i-1][len(entries[i-1])-8:]),
					binary.BigEndian.Uint64(entries[i][len(entries[i])-8:]))
			}
		}
	}
	if err := buildSecondaryFromEntries(table.handle, m, entries); err != nil {
		return err
	}
	table.indexMeta.saveMeta(m)
	table.indexMeta.InvalidateRowCache()
	return nil
}
func (table *Table) DropIndex(idx string) error {
	defer table.db.lockOp()()
	// DROP INDEX 必须把整棵索引树还给空闲链。
	// 旧实现只把元数据打上软删除标记 (Flag|=0x80) 并从 metas 摘掉: 根指针一丢, 整棵树
	// (内部节点 + 叶子) 就既不可达、又不在空闲链上 —— 变成永久孤儿页, 文件只涨不缩。
	// 实测 20 万行的一棵二级索引约 197 页; bench_10m 里 9.9M 行的 Dept 索引是数万页,
	// 正是那批孤儿页的主因。
	// 注意: indexMeta.DropIndex 会用 append 挪动 metas 元素, 之后 &metas[i] 就失效了,
	// 所以必须在摘除之前把 IndexType / EntryPage 快照出来。
	var itype page.IndexType
	var rootPID uint32
	if m := table.indexMeta.findIndexMetaByName(idx); m != nil {
		itype, rootPID = m.IndexType, m.EntryPage
	}
	if err := table.indexMeta.DropIndex(idx); err != nil {
		return err
	}
	// 旧结构此刻已不可达 (元数据已摘), 归还它的全部页。
	releaseIndexPages(table, itype, rootPID)
	return nil
}

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

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

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

// UpdateRow 通过索引更新一行
func (table *Table) UpdateRow(idxName string, key []byte, values []types.FieldValue) error {
	defer table.db.lockOp()()
	// 先取旧行: 动态字段可复用旧行已占用的堆块, 避免每次更新都产生堆垃圾。
	// 该行同时下传给 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
}

// SearchByIndexWithPK 同 SearchByIndex, 但额外把该行的真实行号一并返回。
//
// 行号是行的 8 字节前缀, 不在字段值里 —— 上层要把它带回对象, 之后才能按行号
// 精确 Update/Delete (见 ORM 的 modelCore.rowID)。
func (table *Table) SearchByIndexWithPK(idxName string, key []byte) (uint64, []types.FieldValue, error) {
	defer table.db.lockOp()()
	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
}