MySql support.

.gitignore

@@ -88,3 +88,6 @@ fabric.properties
 
 # Nuget assets
 /nupkgs
+
+# Visual Studio Files
+.vs

PhenX.EntityFrameworkCore.BulkInsert.sln

@@ -1,5 +1,8 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 17
+VisualStudioVersion = 17.12.35707.178 d17.12
+MinimumVisualStudioVersion = 10.0.40219.1
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PhenX.EntityFrameworkCore.BulkInsert", "src\PhenX.EntityFrameworkCore.BulkInsert\PhenX.EntityFrameworkCore.BulkInsert.csproj", "{56CA0AE2-6EAB-4394-9E06-132558551251}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PhenX.EntityFrameworkCore.BulkInsert.PostgreSql", "src\PhenX.EntityFrameworkCore.BulkInsert.PostgreSql\PhenX.EntityFrameworkCore.BulkInsert.PostgreSql.csproj", "{F37308A8-1C3C-44D2-9440-670DF76A8C31}"
@@ -30,6 +33,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "root", "root", "{45366E91-4
 		README.md = README.md
 	EndProjectSection
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PhenX.EntityFrameworkCore.BulkInsert.MySql", "src\PhenX.EntityFrameworkCore.BulkInsert.MySql\PhenX.EntityFrameworkCore.BulkInsert.MySql.csproj", "{17649766-EA68-4333-8DA8-47B014A8B2CC}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -60,6 +65,13 @@ Global
 		{450E859C-411F-4D67-A0B4-4E02C3D30E14}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{450E859C-411F-4D67-A0B4-4E02C3D30E14}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{450E859C-411F-4D67-A0B4-4E02C3D30E14}.Release|Any CPU.Build.0 = Release|Any CPU
