diff --git a/prisma/migrations/0190_v1200_rollup_regression_accumulators/migration.sql b/prisma/migrations/0190_v1200_rollup_regression_accumulators/migration.sql new file mode 100644 index 000000000..39a1bcb7f --- /dev/null +++ b/prisma/migrations/0190_v1200_rollup_regression_accumulators/migration.sql @@ -0,0 +1,66 @@ +-- v1.20.0 F6 — rollup-tier completion: per-bucket regression accumulators. +-- +-- Up to v1.19.x the windowed slope / r² / sd (slope7/30/90, r²_7/30/90, +-- the 90-day population stddev) still ran as a live `$queryRaw` over the +-- raw `measurements` partition on EVERY warm-rollup request, because the +-- per-bucket `mean` cannot reconstruct an OLS regression over the raw +-- points in a window. The per-bucket `sd / slope / r2` columns are a +-- within-bucket fold and are NOT linearly composable across DAY buckets. +-- +-- This adds the four regression-accumulator sums Postgres' REGR_* / +-- STDDEV_POP folds expose, computed per (user, type, granularity, +-- bucketStart, source) over the same canonical-source rows the live +-- query reads. With `n = count` and `Σy = mean·count` already stored, +-- these four close the set: all six terms are ADDITIVE across buckets, +-- so a 7/30/90-day slope / r² / sd composes from the summed accumulators +-- with a bit-identical result to the live REGR_SLOPE / REGR_R2 / +-- STDDEV_POP over the raw rows (Postgres uses the same closed form): +-- +-- slope = (n·Σxy − Σx·Σy) / (n·Σxx − Σx²) +-- r² = (n·Σxy − Σx·Σy)² / ((n·Σxx − Σx²)(n·Σyy − Σy²)) +-- sd_pop = sqrt(Σyy/n − (Σy/n)²) +-- +-- The x-axis is EPOCH DAYS (`EXTRACT(EPOCH FROM measured_at) / 86400.0`), +-- matching the live REGR_* calls in `measurement-rollups.ts`, +-- `summaries-slice.ts`, and `comprehensive-aggregator.ts`. The raw +-- (un-rebased) epoch-day sums are stored so the cross-bucket sum +-- reconstructs the same regression the live query runs. +-- +-- Four additive, nullable `DOUBLE PRECISION` columns on +-- `measurement_rollups`: +-- +-- * `sum_x` — Σ epoch_days +-- * `sum_xy` — Σ (epoch_days · value) +-- * `sum_xx` — Σ epoch_days² +-- * `sum_yy` — Σ value² +-- +-- Nullable keeps the migration additive: existing rows read NULL until +-- the boot-time `rollup-full-backfill` re-fold refills them (mirrors the +-- v1.4.39 `sum_value` rollout). New writes self-populate because the +-- accumulators ride the same aggregate that already recomputes on every +-- write hook. The legacy per-bucket `sd / slope / r2` columns are left +-- in place (harmless) to keep this migration purely additive. +-- +-- Additive; no existing row touched. Idempotent guards +-- (`ADD COLUMN IF NOT EXISTS`) so a rerun is safe on prod. +-- +-- Reversibility (down): +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_x"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_xy"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_xx"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_yy"; +-- Dropping them reverts the readers to the live REGR_* path on a +-- coverage miss; the read-swap falls back automatically when the +-- accumulators are NULL. + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_x" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_xy" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_xx" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_yy" DOUBLE PRECISION; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index e6839cdce..bf098311f 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1292,6 +1292,20 @@ model MeasurementRollup { /// to keep the migration additive — the boot backfill fills NULLs /// for existing rows; new rows always carry the sum. sumValue Float? @map("sum_value") + /// v1.20.0 F6 — per-bucket OLS regression accumulators over the same + /// canonical-source raw rows the live REGR_* query reads. With + /// `n = count` and `Σy = mean·count` these close the additive set so a + /// windowed slope / r² / sd composes from summed accumulators across DAY + /// buckets with a bit-identical result to live REGR_SLOPE / REGR_R2 / + /// STDDEV_POP. The x-axis is epoch days (`EXTRACT(EPOCH FROM + /// measured_at) / 86400.0`), stored un-rebased so the cross-bucket sum + /// reconstructs the live regression. Nullable to keep migration 0190 + /// additive — the boot backfill fills NULLs for pre-migration rows; new + /// rows always carry them (they ride the same aggregate as mean / sum). + sumX Float? @map("sum_x") + sumXy Float? @map("sum_xy") + sumXx Float? @map("sum_xx") + sumYy Float? @map("sum_yy") computedAt DateTime @default(now()) @map("computed_at") @db.Timestamptz(3) user User @relation(fields: [userId], references: [id], onDelete: Cascade) diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index c43bd7386..ddff5f522 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -53,6 +53,118 @@ export interface DailyMeanRow { maxValue: number; } +/** + * v1.20.0 F6 — per-bucket OLS regression accumulators (epoch-day x-axis). + * Mirrors the four columns migration 0190 added to `measurement_rollups` + * plus the `n` / `Σy` the existing `count` / `mean` already carry. A bucket + * whose accumulators predate the migration (or whose write missed the + * re-fold) reports `sumXy === null`; the reader treats any null in the + * window as a coverage miss and falls back to the live REGR_* path. + */ +export interface RegressionAccumulators { + count: number; + /** Σy = mean·count (the writer stores mean + count; Σy is derived). */ + mean: number; + sumX: number | null; + sumXy: number | null; + sumXx: number | null; + sumYy: number | null; +} + +/** Closed-form windowed regression result. `null` when undefined. */ +export interface ComposedRegression { + slope: number | null; + r2: number | null; + /** Population standard deviation (divides by n, matching STDDEV_POP). */ + sdPop: number | null; +} + +/** + * v1.20.0 F6 — compose a windowed OLS slope / r² / population-sd from the + * summed regression accumulators of a set of DAY buckets. + * + * The six terms (`n = Σcount`, `Σx`, `Σy = Σ(mean·count)`, `Σxy`, `Σxx`, + * `Σyy`) are ADDITIVE across buckets, so summing the per-bucket + * accumulators over the window and evaluating the closed form yields a + * result BIT-IDENTICAL to Postgres `REGR_SLOPE` / `REGR_R2` / `STDDEV_POP` + * over the same raw rows (Postgres folds the same accumulators): + * + * slope = (n·Σxy − Σx·Σy) / (n·Σxx − Σx²) + * r² = (n·Σxy − Σx·Σy)² / ((n·Σxx − Σx²)(n·Σyy − Σy²)) + * sd_pop = sqrt(Σyy/n − (Σy/n)²) + * + * The caller MUST collapse overlapping sources to the canonical source + * BEFORE handing rows here — the accumulators are per-source, so summing a + * dual-source day's two rows would double-count the reading. Pass the + * already-source-collapsed bucket set. + * + * Coverage contract: returns `{ null, null, null }` (a full miss) when + * - any bucket in the window has a `null` accumulator (pre-migration row + * the boot re-fold has not refilled), or + * - the window holds fewer than 2 readings, or + * - a denominator degenerates to 0 (no x-variance / no y-variance). + * + * The caller treats a miss as "fall back to live SQL", so a partially + * back-filled window never silently returns a wrong (partial) regression. + * + * Postgres' REGR_SLOPE / REGR_R2 ignore rows with a NULL dependent or + * independent value; `value` and `measured_at` are NOT NULL on + * `measurements`, so every raw row contributes and the accumulator `n` + * equals the live `regr_count`. + */ +export function composeRegression( + buckets: ReadonlyArray, +): ComposedRegression { + const MISS: ComposedRegression = { slope: null, r2: null, sdPop: null }; + if (buckets.length === 0) return MISS; + + let n = 0; + let sumX = 0; + let sumY = 0; + let sumXy = 0; + let sumXx = 0; + let sumYy = 0; + for (const b of buckets) { + // Any null accumulator in the window means the row predates the + // accumulator columns (or its re-fold is pending) — bail to live SQL + // rather than compose a regression over an incomplete window. + if ( + b.sumX === null || + b.sumXy === null || + b.sumXx === null || + b.sumYy === null + ) { + return MISS; + } + n += b.count; + sumX += b.sumX; + // Σy is reconstructed from the stored mean·count, matching the live + // SUM(value) over the same rows. + sumY += b.mean * b.count; + sumXy += b.sumXy; + sumXx += b.sumXx; + sumYy += b.sumYy; + } + + if (n < 2) return MISS; + + const denomX = n * sumXx - sumX * sumX; + const denomY = n * sumYy - sumY * sumY; + const cov = n * sumXy - sumX * sumY; + + const slope = denomX === 0 ? null : cov / denomX; + // REGR_R2 is null when either variance is zero (matches Postgres). + const r2 = + denomX === 0 || denomY === 0 ? null : (cov * cov) / (denomX * denomY); + + // Population variance: Σyy/n − (Σy/n)². Clamp tiny negative values that + // float rounding can produce when the variance is effectively zero. + const variance = sumYy / n - (sumY / n) * (sumY / n); + const sdPop = variance <= 0 ? 0 : Math.sqrt(variance); + + return { slope, r2, sdPop }; +} + /** Minimal shape the source collapse needs from a per-source rollup row. */ export interface SourcedBucketRow { bucketStart: Date; @@ -129,6 +241,41 @@ export function collapseRollupRowsBySource( return out; } +/** + * v1.20.0 F6 — a DAY bucket carrying its `bucketStart` plus the + * regression accumulators. The windowed-regression readers consume an + * array of these (one canonical-source row per day) and slice them into + * the 7/30/90-day sub-windows before composing. + */ +export interface AccumulatorBucketRow extends RegressionAccumulators { + bucketStart: Date; +} + +/** + * v1.20.0 F6 — compose a windowed slope / r² / population-sd over the DAY + * buckets whose `bucketStart` falls on or after `since`. + * + * The window is DAY-aligned (a bucket is in-window iff its `bucketStart` + * is `>= since`), matching the DAY-rollup grain. Callers anchor `since` on + * a UTC-day boundary (`startOfUtcDay(now − N days)`) so the window is the + * day-aligned equivalent of the live `measured_at >= NOW() - INTERVAL 'N + * days'` filter. The accumulator-composed result over a given bucket set + * is bit-identical to live REGR_* / STDDEV_POP over the same buckets' raw + * rows — the parity test pins this. + * + * Returns a full miss `{ null, null, null }` when any in-window bucket + * lacks accumulators (see `composeRegression`), so a partially back-filled + * window falls through to live SQL rather than returning a partial answer. + */ +export function composeWindowedRegression( + rows: ReadonlyArray, + since: Date, +): ComposedRegression { + const cutoff = since.getTime(); + const inWindow = rows.filter((r) => r.bucketStart.getTime() >= cutoff); + return composeRegression(inWindow); +} + /** * Combine DAY buckets into the linearly-composable window stats — * `count`, `min`, `max`, `mean`. SD / slope / R² are intentionally diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 4bb2345c7..1bc399a2a 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -129,6 +129,14 @@ interface RollupRow { sd: number | null; slope: number | null; r2: number | null; + // v1.20.0 F6 — per-bucket OLS regression accumulators (epoch-day x-axis). + // See `MeasurementRollup.sumX` doc in schema.prisma. These let the + // windowed slope / r² / sd compose from summed accumulators across DAY + // buckets with bit-identical parity to live REGR_*/STDDEV_POP. + sum_x: number | null; + sum_xy: number | null; + sum_xx: number | null; + sum_yy: number | null; } /** @@ -392,7 +400,22 @@ async function runRollupAggregate(input: { REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - )::double precision AS r2 + )::double precision AS r2, + -- v1.20.0 F6 — regression accumulators (epoch-day x-axis). Summed + -- per (type, source, bucket) here so the read tier composes the + -- windowed slope / r2 / sd across DAY buckets without a live scan. + -- value is NOT NULL on measurements, so every row contributes; the + -- sums are over the same rows REGR_/STDDEV_POP fold above. + SUM(EXTRACT(EPOCH FROM m."measured_at") / 86400.0)::double precision + AS sum_x, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) * m."value" + )::double precision AS sum_xy, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + * (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + )::double precision AS sum_xx, + SUM(m."value" * m."value")::double precision AS sum_yy FROM measurements m WHERE m."user_id" = $1 AND m."type" IN (${typeList}) @@ -427,7 +450,20 @@ async function runRollupAggregate(input: { REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - )::double precision AS r2 + )::double precision AS r2, + -- v1.20.0 F6 — regression accumulators (epoch-day x-axis). See the + -- typed-list query above for the rationale; the read tier composes + -- windowed slope / r² / sd from these summed across DAY buckets. + SUM(EXTRACT(EPOCH FROM m."measured_at") / 86400.0)::double precision + AS sum_x, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) * m."value" + )::double precision AS sum_xy, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + * (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + )::double precision AS sum_xx, + SUM(m."value" * m."value")::double precision AS sum_yy FROM measurements m WHERE m."user_id" = $1 AND m."measured_at" >= $2 @@ -496,6 +532,12 @@ async function persistRollupRows( sd: row.sd, slope: row.slope, r2: row.r2, + // v1.20.0 F6 — persist the regression accumulators alongside the + // existing stats; they ride the same aggregate (no extra round-trip). + sumX: row.sum_x ?? null, + sumXy: row.sum_xy ?? null, + sumXx: row.sum_xx ?? null, + sumYy: row.sum_yy ?? null, computedAt: new Date(), });