Nonoptimal database query

src/providers/WorkflowCore.Persistence.EntityFramework/Services/LargeDataOptimizedEntityFrameworkPersistenceProvider.cs

@@ -0,0 +1,135 @@
+using Microsoft.EntityFrameworkCore;
+using System;
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Threading.Tasks;
+using WorkflowCore.Persistence.EntityFramework.Models;
+using WorkflowCore.Models;
+using WorkflowCore.Persistence.EntityFramework.Interfaces;
+using System.Threading;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.Persistence.EntityFramework.Services
+{
+    public sealed class LargeDataOptimizedEntityFrameworkPersistenceProvider : EntityFrameworkPersistenceProvider, IPersistenceProvider
+    {
+        private readonly IWorkflowDbContextFactory _contextFactory;
+
+        public LargeDataOptimizedEntityFrameworkPersistenceProvider(IWorkflowDbContextFactory contextFactory, bool canCreateDb, bool canMigrateDb)
+            : base(contextFactory, canCreateDb, canMigrateDb)
+        {
+            _contextFactory = contextFactory;
+        }
+
+        /// <inheritdoc/>
+        public new async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)
+        {
+            using (var db = _contextFactory.Build())
+            {
+                IQueryable<PersistedWorkflow> query = db.Set<PersistedWorkflow>()
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .AsSplitQuery()
+                    .AsQueryable();
+
+                if (status.HasValue)
+                {
+                    query = query.Where(x => x.Status == status.Value);
+                }
+
+                if (!string.IsNullOrEmpty(type))
+                {
+                    query = query.Where(x => x.WorkflowDefinitionId == type);
+                }
+
+                if (createdFrom.HasValue)
+                {
+                    query = query.Where(x => x.CreateTime >= createdFrom.Value);
+                }
+
+                if (createdTo.HasValue)
+                {
+                    query = query.Where(x => x.CreateTime <= createdTo.Value);
+                }
+
+                var rawResult = await query.OrderBy(x => x.PersistenceId).Skip(skip).Take(take).ToListAsync();
+
+                var result = new List<WorkflowInstance>(rawResult.Count);
+
+                foreach (var item in rawResult)
+                {
+                    result.Add(item.ToWorkflowInstance());
+                }
+
+                return result;
+            }
+        }
+
+        /// <inheritdoc/>
+        public new async Task<WorkflowInstance> GetWorkflowInstance(string id, CancellationToken cancellationToken = default)
+        {
+            using (var db = _contextFactory.Build())
+            {
+                var uid = new Guid(id);
+                var raw = await db.Set<PersistedWorkflow>()
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .AsSplitQuery()
+                    .FirstAsync(x => x.InstanceId == uid, cancellationToken);
+
+                return raw?.ToWorkflowInstance();
+            }
+        }
+
+        /// <inheritdoc/>
+        public new async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids, CancellationToken cancellationToken = default)
+        {
+            if (ids == null)
+            {
+                return Array.Empty<WorkflowInstance>();
+            }
+
+            using (var db = _contextFactory.Build())
+            {
+                var uids = ids.Select(i => new Guid(i));
+                var raw = db.Set<PersistedWorkflow>()
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .AsSplitQuery()
+                    .Where(x => uids.Contains(x.InstanceId));
+
+                var persistedWorkflows = await raw.ToListAsync(cancellationToken);
+
+                return persistedWorkflows.Select(i => i.ToWorkflowInstance());
+            }
+        }
+
+        /// <inheritdoc/>
+        public new async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken cancellationToken = default)
+        {
+            using (var db = _contextFactory.Build())
+            using (var transaction = await db.Database.BeginTransactionAsync(IsolationLevel.RepeatableRead, cancellationToken))
+            {
+                var uid = new Guid(workflow.Id);
+                var existingEntity = await db.Set<PersistedWorkflow>()
+                    .Where(x => x.InstanceId == uid)
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .AsSplitQuery()
+                    .AsTracking()
+                    .FirstAsync(cancellationToken);
+
+                _ = workflow.ToPersistable(existingEntity);
+
+                await db.SaveChangesAsync(cancellationToken);
+
+                await transaction.CommitAsync(cancellationToken);
+            }
+        }
+    }
+}

src/providers/WorkflowCore.Persistence.EntityFramework/WorkflowCore.Persistence.EntityFramework.csproj

@@ -30,8 +30,8 @@
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.1' ">
-    <PackageReference Include="Microsoft.EntityFrameworkCore" Version="3.1.2" />
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="3.1.2" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore" Version="5.0.1" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="5.0.1" />
   </ItemGroup>
 
   <ItemGroup>

