fix(historian): #671 review — raw-row cap, reverse Interpolative, SDK status rule, Partial bit
M1: the time-weighted family (incl. Total, formerly a pushed-down SUM) streams raw window rows into the C# evaluator. New shared knob ServerHistorian:MaxTimeWeightedRawRows (default 1,000,000; <=0 unbounded with a Validate warning) flows through HistorianBackendRegistry into both providers. The window query is capped at cap+1 (SQL Server TOP (@maxRows), QuestDB LIMIT $4); one row past the cap throws the new Core.Abstractions HistoryReadLimitExceededException, which the node manager maps to BadTooManyOperations for that node. Not counted as a backend health failure; bound rows are not counted; per-sample aggregates are not capped. M2: reverse Interpolative is anchored at the later time (Part 13): points at End - i*interval, emitted latest first. 10:00->09:00 @1h yields 10:00. L1: Partial bit (0x4, = Opc.Ua.AggregateBits.Partial) on non-Bad results of a truncated last interval (time-weighted family; per-sample family is a follow-up). L2/L3: aligned with the SDK AggregateCalculator.GetTimeBasedStatusCode (decompiled from Opc.Ua.Server 1.5.378.106): TreatUncertainAsBad=false counts Uncertain time (Uncertain samples, held-into-outage, extrapolation) as Good time; data but no non-Bad time => generic Bad (0x80000000); BadNoData only when there is no data at all. Consequence: a live tag's latest interval reads Good (documented). L5: release note no longer claims Total matches the Gateway Integral exactly (AVEVA's divisor is unverified); it notes Total is now evaluated in C# and subject to the cap. Tests: unit (reverse Interpolative x3, Partial, Bad vs BadNoData incl. an open-ended outage, cap helpers, SDK bit oracles); conformance (reverse Interpolative, raw-row cap at/over/0, bounds not counted, health unaffected); node manager BadTooManyOperations; options default + warning. Each new behaviour falsified by mutation (aggregator x9, data-source cap checks on both backends against the real DBs, node-manager catch).
This commit is contained in:
Joseph Doherty
@@ -71,6 +71,22 @@ public sealed class HistoryAggregateNotSupportedException : NotSupportedExceptio
|
||||
=> Aggregate = aggregate;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by a history backend when a read would exceed a configured size limit (e.g. the raw-row cap of the SQL
|
||||
/// backends' time-weighted aggregates, #671). The Server HistoryRead path translates it to
|
||||
/// <c>BadTooManyOperations</c> for that node only; it is a client-request limit, not a backend fault.
|
||||
/// </summary>
|
||||
public sealed class HistoryReadLimitExceededException : InvalidOperationException
|
||||
{
|
||||
/// <summary>The configured limit that was exceeded.</summary>
|
||||
public long Limit { get; }
|
||||
|
||||
/// <summary>Creates the exception.</summary>
|
||||
/// <param name="limit">The configured limit.</param>
|
||||
/// <param name="message">Operator-facing description.</param>
|
||||
public HistoryReadLimitExceededException(long limit, string message) : base(message) => Limit = limit;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One row returned by <see cref="IHistorianDataSource.ReadEventsAsync"/> — a historical
|
||||
/// alarm/event record, not the OPC UA live-event stream. Fields match the minimum set the
|
||||
|
||||
@@ -125,7 +125,8 @@ internal static class QuestDbAggregateQueries
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Time-weighted family (#671), window statement: every live sample in <c>[$2, $3)</c> for tag <c>$1</c>, ascending.
|
||||
/// Time-weighted family (#671), window statement: the live samples in <c>[$2, $3)</c> for tag <c>$1</c>, ascending,
|
||||
/// at most <c>$4</c> (the raw-row cap + 1).
|
||||
/// Columns: <c>ts, kind, num_value, int_value, str_value, status, server_ts</c>. The caller feeds the before-bounds,
|
||||
/// these rows and the after-bounds, in that order, into <see cref="TimeWeightedAggregator"/> — the same evaluator
|
||||
/// the SQL Server backend uses, so the two agree exactly. Trade-off: the window's rows travel to the server rather
|
||||
@@ -135,7 +136,7 @@ internal static class QuestDbAggregateQueries
|
||||
/// <param name="sample">Prefixed sample table name.</param>
|
||||
/// <returns>The SQL text.</returns>
|
||||
public static string TimeWeightedWindow(string sample) =>
|
||||
$"SELECT {SampleColumns} FROM {sample} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts ASC";
|
||||
$"SELECT {SampleColumns} FROM {sample} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts ASC LIMIT $4";
|
||||
|
||||
/// <summary>The row columns the time-weighted statements return (after <c>k</c> for the bounds).</summary>
|
||||
private const string SampleColumns = "ts, kind, num_value, int_value, str_value, status, server_ts";
|
||||
|
||||
+16
-1
@@ -64,6 +64,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
private readonly ILogger _log;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly HistorianHealthCounters _health = new();
|
||||
private readonly int _maxTimeWeightedRawRows;
|
||||
|
||||
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
@@ -71,10 +72,12 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
/// <param name="catalog">Tag catalog (reads use only its read-only <see cref="QuestDbTagCatalog.ExistsAsync"/>).</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
|
||||
/// <param name="maxTimeWeightedRawRows">Raw in-window row cap of one time-weighted read (<= 0 = unbounded).</param>
|
||||
internal QuestDbHistorianDataSource(
|
||||
QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, ILogger? log = null,
|
||||
TimeProvider? timeProvider = null)
|
||||
TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
|
||||
{
|
||||
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
|
||||
_conn = conn ?? throw new ArgumentNullException(nameof(conn));
|
||||
_bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap));
|
||||
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
|
||||
@@ -387,6 +390,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
RecordSuccess();
|
||||
return new HistoryReadResult(result, ContinuationPoint: null);
|
||||
}
|
||||
catch (HistoryReadLimitExceededException)
|
||||
{
|
||||
// A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
@@ -420,9 +428,16 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
QuestDbConnection.BindText(cmd, tag);
|
||||
QuestDbConnection.BindTimestamp(cmd, plan.StartUtc);
|
||||
QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
|
||||
cmd.Parameters.Add(new NpgsqlParameter
|
||||
{ NpgsqlDbType = NpgsqlDbType.Integer, Value = TimeWeightedAggregator.WindowFetchLimit(_maxTimeWeightedRawRows) });
|
||||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
long windowRows = 0;
|
||||
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
// LIMIT $4 = cap + 1: one row past the cap fails the node without reading on.
|
||||
TimeWeightedAggregator.CheckWindowRows(++windowRows, _maxTimeWeightedRawRows, tag);
|
||||
tw.Add(ReadStored(reader, 0));
|
||||
}
|
||||
}
|
||||
if (bounds[2] is { } anyAfter) tw.Add(anyAfter);
|
||||
if (bounds[3] is { } usableAfter) tw.Add(usableAfter);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
using System.Collections.Frozen;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
@@ -33,11 +34,16 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync
|
||||
|
||||
/// <summary>Creates the provider over the <c>ServerHistorian:QuestDb</c> options (validated elsewhere).</summary>
|
||||
/// <param name="options">The <c>ServerHistorian:QuestDb</c> options.</param>
|
||||
public QuestDbHistorianProvider(QuestDbHistorianOptions options)
|
||||
/// <param name="maxTimeWeightedRawRows">The shared <c>ServerHistorian:MaxTimeWeightedRawRows</c> cap (<= 0 = unbounded).</param>
|
||||
public QuestDbHistorianProvider(QuestDbHistorianOptions options,
|
||||
int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
|
||||
{
|
||||
_options = options ?? throw new ArgumentNullException(nameof(options));
|
||||
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
|
||||
}
|
||||
|
||||
private readonly int _maxTimeWeightedRawRows;
|
||||
|
||||
/// <inheritdoc />
|
||||
public string Name => "QuestDb";
|
||||
|
||||
@@ -69,7 +75,7 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync
|
||||
{
|
||||
var s = Get(sp);
|
||||
return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
|
||||
Loggers(sp).CreateLogger<QuestDbHistorianDataSource>());
|
||||
Loggers(sp).CreateLogger<QuestDbHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -12,6 +12,12 @@ public static class HistorianInfoBits
|
||||
/// <summary>The value was interpolated.</summary>
|
||||
public const uint HistorianInterpolated = 0x2u;
|
||||
|
||||
/// <summary>
|
||||
/// Partial (Part 4 §7.39 / <c>Opc.Ua.AggregateBits.Partial</c>): the calculated value covers an interval shorter
|
||||
/// than the processing interval (the truncated last interval of a processed read).
|
||||
/// </summary>
|
||||
public const uint HistorianPartial = 0x4u;
|
||||
|
||||
/// <summary>
|
||||
/// Marks <paramref name="status"/> as a Calculated data value. Any historian bits the status already carries
|
||||
/// (e.g. a stored Interpolated sample) are cleared first, so the result is never both.
|
||||
|
||||
@@ -10,6 +10,9 @@ public static class HistorianStatusCodes
|
||||
public const uint GoodEntryInserted = 0x00A20000u;
|
||||
public const uint GoodEntryReplaced = 0x00A30000u;
|
||||
public const uint BadNoData = 0x809B0000u;
|
||||
|
||||
/// <summary>Generic Bad: a time-weighted interval holding data but no non-Bad time (the SDK aggregate rule).</summary>
|
||||
public const uint Bad = 0x80000000u;
|
||||
public const uint BadEntryExists = 0x809F0000u;
|
||||
public const uint BadNoEntryExists = 0x80A00000u;
|
||||
public const uint BadInvalidTimestamp = 0x80230000u;
|
||||
|
||||
+103
-37
@@ -39,17 +39,23 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
/// <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>
|
||||
/// <item><term>Interpolative</term><description>The interpolated-bounding signal's value at each interval start —
|
||||
/// for a reverse read, at <c>End − i·interval</c> (Part 13 anchors a reverse read at its later time), emitted
|
||||
/// latest first.</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).
|
||||
/// <b>Status (time-weighted four)</b> — the SDK <c>AggregateCalculator.GetTimeBasedStatusCode</c> rule with
|
||||
/// TreatUncertainAsBad = false, so Uncertain time (Uncertain samples, a value held up to an outage, stepped
|
||||
/// extrapolation past the last sample) counts as Good time: every instant of the interval non-Bad ⇒
|
||||
/// <c>Good</c>; no non-Bad time but some Bad (outage) time ⇒ <c>Bad</c>; no data at all ⇒ <c>BadNoData</c>;
|
||||
/// otherwise (PercentDataGood = PercentDataBad = 100) <c>UncertainDataSubNormal</c>. Time before the first sample
|
||||
/// counts as Bad (no data) time. A usable String/DateTime sample contributing to the interval ⇒
|
||||
/// <c>BadAggregateInvalidInputs</c>. Every status carries the Calculated bit, plus the Partial bit on a non-Bad
|
||||
/// result of a truncated last interval; 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
|
||||
/// <b>Status (Interpolative).</b> A usable sample exactly at the point ⇒ 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)
|
||||
@@ -60,26 +66,54 @@ public sealed class TimeWeightedAggregator
|
||||
{
|
||||
private const double TicksPerSecond = TimeSpan.TicksPerSecond;
|
||||
|
||||
/// <summary>
|
||||
/// Default cap on the raw in-window rows one time-weighted read may stream (<c>ServerHistorian:MaxTimeWeightedRawRows</c>).
|
||||
/// </summary>
|
||||
public const int DefaultMaxWindowRows = 1_000_000;
|
||||
|
||||
/// <summary>
|
||||
/// The <c>TOP</c>/<c>LIMIT</c> a backend puts on its window query for a configured cap: <c>cap + 1</c> (so one
|
||||
/// row past the cap proves it was exceeded without reading on), or <see cref="int.MaxValue"/> when the cap is
|
||||
/// <= 0 (unbounded).
|
||||
/// </summary>
|
||||
/// <param name="maxWindowRows">The configured cap.</param>
|
||||
/// <returns>The row limit to put on the window query.</returns>
|
||||
public static int WindowFetchLimit(int maxWindowRows) =>
|
||||
maxWindowRows <= 0 || maxWindowRows == int.MaxValue ? int.MaxValue : maxWindowRows + 1;
|
||||
|
||||
/// <summary>Throws when <paramref name="rowsRead"/> has passed a positive <paramref name="maxWindowRows"/>.</summary>
|
||||
/// <param name="rowsRead">Window rows read so far.</param>
|
||||
/// <param name="maxWindowRows">The configured cap (<= 0 = unbounded).</param>
|
||||
/// <param name="tag">The tag, for the message.</param>
|
||||
/// <exception cref="HistoryReadLimitExceededException">The cap is exceeded.</exception>
|
||||
public static void CheckWindowRows(long rowsRead, int maxWindowRows, string tag)
|
||||
{
|
||||
if (maxWindowRows > 0 && rowsRead > maxWindowRows)
|
||||
throw new HistoryReadLimitExceededException(maxWindowRows,
|
||||
$"The time-weighted aggregate read of '{tag}' covers more than {maxWindowRows} raw samples " +
|
||||
"(ServerHistorian:MaxTimeWeightedRawRows); narrow the window or use a pushed-down aggregate such as Average.");
|
||||
}
|
||||
|
||||
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.
|
||||
// Simple-bounding (raw) signal, per bucket: ∫ over non-Bad time, non-Bad ticks, ticks inside an outage.
|
||||
private readonly double[] _rawArea;
|
||||
private readonly long[] _usableTicks;
|
||||
private readonly long[] _goodTicks;
|
||||
private readonly long[] _badTicks;
|
||||
// Interpolated-bounding (bridged) signal, per bucket.
|
||||
private readonly double[] _bridgedArea;
|
||||
private readonly long[] _bridgedTicks;
|
||||
private readonly bool[] _invalid;
|
||||
// Interpolative, per bucket start.
|
||||
// Interpolative, per point (index = ascending time; see PointTicks).
|
||||
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 long _nextPoint; // next point (ascending time) not yet classified
|
||||
private readonly List<PendingPoint> _pending = []; // points awaiting the next usable sample
|
||||
private bool _finished;
|
||||
|
||||
/// <summary>
|
||||
@@ -106,7 +140,7 @@ public sealed class TimeWeightedAggregator
|
||||
var n = (int)plan.BucketCount;
|
||||
_rawArea = new double[n];
|
||||
_usableTicks = new long[n];
|
||||
_goodTicks = new long[n];
|
||||
_badTicks = new long[n];
|
||||
_bridgedArea = new double[n];
|
||||
_bridgedTicks = new long[n];
|
||||
_invalid = new bool[n];
|
||||
@@ -142,15 +176,13 @@ public sealed class TimeWeightedAggregator
|
||||
// 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
|
||||
}
|
||||
// Toward a usable sample: sloped or stepped; toward a Bad one: held up to the outage.
|
||||
var sloped = s.IsUsable && IsSloped(p, s);
|
||||
AddRaw(p, t, sloped ? s.Value.Number : p.Value.Number);
|
||||
}
|
||||
else if (_prevAny is { } bad)
|
||||
{
|
||||
AddBad(bad, t);
|
||||
}
|
||||
|
||||
if (!s.IsUsable) DegradePending(); // the segment the pending points sit on runs into an outage
|
||||
@@ -179,7 +211,9 @@ public sealed class TimeWeightedAggregator
|
||||
_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
|
||||
AddRaw(p, _end, p.Value.Number); // stepped extrapolation
|
||||
else if (_prevAny is { } bad && bad.TimeUtc.Ticks < _end)
|
||||
AddBad(bad, _end); // the outage runs to the window end
|
||||
if (_prevUsable is { } pu && pu.TimeUtc.Ticks < _end)
|
||||
AddBridged(pu, _end, pu.Value.Number);
|
||||
ResolvePending(_prevUsable, next: null);
|
||||
@@ -195,20 +229,26 @@ public sealed class TimeWeightedAggregator
|
||||
|
||||
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);
|
||||
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, PointTime(i), now);
|
||||
|
||||
DataValueSnapshot Calc(object? value, uint status) =>
|
||||
new(value, HistorianInfoBits.WithCalculated(status), bucketStart, now);
|
||||
var bucketStart = _plan.BucketStart(i);
|
||||
var duration = BucketEnd(i) - BucketStartTicks(i);
|
||||
var partial = duration < _plan.IntervalTicks; // only the truncated last interval
|
||||
DataValueSnapshot Calc(object? value, uint status)
|
||||
{
|
||||
var code = HistorianInfoBits.WithCalculated(status);
|
||||
if (partial && !StatusBits.IsBad(status)) code |= HistorianInfoBits.HistorianPartial;
|
||||
return new DataValueSnapshot(value, code, bucketStart, now);
|
||||
}
|
||||
|
||||
if (_invalid[i]) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
|
||||
if (_usableTicks[i] == 0) return AggregateEvaluator.NoData(bucketStart, now);
|
||||
if (_usableTicks[i] == 0)
|
||||
return _badTicks[i] > 0 ? Calc(null, HistorianStatusCodes.Bad) : AggregateEvaluator.NoData(bucketStart, now);
|
||||
|
||||
var duration = BucketEnd(i) - BucketStartTicks(i);
|
||||
var status = _goodTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
|
||||
var status = _usableTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
|
||||
return _aggregate switch
|
||||
{
|
||||
HistoryAggregateType.TimeAverage => Calc(_bridgedArea[i] / _bridgedTicks[i], status),
|
||||
@@ -233,14 +273,29 @@ public sealed class TimeWeightedAggregator
|
||||
|
||||
private long BucketEnd(long i) => Math.Min(BucketStartTicks(i) + _plan.IntervalTicks, _end);
|
||||
|
||||
private void AddRaw(StoredSample from, long toTicks, double toValue, bool good) =>
|
||||
private void AddRaw(StoredSample from, long toTicks, double toValue) =>
|
||||
Distribute(from, toTicks, toValue, (i, area, ticks) =>
|
||||
{
|
||||
_rawArea[i] += area;
|
||||
_usableTicks[i] += ticks;
|
||||
if (good) _goodTicks[i] += ticks;
|
||||
});
|
||||
|
||||
/// <summary>Counts <c>[from.t, toTicks)</c> as outage (Bad) time; no value, never marks a bucket invalid.</summary>
|
||||
private void AddBad(StoredSample from, long toTicks)
|
||||
{
|
||||
var lo = Math.Max(from.TimeUtc.Ticks, _start);
|
||||
var hi = Math.Min(toTicks, _end);
|
||||
if (lo >= hi || _plan.BucketCount == 0) return;
|
||||
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) _badTicks[i] += b - a;
|
||||
}
|
||||
}
|
||||
|
||||
private void AddBridged(StoredSample from, long toTicks, double toValue) =>
|
||||
Distribute(from, toTicks, toValue, (i, area, ticks) =>
|
||||
{
|
||||
@@ -279,13 +334,24 @@ public sealed class TimeWeightedAggregator
|
||||
|
||||
// ------------------------------------------------------------------------------------------ Interpolative
|
||||
|
||||
/// <summary>Classifies every unclassified bucket start strictly before <paramref name="beforeTicks"/>.</summary>
|
||||
/// <summary>
|
||||
/// The time of Interpolative point <paramref name="j"/> (ascending time). Forward: the bucket starts
|
||||
/// <c>start + j·interval</c>. Reverse (Part 13 anchors a reverse read at its later time, the request's start):
|
||||
/// <c>End − (n−1−j)·interval</c>, i.e. <c>End, End − interval, …</c> once <see cref="BucketPlan.Assemble"/> has
|
||||
/// reversed the order — so a reverse read 10:00 → 09:00 at 1 h yields the value at 10:00.
|
||||
/// </summary>
|
||||
private long PointTicks(long j) =>
|
||||
_plan.Reversed ? _end - (_plan.BucketCount - 1 - j) * _plan.IntervalTicks : BucketStartTicks(j);
|
||||
|
||||
private DateTime PointTime(long j) => new(PointTicks(j), DateTimeKind.Utc);
|
||||
|
||||
/// <summary>Classifies every unclassified point 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);
|
||||
var a = PointTicks(_nextPoint);
|
||||
if (a >= beforeTicks) break;
|
||||
|
||||
if (_prevAny is not { } p)
|
||||
@@ -296,7 +362,7 @@ public sealed class TimeWeightedAggregator
|
||||
{
|
||||
// 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)
|
||||
? new DataValueSnapshot(p.Value.Number, p.Status, PointTime(_nextPoint), default)
|
||||
: InvalidPoint(_nextPoint);
|
||||
}
|
||||
else if (!p.IsUsable)
|
||||
@@ -318,7 +384,7 @@ public sealed class TimeWeightedAggregator
|
||||
if (_pending.Count == 0) return;
|
||||
foreach (var pt in _pending)
|
||||
{
|
||||
var bucketStart = _plan.BucketStart(pt.Bucket);
|
||||
var bucketStart = PointTime(pt.Bucket);
|
||||
if (p is not { } prev || pt.Quality == PointQuality.NoPrior)
|
||||
{
|
||||
_points[pt.Bucket] = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, bucketStart, default);
|
||||
@@ -360,5 +426,5 @@ public sealed class TimeWeightedAggregator
|
||||
}
|
||||
|
||||
private DataValueSnapshot InvalidPoint(long bucket) =>
|
||||
new(null, HistorianStatusCodes.BadAggregateInvalidInputs, _plan.BucketStart(bucket), default);
|
||||
new(null, HistorianStatusCodes.BadAggregateInvalidInputs, PointTime(bucket), default);
|
||||
}
|
||||
|
||||
@@ -95,7 +95,7 @@ internal static class SqlAggregateQueries
|
||||
/// Time-weighted family (#671: TimeAverage / TimeAverage2 / Total / Total2 / Interpolative) — <b>not</b> pushed down.
|
||||
/// One batch of five result sets, each <c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode</c>:
|
||||
/// (1) the latest usable sample before <c>@start</c>, (2) the latest sample of any quality before <c>@start</c>,
|
||||
/// (3) every sample in <c>[@start, @end)</c> ascending, (4) the first sample of any quality at or after
|
||||
/// (3) the samples in <c>[@start, @end)</c> ascending, at most <c>@maxRows</c> (the raw-row cap + 1), (4) the first sample of any quality at or after
|
||||
/// <c>@end</c>, (5) the first usable one at or after <c>@end</c>. Each bound is a <c>TOP 1</c> seek on the
|
||||
/// <c>(TagId, SourceTimeUtc)</c> key. The caller streams them, in that order, into
|
||||
/// <see cref="TimeWeightedAggregator"/> (shared with QuestDB, so both backends agree exactly).
|
||||
@@ -116,7 +116,7 @@ internal static class SqlAggregateQueries
|
||||
return
|
||||
$"SELECT TOP 1 {cols} {key}SourceTimeUtc < @start{usable} ORDER BY SourceTimeUtc DESC; " +
|
||||
$"SELECT TOP 1 {cols} {key}SourceTimeUtc < @start ORDER BY SourceTimeUtc DESC; " +
|
||||
$"SELECT {cols} {key}SourceTimeUtc >= @start AND SourceTimeUtc < @end ORDER BY SourceTimeUtc ASC; " +
|
||||
$"SELECT TOP (@maxRows) {cols} {key}SourceTimeUtc >= @start AND SourceTimeUtc < @end ORDER BY SourceTimeUtc ASC; " +
|
||||
$"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end ORDER BY SourceTimeUtc ASC; " +
|
||||
$"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end{usable} ORDER BY SourceTimeUtc ASC;";
|
||||
}
|
||||
|
||||
+20
-3
@@ -59,6 +59,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
private readonly TimeProvider _time;
|
||||
|
||||
private readonly HistorianHealthCounters _health = new();
|
||||
private readonly int _maxTimeWeightedRawRows;
|
||||
|
||||
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
@@ -66,10 +67,12 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
/// <param name="catalog">Tag catalog (shared; reads use only its read-only lookup).</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
|
||||
/// <param name="maxTimeWeightedRawRows">Raw in-window row cap of one time-weighted read (<= 0 = unbounded).</param>
|
||||
internal SqlServerHistorianDataSource(
|
||||
SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null,
|
||||
TimeProvider? timeProvider = null)
|
||||
TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
|
||||
{
|
||||
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
|
||||
_conn = conn ?? throw new ArgumentNullException(nameof(conn));
|
||||
_bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap));
|
||||
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
|
||||
@@ -321,7 +324,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
if (family == AggregateFamily.TimeWeighted)
|
||||
{
|
||||
var tw = new TimeWeightedAggregator(aggregate, plan);
|
||||
if (tagId is { } twId) await StreamTimeWeightedAsync(twId, plan, tw, cancellationToken).ConfigureAwait(false);
|
||||
if (tagId is { } twId) await StreamTimeWeightedAsync(twId, fullReference, plan, tw, cancellationToken).ConfigureAwait(false);
|
||||
var twResult = tw.Finish(now);
|
||||
RecordSuccess();
|
||||
return new HistoryReadResult(twResult, ContinuationPoint: null);
|
||||
@@ -354,6 +357,11 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
RecordSuccess();
|
||||
return new HistoryReadResult(result, ContinuationPoint: null);
|
||||
}
|
||||
catch (HistoryReadLimitExceededException)
|
||||
{
|
||||
// A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
@@ -365,18 +373,27 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
/// Streams the time-weighted batch (<see cref="SqlAggregateQueries.TimeWeighted"/>) into <paramref name="tw"/>: the
|
||||
/// two before-bounds, the window, the two after-bounds — ascending, with duplicates dropped by the aggregator.
|
||||
/// </summary>
|
||||
private async Task StreamTimeWeightedAsync(int id, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct)
|
||||
private async Task StreamTimeWeightedAsync(int id, string tag, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct)
|
||||
{
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
await using var cmd = _conn.CreateCommand(conn, SqlAggregateQueries.TimeWeighted(_conn.Q("Sample")));
|
||||
cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
|
||||
SqlHistorianConnection.AddDateTime2(cmd, "@start", DateTime.SpecifyKind(plan.StartUtc, DateTimeKind.Utc));
|
||||
SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(plan.EndUtc, DateTimeKind.Utc));
|
||||
cmd.Parameters.Add(new SqlParameter("@maxRows", SqlDbType.Int)
|
||||
{ Value = TimeWeightedAggregator.WindowFetchLimit(_maxTimeWeightedRawRows) });
|
||||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
var resultSet = 0;
|
||||
do
|
||||
{
|
||||
long windowRows = 0;
|
||||
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
// Result set 2 is the window (TOP (@maxRows) = cap + 1): one row past the cap fails the node.
|
||||
if (resultSet == 2) TimeWeightedAggregator.CheckWindowRows(++windowRows, _maxTimeWeightedRawRows, tag);
|
||||
tw.Add(ReadStored(reader, 0));
|
||||
}
|
||||
resultSet++;
|
||||
} while (await reader.NextResultAsync(ct).ConfigureAwait(false));
|
||||
}
|
||||
|
||||
|
||||
+8
-2
@@ -1,3 +1,4 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
using System.Collections.Frozen;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
@@ -29,11 +30,16 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
|
||||
|
||||
/// <summary>Creates the provider over the <c>ServerHistorian:SqlServer</c> options (validated elsewhere).</summary>
|
||||
/// <param name="options">The <c>ServerHistorian:SqlServer</c> options.</param>
|
||||
public SqlServerHistorianProvider(SqlServerHistorianOptions options)
|
||||
/// <param name="maxTimeWeightedRawRows">The shared <c>ServerHistorian:MaxTimeWeightedRawRows</c> cap (<= 0 = unbounded).</param>
|
||||
public SqlServerHistorianProvider(SqlServerHistorianOptions options,
|
||||
int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
|
||||
{
|
||||
_options = options ?? throw new ArgumentNullException(nameof(options));
|
||||
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
|
||||
}
|
||||
|
||||
private readonly int _maxTimeWeightedRawRows;
|
||||
|
||||
/// <inheritdoc />
|
||||
public string Name => "SqlServer";
|
||||
|
||||
@@ -63,7 +69,7 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
|
||||
{
|
||||
var s = Get(sp);
|
||||
return new SqlServerHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
|
||||
Loggers(sp).CreateLogger<SqlServerHistorianDataSource>());
|
||||
Loggers(sp).CreateLogger<SqlServerHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -41,8 +41,8 @@ public static class HistorianBackendRegistry
|
||||
new Dictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>>(StringComparer.OrdinalIgnoreCase)
|
||||
{
|
||||
[ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o),
|
||||
[SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)),
|
||||
[QuestDbBackendName] = (_, c) => new QuestDbHistorianProvider(BindQuestDb(c)),
|
||||
[SqlServerBackendName] = (o, c) => new SqlServerHistorianProvider(BindSqlServer(c), o.MaxTimeWeightedRawRows),
|
||||
[QuestDbBackendName] = (o, c) => new QuestDbHistorianProvider(BindQuestDb(c), o.MaxTimeWeightedRawRows),
|
||||
};
|
||||
|
||||
/// <summary>The registered backend names, for validation messages.</summary>
|
||||
|
||||
@@ -5779,6 +5779,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
errors[handle.Index] = StatusCodes.BadAggregateNotSupported;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadAggregateNotSupported };
|
||||
}
|
||||
catch (HistoryReadLimitExceededException ex)
|
||||
{
|
||||
// The read would exceed a backend size limit (#671: the time-weighted raw-row cap) — refuse THIS node
|
||||
// explicitly instead of a generic fault or a deadline-elapsed BadTimeout.
|
||||
_logger.LogWarning("OtOpcUaNodeManager: HistoryRead limit exceeded for node {NodeId}: {Message}", handle.NodeId, ex.Message);
|
||||
errors[handle.Index] = StatusCodes.BadTooManyOperations;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// One node's backend failure (throw / timeout) must not throw out of the batch — surface a
|
||||
|
||||
@@ -112,6 +112,16 @@ public sealed class ServerHistorianOptions
|
||||
/// </summary>
|
||||
public TimeSpan HistoryReadDeadline { get; init; } = TimeSpan.FromSeconds(60);
|
||||
|
||||
/// <summary>
|
||||
/// <c>SqlServer</c> / <c>QuestDb</c> backends only (#671): the most raw in-window samples one Part 13
|
||||
/// time-weighted read (TimeAverage, TimeAverage2, Total, Total2, Interpolative) may stream into the server-side
|
||||
/// evaluator — that family is not pushed down to SQL. A read past the cap stops at cap + 1 rows and fails THAT
|
||||
/// node with <c>BadTooManyOperations</c> instead of holding a pooled connection until
|
||||
/// <see cref="HistoryReadDeadline"/>. Non-positive = unbounded (a <see cref="Validate"/> warning). Default
|
||||
/// 1 000 000 (≈ 11.5 days of a 1 Hz tag).
|
||||
/// </summary>
|
||||
public int MaxTimeWeightedRawRows { get; init; } = 1_000_000;
|
||||
|
||||
/// <summary>Returns operator-facing misconfiguration warnings for an <c>Enabled</c> historian
|
||||
/// (empty when disabled or correctly configured). Pure — the registration logs each entry.
|
||||
/// <para>
|
||||
@@ -154,6 +164,8 @@ public sealed class ServerHistorianOptions
|
||||
warnings.Add($"ServerHistorian:HistoryReadBatchParallelism is {HistoryReadBatchParallelism} — must be > 0; HistoryRead falls back to serving each node sequentially.");
|
||||
if (MaxConcurrentHistoryReadBatches <= 0)
|
||||
warnings.Add($"ServerHistorian:MaxConcurrentHistoryReadBatches is {MaxConcurrentHistoryReadBatches} — must be > 0; the process-wide HistoryRead batch limiter is disabled.");
|
||||
if (MaxTimeWeightedRawRows <= 0)
|
||||
warnings.Add($"ServerHistorian:MaxTimeWeightedRawRows is {MaxTimeWeightedRawRows} — must be > 0; time-weighted aggregate reads on the SqlServer/QuestDb backends stream an unbounded number of raw samples.");
|
||||
if (HistoryReadDeadline <= TimeSpan.Zero)
|
||||
warnings.Add($"ServerHistorian:HistoryReadDeadline is {HistoryReadDeadline} — must be > 0; HistoryRead requests run unbounded (no per-request deadline).");
|
||||
return warnings;
|
||||
|
||||
Reference in New Issue
Block a user