• • •

Transactional to Business Intelligence

The Problem: Reduced Maintenance Windows

All developers of Transactional(OLTP) systems have had to carry out certain processes for the feed of Business Intelligence(BI) databases. This type of process can be more or less complicated depending on the manager of the OLTP system and the manager of the business analysis system. Let me explain: Normally, the necessary information in both types of system is not usually the same, so, either the OLTP team or the business intelligence team, they must analyze, select and transfer the information available from one system to another. And depending on the managers, this task will be done by one or other equipment.

In the event that this task must be carried out by the OLTP team, normally, there is usually a additional requirement: efficiency. This is usually due to the fact that OLTP systems are usually very busy during the day with online work, and at night with data uploads and reporting varied. Thus, the time window given for this type of process is usually very small, so it is necessary to optimizing the entire transfer process as much as possible.

Here we will describe a work methodology using the Insert Bulk Bestia libraries and stored procedures in such a way that the analysis, selection and transmission of information from one system to another is minimized.

The code shown in this chapter is presented as an example of the use of the SQLServerAsyncExecProcedures library included in Insert Bulk Bestia.
MalaCharger is not responsible for the use of such code.

Problem Statement

You want to perform a daily process to send certain information from one OLTP system to another system for BI (both are on different hosts). Said information must be marked indicating for each record if it is new, has been modified or deleted since the last time the process was executed.

To do this, a series of repository tables will be created in both systems. In BI, a table called "ACTUALIZACIONES" will be created, where records inserted, modified, or deleted will be logged for each execution of the process. BI will check this table and must process the information indicated there.

Note: All created objects will be made in the "rca" scheme.

It is also based on the hypothesis that Insert Bulk Bestia libraries are installed in the OLTP and not in the BI system.

Repository in OLTP

In the OLTP system's database, a repository of tables will be created with the information to be transmitted (rca.Table1, rca.Table2, ...), and associated with each table, there will be another table where a pair of HASHes of the information will be stored (rca.Table1_HASH, rca.Table2_HASH, ...). These two HASHes will facilitate and speed up the process of comparing information compared to using queries that compare field by field, to which additional queries must be added to determine which records are new and which are deleted.

In addition to the repository, a Sequence used to name the successive executions and a HASH calculation function will be included in the code for those versions of SQL Server (2014 and lower) whose HASHBYTES function is limited to 4000 characters (NVARCHAR) . Let's take a look at it in code (for brevity, only a couple of example tables will be included):

 IF EXISTS(select NULL from sys.sequences where object_id = object_id('rca.SEQ_PARALELO'))
  DROP SEQUENCE rca.SEQ_PARALELO
GO
CREATE SEQUENCE rca.SEQ_PARALELO AS [bigint]
  START WITH 1
  INCREMENT BY 1
  MINVALUE 1
  MAXVALUE 9223372036854775807
  CACHE 10 
GO

--SQL Server 2014 and earlier versions:
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.fnHashBytes') IS NOT NULL
  DROP FUNCTION rca.fnHashBytes
GO
CREATE FUNCTION rca.fnHashBytes (
  @EncryptAlgorithm NVARCHAR(50),
  @DataToEncrypt nvarchar(max))
RETURNS VARBINARY(MAX) WITH SCHEMABINDING
AS BEGIN
  DECLARE @Index INT = 1, @DataToEncryptLength INT
  DECLARE @EncryptedResult varbinary(max) = NULL

  IF @DataToEncrypt IS NOT NULL
  BEGIN
    SELECT @EncryptedResult = 0x, @DataToEncryptLength = DATALENGTH(@DataToEncrypt)
    WHILE @Index <= @DataToEncryptLength
    BEGIN
      SELECT @EncryptedResult = @EncryptedResult + HASHBYTES(@EncryptAlgorithm,
             SUBSTRING(@DataToEncrypt, @Index, 4000)), @Index = @Index + 4000
    END
  END
  RETURN @EncryptedResult
END
GO

--As can be seen, the "rca" schema is used to define the objects used for the repository.

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Tabla1') IS NOT NULL
  DROP TABLE rca.Tabla1
