Query: Adds Distribution for MakeList and MakeSet

Microsoft.Azure.Cosmos/src/Query/Core/Pipeline/Aggregate/AggregateOperator.cs

@@ -8,6 +8,8 @@ internal enum AggregateOperator
     {
         Average,
         Count,
+        MakeList,
+        MakeSet,
         Max,
         Min,
         Sum,

Microsoft.Azure.Cosmos/src/Query/Core/Pipeline/Aggregate/Aggregators/MakeListAggregator.cs

@@ -0,0 +1,72 @@
+//------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+//------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.Query.Core.Pipeline.Aggregate.Aggregators
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Diagnostics;
+    using System.Globalization;
+    using System.Linq;
+    using System.Text;
+    using Microsoft.Azure.Cosmos.CosmosElements;
+    using Microsoft.Azure.Cosmos.CosmosElements.Numbers;
+    using Microsoft.Azure.Cosmos.Query.Core.Exceptions;
+    using Microsoft.Azure.Cosmos.Query.Core.Monads;
+
+    internal sealed class MakeListAggregator : IAggregator
+    {
+        private readonly List<CosmosElement> globalList;
+
+        private MakeListAggregator(CosmosArray initialList)
+        {
+            this.globalList = new List<CosmosElement>();
+            foreach (CosmosElement listItem in initialList)
+            {
+                this.globalList.Add(listItem);
+            }
+        }
+
+        public void Aggregate(CosmosElement localList)
+        {
+            if (!(localList is CosmosArray cosmosArray))
+            {
+                throw new ArgumentException($"{nameof(localList)} must be an array.");
+            }
+
+            this.globalList.AddRange(cosmosArray.ToList<CosmosElement>());
+        }
+
+        public CosmosElement GetResult()
+        {
+            return CosmosArray.Create(this.globalList);
+        }
+
+        public static TryCatch<IAggregator> TryCreate(CosmosElement continuationToken)
+        {
+            CosmosArray partialList;
+            if (continuationToken != null)
+            {
+                if (!(continuationToken is CosmosArray cosmosPartialList))
+                {
+                    return TryCatch<IAggregator>.FromException(
+                        new MalformedContinuationTokenException($@"Invalid MakeList continuation token: ""{continuationToken}""."));
+                }
+
+                partialList = cosmosPartialList;
+            }
+            else
+            {
+                partialList = CosmosArray.Empty;
+            }
+
+            return TryCatch<IAggregator>.FromResult(new MakeListAggregator(initialList: partialList));
+        }
+
+        public CosmosElement GetCosmosElementContinuationToken()
+        {
+            return this.GetResult();
+        }
+    }
+}

Microsoft.Azure.Cosmos/src/Query/Core/Pipeline/Aggregate/Aggregators/MakeSetAggregator.cs

@@ -0,0 +1,77 @@
+//------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+//------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.Query.Core.Pipeline.Aggregate.Aggregators
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Diagnostics;
+    using System.Globalization;
+    using System.Linq;
+    using System.Text;
+    using Microsoft.Azure.Cosmos.CosmosElements;
+    using Microsoft.Azure.Cosmos.CosmosElements.Numbers;
+    using Microsoft.Azure.Cosmos.Query.Core.Exceptions;
+    using Microsoft.Azure.Cosmos.Query.Core.Monads;
+
+    internal sealed class MakeSetAggregator : IAggregator
+    {
+        private readonly HashSet<CosmosElement> globalSet;
+
+        private MakeSetAggregator(CosmosArray initialSet)
+        {
+            this.globalSet = new HashSet<CosmosElement>();
+            foreach (CosmosElement setItem in initialSet)
+            {
+                this.globalSet.Add(setItem);
+            }
+        }
+
+        public void Aggregate(CosmosElement localSet)
+        {
+            if (!(localSet is CosmosArray cosmosArray))
+            {
+                throw new ArgumentException($"{nameof(localSet)} must be an array.");
+            }
+
+            this.globalSet.UnionWith(cosmosArray.ToList<CosmosElement>());
+        }
+
+        public CosmosElement GetResult()
+        {
+            return CosmosArray.Create(this.globalSet);
+        }
+
+        public string GetContinuationToken()
+        {
+            return this.globalSet.ToString();
+        }
+
+        public static TryCatch<IAggregator> TryCreate(CosmosElement continuationToken)
+        {
+            CosmosArray partialSet;
+            if (continuationToken != null)
+            {
+                if (!(continuationToken is CosmosArray cosmosPartialSet))
+                {
+                    return TryCatch<IAggregator>.FromException(
+                        new MalformedContinuationTokenException($@"Invalid MakeSet continuation token: ""{continuationToken}""."));
+                }
+
+                partialSet = cosmosPartialSet;
+            }
+            else
+            {
+                partialSet = CosmosArray.Empty;
+            }
+
+            return TryCatch<IAggregator>.FromResult(new MakeSetAggregator(initialSet: partialSet));
+        }
+
+        public CosmosElement GetCosmosElementContinuationToken()
+        {
+            return this.GetResult();
+        }
+    }
+}

Microsoft.Azure.Cosmos/src/Query/Core/Pipeline/Aggregate/Aggregators/SingleGroupAggregator.cs

@@ -370,6 +370,14 @@ public static TryCatch<AggregateValue> TryCreate(
                             tryCreateAggregator = CountAggregator.TryCreate(continuationToken);
                             break;
 
+                        case AggregateOperator.MakeList:
+                            tryCreateAggregator = MakeListAggregator.TryCreate(continuationToken);
+                            break;
+
+                        case AggregateOperator.MakeSet:
+                            tryCreateAggregator = MakeSetAggregator.TryCreate(continuationToken);
+                            break;
+
                         case AggregateOperator.Max:
                             tryCreateAggregator = MinMaxAggregator.TryCreateMaxAggregator(continuationToken);
                             break;