Fix shortcomings of QueryBatcher initial implementation

 * AutoFlush is handled too late, causing stale data to be yielded
 * Cacheable result transformation is not fully done
 * Auto discovery of types fails
 * InitializeEntitiesAndCollections may be wrongly skipped or called
 * DoExecute should be renamed ExecuteBatched
 * BeginProcess should span the whole processing

 Follow up to #1718. See its reviews for more details on the above
 shortcomings.

Author: fredericDelaporte

src/NHibernate.Test/Async/Futures/QueryBatchFixture.cs

@@ -278,6 +278,115 @@ public async Task FutureForEagerMappedCollectionAsync()
 			}
 		}
 
+		[Test]
+		public async Task AutoDiscoverWorksWithFutureAsync()
+		{
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var future =
+					s
+						.CreateSQLQuery("select count(*) as count from EntitySimpleChild where Name like :pattern")
+						.AddScalar("count", NHibernateUtil.Int64)
+						.SetString("pattern", "Chi%")
+						.SetCacheable(true)
+					.FutureValue<long>();
+
+				Assert.That(await (future.GetValueAsync()), Is.EqualTo(2L), "From DB");
+				await (t.CommitAsync());
+			}
+
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var future =
+					s
+						.CreateSQLQuery("select count(*) as count from EntitySimpleChild where Name like :pattern")
+						.AddScalar("count", NHibernateUtil.Int64)
+						.SetString("pattern", "Chi%")
+						.SetCacheable(true)
+						.FutureValue<long>();
+
+				Assert.That(await (future.GetValueAsync()), Is.EqualTo(2L), "From cache");
+				await (t.CommitAsync());
+			}
+		}
+
+		[Test]
+		public async Task AutoFlushCacheInvalidationWorksWithFutureAsync()
+		{
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var futureResults =
+					(await (s
+						.CreateQuery("from EntitySimpleChild")
+						.SetCacheable(true)
+						.Future<EntitySimpleChild>()
+						.GetEnumerableAsync()))
+						.ToList();
+
+				Assert.That(futureResults, Has.Count.EqualTo(2), "First call");
+
+				await (t.CommitAsync());
+			}
+
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var deleted = await (s.Query<EntitySimpleChild>().FirstAsync());
+				// We need to get rid of a referencing entity for the delete.
+				deleted.Parent.Child1 = null;
+				deleted.Parent.Child2 = null;
+				await (s.DeleteAsync(deleted));
+
+				var future =
+					s
+						.CreateQuery("from EntitySimpleChild")
+						.SetCacheable(true)
+						.Future<EntitySimpleChild>();
+
+				Assert.That((await (future.GetEnumerableAsync())).ToList(), Has.Count.EqualTo(1), "After delete");
+				await (t.CommitAsync());
+			}
+		}
+
+		[Test]
+		public async Task UsingHqlToFutureWithCacheAndTransformerDoesntThrowAsync()
+		{
+			// Adapted from #383
+			using (var session = OpenSession())
+			using (var t = session.BeginTransaction())
+			{
+				//store values in cache
+				await (session
+					.CreateQuery("from EntitySimpleChild")
+					.SetResultTransformer(Transformers.DistinctRootEntity)
+					.SetCacheable(true)
+					.SetCacheMode(CacheMode.Normal)
+					.Future<EntitySimpleChild>()
+					.GetEnumerableAsync());
+				await (t.CommitAsync());
+			}
+
+			using (var session = OpenSession())
+			using (var t = session.BeginTransaction())
+			{
+				//get values from cache
+				var results =
+					(await (session
+						.CreateQuery("from EntitySimpleChild")
+						.SetResultTransformer(Transformers.DistinctRootEntity)
+						.SetCacheable(true)
+						.SetCacheMode(CacheMode.Normal)
+						.Future<EntitySimpleChild>()
+						.GetEnumerableAsync()))
+						.ToList();
+
+				Assert.That(results.Count, Is.EqualTo(2));
+			}
+		}
+
 		#region Test Setup
 
 		protected override HbmMapping GetMappings()

src/NHibernate.Test/Futures/QueryBatchFixture.cs

@@ -266,6 +266,116 @@ public void FutureForEagerMappedCollection()
 			}
 		}
 
