SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Fix Comments and Casting logic of NVARCHAR2

Browse source
This commit is contained in:
MohitGuptaC 2026-06-24 10:24:05 +05:30
parent 82e069c265
commit 6b7296fa29
No known key found for this signature in database
1 changed files with 111 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

198
src/modules/Elsa.Persistence.EFCore.Common/Extensions/BulkUpsertExtensions.cs
View file

@ -16,6 +16,12 @@ public static class BulkUpsertExtensions
/// <summary>
/// Performs a bulk upsert operation on a list of entities in the specified database context using a key selector.
/// </summary>
/// <typeparam name="TDbContext">The type of the database context.</typeparam>
/// <typeparam name="TEntity">The type of the entity being upserted.</typeparam>
/// <param name="dbContext">The database context where the bulk upsert operation will be executed.</param>
/// <param name="entities">The list of entities to be upserted.</param>
/// <param name="keySelector">An expression used to determine the key for upsert operations.</param>
/// <param name="cancellationToken">A token to observe while waiting for the operation to complete.</param>
public static async Task BulkUpsertAsync<TDbContext, TEntity>(
this TDbContext dbContext,
IList<TEntity> entities,
@ -30,6 +36,13 @@ public static class BulkUpsertExtensions
/// <summary>
/// Performs a bulk upsert operation on a list of entities in the specified database context using a key selector and optional batch size.
/// </summary>
/// <typeparam name="TDbContext">The type of the database context.</typeparam>
/// <typeparam name="TEntity">The type of the entity being upserted.</typeparam>
/// <param name="dbContext">The database context where the bulk upsert operation will be executed.</param>
/// <param name="entities">The list of entities to be upserted.</param>
/// <param name="keySelector">An expression used to determine the key for upsert operations.</param>
/// <param name="batchSize">The size of each batch for processing the upsert operation. Defaults to 50.</param>
/// <param name="cancellationToken">A token to observe while waiting for the operation to complete.</param>
/// <exception cref="NotSupportedException">Thrown if the database provider for the context is not supported.</exception>
public static async Task BulkUpsertAsync<TDbContext, TEntity>(
this TDbContext dbContext,
@ -43,29 +56,30 @@ public static class BulkUpsertExtensions
if (entities.Count == 0)
return;
// Identify the current provider (e.g., "Microsoft.EntityFrameworkCore.SqlServer")
var providerName = dbContext.Database.ProviderName?.ToLowerInvariant() ?? string.Empty;
// Determine the method for generating SQL based on the provider
Func<DbContext, IList<TEntity>, Expression<Func<TEntity, string>>, (string, object[])> generateSql = providerName switch
{
var pn when pn.Contains("sqlserver") => GenerateSqlServerUpsert,
var pn when pn.Contains("sqlite") => GenerateSqliteUpsert,
var pn when pn.Contains("postgres") => GeneratePostgresUpsert,
var pn when pn.Contains("mysql") => GenerateMySqlUpsert,
var pn when pn.Contains("oracle") => GenerateOracleUpsert,
var pn when pn.Contains("sqlite") => GenerateSqliteUpsert,
var pn when pn.Contains("postgres") => GeneratePostgresUpsert,
var pn when pn.Contains("mysql") => GenerateMySqlUpsert,
var pn when pn.Contains("oracle") => GenerateOracleUpsert,
_ => throw new NotSupportedException($"Provider '{providerName}' is not supported.")
};
// Loop through batched entities
foreach (var batch in entities.Chunk(batchSize))
{
// Generate SQL and parameters
var (sql, parameters) = generateSql(dbContext, batch, keySelector);
await dbContext.Database.ExecuteSqlRawAsync(sql, parameters, cancellationToken);
}
}
// -------------------------------------------------------------------------
// SQL Server
// -------------------------------------------------------------------------
private static (string, object[]) GenerateSqlServerUpsert<TEntity>(
DbContext dbContext,
IList<TEntity> entities,
@ -78,7 +92,9 @@ public static class BulkUpsertExtensions
var props = entityType.GetProperties().ToList();
var keyProp = entityType.FindProperty(keySelector.GetMemberAccess().Name)!;
var keyColumnName = $"[{keyProp.GetColumnName(storeObject)}]";
var columnNames = props.Select(p => $"[{p.GetColumnName(storeObject)}]").ToList();
var columnNames = props
.Select(p => $"[{p.GetColumnName(storeObject)}]")
.ToList();
var mergeSql = new StringBuilder();
mergeSql.AppendLine($"MERGE {tableName} AS Target");
@ -95,21 +111,27 @@ public static class BulkUpsertExtensions
foreach (var property in props)
{
var paramName = $"{{{parameterCount++}}}";
// If it's a shadow property, retrieve value via Entry(..).Property(..)
var value = property.IsShadowProperty()
? dbContext.Entry(entity).Property(property.Name).CurrentValue
: property.PropertyInfo?.GetValue(entity);
var converter = property.GetTypeMapping().Converter;
if (converter != null) value = converter.ConvertToProvider(value)!;
var converter = property.GetTypeMapping().Converter;
if (converter != null)
value = converter.ConvertToProvider(value)!;
// Explicitly cast null values for varbinary columns
if (property.GetColumnType().StartsWith("varbinary", StringComparison.OrdinalIgnoreCase) && value is null)
values.Add("CAST(NULL AS varbinary(max))");
values.Add("CAST(NULL AS varbinary(max))"); // Explicitly cast null
else
values.Add(paramName);
parameters.Add(value!);
}
mergeSql.AppendLine($"({string.Join(", ", values)}){(i < entities.Count - 1 ? "," : string.Empty)}");
var line = $"({string.Join(", ", values)}){(i < entities.Count - 1 ? "," : string.Empty)}";
mergeSql.AppendLine(line);
}
mergeSql.AppendLine($") AS Source ({string.Join(", ", columnNames)})");
@ -123,10 +145,6 @@ public static class BulkUpsertExtensions
return (mergeSql.ToString(), parameters.ToArray());
}
// -------------------------------------------------------------------------
// SQLite
// -------------------------------------------------------------------------
private static (string, object[]) GenerateSqliteUpsert<TEntity>(
DbContext dbContext,
IList<TEntity> entities,
@ -139,7 +157,9 @@ public static class BulkUpsertExtensions
var props = entityType.GetProperties().ToList();
var keyProp = entityType.FindProperty(keySelector.GetMemberAccess().Name)!;
var keyColumnName = keyProp.GetColumnName(storeObject);
var columnNames = props.Select(p => p.GetColumnName(storeObject)!).ToList();
var columnNames = props
.Select(p => p.GetColumnName(storeObject)!)
.ToList();
var sb = new StringBuilder();
var parameters = new List<object>();
@ -155,30 +175,36 @@ public static class BulkUpsertExtensions
foreach (var property in props)
{
var paramName = $"{{{parameterCount++}}}";
var value = property.IsShadowProperty()
? dbContext.Entry(entity).Property(property.Name).CurrentValue
: property.PropertyInfo?.GetValue(entity);
var converter = property.GetTypeMapping().Converter;
if (converter != null) value = converter.ConvertToProvider(value);
if (converter != null)
value = converter.ConvertToProvider(value);
placeholders.Add(paramName);
parameters.Add(value!);
}
sb.Append($"({string.Join(", ", placeholders)})");
if (i < entities.Count - 1) sb.Append(", ");
if (i < entities.Count - 1)
sb.Append(", ");
}
sb.AppendLine();
sb.AppendLine($"ON CONFLICT(\"{keyColumnName}\") DO UPDATE SET");
sb.AppendLine(string.Join(", ", columnNames.Where(c => c != keyColumnName).Select(c => $"\"{c}\"=excluded.\"{c}\"")) + ";");
var updateAssignments = columnNames
.Where(c => c != keyColumnName)
.Select(c => $"\"{c}\"=excluded.\"{c}\"");
sb.AppendLine(string.Join(", ", updateAssignments) + ";");
return (sb.ToString(), parameters.ToArray());
}
// -------------------------------------------------------------------------
// PostgreSQL
// -------------------------------------------------------------------------
private static (string, object[]) GeneratePostgresUpsert<TEntity>(
DbContext dbContext,
IList<TEntity> entities,
@ -188,10 +214,14 @@ public static class BulkUpsertExtensions
var entityType = dbContext.Model.FindEntityType(typeof(TEntity))!;
var tableName = entityType.GetTableName();
var storeObject = StoreObjectIdentifier.Table(tableName!, entityType.GetSchema());
var props = entityType.GetProperties().ToList();
var keyProp = entityType.FindProperty(keySelector.GetMemberAccess().Name)!;
var keyColumnName = keyProp.GetColumnName(storeObject);
var columnNames = props.Select(p => p.GetColumnName(storeObject)!).ToList();
var columnNames = props
.Select(p => p.GetColumnName(storeObject)!)
.ToList();
var sb = new StringBuilder();
var parameters = new List<object>();
@ -207,12 +237,16 @@ public static class BulkUpsertExtensions
foreach (var property in props)
{
var paramName = $"{{{parameterCount++}}}";
var value = property.IsShadowProperty()
? dbContext.Entry(entity).Property(property.Name).CurrentValue
: property.PropertyInfo?.GetValue(entity);
var converter = property.GetTypeMapping().Converter;
if (converter != null) value = converter.ConvertToProvider(value);
var converter = property.GetTypeMapping().Converter;
if (converter != null)
value = converter.ConvertToProvider(value);
// Detect json/jsonb column types and cast the parameter so PostgreSQL accepts it.
var columnType = property.GetColumnType();
if (columnType.StartsWith("jsonb", StringComparison.OrdinalIgnoreCase))
placeholders.Add($"CAST({paramName} AS jsonb)");
@ -220,25 +254,27 @@ public static class BulkUpsertExtensions
placeholders.Add($"CAST({paramName} AS json)");
else
placeholders.Add(paramName);
parameters.Add(value!);
}
sb.Append($"({string.Join(", ", placeholders)})");
if (i < entities.Count - 1) sb.Append(", ");
if (i < entities.Count - 1)
sb.Append(", ");
}
sb.AppendLine();
sb.AppendLine($"ON CONFLICT (\"{keyColumnName}\") DO UPDATE SET");
sb.AppendLine(string.Join(", ", columnNames.Where(c => c != keyColumnName).Select(c => $"\"{c}\" = EXCLUDED.\"{c}\"")) + ";");
var updateAssignments = columnNames
.Where(c => c != keyColumnName)
.Select(c => $"\"{c}\" = EXCLUDED.\"{c}\"");
sb.AppendLine(string.Join(", ", updateAssignments) + ";");
return (sb.ToString(), parameters.ToArray());
}
// -------------------------------------------------------------------------
// MySQL
// -------------------------------------------------------------------------
private static (string, object[]) GenerateMySqlUpsert<TEntity>(
DbContext dbContext,
IList<TEntity> entities,
@ -248,10 +284,14 @@ public static class BulkUpsertExtensions
var entityType = dbContext.Model.FindEntityType(typeof(TEntity))!;
var tableName = entityType.GetTableName();
var storeObject = StoreObjectIdentifier.Table(tableName!, entityType.GetSchema());
var props = entityType.GetProperties().ToList();
var keyProp = entityType.FindProperty(keySelector.GetMemberAccess().Name)!;
var keyColumnName = keyProp.GetColumnName(storeObject);
var columnNames = props.Select(p => p.GetColumnName(storeObject)!).ToList();
var columnNames = props
.Select(p => p.GetColumnName(storeObject)!)
.ToList();
var sb = new StringBuilder();
var parameters = new List<object>();
@ -267,30 +307,36 @@ public static class BulkUpsertExtensions
foreach (var property in props)
{
var paramName = $"{{{parameterCount++}}}";
var value = property.IsShadowProperty()
? dbContext.Entry(entity).Property(property.Name).CurrentValue
: property.PropertyInfo?.GetValue(entity);
var converter = property.GetTypeMapping().Converter;
if (converter != null) value = converter.ConvertToProvider(value);
if (converter != null)
value = converter.ConvertToProvider(value);
placeholders.Add(paramName);
parameters.Add(value!);
}
sb.Append($"({string.Join(", ", placeholders)})");
if (i < entities.Count - 1) sb.Append(", ");
if (i < entities.Count - 1)
sb.Append(", ");
}
sb.AppendLine();
sb.AppendLine("ON DUPLICATE KEY UPDATE");
sb.AppendLine(string.Join(", ", columnNames.Where(c => c != keyColumnName).Select(c => $"`{c}` = VALUES(`{c}`)")) + ";");
var updateAssignments = columnNames
.Where(c => c != keyColumnName)
.Select(c => $"`{c}` = VALUES(`{c}`)");
sb.AppendLine(string.Join(", ", updateAssignments) + ";");
return (sb.ToString(), parameters.ToArray());
}
// -------------------------------------------------------------------------
// Oracle
// -------------------------------------------------------------------------
private static (string, object[]) GenerateOracleUpsert<TEntity>(
DbContext dbContext,
IList<TEntity> entities,
@ -301,18 +347,19 @@ public static class BulkUpsertExtensions
var schema = entityType.GetSchema();
var tableName = entityType.GetTableName()!;
var storeObject = StoreObjectIdentifier.Table(tableName, schema);
// Both schema and table must be quoted so Oracle treats them as
// case-sensitive identifiers, matching what EF Core migrations create.
// Both schema and table must be quoted so Oracle treats them as case-sensitive
// identifiers, matching what EF Core migrations create.
var fullName = !string.IsNullOrEmpty(schema)
? $"\"{schema}\".\"{tableName}\""
: $"\"{tableName}\"";
var props = entityType.GetProperties().ToList();
var keyProp = entityType.FindProperty(keySelector.GetMemberAccess().Name)!;
var keyColumnName = keyProp.GetColumnName(storeObject)!;
// Pre-build quoted column name list once; reuse throughout.
// Pre-build quoted column names once and reuse throughout all clauses.
var quotedColumnNames = props
.Select(p => $"\"{p.GetColumnName(storeObject)}\"")
.ToList();
@ -343,30 +390,23 @@ public static class BulkUpsertExtensions
value = converter.ConvertToProvider(value);
parameters.Add(value!);
// Alias must be quoted so Oracle preserves case, matching the
// quoted references in the WHEN MATCHED / WHEN NOT MATCHED clauses.
// Aliases must be quoted so Oracle preserves their case, matching
// the quoted references in ON, UPDATE SET, and INSERT/VALUES below.
var quotedAlias = $"\"{property.GetColumnName(storeObject)}\"";
// Oracle cannot infer the bind parameter type from a bare SELECT …
// FROM DUAL — there is no target column to derive it from. For
// NVARCHAR2 columns this causes ODP.NET to default to VARCHAR2,
// which leads to datatype mismatch errors in the MERGE. An explicit
// CAST restores the correct type. The length is read from the EF
// column type string (e.g. "NVARCHAR2(500)") so it matches the
// actual column definition rather than an arbitrary hardcoded value.
string expr;
// In a SELECT … FROM DUAL subquery, ODP.NET has no target column to
// derive bind parameter types from and defaults to VARCHAR2 for .NET
// strings. Elsa's Oracle migrations define string columns as NVARCHAR2,
// so an explicit CAST is required to avoid a datatype mismatch error.
// The full EF Core column type string (e.g. "NVARCHAR2(450 CHAR)") is
// used directly in the CAST so all Oracle type variants are handled
// correctly without any string parsing.
var columnType = property.GetColumnType() ?? string.Empty;
if (columnType.StartsWith("NVARCHAR2", StringComparison.OrdinalIgnoreCase))
{
var length = ParseNVarchar2Length(columnType);
expr = $"CAST({paramName} AS NVARCHAR2({length}))";
}
else
{
expr = paramName;
}
var expr = columnType.StartsWith("NVARCHAR2", StringComparison.OrdinalIgnoreCase)
? $"CAST({paramName} AS {columnType})"
: paramName;
lineParts.Add($"{expr} AS {quotedAlias}");
}
@ -385,20 +425,4 @@ public static class BulkUpsertExtensions
return (sb.ToString(), parameters.ToArray());
}
/// <summary>
/// Extracts the maximum length from an Oracle NVARCHAR2 column type string.
/// For example, "NVARCHAR2(500)" returns 500.
/// Falls back to 2000 (Oracle's maximum for NVARCHAR2) if the string is malformed.
/// </summary>
private static int ParseNVarchar2Length(string columnType)
{
var open = columnType.IndexOf('(');
var close = columnType.IndexOf(')');
if (open >= 0 && close > open &&
int.TryParse(columnType.AsSpan(open + 1, close - open - 1), out var length))
return length;
return 2000;
}
}
}