Skip to content

Commit 32c4c37

Browse files
danielgerlagdanielgerlag
authored
Merge pull request #1399 from danielgerlag/copilot/fix-bcaf8fbb-32f9-4231-98a7-c72e191619e8
Add LINQ database query optimizations for large data scenarios
2 parents 76c011c + b31d6b5 commit 32c4c37

File tree

4 files changed

+167
-7
lines changed

4 files changed

+167
-7
lines changed

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

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
using Microsoft.EntityFrameworkCore;
2+
using System;
3+
using System.Collections.Generic;
4+
using System.Data;
5+
using System.Linq;
6+
using System.Threading.Tasks;
7+
using WorkflowCore.Persistence.EntityFramework.Models;
8+
using WorkflowCore.Models;
9+
using WorkflowCore.Persistence.EntityFramework.Interfaces;
10+
using System.Threading;
11+
using WorkflowCore.Interface;
12+
13+
namespace WorkflowCore.Persistence.EntityFramework.Services
14+
{
15+
public sealed class LargeDataOptimizedEntityFrameworkPersistenceProvider : EntityFrameworkPersistenceProvider, IPersistenceProvider
16+
{
17+
private readonly IWorkflowDbContextFactory _contextFactory;
18+
19+
public LargeDataOptimizedEntityFrameworkPersistenceProvider(IWorkflowDbContextFactory contextFactory, bool canCreateDb, bool canMigrateDb)
20+
: base(contextFactory, canCreateDb, canMigrateDb)
21+
{
22+
_contextFactory = contextFactory;
23+
}
24+
25+
/// <inheritdoc/>
26+
public new async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)
27+
{
28+
using (var db = _contextFactory.Build())
29+
{
30+
IQueryable<PersistedWorkflow> query = db.Set<PersistedWorkflow>()
31+
.Include(wf => wf.ExecutionPointers)
32+
.ThenInclude(ep => ep.ExtensionAttributes)
33+
.Include(wf => wf.ExecutionPointers)
34+
.AsSplitQuery()
35+
.AsQueryable();
36+
37+
if (status.HasValue)
38+
{
39+
query = query.Where(x => x.Status == status.Value);
40+
}
41+
42+
if (!string.IsNullOrEmpty(type))
43+
{
44+
query = query.Where(x => x.WorkflowDefinitionId == type);
45+
}
46+
47+
if (createdFrom.HasValue)
48+
{
49+
query = query.Where(x => x.CreateTime >= createdFrom.Value);
50+
}
51+
52+
if (createdTo.HasValue)
53+
{
54+
query = query.Where(x => x.CreateTime <= createdTo.Value);
55+
}
56+
57+
var rawResult = await query.OrderBy(x => x.PersistenceId).Skip(skip).Take(take).ToListAsync();
58+
59+
var result = new List<WorkflowInstance>(rawResult.Count);
60+
61+
foreach (var item in rawResult)
62+
{
63+
result.Add(item.ToWorkflowInstance());
64+
}
65+
66+
return result;
67+
}
68+
}
69+
70+
/// <inheritdoc/>
71+
public new async Task<WorkflowInstance> GetWorkflowInstance(string id, CancellationToken cancellationToken = default)
72+
{
73+
using (var db = _contextFactory.Build())
74+
{
75+
var uid = new Guid(id);
76+
var raw = await db.Set<PersistedWorkflow>()
77+
.Include(wf => wf.ExecutionPointers)
78+
.ThenInclude(ep => ep.ExtensionAttributes)
79+
.Include(wf => wf.ExecutionPointers)
80+
.AsSplitQuery()
81+
.FirstAsync(x => x.InstanceId == uid, cancellationToken);
82+
83+
return raw?.ToWorkflowInstance();
84+
}
85+
}
86+
87+
/// <inheritdoc/>
88+
public new async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids, CancellationToken cancellationToken = default)
89+
{
90+
if (ids == null)
91+
{
92+
return Array.Empty<WorkflowInstance>();
93+
}
94+
95+
using (var db = _contextFactory.Build())
96+
{
97+
var uids = ids.Select(i => new Guid(i));
98+
var raw = db.Set<PersistedWorkflow>()
99+
.Include(wf => wf.ExecutionPointers)
100+
.ThenInclude(ep => ep.ExtensionAttributes)
101+
.Include(wf => wf.ExecutionPointers)
102+
.AsSplitQuery()
103+
.Where(x => uids.Contains(x.InstanceId));
104+
105+
var persistedWorkflows = await raw.ToListAsync(cancellationToken);
106+
107+
return persistedWorkflows.Select(i => i.ToWorkflowInstance());
108+
}
109+
}
110+
111+
/// <inheritdoc/>
112+
public new async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken cancellationToken = default)
113+
{
114+
using (var db = _contextFactory.Build())
115+
using (var transaction = await db.Database.BeginTransactionAsync(IsolationLevel.RepeatableRead, cancellationToken))
116+
{
117+
var uid = new Guid(workflow.Id);
118+
var existingEntity = await db.Set<PersistedWorkflow>()
119+
.Where(x => x.InstanceId == uid)
120+
.Include(wf => wf.ExecutionPointers)
121+
.ThenInclude(ep => ep.ExtensionAttributes)
122+
.Include(wf => wf.ExecutionPointers)
123+
.AsSplitQuery()
124+
.AsTracking()
125+
.FirstAsync(cancellationToken);
126+
127+
_ = workflow.ToPersistable(existingEntity);
128+
129+
await db.SaveChangesAsync(cancellationToken);
130+
131+
await transaction.CommitAsync(cancellationToken);
132+
}
133+
}
134+
}
135+
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@
3030
</ItemGroup>
3131

