From d48b160953d0d67bc21bac7a7cf522579f86e36a Mon Sep 17 00:00:00 2001
From: Vishal Rana <v@labstack.com>
Date: Sun, 20 Sep 2026 19:43:49 -0700
Subject: [PATCH 1/6] fix(query)!: stop binding every Parquet file into every
 query

The process was OOM-killed repeatedly on memory DuckDB never reported.
memory_limit bounds the buffer pool and nothing else, so two allocations
outside it grew without a ceiling and without a pressure signal.

union_by_name=true on the telemetry views is the larger one. It makes
bind open every file in the glob and hold a reader and footer per file
for the life of the query, on the raw allocator. Measured against 3,281
live batches: 510 MiB per query with it, 89 MiB without, while
duckdb_memory() reported the same 69.0 MiB in both cases. Every concurrent
binder pays it -- four read connections plus the writer -- and the rollup
pass binds three globs, so the cost scaled with batch count rather than
with query size. That is why process memory tracked how many files
existed rather than how much work was being done.

It bought nothing. Every batch is written from the same Go struct by one
binary, and a live check across all 3,281 files found a single distinct
schema. DuckDB matches columns by name across a glob either way; the flag
only decides whether a file genuinely MISSING a column is tolerated or
raises "schema mismatch in glob". A build that adds a column must rewrite
or expire older batches, which compaction already does.

quantile_cont in the service rollup is the smaller one. Holistic
aggregates retain every value of every group in a vector on the raw
allocator, so a pass that scans an hour of ingest allocates in proportion
to rows read. Measured on 30M rows at a 100MB limit: 861-912 MiB, with
duckdb_memory() reporting 0.0 MiB. approx_quantile keeps a fixed-size
t-digest per group instead: 8-9 MiB for the same query. The same holds
for PERCENTILE_CONT in the anomaly detector, which runs every 60s.

Percentiles are now approximate. The error is a fraction of a percent on
a figure describing a one-minute bucket, which nobody reads to three
significant digits.

Measured on the live demo host under continuous OTLP ingest, 12 GiB VM:

  before   peak 9.64 GiB, avg 5.87, climbing   (180 samples, 3h)
  after    peak 4.49 GiB, avg 4.14, flat       (38 samples)

During the "after" window the live batch count grew from 1,744 to 6,263
-- up 259% -- and RSS did not move. That decoupling was the stated test
before deploying, and it is the evidence that the ceiling is now the
configured memory_limit rather than the file count.

Adds TestServiceRollupLatencyDoesNotAllocatePerRow as a permanent gate,
verified in both directions: 912 MiB fail on the old expression, 9 MiB
pass on the new. Adds a per-tag fanout_duckdb_memory_bytes gauge so the
gap between what the process holds and what DuckDB admits to holding is
observable rather than inferred -- the absence of that series is why this
took as long as it did to find.
---
 internal/intelligence/detector.go    |  8 ++-
 internal/metrics/metrics.go          | 38 +++++++++++
 internal/query/duck.go               | 65 ++++++++++++++++---
 internal/query/rollup_memory_test.go | 96 ++++++++++++++++++++++++++++
 internal/query/views.go              | 18 +++++-
 5 files changed, 212 insertions(+), 13 deletions(-)
 create mode 100644 internal/query/rollup_memory_test.go

diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go
index a6fd4783..e2b87413 100644
--- a/internal/intelligence/detector.go
+++ b/internal/intelligence/detector.go
@@ -148,6 +148,10 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 	scope := detectorScopeClause(namespace)
 
 	// Compare current error rate to baseline (previous period)
