[Backport master] Add multi terms aggregation

docs/aggregations.asciidoc

@@ -168,6 +168,8 @@ In addition to the buckets themselves, the bucket aggregations also compute and
 
 * <<missing-aggregation-usage,Missing Aggregation Usage>>
 
+* <<multi-terms-aggregation-usage,Multi Terms Aggregation Usage>>
+
 * <<nested-aggregation-usage,Nested Aggregation Usage>>
 
 * <<parent-aggregation-usage,Parent Aggregation Usage>>
@@ -231,6 +233,8 @@ include::aggregations/bucket/ip-range/ip-range-aggregation-usage.asciidoc[]
 
 include::aggregations/bucket/missing/missing-aggregation-usage.asciidoc[]
 
+include::aggregations/bucket/multi-terms/multi-terms-aggregation-usage.asciidoc[]
+
 include::aggregations/bucket/nested/nested-aggregation-usage.asciidoc[]
 
 include::aggregations/bucket/parent/parent-aggregation-usage.asciidoc[]

docs/aggregations/bucket/multi-terms/multi-terms-aggregation-usage.asciidoc

@@ -0,0 +1,131 @@
+:ref_current: https://www.elastic.co/guide/en/elasticsearch/reference/7.x
+
+:github: https://github.com/elastic/elasticsearch-net
+
+:nuget: https://www.nuget.org/packages
+
+////
+IMPORTANT NOTE
+==============
+This file has been generated from https://github.com/elastic/elasticsearch-net/tree/7.x/src/Tests/Tests/Aggregations/Bucket/MultiTerms/MultiTermsAggregationUsageTests.cs. 
+If you wish to submit a PR for any spelling mistakes, typos or grammatical errors for this file,
+please modify the original csharp file found at the link and submit the PR with that change. Thanks!
+////
+
+[[multi-terms-aggregation-usage]]
+=== Multi Terms Aggregation Usage
+
+A multi-bucket value source based aggregation where buckets are dynamically built - one per unique set of values.
+
+See the Elasticsearch documentation on {ref_current}//search-aggregations-bucket-multi-terms-aggregation.html[multi terms aggregation] for more detail.
+
+==== Fluent DSL example
+
+[source,csharp]
+----
+a => a
+.MultiTerms("states", st => st
+    .CollectMode(TermsAggregationCollectMode.BreadthFirst)
+    .Terms(t => t.Field(f => f.Name), t => t.Field(f => f.NumberOfCommits).Missing(0))
+    .MinimumDocumentCount(1)
+    .Size(5)
+    .ShardSize(100)
+    .ShardMinimumDocumentCount(1)
+    .ShowTermDocCountError(true)
+    .Order(o => o
+        .KeyAscending()
+        .CountDescending()
+    )
+    .Meta(m => m
+        .Add("foo", "bar")
+    )
+)
+----
+
+==== Object Initializer syntax example
+
+[source,csharp]
+----
+new MultiTermsAggregation("states")
+{
+    CollectMode = TermsAggregationCollectMode.BreadthFirst,
+    Terms = new List<Term>
+    {
+        new() {Field = Field<Project>(f => f.Name) },
+        new() {Field = Field<Project>(f => f.NumberOfCommits), Missing = 0 }
+    },
+    MinimumDocumentCount = 1,
+    Size = 5,
+    ShardSize = 100,
+    ShardMinimumDocumentCount = 1,
+    ShowTermDocCountError = true,
+    Order = new List<TermsOrder>
+    {
+        TermsOrder.KeyAscending,
+        TermsOrder.CountDescending
+    },
+    Meta = new Dictionary<string, object>
+    {
+        { "foo", "bar" }
+    }
+}
+----
+
+[source,javascript]
+.Example json output
+----
+{
+  "states": {
+    "meta": {
+      "foo": "bar"
+    },
+    "multi_terms": {
+      "collect_mode": "breadth_first",
+      "terms": [
+        {
+          "field": "name"
+        },
+        {
+          "field": "numberOfCommits",
+          "missing": 0
+        }
+      ],
+      "min_doc_count": 1,
+      "shard_min_doc_count": 1,
+      "size": 5,
+      "shard_size": 100,
+      "show_term_doc_count_error": true,
+      "order": [
+        {
+          "_key": "asc"
+        },
+        {
+          "_count": "desc"
+        }
+      ]
+    }
+  }
+}
+----
+
+==== Handling Responses
+
+[source,csharp]
+----
+response.ShouldBeValid();
+var states = response.Aggregations.MultiTerms("states");
+states.Should().NotBeNull();
+states.DocCountErrorUpperBound.Should().HaveValue();
+states.SumOtherDocCount.Should().HaveValue();
+states.Buckets.Should().NotBeNull();
+states.Buckets.Count.Should().BeGreaterThan(0);
+foreach (var item in states.Buckets)
+{
+    item.Key.Should().NotBeNullOrEmpty();
+    item.DocCount.Should().BeGreaterOrEqualTo(1);
+    item.KeyAsString.Should().NotBeNullOrEmpty();
+}
+states.Meta.Should().NotBeNull().And.HaveCount(1);
+states.Meta["foo"].Should().Be("bar");
+----
+

