MongoDB聚合管道：筛选匹配搜索集且无交集的随机文档集合

问题背景 #

现有搜索集{a, b}，以及如下MongoDB文档集合：

[
  { "owner": "anna", "letters": ["c"] },
  { "owner": "bob", "letters": ["b", "c"] },
  { "owner": "cai", "letters": ["a", "b"] },
  { "owner": "dora", "letters": ["a", "d"] },
  { "owner": "emil", "letters": ["a"] },
  { "owner": "fry", "letters": ["b"] }
]

需求规则 #

需要选出一组随机文档，满足两个核心条件：

1. 搜索集{a, b}是所选文档letters字段累积去重后的子集（即累积起来必须覆盖a和b）；
2. 所选文档的letters数组之间无交集（任意两个文档的letters没有共同元素）。

有效/无效示例 #

有效示例

示例1：单个文档覆盖全部搜索集

{ "owner": "cai", "letters": ["a", "b"] }

示例2：两个无交集文档分别覆盖a和b

[
  { "owner": "emil", "letters": ["a"] },
  { "owner": "fry", "letters": ["b"] }
]

示例3：两个无交集文档，一个覆盖a，另一个覆盖b+额外元素

[
  { "owner": "emil", "letters": ["a"] },
  { "owner": "bob", "letters": ["b", "c"] }
]

示例4：两个无交集文档，分别覆盖b+额外、a+额外元素

[
  { "owner": "bob", "letters": ["b", "c"] },
  { "owner": "dora", "letters": ["a", "d"] }
]

无效示例

示例1：两个文档的letters有交集（都包含b）

[
  { "owner": "bob", "letters": ["b", "c"] },
  { "owner": "cai", "letters": ["a", "b"] }
]

示例2：两个文档的letters有交集（都包含a）

[
  { "owner": "emil", "letters": ["a"] },
  { "owner": "cai", "letters": ["a", "b"] }
]

已尝试操作与卡点 #

已尝试在聚合管道中按letters分组并随机化文档顺序，但无法实现上述集合规则的筛选逻辑，不知道如何动态确保所选文档的letters无交集且覆盖搜索集。

解决方案 #

方案1：聚合管道实现（适合小数据集） #

通过聚合管道的自关联、集合运算，生成所有有效组合后随机选取：

db.yourCollection.aggregate([
  // 预处理：将letters转为集合，标记是否包含a/b
  {
    $addFields: {
      letterSet: { $setUnion: ["$letters"] },
      hasA: { $in: ["a", "$letters"] },
      hasB: { $in: ["b", "$letters"] }
    }
  },
  // 自关联查找与当前文档无交集的其他文档
  {
    $lookup: {
      from: "yourCollection",
      let: { currentLetters: "$letterSet" },
      pipeline: [
        {
          $match: {
            $expr: { $eq: [{ $setIntersection: ["$letterSet", "$$currentLetters"] }, []] }
          }
        }
      ],
      as: "possibleMatches"
    }
  },
  // 筛选出能组成有效组合的文档
  {
    $addFields: {
      validPairs: {
        $filter: {
          input: "$possibleMatches",
          cond: {
            $or: [
              // 当前文档本身已覆盖a+b，单独有效
              { $and: ["$hasA", "$hasB"] },
              // 组合后覆盖a+b
              { $or: [
                { $and: ["$hasA", "$$this.hasB"] },
                { $and: ["$hasB", "$$this.hasA"] }
              ]}
            ]
          }
        }
      }
    }
  },
  // 过滤掉无有效组合的文档
  { $match: { $or: [{ $and: ["$hasA", "$hasB"] }, { $ne: ["$validPairs", []] }] } },
  // 随机选取一个基准文档
  { $sample: { size: 1 } },
  // 整理结果：单个文档或随机配对的组合
  {
    $project: {
      result: {
        $cond: {
          if: { $and: ["$hasA", "$hasB"] },
          then: [ { owner: "$owner", letters: "$letters" } ],
          else: [
            { owner: "$owner", letters: "$letters" },
            { $arrayElemAt: ["$validPairs", { $floor: { $multiply: [{ $rand: {} }, { $size: "$validPairs" }] } }] }
          ]
        }
      }
    }
  }
])

方案2：应用层迭代筛选（适合大数据集） #

如果数据集较大，聚合管道生成全组合会影响性能，建议在应用层实现迭代筛选逻辑（伪代码示例）：

const searchSet = new Set(['a', 'b']);
const collection = db.collection('yourCollection');
let selectedDocs = [];
let coveredLetters = new Set();

while (coveredLetters.size < searchSet.size) {
  // 生成筛选条件：与已选文档无交集，且包含未覆盖的搜索元素
  const excludeLetters = Array.from(coveredLetters);
  const requiredFilters = Array.from(searchSet).map(letter => ({ letters: letter }));
  
  const candidates = await collection.find({
    letters: { $not: { $elemMatch: { $in: excludeLetters } } },
    $or: requiredFilters
  }).toArray();

  // 无候选则重置重新选
  if (candidates.length === 0) {
    selectedDocs = [];
    coveredLetters = new Set();
    continue;
  }

  // 随机选一个候选文档
  const randomIndex = Math.floor(Math.random() * candidates.length);
  const pickedDoc = candidates[randomIndex];
  
  selectedDocs.push(pickedDoc);
  // 更新已覆盖的字母集合
  pickedDoc.letters.forEach(letter => coveredLetters.add(letter));
}

console.log(selectedDocs);

核心逻辑说明 #

两种方案的核心都是：

1. 避免交集：通过$setIntersection（聚合）或$elemMatch（查询）确保新选文档的letters与已选集合无重叠；
2. 覆盖搜索集：逐步累积已覆盖的字母，直到包含a和b；
3. 随机性：通过$sample（聚合）或Math.random()（应用层）实现随机选取。

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