Issue #1239 resolution - QueryAsync missing buffered parameter

Dapper.EntityFramework.StrongName/Dapper.EntityFramework.StrongName.csproj

@@ -4,7 +4,7 @@
     <Title>Dapper: Entity Framework type handlers (with a strong name)</Title>
     <Description>Extension handlers for entity framework</Description>
     <Authors>Marc Gravell;Nick Craver</Authors>
-    <TargetFrameworks>net461</TargetFrameworks>
+    <TargetFrameworks>net462</TargetFrameworks>
     <AssemblyOriginatorKeyFile>../Dapper.snk</AssemblyOriginatorKeyFile>
     <SignAssembly>true</SignAssembly>
     <PublicSign Condition=" '$(OS)' != 'Windows_NT' ">true</PublicSign>

Dapper.EntityFramework/Dapper.EntityFramework.csproj

@@ -5,7 +5,7 @@
     <AssemblyTitle>Dapper entity framework type handlers</AssemblyTitle>
     <VersionPrefix>1.50.2</VersionPrefix>
     <Authors>Marc Gravell;Nick Craver</Authors>
-    <TargetFrameworks>net461</TargetFrameworks>
+    <TargetFrameworks>net462</TargetFrameworks>
     <PackageTags>orm;sql;micro-orm</PackageTags>
   </PropertyGroup>
   <ItemGroup>

Dapper.Rainbow/Dapper.Rainbow.csproj

@@ -6,17 +6,18 @@
     <Description>Trivial micro-orm implemented on Dapper, provides with CRUD helpers.</Description>
     <Authors>Sam Saffron</Authors>
     <Copyright>2017 Sam Saffron</Copyright>
-    <TargetFrameworks>net461;netstandard2.0;net5.0</TargetFrameworks>
+    <TargetFrameworks>net48;netstandard2.1;net5.0</TargetFrameworks>
     <!-- TODO: Docs -->
     <TreatWarningsAsErrors>false</TreatWarningsAsErrors>
   </PropertyGroup>
   <ItemGroup>
     <ProjectReference Include="..\Dapper\Dapper.csproj" />
+    <PackageReference Include="System.Linq.Async" Version="6.0.1" />
   </ItemGroup>
-  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
+  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.1'">
     <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
   </ItemGroup>
-  <ItemGroup Condition=" '$(TargetFramework)' == 'net461' ">
+  <ItemGroup Condition=" '$(TargetFramework)' == 'net48' ">
     <Reference Include="Microsoft.CSharp" />
   </ItemGroup>
 </Project>

Dapper.Rainbow/Database.Async.cs

@@ -26,7 +26,7 @@ public partial class Table<T, TId>
                 string colsParams = string.Join(",", paramNames.Select(p => "@" + p));
                 var sql = "set nocount on insert " + TableName + " (" + cols + ") values (" + colsParams + ") select cast(scope_identity() as int)";
 
-                return (await database.QueryAsync<int?>(sql, o).ConfigureAwait(false)).Single();
+                return await database.QueryAsync<int?>(sql, o).SingleAsync().ConfigureAwait(false);
             }
 
             /// <summary>
@@ -77,7 +77,7 @@ public virtual Task<T> FirstAsync() =>
             /// Asynchronously gets the all rows from this table.
             /// </summary>
             /// <returns>Data from all table rows.</returns>
-            public Task<IEnumerable<T>> AllAsync() =>
+            public IAsyncEnumerable<T> AllAsync() =>
                 database.QueryAsync<T>("select * from " + TableName);
         }
 
@@ -97,7 +97,7 @@ public Task<int> ExecuteAsync(string sql, dynamic param = null) =>
         /// <param name="sql">The SQL to execute.</param>
         /// <param name="param">The parameters to use.</param>
         /// <returns>An enumerable of <typeparamref name="T"/> for the rows fetched.</returns>
-        public Task<IEnumerable<T>> QueryAsync<T>(string sql, dynamic param = null) =>
+        public IAsyncEnumerable<T> QueryAsync<T>(string sql, dynamic param = null) =>
             _connection.QueryAsync<T>(sql, param as object, _transaction, _commandTimeout);
 
         /// <summary>
