1
// 跳表节点
2
struct SkipNode {
3
    Key     key;
4
    Value   value;
5
    int     level;
6
    SkipNode *forward[];  // 每层的前向指针
7
};
8

9
// 跳表
10
struct SkipList {
11
    int         max_level;
12
    int         current_level;
13
    SkipNode   *header;
14
    double      probability;  // 通常 0.25
15
};
16

17
// 查找：O(log N)
18
SkipNode *skiplist_find(SkipList *list, Key key) {
19
    SkipNode *current = list->header;
20
    for (int i = list->current_level - 1; i >= 0; i--) {
21
        while (current->forward[i] && current->forward[i]->key < key) {
22
            current = current->forward[i];
23
        }
24
    }
25
    current = current->forward[0];
26
    if (current && current->key == key) return current;
27
    return NULL;
28
}
29

30
// 插入：O(log N)
31
void skiplist_insert(SkipList *list, Key key, Value value) {
32
    SkipNode *update[MAX_LEVEL];
33
    SkipNode *current = list->header;
34

35
    // 查找插入位置
36
    for (int i = list->current_level - 1; i >= 0; i--) {
37
        while (current->forward[i] && current->forward[i]->key < key) {
38
            current = current->forward[i];
39
        }
40
        update[i] = current;
41
    }
42

43
    // 随机决定层数
44
    int new_level = random_level(list->probability);
45
    SkipNode *new_node = create_node(key, value, new_level);
46

47
    // 更新指针
48
    for (int i = 0; i < new_level; i++) {
49
        new_node->forward[i] = update[i]->forward[i];
50
        update[i]->forward[i] = new_node;
51
    }
52
}

1
// SSTable 文件布局（简化）
2
// ┌─────────────────────────────────────┐
3
// │ Data Block 0  [key0, val0] ...      │ 4KB
4
// │ Data Block 1  [keyN, valN] ...      │ 4KB
5
// │ ...                                  │
6
// │ Data Block M                         │ 4KB
7
// ├─────────────────────────────────────┤
8
// │ Meta Block 0  (Bloom Filter)        │
9
// │ Meta Block 1  (Properties)          │
10
// ├─────────────────────────────────────┤
11
// │ Index Block                          │
12
// │   [block_min_key → offset, size]    │
13
// ├─────────────────────────────────────┤
14
// │ Footer (48 bytes)                    │
15
// │   meta_index_offset, index_offset   │
16
// │   magic_number                       │
17
// └─────────────────────────────────────┘
18

19
struct SSTableFooter {
20
    uint64_t  meta_index_offset;    // 元数据索引偏移
21
    uint64_t  meta_index_size;
22
    uint64_t  index_offset;         // 数据索引偏移
23
    uint64_t  index_size;
24
    uint8_t   magic[8];             // 魔数校验
25
};

1
# RocksDB 的键值编码
2
# Internal Key = User Key + Sequence Number + Type
3
# Type: Put(0) 或 Delete(1)
4

5
def encode_internal_key(user_key, seq_num, op_type):
6
    """编码内部键"""
7
    # User Key: 用户可见的键
8
    # Sequence Number: 7 字节，单调递增
9
    # Type: 1 字节，Put 或 Delete
10
    internal_key = user_key + struct.pack(">Q", (seq_num << 8) | op_type)
11
    return internal_key
12

13
# 键排序规则：
14
# 1. 按 User Key 升序
15
# 2. 相同 User Key，按 Sequence Number 降序（新的在前）
16
# 3. 相同 User Key + Sequence Number，Put 优先于 Delete
17

18
# 这保证了读取时先看到最新版本

1
# Leveled Compaction 触发条件
2
def should_compact_leveled(level, level_size, max_size):
3
    """判断是否需要 Compaction"""
4
    # L0: SSTable 数量超过 4
5
    if level == 0 and len(sstables[level]) >= 4:
6
        return True
7

8
    # L1+: 层级大小超过限制
9
    if level_size > max_size:
10
        return True
11

12
    return False
13

14
# Leveled Compaction 过程
15
def leveled_compaction(level):
16
    """Leveled Compaction"""
17
    # 1. 选择一个 SSTable 作为输入
18
    input_sst = pick_compaction_input(level)
19

20
    # 2. 找到下一层与输入范围重叠的 SSTable
21
    overlap_ssts = find_overlaps(level + 1, input_sst.key_range)
22

23
    # 3. 多路归并排序
24
    new_ssts = merge_sort([input_sst] + overlap_ssts)
25

26
    # 4. 写入新的 SSTable
27
    write_new_sstables(level + 1, new_ssts)
28

29
    # 5. 删除旧的 SSTable
30
    delete_old_sstables([input_sst] + overlap_ssts)

1
# Leveled Compaction 写放大分析
2
def leveled_write_amplification(n_levels=7, size_ratio=10):
3
    """计算 Leveled Compaction 的写放大"""
