Add sync mode (Fixes #4)

Author: fabien.menager

src/PhenX.EntityFrameworkCore.BulkInsert.PostgreSql/PostgreSqlBulkInsertProvider.cs

@@ -28,6 +28,7 @@ private string GetBinaryImportCommand(DbContext context, Type entityType, string
 
     /// <inheritdoc />
     protected override async Task BulkInsert<T>(
+        bool sync,
         DbContext context,
         IEnumerable<T> entities,
         string tableName,
@@ -39,20 +40,51 @@ protected override async Task BulkInsert<T>(
 
         var importCommand = GetBinaryImportCommand(context, typeof(T), tableName);
 
-        await using var writer = await connection.BeginBinaryImportAsync(importCommand, ctk);
+        var writer = sync
+            // ReSharper disable once MethodHasAsyncOverloadWithCancellation
+            ? connection.BeginBinaryImport(importCommand)
+            : await connection.BeginBinaryImportAsync(importCommand, ctk);
 
         foreach (var entity in entities)
         {
-            await writer.StartRowAsync(ctk);
+            if (sync)
+            {
+                // ReSharper disable once MethodHasAsyncOverloadWithCancellation
+                writer.StartRow();
+            }
+            else
+            {
+                await writer.StartRowAsync(ctk);
+            }
 
             foreach (var property in properties)
             {
                 var value = property.GetValue(entity);
 
-                await writer.WriteAsync(value, ctk);
+                if (sync)
+                {
+                    // ReSharper disable once MethodHasAsyncOverloadWithCancellation
+                    writer.Write(value);
+                }
+                else
+                {
+                    await writer.WriteAsync(value, ctk);
+                }
             }
         }
 
-        await writer.CompleteAsync(ctk);
+        if (sync)
+        {
+            // ReSharper disable once MethodHasAsyncOverloadWithCancellation
+            writer.Complete();
+            // ReSharper disable once MethodHasAsyncOverload
+            writer.Dispose();
+        }
+        else
+        {
+            await writer.CompleteAsync(ctk);
+            await writer.DisposeAsync();
+        }
+
     }
 }

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

@@ -23,9 +23,15 @@ internal class SqlServerBulkInsertProvider : BulkInsertProviderBase<SqlServerDia
     protected override string GetTempTableName(string tableName) => $"#_temp_bulk_insert_{tableName}";
 
     /// <inheritdoc />
-    protected override async Task BulkInsert<T>(DbContext context, IEnumerable<T> entities,
+    protected override async Task BulkInsert<T>(
+        bool sync,
+        DbContext context,
+        IEnumerable<T> entities,
         string tableName,
-        PropertyAccessor[] properties, BulkInsertOptions options, CancellationToken ctk)
+        PropertyAccessor[] properties,
+        BulkInsertOptions options,
+        CancellationToken ctk
+    )
     {
         var connection = context.Database.GetDbConnection();
         var sqlTransaction = context.Database.CurrentTransaction!.GetDbTransaction() as SqlTransaction;
@@ -40,6 +46,14 @@ protected override async Task BulkInsert<T>(DbContext context, IEnumerable<T> en
             bulkCopy.ColumnMappings.Add(prop.Name, SqlDialect.Quote(prop.ColumnName));
         }
 
-        await bulkCopy.WriteToServerAsync(new EnumerableDataReader<T>(entities, properties), ctk);
+        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.Sqlite/SqliteBulkInsertProvider.cs

@@ -25,11 +25,12 @@ internal class SqliteBulkInsertProvider : BulkInsertProviderBase<SqliteDialectBu
     protected override string AddTableCopyBulkInsertId => "--"; // No need to add an ID column in SQLite
 
     /// <inheritdoc />
-    protected override Task AddBulkInsertIdColumn<T>(DbContext context, CancellationToken cancellationToken,
-        string tempTableName) where T : class
-    {
-        return Task.CompletedTask;
-    }
+    protected override Task AddBulkInsertIdColumn<T>(
+        bool sync,
+        DbContext context,
+        string tempTableName,
+        CancellationToken cancellationToken
+    ) where T : class => Task.CompletedTask;
 
     /// <summary>
     /// Taken from https://github.com/dotnet/efcore/blob/667c569c49a1ab7e142621395d3f14f2af0508b4/src/Microsoft.Data.Sqlite.Core/SqliteValueBinder.cs#L231
@@ -116,8 +117,15 @@ private DbCommand GetInsertCommand(DbContext context, Type entityType, string ta
     }
 
     /// <inheritdoc />
-    protected override async Task BulkInsert<T>(DbContext context, IEnumerable<T> entities,
-        string tableName, PropertyAccessor[] properties, BulkInsertOptions options, CancellationToken ctk) where T : class
+    protected override async Task BulkInsert<T>(
+        bool sync,
+        DbContext context,
+        IEnumerable<T> entities,
+        string tableName,
+        PropertyAccessor[] properties,
+        BulkInsertOptions options,
+        CancellationToken ctk
+    ) where T : class
     {
         const int maxParams = 1000;
         var batchSize = options.BatchSize ?? 5;
@@ -131,20 +139,32 @@ protected override async Task BulkInsert<T>(DbContext context, IEnumerable<T> en
             if (chunk.Length == batchSize)
             {
                 FillValues(chunk, insertCommand.Parameters, properties);
-
-                await insertCommand.ExecuteNonQueryAsync(ctk);
+                await ExecuteCommand(sync, insertCommand, ctk);
             }
             // Last chunk
             else
             {
                 var partialInsertCommand = GetInsertCommand(context, typeof(T), tableName, chunk.Length);
-                FillValues(chunk, partialInsertCommand.Parameters, properties);
 
-                await partialInsertCommand.ExecuteNonQueryAsync(ctk);
+                FillValues(chunk, partialInsertCommand.Parameters, properties);
+                await ExecuteCommand(sync, partialInsertCommand, ctk);
             }
         }
     }
 
+    private static async Task ExecuteCommand(bool sync, DbCommand insertCommand, CancellationToken ctk)
+    {
+        if (sync)
+        {
+            // ReSharper disable once MethodHasAsyncOverloadWithCancellation
+            insertCommand.ExecuteNonQuery();
+        }
+        else
+        {
+            await insertCommand.ExecuteNonQueryAsync(ctk);
+        }
+    }
+
     private static void FillValues<T>(T[] chunk, DbParameterCollection parameters, PropertyAccessor[] properties) where T : class
     {
         var index = 0;

src/PhenX.EntityFrameworkCore.BulkInsert/Abstractions/IBulkInsertProvider.cs

@@ -12,7 +12,8 @@ internal interface IBulkInsertProvider
     /// <summary>
     /// Calls the provider to perform a bulk insert operation.
     /// </summary>
-    internal Task<List<T>> BulkInsertWithIdentityAsync<T>(
+    internal Task<List<T>> BulkInsertReturnEntities<T>(
+        bool sync,
         DbContext context,
         IEnumerable<T> entities,
         BulkInsertOptions options,
@@ -23,7 +24,8 @@ internal Task<List<T>> BulkInsertWithIdentityAsync<T>(
     /// <summary>
     /// Calls the provider to perform a bulk insert operation without returning the inserted entities.
     /// </summary>
-    internal Task BulkInsertWithoutReturnAsync<T>(
+    internal Task BulkInsert<T>(
+        bool sync,
         DbContext context,
         IEnumerable<T> entities,
         BulkInsertOptions options,