如何将 MongoDB aggregation 转为 Spring WebFlux 代码

结论

问题在于 $replaceRoot 的构造方式。ObjectOperators.valueOf("payment").mergeWith(Aggregation.ROOT) 执行的是对象合并，无法构造出下面的新文档：

{
  "_id": "$_id",
  "type": "payment",
  "payment": "$$ROOT"
}

应改用 replaceRoot().withDocument() 明确定义新的根文档，并分别处理字段引用、字符串常量和 $$ROOT。$unionWith 中的 pipeline 也需要使用相同的写法。

正确的聚合代码

import static org.springframework.data.mongodb.core.aggregation.Aggregation.limit;
import static org.springframework.data.mongodb.core.aggregation.Aggregation.newAggregation;
import static org.springframework.data.mongodb.core.aggregation.Aggregation.replaceRoot;

import java.util.List;

import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
import org.springframework.data.mongodb.core.aggregation.LimitOperation;
import org.springframework.data.mongodb.core.aggregation.ReplaceRootOperation;
import org.springframework.data.mongodb.core.aggregation.UnionWithOperation;

import reactor.core.publisher.Flux;

ReplaceRootOperation replacePayment = replaceRoot()
        .withDocument()
        .andValueOf("_id").as("_id")
        .andValue("payment").as("type")
        .andValueOf(Aggregation.ROOT).as("payment");

ReplaceRootOperation replaceRefund = replaceRoot()
        .withDocument()
        .andValueOf("_id").as("_id")
        .andValue("refund").as("type")
        .andValueOf(Aggregation.ROOT).as("refund");

UnionWithOperation unionWithRefund = UnionWithOperation
        .unionWith("refund")
        .pipeline(replaceRefund);

LimitOperation limitOperation = limit(pageSize);

List<AggregationOperation> operations = List.of(
        replacePayment,
        unionWithRefund,
        limitOperation
);

Aggregation aggregation = newAggregation(operations);

Flux<Transaction> output = mongoTemplate.aggregate(
        aggregation,
        "payment",
        Transaction.class
);

生成的核心 pipeline 应与下面的结构等价：

[
  {
    $replaceRoot: {
      newRoot: {
        _id: "$_id",
        type: "payment",
        payment: "$$ROOT"
      }
    }
  },
  {
    $unionWith: {
      coll: "refund",
      pipeline: [
        {
          $replaceRoot: {
            newRoot: {
              _id: "$_id",
              type: "refund",
              refund: "$$ROOT"
            }
          }
        }
      ]
    }
  },
  {
    $limit: pageSize
  }
]

为什么原来的写法无法得到预期结果

这段代码：

ObjectOperators.valueOf("payment").mergeWith(Aggregation.ROOT)

表示对象合并操作。其中，"payment" 会作为字段引用参与表达式计算，而不是创建一个名为 payment 的字段。mergeWith() 也不会自动添加 type、payment 或 refund 等包装字段。

因此，聚合结果仍接近原始的 payment 文档。MongoDB 能够返回 _id，所以 Transaction.id 有值；但结果中没有顶层的 type、payment 和 refund，Spring Data 映射后，这些属性就会是 null。

使用 withDocument()，相当于明确告诉 Spring Data，新根节点是一个重新构造的文档。

下面三种写法的含义并不相同：

.andValueOf("_id").as("_id")                 // 引用当前文档的 $_id
.andValue("payment").as("type")              // 字符串常量 "payment"
.andValueOf(Aggregation.ROOT).as("payment")  // 引用整个当前文档 $$ROOT

当 Spring Data 版本的 Builder 行为不一致时

如果当前 Spring Data MongoDB 版本不支持上述 withDocument() 链式 API，或者 $$ROOT 没有被正确渲染，可以直接构造 BSON stage。这种方式更贴近原始 MongoDB script，排查问题也更直接。

import org.bson.Document;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
import org.springframework.data.mongodb.core.aggregation.UnionWithOperation;

AggregationOperation replacePayment = context ->
        new Document("$replaceRoot",
                new Document("newRoot",
                        new Document("_id", "$_id")
                                .append("type", "payment")
                                .append("payment", "$$ROOT")));

AggregationOperation replaceRefund = context ->
        new Document("$replaceRoot",
                new Document("newRoot",
                        new Document("_id", "$_id")
                                .append("type", "refund")
                                .append("refund", "$$ROOT")));

UnionWithOperation unionWithRefund = UnionWithOperation
        .unionWith("refund")
        .pipeline(replaceRefund);

Aggregation aggregation = Aggregation.newAggregation(
        replacePayment,
        unionWithRefund,
        Aggregation.limit(pageSize)
);

Flux<Transaction> output = mongoTemplate.aggregate(
        aggregation,
        "payment",
        Transaction.class
);

Transaction 映射结构

Transaction 的属性名称和类型必须与聚合输出保持一致，例如：

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;

@Document("transaction")
public class Transaction {

    @Id
    private String id;

    private String type;

    private Payment payment;

    private Refund refund;

    // getters and setters
}

如果 MongoDB 中的 _id 不是字符串，而是 ObjectId、UUID 或其他类型，id 应改为对应的 Java 类型。嵌套的 Payment 和 Refund 类也需要能够映射原 collection 中的字段。

每条结果只会填充一个业务字段：

type = payment  -> payment 有值，refund 为 null
type = refund   -> refund 有值，payment 为 null

这符合当前聚合结构的预期。

WebFlux 中不要调用 blockFirst()

下面的代码可以用于临时测试：

Transaction transaction = output.blockFirst();

但不要在 WebFlux 请求处理线程中执行阻塞操作，应直接返回响应式类型：

public Flux<Transaction> findTransactions(int pageSize) {
    return mongoTemplate.aggregate(
            aggregation,
            "payment",
            Transaction.class
    );
}

如果只需要第一条记录，可以返回：

public Mono<Transaction> findFirstTransaction(int pageSize) {
    return mongoTemplate.aggregate(
            aggregation,
            "payment",
            Transaction.class
    ).next();
}

是否会创建 transaction collection

aggregate(..., Transaction.class) 中的 Transaction.class 只用于结果映射，不会创建或写入名为 transaction 的 collection。当前 pipeline 只是合并 payment 和 refund，再以 Transaction 结构返回结果。

如果需要将结果持久化到 transaction collection，还要在 pipeline 末尾添加 $merge 或 $out。这会产生数据库写操作，因此需要另外处理重复 _id、覆盖策略和执行权限。

备注：内容仅供参考。