3232
<ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.1' ">
33-
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="3.1.2" />
34-
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="3.1.2" />
33+
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="5.0.1" />
34+
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="5.0.1" />
3535
</ItemGroup>
3636

3737
<ItemGroup>

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

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,22 @@ namespace Microsoft.Extensions.DependencyInjection
99
{
1010
public static class ServiceCollectionExtensions
1111
{
12-
public static WorkflowOptions UsePostgreSQL(this WorkflowOptions options,
13-
string connectionString, bool canCreateDB, bool canMigrateDB, string schemaName="wfc")
12+
private static readonly Func<PostgresContextFactory, bool, bool, IPersistenceProvider> DefaultProviderFactory =
13+
(sqlContextFactory, canCreateDb, canMigrateDb) =>
14+
new EntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
15+
16+
private static readonly Func<PostgresContextFactory, bool, bool, IPersistenceProvider> OptimizedProviderFactory =
17+
(sqlContextFactory, canCreateDb, canMigrateDb) =>
18+
new LargeDataOptimizedEntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
19+
20+
public static WorkflowOptions UsePostgreSQL(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, string schemaName = "wfc") =>
21+
options.UsePostgreSQL(connectionString, canCreateDB, canMigrateDB, false, schemaName);
22+
23+
public static WorkflowOptions UsePostgreSQL(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, bool largeDataOptimized, string schemaName="wfc")
1424
{
15-
options.UsePersistence(sp => new EntityFrameworkPersistenceProvider(new PostgresContextFactory(connectionString, schemaName), canCreateDB, canMigrateDB));
25+
var providerFactory = largeDataOptimized ? OptimizedProviderFactory : DefaultProviderFactory;
26+
27+
options.UsePersistence(_ => providerFactory(new PostgresContextFactory(connectionString, schemaName), canCreateDB, canMigrateDB));
1628
options.Services.AddTransient<IWorkflowPurger>(sp => new WorkflowPurger(new PostgresContextFactory(connectionString, schemaName)));
1729
return options;
1830
}

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,9 +9,22 @@ namespace Microsoft.Extensions.DependencyInjection
99
{
1010
public static class ServiceCollectionExtensions
1111
{
12-
public static WorkflowOptions UseSqlServer(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, Action<DbConnection> initAction = null)
12+
private static readonly Func<SqlContextFactory, bool, bool, IPersistenceProvider> DefaultProviderFactory =
13+
(sqlContextFactory, canCreateDb, canMigrateDb) =>
14+
new EntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
15+
16+
private static readonly Func<SqlContextFactory, bool, bool, IPersistenceProvider> OptimizedProviderFactory =
17+
(sqlContextFactory, canCreateDb, canMigrateDb) =>
18+
new LargeDataOptimizedEntityFrameworkPersistenceProvider(sqlContextFactory, canCreateDb, canMigrateDb);
19+
20+
public static WorkflowOptions UseSqlServer(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, Action<DbConnection> initAction = null) =>
21+
options.UseSqlServer(connectionString, canCreateDB, canMigrateDB, false, initAction);
22+
23+
public static WorkflowOptions UseSqlServer(this WorkflowOptions options, string connectionString, bool canCreateDB, bool canMigrateDB, bool largeDataOptimized, Action<DbConnection> initAction = null)
1324
{
14-
options.UsePersistence(sp => new EntityFrameworkPersistenceProvider(new SqlContextFactory(connectionString, initAction), canCreateDB, canMigrateDB));
25+
var providerFactory = largeDataOptimized ? OptimizedProviderFactory : DefaultProviderFactory;
26+
27+
options.UsePersistence(_ => providerFactory(new SqlContextFactory(connectionString, initAction), canCreateDB, canMigrateDB));
1528
options.Services.AddTransient<IWorkflowPurger>(sp => new WorkflowPurger(new SqlContextFactory(connectionString, initAction)));
1629
return options;
1730
}

0 commit comments

Comments
 (0)