From 91dbb31a3f3d8b34bfe45683ce4bbea70aab4e35 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Sat, 12 Sep 2026 16:23:34 +0000 Subject: [PATCH] feat(search): metric aggregations sum/min/max/avg over a numeric field, carried as the proto MetricDefinition/Metric message pair end to end: the engines emit a Metric result (avg as sum/count accumulators), the service layer reduces metrics across spaces, the graph layer maps them from and to the libregraph metricDefinition/searchMetric objects. --- services/search/pkg/bleve/aggregations.go | 193 +++++++++++++++++- services/search/pkg/bleve/backend.go | 16 +- .../pkg/opensearch/internal/aggs/aggs.go | 63 +++++- .../pkg/opensearch/internal/aggs/aggs_test.go | 80 ++++++++ services/search/pkg/parity/README.md | 1 + .../search/pkg/parity/aggregations_test.go | 8 + services/search/pkg/search/service.go | 53 ++++- 7 files changed, 400 insertions(+), 14 deletions(-) diff --git a/services/search/pkg/bleve/aggregations.go b/services/search/pkg/bleve/aggregations.go index 149cdef4ab..9e39ef7611 100644 --- a/services/search/pkg/bleve/aggregations.go +++ b/services/search/pkg/bleve/aggregations.go @@ -1,22 +1,30 @@ package bleve import ( + "context" + "errors" "fmt" "strconv" "time" "github.com/blevesearch/bleve/v2" + "github.com/blevesearch/bleve/v2/numeric" bleveSearch "github.com/blevesearch/bleve/v2/search" + "github.com/blevesearch/bleve/v2/search/collector" + index "github.com/blevesearch/bleve_index_api" searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" ) +// Bleve facets count one field, so metrics are folded from doc values by +// aggCollector, hooked into the collector walk through bleve's +// document-match-handler context key. Loading hits for them instead costs a +// stored-document decode per match. + // defaultFacetSize is used when no size is requested; the service layer trims // after cross-space merge. const defaultFacetSize = 1000 -// collected reports an aggregation bleve facets cannot answer: a metric or one -// with sub-aggregations. Facets count one field and cannot nest. func collected(agg *searchService.AggregationOption) bool { return agg.GetMetricDefinition() != nil || len(agg.GetSubAggregations()) > 0 } @@ -134,12 +142,189 @@ func facetBuckets(fr *bleveSearch.FacetResult, agg *searchService.AggregationOpt return buckets } -func extractBleveAggregations(res *bleve.SearchResult, aggs []*searchService.AggregationOption) []*searchService.AggregationResult { +type aggLevel struct { + opt *searchService.AggregationOption +} + +func newAggLevel(opt *searchService.AggregationOption) (*aggLevel, error) { + if opt.GetMetricDefinition() == nil { + return nil, fmt.Errorf("sub-aggregations are not supported by bleve yet") + } + return &aggLevel{opt: opt}, nil +} + +// fieldValues is per-document scratch, reused across documents. +type fieldValues struct { + numbers []int64 +} + +type bucketAcc struct { + value float64 // SUM/MIN/MAX + sum float64 // AVG numerator + count int64 // AVG denominator + seen bool +} + +func newBucketAcc(*aggLevel) *bucketAcc { + return &bucketAcc{} +} + +// aggCollector serves one search; bleve's collector is single-threaded. +type aggCollector struct { + fields map[string]*fieldValues + fieldNames []string + roots map[int]*aggRoot // by position in the request's aggregations +} + +type aggRoot struct { + level *aggLevel + acc *bucketAcc +} + +func newAggCollector(aggs []*searchService.AggregationOption) (*aggCollector, error) { + c := &aggCollector{fields: map[string]*fieldValues{}, roots: map[int]*aggRoot{}} + for i, agg := range aggs { + if !collected(agg) { + continue + } + l, err := newAggLevel(agg) + if err != nil { + return nil, err + } + c.register(l) + c.roots[i] = &aggRoot{level: l, acc: newBucketAcc(l)} + } + if len(c.roots) == 0 { + return nil, nil + } + return c, nil +} + +func (c *aggCollector) register(l *aggLevel) { + if _, ok := c.fields[l.opt.GetField()]; !ok { + c.fields[l.opt.GetField()] = &fieldValues{} + c.fieldNames = append(c.fieldNames, l.opt.GetField()) + } +} + +// The handler runs for every match, before the top-n cut. +func (c *aggCollector) withContext(ctx context.Context) context.Context { + maker := bleveSearch.MakeDocumentMatchHandler(func(sc *bleveSearch.SearchContext) (bleveSearch.DocumentMatchHandler, bool, error) { + inner, loadID, err := collector.MakeTopNDocumentMatchHandler(sc) + if err != nil { + return nil, false, err + } + if inner == nil { + return nil, false, errors.New("aggregations need the top-n collector") + } + dvr, err := sc.IndexReader.DocValueReader(c.fieldNames) + if err != nil { + return nil, false, err + } + return func(d *bleveSearch.DocumentMatch) error { + if d != nil { + if err := c.collect(sc.IndexReader, dvr, d); err != nil { + return err + } + } + return inner(d) + }, loadID, nil + }) + return context.WithValue(ctx, bleveSearch.MakeDocumentMatchHandlerKey, maker) +} + +func (c *aggCollector) collect(reader index.IndexReader, dvr index.DocValueReader, d *bleveSearch.DocumentMatch) error { + if d.IndexInternalID == nil { + id, err := reader.InternalID(d.ID) + if err != nil { + return err + } + d.IndexInternalID = id + } + for _, fv := range c.fields { + fv.numbers = fv.numbers[:0] + } + if err := dvr.VisitDocValues(d.IndexInternalID, c.visit); err != nil { + return err + } + for _, root := range c.roots { + c.fold(root.acc, root.level) + } + return nil +} + +// Numeric and date doc values are prefix-coded at several precisions; only +// shift 0 carries the exact value. +func (c *aggCollector) visit(field string, term []byte) { + fv, ok := c.fields[field] + if !ok { + return + } + pc := numeric.PrefixCoded(term) + if shift, err := pc.Shift(); err == nil && shift == 0 { + if v, err := pc.Int64(); err == nil { + fv.numbers = append(fv.numbers, v) + } + } +} + +func (c *aggCollector) fold(a *bucketAcc, l *aggLevel) { + for _, raw := range c.fields[l.opt.GetField()].numbers { + a.addMetric(l.opt.GetMetricDefinition().GetKind(), numeric.Int64ToFloat64(raw)) + } +} + +func (a *bucketAcc) addMetric(kind searchService.MetricKind, v float64) { + switch kind { + case searchService.MetricKind_METRIC_KIND_SUM: + a.value += v + case searchService.MetricKind_METRIC_KIND_MIN: + if !a.seen || v < a.value { + a.value = v + } + case searchService.MetricKind_METRIC_KIND_MAX: + if !a.seen || v > a.value { + a.value = v + } + case searchService.MetricKind_METRIC_KIND_AVG: + a.sum += v + a.count++ + } + a.seen = true +} + +func (a *bucketAcc) result(l *aggLevel) *searchService.AggregationResult { + if !a.seen { + return nil + } + m := &searchService.Metric{Kind: l.opt.GetMetricDefinition().GetKind()} + r := &searchService.AggregationResult{Field: l.opt.GetField(), Metric: m} + if m.Kind == searchService.MetricKind_METRIC_KIND_AVG { + m.Sum = a.sum + m.Count = a.count + } else { + m.Value = a.value + } + return r +} + +func extractBleveAggregations(res *bleve.SearchResult, aggs []*searchService.AggregationOption, c *aggCollector) []*searchService.AggregationResult { if len(aggs) == 0 { return nil } out := make([]*searchService.AggregationResult, 0, len(aggs)) - for _, agg := range aggs { + for i, agg := range aggs { + if collected(agg) { + if c == nil { + continue + } + if root, ok := c.roots[i]; ok { + if r := root.acc.result(root.level); r != nil { + out = append(out, r) + } + } + continue + } fr, ok := res.Facets[agg.GetField()] if !ok { continue diff --git a/services/search/pkg/bleve/backend.go b/services/search/pkg/bleve/backend.go index 889a034755..de86a79ca9 100644 --- a/services/search/pkg/bleve/backend.go +++ b/services/search/pkg/bleve/backend.go @@ -2,7 +2,6 @@ package bleve import ( "context" - "errors" "math" "time" @@ -43,7 +42,7 @@ func NewBackend(index bleve.Index, queryCreator searchQuery.Creator[query.Query] // Search executes a search request operation within the index. // Returns a SearchIndexResponse object or an error. -func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error) { +func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequest) (*searchService.SearchIndexResponse, error) { createdQuery, err := b.queryCreator.Create(sir.Query) if err != nil { if kql.IsValidationError(err) { @@ -97,7 +96,7 @@ func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexReques for _, agg := range sir.GetAggregations() { if collected(agg) { - return nil, errors.New("metric and nested aggregations are not supported by bleve yet") + continue } fr, err := newBleveFacetRequest(agg) if err != nil { @@ -105,9 +104,16 @@ func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexReques } bleveReq.AddFacet(agg.GetField(), fr) } + aggs, err := newAggCollector(sir.GetAggregations()) + if err != nil { + return nil, err + } + if aggs != nil { + ctx = aggs.withContext(ctx) + } bleveReq.Fields = []string{"*"} - res, err := b.index.Search(bleveReq) + res, err := b.index.SearchInContext(ctx, bleveReq) if err != nil { return nil, err } @@ -163,7 +169,7 @@ func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexReques return &searchService.SearchIndexResponse{ Matches: matches, TotalMatches: int32(totalMatches), - Aggregations: extractBleveAggregations(res, sir.GetAggregations()), + Aggregations: extractBleveAggregations(res, sir.GetAggregations(), aggs), }, nil } diff --git a/services/search/pkg/opensearch/internal/aggs/aggs.go b/services/search/pkg/opensearch/internal/aggs/aggs.go index 4e77402511..c4beb41cd9 100644 --- a/services/search/pkg/opensearch/internal/aggs/aggs.go +++ b/services/search/pkg/opensearch/internal/aggs/aggs.go @@ -17,7 +17,7 @@ import ( const DefaultFacetSize = 1000 // Build translates AggregationOptions into the OpenSearch aggregation DSL -// (terms, range, date_range). Entries get an index-derived +// (terms, range, date_range, metric). Entries get an index-derived // name so repeated aggs on one field don't collide. A range bound that is // neither a number nor a date is an error. func Build(opts []*searchsvc.AggregationOption) (map[string]any, error) { @@ -47,6 +47,10 @@ func buildLevel(opts []*searchsvc.AggregationOption, prefix string) (map[string] func buildOne(opt *searchsvc.AggregationOption) (map[string]any, error) { field := opt.GetField() + if md := opt.GetMetricDefinition(); md != nil { + mk := md.GetKind() + return buildMetric(field, mk), nil + } var entry map[string]any if ranges := rangesOf(opt); len(ranges) > 0 { built, kind, err := buildRanges(field, ranges) @@ -74,6 +78,22 @@ func buildOne(opt *searchsvc.AggregationOption) (map[string]any, error) { return entry, nil } +// buildMetric emits the sum/min/max metric. AVG uses a stats agg to transport +// (sum, count) for the cross-space merge; the service layer collapses to the average. +func buildMetric(field string, kind searchsvc.MetricKind) map[string]any { + switch kind { + case searchsvc.MetricKind_METRIC_KIND_SUM: + return map[string]any{"sum": map[string]any{"field": field}} + case searchsvc.MetricKind_METRIC_KIND_MIN: + return map[string]any{"min": map[string]any{"field": field}} + case searchsvc.MetricKind_METRIC_KIND_MAX: + return map[string]any{"max": map[string]any{"field": field}} + case searchsvc.MetricKind_METRIC_KIND_AVG: + return map[string]any{"stats": map[string]any{"field": field}} + } + return nil +} + func rangesOf(opt *searchsvc.AggregationOption) []*searchsvc.BucketRange { bd := opt.GetBucketDefinition() if bd == nil { @@ -188,6 +208,10 @@ func parseLevel(node aggNode, opts []*searchsvc.AggregationOption, prefix string func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption) *searchsvc.AggregationResult { field := opt.GetField() + if md := opt.GetMetricDefinition(); md != nil { + mk := md.GetKind() + return parseMetric(raw, field, mk) + } var body struct { Buckets []json.RawMessage `json:"buckets"` } @@ -221,6 +245,43 @@ func parseBucket(raw json.RawMessage) *searchsvc.Bucket { return b } +func parseMetric(raw json.RawMessage, field string, kind searchsvc.MetricKind) *searchsvc.AggregationResult { + switch kind { + case searchsvc.MetricKind_METRIC_KIND_SUM, + searchsvc.MetricKind_METRIC_KIND_MIN, + searchsvc.MetricKind_METRIC_KIND_MAX: + var body struct { + Value *float64 `json:"value"` + } + if err := json.Unmarshal(raw, &body); err != nil { + return nil + } + res := &searchsvc.AggregationResult{ + Field: field, + Metric: &searchsvc.Metric{Kind: kind}, + } + if body.Value != nil { + res.Metric.Value = *body.Value + } + return res + case searchsvc.MetricKind_METRIC_KIND_AVG: + var body struct { + Sum float64 `json:"sum"` + Count int64 `json:"count"` + } + if err := json.Unmarshal(raw, &body); err != nil { + return nil + } + return &searchsvc.AggregationResult{ + Field: field, + Metric: &searchsvc.Metric{Kind: kind, Sum: body.Sum, Count: body.Count}, + } + } + return nil +} + +// bucketKeyToString normalises a response key to a string (terms are strings, +// ranges use our "from-to" key, numeric terms come back as JSON numbers). func bucketKeyToString(v any) string { switch x := v.(type) { case string: diff --git a/services/search/pkg/opensearch/internal/aggs/aggs_test.go b/services/search/pkg/opensearch/internal/aggs/aggs_test.go index c7986fdfc6..6e4e9a27b2 100644 --- a/services/search/pkg/opensearch/internal/aggs/aggs_test.go +++ b/services/search/pkg/opensearch/internal/aggs/aggs_test.go @@ -2,6 +2,7 @@ package aggs_test import ( "encoding/json" + "fmt" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -89,6 +90,29 @@ var _ = Describe("Build", func() { Expect(ranges[2]).ToNot(HaveKey("to")) // open upper bound }) + DescribeTable("builds single-value metric aggregations", + func(kind searchsvc.MetricKind, esKind string) { + res := build([]*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricDefinition: &searchsvc.MetricDefinition{Kind: kind}}, + }) + body, ok := res["a_0"].(map[string]any)[esKind].(map[string]any) + Expect(ok).To(BeTrue()) + Expect(body["field"]).To(Equal("audio.duration")) + }, + Entry("sum", searchsvc.MetricKind_METRIC_KIND_SUM, "sum"), + Entry("min", searchsvc.MetricKind_METRIC_KIND_MIN, "min"), + Entry("max", searchsvc.MetricKind_METRIC_KIND_MAX, "max"), + ) + + It("uses a stats aggregation for AVG", func() { + res := build([]*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricDefinition: &searchsvc.MetricDefinition{Kind: searchsvc.MetricKind_METRIC_KIND_AVG}}, + }) + stats, ok := res["a_0"].(map[string]any)["stats"].(map[string]any) + Expect(ok).To(BeTrue()) + Expect(stats["field"]).To(Equal("audio.duration")) + }) + }) var _ = Describe("Parse", func() { @@ -124,4 +148,60 @@ var _ = Describe("Parse", func() { // numeric term key stringified without trailing zeros Expect(out[2].Buckets[0].Key).To(Equal("9")) }) + + DescribeTable("parses single-value metrics", + func(kind searchsvc.MetricKind, value float64) { + raw := json.RawMessage(fmt.Sprintf(`{"a_0": {"value": %g}}`, value)) + out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricDefinition: &searchsvc.MetricDefinition{Kind: kind}}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(1)) + Expect(out[0].GetMetric().GetKind()).To(Equal(kind)) + Expect(out[0].GetMetric().GetValue()).To(Equal(value)) + }, + Entry("sum", searchsvc.MetricKind_METRIC_KIND_SUM, 1234.5), + Entry("min", searchsvc.MetricKind_METRIC_KIND_MIN, 10.0), + Entry("max", searchsvc.MetricKind_METRIC_KIND_MAX, 99.0), + ) + + It("decodes a null metric value to zero", func() { + // OpenSearch returns value: null when a metric has no matching docs. + out, err := aggs.Parse(json.RawMessage(`{"a_0": {"value": null}}`), + []*searchsvc.AggregationOption{{Field: "audio.duration", MetricDefinition: &searchsvc.MetricDefinition{Kind: searchsvc.MetricKind_METRIC_KIND_SUM}}}) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(1)) + Expect(out[0].GetMetric().GetValue()).To(BeZero()) + Expect(out[0].GetMetric().GetKind()).To(Equal(searchsvc.MetricKind_METRIC_KIND_SUM)) + }) + + It("carries avg transport (sum + count) from a stats response", func() { + raw := json.RawMessage(`{ + "a_0": {"count": 100, "min": 30000.0, "max": 500000.0, "avg": 245000.0, "sum": 24500000.0} + }`) + out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricDefinition: &searchsvc.MetricDefinition{Kind: searchsvc.MetricKind_METRIC_KIND_AVG}}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(1)) + Expect(out[0].GetMetric().GetKind()).To(Equal(searchsvc.MetricKind_METRIC_KIND_AVG)) + Expect(out[0].GetMetric().GetSum()).To(Equal(24500000.0)) + Expect(out[0].GetMetric().GetCount()).To(Equal(int64(100))) + }) + + It("returns nil for empty raw or empty options", func() { + got, err := aggs.Parse(nil, []*searchsvc.AggregationOption{{Field: "x"}}) + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(BeNil()) + + got, err = aggs.Parse(json.RawMessage(`{}`), nil) + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(BeNil()) + }) + + It("errors on malformed json and returns no result", func() { + got, err := aggs.Parse(json.RawMessage(`not-json`), []*searchsvc.AggregationOption{{Field: "x"}}) + Expect(err).To(HaveOccurred()) + Expect(got).To(BeNil()) + }) }) diff --git a/services/search/pkg/parity/README.md b/services/search/pkg/parity/README.md index 84cee680d2..ddf01ec340 100644 --- a/services/search/pkg/parity/README.md +++ b/services/search/pkg/parity/README.md @@ -741,6 +741,7 @@ Fixtures: | AGG-03 | `mediatype:audio` reads `artist and album buckets in one request` | audio.albu... Steel=2, audio.albu...Spades=1, audio.album Bomber=2, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.albu... Steel=2, audio.albu...Spades=1, audio.album Bomber=2, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.albu... Steel=2, audio.albu...Spades=1, audio.album Bomber=2, audio.artist Motörhead=3, audio.artist Saxon=2 | ✅ | | AGG-04 | `mediatype:audio` reads `audio.year buckets per decade` | audio.year 1970-1980=2, audio.year 1980-1990=1, audio.year 1990-2000=1, audio.year 2000-2010=3 | audio.year 1970-1980=2, audio.year 1980-1990=1, audio.year 1990-2000=1, audio.year 2000-2010=3 | audio.year 1970-1980=2, audio.year 1980-1990=1, audio.year 1990-2000=1, audio.year 2000-2010=3 | ✅ | | AGG-05 | `mediatype:audio` reads `open-ended audio.year ranges` | audio.year -1990=3, audio.year 2000-=3 | audio.year -1990=3, audio.year 2000-=3 | audio.year -1990=3, audio.year 2000-=3 | ✅ | +| AGG-06 | `mediatype:audio` reads `top-level metrics on audio.year` | audio.year max=2009, audio.year min=1971, audio.year sum=13942, audio.year... count=7 | audio.year max=2009, audio.year min=1971, audio.year sum=13942, audio.year... count=7 | audio.year max=2009, audio.year min=1971, audio.year sum=13942, audio.year... count=7 | ✅ | | AGG-07 | `mediatype:image` reads `photo.takenDateTime buckets per date range` | photo.take...-01-01=1, photo.take...-09-01=2, photo.take...00:00Z=2 | photo.take...-01-01=1, photo.take...-09-01=2, photo.take...00:00Z=2 | photo.take...-01-01=1, photo.take...-09-01=2, photo.take...00:00Z=2 | ✅ | | AGG-08 | `mediatype:image` reads `open-ended date ranges` | photo.take...-01-01=3, photo.take...01-01-=1 | photo.take...-01-01=3, photo.take...01-01-=1 | photo.take...-01-01=3, photo.take...01-01-=1 | ✅ | | AGG-09 | `mediatype:image` reads `malformed date range bound` | error | error | error | ✅ | diff --git a/services/search/pkg/parity/aggregations_test.go b/services/search/pkg/parity/aggregations_test.go index ea901b8db9..4bac338058 100644 --- a/services/search/pkg/parity/aggregations_test.go +++ b/services/search/pkg/parity/aggregations_test.go @@ -96,6 +96,14 @@ func aggregationCases() []aggCase { &searchService.BucketRange{From: "2000"}, )}}, want: []string{"audio.year -1990=3", "audio.year 2000-=3"}}, + {id: 6, query: "mediatype:audio", reads: "top-level metrics on audio.year", + aggs: []*searchService.AggregationOption{ + {Field: "audio.year", MetricDefinition: &searchService.MetricDefinition{Kind: searchService.MetricKind_METRIC_KIND_SUM}}, + {Field: "audio.year", MetricDefinition: &searchService.MetricDefinition{Kind: searchService.MetricKind_METRIC_KIND_MIN}}, + {Field: "audio.year", MetricDefinition: &searchService.MetricDefinition{Kind: searchService.MetricKind_METRIC_KIND_MAX}}, + {Field: "audio.year", MetricDefinition: &searchService.MetricDefinition{Kind: searchService.MetricKind_METRIC_KIND_AVG}}, + }, + want: []string{"audio.year sum=13942", "audio.year min=1971", "audio.year max=2009", "audio.year avg sum=13942 count=7"}}, {id: 7, query: "mediatype:image", reads: "photo.takenDateTime buckets per date range", aggs: []*searchService.AggregationOption{{Field: "photo.takenDateTime", BucketDefinition: ranges( &searchService.BucketRange{From: "2018-08-11T00:00:00Z", To: "2018-08-12T00:00:00Z"}, diff --git a/services/search/pkg/search/service.go b/services/search/pkg/search/service.go index bdb240548a..fcf8bdb774 100644 --- a/services/search/pkg/search/service.go +++ b/services/search/pkg/search/service.go @@ -100,11 +100,10 @@ func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng E // Search processes a search request and passes it down to the engine. func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*searchsvc.SearchResponse, error) { - // bucket aggregations only for now: the engines do not evaluate metrics - // and sub-aggregations yet + // the engines do not evaluate sub-aggregations yet for _, opt := range req.GetAggregations() { - if opt.GetMetricDefinition() != nil || len(opt.GetSubAggregations()) > 0 { - return nil, errtypes.BadRequest("metric and nested aggregations are not supported yet") + if len(opt.GetSubAggregations()) > 0 { + return nil, errtypes.BadRequest("nested aggregations are not supported yet") } } s.logger.Debug().Str("query", req.Query).Msg("performing a search") @@ -145,6 +144,7 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se return nil, errtypes.BadRequest("empty query provided") } req.Query = query + if len(scope) > 0 { scopedID, err := storagespace.ParseID(scope) if err != nil { @@ -298,6 +298,7 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se } mergedAggregations := map[string]map[string]*searchmsgBucket{} + mergedMetrics := map[string]*searchsvc.AggregationResult{} for _, res := range responses { if res == nil { continue @@ -307,6 +308,25 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se matches = append(matches, match) } for _, agg := range res.GetAggregations() { + // Top-level metric: reduce across spaces; keyed by field+kind so + // several metrics on the same field stay separate. + if m := agg.GetMetric(); m != nil { + kind := m.GetKind() + key := agg.GetField() + "|" + kind.String() + existing, ok := mergedMetrics[key] + if !ok { + mergedMetrics[key] = agg + continue + } + em := existing.GetMetric() + if kind == searchsvc.MetricKind_METRIC_KIND_AVG { + em.Sum += m.GetSum() + em.Count += m.GetCount() + } else { + em.Value = reduceMetric(kind, em.GetValue(), m.GetValue()) + } + continue + } field := agg.GetField() if _, ok := mergedAggregations[field]; !ok { mergedAggregations[field] = map[string]*searchmsgBucket{} @@ -337,6 +357,12 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se aggregations := make([]*searchsvc.AggregationResult, 0, len(req.GetAggregations())) for _, opt := range req.GetAggregations() { field := opt.GetField() + if md := opt.GetMetricDefinition(); md != nil { + if m, ok := mergedMetrics[field+"|"+md.GetKind().String()]; ok { + aggregations = append(aggregations, m) + } + continue + } bucketMap := mergedAggregations[field] buckets := make([]*searchsvc.Bucket, 0, len(bucketMap)) for _, b := range bucketMap { @@ -358,6 +384,25 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se type searchmsgBucket = searchsvc.Bucket +// reduceMetric is only called when both sides carry a value. +func reduceMetric(kind searchsvc.MetricKind, a, b float64) float64 { + switch kind { + case searchsvc.MetricKind_METRIC_KIND_SUM: + return a + b + case searchsvc.MetricKind_METRIC_KIND_MIN: + if b < a { + return b + } + return a + case searchsvc.MetricKind_METRIC_KIND_MAX: + if b > a { + return b + } + return a + } + return a +} + // postProcessBuckets applies the BucketDefinition; defaults to count-descending. func postProcessBuckets(buckets []*searchsvc.Bucket, opt *searchsvc.AggregationOption) []*searchsvc.Bucket { bd := opt.GetBucketDefinition()