diff --git a/README.md b/README.md index f7cec1b..8c935c4 100644 --- a/README.md +++ b/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 `_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` - 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` that also projects a `sum` or a `count` 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. + - `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 diff --git a/aggregate.go b/aggregate.go index ec45446..ec93b38 100644 --- a/aggregate.go +++ b/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 "_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 { diff --git a/aggregate_test.go b/aggregate_test.go index b10f0ee..1e05552 100644 --- a/aggregate_test.go +++ b/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{ diff --git a/e2e_query_test.go b/e2e_query_test.go index 001d6bf..f95606e 100644 --- a/e2e_query_test.go +++ b/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 "_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. diff --git a/reports.go b/reports.go index a12b7d6..e9ee47d 100644 --- a/reports.go +++ b/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{} - for k, vs := range in { - out[k] = append([]string(nil), vs...) - } + out := copyParams(in) 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{} diff --git a/server.go b/server.go index ab42430..2d38958 100644 --- a/server.go +++ b/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. diff --git a/server_test.go b/server_test.go index 79c025b..8768c59 100644 --- a/server_test.go +++ b/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.