Add support for conflict resolution in bulk insert operations with PG

Author: fabien.menager

src/EntityFrameworkCore.ExecuteInsert.PostgreSql/PostgreSqlBulkInsertProvider.cs

@@ -1,6 +1,11 @@
 using System.Data.Common;
+using System.Linq.Expressions;
+using System.Text;
+
+using EntityFrameworkCore.ExecuteInsert.OnConflict;
 
 using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata;
 
 using Npgsql;
 
@@ -15,7 +20,7 @@ public class PostgreSqlBulkInsertProvider : BulkInsertProviderBase
     protected override string CreateTableCopySql => "CREATE TEMPORARY TABLE {0} AS TABLE {1} WITH NO DATA;";
 
     //language=sql
-    protected override string AddTableCopyBulkInsertId => "ALTER TABLE {0} ADD COLUMN _bulk_insert_id SERIAL PRIMARY KEY;";
+    protected override string AddTableCopyBulkInsertId => $"ALTER TABLE {{0}} ADD COLUMN {BulkInsertId} SERIAL PRIMARY KEY;";
 
     private string GetBinaryImportCommand(DbContext context, Type entityType, string tableName)
     {
@@ -24,6 +29,204 @@ private string GetBinaryImportCommand(DbContext context, Type entityType, string
         return $"COPY {tableName} ({string.Join(", ", columns)}) FROM STDIN (FORMAT BINARY)";
     }
 
+    protected override string BuildInsertSelectQuery<T>(string tableName,
+        string targetTableName,
+        IProperty[] insertedProperties,
+        IProperty[] properties,
+        BulkInsertOptions options, OnConflictOptions? onConflict = null)
+    {
+        var insertedColumns = insertedProperties.Select(p => Escape(p.GetColumnName()));
+        var insertedColumnList = string.Join(", ", insertedColumns);
+
+        var returnedColumns = properties.Select(p => $"{Escape(p.GetColumnName())} AS {Escape(p.Name)}");
+        var columnList = string.Join(", ", returnedColumns);
+
+        var q = new StringBuilder();
+
+        if (options.MoveRows)
+        {
+            q.AppendLine($"""
+                    WITH moved_rows AS (
+                       DELETE FROM {tableName}
+                           RETURNING {insertedColumnList}
+                    )
+                    """);
+            tableName = "moved_rows";
+        }
+
+        q.AppendLine($"""
+                      INSERT INTO {targetTableName} ({insertedColumnList})
+                      SELECT {insertedColumnList}
+                      FROM {tableName}
+                      """);
+
+        if (onConflict is OnConflictOptions<T> onConflictTyped)
+        {
+            q.AppendLine("ON CONFLICT");
+
+            if (onConflictTyped.Update != null)
+            {
+                if (onConflictTyped.Match != null)
+                {
+                    q.AppendLine($"({string.Join(", ", GetColumns(onConflictTyped.Match).Select(Escape))})");
+                }
+
+                if (onConflictTyped.Update != null)
+                {
+                    q.AppendLine($"DO UPDATE SET {string.Join(", ", GetUpdates(onConflictTyped.Update))}");
+                }
+
+                if (onConflictTyped.Condition != null)
+                {
+                    q.AppendLine($"WHERE {onConflictTyped.Condition}");
+                }
+            }
+            else
+            {
+                q.AppendLine("DO NOTHING");
+            }
+        }
+
+        if (columnList.Length != 0)
+        {
+            q.AppendLine($"RETURNING {columnList}");
+        }
+
+        q.AppendLine(";");
+
+        return q.ToString();
+    }
+
+    private IEnumerable<string> GetUpdates<T>(Expression<Func<T, object>> update)
+    {
+        if (update.Body is NewExpression { Members: not null } newExpr)
+        {
+            foreach (var arg in newExpr.Arguments.Zip(newExpr.Members, (expr, member) => (expr, member)))
+            {
+                yield return $"{Escape(arg.member.Name)} = {ToSqlExpression(arg.expr)}";
+            }
+        }
+        else if (update.Body is MemberInitExpression memberInit)
+        {
+            foreach (var binding in memberInit.Bindings.OfType<MemberAssignment>())
+            {
+                yield return $"{Escape(binding.Member.Name)} = {ToSqlExpression(binding.Expression)}";
+            }
+        }
+        else if (update.Body is MemberExpression memberExpr)
+        {
+            yield return $"{Escape(memberExpr.Member.Name)} = {ToSqlExpression(memberExpr)}";
+        }
+        else
+        {
+            throw new NotSupportedException("Unsupported expression type for update");
+        }
+    }
+
+    private string ToSqlExpression(Expression expr)
+    {
+        switch (expr)
+        {
+            case MemberExpression m:
+                var prefix = "EXCLUDED";
+                return $"{prefix}.{Escape(m.Member.Name)}";
+
+            case BinaryExpression b:
+                var left = ToSqlExpression(b.Left);
+                var right = ToSqlExpression(b.Right);
+                var op = b.NodeType switch
+                {
+                    ExpressionType.Add => b.Type == typeof(string) ? "||" : "+",
+                    ExpressionType.Subtract => "-",
+                    ExpressionType.Multiply => "*",
+                    ExpressionType.Divide => "/",
+                    ExpressionType.Modulo => "%",
+                    ExpressionType.AndAlso => "AND",
+                    ExpressionType.OrElse => "OR",
+                    ExpressionType.Equal => "=",
+                    ExpressionType.NotEqual => "<>",
+                    ExpressionType.LessThan => "<",
+                    ExpressionType.LessThanOrEqual => "<=",
+                    ExpressionType.GreaterThan => ">",
+                    ExpressionType.GreaterThanOrEqual => ">=",
+                    _ => throw new NotSupportedException($"Opérateur non supporté: {b.NodeType}")
+                };
+                return $"({left} {op} {right})";
+
+            case ConstantExpression c:
+                if (c.Type == typeof(RawSqlValue) && c.Value != null)
+                {
+                    return ((RawSqlValue)c.Value!).Sql;
+                }
+
+                if (c.Type == typeof(string) ||
+                    c.Type == typeof(Guid))
+                {
+                    return $"'{c.Value}'";
+                }
+
+                if (c.Type == typeof(bool))
+                {
+                    return (bool)c.Value! ? "TRUE" : "FALSE";
+                }
+
+                return c.Value?.ToString() ?? "NULL";
+
+            case UnaryExpression u:
+                if (u.NodeType == ExpressionType.Convert)
+                {
+                    return ToSqlExpression(u.Operand);
+                }
+                if (u.NodeType == ExpressionType.Not)
+                {
+                    return $"NOT ({ToSqlExpression(u.Operand)})";
+                }
+                throw new NotSupportedException($"Unary operator not supported: {u.NodeType}");
+
+            case MethodCallExpression mce:
+                // Supporte quelques méthodes courantes (ToLower, ToUpper, Trim, etc.)
+                var objSql = mce.Object != null ? ToSqlExpression(mce.Object) : null;
+                var argsSql = mce.Arguments.Select(ToSqlExpression).ToArray();
+                switch (mce.Method.Name)
+                {
+                    case "ToLower":
+                        return $"LOWER({objSql})";
+                    case "ToUpper":
+                        return $"UPPER({objSql})";
+                    case "Trim":
+                        return $"BTRIM({objSql})";
+                    case "Contains" when mce is { Object: not null, Arguments.Count: 1 }:
+                        return $"{objSql} LIKE '%' || {argsSql[0]} || '%'";
+                    case "StartsWith" when mce is { Object: not null, Arguments.Count: 1 }:
+                        return $"{objSql} LIKE {argsSql[0]} || '%'";
+                    case "EndsWith" when mce is { Object: not null, Arguments.Count: 1 }:
+                        return $"{objSql} LIKE '%' || {argsSql[0]}";
+                    default:
+                        throw new NotSupportedException($"Method not supported: {mce.Method.Name}");
+                }
+
+            case ParameterExpression p:
+                return Escape(p.Name ?? "param");
+
+            default:
+                throw new NotSupportedException($"Expression not supported: {expr.NodeType}");
+        }
+    }
+
+    private string[] GetColumns<T>(Expression<Func<T, object>> columns)
+    {
+        return columns.Body switch
+        {
+            NewExpression newExpression => newExpression.Arguments.OfType<MemberExpression>()
+                .Select(m => m.Member.Name)
+                .ToArray(),
+            MemberExpression memberExpression => [
+                memberExpression.Member.Name
+            ],
+            _ => throw new NotSupportedException("Unsupported expression type")
+        };
+    }
+
     protected override async Task BulkImport<T>(DbContext context, DbConnection connection, IEnumerable<T> entities,
         string tableName, PropertyAccessor[] properties, CancellationToken ctk) where T : class
     {

src/EntityFrameworkCore.ExecuteInsert.SqlServer/SqlServerBulkInsertProvider.cs

@@ -1,7 +1,9 @@
 using System.Data;
 using System.Data.Common;
+using System.Text;
 
 using EntityFrameworkCore.ExecuteInsert.Helpers;
+using EntityFrameworkCore.ExecuteInsert.OnConflict;
 
 using Microsoft.Data.SqlClient;
 using Microsoft.EntityFrameworkCore;
@@ -18,12 +20,9 @@ public class SqlServerBulkInsertProvider : BulkInsertProviderBase
     protected override string CreateTableCopySql => "SELECT {2} INTO {0} FROM {1} WHERE 1 = 0;";
 
     //language=sql
-    protected override string AddTableCopyBulkInsertId => "ALTER TABLE {0} ADD _bulk_insert_id INT IDENTITY PRIMARY KEY;";
+    protected override string AddTableCopyBulkInsertId => $"ALTER TABLE {{0}} ADD {BulkInsertId} INT IDENTITY PRIMARY KEY;";
 
-    protected override string GetTempTableName<T>(string tableName) where T : class
-    {
-        return $"#_temp_bulk_insert_{tableName}";
-    }
+    protected override string GetTempTableName(string tableName) => $"#_temp_bulk_insert_{tableName}";
 
     protected override async Task BulkImport<T>(DbContext context, DbConnection connection, IEnumerable<T> entities, string tableName,
         PropertyAccessor[] properties, CancellationToken ctk)
@@ -45,35 +44,48 @@ protected override async Task BulkImport<T>(DbContext context, DbConnection conn
         await t.CommitAsync(ctk);
     }
 
-    protected override string BuildInsertSelectQuery(string tempTableName, string targetTableName,
-        IProperty[] insertedProperties, IProperty[] properties, bool moveRows)
+    protected override string BuildInsertSelectQuery<T>(string tableName,
+        string targetTableName,
+        IProperty[] insertedProperties,
+        IProperty[] properties,
+        BulkInsertOptions options, OnConflictOptions? onConflict = null)
     {
-        var insertedColumns = insertedProperties.Select(p => Escape(p.GetColumnName()));
+        var insertedColumns = insertedProperties.Select(p => Escape(p.GetColumnName())).ToArray();
         var insertedColumnList = string.Join(", ", insertedColumns);
-        var columnList = string.Join(", ", properties.Select(p => $"INSERTED.{p.GetColumnName()}"));
 
-//         if (moveRows)
+        var returnedColumns = properties.Select(p => $"INSERTED.{p.GetColumnName()} AS [{p.Name}]");
+        var columnList = string.Join(", ", returnedColumns);
+
+        var q = new StringBuilder();
+
+//         if (options.MoveRows)
 //         {
-//             //language=sql
-//             return $"""
-//                     WITH moved_rows AS (
-//                        DELETE FROM {tempTableName}
-//                            OUTPUT {insertedColumnList}
-//                     )
-//                     INSERT INTO {targetTableName} ({insertedColumnList})
-//                     SELECT {insertedColumnList}
-//                     FROM moved_rows
-//                         RETURNING {columnList};
-//                     """;
+//             var deletedColumnList = string.Join(", ", insertedColumns.Select(c => $"DELETED.{c}"));
+//
+//             q.AppendLine($"""
+//                 DELETE FROM {tableName}
+//                 OUTPUT {deletedColumnList}
+//                 """);
 //         }
 
-        //language=sql
-        return $"""
-                INSERT INTO {targetTableName} ({insertedColumnList})
-                OUTPUT {columnList}
-                SELECT {insertedColumnList}
-                FROM {tempTableName};
-                """;
+        q.AppendLine($"INSERT INTO {targetTableName} ({insertedColumnList})");
+
+        if (columnList.Length != 0)
+        {
+            q.AppendLine($"OUTPUT {columnList}");
+        }
+
+        q.AppendLine($"""
+            SELECT {insertedColumnList}
+            FROM {tableName}
+            """);
+
+        // SQL Server ne supporte pas ON CONFLICT DO NOTHING, mais on garde la signature pour homogénéité
+        // if (options.OnConflictIgnore) { ... }
+
+        q.AppendLine(";");
+
+        return q.ToString();
     }
 
     private DataTable ConvertToDataTable<T>(PropertyAccessor[] properties)

src/EntityFrameworkCore.ExecuteInsert/Abstractions/BulkInsertExtensions.cs

@@ -1,4 +1,6 @@
-using Microsoft.EntityFrameworkCore;
+using EntityFrameworkCore.ExecuteInsert.OnConflict;
+
+using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Infrastructure;
 
 namespace EntityFrameworkCore.ExecuteInsert.Abstractions;
@@ -9,23 +11,24 @@ public static async Task<List<T>> ExecuteInsertWithIdentityAsync<T>(
         this DbSet<T> dbSet,
         IEnumerable<T> entities,
         Action<BulkInsertOptions>? configure = null,
+        OnConflictOptions? onConflict = null,
         CancellationToken ctk = default
     ) where T : class
     {
         var provider = InitProvider(dbSet, configure, out var context, out var options);
 
-        return await provider.BulkInsertWithIdentityAsync(context, entities, options, ctk);
+        return await provider.BulkInsertWithIdentityAsync(context, entities, options, onConflict, ctk);
     }
 
-    public static async Task ExecuteInsertWithIdentityAsync<T>(this DbContext dbContext, IEnumerable<T> entities, Action<BulkInsertOptions>? configure = null, CancellationToken cancellationToken = default) where T : class
+    public static async Task ExecuteInsertWithIdentityAsync<T>(this DbContext dbContext, IEnumerable<T> entities, Action<BulkInsertOptions>? configure = null, 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.ExecuteInsertWithIdentityAsync(entities, configure, cancellationToken);
+        await dbSet.ExecuteInsertWithIdentityAsync(entities, configure, onConflict, cancellationToken);
     }
 
     // public static async Task<List<object>> ExecuteInsertWithPrimaryKeyAsync<T>(
@@ -55,23 +58,24 @@ public static async Task ExecuteInsertAsync<T>(
         this DbSet<T> dbSet,
         IEnumerable<T> entities,
         Action<BulkInsertOptions>? configure = null,
+        OnConflictOptions? onConflict = null,
         CancellationToken ctk = default
     ) where T : class
     {
         var provider = InitProvider(dbSet, configure, out var context, out var options);
 
-        await provider.BulkInsertWithoutReturnAsync(context, entities, options, ctk);
+        await provider.BulkInsertWithoutReturnAsync(context, entities, options, onConflict, ctk);
     }
 
-    public static async Task ExecuteInsertAsync<T>(this DbContext dbContext, IEnumerable<T> entities, Action<BulkInsertOptions>? configure = null, CancellationToken cancellationToken = default) where T : class
+    public static async Task ExecuteInsertAsync<T>(this DbContext dbContext, IEnumerable<T> entities, Action<BulkInsertOptions>? configure = null, 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.ExecuteInsertAsync(entities, configure, cancellationToken);
+        await dbSet.ExecuteInsertAsync(entities, configure, onConflict, cancellationToken);
     }
 
     private static DbContext GetDbContext<T>(this DbSet<T> dbSet) where T : class