图数据库引擎内部剖析：从邻接表、CSR 到原生图存储与遍历执行引擎

from typing import List

class CSRGraph:
    """压缩稀疏行: 用 offsets + neighbors 两个数组连续存放邻接"""
    def __init__(self, edges: List<tuple>, n: int):
        self.n = n
        # 统计每个顶点的出度
        deg = [0] * (n + 1)
        for u, v in edges:
            deg[u + 1] += 1
        # 前缀和 -> offsets[i] 是顶点 i 的邻居在 neighbors 中的起点
        for i in range(1, n + 1):
            deg[i] += deg[i - 1]
        self.offsets = deg            # 长度 n+1
        self.neighbors = [0] * (len(edges))
        cursor = self.offsets[:]     # 临时写指针
        for u, v in edges:
            self.neighbors[cursor[u]] = v
            cursor[u] += 1

    def out_neighbors(self, u: int) -> List<int>:
        # O(1) 定位区间, 连续内存读取 -> 缓存友好
        return self.neighbors[self.offsets[u]:self.offsets[u + 1]]

g = CSRGraph([(0, 1), (0, 2), (1, 2), (2, 3)], n=4)
print(g.out_neighbors(0))   # [1, 2]
print(g.out_neighbors(3))   # []

// 原生图存储的简化结构: 顶点记录内联指向它的边区间
struct NativeGraph {
    nodes: Vec<Node>,
    // offsets[i]..offsets[i+1] 是节点 i 的邻边在 edges 中的区间
    offsets: Vec<usize>,
    edges: Vec<Edge>,
}

impl NativeGraph {
    // 取邻居即一次切片, 无索引回表
    fn neighbors(&self, v: usize) -> &[Edge] {
        &self.edges[self.offsets[v]..self.offsets[v + 1]]
    }
}

from collections import deque

def bfs(graph, start):
    """广度优先遍历, 返回 (访问顺序, 跳数)"""
    seen, order, q = {start}, [start], deque([start])
    dist = {start: 0}
    while q:
        u = q.popleft()
        for v in graph.out_neighbors(u):
            if v not in seen:
                seen.add(v); order.append(v); q.append(v)
                dist[v] = dist[u] + 1
    return order, dist

import heapq
def dijkstra(graph, start, goal):
    """带权最短路径: 优先队列驱动, 遍历即松弛边"""
    dist = {start: 0}; pq = [(0, start)]
    while pq:
        d, u = heapq.heappop(pq)
        if u == goal: return d
        if d > dist.get(u, float('inf')): continue
        for v in graph.out_neighbors(u):
            nd = d + 1   # 边权此处取 1, 可替换为权重属性
            if nd < dist.get(v, float('inf')):
                dist[v] = nd; heapq.heappush(pq, (nd, v))
    return float('inf')

def plan_match(where_selectivity, has_index, est_vertex_count):
    """极简 Cypher MATCH 优化器: 选 '索引扫描' 还是 '全图扫描'"""
    # 选择性 = 命中比例; 索引扫描代价 ~ 选择性 * log(N)
    idx_cost = where_selectivity * est_vertex_count * 0.2 + 8
    scan_cost = est_vertex_count                       # 全表扫描线性代价
    if has_index and idx_cost < scan_cost:
        return "IndexScan", idx_cost
    return "FullScan", scan_cost

# 例: 4600 万顶点, 选择性 0.001%, 有索引
print(plan_match(0.00001, True, 46_000_000))   # ('IndexScan', ~14)
# 无索引时退化为全表扫描
print(plan_match(0.5, False, 46_000_000))       # ('FullScan', 46000000)

import hashlib

class ConsistentPartitioner:
    """一致性哈希: 把顶点映射到固定数量的分区, 增删节点只动少量数据"""
    def __init__(self, replicas, partitions):
        self.parts = partitions
        self.ring = []
        for p in range(partitions):
            for r in range(replicas):
                h = int.from_bytes(hashlib.md5(b"part%d-%d" % (p, r)).digest()[:4], "big")
                self.ring.append((h, p))
        self.ring.sort()

    def partition_of(self, vid):
        h = int.from_bytes(hashlib.md5(b"v%d" % vid).digest()[:4], "big")
        # 顺时针找第一个 >= h 的虚拟节点
        for key, p in self.ring:
            if key >= h:
                return p
        return self.ring[0][1]

# 跨分区 BFS: 每次展开邻居时, 本地分区直接取, 远端入 RPC 队列
def cross_partition_bfs(graph, part, start):
    rpcs = 0; local = 0; seen = {start}; q = [start]
    while q:
        u = q.pop(0)
        for v in graph.out_neighbors(u):
            if v in seen: continue
            seen.add(v)
            if part.partition_of(v) == part.partition_of(u):
                local += 1; q.append(v)
            else:
                rpcs += 1   # 触发一次跨分区 RPC
    return len(seen), local, rpcs

part = ConsistentPartitioner(replicas=20, partitions=4)
print(part.partition_of(10086))   # 0..3 之间的分区号

def pagerank(graph, iters=20, damping=0.85):
    """在 CSR 图上迭代 PageRank, 每次沿出边分配权重"""
    n = graph.n
    pr = [1.0 / n] * n
    out_deg = [len(graph.out_neighbors(i)) for i in range(n)]
    for _ in range(iters):
        new = [0.0] * n
        for u in range(n):
            if out_deg[u] == 0:
                # 悬挂节点: 把权重均分给所有人
                share = pr[u] / n
                for v in range(n): new[v] += share
                continue
            share = damping * pr[u] / out_deg[u]
            for v in graph.out_neighbors(u):
                new[v] += share
        # 补偿随机跳转的 (1-damping) 部分
        base = (1 - damping) / n
        pr = [base + x for x in new]
    return pr

# 在前面构造的 4 点图上跑
g = CSRGraph([(0, 1), (0, 2), (1, 2), (2, 3)], n=4)
print([round(x, 3) for x in pagerank(g)])
# -> 与 PageRank 收敛值一致: 入度高的点(2,3)权重更高