TsDB2/DB/core/IndexMetaPage.go

package core

import (
	"TsDB2/DB/core/index"
	"TsDB2/DB/core/index/btree"
	"TsDB2/DB/core/index/datapage"
	"TsDB2/DB/core/page"
	"TsDB2/DB/core/types"
	"TsDB2/DB/core/util"
	"encoding/binary"
	"unsafe"
)

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

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

	handle      page.HandleDb
	hdrSize     uint32
	metaSize    uint32
	lastPageID  uint32
	primaryLink map[string]string // 二级索引名 → 一级索引名
	primaryDPs  []uint32          // 主键 B+ 树所有 DataPage ID 缓存
}

// 初始化
func (im *IndexMetaPage) Init(pageID uint32) {
	im.hdrSize = uint32(unsafe.Sizeof(page.PageHdr{}))
	im.metaSize = uint32(unsafe.Sizeof(IndexMeta{}))
	im.metas = make([]IndexMeta, 0)
	im.hdr.PageID = pageID
	im.hdr.Type = page.IndexMetaIndex
	im.hdr.NextPage = 0
	if pageID == 0 {
		return
	}
	for pageID != 0 {
		im.lastPageID = pageID
		buf := make([]byte, PageSize)
		if err := im.handle.ReadPage(pageID, buf); err != nil {
			break
		}
		hdr := (*page.PageHdr)(unsafe.Pointer(&buf[0]))
		offset := im.hdrSize

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

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

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

// 创建索引（纯元数据，不碰IndexPage）
func (im *IndexMetaPage) CreateIndex(indexName string, itype page.IndexType, fieldIDs []byte) (*IndexMeta, error) {
	fc := byte(len(fieldIDs))
	if fc == 0 || fc > 16 {
		return nil, ErrIndexFieldCount
	}

	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.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.PageHdr)(unsafe.Pointer(&buf[0]))
	}

	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)
	return &m, 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.PageHdr)(unsafe.Pointer(&buf[0]))
		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:]...)
						break
					}
				}
				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.PageHdr)(unsafe.Pointer(&buf[0]))
	hdr.Type = page.IndexMetaIndex
	hdr.PageID = pageID
	hdr.UseSize = uint32(unsafe.Sizeof(*hdr))
	im.handle.SavePage(pageID, buf)
	return buf, hdr
}

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

func (im *IndexMetaPage) buildPrimaryLink() {
	im.primaryLink = make(map[string]string)
	var primaryName string
	for _, m := range im.metas {
		if m.Flag&0x80 != 0 {
			continue
		}
		name := util.BytesToString(m.IndexName[:])
		if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
			primaryName = name
		} else {
			if primaryName != "" {
				im.primaryLink[name] = primaryName
			}
		}
	}
}

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) getPrimaryFor(secName string) *IndexMeta {
	if im.primaryLink == nil {
		im.buildPrimaryLink()
	}
	if primName, ok := im.primaryLink[secName]; ok {
		return im.findIndexMetaByName(primName)
	}
	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.PageHdr)(unsafe.Pointer(&buf[0]))
		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
	}
}

// extractIndexKey 从行数据中提取索引键 (版本感知)
// rowData 格式: [schemaVer(1B)] [field_data...]
func extractIndexKey(rowData []byte, meta *IndexMeta, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) []byte {
	if meta.FieldCnt == 0 || len(rowData) < 1 {
		return nil
	}
	fieldIdx := int(meta.FieldIDs[0])

	// 从 rowData 读取该行的 schema 版本
	ver := rowData[0]
	var sv *SchemaVersion
	for i := range versions {
		if versions[i].Version == ver {
			sv = &versions[i]
			break
		}
	}
	if sv == nil {
		return nil
	}

	// 按该版本的字段顺序计算偏移
	off := 1 // 跳过 schemaVer
	for _, idx := range sv.FieldIndices {
		if int(idx) == fieldIdx {
			keyLen := int(getFieldSize(fields[fieldIdx].FieldType))
			if off+keyLen > len(rowData) {
				return nil
			}
			key := make([]byte, keyLen)
			copy(key, rowData[off:off+keyLen])
			return key
		}
		off += int(getFieldSize(fields[idx].FieldType))
	}
	return nil
}

