From 2b8eb2c1c228099ff9778e618b0b5b84c816cb3a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 25 Oct 2026 05:46:01 +0100 Subject: [PATCH 1/3] perf(analytics): pick the canonical source with one window over one scan The canonical-source pick grouped the window to (type, day, source), chose each day's source with DISTINCT ON, and joined the choices back to the rows. Whether that join ran as a hash join or as a nested loop that rescanned the whole pick for every row was the planner's call, made from table statistics. On a measurements table without them (a new install, or an import still under the autoanalyze threshold) it estimated one row on each side and took the nested loop. Binding the request's clock as a parameter instead of reading NOW() flipped which side of that loop the pick landed on: the analytics read over 6,000 freshly written blood-pressure rows went from 1.6 s to 8.4 s, and the demo seed's dashboard headline to 20 s. On a ten-year account without statistics the 90-day read took 4 s with either clock and the all-time read ran past two minutes. The pick is now one window function over one scan: FIRST_VALUE of the source per (type, UTC day) in ladder order, keeping the rows of that source. Same rows, same order of preference, so every day keeps its source; the parity suite against the DISTINCT ON statement holds that. The analytics read on the fixture above takes about 0.1 s, the headline 0.4 s. On the analysed ten-year account the 90-day and all-time reads are as fast as the join was, and they no longer depend on statistics. The rows carry one extra column, canonical_source; every reader selects its columns by name. A new integration test runs every reader shape under EXPLAIN ANALYZE on a table with no statistics and bounds the rows the plan produces or discards per row seeded. The statement count never moved, so a statement budget could not see this; the join measured 200 to 8,600 row visits per row seeded and the window stays at 4. --- .../__tests__/bp-in-target-fast-path.test.ts | 2 +- .../__tests__/source-rank-sql.test.ts | 21 +- src/lib/analytics/source-rank-sql.ts | 115 +++++---- src/lib/analytics/summaries-slice.ts | 8 +- src/lib/insights/comprehensive-aggregator.ts | 10 +- .../__tests__/series-canonical.test.ts | 13 +- .../__tests__/source-provenance.test.ts | 8 +- .../canonical-source-pick-parity.test.ts | 6 +- .../canonical-source-pick-unanalysed.test.ts | 233 ++++++++++++++++++ 9 files changed, 324 insertions(+), 92 deletions(-) create mode 100644 tests/integration/canonical-source-pick-unanalysed.test.ts diff --git a/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts b/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts index 2b53ebd7c..ab43556da 100644 --- a/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts +++ b/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts @@ -316,7 +316,7 @@ describe("computeBpInTargetFastPath", () => { // a raw `findMany` (double-counting) fails here. for (const call of QUERY_RAW.mock.calls) { const sql = String(call[0]); - expect(sql).toContain("DISTINCT ON"); + expect(sql).toContain("canonical_source"); expect(sql).toContain("date_trunc('day'"); } }); diff --git a/src/lib/analytics/__tests__/source-rank-sql.test.ts b/src/lib/analytics/__tests__/source-rank-sql.test.ts index 5a5f0cebc..33307d722 100644 --- a/src/lib/analytics/__tests__/source-rank-sql.test.ts +++ b/src/lib/analytics/__tests__/source-rank-sql.test.ts @@ -7,8 +7,10 @@ * self-join. These tests pin: * 1. the alias-wrapped `canonicalMeasurementsFrom` is exactly the CTE body * wrapped in `( … ) m` — so the single-reference call site is unchanged; - * 2. the canonical body carries exactly ONE DISTINCT-ON pick — so a fold - * that references the CTE twice runs the expensive pick once; + * 2. the canonical body reads `measurements` once and picks with one window + * function, with no join for the planner to run as a nested loop — so a + * fold that references the CTE twice runs the pick once, and a table + * without statistics cannot turn it quadratic; * 3. the interval whitelist still rejects an unsafe splice; * 4. the window counts from an instant the caller took, never from the * database clock. @@ -34,21 +36,22 @@ describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => { expect(from).toBe(`(${cte}\n ) m`); }); - it("runs exactly one DISTINCT-ON canonical pick in the body", () => { + it("picks with one window over one scan, with no join", () => { const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS); - const matches = cte.match(/DISTINCT ON/g) ?? []; - expect(matches).toHaveLength(1); + expect(cte.match(/FROM measurements\b/g) ?? []).toHaveLength(1); + expect(cte.match(/FIRST_VALUE\(/g) ?? []).toHaveLength(1); + expect(cte).not.toMatch(/\bJOIN\b|DISTINCT ON/); }); - it("binds the user id as $1 and scopes both the inner and outer windows", () => { + it("binds the user id as $1 and scopes the scan to the window", () => { const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS); // user id is always a bound parameter, never spliced. expect(cte).toContain('"user_id" = $1'); - // The trailing-window bound appears on both the inner pick and the outer - // filter so the folded references stay scoped identically. + // The trailing-window bound scopes the one scan, so the pick and the rows + // it keeps are cut by the same window. const intervalMatches = cte.match(/>= \$2::timestamp - INTERVAL '90 days'/g) ?? []; - expect(intervalMatches).toHaveLength(2); + expect(intervalMatches).toHaveLength(1); }); it("never cuts its window on the database clock", () => { diff --git a/src/lib/analytics/source-rank-sql.ts b/src/lib/analytics/source-rank-sql.ts index 58b015bf5..0c8249d7a 100644 --- a/src/lib/analytics/source-rank-sql.ts +++ b/src/lib/analytics/source-rank-sql.ts @@ -121,35 +121,45 @@ export interface CanonicalWindow { /** * Build the canonical-source SELECT body — the raw `measurements` rows * restricted to the ladder-winning source per (type, day) — WITHOUT the - * outer `( … ) m` alias wrapper. The inner `DISTINCT ON` picks the - * canonical source for each day and the join keeps only that source's - * readings. + * outer `( … ) m` alias wrapper. One scan reads the window's rows; a window + * function names each (type, day)'s canonical source on every row, and the + * rows of any other source are dropped. * * Use this when the same canonical set must be referenced more than once in * one statement (e.g. a `window_stats` CTE plus the outer aggregate): bind it * once via `WITH cm AS (${canonicalMeasurementsCte(...)})` and select - * `FROM cm m` everywhere, so the expensive DISTINCT-ON self-join runs once - * rather than per reference. For the single-reference case use - * `canonicalMeasurementsFrom`, which wraps this body in the alias. + * `FROM cm m` everywhere, so the pick runs once rather than per reference. + * For the single-reference case use `canonicalMeasurementsFrom`, which wraps + * this body in the alias. + * + * The rows carry every `measurements` column plus `canonical_source`, the + * picked source (equal to `source` on every row that survives). Readers + * select the columns they need, so the extra one never reaches them. * * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with - * unqualified `"type"`/`"source"` columns. `window`, when given, scopes both - * the inner pick and the outer filter to the readings from `from` on, or from - * `interval` (e.g. `"90 days"`, whitelisted) before it. `from` is an instant - * the caller took (`boundInstantSql` or `instantLiteralSql`, see - * `request-clock-sql.ts`), the request's `now` for a trailing window, so the - * cut runs on the caller's clock and never on the database's. `onlyType`, when given, scopes both - * halves to that one metric. + * unqualified `"type"`/`"source"` columns. `window`, when given, scopes the + * scan to the readings from `from` on, or from `interval` (e.g. `"90 days"`, + * whitelisted) before it. `from` is an instant the caller took + * (`boundInstantSql` or `instantLiteralSql`, see `request-clock-sql.ts`), the + * request's `now` for a trailing window, so the cut runs on the caller's + * clock and never on the database's. `onlyType`, when given, scopes the scan + * to that one metric. * - * The pick groups before it sorts. Its order (rank, then source name) depends - * only on a row's `(type, source)`, so every reading of one source on one day - * sorts alike, and the distinct `(type, day, source)` triples carry the whole - * decision. A hash aggregate reduces the window's rows to those triples (a - * few thousand) and `DISTINCT ON` sorts only them; sorting every raw row by - * a CASE expression no index can supply spilled to disk on a large account. - * The picked source per day is the same by construction; - * `tests/integration/canonical-source-pick-parity.test.ts` holds the old - * statement as its reference. + * The pick is the first source of the day in ladder order (rank, then source + * name), the same order the `DISTINCT ON` pick it replaced used, over the + * same rows, so every (type, day) keeps the same source; + * `tests/integration/canonical-source-pick-parity.test.ts` holds the + * `DISTINCT ON` statement as its reference. + * + * Why no join: the earlier shape grouped the window to `(type, day, source)`, + * picked per day with `DISTINCT ON`, and joined the picks back to the rows. + * Whether that join ran as a hash join or as a nested loop that rescanned the + * pick for every row was the planner's call, made from table statistics. On + * a table without them (a new install, or an import still under the + * autoanalyze threshold) it estimated one row and chose the nested loop: on a + * 6,000-row fixture the all-time read took seconds, on a ten-year account it + * ran past the statement timeout. A window over one scan has no such choice + * to get wrong; `canonical-source-pick-unanalysed.test.ts` holds that. */ export function canonicalMeasurementsCte( rankUnqualified: string, @@ -161,52 +171,37 @@ export function canonicalMeasurementsCte( : window.interval === undefined ? assertInstantSql(window.from) : `${assertInstantSql(window.from)} - INTERVAL '${assertInterval(window.interval)}'`; - const sinceInner = bound ? `AND "measured_at" >= ${bound}` : ""; - const sinceOuter = bound ? `AND mm."measured_at" >= ${bound}` : ""; - // A single-type reader (a series for one metric) scopes both halves to that - // type, so the pick scans one metric's rows instead of every row the user + const since = bound ? `AND mm."measured_at" >= ${bound}` : ""; + // A single-type reader (a series for one metric) scopes the scan to that + // type, so the pick reads one metric's rows instead of every row the user // has in the window. - const typeInner = onlyType - ? `AND "type" = '${assertEnumLiteral(onlyType)}'::"measurement_type"` - : ""; - const typeOuter = onlyType + const typeOnly = onlyType ? `AND mm."type" = '${assertEnumLiteral(onlyType)}'::"measurement_type"` : ""; return ` - SELECT mm.* - FROM measurements mm - JOIN ( - SELECT DISTINCT ON ("type", d) - "type" AS t, - d, - "source" AS canon - FROM ( - SELECT "type", date_trunc('day', "measured_at") AS d, "source" - FROM measurements - WHERE "user_id" = $1 - AND "deleted_at" IS NULL - ${sinceInner} - ${typeInner} - GROUP BY "type", date_trunc('day', "measured_at"), "source" - ) g - ORDER BY "type", d, (${rankUnqualified}), "source"::text - ) c - ON c.t = mm."type" - AND c.d = date_trunc('day', mm."measured_at") - AND c.canon = mm."source" - WHERE mm."user_id" = $1 - AND mm."deleted_at" IS NULL - ${sinceOuter} - ${typeOuter}`; + SELECT w.* + FROM ( + SELECT mm.*, + FIRST_VALUE(mm."source") OVER ( + PARTITION BY mm."type", date_trunc('day', mm."measured_at") + ORDER BY (${rankUnqualified}), mm."source"::text + ) AS canonical_source + FROM measurements mm + WHERE mm."user_id" = $1 + AND mm."deleted_at" IS NULL + ${since} + ${typeOnly} + ) w + WHERE w."source" = w.canonical_source`; } /** * Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements` - * to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks - * the ladder-winning source for each day, and the join keeps only that source's - * readings. Use it in place of `FROM measurements m` so a live aggregate over - * an overlapping vital never blends two devices — the same collapse the rollup - * readers apply, kept in lockstep for live/rollup parity. + * to the canonical-source rows per (type, day): the ladder-winning source for + * each day, and only that source's readings. Use it in place of + * `FROM measurements m` so a live aggregate over an overlapping vital never + * blends two devices — the same collapse the rollup readers apply, kept in + * lockstep for live/rollup parity. * * The user id is bound as `$1`; `rankUnqualified`, `window` and `onlyType` are * those of `canonicalMeasurementsCte`. diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 3d506ac52..54c5fef72 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -573,10 +573,10 @@ async function computeFromRollups( const [narrows, latests, dayBuckets, accumulatorRows, preFoldRemainder] = await Promise.all([ - // The FROM is restricted to the canonical-source rows per (type, day): the - // inner DISTINCT ON picks the ladder-winning source for each day, and the - // join keeps only that source's readings so the 90-day AVG / median / slope - // never blend two devices that both reported the same vital. + // The FROM is restricted to the canonical-source rows per (type, day): + // the ladder-winning source for each day, and only that source's + // readings, so the 90-day AVG / median / slope never blend two devices + // that both reported the same vital. // The window means weigh each day once for an hourly-mean type // (`windowedMean`); the median stays over every reading. prisma.$queryRawUnsafe( diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 6dcef4f28..09a7c6ebb 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -403,10 +403,10 @@ async function buildFromRollups( timeSubquery(timings, "narrow", () => prisma.$queryRawUnsafe( ` - -- v1.18.11 perf#3a — the canonical-source self-join is the dominant + -- v1.18.11 perf#3a — the canonical-source pick is the dominant -- cost; bind it ONCE as a CTE and reference it from both window_stats - -- and the outer aggregate so the DISTINCT-ON pick runs a single time - -- instead of twice per rebuild. Output is identical. + -- and the outer aggregate so the pick runs a single time instead of + -- twice per rebuild. Output is identical. WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)} ), -- The z-score's mean and spread stay over every reading: they measure @@ -626,9 +626,9 @@ async function buildFromLiveAggregate( timeSubquery(timings, "heavy", () => prisma.$queryRawUnsafe( ` - -- v1.18.11 perf#3a — bind the canonical-source self-join once (see the + -- v1.18.11 perf#3a — bind the canonical-source pick once (see the -- warm narrow query). window_stats + the outer aggregate both read cm, - -- so the DISTINCT-ON pick runs once. Output is identical. + -- so the pick runs once. Output is identical. WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)} ), -- The z-score's mean and spread stay over every reading: they measure diff --git a/src/lib/measurements/__tests__/series-canonical.test.ts b/src/lib/measurements/__tests__/series-canonical.test.ts index 310f08f2d..6f890608f 100644 --- a/src/lib/measurements/__tests__/series-canonical.test.ts +++ b/src/lib/measurements/__tests__/series-canonical.test.ts @@ -10,9 +10,8 @@ const SINCE = new Date("2026-03-01T06:30:00.000Z"); describe("seriesRowsFrom", () => { it("collapses a type that has a ladder to the ladder's source per day", () => { const sql = seriesRowsFrom(null, "PULSE", SINCE); - expect(sql).toContain("DISTINCT ON"); - // Scoped to the one metric, in both halves of the helper. - expect(sql).toContain(`"type" = 'PULSE'::"measurement_type"`); + expect(sql).toContain("canonical_source"); + // Scoped to the one metric in the helper's one scan. expect(sql).toContain(`mm."type" = 'PULSE'::"measurement_type"`); expect(sql.trimEnd().endsWith(") m")).toBe(true); }); @@ -21,7 +20,7 @@ describe("seriesRowsFrom", () => { const sql = seriesRowsFrom(null, "WEIGHT", SINCE); expect( sql.match(/>= '2026-03-01 06:30:00.000'::timestamp - INTERVAL '1 day'/g), - ).toHaveLength(2); + ).toHaveLength(1); // Counted from the caller's `since`, never from the database clock. expect(sql).not.toMatch(/NOW\(\)/i); }); @@ -46,7 +45,7 @@ describe("seriesRowsFrom", () => { "keeps every source for %s, which has no ladder", (type) => { const sql = seriesRowsFrom(null, type, SINCE); - expect(sql).not.toContain("DISTINCT ON"); + expect(sql).not.toContain("canonical_source"); expect(sql).toContain(`mm."type" = '${type}'::"measurement_type"`); expect(sql).toContain(`mm."user_id" = $1`); expect(sql).toContain(`mm."deleted_at" IS NULL`); @@ -69,9 +68,9 @@ describe("canonicalMeasurementsCte, one-metric scope", () => { expect(sql).not.toContain("measurement_type"); }); - it("scopes the pick and the join to the type when one is given", () => { + it("scopes the scan to the type when one is given", () => { const sql = canonicalMeasurementsCte(rank, NINETY_DAYS, "PULSE"); - expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(2); + expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(1); }); it("refuses anything that is not a closed-enum literal", () => { diff --git a/src/lib/provenance/__tests__/source-provenance.test.ts b/src/lib/provenance/__tests__/source-provenance.test.ts index 412d93687..d5dc72834 100644 --- a/src/lib/provenance/__tests__/source-provenance.test.ts +++ b/src/lib/provenance/__tests__/source-provenance.test.ts @@ -115,10 +115,12 @@ describe("the provenance query orders like the canonical pick", () => { `ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text`, ), ); - // The shared pick groups to (type, day, source) first and orders the - // groups by the same keys, the UTC day named `d`. + // The shared pick partitions by type and UTC day and orders each day's + // rows by the same keys. expect(squash(canonicalMeasurementsCte(rank))).toContain( - squash(`ORDER BY "type", d, (${rank}), "source"::text`), + squash( + `PARTITION BY mm."type", date_trunc('day', mm."measured_at") ORDER BY (${rank}), mm."source"::text`, + ), ); // Values are bound, never spliced. expect(userId).toBe("u1"); diff --git a/tests/integration/canonical-source-pick-parity.test.ts b/tests/integration/canonical-source-pick-parity.test.ts index da80698f1..b8cded3c2 100644 --- a/tests/integration/canonical-source-pick-parity.test.ts +++ b/tests/integration/canonical-source-pick-parity.test.ts @@ -1,8 +1,8 @@ /** * The canonical-source pick, against a real Postgres: the shipped statement - * (group the window to `(type, day, source)` first, then `DISTINCT ON`) - * against the statement it replaced (`DISTINCT ON` over every raw row), - * kept verbatim below as the reference. Only its window anchor changed: it + * (one window function over one scan names each day's source) against the + * `DISTINCT ON` pick joined back to the rows, kept verbatim below as the + * reference. Only its window anchor changed: it * counts from the same bound instant the shipped helper takes, where it read * `NOW()`, so both cut one window. * diff --git a/tests/integration/canonical-source-pick-unanalysed.test.ts b/tests/integration/canonical-source-pick-unanalysed.test.ts new file mode 100644 index 000000000..3661c4c2c --- /dev/null +++ b/tests/integration/canonical-source-pick-unanalysed.test.ts @@ -0,0 +1,233 @@ +/** + * The canonical-source pick (`canonicalMeasurementsCte`) must not depend on + * planner statistics. A new install, or an account whose import is still + * under the autoanalyze threshold, reads against a `measurements` table the + * planner has no statistics for. The pick used to join its per-day choices + * back to the rows; on such a table the planner estimated one row on each + * side and ran the join as a nested loop, rescanning the whole pick for every + * reading. `GET /api/analytics` on 6,000 blood-pressure rows took seconds, + * and on a ten-year account the all-time read ran past the statement timeout. + * + * Latency on a testcontainer says little, and the statement count did not + * move at all, so this file holds the work instead: every reader shape the + * helper serves is run under `EXPLAIN ANALYZE` on a table with no statistics, + * and the rows its plan nodes produce or discard, times their loops, must + * stay within a small multiple of the rows seeded. A plan that revisits the + * table per row lands thousands of times over that budget, whatever the + * machine. + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import type { + MeasurementSource, + MeasurementType, + Prisma, +} from "@/generated/prisma/client"; +import { + boundInstantSql, + instantLiteralSql, +} from "@/lib/analytics/request-clock-sql"; +import { + buildSourceRankCase, + canonicalMeasurementsCte, + type CanonicalWindow, +} from "@/lib/analytics/source-rank-sql"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const prisma = getPrismaClient(); +const USER = "canonical-pick-unanalysed"; +const DAY = 86_400_000; + +/** + * Rows produced or discarded, summed over the plan, per row seeded. The + * window pick sits at 4 or under on every shape below; the nested loop this + * guards against measured 200 on the 90-day read and over 3,000 on the + * one-type read, and ran past a minute on the all-time read. + */ +const WORK_PER_ROW = 8; + +/** + * Long enough for any linear plan on this fixture many times over, short + * enough that a quadratic one fails here instead of at the test timeout. + */ +const STATEMENT_TIMEOUT = "10s"; + +/** + * Ten years of paired blood pressure from one source with a second source on + * every third day, and a year of pulse from two sources a day: overlapping + * sources, so the pick has days to decide. + */ +function seedRows(): Prisma.MeasurementCreateManyInput[] { + const rows: Prisma.MeasurementCreateManyInput[] = []; + const now = Date.now(); + const push = ( + type: MeasurementType, + source: MeasurementSource, + value: number, + at: number, + ) => + rows.push({ + userId: USER, + type, + source, + value, + unit: type === "PULSE" ? "bpm" : "mmHg", + measuredAt: new Date(at), + }); + for (let d = 0; d < 3000; d += 1) { + const at = now - (d + 0.25) * DAY; + push("BLOOD_PRESSURE_SYS", "WITHINGS", 120 + (d % 30), at); + push("BLOOD_PRESSURE_DIA", "WITHINGS", 75 + (d % 15), at); + if (d % 3 === 0) { + push("BLOOD_PRESSURE_SYS", "MANUAL", 130 + (d % 20), at + 600_000); + push("BLOOD_PRESSURE_DIA", "MANUAL", 80 + (d % 10), at + 600_000); + } + } + for (let d = 0; d < 365; d += 1) { + for (const h of [7, 13, 21]) { + const at = now - d * DAY - h * 3_600_000; + push("PULSE", "WITHINGS", 60 + ((d + h) % 20), at); + push("PULSE", "APPLE_HEALTH", 62 + ((d + h) % 18), at + 60_000); + } + } + return rows; +} + +interface PlanNode { + "Actual Rows"?: number; + "Actual Loops"?: number; + "Rows Removed by Filter"?: number; + "Rows Removed by Join Filter"?: number; + "Rows Removed by Index Recheck"?: number; + Plans?: PlanNode[]; +} + +/** + * Every row a node returned or threw away, over all its loops. EXPLAIN + * reports both per loop, so a node rescanned once per outer row counts every + * rescan. + */ +function rowVisits(node: PlanNode): number { + const perLoop = + (node["Actual Rows"] ?? 0) + + (node["Rows Removed by Filter"] ?? 0) + + (node["Rows Removed by Join Filter"] ?? 0) + + (node["Rows Removed by Index Recheck"] ?? 0); + const own = perLoop * (node["Actual Loops"] ?? 1); + return own + (node.Plans ?? []).reduce((sum, p) => sum + rowVisits(p), 0); +} + +const NOW = new Date(); + +const SHAPES: Array< + [ + label: string, + window: CanonicalWindow | undefined, + type: MeasurementType | undefined, + from: Date | undefined, + ] +> = [ + // The summaries slice's all-time read. + [ + "every type, 15 years", + { from: boundInstantSql(2), interval: "15 years" }, + undefined, + NOW, + ], + // The summaries slice's and the comprehensive aggregator's 90-day reads. + [ + "every type, 90 days", + { from: boundInstantSql(2), interval: "90 days" }, + undefined, + NOW, + ], + // The BP-in-target fast path: from its 365-day start, no interval. + [ + "every type, from an instant", + { from: boundInstantSql(2) }, + undefined, + new Date(NOW.getTime() - 365 * DAY), + ], + // A metric series: one type, from a literal instant one day wider. + [ + "one type, ten years", + { + from: instantLiteralSql(new Date(NOW.getTime() - 3650 * DAY)), + interval: "1 day", + }, + "BLOOD_PRESSURE_SYS", + undefined, + ], + ["every type, no window", undefined, undefined, undefined], +]; + +let seeded = 0; + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements SET (autovacuum_enabled = false)", + ); + // Statistics another file in this worker gathered survive its TRUNCATE. + await prisma.$executeRawUnsafe( + "DELETE FROM pg_statistic WHERE starelid = 'measurements'::regclass", + ); + await prisma.user.create({ data: { id: USER, username: USER } }); + const rows = seedRows(); + seeded = rows.length; + await prisma.measurement.createMany({ data: rows }); +}, 120_000); + +afterAll(async () => { + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements RESET (autovacuum_enabled)", + ); +}); + +describe("canonical-source pick on a table without statistics", () => { + it("runs against a table the planner has no statistics for", async () => { + expect(seeded).toBeGreaterThan(10_000); + const [stats] = await prisma.$queryRawUnsafe< + Array<{ columns: number; reltuples: number }> + >( + `SELECT (SELECT COUNT(*) FROM pg_statistic + WHERE starelid = 'measurements'::regclass)::int AS columns, + reltuples::float8 AS reltuples + FROM pg_class WHERE oid = 'measurements'::regclass`, + ); + expect(stats).toEqual({ columns: 0, reltuples: -1 }); + }); + + it.each(SHAPES)( + "reads the window a bounded number of times: %s", + async (_label, window, onlyType, from) => { + const rank = buildSourceRankCase(null, '"type"', '"source"'); + const sql = `WITH cm AS (${canonicalMeasurementsCte(rank, window, onlyType)} + ) + SELECT COUNT(*)::int AS n, SUM(cm."value")::float8 AS total FROM cm`; + const params: unknown[] = from ? [USER, from] : [USER]; + const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe( + `SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`, + ); + return tx.$queryRawUnsafe< + Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }> + >(`EXPLAIN (ANALYZE, FORMAT JSON) ${sql}`, ...params); + }); + const visits = rowVisits(plan[0].Plan); + expect( + visits, + `${visits} row visits for ${seeded} rows seeded:\n` + + JSON.stringify(plan[0].Plan, null, 1), + ).toBeLessThanOrEqual(WORK_PER_ROW * seeded); + // The pick kept rows and dropped the second source's. + const [{ n }] = await prisma.$queryRawUnsafe>( + sql, + ...params, + ); + expect(n).toBeGreaterThan(0); + expect(n).toBeLessThan(seeded); + }, + ); +}); From 6d338760d871360dbe85d85d440b896f02b4bdd3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 25 Oct 2026 05:57:31 +0100 Subject: [PATCH 2/3] perf(series): pick the blood-pressure source with one window over one scan The blood-pressure series chose each UTC day's source with DISTINCT ON over the day's floor-passing systolic rows and joined the choices back to the readings. That is the shape the analytics pick had, with the same weakness: on a measurements table without planner statistics the planner estimated one row on each side and ran the join as a nested loop that rescanned the pick for every reading. The ten-year series over 6,000 freshly written rows from one source took 0.57 s for the statement alone, 0.86 s with a second source on every third day; the plan visited about 4,900 rows per row stored, so the cost grows with the square of the history. The pick is now one window over one scan: the day's rows ordered systolic first, then by ladder rank, then by source name, and the first row's source taken only when it is a systolic one. The scan starts at the window's first UTC day, so the pick still sees the whole edge day, and a day without a floor-passing systolic still shows nothing. Both fixtures above read in about 10 ms, with or without statistics. A parity suite runs the old statement as the reference against the new one across overlapping and off-ladder sources, half readings, sub-floor rows, tombstones, both DST changes, window starts on and either side of UTC midnight, and four session zones. The unanalysed-table guard now covers the series' ten-year, quarter and week reads. --- .../series/read-blood-pressure.ts | 85 ++++-- .../integration/bp-series-pick-parity.test.ts | 282 ++++++++++++++++++ .../canonical-source-pick-unanalysed.test.ts | 72 ++++- 3 files changed, 397 insertions(+), 42 deletions(-) create mode 100644 tests/integration/bp-series-pick-parity.test.ts diff --git a/src/lib/measurements/series/read-blood-pressure.ts b/src/lib/measurements/series/read-blood-pressure.ts index 89c1f5917..d86e2672b 100644 --- a/src/lib/measurements/series/read-blood-pressure.ts +++ b/src/lib/measurements/series/read-blood-pressure.ts @@ -32,7 +32,8 @@ export async function readBloodPressureSeries( // the whole edge day, not only the slice after `since`. // // `rank` is the closed-enum whitelist splice from `buildSourceRankCase`; - // every request value is a positional parameter. + // every request value is a positional parameter. The statement and why it + // is a window rather than a join are under `bloodPressureSeriesSql`. const rank = buildSourceRankCase( await loadUserSourcePriority(userId), '"type"', @@ -47,33 +48,7 @@ export async function readBloodPressureSeries( measured_at: Date; }> >( - ` - SELECT m."id", m."type"::text AS type, m."source"::text AS source, - m."value", m."measured_at" - FROM measurements m - JOIN ( - SELECT DISTINCT ON (date_trunc('day', "measured_at")) - date_trunc('day', "measured_at") AS d, - "source" AS canon - FROM measurements - WHERE "user_id" = $1 - AND "type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" - AND "deleted_at" IS NULL - AND "value" >= $3 - AND "measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC') - ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text - ) c - ON c.d = date_trunc('day', m."measured_at") - AND c.canon = m."source" - WHERE m."user_id" = $1 - AND m."deleted_at" IS NULL - AND m."measured_at" >= $2 - AND ( - (m."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND m."value" >= $3) - OR (m."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND m."value" >= $4) - ) - ORDER BY m."measured_at" ASC, m."id" ASC - `, + bloodPressureSeriesSql(rank), userId, since, VALUE_RANGES.BLOOD_PRESSURE_SYS.min, @@ -125,3 +100,57 @@ export async function readBloodPressureSeries( }); return { points }; } + +/** + * The series statement: every floor-passing systolic and diastolic row from + * `$2` on, of the source the ladder picks for its UTC day. `$1` is the user, + * `$2` the window start, `$3` and `$4` the systolic and diastolic floors; + * `rank` is a `buildSourceRankCase` CASE over unqualified `"type"` and + * `"source"`. + * + * One scan reads the window from the start of its first UTC day, so the pick + * sees the whole edge day. A window function names each day's source on + * every row: the day's rows are ordered systolic first, then by ladder rank, + * then by source name, and the first row's source is taken only when that + * row is a systolic one, so a day with no floor-passing systolic picks no + * source and shows nothing, as before. Rows of any other source, and rows + * of the edge day before `$2`, are dropped after the pick. + * + * Why no join: the pick used to be a `DISTINCT ON` over the day's systolic + * rows joined back to the readings. On a table without planner statistics + * (a new install, an import still under the autoanalyze threshold) the + * planner estimated one row on each side and ran that join as a nested loop + * that rescanned the pick for every reading: ten years of readings cost + * about 5,000 row visits per row. The window has no such choice to get + * wrong; `tests/integration/canonical-source-pick-unanalysed.test.ts` holds + * that, and `tests/integration/bp-series-pick-parity.test.ts` holds the old + * statement as the reference for the rows it returns. + */ +export function bloodPressureSeriesSql(rank: string): string { + return ` + SELECT w."id", w."type"::text AS type, w."source"::text AS source, + w."value", w."measured_at" + FROM ( + SELECT mm."id", mm."type", mm."source", mm."value", mm."measured_at", + FIRST_VALUE( + CASE WHEN mm."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" + THEN mm."source" END + ) OVER ( + PARTITION BY date_trunc('day', mm."measured_at") + ORDER BY mm."type" <> 'BLOOD_PRESSURE_SYS'::"measurement_type", + (${rank}), mm."source"::text + ) AS canonical_source + FROM measurements mm + WHERE mm."user_id" = $1 + AND mm."deleted_at" IS NULL + AND mm."measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC') + AND ( + (mm."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND mm."value" >= $3) + OR (mm."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND mm."value" >= $4) + ) + ) w + WHERE w."source" = w.canonical_source + AND w."measured_at" >= $2 + ORDER BY w."measured_at" ASC, w."id" ASC + `; +} diff --git a/tests/integration/bp-series-pick-parity.test.ts b/tests/integration/bp-series-pick-parity.test.ts new file mode 100644 index 000000000..ac583ecb0 --- /dev/null +++ b/tests/integration/bp-series-pick-parity.test.ts @@ -0,0 +1,282 @@ +/** + * The blood-pressure series pick, against a real Postgres: the shipped + * statement (one window function over one scan names each UTC day's source) + * against the `DISTINCT ON` pick over the day's systolic rows joined back to + * the readings, kept verbatim below as the reference. + * + * Both must return exactly the same rows in the same order, under the + * default ladder and under user ladders that reorder it, for window starts + * mid-day, on and either side of UTC midnight and on a local midnight, under + * a UTC session and under sessions in other zones (the day cut reads the + * session zone in both statements), over a fixture that carries what could + * tell them apart: overlapping sources on one day, sources off the ladder + * (the name tiebreak decides), a source whose only systolic that day is + * below the floor, a diastolic below its floor, a diastolic with no + * systolic, a systolic with no diastolic, tombstones, readings a minute + * either side of UTC midnight, both DST changes in the user's zone, empty + * days, and several readings per source per day. + */ +import { beforeAll, describe, expect, it } from "vitest"; + +import type { MeasurementSource } from "@/generated/prisma/client"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { bloodPressureSeriesSql } from "@/lib/measurements/series/read-blood-pressure"; +import { VALUE_RANGES } from "@/lib/validations/measurement"; + +import { makeUser } from "../kit/factories"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER_ID = "user-bp-series-parity"; +const DAY_MS = 86_400_000; +const SYS_MIN = VALUE_RANGES.BLOOD_PRESSURE_SYS.min; +const DIA_MIN = VALUE_RANGES.BLOOD_PRESSURE_DIA.min; + +/** The statement as it stood before the window: a pick joined back. */ +function legacyBloodPressureSeriesSql(rank: string): string { + return ` + SELECT m."id", m."type"::text AS type, m."source"::text AS source, + m."value", m."measured_at" + FROM measurements m + JOIN ( + SELECT DISTINCT ON (date_trunc('day', "measured_at")) + date_trunc('day', "measured_at") AS d, + "source" AS canon + FROM measurements + WHERE "user_id" = $1 + AND "type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" + AND "deleted_at" IS NULL + AND "value" >= $3 + AND "measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC') + ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text + ) c + ON c.d = date_trunc('day', m."measured_at") + AND c.canon = m."source" + WHERE m."user_id" = $1 + AND m."deleted_at" IS NULL + AND m."measured_at" >= $2 + AND ( + (m."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND m."value" >= $3) + OR (m."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND m."value" >= $4) + ) + ORDER BY m."measured_at" ASC, m."id" ASC + `; +} + +/** Small seeded generator, so the fixture is the same on every run. */ +function rng(seed: number) { + let a = seed >>> 0; + return () => { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +const SOURCES: MeasurementSource[] = [ + "WITHINGS", + "APPLE_HEALTH", + "MANUAL", + // Off the blood-pressure ladder: rank alike, the name tiebreak decides. + "OURA", + "WHOOP", +]; + +const NOW = Date.now(); +const TODAY_UTC = Math.floor(NOW / DAY_MS) * DAY_MS; + +type Row = { + userId: string; + type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA"; + source: MeasurementSource; + value: number; + unit: string; + measuredAt: Date; + deletedAt: Date | null; +}; + +async function seed(): Promise { + const prisma = getPrismaClient(); + await makeUser({ id: USER_ID, timezone: "Europe/Berlin" }); + const random = rng(7); + const rows: Row[] = []; + const push = ( + type: Row["type"], + source: MeasurementSource, + value: number, + at: number, + ) => + rows.push({ + userId: USER_ID, + type, + source, + value, + unit: "mmHg", + measuredAt: new Date(at), + // A tombstone, sometimes the only row its source had that day. + deletedAt: random() < 0.06 ? new Date() : null, + }); + // 400 days: over both DST changes in the user's zone. + for (let day = 0; day < 400; day++) { + if (day % 13 === 6) continue; // an empty day + const dayStart = TODAY_UTC - day * DAY_MS; + const sources = [...SOURCES] + .sort(() => random() - 0.5) + .slice(0, 1 + Math.floor(random() * 3)); + for (const source of sources) { + const readings = 1 + Math.floor(random() * 3); + for (let r = 0; r < readings; r++) { + // Mostly through the day, sometimes a minute either side of UTC + // midnight. + const at = + dayStart + + (r === 0 && random() < 0.25 + ? random() < 0.5 + ? 60_000 + : DAY_MS - 60_000 + : Math.floor(random() * DAY_MS)); + const shape = random(); + if (shape < 0.08) { + // A systolic below the floor: it must not claim the day. + push( + "BLOOD_PRESSURE_SYS", + source, + Math.floor(random() * SYS_MIN), + at, + ); + push("BLOOD_PRESSURE_DIA", source, 80, at); + } else if (shape < 0.14) { + // A diastolic below its floor. + push("BLOOD_PRESSURE_SYS", source, 125, at); + push( + "BLOOD_PRESSURE_DIA", + source, + Math.floor(random() * DIA_MIN), + at, + ); + } else if (shape < 0.2) { + // A diastolic alone: it never picks a source. + push("BLOOD_PRESSURE_DIA", source, 78, at); + } else if (shape < 0.25) { + // A systolic alone. + push("BLOOD_PRESSURE_SYS", source, 131, at); + } else { + const sys = 100 + Math.round(random() * 600) / 10; + push("BLOOD_PRESSURE_SYS", source, sys, at); + push( + "BLOOD_PRESSURE_DIA", + source, + 60 + Math.round(random() * 300) / 10, + at + Math.floor(random() * 4) * 60_000, + ); + } + } + } + } + // The natural identity is unique; drop accidental collisions. + const seen = new Set(); + const unique = rows.filter((row) => { + const key = `${row.type}|${row.source}|${row.measuredAt.getTime()}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }); + await prisma.measurement.createMany({ data: unique }); + return unique.length; +} + +const LADDERS: Array<[string, unknown]> = [ + ["default ladder", null], + [ + "a user ladder that reorders it", + { bloodPressure: ["MANUAL", "APPLE_HEALTH", "WITHINGS"] }, + ], +]; + +const STARTS: Array<[string, Date]> = [ + ["ten years", new Date(NOW - 3650 * DAY_MS)], + ["a year, mid-day", new Date(TODAY_UTC - 365 * DAY_MS + 13 * 3_600_000)], + ["thirty days, on UTC midnight", new Date(TODAY_UTC - 30 * DAY_MS)], + [ + "a week, a minute before UTC midnight", + new Date(TODAY_UTC - 7 * DAY_MS - 60_000), + ], + [ + "a week, a minute after UTC midnight", + new Date(TODAY_UTC - 7 * DAY_MS + 60_000), + ], + // 22:00 UTC is Berlin midnight in summer, 23:00 in winter. + [ + "ninety days, on a local midnight", + new Date(TODAY_UTC - 91 * DAY_MS + 22 * 3_600_000), + ], + ["two days, mid-day", new Date(TODAY_UTC - 2 * DAY_MS + 9 * 3_600_000)], +]; + +const ZONES = [ + "UTC", + "Europe/Berlin", + "Pacific/Auckland", + "America/Los_Angeles", +]; + +describe("blood-pressure series pick parity", () => { + let seeded = 0; + beforeAll(async () => { + await truncateAllTables(getPrismaClient()); + seeded = await seed(); + }); + + it("seeds a fixture worth comparing", () => { + expect(seeded).toBeGreaterThan(2000); + }); + + for (const [ladderLabel, ladder] of LADDERS) { + for (const zone of ZONES) { + for (const [startLabel, since] of STARTS) { + it(`returns the same rows: ${ladderLabel}, ${zone} session, ${startLabel}`, async () => { + const prisma = getPrismaClient(); + const rank = buildSourceRankCase(ladder, '"type"', '"source"'); + const select = (sql: string) => + prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe(`SET LOCAL TIME ZONE '${zone}'`); + return tx.$queryRawUnsafe< + Array<{ + id: string; + type: string; + source: string; + value: number; + measured_at: Date; + }> + >(sql, USER_ID, since, SYS_MIN, DIA_MIN); + }); + const now = await select(bloodPressureSeriesSql(rank)); + const before = await select(legacyBloodPressureSeriesSql(rank)); + expect(now.length).toBeGreaterThan(0); + expect(now).toEqual(before); + // Not trivial: both halves came back, from more than one source + // over the window, and the pick dropped rows that were live, + // floor-passing and inside the window. + expect(new Set(now.map((r) => r.type)).size).toBe(2); + const live = await prisma.$queryRawUnsafe>( + `SELECT COUNT(*)::int AS n FROM measurements + WHERE "user_id" = $1 AND "deleted_at" IS NULL + AND "measured_at" >= $2 + AND (("type" = 'BLOOD_PRESSURE_SYS' AND "value" >= $3) + OR ("type" = 'BLOOD_PRESSURE_DIA' AND "value" >= $4))`, + USER_ID, + since, + SYS_MIN, + DIA_MIN, + ); + expect(now.length).toBeLessThan(live[0].n); + if (since.getTime() < TODAY_UTC - 30 * DAY_MS) { + expect(new Set(now.map((r) => r.source)).size).toBeGreaterThan(1); + } + }); + } + } + } +}); diff --git a/tests/integration/canonical-source-pick-unanalysed.test.ts b/tests/integration/canonical-source-pick-unanalysed.test.ts index 3661c4c2c..40389cea3 100644 --- a/tests/integration/canonical-source-pick-unanalysed.test.ts +++ b/tests/integration/canonical-source-pick-unanalysed.test.ts @@ -32,6 +32,8 @@ import { canonicalMeasurementsCte, type CanonicalWindow, } from "@/lib/analytics/source-rank-sql"; +import { bloodPressureSeriesSql } from "@/lib/measurements/series/read-blood-pressure"; +import { VALUE_RANGES } from "@/lib/validations/measurement"; import { getPrismaClient, truncateAllTables } from "./setup"; @@ -164,6 +166,28 @@ const SHAPES: Array< let seeded = 0; +/** + * Run `sql` under `EXPLAIN ANALYZE` and hold its row visits to the budget. + * The statement timeout fails a quadratic plan here rather than at the test + * timeout. + */ +async function expectBoundedPlan(sql: string, params: unknown[]) { + const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe( + `SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`, + ); + return tx.$queryRawUnsafe< + Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }> + >(`EXPLAIN (ANALYZE, FORMAT JSON) ${sql}`, ...params); + }); + const visits = rowVisits(plan[0].Plan); + expect( + visits, + `${visits} row visits for ${seeded} rows seeded:\n` + + JSON.stringify(plan[0].Plan, null, 1), + ).toBeLessThanOrEqual(WORK_PER_ROW * seeded); +} + beforeAll(async () => { await truncateAllTables(prisma); await prisma.$executeRawUnsafe( @@ -207,20 +231,7 @@ describe("canonical-source pick on a table without statistics", () => { ) SELECT COUNT(*)::int AS n, SUM(cm."value")::float8 AS total FROM cm`; const params: unknown[] = from ? [USER, from] : [USER]; - const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => { - await tx.$executeRawUnsafe( - `SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`, - ); - return tx.$queryRawUnsafe< - Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }> - >(`EXPLAIN (ANALYZE, FORMAT JSON) ${sql}`, ...params); - }); - const visits = rowVisits(plan[0].Plan); - expect( - visits, - `${visits} row visits for ${seeded} rows seeded:\n` + - JSON.stringify(plan[0].Plan, null, 1), - ).toBeLessThanOrEqual(WORK_PER_ROW * seeded); + await expectBoundedPlan(sql, params); // The pick kept rows and dropped the second source's. const [{ n }] = await prisma.$queryRawUnsafe>( sql, @@ -231,3 +242,36 @@ describe("canonical-source pick on a table without statistics", () => { }, ); }); + +/** + * The blood-pressure series picks its own source per day (from the day's + * systolic rows above the floor, applied to both halves of the reading), so + * it does not go through `canonicalMeasurementsCte` and needs its own row + * here. Its windows are the series route's: ten years, a quarter, a week. + * The `DISTINCT ON` pick joined back to the readings measured about 4,900 + * row visits per row seeded on the ten-year read; the window stays at 4. + */ +describe("blood-pressure series pick on a table without statistics", () => { + it.each([ + ["3650 days", 3650], + ["90 days", 90], + ["7 days", 7], + ])("reads the window a bounded number of times: %s", async (_label, days) => { + const rank = buildSourceRankCase(null, '"type"', '"source"'); + const params: unknown[] = [ + USER, + new Date(NOW.getTime() - days * DAY), + VALUE_RANGES.BLOOD_PRESSURE_SYS.min, + VALUE_RANGES.BLOOD_PRESSURE_DIA.min, + ]; + const sql = bloodPressureSeriesSql(rank); + await expectBoundedPlan(sql, params); + // Both halves came back, and the second source's days were dropped. + const rows = await prisma.$queryRawUnsafe< + Array<{ type: string; source: string }> + >(sql, ...params); + expect(rows.some((r) => r.type === "BLOOD_PRESSURE_SYS")).toBe(true); + expect(rows.some((r) => r.type === "BLOOD_PRESSURE_DIA")).toBe(true); + expect(rows.every((r) => r.source === "WITHINGS")).toBe(true); + }); +}); From 4e79098cc09a26381bbdbae3e4458e9b7e30932c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 25 Oct 2026 06:27:44 +0100 Subject: [PATCH 3/3] perf(measurements): fold per-source days without joining a regrouping back Three readers grouped the same rows twice and joined the second grouping back to the first: the live buckets joined each hourly-mean day's hour means to the per-source days, the all-time extremes joined the day weights the same way, and the doctor report's dense buckets joined each bucket's latest reading and hour means to the bucket aggregates. On a measurements table without planner statistics (a new install, or an import under the autoanalyze threshold) the planner estimated one row on each side and ran every one of those joins as a nested loop that re-ran the inner grouping for each outer row. On ten years of readings over six types (51,000 rows, no statistics) the dense buckets and the every-type live buckets ran into the 60 s statement timeout and the all-time extremes took 11 s. With statistics all three answered in 0.1 to 0.2 s. Each is now one chain of folds over one scan. The live buckets fold the readings per local hour and the hours per local day, carrying the day of hours beside the counts. The all-time extremes take the day weights from the same windowed rows they count, the shape the pre-fold remainder already has. The dense buckets carry each bucket's latest reading and each hour's count and mean on the rows as window values, and one grouping folds them all. Same fixture without statistics: 0.2 s, 0.1 s and 0.1 s, the same as with them. A parity suite runs the joined statements as the reference against the readers across ladders, zones with whole, half-hour and negative offsets, grains, overlapping and off-ladder sources, glucose contexts and device types, tombstones and both DST changes; sums and means agree to floating-point summation order, everything else exactly. The unanalysed-table guard now runs the three readers under EXPLAIN ANALYZE: the joins measured 200 to 3,000 row visits per row stored, the folds about 10. The dense-bucket loader drops under the 200-line function cap, so its baseline entry goes. --- .../baselines/source-function-lines.json | 5 +- src/lib/doctor-report/dense-buckets.ts | 112 +-- src/lib/insights/feature-blocks.ts | 53 +- src/lib/measurements/daily-series-read.ts | 67 +- .../canonical-source-pick-unanalysed.test.ts | 124 ++- .../per-source-fold-parity.test.ts | 752 ++++++++++++++++++ 6 files changed, 965 insertions(+), 148 deletions(-) create mode 100644 tests/integration/per-source-fold-parity.test.ts diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index e29dc0c14..3f0155aef 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", - "count": 349, - "total": 113048, + "count": 348, + "total": 112838, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -275,7 +275,6 @@ "lib/doctor-report-pdf/measurements-charts-section.ts :: buildMeasurementsChartsSection": 384, "lib/doctor-report-pdf/medication-mood-wellness-section.ts :: buildMedicationMoodWellnessSection": 509, "lib/doctor-report/collect.ts :: collectDoctorReportData": 667, - "lib/doctor-report/dense-buckets.ts :: loadDenseMeasurementBuckets": 210, "lib/documents/fenced-chat.ts :: streamFencedReply": 309, "lib/documents/ingest.ts :: ingestDocument": 272, "lib/export/backup-json-scan.ts :: scanBackupJson": 225, diff --git a/src/lib/doctor-report/dense-buckets.ts b/src/lib/doctor-report/dense-buckets.ts index 3810fcda7..c01baed17 100644 --- a/src/lib/doctor-report/dense-buckets.ts +++ b/src/lib/doctor-report/dense-buckets.ts @@ -81,6 +81,13 @@ export async function loadDenseMeasurementBuckets(params: { const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); // Each hour's weight under the active pulse day-mean method (code-built). const hourWeight = Prisma.raw(hourWeightSql("n")); + // One grouping over one scan, no join. The latest reading and the hour + // means used to be groupings of their own, joined back to the buckets; on + // a measurements table without planner statistics (a new install, an + // import under the autoanalyze threshold) those joins ran as nested loops + // that re-ran a grouping for every bucket, and a long report window ran + // into the statement timeout. `tests/integration/per-source-fold-parity.test.ts` + // holds the joined statement as the reference. return prisma.$queryRaw` WITH localized AS ( SELECT @@ -106,6 +113,9 @@ export async function loadDenseMeasurementBuckets(params: { AND m."deleted_at" IS NULL AND m."type" IN (${typeList}) ), + -- Every row carries its bucket's latest reading and its local hour's + -- reading count and mean, with one row per hour marked, so the one + -- grouping below folds them all. scored AS ( SELECT localized.*, @@ -113,72 +123,18 @@ export async function loadDenseMeasurementBuckets(params: { WHEN "value" > 0 THEN 1.509 * (power(ln("value"), 1.084) - 5.381) ELSE NULL - END AS risk_transform + END AS risk_transform, + FIRST_VALUE("value") OVER ( + PARTITION BY local_day, "type", "source", "device_type", "glucose_context" + ORDER BY "measured_at" DESC, "id" DESC + ) AS latest_value, + COUNT(*) OVER hour AS n, + AVG("value") OVER hour AS hour_mean, + ROW_NUMBER() OVER hour = 1 AS hour_first FROM localized - ), - latest_rows AS ( - SELECT DISTINCT ON ( - local_day, - "type", - "source", - "device_type", - "glucose_context" + WINDOW hour AS ( + PARTITION BY local_day, local_hour, "type", "source", "device_type", "glucose_context" ) - local_day, - "type", - "source", - "device_type", - "glucose_context", - "measured_at" AS latest_at, - "value" AS latest_value - FROM localized - ORDER BY - local_day, - "type", - "source", - "device_type", - "glucose_context", - "measured_at" DESC, - "id" DESC - ), - -- The hourly-mean day (pulse): each local hour's mean and reading - -- count, then the weighted sum of the means and the sum of the hour - -- weights per bucket. Empty for every other type. - hour_means AS ( - SELECT - local_day, - "type", - "source", - "device_type", - "glucose_context", - AVG("value") AS hour_mean, - COUNT(*) AS n - FROM localized - WHERE ${hourlyMeanType} - GROUP BY - local_day, - local_hour, - "type", - "source", - "device_type", - "glucose_context" - ), - hour_agg AS ( - SELECT - local_day, - "type", - "source", - "device_type", - "glucose_context", - SUM(hour_mean * ${hourWeight})::double precision AS hour_weighted_sum, - SUM(${hourWeight})::double precision AS hour_weight - FROM hour_means - GROUP BY - local_day, - "type", - "source", - "device_type", - "glucose_context" ), grouped AS ( SELECT @@ -188,6 +144,14 @@ export async function loadDenseMeasurementBuckets(params: { "device_type", "glucose_context", COUNT(*)::int AS sample_count, + MAX("measured_at") AS latest_at, + MAX(latest_value)::double precision AS latest_value, + SUM(hour_mean * ${hourWeight}) + FILTER (WHERE hour_first AND ${hourlyMeanType})::double precision + AS hour_weighted_sum, + SUM(${hourWeight}) + FILTER (WHERE hour_first AND ${hourlyMeanType})::double precision + AS hour_weight, SUM("value")::double precision AS sum_value, MIN("value")::double precision AS min_value, MAX("value")::double precision AS max_value, @@ -242,8 +206,8 @@ export async function loadDenseMeasurementBuckets(params: { grouped.sum_value AS "sumValue", grouped.min_value AS "minValue", grouped.max_value AS "maxValue", - latest_rows.latest_at AS "latestAt", - latest_rows.latest_value::double precision AS "latestValue", + grouped.latest_at AS "latestAt", + grouped.latest_value AS "latestValue", grouped.clinical_count AS "clinicalCount", grouped.clinical_sum AS "clinicalSum", grouped.clinical_sum_squares AS "clinicalSumSquares", @@ -256,21 +220,9 @@ export async function loadDenseMeasurementBuckets(params: { grouped.clinical_tar2_count AS "clinicalTar2Count", grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", - hour_agg.hour_weighted_sum AS "hourWeightedSum", - hour_agg.hour_weight AS "hourWeight" + grouped.hour_weighted_sum AS "hourWeightedSum", + grouped.hour_weight AS "hourWeight" FROM grouped - INNER JOIN latest_rows - ON latest_rows.local_day = grouped.local_day - AND latest_rows."type" = grouped."type" - AND latest_rows."source" = grouped."source" - AND latest_rows."device_type" IS NOT DISTINCT FROM grouped."device_type" - AND latest_rows."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" - LEFT JOIN hour_agg - ON hour_agg.local_day = grouped.local_day - AND hour_agg."type" = grouped."type" - AND hour_agg."source" = grouped."source" - AND hour_agg."device_type" IS NOT DISTINCT FROM grouped."device_type" - AND hour_agg."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" ORDER BY "bucketStart" ASC, grouped."type" ASC `; diff --git a/src/lib/insights/feature-blocks.ts b/src/lib/insights/feature-blocks.ts index acd01bf57..63229ce86 100644 --- a/src/lib/insights/feature-blocks.ts +++ b/src/lib/insights/feature-blocks.ts @@ -12,11 +12,7 @@ import { prisma } from "@/lib/db"; import { Prisma } from "@/generated/prisma/client"; import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; -import { - dayWeightedRowsSql, - foldMeanSql, - isHourlyMeanTypeSql, -} from "@/lib/measurements/day-mean"; +import { dayWeightedRowsSql, foldMeanSql } from "@/lib/measurements/day-mean"; import { loadUserSourcePriority, utcBucketEnd, @@ -207,9 +203,18 @@ export async function readAllTimeExtremes( ORDER BY r."type", r."bucket_start", (${rank("r")}), r."source"::text ) tier`; // A day of pulse is the mean of its hours' means and the full history the - // mean of its days (`day-mean.ts`). `weighted` carries, per source and day, - // the day-weighted sum and the day's weight (one) for those types only; - // every other type keeps the reading-weighted mean below, unchanged. + // mean of its days (`day-mean.ts`). `per_source` carries, per source and + // day, the day-weighted sum and the day's weight (one) beside the reading + // count and sum; the mean below reads the weighted pair for those types + // only, and every other type keeps the reading-weighted mean, unchanged. + // + // One fold of one scan, no join. The weighted pair used to be a second + // grouping of the same rows, joined back to the per-source days before the + // pick; on a measurements table without planner statistics (a new install, + // an import under the autoanalyze threshold) that join ran as a nested + // loop that re-ran the weighting for every source day, and ten years of + // readings took seconds. `tests/integration/per-source-fold-parity.test.ts` + // holds the joined statement as the reference. const mean = Prisma.raw( foldMeanSql({ typeColumn: 'c."type"', @@ -223,30 +228,18 @@ export async function readAllTimeExtremes( Array<{ type: string; mean: number; min: number; max: number }> >` WITH live AS (${Prisma.join(rawArms, " UNION ALL ")}), - hourly_src AS ( - SELECT * FROM live l - WHERE ${Prisma.raw(isHourlyMeanTypeSql('l."type"'))} - ), - weighted AS ( - SELECT - w."type", - w."source", - date_trunc('day', w."measured_at") AS day, - SUM(w."value" * w.day_weight)::double precision AS wsum, - SUM(w.day_weight)::double precision AS wdays - FROM ${dayWeightedRowsSql("hourly_src", null, { bySource: true })} w - GROUP BY w."type", w."source", 3 - ), per_source AS ( SELECT l."type", l."source", - date_trunc('day', l."measured_at") AS day, - COUNT(*)::int AS cnt, - SUM(l."value")::double precision AS total, - MIN(l."value")::double precision AS min_value, - MAX(l."value")::double precision AS max_value - FROM live l + date_trunc('day', l."measured_at") AS day, + COUNT(*)::int AS cnt, + SUM(l."value")::double precision AS total, + MIN(l."value")::double precision AS min_value, + MAX(l."value")::double precision AS max_value, + SUM(l."value" * l.day_weight)::double precision AS wsum, + SUM(l.day_weight)::double precision AS wdays + FROM ${dayWeightedRowsSql("live", null, { bySource: true })} l GROUP BY l."type", l."source", 3 ) SELECT @@ -257,10 +250,8 @@ export async function readAllTimeExtremes( FROM ( SELECT * FROM ( SELECT DISTINCT ON (p."type", p.day) - p."type", p.cnt, p.total, p.min_value, p.max_value, w.wsum, w.wdays + p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays FROM per_source p - LEFT JOIN weighted w - ON w."type" = p."type" AND w."source" = p."source" AND w.day = p.day ORDER BY p."type", p.day, (${rank("p")}), p."source"::text ) raw ${tierCanon} diff --git a/src/lib/measurements/daily-series-read.ts b/src/lib/measurements/daily-series-read.ts index b25cf5a7a..702af8782 100644 --- a/src/lib/measurements/daily-series-read.ts +++ b/src/lib/measurements/daily-series-read.ts @@ -217,13 +217,22 @@ export async function readLiveBuckets(opts: { // the user's zone and truncate there. // // The canonical source is chosen per local DAY, whatever the grain: each - // source is folded per day in one hashed pass (a few rows per day, however - // dense the stream), each day keeps its ladder-canonical source, and only - // then are the days folded into the week or month. Picking the source per + // source is folded per hour and then per day in hashed passes (a few rows + // per day, however dense the stream), each day keeps its ladder-canonical + // source, and only then are the days folded into the week or month. Picking the source per // week or month instead dropped every day the losing source carried, so a // month of manual readings plus one synced reading reported one day. // Folding first and picking second is the same result as collapsing the // raw rows first; it just never sorts more than one row per source per day. + // + // One chain of folds over one scan, no join. The hour means used to be a + // second scan joined back to the per-source days before the pick; on a + // measurements table without planner statistics (a new install, an import + // under the autoanalyze threshold) that join ran as a nested loop that + // re-ran the hour fold for every source day, and a ten-year read of every + // type ran into the statement timeout. + // `tests/integration/per-source-fold-parity.test.ts` holds the joined + // statement as the reference. const buckets = await prisma.$queryRaw< Array<{ type: string; @@ -234,13 +243,20 @@ export async function readLiveBuckets(opts: { max_value: number | null; }> >` - WITH per_source AS ( + -- Each source's readings fold per local hour, then the hours per local + -- day. A day of a type with an activity-dependent sampling rate is the + -- weighted mean of its local hours' means, not the mean of its readings: + -- a workout hour holds hundreds of times as many readings as a resting + -- one. NULL for every other type. + WITH per_hour AS ( SELECT m."type", m."source", date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, + date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, SUM(m."value")::double precision AS total, - COUNT(*)::int AS cnt, + COUNT(*) AS n, + AVG(m."value") AS hour_mean, MIN(m."value") AS min_value, MAX(m."value") AS max_value FROM measurements m @@ -249,40 +265,25 @@ export async function readLiveBuckets(opts: { AND m."measured_at" <= ${to} AND m."deleted_at" IS NULL ${typeFilter} - GROUP BY m."type", m."source", 3 + GROUP BY m."type", m."source", 3, 4 ), - -- A day of a type with an activity-dependent sampling rate is the weighted - -- mean of its local hours' means, not the mean of its readings: a workout - -- hour holds hundreds of times as many readings as a resting one. - -- Empty for every other type. - hour_mean AS ( - SELECT h."type", h."source", h.day, - ${dayOfHours}::double precision AS day_mean - FROM ( - SELECT - m."type", - m."source", - date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, - date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, - AVG(m."value") AS hour_mean, - COUNT(*) AS n - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${from} - AND m."measured_at" <= ${to} - AND m."deleted_at" IS NULL - AND m."type" IN (${hourlyMeanList}) - ${typeFilter} - GROUP BY m."type", m."source", 3, 4 - ) h + per_source AS ( + SELECT + h."type", + h."source", + h.day, + SUM(h.total)::double precision AS total, + SUM(h.n)::int AS cnt, + MIN(h.min_value) AS min_value, + MAX(h.max_value) AS max_value, + (CASE WHEN h."type" IN (${hourlyMeanList}) THEN ${dayOfHours} END)::double precision AS day_mean + FROM per_hour h GROUP BY h."type", h."source", h.day ), canon AS ( SELECT DISTINCT ON (p."type", p.day) - p."type", p.day, p.total, p.cnt, p.min_value, p.max_value, hm.day_mean + p."type", p.day, p.total, p.cnt, p.min_value, p.max_value, p.day_mean FROM per_source p - LEFT JOIN hour_mean hm - ON hm."type" = p."type" AND hm."source" = p."source" AND hm.day = p.day ORDER BY p."type", p.day, (${rankRaw}), p."source"::text ), folded AS ( diff --git a/tests/integration/canonical-source-pick-unanalysed.test.ts b/tests/integration/canonical-source-pick-unanalysed.test.ts index 40389cea3..f968ae8e5 100644 --- a/tests/integration/canonical-source-pick-unanalysed.test.ts +++ b/tests/integration/canonical-source-pick-unanalysed.test.ts @@ -15,8 +15,14 @@ * stay within a small multiple of the rows seeded. A plan that revisits the * table per row lands thousands of times over that budget, whatever the * machine. + * + * The same holds for the readers that fold per source and day without the + * helper (the blood-pressure series, the live buckets, the all-time + * extremes, the doctor report's dense buckets): each used to join a second + * grouping of the same rows back to its days, which the planner ran the same + * way on such a table. */ -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import type { MeasurementSource, @@ -32,6 +38,10 @@ import { canonicalMeasurementsCte, type CanonicalWindow, } from "@/lib/analytics/source-rank-sql"; +import { prisma as appPrisma } from "@/lib/db"; +import { loadDenseMeasurementBuckets } from "@/lib/doctor-report/dense-buckets"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; import { bloodPressureSeriesSql } from "@/lib/measurements/series/read-blood-pressure"; import { VALUE_RANGES } from "@/lib/validations/measurement"; @@ -49,6 +59,14 @@ const DAY = 86_400_000; */ const WORK_PER_ROW = 8; +/** + * The same for a reader that folds every source's rows per hour or per day + * before it picks: a few more passes over the rows (a sort and a grouping + * per level), so about 10 per row seeded. The joins these readers used to + * make measured 200 to 3,000. + */ +const FOLD_WORK_PER_ROW = 16; + /** * Long enough for any linear plan on this fixture many times over, short * enough that a quadratic one fails here instead of at the test timeout. @@ -166,6 +184,55 @@ const SHAPES: Array< let seeded = 0; +/** + * Run `read` with every `$queryRaw` it issues also run under `EXPLAIN + * ANALYZE` on a transaction with the statement timeout, and hold each plan to + * the budget. The reader still gets its real rows. + */ +async function expectBoundedReader( + read: () => Promise, + workPerRow: number, +) { + const original = appPrisma.$queryRaw.bind(appPrisma) as ( + ...args: unknown[] + ) => Promise; + const statements: Array<[TemplateStringsArray, unknown[]]> = []; + vi.spyOn(appPrisma, "$queryRaw").mockImplementation((( + strings: TemplateStringsArray, + ...values: unknown[] + ) => { + statements.push([strings, values]); + return original(strings, ...values); + }) as never); + try { + await read(); + } finally { + vi.restoreAllMocks(); + } + expect(statements.length).toBeGreaterThan(0); + for (const [strings, values] of statements) { + const head = "EXPLAIN (ANALYZE, FORMAT JSON) "; + const explain = Object.assign([head + strings[0], ...strings.slice(1)], { + raw: [head + strings.raw[0], ...strings.raw.slice(1)], + }); + const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe( + `SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`, + ); + return tx.$queryRaw }>>( + explain, + ...values, + ); + }); + const visits = rowVisits(plan[0].Plan); + expect( + visits, + `${visits} row visits for ${seeded} rows seeded:\n` + + JSON.stringify(plan[0].Plan, null, 1), + ).toBeLessThanOrEqual(workPerRow * seeded); + } +} + /** * Run `sql` under `EXPLAIN ANALYZE` and hold its row visits to the budget. * The statement timeout fails a quadratic plan here rather than at the test @@ -275,3 +342,58 @@ describe("blood-pressure series pick on a table without statistics", () => { expect(rows.every((r) => r.source === "WITHINGS")).toBe(true); }); }); + +/** + * The readers that fold each source's days themselves and then pick: the + * live buckets of every type over ten years, the all-time extremes, and the + * doctor report's dense buckets over ten years. Each used to join a second + * grouping of the same rows (hour means, day weights, latest readings) back + * to the per-source days; on this fixture those joins measured about 200 + * (dense buckets), 1,200 (extremes) and 3,000 (live buckets) row visits per + * row seeded. + */ +describe("per-source folds on a table without statistics", () => { + const from = new Date(NOW.getTime() - 3650 * DAY); + it.each<[string, () => Promise>]>([ + [ + "live buckets, every type, ten years", + () => + readLiveBuckets({ + userId: USER, + type: null, + from, + to: NOW, + cap: 100_000, + priorityJson: null, + grain: "daily", + timeZone: "Europe/Berlin", + }), + ], + [ + "all-time extremes", + () => + readAllTimeExtremes(USER, [ + "BLOOD_PRESSURE_SYS", + "BLOOD_PRESSURE_DIA", + "PULSE", + ]), + ], + [ + "dense buckets, ten years", + () => + loadDenseMeasurementBuckets({ + userId: USER, + start: from, + end: NOW, + reportTz: "Europe/Berlin", + denseTypes: ["PULSE"], + }), + ], + ])("reads the window a bounded number of times: %s", async (_label, read) => { + let out: unknown[] | Map = []; + await expectBoundedReader(async () => { + out = await read(); + }, FOLD_WORK_PER_ROW); + expect(out instanceof Map ? out.size : out.length).toBeGreaterThan(0); + }); +}); diff --git a/tests/integration/per-source-fold-parity.test.ts b/tests/integration/per-source-fold-parity.test.ts new file mode 100644 index 000000000..2a4bce63d --- /dev/null +++ b/tests/integration/per-source-fold-parity.test.ts @@ -0,0 +1,752 @@ +/** + * Three per-source folds, against a real Postgres: each reader as it ships + * (one fold over one scan) against the statement it replaced, kept verbatim + * below as the reference, in which a second grouping of the same rows was + * joined back to the per-source days or buckets. + * + * - `readLiveBuckets`: the local hours' means for an hourly-mean type, + * joined to the per-source days before the ladder pick; + * - `readAllTimeExtremes`: the day-weighted sums for an hourly-mean type, + * joined to the per-source days before the ladder pick; + * - `loadDenseMeasurementBuckets`: each bucket's latest reading and its + * hour means, joined to the bucket aggregates. + * + * Every row must match: keys, counts, timestamps and extremes exactly, sums + * and means to within floating-point summation order (each statement adds + * the same readings, but not always in the same order or grouping, and + * neither statement fixed that order). The fixture carries what could tell + * them apart: overlapping sources per day, a source off every ladder, a + * cumulative type, an hourly-mean type with several readings in one hour, + * glucose contexts and device types including null ones, tombstones, + * readings a minute either side of UTC midnight, both DST changes in the + * user's zone, and empty days. Shapes run under the default ladders and a + * user ladder, in zones with whole, half-hour and negative offsets. + * + * The fixture holds no rollups, so `readAllTimeExtremes` reads every type + * from the raw rows; its rollup arm is a separate `UNION ALL` the change did + * not touch. + */ +import { afterEach, beforeAll, describe, expect, it, vi } from "vitest"; + +import { + Prisma, + type GlucoseContext, + type MeasurementSource, + type MeasurementType, +} from "@/generated/prisma/client"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { prisma as appPrisma } from "@/lib/db"; +import { loadDenseMeasurementBuckets } from "@/lib/doctor-report/dense-buckets"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; +import { + dayWeightedRowsSql, + foldMeanSql, + hourWeightSql, + isHourlyMeanTypeSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; +import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; + +import { makeUser } from "../kit/factories"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const prisma = getPrismaClient(); +const USER_ID = "user-per-source-fold-parity"; +const DAY_MS = 86_400_000; +const NOW = Date.now(); +const TODAY_UTC = Math.floor(NOW / DAY_MS) * DAY_MS; + +type Row = Record; + +/** The rows each `$queryRaw` the reader issued returned, in call order. */ +async function capture(read: () => Promise): Promise { + const results: Row[][] = []; + const original = appPrisma.$queryRaw.bind(appPrisma) as ( + ...args: unknown[] + ) => Promise; + vi.spyOn(appPrisma, "$queryRaw").mockImplementation(((...args: unknown[]) => + original(...args).then((rows) => { + results.push(rows); + return rows; + })) as never); + await read(); + return results; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +/** Compare two row sets: numbers to a relative 1e-9, everything else exact. */ +function expectSameRows(now: Row[], before: Row[], keys: string[]) { + const key = (r: Row) => + keys + .map((k) => + r[k] instanceof Date ? (r[k] as Date).toISOString() : String(r[k]), + ) + .join("|"); + const sort = (rows: Row[]) => + [...rows].sort((a, b) => (key(a) < key(b) ? -1 : key(a) > key(b) ? 1 : 0)); + const a = sort(now); + const b = sort(before); + expect(a.map(key)).toEqual(b.map(key)); + for (let i = 0; i < a.length; i++) { + expect(Object.keys(a[i]).sort()).toEqual(Object.keys(b[i]).sort()); + for (const [col, value] of Object.entries(a[i])) { + const other = b[i][col]; + if (typeof value === "number" && typeof other === "number") { + expect( + Math.abs(value - other), + `${col} of ${key(a[i])}: ${value} vs ${other}`, + ).toBeLessThanOrEqual(1e-9 * Math.max(1, Math.abs(other))); + } else { + expect(value, `${col} of ${key(a[i])}`).toEqual(other); + } + } + } +} + +/** Small seeded generator, so the fixture is the same on every run. */ +function rng(seed: number) { + let a = seed >>> 0; + return () => { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +const SOURCES: MeasurementSource[] = [ + "APPLE_HEALTH", + "WITHINGS", + "MANUAL", + "OURA", + // Off every ladder: the name tiebreak decides. + "NIGHTSCOUT", +]; +const TYPES: MeasurementType[] = [ + "PULSE", + "BLOOD_GLUCOSE", + "HEART_RATE_VARIABILITY", + "WEIGHT", + "ACTIVITY_STEPS", + "BLOOD_PRESSURE_SYS", +]; +const CONTEXTS: Array = [null, "FASTING", "RANDOM"]; +const DEVICES: Array = [null, "watch", "phone"]; + +async function seed(): Promise { + await makeUser({ id: USER_ID, timezone: "Europe/Berlin" }); + const random = rng(11); + const rows: Prisma.MeasurementCreateManyInput[] = []; + // 400 days: over both DST changes in the user's zone. + for (let day = 0; day < 400; day++) { + if (day % 17 === 8) continue; // an empty day + const dayStart = TODAY_UTC - day * DAY_MS; + for (const type of TYPES) { + const sources = [...SOURCES] + .sort(() => random() - 0.5) + .slice(0, 1 + Math.floor(random() * 3)); + for (const source of sources) { + const readings = 1 + Math.floor(random() * 6); + // Several readings share an hour, so an hour mean differs from a + // reading mean. + const hour = Math.floor(random() * 24); + for (let r = 0; r < readings; r++) { + const offset = + r === 0 && random() < 0.2 + ? random() < 0.5 + ? 60_000 + : DAY_MS - 60_000 + : random() < 0.5 + ? hour * 3_600_000 + Math.floor(random() * 3_600_000) + : Math.floor(random() * DAY_MS); + rows.push({ + userId: USER_ID, + type, + source, + value: Math.round(random() * 2_000_000) / 10_000 + 1, + unit: "x", + measuredAt: new Date(dayStart + offset), + deviceType: DEVICES[Math.floor(random() * DEVICES.length)], + glucoseContext: + type === "BLOOD_GLUCOSE" + ? CONTEXTS[Math.floor(random() * CONTEXTS.length)] + : null, + // A tombstone, sometimes the only row its source had that day. + deletedAt: random() < 0.06 ? new Date() : null, + }); + } + } + } + } + // The natural identity is unique; drop accidental collisions. + const seen = new Set(); + const unique = rows.filter((row) => { + const key = `${row.type}|${row.source}|${(row.measuredAt as Date).getTime()}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }); + await prisma.measurement.createMany({ data: unique }); + return unique.length; +} + +const LADDERS: Array<[string, Prisma.InputJsonValue | null]> = [ + ["default ladders", null], + [ + "a user ladder", + { + pulse: ["OURA", "WITHINGS", "APPLE_HEALTH"], + weight: ["MANUAL", "WITHINGS"], + steps: ["WITHINGS", "APPLE_HEALTH"], + bloodPressure: ["APPLE_HEALTH", "MANUAL", "WITHINGS"], + }, + ], +]; +const ZONES = ["Europe/Berlin", "America/Los_Angeles", "Asia/Kolkata"]; + +async function setLadder(ladder: Prisma.InputJsonValue | null) { + await prisma.user.update({ + where: { id: USER_ID }, + data: { sourcePriorityJson: ladder ?? Prisma.DbNull }, + }); +} + +// --------------------------------------------------------------------------- +// The references, verbatim but for their inputs, which the readers build the +// same way. + +function legacyLiveBuckets(opts: { + userId: string; + type: MeasurementType | null; + from: Date; + to: Date; + cap: number; + priorityJson: unknown; + unitName: string; + timeZone: string; +}) { + const { userId, type, from, to, cap, priorityJson, timeZone } = opts; + const unit = Prisma.raw(`'${opts.unitName}'`); + const cumulativeList = [...CUMULATIVE_HK_TYPES] + .map((t) => `'${t}'`) + .join(","); + const isCumulative = Prisma.raw(`c."type"::text IN (${cumulativeList})`); + const hourlyMeanList = Prisma.raw( + [...HOURLY_MEAN_DAY_TYPES].map((t) => `'${t}'::measurement_type`).join(","), + ); + const rankRaw = Prisma.raw( + buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), + ); + const dayOfHours = Prisma.raw( + weightedHourMeanSql("h.hour_mean", hourWeightSql("h.n")), + ); + const typeFilter = type + ? Prisma.sql`AND m."type" = ${type}::measurement_type` + : Prisma.empty; + return prisma.$queryRaw` + WITH per_source AS ( + SELECT + m."type", + m."source", + date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, + SUM(m."value")::double precision AS total, + COUNT(*)::int AS cnt, + MIN(m."value") AS min_value, + MAX(m."value") AS max_value + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${from} + AND m."measured_at" <= ${to} + AND m."deleted_at" IS NULL + ${typeFilter} + GROUP BY m."type", m."source", 3 + ), + -- A day of a type with an activity-dependent sampling rate is the weighted + -- mean of its local hours' means, not the mean of its readings: a workout + -- hour holds hundreds of times as many readings as a resting one. + -- Empty for every other type. + hour_mean AS ( + SELECT h."type", h."source", h.day, + ${dayOfHours}::double precision AS day_mean + FROM ( + SELECT + m."type", + m."source", + date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, + date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, + AVG(m."value") AS hour_mean, + COUNT(*) AS n + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${from} + AND m."measured_at" <= ${to} + AND m."deleted_at" IS NULL + AND m."type" IN (${hourlyMeanList}) + ${typeFilter} + GROUP BY m."type", m."source", 3, 4 + ) h + GROUP BY h."type", h."source", h.day + ), + canon AS ( + SELECT DISTINCT ON (p."type", p.day) + p."type", p.day, p.total, p.cnt, p.min_value, p.max_value, hm.day_mean + FROM per_source p + LEFT JOIN hour_mean hm + ON hm."type" = p."type" AND hm."source" = p."source" AND hm.day = p.day + ORDER BY p."type", p.day, (${rankRaw}), p."source"::text + ), + folded AS ( + SELECT + c."type", + date_trunc(${unit}, c.day) AS d, + (CASE + WHEN ${isCumulative} THEN SUM(c.total) + -- Hourly-mean types: every day once, so a window of days does not + -- hand a workout day its thousands of readings as weight. + WHEN COUNT(c.day_mean) = COUNT(*) THEN AVG(c.day_mean) + ELSE SUM(c.total) / SUM(c.cnt) + END)::double precision AS avg, + SUM(c.cnt)::int AS cnt, + MIN(c.min_value) AS min_value, + MAX(c.max_value) AS max_value + FROM canon c + GROUP BY c."type", 2 + ) + SELECT + f."type"::text AS type, + f.d AT TIME ZONE ${timeZone} AS bucket_start, + f.avg, + f.cnt, + f.min_value, + f.max_value + FROM folded f + ORDER BY f.d ASC + LIMIT ${cap} + `; +} + +function legacyAllTimeExtremes( + userId: string, + types: readonly MeasurementType[], + floor: Date | null, + priorityJson: unknown, +) { + const typeList = (list: readonly MeasurementType[]) => + Prisma.join(list.map((t) => Prisma.sql`${t}::measurement_type`)); + const since = + floor === null ? Prisma.empty : Prisma.sql`AND m."measured_at" >= ${floor}`; + // No rollups in the fixture: every type is read raw, in one arm. + const rawArms = [ + Prisma.sql` + SELECT m."type", m."source", m."measured_at", m."value" + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList(types)}) + ${since}`, + ]; + const tierCanon = Prisma.empty; + const rank = (alias: string) => + Prisma.raw( + buildSourceRankCase(priorityJson, `${alias}."type"`, `${alias}."source"`), + ); + const mean = Prisma.raw( + foldMeanSql({ + typeColumn: 'c."type"', + total: "c.total", + count: "c.cnt", + weightedSum: "c.wsum", + weightSum: "c.wdays", + }), + ); + return prisma.$queryRaw` + WITH live AS (${Prisma.join(rawArms, " UNION ALL ")}), + hourly_src AS ( + SELECT * FROM live l + WHERE ${Prisma.raw(isHourlyMeanTypeSql('l."type"'))} + ), + weighted AS ( + SELECT + w."type", + w."source", + date_trunc('day', w."measured_at") AS day, + SUM(w."value" * w.day_weight)::double precision AS wsum, + SUM(w.day_weight)::double precision AS wdays + FROM ${dayWeightedRowsSql("hourly_src", null, { bySource: true })} w + GROUP BY w."type", w."source", 3 + ), + per_source AS ( + SELECT + l."type", + l."source", + date_trunc('day', l."measured_at") AS day, + COUNT(*)::int AS cnt, + SUM(l."value")::double precision AS total, + MIN(l."value")::double precision AS min_value, + MAX(l."value")::double precision AS max_value + FROM live l + GROUP BY l."type", l."source", 3 + ) + SELECT + c."type"::text AS type, + ${mean}::double precision AS mean, + MIN(c.min_value)::double precision AS min, + MAX(c.max_value)::double precision AS max + FROM ( + SELECT * FROM ( + SELECT DISTINCT ON (p."type", p.day) + p."type", p.cnt, p.total, p.min_value, p.max_value, w.wsum, w.wdays + FROM per_source p + LEFT JOIN weighted w + ON w."type" = p."type" AND w."source" = p."source" AND w.day = p.day + ORDER BY p."type", p.day, (${rank("p")}), p."source"::text + ) raw + ${tierCanon} + ) c + GROUP BY c."type" + `; +} + +function legacyDenseBuckets(params: { + userId: string; + start: Date; + end: Date; + reportTz: string; + denseTypes: readonly MeasurementType[]; +}) { + const { userId, start, end, reportTz, denseTypes } = params; + const typeList = Prisma.join( + denseTypes.map((type) => Prisma.sql`${type}::"measurement_type"`), + ); + const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); + const hourWeight = Prisma.raw(hourWeightSql("n")); + return prisma.$queryRaw` + WITH localized AS ( + SELECT + m."id", + m."type", + m."source", + m."device_type", + m."glucose_context", + m."value", + m."measured_at", + date_trunc( + 'day', + (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${reportTz} + ) AS local_day, + date_trunc( + 'hour', + (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${reportTz} + ) AS local_hour + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${start} + AND m."measured_at" <= ${end} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList}) + ), + scored AS ( + SELECT + localized.*, + CASE + WHEN "value" > 0 + THEN 1.509 * (power(ln("value"), 1.084) - 5.381) + ELSE NULL + END AS risk_transform + FROM localized + ), + latest_rows AS ( + SELECT DISTINCT ON ( + local_day, + "type", + "source", + "device_type", + "glucose_context" + ) + local_day, + "type", + "source", + "device_type", + "glucose_context", + "measured_at" AS latest_at, + "value" AS latest_value + FROM localized + ORDER BY + local_day, + "type", + "source", + "device_type", + "glucose_context", + "measured_at" DESC, + "id" DESC + ), + -- The hourly-mean day (pulse): each local hour's mean and reading + -- count, then the weighted sum of the means and the sum of the hour + -- weights per bucket. Empty for every other type. + hour_means AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + AVG("value") AS hour_mean, + COUNT(*) AS n + FROM localized + WHERE ${hourlyMeanType} + GROUP BY + local_day, + local_hour, + "type", + "source", + "device_type", + "glucose_context" + ), + hour_agg AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + SUM(hour_mean * ${hourWeight})::double precision AS hour_weighted_sum, + SUM(${hourWeight})::double precision AS hour_weight + FROM hour_means + GROUP BY + local_day, + "type", + "source", + "device_type", + "glucose_context" + ), + grouped AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + COUNT(*)::int AS sample_count, + SUM("value")::double precision AS sum_value, + MIN("value")::double precision AS min_value, + MAX("value")::double precision AS max_value, + COUNT(*) FILTER (WHERE "value" > 0)::int AS clinical_count, + SUM("value") FILTER (WHERE "value" > 0)::double precision + AS clinical_sum, + SUM("value" * "value") FILTER (WHERE "value" > 0)::double precision + AS clinical_sum_squares, + MIN("measured_at") FILTER (WHERE "value" > 0) AS clinical_first_at, + MAX("measured_at") FILTER (WHERE "value" > 0) AS clinical_last_at, + COUNT(*) FILTER (WHERE "value" BETWEEN 70 AND 180)::int + AS clinical_tir_count, + COUNT(*) FILTER (WHERE "value" > 0 AND "value" < 70)::int + AS clinical_tbr1_count, + COUNT(*) FILTER (WHERE "value" > 0 AND "value" < 54)::int + AS clinical_tbr2_count, + COUNT(*) FILTER (WHERE "value" > 180)::int + AS clinical_tar1_count, + COUNT(*) FILTER (WHERE "value" > 250)::int + AS clinical_tar2_count, + SUM( + CASE + WHEN risk_transform < 0 + THEN 10 * risk_transform * risk_transform + ELSE 0 + END + ) FILTER (WHERE risk_transform IS NOT NULL)::double precision + AS clinical_low_risk_sum, + SUM( + CASE + WHEN risk_transform > 0 + THEN 10 * risk_transform * risk_transform + ELSE 0 + END + ) FILTER (WHERE risk_transform IS NOT NULL)::double precision + AS clinical_high_risk_sum + FROM scored + GROUP BY + local_day, + "type", + "source", + "device_type", + "glucose_context" + ) + SELECT + grouped.local_day AT TIME ZONE ${reportTz} AS "bucketStart", + grouped."type", + grouped."source", + grouped."device_type" AS "deviceType", + grouped."glucose_context" AS "glucoseContext", + grouped.sample_count AS "count", + grouped.sum_value AS "sumValue", + grouped.min_value AS "minValue", + grouped.max_value AS "maxValue", + latest_rows.latest_at AS "latestAt", + latest_rows.latest_value::double precision AS "latestValue", + grouped.clinical_count AS "clinicalCount", + grouped.clinical_sum AS "clinicalSum", + grouped.clinical_sum_squares AS "clinicalSumSquares", + grouped.clinical_first_at AS "clinicalFirstAt", + grouped.clinical_last_at AS "clinicalLastAt", + grouped.clinical_tir_count AS "clinicalTirCount", + grouped.clinical_tbr1_count AS "clinicalTbr1Count", + grouped.clinical_tbr2_count AS "clinicalTbr2Count", + grouped.clinical_tar1_count AS "clinicalTar1Count", + grouped.clinical_tar2_count AS "clinicalTar2Count", + grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", + grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", + hour_agg.hour_weighted_sum AS "hourWeightedSum", + hour_agg.hour_weight AS "hourWeight" + FROM grouped + INNER JOIN latest_rows + ON latest_rows.local_day = grouped.local_day + AND latest_rows."type" = grouped."type" + AND latest_rows."source" = grouped."source" + AND latest_rows."device_type" IS NOT DISTINCT FROM grouped."device_type" + AND latest_rows."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" + LEFT JOIN hour_agg + ON hour_agg.local_day = grouped.local_day + AND hour_agg."type" = grouped."type" + AND hour_agg."source" = grouped."source" + AND hour_agg."device_type" IS NOT DISTINCT FROM grouped."device_type" + AND hour_agg."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" + ORDER BY "bucketStart" ASC, grouped."type" ASC + +`; +} + +// --------------------------------------------------------------------------- + +describe("per-source fold parity", () => { + let seeded = 0; + beforeAll(async () => { + await truncateAllTables(prisma); + seeded = await seed(); + }, 120_000); + + it("seeds a fixture worth comparing", () => { + expect(seeded).toBeGreaterThan(5000); + }); + + describe("readLiveBuckets", () => { + const SHAPES: Array< + [string, MeasurementType | null, "daily" | "weekly" | "monthly", number] + > = [ + ["every type, daily, 400 days", null, "daily", 400], + ["every type, weekly, 400 days", null, "weekly", 400], + ["every type, monthly, 400 days", null, "monthly", 400], + ["pulse, daily, 30 days", "PULSE", "daily", 30], + ["steps, weekly, 120 days", "ACTIVITY_STEPS", "weekly", 120], + ]; + const UNIT = { daily: "day", weekly: "week", monthly: "month" } as const; + for (const [ladderLabel, ladder] of LADDERS) { + for (const zone of ZONES) { + for (const [label, type, grain, days] of SHAPES) { + it(`returns the same buckets: ${ladderLabel}, ${zone}, ${label}`, async () => { + await setLadder(ladder); + // A window edge in the middle of a day. + const from = new Date(NOW - days * DAY_MS - 5 * 3_600_000); + const to = new Date(NOW); + const opts = { + userId: USER_ID, + type, + from, + to, + cap: 100_000, + priorityJson: ladder, + timeZone: zone, + }; + const [now] = await capture(() => + readLiveBuckets({ ...opts, grain }), + ); + const before = await legacyLiveBuckets({ + ...opts, + unitName: UNIT[grain], + }); + expect(now.length).toBeGreaterThan(0); + expectSameRows(now, before, ["type", "bucket_start", "cnt"]); + }); + } + } + } + }); + + describe("readAllTimeExtremes", () => { + const TYPES_READ: MeasurementType[] = [ + "WEIGHT", + "BLOOD_PRESSURE_SYS", + "PULSE", + "ACTIVITY_STEPS", + ]; + for (const [ladderLabel, ladder] of LADDERS) { + for (const [floorLabel, floor] of [ + ["all time", null], + ["from a floor mid-day", new Date(NOW - 90 * DAY_MS - 7 * 3_600_000)], + ] as const) { + it(`returns the same figures: ${ladderLabel}, ${floorLabel}`, async () => { + await setLadder(ladder); + const results = await capture(() => + readAllTimeExtremes(USER_ID, TYPES_READ, floor), + ); + // The tier check first, then the fold; no type came from the tier. + expect(results).toHaveLength(2); + expect(results[0]).toEqual([]); + const before = await legacyAllTimeExtremes( + USER_ID, + TYPES_READ, + floor, + ladder, + ); + expect(results[1]).toHaveLength(TYPES_READ.length); + expectSameRows(results[1], before, ["type"]); + }); + } + } + }); + + describe("loadDenseMeasurementBuckets", () => { + const DENSE: MeasurementType[] = [ + "PULSE", + "BLOOD_GLUCOSE", + "HEART_RATE_VARIABILITY", + ]; + for (const zone of ZONES) { + for (const [label, days] of [ + ["400 days", 400], + ["120 days", 120], + ] as const) { + it(`returns the same buckets: ${zone}, ${label}`, async () => { + const params = { + userId: USER_ID, + start: new Date(NOW - days * DAY_MS - 5 * 3_600_000), + end: new Date(NOW), + reportTz: zone, + denseTypes: DENSE, + }; + const now = (await loadDenseMeasurementBuckets( + params, + )) as unknown as Row[]; + const before = await legacyDenseBuckets(params); + expect(now.length).toBeGreaterThan(0); + // Some buckets carry an hour fold and some do not. + expect(now.some((r) => r.hourWeight !== null)).toBe(true); + expect(now.some((r) => r.hourWeight === null)).toBe(true); + expectSameRows(now, before, [ + "bucketStart", + "type", + "source", + "deviceType", + "glucoseContext", + ]); + }); + } + } + }); +});