Insert rows into the database in parallel? #12057

okyrylchuk · 2022-03-23T10:05:05Z

okyrylchuk
Mar 23, 2022

Hi folks,

In one scenario, we need to insert thousands of records into the database in parallel to do it faster. As known, DbContext doesn't support multithreading. To create several DbContexts, we'd like to use DbContextFactory (maybe with pooling). However, I can't find how to do it correctly in the documentation. Is it possible? Or maybe there are other approaches to insert rows in the database in parallel?

Thank you in advance for the answers.

maliming · 2022-03-27T10:09:08Z

maliming
Mar 27, 2022
Collaborator

hi

You can create multiple new units of work to solve problems.

https://docs.abp.io/en/abp/latest/Unit-Of-Work#begin-a-new-unit-of-work

4 replies

okyrylchuk Mar 30, 2022
Author

@maliming thank you for your response.

There is a code requesting data from API. Then I want to insert the data in parallel (simplified code snippet below). All of the code there is already in the UoW. I tried your advice and I got the exception, it's below after the code.
Any ideas on how I can fix it?

// Code requesting data from API
await resources.ParallelForEachAsync(async (resource) =>
{
    using var uow = _unitOfWorkManager.Begin(requiresNew: true, isTransactional: false);                   
    var result = await InsertDataAsync(resource, ct);
    await uow.CompleteAsync(ct);
 });

Exception:

Microsoft.EntityFrameworkCore.Database.Connection[20004]
      An error occurred using the connection to database 'MyDb' on server '.'.
fail: Microsoft.EntityFrameworkCore.Database.Command[20102]
      Failed executing DbCommand (57ms) [Parameters=[@__ef_filter__p_0='?' (DbType = Boolean), @__id_0='?' (DbType = Guid)], CommandType='Text', CommandTimeout='30']
      SELECT [a0].[TenantId], [a0].[Name], [a0].[Value], [t].[Id]
      FROM (
          SELECT TOP(1) [a].[Id]
          FROM [AbpTenants] AS [a]
          WHERE ((@__ef_filter__p_0 = CAST(1 AS bit)) OR ([a].[IsDeleted] <> CAST(1 AS bit))) AND ([a].[Id] = @__id_0)
          ORDER BY [a].[Id]
      ) AS [t]
      INNER JOIN [AbpTenantConnectionStrings] AS [a0] ON [t].[Id] = [a0].[TenantId]
      ORDER BY [t].[Id]
fail: Microsoft.EntityFrameworkCore.Query[10100]
      An exception occurred while iterating over the results of a query for context type 'Volo.Abp.TenantManagement.EntityFrameworkCore.TenantManagementDbContext'.
      System.Transactions.TransactionException: The operation is not valid for the state of the transaction.
         at Microsoft.Data.SqlClient.SqlCommand.<>c.<ExecuteDbDataReaderAsync>b__169_0(Task`1 result)
         at System.Threading.Tasks.ContinuationResultTaskFromResultTask`2.InnerInvoke()
         at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
      --- End of stack trace from previous location ---
         at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot, Thread threadPoolThread)
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.RelationalShapedQueryCompilingExpressionVisitor.ShaperProcessingExpressionVisitor.<>c__DisplayClass63_0`2.<<PopulateSplitIncludeCollectionAsync>g__InitializeReaderAsync|0>d.MoveNext()
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.SqlServer.Storage.Internal.SqlServerExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.RelationalShapedQueryCompilingExpressionVisitor.ShaperProcessingExpressionVisitor.PopulateSplitIncludeCollectionAsync[TIncludingEntity,TIncludedEntity](Int32 collectionId, RelationalQueryContext queryContext, IExecutionStrategy executionStrategy, Boolean detailedErrorsEnabled, SplitQueryResultCoordinator resultCoordinator, RelationalCommandCache relationalCommandCache, Func`3 childIdentifier, IReadOnlyList`1 identifierValueComparers, Func`5 innerShaper, Func`4 relatedDataLoaders, INavigationBase inverseNavigation, Action`2 fixup, Boolean trackingQuery)
         at Microsoft.EntityFrameworkCore.Query.RelationalShapedQueryCompilingExpressionVisitor.ShaperProcessingExpressionVisitor.TaskAwaiter(Func`1[] taskFactories)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
      System.Transactions.TransactionException: The operation is not valid for the state of the transaction.
         at Microsoft.Data.SqlClient.SqlCommand.<>c.<ExecuteDbDataReaderAsync>b__169_0(Task`1 result)
         at System.Threading.Tasks.ContinuationResultTaskFromResultTask`2.InnerInvoke()
         at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
      --- End of stack trace from previous location ---
         at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot, Thread threadPoolThread)
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.RelationalShapedQueryCompilingExpressionVisitor.ShaperProcessingExpressionVisitor.<>c__DisplayClass63_0`2.<<PopulateSplitIncludeCollectionAsync>g__InitializeReaderAsync|0>d.MoveNext()
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.SqlServer.Storage.Internal.SqlServerExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.RelationalShapedQueryCompilingExpressionVisitor.ShaperProcessingExpressionVisitor.PopulateSplitIncludeCollectionAsync[TIncludingEntity,TIncludedEntity](Int32 collectionId, RelationalQueryContext queryContext, IExecutionStrategy executionStrategy, Boolean detailedErrorsEnabled, SplitQueryResultCoordinator resultCoordinator, RelationalCommandCache relationalCommandCache, Func`3 childIdentifier, IReadOnlyList`1 identifierValueComparers, Func`5 innerShaper, Func`4 relatedDataLoaders, INavigationBase inverseNavigation, Action`2 fixup, Boolean trackingQuery)
         at Microsoft.EntityFrameworkCore.Query.RelationalShapedQueryCompilingExpressionVisitor.ShaperProcessingExpressionVisitor.TaskAwaiter(Func`1[] taskFactories)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
