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

Compare commits

base: pt:spike/duckdb-virtual-server
Branches Tags
pt:master
pt:feature/migrations
pt:feature/static-grain
pt:spike/duckdb-virtual-server
..
compare: pt:master
Branches Tags
pt:feature/migrations
pt:feature/static-grain
pt:spike/duckdb-virtual-server
pt:master

No commits in common. "spike/duckdb-virtual-server" and "master" have entirely different histories.
spike/duck ... master

6 changed files with 22 additions and 1668 deletions
Show Stats Expand all files Collapse all files

1333
package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

1
package.json
View File

@ -13,7 +13,6 @@
"apache-arrow": "^21.1.0",
"cors": "^2.8.5",
"dotenv": "^16.0.0",
"duckdb": "^1.4.4",
"express": "^4.18.2",
"pg": "^8.11.3"
},

105
routes/perspective.js
View File

@ -1,105 +0,0 @@
// SPIKE (Option C): server-side DuckDB virtual-server backend for Perspective.
//
// The browser runs Perspective's VirtualServer + GenericSQLVirtualServerModel
// (WASM) which turns each pivot interaction into a SQL string. That SQL is
// POSTed here and run against a persistent in-process DuckDB that has the
// forecast table materialized from Postgres. Only the generated SQL and the
// aggregated (viewport-sized) Arrow result cross the wire — never all rows.
//
// This is a measurement spike, not production wiring: one global DuckDB
// connection, materialize-once per version (stale after writes until /init
// refresh). See pf_perspective_options.md option C.
const express = require('express');
let duckdb = null;
try { duckdb = require('duckdb'); } catch { /* optional dep */ }
module.exports = function(pool) {
const router = express.Router();
let dbPromise = null; // resolves to { db, conn }
const materialized = new Map(); // versionId -> { duckName, rows, ms }
function getDb() {
if (dbPromise) return dbPromise;
dbPromise = new Promise((resolve, reject) => {
const db = new duckdb.Database(':memory:');
const conn = db.connect();
const PG = `host=${process.env.DB_HOST} port=${process.env.DB_PORT} `
+ `dbname=${process.env.DB_NAME} user=${process.env.DB_USER} password=${process.env.DB_PASSWORD}`;
conn.exec(
`INSTALL postgres; LOAD postgres; `
+ `INSTALL arrow FROM community; LOAD arrow; `
+ `ATTACH '${PG}' AS pg (TYPE postgres, READ_ONLY);`,
(err) => err ? reject(err) : resolve({ db, conn })
);
});
return dbPromise;
}
const allP = (conn, sql) => new Promise((res, rej) => conn.all(sql, (e, r) => e ? rej(e) : res(r)));
const execP = (conn, sql) => new Promise((res, rej) => conn.exec(sql, (e) => e ? rej(e) : res()));
const arrowP = (conn, sql) => new Promise((res, rej) => conn.arrowIPCAll(sql, (e, r) => e ? rej(e) : res(r)));
// bigint-safe JSON for the small metadata queries (schema/size/min-max)
const sendJson = (res, rows) =>
res.type('json').send(JSON.stringify(rows, (k, v) => typeof v === 'bigint' ? Number(v) : v));
async function materialize(conn, versionId, refresh) {
if (materialized.has(versionId) && !refresh) return materialized.get(versionId);
const v = await pool.query(
`SELECT s.tname FROM pf.version v JOIN pf.source s ON s.id = v.source_id WHERE v.id = $1`,
[versionId]
);
if (!v.rows.length) throw new Error(`version ${versionId} not found`);
const tname = v.rows[0].tname;
const fcTable = `fc_${tname}_${versionId}`;
const duckName = `fc_${versionId}`;
const t0 = Date.now();
if (refresh) await execP(conn, `DROP TABLE IF EXISTS "${duckName}";`);
await execP(conn, `CREATE TABLE IF NOT EXISTS "${duckName}" AS SELECT * FROM pg."pf"."${fcTable}";`);
const cnt = await allP(conn, `SELECT count(*) AS c FROM "${duckName}";`);
const info = { duckName, rows: Number(cnt[0].c), ms: Date.now() - t0 };
materialized.set(versionId, info);
return info;
}
// Materialize a version's forecast table into DuckDB and report the hosted id.
router.post('/perspective/init', async (req, res) => {
if (!duckdb) return res.status(501).json({ error: 'duckdb node binding not installed' });
try {
const versionId = parseInt(req.body.versionId);
if (!versionId) return res.status(400).json({ error: 'versionId required' });
const { conn } = await getDb();
const info = await materialize(conn, versionId, !!req.body.refresh);
res.json({ tableId: `memory.${info.duckName}`, rows: info.rows, materialize_ms: info.ms });
} catch (err) {
console.error('[perspective/init]', err.message);
res.status(500).json({ error: err.message });
}
});
// Execute one Perspective-generated SQL statement. arrow=true → Arrow IPC bytes.
router.post('/perspective/sql', async (req, res) => {
if (!duckdb) return res.status(501).json({ error: 'duckdb node binding not installed' });
const { sql, arrow } = req.body || {};
if (!sql) return res.status(400).json({ error: 'sql required' });
try {
const { conn } = await getDb();
if (process.env.PSP_DEBUG_SQL) console.log(`[psp sql${arrow ? ' arrow' : ''}]`, String(sql).replace(/\s+/g, ' ').slice(0, 220));
if (arrow) {
const bufs = await arrowP(conn, sql);
const out = Buffer.concat(bufs, bufs.reduce((a, b) => a + b.length, 0));
res.type('application/octet-stream').send(out);
} else {
sendJson(res, await allP(conn, sql));
}
} catch (err) {
console.error('[perspective/sql]', err.message, '\n SQL:', String(sql).slice(0, 300));
res.status(500).json({ error: err.message });
}
});
return router;
};