src/providers/WorkflowCore.Persistence.PostgreSQL/ServiceCollectionExtensions.cs

@@ -9,10 +9,22 @@ namespace Microsoft.Extensions.DependencyInjection
 {
     public static class ServiceCollectionExtensions
     {
-        public static WorkflowOptions UsePostgreSQL(this WorkflowOptions options,
-            string connectionString, bool canCreateDB, bool canMigrateDB, string schemaName="wfc")
+        private static readonly Func<PostgresContextFactory, bool, bool, IPersistenceProvider> DefaultProviderFactory =
+            (sqlContextFactory, canCreateDb, canMigrateDb) =>
+                new EntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
+
+        private static readonly Func<PostgresContextFactory, bool, bool, IPersistenceProvider> OptimizedProviderFactory =
+            (sqlContextFactory, canCreateDb, canMigrateDb) =>
+                new LargeDataOptimizedEntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
+
+        public static WorkflowOptions UsePostgreSQL(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, string schemaName = "wfc") =>
+            options.UsePostgreSQL(connectionString, canCreateDB, canMigrateDB, false, schemaName);
+
+        public static WorkflowOptions UsePostgreSQL(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, bool largeDataOptimized, string schemaName="wfc")
         {
-            options.UsePersistence(sp => new EntityFrameworkPersistenceProvider(new PostgresContextFactory(connectionString, schemaName), canCreateDB, canMigrateDB));
+            var providerFactory = largeDataOptimized ? OptimizedProviderFactory : DefaultProviderFactory;
+
+            options.UsePersistence(_ => providerFactory(new PostgresContextFactory(connectionString, schemaName), canCreateDB, canMigrateDB));
             options.Services.AddTransient<IWorkflowPurger>(sp => new WorkflowPurger(new PostgresContextFactory(connectionString, schemaName)));
             return options;
         }

src/providers/WorkflowCore.Persistence.SqlServer/ServiceCollectionExtensions.cs

@@ -9,11 +9,24 @@ namespace Microsoft.Extensions.DependencyInjection
 {
     public static class ServiceCollectionExtensions
     {
-        public static WorkflowOptions UseSqlServer(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, Action<DbConnection> initAction = null)
+        private static readonly Func<SqlContextFactory, bool, bool, IPersistenceProvider> DefaultProviderFactory =
+            (sqlContextFactory, canCreateDb, canMigrateDb) =>
+                new EntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
+
+        private static readonly Func<SqlContextFactory, bool, bool, IPersistenceProvider> OptimizedProviderFactory =
+            (sqlContextFactory, canCreateDb, canMigrateDb) =>
+                new LargeDataOptimizedEntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
+
+        public static WorkflowOptions UseSqlServer(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, Action<DbConnection> initAction = null) =>
+            options.UseSqlServer(connectionString, canCreateDB, canMigrateDB, false, initAction);
+
+        public static WorkflowOptions UseSqlServer(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, bool largeDataOptimized, Action<DbConnection> initAction = null)
         {
-            options.UsePersistence(sp => new EntityFrameworkPersistenceProvider(new SqlContextFactory(connectionString, initAction), canCreateDB, canMigrateDB));
+            var providerFactory = largeDataOptimized ? OptimizedProviderFactory : DefaultProviderFactory;
+
+            options.UsePersistence(_ => providerFactory(new SqlContextFactory(connectionString, initAction), canCreateDB, canMigrateDB));
             options.Services.AddTransient<IWorkflowPurger>(sp => new WorkflowPurger(new SqlContextFactory(connectionString, initAction)));
             return options;
         }
     }
-}
+}

test/WorkflowCore.IntegrationTests/Scenarios/ForkScenario.cs

@@ -6,7 +6,7 @@
 
 namespace WorkflowCore.IntegrationTests.Scenarios
 {
-    public class ForkScenario : BaseScenario<ForkScenario.OutcomeFork, Object>
+    public class ForkScenario<TSelf> : BaseScenario<ForkScenario<TSelf>.OutcomeFork, Object>
     {
         static int TaskATicker = 0;
         static int TaskBTicker = 0;

test/WorkflowCore.IntegrationTests/Scenarios/IfScenario.cs

@@ -8,7 +8,7 @@
 
 namespace WorkflowCore.IntegrationTests.Scenarios
 {
-    public class IfScenario : WorkflowTest<IfScenario.IfWorkflow, IfScenario.MyDataClass>
+    public class IfScenario<TSelf> : WorkflowTest<IfScenario<TSelf>.IfWorkflow, IfScenario<TSelf>.MyDataClass>
     {
         internal static int Step1Ticker = 0;
         internal static int Step2Ticker = 0;

test/WorkflowCore.IntegrationTests/Scenarios/UserScenario.cs

@@ -8,7 +8,7 @@
 
 namespace WorkflowCore.IntegrationTests.Scenarios
 {
-    public class UserScenario : WorkflowTest<UserScenario.HumanWorkflow, Object>
+    public class UserScenario<TSelf> : WorkflowTest<UserScenario<TSelf>.HumanWorkflow, Object>
     {
         internal static int ApproveStepTicker = 0;
         internal static int DisapproveStepTicker = 0;

test/WorkflowCore.IntegrationTests/Scenarios/WhenScenario.cs

@@ -8,7 +8,7 @@
 
 namespace WorkflowCore.IntegrationTests.Scenarios
 {
-    public class WhenScenario : WorkflowTest<WhenScenario.WhenWorkflow, WhenScenario.MyDataClass>
+    public class WhenScenario<TSelf> : WorkflowTest<WhenScenario<TSelf>.WhenWorkflow, WhenScenario<TSelf>.MyDataClass>
     {
         internal static int Case1Ticker = 0;
         internal static int Case2Ticker = 0;

test/WorkflowCore.IntegrationTests/Scenarios/WhileScenario.cs

@@ -8,7 +8,7 @@
 
 namespace WorkflowCore.IntegrationTests.Scenarios
 {
-    public class WhileScenario : WorkflowTest<WhileScenario.WhileWorkflow, WhileScenario.MyDataClass>
+    public class WhileScenario<TSelf> : WorkflowTest<WhileScenario<TSelf>.WhileWorkflow, WhileScenario<TSelf>.MyDataClass>
     {
         internal static int Step1Ticker = 0;
         internal static int Step2Ticker = 0;

test/WorkflowCore.Tests.DynamoDB/Scenarios/DynamoIfScenario.cs

@@ -7,8 +7,8 @@
 namespace WorkflowCore.Tests.DynamoDB.Scenarios
 {
     [Collection("DynamoDb collection")]
-    public class DynamoIfScenario : IfScenario
-    {        
+    public class DynamoIfScenario : IfScenario<DynamoIfScenario>
+    {
         protected override void ConfigureServices(IServiceCollection services)
         {
             var cfg = new AmazonDynamoDBConfig {ServiceURL = DynamoDbDockerSetup.ConnectionString};

test/WorkflowCore.Tests.DynamoDB/Scenarios/DynamoWhileScenario.cs

@@ -7,8 +7,8 @@
 namespace WorkflowCore.Tests.DynamoDB.Scenarios
 {
     [Collection("DynamoDb collection")]
-    public class DynamoWhileScenario : WhileScenario
-    {        
+    public class DynamoWhileScenario : WhileScenario<DynamoWhileScenario>
+    {
         protected override void ConfigureServices(IServiceCollection services)
         {
             var cfg = new AmazonDynamoDBConfig {ServiceURL = DynamoDbDockerSetup.ConnectionString};

test/WorkflowCore.Tests.MongoDB/Scenarios/MongoForkScenario.cs

@@ -6,8 +6,8 @@
 namespace WorkflowCore.Tests.MongoDB.Scenarios
 {
     [Collection("Mongo collection")]
-    public class MongoForkScenario : ForkScenario
-    {        
+    public class MongoForkScenario : ForkScenario<MongoForkScenario>
+    {
         protected override void Configure(IServiceCollection services)
         {
             services.AddWorkflow(x => x.UseMongoDB(MongoDockerSetup.ConnectionString, nameof(MongoForkScenario)));

test/WorkflowCore.Tests.MongoDB/Scenarios/MongoIfScenario.cs

@@ -6,8 +6,8 @@
 namespace WorkflowCore.Tests.MongoDB.Scenarios
 {
     [Collection("Mongo collection")]
-    public class MongoIfScenario : IfScenario
-    {        
+    public class MongoIfScenario : IfScenario<MongoIfScenario>
+    {
         protected override void ConfigureServices(IServiceCollection services)
         {
             services.AddWorkflow(x => x.UseMongoDB(MongoDockerSetup.ConnectionString, nameof(MongoIfScenario)));

test/WorkflowCore.Tests.MongoDB/Scenarios/MongoUserScenario.cs

@@ -6,7 +6,7 @@
 namespace WorkflowCore.Tests.MongoDB.Scenarios
 {
     [Collection("Mongo collection")]
-    public class MongoUserScenario : UserScenario
+    public class MongoUserScenario : UserScenario<MongoUserScenario>
     {        
         protected override void ConfigureServices(IServiceCollection services)
         {

test/WorkflowCore.Tests.MongoDB/Scenarios/MongoWhenScenario.cs

@@ -6,7 +6,7 @@
 namespace WorkflowCore.Tests.MongoDB.Scenarios
 {
     [Collection("Mongo collection")]
-    public class MongoWhenScenario : WhenScenario
+    public class MongoWhenScenario : WhenScenario<MongoWhenScenario>
     {        
         protected override void ConfigureServices(IServiceCollection services)
         {

test/WorkflowCore.Tests.MongoDB/Scenarios/MongoWhileScenario.cs

@@ -6,7 +6,7 @@
 namespace WorkflowCore.Tests.MongoDB.Scenarios
 {
     [Collection("Mongo collection")]
-    public class MongoWhileScenario : WhileScenario
+    public class MongoWhileScenario : WhileScenario<MongoWhileScenario>
     {        
         protected override void ConfigureServices(IServiceCollection services)
         {

test/WorkflowCore.Tests.MySQL/Scenarios/MysqlForkScenario.cs

@@ -6,8 +6,8 @@
 namespace WorkflowCore.Tests.MySQL.Scenarios
 {
     [Collection("Mysql collection")]
-    public class MysqlForkScenario : ForkScenario
-    {        
+    public class MysqlForkScenario : ForkScenario<MysqlForkScenario>
+    {
         protected override void Configure(IServiceCollection services)
         {
             services.AddWorkflow(x => x.UseMySQL(MysqlDockerSetup.ScenarioConnectionString, true, true));

test/WorkflowCore.Tests.MySQL/Scenarios/MysqlIfScenario.cs

@@ -6,8 +6,8 @@
 namespace WorkflowCore.Tests.MySQL.Scenarios
 {
     [Collection("Mysql collection")]
-    public class MysqlIfScenario : IfScenario
-    {        
+    public class MysqlIfScenario : IfScenario<MysqlIfScenario>
+    {
         protected override void ConfigureServices(IServiceCollection services)
         {
             services.AddWorkflow(x => x.UseMySQL(MysqlDockerSetup.ScenarioConnectionString, true, true));

test/WorkflowCore.Tests.MySQL/Scenarios/MysqlUserScenario.cs

@@ -6,7 +6,7 @@
 namespace WorkflowCore.Tests.MySQL.Scenarios
 {
     [Collection("Mysql collection")]
-    public class MysqlUserScenario : UserScenario
+    public class MysqlUserScenario : UserScenario<MysqlUserScenario>
     {        
         protected override void ConfigureServices(IServiceCollection services)
         {

test/WorkflowCore.Tests.MySQL/Scenarios/MysqlWhenScenario.cs

@@ -6,7 +6,7 @@
 namespace WorkflowCore.Tests.MySQL.Scenarios
 {
     [Collection("Mysql collection")]
-    public class MysqlWhenScenario : WhenScenario
+    public class MysqlWhenScenario : WhenScenario<MysqlWhenScenario>
     {        
         protected override void ConfigureServices(IServiceCollection services)
         {

test/WorkflowCore.Tests.MySQL/Scenarios/MysqlWhileScenario.cs

@@ -6,7 +6,7 @@
 namespace WorkflowCore.Tests.MySQL.Scenarios
 {
     [Collection("Mysql collection")]
-    public class MysqlWhileScenario : WhileScenario
+    public class MysqlWhileScenario : WhileScenario<MysqlWhileScenario>
     {        
         protected override void ConfigureServices(IServiceCollection services)
         {

test/WorkflowCore.Tests.Oracle/Scenarios/OracleForkScenario.cs

@@ -2,15 +2,14 @@
 using Microsoft.Extensions.DependencyInjection;
 using WorkflowCore.IntegrationTests.Scenarios;
 using WorkflowCore.Persistence.Oracle;
-using WorkflowCore.Tests.Oracle;
 
 using Xunit;
 
 namespace WorkflowCore.Tests.Oracle.Scenarios
 {
     [Collection("Oracle collection")]
-    public class OracleForkScenario : ForkScenario
-    {        
+    public class OracleForkScenario : ForkScenario<OracleForkScenario>
+    {
         protected override void Configure(IServiceCollection services)
         {
             services.AddWorkflow(x => x.UseOracle(OracleDockerSetup.ConnectionString, true, true));