using System;

using System.Data;

using System.Linq;

using System.Collections.Generic;

using System.Diagnostics;

​

/*

Although this should ony take around a second to batch 1 million rows,

so could just be run on all import entities without any performance hit,

this could introduce inefficency for imports that don't cause deadlocks,

depending on the make up the import file.

​

For example if we loaded a file of 50  records for the same 

employee, this would result in 50 batches, fine for absences as that is

what we would have to do in order to avoid the deadlocks, but for payrollchanges

where deadlocks are not an issue, this would effectively result in the rows being 

processed sequentially.

​

Therefore this should also be conifugrable per entity.

​

Can we have a table in ControlDB that we have permanent dbo permissions on?

​

Suggest Table:

​

ImportPerformanceConfig

​

TableName | Stack | SaveDataMaxDegreeOfParallelism | UseDeadlockPreventionBatches

​

Would only need to populate for entities that do not use the defaults (8 and false)

​

Would need to add stack value as orchestrator variable too if not there already, and feed it into the import runbook

​

it would be fairly easy to expand this and add in another column for an additional identifier, such as in the case of absence

having an individual config for AbsenceType = Sickess where deadlockbatches are used and parallelism = 2,

and the rest of absence having batches = true and parellelism = 8, or again a separate one for absencetype = holiday as well

​

This would allow us to truly maximise the potential of all our imports, limiting only where entirely necessary.

​

This could also be done in a way that could deal with an absence file which does have multiple types in one import.

​

*/

​

​

​

public class ImportPerformanceConfig

{   

    public string TableName {get; set;}

    public string Stack {get; set; }

    public int? SaveDataMaxDegreeOfParallelism { get; set; }

    public bool? UseDeadlockPreventionBatches { get; set; }

}

​

public class MockDbResponse

{

    public List<ImportPerformanceConfig> ImportPerformanceConfigs 

    { 

        get

        {

            return new List<ImportPerformanceConfig>

            {

                new ImportPerformanceConfig

                {

                    TableName = "ABSENCE",

                    Stack = "PROD",             

                    SaveDataMaxDegreeOfParallelism = 8,

                    UseDeadlockPreventionBatches = true

                }

            };

        }   

    }   

}

​

​

public class Program

{

    private static int defaultSaveDataMaxDegreeOfParallelism = 8; 

    private static string stack = "Prod";

    static DataTable GetTestData()

    {

        var table = new DataTable();

        table.TableName = "ABSENCE";

        table.Columns.Add("PEOPLE_ID", typeof(string));

        for (var i = 0; i < 7; i ++)

        {

            table.Rows.Add("Mark");

        }

        for (var i = 0; i < 0; i ++)

        {

            table.Rows.Add("Jeremey");

        }       

        for (var i = 0; i < 0; i ++)

        {

            table.Rows.Add("Big Mad Andy");

        }   

        for (var i = 0; i < 0; i ++)

        {

            table.Rows.Add("NAME" + i);

        }               

        return table;

    }

    public static void Main()

    {

        var mockDbResponse = new MockDbResponse();

        var importData = GetTestData().AsEnumerable().ToList(); 

        var tableName = importData.First().Table.TableName;

        var config = mockDbResponse.ImportPerformanceConfigs.FirstOrDefault(config => 

                    String.Equals(config.TableName, tableName, StringComparison.OrdinalIgnoreCase) 

                    && String.Equals(config.Stack, stack, StringComparison.OrdinalIgnoreCase));

        var saveDataMaxDegreeOfParallelism = GetSaveDataMaxDegreeOfParallelism(config, defaultSaveDataMaxDegreeOfParallelism);      

        var useDeadlockPreventionBatches = IsConfigValueTrue(config) && saveDataMaxDegreeOfParallelism > 1 ? true : false;

        var batches = useDeadlockPreventionBatches 

            ? CreateBatches(importData, "PEOPLE_ID", saveDataMaxDegreeOfParallelism)

            : new List<List<DataRow>> { importData };

​

        SaveDataProcessor(batches, saveDataMaxDegreeOfParallelism, importData.Count());

    }

    static List<List<DataRow>> CreateBatches(List<DataRow> importData, string identityField, int optimalBatchMultiplier)

