MongoDB 如何统计各分组最高版本对应的文档数量

问题场景 #

需求为按指定一级字段分组，找到每组内另一字段的最大值，过滤掉组内低于该最大值的历史文档后，再按多维度分组统计文档总数。
本次测试数据集如下：

[
  { "id": 1, "scanner": "bandit", "severity": "low", "version": 1 },
  { "id": 2, "scanner": "bandit", "severity": "low", "version": 1 },
  { "id": 3, "scanner": "bandit", "severity": "medium", "version": 1 },
  { "id": 4, "scanner": "bandit", "severity": "medium", "version": 2 },
  { "id": 5, "scanner": "bandit", "severity": "high", "version": 2 },
  { "id": 6, "scanner": "semgrep", "severity": "critical", "version": 2 },
  { "id": 7, "scanner": "semgrep", "severity": "critical", "version": 3 },
  { "id": 8, "scanner": "semgrep", "severity": "info", "version": 4 },
  { "id": 9, "scanner": "semgrep", "severity": "info", "version": 4 }
]

其中bandit对应的最高version为2，semgrep对应的最高version为4，需要过滤所有低于最高version的文档，按scanner、severity维度分组计数，预期返回结果：

[
    { "scanner": "bandit", "severity": "medium", "count": 1, "version": 2 },
    { "scanner": "bandit", "severity": "high", "count": 1, "version": 2 },
    { "scanner": "semgrep", "severity": "info", "count": 2, "version": 4 }
]

兼容所有MongoDB版本的通用实现 #

通过聚合管道自关联实现，不需要依赖高版本特性，可直接运行，注意将代码中的你的集合名替换为实际集合名称：

db.你的集合名.aggregate([
  // 按一级分组字段scanner分组，计算每个scanner对应的最高version
  {
    $group: {
      _id: "$scanner",
      maxVersion: { $max: "$version" }
    }
  },
  // 关联原集合，筛选出每个scanner下version等于最高值的文档
  {
    $lookup: {
      from: "你的集合名",
      let: { currentScanner: "$_id", topVersion: "$maxVersion" },
      pipeline: [
        {
          $match: {
            $expr: {
              $and: [
                { $eq: ["$scanner", "$$currentScanner"] },
                { $eq: ["$version", "$$topVersion"] }
              ]
            }
          }
        }
      ],
      as: "validDocs"
    }
  },
  // 展开筛选后的有效文档
  { $unwind: "$validDocs" },
  // 提取有效文档为根节点，简化后续分组逻辑
  { $replaceRoot: { newRoot: "$validDocs" } },
  // 按scanner、severity维度分组统计数量
  {
    $group: {
      _id: {
        scanner: "$scanner",
        severity: "$severity"
      },
      count: { $sum: 1 },
      version: { $first: "$version" }
    }
  },
  // 调整输出字段结构，移除自动生成的_id字段
  {
    $project: {
      _id: 0,
      scanner: "$_id.scanner",
      severity: "$_id.severity",
      count: 1,
      version: 1
    }
  }
])

MongoDB 5.0+ 简化实现 #

如果使用的是5.0及以上版本，支持窗口函数，可以省去自关联步骤，执行效率更高：

db.你的集合名.aggregate([
  // 按scanner分区，计算每个分区内的最大version，附加到每条文档上
  {
    $setWindowFields: {
      partitionBy: "$scanner",
      sortBy: { version: -1 },
      output: {
        groupMaxVersion: { $max: "$version" }
      }
    }
  },
  // 过滤掉version不等于所属分组最大值的历史文档
  { $match: { $expr: { $eq: ["$version", "$groupMaxVersion"] } } },
  // 按目标维度分组计数
  {
    $group: {
      _id: { scanner: "$scanner", severity: "$severity" },
      count: { $sum: 1 },
      version: { $first: "$version" }
    }
  },
  // 调整输出结构
  {
    $project: {
      _id: 0,
      scanner: "$_id.scanner",
      severity: "$_id.severity",
      count: 1,
      version: 1
    }
  }
])

逻辑说明 #

• 核心思路是先拿到每个一级分组的阈值（最高version），再基于阈值过滤有效数据，最后做维度统计，避免直接分组时把低版本历史数据计入结果。
• 两种写法执行后返回的结果和预期完全一致，可根据实际MongoDB版本选择使用。

内容的提问来源于stack exchange，提问作者KeeperOfTheSevenKeys