feat: add SqlInsertBatchWriter

Author: BrettJaner

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Dependencies/SinkDependenciesFactory.cs

@@ -52,9 +52,13 @@ internal static SinkDependencies Create(
                     sqlCreateDatabaseWriter, sqlConnectionFactoryNoDb),
                 SqlTableCreator = new SqlTableCreator(
                     sqlCreateTableWriter, sqlConnectionFactory),
-                SqlBulkBatchWriter = new SqlBulkBatchWriter(
-                    sinkOptions.TableName, sinkOptions.SchemaName, columnOptions.DisableTriggers,
-                    sqlConnectionFactory, logEventDataGenerator),
+                SqlBulkBatchWriter = sinkOptions.UseSqlBulkCopy
+                    ? (ISqlBulkBatchWriter)new SqlBulkBatchWriter(
+                        sinkOptions.TableName, sinkOptions.SchemaName, columnOptions.DisableTriggers,
+                        sqlConnectionFactory, logEventDataGenerator)
+                    : (ISqlBulkBatchWriter)new SqlInsertBatchWriter(
+                        sinkOptions.TableName, sinkOptions.SchemaName,
+                        sqlConnectionFactory, logEventDataGenerator),
                 SqlLogEventWriter = new SqlLogEventWriter(
                     sinkOptions.TableName, sinkOptions.SchemaName,
                     sqlConnectionFactory, logEventDataGenerator)

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/MSSqlServerSinkOptions.cs

@@ -17,6 +17,7 @@ public MSSqlServerSinkOptions()
             BatchPostingLimit = MSSqlServerSink.DefaultBatchPostingLimit;
             BatchPeriod = MSSqlServerSink.DefaultPeriod;
             EagerlyEmitFirstEvent = true;
+            UseSqlBulkCopy = true;
         }
 
         internal MSSqlServerSinkOptions(
@@ -77,5 +78,10 @@ internal MSSqlServerSinkOptions(
         /// A switch allowing the pass-through minimum level to be changed at runtime
         /// </summary>
         public LoggingLevelSwitch LevelSwitch { get; set; }
+
+        /// <summary>
+        /// Flag to use <see cref="Microsoft.Data.SqlClient.SqlBulkCopy"/> instead of individual INSERT statements (default: true)
+        /// </summary>
+        public bool UseSqlBulkCopy { get; set; }
     }
 }

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Platform/SqlClient/ISqlCommandWrapper.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Data;
+using System.Threading.Tasks;
 
 namespace Serilog.Sinks.MSSqlServer.Platform.SqlClient
 {
@@ -10,5 +11,6 @@ internal interface ISqlCommandWrapper : IDisposable
 
         void AddParameter(string parameterName, object value);
         int ExecuteNonQuery();
+        Task<int> ExecuteNonQueryAsync();
     }
 }

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Platform/SqlClient/SqlCommandWrapper.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Data;
+using System.Threading.Tasks;
 using Microsoft.Data.SqlClient;
 
 namespace Serilog.Sinks.MSSqlServer.Platform.SqlClient
@@ -43,6 +44,9 @@ public void AddParameter(string parameterName, object value)
         public int ExecuteNonQuery() =>
             _sqlCommand.ExecuteNonQuery();
 
+        public Task<int> ExecuteNonQueryAsync() =>
+            _sqlCommand.ExecuteNonQueryAsync();
+
         protected virtual void Dispose(bool disposing)
         {
             if (!_disposedValue)

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Platform/SqlInsertBatchWriter.cs

@@ -0,0 +1,86 @@
+using System;
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Serilog.Debugging;
+using Serilog.Events;
+using Serilog.Sinks.MSSqlServer.Output;
+using static System.FormattableString;
+
+namespace Serilog.Sinks.MSSqlServer.Platform
+{
+    internal class SqlInsertBatchWriter : ISqlBulkBatchWriter
+    {
+        private readonly string _tableName;
+        private readonly string _schemaName;
+        private readonly ISqlConnectionFactory _sqlConnectionFactory;
+        private readonly ILogEventDataGenerator _logEventDataGenerator;
+
+        public SqlInsertBatchWriter(
+            string tableName,
+            string schemaName,
+            ISqlConnectionFactory sqlConnectionFactory,
+            ILogEventDataGenerator logEventDataGenerator)
+        {
+            _tableName = tableName ?? throw new ArgumentNullException(nameof(tableName));
+            _schemaName = schemaName ?? throw new ArgumentNullException(nameof(schemaName));
+            _sqlConnectionFactory = sqlConnectionFactory ?? throw new ArgumentNullException(nameof(sqlConnectionFactory));
+            _logEventDataGenerator = logEventDataGenerator ?? throw new ArgumentNullException(nameof(logEventDataGenerator));
+        }
+
+        public async Task WriteBatch(IEnumerable<LogEvent> events, DataTable dataTable)
+        {
+            try
+            {
+                using (var cn = _sqlConnectionFactory.Create())
+                {
+                    await cn.OpenAsync().ConfigureAwait(false);
+
+                    foreach (var logEvent in events)
+                    {
+                        using (var command = cn.CreateCommand())
+                        {
+                            command.CommandType = CommandType.Text;
+
+                            var fieldList = new StringBuilder(Invariant($"INSERT INTO [{_schemaName}].[{_tableName}] ("));
+                            var parameterList = new StringBuilder(") VALUES (");
+
+                            var index = 0;
+                            foreach (var field in _logEventDataGenerator.GetColumnsAndValues(logEvent))
+                            {
+                                if (index != 0)
+                                {
+                                    fieldList.Append(',');
+                                    parameterList.Append(',');
+                                }
+
+                                fieldList.Append(Invariant($"[{field.Key}]"));
+                                parameterList.Append("@P");
+                                parameterList.Append(index);
+
+                                command.AddParameter(Invariant($"@P{index}"), field.Value);
+
+                                index++;
+                            }
+
+                            parameterList.Append(')');
+                            fieldList.Append(parameterList);
+
+                            command.CommandText = fieldList.ToString();
+
+                            await command.ExecuteNonQueryAsync().ConfigureAwait(false);
+                        }
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                SelfLog.WriteLine("Unable to write batch of {0} log events to the database due to following error: {1}",
+                    events.Count(), ex);
+                throw;
+            }
+        }
+    }
+}