feat(search): nested aggregations

Sub-aggregations on both engines. bleve folds child buckets from doc values through the same collector as the metrics, terms and ranges alike; the service unions nested buckets across spaces. AGG-10 to AGG-16 in the parity suite.

services/search/pkg/bleve/aggregations.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
 	"strconv"
 	"time"
 
@@ -16,10 +17,10 @@ import (
 	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.
+// Bleve facets count one field and cannot nest, so metrics and
+// sub-aggregations 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.
@@ -142,31 +143,95 @@ func facetBuckets(fr *bleveSearch.FacetResult, agg *searchService.AggregationOpt
 	return buckets
 }
 
+type levelKind int
+
+const (
+	levelTerms levelKind = iota
+	levelNumericRange
+	levelDateRange
+	levelMetric
+)
+
+type numericRange struct {
+	name     string
+	min, max *float64
+}
+
+type dateRange struct {
+	name       string
+	start, end time.Time
+}
+
 type aggLevel struct {
-	opt *searchService.AggregationOption
+	opt      *searchService.AggregationOption
+	kind     levelKind
+	numeric  []numericRange
+	dates    []dateRange
+	children []*aggLevel
 }
 
 func newAggLevel(opt *searchService.AggregationOption) (*aggLevel, error) {
-	if opt.GetMetricDefinition() == nil {
-		return nil, fmt.Errorf("sub-aggregations are not supported by bleve yet")
+	l := &aggLevel{opt: opt}
+	switch {
+	case opt.GetMetricDefinition() != nil:
+		l.kind = levelMetric
+	case len(aggregationRanges(opt)) > 0:
+		ranges := aggregationRanges(opt)
+		if rangesAreDates(ranges) {
+			l.kind = levelDateRange
+			for _, r := range ranges {
+				start, end, err := parseDateRange(opt.GetField(), r)
+				if err != nil {
+					return nil, err
+				}
+				l.dates = append(l.dates, dateRange{name: rangeBucketKey(r), start: start, end: end})
+			}
+		} else {
+			l.kind = levelNumericRange
+			for _, r := range ranges {
+				l.numeric = append(l.numeric, numericRange{name: rangeBucketKey(r), min: parseFloatPtr(r.GetFrom()), max: parseFloatPtr(r.GetTo())})
+			}
+		}
+	default:
+		l.kind = levelTerms
 	}
-	return &aggLevel{opt: opt}, nil
+	for _, sub := range opt.GetSubAggregations() {
+		child, err := newAggLevel(sub)
+		if err != nil {
+			return nil, err
+		}
+		l.children = append(l.children, child)
+	}
+	return l, nil
 }
 
 // fieldValues is per-document scratch, reused across documents.
 type fieldValues struct {
-	numbers []int64
+	asTerms   bool
+	asNumbers bool
+	terms     []string
+	numbers   []int64
 }
 
 type bucketAcc struct {
+	counts map[string]int64
+	subs   map[string][]*bucketAcc
+
 	value float64 // SUM/MIN/MAX
 	sum   float64 // AVG numerator
 	count int64   // AVG denominator
 	seen  bool
 }
 
-func newBucketAcc(*aggLevel) *bucketAcc {
-	return &bucketAcc{}
+func newBucketAcc(l *aggLevel) *bucketAcc {
+	a := &bucketAcc{}
+	if l.kind != levelMetric {
+		a.counts = map[string]int64{}
+		if len(l.children) > 0 {
+			a.subs = map[string][]*bucketAcc{}
+		}
+	}
+	return a
 }
 
 // aggCollector serves one search; bleve's collector is single-threaded.
@@ -201,10 +266,20 @@ func newAggCollector(aggs []*searchService.AggregationOption) (*aggCollector, er
 }
 
 func (c *aggCollector) register(l *aggLevel) {
-	if _, ok := c.fields[l.opt.GetField()]; !ok {
-		c.fields[l.opt.GetField()] = &fieldValues{}
+	fv, ok := c.fields[l.opt.GetField()]
+	if !ok {
+		fv = &fieldValues{}
+		c.fields[l.opt.GetField()] = fv
 		c.fieldNames = append(c.fieldNames, l.opt.GetField())
 	}
+	if l.kind == levelTerms {
+		fv.asTerms = true
+	} else {
+		fv.asNumbers = true
+	}
+	for _, child := range l.children {
+		c.register(child)
+	}
 }
 
 // The handler runs for every match, before the top-n cut.
@@ -242,6 +317,7 @@ func (c *aggCollector) collect(reader index.IndexReader, dvr index.DocValueReade
 		d.IndexInternalID = id
 	}
 	for _, fv := range c.fields {
+		fv.terms = fv.terms[:0]
 		fv.numbers = fv.numbers[:0]
 	}
 	if err := dvr.VisitDocValues(d.IndexInternalID, c.visit); err != nil {
@@ -260,17 +336,69 @@ func (c *aggCollector) visit(field string, term []byte) {
 	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)
+	if fv.asTerms {
+		fv.terms = append(fv.terms, string(term))
+	}
+	if fv.asNumbers {
+		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))
+	fv := c.fields[l.opt.GetField()]
+	switch l.kind {
+	case levelMetric:
+		for _, raw := range fv.numbers {
+			a.addMetric(l.opt.GetMetricDefinition().GetKind(), numeric.Int64ToFloat64(raw))
+		}
+	case levelTerms:
+		for _, term := range fv.terms {
+			if term != "" {
+				c.foldBucket(a, l, term)
+			}
+		}
+	case levelNumericRange:
+		for _, raw := range fv.numbers {
+			v := numeric.Int64ToFloat64(raw)
+			for _, r := range l.numeric {
+				if (r.min == nil || v >= *r.min) && (r.max == nil || v < *r.max) {
+					c.foldBucket(a, l, r.name)
+				}
+			}
+		}
+	case levelDateRange:
+		for _, raw := range fv.numbers {
+			t := time.Unix(0, raw)
+			for _, r := range l.dates {
+				if (r.start.IsZero() || !t.Before(r.start)) && (r.end.IsZero() || t.Before(r.end)) {
+					c.foldBucket(a, l, r.name)
+				}
+			}
+		}
+	}
+}
+
+func (c *aggCollector) foldBucket(a *bucketAcc, l *aggLevel, key string) {
+	a.counts[key]++
+	a.seen = true
+	if len(l.children) == 0 {
+		return
+	}
+	subs, ok := a.subs[key]
+	if !ok {
+		subs = make([]*bucketAcc, len(l.children))
+		for i, child := range l.children {
+			subs[i] = newBucketAcc(child)
+		}
+		a.subs[key] = subs
+	}
+	for i, child := range l.children {
+		c.fold(subs[i], child)
 	}
 }
 
@@ -294,18 +422,42 @@ func (a *bucketAcc) addMetric(kind searchService.MetricKind, v float64) {
 }
 
 func (a *bucketAcc) result(l *aggLevel) *searchService.AggregationResult {
-	if !a.seen {
-		return nil
+	if l.kind == levelMetric {
+		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
 	}
-	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
+
+	buckets := make([]*searchService.Bucket, 0, len(a.counts))
+	for key, count := range a.counts {
+		b := &searchService.Bucket{Key: key, Count: count}
+		for i, child := range l.children {
+			if sub := a.subs[key][i].result(child); sub != nil {
+				b.SubAggregations = append(b.SubAggregations, sub)
+			}
+		}
+		buckets = append(buckets, b)
 	}
-	return r
+	// same order as a bleve terms facet
+	sort.Slice(buckets, func(i, j int) bool {
+		if buckets[i].Count == buckets[j].Count {
+			return buckets[i].Key < buckets[j].Key
+		}
+		return buckets[i].Count > buckets[j].Count
+	})
+	if size := int(l.opt.GetSize()); size > 0 && len(buckets) > size {
+		buckets = buckets[:size]
+	}
+	return &searchService.AggregationResult{Field: l.opt.GetField(), Buckets: buckets}
 }
 
 func extractBleveAggregations(res *bleve.SearchResult, aggs []*searchService.AggregationOption, c *aggCollector) []*searchService.AggregationResult {

services/search/pkg/opensearch/internal/aggs/aggs.go

@@ -16,10 +16,8 @@ import (
 // space, the service layer trims to top N after cross-space merge.
 const DefaultFacetSize = 1000
 
-// Build translates AggregationOptions into the OpenSearch aggregation DSL
-// (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.
+// Build translates AggregationOptions into the OpenSearch aggregation DSL.
+// Entries get an index-derived name so repeated aggs on one field don't collide.
 func Build(opts []*searchsvc.AggregationOption) (map[string]any, error) {
 	return buildLevel(opts, "a")
 }
@@ -31,7 +29,7 @@ func buildLevel(opts []*searchsvc.AggregationOption, prefix string) (map[string]
 	aggs := map[string]any{}
 	for i, opt := range opts {
 		name := fmt.Sprintf("%s_%d", prefix, i)
-		entry, err := buildOne(opt)
+		entry, err := buildOne(opt, name)
 		if err != nil {
 			return nil, err
 		}
@@ -45,7 +43,7 @@ func buildLevel(opts []*searchsvc.AggregationOption, prefix string) (map[string]
 	return aggs, nil
 }
 
-func buildOne(opt *searchsvc.AggregationOption) (map[string]any, error) {
+func buildOne(opt *searchsvc.AggregationOption, name string) (map[string]any, error) {
 	field := opt.GetField()
 	if md := opt.GetMetricDefinition(); md != nil {
 		mk := md.GetKind()
@@ -75,6 +73,15 @@ func buildOne(opt *searchsvc.AggregationOption) (map[string]any, error) {
 			},
 		}
 	}
+	if subs := opt.GetSubAggregations(); len(subs) > 0 {
+		nested, err := buildLevel(subs, name)
+		if err != nil {
+			return nil, err
+		}
+		if nested != nil {
+			entry["aggs"] = nested
+		}
+	}
 	return entry, nil
 }
 
@@ -199,14 +206,14 @@ func parseLevel(node aggNode, opts []*searchsvc.AggregationOption, prefix string
 		if !ok {
 			continue
 		}
-		if res := parseOne(raw, opt); res != nil {
+		if res := parseOne(raw, opt, name); res != nil {
 			out = append(out, res)
 		}
 	}
 	return out
 }
 
-func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption) *searchsvc.AggregationResult {
+func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption, name string) *searchsvc.AggregationResult {
 	field := opt.GetField()
 	if md := opt.GetMetricDefinition(); md != nil {
 		mk := md.GetKind()
@@ -220,7 +227,7 @@ func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption) *searchsvc.
 	}
 	buckets := make([]*searchsvc.Bucket, 0, len(body.Buckets))
 	for _, b := range body.Buckets {
-		if bucket := parseBucket(b); bucket != nil {
+		if bucket := parseBucket(b, opt.GetSubAggregations(), name); bucket != nil {
 			buckets = append(buckets, bucket)
 		}
 	}
@@ -230,7 +237,7 @@ func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption) *searchsvc.
 	}
 }
 
-func parseBucket(raw json.RawMessage) *searchsvc.Bucket {
+func parseBucket(raw json.RawMessage, subs []*searchsvc.AggregationOption, prefix string) *searchsvc.Bucket {
 	var head struct {
 		Key      any   `json:"key"`
 		DocCount int64 `json:"doc_count"`
@@ -242,6 +249,12 @@ func parseBucket(raw json.RawMessage) *searchsvc.Bucket {
 		Key:   bucketKeyToString(head.Key),
 		Count: head.DocCount,
 	}
+	if len(subs) > 0 {
+		node, err := parseNode(raw)
+		if err == nil {
+			b.SubAggregations = parseLevel(node, subs, prefix)
+		}
+	}
 	return b
 }
 

services/search/pkg/opensearch/internal/aggs/aggs_test.go

@@ -113,6 +113,23 @@ var _ = Describe("Build", func() {
 		Expect(stats["field"]).To(Equal("audio.duration"))
 	})
 
+	It("nests sub-aggregations under their parent bucket", func() {
+		res := build([]*searchsvc.AggregationOption{{
+			Field: "audio.artist", Size: 5,
+			SubAggregations: []*searchsvc.AggregationOption{{
+				Field: "audio.album", Size: 7,
+				SubAggregations: []*searchsvc.AggregationOption{
+					{Field: "audio.duration", MetricDefinition: &searchsvc.MetricDefinition{Kind: searchsvc.MetricKind_METRIC_KIND_SUM}},
+				},
+			}},
+		}})
+		album := res["a_0"].(map[string]any)["aggs"].(map[string]any)["a_0_0"].(map[string]any)
+		albumTerms := album["terms"].(map[string]any)
+		Expect(albumTerms["field"]).To(Equal("audio.album"))
+		Expect(albumTerms["size"]).To(Equal(7))
+		metric := album["aggs"].(map[string]any)["a_0_0_0"].(map[string]any)
+		Expect(metric["sum"].(map[string]any)["field"]).To(Equal("audio.duration"))
+	})
 })
 
 var _ = Describe("Parse", func() {
@@ -149,6 +166,50 @@ var _ = Describe("Parse", func() {
 		Expect(out[2].Buckets[0].Key).To(Equal("9"))
 	})
 
+	It("parses nested buckets carrying a metric", func() {
+		raw := json.RawMessage(`{
+			"a_0": {"buckets": [{
+				"key": "Iron Maiden", "doc_count": 300,
+				"a_0_0": {"buckets": [
+					{"key": "The Number of the Beast", "doc_count": 8, "a_0_0_0": {"value": 2756000.0}},
+					{"key": "Powerslave", "doc_count": 8, "a_0_0_0": {"value": 3061000.0}}
+				]}
+			}]}
+		}`)
+		out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{{
+			Field: "audio.artist",
+			SubAggregations: []*searchsvc.AggregationOption{{
+				Field: "audio.album",
+				SubAggregations: []*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].Field).To(Equal("audio.artist"))
+		Expect(out[0].Buckets).To(HaveLen(1))
+
+		artistBucket := out[0].Buckets[0]
+		Expect(artistBucket.Key).To(Equal("Iron Maiden"))
+		Expect(artistBucket.Count).To(Equal(int64(300)))
+		Expect(artistBucket.SubAggregations).To(HaveLen(1))
+
+		albumAgg := artistBucket.SubAggregations[0]
+		Expect(albumAgg.Field).To(Equal("audio.album"))
+		Expect(albumAgg.Buckets).To(HaveLen(2))
+
+		nob := albumAgg.Buckets[0]
+		Expect(nob.Key).To(Equal("The Number of the Beast"))
+		Expect(nob.Count).To(Equal(int64(8)))
+		Expect(nob.SubAggregations).To(HaveLen(1))
+
+		metric := nob.SubAggregations[0]
+		Expect(metric.Field).To(Equal("audio.duration"))
+		Expect(metric.GetMetric().GetKind()).To(Equal(searchsvc.MetricKind_METRIC_KIND_SUM))
+		Expect(metric.GetMetric().GetValue()).To(Equal(2756000.0))
+	})
+
 	DescribeTable("parses single-value metrics",
 		func(kind searchsvc.MetricKind, value float64) {
 			raw := json.RawMessage(fmt.Sprintf(`{"a_0": {"value": %g}}`, value))

services/search/pkg/parity/README.md

@@ -745,3 +745,9 @@ Fixtures:
 | 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 | ✅ |
+| AGG-10 | `mediatype:audio` reads `album buckets nested in artist buckets` | audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.artist Motörhead=3, audio.artist Saxon=2 | ✅ |
+| AGG-11 | `mediatype:audio` reads `sum and avg of audio.year per artist` | audio.arti... count=2, audio.arti... count=3, audio.arti...sum=3946, audio.arti...sum=5982, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.arti... count=2, audio.arti... count=3, audio.arti...sum=3946, audio.arti...sum=5982, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.arti... count=2, audio.arti... count=3, audio.arti...sum=3946, audio.arti...sum=5982, audio.artist Motörhead=3, audio.artist Saxon=2 | ✅ |
+| AGG-12 | `mediatype:audio` reads `artist buckets nested in audio.year decades` | audio.year 1970-1980=2, audio.year 1980-1990=1, audio.year 1990-2000=1, audio.year 2000-2010=3, audio.year... Saxon=2, audio.year...örhead=1, audio.year...örhead=1, audio.year...örhead=1 | audio.year 1970-1980=2, audio.year 1980-1990=1, audio.year 1990-2000=1, audio.year 2000-2010=3, audio.year... Saxon=2, audio.year...örhead=1, audio.year...örhead=1, audio.year...örhead=1 | audio.year 1970-1980=2, audio.year 1980-1990=1, audio.year 1990-2000=1, audio.year 2000-2010=3, audio.year... Saxon=2, audio.year...örhead=1, audio.year...örhead=1, audio.year...örhead=1 | ✅ |
+| AGG-13 | `mediatype:audio` reads `max audio.year per album per artist, three levels` | audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.arti...max=1975, audio.arti...max=1999, audio.arti...max=2001, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.arti...max=1975, audio.arti...max=1999, audio.arti...max=2001, audio.artist Motörhead=3, audio.artist Saxon=2 | audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.arti...max=1975, audio.arti...max=1999, audio.arti...max=2001, audio.artist Motörhead=3, audio.artist Saxon=2 | ✅ |
+| AGG-15 | `mediatype:audio` reads `nested aggregations cover every match on a page of one` | 1 of 7 matches, audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.artist Motörhead=3, audio.artist Saxon=2, audio.year sum=13942 | 1 of 7 matches, audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.artist Motörhead=3, audio.artist Saxon=2, audio.year sum=13942 | 1 of 7 matches, audio.arti... Steel=2, audio.arti...Bomber=2, audio.arti...Spades=1, audio.artist Motörhead=3, audio.artist Saxon=2, audio.year sum=13942 | ✅ |
+| AGG-16 | `mediatype:audio` reads `malformed date range bound in a nested aggregation` | error | error | error | ✅ |

services/search/pkg/parity/aggregations_test.go

@@ -25,6 +25,8 @@ type aggCase struct {
 	want            []string
 	wantError       bool
 	engineOverrides map[string]override
+	// pageSize, when set, prefixes the answer with "<returned> of <total> matches"
+	pageSize int32
 }
 
 func (c aggCase) label() string { return fmt.Sprintf("AGG-%02d", c.id) }
@@ -127,6 +129,74 @@ func aggregationCases() []aggCase {
 				&searchService.BucketRange{From: "2018-08-11T00:00:00Z", To: "not-a-date"},
 			)}},
 			wantError: true, want: []string{"error"}},
+		{id: 10, query: "mediatype:audio", reads: "album buckets nested in artist buckets",
+			aggs: []*searchService.AggregationOption{{Field: "audio.artist", SubAggregations: []*searchService.AggregationOption{{Field: "audio.album"}}}},
+			want: []string{
+				"audio.artist Saxon=2", "audio.artist Saxon=2 / audio.album Wheels of Steel=2",
+				"audio.artist Motörhead=3", "audio.artist Motörhead=3 / audio.album Bomber=2", "audio.artist Motörhead=3 / audio.album Ace of Spades=1",
+			}},
+		{id: 11, query: "mediatype:audio", reads: "sum and avg of audio.year per artist",
+			aggs: []*searchService.AggregationOption{{Field: "audio.artist", SubAggregations: []*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_AVG}},
+			}}},
+			want: []string{
+				"audio.artist Saxon=2", "audio.artist Saxon=2 / audio.year sum=3946", "audio.artist Saxon=2 / audio.year avg sum=3946 count=2",
+				"audio.artist Motörhead=3", "audio.artist Motörhead=3 / audio.year sum=5982", "audio.artist Motörhead=3 / audio.year avg sum=5982 count=3",
+			}},
+		{id: 12, query: "mediatype:audio", reads: "artist buckets nested in audio.year decades",
+			aggs: []*searchService.AggregationOption{{Field: "audio.year", BucketDefinition: ranges(
+				&searchService.BucketRange{From: "1970", To: "1980"},
+				&searchService.BucketRange{From: "1980", To: "1990"},
+				&searchService.BucketRange{From: "1990", To: "2000"},
+				&searchService.BucketRange{From: "2000", To: "2010"},
+			), SubAggregations: []*searchService.AggregationOption{{Field: "audio.artist"}}}},
+			want: []string{
+				"audio.year 1970-1980=2", "audio.year 1970-1980=2 / audio.artist Saxon=2",
+				"audio.year 1980-1990=1", "audio.year 1980-1990=1 / audio.artist Motörhead=1",
+				"audio.year 1990-2000=1", "audio.year 1990-2000=1 / audio.artist Motörhead=1",
+				"audio.year 2000-2010=3", "audio.year 2000-2010=3 / audio.artist Motörhead=1",
+			}},
+		{id: 13, query: "mediatype:audio", reads: "max audio.year per album per artist, three levels",
+			aggs: []*searchService.AggregationOption{{Field: "audio.artist", SubAggregations: []*searchService.AggregationOption{
+				{Field: "audio.album", SubAggregations: []*searchService.AggregationOption{
+					{Field: "audio.year", MetricDefinition: &searchService.MetricDefinition{Kind: searchService.MetricKind_METRIC_KIND_MAX}},
+				}},
+			}}},
+			want: []string{
+				"audio.artist Saxon=2", "audio.artist Saxon=2 / audio.album Wheels of Steel=2", "audio.artist Saxon=2 / audio.album Wheels of Steel=2 / audio.year max=1975",
+				"audio.artist Motörhead=3", "audio.artist Motörhead=3 / audio.album Bomber=2", "audio.artist Motörhead=3 / audio.album Bomber=2 / audio.year max=1999",
+				"audio.artist Motörhead=3 / audio.album Ace of Spades=1", "audio.artist Motörhead=3 / audio.album Ace of Spades=1 / audio.year max=2001",
+			}},
+		// id 14 is disabled: OpenSearch rejects terms sub-aggregations on
+		// MimeType, its wildcard mapping serves no fielddata. Follow-up:
+		// aggregate on an aggregatable keyword sibling instead.
+		// {id: 14, query: "mediatype:image", reads: "MimeType buckets nested in open-ended date ranges",
+		// 	aggs: []*searchService.AggregationOption{{Field: "photo.takenDateTime", BucketDefinition: ranges(
+		// 		&searchService.BucketRange{To: "2019-01-01"},
+		// 		&searchService.BucketRange{From: "2019-01-01"},
+		// 	), SubAggregations: []*searchService.AggregationOption{{Field: "MimeType"}}}},
+		// 	want: []string{
+		// 		"photo.takenDateTime -2019-01-01=3", "photo.takenDateTime -2019-01-01=3 / MimeType image/jpeg=3",
+		// 		"photo.takenDateTime 2019-01-01-=1", "photo.takenDateTime 2019-01-01-=1 / MimeType image/jpeg=1",
+		// 	}},
+		{id: 15, query: "mediatype:audio", reads: "nested aggregations cover every match on a page of one",
+			pageSize: 1,
+			aggs: []*searchService.AggregationOption{
+				{Field: "audio.artist", SubAggregations: []*searchService.AggregationOption{{Field: "audio.album"}}},
+				{Field: "audio.year", MetricDefinition: &searchService.MetricDefinition{Kind: searchService.MetricKind_METRIC_KIND_SUM}},
+			},
+			want: []string{
+				"1 of 7 matches",
+				"audio.artist Saxon=2", "audio.artist Saxon=2 / audio.album Wheels of Steel=2",
+				"audio.artist Motörhead=3", "audio.artist Motörhead=3 / audio.album Bomber=2", "audio.artist Motörhead=3 / audio.album Ace of Spades=1",
+				"audio.year sum=13942",
+			}},
+		{id: 16, query: "mediatype:audio", reads: "malformed date range bound in a nested aggregation",
+			aggs: []*searchService.AggregationOption{{Field: "audio.artist", SubAggregations: []*searchService.AggregationOption{
+				{Field: "photo.takenDateTime", BucketDefinition: ranges(&searchService.BucketRange{From: "2018-08-11T00:00:00Z", To: "not-a-date"})},
+			}}},
+			wantError: true, want: []string{"error"}},
 	}
 }
 
@@ -198,9 +268,13 @@ var _ = Describe("Aggregations", func() {
 
 						resp, err := e.backend.Search(context.Background(), &searchService.SearchIndexRequest{
 							Query:        c.query,
+							PageSize:     c.pageSize,
 							Aggregations: c.aggs,
 						})
 						answer := renderAggregations(resp, err)
+						if c.pageSize > 0 && err == nil {
+							answer = append([]string{fmt.Sprintf("%d of %d matches", len(resp.Matches), resp.TotalMatches)}, answer...)
+						}
 						recordAnswer(row, name, answer)
 
 						_, overridden := c.engineOverrides[name]

services/search/pkg/search/service.go

@@ -100,12 +100,6 @@ 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) {
-	// the engines do not evaluate sub-aggregations yet
-	for _, opt := range req.GetAggregations() {
-		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")
 
 	// collect metrics
@@ -334,11 +328,15 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se
 			for _, b := range agg.GetBuckets() {
 				if existing, ok := mergedAggregations[field][b.GetKey()]; ok {
 					existing.Count += b.GetCount()
+					// union child buckets per sub-aggregation so counts stay
+					// right when a key spans multiple spaces
+					existing.SubAggregations = mergeSubAggregations(existing.GetSubAggregations(), b.GetSubAggregations())
 					continue
 				}
 				mergedAggregations[field][b.GetKey()] = &searchsvc.Bucket{
-					Key:   b.GetKey(),
-					Count: b.GetCount(),
+					Key:             b.GetKey(),
+					Count:           b.GetCount(),
+					SubAggregations: b.GetSubAggregations(),
 				}
 			}
 		}
@@ -384,6 +382,62 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se
 
 type searchmsgBucket = searchsvc.Bucket
 
+func mergeSubAggregations(a, b []*searchsvc.AggregationResult) []*searchsvc.AggregationResult {
+	if len(a) == 0 {
+		return b
+	}
+	if len(b) == 0 {
+		return a
+	}
+	byField := make(map[string]*searchsvc.AggregationResult, len(a))
+	for _, r := range a {
+		byField[r.GetField()] = r
+	}
+	for _, r := range b {
+		existing, ok := byField[r.GetField()]
+		if !ok {
+			byField[r.GetField()] = r
+			continue
+		}
+		if existing.GetMetric() != nil || r.GetMetric() != nil {
+			// Metric result: apply the kind's reducer; prefer existing's kind.
+			if existing.GetMetric() == nil {
+				existing.Metric = &searchsvc.Metric{}
+			}
+			em := existing.Metric
+			if em.GetKind() == searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED {
+				em.Kind = r.GetMetric().GetKind()
+			}
+			rm := r.GetMetric()
+			if em.GetKind() == searchsvc.MetricKind_METRIC_KIND_AVG {
+				em.Sum += rm.GetSum()
+				em.Count += rm.GetCount()
+			} else {
+				em.Value = reduceMetric(em.GetKind(), em.GetValue(), rm.GetValue())
+			}
+			continue
+		}
+		byKey := make(map[string]*searchsvc.Bucket, len(existing.Buckets))
+		for _, bk := range existing.Buckets {
+			byKey[bk.GetKey()] = bk
+		}
+		for _, bk := range r.GetBuckets() {
+			if prev, ok := byKey[bk.GetKey()]; ok {
+				prev.Count += bk.GetCount()
+				prev.SubAggregations = mergeSubAggregations(prev.GetSubAggregations(), bk.GetSubAggregations())
+			} else {
+				existing.Buckets = append(existing.Buckets, bk)
+				byKey[bk.GetKey()] = bk
+			}
+		}
+	}
+	out := make([]*searchsvc.AggregationResult, 0, len(byField))
+	for _, r := range byField {
+		out = append(out, r)
+	}
+	return out
+}
+
 // reduceMetric is only called when both sides carry a value.
 func reduceMetric(kind searchsvc.MetricKind, a, b float64) float64 {
 	switch kind {