Refuse duplicate aggregate columns and page aggregates after the fold

A repeated extract function names one response column twice, which openvoxdb
aliases as <name>_2: unknown to the merge spec, it froze at the first backend's
value. A limit pushed upstream truncated each backend's groups before the
cross-backend fold, so a group could be partly counted or missed.

- Refuses any extract projecting one response column twice, naming the clash
- Fetches every group and applies limit/offset after the fold
- Documents the float64 avg divergence from Postgres numeric

README.md

@@ -117,12 +117,24 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
     | `to_string` | nothing — it is a row function, so it groups like a plain projected column |
     | `jsonb_typeof` | likewise |
 
+  - Because each column is named after its function, an `extract` that projects
+    the **same function twice** — any of them — names one response column twice.
+    openvoxdb aliases the repeat as `<name>_2` (then `_3`, and so on), which is
+    neither a grouping key nor an aggregate `pdbmux` knows to fold, so the first
+    backend's value would freeze into the merged row. Such a query is refused
+    with **400** naming the clashing column, as is one whose plain `extract`
+    field takes the name a projected function would use. Repeating a plain field
+    is not a clash: the copy holds the same value as the key it duplicates.
   - 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`
+    an `extract` that also projects a `sum` or a `count` is refused with **400**
-    more than once, is refused with **400** naming the clash rather than answered
+    naming the clash rather than answered with a wrong number. An `avg` over no
-    with a wrong number. An `avg` over no rows stays `null`, as upstream. An
+    rows stays `null`, as upstream. An `order_by` on `avg` is applied to the
-    `order_by` on `avg` is applied to the merged rows here, not upstream.
+    merged rows here, not upstream.
+  - `avg` is folded as `sum / count` in float64, while a single openvoxdb divides
+    in Postgres `numeric`, which is arbitrary-precision. Whole-number averages
+    round-trip exactly; a fractional one can differ from a single backend's
+    answer in the low-order digits, as can a `sum` beyond 2^53.
   - 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
@@ -137,7 +149,14 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
     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.
+    rows are kept as they came — and, having one row per record rather than per
+    group, they keep the upstream `limit` that bounds them.
+  - `limit` and `offset` are **not** forwarded for an `extract` that folds: a
+    backend's own first N groups are not the merged result's first N, and a group
+    truncated away on one backend would fold to a wrong value. Every group is
+    fetched and the window cut after the fold, which an aggregate's row count —
+    one per distinct group value — keeps affordable. `include_total` still
+    reports the merged group count.
   - 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 combined endpoint reports the **merged** row count,
@@ -275,6 +294,8 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
   the merged set's existing order). A record missing an ordered field sorts first.
 - Backends are asked for the first `offset + limit` records — never an `offset`
   — and the requested window is then cut from the merged, re-sorted set.
+- A folded `extract` aggregate is the exception: neither `limit` nor `offset` is
+  forwarded, since a group truncated on one backend cannot be folded correctly.
 - `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. Combined endpoints report the merged row

aggregate.go

@@ -93,9 +93,18 @@ func parseAggregate(query string) (*aggregateSpec, error) {
 	spec := &aggregateSpec{}
 	var avgArgs []json.RawMessage
 	sawFunction := false
+	projected := map[string]bool{}
+	fnColumns := map[string]bool{}
 	for _, col := range cols {
 		var name string
 		if json.Unmarshal(col, &name) == nil {
+			// A field repeating another field is harmless — the copy holds the same
+			// value as the key it duplicates — but one taking a function's name is
+			// the same clash the other way round.
+			if fnColumns[name] {
+				return nil, columnClash(name)
+			}
+			projected[name] = true
 			spec.keys = appendUnique(spec.keys, name)
 			continue
 		}
@@ -104,13 +113,14 @@ func parseAggregate(query string) (*aggregateSpec, error) {
 			continue
 		}
 		sawFunction = true
+		if projected[fn] {
+			return nil, columnClash(fn)
+		}
+		projected[fn], fnColumns[fn] = true, 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")
 			}
@@ -120,7 +130,7 @@ func parseAggregate(query string) (*aggregateSpec, error) {
 			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})