1
server.js
View File

@ -31,7 +31,6 @@ app.use('/api', require('./routes/sources')(pool));
app.use('/api', require('./routes/versions')(pool));
app.use('/api', require('./routes/operations')(pool));
app.use('/api', require('./routes/log')(pool));
app.use('/api', require('./routes/perspective')(pool)); // SPIKE: server-side DuckDB virtual server
const port = process.env.PORT || 3010;

154
ui/src/duckServerHandler.js
View File

@ -1,154 +0,0 @@
// SPIKE (Option C): a Perspective VirtualServerHandler that runs in the browser
// but executes its generated SQL on a SERVER-SIDE DuckDB (routes/perspective.js).
//
// Perspective's GenericSQLVirtualServerModel (WASM) turns each view interaction
// into a SQL string; we POST that SQL to /api/perspective/sql and feed the
// result back. Metadata queries come back as JSON; viewGetData comes back as
// Arrow IPC and is handed straight to the data slice. Only generated SQL and
// the aggregated viewport cross the wire.
//
// Adapted from @perspective-dev/client/src/ts/virtual_servers/duckdb.ts
// (which targets in-browser duckdb-wasm). Same SQL model, different transport.
const NUMBER_AGGS = [
'sum', 'count', 'any_value', 'arbitrary', 'array_agg', 'avg', 'bit_and',
'bit_or', 'bit_xor', 'bitstring_agg', 'bool_and', 'bool_or', 'countif',
'favg', 'fsum', 'geomean', 'kahan_sum', 'last', 'max', 'min', 'product',
'string_agg', 'sumkahan',
]
const STRING_AGGS = [
'count', 'any_value', 'arbitrary', 'first', 'countif', 'last', 'string_agg',
]
const FILTER_OPS = [
'==', '!=', 'LIKE', 'IS DISTINCT FROM', 'IS NOT DISTINCT FROM',
'>=', '<=', '>', '<',
]
function duckdbTypeToPsp(name) {
name = String(name).toLowerCase()
if (name === 'varchar' || name === 'utf8') return 'string'
if (name === 'double' || name === 'bigint' || name === 'hugeint' ||
name === 'float64' || name.startsWith('decimal')) return 'float'
if (name.startsWith('int')) return 'integer'
if (name.startsWith('date')) return 'date'
if (name.startsWith('bool')) return 'boolean'
if (name.startsWith('timestamp')) return 'datetime'
if (name.startsWith('json') || name.startsWith('struct')) return 'string'
console.warn(`[duck-server] unknown type '${name}'`)
return 'string'
}
// Build a VirtualServerHandler bound to the given perspective module (for the
// SQL generator) and server SQL endpoint. onTiming(ms, bytes) is called for
// every viewGetData round trip so the UI can show interaction latency.
export function makeDuckServerHandler(mod, { sqlUrl = '/api/perspective/sql', onTiming } = {}) {
let sqlBuilder = null
// The SQL generator must come from an INITIALIZED perspective wasm module.
// The <perspective-viewer> element's static __wasm_module__ is initialized
// once the element is registered; the bare client default export is not, so
// `new perspective.GenericSQLVirtualServerModel()` throws on its wasm glue.
// Mirror @perspective-dev's reference DuckDBHandler and resolve from the viewer.
function wasmModule() {
const viewerClass = typeof customElements !== 'undefined'
&& customElements.get('perspective-viewer')
if (viewerClass && viewerClass.__wasm_module__) return viewerClass.__wasm_module__
return mod
}
const builder = () => (sqlBuilder ||= new (wasmModule().GenericSQLVirtualServerModel)())
async function postSql(sql, wantArrow) {
const r = await fetch(sqlUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ sql, arrow: !!wantArrow }),
})
if (!r.ok) {
let msg = r.statusText
try { msg = (await r.json()).error } catch {}
throw new Error(`[duck-server] ${msg}`)
}
return wantArrow ? new Uint8Array(await r.arrayBuffer()) : r.json()
}
return {
getFeatures() {
return {
group_by: true,
split_by: true,
sort: true,
expressions: true,
group_rollup_mode: ['rollup', 'flat', 'total'],
filter_ops: {
integer: FILTER_OPS, float: FILTER_OPS, string: FILTER_OPS,
boolean: FILTER_OPS, date: FILTER_OPS, datetime: FILTER_OPS,
},
aggregates: {
integer: NUMBER_AGGS, float: NUMBER_AGGS, string: STRING_AGGS,
boolean: STRING_AGGS, date: STRING_AGGS, datetime: STRING_AGGS,
},
}
},
async getHostedTables() {
const rows = await postSql(builder().getHostedTables())
return rows.map(r => `${r.database || 'memory'}.${r.name}`)
},
async tableSchema(tableId) {
const rows = await postSql(builder().tableSchema(tableId))
const schema = {}
for (const r of rows) {
if (!String(r.column_name).startsWith('__')) {
schema[r.column_name] = duckdbTypeToPsp(r.column_type)
}
}
return schema
},
async tableSize(tableId) {
const rows = await postSql(builder().tableSize(tableId))
return Number(rows[0]['count_star()'])
},
async tableMakeView(tableId, viewId, config) {
await postSql(builder().tableMakeView(tableId, viewId, config))
},
async tableValidateExpression(tableId, expression) {
const rows = await postSql(builder().tableValidateExpression(tableId, expression))
return duckdbTypeToPsp(rows[0].column_type)
},
async viewColumnSize(viewId, config) {
const rows = await postSql(builder().viewColumnSize(viewId))
const count = Number(Object.values(rows[0])[0])
const gs = config?.group_by?.length || 0
const isFlat = config?.group_rollup_mode === 'flat'
return count - (gs === 0 ? 0 : isFlat ? gs : gs + 1)
},
async viewSize(viewId) {
const rows = await postSql(builder().viewSize(viewId))
return Number(Object.values(rows[0])[0])
},
async viewDelete(viewId) {
await postSql(builder().viewDelete(viewId))
},
async viewGetMinMax(viewId, columnName, config) {
const rows = await postSql(builder().viewGetMinMax(viewId, columnName, config))
let [min, max] = Object.values(rows[0])
if (typeof min === 'bigint') min = Number(min)
if (typeof max === 'bigint') max = Number(max)
return { min: min ?? null, max: max ?? null }
},
async viewGetData(viewId, config, schema, viewport, dataSlice) {
const t0 = performance.now()
const ipc = await postSql(builder().viewGetData(viewId, config, viewport, schema), true)
onTiming?.(performance.now() - t0, ipc.length)
dataSlice.fromArrowIpc(ipc)
},
}
}

