diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs index 27e4f416..f4e00820 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs @@ -31,9 +31,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater /// Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit). internal const int BatchSize = 1000; - /// Request times further apart than this are read with separate range queries (I-2). - internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1); - private const long ResolutionTicks = 10; private const int LockStripes = 64; private readonly SemaphoreSlim[] _locks = Enumerable.Range(0, LockStripes).Select(_ => new SemaphoreSlim(1, 1)).ToArray(); @@ -60,26 +57,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater /// Rows returned by existence reads so far (test/diagnostic counter). internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead); - /// Splits times into clusters wherever consecutive sorted times are further apart than . - /// Request times (any order). - /// Largest gap kept inside one cluster. - /// Inclusive [Lo, Hi] ranges, ascending. - internal static List<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList times, TimeSpan gap) - { - var result = new List<(DateTime, DateTime)>(); - if (times.Count == 0) return result; - var sorted = times.OrderBy(t => t).ToList(); - var lo = sorted[0]; - var prev = lo; - foreach (var t in sorted.Skip(1)) - { - if (t - prev > gap) { result.Add((lo, prev)); lo = t; } - prev = t; - } - result.Add((lo, prev)); - return result; - } - private async Task LockAsync(string tag, CancellationToken ct) { var sem = _locks[(int)((uint)StringComparer.Ordinal.GetHashCode(tag) % LockStripes)]; @@ -231,7 +208,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater { var live = new HashSet(); await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); - foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap)) + foreach (var (lo, hi) in QuestDbTimeClusters.Cluster(times, QuestDbTimeClusters.ExistenceGap)) { await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted"); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs index 791ef010..a7fba836 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs @@ -98,49 +98,13 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter } } - /// A gap between consecutive batch times larger than this starts a new existence-query range. - internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1); - - /// - /// Splits ascending into inclusive [Lo, Hi] ranges: a new range starts where the gap to the - /// previous time exceeds or the range already holds times. One outlier - /// time therefore costs its own tiny query instead of widening one range over every stored row in between. - /// - /// Times in ascending order. - /// Largest gap kept inside one range. - /// Most times in one range (>= 1). - /// The ranges, ascending. - internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList sortedTimes, TimeSpan gap, int maxPerCluster) - { - ArgumentOutOfRangeException.ThrowIfLessThan(maxPerCluster, 1); - var result = new List<(DateTime, DateTime)>(); - if (sortedTimes.Count == 0) return result; - var lo = sortedTimes[0]; - var prev = lo; - var n = 1; - for (var i = 1; i < sortedTimes.Count; i++) - { - var t = sortedTimes[i]; - if (t - prev > gap || n >= maxPerCluster) - { - result.Add((lo, prev)); - lo = t; - n = 0; - } - prev = t; - n++; - } - result.Add((lo, prev)); - return result; - } - // T0 Q11 range form, one query per cluster: tombstones are NOT filtered out (a deleted row must not be resurrected). private async Task> ExistingTicksAsync(string tag, List rows, CancellationToken ct) { var times = rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).OrderBy(t => t).ToList(); var set = new HashSet(); await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); - foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap, BatchRows)) + foreach (var (lo, hi) in QuestDbTimeClusters.Cluster(times, QuestDbTimeClusters.ExistenceGap, BatchRows)) { await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3"); QuestDbConnection.BindText(cmd, tag); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs new file mode 100644 index 00000000..442e7b34 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs @@ -0,0 +1,49 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; + +/// +/// Groups the times of one write or HistoryUpdate request into inclusive [Lo, Hi] ranges so that the +/// "which of these times already exist" check reads one range query per cluster instead of one query spanning +/// [min, max] of the whole request. One far-away outlier time then costs its own tiny query instead of +/// widening a single range over every stored row in between. Shared by the live-value writer and the updater. +/// +internal static class QuestDbTimeClusters +{ + /// Request times further apart than this start a new existence-query range. + internal static readonly TimeSpan ExistenceGap = TimeSpan.FromMinutes(1); + + /// + /// Splits (any order; sorted here) into inclusive ranges, ascending: a new range starts + /// where the gap to the previous time exceeds (exactly stays + /// together), or where the current range already holds times. + /// + /// Request times, in any order. + /// Largest gap kept inside one range. + /// Most times in one range (>= 1); unbounded by default. + /// The ranges, ascending; empty for no times. + internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster( + IReadOnlyList times, TimeSpan gap, int maxPerCluster = int.MaxValue) + { + ArgumentNullException.ThrowIfNull(times); + ArgumentOutOfRangeException.ThrowIfLessThan(maxPerCluster, 1); + var result = new List<(DateTime, DateTime)>(); + if (times.Count == 0) return result; + var sorted = times.Order().ToList(); + var lo = sorted[0]; + var prev = lo; + var n = 1; + for (var i = 1; i < sorted.Count; i++) + { + var t = sorted[i]; + if (t - prev > gap || n >= maxPerCluster) + { + result.Add((lo, prev)); + lo = t; + n = 0; + } + prev = t; + n++; + } + result.Add((lo, prev)); + return result; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.Redelivery.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.Redelivery.cs new file mode 100644 index 00000000..c3b85295 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.Redelivery.cs @@ -0,0 +1,38 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; + +public sealed partial class TombstoneTests +{ + /// + /// The live writer's outbox redelivers batches (at-least-once). A batch replayed after an operator's + /// DeleteAtTime must not bring the deleted sample back: the writer's existence check counts a tombstone as + /// present (it deliberately does not filter NOT deleted), so the redelivered row is skipped. + /// + [Fact] + public async Task Tombstone_is_not_resurrected_by_a_writer_redelivery() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + var writer = new QuestDbHistorianValueWriter(rig.Conn, rig.Tags, + new QuestDbWalVisibility(rig.Conn, TimeSpan.FromSeconds(5))); + HistorizationValue[] batch = + [ + new(T0, HistorianValue.FromDouble(1), HistorianStatusCodes.Good), + new(T0.AddSeconds(1), HistorianValue.FromDouble(2), HistorianStatusCodes.Good), + ]; + + (await writer.WriteLiveValuesAsync("r", batch, default)).ShouldBeTrue(); + (await rig.Updater.DeleteAtTimeAsync("r", [T0], default)).ShouldBe([HistorianStatusCodes.Good]); + (await writer.WriteLiveValuesAsync("r", batch, default)).ShouldBeTrue(); // the redelivery + + (await UpdateTests.RowsAsync(rig, "r", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]); + var all = await UpdateTests.RowsAsync(rig, "r"); + all.Count.ShouldBe(2); + all[0].Deleted.ShouldBeTrue(); + all[0].Num.ShouldBeNull(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs index 2f3f53e3..841b73b7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; /// Soft-delete behaviour. The writer-redelivery case lives in T7 (TombstoneTests.Redelivery.cs). [Collection(QuestDbCollection.Name)] -public sealed class TombstoneTests(QuestDbFixture fx) +public sealed partial class TombstoneTests(QuestDbFixture fx) { private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs index 0f721065..4d4e52a5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs @@ -249,16 +249,6 @@ public sealed class UpdateTests(QuestDbFixture fx) } } - [Fact] - public void Cluster_splits_where_the_gap_exceeds_the_limit() - { - var gap = QuestDbHistorianDataUpdater.ExistenceClusterGap; - var c = QuestDbHistorianDataUpdater.Cluster( - [T0.AddHours(2), T0, T0.AddSeconds(30), T0.AddHours(1)], gap); - c.ShouldBe([(T0, T0.AddSeconds(30)), (T0.AddHours(1), T0.AddHours(1)), (T0.AddHours(2), T0.AddHours(2))]); - QuestDbHistorianDataUpdater.Cluster([], gap).ShouldBeEmpty(); - } - [Fact] public async Task Existence_reads_are_bounded_when_request_times_are_far_apart() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs index 39e75e1b..a090346f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs @@ -274,23 +274,4 @@ public sealed class WriteTests(QuestDbFixture fx) (await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue(); (await RowsAsync(c, "tag.far")).Count.ShouldBe(2); } - - [Fact] - public void Cluster_splits_on_a_gap_over_the_limit_and_at_the_per_cluster_cap() - { - var g = TimeSpan.FromMinutes(1); - QuestDbHistorianValueWriter.Cluster([], g, 10).ShouldBeEmpty(); - QuestDbHistorianValueWriter.Cluster([T0], g, 10).ShouldBe([(T0, T0)]); - // exactly the gap stays together; one tick over splits - QuestDbHistorianValueWriter.Cluster([T0, T0 + g], g, 10).ShouldBe([(T0, T0 + g)]); - QuestDbHistorianValueWriter.Cluster([T0, T0 + g + TimeSpan.FromTicks(10)], g, 10) - .ShouldBe([(T0, T0), (T0 + g + TimeSpan.FromTicks(10), T0 + g + TimeSpan.FromTicks(10))]); - // outlier gets its own range - var far = T0.AddYears(1); - QuestDbHistorianValueWriter.Cluster([T0, T0.AddSeconds(1), far], g, 10).ShouldBe([(T0, T0.AddSeconds(1)), (far, far)]); - // cap: 5 close times, max 2 per cluster - var five = Enumerable.Range(0, 5).Select(i => T0.AddSeconds(i)).ToList(); - QuestDbHistorianValueWriter.Cluster(five, g, 2) - .ShouldBe([(five[0], five[1]), (five[2], five[3]), (five[4], five[4])]); - } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTimeClustersTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTimeClustersTests.cs new file mode 100644 index 00000000..f68f9521 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTimeClustersTests.cs @@ -0,0 +1,62 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests; + +/// +/// The shared existence-query clustering (moved here, names unchanged, from the writer's WriteTests and the +/// updater's UpdateTests when the two private copies were consolidated into ). +/// +public sealed class QuestDbTimeClustersTests +{ + private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); + + [Fact] + public void Cluster_splits_on_a_gap_over_the_limit_and_at_the_per_cluster_cap() + { + var g = TimeSpan.FromMinutes(1); + QuestDbTimeClusters.Cluster([], g, 10).ShouldBeEmpty(); + QuestDbTimeClusters.Cluster([T0], g, 10).ShouldBe([(T0, T0)]); + // exactly the gap stays together; one tick over splits + QuestDbTimeClusters.Cluster([T0, T0 + g], g, 10).ShouldBe([(T0, T0 + g)]); + QuestDbTimeClusters.Cluster([T0, T0 + g + TimeSpan.FromTicks(10)], g, 10) + .ShouldBe([(T0, T0), (T0 + g + TimeSpan.FromTicks(10), T0 + g + TimeSpan.FromTicks(10))]); + // outlier gets its own range + var far = T0.AddYears(1); + QuestDbTimeClusters.Cluster([T0, T0.AddSeconds(1), far], g, 10).ShouldBe([(T0, T0.AddSeconds(1)), (far, far)]); + // cap: 5 close times, max 2 per cluster + var five = Enumerable.Range(0, 5).Select(i => T0.AddSeconds(i)).ToList(); + QuestDbTimeClusters.Cluster(five, g, 2) + .ShouldBe([(five[0], five[1]), (five[2], five[3]), (five[4], five[4])]); + } + + [Fact] + public void Cluster_splits_where_the_gap_exceeds_the_limit() + { + var gap = QuestDbTimeClusters.ExistenceGap; + var c = QuestDbTimeClusters.Cluster( + [T0.AddHours(2), T0, T0.AddSeconds(30), T0.AddHours(1)], gap); + c.ShouldBe([(T0, T0.AddSeconds(30)), (T0.AddHours(1), T0.AddHours(1)), (T0.AddHours(2), T0.AddHours(2))]); + QuestDbTimeClusters.Cluster([], gap).ShouldBeEmpty(); + } + + [Fact] + public void The_existence_gap_is_one_minute_and_the_default_cap_is_unbounded() + { + QuestDbTimeClusters.ExistenceGap.ShouldBe(TimeSpan.FromMinutes(1)); + var many = Enumerable.Range(0, 5000).Select(i => T0.AddMilliseconds(i)).ToList(); + QuestDbTimeClusters.Cluster(many, QuestDbTimeClusters.ExistenceGap).ShouldBe([(many[0], many[^1])]); + } + + [Fact] + public void Unsorted_input_is_sorted_before_capping() + { + var times = new[] { T0.AddSeconds(3), T0, T0.AddSeconds(2), T0.AddSeconds(1) }; + QuestDbTimeClusters.Cluster(times, TimeSpan.FromMinutes(1), 2) + .ShouldBe([(T0, T0.AddSeconds(1)), (T0.AddSeconds(2), T0.AddSeconds(3))]); + } + + [Fact] + public void A_cap_below_one_is_rejected() => + Should.Throw(() => QuestDbTimeClusters.Cluster([T0], TimeSpan.FromMinutes(1), 0)); +}