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

fix(historian): #680 lane B — per-sample Count/Start/End/Delta per Part 13 + forward end-of-data Partial

Browse Source 
Count: Good samples only, value-based status (all Bad => generic Bad), 0/Good on an
empty interval inside the tag's data range [first, last] (both directions).
Start/End: the raw first/last sample of any quality (own value, status, timestamp).
Delta: Uncertain only when the interval's first or last sample is not usable.
Forward end-of-data Partial (SDK m_endOfData, earlier edge inclusive) for every
per-sample aggregate but Average (PercentGood/Bad left to lane A).
Backends push down first/last-any-quality columns and fetch the tag's data range
(earliest <= window end, latest >= window start) only when NeedsDataRange.
SDK parity masks for Start/End/Delta/Count removed; new parity datasets.
This commit is contained in:
Joseph Doherty
2026-10-11 15:52:49 -04:00
parent f6a2628656
commit dbfe2fc4de
13 changed files with 644 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs
+22 -11
View File
@@ -472,7 +472,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
return new HistoryReadResult(twResult, ContinuationPoint: null);
}
var rows = new List<BucketRow>();
DateTime? earliest = null;
DateTime? earliest = null, latest = null;
if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
{
var sample = _conn.SampleTable;
@@ -494,20 +494,19 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
rows.Add(ReadBucket(reader, family));
}
// The tag's earliest live sample before the window end: the start-of-data Partial rule (#677) — forward
// reads of an aggregate other than Average only; skipped otherwise.
if (AggregateEvaluator.NeedsEarliestSample(aggregate, plan))
// The tag's data range (earliest + latest live sample): the start/end-of-data Partial rules (#677/#680 —
// forward reads of an aggregate other than Average) and Count's in-range empty interval (#680); skipped
// otherwise.
if (AggregateEvaluator.NeedsDataRange(aggregate, plan))
{
await using var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.EarliestSample(sample));
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false) && !reader.IsDBNull(0))
earliest = QuestDbConnection.ReadUtc(reader, 0);
earliest = await ReadBoundAsync(conn, QuestDbAggregateQueries.EarliestSample(sample), fullReference,
plan.EndUtc, cancellationToken).ConfigureAwait(false);
latest = await ReadBoundAsync(conn, QuestDbAggregateQueries.LatestSample(sample), fullReference,
plan.StartUtc, cancellationToken).ConfigureAwait(false);
}
}
var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, now);
var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, latest, now);
RecordSuccess();
return new HistoryReadResult(result, ContinuationPoint: null);
@@ -565,6 +564,16 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
if (bounds[3] is { } usableAfter) tw.Add(usableAfter);
}
/// <summary>Runs a one-column <c>ts</c> seek (<see cref="QuestDbAggregateQueries.EarliestSample"/> / <c>LatestSample</c>).</summary>
private async Task<DateTime?> ReadBoundAsync(NpgsqlConnection conn, string sql, string tag, DateTime edge, CancellationToken ct)
{
await using var cmd = _conn.CreateCommand(conn, sql);
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindTimestamp(cmd, edge);
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
return await reader.ReadAsync(ct).ConfigureAwait(false) && !reader.IsDBNull(0) ? QuestDbConnection.ReadUtc(reader, 0) : null;
}
private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family)
{
static double? D(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
@@ -597,6 +606,8 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
case AggregateFamily.StartEnd:
row.First = V(reader, f);
row.Last = V(reader, f + 5);
row.FirstAny = ReadStored(reader, QuestDbAggregateQueries.FirstAnyColumn);
row.LastAny = ReadStored(reader, QuestDbAggregateQueries.FirstAnyColumn + 6);
break;
default:
row.Extreme = D(reader, f);