Allows the use of snapshot isolation level on transactions (#791)

Signed-off-by: John Battye <[email protected]>

Author: battyejp

src/MySqlConnector/Core/StandardEnlistedTransaction.cs

@@ -21,15 +21,16 @@ protected override void OnStart()
 				IsolationLevel.ReadCommitted => "read committed",
 				IsolationLevel.ReadUncommitted => "read uncommitted",
 				IsolationLevel.RepeatableRead => "repeatable read",
-				IsolationLevel.Snapshot => throw new NotSupportedException("IsolationLevel.{0} is not supported.".FormatInvariant(Transaction.IsolationLevel)),
+				IsolationLevel.Snapshot => "repeatable read",
 				IsolationLevel.Chaos => throw new NotSupportedException("IsolationLevel.{0} is not supported.".FormatInvariant(Transaction.IsolationLevel)),
 
 				// "In terms of the SQL:1992 transaction isolation levels, the default InnoDB level is REPEATABLE READ." - http://dev.mysql.com/doc/refman/5.7/en/innodb-transaction-model.html
 				IsolationLevel.Unspecified => "repeatable read",
 				_ => "repeatable read",
 			};
 
-			using var cmd = new MySqlCommand("set transaction isolation level " + isolationLevel + "; start transaction;", Connection);
+			var consistentSnapshotText = Transaction.IsolationLevel == IsolationLevel.Snapshot ? " with consistent snapshot" : string.Empty;
+			using var cmd = new MySqlCommand($"set transaction isolation level {isolationLevel}; start transaction{consistentSnapshotText};", Connection);
 			cmd.ExecuteNonQuery();
 		}
 

src/MySqlConnector/MySql.Data.MySqlClient/MySqlConnection.cs

@@ -63,14 +63,17 @@ private async ValueTask<MySqlTransaction> BeginDbTransactionAsync(IsolationLevel
 				IsolationLevel.ReadCommitted => "read committed",
 				IsolationLevel.RepeatableRead => "repeatable read",
 				IsolationLevel.Serializable => "serializable",
+				IsolationLevel.Snapshot => "repeatable read",
 
 				// "In terms of the SQL:1992 transaction isolation levels, the default InnoDB level is REPEATABLE READ." - http://dev.mysql.com/doc/refman/5.7/en/innodb-transaction-model.html
 				IsolationLevel.Unspecified => "repeatable read",
 
 				_ => throw new NotSupportedException("IsolationLevel.{0} is not supported.".FormatInvariant(isolationLevel))
 			};
 
-			using (var cmd = new MySqlCommand("set transaction isolation level " + isolationLevelValue + "; start transaction;", this))
+			var consistentSnapshotText = isolationLevel == IsolationLevel.Snapshot ? " with consistent snapshot" : string.Empty;
+
+			using (var cmd = new MySqlCommand($"set transaction isolation level {isolationLevelValue}; start transaction{consistentSnapshotText};", this))
 				await cmd.ExecuteNonQueryAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
 
 			var transaction = new MySqlTransaction(this, isolationLevel);
