从零构建现代搜索引擎（26）：把查询分发到多个分片

单机索引有容量天花板。内存装不下全部 HNSW 向量时，要么做量化压缩（第 22 篇），要么把索引拆到多个进程上。量化只能压缩 4 倍，而索引拆分没有理论上限。

本篇把索引拆成多个分片，用 scatter/gather 模式分发查询，在 coordinator 上合并全局 Top-K。重点处理分片内 BM25 统计偏差和跨分片评分可比性。

为什么要分片

单机瓶颈

文档数	倒排索引	HNSW(int8)	Stored fields	合计
10K	50 MB	23 MB	20 MB	~93 MB
100K	500 MB	230 MB	200 MB	~930 MB
1M	5 GB	2.3 GB	2 GB	~9.3 GB
10M	50 GB	23 GB	20 GB	~93 GB

10M 文档需要 93 GB 内存——超出大多数单机配置。即使装得下，查询延迟也随文档数增长。

分片的收益

将 10M 文档拆成 10 个分片，每个分片 1M 文档：

• 每个分片只需 ~9.3 GB 内存
• 查询延迟接近 1M 文档的水平（各分片并行执行）
• 单个分片故障不影响其他分片

固定路由

文档到分片的映射

class ShardRouter {
    private final int numShards;

    ShardRouter(int numShards) {
        this.numShards = numShards;
    }

    int route(String docId) {
        int hash = docId.hashCode() & 0x7FFFFFFF;  // 取绝对值
        return hash % numShards;
    }
}

hashCode() & 0x7FFFFFFF 确保哈希值非负——Java 的 hashCode() 可能返回负数。

固定路由的性质

• 确定性：同一 docId 始终路由到同一分片
• 均匀性：好的哈希函数使各分片文档数接近
• 无状态：不需要维护映射表，任何节点都能计算路由

写入时：router.route(docId) → 发送到对应分片。
更新/删除时：router.route(docId) → 同一分片，不需要广播。

Scatter/Gather

架构

      Client
        │
        ▼
   Coordinator
    ╱    │    ╲
   ▼     ▼     ▼
Shard0 Shard1 Shard2

Coordinator 不持有索引数据，只负责分发查询和合并结果。

Coordinator 实现

class SearchCoordinator {
    private final List<ShardClient> shards;

    SearchCoordinator(List<ShardClient> shards) {
        this.shards = shards;
    }

    SearchResult search(String queryText, int topK) {
        // Scatter: 并行发送查询到所有分片
        List<CompletableFuture<ShardResult>> futures = shards.stream()
            .map(shard -> CompletableFuture.supplyAsync(
                () -> shard.search(queryText, topK)))
            .toList();

        // Gather: 等待所有分片返回
        List<ShardResult> results = futures.stream()
            .map(f -> {
                try {
                    return f.get(200, TimeUnit.MILLISECONDS);
                } catch (Exception e) {
                    return ShardResult.EMPTY;  // 分片超时或失败
                }
            })
            .toList();

        // 合并全局 Top-K
        return mergeResults(results, topK);
    }
}

ShardClient

每个分片对外提供搜索接口。单机多分片用方法调用，跨机器用 HTTP/gRPC：

interface ShardClient {
    ShardResult search(String queryText, int topK);
}

record ShardResult(
    int shardId,
    List<DocScore> bm25Hits,
    List<DocScore> vectorHits,
    boolean partial
) {
    static final ShardResult EMPTY = new ShardResult(-1, List.of(), List.of(), true);
}

每个分片分别返回 BM25 和向量召回的结果——融合在 coordinator 做。

全局 Top-K 合并

K-way Merge

SearchResult mergeResults(List<ShardResult> shardResults, int topK) {
    // 合并 BM25 结果
    List<DocScore> allBm25 = new ArrayList<>();
    for (ShardResult sr : shardResults) {
        allBm25.addAll(sr.bm25Hits());
    }

    // 合并向量结果
    List<DocScore> allVector = new ArrayList<>();
    for (ShardResult sr : shardResults) {
        allVector.addAll(sr.vectorHits());
    }

    // 各自取全局 Top-K（按 score 降序）
    List<DocScore> globalBm25 = topK(allBm25, topK);
    List<DocScore> globalVector = topK(allVector, topK);

    // RRF 融合
    List<DocScore> fused = rrfFusion(globalBm25, globalVector, 60);

    // 重排（在 coordinator 上执行）
    List<DocScore> reranked = rerank(fused, topK);

    // 标记部分失败
    boolean partial = shardResults.stream().anyMatch(ShardResult::partial);

    return new SearchResult(reranked, partial);
}

List<DocScore> topK(List<DocScore> docs, int k) {
    PriorityQueue<DocScore> heap = new PriorityQueue<>(
        Comparator.comparingDouble(DocScore::score));
    for (DocScore doc : docs) {
        heap.offer(doc);
        if (heap.size() > k) heap.poll();
    }
    return heap.stream()
        .sorted(Comparator.comparingDouble(DocScore::score).reversed())
        .toList();
}

稳定排序

多个文档 score 相同时，排序结果不确定。用 docId 做 tiebreaker：

