perf: optimize job_stats storage for timestamps and zero-memory jobs (#8289)

* perf: optimize job_stats storage for timestamps and zero-memory jobs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * chore: update sqlx offline cache nullable metadata Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * refactor: use centisecond offsets for job_stats timestamps (~248 day range) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: update SELECT to use offsets_cs column name Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-08-18 16:02:10 +00:00 · 2026-03-10 04:59:56 +00:00
parent 39e77ecd00
commit 2d8335dc43
9 changed files with 138 additions and 99 deletions
@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid",
+        "Text",
+        "Int4"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0af0e0a1dddeee2021ba060e390e1b60caa3752669636e9fb0817a68121a9451"
+}
@@ -1,17 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Uuid",
-        "Text",
-        "Int4"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "1db82007445ff5f644bb607aa28f5747cb50d193475fff5fcfdde37d1bc74636"
-}
@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid",
+        "Text",
+        "Float4"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "a837494a58ab58bfa18c0385350a861076a5fc4f7eefceb1c0de5cf55293f327"
+}
@@ -1,17 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Uuid",
-        "Text",
-        "Float4"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "d44c37882150532383d1058639f4d3af288a346c50f29809dbc55a34088a0abc"
-}
@@ -0,0 +1,2 @@
+ALTER TABLE job_stats DROP COLUMN IF EXISTS timeseries_start;
+ALTER TABLE job_stats DROP COLUMN IF EXISTS offsets_cs;
@@ -0,0 +1,5 @@
+-- Store timeseries timestamps as a start time + integer centisecond offsets
+-- instead of full TIMESTAMPTZ[] arrays. Saves ~4 bytes per data point.
+-- i32 centiseconds gives ~248 days of range with 10ms precision.
+ALTER TABLE job_stats ADD COLUMN IF NOT EXISTS timeseries_start TIMESTAMPTZ;
+ALTER TABLE job_stats ADD COLUMN IF NOT EXISTS offsets_cs INTEGER[];
@@ -79,7 +79,7 @@ async fn get_job_metrics(
    >,
 ) -> error::JsonResult<JobStatsResponse> {
    let records = sqlx::query_as::<_, JobStatsRecord>(
-        "SELECT * FROM job_stats where workspace_id = $1 and job_id = $2",
+        "SELECT workspace_id, job_id, metric_id, metric_name, metric_kind, scalar_int, scalar_float, timestamps, timeseries_int, timeseries_float, timeseries_start, offsets_cs FROM job_stats WHERE workspace_id = $1 AND job_id = $2",
    )
    .bind(w_id)
    .bind(job_id)
@@ -91,7 +91,7 @@ async fn get_job_metrics(
    let mut timeseries_metrics: Vec<TimeseriesMetric> = vec![];

    for record in records {
-        let metric_id = record.metric_id;
+        let metric_id = record.metric_id.clone();
        match record.metric_kind {
            MetricKind::ScalarInt => {
                let value = record.scalar_int.unwrap_or_default() as f64;
@@ -102,47 +102,43 @@ async fn get_job_metrics(
                scalar_metrics.push(ScalarMetric { metric_id: metric_id.clone(), value });
            }
            MetricKind::TimeseriesInt => {
-                if record.timestamps.clone().unwrap_or_default().len()
-                    != record.timeseries_int.clone().unwrap_or_default().len()
-                {
-                    tracing::warn!("Timeseries metric {} has an invalid shape. It doesn't have one timestamp per measurement. (timestamps: {:?}, measurements: {:?})", metric_id, record.timestamps, record.timeseries_int)
+                let timestamps = resolve_timestamps(&record);
+                let timeseries_int = record.timeseries_int.unwrap_or_default();
+                if timestamps.len() != timeseries_int.len() {
+                    tracing::warn!("Timeseries metric {} has an invalid shape. timestamps: {}, measurements: {}", metric_id, timestamps.len(), timeseries_int.len());
                }
                let (timestamps, timeseries_int) = timeseries_sample(
                    from_timestamp,
                    to_timestamp,
                    timeseries_max_datapoints,
-                    record.timestamps.unwrap_or_default(),
-                    record.timeseries_int.unwrap_or_default(),
+                    timestamps,
+                    timeseries_int,
                );
-                let mut values: Vec<DataPoint> = vec![];
-                for (idx, value) in timeseries_int.iter().enumerate() {
-                    values.push(DataPoint {
-                        timestamp: timestamps[idx],
-                        value: value.to_owned() as f64,
-                    });
-                }
+                let values: Vec<DataPoint> = timestamps
+                    .iter()
+                    .zip(timeseries_int.iter())
+                    .map(|(ts, v)| DataPoint { timestamp: *ts, value: *v as f64 })
+                    .collect();
                timeseries_metrics.push(TimeseriesMetric { metric_id: metric_id.clone(), values });
            }
            MetricKind::TimeseriesFloat => {
-                if record.timestamps.clone().unwrap_or_default().len()
-                    != record.timeseries_int.clone().unwrap_or_default().len()
-                {
-                    tracing::warn!("Timeseries metric {} has an invalid shape. It doesn't have one timestamp per measurement. (timestamps: {:?}, measurements: {:?})", metric_id, record.timestamps, record.timeseries_float)
+                let timestamps = resolve_timestamps(&record);
+                let timeseries_float = record.timeseries_float.unwrap_or_default();
+                if timestamps.len() != timeseries_float.len() {
+                    tracing::warn!("Timeseries metric {} has an invalid shape. timestamps: {}, measurements: {}", metric_id, timestamps.len(), timeseries_float.len());
                }
                let (timestamps, timeseries_float) = timeseries_sample(
                    from_timestamp,
                    to_timestamp,
                    timeseries_max_datapoints,
-                    record.timestamps.unwrap_or_default(),
-                    record.timeseries_float.unwrap_or_default(),
+                    timestamps,
+                    timeseries_float,
                );
-                let mut values: Vec<DataPoint> = vec![];
-                for (idx, value) in timeseries_float.iter().enumerate() {
-                    values.push(DataPoint {
-                        timestamp: timestamps[idx],
-                        value: value.to_owned() as f64,
-                    });
-                }
+                let values: Vec<DataPoint> = timestamps
+                    .iter()
+                    .zip(timeseries_float.iter())
+                    .map(|(ts, v)| DataPoint { timestamp: *ts, value: *v as f64 })
+                    .collect();
                timeseries_metrics.push(TimeseriesMetric { metric_id: metric_id.clone(), values });
            }
        };
@@ -152,6 +148,21 @@ async fn get_job_metrics(
    let response = JobStatsResponse { metrics_metadata, scalar_metrics, timeseries_metrics };
    Ok(Json(response))
 }
+
+/// Reconstruct full timestamps from `timeseries_start` + `offsets_cs` if available,
+/// otherwise fall back to legacy `timestamps` column.
+fn resolve_timestamps(record: &JobStatsRecord) -> Vec<chrono::DateTime<chrono::Utc>> {
+    if let (Some(start), Some(offsets)) = (record.timeseries_start, &record.offsets_cs) {
+        if !offsets.is_empty() {
+            return offsets
+                .iter()
+                .map(|&cs| start + chrono::Duration::milliseconds(cs as i64 * 10))
+                .collect();
+        }
+    }
+    // Legacy fallback: use the full timestamps column
+    record.timestamps.clone().unwrap_or_default()
+}
 #[derive(Deserialize)]
 struct JobProgressSetRequest {
    percent: i32,
@@ -15,9 +15,11 @@ pub struct JobStatsRecord {
    pub timestamps: Option<Vec<chrono::DateTime<chrono::Utc>>>,
    pub timeseries_int: Option<Vec<i32>>,
    pub timeseries_float: Option<Vec<f32>>,
+    pub timeseries_start: Option<chrono::DateTime<chrono::Utc>>,
+    pub offsets_cs: Option<Vec<i32>>,
 }

-#[derive(sqlx::Type, Debug, PartialEq, Deserialize, Serialize)]
+#[derive(sqlx::Type, Debug, Clone, PartialEq, Deserialize, Serialize)]
 #[sqlx(type_name = "METRIC_KIND", rename_all = "snake_case")]
 pub enum MetricKind {
    ScalarInt,
@@ -52,29 +54,21 @@ pub async fn register_metric_for_job(
        return Ok(metric_id);
    }

-    let (scalar_int, scalar_float, timestamps, timeseries_int, timeseries_float) = match metric_kind
-    {
+    let is_timeseries = matches!(
+        metric_kind,
+        MetricKind::TimeseriesInt | MetricKind::TimeseriesFloat
+    );
+
+    let (scalar_int, scalar_float, timeseries_int, timeseries_float) = match metric_kind {
        MetricKind::ScalarInt | MetricKind::ScalarFloat => {
-            (None as Option<i32>, None as Option<f32>, None, None, None)
+            (None as Option<i32>, None as Option<f32>, None, None)
        }
-        MetricKind::TimeseriesInt => (
-            None,
-            None,
-            Some(&[] as &[chrono::DateTime<chrono::Utc>]),
-            Some(&[] as &[i32]),
-            None,
-        ),
-        MetricKind::TimeseriesFloat => (
-            None,
-            None,
-            Some(&[] as &[chrono::DateTime<chrono::Utc>]),
-            None,
-            Some(&[] as &[f32]),
-        ),
+        MetricKind::TimeseriesInt => (None, None, Some(&[] as &[i32]), None),
+        MetricKind::TimeseriesFloat => (None, None, None, Some(&[] as &[f32])),
    };

    sqlx::query(
-        "INSERT INTO job_stats (workspace_id, job_id, metric_id, metric_name, metric_kind, scalar_int, scalar_float, timestamps, timeseries_int, timeseries_float) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
+        "INSERT INTO job_stats (workspace_id, job_id, metric_id, metric_name, metric_kind, scalar_int, scalar_float, timeseries_int, timeseries_float, timeseries_start, offsets_cs) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, CASE WHEN $10 THEN now() ELSE NULL END, CASE WHEN $10 THEN ARRAY[]::int[] ELSE NULL END)",
    )
    .bind(workspace_id)
    .bind(job_id)
@@ -83,9 +77,9 @@ pub async fn register_metric_for_job(
    .bind(metric_kind)
    .bind(scalar_int)
    .bind(scalar_float)
-    .bind(timestamps)
    .bind(timeseries_int)
    .bind(timeseries_float)
+    .bind(is_timeseries)
    .execute(db)
    .warn_after_seconds(1)
    .await?;
@@ -117,6 +111,30 @@ pub async fn record_metric(
    }
    let metric_kind = metric_kind_opt.unwrap();

+    record_metric_impl(db, workspace_id, job_id, metric_id, value, metric_kind).await
+}
+
+/// Record a timeseries metric value without the extra SELECT to look up metric_kind.
+/// Use this when the caller already knows the metric kind (e.g. the worker that registered it).
+pub async fn record_timeseries_value(
+    db: &DB,
+    workspace_id: String,
+    job_id: Uuid,
+    metric_id: String,
+    value: MetricNumericValue,
+    metric_kind: MetricKind,
+) -> error::Result<()> {
+    record_metric_impl(db, workspace_id, job_id, metric_id, value, metric_kind).await
+}
+
+async fn record_metric_impl(
+    db: &DB,
+    workspace_id: String,
+    job_id: Uuid,
+    metric_id: String,
+    value: MetricNumericValue,
+    metric_kind: MetricKind,
+) -> error::Result<()> {
    let (value_int, value_float) = match value {
        MetricNumericValue::Integer(val) => {
            if metric_kind != MetricKind::TimeseriesInt && metric_kind != MetricKind::ScalarInt {
@@ -160,7 +178,7 @@ pub async fn record_metric(
        }
        MetricKind::TimeseriesInt => {
            sqlx::query!(
-                "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
+                "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
                &workspace_id,
                &job_id,
                &metric_id,
@@ -169,7 +187,7 @@ pub async fn record_metric(
        }
        MetricKind::TimeseriesFloat => {
            sqlx::query!(
-                "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
+                "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
                &workspace_id,
                &job_id,
                &metric_id,
@@ -737,21 +737,24 @@ where
                let update_job_row = i == 2 || (!*SLOW_LOGS && (i < 20 || (i < 120 && i % 5 == 0) || i % 10 == 0)) || i % 20 == 0;
                if update_job_row && job_id != Uuid::nil() {
                    if let Connection::Sql(ref db) = conn {
-                        // tracking metric starting at i >= 2 b/c first point it useless and we don't want to track metric for super fast jobs
-                        if i == 2 {
-                            memory_metric_id = job_metrics::register_metric_for_job(
-                                &db,
-                                w_id.to_string(),
-                                job_id,
-                                "memory_kb".to_string(),
-                                job_metrics::MetricKind::TimeseriesInt,
-                                Some("Job Memory Footprint (kB)".to_string()),
-                            )
-                            .await;
-                        }
-                        if let Ok(ref metric_id) = memory_metric_id {
-                            if let Err(err) = job_metrics::record_metric(&db, w_id.to_string(), job_id, metric_id.to_owned(), job_metrics::MetricNumericValue::Integer(current_mem)).await {
-                                tracing::error!("Unable to save memory stat for job {} in workspace {}. Error was: {:?}", job_id, w_id, err);
+                        // Only track memory when it's non-zero (avoids storing all-zero timeseries for jobs that don't report memory)
+                        if current_mem > 0 {
+                            // Register on first non-zero reading (deferred from i==2 to avoid metric for jobs with no memory reporting)
+                            if memory_metric_id.is_err() {
+                                memory_metric_id = job_metrics::register_metric_for_job(
+                                    &db,
+                                    w_id.to_string(),
+                                    job_id,
+                                    "memory_kb".to_string(),
+                                    job_metrics::MetricKind::TimeseriesInt,
+                                    Some("Job Memory Footprint (kB)".to_string()),
+                                )
+                                .await;
+                            }
+                            if let Ok(ref metric_id) = memory_metric_id {
+                                if let Err(err) = job_metrics::record_timeseries_value(&db, w_id.to_string(), job_id, metric_id.to_owned(), job_metrics::MetricNumericValue::Integer(current_mem), job_metrics::MetricKind::TimeseriesInt).await {
+                                    tracing::error!("Unable to save memory stat for job {} in workspace {}. Error was: {:?}", job_id, w_id, err);
+                                }
                            }
                        }
                    }