TsDB2/DB/core/index/hash_handler.go

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<<Level
	TotalEntries uint32 // 活跃条目总数
	DirStartPID  uint32 // 桶目录首页
	DirTailPID   uint32 // 桶目录链尾页 (整条目录链可 O(1) 回收)
	DirPageCount uint32 // 桶目录页数
	KeyLen       uint32 // 索引键长度
	EntryLen     uint32 // 条目长度 = KeyLen + 8
}

// bucketCount 当前可用桶数: 2^Level + SplitPointer
func (md *hashMetaData) bucketCount() uint32 {
	return (uint32(1) << md.Level) + md.SplitPointer
}

// bucketCap 单个桶页可容纳的条目数
func (md *hashMetaData) bucketCap() uint32 {
	if md.EntryLen == 0 {
		return 0
	}
	return uint32(page.IndexPageSize-hashBucketHdrSize) / md.EntryLen
}

// hashSplitThreshold 分裂阈值: 平均每桶活跃条目超过它才分裂一个桶。
//
// 取 min(hashLoadFactorCap, bucketCap*3/4) 而非固定 400:
//   窄键 (8B int, 单页可容 1022 条) 时 400 远未写满, 保留历史值, 分裂不激进;
//   宽键时桶页容量骤降 —— 默认 string 是 256 字节 列, 单页仅容 61 条, 若仍按
//   400 分裂, 平均每桶要摊成 6~7 页溢出链, 等值查找退化成 O(链长), 与哈希
//   索引的卖点相悖。按容量的 3/4 收敛后, 分裂发生在「单页将满」之前,
//   每个桶基本停留在一页内。
func hashSplitThreshold(md *hashMetaData) uint64 {
	cap := uint64(md.bucketCap())
	if cap == 0 {
		return hashLoadFactorCap
	}
	if t := cap * 3 / 4; t < hashLoadFactorCap {
		return t
	}
	return hashLoadFactorCap
}

// hashMergeThreshold 合并阈值: 平均每桶活跃条目低于它才撤销一个桶。
// 与分裂阈值保持 1:4 的固定比值 (历史 400/100), 中间区间不做任何增删桶,
// 避免在阈值附近反复分裂/合并抖动。下限取 1, 防止超宽键下退化为 0 导致
// 「>= 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)<<md.Level) {
		return nil, errHashMetaInvalid
	}
	md.EntryLen = md.KeyLen + 8
	return md, nil
}

func writeHashMeta(handle page.HandleDb, metaPID uint32, md *hashMetaData) error {
	buf := make([]byte, page.IndexPageSize)
	// 新分配的页可能尚未落盘, 读失败是允许的; 页头显式重建, 不依赖页内残留数据
	_ = handle.ReadPage(metaPID, buf)
	hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
	hdr.Hdr.Type = page.IndexDataIndex
	hdr.Hdr.PageID = metaPID
	hdr.Hdr.PrePage = 0
	hdr.Hdr.NextPage = 0
	hdr.Hdr.UseSize = uint32(hashMetaFixedOff + hashMetaSize)

	binary.LittleEndian.PutUint32(buf[hashMetaMagicOff:], hashMagic)
	binary.LittleEndian.PutUint32(buf[hashMetaLevelOff:], md.Level)
	binary.LittleEndian.PutUint32(buf[hashMetaSplitOff:], md.SplitPointer)
	binary.LittleEndian.PutUint32(buf[hashMetaTotalOff:], md.TotalEntries)
	binary.LittleEndian.PutUint32(buf[hashMetaDirOff:], md.DirStartPID)
	binary.LittleEndian.PutUint32(buf[hashMetaDirPagesOff:], md.DirPageCount)
	binary.LittleEndian.PutUint32(buf[hashMetaKeyLenOff:], md.KeyLen)
	binary.LittleEndian.PutUint32(buf[hashMetaDirTailOff:], md.DirTailPID)
	return handle.SavePage(metaPID, buf)
}

// initHashTable 首次初始化 Hash 表 (元数据页 + 初始桶目录)
func initHashTable(handle page.HandleDb, metaPID *uint32, keyLen int) error {
	if keyLen < hashKeyMinLen || keyLen > 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
}