+		[Test]
+		public void AutoDiscoverWorksWithFuture()
+		{
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var future =
+					s
+						.CreateSQLQuery("select count(*) as count from EntitySimpleChild where Name like :pattern")
+						.AddScalar("count", NHibernateUtil.Int64)
+						.SetString("pattern", "Chi%")
+						.SetCacheable(true)
+					.FutureValue<long>();
+
+				Assert.That(future.Value, Is.EqualTo(2L), "From DB");
+				t.Commit();
+			}
+
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var future =
+					s
+						.CreateSQLQuery("select count(*) as count from EntitySimpleChild where Name like :pattern")
+						.AddScalar("count", NHibernateUtil.Int64)
+						.SetString("pattern", "Chi%")
+						.SetCacheable(true)
+						.FutureValue<long>();
+
+				Assert.That(future.Value, Is.EqualTo(2L), "From cache");
+				t.Commit();
+			}
+		}
+
+		[Test]
+		public void AutoFlushCacheInvalidationWorksWithFuture()
+		{
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var futureResults =
+					s
+						.CreateQuery("from EntitySimpleChild")
+						.SetCacheable(true)
+						.Future<EntitySimpleChild>()
+						.GetEnumerable()
+						.ToList();
+
+				Assert.That(futureResults, Has.Count.EqualTo(2), "First call");
+
+				t.Commit();
+			}
+
+			using (var s = OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				var deleted = s.Query<EntitySimpleChild>().First();
+				// We need to get rid of a referencing entity for the delete.
+				deleted.Parent.Child1 = null;
+				deleted.Parent.Child2 = null;
+				s.Delete(deleted);
+
+				var future =
+					s
+						.CreateQuery("from EntitySimpleChild")
+						.SetCacheable(true)
+						.Future<EntitySimpleChild>();
+
+				Assert.That(future.GetEnumerable().ToList(), Has.Count.EqualTo(1), "After delete");
+				t.Commit();
+			}
+		}
+
+		[Test]
+		public void UsingHqlToFutureWithCacheAndTransformerDoesntThrow()
+		{
+			// Adapted from #383
+			using (var session = OpenSession())
+			using (var t = session.BeginTransaction())
+			{
+				//store values in cache
+				session
+					.CreateQuery("from EntitySimpleChild")
+					.SetResultTransformer(Transformers.DistinctRootEntity)
+					.SetCacheable(true)
+					.SetCacheMode(CacheMode.Normal)
+					.Future<EntitySimpleChild>()
+					.GetEnumerable();
+				t.Commit();
+			}
+
+			using (var session = OpenSession())
+			using (var t = session.BeginTransaction())
+			{
+				//get values from cache
+				var results =
+					session
+						.CreateQuery("from EntitySimpleChild")
+						.SetResultTransformer(Transformers.DistinctRootEntity)
+						.SetCacheable(true)
+						.SetCacheMode(CacheMode.Normal)
+						.Future<EntitySimpleChild>()
+						.GetEnumerable()
+						.ToList();
+
+				Assert.That(results.Count, Is.EqualTo(2));
+				t.Commit();
+			}
+		}
+
 		#region Test Setup
 
 		protected override HbmMapping GetMappings()

src/NHibernate/Async/Multi/IQueryBatchItem.cs

@@ -28,6 +28,13 @@ public partial interface IQueryBatchItem
 		/// <param name="cancellationToken">A cancellation token that can be used to cancel the work</param>
 		Task<IEnumerable<ISqlCommand>> GetCommandsAsync(CancellationToken cancellationToken);
 
+		/// <summary>
+		/// Process the result sets generated by <see cref="GetCommands"/>. Advance the results set
+		/// to the next query, or to its end if this is the last query.
+		/// </summary>
+		/// <returns>The number of rows processed.</returns>
+		Task<int> ProcessResultsSetAsync(DbDataReader reader, CancellationToken cancellationToken);
+
 		/// <summary>
 		/// Executed after all commands in batch are processed
 		/// </summary>

src/NHibernate/Async/Multi/QueryBatch.cs

@@ -29,39 +29,40 @@ public async Task ExecuteAsync(CancellationToken cancellationToken)
 			cancellationToken.ThrowIfCancellationRequested();
 			if (_queries.Count == 0)
 				return;
-			var sessionFlushMode = Session.FlushMode;
-			if (FlushMode.HasValue)
-				Session.FlushMode = FlushMode.Value;
-			try
+			using (Session.BeginProcess())
 			{
-				Init();
-
-				if (!Session.Factory.ConnectionProvider.Driver.SupportsMultipleQueries)
+				var sessionFlushMode = Session.FlushMode;
+				if (FlushMode.HasValue)
+					Session.FlushMode = FlushMode.Value;
+				try
 				{
-					foreach (var query in _queries)
+					Init();
+
+					if (!Session.Factory.ConnectionProvider.Driver.SupportsMultipleQueries)
 					{
-						await (query.ExecuteNonBatchedAsync(cancellationToken)).ConfigureAwait(false);
+						foreach (var query in _queries)
+						{
+							await (query.ExecuteNonBatchedAsync(cancellationToken)).ConfigureAwait(false);
+						}
+
+						return;
 					}
-					return;
-				}
 
-				using (Session.BeginProcess())
-				{
-					await (DoExecuteAsync(cancellationToken)).ConfigureAwait(false);
+					await (ExecuteBatchedAsync(cancellationToken)).ConfigureAwait(false);
 				}
-			}
-			finally
-			{
-				if (_autoReset)
+				finally
 				{
-					_queries.Clear();
-					_queriesByKey.Clear();
-				}
-				else
-					_executed = true;
+					if (_autoReset)
+					{
+						_queries.Clear();
+						_queriesByKey.Clear();
+					}
+					else
+						_executed = true;
 
-				if (FlushMode.HasValue)
-					Session.FlushMode = sessionFlushMode;
+					if (FlushMode.HasValue)
+						Session.FlushMode = sessionFlushMode;
+				}
 			}
 		}
 
