feat(hadoop): real Hive engine + filterable, detailed analytics with lineage

Deploys Apache Hive 3.1.3 (Derby metastore + external table over the HDFS
historical CSV, MapReduce exec) on the Hadoop master, so the engine comparison
shows a REAL measured Hive latency (~5.2s) next to live Trino (~0.3s); Impala
stays clearly-labelled representative. The API re-measures Hive over SSH on a
30-min TTL (cached + persisted, with a committed seed). Adds filters
(year/region/category/channel), a region×category heatmap, Trino exec stats,
and a "where is this data read from" lineage panel (Trino->S3/Iceberg/Parquet
with snapshot+files, Hive->HDFS/CSV with namenode+files). Mounts host SSH key
read-only into the api container for the live Hive benchmark.

api/Dockerfile

@@ -4,7 +4,7 @@ WORKDIR /app
 RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
-COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py .
+COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py hive_bench_seed.json .
 RUN mkdir -p /data
 ENV DATABASE_URL=sqlite:////data/atc-agents.db
 EXPOSE 3201

api/hadoop_analytics.py

@@ -1,67 +1,113 @@
 """Hadoop lakehouse analytics for the Command Center 'Hadoop' tab.
 
-Runs live analytical queries on the curated lakehouse table
+- /api/hadoop/filters    -> available filter values (years/regions/categories/channels)
-(iceberg.hadoop.historical_sales, stored as Parquet on Dell ECS S3) via Trino,
+- /api/hadoop/analytics  -> live, filterable KPIs + breakdowns via Trino (with exec stats)
-and exposes a query-engine comparison: Trino is measured live, while Impala and
+- /api/hadoop/source     -> data lineage: exactly where each engine reads from
-Hive are shown as clearly-labelled *representative* figures (those engines are
+- /api/hadoop/engines    -> engine comparison. Trino is measured live; Hive is a REAL
-not deployed on this platform). Results are cached briefly so the UI stays snappy.
+                            measurement (Apache Hive 3.1.3, MapReduce) refreshed over SSH
+                            from the cluster; Impala is a clearly-labelled representative.
+
+Trino reads the curated table iceberg.hadoop.historical_sales (Parquet on Dell ECS S3);
+Hive reads lake.historical_sales (CSV on HDFS). Both ~25k rows -> fair latency contrast.
 """
 
 from __future__ import annotations
 
 import asyncio
+import json
+import os
 import time
+from pathlib import Path
 from typing import Any
 
 import httpx
-from fastapi import APIRouter
+from fastapi import APIRouter, Query
 from fastapi.responses import JSONResponse
 
-import os
-
 TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/")
 TRINO_USER = os.getenv("TRINO_USER", "mo")
 TABLE = "iceberg.hadoop.historical_sales"
+S3_ENDPOINT = os.getenv("S3_ENDPOINT", "http://10.0.20.111:9020")
+HDFS_NN_URL = os.getenv("HDFS_NN_URL", "http://10.0.21.61:9870")
+
+HIVE_SSH_HOST = os.getenv("HIVE_SSH_HOST", "10.0.21.61")
+HIVE_SSH_USER = os.getenv("HIVE_SSH_USER", "root")
+HIVE_BENCH_CMD = os.getenv("HIVE_BENCH_CMD", "/opt/hive/bench.sh")
+HIVE_BENCH_TTL = float(os.getenv("HIVE_BENCH_TTL", "1800"))  # 30 min
+_SEED = Path(__file__).with_name("hive_bench_seed.json")
+_PERSIST = Path("/data/hive_bench.json")
 
 router = APIRouter(prefix="/api/hadoop", tags=["hadoop"])
 
 _cache: dict[str, Any] = {}
-_TTL = 60.0  # seconds
+_TTL = 45.0
+
+# dimension -> Iceberg column
+DIMS = {"years": "order_year", "regions": "region", "categories": "product_category", "channels": "channel"}
 
 
-async def _trino(sql: str, deadline_s: float = 30.0) -> list[list[Any]]:
+# --------------------------------------------------------------------------- Trino
-    """Execute a Trino statement and return all rows (list of lists)."""
+async def _trino(sql: str, deadline_s: float = 30.0) -> dict[str, Any]:
-    headers = {
+    headers = {"X-Trino-User": TRINO_USER, "X-Trino-Catalog": "iceberg", "X-Trino-Schema": "hadoop"}
-        "X-Trino-User": TRINO_USER,
-        "X-Trino-Catalog": "iceberg",
-        "X-Trino-Schema": "hadoop",
-    }
     rows: list[list[Any]] = []
+    cols: list[str] | None = None
+    stats: dict[str, Any] = {}
     start = time.monotonic()
     async with httpx.AsyncClient(timeout=15.0) as client:
         r = await client.post(f"{TRINO_URL}/v1/statement", content=sql.encode(), headers=headers)
         r.raise_for_status()
         payload = r.json()
         while True:
-            err = payload.get("error")
+            if payload.get("error"):
-            if err:
+                raise RuntimeError(payload["error"].get("message", str(payload["error"])))
-                raise RuntimeError(err.get("message", str(err)))
+            if payload.get("columns") and cols is None:
+                cols = [c["name"] for c in payload["columns"]]
             rows.extend(payload.get("data", []) or [])
+            if payload.get("stats"):
+                stats = payload["stats"]
             nxt = payload.get("nextUri")
             if not nxt:
                 break
             if time.monotonic() - start > deadline_s:
                 raise TimeoutError("Trino query exceeded deadline")
-            await asyncio.sleep(0.05)
+            await asyncio.sleep(0.04)
             rr = await client.get(nxt, headers={"X-Trino-User": TRINO_USER})
             rr.raise_for_status()
             payload = rr.json()
-    return rows
+    return {"rows": rows, "columns": cols or [], "stats": stats}
 
 
-def _cached(key: str):
+def _trino_stats(stats: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "processed_rows": stats.get("processedRows"),
+        "processed_bytes": stats.get("processedBytes"),
+        "elapsed_ms": stats.get("elapsedTimeMillis"),
+        "cpu_ms": stats.get("cpuTimeMillis"),
+        "peak_memory_bytes": stats.get("peakMemoryBytes"),
+        "splits": stats.get("totalSplits"),
+    }
+
+
+# --------------------------------------------------------------------------- filters
+def _parse(v: str | None) -> list[str]:
+    return [x.strip() for x in (v or "").split(",") if x.strip()]
+
+
+def _where(years, regions, categories, channels) -> str:
+    clauses = []
+    yrs = [y for y in years if y.lstrip("-").isdigit()]
+    if yrs:
+        clauses.append(f"order_year IN ({','.join(yrs)})")
+    for vals, col in ((regions, "region"), (categories, "product_category"), (channels, "channel")):
+        if vals:
+            esc = ",".join("'" + v.replace("'", "''") + "'" for v in vals)
+            clauses.append(f"{col} IN ({esc})")
+    return ("WHERE " + " AND ".join(clauses)) if clauses else ""
+
+
+def _cached(key: str, ttl: float = _TTL):
     item = _cache.get(key)
-    if item and (time.time() - item["ts"] < _TTL):
+    if item and (time.time() - item["ts"] < ttl):
         return item["data"]
     return None
 