// InsertRow 插入行到所有索引
func (im *IndexMetaPage) InsertRow(fields []types.FieldInfo, pkID uint64, rowData []byte, activeVer byte, versions []SchemaVersion) error {
	im.primaryDPs = nil // 插入可能创建新 DataPage, 失效缓存
	for i := range im.metas {
		m := &im.metas[i]
		if m.Flag&0x80 != 0 {
			continue
		}
		handler, ok := index.Compose.GetHandler(m.IndexType)
		if !ok {
			continue
		}
		indexKey := extractIndexKey(rowData, m, fields, activeVer, versions)
		if indexKey == nil {
			continue
		}
		entryPage := m.EntryPage
		_, err := 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
}

// SearchByIndex 通过索引搜索
func (im *IndexMetaPage) SearchByIndex(idxName string, indexKey []byte, fields []types.FieldInfo) ([]byte, bool) {
	meta := im.findIndexMetaByName(idxName)
	if meta == nil {
		return nil, false
	}
	handler, ok := index.Compose.GetHandler(meta.IndexType)
	if !ok {
		return nil, false
	}
	result, found := handler.Search(im.handle, meta.EntryPage, indexKey, len(indexKey))
	if !found {
		return nil, false
	}
	if meta.IndexType == page.IndexTypePrimary || meta.IndexType == page.IndexTypeTime {
		return result, true
	}
	// 二级索引: result 是 8B BigEndian pkID, 需要回查主键
	pkID := binary.BigEndian.Uint64(result)
	return im.ResolveByPK(pkID, fields)
}

// ResolveByPK 通过主键ID (pkID) 查找行数据
// 注意: pkID 是自增计数器, 不一定等于主键索引字段值
// 因此不能直接用 pkID 搜索主键 B+ 树 (B+ 树 key = 索引字段值)
// 必须扫描所有 DataPage 按存储的 pkID 匹配
func (im *IndexMetaPage) ResolveByPK(pkID uint64, fields []types.FieldInfo) ([]byte, bool) {
	pkKeyBE := make([]byte, 8)
	binary.BigEndian.PutUint64(pkKeyBE, pkID)
	for _, m := range im.metas {
		if m.Flag&0x80 != 0 {
			continue
		}
		if m.IndexType == page.IndexTypePrimary {
			// 主键 B+ 树: DataPage 按 pkID 顺序, 用二分查找定位目标页
			if im.primaryDPs == nil {
				im.primaryDPs = btree.GetDataPageIDs(im.handle, m.EntryPage)
			}
			targetPK := binary.BigEndian.Uint64(pkKeyBE)

			// 二分查找: 找第一个 MaxPK >= targetPK 的 DataPage
			left, right := 0, len(im.primaryDPs)-1
			for left <= right {
				mid := (left + right) / 2
				dp, err := datapage.Load(im.handle, im.primaryDPs[mid])
				if err != nil || dp.Count() == 0 {
					// 空页, 跳过 (但不影响二分逻辑)
					left = mid + 1
					continue
				}
				if dp.MaxPK() < targetPK {
					// 当前页的最大 pkID < 目标, 目标在右侧
					left = mid + 1
				} else {
					// 当前页的最大 pkID >= 目标, 可能是目标页, 继续向左找
					right = mid - 1
				}
			}

			// left 指向第一个可能包含目标的 DataPage
			if left >= 0 && left < len(im.primaryDPs) {
				dp, err := datapage.Load(im.handle, im.primaryDPs[left])
				if err == nil && dp.Count() > 0 {
					if targetPK >= dp.MinPK() && targetPK <= dp.MaxPK() {
						if storedRow, ok := dp.FindRowByPK(pkKeyBE); ok {
							return storedRow, true
						}
					}
				}
			}
			return nil, false
		}
		if m.IndexType == page.IndexTypeTime {
			// 时序索引: 遍历条目链找到 DataPage
			curPageID := m.EntryPage
			for curPageID != 0 {
				buf := make([]byte, PageSize)
				if err := im.handle.ReadPage(curPageID, buf); err != nil {
					break
				}
				hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
				entryLen := int(hdr.EntryLen)
				count := int(hdr.Count)
				entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
				for i := 0; i < count; i++ {
					dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
					if dpID != 0 {
						dp, err := datapage.Load(im.handle, dpID)
						if err == nil {
							if storedRow, ok := dp.FindRowByPK(pkKeyBE); ok {
								return storedRow, true
							}
						}
					}
				}
				curPageID = hdr.Hdr.NextPage
			}
			return nil, false
		}
	}
	return nil, false
}

// DeleteByKey 通过索引删除行（级联处理二级索引）
func (im *IndexMetaPage) DeleteByKey(idxName string, indexKey []byte, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) error {
	im.primaryDPs = nil // 删除可能释放 DataPage, 失效缓存
	if im.primaryLink == nil {
		im.buildPrimaryLink()
	}
	meta := im.findIndexMetaByName(idxName)
	if meta == nil {
		return page.ErrNotFound
	}
	handler, ok := index.Compose.GetHandler(meta.IndexType)
	if !ok {
		return page.ErrNotFound
	}
	if meta.IndexType == page.IndexTypePrimary || meta.IndexType == page.IndexTypeTime {
		// 主键/时序索引: 先读取行数据用于级联删除二级索引
		var rowData []byte
		if meta.IndexType == page.IndexTypePrimary {
			if data, found := handler.Search(im.handle, meta.EntryPage, indexKey, len(indexKey)); found {
				rowData = data
			}
		}
		entryPage := meta.EntryPage
		err := handler.Delete(im.handle, &entryPage, indexKey, len(indexKey))
		if entryPage != meta.EntryPage {
			meta.EntryPage = entryPage
			im.saveMeta(meta)
		}
		// 级联删除二级索引
		if rowData != nil {
			// PrimaryHandler 返回的 rowData 包含 8B LE pkID 前缀, 需要剥离
			origRow := rowData
			if len(origRow) > 8 {
				origRow = origRow[8:]
			}
			primName := util.BytesToString(meta.IndexName[:])
			for i := range im.metas {
				m := &im.metas[i]
				if m.Flag&0x80 != 0 {
					continue
				}
				secName := util.BytesToString(m.IndexName[:])
				if linkedPrim, ok := im.primaryLink[secName]; ok && linkedPrim == primName {
					secHandler, ok := index.Compose.GetHandler(m.IndexType)
					if ok {
						secKey := extractIndexKey(origRow, m, fields, activeVer, versions)
						if secKey != nil {
							secEntryPage := m.EntryPage
							secHandler.Delete(im.handle, &secEntryPage, secKey, len(secKey))
							if secEntryPage != m.EntryPage {
								m.EntryPage = secEntryPage
								im.saveMeta(m)
							}
						}
					}
				}
			}
		}
		return err
	}
	// 二级索引: 先查 pkID, 删主键, 再删二级
	pkIDBuf, found := handler.Search(im.handle, meta.EntryPage, indexKey, len(indexKey))
	if !found {
		return page.ErrNotFound
	}
	pkID := binary.BigEndian.Uint64(pkIDBuf)
	// 找主键索引删除
	primMeta := im.getPrimaryFor(idxName)
	if primMeta != nil {
		primHandler, ok := index.Compose.GetHandler(primMeta.IndexType)
		if ok {
			pkKey := make([]byte, 8)
			binary.BigEndian.PutUint64(pkKey, pkID)
			entryPage := primMeta.EntryPage
			switch primMeta.IndexType {
			case page.IndexTypePrimary:
				// 主键 B+ 树: 用 DeleteByPK 按 pkID 匹配
				primHandler.(*index.PrimaryHandler).DeleteByPK(im.handle, &entryPage, pkID)
			case page.IndexTypeTime:
				// 时序索引: 遍历所有条目, 从 DataPage 按 pkID 删除
				curPageID := entryPage
				for curPageID != 0 {
					buf := make([]byte, page.IndexPageSize)
					if err := im.handle.ReadPage(curPageID, buf); err != nil {
						break
					}
					hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
					entryLen := int(hdr.EntryLen)
					count := int(hdr.Count)
					entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
					for i := 0; i < count; i++ {
						dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
						if dpID != 0 {
							if dp, err := datapage.Load(im.handle, dpID); err == nil {
								if dp.DeleteByPK(pkKey) {
									if dp.AllDeleted() {
										if im.handle.FreePage != nil {
											im.handle.FreePage(dpID)
										}
									} else {
										dp.Save()
									}
									goto timeDeleteDone
								}
							}
						}
					}
					curPageID = hdr.Hdr.NextPage
				}
			timeDeleteDone:
			}
			if entryPage != primMeta.EntryPage {
				primMeta.EntryPage = entryPage
				im.saveMeta(primMeta)
			}
		}
	}
	// 删二级索引条目
	entryPage := meta.EntryPage
	err := handler.Delete(im.handle, &entryPage, indexKey, len(indexKey))
	if entryPage != meta.EntryPage {
		meta.EntryPage = entryPage
		im.saveMeta(meta)
	}
	return err
}

// UpdateByKey 通过索引更新行（级联处理二级索引）
func (im *IndexMetaPage) UpdateByKey(idxName string, indexKey []byte, newRowData []byte, pkID uint64, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) error {
	if im.primaryLink == nil {
		im.buildPrimaryLink()
	}
	// 更新主键索引中的行数据
	for i := range im.metas {
		m := &im.metas[i]
		if m.Flag&0x80 != 0 {
			continue
		}
		if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
			handler, ok := index.Compose.GetHandler(m.IndexType)
			if ok {
				// 主键/时序索引始终使用 pkID(8B BE) 作为 B+ 树 key
				primKey := make([]byte, 8)
				binary.BigEndian.PutUint64(primKey, pkID)
				entryPage := m.EntryPage
				handler.Update(im.handle, &entryPage, primKey, newRowData, pkID, 8)
				if entryPage != m.EntryPage {
					m.EntryPage = entryPage
					im.saveMeta(m)
				}
			}
			break
		}
	}
	// 更新二级索引（如果索引键变了需要删旧插新）
	for i := range im.metas {
		m := &im.metas[i]
		if m.Flag&0x80 != 0 {
			continue
		}
		if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
			continue
		}
		handler, ok := index.Compose.GetHandler(m.IndexType)
		if !ok {
			continue
		}
		newKey := extractIndexKey(newRowData, m, fields, activeVer, versions)
		if newKey == nil {
			continue
		}
		entryPage := m.EntryPage
		handler.Update(im.handle, &entryPage, newKey, newRowData, pkID, len(newKey))
		if entryPage != m.EntryPage {
			m.EntryPage = entryPage
			im.saveMeta(m)
		}
	}
	return nil
}