tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-07 04:12:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(measurements): fold per-source days without joining a regrouping back

Browse Source 
Three readers grouped the same rows twice and joined the second grouping
back to the first: the live buckets joined each hourly-mean day's hour
means to the per-source days, the all-time extremes joined the day
weights the same way, and the doctor report's dense buckets joined each
bucket's latest reading and hour means to the bucket aggregates. On a
measurements table without planner statistics (a new install, or an
import under the autoanalyze threshold) the planner estimated one row on
each side and ran every one of those joins as a nested loop that re-ran
the inner grouping for each outer row.

On ten years of readings over six types (51,000 rows, no statistics) the
dense buckets and the every-type live buckets ran into the 60 s statement
timeout and the all-time extremes took 11 s. With statistics all three
answered in 0.1 to 0.2 s.

Each is now one chain of folds over one scan. The live buckets fold the
readings per local hour and the hours per local day, carrying the day of
hours beside the counts. The all-time extremes take the day weights from
the same windowed rows they count, the shape the pre-fold remainder
already has. The dense buckets carry each bucket's latest reading and each
hour's count and mean on the rows as window values, and one grouping folds
them all. Same fixture without statistics: 0.2 s, 0.1 s and 0.1 s, the same
as with them.

A parity suite runs the joined statements as the reference against the
readers across ladders, zones with whole, half-hour and negative offsets,
grains, overlapping and off-ladder sources, glucose contexts and device
types, tombstones and both DST changes; sums and means agree to
floating-point summation order, everything else exactly. The
unanalysed-table guard now runs the three readers under EXPLAIN ANALYZE:
the joins measured 200 to 3,000 row visits per row stored, the folds about
10.

The dense-bucket loader drops under the 200-line function cap, so its
baseline entry goes.
This commit is contained in:
Marc-André Bombeck committed 2026-10-25 06:27:44 +01:00
1 parent 6d338760d8
commit 4e79098cc0
6 files changed
+965 -148
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/__tests__/baselines/source-function-lines.json
+2 -3
View File
@@ -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,
src/lib/doctor-report/dense-buckets.ts
+32 -80
View File
@@ -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<DenseMeasurementBucket[]>`
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
`;
src/lib/insights/feature-blocks.ts
+22 -31
View File
@@ -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}
src/lib/measurements/daily-series-read.ts
+34 -33
View File
@@ -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 (
tests/integration/canonical-source-pick-unanalysed.test.ts
+123 -1
View File
@@ -15,8 +15,14 @@
* stay within a small multiple of the rows seeded. A plan that revisits the
* table per row lands thousands of times over that budget, whatever the
* machine.
*
* The same holds for the readers that fold per source and day without the
* helper (the blood-pressure series, the live buckets, the all-time
* extremes, the doctor report's dense buckets): each used to join a second
* grouping of the same rows back to its days, which the planner ran the same
* way on such a table.
*/
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import type {
MeasurementSource,
@@ -32,6 +38,10 @@ import {
canonicalMeasurementsCte,
type CanonicalWindow,
} from "@/lib/analytics/source-rank-sql";
import { prisma as appPrisma } from "@/lib/db";
import { loadDenseMeasurementBuckets } from "@/lib/doctor-report/dense-buckets";
import { readAllTimeExtremes } from "@/lib/insights/feature-blocks";
import { readLiveBuckets } from "@/lib/measurements/daily-series-read";
import { bloodPressureSeriesSql } from "@/lib/measurements/series/read-blood-pressure";
import { VALUE_RANGES } from "@/lib/validations/measurement";
@@ -49,6 +59,14 @@ const DAY = 86_400_000;
*/
const WORK_PER_ROW = 8;
/**
* The same for a reader that folds every source's rows per hour or per day
* before it picks: a few more passes over the rows (a sort and a grouping
* per level), so about 10 per row seeded. The joins these readers used to
* make measured 200 to 3,000.
*/
const FOLD_WORK_PER_ROW = 16;
/**
* Long enough for any linear plan on this fixture many times over, short
* enough that a quadratic one fails here instead of at the test timeout.
@@ -166,6 +184,55 @@ const SHAPES: Array<
let seeded = 0;
/**
* Run `read` with every `$queryRaw` it issues also run under `EXPLAIN
* ANALYZE` on a transaction with the statement timeout, and hold each plan to
* the budget. The reader still gets its real rows.
*/
async function expectBoundedReader(
read: () => Promise<unknown>,
workPerRow: number,
) {
const original = appPrisma.$queryRaw.bind(appPrisma) as (
...args: unknown[]
) => Promise<unknown>;
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<Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }>>(
explain,
...values,
);
});
const visits = rowVisits(plan[0].Plan);
expect(
visits,
`${visits} row visits for ${seeded} rows seeded:\n` +
JSON.stringify(plan[0].Plan, null, 1),
).toBeLessThanOrEqual(workPerRow * seeded);
}
}
/**
* Run `sql` under `EXPLAIN ANALYZE` and hold its row visits to the budget.
* The statement timeout fails a quadratic plan here rather than at the test
@@ -275,3 +342,58 @@ describe("blood-pressure series pick on a table without statistics", () => {
expect(rows.every((r) => r.source === "WITHINGS")).toBe(true);
});
});
/**
* The readers that fold each source's days themselves and then pick: the
* live buckets of every type over ten years, the all-time extremes, and the
* doctor report's dense buckets over ten years. Each used to join a second
* grouping of the same rows (hour means, day weights, latest readings) back
* to the per-source days; on this fixture those joins measured about 200
* (dense buckets), 1,200 (extremes) and 3,000 (live buckets) row visits per
* row seeded.
*/
describe("per-source folds on a table without statistics", () => {
const from = new Date(NOW.getTime() - 3650 * DAY);
it.each<[string, () => Promise<unknown[] | Map<unknown, unknown>>]>([
[
"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<unknown, unknown> = [];
await expectBoundedReader(async () => {
out = await read();
}, FOLD_WORK_PER_ROW);
expect(out instanceof Map ? out.size : out.length).toBeGreaterThan(0);
});
});
tests/integration/per-source-fold-parity.test.ts
+752
View File
@@ -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<string, unknown>;
/** The rows each `$queryRaw` the reader issued returned, in call order. */
async function capture(read: () => Promise<unknown>): Promise<Row[][]> {
const results: Row[][] = [];
const original = appPrisma.$queryRaw.bind(appPrisma) as (
...args: unknown[]
) => Promise<Row[]>;
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<GlucoseContext | null> = [null, "FASTING", "RANDOM"];
const DEVICES: Array<string | null> = [null, "watch", "phone"];
async function seed(): Promise<number> {
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<string>();
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<Row[]>`
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<Row[]>`
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<Row[]>`
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",
]);
});
}
}
});
});