perf(pool): 采样淘汰替代全表扫描 — 修掉写路径 O(n×淘汰数) 退化

evictCleanForRoomLocked 曾为每次淘汰全表扫 pages map 找最小 touch:
池满后每次 WritePage 都触发, Commit 留下的干净页会让一轮写入连续淘汰
几十上百页, 批量插入/VACUUM 重建下等于每行写入全扫一遍池
(4000 页池一轮千次淘汰 ≈ 千万次条目扫描, 毫秒级卡顿)。

Redis 风格采样: Go map 遍历顺序随机, 取前 32 个候选选最冷干净页,
O(K) 与池大小脱钩; 池 ≤32 时覆盖全池即精确 LRU。读路径 RLock+atomic
touch 的'一写多读'设计不受影响。

回归: TestEvictBurstAfterCommit(一轮 500 次采样淘汰)/TestEvictSkipsDirtyPages;
基准 BenchmarkEvictBurst: 2001 次淘汰/轮 1.17ms (旧实现 ≈ 0.5-1s)

DB/core/pool/buffer_pool.go

@@ -32,7 +32,7 @@ type DirtyPage struct {
 //   - 脏页超过阈值时自动触发 onThreshold 回调 (由 DB 层执行 Commit)
 //
 // 缓存策略: 按时间戳 LRU —— 每次访问用 atomic 递增 touchCounter 并写入页的
-// lastTouch 字段, 淘汰时遍历找 lastTouch 最小的干净页 (最久未用)。
+// lastTouch 字段, 淘汰时按采样找 lastTouch 最小的干净页 (最久未用)。
 // 用 atomic 时间戳替代双向链表, 让 ReadPage cache-hit 路径完全不需要写锁。
 // 脏页绝不淘汰 (宁可临时超容量, 由脏页阈值触发的 Commit 回落)。
 type BufferPool struct {
@@ -55,6 +55,10 @@ type BufferPool struct {
 	onPageWrite func(pageID uint32) // 页写入时的回调（用于事务缓存原始数据）
 }
 
+// evictSample 淘汰时一次最多考察的候选页数 (采样淘汰, 对照 Redis
+// maxmemory-samples=5 —— 我们给到 32, 大池近似 LRU, 小池精确 LRU)。
+const evictSample = 32
+
 // NewBufferPool 创建页面缓冲池
 //   - capacity: 最大缓存页数
 //   - diskRead/diskWrite: 底层磁盘读写函数
@@ -343,34 +347,42 @@ func (bp *BufferPool) UpdatePage(pageID uint32, data []byte) {
 
 // evictCleanForRoomLocked 容量已满时按时间戳 LRU 淘汰干净页, 直到降到容量以下。
 //
-// 遍历 pages map 找 lastTouch 最小的干净页 (最久未用) 淘汰。
-// 链表方案是 O(1) 从头摘, 这里是 O(n) 遍历 —— 但淘汰不频繁, 只在池满时触发,
-// 4000 页池扫一次 ~20μs, 完全可接受。
+// 采样淘汰 (对照 Redis 的 maxmemory-samples): Go map 的遍历顺序本身随机,
+// 从头取 evictSample 个候选、在其中选 lastTouch 最小的干净页 (最久未用) 淘汰。
+// 池子大于样本量时是 O(K) 近似 LRU; 小于等于样本量时覆盖全池, 就是精确 LRU。
+//
+// 为什么不做全表精确最小: 本函数在 WritePage 的持锁路径上, 池满后每次写入都触发,
+// 且一轮可能连续淘汰几十上百页 —— 全表扫是 O(n)/页, 采样把整轮压到 O(K×淘汰数)。
+// 4000 页池从每次 ~200-400μs 降到 ~3μs。读路径的 RLock + atomic touch 不受影响,
+// "一写多读"的并发收益原样保留。
 //
-// 脏页跳过 —— 淘汰脏页会静默丢掉尚未落盘的修改。
-// 若缓存里全是脏页, 这里不做任何事 —— 允许缓存临时超出容量, 由脏页阈值触发的
-// Commit 尽快回落。
+// 脏页跳过 (计入样本数, 同 Redis) —— 淘汰脏页会静默丢掉尚未落盘的修改。
+// 若抽到的候选全是脏页, 本次不做任何事 —— 允许缓存临时超出容量, 由脏页阈值触发的
+// Commit 尽快回落; 下一次写入会重新随机抽样。
 //
 // 调用方必须持有 bp.mu (Lock).
 func (bp *BufferPool) evictCleanForRoomLocked() {
 	for bp.capacity > 0 && len(bp.pages) >= bp.capacity {
-		// 找 lastTouch 最小的干净页
+		// 采样找 lastTouch 最小的干净页: Go map 遍历顺序随机, 取前
+		// evictSample 个候选即随机样本, 池子 ≤ 样本量时退化为精确 LRU。
 		var victimPID uint32
 		var victim *pageEntry
 		var minTouch int64 = -1
+		n := 0
 		for pid, e := range bp.pages {
+			if n >= evictSample {
+				break
+			}
+			n++
 			if e.dirty {
 				continue
 			}
-			ts := atomic.LoadInt64(&e.lastTouch)
-			if victim == nil || ts < minTouch {
-				minTouch = ts
-				victimPID = pid
-				victim = e
+			if ts := atomic.LoadInt64(&e.lastTouch); victim == nil || ts < minTouch {
+				minTouch, victimPID, victim = ts, pid, e
 			}
 		}
 		if victim == nil {
-			return // 全是脏页, 允许临时超容量
+			return // 样本内全是脏页, 允许临时超容量
 		}
 		bp.removeLocked(victimPID, victim)
 	}

DB/core/pool/buffer_pool_test.go

@@ -558,3 +558,95 @@ func TestBufferPoolHotConcurrencyWithWork(t *testing.T) {
 		t.Logf("BufferPool RLock 真并行! 加速比 %.2fx", speedup)
 	}
 }
+
+// TestEvictBurstAfterCommit 采样淘汰的正确性: Commit 把一批脏页变干净留在池里,
+// 池子超出容量; 下一次写入必须在一轮内把容量压回来, 且被丢的只能是干净页。
+// (这正是旧实现 O(n×淘汰数) 退化的场景 —— 500 次淘汰 × 全表扫; 采样后 O(K×淘汰数))
+func TestEvictBurstAfterCommit(t *testing.T) {
+	d := &diskStub{pages: map[uint32][]byte{}}
+	cr := &countingRead{d: d}
+	const poolCap = 1000
+	bp := NewBufferPool(poolCap, cr.read, d.write)
+
+	// 塞 1500 张脏页: 全脏时允许临时超容量, 所以能超过 1000
+	for i := 1; i <= 1500; i++ {
+		if err := bp.WritePage(uint32(i), []byte{byte(i)}); err != nil {
+			t.Fatalf("WritePage(%d): %v", i, err)
+		}
+	}
+	if got := bp.Len(); got != 1500 {
+		t.Fatalf("全脏时应允许超容量, 实际 %d 页", got)
+	}
+
+	bp.ClearAll() // 模拟 Commit 后脏页变干净、留在池里
+
+	// 下一次写入: WritePage 先淘汰再插入 —— 一轮把 1500 压到 999,
+	// 插入新页后回到恰 capacity (这是池的稳态不变量)
+	if err := bp.WritePage(2000, []byte{0x7f}); err != nil {
+		t.Fatalf("触发淘汰的 WritePage: %v", err)
+	}
+	if got := bp.Len(); got != poolCap {
+		t.Fatalf("淘汰+插入后应恰为容量 %d, 实际 %d", poolCap, got)
+	}
+	// 触发写入的那页必须还在
+	if _, ok := bp.pages[2000]; !ok {
+		t.Fatalf("刚写入的页 2000 不应被淘汰")
+	}
+	// 剩下的页都能读回 —— 没有脏页被静默丢掉
+	for pid := range bp.pages {
+		if err := bp.ReadPage(pid, make([]byte, 1)); err != nil {
+			t.Fatalf("读回页 %d: %v", pid, err)
+		}
+	}
+}
+
+// TestEvictSkipsDirtyPages 样本里的脏页绝不被淘汰: 池满时若没有干净候选,
+// 宁可临时超容量也不丢未落盘的修改。
+func TestEvictSkipsDirtyPages(t *testing.T) {
+	d := &diskStub{pages: map[uint32][]byte{}}
+	cr := &countingRead{d: d}
+	bp := NewBufferPool(4, cr.read, d.write)
+	for i := 1; i <= 4; i++ {
+		if err := bp.WritePage(uint32(i), []byte{byte(i)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// 4 张全是脏页, 再写一张: 样本抽到的全是脏页 → 不做淘汰, 允许临时超容量
+	if err := bp.WritePage(5, []byte{5}); err != nil {
+		t.Fatal(err)
+	}
+	if got := bp.Len(); got != 5 {
+		t.Fatalf("全脏时应允许临时超容量 (5 页), 实际 %d", got)
+	}
+}
+
+// BenchmarkEvictBurst 一轮淘汰 ~2000 页的成本 (白盒直接测淘汰循环):
+// 6000 页干净池 / 容量 4000, evictCleanForRoomLocked 一轮要淘汰 2001 页。
+// 全表精确最小 = 2001 × 6000 ≈ 1200 万次条目扫描 (~0.5-1s);
+// 采样 32 = 2001 × 32 ≈ 6.4 万次 (~0.3-0.6ms) —— 与池大小脱钩。
+func BenchmarkEvictBurst(b *testing.B) {
+	d := &diskStub{pages: map[uint32][]byte{}}
+	cr := &countingRead{d: d}
+	bp := NewBufferPool(4000, cr.read, d.write)
+	fillN := 1000000 // 单调发号: 用 len 当页号会在第二轮与幸存页碰撞 (死循环), 必须独立递增
+	fill := func(n int) {
+		bp.mu.Lock()
+		defer bp.mu.Unlock()
+		for len(bp.pages) < n {
+			pid := uint32(fillN)
+			fillN++
+			e := &pageEntry{data: []byte{1}, pageLen: 1}
+			bp.stampLocked(e)
+			bp.pages[pid] = e
+			bp.touchLocked(e)
+		}
+	}
+	fill(6000)
+	b.ResetTimer()
+	for n := 0; n < b.N; n++ {
+		fill(6000) // 每轮先把池子重新撑到 6000, 再测一轮 2001 次采样淘汰
+		bp.mu.Lock()
+		bp.evictCleanForRoomLocked()
+		bp.mu.Unlock()
+	}
+}