4
    total_wa = 0
5

6
    for level in range(n_levels):
7
        # 每层的数据需要被重写的次数
8
        # 数据从 L0 移到 L1 时，L1 的数据被重写
9
        # 数据从 L1 移到 L2 时，L2 的数据被重写
10
        # ...
11
        wa = size_ratio  # 每层重写约 size_ratio 倍数据
12
        total_wa += wa
13
        print(f"L{level} → L{level+1}: 写放大贡献 = {wa}x")
14

15
    print(f"\n总写放大: {total_wa}x")
16
    return total_wa
17

18
leveled_write_amplification()
19
# L0 → L1: 写放大贡献 = 10x
20
# L1 → L2: 写放大贡献 = 10x
21
# L2 → L3: 写放大贡献 = 10x
22
# L3 → L4: 写放大贡献 = 10x
23
# L4 → L5: 写放大贡献 = 10x
24
# L5 → L6: 写放大贡献 = 10x
25
# L6 → L7: 写放大贡献 = 10x
26
# 总写放大: 70x

1
# 布隆过滤器实现
2
import mmh3  # MurmurHash3
3

4
class BloomFilter:
5
    def __init__(self, expected_items, false_positive_rate=0.01):
6
        # 计算需要的位数
7
        self.size = self._optimal_size(expected_items, false_positive_rate)
8
        self.num_hashes = self._optimal_hashes(self.size, expected_items)
9
        self.bit_array = [0] * self.size
10

11
    def _optimal_size(self, n, p):
12
        """计算最优位数组大小"""
13
        # m = -(n * ln(p)) / (ln(2)^2)
14
        import math
15
        return int(-(n * math.log(p)) / (math.log(2) ** 2))
16

17
    def _optimal_hashes(self, m, n):
18
        """计算最优哈希函数数量"""
19
        # k = (m/n) * ln(2)
20
        import math
21
        return int((m / n) * math.log(2))
22

23
    def add(self, key):
24
        """添加键"""
25
        for i in range(self.num_hashes):
26
            index = mmh3.hash(str(key), i) % self.size
27
            self.bit_array[index] = 1
28

29
    def might_contain(self, key):
30
        """检查键是否可能存在"""
31
        for i in range(self.num_hashes):
32
            index = mmh3.hash(str(key), i) % self.size
33
            if self.bit_array[index] == 0:
34
                return False  # 一定不存在
35
        return True  # 可能存在（有假阳性）
36

37
# 使用示例
38
bf = BloomFilter(expected_items=100000, false_positive_rate=0.01)
39
bf.add("user:1001")
40
print(bf.might_contain("user:1001"))  # True
41
print(bf.might_contain("user:9999"))  # False（大概率）

1
# RocksDB 配置示例（Python）
2
import rocksdb
3

4
opts = rocksdb.Options()
5
opts.create_if_missing = True
6

7
# 写缓冲区（MemTable 大小）
8
opts.write_buffer_size = 64 * 1024 * 1024  # 64MB
9

10
# L0 层文件数触发 Compaction
11
opts.level0_file_num_compaction_trigger = 4
12

13
# L0 层文件数触发慢写入
14
opts.level0_slowdown_writes_trigger = 20
15
opts.level0_stop_writes_trigger = 36
16

17
# 每层大小倍数
18
opts.max_bytes_for_level_base = 256 * 1024 * 1024  # 256MB
19
opts.max_bytes_for_level_multiplier = 10
20

21
# Compaction 策略
22
opts.compaction_style = rocksdb.Options.LEVEL  # Leveled
23

24
# 布隆过滤器
25
table_opts = rocksdb.BlockBasedTableOptions()
26
table_opts.filter_policy = rocksdb.BloomFilterPolicy(10)  # 10 bits/key
27
table_opts.block_cache = rocksdb.LRUCache(512 * 1024 * 1024)  # 512MB
28
opts.table_factory = rocksdb.BlockBasedTableFactory(table_opts)
29

30
# 压缩
31
opts.compression = rocksdb.CompressionType.lz4_compression
32
opts.bottommost_compression = rocksdb.CompressionType.zstd_compression
33

34
db = rocksdb.DB("/data/rocksdb", opts)

1
# 查看 RocksDB 统计
2
# 通过 db.GetProperty() 获取
3
rocksdb.stats
4
rocksdb.compaction-stats
5
rocksdb.dbstats
6

7
# 关键指标：
8
# compaction.bytes_read:    Compaction 读取字节数
9
# compaction.bytes_written: Compaction 写入字节数
10
# stall.micros:             写入停顿时间
11
# block.cache.hit:          Block Cache 命中次数
12
# block.cache.miss:         Block Cache 未命中次数
13
# bloom.filter.useful:      Bloom Filter 有用次数