Mongo C# Driver 聚合使用---深入浅出

聚合查询结构体系

我们都知道Mongo中聚合是由$match，$project等聚合项组成，所以在C# Driver中具有两种类型：聚合管道(PipelineDefinition)和聚合管道项(IPipelineStageDefinition) ，下面先来看一下聚合管道项的结构体系

IPipelineStageDefinition

IPipelineStageDefinition接口是聚合管道项的顶级接口，这个接口中只定义了一些获取输入类型和输出类型的简单的属性

 public interface IPipelineStageDefinition{    //      输入类型
    Type InputType { get; }    //      获取管道操作名称
    string OperatorName { get; }    //      输出类型
    Type OutputType { get; }    //      获取一个IRenderedIRenderedPipelineStageDefinition
    //      IRenderedPipelineStageDefinition是一个Stage提供对象，下面会介绍
    IRenderedPipelineStageDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry);    //      获取当前管道项的字符串格式，例：{ "$match" : { "_id" : "402066782845407232" } }
    string ToString(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry);
}

这个接口具有一个PipelineStageDefinition派生类，这个类是一个抽象类，在这个抽象类中只多了两个隐式转换，

public abstract class PipelineStageDefinition : IPipelineStageDefinition
{    //      将一个Bsondocument对象转换为管道项
    public static implicit operator PipelineStageDefinition(Bsondocument document);    //      将一个json字符串转换为管道项
    public static implicit operator PipelineStageDefinition(string json);
}

用过C# Driver的朋友都应该知道我们使用Driver时经常使用这种隐式转换，例如经常使用FilterDefinition便可使用json字符串直接赋值,这也是Driver强大的地方。

FilterDefinition filter = "{_id:123}";

其实这两个隐式转换如果翻源码就会看到直接创建了这个抽象类的实现类对象

public static implicit operator PipelineStageDefinition(Bsondocument document)
{    if (document == null)        return null;    return new BsondocumentPipelineStageDefinition(document);
}public static implicit operator PipelineStageDefinition(string json)
{    if (json == null)        return null;    return new JsonPipelineStageDefinition(json);
}

也就是说这个抽象类具有这么两个派生类BsondocumentPipelineStageDefinition，JsonPipelineStageDefinition 这两个类型就是使用Bsondocument对象和json字符串进行实例化聚合管道项

PipelineStageDefinition其它派生类

如果仅仅使用，只使用上面那两个派生类即可，但实际上IPipelineStageDefinition的派生类还有两个：

DelegatedPipelineStageDefinition：由一个Func
SortPipelineStageDefinition：排序项的实例对象

其实这两个派生类在使用上根本不需要知道，它们的访问级别是internal,也就是说在使用时根本无法创建这两个派生类的实例对象，其实这两个类都是PipelineStageDefinition实例在调用Match() ，Project() ，Sort() 方法时进行内部创建的，这个下面再说

对于SortPipelineStageDefinition和DelegatedPipelineStageDefinition这两个派生类其实内部特别简单，但是却又扯到了另外一个类型

internal class SortPipelineStageDefinition : PipelineStageDefinition{    public SortPipelineStageDefinition(SortDefinition sort)
    {        Sort = sort;
    }    //      排序条件对象
    public SortDefinition Sort { get; private set; }    //      操作项名称
    public override string OperatorName => "$sort";    public override RenderedPipelineStageDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry)
    {        var renderedSort = Sort.Render(inputSerializer, serializerRegistry);        var document = new Bsondocument(OperatorName, renderedSort);        return new RenderedPipelineStageDefinition(OperatorName, document, inputSerializer);
    }
}internal sealed class DelegatedPipelineStageDefinition : PipelineStageDefinition    {        //      委托缓存
        private readonly Func, IBsonSerializerRegistry, RenderedPipelineStageDefinition> _renderer;        public DelegatedPipelineStageDefinition(string operatorName, Func, IBsonSerializerRegistry, RenderedPipelineStageDefinition> renderer)
        {
            _renderer = renderer;
        }        //      获取RenderedPipelineStageDefinition  直接返回委托调用
        public override RenderedPipelineStageDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry)
        {            return _renderer(inputSerializer, serializerRegistry);
        }
    }

