I am consuming a large csv file with a lot of columns. I would like to split that file into n different types objects and then bulk insert those different objects into n different tables. I would like to optimize as much as possible and not hold the entire csv file in memory while I writing to the db. If there is an error writing to 1 table, all tables must rollback their transactions. I am passing the token source around because in the event that 1 table has an issue, I would like all other tables to stop processing and rollback.
To simplify the object creation/definition I am using ExpandoObjects.
The TableDetails class holds the table name, SqlConnection, and the transaction that will be used to write to a table.
I have included the ObjectDataReader that implements IDataReader to facilitate sending an IEnumerable through a SqlBulkCopy.
Why is it that when I run my Dummy SaveToDb Method, all 3 tables get their own thread to make and write to console, but when I run the real SaveToDb method, all work is done on 1 thread?
What do i need to do to get the same behavior i get on my test method?
dummy tables created as follows
create table Table1(Column1 int)
create table Table2(Column1 int)
create table Table3(Column1 int)
Main Work Here
void Main()
{
var tokenSource = new CancellationTokenSource();
var sqlConnection1 = new SqlConnection("Some Connection String");
sqlConnection1.Open();
var sqlConnection2= new SqlConnection("Some Connection String");
sqlConnection2.Open();
var sqlConnection3 = new SqlConnection("Some Connection String");
sqlConnection3.Open();
var details = new List<TableDetails>()
{
new TableDetails(){ TableName = "Table1", Connection = sqlConnection1, Transaction = sqlConnection1.BeginTransaction(), ColumnMap = new Dictionary<int,string>(){{0, "Colunm1"}}},
new TableDetails(){ TableName = "Table2", Connection = sqlConnection2, Transaction = sqlConnection2.BeginTransaction(), ColumnMap = new Dictionary<int,string>(){{1, "Colunm1"}}},
new TableDetails(){ TableName = "Table3", Connection = sqlConnection3, Transaction = sqlConnection3.BeginTransaction(),ColumnMap = new Dictionary<int,string>(){{2, "Colunm1"}}},
};
var lines = GetLines(100);
var tasks = lines
.SelectMany(e => SplitUp(e, details))
.GroupBy(e => e.Item1)
.Select(e => new { e.Key, Value = e.Select(v => MakeExpando(v.Item2, v.Item1)) })
// .Select(e => SaveToDbTest(e.Key, e.Value));
.Select(e => SaveToDb(e.Value, e.Key, tokenSource));
Task.WhenAll(tasks).Wait();
foreach (var detail in details)
{
detail.Transaction.Commit();
detail.Connection.Dispose();
}
}
public IEnumerable<string> GetLines(int size)
{
var rand = new Random();
for (int i = 0; i < size; i++)
yield return $"{rand.Next(1, 100)},{rand.Next(1, 100)},{rand.Next(1, 100)}";
}
public IEnumerable<(TableDetails, string)> SplitUp(string line, List<TableDetails> details)
{
foreach (var detail in details)
{
yield return (detail, line);
}
}
public ExpandoObject MakeExpando(string line, TableDetails details)
{
Console.WriteLine($"Thread ID:{Thread.CurrentThread.ManagedThreadId} Making Expando for Table {details.TableName}");
var items = line.Split(',');
dynamic retVal = new ExpandoObject();
var r = retVal as IDictionary<string, object>;
object value;
foreach(var map in details.ColumnMap)
{
value = items[map.Key];
r.Add(map.Value, value);
}
return retVal;
}
public Task SaveToDbTest(TableDetails details, IEnumerable<ExpandoObject> items)
{
var retVal = Task.Factory.StartNew(() =>
{
foreach (var i in items)
{
Console.WriteLine($"Thread ID:{Thread.CurrentThread.ManagedThreadId} Saving To Table {details.TableName} => {i}");
}
});
return retVal;
}
private async Task SaveToDb<T>(IEnumerable<T> items, TableDetails details, CancellationTokenSource tokenSource) where T : IDictionary<string, object>
{
var bulkCopy = new SqlBulkCopy(details.Connection, SqlBulkCopyOptions.Default, details.Transaction);
try
{
bulkCopy.DestinationTableName = details.TableName;
bulkCopy.BatchSize = 20;
bulkCopy.BulkCopyTimeout = (int)TimeSpan.FromMinutes(120).TotalSeconds;
bulkCopy.EnableStreaming = true;
var reader = new ObjectDataReader<T>(items, details.ColumnMap.Count());
var stopwatch = new Stopwatch();
stopwatch.Start();
await bulkCopy.WriteToServerAsync(reader, tokenSource.Token);
stopwatch.Stop();
Console.WriteLine($"completed db write in {stopwatch.Elapsed}");
}
catch (Exception ex)
{
if (ex.GetType() != typeof(TaskCanceledException))
tokenSource.Cancel();
throw;
}
}
Table Details
public class TableDetails
{
public string TableName { get; set; }
public SqlConnection Connection { get; set; }
public SqlTransaction Transaction { get; set; }
public Dictionary<int, string> ColumnMap {get; set;}
}
And an IDataReader
public class ObjectDataReader<TData> : IDataReader where TData : IDictionary<string, object>
{
private IEnumerator<TData> _dataEnumerator;
private Dictionary<int, string> _indexToName;
public ObjectDataReader(IEnumerable<TData> data, int propertyCount)
{
_fieldCount = propertyCount;
_dataEnumerator = data.GetEnumerator();
}
#region IDataReader Members
public void Close()
{
Dispose();
}
public int Depth => 1;
public DataTable GetSchemaTable()
{
return null;
}
public bool IsClosed => _dataEnumerator == null;
public bool NextResult()
{
return false;
}
public bool Read()
{
if (IsClosed)
throw new ObjectDisposedException(GetType().Name);
Console.WriteLine($"Thread ID:{Thread.CurrentThread.ManagedThreadId} Reading next item");
return _dataEnumerator.MoveNext();
}
public int RecordsAffected => -1;
#endregion
#region IDisposable Members
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
private void Dispose(bool disposing)
{
if (disposing)
{
if (_dataEnumerator != null)
{
_dataEnumerator.Dispose();
_dataEnumerator = null;
}
}
}
#endregion
#region IDataRecord Members
public int GetOrdinal(string name)
{
throw new NotImplementedException();
}
public object GetValue(int i)
{
if (_dataEnumerator == null)
throw new ObjectDisposedException(GetType().Name);
var item = _dataEnumerator.Current;
if (_indexToName == null)
{
_indexToName = item
.Select((e, id) => new { Index = id, e.Key })
.ToDictionary(k => k.Index, v => v.Key);
}
if (_indexToName.Count <= i)
return null;
return item[_indexToName[i]];
}
private int _fieldCount;
public int FieldCount => _fieldCount; //throw new NotImplementedException(); // s_propertyAccessorCache.Value.Accessors.Count;
#region Not Implemented Members
public bool GetBoolean(int i)
{
throw new NotImplementedException();
}
public byte GetByte(int i)
{
throw new NotImplementedException();
}
public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length)
{
throw new NotImplementedException();
}
public char GetChar(int i)
{
throw new NotImplementedException();
}
public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length)
{
throw new NotImplementedException();
}
public IDataReader GetData(int i)
{
throw new NotImplementedException();
}
public string GetDataTypeName(int i)
{
throw new NotImplementedException();
}
public DateTime GetDateTime(int i)
{
throw new NotImplementedException();
}
public decimal GetDecimal(int i)
{
throw new NotImplementedException();
}
public double GetDouble(int i)
{
throw new NotImplementedException();
}
public Type GetFieldType(int i)
{
throw new NotImplementedException();
}
public float GetFloat(int i)
{
throw new NotImplementedException();
}
public Guid GetGuid(int i)
{
throw new NotImplementedException();
}
public short GetInt16(int i)
{
throw new NotImplementedException();
}
public int GetInt32(int i)
{
throw new NotImplementedException();
}
public long GetInt64(int i)
{
throw new NotImplementedException();
}
public string GetName(int i)
{
throw new NotImplementedException();
}
public string GetString(int i)
{
throw new NotImplementedException();
}
public int GetValues(object[] values)
{
throw new NotImplementedException();
}
public bool IsDBNull(int i)
{
var val = GetValue(i);
return val == null;
throw new NotImplementedException();
}
public object this[string name]
{
get { throw new NotImplementedException(); }
}
public object this[int i]
{
get { throw new NotImplementedException(); }
}
#endregion
#endregion
}
Output when writing to db.
Thread ID:60 Reading next item
Thread ID:60 Making Expando for Table Table1
Thread ID:60 Reading next item
Thread ID:60 Making Expando for Table Table1
...
Thread ID:60 Reading next item
Thread ID:60 Making Expando for Table Table2
Thread ID:60 Reading next item
...
Thread ID:60 Making Expando for Table Table3
Thread ID:60 Reading next item
Thread ID:60 Making Expando for Table Table3
Thread ID:60 Reading next item
Output when writing to console.
Thread ID:62 Making Expando for Table Table2
Thread ID:71 Making Expando for Table Table3
Thread ID:69 Making Expando for Table Table1
Thread ID:62 Saving To Table Table2 => System.Dynamic.ExpandoObject
Thread ID:62 Making Expando for Table Table2
...
Thread ID:71 Saving To Table Table3 => System.Dynamic.ExpandoObject
Thread ID:71 Making Expando for Table Table3
...
Thread ID:62 Making Expando for Table Table2
Thread ID:62 Saving To Table Table2 => System.Dynamic.ExpandoObject
Thread ID:62 Making Expando for Table Table2
Thread ID:62 Saving To Table Table2 => System.Dynamic.ExpandoObject
Thread ID:62 Making Expando for Table Table2