Enhance conflict resolution with support for raw SQL conditions and expressions in OnConflictOptions

Author: fabien.menager

README.md

@@ -43,7 +43,7 @@ Install-Package PhenX.EntityFrameworkCore.BulkInsert.MySql
 
 ## Usage
 
-1. Register the bulk insert provider in your `DbContextOptions`:
+Register the bulk insert provider in your `DbContextOptions`:
 
 ```csharp
 services.AddDbContext<MyDbContext>(options =>
@@ -62,7 +62,7 @@ services.AddDbContext<MyDbContext>(options =>
 });
 ```
 
-2. Use the bulk insert extension method:
+### Very basic usage
 
 ```csharp
 // Asynchronously
@@ -72,7 +72,7 @@ await dbContext.ExecuteBulkInsertAsync(entities);
 dbContext.ExecuteBulkInsert(entities);
 ```
 
-3. You can also configure the bulk insert options:
+### Bulk insert with options
 
 ```csharp
 // Common options
@@ -103,12 +103,46 @@ await dbContext.ExecuteBulkInsertAsync(entities, o =>
 });
 ```
 
-4. You can also return the inserted entities (slower):
+### Returning inserted entities
 
 ```csharp
 await dbContext.ExecuteBulkInsertReturnEntitiesAsync(entities);
 ```
 
