MongoDB聚合查询：识别层级文档中的工作流分支

我尝试编写MongoDB聚合查询，处理包含父子关系活动的复杂文档，目标是从中识别多个工作流分支，但当前查询无法得到预期响应。

输入文档 #

{"activties": [
  { "activityId": "A1", "children": [ { "activityId": "A2" }, { "activityId": "A7" } ] },
  { "activityId": "A2", "children": [ { "activityId": "A3" } ] },
  { "activityId": "A3", "children": [ { "activityId": "A4" } ] },
  { "activityId": "A4", "children": [ { "activityId": "A5" } ] },
  { "activityId": "A5", "children": [] },
  { "activityId": "A7", "children": [] }
]}

注：文档中activties存在拼写错误，正确应为activities

当前聚合查询 #

db.collection.aggregate([
  {
    $graphLookup: {
      from: "collection",
      startWith: "$activityId",
      connectFromField: "children.activityId",
      connectToField: "activityId",
      as: "workflowstreams",
      depthField: "depth"
    }
  },
  {
    $addFields: {
      workflowstreams: {
        $map: {
          input: "$workflowstreams",
          as: "ws",
          in: {
            activityId: "$$ws.activityId",
            depth: "$$ws.depth",
            children: "$$ws.children"
          }
        }
      }
    }
  },
  {
    $addFields: {
      splittedStreams: {
        $reduce: {
          input: "$workflowstreams",
          initialValue: { streams: [], currentStream: [] },
          in: {
            $cond: {
              if: { $eq: ["$$this.children", []] },
              then: {
                streams: { $concatArrays: ["$$value.streams", [["$$value.currentStream", "$$this.activityId"]] ] },
                currentStream: []
              },
              else: {
                streams: "$$value.streams",
                currentStream: { $concatArrays: ["$$value.currentStream", ["$$this.activityId"]] }
              }
            }
          }
        }
      }
    }
  },
  {
    $project: {
      workflowstreams: {
        $concatArrays: [
          "$splittedStreams.streams",
          {
            $cond: {
              if: { $gt: [{ $size: "$splittedStreams.currentStream" }, 0] },
              then: ["$splittedStreams.currentStream"],
              else: []
            }
          }
        ]
      }
    }
  },
  {
    $addFields: {
      workflowstreams: {
        $map: {
          input: "$workflowstreams",
          as: "ws",
          in: {
            $filter: {
              input: "$$ws",
              as: "activity",
              cond: { $ne: ["$$activity", null] }
            }
          }
        }
      }
    }
  },
  {
    $addFields: {
      workflowstreams: {
        $map: {
          input: "$workflowstreams",
          as: "ws",
          in: {
            $filter: {
              input: "$$ws",
              as: "activity",
              cond: { $ne: ["$$activity", []] }
            }
          }
        }
      }
    }
  },
  {
    $group: {
      _id: null,
      workflowstreams: { $first: "$workflowstreams" }
    }
  },
  {
    $project: {
      _id: 0,
      workflowstreams: {
        $map: {
          input: "$workflowstreams",
          as: "ws",
          in: {
            $map: {
              input: "$$ws",
              as: "activity",
              in: "$$activity"
            }
          }
        }
      }
    }
  }
]);

预期响应 #

[
  { "workflowstreams": [ ["A1", "A2", "A3", "A4", "A5"], ["A1", "A7"] ] }
]

问题分析与修正查询 #

现有查询核心问题 #

1. $graphLookup逻辑错误：输入文档是单文档包含活动数组，而非集合中每个活动单独成文档。原查询startWith: "$activityId"指向根文档字段，但根文档不存在该字段，导致无法正确遍历父子关系。
2. 流拆分结构错误：$reduce中拼接路径时错误嵌套数组，导致最终结果结构混乱，无法生成扁平化的工作流分支。

修正后的聚合查询 #

db.collection.aggregate([
  // 展开活动数组，将每个活动转为独立条目
  { $unwind: "$activties" },
  // 遍历获取每个活动的所有关联节点
  {
    $graphLookup: {
      from: "collection",
      startWith: "$activties.activityId",
      connectFromField: "activties.children.activityId",
      connectToField: "activties.activityId",
      as: "relatedActivities",
      depthField: "depth"
    }
  },
  // 筛选出根节点（无父节点的活动A1）
  {
    $match: {
      $expr: {
        $not: {
          $in: ["$activties.activityId", "$relatedActivities.activties.children.activityId"]
        }
      }
    }
  },
  // 递归生成完整工作流分支
  {
    $addFields: {
      workflowstreams: {
        $function: {
          body: function(rootAct, allActs) {
            // 递归生成分支路径
            function buildBranches(activity) {
              if (!activity.children.length) return [[activity.activityId]];
              return activity.children.flatMap(child => {
                const targetAct = allActs.find(item => item.activityId === child.activityId);
                return buildBranches(targetAct).map(branch => [activity.activityId, ...branch]);
              });
            }
            return buildBranches(rootAct);
          },
          args: ["$activties", "$relatedActivities.activties"],
          lang: "js"
        }
      }
    }
  },
  // 整合结果并格式化
  {
    $group: {
      _id: null,
      workflowstreams: { $first: "$workflowstreams" }
    }
  },
  { $project: { _id: 0 } }
]);

说明 #

1. $unwind拆分数组：将单文档内的活动数组拆分为多个文档，为$graphLookup提供正确的遍历基础。
2. $graphLookup关联节点：获取每个活动的所有关联节点，包含完整父子链。
3. 筛选根节点：通过判断活动ID是否存在于其他活动的子节点列表中，定位根节点A1。
4. 递归生成分支：使用$function自定义JS函数递归遍历子节点，生成扁平化的工作流路径。
5. 结果整合：将根节点生成的所有分支整合为预期格式。

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