Add GroupJoin overload returning IGrouping<TOuter, TInner>

src/libraries/System.Linq.AsyncEnumerable/ref/System.Linq.AsyncEnumerable.cs

@@ -73,6 +73,8 @@ public static partial class AsyncEnumerable
         public static System.Collections.Generic.IAsyncEnumerable<TResult> GroupBy<TSource, TKey, TResult>(this System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Func<TSource, TKey> keySelector, System.Func<TKey, System.Collections.Generic.IEnumerable<TSource>, TResult> resultSelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
         public static System.Collections.Generic.IAsyncEnumerable<TResult> GroupBy<TSource, TKey, TElement, TResult>(this System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Func<TSource, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TKey>> keySelector, System.Func<TSource, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TElement>> elementSelector, System.Func<TKey, System.Collections.Generic.IEnumerable<TElement>, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TResult>> resultSelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
         public static System.Collections.Generic.IAsyncEnumerable<TResult> GroupBy<TSource, TKey, TElement, TResult>(this System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Func<TSource, TKey> keySelector, System.Func<TSource, TElement> elementSelector, System.Func<TKey, System.Collections.Generic.IEnumerable<TElement>, TResult> resultSelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
+        public static System.Collections.Generic.IAsyncEnumerable<System.Linq.IGrouping<TOuter, TInner>> GroupJoin<TOuter, TInner, TKey>(this System.Collections.Generic.IAsyncEnumerable<TOuter> outer, System.Collections.Generic.IAsyncEnumerable<TInner> inner, System.Func<TOuter, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TKey>> outerKeySelector, System.Func<TInner, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TKey>> innerKeySelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
+        public static System.Collections.Generic.IAsyncEnumerable<System.Linq.IGrouping<TOuter, TInner>> GroupJoin<TOuter, TInner, TKey>(this System.Collections.Generic.IAsyncEnumerable<TOuter> outer, System.Collections.Generic.IAsyncEnumerable<TInner> inner, System.Func<TOuter, TKey> outerKeySelector, System.Func<TInner, TKey> innerKeySelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
         public static System.Collections.Generic.IAsyncEnumerable<TResult> GroupJoin<TOuter, TInner, TKey, TResult>(this System.Collections.Generic.IAsyncEnumerable<TOuter> outer, System.Collections.Generic.IAsyncEnumerable<TInner> inner, System.Func<TOuter, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TKey>> outerKeySelector, System.Func<TInner, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TKey>> innerKeySelector, System.Func<TOuter, System.Collections.Generic.IEnumerable<TInner>, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<TResult>> resultSelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
         public static System.Collections.Generic.IAsyncEnumerable<TResult> GroupJoin<TOuter, TInner, TKey, TResult>(this System.Collections.Generic.IAsyncEnumerable<TOuter> outer, System.Collections.Generic.IAsyncEnumerable<TInner> inner, System.Func<TOuter, TKey> outerKeySelector, System.Func<TInner, TKey> innerKeySelector, System.Func<TOuter, System.Collections.Generic.IEnumerable<TInner>, TResult> resultSelector, System.Collections.Generic.IEqualityComparer<TKey>? comparer = null) { throw null; }
         public static System.Collections.Generic.IAsyncEnumerable<(int Index, TSource Item)> Index<TSource>(this System.Collections.Generic.IAsyncEnumerable<TSource> source) { throw null; }

src/libraries/System.Linq.AsyncEnumerable/src/System/Linq/GroupJoin.cs

@@ -1,6 +1,7 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System.Collections;
 using System.Collections.Generic;
 using System.Runtime.CompilerServices;
 using System.Threading;
@@ -10,6 +11,120 @@ namespace System.Linq
 {
     public static partial class AsyncEnumerable
     {
+        /// <summary>Correlates the elements of two sequences based on key equality and groups the results.</summary>
+        /// <typeparam name="TOuter">The type of the elements of the first sequence.</typeparam>
+        /// <typeparam name="TInner">The type of the elements of the second sequence.</typeparam>
+        /// <typeparam name="TKey">The type of the keys returned by the key selector functions.</typeparam>
+        /// <param name="outer">The first sequence to join.</param>
+        /// <param name="inner">The sequence to join to the first sequence.</param>
+        /// <param name="outerKeySelector">A function to extract the join key from each element of the first sequence.</param>
+        /// <param name="innerKeySelector">A function to extract the join key from each element of the second sequence.</param>
+        /// <param name="comparer">An <see cref="IEqualityComparer{T}"/> to use to hash and compare keys.</param>
+        /// <returns>
+        /// An <see cref="IAsyncEnumerable{T}"/> that contains elements of type <see cref="IGrouping{TKey, TElement}"/>
+        /// where each grouping contains the outer element as the key and the matching inner elements.
+        /// </returns>
+        /// <exception cref="ArgumentNullException"><paramref name="outer" /> is <see langword="null" />.</exception>
+        /// <exception cref="ArgumentNullException"><paramref name="inner" /> is <see langword="null" />.</exception>
+        /// <exception cref="ArgumentNullException"><paramref name="outerKeySelector" /> is <see langword="null" />.</exception>
+        /// <exception cref="ArgumentNullException"><paramref name="innerKeySelector" /> is <see langword="null" />.</exception>
+        public static IAsyncEnumerable<IGrouping<TOuter, TInner>> GroupJoin<TOuter, TInner, TKey>(
+            this IAsyncEnumerable<TOuter> outer,
+            IAsyncEnumerable<TInner> inner,
+            Func<TOuter, TKey> outerKeySelector,
+            Func<TInner, TKey> innerKeySelector,
+            IEqualityComparer<TKey>? comparer = null)
+        {
+            ArgumentNullException.ThrowIfNull(outer);
+            ArgumentNullException.ThrowIfNull(inner);
+            ArgumentNullException.ThrowIfNull(outerKeySelector);
+            ArgumentNullException.ThrowIfNull(innerKeySelector);
+
+            return
+                outer.IsKnownEmpty() ? Empty<IGrouping<TOuter, TInner>>() :
+                Impl(outer, inner, outerKeySelector, innerKeySelector, comparer, default);
+
+            static async IAsyncEnumerable<IGrouping<TOuter, TInner>> Impl(
+                IAsyncEnumerable<TOuter> outer,
+                IAsyncEnumerable<TInner> inner,
+                Func<TOuter, TKey> outerKeySelector,
+                Func<TInner, TKey> innerKeySelector,
+                IEqualityComparer<TKey>? comparer,
+                [EnumeratorCancellation] CancellationToken cancellationToken)
+            {
+                await using IAsyncEnumerator<TOuter> e = outer.GetAsyncEnumerator(cancellationToken);
+
+                if (await e.MoveNextAsync())
+                {
+                    AsyncLookup<TKey, TInner> lookup = await AsyncLookup<TKey, TInner>.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken);
+                    do
+                    {
+                        TOuter item = e.Current;
+                        yield return new AsyncGroupJoinGrouping<TOuter, TInner>(item, lookup[outerKeySelector(item)]);
+                    }
+                    while (await e.MoveNextAsync());
+                }
+            }
+        }
+
+        /// <summary>Correlates the elements of two sequences based on key equality and groups the results.</summary>
+        /// <typeparam name="TOuter">The type of the elements of the first sequence.</typeparam>
+        /// <typeparam name="TInner">The type of the elements of the second sequence.</typeparam>
+        /// <typeparam name="TKey">The type of the keys returned by the key selector functions.</typeparam>
+        /// <param name="outer">The first sequence to join.</param>
+        /// <param name="inner">The sequence to join to the first sequence.</param>
+        /// <param name="outerKeySelector">A function to extract the join key from each element of the first sequence.</param>
+        /// <param name="innerKeySelector">A function to extract the join key from each element of the second sequence.</param>
+        /// <param name="comparer">An <see cref="IEqualityComparer{T}"/> to use to hash and compare keys.</param>
+        /// <returns>
+        /// An <see cref="IAsyncEnumerable{T}"/> that contains elements of type <see cref="IGrouping{TKey, TElement}"/>
+        /// where each grouping contains the outer element as the key and the matching inner elements.
+        /// </returns>
+        /// <exception cref="ArgumentNullException"><paramref name="outer" /> is <see langword="null" />.</exception>
+        /// <exception cref="ArgumentNullException"><paramref name="inner" /> is <see langword="null" />.</exception>
+        /// <exception cref="ArgumentNullException"><paramref name="outerKeySelector" /> is <see langword="null" />.</exception>
+        /// <exception cref="ArgumentNullException"><paramref name="innerKeySelector" /> is <see langword="null" />.</exception>
+        public static IAsyncEnumerable<IGrouping<TOuter, TInner>> GroupJoin<TOuter, TInner, TKey>(
+            this IAsyncEnumerable<TOuter> outer,
+            IAsyncEnumerable<TInner> inner,
+            Func<TOuter, CancellationToken, ValueTask<TKey>> outerKeySelector,
+            Func<TInner, CancellationToken, ValueTask<TKey>> innerKeySelector,
+            IEqualityComparer<TKey>? comparer = null)
+        {
+            ArgumentNullException.ThrowIfNull(outer);
+            ArgumentNullException.ThrowIfNull(inner);
+            ArgumentNullException.ThrowIfNull(outerKeySelector);
+            ArgumentNullException.ThrowIfNull(innerKeySelector);
+
+            return
+                outer.IsKnownEmpty() ? Empty<IGrouping<TOuter, TInner>>() :
+                Impl(outer, inner, outerKeySelector, innerKeySelector, comparer, default);
+
+            static async IAsyncEnumerable<IGrouping<TOuter, TInner>> Impl(
+                IAsyncEnumerable<TOuter> outer,
+                IAsyncEnumerable<TInner> inner,
+                Func<TOuter, CancellationToken, ValueTask<TKey>> outerKeySelector,
+                Func<TInner, CancellationToken, ValueTask<TKey>> innerKeySelector,
+                IEqualityComparer<TKey>? comparer,
+                [EnumeratorCancellation] CancellationToken cancellationToken)
+            {
+                await using IAsyncEnumerator<TOuter> e = outer.GetAsyncEnumerator(cancellationToken);
+
+                if (await e.MoveNextAsync())
+                {
+                    AsyncLookup<TKey, TInner> lookup = await AsyncLookup<TKey, TInner>.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken);
+                    do
+                    {
+                        TOuter item = e.Current;
+                        yield return new AsyncGroupJoinGrouping<TOuter, TInner>(
+                            item,
+                            lookup[await outerKeySelector(item, cancellationToken)]);
+                    }
+                    while (await e.MoveNextAsync());
+                }
+            }
+        }
+
         /// <summary>Correlates the elements of two sequences based on key equality and groups the results.</summary>
         /// <typeparam name="TOuter"></typeparam>
         /// <typeparam name="TInner"></typeparam>
@@ -143,4 +258,22 @@ lookup[await outerKeySelector(item, cancellationToken)],
             }
         }
     }
+
+    internal sealed class AsyncGroupJoinGrouping<TKey, TElement> : IGrouping<TKey, TElement>
+    {
+        private readonly TKey _key;
+        private readonly IEnumerable<TElement> _elements;
+
+        public AsyncGroupJoinGrouping(TKey key, IEnumerable<TElement> elements)
+        {
+            _key = key;
+            _elements = elements;
+        }
+
+        public TKey Key => _key;
+
+        public IEnumerator<TElement> GetEnumerator() => _elements.GetEnumerator();
+
+        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+    }
 }

src/libraries/System.Linq.AsyncEnumerable/tests/GroupJoinTests.cs

@@ -26,13 +26,34 @@ public void InvalidInputs_Throws()
             AssertExtensions.Throws<ArgumentNullException>("resultSelector", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>(), async (outer, ct) => outer, async (inner, ct) => inner, (Func<string, IEnumerable<string>, CancellationToken, ValueTask<string>>)null));
         }
 
