diff --git a/services/search/pkg/bleve/aggregations.go b/services/search/pkg/bleve/aggregations.go index 9e39ef7611..4d92b02b63 100644 --- a/services/search/pkg/bleve/aggregations.go +++ b/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 { diff --git a/services/search/pkg/opensearch/internal/aggs/aggs.go b/services/search/pkg/opensearch/internal/aggs/aggs.go index c4beb41cd9..64b898c012 100644 --- a/services/search/pkg/opensearch/internal/aggs/aggs.go +++ b/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 } diff --git a/services/search/pkg/opensearch/internal/aggs/aggs_test.go b/services/search/pkg/opensearch/internal/aggs/aggs_test.go index 6e4e9a27b2..b049656b89 100644 --- a/services/search/pkg/opensearch/internal/aggs/aggs_test.go +++ 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)) diff --git a/services/search/pkg/parity/README.md b/services/search/pkg/parity/README.md index ddf01ec340..ef943e45df 100644 --- a/services/search/pkg/parity/README.md +++ b/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 | ✅ | diff --git a/services/search/pkg/parity/aggregations_test.go b/services/search/pkg/parity/aggregations_test.go index 4bac338058..76f107aaa2 100644 --- a/services/search/pkg/parity/aggregations_test.go +++ b/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 " of 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] diff --git a/services/search/pkg/search/service.go b/services/search/pkg/search/service.go index fcf8bdb774..b7f985b622 100644 --- a/services/search/pkg/search/service.go +++ b/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 {