@@ -194,7 +194,7 @@ public Task<IEnumerable<TReturn>> QueryAsync<TFirst, TSecond, TThird, TFourth, T
         /// <param name="sql">The SQL to execute.</param>
         /// <param name="param">The parameters to use.</param>
         /// <remarks>Note: each row can be accessed via "dynamic", or by casting to an IDictionary&lt;string,object&gt;</remarks>
-        public Task<IEnumerable<dynamic>> QueryAsync(string sql, dynamic param = null) =>
+        public IAsyncEnumerable<dynamic> QueryAsync(string sql, dynamic param = null) =>
             _connection.QueryAsync(sql, param as object, _transaction);
 
         /// <summary>

Dapper.SqlBuilder/Dapper.SqlBuilder.csproj

@@ -5,17 +5,17 @@
     <Title>Dapper SqlBuilder component</Title>
     <Description>The Dapper SqlBuilder component, for building SQL queries dynamically.</Description>
     <Authors>Sam Saffron, Johan Danforth</Authors>
-    <TargetFrameworks>net461;netstandard2.0;net5.0</TargetFrameworks>
+    <TargetFrameworks>net462;netstandard2.1;net5.0</TargetFrameworks>
     <GenerateDocumentationFile>false</GenerateDocumentationFile>
     <TreatWarningsAsErrors>false</TreatWarningsAsErrors>
   </PropertyGroup>
   <ItemGroup>
     <ProjectReference Include="..\Dapper\Dapper.csproj" />
   </ItemGroup>
-  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
+  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.1'">
     <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
   </ItemGroup>
-  <ItemGroup Condition=" '$(TargetFramework)' == 'net461' ">
+  <ItemGroup Condition=" '$(TargetFramework)' == 'net462' ">
     <Reference Include="Microsoft.CSharp" />
   </ItemGroup>
 </Project>

Dapper.StrongName/Dapper.StrongName.csproj

@@ -5,7 +5,7 @@
     <Title>Dapper (Strong Named)</Title>
     <Description>A high performance Micro-ORM supporting SQL Server, MySQL, Sqlite, SqlCE, Firebird etc..</Description>
     <Authors>Sam Saffron;Marc Gravell;Nick Craver</Authors>
-    <TargetFrameworks>net461;netstandard2.0;net5.0</TargetFrameworks>
+    <TargetFrameworks>net462;netstandard2.1;net5.0</TargetFrameworks>
     <SignAssembly>true</SignAssembly>
     <PublicSign Condition=" '$(OS)' != 'Windows_NT' ">true</PublicSign>
   </PropertyGroup>
@@ -18,7 +18,11 @@
     <DefineConstants>$(DefineConstants);PLAT_NO_REMOTING;PLAT_SKIP_LOCALS_INIT</DefineConstants>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
-  <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.0'">
+  <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.1'">
     <PackageReference Include="System.Reflection.Emit.Lightweight" Version="4.7.0" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
+  </ItemGroup>
+  <ItemGroup Condition=" '$(TargetFramework)' == 'net462'">
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
   </ItemGroup>
 </Project>

Dapper/Dapper.csproj

@@ -5,7 +5,7 @@
     <PackageTags>orm;sql;micro-orm</PackageTags>
     <Description>A high performance Micro-ORM supporting SQL Server, MySQL, Sqlite, SqlCE, Firebird etc..</Description>
     <Authors>Sam Saffron;Marc Gravell;Nick Craver</Authors>
-    <TargetFrameworks>net461;netstandard2.0;net5.0</TargetFrameworks>
+    <TargetFrameworks>net462;netstandard2.1;net5.0</TargetFrameworks>
   </PropertyGroup>
   <ItemGroup>
     <Folder Include="Properties\" />
@@ -15,7 +15,11 @@
     <DefineConstants>$(DefineConstants);PLAT_NO_REMOTING;PLAT_SKIP_LOCALS_INIT</DefineConstants>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
-  <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.0'">
+  <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.1'">
     <PackageReference Include="System.Reflection.Emit.Lightweight" Version="4.7.0" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
+  </ItemGroup>
+  <ItemGroup Condition=" '$(TargetFramework)' == 'net462'">
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
   </ItemGroup>
 </Project>

Dapper/SqlMapper.Async.cs

@@ -19,19 +19,20 @@ public static partial class SqlMapper
         /// <param name="sql">The SQL to execute for the query.</param>
         /// <param name="param">The parameters to pass, if any.</param>
         /// <param name="transaction">The transaction to use, if any.</param>
