perf(analytics): pick the canonical source with one window over one scan

The canonical-source pick grouped the window to (type, day, source),
chose each day's source with DISTINCT ON, and joined the choices back to
the rows. Whether that join ran as a hash join or as a nested loop that
rescanned the whole pick for every row was the planner's call, made from
table statistics. On a measurements table without them (a new install,
or an import still under the autoanalyze threshold) it estimated one row
on each side and took the nested loop.

Binding the request's clock as a parameter instead of reading NOW()
flipped which side of that loop the pick landed on: the analytics read
over 6,000 freshly written blood-pressure rows went from 1.6 s to 8.4 s,
and the demo seed's dashboard headline to 20 s. On a ten-year account
without statistics the 90-day read took 4 s with either clock and the
all-time read ran past two minutes.

The pick is now one window function over one scan: FIRST_VALUE of the
source per (type, UTC day) in ladder order, keeping the rows of that
source. Same rows, same order of preference, so every day keeps its
source; the parity suite against the DISTINCT ON statement holds that.
The analytics read on the fixture above takes about 0.1 s, the headline
0.4 s. On the analysed ten-year account the 90-day and all-time reads
are as fast as the join was, and they no longer depend on statistics.

The rows carry one extra column, canonical_source; every reader selects
its columns by name.

A new integration test runs every reader shape under EXPLAIN ANALYZE on
a table with no statistics and bounds the rows the plan produces or
discards per row seeded. The statement count never moved, so a statement
budget could not see this; the join measured 200 to 8,600 row visits
per row seeded and the window stays at 4.

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'");
       }
     });

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", () => {

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`.

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<NarrowAggregateRow[]>(

src/lib/insights/comprehensive-aggregator.ts

@@ -403,10 +403,10 @@ async function buildFromRollups(
       timeSubquery(timings, "narrow", () =>
         prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
           `