通过上面代码可以看到这两个派生类型特别简单，感觉就是一个简单的代理，一切都指向于RenderedPipelineStageDefinition这个类型，也就是在真正执行聚合操作时可能使用的就是这个类型，这个在这先留一下悬念，因为RenderedPipelineStageDefinition这个类型还涉及到了整个聚合管道对象和执行操作，等到下面再讲解。

PipelineStageDefinitionBuilder

下面来说一下PipelineStageDefinitionBuilder这个类型，顾名思义，这是一个创建PipelineStageDefinition的类型，它是一个静态类，内部具有创建各种的使用方法，这个类型中方法特别多，也不一一细讲，只讲三个方法，也就是上面提到的Match() ，Project() ，Sort()

public static class PipelineStageDefinitionBuilder{    //      match
     public static PipelineStageDefinition Match(
            FilterDefinition filter)
     {         
         const string operatorName = "$match";         var stage = new DelegatedPipelineStageDefinition(
             operatorName,
             (s, sr) => new RenderedPipelineStageDefinition(operatorName, new Bsondocument(operatorName, filter.Render(s, sr)), s));         return stage;
     }    //          project
    public static PipelineStageDefinition Project(ProjectionDefinition projection)
        {            const string operatorName = "$project";            var stage = new DelegatedPipelineStageDefinition(
                operatorName,
                (s, sr) =>
                {                    var renderedProjection = projection.Render(s, sr);
                    Bsondocument document;                    if (renderedProjection.document == null)                        document = new Bsondocument();                    else
                        document = new Bsondocument(operatorName, renderedProjection.document);                    return new RenderedPipelineStageDefinition(operatorName, document, renderedProjection.ProjectionSerializer);
                });            return stage;
        }    //      sort
    public static PipelineStageDefinition Sort(
        SortDefinition sort)
    {        return new SortPipelineStageDefinition(sort);
    }
}

上面就是这三个方法的源代码，三个方法分别使用FilterDefinition，ProjectionDefinition，SortDefinition实例创建PipelineStageDefinition对象，而所创建的也是后面讲的那两个派生类，这也验证了上面所说的两个类型的用途。

PipelineStageDefinition类总结

从上面一步步可以得知，Driver为我们提供了三种创建聚合项的办法，其实这三种也应用于driver的各种使用上

Bsondocument创建
json字符串创建
使用PipelineStageDefinitionBuilder进行创建

PipelineDefinition

说完管道项，下面就说一下整个聚合管道的操作类PipelineDefinition以及它的派生类

首先PipelineDefinition这个父级类型，它跟PipelineStageDefinition一样是一个抽象类型，并且和PipelineStageDefinition相同的是它也有一个Render方法和两个隐式转换，多了几个静态的创建方法，使得更具有扩展性

 public abstract class PipelineDefinition {        public abstract RenderedPipelineDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry);    
     //     使用管道项集合创建一个PipelineStagePipelineDefinition实例对象
      public static PipelineDefinition Create(
            IEnumerable stages,
            IBsonSerializer outputSerializer = null)
        {            return new PipelineStagePipelineDefinition(stages, outputSerializer);
        }        //      使用Bsondocument集合创建一个BsondocumentStagePipelineDefinition对象
        public static PipelineDefinition Create(
            IEnumerable stages,
            IBsonSerializer outputSerializer = null)
        {            return new BsondocumentStagePipelineDefinition(stages, outputSerializer);
        }        //      使用json字符串集合创建一个BsondocumentStagePipelineDefinition对象
         public static PipelineDefinition Create(
            IEnumerable stages,
            IBsonSerializer outputSerializer = null)
        {            return Create(stages?.Select(s => Bsondocument.Parse(s)), outputSerializer);
        }        //      隐式转换
        
        //      将IPipelineStageDefinition集合隐式转换为PipelineStagePipelineDefinition对象
        public static implicit operator PipelineDefinition(List stages)
        {            return Create(stages);
        }        //      将Bsondocument集合隐式转换为BsondocumentStagePipelineDefinition对象
        public static implicit operator PipelineDefinition(List stages)
        {            return Create(stages);
        }

 }

