From 9d90dc94577b38b09abe2d0dd2c63d992a19b650 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 02:16:14 -0400 Subject: [PATCH] fix(historian): clustered existence reads + tag-predicate tests (T4 review) Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- .../QuestDbHistorianValueWriter.cs | 56 +++++++++++--- .../WriteTests.cs | 75 +++++++++++++++++++ 2 files changed, 122 insertions(+), 9 deletions(-) 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 de1d790c..791ef010 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,19 +98,57 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter } } - // T0 Q11 range form: tombstones are NOT filtered out (a deleted row must not be resurrected). + /// 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 min = rows.Min(r => QuestDbConnection.Round(r.TimeUtc)); - var max = rows.Max(r => QuestDbConnection.Round(r.TimeUtc)); + 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); - 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); - QuestDbConnection.BindTimestamp(cmd, min); - QuestDbConnection.BindTimestamp(cmd, max); - await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); - while (await r.ReadAsync(ct).ConfigureAwait(false)) set.Add(QuestDbConnection.ReadUtc(r, 0).Ticks); + foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap, 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); + QuestDbConnection.BindTimestamp(cmd, lo); + QuestDbConnection.BindTimestamp(cmd, hi); + await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await r.ReadAsync(ct).ConfigureAwait(false)) set.Add(QuestDbConnection.ReadUtc(r, 0).Ticks); + } return set; } 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 3f8a40f7..39e75e1b 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 @@ -218,4 +218,79 @@ public sealed class WriteTests(QuestDbFixture fx) cts.Cancel(); (await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], cts.Token)).ShouldBeFalse(); } + + private static async Task SeedRowAsync(QuestDbConnection c, QuestDbWalVisibility wal, string tag, DateTime t, double v) + { + await using (var conn = await c.OpenAsync(default)) + { + await using var cmd = c.CreateCommand(conn, + $"INSERT INTO {c.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ($1,$2,1,$3,NULL,NULL,0,false,$4)"); + QuestDbConnection.BindTimestamp(cmd, t); + QuestDbConnection.BindText(cmd, tag); + cmd.Parameters.Add(new Npgsql.NpgsqlParameter { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Double, Value = v }); + QuestDbConnection.BindTimestamp(cmd, t); + await cmd.ExecuteNonQueryAsync(); + } + await wal.WaitAsync(c.SampleTable, default); + } + + [Fact] + public async Task A_row_on_another_tag_at_the_same_timestamp_does_not_suppress_the_write() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5)); + (await w.WriteLiveValuesAsync("tag.A", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.B", [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + (await RowsAsync(c, "tag.A")).Single().N.ShouldBe(1d); + (await RowsAsync(c, "tag.B")).Single().N.ShouldBe(2d); + } + + [Fact] + public async Task An_absent_interior_timestamp_is_written_while_both_present_ends_are_left_alone() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5)); + await w.WriteLiveValuesAsync("tag.ends", [new(T0.AddSeconds(100), HistorianValue.FromDouble(0), 0)], CancellationToken.None); // creates the tag/tables + await SeedRowAsync(c, wal, "tag.ends", T0, 10); + await SeedRowAsync(c, wal, "tag.ends", T0.AddSeconds(20), 30); + (await w.WriteLiveValuesAsync("tag.ends", + [new(T0, HistorianValue.FromDouble(-1), 0), new(T0.AddSeconds(10), HistorianValue.FromDouble(20), 0), new(T0.AddSeconds(20), HistorianValue.FromDouble(-3), 0)], + CancellationToken.None)).ShouldBeTrue(); + var rows = (await RowsAsync(c, "tag.ends")).Where(r => r.T <= T0.AddSeconds(20)).ToList(); + rows.Select(r => r.N).ShouldBe([10d, 20d, 30d]); + } + + [Fact] + public async Task An_outlier_timestamp_does_not_widen_the_existence_read_over_the_stored_rows_between() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + // Behavioural check of the whole path with a far-apart batch: both samples land, a redelivery adds nothing. + var far = T0.AddYears(1); + HistorizationValue[] vals = [new(T0, HistorianValue.FromDouble(1), 0), new(far, HistorianValue.FromDouble(2), 0)]; + (await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue(); + (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])]); + } }