Fix high memory traffic while extracting metadata

Orm/Xtensive.Orm/Orm/Providers/Interfaces/ISqlExecutor.cs

@@ -5,6 +5,7 @@
 // Created:    2012.02.29
 
 using System.Collections.Generic;
+using System.Data;
 using System.Data.Common;
 using System.Threading;
 using System.Threading.Tasks;
@@ -21,8 +22,9 @@ public interface ISqlExecutor
     /// Executes the specified query statement. This method is similar to <see cref="DbCommand.ExecuteReader()"/>.
     /// </summary>
     /// <param name="statement">The statement to execute.</param>
+    /// <param name="commandBehavior">Options for statement execution and data retrieval.</param>
     /// <returns>Result of execution.</returns>
-    CommandWithDataReader ExecuteReader(ISqlCompileUnit statement);
+    CommandWithDataReader ExecuteReader(ISqlCompileUnit statement, CommandBehavior commandBehavior = CommandBehavior.Default);
 
     /// <summary>
     /// Asynchronously executes the specified query statement.
@@ -34,13 +36,27 @@ public interface ISqlExecutor
     /// <param name="token">The cancellation token to terminate execution if needed.</param>
     /// <returns>Result of execution.</returns>
     Task<CommandWithDataReader> ExecuteReaderAsync(ISqlCompileUnit statement, CancellationToken token = default);
+
+    /// <summary>
+    /// Asynchronously executes the specified query statement.
+    /// This method is similar to <see cref="DbCommand.ExecuteReaderAsync()"/>.
+    /// </summary>
+    /// <remarks> Multiple active operations are not supported. Use <see langword="await"/>
+    /// to ensure that all asynchronous operations have completed.</remarks>
+    /// <param name="statement">The statement to execute.</param>
+    /// <param name="commandBehavior">Options for statement execution and data retrieval.</param>
+    /// <param name="token">The cancellation token to terminate execution if needed.</param>
+    /// <returns>Result of execution.</returns>
+    Task<CommandWithDataReader> ExecuteReaderAsync(
+      ISqlCompileUnit statement, CommandBehavior commandBehavior, CancellationToken token = default);
 
     /// <summary>
     /// Executes the specified query statement. This method is similar to <see cref="DbCommand.ExecuteReader()"/>.
     /// </summary>
     /// <param name="commandText">The statement to execute.</param>
+    /// <param name="commandBehavior">Options for statement execution and data retrieval.</param>
     /// <returns>Result of execution.</returns>
-    CommandWithDataReader ExecuteReader(string commandText);
+    CommandWithDataReader ExecuteReader(string commandText, CommandBehavior commandBehavior = CommandBehavior.Default);
 
     /// <summary>
     /// Asynchronously executes the specified query statement.
@@ -51,6 +67,17 @@ public interface ISqlExecutor
     /// <returns>Result of execution.</returns>
     Task<CommandWithDataReader> ExecuteReaderAsync(string commandText, CancellationToken token = default);
 
+    /// <summary>
+    /// Asynchronously executes the specified query statement.
+    /// This method is similar to <see cref="DbCommand.ExecuteReaderAsync()"/>.
+    /// </summary>
+    /// <param name="commandText">The statement to execute.</param>
+    /// <param name="commandBehavior">Options for statement execution and data retrieval.</param>
+    /// <param name="token">The cancellation token to terminate execution if needed.</param>
+    /// <returns>Result of execution.</returns>
+    Task<CommandWithDataReader> ExecuteReaderAsync(
+      string commandText, CommandBehavior commandBehavior, CancellationToken token = default);
+
     /// <summary>
     /// Executes the specified scalar statement. This method is similar to <see cref="DbCommand.ExecuteScalar"/>.
     /// </summary>

Orm/Xtensive.Orm/Orm/Providers/SqlExecutor.cs

@@ -5,6 +5,7 @@
 // Created:    2012.02.29
 
 using System.Collections.Generic;
