Merge fix/693-gateway-attime: Gateway AtTime from raw reads + shared evaluator, node type hint (#693)
This commit is contained in:
Joseph Doherty
@@ -81,6 +81,31 @@ public interface IHistorianDataSource : IDisposable
|
||||
IReadOnlyList<DateTime> timestampsUtc,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #693: <see cref="ReadAtTimeAsync(string, IReadOnlyList{DateTime}, CancellationToken)"/> with the value type of
|
||||
/// the node being read, for a backend whose stored values do not carry their type (the Gateway: AVEVA stores every
|
||||
/// historized type as a numeric analog, so a Boolean comes back as a 0/1 Double). Such a backend uses the hint to apply
|
||||
/// the AtTime rules exactly as the typed backends do — a Boolean (or String / DateTime) node is stepped, an integer node
|
||||
/// is interpolated, rounded half away from zero and returned as the node's integer type, a floating-point node is
|
||||
/// interpolated. A backend that stores typed values ignores it: the default forwards to the hint-less overload, so
|
||||
/// SqlServer, QuestDB and the Null source need no change.
|
||||
/// </summary>
|
||||
/// <param name="fullReference">The full reference of the tag to read.</param>
|
||||
/// <param name="timestampsUtc">The list of timestamps to read values at.</param>
|
||||
/// <param name="valueType">
|
||||
/// The scalar value type of the node (<see cref="TypeCode.Boolean"/>, the integer codes, <see cref="TypeCode.Single"/>,
|
||||
/// <see cref="TypeCode.Double"/>, <see cref="TypeCode.String"/>, <see cref="TypeCode.DateTime"/>), or <c>null</c> when
|
||||
/// unknown, not scalar, or not one of those — the backend's own policy then applies.
|
||||
/// </param>
|
||||
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
|
||||
/// <returns>One sample per requested timestamp, in the same order as <paramref name="timestampsUtc"/>.</returns>
|
||||
Task<HistoryReadResult> ReadAtTimeAsync(
|
||||
string fullReference,
|
||||
IReadOnlyList<DateTime> timestampsUtc,
|
||||
TypeCode? valueType,
|
||||
CancellationToken cancellationToken) =>
|
||||
ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Read historical alarm / event records — OPC UA HistoryReadEvents service.
|
||||
/// Distinct from any live event stream; sources here come from the historian's
|
||||
@@ -95,7 +120,7 @@ public interface IHistorianDataSource : IDisposable
|
||||
/// <c>uint maxValuesPerNode</c> because their OPC UA HistoryRead surface has no
|
||||
/// equivalent "use default" sentinel.
|
||||
///
|
||||
/// This surface declares <see cref="ReadAtTimeAsync"/> and <see cref="ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>
|
||||
/// This surface declares <see cref="ReadAtTimeAsync(string, IReadOnlyList{DateTime}, CancellationToken)"/> and <see cref="ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>
|
||||
/// as required members — a server-side historian owns the full read surface.
|
||||
/// </remarks>
|
||||
/// <param name="sourceName">The source name to filter events, or null to return events from all sources.</param>
|
||||
|
||||
@@ -0,0 +1,494 @@
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #693: HistoryRead AtTime on the Gateway backend, served from RAW gateway reads and evaluated with the SQL backends'
|
||||
/// shared <see cref="AtTimeEvaluator"/> (Part 11 §6.5.4) — as #681 did for the per-sample aggregates — so the Gateway
|
||||
/// answers exactly what SqlServer and QuestDB answer for the same stored samples. AVEVA's own interpolated answer (the
|
||||
/// gateway's <c>ReadAtTime</c>) is not used: live, it returned the previous value as Good at a stored NULL's own time,
|
||||
/// null + Bad at a Good sample right after a NULL, a drifted value (99.99999999944) at an exact stored 100, and no
|
||||
/// Interpolated bit.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>What each requested time needs.</b> <c>prev</c> = the latest stored sample at or before it, <c>next</c> = the
|
||||
/// first stored sample after it. <c>next</c> is only read when the evaluator uses it (a usable <c>prev</c> that is not
|
||||
/// an exact hit).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Read plan — O(1) reads for a dense batch, never two round-trips per timestamp by default.</b> The distinct
|
||||
/// requested times are walked in ascending order with ONE forward window read <c>[t − 1 ms, max + 1 ms]</c> of at most
|
||||
/// <see cref="ReadRows"/> rows, which yields every in-window neighbour. The sample before the first time comes from a
|
||||
/// walk-back (below), the sample after the last time from one small forward probe. A window read that fills its budget
|
||||
/// (dense data, widely spread times) resolves the times before its last row; the next unresolved time either
|
||||
/// continues from that row (when the rows in between are expected to fit one read) or jumps there with a read that
|
||||
/// starts a density-sized margin before it (so its <c>prev</c> is normally in the same read). Ten times one second
|
||||
/// apart on a 1 Hz tag cost 1 read (all exact hits) or 3 (interpolated); ten times an hour apart about one read each.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Walk-back</b> (the gateway reads forward only), in two phases. <i>Gallop:</i> segments <c>(lo, hi]</c> stepping
|
||||
/// back from <c>hi</c>, each one forward read, the first sized from the density the window read showed (else
|
||||
/// <see cref="InitialWalkWidth"/>), an empty one growing the next ×4; after two empty segments the whole remainder down
|
||||
/// to the epoch is asked once — a complete answer ends the walk, a FULL one seeds the bisection with its last row.
|
||||
/// <i>Bisect</i> (once a stored row is known below <c>hi</c> — a full segment's last row, or the last row of an earlier
|
||||
/// window read): read <c>(mid, hi]</c>; empty ⇒ <c>hi = mid</c>; full ⇒ the known row moves up to its last row;
|
||||
/// complete with rows ⇒ its last row is the answer. Right after a full read the next segment is aimed at about a
|
||||
/// quarter of a read at the density it showed (one try — an empty aimed segment falls back to halving), and below
|
||||
/// 2 ms the segment is the whole stretch from the known row (never a 100 ns step).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Bounded.</b> One request may take at most <c>64 + 16 × distinct times</c> raw reads (<see cref="BaseReads"/>,
|
||||
/// <see cref="ReadsPerTime"/>); past it — or with a strict-gateway ceiling below <see cref="MinBudget"/> rows — the
|
||||
/// node fails <c>BadTooManyOperations</c> with a message naming the limit. A jump read (to a time far past the previous
|
||||
/// read) asks for only <see cref="JumpRows"/> rows, so a thousand times spread over ten years cost about a thousand
|
||||
/// small reads, not a thousand 1,024-row reads.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Wire shape.</b> Every request bound is a whole millisecond (AVEVA stores milliseconds and truncates request
|
||||
/// times); a row counts only strictly INSIDE its request — AVEVA's synthetic boundary row at the query start (the value
|
||||
/// in effect there, <c>Quality = 133</c>, or a NULL row) can never be taken for a stored sample, nor a possible row at
|
||||
/// the query end; the epoch and before hold nothing. A Bad row (a stored NULL arrives as value 0, <c>q=1 opc=0</c>,
|
||||
/// #690) is value-less. Every stream is drained, never abandoned (a cancelled gateway stream evicts its pooled
|
||||
/// historian session). Nothing is asked past <c>now + </c><see cref="RawAggregateReader.LookAheadHorizon"/> (the
|
||||
/// historian stores nothing further ahead, and the gateway's AVEVA may refuse a far-future end); a strict gateway's
|
||||
/// row ceiling (#691.3) is learned and honoured — a short read is ordinary paging here, never re-probed.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Values.</b> A gateway value is a Double or a String whatever the node's data type (AVEVA stores every historized
|
||||
/// type as a numeric analog — a Boolean as a 0/1 <c>Int2</c>). The node's type hint converts each row BEFORE the
|
||||
/// shared evaluator runs, so its own type rules apply exactly as on the SQL backends: Boolean (value ≠ 0) and String
|
||||
/// are stepped; an integer type is an Int64 — interpolated, rounded half away from zero — and the result is returned
|
||||
/// as the node's integer type; Float / Double (and no hint) interpolate linearly as Double.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class GatewayAtTimeReader
|
||||
{
|
||||
/// <summary>The row budget of a window read (the first, or one continuing from a full one) and a walk-back segment.</summary>
|
||||
internal const int ReadRows = 1024;
|
||||
|
||||
/// <summary>The row budget of a jump read (a window read restarting at a time far past the previous read's last row).</summary>
|
||||
internal const int JumpRows = 64;
|
||||
|
||||
/// <summary>The row budget of the forward probe for the sample after the last time (boundary row + ties + one).</summary>
|
||||
internal const int ProbeRows = 2 + RawAggregateReader.PreWindowAllowance;
|
||||
|
||||
/// <summary>The smallest row budget a read can work with (the boundary row, a tie, one sample beyond).</summary>
|
||||
internal const int MinBudget = 4;
|
||||
|
||||
/// <summary>Raw reads one request may always take …</summary>
|
||||
internal const int BaseReads = 64;
|
||||
|
||||
/// <summary>
|
||||
/// … plus this many per distinct requested time. A runaway guard, not a budget (<c>HistoryReadDeadline</c> bounds the
|
||||
/// wall clock): a time inside a gap of a bursty tag legitimately costs a jump read plus a bisection back across the gap
|
||||
/// (≈ 6 reads — a 1 h/day tag asked at noon on consecutive days), so 4 refused requests AVEVA's own AtTime served.
|
||||
/// </summary>
|
||||
internal const int ReadsPerTime = 16;
|
||||
|
||||
/// <summary>The first walk-back segment's width when the window read showed no density.</summary>
|
||||
internal static readonly TimeSpan InitialWalkWidth = TimeSpan.FromMinutes(1);
|
||||
|
||||
/// <summary>The widest walk-back segment sized from density, before ×4 growth.</summary>
|
||||
private static readonly TimeSpan MaxSeededWidth = TimeSpan.FromDays(1);
|
||||
|
||||
private const long Ms = TimeSpan.TicksPerMillisecond;
|
||||
|
||||
private static readonly TimeSpan OneMs = TimeSpan.FromMilliseconds(1);
|
||||
|
||||
private static readonly DateTime Floor = RawAggregateReader.ProbeFloor;
|
||||
|
||||
/// <summary>One drained read: the rows strictly inside the request (ascending) and whether the stream ended short.</summary>
|
||||
private readonly record struct Read(List<StoredSample> Rows, bool Complete);
|
||||
|
||||
/// <summary>Evaluates AtTime for <paramref name="timestampsUtc"/> (same length and order, duplicates included).</summary>
|
||||
/// <param name="client">The gateway client.</param>
|
||||
/// <param name="ceiling">The data source's learned gateway row ceiling (#691.3).</param>
|
||||
/// <param name="tag">Historian tag name.</param>
|
||||
/// <param name="timestampsUtc">Requested times (any kind; treated as UTC, as on the SQL backends).</param>
|
||||
/// <param name="now">Server timestamp, the beyond-now rule and the look-ahead horizon.</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <param name="valueType">The node's scalar value type (the hinted <c>IHistorianDataSource.ReadAtTimeAsync</c>), or null.</param>
|
||||
/// <param name="maxReads">The request's raw-read cap; null = <see cref="BaseReads"/> + <see cref="ReadsPerTime"/> per distinct time.</param>
|
||||
/// <returns>One value per requested time, stamped with it.</returns>
|
||||
/// <exception cref="HistoryReadLimitExceededException">More than <see cref="ReadRows"/> rows within one millisecond, the
|
||||
/// read cap, or a strict gateway ceiling below <see cref="MinBudget"/>.</exception>
|
||||
public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IHistorianGatewayClient client, GatewayRowCeiling ceiling,
|
||||
string tag, IReadOnlyList<DateTime> timestampsUtc, DateTime now, CancellationToken ct, TypeCode? valueType = null,
|
||||
int? maxReads = null)
|
||||
{
|
||||
var requested = timestampsUtc.Select(t => DateTime.SpecifyKind(t, DateTimeKind.Utc)).ToArray();
|
||||
if (requested.Length == 0) return [];
|
||||
var times = requested.Distinct().Order().ToArray();
|
||||
var ctx = new Ctx(client, ceiling, tag, ct, valueType, maxReads ?? BaseReads + ReadsPerTime * times.Length);
|
||||
var neighbours = await ResolveAsync(ctx, times, now).ConfigureAwait(false);
|
||||
var result = new DataValueSnapshot[requested.Length];
|
||||
for (var i = 0; i < requested.Length; i++)
|
||||
{
|
||||
var t = requested[i];
|
||||
var (prev, next) = neighbours[t.Ticks];
|
||||
// A DateTime node is stepped (as on the SQL backends): a usable non-numeric next sends the evaluator down its
|
||||
// stepped branch, with the next's own status.
|
||||
if (valueType == TypeCode.DateTime && next is { IsUsable: true } n)
|
||||
next = n with { Value = HistorianValue.FromString(string.Empty) };
|
||||
result[i] = ToNodeType(AtTimeEvaluator.Evaluate(t, prev, next, now, now), valueType);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>One request's read context: the client, the type hint, and the read count against its cap.</summary>
|
||||
private sealed class Ctx(IHistorianGatewayClient client, GatewayRowCeiling ceiling, string tag, CancellationToken ct,
|
||||
TypeCode? valueType, int maxReads)
|
||||
{
|
||||
public IHistorianGatewayClient Client { get; } = client;
|
||||
public GatewayRowCeiling Ceiling { get; } = ceiling;
|
||||
public string Tag { get; } = tag;
|
||||
public CancellationToken Ct { get; } = ct;
|
||||
public TypeCode? ValueType { get; } = valueType;
|
||||
public int MaxReads { get; } = maxReads;
|
||||
public int Reads { get; set; }
|
||||
|
||||
/// <summary>The smallest budget a read was served with (a strict gateway's ceiling shows here).</summary>
|
||||
public int SmallestBudget { get; set; } = int.MaxValue;
|
||||
}
|
||||
|
||||
/// <summary>True for the integer type codes (evaluated as Int64, returned as the node's own type).</summary>
|
||||
private static bool IsInteger(TypeCode? t) =>
|
||||
t is TypeCode.SByte or TypeCode.Byte or TypeCode.Int16 or TypeCode.UInt16 or TypeCode.Int32 or TypeCode.UInt32
|
||||
or TypeCode.Int64 or TypeCode.UInt64;
|
||||
|
||||
/// <summary>
|
||||
/// An evaluated Int64 (integer hint) returned as the node's integer type. A value the type cannot hold (only possible
|
||||
/// for a stored value already outside the node's range) stays the evaluator's Int64.
|
||||
/// </summary>
|
||||
private static DataValueSnapshot ToNodeType(DataValueSnapshot v, TypeCode? valueType)
|
||||
{
|
||||
if (v.Value is not long l || !IsInteger(valueType) || valueType == TypeCode.Int64) return v;
|
||||
try
|
||||
{
|
||||
return v with { Value = Convert.ChangeType(l, valueType!.Value, System.Globalization.CultureInfo.InvariantCulture) };
|
||||
}
|
||||
catch (OverflowException)
|
||||
{
|
||||
return v;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>prev / next for every time of <paramref name="times"/> (distinct, ascending).</summary>
|
||||
private static async Task<Dictionary<long, (StoredSample? Prev, StoredSample? Next)>> ResolveAsync(
|
||||
Ctx ctx, DateTime[] times, DateTime now)
|
||||
{
|
||||
var result = new Dictionary<long, (StoredSample?, StoredSample?)>(times.Length);
|
||||
var horizon = MsFloor(DateTime.MaxValue - now > RawAggregateReader.LookAheadHorizon
|
||||
? DateTime.SpecifyKind(now + RawAggregateReader.LookAheadHorizon, DateTimeKind.Utc)
|
||||
: DateTime.MaxValue.AddTicks(-Ms));
|
||||
|
||||
// Nothing is stored at or before the epoch.
|
||||
var i = 0;
|
||||
for (; i < times.Length && times[i] <= Floor; i++) result[times[i].Ticks] = (null, null);
|
||||
if (i == times.Length) return result;
|
||||
|
||||
// Reads are planned on times clamped to the horizon (nothing is stored beyond it, and no read ends past it).
|
||||
var qe = Min(MsFloor(Min(times[^1], horizon)) + OneMs, horizon);
|
||||
var qs = RequestStart(Min(times[i], horizon));
|
||||
var budget = ReadRows;
|
||||
StoredSample? before = null; // the latest stored sample at or before qs …
|
||||
var beforeKnown = false; // … once known
|
||||
DateTime? walkKnown = null; // a stored sample exists here (at or before qs): the walk-back bisects above it
|
||||
var walkWidth = InitialWalkWidth;
|
||||
var probeFrom = -1; // first index whose next lies at or after qe (complete last read)
|
||||
|
||||
while (true)
|
||||
{
|
||||
var read = await ReadTrustedAsync(ctx, qs, qe, budget).ConfigureAwait(false);
|
||||
var rows = read.Rows;
|
||||
if (!read.Complete && rows.Count == 0)
|
||||
throw TooDense(ctx, qs);
|
||||
if (rows.Count >= 2 && !beforeKnown && walkKnown is null)
|
||||
walkWidth = Clamp(TimeSpan.FromTicks(4 * (rows[^1].TimeUtc - rows[0].TimeUtc).Ticks / (rows.Count - 1)));
|
||||
|
||||
// A short stream saw every row up to qe; a full one every row before its last row's time x.
|
||||
var x = read.Complete ? qe : rows[^1].TimeUtc;
|
||||
var j = -1; // index of the last row at or before the current time
|
||||
var k = i;
|
||||
for (; k < times.Length; k++)
|
||||
{
|
||||
var t = times[k];
|
||||
var tr = Min(t, horizon);
|
||||
if (!read.Complete && tr >= x) break;
|
||||
while (j + 1 < rows.Count && rows[j + 1].TimeUtc <= tr) j++;
|
||||
|
||||
StoredSample? prev;
|
||||
if (j >= 0) prev = rows[j];
|
||||
else
|
||||
{
|
||||
if (!beforeKnown)
|
||||
{
|
||||
before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false);
|
||||
beforeKnown = true;
|
||||
}
|
||||
prev = before;
|
||||
}
|
||||
|
||||
StoredSample? next = null;
|
||||
if (NeedsNext(t, prev) && t < horizon)
|
||||
{
|
||||
if (j + 1 < rows.Count) next = rows[j + 1];
|
||||
else if (probeFrom < 0) probeFrom = k; // after every row of a complete read: the probe answers it
|
||||
}
|
||||
result[t.Ticks] = (prev, next);
|
||||
}
|
||||
|
||||
if (k == times.Length) break;
|
||||
|
||||
// Truncated before times[k]: continue from the last row, or jump to times[k] with a density-sized margin and a
|
||||
// small budget (it needs that time's neighbours, and whatever later times follow closely).
|
||||
var tn = Min(times[k], horizon);
|
||||
var spanTicks = Math.Max(1, (x - rows[0].TimeUtc).Ticks);
|
||||
var density = (double)rows.Count / spanTicks; // rows per tick
|
||||
var gapRows = density * (tn - x).Ticks;
|
||||
var margin = Clamp(TimeSpan.FromTicks((long)Math.Min(TimeSpan.MaxValue.Ticks / 2.0, JumpRows / 4.0 / Math.Max(density, 1e-18))));
|
||||
var resume = MsFloor(x) - OneMs;
|
||||
var jump = gapRows >= ReadRows / 2.0;
|
||||
var nextQs = jump ? Max(RequestStart(tn - Min(margin, tn - Floor)), resume) : resume;
|
||||
if (nextQs <= qs) throw TooDense(ctx, x);
|
||||
|
||||
if (nextQs < x)
|
||||
{
|
||||
// Every row before x was seen: the latest at or before nextQs is known.
|
||||
var lastIndex = rows.FindLastIndex(r => r.TimeUtc <= nextQs);
|
||||
if (lastIndex >= 0)
|
||||
{
|
||||
before = rows[lastIndex];
|
||||
beforeKnown = true;
|
||||
}
|
||||
walkKnown = null;
|
||||
}
|
||||
else
|
||||
{
|
||||
beforeKnown = false;
|
||||
before = null;
|
||||
walkKnown = x;
|
||||
walkWidth = margin;
|
||||
}
|
||||
budget = jump ? JumpRows : ReadRows;
|
||||
qs = nextQs;
|
||||
i = k;
|
||||
}
|
||||
|
||||
if (probeFrom >= 0 && qe < horizon)
|
||||
{
|
||||
var after = await FirstAfterAsync(ctx, qe - OneMs, horizon, Min(times[^1], horizon)).ConfigureAwait(false);
|
||||
if (after is not null)
|
||||
for (var k = probeFrom; k < times.Length; k++)
|
||||
{
|
||||
var t = times[k];
|
||||
var (prev, next) = result[t.Ticks];
|
||||
if (next is null && t < horizon && NeedsNext(t, prev) && after.Value.TimeUtc > t)
|
||||
result[t.Ticks] = (prev, after);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>The evaluator reads <c>next</c> only for a usable <c>prev</c> that is not an exact hit.</summary>
|
||||
private static bool NeedsNext(DateTime t, StoredSample? prev) => prev is { IsUsable: true } p && p.TimeUtc != t;
|
||||
|
||||
/// <summary>One walk-back segment <c>(lo, hi]</c>: its oldest and newest rows, and whether the read was complete.</summary>
|
||||
private static async Task<(StoredSample? First, StoredSample? Last, bool Complete)> SegmentAsync(
|
||||
Ctx ctx, DateTime lo, DateTime hi)
|
||||
{
|
||||
var read = await ReadTrustedAsync(ctx, MsFloor(lo), MsFloor(hi) + OneMs, ReadRows).ConfigureAwait(false);
|
||||
StoredSample? first = null, last = null;
|
||||
foreach (var r in read.Rows)
|
||||
if (r.TimeUtc > lo && r.TimeUtc <= hi)
|
||||
{
|
||||
first ??= r;
|
||||
last = r;
|
||||
}
|
||||
return (first, last, read.Complete);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The latest stored sample with time in <c>(Floor, hi]</c> (see the type remarks: gallop, then bisect).
|
||||
/// <paramref name="known"/>: a stored sample exists at that time (≤ <paramref name="hi"/>) — the walk then bisects
|
||||
/// above it straight away, its first segment aimed at <paramref name="width"/>.
|
||||
/// </summary>
|
||||
private static async Task<StoredSample?> WalkBackAsync(Ctx ctx, DateTime hi, DateTime? known, TimeSpan width)
|
||||
{
|
||||
TimeSpan? aim = known is null ? null : width;
|
||||
|
||||
// Gallop: no stored row known below hi yet.
|
||||
var empties = 0;
|
||||
while (known is null)
|
||||
{
|
||||
if (hi <= Floor) return null;
|
||||
var rest = empties >= 2;
|
||||
var lo = rest || (hi - Floor).Ticks <= width.Ticks ? Floor : hi - width;
|
||||
var seg = await SegmentAsync(ctx, lo, hi).ConfigureAwait(false);
|
||||
if (seg.Complete)
|
||||
{
|
||||
if (seg.Last is { } found) return found;
|
||||
if (lo <= Floor) return null;
|
||||
hi = lo;
|
||||
empties++;
|
||||
width = Grow(width);
|
||||
continue;
|
||||
}
|
||||
// Full — a whole-remainder read of deep history too: its last row is a stored row below hi. Bisect above it.
|
||||
if (seg.Last is not { } seen) throw TooDense(ctx, lo);
|
||||
known = seen.TimeUtc;
|
||||
aim = AimFrom(seg.First!.Value.TimeUtc, seen.TimeUtc);
|
||||
}
|
||||
|
||||
// Bisect: a stored row exists at `known`, nothing newer than `hi` matters; the answer is the newest row in [known, hi].
|
||||
var aimMissed = false;
|
||||
while (true)
|
||||
{
|
||||
var k = known.Value;
|
||||
DateTime lo;
|
||||
var aimed = false;
|
||||
if (hi - k < TimeSpan.FromMilliseconds(2))
|
||||
lo = k.AddTicks(-1); // the whole stretch from the known row: never a sub-millisecond step
|
||||
else
|
||||
{
|
||||
lo = k + TimeSpan.FromTicks((hi - k).Ticks / 2);
|
||||
if (aim is { } a && hi - a > lo)
|
||||
{
|
||||
lo = hi - a; // an aimed try right after a full read — until one comes back empty (a gap: halve from then on)
|
||||
aimed = true;
|
||||
}
|
||||
}
|
||||
aim = null;
|
||||
|
||||
var seg = await SegmentAsync(ctx, lo, hi).ConfigureAwait(false);
|
||||
if (seg.Complete)
|
||||
{
|
||||
if (seg.Last is { } found) return found;
|
||||
if (lo < k) return null; // the known row is gone (deleted meanwhile): nothing is left to find
|
||||
hi = lo;
|
||||
aimMissed |= aimed;
|
||||
continue;
|
||||
}
|
||||
if (seg.Last is not { } seen || seen.TimeUtc <= k) throw TooDense(ctx, k);
|
||||
known = seen.TimeUtc;
|
||||
if (!aimMissed) aim = AimFrom(seg.First!.Value.TimeUtc, seen.TimeUtc);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>About a quarter of a read at the density a full read showed (its rows from <paramref name="first"/> to <paramref name="last"/>).</summary>
|
||||
private static TimeSpan AimFrom(DateTime first, DateTime last) =>
|
||||
TimeSpan.FromTicks(Math.Max(Ms, (last - first).Ticks / 4));
|
||||
|
||||
/// <summary>The first stored sample after <paramref name="after"/> in <c>(qs, end)</c>, from a small forward read.</summary>
|
||||
private static async Task<StoredSample?> FirstAfterAsync(Ctx ctx, DateTime qs, DateTime end, DateTime after)
|
||||
{
|
||||
foreach (var budget in new[] { ProbeRows, ReadRows })
|
||||
{
|
||||
var read = await ReadTrustedAsync(ctx, qs, end, budget).ConfigureAwait(false);
|
||||
foreach (var r in read.Rows)
|
||||
if (r.TimeUtc > after) return r;
|
||||
if (read.Complete) return null;
|
||||
}
|
||||
throw TooDense(ctx, qs);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One forward read <c>[qs, qe]</c> (whole milliseconds), drained, keeping only rows strictly inside it and after the
|
||||
/// epoch — counted against the request's read cap. A strict gateway's ceiling is learned on refusal and then asked for
|
||||
/// directly; a full read is not re-probed (the caller pages on, it does not need the rest of this read).
|
||||
/// </summary>
|
||||
private static async Task<Read> ReadTrustedAsync(Ctx ctx, DateTime qs, DateTime qe, int budget)
|
||||
{
|
||||
if (++ctx.Reads > ctx.MaxReads)
|
||||
throw new HistoryReadLimitExceededException(ctx.MaxReads,
|
||||
$"The AtTime read of '{ctx.Tag}' needed more than {ctx.MaxReads} raw reads ({BaseReads} + {ReadsPerTime} per " +
|
||||
"requested time)" + CeilingNote(ctx) + "; the Gateway backend serves AtTime from raw reads — request fewer times.");
|
||||
var (read, used) = await ctx.Ceiling.ReadAsync(budget, b => PassAsync(ctx, qs, qe, b), (_, _) => false)
|
||||
.ConfigureAwait(false);
|
||||
if (used < ctx.SmallestBudget) ctx.SmallestBudget = used;
|
||||
if (used < MinBudget)
|
||||
throw new HistoryReadLimitExceededException(used,
|
||||
$"The HistorianGateway's row ceiling ({used}, its Historian:ReadLimits:MaxValuesCeiling with " +
|
||||
$"AllowStreamingBeyondCeiling=false) is below the {MinBudget} rows an AtTime raw read of '{ctx.Tag}' needs.");
|
||||
return read;
|
||||
}
|
||||
|
||||
private static string CeilingNote(Ctx ctx) =>
|
||||
ctx.SmallestBudget < JumpRows
|
||||
? $", with reads bounded by the HistorianGateway's row ceiling of {ctx.SmallestBudget} (Historian:ReadLimits:MaxValuesCeiling)"
|
||||
: string.Empty;
|
||||
|
||||
private static async Task<Read> PassAsync(Ctx ctx, DateTime qs, DateTime qe, int budget)
|
||||
{
|
||||
var rows = new List<StoredSample>();
|
||||
var received = 0;
|
||||
await foreach (var raw in ctx.Client.ReadRawAsync(ctx.Tag, qs, qe, budget, ctx.Ct).ConfigureAwait(false))
|
||||
{
|
||||
received++; // drained to completion, never broken out of
|
||||
if (raw.Timestamp is null) continue;
|
||||
var t = raw.Timestamp.ToDateTime();
|
||||
if (t > qs && t < qe && t > Floor) rows.Add(ToStored(raw, t, ctx.ValueType));
|
||||
}
|
||||
return new Read(rows, received < budget);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A gateway row as a stored sample of the node's type (see the type remarks); a Bad row carries no value (#690: a
|
||||
/// stored NULL arrives as value 0).
|
||||
/// </summary>
|
||||
private static StoredSample ToStored(HistorianSample raw, DateTime t, TypeCode? valueType)
|
||||
{
|
||||
var status = SampleMapper.ToStatus(raw);
|
||||
var value = StatusBits.IsBad(status) ? HistorianValue.Empty
|
||||
: raw.HasNumericValue ? Typed(raw.NumericValue, valueType)
|
||||
: raw.HasStringValue ? HistorianValue.FromString(raw.StringValue)
|
||||
: HistorianValue.Empty;
|
||||
return StoredSample.FromStored(t, value, status);
|
||||
}
|
||||
|
||||
/// <summary>A gateway number as the hinted kind, so the shared evaluator's own type branch decides stepped vs linear.</summary>
|
||||
private static HistorianValue Typed(double v, TypeCode? valueType) => valueType switch
|
||||
{
|
||||
TypeCode.Boolean => HistorianValue.FromBoolean(v != 0),
|
||||
TypeCode.String => HistorianValue.FromString(v.ToString("R", System.Globalization.CultureInfo.InvariantCulture)),
|
||||
// In Int64 range only (2^63 is exactly representable); NaN, ±∞ or out of range is no integer at all — Empty, which
|
||||
// StoredSample.FromStored makes BadNoData (unusable), never a saturated Good value.
|
||||
_ when IsInteger(valueType) => v >= -9.2233720368547758E18 && v < 9.2233720368547758E18
|
||||
? HistorianValue.FromInt64((long)Math.Round(v, MidpointRounding.AwayFromZero))
|
||||
: HistorianValue.Empty,
|
||||
_ => HistorianValue.FromDouble(v),
|
||||
};
|
||||
|
||||
/// <summary>The window read's start for a time: a millisecond before its millisecond (never before the epoch).</summary>
|
||||
private static DateTime RequestStart(DateTime t)
|
||||
{
|
||||
var s = MsFloor(t) - TimeSpan.FromMilliseconds(1);
|
||||
return s < Floor ? Floor : s;
|
||||
}
|
||||
|
||||
private static DateTime MsFloor(DateTime t) => new(t.Ticks - t.Ticks % Ms, DateTimeKind.Utc);
|
||||
|
||||
private static DateTime Min(DateTime a, DateTime b) => a <= b ? a : b;
|
||||
|
||||
private static DateTime Max(DateTime a, DateTime b) => a >= b ? a : b;
|
||||
|
||||
private static TimeSpan Min(TimeSpan a, TimeSpan b) => a <= b ? a : b;
|
||||
|
||||
private static TimeSpan Clamp(TimeSpan w) =>
|
||||
w < TimeSpan.FromMilliseconds(1) ? TimeSpan.FromMilliseconds(1) : w > MaxSeededWidth ? MaxSeededWidth : w;
|
||||
|
||||
private static TimeSpan Grow(TimeSpan w) => w.Ticks > TimeSpan.MaxValue.Ticks / 4 ? TimeSpan.MaxValue : w * 4;
|
||||
|
||||
private static HistoryReadLimitExceededException TooDense(Ctx ctx, DateTime at)
|
||||
{
|
||||
var rows = Math.Min(ReadRows, ctx.SmallestBudget);
|
||||
return new(rows,
|
||||
$"The AtTime read of '{ctx.Tag}' met more than {rows} stored samples within about one millisecond of {at:O}" +
|
||||
(rows < ReadRows ? $" (reads are bounded by the HistorianGateway's row ceiling of {rows})" : string.Empty) +
|
||||
"; the Gateway backend serves AtTime from raw reads.");
|
||||
}
|
||||
}
|
||||
+64
-2
@@ -46,6 +46,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
private readonly ILogger<GatewayHistorianDataSource> _logger;
|
||||
private readonly int _maxRawRows;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly bool _atTimeFromRaw;
|
||||
|
||||
/// <summary>#693: AtTime from raw reads (true) or AVEVA's own interpolation (the kill switch).</summary>
|
||||
internal bool AtTimeFromRaw => _atTimeFromRaw;
|
||||
|
||||
/// <summary>A strict gateway's row ceiling, learned from its first refusal (#691.3).</summary>
|
||||
private readonly GatewayRowCeiling _ceiling = new();
|
||||
@@ -67,9 +71,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
/// <= 0 = unbounded). Past it the node fails <c>BadTooManyOperations</c>.
|
||||
/// </param>
|
||||
/// <param name="timeProvider">Clock for result server timestamps and the raw look-ahead horizon; null = system.</param>
|
||||
/// <param name="atTimeFromRaw">
|
||||
/// #693: serve AtTime from raw reads with the shared AtTime rules (<see cref="GatewayAtTimeReader"/>; default);
|
||||
/// <c>false</c> = AVEVA's own interpolation passed through (<c>ServerHistorian:GatewayAtTimeFromRaw</c> kill switch).
|
||||
/// </param>
|
||||
public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
|
||||
int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null)
|
||||
int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null,
|
||||
bool atTimeFromRaw = true)
|
||||
{
|
||||
_atTimeFromRaw = atTimeFromRaw;
|
||||
ArgumentNullException.ThrowIfNull(client);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
_client = client;
|
||||
@@ -364,7 +374,59 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
internal static bool IsMaxValuesCeilingRejection(Exception ex) => GatewayReadLimits.IsCeilingRejection(ex);
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// #693: served from RAW gateway reads — each requested time's stored sample at-or-before and after it — evaluated with
|
||||
/// the SQL backends' shared <see cref="AtTimeEvaluator"/> (Part 11 §6.5.4), so the Gateway answers what SqlServer and
|
||||
/// QuestDB answer for the same stored samples; see <see cref="GatewayAtTimeReader"/> for the read plan and its cost.
|
||||
/// AVEVA's own interpolated answer (the gateway's <c>ReadAtTime</c>) returned the previous value as Good at a stored
|
||||
/// NULL's time, null + Bad at a Good sample right after a NULL, a drifted value at an exact stored sample, and no
|
||||
/// Interpolated bit; it is only used when <c>ServerHistorian:GatewayAtTimeFromRaw = false</c> (kill switch).
|
||||
/// A gateway value is a Double (or a String) whatever the node's type (AVEVA stores a Boolean as a 0/1 <c>Int2</c>), so
|
||||
/// without a type hint every numeric tag is interpolated linearly; the node manager passes the node's type through
|
||||
/// <see cref="ReadAtTimeAsync(string, IReadOnlyList{DateTime}, TypeCode?, CancellationToken)"/>.
|
||||
/// </remarks>
|
||||
public Task<HistoryReadResult> ReadAtTimeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
||||
ReadAtTimeAsync(fullReference, timestampsUtc, valueType: null, cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// #693: the hint makes the AtTime rules type-exact, as on the SQL backends — the gateway's Double rows are converted to
|
||||
/// the node's type BEFORE the shared evaluator runs (Boolean = value ≠ 0 ⇒ stepped; an integer type ⇒ Int64 ⇒
|
||||
/// interpolated and rounded half away from zero, then returned as the node's integer type; String ⇒ stepped; Float /
|
||||
/// Double / no hint ⇒ linear Double). Ignored with the kill switch (AVEVA's own answer).
|
||||
/// </remarks>
|
||||
public async Task<HistoryReadResult> ReadAtTimeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, TypeCode? valueType, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(timestampsUtc);
|
||||
if (!_atTimeFromRaw) return await ReadAtTimeNativeAsync(fullReference, timestampsUtc, cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
var now = _time.GetUtcNow().UtcDateTime;
|
||||
var values = await GatewayAtTimeReader
|
||||
.ReadAsync(_client, _ceiling, fullReference, timestampsUtc, now, cancellationToken, valueType)
|
||||
.ConfigureAwait(false);
|
||||
RecordOutcome(success: true, error: null);
|
||||
return new HistoryReadResult(values, ContinuationPoint: null);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw; // caller-requested, not a backend fault
|
||||
}
|
||||
catch (HistoryReadLimitExceededException)
|
||||
{
|
||||
throw; // a client-request limit (a pathological tie cluster), not a backend fault
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The pre-#693 AtTime: AVEVA's own interpolation, aligned to the request (kill switch only).</summary>
|
||||
private async Task<HistoryReadResult> ReadAtTimeNativeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
@@ -494,7 +556,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
|
||||
/// <summary>
|
||||
/// Reconciles a gateway at-time reply against the requested timestamps to honour the
|
||||
/// <see cref="IHistorianDataSource.ReadAtTimeAsync"/> contract: exactly one snapshot per
|
||||
/// <see cref="IHistorianDataSource.ReadAtTimeAsync(string, IReadOnlyList{DateTime}, CancellationToken)"/> contract: exactly one snapshot per
|
||||
/// requested timestamp, in request order. Returned samples are indexed by timestamp ticks;
|
||||
/// any requested timestamp the gateway did not return is filled with a Bad-quality
|
||||
/// (<c>0x80000000</c>) snapshot stamped at the requested time rather than positionally
|
||||
|
||||
+3
-1
@@ -40,7 +40,9 @@ public static class GatewayHistorian
|
||||
HistorianGatewayClientAdapter.Create(options, loggerFactory),
|
||||
logger,
|
||||
// #681: the raw-evaluated per-sample aggregates share the SQL backends' raw-row cap.
|
||||
maxRawRows: options.MaxTimeWeightedRawRows);
|
||||
maxRawRows: options.MaxTimeWeightedRawRows,
|
||||
// #693: AtTime from raw reads + the shared evaluator; false = AVEVA's own interpolation (kill switch).
|
||||
atTimeFromRaw: options.GatewayAtTimeFromRaw);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -49,8 +49,10 @@ public static class AtTimeEvaluator
|
||||
{
|
||||
var fraction = (double)(t.Ticks - p.TimeUtc.Ticks) / (n.TimeUtc.Ticks - p.TimeUtc.Ticks);
|
||||
var v = p.Value.Number + (nv - p.Value.Number) * fraction;
|
||||
// Each arm boxed on its own: `cond ? (long)x : v` is a DOUBLE expression in C# (the long widens), which returned
|
||||
// every interpolated Int64 as a Double until #693.
|
||||
object value = p.Value.Kind == HistorianValueKind.Int64
|
||||
? (long)Math.Round(v, MidpointRounding.AwayFromZero)
|
||||
? (object)(long)Math.Round(v, MidpointRounding.AwayFromZero)
|
||||
: v;
|
||||
var status = p.IsGood && n.IsGood ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
|
||||
return new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), t, serverTime);
|
||||
|
||||
@@ -3399,6 +3399,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
return NodeId.Null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #693: the AtTime value-type hint of a historized variable — its DataType's built-in type as a
|
||||
/// <see cref="TypeCode"/> when the node is a scalar of a type the historian evaluates (Boolean, the integers, Float,
|
||||
/// Double, String, DateTime); <c>null</c> for an array, an unknown node, or any other type. A UNS mirror carries its
|
||||
/// backing raw tag's DataType, so either NodeId gives the same hint.
|
||||
/// </summary>
|
||||
/// <param name="nodeId">The HistoryRead request's node id.</param>
|
||||
/// <returns>The hint, or null.</returns>
|
||||
internal TypeCode? AtTimeValueType(NodeId nodeId)
|
||||
{
|
||||
if (!_variables.TryGetValue(MapKey(nodeId), out var variable) || variable.ValueRank != ValueRanks.Scalar) return null;
|
||||
return TypeInfo.GetBuiltInType(variable.DataType) switch
|
||||
{
|
||||
BuiltInType.Boolean => TypeCode.Boolean,
|
||||
BuiltInType.SByte => TypeCode.SByte,
|
||||
BuiltInType.Byte => TypeCode.Byte,
|
||||
BuiltInType.Int16 => TypeCode.Int16,
|
||||
BuiltInType.UInt16 => TypeCode.UInt16,
|
||||
BuiltInType.Int32 => TypeCode.Int32,
|
||||
BuiltInType.UInt32 => TypeCode.UInt32,
|
||||
BuiltInType.Int64 => TypeCode.Int64,
|
||||
BuiltInType.UInt64 => TypeCode.UInt64,
|
||||
BuiltInType.Float => TypeCode.Single,
|
||||
BuiltInType.Double => TypeCode.Double,
|
||||
BuiltInType.String => TypeCode.String,
|
||||
BuiltInType.DateTime => TypeCode.DateTime,
|
||||
_ => null,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>Map a Tag.DataType string ("Boolean", "Int32", "Float", "Double", "String",
|
||||
/// "DateTime") to the OPC UA built-in NodeId. Unknown names fall back to BaseDataType
|
||||
/// (matches CreateVariable's default for lazy-created nodes).</summary>
|
||||
@@ -5177,9 +5207,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
var work = new List<Func<Task>>(nodesToProcess.Count);
|
||||
foreach (var handle in nodesToProcess)
|
||||
{
|
||||
// #693: the node's scalar value type, for a backend whose stored values carry none (the Gateway).
|
||||
var valueType = AtTimeValueType(handle.NodeId);
|
||||
work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync(
|
||||
tagname,
|
||||
timestamps,
|
||||
valueType,
|
||||
token), ct));
|
||||
}
|
||||
|
||||
|
||||
@@ -97,6 +97,15 @@ public sealed class ServerHistorianOptions
|
||||
/// </summary>
|
||||
public bool HonorReturnBounds { get; init; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #693, <c>Gateway</c> backend only: whether HistoryRead-AtTime is served from RAW gateway reads evaluated with
|
||||
/// the SQL backends' shared AtTime rules (exact stored value at a stored time, linear between usable neighbours with
|
||||
/// the Interpolated bit, held before a Bad neighbour as Uncertain, null + the outage's Bad code inside an outage,
|
||||
/// <c>BadNoData</c> before the first sample). Default <c>true</c>. <c>false</c> is the kill switch back to the
|
||||
/// pre-#693 answer: AVEVA's own interpolation (the gateway's <c>ReadAtTime</c>), passed through unchanged.
|
||||
/// </summary>
|
||||
public bool GatewayAtTimeFromRaw { get; init; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// Maximum number of historized nodes served CONCURRENTLY within one HistoryRead batch (arch-review
|
||||
/// 03/S3). The SDK's <c>HistoryRead*</c> override surface is synchronous, so each arm block-bridges
|
||||
|
||||
Reference in New Issue
Block a user