Dapper Bulk Operations | Batuhan Aydın

If you are using Dapper, and you need to do bulk operations to reduce I/O time, there isn’t many sources how. In this post, we’ll explore some of our options.

SqlBulkCopy

SQLBulkCopy is a .NET framework class that provides a fast and efficient way to copy large amounts of data from a source data store to a destination SQL Server database table. It is designed to quickly transfer large volumes of data from one data source to a SQL Server table, making it a useful tool for applications that need to load large amounts of data quickly and efficiently.

Pros: Easy, fast Cons: Only for SQL Server, updating is cumbersome

To use this, I usually create this static class, and use its methods in the base Dapper class.


    public static class SqlBulkCopyExtensions<T>
    {
        private static PropertyInfo[] Properties => typeof(T).GetProperties();
        private static string TableName => GetTableName();

        public static DataTable GetDataTable()
        {
            var dataTable = new DataTable
            {
                TableName = $"[dbo].[{TableName}]"
            };
            MapDataTableColumns(dataTable);
            return dataTable;
        }

        public static void FillDataTable(DataTable dataTable, IEnumerable<T> items)
        {
            foreach (var item in items)
            {
                var row = dataTable.NewRow();

                foreach (var property in Properties)
                {
                    if (property.GetCustomAttribute<NotMappedAttribute>() != null)
                        continue;
                    row[property.Name] = property.GetValue(item) ?? DBNull.Value;
                }

                dataTable.Rows.Add(row);
            }
        }

        public static void MapColumns(SqlBulkCopy bulkCopy)
        {
            foreach (var property in Properties)
            {
                if (property.GetCustomAttribute<NotMappedAttribute>() != null)
                    continue;
                SqlBulkCopyColumnMapping mapping = new(property.Name, property.Name);
                bulkCopy.ColumnMappings.Add(mapping);
            }
        }

        private static void MapDataTableColumns(DataTable dataTable)
        {
            foreach (var property in Properties)
            {
                if (property.GetCustomAttribute<NotMappedAttribute>() != null)
                    continue;
                dataTable.Columns.Add(property.Name, Nullable.GetUnderlyingType(property.PropertyType) ?? property.PropertyType);
            }
        }

        // Create an attribute named TableName, which holds the table name we're inserting.
        private static string GetTableName()
        {
            var tableNameAttribute = typeof(T).GetCustomAttribute<TableNameAttribute>(false);
            if (tableNameAttribute is null || string.IsNullOrWhiteSpace(tableNameAttribute.Name)) throw new ArgumentNullException(("Null or empty table name"));
            return tableNameAttribute.Name;
        }
    }

    public virtual void BulkInsert<T>(IEnumerable<T> items)
        {
            var dataTable = SqlBulkCopyExtensions<T>.GetDataTable();
            SqlBulkCopyExtensions<T>.FillDataTable(dataTable, items);
            using (var connection = new SqlConnection(_connectionString))
            {
                using var bulkCopy = new SqlBulkCopy(connection);
                try
                {
                    connection.Open();
                    bulkCopy.DestinationTableName = dataTable.TableName;
                    bulkCopy.BulkCopyTimeout = 300;
                    SqlBulkCopyExtensions<T>.MapColumns(bulkCopy);
                    bulkCopy.WriteToServerAsync(dataTable);
                }
                catch
                {
                    return;
                }
                finally
                {
                    bulkCopy.Close();
                    if (connection.State == ConnectionState.Open)
                    {
                        connection.Close();
                    }
                }
            }
        }

The static class is not customizable. In that case, I’d create an abstract class, an interface and implement it for every type that needs to use this. And accept that interface in the base dapper class.

For bulk update, you may create a temporary table, make the insertion to there, then merge the data. However, this is cumbersome. There is a much better alternative for bulk update operations.

User Defined Table Type

Pros: You can do different type of operations Cons: You create many database types

Create User Defined Table type in the database. This will hold the data you’re going to update and the keys to find them.


CREATE TYPE [RandomTable.UDT] AS TABLE
(
    Id bigint NOT NULL,
    Name VARCHAR(50),
    Primary Key(Id)
);

Create a stored procedure that accepts the udt we created, and performs the update operation

CREATE PROCEDURE UpdateData
    @Recent [RandomTable.UDT] READONLY
AS
BEGIN
    UPDATE RandomTable
    SET t.Name = r.Name
    FROM RandomTable t
    INNER JOIN @Recent r ON t.Id = r.Id
END

Call it from Dapper.

I’ve added two methods here. The first one handles everything itself but in some cases you need to do more, something different. The second one is more open to customization, you can create and fill DataTable as you wish

        public virtual int BulkUpdate<T>(IEnumerable<T> items, string spName)
        {
            var dataTable = SqlBulkCopyExtensions<T>.GetDataTable();
            SqlBulkCopyExtensions<T>.FillDataTable(dataTable, items);
            using (var connection = new SqlConnection(_connectionString))
            {
                try
                {
                    connection.Open();
                    var result = connection.Execute($"[dbo].[{spName}]", new
                    {
                        RecentData = dataTable.AsTableValuedParameter()
                    }, commandType: CommandType.StoredProcedure);
                    return result;
                }
                catch
                {
                    return -1;
                }
                finally
                {
                    if (connection.State == ConnectionState.Open)
                    {
                        connection.Close();
                    }
                }
            }
        }

        public virtual int BulkUpdate(DataTable dataTable, string spName)
        {
            using (var connection = new SqlConnection(_connectionString))
            {
                try
                {
                    connection.Open();
                    var result = connection.Execute($"[dbo].[{spName}]", new
                    {
                        RecentData = dataTable.AsTableValuedParameter()
                    }, commandType: CommandType.StoredProcedure);
                    return result;
                }
                catch
                {
                    return -1;
                }
                finally
                {
                    if (connection.State == ConnectionState.Open)
                    {
                        connection.Close();
                    }
                }
            }
        }