MongoDB查询不同阶段如何保留结果计数？

问题背景 #

在多租户MongoDB场景中，聚合管道需要同时返回两类关键计数：

• 租户下的总文档数（第一个$match {tenant: "1"}阶段的输出数）
• 匹配用户搜索条件的文档数（第二个$match {level: "info"}阶段的输出数）
  但由于API采用分页（$skip+$limit），无法直接从返回结果中获取这两个计数，且$facet会忽略索引导致性能问题，explain结果也因查询优化无法提供准确阶段数据。

示例事件数据 #

{ id: 1, tenant: 1, name: "Device online", level: "info", /* ... */ },
{ id: 2, tenant: 1, name: "Device offline", level: "warning", /* ... */ },
{ id: 3, tenant: 1, name: "System failure", level: "error", /* ... */ },
{ id: 4, tenant: 2, name: "Device online", level: "info", /* ... */ },
{ id: 5, tenant: 2, name: "Device online", level: "info", /* ... */ },
{ id: 6, tenant: 2, name: "Device online", level: "info", /* ... */ },
{ id: 7, tenant: 1, name: "Device online", level: "info", /* ... */ },
{ id: 8, tenant: 2, name: "Device offline", level: "warning", /* ... */ },
{ id: 9, tenant: 1, name: "Device online", level: "info", /* ... */ },
{ id: 10, tenant: 1, name: "Device online", level: "info", /* ... */ },
{ id: 11, tenant: 2, name: "Device online", level: "info", /* ... */ },
{ id: 12, tenant: 1, name: "Device online", level: "info", /* ... */ },
{ id: 13, tenant: 1, name: "Device online", level: "info", /* ... */ },
...

解决方案：用$count+$mergeObjects传递计数 #

核心思路是在聚合的早期阶段先计算两个计数，再将计数与分页后的结果合并，同时保证索引能被正常使用。

1. 针对现有管道的修改 #

将计数计算逻辑放在分页之前，通过嵌套$lookup替代顶层$facet（避免全表扫描），确保每个$match都能命中索引：

[
  // 第一步：计算租户总文档数
  { $match: { tenant: "1" } },
  { $group: { _id: null, tenantTotal: { $sum: 1 } } },
  // 第二步：关联过滤计数与分页数据的子管道
  { $lookup: {
      from: "events",
      let: { tenantTotal: "$tenantTotal" },
      pipeline: [
        // 计算搜索条件匹配的总文档数
        { $match: { tenant: "1", level: "info" } },
        { $group: { _id: null, filteredTotal: { $sum: 1 } } },
        // 关联分页数据管道
        { $lookup: {
            from: "events",
            pipeline: [
              { $match: { tenant: "1", level: "info" } },
              { $sort: { timestamp: -1 } },
              { $skip: 15 },
              { $limit: 5 },
              { $lookup: {
                  from: "users",
                  let: { id: "$actorId" },
                  pipeline: [ { $project: { id: 1, name: 1, email: 1 } } ],
                  as: "user"
                }
              },
              { $unwind: { path: "$user", preserveNullAndEmptyArrays: true } }
            ],
            as: "paginatedData"
          }
        },
        { $unwind: { path: "$filteredTotal", preserveNullAndEmptyArrays: true } }
      ],
      as: "filteredData"
    }
  },
  { $unwind: "$filteredData" },
  // 第三步：整合最终输出结构
  { $project: {
      _id: 0,
      tenantTotal: "$tenantTotal",
      filteredTotal: "$filteredData.filteredTotal",
      data: "$filteredData.paginatedData"
    }
  }
]

2. 当$lookup需要提前用于匹配的场景 #

如果需要将$lookup的结果用于后续过滤（比如根据用户属性筛选），可以调整顺序，先完成关联再计算计数：

[
  { $match: { tenant: "1" } },
  // 提前完成用户关联
  { $lookup: {
      from: "users",
      let: { id: "$actorId" },
      pipeline: [ { $project: { id: 1, name: 1, email: 1 } } ],
      as: "user"
    }
  },
  { $unwind: { path: "$user", preserveNullAndEmptyArrays: true } },
  // 收集租户所有文档并计算总数
  { $group: {
      _id: null,
      tenantTotal: { $sum: 1 },
      docs: { $push: "$$ROOT" }
    }
  },
  // 过滤匹配条件的文档并计算数量
  { $project: {
      tenantTotal: 1,
      filteredDocs: {
        $filter: {
          input: "$docs",
          cond: { $eq: ["$$this.level", "info"] }
        }
      }
    }
  },
  // 实现分页并整合结果
  { $project: {
      tenantTotal: 1,
      filteredTotal: { $size: "$filteredDocs" },
      paginatedData: {
        $slice: [
          { $sortArray: { input: "$filteredDocs", sortBy: { timestamp: -1 } } },
          15,
          5
        ]
      }
    }
  },
  { $project: { _id: 0, tenantTotal: 1, filteredTotal: 1, data: "$paginatedData" } }
]

关键注意事项 #

• 确保tenant字段有单独索引，tenant+level有复合索引，所有$match阶段都能命中索引，避免全表扫描。
• 避免使用顶层$facet，它会对每个子管道单独扫描集合，性能低下；嵌套$lookup的方式可复用过滤后的结果，提升效率。
• 数据量极大时，可拆分请求：先单独查询两个计数，再执行分页查询，虽然多一次请求，但性能更可控。

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