聚合

主线位置：查询主线第四、五步的另一种收集器。把 TopDocs 换成 AggregationCollector，同一个查询就从「前 10 篇是谁」变成「这些文档按某个值分桶、每桶多少篇、平均多长」。它照样走 for_segment → 每篇 collect → harvest → merge_fruits，只是每个段读的是快速字段，交出来的果实是一棵可以合并的中间结果树。

flowchart TB
  json["请求 JSON"] --> req["Aggregations"]
  req --> col["AggregationCollector"]
  col --> seg["每段 for_segment"]
  ff[".fast 列"] --> seg
  seg --> c["collect_block 分桶累计"]
  c --> h["harvest 中间结果"]
  h --> m["merge_fruits 各段合并"]
  m --> fin["into_final_result"]
  fin --> out["结果 JSON"]

src/aggregation/ 是 src/ 里最大的目录，0.27.0 有 36320 行（含测试）。本页只走主线，不逐个讲聚合类型。实验在 exp-tantivy/srchb-lab：wiki.json 1000 篇，initial（标题首字母大写，STRING | FAST）和 blen（正文字节数，u64 FAST）两个快速字段，用 NoMergePolicy 每 250 篇提交一次，得到 4 个段；对照组链接 0.26.1。平台 M2 Max、macOS 27、rustc 1.99.0，release。

请求：Elasticsearch 的 JSON

聚合请求的类型是 Aggregations，一个「名字 → 聚合」的哈希表。每个 Aggregation 是聚合本身加可选的子聚合 aggs；聚合本身是 AggregationVariants，用 serde 按键名区分。键名和参数照搬 Elasticsearch，serde_json::from_value 直接得到请求。写错类型名时 serde 的报错正好列出全部 17 种：

unknown agg type: Err("unknown variant `termz`, expected one of `range`, `histogram`, `date_histogram`, `terms`, `filter`, `composite`, `multi_terms`, `avg`, `value_count`, `max`, `min`, `stats`, `extended_stats`, `sum`, `percentiles`, `top_hits`, `cardinality`")

前 7 种是分桶（bucket），可以带子聚合；后 10 种是指标（metric）。multi_terms 是 0.27.0 新增的（CHANGELOG 记为 #2984）。aggregation/mod.rs 的模块文档还只列到 TopHits，也还写着 AggregationCollector::from_aggs(agg_req, None)，而第二个参数早已是 AggContextParams（限额、分词器、计算列），一般传 Default::default()。

聚合只读快速字段。字段没开 FAST 时，在第一个段的 for_segment 上报错：

terms on title: Err("An invalid argument was passed: 'Field \"title\" is not configured as fast field'")

想在查询之前就检查，0.27.0 新增了 validate_aggregation_fields_exist。

请求如下：按首字母分桶取前 3 桶、每桶算平均长度；全体长度的统计；长度按 5000 字节分桶。

  let req = json!({
    "by_initial": {
      "terms": { "field": "initial", "size": 3 },
      "aggs": { "avg_len": { "avg": { "field": "blen" } } }
    },
    "len_stats": { "stats": { "field": "blen" } },
    "len_hist": { "histogram": { "field": "blen", "interval": 5000 } }
  });
  let aggs: Aggregations = serde_json::from_value(req.clone())?;
  let res = s.search(&AllQuery, &AggregationCollector::from_aggs(aggs.clone(), Default::default()))?;
  println!("{}", serde_json::to_string_pretty(&res)?);

结果也是 Elasticsearch 的格式：

