diff --git a/CLAUDE.md b/CLAUDE.md index 669bcb1d..fc86bc04 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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) diff --git a/docs/Client.CLI.md b/docs/Client.CLI.md index 66301d4b..a61094f4 100644 --- a/docs/Client.CLI.md +++ b/docs/Client.CLI.md @@ -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" \ diff --git a/docs/Historian.md b/docs/Historian.md index 0fd60e08..2677061e 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -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,95 @@ otopcua-cli historyread -u opc.tcp://localhost:4840 -U apikey -P otua__ ⚠️ **#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 OPC quality 0. +- **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. +- **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 +858,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` | diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index 95b8067d..aa87cbef 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -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; /// public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable { + /// + /// 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 + /// / / + /// families of : Average, Count, Minimum, + /// Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA + /// (TimeAverage, Total — ). Any other aggregate throws + /// without calling the gateway. It is what + /// returns, so the advertised AggregateFunctions folders and the server's + /// up-front BadAggregateNotSupported read it too (#682). + /// + public static IReadOnlySet ServedAggregates { get; } = + Enum.GetValues() + .Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a)) + .ToFrozenSet(); + private readonly IHistorianGatewayClient _client; private readonly ILogger _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 /// Creates a gateway-backed historian data source. /// The gateway client seam used for all reads. /// Diagnostic logger; failures are recorded without leaking tag/host detail. - public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger) + /// + /// In-window raw-row cap of one raw-evaluated processed read (ServerHistorian:MaxTimeWeightedRawRows; + /// <= 0 = unbounded). Past it the node fails BadTooManyOperations. + /// + /// Clock for result server timestamps and the raw look-ahead horizon; null = system. + public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger, + int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null) { ArgumentNullException.ThrowIfNull(client); ArgumentNullException.ThrowIfNull(logger); _client = client; _logger = logger; + _maxRawRows = maxRawRows; + _time = timeProvider ?? TimeProvider.System; } /// @@ -75,17 +104,38 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } /// + /// + /// + /// Per-sample aggregates (#681) — Average, Count, Minimum, Maximum, Range, Start, End, Delta, + /// Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator + /// (): 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 + /// ServerHistorian:MaxTimeWeightedRawRows (, not counted as + /// a backend failure). + /// + /// + /// TimeAverage / Total stay native AVEVA retrieval modes (), stamped with + /// the interval start and the Calculated bit. Everything else is + /// (see ). + /// + /// + /// A raw-evaluated read whose window holds more than + /// intervals. public async Task 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(); 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 } } + /// Serves a per-sample aggregate from a raw read (#681); see . + private async Task 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; + } + } + + /// + /// True when (or an inner exception — the client package wraps the ) + /// is the gateway's InvalidArgument "max_values exceeds the server ceiling" rejection. + /// + 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; + } + /// public async Task ReadAtTimeAsync( string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) @@ -132,13 +242,6 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// public bool SupportsReverseEvents => false; - /// - /// The aggregates the Gateway backend serves — the single static the advertised AggregateFunctions folders - /// and the server's up-front BadAggregateNotSupported read through (#682). - /// Today: , the members the gateway has a retrieval mode for. - /// - public static readonly IReadOnlySet ServedAggregates = AggregateModeMapper.Served; - /// /// ; every other member answers BadAggregateNotSupported and is not advertised. public IReadOnlySet SupportedAggregates => ServedAggregates; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs index 4df51b33..7927e71d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs @@ -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); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs index 16ba239a..a8cc4d30 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs @@ -6,29 +6,33 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping; /// /// Maps the driver-agnostic (OPC UA Part 13 aggregate) onto the -/// gateway's native . +/// gateway's native — for the aggregates still answered by AVEVA itself. /// /// /// -/// Served: → , -/// Minimum → , Maximum → , -/// Total → . +/// Native (AVEVA-computed): → +/// and → . 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). /// /// -/// Deliberately NOT served (#679): (OPC UA Average is the arithmetic mean; -/// AVEVA's modes include no arithmetic mean — only the time-weighted one, which is TimeAverage) and -/// (AVEVA Counter is a rollover-counter increase, not a sample count). -/// Both surface as BadAggregateNotSupported 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: evaluates them from raw reads +/// with the SQL backends' shared evaluator (Minimum/Maximum used to map to AVEVA's MinimumWithTime / +/// MaximumWithTime). Every other member — incl. TimeAverage2, Total2, Interpolative, PercentGood, PercentBad — is +/// here. /// /// internal static class AggregateModeMapper { /// - /// The aggregates the gateway serves — every defined member maps, derived from it so the - /// advertised set and the dispatch can never disagree. - /// 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 maps, derived + /// from it so the dispatch and the declared set can never disagree. Part of + /// (#681/#682). /// - public static readonly IReadOnlySet Served = + public static readonly FrozenSet NativeAggregates = Enum.GetValues().Where(a => TryMap(a) is not null).ToFrozenSet(); /// Maps an aggregate function to the gateway retrieval mode. @@ -48,12 +52,10 @@ internal static class AggregateModeMapper nameof(aggregate), aggregate, "Unmapped HistoryAggregateType — add a RetrievalMode mapping."); } - /// The single mapping table: the retrieval mode for a served aggregate, or null. + /// The single mapping table: the retrieval mode for a natively served aggregate, or null. 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, }; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs index 0b8fdd94..232b453b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs @@ -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); + } + + /// The OPC UA status of a raw sample (shared by raw reads and the #681 raw-evaluated aggregates). + /// The gateway raw sample. + /// The mapped status code. + 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); } /// Maps a batch of raw samples to value snapshots, in order. @@ -57,20 +65,28 @@ internal static class SampleMapper /// The gateway aggregate sample. /// The driver-agnostic snapshot. /// + /// /// Unlike the legacy Wonderware DTO (a nullable double?), the gateway proto carries a /// non-optional double value, so an unavailable (no-data) bucket cannot be signalled by a /// null value. Instead it is signalled by a non-Good opc_quality: a Good bucket /// (opc_quality >= 192) yields its value with a Good status, anything else maps to /// BadNoData with a null value — preserving the Wonderware aggregate contract (binary /// Good-with-value / BadNoData-null). + /// + /// + /// #681: both statuses carry the Calculated info bits (0x401) 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 start_time from a row field (tail + 43) its + /// reverse-engineering notes still mark unverified. + /// /// 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); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs new file mode 100644 index 00000000..2f4a9a3a --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -0,0 +1,296 @@ +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; + +/// +/// #681: serves the per-sample processed aggregates (the , +/// and families) on the Gateway backend by +/// streaming the window's RAW samples through the gateway and evaluating them with the shared +/// Driver.Historian.Sql.Common code — for the bucket math (forward and reverse, #677) and +/// for values, status and the Partial/Calculated bits — so the Gateway answers +/// exactly what SqlServer and QuestDB answer for the same stored samples. +/// +/// +/// +/// Forward raw reads, each consumed to completion. The gateway's ReadRaw streams oldest-first over an +/// inclusive [start, end] and refuses end < start, so a reverse read streams the same ascending range +/// and assembles the result newest-first. The window stream is classified sample by sample +/// with ([start, end) forward, (start, end] 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 HistorianRpcFaults.IsConnectionFault: any RpcException ⇒ true). +/// +/// +/// The AVEVA start-boundary row. An AVEVA Full raw read emits an initial-value row stamped exactly at +/// the query start (histsdk reverse-engineering: Quality = 133, OpcQuality = 192) carrying the value +/// in effect there, not a stored sample. The window stream is therefore requested from +/// (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 NumericValue = 0 on every +/// numeric row, StringValue = "" on a string one) but with OPC quality 0. +/// +/// +/// The data range. 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 +/// (; a plain Average needs neither). Earlier data is a non-Bad +/// pre-window row, or — when none arrived — a -row read [, +/// start). Later data is a sample exactly at the window end (forward) from the window read, or else a +/// -row read [end, now + ] (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. +/// +/// +/// Bounded. The in-window rows are capped by ServerHistorian:MaxTimeWeightedRawRows (the same knob the +/// SQL backends' time-weighted reads use): past it the node fails with +/// (⇒ BadTooManyOperations) once the stream — bounded by the +/// requested row budget — has been drained. The gateway is asked for at most cap + 2 + +/// rows; a stream that delivers all of them cannot be proven complete and fails the +/// same way rather than answering from a truncated window. +/// +/// +/// Values and status. A sample's value is its numeric value (Double), else its string value, else Empty; its +/// status is — 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 +/// (), exactly as on the SQL backends. +/// +/// +internal static class RawAggregateReader +{ + /// How far before the window the raw stream starts, so the AVEVA initial-value row lands outside it. + internal static readonly TimeSpan BoundaryMargin = TimeSpan.FromMilliseconds(1); + + /// How far past "now" the stream may look for a sample after the window (end-of-data rules). + internal static readonly TimeSpan LookAheadHorizon = TimeSpan.FromDays(1); + + /// Rows the stream may spend before the window (the boundary row plus samples inside the margin). + internal const int PreWindowAllowance = 16; + + /// Start of the "is there data before the window?" probe (the Unix epoch — before any AVEVA history). + internal static readonly DateTime ProbeFloor = DateTime.UnixEpoch; + + /// Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample). + internal const int ProbeRows = 2; + + /// True for the aggregates this reader serves: the per-sample families of . + /// Requested aggregate. + /// Whether the aggregate is evaluated from raw samples. + public static bool Serves(HistoryAggregateType aggregate) => + AggregateFamilies.FamilyOf(aggregate) is AggregateFamily.Simple or AggregateFamily.StartEnd or AggregateFamily.ActualTime; + + /// The gateway max_values for a configured window cap ( when unbounded). + /// The configured cap (<= 0 = unbounded). + /// The row limit to request. + public static int MaxValuesFor(int maxWindowRows) => + maxWindowRows <= 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)maxWindowRows + 2 + PreWindowAllowance); + + /// Maps one gateway raw sample to a stored sample (value + mapped status, Empty clamped Bad). + /// The gateway sample; its timestamp must be set. + /// The stored sample. + 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)); + } + + /// Reads and evaluates one per-sample processed read. + /// The gateway client. + /// Historian tag name. + /// A per-sample aggregate (). + /// The bucket plan. + /// The in-window raw-row cap (<= 0 = unbounded). + /// Server timestamp, and the anchor of the look-ahead horizon. + /// Cancellation token. + /// One value per interval, in the plan's order. + /// The window holds more than rows, + /// or the stream could not be read to completion within the requested row budget. + public static async Task> 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 + // OPC quality 0. + 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); + } + + /// + /// The time of the first stored sample (any quality) strictly between and + /// , read as a -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. + /// + private static async Task 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; + } + + /// + /// The row's OPC DA quality alone (opc_quality ≥ 64: Good or Uncertain) — deliberately without + /// 's fallback to the AVEVA quality field when opc_quality is 0, which + /// is also the OPC DA Bad value a NULL row carries. + /// + private static bool IsNotBadOnTheWire(HistorianSample raw) => raw.OpcQuality >= 64; + + 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."); +} + +/// +/// Builds the per-bucket statistics () the SQL backends' aggregate queries produce, from a stream +/// of raw samples — the C# statement of SqlAggregateQueries' columns: counts by class (usable = non-Bad with a +/// value; good = usable with Good severity), Avg/Min/Max/Sum over the usable numeric values, first/last usable value by +/// time, the extreme usable numeric value with its time (ties ⇒ earliest), and the first/last sample of any quality. +/// +/// +/// 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. +/// +internal sealed class RawBucketAccumulator +{ + private readonly bool _maximum; + private readonly Dictionary _buckets = new(); + + /// Creates an accumulator. + /// Extreme = the maximum (MaximumActualTime); otherwise the minimum. + public RawBucketAccumulator(bool maximum) => _maximum = maximum; + + private sealed class State + { + public int Total, Usable, Good, NonNumeric, Numeric; + public double Sum, Min = double.PositiveInfinity, Max = double.NegativeInfinity; + public StoredSample? First, Last, FirstAny, LastAny, Extreme; + } + + /// Adds one in-window sample to its bucket. Order-independent (picks compare by time). + /// Ascending bucket index (). + /// The sample. + 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++; + return; + } + st.Numeric++; + 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; + } + } + + /// The non-empty buckets' rows, ascending by bucket index. + /// The rows. + public IEnumerable Rows() => + _buckets.OrderBy(kv => kv.Key).Select(kv => + { + var st = kv.Value; + var hasNumeric = st.Numeric > 0; + return new BucketRow + { + Bucket = kv.Key, + Total = st.Total, + Usable = st.Usable, + Good = st.Good, + Bad = st.Total - st.Usable, + NonNumeric = st.NonNumeric, + Avg = hasNumeric ? st.Sum / st.Numeric : null, + Min = hasNumeric ? st.Min : null, + Max = hasNumeric ? st.Max : null, + Sum = hasNumeric ? st.Sum : 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, + }; + }); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj index 262241a0..a6205447 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj @@ -11,6 +11,11 @@ + +