Compare commits
No commits in common. "feature/static-grain" and "master" have entirely different histories.
feature/st
...
master
20
CLAUDE.md
20
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; `in_grain` flags dimension/date columns that define the **display grain** (see below)
|
||||
- **`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.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,23 +64,15 @@ ui/src/
|
||||
## Core data flow
|
||||
|
||||
### Initial load (Forecast view)
|
||||
`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.
|
||||
`GET /api/versions/:id/data` → Arrow IPC binary stream → `worker.table(buffer)` in Perspective WASM
|
||||
|
||||
**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. 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.
|
||||
POST to `/api/versions/:id/{scale|recode|clone}` → SQL executed with `RETURNING *` → new rows returned as JSON → `pspTable.update(rows)` — no full reload.
|
||||
|
||||
### Undo
|
||||
`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.
|
||||
`DELETE /api/log/:logid` → removes rows by logid → **full Perspective reload** (known wart).
|
||||
|
||||
---
|
||||
|
||||
@ -122,8 +114,6 @@ 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. Live server-side aggregation (Path A / DuckDB virtual server) is parked on branch `spike/duckdb-virtual-server`.
|
||||
Baseline replay (`replay: true` returns 501), approval workflow, territory filtering, export, version comparison, multi-DB connections.
|
||||
|
||||
@ -1,10 +1,6 @@
|
||||
// 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}},
|
||||
@ -14,37 +10,6 @@
|
||||
// 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')
|
||||
@ -80,26 +45,8 @@ 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(),
|
||||
@ -112,43 +59,6 @@ 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 => {
|
||||
@ -241,7 +151,7 @@ ilog AS (
|
||||
FROM base
|
||||
RETURNING *
|
||||
)
|
||||
${opTail('ins')}`.trim();
|
||||
SELECT * FROM ins`.trim();
|
||||
}
|
||||
|
||||
function buildRecode() {
|
||||
@ -272,12 +182,7 @@ ilog AS (
|
||||
FROM src
|
||||
RETURNING *
|
||||
)
|
||||
${grain ? `,allrows AS (
|
||||
SELECT * FROM neg
|
||||
UNION ALL
|
||||
SELECT * FROM ins
|
||||
)
|
||||
${opTail('allrows')}` : 'SELECT * FROM neg UNION ALL SELECT * FROM ins'}`.trim();
|
||||
SELECT * FROM neg UNION ALL SELECT * FROM ins`.trim();
|
||||
}
|
||||
|
||||
function buildClone() {
|
||||
@ -300,7 +205,7 @@ ilog AS (
|
||||
{{exclude_clause}}
|
||||
RETURNING *
|
||||
)
|
||||
${opTail('ins')}`.trim();
|
||||
SELECT * FROM ins`.trim();
|
||||
}
|
||||
|
||||
function buildUndo() {
|
||||
@ -404,4 +309,4 @@ function esc(val) {
|
||||
return String(val).replace(/'/g, "''");
|
||||
}
|
||||
|
||||
module.exports = { generateSQL, grainOf, applyTokens, buildWhere, buildExcludeClause, buildSetClause, buildFilterClause, esc };
|
||||
module.exports = { generateSQL, applyTokens, buildWhere, buildExcludeClause, buildSetClause, buildFilterClause, esc };
|
||||
|
||||
52
pf_spec.md
52
pf_spec.md
@ -714,26 +714,13 @@ DELETE FROM pf.log WHERE id = {{logid}};
|
||||
|
||||
---
|
||||
|
||||
## Display-grain pre-aggregation
|
||||
## Display-grain pre-aggregation (planned)
|
||||
|
||||
**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).
|
||||
**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).
|
||||
|
||||
**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
|
||||
@ -764,11 +751,7 @@ 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. 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:
|
||||
table to the stored grain and returns Arrow IPC:
|
||||
|
||||
```sql
|
||||
SELECT
|
||||
@ -800,12 +783,6 @@ 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
|
||||
@ -836,20 +813,9 @@ 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
|
||||
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 {{fc_table}} WHERE pf_logid = {{logid}}
|
||||
RETURNING DISTINCT {{grain_cols}}, pf_iter, pf_logid; -- → pf_gkeys to remove
|
||||
DELETE FROM pf.log WHERE id = {{logid}};
|
||||
```
|
||||
|
||||
|
||||
@ -1,5 +1,4 @@
|
||||
const express = require('express');
|
||||
const { grainOf } = require('../lib/sql_generator');
|
||||
const { fcTable } = require('../lib/utils');
|
||||
|
||||
module.exports = function(pool) {
|
||||
@ -52,7 +51,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, v.source_id
|
||||
SELECT l.*, v.status, s.tname, v.id AS version_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
|
||||
@ -62,43 +61,15 @@ 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 = 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(
|
||||
const deleted = 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(grain
|
||||
? {
|
||||
rows_deleted: deleted.rows[0].rows_deleted,
|
||||
pf_gkeys: deleted.rows[0].pf_gkeys || []
|
||||
}
|
||||
: {
|
||||
res.json({
|
||||
rows_deleted: deleted.rowCount,
|
||||
pf_ids: deleted.rows.map(r => r.pf_id)
|
||||
});
|
||||
|
||||
@ -127,37 +127,6 @@ 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;
|
||||
|
||||
@ -89,15 +89,14 @@ 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, in_grain, opos)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
|
||||
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)
|
||||
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,
|
||||
@ -107,7 +106,6 @@ module.exports = function(pool) {
|
||||
col.is_key || false,
|
||||
col.dim_group || null,
|
||||
col.dim_period_col || null,
|
||||
col.in_grain || false,
|
||||
col.opos || null
|
||||
]);
|
||||
}
|
||||
@ -168,13 +166,6 @@ 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');
|
||||
|
||||
@ -17,6 +17,8 @@ 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
|
||||
|
||||
@ -27,18 +29,10 @@ 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,
|
||||
|
||||
@ -148,11 +148,25 @@ 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)
|
||||
function loadLayouts(vid) {
|
||||
const stored = localStorage.getItem(LAYOUTS_KEY(vid))
|
||||
setLayouts(stored ? JSON.parse(stored) : [])
|
||||
setActiveLayoutId(null)
|
||||
}
|
||||
|
||||
async function initViewer(vid, sid) {
|
||||
const viewer = viewerRef.current
|
||||
if (!viewer) return
|
||||
const myId = ++initIdRef.current
|
||||
setLoading(true)
|
||||
setLargeDataset(false)
|
||||
setLoadProgress(null)
|
||||
setSlice({})
|
||||
expandDepthRef.current = null
|
||||
try {
|
||||
const [perspective, dataResult, meta] = 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
|
||||
@ -177,49 +191,13 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
|
||||
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) : [])
|
||||
setActiveLayoutId(null)
|
||||
}
|
||||
|
||||
async function initViewer(vid, sid) {
|
||||
const viewer = viewerRef.current
|
||||
if (!viewer) return
|
||||
const myId = ++initIdRef.current
|
||||
setLoading(true)
|
||||
setLargeDataset(false)
|
||||
setLoadProgress(null)
|
||||
setSlice({})
|
||||
expandDepthRef.current = null
|
||||
try {
|
||||
// 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(),
|
||||
fetchArrow(`/api/versions/${vid}/${grainMode ? 'agg' : 'data'}`),
|
||||
}),
|
||||
fetch(`/api/sources/${sid}/cols`).then(r => r.json()),
|
||||
])
|
||||
|
||||
const { buffer, rowCount } = dataResult
|
||||
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',
|
||||
]
|
||||
: [
|
||||
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',
|
||||
])
|
||||
@ -243,7 +221,7 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
|
||||
if (stale) await stale.delete()
|
||||
} catch {}
|
||||
|
||||
const opts = { name: tableName, index: indexCol }
|
||||
const opts = { name: tableName, index: 'pf_id' }
|
||||
tableRef.current = await (rowCount > 0 ? worker.table(buffer, opts) : worker.table([], opts))
|
||||
|
||||
if (myId !== initIdRef.current) {
|
||||
@ -545,11 +523,8 @@ 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))
|
||||
// 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)
|
||||
if (data.pf_ids?.length && tableRef.current) {
|
||||
await tableRef.current.remove(data.pf_ids)
|
||||
}
|
||||
flash(`Undone — ${data.rows_deleted} rows removed`)
|
||||
} catch (err) {
|
||||
|
||||
@ -173,11 +173,6 @@ 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">
|
||||
|
||||
@ -283,11 +278,6 @@ 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 && (
|
||||
@ -312,7 +302,6 @@ 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>
|
||||
@ -340,15 +329,6 @@ 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"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user