GO
CREATE TABLE rca.Tabla1(
    Campo1 varchar(16) NOT NULL,
    Campo2 smallint NOT NULL,
    Campo3 varchar(55) NOT NULL,

    ACT_CD_TIPO smallint NOT NULL,

  CONSTRAINT Tabla1_PK PRIMARY KEY CLUSTERED
    (
     Campo1 ASC,
     Campo2 ASC,
     ACT_CD_TIPO ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Tabla1_HASH') IS NOT NULL
  DROP TABLE rca.Tabla1_HASH
GO
CREATE TABLE rca.Tabla1_HASH (
    Campo1 varchar(16) NOT NULL,
    Campo2 smallint NOT NULL,

    HASH_OLD varbinary(max) NULL,
    HASH_NEW varbinary(max) NULL,

 CONSTRAINT Tabla1_HASH_PK PRIMARY KEY CLUSTERED
    (
     Campo1 ASC,
     Campo2 ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Tabla2') IS NOT NULL
  DROP TABLE rca.Tabla2
GO
CREATE TABLE rca.Tabla2(
    Campo1 varchar(16) NOT NULL,
    Campo2 smallint NOT NULL,
    Campo3 varchar(55) NOT NULL,

    ACT_CD_TIPO smallint NOT NULL,

  CONSTRAINT Tabla2_PK PRIMARY KEY CLUSTERED
    (
     Campo1 ASC,
     ACT_CD_TIPO ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Tabla2_HASH') IS NOT NULL
  DROP TABLE rca.Tabla2_HASH
GO
CREATE TABLE rca.Tabla2_HASH (
    Campo1 varchar(16) NOT NULL,

    HASH_OLD varbinary(max) NULL,
    HASH_NEW varbinary(max) NULL,

 CONSTRAINT Tabla2_HASH_PK PRIMARY KEY CLUSTERED
    (
     Campo1 ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

It is assumed that both "Table1" and "Table2" only contain the fields to be transferred. The information will come from the tables "Table1Org" and "Table2Org" respectively, which will contain much more information, but which is not necessary for BI. "Table1" has a primary key that is made up of two fields and "Table2" has a single field.

Both tables contain the ACT_CD_TIPO field, which is part of the primary key. During information processing, records will be duplicated, and this field will be used to distinguish new records from old ones.

The HASH_OLD field will contain the HASH of the record of the last execution of the process, while the HASH_NEW field will contain the HASH of the record of the current execution.
If HASH_OLD is NULL, it means that the record is new, if HAS_NEW is NULL, it means that the record has been removed from the original table.
And if both fields are not NULL and do not match, it means that the record has been modified.

Stored Procedures in OLTP

In the OLTP, a series of stored procedures will be created to update the repository tables that have been created.

It should be noted that no Foreign Key has been created in the repository tables. However, if there were any, before updating the tables they would have to be removed. The update of the repository tables will be done in parallel, so there will be no pre-established order. Thus, first of all, two stored procedures will be created to eliminate and recreate these Foreign Keys. Here is an approach to these stored procedures:

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.EliminaFkLocales') IS NOT NULL
  DROP PROCEDURE rca.EliminaFkLocales
GO
CREATE PROCEDURE rca.EliminaFkLocales
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(),
          @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY

    --Drop all FKs from repository tables for loading
    IF EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'Tabla1_Tabla2_FK')
      ALTER TABLE rca.Tabla1 DROP CONSTRAINT Tabla1_Tabla2_FK

    EXECUTE ibb.Log_Errores_Inserta N'Local Foreign Keys Deleted.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento

  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR deleting the Local Foreign Keys.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.CreaFkLocales') IS NOT NULL
  DROP PROCEDURE rca.CreaFkLocales
GO
CREATE PROCEDURE rca.CreaFkLocales
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(),
          @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY

    --Create all necessary FKs
    --In our example, there are no FKs between tables in the repository
    --IF NOT EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'Tabla1_Tabla2_FK')
    --  ALTER TABLE rca.Tabla1 ADD CONSTRAINT Tabla1_Tabla2_FK FOREIGN KEY(Campo1)
    --    REFERENCES rca.Tabla2 (Campo1)

    EXECUTE ibb.Log_Errores_Inserta N'Local Foreign Keys Created.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento

  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR creating the Local Foreign Keys.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

This is followed by the stored procedures responsible for updating the repository tables. The ones shown here are in charge of updating the contents of the tables "rca.Tabla1", "rca.Tabla1_HASH", "rca.Tabla2" and "rca.Tabla2_HASH". The code comments are descriptive enough.

These stored procedures will run in parallel. Therefore, the following instructions are important:

It is desired that, if an error occurs in them, it does not continue the execution (SET XACT_ABORT ON).
The goal of parallel execution is to make the most of all available CPUs. Therefore, you do not want Querys to be executed in parallel (OPTION (MAXDOP 1)). MERGE is usually a candidate statement to be executed in parallel, so it is very important to tell it not to do it.

Algorithm:

Update ACT_CD_TIPO field to 2 for all records in repository table (Tabla1).
Insert all the records from the original OLTP table (Table1Org) in the repository table (Table1). The ACT_CD_TIPO field is set to 0 to indicate that they are the inserted records.
This Query is where the information from the OLTP is selected and transformed appropriately to meet BI needs.
Delete old records:
- Table1 contains the records of the previous execution marked with ACT_CD_TIPO at 2 and ALL the OLTP records marked with ACT_CD_TIPO at 0.
- The DELETE will delete the records that already existed in Table1 and that continue to exist in the OLTP. For this, the key from Table1 is used.
- In this way, the records with ACT_CD_TIPO = 2 will be the records that have disappeared from the OLTP, and the records with ACT_CD_TIPO = 0 will contain the existing information in the OLTP.
Table1_HASH update:
- The HASH of each record is calculated and compared with the HASH contained in the HASH_NEW field.
- If the record exists in both tables (Table1 and Table1_HASH), but the calculated HASH does not match the HASH_NEW, the HASH_NEW field is updated with the calculated HASH. Those will be the records whose information has been modified in the OLTP. For these records, HASH_OLD and HASH_NEW will not match. We will see why later.
- If the record does not exist in Table1_HASH, it is inserted into that table. It will be the new records. These records will have the HAS_OLD field set to NULL.
- If the record does not exist in Table1 but does exist in Table1_HASH, in Table1_HASH the HASH_NEW field is updated to NULL. These records will be deleted and the HASH_OLD field will not be NULL.

 --NOTE: rca.Table1 has 'Campo1' and 'Campo2' as PRIMARY KEY

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.RellenaTabla1') IS NOT NULL
  DROP PROCEDURE rca.RellenaTabla1
GO
CREATE PROCEDURE rca.RellenaTabla1
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON

--This stored procedure will run in PARALLEL,
--so if an error occurs, it should not continue with the execution.
SET XACT_ABORT ON

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY
    --Mark all records as 2
    UPDATE rca.Tabla1
      SET ACT_CD_TIPO = 2
      OPTION(MAXDOP 1) --As it will run in parallel, we should not block the CPUs

    --Insert all records from the Original Table marked as 0
    INSERT INTO rca.Tabla1
      SELECT Tb1Org.Campo1, Tb1Org.Campo2, Tb1Org.Campo3, 0 AS ACT_CD_TIPO
        FROM dbo.Tabla1Org Tb1Org
      OPTION(MAXDOP 1) --As it will run in parallel, we should not block the CPUs

    --Delete old records:
    --  If record1 exists in both Table1 and TableOrg1, after the previous INSERT, Table1
    --  will contain: record1-2 (from the UPDATE) and record1-0 (from the INSERT). This DELETE
    --  will remove record1-2, leaving the new record.
    DELETE Tb1Del
      FROM rca.Tabla1 Tb1Del
        INNER JOIN rca.Tabla1 Tb1Ins
          ON Tb1Del.Campo1 = Tb1Ins.Campo1 AND Tb1Del.Campo2 = Tb1Ins.Campo2
      WHERE Tb1Del.ACT_CD_TIPO = 2 AND Tb1Ins.ACT_CD_TIPO = 0
      OPTION(MAXDOP 1) --As it will run in parallel, we should not block the CPUs

    --Update of Table1_HASH
    --  The HASH of each record is calculated and compared with the HASH_NEW field
    --    If the record exists in both tables but the HASH does not match, the HASH_NEW field is updated
    --    If the record does not exist in Table1_HASH, it is inserted
    --    If the record does not exist in Table1, the HASH_NEW field is nullified
    MERGE rca.Tabla1_HASH TablaHash USING
          (SELECT MBT.Campo1, MBT.Campo2,
                --SQL Server 2014 and earlier versions:
                --  Here the SQL Server HASHBYTES function is being used, because it is assumed that
                --  the character string being formed will not exceed 4000 characters. If it exceeds this length,
                --  the function rca.fnHashBytes should be used
              HASHBYTES('SHA2_512', (select MBT.* from (values(null))foo(bar) for xml auto)) as HASH_NEW
            FROM rca.Tabla1 AS MBT
            WHERE MBT.ACT_CD_TIPO = 0
          ) AS Hases_Nuevos
            ON TablaHash.Campo1 = Hases_Nuevos.Campo1 AND TablaHash.Campo2 = Hases_Nuevos.Campo2
      WHEN MATCHED AND TablaHash.HASH_NEW <> Hases_Nuevos.HASH_NEW THEN --Modified
        UPDATE SET TablaHash.HASH_NEW = Hases_Nuevos.HASH_NEW
      WHEN NOT MATCHED BY TARGET THEN --New
        INSERT(Campo1, Campo2, HASH_NEW)
        VALUES(Hases_Nuevos.Campo1, Hases_Nuevos.Campo2, Hases_Nuevos.HASH_NEW)
      WHEN NOT MATCHED BY SOURCE THEN --Deleted
        UPDATE SET HASH_NEW = NULL
      OPTION(MAXDOP 1); --As it will run in parallel, we should not block the CPUs

      --MERGE is a statement that could be parallelized by SQL Server. Since many things are already being done
      --in parallel, we specify not to parallelize it to avoid blocking the CPUs

    --It might be advisable to update the indexes and statistics if there are many modifications
    ALTER INDEX ALL ON rca.Tabla1_HASH REBUILD WITH (MAXDOP = 1, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.Tabla1_HASH WITH COLUMNS
    ALTER INDEX ALL ON rca.Tabla1 REBUILD WITH (MAXDOP = 1, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.Tabla1 WITH COLUMNS

    EXECUTE ibb.Log_Errores_Inserta N'Table1 and Table1_HASH have been updated.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento

  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR updating Table1 and Table1_HASH.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO


--NOTE: rca.Table2 has 'Campo1' as PRIMARY KEY

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.RellenaTabla2') IS NOT NULL
  DROP PROCEDURE rca.RellenaTabla2
GO
CREATE PROCEDURE rca.RellenaTabla2
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON

--This stored procedure will run in PARALLEL,
--so if an error occurs, it should not continue with the execution.
SET XACT_ABORT ON

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY
    --Mark all records as 2
    UPDATE rca.Tabla2
      SET ACT_CD_TIPO = 2
      OPTION(MAXDOP 1) --As it will run in parallel, we should not block the CPUs

    --Insert all records from the Original Table marked as 0
    INSERT INTO rca.Tabla2
      SELECT Tb2Org.Campo1, Tb2Org.Campo2, Tb2Org.Campo3, 0 AS ACT_CD_TIPO
        FROM dbo.Tabla2Org Tb2Org
      OPTION(MAXDOP 1) --As it will run in parallel, we should not block the CPUs

    --Delete old records:
    --  If record1 exists in both Table2 and Table2Org, after the previous INSERT, Table1
    --  will contain: record1-2 (from the UPDATE) and record1-0 (from the INSERT). This DELETE
    --  will remove record1-2, leaving the new record.
    DELETE Tb2Del
      FROM rca.Tabla2 Tb2Del
        INNER JOIN rca.Tabla2 Tb2Ins
          ON Tb2Del.Campo1 = Tb2Ins.Campo1
      WHERE Tb2Del.ACT_CD_TIPO = 2 AND Tb2Ins.ACT_CD_TIPO = 0
      OPTION(MAXDOP 1) --As it will run in parallel, we should not block the CPUs

    --Update of Table2_HASH
    --  The HASH of each record is calculated and compared with the HASH_NEW field
    --    If the record exists in both tables but the HASH does not match, the HASH_NEW field is updated
    --    If the record does not exist in Table2_HASH, it is inserted
    --    If the record does not exist in Table2, the HASH_NEW field is nullified
    MERGE rca.Tabla2_HASH TablaHash USING
          (SELECT MBT.Campo1,
                --SQL Server 2014 and earlier versions:
                --  Here the SQL Server HASHBYTES function is being used, because it is assumed that
                --  the character string being formed will not exceed 4000 characters. If it exceeds this length,
                --  the function rca.fnHashBytes should be used
              HASHBYTES('SHA2_512', (select MBT.* from (values(null))foo(bar) for xml auto)) as HASH_NEW
            FROM rca.Tabla2 AS MBT
            WHERE MBT.ACT_CD_TIPO = 0
          ) AS Hases_Nuevos
            ON TablaHash.Campo1 = Hases_Nuevos.Campo1
      WHEN MATCHED AND TablaHash.HASH_NEW <> Hases_Nuevos.HASH_NEW THEN --Modified
        UPDATE SET TablaHash.HASH_NEW = Hases_Nuevos.HASH_NEW
      WHEN NOT MATCHED BY TARGET THEN --New
        INSERT(Campo1, HASH_NEW)
        VALUES(Hases_Nuevos.Campo1, Hases_Nuevos.HASH_NEW)
      WHEN NOT MATCHED BY SOURCE THEN --Deleted
        UPDATE SET HASH_NEW = NULL
      OPTION(MAXDOP 1); --As it will run in parallel, we should not block the CPUs
        --MERGE is a statement that could be parallelized by SQL Server. Since many things are already being done
        --in parallel, we specify not to parallelize it to avoid blocking the CPUs

    --It might be advisable to update the indexes and statistics if there are many modifications
    ALTER INDEX ALL ON rca.Tabla2_HASH REBUILD WITH(MAXDOP = 1, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.Tabla2_HASH WITH COLUMNS
    ALTER INDEX ALL ON rca.Tabla2 REBUILD WITH (MAXDOP = 1, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.Tabla2 WITH COLUMNS

    EXECUTE ibb.Log_Errores_Inserta N'Table2 and Table2_HASH have been updated.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento

  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR updating Table2 and Table2_HASH.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

Repository in BI

Just as a series of tables have been created in the OLTP, their replicas will be created in BI according to the requirements outlined at the beginning.

The replicas of "Table1" and "Table2" are shown below. A new field has been added, ACT_CD_ACTUALIZACION, which corresponds to the key of the table rca.UPDATES. Both tables contain the ACT_CD_TIPO field that will indicate what has happened to the record since the last time the process was run. It will contain the values: 0-Insert, 1-Update, 2-Delete.

The rca.UPDATES table will contain a record for each time the data transfer process has been attempted. If the process was successful, the ACT_FH_FIN field will have a value, and if there have been errors, this field will contain a NULL.

As in OLTP, Foreign Keys have not been defined between the repository tables, however, the FKs that relate these tables to the rca table have been defined. UPDATES.

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Tabla1') IS NOT NULL
  DROP TABLE rca.Tabla1
GO
CREATE TABLE rca.Tabla1(
    Campo1 varchar(16) NOT NULL,
    Campo2 smallint NOT NULL,
    Campo3 varchar(55) NOT NULL,

    ACT_CD_TIPO smallint NOT NULL,
    ACT_CD_ACTUALIZACION bigint NOT NULL,

   CONSTRAINT Tabla1_PK PRIMARY KEY CLUSTERED
    (
     Campo1 ASC,
     Campo2 ASC,
     ACT_CD_ACTUALIZACION ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Tabla2') IS NOT NULL
  DROP TABLE rca.Tabla2
GO
CREATE TABLE rca.Tabla2(
    Campo1 varchar(16) NOT NULL,
    Campo2 smallint NOT NULL,
    Campo3 varchar(55) NOT NULL,

    ACT_CD_TIPO smallint NOT NULL,
    ACT_CD_ACTUALIZACION bigint NOT NULL,

   CONSTRAINT Tabla2_PK PRIMARY KEY CLUSTERED
    (
     Campo1 ASC,
     ACT_CD_ACTUALIZACION ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.ACTUALIZACIONES') IS NOT NULL
  DROP TABLE rca.ACTUALIZACIONES
GO
CREATE TABLE rca.ACTUALIZACIONES(
  ACT_CD_ACTUALIZACION bigint NOT NULL,
  ACT_FH_INICIO datetime2(0) NOT NULL,
  ACT_FH_FIN datetime2(0) NULL,
  ACT_DURACION nvarchar(50) NULL,

  CONSTRAINT ACTUALIZACIONES_PK PRIMARY KEY CLUSTERED 
    (
        ACT_CD_ACTUALIZACION ASC
    )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
GO

CREATE NONCLUSTERED INDEX RCA_ACTUALIZACIONES_FIN_NULO_IDX ON
rca.ACTUALIZACIONES
(
  ACT_FH_FIN ASC
)
INCLUDE(ACT_CD_ACTUALIZACION)
WITH (PAD_INDEX = ON, STATISTICS_NORECOMPUTE = OFF, SORT_IN_TEMPDB = OFF, DROP_EXISTING = OFF, ONLINE = OFF,
      ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 100) ON [PRIMARY]
GO


ALTER TABLE rca.Tabla1 ADD CONSTRAINT ACTUALIZACIONES_Tabla1_FK FOREIGN KEY(ACT_CD_ACTUALIZACION)
  REFERENCES rca.ACTUALIZACIONES (ACT_CD_ACTUALIZACION) ON DELETE CASCADE
GO
ALTER TABLE rca.Tabla2 ADD CONSTRAINT ACTUALIZACIONES_Tabla2_FK FOREIGN KEY(ACT_CD_ACTUALIZACION)
  REFERENCES rca.ACTUALIZACIONES (ACT_CD_ACTUALIZACION) ON DELETE CASCADE
GO

Stored Procedures in BI

It will also be necessary to create certain stored procedures on the BI side:

Cleaning of the repository tables and of the rca.UPDATES table of possible previous executions that did not finish correctly: (ACT_FH_FIN IS NULL, or ACT_CD_ACTUALIZACION does not exist in rca.UPDATES). It also inserts a new record for the current execution in rca.UPDATES.
A stored procedure is also created to update the end date of the process, also indicating that the process has run successfully.

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.LimpiaTablasRemotas') IS NOT NULL
  DROP PROCEDURE rca.LimpiaTablasRemotas
GO
CREATE PROCEDURE rca.LimpiaTablasRemotas
  @fechaInicioProceso NVARCHAR(50),
  @IdParalelo BIGINT,
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON

  SET @error = 0
  BEGIN TRY

    --Delete from Table1 and Table2 any records that may have been inserted from a previous process
    --that may have failed (ACT_FH_FIN IS NULL or ACT_CD_ACTUALIZACION does not exist in rca.ACTUALIZACIONES)
    DELETE rca.Tabla1
      WHERE ACT_CD_ACTUALIZACION IN (SELECT ACT_CD_ACTUALIZACION FROM rca.ACTUALIZACIONES WHERE ACT_FH_FIN IS NULL)
        OR ACT_CD_ACTUALIZACION NOT IN (SELECT ACT_CD_ACTUALIZACION FROM rca.ACTUALIZACIONES)

    DELETE rca.Tabla2
      WHERE ACT_CD_ACTUALIZACION IN (SELECT ACT_CD_ACTUALIZACION FROM rca.ACTUALIZACIONES WHERE ACT_FH_FIN IS NULL)
        OR ACT_CD_ACTUALIZACION NOT IN (SELECT ACT_CD_ACTUALIZACION FROM rca.ACTUALIZACIONES)

    --There is a DELETE CASCADE, but for performance reasons the previous queries are performed.
    DELETE rca.ACTUALIZACIONES
      WHERE ACT_FH_FIN IS NULL

    --The new record is created for the current execution
    INSERT INTO rca.ACTUALIZACIONES (ACT_CD_ACTUALIZACION, ACT_FH_INICIO)
      SELECT @IdParalelo, TRY_CONVERT(datetime2(0), @fechaInicioProceso, 121)

    --For performance reasons, it may be convenient to update the indices and statistics.
    ALTER INDEX ALL ON rca.ACTUALIZACIONES REBUILD WITH (MAXDOP = 0, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.ACTUALIZACIONES WITH COLUMNS
  END TRY
  BEGIN CATCH
    --It would be convenient to create the entire ibb.Log_Errores structure in the BI system and leave the corresponding traces
    SET @error = 1
  END CATCH
END
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.UpdateFinActualizacion') IS NOT NULL
  DROP PROCEDURE rca.UpdateFinActualizacion
GO
CREATE PROCEDURE rca.UpdateFinActualizacion
  @IdParalelo bigint,
  @Duracion nvarchar(50),
  @fFin nvarchar(50),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET REMOTE_PROC_TRANSACTIONS ON
SET XACT_ABORT ON

  SET @error = 0
  BEGIN TRY
    UPDATE rca.ACTUALIZACIONES
        SET ACT_FH_FIN = TRY_CONVERT(datetime2(0), @fFin, 121), ACT_DURACION = @Duracion
      WHERE ACT_CD_ACTUALIZACION = @IdParalelo
  END TRY
  BEGIN CATCH
    --It would be convenient to create the entire ibb.Log_Errores structure in the BI system and leave the corresponding traces
    SET @error = 1
  END CATCH
END
GO

Second, the stored procedures to delete and create the Foreign Keys are created on the remote server. Even if FKs are not defined between the repository tables, FKs have been defined between these tables and the rca.ACTUALIZACIONES table.

A stored procedure has also been created for the regeneration of the indexes and statistics of the BI tables.

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.EliminaFkRemotas') IS NOT NULL
  DROP PROCEDURE rca.EliminaFkRemotas
GO
CREATE PROCEDURE rca.EliminaFkRemotas
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  SET @error = 0
  BEGIN TRY
    --Remove all FKs from repository tables for loading
    IF EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'Tabla1_Tabla2_FK')
      ALTER TABLE rca.Tabla1 DROP CONSTRAINT Tabla1_Tabla2_FK

    --Remove all FKs against the rca.ACTUALIZACIONES table.
    IF EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'ACTUALIZACIONES_Tabla1_FK')
      ALTER TABLE rca.Tabla1 DROP CONSTRAINT ACTUALIZACIONES_Tabla1_FK

    IF EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'ACTUALIZACIONES_Tabla2_FK')
      ALTER TABLE rca.Tabla2 DROP CONSTRAINT ACTUALIZACIONES_Tabla2_FK
  END TRY
  BEGIN CATCH
    --It would be convenient to create the entire ibb.Log_Errores structure in the BI system and leave the corresponding traces
    SET @error = 1
  END CATCH
END
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.CreaFkRemotas') IS NOT NULL
  DROP PROCEDURE rca.CreaFkRemotas
GO
CREATE PROCEDURE rca.CreaFkRemotas
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  SET @error = 0
  BEGIN TRY

    --Create all the necessary FKs
    --In the example, there are no FKs between the repository tables
    --IF NOT EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'Tabla1_Tabla2_FK')
    --ALTER TABLE rca.Tabla1 ADD CONSTRAINT Tabla1_Tabla2_FK FOREIGN KEY(Campo1)
    --  REFERENCES rca.Tabla2 (Campo1)

    --Create all the FKs against the rca.ACTUALIZACIONES table
    IF NOT EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'ACTUALIZACIONES_Tabla1_FK')
      ALTER TABLE rca.Tabla1 ADD CONSTRAINT ACTUALIZACIONES_Tabla1_FK FOREIGN KEY(ACT_CD_ACTUALIZACION)
      REFERENCES rca.ACTUALIZACIONES (ACT_CD_ACTUALIZACION) ON DELETE CASCADE

    IF NOT EXISTS(SELECT NULL FROM sys.foreign_keys WHERE name = 'ACTUALIZACIONES_Tabla2_FK')
      ALTER TABLE rca.Tabla2 ADD CONSTRAINT ACTUALIZACIONES_Tabla2_FK FOREIGN KEY(ACT_CD_ACTUALIZACION)
      REFERENCES rca.ACTUALIZACIONES (ACT_CD_ACTUALIZACION) ON DELETE CASCADE
  END TRY
  BEGIN CATCH
    --It would be convenient to create the entire ibb.Log_Errores structure in the BI system and leave the corresponding traces
    SET @error = 1
  END CATCH
END
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.RegeneraIndicesRemotos') IS NOT NULL
  DROP PROCEDURE rca.RegeneraIndicesRemotos
GO
CREATE PROCEDURE rca.RegeneraIndicesRemotos
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  SET @error = 0
  BEGIN TRY
    --If there are many updates in the repository, it may be advisable to update
    --the indexes and statistics
    ALTER INDEX ALL ON rca.Tabla1 REBUILD WITH (MAXDOP = 0, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.Tabla1 WITH COLUMNS

    ALTER INDEX ALL ON rca.Tabla2 REBUILD WITH (MAXDOP = 0, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.Tabla2 WITH COLUMNS

    ALTER INDEX ALL ON rca.ACTUALIZACIONES REBUILD WITH (MAXDOP = 0, SORT_IN_TEMPDB = ON, ONLINE = OFF, STATISTICS_NORECOMPUTE = OFF)
    UPDATE STATISTICS rca.ACTUALIZACIONES WITH COLUMNS
  END TRY
  BEGIN CATCH
    --It would be convenient to create the entire ibb.Log_Errores structure in the BI system and leave the corresponding traces
    SET @error = 1
  END CATCH
END
GO

Stored Procedures for Transfer of Information

In the OLTP, the stored procedures for the transfer of information are created. It is assumed that you are using a Linked Server called RCA, and a destination database also called RCA. The target tables, as already mentioned, are in the rca schema.

Obviously, the use of a Linked Server is not the most appropriate for doing this kind of thing. However, if this is one of the requirements, one of the ways to speed up the process is to use the 'SQLServerAsyncExecProcedures' libraries so that the available resources are optimized to the maximum, and above all, the transmission of data through the network, as this is often the bottleneck.

It can be seen that with the algorithm used so far it is not necessary for the information to travel in both directions, that is, in the OLTP, the information to be sent to BI is known, and it is not necessary that the BI information must travel to OLTP to make comparisons. Obviously, this generates great efficiency, since the bottleneck is usually in the network.

Thus, the following are the stored procedures that pass the data for Table1 and Table2:

If there are transactions, they should promote to distributed transactions: SET REMOTE_PROC_TRANSACTIONS ON.
It is desired that, if an error occurs in them, the execution does not continue (SET XACT_ABORT ON).
In BI, new records (HASH_OLD IS NULL), modified and deleted records (HASH_NEW! = HASH_OLD) must be inserted.
The ACT_CD_TIPO field will contain the values: 0-Insert(HASH_OLD IS NULL), 1-Update(HASH_NEW != HASH_OLD), 2-Delete(HASH_NEW IS NULL).
The ACT_CD_ACTUALIZACION field will contain the appropriate value as will be seen later.

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Traspasa_Tabla1') IS NOT NULL
  DROP PROCEDURE rca.Traspasa_Tabla1
GO
CREATE PROCEDURE rca.Traspasa_Tabla1
  @idParalelo BIGINT,
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
--If there are transactions, they must be able to promote
SET REMOTE_PROC_TRANSACTIONS ON
SET XACT_ABORT ON

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY
    --Insert of the New (HASH_OLD is null) and Modified (HASH_NEW != HASH_OLD)
    INSERT INTO RCA.RCA.rca.Tabla1
      SELECT tb1.Campo1, tb1.Campo2, tb1.Campo3,
            IIF(ExpH.HASH_OLD IS NULL, 0, IIF(ExpH.HASH_NEW IS NULL, 2, 1)), @idParalelo
        FROM rca.Tabla1 tb1
          INNER JOIN rca.Tabla1_HASH ExpH
            ON ExpH.Campo1 = tb1.Campo1 AND ExpH.Campo2 = tb1.Campo2
        WHERE ExpH.HASH_OLD IS NULL OR ExpH.HASH_NEW != ExpH.HASH_OLD

    EXECUTE ibb.Log_Errores_Inserta N'Table1 has been transferred.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR transferring Table1.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Traspasa_Tabla2') IS NOT NULL
  DROP PROCEDURE rca.Traspasa_Tabla2
GO
CREATE PROCEDURE rca.Traspasa_Tabla2
  @idParalelo BIGINT,
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
--If there are transactions, they must be able to promote
SET REMOTE_PROC_TRANSACTIONS ON
SET XACT_ABORT ON

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY
    --Insert of the New (HASH_OLD is null) and Modified (HASH_NEW != HASH_OLD)
    INSERT INTO RCA.RCA.rca.Tabla2
      SELECT tb1.Campo1, tb1.Campo2, tb1.Campo3,
            IIF(ExpH.HASH_OLD IS NULL, 0, IIF(ExpH.HASH_NEW IS NULL, 2, 1)), @idParalelo
        FROM rca.Tabla2 tb1
          INNER JOIN rca.Tabla2_HASH ExpH
            ON ExpH.Campo1 = tb1.Campo1
        WHERE ExpH.HASH_OLD IS NULL OR ExpH.HASH_NEW != ExpH.HASH_OLD

    EXECUTE ibb.Log_Errores_Inserta N'Table2 has been transferred.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR transferring Table2.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

Stored Procedures for COMMIT

Until now, all stored procedures do not need transactions, since in case of error the algorithm recovers from errors. For example, if it fails to update the OLTP tables or the generation of the HASH_NEWs, in the next execution of the process it will recover from these errors. If the transmission of the tables from OLTP to BI fails, the stored procedure 'RCA.RCA.rca.LimpiaTablasRemotas' will retrieve the remote tables.

However, the stored procedures described here are the ones that actually COMMIT the process, that is, they are stored procedures that need to be done within a transaction, since in case of failure, the process will not be able to recover. Transactions can be of any isolation level, since each stored procedure does not need to "see" the data for the other stored procedures.

The stored procedures 'rca.UpdateHash_Tabla1' and 'rca.UpdateHash_Tabla2' prepare the Table1_HASH and Table2_HASH tables for the next process execution:

They remove from Table1_HASH the records that have disappeared in the OLTP (HASH_NEW IS NULL).
They remove from Table1 those records that do not exist in Table1_HASH.
Updates the HASH_OLD field in Table1_HASH with the content of HASH_NEW.

If these procedures fail, it will not be possible to distinguish which record has been processed correctly from which has not. The transactions that can be defined for these stored procedures will be local to the instance where the OLTP is located.

In this critical step, the remote procedure 'RCA.RCA.rca.UpdateFinActualizacion' must also be called, which updates the field 'ACT_FH_FIN' indicating that the process has been executed correctly. If this method fails, the process will also not be recoverable on the next run. In this other case, the transaction that is associated with this stored procedure will be distributed, so MS DTC will have to be enabled on both servers.

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.ActualizaHash_Tabla1') IS NOT NULL
  DROP PROCEDURE rca.ActualizaHash_Tabla1
GO
CREATE PROCEDURE rca.ActualizaHash_Tabla1
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY

    --Delete from Table1_HASH those records where HASH_NEW is null (deleted).
    DELETE rca.Tabla1_HASH 
      WHERE HASH_NEW IS NULL

    --Delete from Table1 those records that do not exist in Table1_HASH
    DELETE Tbl
      FROM rca.Tabla1 Tbl
        LEFT JOIN rca.Tabla1_HASH TblHs
          ON Tbl.Campo1 = TblHs.Campo1 AND Tbl.Campo2 = TblHs.Campo2
      WHERE TblHs.Campo1 IS NULL AND TblHs.Campo2 IS NULL

    --Update HASH_OLD with HASH_NEW
    UPDATE rca.Tabla1_HASH
      SET HASH_OLD = HASH_NEW

    EXECUTE ibb.Log_Errores_Inserta N'Table1_HASH has been updated.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR updating Table1_HASH.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.ActualizaHash_Tabla2') IS NOT NULL
  DROP PROCEDURE rca.ActualizaHash_Tabla2
GO
CREATE PROCEDURE rca.ActualizaHash_Tabla2
  @NombreParalelo NVARCHAR(256),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET XACT_ABORT ON

  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY

    --Delete from Table2_HASH those records where HASH_NEW is null (deleted).
    DELETE rca.Tabla2_HASH 
      WHERE HASH_NEW IS NULL

    --Delete from Table2 those records that do not exist in Table2_HASH
    DELETE Tbl
      FROM rca.Tabla2 Tbl
        LEFT JOIN rca.Tabla2_HASH TblHs
          ON Tbl.Campo1 = TblHs.Campo1
      WHERE TblHs.Campo1 IS NULL

    --Update HASH_OLD with HASH_NEW
    UPDATE rca.Tabla2_HASH
      SET HASH_OLD = HASH_NEW

    EXECUTE ibb.Log_Errores_Inserta N'Table2_HASH has been updated.',
      0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR updating Table2_HASH.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

For this critical point there are two possibilities:

The first is to execute in parallel each of these stored procedures with independent transactions. If all procedures run without errors, all transactions will be committed. If any fail, a rollback will be made. It should be noted that the transaction required to execute 'RCA.RCA.rca.UpdateFinActualizacion' must be a distributed transaction, so MS DTC needs to be enabled on both servers. Experience says that this possibility has a high probability that transactions fail, not stored procedures, but the transactions themselves.
The second is to create a new stored procedure that calls all of these update HASH stored procedures, along with the stored procedure 'RCA.RCA.rca.UpdateFinUpdate' and put it in a transaction. The transaction is obviously going to promote to a distributed transaction, so you need to have MS DTC enabled on both servers.

Since a demonstration example is being presented here, the code for both solutions will be shown. Based on trial and error tests, you will decide on one or the other.

If the second possibility is chosen, the following stored procedure will have to be created:

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.Actualiza_Hash_Todas_Tablas_Secuencial') IS NOT NULL
  DROP PROCEDURE rca.Actualiza_Hash_Todas_Tablas_Secuencial
GO
CREATE PROCEDURE rca.Actualiza_Hash_Todas_Tablas_Secuencial
  @NombreParalelo NVARCHAR(256),
  @IdParalelo bigint,
  @inicioProceso DATETIME2(7),
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
SET REMOTE_PROC_TRANSACTIONS ON
SET XACT_ABORT ON

  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)
  DECLARE @Duracion nvarchar(50), @fFin DATETIME2(7)

  SET @error = 0
  BEGIN TRY

    EXEC rca.ActualizaHash_Tabla1 @NombreParalelo, @error out
    IF @error = 0
      EXEC rca.ActualizaHash_Tabla2 @NombreParalelo, @error out
    --And so with the rest of the tables

--Finally, update the process completion date
    IF @error = 0
    BEGIN
      SELECT @fFin = SYSDATETIME(), @Duracion = ibb.TimeSpanToNow(@inicioProceso)
      EXEC RCA.RCA.rca.UpdateFinActualizacion @IdParalelo, @Duracion, @fFin, @error out
    END

    IF @error = 0
      EXECUTE ibb.Log_Errores_Inserta N'The HASH of all the Tables has been updated.',
        0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
    ELSE
      EXECUTE ibb.Log_Errores_Inserta N'ERROR updating the HASH of all the Tables.',
        10, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR updating the HASH of all the Tables.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

Parallel Creation and Execution

Finally, the stored procedure that creates the 'Parallel' and executes it is detailed.

If there are transactions, they should promote to distributed transactions: SET REMOTE_PROC_TRANSACTIONS ON.
It is desired that, if an error occurs in them, it does not continue the execution: SET XACT_ABORT ON.
If there are open transactions in the call to this remote procedure, the execution is aborted.
In order to identify each execution by a different name, this name will be created from a sequence. This value will also be used to identify a new execution of the process in BI (ACT_CD_ACTUALIZACION).
The Parallel will be made up of 3 Lots of Works:
- 1º Updating of the OLTP repository tables. No transactions.
  Initial jobs executed sequentially:
  - Create the new record in rca.UPDATES and delete data from past failed runs.
  - If there are FKs in the OLTP repository: Delete them.
  - Delete the FKs from the BI repository.
  Jobs executed in parallel:
  - Update the tables of the OLTP repository in parallel.
    Note: the regeneration of indexes and statistics is included in the stored procedures, if not, a stored procedure would have to be made for it and executed as Final jobs.
  Final jobs to be executed sequentially:
  - If there are FKs in the OLTP repository: Create them
- 2º Transfer of Information from the OLTP repository to the BI repository. No transactions.
  Jobs executed in parallel:
  - Transfer the information from the OLTP repository to BI in parallel.
  Final jobs to be executed sequentially:
  - Create the FKs in BI.
  - Regenerate the indexes in BI.
- 3º Update HAHS_OLD by HASH_NEW and put the end of process indicator. With transactions.
  
  FIRST OPTION: Execute in parallel and with transactions the update of the HASH and the End Date
  Jobs executed in parallel:
  - Update the HASHs of the tables in the OLTP repository.
  - Put the end date in the BI repository.
  Execution options for parallel:
  - Three job batches have been created, and after experimenting several times with our systems and hardware, we deduced the optimal degrees of parallelism:
    The code indicates that for the first batch, the degree of parallelism is 6 (6 simultaneous processes), for the second 20 (data transmission), and for the third 15 (if the FIRST OPTION is chosen).
  - It is indicated that the transactions to be created will be of type 'ReadCommitted' and that the maximum time a transaction will remain open is 5 minutes. This value must be such that it is greater than the total duration of the 3rd batch, as it is the only one with transactions.
  Run parallel
  
  SECOND OPTION: Execute the stored procedure "rca.Update_Hash_All_Tables_Sequential" within a transaction
  Execution options for parallel:
  - Two batches of jobs have been created, and after experimenting several times with our systems and hardware, we deduced the optimal degrees of parallelism:
    The code indicates that for the first batch the degree of parallelism is 6 (6 simultaneous processes), and for the second, 20 (transmission of information). There is no third batch.
  - It is indicated that the transactions to be created will be of type 'ReadCommitted' and that the maximum time that a transaction will last open will be 5 minutes. However, this setting has no effect, since the third batch of jobs has not been created and the rest of the batches have no transactions.
  Run parallel
  If the parallel was successful:
  - Open a transaction
  - Execute the stored procedure created expressly for the SECOND OPTION.
  - If it has been executed successfully, end the transaction, otherwise, undo it.

 SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
IF OBJECT_ID ('rca.RellenaTablasParalelo') IS NOT NULL
  DROP PROCEDURE rca.RellenaTablasParalelo
GO
CREATE PROCEDURE rca.RellenaTablasParalelo
  @error INT OUTPUT
AS BEGIN SET NOCOUNT ON
--SET IMPLICIT_TRANSACTIONS OFF
SET REMOTE_PROC_TRANSACTIONS ON
SET XACT_ABORT ON

  DECLARE @NumTranAntes INT = @@TRANCOUNT
  DECLARE @idParalelo bigint, @NombreParalelo varchar(256) = NULL,
          @Parametros NVARCHAR(MAX) = NULL, @Trabajo NVARCHAR(MAX) = NULL
  DECLARE @Duracion nvarchar(50), @fFin DATETIME2(7)

  --Variables for LOG
  DECLARE @inicio DATETIME2(7) = SYSDATETIME(), @nameDB SYSNAME = DB_NAME(),
          @esquema SYSNAME = OBJECT_SCHEMA_NAME(@@PROCID),
          @procedimiento SYSNAME = OBJECT_NAME(@@PROCID)

  SET @error = 0
  BEGIN TRY

    --If there are open transactions, the system will not work
    IF @NumTranAntes > 0
    BEGIN
      SET @error = 1
      EXECUTE ibb.Log_Errores_Inserta N'The table transfer cannot function with an open transaction.',
        0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
      RETURN
    END

    --In order to identify each execution by a different name, this name will be created from a sequence:
    SELECT @idParalelo = NEXT VALUE FOR rca.SEQ_PARALELO, @NombreParalelo = CONCAT('ParaleloRepositorio', @idParalelo)

    --The parallel process will consist of 3 Job Batches:
    --  1st Update of the tables in the local repository. No transactions.
    --  2nd Transfer of information from the local repository to the remote one. No transactions.
    --  3rd Update of HASH_OLD with HASH_NEW and set the job completion indicator. With transactions.

    --Job Batch 1st:
    --  1st Create the new record in rca.ACTUALIZACIONES and remove data from past failed executions.
    --  2nd If there are local FKs in the repository: Delete them.
    --  3rd Delete the remote FKs.
    SET @Trabajo = CONCAT(N'EXEC SGR.rca.LimpiaTablasRemotas N''', TRY_CONVERT(nvarchar, @inicio, 121), N''', ', @idParalelo, N', @error out')
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 1, N'Iniciales', N'Limpia Tablas Remotas', 0, @Trabajo
    SELECT @Parametros = CONCAT(N'N''', @NombreParalelo, N''', @error out'), @Trabajo = CONCAT(N'EXEC rca.EliminaFkLocales ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 1, N'Iniciales', N'Elimina FKs Locales', 0, @Trabajo
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 1, N'Iniciales', N'Elimina FKs Remotas', 0, N'EXEC SGR.rca.EliminaFkRemotas @error out'

    --  4th Update the tables in the local repository in parallel.
    --    Note: The stored procedures include the regeneration of indexes and statistics; otherwise,
    --          a stored procedure would need to be created for this and executed as Final jobs.
    SET @Trabajo = CONCAT(N'EXEC rca.RellenaTabla1 ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 1, N'Paralelo', N'Carga Tabla1', 0, @Trabajo
    SET @Trabajo = CONCAT(N'EXEC rca.RellenaTabla2 ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 1, N'Paralelo', N'Carga Tabla2', 0, @Trabajo

    --  5th If there are local FKs in the repository: Create them
    SET @Trabajo = CONCAT(N'EXEC rca.CreaFkLocales ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 1, N'Finales', N'Crea FKs Locales', 0, @Trabajo

    --Job Batch 2nd:
    --  1st Transfer the information from local to remote in parallel
    --  2nd Create the remote FKs
    --  3rd Regenerate the remote indexes
    SELECT @Parametros = CONCAT(@idParalelo, N', N''', @NombreParalelo, N''', @error out'), @Trabajo = CONCAT(N'EXEC rca.Traspasa_Tabla1 ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 2, N'Paralelo', N'Traspasa Tabla1', 0, @Trabajo
    SELECT @Trabajo = CONCAT(N'EXEC rca.Traspasa_Tabla2 ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 2, N'Paralelo', N'Traspasa Tabla2', 0, @Trabajo

    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 2, N'Finales', N'Crea FKs Remotas', 0, N'EXEC SGR.rca.CreaFkRemotas @error out'
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 2, N'Finales', N'Regenera Índices Remotos', 0, N'EXEC SGR.rca.RegeneraIndicesRemotos @error out'

    --FIRST OPTION: Execute in parallel and with transactions the update of the HASH and the Completion Date
    --Job Batch 3rd, with transactions:
    --  1st Update the HASHs of the tables
    --  2nd Set the completion date in the remote
    SELECT @Parametros = CONCAT(N'''', @NombreParalelo, N''', @error out'), @Trabajo = CONCAT(N'EXEC rca.ActualizaHash_Tabla1 ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 3, N'Paralelo', N'ActualizaHash Tabla1', 1, @Trabajo
    SELECT @Trabajo = CONCAT(N'EXEC rca.ActualizaHash_Tabla2 ', @Parametros)
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 3, N'Paralelo', N'ActualizaHash Tabla2', 1, @Trabajo
    SELECT @Duracion = ibb.TimeSpanToNow(@inicio), @Trabajo = CONCAT(N'EXEC SGR.rca.UpdateFinActualizacion ', @idParalelo,
      N', ''', @Duracion, N''', ''', TRY_CONVERT(nvarchar, SYSDATETIME(), 121), N''', @error out')
    EXEC ibbclr.Parallel_AddJob @NombreParalelo, 3, N'Paralelo', N'Actualiza Fecha Fin', 1, @Trabajo

    --Options for executing the Jobs
    --  Three job batches have been created, and after experimenting several times, we found that the following degrees
    --  of parallelism are optimal
    EXEC ibbclr.Parallel_SetOption_MaxTaskBatch @NombreParalelo, 1, 6
    EXEC ibbclr.Parallel_SetOption_MaxTaskBatch @NombreParalelo, 2, 20
    EXEC ibbclr.Parallel_SetOption_MaxTaskBatch @NombreParalelo, 3, 15
    --  It is indicated that the transactions to be created will be of type 'ReadCommitted' and that the maximum time a
    --  transaction will remain open is 5 minutes. This value must be such that it is greater than the total duration of the 3rd batch, as it is the
    --  only one with transactions
    EXEC ibbclr.Parallel_SetOptions @NombreParalelo, N'ReadCommitted', 5

    --Execution of the Parallel
    EXEC @error = ibbclr.Parallel_Execute @NombreParalelo

    ----SECOND OPTION: Execute the stored procedure rca.Actualiza_Hash_Todas_Tablas_Secuencial within a transaction
    --BEGIN TRAN
    --EXEC rca.Actualiza_Hash_Todas_Tablas_Secuencial @NombreParalelo, @idParalelo, @inicio, @error out
    --IF @error = 0
    --  COMMIT
    --ELSE
    --  ROLLBACK

    IF @error = 0
      EXECUTE ibb.Log_Errores_Inserta N'The Parallel has been executed.',
        0, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
    ELSE
      EXECUTE ibb.Log_Errores_Inserta N'ERROR executing the Parallel.',
        10, @NombreParalelo, @inicio, @nameDB, @esquema, @procedimiento
  END TRY
  BEGIN CATCH
    DECLARE @ERR_SQL_NUM_ERROR INT = ERROR_NUMBER(), @ERR_SQL_NUM_LINEA INT = ERROR_LINE(),
            @ERR_SQL_MENSAJE NVARCHAR(MAX) = ERROR_MESSAGE(), @ERR_SQL_PROCEDIMIENTO SYSNAME = ERROR_PROCEDURE(),
            @ERR_SQL_SEVERIDAD INT = ERROR_SEVERITY(), @ERR_SQL_ESTADO INT = ERROR_STATE()
    EXECUTE ibb.Log_Errores_Inserta N'ERROR executing the Parallel.',
      10, @NombreParalelo, NULL, @nameDB, @esquema, @procedimiento,
      @ERR_SQL_NUM_ERROR, @ERR_SQL_NUM_LINEA, @ERR_SQL_MENSAJE, @ERR_SQL_PROCEDIMIENTO, @ERR_SQL_SEVERIDAD, @ERR_SQL_ESTADO
    SET @error = @ERR_SQL_NUM_ERROR
  END CATCH
END
GO

Once everything is programmed, the execution is quite simple:

   DECLARE @error int
  EXEC rca.RellenaTablasParalelo @error out

In the table 'ibb.Log_Errores' the LOG messages will appear together with the ERROR messages, if any. The messages are quite long, and contain line breaks, so it is recommended to activate the option to keep CR/LF of SSMS, to be able to copy and paste them where they can be read in full.

It is not within the scope of this chapter, but the algorithm described here can be added functionality for disaster recovery, that is, retrieving the OLTP repository tables from the BI repository tables, and vice versa.