Add static display-grain pre-aggregation (col_meta.in_grain)

Ship rows pre-aggregated to the grain the pivot displays instead of raw
forecast rows. This is Path B from pf_perspective_options.md: it keeps
Perspective's native WASM engine — so expand/collapse/depth/sort/filter
all still work — and fixes load time by cutting rows, not transport.

Measured on pf.fc_osm_stack_20 at pending_rep x customer x smon:
534,902 -> 6,154 rows (~87x), pf_gkey unique across all 6,154, and both
measures reconcile exactly to the raw totals.

The grain is static: flagged once per source in Setup and baked into the
stored pf.sql templates, so load and operations agree by construction.
Sources with no flagged column keep the previous raw-row behaviour, so
this is backward compatible.

- pf.col_meta gains in_grain; grainOf() in lib/sql_generator.js is the
  single definition of the grain and is reused by routes/log.js.
- New get_agg template + GET /api/versions/:id/agg, generated only when a
  grain is defined. Regenerating drops templates no longer produced, so
  clearing the grain falls back to /data.
- scale/recode/clone now aggregate their own new rows to grain before
  returning. Because pf_logid is part of pf_gkey those keys are always
  new, so table.update() appends and the view re-sums — the Excel
  pivot-cache pattern, no bucket recomputation.
- Undo reports pf_gkeys (RETURNING cannot take DISTINCT, so the delete
  feeds a CTE that reduces to distinct keys); the client removes those
  index values and the view re-sums.
- pf_gkey is concat_ws(chr(31), COALESCE(col::text, chr(30)), ...).
  The separator and NULL sentinel are load-bearing: plain concat_ws skips
  NULLs, so ('a',NULL) and (NULL,'a') would collide and silently merge two
  groups into one indexed row.
- Forecast.jsx reads col_meta first to pick /agg vs /data; the Arrow
  streaming logic is extracted to fetchArrow() since both share it.
- Setup.jsx gains a grain checkbox and shows the resulting grain.
- 01_schema.sql: move the col_meta ALTERs after its CREATE TABLE — they
  referenced the table before it existed on a fresh install.

All six generated statements verified to plan against the real forecast
table; the in_grain column has been added to the dev database.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

CLAUDE.md

@@ -49,7 +49,7 @@ ui/src/
 ## Database schema (`pf`)
 
 - **`pf.source`** — registered source tables
-- **`pf.col_meta`** — column roles: `dimension` | `value` | `units` | `date` | `filter` | `ignore`; `is_key` marks dimensions used in slice WHERE clauses; `dim_group` groups functionally dependent columns (e.g. date + its derived year/month dimensions); `dim_period_col` maps a dimension to a `pf.dim_period` column so date-adjacent values are derived at load time rather than copied raw
+- **`pf.col_meta`** — column roles: `dimension` | `value` | `units` | `date` | `filter` | `ignore`; `is_key` marks dimensions used in slice WHERE clauses; `dim_group` groups functionally dependent columns (e.g. date + its derived year/month dimensions); `dim_period_col` maps a dimension to a `pf.dim_period` column so date-adjacent values are derived at load time rather than copied raw; `in_grain` flags dimension/date columns that define the **display grain** (see below)
 - **`pf.version`** — named forecast scenarios; `exclude_iters` (default `["reference"]`) blocks those iter values from all operations
 - **`pf.fc_{tname}_{version_id}`** — one forecast table per version; contains both operational rows (`pf_iter = baseline|scale|recode|clone`) and reference rows (`pf_iter = reference`)
 - **`pf.log`** — audit log; every write gets one entry; `slice` + `params` stored as jsonb
@@ -64,15 +64,23 @@ ui/src/
 ## Core data flow
 
 ### Initial load (Forecast view)
-`GET /api/versions/:id/data` → Arrow IPC binary stream → `worker.table(buffer)` in Perspective WASM
+`Forecast.jsx` fetches col_meta first, then picks the endpoint:
+
+- **grain mode** (any `in_grain` column) — `GET /api/versions/:id/agg`, rows pre-aggregated to the grain, table indexed on `pf_gkey`
+- **raw mode** (no grain) — `GET /api/versions/:id/data`, raw forecast rows, table indexed on `pf_id`
+
+Either way: Arrow IPC binary stream → `worker.table(buffer)` in Perspective WASM. `fetchArrow()` handles both.
 
 **Why one batch (not streaming):** pg returns `bigint`/`numeric` as strings by default — type parsers in `server.js` coerce them to numbers. Per-batch Arrow encoding creates independent dictionaries that cause Perspective WASM to crash on dictionary replacement messages. Server accumulates all rows, emits one record batch.
 
