diff --git a/src/lib/measurements/day-mean.ts b/src/lib/measurements/day-mean.ts index 38b2e2970..9d90908bc 100644 --- a/src/lib/measurements/day-mean.ts +++ b/src/lib/measurements/day-mean.ts @@ -245,34 +245,32 @@ function weightedRowsSql( if (!/^[a-z_][a-z0-9_]*$/.test(source)) { throw new Error(`source must be a CTE name: ${source}`); } - const hourOf = (alias: string) => - `date_trunc('hour', ${frameTs(frame, `${alias}."measured_at"`)})`; - const dayOf = (alias: string) => - `date_trunc('day', ${frameTs(frame, `${alias}."measured_at"`)})`; - const src = bySource ? `, t."source"` : ""; - const part = bySource ? `, x."source"` : ""; - const join = bySource ? ` AND hw."source" = s."source"` : ""; + // One scan of the CTE and window functions only. The earlier shape joined + // the CTE to an hour grouping of itself; with stale or missing statistics + // (a fresh import below the autoanalyze threshold) the planner ran that + // join as a nested loop and a year of pulse ran into the statement timeout. + // + // `dm_hour_n` counts the readings in a row's hour and `dm_hour_first` marks + // one row per hour, so the sum of `dm_hour_first` over a day is the day's + // hour count. Both columns ride along on the row; every caller aggregates. + const partition = (alias: string, unit: "hour" | "day") => { + const keys = bySource + ? `${alias}."type", ${alias}."source"` + : `${alias}."type"`; + const ts = frameTs(frame, `${alias}."measured_at"`); + return `PARTITION BY ${keys}, date_trunc('${unit}', ${ts})`; + }; return `( SELECT s.*, (CASE WHEN ${isHourlyMeanTypeSql('s."type"')} - THEN 1.0 / (hw.n * hw.hours) + THEN 1.0 / (s.dm_hour_n * SUM(s.dm_hour_first) OVER (${partition("s", "day")})) ELSE 1.0 END)::double precision AS day_weight - FROM ${source} s - LEFT JOIN ( - SELECT x."type"${part}, x.local_hour, x.n, - COUNT(*) OVER (PARTITION BY x."type"${part}, x.local_day) AS hours - FROM ( - SELECT t."type"${src}, - ${hourOf("t")} AS local_hour, - ${dayOf("t")} AS local_day, - COUNT(*) AS n - FROM ${source} t - WHERE ${isHourlyMeanTypeSql('t."type"')} - GROUP BY ${bySource ? "1, 2, 3, 4" : "1, 2, 3"} - ) x - ) hw - ON hw."type" = s."type"${join} - AND hw.local_hour = ${hourOf("s")} + FROM ( + SELECT t.*, + COUNT(*) OVER (${partition("t", "hour")}) AS dm_hour_n, + (ROW_NUMBER() OVER (${partition("t", "hour")}) = 1)::int AS dm_hour_first + FROM ${source} t + ) s )`; } diff --git a/tests/integration/day-weight-unanalysed-scale.test.ts b/tests/integration/day-weight-unanalysed-scale.test.ts new file mode 100644 index 000000000..977f32575 --- /dev/null +++ b/tests/integration/day-weight-unanalysed-scale.test.ts @@ -0,0 +1,232 @@ +/** + * The shared pulse day weight (`dayWeightedRows`) must not depend on planner + * statistics. A new account importing a year of pulse lands below the + * autoanalyze threshold, so the first reads run against a table the planner + * believes is empty. The earlier shape joined the CTE to an hour grouping of + * itself; on such a table the planner chose a nested loop and the all-time + * pulse aggregate ran into the 60 s production statement timeout. + * + * This file seeds a year of dense pulse (well over 100k rows) with autovacuum + * off for the table and no ANALYZE, then: + * + * - times the real all-time reader (`readAllTimeExtremes`) and a bare + * day-weighted mean, each well under the timeout; + * - after ANALYZE, compares the weights per day, per frame and per source + * against the earlier self-join SQL kept below as the oracle. + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import type { Prisma } from "@/generated/prisma/client"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { + SESSION_DAY_FRAME, + dayWeightedRows, + zoneDayFrame, + type DayFrame, +} from "@/lib/measurements/day-mean"; + +const prisma = getPrismaClient(); +const USER = "day-weight-scale"; +const DAY = 86_400_000; +const MINUTE = 60_000; +const DAYS = 365; +/** Generous against a 60 s production timeout, far above what one scan takes. */ +const BOUND_MS = 15_000; + +const start = + Date.UTC( + new Date().getUTCFullYear(), + new Date().getUTCMonth(), + new Date().getUTCDate(), + ) - + DAYS * DAY; + +/** + * About 330 readings a day, unevenly spread so hours carry different counts, + * plus a second source on every fifth day so the per-source weights differ. + */ +function seedRows(): Prisma.MeasurementCreateManyInput[] { + const rows: Prisma.MeasurementCreateManyInput[] = []; + let i = 0; + for (let d = 0; d < DAYS; d += 1) { + for (let m = 0; m < 24 * 60; m += 4) { + if ((m + d) % 13 === 0) continue; + const hour = Math.floor(m / 60); + // Night hours sparse, day hours dense. + if (hour < 6 && m % 12 !== 0) continue; + rows.push({ + id: `${USER}-a-${i}`, + userId: USER, + type: "PULSE", + unit: "bpm", + source: "APPLE_HEALTH", + value: 55 + ((i * 7 + d) % 50), + measuredAt: new Date(start + d * DAY + m * MINUTE + (i % 3) * 1000), + }); + i += 1; + } + if (d % 5 === 0) { + for (const h of [7, 7, 13, 22]) { + rows.push({ + id: `${USER}-w-${i}`, + userId: USER, + type: "PULSE", + unit: "bpm", + source: "WITHINGS", + value: 60 + (i % 20), + measuredAt: new Date( + start + d * DAY + h * 3_600_000 + (i % 50) * MINUTE, + ), + }); + i += 1; + } + } + } + return rows; +} + +/** The earlier self-join, kept verbatim as the oracle for the weights. */ +function oracleWeightedRows( + source: string, + frame: DayFrame, + bySource: boolean, +): string { + const ts = (column: string) => + frame.kind === "session" + ? column + : `((${column} AT TIME ZONE 'UTC') AT TIME ZONE ${frame.tzSql})`; + const hourOf = (alias: string) => + `date_trunc('hour', ${ts(`${alias}."measured_at"`)})`; + const dayOf = (alias: string) => + `date_trunc('day', ${ts(`${alias}."measured_at"`)})`; + const src = bySource ? `, t."source"` : ""; + const part = bySource ? `, x."source"` : ""; + const join = bySource ? ` AND hw."source" = s."source"` : ""; + return `( + SELECT s.*, + (CASE WHEN (s."type")::text IN ('PULSE') + THEN 1.0 / (hw.n * hw.hours) + ELSE 1.0 END)::double precision AS day_weight + FROM ${source} s + LEFT JOIN ( + SELECT x."type"${part}, x.local_hour, x.n, + COUNT(*) OVER (PARTITION BY x."type"${part}, x.local_day) AS hours + FROM ( + SELECT t."type"${src}, + ${hourOf("t")} AS local_hour, + ${dayOf("t")} AS local_day, + COUNT(*) AS n + FROM ${source} t + WHERE (t."type")::text IN ('PULSE') + GROUP BY ${bySource ? "1, 2, 3, 4" : "1, 2, 3"} + ) x + ) hw + ON hw."type" = s."type"${join} + AND hw.local_hour = ${hourOf("s")} + )`; +} + +const SRC_CTE = `WITH src AS ( + SELECT m."id", m."type", m."source", m."measured_at", m."value" + FROM measurements m + WHERE m."user_id" = $1 AND m."deleted_at" IS NULL AND m."type" = 'PULSE' +)`; + +async function timed(fn: () => Promise): Promise<[T, number]> { + const t0 = performance.now(); + const out = await fn(); + return [out, performance.now() - t0]; +} + +let rowCount = 0; + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements SET (autovacuum_enabled = false)", + ); + await prisma.user.create({ data: { id: USER, username: USER } }); + const rows = seedRows(); + rowCount = rows.length; + for (let k = 0; k < rows.length; k += 20_000) { + await prisma.measurement.createMany({ data: rows.slice(k, k + 20_000) }); + } +}, 180_000); + +afterAll(async () => { + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements RESET (autovacuum_enabled)", + ); +}); + +describe("day weight on an unanalysed table", () => { + it("seeds a year of dense pulse", () => { + expect(rowCount).toBeGreaterThan(100_000); + }); + + it("answers the all-time pulse aggregate well under the timeout", async () => { + const [out, ms] = await timed(() => readAllTimeExtremes(USER, ["PULSE"])); + console.info(`[day-weight] readAllTimeExtremes ${ms.toFixed(0)} ms`); + expect(out.get("PULSE")?.mean).toBeGreaterThan(55); + expect(ms).toBeLessThan(BOUND_MS); + }); + + it("answers a bare day-weighted mean well under the timeout", async () => { + const [rows, ms] = await timed(() => + prisma.$queryRawUnsafe>( + `${SRC_CTE} + SELECT (SUM(w."value" * w.day_weight) / SUM(w.day_weight))::double precision AS mean + FROM ${dayWeightedRows("src", zoneDayFrame("$2"), { bySource: true })} w`, + USER, + "Europe/Berlin", + ), + ); + console.info(`[day-weight] zone + bySource mean ${ms.toFixed(0)} ms`); + expect(rows[0].mean).toBeGreaterThan(55); + expect(ms).toBeLessThan(BOUND_MS); + }); +}); + +describe("day weight against the earlier self-join", () => { + beforeAll(async () => { + // Statistics now, so the oracle plans the way it did in production. + await prisma.$executeRawUnsafe("ANALYZE measurements"); + }, 120_000); + + const cases: Array<[string, DayFrame, boolean, unknown[]]> = [ + ["session frame", SESSION_DAY_FRAME, false, []], + ["session frame by source", SESSION_DAY_FRAME, true, []], + ["Berlin frame", zoneDayFrame("$2"), false, ["Europe/Berlin"]], + ["Kolkata frame by source", zoneDayFrame("$2"), true, ["Asia/Kolkata"]], + ]; + + it.each(cases)( + "%s gives every row the same weight", + async (_label, frame, bySource, extra) => { + const query = (sub: string) => + prisma.$queryRawUnsafe>( + `${SRC_CTE} + SELECT w."id" AS id, w.day_weight AS w + FROM ${sub} w + ORDER BY w."id"`, + USER, + ...extra, + ); + const fresh = await query(dayWeightedRows("src", frame, { bySource })); + const oracle = await query(oracleWeightedRows("src", frame, bySource)); + expect(fresh.length).toBe(rowCount); + expect(fresh.length).toBe(oracle.length); + let mismatches = 0; + for (let k = 0; k < fresh.length; k += 1) { + if ( + fresh[k].id !== oracle[k].id || + Math.abs(fresh[k].w - oracle[k].w) > 1e-12 + ) { + mismatches += 1; + } + } + expect(mismatches).toBe(0); + }, + ); +}); diff --git a/tests/integration/mcp-metric-series-source-snapshot.test.ts b/tests/integration/mcp-metric-series-source-snapshot.test.ts index 3759189dc..ab95bc21a 100644 --- a/tests/integration/mcp-metric-series-source-snapshot.test.ts +++ b/tests/integration/mcp-metric-series-source-snapshot.test.ts @@ -218,13 +218,10 @@ beforeAll(async () => { // The mood feature warms its rollup tier fire-and-forget on first read; // settle it now so both builds read the same tier. await ensureUserMoodRollupsFresh(userId); - // Planner statistics for the rows just written. Without them the - // all-time pulse aggregate (`readAllTimeExtremes`, day-weighted for pulse) - // plans against an empty table and runs for ~14 s per call until - // autovacuum's analyze happens to land, which on a loaded runner took the - // first case past its 60 s budget. A production table always has - // statistics, so this measures what production runs. - await prisma.$executeRawUnsafe("ANALYZE measurements, measurement_rollups"); + // No ANALYZE here, on purpose: a fresh import below the autoanalyze + // threshold is what production runs too, and the all-time pulse aggregate + // must not depend on planner statistics + // (`day-weight-unanalysed-scale.test.ts`). }, 120_000); describe("get_metric_series built from one source's rows", () => {