fix(historian): #691.2 review — forward native TimeAverage/Total on the SQL grid: clipped latest interval read over its exact span, Partial, interval clamped to the window
This commit is contained in:
Joseph Doherty
@@ -1365,7 +1365,7 @@ historian tagname, which is the tag's driver `FullName`.
|
||||
The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the
|
||||
`GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default
|
||||
(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early (epoch-floored) with `NumValuesPerNode + 16` rows (unclamped; 0 ⇒ `int.MaxValue`) and filtered by timestamp — so an empty window is `GoodNoData`, not a fabricated value at StartTime; the forward end is now exclusive (a point read inclusive); pre-window rows that crowd the budget trigger ONE re-read with room for them, else `BadTooManyOperations`; `ReturnBounds` is ignored on every backend; ⚠️ #690: a Gateway Bad row — a stored NULL, which histsdk sends as value 0 — now reads value **null** + Bad on Raw and AtTime, as on the SQL backends; ⚠️ #691.3: a strict gateway (`AllowStreamingBeyondCeiling=false`) that refuses a raw budget is re-asked ONCE at the ceiling its refusal names — raw and per-sample reads alike; the ceiling is then remembered per data source (`GatewayRowCeiling`: refusals count as gateway errors), re-learned from a refusal when lowered, and re-probed once at the full budget when a ceiling-bounded read comes back full (raised / streaming) — so an uncapped (`NumValuesPerNode = 0`) read of a small window is served; a ceiling-bounded stream that is not provably complete is `BadTooManyOperations`, never a truncated answer), Processed
|
||||
(backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27; ⚠️ #691.2: a REVERSE native read — refused by the gateway before — is now served by forward native requests on the later-anchored #677 grid (whole intervals in one read, the clipped earliest interval as its own exact-span read, Partial unless Bad), newest first, later-edge stamped); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over
|
||||
(backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27; ⚠️ #691.2: native reads use the SQL backends' grid in both directions via forward requests only — a REVERSE read (refused by the gateway before) is anchored at its later time (#677), and the clipped interval (earliest reverse, LATEST forward — which used to be averaged past EndTime) is its own exact-span read marked Partial unless Bad; an interval wider than the window is sent as the window, interval 0 as one interval; until historiangw#27 lands both directions return Good-status wrong values); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over
|
||||
alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node);
|
||||
authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization.
|
||||
|
||||
|
||||
+20
-13
@@ -886,19 +886,26 @@ fixed:
|
||||
`HistorianTagDefinition.IntegralDivisor` is `0`, so `GatewayTagProvisioner` now sends **`IntegralDivisor = 1`** explicitly on every
|
||||
`EnsureTags` — the value·seconds unit the SQL backends' `Total` uses. A tag provisioned earlier keeps its divisor until its next
|
||||
`EnsureTags` (an upsert). The live magnitude check against AVEVA (a constant 1 integrated over 1 s reads 1) passed on 2026-10-12 at the AVEVA
|
||||
level; end to end it waits on historiangw#27 (above). ⚠️ **Reverse native reads (#691.2, client-visible).** A reverse processed read (`StartTime > EndTime`) of the AVEVA-native
|
||||
`TimeAverage`/`Total` used to reach the gateway as `start > end` and fail the node (`InvalidArgument`, counted as a backend failure). It
|
||||
now follows #677 like the SQL backends and the raw-evaluated per-sample aggregates (#681): window `(EndTime, StartTime]`, intervals
|
||||
stepping back from `StartTime`, the clipped interval the **earliest**, each stamped with its **later** edge, newest first. It is served
|
||||
by **forward** native requests on that later-anchored grid (`NativeReverseAggregatePlan` — AVEVA anchors a native read's grid at the
|
||||
request start): one read `[O + I·f, StartTime)` with the interval `I` for the whole intervals (`O = StartTime − n·I`), and — when the
|
||||
window is not a whole number of intervals — a second read over exactly the clipped earliest span `[EndTime, O + I)` with its own width
|
||||
as the interval, so AVEVA never averages time before the window. A time-weighted value over `[a, b)` and over `(a, b]` is the same
|
||||
integral (two instants apart), so each forward row is the reverse interval's value; rows are mapped exactly as forward native rows
|
||||
(#685) and the clipped interval carries the **Partial** bit unless Bad (Part 13 §5.3.3.2 Case 3, as on the SQL backends — the forward
|
||||
native read still sets no Partial bit). An interval AVEVA answers no row for is absent, as on a forward native read. ⚠️ Pinned against
|
||||
a native-mode fake only: under historiangw#27 AVEVA answers one row per request today, so end to end a reverse native read is as wrong
|
||||
as a forward one until the gateway is fixed — re-run the reverse leg then. Since #681 every native bucket is stamped with its interval **start**
|
||||
level; end to end it waits on historiangw#27 (above). ⚠️ **Native reads on the SQL backends' interval grid (#691.2, client-visible).** Both directions now use `BucketPlan`, served by
|
||||
**forward** native requests only (`NativeAggregatePlan` — AVEVA anchors a native read's interval grid at the request start, and the
|
||||
gateway refuses `start > end` and `interval <= 0`):
|
||||
- **Reverse** (`StartTime > EndTime`) used to reach the gateway as `start > end` and fail the node (`InvalidArgument`, counted as a backend
|
||||
failure). It now follows #677 like the SQL backends and the raw-evaluated per-sample aggregates (#681): window `(EndTime, StartTime]`,
|
||||
intervals stepping back from `StartTime`, the clipped interval the **earliest**, each stamped with its **later** edge, newest first —
|
||||
one read `[O + I·f, StartTime)` with the interval `I` for the whole intervals (`O = StartTime − n·I`, `f` = 1 when clipped) and a
|
||||
second read over exactly the clipped span `[EndTime, O + I)` with its own width as the interval. A time-weighted value over `[a, b)` and
|
||||
over `(a, b]` is the same integral (two instants apart), so each forward row is the reverse interval's value.
|
||||
- **Forward** windows that are not a whole number of intervals used to send one request whose last AVEVA interval ran **past EndTime**;
|
||||
now the whole intervals `[StartTime, StartTime + (n−1)·I)` are one read and the clipped **latest** interval `[StartTime + (n−1)·I,
|
||||
EndTime)` is its own read over exactly that span. An interval wider than the window is sent as the window, and `interval = 0` as one
|
||||
interval over the window (it used to reach the gateway as 0 and be refused). An empty window (`StartTime == EndTime`) sends nothing.
|
||||
- Stamps match SqlServer/QuestDB (forward: earlier edge, ascending; reverse: later edge, descending), and the clipped interval carries
|
||||
the **Partial** bit unless Bad (Part 13 §5.3.3.2 Case 3, as the SQL backends' time-weighted aggregates — which also mark the intervals
|
||||
the tag's data starts/ends inside; AVEVA's rows carry no data range, so that is not reproduced). Rows are otherwise mapped as before
|
||||
(#685). An interval AVEVA answers no row for is absent (the SQL backends answer it `BadNoData`).
|
||||
- ⚠️ **Until historiangw#27 is fixed, native values are wrong in BOTH directions with a Good status:** AVEVA answers one row per request
|
||||
(the whole span's value), which lands on that request's first interval as a Good, Calculated value while the other intervals are
|
||||
absent. Pinned against a native-mode fake only — re-run the forward and reverse native legs live once the gateway is fixed. Since #681 every native bucket is stamped with its interval **start**
|
||||
(Part 13; it was the interval end) and carries the Calculated info bits (`Good` ⇒ `0x00000401`, no data ⇒ `BadNoData | 0x401`).
|
||||
The live AVEVA gate (2026-10-12) confirmed `start_time` is the interval start (and the `value·s` unit). Previously noted as
|
||||
pending: it reads the aggregate row's `start_time`, which histsdk decodes from a row
|
||||
|
||||
+12
-39
@@ -253,9 +253,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
/// <para>
|
||||
/// <b>TimeAverage / Total</b> stay native AVEVA retrieval modes (<see cref="AggregateModeMapper"/>), stamped with
|
||||
/// the interval start and the Calculated bit; AVEVA's extra row for the interval before the window is dropped and
|
||||
/// an Uncertain interval keeps its value (#685, <see cref="SampleMapper.ToAggregateSnapshots"/>). A reverse read
|
||||
/// (<c>start > end</c>, which the gateway refuses) is served by forward requests on the later-anchored #677 grid
|
||||
/// (#691.2, <see cref="NativeReverseAggregatePlan"/>). Everything else is <see cref="HistoryAggregateNotSupportedException"/>
|
||||
/// an Uncertain interval keeps its value (#685). Both directions use the SQL backends' interval grid with forward
|
||||
/// requests only — a reverse read (which the gateway refuses) anchored at its later time (#677), the clipped interval
|
||||
/// read over exactly its span and marked Partial (#691.2, <see cref="NativeAggregatePlan"/>). Everything else is <see cref="HistoryAggregateNotSupportedException"/>
|
||||
/// (see <see cref="ServedAggregates"/>).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
@@ -272,50 +272,23 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
// Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
|
||||
// request the server answers BadAggregateNotSupported, not a backend fault to count (#285).
|
||||
var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
|
||||
if (startUtc > endUtc)
|
||||
return await ReadNativeReverseAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), mode, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
// TimeAverage / Total are native gateway retrieval modes (TimeWeightedAverage / Integral) — no
|
||||
// client-side scaling.
|
||||
var buckets = new List<HistorianAggregateSample>();
|
||||
await foreach (var bucket in _client
|
||||
.ReadAggregateAsync(fullReference, startUtc, endUtc, mode, interval, cancellationToken)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
buckets.Add(bucket);
|
||||
}
|
||||
|
||||
// Filtered while mapping, after the stream was drained (#685: AVEVA's extra pre-window row).
|
||||
var snapshots = SampleMapper.ToAggregateSnapshots(buckets, startUtc, endUtc);
|
||||
RecordOutcome(success: true, error: null);
|
||||
return new HistoryReadResult(snapshots, ContinuationPoint: null);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw; // caller-requested, not a backend fault
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
throw;
|
||||
}
|
||||
// Planned OUTSIDE the failure-recording try: too many intervals is a client request limit, not a backend fault.
|
||||
return await ReadNativeAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), mode, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A reverse (<c>start > end</c>) native TimeAverage / Total read (#691.2), anchored at its later time as on the SQL
|
||||
/// backends (#677) — the gateway refuses <c>start > end</c>, so it is served by forward native requests on the
|
||||
/// later-anchored grid. See <see cref="NativeReverseAggregatePlan"/>.
|
||||
/// A native TimeAverage / Total read on the SQL backends' interval grid, either direction (#691.2): forward native
|
||||
/// requests only, the clipped interval read over exactly its span. See <see cref="NativeAggregatePlan"/>.
|
||||
/// </summary>
|
||||
private async Task<HistoryReadResult> ReadNativeReverseAsync(string fullReference, BucketPlan plan, RetrievalMode mode,
|
||||
private async Task<HistoryReadResult> ReadNativeAsync(string fullReference, BucketPlan plan, RetrievalMode mode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null);
|
||||
try
|
||||
{
|
||||
var byBucket = new Dictionary<long, DataValueSnapshot>();
|
||||
foreach (var request in NativeReverseAggregatePlan.Requests(plan))
|
||||
foreach (var request in NativeAggregatePlan.Requests(plan))
|
||||
{
|
||||
var buckets = new List<HistorianAggregateSample>();
|
||||
await foreach (var bucket in _client
|
||||
@@ -324,11 +297,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
{
|
||||
buckets.Add(bucket); // drained to completion, mapped afterwards
|
||||
}
|
||||
NativeReverseAggregatePlan.Place(plan, request, buckets, byBucket);
|
||||
NativeAggregatePlan.Place(plan, request, buckets, byBucket);
|
||||
}
|
||||
|
||||
RecordOutcome(success: true, error: null);
|
||||
return new HistoryReadResult(NativeReverseAggregatePlan.Assemble(plan, byBucket), ContinuationPoint: null);
|
||||
return new HistoryReadResult(NativeAggregatePlan.Assemble(plan, byBucket), ContinuationPoint: null);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
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>
|
||||
/// Serves a processed read of the AVEVA-native <c>TimeAverage</c> / <c>Total</c> on the SQL backends' interval grid
|
||||
/// (<see cref="BucketPlan"/>), in either direction: forward — window <c>[start, end)</c>, intervals stepping forward
|
||||
/// from <c>start</c>, the clipped interval the LATEST, each stamped with its earlier edge, oldest first; reverse (#677,
|
||||
/// #691.2) — window <c>(end, start]</c>, intervals stepping back from <c>start</c>, the clipped interval the EARLIEST,
|
||||
/// each stamped with its later edge, newest first. The gateway refuses <c>start > end</c> and <c>interval <= 0</c>,
|
||||
/// so every request it sees is forward with a positive interval.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Forward requests on the plan's grid.</b> AVEVA anchors a native read's interval grid at the request start (live,
|
||||
/// 2026-10-12: rows at <c>start + j·interval</c>, plus one extra row for the interval before it, #685). So the plan's
|
||||
/// whole intervals are requested as ONE forward read spanning exactly them with the interval <c>I</c>, and the
|
||||
/// clipped interval (when the window is not a whole number of intervals) as a second forward read over exactly its
|
||||
/// span with its own width as the interval — AVEVA never averages time outside the window (before it on a reverse
|
||||
/// read, past <c>EndTime</c> on a forward one). An interval wider than the window is sent as the window, and
|
||||
/// <c>interval <= 0</c> as one interval over the whole window (it used to reach the gateway as 0 and be refused).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Soundness of reverse.</b> A time-weighted average / integral over <c>[a, b)</c> and over <c>(a, b]</c> integrate
|
||||
/// the same function over sets differing by two instants, so a forward native row IS the value of the reverse
|
||||
/// interval with the same edges. The one place the half-open side could matter is AVEVA's status for an interval
|
||||
/// whose only non-good sample lies exactly on an edge — accepted, as on every native row.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Mapping.</b> Each row is mapped as before (<see cref="SampleMapper.ToAggregateSnapshot"/>: Uncertain keeps its
|
||||
/// value, no data ⇒ <c>BadNoData</c>, Calculated bits), placed by its interval start (else end), restamped with
|
||||
/// <see cref="BucketPlan.BucketTimestamp"/>, and the clipped interval carries the Partial bit unless Bad (Part 13
|
||||
/// §5.3.3.2 Case 3, as the SQL backends' time-weighted aggregates — which additionally mark the intervals the data
|
||||
/// starts/ends inside; AVEVA's rows carry no data range, so those are not reproduced). An interval AVEVA answers no
|
||||
/// row for is absent (not a fabricated <c>BadNoData</c>, unlike the SQL backends).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// ⚠️ Today historiangw#27 (the interval reaches AVEVA in ticks where it reads milliseconds) makes AVEVA answer ONE
|
||||
/// Good row per request — the whole-intervals read lands on its earliest interval with a whole-span value and the rest
|
||||
/// are absent — in both directions. The mapping needs no change once the gateway is fixed.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class NativeAggregatePlan
|
||||
{
|
||||
/// <summary>One forward native request and the ascending bucket index of its first interval.</summary>
|
||||
/// <param name="StartUtc">Forward request start (inclusive).</param>
|
||||
/// <param name="EndUtc">Forward request end.</param>
|
||||
/// <param name="Interval">The request's interval (> 0).</param>
|
||||
/// <param name="FirstBucket">Ascending plan bucket of the request's first interval.</param>
|
||||
/// <param name="BucketCount">Number of plan buckets the request covers.</param>
|
||||
internal readonly record struct Request(DateTime StartUtc, DateTime EndUtc, TimeSpan Interval, long FirstBucket, long BucketCount);
|
||||
|
||||
/// <summary>The forward requests serving a plan: the whole intervals first, then the clipped one (if any).</summary>
|
||||
/// <param name="plan">A plan with at least one bucket.</param>
|
||||
/// <returns>One or two forward requests.</returns>
|
||||
public static IReadOnlyList<Request> Requests(BucketPlan plan)
|
||||
{
|
||||
var interval = TimeSpan.FromTicks(plan.IntervalTicks);
|
||||
var n = plan.BucketCount;
|
||||
var clipped = plan.IsTruncated(0) ? 0L : plan.IsTruncated(n - 1) ? n - 1 : -1L;
|
||||
if (clipped < 0) return [new Request(plan.StartUtc, plan.EndUtc, interval, 0, n)];
|
||||
|
||||
var clipLo = new DateTime(plan.LowerTicks(clipped), DateTimeKind.Utc);
|
||||
var clipHi = new DateTime(plan.UpperTicks(clipped), DateTimeKind.Utc);
|
||||
var requests = new List<Request>(2);
|
||||
if (n > 1)
|
||||
requests.Add(clipped == 0
|
||||
? new Request(clipHi, plan.EndUtc, interval, 1, n - 1) // reverse: the clipped one is the earliest
|
||||
: new Request(plan.StartUtc, clipLo, interval, 0, n - 1)); // forward: the clipped one is the latest
|
||||
requests.Add(new Request(clipLo, clipHi, clipHi - clipLo, clipped, 1));
|
||||
return requests;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps a request's rows onto the plan's buckets: rows outside the request window (AVEVA's pre-window row, #685)
|
||||
/// are dropped; each in-window row is placed by its interval start (or, without one, its end).
|
||||
/// </summary>
|
||||
/// <param name="plan">The plan.</param>
|
||||
/// <param name="request">The forward request the rows answer.</param>
|
||||
/// <param name="rows">The drained rows.</param>
|
||||
/// <param name="byBucket">Results by ascending bucket index (first row per bucket wins).</param>
|
||||
public static void Place(BucketPlan plan, Request request, IEnumerable<HistorianAggregateSample> rows,
|
||||
Dictionary<long, DataValueSnapshot> byBucket)
|
||||
{
|
||||
var interval = request.Interval.Ticks;
|
||||
foreach (var row in rows)
|
||||
{
|
||||
if (!SampleMapper.IsInWindow(row, request.StartUtc, request.EndUtc)) continue;
|
||||
long offset = row.StartTime is { } s
|
||||
? (s.ToDateTime().Ticks - request.StartUtc.Ticks) / interval
|
||||
: (row.EndTime.ToDateTime().Ticks - request.StartUtc.Ticks - 1) / interval; // IsInWindow ⇒ end in (lo, hi]
|
||||
if (offset < 0 || offset >= request.BucketCount) continue;
|
||||
var index = request.FirstBucket + offset;
|
||||
if (byBucket.ContainsKey(index)) continue;
|
||||
|
||||
var snapshot = SampleMapper.ToAggregateSnapshot(row);
|
||||
var status = snapshot.StatusCode;
|
||||
if (plan.IsTruncated(index) && !StatusBits.IsBad(status)) status |= HistorianInfoBits.HistorianPartial;
|
||||
byBucket[index] = snapshot with { StatusCode = status, SourceTimestampUtc = plan.BucketTimestamp(index) };
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The placed results in the read's order (ascending forward, descending reverse).</summary>
|
||||
/// <param name="plan">The plan.</param>
|
||||
/// <param name="byBucket">Results by ascending bucket index.</param>
|
||||
/// <returns>The results.</returns>
|
||||
public static IReadOnlyList<DataValueSnapshot> Assemble(BucketPlan plan, IReadOnlyDictionary<long, DataValueSnapshot> byBucket) =>
|
||||
(plan.Reversed ? byBucket.OrderByDescending(kv => kv.Key) : byBucket.OrderBy(kv => kv.Key)).Select(kv => kv.Value).ToList();
|
||||
}
|
||||
@@ -1,103 +0,0 @@
|
||||
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>
|
||||
/// #691.2: serves a REVERSE (<c>start > end</c>) processed read of the AVEVA-native <c>TimeAverage</c> / <c>Total</c>
|
||||
/// with the #677 semantics the SQL backends give it — window <c>(end, start]</c>, intervals stepping back from the
|
||||
/// request's later time, the clipped interval the EARLIEST, each result stamped with its later edge, newest first
|
||||
/// (<see cref="BucketPlan"/>, reversed) — although the gateway refuses <c>start > end</c>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Forward requests on the later-anchored grid.</b> AVEVA anchors a native read's interval grid at the request
|
||||
/// start (live, 2026-10-12: rows at <c>start + j·interval</c>, plus one extra row for the interval before it, #685).
|
||||
/// So the whole intervals of the reverse plan — <c>(O + k·I, O + (k+1)·I]</c> with <c>O = later − n·I</c> — are
|
||||
/// requested as ONE forward read <c>[O + I·f, later)</c> with interval <c>I</c> (<c>f</c> = 1 when the earliest
|
||||
/// interval is clipped, else 0), and the clipped earliest interval <c>(earlier, O + I]</c> as a second forward read
|
||||
/// over exactly that span with its own width as the interval — so AVEVA computes it over the clipped span only, never
|
||||
/// over time before the window. Neither request is ever reversed.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Soundness.</b> A time-weighted average / integral over <c>[a, b)</c> and over <c>(a, b]</c> integrate the same
|
||||
/// function over sets differing by two instants (measure zero), so a forward native row IS the value of the
|
||||
/// reverse interval with the same edges. The one place the half-open side could matter is AVEVA's status for an
|
||||
/// interval whose only non-good sample lies exactly on an edge — accepted, as on every native row. Each row is
|
||||
/// mapped exactly as a forward native row (<see cref="SampleMapper.ToAggregateSnapshot"/>: Uncertain keeps its value,
|
||||
/// no data ⇒ <c>BadNoData</c>, Calculated bits), restamped with its interval's LATER edge; the clipped interval
|
||||
/// carries the Partial bit unless Bad (Part 13 §5.3.3.2 Case 3, as the SQL backends' time-weighted aggregates).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Missing rows are not invented.</b> As on a forward native read, an interval AVEVA answers no row for is absent
|
||||
/// from the result (not a fabricated <c>BadNoData</c>). Today historiangw#27 (the interval reaches AVEVA in ticks
|
||||
/// where it reads milliseconds) makes AVEVA answer ONE row per request — the whole-intervals read lands on its
|
||||
/// earliest interval and the rest are absent — exactly as wrong as a forward native read; this mapping needs no
|
||||
/// change once the gateway is fixed.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class NativeReverseAggregatePlan
|
||||
{
|
||||
/// <summary>One forward native request and the ascending bucket index of its first interval.</summary>
|
||||
/// <param name="StartUtc">Forward request start (inclusive).</param>
|
||||
/// <param name="EndUtc">Forward request end.</param>
|
||||
/// <param name="Interval">The request's interval.</param>
|
||||
/// <param name="FirstBucket">Ascending plan bucket of the request's first interval.</param>
|
||||
/// <param name="BucketCount">Number of plan buckets the request covers.</param>
|
||||
internal readonly record struct Request(DateTime StartUtc, DateTime EndUtc, TimeSpan Interval, long FirstBucket, long BucketCount);
|
||||
|
||||
/// <summary>The forward requests serving a reverse plan: the whole intervals first, then the clipped earliest one.</summary>
|
||||
/// <param name="plan">A reverse plan with at least one bucket.</param>
|
||||
/// <returns>One or two forward requests.</returns>
|
||||
public static IReadOnlyList<Request> Requests(BucketPlan plan)
|
||||
{
|
||||
var interval = TimeSpan.FromTicks(plan.IntervalTicks);
|
||||
if (!plan.IsTruncated(0))
|
||||
return [new Request(plan.StartUtc, plan.EndUtc, interval, 0, plan.BucketCount)];
|
||||
|
||||
var firstWholeStart = new DateTime(plan.UpperTicks(0), DateTimeKind.Utc); // O + I
|
||||
var requests = new List<Request>(2);
|
||||
if (plan.BucketCount > 1)
|
||||
requests.Add(new Request(firstWholeStart, plan.EndUtc, interval, 1, plan.BucketCount - 1));
|
||||
requests.Add(new Request(plan.StartUtc, firstWholeStart, firstWholeStart - plan.StartUtc, 0, 1));
|
||||
return requests;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps a request's rows onto the plan's buckets: rows outside the request window (AVEVA's pre-window row, #685)
|
||||
/// are dropped; each in-window row is placed by its interval start (or, without one, its end).
|
||||
/// </summary>
|
||||
/// <param name="plan">The reverse plan.</param>
|
||||
/// <param name="request">The forward request the rows answer.</param>
|
||||
/// <param name="rows">The drained rows.</param>
|
||||
/// <param name="byBucket">Results by ascending bucket index (first row per bucket wins).</param>
|
||||
public static void Place(BucketPlan plan, Request request, IEnumerable<HistorianAggregateSample> rows,
|
||||
Dictionary<long, DataValueSnapshot> byBucket)
|
||||
{
|
||||
var interval = request.Interval.Ticks;
|
||||
foreach (var row in rows)
|
||||
{
|
||||
if (!SampleMapper.IsInWindow(row, request.StartUtc, request.EndUtc)) continue;
|
||||
long offset = row.StartTime is { } s
|
||||
? (s.ToDateTime().Ticks - request.StartUtc.Ticks) / interval
|
||||
: (row.EndTime.ToDateTime().Ticks - request.StartUtc.Ticks - 1) / interval; // IsInWindow ⇒ end in (lo, hi]
|
||||
if (offset < 0 || offset >= request.BucketCount) continue;
|
||||
var index = request.FirstBucket + offset;
|
||||
if (byBucket.ContainsKey(index)) continue;
|
||||
|
||||
var snapshot = SampleMapper.ToAggregateSnapshot(row);
|
||||
var status = snapshot.StatusCode;
|
||||
if (plan.IsTruncated(index) && !StatusBits.IsBad(status)) status |= HistorianInfoBits.HistorianPartial;
|
||||
byBucket[index] = snapshot with { StatusCode = status, SourceTimestampUtc = plan.BucketTimestamp(index) };
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The placed results newest first (the reverse read's order).</summary>
|
||||
/// <param name="plan">The reverse plan.</param>
|
||||
/// <param name="byBucket">Results by ascending bucket index.</param>
|
||||
/// <returns>The results, descending.</returns>
|
||||
public static IReadOnlyList<DataValueSnapshot> Assemble(BucketPlan plan, IReadOnlyDictionary<long, DataValueSnapshot> byBucket) =>
|
||||
byBucket.OrderByDescending(kv => kv.Key).Select(kv => kv.Value).ToList();
|
||||
}
|
||||
+3
-3
@@ -145,11 +145,11 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
ReadAggregateCallCount++;
|
||||
AggregateCalls.Add((startUtc, endUtc, mode, interval));
|
||||
if (AggregateSource is null) return ToAsyncStream(AggregateSamples, ThrowOnRead, ct);
|
||||
if (startUtc > endUtc)
|
||||
if (startUtc > endUtc || interval <= TimeSpan.Zero)
|
||||
{
|
||||
// The real gateway's ValidateRange refusal, raised before any row.
|
||||
// The real gateway's request validation (ValidateRange; interval must be > 0), raised before any row.
|
||||
var rpc = new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument,
|
||||
"start_time must be less than or equal to end_time."));
|
||||
startUtc > endUtc ? "start_time must be less than or equal to end_time." : "interval must be greater than zero."));
|
||||
return ToAsyncStream(Array.Empty<HistorianAggregateSample>(),
|
||||
new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException(rpc.Status.Detail, rpc), ct);
|
||||
}
|
||||
|
||||
+4
-2
@@ -29,12 +29,14 @@ public sealed class GatewayHistorianDataSourceTests
|
||||
Assert.Equal(100 + RawAggregateReader.PreWindowAllowance, fake.LastReadRawMaxValues);
|
||||
}
|
||||
|
||||
private static readonly DateTime Hour0 = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
[Fact]
|
||||
public async Task ReadProcessed_uses_aggregate_mode_mapping()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient();
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
await ds.ReadProcessedAsync("T", default, default, TimeSpan.FromSeconds(60), HistoryAggregateType.Total, TestContext.Current.CancellationToken);
|
||||
await ds.ReadProcessedAsync("T", Hour0, Hour0.AddHours(1), TimeSpan.FromSeconds(60), HistoryAggregateType.Total, TestContext.Current.CancellationToken); // an empty window sends nothing
|
||||
Assert.Equal(RetrievalMode.Integral, fake.LastAggregateMode);
|
||||
Assert.Equal(TimeSpan.FromSeconds(60), fake.LastAggregateInterval);
|
||||
}
|
||||
@@ -62,7 +64,7 @@ public sealed class GatewayHistorianDataSourceTests
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient();
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
await ds.ReadProcessedAsync("T", default, default, TimeSpan.FromSeconds(60), HistoryAggregateType.TimeAverage, TestContext.Current.CancellationToken);
|
||||
await ds.ReadProcessedAsync("T", Hour0, Hour0.AddHours(1), TimeSpan.FromSeconds(60), HistoryAggregateType.TimeAverage, TestContext.Current.CancellationToken); // an empty window sends nothing
|
||||
Assert.Equal(RetrievalMode.TimeWeightedAverage, fake.LastAggregateMode);
|
||||
}
|
||||
|
||||
|
||||
+53
-5
@@ -8,16 +8,17 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// #691.2: a reverse processed read (<c>start > end</c>) of the AVEVA-native <c>TimeAverage</c> / <c>Total</c>. The
|
||||
/// gateway refuses <c>start > end</c>, so the read is served by FORWARD native requests whose bucket grid is anchored
|
||||
/// at the request's later time (#677: window <c>(end, start]</c>, intervals stepping back from <c>start</c>, the clipped
|
||||
/// interval the earliest, each stamped with its later edge, results newest first). The fake is a native-mode AVEVA
|
||||
/// #691.2: a forward or reverse processed read of the AVEVA-native <c>TimeAverage</c> / <c>Total</c> on the SQL
|
||||
/// backends' interval grid. The gateway refuses <c>start > end</c>, so a reverse read is served by FORWARD native
|
||||
/// requests whose bucket grid is anchored at the request's later time (#677: window <c>(end, start]</c>, intervals
|
||||
/// stepping back from <c>start</c>, the clipped interval the earliest, each stamped with its later edge, results newest
|
||||
/// first); a forward read's clipped LATEST interval is read over exactly its span (review). The fake is a native-mode AVEVA
|
||||
/// model: a step-held stored series; for a forward request <c>[s, e)</c> with interval <c>I</c> it answers one row per
|
||||
/// interval of the grid anchored at <c>s</c> (plus AVEVA's extra row for the interval before <c>s</c>, #685), each the
|
||||
/// time-weighted average / integral of exactly that interval — the per-interval behaviour the gateway will have once
|
||||
/// historiangw#27 is fixed.
|
||||
/// </summary>
|
||||
public sealed class GatewayReverseNativeAggregateTests
|
||||
public sealed class GatewayNativeAggregateGridTests
|
||||
{
|
||||
private static readonly CancellationToken Ct = CancellationToken.None;
|
||||
private static readonly DateTime T0 = new(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc);
|
||||
@@ -156,6 +157,53 @@ public sealed class GatewayReverseNativeAggregateTests
|
||||
s[2].Value.ShouldBeNull();
|
||||
}
|
||||
|
||||
// ---- forward reads on the same grid (#691.2 review): the clipped LATEST interval is its own exact-span read ----------
|
||||
|
||||
[Theory]
|
||||
[InlineData(HistoryAggregateType.TimeAverage, RetrievalMode.TimeWeightedAverage)]
|
||||
[InlineData(HistoryAggregateType.Total, RetrievalMode.Integral)]
|
||||
public async Task A_forward_read_with_a_partial_interval_clips_the_LATEST_one_and_marks_it_Partial(HistoryAggregateType agg,
|
||||
RetrievalMode mode)
|
||||
{
|
||||
var fake = Aveva();
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(5), At(70), TimeSpan.FromSeconds(20), agg, Ct)).Samples;
|
||||
|
||||
// [5, 70): [5,25), [25,45), [45,65) and the clipped [65,70) — stamped with each EARLIER edge, oldest first, as on
|
||||
// SqlServer/QuestDB (BucketPlan forward; the truncated interval Partial unless Bad).
|
||||
s.Select(x => x.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(5), At(25), At(45), At(65) });
|
||||
s.Select(x => (double)x.Value!).ShouldBe(
|
||||
[Expected(mode, At(5), At(25))!.Value, Expected(mode, At(25), At(45))!.Value, Expected(mode, At(45), At(65))!.Value,
|
||||
Expected(mode, At(65), At(70))!.Value],
|
||||
tolerance: 1e-9);
|
||||
s.Select(x => x.StatusCode).ShouldBe([GoodCalc, GoodCalc, GoodCalc, GoodCalc | Partial]);
|
||||
// AVEVA never averages past EndTime: the whole intervals, then the clipped one over exactly its span.
|
||||
fake.AggregateCalls.ShouldBe([
|
||||
(At(5), At(65), mode, TimeSpan.FromSeconds(20)),
|
||||
(At(65), At(70), mode, TimeSpan.FromSeconds(5)),
|
||||
]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_forward_interval_wider_than_the_window_is_sent_as_the_window()
|
||||
{
|
||||
var fake = Aveva();
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(12), At(30), TimeSpan.FromSeconds(60),
|
||||
HistoryAggregateType.Total, Ct)).Samples;
|
||||
s.Single().SourceTimestampUtc.ShouldBe(At(12));
|
||||
s.Single().StatusCode.ShouldBe(GoodCalc);
|
||||
fake.AggregateCalls.ShouldBe([(At(12), At(30), RetrievalMode.Integral, TimeSpan.FromSeconds(18))]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_zero_interval_forward_read_is_one_interval_over_the_window_not_a_zero_interval_request()
|
||||
{
|
||||
var fake = Aveva();
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(0), At(30), TimeSpan.Zero,
|
||||
HistoryAggregateType.TimeAverage, Ct)).Samples;
|
||||
s.Single().SourceTimestampUtc.ShouldBe(At(0));
|
||||
fake.AggregateCalls.Single().Interval.ShouldBe(TimeSpan.FromSeconds(30));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_forward_read_is_unchanged()
|
||||
{
|
||||
Reference in New Issue
Block a user