jdescopingtool/OLD/WorkerService/Process/UpdateProcessor.TableManagement.cs

using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Linq;
using System.Reflection;
using System.Text;
using Dapper;
using WorkerService.Models;

namespace WorkerService.Process
{
    /// <summary>
    /// Table management functionality for data update processor
    /// </summary>
    public partial class UpdateProcessor
    {
        /// <summary>
        /// Creates staging table with matching column layout of given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to create staging table for</param>
        /// <returns>Name of created temporary table</returns>
        public static string CreateStagingTable(SqlConnection connection, string tableName)
        {
            try
            {
                //Get table specification
                TableSpec tableSpec = GetTableSpec(connection, tableName);

                //Drop temp table if it already exists
                connection.Execute($"IF OBJECT_ID('tempdb..{tableSpec.StagingTableName}') IS NOT NULL DROP TABLE {tableSpec.StagingTableName};");

                //Create temp table
                connection.Execute($"CREATE TABLE {tableSpec.StagingTableName}({string.Join(",", tableSpec.Columns.Select(c => $"{c.Name} {c.Definition}"))});");

                //Create indicies on temp table
                StringBuilder builder = new StringBuilder();
                builder.Append($"CREATE INDEX IDX_STAGING_{tableSpec.Name} ON {tableSpec.StagingTableName}(");
                builder.Append(string.Join(",", tableSpec.PrimaryKey.Select(c => $"{c.Name}")));
                if (tableSpec.Columns.Any(c => c.Name.Equals("LastUpdateDT", StringComparison.CurrentCultureIgnoreCase)))
                {
                    builder.Append(", LastUpdateDT DESC");
                }
                else if (tableSpec.Columns.Any(c => c.Name.Equals("ReleaseDate", StringComparison.CurrentCultureIgnoreCase)))
                {
                    builder.Append(", ReleaseDate DESC");
                }
                builder.Append(");");
         
                connection.Execute(builder.ToString());

                //Disable indicies on temp table
                DisableIndicies(connection, tableSpec.StagingTableName);

                return tableSpec.StagingTableName;
            }
            catch (Exception error)
            {
                //Log and forward error
                logger.Error("GetStagingTable: failed to create staging table for '{0}': {1}.", tableName, error.Message);
                throw;
            }
        }

        /// <summary>
        /// Creates temporary table with matching column layout of given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to create temporary table for</param>
        /// <returns>Name of created temporary table</returns>
        public static string CreateTempTable(SqlConnection connection, string tableName)
        {
            try
            {
                //Get table specification
                TableSpec tableSpec = GetTableSpec(connection, tableName);

                //Drop temp table if it already exists
                connection.Execute($"IF OBJECT_ID('tempdb..{tableSpec.TempTableName}') IS NOT NULL DROP TABLE {tableSpec.TempTableName};");

                //Create temp table
                connection.Execute($"CREATE TABLE {tableSpec.TempTableName}({string.Join(",", tableSpec.Columns.Select(c => $"{c.Name} {c.Definition}"))}, CONSTRAINT PK_{tableSpec.TempTableName} PRIMARY KEY CLUSTERED({string.Join(",", tableSpec.PrimaryKey.Select(c => $"{c.Name}"))}));");

                StringBuilder builder = new StringBuilder();

                builder.AppendLine("WITH StagingCTE AS (");
                builder.Append($"SELECT st.*, ROW_NUMBER() OVER(PARTITION BY {string.Join(", ", tableSpec.PrimaryKey.Select(c=>c.Name))} ORDER BY {tableSpec.Columns.FirstOrDefault(c => c.Name.Equals("LastUpdateDT", StringComparison.CurrentCultureIgnoreCase) || c.Name.Equals("ReleaseDate", StringComparison.CurrentCultureIgnoreCase))?.Name}) RN FROM {tableSpec.StagingTableName} st");
                builder.AppendLine(")");
                builder.AppendLine($"INSERT INTO {tableSpec.TempTableName}({string.Join(", ", tableSpec.Columns.Select(c=>c.Name))})");
                builder.AppendLine($"SELECT {string.Join(", ", tableSpec.Columns.Select(c => c.Name))} FROM StagingCTE WHERE RN = 1 ORDER BY {string.Join(", ", tableSpec.PrimaryKey.Select(c=>c.Name))};");

                connection.Execute(builder.ToString(),commandTimeout:600);
               
                return tableSpec.TempTableName;
            }
            catch (Exception error)
            {
                //Log and forward error
                logger.Error("CreateTableTable: failed to create temporary table for '{0}': {1}.", tableName, error.Message);
                throw;
            }
        }

