Paul Trowbridge
0e87f22e24
The generator picked columns with find(), so a source with more than one
date or value column silently loaded only the first of each. The version's
DDL is built from col_meta separately, so the extra columns existed in the
forecast table and stayed null for its whole life -- gs.osm_skinny lost
sdate and stdcost_usd across 295k rows that way.
Loads copy the source row wholesale, so they now carry all of them, with
the offset shifting every date column together. The adjustment operations
are genuinely single-measure -- scale distributes one {{value_incr}} -- so
they keep the narrow column list until that has a defined meaning.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
362 lines
14 KiB
JavaScript
362 lines
14 KiB
JavaScript
// Generates operation SQL for a source table, baking in column names from col_meta.
|
|
// Runtime values are left as {{token}} substitution points.
|
|
//
|
|
// 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}},
|
|
// {{params}}, {{slice}}, {{date_from}}, {{date_to}},
|
|
// {{value_incr}}, {{units_incr}}, {{set_clause}}, {{scale_factor}}
|
|
|
|
// wrap a column name in double quotes for safe use in SQL
|
|
function q(name) { return `"${name}"`; }
|
|
|
|
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];
|
|
const dateCol = dateCols[0];
|
|
|
|
if (!valueCol) throw new Error('No value column defined in col_meta');
|
|
if (!dateCol) 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}"`;
|
|
const dataCols = [...dims, dateCol, 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(', ');
|
|
|
|
// 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.
|
|
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: if the date column is the is_key of a dim_group,
|
|
// dimension siblings with dim_period_col set are derived from pf.dim_period
|
|
// instead of being copied raw from the source on baseline/reference load.
|
|
const dateKeyGroup = colMeta.find(c => c.role === 'date' && c.is_key && c.dim_group)?.dim_group;
|
|
const dimPeriodMap = new Map(
|
|
dateKeyGroup
|
|
? colMeta
|
|
.filter(c => c.role === 'dimension' && c.dim_group === dateKeyGroup && c.dim_period_col)
|
|
.map(c => [c.cname, c.dim_period_col])
|
|
: []
|
|
);
|
|
const hasDimPeriod = dimPeriodMap.size > 0;
|
|
|
|
return {
|
|
get_data: buildGetData(),
|
|
baseline: buildBaseline(),
|
|
reference: buildReference(),
|
|
scale: buildScale(),
|
|
recode: buildRecode(),
|
|
clone: buildClone(),
|
|
undo: buildUndo()
|
|
};
|
|
|
|
function buildGetData() {
|
|
return `SELECT * FROM {{fc_table}}`;
|
|
}
|
|
|
|
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)) return `dp.${q(dimPeriodMap.get(c))} AS ${q(c)}`;
|
|
return `${pfx}${q(c)}`;
|
|
}).join(',\n ');
|
|
}
|
|
|
|
function buildFromClause() {
|
|
if (!hasDimPeriod) return srcTable;
|
|
return `${srcTable} s\n JOIN pf.dim_period dp`
|
|
+ ` ON dp.drange @> (s.${q(dateCol)} + '{{date_offset}}'::interval)::date`;
|
|
}
|
|
|
|
function buildBaseline() {
|
|
return `
|
|
WITH
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'baseline', NULL, '{{params}}'::jsonb, '{{note}}')
|
|
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 FROM ins`.trim();
|
|
}
|
|
|
|
function buildReference() {
|
|
return `
|
|
WITH
|
|
ilog AS (
|
|
INSERT INTO pf.log (version_id, pf_user, operation, slice, params, note)
|
|
VALUES ({{version_id}}, '{{pf_user}}', 'reference', NULL, '{{params}}'::jsonb, '{{note}}')
|
|
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 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`;
|
|
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
|
|
].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}}
|
|
)
|
|
,ins AS (
|
|
INSERT INTO {{fc_table}} (${insertCols})
|
|
SELECT
|
|
${[dimsJoined, q(dateCol), ...(effectiveValue ? [vSel] : []), ...(effectiveUnits ? [uSel] : [])].join(',\n ')},
|
|
'scale', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM base
|
|
RETURNING *
|
|
)
|
|
SELECT * FROM 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 ${selectData}
|
|
FROM {{fc_table}}
|
|
WHERE {{where_clause}}
|
|
{{exclude_clause}}
|
|
)
|
|
,neg AS (
|
|
INSERT INTO {{fc_table}} (${insertCols})
|
|
SELECT ${dimsJoined}, ${q(dateCol)}, ${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}}, ${q(dateCol)}, ${effectiveValue ? q(effectiveValue) : '0'}${effectiveUnits ? `, ${q(effectiveUnits)}` : ''},
|
|
'recode', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM src
|
|
RETURNING *
|
|
)
|
|
SELECT * FROM neg UNION ALL SELECT * FROM ins`.trim();
|
|
}
|
|
|
|
function buildClone() {
|
|
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}} (${insertCols})
|
|
SELECT
|
|
{{set_clause}},
|
|
${q(dateCol)},
|
|
${effectiveValue ? `round(${q(effectiveValue)} * {{scale_factor}}, 2)` : '0'}${effectiveUnits ? `,\n round(${q(effectiveUnits)} * {{scale_factor}}, 5)` : ''},
|
|
'clone', (SELECT id FROM ilog), '{{pf_user}}', now()
|
|
FROM {{fc_table}}
|
|
WHERE {{where_clause}}
|
|
{{exclude_clause}}
|
|
RETURNING *
|
|
)
|
|
SELECT * FROM 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
|
|
function buildWhere(slice, dimCols) {
|
|
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 (!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';
|
|
}
|
|
|
|
// 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) {
|
|
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);
|
|
|
|
const groups = list
|
|
.map(s => buildWhere(s, dimCols))
|
|
.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
|
|
function buildSetClause(dimCols, setObj) {
|
|
return dimCols.map(col => {
|
|
if (setObj && setObj[col] !== undefined) {
|
|
return `'${esc(setObj[col])}' AS "${col}"`;
|
|
}
|
|
return `"${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, applyTokens, buildWhere, buildWhereAny, buildExcludeClause, buildExcludePredicate, buildSetClause, buildFilterClause, esc };
|