注：PipelineDefinition类中还封装了数组参数和其它内容，有兴趣的朋友可以自己去看看

上面类型可以看出PipelineDefinition做了很多封装，为了使用更加便捷。从上面也看到了两个派生类型：PipelineStagePipelineDefinition和BsondocumentStagePipelineDefinition

其实PipelineDefinition派生类型一共有7个，我们能用到的是6个，我将这个7个类型分为：创建性，改变性和外部不可用性这三种，下面先来看看创建性

注：其实严格意义上是两种，外部不可用的派生类型属于创建性，外部不可用的派生类型也只是在特定情况下被内部用到。

创建性派PipelineDefinition

创建性有3个，其中两个就是上面基类中创建的两个派生类型，另外一个是EmptyPipelineDefinition，顾名思义这是一个空的管道，这个跟创建空条件那个是极其相似的( Builders)，

//      EmptyPipelineDefinitionpublic sealed class EmptyPipelineDefinition : PipelineDefinition{    public override IEnumerable Stages =>         Enumerable.Empty();    //      
    public override RenderedPipelineDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry)
    {        var documents = Enumerable.Empty();        return new RenderedPipelineDefinition(documents, _inputSerializer ?? inputSerializer);
    }
}//      PipelineStagePipelineDefinition
  public sealed class PipelineStagePipelineDefinition : PipelineDefinition    {        private readonly IList _stages;        public PipelineStagePipelineDefinition(IEnumerable stages, IBsonSerializer outputSerializer = null)
        {
            _stages = stages;
            _outputSerializer = outputSerializer;
        }        public override IEnumerable Stages => _stages;        public override RenderedPipelineDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry)
        {            //      当前集合进行存储当前聚合管道所有聚合项的Bsondocument
            var pipeline = new List();            IBsonSerializer currentSerializer = inputSerializer;
            foreach (var stage in _stages)
            {                //      获取每一个聚合项的RenderedPipelineDefinition
                //      然后获取每个聚合项RenderedPipelineDefinition中的Bsondocument
                var renderedStage = stage.Render(currentSerializer, serializerRegistry);
                currentSerializer = renderedStage.OutputSerializer;                if (renderedStage.document.ElementCount > 0)
                {
                    pipeline.Add(renderedStage.document);
                }
            }            return new RenderedPipelineDefinition(
                pipeline,
                _outputSerializer ?? (currentSerializer as IBsonSerializer) ?? serializerRegistry.GetSerializer());
        }//      BsondocumentStagePipelineDefinition
  public sealed class BsondocumentStagePipelineDefinition : PipelineDefinition    {        private readonly List _stages;        public BsondocumentStagePipelineDefinition(IEnumerable stages, IBsonSerializer outputSerializer = null)
        {
            _stages = stages;
            _outputSerializer = outputSerializer;
        }        public override IBsonSerializer OutputSerializer => _outputSerializer;        public IList documents
        {            get { return _stages; }
        }        //      获取当前聚合的所有聚合项
        public override IEnumerable Stages => _stages.Select(s => new BsondocumentPipelineStageDefinition(s, _outputSerializer));   
        public override RenderedPipelineDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry)
        {            return new RenderedPipelineDefinition(
                _stages,
                _outputSerializer ?? (inputSerializer as IBsonSerializer) ?? serializerRegistry.GetSerializer());
        }
    }

上面是这个三个派生类型基本实现，基本上也都没什么特别的地方，而逻辑也是在Render()这个方法中，EmptyPipelineDefinition中创建了一个空的Bsondocument对象集合实例化的RenderedPipelineDefinition，而BsondocumentStagePipelineDefinition和PipelineStagePipelineDefinition分别以传入的Bsondocument集合和从管道项对象中调用的Render()中获取Bsondocument集合。从这里可以得出2点

