From 830b47f624345026520b420e3e8523f2774a9c1d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 16:55:01 -0400 Subject: [PATCH] =?UTF-8?q?fix(historian):=20#691.2=20review=20=E2=80=94?= =?UTF-8?q?=20forward=20native=20TimeAverage/Total=20on=20the=20SQL=20grid?= =?UTF-8?q?:=20clipped=20latest=20interval=20read=20over=20its=20exact=20s?= =?UTF-8?q?pan,=20Partial,=20interval=20clamped=20to=20the=20window?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CLAUDE.md | 2 +- docs/Historian.md | 33 ++++-- .../GatewayHistorianDataSource.cs | 51 ++------ .../NativeAggregatePlan.cs | 112 ++++++++++++++++++ .../NativeReverseAggregatePlan.cs | 103 ---------------- .../FakeHistorianGatewayClient.cs | 6 +- .../GatewayHistorianDataSourceTests.cs | 6 +- ....cs => GatewayNativeAggregateGridTests.cs} | 58 ++++++++- 8 files changed, 205 insertions(+), 166 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeAggregatePlan.cs delete mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeReverseAggregatePlan.cs rename tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/{GatewayReverseNativeAggregateTests.cs => GatewayNativeAggregateGridTests.cs} (71%) diff --git a/CLAUDE.md b/CLAUDE.md index 5f302a95..f29c3758 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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. diff --git a/docs/Historian.md b/docs/Historian.md index 35d4c5c3..637b32f6 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index 4d182f2e..5685ed32 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -253,9 +253,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// /// TimeAverage / Total stay native AVEVA retrieval modes (), 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, ). A reverse read - /// (start > end, which the gateway refuses) is served by forward requests on the later-anchored #677 grid - /// (#691.2, ). Everything else is + /// 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, ). Everything else is /// (see ). /// /// @@ -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(); - 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); } /// - /// A reverse (start > end) native TimeAverage / Total read (#691.2), anchored at its later time as on the SQL - /// backends (#677) — the gateway refuses start > end, so it is served by forward native requests on the - /// later-anchored grid. See . + /// 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 . /// - private async Task ReadNativeReverseAsync(string fullReference, BucketPlan plan, RetrievalMode mode, + private async Task ReadNativeAsync(string fullReference, BucketPlan plan, RetrievalMode mode, CancellationToken cancellationToken) { if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null); try { var byBucket = new Dictionary(); - foreach (var request in NativeReverseAggregatePlan.Requests(plan)) + foreach (var request in NativeAggregatePlan.Requests(plan)) { var buckets = new List(); 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) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeAggregatePlan.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeAggregatePlan.cs new file mode 100644 index 00000000..35dc1573 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeAggregatePlan.cs @@ -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; + +/// +/// Serves a processed read of the AVEVA-native TimeAverage / Total on the SQL backends' interval grid +/// (), in either direction: forward — window [start, end), intervals stepping forward +/// from start, the clipped interval the LATEST, each stamped with its earlier edge, oldest first; reverse (#677, +/// #691.2) — window (end, start], intervals stepping back from start, the clipped interval the EARLIEST, +/// each stamped with its later edge, newest first. The gateway refuses start > end and interval <= 0, +/// so every request it sees is forward with a positive interval. +/// +/// +/// +/// Forward requests on the plan's grid. AVEVA anchors a native read's interval grid at the request start (live, +/// 2026-10-12: rows at start + j·interval, 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 I, 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 EndTime on a forward one). An interval wider than the window is sent as the window, and +/// interval <= 0 as one interval over the whole window (it used to reach the gateway as 0 and be refused). +/// +/// +/// Soundness of reverse. A time-weighted average / integral over [a, b) and over (a, b] 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. +/// +/// +/// Mapping. Each row is mapped as before (: Uncertain keeps its +/// value, no data ⇒ BadNoData, Calculated bits), placed by its interval start (else end), restamped with +/// , 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 BadNoData, unlike the SQL backends). +/// +/// +/// ⚠️ 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. +/// +/// +internal static class NativeAggregatePlan +{ + /// One forward native request and the ascending bucket index of its first interval. + /// Forward request start (inclusive). + /// Forward request end. + /// The request's interval (> 0). + /// Ascending plan bucket of the request's first interval. + /// Number of plan buckets the request covers. + internal readonly record struct Request(DateTime StartUtc, DateTime EndUtc, TimeSpan Interval, long FirstBucket, long BucketCount); + + /// The forward requests serving a plan: the whole intervals first, then the clipped one (if any). + /// A plan with at least one bucket. + /// One or two forward requests. + public static IReadOnlyList 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(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; + } + + /// + /// 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). + /// + /// The plan. + /// The forward request the rows answer. + /// The drained rows. + /// Results by ascending bucket index (first row per bucket wins). + public static void Place(BucketPlan plan, Request request, IEnumerable rows, + Dictionary 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) }; + } + } + + /// The placed results in the read's order (ascending forward, descending reverse). + /// The plan. + /// Results by ascending bucket index. + /// The results. + public static IReadOnlyList Assemble(BucketPlan plan, IReadOnlyDictionary byBucket) => + (plan.Reversed ? byBucket.OrderByDescending(kv => kv.Key) : byBucket.OrderBy(kv => kv.Key)).Select(kv => kv.Value).ToList(); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeReverseAggregatePlan.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeReverseAggregatePlan.cs deleted file mode 100644 index ab1d29e6..00000000 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/NativeReverseAggregatePlan.cs +++ /dev/null @@ -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; - -/// -/// #691.2: serves a REVERSE (start > end) processed read of the AVEVA-native TimeAverage / Total -/// with the #677 semantics the SQL backends give it — window (end, start], intervals stepping back from the -/// request's later time, the clipped interval the EARLIEST, each result stamped with its later edge, newest first -/// (, reversed) — although the gateway refuses start > end. -/// -/// -/// -/// Forward requests on the later-anchored grid. AVEVA anchors a native read's interval grid at the request -/// start (live, 2026-10-12: rows at start + j·interval, plus one extra row for the interval before it, #685). -/// So the whole intervals of the reverse plan — (O + k·I, O + (k+1)·I] with O = later − n·I — are -/// requested as ONE forward read [O + I·f, later) with interval I (f = 1 when the earliest -/// interval is clipped, else 0), and the clipped earliest interval (earlier, O + I] 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. -/// -/// -/// Soundness. A time-weighted average / integral over [a, b) and over (a, b] 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 (: Uncertain keeps its value, -/// no data ⇒ BadNoData, 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). -/// -/// -/// Missing rows are not invented. As on a forward native read, an interval AVEVA answers no row for is absent -/// from the result (not a fabricated BadNoData). 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. -/// -/// -internal static class NativeReverseAggregatePlan -{ - /// One forward native request and the ascending bucket index of its first interval. - /// Forward request start (inclusive). - /// Forward request end. - /// The request's interval. - /// Ascending plan bucket of the request's first interval. - /// Number of plan buckets the request covers. - internal readonly record struct Request(DateTime StartUtc, DateTime EndUtc, TimeSpan Interval, long FirstBucket, long BucketCount); - - /// The forward requests serving a reverse plan: the whole intervals first, then the clipped earliest one. - /// A reverse plan with at least one bucket. - /// One or two forward requests. - public static IReadOnlyList 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(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; - } - - /// - /// 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). - /// - /// The reverse plan. - /// The forward request the rows answer. - /// The drained rows. - /// Results by ascending bucket index (first row per bucket wins). - public static void Place(BucketPlan plan, Request request, IEnumerable rows, - Dictionary 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) }; - } - } - - /// The placed results newest first (the reverse read's order). - /// The reverse plan. - /// Results by ascending bucket index. - /// The results, descending. - public static IReadOnlyList Assemble(BucketPlan plan, IReadOnlyDictionary byBucket) => - byBucket.OrderByDescending(kv => kv.Key).Select(kv => kv.Value).ToList(); -} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index a9888666..6cb69cc8 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs @@ -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(), new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException(rpc.Status.Detail, rpc), ct); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs index 8ed951c2..0920a0d4 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs @@ -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.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.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); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayReverseNativeAggregateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateGridTests.cs similarity index 71% rename from tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayReverseNativeAggregateTests.cs rename to tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateGridTests.cs index 2b2d28d9..de71843d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayReverseNativeAggregateTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateGridTests.cs @@ -8,16 +8,17 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; /// -/// #691.2: a reverse processed read (start > end) of the AVEVA-native TimeAverage / Total. The -/// gateway refuses start > end, so the read is served by FORWARD native requests whose bucket grid is anchored -/// at the request's later time (#677: window (end, start], intervals stepping back from start, 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 TimeAverage / Total on the SQL +/// backends' interval grid. The gateway refuses start > end, so a reverse read is served by FORWARD native +/// requests whose bucket grid is anchored at the request's later time (#677: window (end, start], intervals +/// stepping back from start, 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 [s, e) with interval I it answers one row per /// interval of the grid anchored at s (plus AVEVA's extra row for the interval before s, #685), each the /// time-weighted average / integral of exactly that interval — the per-interval behaviour the gateway will have once /// historiangw#27 is fixed. /// -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() {