Xor Filter 与 Ribbon Filter 内部剖析：用近信息论下界超越布隆过滤器的静态集合近似查询

m = n * log2(1/ε) * log2(e) ≈ 1.44 * n * log2(1/ε)   # 位
k = log2(1/ε)                                          # 最优哈希次数

A[p1] ^ A[p2] ^ A[p3] = f(x)        # 三个槽值的异或 == 该元素指纹

# 伪代码：对元素 key 取 (p1, p2, p3, fingerprint)
seg = m // 3
h   = hash64(key, seed)
p1  = h % seg                     # 落在第 0 段
p2  = (h // seg) % seg + seg      # 落在第 1 段
p3  = (h // (seg*seg)) % seg + 2*seg  # 落在第 2 段
fp  = hash64(key, seed+1) & ((1 << w) - 1)   # w 位指纹

# 剥离主循环（伪代码）
q = 度数为 1 的槽队列
while q:
    v = q.pop()
    e = v 唯一相连的活边
    live.remove(e);  stack.push((v, e))
    for p in e 的另两个端点:
        degree[p] -= 1
        if degree[p] == 1: q.push(p)
if live 非空: 失败（有环）        # 换 seed 重试
# 逆序赋值
V = [0]*m
for (v, e) in reversed(stack):
    V[v] = f(e) ^ V[e.另两个端点]

总位宽 ≈ 1.23 * n * w   ，其中 w = ceil(log2(1/ε))
每元素 ≈ 1.23 * log2(1/ε) 位

A · x = b       （所有运算在 GF(2) 上，即异或）
# 查询：对元素 y，取它对应的行（r 个列），把 x 中这些列的值异或，与 b[y] 比对

# Ribbon 构建（伪代码，GF(2) 运算）
# 1) 行按首个 1 的列号排序 → 矩阵呈带状（ribbon width r）
# 2) 正向消元：自上而下，把每行主元下方的 1 消去（只动带状窗口内）
# 3) 回代：自下而上解出 x（每一列的值）

# 查询：
def contains(y):
    cols = row_of(y)          # y 对应的 r 个列
    acc  = 0
    for c in cols: acc ^= x[c]
    return acc == fingerprint(y)

import hashlib, random
from collections import deque

def _hash(key: bytes, salt: int) -> int:
    h = hashlib.blake2b(key, digest_size=8,
                        salt=salt.to_bytes(8, 'big')).digest()
    return int.from_bytes(h[:8], 'big')

class XorFilter:
    def __init__(self, keys, w=16, scale=1.25, seed=0):
        self.w = w
        self.mask = (1 << w) - 1
        self.keys = [k if isinstance(k, bytes) else str(k).encode()
                     for k in keys]
        self.build(scale, seed)

    def _fingerprint(self, key, salt):
        return (_hash(key, salt) & self.mask) | 1   # 非零指纹

    def _positions(self, key, salt, seg):
        h = _hash(key, salt)
        p1 = h % seg
        p2 = (h // seg) % seg + seg
        p3 = (h // (seg * seg)) % seg + 2 * seg
        return p1, p2, p3

    def build(self, scale, seed):
        n = len(self.keys)
        seg = max(1, int(scale * n) // 3)
        m = seg * 3
        for attempt in range(300):
            s = seed + attempt
            pos, fp = [], []
            for k in self.keys:
                pos.append(self._positions(k, s, seg))
                fp.append(self._fingerprint(k, s + 999))
            slot_edges = [[] for _ in range(m)]
            for i, (p1, p2, p3) in enumerate(pos):
                slot_edges[p1].append(i)
                slot_edges[p2].append(i)
                slot_edges[p3].append(i)
            degree = [len(e) for e in slot_edges]
            live = set(range(n))
            q = deque(v for v in range(m) if degree[v] == 1)
            stack = []
            while q:
                v = q.popleft()
                if degree[v] != 1:
                    continue
                e = next((c for c in slot_edges[v] if c in live), None)
                if e is None:
                    continue
                live.discard(e)
                stack.append((v, e))
                p1, p2, p3 = pos[e]
                for p in (p1, p2, p3):
                    if p != v:
                        degree[p] -= 1
                        if degree[p] == 1:
                            q.append(p)
            if live:
                continue                      # 出现环，换种子重试
            V = [0] * m
            for v, e in reversed(stack):       # 逆序赋值
                p1, p2, p3 = pos[e]
                acc = fp[e]
                for p in (p1, p2, p3):
                    if p != v:
                        acc ^= V[p]
                V[v] = acc
            ok = all((V[a] ^ V[b] ^ V[c]) == fp[e]
                     for e, (a, b, c) in enumerate(pos))
            if ok:
                self.M, self.V, self.seed = m, V, s
                return
        raise RuntimeError("xor filter 构建失败：反复出现环")

    def contains(self, key):
        key = key if isinstance(key, bytes) else str(key).encode()
        seg = self.M // 3
        p1, p2, p3 = self._positions(key, self.seed, seg)
        f = self._fingerprint(key, self.seed + 999)
        return (self.V[p1] ^ self.V[p2] ^ self.V[p3]) == f


if __name__ == '__main__':
    keys = [f"user:{i}".encode() for i in range(500)]
    xf = XorFilter(keys, w=16)
    print("数组长度 M =", xf.M, " 每元素 ≈",
          round(xf.M * 16 / len(keys), 2), "位 (w=16)")
    print("命中查询 user:0   :", xf.contains(b"user:0"))     # True
    print("命中查询 user:499 :", xf.contains(b"user:499"))   # True
    # 误报率实测：100 万个非成员里的误报比例应接近 1/2^16
    fp = sum(1 for i in range(1_000_000)
             if xf.contains(f"ghost:{i}".encode()))
    print("百万非成员误报数 :", fp, " 实测 FPR ≈", fp / 1_000_000)