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/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/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/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/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/__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/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/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/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/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-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..f968ae8e5 --- /dev/null +++ b/tests/integration/canonical-source-pick-unanalysed.test.ts @@ -0,0 +1,399 @@ +/** + * 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. + * + * 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, vi } 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 { 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"; + +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; + +/** + * 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. + */ +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; + +/** + * 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 + * 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( + "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]; + await expectBoundedPlan(sql, params); + // 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); + }, + ); +}); + +/** + * 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); + }); +}); + +/** + * 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", + ]); + }); + } + } + }); +});