diff --git a/services/search/pkg/opensearch/backend.go b/services/search/pkg/opensearch/backend.go index a7f28bbdfa..59bc4f6319 100644 --- a/services/search/pkg/opensearch/backend.go +++ b/services/search/pkg/opensearch/backend.go @@ -16,6 +16,7 @@ import ( "github.com/opencloud-eu/opencloud/pkg/log" searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0" searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/aggs" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/convert" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/osu" "github.com/opencloud-eu/opencloud/services/search/pkg/search" @@ -124,6 +125,7 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ }, }, }, + Aggs: aggs.Build(sir.GetAggregations()), }, ) if err != nil { @@ -157,9 +159,15 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ matches = append(matches, match) } + aggResults, err := aggs.Parse(resp.Aggregations, sir.GetAggregations()) + if err != nil { + return nil, fmt.Errorf("failed to parse aggregations: %w", err) + } + return &searchService.SearchIndexResponse{ Matches: matches, TotalMatches: int32(totalMatches), + Aggregations: aggResults, }, nil } diff --git a/services/search/pkg/opensearch/internal/aggs/aggs.go b/services/search/pkg/opensearch/internal/aggs/aggs.go new file mode 100644 index 0000000000..5c3a8f88f3 --- /dev/null +++ b/services/search/pkg/opensearch/internal/aggs/aggs.go @@ -0,0 +1,258 @@ +// Package aggs translates proto aggregation options into the OpenSearch +// aggregation DSL and parses the response. Internal subpackage so its unit +// tests skip the parent package's Docker OpenSearch container. +package aggs + +import ( + "encoding/json" + "fmt" + "strconv" + + searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" +) + +// DefaultFacetSize matches the bleve backend: pull a generous bucket count per +// 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, metric, nested). Entries get an index-derived name so repeated +// aggs on one field don't collide. +func Build(opts []*searchsvc.AggregationOption) map[string]any { + return buildLevel(opts, "a") +} + +func buildLevel(opts []*searchsvc.AggregationOption, prefix string) map[string]any { + if len(opts) == 0 { + return nil + } + aggs := map[string]any{} + for i, opt := range opts { + name := fmt.Sprintf("%s_%d", prefix, i) + if entry := buildOne(opt, name); entry != nil { + aggs[name] = entry + } + } + if len(aggs) == 0 { + return nil + } + return aggs +} + +func buildOne(opt *searchsvc.AggregationOption, name string) map[string]any { + field := opt.GetField() + if mk := opt.GetMetricKind(); mk != searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED { + return buildMetric(field, mk) + } + var entry map[string]any + if ranges := rangesOf(opt); len(ranges) > 0 { + entry = map[string]any{ + "range": map[string]any{ + "field": field, + "ranges": buildRanges(ranges), + }, + } + } else { + size := int(opt.GetSize()) + if size <= 0 { + size = DefaultFacetSize + } + entry = map[string]any{ + "terms": map[string]any{ + "field": field, + "size": size, + }, + } + } + if subs := opt.GetSubAggregations(); len(subs) > 0 { + if nested := buildLevel(subs, name); nested != nil { + entry["aggs"] = nested + } + } + return entry +} + +// 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 { + return nil + } + return bd.GetRanges() +} + +func buildRanges(ranges []*searchsvc.BucketRange) []map[string]any { + out := make([]map[string]any, 0, len(ranges)) + for _, r := range ranges { + entry := map[string]any{ + "key": RangeKey(r), + } + if v, err := strconv.ParseFloat(r.GetFrom(), 64); err == nil { + entry["from"] = v + } + if v, err := strconv.ParseFloat(r.GetTo(), 64); err == nil { + entry["to"] = v + } + out = append(out, entry) + } + return out +} + +// RangeKey mirrors the bleve backend so cross-space merging keys match. +func RangeKey(r *searchsvc.BucketRange) string { + return r.GetFrom() + "-" + r.GetTo() +} + +// Parse converts the response aggregations block into proto results, preserving +// request order and recursing into sub-aggregations. Empty input yields +// (nil, nil); invalid JSON yields an error. +func Parse(raw json.RawMessage, opts []*searchsvc.AggregationOption) ([]*searchsvc.AggregationResult, error) { + if len(raw) == 0 || len(opts) == 0 { + return nil, nil + } + node, err := parseNode(raw) + if err != nil { + return nil, err + } + return parseLevel(node, opts, "a"), nil +} + +// aggNode is a lazily-decoded cursor over one level of the aggs response. +type aggNode map[string]json.RawMessage + +func parseNode(raw json.RawMessage) (aggNode, error) { + var m aggNode + if err := json.Unmarshal(raw, &m); err != nil { + return nil, fmt.Errorf("decode opensearch aggregations: %w", err) + } + return m, nil +} + +func parseLevel(node aggNode, opts []*searchsvc.AggregationOption, prefix string) []*searchsvc.AggregationResult { + out := make([]*searchsvc.AggregationResult, 0, len(opts)) + for i, opt := range opts { + name := fmt.Sprintf("%s_%d", prefix, i) + raw, ok := node[name] + if !ok { + continue + } + if res := parseOne(raw, opt, name); res != nil { + out = append(out, res) + } + } + return out +} + +func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption, name string) *searchsvc.AggregationResult { + field := opt.GetField() + if mk := opt.GetMetricKind(); mk != searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED { + return parseMetric(raw, field, mk) + } + var body struct { + Buckets []json.RawMessage `json:"buckets"` + } + if err := json.Unmarshal(raw, &body); err != nil { + return nil + } + buckets := make([]*searchsvc.Bucket, 0, len(body.Buckets)) + for _, b := range body.Buckets { + if bucket := parseBucket(b, opt.GetSubAggregations(), name); bucket != nil { + buckets = append(buckets, bucket) + } + } + return &searchsvc.AggregationResult{ + Field: field, + Buckets: buckets, + } +} + +func parseBucket(raw json.RawMessage, subs []*searchsvc.AggregationOption, prefix string) *searchsvc.Bucket { + var head struct { + Key any `json:"key"` + DocCount int64 `json:"doc_count"` + } + if err := json.Unmarshal(raw, &head); err != nil { + return nil + } + b := &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 +} + +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, + MetricKind: kind, + } + if body.Value != nil { + res.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, + Sum: body.Sum, + Count: body.Count, + MetricKind: kind, + } + } + 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: + return x + case float64: + // format without trailing zeros so keys match filter values + return strconv.FormatFloat(x, 'f', -1, 64) + case bool: + return strconv.FormatBool(x) + case nil: + return "" + default: + return "" + } +} diff --git a/services/search/pkg/opensearch/internal/aggs/aggs_suite_test.go b/services/search/pkg/opensearch/internal/aggs/aggs_suite_test.go new file mode 100644 index 0000000000..11337ad5d3 --- /dev/null +++ b/services/search/pkg/opensearch/internal/aggs/aggs_suite_test.go @@ -0,0 +1,13 @@ +package aggs_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestAggs(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Aggs Suite") +} diff --git a/services/search/pkg/opensearch/internal/aggs/aggs_test.go b/services/search/pkg/opensearch/internal/aggs/aggs_test.go new file mode 100644 index 0000000000..a1f080411d --- /dev/null +++ b/services/search/pkg/opensearch/internal/aggs/aggs_test.go @@ -0,0 +1,227 @@ +package aggs_test + +import ( + "encoding/json" + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/aggs" +) + +var _ = Describe("Build", func() { + It("builds a terms aggregation", func() { + res := aggs.Build([]*searchsvc.AggregationOption{ + {Field: "audio.artist", Size: 10}, + }) + Expect(res).ToNot(BeNil()) + entry, ok := res["a_0"].(map[string]any) + Expect(ok).To(BeTrue()) + terms, ok := entry["terms"].(map[string]any) + Expect(ok).To(BeTrue()) + Expect(terms["field"]).To(Equal("audio.artist")) + Expect(terms["size"]).To(Equal(10)) + }) + + It("builds a range aggregation with open-ended bounds", func() { + res := aggs.Build([]*searchsvc.AggregationOption{{ + Field: "audio.year", + BucketDefinition: &searchsvc.BucketDefinition{ + Ranges: []*searchsvc.BucketRange{ + {From: "1970", To: "1980"}, + {To: "1970"}, + {From: "2020"}, + }, + }, + }}) + r := res["a_0"].(map[string]any)["range"].(map[string]any) + Expect(r["field"]).To(Equal("audio.year")) + ranges := r["ranges"].([]map[string]any) + Expect(ranges).To(HaveLen(3)) + Expect(ranges[0]).To(SatisfyAll( + HaveKeyWithValue("key", "1970-1980"), + HaveKeyWithValue("from", 1970.0), + HaveKeyWithValue("to", 1980.0), + )) + Expect(ranges[1]).ToNot(HaveKey("from")) // open lower bound + Expect(ranges[2]).ToNot(HaveKey("to")) // open upper bound + }) + + DescribeTable("builds single-value metric aggregations", + func(kind searchsvc.MetricKind, esKind string) { + res := aggs.Build([]*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricKind: 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 := aggs.Build([]*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricKind: 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")) + }) + + It("nests sub-aggregations under their parent bucket", func() { + res := aggs.Build([]*searchsvc.AggregationOption{{ + Field: "audio.artist", Size: 5, + SubAggregations: []*searchsvc.AggregationOption{{ + Field: "audio.album", Size: 7, + SubAggregations: []*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricKind: 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() { + It("parses flat term and range buckets, stringifying numeric keys", func() { + raw := json.RawMessage(`{ + "a_0": {"buckets": [ + {"key": "Pink Floyd", "doc_count": 42}, + {"key": "Motörhead", "doc_count": 35} + ]}, + "a_1": {"buckets": [ + {"key": "1970-1980", "from": 1970.0, "to": 1980.0, "doc_count": 12} + ]}, + "a_2": {"buckets": [ + {"key": 9, "doc_count": 3} + ]} + }`) + out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{ + {Field: "audio.artist"}, + {Field: "audio.year"}, + {Field: "audio.track"}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(3)) + + Expect(out[0].Field).To(Equal("audio.artist")) + Expect(out[0].Buckets).To(HaveLen(2)) + Expect(out[0].Buckets[0].Key).To(Equal("Pink Floyd")) + Expect(out[0].Buckets[0].Count).To(Equal(int64(42))) + + Expect(out[1].Buckets[0].Key).To(Equal("1970-1980")) + Expect(out[1].Buckets[0].Count).To(Equal(int64(12))) + + // numeric term key stringified without trailing zeros + 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", MetricKind: 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.MetricKind).To(Equal(searchsvc.MetricKind_METRIC_KIND_SUM)) + Expect(metric.Value).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)) + out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{ + {Field: "audio.duration", MetricKind: kind}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(1)) + Expect(out[0].MetricKind).To(Equal(kind)) + Expect(out[0].Value).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", MetricKind: searchsvc.MetricKind_METRIC_KIND_SUM}}) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(1)) + Expect(out[0].Value).To(BeZero()) + Expect(out[0].MetricKind).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", MetricKind: searchsvc.MetricKind_METRIC_KIND_AVG}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(1)) + Expect(out[0].MetricKind).To(Equal(searchsvc.MetricKind_METRIC_KIND_AVG)) + Expect(out[0].Sum).To(Equal(24500000.0)) + Expect(out[0].Count).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/opensearch/internal/osu/request.go b/services/search/pkg/opensearch/internal/osu/request.go index 927da7ac54..f6f4e238ee 100644 --- a/services/search/pkg/opensearch/internal/osu/request.go +++ b/services/search/pkg/opensearch/internal/osu/request.go @@ -95,6 +95,7 @@ func BuildSearchReq(req *opensearchgoAPI.SearchReq, q Builder, p ...SearchBodyPa type SearchBodyParams struct { Highlight *BodyParamHighlight `json:"highlight,omitempty"` + Aggs map[string]any `json:"aggs,omitempty"` } //----------------------------------------------------------------------------//