dohertj2/lmxopcua
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(historian): page SQL Server HistoryUpdate DeleteRaw (lmxopcua#746)

Browse Source 
DeleteRaw deleted an unbounded range in one statement and transaction with
OUTPUT INTO the modification trail. Over years of a 1 Hz tag that is tens of
millions of rows deleted and audited at once: lock escalation on Sample
(blocking the live writer), transaction-log growth, and a 30 s
CommandTimeout that rolled the whole thing back.

It now runs DELETE TOP (4000) batches, each its own transaction with its
trail rows in the same statement (every deleted row is still audited),
cancellable between batches. A failure part-way leaves the committed batches
deleted and reports the failure; re-issuing completes it (documented, same
as QuestDB's paged tombstoning). The existing atomicity tests still hold per
batch.

Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22
in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-10-22 17:11:43 -04:00
parent bc2f5e1889
commit 9c3c9b0d61
3 changed files with 118 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataUpdater.cs
+40 -12
View File
@@ -5,7 +5,7 @@ using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <summary>
/// OPC UA HistoryUpdate against the SQL Server sample table (spec §4 "Updates"). One transaction per call;
/// OPC UA HistoryUpdate against the SQL Server sample table (spec §4 "Updates"). One transaction per call (DeleteRaw: per batch);
/// exactly one status per input item, in input order.
/// <para><b>Duplicate timestamps within one request</b> (compared at full 100 ns tick precision):
/// Insert — the first occurrence is applied, later ones are <c>BadEntryExists</c>; Replace/Update — last wins,
@@ -22,6 +22,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// is written.</para>
/// <para><b>Unknown tag.</b> Insert/Update create it via <see cref="SqlServerTagCatalog"/>; Replace and the
/// deletes report <c>BadNoEntryExists</c> (DeleteRaw: <c>BadNoData</c>) without creating anything.</para>
/// <para><b>DeleteRaw is paged</b> (<see cref="DeleteRawBatchSize"/> rows per batch, one transaction each, the trail
/// rows in the same statement), so a range over years of data never runs as one giant statement/transaction.</para>
/// <para><b>Modified-value history (Part 11).</b> Every row the MERGE inserts or updates and every row a
/// delete removes is copied into <c>SampleModified</c> by an <c>OUTPUT … INTO</c> on the SAME statement — the prior
/// value from <c>deleted.*</c> (NULL for an insert), the caller's <see cref="HistoryUpdateContext"/>, the
@@ -96,23 +98,49 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument;
var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
if (tagId is null) return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0);
// Paged: DELETE TOP (n) batches, each its own transaction with its trail rows (OUTPUT INTO on the same statement, so
// every deleted row still has its audit row). One statement over years of a 1 Hz tag deleted — and audited — tens
// of millions of rows in one transaction: lock escalation on Sample (blocking the live writer), transaction-log
// growth, and a CommandTimeout that rolled the whole thing back. A failure part-way leaves the batches already
// committed deleted (each with its trail rows) and is reported as the call's failure; re-issuing the same DeleteRaw
// completes it.
long rows = 0;
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
int rows;
await using (var cmd = _conn.CreateCommand(conn,
$"DELETE FROM {_conn.Q("Sample")} {AuditInto(HistoryModificationType.Delete)} " +
"WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
while (true)
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
AddAuditParameters(cmd, context);
SqlHistorianConnection.AddDateTime2(cmd, "@s", startUtc);
SqlHistorianConnection.AddDateTime2(cmd, "@e", endUtc);
rows = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
ct.ThrowIfCancellationRequested();
int batch;
await using (var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false))
{
await using (var cmd = _conn.CreateCommand(conn,
$"DELETE TOP (@n) FROM {_conn.Q("Sample")} {AuditInto(HistoryModificationType.Delete)} " +
"WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
{
cmd.Parameters.Add(new SqlParameter("@n", System.Data.SqlDbType.Int) { Value = DeleteRawBatchSize });
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
AddAuditParameters(cmd, context);
SqlHistorianConnection.AddDateTime2(cmd, "@s", startUtc);
SqlHistorianConnection.AddDateTime2(cmd, "@e", endUtc);
batch = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
await tx.CommitAsync(ct).ConfigureAwait(false);
}
rows += batch;
DeleteRawBatches++;
if (batch < DeleteRawBatchSize) break;
}
await tx.CommitAsync(ct).ConfigureAwait(false);
return HistoryUpdatePlanner.DeleteRawResult(tagKnown: true, rows);
}
/// <summary>
/// Rows per DeleteRaw batch: under SQL Server's ~5,000-lock escalation threshold (each deleted row also inserts its
/// trail row), so a range delete never takes a table lock. Settable for tests.
/// </summary>
internal int DeleteRawBatchSize { get; set; } = 4000;
/// <summary>DeleteRaw batches run so far (lifetime) — test/diagnostic.</summary>
internal int DeleteRawBatches { get; private set; }
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc,
HistoryUpdateContext context, CancellationToken ct)