        /// <summary>
        /// Generates merge statement for given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to generate merge statement for</param>
        /// <returns>Merge statement for given table</returns>
        private static string GenerateMerge(SqlConnection connection, string tableName)
        {
            try
            {
                //Get table specification
                TableSpec tableSpec = GetTableSpec(connection, tableName);

                StringBuilder builder = new StringBuilder();

                builder.AppendFormat("MERGE {0} AS TARGET", tableSpec.Name);
                builder.AppendLine("");

                builder.AppendFormat("USING {0} AS SOURCE ON({1})", tableSpec.TempTableName, string.Join(" AND ", tableSpec.PrimaryKey.Select(c => $"TARGET.{c.Name} = SOURCE.{c.Name}")));
                builder.AppendLine("");

                builder.Append("WHEN MATCHED");
                if (tableSpec.Columns.Exists(c => c.Name.Equals("LastUpdateDT", StringComparison.CurrentCultureIgnoreCase)))
                {
                    builder.Append(" AND TARGET.LastUpdateDT < SOURCE.LastUpdateDT");
                }
                builder.AppendLine(" THEN");

                builder.Append("UPDATE SET ");
                builder.Append(string.Join(", ", tableSpec.Columns.Where(c => !tableSpec.PrimaryKey.Contains(c)).Select(c => $"TARGET.{c.Name} = SOURCE.{c.Name}")));
                builder.AppendLine();

                builder.AppendLine("WHEN NOT MATCHED BY TARGET THEN");

                builder.Append("INSERT(");
                builder.Append(string.Join(", ", tableSpec.Columns.Select(c => c.Name)));
                builder.AppendLine(")");

                builder.Append("VALUES(");
                builder.Append(string.Join(", ", tableSpec.Columns.Select(c => $"SOURCE.{c.Name}")));
                builder.AppendLine(");");

                return builder.ToString();
            }
            catch (Exception error)
            {
                //Log and forward error
                logger.Error("GenerateMerge: failed to generate merge statement for '{0}': {1}.", tableName, error.Message);
                throw;
            }
        }

        /// <summary>
        /// Generates bulk copy specification for given table
        /// </summary>
        /// <typeparam name="T">Data type being bulk copied</typeparam>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to generate bulk copy for</param>
        /// <returns>Bulk copy for given table</returns>
        private static SqlBulkCopy GenerateBulkCopy<T>(SqlConnection connection, string tableName)
        {
            return GenerateBulkCopy(connection, tableName, typeof(T));
        }

        /// <summary>
        /// Generates bulk copy specification for given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to generate bulk copy for</param>
        /// <param name="type">Data type being bulk copied</param>
        /// <returns>Bulk copy for given table</returns>
        private static SqlBulkCopy GenerateBulkCopy(SqlConnection connection, string tableName, Type type)
        {
            try
            {
                //Get table specification
                TableSpec tableSpec = GetTableSpec(connection, tableName);

                //Get class properties
                List<string> properties = type.GetProperties(BindingFlags.Instance | BindingFlags.Public).Select(p => p.Name).ToList();

                //Build bulk copy specification
                SqlBulkCopy bulkCopy = new SqlBulkCopy(connection)
                {
                    BatchSize = 10000,
                    NotifyAfter = 5000,
                    EnableStreaming = true,
                    DestinationTableName = tableSpec.StagingTableName
                };

                foreach (ColumnSpec columnSpec in tableSpec.Columns)
                {
                    string property = properties.FirstOrDefault(p => p.Equals(columnSpec.Name, StringComparison.CurrentCultureIgnoreCase));
                    if (!string.IsNullOrEmpty(property))
                    {
                        bulkCopy.ColumnMappings.Add(property, columnSpec.Name);
                    }
                }

                return bulkCopy;
            }
            catch (Exception error)
            {
                //Log and forward error
                logger.Error("GenerateBulkCopy: failed to generate bulk copy for '{0}': {1}.", tableName, error.Message);
                throw;
            }
        }

        /// <summary>
        /// SQL to get the columns for the given table
        /// </summary>
        private const string SQL_GET_TABLE_COLUMNS = @"
            SELECT  c.name AS Name,
                    CASE t2.name
                      WHEN 'char' THEN 'CHAR(' + CAST(c.max_length AS VARCHAR(10)) + ')'
                      WHEN 'varchar' THEN 'VARCHAR(' + CASE c.max_length WHEN -1 THEN 'MAX' ELSE CAST(c.max_length AS VARCHAR(10)) END + ')'
                      WHEN 'decimal' THEN 'DECIMAL(' + CAST(c.precision AS VARCHAR(4)) + ',' + CAST(c.scale AS VARCHAR(4)) + ')'
                      ELSE UPPER(t2.name)
                    END AS Definition
            FROM    sys.columns c INNER JOIN 
                    sys.types AS t2 ON (c.system_type_id = t2.system_type_id) INNER JOIN 
                    sys.tables t ON (c.object_id = t.object_id)
            WHERE   t.name = @tableName
            ORDER BY c.column_id";