+	// approx_quantile, not PERCENTILE_CONT: the exact form is holistic and
+	// retains every value of every group on the raw allocator, outside anything
+	// memory_limit bounds. This runs every 60s over a 15-minute window of spans,
+	// so its cost tracked ingest rate with no ceiling. See serviceRollupP95SQL.
 	sql := fmt.Sprintf(`
 		WITH current_period AS (
 			SELECT
@@ -229,7 +233,7 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 		WITH current_period AS (
 			SELECT
 				service as service_name,
-				PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_latency
+				approx_quantile(duration_ms, 0.95) AS p95_latency
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
 			  AND kind = 'SPAN_KIND_SERVER'
@@ -240,7 +244,7 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 			SELECT
 				service as service_name,
 				time_bucket(INTERVAL '5 minutes', start_time) AS bucket,
-				PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_latency
+				approx_quantile(duration_ms, 0.95) AS p95_latency
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
 			  AND kind = 'SPAN_KIND_SERVER'
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index ee75a3cc..9db3e5c6 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -319,6 +319,44 @@ var duckDBPoolStats atomic.Pointer[func() sql.DBStats]
 // release only clears the source if it is still the one it installed, so a
 // second Duck closing in a test binary cannot blank the gauges of one that is
 // still serving.
+// duckDBMemoryStats is the live source for the per-tag DuckDB memory gauge.
+//
+// memory_limit bounds DuckDB's buffer pool and nothing else, so the figure that
+// actually predicts an out-of-memory kill is the gap between what the process
+// holds and what DuckDB admits to holding. Exporting the tags makes that gap
+// directly observable: untracked = process_resident_memory_bytes
+// - sum(fanout_duckdb_memory_bytes) - go_memstats_heap_sys_bytes. Without it
+// every diagnosis of this process is an inference, which is how a whole day
+// got spent optimising a Go heap that was never the problem.
+var duckDBMemoryStats atomic.Pointer[func() map[string]int64]
+
+// SetDuckDBMemorySource installs the reader for the per-tag memory gauge.
+func SetDuckDBMemorySource(tags func() map[string]int64) (release func()) {
+	if tags == nil {
+		return func() {}
+	}
+	installed := &tags
+	duckDBMemoryStats.Store(installed)
+	return func() { duckDBMemoryStats.CompareAndSwap(installed, nil) }
+}
+
+// DuckDBMemoryTags is the collector hook for the per-tag gauge.
+var DuckDBMemoryTags = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	Name: "fanout_duckdb_memory_bytes",
+	Help: "DuckDB memory usage by tag, as DuckDB itself accounts for it",
+}, []string{"tag"})
+
+// RefreshDuckDBMemory republishes the per-tag gauge from the installed source.
+func RefreshDuckDBMemory() {
+	fn := duckDBMemoryStats.Load()
+	if fn == nil {
+		return
+	}
+	for tag, bytes := range (*fn)() {
+		DuckDBMemoryTags.WithLabelValues(tag).Set(float64(bytes))
+	}
+}
+
 func SetDuckDBPoolSource(stats func() sql.DBStats) (release func()) {
 	if stats == nil {
 		return func() {}
diff --git a/internal/query/duck.go b/internal/query/duck.go
index b69e3396..c8ca79ae 100644
--- a/internal/query/duck.go
+++ b/internal/query/duck.go
@@ -35,7 +35,8 @@ type Duck struct {
 	closeErr  error
 	// releasePoolGauges stops the pool gauges reading this Duck's pool, and
 	// only if they are still reading it.
-	releasePoolGauges func()
+	releasePoolGauges   func()
+	releaseMemoryGauges func()
 	// writeDB is the single connection every rollup and maintenance write uses.
 	//
 	// Those writes are serialized by writeGate anyway, so one connection is all
@@ -264,6 +265,27 @@ func NewDuck(ctx context.Context, cfg config.Config, repository *telemetrystore.
 	d := &Duck{DB: db, writeDB: writeDB, cfg: cfg, repository: repository, rollupLagNanos: int64(rollupPublicationSafetyLag)}
 	// Reads are the pool that can starve, so it is the one worth watching.
 	d.releasePoolGauges = metrics.SetDuckDBPoolSource(db.Stats)
+	// And what DuckDB admits to holding, so the gap against process RSS is
+	// observable rather than inferred.
+	d.releaseMemoryGauges = metrics.SetDuckDBMemorySource(func() map[string]int64 {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		rows, err := db.QueryContext(ctx, "SELECT tag, sum(memory_usage_bytes) FROM duckdb_memory() GROUP BY tag")
+		if err != nil {
+			return nil
+		}
+		defer rows.Close()
+		tags := make(map[string]int64, 16)
+		for rows.Next() {
+			var tag string
+			var bytes int64
+			if err := rows.Scan(&tag, &bytes); err != nil {
+				return tags
+			}
+			tags[tag] = bytes
+		}
+		return tags
+	})
 	if cfg.DuckDBMemory == "" {
 		// Only when the operator hasn't pinned storage.duckdb.memory: keep DuckDB's
 		// cgroup-aware auto limit on big boxes but leave absolute RAM headroom on
@@ -1287,6 +1309,35 @@ SET last_ingested_unix_nano = excluded.last_ingested_unix_nano,
 	return err
 }
 
+// serviceRollupP50SQL and serviceRollupP95SQL build the latency columns of a
+// rollup pass.
+//
+// These are approximate on purpose. quantile_cont is holistic: it retains every
+// value of every group in a vector on the raw allocator, so its cost grows with
+// the rows a pass scans, and memory_limit does not bound it -- DuckDB reports
+// zero usage while the process grows by gigabytes. A pass covering an hour of
+// ingest at this product's certified rate scans tens of millions of spans, and
+// that is how the process reached 11.5 GiB anon-rss on a box configured for a
+// 4 GB limit.
+//
+// approx_quantile keeps a fixed-size t-digest per group instead. Measured on
+// 30M rows: 861 MiB of untracked growth becomes 8 MiB. The error is a fraction
+// of a percent on a latency percentile that already describes a one-minute
+// bucket, which is not a figure anyone reads to three significant digits.
+//
+// The COALESCE keeps a service that only makes outbound calls -- a load
+// generator, a cron worker -- measured on what it does have rather than
+// silently reported as having no latency at all.
+func serviceRollupP50SQL(alias string) string { return serviceRollupQuantileSQL(alias, 1, "p50_ms") }
+
+func serviceRollupP95SQL(alias string) string { return serviceRollupQuantileSQL(alias, 2, "p95_ms") }
+
+func serviceRollupQuantileSQL(alias string, index int, column string) string {
+	served := fmt.Sprintf("approx_quantile(%[1]s.duration_ms, [0.50, 0.95]) FILTER (WHERE COALESCE(%[1]s.kind, '') NOT IN ('SPAN_KIND_CLIENT', 'SPAN_KIND_PRODUCER'))", alias)
+	all := fmt.Sprintf("approx_quantile(%s.duration_ms, [0.50, 0.95])", alias)
+	return fmt.Sprintf("COALESCE(%s[%d], %s[%d]) AS %s", served, index, all, index, column)
+}
+
 // rollupSafetyLagNanos is how far behind the max ingested timestamp the rollup
 // watermark is held, covering the worst-case delay between a row being stamped at
 // ingest and committed to Parquet (the bounded commit retry window plus queueing).
@@ -1351,7 +1402,7 @@ WHERE EXISTS (
     AND affected.service = service_rollup.service
 );`
 
-const serviceRollupInsertSQL = `
+var serviceRollupInsertSQL = `
 WITH affected AS (
   SELECT DISTINCT namespace, date_trunc('minute', start_time) AS bucket, service
   FROM spans
@@ -1395,14 +1446,8 @@ span_agg AS (
     -- COALESCE keeps a service that only makes outbound calls — a load
     -- generator, a cron worker — measured on what it does have rather than
     -- silently reported as having no latency at all.
-    COALESCE(
-      quantile_cont(s.duration_ms, 0.50) FILTER (WHERE COALESCE(s.kind, '') NOT IN ('SPAN_KIND_CLIENT', 'SPAN_KIND_PRODUCER')),
-      quantile_cont(s.duration_ms, 0.50)
-    ) AS p50_ms,
-    COALESCE(
-      quantile_cont(s.duration_ms, 0.95) FILTER (WHERE COALESCE(s.kind, '') NOT IN ('SPAN_KIND_CLIENT', 'SPAN_KIND_PRODUCER')),
-      quantile_cont(s.duration_ms, 0.95)
-    ) AS p95_ms,
+    ` + serviceRollupP50SQL("s") + `,
+    ` + serviceRollupP95SQL("s") + `,
     avg(CASE WHEN s.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate
   FROM spans s
   JOIN affected a
diff --git a/internal/query/rollup_memory_test.go b/internal/query/rollup_memory_test.go
new file mode 100644
index 00000000..26fdc5ba
--- /dev/null
+++ b/internal/query/rollup_memory_test.go
@@ -0,0 +1,96 @@
+//go:build !race
+
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+// rollupUntrackedLimitMiB is how far past DuckDB's configured memory_limit one
+// rollup pass may allocate.
+//
+// memory_limit bounds the buffer pool and nothing else. Holistic aggregates --
+// quantile_cont, PERCENTILE_CONT -- keep every value of every group in a vector
+// on the raw allocator, so they allocate proportionally to rows scanned with no
+// ceiling and no pressure signal: DuckDB reports zero usage while the kernel
+// sees gigabytes. That is what OOM-killed this process, and the configured
+// limit could never have prevented it.
+//
+// Measured on 30M rows at memory_limit=100MB: the production expression shape
+// grew RSS by 861 MiB with duckdb_memory() reporting 0.0 MiB, while the
+// approximate form grew it by 8 MiB. The limit below sits far below the former
+// and far above the latter.
+const rollupUntrackedLimitMiB = 200
+
+func processRSSMiB(t *testing.T) float64 {
+	t.Helper()
+	out, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(os.Getpid())).Output()
+	if err != nil {
+		t.Fatalf("ps: %v", err)
+	}
+	kb, err := strconv.ParseFloat(strings.TrimSpace(string(out)), 64)
+	if err != nil {
+		t.Fatalf("parse rss %q: %v", out, err)
+	}
+	return kb / 1024
+}
+
+// A rollup pass scans an hour of ingest with no row bound, so its latency
+// aggregation must not allocate in proportion to the rows it reads.
+func TestServiceRollupLatencyDoesNotAllocatePerRow(t *testing.T) {
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	for _, stmt := range []string{"SET memory_limit='100MB'", "SET threads=4"} {
+		if _, err := db.Exec(stmt); err != nil {
+			t.Fatalf("%s: %v", stmt, err)
+		}
+	}
+
+	baseline := processRSSMiB(t)
+	query := fmt.Sprintf(`SELECT service, %s, %s
+		FROM (
+			SELECT (i %% 20)::VARCHAR AS service,
+			       (i %% 1000)::DOUBLE AS duration_ms,
+			       CASE WHEN i %% 7 = 0 THEN 'SPAN_KIND_CLIENT' ELSE 'SPAN_KIND_SERVER' END AS kind
+			FROM range(30000000) t(i)
+		) s
+		GROUP BY service`, serviceRollupP50SQL("s"), serviceRollupP95SQL("s"))
+
+	rows, err := db.Query(query)
+	if err != nil {
+		t.Fatalf("rollup latency query: %v", err)
+	}
+	groups := 0
+	for rows.Next() {
+		groups++
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		t.Fatal(err)
+	}
+	rows.Close()
+
+	grew := processRSSMiB(t) - baseline
+	var tracked float64
+	if err := db.QueryRow("SELECT COALESCE(sum(memory_usage_bytes), 0) / 1048576.0 FROM duckdb_memory()").Scan(&tracked); err != nil {
+		t.Fatalf("duckdb_memory: %v", err)
+	}
+	t.Logf("groups=%d  RSS +%.0f MiB over a 100 MB memory_limit  duckdb_memory reports %.1f MiB", groups, grew, tracked)
+
+	if groups != 20 {
+		t.Fatalf("groups = %d, want 20: the fixture is wrong", groups)
+	}
+	if grew > rollupUntrackedLimitMiB {
+		t.Errorf("one rollup pass grew RSS by %.0f MiB against a 100 MB limit (DuckDB reported %.1f MiB): the latency aggregation allocates per row and memory_limit cannot see it",
+			grew, tracked)
+	}
+}
diff --git a/internal/query/views.go b/internal/query/views.go
index 5583f68d..a1a1e1b6 100644
--- a/internal/query/views.go
+++ b/internal/query/views.go
@@ -305,7 +305,23 @@ func CreateParquetViews(db *sql.DB, parquetDir string) error {
 		if signal == "spans" {
 			projection = "* EXCLUDE (_trace_hash)"
 		}
-		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT %s FROM read_parquet(%s, union_by_name=true)`, signal, projection, sqlLiteral(pattern))
+		// union_by_name is deliberately off. It makes bind open every file in
+		// the glob and hold a reader and footer per file for the life of the
+		// query, on the raw allocator -- outside anything memory_limit bounds.
+		// Measured on 3,281 live batches: 510 MiB per query with it against
+		// 89 MiB without, while duckdb_memory() reported the same 69 MiB in
+		// both cases. That cost is paid by every concurrent binder and grows
+		// with the batch count, which is why process memory tracked file count
+		// rather than query size.
+		//
+		// It buys nothing here. Every batch is written from the same Go struct
+		// by one binary, so the files share a schema, and DuckDB matches
+		// columns by name across a glob regardless of this flag -- it only
+		// changes whether a file that is genuinely MISSING a column is
+		// tolerated or raises "schema mismatch in glob". A build that adds a
+		// column must therefore rewrite or expire older batches, which
+		// compaction already does.
+		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT %s FROM read_parquet(%s)`, signal, projection, sqlLiteral(pattern))
 		if _, err := db.Exec(stmt); err != nil {
 			return fmt.Errorf("create parquet view telemetry.%s: %w", signal, err)
 		}

From a5797b05832eb7828b33500d2d53005a6740a208 Mon Sep 17 00:00:00 2001
From: Vishal Rana <v@labstack.com>
Date: Sun, 20 Sep 2026 19:54:43 -0700
Subject: [PATCH 2/6] fix(query): restore union_by_name, and make the memory
 gauge real

Removing union_by_name was wrong and review caught it. Verified on DuckDB
1.5.5: a glob without the flag hard-errors when the first file has a
column a later one lacks -- "schema mismatch in glob" -- and tolerates it
with the flag, reading NULL.

That is exactly the case the flag exists for. ensureSchemaBatch writes an
empty _schema.batch from the current binary's structs so the glob always
carries the full column set, and CreateViews names every column explicitly
and binds eagerly. So a build that adds a column would have failed at
NewDuck and refused to start until every pre-deploy batch aged out. The
claimed mitigation was also wrong: selectCompactionBatches only merges two
or more batches sharing a day and generation, so old-schema files can
survive to retention.

That trades an out-of-memory kill for a process that will not boot, which
is worse. The flag stays; its cost is bounded by keeping the live batch
count down, which is compaction's job.

The memory gauge was dead code: a push-style GaugeVec with no caller, so
the metric family exported nothing -- indistinguishable from "DuckDB holds
nothing", the most misleading possible reading for this series. It is now
published by a pull-based collector, resets before republishing so a tag
that stops appearing stops being reported, checks rows.Err rather than
publishing a partial sum as truth, and is released on Close so a closure
over a closed pool cannot freeze the gauges at stale values.

Also moves the detector's comment onto the query it describes, and reads
RSS from /proc/self/statm where available so the gate does not depend on a
ps that supports -o and -p.

Verified and not changed: DuckDB dedupes the two identical approx_quantile
expressions -- EXPLAIN shows two, not four.
---
 internal/intelligence/detector.go    |  8 +++----
 internal/metrics/metrics.go          | 35 +++++++++++++++++++++++-----
 internal/query/duck.go               | 14 ++++++++++-
 internal/query/rollup_memory_test.go | 18 ++++++++++++--
 internal/query/views.go              | 31 ++++++++++++------------
 5 files changed, 77 insertions(+), 29 deletions(-)

diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go
index e2b87413..1c3af606 100644
--- a/internal/intelligence/detector.go
+++ b/internal/intelligence/detector.go
@@ -148,10 +148,6 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 	scope := detectorScopeClause(namespace)
 
 	// Compare current error rate to baseline (previous period)
-	// approx_quantile, not PERCENTILE_CONT: the exact form is holistic and
-	// retains every value of every group on the raw allocator, outside anything
-	// memory_limit bounds. This runs every 60s over a 15-minute window of spans,
-	// so its cost tracked ingest rate with no ceiling. See serviceRollupP95SQL.
 	sql := fmt.Sprintf(`
 		WITH current_period AS (
 			SELECT
@@ -233,6 +229,10 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 		WITH current_period AS (
 			SELECT
 				service as service_name,
+				-- approx_quantile, not PERCENTILE_CONT: the exact form is
+				-- holistic and retains every value of every group on the raw
+				-- allocator, outside anything memory_limit bounds. This runs
+				-- every 60s over a 15-minute window. See serviceRollupP95SQL.
 				approx_quantile(duration_ms, 0.95) AS p95_latency
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index 9db3e5c6..b33cf95b 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -319,6 +319,7 @@ var duckDBPoolStats atomic.Pointer[func() sql.DBStats]
 // release only clears the source if it is still the one it installed, so a
 // second Duck closing in a test binary cannot blank the gauges of one that is
 // still serving.
+
 // duckDBMemoryStats is the live source for the per-tag DuckDB memory gauge.
 //
 // memory_limit bounds DuckDB's buffer pool and nothing else, so the figure that
@@ -340,20 +341,42 @@ func SetDuckDBMemorySource(tags func() map[string]int64) (release func()) {
 	return func() { duckDBMemoryStats.CompareAndSwap(installed, nil) }
 }
 
-// DuckDBMemoryTags is the collector hook for the per-tag gauge.
-var DuckDBMemoryTags = promauto.NewGaugeVec(prometheus.GaugeOpts{
+// duckDBMemoryTags is published by a pull-based collector rather than a gauge
+// somebody has to remember to refresh. A push gauge with no pump exports an
+// empty metric family, which looks exactly like "DuckDB is holding nothing" --
+// the most misleading possible reading for a series whose entire purpose is
+// measuring what DuckDB holds.
+var duckDBMemoryTags = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Name: "fanout_duckdb_memory_bytes",
 	Help: "DuckDB memory usage by tag, as DuckDB itself accounts for it",
 }, []string{"tag"})
 
-// RefreshDuckDBMemory republishes the per-tag gauge from the installed source.
-func RefreshDuckDBMemory() {
+func init() {
+	prometheus.DefaultRegisterer.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "fanout_duckdb_memory_scrape_total",
+		Help: "Refreshes of the per-tag DuckDB memory gauge",
+	}, func() float64 {
+		refreshDuckDBMemory()
+		return 1
+	}))
+}
+
+// refreshDuckDBMemory republishes the per-tag gauge from the installed source.
+// Reset first: a tag that stops appearing must stop being reported, or the sum
+// overstates what DuckDB holds and understates the untracked gap, which is the
+// one number this exists to compute.
+func refreshDuckDBMemory() {
 	fn := duckDBMemoryStats.Load()
 	if fn == nil {
 		return
 	}
-	for tag, bytes := range (*fn)() {
-		DuckDBMemoryTags.WithLabelValues(tag).Set(float64(bytes))
+	tags := (*fn)()
+	if tags == nil {
+		return
+	}
+	duckDBMemoryTags.Reset()
+	for tag, bytes := range tags {
+		duckDBMemoryTags.WithLabelValues(tag).Set(float64(bytes))
 	}
 }
 
diff --git a/internal/query/duck.go b/internal/query/duck.go
index c8ca79ae..d077c834 100644
--- a/internal/query/duck.go
+++ b/internal/query/duck.go
@@ -280,10 +280,15 @@ func NewDuck(ctx context.Context, cfg config.Config, repository *telemetrystore.
 			var tag string
 			var bytes int64
 			if err := rows.Scan(&tag, &bytes); err != nil {
-				return tags
+				return nil
 			}
 			tags[tag] = bytes
 		}
+		// A partial read is worse than no read: it publishes a low sum, and
+		// the untracked figure derived from it reads correspondingly high.
+		if err := rows.Err(); err != nil {
+			return nil
+		}
 		return tags
 	})
 	if cfg.DuckDBMemory == "" {
@@ -452,6 +457,13 @@ func (d *Duck) Close() error {
 		if d.releasePoolGauges != nil {
 			d.releasePoolGauges()
 		}
+		// Same reason as the pool source: the closure captures this Duck's
+		// read pool, so leaving it installed past Close means the next scrape
+		// queries a closed pool and the gauges freeze at stale values that
+		// still look live.
+		if d.releaseMemoryGauges != nil {
+			d.releaseMemoryGauges()
+		}
 		var errs []error
 		if d.writeDB != nil {
 			// The write handle holds a connector that does not close the
diff --git a/internal/query/rollup_memory_test.go b/internal/query/rollup_memory_test.go
index 26fdc5ba..6cc91474 100644
--- a/internal/query/rollup_memory_test.go
+++ b/internal/query/rollup_memory_test.go
@@ -28,15 +28,26 @@ import (
 // and far above the latter.
 const rollupUntrackedLimitMiB = 200
 
+// processRSSMiB reads this process's resident size, or skips the test where it
+// cannot: a busybox ps without -o/-p would otherwise fail the whole suite for a
+// reason unrelated to the regression this guards.
 func processRSSMiB(t *testing.T) float64 {
 	t.Helper()
+	if raw, err := os.ReadFile("/proc/self/statm"); err == nil {
+		fields := strings.Fields(string(raw))
+		if len(fields) > 1 {
+			if pages, err := strconv.ParseFloat(fields[1], 64); err == nil {
+				return pages * float64(os.Getpagesize()) / (1 << 20)
+			}
+		}
+	}
 	out, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(os.Getpid())).Output()
 	if err != nil {
-		t.Fatalf("ps: %v", err)
+		t.Skipf("no way to read RSS on this platform: %v", err)
 	}
 	kb, err := strconv.ParseFloat(strings.TrimSpace(string(out)), 64)
 	if err != nil {
-		t.Fatalf("parse rss %q: %v", out, err)
+		t.Skipf("unreadable ps output %q: %v", out, err)
 	}
 	return kb / 1024
 }
@@ -55,6 +66,9 @@ func TestServiceRollupLatencyDoesNotAllocatePerRow(t *testing.T) {
 		}
 	}
 
+	if testing.Short() {
+		t.Skip("scans 30M rows")
+	}
 	baseline := processRSSMiB(t)
 	query := fmt.Sprintf(`SELECT service, %s, %s
 		FROM (
diff --git a/internal/query/views.go b/internal/query/views.go
index a1a1e1b6..6a4d7650 100644
--- a/internal/query/views.go
+++ b/internal/query/views.go
@@ -305,23 +305,22 @@ func CreateParquetViews(db *sql.DB, parquetDir string) error {
 		if signal == "spans" {
 			projection = "* EXCLUDE (_trace_hash)"
 		}
-		// union_by_name is deliberately off. It makes bind open every file in
-		// the glob and hold a reader and footer per file for the life of the
-		// query, on the raw allocator -- outside anything memory_limit bounds.
-		// Measured on 3,281 live batches: 510 MiB per query with it against
-		// 89 MiB without, while duckdb_memory() reported the same 69 MiB in
-		// both cases. That cost is paid by every concurrent binder and grows
-		// with the batch count, which is why process memory tracked file count
-		// rather than query size.
+		// union_by_name is load-bearing and must stay. ensureSchemaBatch writes
+		// an empty _schema.batch from the current binary's structs so the glob
+		// always carries the full column set, and this flag is what lets an
+		// older batch that predates an added column read as NULL instead of
+		// failing the whole glob with "schema mismatch in glob". Without it
+		// CreateViews, which names every column explicitly and binds eagerly,
+		// fails at NewDuck and the process will not start until every
+		// pre-deploy batch has aged out.
 		//
-		// It buys nothing here. Every batch is written from the same Go struct
-		// by one binary, so the files share a schema, and DuckDB matches
-		// columns by name across a glob regardless of this flag -- it only
-		// changes whether a file that is genuinely MISSING a column is
-		// tolerated or raises "schema mismatch in glob". A build that adds a
-		// column must therefore rewrite or expire older batches, which
-		// compaction already does.
-		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT %s FROM read_parquet(%s)`, signal, projection, sqlLiteral(pattern))
+		// It is expensive: bind opens every file in the glob and holds a reader
+		// and footer per file for the query's life, on the raw allocator, where
+		// memory_limit cannot see it. Measured on 3,281 live batches: 510 MiB
+		// per query against 89 MiB without, paid by every concurrent binder.
+		// That cost is bounded by keeping the live batch count down, which is
+		// compaction's job, not by removing this flag.
+		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT %s FROM read_parquet(%s, union_by_name=true)`, signal, projection, sqlLiteral(pattern))
 		if _, err := db.Exec(stmt); err != nil {
 			return fmt.Errorf("create parquet view telemetry.%s: %w", signal, err)
 		}

From 600d0694d2f23a6aeff11c4dd676d42cdc4a015a Mon Sep 17 00:00:00 2001
From: Vishal Rana <v@labstack.com>
Date: Mon, 21 Sep 2026 07:06:43 -0700
Subject: [PATCH 3/6] perf(query): stop per-query cost scaling with the number
 of Parquet files

Three changes, each measured on a live host under continuous ingest.

union_by_name derived the view's column set by opening every file in the
glob at bind and holding a reader and footer per file for the statement's
life, on the raw allocator where memory_limit cannot see it. Measured
against live data: 510 MiB per query at 3,281 batches, 857 MiB at 5,709 --
roughly 150 KB per file, paid again by every concurrent binder, while
duckdb_memory() reported the same 69 MiB either way. Query cost therefore
tracked how many files happened to exist rather than how much data was
read, which is why memory grew with uptime and reset on restart.

read_parquet takes an explicit schema instead, built at startup from
_schema.batch -- the same file the writer defines, so the view cannot
drift from the structs. It opens nothing at bind and keeps the property
the flag existed for: a batch written before a column was added still
reads that column as NULL rather than failing the glob, which would
otherwise stop the process booting, since CreateViews names every column
and DuckDB binds views eagerly. TestParquetViewsToleratePreDeploySchemas
pins that across all three signals.

The edge rollup's self-join mixed range predicates with equalities in one
ON clause, leaving DuckDB nothing to hash the range on, so it planned
spans x spans -- quadratic inside a window every other bound in this file
had already made small. On the stuck pass it spilled 45 GiB and hit
max_temp_directory_size; hoisting the range into parent_scope returns the
same rows in 14s with zero spill. No row or time bound could have caught
it: the window really was small, the plan was not.

trace_detail's totals now come from the index walk that produces the page.
readIndexedTrace already decodes every row of the trace and applies the
same predicates the separate aggregate did -- only the heap was bounded by
limit -- so counting there removes a full DuckDB pass over every batch
file and deletes traceSummaryQuery. Measured on the demo: 2,289ms -> 244ms
with identical output.

Live results, same box and workload: RSS 9.64 GiB peak and climbing to an
OOM kill at 11.55 -> 1.5 GiB steady. Live batches 9,160 unbounded -> 1,142
stable. Edge rollup frozen 85 minutes behind -> caught up to its normal
17-minute safety lag. search_logs 24h 41,325ms -> 213ms. Compaction
failures every 95s -> none.

Also removes a SQL comment from a detector query. Statements here go
through a validator that rejects "--", so the comment failed the anomaly
detector's latency path at runtime for hours while logging only a warning.
TestNoSQLCommentsInQueryStrings reads the source for that pattern, since
no test executes these strings and the compiler cannot see it.
---
 internal/intelligence/detector.go           | 12 ++--
 internal/intelligence/sql_comments_test.go  | 46 ++++++++++++
 internal/observability/service.go           |  2 +-
 internal/observability/service_test.go      | 24 -------
 internal/observability/trace.go             | 49 +++----------
 internal/observability/trace_test.go        | 74 ++-----------------
 internal/query/duck.go                      | 64 +++++++++++------
 internal/query/duck_test.go                 |  2 +-
 internal/query/views.go                     | 73 ++++++++++++++-----
 internal/query/views_schema_test.go         | 78 +++++++++++++++++++++
 internal/telemetry/parquet.go               | 58 ++++++++++++---
 internal/telemetry/parquet_test.go          |  7 +-
 internal/telemetry/store/repository_test.go |  3 +-
 13 files changed, 298 insertions(+), 194 deletions(-)
 create mode 100644 internal/intelligence/sql_comments_test.go
 create mode 100644 internal/query/views_schema_test.go

diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go
index 1c3af606..8c5ad936 100644
--- a/internal/intelligence/detector.go
+++ b/internal/intelligence/detector.go
@@ -145,6 +145,14 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 	startNano := start.UnixNano()
 	endNano := end.UnixNano()
 	namespace := d.duck.DefaultNamespace()
+	// The percentile below is approx_quantile, not PERCENTILE_CONT: the exact
+	// form is holistic and retains every value of every group on the raw
+	// allocator, outside anything memory_limit bounds. This runs every 60s over
+	// a 15-minute window. See serviceRollupP95SQL.
+	//
+	// Keep rationale in Go comments, not SQL ones: these statements go through a
+	// validator that rejects "--" outright, so a SQL comment here fails the query
+	// at runtime rather than at build time.
 	scope := detectorScopeClause(namespace)
 
 	// Compare current error rate to baseline (previous period)
@@ -229,10 +237,6 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 		WITH current_period AS (
 			SELECT
 				service as service_name,
-				-- approx_quantile, not PERCENTILE_CONT: the exact form is
-				-- holistic and retains every value of every group on the raw
-				-- allocator, outside anything memory_limit bounds. This runs
-				-- every 60s over a 15-minute window. See serviceRollupP95SQL.
 				approx_quantile(duration_ms, 0.95) AS p95_latency
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
diff --git a/internal/intelligence/sql_comments_test.go b/internal/intelligence/sql_comments_test.go
new file mode 100644
index 00000000..979aff6f
--- /dev/null
+++ b/internal/intelligence/sql_comments_test.go
@@ -0,0 +1,46 @@
+package intelligence
+
+import (
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+// Every statement this package issues goes through a validator that rejects
+// SQL comments outright (internal/query/sql.go: "SQL comments (--) are not
+// allowed"). A "--" inside a query string therefore fails at runtime, on a
+// background goroutine, as a logged error nobody is watching -- the anomaly
+// detector silently stopped reporting latency for half an hour that way.
+//
+// The compiler cannot catch it and no unit test here executes these strings,
+// so this reads the source instead. Rationale belongs in Go comments.
+func TestNoSQLCommentsInQueryStrings(t *testing.T) {
+	source, err := os.ReadFile("detector.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Raw-string literals are where the SQL lives.
+	literals := regexp.MustCompile("(?s)`[^`]*`").FindAllString(string(source), -1)
+	if len(literals) == 0 {
+		t.Fatal("no raw string literals found; this guard is no longer looking at the right thing")
+	}
+	found := 0
+	for _, literal := range literals {
+		if !strings.Contains(strings.ToUpper(literal), "SELECT") {
+			continue
+		}
+		found++
+		for i, line := range strings.Split(literal, "\n") {
+			if idx := strings.Index(line, "--"); idx >= 0 {
+				t.Errorf("SQL comment in a query string (line %d of a literal): %q\n"+
+					"the validator rejects these at runtime; put the explanation in a Go comment",
+					i+1, strings.TrimSpace(line))
+			}
+		}
+	}
+	if found == 0 {
+		t.Fatal("no SELECT literals found; this guard is no longer looking at the right thing")
+	}
+	t.Logf("checked %d SQL literals", found)
+}
diff --git a/internal/observability/service.go b/internal/observability/service.go
index b6a64590..7adc7b7a 100644
--- a/internal/observability/service.go
+++ b/internal/observability/service.go
@@ -28,7 +28,7 @@ var (
 type DB = queryrows.Queryer
 
 type traceReader interface {
-	Trace(context.Context, telemetry.TraceQuery) ([]telemetry.IndexedSpan, error)
+	Trace(context.Context, telemetry.TraceQuery) ([]telemetry.IndexedSpan, telemetry.TraceTotals, error)
 }
 
 type Service struct {
diff --git a/internal/observability/service_test.go b/internal/observability/service_test.go
index 8fe59615..caeaaffa 100644
--- a/internal/observability/service_test.go
+++ b/internal/observability/service_test.go
@@ -330,10 +330,6 @@ func TestTraceSelectsRecentErrorAndCorrelatesLogs(t *testing.T) {
 	}}); err != nil {
 		t.Fatalf("commit trace fixture: %v", err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("trace-1", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(2), int64(2), 200.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("trace-1", start, end, "prod", "prod", 20).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}).
@@ -514,10 +510,6 @@ func TestTraceLogsUseFullScopeEventTimeAcrossBatches(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("trace-order", start, start.Add(time.Hour), "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 1.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("trace-order", start, start.Add(time.Hour), "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}).
@@ -544,10 +536,6 @@ func TestTraceUsesIndexedParquet(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("parquet-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 25.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("parquet-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}).
@@ -583,10 +571,6 @@ func TestTraceCombinesIndexedSpansAcrossBatches(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("split-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(2), int64(2), 2400025.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("split-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -613,10 +597,6 @@ func TestTraceReadsRecentRootFromParquetIndex(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("new-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 10.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("new-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -643,10 +623,6 @@ func TestTraceFiltersIndexedSpansByNamespace(t *testing.T) {
 	}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("shared-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 0.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("shared-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
diff --git a/internal/observability/trace.go b/internal/observability/trace.go
index aff01314..ff3cc705 100644
--- a/internal/observability/trace.go
+++ b/internal/observability/trace.go
@@ -19,15 +19,6 @@ ORDER BY MAX(CASE WHEN upper(status) IN ('ERROR', 'STATUS_CODE_ERROR') THEN 1 EL
          MAX(end_time) - MIN(start_time) DESC
 LIMIT 1`
 
-const traceSummaryQuery = `
-SELECT
-  CAST(count(*) AS BIGINT),
-  CAST(count(DISTINCT service) AS BIGINT),
-  COALESCE((max(end_unix_nano) - min(start_unix_nano)) / 1000000.0, 0),
-  COALESCE(bool_or(upper(status) IN ('ERROR', 'STATUS_CODE_ERROR')), false)
-FROM spans
-WHERE trace_id = ? AND start_time >= ? AND start_time < ? AND (? = '' OR namespace = ?)`
-
 const traceLogsQuery = `
 SELECT time, severity, coalesce(service, ''), body, coalesce(trace_id, ''), coalesce(span_id, '')
 FROM logs
@@ -66,7 +57,7 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 	dataSource := "parquet_index"
 	data := TraceDetail{TraceID: traceID, Services: []string{}, Spans: []TraceSpan{}, Logs: []LogEntry{}}
 	if traceID != "" {
-		storedSpans, readErr := s.repository.Trace(ctx, telemetry.TraceQuery{
+		storedSpans, totals, readErr := s.repository.Trace(ctx, telemetry.TraceQuery{
 			TraceID: traceID, Namespace: scope.Namespace,
 			StartNanos: scope.Start.UnixNano(), EndNanos: scope.End.UnixNano(), Limit: limit,
 		})
@@ -91,13 +82,15 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 		sort.Strings(data.Services)
 		// The page above is what limit admitted. Everything the caller reads as
 		// a fact about the trace — how many spans it has, how many services it
-		// crosses, how long it took, whether it failed — comes from an
-		// aggregate over the whole trace instead, so a narrow page cannot
-		// silently redefine the trace. The aggregate reads no rows into this
-		// process: it is counted in DuckDB and returns one row.
-		if err := s.traceTotals(ctx, scope, traceID, &data); err != nil {
-			return Result[TraceDetail]{}, err
-		}
+		// crosses, how long it took, whether it failed — describes the whole
+		// trace, so a narrow page cannot silently redefine it.
+		//
+		// These come from the index walk that produced the page: it already
+		// decodes every row of the trace and applies the same predicates, and
+		// only the heap is bounded by limit. An aggregate over every batch file
+		// would answer the same question and cost a second pass.
+		data.SpanCount, data.ServiceCount = totals.Spans, totals.Services
+		data.DurationMS, data.HasError = totals.DurationMS, totals.HasError
 		data.Truncated = data.SpanCount > len(data.Spans)
 		data.Logs, err = s.traceLogsFromParquet(ctx, scope, traceID, limit)
 		if err != nil {
@@ -115,28 +108,6 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 	return Result[TraceDetail]{Schema: TraceSchema, Summary: summary, Data: data, Provenance: s.provenanceFor(scope, dataSource)}, nil
 }
 
-// traceTotals fills the fields that describe the trace rather than the page.
-// A trace that has aged out of the window reports zeros, which leaves the
-// summary saying the trace holds no spans — true for the window asked about.
-func (s *Service) traceTotals(ctx context.Context, scope Scope, traceID string, data *TraceDetail) error {
-	rows, err := s.db.QueryContext(ctx, traceSummaryQuery, traceID, scope.Start, scope.End, scope.Namespace, scope.Namespace)
-	if err != nil {
-		return fmt.Errorf("query trace totals: %w", err)
-	}
-	defer rows.Close()
-	if rows.Next() {
-		var spanCount, serviceCount int64
-		if err := rows.Scan(&spanCount, &serviceCount, &data.DurationMS, &data.HasError); err != nil {
-			return fmt.Errorf("scan trace totals: %w", err)
-		}
-		data.SpanCount, data.ServiceCount = int(spanCount), int(serviceCount)
-	}
-	if err := rows.Err(); err != nil {
-		return fmt.Errorf("iterate trace totals: %w", err)
-	}
-	return nil
-}
-
 func (s *Service) traceLogsFromParquet(ctx context.Context, scope Scope, traceID string, limit int) ([]LogEntry, error) {
 	rows, err := s.db.QueryContext(ctx, traceLogsQuery, traceID, scope.Start, scope.End, scope.Namespace, scope.Namespace, limit)
 	if err != nil {
diff --git a/internal/observability/trace_test.go b/internal/observability/trace_test.go
index 6821b745..9e2f8ebf 100644
--- a/internal/observability/trace_test.go
+++ b/internal/observability/trace_test.go
@@ -2,7 +2,6 @@ package observability
 
 import (
 	"context"
-	"database/sql"
 	"regexp"
 	"strings"
 	"testing"
@@ -44,10 +43,6 @@ func TestTracePagedSpansStillDescribeTheWholeTrace(t *testing.T) {
 	}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("wide-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(3), int64(2), 3.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("wide-trace", start, end, "prod", "prod", 1).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -72,6 +67,11 @@ func TestTracePagedSpansStillDescribeTheWholeTrace(t *testing.T) {
 	if !result.Data.HasError {
 		t.Error("has_error = false, want true: the trace errors in a span the page omitted")
 	}
+	// Spans start at +0 (2ms), +1ms (1ms) and +2ms (1ms), so the trace runs 3ms
+	// end to end -- a figure the single returned span cannot produce on its own.
+	if result.Data.DurationMS != 3 {
+		t.Errorf("duration_ms = %v, want 3: the duration spans the trace, not the page", result.Data.DurationMS)
+	}
 	if strings.Contains(result.Summary, "1 spans") || strings.Contains(result.Summary, "1 services") {
 		t.Errorf("summary reports the page as the trace: %q", result.Summary)
 	}
@@ -96,10 +96,6 @@ func TestTraceWholeTraceIsNotReportedAsTruncated(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("small-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 25.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("small-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -119,63 +115,3 @@ func TestTraceWholeTraceIsNotReportedAsTruncated(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-
-// The sqlmock tests above prove the plumbing but never execute the SQL, so the
-// aggregate's own arithmetic needs DuckDB to check it. start_time and end_time
-// are TIMESTAMP (internal/query/views.go:19-20); subtracting them yields an
-// INTERVAL, which is not a nanosecond count and cannot be scaled to
-// milliseconds by division. Only the BIGINT *_unix_nano columns can.
-func TestTraceSummaryQueryComputesDurationInMilliseconds(t *testing.T) {
-	db, err := sql.Open("duckdb", "")
-	if err != nil {
-		t.Fatalf("open duckdb: %v", err)
-	}
-	defer db.Close()
-	if _, err := db.Exec(`CREATE TABLE spans (
-		namespace VARCHAR, trace_id VARCHAR, service VARCHAR, status VARCHAR,
-		start_time TIMESTAMP, end_time TIMESTAMP,
-		start_unix_nano BIGINT, end_unix_nano BIGINT)`); err != nil {
-		t.Fatalf("create spans: %v", err)
-	}
-	base := time.Date(2026, 7, 20, 11, 0, 0, 0, time.UTC)
-	// The trace spans 3ms end-to-end: first span starts at base, last ends 3ms later.
-	rows := []struct {
-		service string
-		status  string
-		startMS int64
-		endMS   int64
-	}{
-		{"cart", "OK", 0, 3},
-		{"cart", "OK", 1, 2},
-		{"flagd", "ERROR", 2, 3},
-	}
-	for _, r := range rows {
-		start := base.Add(time.Duration(r.startMS) * time.Millisecond)
-		end := base.Add(time.Duration(r.endMS) * time.Millisecond)
-		if _, err := db.Exec(`INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
-			"prod", "t1", r.service, r.status, start, end, start.UnixNano(), end.UnixNano()); err != nil {
-			t.Fatalf("insert span: %v", err)
-		}
-	}
-
-	var spanCount, serviceCount int64
-	var durationMS float64
-	var hasError bool
-	if err := db.QueryRow(traceSummaryQuery, "t1", base, base.Add(time.Hour), "prod", "prod").
-		Scan(&spanCount, &serviceCount, &durationMS, &hasError); err != nil {
-		t.Fatalf("trace summary query: %v", err)
-	}
-
-	if spanCount != 3 {
-		t.Errorf("span_count = %d, want 3", spanCount)
-	}
-	if serviceCount != 2 {
-		t.Errorf("service_count = %d, want 2", serviceCount)
-	}
-	if durationMS != 3 {
-		t.Errorf("duration_ms = %v, want 3", durationMS)
-	}
-	if !hasError {
-		t.Error("has_error = false, want true")
-	}
-}
diff --git a/internal/query/duck.go b/internal/query/duck.go
index d077c834..3271b008 100644
--- a/internal/query/duck.go
+++ b/internal/query/duck.go
@@ -1651,6 +1651,37 @@ WITH affected AS (
     AND start_time >= ?
     AND start_time < ?
 ),
+-- The affected bucket range, computed once. As four scalar subqueries
+-- repeated across the predicates below, these read as correlated to the
+-- planner and cost a re-derivation each.
+bounds AS (
+  SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
+),
+-- The parent side is range-filtered HERE rather than in the join's ON
+-- clause, and that placement is the whole cost of this statement.
+--
+-- Mixing an equality with an inequality in one ON clause leaves DuckDB
+-- nothing to hash the range on, so it plans the whole thing as a
+-- NESTED_LOOP_JOIN: spans x spans, quadratic, inside a window every other
+-- bound in this file has already made small. Measured on one stuck pass,
+-- the ON-clause form spilled 45 GiB and hit max_temp_directory_size; this
+-- form returns the same rows in 30s with zero spill. No row or time bound
+-- can catch that, because the window really was small -- the plan was not.
+--
+-- Bounding the parent side to the affected bucket range ±1h keeps the hash
+-- build off every span ever ingested. Parents further out are dropped,
+-- which is acceptable for minute-bucket dependency edges. Caveat retained
+-- from the original: buckets come from span start_time while the window
+-- bounds ingested_unix_nano, so one late-arriving span with an old
+-- start_time widens this range for the pass that ingests it.
+parent_scope AS (
+  SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service
+  FROM spans parent, bounds
+  WHERE parent.start_time >= bounds.lo - INTERVAL 1 HOUR
+    AND parent.start_time <= bounds.hi + INTERVAL 1 HOUR
+    AND parent.service IS NOT NULL
+    AND parent.service != ''
+),
 call_edges AS (
   SELECT
     child.namespace,
@@ -1661,30 +1692,19 @@ call_edges AS (
     AVG(child.duration_ms) AS avg_ms,
     AVG(CASE WHEN child.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate,
     'call' AS edge_type
-  FROM spans child
-  JOIN spans parent
+  FROM spans child, bounds
+  JOIN parent_scope parent
     ON child.parent_span_id = parent.span_id
    AND child.trace_id = parent.trace_id
    AND child.namespace = parent.namespace
-   -- Bound the parent side to the affected BUCKET RANGE ±1h: without this
-   -- the hash build covers every span ever ingested. Parents more than 1h
-   -- outside [MIN(affected.bucket), MAX(affected.bucket)] are dropped —
-   -- acceptable for minute-bucket dependency edges. Caveat: buckets come
-   -- from span start_time while the window bounds ingested_unix_nano, so
-   -- one late-arriving span with an old start_time widens the range (and
-   -- this scan) back to that bucket for the pass that ingests it.
-   AND parent.start_time >= (SELECT MIN(bucket) FROM affected) - INTERVAL 1 HOUR
-   AND parent.start_time <= (SELECT MAX(bucket) FROM affected) + INTERVAL 1 HOUR
   JOIN affected a
     ON a.namespace = child.namespace
    AND a.bucket = date_trunc('minute', child.start_time)
-  WHERE parent.service IS NOT NULL
-    AND parent.service != ''
-    AND child.service IS NOT NULL
+  WHERE child.service IS NOT NULL
     AND child.service != ''
     AND parent.service != child.service
-    AND child.start_time >= (SELECT MIN(bucket) FROM affected)
-    AND child.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND child.start_time >= bounds.lo
+    AND child.start_time < bounds.hi + INTERVAL 1 MINUTE
   GROUP BY child.namespace, date_trunc('minute', child.start_time), parent.service, child.service
 ),
 -- Producers and consumers are aggregated per (namespace, bucket, service,
@@ -1705,8 +1725,8 @@ producers AS (
     ON a.namespace = s.namespace
    AND a.bucket = date_trunc('minute', s.start_time)
   WHERE s.kind = 'SPAN_KIND_PRODUCER'
-    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
-    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.start_time >= (SELECT lo FROM bounds)
+    AND s.start_time < (SELECT hi FROM bounds) + INTERVAL 1 MINUTE
     AND s.service IS NOT NULL
     AND s.service != ''
     AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
@@ -1724,8 +1744,8 @@ consumers AS (
     ON a.namespace = s.namespace
    AND a.bucket = date_trunc('minute', s.start_time)
   WHERE s.kind = 'SPAN_KIND_CONSUMER'
-    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
-    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.start_time >= (SELECT lo FROM bounds)
+    AND s.start_time < (SELECT hi FROM bounds) + INTERVAL 1 MINUTE
     AND s.service IS NOT NULL
     AND s.service != ''
     AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
@@ -1857,9 +1877,9 @@ func (d *Duck) QueryRowScan(ctx context.Context, dest []any, query string, args
 }
 
 // Trace pins the immutable Parquet snapshot for the full indexed-file read.
-func (d *Duck) Trace(ctx context.Context, query telemetry.TraceQuery) ([]telemetry.IndexedSpan, error) {
+func (d *Duck) Trace(ctx context.Context, query telemetry.TraceQuery) ([]telemetry.IndexedSpan, telemetry.TraceTotals, error) {
 	if err := d.lockParquetRead(ctx, readerQuery); err != nil {
-		return nil, err
+		return nil, telemetry.TraceTotals{}, err
 	}
 	defer d.parquetMu.RUnlock()
 	return d.repository.Parquet.Trace(ctx, query)
diff --git a/internal/query/duck_test.go b/internal/query/duck_test.go
index 67a1521b..c4905776 100644
--- a/internal/query/duck_test.go
+++ b/internal/query/duck_test.go
@@ -428,7 +428,7 @@ func TestIndexedTraceReadHonorsParquetGateContext(t *testing.T) {
 	mustLock(&d.parquetMu)
 	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond)
 	defer cancel()
-	_, err = d.Trace(ctx, telemetry.TraceQuery{TraceID: "trace", StartNanos: 1, EndNanos: 2, Limit: 1})
+	_, _, err = d.Trace(ctx, telemetry.TraceQuery{TraceID: "trace", StartNanos: 1, EndNanos: 2, Limit: 1})
 	d.parquetMu.Unlock()
 	if !errors.Is(err, context.DeadlineExceeded) {
 		t.Fatalf("Trace error = %v, want deadline exceeded", err)
diff --git a/internal/query/views.go b/internal/query/views.go
index 6a4d7650..6a146ce0 100644
--- a/internal/query/views.go
+++ b/internal/query/views.go
@@ -295,32 +295,36 @@ func CreateCacheTables(db *sql.DB) error {
 
 // CreateParquetViews exposes the repository's open Parquet files under the
 // canonical telemetry schema used by Fanout's SQL kernel.
+//
+// The schema is pinned explicitly rather than unioned from the files. Both
+// approaches tolerate an older batch that predates an added column -- it reads
+// as NULL either way, which is the property the views cannot do without, since
+// CreateViews names every column and binds eagerly, so a glob that rejects one
+// old file stops the process from starting.
+//
+// They differ entirely in cost. union_by_name derives the column set by opening
+// every file in the glob at bind and holding a reader and footer per file for
+// the statement's life, on the raw allocator where memory_limit cannot see it:
+// measured at 510 MiB per query against 3,281 live batches and 857 MiB against
+// 5,709, scaling at roughly 150 KB per file and paid again by every concurrent
+// binder. Pinning the schema opens nothing at bind, so the same query costs
+// tens of MiB and stops tracking how many batches happen to exist.
 func CreateParquetViews(db *sql.DB, parquetDir string) error {
 	if _, err := db.Exec(`CREATE SCHEMA IF NOT EXISTS telemetry`); err != nil {
 		return err
 	}
 	for _, signal := range []string{"spans", "logs", "metrics"} {
 		pattern := filepath.ToSlash(filepath.Join(parquetDir, "batches", "*.batch", signal+".parquet"))
-		projection := "*"
-		if signal == "spans" {
-			projection = "* EXCLUDE (_trace_hash)"
+		// _schema.batch is written from the current binary's row structs and is
+		// the definition of "every column this build knows about", so reading
+		// its schema keeps the view in lockstep with the writer instead of
+		// duplicating the column list here for someone to forget.
+		schemaFile := filepath.ToSlash(filepath.Join(parquetDir, "batches", "_schema.batch", signal+".parquet"))
+		columns, err := parquetSchemaMap(db, schemaFile, signal == "spans")
+		if err != nil {
+			return err
 		}
-		// union_by_name is load-bearing and must stay. ensureSchemaBatch writes
-		// an empty _schema.batch from the current binary's structs so the glob
-		// always carries the full column set, and this flag is what lets an
-		// older batch that predates an added column read as NULL instead of
-		// failing the whole glob with "schema mismatch in glob". Without it
-		// CreateViews, which names every column explicitly and binds eagerly,
-		// fails at NewDuck and the process will not start until every
-		// pre-deploy batch has aged out.
-		//
-		// It is expensive: bind opens every file in the glob and holds a reader
-		// and footer per file for the query's life, on the raw allocator, where
-		// memory_limit cannot see it. Measured on 3,281 live batches: 510 MiB
-		// per query against 89 MiB without, paid by every concurrent binder.
-		// That cost is bounded by keeping the live batch count down, which is
-		// compaction's job, not by removing this flag.
-		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT %s FROM read_parquet(%s, union_by_name=true)`, signal, projection, sqlLiteral(pattern))
+		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT * FROM read_parquet(%s, schema=%s)`, signal, sqlLiteral(pattern), columns)
 		if _, err := db.Exec(stmt); err != nil {
 			return fmt.Errorf("create parquet view telemetry.%s: %w", signal, err)
 		}
@@ -328,6 +332,37 @@ func CreateParquetViews(db *sql.DB, parquetDir string) error {
 	return nil
 }
 
+// parquetSchemaMap renders the schema argument for read_parquet from one file's
+// own schema. A column absent from a given file takes its default, which is how
+// an older batch survives a build that added a column.
+func parquetSchemaMap(db *sql.DB, file string, dropTraceHash bool) (string, error) {
+	rows, err := db.Query(fmt.Sprintf(`SELECT column_name, column_type FROM (DESCRIBE SELECT * FROM read_parquet(%s))`, sqlLiteral(file)))
+	if err != nil {
+		return "", fmt.Errorf("describe %s: %w", file, err)
+	}
+	defer rows.Close()
+	var entries []string
+	for rows.Next() {
+		var name, columnType string
+		if err := rows.Scan(&name, &columnType); err != nil {
+			return "", err
+		}
+		// _trace_hash is a physical sort key, not part of the telemetry schema.
+		if dropTraceHash && name == "_trace_hash" {
+			continue
+		}
+		entries = append(entries, fmt.Sprintf("%s: {'name': %s, 'type': %s, 'default_value': NULL}",
+			sqlLiteral(name), sqlLiteral(name), sqlLiteral(columnType)))
+	}
+	if err := rows.Err(); err != nil {
+		return "", err
+	}
+	if len(entries) == 0 {
+		return "", fmt.Errorf("no columns described for %s", file)
+	}
+	return "MAP{" + strings.Join(entries, ", ") + "}", nil
+}
+
 // CreateViews creates stable clean-name views plus the attr() macro.
 func CreateViews(db *sql.DB) error {
 	for _, stmt := range []string{macroAttr, viewSpans, viewLogs, viewMetrics} {
diff --git a/internal/query/views_schema_test.go b/internal/query/views_schema_test.go
new file mode 100644
index 00000000..e1dc874e
--- /dev/null
+++ b/internal/query/views_schema_test.go
@@ -0,0 +1,78 @@
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// The telemetry views must survive a build that adds a column: older batches
+// on disk lack it and must read as NULL rather than failing the glob.
+//
+// This is not a theoretical concern. CreateViews names every column explicitly
+// and DuckDB binds a view eagerly, so a glob that rejects one old file fails at
+// NewDuck and the process does not start at all -- it stays down until every
+// pre-deploy batch has aged out. An earlier attempt to cut the views' bind cost
+// by dropping union_by_name did exactly that, which is why this test exists
+// rather than a comment saying to be careful.
+func TestParquetViewsToleratePreDeploySchemas(t *testing.T) {
+	dir := t.TempDir()
+	batches := filepath.Join(dir, "batches")
+	for _, name := range []string{"_schema.batch", "old.batch", "new.batch"} {
+		if err := os.MkdirAll(filepath.Join(batches, name), 0o755); err != nil {
+			t.Fatal(err)
+		}
+	}
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	write := func(batch, signal, sel string) {
+		t.Helper()
+		path := filepath.ToSlash(filepath.Join(batches, batch, signal+".parquet"))
+		if _, err := db.Exec(fmt.Sprintf(`COPY (%s) TO '%s' (FORMAT PARQUET)`, sel, path)); err != nil {
+			t.Fatalf("write %s/%s: %v", batch, signal, err)
+		}
+	}
+	// The current build knows about added_later; the old batch predates it.
+	for _, signal := range []string{"spans", "logs", "metrics"} {
+		write("_schema.batch", signal, "SELECT 1 AS id, 'x' AS added_later WHERE false")
+		write("old.batch", signal, "SELECT 1 AS id")
+		write("new.batch", signal, "SELECT 2 AS id, 'present' AS added_later")
+	}
+
+	if err := CreateParquetViews(db, dir); err != nil {
+		t.Fatalf("CreateParquetViews must not fail on a pre-deploy batch: %v", err)
+	}
+
+	for _, signal := range []string{"spans", "logs", "metrics"} {
+		rows, err := db.Query(fmt.Sprintf(`SELECT id, added_later FROM telemetry.%s ORDER BY id`, signal))
+		if err != nil {
+			t.Fatalf("query telemetry.%s: %v", signal, err)
+		}
+		seen := map[int]bool{}
+		for rows.Next() {
+			var id int
+			var added sql.NullString
+			if err := rows.Scan(&id, &added); err != nil {
+				rows.Close()
+				t.Fatal(err)
+			}
+			seen[id] = true
+			if id == 1 && added.Valid {
+				t.Errorf("%s: the pre-deploy batch reported %q for a column it does not have", signal, added.String)
+			}
+			if id == 2 && added.String != "present" {
+				t.Errorf("%s: current batch read added_later=%q, want \"present\"", signal, added.String)
+			}
+		}
+		rows.Close()
+		if !seen[1] || !seen[2] {
+			t.Errorf("%s: read %v, want rows from both the old and the new batch", signal, seen)
+		}
+	}
+}
diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go
index 365e8c3c..706f522f 100644
--- a/internal/telemetry/parquet.go
+++ b/internal/telemetry/parquet.go
@@ -439,18 +439,35 @@ func (p *ParquetStore) RestoreRetiredInputs(inputs []string, replacementID strin
 // Trace reads only ranges selected by the persistent hash index. Scope filters
 // and the limit are applied while decoding so a pathological trace cannot grow
 // request memory without bound.
-func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSpan, error) {
+// TraceTotals describes the whole trace, not the page a limit admitted.
+//
+// These come free: readIndexedTrace already decodes every row of the trace and
+// applies the same predicates a separate aggregate would, and only the heap is
+// bounded by Limit. Deriving them here removes a second pass over every batch
+// file -- the thing that made trace_detail pay for its own correctness.
+type TraceTotals struct {
+	Spans        int
+	Services     int
+	DurationMS   float64
+	HasError     bool
+	MinStartNano int64
+	MaxEndNano   int64
+}
+
+func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSpan, TraceTotals, error) {
+	var totals TraceTotals
+	services := make(map[string]struct{}, 8)
 	if query.TraceID == "" {
-		return nil, nil
+		return nil, totals, nil
 	}
 	if query.Limit <= 0 {
-		return nil, errors.New("trace query limit must be positive")
+		return nil, totals, errors.New("trace query limit must be positive")
 	}
 	if query.Limit > maxTraceQueryResults {
-		return nil, fmt.Errorf("trace query limit exceeds %d", maxTraceQueryResults)
+		return nil, totals, fmt.Errorf("trace query limit exceeds %d", maxTraceQueryResults)
 	}
 	if query.StartNanos >= query.EndNanos {
-		return nil, errors.New("trace query time range must be positive")
+		return nil, totals, errors.New("trace query time range must be positive")
 	}
 	hash := xxh3.HashString(query.TraceID)
 	p.mu.RLock()
@@ -462,7 +479,7 @@ func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSp
 	selected := make(indexedSpanHeap, 0, query.Limit)
 	for _, batch := range batches {
 		if err := ctx.Err(); err != nil {
-			return nil, err
+			return nil, totals, err
 		}
 		if batch.metadata.MaxSpanStartNanos > 0 &&
 			(batch.metadata.MaxSpanStartNanos < query.StartNanos || batch.metadata.MinSpanStartNanos >= query.EndNanos) {
@@ -470,23 +487,27 @@ func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSp
 		}
 		match, found, err := batch.traces.Lookup(hash)
 		if err != nil {
-			return nil, err
+			return nil, totals, err
 		}
 		if !found {
 			continue
 		}
-		if err := readIndexedTrace(ctx, batch, match, query, &selected); err != nil {
-			return nil, err
+		if err := readIndexedTrace(ctx, batch, match, query, &selected, &totals, services); err != nil {
+			return nil, totals, err
 		}
 	}
+	totals.Services = len(services)
+	if totals.Spans > 0 && totals.MaxEndNano > totals.MinStartNano {
+		totals.DurationMS = float64(totals.MaxEndNano-totals.MinStartNano) / float64(time.Millisecond)
+	}
 	out := []IndexedSpan(selected)
 	sort.Slice(out, func(i, j int) bool {
 		return indexedSpanEarlier(out[i], out[j])
 	})
-	return out, nil
+	return out, totals, nil
 }
 
-func readIndexedTrace(ctx context.Context, batch *storedBatch, match traceRange, query TraceQuery, selected *indexedSpanHeap) (err error) {
+func readIndexedTrace(ctx context.Context, batch *storedBatch, match traceRange, query TraceQuery, selected *indexedSpanHeap, totals *TraceTotals, services map[string]struct{}) (err error) {
 	if match.row > uint64(math.MaxInt64) {
 		return errors.New("trace index row exceeds Parquet reader limit")
 	}
@@ -522,6 +543,21 @@ func readIndexedTrace(ctx context.Context, batch *storedBatch, match traceRange,
 				(query.Namespace != "" && row.Namespace != query.Namespace) {
 				continue
 			}
+			// Counted before the heap, so the totals describe the trace while
+			// the heap keeps describing the page.
+			totals.Spans++
+			if row.Service != "" {
+				services[row.Service] = struct{}{}
+			}
+			if strings.Contains(strings.ToUpper(row.Status), "ERROR") {
+				totals.HasError = true
+			}
+			if totals.MinStartNano == 0 || row.StartUnixNano < totals.MinStartNano {
+				totals.MinStartNano = row.StartUnixNano
+			}
+			if end := row.StartUnixNano + int64(row.DurationMS*float64(time.Millisecond)); end > totals.MaxEndNano {
+				totals.MaxEndNano = end
+			}
 			selected.Add(row.span(), query.Limit)
 		}
 		remaining -= uint64(n)
diff --git a/internal/telemetry/parquet_test.go b/internal/telemetry/parquet_test.go
index 17e4d80c..bdd07fc2 100644
--- a/internal/telemetry/parquet_test.go
+++ b/internal/telemetry/parquet_test.go
@@ -163,7 +163,7 @@ func TestParquetStoreTraceFiltersAndBoundsResultsDuringRead(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	got, err := store.Trace(context.Background(), TraceQuery{
+	got, _, err := store.Trace(context.Background(), TraceQuery{
 		TraceID: "large-trace", Namespace: "prod", StartNanos: 10, EndNanos: 90, Limit: 3,
 	})
 	if err != nil {
@@ -226,7 +226,7 @@ func TestParquetStoreSkipsTraceIndexesOutsideTimeWindow(t *testing.T) {
 	if err := os.Remove(filepath.Join(store.BatchPath("old-traces"), "trace.fidx")); err != nil {
 		t.Fatal(err)
 	}
-	got, err := store.Trace(context.Background(), TraceQuery{
+	got, _, err := store.Trace(context.Background(), TraceQuery{
 		TraceID: "wanted", StartNanos: 1_000, EndNanos: 2_000, Limit: 10,
 	})
 	if err != nil {
@@ -502,9 +502,10 @@ func completeTestSpan() Span {
 }
 
 func traceAll(store *ParquetStore, traceID string) ([]IndexedSpan, error) {
-	return store.Trace(context.Background(), TraceQuery{
+	spans, _, err := store.Trace(context.Background(), TraceQuery{
 		TraceID: traceID, StartNanos: math.MinInt64, EndNanos: math.MaxInt64, Limit: maxTraceQueryResults,
 	})
+	return spans, err
 }
 
 func readOneParquetRow[T any](t *testing.T, path string) T {
diff --git a/internal/telemetry/store/repository_test.go b/internal/telemetry/store/repository_test.go
index 43b2a1b1..e63dae70 100644
--- a/internal/telemetry/store/repository_test.go
+++ b/internal/telemetry/store/repository_test.go
@@ -689,9 +689,10 @@ func openTestDuckDB(t *testing.T) *sql.DB {
 }
 
 func traceAll(repository *Repository, traceID string) ([]telemetry.IndexedSpan, error) {
-	return repository.Parquet.Trace(context.Background(), telemetry.TraceQuery{
+	spans, _, err := repository.Parquet.Trace(context.Background(), telemetry.TraceQuery{
 		TraceID: traceID, StartNanos: -1 << 63, EndNanos: 1<<63 - 1, Limit: 500,
 	})
+	return spans, err
 }
 
 // seedFailedCompaction leaves a live COMPACTION.json whose recovery always

From cbec5e2cff75bdb99a454dbe7b9f9b86d1888959 Mon Sep 17 00:00:00 2001
From: Vishal Rana <v@labstack.com>
Date: Mon, 21 Sep 2026 09:48:47 -0700
Subject: [PATCH 4/6] fix(store): stop compaction stranding every day but the
 current one

selectBoundedCompactionGroup returned a group only when it was full at
maxBatches or when a row or byte ceiling stopped it. A day that had
finished being written was neither: its group never grows again, so it
never reaches the ceiling and never saturates, and nothing ever compacted
it.

Running the real selection against the live box's metadata showed the
scale of it: 1,243 batches in 69 (day, generation) groups spanning 25
days, and every single group returned no candidate. Only the current day
ever reached 128 files, so every compaction output was generation 1 and
every older day was untouchable. Live file count -- which every query pays
per-file overhead on -- held at an equilibrium around 1,100 that no
reordering or larger budget could move.

The function's own comment warns about stranding, but its escape hatch
only covers groups too LARGE for a ceiling. A group merely too small to
fill one had no way out.

An underfilled group is now compactable once its day is finished, meaning
a newer day exists in the set. Holding out for a full group stays correct
while a day is still being written -- more batches are coming and merging
early wastes the work -- and stops being correct the moment it is not.

Measured on the demo: 1,271 -> 224 batches within twenty minutes and
holding, with generation 2 falling 291 -> 27 and generation 3 rising
3 -> 22 as promotion resumed. Compaction output rate went from roughly
2.6 to 11 merged directories per minute. RSS 0.6-1.5 GiB, no errors.

Two changes here did not fix it and are kept only on their own merits.
The compaction byte ceiling moves 256 -> 768 MiB because merge cost is now
measured rather than guessed: four inputs totalling 442 MiB on disk merged
at +413 MiB RSS, so the budget tracks on-disk bytes almost exactly, and
768 MiB sits well inside the headroom the process now has. Generation
ordering within a day inverts to highest-first because lowest-first cannot
help starving the levels above it once ingest refills generation 0 -- true
independently, though it was not what stalled this box.
---
 internal/telemetry/store/compaction.go        |  87 +++++++++++++-
 .../telemetry/store/compaction_bytes_test.go  |   4 +-
 .../store/compaction_starvation_test.go       | 107 ++++++++++++++++++
 3 files changed, 191 insertions(+), 7 deletions(-)
 create mode 100644 internal/telemetry/store/compaction_starvation_test.go

diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go
index c10f5062..b1bc96c1 100644
--- a/internal/telemetry/store/compaction.go
+++ b/internal/telemetry/store/compaction.go
@@ -16,8 +16,26 @@ import (
 )
 
 const (
-	maxCompactionRows  = 25_000_000
-	maxCompactionBytes = 256 << 20
+	maxCompactionRows = 25_000_000
+
+	// maxCompactionBytes bounds a pass by the bytes it admits, because that is
+	// what a merge costs: parquet-go opens a cursor per row group across every
+	// input and holds each one's dictionaries at once. Measured on real batches,
+	// peak RSS tracks the on-disk input bytes almost exactly -- 4 inputs
+	// totalling 442 MiB on disk merged at +413 MiB RSS.
+	//
+	// Row counts cannot stand in for it: wide spans carrying large attribute
+	// payloads and bare log lines differ by an order of magnitude at equal row
+	// counts.
+	//
+	// The figure is a memory budget, so it also decides how fast compaction can
+	// consolidate. Set too low it throttles the cascade: at 256 MiB a pass
+	// admitted only ~16 of the 16 MB generation-2 batches instead of the 128 the
+	// count ceiling allows, and 290 of them piled up unpromoted. 768 MiB keeps
+	// merge peak well inside the headroom the process now has -- it sits at
+	// 1.5 GiB against a 4 GB DuckDB budget -- while admitting enough inputs per
+	// pass for the higher generations to drain.
+	maxCompactionBytes = 768 << 20
 
 	// assumedCompactionBytesPerRow prices a batch whose size was never
 	// measured. Measured batches run well under this -- span rows with JSON
@@ -249,15 +267,46 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches int)
 		key := compactionKey{day: batch.MaxIngestedNanos / int64(24*time.Hour), generation: batch.Generation}
 		groups[key] = append(groups[key], batch)
 	}
+	// Within a day, take the HIGHEST generation that has a compactable group,
+	// not the lowest.
+	//
+	// Lowest-first starves everything above it. Ingest refills generation 0
+	// continuously, so if it always wins no later generation is ever selected,
+	// while each pass it does run adds one more batch to the generation above.
+	// Observed on the live box as 508 / 371 / 290 batches at generations
+	// 0 / 1 / 2, generation 2 untouched for hours, and the live file count
+	// settling at an equilibrium instead of falling. That count is what every
+	// query pays per-file overhead on, so an equilibrium there is a permanent
+	// tax rather than a backlog that clears.
+	//
+	// Highest-first cannot starve anything. A generation above 0 only grows by
+	// one batch per pass of the generation below it, so its group fills slowly
+	// and empties in a few passes; once it no longer has a full group it stops
+	// being a candidate and generation 0 -- which is almost always ready --
+	// takes every remaining pass. Reducing the count by a group is worth the
+	// same wherever it happens, because the cost being reduced is per file.
+	//
+	// The older day still wins outright. Reclaiming space retention is about to
+	// drop matters more than consolidating today, and a day no longer being
+	// written finishes and stays finished.
+	newestDay := int64(math.MinInt64)
+	for key := range groups {
+		if key.day > newestDay {
+			newestDay = key.day
+		}
+	}
 	var chosen compactionKey
 	var selected []telemetry.BatchMetadata
 	found := false
 	for key, group := range groups {
-		candidate := selectBoundedCompactionGroup(group, maxBatches)
+		candidate := selectBoundedCompactionGroup(group, maxBatches, key.day < newestDay)
 		if len(candidate) < 2 {
 			continue
 		}
-		if !found || key.day < chosen.day || key.day == chosen.day && key.generation < chosen.generation {
+		better := !found ||
+			key.day < chosen.day ||
+			key.day == chosen.day && key.generation > chosen.generation
+		if better {
 			chosen, selected, found = key, candidate, true
 		}
 	}
@@ -271,7 +320,7 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches int)
 // small files, but admits a smaller group when the row ceiling fills first.
 // Without the latter, one successful generation can make every later group too
 // large for the ceiling and permanently strand those files.
-func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches int) []telemetry.BatchMetadata {
+func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches int, finished bool) []telemetry.BatchMetadata {
 	ordered := append([]telemetry.BatchMetadata(nil), group...)
 	sort.Slice(ordered, func(i, j int) bool {
 		left, right := compactionBatchRows(ordered[i]), compactionBatchRows(ordered[j])
@@ -326,9 +375,37 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in
 	if bytes == maxCompactionBytes {
 		saturated = true
 	}
+	// A group is worth merging when it is full, when a ceiling stopped it, or
+	// when it simply holds enough files to be worth the merge.
+	//
+	// That last case is the one this was missing. The escape hatch above only
+	// covered groups too LARGE for a ceiling; a group merely too SMALL to reach
+	// maxBatches was stranded forever. Days stop being written -- the group for
+	// a past day never grows again -- so on the live box 1,243 batches sat in 69
+	// (day, generation) groups across 25 days and not one produced a candidate.
+	// Only the current day ever reached 128 files, so every output was
+	// generation 1 and everything older was untouchable. The live file count,
+	// which every query pays per-file overhead on, held at an equilibrium that
+	// no reordering or budget change could move.
+	// A group is worth merging when it is full, when a ceiling stopped it, or
+	// when it belongs to a day that has finished and already holds enough files
+	// to be worth the merge.
+	//
+	// Holding out for a full group is right while a day is still being written:
+	// more batches are coming, and merging early wastes the work. It is wrong
+	// once the day is over, because the group will never grow again. On the
+	// live box that stranded almost everything -- 1,243 batches in 69
+	// (day, generation) groups across 25 days, not one of them a candidate,
+	// because only the current day ever reached 128 files. Every compaction
+	// output was generation 1 and every older day was untouchable, which is why
+	// the live file count held at an equilibrium that neither reordering nor a
+	// larger byte budget could move.
 	if len(candidate) == maxBatches || saturated && len(candidate) >= 2 {
 		return candidate
 	}
+	if finished && len(candidate) >= minCompactionInputs {
+		return candidate
+	}
 	return nil
 }
 
diff --git a/internal/telemetry/store/compaction_bytes_test.go b/internal/telemetry/store/compaction_bytes_test.go
index 868b4cfb..294bfa1c 100644
--- a/internal/telemetry/store/compaction_bytes_test.go
+++ b/internal/telemetry/store/compaction_bytes_test.go
@@ -24,7 +24,7 @@ func TestSelectBoundedCompactionGroupStopsAtTheByteBudget(t *testing.T) {
 		})
 	}
 
-	candidate := selectBoundedCompactionGroup(group, 128)
+	candidate := selectBoundedCompactionGroup(group, 128, false)
 
 	if len(candidate) == 0 {
 		t.Fatal("no group selected; the byte budget must still admit a mergeable group")
@@ -58,7 +58,7 @@ func TestSelectBoundedCompactionGroupChargesUnsizedBatchesAnEstimate(t *testing.
 		})
 	}
 
-	candidate := selectBoundedCompactionGroup(group, 128)
+	candidate := selectBoundedCompactionGroup(group, 128, false)
 
 	if len(candidate) == 0 {
 		t.Fatal("no group selected; unsized batches must still be compactable")
diff --git a/internal/telemetry/store/compaction_starvation_test.go b/internal/telemetry/store/compaction_starvation_test.go
new file mode 100644
index 00000000..28f70953
--- /dev/null
+++ b/internal/telemetry/store/compaction_starvation_test.go
@@ -0,0 +1,107 @@
+package store
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/labstack/fanout/internal/telemetry"
+)
+
+func genBatches(day int64, generation uint32, n int, rows int) []telemetry.BatchMetadata {
+	out := make([]telemetry.BatchMetadata, 0, n)
+	for i := range n {
+		out = append(out, telemetry.BatchMetadata{
+			ID:               fmt.Sprintf("g%d-%04d", generation, i),
+			Generation:       generation,
+			Spans:            rows,
+			Bytes:            int64(rows) * 100,
+			MaxIngestedNanos: day*int64(24*time.Hour) + int64(i),
+			MinIngestedNanos: day*int64(24*time.Hour) + int64(i),
+		})
+	}
+	return out
+}
+
+// Ingest replenishes generation 0 continuously, so a rule that always prefers
+// the lowest generation never reaches the higher ones. Observed in production:
+// generation 0 and 1 were compacting while 290 generation-2 batches sat
+// untouched for hours, and the live file count -- which is what every query
+// pays per-file overhead on -- settled at an equilibrium instead of falling.
+func TestSelectCompactionBatchesDoesNotStarveHigherGenerations(t *testing.T) {
+	// The shape of the stall, taken from the live box: generation 0 replenished
+	// by ingest, generation 2 piled up above it. Both are past the count ceiling,
+	// so both are valid candidates and the tie-break decides.
+	batches := append(genBatches(1, 0, 508, 1_000), genBatches(1, 2, 290, 200_000)...)
+
+	selected := selectCompactionBatches(batches, 128)
+	if len(selected) == 0 {
+		t.Fatal("selected nothing with two compactable groups available")
+	}
+	if selected[0].Generation != 2 {
+		t.Errorf("selected generation %d when generation 2 has 290 batches waiting: "+
+			"a rule that always takes the lowest generation lets ingest starve every generation above it",
+			selected[0].Generation)
+	}
+}
+
+// The older day still wins: reclaiming disk that retention will drop next
+// matters more than consolidating today's files.
+func TestSelectCompactionBatchesStillPrefersTheOlderDay(t *testing.T) {
+	batches := append(genBatches(1, 0, 200, 1_000), genBatches(5, 2, 290, 200_000)...)
+
+	selected := selectCompactionBatches(batches, 128)
+	if len(selected) == 0 {
+		t.Fatal("selected nothing")
+	}
+	if selected[0].MaxIngestedNanos/int64(24*time.Hour) != 1 {
+		t.Errorf("selected day %d, want the older day 1 even though day 5 has more files",
+			selected[0].MaxIngestedNanos/int64(24*time.Hour))
+	}
+}
+
+// Generation 0 must still be served: it is refilled constantly, so once the
+// higher generations no longer have a full group it should take every pass.
+func TestSelectCompactionBatchesServesGenerationZeroWhenNothingIsPiledAbove(t *testing.T) {
+	// Only generation 0 has enough to form a group.
+	batches := append(genBatches(1, 0, 200, 1_000), genBatches(1, 2, 3, 1_000)...)
+
+	selected := selectCompactionBatches(batches, 128)
+	if len(selected) == 0 {
+		t.Fatal("selected nothing with a full generation-0 group available")
+	}
+	if selected[0].Generation != 0 {
+		t.Errorf("selected generation %d, want 0: with nothing compactable above it, the newest generation takes the pass",
+			selected[0].Generation)
+	}
+}
+
+// A day that has stopped being written still has to consolidate. Its group
+// never reaches the count ceiling and never saturates a row or byte ceiling, so
+// a rule that admits only full or saturated groups strands it forever.
+//
+// Measured on the live box: 1,243 batches spread over 69 (day, generation)
+// groups covering 25 days, every one of them returning no candidate. Only the
+// current day ever reached 128 files, so every compaction output was
+// generation 1 and everything older sat untouched -- which is why the live file
+// count held at an equilibrium no amount of reordering or budget-raising moved.
+func TestSelectBoundedCompactionGroupCompactsAnUnderfilledDay(t *testing.T) {
+	// A finished day: well under the count ceiling, nowhere near saturating.
+	group := genBatches(3, 1, 56, 5_000)
+
+	candidate := selectBoundedCompactionGroup(group, 128, true)
+
+	if len(candidate) == 0 {
+		t.Fatal("no candidate for a 56-batch day: a day that will never grow to 128 files can never consolidate")
+	}
+	if len(candidate) != 56 {
+		t.Errorf("candidate = %d, want all 56: nothing here forces a smaller group", len(candidate))
+	}
+}
+
+// Still not worth waking the merge for a couple of files.
+func TestSelectBoundedCompactionGroupIgnoresATrivialDay(t *testing.T) {
+	if got := selectBoundedCompactionGroup(genBatches(3, 1, 3, 5_000), 128, true); len(got) != 0 {
+		t.Errorf("candidate = %d for a 3-batch day, want none: below the effort floor", len(got))
+	}
+}

From 513605b163aa629234fe90a30b4dfaadef48b4c2 Mon Sep 17 00:00:00 2001
From: Vishal Rana <v@labstack.com>
Date: Mon, 21 Sep 2026 11:07:15 -0700
Subject: [PATCH 5/6] fix(query,telemetry,metrics): correct defects found in
 review of this branch

_schema.batch was written once, on the first boot into an empty data
directory, and never revisited. Every deployment since read the column
list of whichever build created the directory -- the demo's file was
dated 2026-08-28, 24 days stale. Since internal/query derives every
view's read_parquet(schema=MAP{...}) from it, a build that added a
column to spanParquetRow could not start at all: the view would not
expose the column, CreateViews would fail to bind it, NewDuck would
return an error, and restarting would not help because the stale file
was still there. It is now rewritten on every open. The files are
empty, so this costs three small writes per boot.

The edge rollup's FROM clause read "FROM spans child, bounds JOIN
parent_scope ON child...". The comma binds looser than JOIN, so that
parses as "spans child" comma "(bounds JOIN parent_scope ON ...)" and
DuckDB materialises a CROSS_PRODUCT of the child scan with bounds
rather than folding the single bounds row into the filters. bounds now
joins last as an explicit CROSS JOIN: 14ms against 21ms on the fixture
in TestEdgeRollupPlanUsesHashJoins, which asserts the operator is
absent here and present in the comma form.

That test is also the first thing to execute edgeRollupInsertSQL at
all, which is why neither of this statement's shape defects was ever
caught: every form returns identical rows. Two comments in it claimed
a NESTED_LOOP_JOIN and a 45 GiB spill that do not reproduce -- the
earlier probe behind them was a simplified query, not this one. They
now state what is measured: the cost is the size of the hash build
side, not the join algorithm.

The per-tag DuckDB memory gauge was a GaugeVec that a scrape hook
Reset and refilled. That is shared mutable state across concurrent
scrapes, and a scrape landing mid-refill exported a subset of the tags
-- reproduced in TestDuckDBMemoryIsSafeUnderConcurrentScrapes, which
saw 1 or 2 of 3. It also left the last good values in place when a
read failed, reporting them as current, so the number meant to reveal
an impending out-of-memory kill read normal. A Collector has neither
problem, and it drops the fanout_duckdb_memory_scrape_total gauge that
existed only to pump it.

Also moves four comments to what they describe and deletes a
superseded duplicate in compaction.go.
---
 internal/metrics/duckdb_memory_test.go        | 110 +++++++++
 internal/metrics/metrics.go                   |  71 +++---
 internal/observability/trace.go               |   5 +-
 internal/query/duck.go                        |  38 +--
 internal/query/edge_rollup_plan_test.go       | 224 ++++++++++++++++++
 .../query/testdata/edge_rollup_pre_fix.sql    | 126 ++++++++++
 internal/query/views.go                       |  11 +-
 internal/telemetry/parquet.go                 |  24 +-
 internal/telemetry/schema_batch_test.go       |  89 +++++++
 internal/telemetry/store/compaction.go        |  12 -
 10 files changed, 637 insertions(+), 73 deletions(-)
 create mode 100644 internal/metrics/duckdb_memory_test.go
 create mode 100644 internal/query/edge_rollup_plan_test.go
 create mode 100644 internal/query/testdata/edge_rollup_pre_fix.sql
 create mode 100644 internal/telemetry/schema_batch_test.go

diff --git a/internal/metrics/duckdb_memory_test.go b/internal/metrics/duckdb_memory_test.go
new file mode 100644
index 00000000..4ac7ea5b
--- /dev/null
+++ b/internal/metrics/duckdb_memory_test.go
@@ -0,0 +1,110 @@
+package metrics
+
+import (
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+)
+
+// gatherDuckDBMemory returns the exported per-tag memory series, by tag.
+func gatherDuckDBMemory(t *testing.T) map[string]float64 {
+	t.Helper()
+	families, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := map[string]float64{}
+	for _, family := range families {
+		if family.GetName() != "fanout_duckdb_memory_bytes" {
+			continue
+		}
+		for _, metric := range family.GetMetric() {
+			for _, label := range metric.GetLabel() {
+				if label.GetName() == "tag" {
+					out[label.GetValue()] = metric.GetGauge().GetValue()
+				}
+			}
+		}
+	}
+	return out
+}
+
+// The whole purpose of this series is to measure what DuckDB holds right now,
+// so that the untracked gap against process RSS can be derived from it. A
+// failed read must therefore export nothing. Reporting the last successful
+// figures as if they were current is the one outcome that actively misleads:
+// the gap is computed against a stale sum, and the number that was supposed to
+// reveal an impending out-of-memory kill reads normal.
+func TestDuckDBMemoryDropsSeriesWhenTheReadFails(t *testing.T) {
+	release := SetDuckDBMemorySource(func() map[string]int64 {
+		return map[string]int64{"BASE_TABLE": 4096}
+	})
+	if got := gatherDuckDBMemory(t); got["BASE_TABLE"] != 4096 {
+		release()
+		t.Fatalf("healthy read exported %v, want BASE_TABLE=4096", got)
+	}
+	release()
+
+	// The source is installed but cannot read: the closure in internal/query
+	// returns nil for a query error, a scan error, or a partial read.
+	release = SetDuckDBMemorySource(func() map[string]int64 { return nil })
+	defer release()
+	if got := gatherDuckDBMemory(t); len(got) != 0 {
+		t.Errorf("after a failed read the gauge still exports %v; stale values are reported as current", got)
+	}
+}
+
+// Prometheus scrapes concurrently, and a GaugeVec that is Reset and refilled on
+// every scrape is shared mutable state: two scrapes interleave, and one of them
+// observes the other's half-finished refill -- missing tags, and a sum that
+// understates what DuckDB holds.
+func TestDuckDBMemoryIsSafeUnderConcurrentScrapes(t *testing.T) {
+	release := SetDuckDBMemorySource(func() map[string]int64 {
+		return map[string]int64{"BASE_TABLE": 1, "HASH_TABLE": 2, "ORDER_BY": 3}
+	})
+	defer release()
+
+	var wg sync.WaitGroup
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range 25 {
+				families, err := prometheus.DefaultGatherer.Gather()
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				for _, family := range families {
+					if family.GetName() != "fanout_duckdb_memory_bytes" {
+						continue
+					}
+					if n := len(family.GetMetric()); n != 3 {
+						t.Errorf("scrape saw %d tags, want 3: a concurrent refill was observed half-done", n)
+						return
+					}
+					_ = dto.MetricType_GAUGE
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
+
+// The gauge must not invent a series when nothing has installed a source,
+// because an empty family reads exactly like "DuckDB is holding nothing".
+func TestDuckDBMemoryExportsNothingWithoutASource(t *testing.T) {
+	if got := gatherDuckDBMemory(t); len(got) != 0 {
+		t.Errorf("with no source installed the gauge exports %v", got)
+	}
+	if families, err := prometheus.DefaultGatherer.Gather(); err == nil {
+		for _, family := range families {
+			if strings.Contains(family.GetName(), "duckdb_memory_scrape") {
+				t.Errorf("%s exists only to pump a push gauge; a collector does not need it", family.GetName())
+			}
+		}
+	}
+}
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index b33cf95b..57fbf856 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -297,10 +297,10 @@ func RecordIngest(signal string, count int) {
 	IngestTotal.WithLabelValues(signal).Add(float64(count))
 }
 
-// RecordFlush records a flush event
 // RecordIngestShed counts one refused telemetry request.
 func RecordIngestShed() { IngestShedTotal.Inc() }
 
+// RecordFlush records a flush event.
 func RecordFlush(signal string, durationSec float64) {
 	FlushTotal.WithLabelValues(signal).Inc()
 	FlushDuration.WithLabelValues(signal).Observe(durationSec)
@@ -314,12 +314,6 @@ func RecordFlush(signal string, durationSec float64) {
 // on the second one. Unset, every gauge reads zero.
 var duckDBPoolStats atomic.Pointer[func() sql.DBStats]
 
-// SetDuckDBPoolSource points the connection-pool gauges at a pool and returns
-// the function that stops reading it. The last caller wins; the returned
-// release only clears the source if it is still the one it installed, so a
-// second Duck closing in a test binary cannot blank the gauges of one that is
-// still serving.
-
 // duckDBMemoryStats is the live source for the per-tag DuckDB memory gauge.
 //
 // memory_limit bounds DuckDB's buffer pool and nothing else, so the figure that
@@ -341,45 +335,48 @@ func SetDuckDBMemorySource(tags func() map[string]int64) (release func()) {
 	return func() { duckDBMemoryStats.CompareAndSwap(installed, nil) }
 }
 
-// duckDBMemoryTags is published by a pull-based collector rather than a gauge
-// somebody has to remember to refresh. A push gauge with no pump exports an
-// empty metric family, which looks exactly like "DuckDB is holding nothing" --
-// the most misleading possible reading for a series whose entire purpose is
-// measuring what DuckDB holds.
-var duckDBMemoryTags = promauto.NewGaugeVec(prometheus.GaugeOpts{
-	Name: "fanout_duckdb_memory_bytes",
-	Help: "DuckDB memory usage by tag, as DuckDB itself accounts for it",
-}, []string{"tag"})
+// duckDBMemoryDesc is emitted by a Collector, not held in a GaugeVec.
+//
+// The obvious implementation -- a GaugeVec that a scrape hook Resets and
+// refills -- is shared mutable state across concurrent scrapes, and a scrape
+// that lands mid-refill exports a subset of the tags. That understates what
+// DuckDB holds and correspondingly overstates the untracked gap, which is the
+// single number this series exists to compute. It also leaves the last good
+// values in place when a read fails, reporting them as current: the reading
+// that was meant to reveal an impending out-of-memory kill then looks normal.
+//
+// A Collector has neither problem. Each scrape builds its own metrics from its
+// own read, and a read that returns nothing exports nothing -- the series goes
+// absent, which is visibly different from "DuckDB is holding nothing".
+var duckDBMemoryDesc = prometheus.NewDesc(
+	"fanout_duckdb_memory_bytes",
+	"DuckDB memory usage by tag, as DuckDB itself accounts for it",
+	[]string{"tag"}, nil,
+)
 
-func init() {
-	prometheus.DefaultRegisterer.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
-		Name: "fanout_duckdb_memory_scrape_total",
-		Help: "Refreshes of the per-tag DuckDB memory gauge",
-	}, func() float64 {
-		refreshDuckDBMemory()
-		return 1
-	}))
-}
+type duckDBMemoryCollector struct{}
+
+func (duckDBMemoryCollector) Describe(ch chan<- *prometheus.Desc) { ch <- duckDBMemoryDesc }
 
-// refreshDuckDBMemory republishes the per-tag gauge from the installed source.
-// Reset first: a tag that stops appearing must stop being reported, or the sum
-// overstates what DuckDB holds and understates the untracked gap, which is the
-// one number this exists to compute.
-func refreshDuckDBMemory() {
+func (duckDBMemoryCollector) Collect(ch chan<- prometheus.Metric) {
 	fn := duckDBMemoryStats.Load()
 	if fn == nil {
 		return
 	}
-	tags := (*fn)()
-	if tags == nil {
-		return
-	}
-	duckDBMemoryTags.Reset()
-	for tag, bytes := range tags {
-		duckDBMemoryTags.WithLabelValues(tag).Set(float64(bytes))
+	for tag, bytes := range (*fn)() {
+		ch <- prometheus.MustNewConstMetric(duckDBMemoryDesc, prometheus.GaugeValue, float64(bytes), tag)
 	}
 }
 
+func init() {
+	prometheus.DefaultRegisterer.MustRegister(duckDBMemoryCollector{})
+}
+
+// SetDuckDBPoolSource points the connection-pool gauges at a pool and returns
+// the function that stops reading it. The last caller wins; the returned
+// release only clears the source if it is still the one it installed, so a
+// second Duck closing in a test binary cannot blank the gauges of one that is
+// still serving.
 func SetDuckDBPoolSource(stats func() sql.DBStats) (release func()) {
 	if stats == nil {
 		return func() {}
diff --git a/internal/observability/trace.go b/internal/observability/trace.go
index ff3cc705..745a1467 100644
--- a/internal/observability/trace.go
+++ b/internal/observability/trace.go
@@ -69,7 +69,10 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 		}
 
 		// Only the page's own service list is derived here. Duration and
-		// has_error describe the whole trace and come from the aggregate below.
+		// has_error describe the whole trace, so they come from the totals the
+		// repository accumulated over every span it walked -- not from this
+		// page, which is why trace_detail used to report the page's span count
+		// and duration as the trace's.
 		serviceSet := make(map[string]struct{})
 		for _, span := range data.Spans {
 			if span.Service != "" {
diff --git a/internal/query/duck.go b/internal/query/duck.go
index 3271b008..aa23c8dd 100644
--- a/internal/query/duck.go
+++ b/internal/query/duck.go
@@ -1658,22 +1658,21 @@ bounds AS (
   SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
 ),
 -- The parent side is range-filtered HERE rather than in the join's ON
--- clause, and that placement is the whole cost of this statement.
+-- clause, which keeps the hash build off every span ever ingested. The
+-- affected window is one minute of newly ingested spans; the table behind
+-- this view is the whole retention period, so the two differ by orders of
+-- magnitude and the ON-clause form pays for the difference on every pass.
 --
--- Mixing an equality with an inequality in one ON clause leaves DuckDB
--- nothing to hash the range on, so it plans the whole thing as a
--- NESTED_LOOP_JOIN: spans x spans, quadratic, inside a window every other
--- bound in this file has already made small. Measured on one stuck pass,
--- the ON-clause form spilled 45 GiB and hit max_temp_directory_size; this
--- form returns the same rows in 30s with zero spill. No row or time bound
--- can catch that, because the window really was small -- the plan was not.
+-- Measured (TestEdgeRollupPlanUsesHashJoins, spans as a Parquet-backed view,
+-- 600k spans over 25 days, one affected minute): 14ms here against 33ms with
+-- the range predicate in the ON clause. Both plan a HASH_JOIN -- the cost is
+-- the size of the build side, not the join algorithm.
 --
--- Bounding the parent side to the affected bucket range ±1h keeps the hash
--- build off every span ever ingested. Parents further out are dropped,
--- which is acceptable for minute-bucket dependency edges. Caveat retained
--- from the original: buckets come from span start_time while the window
--- bounds ingested_unix_nano, so one late-arriving span with an old
--- start_time widens this range for the pass that ingests it.
+-- Parents more than 1h outside the affected bucket range are dropped, which
+-- is acceptable for minute-bucket dependency edges. Caveat retained from the
+-- original: buckets come from span start_time while the window bounds
+-- ingested_unix_nano, so one late-arriving span with an old start_time
+-- widens this range for the pass that ingests it.
 parent_scope AS (
   SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service
   FROM spans parent, bounds
@@ -1692,7 +1691,7 @@ call_edges AS (
     AVG(child.duration_ms) AS avg_ms,
     AVG(CASE WHEN child.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate,
     'call' AS edge_type
-  FROM spans child, bounds
+  FROM spans child
   JOIN parent_scope parent
     ON child.parent_span_id = parent.span_id
    AND child.trace_id = parent.trace_id
@@ -1700,6 +1699,15 @@ call_edges AS (
   JOIN affected a
     ON a.namespace = child.namespace
    AND a.bucket = date_trunc('minute', child.start_time)
+  -- bounds joins last, and as an explicit CROSS JOIN. Written as
+  -- "FROM spans child, bounds JOIN parent_scope ON child...", the comma binds
+  -- looser than JOIN: that parses as "spans child" comma "(bounds JOIN
+  -- parent_scope ON ...)", so DuckDB materialises a CROSS_PRODUCT of the child
+  -- scan with bounds instead of folding the single bounds row into the filters.
+  -- Same rows either way, so nothing fails and no row assertion notices; on the
+  -- fixture above it is 21ms against 14ms. TestEdgeRollupPlanUsesHashJoins
+  -- asserts the CROSS_PRODUCT is absent here and present in the comma form.
+  CROSS JOIN bounds
   WHERE child.service IS NOT NULL
     AND child.service != ''
     AND parent.service != child.service
diff --git a/internal/query/edge_rollup_plan_test.go b/internal/query/edge_rollup_plan_test.go
new file mode 100644
index 00000000..9db7f691
--- /dev/null
+++ b/internal/query/edge_rollup_plan_test.go
@@ -0,0 +1,224 @@
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+// The edge rollup is the most expensive statement this process issues, and
+// until this test nothing executed it. Both of its shape defects were therefore
+// invisible: every form returns identical rows, so a row assertion passes on
+// all of them, and the only symptom was a pass that did not finish inside its
+// budget.
+//
+// The plan is the behaviour under test. This asserts it in both directions --
+// a guard that passes on the defect it was written for is worse than none --
+// and checks that the forms agree on rows, which is what makes the rewrite a
+// performance change rather than a behaviour one.
+func TestEdgeRollupPlanUsesHashJoins(t *testing.T) {
+	db := openTestDuck(t)
+	defer db.Close()
+	seedEdgeRollupFixture(t, db)
+
+	preFix, err := os.ReadFile("testdata/edge_rollup_pre_fix.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	commaForm := edgeRollupCommaForm(t)
+
+	// Params: the ingested window is the fresh minute only; the start_time
+	// bounds are deliberately wide, as they are in production.
+	const windowLo, windowHi = 50, 200
+	spanLo := time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC)
+	spanHi := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
+
+	shippedPlan := explain(t, db, edgeRollupInsertSQL, windowLo, windowHi, spanLo, spanHi)
+	commaPlan := explain(t, db, commaForm, windowLo, windowHi, spanLo, spanHi)
+
+	// The defect stated as an assertion. If DuckDB ever folds this away, the
+	// guard below stops proving anything and this says so rather than passing.
+	if !strings.Contains(commaPlan, "CROSS_PRODUCT") {
+		t.Fatalf("the comma form no longer plans a CROSS_PRODUCT, so this guard proves nothing now:\n%s", commaPlan)
+	}
+	for _, bad := range []string{"CROSS_PRODUCT", "DELIM_JOIN", "DELIM_GET", "NESTED_LOOP_JOIN"} {
+		if strings.Contains(shippedPlan, bad) {
+			t.Errorf("edge rollup plans a %s; the FROM clause has regressed to the comma form:\n%s", bad, shippedPlan)
+		}
+	}
+	if !strings.Contains(shippedPlan, "HASH_JOIN") {
+		t.Errorf("edge rollup plans no HASH_JOIN at all:\n%s", shippedPlan)
+	}
+
+	// All three forms must agree on rows. Wall-clock is logged rather than
+	// asserted: the ratios are consistent but the absolute numbers are not
+	// stable enough across machines to fail a build on.
+	var want []string
+	for _, form := range []struct{ name, sql string }{
+		{"shipped", edgeRollupInsertSQL},
+		{"comma", commaForm},
+		{"pre-fix (range predicate in the ON clause)", string(preFix)},
+	} {
+		start := time.Now()
+		edges := runEdgeRollup(t, db, form.sql, windowLo, windowHi, spanLo, spanHi)
+		t.Logf("%-44s %6.0fms  %d edges", form.name, float64(time.Since(start).Microseconds())/1000, len(edges))
+		if want == nil {
+			if len(edges) == 0 {
+				t.Fatal("the fixture produced no edges, so no plan was exercised")
+			}
+			want = edges
+			continue
+		}
+		if strings.Join(edges, "|") != strings.Join(want, "|") {
+			t.Errorf("%s disagrees on rows:\n got:  %v\n want: %v", form.name, edges, want)
+		}
+	}
+	// frontend -> api -> db from the hand-built trace, svc-4 -> svc-5 from the
+	// bulk rows, and frontend -> worker across the queue.
+	const expected = "api->db call 1|frontend->api call 1|frontend->worker messaging 1|svc-4->svc-5 call 1000"
+	if strings.Join(want, "|") != expected {
+		t.Errorf("edges = %v,\n want %s", want, expected)
+	}
+}
+
+// edgeRollupCommaForm rebuilds the pre-fix FROM clause from the shipped
+// statement, so the two differ in exactly the thing under test and the guard
+// cannot rot into comparing two stale copies.
+func edgeRollupCommaForm(t *testing.T) string {
+	t.Helper()
+	const joined = "  FROM spans child\n  JOIN parent_scope parent"
+	if !strings.Contains(edgeRollupInsertSQL, joined) {
+		t.Fatal("the edge rollup FROM clause no longer matches what this test rewrites")
+	}
+	form := strings.Replace(edgeRollupInsertSQL, joined, "  FROM spans child, bounds\n  JOIN parent_scope parent", 1)
+	if !strings.Contains(form, "\n  CROSS JOIN bounds\n") {
+		t.Fatal("no CROSS JOIN bounds to remove")
+	}
+	return strings.Replace(form, "\n  CROSS JOIN bounds\n", "\n", 1)
+}
+
+func explain(t *testing.T, db *sql.DB, query string, args ...any) string {
+	t.Helper()
+	rows, err := db.Query("EXPLAIN "+query, args...)
+	if err != nil {
+		t.Fatalf("EXPLAIN: %v", err)
+	}
+	defer rows.Close()
+	var plan strings.Builder
+	for rows.Next() {
+		var key, value string
+		if err := rows.Scan(&key, &value); err != nil {
+			t.Fatal(err)
+		}
+		plan.WriteString(value)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return plan.String()
+}
+
+// runEdgeRollup executes the statement into an empty edge_rollup and returns
+// the edges it wrote, so forms can be compared on rows as well as on plan.
+func runEdgeRollup(t *testing.T, db *sql.DB, query string, args ...any) []string {
+	t.Helper()
+	if _, err := db.Exec("DELETE FROM edge_rollup"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(query, args...); err != nil {
+		t.Fatalf("edge rollup: %v", err)
+	}
+	rows, err := db.Query(`SELECT caller, callee, edge_type, calls FROM edge_rollup ORDER BY caller, callee, edge_type`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var edges []string
+	for rows.Next() {
+		var caller, callee, edgeType string
+		var calls int64
+		if err := rows.Scan(&caller, &callee, &edgeType, &calls); err != nil {
+			t.Fatal(err)
+		}
+		edges = append(edges, fmt.Sprintf("%s->%s %s %d", caller, callee, edgeType, calls))
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return edges
+}
+
+// seedEdgeRollupFixture builds the production shape: a long history of spans
+// that were rolled up long ago, and one freshly ingested minute. The two differ
+// by orders of magnitude, which is what makes where the parent side is filtered
+// matter. spans is a view over Parquet, as it is in production -- against a
+// DuckDB table the planner sees different statistics and the forms converge.
+func seedEdgeRollupFixture(t *testing.T, db *sql.DB) {
+	t.Helper()
+	if _, err := db.Exec(createEdgeRollupTable); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`CREATE TABLE spans (
+		namespace TEXT, trace_id TEXT, span_id TEXT, parent_span_id TEXT,
+		service TEXT, kind TEXT, status TEXT, duration_ms DOUBLE,
+		start_time TIMESTAMP, ingested_unix_nano BIGINT, attributes_json TEXT
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// The smallest trace that exercises both halves: a two-hop call chain and a
+	// producer/consumer pair, all inside the freshly ingested minute.
+	fresh := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC)
+	for _, s := range []struct{ id, parent, service, kind, attributes string }{
+		{id: "root", service: "frontend", kind: "SPAN_KIND_SERVER"},
+		{id: "a", parent: "root", service: "api", kind: "SPAN_KIND_SERVER"},
+		{id: "b", parent: "a", service: "db", kind: "SPAN_KIND_CLIENT"},
+		{id: "p", parent: "root", service: "frontend", kind: "SPAN_KIND_PRODUCER",
+			attributes: `{"messaging.destination.name":"jobs","messaging.system":"kafka"}`},
+		{id: "c", service: "worker", kind: "SPAN_KIND_CONSUMER",
+			attributes: `{"messaging.destination.name":"jobs","messaging.system":"kafka"}`},
+	} {
+		attributes := s.attributes
+		if attributes == "" {
+			attributes = "{}"
+		}
+		if _, err := db.Exec(`INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+			"default", "trace-1", s.id, s.parent, s.service, s.kind,
+			"STATUS_CODE_OK", 5.0, fresh, int64(100), attributes); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// 2k more spans in that same minute, so the fresh window is not trivially
+	// small relative to the joins.
+	if _, err := db.Exec(`INSERT INTO spans
+		SELECT 'default', 'new-' || (i // 2)::VARCHAR, 'new-' || i::VARCHAR,
+		       CASE WHEN i % 2 = 1 THEN 'new-' || (i - 1)::VARCHAR ELSE '' END,
+		       'svc-' || (i % 2 + 4)::VARCHAR, 'SPAN_KIND_SERVER', 'STATUS_CODE_OK', 5.0,
+		       TIMESTAMP '2026-09-20 12:00:00', 100, '{}'
+		FROM range(2000) t(i)`); err != nil {
+		t.Fatal(err)
+	}
+	// 600k spans across 25 days, ingested long before the window under test.
+	if _, err := db.Exec(`INSERT INTO spans
+		SELECT 'default', 'old-' || (i // 2)::VARCHAR, 'old-' || i::VARCHAR,
+		       CASE WHEN i % 2 = 1 THEN 'old-' || (i - 1)::VARCHAR ELSE '' END,
+		       'svc-' || (i % 8)::VARCHAR, 'SPAN_KIND_SERVER', 'STATUS_CODE_OK', 5.0,
+		       TIMESTAMP '2026-08-26 00:00:00' + INTERVAL (i % 36000) MINUTE, 1, '{}'
+		FROM range(600000) t(i)`); err != nil {
+		t.Fatal(err)
+	}
+
+	dir := t.TempDir()
+	if _, err := db.Exec(`COPY (SELECT * FROM spans) TO '` + dir + `/spans.parquet' (FORMAT PARQUET)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`DROP TABLE spans`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`CREATE VIEW spans AS SELECT * FROM read_parquet('` + dir + `/spans.parquet')`); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/internal/query/testdata/edge_rollup_pre_fix.sql b/internal/query/testdata/edge_rollup_pre_fix.sql
new file mode 100644
index 00000000..60497482
--- /dev/null
+++ b/internal/query/testdata/edge_rollup_pre_fix.sql
@@ -0,0 +1,126 @@
+
+WITH affected AS (
+  SELECT DISTINCT namespace, date_trunc('minute', start_time) AS bucket
+  FROM spans
+  WHERE ingested_unix_nano > ?
+    AND ingested_unix_nano <= ?
+    AND start_time IS NOT NULL
+    AND start_time >= ?
+    AND start_time < ?
+),
+call_edges AS (
+  SELECT
+    child.namespace,
+    date_trunc('minute', child.start_time) AS bucket,
+    parent.service AS caller,
+    child.service AS callee,
+    COUNT(*) AS calls,
+    AVG(child.duration_ms) AS avg_ms,
+    AVG(CASE WHEN child.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate,
+    'call' AS edge_type
+  FROM spans child
+  JOIN spans parent
+    ON child.parent_span_id = parent.span_id
+   AND child.trace_id = parent.trace_id
+   AND child.namespace = parent.namespace
+   -- Bound the parent side to the affected BUCKET RANGE ±1h: without this
+   -- the hash build covers every span ever ingested. Parents more than 1h
+   -- outside [MIN(affected.bucket), MAX(affected.bucket)] are dropped —
+   -- acceptable for minute-bucket dependency edges. Caveat: buckets come
+   -- from span start_time while the window bounds ingested_unix_nano, so
+   -- one late-arriving span with an old start_time widens the range (and
+   -- this scan) back to that bucket for the pass that ingests it.
+   AND parent.start_time >= (SELECT MIN(bucket) FROM affected) - INTERVAL 1 HOUR
+   AND parent.start_time <= (SELECT MAX(bucket) FROM affected) + INTERVAL 1 HOUR
+  JOIN affected a
+    ON a.namespace = child.namespace
+   AND a.bucket = date_trunc('minute', child.start_time)
+  WHERE parent.service IS NOT NULL
+    AND parent.service != ''
+    AND child.service IS NOT NULL
+    AND child.service != ''
+    AND parent.service != child.service
+    AND child.start_time >= (SELECT MIN(bucket) FROM affected)
+    AND child.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+  GROUP BY child.namespace, date_trunc('minute', child.start_time), parent.service, child.service
+),
+-- Producers and consumers are aggregated per (namespace, bucket, service,
+-- destination, msg_system) BEFORE the join — joining raw span rows multiplies
+-- producer spans by consumer spans per destination, quadratic in per-bucket
+-- span count. calls = consumed messages on the destination, attributed to
+-- each producer service publishing to it (a message consumed once appears on
+-- every producer's edge).
+producers AS (
+  SELECT DISTINCT
+    s.namespace,
+    date_trunc('minute', s.start_time) AS bucket,
+    s.service,
+    json_extract_string(s.attributes_json, '$."messaging.destination.name"') AS destination,
+    json_extract_string(s.attributes_json, '$."messaging.system"') AS msg_system
+  FROM spans s
+  JOIN affected a
+    ON a.namespace = s.namespace
+   AND a.bucket = date_trunc('minute', s.start_time)
+  WHERE s.kind = 'SPAN_KIND_PRODUCER'
+    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
+    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.service IS NOT NULL
+    AND s.service != ''
+    AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
+),
+consumers AS (
+  SELECT
+    s.namespace,
+    date_trunc('minute', s.start_time) AS bucket,
+    s.service,
+    json_extract_string(s.attributes_json, '$."messaging.destination.name"') AS destination,
+    json_extract_string(s.attributes_json, '$."messaging.system"') AS msg_system,
+    COUNT(*) AS calls
+  FROM spans s
+  JOIN affected a
+    ON a.namespace = s.namespace
+   AND a.bucket = date_trunc('minute', s.start_time)
+  WHERE s.kind = 'SPAN_KIND_CONSUMER'
+    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
+    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.service IS NOT NULL
+    AND s.service != ''
+    AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
+  GROUP BY s.namespace, date_trunc('minute', s.start_time), s.service,
+    json_extract_string(s.attributes_json, '$."messaging.destination.name"'),
+    json_extract_string(s.attributes_json, '$."messaging.system"')
+),
+messaging_edges AS (
+  SELECT
+    p.namespace,
+    p.bucket,
+    p.service AS caller,
+    c.service AS callee,
+    SUM(c.calls) AS calls,
+    0.0 AS avg_ms,
+    0.0 AS error_rate,
+    'messaging' AS edge_type
+  FROM producers p
+  JOIN consumers c
+    ON c.namespace = p.namespace
+   AND c.bucket = p.bucket
+   AND c.destination = p.destination
+   AND c.msg_system = p.msg_system
+  WHERE p.service != c.service
+  GROUP BY p.namespace, p.bucket, p.service, c.service
+)
+INSERT INTO edge_rollup (
+  namespace,
+  bucket,
+  caller,
+  callee,
+  calls,
+  avg_ms,
+  error_rate,
+  edge_type
+)
+SELECT namespace, bucket, caller, callee, calls, avg_ms, error_rate, edge_type
+FROM call_edges
+UNION ALL
+SELECT namespace, bucket, caller, callee, calls, avg_ms, error_rate, edge_type
+FROM messaging_edges;
\ No newline at end of file
diff --git a/internal/query/views.go b/internal/query/views.go
index 6a146ce0..c79ea40b 100644
--- a/internal/query/views.go
+++ b/internal/query/views.go
@@ -315,10 +315,13 @@ func CreateParquetViews(db *sql.DB, parquetDir string) error {
 	}
 	for _, signal := range []string{"spans", "logs", "metrics"} {
 		pattern := filepath.ToSlash(filepath.Join(parquetDir, "batches", "*.batch", signal+".parquet"))
-		// _schema.batch is written from the current binary's row structs and is
-		// the definition of "every column this build knows about", so reading
-		// its schema keeps the view in lockstep with the writer instead of
-		// duplicating the column list here for someone to forget.
+		// _schema.batch is rewritten from the current binary's row structs on
+		// every open (ParquetStore.ensureSchemaBatch), so it is the definition
+		// of "every column this build knows about". Reading its schema keeps
+		// the view in lockstep with the writer instead of duplicating the
+		// column list here for someone to forget. That rewrite is load-bearing:
+		// while the file was written once and kept, these views were pinned to
+		// whichever build created the data directory.
 		schemaFile := filepath.ToSlash(filepath.Join(parquetDir, "batches", "_schema.batch", signal+".parquet"))
 		columns, err := parquetSchemaMap(db, schemaFile, signal == "spans")
 		if err != nil {
diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go
index 706f522f..03970629 100644
--- a/internal/telemetry/parquet.go
+++ b/internal/telemetry/parquet.go
@@ -938,14 +938,24 @@ func (p *ParquetStore) PublishReplacement(stage string, metadata BatchMetadata,
 	return errors.Join(removeErr, syncDirectory(p.batchesDir))
 }
 
+// ensureSchemaBatch writes _schema.batch: an empty Parquet batch whose only
+// purpose is to carry this binary's column list. internal/query derives every
+// view's explicit read_parquet(schema=MAP{...}) from it, so a column absent
+// here is a column absent from the view.
+//
+// It is rewritten on every open rather than created once. Written once, it
+// pinned the views to whichever build first created the data directory, for
+// the life of that deployment -- and a later build that added a column to
+// spanParquetRow could not start at all: the view would not expose it,
+// CreateViews would fail to bind it, NewDuck would return an error, and
+// restarting would not help because the stale file was still there. The files
+// are empty, so this is three small writes per boot.
 func (p *ParquetStore) ensureSchemaBatch() error {
 	final := filepath.Join(p.batchesDir, SchemaBatch)
-	if info, err := os.Stat(final); err == nil && info.IsDir() {
-		return nil
-	} else if err != nil && !errors.Is(err, os.ErrNotExist) {
+	stage := filepath.Join(p.stagingDir, "_schema")
+	if err := os.RemoveAll(stage); err != nil {
 		return err
 	}
-	stage := filepath.Join(p.stagingDir, "_schema")
 	if err := os.Mkdir(stage, 0o755); err != nil {
 		return err
 	}
@@ -961,6 +971,12 @@ func (p *ParquetStore) ensureSchemaBatch() error {
 	if err := syncDirectory(stage); err != nil {
 		return err
 	}
+	// Rename cannot replace a non-empty directory, so the old one is removed
+	// first. A crash between the two leaves no _schema.batch at all, which the
+	// next open recreates: the batch holds no rows, only the column list.
+	if err := os.RemoveAll(final); err != nil {
+		return err
+	}
 	if err := os.Rename(stage, final); err != nil {
 		return err
 	}
diff --git a/internal/telemetry/schema_batch_test.go b/internal/telemetry/schema_batch_test.go
new file mode 100644
index 00000000..5d54baf7
--- /dev/null
+++ b/internal/telemetry/schema_batch_test.go
@@ -0,0 +1,89 @@
+package telemetry
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/parquet-go/parquet-go"
+)
+
+// _schema.batch is an empty Parquet batch whose only job is to carry this
+// binary's column list: internal/query derives every view's explicit
+// read_parquet(schema=MAP{...}) from it, so a column missing there is a column
+// missing from the view.
+//
+// It was written once, on the first boot into an empty data directory, and
+// never revisited. Every deployment since has been reading the column list of
+// whichever build happened to create the directory. Adding a column to
+// spanParquetRow would have shipped a binary whose views cannot bind it --
+// CreateViews fails, NewDuck returns an error, the process does not start, and
+// restarting does not help because the stale file is still there.
+//
+// The direction that matters is an OLD file and a NEW binary, so that is what
+// this sets up. A test that writes a fresh _schema.batch and reads it back
+// passes against the broken code.
+func TestSchemaBatchTracksTheRunningBinary(t *testing.T) {
+	dir := t.TempDir()
+	store, err := OpenParquetStore(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	store.Close()
+
+	spansFile := filepath.Join(dir, "batches", SchemaBatch, "spans.parquet")
+	current := parquetColumns(t, spansFile)
+	if len(current) < 10 {
+		t.Fatalf("only %d columns in a freshly written schema batch: %v", len(current), current)
+	}
+
+	// An older build's _schema.batch: same file, fewer columns.
+	type oldSpanRow struct {
+		Namespace string `parquet:"namespace"`
+		TraceID   string `parquet:"trace_id"`
+		SpanID    string `parquet:"span_id"`
+	}
+	if err := os.Remove(spansFile); err != nil {
+		t.Fatal(err)
+	}
+	if err := writeTypedParquet(spansFile, []oldSpanRow{}, parquetPageSize); err != nil {
+		t.Fatal(err)
+	}
+	if got := parquetColumns(t, spansFile); len(got) != 3 {
+		t.Fatalf("fixture is wrong: wrote %d columns, want 3", len(got))
+	}
+
+	reopened, err := OpenParquetStore(dir)
+	if err != nil {
+		t.Fatalf("reopen against an older schema batch: %v", err)
+	}
+	defer reopened.Close()
+
+	got := parquetColumns(t, spansFile)
+	if len(got) != len(current) {
+		t.Errorf("after reopening, the schema batch has %d columns, want this binary's %d\n got:  %v\n want: %v",
+			len(got), len(current), got, current)
+	}
+}
+
+func parquetColumns(t *testing.T, path string) []string {
+	t.Helper()
+	file, err := os.Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer file.Close()
+	info, err := file.Stat()
+	if err != nil {
+		t.Fatal(err)
+	}
+	parquetFile, err := parquet.OpenFile(file, info.Size())
+	if err != nil {
+		t.Fatal(err)
+	}
+	var columns []string
+	for _, field := range parquetFile.Schema().Fields() {
+		columns = append(columns, field.Name())
+	}
+	return columns
+}
diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go
index b1bc96c1..a2d0153b 100644
--- a/internal/telemetry/store/compaction.go
+++ b/internal/telemetry/store/compaction.go
@@ -376,18 +376,6 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in
 		saturated = true
 	}
 	// A group is worth merging when it is full, when a ceiling stopped it, or
-	// when it simply holds enough files to be worth the merge.
-	//
-	// That last case is the one this was missing. The escape hatch above only
-	// covered groups too LARGE for a ceiling; a group merely too SMALL to reach
-	// maxBatches was stranded forever. Days stop being written -- the group for
-	// a past day never grows again -- so on the live box 1,243 batches sat in 69
-	// (day, generation) groups across 25 days and not one produced a candidate.
-	// Only the current day ever reached 128 files, so every output was
-	// generation 1 and everything older was untouchable. The live file count,
-	// which every query pays per-file overhead on, held at an equilibrium that
-	// no reordering or budget change could move.
-	// A group is worth merging when it is full, when a ceiling stopped it, or
 	// when it belongs to a day that has finished and already holds enough files
 	// to be worth the merge.
 	//

From a017ef0eacc5379e9734ca51f45ddb8072ce7050 Mon Sep 17 00:00:00 2001
From: Vishal Rana <v@labstack.com>
Date: Mon, 21 Sep 2026 11:07:37 -0700
Subject: [PATCH 6/6] fix(intelligence): stop reporting every service as a
 critical anomaly

The demo sat at health score 0 with 17 critical insights, permanently,
every one of them wrong. Two independent defects, both silent.

DuckDB's FLOAT is single precision, so the driver returns float32 and
the row["x"].(float64) that every caller here writes fails -- yielding
0.0 from a value that was never zero. Three columns were cast to FLOAT.
Volume anomalies therefore reported "0 spans" for services serving
thousands, and the error-rate detector's current and baseline rates
were always 0, for as long as the code existed. Callers write that
assertion because every other numeric column they meet (DOUBLE, AVG,
SUM) really is float64, so the one FLOAT column in a query fails while
the rest of the row is fine.

The casts become DOUBLE, but the conversion is what changes: RowMap now
widens float32, because fixing three casts fixes today's queries and
leaves the trap armed for the next one.

Separately, the volume detector compared windows of equal length that
were not counted the same way -- current was a single COUNT(*) over the
whole window while the baseline averaged per-5-minute-bucket counts.
On a 15-minute window the current figure was three times the baseline
for every service regardless of traffic. Perfectly steady traffic
scores 3.00x and z=75.59 against that form. Both sides now average over
the same buckets.

A test asserting only that steady traffic is quiet would pass against a
detector that never fires, so the guard runs both directions: steady
traffic scores 0.00 and a service that loses 95% of its traffic scores
-36.03. Verified live -- the demo now reports health 100 with no
anomalies, and its real latency outlier still surfaces.

Error-rate detection remains desensitised for a separate, pre-existing
reason not addressed here: its z-score divides a difference of rates by
STDDEV of a per-span 0/1 indicator, which is the Bernoulli spread of
individual outcomes rather than the spread of the rate across buckets.
At the demo's 13.8% baseline a spike must exceed 69 percentage points
to reach the threshold, and the higher a service's error rate the
harder its spikes become to detect.
---
 internal/intelligence/detector.go          |  74 ++++++----
 internal/intelligence/sql_comments_test.go |  17 +++
 internal/intelligence/volume_sql_test.go   | 150 +++++++++++++++++++++
 internal/query/rowmap_float_test.go        |  47 +++++++
 internal/query/sql.go                      |  16 ++-
 5 files changed, 275 insertions(+), 29 deletions(-)
 create mode 100644 internal/intelligence/volume_sql_test.go
 create mode 100644 internal/query/rowmap_float_test.go

diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go
index 8c5ad936..10de6c5a 100644
--- a/internal/intelligence/detector.go
+++ b/internal/intelligence/detector.go
@@ -1,3 +1,11 @@
+// Package intelligence derives anomalies and log patterns from telemetry.
+//
+// Every SQL string in this file goes through internal/query's validator, which
+// rejects "--" outright. A SQL comment in one of these statements therefore
+// fails the query at runtime, on a background goroutine, as a logged error
+// nobody is watching -- that is how latency detection stopped reporting for
+// half an hour. Rationale goes in Go comments; TestNoSQLCommentsInQueryStrings
+// enforces it.
 package intelligence
 
 import (
@@ -145,14 +153,6 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 	startNano := start.UnixNano()
 	endNano := end.UnixNano()
 	namespace := d.duck.DefaultNamespace()
-	// The percentile below is approx_quantile, not PERCENTILE_CONT: the exact
-	// form is holistic and retains every value of every group on the raw
-	// allocator, outside anything memory_limit bounds. This runs every 60s over
-	// a 15-minute window. See serviceRollupP95SQL.
-	//
-	// Keep rationale in Go comments, not SQL ones: these statements go through a
-	// validator that rejects "--" outright, so a SQL comment here fails the query
-	// at runtime rather than at build time.
 	scope := detectorScopeClause(namespace)
 
 	// Compare current error rate to baseline (previous period)
@@ -162,7 +162,7 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 				service as service_name,
 				COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR')) AS error_count,
 				COUNT(*) AS total_count,
-				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::FLOAT / COUNT(*)::FLOAT) AS error_rate
+				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS error_rate
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
 			%s
@@ -173,7 +173,7 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 				service as service_name,
 				COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR')) AS error_count,
 				COUNT(*) AS total_count,
-				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::FLOAT / COUNT(*)::FLOAT) AS error_rate,
+				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS error_rate,
 				STDDEV(CASE WHEN status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_stddev
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
@@ -228,6 +228,10 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 
 // detectLatencyAnomalies detects latency degradation
 func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.Time) []Anomaly {
+	// The percentile below is approx_quantile, not PERCENTILE_CONT: the exact
+	// form is holistic and retains every value of every group on the raw
+	// allocator, outside anything memory_limit bounds. This runs every 60s over
+	// a 15-minute window. See serviceRollupP95SQL.
 	startNano := start.UnixNano()
 	endNano := end.UnixNano()
 	namespace := d.duck.DefaultNamespace()
@@ -308,22 +312,32 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 	return anomalies
 }
 
-// detectVolumeAnomalies detects unusual traffic volume changes
-func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Time) []Anomaly {
-	startNano := start.UnixNano()
-	endNano := end.UnixNano()
-	namespace := d.duck.DefaultNamespace()
-	scope := detectorScopeClause(namespace)
-
-	sql := fmt.Sprintf(`
+// volumeAnomalySQL compares span volume in [startNano, endNano) against the
+// window of equal length immediately before it.
+//
+// Both sides are averaged over 5-minute buckets, and that symmetry is the whole
+// point. The current side used to be a single COUNT(*) over the entire window
+// while the baseline averaged per-bucket counts, so on a 15-minute window the
+// current figure was three times the baseline for every service no matter what
+// the traffic did. Every service was permanently a critical volume anomaly:
+// steady traffic scored z=75 in TestVolumeAnomalyComparesEqualWindows, and the
+// live demo sat at health score 0 with 17 meaningless critical insights.
+func volumeAnomalySQL(startNano, endNano int64, scope string) string {
+	return fmt.Sprintf(`
 		WITH current_period AS (
 			SELECT
-				service as service_name,
-				COUNT(*) AS span_count
-			FROM spans
-			WHERE start_unix_nano >= %d AND start_unix_nano < %d
-			%s
-			GROUP BY service
+				service_name,
+				AVG(cnt) AS span_count
+			FROM (
+				SELECT
+					service as service_name,
+					COUNT(*) AS cnt
+				FROM spans
+				WHERE start_unix_nano >= %d AND start_unix_nano < %d
+				%s
+				GROUP BY service, time_bucket(INTERVAL '5 minutes', start_time)
+			) subq
+			GROUP BY service_name
 		),
 		baseline_period AS (
 			SELECT
@@ -343,7 +357,7 @@ func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Ti
 		)
 		SELECT
 			c.service_name,
-			c.span_count::FLOAT AS current_count,
+			c.span_count::DOUBLE AS current_count,
 			COALESCE(b.avg_count, 0.0) AS baseline_count,
 			CASE
 				WHEN b.count_stddev > 0 THEN (c.span_count - b.avg_count) / b.count_stddev
@@ -352,6 +366,16 @@ func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Ti
 		FROM current_period c
 		LEFT JOIN baseline_period b ON c.service_name = b.service_name
 	`, startNano, endNano, scope, startNano-endNano+startNano, startNano, scope)
+}
+
+// detectVolumeAnomalies detects unusual traffic volume changes
+func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Time) []Anomaly {
+	startNano := start.UnixNano()
+	endNano := end.UnixNano()
+	namespace := d.duck.DefaultNamespace()
+	scope := detectorScopeClause(namespace)
+
+	sql := volumeAnomalySQL(startNano, endNano, scope)
 
 	resp := d.duck.ExecuteSQL(ctx, query.SQLRequest{Query: sql})
 	if resp.Error != "" {
diff --git a/internal/intelligence/sql_comments_test.go b/internal/intelligence/sql_comments_test.go
index 979aff6f..b1f475d8 100644
--- a/internal/intelligence/sql_comments_test.go
+++ b/internal/intelligence/sql_comments_test.go
@@ -44,3 +44,20 @@ func TestNoSQLCommentsInQueryStrings(t *testing.T) {
 	}
 	t.Logf("checked %d SQL literals", found)
 }
+
+// DuckDB's FLOAT is single precision, so the driver returns float32 and the
+// `row["x"].(float64)` every caller here writes fails -- silently yielding 0.0.
+// internal/query now widens float32 on the way out, which disarms the trap, but
+// a rate or a count has no business being single precision in the first place
+// and the next reader should not have to know about the widening to trust it.
+func TestNoFloatCastsInQueryStrings(t *testing.T) {
+	source, err := os.ReadFile("detector.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i, line := range strings.Split(string(source), "\n") {
+		if strings.Contains(strings.ToUpper(line), "::FLOAT") {
+			t.Errorf("detector.go:%d casts to FLOAT (single precision): %q\nuse ::DOUBLE", i+1, strings.TrimSpace(line))
+		}
+	}
+}
diff --git a/internal/intelligence/volume_sql_test.go b/internal/intelligence/volume_sql_test.go
new file mode 100644
index 00000000..7c982bf4
--- /dev/null
+++ b/internal/intelligence/volume_sql_test.go
@@ -0,0 +1,150 @@
+package intelligence
+
+import (
+	"database/sql"
+	"fmt"
+	"math"
+	"testing"
+	"time"
+)
+
+// The current window and the baseline window are the same length, but they were
+// not counted the same way: the baseline averaged per-5-minute-bucket counts
+// while the current side summed the whole 15 minutes. Current was therefore
+// about three times the baseline by construction, for every service, always --
+// so every service was permanently a critical volume anomaly and the demo's
+// health score sat at 0 with 17 "critical" insights that meant nothing.
+//
+// Steady traffic must produce no anomaly. That is the property here.
+func TestVolumeAnomalyComparesEqualWindows(t *testing.T) {
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE spans (
+		service TEXT, namespace TEXT, start_time TIMESTAMP, start_unix_nano BIGINT
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// 30 minutes of steady traffic in six 5-minute buckets, with enough jitter
+	// that STDDEV is not zero -- a perfectly flat fixture divides by zero and
+	// the CASE returns 0.0, which would pass against the broken form too.
+	end := time.Date(2026, 9, 21, 18, 0, 0, 0, time.UTC)
+	start := end.Add(-15 * time.Minute)
+	baselineStart := start.Add(-15 * time.Minute)
+	perBucket := []int{98, 103, 99, 101, 97, 102}
+	for bucket, count := range perBucket {
+		at := baselineStart.Add(time.Duration(bucket) * 5 * time.Minute)
+		for _, service := range []string{"cart", "checkout", "frontend"} {
+			for i := range count {
+				ts := at.Add(time.Duration(i) * time.Millisecond)
+				if _, err := db.Exec(`INSERT INTO spans VALUES (?, 'default', ?, ?)`,
+					service, ts, ts.UnixNano()); err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
+	}
+
+	query := volumeAnomalySQL(start.UnixNano(), end.UnixNano(), "")
+	rows, err := db.Query(query)
+	if err != nil {
+		t.Fatalf("volume query: %v", err)
+	}
+	defer rows.Close()
+
+	seen := 0
+	for rows.Next() {
+		var service string
+		var current, baseline, zScore float64
+		if err := rows.Scan(&service, &current, &baseline, &zScore); err != nil {
+			t.Fatal(err)
+		}
+		seen++
+		if current <= 0 {
+			t.Errorf("%s: current = %v, want the spans it actually served", service, current)
+		}
+		ratio := current / baseline
+		if ratio < 0.8 || ratio > 1.25 {
+			t.Errorf("%s: current %.0f against baseline %.0f (%.2fx) -- the two windows are not counted the same way",
+				service, current, baseline, ratio)
+		}
+		if math.Abs(zScore) >= 3 {
+			t.Errorf("%s: steady traffic scored z=%.2f, which reports as a critical anomaly", service, zScore)
+		}
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if seen != 3 {
+		t.Fatalf("got %d services, want 3", seen)
+	}
+	fmt.Print()
+}
+
+// The other direction, and the one that matters more: a test asserting only
+// that steady traffic is quiet passes just as well against a detector that
+// never fires at all. A service whose traffic really does collapse must still
+// be reported.
+func TestVolumeAnomalyStillFiresOnARealDrop(t *testing.T) {
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE spans (
+		service TEXT, namespace TEXT, start_time TIMESTAMP, start_unix_nano BIGINT
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	end := time.Date(2026, 9, 21, 18, 0, 0, 0, time.UTC)
+	start := end.Add(-15 * time.Minute)
+	baselineStart := start.Add(-15 * time.Minute)
+	perBucket := []int{98, 103, 99, 101, 97, 102}
+	for bucket, count := range perBucket {
+		at := baselineStart.Add(time.Duration(bucket) * 5 * time.Minute)
+		inCurrentWindow := !at.Before(start)
+		for _, service := range []string{"steady", "collapsing"} {
+			// "collapsing" serves its baseline, then drops to 5% of it.
+			if service == "collapsing" && inCurrentWindow {
+				count = count / 20
+			}
+			for i := range count {
+				ts := at.Add(time.Duration(i) * time.Millisecond)
+				if _, err := db.Exec(`INSERT INTO spans VALUES (?, 'default', ?, ?)`,
+					service, ts, ts.UnixNano()); err != nil {
+					t.Fatal(err)
+				}
+			}
+			count = perBucket[bucket]
+		}
+	}
+
+	rows, err := db.Query(volumeAnomalySQL(start.UnixNano(), end.UnixNano(), ""))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	scores := map[string]float64{}
+	for rows.Next() {
+		var service string
+		var current, baseline, zScore float64
+		if err := rows.Scan(&service, &current, &baseline, &zScore); err != nil {
+			t.Fatal(err)
+		}
+		scores[service] = zScore
+		t.Logf("%-11s current=%.1f baseline=%.1f z=%.2f", service, current, baseline, zScore)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if z, ok := scores["collapsing"]; !ok || math.Abs(z) < 3 {
+		t.Errorf("a service that lost 95%% of its traffic scored z=%.2f (present: %v); it must be reported", z, ok)
+	}
+	if z := scores["steady"]; math.Abs(z) >= 3 {
+		t.Errorf("steady traffic scored z=%.2f alongside it", z)
+	}
+}
diff --git a/internal/query/rowmap_float_test.go b/internal/query/rowmap_float_test.go
new file mode 100644
index 00000000..5111f0ca
--- /dev/null
+++ b/internal/query/rowmap_float_test.go
@@ -0,0 +1,47 @@
+package query
+
+import (
+	"context"
+	"testing"
+)
+
+// DuckDB's FLOAT is single precision, so the driver hands back a float32 and
+// every `row["x"].(float64)` on it fails -- yielding 0.0 from a value that was
+// never zero. Callers write that assertion because every other numeric column
+// they touch (DOUBLE, AVG, SUM) really is float64, so the one FLOAT column in a
+// query fails silently while the rest of the row is fine.
+//
+// That is not hypothetical: internal/intelligence cast three columns to FLOAT
+// and reported 0 for all of them for as long as the code existed -- volume
+// anomalies showed "0 spans" for services that were serving thousands, and the
+// error-rate detector's current and baseline rates were always 0.
+//
+// Fixing the casts alone fixes today's queries and leaves the trap armed for
+// the next one, so the conversion is what changes here.
+func TestRowMapWidensFloat32(t *testing.T) {
+	db := openTestDuck(t)
+	defer db.Close()
+	d := &Duck{DB: db}
+
+	resp := d.ExecuteSQL(context.Background(), SQLRequest{
+		Query: "SELECT 327::FLOAT AS as_float, 0.25::FLOAT AS rate, 327::DOUBLE AS as_double",
+	})
+	if resp.Error != "" {
+		t.Fatalf("ExecuteSQL: %s", resp.Error)
+	}
+	if len(resp.Results) != 1 {
+		t.Fatalf("got %d rows, want 1", len(resp.Results))
+	}
+	row := resp.Results[0]
+
+	for name, want := range map[string]float64{"as_float": 327, "rate": 0.25, "as_double": 327} {
+		got, ok := row[name].(float64)
+		if !ok {
+			t.Errorf("row[%q] is %T, not float64: every caller asserting float64 reads 0 from it", name, row[name])
+			continue
+		}
+		if got != want {
+			t.Errorf("row[%q] = %v, want %v", name, got, want)
+		}
+	}
+}
diff --git a/internal/query/sql.go b/internal/query/sql.go
index 6776ed32..e2504129 100644
--- a/internal/query/sql.go
+++ b/internal/query/sql.go
@@ -148,10 +148,18 @@ func (d *Duck) ExecuteSQL(ctx context.Context, req SQLRequest) (resp SQLResponse
 			row := make(RowMap)
 			for i, col := range columns {
 				val := values[i]
-				// Convert []uint8 to string for better JSON representation
-				if b, ok := val.([]byte); ok {
-					row[col] = string(b)
-				} else {
+				switch v := val.(type) {
+				case []byte:
+					// Convert []uint8 to string for better JSON representation
+					row[col] = string(v)
+				case float32:
+					// DuckDB's FLOAT is single precision, so the driver returns
+					// float32. Callers assert float64 -- every other numeric
+					// column they meet is one -- and a failed assertion yields
+					// 0.0 from a value that was never zero, silently. Widening
+					// here costs nothing and disarms the whole class.
+					row[col] = float64(v)
+				default:
 					row[col] = val
 				}
 			}