public async Task<int> BulkInsertAsync<T>(string tableName, IEnumerable<T> data, int batchSize = 1000)
{
int totalRowsAffected = 0;
var batches = data.Batch(batchSize);
foreach (var batch in batches)
{
string connstring = "connection string";
using SnowflakeDbConnection conn = new SnowflakeDbConnection();
conn.ConnectionString = connstring;
conn.Open();
using var transaction = conn.BeginTransaction();
try
{
var dataTable = batch.ToDataTable();
var sql = GenerateBulkInsertSql(tableName, dataTable);
var rowsAffected = await conn.ExecuteAsync(sql, dataTable, transaction: transaction);
transaction.Commit();
totalRowsAffected += rowsAffected;
}
catch
{
transaction.Rollback();
throw;
}
}
return totalRowsAffected;
}
private string GenerateBulkInsertSql(string tableName, DataTable dataTable)
{
var columns = string.Join(", ", dataTable.Columns.Cast<DataColumn>().Select(c => c.ColumnName));
var values = string.Join(", ", dataTable.Columns.Cast<DataColumn>().Select(c => ":" + c.ColumnName));
return $"INSERT INTO {tableName} ({columns}) VALUES ({values})";
}
public static DataTable ToDataTable<T>(this IEnumerable<T> data)
{
var dataTable = new DataTable();
var properties = typeof(T).GetProperties();
foreach (var prop in properties)
{
dataTable.Columns.Add(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
}
foreach (var item in data)
{
var row = dataTable.NewRow();
foreach (var prop in properties)
{
row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
}
dataTable.Rows.Add(row);
}
return dataTable;
}
Can anyone assist with performing BulkInsert and BulkUpdate operations on Snowflake tables using Dapper and C#? I am encountering issues where the values are not being inserted into the Snowflake tables. Any help would be greatly appreciated. Thanks in advance!
7
Lose the DataTable. It would be nice if this was hand-written SQL, but it probably isn’t a huge problem in this case, but, something along the lines of:
private static string GenerateBulkInsertSql<T>(string tableName)
{
var propertyNames = typeof(T).GetProperties().Select(p => p.Name);
var columns = string.Join(", ", propertyNames);
var values = string.Join(", ", propertyNames.Select(name => ":" + name));
return $"INSERT INTO {tableName} ({columns}) VALUES ({values})";
}
which should be enough to get you the SQL (the same SQL for every operation), then you should be able to use batch as the parameter value:
var sql = GenerateBulkInsertSql<T>(tableName);
foreach (var batch in batches)
{
string connstring = "connection string";
using SnowflakeDbConnection conn = new SnowflakeDbConnection();
conn.ConnectionString = connstring;
conn.Open();
using var transaction = conn.BeginTransaction();
try
{
var rowsAffected = await conn.ExecuteAsync(sql, batch, transaction: transaction);
transaction.Commit();
totalRowsAffected += rowsAffected;
}
catch
{
transaction.Rollback();
throw;
}
}
2