        /// <summary>
        /// SQL to get the primary key columns for the given table
        /// </summary>
        private const string SQL_GET_TABLE_PRIMARY_KEY = @"
            SELECT  COLUMN_NAME AS Name
            FROM    INFORMATION_SCHEMA.KEY_COLUMN_USAGE
            WHERE   OBJECTPROPERTY(OBJECT_ID(CONSTRAINT_SCHEMA + '.' + QUOTENAME(CONSTRAINT_NAME)), 'IsPrimaryKey') = 1 AND
                    TABLE_NAME = @tableName
            ORDER BY ORDINAL_POSITION";

        /// <summary>
        /// Gets the table specification for the given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to get specification for</param>
        /// <returns>Table specification for the given table</returns>
        public static TableSpec GetTableSpec(SqlConnection connection, string tableName)
        {
            TableSpec tableSpec = new TableSpec() { Name = tableName };

            //Load columns
            tableSpec.Columns.AddRange(connection.Query<ColumnSpec>(SQL_GET_TABLE_COLUMNS, new { tableName }));

            //Load primary key
            tableSpec.PrimaryKey.AddRange(connection.Query<string>(SQL_GET_TABLE_PRIMARY_KEY, new { tableName }).Select(cn => tableSpec.Columns.First(c => c.Name.Equals(cn, StringComparison.CurrentCultureIgnoreCase))));

            return tableSpec;
        }

        /// <summary>
        /// Truncates given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to truncate</param>
        private static void TruncateTable(SqlConnection connection, string tableName)
        {
            try
            {
                logger.Debug("TruncateTable: truncating table '{0}'", tableName);

                //Generate and execute SQL to truncate table
                string sql = $"TRUNCATE TABLE {tableName};";
                connection.Execute(sql);
            }
            catch (Exception error)
            {
                //Log but do not forward error
                logger.Error("TruncateTable: failed to truncate table '{0}': {1}.", tableName, error.Message);
            }
        }

        /// <summary>
        /// SQL to get indices on given table
        /// </summary>
        private const string SQL_GET_INDICES = @"
            SELECT  DISTINCT
                    ind.name AS Name,
                    ind.is_primary_key AS IsPrimaryKey,
                    ind.is_unique AS IsUnique,
                    ind.is_unique_constraint AS IsUniqueConstraint
            FROM    sys.indexes ind INNER JOIN 
                    sys.index_columns ic ON (ind.object_id = ic.object_id AND ind.index_id = ic.index_id) INNER JOIN
                    sys.columns col ON (ic.object_id = col.object_id AND ic.column_id = col.column_id) INNER JOIN 
                    sys.tables t ON (ind.object_id = t.object_id)
            WHERE   t.name = @tableName";

        /// <summary>
        /// Disables all non-PK indices on given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>
        /// <param name="tableName">Name of table to disable non-PK indices for</param>
        private static void DisableIndicies(SqlConnection connection, string tableName)
        {
            try
            {
                //Get all indices on table
                List<Index> indices = connection.Query<Index>(SQL_GET_INDICES, new { tableName }).ToList();

                //Loop through all non-PK/non-cluster indices
                foreach (Index index in indices.Where(i => !i.IsPrimaryKey && !i.IsUnique && !i.IsUniqueConstraint))
                {
                    //Generate and execute SQL to disable index
                    string sql = $"ALTER INDEX {index.Name} ON {tableName} DISABLE;";
                    connection.Execute(sql);
                }
            }
            catch (Exception error)
            {
                //Log but do not forward error
                logger.Error("DisableIndicies: failed to disable non-PK indicies on table '{0}': {1}.", tableName, error.Message);
            }
        }

        /// <summary>
        /// Rebuilds all indices on given table
        /// </summary>
        /// <param name="connection">SQL connection to execute commands on</param>So
        /// <param name="tableName">Name of table to rebuild indices for</param>
        private static void RebuildIndicies(SqlConnection connection, string tableName)
        {
            try
            {
                //Get all indices on table
                List<Index> indices = connection.Query<Index>(SQL_GET_INDICES, new { tableName }).ToList();

                //Loop through indices
                foreach (Index index in indices.Where(i => !i.IsPrimaryKey && !i.IsUnique && !i.IsUniqueConstraint))
                {
                    //Generate and execute SQL to rebuild index
                    string sql = $"ALTER INDEX {index.Name} ON {tableName} REBUILD;";
                    connection.Execute(sql);
                }
            }
            catch (Exception error)
            {
                //Log but do not forward error
                logger.Error("RebuildIndicies: failed to rebuild indicies on table '{0}': {1}.", tableName, error.Message);
            }
        }
    }
}