Implement batching for MySqlDataAdapter. Fixes #635

Author: bgrainger

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

@@ -1,5 +1,6 @@
 #if !NETSTANDARD1_3
 using System;
+using System.Collections.Generic;
 using System.Data;
 using System.Data.Common;
 
@@ -63,6 +64,41 @@ public MySqlDataAdapter(string selectCommandText, string connectionString)
 		protected override RowUpdatingEventArgs CreateRowUpdatingEvent(DataRow dataRow, IDbCommand command, StatementType statementType, DataTableMapping tableMapping) => new MySqlRowUpdatingEventArgs(dataRow, command, statementType, tableMapping);
 
 		protected override RowUpdatedEventArgs CreateRowUpdatedEvent(DataRow dataRow, IDbCommand command, StatementType statementType, DataTableMapping tableMapping) => new MySqlRowUpdatedEventArgs(dataRow, command, statementType, tableMapping);
+
+		public override int UpdateBatchSize { get; set; }
+
+		protected override void InitializeBatching() => m_batchCommands = new List<MySqlCommand>();
+
+		protected override void TerminateBatching()
+		{
+			if (m_batchCommands is object)
+				ClearBatch();
+			m_batchCommands = null;
+		}
+
+		protected override int AddToBatch(IDbCommand command)
+		{
+			var count = m_batchCommands.Count;
+			m_batchCommands.Add(((MySqlCommand) command).Clone());
+			return count;
+		}
+
+		protected override void ClearBatch()
+		{
+			foreach (var command in m_batchCommands)
+				command.Dispose();
+			m_batchCommands.Clear();
+		}
+
+		protected override int ExecuteBatch()
+		{
+			var result = 0;
+			foreach (var command in m_batchCommands)
+				result += command.ExecuteNonQuery();
+			return result;
+		}
+
+		List<MySqlCommand> m_batchCommands;
 	}
 
 	public delegate void MySqlRowUpdatingEventHandler(object sender, MySqlRowUpdatingEventArgs e);

tests/SideBySide/DataAdapterTests.cs

@@ -191,6 +191,72 @@ public void InsertWithDataSet()
 			}
 		}
 
+		[Fact]
+		public void BatchUpdate()
+		{
+			using (var ds = new DataSet())
+			using (var da = new MySqlDataAdapter("SELECT * FROM data_adapter", m_connection))
+			{
+				da.Fill(ds);
+
+				da.UpdateCommand = new MySqlCommand("UPDATE data_adapter SET int_value=@int, text_value=@text WHERE id=@id", m_connection)
+				{
+					Parameters =
+					{
+						new MySqlParameter("@int", MySqlDbType.Int32) { Direction = ParameterDirection.Input, SourceColumn = "int_value" },
+						new MySqlParameter("@text", MySqlDbType.String) { Direction = ParameterDirection.Input, SourceColumn = "text_value" },
+						new MySqlParameter("@id", MySqlDbType.Int64) { Direction = ParameterDirection.Input, SourceColumn = "id" },
+					},
+					UpdatedRowSource = UpdateRowSource.None,
+				};
+
+				da.UpdateBatchSize = 10;
+
+				var dt = ds.Tables[0];
+				dt.Rows[0][1] = 2;
+				dt.Rows[0][2] = "two";
+				dt.Rows[1][1] = 3;
+				dt.Rows[1][2] = "three";
+				dt.Rows[2][1] = 4;
+				dt.Rows[2][2] = "four";
+
+				da.Update(ds);
+			}
+
+			Assert.Equal(new[] { "two", "three", "four" }, m_connection.Query<string>("SELECT text_value FROM data_adapter ORDER BY id"));
+		}
+
+
+		[Fact]
+		public void BatchInsert()
+		{
+			using (var ds = new DataSet())
+			using (var da = new MySqlDataAdapter("SELECT * FROM data_adapter", m_connection))
+			{
+				da.Fill(ds);
+
+				da.InsertCommand = new MySqlCommand("INSERT INTO data_adapter(int_value, text_value) VALUES(@int, @text);", m_connection)
+				{
+					Parameters =
+					{
+						new MySqlParameter("@int", MySqlDbType.Int32) { Direction = ParameterDirection.Input, SourceColumn = "int_value" },
+						new MySqlParameter("@text", MySqlDbType.String) { Direction = ParameterDirection.Input, SourceColumn = "text_value" },
+					},
+					UpdatedRowSource = UpdateRowSource.None,
+				};
+	
+				da.UpdateBatchSize = 10;
+
+				var dt = ds.Tables[0];
+				dt.Rows.Add(0, 2, "two");
+				dt.Rows.Add(0, 3, "three");
+				dt.Rows.Add(0, 4, "four");
+
+				da.Update(ds);
+			}
+
+			Assert.Equal(new[] { null, "", "one", "two", "three", "four" }, m_connection.Query<string>("SELECT text_value FROM data_adapter ORDER BY id"));
+		}
 		readonly MySqlConnection m_connection;
 	}
 }