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.

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
+  `;
+}

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<number> {
+  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<string>();
+  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<Array<{ n: number }>>(
+            `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);
+          }
+        });
+      }
+    }
+  }
+});

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<Array<{ n: number }>>(
         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);
+  });
+});