MongoDB层级文档中工作流流识别的聚合查询实现

需求说明 #

需编写MongoDB聚合查询，处理包含父子层级关系的活动集合文档，识别出所有从根节点到叶子节点的完整工作流路径，并按原文档分组输出。

输入文档 #

[
  {
    "id": "123",
    "activities": [
      {
        "activityId": "A1",
        "children": [
          { "activityId": "A2" },
          { "activityId": "A7" }
        ]
      },
      {
        "activityId": "A2",
        "children": [ { "activityId": "A3" } ]
      },
      {
        "activityId": "A3",
        "children": [ { "activityId": "A4" } ]
      },
      {
        "activityId": "A4",
        "children": [ { "activityId": "A5" } ]
      },
      {
        "activityId": "A4",
        "children": [ { "activityId": "A6" } ]
      },
      { "activityId": "A6", "children": [] },
      { "activityId": "A5", "children": [] },
      { "activityId": "A7", "children": [] },
      {
        "activityId": "B1",
        "children": [
          { "activityId": "B2" },
          { "activityId": "B3" }
        ]
      },
      { "activityId": "B2", "children": [] },
      { "activityId": "B3", "children": [] },
      { "activityId": "C1", "children": [] }
    ]
  },
  {
    "id": "897",
    "activities": [
      {
        "activityId": "X1",
        "children": [
          { "activityId": "X2" },
          { "activityId": "X7" }
        ]
      },
      {
        "activityId": "X2",
        "children": [ { "activityId": "X3" } ]
      },
      {
        "activityId": "X3",
        "children": [ { "activityId": "X4" } ]
      },
      {
        "activityId": "X4",
        "children": [ { "activityId": "X5" } ]
      },
      {
        "activityId": "X4",
        "children": [ { "activityId": "X6" } ]
      },
      { "activityId": "X6", "children": [] },
      { "activityId": "X5", "children": [] },
      { "activityId": "X7", "children": [] },
      {
        "activityId": "Y1",
        "children": [
          { "activityId": "Y2" },
          { "activityId": "Y3" }
        ]
      },
      { "activityId": "Y2", "children": [] },
      { "activityId": "Y3", "children": [] },
      { "activityId": "Z1", "children": [] }
    ]
  }
]

当前聚合查询（无法得到预期结果） #

db.collection.aggregate([
  {
    "$match": {
      "activities.children": []
    }
  },
  {
    "$graphLookup": {
      "from": "collection",
      "startWith": "$activities.activityId",
      "connectFromField": "activities.activityId",
      "connectToField": "activities.children.activityId",
      "as": "workflowStreams"
    }
  },
  {
    "$set": {
      "workflowStreams": {
        "$setUnion": [
          [ "$activities.activityId" ],
          "$workflowStreams.activities.activityId"
        ]
      }
    }
  },
  {
    "$group": {
      "_id": "$id",
      "workflowStreams": { "$push": "$workflowStreams" }
    }
  }
])

预期响应 #

[
  {
    "id": "123",
    "workflowStreams": [
      [ "A1", "A2", "A3", "A4", "A6" ],
      [ "A1", "A2", "A3", "A4", "A5" ],
      [ "A1", "A7" ],
      [ "B1", "B2" ],
      [ "B1", "B3" ],
      [ "C1" ]
    ]
  },
  {
    "id": "897",
    "workflowStreams": [
      [ "X1", "X2", "X3", "X4", "X6" ],
      [ "X1", "X2", "X3", "X4", "X5" ],
      [ "X1", "X7" ],
      [ "Y1", "Y2" ],
      [ "Y1", "Y3" ],
      [ "Z1" ]
    ]
  }
]

修正后的聚合查询 #

原查询错误在于$graphLookup是跨文档查询逻辑，而所有活动都嵌套在单个文档的activities数组中，无法直接生效。以下是适配嵌套结构的解决方案：

db.collection.aggregate([
  // 拆分activities数组为独立活动条目
  { $unwind: "$activities" },
  // 构建活动映射表：记录每个活动的子节点、是否为叶子节点
  {
    $group: {
      _id: "$id",
      activityMap: {
        $push: {
          k: "$activities.activityId",
          v: {
            children: "$activities.children.activityId",
            isLeaf: { $eq: [ "$activities.children", [] ] }
          }
        }
      }
    }
  },
  // 将映射数组转为对象，方便后续访问
  { $set: { activityMap: { $arrayToObject: "$activityMap" } } },
  // 自定义递归函数生成所有根到叶子的路径
  {
    $set: {
      workflowStreams: {
        $function: {
          body: function(activityMap) {
            // 递归生成单个节点的所有下游路径
            function generatePaths(nodeId) {
              const node = activityMap[nodeId];
              if (node.isLeaf) return [[nodeId]];
              const paths = [];
              node.children.forEach(childId => {
                generatePaths(childId).forEach(path => paths.push([nodeId, ...path]));
              });
              return paths;
            }

            // 筛选所有根节点（无父节点的活动）
            const allNodes = Object.keys(activityMap);
            const childNodes = new Set();
            Object.values(activityMap).forEach(node => node.children.forEach(child => childNodes.add(child)));
            const rootNodes = allNodes.filter(node => !childNodes.has(node));

            // 生成所有根节点的完整路径
            let allPaths = [];
            rootNodes.forEach(root => allPaths = allPaths.concat(generatePaths(root)));
            return allPaths;
          },
          args: ["$activityMap"],
          lang: "js"
        }
      }
    }
  },
  // 调整输出字段格式，匹配预期结构
  { $project: { _id: 0, id: "$_id", workflowStreams: 1 } }
])

关键逻辑说明 #

1. 拆分与映射：通过$unwind和$group将嵌套的活动转为可快速访问的映射表，记录每个活动的子节点和叶子节点标识。
2. 递归路径生成：用$function自定义JS逻辑，先识别所有根节点，再递归遍历每个根节点到叶子节点的所有分支，拼接成完整工作流路径。
3. 格式适配：最后调整输出字段，移除MongoDB默认的_id，改用原文档的id字段。

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