如何将MongoDB Compass聚合管道转换为MongoTemplate实现？

我在MongoDB Compass中编写的聚合管道可正常运行，但转换为Spring Data MongoDB的MongoTemplate实现时持续报错，尝试多种方式均未解决，也未找到直接在代码中复用原管道的方法，请求帮忙排查。

Compass中可正常运行的聚合管道 #

[
  {
    "$match": {
      "$or": [
        { "riskReport": { "$ne": null } },
        { "riskOfDealReport": { "$ne": null } },
        { "controlProcedureReport": { "$ne": null } },
        { "matrixReport": { "$ne": null } }
      ]
    }
  },
  {
    "$project": {
      "controlProcedureReport": "$controlProcedureReport",
      "riskReport": "$riskReport",
      "matrixReport": "$matrixReport",
      "documentName": {
        "$concatArrays": [
          "$riskReport.ids.value",
          "$controlProcedureReport.ids.value",
          "$matrixReport.ids"
        ]
      }
    }
  },
  {
    "$unwind": { "path": "$documentName" }
  },
  {
    "$project": {
      "year": "$_id.year",
      "periodCode": "$_id.periodCode",
      "serialNumber": "$_id.serialNumber",
      "formationDate": "$formationDate",
      "status": "$status",
      "documentName": "$documentName",
      "fileId": {
        "$cond": {
          "if": { "$in": ["$documentName", "$controlProcedureReport.ids.value"] },
          "then": "$controlProcedureReport.fileId",
          "else": {
            "$cond": {
              "if": { "$in": ["$documentName", "$riskReport.ids.value"] },
              "then": "$riskReport.fileId",
              "else": {
                "$cond": {
                  "if": { "$in": ["$documentName", "$matrixReport.ids"] },
                  "then": "$matrixReport.fileId",
                  "else": ""
                }
              }
            }
          }
        }
      }
    }
  }
]

尝试的Java代码（存在问题） #

ProjectionOperation project2 = Aggregation.project()
        .and("_id.year").as("year")
        .and("_id.periodCode").as("periodCode")
        .and("_id.serialNumber").as("serialNumber")
        .and("formationDate").as("formationDate")
        .and("status").as("status")
        .and("documentName").as("documentName")
        .and(ConditionalOperators.when(
                     ArrayOperators.arrayOf("controlProcedureReport.ids.value").containsValue("$documentName")
             ).thenValueOf("controlProcedureReport.fileId")
             .otherwise(
                     ConditionalOperators.when(
                                     ArrayOperators.arrayOf("riskReport.ids.value").containsValue("$documentName")
                             ).thenValueOf("riskReport.fileId")
                             .otherwise("")
                     )
        ).as("fileId");

Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(packageCriteria),
        project1,
        unwind,
        project2,
        procedureResults,
        icsDisclosures,
        icsDisclosureRegulations
);

List<Document> mappedResults = mongoTemplate.aggregate(aggregation, DOCUMENT_PACKAGE_COLLECTION, Document.class).getMappedResults();

问题分析与修正方案 #

核心问题 #

1. 字段引用错误：ArrayOperators.arrayOf("controlProcedureReport.ids.value")直接传入字符串会被解析为字面量，而非文档字段引用，导致$in判断逻辑失效。
2. 缺失matrixReport分支：原管道中包含对matrixReport.ids的判断，但Java代码里遗漏了这个分支，逻辑不完整。
3. project1和unwind未定义：代码中引用的project1和unwind未给出，需补全对应逻辑。

修正后的完整Java代码 #

// 1. 构建$match条件，与Compass中的逻辑一致
Criteria packageCriteria = new Criteria().orOperator(
        Criteria.where("riskReport").ne(null),
        Criteria.where("riskOfDealReport").ne(null),
        Criteria.where("controlProcedureReport").ne(null),
        Criteria.where("matrixReport").ne(null)
);

// 2. 构建第一个$project操作（对应Compass中的第二个阶段）
ProjectionOperation project1 = Aggregation.project()
        .and("controlProcedureReport").as("controlProcedureReport")
        .and("riskReport").as("riskReport")
        .and("matrixReport").as("matrixReport")
        .and(ArrayOperators.ConcatArrays.arrayOf("riskReport.ids.value")
                .concat("controlProcedureReport.ids.value")
                .concat("matrixReport.ids"))
        .as("documentName");

// 3. 构建$unwind操作
UnwindOperation unwind = Aggregation.unwind("documentName");

// 4. 构建第二个$project操作，修复条件判断逻辑
ProjectionOperation project2 = Aggregation.project()
        .and("_id.year").as("year")
        .and("_id.periodCode").as("periodCode")
        .and("_id.serialNumber").as("serialNumber")
        .and("formationDate").as("formationDate")
        .and("status").as("status")
        .and("documentName").as("documentName")
        .and(ConditionalOperators.when(
                // 正确构建$in表达式：判断documentName是否在controlProcedureReport.ids.value数组中
                AggregationExpression.from(MongoExpression.create("{ $in: [ '$documentName', '$controlProcedureReport.ids.value' ] }"))
        ).thenValueOf("controlProcedureReport.fileId")
        .otherwise(
                ConditionalOperators.when(
                        AggregationExpression.from(MongoExpression.create("{ $in: [ '$documentName', '$riskReport.ids.value' ] }"))
                ).thenValueOf("riskReport.fileId")
                .otherwise(
                        ConditionalOperators.when(
                                AggregationExpression.from(MongoExpression.create("{ $in: [ '$documentName', '$matrixReport.ids' ] }"))
                        ).thenValueOf("matrixReport.fileId")
                        .otherwise("")
                )
        )).as("fileId");

// 5. 构建完整聚合管道（如果procedureResults等操作是额外需求，需确保它们的逻辑正确）
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(packageCriteria),
        project1,
        unwind,
        project2
        // 如果需要保留procedureResults, icsDisclosures, icsDisclosureRegulations，请确保这些操作已正确定义
);

// 执行聚合查询
List<Document> mappedResults = mongoTemplate.aggregate(aggregation, "document_package_collection", Document.class).getMappedResults();

关键说明 #

• 使用AggregationExpression.from(MongoExpression.create(...))直接构建MongoDB原生的$in表达式，避免Spring Data封装类的字段引用问题。
• 补全了matrixReport的条件分支，确保逻辑与Compass中的管道完全一致。
• 明确了project1和unwind的实现，保证管道阶段完整。

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