Use "ambient" transaction in bulk loader. Fixes #300

Author: bgrainger

src/MySqlConnector/MySqlClient/MySqlBulkLoader.cs

@@ -5,7 +5,6 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using MySql.Data.Serialization;
 using MySql.Data.Protocol.Serialization;
 
 namespace MySql.Data.MySqlClient
@@ -23,6 +22,7 @@ public class MySqlBulkLoader
         public List<string> Columns { get; }
         public MySqlBulkLoaderConflictOption ConflictOption { get; set; }
         public MySqlConnection Connection { get; set; }
+		public MySqlTransaction Transaction { get; set; }
         public char EscapeCharacter { get; set; }
         public List<string> Expressions { get; }
         public char FieldQuotationCharacter { get; set; }
@@ -60,6 +60,7 @@ public MySqlBulkLoader(MySqlConnection connection)
             ConflictOption = MySqlBulkLoaderConflictOption.None;
             Columns = new List<string>();
             Expressions = new List<string>();
+	        Transaction = connection.CurrentTransaction;
         }
 
         private string BuildSqlCommand()
@@ -182,9 +183,9 @@ private async Task<int> LoadAsync(IOBehavior ioBehavior, CancellationToken cance
             try
             {
                 var commandString = BuildSqlCommand();
-                var cmd = new MySqlCommand(commandString, Connection)
+                var cmd = new MySqlCommand(commandString, Connection, Transaction)
                 {
-                    CommandTimeout = Timeout
+                    CommandTimeout = Timeout,
                 };
                 return await cmd.ExecuteNonQueryAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
             }

tests/SideBySide/BulkLoaderAsync.cs

@@ -208,6 +208,78 @@ public async Task BulkLoadLocalCsvFileNotFound()
 			};
 		}
 
+		[Fact]
+		public async Task BulkLoadLocalCsvFileInTransactionWithCommit()
+		{
+			try
+			{
+				await m_database.Connection.OpenAsync();
+				using (var transaction = m_database.Connection.BeginTransaction())
+				{
+					var bulkLoader = new MySqlBulkLoader(m_database.Connection)
+					{
+						FileName = AppConfig.MySqlBulkLoaderLocalCsvFile,
+						TableName = m_testTable,
+						CharacterSet = "UTF8",
+						NumberOfLinesToSkip = 1,
+						FieldTerminator = ",",
+						FieldQuotationCharacter = '"',
+						FieldQuotationOptional = true,
+						Local = true,
+					};
+					bulkLoader.Expressions.Add("five = UNHEX(five)");
+					bulkLoader.Columns.AddRange(new[] { "one", "two", "three", "four", "five" });
+
+					var rowCount = await bulkLoader.LoadAsync();
+					Assert.Equal(20, rowCount);
+
+					transaction.Commit();
+				}
+
+				Assert.Equal(20, await m_database.Connection.ExecuteScalarAsync<int>($@"select count(*) from {m_testTable};"));
+			}
+			finally
+			{
+				m_database.Connection.Close();
+			}
+		}
+
+		[Fact]
+		public async Task BulkLoadLocalCsvFileInTransactionWithRollback()
+		{
+			try
+			{
+				await m_database.Connection.OpenAsync();
+				using (var transaction = m_database.Connection.BeginTransaction())
+				{
+					var bulkLoader = new MySqlBulkLoader(m_database.Connection)
+					{
+						FileName = AppConfig.MySqlBulkLoaderLocalCsvFile,
+						TableName = m_testTable,
+						CharacterSet = "UTF8",
+						NumberOfLinesToSkip = 1,
+						FieldTerminator = ",",
+						FieldQuotationCharacter = '"',
+						FieldQuotationOptional = true,
+						Local = true,
+					};
+					bulkLoader.Expressions.Add("five = UNHEX(five)");
+					bulkLoader.Columns.AddRange(new[] { "one", "two", "three", "four", "five" });
+
+					var rowCount = await bulkLoader.LoadAsync();
+					Assert.Equal(20, rowCount);
+
+					transaction.Rollback();
+				}
+
+				Assert.Equal(0, await m_database.Connection.ExecuteScalarAsync<int>($@"select count(*) from {m_testTable};"));
+			}
+			finally
+			{
+				m_database.Connection.Close();
+			}
+		}
+
 		[Fact]
 		public async Task BulkLoadMissingFileName()
 		{

tests/SideBySide/BulkLoaderSync.cs

@@ -210,6 +210,78 @@ public void BulkLoadLocalCsvFileNotFound()
 			};
 		}
 
+		[Fact]
+		public void BulkLoadLocalCsvFileInTransactionWithCommit()
+		{
+			try
+			{
+				m_database.Connection.Open();
+				using (var transaction = m_database.Connection.BeginTransaction())
+				{
+					var bulkLoader = new MySqlBulkLoader(m_database.Connection)
+					{
+						FileName = AppConfig.MySqlBulkLoaderLocalCsvFile,
+						TableName = m_testTable,
+						CharacterSet = "UTF8",
+						NumberOfLinesToSkip = 1,
+						FieldTerminator = ",",
+						FieldQuotationCharacter = '"',
+						FieldQuotationOptional = true,
+						Local = true,
+					};
+					bulkLoader.Expressions.Add("five = UNHEX(five)");
+					bulkLoader.Columns.AddRange(new[] { "one", "two", "three", "four", "five" });
+
+					var rowCount = bulkLoader.Load();
+					Assert.Equal(20, rowCount);
+
+					transaction.Commit();
+				}
+
+				Assert.Equal(20, m_database.Connection.ExecuteScalar<int>($@"select count(*) from {m_testTable};"));
+			}
+			finally
+			{
+				m_database.Connection.Close();
+			}
+		}
+
+		[Fact]
+		public void BulkLoadLocalCsvFileInTransactionWithRollback()
+		{
+			try
+			{
+				m_database.Connection.Open();
+				using (var transaction = m_database.Connection.BeginTransaction())
+				{
+					var bulkLoader = new MySqlBulkLoader(m_database.Connection)
+					{
+						FileName = AppConfig.MySqlBulkLoaderLocalCsvFile,
+						TableName = m_testTable,
+						CharacterSet = "UTF8",
+						NumberOfLinesToSkip = 1,
+						FieldTerminator = ",",
+						FieldQuotationCharacter = '"',
+						FieldQuotationOptional = true,
+						Local = true,
+					};
+					bulkLoader.Expressions.Add("five = UNHEX(five)");
+					bulkLoader.Columns.AddRange(new[] { "one", "two", "three", "four", "five" });
+
+					var rowCount = bulkLoader.Load();
+					Assert.Equal(20, rowCount);
+
+					transaction.Rollback();
+				}
+
+				Assert.Equal(0, m_database.Connection.ExecuteScalar<int>($@"select count(*) from {m_testTable};"));
+			}
+			finally
+			{
+				m_database.Connection.Close();
+			}
+		}
+
 		[Fact]
 		public void BulkLoadMissingFileName()
 		{