+using System.Data;
 using System.Data.Common;
 using System.Linq;
 using System.Threading;
@@ -23,17 +24,23 @@ internal sealed class SqlExecutor : ISqlExecutor
     private readonly StorageDriver driver;
     private readonly Session session;
 
-    public CommandWithDataReader ExecuteReader(ISqlCompileUnit statement)
+    public CommandWithDataReader ExecuteReader(
+      ISqlCompileUnit statement, CommandBehavior commandBehavior = CommandBehavior.Default)
     {
       EnsureConnectionIsOpen();
-      return ExecuteReader(connection.CreateCommand(Compile(statement)));
+      return ExecuteReader(connection.CreateCommand(Compile(statement)), commandBehavior);
     }
 
+    public Task<CommandWithDataReader> ExecuteReaderAsync(
+      ISqlCompileUnit statement, CancellationToken token = default) =>
+      ExecuteReaderAsync(statement, CommandBehavior.Default, token);
+
     public async Task<CommandWithDataReader> ExecuteReaderAsync(
-      ISqlCompileUnit statement, CancellationToken token = default)
+      ISqlCompileUnit statement, CommandBehavior commandBehavior, CancellationToken token = default)
     {
       await EnsureConnectionIsOpenAsync(token).ConfigureAwait(false);
-      return await ExecuteReaderAsync(connection.CreateCommand(Compile(statement)), token).ConfigureAwait(false);
+      return await ExecuteReaderAsync(
+        connection.CreateCommand(Compile(statement)), commandBehavior, token).ConfigureAwait(false);
     }
 
     public int ExecuteNonQuery(ISqlCompileUnit statement)
@@ -68,16 +75,22 @@ public async Task<object> ExecuteScalarAsync(ISqlCompileUnit statement, Cancella
       }
     }
 
-    public CommandWithDataReader ExecuteReader(string commandText)
+    public CommandWithDataReader ExecuteReader(
+      string commandText, CommandBehavior commandBehavior = CommandBehavior.Default)
     {
       EnsureConnectionIsOpen();
-      return ExecuteReader(connection.CreateCommand(commandText));
+      return ExecuteReader(connection.CreateCommand(commandText), commandBehavior);
     }
 
-    public async Task<CommandWithDataReader> ExecuteReaderAsync(string commandText, CancellationToken token = default)
+    public Task<CommandWithDataReader> ExecuteReaderAsync(string commandText, CancellationToken token = default) =>
+      ExecuteReaderAsync(commandText, CommandBehavior.Default, token);
+
+    public async Task<CommandWithDataReader> ExecuteReaderAsync(
+      string commandText, CommandBehavior commandBehavior, CancellationToken token = default)
     {
       await EnsureConnectionIsOpenAsync(token).ConfigureAwait(false);
-      return await ExecuteReaderAsync(connection.CreateCommand(commandText), token).ConfigureAwait(false);
+      return await ExecuteReaderAsync(
+        connection.CreateCommand(commandText), commandBehavior, token).ConfigureAwait(false);
     }
 
     public int ExecuteNonQuery(string commandText)
@@ -242,11 +255,11 @@ private string Compile(ISqlCompileUnit statement)
       return driver.Compile(statement, session.StorageNode.Configuration).GetCommandText();
     }
 