{
  "by_initial": {
    "buckets": [
      {
        "key": "S",
        "doc_count": 95,
        "avg_len": {
          "value": 1003.7263157894737
        }
      },
      {
        "key": "M",
        "doc_count": 72,
        "avg_len": {
          "value": 1193.9166666666667
        }
      },
      {
        "key": "A",
        "doc_count": 70,
        "avg_len": {
          "value": 1127.2857142857142
        }
      }
    ],
    "sum_other_doc_count": 763,
    "doc_count_error_upper_bound": 0
  },
  "len_hist": {
    "buckets": [
      {
        "key": 0.0,
        "doc_count": 967
      },
      {
        "key": 5000.0,
        "doc_count": 26
      },
      {
        "key": 10000.0,
        "doc_count": 5
      },
      {
        "key": 15000.0,
        "doc_count": 2
      }
    ]
  },
  "len_stats": {
    "count": 1000,
    "sum": 1048829.0,
    "min": 0.0,
    "max": 18363.0,
    "avg": 1048.829
  }
}

查询换成 river（标题字段）就只在 2 篇命中的文档上分桶，和 Count 组成元组一次拿到两样：

query river: 2 docs, by_initial = {"buckets":[{"avg_len":{"value":1725.0},"doc_count":1,"key":"G"},{"avg_len":{"value":2083.0},"doc_count":1,"key":"W"}],"doc_count_error_upper_bound":0,"sum_other_doc_count":0}

每个段：读列、分桶、累计

AggregationCollector 是一个普通的收集器，requires_scoring 返回 false，所以查询不算 BM25，文档按 64 篇一块交给 collect_block。每个段上，from_agg_req_and_reader 做两步：

1. build_aggregations_data_from_req 把请求树和这个段对上：按字段名打开快速字段的列（JSON 字段的同一路径在不同段里可能是不同类型的列），检查类型，得到一棵 AggNode 树。
2. build_segment_agg_collectors 为每个节点建一个 SegmentAggregationCollector。

这些收集器的接口和普通的 SegmentCollector 不同：collect(parent_bucket_id, docs, ..) 多一个父桶号。桶号 BucketId 是一个 u32，在一个聚合里连续分配（terms 聚合每遇到一个新词给一个号）；子聚合不是每个桶一份对象，而是一个收集器、内部用桶号下标的 Vec 存每个桶的累计值。上例里 avg_len 只有一个收集器，by_initial 每把一篇文档分进某个桶，就把（桶号，文档号）交给它。父聚合先把这些对缓存起来，攒到 FLUSH_THRESHOLD = 2048 篇再按桶成批交给子聚合，减少动态分发。

指标聚合在段内只记能合并的量。avg 的中间结果 IntermediateAverage 里是一个 IntermediateStats：个数、和、Kahan 求和的补偿项、最小、最大。平均值要到最后一步才除，各段的平均值是不能直接平均的。

合并：中间结果树

harvest 把段内状态转成 IntermediateAggregationResults：还是「名字 → 结果」的表，桶的键换成和列类型无关的 IntermediateKey（字符串、u64、i64、f64、布尔、IP），因为不同段的字典序号、不同段的列类型都不能直接比。merge_fruits 取出一个段的结果，把其余的逐个 merge_fruits 进去：同名的递归合并，terms 的桶按键相加（IntermediateTermBucketResult::merge 第一次合并时建一张键到下标的表，之后每个桶只查一次），统计量相加。最后 into_final_result 按请求做收尾：terms 按 doc_count 或键排序、截到 size、算 sum_other_doc_count，平均值做除法，再检查桶总数没超过 DEFAULT_BUCKET_LIMIT（65000）。内存另有一个默认 500 MB 的上限（DEFAULT_MEMORY_LIMIT），在收集过程中检查。

AggregationCollector 的 merge_fruits 在合并之后立刻调 into_final_result；DistributedAggregationCollector 停在中间结果，留给调用方跨索引、跨机器再合并；模块文档的说法是「When the data is distributed on different Index instances」用它。中间结果实现了 serde，但模块文档写着不要用 JSON（「Postcard is a good choice」），实测确实不行，同时验证一下中间结果手动收尾和直接用 AggregationCollector 一致：

intermediate keys: ["by_initial", "len_hist", "len_stats"]
intermediate as JSON: Err("key must be a string")
into_final_result == AggregationCollector: true

