Paul Trowbridge
19e52fa3fe
Scale and recode wrote dataCols, which held dateCols[0] and nothing else. On a source with order, requested and ship dates that means every scale and recode row was written with rdate and sdate_e null -- not stale, absent -- while the period dimensions derived from them (rseas, smon_e) were copied with the other dimensions and so read perfectly, which is what kept it out of sight. Both columns are role 'date' and therefore sliceable, so an adjustment scoped to a ship date matched no adjustment row that had ever been written, and anything dating the forecast by ship date dropped every adjustment without saying so. Clone already carried them all, for the date_offset it has to shift; the loads always did. The GROUP BY widens with the select list. On osm_skinny that is 971,185 groups against 971,352 -- ordnum and ordline already make a row nearly unique -- so the row count is unaffected in practice. Existing rows are not repaired; see Known issues. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
792 lines
35 KiB
JavaScript
792 lines
35 KiB
JavaScript
// 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}},
|
|
// {{label}}, {{bucket}}, {{tag}}, {{territory_clause}},
|
|
// {{params}}, {{slice}}, {{date_from}}, {{date_to}},
|
|
// {{value_incr}}, {{units_incr}}, {{set_clause}}, {{scale_factor}}
|
|
|
|
// What the pivot shows for a row's segment and its bucket.
|
|
//
|
|
// The ordering prefix is part of the stored text, not computed here. Perspective
|
|
// orders column groups by the value string, so "01 - Actual" is the only way an
|
|
// arbitrary order can be expressed -- and l.label is where a person types it.
|
|
// Nothing derives it, which is deliberate: an earlier design built the prefix from
|
|
// a separate seq column, as Perspective expressions on the client, and the prefix
|
|
// then existed only inside the pivot -- so every other reader disagreed with it,
|
|
// and a label that could not be expressed in ExprTK's printable-ASCII-per-byte
|
|
// string scanner could not be ordered at all. Stored text has neither problem.
|
|
//
|
|
// The single exception is the adjustment fallback, whose 99 keeps unlabelled
|
|
// adjustments last. Labelling an adjustment's log row overrides it, which is how
|
|
// one kind of adjustment is split out from the rest -- l.label rather than tag or
|
|
// note, so a segment name stays separable from adjustment commentary (pf_note).
|
|
//
|
|
// ---------------------------------------------------------------------------
|
|
// DISPLAY DEFAULTS -- every hardcoded name the pivot can show.
|
|
//
|
|
// These are the values a row falls back to when nobody has named it. They are
|
|
// the complete list: if a segment or bucket appears in the pivot under a name
|
|
// that is not in pf.log, it came from here. CLAUDE.md has the same list under
|
|
// "Hardcoded display names".
|
|
//
|
|
// They live on pf.version -- adjustment_segment, adjustment_bucket,
|
|
// unlabeled_load -- and the constants below are only the fallback for a version
|
|
// that has not set one. Read through a join at query time rather than
|
|
// substituted at generation: pf.sql templates are keyed on (source_id,
|
|
// operation) and shared by every version of a source, so a value baked in could
|
|
// not vary by version and regenerating for one would change the others.
|
|
//
|
|
// Exported because /data builds its own statement in routes/operations.js while
|
|
// /agg is generated here, and the two have to agree.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// An adjustment with no label of its own. The 99 keeps it after every numbered
|
|
// segment -- ordering is string ordering, so this only works while the loads
|
|
// carry 01-0n. Labelling an adjustment's log row overrides it, which is how one
|
|
// kind of adjustment is split out from the rest.
|
|
const ADJUSTMENT_SEGMENT = '99 - Adjustments';
|
|
|
|
// What an adjustment counts toward. Prefixed to match the segments it adjusts:
|
|
// unprefixed it read 'Forecast' while the loads read '04 - Forecast', and the
|
|
// column split in two -- the adjustments sitting apart from the rows they
|
|
// adjust. The number is a guess at the convention in use, which is the clearest
|
|
// argument for making this per-version.
|
|
const ADJUSTMENT_BUCKET = '04 - Forecast';
|
|
|
|
// A load nobody named. No prefix, so it sorts after everything numbered --
|
|
// letters follow digits in ASCII. The old '(unlabeled load)' sorted *first*,
|
|
// since '(' is 0x28 and digits begin at 0x30.
|
|
const UNLABELED_LOAD = 'Unlabeled';
|
|
|
|
const LOAD_SEGMENT = `COALESCE(NULLIF(l.label, ''), NULLIF(l.tag, ''), NULLIF(l.note, ''),
|
|
NULLIF(v.unlabeled_load, ''), '${UNLABELED_LOAD}')`;
|
|
|
|
const SEGMENT_EXPR = `CASE WHEN l.operation IN ('baseline','reference')
|
|
THEN ${LOAD_SEGMENT}
|
|
ELSE COALESCE(NULLIF(l.label, ''), NULLIF(v.adjustment_segment, ''), '${ADJUSTMENT_SEGMENT}')
|
|
END`;
|
|
|
|
// What the row counts towards. A load falls back to its own name until it is
|
|
// bucketed; an adjustment falls back to the forecast bucket, because that is
|
|
// what an adjustment is -- exclude_iters keeps operations off the reference
|
|
// segments, so there is no adjustment that is not part of the forecast.
|
|
const BUCKET_EXPR = `COALESCE(NULLIF(l.bucket, ''),
|
|
CASE WHEN l.operation IN ('baseline','reference')
|
|
THEN ${LOAD_SEGMENT}
|
|
ELSE COALESCE(NULLIF(v.adjustment_bucket, ''), '${ADJUSTMENT_BUCKET}')
|
|
END)`;
|
|
|
|
const NOTE_EXPR = `CASE WHEN l.operation IN ('baseline','reference')
|
|
THEN NULL
|
|
ELSE COALESCE(NULLIF(l.tag, ''), NULLIF(l.note, ''))
|
|
END`;
|
|
|
|
// Every pf.log column the two expressions above read, for /agg's GROUP BY: they
|
|
// are functionally dependent on pf_logid, which is in the grain, but Postgres
|
|
// will not infer that.
|
|
const LABEL_GROUP_COLS = ['l.operation', 'l.label', 'l.tag', 'l.note', 'l.bucket',
|
|
'v.adjustment_segment', 'v.adjustment_bucket', 'v.unlabeled_load'];
|
|
|
|
// The version carries the fallback names, so every statement that reads the
|
|
// expressions above needs it in scope as `v`. LEFT, not inner: a forecast row
|
|
// whose log entry somehow has no version should still come back, named by the
|
|
// constants.
|
|
const VERSION_JOIN = `
|
|
LEFT JOIN pf.version v
|
|
ON v.id = l.version_id`;
|
|
|
|
// 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.
|
|
//
|
|
// md5 of that, rather than the concatenation itself, because the key is an
|
|
// opaque handle -- nothing reads it but table.update() and table.remove().
|
|
// The raw form averaged 233 chars on a 24-column grain and, being unique per
|
|
// row, defeated Arrow's dictionary encoding: 65.6 MB of a 109 MB payload,
|
|
// more than every other column combined. 128 bits keeps collisions unreachable.
|
|
const key = (pfx = '') => `md5(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 };
|
|
}
|
|
|
|
// Date columns that anchor a dim_group, each with the dimensions derived from
|
|
// pf.dim_period for it. Shared so the generator and the routes agree on both the
|
|
// membership and the join aliases -- the same reason grainOf is a single function.
|
|
function dateGroupsOf(colMeta) {
|
|
return colMeta
|
|
.filter(c => c.role === 'date' && c.is_key && c.dim_group)
|
|
.map((keyCol, i) => ({
|
|
alias: `dp${i + 1}`,
|
|
dateCol: keyCol.cname,
|
|
group: keyCol.dim_group,
|
|
derived: colMeta
|
|
.filter(c => c.role === 'dimension'
|
|
&& c.dim_group === keyCol.dim_group
|
|
&& c.dim_period_col)
|
|
.map(c => ({ cname: c.cname, periodCol: c.dim_period_col })),
|
|
}))
|
|
.filter(g => g.derived.length > 0);
|
|
}
|
|
|
|
// cname -> which join it comes from and which of its columns
|
|
function dimPeriodMapOf(dateGroups) {
|
|
return new Map(
|
|
dateGroups.flatMap(g => g.derived.map(d => [d.cname, { alias: g.alias, periodCol: d.periodCol }]))
|
|
);
|
|
}
|
|
|
|
// The joins themselves, against a date that has already had {{date_offset}}
|
|
// applied. LEFT because a null date -- an order not yet shipped has no ship date
|
|
// -- must leave the period columns empty rather than drop the row.
|
|
function dimPeriodJoins(dateGroups, alias = 's') {
|
|
return dateGroups.map(g =>
|
|
`\n LEFT JOIN pf.dim_period ${g.alias}`
|
|
+ ` ON ${g.alias}.drange @> (${alias}."${g.dateCol}" + '{{date_offset}}'::interval)::date`
|
|
).join('');
|
|
}
|
|
|
|
function generateSQL(source, colMeta) {
|
|
const dims = colMeta
|
|
.filter(c => c.role === 'dimension')
|
|
.sort((a, b) => (a.opos || 0) - (b.opos || 0))
|
|
.map(c => c.cname);
|
|
|
|
// Every column of each measure/date role, in col_meta order. Loads carry all of
|
|
// them; the adjustment operations are single-measure (scale distributes one
|
|
// {{value_incr}}) and use only the first of each, below.
|
|
const byRole = role => colMeta
|
|
.filter(c => c.role === role)
|
|
.sort((a, b) => (a.opos || 0) - (b.opos || 0))
|
|
.map(c => c.cname);
|
|
|
|
const valueCols = byRole('value');
|
|
const unitsCols = byRole('units');
|
|
const dateCols = byRole('date');
|
|
|
|
const valueCol = valueCols[0];
|
|
const unitsCol = unitsCols[0];
|
|
|
|
if (!valueCol) throw new Error('No value column defined in col_meta');
|
|
if (!dateCols.length) throw new Error('No date column defined in col_meta');
|
|
if (dims.length === 0) throw new Error('No dimension columns defined in col_meta');
|
|
|
|
const srcTable = `"${source.schema}"."${source.tname}"`;
|
|
// Every date column, not just the primary one. Scale and recode used to
|
|
// carry dateCols[0] alone, so a source with order, requested and ship dates
|
|
// wrote adjustment rows whose rdate and sdate_e were *null* -- not stale,
|
|
// absent. Those columns are role 'date' and therefore sliceable, so a later
|
|
// adjustment scoped to a ship date matched no adjustment row ever written,
|
|
// and any reader that dated the forecast by ship date lost every
|
|
// adjustment silently. The derived period dimensions (rseas, smon_e) were
|
|
// fine throughout, being dimensions and copied with the rest -- which is
|
|
// what made it invisible: the labels were right and the dates underneath
|
|
// them were gone.
|
|
const dataCols = [...dims, ...dateCols, valueCol, unitsCol].filter(Boolean);
|
|
const effectiveValue = dataCols.includes(valueCol) ? valueCol : null;
|
|
const effectiveUnits = dataCols.includes(unitsCol) ? unitsCol : null;
|
|
const insertCols = [...dataCols.map(q), 'pf_iter', 'pf_logid', 'pf_user', 'pf_created_at'].join(', ');
|
|
const selectData = dataCols.map(q).join(', ');
|
|
const dimsJoined = dims.map(q).join(', ');
|
|
// The date columns as a select list, in the same order dataCols lists them,
|
|
// so it lines up with insertCols positionally.
|
|
const datesJoined = dateCols.map(q).join(', ');
|
|
|
|
// Baseline and reference copy the source row wholesale, so they carry every
|
|
// measure and every date — not just the primary one the operations act on.
|
|
// Dropping the others would leave those columns null for the life of the version.
|
|
// Clone carries every date column, not just the primary one, because it is the
|
|
// operation that moves rows through time: {{date_offset}} shifts them all
|
|
// together, and the period dimensions are re-derived from pf.dim_period against
|
|
// the shifted dates rather than copied from the row being cloned. Cloning last
|
|
// year's mix forward a year otherwise produces rows dated 2027 still labelled
|
|
// with 2026's periods.
|
|
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);
|
|
|
|
// dim_period JOIN support: a date column that is the is_key of a dim_group
|
|
// anchors that group, and dimension siblings with dim_period_col set are
|
|
// derived from pf.dim_period instead of copied raw. Derivation is against the
|
|
// date *after* {{date_offset}}, which is the whole point -- shift a baseline
|
|
// forward a year and its period columns follow, rather than still naming the
|
|
// year it came from.
|
|
//
|
|
// Every such group, not just the first. This used to be a find(), so a source
|
|
// with order, requested and ship date groups derived the order one and copied
|
|
// the other two raw -- shifted dates against unshifted period labels.
|
|
const dateGroups = dateGroupsOf(colMeta);
|
|
const dimPeriodMap = dimPeriodMapOf(dateGroups);
|
|
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(),
|
|
recode: buildRecode(),
|
|
clone: buildClone(),
|
|
undo: buildUndo()
|
|
};
|
|
|
|
function buildGetData() {
|
|
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() {
|
|
// pf_logid is part of the grain, so joining pf.log adds no rows — each group
|
|
// already belongs to exactly one log entry. Without this the segment labels
|
|
// that /data surfaces would vanish the moment a source declares a grain.
|
|
return `
|
|
SELECT
|
|
${grainSelect('t.')}
|
|
,${SEGMENT_EXPR} AS pf_segment
|
|
,${BUCKET_EXPR} AS pf_bucket
|
|
,${NOTE_EXPR} AS pf_note
|
|
,l.operation AS pf_op
|
|
FROM {{fc_table}} t
|
|
LEFT JOIN pf.log l
|
|
ON l.id = t.pf_logid${VERSION_JOIN}
|
|
WHERE {{territory_clause}}
|
|
GROUP BY
|
|
${grain.groupCols('t.').join('\n ,')}
|
|
,${LABEL_GROUP_COLS.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)
|
|
// The offset shifts every date column, so order date and ship date stay in step.
|
|
return loadCols.map(c => {
|
|
if (dateColSet.has(c)) return `(${pfx}${q(c)} + '{{date_offset}}'::interval)::date`;
|
|
if (dimPeriodMap.has(c)) {
|
|
const { alias, periodCol } = dimPeriodMap.get(c);
|
|
return `${alias}.${q(periodCol)} AS ${q(c)}`;
|
|
}
|
|
return `${pfx}${q(c)}`;
|
|
}).join(',\n ');
|
|
}
|
|
|
|
function buildFromClause() {
|
|
if (!hasDimPeriod) return srcTable;
|
|
return srcTable + ' s' + dimPeriodJoins(dateGroups);
|
|
}
|
|
|
|
function buildBaseline() {
|
|
return `
|
|
WITH
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note, label, bucket, tag)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'baseline', NULL, '{{params}}'::jsonb, '{{note}}',
|
|
NULLIF('{{label}}', ''), NULLIF('{{bucket}}', ''), NULLIF('{{tag}}', ''))
|
|
RETURNING id
|
|
)
|
|
,ins AS (
|
|
INSERT INTO {{fc_table}} (${loadInsertCols})
|
|
SELECT
|
|
${buildLoadSelect(hasDimPeriod ? 's.' : '')},
|
|
'baseline', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM ${buildFromClause()}
|
|
WHERE {{filter_clause}}
|
|
RETURNING *
|
|
)
|
|
SELECT count(*) AS rows_affected, (SELECT id FROM ilog) AS log_id FROM ins`.trim();
|
|
}
|
|
|
|
function buildReference() {
|
|
return `
|
|
WITH
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note, label, bucket, tag)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'reference', NULL, '{{params}}'::jsonb, '{{note}}',
|
|
NULLIF('{{label}}', ''), NULLIF('{{bucket}}', ''), NULLIF('{{tag}}', ''))
|
|
RETURNING id
|
|
)
|
|
,ins AS (
|
|
INSERT INTO {{fc_table}} (${loadInsertCols})
|
|
SELECT
|
|
${buildLoadSelect(hasDimPeriod ? 's.' : '')},
|
|
'reference', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM ${buildFromClause()}
|
|
WHERE {{filter_clause}}
|
|
RETURNING *
|
|
)
|
|
SELECT count(*) AS rows_affected, (SELECT id FROM ilog) AS log_id FROM ins`.trim();
|
|
}
|
|
|
|
function buildScale() {
|
|
const vSel = effectiveValue
|
|
? `round((${q(effectiveValue)} / NULLIF(total_value, 0)) * {{value_incr}}, 2)`
|
|
: `0`;
|
|
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] : [],
|
|
datesJoined,
|
|
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
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'scale', '{{slice}}'::jsonb, '{{params}}'::jsonb, '{{note}}')
|
|
RETURNING id
|
|
)
|
|
,base AS (
|
|
SELECT
|
|
${baseSelectParts}
|
|
FROM {{fc_table}}
|
|
WHERE {{where_clause}}
|
|
{{exclude_clause}}
|
|
GROUP BY
|
|
${groupCols([...dims, ...dateCols])}
|
|
)
|
|
,ins AS (
|
|
INSERT INTO {{fc_table}} (${insertCols})
|
|
SELECT
|
|
${[dimsJoined, datesJoined, ...(effectiveValue ? [vSel] : []), ...(effectiveUnits ? [uSel] : [])].join(',\n ')},
|
|
'scale', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM base
|
|
RETURNING *
|
|
)
|
|
${opTail('ins')}`.trim();
|
|
}
|
|
|
|
function buildRecode() {
|
|
return `
|
|
WITH
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'recode', '{{slice}}'::jsonb, '{{params}}'::jsonb, '{{note}}')
|
|
RETURNING id
|
|
)
|
|
,src AS (
|
|
SELECT
|
|
${dimsJoined},
|
|
${datesJoined}${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])}
|
|
)
|
|
,neg AS (
|
|
INSERT INTO {{fc_table}} (${insertCols})
|
|
SELECT ${dimsJoined}, ${datesJoined}, ${effectiveValue ? `-${q(effectiveValue)}` : '0'}${effectiveUnits ? `, -${q(effectiveUnits)}` : ''},
|
|
'recode', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM src
|
|
RETURNING *
|
|
)
|
|
,ins AS (
|
|
INSERT INTO {{fc_table}} (${insertCols})
|
|
SELECT {{set_clause}}, ${datesJoined}, ${effectiveValue ? q(effectiveValue) : '0'}${effectiveUnits ? `, ${q(effectiveUnits)}` : ''},
|
|
'recode', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
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();
|
|
}
|
|
|
|
function buildClone() {
|
|
const select = [
|
|
// dims: whatever {{set_clause}} resolves them to. The route builds it,
|
|
// and substitutes the dim_period expression for any derived dimension
|
|
// the caller has not overridden outright.
|
|
'{{set_clause}}',
|
|
...dateCols.map(c => `(s.${q(c)} + '{{date_offset}}'::interval)::date`),
|
|
effectiveValue ? `round(s.${q(effectiveValue)} * {{scale_factor}}, 2)` : null,
|
|
effectiveUnits ? `round(s.${q(effectiveUnits)} * {{scale_factor}}, 5)` : null,
|
|
].filter(Boolean).join(',\n ');
|
|
|
|
return `
|
|
WITH
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'clone', '{{slice}}'::jsonb, '{{params}}'::jsonb, '{{note}}')
|
|
RETURNING id
|
|
)
|
|
,ins AS (
|
|
INSERT INTO {{fc_table}} (${cloneInsertCols})
|
|
SELECT
|
|
${select},
|
|
'clone', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
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();
|
|
}
|
|
|
|
function buildUndo() {
|
|
// undo is executed as two separate queries in the route handler
|
|
// (delete from fc_table first, then delete from pf.log) to avoid
|
|
// FK constraint ordering issues within a single CTE statement.
|
|
// This entry is a placeholder — the undo route uses it as a template reference.
|
|
return `
|
|
-- step 1 (run first):
|
|
DELETE FROM {{fc_table}} WHERE pf_logid = {{logid}};
|
|
-- step 2 (run after step 1):
|
|
DELETE FROM pf.log WHERE id = {{logid}};`.trim();
|
|
}
|
|
}
|
|
|
|
// substitute {{token}} placeholders in a SQL string
|
|
function applyTokens(sql, tokens) {
|
|
let result = sql;
|
|
for (const [key, value] of Object.entries(tokens)) {
|
|
result = result.replace(new RegExp(`\\{\\{${key}\\}\\}`, 'g'), value ?? '');
|
|
}
|
|
return result;
|
|
}
|
|
|
|
// build a SQL WHERE clause string from a slice object
|
|
// only dimension columns are included; unrecognised keys are silently skipped
|
|
// pf_segment and pf_bucket are not columns on the forecast table -- they are
|
|
// computed at read time from the row's pf.log entry -- so a slice naming one
|
|
// cannot be compared directly. It resolves to a set of log ids instead, which is
|
|
// exact: the name lives on the log row, and every forecast row carries the
|
|
// pf_logid that points at it.
|
|
//
|
|
// Without this they were dropped from the slice, and clicking a single bucket's
|
|
// cell scaled every bucket at that dimension intersection while the panel showed
|
|
// only the one clicked.
|
|
const COMPUTED_SLICE_COLS = { pf_segment: SEGMENT_EXPR, pf_bucket: BUCKET_EXPR };
|
|
|
|
function computedSlicePredicate(col, val, versionId) {
|
|
if (versionId == null) {
|
|
const err = new Error(`Cannot filter on ${col} without a version`);
|
|
err.status = 500;
|
|
throw err;
|
|
}
|
|
const vals = (Array.isArray(val) ? val : [val]).map(v => `'${esc(v)}'`).join(', ');
|
|
return `pf_logid IN (
|
|
SELECT l.id
|
|
FROM pf.log l${VERSION_JOIN}
|
|
WHERE l.version_id = ${parseInt(versionId)}
|
|
AND ${COMPUTED_SLICE_COLS[col]} IN (${vals})
|
|
)`;
|
|
}
|
|
|
|
function buildWhere(slice, dimCols, versionId) {
|
|
if (!slice || Object.keys(slice).length === 0) return 'TRUE';
|
|
|
|
const allowed = new Set(dimCols);
|
|
const parts = [];
|
|
|
|
for (const [col, val] of Object.entries(slice)) {
|
|
if (COMPUTED_SLICE_COLS[col]) {
|
|
parts.push(computedSlicePredicate(col, val, versionId));
|
|
continue;
|
|
}
|
|
// A pf_ key this does not understand is refused rather than skipped.
|
|
// Skipping is how a selection silently widened: the operation ran against
|
|
// everything the dropped key would have excluded. pf_iter is the one
|
|
// exception -- the client strips it deliberately, since two cells that
|
|
// differ only by iter band are the same slice.
|
|
if (col.startsWith('pf_') && col !== 'pf_iter') {
|
|
const err = new Error(`Slice names ${col}, which cannot be filtered on`);
|
|
err.status = 400;
|
|
throw err;
|
|
}
|
|
if (!allowed.has(col)) continue;
|
|
if (Array.isArray(val)) {
|
|
const escaped = val.map(v => esc(v));
|
|
parts.push(`"${col}" IN ('${escaped.join("', '")}')`);
|
|
} else {
|
|
parts.push(`"${col}" = '${esc(val)}'`);
|
|
}
|
|
}
|
|
|
|
return parts.length ? parts.join('\nAND ') : 'TRUE';
|
|
}
|
|
|
|
// The pivot's own filter, carried alongside the slices so an operation writes
|
|
// exactly the rows the ledger counted.
|
|
//
|
|
// A slice is {col: value} and can only ever mean equality, so a view filtered to
|
|
// sseas_e <= 2027 could not be expressed as one. Refusing it was safe but
|
|
// useless -- a bounded season is an ordinary way to scope a forecast -- so the
|
|
// operators travel as [col, op, value] triples instead.
|
|
//
|
|
// Perspective's operator names, not SQL's, since that is where these come from.
|
|
// Anything outside this list is refused rather than ignored: a scope silently
|
|
// dropped is a write that is wider than the panel that authorised it.
|
|
const SCOPE_OPS = {
|
|
'==': (c, v) => `${c} = ${v[0]}`,
|
|
'!=': (c, v) => `${c} != ${v[0]}`,
|
|
'>': (c, v) => `${c} > ${v[0]}`,
|
|
'>=': (c, v) => `${c} >= ${v[0]}`,
|
|
'<': (c, v) => `${c} < ${v[0]}`,
|
|
'<=': (c, v) => `${c} <= ${v[0]}`,
|
|
'in': (c, v) => `${c} IN (${v.join(', ')})`,
|
|
'not in': (c, v) => `${c} NOT IN (${v.join(', ')})`,
|
|
'is null': (c) => `${c} IS NULL`,
|
|
'is not null': (c) => `${c} IS NOT NULL`,
|
|
};
|
|
|
|
function buildScopeClause(scope, dimCols, versionId) {
|
|
if (!Array.isArray(scope) || scope.length === 0) return '';
|
|
const allowed = new Set(dimCols);
|
|
const parts = scope.map((entry) => {
|
|
if (!Array.isArray(entry) || entry.length < 2) {
|
|
const err = new Error(`Malformed scope entry ${JSON.stringify(entry)}`);
|
|
err.status = 400; throw err;
|
|
}
|
|
const [col, op, ...rest] = entry;
|
|
const vals = (Array.isArray(rest[0]) ? rest[0] : rest).filter(v => v !== undefined);
|
|
|
|
if (COMPUTED_SLICE_COLS[col]) {
|
|
if (op !== '==' && op !== 'in') {
|
|
const err = new Error(`${col} can only be scoped with == or in, not ${op}`);
|
|
err.status = 400; throw err;
|
|
}
|
|
return computedSlicePredicate(col, vals, versionId);
|
|
}
|
|
if (!allowed.has(col)) {
|
|
const err = new Error(`Column "${col}" is not available for filtering`);
|
|
err.status = 400; throw err;
|
|
}
|
|
const fn = SCOPE_OPS[op];
|
|
if (!fn) {
|
|
const err = new Error(`Unsupported filter operator "${op}" on ${col}`);
|
|
err.status = 400; throw err;
|
|
}
|
|
return fn(`"${col}"`, vals.map(v => `'${esc(String(v))}'`));
|
|
});
|
|
return parts.join('\nAND ');
|
|
}
|
|
|
|
// The territory predicate: what this account may see and change.
|
|
//
|
|
// Server-side by construction. Today's `scope` is sent by the browser, which is
|
|
// right for a filter the user chose and would be fatal for a permission -- so
|
|
// this one is built from the session and ANDed on last, where nothing in the
|
|
// request can remove it.
|
|
//
|
|
// FALSE, not TRUE, for an account with no territory. The whole point is that a
|
|
// missing grant means no rows: an empty list that fell through to TRUE would
|
|
// hand the entire book to the first account somebody forgot to configure.
|
|
function buildTerritoryClause(territory, territoryCol, alias = '') {
|
|
if (!territory || territory.admin) return '';
|
|
if (!territoryCol) return 'FALSE';
|
|
const vals = (territory.values || []).filter(v => v != null && v !== '');
|
|
if (vals.length === 0) return 'FALSE';
|
|
const pfx = alias ? `${alias}.` : '';
|
|
return `${pfx}"${territoryCol}" IN (${vals.map(v => `'${esc(String(v))}'`).join(', ')})`;
|
|
}
|
|
|
|
// build a WHERE clause spanning several slices — an OR of AND-groups.
|
|
// A union of slices cannot be flattened into one IN list per column: slices
|
|
// {Region:East, State:NY} and {Region:West, State:CA} would become
|
|
// Region IN (East,West) AND State IN (NY,CA), which also matches East/CA.
|
|
function buildWhereAny(slices, dimCols, versionId) {
|
|
const list = (slices || []).filter(s => s && Object.keys(s).length > 0);
|
|
if (list.length === 0) return 'TRUE';
|
|
if (list.length === 1) return buildWhere(list[0], dimCols, versionId);
|
|
|
|
const groups = list
|
|
.map(s => buildWhere(s, dimCols, versionId))
|
|
.filter(w => w !== 'TRUE');
|
|
|
|
// any slice that reduced to TRUE selects everything, so the union does too
|
|
if (groups.length !== list.length) return 'TRUE';
|
|
|
|
// outer parens matter: the caller appends `AND pf_iter NOT IN (...)`,
|
|
// and AND binds tighter than OR
|
|
return `(${groups.map(g => `(${g.replace(/\n/g, ' ')})`).join('\n OR ')})`;
|
|
}
|
|
|
|
// the bare predicate for "this row participates in operations", for use in a
|
|
// FILTER clause where the excluded rows still need to be counted separately
|
|
function buildExcludePredicate(excludeIters) {
|
|
if (!excludeIters || excludeIters.length === 0) return 'TRUE';
|
|
const list = excludeIters.map(i => `'${esc(i)}'`).join(', ');
|
|
return `pf_iter NOT IN (${list})`;
|
|
}
|
|
|
|
// build AND iter NOT IN (...) from a version's exclude_iters array
|
|
function buildExcludeClause(excludeIters) {
|
|
if (!excludeIters || excludeIters.length === 0) return '';
|
|
const list = excludeIters.map(i => `'${esc(i)}'`).join(', ');
|
|
return `AND pf_iter NOT IN (${list})`;
|
|
}
|
|
|
|
// build the dimension columns portion of a SELECT for recode/clone
|
|
// replaces named dimensions with literal values, passes others through unchanged
|
|
// derivedExprs: cname -> a SQL expression to use when the caller has not set the
|
|
// column outright. Clone passes the dim_period expressions through here, so a
|
|
// cloned row's period dimensions come from the calendar against its shifted date
|
|
// rather than from the row it was copied from.
|
|
function buildSetClause(dimCols, setObj, opts = {}) {
|
|
const { derivedExprs, alias } = opts;
|
|
const pfx = alias ? `${alias}.` : '';
|
|
return dimCols.map(col => {
|
|
if (setObj && setObj[col] !== undefined) {
|
|
return `'${esc(setObj[col])}' AS "${col}"`;
|
|
}
|
|
if (derivedExprs && derivedExprs[col]) {
|
|
return `${derivedExprs[col]} AS "${col}"`;
|
|
}
|
|
return `${pfx}"${col}"`;
|
|
}).join(', ');
|
|
}
|
|
|
|
// build a SQL WHERE clause from an array of filter objects { col, op, values }
|
|
// only allows columns with role 'date' or 'filter'
|
|
function buildFilterClause(filters, colMeta) {
|
|
if (!filters || filters.length === 0) {
|
|
const err = new Error('At least one filter is required');
|
|
err.status = 400; throw err;
|
|
}
|
|
const allowed = new Set(
|
|
colMeta.filter(c => c.role !== 'ignore').map(c => c.cname)
|
|
);
|
|
const parts = filters.map(({ col, op, values = [] }) => {
|
|
if (!allowed.has(col)) {
|
|
const err = new Error(`Column "${col}" is not available for baseline filtering`);
|
|
err.status = 400; throw err;
|
|
}
|
|
const c = `"${col}"`;
|
|
const v = values.map(x => `'${esc(String(x))}'`);
|
|
switch (op) {
|
|
case '=': return `${c} = ${v[0]}`;
|
|
case '!=': return `${c} != ${v[0]}`;
|
|
case 'IN': return `${c} IN (${v.join(', ')})`;
|
|
case 'NOT IN': return `${c} NOT IN (${v.join(', ')})`;
|
|
case 'BETWEEN': return `${c} BETWEEN ${v[0]} AND ${v[1]}`;
|
|
case 'IS NULL': return `${c} IS NULL`;
|
|
case 'IS NOT NULL': return `${c} IS NOT NULL`;
|
|
default: {
|
|
const err = new Error(`Unsupported operator "${op}"`);
|
|
err.status = 400; throw err;
|
|
}
|
|
}
|
|
});
|
|
return parts.join('\nAND ');
|
|
}
|
|
|
|
// escape a value for safe SQL string substitution
|
|
function esc(val) {
|
|
if (val === null || val === undefined) return '';
|
|
return String(val).replace(/'/g, "''");
|
|
}
|
|
|
|
module.exports = { generateSQL, grainOf, COMPUTED_SLICE_COLS, buildScopeClause, buildTerritoryClause,
|
|
SEGMENT_EXPR, BUCKET_EXPR, NOTE_EXPR, LABEL_GROUP_COLS, VERSION_JOIN,
|
|
ADJUSTMENT_SEGMENT, ADJUSTMENT_BUCKET, UNLABELED_LOAD, dateGroupsOf, dimPeriodMapOf, dimPeriodJoins, applyTokens, buildWhere, buildWhereAny, buildExcludeClause, buildExcludePredicate, buildSetClause, buildFilterClause, esc };
|