@@ -71,42 +117,62 @@ def _store(key: str, data: Any):
     return data
 
 
+@router.get("/filters")
+async def filters():
+    cached = _cached("filters", ttl=300)
+    if cached is not None:
+        return JSONResponse(cached)
+    try:
+        out = {}
+        q = await asyncio.gather(
+            _trino(f"SELECT DISTINCT order_year FROM {TABLE} ORDER BY 1"),
+            _trino(f"SELECT DISTINCT region FROM {TABLE} ORDER BY 1"),
+            _trino(f"SELECT DISTINCT product_category FROM {TABLE} ORDER BY 1"),
+            _trino(f"SELECT DISTINCT channel FROM {TABLE} ORDER BY 1"),
+        )
+        out = {
+            "ok": True,
+            "years": [r[0] for r in q[0]["rows"]],
+            "regions": [r[0] for r in q[1]["rows"]],
+            "categories": [r[0] for r in q[2]["rows"]],
+            "channels": [r[0] for r in q[3]["rows"]],
+        }
+        return JSONResponse(_store("filters", out))
+    except Exception as e:
+        return JSONResponse({"ok": False, "error": str(e)})
+
+
 @router.get("/analytics")
-async def analytics():
+async def analytics(
-    cached = _cached("analytics")
+    years: str | None = Query(None),
+    regions: str | None = Query(None),
+    categories: str | None = Query(None),
+    channels: str | None = Query(None),
+):
+    where = _where(_parse(years), _parse(regions), _parse(categories), _parse(channels))
+    key = "an:" + where
+    cached = _cached(key)
     if cached is not None:
         return JSONResponse(cached)
     try:
         kpis_q = (
-            f"SELECT count(*) AS orders, sum(amount) AS revenue, avg(amount) AS aov, "
+            f"SELECT count(*), sum(amount), avg(amount), sum(quantity), count(DISTINCT region), "
-            f"sum(quantity) AS units, count(DISTINCT region) AS regions, "
+            f"min(order_year), max(order_year), avg(unit_price) FROM {TABLE} {where}"
-            f"min(order_year) AS min_y, max(order_year) AS max_y FROM {TABLE}"
         )
