Throw exception if not all rows were copied. Fixes #814

Author: bgrainger

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

@@ -128,6 +128,7 @@ private async ValueTask WriteToServerAsync(IOBehavior ioBehavior, CancellationTo
 #endif
 		{
 			var tableName = DestinationTableName ?? throw new InvalidOperationException("DestinationTableName must be set before calling WriteToServer");
+			m_wasAborted = false;
 
 			Log.Info("Starting bulk copy to {0}", tableName);
 			var bulkLoader = new MySqlBulkLoader(m_connection)
@@ -215,13 +216,19 @@ private async ValueTask WriteToServerAsync(IOBehavior ioBehavior, CancellationTo
 					throw new InvalidOperationException("SourceOrdinal {0} is an invalid value".FormatInvariant(columnMapping.SourceOrdinal));
 			}
 
-			await bulkLoader.LoadAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
+			var rowsInserted = await bulkLoader.LoadAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
 
 			if (closeConnection)
 				m_connection.Close();
 
 			Log.Info("Finished bulk copy to {0}", tableName);
 
+			if (!m_wasAborted && rowsInserted != RowsCopied)
+			{
+				Log.Error("Bulk copy to DestinationTableName={0} failed; RowsCopied={1}; RowsInserted={2}", tableName, RowsCopied, rowsInserted);
+				throw new MySqlException(MySqlErrorCode.BulkCopyFailed, "{0} rows were copied to {1} but only {2} were inserted.".FormatInvariant(RowsCopied, tableName, rowsInserted));
+			}
+
 #if !NETSTANDARD2_1 && !NETCOREAPP3_0
 			return default;
 #endif
@@ -330,6 +337,7 @@ await m_valuesEnumerator.MoveNextAsync().ConfigureAwait(false) :
 			finally
 			{
 				ArrayPool<byte>.Shared.Return(buffer);
+				m_wasAborted = eventArgs?.Abort ?? false;
 			}
 
 			static bool WriteValue(MySqlConnection connection, object? value, Span<byte> output, out int bytesWritten)
@@ -558,5 +566,6 @@ static bool WriteBytes(ReadOnlySpan<byte> value, Span<byte> output, out int byte
 		readonly MySqlConnection m_connection;
 		readonly MySqlTransaction? m_transaction;
 		IValuesEnumerator? m_valuesEnumerator;
+		bool m_wasAborted;
 	}
 }

src/MySqlConnector/MySql.Data.MySqlClient/MySqlErrorCode.g.cs

@@ -6,6 +6,11 @@ namespace MySql.Data.MySqlClient
 	[System.CodeDom.Compiler.GeneratedCode("https://gist.github.com/bgrainger/791cecb647d514a9dd2f3d83b2387e49", "2")]
 	public enum MySqlErrorCode
 	{
+		/// <summary>
+		/// Not all rows from the source supplied to <see cref="MySqlBulkCopy"/> were copied to <see cref="MySqlBulkCopy.DestinationTableName"/>.
+		/// </summary>
+		BulkCopyFailed = -2,
+
 		/// <summary>
 		/// The timeout period specified by <see cref="MySqlCommand.CommandTimeout"/> elapsed before the operation completed.
 		/// </summary>

tests/SideBySide/BulkLoaderSync.cs

@@ -918,6 +918,45 @@ public void BulkCopyColumnMappingsInvalidDestinationColumn()
 
 			Assert.Throws<InvalidOperationException>(() => bulkCopy.WriteToServer(dataTable));
 		}
+
+		[Fact]
+		public void BulkCopyDoesNotInsertAllRows()
+		{
+			using var connection = new MySqlConnection(GetLocalConnectionString());
+			connection.Open();
+
+			connection.Execute(@"drop table if exists bulk_copy_duplicate_pk;
+create table bulk_copy_duplicate_pk(id integer primary key, value text not null);");
+
+			var bcp = new MySqlBulkCopy(connection)
+			{
+				DestinationTableName = "bulk_copy_duplicate_pk"
+			};
+
+			var dataTable = new DataTable()
+			{
+				Columns =
+				{
+					new DataColumn("id", typeof(int)),
+					new DataColumn("value", typeof(string)),
+				},
+				Rows =
+				{
+					new object[] { 1, "a" },
+					new object[] { 1, "b" },
+					new object[] { 3, "c" },
+				}
+			};
+
+			try
+			{
+				bcp.WriteToServer(dataTable);
+				Assert.True(false, "Exception wasn't thrown");
+			}
+			catch (MySqlException ex) when (ex.Number == (int) MySqlErrorCode.BulkCopyFailed)
+			{
+			}
+		}
 #endif
 
 		[Fact]