-      -- 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<HeavyAggregateRow[]>(
         `
-      -- 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

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", () => {

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");

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.
  *

tests/integration/canonical-source-pick-unanalysed.test.ts

@@ -0,0 +1,233 @@
+/**
+ * The canonical-source pick (`canonicalMeasurementsCte`) must not depend on
+ * planner statistics. A new install, or an account whose import is still
+ * under the autoanalyze threshold, reads against a `measurements` table the
+ * planner has no statistics for. The pick used to join its per-day choices
+ * back to the rows; on such a table the planner estimated one row on each
+ * side and ran the join as a nested loop, rescanning the whole pick for every
+ * reading. `GET /api/analytics` on 6,000 blood-pressure rows took seconds,
+ * and on a ten-year account the all-time read ran past the statement timeout.
+ *
+ * Latency on a testcontainer says little, and the statement count did not
+ * move at all, so this file holds the work instead: every reader shape the
+ * helper serves is run under `EXPLAIN ANALYZE` on a table with no statistics,
+ * and the rows its plan nodes produce or discard, times their loops, must
+ * stay within a small multiple of the rows seeded. A plan that revisits the
+ * table per row lands thousands of times over that budget, whatever the
+ * machine.
+ */
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import type {
+  MeasurementSource,
+  MeasurementType,
+  Prisma,
+} from "@/generated/prisma/client";
+import {
+  boundInstantSql,
+  instantLiteralSql,
+} from "@/lib/analytics/request-clock-sql";
+import {
+  buildSourceRankCase,
+  canonicalMeasurementsCte,
+  type CanonicalWindow,
+} from "@/lib/analytics/source-rank-sql";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+const prisma = getPrismaClient();
+const USER = "canonical-pick-unanalysed";
+const DAY = 86_400_000;
+
+/**
+ * Rows produced or discarded, summed over the plan, per row seeded. The
+ * window pick sits at 4 or under on every shape below; the nested loop this
+ * guards against measured 200 on the 90-day read and over 3,000 on the
+ * one-type read, and ran past a minute on the all-time read.
+ */
+const WORK_PER_ROW = 8;
+
+/**
+ * Long enough for any linear plan on this fixture many times over, short
+ * enough that a quadratic one fails here instead of at the test timeout.
+ */
+const STATEMENT_TIMEOUT = "10s";
+
+/**
+ * Ten years of paired blood pressure from one source with a second source on
+ * every third day, and a year of pulse from two sources a day: overlapping
+ * sources, so the pick has days to decide.
+ */
+function seedRows(): Prisma.MeasurementCreateManyInput[] {
+  const rows: Prisma.MeasurementCreateManyInput[] = [];
+  const now = Date.now();
+  const push = (
+    type: MeasurementType,
+    source: MeasurementSource,
+    value: number,
+    at: number,
+  ) =>
+    rows.push({
+      userId: USER,
+      type,
+      source,
+      value,
+      unit: type === "PULSE" ? "bpm" : "mmHg",
+      measuredAt: new Date(at),
+    });
+  for (let d = 0; d < 3000; d += 1) {
+    const at = now - (d + 0.25) * DAY;
+    push("BLOOD_PRESSURE_SYS", "WITHINGS", 120 + (d % 30), at);
+    push("BLOOD_PRESSURE_DIA", "WITHINGS", 75 + (d % 15), at);
+    if (d % 3 === 0) {
+      push("BLOOD_PRESSURE_SYS", "MANUAL", 130 + (d % 20), at + 600_000);
+      push("BLOOD_PRESSURE_DIA", "MANUAL", 80 + (d % 10), at + 600_000);
+    }
+  }
+  for (let d = 0; d < 365; d += 1) {
+    for (const h of [7, 13, 21]) {
+      const at = now - d * DAY - h * 3_600_000;
+      push("PULSE", "WITHINGS", 60 + ((d + h) % 20), at);
+      push("PULSE", "APPLE_HEALTH", 62 + ((d + h) % 18), at + 60_000);
+    }
+  }
+  return rows;
+}
+
+interface PlanNode {
+  "Actual Rows"?: number;
+  "Actual Loops"?: number;
+  "Rows Removed by Filter"?: number;
+  "Rows Removed by Join Filter"?: number;
+  "Rows Removed by Index Recheck"?: number;
+  Plans?: PlanNode[];
+}
+
+/**
+ * Every row a node returned or threw away, over all its loops. EXPLAIN
+ * reports both per loop, so a node rescanned once per outer row counts every
+ * rescan.
+ */
+function rowVisits(node: PlanNode): number {
+  const perLoop =
+    (node["Actual Rows"] ?? 0) +
+    (node["Rows Removed by Filter"] ?? 0) +
+    (node["Rows Removed by Join Filter"] ?? 0) +
+    (node["Rows Removed by Index Recheck"] ?? 0);
+  const own = perLoop * (node["Actual Loops"] ?? 1);
+  return own + (node.Plans ?? []).reduce((sum, p) => sum + rowVisits(p), 0);
+}
+
+const NOW = new Date();
+
+const SHAPES: Array<
+  [
+    label: string,
+    window: CanonicalWindow | undefined,
+    type: MeasurementType | undefined,
+    from: Date | undefined,
+  ]
+> = [
+  // The summaries slice's all-time read.
+  [
+    "every type, 15 years",
+    { from: boundInstantSql(2), interval: "15 years" },
+    undefined,
+    NOW,
+  ],
+  // The summaries slice's and the comprehensive aggregator's 90-day reads.
+  [
+    "every type, 90 days",
+    { from: boundInstantSql(2), interval: "90 days" },
+    undefined,
+    NOW,
+  ],
+  // The BP-in-target fast path: from its 365-day start, no interval.
+  [
+    "every type, from an instant",
+    { from: boundInstantSql(2) },
+    undefined,
+    new Date(NOW.getTime() - 365 * DAY),
+  ],
+  // A metric series: one type, from a literal instant one day wider.
+  [
+    "one type, ten years",
+    {
+      from: instantLiteralSql(new Date(NOW.getTime() - 3650 * DAY)),
+      interval: "1 day",
+    },
+    "BLOOD_PRESSURE_SYS",
+    undefined,
+  ],
+  ["every type, no window", undefined, undefined, undefined],
+];
+
+let seeded = 0;
+
+beforeAll(async () => {
+  await truncateAllTables(prisma);
+  await prisma.$executeRawUnsafe(
+    "ALTER TABLE measurements SET (autovacuum_enabled = false)",
+  );
+  // Statistics another file in this worker gathered survive its TRUNCATE.
+  await prisma.$executeRawUnsafe(
+    "DELETE FROM pg_statistic WHERE starelid = 'measurements'::regclass",
+  );
+  await prisma.user.create({ data: { id: USER, username: USER } });
+  const rows = seedRows();
+  seeded = rows.length;
+  await prisma.measurement.createMany({ data: rows });
+}, 120_000);
+
+afterAll(async () => {
+  await prisma.$executeRawUnsafe(
+    "ALTER TABLE measurements RESET (autovacuum_enabled)",
+  );
+});
+
+describe("canonical-source pick on a table without statistics", () => {
+  it("runs against a table the planner has no statistics for", async () => {
+    expect(seeded).toBeGreaterThan(10_000);
+    const [stats] = await prisma.$queryRawUnsafe<
+      Array<{ columns: number; reltuples: number }>
+    >(
+      `SELECT (SELECT COUNT(*) FROM pg_statistic
+                WHERE starelid = 'measurements'::regclass)::int AS columns,
+              reltuples::float8 AS reltuples
+         FROM pg_class WHERE oid = 'measurements'::regclass`,
+    );
+    expect(stats).toEqual({ columns: 0, reltuples: -1 });
+  });
+
+  it.each(SHAPES)(
+    "reads the window a bounded number of times: %s",
+    async (_label, window, onlyType, from) => {
+      const rank = buildSourceRankCase(null, '"type"', '"source"');
+      const sql = `WITH cm AS (${canonicalMeasurementsCte(rank, window, onlyType)}
+        )
+        SELECT COUNT(*)::int AS n, SUM(cm."value")::float8 AS total FROM cm`;
+      const params: unknown[] = from ? [USER, from] : [USER];
+      const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => {
+        await tx.$executeRawUnsafe(
+          `SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`,
+        );
+        return tx.$queryRawUnsafe<
+          Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }>
+        >(`EXPLAIN (ANALYZE, FORMAT JSON) ${sql}`, ...params);
+      });
+      const visits = rowVisits(plan[0].Plan);
+      expect(
+        visits,
+        `${visits} row visits for ${seeded} rows seeded:\n` +
+          JSON.stringify(plan[0].Plan, null, 1),
+      ).toBeLessThanOrEqual(WORK_PER_ROW * seeded);
+      // The pick kept rows and dropped the second source's.
+      const [{ n }] = await prisma.$queryRawUnsafe<Array<{ n: number }>>(
+        sql,
+        ...params,
+      );
+      expect(n).toBeGreaterThan(0);
+      expect(n).toBeLessThan(seeded);
+    },
+  );
+});