-        by_year_q = (
+        kpi_res = await _trino(kpis_q)
-            f"SELECT order_year, sum(amount) AS revenue, count(*) AS orders "
+        k = kpi_res["rows"][0] if kpi_res["rows"] else [0] * 8
-            f"FROM {TABLE} GROUP BY order_year ORDER BY order_year"
+        by_year, by_region, by_category, by_channel, by_status = await asyncio.gather(
+            _trino(f"SELECT order_year, sum(amount), count(*) FROM {TABLE} {where} GROUP BY order_year ORDER BY order_year"),
+            _trino(f"SELECT region, sum(amount), count(*) FROM {TABLE} {where} GROUP BY region ORDER BY 2 DESC"),
+            _trino(f"SELECT product_category, sum(amount), count(*) FROM {TABLE} {where} GROUP BY product_category ORDER BY 2 DESC"),
+            _trino(f"SELECT channel, sum(amount), count(*) FROM {TABLE} {where} GROUP BY channel ORDER BY 2 DESC"),
+            _trino(f"SELECT region, product_category, sum(amount) FROM {TABLE} {where} GROUP BY region, product_category ORDER BY 1,2"),
         )
-        by_region_q = (
-            f"SELECT region, sum(amount) AS revenue FROM {TABLE} "
-            f"GROUP BY region ORDER BY revenue DESC"
-        )
-        by_category_q = (
-            f"SELECT product_category, sum(amount) AS revenue FROM {TABLE} "
-            f"GROUP BY product_category ORDER BY revenue DESC"
-        )
-        by_channel_q = (
-            f"SELECT channel, sum(amount) AS revenue, count(*) AS orders FROM {TABLE} "
-            f"GROUP BY channel ORDER BY revenue DESC"
-        )
-        kpis, by_year, by_region, by_category, by_channel = await asyncio.gather(
-            _trino(kpis_q), _trino(by_year_q), _trino(by_region_q),
-            _trino(by_category_q), _trino(by_channel_q),
-        )
-        k = kpis[0] if kpis else [0, 0, 0, 0, 0, None, None]
         data = {
             "ok": True,
             "table": TABLE,
-            "location": "s3://data/hadoop/historical_sales (Iceberg/Parquet on Dell ECS)",
+            "filtered": bool(where),
+            "where": where,
             "kpis": {
                 "orders": int(k[0] or 0),
                 "revenue": float(k[1] or 0),
@@ -115,76 +181,182 @@ async def analytics():
                 "regions": int(k[4] or 0),
                 "year_min": k[5],
                 "year_max": k[6],
+                "avg_unit_price": float(k[7] or 0),
             },
-            "by_year": [{"year": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_year],
+            "by_year": [{"year": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_year["rows"]],
-            "by_region": [{"region": r[0], "revenue": float(r[1] or 0)} for r in by_region],
+            "by_region": [{"region": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_region["rows"]],
-            "by_category": [{"category": r[0], "revenue": float(r[1] or 0)} for r in by_category],
+            "by_category": [{"category": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_category["rows"]],
-            "by_channel": [{"channel": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_channel],
+            "by_channel": [{"channel": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_channel["rows"]],
+            "matrix": [{"region": r[0], "category": r[1], "revenue": float(r[2] or 0)} for r in by_status["rows"]],
+            "trino_stats": _trino_stats(kpi_res["stats"]),
         }
-        return JSONResponse(_store("analytics", data))
+        return JSONResponse(_store(key, data))
     except Exception as e:
-        return JSONResponse({"ok": False, "error": str(e)}, status_code=200)
+        return JSONResponse({"ok": False, "error": str(e)})
+
+
+@router.get("/source")
+async def source():
+    cached = _cached("source", ttl=120)
+    if cached is not None:
+        return JSONResponse(cached)
+    trino_src: dict[str, Any] = {
+        "engine": "Trino",
+        "catalog": "iceberg",
+        "schema": "hadoop",
+        "table": "historical_sales",
+        "format": "Apache Iceberg · Parquet",
+        "storage": "Dell ECS S3 (object store)",
+        "location": "s3://data/hadoop/historical_sales",
+        "s3_endpoint": S3_ENDPOINT,
+        "metastore": "Iceberg (file metadata on S3)",
+    }
+    try:
+        f = await _trino('SELECT count(*), coalesce(sum(file_size_in_bytes),0) FROM iceberg.hadoop."historical_sales$files"')
+        trino_src["data_files"] = int(f["rows"][0][0]) if f["rows"] else None
+        trino_src["data_bytes"] = int(f["rows"][0][1]) if f["rows"] else None
+        s = await _trino('SELECT snapshot_id, committed_at FROM iceberg.hadoop."historical_sales$snapshots" ORDER BY committed_at DESC LIMIT 1')
+        if s["rows"]:
+            trino_src["snapshot_id"] = str(s["rows"][0][0])
+            trino_src["snapshot_at"] = str(s["rows"][0][1])
+        c = await _trino(f"SELECT count(*) FROM {TABLE}")
+        trino_src["rows"] = int(c["rows"][0][0]) if c["rows"] else None
+    except Exception as e:
+        trino_src["error"] = str(e)
+
+    hv = _hive_cache["data"] or {}
+    hive_src = {
+        "engine": "Hive",
+        "database": "lake",
+        "table": "historical_sales",
+        "format": hv.get("storage", "HDFS · CSV (TEXTFILE)"),
+        "storage": "HDFS (Hadoop cluster)",
+        "location": hv.get("hdfs_location", "hdfs://atc-hadoop-m01:8020/data/historical/sales_orders"),
+        "namenode": HDFS_NN_URL,
+        "metastore": "Hive Metastore (Derby)",
+        "exec": hv.get("exec", "MapReduce"),
+        "data_files": hv.get("hdfs_files"),
+        "data_bytes": hv.get("hdfs_size_bytes"),
+        "rows": hv.get("rows"),
+        "measured_at": hv.get("measured_at"),
+    }
+    data = {"ok": True, "sources": [trino_src, hive_src], "hdfs_namenode": HDFS_NN_URL}
+    return JSONResponse(_store("source", data))
+
+
+# --------------------------------------------------------------------------- Hive (real, via SSH)
+def _load_seed() -> dict[str, Any]:
+    for p in (_PERSIST, _SEED):
+        try:
+            if p.exists():
+                return json.loads(p.read_text())
+        except Exception:
+            pass
+    return {}
+
+
+_hive_cache: dict[str, Any] = {"data": _load_seed(), "ts": 0.0, "refreshing": False}
+
+
+def _ssh_hive_bench() -> dict[str, Any]:
+    import paramiko
+
+    cli = paramiko.SSHClient()
+    cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    cli.connect(HIVE_SSH_HOST, username=HIVE_SSH_USER, timeout=10, look_for_keys=True, allow_agent=False)
+    try:
+        _stdin, stdout, _stderr = cli.exec_command(HIVE_BENCH_CMD, timeout=120)
+        out = stdout.read().decode("utf-8", "replace").strip()
+    finally:
+        cli.close()
+    line = [ln for ln in out.splitlines() if ln.strip().startswith("{")]
+    return json.loads(line[-1]) if line else {}
+
+
+async def _refresh_hive() -> None:
+    if _hive_cache["refreshing"]:
+        return
+    _hive_cache["refreshing"] = True
+    try:
+        res = await asyncio.to_thread(_ssh_hive_bench)
+        if res and res.get("ok"):
+            _hive_cache["data"] = res
+            _hive_cache["ts"] = time.time()
+            try:
+                _PERSIST.write_text(json.dumps(res))
+            except Exception:
+                pass
+    except Exception:
+        pass
+    finally:
+        _hive_cache["refreshing"] = False
 
 
 @router.get("/engines")
 async def engines():
-    """Live-measured Trino latency vs representative Impala/Hive figures."""
+    # Trino measured live (same workload as Hive: revenue by region)
-    cached = _cached("engines")
+    bench_sql = f"SELECT region, sum(amount) AS revenue, count(*) AS n FROM {TABLE} GROUP BY region ORDER BY revenue DESC"
-    if cached is not None:
+    trino_entry: dict[str, Any] = {"name": "Trino", "measured": True, "ok": False}
-        return JSONResponse(cached)
-    bench_sql = (
-        f"SELECT region, product_category, sum(amount) AS revenue, "
-        f"avg(unit_price) AS avg_price, count(*) AS n "
-        f"FROM {TABLE} GROUP BY region, product_category ORDER BY revenue DESC"
-    )
     try:
-        # warm + measure (best of 2 to dampen JIT/scheduling noise)
+        await _trino(bench_sql)  # warm
-        await _trino(bench_sql)
         t0 = time.monotonic()
-        await _trino(bench_sql)
+        res = await _trino(bench_sql)
         trino_ms = round((time.monotonic() - t0) * 1000)
-
+        st = _trino_stats(res["stats"])
-        cnt = await _trino(f"SELECT count(*) FROM {TABLE}")
+        trino_entry.update({
-        rows_scanned = int(cnt[0][0]) if cnt else 0
-
-        # Representative multipliers for an analytical aggregate over columnar
-        # Parquet of this size. Impala (MPP C++ daemons) is in the same league
-        # as Trino; Hive (Tez/MR batch) pays heavy job-startup cost.
-        data = {
             "ok": True,
-            "benchmark_sql": bench_sql,
+            "latency_ms": trino_ms,
-            "rows_scanned": rows_scanned,
+            "model": "MPP · in-memory pipelined",
-            "measured_engine": "Trino",
+            "storage": "Iceberg / Parquet on S3",
-            "engines": [
+            "best_for": "Interactive federated SQL & lakehouse BI",
-                {
+            "note": "Live query on iceberg.hadoop.historical_sales",
-                    "name": "Trino",
+            "processed_rows": st["processed_rows"],
-                    "measured": True,
+            "processed_bytes": st["processed_bytes"],
-                    "latency_ms": trino_ms,
+            "by_region": [{"region": r[0], "revenue": float(r[1] or 0), "n": int(r[2] or 0)} for r in res["rows"]],
-                    "model": "MPP · in-memory pipelined",
+        })
-                    "storage": "Iceberg / Parquet on S3",
-                    "best_for": "Interactive federated SQL & lakehouse BI",
-                    "note": "Live query on iceberg.hadoop.historical_sales",
-                },
-                {
-                    "name": "Impala",
-                    "measured": False,
-                    "latency_ms": max(1, round(trino_ms * 0.9)),
-                    "model": "MPP · C++ daemons (LLVM codegen)",
-                    "storage": "Parquet on HDFS / S3 (HMS)",
-                    "best_for": "Low-latency interactive BI on Hadoop",
-                    "note": "Representative — no Impala daemon deployed",
-                },
-                {
-                    "name": "Hive",
-                    "measured": False,
-                    "latency_ms": max(1, round(trino_ms * 8)),
-                    "model": "Batch · Tez / MapReduce",
-                    "storage": "ORC / Parquet on HDFS (HMS)",
-                    "best_for": "Large ETL / batch transforms",
-                    "note": "Representative — no HiveServer2 deployed",
-                },
-            ],
-        }
-        return JSONResponse(_store("engines", data))
     except Exception as e:
-        return JSONResponse({"ok": False, "error": str(e)}, status_code=200)
+        trino_entry["error"] = str(e)
+        trino_ms = 0
+
+    # Hive: REAL measurement, cached + refreshed over SSH
+    if time.time() - _hive_cache["ts"] > HIVE_BENCH_TTL:
+        asyncio.create_task(_refresh_hive())
+    hv = _hive_cache["data"] or {}
+    hive_ms = int(hv.get("query_time_ms") or 0)
+    hive_entry = {
+        "name": "Hive",
+        "measured": bool(hv.get("ok")),
+        "ok": bool(hv.get("ok")),
+        "latency_ms": hive_ms,
+        "wall_ms": hv.get("wall_ms"),
+        "model": f"Batch · {hv.get('exec', 'MapReduce')}",
+        "storage": hv.get("storage", "HDFS · CSV (TEXTFILE)"),
+        "best_for": "Large ETL / batch transforms",
+        "note": f"Live measurement · Apache Hive {hv.get('version', '3.1.3')} on HDFS",
+        "version": hv.get("version"),
+        "measured_at": hv.get("measured_at"),
+        "rows": hv.get("rows"),
+        "by_region": hv.get("by_region", []),
+    }
+
+    # Impala: representative (not deployed on this cluster)
+    base = trino_ms or 300
+    impala_entry = {
+        "name": "Impala",
+        "measured": False,
+        "ok": True,
+        "latency_ms": max(1, round(base * 0.9)),
+        "model": "MPP · C++ daemons (LLVM codegen)",
+        "storage": "Parquet on HDFS / S3 (HMS)",
+        "best_for": "Low-latency interactive BI on Hadoop",
+        "note": "Representative — Impala daemons not deployed on this cluster",
+    }
+
+    data = {
+        "ok": True,
+        "benchmark": "Revenue by region over ~25k historical rows",
+        "trino_sql": bench_sql,
+        "hive_sql": hv.get("query"),
+        "engines": [trino_entry, hive_entry, impala_entry],
+    }
+    return JSONResponse(data)

api/hive_bench_seed.json

@@ -0,0 +1 @@
+{"engine":"Hive","ok":true,"attempts":1,"version":"3.1.3","exec":"MapReduce (local)","storage":"HDFS · CSV (TEXTFILE)","table":"lake.historical_sales","hdfs_location":"hdfs://atc-hadoop-m01:8020/data/historical/sales_orders","hdfs_size_bytes":2496372,"hdfs_files":5,"query":"SELECT region, round(sum(amount),2) AS revenue, count(*) AS n FROM lake.historical_sales GROUP BY region ORDER BY revenue DESC","query_time_ms":5340,"wall_ms":14440,"rows":25000,"measured_at":"2026-06-26T17:00:37Z","by_region":[{"region":"LATAM","revenue":2.549512032E7,"n":5050},{"region":"EMEA","revenue":2.525153502E7,"n":5047},{"region":"NA","revenue":2.516295049E7,"n":5021},{"region":"EU","revenue":2.489941114E7,"n":4976},{"region":"APAC","revenue":2.474656452E7,"n":4906}]}

docker-compose.yml

@@ -53,6 +53,7 @@ services:
       ELASTIC_PASSWORD: ${ELASTIC_PASSWORD:-}
     volumes:
       - api_data:/data
+      - /root/.ssh:/root/.ssh:ro
     depends_on:
       redis:
         condition: service_started

infra/hadoop/hive_bench.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+# Real Hive benchmark on the HDFS historical dataset. Emits one JSON line.
+# Retries because Hive's LocalJobRunner can flake on a cold first job.
+export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk
+export HIVE_HOME=/opt/hive
+export HADOOP_HOME=/usr/lib/hadoop
+export HADOOP_CONF_DIR=/opt/hive/hadoopconf
+export HADOOP_USER_NAME=hdfs
+export HADOOP_CLIENT_OPTS="-Xmx3g"
+export PATH=$JAVA_HOME/bin:$HIVE_HOME/bin:$PATH
+
+Q="SELECT region, round(sum(amount),2) AS revenue, count(*) AS n FROM lake.historical_sales GROUP BY region ORDER BY revenue DESC"
+ISO=$(date -u +%Y-%m-%dT%H:%M:%SZ)
+HSIZE=$(hdfs dfs -du -s /data/historical/sales_orders 2>/dev/null | awk '{print $1}'); [ -z "$HSIZE" ] && HSIZE=0
+HFILES=$(hdfs dfs -ls -R /data/historical/sales_orders 2>/dev/null | grep -c "part-"); [ -z "$HFILES" ] && HFILES=0
+
+ERR=$(mktemp); OUT=$(mktemp); RC=1; WALL=0; TT=0
+for attempt in 1 2 3; do
+  T0=$(date +%s%3N)
+  $HIVE_HOME/bin/hive -e "$Q" >"$OUT" 2>"$ERR"
+  RC=$?
+  T1=$(date +%s%3N)
+  WALL=$((T1-T0))
+  if [ $RC -eq 0 ] && [ -s "$OUT" ]; then
+    TT=$(grep -oE "Time taken: [0-9.]+ seconds" "$ERR" | grep -oE "[0-9.]+" | sort -nr | head -1)
+    break
+  fi
+  sleep 2
+done
+[ -z "$TT" ] && TT=0
+QMS=$(awk "BEGIN{printf \"%d\", $TT*1000}")
+ROWS_JSON=$(awk -F'\t' 'NF>=3{printf "%s{\"region\":\"%s\",\"revenue\":%s,\"n\":%s}", (c++? ",":""), $1,$2,$3}' "$OUT")
+cat <<JSON
+{"engine":"Hive","ok":$([ $RC -eq 0 ] && echo true || echo false),"attempts":$attempt,"version":"3.1.3","exec":"MapReduce (local)","storage":"HDFS · CSV (TEXTFILE)","table":"lake.historical_sales","hdfs_location":"hdfs://atc-hadoop-m01:8020/data/historical/sales_orders","hdfs_size_bytes":$HSIZE,"hdfs_files":$HFILES,"query":"$Q","query_time_ms":$QMS,"wall_ms":$WALL,"rows":25000,"measured_at":"$ISO","by_region":[${ROWS_JSON}]}
+JSON
+rm -f "$ERR" "$OUT"

infra/hadoop/hive_setup.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+set -e
+export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk
+export HIVE_HOME=/opt/hive
+# Bigtop hadoop: pick a HADOOP_HOME that contains bin/hadoop
+if [ -x /usr/lib/hadoop/bin/hadoop ]; then export HADOOP_HOME=/usr/lib/hadoop; else export HADOOP_HOME=/usr; fi
+export HADOOP_CONF_DIR=/etc/hadoop/conf
+export PATH=$JAVA_HOME/bin:$HIVE_HOME/bin:$PATH
+echo "JAVA_HOME=$JAVA_HOME  HADOOP_HOME=$HADOOP_HOME"
+
+echo "=== guava fix ==="
+rm -f /opt/hive/lib/guava-19.0.jar
+cp -f /usr/lib/hadoop/lib/guava-27.0-jre.jar /opt/hive/lib/ 2>/dev/null || cp -f /usr/lib/hadoop/client/guava-27.0-jre.jar /opt/hive/lib/
+ls /opt/hive/lib/guava-*.jar
+
+echo "=== hive-site.xml ==="
+cat > /opt/hive/conf/hive-site.xml <<'XML'
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<configuration>
+  <property><name>javax.jdo.option.ConnectionURL</name><value>jdbc:derby:;databaseName=/opt/hive/metastore_db;create=true</value></property>
+  <property><name>javax.jdo.option.ConnectionDriverName</name><value>org.apache.derby.jdbc.EmbeddedDriver</value></property>
+  <property><name>hive.metastore.warehouse.dir</name><value>/user/hive/warehouse</value></property>
+  <property><name>hive.execution.engine</name><value>mr</value></property>
+  <property><name>mapreduce.framework.name</name><value>local</value></property>
+  <property><name>hive.exec.mode.local.auto</name><value>true</value></property>
+  <property><name>hive.exec.submitviachild</name><value>false</value></property>
+  <property><name>hive.metastore.schema.verification</name><value>false</value></property>
+  <property><name>datanucleus.schema.autoCreateAll</name><value>true</value></property>
+  <property><name>hive.server2.enable.doAs</name><value>false</value></property>
+  <property><name>hive.stats.autogather</name><value>false</value></property>
+  <property><name>hive.metastore.event.db.notification.api.auth</name><value>false</value></property>
+</configuration>
+XML
+
+cat > /opt/hive/conf/hive-env.sh <<ENV
+export JAVA_HOME=$JAVA_HOME
+export HADOOP_HOME=$HADOOP_HOME
+export HADOOP_CONF_DIR=$HADOOP_CONF_DIR
+export HIVE_HOME=$HIVE_HOME
+ENV
+
+# reusable wrapper for running hive non-interactively
+cat > /opt/hive/runhive.sh <<WRAP
+#!/bin/bash
+export JAVA_HOME=$JAVA_HOME
+export HIVE_HOME=$HIVE_HOME
+export HADOOP_HOME=$HADOOP_HOME
+export HADOOP_CONF_DIR=$HADOOP_CONF_DIR
+export HADOOP_CLIENT_OPTS="-Xmx2g \$HADOOP_CLIENT_OPTS"
+export PATH=\$JAVA_HOME/bin:\$HIVE_HOME/bin:\$PATH
+exec hive "\$@"
+WRAP
+chmod +x /opt/hive/runhive.sh
+
+echo "=== init derby metastore schema ==="
+rm -rf /opt/hive/metastore_db
+cd /opt/hive
+$HIVE_HOME/bin/schematool -dbType derby -initSchema >/tmp/schematool.log 2>&1 && echo "schema init OK" || { echo "schema init FAILED"; tail -25 /tmp/schematool.log; exit 1; }

ui/src/components/features/HadoopAnalytics.tsx

@@ -1,62 +1,52 @@
-import { useCallback, useEffect, useState } from 'react'
+import { useCallback, useEffect, useMemo, useState } from 'react'
-import { Activity, BarChart3, Database, Gauge, Layers, Loader2, RefreshCw, Zap } from 'lucide-react'
+import {
+  Activity, BarChart3, Boxes, Cloud, Database, Gauge, GitBranch, HardDrive,
+  Layers, Loader2, RefreshCw, Server, Tag, Zap,
+} from 'lucide-react'
 import { cn } from '../../lib/utils'
-import { subTabIdle } from '../../lib/tabActive'
+import { subTabActive, subTabIdle } from '../../lib/tabActive'
 
 type Kpis = {
-  orders: number
+  orders: number; revenue: number; aov: number; units: number; regions: number
-  revenue: number
+  year_min: number | null; year_max: number | null; avg_unit_price: number
-  aov: number
-  units: number
-  regions: number
-  year_min: number | null
-  year_max: number | null
 }
-
 type Analytics = {
-  ok: boolean
+  ok: boolean; filtered?: boolean; where?: string; kpis?: Kpis
-  table?: string
-  location?: string
-  kpis?: Kpis
   by_year?: { year: number; revenue: number; orders: number }[]
-  by_region?: { region: string; revenue: number }[]
+  by_region?: { region: string; revenue: number; orders: number }[]
-  by_category?: { category: string; revenue: number }[]
+  by_category?: { category: string; revenue: number; orders: number }[]
   by_channel?: { channel: string; revenue: number; orders: number }[]
+  matrix?: { region: string; category: string; revenue: number }[]
+  trino_stats?: { processed_rows?: number; processed_bytes?: number; elapsed_ms?: number; cpu_ms?: number; peak_memory_bytes?: number; splits?: number }
   error?: string
 }
-
+type Filters = { ok: boolean; years: number[]; regions: string[]; categories: string[]; channels: string[] }
+type SourceItem = {
+  engine: string; catalog?: string; schema?: string; database?: string; table: string
+  format: string; storage: string; location: string; s3_endpoint?: string; namenode?: string
+  metastore?: string; exec?: string; data_files?: number; data_bytes?: number; rows?: number
+  snapshot_id?: string; snapshot_at?: string; measured_at?: string; error?: string
+}
+type Source = { ok: boolean; sources?: SourceItem[]; hdfs_namenode?: string }
 type Engine = {
-  name: string
+  name: string; measured: boolean; ok: boolean; latency_ms: number; wall_ms?: number
-  measured: boolean
+  model: string; storage: string; best_for: string; note: string
-  latency_ms: number
+  processed_rows?: number; processed_bytes?: number; version?: string; measured_at?: string
-  model: string
+  by_region?: { region: string; revenue: number; n: number }[]
-  storage: string
-  best_for: string
-  note: string
 }
+type Engines = { ok: boolean; benchmark?: string; trino_sql?: string; hive_sql?: string; engines?: Engine[] }
 
-type Engines = {
+const usd = (n: number) => (n >= 1e9 ? `$${(n / 1e9).toFixed(2)}B` : n >= 1e6 ? `$${(n / 1e6).toFixed(1)}M` : n >= 1e3 ? `$${(n / 1e3).toFixed(0)}K` : `$${n.toFixed(0)}`)
-  ok: boolean
+const num = (n?: number | null) => (n == null ? '—' : n.toLocaleString('en-US'))
-  benchmark_sql?: string
+const bytes = (n?: number | null) => {
-  rows_scanned?: number
+  if (n == null) return '—'
-  measured_engine?: string
+  if (n >= 1e9) return `${(n / 1e9).toFixed(2)} GB`
-  engines?: Engine[]
+  if (n >= 1e6) return `${(n / 1e6).toFixed(2)} MB`
-  error?: string
+  if (n >= 1e3) return `${(n / 1e3).toFixed(1)} KB`
-}
+  return `${n} B`
-
-const usd = (n: number) => {
-  if (n >= 1e9) return `$${(n / 1e9).toFixed(2)}B`
-  if (n >= 1e6) return `$${(n / 1e6).toFixed(1)}M`
-  if (n >= 1e3) return `$${(n / 1e3).toFixed(0)}K`
-  return `$${n.toFixed(0)}`
-}
-const num = (n: number) => n.toLocaleString('en-US')
-
-const ENGINE_COLORS: Record<string, string> = {
-  Trino: '#22d3ee',
-  Impala: '#fb923c',
-  Hive: '#a78bfa',
 }
+const ms = (n?: number | null) => (n == null ? '—' : n >= 1000 ? `${(n / 1000).toFixed(2)} s` : `${Math.round(n)} ms`)
+const ENGINE_COLORS: Record<string, string> = { Trino: '#22d3ee', Impala: '#fb923c', Hive: '#a78bfa' }
 
 function Kpi({ icon: Icon, label, value, sub }: { icon: typeof Gauge; label: string; value: string; sub?: string }) {
   return (
@@ -70,7 +60,7 @@ function Kpi({ icon: Icon, label, value, sub }: { icon: typeof Gauge; label: str
   )
 }
 
-function BarRow({ label, value, max, display, color = '#34d399' }: { label: string; value: number; max: number; display: string; color?: string }) {
+function BarRow({ label, value, max, display, sub, color = '#34d399' }: { label: string; value: number; max: number; display: string; sub?: string; color?: string }) {
   const pct = max > 0 ? Math.max(2, (value / max) * 100) : 0
   return (
     <div className="flex items-center gap-2 text-[10px]">
@@ -78,7 +68,7 @@ function BarRow({ label, value, max, display, color = '#34d399' }: { label: stri
       <div className="relative h-3.5 flex-1 overflow-hidden rounded-sm bg-surface-overlay/60">
         <div className="h-full rounded-sm" style={{ width: `${pct}%`, background: color }} />
       </div>
-      <span className="w-16 shrink-0 text-right font-mono text-foreground">{display}</span>
+      <span className="w-28 shrink-0 text-right font-mono text-foreground">{display}{sub && <span className="text-foreground-faint"> · {sub}</span>}</span>
     </div>
   )
 }
@@ -94,20 +84,39 @@ function ChartCard({ title, icon: Icon, children }: { title: string; icon: typeo
   )
 }
 
+function Chip({ active, onClick, children }: { active: boolean; onClick: () => void; children: React.ReactNode }) {
+  return (
+    <button type="button" onClick={onClick} className={cn('rounded-full border px-2.5 py-0.5 text-[10px] font-medium transition', active ? 'border-emerald-400/60 bg-emerald-400/20 text-emerald-300' : 'border-border text-foreground-muted hover:border-emerald-400/40 hover:text-foreground')}>
+      {children}
+    </button>
+  )
+}
+
 export function HadoopAnalytics() {
+  const [filters, setFilters] = useState<Filters | null>(null)
   const [data, setData] = useState<Analytics | null>(null)
+  const [source, setSource] = useState<Source | null>(null)
   const [engines, setEngines] = useState<Engines | null>(null)
   const [loading, setLoading] = useState(false)
+  const [selYears, setSelYears] = useState<number[]>([])
+  const [selRegions, setSelRegions] = useState<string[]>([])
+  const [selCats, setSelCats] = useState<string[]>([])
+  const [selChannels, setSelChannels] = useState<string[]>([])
 
-  const load = useCallback(async () => {
+  const qs = useMemo(() => {
+    const p = new URLSearchParams()
+    if (selYears.length) p.set('years', selYears.join(','))
+    if (selRegions.length) p.set('regions', selRegions.join(','))
+    if (selCats.length) p.set('categories', selCats.join(','))
+    if (selChannels.length) p.set('channels', selChannels.join(','))
+    return p.toString()
+  }, [selYears, selRegions, selCats, selChannels])
+
+  const loadAnalytics = useCallback(async (query: string) => {
     setLoading(true)
     try {
-      const [a, e] = await Promise.all([
+      const a = await fetch(`/api/hadoop/analytics${query ? `?${query}` : ''}`).then((r) => r.json())
-        fetch('/api/hadoop/analytics').then((r) => r.json()),
-        fetch('/api/hadoop/engines').then((r) => r.json()),
-      ])
       setData(a)
-      setEngines(e)
     } catch {
       setData({ ok: false, error: 'Analytics API unavailable' })
     } finally {
@@ -115,9 +124,24 @@ export function HadoopAnalytics() {
     }
   }, [])
 
-  useEffect(() => {
+  const loadSideband = useCallback(async () => {
-    load()
+    try {
-  }, [load])
+      const [f, s, e] = await Promise.all([
+        fetch('/api/hadoop/filters').then((r) => r.json()),
+        fetch('/api/hadoop/source').then((r) => r.json()),
+        fetch('/api/hadoop/engines').then((r) => r.json()),
+      ])
+      setFilters(f); setSource(s); setEngines(e)
+    } catch { /* ignore */ }
+  }, [])
+
+  useEffect(() => { loadSideband() }, [loadSideband])
+  useEffect(() => { loadAnalytics(qs) }, [qs, loadAnalytics])
+
+  const toggle = <T,>(v: T, list: T[], set: (x: T[]) => void) =>
+    set(list.includes(v) ? list.filter((x) => x !== v) : [...list, v])
+  const reset = () => { setSelYears([]); setSelRegions([]); setSelCats([]); setSelChannels([]) }
+  const anyFilter = selYears.length + selRegions.length + selCats.length + selChannels.length > 0
 
   const k = data?.kpis
   const maxYear = Math.max(1, ...(data?.by_year || []).map((r) => r.revenue))
@@ -125,110 +149,166 @@ export function HadoopAnalytics() {
   const maxCat = Math.max(1, ...(data?.by_category || []).map((r) => r.revenue))
   const maxChan = Math.max(1, ...(data?.by_channel || []).map((r) => r.revenue))
   const maxLat = Math.max(1, ...(engines?.engines || []).map((e) => e.latency_ms))
+  const matrixRegions = useMemo(() => [...new Set((data?.matrix || []).map((m) => m.region))], [data])
+  const matrixCats = useMemo(() => [...new Set((data?.matrix || []).map((m) => m.category))], [data])
+  const matrixMax = Math.max(1, ...(data?.matrix || []).map((m) => m.revenue))
+  const matrixVal = (r: string, c: string) => data?.matrix?.find((m) => m.region === r && m.category === c)?.revenue || 0
 
   return (
     <div className="scrollbar-thin min-h-0 flex-1 overflow-y-auto p-3">
-      <div className="mb-3 flex flex-wrap items-center justify-between gap-2">
+      {/* Filter bar */}
-        <div>
+      <div className="mb-3 rounded-lg border border-border bg-surface/60 p-2.5">
-          <p className="text-[11px] font-semibold text-foreground">Lakehouse Analytics</p>
+        <div className="mb-1.5 flex items-center justify-between">
-          <p className="font-mono text-[9px] text-foreground-muted">{data?.location || 'iceberg.hadoop.historical_sales'}</p>
+          <span className="flex items-center gap-1.5 text-[11px] font-semibold text-foreground"><Tag className="h-3.5 w-3.5 text-emerald-400" /> Filters</span>
+          <div className="flex items-center gap-2">
+            {anyFilter && <button type="button" onClick={reset} className="text-[10px] text-foreground-muted underline hover:text-foreground">Reset</button>}
+            <button type="button" onClick={() => { loadSideband(); loadAnalytics(qs) }} className={cn('rounded-md px-2.5 py-1 text-[10px]', subTabIdle)}>
+              <RefreshCw className={cn('inline h-3 w-3', loading && 'animate-spin')} /> Refresh
+            </button>
+          </div>
+        </div>
+        <div className="space-y-1.5">
+          <FilterRow label="Year" items={(filters?.years || []).map(String)} sel={selYears.map(String)} onToggle={(v) => toggle(Number(v), selYears, setSelYears)} />
+          <FilterRow label="Region" items={filters?.regions || []} sel={selRegions} onToggle={(v) => toggle(v, selRegions, setSelRegions)} />
+          <FilterRow label="Category" items={filters?.categories || []} sel={selCats} onToggle={(v) => toggle(v, selCats, setSelCats)} />
+          <FilterRow label="Channel" items={filters?.channels || []} sel={selChannels} onToggle={(v) => toggle(v, selChannels, setSelChannels)} />
         </div>
-        <button type="button" onClick={load} className={cn('rounded-md px-3 py-1.5 text-[11px]', subTabIdle)}>
-          <RefreshCw className={cn('inline h-3 w-3', loading && 'animate-spin')} /> Refresh
-        </button>
       </div>
 
-      {loading && !data && (
+      {loading && !data && <p className="flex items-center gap-2 text-[11px] text-foreground-muted"><Loader2 className="h-4 w-4 animate-spin" /> Querying Trino…</p>}
-        <p className="flex items-center gap-2 text-[11px] text-foreground-muted">
-          <Loader2 className="h-4 w-4 animate-spin" /> Querying Trino…
-        </p>
-      )}
       {data && !data.ok && <p className="text-[11px] text-danger">{data.error}</p>}
 
       {k && (
         <>
-          <div className="mb-3 grid grid-cols-2 gap-2 sm:grid-cols-3 lg:grid-cols-6">
+          <div className="mb-3 grid grid-cols-2 gap-2 sm:grid-cols-4 lg:grid-cols-7">
             <Kpi icon={Zap} label="Revenue" value={usd(k.revenue)} sub={`${k.year_min}–${k.year_max}`} />
             <Kpi icon={Activity} label="Orders" value={num(k.orders)} />
             <Kpi icon={Gauge} label="Avg order" value={usd(k.aov)} />
             <Kpi icon={Layers} label="Units" value={num(k.units)} />
+            <Kpi icon={Tag} label="Avg unit $" value={usd(k.avg_unit_price)} />
             <Kpi icon={Database} label="Regions" value={String(k.regions)} />
-            <Kpi icon={BarChart3} label="Years" value={`${(k.year_max ?? 0) - (k.year_min ?? 0) + 1}`} sub={`${k.year_min}–${k.year_max}`} />
+            <Kpi icon={BarChart3} label="Scanned" value={num(data?.trino_stats?.processed_rows)} sub={bytes(data?.trino_stats?.processed_bytes)} />
           </div>
 
           <div className="grid grid-cols-1 gap-2 lg:grid-cols-2">
             <ChartCard title="Revenue by year" icon={BarChart3}>
-              {(data.by_year || []).map((r) => (
+              {(data.by_year || []).map((r) => <BarRow key={r.year} label={String(r.year)} value={r.revenue} max={maxYear} display={usd(r.revenue)} sub={`${num(r.orders)} ord`} color="#22d3ee" />)}
-                <BarRow key={r.year} label={String(r.year)} value={r.revenue} max={maxYear} display={usd(r.revenue)} color="#22d3ee" />
-              ))}
             </ChartCard>
             <ChartCard title="Revenue by region" icon={BarChart3}>
-              {(data.by_region || []).map((r) => (
+              {(data.by_region || []).map((r) => <BarRow key={r.region} label={r.region} value={r.revenue} max={maxRegion} display={usd(r.revenue)} sub={`${num(r.orders)} ord`} color="#34d399" />)}
-                <BarRow key={r.region} label={r.region} value={r.revenue} max={maxRegion} display={usd(r.revenue)} color="#34d399" />
-              ))}
             </ChartCard>
             <ChartCard title="Revenue by product category" icon={BarChart3}>
-              {(data.by_category || []).map((r) => (
+              {(data.by_category || []).map((r) => <BarRow key={r.category} label={r.category} value={r.revenue} max={maxCat} display={usd(r.revenue)} sub={`${num(r.orders)} ord`} color="#fbbf24" />)}
-                <BarRow key={r.category} label={r.category} value={r.revenue} max={maxCat} display={usd(r.revenue)} color="#fbbf24" />
-              ))}
             </ChartCard>
             <ChartCard title="Revenue by channel" icon={BarChart3}>
-              {(data.by_channel || []).map((r) => (
+              {(data.by_channel || []).map((r) => <BarRow key={r.channel} label={r.channel} value={r.revenue} max={maxChan} display={usd(r.revenue)} sub={`${num(r.orders)} ord`} color="#a78bfa" />)}
-                <BarRow key={r.channel} label={r.channel} value={r.revenue} max={maxChan} display={usd(r.revenue)} color="#a78bfa" />
-              ))}
             </ChartCard>
           </div>
+
+          {/* Region x Category heatmap */}
+          {matrixRegions.length > 0 && (
+            <div className="mt-2 rounded-lg border border-border bg-surface/60 p-3">
+              <h3 className="mb-2 flex items-center gap-1.5 text-[11px] font-semibold text-foreground"><Boxes className="h-3.5 w-3.5 text-emerald-400" /> Revenue · region × category</h3>
+              <div className="overflow-x-auto">
+                <table className="text-[10px]">
+                  <thead><tr><th className="px-2 py-1 text-left text-foreground-faint">Region \ Category</th>{matrixCats.map((c) => <th key={c} className="px-2 py-1 text-right text-foreground-muted">{c}</th>)}</tr></thead>
+                  <tbody>
+                    {matrixRegions.map((r) => (
+                      <tr key={r}>
+                        <td className="px-2 py-1 font-medium text-foreground-muted">{r}</td>
+                        {matrixCats.map((c) => {
+                          const v = matrixVal(r, c)
+                          const a = Math.max(0.06, v / matrixMax)
+                          return <td key={c} className="px-2 py-1 text-right font-mono text-foreground" style={{ background: `rgba(52,211,153,${a})` }}>{usd(v)}</td>
+                        })}
+                      </tr>
+                    ))}
+                  </tbody>
+                </table>
+              </div>
+            </div>
+          )}
         </>
       )}
 
+      {/* Data lineage: where is this read from */}
+      <div className="mt-3 rounded-lg border border-border bg-surface/60 p-3">
+        <h3 className="mb-2 flex items-center gap-1.5 text-[11px] font-semibold text-foreground"><GitBranch className="h-3.5 w-3.5 text-cyan-400" /> Where is this data read from?</h3>
+        <div className="grid grid-cols-1 gap-2 lg:grid-cols-2">
+          {(source?.sources || []).map((s) => {
+            const isTrino = s.engine === 'Trino'
+            return (
+              <div key={s.engine} className="rounded-md border border-border/60 bg-surface-overlay/30 p-2.5">
+                <div className="mb-1.5 flex items-center gap-1.5 text-[11px] font-semibold" style={{ color: ENGINE_COLORS[s.engine] || '#94a3b8' }}>
+                  {isTrino ? <Cloud className="h-3.5 w-3.5" /> : <HardDrive className="h-3.5 w-3.5" />} {s.engine} → {s.storage}
+                </div>
+                <dl className="grid grid-cols-[88px_1fr] gap-x-2 gap-y-0.5 text-[10px]">
+                  <Dt>Table</Dt><Dd mono>{(s.catalog || s.database)}.{s.schema ? `${s.schema}.` : ''}{s.table}</Dd>
+                  <Dt>Format</Dt><Dd>{s.format}</Dd>
+                  <Dt>Location</Dt><Dd mono>{s.location}</Dd>
+                  {isTrino ? <><Dt>S3 endpoint</Dt><Dd mono>{s.s3_endpoint}</Dd></> : <><Dt>NameNode</Dt><Dd mono>{s.namenode}</Dd></>}
+                  <Dt>Metastore</Dt><Dd>{s.metastore}</Dd>
+                  <Dt>Data files</Dt><Dd>{num(s.data_files)} · {bytes(s.data_bytes)}</Dd>
+                  <Dt>Rows</Dt><Dd>{num(s.rows)}</Dd>
+                  {s.snapshot_id && <><Dt>Snapshot</Dt><Dd mono>{s.snapshot_id} @ {s.snapshot_at}</Dd></>}
+                  {s.exec && <><Dt>Exec</Dt><Dd>{s.exec}</Dd></>}
+                </dl>
+                {s.error && <p className="mt-1 text-[9px] text-amber-400">{s.error}</p>}
+              </div>
+            )
+          })}
+        </div>
+      </div>
+
       {/* Engine comparison */}
       <div className="mt-3 rounded-lg border border-border bg-surface/60 p-3">
         <div className="mb-1 flex items-center justify-between gap-2">
-          <h3 className="flex items-center gap-1.5 text-[11px] font-semibold text-foreground">
+          <h3 className="flex items-center gap-1.5 text-[11px] font-semibold text-foreground"><Server className="h-3.5 w-3.5 text-cyan-400" /> Query engines · Impala &amp; Hive vs Trino</h3>
-            <Zap className="h-3.5 w-3.5 text-cyan-400" /> Query engines · Impala &amp; Hive vs Trino
+          <span className="text-[9px] text-foreground-faint">{engines?.benchmark} · lower = faster</span>
-          </h3>
-          {engines?.rows_scanned ? (
-            <span className="text-[9px] text-foreground-faint">{num(engines.rows_scanned)} rows · lower = faster</span>
-          ) : null}
         </div>
         <p className="mb-2 text-[9px] text-foreground-muted">
-          Same aggregation over the curated lakehouse table.{' '}
+          <span className="text-cyan-400">Trino</span> (S3/Iceberg) and <span className="text-[#a78bfa]">Hive</span> (HDFS/CSV, Apache Hive 3.1.3 · MapReduce) are both <b>measured live</b>;
-          <span className="text-cyan-400">Trino is measured live</span>; Impala &amp; Hive are representative
+          Impala is a representative reference (daemons not deployed on this cluster).
-          reference figures (those engines are not deployed on this platform).
         </p>
         <div className="space-y-2">
           {(engines?.engines || []).map((e) => (
             <div key={e.name} className="rounded-md border border-border/60 bg-surface-overlay/30 px-2.5 py-2">
               <div className="flex items-center gap-2 text-[10px]">
                 <span className="w-14 shrink-0 font-semibold" style={{ color: ENGINE_COLORS[e.name] || '#94a3b8' }}>{e.name}</span>
-                <span
+                <span className={cn('rounded px-1.5 py-px text-[8px] font-medium uppercase tracking-wide', e.measured ? 'bg-cyan-500/20 text-cyan-300' : 'bg-foreground-faint/15 text-foreground-faint')}>{e.measured ? 'live' : 'representative'}</span>
-                  className={cn(
-                    'rounded px-1.5 py-px text-[8px] font-medium uppercase tracking-wide',
-                    e.measured ? 'bg-cyan-500/20 text-cyan-300' : 'bg-foreground-faint/15 text-foreground-faint',
-                  )}
-                >
-                  {e.measured ? 'live' : 'representative'}
-                </span>
                 <div className="relative h-3.5 flex-1 overflow-hidden rounded-sm bg-surface-overlay/60">
-                  <div
+                  <div className="h-full rounded-sm" style={{ width: `${Math.max(3, (e.latency_ms / maxLat) * 100)}%`, background: ENGINE_COLORS[e.name] || '#94a3b8' }} />
-                    className="h-full rounded-sm"
-                    style={{ width: `${Math.max(3, (e.latency_ms / maxLat) * 100)}%`, background: ENGINE_COLORS[e.name] || '#94a3b8' }}
-                  />
                 </div>
-                <span className="w-16 shrink-0 text-right font-mono text-foreground">{num(e.latency_ms)} ms</span>
+                <span className="w-16 shrink-0 text-right font-mono text-foreground">{ms(e.latency_ms)}</span>
               </div>
-              <div className="mt-1 grid grid-cols-1 gap-x-4 gap-y-0.5 pl-16 text-[9px] text-foreground-muted sm:grid-cols-3">
+              <div className="mt-1 grid grid-cols-1 gap-x-4 gap-y-0.5 pl-16 text-[9px] text-foreground-muted sm:grid-cols-2 lg:grid-cols-4">
                 <span><span className="text-foreground-faint">Model:</span> {e.model}</span>
                 <span><span className="text-foreground-faint">Storage:</span> {e.storage}</span>
-                <span><span className="text-foreground-faint">Best for:</span> {e.best_for}</span>
+                {e.processed_rows != null && <span><span className="text-foreground-faint">Scanned:</span> {num(e.processed_rows)} rows · {bytes(e.processed_bytes)}</span>}
+                {e.version && <span><span className="text-foreground-faint">Engine:</span> v{e.version}{e.wall_ms ? ` · wall ${ms(e.wall_ms)}` : ''}</span>}
+                {e.measured_at && <span><span className="text-foreground-faint">Measured:</span> {e.measured_at.replace('T', ' ').replace('Z', ' UTC')}</span>}
+                <span className="lg:col-span-4 text-foreground-faint">{e.note}</span>
               </div>
             </div>
           ))}
         </div>
-        {engines?.benchmark_sql && (
+        {engines?.trino_sql && <pre className="mt-2 overflow-x-auto rounded bg-black/30 p-2 font-mono text-[9px] leading-relaxed text-foreground-faint">Trino:  {engines.trino_sql}{engines.hive_sql ? `\nHive:   ${engines.hive_sql}` : ''}</pre>}
-          <pre className="mt-2 overflow-x-auto rounded bg-black/30 p-2 font-mono text-[9px] leading-relaxed text-foreground-faint">{engines.benchmark_sql}</pre>
-        )}
       </div>
     </div>
   )
 }
+
+function FilterRow({ label, items, sel, onToggle }: { label: string; items: string[]; sel: string[]; onToggle: (v: string) => void }) {
+  if (!items.length) return null
+  return (
+    <div className="flex items-start gap-2">
+      <span className="mt-0.5 w-16 shrink-0 text-[10px] font-medium text-foreground-faint">{label}</span>
+      <div className="flex flex-wrap gap-1">
+        {items.map((it) => <Chip key={it} active={sel.includes(it)} onClick={() => onToggle(it)}>{it}</Chip>)}
+      </div>
+    </div>
+  )
+}
+
+function Dt({ children }: { children: React.ReactNode }) { return <dt className="text-foreground-faint">{children}</dt> }
+function Dd({ children, mono }: { children: React.ReactNode; mono?: boolean }) { return <dd className={cn('truncate text-foreground-muted', mono && 'font-mono text-[9px]')} title={typeof children === 'string' ? children : undefined}>{children}</dd> }