-    private CommandWithDataReader ExecuteReader(DbCommand command)
+    private CommandWithDataReader ExecuteReader(DbCommand command, CommandBehavior commandBehavior)
     {
       DbDataReader reader;
       try {
-        reader = driver.ExecuteReader(session, command);
+        reader = driver.ExecuteReader(session, command, commandBehavior);
       }
       catch {
         command.Dispose();
@@ -255,11 +268,11 @@ private CommandWithDataReader ExecuteReader(DbCommand command)
       return new CommandWithDataReader(command, reader);
     }
 
-    private async Task<CommandWithDataReader> ExecuteReaderAsync(DbCommand command, CancellationToken token)
+    private async Task<CommandWithDataReader> ExecuteReaderAsync(DbCommand command, CommandBehavior commandBehavior, CancellationToken token)
     {
       DbDataReader reader;
       try {
-        reader = await driver.ExecuteReaderAsync(session, command, token).ConfigureAwait(false);
+        reader = await driver.ExecuteReaderAsync(session, command, commandBehavior, token).ConfigureAwait(false);
       }
       catch {
         await command.DisposeAsync().ConfigureAwait(false);

Orm/Xtensive.Orm/Orm/Providers/StorageDriver.Operations.cs

@@ -392,45 +392,40 @@ public async ValueTask ReleaseSavepointAsync(
     #region Sync Execute methods
 
     public int ExecuteNonQuery(Session session, DbCommand command) =>
-      ExecuteCommand(session, command, c => c.ExecuteNonQuery());
+      ExecuteCommand(session, command, CommandBehavior.Default, (c, cb) => c.ExecuteNonQuery());
 
     public object ExecuteScalar(Session session, DbCommand command) =>
-      ExecuteCommand(session, command, c => c.ExecuteScalar());
+      ExecuteCommand(session, command, CommandBehavior.Default, (c, cb) => c.ExecuteScalar());
 
-    public DbDataReader ExecuteReader(Session session, DbCommand command) =>
-      ExecuteCommand(session, command, c => c.ExecuteReader());
+    public DbDataReader ExecuteReader(Session session, DbCommand command,
+      CommandBehavior behavior = CommandBehavior.Default) =>
+      ExecuteCommand(session, command, behavior, (c, cb) => c.ExecuteReader(cb));
 
     #endregion
 
     #region Async Execute methods
 
-    public Task<int> ExecuteNonQueryAsync(Session session, DbCommand command) =>
-      ExecuteCommandAsync(session, command, CancellationToken.None,
-        (c, ct) => c.ExecuteNonQueryAsync(ct));
+    public Task<int> ExecuteNonQueryAsync(Session session, DbCommand command, CancellationToken cancellationToken = default) =>
+      ExecuteCommandAsync(session, command, CommandBehavior.Default, cancellationToken,
+        (c, cb, ct) => c.ExecuteNonQueryAsync(ct));
 
-    public Task<int> ExecuteNonQueryAsync(Session session, DbCommand command, CancellationToken cancellationToken) =>
-      ExecuteCommandAsync(session, command, cancellationToken,
-        (c, ct) => c.ExecuteNonQueryAsync(ct));
+    public Task<object> ExecuteScalarAsync(Session session, DbCommand command, CancellationToken cancellationToken = default) =>
+      ExecuteCommandAsync(session, command, CommandBehavior.Default, cancellationToken,
+        (c, cb, ct) => c.ExecuteScalarAsync(ct));
 
-    public Task<object> ExecuteScalarAsync(Session session, DbCommand command) =>
-      ExecuteCommandAsync(session, command, CancellationToken.None,
-        (c, ct) => c.ExecuteScalarAsync(ct));
+    public Task<DbDataReader> ExecuteReaderAsync(Session session, DbCommand command,
+      CancellationToken cancellationToken = default) =>
+      ExecuteReaderAsync(session, command, CommandBehavior.Default, cancellationToken);
 
-    public Task<object> ExecuteScalarAsync(Session session, DbCommand command, CancellationToken cancellationToken) =>
-      ExecuteCommandAsync(session, command, cancellationToken,
-        (c, ct) => c.ExecuteScalarAsync(ct));
-
-    public Task<DbDataReader> ExecuteReaderAsync(Session session, DbCommand command) =>
-      ExecuteCommandAsync(session, command, CancellationToken.None,
-        (c, ct) => c.ExecuteReaderAsync(ct));
-
-    public Task<DbDataReader> ExecuteReaderAsync(Session session, DbCommand command, CancellationToken cancellationToken) =>
-      ExecuteCommandAsync(session, command, cancellationToken,
-        (c, ct) => c.ExecuteReaderAsync(ct));
+    public Task<DbDataReader> ExecuteReaderAsync(
+      Session session, DbCommand command, CommandBehavior behavior, CancellationToken cancellationToken = default) =>
+      ExecuteCommandAsync(session, command, behavior, cancellationToken,
+        (c, cb, ct) => c.ExecuteReaderAsync(cb, ct));
 
     #endregion
 
-    private TResult ExecuteCommand<TResult>(Session session, DbCommand command, Func<DbCommand, TResult> action)
+    private TResult ExecuteCommand<TResult>(
+      Session session, DbCommand command, CommandBehavior commandBehavior, Func<DbCommand, CommandBehavior, TResult> action)
     {
       if (isLoggingEnabled) {
         SqlLog.Info(Strings.LogSessionXQueryY, session.ToStringSafely(), command.ToHumanReadableString());
@@ -440,7 +435,7 @@ private TResult ExecuteCommand<TResult>(Session session, DbCommand command, Func
 
       TResult result;
       try {
-        result = action.Invoke(command);
+        result = action.Invoke(command, commandBehavior);
       }
       catch (Exception exception) {
         var wrapped = ExceptionBuilder.BuildException(exception, command.ToHumanReadableString());
@@ -453,8 +448,9 @@ private TResult ExecuteCommand<TResult>(Session session, DbCommand command, Func
       return result;
     }
 
-    private async Task<TResult> ExecuteCommandAsync<TResult>(Session session, DbCommand command,
-      CancellationToken cancellationToken, Func<DbCommand, CancellationToken, Task<TResult>> action)
+    private async Task<TResult> ExecuteCommandAsync<TResult>(Session session,
+      DbCommand command, CommandBehavior commandBehavior,
+      CancellationToken cancellationToken, Func<DbCommand, CommandBehavior, CancellationToken, Task<TResult>> action)
     {
       if (isLoggingEnabled) {
         SqlLog.Info(Strings.LogSessionXQueryY, session.ToStringSafely(), command.ToHumanReadableString());
@@ -465,7 +461,7 @@ private async Task<TResult> ExecuteCommandAsync<TResult>(Session session, DbComm
 
       TResult result;
       try {
-        result = await action(command, cancellationToken).ConfigureAwait(false);
+        result = await action(command, commandBehavior, cancellationToken).ConfigureAwait(false);
       }
       catch (OperationCanceledException) {
         throw;

Orm/Xtensive.Orm/Orm/Upgrade/Internals/Metadata/MetadataExtractor.cs

@@ -6,6 +6,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Data;
 using System.Data.Common;
 using System.Linq;
 using System.Threading;
@@ -130,7 +131,7 @@ private TypeMetadata ParseType(DbDataReader reader)
 
     private void ExecuteQuery<T>(ICollection<T> output, ISqlCompileUnit query, Func<DbDataReader, T> parser)
     {
-      using var command = executor.ExecuteReader(query);
+      using var command = executor.ExecuteReader(query, CommandBehavior.SequentialAccess);
       var reader = command.Reader;
       while (reader.Read()) {
         output.Add(parser.Invoke(reader));
@@ -140,7 +141,7 @@ private void ExecuteQuery<T>(ICollection<T> output, ISqlCompileUnit query, Func<
     private async Task ExecuteQueryAsync<T>(ICollection<T> output, ISqlCompileUnit query, Func<DbDataReader, T> parser,
       CancellationToken token)
     {
-      var command = await executor.ExecuteReaderAsync(query, token).ConfigureAwait(false);
+      var command = await executor.ExecuteReaderAsync(query, CommandBehavior.SequentialAccess, token).ConfigureAwait(false);
       await using (command.ConfigureAwait(false)) {
         var reader = command.Reader;
         while (await reader.ReadAsync(token).ConfigureAwait(false)) {