Getting Started with Aggregation Pipelines in Rust

1
{
2
   '_id': ObjectId('573a1392f29313caabcdb497'),
3
   'awards': {'nominations': 7,
4
               'text': 'Won 1 Oscar. Another 2 wins & 7 nominations.',
5
               'wins': 3},
6
   'cast': ['Janet Gaynor', 'Fredric March', 'Adolphe Menjou', 'May Robson'],
7
   'countries': ['USA'],
8
   'directors': ['William A. Wellman', 'Jack Conway'],
9
   'fullplot': 'Esther Blodgett is just another starry-eyed farm kid trying to '
10
               'break into the movies. Waitressing at a Hollywood party, she '
11
               'catches the eye of alcoholic star Norman Maine, is given a test, '
12
               'and is caught up in the Hollywood glamor machine (ruthlessly '
13
               'satirized). She and her idol Norman marry; but his career '
14
               'abruptly dwindles to nothing',
15
   'genres': ['Drama'],
16
   'imdb': {'id': 29606, 'rating': 7.7, 'votes': 5005},
17
   'languages': ['English'],
18
   'lastupdated': '2015-09-01 00:55:54.333000000',
19
   'plot': 'A young woman comes to Hollywood with dreams of stardom, but '
20
            'achieves them only with the help of an alcoholic leading man whose '
21
            'best days are behind him.',
22
   'poster': 'https://m.media-amazon.com/images/M/MV5BMmE5ODI0NzMtYjc5Yy00MzMzLTk5OTQtN2Q3MzgwOTllMTY3XkEyXkFqcGdeQXVyNjc0MzMzNjA@._V1_SY1000_SX677_AL_.jpg',
23
   'rated': 'NOT RATED',
24
   'released': datetime.datetime(1937, 4, 27, 0, 0),
25
   'runtime': 111,
26
   'title': 'A Star Is Born',
27
   'tomatoes': {'critic': {'meter': 100, 'numReviews': 11, 'rating': 7.4},
28
               'dvd': datetime.datetime(2004, 11, 16, 0, 0),
29
               'fresh': 11,
30
               'lastUpdated': datetime.datetime(2015, 8, 26, 18, 58, 34),
31
               'production': 'Image Entertainment Inc.',
32
               'rotten': 0,
33
               'viewer': {'meter': 79, 'numReviews': 2526, 'rating': 3.6},
34
               'website': 'http://www.vcientertainment.com/Film-Categories?product_id=73'},
35
   'type': 'movie',
36
   'writers': ['Dorothy Parker (screen play)',
37
               'Alan Campbell (screen play)',
38
               'Robert Carson (screen play)',
39
               'William A. Wellman (from a story by)',
40
               'Robert Carson (from a story by)'],
41
   'year': 1937}

1
// Load the MongoDB connection string from an environment variable:
2
let client_uri =
3
   env::var("MONGODB_URI").expect("You must set the MONGODB_URI environment var!");
4
5
// An extra line of code to work around a DNS issue on Windows:
6
let options =
7
   ClientOptions::parse_with_resolver_config(&client_uri, ResolverConfig::cloudflare())
8
      .await?;
9
let client = mongodb::Client::with_options(options)?;
10
11
// Get the 'movies' collection from the 'sample_mflix' database:
12
let movies = client.database("sample_mflix").collection("movies");

1
// Usually implemented outside your main function:
2
#[derive(Deserialize)]
3
struct MovieSummary {
4
   title: String,
5
   cast: Vec<String>,
6
   year: i32,
7
}
8
9
impl fmt::Display for MovieSummary {
10
   fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
11
      write!(
12
         f,
13
         "{}, {}, {}",
14
         self.title,
15
         self.cast.get(0).unwrap_or(&"- no cast -".to_owned()),
16
         self.year
17
      )
18
   }
19
}
20
21
// Inside main():
22
let pipeline = vec![
23
   doc! {
24
      // filter on movie title:
25
      "$match": {
26
         "title": "A Star Is Born"
27
      }
28
   },
29
   doc! {
30
      // sort by year, ascending:
31
      "$sort": {
32
         "year": 1
33
      }
34
   },
35
];
36
37
// Look up "A Star is Born" in ascending year order:
38
let mut results = movies.aggregate(pipeline, None).await?;
39
// Loop through the results, convert each of them to a MovieSummary, and then print out.
40
while let Some(result) = results.next().await {
41
   // Use serde to deserialize into the MovieSummary struct:
42
   let doc: MovieSummary = bson::from_document(result?)?;
43
   println!("* {}", doc);
44
}