@@ -90,7 +91,7 @@ private async Task<IList<TResult>> GetResultsAsync<TResult>(IQueryBatchItem quer
 			cancellationToken.ThrowIfCancellationRequested();
 			if (!_executed)
 				await (ExecuteAsync(cancellationToken)).ConfigureAwait(false);
-			return ((IQueryBatchItem<TResult>)query).GetResults();
+			return ((IQueryBatchItem<TResult>) query).GetResults();
 		}
 
 		private async Task CombineQueriesAsync(IResultSetsCommand resultSetsCommand, CancellationToken cancellationToken)
@@ -103,25 +104,29 @@ private async Task CombineQueriesAsync(IResultSetsCommand resultSetsCommand, Can
 			}
 		}
 
-		protected async Task DoExecuteAsync(CancellationToken cancellationToken)
+		protected async Task ExecuteBatchedAsync(CancellationToken cancellationToken)
 		{
 			cancellationToken.ThrowIfCancellationRequested();
-			var resultSetsCommand = Session.Factory.ConnectionProvider.Driver.GetResultSetsCommand(Session);
-			await (CombineQueriesAsync(resultSetsCommand, cancellationToken)).ConfigureAwait(false);
-
 			var querySpaces = new HashSet<string>(_queries.SelectMany(t => t.GetQuerySpaces()));
-			if (resultSetsCommand.HasQueries)
+			if (querySpaces.Count > 0)
 			{
 				await (Session.AutoFlushIfRequiredAsync(querySpaces, cancellationToken)).ConfigureAwait(false);
 			}
 
+			var resultSetsCommand = Session.Factory.ConnectionProvider.Driver.GetResultSetsCommand(Session);
+			// CombineQueries queries the second level cache, which may contain stale data in regard to
+			// the session changes. For having them invalidated, auto-flush must have been handled before
+			// calling CombineQueries.
+			await (CombineQueriesAsync(resultSetsCommand, cancellationToken)).ConfigureAwait(false);
+
 			bool statsEnabled = Session.Factory.Statistics.IsStatisticsEnabled;
 			Stopwatch stopWatch = null;
 			if (statsEnabled)
 			{
 				stopWatch = new Stopwatch();
 				stopWatch.Start();
 			}
+
 			if (Log.IsDebugEnabled())
 			{
 				Log.Debug("Multi query with {0} queries: {1}", _queries.Count, resultSetsCommand.Sql);
@@ -136,11 +141,7 @@ protected async Task DoExecuteAsync(CancellationToken cancellationToken)
 					{
 						foreach (var multiSource in _queries)
 						{
-							foreach (var resultSetHandler in multiSource.GetResultSetHandler())
-							{
-								rowCount += resultSetHandler(reader);
-								await (reader.NextResultAsync(cancellationToken)).ConfigureAwait(false);
-							}
+							rowCount += await (multiSource.ProcessResultsSetAsync(reader, cancellationToken)).ConfigureAwait(false);
 						}
 					}
 				}
@@ -154,13 +155,20 @@ protected async Task DoExecuteAsync(CancellationToken cancellationToken)
 			catch (Exception sqle)
 			{
 				Log.Error(sqle, "Failed to execute multi query: [{0}]", resultSetsCommand.Sql);
-				throw ADOExceptionHelper.Convert(Session.Factory.SQLExceptionConverter, sqle, "Failed to execute multi query", resultSetsCommand.Sql);
+				throw ADOExceptionHelper.Convert(
+					Session.Factory.SQLExceptionConverter,
+					sqle,
+					"Failed to execute multi query",
+					resultSetsCommand.Sql);
 			}
 
 			if (statsEnabled)
 			{
 				stopWatch.Stop();
-				Session.Factory.StatisticsImplementor.QueryExecuted($"{_queries.Count} queries", rowCount, stopWatch.Elapsed);
+				Session.Factory.StatisticsImplementor.QueryExecuted(
+					$"{_queries.Count} queries",
+					rowCount,
+					stopWatch.Elapsed);
 			}
 		}
 	}