List<DocScore> topKStable(List<DocScore> docs, int k) {
    PriorityQueue<DocScore> heap = new PriorityQueue<>((a, b) -> {
        int cmp = Double.compare(a.score(), b.score());
        if (cmp != 0) return cmp;
        return a.docId().compareTo(b.docId());  // tiebreaker
    });
    for (DocScore doc : docs) {
        heap.offer(doc);
        if (heap.size() > k) heap.poll();
    }
    return heap.stream()
        .sorted(Comparator.comparingDouble(DocScore::score).reversed()
            .thenComparing(DocScore::docId))
        .toList();
}

没有 tiebreaker 时，分页请求的结果可能不一致——翻到第二页时，第一页的文档排序变了，导致某些文档既不在第一页也不在第二页。

分片内统计偏差

IDF 偏差

BM25 的 IDF 公式：log((N - df + 0.5) / (df + 0.5))

N 是文档总数，df 是包含该 term 的文档数。分片内的 N 和 df 是局部值。

假设 3 个分片，term “搜索” 的分布：

	分片 0	分片 1	分片 2	全局
N	10000	10000	10000	30000
df	5000	100	50	5150
IDF	0.69	4.60	5.30	1.76

分片 0 认为 “搜索” 很常见（IDF=0.69），分片 2 认为很罕见（IDF=5.30）。同一个 term 在不同分片中的权重差 7.7 倍——跨分片的 BM25 score 不可比。

实践中的处理

方案 1：忽略（默认选择）

如果文档按 docId 哈希路由，term 分布大致均匀，IDF 偏差很小。只有当文档分布严重倾斜时（如按语言分片、按时间分片）才需要处理。

方案 2：两阶段查询

SearchResult searchWithGlobalStats(String queryText, int topK) {
    // 阶段 1: 收集各分片的 term 统计
    List<CompletableFuture<TermStats>> statsFutures = shards.stream()
        .map(shard -> CompletableFuture.supplyAsync(
            () -> shard.getTermStats(queryText)))
        .toList();

    // 合并为全局统计
    TermStats globalStats = mergeTermStats(
        statsFutures.stream().map(CompletableFuture::join).toList());

    // 阶段 2: 带全局统计的查询
    List<CompletableFuture<ShardResult>> searchFutures = shards.stream()
        .map(shard -> CompletableFuture.supplyAsync(
            () -> shard.searchWithStats(queryText, topK, globalStats)))
        .toList();

    return mergeResults(
        searchFutures.stream().map(CompletableFuture::join).toList(),
        topK);
}

代价：多一次网络往返。Elasticsearch 的 dfs_query_then_fetch 就是这个方案。

向量搜索无此问题

余弦相似度和内积是绝对值——不依赖分片内统计。分片 0 的 0.85 和分片 1 的 0.82 可以直接比较。

分片上的完整搜索流程

每个分片的处理

class ShardSearchService {
    private final IndexSearcher searcher;
    private final EmbeddingClient embeddingClient;
    private final int shardId;

    ShardResult search(String queryText, int topK) {
        // BM25 召回
        Query bm25Query = parseQuery(queryText);
        TopDocs bm25Docs = searcher.search(bm25Query, topK);
        List<DocScore> bm25Hits = toDocScores(bm25Docs);

        // 向量召回
        float[] queryVec = embeddingClient.embed(
            List.of(queryText), "query")[0];
        KnnFloatVectorQuery knnQuery = new KnnFloatVectorQuery(
            "embedding", queryVec, topK);
        TopDocs knnDocs = searcher.search(knnQuery, topK);
        List<DocScore> vectorHits = toDocScores(knnDocs);

        return new ShardResult(shardId, bm25Hits, vectorHits, false);
    }
}

融合和重排不在分片上做——coordinator 需要看到所有分片的结果才能正确融合。

单分片/多分片对照

验证

void compareShardingImpact(List<String> queries, int topK) {
    for (String query : queries) {
        // 单分片（全量索引）
        SearchResult singleShard = singleShardSearch(query, topK);

        // 多分片
        SearchResult multiShard = coordinator.search(query, topK);

        // 比较 Top-K 结果的重叠度
        Set<String> singleIds = singleShard.docIds();
        Set<String> multiIds = multiShard.docIds();
        double overlap = intersection(singleIds, multiIds).size()
            / (double) topK;

        System.out.printf("query='%s' overlap=%.2f\n", query, overlap);
    }
}

预期：文档均匀分布时，overlap > 0.95——分片对结果影响很小。

当前局限

• 固定分片数——增减分片需要全量 rehash
• Coordinator 是单点——需要多个 coordinator 做负载均衡
• 分片间不共享 embedding 缓存——相同查询在每个分片都要编码一次
• 没有实现跨分片的全局统计——依赖文档均匀分布

练习

1. 将索引拆成 3 个分片，验证文档分布是否均匀
2. 实现 scatter/gather 搜索，测量多分片 vs 单分片的延迟差异
3. 构造 term 分布不均的场景，观察 IDF 偏差对排序的影响
4. 实现稳定排序，验证分页结果的一致性
5. 模拟一个分片超时，验证部分失败的降级行为

延伸阅读

• Elasticsearch: Index Sharding and Routing
• Apache Lucene: ParallelLeafReader（单机多线程搜索）
• Jeff Dean, “Achieving Rapid Response Times in Large Online Services”