+			spec.aggs = append(spec.aggs, aggColumn{name: fn, op: combine})
 		}
 	}
 	if !sawFunction {
@@ -139,6 +149,14 @@ func parseAggregate(query string) (*aggregateSpec, error) {
 	return spec, nil
 }
 
+// columnClash refuses a projection naming one response column twice. openvoxdb
+// aliases every extract column after its function, so a repeat comes back as an
+// order-dependent "<name>_2" that is neither a grouping key nor a folded
+// aggregate, leaving the first backend's value frozen in the merged row.
+func columnClash(name string) error {
+	return fmt.Errorf("extract projects the column %q more than once: openvoxdb returns the repeat as %q, which pdbmux can neither key on nor fold", name, name+"_2")
+}
+
 // 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.
@@ -242,13 +260,6 @@ func appendUnique(s []string, v string) []string {
 	return append(s, v)
 }
 
-func appendUniqueAgg(s []aggColumn, v aggColumn) []aggColumn {
-	if hasAgg(s, v.name) {
-		return s
-	}
-	return append(s, v)
-}
-
 func hasAgg(s []aggColumn, name string) bool {
 	for _, x := range s {
 		if x.name == name {

aggregate_test.go

@@ -371,6 +371,44 @@ func TestParseAggregate_TwoAvgColumnsAreRefused(t *testing.T) {
 	}
 }
 
+// Every extract function names its column after itself, so a second one of the
+// same name clashes whatever the function is — not only for avg.
+func TestParseAggregate_RepeatedFunctionNameIsRefused(t *testing.T) {
+	for _, fn := range []string{"count", "sum", "min", "max", "to_string", "jsonb_typeof", "avg"} {
+		q := `["extract",[["function","` + fn + `","line"],["function","` + fn + `","type"]],["group_by","line","type"]]`
+		spec, err := parseAggregate(q)
+		if err == nil {
+			t.Errorf("parseAggregate(%s) = %+v, want a refusal", q, spec)
+			continue
+		}
+		if !strings.Contains(err.Error(), fn) {
+			t.Errorf("error %q does not name the clashing column %q", err, fn)
+		}
+	}
+}
+
+// The same clash the other way round: a plain field takes the response key a
+// later function column would name.
+func TestParseAggregate_FieldClashingWithAFunctionIsRefused(t *testing.T) {
+	const q = `["extract",["count",["function","count","certname"]],["group_by","count"]]`
+	spec, err := parseAggregate(q)
+	if err == nil {
+		t.Fatalf("parseAggregate(%s) = %+v, want a refusal", q, spec)
+	}
+	if !strings.Contains(err.Error(), "count") {
+		t.Errorf("error %q does not name the clashing column", err)
+	}
+}
+
+// A repeated plain field projects the same value twice, so it is no clash: the
+// duplicate carries nothing the grouping key has not already got.
+func TestParseAggregate_RepeatedPlainFieldIsKept(t *testing.T) {
+	spec := mustAggregate(t, `["extract",["type","type",["function","count","certname"]],["group_by","type"]]`)
+	if !slices.Equal(spec.keys, []string{"type"}) {
+		t.Errorf("keys = %v, want [type]", spec.keys)
+	}
+}
+
 func TestCombineRows_DisjointKeysAreKept(t *testing.T) {
 	spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
 	merged := combineRows([]backendResult{

e2e_query_test.go

@@ -532,6 +532,31 @@ func TestUnmergeableAggregateIsRefused(t *testing.T) {
 	}
 }
 
+// openvoxdb names every extract column after its function and aliases a repeat
+// as "<name>_2", which pdbmux's spec never learns about: it would be neither a
+// grouping key nor a folded aggregate, so the first backend's value would
+// freeze into the merged row. The backend answers such a query, which is
+// exactly why pdbmux has to refuse it.
+func TestRepeatedFunctionColumnIsRefused(t *testing.T) {
+	const q = `["extract",[["function","count","certname"],["function","count","catalog_environment"]]]`
+
+	resp := rawGet(t, nodesPath, query(q))
+	if resp.status != http.StatusBadRequest {
+		t.Fatalf("status %d, want 400: %s", resp.status, resp.body)
+	}
+	if !strings.Contains(string(resp.body), "count") {
+		t.Errorf("refusal %q does not name the clashing column", resp.body)
+	}
+
+	rows := h.a.query(context.Background(), t, nodesPath, query(q))
+	if len(rows) != 1 {
+		t.Fatalf("backend %s returned %d rows for the repeated projection, want 1", h.a.name, len(rows))
+	}
+	if _, ok := rows[0]["count_2"]; !ok {
+		t.Errorf("backend %s row = %v, want the aliased count_2 column the refusal exists for", h.a.name, rows[0])
+	}
+}
+
 // 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.

reports.go

@@ -219,10 +219,7 @@ func parsePaging(v url.Values) (paging, error) {
 
 // Backends are asked for the first offset+limit records with no offset; the offset is applied to the union instead.
 func (p paging) upstreamParams(in url.Values) url.Values {
-	out := url.Values{}
+	out := copyParams(in)
-	for k, vs := range in {
-		out[k] = append([]string(nil), vs...)
-	}
 	out.Del("offset")
 	if p.limit >= 0 {
 		out.Set("limit", strconv.Itoa(p.limit+p.offset))
@@ -230,6 +227,25 @@ func (p paging) upstreamParams(in url.Values) url.Values {
 	return out
 }
 
+// unpagedParams is upstreamParams for a response whose rows are folded together:
+// a backend's own first N groups are not the merged result's first N, and a
+// group truncated away on one backend folds to a wrong value, so every group is
+// fetched and the window is cut after the fold.
+func unpagedParams(in url.Values) url.Values {
+	out := copyParams(in)
+	out.Del("offset")
+	out.Del("limit")
+	return out
+}
+
+func copyParams(in url.Values) url.Values {
+	out := url.Values{}
+	for k, vs := range in {
+		out[k] = append([]string(nil), vs...)
+	}
+	return out
+}
+
 func (p paging) apply(recs []json.RawMessage) []json.RawMessage {
 	if p.offset >= len(recs) {
 		return []json.RawMessage{}

server.go

@@ -471,6 +471,11 @@ func (s *Server) serveCombined(w http.ResponseWriter, r *http.Request, path stri
 		return
 	}
 	upstream := page.upstreamParams(in)
+	if spec != nil && len(spec.aggs) > 0 {
+		// An aggregate returns one row per distinct group, so the whole result is
+		// fetched and paged locally rather than truncated per backend.
+		upstream = unpagedParams(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.

server_test.go

@@ -1114,6 +1114,100 @@ func TestHandler_FactsAggregateRecordsIsMergedRowCount(t *testing.T) {
 	}
 }
 
+// aggregatePagingBackends hold group counts a per-backend limit would truncate
+// to the wrong answer: backend a's own first row is Exec, so a limit pushed
+// upstream drops the File rows that together make File the real top group.
+func aggregatePagingBackends(t *testing.T) (*fakeBackend, *fakeBackend) {
+	t.Helper()
+	a := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[resourcesPath] = `[{"count":6,"type":"Exec"},{"count":5,"type":"File"}]`
+	b := newFakeBackend(t, `[]`, `[]`)
+	b.bodies[resourcesPath] = `[{"count":4,"type":"File"}]`
+	return a, b
+}
+
+const aggregateCountByType = `["extract",[["function","count","certname"],"type"],["group_by","type"]]`
+
+// A group truncated away on one backend would fold to a wrong total, so the
+// whole aggregate is fetched and the window cut after the fold.
+func TestHandler_ResourcesAggregateLimitIsAppliedAfterTheFold(t *testing.T) {
+	a, b := aggregatePagingBackends(t)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{
+		"query":    {aggregateCountByType},
+		"order_by": {`[{"field":"count","order":"desc"}]`},
+		"limit":    {"1"},
+	})
+	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{{"count": float64(9), "type": "File"}}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("top group = %v, want %v", got, want)
+	}
+	for name, fb := range map[string]*fakeBackend{"a": a, "b": b} {
+		p, ok := fb.params(resourcesPath)
+		if !ok {
+			t.Fatalf("%s backend was not queried", name)
+		}
+		if p.Has("limit") || p.Has("offset") {
+			t.Errorf("%s backend got limit=%q offset=%q, want both applied locally", name, p.Get("limit"), p.Get("offset"))
+		}
+	}
+}
+
+// The merged row count is the whole aggregate's, not the paged window's.
+func TestHandler_ResourcesAggregateIncludeTotalWithLocalPaging(t *testing.T) {
+	a, b := aggregatePagingBackends(t)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{
+		"query":         {aggregateCountByType},
+		"order_by":      {`[{"field":"count","order":"desc"}]`},
+		"limit":         {"1"},
+		"offset":        {"1"},
+		"include_total": {"true"},
+	})
+	if got := rec.Header().Get(recordsHeader); got != "2" {
+		t.Errorf("%s = %q, want 2 merged groups", recordsHeader, got)
+	}
+	var got []map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	want := []map[string]any{{"count": float64(6), "type": "Exec"}}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("offset window = %v, want %v", got, want)
+	}
+}
+
+// Row functions return one row per record rather than per group, so nothing is
+// folded and the upstream limit that bounds them still applies.
+func TestHandler_ResourcesRowFunctionKeepsUpstreamLimit(t *testing.T) {
+	a, b := aggregatePagingBackends(t)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{
+		"query": {`["extract",[["function","to_string","line"],"type"]]`},
+		"limit": {"1"},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	p, ok := a.params(resourcesPath)
+	if !ok {
+		t.Fatal("backend a was not queried")
+	}
+	if p.Get("limit") != "1" {
+		t.Errorf("backend got limit=%q, want it forwarded", p.Get("limit"))
+	}
+}
+
 func TestHandler_FactsNonAggregateStillMergedByCertname(t *testing.T) {
 	// Regression: routing aggregates to the summing path must not divert plain
 	// queries, including an extract projection that carries no function column.