+### Display grain
+Aggregating to the grain the pivot actually displays is the load-time fix — measured 534,902 → 6,154 rows on `osm_stack`. It keeps the **native** Perspective engine, so expand/collapse/depth/sort/filter all still work. Set the grain in Setup (`in_grain` per column); it is baked into `pf.sql` at Generate SQL time so load and operations agree. `grainOf()` in `lib/sql_generator.js` is the single definition of what the grain is — `Setup.jsx` and `routes/log.js` mirror it. Full design: `pf_spec.md` → §Display-grain pre-aggregation. Why not a DuckDB virtual server: `pf_perspective_options.md` → §Spike findings.
+
 ### Forecast operations
-POST to `/api/versions/:id/{scale|recode|clone}` → SQL executed with `RETURNING *` → new rows returned as JSON → `pspTable.update(rows)` — no full reload.
+POST to `/api/versions/:id/{scale|recode|clone}` → SQL executed with `RETURNING *` → new rows returned as JSON → `pspTable.update(rows)` — no full reload. In grain mode the operation's final CTE aggregates its own new rows to grain first; since `pf_logid` is part of `pf_gkey` those keys are always new, so `update()` **appends** and the view re-sums.
 
 ### Undo
-`DELETE /api/log/:logid` → removes rows by logid → **full Perspective reload** (known wart).
+`DELETE /api/log/:logid` → removes rows by logid → `table.remove()` of the affected index values (`pf_gkeys` in grain mode, `pf_ids` in raw mode); the view re-sums. No full reload.
 
 ---
 
@@ -114,6 +122,8 @@ Theme state lives in `ui/src/theme.jsx` — a React context (`ThemeContext`) wit
 - Default pivot layout should be configurable per source (currently hardcodes first 2 dimensions)
 - Source/version selection doesn't persist across page reload
 - Col_meta / version schema drift: if col_meta roles change after a version's forecast table is created, SQL and DDL go out of sync — workaround is to delete and recreate the version
