Add MultiTerms bucket aggregation

This commit adds the Multi Terms bucket aggregation as described here:

https://www.elastic.co/guide/en/elasticsearch/reference/7.13/search-aggregations-bucket-multi-terms-aggregation.html

Author: olivere

search_aggs.go

@@ -369,6 +369,21 @@ func (a Aggregations) Terms(name string) (*AggregationBucketKeyItems, bool) {
 	return nil, false
 }
 
+// MultiTerms returns multi terms aggregation results.
+// See: https://www.elastic.co/guide/en/elasticsearch/reference/7.13/search-aggregations-bucket-multi-terms-aggregation.html
+func (a Aggregations) MultiTerms(name string) (*AggregationBucketMultiKeyItems, bool) {
+	if raw, found := a[name]; found {
+		agg := new(AggregationBucketMultiKeyItems)
+		if raw == nil {
+			return agg, true
+		}
+		if err := json.Unmarshal(raw, agg); err == nil {
+			return agg, true
+		}
+	}
+	return nil, false
+}
+
 // SignificantTerms returns significant terms aggregation results.
 // See: https://www.elastic.co/guide/en/elasticsearch/reference/7.0/search-aggregations-bucket-significantterms-aggregation.html
 func (a Aggregations) SignificantTerms(name string) (*AggregationBucketSignificantTerms, bool) {
@@ -1342,6 +1357,71 @@ func (a *AggregationBucketKeyItem) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+// AggregationBucketMultiKeyItems is a bucket aggregation that is returned
+// with a multi terms aggregation.
+type AggregationBucketMultiKeyItems struct {
+	Aggregations
+
+	DocCountErrorUpperBound int64                            //`json:"doc_count_error_upper_bound"`
+	SumOfOtherDocCount      int64                            //`json:"sum_other_doc_count"`
+	Buckets                 []*AggregationBucketMultiKeyItem //`json:"buckets"`
+	Meta                    map[string]interface{}           // `json:"meta,omitempty"`
+}
+
+// UnmarshalJSON decodes JSON data and initializes an AggregationBucketMultiKeyItems structure.
+func (a *AggregationBucketMultiKeyItems) UnmarshalJSON(data []byte) error {
+	var aggs map[string]json.RawMessage
+	if err := json.Unmarshal(data, &aggs); err != nil {
+		return err
+	}
+	if v, ok := aggs["doc_count_error_upper_bound"]; ok && v != nil {
+		json.Unmarshal(v, &a.DocCountErrorUpperBound)
+	}
+	if v, ok := aggs["sum_other_doc_count"]; ok && v != nil {
+		json.Unmarshal(v, &a.SumOfOtherDocCount)
+	}
+	if v, ok := aggs["buckets"]; ok && v != nil {
+		json.Unmarshal(v, &a.Buckets)
+	}
+	if v, ok := aggs["meta"]; ok && v != nil {
+		json.Unmarshal(v, &a.Meta)
+	}
+	a.Aggregations = aggs
+	return nil
+}
+
+// AggregationBucketMultiKeyItem is a single bucket of an AggregationBucketMultiKeyItems structure.
+type AggregationBucketMultiKeyItem struct {
+	Aggregations
+
+	Key         []interface{} //`json:"key"`
+	KeyAsString *string       //`json:"key_as_string"`
+	KeyNumber   []json.Number
+	DocCount    int64 //`json:"doc_count"`
+}
+
+// UnmarshalJSON decodes JSON data and initializes an AggregationBucketMultiKeyItem structure.
+func (a *AggregationBucketMultiKeyItem) UnmarshalJSON(data []byte) error {
+	var aggs map[string]json.RawMessage
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.UseNumber()
+	if err := dec.Decode(&aggs); err != nil {
+		return err
+	}
+	if v, ok := aggs["key"]; ok && v != nil {
+		json.Unmarshal(v, &a.Key)
+		json.Unmarshal(v, &a.KeyNumber)
+	}
+	if v, ok := aggs["key_as_string"]; ok && v != nil {
+		json.Unmarshal(v, &a.KeyAsString)
+	}
+	if v, ok := aggs["doc_count"]; ok && v != nil {
+		json.Unmarshal(v, &a.DocCount)
+	}
+	a.Aggregations = aggs
+	return nil
+}
+
 // -- Bucket types for significant terms --
 
 // AggregationBucketSignificantTerms is a bucket aggregation returned

search_aggs_bucket_multi_terms.go

@@ -4,17 +4,17 @@
 
 package elastic
 
-import "fmt"
-
-// A multi-bucket value source based aggregation where buckets are dynamically built - one per
-// unique set of values. The multi terms aggregation is very similar to the terms aggregation,
-// however in most cases it will be slower than the terms aggregation and will consume more
-// memory. Therefore, if the same set of fields is constantly used, it would be more efficient to
-// index a combined key for this fields as a separate field and use the terms aggregation on this field.
+// MultiTermsAggregation is a multi-bucket value source based aggregation
+// where buckets are dynamically built - one per unique set of values.
+// The multi terms aggregation is very similar to the terms aggregation,
+// however in most cases it will be slower than the terms aggregation and will
+// consume more memory. Therefore, if the same set of fields is constantly
+// used, it would be more efficient to index a combined key for this fields
+// as a separate field and use the terms aggregation on this field.
 //
-// See: https://www.elastic.co/guide/en/elasticsearch/reference/7.12/search-aggregations-bucket-multi-terms-aggregation.html
+// See https://www.elastic.co/guide/en/elasticsearch/reference/7.13/search-aggregations-bucket-multi-terms-aggregation.html
 type MultiTermsAggregation struct {
-	multiTerms      []*MultiTerm
+	multiTerms      []MultiTerm
 	subAggregations map[string]Aggregation
 	meta            map[string]interface{}
 
@@ -27,14 +27,30 @@ type MultiTermsAggregation struct {
 	order                 []MultiTermsOrder
 }
 
+// NewMultiTermsAggregation initializes a new MultiTermsAggregation.
 func NewMultiTermsAggregation() *MultiTermsAggregation {
 	return &MultiTermsAggregation{
 		subAggregations: make(map[string]Aggregation),
 	}
 }
 
-func (a *MultiTermsAggregation) Terms(multiTerm ...*MultiTerm) *MultiTermsAggregation {
-	a.multiTerms = multiTerm
+// Terms adds a slice of field names to return in the aggregation.
+//
+// Notice that it appends to existing terms, so you can use Terms more than
+// once, and mix with MultiTerms method.
+func (a *MultiTermsAggregation) Terms(fields ...string) *MultiTermsAggregation {
+	for _, field := range fields {
+		a.multiTerms = append(a.multiTerms, MultiTerm{Field: field})
+	}
+	return a
+}
+
+// MultiTerms adds a slice of MultiTerm instances to return in the aggregation.
+//
+// Notice that it appends to existing terms, so you can use MultiTerms more
+// than once, and mix with Terms method.
+func (a *MultiTermsAggregation) MultiTerms(multiTerms ...MultiTerm) *MultiTermsAggregation {
+	a.multiTerms = append(a.multiTerms, multiTerms...)
 	return a
 }
 
@@ -206,9 +222,6 @@ func (a *MultiTermsAggregation) Source() (interface{}, error) {
 	// ValuesSourceAggregationBuilder
 	terms := make([]interface{}, len(a.multiTerms))
 	for i := range a.multiTerms {
-		if a.multiTerms[i] == nil {
-			return nil, fmt.Errorf("expected a multiterm but found a nil multiterm")
-		}
 		s, err := a.multiTerms[i].Source()
 		if err != nil {
 			return nil, err
@@ -288,26 +301,16 @@ func (order *MultiTermsOrder) Source() (interface{}, error) {
 
 // MultiTerm specifies a single term field for a multi terms aggregation.
 type MultiTerm struct {
-	field   string
-	missing interface{}
+	Field   string
+	Missing interface{}
 }
 
 // Source returns serializable JSON of the MultiTerm.
 func (term *MultiTerm) Source() (interface{}, error) {
 	source := make(map[string]interface{})
-	source["field"] = term.field
-	if term.missing != nil {
-		source["missing"] = term.missing
+	source["field"] = term.Field
+	if term.Missing != nil {
+		source["missing"] = term.Missing
 	}
 	return source, nil
 }
-
-// Missing configures the value to use when document miss a value
-func (term *MultiTerm) Missing(missing interface{}) *MultiTerm {
-	term.missing = missing
-	return term
-}
-
-func NewMultiTerm(field string) *MultiTerm {
-	return &MultiTerm{field: field}
-}

search_aggs_bucket_multi_terms_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestMultiTermsAggregation(t *testing.T) {
-	agg := NewMultiTermsAggregation().Terms(NewMultiTerm("genre"), NewMultiTerm("product"))
+	agg := NewMultiTermsAggregation().Terms("genre", "product")
 	src, err := agg.Source()
 	if err != nil {
 		t.Fatal(err)
@@ -26,9 +26,29 @@ func TestMultiTermsAggregation(t *testing.T) {
 	}
 }
 
+func TestMultiTermsAggregationWithMultiTerms(t *testing.T) {
+	agg := NewMultiTermsAggregation().MultiTerms(
+		MultiTerm{Field: "genre", Missing: "n/a"},
+		MultiTerm{Field: "product", Missing: "n/a"},
+	)
+	src, err := agg.Source()
+	if err != nil {
+		t.Fatal(err)
+	}
+	data, err := json.Marshal(src)
+	if err != nil {
+		t.Fatalf("marshaling to JSON failed: %v", err)
+	}
+	got := string(data)
+	expected := `{"multi_terms":{"terms":[{"field":"genre","missing":"n/a"},{"field":"product","missing":"n/a"}]}}`
+	if got != expected {
+		t.Errorf("expected\n%s\n,got:\n%s", expected, got)
+	}
+}
+
 func TestMultiTermsAggregationWithSubAggregation(t *testing.T) {
 	subAgg := NewAvgAggregation().Field("height")
-	agg := NewMultiTermsAggregation().Terms(NewMultiTerm("genre"), NewMultiTerm("product")).Size(10).
+	agg := NewMultiTermsAggregation().Terms("genre", "product").Size(10).
 		OrderByAggregation("avg_height", false)
 	agg = agg.SubAggregation("avg_height", subAgg)
 	src, err := agg.Source()
@@ -49,7 +69,7 @@ func TestMultiTermsAggregationWithSubAggregation(t *testing.T) {
 func TestMultiTermsAggregationWithMultipleSubAggregation(t *testing.T) {
 	subAgg1 := NewAvgAggregation().Field("height")
 	subAgg2 := NewAvgAggregation().Field("width")
-	agg := NewMultiTermsAggregation().Terms(NewMultiTerm("genre"), NewMultiTerm("product")).Size(10).
+	agg := NewMultiTermsAggregation().Terms("genre", "product").Size(10).
 		OrderByAggregation("avg_height", false)
 	agg = agg.SubAggregation("avg_height", subAgg1)
 	agg = agg.SubAggregation("avg_width", subAgg2)
@@ -69,7 +89,7 @@ func TestMultiTermsAggregationWithMultipleSubAggregation(t *testing.T) {
 }
 
 func TestMultiTermsAggregationWithMetaData(t *testing.T) {
-	agg := NewMultiTermsAggregation().Terms(NewMultiTerm("genre"), NewMultiTerm("product")).Size(10).OrderByKeyDesc()
+	agg := NewMultiTermsAggregation().Terms("genre", "product").Size(10).OrderByKeyDesc()
 	agg = agg.Meta(map[string]interface{}{"name": "Oliver"})
 	src, err := agg.Source()
 	if err != nil {
@@ -87,7 +107,10 @@ func TestMultiTermsAggregationWithMetaData(t *testing.T) {
 }
 
 func TestMultiTermsAggregationWithMissing(t *testing.T) {
-	agg := NewMultiTermsAggregation().Terms(NewMultiTerm("genre"), NewMultiTerm("product").Missing("n/a")).Size(10)
+	agg := NewMultiTermsAggregation().MultiTerms(
+		MultiTerm{Field: "genre"},
+		MultiTerm{Field: "product", Missing: "n/a"},
+	).Size(10)
 	src, err := agg.Source()
 	if err != nil {
 		t.Fatal(err)

search_aggs_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 // TestAggs is an integration test for most aggregation types.
-func TestAggs(t *testing.T) {
+func TestAggs123(t *testing.T) {
 	client := setupTestClientAndCreateIndex(t) //, SetTraceLog(log.New(os.Stdout, "", log.LstdFlags)))
 
 	tweet1 := tweet{
@@ -79,6 +79,7 @@ func TestAggs(t *testing.T) {
 	globalAgg := NewGlobalAggregation()
 	usersAgg := NewTermsAggregation().Field("user").Size(10).OrderByCountDesc()
 	retweetsAgg := NewTermsAggregation().Field("retweets").Size(10)
+	multiTermsAgg := NewMultiTermsAggregation().Terms("user").MultiTerms(MultiTerm{Field: "tags", Missing: "unclassified"}).Size(10)
 	avgRetweetsAgg := NewAvgAggregation().Field("retweets")
 	avgRetweetsWithMetaAgg := NewAvgAggregation().Field("retweetsMeta").Meta(map[string]interface{}{"meta": true})
 	weightedAvgRetweetsAgg := NewWeightedAvgAggregation().
@@ -127,6 +128,7 @@ func TestAggs(t *testing.T) {
 	builder = builder.Aggregation("global", globalAgg)
 	builder = builder.Aggregation("users", usersAgg)
 	builder = builder.Aggregation("retweets", retweetsAgg)
+	builder = builder.Aggregation("multiterms", multiTermsAgg)
 	builder = builder.Aggregation("avgRetweets", avgRetweetsAgg)
 	builder = builder.Aggregation("avgRetweetsWithMeta", avgRetweetsWithMetaAgg)
 	builder = builder.Aggregation("weightedAvgRetweets", weightedAvgRetweetsAgg)
@@ -339,6 +341,38 @@ func TestAggs(t *testing.T) {
 		}
 	}
 
+	// A multi terms aggregate
+	{
+		agg, found := agg.MultiTerms("multiterms")
+		if !found {
+			t.Errorf("expected %v; got: %v", true, found)
+		}
+		if agg == nil {
+			t.Fatalf("expected != nil; got: nil")
+		}
+		if len(agg.Buckets) != 4 {
+			t.Fatalf("expected %d; got: %d", 4, len(agg.Buckets))
+		}
+		if want, have := 2, len(agg.Buckets[0].Key); want != have {
+			t.Errorf("expected %v; got: %v", want, have)
+		}
+		if want, have := "olivere", agg.Buckets[0].Key[0]; want != have {
+			t.Errorf("expected %v; got: %v", want, have)
+		}
+		if want, have := "golang", agg.Buckets[0].Key[1]; want != have {
+			t.Errorf("expected %v; got: %v", want, have)
+		}
+		if agg.Buckets[0].KeyAsString == nil {
+			t.Fatal("expected string not nil")
+		}
+		if want, have := "olivere|golang", *agg.Buckets[0].KeyAsString; want != have {
+			t.Errorf("expected %v; got: %v", want, have)
+		}
+		if agg.Buckets[0].DocCount != 2 {
+			t.Errorf("expected %d; got: %d", 2, agg.Buckets[0].DocCount)
+		}
+	}
+
 	// avgRetweets
 	{
 		agg, found := agg.Avg("avgRetweets")