11592275723

Committed 30 Oct 2024 11:28AM UTC coverage: 57.386% (+0.004%) from 57.382%

Build # 11592275723

Build Type

Pull #2006

github

Committed by

JFriel

Commit Message

Merge branch 'develop' of https://github.com/HicServices/RDMP into release/8.4.0

Pull Request Pull Request #2006: Release: 8.4.0

Run Details

11208 of 21044 branches covered (53.26%)

Branch coverage included in aggregate %.

6 of 7 new or added lines in 4 files covered. (85.71%)

30 existing lines in 3 files now uncovered.

31734 of 53786 relevant lines covered (59.0%)

8216.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.11

/Rdmp.Core/DataLoad/Engine/Migration/QueryBuilding/OverwriteMigrationStrategy.cs

// Copyright (c) The University of Dundee 2018-2019
// This file is part of the Research Data Management Platform (RDMP).
// RDMP is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
// RDMP is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
// You should have received a copy of the GNU General Public License along with RDMP. If not, see <https://www.gnu.org/licenses/>.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using FAnsi;
using FAnsi.Connections;
using FAnsi.Discovery;
using FAnsi.Discovery.QuerySyntax;
using Rdmp.Core.DataFlowPipeline;
using Rdmp.Core.DataLoad.Engine.Job;
using Rdmp.Core.DataLoad.Triggers;
using Rdmp.Core.ReusableLibraryCode.Progress;

namespace Rdmp.Core.DataLoad.Engine.Migration.QueryBuilding;

/// <summary>
/// Migrates from STAGING to LIVE a single table (with a MigrationColumnSet).  This is an UPSERT (new replaces old) operation achieved (in SQL) with MERGE and
/// UPDATE (based on primary key).  Both tables must be on the same server.  A MERGE sql statement will be created using LiveMigrationQueryHelper and executed
/// within a transaction.
/// </summary>
public class OverwriteMigrationStrategy : DatabaseMigrationStrategy
{
    public OverwriteMigrationStrategy(IManagedConnection managedConnection)
        : base(managedConnection)
    {
    }

