Add support for provider-specifig options

PhenX-Pro · PhenX-Pro · commit f037a4ce86b0 · 2025-05-22T16:37:42.000+02:00
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlBulkInsertOptions.cs b/src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlBulkInsertOptions.cs
@@ -0,0 +1,11 @@
+﻿using PhenX.EntityFrameworkCore.BulkInsert.Options;
+
+namespace PhenX.EntityFrameworkCore.BulkInsert.MySql;
+
+/// <summary>
+/// Options specific to MySQL bulk insert.
+/// </summary>
+public class MySqlBulkInsertOptions : BulkInsertOptions
+{
+
+}
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlBulkInsertProvider.cs b/src/PhenX.EntityFrameworkCore.BulkInsert.MySql/MySqlBulkInsertProvider.cs
@@ -8,7 +8,7 @@
 
 namespace PhenX.EntityFrameworkCore.BulkInsert.MySql;
 
-internal class MySqlBulkInsertProvider : BulkInsertProviderBase<MySqlServerDialectBuilder>
+internal class MySqlBulkInsertProvider : BulkInsertProviderBase<MySqlServerDialectBuilder, MySqlBulkInsertOptions>
 {
     public MySqlBulkInsertProvider(ILogger<MySqlBulkInsertProvider>? logger = null) : base(logger)
     {
@@ -26,7 +26,7 @@ public MySqlBulkInsertProvider(ILogger<MySqlBulkInsertProvider>? logger = null)
     protected override string GetTempTableName(string tableName) => $"#_temp_bulk_insert_{tableName}";
 
     /// <inheritdoc />
-    public override BulkInsertOptions GetDefaultOptions() => new();
+    protected override MySqlBulkInsertOptions GetDefaultOptions() => new();
 
     /// <inheritdoc />
     public override Task<List<T>> BulkInsertReturnEntities<T>(
@@ -47,21 +47,25 @@ protected override async Task BulkInsert<T>(
         IEnumerable<T> entities,
         string tableName,
         PropertyAccessor[] properties,
-        BulkInsertOptions options,
+        MySqlBulkInsertOptions options,
         CancellationToken ctk
     )
     {
         var connection = (MySqlConnection)context.Database.GetDbConnection();
-        var sqlTransaction = context.Database.CurrentTransaction!.GetDbTransaction()
+
+        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 bulkCopy = new MySqlBulkCopy(connection, mySqlTransaction)
+        {
+            DestinationTableName = tableName,
+            BulkCopyTimeout = options.GetCopyTimeoutInSeconds(),
+        };
 
         var sourceOrdinal = 0;
         foreach (var prop in properties)
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert.PostgreSql/PostgreSqlBulkInsertProvider.cs b/src/PhenX.EntityFrameworkCore.BulkInsert.PostgreSql/PostgreSqlBulkInsertProvider.cs
@@ -10,7 +10,7 @@
 namespace PhenX.EntityFrameworkCore.BulkInsert.PostgreSql;
 
 [UsedImplicitly]
-internal class PostgreSqlBulkInsertProvider : BulkInsertProviderBase<PostgreSqlDialectBuilder>
+internal class PostgreSqlBulkInsertProvider : BulkInsertProviderBase<PostgreSqlDialectBuilder, BulkInsertOptions>
 {
     public PostgreSqlBulkInsertProvider(ILogger<PostgreSqlBulkInsertProvider>? logger = null) : base(logger)
     {
@@ -32,7 +32,7 @@ private string GetBinaryImportCommand(DbContext context, Type entityType, string
     }
 
     /// <inheritdoc />
-    public override BulkInsertOptions GetDefaultOptions() => new()
+    protected override BulkInsertOptions GetDefaultOptions() => new()
     {
         BatchSize = 50_000,
     };
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert.SqlServer/SqlServerBulkInsertOptions.cs b/src/PhenX.EntityFrameworkCore.BulkInsert.SqlServer/SqlServerBulkInsertOptions.cs
@@ -10,6 +10,9 @@ namespace PhenX.EntityFrameworkCore.BulkInsert.SqlServer;
 public class SqlServerBulkInsertOptions : BulkInsertOptions
 {
     /// <inheritdoc cref="SqlBulkCopyOptions"/>
-    public SqlBulkCopyOptions CopyOptions { get; init; } = SqlBulkCopyOptions.Default;
+    public SqlBulkCopyOptions CopyOptions { get; set; } = SqlBulkCopyOptions.Default;
+
+    /// <inheritdoc cref="SqlBulkCopy.EnableStreaming"/>
+    public bool EnableStreaming { get; set; } = false;
 
 }
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert.SqlServer/SqlServerBulkInsertProvider.cs b/src/PhenX.EntityFrameworkCore.BulkInsert.SqlServer/SqlServerBulkInsertProvider.cs
@@ -10,7 +10,7 @@
 namespace PhenX.EntityFrameworkCore.BulkInsert.SqlServer;
 
 [UsedImplicitly]
-internal class SqlServerBulkInsertProvider : BulkInsertProviderBase<SqlServerDialectBuilder>
+internal class SqlServerBulkInsertProvider : BulkInsertProviderBase<SqlServerDialectBuilder, SqlServerBulkInsertOptions>
 {
     public SqlServerBulkInsertProvider(ILogger<SqlServerBulkInsertProvider>? logger = null) : base(logger)
     {
@@ -27,7 +27,7 @@ public SqlServerBulkInsertProvider(ILogger<SqlServerBulkInsertProvider>? logger
     /// <inheritdoc />
     protected override string GetTempTableName(string tableName) => $"#_temp_bulk_insert_{tableName}";
 
-    public override BulkInsertOptions GetDefaultOptions() => new SqlServerBulkInsertOptions
+    protected override SqlServerBulkInsertOptions GetDefaultOptions() => new()
     {
         BatchSize = 50_000,
     };
@@ -39,17 +39,19 @@ protected override async Task BulkInsert<T>(
         IEnumerable<T> entities,
         string tableName,
         PropertyAccessor[] properties,
-        BulkInsertOptions options,
+        SqlServerBulkInsertOptions options,
         CancellationToken ctk
     )
     {
         var connection = (SqlConnection) context.Database.GetDbConnection();
         var sqlTransaction = context.Database.CurrentTransaction!.GetDbTransaction() as SqlTransaction;
 
-        using var bulkCopy = new SqlBulkCopy(connection, SqlBulkCopyOptions.TableLock, sqlTransaction);
+        using var bulkCopy = new SqlBulkCopy(connection, options.CopyOptions, sqlTransaction);
+
         bulkCopy.DestinationTableName = tableName;
         bulkCopy.BatchSize = options.BatchSize;
         bulkCopy.BulkCopyTimeout = options.GetCopyTimeoutInSeconds();
+        bulkCopy.EnableStreaming = options.EnableStreaming;
 
         foreach (var prop in properties)
         {
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert.Sqlite/SqliteBulkInsertProvider.cs b/src/PhenX.EntityFrameworkCore.BulkInsert.Sqlite/SqliteBulkInsertProvider.cs
@@ -12,7 +12,7 @@
 namespace PhenX.EntityFrameworkCore.BulkInsert.Sqlite;
 
 [UsedImplicitly]
-internal class SqliteBulkInsertProvider : BulkInsertProviderBase<SqliteDialectBuilder>
+internal class SqliteBulkInsertProvider : BulkInsertProviderBase<SqliteDialectBuilder, BulkInsertOptions>
 {
     public SqliteBulkInsertProvider(ILogger<SqliteBulkInsertProvider>? logger = null) : base(logger)
     {
@@ -30,7 +30,7 @@ public SqliteBulkInsertProvider(ILogger<SqliteBulkInsertProvider>? logger = null
     protected override string AddTableCopyBulkInsertId => "--"; // No need to add an ID column in SQLite
 
     /// <inheritdoc />
-    public override BulkInsertOptions GetDefaultOptions() => new()
+    protected override BulkInsertOptions GetDefaultOptions() => new()
     {
         BatchSize = 5,
     };
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert/Abstractions/IBulkInsertProvider.cs b/src/PhenX.EntityFrameworkCore.BulkInsert/Abstractions/IBulkInsertProvider.cs
@@ -19,7 +19,8 @@ internal Task<List<T>> BulkInsertReturnEntities<T>(
         BulkInsertOptions options,
         OnConflictOptions? onConflict = null,
         CancellationToken ctk = default
-    ) where T : class;
+    )
+        where T : class;
 
     /// <summary>
     /// Calls the provider to perform a bulk insert operation without returning the inserted entities.
@@ -31,10 +32,11 @@ internal Task BulkInsert<T>(
         BulkInsertOptions options,
         OnConflictOptions? onConflict = null,
         CancellationToken ctk = default
-    ) where T : class;
+    )
+        where T : class;
 
     /// <summary>
     /// Make the default options for the provider, can be a subclass of <see cref="BulkInsertOptions"/>.
     /// </summary>
-    internal BulkInsertOptions GetDefaultOptions();
+    internal BulkInsertOptions InternalGetDefaultOptions();
 }
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert/BulkInsertProviderBase.cs b/src/PhenX.EntityFrameworkCore.BulkInsert/BulkInsertProviderBase.cs
@@ -12,26 +12,27 @@
 
 namespace PhenX.EntityFrameworkCore.BulkInsert;
 
-internal abstract class BulkInsertProviderBase<TDialect> : IBulkInsertProvider
+internal abstract class BulkInsertProviderBase<TDialect, TOptions> : IBulkInsertProvider
     where TDialect : SqlDialectBuilder, new()
+    where TOptions : BulkInsertOptions, new()
 {
     protected readonly TDialect SqlDialect = new();
-    private readonly ILogger<BulkInsertProviderBase<TDialect>>? Logger;
+    private readonly ILogger<BulkInsertProviderBase<TDialect, TOptions>>? Logger;
 
     protected virtual string BulkInsertId => "_bulk_insert_id";
 
     protected abstract string CreateTableCopySql { get; }
     protected abstract string AddTableCopyBulkInsertId { get; }
 
-    protected BulkInsertProviderBase(ILogger<BulkInsertProviderBase<TDialect>>? logger = null)
+    protected BulkInsertProviderBase(ILogger<BulkInsertProviderBase<TDialect, TOptions>>? logger = null)
     {
         Logger = logger;
     }
 
     protected async Task<string> CreateTableCopyAsync<T>(
         bool sync,
         DbContext context,
-        BulkInsertOptions options,
+        TOptions options,
         CancellationToken cancellationToken = default) where T : class
     {
         var tableInfo = GetTableInfo(context, typeof(T));
@@ -148,11 +149,16 @@ public virtual async Task<List<T>> BulkInsertReturnEntities<T>(
         CancellationToken ctk = default
     ) where T : class
     {
+        if (options is not TOptions providerOptions)
+        {
+            throw new InvalidOperationException($"Invalid options type: {options.GetType().Name}. Expected: {typeof(TOptions).Name}");
+        }
+
         var (connection, wasClosed, transaction, wasBegan) = await context.GetConnection(sync, ctk);
 
-        var (tableName, _) = await PerformBulkInsertAsync(sync, context, entities, options, tempTableRequired: true, ctk: ctk);
+        var (tableName, _) = await PerformBulkInsertAsync(sync, context, entities, providerOptions, tempTableRequired: true, ctk: ctk);
 
-        var result = await CopyFromTempTableAsync<T>(sync, context, tableName, true, options, onConflict, cancellationToken: ctk);
+        var result = await CopyFromTempTableAsync<T>(sync, context, tableName, true, providerOptions, onConflict, cancellationToken: ctk);
 
         await Finish(sync, connection, wasClosed, transaction, wasBegan, ctk);
 
@@ -200,29 +206,36 @@ public virtual async Task BulkInsert<T>(
         CancellationToken ctk = default
     ) where T : class
     {
+        if (options is not TOptions providerOptions)
+        {
+            throw new InvalidOperationException($"Invalid options type: {options.GetType().Name}. Expected: {typeof(TOptions).Name}");
+        }
+
         if (onConflict != null)
         {
             var (connection, wasClosed, transaction, wasBegan) = await context.GetConnection(sync, ctk);
 
-            var (tableName, _) = await PerformBulkInsertAsync(sync, context, entities, options, tempTableRequired: true, ctk: ctk);
+            var (tableName, _) = await PerformBulkInsertAsync(sync, context, entities, providerOptions, tempTableRequired: true, ctk: ctk);
 
             await CopyFromTempTableAsync<T>(sync, context, tableName, false, options, onConflict, ctk);
 
             await Finish(sync, connection, wasClosed, transaction, wasBegan, ctk);
         }
         else
         {
-            await PerformBulkInsertAsync(sync, context, entities, options, tempTableRequired: false, ctk: ctk);
+            await PerformBulkInsertAsync(sync, context, entities, providerOptions, tempTableRequired: false, ctk: ctk);
         }
     }
 
-    public abstract BulkInsertOptions GetDefaultOptions();
+    public BulkInsertOptions InternalGetDefaultOptions() => GetDefaultOptions();
+
+    protected abstract TOptions GetDefaultOptions();
 
     private async Task<(string TableName, DbConnection Connection)> PerformBulkInsertAsync<T>(
         bool sync,
         DbContext context,
         IEnumerable<T> entities,
-        BulkInsertOptions options,
+        TOptions options,
         bool tempTableRequired,
         CancellationToken ctk = default) where T : class
     {
@@ -258,7 +271,7 @@ protected abstract Task BulkInsert<T>(
         IEnumerable<T> entities,
         string tableName,
         PropertyAccessor[] properties,
-        BulkInsertOptions options,
+        TOptions options,
         CancellationToken ctk
     ) where T : class;
 
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert/Enums/ProviderType.cs b/src/PhenX.EntityFrameworkCore.BulkInsert/Enums/ProviderType.cs
@@ -0,0 +1,27 @@
+﻿namespace PhenX.EntityFrameworkCore.BulkInsert.Enums;
+
+/// <summary>
+/// Enumeration of supported database providers.
+/// </summary>
+public enum ProviderType
+{
+    /// <summary>
+    /// SQL Server provider.
+    /// </summary>
+    SqlServer,
+
+    /// <summary>
+    /// PostgreSQL provider.
+    /// </summary>
+    PostgreSql,
+
+    /// <summary>
+    /// SQLite provider.
+    /// </summary>
+    Sqlite,
+
+    /// <summary>
+    /// MySQL provider.
+    /// </summary>
+    MySql,
+}
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert/Extensions/DbContextExtensions.cs b/src/PhenX.EntityFrameworkCore.BulkInsert/Extensions/DbContextExtensions.cs
@@ -5,6 +5,8 @@
 using Microsoft.EntityFrameworkCore.Metadata;
 using Microsoft.EntityFrameworkCore.Storage;
 
+using PhenX.EntityFrameworkCore.BulkInsert.Enums;
+
 namespace PhenX.EntityFrameworkCore.BulkInsert.Extensions;
 
 internal static class DbContextExtensions
@@ -61,4 +63,17 @@ internal static IProperty[] GetProperties(this DbContext context, Type entityTyp
 
         return (connection, wasClosed, transaction, wasBegan);
     }
+
+    /// <summary>
+    /// Tells if the current provider is the specified provider type.
+    /// </summary>
+    internal static bool IsProvider(this DbContext context, ProviderType providerType)
+    {
+        if (context.Database.ProviderName == null)
+        {
+            throw new InvalidOperationException("Database provider name is null.");
+        }
+
+        return context.Database.ProviderName.Contains(providerType.ToString(), StringComparison.OrdinalIgnoreCase);
+    }
 }
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert/Extensions/DbSetExtensions.cs b/src/PhenX.EntityFrameworkCore.BulkInsert/Extensions/DbSetExtensions.cs
@@ -126,17 +126,8 @@ public static async Task ExecuteBulkInsertAsync<T>(
         Action<BulkInsertOptions> configure,
         OnConflictOptions? onConflict = null,
         CancellationToken cancellationToken = default
-    )
-        where T : class
-    {
-        var dbSet = dbContext.Set<T>();
-        if (dbSet == null)
-        {
-            throw new InvalidOperationException($"DbSet of type {typeof(T).Name} not found in DbContext.");
-        }
-
-        await dbSet.ExecuteBulkInsertAsync(entities, configure, onConflict, cancellationToken);
-    }
+    ) where T : class
+        => await ExecuteBulkInsertAsync<T, BulkInsertOptions>(dbContext, entities, configure, onConflict, cancellationToken);
 
     /// <summary>
     /// Executes a bulk insert operation without returning the inserted/updated entities, from the DbContext.
@@ -147,14 +138,28 @@ public static async Task ExecuteBulkInsertAsync<T>(
         OnConflictOptions? onConflict = null,
         CancellationToken cancellationToken = default
     ) where T : class
+        => await ExecuteBulkInsertAsync<T, BulkInsertOptions>(dbContext, entities, _ => { }, onConflict, cancellationToken);
+
+    /// <summary>
+    /// Executes a bulk insert operation without returning the inserted/updated entities, from the DbContext.
+    /// </summary>
+    public static async Task ExecuteBulkInsertAsync<T, TOptions>(
+        this DbContext dbContext,
+        IEnumerable<T> entities,
+        Action<TOptions> configure,
+        OnConflictOptions? onConflict = null,
+        CancellationToken cancellationToken = default
+    )
+        where T : class
+        where TOptions : BulkInsertOptions
     {
         var dbSet = dbContext.Set<T>();
         if (dbSet == null)
         {
             throw new InvalidOperationException($"DbSet of type {typeof(T).Name} not found in DbContext.");
         }
 
-        await dbSet.ExecuteBulkInsertAsync<T, BulkInsertOptions>(entities, _ => { }, onConflict, cancellationToken);
+        await dbSet.ExecuteBulkInsertAsync(entities, configure, onConflict, cancellationToken);
     }
 
     /// <summary>
@@ -237,7 +242,7 @@ private static IBulkInsertProvider InitProvider<T, TOptions>(DbSet<T> dbSet, Act
         context = dbSet.GetDbContext();
         var provider = context.GetService<IBulkInsertProvider>();
 
-        var defaultOptions = provider.GetDefaultOptions();
+        var defaultOptions = provider.InternalGetDefaultOptions();
 
         if (defaultOptions is not TOptions castedOptions)
         {
diff --git a/src/PhenX.EntityFrameworkCore.BulkInsert/PhenX.EntityFrameworkCore.BulkInsert.csproj b/src/PhenX.EntityFrameworkCore.BulkInsert/PhenX.EntityFrameworkCore.BulkInsert.csproj
@@ -6,6 +6,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <InternalsVisibleTo Include="PhenX.EntityFrameworkCore.BulkInsert.Tests"/>
     <InternalsVisibleTo Include="PhenX.EntityFrameworkCore.BulkInsert.MySql"/>
     <InternalsVisibleTo Include="PhenX.EntityFrameworkCore.BulkInsert.PostgreSql"/>
     <InternalsVisibleTo Include="PhenX.EntityFrameworkCore.BulkInsert.SqlServer"/>
diff --git a/tests/PhenX.EntityFrameworkCore.BulkInsert.Tests/Tests/Basic/BasicTestsBase.cs b/tests/PhenX.EntityFrameworkCore.BulkInsert.Tests/Tests/Basic/BasicTestsBase.cs

Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@`
`10`	`10`	`namespace PhenX.EntityFrameworkCore.BulkInsert.PostgreSql;`
`11`	`11`
`12`	`12`	`[UsedImplicitly]`
`13`		`-internal class PostgreSqlBulkInsertProvider : BulkInsertProviderBase<PostgreSqlDialectBuilder>`
	`13`	`+internal class PostgreSqlBulkInsertProvider : BulkInsertProviderBase<PostgreSqlDialectBuilder, BulkInsertOptions>`
`14`	`14`	`{`
`15`	`15`	`public PostgreSqlBulkInsertProvider(ILogger<PostgreSqlBulkInsertProvider>? logger = null) : base(logger)`
`16`	`16`	`{`
`@@ -32,7 +32,7 @@ private string GetBinaryImportCommand(DbContext context, Type entityType, string`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`/// <inheritdoc />`
`35`		`- public override BulkInsertOptions GetDefaultOptions() => new()`
	`35`	`+ protected override BulkInsertOptions GetDefaultOptions() => new()`
`36`	`36`	`{`
`37`	`37`	`BatchSize = 50_000,`
`38`	`38`	`};`
Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,9 @@ namespace PhenX.EntityFrameworkCore.BulkInsert.SqlServer;`
`10`	`10`	`public class SqlServerBulkInsertOptions : BulkInsertOptions`
`11`	`11`	`{`
`12`	`12`	`/// <inheritdoc cref="SqlBulkCopyOptions"/>`
`13`		`- public SqlBulkCopyOptions CopyOptions { get; init; } = SqlBulkCopyOptions.Default;`
	`13`	`+ public SqlBulkCopyOptions CopyOptions { get; set; } = SqlBulkCopyOptions.Default;`
	`14`	`+`
	`15`	`+ /// <inheritdoc cref="SqlBulkCopy.EnableStreaming"/>`
	`16`	`+ public bool EnableStreaming { get; set; } = false;`
`14`	`17`
`15`	`18`	`}`
Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,7 @@`
`12`	`12`	`namespace PhenX.EntityFrameworkCore.BulkInsert.Sqlite;`
`13`	`13`
`14`	`14`	`[UsedImplicitly]`
`15`		`-internal class SqliteBulkInsertProvider : BulkInsertProviderBase<SqliteDialectBuilder>`
	`15`	`+internal class SqliteBulkInsertProvider : BulkInsertProviderBase<SqliteDialectBuilder, BulkInsertOptions>`
`16`	`16`	`{`
`17`	`17`	`public SqliteBulkInsertProvider(ILogger<SqliteBulkInsertProvider>? logger = null) : base(logger)`
`18`	`18`	`{`
`@@ -30,7 +30,7 @@ public SqliteBulkInsertProvider(ILogger<SqliteBulkInsertProvider>? logger = null`
`30`	`30`	`protected override string AddTableCopyBulkInsertId => "--"; // No need to add an ID column in SQLite`
`31`	`31`
`32`	`32`	`/// <inheritdoc />`
`33`		`- public override BulkInsertOptions GetDefaultOptions() => new()`
	`33`	`+ protected override BulkInsertOptions GetDefaultOptions() => new()`
`34`	`34`	`{`
`35`	`35`	`BatchSize = 5,`
`36`	`36`	`};`