PyMongo聚合$merge更新时嵌套字段丢失问题求助

问题背景 #

使用MongoDB 5+和最新版PyMongo，执行聚合查询替换configuration.sql字段内容后，通过$merge更新文档时，configuration对象下的其他字段（如identifier、steps等）全部丢失。

输入文档结构 #

{
  "_id": {
    "$oid": "639a60"
  },
  "status": "inactive",
  "version": "0.1",
  "configuration": {
    "identifier": "backfill-test",
    "secondaryIdentifier": "platform_type",
    "sql": "select * from test where lastupd_ts >= '2021-09-10 18:00:00' and lastupd_ts <= '2021-09-10 19:00:00'",
    "steps": [
      {
        "service": "Publish",
        "order": 1,
        "configuration": {
          "topic": "platform-type",
          "type": "PlatformType",
          "action": "N",
          "keyDeserializer": "serializers.Kafka",
          "valueDeserializer": "serializers.Deserializer"
        }
      }
    ]
  },
  "name": "data-exporter-svc"
}

原错误查询 #

database.collection_name.aggregate(
[
        {"$match": {"configuration.identifier": "backfill-test"}},
        {
            "$project":
            {
                "configuration.sql": {"$replaceAll": {"input": "$configuration.sql", "find": "lastupd_ts >= \'2021-09-10 18:00:00\' and lastupd_ts <= \'2021-09-10 19:00:00\'", "replacement": "lastupd_ts >= \'2024-00-10 18:00:00\' and lastupd_ts <= \'2024-00-10 19:00:00\'"}}
            }
        },
        {
            "$merge": "collection_name"
        },
    ])

当前错误结果 #

{
  "_id": {
    "$oid": "639a60"
  },
  "status": "inactive",
  "version": "0.1",
  "configuration": {
    "sql": "select * from test where lastupd_ts >= '2024-00-10 18:00:00' and lastupd_ts <= '2024-00-10 19:00:00'"
  },
  "name": "data-exporter-svc"
}

期望结果 #

{
  "_id": {
    "$oid": "639a60"
  },
  "status": "inactive",
  "version": "0.1",
  "configuration": {
    "identifier": "backfill-test",
    "secondaryIdentifier": "platform_type",
    "sql": "select * from test where lastupd_ts >= '2024-00-10 18:00:00' and lastupd_ts <= '2024-00-10 19:00:00'",
    "steps": [
      {
        "service": "Publish",
        "order": 1,
        "configuration": {
          "topic": "platform-type",
          "type": "PlatformType",
          "action": "N",
          "keyDeserializer": "serializers.Kafka",
          "valueDeserializer": "serializers.Deserializer"
        }
      }
    ]
  },
  "name": "data-exporter-svc"
}

原因分析 #

原查询中使用$project阶段时，仅显式指定了configuration.sql字段，$project默认会过滤掉所有未显式声明的字段，包括configuration对象内的其他属性。后续$merge阶段会用这个被过滤后的文档覆盖原文档，导致嵌套字段丢失。

修正方案 #

使用$addFields替换$project阶段，$addFields会保留文档中所有原有字段，仅更新或添加指定的字段，完美解决嵌套字段丢失问题。

修正后的查询代码 #

database.collection_name.aggregate(
[
        {"$match": {"configuration.identifier": "backfill-test"}},
        {
            "$addFields": {
                "configuration": {
                    # 保留原configuration的所有字段，仅替换sql值
                    "$mergeObjects": [
                        "$configuration",
                        {
                            "sql": {
                                "$replaceAll": {
                                    "input": "$configuration.sql",
                                    "find": "lastupd_ts >= '2021-09-10 18:00:00' and lastupd_ts <= '2021-09-10 19:00:00'",
                                    "replacement": "lastupd_ts >= '2024-00-10 18:00:00' and lastupd_ts <= '2024-00-10 19:00:00'"
                                }
                            }
                        }
                    ]
                }
            }
        },
        {
            "$merge": {
                "into": "collection_name",
                # 指定匹配条件为_id，确保只更新目标文档
                "on": "_id",
                # 当匹配到文档时，仅合并更新的字段（可选，进一步确保安全）
                "whenMatched": "merge"
            }
        },
    ])

说明 #

1. $mergeObjects：将原configuration对象与包含新sql值的对象合并，保留原对象所有字段，仅覆盖sql字段。
2. $addFields：保留文档中所有原有顶级字段（如status、version、name），同时更新嵌套的configuration对象。
3. $merge的whenMatched: "merge"：确保当匹配到现有文档时，仅合并更新的字段，而非完全替换整个文档，进一步保障数据安全。

内容的提问来源于stack exchange，提问作者Shubham Dubey