    public override void MigrateTable(IDataLoadJob job, MigrationColumnSet columnsToMigrate, int dataLoadInfoID,
        GracefulCancellationToken cancellationToken, ref int inserts, ref int updates)
    {
        var server = columnsToMigrate.DestinationTable.Database.Server;

        //see CrossDatabaseMergeCommandTest

        /*          ------------MIGRATE NEW RECORDS (novel by primary key)--------
         *

INSERT INTO CrossDatabaseMergeCommandTo..ToTable (Name,Age,Postcode,hic_dataLoadRunID)
SELECT
[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Name,
[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Age,
[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Postcode,
1
FROM
[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING
left join
CrossDatabaseMergeCommandTo..ToTable
on
[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Age = CrossDatabaseMergeCommandTo..ToTable.Age
AND
[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Name = CrossDatabaseMergeCommandTo..ToTable.Name
WHERE
CrossDatabaseMergeCommandTo..ToTable.Age is null
*/

        var sbInsert = new StringBuilder();
        var syntax = server.GetQuerySyntaxHelper();


        sbInsert.AppendLine(
            $"INSERT INTO {columnsToMigrate.DestinationTable.GetFullyQualifiedName()} ({string.Join(",", columnsToMigrate.FieldsToUpdate.Select(c => syntax.EnsureWrapped(c.GetRuntimeName())))}");

        //if we are not ignoring the trigger then we should record the data load run ID
        if (!job.LoadMetadata.IgnoreTrigger)
            sbInsert.AppendLine($",{syntax.EnsureWrapped(SpecialFieldNames.DataLoadRunID)}");

        sbInsert.AppendLine(")");

        sbInsert.AppendLine("SELECT");

        // Add the columns we are migrating
        sbInsert.AppendLine(string.Join($",{Environment.NewLine}",
            columnsToMigrate.FieldsToUpdate.Select(c => c.GetFullyQualifiedName())));

        // If we are using trigger also add the run ID e.g. ",50"
        if (!job.LoadMetadata.IgnoreTrigger)
            sbInsert.AppendLine($",{dataLoadInfoID}");

        sbInsert.AppendLine("FROM");
        sbInsert.AppendLine(columnsToMigrate.SourceTable.GetFullyQualifiedName());
        sbInsert.AppendLine("LEFT JOIN");
        sbInsert.AppendLine(columnsToMigrate.DestinationTable.GetFullyQualifiedName());
        sbInsert.AppendLine("ON");

        sbInsert.AppendLine(
            string.Join($" AND {Environment.NewLine}",
                columnsToMigrate.PrimaryKeys.Select(
                    pk =>
                        string.Format("{0}.{1}={2}.{1}", columnsToMigrate.SourceTable.GetFullyQualifiedName(),
                            syntax.EnsureWrapped(pk.GetRuntimeName()),
                            columnsToMigrate.DestinationTable.GetFullyQualifiedName()))));

        sbInsert.AppendLine("WHERE");
        sbInsert.AppendLine(
            $"{columnsToMigrate.DestinationTable.GetFullyQualifiedName()}.{syntax.EnsureWrapped(columnsToMigrate.PrimaryKeys.First().GetRuntimeName())} IS NULL");

        //right at the end of the SELECT
        if (columnsToMigrate.DestinationTable.Database.Server.DatabaseType == DatabaseType.MySql)
            sbInsert.Append(" FOR UPDATE");

        var insertSql = sbInsert.ToString();

        var cmd = server.GetCommand(insertSql, _managedConnection);
        cmd.CommandTimeout = Timeout;

        job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information,
            $"INSERT query: {Environment.NewLine}{insertSql}"));

        cancellationToken.ThrowIfCancellationRequested();


        try
        {
            inserts = cmd.ExecuteNonQuery();

            var sqlLines = new List<CustomLine>();

            var toSet = columnsToMigrate.FieldsToUpdate.Where(c => !c.IsPrimaryKey).Select(c =>
                string.Format("t1.{0} = t2.{0}", syntax.EnsureWrapped(c.GetRuntimeName()))).ToArray();

            if (!toSet.Any())
            {
                job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Warning,
                    $"Table {columnsToMigrate.DestinationTable} is entirely composed of PrimaryKey columns or hic_ columns so UPDATE will NOT take place"));
                return;
            }

            var toDiff = columnsToMigrate.FieldsToDiff.Where(c => !c.IsPrimaryKey).ToArray();

            if (!toDiff.Any())
            {
                job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Warning,
                    $"Table {columnsToMigrate.DestinationTable} is entirely composed of PrimaryKey columns or hic_ columns/ other non DIFF columns that will not result in an UPDATE will NOT take place"));
                return;
            }

            //t1.Name = t2.Name, t1.Age=T2.Age etc
            sqlLines.Add(new CustomLine(string.Join(",", toSet), QueryComponent.SET));

            //also update the hic_dataLoadRunID field
            if (!job.LoadMetadata.IgnoreTrigger)
                sqlLines.Add(new CustomLine(
                    $"t1.{syntax.EnsureWrapped(SpecialFieldNames.DataLoadRunID)}={dataLoadInfoID}",
                    QueryComponent.SET));

            //t1.Name <> t2.Name AND t1.Age <> t2.Age etc
            sqlLines.Add(new CustomLine(string.Join(" OR ", toDiff.Select(c => GetORLine(c, syntax))),
                QueryComponent.WHERE));

            //the join
            sqlLines.AddRange(columnsToMigrate.PrimaryKeys.Select(p =>
                new CustomLine(string.Format("t1.{0} = t2.{0}", syntax.EnsureWrapped(p.GetRuntimeName())),
                    QueryComponent.JoinInfoJoin)));

            var updateHelper = columnsToMigrate.DestinationTable.Database.Server.GetQuerySyntaxHelper().UpdateHelper;

            var updateQuery = updateHelper.BuildUpdate(
                columnsToMigrate.DestinationTable,
                columnsToMigrate.SourceTable,
                sqlLines);

            job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information,
                $"Update query:{Environment.NewLine}{updateQuery}"));

            var updateCmd = server.GetCommand(updateQuery, _managedConnection);
            updateCmd.CommandTimeout = Timeout;
            cancellationToken.ThrowIfCancellationRequested();

            try
            {
                updates = updateCmd.ExecuteNonQuery();
            }
            catch (Exception e)
            {
                job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Error,
                    $"Did not successfully perform the update queries: {updateQuery}", e));
                throw new Exception($"Did not successfully perform the update queries: {updateQuery} - {e}");
            }
        }
        catch (OperationCanceledException)
        {
            throw; // have to catch and rethrow this because of the catch-all below
        }
        catch (Exception e)
        {
            job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Error,
                $"Failed to migrate {columnsToMigrate.SourceTable} to {columnsToMigrate.DestinationTable}", e));
            throw new Exception(
                $"Failed to migrate {columnsToMigrate.SourceTable} to {columnsToMigrate.DestinationTable}: {e}");
        }
    }

    private static string GetORLine(DiscoveredColumn c, IQuerySyntaxHelper syntax) => string.Format(
        "(t1.{0} <> t2.{0} OR (t1.{0} is null AND t2.{0} is not null) OR (t2.{0} is null AND t1.{0} is not null))",
        syntax.EnsureWrapped(c.GetRuntimeName()));
}

