SPIKE: server-side DuckDB virtual server behind ?engine=duck

Working measurement spike for Option C: aggregation runs in a server-side
DuckDB instead of shipping every row to the browser. Opt-in via
?engine=duck; the default client-side WASM path is untouched.

- routes/perspective.js: POST /perspective/init materializes a version's
  forecast table into a persistent in-process DuckDB (ATTACH postgres,
  so 535k rows land in ~1.5s without streaming through Node); POST
  /perspective/sql executes one Perspective-generated statement, with
  Arrow IPC output for viewport reads.
- ui/src/duckServerHandler.js: VirtualServerHandler that delegates SQL
  generation to Perspective's own GenericSQLVirtualServerModel and POSTs
  the result to the server. Adapted from @perspective-dev/client's
  virtual_servers/duckdb.ts — same SQL model, different transport.
- Forecast.jsx: DUCK_MODE branch replaces the heavy /data fetch with
  /perspective/init, opens the hosted table over a message port, and
  shows a latency overlay. applyDepth routes through viewer.restore()
  since the SQL server cannot handle view.set_depth().

Preserved for a future faster-load effort. Findings and the reason this
is not the shipped path (no interactive tree expand/collapse) are in
pf_perspective_options.md on master.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

package.json

@@ -13,6 +13,7 @@
     "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"
   },

routes/perspective.js

@@ -0,0 +1,105 @@
+// 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;
+};

server.js

@@ -31,6 +31,7 @@ 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;

ui/src/duckServerHandler.js

@@ -0,0 +1,154 @@
+// 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)
+        },
+    }
+}

ui/src/views/Forecast.jsx

@@ -1,9 +1,13 @@
 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
@@ -35,6 +39,7 @@ 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([])
@@ -69,6 +74,7 @@ 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)
@@ -164,9 +170,17 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
     setSlice({})
     expandDepthRef.current = null
     try {
-      const [perspective, dataResult, meta] = await Promise.all([
-        loadPerspective(),
-        fetch(`/api/versions/${vid}/data`).then(async r => {
+      // 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 => {
           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
@@ -191,11 +205,14 @@ 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),
@@ -203,26 +220,44 @@ export default function Forecast({ sources = [], sourceId, versionId, refreshSou
       ])
       const tableName = `fc_${vid}`
 
-      if (rowCount >= 500000) setLargeDataset(true)
-
       if (myId !== initIdRef.current) return
 
-      if (!workerRef.current) workerRef.current = await perspective.worker()
-      const worker = workerRef.current
+      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)
 
-      // 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
+        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))
       }
-      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 {}
@@ -301,6 +336,14 @@ 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()
@@ -741,6 +784,17 @@ 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>