Merge feat/681-gateway-per-sample: Gateway serves the per-sample aggregates from raw reads via the shared evaluator (#681)
This commit is contained in:
Joseph Doherty
@@ -1364,7 +1364,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, Processed
|
||||
(backend-dependent: Gateway serves TimeAverage/Minimum/Maximum/Total, with Average/Count `BadAggregateNotSupported` since AVEVA has no mean or sample count, #679; SqlServer/QuestDb serve seventeen), 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 (now interval-start stamped — pending the live leg — + Calculated); 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.
|
||||
|
||||
@@ -1382,7 +1382,7 @@ authorization uses the standard `AccessLevels.HistoryRead` bit set at materializ
|
||||
| `MaxTieClusterOverfetch` | `65536` | Bounded over-fetch the HistoryRead-Raw paging uses to page within an oversized same-timestamp tie cluster (retained from the prior backend) |
|
||||
| `HistoryReadBatchParallelism` | `4` | Max nodes served concurrently within one HistoryRead batch (03/S3); `⌈N/P⌉ × CallTimeout` worst case instead of sequential |
|
||||
| `MaxConcurrentHistoryReadBatches` | `16` | Process-wide concurrent HistoryRead batch cap (03/S3); flood degrades to `BadTooManyOperations`. Total in-flight reads capped at 16 × 4 = 64 |
|
||||
| `MaxTimeWeightedRawRows` | `1000000` | `SqlServer`/`QuestDb` only (#671): raw in-window rows one time-weighted aggregate read (incl. `Total` and, since #680, PercentGood/PercentBad) may stream into the C# evaluator; window query capped at this + 1, past it ⇒ `BadTooManyOperations` for that node. ≤ 0 = unbounded |
|
||||
| `MaxTimeWeightedRawRows` | `1000000` | `SqlServer`/`QuestDb` (#671): raw in-window rows one time-weighted aggregate read (incl. `Total` and, since #680, PercentGood/PercentBad) may stream into the C# evaluator — and, since #681, every `Gateway` per-sample read (raw-evaluated); window query capped at this + 1, past it ⇒ `BadTooManyOperations` for that node. ≤ 0 = unbounded |
|
||||
| `HistoryReadDeadline` | `00:01:00` | Per-request HistoryRead wall-clock deadline (03/S3); a node still in flight surfaces `BadTimeout`. Non-positive = unbounded |
|
||||
|
||||
### Alarm-history path (`AlarmHistorian` section)
|
||||
|
||||
+2
-2
@@ -210,8 +210,8 @@ otopcua-cli historyread -u opc.tcp://localhost:4840/OtOpcUa \
|
||||
-n "ns=1;s=TestMachine_001.TestHistoryValue" \
|
||||
--start "2026-03-25T00:00:00Z" --end "2026-03-30T00:00:00Z"
|
||||
|
||||
# Aggregate: 1-hour average (SqlServer/QuestDb; on the Gateway historian backend Average is
|
||||
# BadAggregateNotSupported — use --aggregate TimeAverage there, #679)
|
||||
# Aggregate: 1-hour average (every backend; on Gateway it is computed from raw samples since #681 —
|
||||
# capped by ServerHistorian:MaxTimeWeightedRawRows, so read long dense windows in slices)
|
||||
otopcua-cli historyread -u opc.tcp://localhost:4840/OtOpcUa \
|
||||
-n "ns=1;s=TestMachine_001.TestHistoryValue" \
|
||||
--start "2026-03-25T00:00:00Z" --end "2026-03-30T00:00:00Z" \
|
||||
|
||||
+82
-14
@@ -77,7 +77,7 @@ consumes the historian. The existing enable gates (`ServerHistorian:Enabled`, `A
|
||||
| Value kinds recorded | `Double`, `Int64`, `Boolean` (numeric analog only) | all: `Double`, `Int64`, `Boolean`, `String`, `DateTime`, plus `Empty` (a Bad sample with no value) | same as `SqlServer` |
|
||||
| Status code stored | Sent as DA quality but **discarded** by the gateway (C-002); AVEVA stores every sample Good | Stored bit-for-bit (`StatusCode` int column) | Stored bit-for-bit (unsigned, in a `LONG`) |
|
||||
| Comms-loss / driver-Bad publishes | Dropped, counted in `DroppedUnsupportedKind` | Stored as an `Empty` outage-marker sample carrying the Bad code (see [below](#typed-values-real-status-and-outage-markers)) | Stored as an `Empty` outage-marker sample, as on `SqlServer` |
|
||||
| Processed aggregates | TimeAverage, Minimum, Maximum, Total (Average and Count: `BadAggregateNotSupported`, #679) | The seventeen in the [aggregate table](#aggregates-processed-reads) (incl. Part 13 time-weighted, #671) | The same seventeen, identical results |
|
||||
| Processed aggregates | Twelve: the ten per-sample ones (Average, Count, Minimum, Maximum, Range, Start, End, Delta, Min/MaxActualTime) evaluated from raw reads with the SQL backends' evaluator — identical results (#681) — plus TimeAverage and Total from AVEVA's native modes | The seventeen in the [aggregate table](#aggregates-processed-reads) (incl. Part 13 time-weighted, #671) | The same seventeen, identical results |
|
||||
| HistoryUpdate | No (`BadHistoryOperationUnsupported`) | Yes (Insert / Replace / Update / DeleteRaw / DeleteAtTime) | Yes — soft delete (tombstones), see [below](#historyupdate-soft-delete) |
|
||||
| Modified-value history (`IsReadModified`, #668) | No (`BadHistoryOperationUnsupported`) | Yes — `SampleModified`, atomic with the change | Yes — `{prefix}sample_modified`, written before the change (orphan-on-crash) |
|
||||
| Alarm history | gateway `SendEvent` | `Event` table (insert-if-absent) | `{prefix}event` table (dedup upsert on `(ts, event_id)`) |
|
||||
@@ -146,7 +146,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions
|
||||
| `MaxTieClusterOverfetch` | int | `65536` | (Node-manager paging; all backends — a `SqlServer` PK rules out tie clusters, so it is moot there.) Maximum samples the server will fetch in one shot to page through a tie cluster (multiple samples sharing one `SourceTimestamp`). A cluster larger than this ceiling fails `BadHistoryOperationUnsupported`. Raise to handle abnormally large tie clusters; the default covers all normal-data cases. |
|
||||
| `HistoryReadBatchParallelism` | int | `4` | Max historized nodes served **concurrently** within one HistoryRead batch (arch-review 03/S3). The SDK's `HistoryRead*` override surface is synchronous, so each arm bridges once at its boundary; this turns per-node serving from sequential (`N × CallTimeout` worst case) into `⌈N / P⌉ × CallTimeout`. Also caps the gateway-load multiplication a single client can command. `≤ 1` serves sequentially. |
|
||||
| `MaxConcurrentHistoryReadBatches` | int | `16` | Process-wide limit on HistoryRead (and HistoryUpdate) batches served **concurrently** (arch-review 03/S3). A flood of history clients degrades to `BadTooManyOperations` rather than exhausting the SDK request-thread pool. With `HistoryReadBatchParallelism` this caps total in-flight gateway reads at `MaxConcurrentHistoryReadBatches × HistoryReadBatchParallelism` (default 16 × 4 = 64). A batch that cannot get a slot within `min(HistoryReadDeadline, 5 s)` fails every handle `BadTooManyOperations`. |
|
||||
| `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` only (#671). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped. 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). |
|
||||
| `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` (#671), and the `Gateway` per-sample aggregates (#681 — every per-sample read there streams raw samples, so the cap applies to the window's raw rows; see [Aggregate set](#aggregate-set-and-badaggregatenotsupported-285)). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped on the SQL backends (pushed down). 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). |
|
||||
| `HistoryReadDeadline` | TimeSpan | `00:01:00` | Per-request wall-clock deadline for a HistoryRead (and HistoryUpdate) batch (arch-review 03/S3). A node still in flight at the deadline surfaces `BadTimeout`, so one request can never hold an SDK request thread longer than this regardless of node count. Non-positive = unbounded (a `Validate()` warning). |
|
||||
| `HistoryUpdateMaxFutureSkew` | TimeSpan | `1.00:00:00` | #674. How far past the server clock a HistoryUpdate (Insert / Replace / Update) item's `SourceTimestamp` may lie; an item beyond `now + skew` is refused **`BadOutOfRange`** by the node manager before any backend sees it (every backend). Guards against a far-future typo (a 2062 timestamp). The live recorder is not bounded. Non-positive disables the bound (no warning — a deliberate choice). See [HistoryUpdate](#historyupdate). |
|
||||
|
||||
@@ -747,31 +747,99 @@ otopcua-cli historyread -u opc.tcp://localhost:4840 -U apikey -P otua_<keyId>_<s
|
||||
The OPC UA `Total` aggregate is **supported** over the HistorianGateway backend. The gateway exposes a
|
||||
native **`Integral`** retrieval mode, so `Total` maps straight to it (`HistoryAggregateType.Total →
|
||||
RetrievalMode.Integral`) — no client-side scaling. (This replaces the retired Wonderware path, which had no
|
||||
`Total` column and derived it client-side as time-weighted `Average × interval-seconds`.) Bucket status codes and timestamps are preserved unchanged; empty / null
|
||||
buckets surface as `BadNoData`.
|
||||
`Total` column and derived it client-side as time-weighted `Average × interval-seconds`.) Values pass through unchanged; a Good
|
||||
bucket is `Good | Calculated` (`0x00000401`) and an empty / null bucket `BadNoData | Calculated` (#681), stamped with its interval
|
||||
start (see below).
|
||||
|
||||
**Integral unit (#677).** AVEVA's integral is `∫ value dt` (seconds) ÷ the tag's `IntegralDivisor`. The proto3 `double` default for
|
||||
`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) is still open — it needs the
|
||||
VPN. Reverse processed reads on the `Gateway` backend are answered by AVEVA and are not re-anchored by #677.
|
||||
VPN. Reverse processed reads of the AVEVA-native `TimeAverage`/`Total` are not re-anchored by #677 (the gateway refuses `end < start`);
|
||||
the raw-evaluated per-sample aggregates (#681) do follow #677. 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 interval-start stamp is pending the live leg:** it reads the aggregate row's `start_time`, which histsdk decodes from a row
|
||||
field (`tail + 43`) its own reverse-engineering notes still mark unverified (HistorianGateway
|
||||
`docs/plans/2026-07-13-archreview2-D-histsdk-residue-design.md`); if that field proves wrong on live AVEVA the stamp is wrong with it.
|
||||
|
||||
### Aggregate set and `BadAggregateNotSupported` (#285)
|
||||
|
||||
> ⚠️ **#679 (client-visible): on the `Gateway` backend, `Average` and `Count` are now `BadAggregateNotSupported`.** AVEVA's 15
|
||||
> retrieval modes contain no arithmetic mean and no sample count. `Average` used to be mapped to `TimeWeightedAverage` (a different
|
||||
> statistic from the OPC UA mean the SQL backends return); `Count` to `Counter`, which is a rollover-counter increase, not a sample
|
||||
> count. Request **`TimeAverage`** for the time-weighted number (now mapped to `TimeWeightedAverage`). There is no Gateway sample
|
||||
> count. `TimeAverage` magnitude and timestamping against live AVEVA are unverified (needs the VPN).
|
||||
> ⚠️ **#681 (client-visible): on the `Gateway` backend the per-sample aggregates are computed from RAW reads.** `Average`, `Count`
|
||||
> (both `BadAggregateNotSupported` since #679 — AVEVA has no arithmetic mean and no sample count), `Minimum` and `Maximum` (were
|
||||
> AVEVA's `MinimumWithTime` / `MaximumWithTime`), and the newly served `Range`, `Start`, `End`, `Delta`, `MinimumActualTime`,
|
||||
> `MaximumActualTime` stream the window's raw samples through the gateway and are evaluated by the SQL backends' shared
|
||||
> `Sql.Common` evaluator (`BucketPlan` + `AggregateEvaluator`), so Gateway answers exactly what `SqlServer`/`QuestDb` answer for the
|
||||
> same samples — values, status (Calculated / Uncertain / Partial bits, #680 Count/Start/End/Delta semantics), interval-START
|
||||
> timestamps, and reverse reads anchored at their later time (#677). What a client sees change: `Minimum`/`Maximum` values and
|
||||
> statuses may differ from AVEVA's own modes; every per-sample result now carries the Calculated info bits; a per-sample read is
|
||||
> bounded by **`ServerHistorian:MaxTimeWeightedRawRows`** (in-window raw rows; past it ⇒ `BadTooManyOperations` for that node — read
|
||||
> a long window over a dense tag in shorter windows). `TimeAverage` and `Total` stay AVEVA-native (`TimeWeightedAverage` /
|
||||
> `Integral`), now stamped with the interval start (was its end) and carrying the Calculated bit; `TimeAverage2`, `Total2`,
|
||||
> `Interpolative`, `PercentGood` and `PercentBad` remain `BadAggregateNotSupported` on Gateway. The served set is the single
|
||||
> static `GatewayHistorianDataSource.ServedAggregates`.
|
||||
|
||||
**How the Gateway raw evaluation works (#681).** Forward `ReadRaw` reads only (the gateway streams oldest-first and refuses
|
||||
`end < start`; `BucketPlan` assembles a reverse result newest-first), O(buckets) memory, and **every gateway stream is consumed to
|
||||
completion** — breaking out of one cancels the call, and the gateway currently classifies that cancellation as a connection fault
|
||||
and evicts its shared historian session (see the HistorianGateway issue text in the #681 record).
|
||||
|
||||
- **Window read** `[start − 1 ms, end]`. An AVEVA `Full` raw read emits an *initial-value* row stamped exactly at the query start
|
||||
(histsdk evidence: `Quality = 133`, `OpcQuality = 192`) — the value in effect there, not a stored sample. Starting 1 ms early puts
|
||||
that row outside every bucket. A pre-window row only proves "the tag has earlier data" (start-of-data Partial, Count's in-range
|
||||
zero), and only when its **OPC quality is not Bad** — a "no earlier data" NULL row still carries a value on the wire (histsdk sets
|
||||
`NumericValue = 0` on every numeric row and `StringValue = ""` on a string one) but with a Bad OPC quality (captured as 1, no-data; histsdk string-read-layout.md).
|
||||
- **Earlier-data probe.** When the data range matters (every per-sample aggregate but `Average`) and no non-Bad pre-window row
|
||||
arrived, a 2-row read `[1970-01-01, start)` asks whether the tag has data before the window. Normally this fires only for a tag
|
||||
whose history really begins inside the window.
|
||||
- **Look-ahead.** The end-of-data rules need to know whether data exists after the window: a sample exactly at the window end
|
||||
(forward) comes from the window read; otherwise a 2-row read `[end, now + 1 day]`. In both small reads a row stamped exactly at
|
||||
the query start (a boundary row) or the query end (a possible terminal row) is not a sample, whatever its value.
|
||||
- **Bound.** In-window rows are capped by `MaxTimeWeightedRawRows`; the gateway is asked for `cap + 18` rows. Past the cap the rest
|
||||
of the (bounded) stream is drained, then the node fails `BadTooManyOperations`; a stream that delivers the whole budget cannot be
|
||||
proven complete and fails the same way. The gateway's own refusal of that budget (`InvalidArgument` "max_values exceeds the
|
||||
server ceiling" — `Historian:ReadLimits:MaxValuesCeiling`, default 1,000,000, with `AllowStreamingBeyondCeiling=false`; the
|
||||
default `1,000,000 + 18` exceeds it) is also `BadTooManyOperations`, not a backend fault — lower `MaxTimeWeightedRawRows` to
|
||||
`ceiling − 18` on such a gateway. A caller cancellation (client cancel, `HistoryReadDeadline`) is not counted as a backend fault.
|
||||
- **Bucket statistics** are `RawBucketAccumulator`, the C# statement of the SQL backends' per-sample columns — incl. #684:
|
||||
Average / Minimum / Maximum / Range / *ActualTime over **Good** samples only, the Uncertain extremes feeding only the Min/Max
|
||||
status rule, and an Uncertain String/DateTime sample not refusing the Good-value aggregates. A change to those SQL columns
|
||||
must be repeated there.
|
||||
- **Status** is the raw read's quality mapping. C-002: the gateway discards quality on write, so AVEVA history is normally all Good.
|
||||
⚠️ **Known pre-existing limitation:** `SampleMapper` falls back to AVEVA's own `Quality` field when `OpcQuality` is 0 — but 0 is
|
||||
also the OPC DA Bad value, so a Bad row whose AVEVA `Quality` is ≥ 64 (e.g. 133) maps Uncertain/Good. Raw reads have always done
|
||||
this; since #681 such a row also feeds the per-sample aggregates (as a usable sample). The earlier-data evidence rule reads
|
||||
`OpcQuality` alone and is not affected.
|
||||
|
||||
**Cost (client-visible).** A per-sample read now moves every raw sample of the window over the gateway instead of one row per
|
||||
interval — notably `Minimum`/`Maximum`, which were native AVEVA modes before #681. Latency and gateway session time grow with the
|
||||
window's sample count, and the cap makes a long window over a dense tag fail: a month of a 1 Hz tag is ~2.6 M samples ⇒
|
||||
`BadTooManyOperations` at the default 1,000,000 cap (≈ 11.5 days of 1 Hz). Read such windows in shorter slices, or use
|
||||
`TimeAverage` (still native).
|
||||
|
||||
**Why the time-weighted family is not raw-evaluated.** Part 13 time-weighted aggregates need the real *bounding* samples — the time,
|
||||
value and quality of the latest sample before the window (sloped interpolation across the window start, an outage straddling it). The
|
||||
gateway has no "previous stored sample" read: a raw read streams forward from its start, and AVEVA replaces whatever lies before
|
||||
with the initial-value row stamped at the query start, losing the original time. So `TimeAverage`/`Total` keep AVEVA's native modes
|
||||
(exact to AVEVA's own definition, not provably Part 13), and the other five time-weighted members stay unsupported rather than
|
||||
approximated.
|
||||
|
||||
**Live leg (open — needs the VPN to `wonder-sql-vd03`):** (1) that AVEVA's raw read emits the initial-value row with a non-Bad OPC
|
||||
quality when earlier data exists and a Bad (NULL) one when none does; (2) that the window read emits no terminal row at its query
|
||||
end (the window read relies on there being none; the small reads ignore one); (3) that a 2-row raw read starting at `1970-01-01`
|
||||
and one ending at `now + 1 day` are accepted and cheap; (4) the gateway's ceiling rejection text still matches
|
||||
"max_values exceeds the server ceiling"; (5) `Minimum`/`Maximum` against live data vs. the old AVEVA modes; (6)
|
||||
`TimeAverage`/`Total` magnitude and the interval-start stamp (`tail + 43`, above).
|
||||
|
||||
`HistoryAggregateType` carries the wider Part 13 set (Interpolative, TimeAverage, Start, End, Delta, Range,
|
||||
Min/MaxActualTime, Start/EndBound, DeltaBounds, DurationGood/Bad, PercentGood/Bad, WorstQuality,
|
||||
NumberOfTransitions, AnnotationCount, StandardDeviationPopulation, VariancePopulation); `AggregateFunctionMap`
|
||||
(`OpcUaServer`) maps each standard aggregate NodeId (`TimeAverage2` and `Total2` were appended for #671). The **`SqlServer`** and
|
||||
**`QuestDb`** backends serve the seventeen listed in the [aggregate table](#aggregates-processed-reads); everything else
|
||||
(e.g. `StartBound`, `DurationGood`, `StandardDeviationPopulation`) is `BadAggregateNotSupported` for that node. The **HistorianGateway** backend only has retrieval modes for the four
|
||||
(TimeAverage, Minimum, Maximum, Total). A NodeId outside the mapped set (e.g. `StandardDeviationSample`) is still
|
||||
`BadAggregateNotSupported`.
|
||||
(e.g. `StartBound`, `DurationGood`, `StandardDeviationPopulation`) is `BadAggregateNotSupported` for that node. The **HistorianGateway** backend serves twelve (#681 —
|
||||
`GatewayHistorianDataSource.ServedAggregates`): the ten per-sample aggregates from raw reads, plus TimeAverage and Total
|
||||
through AVEVA retrieval modes (`AggregateModeMapper`); for any other member it throws
|
||||
`HistoryAggregateNotSupportedException` without calling the gateway (and since #682 the server refuses it up front), which the HistoryRead path turns into
|
||||
`BadAggregateNotSupported` for that node only (not a generic failure, and not counted as a backend fault). A NodeId outside the mapped set (e.g. `StandardDeviationSample`) is still `BadAggregateNotSupported`.
|
||||
|
||||
### Advertised aggregates (`AggregateFunctions`, #682)
|
||||
|
||||
@@ -794,7 +862,7 @@ aggregate-function objects. A client choosing a history aggregate should read **
|
||||
|
||||
| `ServerHistorian:Backend` | i=11201 (history) | i=2997 (server) | Declared by |
|
||||
|---|---|---|---|
|
||||
| `Gateway` | 4 — TimeAverage, Minimum, Maximum, Total | all 37 | `GatewayHistorianDataSource.ServedAggregates` (today `AggregateModeMapper.Served`, derived from its mapping switch) |
|
||||
| `Gateway` | 12 — the ten per-sample aggregates (raw-evaluated, #681) + TimeAverage, Total (native AVEVA modes) | all 37 | `GatewayHistorianDataSource.ServedAggregates` (the per-sample families of `AggregateFamilies` ∪ `AggregateModeMapper.NativeAggregates`, derived from its mapping switch) |
|
||||
| `SqlServer`, `QuestDb` | 17 — the [aggregate table](#aggregates-processed-reads) | all 37 | `AggregateFamilies.Served` (Sql.Common; every member with a family) |
|
||||
| historian disabled (Null source) | **none** | all 37 | `NullHistorianDataSource` |
|
||||
|
||||
|
||||
+113
-10
@@ -1,7 +1,10 @@
|
||||
using System.Collections.Frozen;
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging;
|
||||
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;
|
||||
|
||||
@@ -23,8 +26,26 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||||
/// </remarks>
|
||||
public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
|
||||
{
|
||||
/// <summary>
|
||||
/// The processed (aggregate) functions the Gateway backend serves — the single source of truth for this backend's
|
||||
/// aggregate set (#681): the per-sample aggregates evaluated from raw reads (every member of the
|
||||
/// <see cref="AggregateFamily.Simple"/> / <see cref="AggregateFamily.StartEnd"/> /
|
||||
/// <see cref="AggregateFamily.ActualTime"/> families of <see cref="AggregateFamilies"/>: Average, Count, Minimum,
|
||||
/// Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA
|
||||
/// (TimeAverage, Total — <see cref="AggregateModeMapper"/>). Any other aggregate throws
|
||||
/// <see cref="HistoryAggregateNotSupportedException"/> without calling the gateway. It is what
|
||||
/// <see cref="SupportedAggregates"/> returns, so the advertised <c>AggregateFunctions</c> folders and the server's
|
||||
/// up-front <c>BadAggregateNotSupported</c> read it too (#682).
|
||||
/// </summary>
|
||||
public static IReadOnlySet<HistoryAggregateType> ServedAggregates { get; } =
|
||||
Enum.GetValues<HistoryAggregateType>()
|
||||
.Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a))
|
||||
.ToFrozenSet();
|
||||
|
||||
private readonly IHistorianGatewayClient _client;
|
||||
private readonly ILogger<GatewayHistorianDataSource> _logger;
|
||||
private readonly int _maxRawRows;
|
||||
private readonly TimeProvider _time;
|
||||
|
||||
private readonly object _healthLock = new();
|
||||
private DateTime? _lastSuccessUtc;
|
||||
@@ -38,12 +59,20 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
/// <summary>Creates a gateway-backed historian data source.</summary>
|
||||
/// <param name="client">The gateway client seam used for all reads.</param>
|
||||
/// <param name="logger">Diagnostic logger; failures are recorded without leaking tag/host detail.</param>
|
||||
public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger)
|
||||
/// <param name="maxRawRows">
|
||||
/// In-window raw-row cap of one raw-evaluated processed read (<c>ServerHistorian:MaxTimeWeightedRawRows</c>;
|
||||
/// <= 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>
|
||||
public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
|
||||
int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(client);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
_client = client;
|
||||
_logger = logger;
|
||||
_maxRawRows = maxRawRows;
|
||||
_time = timeProvider ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -75,17 +104,38 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Per-sample aggregates (#681)</b> — Average, Count, Minimum, Maximum, Range, Start, End, Delta,
|
||||
/// Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator
|
||||
/// (<see cref="RawAggregateReader"/>): identical values, status, Calculated/Partial bits and timestamps to
|
||||
/// SqlServer/QuestDB for the same samples, forward and reverse (#677). The in-window raw rows are capped by
|
||||
/// <c>ServerHistorian:MaxTimeWeightedRawRows</c> (<see cref="HistoryReadLimitExceededException"/>, not counted as
|
||||
/// a backend failure).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>TimeAverage / Total</b> stay native AVEVA retrieval modes (<see cref="AggregateModeMapper"/>), stamped with
|
||||
/// the interval start and the Calculated bit. Everything else is <see cref="HistoryAggregateNotSupportedException"/>
|
||||
/// (see <see cref="ServedAggregates"/>).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <exception cref="ArgumentOutOfRangeException">A raw-evaluated read whose window holds more than
|
||||
/// <see cref="BucketPlan.MaxBuckets"/> intervals.</exception>
|
||||
public async Task<HistoryReadResult> ReadProcessedAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||
{
|
||||
if (RawAggregateReader.Serves(aggregate))
|
||||
return await ReadProcessedFromRawAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
// 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);
|
||||
try
|
||||
{
|
||||
// Total is a native gateway retrieval mode (Integral) — no client-side scaling. Average and Count
|
||||
// have no AVEVA mode and were rejected by the mapper above (#679).
|
||||
// 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)
|
||||
@@ -98,6 +148,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
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);
|
||||
@@ -105,6 +159,62 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Serves a per-sample aggregate from a raw read (#681); see <see cref="RawAggregateReader"/>.</summary>
|
||||
private async Task<HistoryReadResult> ReadProcessedFromRawAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||
{
|
||||
// Planned OUTSIDE the failure-recording try (as on the SQL backends): too many intervals is a client request
|
||||
// limit, not a backend fault.
|
||||
var plan = BucketPlan.Create(startUtc, endUtc, interval);
|
||||
try
|
||||
{
|
||||
var now = _time.GetUtcNow().UtcDateTime;
|
||||
var values = await RawAggregateReader
|
||||
.ReadAsync(_client, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
RecordOutcome(success: true, error: null);
|
||||
return new HistoryReadResult(values, ContinuationPoint: null);
|
||||
}
|
||||
catch (HistoryReadLimitExceededException)
|
||||
{
|
||||
// A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
|
||||
throw;
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// The caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault.
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
|
||||
{
|
||||
// The gateway refused the row budget (Historian:ReadLimits:MaxValuesCeiling with
|
||||
// AllowStreamingBeyondCeiling=false): the same client-request limit as the cap, not a backend fault.
|
||||
throw new HistoryReadLimitExceededException(_maxRawRows,
|
||||
$"The HistorianGateway refused the raw row budget of the processed read of '{fullReference}' (its " +
|
||||
"Historian:ReadLimits:MaxValuesCeiling); lower ServerHistorian:MaxTimeWeightedRawRows below that ceiling " +
|
||||
"or narrow the window.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// True when <paramref name="ex"/> (or an inner exception — the client package wraps the <see cref="RpcException"/>)
|
||||
/// is the gateway's <c>InvalidArgument</c> "max_values exceeds the server ceiling" rejection.
|
||||
/// </summary>
|
||||
internal static bool IsMaxValuesCeilingRejection(Exception ex)
|
||||
{
|
||||
for (var e = ex; e is not null; e = e.InnerException)
|
||||
if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc
|
||||
&& rpc.Status.Detail.Contains("max_values exceeds the server ceiling", StringComparison.Ordinal))
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<HistoryReadResult> ReadAtTimeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
||||
@@ -132,13 +242,6 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
/// </remarks>
|
||||
public bool SupportsReverseEvents => false;
|
||||
|
||||
/// <summary>
|
||||
/// The aggregates the Gateway backend serves — the single static the advertised <c>AggregateFunctions</c> folders
|
||||
/// and the server's up-front <c>BadAggregateNotSupported</c> read through <see cref="SupportedAggregates"/> (#682).
|
||||
/// Today: <see cref="AggregateModeMapper.Served"/>, the members the gateway has a retrieval mode for.
|
||||
/// </summary>
|
||||
public static readonly IReadOnlySet<HistoryAggregateType> ServedAggregates = AggregateModeMapper.Served;
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks><see cref="ServedAggregates"/>; every other member answers <c>BadAggregateNotSupported</c> and is not advertised.</remarks>
|
||||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates => ServedAggregates;
|
||||
|
||||
+3
-1
@@ -38,7 +38,9 @@ public static class GatewayHistorian
|
||||
|
||||
return new GatewayHistorianDataSource(
|
||||
HistorianGatewayClientAdapter.Create(options, loggerFactory),
|
||||
logger);
|
||||
logger,
|
||||
// #681: the raw-evaluated per-sample aggregates share the SQL backends' raw-row cap.
|
||||
maxRawRows: options.MaxTimeWeightedRawRows);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+17
-15
@@ -6,29 +6,33 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
|
||||
|
||||
/// <summary>
|
||||
/// Maps the driver-agnostic <see cref="HistoryAggregateType"/> (OPC UA Part 13 aggregate) onto the
|
||||
/// gateway's native <see cref="RetrievalMode"/>.
|
||||
/// gateway's native <see cref="RetrievalMode"/> — for the aggregates still answered by AVEVA itself.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Served: <see cref="HistoryAggregateType.TimeAverage"/> → <see cref="RetrievalMode.TimeWeightedAverage"/>,
|
||||
/// Minimum → <see cref="RetrievalMode.MinimumWithTime"/>, Maximum → <see cref="RetrievalMode.MaximumWithTime"/>,
|
||||
/// Total → <see cref="RetrievalMode.Integral"/>.
|
||||
/// Native (AVEVA-computed): <see cref="HistoryAggregateType.TimeAverage"/> → <see cref="RetrievalMode.TimeWeightedAverage"/>
|
||||
/// and <see cref="HistoryAggregateType.Total"/> → <see cref="RetrievalMode.Integral"/>. They stay native because an exact
|
||||
/// Part 13 time-weighted evaluation needs the real bounding samples on both sides of the window (time, value and
|
||||
/// quality of the latest sample before it), and the gateway has no "previous stored sample" read — a raw read streams
|
||||
/// forward from its start and AVEVA replaces whatever lies before with an initial-value row stamped at the query start,
|
||||
/// losing the original time a sloped segment across the window start needs (#681).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Deliberately NOT served (#679): <see cref="HistoryAggregateType.Average"/> (OPC UA Average is the arithmetic mean;
|
||||
/// AVEVA's modes include no arithmetic mean — only the time-weighted one, which is TimeAverage) and
|
||||
/// <see cref="HistoryAggregateType.Count"/> (AVEVA <c>Counter</c> is a rollover-counter increase, not a sample count).
|
||||
/// Both surface as <c>BadAggregateNotSupported</c> rather than returning a different statistic under the requested name.
|
||||
/// Since #681 the per-sample aggregates (Average, Count, Minimum, Maximum, Range, Start, End, Delta,
|
||||
/// Minimum/MaximumActualTime) are NOT native: <see cref="GatewayHistorianDataSource"/> evaluates them from raw reads
|
||||
/// with the SQL backends' shared evaluator (Minimum/Maximum used to map to AVEVA's <c>MinimumWithTime</c> /
|
||||
/// <c>MaximumWithTime</c>). Every other member — incl. TimeAverage2, Total2, Interpolative, PercentGood, PercentBad — is
|
||||
/// <see cref="HistoryAggregateNotSupportedException"/> here.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class AggregateModeMapper
|
||||
{
|
||||
/// <summary>
|
||||
/// The aggregates the gateway serves — every defined member <see cref="TryMap"/> maps, derived from it so the
|
||||
/// advertised set and the dispatch can never disagree. <see cref="GatewayHistorianDataSource.SupportedAggregates"/>
|
||||
/// returns it (#682); it is the ONLY declaration of the gateway's served set.
|
||||
/// The aggregates answered by a native AVEVA retrieval mode — every defined member <see cref="TryMap"/> maps, derived
|
||||
/// from it so the dispatch and the declared set can never disagree. Part of
|
||||
/// <see cref="GatewayHistorianDataSource.ServedAggregates"/> (#681/#682).
|
||||
/// </summary>
|
||||
public static readonly IReadOnlySet<HistoryAggregateType> Served =
|
||||
public static readonly FrozenSet<HistoryAggregateType> NativeAggregates =
|
||||
Enum.GetValues<HistoryAggregateType>().Where(a => TryMap(a) is not null).ToFrozenSet();
|
||||
|
||||
/// <summary>Maps an aggregate function to the gateway retrieval mode.</summary>
|
||||
@@ -48,12 +52,10 @@ internal static class AggregateModeMapper
|
||||
nameof(aggregate), aggregate, "Unmapped HistoryAggregateType — add a RetrievalMode mapping.");
|
||||
}
|
||||
|
||||
/// <summary>The single mapping table: the retrieval mode for a served aggregate, or null.</summary>
|
||||
/// <summary>The single mapping table: the retrieval mode for a natively served aggregate, or null.</summary>
|
||||
private static RetrievalMode? TryMap(HistoryAggregateType aggregate) => aggregate switch
|
||||
{
|
||||
HistoryAggregateType.TimeAverage => RetrievalMode.TimeWeightedAverage,
|
||||
HistoryAggregateType.Minimum => RetrievalMode.MinimumWithTime,
|
||||
HistoryAggregateType.Maximum => RetrievalMode.MaximumWithTime,
|
||||
HistoryAggregateType.Total => RetrievalMode.Integral,
|
||||
_ => null,
|
||||
};
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
|
||||
|
||||
@@ -30,16 +31,23 @@ internal static class SampleMapper
|
||||
else
|
||||
value = null;
|
||||
|
||||
return new DataValueSnapshot(
|
||||
Value: value,
|
||||
StatusCode: ToStatus(sample),
|
||||
SourceTimestampUtc: sample.Timestamp?.ToDateTime(), // Utc kind
|
||||
ServerTimestampUtc: DateTime.UtcNow);
|
||||
}
|
||||
|
||||
/// <summary>The OPC UA status of a raw sample (shared by raw reads and the #681 raw-evaluated aggregates).</summary>
|
||||
/// <param name="sample">The gateway raw sample.</param>
|
||||
/// <returns>The mapped status code.</returns>
|
||||
public static uint ToStatus(HistorianSample sample)
|
||||
{
|
||||
// Prefer the OPC DA quality byte (opc_quality); the gateway populates it directly from the
|
||||
// SDK's OpcQuality, so it is the authoritative byte for GatewayQualityMapper. Fall back to
|
||||
// the historian quality field only when opc_quality is unset (0).
|
||||
byte qualityByte = sample.OpcQuality != 0 ? (byte)sample.OpcQuality : (byte)sample.Quality;
|
||||
|
||||
return new DataValueSnapshot(
|
||||
Value: value,
|
||||
StatusCode: GatewayQualityMapper.Map(qualityByte),
|
||||
SourceTimestampUtc: sample.Timestamp?.ToDateTime(), // Utc kind
|
||||
ServerTimestampUtc: DateTime.UtcNow);
|
||||
return GatewayQualityMapper.Map(qualityByte);
|
||||
}
|
||||
|
||||
/// <summary>Maps a batch of raw samples to value snapshots, in order.</summary>
|
||||
@@ -57,20 +65,28 @@ internal static class SampleMapper
|
||||
/// <param name="aggregate">The gateway aggregate sample.</param>
|
||||
/// <returns>The driver-agnostic snapshot.</returns>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Unlike the legacy Wonderware DTO (a nullable <c>double?</c>), the gateway proto carries a
|
||||
/// non-optional <c>double value</c>, so an unavailable (no-data) bucket cannot be signalled by a
|
||||
/// null value. Instead it is signalled by a non-Good <c>opc_quality</c>: a Good bucket
|
||||
/// (<c>opc_quality >= 192</c>) yields its value with a Good status, anything else maps to
|
||||
/// <c>BadNoData</c> with a null value — preserving the Wonderware aggregate contract (binary
|
||||
/// Good-with-value / BadNoData-null).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// #681: both statuses carry the Calculated info bits (<c>0x401</c>) like every other backend's processed value, and
|
||||
/// the bucket is stamped with its interval START (Part 13; it was the interval end), falling back to the end only when
|
||||
/// the start is absent. Pending the live leg: histsdk decodes <c>start_time</c> from a row field (<c>tail + 43</c>) its
|
||||
/// reverse-engineering notes still mark unverified.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static DataValueSnapshot ToAggregateSnapshot(HistorianAggregateSample aggregate)
|
||||
{
|
||||
bool hasData = aggregate.OpcQuality >= GoodQualityFloor;
|
||||
return new DataValueSnapshot(
|
||||
Value: hasData ? aggregate.Value : null, // boxes as System.Double when present
|
||||
StatusCode: hasData ? StatusGood : StatusBadNoData,
|
||||
SourceTimestampUtc: (aggregate.EndTime ?? aggregate.StartTime)?.ToDateTime(), // bucket timestamp
|
||||
StatusCode: HistorianInfoBits.WithCalculated(hasData ? StatusGood : StatusBadNoData),
|
||||
SourceTimestampUtc: (aggregate.StartTime ?? aggregate.EndTime)?.ToDateTime(), // bucket timestamp = interval start
|
||||
ServerTimestampUtc: DateTime.UtcNow);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,312 @@
|
||||
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>
|
||||
/// #681: serves the per-sample processed aggregates (the <see cref="AggregateFamily.Simple"/>,
|
||||
/// <see cref="AggregateFamily.StartEnd"/> and <see cref="AggregateFamily.ActualTime"/> families) on the Gateway backend by
|
||||
/// streaming the window's RAW samples through the gateway and evaluating them with the shared
|
||||
/// <c>Driver.Historian.Sql.Common</c> code — <see cref="BucketPlan"/> for the bucket math (forward and reverse, #677) and
|
||||
/// <see cref="AggregateEvaluator.EvaluateAll"/> for values, status and the Partial/Calculated bits — so the Gateway answers
|
||||
/// exactly what SqlServer and QuestDB answer for the same stored samples.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Forward raw reads, each consumed to completion.</b> The gateway's <c>ReadRaw</c> streams oldest-first over an
|
||||
/// inclusive <c>[start, end]</c> and refuses <c>end < start</c>, so a reverse read streams the same ascending range
|
||||
/// and <see cref="BucketPlan"/> assembles the result newest-first. The window stream is classified sample by sample
|
||||
/// with <see cref="BucketPlan.IndexOf"/> (<c>[start, end)</c> forward, <c>(start, end]</c> reverse) — memory is
|
||||
/// O(buckets), never O(samples). No gateway stream is ever abandoned mid-way: a client-side break cancels the call,
|
||||
/// and the gateway currently classifies that cancellation as a connection fault and evicts its shared historian
|
||||
/// session (HistorianGateway <c>HistorianRpcFaults.IsConnectionFault</c>: any <c>RpcException</c> ⇒ true).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The AVEVA start-boundary row.</b> An AVEVA <c>Full</c> raw read emits an initial-value row stamped exactly at
|
||||
/// the query start (histsdk reverse-engineering: <c>Quality = 133</c>, <c>OpcQuality = 192</c>) carrying the value
|
||||
/// in effect there, not a stored sample. The window stream is therefore requested from <see cref="BoundaryMargin"/>
|
||||
/// (1 ms) BEFORE the window, so that row lands outside every bucket and can never be counted as a sample. A row
|
||||
/// before the window counts only as evidence that the tag has earlier data, and only when its OPC quality is not
|
||||
/// Bad — a "no earlier data" NULL row still arrives WITH a value (histsdk sets <c>NumericValue = 0</c> on every
|
||||
/// numeric row, <c>StringValue = ""</c> on a string one) but with a Bad OPC quality (captured: 1, no-data).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The data range.</b> The start/end-of-data Partial rules and Count's in-range empty interval need the tag's
|
||||
/// earliest sample at or before the window's later edge and latest at or after its earlier edge
|
||||
/// (<see cref="AggregateEvaluator.NeedsDataRange"/>; a plain <c>Average</c> needs neither). Earlier data is a non-Bad
|
||||
/// pre-window row, or — when none arrived — a <see cref="ProbeRows"/>-row read <c>[</c><see cref="ProbeFloor"/><c>,
|
||||
/// start)</c>. Later data is a sample exactly at the window end (forward) from the window read, or else a
|
||||
/// <see cref="ProbeRows"/>-row read <c>[end, now + </c><see cref="LookAheadHorizon"/><c>]</c> (the historian stores
|
||||
/// nothing further in the future). In both small reads a row stamped exactly at the query start (a boundary row) or
|
||||
/// at the query end (a possible terminal row) is not a stored sample, whatever its value.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Bounded.</b> The in-window rows are capped by <c>ServerHistorian:MaxTimeWeightedRawRows</c> (the same knob the
|
||||
/// SQL backends' time-weighted reads use): past it the node fails with
|
||||
/// <see cref="HistoryReadLimitExceededException"/> (⇒ <c>BadTooManyOperations</c>) once the stream — bounded by the
|
||||
/// requested row budget — has been drained. The gateway is asked for at most <c>cap + 2 + </c>
|
||||
/// <see cref="PreWindowAllowance"/> rows; a stream that delivers all of them cannot be proven complete and fails the
|
||||
/// same way rather than answering from a truncated window.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Values and status.</b> A sample's value is its numeric value (Double), else its string value, else Empty; its
|
||||
/// status is <see cref="SampleMapper.ToStatus"/> — the same quality mapping a raw HistoryRead serves (C-002: the
|
||||
/// gateway stores no quality, so AVEVA history is normally all Good). An Empty sample is clamped Bad
|
||||
/// (<see cref="StoredSample.FromStored"/>), exactly as on the SQL backends.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class RawAggregateReader
|
||||
{
|
||||
/// <summary>How far before the window the raw stream starts, so the AVEVA initial-value row lands outside it.</summary>
|
||||
internal static readonly TimeSpan BoundaryMargin = TimeSpan.FromMilliseconds(1);
|
||||
|
||||
/// <summary>How far past "now" the stream may look for a sample after the window (end-of-data rules).</summary>
|
||||
internal static readonly TimeSpan LookAheadHorizon = TimeSpan.FromDays(1);
|
||||
|
||||
/// <summary>Rows the stream may spend before the window (the boundary row plus samples inside the margin).</summary>
|
||||
internal const int PreWindowAllowance = 16;
|
||||
|
||||
/// <summary>Start of the "is there data before the window?" probe (the Unix epoch — before any AVEVA history).</summary>
|
||||
internal static readonly DateTime ProbeFloor = DateTime.UnixEpoch;
|
||||
|
||||
/// <summary>Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample).</summary>
|
||||
internal const int ProbeRows = 2;
|
||||
|
||||
/// <summary>True for the aggregates this reader serves: the per-sample families of <see cref="AggregateFamilies"/>.</summary>
|
||||
/// <param name="aggregate">Requested aggregate.</param>
|
||||
/// <returns>Whether the aggregate is evaluated from raw samples.</returns>
|
||||
public static bool Serves(HistoryAggregateType aggregate) =>
|
||||
AggregateFamilies.FamilyOf(aggregate) is AggregateFamily.Simple or AggregateFamily.StartEnd or AggregateFamily.ActualTime;
|
||||
|
||||
/// <summary>The gateway <c>max_values</c> for a configured window cap (<see cref="int.MaxValue"/> when unbounded).</summary>
|
||||
/// <param name="maxWindowRows">The configured cap (<= 0 = unbounded).</param>
|
||||
/// <returns>The row limit to request.</returns>
|
||||
public static int MaxValuesFor(int maxWindowRows) =>
|
||||
maxWindowRows <= 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)maxWindowRows + 2 + PreWindowAllowance);
|
||||
|
||||
/// <summary>Maps one gateway raw sample to a stored sample (value + mapped status, Empty clamped Bad).</summary>
|
||||
/// <param name="sample">The gateway sample; its timestamp must be set.</param>
|
||||
/// <returns>The stored sample.</returns>
|
||||
public static StoredSample ToStored(HistorianSample sample)
|
||||
{
|
||||
var value = sample.HasNumericValue ? HistorianValue.FromDouble(sample.NumericValue)
|
||||
: sample.HasStringValue ? HistorianValue.FromString(sample.StringValue)
|
||||
: HistorianValue.Empty;
|
||||
return StoredSample.FromStored(sample.Timestamp.ToDateTime(), value, SampleMapper.ToStatus(sample));
|
||||
}
|
||||
|
||||
/// <summary>Reads and evaluates one per-sample processed read.</summary>
|
||||
/// <param name="client">The gateway client.</param>
|
||||
/// <param name="tag">Historian tag name.</param>
|
||||
/// <param name="aggregate">A per-sample aggregate (<see cref="Serves"/>).</param>
|
||||
/// <param name="plan">The bucket plan.</param>
|
||||
/// <param name="maxWindowRows">The in-window raw-row cap (<= 0 = unbounded).</param>
|
||||
/// <param name="now">Server timestamp, and the anchor of the look-ahead horizon.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>One value per interval, in the plan's order.</returns>
|
||||
/// <exception cref="HistoryReadLimitExceededException">The window holds more than <paramref name="maxWindowRows"/> rows,
|
||||
/// or the stream could not be read to completion within the requested row budget.</exception>
|
||||
public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IHistorianGatewayClient client, string tag,
|
||||
HistoryAggregateType aggregate, BucketPlan plan, int maxWindowRows, DateTime now, CancellationToken ct)
|
||||
{
|
||||
if (plan.BucketCount == 0) return [];
|
||||
var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan);
|
||||
var requestStart = plan.StartUtc.Ticks > BoundaryMargin.Ticks ? plan.StartUtc - BoundaryMargin : plan.StartUtc;
|
||||
var maxValues = MaxValuesFor(maxWindowRows);
|
||||
|
||||
var buckets = new RawBucketAccumulator(aggregate == HistoryAggregateType.MaximumActualTime);
|
||||
DateTime? earliest = null, latest = null;
|
||||
long windowRows = 0, received = 0;
|
||||
var overCap = false;
|
||||
// Consumed to COMPLETION (never broken out of): abandoning a gateway stream cancels the call server-side, and the
|
||||
// gateway currently treats that cancellation as a connection fault and evicts its shared historian session. Past
|
||||
// the cap the rest is drained (bounded by maxValues) without being evaluated.
|
||||
await foreach (var raw in client.ReadRawAsync(tag, requestStart, plan.EndUtc, maxValues, ct).ConfigureAwait(false))
|
||||
{
|
||||
received++;
|
||||
if (overCap || raw.Timestamp is null) continue;
|
||||
var s = ToStored(raw);
|
||||
var t = s.TimeUtc;
|
||||
if (plan.IndexOf(t) is { } bucket)
|
||||
{
|
||||
if (maxWindowRows > 0 && ++windowRows > maxWindowRows)
|
||||
{
|
||||
overCap = true;
|
||||
continue;
|
||||
}
|
||||
buckets.Add(bucket, s);
|
||||
}
|
||||
// The data range as the SQL backends define it: earliest stored sample at or before the window's later edge,
|
||||
// latest at or after its earlier edge. A row before the window (the AVEVA boundary row, or a stored sample in
|
||||
// the margin) proves earlier data only when its OPC quality is not Bad: AVEVA's boundary row for a tag with no
|
||||
// earlier data is a NULL row, which arrives WITH a value (histsdk sets NumericValue = 0 / StringValue "") and
|
||||
// a Bad OPC quality (captured: 1, no-data).
|
||||
if (t >= plan.StartUtc || IsNotBadOnTheWire(raw))
|
||||
{
|
||||
if (t <= plan.EndUtc && (earliest is null || t < earliest)) earliest = t;
|
||||
if (t >= plan.StartUtc && (latest is null || t > latest)) latest = t;
|
||||
}
|
||||
}
|
||||
if (overCap || (maxWindowRows > 0 && received >= maxValues))
|
||||
throw LimitExceeded(tag, maxWindowRows);
|
||||
|
||||
if (needsRange)
|
||||
{
|
||||
// Earlier data: no non-Bad pre-window row ⇒ ask explicitly (normally only when the tag's history really starts
|
||||
// inside the window).
|
||||
var earlierSeen = earliest is { } e && (e < plan.StartUtc || (plan.Reversed && e == plan.StartUtc));
|
||||
if (!earlierSeen && plan.StartUtc.AddTicks(-1) > ProbeFloor)
|
||||
if (await FirstSampleAsync(client, tag, ProbeFloor, plan.StartUtc.AddTicks(-1), ct).ConfigureAwait(false) is { } before)
|
||||
earliest = before;
|
||||
|
||||
// Later data: the main read covers a sample exactly at the window end (forward: that is already later data);
|
||||
// otherwise a small separate read from the end to now + horizon.
|
||||
var laterSeen = latest is { } l && (plan.Reversed ? l > plan.EndUtc : l >= plan.EndUtc);
|
||||
var horizon = DateTime.MaxValue - now > LookAheadHorizon
|
||||
? DateTime.SpecifyKind(now + LookAheadHorizon, DateTimeKind.Utc)
|
||||
: DateTime.MaxValue;
|
||||
if (!laterSeen && horizon > plan.EndUtc)
|
||||
if (await FirstSampleAsync(client, tag, plan.EndUtc, horizon, ct).ConfigureAwait(false) is { } after)
|
||||
latest = after;
|
||||
}
|
||||
|
||||
return AggregateEvaluator.EvaluateAll(aggregate, plan, buckets.Rows(), earliest, latest, now);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The time of the first stored sample (any quality) strictly between <paramref name="startUtc"/> and
|
||||
/// <paramref name="endUtc"/>, read as a <see cref="ProbeRows"/>-row raw read consumed to completion. A row stamped
|
||||
/// exactly at the query start is the historian's boundary row (the value in effect there, or a NULL row when there is
|
||||
/// none) and a row exactly at the query end may be a terminal row — neither is a stored sample, whatever its value.
|
||||
/// </summary>
|
||||
private static async Task<DateTime?> FirstSampleAsync(IHistorianGatewayClient client, string tag, DateTime startUtc,
|
||||
DateTime endUtc, CancellationToken ct)
|
||||
{
|
||||
DateTime? found = null;
|
||||
await foreach (var raw in client.ReadRawAsync(tag, startUtc, endUtc, ProbeRows, ct).ConfigureAwait(false))
|
||||
{
|
||||
if (found is not null || raw.Timestamp is null) continue; // drain, never break (see ReadAsync)
|
||||
var t = raw.Timestamp.ToDateTime();
|
||||
if (t > startUtc && t < endUtc) found = t;
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The row's OPC DA quality alone (<c>opc_quality</c> ≥ 64: Good or Uncertain) — deliberately without
|
||||
/// <see cref="SampleMapper.ToStatus"/>'s fallback to the AVEVA <c>quality</c> field when <c>opc_quality</c> is 0, which
|
||||
/// is also the OPC DA Bad value a NULL row carries.
|
||||
/// </summary>
|
||||
private static bool IsNotBadOnTheWire(HistorianSample raw) => (byte)raw.OpcQuality >= 64; // low byte, as SampleMapper.ToStatus reads it
|
||||
|
||||
private static HistoryReadLimitExceededException LimitExceeded(string tag, int maxWindowRows) =>
|
||||
new(maxWindowRows,
|
||||
$"The processed read of '{tag}' covers more than {maxWindowRows} raw samples " +
|
||||
"(ServerHistorian:MaxTimeWeightedRawRows); the Gateway backend evaluates per-sample aggregates from raw reads — " +
|
||||
"narrow the window.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds the per-bucket statistics (<see cref="BucketRow"/>) the SQL backends' aggregate queries produce, from a stream
|
||||
/// of raw samples — the C# statement of <c>SqlAggregateQueries</c>' columns: counts by class (usable = non-Bad with a
|
||||
/// value; good = usable with Good severity; NonNumeric = usable String/DateTime; GoodNonNumeric = good String/DateTime),
|
||||
/// Avg/Min/Max/Sum over the GOOD numeric values and UncertainMin/UncertainMax over the usable-but-not-Good numeric values
|
||||
/// (#684), first/last usable value by time, the extreme GOOD numeric value with its time (ties ⇒ earliest), and the
|
||||
/// first/last sample of any quality.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Keep in step with the SQL backends' per-sample value columns: a change to which samples feed a statistic there
|
||||
/// (e.g. Uncertain handling) must be repeated here, or the Gateway backend silently diverges.
|
||||
/// </remarks>
|
||||
internal sealed class RawBucketAccumulator
|
||||
{
|
||||
private readonly bool _maximum;
|
||||
private readonly Dictionary<long, State> _buckets = new();
|
||||
|
||||
/// <summary>Creates an accumulator.</summary>
|
||||
/// <param name="maximum">Extreme = the maximum (MaximumActualTime); otherwise the minimum.</param>
|
||||
public RawBucketAccumulator(bool maximum) => _maximum = maximum;
|
||||
|
||||
private sealed class State
|
||||
{
|
||||
public int Total, Usable, Good, NonNumeric, GoodNonNumeric, GoodNumeric, UncertainNumeric;
|
||||
public double Sum, Min = double.PositiveInfinity, Max = double.NegativeInfinity;
|
||||
public double UncertainMin = double.PositiveInfinity, UncertainMax = double.NegativeInfinity;
|
||||
public StoredSample? First, Last, FirstAny, LastAny, Extreme;
|
||||
}
|
||||
|
||||
/// <summary>Adds one in-window sample to its bucket. Order-independent (picks compare by time).</summary>
|
||||
/// <param name="bucket">Ascending bucket index (<see cref="BucketPlan.IndexOf"/>).</param>
|
||||
/// <param name="s">The sample.</param>
|
||||
public void Add(long bucket, StoredSample s)
|
||||
{
|
||||
if (!_buckets.TryGetValue(bucket, out var st)) _buckets[bucket] = st = new State();
|
||||
st.Total++;
|
||||
if (st.FirstAny is not { } fa || s.TimeUtc < fa.TimeUtc) st.FirstAny = s;
|
||||
if (st.LastAny is not { } la || s.TimeUtc >= la.TimeUtc) st.LastAny = s;
|
||||
if (!s.IsUsable) return;
|
||||
st.Usable++;
|
||||
if (s.IsGood) st.Good++;
|
||||
if (st.First is not { } f || s.TimeUtc < f.TimeUtc) st.First = s;
|
||||
if (st.Last is not { } l || s.TimeUtc >= l.TimeUtc) st.Last = s;
|
||||
if (!s.Value.TryGetDouble(out var v))
|
||||
{
|
||||
st.NonNumeric++;
|
||||
if (s.IsGood) st.GoodNonNumeric++;
|
||||
return;
|
||||
}
|
||||
if (!s.IsGood)
|
||||
{
|
||||
// #684: an Uncertain value is not part of the value set; it only feeds the Min/Max status rule.
|
||||
st.UncertainNumeric++;
|
||||
if (v < st.UncertainMin) st.UncertainMin = v;
|
||||
if (v > st.UncertainMax) st.UncertainMax = v;
|
||||
return;
|
||||
}
|
||||
st.GoodNumeric++;
|
||||
st.Sum += v;
|
||||
if (v < st.Min) st.Min = v;
|
||||
if (v > st.Max) st.Max = v;
|
||||
if (st.Extreme is not { } e) st.Extreme = s;
|
||||
else
|
||||
{
|
||||
var ev = e.Value.Number;
|
||||
var better = _maximum ? v > ev : v < ev;
|
||||
if (better || (v == ev && s.TimeUtc < e.TimeUtc)) st.Extreme = s;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The non-empty buckets' rows, ascending by bucket index.</summary>
|
||||
/// <returns>The rows.</returns>
|
||||
public IEnumerable<BucketRow> Rows() =>
|
||||
_buckets.OrderBy(kv => kv.Key).Select(kv =>
|
||||
{
|
||||
var st = kv.Value;
|
||||
var hasNumeric = st.GoodNumeric > 0;
|
||||
var hasUncertain = st.UncertainNumeric > 0;
|
||||
return new BucketRow
|
||||
{
|
||||
Bucket = kv.Key,
|
||||
Total = st.Total,
|
||||
Usable = st.Usable,
|
||||
Good = st.Good,
|
||||
Bad = st.Total - st.Usable,
|
||||
NonNumeric = st.NonNumeric,
|
||||
GoodNonNumeric = st.GoodNonNumeric,
|
||||
Avg = hasNumeric ? st.Sum / st.GoodNumeric : null,
|
||||
Min = hasNumeric ? st.Min : null,
|
||||
Max = hasNumeric ? st.Max : null,
|
||||
Sum = hasNumeric ? st.Sum : null,
|
||||
UncertainMin = hasUncertain ? st.UncertainMin : null,
|
||||
UncertainMax = hasUncertain ? st.UncertainMax : null,
|
||||
First = st.First?.Value ?? HistorianValue.Empty,
|
||||
Last = st.Last?.Value ?? HistorianValue.Empty,
|
||||
Extreme = st.Extreme?.Value.Number,
|
||||
ExtremeTime = st.Extreme?.TimeUtc,
|
||||
FirstAny = st.FirstAny,
|
||||
LastAny = st.LastAny,
|
||||
};
|
||||
});
|
||||
}
|
||||
+5
@@ -11,6 +11,11 @@
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.csproj"/>
|
||||
<!-- #681: the per-sample processed aggregates are evaluated from raw reads with the SAME database-independent code the
|
||||
SQL backends use (BucketPlan, BucketRow, AggregateEvaluator), so Gateway matches SqlServer/QuestDB exactly.
|
||||
Sql.Common references only Core.Abstractions + Core.AlarmHistorian (both already referenced here) and no
|
||||
driver, so this adds no cycle. -->
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj"/>
|
||||
<!-- Runtime owns ServerHistorianOptions (the bound ServerHistorian config). The read-cutover
|
||||
factory (GatewayHistorian.CreateDataSource / HistorianGatewayClientAdapter.Create) maps
|
||||
those options onto the package client, so the driver references it. Runtime references no
|
||||
|
||||
+24
-1
@@ -19,6 +19,17 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
public DateTime LastReadRawEndUtc;
|
||||
public int LastReadRawMaxValues;
|
||||
public int ReadRawCallCount;
|
||||
/// <summary>
|
||||
/// When set, ReadRaw serves <c>RawSource(start, end, maxValues)</c> instead of <see cref="RawSamples"/> (a
|
||||
/// store-like fake honouring the window, #681).
|
||||
/// </summary>
|
||||
public Func<DateTime, DateTime, int, IEnumerable<HistorianSample>>? RawSource;
|
||||
/// <summary>Raw samples actually pulled by the consumer across every ReadRaw stream.</summary>
|
||||
public int RawYieldedCount;
|
||||
/// <summary>ReadRaw streams the consumer enumerated to the END (never broke out of — #681 review).</summary>
|
||||
public int RawStreamsCompleted;
|
||||
/// <summary>Every ReadRaw call's (start, end, maxValues), in order.</summary>
|
||||
public readonly List<(DateTime Start, DateTime End, int MaxValues)> RawCalls = new();
|
||||
|
||||
// ---- ReadAggregate -------------------------------------------------------------------------
|
||||
public IReadOnlyList<HistorianAggregateSample> AggregateSamples = Array.Empty<HistorianAggregateSample>();
|
||||
@@ -90,7 +101,9 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
LastReadRawEndUtc = endUtc;
|
||||
LastReadRawMaxValues = maxValues;
|
||||
ReadRawCallCount++;
|
||||
return ToAsyncStream(RawSamples, ThrowOnRead, ct);
|
||||
RawCalls.Add((startUtc, endUtc, maxValues));
|
||||
var items = RawSource is null ? RawSamples : RawSource(startUtc, endUtc, maxValues).ToList();
|
||||
return CountingStream(ToAsyncStream(items, ThrowOnRead, ct));
|
||||
}
|
||||
|
||||
public IAsyncEnumerable<HistorianAggregateSample> ReadAggregateAsync(
|
||||
@@ -193,6 +206,16 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
private async IAsyncEnumerable<HistorianSample> CountingStream(IAsyncEnumerable<HistorianSample> inner)
|
||||
{
|
||||
await foreach (var item in inner.ConfigureAwait(false))
|
||||
{
|
||||
RawYieldedCount++;
|
||||
yield return item;
|
||||
}
|
||||
RawStreamsCompleted++;
|
||||
}
|
||||
|
||||
private static async IAsyncEnumerable<T> ToAsyncStream<T>(
|
||||
IReadOnlyList<T> items,
|
||||
Exception? error,
|
||||
|
||||
+37
-3
@@ -32,11 +32,29 @@ 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.Minimum, TestContext.Current.CancellationToken);
|
||||
Assert.Equal(RetrievalMode.MinimumWithTime, fake.LastAggregateMode);
|
||||
await ds.ReadProcessedAsync("T", default, default, TimeSpan.FromSeconds(60), HistoryAggregateType.Total, TestContext.Current.CancellationToken);
|
||||
Assert.Equal(RetrievalMode.Integral, fake.LastAggregateMode);
|
||||
Assert.Equal(TimeSpan.FromSeconds(60), fake.LastAggregateInterval);
|
||||
}
|
||||
|
||||
[Fact] // #681: the native buckets are stamped with the interval START and carry the Calculated bits
|
||||
public async Task ReadProcessed_native_buckets_are_stamped_at_the_interval_start_with_the_Calculated_bit()
|
||||
{
|
||||
var start = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
var fake = new FakeHistorianGatewayClient
|
||||
{
|
||||
AggregateSamples = new[]
|
||||
{
|
||||
new HistorianAggregateSample { Value = 7.0, OpcQuality = 192, StartTime = Timestamp.FromDateTime(start), EndTime = Timestamp.FromDateTime(start.AddMinutes(1)) },
|
||||
new HistorianAggregateSample { OpcQuality = 0, StartTime = Timestamp.FromDateTime(start.AddMinutes(1)), EndTime = Timestamp.FromDateTime(start.AddMinutes(2)) },
|
||||
},
|
||||
};
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
var r = await ds.ReadProcessedAsync("T", start, start.AddMinutes(2), TimeSpan.FromMinutes(1), HistoryAggregateType.TimeAverage, TestContext.Current.CancellationToken);
|
||||
Assert.Equal(new DateTime?[] { start, start.AddMinutes(1) }, r.Samples.Select(x => x.SourceTimestampUtc).ToArray());
|
||||
Assert.Equal(new[] { 0x00000401u, 0x809B0401u }, r.Samples.Select(x => x.StatusCode).ToArray());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReadProcessed_TimeAverage_requests_TimeWeightedAverage()
|
||||
{
|
||||
@@ -46,9 +64,24 @@ public sealed class GatewayHistorianDataSourceTests
|
||||
Assert.Equal(RetrievalMode.TimeWeightedAverage, fake.LastAggregateMode);
|
||||
}
|
||||
|
||||
[Theory] // #679: no AVEVA equivalent — not supported, gateway never called, no backend failure recorded
|
||||
[Theory] // #681 (was #679 BadAggregateNotSupported): served from a raw read, never a native retrieval mode
|
||||
[InlineData(HistoryAggregateType.Average)]
|
||||
[InlineData(HistoryAggregateType.Count)]
|
||||
[InlineData(HistoryAggregateType.Minimum)]
|
||||
[InlineData(HistoryAggregateType.Maximum)]
|
||||
public async Task ReadProcessed_per_sample_aggregate_reads_raw_not_a_retrieval_mode(HistoryAggregateType aggregate)
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient();
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
await ds.ReadProcessedAsync("T", t0, t0.AddMinutes(1), TimeSpan.FromSeconds(60), aggregate, TestContext.Current.CancellationToken);
|
||||
Assert.Equal(0, fake.ReadAggregateCallCount);
|
||||
Assert.True(fake.ReadRawCallCount >= 1); // the window, plus the earlier-data probe when the range matters
|
||||
}
|
||||
|
||||
[Theory] // no AVEVA mode and not a per-sample aggregate: not supported, gateway never called, no failure recorded
|
||||
[InlineData(HistoryAggregateType.TimeAverage2)]
|
||||
[InlineData(HistoryAggregateType.PercentGood)]
|
||||
public async Task ReadProcessed_unsupported_aggregate_throws_without_calling_gateway_or_recording_failure(HistoryAggregateType aggregate)
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient();
|
||||
@@ -56,6 +89,7 @@ public sealed class GatewayHistorianDataSourceTests
|
||||
await Assert.ThrowsAsync<HistoryAggregateNotSupportedException>(() => ds.ReadProcessedAsync(
|
||||
"T", default, default, TimeSpan.FromSeconds(60), aggregate, TestContext.Current.CancellationToken));
|
||||
Assert.Equal(0, fake.ReadAggregateCallCount);
|
||||
Assert.Equal(0, fake.ReadRawCallCount);
|
||||
var h = ds.GetHealthSnapshot();
|
||||
Assert.Equal(0, h.TotalFailures);
|
||||
Assert.Equal(0, h.TotalQueries);
|
||||
|
||||
+568
@@ -0,0 +1,568 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Grpc.Core;
|
||||
using ZB.MOM.WW.HistorianGateway.Client;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// #681: the Gateway backend serves the per-sample aggregates from raw reads with the SQL backends' shared evaluator.
|
||||
/// The parity oracle is what SqlServer/QuestDB answer for the same stored samples: per-bucket statistics built
|
||||
/// independently in memory (LINQ, <see cref="BucketPlan.IndexOf"/>) — the C# statement of the backends' SQL that
|
||||
/// <c>Sql.Common.Tests.SdkAggregateParityTests</c> pins against the OPC UA SDK's own Part 13 calculators — evaluated by
|
||||
/// <see cref="AggregateEvaluator.EvaluateAll"/> with the data range the SQL backends query (earliest sample ≤ the
|
||||
/// window's later edge, latest ≥ its earlier edge, over EVERY stored sample). The fake gateway serves raw reads from
|
||||
/// the same stored samples, optionally emitting AVEVA's initial-value row stamped at the query start.
|
||||
/// </summary>
|
||||
public sealed class GatewayRawAggregateTests
|
||||
{
|
||||
private static readonly CancellationToken Ct = CancellationToken.None;
|
||||
|
||||
/// <summary>2026-10-09 08:00:00.1234567 UTC — not millisecond-aligned (the SQL fixtures' T0).</summary>
|
||||
private static readonly DateTime T0 = new DateTime(2026, 10, 9, 8, 0, 0, DateTimeKind.Utc).AddTicks(1_234_567);
|
||||
|
||||
private static readonly DateTime Now = At(3600);
|
||||
|
||||
private const uint Good = 0x00000000u;
|
||||
private const uint Bad = 0x80000000u;
|
||||
private const uint Uncertain = 0x40000000u;
|
||||
private const uint GoodCalc = 0x00000401u;
|
||||
private const uint UncertainCalc = 0x40A40401u;
|
||||
private const uint NoDataCalc = 0x809B0401u;
|
||||
private const uint InvalidInputsCalc = 0x80D60401u;
|
||||
private const uint Partial = 0x4u;
|
||||
|
||||
private static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond));
|
||||
|
||||
private static StoredSample S(double seconds, HistorianValue v, uint status) => new(At(seconds), v, status);
|
||||
|
||||
private static StoredSample D(double seconds, double v, uint status = Good) => S(seconds, HistorianValue.FromDouble(v), status);
|
||||
|
||||
// ---- datasets (statuses restricted to codes that round-trip the OPC DA quality byte: Good, Uncertain, Bad) ----------
|
||||
|
||||
/// <summary>Tag A of the conformance suite: +0=10, +10=20, +20=Bad(30), +30=40, +40=50.</summary>
|
||||
private static readonly StoredSample[] A = [D(0, 10), D(10, 20), D(20, 30, Bad), D(30, 40), D(40, 50)];
|
||||
|
||||
/// <summary>Every 2.5 s from −10 to +100, values cycling, one Bad sample at +22.5.</summary>
|
||||
private static readonly StoredSample[] Dense = Enumerable.Range(0, 45)
|
||||
.Select(i => (t: -10 + i * 2.5, i))
|
||||
.Select(x => D(x.t, 10 + (x.i * 7 % 13), x.t == 22.5 ? Bad : Good))
|
||||
.ToArray();
|
||||
|
||||
private static readonly StoredSample[] Gappy =
|
||||
[D(1, 10), D(2, 11, Uncertain), D(25, 12, Bad), D(45, 13), D(70, 1)];
|
||||
|
||||
private static readonly StoredSample[] Bracketed =
|
||||
[
|
||||
D(-5, 1), D(1, 10, Bad), D(3, 11), D(5, 12, Bad), D(7, 13, Uncertain), D(9, 15), D(12, 20),
|
||||
S(14, HistorianValue.Empty, Bad), D(16, 21), S(19, HistorianValue.Empty, Bad), D(40, 1),
|
||||
];
|
||||
|
||||
private static readonly StoredSample[] EarlyEnd = [D(-5, 1), D(3, 2), D(25, 3)];
|
||||
private static readonly StoredSample[] EndOnEdge = [D(-5, 1), D(3, 2), D(20, 3)];
|
||||
|
||||
private static readonly StoredSample[] LateStart = Enumerable.Range(0, 20)
|
||||
.Select(i => D(7 + i * 5, i % 2 == 0 ? 20 : 40)).ToArray();
|
||||
|
||||
private static readonly StoredSample[] OutageEnds = [D(-5, 1), D(3, 2), S(25, HistorianValue.Empty, Bad)];
|
||||
|
||||
/// <summary>A usable String sample among numbers (numeric aggregates ⇒ BadAggregateInvalidInputs in its bucket).</summary>
|
||||
private static readonly StoredSample[] Mixed =
|
||||
[D(-5, 1), D(2, 4), S(12, HistorianValue.FromString("x"), Good), D(15, 6), D(33, 9, Uncertain)];
|
||||
|
||||
/// <summary>Equal extremes in one bucket (ActualTime ties ⇒ earliest).</summary>
|
||||
private static readonly StoredSample[] Ties = [D(1, 5), D(3, 9), D(4, 1), D(6, 9), D(8, 1), D(25, 2)];
|
||||
|
||||
/// <summary>#684 (Sql.Common's DenseUncertain): Dense with every 5th sample an in-range Uncertain 16.</summary>
|
||||
private static readonly StoredSample[] DenseUncertain = Enumerable.Range(0, 45)
|
||||
.Select(i => (t: -10 + i * 2.5, i))
|
||||
.Select(x => x.i % 5 == 0 && x.t != 22.5
|
||||
? D(x.t, 16, Uncertain)
|
||||
: D(x.t, 10 + (x.i * 7 % 13), x.t == 22.5 ? Bad : Good))
|
||||
.ToArray();
|
||||
|
||||
/// <summary>#684 (Sql.Common's UncertainBeyond): Uncertain values below and above the Good extremes in one interval.</summary>
|
||||
private static readonly StoredSample[] UncertainBeyond = [D(-5, 20), D(2, 10), D(4, 5, Uncertain), D(6, 99, Uncertain), D(8, 30), D(25, 20)];
|
||||
|
||||
/// <summary>#684: an Uncertain string among Good numbers (does not refuse the Good-value aggregates) and a Good one.</summary>
|
||||
private static readonly StoredSample[] UncertainText =
|
||||
[
|
||||
D(-5, 1), D(2, 4), S(4, HistorianValue.FromString("u"), Uncertain), D(6, 8), S(14, HistorianValue.FromString("g"), Good),
|
||||
D(16, 3),
|
||||
];
|
||||
|
||||
private static readonly (string Name, StoredSample[] Data)[] Datasets =
|
||||
[
|
||||
("A", A), ("Dense", Dense), ("Gappy", Gappy), ("Bracketed", Bracketed), ("EarlyEnd", EarlyEnd),
|
||||
("EndOnEdge", EndOnEdge), ("LateStart", LateStart), ("OutageEnds", OutageEnds), ("Mixed", Mixed), ("Ties", Ties),
|
||||
("DenseUncertain", DenseUncertain), ("UncertainBeyond", UncertainBeyond), ("UncertainText", UncertainText),
|
||||
];
|
||||
|
||||
private static readonly (double From, double To, double Interval)[] Windows =
|
||||
[
|
||||
(0, 80, 20), (80, 0, 20), (0, 50, 20), (50, 0, 20), (5, 80, 15), (80, 5, 15), (3, 77, 20), (77, 3, 20),
|
||||
(0, 60, 0), (60, 0, 0), (0, 60, 10), (60, 0, 10), (-10, 1, 5), (1, -10, 5), (70, 85, 5), (85, 70, 5), (0, 1, 1),
|
||||
(0, 40, 10), (40, 0, 10), (0, 40, 25), (40, 0, 25),
|
||||
];
|
||||
|
||||
private static readonly HistoryAggregateType[] PerSample =
|
||||
[
|
||||
HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum,
|
||||
HistoryAggregateType.Count, HistoryAggregateType.Range, HistoryAggregateType.Start, HistoryAggregateType.End,
|
||||
HistoryAggregateType.Delta, HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime,
|
||||
];
|
||||
|
||||
// ---- the fake store ----------------------------------------------------------------------------------------------
|
||||
|
||||
private static HistorianSample ToWire(StoredSample s)
|
||||
{
|
||||
var w = new HistorianSample
|
||||
{
|
||||
Tag = "T",
|
||||
Timestamp = Timestamp.FromDateTime(s.TimeUtc),
|
||||
OpcQuality = s.Status switch { Good => 192u, Uncertain => 64u, _ => 0u },
|
||||
};
|
||||
// The REAL wire shape (histsdk): every numeric row carries NumericValue — a NULL / no-data row arrives as 0 with
|
||||
// OPC quality 0 — and a string row always carries StringValue.
|
||||
if (s.Value.Kind == HistorianValueKind.String) w.StringValue = s.Value.Text;
|
||||
else w.NumericValue = s.Value.TryGetDouble(out var d) ? d : 0d;
|
||||
return w;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A gateway over a stored sample set: ReadRaw returns the samples in <c>[start, end]</c> ascending, capped at
|
||||
/// <c>maxValues</c>. With <paramref name="avevaBoundary"/> it first emits what an AVEVA <c>Full</c> read does at
|
||||
/// the query start when nothing is stored exactly there: a row stamped AT the query start carrying the value in
|
||||
/// effect (the last stored sample before it, with its quality; <c>Quality = 133</c>) — or, with no earlier data, a
|
||||
/// NULL row in the real wire shape: <c>NumericValue = 0</c> present, <c>OpcQuality = 0</c>. With
|
||||
/// <paramref name="terminalRow"/> the small (2-row) reads also end with a NULL row stamped at the query END (a
|
||||
/// hypothetical shape the probe / look-ahead must ignore; the window read relies on there being none — live leg).
|
||||
/// </summary>
|
||||
private static FakeHistorianGatewayClient Store(IReadOnlyList<StoredSample> data, bool avevaBoundary = false,
|
||||
bool terminalRow = false)
|
||||
{
|
||||
var sorted = data.OrderBy(s => s.TimeUtc).ToList();
|
||||
return new FakeHistorianGatewayClient
|
||||
{
|
||||
RawSource = (start, end, max) =>
|
||||
{
|
||||
var rows = new List<HistorianSample>();
|
||||
if (avevaBoundary && sorted.All(s => s.TimeUtc != start))
|
||||
{
|
||||
var prior = sorted.LastOrDefault(s => s.TimeUtc < start);
|
||||
var boundary = prior.TimeUtc == default
|
||||
? new HistorianSample { Tag = "T", NumericValue = 0, OpcQuality = 0, Quality = 1 }
|
||||
: ToWire(prior);
|
||||
boundary.Timestamp = Timestamp.FromDateTime(start);
|
||||
if (prior.TimeUtc != default) boundary.Quality = 133;
|
||||
rows.Add(boundary);
|
||||
}
|
||||
rows.AddRange(sorted.Where(s => s.TimeUtc >= start && s.TimeUtc <= end).Select(ToWire));
|
||||
if (terminalRow && max <= 2 && end > start && sorted.All(s => s.TimeUtc != end))
|
||||
rows.Add(new HistorianSample
|
||||
{ Tag = "T", Timestamp = Timestamp.FromDateTime(end), NumericValue = 0, OpcQuality = 0, Quality = 1 });
|
||||
return rows.Take(max);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private sealed class FixedClock(DateTime now) : TimeProvider
|
||||
{
|
||||
public override DateTimeOffset GetUtcNow() => new(now);
|
||||
}
|
||||
|
||||
private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient client, int maxRawRows = 1_000_000) =>
|
||||
new(client, NullLogger<GatewayHistorianDataSource>.Instance, maxRawRows, new FixedClock(Now));
|
||||
|
||||
// ---- the oracle: what the SQL backends answer --------------------------------------------------------------------
|
||||
|
||||
/// <summary>The SQL backends' answer for <paramref name="data"/> (statuses as the gateway quality byte carries them).</summary>
|
||||
private static IReadOnlyList<DataValueSnapshot> SqlBackendAnswer(HistoryAggregateType agg, BucketPlan plan,
|
||||
IReadOnlyList<StoredSample> data)
|
||||
{
|
||||
// Same clamp the backends apply (an Empty sample is Bad).
|
||||
var stored = data.Select(s => StoredSample.FromStored(s.TimeUtc, s.Value, s.Status)).ToList();
|
||||
var earliest = stored.Where(s => s.TimeUtc <= plan.EndUtc).Select(s => (DateTime?)s.TimeUtc).Min();
|
||||
var latest = stored.Where(s => s.TimeUtc >= plan.StartUtc).Select(s => (DateTime?)s.TimeUtc).Max();
|
||||
return AggregateEvaluator.EvaluateAll(agg, plan, InMemoryRows(plan, stored, agg == HistoryAggregateType.MaximumActualTime),
|
||||
earliest, latest, Now);
|
||||
}
|
||||
|
||||
/// <summary>Independent LINQ statement of the per-bucket statistics (copied from Sql.Common.Tests, #684 semantics).</summary>
|
||||
private static IEnumerable<BucketRow> InMemoryRows(BucketPlan plan, IEnumerable<StoredSample> samples, bool maximum)
|
||||
{
|
||||
foreach (var g in samples.Select(s => (s, b: plan.IndexOf(s.TimeUtc))).Where(x => x.b is not null)
|
||||
.GroupBy(x => x.b!.Value).OrderBy(g => g.Key))
|
||||
{
|
||||
var all = g.Select(x => x.s).OrderBy(s => s.TimeUtc).ToList();
|
||||
var usable = all.Where(s => s.IsUsable).ToList();
|
||||
var numeric = usable.Where(s => s.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64
|
||||
or HistorianValueKind.Boolean).ToList();
|
||||
var good = numeric.Where(s => s.IsGood).ToList();
|
||||
var uncertain = numeric.Where(s => !s.IsGood).ToList();
|
||||
var extreme = good.OrderBy(s => maximum ? -s.Value.Number : s.Value.Number).ThenBy(s => s.TimeUtc)
|
||||
.Cast<StoredSample?>().FirstOrDefault();
|
||||
yield return new BucketRow
|
||||
{
|
||||
Bucket = g.Key,
|
||||
Total = all.Count,
|
||||
Usable = usable.Count,
|
||||
Good = all.Count(s => s.IsGood),
|
||||
Bad = all.Count - usable.Count,
|
||||
NonNumeric = usable.Count - numeric.Count,
|
||||
GoodNonNumeric = usable.Count(s => s.IsGood) - numeric.Count(s => s.IsGood),
|
||||
Avg = good.Count == 0 ? null : good.Average(s => s.Value.Number),
|
||||
Min = good.Count == 0 ? null : good.Min(s => s.Value.Number),
|
||||
Max = good.Count == 0 ? null : good.Max(s => s.Value.Number),
|
||||
Sum = good.Count == 0 ? null : good.Sum(s => s.Value.Number),
|
||||
UncertainMin = uncertain.Count == 0 ? null : uncertain.Min(s => s.Value.Number),
|
||||
UncertainMax = uncertain.Count == 0 ? null : uncertain.Max(s => s.Value.Number),
|
||||
First = usable.Count == 0 ? HistorianValue.Empty : usable[0].Value,
|
||||
Last = usable.Count == 0 ? HistorianValue.Empty : usable[^1].Value,
|
||||
Extreme = extreme?.Value.Number,
|
||||
ExtremeTime = extreme?.TimeUtc,
|
||||
FirstAny = all[0],
|
||||
LastAny = all[^1],
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
private static string Describe(IReadOnlyList<DataValueSnapshot> v) =>
|
||||
string.Join(" | ", v.Select(x => $"{x.Value ?? "null"} 0x{x.StatusCode:X8} {(x.SourceTimestampUtc!.Value - T0).TotalSeconds}s"));
|
||||
|
||||
// ---- parity ------------------------------------------------------------------------------------------------------
|
||||
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task Every_per_sample_aggregate_matches_the_SQL_backends_forward_and_reverse(bool avevaBoundary)
|
||||
{
|
||||
var failures = new List<string>();
|
||||
var compared = 0;
|
||||
foreach (var agg in PerSample)
|
||||
foreach (var (name, data) in Datasets)
|
||||
foreach (var (from, to, interval) in Windows)
|
||||
{
|
||||
var plan = BucketPlan.Create(At(from), At(to), TimeSpan.FromSeconds(interval));
|
||||
var expected = SqlBackendAnswer(agg, plan, data);
|
||||
var fake = Store(data, avevaBoundary);
|
||||
var actual = (await Source(fake)
|
||||
.ReadProcessedAsync("T", At(from), At(to), TimeSpan.FromSeconds(interval), agg, Ct)).Samples;
|
||||
compared++;
|
||||
if (fake.RawStreamsCompleted != fake.ReadRawCallCount)
|
||||
failures.Add($"{agg} {name} {from}→{to}/{interval}s: a raw stream was abandoned mid-way");
|
||||
var same = actual.Count == expected.Count && actual.Zip(expected).All(p =>
|
||||
Equals(p.First.Value, p.Second.Value) && p.First.StatusCode == p.Second.StatusCode
|
||||
&& p.First.SourceTimestampUtc == p.Second.SourceTimestampUtc);
|
||||
if (!same)
|
||||
failures.Add($"{agg} {name} {from}→{to}/{interval}s\n ours {Describe(actual)}\n sql {Describe(expected)}");
|
||||
}
|
||||
compared.ShouldBe(PerSample.Length * Datasets.Length * Windows.Length);
|
||||
failures.ShouldBeEmpty(string.Join("\n", failures));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(HistoryAggregateType.Average, 15d, 40d, 50d)]
|
||||
[InlineData(HistoryAggregateType.Minimum, 10d, 40d, 50d)]
|
||||
[InlineData(HistoryAggregateType.Maximum, 20d, 40d, 50d)]
|
||||
[InlineData(HistoryAggregateType.Delta, 10d, 0d, 0d)]
|
||||
[InlineData(HistoryAggregateType.Range, 10d, 0d, 0d)]
|
||||
public async Task Numeric_aggregates_match_the_conformance_suites_hand_computed_literals(HistoryAggregateType agg,
|
||||
double b0, double b1, double b2)
|
||||
{
|
||||
// The literals of ProcessedConformance.Numeric_aggregates_over_the_fixture (SqlServer + QuestDB), over tag A.
|
||||
var s = (await Source(Store(A, avevaBoundary: true))
|
||||
.ReadProcessedAsync("A", At(0), At(80), TimeSpan.FromSeconds(20), agg, Ct)).Samples;
|
||||
s.Select(x => x.Value).ShouldBe([b0, b1, b2, null]);
|
||||
var b2Status = agg == HistoryAggregateType.Average ? GoodCalc : GoodCalc | Partial;
|
||||
s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, b2Status, NoDataCalc]);
|
||||
s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40), At(60)]); // interval STARTS (Part 13)
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Count_is_a_Good_sample_count_as_int_and_zero_inside_the_data_range()
|
||||
{
|
||||
// #679 made Count BadAggregateNotSupported on Gateway; #681 serves it with the SQL backends' #680 semantics.
|
||||
var s = (await Source(Store(Gappy)).ReadProcessedAsync("T", At(0), At(60), TimeSpan.FromSeconds(10),
|
||||
HistoryAggregateType.Count, Ct)).Samples;
|
||||
// [0,10): +1 G, +2 U ⇒ 1, Uncertain (data starts inside ⇒ Partial); [10,20) empty inside the range ⇒ 0 Good;
|
||||
// [20,30): +25 Bad only ⇒ generic Bad; [30,40) 0; [40,50) +45 ⇒ 1; [50,60) 0 (+70 is later data).
|
||||
s.Select(x => x.Value).ShouldBe([1, 0, null, 0, 1, 0]);
|
||||
s.Select(x => x.StatusCode).ShouldBe([UncertainCalc | Partial, GoodCalc, 0x80000401u, GoodCalc, GoodCalc, GoodCalc]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_usable_string_sample_makes_a_numeric_aggregate_BadAggregateInvalidInputs()
|
||||
{
|
||||
var s = (await Source(Store(Mixed)).ReadProcessedAsync("T", At(0), At(40), TimeSpan.FromSeconds(10),
|
||||
HistoryAggregateType.Average, Ct)).Samples;
|
||||
s[1].StatusCode.ShouldBe(InvalidInputsCalc);
|
||||
s[1].Value.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ActualTime_carries_the_extreme_samples_own_time_ties_earliest()
|
||||
{
|
||||
var max = (await Source(Store(Ties)).ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero,
|
||||
HistoryAggregateType.MaximumActualTime, Ct)).Samples.Single();
|
||||
max.Value.ShouldBe(9d);
|
||||
max.SourceTimestampUtc.ShouldBe(At(3));
|
||||
var min = (await Source(Store(Ties)).ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero,
|
||||
HistoryAggregateType.MinimumActualTime, Ct)).Samples.Single();
|
||||
min.Value.ShouldBe(1d);
|
||||
min.SourceTimestampUtc.ShouldBe(At(4));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_reverse_read_is_anchored_at_its_later_time_and_returned_latest_first()
|
||||
{
|
||||
// #677: 50 → 0 at 20 s ⇒ (30,50], (10,30], (0,10] stamped 50, 30, 10.
|
||||
var s = (await Source(Store(A)).ReadProcessedAsync("A", At(50), At(0), TimeSpan.FromSeconds(20),
|
||||
HistoryAggregateType.Maximum, Ct)).Samples;
|
||||
s.Select(x => x.SourceTimestampUtc).ShouldBe([At(50), At(30), At(10)]);
|
||||
s.Select(x => x.Value).ShouldBe([50d, 40d, 20d]);
|
||||
}
|
||||
|
||||
// ---- the raw request ---------------------------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task The_raw_stream_starts_one_millisecond_before_the_window_so_the_AVEVA_boundary_row_is_never_a_sample()
|
||||
{
|
||||
// Data before the window, nothing inside [0,10) but +5: the boundary row (value 1, stamped at the query start) must
|
||||
// not be counted. Count over [0,10) is 1, not 2.
|
||||
var fake = Store([D(-5, 1), D(5, 2)], avevaBoundary: true);
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct))
|
||||
.Samples.Single();
|
||||
s.Value.ShouldBe(1);
|
||||
fake.RawCalls[0].Start.ShouldBe(At(0) - TimeSpan.FromMilliseconds(1));
|
||||
fake.RawCalls.ShouldNotContain(c => c.Start == DateTime.UnixEpoch); // the Good boundary row proved earlier data: no probe
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Without_boundary_evidence_a_probe_asks_for_data_before_the_window()
|
||||
{
|
||||
// No boundary row (a historian that emits none): the second call probes [epoch, start − 1 tick] for 2 rows, and the
|
||||
// sample at −5 it finds keeps the first interval free of the start-of-data Partial bit.
|
||||
var fake = Store([D(-5, 1), D(5, 2), D(15, 3)]);
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(0), At(20), TimeSpan.FromSeconds(10),
|
||||
HistoryAggregateType.Minimum, Ct)).Samples;
|
||||
fake.RawCalls[1].ShouldBe((DateTime.UnixEpoch, At(0).AddTicks(-1), 2));
|
||||
(s[0].StatusCode & Partial).ShouldBe(0u);
|
||||
|
||||
// Data that really starts inside the first interval: the probe finds nothing, the interval is Partial.
|
||||
var late = Store([D(5, 2), D(15, 3)], avevaBoundary: true);
|
||||
var l = (await Source(late).ReadProcessedAsync("T", At(0), At(20), TimeSpan.FromSeconds(10),
|
||||
HistoryAggregateType.Minimum, Ct)).Samples;
|
||||
late.RawCalls[1].Start.ShouldBe(DateTime.UnixEpoch); // the NULL boundary row (value 0, OPC quality 0) is no evidence
|
||||
(l[0].StatusCode & Partial).ShouldBe(Partial);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_raw_request_looks_ahead_past_the_window_only_when_the_data_range_matters()
|
||||
{
|
||||
var count = Store(A);
|
||||
await Source(count).ReadProcessedAsync("A", At(0), At(80), TimeSpan.FromSeconds(20), HistoryAggregateType.Count, Ct);
|
||||
// The window read ends AT the window end; probe + look-ahead are separate 2-row reads, all consumed to completion.
|
||||
count.RawCalls.ShouldBe([
|
||||
(At(0) - TimeSpan.FromMilliseconds(1), At(80), 1_000_000 + 2 + 16),
|
||||
(DateTime.UnixEpoch, At(0).AddTicks(-1), 2),
|
||||
(At(80), Now + TimeSpan.FromDays(1), 2),
|
||||
]);
|
||||
count.RawStreamsCompleted.ShouldBe(3);
|
||||
|
||||
var average = Store(A);
|
||||
await Source(average).ReadProcessedAsync("A", At(0), At(80), TimeSpan.FromSeconds(20), HistoryAggregateType.Average, Ct);
|
||||
average.RawCalls.Single().End.ShouldBe(At(80)); // Average needs no data range: no look-ahead, no probe
|
||||
average.ReadAggregateCallCount.ShouldBe(0);
|
||||
|
||||
var unbounded = Store(A);
|
||||
await Source(unbounded, maxRawRows: 0).ReadProcessedAsync("A", At(0), At(80), TimeSpan.FromSeconds(20),
|
||||
HistoryAggregateType.Count, Ct);
|
||||
unbounded.RawCalls[0].MaxValues.ShouldBe(int.MaxValue);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task No_gateway_stream_is_ever_abandoned_and_later_data_comes_from_a_two_row_look_ahead()
|
||||
{
|
||||
// #681 review: breaking out of a gateway stream cancels the call, which the gateway treats as a connection fault
|
||||
// (it evicts its shared historian session). 1000 samples after the window: the window read never sees them, the
|
||||
// 2-row look-ahead reads two, and every stream runs to its end.
|
||||
var data = Enumerable.Range(0, 1000).Select(i => D(100 + i, i)).Prepend(D(-5, 0)).Prepend(D(5, 1)).ToArray();
|
||||
var fake = Store(data, avevaBoundary: true);
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(0), At(50), TimeSpan.FromSeconds(10),
|
||||
HistoryAggregateType.Minimum, Ct)).Samples;
|
||||
fake.RawStreamsCompleted.ShouldBe(fake.ReadRawCallCount);
|
||||
fake.RawCalls.Last().ShouldBe((At(50), Now + TimeSpan.FromDays(1), 2));
|
||||
fake.RawYieldedCount.ShouldBe(2 + 2); // window read: boundary + +5; look-ahead: boundary at +50 + first sample
|
||||
(s[0].StatusCode & Partial).ShouldBe(0u); // data continues after the window: no end-of-data Partial
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Rows_stamped_at_a_small_reads_query_start_or_end_are_not_samples()
|
||||
{
|
||||
// Data ends at +15: the look-ahead's NULL rows at its own start (+20, boundary) and end (now + 1 day, terminal)
|
||||
// must not count as later data, so (10,20] / [10,20) holds the end of the data ⇒ Partial.
|
||||
var fake = Store([D(-5, 1), D(5, 2), D(15, 3)], avevaBoundary: true, terminalRow: true);
|
||||
var s = (await Source(fake).ReadProcessedAsync("T", At(0), At(20), TimeSpan.FromSeconds(10),
|
||||
HistoryAggregateType.Maximum, Ct)).Samples;
|
||||
(s[1].StatusCode & Partial).ShouldBe(Partial);
|
||||
fake.RawCalls.Last().Start.ShouldBe(At(20));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_window_over_the_raw_row_cap_fails_the_node_without_counting_a_backend_failure()
|
||||
{
|
||||
var data = Enumerable.Range(0, 10).Select(i => D(i, i)).ToArray(); // 10 rows in [0,10)
|
||||
var ds = Source(Store(data), maxRawRows: 9);
|
||||
var ex = await Should.ThrowAsync<HistoryReadLimitExceededException>(() =>
|
||||
ds.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Average, Ct));
|
||||
ex.Limit.ShouldBe(9);
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
|
||||
// Exactly at the cap is served.
|
||||
var ok = await Source(Store(data), maxRawRows: 10)
|
||||
.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Average, Ct);
|
||||
ok.Samples.Single().Value.ShouldBe(4.5);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_stream_that_exhausts_its_row_budget_before_the_window_is_never_answered_from_a_truncated_window()
|
||||
{
|
||||
// 40 rows inside the 1 ms margin before the window eat the whole budget (cap 5 ⇒ 23 rows): the window could not be
|
||||
// read, so the node fails rather than reporting BadNoData.
|
||||
var margin = Enumerable.Range(0, 40).Select(i => new StoredSample(At(0).AddTicks(-1 - i), HistorianValue.FromDouble(i), Good));
|
||||
var data = margin.Append(D(5, 1)).ToArray();
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => Source(Store(data), maxRawRows: 5)
|
||||
.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Past_the_cap_the_window_stream_is_drained_not_abandoned()
|
||||
{
|
||||
var data = Enumerable.Range(0, 10).Select(i => D(i, i)).ToArray();
|
||||
var fake = Store(data);
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => Source(fake, maxRawRows: 3)
|
||||
.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Average, Ct));
|
||||
fake.RawStreamsCompleted.ShouldBe(1);
|
||||
fake.RawYieldedCount.ShouldBe(10);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_gateways_max_values_ceiling_rejection_is_a_request_limit_not_a_backend_fault()
|
||||
{
|
||||
// Historian:ReadLimits:MaxValuesCeiling (1,000,000) with AllowStreamingBeyondCeiling=false refuses the default
|
||||
// cap + 18 budget with InvalidArgument; the client package wraps it in HistorianGatewayException.
|
||||
var rpc = new RpcException(new Status(StatusCode.InvalidArgument,
|
||||
"max_values exceeds the server ceiling of 1000000; read large ranges by time-slicing ..."));
|
||||
var fake = new FakeHistorianGatewayClient { ThrowOnRead = new HistorianGatewayException(rpc.Status.Detail, rpc) };
|
||||
var ds = Source(fake);
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() =>
|
||||
ds.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct));
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
|
||||
// Any other InvalidArgument is still a backend fault.
|
||||
var other = new RpcException(new Status(StatusCode.InvalidArgument, "start_time must be less than or equal to end_time."));
|
||||
var ds2 = Source(new FakeHistorianGatewayClient { ThrowOnRead = new HistorianGatewayException(other.Status.Detail, other) });
|
||||
await Should.ThrowAsync<HistorianGatewayException>(() =>
|
||||
ds2.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct));
|
||||
ds2.GetHealthSnapshot().TotalFailures.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_caller_cancellation_is_not_counted_as_a_backend_failure()
|
||||
{
|
||||
using var cts = new CancellationTokenSource();
|
||||
await cts.CancelAsync();
|
||||
var ds = Source(new FakeHistorianGatewayClient { ThrowOnRead = new OperationCanceledException(cts.Token) });
|
||||
await Should.ThrowAsync<OperationCanceledException>(() =>
|
||||
ds.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, cts.Token));
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
|
||||
// A cancellation the caller did NOT ask for (e.g. the gateway cancelled the call) is still a backend fault.
|
||||
var ds2 = Source(new FakeHistorianGatewayClient { ThrowOnRead = new OperationCanceledException() });
|
||||
await Should.ThrowAsync<OperationCanceledException>(() =>
|
||||
ds2.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct));
|
||||
ds2.GetHealthSnapshot().TotalFailures.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_gateway_fault_on_the_raw_read_is_recorded_and_rethrown()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { ThrowOnRead = new InvalidOperationException("down") };
|
||||
var ds = Source(fake);
|
||||
await Should.ThrowAsync<InvalidOperationException>(() =>
|
||||
ds.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct));
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_empty_window_calls_nothing_and_returns_no_intervals()
|
||||
{
|
||||
var fake = Store(A);
|
||||
var r = await Source(fake).ReadProcessedAsync("A", At(10), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct);
|
||||
r.Samples.ShouldBeEmpty();
|
||||
fake.ReadRawCallCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
// ---- the served set ----------------------------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void The_served_set_is_the_per_sample_families_plus_the_two_native_modes()
|
||||
{
|
||||
GatewayHistorianDataSource.ServedAggregates.OrderBy(a => a).ShouldBe(
|
||||
PerSample.Append(HistoryAggregateType.TimeAverage).Append(HistoryAggregateType.Total).OrderBy(a => a));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Every_member_outside_the_served_set_is_refused_without_calling_the_gateway_and_every_member_inside_is_served()
|
||||
{
|
||||
foreach (var agg in System.Enum.GetValues<HistoryAggregateType>())
|
||||
{
|
||||
var fake = Store(A);
|
||||
var ds = Source(fake);
|
||||
if (GatewayHistorianDataSource.ServedAggregates.Contains(agg))
|
||||
{
|
||||
await ds.ReadProcessedAsync("A", At(0), At(40), TimeSpan.FromSeconds(20), agg, Ct);
|
||||
(fake.ReadRawCallCount + fake.ReadAggregateCallCount).ShouldBeGreaterThan(0, agg.ToString());
|
||||
}
|
||||
else
|
||||
{
|
||||
await Should.ThrowAsync<HistoryAggregateNotSupportedException>(() =>
|
||||
ds.ReadProcessedAsync("A", At(0), At(40), TimeSpan.FromSeconds(20), agg, Ct));
|
||||
(fake.ReadRawCallCount + fake.ReadAggregateCallCount).ShouldBe(0, agg.ToString());
|
||||
ds.GetHealthSnapshot().TotalQueries.ShouldBe(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- the accumulator against the independent oracle, sample order shuffled ---------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void The_bucket_statistics_do_not_depend_on_arrival_order()
|
||||
{
|
||||
var plan = BucketPlan.Create(At(-10), At(100), TimeSpan.FromSeconds(10));
|
||||
foreach (var maximum in new[] { false, true })
|
||||
{
|
||||
var acc = new RawBucketAccumulator(maximum);
|
||||
foreach (var s in DenseUncertain.Reverse())
|
||||
if (plan.IndexOf(s.TimeUtc) is { } b) acc.Add(b, s);
|
||||
var expected = InMemoryRows(plan, DenseUncertain, maximum).ToList();
|
||||
var actual = acc.Rows().ToList();
|
||||
actual.Count.ShouldBe(expected.Count);
|
||||
foreach (var (a, e) in actual.Zip(expected))
|
||||
{
|
||||
(a.Bucket, a.Total, a.Usable, a.Good, a.Bad, a.NonNumeric).ShouldBe((e.Bucket, e.Total, e.Usable, e.Good, e.Bad, e.NonNumeric));
|
||||
(a.Min, a.Max, a.Extreme, a.ExtremeTime).ShouldBe((e.Min, e.Max, e.Extreme, e.ExtremeTime));
|
||||
(a.UncertainMin, a.UncertainMax, a.GoodNonNumeric).ShouldBe((e.UncertainMin, e.UncertainMax, e.GoodNonNumeric));
|
||||
(a.Avg is null).ShouldBe(e.Avg is null);
|
||||
if (a.Avg is { } avg) avg.ShouldBe(e.Avg!.Value, 1e-9);
|
||||
(a.First, a.Last, a.FirstAny, a.LastAny).ShouldBe((e.First, e.Last, e.FirstAny, e.LastAny));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+19
-17
@@ -9,8 +9,6 @@ public sealed class AggregateModeMapperTests
|
||||
{
|
||||
[Theory]
|
||||
[InlineData(HistoryAggregateType.TimeAverage, RetrievalMode.TimeWeightedAverage)]
|
||||
[InlineData(HistoryAggregateType.Minimum, RetrievalMode.MinimumWithTime)]
|
||||
[InlineData(HistoryAggregateType.Maximum, RetrievalMode.MaximumWithTime)]
|
||||
[InlineData(HistoryAggregateType.Total, RetrievalMode.Integral)]
|
||||
public void Maps_each_aggregate(HistoryAggregateType a, RetrievalMode expected)
|
||||
=> Assert.Equal(expected, AggregateModeMapper.ToRetrievalMode(a));
|
||||
@@ -18,11 +16,8 @@ public sealed class AggregateModeMapperTests
|
||||
[Fact] // matrix guard: every member is either mapped or explicitly reported as not supported
|
||||
public void Every_aggregate_member_is_mapped_or_reports_not_supported()
|
||||
{
|
||||
var original = new[]
|
||||
{
|
||||
HistoryAggregateType.Minimum, HistoryAggregateType.Maximum,
|
||||
HistoryAggregateType.Total, HistoryAggregateType.TimeAverage,
|
||||
};
|
||||
var original = new[] { HistoryAggregateType.Total, HistoryAggregateType.TimeAverage };
|
||||
Assert.Equal(original.OrderBy(a => a), AggregateModeMapper.NativeAggregates.OrderBy(a => a));
|
||||
foreach (var a in Enum.GetValues<HistoryAggregateType>())
|
||||
{
|
||||
if (original.Contains(a))
|
||||
@@ -32,26 +27,33 @@ public sealed class AggregateModeMapperTests
|
||||
}
|
||||
}
|
||||
|
||||
[Theory] // #679: AVEVA has no arithmetic mean and no sample count (Counter is a rollover increase)
|
||||
[Theory] // #679: AVEVA has no arithmetic mean and no sample count; #681: the per-sample aggregates are raw-evaluated, not native
|
||||
[InlineData(HistoryAggregateType.Average)]
|
||||
[InlineData(HistoryAggregateType.Count)]
|
||||
public void Average_and_Count_are_not_supported(HistoryAggregateType a)
|
||||
[InlineData(HistoryAggregateType.Minimum)]
|
||||
[InlineData(HistoryAggregateType.Maximum)]
|
||||
public void Per_sample_aggregates_have_no_native_mode(HistoryAggregateType a)
|
||||
=> Assert.Throws<HistoryAggregateNotSupportedException>(() => AggregateModeMapper.ToRetrievalMode(a));
|
||||
|
||||
[Fact] // #682: the served (= advertised) set is exactly what the mapper maps, and the data source returns it
|
||||
[Fact] // #682/#681: the served (= advertised) set is the native members plus the raw-evaluated per-sample families,
|
||||
// and the data source returns exactly that static
|
||||
public void Served_is_exactly_the_mapped_members_and_the_data_source_declares_it()
|
||||
{
|
||||
Assert.Equal(
|
||||
new[] { HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Total, HistoryAggregateType.TimeAverage }
|
||||
.OrderBy(a => a),
|
||||
AggregateModeMapper.Served.OrderBy(a => a));
|
||||
foreach (var a in Enum.GetValues<HistoryAggregateType>().Where(a => !AggregateModeMapper.Served.Contains(a)))
|
||||
Assert.Throws<HistoryAggregateNotSupportedException>(() => AggregateModeMapper.ToRetrievalMode(a));
|
||||
foreach (var a in Enum.GetValues<HistoryAggregateType>())
|
||||
{
|
||||
var native = AggregateModeMapper.NativeAggregates.Contains(a);
|
||||
var perSample = ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.AggregateFamilies.FamilyOf(a) is
|
||||
ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.AggregateFamily.Simple
|
||||
or ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.AggregateFamily.StartEnd
|
||||
or ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.AggregateFamily.ActualTime;
|
||||
Assert.Equal(native || perSample, GatewayHistorianDataSource.ServedAggregates.Contains(a));
|
||||
if (!native) Assert.Throws<HistoryAggregateNotSupportedException>(() => AggregateModeMapper.ToRetrievalMode(a));
|
||||
}
|
||||
Assert.Equal(12, GatewayHistorianDataSource.ServedAggregates.Count);
|
||||
|
||||
var source = new GatewayHistorianDataSource(
|
||||
new FakeHistorianGatewayClient(), Microsoft.Extensions.Logging.Abstractions.NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
Assert.Same(GatewayHistorianDataSource.ServedAggregates, source.SupportedAggregates);
|
||||
Assert.Same(AggregateModeMapper.Served, GatewayHistorianDataSource.ServedAggregates);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
||||
+12
-2
@@ -37,7 +37,7 @@ public sealed class SampleMapperTests
|
||||
{
|
||||
var a = new HistorianAggregateSample { Tag = "T", /* Value unset, no Good quality */ EndTime = Ts(2026, 1, 1, 0, 0, 0) };
|
||||
var snap = SampleMapper.ToAggregateSnapshot(a);
|
||||
Assert.Equal(0x809B0000u, snap.StatusCode); // BadNoData
|
||||
Assert.Equal(0x809B0401u, snap.StatusCode); // BadNoData + Calculated (#681)
|
||||
Assert.Null(snap.Value);
|
||||
}
|
||||
|
||||
@@ -46,10 +46,20 @@ public sealed class SampleMapperTests
|
||||
{
|
||||
var a = new HistorianAggregateSample { Tag = "T", Value = 42.0, OpcQuality = 192, EndTime = Ts(2026, 1, 1, 0, 0, 0) };
|
||||
var snap = SampleMapper.ToAggregateSnapshot(a);
|
||||
Assert.Equal(0x00000000u, snap.StatusCode); // Good
|
||||
Assert.Equal(0x00000401u, snap.StatusCode); // Good + Calculated (#681)
|
||||
Assert.Equal(42.0, Assert.IsType<double>(snap.Value));
|
||||
}
|
||||
|
||||
[Fact] // #681: Part 13 stamps a processed value with its interval START (it used to be the end)
|
||||
public void Aggregate_bucket_is_stamped_with_its_interval_start()
|
||||
{
|
||||
var a = new HistorianAggregateSample
|
||||
{
|
||||
Tag = "T", Value = 1.0, OpcQuality = 192, StartTime = Ts(2026, 1, 1, 0, 0, 0), EndTime = Ts(2026, 1, 1, 0, 1, 0),
|
||||
};
|
||||
Assert.Equal(new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc), SampleMapper.ToAggregateSnapshot(a).SourceTimestampUtc);
|
||||
}
|
||||
|
||||
// Ts(...) builds a Google.Protobuf.WellKnownTypes.Timestamp from UTC parts.
|
||||
private static Timestamp Ts(int y, int mo, int d, int h, int mi, int s)
|
||||
=> Timestamp.FromDateTime(new DateTime(y, mo, d, h, mi, s, DateTimeKind.Utc));
|
||||
|
||||
+7
-1
@@ -149,8 +149,14 @@ public sealed class HistorianBackendRegistryTests
|
||||
IHistorianDataSource Source(ServerHistorianOptions o, IConfiguration c) =>
|
||||
HistorianBackendRegistry.Resolve(o, c).CreateDataSource(sp);
|
||||
|
||||
// #681: the ten per-sample aggregates (raw-evaluated) + the two AVEVA-native ones.
|
||||
Source(Gateway(), Cfg).SupportedAggregates.ShouldBe(
|
||||
[HistoryAggregateType.TimeAverage, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Total],
|
||||
[
|
||||
HistoryAggregateType.Average, HistoryAggregateType.Count, HistoryAggregateType.Minimum,
|
||||
HistoryAggregateType.Maximum, HistoryAggregateType.Range, HistoryAggregateType.Start, HistoryAggregateType.End,
|
||||
HistoryAggregateType.Delta, HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime,
|
||||
HistoryAggregateType.TimeAverage, HistoryAggregateType.Total,
|
||||
],
|
||||
ignoreOrder: true);
|
||||
Source(new ServerHistorianOptions { Backend = "SqlServer", Enabled = true }, SqlCfg())
|
||||
.SupportedAggregates.ShouldBe(SqlServed, ignoreOrder: true);
|
||||
|
||||
+11
-6
@@ -20,7 +20,12 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
/// </summary>
|
||||
public sealed class AdvertisedAggregateFunctionsWireTests
|
||||
{
|
||||
private static readonly string[] GatewayNames = ["Maximum", "Minimum", "TimeAverage", "Total"];
|
||||
// #681: the ten per-sample aggregates (raw-evaluated) + TimeAverage / Total (AVEVA-native).
|
||||
private static readonly string[] GatewayNames =
|
||||
[
|
||||
"Average", "Count", "Delta", "End", "Maximum", "MaximumActualTime", "Minimum", "MinimumActualTime", "Range", "Start",
|
||||
"TimeAverage", "Total",
|
||||
];
|
||||
|
||||
private static readonly string[] SqlNames =
|
||||
[
|
||||
@@ -104,16 +109,16 @@ public sealed class AdvertisedAggregateFunctionsWireTests
|
||||
|
||||
rig.Nm.HistorianDataSource = Backend("SqlServer");
|
||||
(await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(SqlNames);
|
||||
(await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBe(
|
||||
(await ParentsAsync(session, ObjectIds.AggregateFunction_TimeAverage2, ct)).ShouldBe(
|
||||
[ObjectIds.Server_ServerCapabilities_AggregateFunctions, ObjectIds.HistoryServerCapabilities_AggregateFunctions],
|
||||
ignoreOrder: true);
|
||||
|
||||
// Gateway has no arithmetic Average: dropped from the history folder (inverse reference too), kept in the
|
||||
// server folder.
|
||||
// Gateway has no TimeAverage2 (#681: no prior-sample read for Part 13 bounds): dropped from the history folder
|
||||
// (inverse reference too), kept in the server folder.
|
||||
rig.Nm.HistorianDataSource = Backend("Gateway");
|
||||
(await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(GatewayNames);
|
||||
(await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames);
|
||||
(await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBe(
|
||||
(await ParentsAsync(session, ObjectIds.AggregateFunction_TimeAverage2, ct)).ShouldBe(
|
||||
[ObjectIds.Server_ServerCapabilities_AggregateFunctions]);
|
||||
|
||||
// And back: removed references are restored.
|
||||
@@ -125,7 +130,7 @@ public sealed class AdvertisedAggregateFunctionsWireTests
|
||||
|
||||
[Theory]
|
||||
[InlineData("StandardDeviationSample")] // served by no historian backend, never in the history folder
|
||||
[InlineData("Average")] // not served by the Gateway backend wired below
|
||||
[InlineData("TimeAverage2")] // not served by the Gateway backend wired below (#681)
|
||||
public async Task A_live_AggregateFilter_subscription_for_an_unadvertised_aggregate_still_works(string aggregateName)
|
||||
{
|
||||
// The trim touches only the advertised folders, not the AggregateManager's calculator factories, which is what
|
||||
|
||||
Reference in New Issue
Block a user