1
* A Star Is Born, Janet Gaynor, 1937
2
* A Star Is Born, Judy Garland, 1954
3
* A Star Is Born, Barbra Streisand, 1976

1
// Match title = "A Star Is Born":
2
let stage_match_title = doc! {
3
   "$match": {
4
      "title": "A Star Is Born"
5
   }
6
};
7
8
// Sort by year, ascending:
9
let stage_sort_year_ascending = doc! {
10
   "$sort": { "year": 1 }
11
};
12
13
// Now the pipeline is easier to read:
14
let pipeline = vec![stage_match_title, stage_sort_year_ascending];

1
// Sort by year, descending:
2
let stage_sort_year_descending = doc! {
3
   "$sort": {
4
      "year": -1
5
   }
6
};
7
8
// Limit to 1 document:
9
let stage_limit_1 = doc! { "$limit": 1 };
10
11
let pipeline = vec![stage_match_title, stage_sort_year_descending, stage_limit_1];

1
* A Star Is Born, Barbra Streisand, 1976

1
{
2
   '_id': ObjectId('5a9427648b0beebeb69579d3'),
3
   'movie_id': ObjectId('573a1390f29313caabcd4217'),
4
   'date': datetime.datetime(1983, 4, 27, 20, 39, 15),
5
   'email': 'cameron_duran@fakegmail.com',
6
   'name': 'Cameron Duran',
7
   'text': 'Quasi dicta culpa asperiores quaerat perferendis neque. Est animi '
8
           'pariatur impedit itaque exercitationem.'}

1
#[derive(Deserialize)]
2
struct MovieSummary {
3
   title: String,
4
   cast: Vec<String>,
5
   year: i32,
6
   #[serde(default, rename = "related_comments")]
7
   comments: Vec<Comment>,
8
}
9
10
#[derive(Debug, Deserialize)]
11
struct Comment {
12
   email: String,
13
   name: String,
14
   text: String,
15
}

1
// Look up related documents in the 'comments' collection:
2
let stage_lookup_comments = doc! {
3
   "$lookup": {
4
      "from": "comments",
5
      "localField": "_id",
6
      "foreignField": "movie_id",
7
      "as": "related_comments",
8
   }
9
};
10
11
// Limit to the first 5 documents:
12
let stage_limit_5 = doc! { "$limit": 5 };
13
14
let pipeline = vec![
15
   stage_lookup_comments,
16
   stage_limit_5,
17
];
18
19
let mut results = movies.aggregate(pipeline, None).await?;
20
// Loop through the results and print a summary and the comments:
21
while let Some(result) = results.next().await {
22
   let doc: MovieSummary = bson::from_document(result?)?;
23
   println!("* {}, comments={:?}", doc, doc.comments);
24
}

1
// Calculate the number of comments for each movie:
2
let stage_add_comment_count = doc! {
3
   "$addFields": {
4
      "comment_count": {
5
         "$size": "$related_comments"
6
      }
7
   }
8
};
9
10
// Match movie documents with more than 2 comments:
11
let stage_match_with_comments = doc! {
12
   "$match": {
13
      "comment_count": {
14
         "$gt": 2
15
      }
16
   }
17
};

1
let pipeline = vec![
2
   stage_lookup_comments,
3
   stage_add_comment_count,
4
   stage_match_with_comments,
5
   limit_5,
6
]

1
let mut results = movies.aggregate(pipeline, None).await?;
2
// Loop through the results and print a summary and the comments:
3
while let Some(result) = results.next().await {
4
   let doc: MovieSummary = bson::from_document(result?)?;
5
   println!("* {}", doc);
6
   if doc.comments.len() > 0 {
7
      // Print a max of 5 comments per movie:
8
      for comment in doc.comments.iter().take(5) {
9
         println!(
10
            "  - {} <{}>: {}",
11
            comment.name,
12
            comment.email,
13
            comment.text.chars().take(60).collect::<String>(),
14
         );
15
      }
16
   } else {
17
      println!("  - No comments");
18
   }
19
}

