如何用MongoDB高效聚合管道按条件获取多组最新数据？

示例数据 #

[
  {
    "_id": 1,
    "category": "FIRE",
    "time": "2024-05-11T07:11:00Z"
  },
  {
    "_id": 2,
    "category": "FIRE",
    "time": "2024-05-11T08:11:00Z"
  },
  {
    "_id": 3,
    "category": "FIRE",
    "time": "2024-05-11T09:11:00Z"
  },
  {
    "_id": 4,
    "category": "POLICE",
    "time": "2024-05-11T07:22:00Z"
  },
  {
    "_id": 5,
    "category": "POLICE",
    "time": "2024-05-11T08:22:00Z"
  },
  {
    "_id": 6,
    "category": "POLICE",
    "time": "2024-05-11T09:22:00Z"
  },
  {
    "_id": 7,
    "category": "AMBULANCE",
    "time": "2024-05-11T07:33:00Z"
  },
  {
    "_id": 8,
    "category": "AMBULANCE",
    "time": "2024-05-11T08:33:00Z"
  },
  {
    "_id": 9,
    "category": "AMBULANCE",
    "time": "2024-05-11T09:33:00Z"
  }
]

查询需求 #

针对指定类别集合（如["FIRE", "AMBULANCE"]）与时间戳集合（如["2024-05-11T08:15:00Z", "2024-05-11T09:00:00Z"]）的所有组合，获取每个类别在对应时间戳或之前的最新条目。当前已创建[category, time]复合索引，要求用单个高效聚合管道实现。

预期输出 #

[
  {
    "category": "FIRE",
    "time": "2024-05-11T08:15:00Z",
    "last_entry_on_or_before": {
      "_id": 2,
      "category": "FIRE",
      "time": "2024-05-11T08:11:00Z"
    }
  },
  {
    "category": "FIRE",
    "time": "2024-05-11T09:00:00Z",
    "last_entry_on_or_before": {
      "_id": 2,
      "category": "FIRE",
      "time": "2024-05-11T08:11:00Z"
    }
  },
  {
    "category": "AMBULANCE",
    "time": "2024-05-11T08:15:00Z",
    "last_entry_on_or_before": {
      "_id": 7,
      "category": "AMBULANCE",
      "time": "2024-05-11T07:33:00Z"
    }
  },
  {
    "category": "AMBULANCE",
    "time": "2024-05-11T09:00:00Z",
    "last_entry_on_or_before": {
      "_id": 8,
      "category": "AMBULANCE",
      "time": "2024-05-11T08:33:00Z"
    }
  }
]

实现方案 #

完全可以通过单个高效聚合管道实现，且能充分利用已创建的[category, time]复合索引。以下是具体的聚合管道代码：

db.collection.aggregate([
  // 1. 过滤目标类别数据，利用复合索引快速筛选
  {
    $match: {
      category: { $in: ["FIRE", "AMBULANCE"] }
    }
  },
  // 2. 按category和time升序排序，利用索引避免内存排序
  {
    $sort: { category: 1, time: 1 }
  },
  // 3. 按category分组，将每个类别的条目按时间顺序存入数组
  {
    $group: {
      _id: "$category",
      entries: { $push: "$$ROOT" }
    }
  },
  // 4. 生成目标类别与指定时间戳的所有组合
  {
    $crossJoin: {
      timestamps: ["2024-05-11T08:15:00Z", "2024-05-11T09:00:00Z"]
    }
  },
  // 5. 筛选出当前时间戳或之前的条目，取最新的一条
  {
    $project: {
      category: "$_id",
      time: "$timestamps",
      last_entry_on_or_before: {
        $arrayElemAt: [
          {
            $filter: {
              input: "$entries",
              cond: { $lte: ["$$this.time", "$timestamps"] }
            }
          },
          -1
        ]
      },
      _id: 0
    }
  }
])

管道阶段说明 #

• $match：精准过滤目标类别，[category, time]索引会加速这一步的查询，避免全表扫描。
• $sort：借助复合索引的有序性，MongoDB可以直接利用索引返回排序后的结果，无需在内存中进行排序操作，性能更优。
• $group：将同一类别的所有条目按时间顺序存入数组，方便后续筛选。
• $crossJoin：生成指定类别和时间戳的所有组合，确保每个组合都能被处理。
• $project：通过$filter筛选出当前时间戳之前的所有条目，再用$arrayElemAt取最后一个（即最新的），最后整理成预期的输出格式。

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