+        [Fact]
+        public void InvalidInputs_WithoutResultSelector_Throws()
+        {
+            AssertExtensions.Throws<ArgumentNullException>("outer", () => AsyncEnumerable.GroupJoin((IAsyncEnumerable<string>)null, AsyncEnumerable.Empty<string>(), outer => outer, inner => inner));
+            AssertExtensions.Throws<ArgumentNullException>("inner", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), (IAsyncEnumerable<string>)null, outer => outer, inner => inner));
+            AssertExtensions.Throws<ArgumentNullException>("outerKeySelector", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>(), (Func<string, string>)null, inner => inner));
+            AssertExtensions.Throws<ArgumentNullException>("innerKeySelector", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>(), outer => outer, (Func<string, string>)null));
+
+            AssertExtensions.Throws<ArgumentNullException>("outer", () => AsyncEnumerable.GroupJoin((IAsyncEnumerable<string>)null, AsyncEnumerable.Empty<string>(), async (outer, ct) => outer, async (inner, ct) => inner));
+            AssertExtensions.Throws<ArgumentNullException>("inner", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), (IAsyncEnumerable<string>)null, async (outer, ct) => outer, async (inner, ct) => inner));
+            AssertExtensions.Throws<ArgumentNullException>("outerKeySelector", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>(), (Func<string, CancellationToken, ValueTask<string>>)null, async (inner, ct) => inner));
+            AssertExtensions.Throws<ArgumentNullException>("innerKeySelector", () => AsyncEnumerable.GroupJoin(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>(), async (outer, ct) => outer, (Func<string, CancellationToken, ValueTask<string>>)null));
+        }
+
         [Fact]
         public void Empty_ProducesEmpty() // validating an optimization / implementation detail
         {
             Assert.Same(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>().GroupJoin(CreateSource(1, 2, 3), s => s, i => i.ToString(), (s, e) => s));
             Assert.Same(AsyncEnumerable.Empty<string>(), AsyncEnumerable.Empty<string>().GroupJoin(CreateSource(1, 2, 3), async (s, ct) => s, async (i, ct) => i.ToString(), async (s, e, ct) => s));
         }
 
