Merge pull request #2322 from HicServices/task/update-sql-merge

add merge timeout

Author: rdteviotdale

CHANGELOG.md

@@ -4,7 +4,7 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [9.2.0] - 2026-03-09
+## [9.2.0] - 2026-03-10
 - Add Internal Note to Catalogue
 - Fix issue where project associations were not copied when a CIC was cloned
 - Fix issue with using Internal Catalogues in Cohort Identification Configurations

Rdmp.Core/DataExport/DataExtraction/Pipeline/Destinations/MSSqlMergeDestination.cs

@@ -64,6 +64,9 @@ You must have either $a or $d
          ", Mandatory = true, DefaultValue = "$c_$d")]
         public string TableNamingPattern { get; set; }
 
+        [DemandsInitialization("How long should the merge command be allowed to run before timing out, in seconds", DefaultValue = 30000)]
+        public int SQLMergeTimeout { get; set; }
+
 
         private DiscoveredDatabase db;
         private DataTable _toProcess;
@@ -163,7 +166,7 @@ protected override void Open(DataTable toProcess, IDataLoadEventListener job, Gr
 
         protected override void WriteRows(DataTable toProcess, IDataLoadEventListener job, GracefulCancellationToken cancellationToken, Stopwatch stopwatch)
         {
-            _toProcess= toProcess;
+            _toProcess = toProcess;
             var discoveredServer = DataAccessPortal.ExpectServer(TargetDatabaseServer, DataAccessContext.DataExport, false);
 
             //sort out the naming 
@@ -182,7 +185,7 @@ protected override void WriteRows(DataTable toProcess, IDataLoadEventListener jo
                 db.Create();
 
 
-            var tableName = GetTableName(null,toProcess);
+            var tableName = GetTableName(null, toProcess);
             var destinationTable = db.ExpectTable(tableName);
 
             //ensure there are some pks
@@ -254,10 +257,11 @@ protected override void WriteRows(DataTable toProcess, IDataLoadEventListener jo
             //merge
             var tmpTbl = db.CreateTable(
                 out Dictionary<string, Guesser> _dataTypeDictionary,
-                $"mergeTempTable_{tableName}_{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff",CultureInfo.InvariantCulture).Replace('.','-')}",
-                toProcess, null,true, null);
+                $"mergeTempTable_{tableName}_{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture).Replace('.', '-')}",
+                toProcess, null, true, null);
             var _managedConnection = tmpTbl.Database.Server.GetManagedConnection();
             var _bulkcopy = tmpTbl.BeginBulkInsert(CultureInfo.CurrentCulture, _managedConnection.ManagedTransaction);
+            _bulkcopy.Timeout = SQLMergeTimeout;
             _bulkcopy.Upload(toProcess);
             _bulkcopy.Dispose();
             var mergeSql = $"""
@@ -268,17 +272,18 @@ WHEN MATCHED THEN
                     UPDATE SET {string.Join(" , ", nonPkColumns.Select(pkc => $"target.{pkc.ColumnName} = source.{pkc.ColumnName}"))}
                 WHEN NOT MATCHED BY TARGET THEN
                     INSERT ({string.Join(" , ", toProcess.Columns.Cast<DataColumn>().Select(pkc => pkc.ColumnName))})
-                    VALUES ({string.Join(" , ", toProcess.Columns.Cast<DataColumn>().Select(pkc => $"source.{pkc.ColumnName}"))}){(AllowMergeToPerformDeletes?"":";")}
-                {(AllowMergeToPerformDeletes?"""
+                    VALUES ({string.Join(" , ", toProcess.Columns.Cast<DataColumn>().Select(pkc => $"source.{pkc.ColumnName}"))}){(AllowMergeToPerformDeletes ? "" : ";")}
+                {(AllowMergeToPerformDeletes ? """
                 WHEN NOT MATCHED BY SOURCE THEN
                     DELETE;
-                """:"")}
+                """ : "")}
                 """;
             job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information, $"{mergeSql}"));
             var cmd = new SqlCommand(mergeSql, (SqlConnection)_managedConnection.Connection);
+            cmd.CommandTimeout = SQLMergeTimeout;
             var rowCount = cmd.ExecuteNonQuery();
             job.OnNotify(this, new NotifyEventArgs(ProgressEventType.Information, $"Merged {rowCount} rows into {destinationTable.GetFullyQualifiedName()}."));
-            if(DeleteMergeTempTable)tmpTbl.Drop();
+            if (DeleteMergeTempTable) tmpTbl.Drop();
             _managedConnection.Dispose();
         }