feat(historian): Part 13 time-weighted aggregates on SqlServer + QuestDb (#671)
New AggregateFamily.TimeWeighted serving TimeAverage, TimeAverage2, Total, Total2 and Interpolative, evaluated by the shared Sql.Common TimeWeightedAggregator for both SQL backends (identical results by construction). HistoryAggregateType gains TimeAverage2 + Total2 (appended); OtOpcUaNodeManager.MapAggregate maps their NodeIds. Gateway's mapper already refuses unknown defined members => BadAggregateNotSupported per node. Semantics (server defaults, client AggregateConfiguration ignored): TreatUncertainAsBad=false, PercentDataBad=PercentDataGood=100, UseSlopedExtrapolation=false. Sloped for Double/Int64, stepped for Boolean; a usable sample followed by a Bad one is held up to the outage (AtTime rule); stepped Uncertain extrapolation past the last sample. TimeAverage/Total use interpolated bounding (bridge outages); TimeAverage2/Total2 exclude Bad time. Bounding values: latest usable + latest any before start, first any + first usable at/after end (four TOP 1 / LIMIT 1 seeks). CLIENT-VISIBLE: Total on SqlServer/QuestDb was SUM of samples; it is now the time-weighted integral (TimeAverage x interval seconds), matching Gateway's Integral. Release note in docs/Historian.md. Trade-off (deviation from the preferred hybrid pushdown, allowed by the design): the time-weighted family is NOT pushed down. Each read streams the window's rows plus the four bounds (SQL Server: one five-result-set batch; QuestDB: bounds UNION ALL + window select) into the C# evaluator. Network is O(samples in window), memory O(buckets). Writing the per-segment interpolation/edge splitting twice in two SQL dialects and keeping them bit-identical was judged the bigger risk; this also makes the QuestDB lag/lead spike moot. Tests: 35 hand-computed TimeWeightedAggregator unit cases (each of 10 implementation mutations goes red); served-aggregate pin 13 -> 17; conformance sparse fixture (intervals wider and narrower than sample gaps) + an outage straddling both window edges (each of the four SQL seeks disabled in turn goes red); node-manager NodeId mapping test.
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,364 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
/// <summary>
|
||||
/// The OPC UA Part 13 time-weighted aggregates (<see cref="AggregateFamily.TimeWeighted"/>): <c>TimeAverage</c>,
|
||||
/// <c>TimeAverage2</c>, <c>Total</c>, <c>Total2</c> and <c>Interpolative</c>, evaluated over a time-ordered stream of
|
||||
/// stored samples. Both SQL backends fetch the same rows and feed them here, so the two produce identical results
|
||||
/// by construction.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Input.</b> Samples in strictly ascending time: the bounding rows before the window (the latest
|
||||
/// <i>usable</i> sample before <c>start</c>, then the latest sample of any quality before <c>start</c> — they differ
|
||||
/// when an outage straddles the start), every in-window sample, then the bounding rows at/after the window end
|
||||
/// (the first sample of any quality at or after <c>end</c>, then the first <i>usable</i> one). <see cref="Add"/>
|
||||
/// ignores a sample at or before the last one, so a bound fetched twice is harmless. Memory is O(buckets), not O(samples).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The signal (server defaults: TreatUncertainAsBad = false, PercentDataBad = PercentDataGood = 100,
|
||||
/// UseSlopedExtrapolation = false; a client <c>AggregateConfiguration</c> is not honoured).</b> Between a usable
|
||||
/// sample <c>p</c> and the next sample <c>s</c>:
|
||||
/// <list type="bullet">
|
||||
/// <item><c>s</c> usable, both Double/Int64 ⇒ <b>sloped</b> (linear <c>p → s</c>); quality = worse of the two.</item>
|
||||
/// <item><c>s</c> usable, either Boolean ⇒ <b>stepped</b> (<c>p</c> held); quality = <c>p</c>'s.</item>
|
||||
/// <item><c>s</c> Bad ⇒ <c>p</c> held up to the outage (never interpolated into it, as AtTime does); Uncertain.</item>
|
||||
/// <item>no later sample at all ⇒ <c>p</c> held to the window end (stepped extrapolation); Uncertain.</item>
|
||||
/// </list>
|
||||
/// From a Bad sample (incl. an <c>Empty</c> outage marker) to the next sample, and before the first sample, the
|
||||
/// signal is Bad (no data). This is the <i>simple bounding</i> signal.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The <i>interpolated bounding</i> signal (Part 13 "interpolated bounding values") skips Bad samples: it joins
|
||||
/// consecutive <b>usable</b> samples the same way (sloped / stepped) straight across any outage, and holds the
|
||||
/// last usable value to the window end. It is undefined before the first usable sample.
|
||||
/// </para>
|
||||
/// <list type="table">
|
||||
/// <item><term>TimeAverage</term><description>∫ interpolated-bounding signal ÷ its defined duration in the interval.</description></item>
|
||||
/// <item><term>Total</term><description><c>TimeAverage × interval seconds</c> (value·s) — Part 13 §5.4.3.8.</description></item>
|
||||
/// <item><term>TimeAverage2</term><description>∫ simple-bounding signal over its non-Bad parts ÷ non-Bad duration (Bad intervals excluded).</description></item>
|
||||
/// <item><term>Total2</term><description>∫ simple-bounding signal over its non-Bad parts, in value·s (= TimeAverage2 × non-Bad seconds).</description></item>
|
||||
/// <item><term>Interpolative</term><description>The interpolated-bounding signal's value at each interval start.</description></item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// <b>Status (time-weighted four).</b> No non-Bad time in the interval ⇒ <c>BadNoData</c>; the whole interval
|
||||
/// Good ⇒ <c>Good</c>; otherwise <c>UncertainDataSubNormal</c> (PercentDataGood = PercentDataBad = 100). A
|
||||
/// usable String/DateTime sample contributing to the interval ⇒ <c>BadAggregateInvalidInputs</c>. Every status
|
||||
/// carries the Calculated bit; the timestamp is the interval start; the value is a <see cref="double"/>
|
||||
/// (Boolean = 0/1, Int64 as double).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Status (Interpolative).</b> A usable sample exactly at the interval start ⇒ its value with its own status
|
||||
/// (raw, no historian bit). Otherwise no usable sample at or before ⇒ <c>BadNoData</c>; else the interpolated
|
||||
/// value with the Interpolated bit and the status of the simple-bounding signal at that instant: <c>Good</c> on
|
||||
/// a Good segment, <c>UncertainDataSubNormal</c> on an Uncertain one, inside an outage (interpolated across it)
|
||||
/// or past the last sample (extrapolated).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class TimeWeightedAggregator
|
||||
{
|
||||
private const double TicksPerSecond = TimeSpan.TicksPerSecond;
|
||||
|
||||
private readonly HistoryAggregateType _aggregate;
|
||||
private readonly BucketPlan _plan;
|
||||
private readonly long _start;
|
||||
private readonly long _end;
|
||||
|
||||
// Simple-bounding (raw) signal, per bucket: ∫ over non-Bad time, non-Bad ticks, Good ticks.
|
||||
private readonly double[] _rawArea;
|
||||
private readonly long[] _usableTicks;
|
||||
private readonly long[] _goodTicks;
|
||||
// Interpolated-bounding (bridged) signal, per bucket.
|
||||
private readonly double[] _bridgedArea;
|
||||
private readonly long[] _bridgedTicks;
|
||||
private readonly bool[] _invalid;
|
||||
// Interpolative, per bucket start.
|
||||
private readonly DataValueSnapshot?[] _points;
|
||||
|
||||
private StoredSample? _prevAny;
|
||||
private StoredSample? _prevUsable;
|
||||
private long _nextPoint; // next bucket start not yet classified
|
||||
private readonly List<PendingPoint> _pending = []; // bucket starts awaiting the next usable sample
|
||||
private bool _finished;
|
||||
|
||||
/// <summary>
|
||||
/// What the simple-bounding signal is at a pending bucket start: no sample yet; on a segment from a usable sample
|
||||
/// whose end is not known yet; or degraded (inside an outage, or held up to one).
|
||||
/// </summary>
|
||||
private enum PointQuality { NoPrior, OnSegment, Degraded }
|
||||
|
||||
private readonly record struct PendingPoint(long Bucket, long Ticks, PointQuality Quality);
|
||||
|
||||
/// <summary>Starts an evaluation of <paramref name="aggregate"/> over <paramref name="plan"/>.</summary>
|
||||
/// <param name="aggregate">One of the <see cref="AggregateFamily.TimeWeighted"/> aggregates.</param>
|
||||
/// <param name="plan">The bucket plan (forward window).</param>
|
||||
/// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not time-weighted.</exception>
|
||||
public TimeWeightedAggregator(HistoryAggregateType aggregate, BucketPlan plan)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(plan);
|
||||
if (AggregateFamilies.FamilyOf(aggregate) != AggregateFamily.TimeWeighted)
|
||||
throw new HistoryAggregateNotSupportedException(aggregate);
|
||||
_aggregate = aggregate;
|
||||
_plan = plan;
|
||||
_start = plan.StartUtc.Ticks;
|
||||
_end = plan.EndUtc.Ticks;
|
||||
var n = (int)plan.BucketCount;
|
||||
_rawArea = new double[n];
|
||||
_usableTicks = new long[n];
|
||||
_goodTicks = new long[n];
|
||||
_bridgedArea = new double[n];
|
||||
_bridgedTicks = new long[n];
|
||||
_invalid = new bool[n];
|
||||
_points = new DataValueSnapshot?[n];
|
||||
}
|
||||
|
||||
/// <summary>Evaluates <paramref name="aggregate"/> over an already time-ordered sample sequence.</summary>
|
||||
/// <param name="aggregate">One of the <see cref="AggregateFamily.TimeWeighted"/> aggregates.</param>
|
||||
/// <param name="plan">The bucket plan.</param>
|
||||
/// <param name="samples">Bounds + window samples, ascending by time.</param>
|
||||
/// <param name="now">Server timestamp for the results.</param>
|
||||
/// <returns>One value per bucket, in the plan's order (descending for a reverse window).</returns>
|
||||
public static IReadOnlyList<DataValueSnapshot> Evaluate(
|
||||
HistoryAggregateType aggregate, BucketPlan plan, IEnumerable<StoredSample> samples, DateTime now)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(samples);
|
||||
var acc = new TimeWeightedAggregator(aggregate, plan);
|
||||
foreach (var s in samples) acc.Add(s);
|
||||
return acc.Finish(now);
|
||||
}
|
||||
|
||||
/// <summary>Feeds the next sample (ascending time; a sample at or before the previous one is ignored).</summary>
|
||||
/// <param name="s">The sample.</param>
|
||||
public void Add(StoredSample s)
|
||||
{
|
||||
if (_finished) throw new InvalidOperationException("The aggregation is already finished.");
|
||||
var t = s.TimeUtc.Ticks;
|
||||
if (_prevAny is { } last && t <= last.TimeUtc.Ticks) return;
|
||||
|
||||
// Interpolative: classify every bucket start before this sample by the simple-bounding signal there.
|
||||
ClassifyPointsBefore(t);
|
||||
|
||||
// Simple-bounding segment [prevAny, s).
|
||||
if (_prevAny is { IsUsable: true } p)
|
||||
{
|
||||
if (s.IsUsable)
|
||||
{
|
||||
var sloped = IsSloped(p, s);
|
||||
AddRaw(p, t, sloped ? s.Value.Number : p.Value.Number, sloped ? p.IsGood && s.IsGood : p.IsGood);
|
||||
}
|
||||
else
|
||||
{
|
||||
AddRaw(p, t, p.Value.Number, good: false); // held up to the outage
|
||||
}
|
||||
}
|
||||
|
||||
if (!s.IsUsable) DegradePending(); // the segment the pending points sit on runs into an outage
|
||||
|
||||
if (s.IsUsable)
|
||||
{
|
||||
if (_prevUsable is { } pu)
|
||||
{
|
||||
var sloped = IsSloped(pu, s);
|
||||
AddBridged(pu, t, sloped ? s.Value.Number : pu.Value.Number);
|
||||
}
|
||||
ResolvePending(_prevUsable, s);
|
||||
_prevUsable = s;
|
||||
}
|
||||
|
||||
_prevAny = s;
|
||||
}
|
||||
|
||||
/// <summary>Completes the evaluation (extrapolating past the last sample) and assembles the per-bucket results.</summary>
|
||||
/// <param name="now">Server timestamp for the results.</param>
|
||||
/// <returns>One value per bucket, in the plan's order (descending for a reverse window).</returns>
|
||||
public IReadOnlyList<DataValueSnapshot> Finish(DateTime now)
|
||||
{
|
||||
if (!_finished)
|
||||
{
|
||||
_finished = true;
|
||||
ClassifyPointsBefore(long.MaxValue);
|
||||
if (_prevAny is { IsUsable: true } p && p.TimeUtc.Ticks < _end)
|
||||
AddRaw(p, _end, p.Value.Number, good: false); // stepped extrapolation
|
||||
if (_prevUsable is { } pu && pu.TimeUtc.Ticks < _end)
|
||||
AddBridged(pu, _end, pu.Value.Number);
|
||||
ResolvePending(_prevUsable, next: null);
|
||||
}
|
||||
|
||||
var byBucket = new Dictionary<long, DataValueSnapshot>((int)_plan.BucketCount);
|
||||
for (long i = 0; i < _plan.BucketCount; i++)
|
||||
byBucket[i] = Result(i, now);
|
||||
return _plan.Assemble(byBucket, t => AggregateEvaluator.NoData(t, now));
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------------ Results
|
||||
|
||||
private DataValueSnapshot Result(long i, DateTime now)
|
||||
{
|
||||
var bucketStart = _plan.BucketStart(i);
|
||||
if (_aggregate == HistoryAggregateType.Interpolative)
|
||||
return _points[i] is { } point
|
||||
? point with { ServerTimestampUtc = now }
|
||||
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, bucketStart, now);
|
||||
|
||||
DataValueSnapshot Calc(object? value, uint status) =>
|
||||
new(value, HistorianInfoBits.WithCalculated(status), bucketStart, now);
|
||||
|
||||
if (_invalid[i]) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
|
||||
if (_usableTicks[i] == 0) return AggregateEvaluator.NoData(bucketStart, now);
|
||||
|
||||
var duration = BucketEnd(i) - BucketStartTicks(i);
|
||||
var status = _goodTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
|
||||
return _aggregate switch
|
||||
{
|
||||
HistoryAggregateType.TimeAverage => Calc(_bridgedArea[i] / _bridgedTicks[i], status),
|
||||
HistoryAggregateType.Total => Calc(_bridgedArea[i] / _bridgedTicks[i] * (duration / TicksPerSecond), status),
|
||||
HistoryAggregateType.TimeAverage2 => Calc(_rawArea[i] / _usableTicks[i], status),
|
||||
HistoryAggregateType.Total2 => Calc(_rawArea[i] / TicksPerSecond, status),
|
||||
_ => throw new HistoryAggregateNotSupportedException(_aggregate),
|
||||
};
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------------ Segments
|
||||
|
||||
private static bool IsNumeric(StoredSample s) =>
|
||||
s.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64 or HistorianValueKind.Boolean;
|
||||
|
||||
/// <summary>Sloped between two usable samples only when both are Double/Int64; anything Boolean is stepped.</summary>
|
||||
private static bool IsSloped(StoredSample a, StoredSample b) =>
|
||||
a.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64
|
||||
&& b.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64;
|
||||
|
||||
private long BucketStartTicks(long i) => _start + i * _plan.IntervalTicks;
|
||||
|
||||
private long BucketEnd(long i) => Math.Min(BucketStartTicks(i) + _plan.IntervalTicks, _end);
|
||||
|
||||
private void AddRaw(StoredSample from, long toTicks, double toValue, bool good) =>
|
||||
Distribute(from, toTicks, toValue, (i, area, ticks) =>
|
||||
{
|
||||
_rawArea[i] += area;
|
||||
_usableTicks[i] += ticks;
|
||||
if (good) _goodTicks[i] += ticks;
|
||||
});
|
||||
|
||||
private void AddBridged(StoredSample from, long toTicks, double toValue) =>
|
||||
Distribute(from, toTicks, toValue, (i, area, ticks) =>
|
||||
{
|
||||
_bridgedArea[i] += area;
|
||||
_bridgedTicks[i] += ticks;
|
||||
});
|
||||
|
||||
/// <summary>
|
||||
/// Splits the linear segment <c>[from.t, toTicks)</c> (value <c>from.v → toValue</c>) at the bucket edges, clipped to
|
||||
/// the window, and hands each bucket its trapezoid area (value·ticks) and duration. A non-numeric start sample
|
||||
/// marks every bucket it touches invalid instead.
|
||||
/// </summary>
|
||||
private void Distribute(StoredSample from, long toTicks, double toValue, Action<long, double, long> add)
|
||||
{
|
||||
var t0 = from.TimeUtc.Ticks;
|
||||
var lo = Math.Max(t0, _start);
|
||||
var hi = Math.Min(toTicks, _end);
|
||||
if (lo >= hi || _plan.BucketCount == 0) return;
|
||||
|
||||
var numeric = IsNumeric(from);
|
||||
var v0 = from.Value.Number;
|
||||
var span = (double)(toTicks - t0);
|
||||
double ValueAt(long x) => toValue == v0 ? v0 : v0 + (toValue - v0) * ((x - t0) / span);
|
||||
|
||||
for (var i = (lo - _start) / _plan.IntervalTicks; i < _plan.BucketCount; i++)
|
||||
{
|
||||
var bs = BucketStartTicks(i);
|
||||
if (bs >= hi) break;
|
||||
var a = Math.Max(lo, bs);
|
||||
var b = Math.Min(hi, BucketEnd(i));
|
||||
if (a >= b) continue;
|
||||
if (!numeric) { _invalid[i] = true; continue; }
|
||||
add(i, (ValueAt(a) + ValueAt(b)) / 2d * (b - a), b - a);
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------------ Interpolative
|
||||
|
||||
/// <summary>Classifies every unclassified bucket start strictly before <paramref name="beforeTicks"/>.</summary>
|
||||
private void ClassifyPointsBefore(long beforeTicks)
|
||||
{
|
||||
if (_aggregate != HistoryAggregateType.Interpolative) return;
|
||||
for (; _nextPoint < _plan.BucketCount; _nextPoint++)
|
||||
{
|
||||
var a = BucketStartTicks(_nextPoint);
|
||||
if (a >= beforeTicks) break;
|
||||
|
||||
if (_prevAny is not { } p)
|
||||
{
|
||||
_pending.Add(new PendingPoint(_nextPoint, a, PointQuality.NoPrior));
|
||||
}
|
||||
else if (p.IsUsable && p.TimeUtc.Ticks == a)
|
||||
{
|
||||
// A usable sample exactly at the interval start: its own value and status.
|
||||
_points[_nextPoint] = IsNumeric(p)
|
||||
? new DataValueSnapshot(p.Value.Number, p.Status, _plan.BucketStart(_nextPoint), default)
|
||||
: InvalidPoint(_nextPoint);
|
||||
}
|
||||
else if (!p.IsUsable)
|
||||
{
|
||||
_pending.Add(new PendingPoint(_nextPoint, a, PointQuality.Degraded));
|
||||
}
|
||||
else
|
||||
{
|
||||
// Inside a segment that starts at a usable sample; its quality is only known once the next sample
|
||||
// arrives (beforeTicks is that sample's time, or MaxValue at Finish ⇒ extrapolated).
|
||||
_pending.Add(new PendingPoint(_nextPoint, a, PointQuality.OnSegment));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Resolves the pending bucket starts against the bridging pair <c>(p, next)</c>.</summary>
|
||||
private void ResolvePending(StoredSample? p, StoredSample? next)
|
||||
{
|
||||
if (_pending.Count == 0) return;
|
||||
foreach (var pt in _pending)
|
||||
{
|
||||
var bucketStart = _plan.BucketStart(pt.Bucket);
|
||||
if (p is not { } prev || pt.Quality == PointQuality.NoPrior)
|
||||
{
|
||||
_points[pt.Bucket] = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, bucketStart, default);
|
||||
continue;
|
||||
}
|
||||
if (!IsNumeric(prev) || next is { } nn && !IsNumeric(nn))
|
||||
{
|
||||
_points[pt.Bucket] = InvalidPoint(pt.Bucket);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Good only on a segment that runs straight from prev to next (no outage skipped, not extrapolated), and only
|
||||
// when the samples that define the value are Good: both ends when sloped, prev alone when stepped.
|
||||
double value;
|
||||
bool good;
|
||||
if (next is { } n && IsSloped(prev, n))
|
||||
{
|
||||
value = prev.Value.Number + (n.Value.Number - prev.Value.Number)
|
||||
* ((double)(pt.Ticks - prev.TimeUtc.Ticks) / (n.TimeUtc.Ticks - prev.TimeUtc.Ticks));
|
||||
good = pt.Quality == PointQuality.OnSegment && prev.IsGood && n.IsGood;
|
||||
}
|
||||
else
|
||||
{
|
||||
value = prev.Value.Number;
|
||||
good = pt.Quality == PointQuality.OnSegment && next is not null && prev.IsGood;
|
||||
}
|
||||
var status = good ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
|
||||
_points[pt.Bucket] = new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), bucketStart, default);
|
||||
}
|
||||
_pending.Clear();
|
||||
}
|
||||
|
||||
/// <summary>Marks every pending on-segment point degraded: its segment ends at a Bad sample.</summary>
|
||||
private void DegradePending()
|
||||
{
|
||||
for (var k = 0; k < _pending.Count; k++)
|
||||
if (_pending[k].Quality == PointQuality.OnSegment)
|
||||
_pending[k] = _pending[k] with { Quality = PointQuality.Degraded };
|
||||
}
|
||||
|
||||
private DataValueSnapshot InvalidPoint(long bucket) =>
|
||||
new(null, HistorianStatusCodes.BadAggregateInvalidInputs, _plan.BucketStart(bucket), default);
|
||||
}
|
||||
Reference in New Issue
Block a user