Aggregate Functions to Validate Data for SQL Server Replication and ETL Processes

Function Get-CheckSumColumn {
    Param(
    [ValidateLength(7,15)][string]$readserver
    , [ValidateLength(50,4000)][string]$readquery
    , [ValidateRange(0,1)][int]$options
    )
    Process
    {
        $scon = New-Object System.Data.SqlClient.SqlConnection
        $scon.ConnectionString = "Data Source=$readserver;Initial Catalog=master;Integrated Security=true;"

        $cmd = New-Object System.Data.SqlClient.SqlCommand
        $cmd.Connection = $scon
        $cmd.CommandText = $readquery
        $cmd.CommandTimeout = 10
        
        try
        {
            $scon.Open()
            $sqlread = $cmd.ExecuteReader()
    
            while ($sqlread.Read())
            {
                if ($options -eq 0)
                {
                    $checksum_column = $sqlread["COLUMN_NAME"]
                }
                elseif ($options -eq 1)
                {
                    $checksum_column = $sqlread["CheckSumValue"]
                }
            }
        }
        catch [Exception]
        {
            Write-Warning "Get-CheckSumColumn ($options)"
            Write-Warning $_.Exception.Message
        }
        finally
        {
            $cmd.Dispose()
            $scon.Dispose()
        }

        return $checksum_column
    }
}

Function Get-CheckSumQueries {
    Param(
    [ValidateLength(3,75)][string]$readserver
    , [ValidateLength(3,75)][string]$distributiondb
    , [ValidateLength(3,100)][string]$article
    , [ValidateLength(3,100)][string]$source_database
    , [ValidateLength(3,100)][string]$destination_server
    )
    Process
    {
        $checksumquery = "
        SELECT DISTINCT
	        DENSE_RANK() OVER (ORDER BY (dist.publisher + ' ' + s.publisher_db + ' ' + a.article + ' ' + ss.srvname + ' ' + s.subscriber_db + ' ' + a.destination_object)) ID
	        , dist.publisher SourceServer
	        , s.publisher_db SourceDatabase
	        , ss.srvname DestinationServer
	        , s.subscriber_db DestinationDatabase
            , 'SELECT TOP 1 COLUMN_NAME FROM ' + s.publisher_db + '.INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = ''' +  a.article + ''' AND DATA_TYPE IN (''numeric'',''int'',''decimal'',''float'')' AS CheckSumQuerySource
	        , 'SELECT TOP 1 COLUMN_NAME FROM ' + s.subscriber_db + '.INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = ''' +  a.article + ''' AND DATA_TYPE IN (''numeric'',''int'',''decimal'',''float'')' AS CheckSumQueryDestination
        FROM MSArticles a  
	        INNER JOIN (
		        SELECT DISTINCT job_id
			        , publisher
			        , publication_id
			        , publisher_db
		        FROM MSreplication_monitordata
		        ) AS dist ON dist.publication_id = a.publication_id
			        AND dist.publisher_db = a.publisher_db
	        INNER JOIN MSpublications p ON a.publication_id = p.publication_id
		        AND p.publisher_db = a.publisher_db
		        AND a.publisher_id = p.publisher_id
	        INNER JOIN MSsubscriptions s ON s.publisher_db = dist.publisher_db
	        INNER JOIN master..sysservers ss ON s.subscriber_id = ss.srvid
	        INNER JOIN master..sysservers srv ON srv.srvid = s.publisher_id
	        INNER JOIN MSdistribution_agents da ON da.publisher_id = s.publisher_id  
			        AND da.subscriber_id = s.subscriber_id
			        AND da.publisher_db = dist.publisher_db
			        AND da.job_id = dist.job_id
        WHERE p.publication_type = 0
	        AND a.article = '$article'
	        AND s.publisher_db = '$source_database'
	        AND ss.srvname = '$destination_server'
        "
        
        $scon = New-Object System.Data.SqlClient.SqlConnection
        $scon.ConnectionString = "Data Source=$readserver;Initial Catalog=$distributiondb;Integrated Security=true;Connection Timeout=0;"

        $cmd = New-Object System.Data.SqlClient.SqlCommand
        $cmd.Connection = $scon
        $cmd.CommandText = $checksumquery
        $cmd.CommandTimeout = 0
        
        try
        {
            $scon.Open()
            $sqlread = $cmd.ExecuteReader()
    
            while ($sqlread.Read())
            {
                $checksum_source = $sqlread["CheckSumQuerySource"]
                $checksum_destination = $sqlread["CheckSumQueryDestination"]

                $checksum_sourceserver = $sqlread["SourceServer"]
                $checksum_destinationserver = $sqlread["DestinationServer"]
                $checksum_sourcedatabase = $sqlread["SourceDatabase"]
                $checksum_destinationdatabase = $sqlread["DestinationDatabase"]

                $checksum_sret =  Get-CheckSumColumn -readserver $checksum_sourceserver -readquery $checksum_source -options 0
                $checksum_dret = Get-CheckSumColumn -readserver $checksum_destinationserver -readquery $checksum_destination -options 0

                $sourcecheck = "SELECT SUM($checksum_sret) AS CheckSumValue FROM $checksum_sourcedatabase.dbo.$article WITH(NOLOCK)"
                $destinationcheck = "SELECT SUM($checksum_dret) AS CheckSumValue FROM $checksum_destinationdatabase.dbo.$article WITH(NOLOCK)"

                $checksum_svals =  Get-CheckSumColumn -readserver $checksum_sourceserver -readquery $sourcecheck -options 1
                $checksum_dvals = Get-CheckSumColumn -readserver $checksum_destinationserver -readquery $destinationcheck -options 1

                Write-Host $checksum_svals
                Write-Host $checksum_dvals
            }
        }
        catch [Exception]
        {
            Write-Warning "Get-CheckSumQueries"
            Write-Warning $_.Exception.Message
        }
        finally
        {
            $cmd.Dispose()
            $scon.Dispose()
        }
    }
}

Get-CheckSumQueries -readserver "" -distributiondb "" -article "" -source_database "" -destination_server ""