Add SQL query isolation level #6654

Arkatufus · 2023-04-19T21:04:47Z

Changes

Add transaction to all database queries
Add separate read and write isolation level to queries
Add isolation level settings to sql journal and snapshot settings
Add new HOCON key to control journal and snapshot query transaction level

Arkatufus · 2023-04-20T14:12:49Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Extensions/DbConnectionExtensions.cs

+    public static class DbConnectionExtensions
+    {
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public static async Task ExecuteInTransaction(


These extension methods wraps a query delegate inside a controlled transaction

Arkatufus · 2023-04-20T14:12:57Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Extensions/DbConnectionExtensions.cs

+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public static async Task<T> ExecuteInTransaction<T>(


These extension methods wraps a query delegate inside a controlled transaction.
This is the wrapper version that returns a value.

Arkatufus · 2023-04-20T14:13:46Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Extensions/IsolationLevelExtensions.cs

+            => config.GetString(key).ToIsolationLevel();
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public static IsolationLevel ToIsolationLevel(this string level)


Convenience method to convert HOCON/string to IsolationLevel

Arkatufus · 2023-04-20T14:14:24Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

+            WriteIsolationLevel = namingConventions.WriteIsolationLevel;
+
+            // backward compatibility
+            var level = config.GetString("isolation-level");


Backward compatibility for older codes

Arkatufus · 2023-04-20T14:18:33Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

@@ -862,25 +940,24 @@ private void TryProcess()
            {
                _remainingOperations--;

-                var chunk = DequeueChunk(_remainingOperations);
+                var (chunk, isWrite) = DequeueChunk(_remainingOperations);


isWrite is a flag to signal if the current batch chunk is a read or a write chunk.

DequeueChunk automatically separate the read and write requests in the queue, we need to pull the read and write information of the current batch chunk from there.

Arkatufus · 2023-04-20T14:20:04Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

                    .PipeTo(Self, failure: ex => new ChunkExecutionFailure(ex, chunk.Requests, chunk.ChunkId));
            }
        }

-        private async Task<BatchComplete> ExecuteChunk(RequestChunk chunk, IActorContext context)
+        private async Task<BatchComplete> ExecuteChunk(RequestChunk chunk, IActorContext context, bool isWriteOperation)


Need to tell ExecuteBatch the read/write context of the chunk so we can apply the appropriate read/write isolation level

Arkatufus · 2023-04-20T14:21:08Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

@@ -1331,7 +1406,7 @@ protected void AddParameter(TCommand command, string paramName, DbType dbType, o
        /// Select the buffer that has the smallest id on its first item, retrieve a maximum Setup.MaxBatchSize
        /// items from it, and return it as a chunk that needs to be batched
        /// </summary>
-        private RequestChunk DequeueChunk(int chunkId)
+        private (RequestChunk chunk, bool isWrite) DequeueChunk(int chunkId)


isWrite is a flag to signal if the current batch chunk is a read or a write chunk.

DequeueChunk automatically separate the read and write requests in the queue, we need to pull the read and write information of the current batch chunk from there.

Arkatufus · 2023-04-20T14:22:02Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

@@ -323,19 +331,84 @@ protected BatchingSqlJournalSetup(Config config, QueryConfiguration namingConven
        /// <param name="replayFilterSettings">The settings used when replaying events from database back to the persistent actors.</param>
        /// <param name="namingConventions">The naming conventions used by the database to construct valid SQL statements.</param>
        /// <param name="defaultSerializer">The serializer used when no specific type matching can be found.</param>
-        protected BatchingSqlJournalSetup(string connectionString, int maxConcurrentOperations, int maxBatchSize, int maxBufferSize, bool autoInitialize, TimeSpan connectionTimeout, IsolationLevel isolationLevel, CircuitBreakerSettings circuitBreakerSettings, ReplayFilterSettings replayFilterSettings, QueryConfiguration namingConventions, string defaultSerializer)
+        [Obsolete("Use constructor with separate read and write isolation level instead. (since v1.5.2)")]


Marking old constructors as obsolete

Arkatufus · 2023-04-20T14:22:20Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs

@@ -207,6 +217,7 @@ public class QueryConfiguration
        /// <param name="timeout">TBD</param>
        /// <param name="defaultSerializer">The default serializer used when not type override matching is found</param>
        /// <param name="useSequentialAccess">Uses the CommandBehavior.SequentialAccess when creating the command, providing a performance improvement for reading large BLOBS.</param>
+        [Obsolete("Use .ctor that accepts read and write IsolationLevel instead (since v1.5.2)")]


Marking old constructors as obsolete

Arkatufus · 2023-04-20T14:25:29Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs

        {
-            using (var command = GetCommand(connection, AllPersistenceIdsSql))
+            return await connection.ExecuteInTransaction(ReadIsolationLevel, cancellationToken, async (tx, token) =>


Converts old code into delegate and tuck them inside the transaction wrapper.

There should be zero impact from these wrap, the code is marked with aggresive inlining so that they are unrolled as soon as possible either during compile or inside JIT.

No logic were changed, we're just wrapping them inside explicit transaction.

Arkatufus · 2023-04-20T14:26:15Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs

        {
-            using (var command = GetCommand(connection, ByPersistenceIdSql))
+            await connection.ExecuteInTransaction(ReadIsolationLevel, cancellationToken, async (tx, token) =>


Same as above, wrap the old code inside transaction.

Arkatufus · 2023-04-20T14:27:39Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/SqlJournal.cs

@@ -27,7 +27,7 @@ public abstract class SqlJournal : AsyncWriteJournal, IWithUnboundedStash
        private IImmutableDictionary<string, long> _tagSequenceNr = ImmutableDictionary<string, long>.Empty;

        private readonly CancellationTokenSource _pendingRequestsCancellation;
-        private readonly JournalSettings _settings;
+        protected readonly JournalSettings Settings;


Expose JournalSettings as protected so inheriting classes can use them.

Arkatufus · 2023-04-20T14:28:40Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/QueryExecutor.cs

@@ -132,6 +143,7 @@ public class QueryConfiguration
        /// <param name="timeout">TBD</param>
        /// <param name="defaultSerializer">The default serializer used when not type override matching is found</param>
        /// <param name="useSequentialAccess">Uses the CommandBehavior.SequentialAccess when creating the command, providing a performance improvement for reading large BLOBS.</param>
+        [Obsolete("Use the constructor that takes read and write isolation level argument (since v1.5.2)")]


Mark old constructor as obsolete for backward compatibility

Arkatufus · 2023-04-20T14:29:57Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs

@@ -38,7 +38,7 @@ private sealed class Initialized
        /// </summary>
        private readonly CancellationTokenSource _pendingRequestsCancellation;

-        private readonly SnapshotStoreSettings _settings;
+        protected readonly SnapshotStoreSettings Settings;


Expose SnapshotStoreSettings as protected so inheriting classes can use them

Arkatufus · 2023-04-20T14:32:23Z

src/contrib/persistence/Akka.Persistence.Sqlite/sqlite.conf

@@ -31,6 +31,14 @@
 			# timestamp provider used for generation of journal entries timestamps
 			timestamp-provider = "Akka.Persistence.Sql.Common.Journal.DefaultTimestampProvider, Akka.Persistence.Sql.Common"

+			# The isolation level of all database read query
+			# Valid values: "chaos", "read-committed", "read-uncommitted", "repeatable-read", "serializable", "snapshot", or "unspecified"
+			read-isolation-level = unspecified


These new isolation level settings need to be propagated to all Sql.Common implementation when they are upgraded to the next Akka.NET release.

What's the difference between chaos and unspecified?

The differences are covered here: https://learn.microsoft.com/en-us/dotnet/api/system.data.isolationlevel?#fields

Arkatufus · 2023-04-20T14:32:31Z

src/contrib/persistence/Akka.Persistence.Sqlite/sqlite.conf

+
+			# The isolation level of all database write query
+			# Valid values: "chaos", "read-committed", "read-uncommitted", "repeatable-read", "serializable", "snapshot", or "unspecified"
+			write-isolation-level = unspecified


These new isolation level settings need to be propagated to all Sql.Common implementation when they are upgraded to the next Akka.NET release.

Arkatufus · 2023-04-20T14:34:33Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

        /// </summary>
-        public IsolationLevel IsolationLevel { get; }
+        [Obsolete("Use WriteIsolationLevel property instead")]


Make old property obsolete, old isolation level is used for both read and write, we now have fine grain control over both.

Arkatufus · 2023-04-20T14:35:39Z

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

            var stopwatch = new Stopwatch();
            using (var connection = CreateConnection(Setup.ConnectionString))
            {
                await connection.OpenAsync();

                // In the grand scheme of thing, using a transaction in an all read batch operation
                // should not hurt performance by much, because it is done only once at the start.
-                using (var tx = connection.BeginTransaction(Setup.IsolationLevel))
+                using (var tx = connection.BeginTransaction(isWriteOperation ? _writeIsolationLevel : _readIsolationLevel))


The kind of isolation level is controlled via the write flag argument

…akka.net into add_sql_query_isolation_level

Aaronontheweb

LGTM

Arkatufus added 4 commits April 20, 2023 03:42

Add SQL query isolation level

31ff083

Merge branch 'dev' into add_sql_query_isolation_level

16f6c36

Unify default isolation level to unspecified for compatibility

89974f8

Update API Verify list

f613e04

Arkatufus commented Apr 20, 2023

View reviewed changes

Aaronontheweb added akka-persistence-sql-common Akka.Persistence.Sql.Common configuration labels Apr 20, 2023

Arkatufus commented Apr 20, 2023

View reviewed changes

Arkatufus added 5 commits April 20, 2023 21:39

Update sqlite config unit tests

31b3cfa

Merge branch 'dev' into add_sql_query_isolation_level

c978e40

Remove IsolationLevel.Chaos support

e0c6051

Merge branch 'add_sql_query_isolation_level' of github.com:Arkatufus/…

dd27dfe

…akka.net into add_sql_query_isolation_level

Add documentation link in HOCON file

b7e6d5c

Aaronontheweb approved these changes Apr 20, 2023

View reviewed changes

Aaronontheweb enabled auto-merge (squash) April 20, 2023 15:14

Aaronontheweb merged commit 4537b1c into akkadotnet:dev Apr 20, 2023

Arkatufus mentioned this pull request Apr 20, 2023

Update RELEASE_NOTES.md for 1.5.3 release #6656

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add SQL query isolation level #6654

Add SQL query isolation level #6654

Arkatufus commented Apr 19, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Aaronontheweb Apr 20, 2023

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Arkatufus Apr 20, 2023

Aaronontheweb left a comment

Add SQL query isolation level #6654

Add SQL query isolation level #6654

Conversation

Arkatufus commented Apr 19, 2023

Changes

Choose a reason for hiding this comment

Arkatufus Apr 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Arkatufus Apr 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Arkatufus Apr 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Arkatufus Apr 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Aaronontheweb left a comment

Choose a reason for hiding this comment

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023 •

edited

Loading

Arkatufus Apr 20, 2023 •

edited

Loading