JSON 的对象键只能是字符串，而中间结果里的哈希表键是 IntermediateKey 这样的枚举。

[{"doc_count":2,"key":1},{"doc_count":1,"key":-1},{"doc_count":1,"key":18446744073709551615}]

[{"doc_count":1,"key":1},{"doc_count":1,"key":18446744073709551615},{"doc_count":1,"key":1},{"doc_count":1,"key":-1}]

terms 的近似：segment_size

上面 by_initial 的 doc_count_error_upper_bound 是 0，因为每个段交出的桶足够多。terms 聚合在每个段上只交出前 segment_size 个桶（segment_size，也认 Elasticsearch 的 shard_size 和 split_size 这两个别名），默认是 size * 10，至少等于 size。被某个段截掉的词，在别的段里的计数合并上来，就可能少算或整个落选；doc_count_error_upper_bound 是各段「被截掉的桶里最大的那个 doc_count」之和，最终结果里的计数最多可能少这么多。

这就是 Elasticsearch 在分片上的做法，tantivy 把「分片」换成了「段」：一个索引自己内部就有好几个段，单机也要面对同样的近似。默认倍数和 Elasticsearch 不同，后者的 shard_size 默认是 size × 1.5 + 10。

在 jitexpr feature 下，聚合还可以不读某个现成的快速字段，而是读一个用 Cranelift JIT 编译的表达式算出来的「计算列」（JitExprValueSourceProvider，0.27.0 新增，实验性），本卷不展开。

动脑筋答案

不是。segment_size 为 3 时 M 只剩 61，A 被挤出前三，换成了 T；误差上界报 68：

segment_size=3: {"t":{"buckets":[{"key":"S","doc_count":95},{"key":"M","doc_count":61},{"key":"T","doc_count":43}],"sum_other_doc_count":801,"doc_count_error_upper_bound":68}}
segment_size=30: {"t":{"buckets":[{"key":"S","doc_count":95},{"key":"M","doc_count":72},{"key":"A","doc_count":70}],"sum_other_doc_count":763,"doc_count_error_upper_bound":0}}

每个段只交出本段的前 3 个首字母，四个段的前三名分别是 S/T/M、S/A/M、S/L/T、S/A/M。M 在第三个段里落选，那个段的 11 篇没有上报，总数从 72 掉到 61；A 在第一、三个段落选，只剩 42 篇，比同样只有两个段上报的 T（43）少一篇，排名也错了。返回的 68 是四个段各自第 4 名的计数之和（15 + 20 + 17 + 16），真实误差 11 在这个上界之内。设成 30（也就是默认的 size * 10）时，每个段的 26 到 28 个首字母全部交出，没有桶被截掉，误差上界为 0。segment_size 越大越准，代价是每个段交出、合并的桶越多。

小结

• 请求和结果都是 Elasticsearch 的 JSON：Aggregations 是「名字 → 聚合」，17 种类型用 serde 区分，分桶聚合可以嵌套子聚合。只读快速字段。
• AggregationCollector 是普通收集器，不算分；每个段上先把请求树和这个段的列对上，再为每个节点建一个按桶号存状态的段收集器，父聚合按 2048 篇一批把（桶号，文档）交给子聚合。
• 段内只记能合并的量（avg 记个数和总和）；harvest 交出用 IntermediateKey 作键的中间结果树，各段逐个合并，最后 into_final_result 排序、截断、做除法、查桶数上限。DistributedAggregationCollector 停在中间结果，供跨索引合并；中间结果不能用 JSON 序列化。
• terms 每段只交 segment_size（默认 size × 10）个桶，和 Elasticsearch 的 shard_size 同理；太小会少算、排错，doc_count_error_upper_bound 给出上界。
• 0.27.0 修了 JSON 字段里同一个数值因列类型不同而在合并时分成两个桶的问题（#3051）。jitexpr 让聚合可以读 JIT 编译的计算列，属实验特性。