@@ -106,7 +109,7 @@ public override void EnlistTransaction(System.Transactions.Transaction? transact
 				else
 				{
 					m_enlistedTransaction = GetInitializedConnectionSettings().UseXaTransactions ?
-						(EnlistedTransactionBase) new XaEnlistedTransaction(transaction, this) :
+						(EnlistedTransactionBase)new XaEnlistedTransaction(transaction, this) :
 						new StandardEnlistedTransaction(transaction, this);
 					m_enlistedTransaction.Start();
 
@@ -253,7 +256,7 @@ private async Task ChangeDatabaseAsync(IOBehavior ioBehavior, string databaseNam
 			m_session.DatabaseOverride = databaseName;
 		}
 
-		public new MySqlCommand CreateCommand() => (MySqlCommand) base.CreateCommand();
+		public new MySqlCommand CreateCommand() => (MySqlCommand)base.CreateCommand();
 
 		public bool Ping() => PingAsync(IOBehavior.Synchronous, CancellationToken.None).GetAwaiter().GetResult();
 		public Task<bool> PingAsync(CancellationToken cancellationToken = default) => PingAsync(SimpleAsyncIOBehavior, cancellationToken).AsTask();
@@ -400,7 +403,7 @@ private SchemaProvider GetSchemaProvider()
 		SchemaProvider? m_schemaProvider;
 #endif
 
- 		/// <summary>
+		/// <summary>
 		/// Gets the time (in seconds) to wait while trying to establish a connection
 		/// before terminating the attempt and generating an error. This value
 		/// is controlled by <see cref="MySqlConnectionStringBuilder.ConnectionTimeout"/>,
@@ -662,7 +665,7 @@ private async ValueTask<ServerSession> CreateSessionAsync(ConnectionPool? pool,
 				var messageSuffix = (pool?.IsEmpty ?? false) ? " All pooled connections are in use." : "";
 				throw new MySqlException(MySqlErrorCode.UnableToConnectToHost, "Connect Timeout expired." + messageSuffix, ex);
 			}
-			catch (MySqlException ex) when ((timeoutSource?.IsCancellationRequested ?? false) || ((MySqlErrorCode) ex.Number == MySqlErrorCode.CommandTimeoutExpired))
+			catch (MySqlException ex) when ((timeoutSource?.IsCancellationRequested ?? false) || ((MySqlErrorCode)ex.Number == MySqlErrorCode.CommandTimeoutExpired))
 			{
 				throw new MySqlException(MySqlErrorCode.UnableToConnectToHost, "Connect Timeout expired.", ex);
 			}

tests/SideBySide/Transaction.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Data;
 using System.Data.Common;
+using System.Linq;
 using System.Threading.Tasks;
 using Dapper;
 using MySql.Data.MySqlClient;
@@ -50,7 +52,69 @@ public void DbConnectionCommit()
 			Assert.Equal(new[] { 1, 2 }, results);
 		}
 
+		[Theory]
+		[InlineData(IsolationLevel.ReadUncommitted, "read uncommitted")]
+		[InlineData(IsolationLevel.ReadCommitted, "read committed")]
+		[InlineData(IsolationLevel.RepeatableRead, "repeatable read")]
+		[InlineData(IsolationLevel.Serializable, "serializable")]
+		[InlineData(IsolationLevel.Unspecified, "repeatable read")]
 #if !BASELINE
+		[InlineData(IsolationLevel.Snapshot, "repeatable read")]
+#endif
+		public void DbConnectionIsolationLevel(IsolationLevel inputIsolationLevel, string expectedTransactionIsolationLevel)
+		{
+			DbConnection connection = m_connection;
+			m_connection.Execute(@"set global log_output = 'table';");
+			m_connection.Execute(@"set global general_log = 1;");
+			using (var trans = connection.BeginTransaction(inputIsolationLevel))
+			{
+				trans.Commit();
+			}
+
+			m_connection.Execute(@"set global general_log = 0;");
+			var results = connection.Query<string>($"select convert(argument USING utf8) from mysql.general_log where thread_id = {m_connection.ServerThread} order by event_time desc limit 10;");
+			var lastIsolationLevelQuery = results.First(x => x.ToLower().Contains("isolation"));
+
+			if (IsMySqlAndVersionLessThan57(m_connection.ServerVersion))
+			{
+				Assert.Contains("serializable", lastIsolationLevelQuery.ToLower());
+				return;
+			}
+
+			Assert.Contains(expectedTransactionIsolationLevel.ToLower(), lastIsolationLevelQuery.ToLower());
+		}
+
+#if !BASELINE
+		[Theory]
+		[InlineData(IsolationLevel.ReadUncommitted, "start transaction")]
+		[InlineData(IsolationLevel.ReadCommitted, "start transaction")]
+		[InlineData(IsolationLevel.RepeatableRead, "start transaction")]
+		[InlineData(IsolationLevel.Serializable, "start transaction")]
+		[InlineData(IsolationLevel.Unspecified, "start transaction")]
+		[InlineData(IsolationLevel.Snapshot, "start transaction with consistent snapshot")]
+		public void DbConnectionTransactionCommand(IsolationLevel inputIsolationLevel, string expectedTransactionIsolationLevel)
+		{
+			DbConnection connection = m_connection;
+			m_connection.Execute(@"set global log_output = 'table';");
+			m_connection.Execute(@"set global general_log = 1;");
+			using (var trans = connection.BeginTransaction(inputIsolationLevel))
+			{
+				trans.Commit();
+			}
+
+			m_connection.Execute(@"set global general_log = 0;");
+			var results = connection.Query<string>($"select convert(argument USING utf8) from mysql.general_log where thread_id = {m_connection.ServerThread} order by event_time desc limit 10;");
+			var lastStartTransactionQuery = results.First(x => x.ToLower().Contains("start"));
+
+			if (IsMySqlAndVersionLessThan57(m_connection.ServerVersion))
+			{
+				Assert.Contains("start transaction", lastStartTransactionQuery.ToLower());
+				return;
+			}
+
+			Assert.Contains(expectedTransactionIsolationLevel.ToLower(), lastStartTransactionQuery.ToLower());
+		}
+
 		[Fact]
 		public async Task CommitAsync()
 		{
@@ -210,7 +274,7 @@ public void SavepointRollbackUnknownName()
 			}
 			catch (MySqlException ex)
 			{
-				Assert.Equal(MySqlErrorCode.StoredProcedureDoesNotExist, (MySqlErrorCode) ex.Number);
+				Assert.Equal(MySqlErrorCode.StoredProcedureDoesNotExist, (MySqlErrorCode)ex.Number);
 			}
 		}
 
@@ -225,7 +289,7 @@ public void SavepointReleaseUnknownName()
 			}
 			catch (MySqlException ex)
 			{
-				Assert.Equal(MySqlErrorCode.StoredProcedureDoesNotExist, (MySqlErrorCode) ex.Number);
+				Assert.Equal(MySqlErrorCode.StoredProcedureDoesNotExist, (MySqlErrorCode)ex.Number);
 			}
 		}
 
@@ -242,7 +306,7 @@ public void SavepointRollbackReleasedName()
 			}
 			catch (MySqlException ex)
 			{
-				Assert.Equal(MySqlErrorCode.StoredProcedureDoesNotExist, (MySqlErrorCode) ex.Number);
+				Assert.Equal(MySqlErrorCode.StoredProcedureDoesNotExist, (MySqlErrorCode)ex.Number);
 			}
 		}
 
@@ -318,6 +382,20 @@ public async Task DisposeAsync()
 		}
 #endif
 
+		private bool IsMySqlAndVersionLessThan57(string currentVersionStr)
+		{
+			var version = new Version("5.7");
+			Version currentVersion = null;
+
+			if (Version.TryParse(currentVersionStr, out currentVersion))
+			{
+				var result = version.CompareTo(currentVersion);
+				return result > 0;
+			}
+
+			return false;
+		}
+
 		readonly TransactionFixture m_database;
 		readonly MySqlConnection m_connection;
 	}