diff --git a/services/search/pkg/opensearch/backend.go b/services/search/pkg/opensearch/backend.go index abfa7d55a7..a7f43d392a 100644 --- a/services/search/pkg/opensearch/backend.go +++ b/services/search/pkg/opensearch/backend.go @@ -123,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, @@ -141,7 +146,7 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ }, }, }, - Aggs: aggs.Build(sir.GetAggregations()), + Aggs: builtAggs, }, ) if err != nil { diff --git a/services/search/pkg/opensearch/internal/aggs/aggs.go b/services/search/pkg/opensearch/internal/aggs/aggs.go index 5c3a8f88f3..df8829172d 100644 --- a/services/search/pkg/opensearch/internal/aggs/aggs.go +++ b/services/search/pkg/opensearch/internal/aggs/aggs.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "strconv" + "time" searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" ) @@ -16,40 +17,49 @@ import ( 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 { +// (terms, range, date_range, metric, nested). 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 { +func buildLevel(opts []*searchsvc.AggregationOption, prefix string) (map[string]any, error) { if len(opts) == 0 { - return nil + return nil, nil } aggs := map[string]any{} for i, opt := range opts { name := fmt.Sprintf("%s_%d", prefix, i) - if entry := buildOne(opt, name); entry != nil { + entry, err := buildOne(opt, name) + if err != nil { + return nil, err + } + if entry != nil { aggs[name] = entry } } if len(aggs) == 0 { - return nil + return nil, nil } - return aggs + return aggs, nil } -func buildOne(opt *searchsvc.AggregationOption, name string) map[string]any { +func buildOne(opt *searchsvc.AggregationOption, name string) (map[string]any, error) { field := opt.GetField() if mk := opt.GetMetricKind(); mk != searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED { - return buildMetric(field, mk) + return buildMetric(field, mk), nil } var entry map[string]any if ranges := rangesOf(opt); len(ranges) > 0 { + built, kind, err := buildRanges(field, ranges) + if err != nil { + return nil, err + } entry = map[string]any{ - "range": map[string]any{ + kind: map[string]any{ "field": field, - "ranges": buildRanges(ranges), + "ranges": built, }, } } else { @@ -65,11 +75,15 @@ func buildOne(opt *searchsvc.AggregationOption, name string) map[string]any { } } if subs := opt.GetSubAggregations(); len(subs) > 0 { - if nested := buildLevel(subs, name); nested != nil { + nested, err := buildLevel(subs, name) + if err != nil { + return nil, err + } + if nested != nil { entry["aggs"] = nested } } - return entry + return entry, nil } // buildMetric emits the sum/min/max metric. AVG uses a stats agg to transport @@ -96,21 +110,64 @@ func rangesOf(opt *searchsvc.AggregationOption) []*searchsvc.BucketRange { return bd.GetRanges() } -func buildRanges(ranges []*searchsvc.BucketRange) []map[string]any { +// rangeTimeLayouts mirrors the bleve backend's accepted date bound formats. +var rangeTimeLayouts = []string{time.RFC3339, "2006-01-02"} + +func boundIsDate(s string) bool { + for _, layout := range rangeTimeLayouts { + if _, err := time.Parse(layout, s); err == nil { + return true + } + } + return false +} + +// buildRanges renders the ranges and decides between the numeric "range" and +// the "date_range" aggregation: one date-looking bound switches the whole +// aggregation to date mode, exactly like the bleve backend. +func buildRanges(field string, ranges []*searchsvc.BucketRange) ([]map[string]any, string, error) { + dates := false + for _, r := range ranges { + for _, s := range []string{r.GetFrom(), r.GetTo()} { + if s == "" { + continue + } + if _, err := strconv.ParseFloat(s, 64); err != nil { + dates = true + } + } + } + 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 + for side, s := range map[string]string{"from": r.GetFrom(), "to": r.GetTo()} { + if s == "" { + continue + } + if dates { + if !boundIsDate(s) { + return nil, "", fmt.Errorf("invalid date range bound %q on field %q", s, field) + } + entry[side] = s + continue + } + v, err := strconv.ParseFloat(s, 64) + if err != nil { + return nil, "", fmt.Errorf("invalid range bound %q on field %q", s, field) + } + entry[side] = v } out = append(out, entry) } - return out + + kind := "range" + if dates { + kind = "date_range" + } + return out, kind, nil } // RangeKey mirrors the bleve backend so cross-space merging keys match. diff --git a/services/search/pkg/opensearch/internal/aggs/aggs_test.go b/services/search/pkg/opensearch/internal/aggs/aggs_test.go index a1f080411d..59ba9b41ac 100644 --- a/services/search/pkg/opensearch/internal/aggs/aggs_test.go +++ b/services/search/pkg/opensearch/internal/aggs/aggs_test.go @@ -12,8 +12,14 @@ import ( ) 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 := aggs.Build([]*searchsvc.AggregationOption{ + res := build([]*searchsvc.AggregationOption{ {Field: "audio.artist", Size: 10}, }) Expect(res).ToNot(BeNil()) @@ -25,8 +31,43 @@ var _ = Describe("Build", func() { Expect(terms["size"]).To(Equal(10)) }) + It("builds a date_range aggregation for date bounds", func() { + res := build([]*searchsvc.AggregationOption{{ + Field: "photo.takenDateTime", + BucketDefinition: &searchsvc.BucketDefinition{ + Ranges: []*searchsvc.BucketRange{ + {From: "2018-08-01", To: "2018-09-01"}, + {From: "2018-08-11T00:00:00Z"}, + }, + }, + }}) + r := res["a_0"].(map[string]any)["date_range"].(map[string]any) + Expect(r["field"]).To(Equal("photo.takenDateTime")) + ranges := r["ranges"].([]map[string]any) + Expect(ranges).To(HaveLen(2)) + Expect(ranges[0]).To(SatisfyAll( + HaveKeyWithValue("key", "2018-08-01-2018-09-01"), + HaveKeyWithValue("from", "2018-08-01"), + HaveKeyWithValue("to", "2018-09-01"), + )) + Expect(ranges[1]).To(HaveKeyWithValue("from", "2018-08-11T00:00:00Z")) + Expect(ranges[1]).ToNot(HaveKey("to")) + }) + + It("rejects a bound that is neither number nor date", func() { + _, err := aggs.Build([]*searchsvc.AggregationOption{{ + Field: "photo.takenDateTime", + BucketDefinition: &searchsvc.BucketDefinition{ + Ranges: []*searchsvc.BucketRange{ + {From: "2018-08-11T00:00:00Z", To: "not-a-date"}, + }, + }, + }}) + Expect(err).To(HaveOccurred()) + }) + It("builds a range aggregation with open-ended bounds", func() { - res := aggs.Build([]*searchsvc.AggregationOption{{ + res := build([]*searchsvc.AggregationOption{{ Field: "audio.year", BucketDefinition: &searchsvc.BucketDefinition{ Ranges: []*searchsvc.BucketRange{ @@ -51,7 +92,7 @@ var _ = Describe("Build", func() { DescribeTable("builds single-value metric aggregations", func(kind searchsvc.MetricKind, esKind string) { - res := aggs.Build([]*searchsvc.AggregationOption{ + res := build([]*searchsvc.AggregationOption{ {Field: "audio.duration", MetricKind: kind}, }) body, ok := res["a_0"].(map[string]any)[esKind].(map[string]any) @@ -64,7 +105,7 @@ var _ = Describe("Build", func() { ) It("uses a stats aggregation for AVG", func() { - res := aggs.Build([]*searchsvc.AggregationOption{ + res := build([]*searchsvc.AggregationOption{ {Field: "audio.duration", MetricKind: searchsvc.MetricKind_METRIC_KIND_AVG}, }) stats, ok := res["a_0"].(map[string]any)["stats"].(map[string]any) @@ -73,7 +114,7 @@ var _ = Describe("Build", func() { }) It("nests sub-aggregations under their parent bucket", func() { - res := aggs.Build([]*searchsvc.AggregationOption{{ + res := build([]*searchsvc.AggregationOption{{ Field: "audio.artist", Size: 5, SubAggregations: []*searchsvc.AggregationOption{{ Field: "audio.album", Size: 7,