+		{17649766-EA68-4333-8DA8-47B014A8B2CC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{17649766-EA68-4333-8DA8-47B014A8B2CC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{17649766-EA68-4333-8DA8-47B014A8B2CC}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{17649766-EA68-4333-8DA8-47B014A8B2CC}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{56CA0AE2-6EAB-4394-9E06-132558551251} = {CBEBA2A8-79E0-412E-93C1-C88F4473D78B}
@@ -68,5 +80,6 @@ Global
 		{EDCCED5F-D456-45E2-81A6-1077977F042B} = {F8A83782-311C-454D-8B97-B3FB86478BF4}
 		{E4EB1C53-575C-45F8-924A-93DC42E8ACCA} = {F8A83782-311C-454D-8B97-B3FB86478BF4}
 		{450E859C-411F-4D67-A0B4-4E02C3D30E14} = {CBEBA2A8-79E0-412E-93C1-C88F4473D78B}
+		{17649766-EA68-4333-8DA8-47B014A8B2CC} = {CBEBA2A8-79E0-412E-93C1-C88F4473D78B}
 	EndGlobalSection
 EndGlobal

src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlBulkInsertProvider.cs

@@ -0,0 +1,80 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Storage;
+using Microsoft.Extensions.Logging;
+
+using MySqlConnector;
+
+using PhenX.EntityFrameworkCore.BulkInsert.Options;
+
+namespace PhenX.EntityFrameworkCore.BulkInsert.MySql;
+
+internal class MySqlBulkInsertProvider : BulkInsertProviderBase<MySqlServerDialectBuilder>
+{
+    public MySqlBulkInsertProvider(ILogger<MySqlBulkInsertProvider>? logger = null) : base(logger)
+    {
+    }
+
+    //language=sql
+    /// <inheritdoc />
+    protected override string CreateTableCopySql => "CREATE TEMPORARY TABLE {0} SELECT * FROM {1} WHERE 1 = 0;";
+
+    //language=sql
+    /// <inheritdoc />
+    protected override string AddTableCopyBulkInsertId => $"ALTER TABLE {{0}} ADD {BulkInsertId} INT AUTO_INCREMENT PRIMARY KEY;";
+
+    /// <inheritdoc />
+    protected override string GetTempTableName(string tableName) => $"#_temp_bulk_insert_{tableName}";
+
+    /// <inheritdoc />
+    public override Task<List<T>> BulkInsertReturnEntities<T>(
+        bool sync,
+        DbContext context,
+        IEnumerable<T> entities,
+        BulkInsertOptions options,
+        OnConflictOptions? onConflict = null,
+        CancellationToken ctk = default)
+    {
+        throw new NotSupportedException("Provider does not support returning entities.");
+    }
+
+    /// <inheritdoc />
+    protected override async Task BulkInsert<T>(
+        bool sync,
+        DbContext context,
+        IEnumerable<T> entities,
+        string tableName,
+        PropertyAccessor[] properties,
+        BulkInsertOptions options,
+        CancellationToken ctk
+    )
+    {
+        var connection = (MySqlConnection)context.Database.GetDbConnection();
+        var sqlTransaction = context.Database.CurrentTransaction!.GetDbTransaction()
+            ?? throw new InvalidOperationException("No open transaction found.");
+        if (sqlTransaction is not MySqlTransaction mySqlTransaction)
+        {
+            throw new InvalidOperationException($"Invalid transaction foud, got {sqlTransaction.GetType()}.");
+        }
+
+        var bulkCopy = new MySqlBulkCopy(connection, mySqlTransaction);
+        bulkCopy.DestinationTableName = tableName;
+        bulkCopy.BulkCopyTimeout = options.GetCopyTimeoutInSeconds();
+
+        var sourceOrdinal = 0;
+        foreach (var prop in properties)
+        {
+            bulkCopy.ColumnMappings.Add(new MySqlBulkCopyColumnMapping(sourceOrdinal, prop.ColumnName));
+            sourceOrdinal++;
+        }
+
+        if (sync)
+        {
+            // ReSharper disable once MethodHasAsyncOverloadWithCancellation
+            bulkCopy.WriteToServer(new EnumerableDataReader<T>(entities, properties));
+        }
+        else
+        {
+            await bulkCopy.WriteToServerAsync(new EnumerableDataReader<T>(entities, properties), ctk);
+        }
+    }
+}

src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlDbContextOptionsExtensions.cs

@@ -0,0 +1,22 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+
+namespace PhenX.EntityFrameworkCore.BulkInsert.MySql;
+
+/// <summary>
+/// DbContext options extension for MySql.
+/// </summary>
+public static class MySqlDbContextOptionsExtensions
+{
+    /// <summary>
+    /// Configures the DbContext to use the MySql bulk insert provider.
+    /// </summary>
+    public static DbContextOptionsBuilder UseBulkInsertMySql(this DbContextOptionsBuilder optionsBuilder)
+    {
+        var extension = optionsBuilder.Options.FindExtension<BulkInsertOptionsExtension<MySqlBulkInsertProvider>>() ?? new BulkInsertOptionsExtension<MySqlBulkInsertProvider>();
+
+        ((IDbContextOptionsBuilderInfrastructure)optionsBuilder).AddOrUpdateExtension(extension);
+
+        return optionsBuilder;
+    }
+}

src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlDialectBuilder.cs

@@ -0,0 +1,57 @@
+using System.Text;
+
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata;
+
+using PhenX.EntityFrameworkCore.BulkInsert.Dialect;
+using PhenX.EntityFrameworkCore.BulkInsert.Options;
+
+namespace PhenX.EntityFrameworkCore.BulkInsert.MySql;
+
+internal class MySqlServerDialectBuilder : SqlDialectBuilder
+{
+    protected override string OpenDelimiter => "`";
+
+    protected override string CloseDelimiter => "`";
+
+    protected override bool SupportsMoveRows => false;
+
+    protected override void AppendConflictCondition<T>(StringBuilder sql, OnConflictOptions<T> onConflictTyped)
+    {
+        throw new NotSupportedException("Conflict conditions are not supported in MYSQL");
+    }
+
+    protected override void AppendOnConflictUpdate(StringBuilder sql, IEnumerable<string> updates)
+    {
+        sql.AppendLine("UPDATE");
+
+        var i = 0;
+        foreach (var update in updates)
+        {
+            if (i > 0)
+            {
+                sql.Append(", ");
+            }
+
+            sql.Append(update);
+            i++;
+        }
+    }
+
+    protected override void AppendOnConflictStatement(StringBuilder sql)
+    {
+        sql.Append("ON DUPLICATE KEY");
+    }
+
+    protected override void AppendDoNothing(StringBuilder sql, IProperty[] insertedProperties)
+    {
+        var columnName = insertedProperties[0].GetColumnName();
+
+        sql.Append($"UPDATE {Quote(columnName)} = {GetExcludedColumnName(columnName)}");
+    }
+
+    protected override string GetExcludedColumnName(string columnName)
+    {
+        return $"VALUES({Quote(columnName)})";
+    }
+}

src/PhenX.EntityFrameworkCore.BulkInsert.MySql/PhenX.EntityFrameworkCore.BulkInsert.MySql.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <ItemGroup>
+    <ProjectReference Include="..\PhenX.EntityFrameworkCore.BulkInsert\PhenX.EntityFrameworkCore.BulkInsert.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Condition="'$(TargetFramework)' == 'net8.0'" Version="8.0.3" />
+    <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Condition="'$(TargetFramework)' == 'net9.0'" Version="9.0.0-preview.3.efcore.9.0.0" />
+  </ItemGroup>
+
+</Project>

src/PhenX.EntityFrameworkCore.BulkInsert.SqlServer/SqlServerBulkInsertProvider.cs

@@ -48,7 +48,7 @@ CancellationToken ctk
 
         foreach (var prop in properties)
         {
-            bulkCopy.ColumnMappings.Add(prop.Name, SqlDialect.Quote(prop.ColumnName));
+            bulkCopy.ColumnMappings.Add(prop.Name, prop.ColumnName);
         }
 
         if (sync)

src/PhenX.EntityFrameworkCore.BulkInsert.SqlServer/SqlServerDialectBuilder.cs

@@ -1,4 +1,4 @@
-using System.Linq.Expressions;
+using System.Linq.Expressions;
 using System.Text;
 
 using Microsoft.EntityFrameworkCore;
@@ -31,6 +31,11 @@ public override string BuildMoveDataSql<T>(DbContext context, string source,
 
         var q = new StringBuilder();
 
+        if (options.CopyGeneratedColumns)
+        {
+            q.AppendLine($"SET IDENTITY_INSERT {target} ON;");
+        }
+
         // Merge handling
         if (onConflict is OnConflictOptions<T> onConflictTyped && onConflictTyped.Match != null)
         {
@@ -39,7 +44,7 @@ public override string BuildMoveDataSql<T>(DbContext context, string source,
                 matchColumns.Select(col => $"TARGET.{col} = SOURCE.{col}"));
 
             var updateSet = onConflictTyped.Update != null
-                ? string.Join(", ", GetUpdates(context, onConflictTyped.Update))
+                ? string.Join(", ", GetUpdates(context, insertedProperties, onConflictTyped.Update))
                 : null;
 
             q.AppendLine($"MERGE INTO {target} AS TARGET");
@@ -79,11 +84,16 @@ public override string BuildMoveDataSql<T>(DbContext context, string source,
 
         q.AppendLine(";");
 
+        if (options.CopyGeneratedColumns)
+        {
+            q.AppendLine($"SET IDENTITY_INSERT {target} OFF;");
+        }
+
         return q.ToString();
     }
 
-    protected override string GetExcludedColumnName<TEntity>(DbContext context, MemberExpression member)
+    protected override string GetExcludedColumnName(string columnName)
     {
-        return $"SOURCE.{GetColumnName<TEntity>(context, member.Member.Name)}";
+        return $"SOURCE.{columnName}";
     }
 }

src/PhenX.EntityFrameworkCore.BulkInsert.Sqlite/SqliteBulkInsertProvider.cs

@@ -82,9 +82,10 @@ private static SqliteType GetSqliteType(Type clrType)
     }
 
     private DbCommand GetInsertCommand(DbContext context, Type entityType, string tableName,
+        BulkInsertOptions options,
         int batchSize)
     {
-        var columns = context.GetProperties(entityType, false);
+        var columns = context.GetProperties(entityType, options.CopyGeneratedColumns);
         var cmd = context.Database.GetDbConnection().CreateCommand();
 
         var sqliteColumns = columns
@@ -136,7 +137,7 @@ CancellationToken ctk
         var batchSize = options.BatchSize ?? 5;
         batchSize = Math.Min(batchSize, maxParams / properties.Length);
 
-        await using var insertCommand = GetInsertCommand(context, typeof(T), tableName, batchSize);
+        await using var insertCommand = GetInsertCommand(context, typeof(T), tableName, options, batchSize);
 
         foreach (var chunk in entities.Chunk(batchSize))
         {
@@ -149,7 +150,7 @@ CancellationToken ctk
             // Last chunk
             else
             {
-                var partialInsertCommand = GetInsertCommand(context, typeof(T), tableName, chunk.Length);
+                var partialInsertCommand = GetInsertCommand(context, typeof(T), tableName, options, chunk.Length);
 
                 FillValues(chunk, partialInsertCommand.Parameters, properties);
                 await ExecuteCommand(sync, partialInsertCommand, ctk);

src/PhenX.EntityFrameworkCore.BulkInsert/BulkInsertProviderBase.cs

@@ -31,13 +31,14 @@ protected BulkInsertProviderBase(ILogger<BulkInsertProviderBase<TDialect>>? logg
     protected async Task<string> CreateTableCopyAsync<T>(
         bool sync,
         DbContext context,
+        BulkInsertOptions options,
         CancellationToken cancellationToken = default) where T : class
     {
         var tableInfo = GetTableInfo(context, typeof(T));
         var tableName = QuoteTableName(tableInfo.SchemaName, tableInfo.TableName);
         var tempTableName = QuoteTableName(null, GetTempTableName(tableInfo.TableName));
 
-        var keptColumns = string.Join(", ", GetQuotedColumns(context, typeof(T), false));
+        var keptColumns = string.Join(", ", GetQuotedColumns(context, typeof(T), options.CopyGeneratedColumns));
         var query = string.Format(CreateTableCopySql, tempTableName, tableName, keptColumns);
 
         await ExecuteAsync(sync, context, query, cancellationToken);
@@ -108,13 +109,21 @@ private async Task<List<TResult>> CopyFromTempTableWithoutKeysAsync<T, TResult>(
     {
         var (schemaName, tableName, _) = GetTableInfo(context, typeof(T));
         var quotedTableName = QuoteTableName(schemaName, tableName);
-
-        var movedProperties = context.GetProperties(typeof(T), false);
+        var movedProperties = context.GetProperties(typeof(T), options.CopyGeneratedColumns);
         var returnedProperties = returnData ? context.GetProperties(typeof(T)) : [];
 
         var query = SqlDialect.BuildMoveDataSql<T>(context, tempTableName, quotedTableName, movedProperties, returnedProperties, options, onConflict);
 
         if (returnData)
+        {
+            return await QueryAsync(sync, context, query, cancellationToken);
+        }
+
+        // If not returning data, just execute the command
+        await ExecuteAsync(sync, context, query, cancellationToken);
+        return [];
+
+        static async Task<List<TResult>> QueryAsync(bool sync, DbContext context, string query, CancellationToken cancellationToken)
         {
             // Use EF to execute the query and return the results
             IQueryable<TResult> queryable = context
@@ -128,13 +137,9 @@ private async Task<List<TResult>> CopyFromTempTableWithoutKeysAsync<T, TResult>(
 
             return await queryable.ToListAsync(cancellationToken: cancellationToken);
         }
-
-        // If not returning data, just execute the command
-        await ExecuteAsync(sync, context, query, cancellationToken);
-        return [];
     }
 
-    public async Task<List<T>> BulkInsertReturnEntities<T>(
+    public virtual async Task<List<T>> BulkInsertReturnEntities<T>(
         bool sync,
         DbContext context,
         IEnumerable<T> entities,
@@ -163,10 +168,12 @@ private static async Task Finish(bool sync, DbConnection connection, bool wasClo
             {
                 // ReSharper disable once MethodHasAsyncOverloadWithCancellation
                 transaction.Commit();
+                transaction.Dispose();
             }
             else
             {
                 await transaction.CommitAsync(ctk);
+                await transaction.DisposeAsync();
             }
         }
 
@@ -184,7 +191,7 @@ private static async Task Finish(bool sync, DbConnection connection, bool wasClo
         }
     }
 
-    public async Task BulkInsert<T>(
+    public virtual async Task BulkInsert<T>(
         bool sync,
         DbContext context,
         IEnumerable<T> entities,
@@ -225,11 +232,11 @@ public async Task BulkInsert<T>(
         var (connection, wasClosed, transaction, wasBegan) = await context.GetConnection(sync, ctk);
 
         var tableName = tempTableRequired
-            ? await CreateTableCopyAsync<T>(sync, context, ctk)
+            ? await CreateTableCopyAsync<T>(sync, context, options, ctk)
             : GetQuotedTableName(context, typeof(T));
 
         var properties = context
-            .GetProperties(typeof(T), false)
+            .GetProperties(typeof(T), options.CopyGeneratedColumns)
             .Select(p => new PropertyAccessor(p))
             .ToArray();