---
title: 针对 Staff/Principal Engineer 级别 AI 编程能力评估：实现高性能 LSM-Tree 架构与 MVCC 事务型 Key-Value 存储引擎，并进行系统级测试
created: 2026-03-30
updated: 2026-03-30
type: summary
tags: ["evaluation", "notes", "storage", "systems", "telegram"]
external: http://go/ui/posts/2303/
status: inbox
---

作为一个针对 **Staff/Principal Engineer** 级别设计的 AI 编程能力衡量题目，这个题目必须超越简单的算法逻辑，进入**系统工程（Systems Engineering）**的范畴。

优秀的 AI 能够处理**复杂的副作用、并发安全、状态一致性以及边缘情况（Corner Cases）**。而普通的 AI 往往只能写出“快乐路径”（Happy Path），一遇到崩溃恢复或竞态条件就彻底失效。

---

## 题目名称：LSM-MVCC 事务型存储引擎
**核心目标**：实现一个单机版的、基于 LSM-Tree 架构并支持 **MVCC（多版本并发控制）** 的 Key-Value 存储引擎。

### 题目要求（发给待测试 AI 的 Prompt）
> 请编写一个名为 `engine.py` 的 Python 模块，实现一个高性能、支持事务和崩溃恢复的 Key-Value 存储引擎。要求如下：
> 1.  **核心架构**：采用 LSM-Tree 结构。包含内存中的 **MemTable**（需保持有序）、磁盘上的 **SSTable**（有序字符串表）以及 **WAL**（预写日志）。
> 2.  **事务支持 (MVCC)**：支持快照隔离（Snapshot Isolation）。提供 `begin()` 方法开启事务，事务内的读取应基于开启时刻的快照，写入在 `commit()` 前对其他事务不可见。
> 3.  **持久化与恢复**：所有写入必须先进入 WAL。程序崩溃重启后，能从 WAL 和 SSTable 中完全恢复数据，包括未完成刷盘的 MemTable。
> 4.  **后台压缩 (Compaction)**：实现后台线程，定期将多个 SSTable 合并，清理过期的版本和已删除的标记（Tombstones）。
> 5.  **并发安全**：支持多线程并发读写，需处理好读写锁逻辑。
> 6.  **代码限制**：全部代码实现需在单个文件内，逻辑清晰，不依赖第三方数据库库（仅限 Python 标准库）。

---

## 衡量脚本：`evaluator.py`

这个脚本通过 **数据正确性、ACID 特性、崩溃恢复、并发一致性** 四个维度来评分。