+- Grain drift: changing `in_grain` after a load requires Generate SQL + a page reload, since the loaded table's index and columns are fixed at load time. `routes/log.js` derives the grain from live col_meta, so a grain changed mid-session yields `pf_gkeys` that don't match the loaded table and undo silently removes nothing
+- Grain is static per source — a dimension left unflagged cannot be pivoted on. Dynamic per-cut grain (intersect the viewer's field set with the eligible set) is the additive next step; see `pf_spec.md` → §Display-grain pre-aggregation
 
 ## Deferred (not in v1)
-Baseline replay (`replay: true` returns 501), approval workflow, territory filtering, export, version comparison, multi-DB connections.
+Baseline replay (`replay: true` returns 501), approval workflow, territory filtering, export, version comparison, multi-DB connections. Live server-side aggregation (Path A / DuckDB virtual server) is parked on branch `spike/duckdb-virtual-server`.

lib/sql_generator.js

@@ -1,6 +1,10 @@
 // Generates operation SQL for a source table, baking in column names from col_meta.
 // Runtime values are left as {{token}} substitution points.
 //
+// Columns flagged col_meta.in_grain define a display grain. When one is set the
+// initial load (get_agg) and every operation return rows pre-aggregated to that
+// grain and keyed on pf_gkey, instead of raw forecast rows keyed on pf_id.
+//
 // Tokens baked in at generation time:  column names, source schema.table
 // Tokens substituted at request time:  {{fc_table}}, {{where_clause}}, {{exclude_clause}},
 //                                      {{version_id}}, {{logid}}, {{pf_user}}, {{note}},
@@ -10,6 +14,37 @@
 // wrap a column name in double quotes for safe use in SQL
 function q(name) { return `"${name}"`; }
 
+// The display grain: dimension/date columns flagged in_grain, plus pf_iter and
+// pf_logid which are always part of it. Returns null when nothing is flagged —
+// that is raw-row mode, where operations return whole rows and the client
+// indexes on pf_id (the pre-grain behaviour).
+//
+// Keeping pf_logid in the grain is what makes the append model work: each
+// operation's contribution stays a distinct row, so table.update() accumulates
+// rather than replacing a bucket total, and undo can remove exactly that
+// operation's rows.
+function grainOf(colMeta) {
+    const cols = colMeta
+        .filter(c => c.in_grain && (c.role === 'dimension' || c.role === 'date'))
+        .sort((a, b) => (a.opos || 0) - (b.opos || 0))
+        .map(c => c.cname);
+    if (cols.length === 0) return null;
+
+    // pf_gkey must be unique per grain tuple. chr(31) (unit separator) joins the
+    // parts and chr(30) stands in for NULL, so ('a', NULL) cannot collide with
+    // (NULL, 'a') and a NULL stays distinct from an empty string — a collision
+    // would silently merge two groups into one indexed row.
+    const key = (pfx = '') => `concat_ws(chr(31), ${[
+        ...cols.map(c => `COALESCE(${pfx}${q(c)}::text, chr(30))`),
+        `${pfx}pf_iter`,
+        `${pfx}pf_logid::text`
+    ].join(', ')})`;
+
+    const groupCols = (pfx = '') => [...cols.map(c => `${pfx}${q(c)}`), `${pfx}pf_iter`, `${pfx}pf_logid`];
+
+    return { cols, key, groupCols };
+}
+
 function generateSQL(source, colMeta) {
     const dims = colMeta
         .filter(c => c.role === 'dimension')
@@ -45,8 +80,26 @@ function generateSQL(source, colMeta) {
     );
     const hasDimPeriod = dimPeriodMap.size > 0;
 
+    // display grain — when set, initial load and operations both return rows
+    // pre-aggregated to it instead of raw forecast rows
+    const grain = grainOf(colMeta);
+    // A flag on anything other than a dimension/date column is ignored by grainOf,
+    // which is what we want — the role change is the source of truth, not a stale flag.
+    if (grain) {
+        const missing = grain.cols.filter(c => !dataCols.includes(c));
+        if (missing.length > 0) {
+            throw new Error(
+                `Grain columns are never populated in the forecast table: ${missing.join(', ')}`
+            );
+        }
+        if (!effectiveValue && !effectiveUnits) {
+            throw new Error('A grain requires at least one value or units column to aggregate');
+        }
+    }
+
     return {
         get_data:  buildGetData(),
+        ...(grain ? { get_agg: buildGetAgg() } : {}),
         baseline:  buildBaseline(),
         reference: buildReference(),
         scale:     buildScale(),
@@ -59,6 +112,43 @@ function generateSQL(source, colMeta) {
         return `SELECT * FROM {{fc_table}}`;
     }
 
+    // Aggregate the whole forecast table to the display grain. This is the initial
+    // load for grain sources — the client loads the result into a native Perspective
+    // table indexed on pf_gkey and its view sums across these rows, exactly as an
+    // Excel pivot cache sums its data tab.
+    function buildGetAgg() {
+        return `
+SELECT
+    ${grainSelect()}
+FROM {{fc_table}}
+GROUP BY
+    ${grain.groupCols().join('\n    ,')}`.trim();
+    }
+
+    // grain columns + pf_gkey + summed measures, in the leading-comma style the
+    // rest of the generated SQL uses
+    function grainSelect(pfx = '') {
+        return [
+            ...grain.groupCols(pfx),
+            `${grain.key(pfx)} AS pf_gkey`,
+            effectiveValue ? `SUM(${pfx}${q(effectiveValue)}) AS ${q(effectiveValue)}` : null,
+            effectiveUnits ? `SUM(${pfx}${q(effectiveUnits)}) AS ${q(effectiveUnits)}` : null
+        ].filter(Boolean).join('\n    ,');
+    }
+
+    // Tail of an operation statement: in grain mode the inserted rows come back
+    // aggregated to grain (the client appends them and lets the view re-sum);
+    // otherwise whole rows come back as before.
+    function opTail(cte) {
+        if (!grain) return `SELECT * FROM ${cte}`;
+        return `
+SELECT
+    ${grainSelect()}
+FROM ${cte}
+GROUP BY
+    ${grain.groupCols().join('\n    ,')}`.trim();
+    }
+
     function buildLoadSelect(pfx) {
         // pfx: table alias prefix ('s.' when joining dim_period, '' otherwise)
         return dataCols.map(c => {
@@ -151,7 +241,7 @@ ilog AS (
     FROM base
     RETURNING *
 )
-SELECT * FROM ins`.trim();
+${opTail('ins')}`.trim();
     }
 
     function buildRecode() {
@@ -182,7 +272,12 @@ ilog AS (
     FROM src
     RETURNING *
 )
-SELECT * FROM neg UNION ALL SELECT * FROM ins`.trim();
+${grain ? `,allrows AS (
+    SELECT * FROM neg
+    UNION ALL
+    SELECT * FROM ins
+)
+${opTail('allrows')}` : 'SELECT * FROM neg UNION ALL SELECT * FROM ins'}`.trim();
     }
 
     function buildClone() {
@@ -205,7 +300,7 @@ ilog AS (
     {{exclude_clause}}
     RETURNING *
 )
-SELECT * FROM ins`.trim();
+${opTail('ins')}`.trim();
     }
 
     function buildUndo() {
@@ -309,4 +404,4 @@ function esc(val) {
     return String(val).replace(/'/g, "''");
 }
 
-module.exports = { generateSQL, applyTokens, buildWhere, buildExcludeClause, buildSetClause, buildFilterClause, esc };
+module.exports = { generateSQL, grainOf, applyTokens, buildWhere, buildExcludeClause, buildSetClause, buildFilterClause, esc };

pf_spec.md

@@ -714,13 +714,26 @@ DELETE FROM pf.log WHERE id = {{logid}};
 
 ---
 
-## Display-grain pre-aggregation (planned)
+## Display-grain pre-aggregation
 
-**Status:** designed, not yet built. This is the concrete design for **Path B**
+**Status:** built (static grain). This is **Path B** (pre-aggregated extract →
-(pre-aggregated extract → native Perspective table) of two candidate designs;
+native Perspective table) of two candidate designs; rationale, the Path A
-rationale, the Path A alternative (live virtual-server aggregation), the spike
+alternative (live virtual-server aggregation), the spike evidence, and the
-evidence, and the A-vs-B trade-off live in `pf_perspective_options.md`
+A-vs-B trade-off live in `pf_perspective_options.md` (§Two candidate designs,
-(§Two candidate designs, §Spike findings).
+§Spike findings).
+
+The grain is **static** — set once per source in Setup and baked into the stored
+`pf.sql` templates, so load and operations agree by construction. `in_grain`
+means *eligible for the grain*, and in this version the grain is exactly the set
+of eligible columns. Deriving a narrower grain per pivot at request time (the
+dynamic variant) is then additive: intersect the viewer's field set with the
+eligible set. Leaving high-cardinality columns (`part`, raw day dates,
+currency-level detail) unflagged is what keeps the grain from exploding back
+toward raw, regardless of what a user drags into the pivot.
+
+**Measured on `pf.fc_osm_stack_20`** at `pending_rep × customer × smon`:
+534,902 → **6,154** rows (≈87×), `pf_gkey` unique across all 6,154, and both
+measures reconcile exactly to the raw totals (283,296,087.67 / 962,142,261.46).
 
 **Problem it solves.** The current transport ships every raw forecast row to the
 browser (≈535k rows / ~250 MB / ~2 min on `osm_stack`). Perspective then pivots
@@ -751,7 +764,11 @@ the stored `pf.sql` templates, so initial load and operations agree on it.
 ### Initial load — `GET /api/versions/:id/agg`
 
 Replaces the raw `/data` stream for grain-based versions. Aggregates the forecast
-table to the stored grain and returns Arrow IPC:
+table to the stored grain and returns Arrow IPC. The template is stored in
+`pf.sql` as operation `get_agg`, generated only when a grain is defined; clearing
+the grain and regenerating removes it, and the client falls back to `/data`.
+Both endpoints speak the same protocol (one record batch plus an `X-Row-Count`
+header), so the client only chooses the URL:
 
 ```sql
 SELECT
@@ -783,6 +800,12 @@ the smallest change from today's code, which already appends operation results v
 accumulate (rather than replacing a bucket) and a delete can remove exactly that
 operation's rows.
 
+As built, the concatenation is
+`concat_ws(chr(31), COALESCE(col::text, chr(30)), …, pf_iter, pf_logid::text)`.
+The separator and NULL sentinel matter: plain `concat_ws` skips NULLs, so
+`('a', NULL)` and `(NULL, 'a')` would produce the same key and silently merge two
+groups into one indexed row. `chr(30)` also keeps NULL distinct from `''`.
+
 ### Write path (scale / recode / clone) — append the new log entry's rows
 
 Operations INSERT raw rows into `{{fc_table}}` under a new `pf_logid` as today; the
@@ -813,9 +836,20 @@ because each row carries the new, unique `pf_logid`.)
 A logid's rows are uniquely keyed, so undo just removes them and lets the view
 re-sum — no re-aggregation, no emptied-bucket handling, no snapshot caveat:
 
+`RETURNING` does not accept `DISTINCT`, so the delete feeds a CTE that reduces its
+output to the distinct grain keys. `rows_deleted` still counts raw rows removed:
+
 ```sql
-DELETE FROM {{fc_table}} WHERE pf_logid = {{logid}}
+WITH
-RETURNING DISTINCT {{grain_cols}}, pf_iter, pf_logid;   -- → pf_gkeys to remove
+del AS (
+    DELETE FROM {{fc_table}}
+    WHERE pf_logid = {{logid}}
+    RETURNING {{grain_cols}}, pf_iter, pf_logid
+)
+SELECT
+     count(*)::int                    AS rows_deleted
+    ,array_agg(DISTINCT {{grain_key}}) AS pf_gkeys
+FROM del;
 DELETE FROM pf.log WHERE id = {{logid}};
 ```
 

routes/log.js

@@ -1,4 +1,5 @@
 const express = require('express');
+const { grainOf } = require('../lib/sql_generator');
 const { fcTable } = require('../lib/utils');
 
 module.exports = function(pool) {
@@ -51,7 +52,7 @@ module.exports = function(pool) {
         const logId = parseInt(req.params.logid);
         try {
             const logResult = await pool.query(`
-                SELECT l.*, v.status, s.tname, v.id AS version_id
+                SELECT l.*, v.status, s.tname, v.id AS version_id, v.source_id
                 FROM pf.log l
                 JOIN pf.version v ON v.id = l.version_id
                 JOIN pf.source s ON s.id = v.source_id
@@ -61,18 +62,46 @@ module.exports = function(pool) {
             const log = logResult.rows[0];
             if (log.status === 'closed') return res.status(403).json({ error: 'Version is closed' });
             const table = fcTable(log.tname, log.version_id);
+
+            // In grain mode the client's table is indexed on pf_gkey, so undo has to
+            // report the grain keys to remove rather than raw pf_ids. The keys are
+            // distinct while rows_deleted still counts the raw rows removed.
+            const colMeta = await pool.query(
+                `SELECT cname, role, in_grain, opos FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`,
+                [log.source_id]
+            );
+            const grain = grainOf(colMeta.rows);
+
             const client = await pool.connect();
             try {
                 await client.query('BEGIN');
-                const deleted = await client.query(
+                const deleted = grain
-                    `DELETE FROM ${table} WHERE pf_logid = $1 RETURNING pf_id`, [logId]
+                    ? await client.query(`
-                );
+                        WITH
+                        del AS (
+                            DELETE FROM ${table}
+                            WHERE pf_logid = $1
+                            RETURNING ${grain.groupCols().join(', ')}
+                        )
+                        SELECT
+                             count(*)::int                        AS rows_deleted
+                            ,array_agg(DISTINCT ${grain.key()})   AS pf_gkeys
+                        FROM del
+                      `, [logId])
+                    : await client.query(
+                        `DELETE FROM ${table} WHERE pf_logid = $1 RETURNING pf_id`, [logId]
+                      );
                 await client.query('DELETE FROM pf.log WHERE id = $1', [logId]);
                 await client.query('COMMIT');
-                res.json({
+                res.json(grain
-                    rows_deleted: deleted.rowCount,
+                    ? {
-                    pf_ids: deleted.rows.map(r => r.pf_id)
+                        rows_deleted: deleted.rows[0].rows_deleted,
-                });
+                        pf_gkeys:     deleted.rows[0].pf_gkeys || []
+                      }
+                    : {
+                        rows_deleted: deleted.rowCount,
+                        pf_ids:       deleted.rows.map(r => r.pf_id)
+                      });
             } catch (err) {
                 await client.query('ROLLBACK');
                 throw err;

routes/operations.js

@@ -127,6 +127,37 @@ module.exports = function(pool) {
         }
     });
 
+    // Aggregate a version to its display grain and return it as Arrow IPC.
+    // This replaces /data for sources that define a grain (col_meta.in_grain):
+    // the aggregation collapses the row count by orders of magnitude, so the
+    // result loads as one small native Perspective table indexed on pf_gkey and
+    // the WASM view still does all rollup/expand/collapse locally.
+    router.get('/versions/:id/agg', async (req, res) => {
+        try {
+            const ctx = await getContext(parseInt(req.params.id), 'get_agg');
+            const sql = applyTokens(ctx.sql, { fc_table: ctx.table });
+            const { rows } = await runSQL(sql);
+
+            res.setHeader('Content-Type', 'application/vnd.apache.arrow.stream');
+            res.setHeader('X-Row-Count', String(rows.length));
+            if (rows.length === 0) { res.end(); return; }
+
+            // column arrays, one Arrow record batch — same constraint as /data:
+            // per-batch dictionaries crash Perspective's Arrow reader
+            const colArrays = Object.fromEntries(Object.keys(rows[0]).map(k => [k, []]));
+            for (const row of rows) {
+                for (const k of Object.keys(colArrays)) colArrays[k].push(row[k]);
+            }
+            const buf = tableToIPC(tableFromArrays(colArrays), 'stream');
+            res.setHeader('Content-Length', String(buf.byteLength));
+            res.end(Buffer.from(buf.buffer, buf.byteOffset, buf.byteLength));
+        } catch (err) {
+            console.error(err);
+            if (!res.headersSent) res.status(err.status || 500).json({ error: err.message });
+            else res.destroy();
+        }
+    });
+
     // load baseline rows from source table — additive, no delete
     router.post('/versions/:id/baseline', async (req, res) => {
         const { where_clause, date_offset, pf_user, note, filters, raw_where } = req.body;

routes/sources.js

@@ -89,14 +89,15 @@ module.exports = function(pool) {
             await client.query('BEGIN');
             for (const col of cols) {
                 await client.query(`
-                    INSERT INTO pf.col_meta (source_id, cname, label, role, is_key, dim_group, dim_period_col, opos)
+                    INSERT INTO pf.col_meta (source_id, cname, label, role, is_key, dim_group, dim_period_col, in_grain, opos)
-                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                     ON CONFLICT (source_id, cname) DO UPDATE SET
                         label          = EXCLUDED.label,
                         role           = EXCLUDED.role,
                         is_key         = EXCLUDED.is_key,
                         dim_group      = EXCLUDED.dim_group,
                         dim_period_col = EXCLUDED.dim_period_col,
+                        in_grain       = EXCLUDED.in_grain,
                         opos           = EXCLUDED.opos
                 `, [
                     sourceId,
@@ -106,6 +107,7 @@ module.exports = function(pool) {
                     col.is_key         || false,
                     col.dim_group      || null,
                     col.dim_period_col || null,
+                    col.in_grain       || false,
                     col.opos           || null
                 ]);
             }
@@ -166,6 +168,13 @@ module.exports = function(pool) {
                             generated_at = EXCLUDED.generated_at
                     `, [sourceId, operation, sql]);
                 }
+                // drop operations this generation no longer produces — e.g. get_agg
+                // after the grain has been cleared, which would otherwise leave a
+                // stale template the load path would still pick up
+                await client.query(
+                    `DELETE FROM pf.sql WHERE source_id = $1 AND operation <> ALL($2::text[])`,
+                    [sourceId, Object.keys(sqls)]
+                );
                 await client.query('COMMIT');
             } catch (err) {
                 await client.query('ROLLBACK');

setup_sql/01_schema.sql

@@ -17,8 +17,6 @@ CREATE TABLE IF NOT EXISTS pf.source (
 
 -- backfill columns for existing installs
 ALTER TABLE pf.source ADD COLUMN IF NOT EXISTS default_layout jsonb;
-ALTER TABLE pf.col_meta ADD COLUMN IF NOT EXISTS dim_group text;
-ALTER TABLE pf.col_meta ADD COLUMN IF NOT EXISTS dim_period_col text;
 
 -- pf.dim_period: run setup_sql/gen_dim_period.sql to create and populate
 
@@ -29,10 +27,18 @@ CREATE TABLE IF NOT EXISTS pf.col_meta (
     label       text,
     role        text NOT NULL DEFAULT 'ignore',   -- dimension | value | units | date | ignore
     is_key      boolean NOT NULL DEFAULT false,   -- true = usable in WHERE slice
+    dim_group   text,                             -- groups functionally dependent columns
+    dim_period_col text,                          -- pf.dim_period column this dimension derives from
+    in_grain    boolean NOT NULL DEFAULT false,   -- true = column defines the display grain
     opos        integer,
     UNIQUE (source_id, cname)
 );
 
+-- backfill columns for existing installs (must follow the CREATE above)
+ALTER TABLE pf.col_meta ADD COLUMN IF NOT EXISTS dim_group text;
+ALTER TABLE pf.col_meta ADD COLUMN IF NOT EXISTS dim_period_col text;
+ALTER TABLE pf.col_meta ADD COLUMN IF NOT EXISTS in_grain boolean NOT NULL DEFAULT false;
+
 CREATE TABLE IF NOT EXISTS pf.version (
     id              serial PRIMARY KEY,
     source_id       integer NOT NULL REFERENCES pf.source(id) ON DELETE RESTRICT,

ui/src/views/Forecast.jsx

@@ -148,6 +148,37 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
     }
   }
 
+  // Stream an Arrow IPC endpoint into one buffer, reporting download progress.
+  // Both /data and /agg speak the same protocol — a single record batch plus an
+  // X-Row-Count header — so the caller only picks the URL.
+  async function fetchArrow(url) {
+    const r = await fetch(url)
+    if (!r.ok) { const { error } = await r.json(); throw new Error(error || 'Failed to load data') }
+    const rowCount = parseInt(r.headers.get('X-Row-Count') || '0')
+    const total    = parseInt(r.headers.get('Content-Length') || '0') || null
+    const reader   = r.body.getReader()
+    const chunks   = []
+    let received   = 0
+    let lastUpdate = 0
+    setLoadProgress({ received: 0, total })
+    while (true) {
+      const { done, value } = await reader.read()
+      if (done) break
+      chunks.push(value)
+      received += value.byteLength
+      const now = Date.now()
+      if (now - lastUpdate >= 100) {
+        setLoadProgress({ received, total })
+        lastUpdate = now
+      }
+    }
+    setLoadProgress({ received, total })
+    const merged = new Uint8Array(received)
+    let pos = 0
+    for (const c of chunks) { merged.set(c, pos); pos += c.byteLength }
+    return { buffer: merged.buffer, rowCount }
+  }
+
   function loadLayouts(vid) {
     const stored = localStorage.getItem(LAYOUTS_KEY(vid))
     setLayouts(stored ? JSON.parse(stored) : [])
@@ -164,43 +195,34 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
     setSlice({})
     expandDepthRef.current = null
     try {
-      const [perspective, dataResult, meta] = await Promise.all([
+      // col_meta first — it decides which endpoint to load from, and it is a tiny
+      // query next to the data fetch it gates.
+      const meta = await fetch(`/api/sources/${sid}/cols`).then(r => r.json())
+      colMetaRef.current = meta
+
+      // Grain mode: the source declares a display grain, so the server ships rows
+      // already aggregated to it and the table is indexed on pf_gkey. Without a
+      // grain we load raw forecast rows indexed on pf_id, as before.
+      const grainMeta  = meta.filter(c => c.in_grain && ['dimension','date'].includes(c.role))
+      const grainMode  = grainMeta.length > 0
+      const indexCol   = grainMode ? 'pf_gkey' : 'pf_id'
+
+      const [perspective, dataResult] = await Promise.all([
         loadPerspective(),
-        fetch(`/api/versions/${vid}/data`).then(async r => {
+        fetchArrow(`/api/versions/${vid}/${grainMode ? 'agg' : 'data'}`),
-          if (!r.ok) { const { error } = await r.json(); throw new Error(error || 'Failed to load data') }
-          const rowCount = parseInt(r.headers.get('X-Row-Count') || '0')
-          const total    = parseInt(r.headers.get('Content-Length') || '0') || null
-          const reader   = r.body.getReader()
-          const chunks   = []
-          let received   = 0
-          let lastUpdate = 0
-          setLoadProgress({ received: 0, total })
-          while (true) {
-            const { done, value } = await reader.read()
-            if (done) break
-            chunks.push(value)
-            received += value.byteLength
-            const now = Date.now()
-            if (now - lastUpdate >= 100) {
-              setLoadProgress({ received, total })
-              lastUpdate = now
-            }
-          }
-          setLoadProgress({ received, total })
-          const merged = new Uint8Array(received)
-          let pos = 0
-          for (const c of chunks) { merged.set(c, pos); pos += c.byteLength }
-          return { buffer: merged.buffer, rowCount }
-        }),
-        fetch(`/api/sources/${sid}/cols`).then(r => r.json()),
       ])
 
       const { buffer, rowCount } = dataResult
-      colMetaRef.current = meta
+      const validCols = new Set(grainMode
-      const validCols = new Set([
+        ? [
-        ...meta.filter(c => ['dimension','value','units','date'].includes(c.role)).map(c => c.cname),
+            ...grainMeta.map(c => c.cname),
-        'pf_id', 'pf_iter', 'pf_logid', 'pf_user', 'created_at',
+            ...meta.filter(c => ['value','units'].includes(c.role)).map(c => c.cname),
-      ])
+            'pf_gkey', 'pf_iter', 'pf_logid',
+          ]
+        : [
+            ...meta.filter(c => ['dimension','value','units','date'].includes(c.role)).map(c => c.cname),
+            'pf_id', 'pf_iter', 'pf_logid', 'pf_user', 'created_at',
+          ])
       const tableName = `fc_${vid}`
 
       if (rowCount >= 500000) setLargeDataset(true)
@@ -221,7 +243,7 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
         if (stale) await stale.delete()
       } catch {}
 
-      const opts = { name: tableName, index: 'pf_id' }
+      const opts = { name: tableName, index: indexCol }
       tableRef.current = await (rowCount > 0 ? worker.table(buffer, opts) : worker.table([], opts))
 
       if (myId !== initIdRef.current) {
@@ -523,8 +545,11 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
       const data = await res.json()
       if (!res.ok) { flash(data.error, 'error'); return }
       setLogEntries(prev => prev.filter(e => e.id !== logId))
-      if (data.pf_ids?.length && tableRef.current) {
+      // grain versions report pf_gkeys, raw versions pf_ids — either way these are
-        await tableRef.current.remove(data.pf_ids)
+      // the index values of the rows to drop, and the view re-sums what remains
+      const removed = data.pf_gkeys ?? data.pf_ids
+      if (removed?.length && tableRef.current) {
+        await tableRef.current.remove(removed)
       }
       flash(`Undone — ${data.rows_deleted} rows removed`)
     } catch (err) {

ui/src/views/Setup.jsx

@@ -173,6 +173,11 @@ export default function Setup({ refreshSources }) {
 
   const registeredKeys = new Set(sources.map(s => `${s.schema}.${s.tname}`))
 
+  // display grain — must match grainOf() in lib/sql_generator.js
+  const grainCols = editedCols
+    .filter(c => c.in_grain && (c.role === 'dimension' || c.role === 'date'))
+    .map(c => c.cname)
+
   return (
     <div className="h-full flex overflow-hidden text-sm">
 
@@ -278,6 +283,11 @@ export default function Setup({ refreshSources }) {
               <div className="px-3 py-2 border-b border-gray-100 flex items-center justify-between shrink-0">
                 <span className="text-xs font-medium text-gray-500 uppercase tracking-wide">
                   Col Meta — <span className="text-gray-700 normal-case">{selectedSource.schema}.{selectedSource.tname}</span>
+                  <span className="ml-3 normal-case font-normal text-gray-400" title="Columns the forecast load is pre-aggregated to">
+                    grain: {grainCols.length
+                      ? <span className="font-mono text-gray-600">{grainCols.join(' × ')}</span>
+                      : <span className="italic">none — raw rows</span>}
+                  </span>
                 </span>
                 <div className="flex items-center gap-2">
                   {colsDirty && (
@@ -302,6 +312,7 @@ export default function Setup({ refreshSources }) {
                       <th className="px-3 py-1.5 font-medium">column</th>
                       <th className="px-3 py-1.5 font-medium">role</th>
                       <th className="px-3 py-1.5 font-medium text-center">key</th>
+                      <th className="px-3 py-1.5 font-medium text-center" title="Include this column in the display grain — the load is pre-aggregated to the flagged columns">grain</th>
                       <th className="px-3 py-1.5 font-medium">group</th>
                       <th className="px-3 py-1.5 font-medium">period col</th>
                       <th className="px-3 py-1.5 font-medium">label</th>
@@ -329,6 +340,15 @@ export default function Setup({ refreshSources }) {
                             className="cursor-pointer disabled:opacity-20"
                           />
                         </td>
+                        <td className="px-3 py-1.5 text-center">
+                          <input
+                            type="checkbox"
+                            checked={!!col.in_grain}
+                            onChange={e => updateCol(i, 'in_grain', e.target.checked)}
+                            disabled={col.role !== 'dimension' && col.role !== 'date'}
+                            className="cursor-pointer disabled:opacity-20"
+                          />
+                        </td>
                         <td className="px-3 py-1.5">
                           <input
                             type="text"