src/Nest/Aggregations/AggregateDictionary.cs

@@ -221,6 +221,22 @@ public MultiBucketAggregate<RareTermsBucket<TKey>> RareTerms<TKey>(string key)
 
 		public MultiBucketAggregate<DateHistogramBucket> DateHistogram(string key) => GetMultiBucketAggregate<DateHistogramBucket>(key);
 
+		public MultiTermsAggregate<string> MultiTerms(string key) => MultiTerms<string>(key);
+
+		public MultiTermsAggregate<TKey> MultiTerms<TKey>(string key)
+		{
+			var bucket = TryGet<BucketAggregate>(key);
+			return bucket == null
+				? null
+				: new MultiTermsAggregate<TKey>
+				{
+					DocCountErrorUpperBound = bucket.DocCountErrorUpperBound,
+					SumOtherDocCount = bucket.SumOtherDocCount,
+					Buckets = GetMultiTermsBuckets<TKey>(bucket.Items).ToList(),
+					Meta = bucket.Meta
+				};
+		}
+
 		public AutoDateHistogramAggregate AutoDateHistogram(string key)
 		{
 			var bucket = TryGet<BucketAggregate>(key);
@@ -324,6 +340,28 @@ private IEnumerable<RareTermsBucket<TKey>> GetRareTermsBuckets<TKey>(IEnumerable
 				};
 		}
 
+		private static IEnumerable<MultiTermsBucket<TKey>> GetMultiTermsBuckets<TKey>(IEnumerable<IBucket> items)
+		{
+			var buckets = items.Cast<KeyedBucket<object>>();
+
+			foreach (var bucket in buckets)
+			{
+				var aggregates = new MultiTermsBucket<TKey>(bucket.BackingDictionary)
+				{
+					DocCount = bucket.DocCount,
+					DocCountErrorUpperBound = bucket.DocCountErrorUpperBound,
+					KeyAsString = bucket.KeyAsString
+				};
+
+				if (bucket.Key is List<object> allKeys)
+					aggregates.Key = allKeys.Select(GetKeyFromBucketKey<TKey>).ToList();
+				else
+					aggregates.Key = new[] { GetKeyFromBucketKey<TKey>(bucket.Key) };
+
+				yield return aggregates;
+			}
+		}
+
 		private static TKey GetKeyFromBucketKey<TKey>(object key) =>
 			typeof(TKey).IsEnum
 				? (TKey)Enum.Parse(typeof(TKey), key.ToString(), true)

src/Nest/Aggregations/AggregateFormatter.cs

@@ -935,9 +935,7 @@ private IBucket GetDateHistogramBucket(ref JsonReader reader, IJsonFormatterReso
 
 			return dateHistogram;
 		}