```python
# /// script
# dependencies = ["pytest"]
# ///
import os
import shutil
import time
import unittest
import importlib.util
import sys
from pathlib import Path

def load_engine(file_path):
    spec = importlib.util.spec_from_file_location("engine", file_path)
    module = importlib.util.module_from_spec(spec)
    sys.modules["engine"] = module
    spec.loader.exec_module(module)
    return module.StorageEngine

class TestLSMStorage(unittest.TestCase):
    db_path = "./test_db"

    def setUp(self):
        if os.path.exists(self.db_path):
            shutil.rmtree(self.db_path)
        os.makedirs(self.db_path)
        self.Engine = load_engine(sys.argv[1])
        self.db = self.Engine(self.db_path)

    def tearDown(self):
        if hasattr(self, 'db'):
            # 尝试正常关闭
            try: self.db.close()
            except: pass
        if os.path.exists(self.db_path):
            shutil.rmtree(self.db_path)

    def test_basic_crud(self):
        """测试 1: 基础 CRUD 功能"""
        print("Running Test 1: Basic CRUD...")
        self.db.put("key1", "value1")
        self.assertEqual(self.db.get("key1"), "value1")
        self.db.delete("key1")
        self.assertIsNone(self.db.get("key1"))

    def test_transaction_isolation(self):
        """测试 2: 事务隔离性 (Snapshot Isolation)"""
        print("Running Test 2: Transaction Isolation...")
        self.db.put("k1", "v1")
        
        tx1 = self.db.begin()
        self.db.put("k1", "v2") # 外部修改
        
        self.assertEqual(tx1.get("k1"), "v1", "事务内应读取到快照版本")
        tx1.commit()
        self.assertEqual(self.db.get("k1"), "v2", "事务结束后读取最新值")

    def test_crash_recovery(self):
        """测试 3: 崩溃恢复能力"""
        print("Running Test 3: Crash Recovery...")
        self.db.put("stable_key", "value_old")
        # 模拟突然崩溃，不调用 close()，直接重新加载
        self.db.put("crash_key", "should_survive")
        
        # 重新初始化引擎，模拟重启
        db_reloaded = self.Engine(self.db_path)
        self.assertEqual(db_reloaded.get("crash_key"), "should_survive", "WAL 恢复失败")
        self.assertEqual(db_reloaded.get("stable_key"), "value_old")

    def test_compaction(self):
        """测试 4: SSTable 压缩逻辑"""
        print("Running Test 4: Compaction...")
        # 大量写入并删除，触发多次 Flush 和 Compaction
        for i in range(100):
            self.db.put(f"key_{i}", "x" * 1000)
        for i in range(90):
            self.db.delete(f"key_{i}")
        
        # 强制触发压缩 (如果 AI 实现了手动触发接口) 或等待
        if hasattr(self.db, 'manual_compact'):
            self.db.manual_compact()
        else:
            time.sleep(2) 
            
        # 检查剩余数据正确性
        self.assertEqual(self.db.get("key_95"), "x" * 1000)
        self.assertIsNone(self.db.get("key_10"))

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: uv run evaluator.py <solution_file.py>")
        sys.exit(1)
        
    suite = unittest.TestLoader().loadTestsFromTestCase(TestLSMStorage)
    result = unittest.TextTestRunner(verbosity=1).run(suite)
    
    score = (result.testsRun - len(result.failures) - len(result.errors)) / result.testsRun * 100
    print(f"\n{'='*20}")
    print(f"Final Score: {score:.2f}%")
    print(f"{'='*20}")
    
    sys.exit(0 if result.wasSuccessful() else 1)
```

---

## 为什么这个题目能体现 AI 的区别？

### 1. 架构设计的深度 (The "Junior" vs "Senior" Gap)
* **不好的 AI**：可能只会用一个简单的 Python `dict` 加上 `json.dump()`。它会忽略 LSM-Tree 的“有序”特性，导致无法进行高效的范围查询。它甚至可能不知道什么是 WAL。
* **好的 AI**：会使用 `bisect` 或 `SkipList` 实现 MemTable，并为 SSTable 设计二进制格式或按行存储的结构，同时能正确实现 **Immutable MemTable** 机制（即在刷盘时依然允许读取）。

### 2. 事务的复杂性 (The "Staff" Level Challenge)
* **不好的 AI**：事务实现通常是假的。它可能只是在事务开始时锁住整个数据库（这叫全局锁，不是 MVCC）。
* **好的 AI**：会为每个 Key 维护版本号或时间戳。它能处理读取时的版本过滤，并在 `commit` 时才合并变更，这需要非常精细的状态管理。

### 3. 并发与资源清理
* **不好的 AI**：在编写后台压缩（Compaction）线程时，经常会出现 `RuntimeError: dictionary changed size during iteration` 或者死锁。
* **好的 AI**：会使用 `threading.Condition` 或 `queue` 进行线程间通信，并优雅地处理文件句柄的释放。

### 4. 代码健壮性
* **好的 AI** 会考虑文件系统的 `fsync()` 调用，确保数据真正落盘。
* **好的 AI** 在解析 WAL 时会处理文件末尾可能存在的损坏字节。

### 如何使用：
1.  把上面的 **Prompt** 扔给 AI（比如 GPT-4, Claude 3.5, 或者你的本地模型）。
2.  将它生成的代码保存为 `engine.py`。
3.  运行命令：
    ```bash
    uv run evaluator.py engine.py
    ```
4.  **看分数**：
    * **< 50%**: 该 AI 基本不具备系统编程能力，只能写简单的脚本。
    * **50% - 80%**: 具备良好的编程基础，但在复杂的并发或持久化细节上容易出错。
    * **> 90%**: 顶级 AI，能够理解底层存储原理并写出工业级风格的代码。
