diff --git a/api/trino_federated.py b/api/trino_federated.py index 7deecde..46100fe 100644 --- a/api/trino_federated.py +++ b/api/trino_federated.py @@ -235,6 +235,116 @@ async def get_catalogs(): return {"ok": True, "catalogs": out, "source_totals": totals, "count": len(out)} +# ────────────────────────────────────────────────────────────────────────────── +# Realtime business dashboard — fast: instant source estimates + short-TTL +# cached aggregations over the small materialized Hadoop lake tables. +# ────────────────────────────────────────────────────────────────────────────── +_live_aggs: dict[str, Any] = {"ts": 0.0, "data": None} +_LIVE_AGG_TTL = 6.0 +_avg_order_cache: dict[str, Any] = {"ts": 0.0, "val": 0.0} + + +def _avg_order_value() -> float: + now = time.time() + if now - _avg_order_cache["ts"] < 300 and _avg_order_cache["val"]: + return _avg_order_cache["val"] + res = _trino("SELECT avg(amount) FROM iceberg.hadoop.orders_ext", 1) + val = 0.0 + if res.get("ok"): + try: + val = float((res.get("rows") or [[0]])[0][0] or 0) + except Exception: + val = 0.0 + _avg_order_cache["val"] = val + _avg_order_cache["ts"] = now + return val + + +def _region_matrix_from_lake() -> list[dict]: + o = _terms("SELECT region k, count(*) c, sum(amount) rev FROM iceberg.hadoop.orders_ext GROUP BY region", "k", "c", "rev") + e = _terms("SELECT region k, count(*) c FROM iceberg.hadoop.employees_ext GROUP BY region", "k", "c") + sup = _terms("SELECT region k, count(*) c FROM iceberg.hadoop.supply_events_ext GROUP BY region", "k", "c") + regions: dict[str, dict] = {} + for x in o: + regions.setdefault(x["key"], {})["orders"] = x["count"] + regions[x["key"]]["revenue"] = x.get("value", 0) + for x in e: + regions.setdefault(x["key"], {})["hr_events"] = x["count"] + for x in sup: + regions.setdefault(x["key"], {})["supply_events"] = x["count"] + rows = [{"region": k, "orders": v.get("orders", 0), "revenue": v.get("revenue", 0), + "hr_events": v.get("hr_events", 0), "supply_events": v.get("supply_events", 0)} + for k, v in regions.items() if k] + rows.sort(key=lambda r: r.get("revenue") or 0, reverse=True) + return rows + + +def _live_business_aggs() -> dict[str, Any]: + now = time.time() + if _live_aggs["data"] is not None and now - _live_aggs["ts"] < _LIVE_AGG_TTL: + return _live_aggs["data"] + data = { + "orders_by_region": _terms("SELECT region, count(*) c, sum(amount) rev FROM iceberg.hadoop.orders_ext GROUP BY region ORDER BY rev DESC", "region", "c", "rev"), + "orders_by_status": _terms("SELECT order_status k, count(*) c FROM iceberg.hadoop.orders_ext GROUP BY order_status ORDER BY c DESC", "k", "c"), + "orders_by_channel": _terms("SELECT sales_channel k, count(*) c, sum(amount) rev FROM iceberg.hadoop.orders_ext GROUP BY sales_channel ORDER BY rev DESC", "k", "c", "rev"), + "top_customers": _terms("SELECT customer_name k, count(*) c, sum(amount) rev FROM iceberg.hadoop.orders_ext GROUP BY customer_name ORDER BY rev DESC LIMIT 8", "k", "c", "rev"), + "telemetry_by_metric": _terms("SELECT metric_type k, count(*) c, avg(metric_value) v FROM iceberg.hadoop.telemetry_ext GROUP BY metric_type ORDER BY c DESC", "k", "c", "v"), + "supply_by_type": _terms("SELECT type k, count(*) c FROM iceberg.hadoop.supply_events_ext GROUP BY type ORDER BY c DESC", "k", "c"), + "region_matrix": _region_matrix_from_lake(), + } + _live_aggs["data"] = data + _live_aggs["ts"] = now + return data + + +@router.get("/live") +async def get_live(): + import sql_console as s + orders = s._table_row_count("postgres", "public.sales_orders") or 0 + # MySQL event_id is monotonic, so max(event_id) tracks inserts in real time + # (the planner estimate only refreshes after ANALYZE). + hr_res = _trino("SELECT max(event_id) FROM mysql_hr.hr.employee_events", 1) + hr = 0 + if hr_res.get("ok"): + try: + hr = int((hr_res.get("rows") or [[0]])[0][0] or 0) + except Exception: + hr = 0 + if not hr: + hr = s._table_row_count("mysql", "hr.employee_events") or 0 + supply = s._table_row_count("mongodb", "supplychain.events") or 0 + # Cassandra has no cheap estimate — reuse the exact count from the cached + # federated matrix query when available. + telemetry = 0 + if _marquee.get("data") is None: + _load_marquee() + mq = _marquee.get("data") or {} + for r in ((mq.get("matrix") or {}).get("rows") or []): + if r.get("catalog") == "cassandra_telemetry": + try: + telemetry = int(r.get("records") or 0) + except Exception: + telemetry = 0 + avg_order = _avg_order_value() + sources = [ + {"key": "orders", "label": "Orders", "engine": "PostgreSQL", "catalog": "postgres_sales", "rows": orders, "color": "#fbbf24"}, + {"key": "hr_events", "label": "HR events", "engine": "MySQL", "catalog": "mysql_hr", "rows": hr, "color": "#60a5fa"}, + {"key": "supply_events", "label": "Supply events", "engine": "MongoDB", "catalog": "mongodb_supplychain", "rows": supply, "color": "#a78bfa"}, + {"key": "telemetry", "label": "Telemetry", "engine": "Cassandra", "catalog": "cassandra_telemetry", "rows": telemetry, "color": "#22d3ee"}, + ] + return { + "ok": True, + "ts": datetime.now(timezone.utc).isoformat(), + "sources": sources, + "totals": { + "records": orders + hr + supply + telemetry, + "revenue_est": round(orders * avg_order, 2), + "avg_order": round(avg_order, 2), + }, + "business": _live_business_aggs(), + } + + # ────────────────────────────────────────────────────────────────────────────── # Hadoop lake analytics (live over the small materialized external tables) # ────────────────────────────────────────────────────────────────────────────── diff --git a/ui/src/components/features/DataExplorerView.tsx b/ui/src/components/features/DataExplorerView.tsx index 5df36c6..fe735e0 100644 --- a/ui/src/components/features/DataExplorerView.tsx +++ b/ui/src/components/features/DataExplorerView.tsx @@ -15,15 +15,18 @@ import { Network, HardDrive, ShieldCheck, + Radio, } from 'lucide-react' import { cn } from '../../lib/utils' import { TrinoFederationView, type SubTab } from './TrinoFederationView' +import { LiveDashboard } from './LiveDashboard' -type ExplorerTab = 'business' | SubTab +type ExplorerTab = 'business' | 'live' | SubTab -const TABS: { id: ExplorerTab; label: string; icon: typeof Users; hint: string }[] = [ +const TABS: { id: ExplorerTab; label: string; icon: typeof Users; hint: string; live?: boolean }[] = [ { id: 'business', label: 'Business Overview', icon: BarChart3, hint: 'Customers, orders, workforce, supply chain & telemetry across every source' }, - { id: 'federated', label: 'Federated (Trino)', icon: Network, hint: 'One SQL engine joining PostgreSQL, MySQL & MongoDB live — region scorecard' }, + { id: 'live', label: 'Live', icon: Radio, hint: 'Realtime business activity — live counters, ingestion throughput & region matrix', live: true }, + { id: 'federated', label: 'Federated (Trino)', icon: Network, hint: 'One SQL across all 5 databases + region scorecard joined live' }, { id: 'lake', label: 'Hadoop Lake', icon: HardDrive, hint: 'All business data mirrored as external Iceberg tables on HDFS' }, { id: 'dictionary', label: 'Data Dictionary', icon: ShieldCheck, hint: 'Every table & column with PII / masking status — exactly what the assistant sees' }, ] @@ -263,7 +266,7 @@ export function DataExplorerView() { {/* tab bar */}
- {TABS.map(({ id, label, icon: Icon }) => ( + {TABS.map(({ id, label, icon: Icon, live }) => ( ))}
+ {/* Live realtime dashboard */} + {view === 'live' && } + {/* Trino federation / lake / dictionary tabs */} - {view !== 'business' && } + {(view === 'federated' || view === 'lake' || view === 'dictionary') && } {/* ───────── BUSINESS OVERVIEW ───────── */} {view === 'business' && ( diff --git a/ui/src/components/features/LiveDashboard.tsx b/ui/src/components/features/LiveDashboard.tsx new file mode 100644 index 0000000..d1c5f02 --- /dev/null +++ b/ui/src/components/features/LiveDashboard.tsx @@ -0,0 +1,333 @@ +import { useCallback, useEffect, useRef, useState } from 'react' +import { Activity, Pause, Play, ShoppingCart, Users, Boxes, Cpu, DollarSign, Database, Gauge } from 'lucide-react' +import { cn } from '../../lib/utils' + +type Bucket = { key: string; count: number; value?: number } +type Source = { key: string; label: string; engine: string; catalog: string; rows: number; color: string } +type RegionRow = { region: string; orders: number; revenue: number; hr_events: number; supply_events: number } +type Live = { + ok: boolean + ts: string + sources: Source[] + totals: { records: number; revenue_est: number; avg_order: number } + business: { + orders_by_region: Bucket[] + orders_by_status: Bucket[] + orders_by_channel: Bucket[] + top_customers: Bucket[] + telemetry_by_metric: Bucket[] + supply_by_type: Bucket[] + region_matrix: RegionRow[] + } +} + +const COLORS = ['#34d399', '#60a5fa', '#f472b6', '#fbbf24', '#a78bfa', '#22d3ee', '#fb7185'] +const POLL_MS = 2500 + +function fmtNum(n?: number | string | null) { + if (n == null) return '—' + const v = typeof n === 'number' ? n : Number(n) + if (Number.isNaN(v)) return String(n) + if (Math.abs(v) >= 1e9) return `${(v / 1e9).toFixed(2)}B` + if (Math.abs(v) >= 1e6) return `${(v / 1e6).toFixed(2)}M` + if (Math.abs(v) >= 1e3) return `${(v / 1e3).toFixed(1)}K` + return `${Math.round(v)}` +} +const fmtMoney = (n?: number | string | null) => (n == null ? '—' : `€${fmtNum(n)}`) + +function useTween(target: number, ms = 700) { + const [disp, setDisp] = useState(target) + const cur = useRef(target) + const startVal = useRef(target) + const start = useRef(0) + const raf = useRef(0) + useEffect(() => { + startVal.current = cur.current + start.current = performance.now() + cancelAnimationFrame(raf.current) + const tick = (now: number) => { + const p = Math.min(1, (now - start.current) / ms) + const e = 1 - Math.pow(1 - p, 3) + const val = startVal.current + (target - startVal.current) * e + cur.current = val + setDisp(val) + if (p < 1) raf.current = requestAnimationFrame(tick) + } + raf.current = requestAnimationFrame(tick) + return () => cancelAnimationFrame(raf.current) + }, [target, ms]) + return disp +} + +function Spark({ data, color = '#34d399', height = 44 }: { data: number[]; color?: string; height?: number }) { + if (data.length < 2) return
collecting…
+ const w = 240 + const max = Math.max(1, ...data) + const step = w / (data.length - 1) + const coords = data.map((v, i) => [i * step, height - (v / max) * (height - 6) - 3]) + const line = coords.map((c, i) => `${i === 0 ? 'M' : 'L'}${c[0].toFixed(1)},${c[1].toFixed(1)}`).join(' ') + const area = `${line} L${w},${height} L0,${height} Z` + const gid = `sg-${color.replace('#', '')}` + return ( + + + + + + + + + + + + ) +} + +function Bars({ data, valueKind, colorByIndex }: { data?: Bucket[]; valueKind?: 'money' | 'num'; colorByIndex?: boolean }) { + const d = data || [] + const useVal = valueKind != null + const max = Math.max(1, ...d.map((x) => (useVal && x.value != null ? x.value : x.count))) + if (!d.length) return

No data

+ return ( +
+ {d.map((x, i) => { + const metric = useVal && x.value != null ? x.value : x.count + const pct = Math.max(2, (metric / max) * 100) + const label = useVal && x.value != null ? (valueKind === 'money' ? fmtMoney(x.value) : fmtNum(x.value)) : fmtNum(x.count) + return ( +
+ {x.key ?? '—'} +
+
+
+ {label} +
+ ) + })} +
+ ) +} + +function Donut({ data }: { data?: Bucket[] }) { + const d = data || [] + const total = d.reduce((s, x) => s + x.count, 0) || 1 + let acc = 0 + const r = 42 + const c = 2 * Math.PI * r + if (!d.length) return

No data

+ return ( +
+ + {d.map((x, i) => { + const dash = (x.count / total) * c + const seg = + acc += dash + return seg + })} + +
+ {d.slice(0, 7).map((x, i) => ( +
+ + {x.key ?? '—'} + {((x.count / total) * 100).toFixed(0)}% +
+ ))} +
+
+ ) +} + +function Panel({ title, subtitle, icon: Icon, children, className }: { title: string; subtitle?: string; icon?: typeof Database; children: React.ReactNode; className?: string }) { + return ( +
+
+ {Icon && } +

{title}

+ {subtitle && {subtitle}} +
+ {children} +
+ ) +} + +function LiveCounter({ value, label, sub, accent, icon: Icon, money }: { value: number; label: string; sub?: string; accent: string; icon: typeof Users; money?: boolean }) { + const tv = useTween(value) + return ( +
+
+ +
+
+

{label}

+

{money ? fmtMoney(tv) : fmtNum(tv)}

+ {sub &&

{sub}

} +
+
+ ) +} + +const SRC_ICON: Record = { orders: ShoppingCart, hr_events: Users, supply_events: Boxes, telemetry: Cpu } + +export function LiveDashboard() { + const [data, setData] = useState(null) + const [paused, setPaused] = useState(false) + const [err, setErr] = useState(false) + const [totalHist, setTotalHist] = useState([]) + const [rateBySrc, setRateBySrc] = useState>({}) + const [added, setAdded] = useState(0) + const prev = useRef<{ ts: number; rows: Record; total: number } | null>(null) + const startTotal = useRef(null) + + const poll = useCallback(async () => { + try { + const r = await fetch('/api/federated/live') + if (!r.ok) { setErr(true); return } + const d: Live = await r.json() + setErr(false) + const now = Date.parse(d.ts) || Date.now() + const total = d.totals.records + if (prev.current) { + const dt = Math.max(0.5, (now - prev.current.ts) / 1000) + const totRate = Math.max(0, (total - prev.current.total) / dt) + setTotalHist((h) => [...h, totRate].slice(-90)) + const rmap: Record = {} + d.sources.forEach((s) => { + const p = prev.current!.rows[s.key] ?? s.rows + rmap[s.key] = Math.max(0, (s.rows - p) / dt) + }) + setRateBySrc(rmap) + } + if (startTotal.current == null) startTotal.current = total + setAdded(Math.max(0, total - (startTotal.current || total))) + prev.current = { ts: now, rows: Object.fromEntries(d.sources.map((s) => [s.key, s.rows])), total } + setData(d) + } catch { + setErr(true) + } + }, []) + + useEffect(() => { + poll() + if (paused) return + const t = setInterval(poll, POLL_MS) + return () => clearInterval(t) + }, [poll, paused]) + + const b = data?.business + const totalRate = totalHist.length ? totalHist[totalHist.length - 1] : 0 + const matrix = b?.region_matrix || [] + const maxRev = Math.max(1, ...matrix.map((m) => Number(m.revenue) || 0)) + + return ( +
+ {/* live status bar */} +
+ + + {!paused && } + + + {paused ? 'PAUSED' : 'LIVE'} + + + {fmtNum(totalRate)} rows/s + + · + +{fmtNum(added)} since opened + {err && reconnecting…} + {data ? `updated ${new Date(data.ts).toLocaleTimeString()}` : 'connecting…'} + +
+ + {/* headline counters */} +
+ + + {(data?.sources || []).map((s) => ( + + ))} +
+ + {/* throughput + per-source rates */} +
+ + +
+ ~{(POLL_MS / 1000) * 90}s window + peak {fmtNum(Math.max(0, ...totalHist))}/s +
+ + +
+ {(data?.sources || []).map((s) => { + const rate = rateBySrc[s.key] || 0 + const max = Math.max(1, ...Object.values(rateBySrc)) + return ( +
+ {s.engine} +
+
+
+ {fmtNum(rate)}/s +
+ ) + })} +
+ +
+ + {/* region matrix */} + + {matrix.length ? ( +
+ + + + + + + + + + + + {matrix.map((m, i) => { + const heat = (Number(m.revenue) || 0) / maxRev + return ( + + + + + + + + ) + })} + +
RegionOrdersRevenueHR eventsSupply events
+ {m.region} + {fmtNum(m.orders)}{fmtMoney(m.revenue)}{fmtNum(m.hr_events)}{fmtNum(m.supply_events)}
+
+ ) : ( +

Building matrix…

+ )} + + + {/* business breakdown charts */} +
+ + + + + + +
+

+ Counters & throughput are live source estimates (Trino over PostgreSQL, MySQL, MongoDB & Cassandra); breakdown charts aggregate the materialized Hadoop lake. Polling every {POLL_MS / 1000}s. +

+
+ ) +} diff --git a/ui/src/components/features/TrinoFederationView.tsx b/ui/src/components/features/TrinoFederationView.tsx index 4dd4622..9a143f7 100644 --- a/ui/src/components/features/TrinoFederationView.tsx +++ b/ui/src/components/features/TrinoFederationView.tsx @@ -36,7 +36,7 @@ const fmtMoney = (n?: number | string | null) => (n == null ? '—' : `€${fmtN function Panel({ title, subtitle, icon: Icon, children }: { title: string; subtitle?: string; icon?: typeof Database; children: React.ReactNode }) { return ( -
+
{Icon && }

{title}

@@ -375,10 +375,10 @@ export function TrinoFederationView({ embedded = false, activeTab }: { embedded?
-

+

This is exactly what the assistant knows about your data — every column, its type, and whether it is masked or visible.

-
+
{(dict?.tables || []).map((t: any) => (

{t.engine} · {t.desc}