From a6bdd40fc83d4bc46105dd45644752f5570df06d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 23 Oct 2026 18:29:51 +0000
Subject: [PATCH 1/3] Store changes

---
 internal/store/chartedges_test.go |   2 +-
 internal/store/purge.go           |   2 +-
 internal/store/query.go           | 160 +++++++++++-----
 internal/store/rings.go           |  33 ++--
 internal/store/series.go          |  34 ++--
 internal/store/store.go           |  36 +++-
 internal/store/views.go           | 308 ++++++++++++++++++++++++++++++
 internal/store/views_test.go      | 104 ++++++++++
 8 files changed, 602 insertions(+), 77 deletions(-)
 create mode 100644 internal/store/views.go
 create mode 100644 internal/store/views_test.go

diff --git a/internal/store/chartedges_test.go b/internal/store/chartedges_test.go
index 278cedf..36f45f4 100644
--- a/internal/store/chartedges_test.go
+++ b/internal/store/chartedges_test.go
@@ -79,7 +79,7 @@ func TestFirstBucketCountsOnlyTheCoveredPart(t *testing.T) {
 		t.Fatal(err)
 	}
 	q := Query{From: base.Add(-2 * time.Hour), To: base.Add(4 * time.Hour)}
-	if q.usesRollup() {
+	if s.rollupOK(q) {
 		t.Skip("range answered from the summaries")
 	}
 	for name, get := range map[string]func() (*Series, error){
diff --git a/internal/store/purge.go b/internal/store/purge.go
index 96429bb..93dff1f 100644
--- a/internal/store/purge.go
+++ b/internal/store/purge.go
@@ -132,7 +132,7 @@ func (s *Store) PurgeBefore(before time.Time) (PurgeSize, error) {
 	} else {
 		return done, err
 	}
-	for _, t := range []string{"r_ts", "r_host", "r_dim", "ifc"} {
+	for _, t := range append([]string{"r_ts", "r_host", "r_dim", "ifc"}, viewTables...) {
 		if _, err := s.DB.Exec(`DELETE FROM `+t+` WHERE `+cond, args...); err != nil {
 			return done, err
 		}
diff --git a/internal/store/query.go b/internal/store/query.go
index fc3e0f4..38e4345 100644
--- a/internal/store/query.go
+++ b/internal/store/query.go
@@ -57,8 +57,6 @@ func (q Query) bucketSQL(step time.Duration) string {
 
 func (q Query) span() time.Duration { return q.To.Sub(q.From) }
 
-// usesRollup reports whether this query can be answered from rollups.
-func (q Query) usesRollup() bool { return len(q.Filters) == 0 && q.span() > 6*time.Hour }
 
 var dirNames = map[string]int{"outbound": DirOutbound, "inbound": DirInbound, "internal": DirInternal, "transit": DirTransit}
 
@@ -391,65 +389,114 @@ func (s *Store) TopNBy(q Query, dim string, limit int, by string, asc ...bool) (
 	if limit <= 0 || limit > 1000 {
 		limit = 66
 	}
-	if q.usesRollup() && d.rollup != "" {
+	if s.rollupOK(q) && d.rollup != "" {
 		return s.topRollup(q, d.rollup, limit, ord[1]+" "+dir)
 	}
 	where, args, err := q.where()
 	if err != nil {
 		return nil, err
 	}
+	cond := where
+	if dim == "threat" {
+		cond += " AND threat <> ''"
+	}
 	// Group by cheap keys (numeric IPv4, port and protocol as integers) and
 	// format for display afterwards: grouping on address text is several
-	// times slower. IPv4 and IPv6 rows never share a group, so they are
-	// aggregated separately and merged.
-	var inner, groups, inner6, groups6, outer, v4conds []string
+	// times slower. An IPv4 address is grouped by its number with no text,
+	// an IPv6 one by its text with the number 0, so the two never share a
+	// group.
+	var parts []keyPart
+	var v4conds []string
 	for i, e := range []string{d.expr, d.expr2, d.expr3} {
 		if e == "" {
-			outer = append(outer, "'' AS k"+strconv.Itoa(i))
 			continue
 		}
 		kp := keyPartFor(e, i)
-		inner = append(inner, kp.inner...)
-		groups = append(groups, kp.groups...)
-		inner6 = append(inner6, kp.inner6...)
-		groups6 = append(groups6, kp.groups6...)
+		parts = append(parts, kp)
 		if kp.v4cond != "" {
 			v4conds = append(v4conds, kp.v4cond)
 		}
-		outer = append(outer, kp.outer+" AS k"+strconv.Itoa(i))
+	}
+	v4 := "true"
+	if len(v4conds) > 0 {
+		v4 = strings.Join(v4conds, " AND ")
+	}
+	var keySel, groups, outer, on []string
+	for _, kp := range parts {
+		for j, g := range kp.groups {
+			if kp.v4cond != "" {
+				// the number for IPv4 rows, the text for the others
+				n, t := kp.inner[0], kp.inner6[1]
+				n, t = n[:strings.LastIndex(n, " AS ")], t[:strings.LastIndex(t, " AS ")]
+				keySel = append(keySel, "CASE WHEN "+v4+" THEN "+n+" ELSE 0 END AS "+kp.groups[0],
+					"CASE WHEN "+v4+" THEN NULL ELSE "+t+" END AS "+kp.groups6[0])
+				groups = append(groups, kp.groups[0], kp.groups6[0])
+				break
+			}
+			keySel = append(keySel, kp.inner[j])
+			groups = append(groups, g)
+		}
+	}
+	for i, e := range []string{d.expr, d.expr2, d.expr3} {
+		if e == "" {
+			outer = append(outer, "'' AS k"+strconv.Itoa(i))
+		} else {
+			outer = append(outer, keyPartFor(e, i).outer+" AS k"+strconv.Itoa(i))
+		}
+	}
+	// pg: the group columns of a step joined back to t, renamed so the
+	// two sides of the join do not both have them
+	var pg, onP []string
+	for _, g := range groups {
+		on = append(on, "t."+g+" IS NOT DISTINCT FROM x."+g)
+		pg = append(pg, "x."+g+" AS p_"+g)
+		onP = append(onP, "t."+g+" IS NOT DISTINCT FROM p.p_"+g)
 	}
 	extra := "''"
 	if d.extra != "" {
 		extra = "coalesce(CAST(" + d.extra + " AS VARCHAR), '')"
 	}
-	peers := "0"
-	if d.peers != "" && q.span() <= 7*24*time.Hour {
-		peers = "approx_count_distinct(" + d.peers + ")"
-	}
 	intc := "false"
 	if d.intCol != "" {
 		intc = d.intCol
 	}
-	cond := where
-	if dim == "threat" {
-		cond += " AND threat <> ''"
-	}
-	src := s.Source(q.From, q.To)
-	aggs := fmt.Sprintf(`%s AS x_extra, sum(bytes) AS x_b, sum(wire) AS x_w, sum(pkts) AS x_p, sum(flows) AS x_f, %s AS x_peers, %s AS x_int`, extra, peers, intc)
-	part := func(sel, grp []string, c string) string {
-		return fmt.Sprintf(`(SELECT %s, %s FROM %s WHERE %s GROUP BY %s ORDER BY %s %s, x_w DESC LIMIT %d)`,
-			strings.Join(sel, ", "), aggs, src, c, strings.Join(grp, ", "), ord[0], dir, limit)
-	}
-	body := part(inner, groups, cond)
-	qargs := args
-	if len(v4conds) > 0 {
-		v4 := strings.Join(v4conds, " AND ")
-		body = part(inner, groups, "("+cond+") AND "+v4) + " UNION ALL " + part(inner6, groups6, "("+cond+") AND NOT ("+v4+")")
-		qargs = append(append([]any{}, args...), args...)
-	}
-	sqlq := fmt.Sprintf(`SELECT %s, x_extra, x_b, x_w, x_p, x_f, x_peers, x_int FROM (%s) ORDER BY %s %s, x_w DESC LIMIT %d`,
-		strings.Join(outer, ", "), body, ord[0], dir, limit)
-	rows, err := s.DB.Query(sqlq, qargs...)
+	peerCol := "NULL"
+	if d.peers != "" {
+		peerCol = d.peers
+	}
+	// x is the rows, with their keys, read again by each step below rather
+	// than kept: kept, a week of rows is more than the memory there is.
+	// only the columns the steps below use: the flow files are read by
+	// column, and every other one would be read for nothing
+	x := fmt.Sprintf(`x AS NOT MATERIALIZED (SELECT bytes, wire, pkts, flows, peer_cc, peer_org, app, dir, cli_int, srv_int, %s, %s AS x_pc FROM %s WHERE %s)`,
+		strings.Join(keySel, ", "), peerCol, s.Source(q.From, q.To), cond)
+	g := strings.Join(groups, ", ")
+	sums := fmt.Sprintf(`%s AS x_extra, sum(bytes) AS x_b, sum(wire) AS x_w, sum(pkts) AS x_p, sum(flows) AS x_f, %s AS x_int`, extra, intc)
+	// Rank on what can be summed, which the database can do in any amount
+	// of memory, and count the peers only for the rows that made the list.
+	// Counting them for every group first (two million clients over a week)
+	// is what ran out of memory. Ranked by peers, the distinct pairs are
+	// counted instead, which spills to disk as well.
+	var with string
+	if by == "peers" {
+		with = fmt.Sprintf(`WITH %s,
+			t AS (SELECT %s, count(*) AS x_peers FROM (SELECT DISTINCT %s, x_pc FROM x WHERE x_pc IS NOT NULL) GROUP BY ALL ORDER BY x_peers %s LIMIT %d),
+			p AS (SELECT %s, %s FROM x SEMI JOIN t ON %s GROUP BY ALL)
+			SELECT %s, x_extra, x_b, x_w, x_p, x_f, x_peers, x_int FROM t JOIN p ON %s ORDER BY x_peers %s, x_w DESC`,
+			x, g, g, dir, limit, strings.Join(pg, ", "), sums, strings.Join(on, " AND "), strings.Join(outer, ", "), strings.Join(onP, " AND "), dir)
+	} else {
+		peers := "0"
+		if d.peers != "" {
+			peers = "approx_count_distinct(x_pc)"
+		}
+		with = fmt.Sprintf(`WITH %s,
+			t AS (SELECT %s, %s FROM x GROUP BY ALL ORDER BY %s %s, x_w DESC LIMIT %d),
+			p AS (SELECT %s, %s AS x_peers FROM x SEMI JOIN t ON %s GROUP BY ALL)
+			SELECT %s, x_extra, x_b, x_w, x_p, x_f, coalesce(x_peers, 0), x_int FROM t LEFT JOIN p ON %s ORDER BY %s %s, x_w DESC`,
+			x, g, sums, ord[0], dir, limit, strings.Join(pg, ", "), peers, strings.Join(on, " AND "), strings.Join(outer, ", "),
+			strings.Join(onP, " AND "), ord[0], dir)
+	}
+	rows, err := s.DB.Query(with, args...)
 	if err != nil {
 		return nil, err
 	}
@@ -472,14 +519,18 @@ func (s *Store) topRollup(q Query, rollup string, limit int, order string) ([]To
 	args := []any{q.From.UTC(), q.To.UTC()}
 	if strings.HasPrefix(rollup, "host:") {
 		role, _ := strconv.Atoi(rollup[5:])
-		sqlq = fmt.Sprintf(`SELECT ip, sum(bytes), sum(wire), sum(pkts), sum(flows), bool_or(internal) FROM r_host
+		sqlq = fmt.Sprintf(`SELECT ip, sum(bytes), sum(wire), sum(pkts), sum(flows), bool_or(internal) FROM {r_host}
 			WHERE ts >= ? AND ts < ? AND role = %d GROUP BY 1 ORDER BY %s LIMIT %d`, role, order, limit)
 	} else {
-		sqlq = fmt.Sprintf(`SELECT val, sum(bytes), sum(wire), sum(pkts), sum(flows), false FROM r_dim
+		sqlq = fmt.Sprintf(`SELECT val, sum(bytes), sum(wire), sum(pkts), sum(flows), false FROM {r_dim}
 			WHERE ts >= ? AND ts < ? AND dim = ? GROUP BY 1 ORDER BY %s LIMIT %d`, order, limit)
 		args = append(args, rollup)
 	}
-	rows, err := s.DB.Query(sqlq, args...)
+	sqlq, pre, err := s.rollups(q, sqlq)
+	if err != nil {
+		return nil, err
+	}
+	rows, err := s.DB.Query(sqlq, append(pre, args...)...)
 	if err != nil {
 		return nil, err
 	}
@@ -629,12 +680,17 @@ func StepFor(span time.Duration) time.Duration {
 func (s *Store) SeriesByApp(q Query, k int) (*Series, error) {
 	step := StepFor(q.span())
 	q.From = q.localTrunc(q.From, step)
-	settled := settledTo(q.To, q.usesRollup())
+	rollup := s.rollupOK(q)
+	settled := settledTo(q.To, rollup)
 	var rows *sql.Rows
 	var err error
 	bucket := q.bucketSQL(step)
-	if q.usesRollup() {
-		rows, err = s.DB.Query(fmt.Sprintf(`SELECT %s AS b, app, sum(wire) FROM r_ts WHERE ts >= ? AND ts < ? GROUP BY 1,2`, bucket), q.From.UTC(), q.To.UTC())
+	if rollup {
+		sqlq, pre, rerr := s.rollups(q, fmt.Sprintf(`SELECT %s AS b, app, sum(wire) FROM {r_ts} WHERE ts >= ? AND ts < ? GROUP BY 1,2`, bucket))
+		if rerr != nil {
+			return nil, rerr
+		}
+		rows, err = s.DB.Query(sqlq, append(pre, q.From.UTC(), q.To.UTC())...)
 	} else {
 		where, args, werr := q.where()
 		if werr != nil {
@@ -730,14 +786,22 @@ type Totals struct {
 
 func (s *Store) Totals(q Query) (Totals, error) {
 	var t Totals
-	if q.usesRollup() {
-		err := s.DB.QueryRow(`SELECT coalesce(sum(bytes),0), coalesce(sum(wire),0), coalesce(sum(pkts),0), coalesce(sum(flows),0) FROM r_ts WHERE ts >= ? AND ts < ?`, q.From.UTC(), q.To.UTC()).
-			Scan(&t.Bytes, &t.Wire, &t.Pkts, &t.Flows)
-		if err != nil {
+	if s.rollupOK(q) {
+		one := func(sqlq string, dst ...any) error {
+			sqlq, pre, err := s.rollups(q, sqlq)
+			if err != nil {
+				return err
+			}
+			return s.DB.QueryRow(sqlq, append(pre, q.From.UTC(), q.To.UTC())...).Scan(dst...)
+		}
+		if err := one(`SELECT coalesce(sum(bytes),0), coalesce(sum(wire),0), coalesce(sum(pkts),0), coalesce(sum(flows),0) FROM {r_ts} WHERE ts >= ? AND ts < ?`,
+			&t.Bytes, &t.Wire, &t.Pkts, &t.Flows); err != nil {
 			return t, err
 		}
-		s.DB.QueryRow(`SELECT count(DISTINCT ip) FILTER (WHERE internal), count(DISTINCT ip) FILTER (WHERE NOT internal) FROM r_host WHERE ts >= ? AND ts < ?`, q.From.UTC(), q.To.UTC()).Scan(&t.Hosts, &t.Peers)
-		s.DB.QueryRow(`SELECT count(DISTINCT val) FROM r_dim WHERE ts >= ? AND ts < ? AND dim = 'country' AND val NOT IN ('', '__internal__')`, q.From.UTC(), q.To.UTC()).Scan(&t.Countries)
+		// the addresses counted by grouping them, which spills to disk
+		// where count(DISTINCT) would not
+		one(`SELECT count(*) FILTER (WHERE i), count(*) FILTER (WHERE NOT i) FROM (SELECT ip, internal AS i FROM {r_host} WHERE ts >= ? AND ts < ? GROUP BY ALL)`, &t.Hosts, &t.Peers)
+		one(`SELECT count(DISTINCT val) FROM {r_dim} WHERE ts >= ? AND ts < ? AND dim = 'country' AND val NOT IN ('', '__internal__')`, &t.Countries)
 		return t, nil
 	}
 	where, args, err := q.where()
diff --git a/internal/store/rings.go b/internal/store/rings.go
index 5b87053..bfb7be5 100644
--- a/internal/store/rings.go
+++ b/internal/store/rings.go
@@ -43,10 +43,17 @@ func (s *Store) Rings(q Query, inner, outer string, nInner, nOuter int) (*RingSe
 	if err != nil {
 		return nil, err
 	}
-	sq := fmt.Sprintf(`WITH p AS (SELECT %s AS i, %s AS o, sum(wire) AS w FROM %s WHERE %s AND %s GROUP BY 1,2),
-		ti AS (SELECT i FROM p GROUP BY i ORDER BY sum(w) DESC LIMIT %d)
-		SELECT CASE WHEN i IN (SELECT i FROM ti) THEN i ELSE '__other__' END, CASE WHEN i IN (SELECT i FROM ti) THEN o ELSE '__other__' END, CAST(sum(w) AS DOUBLE)
-		FROM p GROUP BY 1,2`, ie, oe, s.Source(q.From, q.To), where, cond, nInner)
+	// The rows are read again by each step rather than kept, and the
+	// pairs are added up only for the top inner values: adding up every
+	// pair first (each of a week's clients with each server it reached)
+	// is what ran out of memory. The database keeps each inner value's
+	// top outer values and their total; the rest is the difference.
+	sq := fmt.Sprintf(`WITH x AS NOT MATERIALIZED (SELECT %s AS i, %s AS o, wire AS w FROM %s WHERE %s AND %s),
+		ti AS (SELECT i FROM x GROUP BY i ORDER BY sum(w) DESC LIMIT %d),
+		p AS (SELECT CASE WHEN i IN (SELECT i FROM ti) THEN i ELSE '__other__' END AS pi,
+			CASE WHEN i IN (SELECT i FROM ti) THEN o ELSE '__other__' END AS po, sum(w) AS w FROM x GROUP BY 1,2)
+		SELECT pi, po, CAST(w AS DOUBLE), CAST(sum(w) OVER (PARTITION BY pi) AS DOUBLE) FROM p
+		QUALIFY row_number() OVER (PARTITION BY pi ORDER BY w DESC) <= %d`, ie, oe, s.Source(q.From, q.To), where, cond, nInner, nOuter+1)
 	rows, err := s.DB.Query(sq, args...)
 	if err != nil {
 		return nil, err
@@ -57,12 +64,15 @@ func (s *Store) Rings(q Query, inner, outer string, nInner, nOuter int) (*RingSe
 	rs := &RingSet{Inner: []Ring{}, Outer: []Ring{}}
 	for rows.Next() {
 		var r Ring
-		if err := rows.Scan(&r.Parent, &r.Key, &r.Wire); err != nil {
+		var all float64
+		if err := rows.Scan(&r.Parent, &r.Key, &r.Wire, &all); err != nil {
 			return nil, err
 		}
 		by[r.Parent] = append(by[r.Parent], r)
-		tot[r.Parent] += r.Wire
-		rs.Total += r.Wire
+		if _, seen := tot[r.Parent]; !seen {
+			tot[r.Parent] = all
+			rs.Total += all
+		}
 	}
 	if err := rows.Err(); err != nil {
 		return nil, err
@@ -82,15 +92,16 @@ func (s *Store) Rings(q Query, inner, outer string, nInner, nOuter int) (*RingSe
 		rs.Inner = append(rs.Inner, Ring{Key: k, Wire: tot[k]})
 		os := by[k]
 		sort.Slice(os, func(a, b int) bool { return os[a].Wire > os[b].Wire })
-		var rest float64
+		// what is not among the outer values kept goes to Other: those
+		// the database left out and an outer value called Other itself
+		rest := tot[k]
 		for i, o := range os {
 			if i < nOuter && o.Key != "__other__" {
 				rs.Outer = append(rs.Outer, o)
-			} else {
-				rest += o.Wire
+				rest -= o.Wire
 			}
 		}
-		if rest > 0 {
+		if rest > 0.5 {
 			rs.Outer = append(rs.Outer, Ring{Parent: k, Key: "__other__", Wire: rest})
 		}
 	}
diff --git a/internal/store/series.go b/internal/store/series.go
index 63045ae..8432b8c 100644
--- a/internal/store/series.go
+++ b/internal/store/series.go
@@ -23,19 +23,19 @@ const peerExpr = "CASE WHEN dir = 2 THEN client ELSE server END"
 var seriesDims = map[string]seriesDim{
 	"app": {key: "app", cond: "true",
 		rollup: func(b string) string {
-			return `SELECT ` + b + `, app, sum(wire), sum(pkts) FROM r_ts WHERE ts >= ? AND ts < ? GROUP BY 1,2`
+			return `SELECT ` + b + `, app, sum(wire), sum(pkts) FROM {r_ts} WHERE ts >= ? AND ts < ? GROUP BY 1,2`
 		}},
 	"client": {key: "client", cond: "true", hourly: true,
 		rollup: func(b string) string {
-			return `SELECT ` + b + `, ip, sum(wire), sum(pkts) FROM r_host WHERE ts >= ? AND ts < ? AND role = 1 GROUP BY 1,2`
+			return `SELECT ` + b + `, ip, sum(wire), sum(pkts) FROM {r_host} WHERE ts >= ? AND ts < ? AND role = 1 GROUP BY 1,2`
 		}},
 	"server": {key: "server", cond: "true", hourly: true,
 		rollup: func(b string) string {
-			return `SELECT ` + b + `, ip, sum(wire), sum(pkts) FROM r_host WHERE ts >= ? AND ts < ? AND role = 2 GROUP BY 1,2`
+			return `SELECT ` + b + `, ip, sum(wire), sum(pkts) FROM {r_host} WHERE ts >= ? AND ts < ? AND role = 2 GROUP BY 1,2`
 		}},
 	"service": {key: portProtoExpr + " || chr(9) || app", cond: "svc_port <> 0", hourly: true,
 		rollup: func(b string) string {
-			return `SELECT ` + b + `, val, sum(wire), sum(pkts) FROM r_dim WHERE ts >= ? AND ts < ? AND dim = 'port' AND NOT starts_with(val, '0/') GROUP BY 1,2`
+			return `SELECT ` + b + `, val, sum(wire), sum(pkts) FROM {r_dim} WHERE ts >= ? AND ts < ? AND dim = 'port' AND NOT starts_with(val, '0/') GROUP BY 1,2`
 		}},
 	// AS of the sender and of the receiver of each packet, when that side
 	// is the external one.
@@ -70,7 +70,7 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 	}
 	// decide before the start is rounded down to a bucket: rounding makes a
 	// 6-hour range a little longer, which must not switch it to the summaries
-	rollup := q.usesRollup() && d.rollup != nil
+	rollup := s.rollupOK(q) && d.rollup != nil
 	step := StepFor(q.span())
 	if rollup && d.hourly && step < time.Hour {
 		// hourly summaries in 5-minute buckets would show one spike an hour
@@ -82,14 +82,20 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 	var inner string
 	var args []any
 	if rollup {
-		inner = d.rollup(bucket)
-		args = []any{q.From.UTC(), q.To.UTC()}
+		r, pre, err := s.rollups(q, d.rollup(bucket))
+		if err != nil {
+			return nil, err
+		}
+		inner = r
+		args = append(pre, q.From.UTC(), q.To.UTC())
 	} else {
 		where, wargs, err := q.where()
 		if err != nil {
 			return nil, err
 		}
-		inner = fmt.Sprintf(`SELECT %s, %s, sum(wire), sum(pkts) FROM %s WHERE %s AND %s GROUP BY 1,2`, bucket, d.key, s.Source(q.From, q.To), where, d.cond)
+		// the rows themselves: grouped by bucket and value before the top
+		// values are known, a week by client is a group for nearly every row
+		inner = fmt.Sprintf(`SELECT %s, %s, wire, pkts FROM %s WHERE %s AND %s`, bucket, d.key, s.Source(q.From, q.To), where, d.cond)
 		args = wargs
 	}
 	m := "w"
@@ -98,10 +104,14 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 	}
 	// keep the top k values apart, everything else as one series, and count
 	// the values
-	sqlq := fmt.Sprintf(`WITH b AS (%s), b2 AS (SELECT * FROM b AS x(t, k, w, p)),
-			top AS (SELECT k FROM b2 GROUP BY k ORDER BY sum(%s) DESC LIMIT %d)
-		SELECT t, CASE WHEN k IN (SELECT k FROM top) THEN k ELSE '__other__' END AS kk, CAST(sum(%s) AS DOUBLE), (SELECT count(DISTINCT k) FROM b2)
-		FROM b2 GROUP BY 1,2`, inner, m, k, m)
+	// b is read again by each step rather than kept: kept, a week of
+	// buckets by client is more than the memory there is. The values are
+	// counted by grouping them, which spills to disk where count(DISTINCT)
+	// would not.
+	sqlq := fmt.Sprintf(`WITH b AS NOT MATERIALIZED (SELECT * FROM (%s) AS x(t, k, w, p)),
+			top AS (SELECT k FROM b GROUP BY k ORDER BY sum(%s) DESC LIMIT %d)
+		SELECT t, CASE WHEN k IN (SELECT k FROM top) THEN k ELSE '__other__' END AS kk, CAST(sum(%s) AS DOUBLE), (SELECT count(*) FROM (SELECT k FROM b GROUP BY k))
+		FROM b GROUP BY 1,2`, inner, m, k, m)
 	rows, err := s.DB.Query(sqlq, args...)
 	if err != nil {
 		return nil, err
diff --git a/internal/store/store.go b/internal/store/store.go
index 3952bdd..ddf9012 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -14,6 +14,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	duckdb "github.com/duckdb/duckdb-go/v2"
@@ -112,6 +113,14 @@ type Store struct {
 	segMu    sync.RWMutex
 	segCache []segment
 
+	// the summaries by device and interface (views.go): ready once the
+	// flow files sealed before they existed are summarised too
+	viewsReady atomic.Bool
+	closing    atomic.Bool
+	bg         sync.WaitGroup
+	bgCtx      context.Context
+	bgStop     context.CancelFunc
+
 	dsMu sync.Mutex // guards the cached start of the data
 	dsAt time.Time  // when dsTS was read
 	dsTS time.Time
@@ -183,7 +192,8 @@ func Open(opt Options) (*Store, error) {
 		return nil, fmt.Errorf("cannot open the database in %s: %w", opt.Dir, err)
 	}
 	s := &Store{opt: opt, conn: c, DB: sql.OpenDB(c)}
-	for _, q := range schema {
+	s.bgCtx, s.bgStop = context.WithCancel(context.Background())
+	for _, q := range append(schema, viewSchema...) {
 		if _, err := s.DB.Exec(q); err != nil {
 			return nil, fmt.Errorf("schema: %w", err)
 		}
@@ -197,6 +207,8 @@ func Open(opt Options) (*Store, error) {
 	if err := s.repairWrapped(); err != nil {
 		log.Printf("store: %v", err)
 	}
+	s.bg.Add(1)
+	go s.buildViews()
 	log.Printf("store: %s, DuckDB memory limit %d MB, %d threads", opt.Dir, memMB, opt.Threads)
 	return s, nil
 }
@@ -322,10 +334,23 @@ func (s *Store) Seal(now time.Time) error {
 	if _, err := s.DB.Exec(q, boundary); err != nil {
 		return fmt.Errorf("seal: %w", err)
 	}
-	if _, err := s.DB.Exec(`INSERT INTO segments VALUES (?, ?, ?, ?)`, path, mn.Time, mx.Time, n); err != nil {
+	// the file, the summaries of what is in it and the rows leaving the hot
+	// table change together, so no query counts them twice or not at all
+	tx, err := s.DB.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if _, err := tx.Exec(`INSERT INTO segments VALUES (?, ?, ?, ?)`, path, mn.Time, mx.Time, n); err != nil {
+		return err
+	}
+	if err := sealViews(tx, boundary); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(`DELETE FROM hot WHERE ts < ?`, boundary); err != nil {
 		return err
 	}
-	if _, err := s.DB.Exec(`DELETE FROM hot WHERE ts < ?`, boundary); err != nil {
+	if err := tx.Commit(); err != nil {
 		return err
 	}
 	s.DB.Exec(`CHECKPOINT`)
@@ -402,7 +427,7 @@ func (s *Store) Retain(now time.Time) error {
 		rows += n
 		return nil
 	}
-	for _, t := range []string{"r_ts", "r_host", "r_dim"} {
+	for _, t := range append([]string{"r_ts", "r_host", "r_dim"}, viewTables...) {
 		if err := del(t, rollCut); err != nil {
 			return err
 		}
@@ -501,6 +526,9 @@ func (s *Store) Usage() Usage {
 
 // Close flushes and closes everything.
 func (s *Store) Close() error {
+	s.closing.Store(true)
+	s.bgStop()
+	s.bg.Wait()
 	s.wmu.Lock()
 	defer s.wmu.Unlock()
 	for _, a := range []*duckdb.Appender{s.appHot, s.appTS, s.appHost, s.appDim, s.appIfc} {
diff --git a/internal/store/views.go b/internal/store/views.go
new file mode 100644
index 0000000..5faeca7
--- /dev/null
+++ b/internal/store/views.go
@@ -0,0 +1,308 @@
+package store
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Summaries by device and by interface.
+//
+// The summaries in r_ts, r_host and r_dim are of all the traffic. Narrowed
+// to one device or one interface, which is how most of the pages are read,
+// a range of days used to be answered from the flow records themselves: a
+// week is tens of millions of rows, read two or three times over for one
+// chart. These tables hold the same summaries once per device and once per
+// interface, so those ranges are read from them instead.
+//
+// iface is the exporter for a device ("10.0.0.1") or "exporter/ifindex"
+// for an interface. side says how the traffic used the interface:
+// sideIn came in through it, sideOut went out through it, sideBoth came in
+// and went out through it (in and out the same interface). For a device
+// side is sideDevice. A flow row is counted once per side, so an interface
+// filter adds up sides 1 to 3 without counting anything twice.
+//
+// They are written when the flow records are sealed into a file, from
+// exactly the rows sealed, in the same transaction that takes them out of
+// the hot table; the rows not sealed yet are read from the hot table, which
+// is small. They count a device's own traffic: rows marked as duplicates of
+// another device's report are included, as they are for any query that
+// keeps to one device.
+const (
+	sideDevice = 0
+	sideIn     = 1
+	sideOut    = 2
+	sideBoth   = 3
+)
+
+var viewSchema = []string{
+	`CREATE TABLE IF NOT EXISTS r_ts_if (ts TIMESTAMP, iface VARCHAR, side UTINYINT, app VARCHAR, bytes UBIGINT, wire UBIGINT, pkts UBIGINT, flows UBIGINT)`,
+	`CREATE TABLE IF NOT EXISTS r_host_if (ts TIMESTAMP, iface VARCHAR, side UTINYINT, ip VARCHAR, role UTINYINT, internal BOOLEAN, bytes UBIGINT, wire UBIGINT, pkts UBIGINT, flows UBIGINT)`,
+	`CREATE TABLE IF NOT EXISTS r_dim_if (ts TIMESTAMP, iface VARCHAR, side UTINYINT, dim VARCHAR, val VARCHAR, bytes UBIGINT, wire UBIGINT, pkts UBIGINT, flows UBIGINT)`,
+}
+
+var viewTables = []string{"r_ts_if", "r_host_if", "r_dim_if"}
+
+// viewsDone is the meta key set once the summaries cover every flow file.
+const viewsDone = "views_v1"
+
+// tsStep is the bucket of r_ts_if: the finest step a range long enough
+// for the summaries is drawn with.
+const tsStep = "INTERVAL 5 MINUTE"
+
+// The dimensions of r_dim_if, as r_dim has them.
+var viewDims = [][2]string{
+	{"app", "app"},
+	{"port", portProtoExpr + " || chr(9) || app"},
+	{"country", "CASE WHEN dir = 3 THEN '__internal__' ELSE peer_cc END"},
+	{"proto", "CAST(proto AS VARCHAR)"},
+	{"dir", "CAST(dir AS VARCHAR)"},
+	{"exporter", "exporter"},
+	{"segment", "segment"},
+	{"encap", "CAST(encap AS VARCHAR)"},
+	{"asn", "CAST(peer_asn AS VARCHAR) || chr(9) || peer_org"},
+}
+
+// viewInserts are the statements that add the summaries of the flow rows
+// in src (a table or subquery) to the tables. Each reads src once: a row
+// is repeated for the device, the interface it came in through and the
+// one it went out through (and for r_dim once more for each dimension)
+// with unnest, rather than src being read again for each of them.
+func viewInserts(src string) []string {
+	// a counter that wrapped below zero (written by 1.6.0 and earlier) is
+	// left out rather than allowed to overflow the sum
+	ok := func(c string) string { return fmt.Sprintf("sum(CASE WHEN %s > %d THEN 0 ELSE %s END)::UBIGINT", c, wrapLimit, c) }
+	sums := ok("bytes") + ", " + ok("wire") + ", " + ok("pkts") + ", " + ok("flows")
+	v := `(SELECT *, unnest(list_filter([
+			{'f': exporter, 's': 0::UTINYINT, 'ok': true},
+			{'f': exporter || '/' || in_if, 's': (CASE WHEN in_if = out_if THEN 3 ELSE 1 END)::UTINYINT, 'ok': in_if <> 0},
+			{'f': exporter || '/' || out_if, 's': 2::UTINYINT, 'ok': out_if <> 0 AND out_if <> in_if}], x -> x.ok)) AS v FROM ` + src + `)`
+	var dims []string
+	for _, d := range viewDims {
+		okd := "true"
+		if d[0] == "asn" {
+			okd = "peer_asn <> 0"
+		}
+		dims = append(dims, fmt.Sprintf(`{'d': '%s', 'v': %s, 'ok': %s}`, d[0], d[1], okd))
+	}
+	return []string{
+		`INSERT INTO r_ts_if SELECT time_bucket(` + tsStep + `, ts), v.f, v.s, app, ` + sums + ` FROM ` + v + ` GROUP BY ALL`,
+		`INSERT INTO r_host_if SELECT date_trunc('hour', ts), v.f, v.s, h.ip, h.role, h.internal, ` + sums + ` FROM (
+			SELECT *, unnest([{'ip': client, 'role': 1::UTINYINT, 'internal': cli_int}, {'ip': server, 'role': 2::UTINYINT, 'internal': srv_int}]) AS h FROM ` + v + `) GROUP BY ALL`,
+		`INSERT INTO r_dim_if SELECT date_trunc('hour', ts), v.f, v.s, d.d, d.v, ` + sums + ` FROM (
+			SELECT *, unnest(list_filter([` + strings.Join(dims, ", ") + `], x -> x.ok)) AS d FROM ` + v + `) GROUP BY ALL`,
+	}
+}
+
+// view is which summaries answer a query: all traffic (key "") or one
+// device or interface and the sides of it.
+type view struct {
+	key   string
+	sides []int
+}
+
+// view works out the view a query's filters describe. ok is false when
+// they narrow it some other way, and only the flow records can answer.
+func (q Query) view() (view, bool) {
+	if len(q.Filters) == 0 {
+		return view{}, true
+	}
+	var key string
+	sides := []int{sideIn, sideOut, sideBoth}
+	device := ""
+	for _, f := range q.Filters {
+		if f.Neg {
+			return view{}, false
+		}
+		var want []int
+		switch f.Field {
+		case "exporter":
+			if device != "" && device != f.Value {
+				return view{}, false
+			}
+			device = f.Value
+			continue
+		case "iface":
+			want = []int{sideIn, sideOut, sideBoth}
+		case "iface_in":
+			want = []int{sideIn, sideBoth}
+		case "iface_out":
+			want = []int{sideOut, sideBoth}
+		default:
+			return view{}, false
+		}
+		if key != "" && key != f.Value {
+			return view{}, false
+		}
+		key = f.Value
+		sides = slices.DeleteFunc(sides, func(x int) bool { return !slices.Contains(want, x) })
+	}
+	switch {
+	case key == "":
+		return view{key: device, sides: []int{sideDevice}}, true
+	case device != "" && !strings.HasPrefix(key, device+"/"):
+		return view{}, false
+	case len(sides) == 0:
+		return view{}, false
+	}
+	return view{key: key, sides: sides}, true
+}
+
+// rollupOK reports whether the summaries answer q.
+func (s *Store) rollupOK(q Query) bool {
+	if q.span() <= 6*time.Hour {
+		return false
+	}
+	v, ok := q.view()
+	return ok && (v.key == "" || s.viewsReady.Load())
+}
+
+// rollups replaces the placeholders {r_ts}, {r_host} and {r_dim} in sqlq
+// with the tables or subqueries that hold q's summaries, and returns the
+// arguments those need, which come before any of sqlq's own.
+func (s *Store) rollups(q Query, sqlq string) (string, []any, error) {
+	v, _ := q.view()
+	var args []any
+	for _, t := range []string{"r_ts", "r_host", "r_dim"} {
+		ph := "{" + t + "}"
+		if !strings.Contains(sqlq, ph) {
+			continue
+		}
+		if v.key == "" {
+			sqlq = strings.ReplaceAll(sqlq, ph, t)
+			continue
+		}
+		src, a, err := s.viewSource(q, v, t)
+		if err != nil {
+			return "", nil, err
+		}
+		for range strings.Count(sqlq, ph) {
+			args = append(args, a...)
+		}
+		sqlq = strings.ReplaceAll(sqlq, ph, src)
+	}
+	return sqlq, args, nil
+}
+
+// viewSource is one summary table for a view, shaped like the table of all
+// traffic it stands in for: the sealed rows from the view's table and the
+// rows still in the hot table, summarised on the fly.
+func (s *Store) viewSource(q Query, v view, table string) (string, []any, error) {
+	where, wargs, err := q.where()
+	if err != nil {
+		return "", nil, err
+	}
+	sides := make([]string, len(v.sides))
+	for i, x := range v.sides {
+		sides[i] = strconv.Itoa(x)
+	}
+	sel := fmt.Sprintf(`FROM %s_if WHERE iface = ? AND side IN (%s)`, table, strings.Join(sides, ","))
+	args := []any{v.key}
+	var hot []string
+	switch table {
+	case "r_ts":
+		sel = "SELECT ts, app, bytes, wire, pkts, flows " + sel
+		hot = []string{"SELECT ts, app, bytes, wire, pkts, flows FROM hot WHERE " + where}
+	case "r_host":
+		sel = "SELECT ts, ip, role, internal, bytes, wire, pkts, flows " + sel
+		hot = []string{
+			"SELECT ts, client, 1::UTINYINT, cli_int, bytes, wire, pkts, flows FROM hot WHERE " + where,
+			"SELECT ts, server, 2::UTINYINT, srv_int, bytes, wire, pkts, flows FROM hot WHERE " + where,
+		}
+	case "r_dim":
+		sel = "SELECT ts, dim, val, bytes, wire, pkts, flows " + sel
+		for _, d := range viewDims {
+			c := where
+			if d[0] == "asn" {
+				c += " AND peer_asn <> 0"
+			}
+			hot = append(hot, fmt.Sprintf("SELECT ts, '%s', %s, bytes, wire, pkts, flows FROM hot WHERE %s", d[0], d[1], c))
+		}
+	}
+	for range hot {
+		args = append(args, wargs...)
+	}
+	return "(" + sel + " UNION ALL " + strings.Join(hot, " UNION ALL ") + ")", args, nil
+}
+
+// sealViews adds the summaries of the hot rows before boundary, in the
+// transaction that deletes them from the hot table.
+func sealViews(tx *sql.Tx, boundary time.Time) error {
+	src := fmt.Sprintf("(SELECT * FROM hot WHERE ts < TIMESTAMP '%s')", boundary.UTC().Format("2006-01-02 15:04:05.999999"))
+	for _, q := range viewInserts(src) {
+		if _, err := tx.Exec(q); err != nil {
+			return fmt.Errorf("summaries: %w", err)
+		}
+	}
+	return nil
+}
+
+// buildViews summarises the flow files sealed before the summaries
+// existed, once per data directory, a file at a time so the memory it
+// needs is a file's worth. Until it is done the pages read those ranges
+// from the flow records as before.
+func (s *Store) buildViews() {
+	defer s.bg.Done()
+	if s.Meta(viewsDone) != "" {
+		s.viewsReady.Store(true)
+		return
+	}
+	// Start over: an earlier run may have stopped half way. The files
+	// sealed from here on are summarised by the seal itself, so the list
+	// is taken, and the tables emptied, with sealing held off.
+	s.wmu.Lock()
+	for _, t := range viewTables {
+		if _, err := s.DB.Exec(`DELETE FROM ` + t); err != nil {
+			s.wmu.Unlock()
+			log.Printf("store: summaries: %v", err)
+			return
+		}
+	}
+	s.segMu.RLock()
+	segs := append([]segment(nil), s.segCache...)
+	s.segMu.RUnlock()
+	s.wmu.Unlock()
+	t0 := time.Now()
+	for i, g := range segs {
+		if s.closing.Load() {
+			return
+		}
+		src := "(SELECT * FROM read_parquet('" + sqlPath(g.path) + "'))"
+		tx, err := s.DB.BeginTx(s.bgCtx, nil)
+		if err != nil {
+			log.Printf("store: summaries: %v", err)
+			return
+		}
+		for _, q := range viewInserts(src) {
+			if _, err = tx.ExecContext(s.bgCtx, q); err != nil {
+				break
+			}
+		}
+		if err != nil {
+			tx.Rollback()
+			if !s.closing.Load() {
+				log.Printf("store: summaries of %s: %v", g.path, err)
+			}
+			return
+		}
+		if err := tx.Commit(); err != nil {
+			log.Printf("store: summaries: %v", err)
+			return
+		}
+		if (i+1)%50 == 0 {
+			log.Printf("store: summarising flow files by device and interface: %d of %d", i+1, len(segs))
+		}
+	}
+	if err := s.SetMeta(viewsDone, time.Now().UTC().Format(time.RFC3339)); err != nil {
+		log.Printf("store: summaries: %v", err)
+		return
+	}
+	s.viewsReady.Store(true)
+	if len(segs) > 0 {
+		log.Printf("store: summarised %d flow files by device and interface in %s", len(segs), time.Since(t0).Round(time.Second))
+	}
+}
diff --git a/internal/store/views_test.go b/internal/store/views_test.go
new file mode 100644
index 0000000..6e30a04
--- /dev/null
+++ b/internal/store/views_test.go
@@ -0,0 +1,104 @@
+package store
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+)
+
+// A device or an interface over days is answered from its summaries, and
+// the answer is the one the flow records give: sealed rows and rows still
+// in the hot table, in, out and in-and-out of the same interface, and the
+// device's own duplicates.
+func TestViewsMatchRecords(t *testing.T) {
+	s, err := Open(Options{Dir: t.TempDir(), Threads: 2})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	now := time.Now().UTC().Truncate(time.Minute)
+	var rows []Row
+	for m := 0; m < 12*60; m += 7 {
+		ts := now.Add(-time.Duration(m) * time.Minute)
+		for i := 0; i < 6; i++ {
+			r := Row{TS: ts, Client: fmt.Sprintf("198.51.100.%d", i), Server: fmt.Sprintf("203.0.113.%d", m%5),
+				Cli4: 0xC6336400 + uint32(i), Srv4: 0xCB007100 + uint32(m%5), SvcPort: uint16(443 + i%2), Proto: 6,
+				App: []string{"HTTPS", "HTTP"}[i%2], Exporter: "10.0.0.1", InIf: uint32(1 + i%3), OutIf: uint32(1 + m%3),
+				Dir: DirTransit, PeerCC: "US", PeerASN: uint32(64500 + i), PeerOrg: "Org",
+				Bytes: uint64(1000 * (i + 1)), Wire: uint64(1100 * (i + 1)), Pkts: uint64(i + 1), Flows: 1, Dup: i == 5}
+			r.Src, r.Dst = r.Client, r.Server
+			rows = append(rows, r)
+		}
+	}
+	if err := s.AppendRows(rows); err != nil {
+		t.Fatal(err)
+	}
+	// everything up to two hours ago goes into a file and the summaries
+	if err := s.Seal(now.Add(-2 * time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	for !s.viewsReady.Load() {
+		time.Sleep(10 * time.Millisecond)
+	}
+	// long ranges start on the hour, as the API makes them
+	from, to := now.Add(-11*time.Hour).Truncate(time.Hour), now.Add(time.Minute)
+	for _, f := range [][]Filter{
+		{{Field: "exporter", Value: "10.0.0.1"}},
+		{{Field: "iface", Value: "10.0.0.1/1"}},
+		{{Field: "iface_in", Value: "10.0.0.1/2"}},
+		{{Field: "iface_out", Value: "10.0.0.1/2"}},
+		{{Field: "iface", Value: "10.0.0.1/3"}, {Field: "iface_in", Value: "10.0.0.1/3"}},
+	} {
+		q := Query{From: from, To: to, Filters: f}
+		if !s.rollupOK(q) {
+			t.Fatalf("%v: not answered from the summaries", f)
+		}
+		answer := func() string {
+			tot, err := s.Totals(q)
+			if err != nil {
+				t.Fatal(err)
+			}
+			// (the host and peer counts are worked out differently from
+			// summaries, for all traffic as well, so they are not compared)
+			out := fmt.Sprintf("totals %d %d %d %d cc %d\n", tot.Bytes, tot.Wire, tot.Pkts, tot.Flows, tot.Countries)
+			for _, dim := range []string{"client", "server", "app", "port", "country", "asn"} {
+				top, err := s.TopN(q, dim, 10)
+				if err != nil {
+					t.Fatal(err)
+				}
+				for _, r := range top {
+					out += fmt.Sprintf("%s %s %d %d\n", dim, r.Key, r.Wire, r.Pkts)
+				}
+			}
+			for _, by := range []string{"app", "client", "service"} {
+				se, err := s.SeriesBy(q, by, "wire", 8)
+				if err != nil {
+					t.Fatal(err)
+				}
+				out += fmt.Sprintf("series %s %v %v\n", by, se.Names, se.Totals)
+			}
+			se, err := s.SeriesByApp(q, 8)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var sum float64
+			for _, v := range se.Values {
+				for _, x := range v {
+					sum += x
+				}
+			}
+			return out + fmt.Sprintf("apps %v %.3f\n", se.Names, sum)
+		}
+		// equal values may come in either order
+		sorted := func(x string) string { l := strings.Split(x, "\n"); sort.Strings(l); return strings.Join(l, "\n") }
+		got := sorted(answer())
+		s.viewsReady.Store(false) // the same from the flow records
+		want := sorted(answer())
+		s.viewsReady.Store(true)
+		if got != want {
+			t.Errorf("%v:\nsummaries:\n%s\nrecords:\n%s", f, got, want)
+		}
+	}
+}

From e34568d537e3b2ff450433fd38df7115eae62a5d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 23 Oct 2026 23:44:38 +0000
Subject: [PATCH 2/3] Store changes

---
 CHANGELOG.md                       |   4 +
 cmd/traffic66/main.go              |   2 +-
 internal/pipeline/pipeline_test.go |   9 +-
 internal/store/chartedges_test.go  |   2 +-
 internal/store/query.go            | 132 ++++++++++++++++++++---------
 internal/store/rings.go            |  46 +++++++---
 internal/store/series.go           |  34 +++++++-
 internal/store/views.go            |  30 +++----
 internal/store/views_test.go       |  38 ++++++++-
 9 files changed, 224 insertions(+), 73 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b73c599..43bf721 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,10 @@
 The release notes on GitHub are taken from this file: the section whose
 heading is the version number.
 
+## 2.5.1
+
+- Long ranges on large installations no longer run out of database memory, and a device or an interface over days is read from summaries of its own (built once in the background for the data already stored).
+
 ## 2.5.0
 
 - Overview: inbound and outbound follow the interface charted above.
diff --git a/cmd/traffic66/main.go b/cmd/traffic66/main.go
index 8ab754a..c36bcaf 100644
--- a/cmd/traffic66/main.go
+++ b/cmd/traffic66/main.go
@@ -55,7 +55,7 @@ var version = ""
 // CHANGELOG.md. TestDevVersionMatchesChangelog keeps the two in step: the
 // default used to be a hardcoded 0.1.0-dev, and because nothing checked it
 // it stayed that way for two major versions.
-const devVersion = "2.5.0"
+const devVersion = "2.5.1"
 
 func init() {
 	if version == "" {
diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go
index bbc4373..3661f94 100644
--- a/internal/pipeline/pipeline_test.go
+++ b/internal/pipeline/pipeline_test.go
@@ -250,10 +250,17 @@ func TestRetention(t *testing.T) {
 	if err := st.Retain(now.Add(48 * time.Hour)); err != nil {
 		t.Fatal(err)
 	}
-	tot, _ = st.Totals(all)
+	// the flow records are gone; the device's summaries stay, as the
+	// summaries of all traffic do
+	records := all
+	records.Filters = append(records.Filters, store.Filter{Field: "client", Value: "10.1.1.1"})
+	tot, _ = st.Totals(records)
 	if tot.Flows != 0 || st.Usage().Segments != 0 {
 		t.Fatalf("expired data still visible: %+v, %+v", tot, st.Usage())
 	}
+	if tot, _ = st.Totals(all); tot.Flows != 2 {
+		t.Fatalf("the device's summaries went with its records: %+v", tot)
+	}
 }
 
 // TestLongRangeConsistent checks that for a range served from summaries,
diff --git a/internal/store/chartedges_test.go b/internal/store/chartedges_test.go
index 36f45f4..b699baa 100644
--- a/internal/store/chartedges_test.go
+++ b/internal/store/chartedges_test.go
@@ -79,7 +79,7 @@ func TestFirstBucketCountsOnlyTheCoveredPart(t *testing.T) {
 		t.Fatal(err)
 	}
 	q := Query{From: base.Add(-2 * time.Hour), To: base.Add(4 * time.Hour)}
-	if s.rollupOK(q) {
+	if s.rollupOK(q, false) {
 		t.Skip("range answered from the summaries")
 	}
 	for name, get := range map[string]func() (*Series, error){
diff --git a/internal/store/query.go b/internal/store/query.go
index 38e4345..39c21f1 100644
--- a/internal/store/query.go
+++ b/internal/store/query.go
@@ -57,7 +57,6 @@ func (q Query) bucketSQL(step time.Duration) string {
 
 func (q Query) span() time.Duration { return q.To.Sub(q.From) }
 
-
 var dirNames = map[string]int{"outbound": DirOutbound, "inbound": DirInbound, "internal": DirInternal, "transit": DirTransit}
 
 // DirName is the stored direction's API name.
@@ -389,7 +388,7 @@ func (s *Store) TopNBy(q Query, dim string, limit int, by string, asc ...bool) (
 	if limit <= 0 || limit > 1000 {
 		limit = 66
 	}
-	if s.rollupOK(q) && d.rollup != "" {
+	if d.rollup != "" && s.rollupOK(q, strings.HasPrefix(d.rollup, "host:")) {
 		return s.topRollup(q, d.rollup, limit, ord[1]+" "+dir)
 	}
 	where, args, err := q.where()
@@ -421,21 +420,23 @@ func (s *Store) TopNBy(q Query, dim string, limit int, by string, asc ...bool) (
 	if len(v4conds) > 0 {
 		v4 = strings.Join(v4conds, " AND ")
 	}
-	var keySel, groups, outer, on []string
+	// keySel and keySel6: the keys of the IPv4 rows (an address as its
+	// number, its text NULL) and of the others (the number 0, the text),
+	// read as two parts so the IPv4 rows never read the address text,
+	// which takes longer than everything else the query reads
+	var keySel, keySel6, groups, outer, on []string
 	for _, kp := range parts {
-		for j, g := range kp.groups {
-			if kp.v4cond != "" {
-				// the number for IPv4 rows, the text for the others
-				n, t := kp.inner[0], kp.inner6[1]
-				n, t = n[:strings.LastIndex(n, " AS ")], t[:strings.LastIndex(t, " AS ")]
-				keySel = append(keySel, "CASE WHEN "+v4+" THEN "+n+" ELSE 0 END AS "+kp.groups[0],
-					"CASE WHEN "+v4+" THEN NULL ELSE "+t+" END AS "+kp.groups6[0])
-				groups = append(groups, kp.groups[0], kp.groups6[0])
-				break
-			}
-			keySel = append(keySel, kp.inner[j])
-			groups = append(groups, g)
+		if kp.v4cond != "" {
+			n := kp.inner[0][:strings.LastIndex(kp.inner[0], " AS ")]
+			t := kp.inner6[1][:strings.LastIndex(kp.inner6[1], " AS ")]
+			keySel = append(keySel, n+" AS "+kp.groups[0], "NULL::VARCHAR AS "+kp.groups6[0])
+			keySel6 = append(keySel6, "0::UINTEGER AS "+kp.groups[0], t+" AS "+kp.groups6[0])
+			groups = append(groups, kp.groups[0], kp.groups6[0])
+			continue
 		}
+		keySel = append(keySel, kp.inner...)
+		keySel6 = append(keySel6, kp.inner...)
+		groups = append(groups, kp.groups...)
 	}
 	for i, e := range []string{d.expr, d.expr2, d.expr3} {
 		if e == "" {
@@ -447,10 +448,18 @@ func (s *Store) TopNBy(q Query, dim string, limit int, by string, asc ...bool) (
 	// pg: the group columns of a step joined back to t, renamed so the
 	// two sides of the join do not both have them
 	var pg, onP []string
+	// plain equality, which the database joins by hashing; the address
+	// text is NULL for IPv4 rows, so it is compared as '' there
 	for _, g := range groups {
-		on = append(on, "t."+g+" IS NOT DISTINCT FROM x."+g)
+		text := strings.HasSuffix(g, "s") // an address as text: NULL for IPv4 rows
+		if text {
+			on = append(on, "coalesce(t."+g+", '') = coalesce(x."+g+", '')")
+			onP = append(onP, "coalesce(t."+g+", '') = coalesce(p.p_"+g+", '')")
+		} else {
+			on = append(on, "t."+g+" = x."+g)
+			onP = append(onP, "t."+g+" = p.p_"+g)
+		}
 		pg = append(pg, "x."+g+" AS p_"+g)
-		onP = append(onP, "t."+g+" IS NOT DISTINCT FROM p.p_"+g)
 	}
 	extra := "''"
 	if d.extra != "" {
@@ -460,16 +469,39 @@ func (s *Store) TopNBy(q Query, dim string, limit int, by string, asc ...bool) (
 	if d.intCol != "" {
 		intc = d.intCol
 	}
-	peerCol := "NULL"
+	// the peer as a hash of its number, or of its text in the IPv6 part,
+	// for the same reason the keys are read that way; a flow's two ends
+	// are always the same family
+	peerCol, peerCol6 := "NULL::UBIGINT", "NULL::UBIGINT"
 	if d.peers != "" {
-		peerCol = d.peers
+		peerCol = "hash(" + map[string]string{"client": "cli4", "server": "srv4"}[d.peers] + ")"
+		peerCol6 = "hash(" + d.peers + ")"
+		if len(v4conds) == 0 {
+			peerCol = peerCol6
+		}
 	}
 	// x is the rows, with their keys, read again by each step below rather
 	// than kept: kept, a week of rows is more than the memory there is.
 	// only the columns the steps below use: the flow files are read by
 	// column, and every other one would be read for nothing
-	x := fmt.Sprintf(`x AS NOT MATERIALIZED (SELECT bytes, wire, pkts, flows, peer_cc, peer_org, app, dir, cli_int, srv_int, %s, %s AS x_pc FROM %s WHERE %s)`,
-		strings.Join(keySel, ", "), peerCol, s.Source(q.From, q.To), cond)
+	src := s.Source(q.From, q.To)
+	// only the columns the steps use: the parts of x are not trimmed to
+	// what is used after it, and every text column read costs as much as
+	// the rest of the query
+	cols := "bytes, wire, pkts, flows"
+	for _, c := range []string{"peer_cc", "peer_org", "app", "dir", "cli_int", "srv_int"} {
+		if strings.Contains(extra+intc, c) {
+			cols += ", " + c
+		}
+	}
+	x := fmt.Sprintf(`x AS NOT MATERIALIZED (SELECT %s, %s, %s AS x_pc FROM %s WHERE (%s) AND %s`,
+		cols, strings.Join(keySel, ", "), peerCol, src, cond, v4)
+	if len(v4conds) > 0 {
+		x += fmt.Sprintf(` UNION ALL SELECT %s, %s, %s FROM %s WHERE (%s) AND NOT (%s)`,
+			cols, strings.Join(keySel6, ", "), peerCol6, src, cond, v4)
+		args = append(append([]any{}, args...), args...)
+	}
+	x += ")"
 	g := strings.Join(groups, ", ")
 	sums := fmt.Sprintf(`%s AS x_extra, sum(bytes) AS x_b, sum(wire) AS x_w, sum(pkts) AS x_p, sum(flows) AS x_f, %s AS x_int`, extra, intc)
 	// Rank on what can be summed, which the database can do in any amount
@@ -489,12 +521,16 @@ func (s *Store) TopNBy(q Query, dim string, limit int, by string, asc ...bool) (
 		if d.peers != "" {
 			peers = "approx_count_distinct(x_pc)"
 		}
+		// ranked on the one measure alone: the groups of a week by client
+		// are millions, and each sum carried for all of them made the
+		// ranking several times slower; the sums are taken for the top
+		// rows only, with their peers
 		with = fmt.Sprintf(`WITH %s,
-			t AS (SELECT %s, %s FROM x GROUP BY ALL ORDER BY %s %s, x_w DESC LIMIT %d),
-			p AS (SELECT %s, %s AS x_peers FROM x SEMI JOIN t ON %s GROUP BY ALL)
-			SELECT %s, x_extra, x_b, x_w, x_p, x_f, coalesce(x_peers, 0), x_int FROM t LEFT JOIN p ON %s ORDER BY %s %s, x_w DESC`,
-			x, g, sums, ord[0], dir, limit, strings.Join(pg, ", "), peers, strings.Join(on, " AND "), strings.Join(outer, ", "),
-			strings.Join(onP, " AND "), ord[0], dir)
+			t AS (SELECT %s, %s AS x_rank FROM x GROUP BY ALL ORDER BY x_rank %s LIMIT %d),
+			p AS (SELECT %s, %s, %s AS x_peers FROM x SEMI JOIN t ON %s GROUP BY ALL)
+			SELECT %s, x_extra, x_b, x_w, x_p, x_f, x_peers, x_int FROM t JOIN p ON %s ORDER BY x_rank %s, x_w DESC`,
+			x, g, ord[1], dir, limit, strings.Join(pg, ", "), sums, peers, strings.Join(on, " AND "), strings.Join(outer, ", "),
+			strings.Join(onP, " AND "), dir)
 	}
 	rows, err := s.DB.Query(with, args...)
 	if err != nil {
@@ -680,7 +716,7 @@ func StepFor(span time.Duration) time.Duration {
 func (s *Store) SeriesByApp(q Query, k int) (*Series, error) {
 	step := StepFor(q.span())
 	q.From = q.localTrunc(q.From, step)
-	rollup := s.rollupOK(q)
+	rollup := s.rollupOK(q, false)
 	settled := settledTo(q.To, rollup)
 	var rows *sql.Rows
 	var err error
@@ -786,7 +822,7 @@ type Totals struct {
 
 func (s *Store) Totals(q Query) (Totals, error) {
 	var t Totals
-	if s.rollupOK(q) {
+	if s.rollupOK(q, false) {
 		one := func(sqlq string, dst ...any) error {
 			sqlq, pre, err := s.rollups(q, sqlq)
 			if err != nil {
@@ -798,11 +834,19 @@ func (s *Store) Totals(q Query) (Totals, error) {
 			&t.Bytes, &t.Wire, &t.Pkts, &t.Flows); err != nil {
 			return t, err
 		}
-		// the addresses counted by grouping them, which spills to disk
-		// where count(DISTINCT) would not
-		one(`SELECT count(*) FILTER (WHERE i), count(*) FILTER (WHERE NOT i) FROM (SELECT ip, internal AS i FROM {r_host} WHERE ts >= ? AND ts < ? GROUP BY ALL)`, &t.Hosts, &t.Peers)
 		one(`SELECT count(DISTINCT val) FROM {r_dim} WHERE ts >= ? AND ts < ? AND dim = 'country' AND val NOT IN ('', '__internal__')`, &t.Countries)
-		return t, nil
+		if v, _ := q.view(); v.key == "" {
+			one(`SELECT count(DISTINCT ip) FILTER (WHERE internal), count(DISTINCT ip) FILTER (WHERE NOT internal) FROM {r_host} WHERE ts >= ? AND ts < ?`, &t.Hosts, &t.Peers)
+			return t, nil
+		}
+		// a device or an interface has no summary by address: its
+		// addresses are counted from the flow records
+		where, args, err := q.where()
+		if err != nil {
+			return t, err
+		}
+		err = s.DB.QueryRow(s.hostCounts(q, where), args...).Scan(&t.Hosts, &t.Peers)
+		return t, err
 	}
 	where, args, err := q.where()
 	if err != nil {
@@ -810,20 +854,27 @@ func (s *Store) Totals(q Query) (Totals, error) {
 	}
 	src := s.Source(q.From, q.To)
 	err = s.DB.QueryRow(fmt.Sprintf(`SELECT coalesce(sum(bytes),0), coalesce(sum(wire),0), coalesce(sum(pkts),0), coalesce(sum(flows),0),
-		count(DISTINCT CASE WHEN NOT srv_int THEN server WHEN NOT cli_int THEN client END),
 		count(DISTINCT nullif(peer_cc, ''))
 		FROM %s WHERE %s`, src, where), args...).
-		Scan(&t.Bytes, &t.Wire, &t.Pkts, &t.Flows, &t.Peers, &t.Countries)
+		Scan(&t.Bytes, &t.Wire, &t.Pkts, &t.Flows, &t.Countries)
 	if err != nil {
 		return t, err
 	}
-	// internal addresses on either side, each counted once (a host is often
-	// a client in one conversation and a server in another)
-	err = s.DB.QueryRow(fmt.Sprintf(`SELECT count(*) FROM (SELECT client FROM %s WHERE %s AND cli_int UNION SELECT server FROM %s WHERE %s AND srv_int)`,
-		src, where, src, where), append(append([]any{}, args...), args...)...).Scan(&t.Hosts)
+	err = s.DB.QueryRow(s.hostCounts(q, where), args...).Scan(&t.Hosts, &t.Peers)
 	return t, err
 }
 
+// hostCounts counts the internal addresses on either side, each once (a
+// host is often a client in one conversation and a server in another),
+// and the external ones. They are counted by grouping them, which spills
+// to disk where count(DISTINCT) over a week of addresses ran out of memory.
+func (s *Store) hostCounts(q Query, where string) string {
+	return fmt.Sprintf(`WITH x AS NOT MATERIALIZED (SELECT client, server, cli_int, srv_int FROM %s WHERE %s)
+		SELECT (SELECT count(*) FROM (SELECT client FROM x WHERE cli_int UNION SELECT server FROM x WHERE srv_int)),
+			(SELECT count(*) FROM (SELECT CASE WHEN NOT srv_int THEN server ELSE client END FROM x WHERE NOT srv_int OR NOT cli_int GROUP BY 1))`,
+		s.Source(q.From, q.To), where)
+}
+
 // Record is one detail row for the flow records view.
 type Record struct {
 	TS       int64   `json:"ts"`
@@ -909,7 +960,10 @@ func (s *Store) Sankey(q Query, mode string, n0, n2 int) ([]SankeyLink, error) {
 	if !ok {
 		return nil, fmt.Errorf("unknown flow paths mode %q", mode)
 	}
-	sqlq := fmt.Sprintf(`WITH b AS (SELECT %s AS c0, %s AS c1, %s AS c2, sum(wire) AS w FROM %s WHERE %s GROUP BY 1,2,3),
+	// b is the rows, read again by each step rather than summed by all
+	// three columns first: by client and server, a week is a group for
+	// nearly every row, more than the memory there is
+	sqlq := fmt.Sprintf(`WITH b AS NOT MATERIALIZED (SELECT %s AS c0, %s AS c1, %s AS c2, wire AS w FROM %s WHERE %s),
 			t0 AS (SELECT c0 FROM b GROUP BY c0 ORDER BY sum(w) DESC LIMIT %d),
 			t2 AS (SELECT c2 FROM b GROUP BY c2 ORDER BY sum(w) DESC LIMIT %d)
 		SELECT CASE WHEN c0 IN (SELECT c0 FROM t0) THEN c0 ELSE '__other__' END,
diff --git a/internal/store/rings.go b/internal/store/rings.go
index bfb7be5..d24b1ac 100644
--- a/internal/store/rings.go
+++ b/internal/store/rings.go
@@ -3,6 +3,7 @@ package store
 import (
 	"fmt"
 	"sort"
+	"strings"
 )
 
 // Ring is one segment of a two-ring chart: an inner value, or an outer value
@@ -20,12 +21,22 @@ type RingSet struct {
 	Total float64 `json:"total"`
 }
 
-var ringDims = map[string]string{
-	"server":  "server",
-	"client":  "client",
-	"service": portProtoExpr + " || chr(9) || app",
+// ringDims are the dimensions of the rings, each as a number and a text
+// for the IPv4 rows and for the others, and how the two are shown. An
+// address is its number where it has one: reading the address text of
+// every row takes longer than everything else the query reads.
+var ringDims = map[string]struct{ num, text, num6, text6, show string }{
+	"server":  {"srv4", "NULL::VARCHAR", "0::UINTEGER", "server", addrText},
+	"client":  {"cli4", "NULL::VARCHAR", "0::UINTEGER", "client", addrText},
+	"service": {"svc_port::UINTEGER * 256 + proto", "app", "svc_port::UINTEGER * 256 + proto", "app", serviceText},
 }
 
+// shown from {n} (the number) and {s} (the text) of a ring value
+const (
+	addrText    = "CASE WHEN {n} <> 0 THEN concat({n} >> 24, '.', ({n} >> 16) & 255, '.', ({n} >> 8) & 255, '.', {n} & 255) ELSE {s} END"
+	serviceText = "concat({n} >> 8, '/', CASE {n} & 255 WHEN 6 THEN 'tcp' WHEN 17 THEN 'udp' ELSE CAST({n} & 255 AS VARCHAR) END, chr(9), {s})"
+)
+
 // Rings splits the traffic by an inner dimension (its top nInner values,
 // the rest as "__other__") and each inner value by an outer dimension (its
 // top nOuter values, the rest as "__other__").
@@ -47,13 +58,26 @@ func (s *Store) Rings(q Query, inner, outer string, nInner, nOuter int) (*RingSe
 	// pairs are added up only for the top inner values: adding up every
 	// pair first (each of a week's clients with each server it reached)
 	// is what ran out of memory. The database keeps each inner value's
-	// top outer values and their total; the rest is the difference.
-	sq := fmt.Sprintf(`WITH x AS NOT MATERIALIZED (SELECT %s AS i, %s AS o, wire AS w FROM %s WHERE %s AND %s),
-		ti AS (SELECT i FROM x GROUP BY i ORDER BY sum(w) DESC LIMIT %d),
-		p AS (SELECT CASE WHEN i IN (SELECT i FROM ti) THEN i ELSE '__other__' END AS pi,
-			CASE WHEN i IN (SELECT i FROM ti) THEN o ELSE '__other__' END AS po, sum(w) AS w FROM x GROUP BY 1,2)
-		SELECT pi, po, CAST(w AS DOUBLE), CAST(sum(w) OVER (PARTITION BY pi) AS DOUBLE) FROM p
-		QUALIFY row_number() OVER (PARTITION BY pi ORDER BY w DESC) <= %d`, ie, oe, s.Source(q.From, q.To), where, cond, nInner, nOuter+1)
+	// top outer values and their total; the rest is the difference. A
+	// flow's two ends are the same family, so cli4 <> 0 tells the IPv4
+	// rows for either end.
+	src := s.Source(q.From, q.To)
+	show := func(d struct{ num, text, num6, text6, show string }, n, t string) string {
+		return strings.NewReplacer("{n}", n, "{s}", t).Replace(d.show)
+	}
+	sq := fmt.Sprintf(`WITH x AS NOT MATERIALIZED (
+			SELECT %[1]s AS in_, %[2]s AS is_, %[3]s AS on_, %[4]s AS os_, wire AS w FROM %[5]s WHERE (%[6]s) AND %[7]s AND cli4 <> 0
+			UNION ALL SELECT %[8]s, %[9]s, %[10]s, %[11]s, wire FROM %[5]s WHERE (%[6]s) AND %[7]s AND cli4 = 0),
+		ti AS (SELECT in_, is_, true AS hit FROM x GROUP BY ALL ORDER BY sum(w) DESC LIMIT %[12]d),
+		p AS (SELECT CASE WHEN ti.hit THEN x.in_ END AS pn, CASE WHEN ti.hit THEN x.is_ END AS ps,
+				CASE WHEN ti.hit THEN x.on_ END AS qn, CASE WHEN ti.hit THEN x.os_ END AS qs, sum(x.w) AS w
+			FROM x LEFT JOIN ti ON x.in_ = ti.in_ AND coalesce(x.is_, '') = coalesce(ti.is_, '') GROUP BY ALL)
+		SELECT CASE WHEN pn IS NULL THEN '__other__' ELSE %[13]s END, CASE WHEN qn IS NULL THEN '__other__' ELSE %[14]s END,
+			CAST(w AS DOUBLE), CAST(sum(w) OVER (PARTITION BY pn, ps) AS DOUBLE) FROM p
+		QUALIFY row_number() OVER (PARTITION BY pn, ps ORDER BY w DESC) <= %[15]d`,
+		ie.num, ie.text, oe.num, oe.text, src, where, cond, ie.num6, ie.text6, oe.num6, oe.text6, nInner,
+		show(ie, "pn", "ps"), show(oe, "qn", "qs"), nOuter+1)
+	args = append(append([]any{}, args...), args...)
 	rows, err := s.DB.Query(sq, args...)
 	if err != nil {
 		return nil, err
diff --git a/internal/store/series.go b/internal/store/series.go
index 8432b8c..48855d3 100644
--- a/internal/store/series.go
+++ b/internal/store/series.go
@@ -3,6 +3,7 @@ package store
 import (
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 )
 
@@ -70,7 +71,7 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 	}
 	// decide before the start is rounded down to a bucket: rounding makes a
 	// 6-hour range a little longer, which must not switch it to the summaries
-	rollup := s.rollupOK(q) && d.rollup != nil
+	rollup := d.rollup != nil && s.rollupOK(q, by == "client" || by == "server")
 	step := StepFor(q.span())
 	if rollup && d.hourly && step < time.Hour {
 		// hourly summaries in 5-minute buckets would show one spike an hour
@@ -106,12 +107,15 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 	// the values
 	// b is read again by each step rather than kept: kept, a week of
 	// buckets by client is more than the memory there is. The values are
-	// counted by grouping them, which spills to disk where count(DISTINCT)
-	// would not.
+	// counted approximately, in a few kilobytes, where counting them
+	// exactly holds every one of them.
 	sqlq := fmt.Sprintf(`WITH b AS NOT MATERIALIZED (SELECT * FROM (%s) AS x(t, k, w, p)),
 			top AS (SELECT k FROM b GROUP BY k ORDER BY sum(%s) DESC LIMIT %d)
-		SELECT t, CASE WHEN k IN (SELECT k FROM top) THEN k ELSE '__other__' END AS kk, CAST(sum(%s) AS DOUBLE), (SELECT count(*) FROM (SELECT k FROM b GROUP BY k))
+		SELECT t, CASE WHEN k IN (SELECT k FROM top) THEN k ELSE '__other__' END AS kk, CAST(sum(%s) AS DOUBLE), (SELECT approx_count_distinct(k) FROM b)
 		FROM b GROUP BY 1,2`, inner, m, k, m)
+	if !rollup && (by == "client" || by == "server") {
+		sqlq, args = s.addrSeries(q, by, bucket, m, k)
+	}
 	rows, err := s.DB.Query(sqlq, args...)
 	if err != nil {
 		return nil, err
@@ -181,3 +185,25 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 	}
 	return se, nil
 }
+
+// addrSeries is the query of SeriesBy for clients or servers from the flow
+// records. An IPv4 address is read and ranked as its number, and made into
+// text only for the top values: reading the address text of every row
+// takes longer than everything else the query reads. The other addresses
+// (IPv6) are read as text, in a part of their own.
+func (s *Store) addrSeries(q Query, by, bucket, m string, k int) (string, []any) {
+	num := map[string]string{"client": "cli4", "server": "srv4"}[by]
+	where, args, _ := q.where()
+	src := s.Source(q.From, q.To)
+	text := keyPartFor(by, 0).outer
+	text = strings.NewReplacer("g0n", "x.kn", "g0s", "x.ks").Replace(text)
+	sqlq := fmt.Sprintf(`WITH x AS NOT MATERIALIZED (
+			SELECT %[1]s AS t, %[2]s AS kn, NULL::VARCHAR AS ks, wire AS w, pkts AS p FROM %[3]s WHERE (%[4]s) AND %[2]s <> 0
+			UNION ALL SELECT %[1]s, 0::UINTEGER, %[5]s, wire, pkts FROM %[3]s WHERE (%[4]s) AND %[2]s = 0),
+			top AS (SELECT kn, ks, true AS hit FROM x GROUP BY kn, ks ORDER BY sum(%[6]s) DESC LIMIT %[7]d)
+		SELECT x.t, CASE WHEN top.hit THEN %[8]s ELSE '__other__' END, CAST(sum(x.%[6]s) AS DOUBLE),
+			(SELECT approx_count_distinct(hash(kn, ks)) FROM x)
+		FROM x LEFT JOIN top ON x.kn = top.kn AND x.ks IS NOT DISTINCT FROM top.ks GROUP BY 1,2`,
+		bucket, num, src, where, by, m, k, text)
+	return sqlq, append(append([]any{}, args...), args...)
+}
diff --git a/internal/store/views.go b/internal/store/views.go
index 5faeca7..da23026 100644
--- a/internal/store/views.go
+++ b/internal/store/views.go
@@ -12,13 +12,19 @@ import (
 
 // Summaries by device and by interface.
 //
-// The summaries in r_ts, r_host and r_dim are of all the traffic. Narrowed
+// The summaries in r_ts and r_dim are of all the traffic. Narrowed
 // to one device or one interface, which is how most of the pages are read,
 // a range of days used to be answered from the flow records themselves: a
 // week is tens of millions of rows, read two or three times over for one
 // chart. These tables hold the same summaries once per device and once per
 // interface, so those ranges are read from them instead.
 //
+// There is no such table for addresses: a client is seldom seen more than
+// a few times in an hour, so an hourly summary by address and interface
+// has more rows than the flow records it sums. Questions about addresses
+// are read from the flow records, in queries written to stay within the
+// memory there is (query.go, series.go, rings.go).
+//
 // iface is the exporter for a device ("10.0.0.1") or "exporter/ifindex"
 // for an interface. side says how the traffic used the interface:
 // sideIn came in through it, sideOut went out through it, sideBoth came in
@@ -41,11 +47,10 @@ const (
 
 var viewSchema = []string{
 	`CREATE TABLE IF NOT EXISTS r_ts_if (ts TIMESTAMP, iface VARCHAR, side UTINYINT, app VARCHAR, bytes UBIGINT, wire UBIGINT, pkts UBIGINT, flows UBIGINT)`,
-	`CREATE TABLE IF NOT EXISTS r_host_if (ts TIMESTAMP, iface VARCHAR, side UTINYINT, ip VARCHAR, role UTINYINT, internal BOOLEAN, bytes UBIGINT, wire UBIGINT, pkts UBIGINT, flows UBIGINT)`,
 	`CREATE TABLE IF NOT EXISTS r_dim_if (ts TIMESTAMP, iface VARCHAR, side UTINYINT, dim VARCHAR, val VARCHAR, bytes UBIGINT, wire UBIGINT, pkts UBIGINT, flows UBIGINT)`,
 }
 
-var viewTables = []string{"r_ts_if", "r_host_if", "r_dim_if"}
+var viewTables = []string{"r_ts_if", "r_dim_if"}
 
 // viewsDone is the meta key set once the summaries cover every flow file.
 const viewsDone = "views_v1"
@@ -75,7 +80,9 @@ var viewDims = [][2]string{
 func viewInserts(src string) []string {
 	// a counter that wrapped below zero (written by 1.6.0 and earlier) is
 	// left out rather than allowed to overflow the sum
-	ok := func(c string) string { return fmt.Sprintf("sum(CASE WHEN %s > %d THEN 0 ELSE %s END)::UBIGINT", c, wrapLimit, c) }
+	ok := func(c string) string {
+		return fmt.Sprintf("sum(CASE WHEN %s > %d THEN 0 ELSE %s END)::UBIGINT", c, wrapLimit, c)
+	}
 	sums := ok("bytes") + ", " + ok("wire") + ", " + ok("pkts") + ", " + ok("flows")
 	v := `(SELECT *, unnest(list_filter([
 			{'f': exporter, 's': 0::UTINYINT, 'ok': true},
@@ -91,8 +98,6 @@ func viewInserts(src string) []string {
 	}
 	return []string{
 		`INSERT INTO r_ts_if SELECT time_bucket(` + tsStep + `, ts), v.f, v.s, app, ` + sums + ` FROM ` + v + ` GROUP BY ALL`,
-		`INSERT INTO r_host_if SELECT date_trunc('hour', ts), v.f, v.s, h.ip, h.role, h.internal, ` + sums + ` FROM (
-			SELECT *, unnest([{'ip': client, 'role': 1::UTINYINT, 'internal': cli_int}, {'ip': server, 'role': 2::UTINYINT, 'internal': srv_int}]) AS h FROM ` + v + `) GROUP BY ALL`,
 		`INSERT INTO r_dim_if SELECT date_trunc('hour', ts), v.f, v.s, d.d, d.v, ` + sums + ` FROM (
 			SELECT *, unnest(list_filter([` + strings.Join(dims, ", ") + `], x -> x.ok)) AS d FROM ` + v + `) GROUP BY ALL`,
 	}
@@ -152,13 +157,14 @@ func (q Query) view() (view, bool) {
 	return view{key: key, sides: sides}, true
 }
 
-// rollupOK reports whether the summaries answer q.
-func (s *Store) rollupOK(q Query) bool {
+// rollupOK reports whether the summaries answer q. hosts: the answer needs
+// the summaries by address, which only all traffic has.
+func (s *Store) rollupOK(q Query, hosts bool) bool {
 	if q.span() <= 6*time.Hour {
 		return false
 	}
 	v, ok := q.view()
-	return ok && (v.key == "" || s.viewsReady.Load())
+	return ok && (v.key == "" || (!hosts && s.viewsReady.Load()))
 }
 
 // rollups replaces the placeholders {r_ts}, {r_host} and {r_dim} in sqlq
@@ -207,12 +213,6 @@ func (s *Store) viewSource(q Query, v view, table string) (string, []any, error)
 	case "r_ts":
 		sel = "SELECT ts, app, bytes, wire, pkts, flows " + sel
 		hot = []string{"SELECT ts, app, bytes, wire, pkts, flows FROM hot WHERE " + where}
-	case "r_host":
-		sel = "SELECT ts, ip, role, internal, bytes, wire, pkts, flows " + sel
-		hot = []string{
-			"SELECT ts, client, 1::UTINYINT, cli_int, bytes, wire, pkts, flows FROM hot WHERE " + where,
-			"SELECT ts, server, 2::UTINYINT, srv_int, bytes, wire, pkts, flows FROM hot WHERE " + where,
-		}
 	case "r_dim":
 		sel = "SELECT ts, dim, val, bytes, wire, pkts, flows " + sel
 		for _, d := range viewDims {
diff --git a/internal/store/views_test.go b/internal/store/views_test.go
index 6e30a04..6ef98ff 100644
--- a/internal/store/views_test.go
+++ b/internal/store/views_test.go
@@ -28,6 +28,9 @@ func TestViewsMatchRecords(t *testing.T) {
 				App: []string{"HTTPS", "HTTP"}[i%2], Exporter: "10.0.0.1", InIf: uint32(1 + i%3), OutIf: uint32(1 + m%3),
 				Dir: DirTransit, PeerCC: "US", PeerASN: uint32(64500 + i), PeerOrg: "Org",
 				Bytes: uint64(1000 * (i + 1)), Wire: uint64(1100 * (i + 1)), Pkts: uint64(i + 1), Flows: 1, Dup: i == 5}
+			if i == 4 { // an IPv6 client among the IPv4 ones
+				r.Client, r.Cli4 = "2001:db8::4", 0
+			}
 			r.Src, r.Dst = r.Client, r.Server
 			rows = append(rows, r)
 		}
@@ -52,7 +55,7 @@ func TestViewsMatchRecords(t *testing.T) {
 		{{Field: "iface", Value: "10.0.0.1/3"}, {Field: "iface_in", Value: "10.0.0.1/3"}},
 	} {
 		q := Query{From: from, To: to, Filters: f}
-		if !s.rollupOK(q) {
+		if !s.rollupOK(q, false) {
 			t.Fatalf("%v: not answered from the summaries", f)
 		}
 		answer := func() string {
@@ -91,8 +94,41 @@ func TestViewsMatchRecords(t *testing.T) {
 			}
 			return out + fmt.Sprintf("apps %v %.3f\n", se.Names, sum)
 		}
+		if f[0].Field == "exporter" {
+			tot, _ := s.Totals(q)
+			rs, err := s.Rings(q, "client", "server", 3, 2)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var inner, outer float64
+			keys := map[string]bool{}
+			for _, r := range rs.Inner {
+				inner += r.Wire
+				keys[r.Key] = true
+			}
+			for _, r := range rs.Outer {
+				outer += r.Wire
+			}
+			if rs.Total != float64(tot.Wire) || inner != rs.Total || outer != rs.Total || !keys["__other__"] || len(rs.Inner) != 4 {
+				t.Errorf("rings: total %v inner %v outer %v of %d, inner %+v", rs.Total, inner, outer, tot.Wire, rs.Inner)
+			}
+			sv, err := s.Rings(q, "service", "client", 8, 8)
+			if err != nil {
+				t.Fatal(err)
+			}
+			seen6 := false
+			for _, r := range sv.Outer {
+				seen6 = seen6 || r.Key == "2001:db8::4"
+			}
+			if sv.Inner[0].Key != "444/tcp\tHTTP" || !seen6 {
+				t.Errorf("service rings: %+v %+v", sv.Inner, sv.Outer)
+			}
+		}
 		// equal values may come in either order
 		sorted := func(x string) string { l := strings.Split(x, "\n"); sort.Strings(l); return strings.Join(l, "\n") }
+		if a := answer(); f[0].Field == "exporter" && (!strings.Contains(a, "client 2001:db8::4 ") || !strings.Contains(a, " 2001:db8::4 198.")) {
+			t.Errorf("%v: the IPv6 client is missing:\n%s", f, a)
+		}
 		got := sorted(answer())
 		s.viewsReady.Store(false) // the same from the flow records
 		want := sorted(answer())

From 769fede683ea6f1909b963b7241aeacd30575fe4 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 23 Oct 2026 23:53:13 +0000
Subject: [PATCH 3/3] Store changes

---
 internal/store/query.go  |  8 +++++++-
 internal/store/series.go |  8 +++++++-
 internal/store/store.go  | 20 ++++++++++++++++++--
 internal/store/views.go  |  4 ----
 4 files changed, 32 insertions(+), 8 deletions(-)

diff --git a/internal/store/query.go b/internal/store/query.go
index 39c21f1..6400b40 100644
--- a/internal/store/query.go
+++ b/internal/store/query.go
@@ -759,7 +759,13 @@ func (s *Store) SeriesByApp(q Query, k int) (*Series, error) {
 	for n := range tot {
 		names = append(names, n)
 	}
-	sort.Slice(names, func(i, j int) bool { return tot[names[i]] > tot[names[j]] })
+	// equal totals by name, so the legend keeps its order between redraws
+	sort.Slice(names, func(i, j int) bool {
+		if tot[names[i]] != tot[names[j]] {
+			return tot[names[i]] > tot[names[j]]
+		}
+		return names[i] < names[j]
+	})
 	if len(names) > k {
 		names = append(names[:k:k], "__other__")
 	}
diff --git a/internal/store/series.go b/internal/store/series.go
index 48855d3..92653d7 100644
--- a/internal/store/series.go
+++ b/internal/store/series.go
@@ -148,7 +148,13 @@ func (s *Store) SeriesBy(q Query, by, measure string, k int) (*Series, error) {
 			names = append(names, n)
 		}
 	}
-	sort.Slice(names, func(i, j int) bool { return tot[names[i]] > tot[names[j]] })
+	// equal totals by name, so the legend keeps its order between redraws
+	sort.Slice(names, func(i, j int) bool {
+		if tot[names[i]] != tot[names[j]] {
+			return tot[names[i]] > tot[names[j]]
+		}
+		return names[i] < names[j]
+	})
 	if _, ok := tot["__other__"]; ok {
 		names = append(names, "__other__")
 	}
diff --git a/internal/store/store.go b/internal/store/store.go
index ddf9012..1e071f4 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -207,8 +207,24 @@ func Open(opt Options) (*Store, error) {
 	if err := s.repairWrapped(); err != nil {
 		log.Printf("store: %v", err)
 	}
-	s.bg.Add(1)
-	go s.buildViews()
+	// the summaries are ready at once where there is nothing to summarise
+	// first (they were built before, or there are no flow files yet), and
+	// built in the background otherwise
+	s.segMu.RLock()
+	nsegs := len(s.segCache)
+	s.segMu.RUnlock()
+	switch {
+	case s.Meta(viewsDone) != "":
+		s.viewsReady.Store(true)
+	case nsegs == 0:
+		if err := s.SetMeta(viewsDone, time.Now().UTC().Format(time.RFC3339)); err != nil {
+			return nil, err
+		}
+		s.viewsReady.Store(true)
+	default:
+		s.bg.Add(1)
+		go s.buildViews()
+	}
 	log.Printf("store: %s, DuckDB memory limit %d MB, %d threads", opt.Dir, memMB, opt.Threads)
 	return s, nil
 }
diff --git a/internal/store/views.go b/internal/store/views.go
index da23026..a585502 100644
--- a/internal/store/views.go
+++ b/internal/store/views.go
@@ -247,10 +247,6 @@ func sealViews(tx *sql.Tx, boundary time.Time) error {
 // from the flow records as before.
 func (s *Store) buildViews() {
 	defer s.bg.Done()
-	if s.Meta(viewsDone) != "" {
-		s.viewsReady.Store(true)
-		return
-	}
 	// Start over: an earlier run may have stopped half way. The files
 	// sealed from here on are summarised by the seal itself, so the list
 	// is taken, and the tables emptied, with sealing held off.