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

fix(historian): clustered existence reads + tag-predicate tests (T4 review)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 02:16:14 -04:00
parent 4c053cc4d0
commit 9d90dc9457
2 changed files with 122 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs
+47 -9
View File
@@ -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).
/// <summary>A gap between consecutive batch times larger than this starts a new existence-query range.</summary>
internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1);
/// <summary>
/// Splits ascending <paramref name="sortedTimes"/> into inclusive <c>[Lo, Hi]</c> ranges: a new range starts where the gap to the
/// previous time exceeds <paramref name="gap"/> or the range already holds <paramref name="maxPerCluster"/> times. One outlier
/// time therefore costs its own tiny query instead of widening one range over every stored row in between.
/// </summary>
/// <param name="sortedTimes">Times in ascending order.</param>
/// <param name="gap">Largest gap kept inside one range.</param>
/// <param name="maxPerCluster">Most times in one range (&gt;= 1).</param>
/// <returns>The ranges, ascending.</returns>
internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList<DateTime> 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<HashSet<long>> ExistingTicksAsync(string tag, List<HistorianSampleRow> 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<long>();
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;
}