Stamp each log entry with what it did

The change log joined the whole forecast table on every open to total rows
it had just written -- 2.5M rows to report a few thousand, and the Baseline
page's row and value columns paid the same cost again.

The totals go onto pf.log at write time instead. This is not a cache that
can drift: a log entry's forecast rows never change once written, because
only the operation owning the logid inserts them and the only thing that
removes them is undo, which deletes the log row too. measure_cols records
which columns the figures 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.

Stamping is best-effort and runs after the commit: a failure to record what
happened must not roll back the thing that happened. The loads return their
new log id to make it possible, the adjustments take theirs from the rows
they return, and apply_mode 'each' writes one entry per slice, so it is a
set rather than a single id.

?recount=1 does it the old way and writes back what it finds. Stored totals
cannot drift on their own, but nothing stops someone deleting forecast rows
by hand, and a stored figure has no way to notice -- so there is a way back,
which doubles as the backfill for entries written before the columns
existed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

lib/sql_generator.js

@@ -373,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() {
@@ -394,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() {

routes/log.js

@@ -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,17 +56,41 @@ module.exports = function(pool) {
                 ? `AND l.operation NOT IN ('baseline', 'reference')`
                 : '';
 
-            const result = await pool.query(`
+            const result = stamped
-                SELECT l.*, ${aggCols},
+                ? await pool.query(`
-                       $2::text AS value_col,
+                    SELECT l.*,
-                       $3::text AS units_col
+                           $2::text AS value_col,
-                FROM pf.log l
+                           $3::text AS units_col
-                LEFT JOIN ${table} f ON f.pf_logid = l.id
+                    FROM pf.log l
-                WHERE l.version_id = $1
+                    WHERE l.version_id = $1
-                ${opFilter}
+                    ${opFilter}
-                GROUP BY l.id
+                    ORDER BY l.id DESC
-                ORDER BY l.id DESC
+                `, [versionId, valueCol || null, unitsCol || null])
-            `, [versionId, valueCol || null, unitsCol || null]);
+                : await pool.query(`
+                    SELECT l.*, ${aggCols},
+                           $2::text AS value_col,
+                           $3::text AS units_col
+                    FROM pf.log l
+                    LEFT JOIN ${table} f ON f.pf_logid = l.id
+                    WHERE l.version_id = $1
+                    ${opFilter}
+                    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);

routes/operations.js

@@ -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 });

setup_sql/01_schema.sql

@@ -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