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

Question

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

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

现在，我需要在 Spring boot 中达到相同的效果，为此我想出了以下内容，理论上应该没有什么不同。

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);

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

有趣的是，当我用 替换 merge 时$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

当在管道中使用时，该方法的aggregate(pipeline)行为看起来不像终端操作。$merge为了使合并起作用，应该调用一些操作，例如void toCollection()或。void forEach(Consumer<? extends T> consumer)因此，toCollection()在 from 返回时调用aggregate()就可以了。

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.

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