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

@@ -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

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
 }
 

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:

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())
+	})
 })

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 | ✅ |

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"},

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()