Combine aggregate columns per function instead of summing every one

sumRows folded every numeric column by addition, which is only correct
for count and sum, so min/max returned a sum, avg an average of
averages, and a to_string extract collapsed into one empty-key row.

- Combine count and sum by adding, min and max by the extreme, on text
  columns as well as numeric ones
- Rewrite an avg extract into an upstream sum and count and divide the
  totals, answering under the avg key the client asked for
- Refuse an aggregate pdbmux cannot merge with 400 naming the clash
- Treat to_string and jsonb_typeof as row functions that group rather
  than fold, and key groups on every non-aggregate projected column
- Give the e2e fixture per-node resource line numbers and titles whose
  extremes differ per backend

README.md

@@ -24,11 +24,11 @@ not PQL) is forwarded verbatim.
 
 | Path | Behaviour |
 |---|---|
-| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract`/`count` query is **summed** instead. |
-| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. An `extract`/`count` query is **summed** instead. |
-| `GET /pdb/query/v4/facts/<name>[/<value>]` | Same fan-out and merge as `/facts`, and an `extract`/`count` query is **summed** the same way. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added — except on the fact's own path, which is **synthesised** from the `/facts` merge (see provenance). |
+| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract` query with a `function` column is **combined** instead. |
+| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. An `extract` query with a `function` column is **combined** instead. |
+| `GET /pdb/query/v4/facts/<name>[/<value>]` | Same fan-out and merge as `/facts`, and an `extract` query with a `function` column is **combined** the same way. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added — except on the fact's own path, which is **synthesised** from the `/facts` merge (see provenance). |
 | `GET /pdb/query/v4/fact-names` | Fan out to all and serve the **union** of the flat name arrays, deduped and re-sorted, re-paged across backends, plus the `pdbmux_source` name while injection is on. `order_by` is only valid on `name`. |
