pt/pf_app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge adjustment row collapse and stamped log totals

Browse Source 
Adjustments write one row per coordinate rather than one per row read, so an
operation no longer inherits the row count of every layer before it.
Verified live: a scale reading 2,948 rows wrote 908.

Each log entry records what it did -- row count, value and units, and the
columns those are denominated in -- stamped after the write rather than
recomputed by joining the whole forecast table every time the change log is
opened.
This commit is contained in:
Paul Trowbridge 2026-09-18 10:57:06 -04:00
commit b173a3ddaf
4 changed files with 157 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
lib/sql_generator.js
View File

@ -224,6 +224,22 @@ function generateSQL(source, colMeta) {
const cloneCols = [...dims, ...dateCols, effectiveValue, effectiveUnits].filter(Boolean);
const cloneInsertCols = [...cloneCols.map(q), 'pf_iter', 'pf_logid', 'pf_user', 'pf_created_at'].join(', ');
// An adjustment writes one row per *coordinate* it touches, not one per row
// it reads.
//
// Reading rows one-for-one meant every operation inherited the row count of
// everything before it: the baseline's rows plus every prior adjustment's
// rows at the same coordinate, so the table grew super-linearly with how
// much work had been done on it. Eight Pull Forward entries and the next
// scale over the same slice writes nine times what it needs to.
//
// Collapsing is over pf_logid and pf_iter only -- every stored dimension and
// date stays in the GROUP BY -- so no column goes null and nothing becomes
// unsliceable later. The distribution maths is untouched either way, since
// the window sums see the same totals whether or not the rows underneath
// them have been added up first.
const groupCols = (cols) => cols.map(q).join(',\n ');
const loadCols = [...dims, ...dateCols, ...valueCols, ...unitsCols];
const loadInsertCols = [...loadCols.map(q), 'pf_iter', 'pf_logid', 'pf_user', 'pf_created_at'].join(', ');
const dateColSet = new Set(dateCols);
@ -357,7 +373,7 @@ ilog AS (
WHERE {{filter_clause}}
RETURNING *
)
SELECT count(*) AS rows_affected FROM ins`.trim();
SELECT count(*) AS rows_affected, (SELECT id FROM ilog) AS log_id FROM ins`.trim();
}
function buildReference() {
@ -378,7 +394,7 @@ ilog AS (
WHERE {{filter_clause}}
RETURNING *
)
SELECT count(*) AS rows_affected FROM ins`.trim();
SELECT count(*) AS rows_affected, (SELECT id FROM ilog) AS log_id FROM ins`.trim();
}
function buildScale() {
@ -388,13 +404,16 @@ SELECT count(*) AS rows_affected FROM ins`.trim();
const uSel = effectiveUnits
? `round((${q(effectiveUnits)} / NULLIF(total_units, 0)) * {{units_incr}}, 5)`
: `0`;
// sum(sum(x)) OVER () is the aggregate of the aggregates: the window runs
// after the GROUP BY, so the total is over collapsed coordinates and
// comes to the same figure the ungrouped window produced.
const baseSelectParts = [
...dimsJoined ? [dimsJoined] : [],
q(dateCol),
effectiveValue ? q(effectiveValue) : null,
effectiveUnits ? q(effectiveUnits) : null,
effectiveValue ? `sum(${q(effectiveValue)}) OVER () AS total_value` : null,
effectiveUnits ? `sum(${q(effectiveUnits)}) OVER () AS total_units` : null
effectiveValue ? `sum(${q(effectiveValue)}) AS ${q(effectiveValue)}` : null,
effectiveUnits ? `sum(${q(effectiveUnits)}) AS ${q(effectiveUnits)}` : null,
effectiveValue ? `sum(sum(${q(effectiveValue)})) OVER () AS total_value` : null,
effectiveUnits ? `sum(sum(${q(effectiveUnits)})) OVER () AS total_units` : null
].filter(Boolean).join(',\n ');
return `
WITH
@ -409,6 +428,8 @@ ilog AS (
FROM {{fc_table}}
WHERE {{where_clause}}
{{exclude_clause}}
GROUP BY
${groupCols([...dims, dateCol])}
)
,ins AS (
INSERT INTO {{fc_table}} (${insertCols})
@ -430,10 +451,14 @@ ilog AS (
RETURNING id
)
,src AS (
SELECT ${selectData}
SELECT
${dimsJoined},
${q(dateCol)}${effectiveValue ? `,\n sum(${q(effectiveValue)}) AS ${q(effectiveValue)}` : ''}${effectiveUnits ? `,\n sum(${q(effectiveUnits)}) AS ${q(effectiveUnits)}` : ''}
FROM {{fc_table}}
WHERE {{where_clause}}
{{exclude_clause}}
GROUP BY
${groupCols([...dims, dateCol])}
)
,neg AS (
INSERT INTO {{fc_table}} (${insertCols})
@ -480,9 +505,15 @@ ilog AS (
SELECT
${select},
'clone', (SELECT id FROM ilog), '{{pf_user}}', now()
FROM {{fc_table}} s${hasDimPeriod ? dimPeriodJoins(dateGroups) : ''}
FROM (
SELECT
${groupCols([...dims, ...dateCols])}${effectiveValue ? `,\n sum(${q(effectiveValue)}) AS ${q(effectiveValue)}` : ''}${effectiveUnits ? `,\n sum(${q(effectiveUnits)}) AS ${q(effectiveUnits)}` : ''}
FROM {{fc_table}}
WHERE {{where_clause}}
{{exclude_clause}}
GROUP BY
${groupCols([...dims, ...dateCols])}
) s${hasDimPeriod ? dimPeriodJoins(dateGroups) : ''}
RETURNING *
)
${opTail('ins')}`.trim();

41
routes/log.js
View File

@ -30,6 +30,21 @@ module.exports = function(pool) {
unitsCol ? `sum(f."${unitsCol}")::float8 AS units_total` : `NULL::float8 AS units_total`
].join(', ');
// The totals are stamped onto the entry when it is written, so the
// normal read is a scan of a few dozen log rows rather than a join
// against millions of forecast rows.
//
// ?recount=1 does it the old way. Stored totals are fixed at write
// time and cannot drift on their own, but nothing stops someone
// deleting forecast rows by hand, and a stored figure has no way to
// notice. This is the way back -- and the backfill for entries
// written before the columns existed.
const recount = req.query.recount === '1' || req.query.recount === 'true';
const stamped = !recount && (await pool.query(
`SELECT count(*)::int AS n FROM pf.log
WHERE version_id = $1 AND row_count IS NULL`, [versionId]
)).rows[0].n === 0;
// ?kind=adjustments drops the baseline and reference entries. That is not
// only about what gets listed: the aggregate below joins the whole
// forecast table, and on a real version the load entries own almost
@ -41,7 +56,17 @@ module.exports = function(pool) {
? `AND l.operation NOT IN ('baseline', 'reference')`
: '';
const result = await pool.query(`
const result = stamped
? await pool.query(`
SELECT l.*,
$2::text AS value_col,
$3::text AS units_col
FROM pf.log l
WHERE l.version_id = $1
${opFilter}
ORDER BY l.id DESC
`, [versionId, valueCol || null, unitsCol || null])
: await pool.query(`
SELECT l.*, ${aggCols},
$2::text AS value_col,
$3::text AS units_col
@ -52,6 +77,20 @@ module.exports = function(pool) {
GROUP BY l.id
ORDER BY l.id DESC
`, [versionId, valueCol || null, unitsCol || null]);
// A recount is also a repair: write back what it found, so the next
// read is cheap again and the stored figure matches the rows.
if (recount) {
for (const r of result.rows) {
await pool.query(
`UPDATE pf.log SET row_count = $2, value_total = $3, units_total = $4,
measure_cols = $5::jsonb
WHERE id = $1`,
[r.id, r.row_count, r.value_total, r.units_total,
JSON.stringify({ value: valueCol || null, units: unitsCol || null })]
);
}
}
res.json(result.rows);
} catch (err) {
console.error(err);

49
routes/operations.js
View File

@ -93,6 +93,40 @@ module.exports = function(pool) {
});
}
// Stamp what the entry did onto the entry itself.
//
// An indexed lookup on pf_logid, run once at write time, in place of the
// change log joining the whole forecast table on every open. Safe to store
// rather than derive because these rows never change: only this operation
// inserts them, and the only thing that removes them is undo, which deletes
// the log row too.
//
// Best-effort by design. A failure here must not roll back a write that
// succeeded -- the totals can always be recomputed, the adjustment cannot.
async function stampLogTotals(client, ctx, logId) {
if (!logId) return;
const v = ctx.valueCol, u = ctx.unitsCol;
try {
await client.query(`
UPDATE pf.log SET
row_count = t.n,
value_total = t.v,
units_total = t.u,
measure_cols = $2::jsonb
FROM (
SELECT count(*)::int AS n
,${v ? `sum(f."${v}")::float8` : 'NULL::float8'} AS v
,${u ? `sum(f."${u}")::float8` : 'NULL::float8'} AS u
FROM ${ctx.table} f
WHERE f.pf_logid = $1
) t
WHERE pf.log.id = $1
`, [logId, JSON.stringify({ value: v || null, units: u || null })]);
} catch (err) {
console.error('[stampLogTotals]', err);
}
}
// echo back what the caller asked for, for the audit log
function pickIntent(body) {
const keys = ['mode', 'target_basis', 'value_incr', 'units_incr', 'value_pct', 'units_pct', 'pct',
@ -443,6 +477,7 @@ module.exports = function(pool) {
});
const result = await runSQL(sql);
await stampLogTotals(pool, ctx, result.rows[0]?.log_id);
res.json(result.rows[0]);
} catch (err) {
console.error(err);
@ -523,6 +558,7 @@ module.exports = function(pool) {
await client.query(`DELETE FROM pf.log WHERE id = $1`, [logid]);
const insResult = await client.query(sql);
await client.query('COMMIT');
await stampLogTotals(pool, ctx, insResult.rows[0]?.log_id);
res.json({
rows_deleted: delRows.rowCount,
@ -600,6 +636,7 @@ module.exports = function(pool) {
});
const result = await runSQL(sql);
await stampLogTotals(pool, ctx, result.rows[0]?.log_id);
res.json(result.rows[0]);
} catch (err) {
console.error(err);
@ -677,6 +714,10 @@ module.exports = function(pool) {
await client.query('COMMIT');
committed = true;
// one log id per unit: apply_mode 'each' writes an entry per slice
for (const id of new Set(allRows.map(r => r.pf_logid).filter(Boolean))) {
await stampLogTotals(pool, ctx, id);
}
const opLabel = (req.body.tag || '').trim() || note || null;
const rows = allRows.map(r => ({ ...r, pf_segment: ADJUSTMENT_SEGMENT, pf_bucket: ADJUSTMENT_BUCKET, pf_note: opLabel, pf_op: 'scale' }));
res.json({
@ -737,6 +778,10 @@ module.exports = function(pool) {
}
await client.query('COMMIT');
committed = true;
// one log id per unit: apply_mode 'each' writes an entry per slice
for (const id of new Set(allRows.map(r => r.pf_logid).filter(Boolean))) {
await stampLogTotals(pool, ctx, id);
}
const opLabel = (req.body.tag || '').trim() || note || null;
const rows = allRows.map(r => ({ ...r, pf_segment: ADJUSTMENT_SEGMENT, pf_bucket: ADJUSTMENT_BUCKET, pf_note: opLabel, pf_op: 'recode' }));
res.json({ rows, rows_affected: rows.length, slices_applied: units.length });
@ -832,6 +877,10 @@ module.exports = function(pool) {
}
await client.query('COMMIT');
committed = true;
// one log id per unit: apply_mode 'each' writes an entry per slice
for (const id of new Set(allRows.map(r => r.pf_logid).filter(Boolean))) {
await stampLogTotals(pool, ctx, id);
}
const opLabel = (req.body.tag || '').trim() || note || null;
const rows = allRows.map(r => ({ ...r, pf_segment: ADJUSTMENT_SEGMENT, pf_bucket: ADJUSTMENT_BUCKET, pf_note: opLabel, pf_op: 'clone' }));
res.json({ rows, rows_affected: rows.length, slices_applied: units.length });

17
setup_sql/01_schema.sql
View File

@ -122,6 +122,23 @@ ALTER TABLE pf.version ADD COLUMN IF NOT EXISTS adjustment_segment text;
ALTER TABLE pf.version ADD COLUMN IF NOT EXISTS adjustment_bucket text;
ALTER TABLE pf.version ADD COLUMN IF NOT EXISTS unlabeled_load text;
-- What the entry did, stamped when it did it.
--
-- Not a cache: a log entry's forecast rows never change after it is written.
-- Only the operation that owns the logid inserts them, and the only thing that
-- removes them is undo, which deletes this row too -- so these totals are fixed
-- at write time rather than derived from something that can move underneath
-- them. The change log was joining the whole forecast table to recompute them
-- on every open, 2.5M rows to total a few thousand.
--
-- measure_cols records which columns they are denominated in, since the value
-- and units roles can be reassigned in col_meta and the numbers would otherwise
-- quietly come to mean something else.
ALTER TABLE pf.log ADD COLUMN IF NOT EXISTS row_count integer;
ALTER TABLE pf.log ADD COLUMN IF NOT EXISTS value_total double precision;
ALTER TABLE pf.log ADD COLUMN IF NOT EXISTS units_total double precision;
ALTER TABLE pf.log ADD COLUMN IF NOT EXISTS measure_cols jsonb;
-- Master data for a dim_group: one row per key value, with its sibling columns.
--
-- The source is transactional and often a view over all history, so deriving a