1.RenderedPipelineStageDefinition的作用是为了提供其内部的Bsondocument然后创建RenderedPipelineDefinition对象
2.RenderedPipelineStageDefinition和RenderedPipelineDefinition的关系就像BsondocumentPipelineStageDefinition和BsondocumentStagePipelineDefinition关系类似，一个对应管道项，一个对应管道

至此，一切的源头都指向了RenderedPipelineDefinition这个类，但是这个类在下面再介绍，先来看一下改变性的PipelineDefinition

改变性PipelineDefinition

为什么我叫它为改变性呢，因为它是在一个已有PipelineDefinition基础上进行的添加或者替换，下面来看看这三个派生类型

PrependedStagePipelineDefinition：在一个PipelineDefinition管道前面添加一个管道项
AppendedStagePipelineDefinition：在一个PipelineDefinition管道后面添加一个管道项
ReplaceOutputSerializerPipelineDefinition：替换一个PipelineDefinition的序列化对象类型

其实看到这三个派生类就知道其作用了，所以在这里也不进行详细介绍了，只贴出它们的构造方法，有兴趣的朋友可以翻阅源码

//      PrependedStagePipelineDefinitionpublic sealed class PrependedStagePipelineDefinition : PipelineDefinition
 {    public PrependedStagePipelineDefinition(
            PipelineStageDefinition stage,
            PipelineDefinition pipeline,
            IBsonSerializer outputSerializer = null)
        {
        }
 } //     AppendedStagePipelineDefinition
 public sealed class AppendedStagePipelineDefinition : PipelineDefinition
 {    public AppendedStagePipelineDefinition(
            PipelineDefinition pipeline,
            PipelineStageDefinition stage,
            IBsonSerializer outputSerializer = null)
        {
        }
 } 
//  ReplaceOutputSerializerPipelineDefinition
 public sealed class ReplaceOutputSerializerPipelineDefinition : PipelineDefinition
 {        public ReplaceOutputSerializerPipelineDefinition(
            PipelineDefinition pipeline,
            IBsonSerializer outputSerializer = null)
        {
        }
 }

外部不可用派生类

这个外部不可用的派生类型是OptimizingPipelineDefinition ，按照翻译看起来像最优的管道，其实在执行操作时都会现将外部定义的PipelineDefinition转换为OptimizingPipelineDefinition 类型，首先先看看这个类型的定义

internal class OptimizingPipelineDefinition : PipelineDefinition    {        private readonly PipelineDefinition _wrapped;        public OptimizingPipelineDefinition(PipelineDefinition wrapped)
        {
            _wrapped = wrapped;
        }        /// 
        public override IEnumerable Stages => _wrapped.Stages;        
        public override RenderedPipelineDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry)
        {            var rendered = _wrapped.Render(inputSerializer, serializerRegistry);            if (rendered.documents.Count > 1)
            {                //      如果有可能，进行组合一下$match
                var firstStage = rendered.documents[0].GetElement(0);                var secondStage = rendered.documents[1].GetElement(0);                if (firstStage.Name == "$match" && secondStage.Name == "$match")
                {                    var combinedFilter = Builders.Filter.And(
                        (Bsondocument)firstStage.Value,
                        (Bsondocument)secondStage.Value);                    var combinedStage = new Bsondocument("$match", combinedFilter.Render(BsondocumentSerializer.Instance, serializerRegistry));

                    rendered.documents[0] = combinedStage;
                    rendered.documents.RemoveAt(1);
                }
            }            return rendered;
        }
    }

可以看到只是在Render()代码中进行了一个轻微的优化操作，这个优化类是针对OfType情况进行优化的，唯一的使用地方是在FilteredMongoCollectionbase这个抽象类中，而这个抽象类的实现类是OfTypeMongoCollection

 internal abstract class FilteredMongoCollectionbase : MongoCollectionbase, IFilteredMongoCollection
 {    //      创建OptimizingPipelineDefinition
      private PipelineDefinition CreateFilteredPipeline(PipelineDefinition pipeline)
        {
            var filterStage = PipelineStageDefinitionBuilder.Match(_filter);
            var filteredPipeline = new PrependedStagePipelineDefinition(filterStage, pipeline);            return new OptimizingPipelineDefinition(filteredPipeline);
        }
 }
 
