diff --git a/crates/cli/production/db-backfill/src/operations/calculate_model_weights_usages/run_migration.rs b/crates/cli/production/db-backfill/src/operations/calculate_model_weights_usages/run_migration.rs index 9f5e7c28f6..38541913d7 100644 --- a/crates/cli/production/db-backfill/src/operations/calculate_model_weights_usages/run_migration.rs +++ b/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) -> 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(), diff --git a/crates/lib/mysql_queries/src/queries/model_weights/count/count_all_model_usages_on_date.rs b/crates/lib/mysql_queries/src/queries/model_weights/count/count_all_model_usages_on_date.rs index 839ff14d38..88e6551508 100644 --- a/crates/lib/mysql_queries/src/queries/model_weights/count/count_all_model_usages_on_date.rs +++ b/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, } struct RawResult { token: Option, usage_count: i64, + maybe_previously_cached_model_use_count: Option, } 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(); diff --git a/crates/service/job/analytics_job/src/tasks/update_weekly_model_usage_counts/update_weekly_model_usage_counts.rs b/crates/service/job/analytics_job/src/tasks/update_weekly_model_usage_counts/update_weekly_model_usage_counts.rs index 5e085ab81b..a3b3f0dba7 100644 --- a/crates/service/job/analytics_job/src/tasks/update_weekly_model_usage_counts/update_weekly_model_usage_counts.rs +++ b/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; }