+        /// <param name="buffered">Whether to buffer the results in memory.</param>
         /// <param name="commandTimeout">The command timeout (in seconds).</param>
         /// <param name="commandType">The type of command to execute.</param>
         /// <remarks>Note: each row can be accessed via "dynamic", or by casting to an IDictionary&lt;string,object&gt;</remarks>
-        public static Task<IEnumerable<dynamic>> QueryAsync(this IDbConnection cnn, string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null) =>
-            QueryAsync<dynamic>(cnn, typeof(DapperRow), new CommandDefinition(sql, param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default));
+        public static IAsyncEnumerable<dynamic> QueryAsync(this IDbConnection cnn, string sql, object param = null, IDbTransaction transaction = null, bool buffered = true, int ? commandTimeout = null, CommandType? commandType = null) =>
+            QueryAsync<dynamic>(cnn, typeof(DapperRow), new CommandDefinition(sql, param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default));
 
         /// <summary>
         /// Execute a query asynchronously using Task.
         /// </summary>
         /// <param name="cnn">The connection to query on.</param>
         /// <param name="command">The command used to query on this connection.</param>
         /// <remarks>Note: each row can be accessed via "dynamic", or by casting to an IDictionary&lt;string,object&gt;</remarks>
-        public static Task<IEnumerable<dynamic>> QueryAsync(this IDbConnection cnn, CommandDefinition command) =>
+        public static IAsyncEnumerable<dynamic> QueryAsync(this IDbConnection cnn, CommandDefinition command) =>
             QueryAsync<dynamic>(cnn, typeof(DapperRow), command);
 
         /// <summary>
@@ -84,7 +85,7 @@ public static Task<dynamic> QuerySingleOrDefaultAsync(this IDbConnection cnn, Co
         /// A sequence of data of <typeparamref name="T"/>; if a basic type (int, string, etc) is queried then the data from the first column is assumed, otherwise an instance is
         /// created per row, and a direct column-name===member-name mapping is assumed (case insensitive).
         /// </returns>
-        public static Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn, string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null) =>
+        public static IAsyncEnumerable<T> QueryAsync<T>(this IDbConnection cnn, string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null) =>
             QueryAsync<T>(cnn, typeof(T), new CommandDefinition(sql, param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default));
 
         /// <summary>
@@ -195,13 +196,14 @@ public static Task<dynamic> QuerySingleOrDefaultAsync(this IDbConnection cnn, st
         /// <param name="sql">The SQL to execute for the query.</param>
         /// <param name="param">The parameters to pass, if any.</param>
         /// <param name="transaction">The transaction to use, if any.</param>
+        /// <param name="buffered">Whether to buffer the results in memory.</param>
         /// <param name="commandTimeout">The command timeout (in seconds).</param>
         /// <param name="commandType">The type of command to execute.</param>
         /// <exception cref="ArgumentNullException"><paramref name="type"/> is <c>null</c>.</exception>
-        public static Task<IEnumerable<object>> QueryAsync(this IDbConnection cnn, Type type, string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null)
+        public static IAsyncEnumerable<object> QueryAsync(this IDbConnection cnn, Type type, string sql, object param = null, IDbTransaction transaction = null, bool buffered = true, int ? commandTimeout = null, CommandType? commandType = null)
         {
             if (type == null) throw new ArgumentNullException(nameof(type));
-            return QueryAsync<object>(cnn, type, new CommandDefinition(sql, param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default));
+            return QueryAsync<object>(cnn, type, new CommandDefinition(sql, param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default));
         }
 
         /// <summary>
@@ -279,7 +281,7 @@ public static Task<object> QuerySingleOrDefaultAsync(this IDbConnection cnn, Typ
         /// A sequence of data of <typeparamref name="T"/>; if a basic type (int, string, etc) is queried then the data from the first column is assumed, otherwise an instance is
         /// created per row, and a direct column-name===member-name mapping is assumed (case insensitive).
         /// </returns>
-        public static Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn, CommandDefinition command) =>
+        public static IAsyncEnumerable<T> QueryAsync<T>(this IDbConnection cnn, CommandDefinition command) =>
             QueryAsync<T>(cnn, typeof(T), command);
 
         /// <summary>
@@ -288,7 +290,7 @@ public static Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn, Command
         /// <param name="cnn">The connection to query on.</param>
         /// <param name="type">The type to return.</param>
         /// <param name="command">The command used to query on this connection.</param>
-        public static Task<IEnumerable<object>> QueryAsync(this IDbConnection cnn, Type type, CommandDefinition command) =>
+        public static IAsyncEnumerable<object> QueryAsync(this IDbConnection cnn, Type type, CommandDefinition command) =>
             QueryAsync<object>(cnn, type, command);
 
         /// <summary>
@@ -403,7 +405,7 @@ private static DbCommand TrySetupAsyncCommand(this CommandDefinition command, ID
             }
         }
 