-
-
-
+
 		private IBucket GetKeyedBucket(ref JsonReader reader, IJsonFormatterResolver formatterResolver)
 		{
 			var token = reader.GetCurrentJsonToken();
@@ -947,6 +945,32 @@ private IBucket GetKeyedBucket(ref JsonReader reader, IJsonFormatterResolver for
 			object key;
 			if (token == JsonToken.String)
 				key = reader.ReadString();
+			else if (token == JsonToken.BeginArray) // we're in a multi-terms bucket
+			{
+				var keys = new List<object>();
+				var count = 0;
+
+				while(reader.ReadIsInArray(ref count))
+				{
+					object keyItem;
+
+					var keyToken = reader.GetCurrentJsonToken();
+
+					if (keyToken == JsonToken.String)
+						keyItem = reader.ReadString();
+					else
+					{
+						var numberKey = reader.ReadNumberSegment();
+						if (numberKey.IsLong())
+							keyItem = NumberConverter.ReadInt64(numberKey.Array, numberKey.Offset, out _);
+						else
+							keyItem = NumberConverter.ReadDouble(numberKey.Array, numberKey.Offset, out _);
+					}
+
+					keys.Add(keyItem);
+				}
+				key = keys;
+			}
 			else
 			{
 				var numberSegment = reader.ReadNumberSegment();

src/Nest/Aggregations/AggregationContainer.cs

@@ -294,6 +294,9 @@ public interface IAggregationContainer
 		[DataMember(Name = "top_metrics")]
 		ITopMetricsAggregation TopMetrics { get; set; }
 
+		[DataMember(Name = "multi_terms")]
+		IMultiTermsAggregation MultiTerms { get; set; }
+
 		void Accept(IAggregationVisitor visitor);
 	}
 
@@ -442,6 +445,8 @@ public class AggregationContainer : IAggregationContainer
 
 		public ITopMetricsAggregation TopMetrics { get; set; }
 
+		public IMultiTermsAggregation MultiTerms { get; set; }
+
 		public void Accept(IAggregationVisitor visitor)
 		{
 			if (visitor.Scope == AggregationVisitorScope.Unknown) visitor.Scope = AggregationVisitorScope.Aggregation;
@@ -555,6 +560,8 @@ public class AggregationContainerDescriptor<T> : DescriptorBase<AggregationConta
 
 		IMovingPercentilesAggregation IAggregationContainer.MovingPercentiles { get; set; }
 
+		IMultiTermsAggregation IAggregationContainer.MultiTerms { get; set; }
+
 		INestedAggregation IAggregationContainer.Nested { get; set; }
 
 		INormalizeAggregation IAggregationContainer.Normalize { get; set; }
@@ -719,6 +726,11 @@ Func<MissingAggregationDescriptor<T>, IMissingAggregation> selector
 		) =>
 			_SetInnerAggregation(name, selector, (a, d) => a.Missing = d);
 
+		public AggregationContainerDescriptor<T> MultiTerms(string name,
+			Func<MultiTermsAggregationDescriptor<T>, IMultiTermsAggregation> selector
+		) =>
+			_SetInnerAggregation(name, selector, (a, d) => a.MultiTerms = d);
+
 		public AggregationContainerDescriptor<T> Nested(string name,
 			Func<NestedAggregationDescriptor<T>, INestedAggregation> selector
 		) =>

src/Nest/Aggregations/Bucket/MultiTerms/MultiTermsAggregate.cs

@@ -0,0 +1,12 @@
+// Licensed to Elasticsearch B.V under one or more agreements.
+// Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information
+
+namespace Nest
+{
+	public class MultiTermsAggregate<TKey> : MultiBucketAggregate<MultiTermsBucket<TKey>>
+	{
+		public long? DocCountErrorUpperBound { get; set; }
+		public long? SumOtherDocCount { get; set; }
+	}
+}