Fix mappings for extra columns in schema (#1594). Fixes #1568

This creates/updates column mappings for columns in the DB schema that have a higher ordinal number than columns in the source data.

Signed-off-by: Bradley Grainger <[email protected]>

Author: bgrainger

src/MySqlConnector/MySqlBulkCopy.cs

@@ -266,7 +266,7 @@ private async ValueTask<MySqlBulkCopyResult> WriteToServerAsync(IOBehavior ioBeh
 		using (var reader = await cmd.ExecuteReaderAsync(CommandBehavior.SchemaOnly, ioBehavior, cancellationToken).ConfigureAwait(false))
 		{
 			var schema = reader.GetColumnSchema();
-			for (var i = 0; i < Math.Min(m_valuesEnumerator!.FieldCount, schema.Count); i++)
+			for (var i = 0; i < schema.Count; i++)
 			{
 				var destinationColumn = reader.GetName(i);
 				if (schema[i].DataTypeName == "BIT")
@@ -295,7 +295,7 @@ private async ValueTask<MySqlBulkCopyResult> WriteToServerAsync(IOBehavior ioBeh
 		}
 
 		// set columns and expressions from the column mappings
-		for (var i = 0; i < m_valuesEnumerator.FieldCount; i++)
+		for (var i = 0; i < m_valuesEnumerator!.FieldCount; i++)
 		{
 			var columnMapping = columnMappings.FirstOrDefault(x => x.SourceOrdinal == i);
 			if (columnMapping is null)

tests/IntegrationTests/BulkLoaderAsync.cs

@@ -635,6 +635,45 @@ public async Task BulkCopyNullDataReader()
 		var bulkCopy = new MySqlBulkCopy(connection);
 		await Assert.ThrowsAsync<ArgumentNullException>(async () => await bulkCopy.WriteToServerAsync(default(DbDataReader)));
 	}
+
+	[Fact]
+	public async Task BulkCopyGeometryAsync()
+	{
+		var dataTable = new DataTable()
+		{
+			Columns =
+			{
+				new DataColumn("geo_data", typeof(MySqlGeometry)),
+			},
+			Rows =
+			{
+				new object[] { MySqlGeometry.FromWkb(0, [1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 240, 63, 0, 0, 0, 0, 0, 0, 240, 63]) },
+			},
+		};
+
+		using var connection = new MySqlConnection(GetLocalConnectionString());
+		await connection.OpenAsync();
+		using (var cmd = new MySqlCommand(@"drop table if exists bulk_load_data_table;
+create table bulk_load_data_table(id BIGINT UNIQUE NOT NULL AUTO_INCREMENT, geo_data GEOMETRY NOT NULL);", connection))
+		{
+			await cmd.ExecuteNonQueryAsync();
+		}
+
+		var bc = new MySqlBulkCopy(connection)
+		{
+			DestinationTableName = "bulk_load_data_table",
+			ColumnMappings =
+			{
+				new()
+				{
+					SourceOrdinal = 0,
+					DestinationColumn = "geo_data",
+				},
+			},
+		};
+
+		await bc.WriteToServerAsync(dataTable);
+	}
 #endif
 
 	private static string GetConnectionString() => BulkLoaderSync.GetConnectionString();

tests/IntegrationTests/BulkLoaderSync.cs

@@ -1284,6 +1284,45 @@ public void BulkCopyDataTableConflictOption(MySqlBulkLoaderConflictOption confli
 		using (var cmd = new MySqlCommand("select b from bulk_load_data_table;", connection))
 			Assert.Equal(expected, cmd.ExecuteScalar());
 	}
+
+	[Fact]
+	public void BulkCopyGeometry()
+	{
+		var dataTable = new DataTable()
+		{
+			Columns =
+			{
+				new DataColumn("geo_data", typeof(MySqlGeometry)),
+			},
+			Rows =
+			{
+				new object[] { MySqlGeometry.FromWkb(0, [1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 240, 63, 0, 0, 0, 0, 0, 0, 240, 63]) },
+			},
+		};
+
+		using var connection = new MySqlConnection(GetLocalConnectionString());
+		connection.Open();
+		using (var cmd = new MySqlCommand(@"drop table if exists bulk_load_data_table;
+create table bulk_load_data_table(id BIGINT UNIQUE NOT NULL AUTO_INCREMENT, geo_data GEOMETRY NOT NULL);", connection))
+		{
+			cmd.ExecuteNonQuery();
+		}
+
+		var bc = new MySqlBulkCopy(connection)
+		{
+			DestinationTableName = "bulk_load_data_table",
+			ColumnMappings =
+			{
+				new()
+				{
+					SourceOrdinal = 0,
+					DestinationColumn = "geo_data",
+				},
+			},
+		};
+
+		bc.WriteToServer(dataTable);
+	}
 #endif
 
 	internal static string GetConnectionString() => AppConfig.ConnectionString;

tests/IntegrationTests/IntegrationTests.csproj

@@ -23,7 +23,7 @@
     <ServerGarbageCollection>true</ServerGarbageCollection>
     <ConcurrentGarbageCollection>true</ConcurrentGarbageCollection>
     <ThreadPoolMinThreads>64</ThreadPoolMinThreads>
-    <LangVersion>11.0</LangVersion>
+    <LangVersion>12.0</LangVersion>
     <ImplicitUsings>enable</ImplicitUsings>
     <NoWarn>$(NoWarn);SA0001;SA1021;SA1133;xUnit1030</NoWarn>
   </PropertyGroup>