package index import ( "TsDB2/DB/core/page" "bytes" "encoding/binary" "errors" "sort" "unsafe" ) // ============================================================================= // Hash 索引 — 磁盘持久化「线性哈希 (Linear Hashing)」二级索引 // // 为什么不用「整体翻倍 + 全表重哈希」: // 旧实现一旦有任意一个桶超阈值, 就把桶数翻倍并把所有条目收进内存重新路由。 // 路由是 hash & (bucketCount-1), 同一个 key 的 hash 相同, 翻倍后仍旧全部落到 // 同一个桶; 于是每插入一行都会再次超阈值、再次翻倍 → 桶数指数膨胀直至卡死。 // 低基数字段 (Age/Dept 只有几十个值) 必然踩中。 // // 线性哈希的做法: // - 维护 level / splitPointer。桶号 = h & (2^level-1), 若结果 < splitPointer // 则再看下一位: h & (2^(level+1)-1)。可用桶数 = 2^level + splitPointer。 // - 每次插入最多分裂「splitPointer 指向的那一个桶」, 把它的条目按新增的那一位 // 一分为二 (原地/增量), 绝不重哈希全表, 也绝不因重复 key 无限翻倍。 // - 是否分裂由「全局负载因子」决定 (平均每桶条目数), 而非单个热点桶的绝对条数, // 因此重复 key 只会让该桶的溢出链变长 (查找仍 O(链长)), 不会引发无意义分裂。 // - 每次删除最多撤销 (合并) 一个桶, 同样只动两个桶。 // // 页面布局 // --------------------------------------------------------------------------- // 元数据页 (即 entryPage, 注册为 IndexDataIndex): // [0, 32) page.IndexHead // [32,36) magic 格式标识, 用于识别旧版本/不兼容布局 // [36,40) level // [40,44) splitPointer // [44,48) totalEntries 活跃条目总数 (驱动负载因子) // [48,52) dirStartPID 桶目录首页 // [52,56) dirPageCount 桶目录页数 // [56,60) keyLen 索引键字节长度 (条目长度 = keyLen + 8) // [60,64) dirTailPID 桶目录链尾页 (整条目录页链可 O(1) 回收) // // 桶目录页 (多页, 用 PageHdr.NextPage 串链; 槽位区从 PageHdrSize 开始): // [0, 20) page.PageHdr // [20, ..) 槽位数组, 每槽 12 字节: [链头(4)][链尾(4)][链长(4)] // 槽 i 描述桶 i 的溢出链 (链头/链尾页 ID 与链长; 0/0/0 = 空桶) // 每页槽位数 = (16384 - 20) / 12 = 1362 // 关键: 链尾与链长随链头一起持久化 —— 回收整条桶链只需 (head, tail, count) // 一次拼接 (O(1)), 不必读链内任何一页。 // // 桶页 (同一桶的溢出链, 用 PageHdr.NextPage 串链 —— 与空闲链同一字段): // [0, 20) page.PageHdr (NextPage 即同桶下一页) // [20,24) BucketID // [24,28) Count 本页活跃条目数 // [28, ..) 条目, 每条 keyLen + 8 字节: [key(keyLen)][pkID(8B BE, 最高位=删除标记)] // 关键一: 条目里保存「原始 key」, 因此等值查找可做精确比较 (无碰撞误判), // 范围扫描也能做真正的键边界过滤。 // 关键二: 桶页的 next 就是公共头 PageHdr.NextPage, 与空闲链完全同一字段 —— // 整条桶链可直接交给 FreeChain 拼接进空闲链, 无需逐页改写回收。 // // 页回收策略: // 分裂/合并/压缩一律「复用旧页」而不是释放后重申请 (freePage 会把页写回空闲 // 链表, 而与页缓存的一致性无保证, 直接释放有让陈旧缓存覆盖空闲链表的风险)。 // 具体做法: 用旧链的页去承载新链, 用不完的页挂到链尾作为空页 (仍可达、可复用), // 保证不产生孤儿页 → DB 文件不会随分裂/删除无限膨胀。 // // 范围扫描: // 哈希不保序, 但条目存了原始 key, 因此 RangeScan 做「(等值则定位单桶 / 否则 // 全桶扫描) + 键边界过滤 + 按 key 排序」, 结果正确且满足 ORDER BY 语义。 // ============================================================================= const ( hashMagic = uint32(0x54484833) // "THH3" (v3: 桶页统一 PageHdr.NextPage, 目录槽记录链尾/链长) hashMinLevel = 4 // 最小层级 hashMinBucketCount = 1 << hashMinLevel // 最小桶数 = 16 hashMaxBuckets = 1 << 22 // 桶数上限 (目录约 48MB: 12B/槽) // 分裂/合并阈值都是「平均每桶活跃条目」的上界, 由 hashSplitThreshold / // hashMergeThreshold 依据桶页容量推导, 不是单纯的绝对值。 hashLoadFactorCap = 400 // 分裂阈值上限 (窄键沿用此历史值) hashMergeRatio = 4 // 合并阈值 = 分裂阈值 / 4, 与分裂阈值构成滞后防抖区间 hashKeyMinLen = 1 hashKeyMaxLen = 4096 hashDeletedFlag = uint64(1) << 63 // pkID 最高位 = 软删除标记 ) // 元数据页字段偏移 const ( hashMetaFixedOff = page.IndexHeadSize // 32 hashMetaMagicOff = hashMetaFixedOff + 0 hashMetaLevelOff = hashMetaFixedOff + 4 hashMetaSplitOff = hashMetaFixedOff + 8 hashMetaTotalOff = hashMetaFixedOff + 12 hashMetaDirOff = hashMetaFixedOff + 16 hashMetaDirPagesOff = hashMetaFixedOff + 20 hashMetaKeyLenOff = hashMetaFixedOff + 24 hashMetaDirTailOff = hashMetaFixedOff + 28 // 目录链尾页 (原 reserved 字段) hashMetaSize = 32 ) // 桶页头部: 公共 PageHdr + BucketID + Count (next 用公共头 PageHdr.NextPage) const hashBucketHdrSize = int(page.PageHdrSize) + 8 // 20 + 4 + 4 = 28 // 桶目录: 槽位从 PageHdrSize 开始, 每槽 12 字节 [head][tail][count] const ( hashDirSlotSize = 12 hashDirSlotsPerPage = (page.IndexPageSize - int(page.PageHdrSize)) / hashDirSlotSize ) // 编译期布局校验: 桶页头必须恰好宽 28 字节 var _ [unsafe.Sizeof(hashBucketHdr{}) - 28]struct{} var _ [28 - unsafe.Sizeof(hashBucketHdr{})]struct{} var ( errHashFormatIncompatible = errors.New("hash index format incompatible: drop and recreate the index") errHashKeyLenMismatch = errors.New("hash index key length mismatch with index definition") errHashMetaInvalid = errors.New("hash index metadata invalid") errHashCorrupted = errors.New("hash index corrupted") ) // hashDirSlot 一条桶溢出链在目录槽里的完整描述。 // 链尾与链长和链头一起持久化: 回收整条桶链时无需再遍历链, (Head, Tail, Count) // 可直接交给 handle.FreeChain 一次拼接进空闲链。 type hashDirSlot struct { Head uint32 Tail uint32 Count uint32 } // hashBucketHdr 桶页头部 (从页缓冲区 unsafe 解析)。 // next 指针不再自持, 统一用公共头 Hdr.NextPage —— 与空闲链同一字段, 便于整链回收。 type hashBucketHdr struct { Hdr page.PageHdr BucketID uint32 Count uint32 } // hashMetaData Hash 索引元数据 (内存表示) type hashMetaData struct { Level uint32 // 层级, 桶数基数 = 1 << Level SplitPointer uint32 // 本轮已分裂桶数, 0 <= SplitPointer < 1<= 0 恒真」而永不合并。 func hashMergeThreshold(md *hashMetaData) uint64 { if t := hashSplitThreshold(md) / hashMergeRatio; t >= 1 { return t } return 1 } // ============================================================================= // Hash / 路由 // ============================================================================= // hashOf 计算索引键的 64 位哈希 (FNV-1a + murmur3 雪崩混合, 低位分布均匀) // 线性哈希取的是哈希的低位, 因此雪崩混合是必要的。 func hashOf(key []byte) uint64 { const ( offset64 = 14695981039346656037 prime64 = 1099511628211 ) h := uint64(offset64) for _, b := range key { h ^= uint64(b) h *= prime64 } h ^= h >> 33 h *= 0xff51afd7ed558ccd h ^= h >> 33 h *= 0xc4ceb9fe1a85ec53 h ^= h >> 33 return h } // hashBucketIdx 线性哈希路由: 返回 [0, 2^Level + split) 内的桶号 func hashBucketIdx(h uint64, level, split uint32) uint32 { idx := uint32(h) & ((uint32(1) << level) - 1) if idx < split { idx = uint32(h) & ((uint32(1) << (level + 1)) - 1) } return idx } // ============================================================================= // 元数据页读写 // ============================================================================= func readHashMeta(handle page.HandleDb, metaPID uint32) (*hashMetaData, error) { buf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(metaPID, buf); err != nil { return nil, err } if binary.LittleEndian.Uint32(buf[hashMetaMagicOff:]) != hashMagic { return nil, errHashFormatIncompatible } md := &hashMetaData{ Level: binary.LittleEndian.Uint32(buf[hashMetaLevelOff:]), SplitPointer: binary.LittleEndian.Uint32(buf[hashMetaSplitOff:]), TotalEntries: binary.LittleEndian.Uint32(buf[hashMetaTotalOff:]), DirStartPID: binary.LittleEndian.Uint32(buf[hashMetaDirOff:]), DirTailPID: binary.LittleEndian.Uint32(buf[hashMetaDirTailOff:]), DirPageCount: binary.LittleEndian.Uint32(buf[hashMetaDirPagesOff:]), KeyLen: binary.LittleEndian.Uint32(buf[hashMetaKeyLenOff:]), } if md.Level < hashMinLevel || md.Level > 30 || md.KeyLen < hashKeyMinLen || md.KeyLen > hashKeyMaxLen || md.SplitPointer >= (uint32(1)< hashKeyMaxLen { return errHashKeyLenMismatch } pid := handle.NewPage() md := &hashMetaData{ Level: hashMinLevel, SplitPointer: 0, TotalEntries: 0, DirStartPID: 0, DirPageCount: 0, KeyLen: uint32(keyLen), EntryLen: uint32(keyLen) + 8, } if err := hashDirEnsure(handle, md, md.bucketCount()); err != nil { return err } if err := writeHashMeta(handle, pid, md); err != nil { return err } *metaPID = pid return nil } // ============================================================================= // 桶目录 (槽位数组, 多页以 PageHdr.NextPage 串链) // ============================================================================= func hashDirPageIndex(idx uint32) (uint32, uint32) { pp := uint32(hashDirSlotsPerPage) return idx / pp, idx % pp } // hashDirPage 返回第 pageIdx 个目录页的页ID与缓冲 func hashDirPage(handle page.HandleDb, md *hashMetaData, pageIdx uint32) (uint32, []byte, error) { cur := md.DirStartPID if cur == 0 { return 0, nil, errHashMetaInvalid } for i := uint32(0); i < pageIdx; i++ { buf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(cur, buf); err != nil { return 0, nil, err } cur = page.HdrOf(buf).NextPage if cur == 0 { return 0, nil, errHashMetaInvalid } } buf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(cur, buf); err != nil { return 0, nil, err } return cur, buf, nil } // hashDirSlotBase 返回桶 idx 的槽在目录页缓冲区内的字节偏移 func hashDirSlotBase(off uint32) int { return int(page.PageHdrSize) + int(off)*hashDirSlotSize } // hashDirGetSlot 读取桶 idx 的整条链描述 (head, tail, count) func hashDirGetSlot(handle page.HandleDb, md *hashMetaData, idx uint32) (hashDirSlot, error) { pi, off := hashDirPageIndex(idx) _, buf, err := hashDirPage(handle, md, pi) if err != nil { return hashDirSlot{}, err } base := hashDirSlotBase(off) return hashDirSlot{ Head: binary.LittleEndian.Uint32(buf[base:]), Tail: binary.LittleEndian.Uint32(buf[base+4:]), Count: binary.LittleEndian.Uint32(buf[base+8:]), }, nil } // hashDirGet 只取链头 (查找路径只需要链头) func hashDirGet(handle page.HandleDb, md *hashMetaData, idx uint32) (uint32, error) { s, err := hashDirGetSlot(handle, md, idx) if err != nil { return 0, err } return s.Head, nil } // hashDirSetSlot 写回桶 idx 的整条链描述 func hashDirSetSlot(handle page.HandleDb, md *hashMetaData, idx uint32, s hashDirSlot) error { pi, off := hashDirPageIndex(idx) pid, buf, err := hashDirPage(handle, md, pi) if err != nil { return err } base := hashDirSlotBase(off) binary.LittleEndian.PutUint32(buf[base:], s.Head) binary.LittleEndian.PutUint32(buf[base+4:], s.Tail) binary.LittleEndian.PutUint32(buf[base+8:], s.Count) return handle.SavePage(pid, buf) } // hashDirEnsure 保证桶目录容量 >= needSlots 个槽 func hashDirEnsure(handle page.HandleDb, md *hashMetaData, needSlots uint32) error { needPages := (needSlots + uint32(hashDirSlotsPerPage) - 1) / uint32(hashDirSlotsPerPage) for md.DirPageCount < needPages { pid := handle.NewPage() buf := make([]byte, page.IndexPageSize) hdr := page.HdrOf(buf) hdr.Type = page.IndexDataIndex hdr.PageID = pid hdr.PrePage = 0 hdr.NextPage = 0 hdr.UseSize = page.PageHdrSize if err := handle.SavePage(pid, buf); err != nil { return err } if md.DirStartPID == 0 { md.DirStartPID = pid } else { // O(1) 接到目录链尾: 尾页 ID 就存在元数据里, 不必从头遍历 tail := md.DirTailPID if tail == 0 { // 兜底: 元数据缺链尾记录时退回遍历定位 t, _, err := hashDirPage(handle, md, md.DirPageCount-1) if err != nil { return err } tail = t } lastBuf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(tail, lastBuf); err != nil { return err } page.HdrOf(lastBuf).NextPage = pid if err := handle.SavePage(tail, lastBuf); err != nil { return err } } md.DirTailPID = pid md.DirPageCount++ } return nil } // ============================================================================= // 桶页 // ============================================================================= func hashReadBucketPage(handle page.HandleDb, pid uint32) (*hashBucketHdr, []byte, error) { buf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(pid, buf); err != nil { return nil, nil, err } return (*hashBucketHdr)(unsafe.Pointer(&buf[0])), buf, nil } // hashInitBucketPage 按空桶页初始化页头 (条目区从 hashBucketHdrSize 开始)。 // next 写在公共头 PageHdr.NextPage 上 —— 与空闲链同字段, 因此整条桶链可直接 // 交给 FreeChain 拼接进空闲链, 回收时无需逐页改写。 func hashInitBucketPage(buf []byte, pid, bucketID, next uint32) { hdr := (*hashBucketHdr)(unsafe.Pointer(&buf[0])) hdr.Hdr.Type = page.IndexDataIndex hdr.Hdr.PageID = pid hdr.Hdr.PrePage = 0 hdr.Hdr.NextPage = next hdr.Hdr.UseSize = uint32(hashBucketHdrSize) hdr.BucketID = bucketID hdr.Count = 0 } // hashBucketSlots 安全计算桶页物理条目数 (UseSize 损坏时返回 0) func hashBucketSlots(hdr *hashBucketHdr, entryLen uint32) uint32 { us := hdr.Hdr.UseSize if us < uint32(hashBucketHdrSize) || us > page.IndexPageSize || entryLen == 0 { return 0 } n := (us - uint32(hashBucketHdrSize)) / entryLen limit := (page.IndexPageSize - uint32(hashBucketHdrSize)) / entryLen if n > limit { n = limit } return n } // hashGatherBucket 遍历桶链: 收集所有活跃条目(平铺, 每条 EntryLen 字节)与全部页ID // 页ID列表用于后续复用这些页, 因此空页也会被收集进来 (保持链的可达性) func hashGatherBucket(handle page.HandleDb, head uint32, entryLen uint32) (flat []byte, count int, pages []uint32, err error) { cur := head for cur != 0 { hdr, buf, rerr := hashReadBucketPage(handle, cur) if rerr != nil { return nil, 0, nil, rerr } pages = append(pages, cur) n := hashBucketSlots(hdr, entryLen) for i := uint32(0); i < n; i++ { off := hashBucketHdrSize + int(i*entryLen) ent := buf[off : off+int(entryLen)] if binary.BigEndian.Uint64(ent[int(entryLen)-8:])&hashDeletedFlag != 0 { continue } flat = append(flat, ent...) count++ } cur = hdr.Hdr.NextPage } return flat, count, pages, nil } // hashWriteChain 用页池 pool 为 entries 构建桶链, 返回链描述 (head, tail, 链长) // 与未用完的页。pool 中的页仍处于已分配状态, 页数不够时 NewPage 不会返回它们。 func hashWriteChain(handle page.HandleDb, bucketID uint32, flat []byte, count int, md *hashMetaData, pool []uint32) (head, tail uint32, chain int, leftover []uint32, err error) { capPerPage := int(md.bucketCap()) if capPerPage <= 0 { return 0, 0, 0, nil, errHashMetaInvalid } if count == 0 { return 0, 0, 0, pool, nil } numPages := (count + capPerPage - 1) / capPerPage ids := make([]uint32, numPages) for i := 0; i < numPages; i++ { if i < len(pool) { ids[i] = pool[i] } else { ids[i] = handle.NewPage() } } if numPages <= len(pool) { leftover = pool[numPages:] } entryLen := int(md.EntryLen) for i := 0; i < numPages; i++ { var next uint32 if i+1 < numPages { next = ids[i+1] } buf := make([]byte, page.IndexPageSize) hashInitBucketPage(buf, ids[i], bucketID, next) hdr := (*hashBucketHdr)(unsafe.Pointer(&buf[0])) start := i * capPerPage n := count - start if n > capPerPage { n = capPerPage } off := hashBucketHdrSize for j := 0; j < n; j++ { src := (start + j) * entryLen copy(buf[off:off+entryLen], flat[src:src+entryLen]) off += entryLen } hdr.Count = uint32(n) hdr.Hdr.UseSize = uint32(off) if err := handle.SavePage(ids[i], buf); err != nil { return 0, 0, 0, nil, err } } return ids[0], ids[numPages-1], numPages, leftover, nil } // hashLinkEmptyPages 把 pages 接到链 (head, tail, chainCount) 的尾部, 返回新的链描述。 // 有了链尾就无需遍历找尾; 空链 (head==0) 时用 pages 自建一条链。 func hashLinkEmptyPages(handle page.HandleDb, head, tail, chainCount, bucketID uint32, pages []uint32) (hashDirSlot, error) { if len(pages) == 0 { return hashDirSlot{Head: head, Tail: tail, Count: chainCount}, nil } for i, pid := range pages { var next uint32 if i+1 < len(pages) { next = pages[i+1] } buf := make([]byte, page.IndexPageSize) hashInitBucketPage(buf, pid, bucketID, next) if err := handle.SavePage(pid, buf); err != nil { return hashDirSlot{}, err } } if head == 0 { return hashDirSlot{Head: pages[0], Tail: pages[len(pages)-1], Count: uint32(len(pages))}, nil } lastBuf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(tail, lastBuf); err != nil { return hashDirSlot{}, err } page.HdrOf(lastBuf).NextPage = pages[0] if err := handle.SavePage(tail, lastBuf); err != nil { return hashDirSlot{}, err } return hashDirSlot{Head: head, Tail: pages[len(pages)-1], Count: chainCount + uint32(len(pages))}, nil } // hashScatter 按桶号重建若干条桶链: 复用 pool 中的页, 多余的页挂到最小桶号的链尾 // 分裂、合并、压缩都走这一条路径, 因此不会产生孤儿页 func hashScatter(handle page.HandleDb, md *hashMetaData, groups map[uint32][]byte, pool []uint32) error { if len(groups) == 0 { return nil } order := make([]uint32, 0, len(groups)) for k := range groups { order = append(order, k) } sort.Slice(order, func(i, j int) bool { return order[i] < order[j] }) leftover := pool for _, bid := range order { flat := groups[bid] head, tail, chain, lo, err := hashWriteChain(handle, bid, flat, len(flat)/int(md.EntryLen), md, leftover) if err != nil { return err } leftover = lo if err := hashDirSetSlot(handle, md, bid, hashDirSlot{Head: head, Tail: tail, Count: uint32(chain)}); err != nil { return err } } if len(leftover) > 0 { bid := order[0] s, err := hashDirGetSlot(handle, md, bid) if err != nil { return err } ns, err := hashLinkEmptyPages(handle, s.Head, s.Tail, s.Count, bid, leftover) if err != nil { return err } if err := hashDirSetSlot(handle, md, bid, ns); err != nil { return err } } return nil } // hashAppendEntry 向桶 idx 追加一个条目, 并同步该桶的链描述。 // 有链尾后不必再从头找空隙: 尾页未满就地追加, 满了才新建页接到链尾。 func hashAppendEntry(handle page.HandleDb, md *hashMetaData, idx uint32, entry []byte) error { s, err := hashDirGetSlot(handle, md, idx) if err != nil { return err } if s.Head == 0 { pid := handle.NewPage() buf := make([]byte, page.IndexPageSize) hashInitBucketPage(buf, pid, idx, 0) copy(buf[hashBucketHdrSize:], entry) hdr := (*hashBucketHdr)(unsafe.Pointer(&buf[0])) hdr.Count = 1 hdr.Hdr.UseSize = uint32(hashBucketHdrSize) + md.EntryLen if err := handle.SavePage(pid, buf); err != nil { return err } return hashDirSetSlot(handle, md, idx, hashDirSlot{Head: pid, Tail: pid, Count: 1}) } tailBuf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(s.Tail, tailBuf); err != nil { return err } tailHdr := (*hashBucketHdr)(unsafe.Pointer(&tailBuf[0])) if tailHdr.Hdr.UseSize+md.EntryLen <= page.IndexPageSize { copy(tailBuf[int(tailHdr.Hdr.UseSize):], entry) tailHdr.Count++ tailHdr.Hdr.UseSize += md.EntryLen return handle.SavePage(s.Tail, tailBuf) } pid := handle.NewPage() buf := make([]byte, page.IndexPageSize) hashInitBucketPage(buf, pid, idx, 0) copy(buf[hashBucketHdrSize:], entry) hdr := (*hashBucketHdr)(unsafe.Pointer(&buf[0])) hdr.Count = 1 hdr.Hdr.UseSize = uint32(hashBucketHdrSize) + md.EntryLen if err := handle.SavePage(pid, buf); err != nil { return err } tailHdr.Hdr.NextPage = pid if err := handle.SavePage(s.Tail, tailBuf); err != nil { return err } return hashDirSetSlot(handle, md, idx, hashDirSlot{Head: s.Head, Tail: pid, Count: s.Count + 1}) } // hashFindEntry 返回桶链中第一个 key 匹配且未删除的 pkID func hashFindEntry(handle page.HandleDb, head uint32, md *hashMetaData, key []byte) (uint64, bool) { entryLen := md.EntryLen keyLen := int(md.KeyLen) cur := head for cur != 0 { hdr, buf, err := hashReadBucketPage(handle, cur) if err != nil { return 0, false } n := hashBucketSlots(hdr, entryLen) for i := uint32(0); i < n; i++ { off := hashBucketHdrSize + int(i*entryLen) ent := buf[off : off+int(entryLen)] pk := binary.BigEndian.Uint64(ent[int(entryLen)-8:]) if pk&hashDeletedFlag != 0 { continue } if bytes.Equal(ent[:keyLen], key) { return pk, true } } cur = hdr.Hdr.NextPage } return 0, false } // hashCollect 遍历桶链, 对活跃且 match 命中的条目调用 emit func hashCollect(handle page.HandleDb, head uint32, md *hashMetaData, match func(key []byte, pkID uint64) bool, emit func(key []byte, pkID uint64)) error { entryLen := md.EntryLen keyLen := int(md.KeyLen) cur := head for cur != 0 { hdr, buf, err := hashReadBucketPage(handle, cur) if err != nil { return err } n := hashBucketSlots(hdr, entryLen) for i := uint32(0); i < n; i++ { off := hashBucketHdrSize + int(i*entryLen) ent := buf[off : off+int(entryLen)] pk := binary.BigEndian.Uint64(ent[int(entryLen)-8:]) if pk&hashDeletedFlag != 0 { continue } if match(ent[:keyLen], pk) { emit(ent[:keyLen], pk) } } cur = hdr.Hdr.NextPage } return nil } // hashDeleteMatch 软删除桶链中第一个 match 命中的活跃条目 // 同时统计整条链的物理条目数与活跃条目数 (用于判断是否需要压缩) func hashDeleteMatch(handle page.HandleDb, head uint32, md *hashMetaData, match func(key []byte, pk uint64) bool) (deleted bool, physical, active int, err error) { entryLen := md.EntryLen keyLen := int(md.KeyLen) cur := head for cur != 0 { hdr, buf, rerr := hashReadBucketPage(handle, cur) if rerr != nil { return false, 0, 0, rerr } n := hashBucketSlots(hdr, entryLen) marked := false for i := uint32(0); i < n; i++ { off := hashBucketHdrSize + int(i*entryLen) ent := buf[off : off+int(entryLen)] pkOff := int(entryLen) - 8 pk := binary.BigEndian.Uint64(ent[pkOff:]) physical++ if pk&hashDeletedFlag != 0 { continue } active++ if !deleted && match(ent[:keyLen], pk) { binary.BigEndian.PutUint64(ent[pkOff:], pk|hashDeletedFlag) if hdr.Count > 0 { hdr.Count-- } deleted = true marked = true } } if marked { if err := handle.SavePage(cur, buf); err != nil { return deleted, physical, active, err } } cur = hdr.Hdr.NextPage } return deleted, physical, active, nil } // ============================================================================= // 分裂 / 合并 / 压缩 // ============================================================================= // trySplit 按全局负载因子判断, 分裂 splitPointer 指向的桶 (增量, 每次最多一个) func (h *HashHandler) trySplit(handle page.HandleDb, metaPID uint32, md *hashMetaData) error { if md.bucketCount() >= hashMaxBuckets { return nil } if uint64(md.TotalEntries) <= uint64(md.bucketCount())*hashSplitThreshold(md) { return nil } src := md.SplitPointer newIdx := (uint32(1) << md.Level) + src srcHead, err := hashDirGet(handle, md, src) if err != nil { return err } flat, _, pool, err := hashGatherBucket(handle, srcHead, md.EntryLen) if err != nil { return err } // 先推进指针, 之后按新参数重新路由 md.SplitPointer++ if md.SplitPointer == (uint32(1) << md.Level) { md.Level++ md.SplitPointer = 0 } if err := hashDirEnsure(handle, md, md.bucketCount()); err != nil { return err } groups := map[uint32][]byte{src: nil, newIdx: nil} entryLen := int(md.EntryLen) for off := 0; off+entryLen <= len(flat); off += entryLen { ent := flat[off : off+entryLen] idx := hashBucketIdx(hashOf(ent[:md.KeyLen]), md.Level, md.SplitPointer) if idx != src && idx != newIdx { // 不变式被破坏: 该桶的条目只可能落到这两个桶 return errHashCorrupted } groups[idx] = append(groups[idx], ent...) } if err := hashScatter(handle, md, groups, pool); err != nil { return err } return writeHashMeta(handle, metaPID, md) } // tryMerge 按全局负载因子判断, 撤销最后一个桶并并入其伙伴桶 (每次最多一个) func (h *HashHandler) tryMerge(handle page.HandleDb, metaPID uint32, md *hashMetaData) error { if md.bucketCount() <= hashMinBucketCount { return nil } if uint64(md.TotalEntries) >= uint64(md.bucketCount())*hashMergeThreshold(md) { return nil } // 回退指针 if md.SplitPointer > 0 { md.SplitPointer-- } else { md.Level-- md.SplitPointer = (uint32(1) << md.Level) - 1 } partner := md.SplitPointer src := (uint32(1) << md.Level) + partner pHead, err := hashDirGet(handle, md, partner) if err != nil { return err } sHead, err := hashDirGet(handle, md, src) if err != nil { return err } pFlat, _, pPool, err := hashGatherBucket(handle, pHead, md.EntryLen) if err != nil { return err } sFlat, _, sPool, err := hashGatherBucket(handle, sHead, md.EntryLen) if err != nil { return err } pool := append(append(make([]uint32, 0, len(pPool)+len(sPool)), pPool...), sPool...) groups := map[uint32][]byte{src: nil, partner: nil} entryLen := int(md.EntryLen) combined := append(append(make([]byte, 0, len(pFlat)+len(sFlat)), pFlat...), sFlat...) for off := 0; off+entryLen <= len(combined); off += entryLen { ent := combined[off : off+entryLen] idx := hashBucketIdx(hashOf(ent[:md.KeyLen]), md.Level, md.SplitPointer) if idx != partner { return errHashCorrupted } groups[partner] = append(groups[partner], ent...) } if err := hashScatter(handle, md, groups, pool); err != nil { return err } return writeHashMeta(handle, metaPID, md) } // compactBucket 就地重写某个桶链 (复用原页), 控制软删除墓碑的堆积 func (h *HashHandler) compactBucket(handle page.HandleDb, md *hashMetaData, idx, head uint32) error { flat, _, pool, err := hashGatherBucket(handle, head, md.EntryLen) if err != nil { return err } return hashScatter(handle, md, map[uint32][]byte{idx: flat}, pool) } // ============================================================================= // HashHandler — 实现 IndexHandler 接口 // ============================================================================= // HashHandler 线性哈希二级索引 (IndexTypeHash) // 条目 = [key(keyLen) + pkID(8B BE)], 软删除用 pkID 最高位标记 type HashHandler struct{} func init() { Compose.RegisterHandler(page.IndexTypeHash, &HashHandler{}) } func (h *HashHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) { if *entryPage == 0 { if err := initHashTable(handle, entryPage, keyLen); err != nil { return 0, err } } md, err := readHashMeta(handle, *entryPage) if err != nil { return 0, err } if uint32(keyLen) != md.KeyLen || len(indexKey) != int(md.KeyLen) { return 0, errHashKeyLenMismatch } idx := hashBucketIdx(hashOf(indexKey), md.Level, md.SplitPointer) entry := make([]byte, md.EntryLen) copy(entry, indexKey) binary.BigEndian.PutUint64(entry[int(md.EntryLen)-8:], pkID&^hashDeletedFlag) if err := hashAppendEntry(handle, md, idx, entry); err != nil { return 0, err } md.TotalEntries++ if err := writeHashMeta(handle, *entryPage, md); err != nil { return 0, err } // 分裂错误必须上抛, 否则会留下半迁移状态却报告成功 if err := h.trySplit(handle, *entryPage, md); err != nil { return 0, err } return *entryPage, nil } func (h *HashHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int, resolve RowResolver) ([]byte, bool) { if entryPage == 0 { return nil, false } md, err := readHashMeta(handle, entryPage) if err != nil { return nil, false } if uint32(keyLen) != md.KeyLen || len(indexKey) != int(md.KeyLen) { return nil, false } idx := hashBucketIdx(hashOf(indexKey), md.Level, md.SplitPointer) head, err := hashDirGet(handle, md, idx) if err != nil || head == 0 { return nil, false } pk, found := hashFindEntry(handle, head, md, indexKey) if !found { return nil, false } // 二级索引不存行数据: 回表由索引自己完成 if resolve != nil { if row, ok := resolve(pk); ok { return row, true } return nil, false } buf := make([]byte, 8) binary.BigEndian.PutUint64(buf, pk) return buf, true } // SearchAll 返回该 key 的全部 pkID (二级索引允许多行同 key) // 说明: 不在 IndexHandler 接口内, 作为 hash 索引的能力扩展提供给上层按需断言 func (h *HashHandler) SearchAll(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int) ([]uint64, bool) { if entryPage == 0 { return nil, false } md, err := readHashMeta(handle, entryPage) if err != nil { return nil, false } if uint32(keyLen) != md.KeyLen || len(indexKey) != int(md.KeyLen) { return nil, false } idx := hashBucketIdx(hashOf(indexKey), md.Level, md.SplitPointer) head, err := hashDirGet(handle, md, idx) if err != nil || head == 0 { return nil, false } var out []uint64 err = hashCollect(handle, head, md, func(k []byte, _ uint64) bool { return bytes.Equal(k, indexKey) }, func(_ []byte, pk uint64) { out = append(out, pk) }) if err != nil || len(out) == 0 { return nil, false } return out, true } // deleteWith 在匹配 key 的桶链上软删除第一个 match 命中的条目, 并完成 // 计数维护 / 墓碑压缩 / 桶合并 等收尾工作 func (h *HashHandler) deleteWith(handle page.HandleDb, entryPage *uint32, indexKey []byte, keyLen int, match func(key []byte, pk uint64) bool) error { if *entryPage == 0 { return page.ErrNotFound } md, err := readHashMeta(handle, *entryPage) if err != nil { return err } if uint32(keyLen) != md.KeyLen || len(indexKey) != int(md.KeyLen) { return page.ErrNotFound } idx := hashBucketIdx(hashOf(indexKey), md.Level, md.SplitPointer) head, err := hashDirGet(handle, md, idx) if err != nil { return err } if head == 0 { return page.ErrNotFound } deleted, physical, active, err := hashDeleteMatch(handle, head, md, match) if err != nil { return err } if !deleted { return page.ErrNotFound } if md.TotalEntries > 0 { md.TotalEntries-- } // 墓碑占多数且超过一页容量时就地压缩 if active*2 < physical && physical > int(md.bucketCap()) { if err := h.compactBucket(handle, md, idx, head); err != nil { return err } } if err := writeHashMeta(handle, *entryPage, md); err != nil { return err } if err := h.tryMerge(handle, *entryPage, md); err != nil { return err } return nil } func (h *HashHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, keyLen int) error { // 二级索引允许重复 key, 用 (key, pkID) 精确定位本行的条目; pkID 未知(为 0)时按 key 删第一个 return h.deleteWith(handle, entryPage, indexKey, keyLen, func(k []byte, pk uint64) bool { return bytes.Equal(k, indexKey) && (pkID == 0 || pk == pkID) }) } func (h *HashHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64, keyLen int) error { // 二级索引条目只存 [key + pkID]: 键未变则无事可做, 键变了则删旧插新 if bytes.Equal(oldKey, newKey) { return nil } if oldKey != nil { if err := h.Delete(handle, entryPage, oldKey, pkID, len(oldKey)); err != nil && err != page.ErrNotFound { return err } } _, err := h.Insert(handle, entryPage, newKey, pkID, rowData, len(newKey)) return err } // IsRowStore Hash 二级索引不承载行数据 func (h *HashHandler) IsRowStore() bool { return false } // ResolveRow Hash 二级索引无法按 pkID 取行, 由调用方的回表函数负责 func (h *HashHandler) ResolveRow(handle page.HandleDb, entryPage uint32, pkID uint64, cache *[]DPRange) ([]byte, bool) { return nil, false } func (h *HashHandler) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 { // Hash 二级索引没有数据页 (叶层直接存 pkID) return nil } func (h *HashHandler) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte { return nil } // CollectPageIDs 实现 index.PageCollector: 枚举该 Hash 索引占用的全部页 —— // 元数据页 + 桶目录页链 + 每个桶的溢出链。 // // 供索引整体被换根后归还旧结构。上层无法自己做这件事: 哈希的三种页 (元数据/ // 目录/桶) 头部各不相同, 按 B+ 节点的 IndexHead 去解析会把 BucketID/NextPage // 当成 EntryLen/IsLeaf, 遍历出一串垃圾页号 —— 而那些页号很可能仍然可达, 还进 // 空闲链表就是静默数据损坏。 // // 任何一步解析失败都立即停手并只返回已确认的部分 (宁可泄漏): 例如元数据魔数 // 对不上 (旧版本布局 / 页损坏) 时直接返回 nil, 连元数据页都不还。 func (h *HashHandler) CollectPageIDs(handle page.HandleDb, entryPage uint32) []uint32 { if entryPage == 0 { return nil } md, err := readHashMeta(handle, entryPage) if err != nil { return nil } var out []uint32 seen := make(map[uint32]bool) add := func(pid uint32) bool { if pid == 0 || seen[pid] { return false } seen[pid] = true out = append(out, pid) return true } add(entryPage) // 同一桶的溢出页统一用公共头 PageHdr.NextPage 串链 walkBucket := func(head uint32) { for add(head) { hdr, _, rerr := hashReadBucketPage(handle, head) if rerr != nil { return } head = hdr.Hdr.NextPage } } // 目录页之间用 PageHdr.NextPage 串链; 顺带把各槽位指向的桶链走完。 // 桶数之外的槽位是 0 (目录页按零值初始化), 会被 add 跳过。 dir := md.DirStartPID for dir != 0 && add(dir) { buf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(dir, buf); err != nil { break } for slot := 0; slot < hashDirSlotsPerPage; slot++ { off := int(page.PageHdrSize) + slot*hashDirSlotSize if off+hashDirSlotSize > len(buf) { break } walkBucket(binary.LittleEndian.Uint32(buf[off:])) } dir = page.HdrOf(buf).NextPage } return out } // CollectChains 实现 index.ChainCollector: 以「整链」而非「逐页」汇报该 Hash 索引 // 占用的全部页 —— 一条链 = (head, tail, 链长)。回收方对每条链只需一次 FreeChain // (O(1) 拼接进空闲链), 完全不必读链内任何一页。 // // 之所以能这么做: 桶页的 next 已经统一到公共头 PageHdr.NextPage, 与空闲链同一 // 字段; 且每条桶链的 (head, tail, count) 都持久化在目录槽里。因此本方法只读 // 元数据页 + 桶目录页, 成本与桶链长度无关。 func (h *HashHandler) CollectChains(handle page.HandleDb, entryPage uint32) []PageChain { if entryPage == 0 { return nil } md, err := readHashMeta(handle, entryPage) if err != nil { return nil } out := make([]PageChain, 0, 8) seen := make(map[uint32]bool) addChain := func(head, tail, count uint32) { if head == 0 || count == 0 { return } if seen[head] { return } seen[head] = true if tail == 0 { tail = head } out = append(out, PageChain{Head: head, Tail: tail, Count: count}) } // 元数据页: 单页链 addChain(entryPage, entryPage, 1) // 桶目录页链 + 每条桶溢出链 (链头/链尾/链长全部来自目录槽, 不读桶页) if md.DirStartPID != 0 && md.DirPageCount > 0 { dirTail := md.DirTailPID if dirTail == 0 { dirTail = md.DirStartPID } addChain(md.DirStartPID, dirTail, md.DirPageCount) dir := md.DirStartPID for dir != 0 { buf := make([]byte, page.IndexPageSize) if err := handle.ReadPage(dir, buf); err != nil { break } for slot := 0; slot < hashDirSlotsPerPage; slot++ { off := int(page.PageHdrSize) + slot*hashDirSlotSize if off+hashDirSlotSize > len(buf) { break } addChain(binary.LittleEndian.Uint32(buf[off:]), binary.LittleEndian.Uint32(buf[off+4:]), binary.LittleEndian.Uint32(buf[off+8:])) } next := page.HdrOf(buf).NextPage if next == md.DirStartPID { break } dir = next } } return out } // RangeScan 范围扫描 // 哈希不保序, 但条目内保存了原始 key, 因此这里做: // - 等值条件: 直接定位到桶 (O(链长)) // - 其它条件: 扫描全部桶 + 键边界过滤, 再按 key 排序 (保证 ORDER BY 语义正确) // // 每个回调参数为 8 字节 BigEndian 的 pkID (二级索引契约) func (h *HashHandler) RangeScan(handle page.HandleDb, entryPage uint32, startKey, endKey []byte, startInclusive, endInclusive bool, reverse bool, callback func(row []byte) bool) error { if entryPage == 0 { return nil } md, err := readHashMeta(handle, entryPage) if err != nil { return err } inRange := func(k []byte) bool { if startKey != nil { c := bytes.Compare(k, startKey) if c < 0 || (c == 0 && !startInclusive) { return false } } if endKey != nil { c := bytes.Compare(k, endKey) if c > 0 || (c == 0 && !endInclusive) { return false } } return true } type kvPair struct { key []byte pk uint64 } var pairs []kvPair collect := func(head uint32) error { return hashCollect(handle, head, md, func(k []byte, _ uint64) bool { return inRange(k) }, func(k []byte, pk uint64) { pairs = append(pairs, kvPair{key: append([]byte(nil), k...), pk: pk}) }) } equality := startKey != nil && endKey != nil && startInclusive && endInclusive && bytes.Equal(startKey, endKey) if equality { idx := hashBucketIdx(hashOf(startKey), md.Level, md.SplitPointer) head, gerr := hashDirGet(handle, md, idx) if gerr != nil { return gerr } if head != 0 { if cerr := collect(head); cerr != nil { return cerr } } } else { n := md.bucketCount() for i := uint32(0); i < n; i++ { head, gerr := hashDirGet(handle, md, i) if gerr != nil { return gerr } if head == 0 { continue } if cerr := collect(head); cerr != nil { return cerr } } sort.Slice(pairs, func(i, j int) bool { c := bytes.Compare(pairs[i].key, pairs[j].key) if reverse { return c > 0 } return c < 0 }) } buf := make([]byte, 8) for _, p := range pairs { binary.BigEndian.PutUint64(buf, p.pk) if !callback(buf) { return nil } } return nil }