Keep master data for a dim_group instead of re-deriving it from the source
Every question about a part -- what values exist, what attributes go with one -- was answered by querying the source, and the source is the wrong place to ask. It is a view over a transaction table, so the query is slow (76s for one ILIKE against 6.9M rows), it describes only what was transacted, and it cannot express intent: there is no way to say a part is discontinued, or to name one that has not sold yet. pf.dim_member holds the app's own list: one row per key value per group, siblings in jsonb, keyed on (source_id, dim_group, key_value). Refresh is a merge rather than a replace, so curation survives it -- members absent from the source are marked source_seen = false, not deleted. Triggered from Setup, next to Generate SQL, because it reads the whole source and the answer only changes when the catalogue does. A key can carry several attribute sets across history -- 11,290 parts against 13,662 combinations on osm_skinny -- so the refresh takes the most recent by the source's date column. That also fixes the sibling autofill, which used to run a DISTINCT ... LIMIT 2 against the source and silently fill nothing whenever a part came back ambiguous. A member row is one definition by construction. The client fetches each group's list once per source and does both completion and autofill against it in memory, so neither costs a request. Columns outside a group, or a group never refreshed, still fall back to the version's values endpoint. Run 01_schema.sql to create the table. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
401b6b9b42
commit
07d92ccd74
@ -253,6 +253,142 @@ module.exports = function(pool) {
|
||||
}
|
||||
});
|
||||
|
||||
// Resolve a dim_group to its key column and siblings, or explain why it cannot be.
|
||||
async function resolveGroup(sourceId, group) {
|
||||
const { rows: meta } = await pool.query(
|
||||
`SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`, [sourceId]);
|
||||
const members = meta.filter(c => c.dim_group === group);
|
||||
if (!members.length) {
|
||||
const err = new Error(`No columns are grouped as "${group}" on this source`);
|
||||
err.status = 404; throw err;
|
||||
}
|
||||
const keyCol = members.find(c => c.is_key);
|
||||
if (!keyCol) {
|
||||
const err = new Error(
|
||||
`Group "${group}" has no is_key column, so its members have nothing to be keyed on`);
|
||||
err.status = 400; throw err;
|
||||
}
|
||||
return {
|
||||
keyCol,
|
||||
siblings: members.filter(c => c.cname !== keyCol.cname),
|
||||
// recency column: the source's primary date, the same one the generator
|
||||
// treats as the date for loads
|
||||
dateCol: meta.find(c => c.role === 'date')?.cname || null,
|
||||
};
|
||||
}
|
||||
|
||||
// The member list for a group, as one array. Small enough to send whole --
|
||||
// 11,290 parts on osm_skinny -- so the client holds it and filters locally
|
||||
// instead of querying per keystroke.
|
||||
router.get('/sources/:id/dim/:group', async (req, res) => {
|
||||
try {
|
||||
const sourceId = parseInt(req.params.id);
|
||||
const { keyCol, siblings } = await resolveGroup(sourceId, req.params.group);
|
||||
const includeInactive = req.query.all === '1';
|
||||
|
||||
const { rows } = await pool.query(`
|
||||
SELECT key_value, attrs, is_active, source_seen
|
||||
FROM pf.dim_member
|
||||
WHERE source_id = $1 AND dim_group = $2
|
||||
${includeInactive ? '' : 'AND is_active'}
|
||||
ORDER BY key_value
|
||||
`, [sourceId, req.params.group]);
|
||||
|
||||
res.json({
|
||||
group: req.params.group,
|
||||
key_col: keyCol.cname,
|
||||
siblings: siblings.map(c => c.cname),
|
||||
members: rows,
|
||||
});
|
||||
} catch (err) {
|
||||
console.error(err);
|
||||
res.status(err.status || 500).json({ error: err.message });
|
||||
}
|
||||
});
|
||||
|
||||
// Rebuild a group's members from the source. A merge, not a replace: curation
|
||||
// (a member deactivated by hand, or added before it ever sold) has to survive a
|
||||
// refresh, so absent members are marked source_seen = false rather than deleted.
|
||||
//
|
||||
// Slow by nature -- it reads the whole source, which for a view over a
|
||||
// transaction table is millions of rows -- so it is a deliberate action rather
|
||||
// than something that happens on a page load.
|
||||
router.post('/sources/:id/dim/:group/refresh', async (req, res) => {
|
||||
const sourceId = parseInt(req.params.id);
|
||||
const group = req.params.group;
|
||||
try {
|
||||
const srcResult = await pool.query(
|
||||
`SELECT schema, tname FROM pf.source WHERE id = $1`, [sourceId]);
|
||||
if (!srcResult.rows.length) return res.status(404).json({ error: 'Source not found' });
|
||||
const { schema, tname } = srcResult.rows[0];
|
||||
|
||||
const { keyCol, siblings, dateCol } = await resolveGroup(sourceId, group);
|
||||
if (!siblings.length) {
|
||||
return res.status(400).json({ error: `Group "${group}" has no sibling columns to store` });
|
||||
}
|
||||
|
||||
const q = (n) => `"${n}"`;
|
||||
const attrs = siblings.map(c => `'${c.cname}', s.${q(c.cname)}::text`).join(', ');
|
||||
// A key can carry more than one attribute set across history -- 11,290
|
||||
// parts against 13,662 combinations on osm_skinny. Take the most recent
|
||||
// by the source's date column, which is the live definition.
|
||||
const recency = dateCol ? `s.${q(dateCol)} DESC NULLS LAST` : `1`;
|
||||
|
||||
const started = Date.now();
|
||||
// An explicit stamp rather than now(): inside a transaction now() is the
|
||||
// transaction's start time, so "refreshed in this run" and "refreshed in
|
||||
// a run that began at the same instant" would be indistinguishable.
|
||||
const runAt = new Date();
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
const { rows: [{ n }] } = await client.query(`
|
||||
WITH ranked AS (
|
||||
SELECT s.${q(keyCol.cname)}::text AS key_value,
|
||||
jsonb_build_object(${attrs}) AS attrs,
|
||||
row_number() OVER (
|
||||
PARTITION BY s.${q(keyCol.cname)} ORDER BY ${recency}
|
||||
) AS rn
|
||||
FROM ${q(schema)}.${q(tname)} s
|
||||
WHERE s.${q(keyCol.cname)} IS NOT NULL
|
||||
)
|
||||
,upserted AS (
|
||||
INSERT INTO pf.dim_member
|
||||
(source_id, dim_group, key_value, attrs, refreshed_at, source_seen)
|
||||
SELECT $1, $2, key_value, attrs, $3, true
|
||||
FROM ranked WHERE rn = 1
|
||||
ON CONFLICT (source_id, dim_group, key_value) DO UPDATE SET
|
||||
attrs = EXCLUDED.attrs,
|
||||
source_seen = true,
|
||||
refreshed_at = $3,
|
||||
updated_at = now()
|
||||
RETURNING 1
|
||||
)
|
||||
SELECT count(*)::int AS n FROM upserted
|
||||
`, [sourceId, group, runAt]);
|
||||
|
||||
// anything this run did not touch is no longer in the source
|
||||
const { rowCount: dropped } = await client.query(`
|
||||
UPDATE pf.dim_member
|
||||
SET source_seen = false, updated_at = now()
|
||||
WHERE source_id = $1 AND dim_group = $2 AND source_seen
|
||||
AND refreshed_at IS DISTINCT FROM $3
|
||||
`, [sourceId, group, runAt]);
|
||||
|
||||
await client.query('COMMIT');
|
||||
res.json({ group, members: n, no_longer_in_source: dropped, ms: Date.now() - started });
|
||||
} catch (err) {
|
||||
await client.query('ROLLBACK');
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(err);
|
||||
res.status(err.status || 500).json({ error: err.message });
|
||||
}
|
||||
});
|
||||
|
||||
// given a key column value, look up sibling dim_group column values from source
|
||||
// returns { sibling_col: value, ... } if exactly one match, null if none or ambiguous
|
||||
router.get('/sources/:id/lookup', async (req, res) => {
|
||||
|
||||
@ -81,6 +81,30 @@ WHERE TRUE
|
||||
AND note <> ''
|
||||
AND operation IN ('baseline', 'reference');
|
||||
|
||||
-- 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
|
||||
-- member list from it is both slow and wrong -- slow because it means scanning
|
||||
-- millions of rows, wrong because it can only describe what was transacted and
|
||||
-- has no way to say a part is discontinued or that a new one exists before it
|
||||
-- has sold. This table is the app's own list, refreshed from the source but
|
||||
-- curatable independently of it.
|
||||
CREATE TABLE IF NOT EXISTS pf.dim_member (
|
||||
source_id integer NOT NULL REFERENCES pf.source(id) ON DELETE CASCADE,
|
||||
dim_group text NOT NULL, -- matches pf.col_meta.dim_group
|
||||
key_value text NOT NULL, -- the is_key column's value
|
||||
attrs jsonb NOT NULL DEFAULT '{}'::jsonb, -- the sibling columns
|
||||
is_active boolean NOT NULL DEFAULT true,
|
||||
source_seen boolean NOT NULL DEFAULT true, -- present in the source at last refresh
|
||||
added_at timestamptz NOT NULL DEFAULT now(),
|
||||
updated_at timestamptz NOT NULL DEFAULT now(),
|
||||
refreshed_at timestamptz,
|
||||
PRIMARY KEY (source_id, dim_group, key_value)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dim_member_active_idx
|
||||
ON pf.dim_member (source_id, dim_group) WHERE is_active;
|
||||
|
||||
-- generated operation SQL per source, stored after col_meta is configured
|
||||
CREATE TABLE IF NOT EXISTS pf.sql (
|
||||
id serial PRIMARY KEY,
|
||||
|
||||
@ -541,24 +541,37 @@ function ScaleLedger({ currentTotals, scaleInputs, setScaleInputs, scalePlug, se
|
||||
// Completion for a key dimension. The list is fetched as you type rather than up
|
||||
// front: part alone has 11,290 distinct values, and a native datalist given all of
|
||||
// them is slow to open and no easier to read than a short filtered one.
|
||||
function DimValueInput({ col, versionId, value, onChange, onBlur, className }) {
|
||||
const [options, setOptions] = useState([])
|
||||
function DimValueInput({ col, members, versionId, value, onChange, onBlur, className }) {
|
||||
const [fallback, setFallback] = useState([])
|
||||
const listId = `pf-vals-${col.cname}`
|
||||
const hasList = !!members?.length
|
||||
|
||||
// Without a member list for this group -- never refreshed, or the column is not
|
||||
// in one -- fall back to the version's own values. That is a 2s scan held in
|
||||
// memory server-side, so it stays debounced rather than firing per keystroke.
|
||||
useEffect(() => {
|
||||
if (!versionId || !col.is_key) return
|
||||
if (hasList || !versionId || !col.is_key) return
|
||||
let cancelled = false
|
||||
// wait for a pause in typing; every keystroke would otherwise be a request
|
||||
const t = setTimeout(async () => {
|
||||
try {
|
||||
const url = `/api/versions/${versionId}/values/${encodeURIComponent(col.cname)}`
|
||||
+ `?limit=50${value ? `&q=${encodeURIComponent(value)}` : ''}`
|
||||
const rows = await fetch(url).then(r => r.ok ? r.json() : [])
|
||||
if (!cancelled) setOptions(Array.isArray(rows) ? rows : [])
|
||||
} catch { if (!cancelled) setOptions([]) }
|
||||
if (!cancelled) setFallback(Array.isArray(rows) ? rows : [])
|
||||
} catch { if (!cancelled) setFallback([]) }
|
||||
}, 200)
|
||||
return () => { cancelled = true; clearTimeout(t) }
|
||||
}, [versionId, col.cname, col.is_key, value])
|
||||
}, [hasList, versionId, col.cname, col.is_key, value])
|
||||
|
||||
// The member list is already in memory, so filtering it costs nothing and needs
|
||||
// no debounce -- the options move with the keystroke.
|
||||
const options = hasList
|
||||
? (() => {
|
||||
const q = (value || '').trim().toLowerCase()
|
||||
const all = members.map(m => m.key_value)
|
||||
return (q ? all.filter(v => v.toLowerCase().includes(q)) : all).slice(0, 50)
|
||||
})()
|
||||
: fallback
|
||||
|
||||
return (
|
||||
<>
|
||||
@ -580,7 +593,7 @@ function DimValueInput({ col, versionId, value, onChange, onBlur, className }) {
|
||||
|
||||
// ── 2b. Recode / clone form ─────────────────────────────────────────────────
|
||||
// Same pairing: the dimension's current value sits beside the box that replaces it.
|
||||
function DimForm({ dimCols, setObj, setSet, slices, lookupDerivedCols, versionId, extra }) {
|
||||
function DimForm({ dimCols, setObj, setSet, slices, lookupDerivedCols, dimMembers, versionId, extra }) {
|
||||
const multi = slices.length > 1
|
||||
const first = slices[0] || {}
|
||||
return (
|
||||
@ -605,6 +618,7 @@ function DimForm({ dimCols, setObj, setSet, slices, lookupDerivedCols, versionId
|
||||
<td className="pl-2 py-0.5">
|
||||
<DimValueInput
|
||||
col={c}
|
||||
members={c.dim_group ? dimMembers?.[c.dim_group]?.members : null}
|
||||
versionId={versionId}
|
||||
value={setObj[c.cname] || ''}
|
||||
onChange={e => setSet(s => ({ ...s, [c.cname]: e.target.value }))}
|
||||
@ -702,7 +716,7 @@ export default function OperationPanel({
|
||||
cloneSet, setCloneSet,
|
||||
cloneScale, setCloneScale,
|
||||
cloneNote, setCloneNote,
|
||||
dimCols, lookupDerivedCols, versionId,
|
||||
dimCols, lookupDerivedCols, dimMembers, versionId,
|
||||
buildPayload, submitOp,
|
||||
}) {
|
||||
const hasSlice = slices.length > 0
|
||||
@ -763,12 +777,12 @@ export default function OperationPanel({
|
||||
)}
|
||||
{activeOp === 'recode' && (
|
||||
<DimForm dimCols={dimCols} setObj={recodeSet} setSet={setRecodeSet}
|
||||
slices={slices} lookupDerivedCols={lookupDerivedCols} versionId={versionId}
|
||||
slices={slices} lookupDerivedCols={lookupDerivedCols} dimMembers={dimMembers} versionId={versionId}
|
||||
extra={<MovingTotal currentTotals={currentTotals} verb="Moving" />} />
|
||||
)}
|
||||
{activeOp === 'clone' && (
|
||||
<DimForm dimCols={dimCols} setObj={cloneSet} setSet={setCloneSet}
|
||||
slices={slices} lookupDerivedCols={lookupDerivedCols} versionId={versionId}
|
||||
slices={slices} lookupDerivedCols={lookupDerivedCols} dimMembers={dimMembers} versionId={versionId}
|
||||
extra={
|
||||
<div className="flex flex-col gap-2">
|
||||
<div className="flex items-center gap-2">
|
||||
|
||||
@ -66,6 +66,8 @@ export default function Forecast({ sources = [], sourceId, versions = [], versio
|
||||
const [scalePlug, setScalePlug] = useState(() => localStorage.getItem('pf_scale_plug') || 'price')
|
||||
// which change-log row has its payload open, if any
|
||||
const [expandedLog, setExpandedLog] = useState(null)
|
||||
// master data per dim_group, fetched once per source: { part: { key_col, siblings, members } }
|
||||
const [dimMembers, setDimMembers] = useState({})
|
||||
// what a target/percentage is measured against: the rows this operation can
|
||||
// write, or everything the pivot shows for the slice (excluded rows included)
|
||||
const [targetBasis, setTargetBasis] = useState('selected')
|
||||
@ -660,6 +662,27 @@ export default function Forecast({ sources = [], sourceId, versions = [], versio
|
||||
} catch { setKnownTags([]) }
|
||||
}
|
||||
|
||||
// One request per dim_group that has an is_key. The lists are small enough to
|
||||
// hold whole -- 11,290 parts -- which is what lets completion and the sibling
|
||||
// autofill run locally instead of querying the source per keystroke.
|
||||
useEffect(() => {
|
||||
if (!sourceId) { setDimMembers({}); return }
|
||||
let cancelled = false
|
||||
;(async () => {
|
||||
const meta = colMetaRef.current
|
||||
const groups = [...new Set(meta.filter(c => c.dim_group && c.is_key).map(c => c.dim_group))]
|
||||
const loaded = {}
|
||||
for (const g of groups) {
|
||||
try {
|
||||
const r = await fetch(`/api/sources/${sourceId}/dim/${encodeURIComponent(g)}`)
|
||||
if (r.ok) loaded[g] = await r.json()
|
||||
} catch { /* a group with no members yet just falls back */ }
|
||||
}
|
||||
if (!cancelled) setDimMembers(loaded)
|
||||
})()
|
||||
return () => { cancelled = true }
|
||||
}, [sourceId, versionId])
|
||||
|
||||
useEffect(() => { refreshLogMeta(versionId) }, [versionId])
|
||||
useEffect(() => { refreshTags(sourceId) }, [sourceId])
|
||||
|
||||
@ -1054,9 +1077,22 @@ export default function Forecast({ sources = [], sourceId, versions = [], versio
|
||||
|
||||
async function lookupDerivedCols(col, value, setter) {
|
||||
if (!sourceId || !value.trim()) return
|
||||
const meta = colMetaRef.current.find(c => c.cname === col)
|
||||
const group = meta?.dim_group && dimMembers[meta.dim_group]
|
||||
|
||||
// With the member list in hand this is a local lookup. It also answers where
|
||||
// the source query could not: the source holds every attribute set a part ever
|
||||
// had, so anything with more than one came back ambiguous and filled nothing.
|
||||
// A member row is a single definition by construction.
|
||||
let derived = null
|
||||
if (group) {
|
||||
derived = group.members.find(m => m.key_value === value.trim())?.attrs || null
|
||||
if (!derived) return
|
||||
} else {
|
||||
const res = await fetch(`/api/sources/${sourceId}/lookup?col=${encodeURIComponent(col)}&value=${encodeURIComponent(value)}`)
|
||||
if (!res.ok) return
|
||||
const derived = await res.json()
|
||||
derived = await res.json()
|
||||
}
|
||||
if (!derived) return
|
||||
setter(prev => {
|
||||
const next = { ...prev }
|
||||
@ -1257,7 +1293,7 @@ export default function Forecast({ sources = [], sourceId, versions = [], versio
|
||||
cloneSet, setCloneSet,
|
||||
cloneScale, setCloneScale,
|
||||
cloneNote, setCloneNote,
|
||||
dimCols, lookupDerivedCols, versionId,
|
||||
dimCols, lookupDerivedCols, dimMembers, versionId,
|
||||
buildPayload, submitOp,
|
||||
}
|
||||
|
||||
|
||||
@ -25,6 +25,7 @@ export default function Setup({ refreshSources }) {
|
||||
const [sqlStatus, setSqlStatus] = useState({}) // sourceId -> bool
|
||||
const [saving, setSaving] = useState(false)
|
||||
const [generating, setGenerating] = useState(false)
|
||||
const [refreshingDims, setRefreshingDims] = useState(false)
|
||||
const [msg, setMsg] = useState(null)
|
||||
const [dimPeriodCols, setDimPeriodCols] = useState([])
|
||||
const [openPeriodIdx, setOpenPeriodIdx] = useState(null)
|
||||
@ -153,6 +154,35 @@ export default function Setup({ refreshSources }) {
|
||||
}
|
||||
}
|
||||
|
||||
// Rebuild every keyed dim_group's member list from the source. Deliberate rather
|
||||
// than automatic: it reads the whole source, which for a view over a transaction
|
||||
// table is millions of rows, and the answer only changes when the catalogue does.
|
||||
async function refreshDimMembers() {
|
||||
const groups = [...new Set(cols.filter(c => c.dim_group && c.is_key).map(c => c.dim_group))]
|
||||
if (groups.length === 0) {
|
||||
flash('No dim_group has an is_key column, so there is nothing to build a list from', 'error')
|
||||
return
|
||||
}
|
||||
setRefreshingDims(true)
|
||||
try {
|
||||
const done = []
|
||||
for (const g of groups) {
|
||||
const res = await fetch(`/api/sources/${selectedSource.id}/dim/${encodeURIComponent(g)}/refresh`,
|
||||
{ method: 'POST' })
|
||||
const data = await res.json()
|
||||
if (!res.ok) { flash(`${g}: ${data.error}`, 'error'); return }
|
||||
done.push(`${g}: ${data.members.toLocaleString()} members`
|
||||
+ (data.no_longer_in_source ? `, ${data.no_longer_in_source} no longer in source` : '')
|
||||
+ ` (${(data.ms / 1000).toFixed(1)}s)`)
|
||||
}
|
||||
flash(done.join(' · '))
|
||||
} catch (err) {
|
||||
flash(err.message, 'error')
|
||||
} finally {
|
||||
setRefreshingDims(false)
|
||||
}
|
||||
}
|
||||
|
||||
async function deleteSource(id, e) {
|
||||
e.stopPropagation()
|
||||
if (!confirm('Deregister this source? Existing forecast tables are not affected.')) return
|
||||
@ -295,6 +325,17 @@ export default function Setup({ refreshSources }) {
|
||||
{saving ? 'Saving…' : 'Save'}
|
||||
</button>
|
||||
)}
|
||||
{cols.some(c => c.dim_group && c.is_key) && (
|
||||
<button
|
||||
onClick={refreshDimMembers}
|
||||
disabled={refreshingDims || colsDirty}
|
||||
className="text-xs border border-gray-200 px-3 py-1 rounded hover:bg-gray-50 disabled:opacity-50"
|
||||
title={colsDirty ? 'Save col meta first'
|
||||
: 'Rebuild the member list for each keyed dim_group from the source. Reads the whole source, so it takes a while.'}
|
||||
>
|
||||
{refreshingDims ? 'Refreshing…' : 'Refresh master data'}
|
||||
</button>
|
||||
)}
|
||||
<button
|
||||
onClick={generateSQL}
|
||||
disabled={generating || colsDirty}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user