Skip to content

Improve journal delete performance #538

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Aaronontheweb merged 4 commits into from
Jun 6, 2025
Merged

Improve journal delete performance #538

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6c9c78b
Improve journal delete performance
Arkatufus May 21, 2025
9b5fb98
Optimize code
Arkatufus May 21, 2025
3893fe0
Merge branch 'dev' into improve-delete-performance
Arkatufus May 21, 2025
a622d65
Merge branch 'dev' into improve-delete-performance
Arkatufus May 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/Akka.Persistence.Sql/Db/AkkaPersistenceDataConnectionFactory.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public AkkaPersistenceDataConnectionFactory(IProviderConfig<JournalTableConfig>

_useCloneDataConnection = config.UseCloneConnection;

if (_opts.RetryPolicyOptions.RetryPolicy is null && _opts.ConnectionOptions.ProviderName!.ToLowerInvariant().StartsWith("sqlserver"))
if (_opts.RetryPolicyOptions.RetryPolicy is null && _opts.RetryPolicyOptions.Factory is null && _opts.ConnectionOptions.ProviderName!.ToLowerInvariant().StartsWith("sqlserver"))

@Arkatufus Arkatufus May 21, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add missing retry policy factory check. The Factory (if not null) is used internally by Linq2Db to populate DataConnection.RetryPolicy if the DataOption.RetryPolicyOptions.RetryPolicy is null.

_opts = _opts.WithOptions( _opts.RetryPolicyOptions with { RetryPolicy = new SqlServerRetryPolicy() } );

_cloneConnection = new Lazy<AkkaDataConnection>(
Expand Down
44 changes: 24 additions & 20 deletions src/Akka.Persistence.Sql/Journal/Dao/BaseByteArrayJournalDao.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,22 +115,37 @@ public async Task<IImmutableList<Exception>> AsyncWriteMessages(
public async Task Delete(string persistenceId, long maxSequenceNr, CancellationToken cancellationToken)
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, ShutdownToken);
long maxMarkedDeletion;

// no need to use transaction
await using (var connection = ConnectionFactory.GetConnection())
{
maxMarkedDeletion = await MaxMarkedForDeletionMaxPersistenceIdQuery(connection, persistenceId, maxSequenceNr).FirstOrDefaultAsync(cts.Token);
}

if (maxMarkedDeletion is 0)
return;
Comment on lines +126 to +127

@Arkatufus Arkatufus May 21, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Early bail out condition, just return if there's nothing to delete.


await ConnectionFactory.ExecuteWithTransactionAsync(
WriteIsolationLevel,
cts.Token,
async (connection, token) =>
{
await connection
.GetTable<JournalRow>()
var journalTable = connection.GetTable<JournalRow>();
await journalTable
.Where(
r =>
r.PersistenceId == persistenceId &&
r.SequenceNumber <= maxSequenceNr)
r.SequenceNumber == maxMarkedDeletion)

@Arkatufus Arkatufus May 21, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of marking all of the records less than the requested sequence number as deleted, we mark the highest affected instead. Update 1 record instead of many.

.Set(r => r.Deleted, true)
.UpdateAsync(token);

var maxMarkedDeletion =
await MaxMarkedForDeletionMaxPersistenceIdQuery(connection, persistenceId).FirstOrDefaultAsync(token);
await journalTable
.Where(
r =>
r.PersistenceId == persistenceId &&
r.SequenceNumber < maxMarkedDeletion)
.DeleteAsync(token);
Comment on lines +143 to +148

@Arkatufus Arkatufus May 21, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Simplify the deletion query, just delete anything that is less than the highest affected row.


if (JournalConfig.DaoConfig.SqlCommonCompatibilityMode)
{
Expand All @@ -149,19 +164,7 @@ await connection
SequenceNumber = maxMarkedDeletion,
},
token: token);
}

await connection
.GetTable<JournalRow>()
.Where(
r =>
r.PersistenceId == persistenceId &&
r.SequenceNumber <= maxSequenceNr &&
r.SequenceNumber < maxMarkedDeletion)
.DeleteAsync(token);

if (JournalConfig.DaoConfig.SqlCommonCompatibilityMode)
{
await connection
.GetTable<JournalMetaData>()
.Where(
Expand All @@ -177,7 +180,7 @@ await connection
.GetTable<JournalTagRow>()
.Where(
r =>
r.SequenceNumber <= maxSequenceNr &&
r.SequenceNumber < maxMarkedDeletion &&
r.PersistenceId == persistenceId)
.DeleteAsync(token);
}
Expand Down Expand Up @@ -442,10 +445,11 @@ protected static ImmutableList<Exception> BuildWriteRejections(List<Util.Try<Jou
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static IQueryable<long> MaxMarkedForDeletionMaxPersistenceIdQuery(
AkkaDataConnection connection,
string persistenceId)
string persistenceId,
long maxSequenceNr)
=> connection
.GetTable<JournalRow>()
.Where(r => r.PersistenceId == persistenceId && r.Deleted)
.Where(r => r.PersistenceId == persistenceId && r.SequenceNumber <= maxSequenceNr)

@Arkatufus Arkatufus May 21, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Optimization, instead of checking against persistence id and deleted field, check against persistence id and sequence number instead, as they are indexed.

.OrderByDescending(r => r.SequenceNumber)
.Select(r => r.SequenceNumber)
.Take(1);
Expand Down