fail: Microsoft.EntityFrameworkCore.Query[10100]
      An exception occurred while iterating over the results of a query for context type 'Volo.Abp.TenantManagement.EntityFrameworkCore.TenantManagementDbContext'.
      System.PlatformNotSupportedException: This platform does not support distributed transactions.
         at System.Transactions.Distributed.DistributedTransactionManager.GetDistributedTransactionFromTransmitterPropagationToken(Byte[] propagationToken)
         at System.Transactions.TransactionInterop.GetDistributedTransactionFromTransmitterPropagationToken(Byte[] propagationToken)
         at System.Transactions.TransactionStatePSPEOperation.PSPEPromote(InternalTransaction tx)
         at System.Transactions.TransactionStateDelegatedBase.EnterState(InternalTransaction tx)
         at System.Transactions.EnlistableStates.Promote(InternalTransaction tx)
         at System.Transactions.Transaction.Promote()
         at System.Transactions.TransactionInterop.ConvertToDistributedTransaction(Transaction transaction)
         at System.Transactions.TransactionInterop.GetExportCookie(Transaction transaction, Byte[] whereabouts)
         at Microsoft.Data.SqlClient.SqlInternalConnection.GetTransactionCookie(Transaction transaction, Byte[] whereAbouts)
         at Microsoft.Data.SqlClient.SqlInternalConnection.EnlistNonNull(Transaction tx)
         at Microsoft.Data.SqlClient.SqlInternalConnection.Enlist(Transaction tx)
         at Microsoft.Data.SqlClient.SqlInternalConnectionTds.Activate(Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionInternal.ActivateConnection(Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionPool.PrepareConnection(DbConnection owningObject, DbConnectionInternal obj, Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionPool.TryGetConnection(DbConnection owningObject, UInt32 waitForMultipleObjectsTimeout, Boolean allowCreate, Boolean onlyOneCheckConnection, DbConnectionOptions userOptions, DbConnectionInternal& connection)
         at Microsoft.Data.ProviderBase.DbConnectionPool.WaitForPendingOpen()
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenInternalAsync(Boolean errorsExpected, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenInternalAsync(Boolean errorsExpected, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenAsync(CancellationToken cancellationToken, Boolean errorsExpected)
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.InitializeReaderAsync(DbContext _, Boolean result, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.SqlServer.Storage.Internal.SqlServerExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
      System.PlatformNotSupportedException: This platform does not support distributed transactions.
         at System.Transactions.Distributed.DistributedTransactionManager.GetDistributedTransactionFromTransmitterPropagationToken(Byte[] propagationToken)
         at System.Transactions.TransactionInterop.GetDistributedTransactionFromTransmitterPropagationToken(Byte[] propagationToken)
         at System.Transactions.TransactionStatePSPEOperation.PSPEPromote(InternalTransaction tx)
         at System.Transactions.TransactionStateDelegatedBase.EnterState(InternalTransaction tx)
         at System.Transactions.EnlistableStates.Promote(InternalTransaction tx)
         at System.Transactions.Transaction.Promote()
         at System.Transactions.TransactionInterop.ConvertToDistributedTransaction(Transaction transaction)
         at System.Transactions.TransactionInterop.GetExportCookie(Transaction transaction, Byte[] whereabouts)
         at Microsoft.Data.SqlClient.SqlInternalConnection.GetTransactionCookie(Transaction transaction, Byte[] whereAbouts)
         at Microsoft.Data.SqlClient.SqlInternalConnection.EnlistNonNull(Transaction tx)
         at Microsoft.Data.SqlClient.SqlInternalConnection.Enlist(Transaction tx)
         at Microsoft.Data.SqlClient.SqlInternalConnectionTds.Activate(Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionInternal.ActivateConnection(Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionPool.PrepareConnection(DbConnection owningObject, DbConnectionInternal obj, Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionPool.TryGetConnection(DbConnection owningObject, UInt32 waitForMultipleObjectsTimeout, Boolean allowCreate, Boolean onlyOneCheckConnection, DbConnectionOptions userOptions, DbConnectionInternal& connection)
         at Microsoft.Data.ProviderBase.DbConnectionPool.WaitForPendingOpen()
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenInternalAsync(Boolean errorsExpected, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenInternalAsync(Boolean errorsExpected, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenAsync(CancellationToken cancellationToken, Boolean errorsExpected)
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.InitializeReaderAsync(DbContext _, Boolean result, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.SqlServer.Storage.Internal.SqlServerExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
      Error occured when running job 'MyJob'
      System.PlatformNotSupportedException: This platform does not support distributed transactions.
         at System.Transactions.Distributed.DistributedTransactionManager.GetDistributedTransactionFromTransmitterPropagationToken(Byte[] propagationToken)
         at System.Transactions.TransactionInterop.GetDistributedTransactionFromTransmitterPropagationToken(Byte[] propagationToken)
         at System.Transactions.TransactionStatePSPEOperation.PSPEPromote(InternalTransaction tx)
         at System.Transactions.TransactionStateDelegatedBase.EnterState(InternalTransaction tx)
         at System.Transactions.EnlistableStates.Promote(InternalTransaction tx)
         at System.Transactions.Transaction.Promote()
         at System.Transactions.TransactionInterop.ConvertToDistributedTransaction(Transaction transaction)
         at System.Transactions.TransactionInterop.GetExportCookie(Transaction transaction, Byte[] whereabouts)
         at Microsoft.Data.SqlClient.SqlInternalConnection.GetTransactionCookie(Transaction transaction, Byte[] whereAbouts)
         at Microsoft.Data.SqlClient.SqlInternalConnection.EnlistNonNull(Transaction tx)
         at Microsoft.Data.SqlClient.SqlInternalConnection.Enlist(Transaction tx)
         at Microsoft.Data.SqlClient.SqlInternalConnectionTds.Activate(Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionInternal.ActivateConnection(Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionPool.PrepareConnection(DbConnection owningObject, DbConnectionInternal obj, Transaction transaction)
         at Microsoft.Data.ProviderBase.DbConnectionPool.TryGetConnection(DbConnection owningObject, UInt32 waitForMultipleObjectsTimeout, Boolean allowCreate, Boolean onlyOneCheckConnection, DbConnectionOptions userOptions, DbConnectionInternal& connection)
         at Microsoft.Data.ProviderBase.DbConnectionPool.WaitForPendingOpen()
      --- End of stack trace from previous location ---
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenInternalAsync(Boolean errorsExpected, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenInternalAsync(Boolean errorsExpected, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Storage.RelationalConnection.OpenAsync(CancellationToken cancellationToken, Boolean errorsExpected)
         at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.InitializeReaderAsync(DbContext _, Boolean result, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.SqlServer.Storage.Internal.SqlServerExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.Internal.SplitQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
         at Microsoft.EntityFrameworkCore.Query.ShapedQueryCompilingExpressionVisitor.SingleOrDefaultAsync[TSource](IAsyncEnumerable`1 asyncEnumerable, CancellationToken cancellationToken)
         at Microsoft.EntityFrameworkCore.Query.ShapedQueryCompilingExpressionVisitor.SingleOrDefaultAsync[TSource](IAsyncEnumerable`1 asyncEnumerable, CancellationToken cancellationToken)
         at Volo.Abp.Domain.Repositories.EntityFrameworkCore.EfCoreRepository`3.FindAsync(TKey id, Boolean includeDetails, CancellationToken cancellationToken)
         at Castle.DynamicProxy.AsyncInterceptorBase.ProceedAsynchronous[TResult](IInvocation invocation, IInvocationProceedInfo proceedInfo)
         at Volo.Abp.Castle.DynamicProxy.CastleAbpMethodInvocationAdapterWithReturnValue`1.ProceedAsync()
         at Volo.Abp.Uow.UnitOfWorkInterceptor.InterceptAsync(IAbpMethodInvocation invocation)
         at Volo.Abp.Castle.DynamicProxy.CastleAsyncAbpInterceptorAdapter`1.InterceptAsync[TResult](IInvocation invocation, IInvocationProceedInfo proceedInfo, Func`3 proceed)
         at Volo.Abp.TenantManagement.TenantStore.FindAsync(Guid id)
         at Volo.Abp.MultiTenancy.MultiTenantConnectionStringResolver.ResolveAsync(String connectionStringName)
         at Volo.Abp.Uow.EntityFrameworkCore.UnitOfWorkDbContextProvider`1.GetDbContextAsync()
         at Volo.Abp.Domain.Repositories.EntityFrameworkCore.EfCoreRepository`2.GetDbSetAsync()
         at Volo.Abp.Domain.Repositories.EntityFrameworkCore.EfCoreRepository`2.FindAsync(Expression`1 predicate, Boolean includeDetails, CancellationToken cancellationToken)
         at Castle.DynamicProxy.AsyncInterceptorBase.ProceedAsynchronous[TResult](IInvocation invocation, IInvocationProceedInfo proceedInfo)
         at Volo.Abp.Castle.DynamicProxy.CastleAbpMethodInvocationAdapterWithReturnValue`1.ProceedAsync()
         at Volo.Abp.Uow.UnitOfWorkInterceptor.InterceptAsync(IAbpMethodInvocation invocation)
         at Volo.Abp.Castle.DynamicProxy.CastleAsyncAbpInterceptorAdapter`1.InterceptAsync[TResult](IInvocation invocation, IInvocationProceedInfo proceedInfo, Func`3 proceed)

maliming Mar 31, 2022
Collaborator

Your ABP Framework version.

okyrylchuk Mar 31, 2022
Author

4.2.0

Sorry for missing the version :)

maliming Apr 1, 2022
Collaborator

hi

Can you share a project that uses the template project(local SQLLocaldb)?

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Insert rows into the database in parallel? #12057

Uh oh!

{{title}}

Uh oh!

Replies: 1 comment 4 replies

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

Insert rows into the database in parallel? #12057

Uh oh!

okyrylchuk Mar 23, 2022

Replies: 1 comment · 4 replies

Uh oh!

maliming Mar 27, 2022 Collaborator

Uh oh!

okyrylchuk Mar 30, 2022 Author

Uh oh!

maliming Mar 31, 2022 Collaborator

Uh oh!

okyrylchuk Mar 31, 2022 Author

Uh oh!

maliming Apr 1, 2022 Collaborator

okyrylchuk
Mar 23, 2022

Replies: 1 comment 4 replies

maliming
Mar 27, 2022
Collaborator

okyrylchuk Mar 30, 2022
Author

maliming Mar 31, 2022
Collaborator

okyrylchuk Mar 31, 2022
Author

maliming Apr 1, 2022
Collaborator