Speed up inserts with MySqlDataAdapter. Fixes #1124

Author: bgrainger

src/MySqlConnector/MySqlDataAdapter.cs

@@ -1,3 +1,8 @@
+using System.Globalization;
+using System.Text;
+using System.Text.RegularExpressions;
+using MySqlConnector.Core;
+
 namespace MySqlConnector;
 
 public sealed class MySqlDataAdapter : DbDataAdapter
@@ -96,7 +101,134 @@ protected override int AddToBatch(IDbCommand command)
 
 	protected override void ClearBatch() => m_batch!.BatchCommands.Clear();
 
-	protected override int ExecuteBatch() => m_batch!.ExecuteNonQuery();
+	protected override int ExecuteBatch()
+	{
+		if (TryConvertToCommand(m_batch!) is MySqlCommand command)
+		{
+			command.Connection = m_batch!.Connection;
+			command.Transaction = m_batch.Transaction;
+			return command.ExecuteNonQuery();
+		}
+		else
+		{
+			return m_batch!.ExecuteNonQuery();
+		}
+	}
+
+	// Detects if the commands in 'batch' are all "INSERT" commands that can be combined into one large value list;
+	// returns a MySqlCommand with the combined SQL if so.
+	internal static MySqlCommand? TryConvertToCommand(MySqlBatch batch)
+	{
+		// ensure there are at least two commands
+		if (batch.BatchCommands.Count < 1)
+			return null;
+
+		// check for a parameterized command
+		var firstCommand = batch.BatchCommands[0];
+		if (firstCommand.Parameters.Count == 0)
+			return null;
+		firstCommand.Batch = batch;
+
+		// check that all commands have the same SQL
+		var sql = firstCommand.CommandText;
+		for (var i = 1; i < batch.BatchCommands.Count; i++)
+		{
+			if (batch.BatchCommands[i].CommandText != sql)
+				return null;
+		}
+
+		// check that it's an INSERT statement
+		if (!sql.StartsWith("INSERT INTO ", StringComparison.OrdinalIgnoreCase))
+			return null;
+
+		// check for "VALUES(...)" clause
+		var match = Regex.Match(sql, @"\bVALUES\s*\([^)]+\)\s*;?\s*$", RegexOptions.Singleline | RegexOptions.IgnoreCase | RegexOptions.CultureInvariant);
+		if (!match.Success)
+			return null;
+
+		// extract the parameters
+		var parser = new InsertSqlParser(firstCommand);
+		parser.Parse(sql);
+
+		// record the parameter indexes that were found
+		foreach (var parameterIndex in parser.ParameterIndexes)
+		{
+			if (parameterIndex < 0 || parameterIndex >= firstCommand.Parameters.Count)
+				return null;
+		}
+
+		// ensure that the VALUES(...) clause contained only parameters, and that all were consumed
+		var remainingValues = parser.CommandText.Substring(match.Index + 6).Trim();
+		remainingValues = remainingValues.TrimEnd(';').Trim().TrimStart('(').TrimEnd(')');
+		remainingValues = remainingValues.Replace(",", "");
+		if (!string.IsNullOrWhiteSpace(remainingValues))
+			return null;
+
+		// build one INSERT statement with concatenated VALUES
+		var combinedCommand = new MySqlCommand();
+		var sqlBuilder = new StringBuilder(sql.Substring(0, match.Index + 6));
+		var combinedParameterIndex = 0;
+		for (var i = 0; i < batch.BatchCommands.Count; i++)
+		{
+			var command = batch.BatchCommands[i];
+			if (i != 0)
+				sqlBuilder.Append(',');
+			sqlBuilder.Append('(');
+
+			for (var parameterIndex = 0; parameterIndex < parser.ParameterIndexes.Count; parameterIndex++)
+			{
+				if (parameterIndex != 0)
+					sqlBuilder.Append(',');
+				var parameterName = "@p" + combinedParameterIndex.ToString(CultureInfo.InvariantCulture);
+				sqlBuilder.Append(parameterName);
+				combinedParameterIndex++;
+				var parameter = command.Parameters[parser.ParameterIndexes[parameterIndex]].Clone();
+				parameter.ParameterName = parameterName;
+				combinedCommand.Parameters.Add(parameter);
+			}
+
+			sqlBuilder.Append(')');
+		}
+		sqlBuilder.Append(';');
+
+		combinedCommand.CommandText = sqlBuilder.ToString();
+		return combinedCommand;
+	}
+
+	internal sealed class InsertSqlParser : SqlParser
+	{
+		public InsertSqlParser(IMySqlCommand command)
+			: base(new StatementPreparer(command.CommandText!, null, command.CreateStatementPreparerOptions()))
+		{
+			CommandText = command.CommandText!;
+			m_parameters = command.RawParameters;
+			ParameterIndexes = new();
+		}
+
+		public List<int> ParameterIndexes { get; }
+
+		public string CommandText { get; private set; }
+
+		protected override void OnNamedParameter(int index, int length)
+		{
+			var name = CommandText.Substring(index, length);
+			var parameterIndex = m_parameters?.NormalizedIndexOf(name) ?? -1;
+			ParameterIndexes.Add(parameterIndex);
+
+			// overwrite the parameter name with spaces
+			CommandText = CommandText.Substring(0, index) + new string(' ', length) + CommandText.Substring(index + length);
+		}
+
+		protected override void OnPositionalParameter(int index)
+		{
+			ParameterIndexes.Add(ParameterIndexes.Count);
+
+			// overwrite the parameter placeholder with a space
+			CommandText = CommandText.Substring(0, index) + " " + CommandText.Substring(index + 1);
+		}
+
+		readonly MySqlParameterCollection? m_parameters;
+	}
 
 	MySqlBatch? m_batch;
 }

