c# - 如何使用 AggregateToCollection() 将 IMongoQueryable 的结果存储在集合中

Question

除非我遗漏了什么，C# MongoDb 驱动程序的文档中似乎有很大的空白。
我正在尝试接受 IMongoQueryable (这是各种 LINQ Where、Select 等操作的结果)并将结果存储在数据库端的集合中。当然，我可以在客户端遍历它并以这种方式持久化，但即使是效率不高的批处理，在 shell 中使用 $merge 也很容易操作。或 $out .
在集合上，有方法 AggregateToCollection<TResult>()我相信这正是我想要的，但它需要一个 PiplineDefinition<TDocument, TResult>我不知道如何从 IMongoQueryable 实际生成的参数.
我一直在使用 GetExecutionModel() 的兔子洞在 IMongoQueryable获得 BsonDocument或 Json 字符串，但我仍然不知道如何将其转换为 PipelineDefinition我需要的!
我以为我会找到的是 IMongoQueryable 上的扩展方法这将允许我发送它以合并到一个集合中。
至此就可以发一个IMongoQueryable对于一个集合，我有一些类似的东西:

    var executionModelDocument = queryable.GetExecutionModel().ToBsonDocument();

    // somehow turn the document into pipeline stages and a pipeline??

    await _database.GetCollection<TDocument>().AggregateToCollectionAsync<TResult>(pipeline);

我如何才能真正做到这一点？

Answer 1

好吧，这最终比预期的要容易，但仍然令人沮丧。
首先，事实证明您可以从 IMongoQueryable 获取管道。作为 JSON 只需调用 ToString()在上面。虽然我说的是 JSON，但它并不完全是 - 输出包含作为阶段数组的管道的完整 JSON，但前面有一个标签，并在括号内包含 JSON。我在这里走了一条捷径，只是拿了一个脏子串:

var queryableJson = queryable.ToString();
var trimmedDocument = queryableJson.Substring(10, queryableJson.Length - 11); // TODO: more reliably get the true json rather than blindly removing what should be "aggregate(" and ")"

接下来，我将 JSON 重新序列化回 BsonDocument 的数组。并做了一个 PipelineDefinition除了它(一个 BsonDocument[] 可以隐式转换为一个 PipelineDefinition ):

PipelineDefinition<TDocument, TResult> pipelineQueryable = BsonSerializer.Deserialize<BsonDocument[]>(trimmedDocument);

请注意，虽然文档位于 http://mongodb.github.io/mongo-csharp-driver/2.4/reference/driver/definitions/#pipelines说我可以隐式地转换一个 BsonDocument对于管道而不是阶段数组，这不是真的，除非我没有找到另一个命名空间中的重载。
现在我们已经为 IMongoQueryable 定义了管道，我们可以简单地向它添加阶段来实现我们想要的结果(在这种情况下，将管道的结果合并到另一个集合中)。您可以指定 MergeStageOptions<TResult> 的属性对象控制行为，但默认值对我来说很好用:

var stageMerge = PipelineStageDefinitionBuilder.Merge<TResult, TResult>(_database.GetCollection<TResult>(), new MergeStageOptions<TResult>());
var mergePipeline = pipelineQueryable.AppendStage(stageMerge);

使用我们新增强的管道，我们可以将其应用于源集合以将输出合并到目标集合中:

_database.GetCollection<TDocument>().AggregateToCollection(mergePipeline);

为简单起见，我在这里演示的不是异步的，但是我在实际代码中利用了驱动程序中的可等待方法，因为同步方法只是包装了可等待版本。
这就是它的全部内容!当我有更多时间时，我会回去尝试跳过序列化-反序列化步骤，因为它明显很慢并且没有必要。为了流畅，我也打算把它变成一种扩展方法。