1	// Copyright (c) The University of Dundee 2018-2019
2	// This file is part of the Research Data Management Platform (RDMP).
3	// RDMP is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
4	// RDMP is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
5	// You should have received a copy of the GNU General Public License along with RDMP. If not, see <https://www.gnu.org/licenses/>.
6
7	using System;
8	using System.Collections.Generic;
9	using System.Linq;
10	using System.Text;
11	using FAnsi;
12	using FAnsi.Connections;
13	using FAnsi.Discovery;
14	using FAnsi.Discovery.QuerySyntax;
15	using Rdmp.Core.DataFlowPipeline;
16	using Rdmp.Core.DataLoad.Engine.Job;
17	using Rdmp.Core.DataLoad.Triggers;
18	using Rdmp.Core.ReusableLibraryCode.Progress;
19
20	namespace Rdmp.Core.DataLoad.Engine.Migration.QueryBuilding;
21
22	/// <summary>
23	/// Migrates from STAGING to LIVE a single table (with a MigrationColumnSet). This is an UPSERT (new replaces old) operation achieved (in SQL) with MERGE and
24	/// UPDATE (based on primary key). Both tables must be on the same server. A MERGE sql statement will be created using LiveMigrationQueryHelper and executed
25	/// within a transaction.
26	/// </summary>
27	public class OverwriteMigrationStrategy : DatabaseMigrationStrategy
28	{
29	public OverwriteMigrationStrategy(IManagedConnection managedConnection)
30	: base(managedConnection)	48✔
31	{
32	}	48✔
33
34	public override void MigrateTable(IDataLoadJob job, MigrationColumnSet columnsToMigrate, int dataLoadInfoID,
35	GracefulCancellationToken cancellationToken, ref int inserts, ref int updates)
36	{
37	var server = columnsToMigrate.DestinationTable.Database.Server;	50✔
38
39	//see CrossDatabaseMergeCommandTest
40
41	/* ------------MIGRATE NEW RECORDS (novel by primary key)--------
42	*
43
44	INSERT INTO CrossDatabaseMergeCommandTo..ToTable (Name,Age,Postcode,hic_dataLoadRunID)
45	SELECT
46	[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Name,
47	[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Age,
48	[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Postcode,
49	1
50	FROM
51	[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING
52	left join
53	CrossDatabaseMergeCommandTo..ToTable
54	on
55	[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Age = CrossDatabaseMergeCommandTo..ToTable.Age
56	AND
57	[CrossDatabaseMergeCommandFrom]..CrossDatabaseMergeCommandTo_ToTable_STAGING.Name = CrossDatabaseMergeCommandTo..ToTable.Name
58	WHERE
59	CrossDatabaseMergeCommandTo..ToTable.Age is null
60	*/
61
62	var sbInsert = new StringBuilder();	50✔
63	var syntax = server.GetQuerySyntaxHelper();	50✔
64
65
66	sbInsert.AppendLine(	50✔
67	$"INSERT INTO {columnsToMigrate.DestinationTable.GetFullyQualifiedName()} ({string.Join(",", columnsToMigrate.FieldsToUpdate.Select(c => syntax.EnsureWrapped(c.GetRuntimeName())))}");	352✔
68
69	//if we are not ignoring the trigger then we should record the data load run ID
70	if (!job.LoadMetadata.IgnoreTrigger)	50✔
71	sbInsert.AppendLine($",{syntax.EnsureWrapped(SpecialFieldNames.DataLoadRunID)}");	44✔
72
73	sbInsert.AppendLine(")");	50✔
74
75	sbInsert.AppendLine("SELECT");	50✔
76
77	// Add the columns we are migrating
78	sbInsert.AppendLine(string.Join($",{Environment.NewLine}",	50✔
79	columnsToMigrate.FieldsToUpdate.Select(c => c.GetFullyQualifiedName())));	352✔
80
81	// If we are using trigger also add the run ID e.g. ",50"
82	if (!job.LoadMetadata.IgnoreTrigger)	50✔
83	sbInsert.AppendLine($",{dataLoadInfoID}");	44✔
84
85	sbInsert.AppendLine("FROM");	50✔
86	sbInsert.AppendLine(columnsToMigrate.SourceTable.GetFullyQualifiedName());	50✔
87	sbInsert.AppendLine("LEFT JOIN");	50✔
88	sbInsert.AppendLine(columnsToMigrate.DestinationTable.GetFullyQualifiedName());	50✔
89	sbInsert.AppendLine("ON");	50✔
90
91	sbInsert.AppendLine(	50✔
92	string.Join($" AND {Environment.NewLine}",	50✔
93	columnsToMigrate.PrimaryKeys.Select(	50✔
94	pk =>	50✔
95	string.Format("{0}.{1}={2}.{1}", columnsToMigrate.SourceTable.GetFullyQualifiedName(),	142✔
96	syntax.EnsureWrapped(pk.GetRuntimeName()),	142✔
97	columnsToMigrate.DestinationTable.GetFullyQualifiedName()))));	142✔
98
99	sbInsert.AppendLine("WHERE");	50✔
100	sbInsert.AppendLine(	50✔
101	$"{columnsToMigrate.DestinationTable.GetFullyQualifiedName()}.{syntax.EnsureWrapped(columnsToMigrate.PrimaryKeys.First().GetRuntimeName())} IS NULL");	50✔
102
103	//right at the end of the SELECT
104	if (columnsToMigrate.DestinationTable.Database.Server.DatabaseType == DatabaseType.MySql)	50✔
105	sbInsert.Append(" FOR UPDATE");	18✔
106
107	var insertSql = sbInsert.ToString();	50✔
108
109	var cmd = server.GetCommand(insertSql, _managedConnection);	50✔
110	cmd.CommandTimeout = Timeout;	50✔
111
112	job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information,	50✔
113	$"INSERT query: {Environment.NewLine}{insertSql}"));	50✔
114
115	cancellationToken.ThrowIfCancellationRequested();	50✔
116
117
118	try
119	{
120	inserts = cmd.ExecuteNonQuery();	50✔
121
122	var sqlLines = new List<CustomLine>();	50✔
123
124	var toSet = columnsToMigrate.FieldsToUpdate.Where(c => !c.IsPrimaryKey).Select(c =>	352✔
125	string.Format("t1.{0} = t2.{0}", syntax.EnsureWrapped(c.GetRuntimeName()))).ToArray();	260✔
126
127	if (!toSet.Any())	50✔
128	{
129	job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Warning,	8✔
130	$"Table {columnsToMigrate.DestinationTable} is entirely composed of PrimaryKey columns or hic_ columns so UPDATE will NOT take place"));	8✔
131	return;	8✔
132	}
133
134	var toDiff = columnsToMigrate.FieldsToDiff.Where(c => !c.IsPrimaryKey).ToArray();	326✔
135
136	if (!toDiff.Any())	42✔
137	{
138	job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Warning,	2✔
139	$"Table {columnsToMigrate.DestinationTable} is entirely composed of PrimaryKey columns or hic_ columns/ other non DIFF columns that will not result in an UPDATE will NOT take place"));	2✔
140	return;	2✔
141	}
142
143	//t1.Name = t2.Name, t1.Age=T2.Age etc
144	sqlLines.Add(new CustomLine(string.Join(",", toSet), QueryComponent.SET));	40✔
145
146	//also update the hic_dataLoadRunID field
147	if (!job.LoadMetadata.IgnoreTrigger)	40✔
148	sqlLines.Add(new CustomLine(	34✔
149	$"t1.{syntax.EnsureWrapped(SpecialFieldNames.DataLoadRunID)}={dataLoadInfoID}",	34✔
150	QueryComponent.SET));	34✔
151
152	//t1.Name <> t2.Name AND t1.Age <> t2.Age etc
153	sqlLines.Add(new CustomLine(string.Join(" OR ", toDiff.Select(c => GetORLine(c, syntax))),	248✔
154	QueryComponent.WHERE));	40✔
155
156	//the join
157	sqlLines.AddRange(columnsToMigrate.PrimaryKeys.Select(p =>	40✔
158	new CustomLine(string.Format("t1.{0} = t2.{0}", syntax.EnsureWrapped(p.GetRuntimeName())),	112✔
159	QueryComponent.JoinInfoJoin)));	112✔
160
161	var updateHelper = columnsToMigrate.DestinationTable.Database.Server.GetQuerySyntaxHelper().UpdateHelper;	40✔
162
163	var updateQuery = updateHelper.BuildUpdate(	40✔
164	columnsToMigrate.DestinationTable,	40✔
165	columnsToMigrate.SourceTable,	40✔
166	sqlLines);	40✔
167
168	job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information,	40✔
169	$"Update query:{Environment.NewLine}{updateQuery}"));	40✔
170
171	var updateCmd = server.GetCommand(updateQuery, _managedConnection);	40✔
172	updateCmd.CommandTimeout = Timeout;	40✔
173	cancellationToken.ThrowIfCancellationRequested();	40✔
174
175	try
176	{
177	updates = updateCmd.ExecuteNonQuery();	40✔
178	}	40✔
UNCOV 179	catch (Exception e)	×
180	{
UNCOV 181	job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Error,	×
182	$"Did not successfully perform the update queries: {updateQuery}", e));	×
UNCOV 183	throw new Exception($"Did not successfully perform the update queries: {updateQuery} - {e}");	×
184	}
185	}	40✔
186	catch (OperationCanceledException)	×
187	{
UNCOV 188	throw; // have to catch and rethrow this because of the catch-all below	×
189	}
UNCOV 190	catch (Exception e)	×
191	{
UNCOV 192	job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Error,	×
193	$"Failed to migrate {columnsToMigrate.SourceTable} to {columnsToMigrate.DestinationTable}", e));	×
UNCOV 194	throw new Exception(	×
195	$"Failed to migrate {columnsToMigrate.SourceTable} to {columnsToMigrate.DestinationTable}: {e}");	×
196	}
197	}	50✔
198
199	private static string GetORLine(DiscoveredColumn c, IQuerySyntaxHelper syntax) => string.Format(	208✔
200	"(t1.{0} <> t2.{0} OR (t1.{0} is null AND t2.{0} is not null) OR (t2.{0} is null AND t1.{0} is not null))",	208✔
201	syntax.EnsureWrapped(c.GetRuntimeName()));	208✔
202	}

HicServices / RDMP / 11592275723

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous