From 654a3686720f49b310b3c315a2d9f7859dfd43e0 Mon Sep 17 00:00:00 2001 From: Paul Trowbridge Date: Tue, 18 Aug 2026 16:25:38 -0400 Subject: [PATCH] Add static display-grain pre-aggregation (col_meta.in_grain) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CLAUDE.md | 20 ++++++-- lib/sql_generator.js | 103 ++++++++++++++++++++++++++++++++++++-- pf_spec.md | 52 +++++++++++++++---- routes/log.js | 45 ++++++++++++++--- routes/operations.js | 31 ++++++++++++ routes/sources.js | 13 ++++- setup_sql/01_schema.sql | 10 +++- ui/src/views/Forecast.jsx | 97 ++++++++++++++++++++++------------- ui/src/views/Setup.jsx | 20 ++++++++ 9 files changed, 325 insertions(+), 66 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 9591095..0706797 100644 --- a/CLAUDE.md +++ b/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`. diff --git a/lib/sql_generator.js b/lib/sql_generator.js index fdba4d1..58c0ffb 100644 --- a/lib/sql_generator.js +++ b/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 }; diff --git a/pf_spec.md b/pf_spec.md index 62b2362..09ce5f2 100644 --- a/pf_spec.md +++ b/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** -(pre-aggregated extract → native Perspective table) of two candidate designs; -rationale, the Path A alternative (live virtual-server aggregation), the spike -evidence, and the A-vs-B trade-off live in `pf_perspective_options.md` -(§Two candidate designs, §Spike findings). +**Status:** built (static grain). This is **Path B** (pre-aggregated extract → +native Perspective table) of two candidate designs; rationale, the Path A +alternative (live virtual-server aggregation), the spike evidence, and the +A-vs-B trade-off live in `pf_perspective_options.md` (§Two candidate designs, +§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}} -RETURNING DISTINCT {{grain_cols}}, pf_iter, pf_logid; -- → pf_gkeys to remove +WITH +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}}; ``` diff --git a/routes/log.js b/routes/log.js index d14d0f1..5ea0f07 100644 --- a/routes/log.js +++ b/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( - `DELETE FROM ${table} WHERE pf_logid = $1 RETURNING pf_id`, [logId] - ); + const deleted = grain + ? 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({ - rows_deleted: deleted.rowCount, - pf_ids: deleted.rows.map(r => r.pf_id) - }); + res.json(grain + ? { + 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; diff --git a/routes/operations.js b/routes/operations.js index b7da282..b369061 100644 --- a/routes/operations.js +++ b/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; diff --git a/routes/sources.js b/routes/sources.js index f09d77e..9ef24ce 100644 --- a/routes/sources.js +++ b/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) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + 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, $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'); diff --git a/setup_sql/01_schema.sql b/setup_sql/01_schema.sql index b555002..2a9e0d9 100644 --- a/setup_sql/01_schema.sql +++ b/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, diff --git a/ui/src/views/Forecast.jsx b/ui/src/views/Forecast.jsx index b83e459..67387a5 100644 --- a/ui/src/views/Forecast.jsx +++ b/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 => { - 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()), + fetchArrow(`/api/versions/${vid}/${grainMode ? 'agg' : 'data'}`), ]) const { buffer, rowCount } = dataResult - colMetaRef.current = meta - const validCols = new Set([ - ...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 validCols = new Set(grainMode + ? [ + ...grainMeta.map(c => c.cname), + ...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) { - await tableRef.current.remove(data.pf_ids) + // grain versions report pf_gkeys, raw versions pf_ids — either way these are + // 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) { diff --git a/ui/src/views/Setup.jsx b/ui/src/views/Setup.jsx index 60c8177..522f369 100644 --- a/ui/src/views/Setup.jsx +++ b/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 (
@@ -278,6 +283,11 @@ export default function Setup({ refreshSources }) {
Col Meta — {selectedSource.schema}.{selectedSource.tname} + + grain: {grainCols.length + ? {grainCols.join(' × ')} + : none — raw rows} +
{colsDirty && ( @@ -302,6 +312,7 @@ export default function Setup({ refreshSources }) { column role key + grain group period col label @@ -329,6 +340,15 @@ export default function Setup({ refreshSources }) { className="cursor-pointer disabled:opacity-20" /> + + updateCol(i, 'in_grain', e.target.checked)} + disabled={col.role !== 'dimension' && col.role !== 'date'} + className="cursor-pointer disabled:opacity-20" + /> +