96
ui/src/views/Forecast.jsx
View File

@ -1,13 +1,9 @@
import { useState, useEffect, useRef } from 'react'
import useTheme from '../theme.jsx'
import { makeDuckServerHandler } from '../duckServerHandler.js'
const LAYOUT_KEY = (vid) => `pf_layout_v${vid}` // last-used layout (auto restore)
const LAYOUTS_KEY = (vid) => `pf_layouts_v${vid}` // named layout list
// SPIKE: opt-in server-side DuckDB engine via ?engine=duck (see pf_perspective_options.md, option C)
const DUCK_MODE = new URLSearchParams(window.location.search).get('engine') === 'duck'
let perspectivePromise = null
function loadPerspective() {
if (perspectivePromise) return perspectivePromise
@ -39,7 +35,6 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
const [largeDataset, setLargeDataset] = useState(false)
const [loadProgress, setLoadProgress] = useState(null) // { received, total }
const [msg, setMsg] = useState(null)
const [duckStats, setDuckStats] = useState(null) // SPIKE: { rows, materializeMs, lastMs, lastBytes, queries }
// layouts
const [layouts, setLayouts] = useState([])
@ -74,7 +69,6 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
const viewerRef = useRef(null)
const workerRef = useRef(null)
const duckClientRef = useRef(null) // SPIKE: virtual-server client
const tableRef = useRef(null)
const colMetaRef = useRef([])
const expandDepthRef = useRef(null)
@ -170,17 +164,9 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
setSlice({})
expandDepthRef.current = null
try {
// In DUCK_MODE the heavy /data fetch is replaced by a tiny /perspective/init
// (materialize on server, returns hosted table id) — no rows cross the wire.
const dataPromise = DUCK_MODE
? fetch('/api/perspective/init', {
method: 'POST', headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ versionId: vid }),
}).then(async r => {
if (!r.ok) { const { error } = await r.json(); throw new Error(error || 'init failed') }
return r.json() // { tableId, rows, materialize_ms }
})
: fetch(`/api/versions/${vid}/data`).then(async r => {
const [perspective, dataResult, meta] = await Promise.all([
loadPerspective(),
fetch(`/api/versions/${vid}/data`).then(async r => {
if (!r.ok) { const { error } = await r.json(); throw new Error(error || 'Failed to load data') }
const rowCount = parseInt(r.headers.get('X-Row-Count') || '0')
const total = parseInt(r.headers.get('Content-Length') || '0') || null
@ -205,14 +191,11 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
let pos = 0
for (const c of chunks) { merged.set(c, pos); pos += c.byteLength }
return { buffer: merged.buffer, rowCount }
})
const [perspective, dataResult, meta] = await Promise.all([
loadPerspective(),
dataPromise,
}),
fetch(`/api/sources/${sid}/cols`).then(r => r.json()),
])
const { buffer, rowCount } = dataResult
colMetaRef.current = meta
const validCols = new Set([
...meta.filter(c => ['dimension','value','units','date'].includes(c.role)).map(c => c.cname),
@ -220,44 +203,26 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
])
const tableName = `fc_${vid}`
if (rowCount >= 500000) setLargeDataset(true)
if (myId !== initIdRef.current) return
if (DUCK_MODE) {
// ---- SPIKE: server-side DuckDB virtual server ----
setDuckStats({ rows: dataResult.rows, materializeMs: dataResult.materialize_ms,
queries: 0, lastMs: null, lastBytes: null })
const handler = makeDuckServerHandler(perspective, {
onTiming: (ms, bytes) => setDuckStats(s => ({
...(s || {}), lastMs: Math.round(ms), lastBytes: bytes, queries: (s?.queries || 0) + 1,
})),
})
const port = await perspective.createMessageHandler(handler)
const client = await perspective.worker(Promise.resolve(port))
duckClientRef.current = client
if (tableRef.current) { try { await tableRef.current.delete() } catch {} tableRef.current = null }
tableRef.current = await client.open_table(dataResult.tableId)
} else {
// ---- client-side WASM path (rows loaded into the browser) ----
const { buffer, rowCount } = dataResult
if (rowCount >= 500000) setLargeDataset(true)
if (!workerRef.current) workerRef.current = await perspective.worker()
const worker = workerRef.current
if (!workerRef.current) workerRef.current = await perspective.worker()
const worker = workerRef.current
// Clean up the previous table — by JS reference first, then by name in the
// worker registry (covers the case where the ref was lost or delete failed).
if (tableRef.current) {
try { await tableRef.current.delete() } catch {}
tableRef.current = null
}
try {
const stale = await worker.open_table(tableName)
if (stale) await stale.delete()
} catch {}
const opts = { name: tableName, index: 'pf_id' }
tableRef.current = await (rowCount > 0 ? worker.table(buffer, opts) : worker.table([], opts))
// Clean up the previous table — by JS reference first, then by name in the
// worker registry (covers the case where the ref was lost or delete failed).
if (tableRef.current) {
try { await tableRef.current.delete() } catch {}
tableRef.current = null
}
try {
const stale = await worker.open_table(tableName)
if (stale) await stale.delete()
} catch {}
const opts = { name: tableName, index: 'pf_id' }
tableRef.current = await (rowCount > 0 ? worker.table(buffer, opts) : worker.table([], opts))
if (myId !== initIdRef.current) {
try { await tableRef.current.delete() } catch {}
@ -336,14 +301,6 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
async function applyDepth(d) {
const viewer = viewerRef.current
if (!viewer) return
if (DUCK_MODE) {
// The SQL virtual server has no handler for the imperative ViewSetDepthReq
// that view.set_depth() emits. Drive depth declaratively through the view
// config instead — tableMakeView bakes group_by_depth into the query.
await viewer.restore({ group_by_depth: d })
expandDepthRef.current = d
return
}
const view = await viewer.getView()
await view.set_depth(d)
const plugin = await viewer.getPlugin()
@ -784,17 +741,6 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
Large dataset — pivot may take a moment to render
</div>
)}
{DUCK_MODE && duckStats && (
<div className="absolute top-2 right-2 z-10 bg-emerald-50 border border-emerald-300 text-emerald-900 text-[11px] font-mono px-2.5 py-1.5 rounded shadow-sm leading-tight">
<div className="font-semibold">DuckDB server engine</div>
<div>{duckStats.rows?.toLocaleString()} rows · materialize {duckStats.materializeMs}ms</div>
<div>
last query: {duckStats.lastMs == null ? '—' : `${duckStats.lastMs}ms`}
{duckStats.lastBytes != null ? ` · ${fmtBytes(duckStats.lastBytes)}` : ''}
{' '}· {duckStats.queries} total
</div>
</div>
)}
<perspective-viewer ref={viewerRef} style={{ position: 'absolute', inset: 0 }} />
</div>