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] 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); + }); +});