-| `GET /pdb/query/v4/resources` | An `extract`/`count` query is fanned out and **summed**; any other query is an unmerged pass-through. |
+| `GET /pdb/query/v4/resources` | An `extract` query with a `function` column is fanned out and **combined**; any other query is an unmerged pass-through. |
 | `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. |
 | `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. |
 | `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. |
@@ -87,36 +87,60 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
   events, which carry no id of their own, dedupe on the verbatim record (a node
   reporting to more than one backend stores identical records in each).
 - **Aggregates** — `extract`/`group_by` rows are counts, not records, so each
-  backend returns a partial answer that has to be **added**, not deduped. This
+  backend returns a partial answer that has to be **combined**, not deduped. This
   covers `/event-counts`, `/aggregate-event-counts`, and any `/reports`,
   `/nodes`, `/resources`, `/facts` or `/facts/<name>[/<value>]` query whose
   `extract` carries a `["function", ...]` column.
-  - The grouping key is the row's non-aggregate fields: for `/reports`,
-    `/nodes`, `/resources`, `/facts` and `/facts/<name>` they come from the
-    query — the plain `extract` fields plus any `group_by` clause — and for the
-    event-count endpoints from the row itself (`subject_type`/`subject`, or
-    `summarize_by`), whose remaining fields are all counts.
+  - The grouping key is the row's full set of non-aggregate columns: for
+    `/reports`, `/nodes`, `/resources`, `/facts` and `/facts/<name>` they come
+    from the query — the plain `extract` fields, the row-function columns and
+    any `group_by` clause — and for the event-count endpoints from the row
+    itself (`subject_type`/`subject`, or `summarize_by`), whose remaining fields
+    are all counts.
   - On `/nodes`, `/facts` and `/facts/<name>[/<value>]` this takes precedence
     over the `certname` merge: an aggregate row has no `certname`, so deduping
     would collapse every backend's rows into one backend's numbers. A query with
     no `function` column — including a plain `extract` projection — still merges
     by `certname`.
   - PuppetDB accepts `count`, `sum`, `avg`, `min`, `max`, `to_string` and
-    `jsonb_typeof` as `extract` functions. Only `count` and `sum` are additive,
-    so only those two merge correctly. `avg`, `min` and `max` are folded like any
-    other numeric column and their merged value is **wrong**; `to_string` and
-    `jsonb_typeof` are row functions whose non-numeric column is carried through
-    from the first backend to report the key. Query a single backend directly
-    for any of those five.
+    `jsonb_typeof` as `extract` functions, and names each response column after
+    the function itself. Each is combined by its own rule rather than by a
+    blanket sum:
+
+    | function | merged across backends by |
+    | --- | --- |
+    | `count` | adding |
+    | `sum` | adding |
+    | `min` | the smallest value any backend reported, on text columns as well as numeric ones |
+    | `max` | the largest value any backend reported, likewise |
+    | `avg` | rewriting the upstream query into `sum` + `count` of the same column and dividing the totals, so the answer is the estate's true weighted average, not an average of averages |
+    | `to_string` | nothing — it is a row function, so it groups like a plain projected column |
+    | `jsonb_typeof` | likewise |
+
+  - The `avg` rewrite is invisible to the client: the request still answers under
+    the `avg` key. It needs the `sum` and `count` response columns for itself, so
+    an `extract` that also projects a `sum` or a `count`, or that projects `avg`
+    more than once, is refused with **400** naming the clash rather than answered
+    with a wrong number. An `avg` over no rows stays `null`, as upstream. An
+    `order_by` on `avg` is applied to the merged rows here, not upstream.
+  - An `extract` function `pdbmux` has no combiner for is refused with **400**
+    rather than folded on a guess.
   - `/resources` has no cross-backend record identity to dedupe on, so only its
     aggregate queries merge; everything else stays an unmerged pass-through.
-  - Rows sharing a key collapse into one with their numeric columns summed. A key
-    only one backend reported is passed through byte-for-byte. An aggregate column
-    that is absent or non-numeric in a row is skipped, never zeroed, so the
-    backends that did report a number still count.
+  - Rows sharing a key collapse into one with each aggregate column combined by
+    its own rule. A key only one backend reported is passed through
+    byte-for-byte. An aggregate column that is absent or `null` in a row is
+    skipped, never zeroed or treated as an extreme, so the backends that did
+    report a value still count.
+  - `to_string` and `jsonb_typeof` compile to scalar expressions upstream, so
+    they return one row per record rather than an aggregate. They form part of
+    the grouping key alongside the plain `extract` fields and the `group_by`
+    clause — including a `group_by` that names the function itself. An `extract`
+    of nothing but row functions has no aggregate to fold, so every backend's
+    rows are kept as they came.
   - A `/reports` query with no `function` column is a projection of real reports,
     not an aggregate, and stays on the union path.
-  - `include_total=true` on a summed endpoint reports the **merged** row count,
+  - `include_total=true` on a combined endpoint reports the **merged** row count,
     not the sum of the backends' `X-Records`, since shared keys collapse.
 
 ### Provenance: the `pdbmux_source` fact
@@ -191,7 +215,7 @@ answered `[]` from the configured names alone, with no fan-out at all, and a
 value naming one filters a record set fetched under a key the value is not part
 of. The record set is a property of the estate rather than of the filter, so
 every value of it — and the unfiltered path — share one entry and one fetch.
-An `extract`/`count` query still takes the summing branch, and the gated query
+An `extract` query with a `function` column still takes the combining branch, and the gated query
 shapes above still answer `[]`, as does every form while injection is off — with
 the name kept out of `/fact-names`, since nothing then produces it.
 
@@ -253,7 +277,7 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
   — and the requested window is then cut from the merged, re-sorted set.
 - `include_total=true` on a union endpoint makes `pdbmux` sum each backend's
   `X-Records` header into one merged header. Deduped records are counted once per
-  backend, so the total is an upper bound. Summed endpoints report the merged row
+  backend, so the total is an upper bound. Combined endpoints report the merged row
   count instead.
 - A malformed `limit`, `offset` or `order_by` gets a `400` rather than being
   forwarded.
@@ -364,7 +388,7 @@ not answering.
 `/fact-names` record sets **in memory** so a busy Puppetboard does not re-fan-out
 the same query every few seconds — its facts overview and fact drilldown are two
 of the pages that hit hardest. Everything else runs uncached — including
-`extract`/`count` aggregates on those paths, and
+`extract` aggregates on those paths, and
 the `/pdb/meta/v1/*` and `/metrics/*` endpoints, which are served live on every
 request. The cache is an interface, and `/reports` gets its own (S3-backed)
 backend later without further handler changes.

aggregate.go

@@ -2,83 +2,216 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"sort"
 	"strconv"
 	"strings"
 )
 
-// aggregateSpec names the columns of an extract/group_by result set: keys
-// identify a row across backends, sums are the numeric aggregate columns added
-// together.
-type aggregateSpec struct {
-	keys []string
-	sums []string
+// combineOp is how one aggregate column's per-backend values become one value.
+type combineOp int
+
+const (
+	opAdd combineOp = iota
+	opMin
+	opMax
+)
+
+// aggColumn is one aggregate column of a result row: the JSON key openvoxdb
+// names it and the operation that folds the backends' values together.
+type aggColumn struct {
+	name string
+	op   combineOp
 }
 
-// parseAggregate reads a PuppetDB AST query and returns the aggregate shape of
-// its response, or nil when the query is not an aggregate — only a top-level
-// `extract` carrying at least one `["function", ...]` column produces summable
-// rows. Key columns are the plain (non-function) extract fields, unioned with an
-// explicit `group_by` clause when the query has one.
-func parseAggregate(query string) *aggregateSpec {
+// rowShape splits a result row into the columns that identify it and the
+// aggregate columns that are folded. finish, when set, rewrites the merged row
+// before it is encoded and forces re-encoding even for a single-backend row.
+type rowShape struct {
+	keys   []string
+	aggs   []aggColumn
+	finish func(map[string]json.RawMessage)
+}
+
+// openvoxdb's accepted extract functions are the keys of pdb-fns->pg-fns,
+// src/puppetlabs/puppetdb/query_eng/engine.clj:234-241, and each response column
+// is named after the function itself (compile-fnexpression, engine.clj:1488-1495).
+// count and sum add across backends; min and max take the extreme, on text
+// columns as well as numeric ones, which openvoxdb allows because the numeric
+// guard applies only to comparison clauses (engine.clj:2483-2490). avg is not
+// combinable from the shard rows at all, so it is rewritten upstream — see
+// rewriteAvg.
+var aggregateOps = map[string]combineOp{
+	"count": opAdd,
+	"sum":   opAdd,
+	"min":   opMin,
+	"max":   opMax,
+}
+
+// to_string and jsonb_typeof compile to scalar Postgres expressions, so they
+// yield one row per input row rather than an aggregate: they name an ordinary
+// projected column and identify a row instead of being folded into one.
+var rowFns = map[string]bool{
+	"to_string":    true,
+	"jsonb_typeof": true,
+}
+
+const avgColumn = "avg"
+
+// aggregateSpec is the merge shape of an extract query's response.
+type aggregateSpec struct {
+	keys  []string
+	aggs  []aggColumn
+	query string // rewritten upstream query, empty when the request's own is used
+	avg   bool   // the client asked for avg; sum and count are pdbmux's helpers
+}
+
+// parseAggregate reads a PuppetDB AST query and returns the merge shape of its
+// response, or nil when the query is not an extract carrying a `["function",...]`
+// column. A non-nil error means the query is an aggregate pdbmux cannot merge
+// and must be refused rather than answered with a wrong number.
+//
+// Key columns are the plain extract fields, the row-function columns and any
+// `group_by` clause; only genuine aggregates are folded.
+func parseAggregate(query string) (*aggregateSpec, error) {
 	if strings.TrimSpace(query) == "" {
-		return nil
+		return nil, nil
 	}
 	var ast []json.RawMessage
 	if json.Unmarshal([]byte(query), &ast) != nil || len(ast) < 2 {
-		return nil
+		return nil, nil
 	}
 	var op string
 	if json.Unmarshal(ast[0], &op) != nil || op != "extract" {
-		return nil
+		return nil, nil
 	}
 	var cols []json.RawMessage
 	if json.Unmarshal(ast[1], &cols) != nil {
-		return nil
+		return nil, nil
 	}
 
 	spec := &aggregateSpec{}
+	var avgArgs []json.RawMessage
+	sawFunction := false
 	for _, col := range cols {
 		var name string
 		if json.Unmarshal(col, &name) == nil {
 			spec.keys = appendUnique(spec.keys, name)
 			continue
 		}
-		if fn, ok := functionName(col); ok {
-			spec.sums = appendUnique(spec.sums, fn)
+		fn, args, ok := functionColumn(col)
+		if !ok {
+			continue
+		}
+		sawFunction = true
+		switch {
+		case rowFns[fn]:
+			spec.keys = appendUnique(spec.keys, fn)
+		case fn == avgColumn:
+			if spec.avg {
+				return nil, fmt.Errorf("extract projects %q more than once, which openvoxdb answers with order-dependent columns", avgColumn)
+			}
+			if len(args) == 0 {
+				return nil, fmt.Errorf("extract function avg needs a column to average")
+			}
+			spec.avg, avgArgs = true, args
+		default:
+			combine, known := aggregateOps[fn]
+			if !known {
+				return nil, fmt.Errorf("extract function %q cannot be merged across backends", fn)
+			}
+			spec.aggs = appendUniqueAgg(spec.aggs, aggColumn{name: fn, op: combine})
 		}
 	}
-	if len(spec.sums) == 0 {
-		return nil
+	if !sawFunction {
+		return nil, nil
 	}
 	for _, node := range ast[2:] {
 		for _, f := range groupByFields(node) {
 			spec.keys = appendUnique(spec.keys, f)
 		}
 	}
-	return spec
+	if spec.avg {
+		if err := spec.rewriteAvg(ast, cols, avgArgs); err != nil {
+			return nil, err
+		}
+	}
+	return spec, nil
 }
 
-// functionName returns the response column an extract function produces, which
-// PuppetDB names after the function itself: ["function","count","certname"]
-// yields a "count" column.
-func functionName(col json.RawMessage) (string, bool) {
+// rewriteAvg replaces the client's avg column with the sum and count of the same
+// expression, so the true weighted average can be computed from the shards:
+// Postgres avg(x) is sum(x)/count(x), and both of those do combine.
+func (a *aggregateSpec) rewriteAvg(ast, cols []json.RawMessage, args []json.RawMessage) error {
+	for _, taken := range []string{"sum", "count"} {
+		// openvoxdb names a column after its function, so a second one of the same
+		// name comes back as an order-dependent "<name>_2" rather than its own key.
+		if contains(a.keys, taken) || hasAgg(a.aggs, taken) {
+			return fmt.Errorf("avg cannot be merged alongside a %q column: pdbmux rewrites avg into an upstream sum and count, which openvoxdb would return under the same response key", taken)
+		}
+	}
+	sumCol, err := functionNode("sum", args)
+	if err != nil {
+		return err
+	}
+	countCol, err := functionNode("count", args)
+	if err != nil {
+		return err
+	}
+
+	out := make([]json.RawMessage, 0, len(cols)+1)
+	for _, col := range cols {
+		if fn, _, ok := functionColumn(col); ok && fn == avgColumn {
+			out = append(out, sumCol)
+			continue
+		}
+		out = append(out, col)
+	}
+	out = append(out, countCol)
+
+	rewritten := append([]json.RawMessage(nil), ast...)
+	encoded, err := json.Marshal(out)
+	if err != nil {
+		return err
+	}
+	rewritten[1] = encoded
+	query, err := json.Marshal(rewritten)
+	if err != nil {
+		return err
+	}
+	a.query = string(query)
+	a.aggs = append(a.aggs, aggColumn{name: "sum", op: opAdd}, aggColumn{name: "count", op: opAdd})
+	return nil
+}
+
+func functionNode(fn string, args []json.RawMessage) (json.RawMessage, error) {
+	parts := make([]json.RawMessage, 0, len(args)+2)
+	parts = append(parts, json.RawMessage(`"function"`), json.RawMessage(`"`+fn+`"`))
+	parts = append(parts, args...)
+	return json.Marshal(parts)
+}
+
+// functionColumn returns the response column an extract function produces, which
+// openvoxdb names after the function itself — ["function","count","certname"]
+// yields a "count" column — along with the function's arguments.
+func functionColumn(col json.RawMessage) (string, []json.RawMessage, bool) {
 	var parts []json.RawMessage
 	if json.Unmarshal(col, &parts) != nil || len(parts) < 2 {
-		return "", false
+		return "", nil, false
 	}
 	var head, name string
 	if json.Unmarshal(parts[0], &head) != nil || head != "function" {
-		return "", false
+		return "", nil, false
 	}
 	if json.Unmarshal(parts[1], &name) != nil || name == "" {
-		return "", false
+		return "", nil, false
 	}
-	return name, true
+	return name, parts[2:], true
 }
 
-// groupByFields returns the field names of a ["group_by", ...] AST node, or nil
-// for any other node.
+// groupByFields returns the response columns a ["group_by", ...] AST node names,
+// or nil for any other node. An entry may be a plain field or a function node,
+// which groups on that function's own column.
 func groupByFields(node json.RawMessage) []string {
 	var parts []json.RawMessage
 	if json.Unmarshal(node, &parts) != nil || len(parts) < 2 {
@@ -93,6 +226,10 @@ func groupByFields(node json.RawMessage) []string {
 		var name string
 		if json.Unmarshal(p, &name) == nil {
 			out = append(out, name)
+			continue
+		}
+		if fn, _, ok := functionColumn(p); ok {
+			out = append(out, fn)
 		}
 	}
 	return out
@@ -105,19 +242,53 @@ func appendUnique(s []string, v string) []string {
 	return append(s, v)
 }
 
-// columns reports which fields of a row form its grouping key and which are
-// summed. A spec is fixed by the query, so the row is ignored.
-func (a *aggregateSpec) columns(map[string]json.RawMessage) ([]string, []string) {
-	return a.keys, a.sums
+func appendUniqueAgg(s []aggColumn, v aggColumn) []aggColumn {
+	if hasAgg(s, v.name) {
+		return s
+	}
+	return append(s, v)
 }
 
-// inferredColumns derives an event-counts row's shape from the row itself: the
+func hasAgg(s []aggColumn, name string) bool {
+	for _, x := range s {
+		if x.name == name {
+			return true
+		}
+	}
+	return false
+}
+
+// shape reports a row's merge shape. A spec is fixed by the query, so the row is
+// ignored.
+func (a *aggregateSpec) shape(map[string]json.RawMessage) rowShape {
+	sh := rowShape{keys: a.keys, aggs: a.aggs}
+	if a.avg {
+		sh.finish = finishAvg
+	}
+	return sh
+}
+
+// finishAvg turns the summed helper columns into the avg key the client asked
+// for. An empty count is openvoxdb's own answer for an average over no rows.
+func finishAvg(row map[string]json.RawMessage) {
+	sum, sumOK := numberOf(row["sum"])
+	count, countOK := numberOf(row["count"])
+	delete(row, "sum")
+	delete(row, "count")
+	if !sumOK || !countOK || count == 0 {
+		row[avgColumn] = json.RawMessage("null")
+		return
+	}
+	row[avgColumn] = json.RawMessage(strconv.FormatFloat(sum/count, 'f', -1, 64))
+}
+
+// inferredShape derives an event-counts row's shape from the row itself: the
 // counts to add (successes, failures, noops, skips, total) are its numeric
-// fields, plus any null one — PuppetDB nulls an aggregate column when a backend
+// fields, plus any null one — openvoxdb nulls an aggregate column when a backend
 // matched nothing — and everything else, subject_type/subject/summarize_by,
 // identifies the row. Those endpoints have a fixed response shape with no
 // numeric key field, so nothing summable is mistaken for identity.
-func inferredColumns(row map[string]json.RawMessage) ([]string, []string) {
+func inferredShape(row map[string]json.RawMessage) rowShape {
 	var keys, sums []string
 	for name, val := range row {
 		if isJSONNumber(val) || isJSONNull(val) {
@@ -128,7 +299,11 @@ func inferredColumns(row map[string]json.RawMessage) ([]string, []string) {
 	}
 	sort.Strings(keys)
 	sort.Strings(sums)
-	return keys, sums
+	aggs := make([]aggColumn, 0, len(sums))
+	for _, s := range sums {
+		aggs = append(aggs, aggColumn{name: s, op: opAdd})
+	}
+	return rowShape{keys: keys, aggs: aggs}
 }
 
 // isJSONNumber reports whether a raw JSON value is a number.
@@ -144,31 +319,33 @@ func isJSONNull(raw json.RawMessage) bool {
 	return strings.TrimSpace(string(raw)) == "null"
 }
 
-// sumGroup accumulates the rows sharing one grouping key.
-type sumGroup struct {
-	raw    json.RawMessage            // first contributing row, verbatim
-	row    map[string]json.RawMessage // its decoded fields
-	totals map[string]float64         // running sum per aggregate column
-	merged bool                       // a second row was folded in
+// mergeGroup accumulates the rows sharing one grouping key.
+type mergeGroup struct {
+	raw      json.RawMessage            // first contributing row, verbatim
+	row      map[string]json.RawMessage // its decoded fields
+	totals   map[string]float64         // running sum per additive column
+	extremes map[string]json.RawMessage // running min/max per extreme column
+	finish   func(map[string]json.RawMessage)
+	merged   bool // a later row changed the group
 }
 
-// sumRows folds each backend's aggregate rows into one row per grouping key,
-// adding the numeric aggregate columns. columns decides, per row, which fields
-// are the key and which are summed.
+// combineRows folds each backend's aggregate rows into one row per grouping key,
+// combining each aggregate column by its own operation. shape decides, per row,
+// which fields are the key and how each aggregate column combines.
 //
-// A row that is not a JSON object passes through untouched, as does the sole row
-// of a key only one backend reported — those keep their upstream bytes. An
-// aggregate column that is absent or non-numeric in a later row is left at the
-// earlier backend's value rather than being coerced to zero. results come in
-// configured backend order and the output keeps first-seen order, a tie-break
-// only.
-func sumRows(results []backendResult, columns func(map[string]json.RawMessage) ([]string, []string)) []json.RawMessage {
+// A row that is not a JSON object passes through untouched, as does the sole
+// row of a key only one backend reported and every row of a shape with no
+// aggregate column to fold — those keep their upstream bytes. An aggregate
+// column that is absent or null in a later row leaves the group's value alone
+// rather than being coerced to zero. results come in configured backend order
+// and the output keeps first-seen order, a tie-break only.
+func combineRows(results []backendResult, shape func(map[string]json.RawMessage) rowShape) []json.RawMessage {
 	type slot struct {
 		raw   json.RawMessage // passthrough row, when group is nil
-		group *sumGroup
+		group *mergeGroup
 	}
 	var order []slot
-	groups := map[string]*sumGroup{}
+	groups := map[string]*mergeGroup{}
 
 	for _, res := range results {
 		for _, rec := range res.records {
@@ -177,32 +354,22 @@ func sumRows(results []backendResult, columns func(map[string]json.RawMessage) (
 				order = append(order, slot{raw: rec.Raw})
 				continue
 			}
-			keys, sums := columns(row)
-			k := groupKey(row, keys)
+			sh := shape(row)
+			if len(sh.aggs) == 0 {
+				// Only row functions were projected, so there is nothing to fold
+				// and each backend's rows stand on their own.
+				order = append(order, slot{raw: rec.Raw})
+				continue
+			}
+			k := groupKey(row, sh.keys)
 			g, ok := groups[k]
 			if !ok {
-				g = &sumGroup{raw: rec.Raw, row: row, totals: map[string]float64{}}
-				for _, s := range sums {
-					if n, ok := numberOf(row[s]); ok {
-						g.totals[s] = n
-					}
-				}
+				g = newMergeGroup(rec.Raw, row, sh)
 				groups[k] = g
 				order = append(order, slot{group: g})
 				continue
 			}
-			for _, s := range sums {
-				n, ok := numberOf(row[s])
-				if !ok {
-					continue
-				}
-				if _, seen := g.totals[s]; !seen {
-					// First numeric value for a column the earlier row lacked.
-					g.totals[s] = 0
-				}
-				g.totals[s] += n
-				g.merged = true
-			}
+			g.fold(row, sh.aggs)
 		}
 	}
 
@@ -217,10 +384,75 @@ func sumRows(results []backendResult, columns func(map[string]json.RawMessage) (
 	return out
 }
 
+func newMergeGroup(raw json.RawMessage, row map[string]json.RawMessage, sh rowShape) *mergeGroup {
+	g := &mergeGroup{
+		raw:      raw,
+		row:      row,
+		totals:   map[string]float64{},
+		extremes: map[string]json.RawMessage{},
+		finish:   sh.finish,
+	}
+	for _, c := range sh.aggs {
+		if c.op == opAdd {
+			if n, ok := numberOf(row[c.name]); ok {
+				g.totals[c.name] = n
+			}
+			continue
+		}
+		if v, ok := row[c.name]; ok && !isJSONNull(v) {
+			g.extremes[c.name] = v
+		}
+	}
+	return g
+}
+
+func (g *mergeGroup) fold(row map[string]json.RawMessage, aggs []aggColumn) {
+	for _, c := range aggs {
+		if c.op == opAdd {
+			n, ok := numberOf(row[c.name])
+			if !ok {
+				continue
+			}
+			if _, seen := g.totals[c.name]; !seen {
+				// First numeric value for a column the earlier row lacked.
+				g.totals[c.name] = 0
+			}
+			g.totals[c.name] += n
+			g.merged = true
+			continue
+		}
+		v, ok := row[c.name]
+		if !ok || isJSONNull(v) {
+			continue
+		}
+		cur, seen := g.extremes[c.name]
+		if seen && !extremeWins(v, cur, c.op) {
+			continue
+		}
+		g.extremes[c.name] = v
+		g.merged = true
+	}
+}
+
+// extremeWins reports whether candidate replaces the running min or max. Values
+// are compared decoded, so min/max works on the text columns openvoxdb allows
+// them on as well as on numbers.
+func extremeWins(candidate, current json.RawMessage, op combineOp) bool {
+	var a, b any
+	if json.Unmarshal(candidate, &a) != nil || json.Unmarshal(current, &b) != nil {
+		return false
+	}
+	c := compareValues(a, b)
+	if op == opMin {
+		return c < 0
+	}
+	return c > 0
+}
+
 // encode renders a group back to JSON, reusing the first row's bytes when
-// nothing was added to it.
-func (g *sumGroup) encode() json.RawMessage {
-	if !g.merged {
+// nothing changed it and no finish step has to rewrite it.
+func (g *mergeGroup) encode() json.RawMessage {
+	if !g.merged && g.finish == nil {
 		return g.raw
 	}
 	row := make(map[string]json.RawMessage, len(g.row))
@@ -230,6 +462,12 @@ func (g *sumGroup) encode() json.RawMessage {
 	for col, total := range g.totals {
 		row[col] = json.RawMessage(strconv.FormatFloat(total, 'f', -1, 64))
 	}
+	for col, v := range g.extremes {
+		row[col] = v
+	}
+	if g.finish != nil {
+		g.finish(row)
+	}
 	raw, err := json.Marshal(row)
 	if err != nil {
 		return g.raw

aggregate_test.go

@@ -2,8 +2,10 @@ package main
 
 import (
 	"encoding/json"
+	"net/url"
 	"reflect"
 	"slices"
+	"strings"
 	"testing"
 )
 
@@ -29,29 +31,82 @@ func decodeRows(t *testing.T, raws []json.RawMessage) []map[string]any {
 	return out
 }
 
-func TestParseAggregate_ExtractWithGroupBy(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
-	if spec == nil {
-		t.Fatal("expected an aggregate spec")
+// mustAggregate parses a query that has to be a mergeable aggregate.
+func mustAggregate(t *testing.T, q string) *aggregateSpec {
+	t.Helper()
+	spec, err := parseAggregate(q)
+	if err != nil {
+		t.Fatalf("parseAggregate(%s): %v", q, err)
 	}
+	if spec == nil {
+		t.Fatalf("parseAggregate(%s) = nil, want an aggregate spec", q)
+	}
+	return spec
+}
+
+func aggNames(aggs []aggColumn) []string {
+	out := make([]string, 0, len(aggs))
+	for _, a := range aggs {
+		out = append(out, a.name)
+	}
+	return out
+}
+
+func TestParseAggregate_ExtractWithGroupBy(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
 	if !slices.Equal(spec.keys, []string{"status"}) {
 		t.Errorf("keys = %v, want [status]", spec.keys)
 	}
-	if !slices.Equal(spec.sums, []string{"count"}) {
-		t.Errorf("sums = %v, want [count]", spec.sums)
+	if !slices.Equal(aggNames(spec.aggs), []string{"count"}) {
+		t.Errorf("aggs = %v, want [count]", aggNames(spec.aggs))
+	}
+	if spec.aggs[0].op != opAdd {
+		t.Errorf("count op = %v, want opAdd", spec.aggs[0].op)
 	}
 }
 
 func TestParseAggregate_GroupByAddsUnextractedField(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count","certname"]],["~","certname",".*"],["group_by","status"]]`)
-	if spec == nil {
-		t.Fatal("expected an aggregate spec")
-	}
+	spec := mustAggregate(t, `["extract",[["function","count","certname"]],["~","certname",".*"],["group_by","status"]]`)
 	if !slices.Equal(spec.keys, []string{"status"}) {
 		t.Errorf("keys = %v, want [status] from the group_by clause", spec.keys)
 	}
-	if !slices.Equal(spec.sums, []string{"count"}) {
-		t.Errorf("sums = %v, want [count]", spec.sums)
+	if !slices.Equal(aggNames(spec.aggs), []string{"count"}) {
+		t.Errorf("aggs = %v, want [count]", aggNames(spec.aggs))
+	}
+}
+
+func TestParseAggregate_OpPerFunction(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","min","line"],["function","max","line"],["function","sum","line"],["function","count"]]]`)
+	want := []aggColumn{
+		{name: "min", op: opMin},
+		{name: "max", op: opMax},
+		{name: "sum", op: opAdd},
+		{name: "count", op: opAdd},
+	}
+	if !reflect.DeepEqual(spec.aggs, want) {
+		t.Errorf("aggs = %v, want %v", spec.aggs, want)
+	}
+	if len(spec.keys) != 0 {
+		t.Errorf("keys = %v, want none", spec.keys)
+	}
+}
+
+// to_string and jsonb_typeof are row functions, so they identify a row rather
+// than being folded into it.
+func TestParseAggregate_RowFunctionsAreKeys(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","jsonb_typeof","value"],["function","count"]]]`)
+	if !slices.Equal(spec.keys, []string{"to_string", "jsonb_typeof"}) {
+		t.Errorf("keys = %v, want [to_string jsonb_typeof]", spec.keys)
+	}
+	if !slices.Equal(aggNames(spec.aggs), []string{"count"}) {
+		t.Errorf("aggs = %v, want [count]", aggNames(spec.aggs))
+	}
+}
+
+func TestParseAggregate_GroupByOnAFunctionColumn(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDD"]]]`)
+	if !slices.Equal(spec.keys, []string{"to_string"}) {
+		t.Errorf("keys = %v, want [to_string] once", spec.keys)
 	}
 }
 
@@ -64,18 +119,33 @@ func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) {
 		`["extract"]`,
 		`{"not":"an array"}`,
 	} {
-		if spec := parseAggregate(q); spec != nil {
+		spec, err := parseAggregate(q)
+		if err != nil {
+			t.Errorf("parseAggregate(%q): unexpected error %v", q, err)
+		}
+		if spec != nil {
 			t.Errorf("parseAggregate(%q) = %+v, want nil", q, spec)
 		}
 	}
 }
 
-func TestSumRows_SharedKeysAreAdded(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
-	merged := sumRows([]backendResult{
+// A function pdbmux has no combiner for is refused, never folded on a guess.
+func TestParseAggregate_UnknownFunctionIsRefused(t *testing.T) {
+	spec, err := parseAggregate(`["extract",[["function","stddev","line"]]]`)
+	if err == nil {
+		t.Fatalf("parseAggregate = %+v, want an error", spec)
+	}
+	if !strings.Contains(err.Error(), "stddev") {
+		t.Errorf("error %q does not name the function", err)
+	}
+}
+
+func TestCombineRows_SharedKeysAreAdded(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)},
 		{name: "b", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)},
-	}, spec.columns)
+	}, spec.shape)
 
 	want := []map[string]any{
 		{"count": float64(7), "status": "changed"},
@@ -86,12 +156,227 @@ func TestSumRows_SharedKeysAreAdded(t *testing.T) {
 	}
 }
 
-func TestSumRows_DisjointKeysAreKept(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
-	merged := sumRows([]backendResult{
+func TestCombineRows_SumIsAdded(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","sum","line"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"sum":30}`)},
+		{name: "b", records: rows(`{"sum":12}`)},
+	}, spec.shape)
+
+	want := []map[string]any{{"sum": float64(42)}}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// The bug this replaces: max folded by addition returned the sum of the
+// backends' maxima, a number no backend ever held.
+func TestCombineRows_MaxIsNotASum(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","max","line"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"max":20}`)},
+		{name: "b", records: rows(`{"max":50}`)},
+	}, spec.shape)
+
+	got := decodeRows(t, merged)
+	want := []map[string]any{{"max": float64(50)}}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("merged = %v, want %v", got, want)
+	}
+	if got[0]["max"] == float64(70) {
+		t.Error("max was summed")
+	}
+}
+
+func TestCombineRows_MinIsTheSmallest(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","min","line"]]]`)
+	for _, tc := range []struct {
+		name string
+		a, b string
+		want any
+	}{
+		{"later backend wins", `{"min":20}`, `{"min":5}`, float64(5)},
+		{"earlier backend wins", `{"min":5}`, `{"min":20}`, float64(5)},
+		{"negative values", `{"min":-1}`, `{"min":-9}`, float64(-9)},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			merged := combineRows([]backendResult{
+				{name: "a", records: rows(tc.a)},
+				{name: "b", records: rows(tc.b)},
+			}, spec.shape)
+			got := decodeRows(t, merged)
+			if len(got) != 1 || got[0]["min"] != tc.want {
+				t.Errorf("merged = %v, want min %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// openvoxdb allows min/max on text columns, where the wrong answer used to be
+// whichever backend answered first.
+func TestCombineRows_MinMaxOnTextColumns(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","min","name"],["function","max","name"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"min":"kernel","max":"role"}`)},
+		{name: "b", records: rows(`{"min":"extra_b","max":"uptime"}`)},
+	}, spec.shape)
+
+	want := []map[string]any{{"min": "extra_b", "max": "uptime"}}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// A backend that matched nothing answers null, which is not an extreme.
+func TestCombineRows_MinMaxIgnoreNulls(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","max","line"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"max":null}`)},
+		{name: "b", records: rows(`{"max":7}`)},
+	}, spec.shape)
+
+	want := []map[string]any{{"max": float64(7)}}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+func TestCombineRows_MinMaxWithGroupBy(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","min","line"],["function","max","line"],"type"],["group_by","type"]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"min":10,"max":20,"type":"File"}`, `{"min":1,"max":2,"type":"Stage"}`)},
+		{name: "b", records: rows(`{"min":30,"max":50,"type":"File"}`)},
+	}, spec.shape)
+
+	want := []map[string]any{
+		{"min": float64(10), "max": float64(50), "type": "File"},
+		{"min": float64(1), "max": float64(2), "type": "Stage"},
+	}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// avg is rewritten upstream into the sum and count of the same expression.
+func TestParseAggregate_AvgIsRewrittenToSumAndCount(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","avg","line"]],["=","type","File"]]`)
+	if !spec.avg {
+		t.Fatal("spec does not record the avg rewrite")
+	}
+	const want = `["extract",[["function","sum","line"],["function","count","line"]],["=","type","File"]]`
+	if spec.query != want {
+		t.Errorf("upstream query = %s, want %s", spec.query, want)
+	}
+	if !slices.Equal(aggNames(spec.aggs), []string{"sum", "count"}) {
+		t.Errorf("aggs = %v, want the sum and count helpers", aggNames(spec.aggs))
+	}
+}
+
+func TestParseAggregate_AvgKeepsCompanionColumnsAndGroupBy(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","avg","line"],"type"],["group_by","type"]]`)
+	const want = `["extract",[["function","sum","line"],"type",["function","count","line"]],["group_by","type"]]`
+	if spec.query != want {
+		t.Errorf("upstream query = %s, want %s", spec.query, want)
+	}
+	if !slices.Equal(spec.keys, []string{"type"}) {
+		t.Errorf("keys = %v, want [type]", spec.keys)
+	}
+}
+
+func TestCombineRows_AvgIsWeightedByCount(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","avg","line"]]]`)
+	// Backend a averages 10 over 1 row, b averages 20 over 3: the true average
+	// is 70/4, not the 15 an average of averages gives.
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"sum":10,"count":1}`)},
+		{name: "b", records: rows(`{"sum":60,"count":3}`)},
+	}, spec.shape)
+
+	want := []map[string]any{{"avg": float64(17.5)}}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// A single backend still gets the avg key back, not the helper columns.
+func TestCombineRows_AvgFromOneBackend(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","avg","line"],"type"],["group_by","type"]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"sum":30,"count":4,"type":"File"}`)},
+		{name: "b", records: nil},
+	}, spec.shape)
+
+	want := []map[string]any{{"avg": float64(7.5), "type": "File"}}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+func TestCombineRows_AvgOverNoRowsIsNull(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","avg","line"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"sum":null,"count":0}`)},
+		{name: "b", records: rows(`{"sum":null,"count":0}`)},
+	}, spec.shape)
+
+	want := []map[string]any{{"avg": nil}}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+func TestCombineRows_AvgWithGroupBy(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","avg","line"],"type"],["group_by","type"]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"sum":10,"count":1,"type":"File"}`, `{"sum":8,"count":2,"type":"Stage"}`)},
+		{name: "b", records: rows(`{"sum":60,"count":3,"type":"File"}`)},
+	}, spec.shape)
+
+	want := []map[string]any{
+		{"avg": float64(17.5), "type": "File"},
+		{"avg": float64(4), "type": "Stage"},
+	}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// The helper columns take the response keys openvoxdb would give the client's
+// own sum or count, so the pair cannot be served together.
+func TestParseAggregate_AvgWithSumOrCountIsRefused(t *testing.T) {
+	for _, q := range []string{
+		`["extract",[["function","avg","line"],["function","sum","line"]]]`,
+		`["extract",[["function","avg","line"],["function","count"]]]`,
+	} {
+		spec, err := parseAggregate(q)
+		if err == nil {
+			t.Errorf("parseAggregate(%s) = %+v, want a refusal", q, spec)
+			continue
+		}
+		if !strings.Contains(err.Error(), "avg") {
+			t.Errorf("error %q does not name avg", err)
+		}
+	}
+}
+
+func TestParseAggregate_AvgWithoutAColumnIsRefused(t *testing.T) {
+	if spec, err := parseAggregate(`["extract",[["function","avg"]]]`); err == nil {
+		t.Errorf("parseAggregate = %+v, want a refusal", spec)
+	}
+}
+
+func TestParseAggregate_TwoAvgColumnsAreRefused(t *testing.T) {
+	if spec, err := parseAggregate(`["extract",[["function","avg","line"],["function","avg","value"]]]`); err == nil {
+		t.Errorf("parseAggregate = %+v, want a refusal", spec)
+	}
+}
+
+func TestCombineRows_DisjointKeysAreKept(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(`{"count":3,"status":"changed"}`)},
 		{name: "b", records: rows(`{"count":2,"status":"skipped"}`)},
-	}, spec.columns)
+	}, spec.shape)
 
 	want := []map[string]any{
 		{"count": float64(3), "status": "changed"},
@@ -102,25 +387,25 @@ func TestSumRows_DisjointKeysAreKept(t *testing.T) {
 	}
 }
 
-func TestSumRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
+func TestCombineRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
 	const raw = `{"count":3,"status":"changed","extra":{"kept":true}}`
-	merged := sumRows([]backendResult{
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(raw)},
 		{name: "b", records: nil},
-	}, spec.columns)
+	}, spec.shape)
 
 	if len(merged) != 1 || string(merged[0]) != raw {
 		t.Errorf("merged = %s, want the row verbatim %s", merged, raw)
 	}
 }
 
-func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
-	merged := sumRows([]backendResult{
+func TestCombineRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(`{"count":5,"status":"changed"}`)},
 		{name: "b", records: rows(`{"count":null,"status":"changed"}`)},
-	}, spec.columns)
+	}, spec.shape)
 
 	want := []map[string]any{{"count": float64(5), "status": "changed"}}
 	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
@@ -128,12 +413,12 @@ func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
 	}
 }
 
-func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
-	merged := sumRows([]backendResult{
+func TestCombineRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(`{"status":"changed"}`)},
 		{name: "b", records: rows(`{"count":6,"status":"changed"}`)},
-	}, spec.columns)
+	}, spec.shape)
 
 	want := []map[string]any{{"count": float64(6), "status": "changed"}}
 	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
@@ -141,29 +426,26 @@ func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) {
 	}
 }
 
-func TestSumRows_NonObjectRowsPassThrough(t *testing.T) {
-	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
-	merged := sumRows([]backendResult{
+func TestCombineRows_NonObjectRowsPassThrough(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(`"surprise"`)},
 		{name: "b", records: rows(`{"count":1,"status":"changed"}`)},
-	}, spec.columns)
+	}, spec.shape)
 
 	if len(merged) != 2 || string(merged[0]) != `"surprise"` {
 		t.Fatalf("merged = %s, want the non-object row kept as-is", merged)
 	}
 }
 
-func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) {
+func TestCombineRows_NoKeyColumnsCollapseToOneRow(t *testing.T) {
 	// ["extract",[["function","count"]],...] is a whole-estate count: one row
 	// per backend, and the merged answer is their sum.
-	spec := parseAggregate(`["extract",[["function","count"]],["=","certname","h1"]]`)
-	if spec == nil {
-		t.Fatal("expected an aggregate spec")
-	}
-	merged := sumRows([]backendResult{
+	spec := mustAggregate(t, `["extract",[["function","count"]],["=","certname","h1"]]`)
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(`{"count":10}`)},
 		{name: "b", records: rows(`{"count":32}`)},
-	}, spec.columns)
+	}, spec.shape)
 
 	want := []map[string]any{{"count": float64(42)}}
 	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
@@ -171,23 +453,96 @@ func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) {
 	}
 }
 
-func TestInferredColumns_SplitsCountsFromIdentity(t *testing.T) {
+// to_string compiles to a scalar expression, so openvoxdb answers one row per
+// input row; with nothing to fold, every backend's rows are kept rather than
+// collapsing into a single empty-key bucket.
+func TestCombineRows_ToStringAloneDoesNotCollapse(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"to_string":"01"}`, `{"to_string":"02"}`)},
+		{name: "b", records: rows(`{"to_string":"02"}`, `{"to_string":"03"}`)},
+	}, spec.shape)
+
+	want := []map[string]any{
+		{"to_string": "01"}, {"to_string": "02"},
+		{"to_string": "02"}, {"to_string": "03"},
+	}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want every row kept %v", got, want)
+	}
+}
+
+func TestCombineRows_ToStringGroupsItsCompanionCount(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDD"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"to_string":"01","count":2}`, `{"to_string":"02","count":3}`)},
+		{name: "b", records: rows(`{"to_string":"02","count":4}`)},
+	}, spec.shape)
+
+	want := []map[string]any{
+		{"to_string": "01", "count": float64(2)},
+		{"to_string": "02", "count": float64(7)},
+	}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+func TestCombineRows_JsonbTypeofGroups(t *testing.T) {
+	spec := mustAggregate(t, `["extract",[["function","jsonb_typeof","value"],["function","count"]],["group_by",["function","jsonb_typeof","value"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"jsonb_typeof":"string","count":5}`)},
+		{name: "b", records: rows(`{"jsonb_typeof":"string","count":6}`, `{"jsonb_typeof":"number","count":1}`)},
+	}, spec.shape)
+
+	want := []map[string]any{
+		{"jsonb_typeof": "string", "count": float64(11)},
+		{"jsonb_typeof": "number", "count": float64(1)},
+	}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// A plain projected column keeps rows apart even when the query has no group_by.
+func TestCombineRows_PlainColumnsAreGroupingKeys(t *testing.T) {
+	spec := mustAggregate(t, `["extract",["type","title",["function","count"]]]`)
+	merged := combineRows([]backendResult{
+		{name: "a", records: rows(`{"type":"File","title":"/tmp/a","count":1}`)},
+		{name: "b", records: rows(`{"type":"File","title":"/tmp/b","count":1}`, `{"type":"File","title":"/tmp/a","count":2}`)},
+	}, spec.shape)
+
+	want := []map[string]any{
+		{"type": "File", "title": "/tmp/a", "count": float64(3)},
+		{"type": "File", "title": "/tmp/b", "count": float64(1)},
+	}
+	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+func TestInferredShape_SplitsCountsFromIdentity(t *testing.T) {
 	var row map[string]json.RawMessage
 	if err := json.Unmarshal([]byte(`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"skips":null}`), &row); err != nil {
 		t.Fatal(err)
 	}
-	keys, sums := inferredColumns(row)
-	if !slices.Equal(keys, []string{"subject", "subject_type"}) {
-		t.Errorf("keys = %v, want [subject subject_type]", keys)
+	sh := inferredShape(row)
+	if !slices.Equal(sh.keys, []string{"subject", "subject_type"}) {
+		t.Errorf("keys = %v, want [subject subject_type]", sh.keys)
 	}
 	// A null count is an empty aggregate, not part of the row's identity.
-	if !slices.Equal(sums, []string{"failures", "skips", "successes"}) {
-		t.Errorf("sums = %v, want [failures skips successes]", sums)
+	if !slices.Equal(aggNames(sh.aggs), []string{"failures", "skips", "successes"}) {
+		t.Errorf("aggs = %v, want [failures skips successes]", aggNames(sh.aggs))
+	}
+	for _, a := range sh.aggs {
+		if a.op != opAdd {
+			t.Errorf("%s op = %v, want opAdd", a.name, a.op)
+		}
 	}
 }
 
-func TestSumRows_EventCountsPerSubject(t *testing.T) {
-	merged := sumRows([]backendResult{
+func TestCombineRows_EventCountsPerSubject(t *testing.T) {
+	merged := combineRows([]backendResult{
 		{name: "a", records: rows(
 			`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"noops":0,"skips":0}`,
 			`{"subject_type":"certname","subject":{"title":"h2"},"failures":0,"successes":5,"noops":0,"skips":0}`,
@@ -195,7 +550,7 @@ func TestSumRows_EventCountsPerSubject(t *testing.T) {
 		{name: "b", records: rows(
 			`{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`,
 		)},
-	}, inferredColumns)
+	}, inferredShape)
 
 	got := decodeRows(t, merged)
 	want := []map[string]any{
@@ -208,3 +563,22 @@ func TestSumRows_EventCountsPerSubject(t *testing.T) {
 		t.Errorf("merged = %v, want %v", got, want)
 	}
 }
+
+func TestDropOrderBy(t *testing.T) {
+	for _, tc := range []struct {
+		name, in, want string
+	}{
+		{"removes the named field", `[{"field":"avg","order":"desc"}]`, ``},
+		{"keeps the others", `[{"field":"avg"},{"field":"type"}]`, `[{"field":"type"}]`},
+		{"leaves an unrelated order_by alone", `[{"field":"type"}]`, `[{"field":"type"}]`},
+		{"drops an unparseable order_by", `not json`, ``},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			params := url.Values{"order_by": {tc.in}}
+			dropOrderBy(params, "avg")
+			if got := params.Get("order_by"); got != tc.want {
+				t.Errorf("order_by = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}

e2e_fixture_test.go

@@ -4,6 +4,7 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"time"
 )
 
@@ -48,28 +49,32 @@ type nodeFixture struct {
 	facts    map[string]any
 	// reportEnd is the report's end_time, and so the node's report_timestamp.
 	reportEnd string
+	// lines are the catalog File resources' line numbers, one resource each. They
+	// give /resources a numeric column whose per-backend minimum, maximum, sum and
+	// row count all differ, so no combiner can be mistaken for another.
+	lines []int
 }
 
 // Backend A: 2 nodes, 7 facts. Backend B: 3 nodes, 10 facts. The counts are
 // deliberately unequal so a summed aggregate cannot be mistaken for either
 // backend's own number.
 var fixtureA = []nodeFixture{
-	{certname: nodeAlpha, reportEnd: tsAlpha, facts: map[string]any{
+	{certname: nodeAlpha, reportEnd: tsAlpha, lines: []int{10, 12, 14}, facts: map[string]any{
 		"osfamily": "RedHat", "kernel": "Linux", "role": "web", "only_a": "yes",
 	}},
-	{certname: nodeShared, reportEnd: tsSharedOnA, facts: map[string]any{
+	{certname: nodeShared, reportEnd: tsSharedOnA, lines: []int{20}, facts: map[string]any{
 		"osfamily": "RedHat", "kernel": "Linux", "owner": backendAName,
 	}},
 }
 
 var fixtureB = []nodeFixture{
-	{certname: nodeBeta, reportEnd: tsBeta, facts: map[string]any{
+	{certname: nodeBeta, reportEnd: tsBeta, lines: []int{30}, facts: map[string]any{
 		"osfamily": "Debian", "kernel": "Linux", "role": "db", "only_b": "yes", "extra_b": "1",
 	}},
-	{certname: nodeGamma, reportEnd: tsGamma, facts: map[string]any{
+	{certname: nodeGamma, reportEnd: tsGamma, lines: []int{40}, facts: map[string]any{
 		"osfamily": "Debian", "kernel": "Linux",
 	}},
-	{certname: nodeShared, reportEnd: tsSharedOnB, facts: map[string]any{
+	{certname: nodeShared, reportEnd: tsSharedOnB, lines: []int{50}, facts: map[string]any{
 		"osfamily": "Debian", "kernel": "Linux", "owner": backendBName,
 	}},
 }
@@ -84,7 +89,7 @@ func loadFixtures(ctx context.Context, t fatalf, a, b *backend) {
 	}
 	// A deactivated node proves the merged view reflects each backend's own
 	// filtering rather than a raw union of everything ever stored.
-	loadNode(ctx, t, a, nodeFixture{certname: nodeGone, reportEnd: tsGone, facts: map[string]any{"osfamily": "RedHat"}})
+	loadNode(ctx, t, a, nodeFixture{certname: nodeGone, reportEnd: tsGone, lines: []int{100}, facts: map[string]any{"osfamily": "RedHat"}})
 	a.submit(ctx, t, cmdDeactivateNode, verDeactivateNode, nodeGone, tsDeactivation, map[string]any{
 		"certname":           nodeGone,
 		"producer_timestamp": tsDeactivation,
@@ -105,7 +110,26 @@ func loadNode(ctx context.Context, t fatalf, b *backend, n nodeFixture) {
 // a catalog_environment and populate /resources, which the aggregate assertions
 // and Puppetboard's index both read.
 func catalogPayload(n nodeFixture) map[string]any {
-	title := "/tmp/" + n.certname
+	resources := []any{
+		map[string]any{
+			"type": "Stage", "title": "main", "aliases": []string{}, "exported": false,
+			"file": nil, "line": nil, "tags": []string{"stage"}, "parameters": map[string]any{},
+		},
+	}
+	edges := []any{}
+	for i, line := range n.lines {
+		title := fileTitle(n.certname, i)
+		resources = append(resources, map[string]any{
+			"type": "File", "title": title, "aliases": []string{}, "exported": false,
+			"file": "/etc/puppetlabs/code/site.pp", "line": line, "tags": []string{"file"},
+			"parameters": map[string]any{"ensure": "present"},
+		})
+		edges = append(edges, map[string]any{
+			"source":       map[string]any{"type": "Stage", "title": "main"},
+			"target":       map[string]any{"type": "File", "title": title},
+			"relationship": "contains",
+		})
+	}
 	return map[string]any{
 		"certname":           n.certname,
 		"version":            "1",
@@ -115,27 +139,21 @@ func catalogPayload(n nodeFixture) map[string]any {
 		"code_id":            nil,
 		"producer_timestamp": n.reportEnd,
 		"producer":           "pdbmux-e2e",
-		"edges": []any{
-			map[string]any{
-				"source":       map[string]any{"type": "Stage", "title": "main"},
-				"target":       map[string]any{"type": "File", "title": title},
-				"relationship": "contains",
-			},
-		},
-		"resources": []any{
-			map[string]any{
-				"type": "Stage", "title": "main", "aliases": []string{}, "exported": false,
-				"file": nil, "line": nil, "tags": []string{"stage"}, "parameters": map[string]any{},
-			},
-			map[string]any{
-				"type": "File", "title": title, "aliases": []string{}, "exported": false,
-				"file": "/etc/puppetlabs/code/site.pp", "line": 1, "tags": []string{"file"},
-				"parameters": map[string]any{"ensure": "present"},
-			},
-		},
+		"edges":              edges,
+		"resources":          resources,
 	}
 }
 
+// The first File resource keeps the plain /tmp/<certname> title the report's
+// event names; the rest sort after every other fixture title, so one backend
+// holds the estate's largest resource title and the other its smallest.
+func fileTitle(certname string, i int) string {
+	if i == 0 {
+		return "/tmp/" + certname
+	}
+	return fmt.Sprintf("/tmp/zz-%s-%d", certname, i)
+}
+
 // factsPayload is the "replace facts" v5 wire format: certname, environment,
 // producer, producer_timestamp and the fact values.
 func factsPayload(n nodeFixture) map[string]any {

e2e_query_test.go

@@ -6,10 +6,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"net/http"
 	"net/url"
 	"reflect"
 	"sort"
+	"strings"
 	"testing"
 )
 
@@ -361,6 +363,212 @@ func TestResourcesAggregatesAreSummed(t *testing.T) {
 	}
 }
 
+// oneAggregateRow reads the single row a whole-estate aggregate returns.
+func oneAggregateRow(t *testing.T, rows []map[string]any) map[string]any {
+	t.Helper()
+	if len(rows) != 1 {
+		t.Fatalf("want exactly one aggregate row, got %d: %v", len(rows), rows)
+	}
+	return rows[0]
+}
+
+// aggregateNumber reads one numeric aggregate column, which openvoxdb nulls when
+// the backend matched nothing.
+func aggregateNumber(t *testing.T, row map[string]any, column string) float64 {
+	t.Helper()
+	n, ok := row[column].(float64)
+	if !ok {
+		t.Fatalf("aggregate row has no numeric %s: %v", column, row)
+	}
+	return n
+}
+
+// backendAggregate runs a whole-estate aggregate against one backend directly.
+func backendAggregate(ctx context.Context, t *testing.T, b *backend, path, q string) map[string]any {
+	t.Helper()
+	return oneAggregateRow(t, b.query(ctx, t, path, query(q)))
+}
+
+// min and max are extremes, not sums: the merged answer has to be the smallest
+// and largest the backends reported, both of which the fixture makes distinct.
+func TestResourcesMinMaxAreExtremes(t *testing.T) {
+	ctx := context.Background()
+	const q = `["extract",[["function","min","line"],["function","max","line"]]]`
+
+	a := backendAggregate(ctx, t, h.a, resourcesPath, q)
+	b := backendAggregate(ctx, t, h.b, resourcesPath, q)
+	minA, maxA := aggregateNumber(t, a, "min"), aggregateNumber(t, a, "max")
+	minB, maxB := aggregateNumber(t, b, "min"), aggregateNumber(t, b, "max")
+	if minA == minB || maxA == maxB {
+		t.Fatalf("the fixture gives both backends the same extremes (min %v/%v, max %v/%v)", minA, minB, maxA, maxB)
+	}
+
+	got := oneAggregateRow(t, get(t, resourcesPath, query(q)).rows(t))
+	if wantMin := math.Min(minA, minB); aggregateNumber(t, got, "min") != wantMin {
+		t.Errorf("/resources min = %v, want %v (%s=%v, %s=%v)", got["min"], wantMin, h.a.name, minA, h.b.name, minB)
+	}
+	wantMax := math.Max(maxA, maxB)
+	if gotMax := aggregateNumber(t, got, "max"); gotMax != wantMax {
+		t.Errorf("/resources max = %v, want %v (%s=%v, %s=%v)", gotMax, wantMax, h.a.name, maxA, h.b.name, maxB)
+	} else if gotMax == maxA+maxB {
+		t.Errorf("/resources max = %v, which is the sum of the backends' maxima", gotMax)
+	}
+}
+
+// openvoxdb allows min/max on text columns, where the merged answer used to be
+// whichever backend happened to answer first. The fixture puts the estate's
+// smallest resource title in one backend and its largest in the other, so a
+// first-wins merge gets one of the two wrong whichever way round it reads them.
+func TestResourcesMinMaxOnATextColumn(t *testing.T) {
+	ctx := context.Background()
+	const q = `["extract",[["function","min","title"],["function","max","title"]],["=","type","File"]]`
+
+	a := backendAggregate(ctx, t, h.a, resourcesPath, q)
+	b := backendAggregate(ctx, t, h.b, resourcesPath, q)
+	minA, minB := aggregateString(t, a, "min"), aggregateString(t, b, "min")
+	maxA, maxB := aggregateString(t, a, "max"), aggregateString(t, b, "max")
+	if minA == minB || maxA == maxB {
+		t.Fatalf("the fixture gives both backends the same extremes (min %q/%q, max %q/%q)", minA, minB, maxA, maxB)
+	}
+
+	got := oneAggregateRow(t, get(t, resourcesPath, query(q)).rows(t))
+	if want := min(minA, minB); got["min"] != want {
+		t.Errorf("/resources min(title) = %v, want %q (%s=%q, %s=%q)", got["min"], want, h.a.name, minA, h.b.name, minB)
+	}
+	if want := max(maxA, maxB); got["max"] != want {
+		t.Errorf("/resources max(title) = %v, want %q (%s=%q, %s=%q)", got["max"], want, h.a.name, maxA, h.b.name, maxB)
+	}
+}
+
+func aggregateString(t *testing.T, row map[string]any, column string) string {
+	t.Helper()
+	s, ok := row[column].(string)
+	if !ok {
+		t.Fatalf("aggregate row has no string %s: %v", column, row)
+	}
+	return s
+}
+
+// avg is decomposed into an upstream sum and count, so the merged answer is the
+// true weighted average of the estate rather than an average of averages.
+func TestResourcesAvgIsWeighted(t *testing.T) {
+	ctx := context.Background()
+	const parts = `["extract",[["function","sum","line"],["function","count","line"]]]`
+
+	a := backendAggregate(ctx, t, h.a, resourcesPath, parts)
+	b := backendAggregate(ctx, t, h.b, resourcesPath, parts)
+	sumA, countA := aggregateNumber(t, a, "sum"), aggregateNumber(t, a, "count")
+	sumB, countB := aggregateNumber(t, b, "sum"), aggregateNumber(t, b, "count")
+	if countA == countB {
+		t.Fatalf("the fixture gives both backends %v rows, so a weighted average is indistinguishable from a plain one", countA)
+	}
+	want := (sumA + sumB) / (countA + countB)
+
+	const q = `["extract",[["function","avg","line"]]]`
+	got := aggregateNumber(t, oneAggregateRow(t, get(t, resourcesPath, query(q)).rows(t)), "avg")
+	if math.Abs(got-want) > 1e-9 {
+		t.Fatalf("/resources avg(line) = %v, want %v ((%v+%v)/(%v+%v))", got, want, sumA, sumB, countA, countB)
+	}
+
+	avgA := aggregateNumber(t, backendAggregate(ctx, t, h.a, resourcesPath, q), "avg")
+	avgB := aggregateNumber(t, backendAggregate(ctx, t, h.b, resourcesPath, q), "avg")
+	if math.Abs(got-(avgA+avgB)) < 1e-9 {
+		t.Errorf("/resources avg(line) = %v, which is the sum of the backends' averages", got)
+	}
+	if math.Abs(got-(avgA+avgB)/2) < 1e-9 {
+		t.Errorf("/resources avg(line) = %v, which is an unweighted average of averages", got)
+	}
+}
+
+func TestResourcesAvgWithGroupBy(t *testing.T) {
+	ctx := context.Background()
+	const parts = `["extract",[["function","sum","line"],["function","count","line"],"type"],["group_by","type"]]`
+	const q = `["extract",[["function","avg","line"],"type"],["group_by","type"]]`
+
+	sums, counts := map[string]float64{}, map[string]float64{}
+	for _, b := range []*backend{h.a, h.b} {
+		for _, row := range b.query(ctx, t, resourcesPath, query(parts)) {
+			typ, ok := row["type"].(string)
+			if !ok {
+				t.Fatalf("grouped row has no type: %v", row)
+			}
+			if n, ok := row["sum"].(float64); ok {
+				sums[typ] += n
+				counts[typ] += aggregateNumber(t, row, "count")
+			}
+		}
+	}
+	if len(sums) == 0 {
+		t.Fatal("no grouped rows carried a sum, so the weighted average is untested")
+	}
+
+	for _, row := range get(t, resourcesPath, query(q)).rows(t) {
+		typ, ok := row["type"].(string)
+		if !ok {
+			t.Fatalf("grouped row has no type: %v", row)
+		}
+		if counts[typ] == 0 {
+			if row["avg"] != nil {
+				t.Errorf("avg(line) for %s = %v, want null over no rows", typ, row["avg"])
+			}
+			continue
+		}
+		want := sums[typ] / counts[typ]
+		if got := aggregateNumber(t, row, "avg"); math.Abs(got-want) > 1e-9 {
+			t.Errorf("avg(line) for %s = %v, want %v", typ, got, want)
+		}
+	}
+}
+
+// A query pdbmux cannot merge is refused outright rather than answered with a
+// plausible wrong number.
+func TestUnmergeableAggregateIsRefused(t *testing.T) {
+	resp := rawGet(t, resourcesPath, query(`["extract",[["function","avg","line"],["function","count"]]]`))
+	if resp.status != http.StatusBadRequest {
+		t.Fatalf("status %d, want 400: %s", resp.status, resp.body)
+	}
+	if !strings.Contains(string(resp.body), "avg") {
+		t.Errorf("refusal %q does not name the limitation", resp.body)
+	}
+}
+
+// to_string is a scalar expression, so it yields one row per record: alone it
+// must not collapse the estate into a single row, and with a companion count it
+// groups.
+func TestReportsToString(t *testing.T) {
+	ctx := context.Background()
+	const bare = `["extract",[["function","to_string","producer_timestamp","FMDAY"]]]`
+
+	wantRows := len(h.a.query(ctx, t, reportsPath, query(bare))) + len(h.b.query(ctx, t, reportsPath, query(bare)))
+	if wantRows < 2 {
+		t.Fatalf("the fixture yields %d to_string rows, so a collapse would be invisible", wantRows)
+	}
+	rows := get(t, reportsPath, query(bare)).rows(t)
+	if len(rows) != wantRows {
+		t.Errorf("/reports to_string returned %d rows, want %d (every backend's rows)", len(rows), wantRows)
+	}
+	for _, row := range rows {
+		if _, ok := row["to_string"].(string); !ok {
+			t.Fatalf("to_string row has no string column: %v", row)
+		}
+	}
+
+	const grouped = `["extract",[["function","to_string","producer_timestamp","FMDAY"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDAY"]]]`
+	want := map[string]float64{}
+	for _, b := range []*backend{h.a, h.b} {
+		for _, row := range b.query(ctx, t, reportsPath, query(grouped)) {
+			want[row["to_string"].(string)] += aggregateNumber(t, row, "count")
+		}
+	}
+	got := map[string]float64{}
+	for _, row := range get(t, reportsPath, query(grouped)).rows(t) {
+		got[row["to_string"].(string)] += aggregateNumber(t, row, "count")
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("grouped /reports to_string counts = %v, want %v", got, want)
+	}
+}
+
 // A fact count row carries no certname, so the per-certname fact merge would
 // keep one backend's rows and drop the other's; only adding the numbers is right.
 func TestFactsAggregatesAreSummed(t *testing.T) {

reports.go

@@ -150,6 +150,41 @@ func valueRank(v any) int {
 	}
 }
 
+// dropOrderBy removes one field from an upstream order_by, for a column the
+// rewritten query no longer projects. An unparseable or emptied order_by is
+// dropped entirely; pdbmux re-sorts the merged rows on the client's own order.
+func dropOrderBy(params url.Values, field string) {
+	raw := params.Get("order_by")
+	if strings.TrimSpace(raw) == "" {
+		return
+	}
+	var entries []map[string]any
+	if json.Unmarshal([]byte(raw), &entries) != nil {
+		params.Del("order_by")
+		return
+	}
+	kept := make([]map[string]any, 0, len(entries))
+	for _, e := range entries {
+		if f, ok := e["field"].(string); ok && f == field {
+			continue
+		}
+		kept = append(kept, e)
+	}
+	if len(kept) == len(entries) {
+		return
+	}
+	if len(kept) == 0 {
+		params.Del("order_by")
+		return
+	}
+	encoded, err := json.Marshal(kept)
+	if err != nil {
+		params.Del("order_by")
+		return
+	}
+	params.Set("order_by", string(encoded))
+}
+
 type paging struct {
 	limit     int // -1 when unset
 	offset    int

server.go

@@ -102,9 +102,9 @@ func (s *Server) StopProbes() { s.health.Stop() }
 func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) {
 	switch {
 	case path == factsPath, path == nodesPath, path == factNamesPath, isFactsSubPath(path):
-		// An aggregate row is a summed count, not the merged record set the
+		// An aggregate row is a combined count, not the merged record set the
 		// cache was built for, so it stays on the live path.
-		if parseAggregate(params.Get("query")) != nil {
+		if spec, err := parseAggregate(params.Get("query")); spec != nil || err != nil {
 			return noopCache{}, false
 		}
 		if s.factsCache != nil {
@@ -143,7 +143,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 	case eventsPath:
 		s.serveUnion(w, r, eventsPath, rawKey)
 	case eventCountsPath, aggregateEventCountsPath:
-		s.serveSummed(w, r, r.URL.Path, inferredColumns)
+		s.serveCombined(w, r, r.URL.Path, nil, inferredShape)
 	default:
 		if name, value, valued := factsSubPath(r.URL.Path); name != "" {
 			s.serveFactsByName(w, r, name, value, valued)
@@ -304,10 +304,14 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string,
 	})
 }
 
-// Aggregate rows carry no certname, so the per-certname fact merge would keep only one backend's; they take the summing path instead.
+// Aggregate rows carry no certname, so the per-certname fact merge would keep only one backend's; they take the combining path instead.
 func (s *Server) serveFacts(w http.ResponseWriter, r *http.Request) {
-	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
-		s.serveSummed(w, r, factsPath, spec.columns)
+	spec, ok := s.aggregateOf(w, r)
+	if !ok {
+		return
+	}
+	if spec != nil {
+		s.serveCombined(w, r, factsPath, spec, spec.shape)
 		return
 	}
 	s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r))
@@ -315,12 +319,16 @@ func (s *Server) serveFacts(w http.ResponseWriter, r *http.Request) {
 
 // Both path forms are the facts entity with a name (and value) constraint ANDed
 // on, so an aggregate over them carries no certname for the per-certname merge
-// to key on, and is summed instead — the same split /nodes makes. A plain query
+// to key on, and is combined instead — the same split /nodes makes. A plain query
 // on the source fact's own path is the one name no backend can answer for, so it
 // is synthesised instead of fanned out as-is.
 func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name, value string, valued bool) {
-	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
-		s.serveSummed(w, r, r.URL.Path, spec.columns)
+	spec, ok := s.aggregateOf(w, r)
+	if !ok {
+		return
+	}
+	if spec != nil {
+		s.serveCombined(w, r, r.URL.Path, spec, spec.shape)
 		return
 	}
 	if inject := s.newSourceInjector(r.URL.Query().Get("query"), true); inject.claims(name) && inject.injects() {
@@ -384,10 +392,14 @@ func (s *Server) hasBackend(name string) bool {
 	return false
 }
 
-// A count row carries no certname, so the certname-keyed merge would collapse every backend's count into one backend's; aggregates take the summing path instead.
+// A count row carries no certname, so the certname-keyed merge would collapse every backend's count into one backend's; aggregates take the combining path instead.
 func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) {
-	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
-		s.serveSummed(w, r, nodesPath, spec.columns)
+	spec, ok := s.aggregateOf(w, r)
+	if !ok {
+		return
+	}
+	if spec != nil {
+		s.serveCombined(w, r, nodesPath, spec, spec.shape)
 		return
 	}
 	s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(r))
@@ -395,37 +407,63 @@ func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) {
 
 // Only aggregates merge: a resource record has no cross-backend identity to dedupe on, so a plain query stays on the pass-through path.
 func (s *Server) serveResources(w http.ResponseWriter, r *http.Request) {
-	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
-		s.serveSummed(w, r, resourcesPath, spec.columns)
+	spec, ok := s.aggregateOf(w, r)
+	if !ok {
+		return
+	}
+	if spec != nil {
+		s.serveCombined(w, r, resourcesPath, spec, spec.shape)
 		return
 	}
 	s.proxyUnmerged(w, r)
 }
 
-// An `extract` query with a `function` column returns synthetic aggregate rows that carry no identity, so they are summed rather than unioned.
+// An `extract` query with a `function` column returns synthetic aggregate rows that carry no identity, so they are combined rather than unioned.
 func (s *Server) serveReports(w http.ResponseWriter, r *http.Request) {
-	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
-		s.serveSummed(w, r, reportsPath, spec.columns)
+	spec, ok := s.aggregateOf(w, r)
+	if !ok {
+		return
+	}
+	if spec != nil {
+		s.serveCombined(w, r, reportsPath, spec, spec.shape)
 		return
 	}
 	s.serveUnion(w, r, reportsPath, reportKey)
 }
 
+// aggregateOf reads the request's merge shape, answering 400 itself when the
+// query is an aggregate pdbmux cannot merge; ok=false means it already replied.
+func (s *Server) aggregateOf(w http.ResponseWriter, r *http.Request) (*aggregateSpec, bool) {
+	spec, err := parseAggregate(r.URL.Query().Get("query"))
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return nil, false
+	}
+	return spec, true
+}
+
 // Merged rows are fewer than the backends' combined records, so include_total reports the merged count rather than a sum of X-Records.
-func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string, columns func(map[string]json.RawMessage) ([]string, []string)) {
+func (s *Server) serveCombined(w http.ResponseWriter, r *http.Request, path string, spec *aggregateSpec, shape func(map[string]json.RawMessage) rowShape) {
 	in := r.URL.Query()
 	page, err := parsePaging(in)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
+	upstream := page.upstreamParams(in)
+	if spec != nil && spec.query != "" {
+		// The backends answer the rewritten query, so they no longer carry the
+		// column the client's order_by may name; the merged rows are sorted here.
+		upstream.Set("query", spec.query)
+		dropOrderBy(upstream, avgColumn)
+	}
 
 	s.serveCached(w, r, path, in, func(ctx context.Context) (cachedResponse, error) {
-		alive, err := s.aliveResults(ctx, path, page.upstreamParams(in))
+		alive, err := s.aliveResults(ctx, path, upstream)
 		if err != nil {
 			return cachedResponse{}, err
 		}
-		merged := sumRows(alive, columns)
+		merged := combineRows(alive, shape)
 		sortRecords(merged, page.order)
 		resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1}
 		s.countBackends(&resp, alive)

server_test.go

@@ -810,6 +810,114 @@ func TestHandler_ReportsAggregateSummed(t *testing.T) {
 	}
 }
 
+// The backends never see the client's avg: they are asked for the sum and count
+// it decomposes into, and the weighted average is computed from those.
+func TestHandler_AvgIsRewrittenUpstreamAndWeighted(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[resourcesPath] = `[{"sum":10,"count":1}]`
+	b := newFakeBackend(t, `[]`, `[]`)
+	b.bodies[resourcesPath] = `[{"sum":60,"count":3}]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{
+		"query": {`["extract",[["function","avg","line"]]]`},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	const wantUpstream = `["extract",[["function","sum","line"],["function","count","line"]]]`
+	for _, fb := range []*fakeBackend{a, b} {
+		if got := fb.gotQuery(resourcesPath); got != wantUpstream {
+			t.Errorf("backend query = %s, want %s", got, wantUpstream)
+		}
+	}
+	var got []map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	want := []map[string]any{{"avg": float64(17.5)}}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("body = %v, want %v (70/4, not the 15 an average of averages gives)", got, want)
+	}
+}
+
+// order_by names a column the rewritten query no longer projects, so it is
+// dropped upstream and applied to the merged rows here instead.
+func TestHandler_AvgOrderByIsDroppedUpstream(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[resourcesPath] = `[{"sum":10,"count":1,"type":"File"},{"sum":8,"count":2,"type":"Stage"}]`
+	b := newFakeBackend(t, `[]`, `[]`)
+	b.bodies[resourcesPath] = `[{"sum":60,"count":3,"type":"File"}]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{
+		"query":    {`["extract",[["function","avg","line"],"type"],["group_by","type"]]`},
+		"order_by": {`[{"field":"avg","order":"desc"}]`},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	if params, _ := a.params(resourcesPath); params.Get("order_by") != "" {
+		t.Errorf("backend got order_by %q, want it dropped", params.Get("order_by"))
+	}
+	var got []map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	want := []map[string]any{
+		{"avg": float64(17.5), "type": "File"},
+		{"avg": float64(4), "type": "Stage"},
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("body = %v, want %v sorted by avg descending", got, want)
+	}
+}
+
+func TestHandler_UnmergeableAggregateIsRefused(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	for _, path := range []string{nodesPath, factsPath, reportsPath, resourcesPath, factsPath + "/uptime"} {
+		rec := doGetParams(t, srv.Handler(), path, url.Values{
+			"query": {`["extract",[["function","avg","line"],["function","count"]]]`},
+		})
+		if rec.Code != http.StatusBadRequest {
+			t.Errorf("%s: status %d, want 400: %s", path, rec.Code, rec.Body.String())
+		}
+		if !strings.Contains(rec.Body.String(), "avg") {
+			t.Errorf("%s: body %q does not name the limitation", path, rec.Body.String())
+		}
+	}
+	if _, asked := a.params(nodesPath); asked {
+		t.Error("a refused query was still fanned out")
+	}
+}
+
+// max used to be folded by addition, returning a number no backend held.
+func TestHandler_MaxIsNotSummed(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[resourcesPath] = `[{"max":20,"min":10}]`
+	b := newFakeBackend(t, `[]`, `[]`)
+	b.bodies[resourcesPath] = `[{"max":50,"min":30}]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{
+		"query": {`["extract",[["function","max","line"],["function","min","line"]]]`},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	var got []map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	want := []map[string]any{{"max": float64(50), "min": float64(10)}}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("body = %v, want %v", got, want)
+	}
+}
+
 func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) {
 	a := newFakeBackend(t, `[]`, `[]`)
 	a.bodies[reportsPath] = `[{"count":4,"status":"changed"}]`