Add MySqlBulkCopyColumnMapping. Fixes #773

Author: bgrainger

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

@@ -1,5 +1,5 @@
 ---
-lastmod: 2020-04-04
+lastmod: 2020-05-02
 date: 2019-11-11
 menu:
   main:
@@ -17,9 +17,6 @@ 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.
 
-`MySqlBulkCopy` currently requires that the source data (`DataTable` or `IDataReader`) contain all
-the columns of the destination table in the same order.
-
 For data that is in CSV or TSV format, use [`MySqlBulkLoader`](api/mysql-bulk-loader/) to bulk load the file.
 
 **Note:** This API is a unique feature of MySqlConnector; you must [switch to MySqlConnector](../../overview/installing/)
@@ -63,6 +60,14 @@ Name of the destination table on the server. (This name shouldn't be quoted or e
 
 If non-zero, this defines the number of rows to be processed before generating a notification event.
 
+`public List<MySqlBulkCopyColumnMapping> ColumnMappings { get; }`
+
+A collection of `MySqlBulkCopyColumnMapping` objects. If the columns being copied from the
+data source line up one-to-one with the columns in the destination table then populating this collection is
+unnecessary. Otherwise, this should be filled with a collection of `MySqlBulkCopyColumnMapping` objects
+specifying how source columns are to be mapped onto destination columns. If one column mapping is specified,
+then all must be specified.
+
 ### Methods
 
 `public void WriteToServer(DataTable dataTable);`
@@ -101,3 +106,41 @@ Receipt of a `RowsCopied` event does not imply that any rows have been sent to t
 
 The `MySqlRowsCopiedEventArgs.Abort` property can be set to `true` by the event handler to abort
 the copy.
+
+## MySqlBulkCopyColumnMapping
+
+Use `MySqlBulkCopyColumnMapping` to specify how to map columns in the source data to
+columns in the destination table.
+
+Set `SourceOrdinal` to the index of the source column to map. Set `DestinationColumn` to
+either the name of a column in the destination table, or the name of a user-defined variable.
+If a user-defined variable, you can use `Expression` to specify a MySQL expression that sets
+a destination column.
+
+Source columns that don't have an entry in `MySqlBulkCopy.ColumnMappings` will be ignored
+(unless the `ColumnMappings` collection is empty, in which case all columns will be mapped
+one-to-one).
+
+Columns containing binary data must be mapped using an expression that uses the `UNHEX` function.
+
+### Examples
+
+```csharp
+new MySqlBulkCopyColumnMapping
+{
+    SourceOrdinal = 2,
+    DestinationColumn = "user_name",
+},
+new MySqlBulkCopyColumnMapping
+{
+    SourceOrdinal = 0,
+    DestinationColumn = "@tmp",
+    Expression = "SET column_value = @tmp * 2",
+},
+new MySqlBulkCopyColumnMapping
+{
+    SourceOrdinal = 1,
+    DestinationColumn = "@tmp2",
+    Expression = "SET binary_column = UNHEX(@tmp2)",
+},
+```

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

@@ -3,6 +3,7 @@
 using System.Buffers.Text;
 using System.Collections.Generic;
 using System.Data;
+using System.Linq;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -20,6 +21,7 @@ public MySqlBulkCopy(MySqlConnection connection, MySqlTransaction? transaction =
 		{
 			m_connection = connection ?? throw new ArgumentNullException(nameof(connection));
 			m_transaction = transaction;
+			ColumnMappings = new List<MySqlBulkCopyColumnMapping>();
 		}
 
 		public int BulkCopyTimeout { get; set; }
@@ -44,6 +46,15 @@ public MySqlBulkCopy(MySqlConnection connection, MySqlTransaction? transaction =
 		/// </remarks>
 		public event MySqlRowsCopiedEventHandler? RowsCopied;
 
+		/// <summary>
+		/// A collection of <see cref="MySqlBulkCopyColumnMapping"/> objects. If the columns being copied from the
+		/// data source line up one-to-one with the columns in the destination table then populating this collection is
+		/// unnecessary. Otherwise, this should be filled with a collection of <see cref="MySqlBulkCopyColumnMapping"/> objects
+		/// specifying how source columns are to be mapped onto destination columns. If one column mapping is specified,
+		/// then all must be specified.
+		/// </summary>
+		public List<MySqlBulkCopyColumnMapping> ColumnMappings { get; }
+
 #if !NETSTANDARD1_3
 		public void WriteToServer(DataTable dataTable)
 		{
@@ -134,16 +145,17 @@ private async ValueTask WriteToServerAsync(IOBehavior ioBehavior, CancellationTo
 				closeConnection = true;
 			}
 
-			using (var cmd = new MySqlCommand("select * from " + QuoteIdentifier(tableName) + ";", m_connection, m_transaction))
-			using (var reader = (MySqlDataReader) await cmd.ExecuteReaderAsync(CommandBehavior.SchemaOnly, ioBehavior, cancellationToken).ConfigureAwait(false))
+			// if no user-supplied column mappings, compute them from the destination schema
+			if (ColumnMappings.Count == 0)
 			{
+				using var cmd = new MySqlCommand("select * from " + QuoteIdentifier(tableName) + ";", m_connection, m_transaction);
+				using var reader = (MySqlDataReader) await cmd.ExecuteReaderAsync(CommandBehavior.SchemaOnly, ioBehavior, cancellationToken).ConfigureAwait(false);
 				var schema = reader.GetColumnSchema();
 				for (var i = 0; i < schema.Count; i++)
 				{
 					if (schema[i].DataTypeName == "BIT")
 					{
-						bulkLoader.Columns.Add($"@col{i}");
-						bulkLoader.Expressions.Add($"`{reader.GetName(i)}` = CAST(@col{i} AS UNSIGNED)");
+						ColumnMappings.Add(new MySqlBulkCopyColumnMapping(i, $"@col{i}", $"`{reader.GetName(i)}` = CAST(@col{i} AS UNSIGNED)"));
 					}
 					else if (schema[i].DataTypeName == "YEAR")
 					{
@@ -155,17 +167,37 @@ private async ValueTask WriteToServerAsync(IOBehavior ioBehavior, CancellationTo
 						var type = schema[i].DataType;
 						if (type == typeof(byte[]) || (type == typeof(Guid) && (m_connection.GuidFormat == MySqlGuidFormat.Binary16 || m_connection.GuidFormat == MySqlGuidFormat.LittleEndianBinary16 || m_connection.GuidFormat == MySqlGuidFormat.TimeSwapBinary16)))
 						{
-							bulkLoader.Columns.Add($"@col{i}");
-							bulkLoader.Expressions.Add($"`{reader.GetName(i)}` = UNHEX(@col{i})");
+							ColumnMappings.Add(new MySqlBulkCopyColumnMapping(i, $"@col{i}", $"`{reader.GetName(i)}` = UNHEX(@col{i})"));
 						}
 						else
 						{
-							bulkLoader.Columns.Add(QuoteIdentifier(reader.GetName(i)));
+							ColumnMappings.Add(new MySqlBulkCopyColumnMapping(i, reader.GetName(i)));
 						}
 					}
 				}
 			}
 
+			// set columns and expressions from the column mappings
+			for (var i = 0; i < m_valuesEnumerator!.FieldCount; i++)
+			{
+				var columnMapping = ColumnMappings.FirstOrDefault(x => x.SourceOrdinal == i);
+				if (columnMapping is null)
+				{
+					bulkLoader.Columns.Add("@`\uE002\bignore`");
+				}
+				else
+				{
+					if (columnMapping.DestinationColumn.Length == 0)
+						throw new InvalidOperationException("MySqlBulkCopyColumnMapping.DestinationName is not set.");
+					if (columnMapping.DestinationColumn[0] == '@')
+						bulkLoader.Columns.Add(columnMapping.DestinationColumn);
+					else
+						bulkLoader.Columns.Add(QuoteIdentifier(columnMapping.DestinationColumn));
+					if (columnMapping.Expression is object)
+						bulkLoader.Expressions.Add(columnMapping.Expression);
+				}
+			}
+
 			await bulkLoader.LoadAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
 
 			if (closeConnection)

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

@@ -0,0 +1,50 @@
+using System;
+
+namespace MySql.Data.MySqlClient
+{
+	/// <summary>
+	/// <see cref="MySqlBulkCopyColumnMapping"/> specifies how to map columns in the source data to
+	/// destination columns when using <see cref="MySqlBulkCopy"/>.
+	/// </summary>
+	public sealed class MySqlBulkCopyColumnMapping
+	{
+		/// <summary>
+		/// Initializes <see cref="MySqlBulkCopyColumnMapping"/> with the default values.
+		/// </summary>
+		public MySqlBulkCopyColumnMapping()
+		{
+			DestinationColumn = "";
+		}
+
+		/// <summary>
+		/// Initializes <see cref="MySqlBulkCopyColumnMapping"/> to the specified values.
+		/// </summary>
+		/// <param name="sourceOrdinal">The ordinal position of the source column.</param>
+		/// <param name="destinationColumn">The name of the destination column.</param>
+		/// <param name="expression">The optional expression to be used to set the destination column.</param>
+		public MySqlBulkCopyColumnMapping(int sourceOrdinal, string destinationColumn, string? expression = null)
+		{
+			SourceOrdinal = sourceOrdinal;
+			DestinationColumn = destinationColumn ?? throw new ArgumentNullException(nameof(destinationColumn));
+			Expression = expression;
+		}
+
+		/// <summary>
+		/// The ordinal position of the source column to map from.
+		/// </summary>
+		public int SourceOrdinal { get; set; }
+
+		/// <summary>
+		/// The name of the destination column to copy to. To use an expression, this should be the name of a unique user-defined variable.
+		/// </summary>
+		public string DestinationColumn { get; set; }
+
+		/// <summary>
+		/// An optional expression for setting a destination column. To use an expression, the <see cref="DestinationColumn"/> should
+		/// be set to the name of a user-defined variable and this expression should set a column using that variable.
+		/// </summary>
+		/// <remarks>To populate a binary column, you must set <see cref="DestinationColumn"/> to a variable name, and <see cref="Expression"/> to an
+		/// expression that uses <code>UNHEX</code> to set the column value, e.g., <code>`destColumn` = UNHEX(@variableName)</code>.</remarks>
+		public string? Expression { get; set; }
+	}
+}

tests/SideBySide/BulkLoaderSync.cs

@@ -784,6 +784,59 @@ public void BulkCopyAbort(int notifyAfter, int rowCount, int abortAfter, int exp
 			using (var cmd = new MySqlCommand("select count(value) from bulk_copy_abort;", connection))
 				Assert.Equal(expectedCount, cmd.ExecuteScalar());
 		}
+
+		[Fact]
+		public void BulkCopyColumnMappings()
+		{
+			using var connection = new MySqlConnection(GetLocalConnectionString());
+			connection.Open();
+			using (var cmd = new MySqlCommand(@"drop table if exists bulk_copy_column_mapping;
+				create table bulk_copy_column_mapping(intvalue int, `text` text);", connection))
+			{
+				cmd.ExecuteNonQuery();
+			}
+
+			var bulkCopy = new MySqlBulkCopy(connection)
+			{
+				DestinationTableName = "bulk_copy_column_mapping",
+				ColumnMappings =
+				{
+					new MySqlBulkCopyColumnMapping(1, "@val", "intvalue = @val + 1"),
+					new MySqlBulkCopyColumnMapping(3, "text"),
+				},
+			};
+
+			var dataTable = new DataTable()
+			{
+				Columns =
+				{
+					new DataColumn("c1", typeof(int)),
+					new DataColumn("c2", typeof(int)),
+					new DataColumn("c3", typeof(string)),
+					new DataColumn("c4", typeof(string)),
+				},
+				Rows =
+				{
+					new object[] { 1, 100, "a", "A" },
+					new object[] { 2, 200, "bb", "BB" },
+					new object[] { 3, 300, "ccc", "CCC" },
+				}
+			};
+
+			bulkCopy.WriteToServer(dataTable);
+
+			using var reader = connection.ExecuteReader(@"select * from bulk_copy_column_mapping;");
+			Assert.True(reader.Read());
+			Assert.Equal(101, reader.GetValue(0));
+			Assert.Equal("A", reader.GetValue(1));
+			Assert.True(reader.Read());
+			Assert.Equal(201, reader.GetValue(0));
+			Assert.Equal("BB", reader.GetValue(1));
+			Assert.True(reader.Read());
+			Assert.Equal(301, reader.GetValue(0));
+			Assert.Equal("CCC", reader.GetValue(1));
+			Assert.False(reader.Read());
+		}
 #endif
 
 		[Fact]