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

refactor(historian): one QuestDbTimeClusters helper for writer + updater existence reads; redelivery tombstone test

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 02:32:11 -04:00
parent d2be213e86
commit 3244f9781a
8 changed files with 152 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs
+1 -24
View File
@@ -31,9 +31,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
/// <summary>Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit).</summary>
internal const int BatchSize = 1000;
/// <summary>Request times further apart than this are read with separate range queries (I-2).</summary>
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
/// <summary>Rows returned by existence reads so far (test/diagnostic counter).</summary>
internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead);
/// <summary>Splits times into clusters wherever consecutive sorted times are further apart than <paramref name="gap"/>.</summary>
/// <param name="times">Request times (any order).</param>
/// <param name="gap">Largest gap kept inside one cluster.</param>
/// <returns>Inclusive [Lo, Hi] ranges, ascending.</returns>
internal static List<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList<DateTime> 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<IDisposable> 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<long>();
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");
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs
+1 -37
View File
@@ -98,49 +98,13 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter
}
}
/// <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 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);
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);
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs
+49
View File
@@ -0,0 +1,49 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <summary>
/// Groups the times of one write or HistoryUpdate request into inclusive <c>[Lo, Hi]</c> ranges so that the
/// "which of these times already exist" check reads one range query per cluster instead of one query spanning
/// <c>[min, max]</c> 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.
/// </summary>
internal static class QuestDbTimeClusters
{
/// <summary>Request times further apart than this start a new existence-query range.</summary>
internal static readonly TimeSpan ExistenceGap = TimeSpan.FromMinutes(1);
/// <summary>
/// Splits <paramref name="times"/> (any order; sorted here) into inclusive ranges, ascending: a new range starts
/// where the gap to the previous time exceeds <paramref name="gap"/> (exactly <paramref name="gap"/> stays
/// together), or where the current range already holds <paramref name="maxPerCluster"/> times.
/// </summary>
/// <param name="times">Request times, in any order.</param>
/// <param name="gap">Largest gap kept inside one range.</param>
/// <param name="maxPerCluster">Most times in one range (&gt;= 1); unbounded by default.</param>
/// <returns>The ranges, ascending; empty for no times.</returns>
internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(
IReadOnlyList<DateTime> 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;
}
}