+### Conflict resolution / merge / upsert
+
+Conflict resolution works by specifying columns that should be used to detect conflicts and the action to take when
+a conflict is detected (e.g., update existing rows), using the `onConflict` parameter.
+
+ * The conflicting columns are specified with the `Match` property and must have a unique constraint in the database.
+ * The action to take when a conflict is detected is specified with the `Update` property. If not specified, the default action is to do nothing (i.e., skip the conflicting rows).
+ * You can also specify the condition for the update action using either the `Where` or the `RawWhere` property. If not specified, the update action will be applied to all conflicting rows.
+
+```csharp
+await dbContext.ExecuteBulkInsertAsync(entities, onConflict: new OnConflictOptions<TestEntity>
+{
+    Match = e => new
+    {
+        e.Name,
+        // ...other columns to match on
+    },
+
+    // Optional: specify the update action, if not specified, the default action is to do nothing
+    Update = e => new TestEntity
+    {
+        Price = e.Price // Update the Price column with the new value
+    },
+
+    // Optional: specify the condition for the update action
+    // Excluded is the row being inserted which is in conflict, and Inserted is the row already in the database.
+    // Using raw SQL condition
+    RawWhere = "EXCLUDED.some_price > INSERTED.some_price",
+
+    // OR using a lambda expression
+    Where = (inserted, excluded) => excluded.Price > inserted.Price,
+});
+```
+
 ## Roadmap
 
 - [ ] [Add support for navigation properties](https://github.com/PhenX/PhenX.EntityFrameworkCore.BulkInsert/issues/2)

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

@@ -1,5 +1,7 @@
 using System.Text;
 
+using Microsoft.EntityFrameworkCore;
+
 using PhenX.EntityFrameworkCore.BulkInsert.Dialect;
 using PhenX.EntityFrameworkCore.BulkInsert.Metadata;
 using PhenX.EntityFrameworkCore.BulkInsert.Options;
@@ -12,14 +14,27 @@ internal class MySqlServerDialectBuilder : SqlDialectBuilder
 
     protected override string CloseDelimiter => "`";
 
+    /// <summary>
+    /// Indicates whether the dialect supports moving rows from temporary table to the final table, in order to
+    /// theoretically reduce disk space requirements.
+    /// </summary>
     protected override bool SupportsMoveRows => false;
 
+    /// <summary>
+    /// Indicates whether the dialect supports INSERT INTO table AS alias.
+    /// </summary>
+    protected override bool SupportsInsertIntoAlias => false;
+
     public override string CreateTableCopySql(string tempNameName, TableMetadata tableInfo, IReadOnlyList<ColumnMetadata> columns)
     {
         return $"CREATE TEMPORARY TABLE {tempNameName} SELECT * FROM {tableInfo.QuotedTableName} WHERE 1 = 0;";
     }
 
-    protected override void AppendConflictCondition<T>(StringBuilder sql, OnConflictOptions<T> onConflictTyped)
+    protected override void AppendConflictCondition<T>(
+        StringBuilder sql,
+        TableMetadata target,
+        DbContext context,
+        OnConflictOptions<T> onConflictTyped)
     {
         throw new NotSupportedException("Conflict conditions are not supported in MYSQL");
     }

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

@@ -1,5 +1,7 @@
 using System.Text;
 
+using Microsoft.EntityFrameworkCore;
+
 using PhenX.EntityFrameworkCore.BulkInsert.Dialect;
 using PhenX.EntityFrameworkCore.BulkInsert.Metadata;
 using PhenX.EntityFrameworkCore.BulkInsert.Options;
@@ -35,6 +37,7 @@ public override string CreateTableCopySql(string tempTableName, TableMetadata ta
     }
 
     public override string BuildMoveDataSql<T>(
+        DbContext context,
         TableMetadata target,
         string source,
         IReadOnlyList<ColumnMetadata> insertedColumns,
@@ -66,24 +69,31 @@ public override string BuildMoveDataSql<T>(
                 throw new InvalidOperationException("Table has no primary key that can be used for conflict detection.");
             }
 
-            q.AppendLine($"MERGE INTO {target.QuotedTableName} AS TARGET");
+            q.AppendLine($"MERGE INTO {target.QuotedTableName} AS INSERTED");
 
             q.Append("USING (SELECT ");
             q.AppendColumns(insertedColumns);
-            q.Append($" FROM {source}) AS SOURCE (");
+            q.Append($" FROM {source}) AS EXCLUDED (");
             q.AppendColumns(insertedColumns);
             q.AppendLine(")");
 
             q.Append("ON ");
-            q.AppendJoin(" AND ", matchColumns, (b, col) => b.Append($"TARGET.{col} = SOURCE.{col}"));
+            q.AppendJoin(" AND ", matchColumns, (b, col) => b.Append($"INSERTED.{col} = EXCLUDED.{col}"));
             q.AppendLine();
 
             if (onConflictTyped.Update != null)
             {
                 var columns = target.GetColumns(false);
 
-                q.AppendLine("WHEN MATCHED THEN UPDATE SET ");
-                q.AppendJoin(", ", GetUpdates(target, columns, onConflictTyped.Update));
+                q.AppendLine("WHEN MATCHED ");
+
+                if (!string.IsNullOrEmpty(onConflictTyped.RawWhere))
+                {
+                    q.Append($"AND {onConflictTyped.RawWhere} ");
+                }
+
+                q.AppendLine("THEN UPDATE SET ");
+                q.AppendJoin(", ", GetUpdates(context, target, columns, onConflictTyped.Update));
                 q.AppendLine();
             }
 
@@ -92,7 +102,7 @@ public override string BuildMoveDataSql<T>(
             q.AppendLine(")");
 
             q.Append("VALUES (");
-            q.AppendJoin(", ", insertedColumns, (b, col) => b.Append($"SOURCE.{col.QuotedColumName}"));
+            q.AppendJoin(", ", insertedColumns, (b, col) => b.Append($"EXCLUDED.{col.QuotedColumName}"));
             q.AppendLine(")");
 
             if (returnedColumns.Count != 0)
@@ -134,9 +144,4 @@ public override string BuildMoveDataSql<T>(
         var result = q.ToString();
         return result;
     }
-
-    protected override string GetExcludedColumnName(string columnName)
-    {
-        return $"SOURCE.{columnName}";
-    }
 }

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

@@ -15,4 +15,9 @@ public override string CreateTableCopySql(string tempNameName, TableMetadata tab
     {
         return $"CREATE TEMP TABLE {tempNameName} AS SELECT * FROM {tableInfo.QuotedTableName} WHERE 0;";
     }
+
+    protected override string Trim(string lhs)
+    {
+        return $"TRIM({lhs})";
+    }
 }

src/PhenX.EntityFrameworkCore.BulkInsert/BulkInsertProviderBase.cs

@@ -1,5 +1,4 @@
 using System.Runtime.CompilerServices;
-using System.Runtime.InteropServices;
 
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Storage;
@@ -202,6 +201,7 @@ protected virtual async Task AddBulkInsertIdColumn<T>(
     {
         var query =
             SqlDialect.BuildMoveDataSql<T>(
+                context,
                 tableInfo,
                 tempTableName,
                 tableInfo.GetColumns(options.CopyGeneratedColumns),