- match comment_update
- .set(comment_aggregates::hot_rank.eq(hot_rank(
- comment_aggregates::score,
- comment_aggregates::published,
- )))
- .execute(conn)
- {
- Ok(_) => {}
- Err(e) => {
- error!("Failed to update comment_aggregates hot_ranks: {}", e)
+/// Runs the hot rank update query in batches until all rows have been processed.
+/// In `where_clause` and `set_clause`, "a" will refer to the current aggregates table.
+/// Locked rows are skipped in order to prevent deadlocks (they will likely get updated on the next
+/// run)
+fn process_hot_ranks_in_batches(
+ conn: &mut PgConnection,
+ table_name: &str,
+ where_clause: &str,
+ set_clause: &str,
+) {
+ let process_start_time = NaiveDateTime::from_timestamp_opt(0, 0).expect("0 timestamp creation");
+
+ let update_batch_size = 1000; // Bigger batches than this tend to cause seq scans
+ let mut processed_rows_count = 0;
+ let mut previous_batch_result = Some(process_start_time);
+ while let Some(previous_batch_last_published) = previous_batch_result {
+ // Raw `sql_query` is used as a performance optimization - Diesel does not support doing this
+ // in a single query (neither as a CTE, nor using a subquery)
+ let result = sql_query(format!(
+ r#"WITH batch AS (SELECT a.id
+ FROM {aggregates_table} a
+ WHERE a.published > $1 AND ({where_clause})
+ ORDER BY a.published
+ LIMIT $2
+ FOR UPDATE SKIP LOCKED)
+ UPDATE {aggregates_table} a {set_clause}
+ FROM batch WHERE a.id = batch.id RETURNING a.published;
+ "#,
+ aggregates_table = table_name,
+ set_clause = set_clause,
+ where_clause = where_clause
+ ))
+ .bind::<Timestamp, _>(previous_batch_last_published)
+ .bind::<Integer, _>(update_batch_size)
+ .get_results::<HotRanksUpdateResult>(conn);
+
+ match result {
+ Ok(updated_rows) => {
+ processed_rows_count += updated_rows.len();
+ previous_batch_result = updated_rows.last().map(|row| row.published);
+ }
+ Err(e) => {
+ error!("Failed to update {} hot_ranks: {}", table_name, e);
+ break;
+ }