在基于文档的聚合管道中使用 $merge 不起作用

Question

我有一个集合，我想对其执行聚合并将结果放入同一数据库中的单独集合中。在仔细检查文档时，我偶然发现了 $merge，它的工作方式正是我想要的。我想出了以下 mongo shell 管道，它工作得很好。

db.getCollection('SOURCE_COLLECTION').aggregate([
  {
    "$match": {type: 'ABC'}
  },
  {
    "$merge": {
      "into": "OUTPUT_COLLECTION",
      "whenMatched": "replace"
    }
  }
])

现在，我需要在Spring启动时有同样的效果，为此我想出了下面的，理论上应该没有什么不同。

final ArrayList<Document> pipeline = new ArrayList<>();

pipeline.add(Document.parse("{$match: {type: 'ABC'}}"));
pipeline.add(Document.parse("{$merge: {into: 'OUTPUT_COLLECTION', whenMatched: 'replace'}}"));

mongoTemplate.getDb()
    .getCollection("SOURCE_COLLECTTION", Document.class)
    .aggregate(pipeline);

尽管如此，这是行不通的。可以看出，我正在使用 MongoCollection<T>.aggregate() 方法，该方法采用 List<Document> 作为管道。管道中的每个阶段都是通过将 JSON 字符串解析为文档生成的。

有趣的是，当我用 $out 替换合并时，它没有任何问题。

final ArrayList<Document> pipeline = new ArrayList<>();

pipeline.add(Document.parse("{$match: {type: 'ABC'}}"));
pipeline.add(Document.parse("{$out: 'OUTPUT_COLLECTION'}"));

mongoTemplate.getDb()
    .getCollection("SOURCE_COLLECTTION", Document.class)
    .aggregate(pipeline);

但这对我没有好处，因为这个聚合将被执行多次（实际上我试图在这里填充一个物化视图类型的集合）。我需要 $merge 才能工作，但事实并非如此。我错过了什么？有人能看到我没有看到的东西吗？

Answer 1

当在管道中使用 $merge 时，方法 aggregate(pipeline) 看起来不像终端操作。要使合并生效，应该调用 void toCollection() 或 void forEach(Consumer<? extends T> consumer) 等操作。因此，从 aggregate() 调用 return 上的 toCollection() 就成功了。

final ArrayList<Document> pipeline = new ArrayList<>();

pipeline.add(Document.parse("{$match: {type: 'ABC'}}"));
pipeline.add(Document.parse("{$out: 'OUTPUT_COLLECTION'}"));

mongoTemplate.getDb()
    .getCollection("SOURCE_COLLECTTION", Document.class)
    .aggregate(pipeline)
    .toCollection(); // This invocation completes the pipeline when it ends with $merge.

有趣的是，在文档中似乎没有提及此类内容（至少在我能找到的任何文档中都没有）。我什至不确定这是否是预期的行为。

在基于文档的聚合管道中使用 $merge 不起作用

Using $merge in Document-based aggregation pipeline is not working

java

aggregation

mongodb

spring-boot