+        [Fact]
+        public void Empty_WithoutResultSelector_ProducesEmpty()
+        {
+            Assert.Same(AsyncEnumerable.Empty<IGrouping<string, int>>(), AsyncEnumerable.Empty<string>().GroupJoin(CreateSource(1, 2, 3), s => s, i => i.ToString()));
+            Assert.Same(AsyncEnumerable.Empty<IGrouping<string, int>>(), AsyncEnumerable.Empty<string>().GroupJoin(CreateSource(1, 2, 3), async (s, ct) => s, async (i, ct) => i.ToString()));
+        }
+
         [Fact]
         public async Task VariousValues_MatchesEnumerable_String()
         {
@@ -55,6 +76,36 @@ await AssertEqual(
             }
         }
 
+        [Fact]
+        public async Task VariousValues_WithoutResultSelector_MatchesEnumerable()
+        {
+            int[] outer = [1, 2, 3];
+            int[] inner = [1, 2, 2, 3, 3, 3];
+
+            foreach (IAsyncEnumerable<int> outerSource in CreateSources(outer))
+            foreach (IAsyncEnumerable<int> innerSource in CreateSources(inner))
+            {
+                var expected = outer.GroupJoin(inner, o => o, i => i);
+                var result = await outerSource.GroupJoin(innerSource, o => o, i => i).ToListAsync();
+
+                Assert.Equal(expected.Count(), result.Count);
+                foreach (var (exp, act) in expected.Zip(result))
+                {
+                    Assert.Equal(exp.Key, act.Key);
+                    Assert.Equal(exp.ToList(), act.ToList());
+                }
+
+                var resultAsync = await outerSource.GroupJoin(innerSource, async (o, ct) => o, async (i, ct) => i).ToListAsync();
+
+                Assert.Equal(expected.Count(), resultAsync.Count);
+                foreach (var (exp, act) in expected.Zip(resultAsync))
+                {
+                    Assert.Equal(exp.Key, act.Key);
+                    Assert.Equal(exp.ToList(), act.ToList());
+                }
+            }
+        }
+
         [Fact]
         public async Task Cancellation_Cancels()
         {
@@ -167,5 +218,31 @@ public async Task InterfaceCalls_ExpectedCounts()
             Assert.Equal(4, inner.CurrentCount);
             Assert.Equal(1, inner.DisposeAsyncCount);
         }