1
* Midnight, Claudette Colbert, 1939
2
  - Sansa Stark <sansa_stark@fakegmail.com>: Error ex culpa dignissimos assumenda voluptates vel. Qui inventore 
3
  - Theon Greyjoy <theon_greyjoy@fakegmail.com>: Animi dolor minima culpa sequi voluptate. Possimus necessitatibu
4
  - Donna Smith <donna_smith@fakegmail.com>: Et esse nulla ducimus tempore aliquid. Suscipit iste dignissimos v

1
// Define a struct to hold grouped data by year:
2
#[derive(Debug, Deserialize)]
3
struct YearSummary {
4
   _id: i32,
5
   #[serde(default)]
6
   movie_count: i64,
7
   #[serde(default)]
8
   movie_titles: Vec<String>,
9
}
10
11
// Some movies have "year" values ending with 'è'.
12
// This stage will filter them out:
13
let stage_filter_valid_years = doc! {
14
   "$match": {
15
      "year": {
16
         "$type": "number",
17
      }
18
   }
19
};
20
21
/*
22
* Group movies by year, producing 'year-summary' documents that look like:
23
* {
24
*     '_id': 1917,
25
* }
26
*/
27
let stage_group_year = doc! {
28
   "$group": {
29
      "_id": "$year",
30
   }
31
};
32
33
let pipeline = vec![stage_filter_valid_years, stage_group_year];
34
35
// Loop through the 'year-summary' documents:
36
let mut results = movies.aggregate(pipeline, None).await?;
37
// Loop through the yearly summaries and print their debug representation:
38
while let Some(result) = results.next().await {
39
   let doc: YearSummary = bson::from_document(result?)?;
40
   println!("* {:?}", doc);
41
}

1
* YearSummary { _id: 1959, movie_count: 0, movie_titles: [] }
2
* YearSummary { _id: 1980, movie_count: 0, movie_titles: [] }
3
* YearSummary { _id: 1977, movie_count: 0, movie_titles: [] }
4
* YearSummary { _id: 1933, movie_count: 0, movie_titles: [] }
5
* YearSummary { _id: 1998, movie_count: 0, movie_titles: [] }
6
* YearSummary { _id: 1922, movie_count: 0, movie_titles: [] }
7
* YearSummary { _id: 1948, movie_count: 0, movie_titles: [] }
8
* YearSummary { _id: 1965, movie_count: 0, movie_titles: [] }
9
* YearSummary { _id: 1950, movie_count: 0, movie_titles: [] }
10
* YearSummary { _id: 1968, movie_count: 0, movie_titles: [] }
11
...

1
let stage_group_year = doc! {
2
   "$group": {
3
      "_id": "$year",
4
      // Count the number of movies in the group:
5
      "movie_count": { "$sum": 1 },
6
   }
7
};

1
* YearSummary { _id: 2005, movie_count: 758, movie_titles: [] }
2
* YearSummary { _id: 1999, movie_count: 542, movie_titles: [] }
3
* YearSummary { _id: 1943, movie_count: 36, movie_titles: [] }
4
* YearSummary { _id: 1926, movie_count: 9, movie_titles: [] }
5
* YearSummary { _id: 1935, movie_count: 40, movie_titles: [] }
6
* YearSummary { _id: 1966, movie_count: 116, movie_titles: [] }
7
* YearSummary { _id: 1971, movie_count: 116, movie_titles: [] }
8
* YearSummary { _id: 1952, movie_count: 58, movie_titles: [] }
9
* YearSummary { _id: 2013, movie_count: 1221, movie_titles: [] }
10
* YearSummary { _id: 1912, movie_count: 2, movie_titles: [] }
11
...

1
* YearSummary { _id: 1986, movie_count: 206, movie_titles: ["Defense of the Realm", "F/X", "Mala Noche", "Witch from Nepal", ... ]}

1
let stage_sort_year_ascending = doc! {
2
   "$sort": {"_id": 1}
3
};
4
5
let pipeline = vec! [
6
   stage_filter_valid_years,  // Match numeric years
7
   stage_group_year,
8
   stage_sort_year_ascending, // Sort by year (which is the unique _id field)
9
]