From 437574f0bb6259cb8778c194dd9af1aec0501796 Mon Sep 17 00:00:00 2001 From: mo Date: Sun, 28 Jun 2026 18:51:55 +0000 Subject: [PATCH] feat: federated query spans all 5 databases (not just 3) The marquee panel only joined 3 region-keyed sources. Add a "one SQL across every database" reach matrix that fans a single Trino query out to PostgreSQL, MySQL, MongoDB, Cassandra and the Hadoop/Iceberg lake in one UNION ALL (telemetry has no region, so a per-source summary is used instead of a misleading join). - New MATRIX_SQL + concurrent execution alongside the region scorecard so total latency stays ~ the slower query - Federated tab shows the 5-source matrix (records + headline metric per engine) above the relabelled 3-source region scorecard --- api/trino_federated.py | 45 +++++++++++++- .../features/TrinoFederationView.tsx | 62 ++++++++++++++++++- 2 files changed, 103 insertions(+), 4 deletions(-) diff --git a/api/trino_federated.py b/api/trino_federated.py index c467be4..7deecde 100644 --- a/api/trino_federated.py +++ b/api/trino_federated.py @@ -68,6 +68,25 @@ MARQUEE_SQL = ( "ORDER BY o.revenue DESC NULLS LAST" ) +# Federated reach - ONE SQL touching every database/engine in the stack. +# Telemetry has no region dimension, so instead of a misleading join we +# summarise each source side-by-side in a single UNION ALL query. +MATRIX_SQL = ( + "SELECT 1 ord, 'PostgreSQL' source, 'postgres_sales' catalog, 'public.sales_orders' dataset,\n" + " count(*) records, CAST(sum(amount) AS double) metric, 'total revenue' metric_label\n" + "FROM postgres_sales.public.sales_orders\n" + "UNION ALL SELECT 2,'MySQL','mysql_hr','hr.employee_events',count(*),\n" + " CAST(count(DISTINCT employee_id) AS double),'distinct employees' FROM mysql_hr.hr.employee_events\n" + "UNION ALL SELECT 3,'MongoDB','mongodb_supplychain','supplychain.events',count(*),\n" + " CAST(sum(amount) AS double),'event value' FROM mongodb_supplychain.supplychain.events\n" + "UNION ALL SELECT 4,'Cassandra','cassandra_telemetry','telemetry.device_metrics',count(*),\n" + " CAST(avg(metric_value) AS double),'avg metric value' FROM cassandra_telemetry.telemetry.device_metrics\n" + "UNION ALL SELECT 5,'Hadoop / Iceberg','iceberg','hadoop.orders_ext',count(*),\n" + " CAST(sum(amount) AS double),'lake revenue' FROM iceberg.hadoop.orders_ext\n" + "ORDER BY ord" +) +MATRIX_CATALOGS = ["postgres_sales", "mysql_hr", "mongodb_supplychain", "cassandra_telemetry", "iceberg"] + def _trino(sql: str, limit: int = 500) -> dict[str, Any]: import sql_console as s @@ -98,15 +117,35 @@ def _load_marquee() -> None: def _marquee_worker() -> None: try: - a = time.time() - res = _trino(MARQUEE_SQL, 50) - elapsed = int((time.time() - a) * 1000) + # Run the two federated queries concurrently so the total wait stays + # close to the slower of the two (region scorecard ~ matrix). + from concurrent.futures import ThreadPoolExecutor + + def timed(sql: str) -> tuple[dict, int]: + a = time.time() + r = _trino(sql, 50) + return r, int((time.time() - a) * 1000) + + with ThreadPoolExecutor(max_workers=2) as ex: + f_region = ex.submit(timed, MARQUEE_SQL) + f_matrix = ex.submit(timed, MATRIX_SQL) + res, elapsed = f_region.result() + mres, melapsed = f_matrix.result() + data = { "ok": res.get("ok", False), "sql": MARQUEE_SQL, "catalogs": ["postgres_sales", "mysql_hr", "mongodb_supplychain"], "elapsed_ms": elapsed, "rows": _rows_as_dicts(res) if res.get("ok") else [], + "matrix": { + "ok": mres.get("ok", False), + "sql": MATRIX_SQL, + "catalogs": MATRIX_CATALOGS, + "elapsed_ms": melapsed, + "rows": _rows_as_dicts(mres) if mres.get("ok") else [], + "error": None if mres.get("ok") else str(mres.get("error", ""))[:300], + }, "error": None if res.get("ok") else str(res.get("error", ""))[:300], "generated_at": datetime.now(timezone.utc).isoformat(), } diff --git a/ui/src/components/features/TrinoFederationView.tsx b/ui/src/components/features/TrinoFederationView.tsx index 4fb76d9..4dd4622 100644 --- a/ui/src/components/features/TrinoFederationView.tsx +++ b/ui/src/components/features/TrinoFederationView.tsx @@ -189,6 +189,9 @@ export function TrinoFederationView({ embedded = false, activeTab }: { embedded? const m = marquee?.marquee const totals = catalogs?.source_totals || {} + const mxRows: any[] = m?.matrix?.rows || [] + const mxMax = Math.max(1, ...mxRows.map((r) => Number(r.records) || 0)) + const mxCatalogs: string[] = m?.matrix?.catalogs || ['postgres_sales', 'mysql_hr', 'mongodb_supplychain', 'cassandra_telemetry', 'iceberg'] const body = ( <> @@ -218,7 +221,64 @@ export function TrinoFederationView({ embedded = false, activeTab }: { embedded? +

+ A single Trino query fanned out to five engines at once — relational, document, wide-column and the Hadoop lakehouse — no copies, no ETL. +

+
+ {mxCatalogs.map((c) => ( + {c} + ))} + +
+ 
{m?.matrix?.sql}
+ {mxRows.length ? ( +
+ + + + + + + + + + + {mxRows.map((r, i) => { + const isMoney = /revenue|value/i.test(r.metric_label || '') + const pct = Math.max(3, (Number(r.records) / mxMax) * 100) + const color = COLORS[i % COLORS.length] + return ( + + + + + + + ) + })} + +
SourceCatalog · datasetRecordsHeadline metric
+ {r.source} + {r.catalog} · {r.dataset} +
+
+ {fmtNum(r.records)} +
+
{isMoney ? fmtMoney(r.metric) : fmtNum(r.metric)} {r.metric_label}
+
+ ) : ( +

{marquee?.running ? <> Federating across all databases… : 'No result yet — click re-run.'}

+ )} + + +