    {

        var timer = new Stopwatch();

        timer.Start();

        var batches = new List<List<DataRow>>();

        var rowGroups = importData

            .GroupBy(row => row.Field<string>(identityField))

            .OrderByDescending(grp => grp.Count())

            .ToList();

        if (rowGroups.Count() == 0)

            return batches;

        var multiRowGroups = rowGroups.Where(grp => grp.Count() > 1).ToList();

        if (multiRowGroups.Count() == 0)

        {

            batches.Add(importData);

            return batches;

        }

        foreach (var row in multiRowGroups.First())

        {

            batches.Add(new List<DataRow>{ row });

        }           

​

​

        foreach (var rowGroup in multiRowGroups.Skip(1))

        {

            var rowsInGroup = rowGroup.ToList();

​

            for (var i = 0; i < rowsInGroup.Count(); i++)

            {

                batches[i].Add(rowsInGroup[i]);

            }

        }           

        var singleRows = rowGroups.Where(grp => grp.Count() == 1).Select(grp => grp.First()).ToList();

​

        var batchIndex = 0;

        var rowIndex = 0;

        var remainingRows = singleRows.Count();

​

        while (remainingRows != 0 && batchIndex < batches.Count)

        {

            var availableRowsInBatch = (optimalBatchMultiplier - batches[batchIndex].Count() % optimalBatchMultiplier);

​

            if (availableRowsInBatch != optimalBatchMultiplier)

            {

                var rowsToAdd = Math.Min(remainingRows, availableRowsInBatch);

                batches[batchIndex].AddRange(singleRows.Skip(rowIndex).Take(rowsToAdd));

                rowIndex += rowsToAdd;

                remainingRows = singleRows.Count() - rowIndex;

            }

​

            batchIndex++;

        }

​

        batches.Last().AddRange(singleRows.Skip(rowIndex));

        timer.Stop();

​

        var timeTaken = timer.Elapsed;

        Console.WriteLine("Time taken: " + timeTaken.ToString(@"m\:ss\.fff"));      

        return batches;

    }

    static void SaveDataProcessor(List<List<DataRow>> importBatches, int saveDataMaxDegreeOfParallelism, int initialCountForValidation)

    {

        //This is just to demonstrate how the records would be sent to save data

        //No changes required to actual functionality as it should handle the batches as is

        var totalRecords = 0;

        foreach(var batch in importBatches)

        {

            var rowIndex = 0;

            totalRecords += batch.Count();

            while (rowIndex < batch.Count())

            {

                var remainingRows = batch.Count() - rowIndex;

                var rowsToSend = Math.Min(saveDataMaxDegreeOfParallelism, remainingRows);

                Console.WriteLine("Sending " + rowsToSend + " rows to save data in parallel");

                Console.WriteLine();

                var rows = batch.Skip(rowIndex).Take(rowsToSend);

                foreach (var row in rows)

                {

                    Console.WriteLine(row["PEOPLE_ID"].ToString());

                }

                Console.WriteLine();

                rowIndex += rowsToSend;

            }

        }

        Console.WriteLine("Total records in initial entity: " + initialCountForValidation);

        Console.WriteLine("Total batches: " + importBatches.Count());

        Console.WriteLine("Total records in batches: " + totalRecords);     

    }

    static int GetSaveDataMaxDegreeOfParallelism(ImportPerformanceConfig? config, int defaultMaxValue)

    {

        //To keep this tidier we will use the IsPositiveValue extension in the setter when getting the default value from Orchestrator config,

        // therefore we will already know that the default is safe.

        var saveDataMaxDegreeOfParallelism = defaultMaxValue; 

​

        if (config != null)

            saveDataMaxDegreeOfParallelism = IsPositiveValue(config.SaveDataMaxDegreeOfParallelism) ? config.SaveDataMaxDegreeOfParallelism.Value : saveDataMaxDegreeOfParallelism;

        return saveDataMaxDegreeOfParallelism;

    }

    static bool IsConfigValueTrue(ImportPerformanceConfig? config) => config != null && (config.UseDeadlockPreventionBatches ?? false) == true;

    //Add as int extension method

    static bool IsPositiveValue(int? inputValue) => (inputValue ?? 0) >= 1; 

}

​

​

​

​