Support bulk loading from IDataReader.

Author: bgrainger

docs/content/api/mysql-bulk-copy.md

@@ -0,0 +1,68 @@
+---
+date: 2019-11-11
+menu:
+  main:
+    parent: api
+title: MySqlBulkCopy
+weight: 15
+---
+
+# MySqlBulkCopy
+
+`MySqlBulkCopy` lets you efficiently load a MySQL Server Table with data from another source.
+It is similar to the [`SqlBulkCopy`](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlbulkcopy) class
+for SQL Server.
+
+Due to [security features](../troubleshooting/load-data-local-infile/) in MySQL Server, the connection string
+**must** have `AllowLoadLocalInfile=true` in order to use this class.
+
+## Example Code
+
+```csharp
+// NOTE: to copy data between tables in the same database, use INSERT ... SELECT
+// https://dev.mysql.com/doc/refman/8.0/en/insert-select.html
+var dataTable = GetDataTableFromExternalSource();
+
+using (var connection = new MySqlConnection("...;AllowLoadLocalInfile=True"))
+{
+	await connection.OpenAsync();
+	var bulkCopy = new MySqlBulkCopy(connection);
+	bulkCopy.DestinationTableName = "some_table_name";
+	await bulkCopy.WriteToServerAsync(dataTable);
+}
+```
+
+## API Reference
+
+### Constructors
+
+`public MySqlBulkCopy(MySqlConnection connection, MySqlTransaction transaction = null)`
+
+Initializes a `MySqlBulkCopy` with the specified connection, and optionally the active transaction.
+
+### Properties
+
+`public int BulkCopyTimeout { get; set; }`
+
+The number of seconds for the operation to complete before it times out, or `0` for no timeout.
+
+`public string DestinationTableName { get; set; }`
+
+Name of the destination table on the server.
+
+### Methods
+
+`public void WriteToServer(DataTable dataTable);`
+
+`public Task WriteToServerAsync(DataTable dataTable, CancellationToken cancellationToken = default);`
+
+Copies all rows in the supplied `DataTable` to the destination table specified by the `DestinationTableName` property of the `MySqlBulkCopy` object.
+(This method is not available on `netstandard1.3`.)
+
+***
+
+`public void WriteToServer(IDataReader dataReader);`
+
+`public Task WriteToServerAsync(IDataReader dataReader, CancellationToken cancellationToken = default);`
+
+Copies all rows in the supplied `IDataReader` to the destination table specified by the `DestinationTableName` property of the `MySqlBulkCopy` object.

src/MySqlConnector/Core/IMySqlCommand.cs

@@ -11,6 +11,7 @@ internal interface IMySqlCommand
 	{
 		string? CommandText { get; }
 		CommandType CommandType { get; }
+		bool AllowUserVariables { get; }
 		CommandBehavior CommandBehavior { get; }
 		MySqlParameterCollection? RawParameters { get; }
 		PreparedStatements? TryGetPreparedStatements();
@@ -28,7 +29,7 @@ public static StatementPreparerOptions CreateStatementPreparerOptions(this IMySq
 		{
 			var connection = command.Connection!;
 			var statementPreparerOptions = StatementPreparerOptions.None;
-			if (connection.AllowUserVariables || command.CommandType == CommandType.StoredProcedure)
+			if (connection.AllowUserVariables || command.CommandType == CommandType.StoredProcedure || command.AllowUserVariables)
 				statementPreparerOptions |= StatementPreparerOptions.AllowUserVariables;
 			if (connection.DateTimeKind == DateTimeKind.Utc)
 				statementPreparerOptions |= StatementPreparerOptions.DateTimeUtc;

src/MySqlConnector/Core/ResultSet.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Buffers;
 using System.Collections.Generic;
+using System.Data;
 using System.Globalization;
 using System.IO;
 using System.Threading;
@@ -80,19 +82,37 @@ public async Task ReadResultSetHeaderAsync(IOBehavior ioBehavior)
 								MySqlBulkLoader.GetAndRemoveSource(localInfile.FileName) :
 								File.OpenRead(localInfile.FileName);
 
-							if (source is Stream stream)
+							IDisposable? disposable = null;
+							byte[]? buffer = null;
+							try
 							{
-								using (stream)
+								switch (source)
 								{
-									var readBuffer = new byte[65536];
+								case Stream stream:
+									disposable = stream;
+									buffer = ArrayPool<byte>.Shared.Rent(1048576);
 									int byteCount;
-									while ((byteCount = await stream.ReadAsync(readBuffer, 0, readBuffer.Length).ConfigureAwait(false)) > 0)
+									while ((byteCount = await stream.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false)) > 0)
 									{
-										payload = new PayloadData(new ArraySegment<byte>(readBuffer, 0, byteCount));
+										payload = new PayloadData(new ArraySegment<byte>(buffer, 0, byteCount));
 										await Session.SendReplyAsync(payload, ioBehavior, CancellationToken.None).ConfigureAwait(false);
 									}
+									break;
+
+								case IDataReader dataReader:
+									await MySqlBulkCopy.SendDataReaderAsync(Connection, dataReader, ioBehavior, CancellationToken.None).ConfigureAwait(false);
+									break;
+
+								default:
+									throw new InvalidOperationException("Unsupported Source type: {0}".FormatInvariant(source.GetType().Name));
 								}
 							}
+							finally
+							{
+								if (buffer is object)
+									ArrayPool<byte>.Shared.Return(buffer);
+								disposable?.Dispose();
+							}
 						}
 						catch (Exception ex)
 						{

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

@@ -18,6 +18,7 @@ public MySqlBatchCommand(string? commandText)
 
 		public string? CommandText { get; set; }
 		public CommandType CommandType { get; set; }
+		public bool AllowUserVariables => false;
 		public CommandBehavior CommandBehavior { get; set; }
 		public int RecordsAffected { get; set; }