Merge pull request #57 from Bilal-io/bugfix/sql-server-batches

Fixes #56 SqlServer Sink not working with batches

Author: bowencode

Extensions/SqlServer/Cosmos.DataTransfer.SqlServerExtension/SqlServerDataSinkExtension.cs

@@ -36,11 +36,12 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
                         bulkCopy.ColumnMappings.Add(new SqlBulkCopyColumnMapping(dbColumn.ColumnName, dbColumn.ColumnName));
                     }
 
+                    var dataTable = new DataTable();
+                    dataTable.Columns.AddRange(dataColumns.Values.ToArray());
+
                     var batches = dataItems.Buffer(settings.BatchSize);
                     await foreach (var batch in batches.WithCancellation(cancellationToken))
                     {
-                        var dataTable = new DataTable();
-                        dataTable.Columns.AddRange(dataColumns.Values.ToArray());
                         foreach (var item in batch)
                         {
                             var fieldNames = item.GetFieldNames().ToList();
@@ -77,6 +78,7 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
                             dataTable.Rows.Add(row);
                         }
                         await bulkCopy.WriteToServerAsync(dataTable, cancellationToken);
+                        dataTable.Clear();
                     }
 
                     await transaction.CommitAsync(cancellationToken);
@@ -91,4 +93,4 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
             await connection.CloseAsync();
         }
     }
-}
+}