update analytics job to skip rewriting model_usage_counts

crates/cli/production/db-backfill/src/operations/calculate_model_weights_usages/run_migration.rs

@@ -45,16 +45,16 @@ pub async fn run_migration(mysql: Pool<MySql>) -> AnyhowResult<()> {
     info!("Records found: {}", usages.counts.len());
 
     for usage in usages.counts {
-      if usage.record_count == 0 {
+      if usage.latest_usage_count == 0 {
         continue;
       }
 
-      info!("Date: {} Token: {} Uses: {} (batch elapsed: {} seconds)", date, usage.token.as_str(), usage.record_count, start.elapsed().as_secs());
+      info!("Date: {} Token: {} Uses: {} (batch elapsed: {} seconds)", date, usage.token.as_str(), usage.latest_usage_count, start.elapsed().as_secs());
 
       upsert_model_weight_usage_count_for_date(Args {
         model_token: &usage.token,
         date,
-        usage_count: usage.record_count,
+        usage_count: usage.latest_usage_count,
         insert_on_zero: false,
         mysql_executor: &mut *connection,
         phantom: Default::default(),

crates/lib/mysql_queries/src/queries/model_weights/count/count_all_model_usages_on_date.rs

@@ -14,12 +14,22 @@ pub struct ModelUseCounts {
 #[derive(Clone)]
 pub struct ModelUseCount {
   pub token: ModelWeightToken,
-  pub record_count: u64,
+
+  /// This is the actual calculated usage count.
+  /// This is the source of truth and should be written over any inaccurate data.
+  pub latest_usage_count: u64,
+
+  /// This is a join against any potentially previously stored calculation.
+  /// Previous calculations may have happened before end of day and may not be
+  /// correct for the full day totals. If this cached count matches the actual
+  /// count, however, then we can skip rewriting the record.
+  pub maybe_previously_cached_model_use_count: Option<u64>,
 }
 
 struct RawResult {
   token: Option<String>,
   usage_count: i64,
+  maybe_previously_cached_model_use_count: Option<u32>,
 }
 
 pub async fn count_all_model_usages_on_date<'e, 'c, E>(
@@ -38,6 +48,7 @@ pub async fn count_all_model_usages_on_date<'e, 'c, E>(
         r#"
 SELECT
   distinct coalesce(w.token, w_old.token) as token,
+  mwuc.usage_count as maybe_previously_cached_model_use_count,
   COUNT(*) as usage_count
 FROM
     media_files as f
@@ -45,14 +56,18 @@ LEFT OUTER JOIN model_weights as w
    ON f.maybe_origin_model_token = w.token
 LEFT OUTER JOIN model_weights as w_old
    ON f.maybe_origin_model_token = w_old.maybe_migration_old_model_token
+LEFT OUTER JOIN model_weight_usage_counts as mwuc
+   ON coalesce(w.token, w_old.token) = mwuc.token
+   AND mwuc.on_date = ?
 WHERE f.maybe_origin_model_token IS NOT NULL
   AND f.created_at >= ?
   AND f.created_at < ? + INTERVAL 1 DAY
-GROUP BY coalesce(w.token, w_old.token)
+GROUP BY coalesce(w.token, w_old.token), mwuc.usage_count
 ORDER BY usage_count DESC
         "#,
       date,
       date,
+      date,
     );
 
   let results = query.fetch_all(mysql_executor).await?;
@@ -62,7 +77,9 @@ ORDER BY usage_count DESC
         None => None, // NB: This shouldn't happen, but SQLx is strict!
         Some(token) => Some(ModelUseCount {
           token: ModelWeightToken::new_from_str(&token),
-          record_count: try_i64_to_u64_or_min(record.usage_count),
+          latest_usage_count: try_i64_to_u64_or_min(record.usage_count),
+          maybe_previously_cached_model_use_count: record.maybe_previously_cached_model_use_count
+              .map(|n| n as u64),
         })
       })
       .collect();

crates/service/job/analytics_job/src/tasks/update_weekly_model_usage_counts/update_weekly_model_usage_counts.rs

@@ -38,23 +38,34 @@ pub async fn calculate_old_model_analytics(job_state: &JobState) -> AnyhowResult
 
     info!("Records found: {}", usages.counts.len());
 
+    let mut skipped_sparse_updates = 0;
+    let mut skipped_noop_updates = 0;
+
     for usage in usages.counts {
-      if usage.record_count == 0 {
+      if usage.latest_usage_count == 0 {
+        skipped_sparse_updates += 1;
+        continue;
+      }
+      if usage.latest_usage_count == usage.maybe_previously_cached_model_use_count.unwrap_or(0) {
+        skipped_noop_updates += 1;
         continue;
       }
 
-      info!("Date: {} Token: {} Uses: {}", date, usage.token.as_str(), usage.record_count);
+      info!("Date: {} Token: {} Uses: {}", date, usage.token.as_str(), usage.latest_usage_count);
 
       upsert_model_weight_usage_count_for_date(Args {
         model_token: &usage.token,
         date,
-        usage_count: usage.record_count,
+        usage_count: usage.latest_usage_count,
         insert_on_zero: false,
         mysql_executor: &mut *connection,
         phantom: Default::default(),
       }).await?;
     }
 
+    info!("Skipped sparse updates: {}", skipped_sparse_updates);
+    info!("Skipped no-op updates: {}", skipped_noop_updates);
+
     tokio::time::sleep(Duration::from_millis(job_state.sleep_config.between_job_batch_wait_millis)).await;
   }