unkin-agent
3eb7d53fe8
A group_by key that repeats a folded column made the column a grouping key and an aggregate at once, and for avg it left the upstream query grouping on a column the rewrite had removed, so the request failed as an opaque 502. - refuse a group_by field that names a folded aggregate or the avg column - cover the function-then-field ordering of the existing clash check
666 lines
25 KiB
Go
666 lines
25 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"net/url"
|
|
"reflect"
|
|
"slices"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
func rows(raws ...string) []record {
|
|
out := make([]record, 0, len(raws))
|
|
for _, r := range raws {
|
|
out = append(out, record{Raw: json.RawMessage(r)})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// decodeRows turns a merged result set into comparable maps.
|
|
func decodeRows(t *testing.T, raws []json.RawMessage) []map[string]any {
|
|
t.Helper()
|
|
out := make([]map[string]any, 0, len(raws))
|
|
for _, raw := range raws {
|
|
var m map[string]any
|
|
if err := json.Unmarshal(raw, &m); err != nil {
|
|
t.Fatalf("unmarshal %s: %v", raw, err)
|
|
}
|
|
out = append(out, m)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// 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(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 := 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(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)
|
|
}
|
|
}
|
|
|
|
func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) {
|
|
for _, q := range []string{
|
|
``,
|
|
`["=","certname","h1"]`,
|
|
`["extract",["certname","hash"],["=","certname","h1"]]`, // projection, still real reports
|
|
`not json`,
|
|
`["extract"]`,
|
|
`{"not":"an array"}`,
|
|
} {
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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.shape)
|
|
|
|
want := []map[string]any{
|
|
{"count": float64(7), "status": "changed"},
|
|
{"count": float64(3), "status": "failed"},
|
|
}
|
|
if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("merged = %v, want %v", got, want)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|
|
|
|
// The ordering the sibling case does not cover: the function column comes first
|
|
// and a later plain field takes the response key it already named.
|
|
func TestParseAggregate_FunctionClashingWithALaterFieldIsRefused(t *testing.T) {
|
|
const q = `["extract",[["function","count","certname"],"count"],["=","certname","h1"]]`
|
|
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 group_by naming a folded column would make it a grouping key and an
|
|
// aggregate at once, so it is refused rather than sent upstream to fail.
|
|
func TestParseAggregate_GroupByOnAnAggregateColumnIsRefused(t *testing.T) {
|
|
for _, fn := range []string{"count", "sum", "min", "max", "avg"} {
|
|
q := `["extract",[["function","` + fn + `","line"]],["group_by","` + fn + `"]]`
|
|
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 refusal is limited to the folded columns: grouping on a plain field or on
|
|
// a row function's own column stays legitimate.
|
|
func TestParseAggregate_GroupByOnANonAggregateColumnIsKept(t *testing.T) {
|
|
for _, q := range []string{
|
|
`["extract",[["function","count","certname"],"status"],["group_by","status"]]`,
|
|
`["extract",[["function","to_string","producer_timestamp","FMDD"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDD"]]]`,
|
|
`["extract",[["function","avg","line"],"type"],["group_by","type"]]`,
|
|
} {
|
|
if spec, err := parseAggregate(q); err != nil || spec == nil {
|
|
t.Errorf("parseAggregate(%s) = %+v, %v, want an accepted spec", q, spec, 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{
|
|
{name: "a", records: rows(`{"count":3,"status":"changed"}`)},
|
|
{name: "b", records: rows(`{"count":2,"status":"skipped"}`)},
|
|
}, spec.shape)
|
|
|
|
want := []map[string]any{
|
|
{"count": float64(3), "status": "changed"},
|
|
{"count": float64(2), "status": "skipped"},
|
|
}
|
|
if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("merged = %v, want %v", got, want)
|
|
}
|
|
}
|
|
|
|
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 := combineRows([]backendResult{
|
|
{name: "a", records: rows(raw)},
|
|
{name: "b", records: nil},
|
|
}, spec.shape)
|
|
|
|
if len(merged) != 1 || string(merged[0]) != raw {
|
|
t.Errorf("merged = %s, want the row verbatim %s", merged, raw)
|
|
}
|
|
}
|
|
|
|
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.shape)
|
|
|
|
want := []map[string]any{{"count": float64(5), "status": "changed"}}
|
|
if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("merged = %v, want the numeric value preserved %v", got, want)
|
|
}
|
|
}
|
|
|
|
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.shape)
|
|
|
|
want := []map[string]any{{"count": float64(6), "status": "changed"}}
|
|
if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("merged = %v, want %v", got, want)
|
|
}
|
|
}
|
|
|
|
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.shape)
|
|
|
|
if len(merged) != 2 || string(merged[0]) != `"surprise"` {
|
|
t.Fatalf("merged = %s, want the non-object row kept as-is", merged)
|
|
}
|
|
}
|
|
|
|
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 := mustAggregate(t, `["extract",[["function","count"]],["=","certname","h1"]]`)
|
|
merged := combineRows([]backendResult{
|
|
{name: "a", records: rows(`{"count":10}`)},
|
|
{name: "b", records: rows(`{"count":32}`)},
|
|
}, spec.shape)
|
|
|
|
want := []map[string]any{{"count": float64(42)}}
|
|
if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
|
|
t.Errorf("merged = %v, want %v", got, want)
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
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(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 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}`,
|
|
)},
|
|
{name: "b", records: rows(
|
|
`{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`,
|
|
)},
|
|
}, inferredShape)
|
|
|
|
got := decodeRows(t, merged)
|
|
want := []map[string]any{
|
|
{"subject_type": "certname", "subject": map[string]any{"title": "h1"},
|
|
"failures": float64(4), "successes": float64(6), "noops": float64(1), "skips": float64(0)},
|
|
{"subject_type": "certname", "subject": map[string]any{"title": "h2"},
|
|
"failures": float64(0), "successes": float64(5), "noops": float64(0), "skips": float64(0)},
|
|
}
|
|
if !reflect.DeepEqual(got, want) {
|
|
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)
|
|
}
|
|
})
|
|
}
|
|
}
|