tests/SideBySide/DataAdapterTests.cs

@@ -1,3 +1,5 @@
+using System.Globalization;
+
 namespace SideBySide;
 
 public class DataAdapterTests : IClassFixture<DatabaseFixture>, IDisposable
@@ -235,5 +237,109 @@ public void BatchInsert()
 		Assert.Equal(new[] { null, "", "one", "two", "three", "four" }, m_connection.Query<string>("SELECT text_value FROM data_adapter ORDER BY id"));
 	}
 
+#if !BASELINE
+	[Theory]
+	[InlineData("INSERT INTO table(col1, col2) VALUES(@col1, @col2);", "@col1,@col2", "0,1")]
+	[InlineData("INSERT INTO table(col1, col2) VALUES(@col1, @col2);", "@col2,@col1", "1,0")]
+	[InlineData("INSERT INTO table(col1, col2) VALUES(@col1, @col2);", "@col1,@col2,@col3", "0,1")]
+	[InlineData("INSERT INTO table(col1, col2) VALUES(@col2, @col3);", "@col1,@col2,@col3", "1,2")]
+	[InlineData("INSERT INTO table(col1, col2) VALUES(?, ?);", "@col1,@col2", "0,1")]
+	[InlineData("INSERT INTO table(col1, col2)\nVALUES(@col1, @col2);", "@col1,@col2", "0,1")]
+	public void ExtractParameterIndexes(string sql, string parameterNames, string expectedIndexes)
+	{
+		var command = new MySqlCommand(sql, m_connection);
+		foreach (var parameterName in parameterNames.Split(','))
+			command.Parameters.Add(new MySqlParameter(parameterName, MySqlDbType.Int32));
+		var parser = new MySqlDataAdapter.InsertSqlParser(command);
+		parser.Parse(sql);
+		Assert.Equal(expectedIndexes.Split(',').Select(x => int.Parse(x, CultureInfo.InvariantCulture)), parser.ParameterIndexes);
+	}
+
+	[Theory]
+	[InlineData("SELECT * FROM table;", 2, 2, null)]
+	[InlineData("SELECT * FROM table VALUES;", 2, 2, null)]
+	[InlineData("INSERT INTO table VALUES(@param0, @param1)", 2, 2, "INSERT INTO table VALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT INTO table VALUES(@param0, @param1)", 3, 2, "INSERT INTO table VALUES(@p0,@p1),(@p2,@p3),(@p4,@p5);")]
+	[InlineData("INSERT INTO table VALUES(@param0, @param1)", 2, 3, "INSERT INTO table VALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT INTO table VALUES(@param0, @param1, @param2)", 2, 3, "INSERT INTO table VALUES(@p0,@p1,@p2),(@p3,@p4,@p5);")]
+	[InlineData("INSERT INTO table VALUES(@param0, @param1);", 2, 2, "INSERT INTO table VALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT INTO table VALUES(?, ?)", 2, 2, "INSERT INTO table VALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT INTO table  VALUES  (  @param0 , \n @param1  ) ;  ", 2, 2, "INSERT INTO table  VALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT INTO table VALUES(@param0, @param2);", 2, 2, null)]
+	[InlineData("INSERT INTO table\nVALUES\n(@param0, @param1);", 2, 2, "INSERT INTO table\nVALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT INTO `table` VALUES  (@\"param0\", @\"param1\");", 2, 2, "INSERT INTO `table` VALUES(@p0,@p1),(@p2,@p3);")]
+	[InlineData("INSERT\nINTO\ntable\nVALUES\n(@param0, @param1);", 2, 2, null)] // ideally should work but \n not supported
+	[InlineData("INSERT INTO table VALUES(1, @param0, @param1);", 2, 2, null)]
+	public void ConvertBatchToCommand(string insertSql, int commandCount, int parameterCount, string expected)
+	{
+		var batch = new MySqlBatch(m_connection);
+		for (var i = 0; i < commandCount; i++)
+		{
+			var batchCommand = new MySqlBatchCommand(insertSql);
+			for (var j = 0; j < parameterCount; j++)
+			{
+				batchCommand.Parameters.Add(new MySqlParameter($"@param{j}", MySqlDbType.Int32));
+			}
+			batch.BatchCommands.Add(batchCommand);
+		}
+
+		var command = MySqlDataAdapter.TryConvertToCommand(batch);
+		if (expected is null)
+		{
+			Assert.Null(command);
+		}
+		else
+		{
+			Assert.NotNull(command);
+			Assert.Equal(expected, command.CommandText);
+		}
+	}
+
+	[Fact]
+	public void ConvertBatchToCommandParameters()
+	{
+		var insertSql = "INSERT INTO table(col1, col2, col3) VALUES(@c1, @c2, @c1);";
+		var batch = new MySqlBatch(m_connection)
+		{
+			BatchCommands =
+			{
+				new MySqlBatchCommand(insertSql)
+				{
+					Parameters =
+					{
+						new MySqlParameter("@c1", 1),
+						new MySqlParameter("@c2", 2),
+					},
+				},
+				new MySqlBatchCommand(insertSql)
+				{
+					Parameters =
+					{
+						new MySqlParameter("@c1", 3),
+						new MySqlParameter("@c2", 4),
+					},
+				},
+			},
+		};
+
+		var command = MySqlDataAdapter.TryConvertToCommand(batch);
+		Assert.NotNull(command);
+		Assert.Equal("INSERT INTO table(col1, col2, col3) VALUES(@p0,@p1,@p2),(@p3,@p4,@p5);", command.CommandText);
+		Assert.Equal(6, command.Parameters.Count);
+		Assert.Equal("@p0", command.Parameters[0].ParameterName);
+		Assert.Equal(1, command.Parameters[0].Value);
+		Assert.Equal("@p1", command.Parameters[1].ParameterName);
+		Assert.Equal(2, command.Parameters[1].Value);
+		Assert.Equal("@p2", command.Parameters[2].ParameterName);
+		Assert.Equal(1, command.Parameters[2].Value);
+		Assert.Equal("@p3", command.Parameters[3].ParameterName);
+		Assert.Equal(3, command.Parameters[3].Value);
+		Assert.Equal("@p4", command.Parameters[4].ParameterName);
+		Assert.Equal(4, command.Parameters[4].Value);
+		Assert.Equal("@p5", command.Parameters[5].ParameterName);
+		Assert.Equal(3, command.Parameters[5].Value);
+	}
+#endif
+
 	readonly MySqlConnection m_connection;
 }