diff --git a/docs/Historian.md b/docs/Historian.md index 8878bffe..77368e11 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -1842,6 +1842,10 @@ equivalent TTL (`QuestDb:RetentionDays`); the Gateway backend's retention is the `DELETE TOP (@n) FROM SampleModified WHERE TagId = @id AND SourceTimeUtc < @cutoff` (#668; same **source-time** cutoff, clustered seek; the purge itself never writes a trail row) — then events with `DELETE TOP (@n) ... WHERE EventTimeUtc < @cutoff` (seek on `CX_Event_Time`), each looped until a batch comes back short. Cancellation is honoured between batches. The `hist.Tag` catalogue is never purged. + **Only eligible tags get deletes:** one pre-filter query per table (`SELECT TagId FROM Tag t WHERE EXISTS (SELECT 1 FROM Sample s + WHERE s.TagId = t.TagId AND s.SourceTimeUtc < @cutoff)`, a key seek per tag inside one statement) picks the tags holding an + eligible row; the per-tag loop used to send two `DELETE TOP` statements for every tag on every pass — 100 000 for 50 000 tags with + nothing to delete, all under the applock. - **Shared databases.** Several pairs may point at one database; a session `sp_getapplock` (`OtOpcUa.Historian.Retention.`, timeout 0) lets one purger run and the others skip that tick (Debug log). - **Observability.** Counter `otopcua.historian.retention.purged{table=Sample|SampleModified|Event}`; a failed pass logs a Warning and retries next diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs index 7e023878..593d8cde 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs @@ -9,7 +9,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; /// /// Deletes hist.Sample / hist.SampleModified / hist.Event rows older than the retention window, in small /// DELETE TOP(n) batches (under the lock-escalation threshold). Samples are purged per tag so each -/// delete is a seek on the clustered key (TagId, SourceTimeUtc); events seek on CX_Event_Time. +/// delete is a seek on the clustered key (TagId, SourceTimeUtc) — and only for the tags a single pre-filter query +/// finds holding an eligible row, so a pass with nothing to delete costs one query per table, not two per tag; events +/// seek on CX_Event_Time. /// A session sp_getapplock (timeout 0) serialises purgers that share one database; a loser skips. /// To purge another table later, add it to . On a v1 schema (no Part 11 trail) /// the SampleModified purge is skipped and absent from the results. @@ -86,22 +88,24 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention // A v1 schema has no SampleModified table — skip its purge rather than fail the whole pass. var trail = _bootstrap.SupportsModifiedTrail; long samples = 0, modified = 0; - var tagIds = new List(); - await using (var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")}")) - await using (var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false)) - while (await r.ReadAsync(ct).ConfigureAwait(false)) tagIds.Add(r.GetInt32(0)); - - foreach (var id in tagIds) + // Pre-filter: only the tags holding at least one eligible row get delete statements — one statement, a key seek + // per tag on (TagId, SourceTimeUtc). Deleting blindly per tag sent two DELETE TOP statements per tag per pass + // (100 000 for 50 000 tags) on a database with nothing to delete, all under the applock. + var sampleTags = await EligibleTagsAsync(conn, "Sample", cutoff, ct).ConfigureAwait(false); + foreach (var id in sampleTags) { samples += await DeleteLoopAsync(conn, - $"DELETE TOP (@n) FROM {_conn.Q("Sample")} WHERE TagId = @id AND SourceTimeUtc < @cutoff", + "DELETE TOP (@n) FROM " + _conn.Q("Sample") + " WHERE TagId = @id AND SourceTimeUtc < @cutoff", cutoff, id, ct).ConfigureAwait(false); - // The Part 11 modification trail ages out on the SAME source-time cutoff as the samples it - // describes; a purge never writes to it (no trigger, no OUTPUT INTO). - if (trail) + } + // The Part 11 modification trail ages out on the SAME source-time cutoff as the samples it + // describes; a purge never writes to it (no trigger, no OUTPUT INTO). + if (trail) + { + foreach (var id in await EligibleTagsAsync(conn, "SampleModified", cutoff, ct).ConfigureAwait(false)) { modified += await DeleteLoopAsync(conn, - $"DELETE TOP (@n) FROM {_conn.Q("SampleModified")} WHERE TagId = @id AND SourceTimeUtc < @cutoff", + "DELETE TOP (@n) FROM " + _conn.Q("SampleModified") + " WHERE TagId = @id AND SourceTimeUtc < @cutoff", cutoff, id, ct).ConfigureAwait(false); } } @@ -131,6 +135,19 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention } } + /// The tags with at least one row older than . + private async Task> EligibleTagsAsync(SqlConnection conn, string table, DateTime cutoff, CancellationToken ct) + { + var ids = new List(); + await using var cmd = _conn.CreateCommand(conn, + "SELECT t.TagId FROM " + _conn.Q("Tag") + " t WHERE EXISTS (SELECT 1 FROM " + _conn.Q(table) + + " s WHERE s.TagId = t.TagId AND s.SourceTimeUtc < @cutoff)"); + SqlHistorianConnection.AddDateTime2(cmd, "@cutoff", cutoff); + await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await r.ReadAsync(ct).ConfigureAwait(false)) ids.Add(r.GetInt32(0)); + return ids; + } + private async Task DeleteLoopAsync(SqlConnection conn, string sql, DateTime cutoff, int? tagId, CancellationToken ct) { long total = 0; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs index 8fe343ad..8462496c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs @@ -92,6 +92,43 @@ public sealed class RetentionTests(SqlHistorianFixture fx) (await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag")).ShouldBe(3); } + [Fact] + public async Task Only_tags_with_eligible_rows_get_delete_statements() + { + // 50 tags with only fresh data + 1 tag with old samples and an old trail row: the pre-filter must send deletes for + // that one tag only — it used to issue two DELETE TOP statements per tag per pass even with nothing to delete. + fx.SkipIfUnavailable(); + var (c, b, _, s) = Make(); + await b.EnsureAsync(CancellationToken.None); + var old = Now.AddDays(-40).ToString("yyyy-MM-ddTHH:mm:ss"); + var fresh = Now.AddDays(-1).ToString("yyyy-MM-ddTHH:mm:ss"); + await ExecAsync( + $"DECLARE @t int = 0; WHILE @t <= 50 BEGIN INSERT [{s}].Tag (Name, DataType, CreatedUtc, UpdatedUtc) VALUES (CONCAT(N'tag', @t), 1, SYSUTCDATETIME(), SYSUTCDATETIME()); SET @t += 1; END;" + + $"INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, '{fresh}', 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag;" + + $"INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, '{old}', 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag WHERE Name = N'tag7';" + + $"INSERT [{s}].SampleModified (TagId, SourceTimeUtc, ModificationTimeUtc, UpdateType, Principal) SELECT TagId, '{old}', SYSUTCDATETIME(), 1, N'p' FROM [{s}].Tag WHERE Name = N'tag9';"); + + var deletes = 0; + var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now)) + { + AfterBatch = () => deletes++, + }; + var results = await purger.PurgeAsync(CancellationToken.None); + + results.ShouldBe([ + new RetentionPurgeResult("Sample", 1), + new RetentionPurgeResult("SampleModified", 1), + new RetentionPurgeResult("Event", 0), + ]); + deletes.ShouldBe(3, "one Sample delete (tag7), one SampleModified delete (tag9), one Event delete — not 2 × 51 + 1"); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(51); + + // A pass with nothing eligible sends only the Event delete. + deletes = 0; + await purger.PurgeAsync(CancellationToken.None); + deletes.ShouldBe(1); + } + [Fact] public async Task A_second_purge_is_a_no_op() { @@ -207,14 +244,15 @@ public sealed class RetentionTests(SqlHistorianFixture fx) var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now)); var purge = purger.PurgeAsync(cts.Token); - // Cancel only once the server reports a DELETE actually blocked behind the holder — a fixed delay could - // cancel during bootstrap/applock acquisition on a slow host and never exercise the mid-command path. + // Cancel only once the server reports a purge command actually blocked behind the holder — a fixed delay could + // cancel during bootstrap/applock acquisition on a slow host and never exercise the mid-command path. The first + // command to touch Sample is the eligible-tags pre-filter (a SELECT), then the DELETEs; either is mid-command. var deadline = DateTime.UtcNow.AddSeconds(15); while (await CountAsync( - $"SELECT COUNT(*) FROM sys.dm_exec_requests WHERE blocking_session_id = {holderSpid} AND command = 'DELETE'") == 0) + $"SELECT COUNT(*) FROM sys.dm_exec_requests WHERE blocking_session_id = {holderSpid} AND command IN ('DELETE', 'SELECT')") == 0) { purge.IsCompleted.ShouldBeFalse("the purge finished without ever blocking on the holder's table lock"); - DateTime.UtcNow.ShouldBeLessThan(deadline, "the purge's DELETE never blocked behind the holder"); + DateTime.UtcNow.ShouldBeLessThan(deadline, "the purge never blocked behind the holder"); await Task.Delay(50, TestContext.Current.CancellationToken); } cts.Cancel();