From ec40f05943d603ccc2863eae9665cd1499bf9d3f Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Sat, 12 Sep 2026 16:33:47 +0000 Subject: [PATCH] feat(search): terms aggregations Term buckets on both engines: bleve facets, OpenSearch terms aggregations, buckets merged across spaces and post-processed per BucketDefinition (minimum count, sort, size). Terms aggregations on numeric fields are rejected at the graph endpoint. Pinned in the parity suite as AGG-01 to AGG-03. Ranges, metrics and sub-aggregations are declined until the engines evaluate them. --- services/graph/pkg/service/v0/searchquery.go | 26 +++ .../graph/pkg/service/v0/searchquery_test.go | 18 ++ services/search/pkg/bleve/aggregations.go | 52 ++++++ services/search/pkg/bleve/backend.go | 9 + services/search/pkg/opensearch/backend.go | 13 ++ .../pkg/opensearch/internal/aggs/aggs.go | 147 +++++++++++++++ .../internal/aggs/aggs_suite_test.go | 13 ++ .../pkg/opensearch/internal/aggs/aggs_test.go | 61 +++++++ .../pkg/opensearch/internal/osu/request.go | 1 + services/search/pkg/parity/README.md | 24 +++ .../search/pkg/parity/aggregations_test.go | 168 ++++++++++++++++++ services/search/pkg/parity/fixtures_test.go | 4 + services/search/pkg/parity/matrix_test.go | 4 + services/search/pkg/search/fields.go | 71 ++++++++ services/search/pkg/search/fields_test.go | 46 +++++ services/search/pkg/search/service.go | 94 ++++++++++ 16 files changed, 751 insertions(+) create mode 100644 services/search/pkg/bleve/aggregations.go create mode 100644 services/search/pkg/opensearch/internal/aggs/aggs.go create mode 100644 services/search/pkg/opensearch/internal/aggs/aggs_suite_test.go create mode 100644 services/search/pkg/opensearch/internal/aggs/aggs_test.go create mode 100644 services/search/pkg/parity/aggregations_test.go create mode 100644 services/search/pkg/search/fields.go create mode 100644 services/search/pkg/search/fields_test.go diff --git a/services/graph/pkg/service/v0/searchquery.go b/services/graph/pkg/service/v0/searchquery.go index b407948cf3..21f7b1fe56 100644 --- a/services/graph/pkg/service/v0/searchquery.go +++ b/services/graph/pkg/service/v0/searchquery.go @@ -2,6 +2,7 @@ package svc import ( "context" + "fmt" "net/http" "path" "slices" @@ -18,6 +19,7 @@ import ( searchmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0" searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" "github.com/opencloud-eu/opencloud/services/graph/pkg/errorcode" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) // SearchQuery runs the search requests and returns results grouped by request @@ -35,6 +37,13 @@ func (g Graph) SearchQuery(w http.ResponseWriter, r *http.Request) { return } + for _, sr := range req.Requests { + if err := validateAggregations(sr.Aggregations); err != nil { + errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, err.Error()) + return + } + } + th := r.Header.Get(revaCtx.TokenHeader) ctx := revaCtx.ContextSetToken(r.Context(), th) ctx = metadata.Set(ctx, revaCtx.TokenHeader, th) @@ -138,6 +147,23 @@ func clampPagination(fromP, sizeP *int32) (int32, int32) { return from, size } +// validateAggregations rejects terms aggregations on numeric/time fields: bleve +// indexes them as prefix-coded binary, so term buckets are meaningless. +// Classification via search.IsNumericField. +func validateAggregations(aggs []libregraph.AggregationOption) error { + for _, a := range aggs { + if !search.IsNumericField(a.Field) { + continue + } + if a.LibreGraphMetricDefinition != nil { + // metrics reduce numeric values, no term buckets involved + continue + } + return fmt.Errorf("terms aggregation is not supported on numeric field %q", a.Field) + } + return nil +} + func libregraphAggregationsToSearch(in []libregraph.AggregationOption) []*searchsvc.AggregationOption { if len(in) == 0 { return nil diff --git a/services/graph/pkg/service/v0/searchquery_test.go b/services/graph/pkg/service/v0/searchquery_test.go index 11eba9e021..3acff95db3 100644 --- a/services/graph/pkg/service/v0/searchquery_test.go +++ b/services/graph/pkg/service/v0/searchquery_test.go @@ -197,4 +197,22 @@ var _ = ginkgo.Describe("SearchQuery", func() { ginkgo.Entry("oversized size clamps to max", int32Ptr(0), int32Ptr(1000), int32(0), int32(500)), ginkgo.Entry("from+size overflow collapses", int32Ptr(1<<31-1), int32Ptr(500), int32(1<<31-1-500), int32(500)), ) + + ginkgo.It("rejects a terms aggregation on a numeric field with 400", func() { + g := graphWithSearch(stubSearchService{ + search: func(*searchsvc.SearchRequest) (*searchsvc.SearchResponse, error) { + ginkgo.Fail("search service must not be called when validation fails") + return nil, nil + }, + }) + rr := postSearchQuery(g, `{ + "requests": [{ + "entityTypes": ["driveItem"], + "query": {"queryString": "mediatype:audio"}, + "size": 0, + "aggregations": [{"field": "audio.year"}] + }] + }`) + Expect(rr.Code).To(Equal(http.StatusBadRequest), rr.Body.String()) + }) }) diff --git a/services/search/pkg/bleve/aggregations.go b/services/search/pkg/bleve/aggregations.go new file mode 100644 index 0000000000..c16172747c --- /dev/null +++ b/services/search/pkg/bleve/aggregations.go @@ -0,0 +1,52 @@ +package bleve + +import ( + "github.com/blevesearch/bleve/v2" + bleveSearch "github.com/blevesearch/bleve/v2/search" + + searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" +) + +// 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.GetMetricKind() != searchService.MetricKind_METRIC_KIND_UNSPECIFIED || len(agg.GetSubAggregations()) > 0 +} + +func newBleveFacetRequest(agg *searchService.AggregationOption) *bleve.FacetRequest { + size := int(agg.GetSize()) + if size <= 0 { + size = defaultFacetSize + } + return bleve.NewFacetRequest(agg.GetField(), size) +} + +func facetBuckets(fr *bleveSearch.FacetResult) []*searchService.Bucket { + buckets := make([]*searchService.Bucket, 0) + for _, t := range fr.Terms.Terms() { + buckets = append(buckets, &searchService.Bucket{Key: t.Term, Count: int64(t.Count)}) + } + return buckets +} + +func extractBleveAggregations(res *bleve.SearchResult, aggs []*searchService.AggregationOption) []*searchService.AggregationResult { + if len(aggs) == 0 { + return nil + } + out := make([]*searchService.AggregationResult, 0, len(aggs)) + for _, agg := range aggs { + fr, ok := res.Facets[agg.GetField()] + if !ok { + continue + } + out = append(out, &searchService.AggregationResult{ + Field: agg.GetField(), + Buckets: facetBuckets(fr), + }) + } + return out +} diff --git a/services/search/pkg/bleve/backend.go b/services/search/pkg/bleve/backend.go index d8426e5435..878870f78a 100644 --- a/services/search/pkg/bleve/backend.go +++ b/services/search/pkg/bleve/backend.go @@ -2,6 +2,7 @@ package bleve import ( "context" + "errors" "math" "time" @@ -94,6 +95,13 @@ func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexReques bleveReq.Size = int(sir.PageSize) } + for _, agg := range sir.GetAggregations() { + if collected(agg) { + return nil, errors.New("metric and nested aggregations are not supported by bleve yet") + } + bleveReq.AddFacet(agg.GetField(), newBleveFacetRequest(agg)) + } + bleveReq.Fields = []string{"*"} res, err := b.index.Search(bleveReq) if err != nil { @@ -151,6 +159,7 @@ func (b *Backend) Search(_ context.Context, sir *searchService.SearchIndexReques return &searchService.SearchIndexResponse{ Matches: matches, TotalMatches: int32(totalMatches), + Aggregations: extractBleveAggregations(res, sir.GetAggregations()), }, nil } diff --git a/services/search/pkg/opensearch/backend.go b/services/search/pkg/opensearch/backend.go index 330dcd577d..a7f43d392a 100644 --- a/services/search/pkg/opensearch/backend.go +++ b/services/search/pkg/opensearch/backend.go @@ -17,6 +17,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" @@ -122,6 +123,11 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ searchParams.Size = conversions.ToPointer(int(sir.PageSize)) } + builtAggs, err := aggs.Build(sir.GetAggregations()) + if err != nil { + return nil, err + } + req, err := osu.BuildSearchReq(&opensearchgoAPI.SearchReq{ Indices: []string{b.index}, Params: searchParams, @@ -140,6 +146,7 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ }, }, }, + Aggs: builtAggs, }, ) if err != nil { @@ -162,9 +169,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..8bc5cad522 --- /dev/null +++ b/services/search/pkg/opensearch/internal/aggs/aggs.go @@ -0,0 +1,147 @@ +// 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). 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) { + return buildLevel(opts, "a") +} + +func buildLevel(opts []*searchsvc.AggregationOption, prefix string) (map[string]any, error) { + if len(opts) == 0 { + return nil, nil + } + aggs := map[string]any{} + for i, opt := range opts { + name := fmt.Sprintf("%s_%d", prefix, i) + entry, err := buildOne(opt) + if err != nil { + return nil, err + } + if entry != nil { + aggs[name] = entry + } + } + if len(aggs) == 0 { + return nil, nil + } + return aggs, nil +} + +func buildOne(opt *searchsvc.AggregationOption) (map[string]any, error) { + field := opt.GetField() + size := int(opt.GetSize()) + if size <= 0 { + size = DefaultFacetSize + } + return map[string]any{ + "terms": map[string]any{ + "field": field, + "size": size, + }, + }, nil +} + +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); res != nil { + out = append(out, res) + } + } + return out +} + +func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption) *searchsvc.AggregationResult { + field := opt.GetField() + 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); bucket != nil { + buckets = append(buckets, bucket) + } + } + return &searchsvc.AggregationResult{ + Field: field, + Buckets: buckets, + } +} + +func parseBucket(raw json.RawMessage) *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, + } + return b +} + +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..298a41d400 --- /dev/null +++ b/services/search/pkg/opensearch/internal/aggs/aggs_test.go @@ -0,0 +1,61 @@ +package aggs_test + +import ( + "encoding/json" + + . "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() { + build := func(opts []*searchsvc.AggregationOption) map[string]any { + res, err := aggs.Build(opts) + Expect(err).ToNot(HaveOccurred()) + return res + } + + It("builds a terms aggregation", func() { + res := 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)) + }) + +}) + +var _ = Describe("Parse", func() { + It("parses term 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": 9, "doc_count": 3} + ]} + }`) + out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{ + {Field: "audio.artist"}, + {Field: "audio.track"}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(out).To(HaveLen(2)) + + 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))) + + // numeric term key stringified without trailing zeros + Expect(out[1].Buckets[0].Key).To(Equal("9")) + }) +}) 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"` } //----------------------------------------------------------------------------// diff --git a/services/search/pkg/parity/README.md b/services/search/pkg/parity/README.md index 6558e96847..1d7f0754d1 100644 --- a/services/search/pkg/parity/README.md +++ b/services/search/pkg/parity/README.md @@ -703,3 +703,27 @@ Fixtures: | METADATA-01 | `*song*` reads `Audio` | all 16 fields unchanged | all 16 fields unchanged | all 16 fields unchanged | ✅ | | METADATA-02 | `*team*` reads `Location` | all 3 fields unchanged | all 3 fields unchanged | all 3 fields unchanged | ✅ | | METADATA-03 | `*team*` reads `Audio` | none | none | none | ✅ | + +## Aggregations + +### aggregations + +Fixtures: + +- `a.mp3`, MimeType = audio/mpeg +- `b.mp3`, MimeType = audio/mpeg +- `c.mp3`, MimeType = audio/mpeg +- `d.mp3`, MimeType = audio/mpeg +- `e.mp3`, MimeType = audio/mpeg +- `f.mp3`, MimeType = audio/mpeg +- `g.mp3`, MimeType = audio/mpeg +- `a.jpg`, MimeType = image/jpeg +- `b.jpg`, MimeType = image/jpeg +- `c.jpg`, MimeType = image/jpeg +- `d.jpg`, MimeType = image/jpeg + +| Case | Query | expected | bleve | OpenSearch | same? | +|---|---|---|---|---|---| +| AGG-01 | `mediatype:audio` reads `term buckets on audio.artist` | audio.arti... Floyd=2, audio.arti...¶rhead=3 | audio.arti... Floyd=2, audio.arti...¶rhead=3 | audio.arti... Floyd=2, audio.arti...¶rhead=3 | ✅ | +| AGG-02 | `mediatype:audio` reads `no aggregations requested` | no match | no match | no match | ✅ | +| AGG-03 | `mediatype:audio` reads `artist and album buckets in one request` | audio.albu...Spades=1, audio.album Bomber=2, audio.album The Wall=2, audio.arti... Floyd=2, audio.arti...¶rhead=3 | audio.albu...Spades=1, audio.album Bomber=2, audio.album The Wall=2, audio.arti... Floyd=2, audio.arti...¶rhead=3 | audio.albu...Spades=1, audio.album Bomber=2, audio.album The Wall=2, audio.arti... Floyd=2, audio.arti...¶rhead=3 | ✅ | diff --git a/services/search/pkg/parity/aggregations_test.go b/services/search/pkg/parity/aggregations_test.go new file mode 100644 index 0000000000..4edeaac093 --- /dev/null +++ b/services/search/pkg/parity/aggregations_test.go @@ -0,0 +1,168 @@ +package parity + +import ( + "context" + "fmt" + "strings" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + libregraph "github.com/opencloud-eu/libre-graph-api-go" + + searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +// aggCase is one aggregation request both engines have to answer alike. The +// answer is rendered to strings (one per non-empty bucket or metric), so the +// matrix machinery can carry it like any query answer. +type aggCase struct { + id int + query string + aggs []*searchService.AggregationOption + reads string + want []string + wantError bool + engineOverrides map[string]override +} + +func (c aggCase) label() string { return fmt.Sprintf("AGG-%02d", c.id) } + +func withYear(name string, year int32) search.Resource { + return fixtureDoc(name, withMime("audio/mpeg"), withAudio(&libregraph.Audio{Year: libregraph.PtrInt32(year)})) +} + +func withTaken(name, taken string) search.Resource { + t, err := time.Parse(time.RFC3339, taken) + if err != nil { + panic(err) + } + return fixtureDoc(name, withMime("image/jpeg"), withPhoto(&libregraph.Photo{TakenDateTime: &t})) +} + +func song(name, artist, album string, year int32) search.Resource { + return fixtureDoc(name, withMime("audio/mpeg"), withAudio(&libregraph.Audio{ + Artist: libregraph.PtrString(artist), + Album: libregraph.PtrString(album), + Year: libregraph.PtrInt32(year), + })) +} + +func aggregationFixtures() []search.Resource { + return []search.Resource{ + // years: 1971, 1975, 1982, 1999, 2001, 2005, 2009 + song("a.mp3", "Pink Floyd", "The Wall", 1971), + song("b.mp3", "Pink Floyd", "The Wall", 1975), + song("c.mp3", "Motörhead", "Bomber", 1982), + song("d.mp3", "Motörhead", "Bomber", 1999), + song("e.mp3", "Motörhead", "Ace of Spades", 2001), + withYear("f.mp3", 2005), + withYear("g.mp3", 2009), + withTaken("a.jpg", "2018-08-11T09:15:00Z"), + withTaken("b.jpg", "2018-08-11T19:42:00Z"), + withTaken("c.jpg", "2018-09-01T12:00:00Z"), + withTaken("d.jpg", "2021-08-11T08:00:00Z"), + } +} + +func aggregationCases() []aggCase { + return []aggCase{ + {id: 1, query: "mediatype:audio", reads: "term buckets on audio.artist", + aggs: []*searchService.AggregationOption{{Field: "audio.artist", Size: 10}}, + want: []string{"audio.artist Pink Floyd=2", "audio.artist Motörhead=3"}}, + {id: 2, query: "mediatype:audio", reads: "no aggregations requested"}, + {id: 3, query: "mediatype:audio", reads: "artist and album buckets in one request", + aggs: []*searchService.AggregationOption{{Field: "audio.artist"}, {Field: "audio.album"}}, + want: []string{ + "audio.artist Pink Floyd=2", "audio.artist Motörhead=3", + "audio.album The Wall=2", "audio.album Bomber=2", "audio.album Ace of Spades=1", + }}, + } +} + +// renderAggregations flattens an answer into comparable strings; buckets with +// no hits are dropped, the engines differ in whether they emit them at all. +// A nested result renders under its bucket, joined by " / ". +func renderAggregations(resp *searchService.SearchIndexResponse, err error) []string { + if err != nil { + return []string{"error"} + } + + out := []string{} + for _, a := range resp.Aggregations { + out = append(out, renderAggregation("", a)...) + } + + return out +} + +func renderAggregation(prefix string, a *searchService.AggregationResult) []string { + if a.MetricKind != searchService.MetricKind_METRIC_KIND_UNSPECIFIED { + kind := strings.ToLower(strings.TrimPrefix(a.MetricKind.String(), "METRIC_KIND_")) + if a.MetricKind == searchService.MetricKind_METRIC_KIND_AVG { + return []string{prefix + fmt.Sprintf("%s avg sum=%v count=%d", a.Field, a.Sum, a.Count)} + } + return []string{prefix + fmt.Sprintf("%s %s=%v", a.Field, kind, a.Value)} + } + + out := []string{} + for _, b := range a.Buckets { + if b.Count == 0 { + continue + } + line := prefix + fmt.Sprintf("%s %s=%d", a.Field, b.Key, b.Count) + out = append(out, line) + for _, sub := range b.SubAggregations { + out = append(out, renderAggregation(line+" / ", sub)...) + } + } + + return out +} + +var _ = Describe("Aggregations", func() { + Describe("aggregations", Ordered, ContinueOnFailure, func() { + var engines []testEngine + + BeforeAll(func() { + engines = newEngines("opencloud-test-engine-parity-aggregations", aggregationFixtures()) + }) + + for caseAt, c := range aggregationCases() { + row := matrixRow{ + Section: "Aggregations", Group: "aggregations", ID: c.label(), + Query: c.query, Reads: c.reads, + Want: c.want, Overrides: renderOverrides(c.engineOverrides), + GroupAt: 100, CaseAt: caseAt, + } + planRow(row) + + Describe(c.label()+" "+c.reads, func() { + for _, name := range engineNames { + It("on "+name, func() { + e := engineNamed(engines, name) + if e.unavailable != "" { + recordSkip(row, name) + Skip(e.unavailable) + } + + resp, err := e.backend.Search(context.Background(), &searchService.SearchIndexRequest{ + Query: c.query, + Aggregations: c.aggs, + }) + answer := renderAggregations(resp, err) + recordAnswer(row, name, answer) + + _, overridden := c.engineOverrides[name] + if !overridden && !c.wantError { + Expect(err).NotTo(HaveOccurred(), "the aggregation has to answer") + } + + expectAnswer(name, answer, override{want: c.want}, c.engineOverrides) + }) + } + }) + } + }) +}) diff --git a/services/search/pkg/parity/fixtures_test.go b/services/search/pkg/parity/fixtures_test.go index 330fddc3dd..a9f564bc69 100644 --- a/services/search/pkg/parity/fixtures_test.go +++ b/services/search/pkg/parity/fixtures_test.go @@ -50,6 +50,10 @@ func withAudio(audio *libregraph.Audio) fixtureOption { return func(r *search.Resource) { r.Audio = audio } } +func withPhoto(photo *libregraph.Photo) fixtureOption { + return func(r *search.Resource) { r.Photo = photo } +} + func withLocation(location *libregraph.GeoCoordinates) fixtureOption { return func(r *search.Resource) { r.Location = location } } diff --git a/services/search/pkg/parity/matrix_test.go b/services/search/pkg/parity/matrix_test.go index a1d51394d7..1bde6ca2b9 100644 --- a/services/search/pkg/parity/matrix_test.go +++ b/services/search/pkg/parity/matrix_test.go @@ -278,6 +278,10 @@ func matrixFixtures(group string) string { fixtures = g.fixtures } + if group == "aggregations" { + fixtures = aggregationFixtures() + } + if len(fixtures) == 0 { return "Fixtures: none" } diff --git a/services/search/pkg/search/fields.go b/services/search/pkg/search/fields.go new file mode 100644 index 0000000000..adb81bc679 --- /dev/null +++ b/services/search/pkg/search/fields.go @@ -0,0 +1,71 @@ +package search + +import ( + "reflect" + "strings" + "time" +) + +// IsNumericField reports whether the indexed field at the dotted path holds a +// numeric or time value. Terms aggregations on those are rejected: bleve stores +// them as prefix-coded binary, so term buckets are meaningless. The set is built +// by walking the Resource type, so new facet fields are picked up automatically. +func IsNumericField(dottedPath string) bool { + return numericFields[dottedPath] +} + +var numericFields = buildNumericFieldSet() + +var timeType = reflect.TypeOf(time.Time{}) + +func buildNumericFieldSet() map[string]bool { + out := map[string]bool{} + walkStruct(out, "", reflect.TypeOf(Resource{})) + return out +} + +func walkStruct(out map[string]bool, prefix string, t reflect.Type) { + for t.Kind() == reflect.Ptr { + t = t.Elem() + } + if t.Kind() != reflect.Struct { + return + } + for i := 0; i < t.NumField(); i++ { + f := t.Field(i) + if !f.IsExported() { + continue + } + if f.Anonymous { + // embedded: promote fields into the current prefix, like encoding/json. + walkStruct(out, prefix, f.Type) + continue + } + path := prefix + jsonFieldName(f) + ft := f.Type + for ft.Kind() == reflect.Ptr { + ft = ft.Elem() + } + switch ft.Kind() { + case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, + reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, + reflect.Float32, reflect.Float64: + out[path] = true + case reflect.Struct: + if ft == timeType { + // time.Time round-trips as RFC3339; treat as numeric. + out[path] = true + continue + } + walkStruct(out, path+".", ft) + } + } +} + +func jsonFieldName(f reflect.StructField) string { + tag := f.Tag.Get("json") + if tag == "" { + return f.Name + } + return strings.Split(tag, ",")[0] +} diff --git a/services/search/pkg/search/fields_test.go b/services/search/pkg/search/fields_test.go new file mode 100644 index 0000000000..f198c40e49 --- /dev/null +++ b/services/search/pkg/search/fields_test.go @@ -0,0 +1,46 @@ +package search_test + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +var _ = Describe("IsNumericField", func() { + DescribeTable("reports whether a field maps to a numeric type", + func(field string, numeric bool) { + Expect(search.IsNumericField(field)).To(Equal(numeric)) + }, + // top-level numeric fields on Resource / Document + Entry("Size (uint64 via embedded Document)", "Size", true), + Entry("Type (uint64 on Resource)", "Type", true), + // top-level string fields + Entry("Name", "Name", false), + Entry("Path", "Path", false), + Entry("MimeType", "MimeType", false), + // nested audio + Entry("audio.artist", "audio.artist", false), + Entry("audio.album", "audio.album", false), + Entry("audio.year", "audio.year", true), + Entry("audio.bitrate", "audio.bitrate", true), + Entry("audio.track", "audio.track", true), + Entry("audio.hasDrm (bool, not numeric)", "audio.hasDrm", false), + // nested image + Entry("image.width", "image.width", true), + Entry("image.height", "image.height", true), + // nested photo + Entry("photo.cameraMake", "photo.cameraMake", false), + Entry("photo.iso", "photo.iso", true), + Entry("photo.focalLength (float32)", "photo.focalLength", true), + Entry("photo.exposureDenominator (float32)", "photo.exposureDenominator", true), + Entry("photo.takenDateTime (time.Time, treated as numeric)", "photo.takenDateTime", true), + // nested location + Entry("location.altitude", "location.altitude", true), + Entry("location.latitude", "location.latitude", true), + Entry("location.longitude", "location.longitude", true), + // unknown fields the caller may still aggregate on + Entry("nonexistent", "nonexistent", false), + Entry("audio.nonexistent", "audio.nonexistent", false), + ) +}) diff --git a/services/search/pkg/search/service.go b/services/search/pkg/search/service.go index 1b7ff47285..1e49ea808a 100644 --- a/services/search/pkg/search/service.go +++ b/services/search/pkg/search/service.go @@ -97,6 +97,16 @@ 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) { + // terms aggregations only for now: the engines do not evaluate ranges, + // metrics and sub-aggregations yet + for _, opt := range req.GetAggregations() { + switch { + case len(opt.GetBucketDefinition().GetRanges()) > 0: + return nil, errtypes.BadRequest("range aggregations are not supported yet") + case opt.GetMetricKind() != searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED || len(opt.GetSubAggregations()) > 0: + return nil, errtypes.BadRequest("metric and nested aggregations are not supported yet") + } + } s.logger.Debug().Str("query", req.Query).Msg("performing a search") // collect metrics @@ -287,6 +297,7 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se return nil, err } + mergedAggregations := map[string]map[string]*searchmsgBucket{} for _, res := range responses { if res == nil { continue @@ -295,6 +306,22 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se for _, match := range res.Matches { matches = append(matches, match) } + for _, agg := range res.GetAggregations() { + field := agg.GetField() + if _, ok := mergedAggregations[field]; !ok { + mergedAggregations[field] = map[string]*searchmsgBucket{} + } + for _, b := range agg.GetBuckets() { + if existing, ok := mergedAggregations[field][b.GetKey()]; ok { + existing.Count += b.GetCount() + continue + } + mergedAggregations[field][b.GetKey()] = &searchsvc.Bucket{ + Key: b.GetKey(), + Count: b.GetCount(), + } + } + } } // compile one sorted list of matches from all spaces and apply the limit if needed @@ -307,13 +334,80 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se matches = matches[0:limit] } + aggregations := make([]*searchsvc.AggregationResult, 0, len(req.GetAggregations())) + for _, opt := range req.GetAggregations() { + field := opt.GetField() + bucketMap := mergedAggregations[field] + buckets := make([]*searchsvc.Bucket, 0, len(bucketMap)) + for _, b := range bucketMap { + buckets = append(buckets, b) + } + aggregations = append(aggregations, &searchsvc.AggregationResult{ + Field: field, + Buckets: postProcessBuckets(buckets, opt), + }) + } + success = true return &searchsvc.SearchResponse{ Matches: matches, TotalMatches: total, + Aggregations: aggregations, }, nil } +// searchmsgBucket aliases the bucket type for the map-of-maps below. +type searchmsgBucket = searchsvc.Bucket + +// postProcessBuckets applies the BucketDefinition (minimumCount filter, sort by +// count/keyAsString/keyAsNumber, trim to Size). Defaults to count-descending. +func postProcessBuckets(buckets []*searchsvc.Bucket, opt *searchsvc.AggregationOption) []*searchsvc.Bucket { + bd := opt.GetBucketDefinition() + sortBy := "count" + desc := true + var minCount int64 + if bd != nil { + if bd.GetSortBy() != "" { + sortBy = bd.GetSortBy() + } + desc = bd.GetIsDescending() + minCount = int64(bd.GetMinimumCount()) + } + + if minCount > 0 { + filtered := buckets[:0] + for _, b := range buckets { + if b.GetCount() >= minCount { + filtered = append(filtered, b) + } + } + buckets = filtered + } + + sort.SliceStable(buckets, func(i, j int) bool { + less := false + switch sortBy { + case "keyAsString": + less = buckets[i].GetKey() < buckets[j].GetKey() + case "keyAsNumber": + iv, _ := strconv.ParseFloat(buckets[i].GetKey(), 64) + jv, _ := strconv.ParseFloat(buckets[j].GetKey(), 64) + less = iv < jv + default: // "count" + less = buckets[i].GetCount() < buckets[j].GetCount() + } + if desc { + return !less + } + return less + }) + + if size := opt.GetSize(); size > 0 && int32(len(buckets)) > size { + buckets = buckets[:size] + } + return buckets +} + func (s *Service) searchIndex(ctx context.Context, req *searchsvc.SearchRequest, space *provider.StorageSpace, mountpointID string) (*searchsvc.SearchIndexResponse, error) { if req.Ref != nil && (req.Ref.ResourceId.StorageId != space.Root.StorageId ||