diff --git a/Source/Core/Extensions/MappedQueryExtensions.cs b/Source/Core/Extensions/MappedQueryExtensions.cs
index 6bcfc76..ca45046 100644
--- a/Source/Core/Extensions/MappedQueryExtensions.cs
+++ b/Source/Core/Extensions/MappedQueryExtensions.cs
@@ -1,7 +1,7 @@
namespace Open.Database.Extensions;
///
-/// Terminal operations for : they materialize the mapped rows into a collection or take a single row. and its skip and take variants expose the same operations directly.
+/// Terminal operations for : they materialize the mapped rows into a collection or take a single row, and projects them lazily. and its skip and take variants expose the same operations directly.
///
///
///
@@ -412,4 +412,48 @@ public static ValueTask> ToHashSetAsync(this IMappedQuery query
=> query.ReadAsync(
async (items, token) => await items.SingleOrDefaultAsync(token).ConfigureAwait(false),
cancellationToken);
+
+ ///
+ /// Projects each mapped row with , lazily: nothing runs until a terminal of the returned query runs.
+ ///
+ /// The type each row is mapped to.
+ /// The type returns.
+ /// The query to project.
+ /// Produces the projected value for each mapped row. It runs once per row, in row order, only while a terminal is reading the rows.
+ /// A deferred query of the projected rows. Every terminal in works on it, and so does Select again.
+ /// or is .
+ ///
+ ///
+ /// This call executes nothing and does not run . The command runs once for each terminal called on the returned query, exactly as it does on , so a method can return the projected query and let each caller pick the terminal (ToListAsync or FirstOrDefaultAsync, for example). A query read by two terminals executes the command twice.
+ ///
+ ///
+ /// The projection is built on and , so how opens and releases the connection, reads through later result sets and honors SingleResult() and cancellation is unchanged. Skip and Take apply to the rows of , before the selector runs: call them before Select. A terminal that stops early, such as FirstOrDefault, runs only for the rows it reads.
+ ///
+ ///
+ /// An exception thrown by propagates to the caller of the terminal, and the reader and the connection are released as they are when a handler throws. A projection of a projection reads through one reader.
+ ///
+ ///
+ ///
+ ///
+ /// static IMappedQuery<CustomerName> Names(IDbConnectionFactory<SqlConnection> factory)
+ /// => factory
+ /// .Command("SELECT Id, Name FROM Customers")
+ /// .Map<Customer>()
+ /// .Select(c => new CustomerName(c.Id, c.Name));
+ ///
+ /// List<CustomerName> all = await Names(factory).ToListAsync(cancellationToken);
+ /// CustomerName? first = await Names(factory).FirstOrDefaultAsync(cancellationToken);
+ ///
+ ///
+ ///
+ public static IMappedQuery Select(
+ this IMappedQuery query,
+ Func selector)
+ {
+ if (query is null) throw new ArgumentNullException(nameof(query));
+ if (selector is null) throw new ArgumentNullException(nameof(selector));
+ Contract.EndContractBlock();
+
+ return new ProjectedQuery(query, selector);
+ }
}
diff --git a/Source/Core/ProjectedQuery.cs b/Source/Core/ProjectedQuery.cs
new file mode 100644
index 0000000..a9377fc
--- /dev/null
+++ b/Source/Core/ProjectedQuery.cs
@@ -0,0 +1,78 @@
+namespace Open.Database.Extensions;
+
+// What MappedQueryExtensions.Select returns: a lazy projection of another
+// query. It holds the source and the selector and runs nothing itself.
+// Each terminal hands the handler a stream of the source's rows, projected one
+// at a time as the handler enumerates it, so a projection of a projection
+// still reads through one reader.
+// A caller sees only IMappedQuery, so this type stays internal.
+sealed class ProjectedQuery(
+ IMappedQuery source,
+ Func selector)
+ : IMappedQuery
+{
+ public void Read(Action> handler)
+ {
+ if (handler is null) throw new ArgumentNullException(nameof(handler));
+ Contract.EndContractBlock();
+
+ source.Read(rows => handler(Project(rows, selector)));
+ }
+
+ public TResult Read(Func, TResult> handler)
+ {
+ if (handler is null) throw new ArgumentNullException(nameof(handler));
+ Contract.EndContractBlock();
+
+ return source.Read(rows => handler(Project(rows, selector)));
+ }
+
+ public ValueTask ReadAsync(
+ Func, CancellationToken, ValueTask> handler,
+ CancellationToken cancellationToken)
+ {
+ if (handler is null) throw new ArgumentNullException(nameof(handler));
+ Contract.EndContractBlock();
+
+ return source.ReadAsync(
+ (rows, token) => handler(ProjectAsync(rows, selector, token), token),
+ cancellationToken);
+ }
+
+ public ValueTask ReadAsync(
+ Func, CancellationToken, ValueTask> handler,
+ CancellationToken cancellationToken)
+ {
+ if (handler is null) throw new ArgumentNullException(nameof(handler));
+ Contract.EndContractBlock();
+
+ return source.ReadAsync(
+ (rows, token) => handler(ProjectAsync(rows, selector, token), token),
+ cancellationToken);
+ }
+
+ // The foreach disposes the source's enumerator when this one is disposed,
+ // so a handler that stops early still releases the reader.
+ static IEnumerable Project(
+ IEnumerable rows,
+ Func selector)
+ {
+ foreach (TSource row in rows)
+ {
+ yield return selector(row);
+ }
+ }
+
+ static async IAsyncEnumerable ProjectAsync(
+ IAsyncEnumerable rows,
+ Func selector,
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ await foreach (TSource row in rows
+ .WithCancellation(cancellationToken)
+ .ConfigureAwait(false))
+ {
+ yield return selector(row);
+ }
+ }
+}
diff --git a/Source/Core/PublicAPI.Unshipped.txt b/Source/Core/PublicAPI.Unshipped.txt
index 637bd32..1b4b04a 100644
--- a/Source/Core/PublicAPI.Unshipped.txt
+++ b/Source/Core/PublicAPI.Unshipped.txt
@@ -189,3 +189,4 @@ static Open.Database.Extensions.ConnectionExtensions.UsingAsync(thi
Open.Database.Extensions.ExpressiveDbCommandBase.RetrieveAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask!>
Open.Database.Extensions.ExpressiveDbCommandBase.RetrieveAsync(System.Collections.Generic.IEnumerable! ordinals, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask!>
Open.Database.Extensions.ExpressiveDbCommandBase.RetrieveAsync(System.Collections.Generic.IEnumerable! columnNames, bool normalizeColumnOrder = false, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask!>
+static Open.Database.Extensions.MappedQueryExtensions.Select(this Open.Database.Extensions.IMappedQuery! query, System.Func! selector) -> Open.Database.Extensions.IMappedQuery!
diff --git a/Tests/Open.Database.Extensions.Core.Tests/MappedQuerySelectTests.cs b/Tests/Open.Database.Extensions.Core.Tests/MappedQuerySelectTests.cs
new file mode 100644
index 0000000..75d2473
--- /dev/null
+++ b/Tests/Open.Database.Extensions.Core.Tests/MappedQuerySelectTests.cs
@@ -0,0 +1,1219 @@
+using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
+using static Open.Database.Extensions.Tests.ReaderFixtures;
+
+#nullable enable
+
+namespace Open.Database.Extensions.Tests;
+
+// query.Select(selector) is a lazy projection of an IMappedQuery:
+// building it runs nothing, each terminal on it runs the command once,
+// and the selector runs once per row, in row order, only while the rows are read.
+// Every level that has a Map() (expressive command, raw command,
+// reader view, command view, table view) gets it.
+[ExcludeFromCodeCoverage]
+public static class MappedQuerySelectTests
+{
+ sealed record Person
+ {
+ public required int Id { get; init; }
+ public string? Name { get; init; }
+ }
+
+ sealed record Label(int Id, string Text);
+
+ static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+ #region Rigs: one way to reach a mapped query per level
+
+ // A mapped query, and a count of the work done against it:
+ // readers created for a command, rows read for a reader view.
+ // A table has nothing to count.
+ sealed class Rig : IDisposable
+ {
+ public required IMappedQuery Query { get; init; }
+ public Func? Work { get; init; }
+ public DataTable? Table { get; init; }
+
+ public void Dispose() => Table?.Dispose();
+ }
+
+ public static TheoryData Levels
+ => ["expressive", "execute", "reader", "command", "table"];
+
+ // The levels whose query can be read more than once.
+ public static TheoryData Repeatable
+ => ["expressive", "execute", "command", "table"];
+
+ static DataTable Table(int count)
+ {
+ DataTable table = new();
+ table.Columns.Add("Id", typeof(int));
+ table.Columns.Add("Name", typeof(string));
+ for (int i = 1; i <= count; i++)
+ {
+ table.Rows.Add(i, "n" + i);
+ }
+
+ return table;
+ }
+
+ static DbTally Tracked(int rows)
+ => new(new DbScript
+ {
+ Rows = [.. Enumerable.Range(1, rows).Select(i => (object?[])[i, "n" + i])],
+ });
+
+ static Rig Make(string level, int rows = 3)
+ {
+ switch (level)
+ {
+ case "expressive":
+ DbTally tally = Tracked(rows);
+ return new()
+ {
+ Query = tally.Factory.Command("select").Map(),
+ Work = () => tally.Readers.Count,
+ };
+
+ case "execute":
+ FakeExecuteReader execute = new(() => Reader(Rows(rows)));
+ return new()
+ {
+ Query = execute.Map(),
+ Work = () => execute.ExecuteCount,
+ };
+
+ case "reader":
+ TrackedReader reader = TrackedRows(rows);
+ return new()
+ {
+ Query = reader.Map(),
+ Work = () => reader.Reads,
+ };
+
+ case "command":
+ FakeDbCommand command = new(
+ new FakeDbConnection(),
+ () => Reader(Rows(rows)));
+ return new()
+ {
+ Query = command.Map(),
+ Work = () => command.Behaviors.Count,
+ };
+
+ case "table":
+ DataTable table = Table(rows);
+ return new() { Query = table.Map(), Table = table };
+
+ default:
+ throw new ArgumentOutOfRangeException(nameof(level));
+ }
+ }
+
+ // The selector the tests use: it notes which row it was given.
+ static Func Project(List? seen = null)
+ => p =>
+ {
+ seen?.Add(p.Id);
+ return $"{p.Id}:{p.Name}";
+ };
+
+ const string All = "1:n1,2:n2,3:n3";
+
+ static string Flat(object? value)
+ => value switch
+ {
+ null => "null",
+ string s => s,
+ IEnumerable items => string.Join(",", items),
+ _ => throw new ArgumentOutOfRangeException(nameof(value)),
+ };
+
+ #endregion
+
+ #region Nothing runs until a terminal runs
+
+ [Theory, MemberData(nameof(Levels))]
+ public static void BuildingAProjection_RunsNothing(string level)
+ {
+ using Rig rig = Make(level);
+ List seen = [];
+
+ IMappedQuery query = rig.Query.Select(Project(seen));
+ IMappedQuery chained = query.Select(s => s + "!");
+
+ Assert.Empty(seen);
+ if (rig.Work is { } work) Assert.Equal(0, work());
+
+ Assert.Equal("1:n1!,2:n2!,3:n3!", Flat(chained.ToList()));
+ Assert.Equal([1, 2, 3], seen);
+ if (rig.Work is { } done) Assert.True(done() > 0);
+ }
+
+ #endregion
+
+ #region Every terminal
+
+ static readonly string[] Terminals =
+ [
+ "ToList", "ToArray", "ToImmutableArray", "ToHashSet", "ToDictionary",
+ "FirstOrDefault", "Read", "ReadAction",
+ ];
+
+ static readonly string[] AsyncTerminals =
+ [
+ "ToListAsync", "ToArrayAsync", "ToImmutableArrayAsync",
+ "ToHashSetAsync", "ToDictionaryAsync", "FirstOrDefaultAsync",
+ "ReadAsync", "ReadAsyncAction",
+ ];
+
+ static TheoryData Cases(string[] terminals)
+ {
+ TheoryData data = [];
+ foreach (string level in new[] { "expressive", "execute", "reader", "command", "table" })
+ {
+ foreach (string terminal in terminals)
+ {
+ data.Add(level, terminal);
+ }
+ }
+
+ return data;
+ }
+
+ public static TheoryData SyncCases => Cases(Terminals);
+
+ public static TheoryData AsyncCases
+ => Cases(AsyncTerminals);
+
+ static string Expected(string terminal)
+ => terminal.StartsWith("FirstOrDefault", StringComparison.Ordinal)
+ ? "1:n1"
+ : All;
+
+ static object? Run(IMappedQuery query, string terminal)
+ => terminal switch
+ {
+ "ToList" => query.ToList(),
+ "ToArray" => query.ToArray(),
+ "ToImmutableArray" => query.ToImmutableArray(),
+ "ToHashSet" => query.ToHashSet(),
+ "ToDictionary" => query.ToDictionary(s => s).Keys,
+ "FirstOrDefault" => query.FirstOrDefault(),
+ "Read" => query.Read(items => items.ToList()),
+ "ReadAction" => ReadInto(query),
+ _ => throw new ArgumentOutOfRangeException(nameof(terminal)),
+ };
+
+ static List ReadInto(IMappedQuery query)
+ {
+ List got = [];
+ query.Read(items => got.AddRange(items));
+ return got;
+ }
+
+ static async ValueTask RunAsync(
+ IMappedQuery query,
+ string terminal,
+ CancellationToken token)
+ {
+ switch (terminal)
+ {
+ case "ToListAsync":
+ return await query.ToListAsync(token);
+ case "ToArrayAsync":
+ return await query.ToArrayAsync(token);
+ case "ToImmutableArrayAsync":
+ return await query.ToImmutableArrayAsync(token);
+ case "ToHashSetAsync":
+ return await query.ToHashSetAsync(null, token);
+ case "ToDictionaryAsync":
+ return (await query.ToDictionaryAsync(s => s, null, token)).Keys;
+ case "FirstOrDefaultAsync":
+ return await query.FirstOrDefaultAsync(token);
+ case "ReadAsync":
+ return await query.ReadAsync(
+ async (items, t) => await items.ToListAsync(t),
+ token);
+ case "ReadAsyncAction":
+ List got = [];
+ await query.ReadAsync(
+ async (items, t) =>
+ {
+ await foreach (string s in items.WithCancellation(t))
+ {
+ got.Add(s);
+ }
+ },
+ token);
+ return got;
+ default:
+ throw new ArgumentOutOfRangeException(nameof(terminal));
+ }
+ }
+
+ [Theory, MemberData(nameof(SyncCases))]
+ public static void EverySyncTerminal_ReturnsTheProjectedRows(
+ string level,
+ string terminal)
+ {
+ using Rig rig = Make(level);
+ List seen = [];
+
+ object? result = Run(rig.Query.Select(Project(seen)), terminal);
+
+ Assert.Equal(Expected(terminal), Flat(result));
+ }
+
+ [Theory, MemberData(nameof(AsyncCases))]
+ public static async Task EveryAsyncTerminal_ReturnsTheProjectedRows(
+ string level,
+ string terminal)
+ {
+ using Rig rig = Make(level);
+ List seen = [];
+
+ object? result = await RunAsync(
+ rig.Query.Select(Project(seen)),
+ terminal,
+ Ct);
+
+ Assert.Equal(Expected(terminal), Flat(result));
+ }
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task SingleOrDefault_ReadsTheOnlyRow_AndThrowsForMore(
+ string level)
+ {
+ using (Rig one = Make(level, 1))
+ {
+ Assert.Equal("1:n1", one.Query.Select(Project()).SingleOrDefault());
+ }
+
+ using (Rig one = Make(level, 1))
+ {
+ Assert.Equal(
+ "1:n1",
+ await one.Query.Select(Project()).SingleOrDefaultAsync(Ct));
+ }
+
+ using (Rig many = Make(level))
+ {
+ Assert.Throws(
+ () => many.Query.Select(Project()).SingleOrDefault());
+ }
+
+ using (Rig many = Make(level))
+ {
+ await Assert.ThrowsAsync(
+ async () => await many.Query.Select(Project()).SingleOrDefaultAsync(Ct));
+ }
+ }
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task NoRows_GiveNothing(string level)
+ {
+ using (Rig none = Make(level, 0))
+ {
+ IMappedQuery query = none.Query.Select(Project());
+ Assert.Empty(query.ToList());
+ }
+
+ using (Rig none = Make(level, 0))
+ {
+ IMappedQuery query = none.Query.Select(Project());
+ Assert.Null(query.FirstOrDefault());
+ }
+
+ using (Rig none = Make(level, 0))
+ {
+ IMappedQuery query = none.Query.Select(Project());
+ Assert.Empty(await query.ToArrayAsync(Ct));
+ }
+
+ using (Rig none = Make(level, 0))
+ {
+ IMappedQuery query = none.Query.Select(Project());
+ Assert.Null(await query.FirstOrDefaultAsync(Ct));
+ }
+ }
+
+ #endregion
+
+ #region The selector: once per row, in row order, only while reading
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task TheSelector_RunsOncePerRow_InRowOrder(string level)
+ {
+ using (Rig rig = Make(level, 4))
+ {
+ List seen = [];
+ rig.Query.Select(Project(seen)).ToArray();
+ Assert.Equal([1, 2, 3, 4], seen);
+ }
+
+ using (Rig rig = Make(level, 4))
+ {
+ List seen = [];
+ await rig.Query.Select(Project(seen)).ToListAsync(Ct);
+ Assert.Equal([1, 2, 3, 4], seen);
+ }
+ }
+
+ // The projection streams: the selector for a row runs when the handler
+ // asks for it, not before the handler starts.
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task TheSelector_RunsAsTheHandlerEnumerates(
+ string level)
+ {
+ using (Rig rig = Make(level))
+ {
+ List log = [];
+ rig.Query
+ .Select(p =>
+ {
+ log.Add("select" + p.Id);
+ return p.Id;
+ })
+ .Read(items =>
+ {
+ log.Add("start");
+ foreach (int id in items)
+ {
+ log.Add("got" + id);
+ }
+ });
+
+ Assert.Equal(
+ "start,select1,got1,select2,got2,select3,got3",
+ string.Join(",", log));
+ }
+
+ using (Rig rig = Make(level))
+ {
+ List log = [];
+ await rig.Query
+ .Select(p =>
+ {
+ log.Add("select" + p.Id);
+ return p.Id;
+ })
+ .ReadAsync(
+ async (items, token) =>
+ {
+ log.Add("start");
+ await foreach (int id in items.WithCancellation(token))
+ {
+ log.Add("got" + id);
+ }
+ },
+ Ct);
+
+ Assert.Equal(
+ "start,select1,got1,select2,got2,select3,got3",
+ string.Join(",", log));
+ }
+ }
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task AnEarlyExit_RunsTheSelectorOnlyForTheRowsRead(
+ string level)
+ {
+ using (Rig rig = Make(level))
+ {
+ List seen = [];
+ Assert.Equal(
+ "1:n1",
+ rig.Query.Select(Project(seen)).FirstOrDefault());
+ Assert.Equal([1], seen);
+ }
+
+ using (Rig rig = Make(level))
+ {
+ List seen = [];
+ Assert.Equal(
+ "1:n1",
+ await rig.Query.Select(Project(seen)).FirstOrDefaultAsync(Ct));
+ Assert.Equal([1], seen);
+ }
+
+ using (Rig rig = Make(level, 5))
+ {
+ List seen = [];
+ string taken = rig.Query
+ .Select(Project(seen))
+ .Read(items => string.Join(",", items.Take(2)));
+ Assert.Equal("1:n1,2:n2", taken);
+ Assert.Equal([1, 2], seen);
+ }
+
+ using (Rig rig = Make(level, 5))
+ {
+ List seen = [];
+ string taken = await rig.Query
+ .Select(Project(seen))
+ .ReadAsync(
+ async (items, token) =>
+ {
+ List got = [];
+ await foreach (string s in items.WithCancellation(token))
+ {
+ got.Add(s);
+ if (got.Count == 2) break;
+ }
+
+ return string.Join(",", got);
+ },
+ Ct);
+ Assert.Equal("1:n1,2:n2", taken);
+ Assert.Equal([1, 2], seen);
+ }
+ }
+
+ #endregion
+
+ #region Skip, Take and SingleResult belong to the source
+
+ [Fact]
+ public static async Task SkipAndTake_ApplyToTheSourceRows_BeforeTheSelector()
+ {
+ List seen = [];
+ FakeExecuteReader command = new(() => Reader(Rows(5)));
+
+ Assert.Equal(
+ "2:n2,3:n3",
+ Flat(command.Map().Skip(1).Take(2).Select(Project(seen)).ToList()));
+ Assert.Equal([2, 3], seen);
+
+ seen.Clear();
+ Assert.Equal(
+ "3:n3,4:n4,5:n5",
+ Flat(command.Map().Skip(2).Select(Project(seen)).ToList()));
+ Assert.Equal([3, 4, 5], seen);
+
+ seen.Clear();
+ Assert.Equal(
+ "1:n1,2:n2",
+ Flat(await command.Map().Take(2).Select(Project(seen)).ToListAsync(Ct)));
+ Assert.Equal([1, 2], seen);
+
+ seen.Clear();
+ using DataTable table = Table(5);
+ Assert.Equal(
+ "4:n4",
+ Flat(table.Map().Skip(3).Take(1).Select(Project(seen)).ToList()));
+ Assert.Equal([4], seen);
+
+ seen.Clear();
+ Assert.Empty(command.Map().Take(0).Select(Project(seen)).ToList());
+ Assert.Empty(seen);
+ }
+
+ [Fact]
+ public static void SingleResult_IsStillTheSourceQuerysChoice()
+ {
+ FakeExecuteReader command = new(() => Reader(Rows(2)));
+
+ command.Map().Select(Project()).ToList();
+ command.Map().SingleResult().Select(Project()).ToList();
+
+ Assert.Equal(2, command.Behaviors.Count);
+ Assert.False(command.Behaviors[0].HasFlag(CommandBehavior.SingleResult));
+ Assert.True(command.Behaviors[1].HasFlag(CommandBehavior.SingleResult));
+ }
+
+ const int LaterSets = 2;
+
+ static FakeExecuteReader MultiSet(
+ int rows,
+ out Func reader,
+ int? failAt = null)
+ {
+ MultiSetReader? last = null;
+ reader = () => last ?? throw new InvalidOperationException("never executed");
+ return new FakeExecuteReader(
+ () => last = new MultiSetReader(Rows(rows), LaterSets, failAt));
+ }
+
+ // The projection sits on Read: reading to the end reads through the later
+ // result sets, as the source query does, and a failure there is thrown.
+ [Fact]
+ public static async Task AFailureInALaterResultSet_IsStillThrown_AfterTheLastRow()
+ {
+ FakeExecuteReader command = MultiSet(2, out Func reader);
+ Assert.Equal(2, command.Map().Select(Project()).ToList().Count);
+ Assert.Equal(LaterSets + 1, reader().NextResultCalls);
+ Assert.Equal(0, reader().ReadsAfterNextResult);
+
+ command = MultiSet(2, out reader);
+ Assert.Equal(
+ 2,
+ (await command.Map().Select(Project()).ToListAsync(Ct)).Count);
+ Assert.Equal(LaterSets + 1, reader().NextResultCalls);
+
+ command = MultiSet(2, out _, failAt: 1);
+ Assert.Throws(
+ () => command.Map().Select(Project()).ToList());
+
+ command = MultiSet(2, out _, failAt: 1);
+ await Assert.ThrowsAsync(
+ async () => await command.Map().Select(Project()).ToListAsync(Ct));
+ }
+
+ [Fact]
+ public static async Task NoReadThrough_AfterSingleResult_OrAnEarlyExit()
+ {
+ FakeExecuteReader command = MultiSet(2, out Func reader, failAt: 1);
+ Assert.Equal(
+ 2,
+ command.Map().SingleResult().Select(Project()).ToList().Count);
+ Assert.Equal(0, reader().NextResultCalls);
+
+ command = MultiSet(2, out reader, failAt: 1);
+ Assert.Equal(
+ "1:n1",
+ command.Map().Select(Project()).FirstOrDefault());
+ Assert.Equal(0, reader().NextResultCalls);
+
+ command = MultiSet(2, out reader, failAt: 1);
+ Assert.Equal(
+ "1:n1",
+ await command.Map().Select(Project()).FirstOrDefaultAsync(Ct));
+ Assert.Equal(0, reader().NextResultCalls);
+ }
+
+ #endregion
+
+ #region Failure and release
+
+ // The reader, the command and the connection are released exactly as
+ // they are when the caller's own handler throws or stops early.
+ static void AssertReleased(DbTally tally)
+ {
+ LeakAssert.CommandsAndReadersDisposed(tally);
+ LeakAssert.ConnectionsReleased(tally, LeakSource.Factory, tally.Pool, 1);
+ }
+
+ static MappedQuery Expressive(DbTally tally)
+ => tally.Factory.Command("select").Map();
+
+ static string Boom(Person p)
+ => p.Id == 2
+ ? throw new InvalidOperationException("boom")
+ : p.Name!;
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task ASelectorThatThrows_ThrowsFromTheTerminal(string level)
+ {
+ using (Rig rig = Make(level))
+ {
+ InvalidOperationException ex = Assert.Throws(
+ () => rig.Query.Select(Boom).ToList());
+ Assert.Equal("boom", ex.Message);
+ }
+
+ using (Rig rig = Make(level))
+ {
+ InvalidOperationException ex = await Assert.ThrowsAsync(
+ async () => await rig.Query.Select(Boom).ToListAsync(Ct));
+ Assert.Equal("boom", ex.Message);
+ }
+
+ using (Rig rig = Make(level))
+ {
+ // Not the second row: the selector was never asked for it.
+ Assert.Equal("n1", rig.Query.Select(Boom).FirstOrDefault());
+ }
+ }
+
+ [Fact]
+ public static async Task ASelectorThatThrows_ReleasesEverything()
+ {
+ DbTally syncTally = Tracked(3);
+ Assert.Throws(
+ () => Expressive(syncTally).Select(Boom).ToList());
+ Assert.Equal(2, Assert.Single(syncTally.Readers).Reads);
+ AssertReleased(syncTally);
+
+ DbTally asyncTally = Tracked(3);
+ await Assert.ThrowsAsync(
+ async () => await Expressive(asyncTally).Select(Boom).ToListAsync(Ct));
+ AssertReleased(asyncTally);
+ }
+
+ [Fact]
+ public static async Task AnEarlyExit_ReleasesEverything()
+ {
+ DbTally first = Tracked(3);
+ Assert.Equal("1:n1", Expressive(first).Select(Project()).FirstOrDefault());
+ Assert.Equal(1, Assert.Single(first.Readers).Reads);
+ AssertReleased(first);
+
+ DbTally firstAsync = Tracked(3);
+ Assert.Equal(
+ "1:n1",
+ await Expressive(firstAsync).Select(Project()).FirstOrDefaultAsync(Ct));
+ Assert.Equal(1, Assert.Single(firstAsync.Readers).Reads);
+ AssertReleased(firstAsync);
+
+ DbTally taken = Tracked(5);
+ Assert.Equal(
+ "1:n1,2:n2",
+ Expressive(taken)
+ .Select(Project())
+ .Read(items => string.Join(",", items.Take(2))));
+ Assert.Equal(2, Assert.Single(taken.Readers).Reads);
+ AssertReleased(taken);
+ }
+
+ [Fact]
+ public static async Task ATerminalThatCompletes_ReleasesEverything()
+ {
+ DbTally syncTally = Tracked(3);
+ Expressive(syncTally).Select(Project()).ToArray();
+ AssertReleased(syncTally);
+
+ DbTally asyncTally = Tracked(3);
+ await Expressive(asyncTally).Select(Project()).ToArrayAsync(Ct);
+ AssertReleased(asyncTally);
+ }
+
+ [Fact]
+ public static async Task AHandlerThatThrows_ReleasesEverything()
+ {
+ DbTally syncTally = Tracked(3);
+ Assert.Throws(
+ () => Expressive(syncTally)
+ .Select(Project())
+ .Read(items =>
+ {
+ _ = items.First();
+ throw new FormatException();
+ }));
+ AssertReleased(syncTally);
+
+ DbTally asyncTally = Tracked(3);
+ await Assert.ThrowsAsync(
+ async () => await Expressive(asyncTally)
+ .Select(Project())
+ .ReadAsync(
+ async (items, token) =>
+ {
+ _ = await items.FirstAsync(token);
+ throw new FormatException();
+ },
+ Ct));
+ AssertReleased(asyncTally);
+ }
+
+ #endregion
+
+ #region Cancellation
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task ACanceledToken_CancelsTheCall_BeforeTheSelectorRuns(
+ string level)
+ {
+ using Rig rig = Make(level);
+ List seen = [];
+ using CancellationTokenSource cts = new();
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync(
+ async () => await rig.Query.Select(Project(seen)).ToListAsync(cts.Token));
+ Assert.Empty(seen);
+ }
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task CancelingWhileRowsAreRead_StopsTheStream(
+ string level)
+ {
+ using Rig rig = Make(level, 5);
+ using CancellationTokenSource cts = new();
+ List seen = [];
+
+ await Assert.ThrowsAnyAsync(
+ async () => await rig.Query
+ .Select(p =>
+ {
+ seen.Add(p.Id);
+ if (p.Id == 2) cts.Cancel();
+ return p.Id;
+ })
+ .ToListAsync(cts.Token));
+
+ Assert.Equal([1, 2], seen);
+ }
+
+ [Fact]
+ public static async Task CancelingWhileRowsAreRead_ReleasesEverything()
+ {
+ DbTally tally = Tracked(5);
+ using CancellationTokenSource cts = new();
+
+ await Assert.ThrowsAnyAsync(
+ async () => await Expressive(tally)
+ .Select(p =>
+ {
+ if (p.Id == 2) cts.Cancel();
+ return p.Id;
+ })
+ .ToListAsync(cts.Token));
+
+ AssertReleased(tally);
+ }
+
+ // The handler is given the token in effect, as the source query gives it.
+ [Fact]
+ public static async Task TheHandler_IsGivenTheToken_AndCanPassItOn()
+ {
+ FakeExecuteReader command = new(() => Reader(Rows(3)));
+ using CancellationTokenSource cts = new();
+ CancellationToken given = default;
+
+ string joined = await command
+ .Map()
+ .Select(Project())
+ .ReadAsync(
+ async (items, token) =>
+ {
+ given = token;
+ List got = [];
+ await foreach (string s in items.WithCancellation(token))
+ {
+ got.Add(s);
+ }
+
+ return string.Join(",", got);
+ },
+ cts.Token);
+
+ Assert.Equal(All, joined);
+ Assert.Equal(cts.Token, given);
+ }
+
+ // The token of the command and the one of the call both reach the rows.
+ [Fact]
+ public static async Task TheCommandsOwnToken_AlsoCancelsTheStream()
+ {
+ using CancellationTokenSource cts = new();
+ FakeExecuteReader command = new(() => Reader(Rows(5)))
+ {
+ CancellationToken = cts.Token,
+ };
+ List seen = [];
+
+ await Assert.ThrowsAnyAsync(
+ async () => await command
+ .Map()
+ .Select(p =>
+ {
+ seen.Add(p.Id);
+ if (p.Id == 3) cts.Cancel();
+ return p.Id;
+ })
+ .ToListAsync(Ct));
+
+ Assert.Equal([1, 2, 3], seen);
+ }
+
+ #endregion
+
+ #region Composing
+
+ [Theory, MemberData(nameof(Levels))]
+ public static async Task Select_Chains_AndStillReadsThroughOneReader(
+ string level)
+ {
+ using (Rig rig = Make(level))
+ {
+ List log = [];
+ IMappedQuery query = rig.Query
+ .Select(p =>
+ {
+ log.Add("a" + p.Id);
+ return p.Id;
+ })
+ .Select(id =>
+ {
+ log.Add("b" + id);
+ return id * 10;
+ })
+ .Select(n => $"#{n}");
+
+ Assert.Equal("#10,#20,#30", Flat(query.ToList()));
+ Assert.Equal("a1,b1,a2,b2,a3,b3", string.Join(",", log));
+ if (rig.Work is { } work && level != "reader") Assert.Equal(1, work());
+ }
+
+ using (Rig rig = Make(level))
+ {
+ IMappedQuery query = rig.Query
+ .Select(p => p.Id)
+ .Select(id => id * 10)
+ .Select(n => $"#{n}");
+
+ Assert.Equal("#10,#20,#30", Flat(await query.ToListAsync(Ct)));
+ }
+ }
+
+ [Theory, MemberData(nameof(Repeatable))]
+ public static async Task TheSameProjection_RunsTheCommandAgainForEachTerminal(
+ string level)
+ {
+ using Rig rig = Make(level);
+ List seen = [];
+ IMappedQuery query = rig.Query.Select(Project(seen));
+
+ Assert.Equal(All, Flat(query.ToList()));
+ Assert.Equal(All, Flat(await query.ToListAsync(Ct)));
+ Assert.Equal("1:n1", query.FirstOrDefault());
+
+ Assert.Equal([1, 2, 3, 1, 2, 3, 1], seen);
+ if (rig.Work is { } work) Assert.Equal(3, work());
+ }
+
+ [Fact]
+ public static void ASelectedQuery_OfADefaultStruct_FailsAtTheTerminal_NotAtSelect()
+ {
+ IMappedQuery query = default(MappedQuery);
+
+ IMappedQuery projected = query.Select(i => i + 1);
+
+ Assert.Throws(() => projected.ToList());
+ }
+
+ #endregion
+
+ #region Arguments
+
+ [Fact]
+ public static void ANullQueryOrSelector_ThrowsAtTheSelectCall()
+ {
+ IMappedQuery? none = null;
+ using Rig rig = Make("execute");
+
+ Assert.Equal(
+ "query",
+ Assert.Throws(
+ () => none!.Select(p => p.Id)).ParamName);
+ Assert.Equal(
+ "selector",
+ Assert.Throws(
+ () => rig.Query.Select((Func)null!)).ParamName);
+ Assert.Equal(0, rig.Work!());
+ }
+
+ // The projected query guards its own entry points once,
+ // before anything is executed.
+ [Fact]
+ public static async Task ANullHandler_ThrowsAtTheCall_AndRunsNothing()
+ {
+ using Rig rig = Make("execute");
+ IMappedQuery query = rig.Query.Select(p => p.Id);
+
+ Assert.Equal(
+ "handler",
+ Assert.Throws(
+ () => query.Read((Action>)null!)).ParamName);
+ Assert.Equal(
+ "handler",
+ Assert.Throws(
+ () => query.Read((Func, int>)null!)).ParamName);
+ ArgumentNullException noHandler = await Assert.ThrowsAsync(
+ async () => await query.ReadAsync(
+ (Func, CancellationToken, ValueTask>)null!,
+ Ct));
+ Assert.Equal("handler", noHandler.ParamName);
+ noHandler = await Assert.ThrowsAsync(
+ async () => await query.ReadAsync(
+ (Func, CancellationToken, ValueTask>)null!,
+ Ct));
+ Assert.Equal("handler", noHandler.ParamName);
+ Assert.Equal(0, rig.Work!());
+ }
+
+ #endregion
+
+ #region A query factory
+
+ // A data class returns the query; each caller picks the terminal.
+ static IMappedQuery