From 1c3ccceb575b472d5541f7e10cde58ee86fc4279 Mon Sep 17 00:00:00 2001 From: electricessence <5899455+electricessence@users.noreply.github.com> Date: Thu, 12 Nov 2026 04:19:19 -0800 Subject: [PATCH 1/2] Select on IMappedQuery composes a projection without running the query The result is an IMappedQuery built over Read and ReadAsync, so a data class can return the query and the caller picks the terminal. Nothing runs until a terminal does. Co-Authored-By: Claude Sonnet 5.5 --- .../Core/Extensions/MappedQueryExtensions.cs | 46 +- Source/Core/ProjectedQuery.cs | 78 ++ Source/Core/PublicAPI.Unshipped.txt | 1 + .../MappedQuerySelectTests.cs | 1221 +++++++++++++++++ docsite/articles/mapping.md | 12 + docsite/articles/whats-new-11.md | 4 + docsite/samples/Mapping.cs | 35 + docsite/samples/Program.cs | 11 + 8 files changed, 1407 insertions(+), 1 deletion(-) create mode 100644 Source/Core/ProjectedQuery.cs create mode 100644 Tests/Open.Database.Extensions.Core.Tests/MappedQuerySelectTests.cs 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 7bf661a..f0a34b5 100644 --- a/Source/Core/PublicAPI.Unshipped.txt +++ b/Source/Core/PublicAPI.Unshipped.txt @@ -187,3 +187,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..8862e7c --- /dev/null +++ b/Tests/Open.Database.Extensions.Core.Tests/MappedQuerySelectTests.cs @@ -0,0 +1,1221 @@ +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 IMappedQuery 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