internal class OfTypeMongoCollection : FilteredMongoCollectionbase
        where TDeriveddocument : TRootdocument
{

}

PipelineDefinitionBuilder类型

PipelineDefinitionBuilder类型是管道系列的一个帮助类，这个与PipelineStageDefinitionBuilder类相似，但又不尽相同，PipelineDefinitionBuilder中定义的都是PipelineDefinition对象的扩展方法，定义了一系列方便的方法

public static class PipelineDefinitionBuilder{
    //      $match
     public static PipelineDefinition Match(            this PipelineDefinition pipeline,
            FilterDefinition filter)
      {            return pipeline.AppendStage(PipelineStageDefinitionBuilder.Match(filter));
      }    //      $project
     public static PipelineDefinition Project(            this PipelineDefinition pipeline,
            ProjectionDefinition projection)
      {            return pipeline.AppendStage(PipelineStageDefinitionBuilder.Project(projection));
      }    
     //     AppendStage
      public static PipelineDefinition AppendStage     (            this PipelineDefinition pipeline,
            PipelineStageDefinition stage,
            IBsonSerializer outputSerializer = null)
        {            return new AppendedStagePipelineDefinition(pipeline, stage, outputSerializer);
        }    
     public static PipelineDefinition As(            this PipelineDefinition pipeline,
            IBsonSerializer outputSerializer = null)
      {            return new ReplaceOutputSerializerPipelineDefinition(pipeline, outputSerializer);
        }
}

其实可以看出从上面几个个方法可以看出其本质还是使用AppendedStagePipelineDefinition和ReplaceOutputSerializerPipelineDefinition。Match()和Project()都是调用了AppendStage(),而这个方法是创建了一个新的AppendedStagePipelineDefinition对象返回。而As()也是创建了一个新的ReplaceOutputSerializerPipelineDefinition返回。其本质没变，但是可以使得整个driver多了扩展性，更加方便了使用。有的聚合项像$addFields并没有封装方法，可能使用率不大，所以并没有封装，像这样的直接就调用AppendStage()即可

PipelineDefinition类总结

通过上面介绍其实可以看出来了，Mongo的C# Driver中聚合操作使用起来特别方便，使用时先创建聚合项对象再创建聚合管道对象还是直接创建聚合管道对象或者直接使用隐式转换都可以。其实不止聚合，C# Driver中各个操作基本都是如此，使用起来都特别方便，既然创建聚合管道实例的方法特别多，所以在这也就不一一列出，只简单的列出几个

1.先实例化聚合项，再实例化聚合管道对象

2.直接使用隐式转换进行创建聚合管道对象

3.使用扩展方法进行创建

RenderedPipelineStageDefinition和RenderedPipelineDefinition介绍

下面我们来说说RenderedPipelineStageDefinition和RenderedPipelineDefinition这两个类，这两个类叫做聚合项和聚合管道的提供者，它们真正提供了聚合的语句。上面已经简单说过，它们分别是聚合项实例和聚合管道实例创建的，并且在PipelineStagePipelineDefinition中也可以看到RenderedPipelineDefinition是根据RenderedPipelineStageDefinition内部Bsondocument进行实例化的，下面先来看一看这两个类型的内部结构

//      RenderedPipelineStageDefinitionpublic class RenderedPipelineStageDefinition : IRenderedPipelineStageDefinition
{    private string _operatorName;    private Bsondocument _document;    private IBsonSerializer _outputSerializer;    public RenderedPipelineStageDefinition(string operatorName, Bsondocument document, IBsonSerializer outputSerializer)
    {
        _operatorName = Ensure.IsNotNull(operatorName, nameof(operatorName));
        _document = Ensure.IsNotNull(document, nameof(document));
        _outputSerializer = Ensure.IsNotNull(outputSerializer, nameof(outputSerializer));
    }    public Bsondocument document
    {
        get { return _document; }
    }    public IBsonSerializer OutputSerializer
    {
        get { return _outputSerializer; }
    }    public string OperatorName
    {
        get { return _operatorName; }
    }

    IBsonSerializer IRenderedPipelineStageDefinition.OutputSerializer
    {
        get { return _outputSerializer; }
    }
}//      RenderedPipelineDefinitionpublic class RenderedPipelineDefinition
{
    private List _documents;    private IBsonSerializer _outputSerializer;    public RenderedPipelineDefinition(IEnumerable documents, IBsonSerializer outputSerializer)
    {
        _documents = Ensure.IsNotNull(documents, nameof(documents)).ToList();
        _outputSerializer = Ensure.IsNotNull(outputSerializer, nameof(outputSerializer));
    }    public IList documents
    {
        get { return _documents; }
    }    public IBsonSerializer OutputSerializer
    {
        get { return _outputSerializer; }
    }
}

