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

feat(historian): #677 Part 13/SDK reverse anchoring, start-of-data Partial, AggregateConfiguration, IntegralDivisor

Browse Source 
- BucketPlan: reverse processed reads anchored at the later time for every
  aggregate: window (end, start], partial interval earliest, results stamped
  with the later edge, latest first (SDK CreateSlice/GetTimestamp/backward
  GetValues). SqlServer adds @bucketAdjustTicks + flips the window predicate;
  QuestDB binds BucketBaseUnixTicks. IndexOf is the C# statement of the math.
- Partial bit on the interval the tag's data starts inside (SDK m_startOfData),
  every aggregate but Average; per-sample reads fetch the earliest sample.
- Node manager: non-default client AggregateConfiguration =>
  BadAggregateConfigurationRejected per node (ServerAggregateConfiguration,
  pinned to the SDK AggregateManager defaults).
- GatewayTagProvisioner sends IntegralDivisor = 1.
- SDK parity oracle (Opc.Ua.Server calculators) in unit + both backends'
  conformance suites, forward and reverse; remaining divergences documented.
- docs/Historian.md + CLAUDE.md release note (client-visible).
This commit is contained in:
Joseph Doherty
2026-10-11 11:26:00 -04:00
parent 825393299e
commit 27ca6cd791
24 changed files with 1134 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs
+30 -21
View File
@@ -425,12 +425,13 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
/// <remarks>
/// The (rounded) window is planned by <see cref="BucketPlan.Create"/>: bucket <c>i</c> =
/// <c>[start + i·interval, start + (i+1)·interval)</c>, start-inclusive / end-exclusive, the last bucket truncated
/// at <c>end</c>, interval &lt;= 0 ⇒ one bucket, <c>start == end</c> ⇒ none, a reversed window computed forward
/// and returned descending, at most <see cref="BucketPlan.MaxBuckets"/> buckets. One statement per node returns
/// the <see cref="BucketRow"/> statistics of each non-empty bucket (bucket index
/// <c>(cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks</c>, never <c>SAMPLE BY</c>);
/// <see cref="AggregateEvaluator.Evaluate"/> produces each value and <see cref="BucketPlan.Assemble"/> fills the
/// empty buckets with <c>BadNoData</c>. The time-weighted family (#671) instead streams the window plus four bounding
/// at <c>end</c>, interval &lt;= 0 ⇒ one bucket, <c>start == end</c> ⇒ none, at most
/// <see cref="BucketPlan.MaxBuckets"/> buckets. A reversed window (#677) is anchored at its later time per Part 13 /
/// the SDK: window <c>(end, start]</c>, buckets stepping back from <c>start</c>, the partial one earliest, each stamped
/// with its later edge, returned latest first. One statement per node returns the <see cref="BucketRow"/> statistics
/// of each non-empty bucket (bucket index <c>(cast(ts AS LONG)*10 − BucketBaseUnixTicks) / IntervalTicks</c>, never
/// <c>SAMPLE BY</c>); a second fetches the tag's earliest sample for the start-of-data Partial bit; and
/// <see cref="AggregateEvaluator.EvaluateAll"/> produces the values and fills the empty buckets with <c>BadNoData</c>. The time-weighted family (#671) instead streams the window plus four bounding
/// rows into <see cref="TimeWeightedAggregator"/>, exactly as the SQL Server backend does.
/// </remarks>
/// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not one of the seventeen served.</exception>
@@ -446,7 +447,6 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
try
{
var now = _time.GetUtcNow().UtcDateTime;
var evaluated = new Dictionary<long, DataValueSnapshot>();
if (family == AggregateFamily.TimeWeighted)
{
var tw = new TimeWeightedAggregator(aggregate, plan);
@@ -456,31 +456,40 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
RecordSuccess();
return new HistoryReadResult(twResult, ContinuationPoint: null);
}
var rows = new List<BucketRow>();
DateTime? earliest = null;
if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
{
var sample = _conn.SampleTable;
var sql = family switch
{
AggregateFamily.Simple => QuestDbAggregateQueries.Simple(sample),
AggregateFamily.StartEnd => QuestDbAggregateQueries.StartEnd(sample),
_ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime),
AggregateFamily.Simple => QuestDbAggregateQueries.Simple(sample, plan.Reversed),
AggregateFamily.StartEnd => QuestDbAggregateQueries.StartEnd(sample, plan.Reversed),
_ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed),
};
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, sql);
QuestDbConnection.BindLong(cmd, plan.StartUnixTicks);
QuestDbConnection.BindLong(cmd, plan.IntervalTicks);
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, plan.StartUtc);
QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
await using (var cmd = _conn.CreateCommand(conn, sql))
{
var row = ReadBucket(reader, family);
evaluated[row.Bucket] = AggregateEvaluator.Evaluate(aggregate, row, plan.BucketStart(row.Bucket), now);
QuestDbConnection.BindLong(cmd, plan.BucketBaseUnixTicks);
QuestDbConnection.BindLong(cmd, plan.IntervalTicks);
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, plan.StartUtc);
QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
rows.Add(ReadBucket(reader, family));
}
// The tag's earliest live sample: the start-of-data Partial rule (#677).
await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.EarliestSample(sample)))
{
QuestDbConnection.BindText(cmd, fullReference);
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);
}
}
var result = plan.Assemble(evaluated, bucketStart => AggregateEvaluator.NoData(bucketStart, now));
var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, now);
RecordSuccess();
return new HistoryReadResult(result, ContinuationPoint: null);