{

/// <summary>

/// Generate a unique temporary column name for the given schema.

/// Use tag to independently create multiple temporary, unique column

/// names for a single transform.

/// </summary>

public static string GetTempColumnName(this Schema schema, string tag = null)

{

Contracts.CheckValue(schema, nameof(schema));

int col;

if (!string.IsNullOrWhiteSpace(tag) && !schema.TryGetColumnIndex(tag, out col))

return tag;

for (int i = 0; ; i++)

{

string name = string.IsNullOrWhiteSpace(tag) ?

string.Format("temp_{0:000}", i) :

string.Format("temp_{0}_{1:000}", tag, i);

if (!schema.TryGetColumnIndex(name, out col))

return name;

}

}

/// <summary>

/// Generate n unique temporary column names for the given schema.

/// Use tag to independently create multiple temporary, unique column

/// names for a single transform.

/// </summary>

public static string[] GetTempColumnNames(this Schema schema, int n, string tag = null)

{

Contracts.CheckValue(schema, nameof(schema));

Contracts.Check(n > 0, "n");

var res = new string[n];

int j = 0;

for (int i = 0; i < n; i++)

{

for (; ; )

{

string name = string.IsNullOrWhiteSpace(tag) ?

string.Format("temp_{0:000}", j) :

string.Format("temp_{0}_{1:000}", tag, j);

j++;

int col;

if (!schema.TryGetColumnIndex(name, out col))

{

res[i] = name;

break;

}

}

}

return res;

}

/// <summary>

/// Get the row count from the input view by any means necessary, even explicit enumeration

/// and counting if <see cref="IDataView.GetRowCount"/> insists on returning <c>null</c>.

/// </summary>

public static long ComputeRowCount(IDataView view)

{

long? countNullable = view.GetRowCount();

if (countNullable != null)

return countNullable.Value;

long count = 0;

using (var cursor = view.GetRowCursor(col => false))

{

while (cursor.MoveNext())

count++;

}

return count;

}

/// <summary>

/// Get the target number of threads to use, given a host and another indicator of thread count.

/// When num > 0, this uses num limited to twice what the host says. Otherwise, if preferOne

/// is true, it returns 1. Otherwise, it returns what the host says.

/// </summary>

public static int GetThreadCount(IHost host, int num = 0, bool preferOne = false)

{

Contracts.CheckValue(host, nameof(host));

int conc = host.ConcurrencyFactor;

if (conc <= 0)

conc = Math.Max(2, Environment.ProcessorCount - 1);

if (num > 0)

return Math.Min(num, 2 * conc);

if (preferOne)

return 1;

return conc;

}

/// <summary>

/// Try to create a cursor set from upstream and consolidate it here. The host determines

/// the target cardinality of the cursor set.

/// </summary>

public static bool TryCreateConsolidatingCursor(out RowCursor curs,

IDataView view, Func<int, bool> predicate, IHost host, Random rand)

{

Contracts.CheckValue(host, nameof(host));

host.CheckValue(view, nameof(view));

host.CheckValue(predicate, nameof(predicate));

int cthd = GetThreadCount(host);

host.Assert(cthd > 0);

if (cthd == 1 || !AllCachable(view.Schema, predicate))

{

curs = null;

return false;

}

var inputs = view.GetRowCursorSet(predicate, cthd, rand);

host.Check(Utils.Size(inputs) > 0);

if (inputs.Length == 1)

curs = inputs[0];

else

{

// We have a somewhat arbitrary batch size of about 64 for buffering results from the

// intermediate cursors, since that at least empirically for most datasets seems to

// strike a nice balance between a size large enough to benefit from parallelism but

// small enough so as to not be too onerous to keep in memory.

const int batchSize = 64;

curs = DataViewUtils.ConsolidateGeneric(host, inputs, batchSize);

}

return true;

}

/// <summary>

/// From the given input cursor, split it into a cursor set with the given

/// cardinality. If not all the active columns are cachable, this will only

/// produce the given input cursor.

/// </summary>

public static RowCursor[] CreateSplitCursors(IChannelProvider provider, RowCursor input, int num)

{

Contracts.CheckValue(provider, nameof(provider));

provider.CheckValue(input, nameof(input));

if (num <= 1)

return new RowCursor[1] { input };

// If any active columns are not cachable, we can't split.

if (!AllCachable(input.Schema, input.IsColumnActive))

return new RowCursor[1] { input };

// REVIEW: Should we limit the cardinality to some reasonable size?

// REVIEW: Ideally a splitter should be owned by a data view

// we might split, so that we can share the cache pools among multiple

// cursors.

// REVIEW: Keep the utility method here, move this splitter stuff

// to some other file.

return Splitter.Split(provider, input.Schema, input, num);

}

/// <summary>

/// Return whether all the active columns, as determined by the predicate, are

/// cachable - either primitive types or vector types.

/// </summary>

public static bool AllCachable(Schema schema, Func<int, bool> predicate)

{

Contracts.CheckValue(schema, nameof(schema));

Contracts.CheckValue(predicate, nameof(predicate));

for (int col = 0; col < schema.Count; col++)

{

if (!predicate(col))

continue;

var type = schema[col].Type;

if (!IsCachable(type))

return false;

}

return true;

}

/// <summary>

/// Determine whether the given type is cachable - either a primitive type or a vector type.

/// </summary>

public static bool IsCachable(this ColumnType type)

{

return type != null && (type.IsPrimitive || type.IsVector);

}

/// <summary>

/// Tests whether the cursors are mutually compatible for consolidation,

/// that is, they all are non-null, have the same schemas, and the same

/// set of columns are active.

/// </summary>

public static bool SameSchemaAndActivity(RowCursor[] cursors)

{

// There must be something to actually consolidate.

if (Utils.Size(cursors) == 0)

return true;

var firstCursor = cursors[0];

if (firstCursor == null)

return false;

if (cursors.Length == 1)

return true;

var schema = firstCursor.Schema;

// All cursors must have the same schema.

for (int i = 1; i < cursors.Length; ++i)

{

if (cursors[i] == null || cursors[i].Schema != schema)

return false;

}

// All cursors must have the same columns active.

for (int c = 0; c < schema.Count; ++c)

{

bool active = firstCursor.IsColumnActive(c);

for (int i = 1; i < cursors.Length; ++i)

{

if (cursors[i].IsColumnActive(c) != active)

return false;

}

}

return true;

}

/// <summary>

/// Given a parallel cursor set, this consolidates them into a single cursor. The batchSize

/// is a hint used for efficiency.

/// </summary>

public static RowCursor ConsolidateGeneric(IChannelProvider provider, RowCursor[] inputs, int batchSize)

{

Contracts.CheckValue(provider, nameof(provider));

provider.CheckNonEmpty(inputs, nameof(inputs));

provider.CheckParam(batchSize >= 0, nameof(batchSize));

if (inputs.Length == 1)

return inputs[0];

object[] pools = null;

return Splitter.Consolidate(provider, inputs, batchSize, ref pools);

}

/// <summary>

/// A convenience class to facilitate the creation of a split, as well as a convenient

/// place to store shared resources that can be reused among multiple splits of a cursor

/// with the same schema. Since splitting also returns a consolidator, this also contains

/// a consolidating logic.

///

/// In a very rough sense, both the splitters and consolidators are written in the same way:

/// For all input cursors, and all active columns, an "in pipe" is created. A worker thread

/// per input cursor busily retrieves values from the cursors and stores them in the "in

/// pipe." At appropriate times, "batch" objects are synthesized from the inputs consumed

/// thusfar, and inserted into a blocking collection. The output cursor or cursors likewise

/// have a set of "out pipe" instances, one per each of the active columns, through which

/// successive batches are presented for consumption by the user of the output cursors. Of

/// course, both split and consolidate have many details from which they differ, for example, the

/// consolidator must accept batches as they come and reconcile them among multiple inputs,

/// while the splitter is more free.

///

/// It is ideal if a data view that could be split retains one of these objects itself,

/// so that multiple splittings will have the capability of sharing buffers from cursoring

/// to cursoring, but this is not required.

/// </summary>

private sealed class Splitter

{

private readonly Schema _schema;

private readonly object[] _cachePools;

/// <summary>

/// Pipes, in addition to column values, will also communicate extra information

/// enumerated within this. This enum serves the purpose of providing nice readable

/// indices to these "extra" information in pipes.

/// </summary>

private enum ExtraIndex

{

Id,

#pragma warning disable MSML_GeneralName // Allow for this private enum.

_Lim

#pragma warning restore MSML_GeneralName

}

private Splitter(Schema schema)

{

Contracts.AssertValue(schema);

_schema = schema;

_cachePools = new object[_schema.Count + (int)ExtraIndex._Lim];

}

public static RowCursor Consolidate(IChannelProvider provider, RowCursor[] inputs, int batchSize, ref object[] ourPools)

{

Contracts.AssertValue(provider);

using (var ch = provider.Start("Consolidate"))

{

return ConsolidateCore(provider, inputs, ref ourPools, ch);

}

}

private static RowCursor ConsolidateCore(IChannelProvider provider, RowCursor[] inputs, ref object[] ourPools, IChannel ch)

{

ch.CheckNonEmpty(inputs, nameof(inputs));

if (inputs.Length == 1)

return inputs[0];

ch.CheckParam(SameSchemaAndActivity(inputs), nameof(inputs), "Inputs not compatible for consolidation");

RowCursor cursor = inputs[0];

var schema = cursor.Schema;

ch.CheckParam(AllCachable(schema, cursor.IsColumnActive), nameof(inputs), "Inputs had some uncachable input columns");

int[] activeToCol;

int[] colToActive;

Utils.BuildSubsetMaps(schema.Count, cursor.IsColumnActive, out activeToCol, out colToActive);

// Because the schema of the consolidator is not necessary fixed, we are merely

// opportunistic about buffer sharing, from cursoring to cursoring. If we can do

// it easily, great, if not, no big deal.

if (Utils.Size(ourPools) != schema.Count)

ourPools = new object[schema.Count + (int)ExtraIndex._Lim];

// Create the out pipes.

OutPipe[] outPipes = new OutPipe[activeToCol.Length + (int)ExtraIndex._Lim];

for (int i = 0; i < activeToCol.Length; ++i)

{

int c = activeToCol[i];

ColumnType type = schema[c].Type;

var pool = GetPool(type, ourPools, c);

outPipes[i] = OutPipe.Create(type, pool);

}

int idIdx = activeToCol.Length + (int)ExtraIndex.Id;

outPipes[idIdx] = OutPipe.Create(NumberType.UG, GetPool(NumberType.UG, ourPools, idIdx));

// Create the structures to synchronize between the workers and the consumer.

const int toConsumeBound = 4;

var toConsume = new BlockingCollection<Batch>(toConsumeBound);

var batchColumnPool = new MadeObjectPool<BatchColumn[]>(() => new BatchColumn[outPipes.Length]);

Thread[] workers = new Thread[inputs.Length];

MinWaiter waiter = new MinWaiter(workers.Length);

bool done = false;

for (int t = 0; t < workers.Length; ++t)

{

var localCursor = inputs[t];

ch.Assert(localCursor.State == CursorState.NotStarted);

// Note that these all take ownership of their respective cursors,

// so they all handle their disposal internal to the thread.

workers[t] = Utils.CreateBackgroundThread(() =>

{

// This will be the last batch sent in the finally. If iteration procedes without

// error, it will remain null, and be sent as a sentinel. If iteration results in

// an exception that we catch, the exception catching block will set this to an

// exception bearing block, and that will be passed along as the last block instead.

Batch lastBatch = null;

try

{

using (localCursor)

{

InPipe[] inPipes = new InPipe[outPipes.Length];

for (int i = 0; i < activeToCol.Length; ++i)

inPipes[i] = outPipes[i].CreateInPipe(RowCursorUtils.GetGetterAsDelegate(localCursor, activeToCol[i]));

inPipes[idIdx] = outPipes[idIdx].CreateInPipe(localCursor.GetIdGetter());

long oldBatch = 0;

int count = 0;

// This event is used to synchronize ourselves using a MinWaiter

// so that we add batches to the consumer queue at the appropriate time.

ManualResetEventSlim waiterEvent = null;

Action pushBatch = () =>

{

if (count > 0)

{

var batchColumns = batchColumnPool.Get();

for (int i = 0; i < inPipes.Length; ++i)

batchColumns[i] = inPipes[i].GetBatchColumnAndReset();

// REVIEW: Is it worth not allocating new Batch object for each batch?

var batch = new Batch(batchColumnPool, batchColumns, count, oldBatch);

count = 0;

// The waiter event should never be null since this is only

// called after a point where waiter.Register has been called.

ch.AssertValue(waiterEvent);

waiterEvent.Wait();

waiterEvent = null;

toConsume.Add(batch);

}

};

// Handle the first one separately, then go into the main loop.

if (localCursor.MoveNext() && !done)

{

oldBatch = localCursor.Batch;

foreach (var pipe in inPipes)

pipe.Fill();

count++;

// Register with the min waiter that we want to wait on this batch number.

waiterEvent = waiter.Register(oldBatch);

while (localCursor.MoveNext() && !done)

{

if (oldBatch != localCursor.Batch)

{

ch.Assert(count == 0 || localCursor.Batch > oldBatch);

pushBatch();

oldBatch = localCursor.Batch;

waiterEvent = waiter.Register(oldBatch);

}

foreach (var pipe in inPipes)

pipe.Fill();

count++;

}

pushBatch();

}

}

}

catch (Exception ex)

{

// Whoops, we won't be sending null as the sentinel now.

lastBatch = new Batch(ex);

toConsume.Add(new Batch(ex));

}

finally

{

if (waiter.Retire() == 0)

{

if (lastBatch == null)

{

// If it wasn't null, this already sent along an exception bearing batch, in which

// case sending the sentinel is unnecessary and unhelpful.

toConsume.Add(null);

}

toConsume.CompleteAdding();

}

}

});

workers[t].Start();

}

Action quitAction = () =>

{

done = true;

var myOutPipes = outPipes;

foreach (var batch in toConsume.GetConsumingEnumerable())

{

if (batch == null)

continue;

batch.SetAll(myOutPipes);

foreach (var outPipe in myOutPipes)

outPipe.Unset();

}

foreach (Thread thread in workers)

thread.Join();

};

return new Cursor(provider, schema, activeToCol, colToActive, outPipes, toConsume, quitAction);

}

private static object GetPool(ColumnType type, object[] pools, int poolIdx)

{

Func<object[], int, object> func = GetPoolCore<int>;

var method = func.GetMethodInfo().GetGenericMethodDefinition().MakeGenericMethod(type.RawType);

return method.Invoke(null, new object[] { pools, poolIdx });

}

private static MadeObjectPool<T[]> GetPoolCore<T>(object[] pools, int poolIdx)

{

var pool = pools[poolIdx] as MadeObjectPool<T[]>;

if (pool == null)

pools[poolIdx] = pool = new MadeObjectPool<T[]>(() => null);

return pool;

}

public static RowCursor[] Split(IChannelProvider provider, Schema schema, RowCursor input, int cthd)

{

Contracts.AssertValue(provider, "provider");

var splitter = new Splitter(schema);

using (var ch = provider.Start("CursorSplitter"))

{

var result = splitter.SplitCore(provider, input, cthd);

return result;

}

}

private RowCursor[] SplitCore(IChannelProvider ch, RowCursor input, int cthd)

{

Contracts.AssertValue(ch);

ch.AssertValue(input);

ch.Assert(input.Schema == _schema);

ch.Assert(cthd >= 2);

ch.Assert(AllCachable(_schema, input.IsColumnActive));

// REVIEW: Should the following be configurable?

// How would we even expose these sorts of parameters to a user?

const int maxBatchCount = 128;

const int toConsumeBound = 4;

// Create the mappings between active column index, and column index.

int[] activeToCol;

int[] colToActive;

Utils.BuildSubsetMaps(_schema.Count, input.IsColumnActive, out activeToCol, out colToActive);

Func<RowCursor, int, InPipe> createFunc = CreateInPipe<int>;

var inGenMethod = createFunc.GetMethodInfo().GetGenericMethodDefinition();

object[] arguments = new object[] { input, 0 };

// Only one set of in-pipes, one per column, as well as for extra side information.

InPipe[] inPipes = new InPipe[activeToCol.Length + (int)ExtraIndex._Lim];

// There are as many sets of out pipes as there are output cursors.

OutPipe[][] outPipes = new OutPipe[cthd][];

for (int i = 0; i < cthd; ++i)

outPipes[i] = new OutPipe[inPipes.Length];

// For each column, create the InPipe, and all OutPipes per output cursor.

for (int c = 0; c < activeToCol.Length; ++c)

{

ch.Assert(0 <= activeToCol[c] && activeToCol[c] < _schema.Count);

ch.Assert(c == 0 || activeToCol[c - 1] < activeToCol[c]);

ch.Assert(input.IsColumnActive(activeToCol[c]));

var type = input.Schema[activeToCol[c]].Type;

ch.Assert(type.IsCachable());

arguments[1] = activeToCol[c];

var inPipe = inPipes[c] =

(InPipe)inGenMethod.MakeGenericMethod(type.RawType).Invoke(this, arguments);

for (int i = 0; i < cthd; ++i)

outPipes[i][c] = inPipe.CreateOutPipe(type);

}

// Beyond the InPipes corresponding to column values, we have extra side info pipes.

int idIdx = activeToCol.Length + (int)ExtraIndex.Id;

inPipes[idIdx] = CreateIdInPipe(input);

for (int i = 0; i < cthd; ++i)

outPipes[i][idIdx] = inPipes[idIdx].CreateOutPipe(NumberType.UG);

var toConsume = new BlockingCollection<Batch>(toConsumeBound);

var batchColumnPool = new MadeObjectPool<BatchColumn[]>(() => new BatchColumn[inPipes.Length]);

bool done = false;

int outputsRunning = cthd;

// Set up and start the thread that consumes the input, and utilizes the InPipe

// instances to compose the Batch objects. The thread takes ownership of the

// cursor, and so handles its disposal.

Thread thread = Utils.CreateBackgroundThread(

() =>

{

Batch lastBatch = null;

try

{

using (input)

{

long batchId = 0;

int count = 0;

Action pushBatch = () =>

{

var batchColumns = batchColumnPool.Get();

for (int c = 0; c < inPipes.Length; ++c)

batchColumns[c] = inPipes[c].GetBatchColumnAndReset();

// REVIEW: Is it worth not allocating new Batch object for each batch?

var batch = new Batch(batchColumnPool, batchColumns, count, batchId++);

count = 0;

toConsume.Add(batch);

};

while (input.MoveNext() && !done)

{

foreach (var pipe in inPipes)

pipe.Fill();

if (++count >= maxBatchCount)

pushBatch();

}

if (count > 0)

pushBatch();

}

}

catch (Exception ex)

{

lastBatch = new Batch(ex);

}

finally

{

// The last batch might be an exception, as in the above case. We pass along the exception

// bearing batch as the first of the last batches, so that the first worker to encounter this

// will know to throw. The remaining get the regular "stop working" null sentinel.

toConsume.Add(lastBatch);

for (int i = 1; i < cthd; ++i)

toConsume.Add(null);

toConsume.CompleteAdding();

}

});

thread.Start();

Action quitAction = () =>

{

int remaining = Interlocked.Decrement(ref outputsRunning);

ch.Assert(remaining >= 0);

if (remaining == 0)

{

done = true;

// A relatively quick and convenient way to dispose of batches, is to use

// set/unset on some output pipes repeatedly. Since all have been disposed,

// we may as well use the first set of output pipes.

var myOutPipes = outPipes[0];

foreach (var batch in toConsume.GetConsumingEnumerable())

{

if (batch == null)

continue;

batch.SetAll(myOutPipes);

foreach (var outPipe in myOutPipes)

outPipe.Unset();

}

thread.Join();

}

};

var cursors = new Cursor[cthd];

for (int i = 0; i < cthd; ++i)

cursors[i] = new Cursor(ch, _schema, activeToCol, colToActive, outPipes[i], toConsume, quitAction);

return cursors;

}

/// <summary>

/// An in pipe creator intended to be used from the splitter only.

/// </summary>

private InPipe CreateInPipe<T>(Row input, int col)

{

Contracts.AssertValue(input);

Contracts.Assert(0 <= col && col < _schema.Count);

return CreateInPipeCore(col, input.GetGetter<T>(col));

}

/// <summary>

/// An in pipe creator intended to be used from the splitter only.

/// </summary>

private InPipe CreateIdInPipe(Row input)

{

Contracts.AssertValue(input);

return CreateInPipeCore(_schema.Count + (int)ExtraIndex.Id, input.GetIdGetter());

}

private InPipe CreateInPipeCore<T>(int poolIdx, ValueGetter<T> getter)

{

Contracts.Assert(0 <= poolIdx && poolIdx < _cachePools.Length);

Contracts.AssertValue(getter);

var pool = (MadeObjectPool<T[]>)_cachePools[poolIdx];

if (pool == null)

{

// REVIEW: If we changed our InPipe behavior so that it only worked over a maximum size

// in all scenarios, both during splitting and consolidating, it would be possible to set this

// to be of fixed size so we don't have to do reallocation.

Interlocked.CompareExchange(ref _cachePools[poolIdx], new MadeObjectPool<T[]>(() => null), null);

pool = (MadeObjectPool<T[]>)_cachePools[poolIdx];

}

return InPipe.Create(pool, getter);

}

/// <summary>

/// There is one of these created per input cursor, per "channel" of information

/// (necessary channels include values from active columns, as well as additional

/// side information), in both splitting and consolidating. This is a running buffer

/// of the input cursor's values. It is used to create <see cref="BatchColumn"/> objects.

/// </summary>

private abstract class InPipe

{

public int Count { get; protected set; }

private InPipe()

{

}

public abstract void Fill();

public abstract BatchColumn GetBatchColumnAndReset();

public static InPipe Create<T>(MadeObjectPool<T[]> pool, ValueGetter<T> getter)

{

return new Impl<T>(pool, getter);

}

/// <summary>

/// Creates an out pipe corresponding to the in pipe. This is useful for the splitter,

/// when we are creating an in pipe.

/// </summary>

public abstract OutPipe CreateOutPipe(ColumnType type);

private sealed class Impl<T> : InPipe

{

private readonly MadeObjectPool<T[]> _pool;

private readonly ValueGetter<T> _getter;

private T[] _values;

public Impl(MadeObjectPool<T[]> pool, ValueGetter<T> getter)

{

Contracts.AssertValue(pool);

Contracts.AssertValue(getter);

_pool = pool;

_getter = getter;

_values = _pool.Get();

Contracts.AssertValueOrNull(_values);

}

public override void Fill()

{

Utils.EnsureSize(ref _values, Count + 1, keepOld: true);

_getter(ref _values[Count++]);

}

public override BatchColumn GetBatchColumnAndReset()

{

// REVIEW: Is it worth avoiding an allocation of these new BatchColumn objects?

var retval = new BatchColumn.Impl<T>(_values, Count);

_values = _pool.Get();

Count = 0;

return retval;

}

public override OutPipe CreateOutPipe(ColumnType type)

{

Contracts.AssertValue(type);

Contracts.Assert(typeof(T) == type.RawType);

return OutPipe.Create(type, _pool);

}

}

}

/// <summary>

/// These are objects continuously created by the <see cref="InPipe"/> to spin off the

/// values they have collected. They are collected into a <see cref="Batch"/>

/// object, and eventually one is consumed by an <see cref="OutPipe"/> instance.

/// </summary>

private abstract class BatchColumn

{

public readonly int Count;

private BatchColumn(int count)

{

Contracts.Assert(count > 0);

Count = count;

}

public sealed class Impl<T> : BatchColumn

{

public readonly T[] Values;

public Impl(T[] values, int count)

: base(count)

{

Contracts.Assert(Utils.Size(values) >= count);

Values = values;

}

}

}

/// <summary>

/// This holds a collection of <see cref="BatchColumn"/> objects, which together hold all

/// the values from a set of rows from the input cursor. These are produced as needed

/// by the input cursor reader, and consumed by each of the output cursors.

///

/// This class also serves a secondary role in marshalling exceptions thrown in the workers

/// producing batches, into the threads consuming these batches.

/// <see cref="HasException"/> lets us know if this is one of these "special" batches.

/// If it is, then the <see cref="SetAll"/> method will throw whenever it is called, by the

/// consumer of the batches.

/// </summary>

private sealed class Batch

{

private readonly MadeObjectPool<BatchColumn[]> _pool;

private readonly BatchColumn[] _batchColumns;

public readonly int Count;

public readonly long BatchId;

private readonly Exception _ex;

public bool HasException { get { return _ex != null; } }

/// <summary>

/// Construct a batch object to communicate the <see cref="BatchColumn"/> objects to consumers.

/// </summary>

public Batch(MadeObjectPool<BatchColumn[]> pool, BatchColumn[] batchColumns, int count, long batchId)

{

Contracts.AssertValue(pool);

Contracts.AssertValue(batchColumns);

Contracts.Assert(count > 0);

Contracts.Assert(batchId >= 0);

Contracts.Assert(batchColumns.All(bc => bc.Count == count));

_pool = pool;

_batchColumns = batchColumns;

Count = count;

BatchId = batchId;

Contracts.Assert(!HasException);

}

/// <summary>

/// Construct a batch object to communicate that something went wrong. In this case all other fields

/// will have default values.

/// </summary>

public Batch(Exception ex)

{