MongoDB $lookup关联可选字段localField不存在时如何保留全量查询结果

问题根因 #

你当前脚本只返回存在car_id用户的核心原因是$unwind操作的默认行为：当待展开的字段为空数组或不存在时，$unwind会直接丢弃整条文档。没有car_id的用户关联cars集合后得到的car字段是空数组，经过{"$unwind": "$car"}阶段就会被过滤。

另外注意你给出的houses集合示例存在笔误，三条数据的_id都写为1，实际使用时需要改为1、2、3才能正常匹配house_id为2、3的用户。

修正后的聚合查询语句 #

修改点如下：

1. 给car字段的$unwind添加preserveNullAndEmptyArrays: true配置，保留car为空数组的用户数据
2. 新增$project阶段，自动删除值为null的car字段，符合无car_id就不返回该字段的需求
3. 将房屋关联的as字段改为houses，匹配你给出的预期输出格式

db.users.aggregate([
  {                                   
    "$lookup": {
      "from": "houses",
      "localField": "house_id",
      "foreignField": "_id",
      "as": "houses"
    }
  },
  {"$unwind": "$houses"},
  {                                   
    "$lookup": {
      "from": "cars",
      "localField": "car_id",
      "foreignField": "_id",
      "as": "car"
    }
  },
  {
    "$unwind": {
      "path": "$car",
      "preserveNullAndEmptyArrays": true
    }
  },
  {
    "$project": {
      "name": 1,
      "house_id": 1,
      "houses": 1,
      "car": {
        "$cond": {
          "if": {"$eq": ["$car", null]},
          "then": "$$REMOVE",
          "else": "$car"
        }
      }
    }
  }
])

pymongo 调用示例 #

你使用pymongo的场景下直接传入上述聚合管道即可，示例代码如下：

from pymongo import MongoClient

# 初始化连接
client = MongoClient("mongodb://你的MongoDB地址:端口")
db = client["你的数据库名"]

# 定义聚合管道
pipeline = [
  {                                   
    "$lookup": {
      "from": "houses",
      "localField": "house_id",
      "foreignField": "_id",
      "as": "houses"
    }
  },
  {"$unwind": "$houses"},
  {                                   
    "$lookup": {
      "from": "cars",
      "localField": "car_id",
      "foreignField": "_id",
      "as": "car"
    }
  },
  {
    "$unwind": {
      "path": "$car",
      "preserveNullAndEmptyArrays": True
    }
  },
  {
    "$project": {
      "name": 1,
      "house_id": 1,
      "houses": 1,
      "car": {
        "$cond": {
          "if": {"$eq": ["$car", None]},
          "then": "$$REMOVE",
          "else": "$car"
        }
      }
    }
  }
]

# 执行查询
result = list(db.users.aggregate(pipeline))

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