Skip to content

Task/rdmp 376 extraction structural changes #2343

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
JFriel wants to merge 15 commits into
base: develop
Choose a base branch
from
Draft

Task/rdmp 376 extraction structural changes #2343

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
acfae60
working add
JFriel Jun 3, 2026
ba4865d
col working
JFriel Jun 4, 2026
0a6e3f3
null columns
JFriel Jun 4, 2026
c24f37d
working triggr
JFriel Jun 4, 2026
2cfbe07
tidy up
JFriel Jun 4, 2026
c0c381e
update codeql
JFriel Jun 9, 2026
6181033
add tests
JFriel Jun 10, 2026
f7cb8b8
update test pipelines
JFriel Jun 10, 2026
e733b9c
Merge branch 'develop' into task/RDMP-376-extraction-structural-changes
JFriel Jul 2, 2026
21bcd74
update merge component
JFriel Jul 2, 2026
4b7bd9f
add tests
JFriel Jul 2, 2026
61aceed
update merge
JFriel Jul 6, 2026
8ddd426
fix some tests
JFriel Jul 6, 2026
be04f3f
update tests
JFriel Jul 6, 2026
97eeb0c
update test
JFriel Jul 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1,926 changes: 1,926 additions & 0 deletions ...t/DataExtraction/ExecuteFullExtractionToDatabaseMSSqlDestinationWithArchiveTriggerTest.cs
View file
Open in desktop

Large diffs are not rendered by default.

1,967 changes: 1,966 additions & 1 deletion Rdmp.Core.Tests/DataExport/DataExtraction/MSSQLMergeDestinationTests.cs
View file
Open in desktop

Large diffs are not rendered by default.

89 changes: 69 additions & 20 deletions ...e/DataExport/DataExtraction/Pipeline/Destinations/ExecuteFullExtractionToDatabaseMSSql.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
// 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 Amazon.Auth.AccessControlPolicy;
using FAnsi.Discovery;
using Rdmp.Core.CommandExecution;
using Rdmp.Core.Curation.Data;
Expand All @@ -14,8 +13,8 @@
using Rdmp.Core.DataExport.DataRelease.Pipeline;
using Rdmp.Core.DataExport.DataRelease.Potential;
using Rdmp.Core.DataFlowPipeline;
using Rdmp.Core.DataLoad.Engine.Job.Scheduling;
using Rdmp.Core.DataLoad.Engine.Pipeline.Destinations;
using Rdmp.Core.DataLoad.Triggers;
using Rdmp.Core.DataLoad.Triggers.Exceptions;
using Rdmp.Core.DataLoad.Triggers.Implementations;
using Rdmp.Core.MapsDirectlyToDatabaseTable;
Expand All @@ -31,7 +30,6 @@
using System.IO;
using System.Linq;
using System.Text.RegularExpressions;
using YamlDotNet.Core;

namespace Rdmp.Core.DataExport.DataExtraction.Pipeline.Destinations;

Expand Down Expand Up @@ -169,6 +167,15 @@ protected override void WriteRows(DataTable toProcess, IDataLoadEventListener jo
LinesWritten += toProcess.Rows.Count;
}


private bool hasStructuralChanges(DataTable source, DiscoveredTable destination)
{
var sourceColumns = source.Columns.Cast<DataColumn>().Select(c => c.ColumnName).ToList();
var destinationColumns = destination.DiscoverColumns().Select(c => c.GetRuntimeName()).ToList();
return !sourceColumns.All(destinationColumns.Contains) || !destinationColumns.All(sourceColumns.Contains);
}