+
+        [Fact]
+        public async Task InterfaceCalls_WithoutResultSelector_ExpectedCounts()
+        {
+            TrackingAsyncEnumerable<int> outer, inner;
+
+            outer = CreateSource(2, 4, 8, 16).Track();
+            inner = CreateSource(1, 2, 3, 4).Track();
+            await ConsumeAsync(outer.GroupJoin(inner, outer => outer, inner => inner));
+            Assert.Equal(5, outer.MoveNextAsyncCount);
+            Assert.Equal(4, outer.CurrentCount);
+            Assert.Equal(1, outer.DisposeAsyncCount);
+            Assert.Equal(5, inner.MoveNextAsyncCount);
+            Assert.Equal(4, inner.CurrentCount);
+            Assert.Equal(1, inner.DisposeAsyncCount);
+
+            outer = CreateSource(2, 4, 8, 16).Track();
+            inner = CreateSource(1, 2, 3, 4).Track();
+            await ConsumeAsync(outer.GroupJoin(inner, async (outer, ct) => outer, async (inner, ct) => inner));
+            Assert.Equal(5, outer.MoveNextAsyncCount);
+            Assert.Equal(4, outer.CurrentCount);
+            Assert.Equal(1, outer.DisposeAsyncCount);
+            Assert.Equal(5, inner.MoveNextAsyncCount);
+            Assert.Equal(4, inner.CurrentCount);
+            Assert.Equal(1, inner.DisposeAsyncCount);
+        }
     }
 }