可以看到其实最重要是就是内部的Bsondocument这个属性，那么这个属性里面是什么呢，我们先来看一下

可以看出Bsondocument其实存放就是一个聚合项的json字符串，也就是

注：这个Render()是以序列化器类型实例和序列化注册实例进行序列化为字符串的

然后我来验证聚合的最后执行操作，也就是RenderedPipelineDefinition的作用，这个操作是在MongoCollectionImpl中，从下面代码可以看出，使用Render()方法获取聚合管道的真实语句。然后由此语句执行，由此可以看出其实一切的PipelineDefinition对象最终都是生成RenderedPipelineDefinition对象，这个对象携带着执行语句的json字符串形式。

 internal sealed class MongoCollectionImpl : MongoCollectionbase {
   public override IAsyncCursor Aggregate(IClientSessionHandle session, PipelineDefinition pipeline, AggregateOptions options, CancellationToken cancellationToken = default(CancellationToken))
        {        //      获取当前聚合管道对象的语句
            var renderedPipeline = pipeline.Render(_documentSerializer, _settings.SerializerRegistry);
            
            options = options ?? new AggregateOptions();            var last = renderedPipeline.documents.LastOrDefault();            if (last != null && last.GetElement(0).Name == "$out")
            {                var aggregateOperation = CreateAggregateToCollectionOperation(renderedPipeline, options);
                ExecuteWriteOperation(session, aggregateOperation, cancellationToken);                var findOperation = CreateAggregateToCollectionFindOperation(last, renderedPipeline.OutputSerializer, options);                var forkedSession = session.Fork();                var deferredCursor = new DeferredAsyncCursor(
                    () => forkedSession.Dispose(),
                    ct => ExecuteReadOperation(forkedSession, findOperation, ReadPreference.Primary, ct),
                    ct => ExecuteReadOperationAsync(forkedSession, findOperation, ReadPreference.Primary, ct));                return deferredCursor;
            }            else
            {                var aggregateOperation = CreateAggregateOperation(renderedPipeline, options);                return ExecuteReadOperation(session, aggregateOperation, cancellationToken);
            }
        }
 }

聚合操作的执行方法

上面说了整个聚合管道类的体系，下面说一下最后调用的执行方法

执行方法调用的是IMongoCollection对象的Aggregate()方法，这个方法在IMongoCollection类中具有两个重载，都是需要PipelineDefinition为参数的。

在这个方法中还有一个AggregateOptions参数。这个类是执行聚合的一些选择操作。比如是否使用游标，如果内存不足情况下是否允许使用磁盘等等。。

IAsyncCursor Aggregate(PipelineDefinition pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken));

Aggregate()方法会返回一个IAsyncCursor实例，这个对象代表一个游标。

其实在IMongoCollectionExtensions这个扩展类中还具有Aggregate()方法，这个方法也算是另外一种用法。因为这个方法参数并没有PipelineDefinition对象，并且返回类型也不再是IAsyncCursor，而是一个IAggregateFluent类型。IAggregateFluent类型具有一系列方法

  public static IAggregateFluent Aggregate(this IMongoCollection collection, AggregateOptions options = null);

原文出处：https://www.cnblogs.com/yan7/p/10338783.html

作者：莫问今朝·

Mongo C# Driver 聚合使用---深入浅出

C/C++/C#相关栏目本月热门文章