Handling Complex Aggregation Pipelines With C#

1
[BsonIgnoreExtraElements]
2
[BsonNoId]
3
public class Movie
4
{
5
    [BsonElement("title")]
6
    public required string Title { get; set; }
7
8
    [BsonElement("year")]
9
    public required int Year { get; set; }
10
11
    [BsonElement("cast")]
12
    public List<string> Cast { get; set; } = new();
13
14
    [BsonElement("imdb")]
15
    public Imdb Imdb { get; set; } = new();
16
}
17
18
[BsonIgnoreExtraElements]
19
public class Imdb
20
{
21
    [BsonElement("rating")]
22
    public double Rating { get; set; }
23
}
24
25
public class RatingByYear
26
{
27
    [BsonId]
28
    public int Year { get; set; }
29
30
    [BsonElement("rating")]
31
    public double AvgRating { get; set; }
32
}

1
var pipelineStr = """
2
    [
3
      {
4
        $match: {
5
          cast: "Robert De Niro"
6
        }
7
      },
8
      {
9
        $group: {
10
          _id: "$year",
11
          rating: { $avg: "$imdb.rating" }
12
        }
13
      },
14
      {
15
        $sort: {
16
          rating: -1
17
        }
18
      }
19
    ]
20
    """;
21
22
PipelineDefinition<Movie, RatingByYear> pipeline = BsonSerializer.Deserialize<BsonArray>(pipelineStr)
23
    .Select(x => x.AsBsonDocument)
24
    .ToArray();
25
26
var result = await (await movies.AggregateAsync<RatingByYear>(pipeline)).ToListAsync();

1
// Be sure to use the correct resource name, the schema is as follows: 
2
// <PROJECT ROOT NAMESPACE>.<Folders>.<FileName>
3
// If the resource name is invalid, null is returned 
4
using var stream = typeof(Program).Assembly.GetManifestResourceStream("MyNamespace.Pipelines.RatingByYear.json");
5
using var reader = new StreamReader(stream!);
6
var pipelineStr = await reader.ReadToEndAsync();
7
8
PipelineDefinition<Movie, RatingByYear> pipeline = BsonSerializer.Deserialize<BsonArray>(pipelineStr)
9
    .Select(x => x.AsBsonDocument)
10
    .ToArray();
11
12
var result = await (await movies.AggregateAsync(pipeline)).ToListAsync();

1
[
2
  {
3
    "$match": {
4
      "$expr": {
5
        "$in": [
6
          "$$castMember",
7
          {
8
            "$ifNull": [
9
              "$cast",
10
              []
11
            ]
12
          }
13
        ]
14
      }
15
    }
16
  },
17
  {
18
    "$group": {
19
      "_id": "$year",
20
      "rating": { "$avg": "$imdb.rating" }
21
    }
22
  },
23
  {
24
    "$setWindowFields": {
25
      "sortBy": { "rating": -1 },
26
      "output": {
27
        "docNo": {
28
          "$documentNumber": {}
29
        }
30
      }
31
    }
32
  },
33
  {
34
    "$match": {
35
      "$expr": {
36
        "$lte": [ "$docNo", "$$topN" ]
37
      }
38
    }
39
  },
40
  {
41
    "$unset": "docNo"
42
  }
43
]

1
using var stream = typeof(Program).Assembly.GetManifestResourceStream("MyNamespace.Pipelines.RatingByYearParameterized.json");
2
using var reader = new StreamReader(stream!);
3
var pipelineStr = await reader.ReadToEndAsync();
4
5
PipelineDefinition<Movie, RatingByYear> pipeline = BsonSerializer.Deserialize<BsonArray>(pipelineStr)
6
    .Select(x => x.AsBsonDocument)
7
    .ToArray();
8
9
var options = new AggregateOptions()
10
{
11
    Let = new BsonDocument()
12
    {
13
        { "castMember", "Al Pacino" },
14
        { "topN", 10 },
15
    }
16
};
17
18
var result = await (await movies.AggregateAsync(pipeline, options)).ToListAsync();

1
{
2
  $match: {
3
    cast: "$$castMember"
4
  }
5
},

1
{
2
  "$match": {
3
    "$expr": {
4
      "$in": [
5
        "$$castMember",
6
        {
7
          "$ifNull": [
8
            "$cast",
9
            []
10
          ]
11
        }
12
      ]
13
    }
14
  }
15
},

1
{
2
  "$setWindowFields": {
3
    "sortBy": { "rating": -1 },
4
    "output": {
5
      "docNo": {
6
        "$documentNumber": {}
7
      }
8
    }
9
  }
10
},
11
{
12
  "$match": {
13
    "$expr": {
14
      "$lte": [ "$docNo", "$$topN" ]
15
    }
16
  }
17
},
18
{
19
  "$unset": "docNo"
20
}