-        private static async Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn, Type effectiveType, CommandDefinition command)
+        private static async IAsyncEnumerable<T> QueryAsync<T>(this IDbConnection cnn, Type effectiveType, CommandDefinition command)
         {
             object param = command.Parameters;
             var identity = new Identity(command.CommandText, command.CommandType, cnn, effectiveType, param?.GetType());
@@ -412,6 +414,7 @@ private static async Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn,
             var cancel = command.CancellationToken;
             using var cmd = command.TrySetupAsyncCommand(cnn, info.ParamReader);
             DbDataReader reader = null;
+
             try
             {
                 if (wasClosed) await cnn.TryOpenAsync(cancel).ConfigureAwait(false);
@@ -422,7 +425,7 @@ private static async Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn,
                 if (tuple.Func == null || tuple.Hash != hash)
                 {
                     if (reader.FieldCount == 0)
-                        return Enumerable.Empty<T>();
+                        yield break;
                     tuple = info.Deserializer = new DeserializerState(hash, GetDeserializer(effectiveType, reader, 0, -1, false));
                     if (command.AddToCache) SetQueryCache(identity, info);
                 }
@@ -433,27 +436,52 @@ private static async Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn,
                 {
                     var buffer = new List<T>();
                     var convertToType = Nullable.GetUnderlyingType(effectiveType) ?? effectiveType;
-                    while (await reader.ReadAsync(cancel).ConfigureAwait(false))
+
+#if (NETSTANDARD2_1_OR_GREATER || NETCOREAPP3_0_OR_GREATER)
+                    await using (reader)
+#else
+                    using (reader)
+#endif
                     {
-                        object val = func(reader);
-                        buffer.Add(GetValue<T>(reader, effectiveType, val));
+                        while (await reader.ReadAsync(cancel).ConfigureAwait(false))
+                        {
+                            object val = func(reader);
+                            buffer.Add(GetValue<T>(reader, effectiveType, val));
+                        }
+                        while (await reader.NextResultAsync(cancel).ConfigureAwait(false)) { /* ignore subsequent result sets */ }
                     }
-                    while (await reader.NextResultAsync(cancel).ConfigureAwait(false)) { /* ignore subsequent result sets */ }
+
+                    reader = null;
                     command.OnCompleted();
-                    return buffer;
+                    if (wasClosed)
+                    {
+                        cnn.Close();
+                        wasClosed = false;
+                    }
+
+                    foreach (var item in buffer)
+                    {
+                        yield return item;
+                    }
                 }
                 else
                 {
-                    // can't use ReadAsync / cancellation; but this will have to do
-                    wasClosed = false; // don't close if handing back an open reader; rely on the command-behavior
-                    var deferred = ExecuteReaderSync<T>(reader, func, command.Parameters);
-                    reader = null; // to prevent it being disposed before the caller gets to see it
-                    return deferred;
+                    while (await reader.ReadAsync(cancel).ConfigureAwait(false))
+                    {
+                        object val = func(reader);
+                        yield return GetValue<T>(reader, effectiveType, val);
+                    }
+                    while (await reader.NextResultAsync(cancel).ConfigureAwait(false)) { /* ignore subsequent result sets */ }
+                    command.OnCompleted();
                 }
             }
             finally
             {
+#if (NETSTANDARD2_1_OR_GREATER || NETCOREAPP3_0_OR_GREATER)
+                await using (reader) { /* dispose if non-null */ }
+#else
                 using (reader) { /* dispose if non-null */ }
+#endif
                 if (wasClosed) cnn.Close();
             }
         }