private DataTableUploadDestination PrepareDestination(IDataLoadEventListener listener, DataTable toProcess)
{
//see if the user has entered an extraction server/database
Expand All @@ -191,7 +198,17 @@ private DataTableUploadDestination PrepareDestination(IDataLoadEventListener lis
if (existing.Exists())
{
var hasPKs = existing.DiscoverColumns().Any(col => col.IsPrimaryKey);

TriggerImplementerFactory triggerFactory = new TriggerImplementerFactory(FAnsi.DatabaseType.MicrosoftSQLServer);
var implementor = triggerFactory.Create(existing);
bool triggerPresent;
try
{
triggerPresent = implementor.GetTriggerStatus() == DataLoad.Triggers.TriggerStatus.Enabled;
}
catch (TriggerMissingException)
{
triggerPresent = false;
}
if (!AlwaysDropExtractionTables)
{
//check the PKs are the same
Expand All @@ -207,6 +224,47 @@ private DataTableUploadDestination PrepareDestination(IDataLoadEventListener lis
"""));
return null;
}
if (hasStructuralChanges(toProcess, existing))
{
var sourceColumns = toProcess.Columns.Cast<DataColumn>().Select(c => c.ColumnName).ToList();
var destinationColumns = existing.DiscoverColumns().Select(c => c.GetRuntimeName()).ToList();

if (triggerPresent && destinationColumns.Except(sourceColumns).Where(c => !SpecialFieldNames.IsHicPrefixed(c)).Any())//only mess about with column removal if there is an archive trigger
{

//move everything into the archive - do this by updating the HIC_validfrom
var sql = $"UPDATE {existing.GetFullyQualifiedName()} set {SpecialFieldNames.ValidFrom} = GETDATE()";
using var con = _destinationDatabase.Server.GetConnection();
con.Open();
using var cmd = _destinationDatabase.Server.GetCommand(sql, con);
cmd.CommandTimeout = 30000;
cmd.ExecuteNonQuery();

var removedColumns = destinationColumns.Except(sourceColumns).Where(c => !SpecialFieldNames.IsHicPrefixed(c));
foreach (var column in removedColumns.Select(c => existing.DiscoverColumn(c)))
{
existing.DropColumn(column);
}
string triggerProblems = "";
string triggerOK = "";
implementor.DropTrigger(out triggerProblems, out triggerOK);
if (triggerProblems != "")
{
listener.OnNotify(this, new NotifyEventArgs(ProgressEventType.Error, triggerProblems));
}

existing = _destinationDatabase.ExpectTable(tblName);
implementor = triggerFactory.Create(existing);
try
{
triggerPresent = implementor.GetTriggerStatus() == DataLoad.Triggers.TriggerStatus.Enabled;
}
catch (TriggerMissingException)
{
triggerPresent = false;
}
}
}
}

if (_request.IsBatchResume)
Expand All @@ -228,18 +286,6 @@ private DataTableUploadDestination PrepareDestination(IDataLoadEventListener lis
}
else if (UseArchiveTrigger && hasPKs)
{

TriggerImplementerFactory triggerFactory = new TriggerImplementerFactory(FAnsi.DatabaseType.MicrosoftSQLServer);
var implementor = triggerFactory.Create(existing);
bool present;
try
{
present = implementor.GetTriggerStatus() == DataLoad.Triggers.TriggerStatus.Enabled;
}
catch (TriggerMissingException)
{
present = false;
}
//check the columns are correct, we might have added some
var existingColumns = existing.DiscoverColumns();
var existingColumnNames = existingColumns.Select(ec => ec.GetRuntimeName());
Expand All @@ -253,9 +299,12 @@ private DataTableUploadDestination PrepareDestination(IDataLoadEventListener lis
foreach (var column in newColumns)
{
existing.AddColumn(column, new TypeGuesser.DatabaseTypeRequest(toProcess.Columns[column].DataType), true, 30000);
archiveTable.AddColumn(column, new TypeGuesser.DatabaseTypeRequest(toProcess.Columns[column].DataType), true, 30000);
if (archiveTable.DiscoverColumns().All(col => col.GetRuntimeName() != column))
{
archiveTable.AddColumn(column, new TypeGuesser.DatabaseTypeRequest(toProcess.Columns[column].DataType), true, 30000);
}
}
if (present)
if (triggerPresent)
{
string triggerProblems = "";
string triggerOK = "";
Expand All @@ -267,12 +316,12 @@ private DataTableUploadDestination PrepareDestination(IDataLoadEventListener lis

existing = _destinationDatabase.ExpectTable(tblName);
implementor = triggerFactory.Create(existing);
present = false;
triggerPresent = false;
}
}
}

if (!present)
if (!triggerPresent)
{
implementor.CreateTrigger(ThrowImmediatelyCheckNotifier.Quiet);
}
Expand Down
108 changes: 105 additions & 3 deletions Rdmp.Core/DataExport/DataExtraction/Pipeline/Destinations/MSSqlMergeDestination.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,13 @@ public MSSqlMergeDestination() : base(false)
{
}

private bool hasStructuralChanges(DataTable source, DiscoveredTable destination)
{
var sourceColumns = source.Columns.Cast<DataColumn>().Select(c => c.ColumnName).ToList();
var destinationColumns = destination.DiscoverColumns().Select(c => c.GetRuntimeName()).ToList();
return !sourceColumns.All(destinationColumns.Contains) || !destinationColumns.All(sourceColumns.Contains);
}

private string GetTableName(string suffix, DataTable dt)
{
string tblName = TableNamingPattern;
Expand Down Expand Up @@ -136,7 +143,7 @@ public override void Abort(IDataLoadEventListener listener)



public static string GetMergeSQL(string destinationTableName,string tempTableName, DataColumn[] pkColumns, DataColumn[] nonPkColumns, bool performDeletes,IQuerySyntaxHelper syntaxHelper)
public static string GetMergeSQL(string destinationTableName, string tempTableName, DataColumn[] pkColumns, DataColumn[] nonPkColumns, bool performDeletes, IQuerySyntaxHelper syntaxHelper)
{
List<DataColumn> columns = new();
columns.AddRange(pkColumns);
Expand Down Expand Up @@ -230,6 +237,94 @@ protected override void Open(DataTable toProcess, IDataLoadEventListener job, Gr
Destination PKs: {string.Join(", ", remotePKs)}
""");
}
TriggerImplementerFactory triggerFactory = new TriggerImplementerFactory(FAnsi.DatabaseType.MicrosoftSQLServer);
var implementor = triggerFactory.Create(existing);
bool triggerPresent;
try
{
triggerPresent = implementor.GetTriggerStatus() == DataLoad.Triggers.TriggerStatus.Enabled;
}
catch (TriggerMissingException)
{
triggerPresent = false;
}
if (hasStructuralChanges(toProcess, existing))
{
var sourceColumns = toProcess.Columns.Cast<DataColumn>().Select(c => c.ColumnName).ToList();
var destinationColumns = existing.DiscoverColumns().Select(c => c.GetRuntimeName()).ToList();

if (triggerPresent && destinationColumns.Except(sourceColumns).Where(c => !SpecialFieldNames.IsHicPrefixed(c)).Any())//only mess about with column removal if there is an archive trigger
{

//move everything into the archive - do this by updating the HIC_validfrom
var sql = $"UPDATE {existing.GetFullyQualifiedName()} set {SpecialFieldNames.ValidFrom} = GETDATE()";
using var con = targetDb.Server.GetConnection();
con.Open();
using var cmd = targetDb.Server.GetCommand(sql, con);
cmd.CommandTimeout = 30000;
cmd.ExecuteNonQuery();

var removedColumns = destinationColumns.Except(sourceColumns).Where(c => !SpecialFieldNames.IsHicPrefixed(c));
foreach (var column in removedColumns.Select(c => existing.DiscoverColumn(c)))
{
existing.DropColumn(column);
}
string triggerProblems = "";
string triggerOK = "";
implementor.DropTrigger(out triggerProblems, out triggerOK);
if (triggerProblems != "")
{
throw new Exception(triggerProblems);
}

existing = targetDb.ExpectTable(tblName);
implementor = triggerFactory.Create(existing);
try
{
triggerPresent = implementor.GetTriggerStatus() == DataLoad.Triggers.TriggerStatus.Enabled;
}
catch (TriggerMissingException)
{
triggerPresent = false;
}
}
if (triggerPresent && sourceColumns.Except(destinationColumns).Where(c => !SpecialFieldNames.IsHicPrefixed(c)).Any())
{
var addedColumns = sourceColumns.Except(destinationColumns).Where(c => !SpecialFieldNames.IsHicPrefixed(c));
var archiveTable = existing.Database.ExpectTable(tblName + "_Archive");
foreach (var column in addedColumns)
{
var colType = toProcess.Columns[column].DataType;
existing.AddColumn(column, new TypeGuesser.DatabaseTypeRequest(colType), true, 30000);
if (archiveTable.Exists() && archiveTable.DiscoverColumns().All(col => col.GetRuntimeName() != column))
{
archiveTable.AddColumn(column, new TypeGuesser.DatabaseTypeRequest(toProcess.Columns[column].DataType), true, 30000);
}
}
string triggerProblems = "";
string triggerOK = "";
implementor.DropTrigger(out triggerProblems, out triggerOK);
if (triggerProblems != "")
{
throw new Exception(triggerProblems);
}

existing = targetDb.ExpectTable(tblName);
implementor = triggerFactory.Create(existing);
try
{
triggerPresent = implementor.GetTriggerStatus() == DataLoad.Triggers.TriggerStatus.Enabled;
}
catch (TriggerMissingException)
{
triggerPresent = false;
}
}
}
if (!triggerPresent && UseArchiveTrigger)
{
implementor.CreateTrigger(ThrowImmediatelyCheckNotifier.Quiet);
}
}
}
_toProcess = toProcess;
Expand Down Expand Up @@ -287,7 +382,7 @@ protected override void WriteRows(DataTable toProcess, IDataLoadEventListener jo
if (UseArchiveTrigger)
{
TriggerImplementerFactory triggerFactory = new TriggerImplementerFactory(FAnsi.DatabaseType.MicrosoftSQLServer);
var implementor = triggerFactory.Create(destinationTable, false, true);
var implementor = triggerFactory.Create(destinationTable);
bool present;
try
{
Expand All @@ -299,7 +394,14 @@ protected override void WriteRows(DataTable toProcess, IDataLoadEventListener jo
}
if (!present)
{
implementor.CreateTrigger(ThrowImmediatelyCheckNotifier.Quiet);
try
{
implementor.CreateTrigger(ThrowImmediatelyCheckNotifier.Quiet);
}
catch (Exception e)
{
job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Error, $"Failed to create archive trigger on {destinationTable.GetFullyQualifiedName()}: {e.Message}"));
}
}
}

Expand Down
41 changes: 31 additions & 10 deletions Rdmp.Core/DataLoad/Triggers/Implementations/MicrosoftSQLTriggerImplementer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,23 +259,39 @@ private void CreateViewOldVersionsTableValuedFunction(string sqlUsedToCreateArch

//these were added during transaction so we have to specify them again here because transaction will not have been committed yet
sqlToRun = sqlToRun.Trim();
sqlToRun += $",{Environment.NewLine}";
sqlToRun += $"\thic_validTo datetime,{Environment.NewLine}";
sqlToRun += "\thic_userID varchar(128),";
sqlToRun += "\thic_status char(1)";
if (!sqlToRun.Contains("hic_validTo"))
{
sqlToRun += $",{Environment.NewLine}";
sqlToRun += $"hic_validTo datetime";
}
if (!sqlToRun.Contains("hic_userID"))
{
sqlToRun += $",{Environment.NewLine}";

sqlToRun += "hic_userID varchar(128)";
}
if (!sqlToRun.Contains("hic_status"))
{
sqlToRun += $",{Environment.NewLine}";
sqlToRun += "hic_status char(1)";
}


sqlToRun += $"){Environment.NewLine}";
sqlToRun += $"AS{Environment.NewLine}";
sqlToRun += $"BEGIN{Environment.NewLine}";
sqlToRun += Environment.NewLine;

var liveCols = _columns.Select(c => $"[{c.GetRuntimeName()}]").Union(new string[]
{
$"[{SpecialFieldNames.DataLoadRunID}]", $"[{SpecialFieldNames.ValidFrom}]"
}).Where(col => _dontAddDataLoadRunId ? col != $"[{SpecialFieldNames.DataLoadRunID}]" : true).ToArray();

var archiveCols = $"{string.Join(",", liveCols)},hic_validTo,hic_userID,hic_status";
var liveCols = _archiveTable.DiscoverColumns().Select(c => $"[{c.GetRuntimeName()}]")
.ToList();

if (!liveCols.Contains($"[{SpecialFieldNames.DataLoadRunID}]")) liveCols.Add($"[{SpecialFieldNames.DataLoadRunID}]");
if (!liveCols.Contains($"[{SpecialFieldNames.ValidFrom}]")) liveCols.Add($"[{SpecialFieldNames.ValidFrom}]");
liveCols = liveCols.Where(col => !_dontAddDataLoadRunId || col != $"[{SpecialFieldNames.DataLoadRunID}]").ToList();

var archiveCols = $"{string.Join(",", liveCols)}";

var cDotArchiveCols = string.Join(",", liveCols.Select(s => $"c.{s}"));


Expand All @@ -285,9 +301,14 @@ private void CreateViewOldVersionsTableValuedFunction(string sqlUsedToCreateArch
", '1899/01/01') AND hic_validTo" + Environment.NewLine, _archiveTable);
sqlToRun += Environment.NewLine;

//these are columns that exist in the archive, but not the table. So have to be set as NULL as <value>
var nullColumns = _archiveTable.DiscoverColumns().Select(c => $"[{c.GetRuntimeName()}]").Except(_table.DiscoverColumns().Select(c => $"[{c.GetRuntimeName()}]")).ToList();

cDotArchiveCols = string.Join(",", liveCols.Select(s => nullColumns.Contains(s) ? $"NULL AS {s}" : $"c.{s}"));

sqlToRun += $"\tINSERT @returntable{Environment.NewLine}";
sqlToRun +=
$"\tSELECT {cDotArchiveCols},NULL AS hic_validTo, NULL AS hic_userID, 'C' AS hic_status{Environment.NewLine}"; //c is for current
$"\tSELECT {cDotArchiveCols}";
sqlToRun += string.Format("\tFROM [{0}] c" + Environment.NewLine, _table.GetRuntimeName());
sqlToRun += $"\tLEFT OUTER JOIN @returntable a ON {Environment.NewLine}";

Expand Down
Loading
Loading