MongoDB文档分组及条件式数组推送实现方法

我有一个phrases集合，结构如下：

• phrase：字符串类型
• phraseLength：短语字符串的长度（数值类型）
• author：字符串类型

需求是按author对短语进行分组，且每个作者的所有短语总长度不超过200字符。具体逻辑为：遍历每个作者的短语，若当前累计总长度小于200，则将该短语加入该作者的phrases数组；累计超过200后，后续短语不再加入。

示例输入文档 #

[
    {
        "phrase": "This is phrase 1 of author 1",
        "phraseLength": 50,
        "author": "Author 1"
    },
    {
        "phrase": "This is phrase 1 of author 1",
        "phraseLength": 150,
        "author": "Author 1"
    },
    {
        "phrase": "This is phrase 1 of author 1",
        "phraseLength": 10,
        "author": "Author 1"
    },
    {
        "phrase": "This is phrase 1 of author 2",
        "phraseLength": 20,
        "author": "Author 2"
    },
    {
        "phrase": "This is phrase 2 of author 2",
        "phraseLength": 180,
        "author": "Author 2"
    },
    {
        "phrase": "This is phrase 3 of author 2",
        "phraseLength": 50,
        "author": "Author 2"
    }
]

期望输出 #

[
    {
        "_id": "Author 1",
        "phrases": [
            {
                "phrase": "This is phrase 1 of author 1",
                "phraseLength": 50,
                "author": "Author 1"
            },
            {
                "phrase": "This is phrase 1 of author 1",
                "phraseLength": 150,
                "author": "Author 1"
            }
        ]
    },
    {
        "_id": "Author 2",
        "phrases": [
            {
                "phrase": "This is phrase 1 of author 2",
                "phraseLength": 20,
                "author": "Author 2"
            },
            {
                "phrase": "This is phrase 2 of author 2",
                "phraseLength": 180,
                "author": "Author 2"
            }
        ]
    }
]

之前的尝试 #

仅实现分组的基础查询：

db.phrases.aggregate([
    {
        $group: {
            _id: '$author',
            phrases: {
                $push: { phrase: '$phrase', phraseLength: '$phraseLength', author: '$author' }
            }
        }
    }
])

错误的限制长度尝试（语法与逻辑均不成立）：

db.phrases.aggregate([
    {
        $group: {
            _id: "$author",
            runningTotalPhraseLength: { $sum: '$phraseLength' },
            $cond: {
                if: {
                    $lte: [{ $size: '$runningTotalPhraseLength' }, 200]
                },
                then: {
                    phrases: {
                        $push: { phrase: '$phrase', phraseLength: '$phraseLength', author: '$author' }
                    }
                },
                else: {},
            },
        }
    }
])

解决方案 #

要实现逐次累计判断的逻辑，需借助窗口函数计算每个文档在所属作者组内的累计长度，再过滤后分组。具体聚合查询如下：

db.phrases.aggregate([
    // 1. 按author排序，保证同一作者的文档连续处理（可选但建议，确保顺序一致性）
    { $sort: { author: 1 } },
    // 2. 计算每个作者组内的累计phraseLength
    {
        $setWindowFields: {
            partitionBy: "$author",
            sortBy: { _id: 1 }, // 可替换为实际业务需要的排序字段，如创建时间
            output: {
                cumulativeLength: {
                    $sum: "$phraseLength",
                    window: { documents: ["unbounded", "current"] }
                }
            }
        }
    },
    // 3. 过滤掉累计长度超过200的文档
    {
        $match: {
            $expr: { $lte: ["$cumulativeLength", 200] }
        }
    },
    // 4. 按author分组，收集符合条件的短语
    {
        $group: {
            _id: "$author",
            phrases: {
                $push: { phrase: "$phrase", phraseLength: "$phraseLength", author: "$author" }
            }
        }
    }
])

逻辑说明 #

1. $sort阶段：按author排序，确保同一作者的文档被连续处理，保证结果顺序与输入一致。
2. $setWindowFields阶段：
   • partitionBy: "$author"：将文档按作者分组
   • sortBy：定义组内文档的处理顺序（示例用_id，可根据业务调整）
   • cumulativeLength：计算从组内第一个文档到当前文档的phraseLength总和
3. $match阶段：仅保留累计长度≤200的文档，超过阈值的直接过滤
4. $group阶段：将过滤后的文档按作者分组，收集到phrases数组中

失败原因分析 #

• $group是批量聚合阶段，无法在分组过程中逐次判断累计长度并决定是否添加文档；
• 错误将$cond作为$group的顶级字段，$group仅允许聚合操作符（如$sum、$push）作为字段值；
• $size: '$runningTotalPhraseLength'语法错误，runningTotalPhraseLength是数值类型，$size仅适用于数组。

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