diff --git a/CLAUDE.md b/CLAUDE.md index 2a385dac..96a304f6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1343,7 +1343,7 @@ historian tagname, which is the tag's driver `FullName`. ### Pluggable backends, SQL Server and HistoryUpdate (2026-10) - **Selection:** `ServerHistorian:Backend` (`Gateway` default | `SqlServer` | `QuestDb`; unknown fails host start). `HistorianBackendRegistry` → `IHistorianBackendProvider` builds all five seams; the gateway keys are only required for `Gateway`. `SqlServer` uses `ServerHistorian:SqlServer:{ConnectionString (required), Schema=hist, CommandTimeout=30s, AutoCreateSchema=true}`. -- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; aggregates (13, simple per-sample definitions, not Part 13) are pushed down to T-SQL with bucket math in **100 ns ticks**. Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. +- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (12, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted five (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM`) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; client `AggregateConfiguration` ignored; status = the SDK `AggregateCalculator` time-based rule, Uncertain time counts as Good; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`). Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. @@ -1372,6 +1372,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`) may stream into the C# evaluator; 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/Historian.md b/docs/Historian.md index fbc3808e..6a93b4df 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 | Average, Minimum, Maximum, Total, Count | The thirteen in the [aggregate table](#aggregates-processed-reads) | The thirteen, pushed down to QuestDB SQL | +| Processed aggregates | Average, Minimum, Maximum, Total, Count | 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) | | Alarm history | gateway `SendEvent` | `Event` table (insert-if-absent) | `{prefix}event` table (dedup upsert on `(ts, event_id)`) | | Provisioning (`EnsureTags`) | numeric types only; others skipped | `MERGE` upsert of every request, all types | Upsert into `{prefix}tag`, all types | @@ -110,6 +110,7 @@ and all HistoryRead calls on historized nodes return `GoodNoData` (empty, not an "HistoryReadBatchParallelism": 4, "MaxConcurrentHistoryReadBatches": 16, "HistoryReadDeadline": "00:01:00", + "MaxTimeWeightedRawRows": 1000000, "SqlServer": { "ConnectionString": "", "Schema": "hist", @@ -141,6 +142,7 @@ and all HistoryRead calls on historized nodes return `GoodNoData` (empty, not an | `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) 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). | | `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). | **`ServerHistorian:SqlServer`** (read only when `Backend=SqlServer`; bound by `HistorianBackendRegistry.BindSqlServer`): @@ -515,7 +517,7 @@ The server supports all four OPC UA HistoryRead variants: | Variant | Node type | CLI `--aggregate` | |---|---|---| | **Raw** | Historized variable | (omit `--aggregate`) | -| **Processed** | Historized variable | `Average`, `Minimum`, `Maximum`, `Count`, `Start`, `End`, `StandardDeviation` (the CLI's own list; `SqlServer` serves thirteen aggregates — see the [aggregate table](#aggregates-processed-reads)) | +| **Processed** | Historized variable | `Average`, `Minimum`, `Maximum`, `Count`, `Start`, `End`, `StandardDeviation` (the CLI's own list; `SqlServer` / `QuestDb` serve seventeen aggregates — see the [aggregate table](#aggregates-processed-reads)) | | **AtTime** | Historized variable | n/a (client supplies exact timestamps) | | **Events** | Equipment folder (event notifier) | n/a | @@ -649,9 +651,9 @@ buckets surface as `BadNoData`. `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); `OtOpcUaNodeManager.MapAggregate` -maps each standard aggregate NodeId. The **`SqlServer`** backend serves the thirteen listed in the -[aggregate table](#aggregates-processed-reads); everything else (e.g. `Interpolative`, `TimeAverage`, -`StandardDeviationPopulation`) is `BadAggregateNotSupported` for that node. The **HistorianGateway** backend only has retrieval modes for the original +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 original five (Average, Minimum, Maximum, Total, Count): for any other member `AggregateModeMapper` throws `HistoryAggregateNotSupportedException`, which the HistoryRead path turns into `BadAggregateNotSupported` for that node only (not a generic failure, and not counted as a backend fault). The OPC UA Client driver forwards every @@ -782,10 +784,21 @@ point (see [Continuation-point paging (Events)](#continuation-point-paging-event ### Aggregates (Processed reads) -> **These are simple per-sample definitions, not full OPC UA Part 13.** There is no time-weighting, no interpolated interval-boundary -> value and no partial-interval status. They are pushed down to T-SQL and suit trend-style consumption. Any `HistoryAggregateType` not -> in the table throws `HistoryAggregateNotSupportedException` ⇒ `BadAggregateNotSupported` for that node (e.g. `Interpolative`, -> `TimeAverage`, `StartBound`, `DurationGood`, `StandardDeviationPopulation`). +> **Two kinds.** The *per-sample* aggregates (Average, Minimum, Maximum, Count, Range, Start/End/Delta, Min/MaxActualTime, +> PercentGood/Bad) are simple per-sample definitions — no time-weighting, no interval-boundary values — pushed down to SQL. The +> *time-weighted* aggregates (TimeAverage, TimeAverage2, Total, Total2, Interpolative; #671) follow OPC UA Part 13 with bounding values +> — see [Time-weighted aggregates](#time-weighted-aggregates-part-13-671). Any `HistoryAggregateType` not in the tables throws +> `HistoryAggregateNotSupportedException` ⇒ `BadAggregateNotSupported` for that node (e.g. `StartBound`, `DurationGood`, +> `StandardDeviationPopulation`). +> +> ⚠️ **Release note / migration (#671): `Total` changed meaning on the `SqlServer` and `QuestDb` backends.** It was the plain `SUM` of +> the usable samples in the interval, pushed down to SQL; it is now the Part 13 time-weighted integral, `TimeAverage × interval seconds` +> (value·seconds), **evaluated in C# from the raw rows** (not pushed down) and therefore subject to `ServerHistorian:MaxTimeWeightedRawRows` +> — a long `Total` over a dense tag that used to succeed may now fail `BadTooManyOperations`; read it in shorter windows. This brings +> `Total` closer to the `Gateway` backend, whose `Total` maps to AVEVA's native time-weighted `Integral` retrieval mode — but AVEVA's +> integral time unit/divisor has not been verified, so the two backends may differ by a constant factor for the same data. A client +> that relied on the old sum gets a different number for the same data (e.g. two samples 10 and 20 ten seconds apart no longer total +> 30). There is no switch back; `Count` × `Average` reproduces the old sum. **Bucketing.** One result per interval of `[start, end)`; bucket *i* = `[start + i·interval, start + (i+1)·interval)`, start-inclusive / end-exclusive to the tick, the last bucket truncated at `end`. `interval <= 0` ⇒ a single interval `[start, end)`; `start == end` ⇒ no @@ -801,7 +814,7 @@ status carries the historian *Calculated* info bit, `BadNoData` included. | Aggregate | Computed over | Value timestamp | |---|---|---| -| `Average`, `Minimum`, `Maximum`, `Total` | `AVG` / `MIN` / `MAX` / `SUM` of `NumValue` over usable samples | bucket start | +| `Average`, `Minimum`, `Maximum` | `AVG` / `MIN` / `MAX` of `NumValue` over usable samples | bucket start | | `Count` | number of usable samples (`Int32`) | bucket start | | `Range` | `Maximum − Minimum` | bucket start | | `Start`, `End` | first / last usable sample by time (keeps the sample's type) | bucket start | @@ -818,6 +831,86 @@ Status and edge rules: - **`PercentGood` / `PercentBad` and Uncertain samples:** Uncertain samples are counted in the denominator but are **neither Good nor Bad**, so `PercentGood + PercentBad < 100` whenever a bucket holds Uncertain samples. Both report status `Good` for any non-empty bucket. - Everything except `Count` and `Start`/`End` is a `double`; `Int64` samples aggregate as doubles and `Boolean` as 0/1. +### Time-weighted aggregates (Part 13, #671) + +`TimeAverage`, `TimeAverage2`, `Total`, `Total2` and `Interpolative`, evaluated by the shared +`Driver.Historian.Sql.Common/TimeWeightedAggregator.cs` for **both** SQL backends (so `SqlServer` and `QuestDb` agree exactly). +**This whole family — `Total` included — is evaluated in C#, not pushed down to SQL** (see *Pushdown trade-off* and the raw-row cap +below). Bucketing (edges, truncated last bucket, the 100 000-interval cap) is the same as above; reverse windows are covered below. + +**Server defaults, fixed:** `TreatUncertainAsBad = false`, `PercentDataBad = PercentDataGood = 100`, `UseSlopedExtrapolation = false`. +**A client's `AggregateConfiguration` in the HistoryRead request is ignored** (the node manager does not forward it); every read uses +these defaults. + +**Bounding values.** Besides the window `[start, end)`, each read fetches four single-row seeks: the latest *usable* sample before +`start`, the latest sample of any quality before `start` (they differ when an outage straddles the start), the first sample of any +quality at or after `end`, and the first *usable* one at or after `end`. So a bucket with no samples of its own, between two samples, +still gets a value. + +**The signal.** Between a usable sample `p` and the next sample `s`: both `Double`/`Int64` ⇒ **sloped** (linear), segment quality the +worse of the two; either `Boolean` ⇒ **stepped** (`p` held), quality `p`'s; `s` Bad ⇒ `p` held up to the outage (never interpolated into +it — the AtTime rule), Uncertain; no later sample at all ⇒ `p` held to the window end (stepped extrapolation), Uncertain. From a Bad +sample (incl. an `Empty` outage marker) to the next sample the signal is **Bad** (outage); before the first sample there is **no +data**. That is the +*simple-bounding* signal. The *interpolated-bounding* signal skips Bad samples: it joins consecutive usable samples (sloped / stepped) +straight across an outage. + +| Aggregate | Value | Bad intervals | +|---|---|---| +| `TimeAverage` | ∫ interpolated-bounding signal ÷ its defined duration | interpolated across (Part 13) | +| `Total` | `TimeAverage × interval seconds` (value·s; the truncated last bucket uses its own length) | interpolated across | +| `TimeAverage2` | ∫ simple-bounding signal over non-Bad time ÷ non-Bad duration | excluded | +| `Total2` | ∫ simple-bounding signal over non-Bad time (value·s) | excluded | +| `Interpolative` | interpolated-bounding value **at the interval start** (reverse read: see below) | interpolated across | + +**Status, the four time-weighted values** — aligned with the OPC Foundation SDK's `AggregateCalculator.GetTimeBasedStatusCode` +(inspected in `Opc.Ua.Server` 1.5.378.106) under the fixed defaults. With `TreatUncertainAsBad = false`, **Uncertain time counts as +Good time**: Uncertain samples, a value held up to an outage and the stepped extrapolation past the last sample all count toward +PercentDataGood. + +- Every instant of the interval non-Bad ⇒ `Good`. +- Some non-Bad time but not all (an outage, or time before the first sample) ⇒ `UncertainDataSubNormal`. +- No non-Bad time but some outage time ⇒ generic **`Bad`** (`0x80000000`) — the SDK's all-Bad result, value null. +- No data at all ⇒ `BadNoData`. +- A usable `String`/`DateTime` sample feeding the interval ⇒ `BadAggregateInvalidInputs`. + +All carry the Calculated bit; a non-Bad result for the **truncated last interval** (shorter than the processing interval) also carries +the **Partial** bit (`0x4`, `AggregateBits.Partial`). Timestamp = interval start; value `double` (Boolean 0/1). (The per-sample +aggregates do not set Partial — a follow-up.) + +> **A live tag's latest interval.** Past the tag's last stored sample the value is held (stepped extrapolation, Uncertain). Under +> the SDK rule that Uncertain time is Good time, so the latest interval of a live tag normally reads **`Good`** — not Uncertain — +> with the last value held to the interval end. It reads Uncertain only if the interval also contains outage time or starts +> before the tag's first sample. Do not take a Good latest interval as proof that data arrived up to its end. If it was truncated +> by the request's end time it also carries **Partial**. + +**Status, `Interpolative`** (point-based, not the PercentDataGood rule): +- A usable sample exactly at the point ⇒ that value with **its own stored status** (no historian bit). +- No usable sample at or before the point ⇒ plain `BadNoData`. +- Otherwise the value with the **Interpolated** bit: + - `Good` on a segment whose defining samples are Good; + - `UncertainDataSubNormal` inside or up to an outage (bridged) and past the last sample (extrapolated). + +**Reverse reads (`end < start`).** Part 13 anchors a reverse read at the request's start, which is the later time. `Interpolative` +follows that: its points are `start, start − interval, …` (in the forward window's terms `End − i·interval`), latest first. A +reverse read 10:00 → 09:00 at 1 h returns the value at **10:00**, not 09:00, and no point ever falls before the window. The interval +aggregates (TimeAverage, Total, …) keep the existing backend-wide reverse rule: computed forward with intervals anchored at the +earlier time, returned in descending order. Re-anchoring those at the later time for every aggregate is a follow-up. + +**Pushdown trade-off.** Unlike the per-sample aggregates, the time-weighted family is **not** pushed down to SQL: the backend streams +the window's rows plus the four bounds (one batch of five result sets on SQL Server; a bounds `UNION ALL` + a window select on QuestDB) +and the evaluator splits segments at bucket edges in C#. Network traffic is therefore O(samples in the window) rather than O(buckets); +evaluation memory stays O(buckets) (rows are streamed, never materialised). The alternative — the same Part 13 segment +interpolation, edge splitting and two bounding definitions written twice, in T-SQL and QuestDB SQL, and kept bit-identical — was +judged the larger risk. For very long, densely sampled windows prefer the pushed-down `Average` (or a coarser read) over +`TimeAverage`. + +**Raw-row cap.** The window query is limited to `ServerHistorian:MaxTimeWeightedRawRows` + 1 rows (default 1 000 000, shared by both +SQL backends; `TOP (@maxRows)` / `LIMIT $4`). When that extra row arrives the read stops and the node fails with +**`BadTooManyOperations`** — for that node only; it is not counted in the backend's health failures. Without the cap a one-year +`Total` on a 1 Hz tag (≈31 M rows) would hold a pooled connection until `HistoryReadDeadline` and then return `BadTimeout`. The four +bound rows do not count toward the cap, and the per-sample aggregates are never capped. Raise the cap, or read in shorter windows. + ### Health The read client reports `GetHealthSnapshot()` (query/failure counters, consecutive failures, `LastError` — e.g. a missing schema under @@ -900,7 +993,7 @@ the spike that pinned every QuestDB-specific form: [plans/2026-10-10-questdb-spi `questdb/questdb:10.0.1`. QuestDB serves every value kind (`Double`, `Int64`, `Boolean`, `String`, `DateTime`, `Empty`), stores the real status, records comms-loss -outage markers, serves the thirteen aggregates and supports **HistoryUpdate** (soft delete), exactly like `SqlServer`. The same conformance +outage markers, serves the seventeen aggregates (time-weighted ones through the same shared evaluator, so results are identical) and supports **HistoryUpdate** (soft delete), exactly like `SqlServer`. The same conformance suite (`tests/Drivers/Historian.Conformance`) runs against both backends; the differences below are the only ones. ### Configuration (`ServerHistorian:QuestDb`) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs index 81096ec3..0be71792 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs @@ -47,6 +47,12 @@ public enum HistoryAggregateType AnnotationCount, StandardDeviationPopulation, VariancePopulation, + + /// Part 13 TimeAverage2: time-weighted average over the interval's non-Bad data only (#671). + TimeAverage2, + + /// Part 13 Total2: time-weighted integral (value·seconds) over the interval's non-Bad data only (#671). + Total2, } /// @@ -65,6 +71,22 @@ public sealed class HistoryAggregateNotSupportedException : NotSupportedExceptio => Aggregate = aggregate; } +/// +/// Thrown by a history backend when a read would exceed a configured size limit (e.g. the raw-row cap of the SQL +/// backends' time-weighted aggregates, #671). The Server HistoryRead path translates it to +/// BadTooManyOperations for that node only; it is a client-request limit, not a backend fault. +/// +public sealed class HistoryReadLimitExceededException : InvalidOperationException +{ + /// The configured limit that was exceeded. + public long Limit { get; } + + /// Creates the exception. + /// The configured limit. + /// Operator-facing description. + public HistoryReadLimitExceededException(long limit, string message) : base(message) => Limit = limit; +} + /// /// One row returned by — a historical /// alarm/event record, not the OPC UA live-event stream. Fields match the minimum set the diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs index afe02efa..85684bd9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs @@ -105,6 +105,42 @@ internal static class QuestDbAggregateQueries "SELECT st.*, r.num_value, r.ts " + "FROM st LEFT JOIN (SELECT * FROM r WHERE rn = 1) r ON r.b = st.b ORDER BY st.b"; + /// + /// Time-weighted family (#671), bounds statement — not pushed down (see ). + /// Four LIMIT 1 seeks UNION ALL-ed, each tagged with k: 0 = latest usable sample before the + /// start, 1 = latest sample of any quality before the start, 2 = first sample of any quality at or after the end, + /// 3 = first usable one at or after the end. Parameters: $1 tag, $2/$3 the forward window + /// start / end. Columns: k, then ts, kind, num_value, int_value, str_value, status, server_ts. + /// + /// Prefixed sample table name. + /// The SQL text. + public static string TimeWeightedBounds(string sample) + { + var usable = $" AND status < {NotBadLimit} AND kind <> {EmptyKind}"; + string Seek(int k, string cmp, string filter, string order) => + $"SELECT * FROM (SELECT {k.ToString(CultureInfo.InvariantCulture)} AS k, {SampleColumns} FROM {sample} " + + $"WHERE tag = $1 AND ts {cmp}{filter} AND NOT deleted ORDER BY ts {order} LIMIT 1)"; + return Seek(0, "< $2", usable, "DESC") + " UNION ALL " + Seek(1, "< $2", "", "DESC") + " UNION ALL " + + Seek(2, ">= $3", "", "ASC") + " UNION ALL " + Seek(3, ">= $3", usable, "ASC"); + } + + /// + /// Time-weighted family (#671), window statement: the live samples in [$2, $3) for tag $1, ascending, + /// at most $4 (the raw-row cap + 1). + /// Columns: ts, kind, num_value, int_value, str_value, status, server_ts. The caller feeds the before-bounds, + /// these rows and the after-bounds, in that order, into — the same evaluator + /// the SQL Server backend uses, so the two agree exactly. Trade-off: the window's rows travel to the server rather + /// than one row per bucket; Part 13's segment interpolation and bucket-edge splitting are then written once + /// instead of twice in two SQL dialects that must stay bit-identical. + /// + /// Prefixed sample table name. + /// The SQL text. + public static string TimeWeightedWindow(string sample) => + $"SELECT {SampleColumns} FROM {sample} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts ASC LIMIT $4"; + + /// The row columns the time-weighted statements return (after k for the bounds). + private const string SampleColumns = "ts, kind, num_value, int_value, str_value, status, server_ts"; + /// The bucket-index expression (exposed for tests). internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index b1d92d01..28042318 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -64,6 +64,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource private readonly ILogger _log; private readonly TimeProvider _time; private readonly HistorianHealthCounters _health = new(); + private readonly int _maxTimeWeightedRawRows; /// Creates the data source over the provider's shared connection, bootstrapper and tag catalog. /// Connection helper. @@ -71,10 +72,12 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// Tag catalog (reads use only its read-only ). /// Optional logger. /// Clock for "now" (AtTime beyond the last sample, server timestamps); system by default. + /// Raw in-window row cap of one time-weighted read (<= 0 = unbounded). internal QuestDbHistorianDataSource( QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, ILogger? log = null, - TimeProvider? timeProvider = null) + TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) { + _maxTimeWeightedRawRows = maxTimeWeightedRawRows; _conn = conn ?? throw new ArgumentNullException(nameof(conn)); _bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap)); _catalog = catalog ?? throw new ArgumentNullException(nameof(catalog)); @@ -332,9 +335,10 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// the statistics of each non-empty bucket (bucket index /// (cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks, never SAMPLE BY); /// produces each value and fills the - /// empty buckets with BadNoData. + /// empty buckets with BadNoData. The time-weighted family (#671) instead streams the window plus four bounding + /// rows into , exactly as the SQL Server backend does. /// - /// The aggregate is not one of the thirteen served. + /// The aggregate is not one of the seventeen served. /// The window holds more than intervals. public async Task ReadProcessedAsync( string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, @@ -348,6 +352,15 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource { var now = _time.GetUtcNow().UtcDateTime; var evaluated = new Dictionary(); + if (family == AggregateFamily.TimeWeighted) + { + var tw = new TimeWeightedAggregator(aggregate, plan); + if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false)) + await StreamTimeWeightedAsync(fullReference, plan, tw, cancellationToken).ConfigureAwait(false); + var twResult = tw.Finish(now); + RecordSuccess(); + return new HistoryReadResult(twResult, ContinuationPoint: null); + } if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false)) { var sample = _conn.SampleTable; @@ -377,6 +390,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource RecordSuccess(); return new HistoryReadResult(result, ContinuationPoint: null); } + catch (HistoryReadLimitExceededException) + { + // A client-request limit (the raw-row cap), not a backend fault: not counted as a failure. + throw; + } catch (Exception ex) { RecordFailure(ex); @@ -384,6 +402,47 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource } } + /// + /// Streams the time-weighted bounds () and window + /// () into in time order: the two + /// before-bounds, the window, the two after-bounds (duplicates are dropped by the aggregator). + /// + private async Task StreamTimeWeightedAsync(string tag, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct) + { + var bounds = new StoredSample?[4]; + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.TimeWeightedBounds(_conn.SampleTable))) + { + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, plan.StartUtc); + QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); + await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await reader.ReadAsync(ct).ConfigureAwait(false)) + bounds[Convert.ToInt32(reader.GetValue(0), CultureInfo.InvariantCulture)] = ReadStored(reader, 1); + } + + if (bounds[0] is { } usableBefore) tw.Add(usableBefore); + if (bounds[1] is { } anyBefore) tw.Add(anyBefore); + await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.TimeWeightedWindow(_conn.SampleTable))) + { + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, plan.StartUtc); + QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); + cmd.Parameters.Add(new NpgsqlParameter + { NpgsqlDbType = NpgsqlDbType.Integer, Value = TimeWeightedAggregator.WindowFetchLimit(_maxTimeWeightedRawRows) }); + await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + long windowRows = 0; + while (await reader.ReadAsync(ct).ConfigureAwait(false)) + { + // LIMIT $4 = cap + 1: one row past the cap fails the node without reading on. + TimeWeightedAggregator.CheckWindowRows(++windowRows, _maxTimeWeightedRawRows, tag); + tw.Add(ReadStored(reader, 0)); + } + } + if (bounds[2] is { } anyAfter) tw.Add(anyAfter); + if (bounds[3] is { } usableAfter) tw.Add(usableAfter); + } + private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family) { static double? D(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs index f56d2205..19230a6b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs @@ -1,3 +1,4 @@ +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; using System.Collections.Frozen; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; @@ -33,11 +34,16 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync /// Creates the provider over the ServerHistorian:QuestDb options (validated elsewhere). /// The ServerHistorian:QuestDb options. - public QuestDbHistorianProvider(QuestDbHistorianOptions options) + /// The shared ServerHistorian:MaxTimeWeightedRawRows cap (<= 0 = unbounded). + public QuestDbHistorianProvider(QuestDbHistorianOptions options, + int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) { _options = options ?? throw new ArgumentNullException(nameof(options)); + _maxTimeWeightedRawRows = maxTimeWeightedRawRows; } + private readonly int _maxTimeWeightedRawRows; + /// public string Name => "QuestDb"; @@ -69,7 +75,7 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync { var s = Get(sp); return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog, - Loggers(sp).CreateLogger()); + Loggers(sp).CreateLogger(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs index f79c3747..ff097de9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs @@ -17,7 +17,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is /// computed over the numeric value (Boolean = 0/1). These are the simple per-sample -/// definitions, not Part 13's interval-boundary (interpolated bound / time-weighted) semantics. +/// definitions; the Part 13 time-weighted aggregates (TimeAverage, TimeAverage2, Total, Total2, Interpolative) +/// are evaluated by , not here. /// /// public static class AggregateEvaluator @@ -28,7 +29,8 @@ public static class AggregateEvaluator /// The bucket's start time (the result timestamp). /// Server timestamp. /// The calculated value. - /// The aggregate is not one of the thirteen served. + /// The aggregate is not one of the per-sample (Simple / + /// StartEnd / ActualTime family) aggregates. public static DataValueSnapshot Evaluate(HistoryAggregateType aggregate, BucketRow r, DateTime bucketStart, DateTime now) { DataValueSnapshot Calc(object? value, uint status, DateTime? t = null) => @@ -53,7 +55,6 @@ public static class AggregateEvaluator case HistoryAggregateType.Average: return Calc(r.Avg, status); case HistoryAggregateType.Minimum: return Calc(r.Min, status); case HistoryAggregateType.Maximum: return Calc(r.Max, status); - case HistoryAggregateType.Total: return Calc(r.Sum, status); case HistoryAggregateType.Range: return Calc(r.Max - r.Min, status); case HistoryAggregateType.Start: return Calc(r.First.ToObject(), status); case HistoryAggregateType.End: return Calc(r.Last.ToObject(), status); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs index 761a0cab..4b94d3d8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs @@ -5,7 +5,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// The query shape that serves an aggregate. One query per family. public enum AggregateFamily { - /// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad: one GROUP BY. + /// Average / Minimum / Maximum / Count / Range / PercentGood / PercentBad: one GROUP BY. Simple, /// Start / End / Delta: first + last usable sample per bucket. @@ -13,6 +13,13 @@ public enum AggregateFamily /// MinimumActualTime / MaximumActualTime: extreme usable sample per bucket, ties ⇒ earliest. ActualTime, + + /// + /// TimeAverage / TimeAverage2 / Total / Total2 / Interpolative: Part 13 time-weighted, with bounding values. + /// The backend streams the window's samples plus the bounding rows and + /// evaluates them (identically for every backend). + /// + TimeWeighted, } /// Which aggregates the SQL historian backends serve, and by which family. @@ -24,10 +31,12 @@ public static class AggregateFamilies public static AggregateFamily? FamilyOf(HistoryAggregateType aggregate) => aggregate switch { HistoryAggregateType.Average or HistoryAggregateType.Minimum or HistoryAggregateType.Maximum - or HistoryAggregateType.Total or HistoryAggregateType.Count or HistoryAggregateType.Range + or HistoryAggregateType.Count or HistoryAggregateType.Range or HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad => AggregateFamily.Simple, HistoryAggregateType.Start or HistoryAggregateType.End or HistoryAggregateType.Delta => AggregateFamily.StartEnd, HistoryAggregateType.MinimumActualTime or HistoryAggregateType.MaximumActualTime => AggregateFamily.ActualTime, + HistoryAggregateType.TimeAverage or HistoryAggregateType.TimeAverage2 or HistoryAggregateType.Total + or HistoryAggregateType.Total2 or HistoryAggregateType.Interpolative => AggregateFamily.TimeWeighted, _ => null, }; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianInfoBits.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianInfoBits.cs index 1a067546..3e6b9d6d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianInfoBits.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianInfoBits.cs @@ -12,6 +12,12 @@ public static class HistorianInfoBits /// The value was interpolated. public const uint HistorianInterpolated = 0x2u; + /// + /// Partial (Part 4 §7.39 / Opc.Ua.AggregateBits.Partial): the calculated value covers an interval shorter + /// than the processing interval (the truncated last interval of a processed read). + /// + public const uint HistorianPartial = 0x4u; + /// /// Marks as a Calculated data value. Any historian bits the status already carries /// (e.g. a stored Interpolated sample) are cleared first, so the result is never both. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianStatusCodes.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianStatusCodes.cs index e8896f66..9fa08bcf 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianStatusCodes.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/HistorianStatusCodes.cs @@ -10,6 +10,9 @@ public static class HistorianStatusCodes public const uint GoodEntryInserted = 0x00A20000u; public const uint GoodEntryReplaced = 0x00A30000u; public const uint BadNoData = 0x809B0000u; + + /// Generic Bad: a time-weighted interval holding data but no non-Bad time (the SDK aggregate rule). + public const uint Bad = 0x80000000u; public const uint BadEntryExists = 0x809F0000u; public const uint BadNoEntryExists = 0x80A00000u; public const uint BadInvalidTimestamp = 0x80230000u; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs new file mode 100644 index 00000000..a4737fa8 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs @@ -0,0 +1,430 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +/// +/// The OPC UA Part 13 time-weighted aggregates (): TimeAverage, +/// TimeAverage2, Total, Total2 and Interpolative, evaluated over a time-ordered stream of +/// stored samples. Both SQL backends fetch the same rows and feed them here, so the two produce identical results +/// by construction. +/// +/// +/// +/// Input. Samples in strictly ascending time: the bounding rows before the window (the latest +/// usable sample before start, then the latest sample of any quality before start — they differ +/// when an outage straddles the start), every in-window sample, then the bounding rows at/after the window end +/// (the first sample of any quality at or after end, then the first usable one). +/// ignores a sample at or before the last one, so a bound fetched twice is harmless. Memory is O(buckets), not O(samples). +/// +/// +/// The signal (server defaults: TreatUncertainAsBad = false, PercentDataBad = PercentDataGood = 100, +/// UseSlopedExtrapolation = false; a client AggregateConfiguration is not honoured). Between a usable +/// sample p and the next sample s: +/// +/// s usable, both Double/Int64 ⇒ sloped (linear p → s); quality = worse of the two. +/// s usable, either Boolean ⇒ stepped (p held); quality = p's. +/// s Bad ⇒ p held up to the outage (never interpolated into it, as AtTime does); Uncertain. +/// no later sample at all ⇒ p held to the window end (stepped extrapolation); Uncertain. +/// +/// From a Bad sample (incl. an Empty outage marker) to the next sample, and before the first sample, the +/// signal is Bad (no data). This is the simple bounding signal. +/// +/// +/// The interpolated bounding signal (Part 13 "interpolated bounding values") skips Bad samples: it joins +/// consecutive usable samples the same way (sloped / stepped) straight across any outage, and holds the +/// last usable value to the window end. It is undefined before the first usable sample. +/// +/// +/// TimeAverage∫ interpolated-bounding signal ÷ its defined duration in the interval. +/// TotalTimeAverage × interval seconds (value·s) — Part 13 §5.4.3.8. +/// TimeAverage2∫ simple-bounding signal over its non-Bad parts ÷ non-Bad duration (Bad intervals excluded). +/// Total2∫ simple-bounding signal over its non-Bad parts, in value·s (= TimeAverage2 × non-Bad seconds). +/// InterpolativeThe interpolated-bounding signal's value at each interval start — +/// for a reverse read, at End − i·interval (Part 13 anchors a reverse read at its later time), emitted +/// latest first. +/// +/// +/// Status (time-weighted four) — the SDK AggregateCalculator.GetTimeBasedStatusCode rule with +/// TreatUncertainAsBad = false, so Uncertain time (Uncertain samples, a value held up to an outage, stepped +/// extrapolation past the last sample) counts as Good time: every instant of the interval non-Bad ⇒ +/// Good; no non-Bad time but some Bad (outage) time ⇒ Bad; no data at all ⇒ BadNoData; +/// otherwise (PercentDataGood = PercentDataBad = 100) UncertainDataSubNormal. Time before the first sample +/// counts as Bad (no data) time. A usable String/DateTime sample contributing to the interval ⇒ +/// BadAggregateInvalidInputs. Every status carries the Calculated bit, plus the Partial bit on a non-Bad +/// result of a truncated last interval; the timestamp is the interval start; the value is a +/// (Boolean = 0/1, Int64 as double). +/// +/// +/// Status (Interpolative). A usable sample exactly at the point ⇒ its value with its own status +/// (raw, no historian bit). Otherwise no usable sample at or before ⇒ BadNoData; else the interpolated +/// value with the Interpolated bit and the status of the simple-bounding signal at that instant: Good on +/// a Good segment, UncertainDataSubNormal on an Uncertain one, inside an outage (interpolated across it) +/// or past the last sample (extrapolated). +/// +/// +public sealed class TimeWeightedAggregator +{ + private const double TicksPerSecond = TimeSpan.TicksPerSecond; + + /// + /// Default cap on the raw in-window rows one time-weighted read may stream (ServerHistorian:MaxTimeWeightedRawRows). + /// + public const int DefaultMaxWindowRows = 1_000_000; + + /// + /// The TOP/LIMIT a backend puts on its window query for a configured cap: cap + 1 (so one + /// row past the cap proves it was exceeded without reading on), or when the cap is + /// <= 0 (unbounded). + /// + /// The configured cap. + /// The row limit to put on the window query. + public static int WindowFetchLimit(int maxWindowRows) => + maxWindowRows <= 0 || maxWindowRows == int.MaxValue ? int.MaxValue : maxWindowRows + 1; + + /// Throws when has passed a positive . + /// Window rows read so far. + /// The configured cap (<= 0 = unbounded). + /// The tag, for the message. + /// The cap is exceeded. + public static void CheckWindowRows(long rowsRead, int maxWindowRows, string tag) + { + if (maxWindowRows > 0 && rowsRead > maxWindowRows) + throw new HistoryReadLimitExceededException(maxWindowRows, + $"The time-weighted aggregate read of '{tag}' covers more than {maxWindowRows} raw samples " + + "(ServerHistorian:MaxTimeWeightedRawRows); narrow the window or use a pushed-down aggregate such as Average."); + } + + private readonly HistoryAggregateType _aggregate; + private readonly BucketPlan _plan; + private readonly long _start; + private readonly long _end; + + // Simple-bounding (raw) signal, per bucket: ∫ over non-Bad time, non-Bad ticks, ticks inside an outage. + private readonly double[] _rawArea; + private readonly long[] _usableTicks; + private readonly long[] _badTicks; + // Interpolated-bounding (bridged) signal, per bucket. + private readonly double[] _bridgedArea; + private readonly long[] _bridgedTicks; + private readonly bool[] _invalid; + // Interpolative, per point (index = ascending time; see PointTicks). + private readonly DataValueSnapshot?[] _points; + + private StoredSample? _prevAny; + private StoredSample? _prevUsable; + private long _nextPoint; // next point (ascending time) not yet classified + private readonly List _pending = []; // points awaiting the next usable sample + private bool _finished; + + /// + /// What the simple-bounding signal is at a pending bucket start: no sample yet; on a segment from a usable sample + /// whose end is not known yet; or degraded (inside an outage, or held up to one). + /// + private enum PointQuality { NoPrior, OnSegment, Degraded } + + private readonly record struct PendingPoint(long Bucket, long Ticks, PointQuality Quality); + + /// Starts an evaluation of over . + /// One of the aggregates. + /// The bucket plan (forward window). + /// The aggregate is not time-weighted. + public TimeWeightedAggregator(HistoryAggregateType aggregate, BucketPlan plan) + { + ArgumentNullException.ThrowIfNull(plan); + if (AggregateFamilies.FamilyOf(aggregate) != AggregateFamily.TimeWeighted) + throw new HistoryAggregateNotSupportedException(aggregate); + _aggregate = aggregate; + _plan = plan; + _start = plan.StartUtc.Ticks; + _end = plan.EndUtc.Ticks; + var n = (int)plan.BucketCount; + _rawArea = new double[n]; + _usableTicks = new long[n]; + _badTicks = new long[n]; + _bridgedArea = new double[n]; + _bridgedTicks = new long[n]; + _invalid = new bool[n]; + _points = new DataValueSnapshot?[n]; + } + + /// Evaluates over an already time-ordered sample sequence. + /// One of the aggregates. + /// The bucket plan. + /// Bounds + window samples, ascending by time. + /// Server timestamp for the results. + /// One value per bucket, in the plan's order (descending for a reverse window). + public static IReadOnlyList Evaluate( + HistoryAggregateType aggregate, BucketPlan plan, IEnumerable samples, DateTime now) + { + ArgumentNullException.ThrowIfNull(samples); + var acc = new TimeWeightedAggregator(aggregate, plan); + foreach (var s in samples) acc.Add(s); + return acc.Finish(now); + } + + /// Feeds the next sample (ascending time; a sample at or before the previous one is ignored). + /// The sample. + public void Add(StoredSample s) + { + if (_finished) throw new InvalidOperationException("The aggregation is already finished."); + var t = s.TimeUtc.Ticks; + if (_prevAny is { } last && t <= last.TimeUtc.Ticks) return; + + // Interpolative: classify every bucket start before this sample by the simple-bounding signal there. + ClassifyPointsBefore(t); + + // Simple-bounding segment [prevAny, s). + if (_prevAny is { IsUsable: true } p) + { + // Toward a usable sample: sloped or stepped; toward a Bad one: held up to the outage. + var sloped = s.IsUsable && IsSloped(p, s); + AddRaw(p, t, sloped ? s.Value.Number : p.Value.Number); + } + else if (_prevAny is { } bad) + { + AddBad(bad, t); + } + + if (!s.IsUsable) DegradePending(); // the segment the pending points sit on runs into an outage + + if (s.IsUsable) + { + if (_prevUsable is { } pu) + { + var sloped = IsSloped(pu, s); + AddBridged(pu, t, sloped ? s.Value.Number : pu.Value.Number); + } + ResolvePending(_prevUsable, s); + _prevUsable = s; + } + + _prevAny = s; + } + + /// Completes the evaluation (extrapolating past the last sample) and assembles the per-bucket results. + /// Server timestamp for the results. + /// One value per bucket, in the plan's order (descending for a reverse window). + public IReadOnlyList Finish(DateTime now) + { + if (!_finished) + { + _finished = true; + ClassifyPointsBefore(long.MaxValue); + if (_prevAny is { IsUsable: true } p && p.TimeUtc.Ticks < _end) + AddRaw(p, _end, p.Value.Number); // stepped extrapolation + else if (_prevAny is { } bad && bad.TimeUtc.Ticks < _end) + AddBad(bad, _end); // the outage runs to the window end + if (_prevUsable is { } pu && pu.TimeUtc.Ticks < _end) + AddBridged(pu, _end, pu.Value.Number); + ResolvePending(_prevUsable, next: null); + } + + var byBucket = new Dictionary((int)_plan.BucketCount); + for (long i = 0; i < _plan.BucketCount; i++) + byBucket[i] = Result(i, now); + return _plan.Assemble(byBucket, t => AggregateEvaluator.NoData(t, now)); + } + + // ------------------------------------------------------------------------------------------ Results + + private DataValueSnapshot Result(long i, DateTime now) + { + if (_aggregate == HistoryAggregateType.Interpolative) + return _points[i] is { } point + ? point with { ServerTimestampUtc = now } + : new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, PointTime(i), now); + + var bucketStart = _plan.BucketStart(i); + var duration = BucketEnd(i) - BucketStartTicks(i); + var partial = duration < _plan.IntervalTicks; // only the truncated last interval + DataValueSnapshot Calc(object? value, uint status) + { + var code = HistorianInfoBits.WithCalculated(status); + if (partial && !StatusBits.IsBad(status)) code |= HistorianInfoBits.HistorianPartial; + return new DataValueSnapshot(value, code, bucketStart, now); + } + + if (_invalid[i]) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs); + if (_usableTicks[i] == 0) + return _badTicks[i] > 0 ? Calc(null, HistorianStatusCodes.Bad) : AggregateEvaluator.NoData(bucketStart, now); + + var status = _usableTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal; + return _aggregate switch + { + HistoryAggregateType.TimeAverage => Calc(_bridgedArea[i] / _bridgedTicks[i], status), + HistoryAggregateType.Total => Calc(_bridgedArea[i] / _bridgedTicks[i] * (duration / TicksPerSecond), status), + HistoryAggregateType.TimeAverage2 => Calc(_rawArea[i] / _usableTicks[i], status), + HistoryAggregateType.Total2 => Calc(_rawArea[i] / TicksPerSecond, status), + _ => throw new HistoryAggregateNotSupportedException(_aggregate), + }; + } + + // ------------------------------------------------------------------------------------------ Segments + + private static bool IsNumeric(StoredSample s) => + s.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64 or HistorianValueKind.Boolean; + + /// Sloped between two usable samples only when both are Double/Int64; anything Boolean is stepped. + private static bool IsSloped(StoredSample a, StoredSample b) => + a.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64 + && b.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64; + + private long BucketStartTicks(long i) => _start + i * _plan.IntervalTicks; + + private long BucketEnd(long i) => Math.Min(BucketStartTicks(i) + _plan.IntervalTicks, _end); + + private void AddRaw(StoredSample from, long toTicks, double toValue) => + Distribute(from, toTicks, toValue, (i, area, ticks) => + { + _rawArea[i] += area; + _usableTicks[i] += ticks; + }); + + /// Counts [from.t, toTicks) as outage (Bad) time; no value, never marks a bucket invalid. + private void AddBad(StoredSample from, long toTicks) + { + var lo = Math.Max(from.TimeUtc.Ticks, _start); + var hi = Math.Min(toTicks, _end); + if (lo >= hi || _plan.BucketCount == 0) return; + for (var i = (lo - _start) / _plan.IntervalTicks; i < _plan.BucketCount; i++) + { + var bs = BucketStartTicks(i); + if (bs >= hi) break; + var a = Math.Max(lo, bs); + var b = Math.Min(hi, BucketEnd(i)); + if (a < b) _badTicks[i] += b - a; + } + } + + private void AddBridged(StoredSample from, long toTicks, double toValue) => + Distribute(from, toTicks, toValue, (i, area, ticks) => + { + _bridgedArea[i] += area; + _bridgedTicks[i] += ticks; + }); + + /// + /// Splits the linear segment [from.t, toTicks) (value from.v → toValue) at the bucket edges, clipped to + /// the window, and hands each bucket its trapezoid area (value·ticks) and duration. A non-numeric start sample + /// marks every bucket it touches invalid instead. + /// + private void Distribute(StoredSample from, long toTicks, double toValue, Action add) + { + var t0 = from.TimeUtc.Ticks; + var lo = Math.Max(t0, _start); + var hi = Math.Min(toTicks, _end); + if (lo >= hi || _plan.BucketCount == 0) return; + + var numeric = IsNumeric(from); + var v0 = from.Value.Number; + var span = (double)(toTicks - t0); + double ValueAt(long x) => toValue == v0 ? v0 : v0 + (toValue - v0) * ((x - t0) / span); + + for (var i = (lo - _start) / _plan.IntervalTicks; i < _plan.BucketCount; i++) + { + var bs = BucketStartTicks(i); + if (bs >= hi) break; + var a = Math.Max(lo, bs); + var b = Math.Min(hi, BucketEnd(i)); + if (a >= b) continue; + if (!numeric) { _invalid[i] = true; continue; } + add(i, (ValueAt(a) + ValueAt(b)) / 2d * (b - a), b - a); + } + } + + // ------------------------------------------------------------------------------------------ Interpolative + + /// + /// The time of Interpolative point (ascending time). Forward: the bucket starts + /// start + j·interval. Reverse (Part 13 anchors a reverse read at its later time, the request's start): + /// End − (n−1−j)·interval, i.e. End, End − interval, … once has + /// reversed the order — so a reverse read 10:00 → 09:00 at 1 h yields the value at 10:00. + /// + private long PointTicks(long j) => + _plan.Reversed ? _end - (_plan.BucketCount - 1 - j) * _plan.IntervalTicks : BucketStartTicks(j); + + private DateTime PointTime(long j) => new(PointTicks(j), DateTimeKind.Utc); + + /// Classifies every unclassified point strictly before . + private void ClassifyPointsBefore(long beforeTicks) + { + if (_aggregate != HistoryAggregateType.Interpolative) return; + for (; _nextPoint < _plan.BucketCount; _nextPoint++) + { + var a = PointTicks(_nextPoint); + if (a >= beforeTicks) break; + + if (_prevAny is not { } p) + { + _pending.Add(new PendingPoint(_nextPoint, a, PointQuality.NoPrior)); + } + else if (p.IsUsable && p.TimeUtc.Ticks == a) + { + // A usable sample exactly at the interval start: its own value and status. + _points[_nextPoint] = IsNumeric(p) + ? new DataValueSnapshot(p.Value.Number, p.Status, PointTime(_nextPoint), default) + : InvalidPoint(_nextPoint); + } + else if (!p.IsUsable) + { + _pending.Add(new PendingPoint(_nextPoint, a, PointQuality.Degraded)); + } + else + { + // Inside a segment that starts at a usable sample; its quality is only known once the next sample + // arrives (beforeTicks is that sample's time, or MaxValue at Finish ⇒ extrapolated). + _pending.Add(new PendingPoint(_nextPoint, a, PointQuality.OnSegment)); + } + } + } + + /// Resolves the pending bucket starts against the bridging pair (p, next). + private void ResolvePending(StoredSample? p, StoredSample? next) + { + if (_pending.Count == 0) return; + foreach (var pt in _pending) + { + var bucketStart = PointTime(pt.Bucket); + if (p is not { } prev || pt.Quality == PointQuality.NoPrior) + { + _points[pt.Bucket] = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, bucketStart, default); + continue; + } + if (!IsNumeric(prev) || next is { } nn && !IsNumeric(nn)) + { + _points[pt.Bucket] = InvalidPoint(pt.Bucket); + continue; + } + + // Good only on a segment that runs straight from prev to next (no outage skipped, not extrapolated), and only + // when the samples that define the value are Good: both ends when sloped, prev alone when stepped. + double value; + bool good; + if (next is { } n && IsSloped(prev, n)) + { + value = prev.Value.Number + (n.Value.Number - prev.Value.Number) + * ((double)(pt.Ticks - prev.TimeUtc.Ticks) / (n.TimeUtc.Ticks - prev.TimeUtc.Ticks)); + good = pt.Quality == PointQuality.OnSegment && prev.IsGood && n.IsGood; + } + else + { + value = prev.Value.Number; + good = pt.Quality == PointQuality.OnSegment && next is not null && prev.IsGood; + } + var status = good ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal; + _points[pt.Bucket] = new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), bucketStart, default); + } + _pending.Clear(); + } + + /// Marks every pending on-segment point degraded: its segment ends at a Bad sample. + private void DegradePending() + { + for (var k = 0; k < _pending.Count; k++) + if (_pending[k].Quality == PointQuality.OnSegment) + _pending[k] = _pending[k] with { Quality = PointQuality.Degraded }; + } + + private DataValueSnapshot InvalidPoint(long bucket) => + new(null, HistorianStatusCodes.BadAggregateInvalidInputs, PointTime(bucket), default); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs index 050cb231..4beb6b15 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs @@ -91,6 +91,36 @@ internal static class SqlAggregateQueries "SELECT st.*, r.NumValue, r.SourceTimeUtc " + "FROM st LEFT JOIN r ON r.b = st.b AND r.rn = 1 ORDER BY st.b;"; + /// + /// Time-weighted family (#671: TimeAverage / TimeAverage2 / Total / Total2 / Interpolative) — not pushed down. + /// One batch of five result sets, each SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode: + /// (1) the latest usable sample before @start, (2) the latest sample of any quality before @start, + /// (3) the samples in [@start, @end) ascending, at most @maxRows (the raw-row cap + 1), (4) the first sample of any quality at or after + /// @end, (5) the first usable one at or after @end. Each bound is a TOP 1 seek on the + /// (TagId, SourceTimeUtc) key. The caller streams them, in that order, into + /// (shared with QuestDB, so both backends agree exactly). + /// + /// + /// Trade-off: the window's rows travel to the server instead of one row per bucket. Part 13's per-segment + /// interpolation, bucket-edge splitting and the two bounding-value definitions would otherwise have to be written + /// twice (T-SQL and QuestDB SQL) and kept bit-identical; evaluating once in C# keeps the backends equal by + /// construction. Memory stays O(buckets): the rows are streamed, never materialised. + /// + /// Schema-qualified Sample table name. + /// The SQL batch. + public static string TimeWeighted(string sample) + { + const string cols = "SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode"; + const string usable = " AND StatusCode >= 0 AND ValueKind <> 0"; + var key = $"FROM {sample} WHERE TagId = @id AND "; + return + $"SELECT TOP 1 {cols} {key}SourceTimeUtc < @start{usable} ORDER BY SourceTimeUtc DESC; " + + $"SELECT TOP 1 {cols} {key}SourceTimeUtc < @start ORDER BY SourceTimeUtc DESC; " + + $"SELECT TOP (@maxRows) {cols} {key}SourceTimeUtc >= @start AND SourceTimeUtc < @end ORDER BY SourceTimeUtc ASC; " + + $"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end ORDER BY SourceTimeUtc ASC; " + + $"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end{usable} ORDER BY SourceTimeUtc ASC;"; + } + /// The s CTE: the window's samples with bucket index and the usable/good flags. private static string Classified(string sample) => "WITH s AS (SELECT (DATEDIFF_BIG(microsecond, @start, SourceTimeUtc) * 10 " + diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index c580cdea..0dbb7f05 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -59,6 +59,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource private readonly TimeProvider _time; private readonly HistorianHealthCounters _health = new(); + private readonly int _maxTimeWeightedRawRows; /// Creates the data source over the provider's shared connection, bootstrapper and tag catalog. /// Connection helper. @@ -66,10 +67,12 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// Tag catalog (shared; reads use only its read-only lookup). /// Optional logger. /// Clock for "now" (AtTime beyond the last sample, server timestamps); system by default. + /// Raw in-window row cap of one time-weighted read (<= 0 = unbounded). internal SqlServerHistorianDataSource( SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null, - TimeProvider? timeProvider = null) + TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) { + _maxTimeWeightedRawRows = maxTimeWeightedRawRows; _conn = conn ?? throw new ArgumentNullException(nameof(conn)); _bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap)); _catalog = catalog ?? throw new ArgumentNullException(nameof(catalog)); @@ -298,10 +301,12 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is /// computed over NumValue (Boolean = 0/1). These are the simple per-sample - /// definitions, not Part 13's interval-boundary (interpolated bound / time-weighted) semantics. + /// definitions. The Part 13 time-weighted family (TimeAverage, TimeAverage2, Total, Total2, Interpolative, #671) is + /// not pushed down: the window's rows plus four bounding rows are streamed into the shared + /// , which fills every interval (an empty one between bounds gets a value). /// /// - /// The aggregate is not one of the thirteen served. + /// The aggregate is not one of the seventeen served. /// The window holds more than intervals. public async Task ReadProcessedAsync( string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, @@ -316,6 +321,14 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource var now = _time.GetUtcNow().UtcDateTime; var evaluated = new Dictionary(); var tagId = plan.BucketCount == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false); + if (family == AggregateFamily.TimeWeighted) + { + var tw = new TimeWeightedAggregator(aggregate, plan); + if (tagId is { } twId) await StreamTimeWeightedAsync(twId, fullReference, plan, tw, cancellationToken).ConfigureAwait(false); + var twResult = tw.Finish(now); + RecordSuccess(); + return new HistoryReadResult(twResult, ContinuationPoint: null); + } if (tagId is { } id) { var sample = _conn.Q("Sample"); @@ -344,6 +357,11 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource RecordSuccess(); return new HistoryReadResult(result, ContinuationPoint: null); } + catch (HistoryReadLimitExceededException) + { + // A client-request limit (the raw-row cap), not a backend fault: not counted as a failure. + throw; + } catch (Exception ex) { RecordFailure(ex); @@ -351,6 +369,34 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource } } + /// + /// Streams the time-weighted batch () into : the + /// two before-bounds, the window, the two after-bounds — ascending, with duplicates dropped by the aggregator. + /// + private async Task StreamTimeWeightedAsync(int id, string tag, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct) + { + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, SqlAggregateQueries.TimeWeighted(_conn.Q("Sample"))); + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id }); + SqlHistorianConnection.AddDateTime2(cmd, "@start", DateTime.SpecifyKind(plan.StartUtc, DateTimeKind.Utc)); + SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(plan.EndUtc, DateTimeKind.Utc)); + cmd.Parameters.Add(new SqlParameter("@maxRows", SqlDbType.Int) + { Value = TimeWeightedAggregator.WindowFetchLimit(_maxTimeWeightedRawRows) }); + await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + var resultSet = 0; + do + { + long windowRows = 0; + while (await reader.ReadAsync(ct).ConfigureAwait(false)) + { + // Result set 2 is the window (TOP (@maxRows) = cap + 1): one row past the cap fails the node. + if (resultSet == 2) TimeWeightedAggregator.CheckWindowRows(++windowRows, _maxTimeWeightedRawRows, tag); + tw.Add(ReadStored(reader, 0)); + } + resultSet++; + } while (await reader.NextResultAsync(ct).ConfigureAwait(false)); + } + private static BucketRow ReadBucket(SqlDataReader reader, AggregateFamily family) { static double? D(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs index 8076a9f3..8b6b5bcd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs @@ -1,3 +1,4 @@ +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; using System.Collections.Frozen; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; @@ -29,11 +30,16 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider /// Creates the provider over the ServerHistorian:SqlServer options (validated elsewhere). /// The ServerHistorian:SqlServer options. - public SqlServerHistorianProvider(SqlServerHistorianOptions options) + /// The shared ServerHistorian:MaxTimeWeightedRawRows cap (<= 0 = unbounded). + public SqlServerHistorianProvider(SqlServerHistorianOptions options, + int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) { _options = options ?? throw new ArgumentNullException(nameof(options)); + _maxTimeWeightedRawRows = maxTimeWeightedRawRows; } + private readonly int _maxTimeWeightedRawRows; + /// public string Name => "SqlServer"; @@ -63,7 +69,7 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider { var s = Get(sp); return new SqlServerHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog, - Loggers(sp).CreateLogger()); + Loggers(sp).CreateLogger(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs index a14e5bd6..52056ba9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs @@ -41,8 +41,8 @@ public static class HistorianBackendRegistry new Dictionary>(StringComparer.OrdinalIgnoreCase) { [ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o), - [SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)), - [QuestDbBackendName] = (_, c) => new QuestDbHistorianProvider(BindQuestDb(c)), + [SqlServerBackendName] = (o, c) => new SqlServerHistorianProvider(BindSqlServer(c), o.MaxTimeWeightedRawRows), + [QuestDbBackendName] = (o, c) => new QuestDbHistorianProvider(BindQuestDb(c), o.MaxTimeWeightedRawRows), }; /// The registered backend names, for validation messages. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index e9a9a123..bb0ea345 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5878,6 +5878,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 errors[handle.Index] = StatusCodes.BadAggregateNotSupported; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadAggregateNotSupported }; } + catch (HistoryReadLimitExceededException ex) + { + // The read would exceed a backend size limit (#671: the time-weighted raw-row cap) — refuse THIS node + // explicitly instead of a generic fault or a deadline-elapsed BadTimeout. + _logger.LogWarning("OtOpcUaNodeManager: HistoryRead limit exceeded for node {NodeId}: {Message}", handle.NodeId, ex.Message); + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; + } catch (Exception ex) { // One node's backend failure (throw / timeout) must not throw out of the batch — surface a @@ -6174,6 +6182,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (aggregateNodeId == ObjectIds.AggregateFunction_AnnotationCount) return HistoryAggregateType.AnnotationCount; if (aggregateNodeId == ObjectIds.AggregateFunction_StandardDeviationPopulation) return HistoryAggregateType.StandardDeviationPopulation; if (aggregateNodeId == ObjectIds.AggregateFunction_VariancePopulation) return HistoryAggregateType.VariancePopulation; + if (aggregateNodeId == ObjectIds.AggregateFunction_TimeAverage2) return HistoryAggregateType.TimeAverage2; + if (aggregateNodeId == ObjectIds.AggregateFunction_Total2) return HistoryAggregateType.Total2; return null; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs index c9db04a0..fc6cc012 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -112,6 +112,16 @@ public sealed class ServerHistorianOptions /// public TimeSpan HistoryReadDeadline { get; init; } = TimeSpan.FromSeconds(60); + /// + /// SqlServer / QuestDb backends only (#671): the most raw in-window samples one Part 13 + /// time-weighted read (TimeAverage, TimeAverage2, Total, Total2, Interpolative) may stream into the server-side + /// evaluator — that family is not pushed down to SQL. A read past the cap stops at cap + 1 rows and fails THAT + /// node with BadTooManyOperations instead of holding a pooled connection until + /// . Non-positive = unbounded (a warning). Default + /// 1 000 000 (≈ 11.5 days of a 1 Hz tag). + /// + public int MaxTimeWeightedRawRows { get; init; } = 1_000_000; + /// Returns operator-facing misconfiguration warnings for an Enabled historian /// (empty when disabled or correctly configured). Pure — the registration logs each entry. /// @@ -154,6 +164,8 @@ public sealed class ServerHistorianOptions warnings.Add($"ServerHistorian:HistoryReadBatchParallelism is {HistoryReadBatchParallelism} — must be > 0; HistoryRead falls back to serving each node sequentially."); if (MaxConcurrentHistoryReadBatches <= 0) warnings.Add($"ServerHistorian:MaxConcurrentHistoryReadBatches is {MaxConcurrentHistoryReadBatches} — must be > 0; the process-wide HistoryRead batch limiter is disabled."); + if (MaxTimeWeightedRawRows <= 0) + warnings.Add($"ServerHistorian:MaxTimeWeightedRawRows is {MaxTimeWeightedRawRows} — must be > 0; time-weighted aggregate reads on the SqlServer/QuestDb backends stream an unbounded number of raw samples."); if (HistoryReadDeadline <= TimeSpan.Zero) warnings.Add($"ServerHistorian:HistoryReadDeadline is {HistoryReadDeadline} — must be > 0; HistoryRead requests run unbounded (no per-request deadline)."); return warnings; diff --git a/tests/Drivers/Historian.Conformance/IConformanceBackend.cs b/tests/Drivers/Historian.Conformance/IConformanceBackend.cs index 55077224..7e817533 100644 --- a/tests/Drivers/Historian.Conformance/IConformanceBackend.cs +++ b/tests/Drivers/Historian.Conformance/IConformanceBackend.cs @@ -22,6 +22,10 @@ public interface IConformanceBackend : IAsyncDisposable /// Its "now"; null means the system clock. IHistorianDataSource CreateDataSource(TimeProvider? clock = null); + /// A fresh data source whose time-weighted raw-row cap (ServerHistorian:MaxTimeWeightedRawRows) is . + /// The cap (<= 0 = unbounded). + IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows); + /// A fresh live-value writer over this backend's store. /// The clock used to stamp values that carry no timestamp; null means the system clock. IHistorianValueWriter CreateValueWriter(Func? utcNow = null); diff --git a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs index cd579922..9ce58c18 100644 --- a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs +++ b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs @@ -39,7 +39,6 @@ public abstract class ProcessedConformance : ConformanceBase [InlineData(HistoryAggregateType.Average, 15d, 40d, 50d)] [InlineData(HistoryAggregateType.Minimum, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.Maximum, 20d, 40d, 50d)] - [InlineData(HistoryAggregateType.Total, 30d, 40d, 50d)] [InlineData(HistoryAggregateType.Start, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.End, 20d, 40d, 50d)] [InlineData(HistoryAggregateType.Delta, 10d, 0d, 0d)] @@ -121,8 +120,8 @@ public abstract class ProcessedConformance : ConformanceBase (edge, HistorianValue.FromDouble(4), Good), // first tick of B1 (At(40).AddTicks(-B.ResolutionTicks), HistorianValue.FromDouble(8), Good), // last tick of B1 (At(40), HistorianValue.FromDouble(2000), Good)); // == end: excluded - var s = (await B.CreateDataSource().ReadProcessedAsync("E", At(0), At(40), Twenty, HistoryAggregateType.Total, Ct)).Samples; - s.Select(x => x.Value).ShouldBe([3d, 12d]); + var s = (await B.CreateDataSource().ReadProcessedAsync("E", At(0), At(40), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Select(x => x.Value).ShouldBe([1.5d, 6d]); s.ShouldAllBe(x => x.StatusCode == GoodCalc); } @@ -164,11 +163,15 @@ public abstract class ProcessedConformance : ConformanceBase (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.Count, Ct)) .Samples.Single().Value.ShouldBe(4); // Also a whole-DateTime-range window (0001 → 9999-12-31T23:59:59.9999999). - (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Total, Ct)) - .Samples.Single().Value.ShouldBe(120d); + (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); // An interval wider than the window (here TimeSpan.MaxValue) is the same single interval, no overflow. - (await src.ReadProcessedAsync("A", start, end, TimeSpan.MaxValue, HistoryAggregateType.Total, Ct)) - .Samples.Single().Value.ShouldBe(120d); + (await src.ReadProcessedAsync("A", start, end, TimeSpan.MaxValue, HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); + // The time-weighted path over the whole range: ≈ the extrapolated 50 that dominates eight millennia. + var tw = (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.TimeAverage2, Ct)).Samples.Single(); + ((double)tw.Value!).ShouldBe(50d, 1e-6); + tw.StatusCode.ShouldBe(UncertainCalc); src.GetHealthSnapshot().TotalFailures.ShouldBe(0); } @@ -303,7 +306,8 @@ public abstract class ProcessedConformance : ConformanceBase async Task One(HistoryAggregateType a) => (await src.ReadProcessedAsync("B", At(0), At(20), Twenty, a, Ct)).Samples.Single(); (await One(HistoryAggregateType.Average)).Value.ShouldBe(0.75d); - (await One(HistoryAggregateType.Total)).Value.ShouldBe(3d); + // Total is time-weighted (#671): stepped 1 over [0,5), 0 over [5,10), 1 over [10,20) ⇒ 15 value·s. + (await One(HistoryAggregateType.Total)).Value.ShouldBe(15d); (await One(HistoryAggregateType.Start)).Value.ShouldBe(true); (await One(HistoryAggregateType.Count)).Value.ShouldBe(4); } @@ -317,8 +321,6 @@ public abstract class ProcessedConformance : ConformanceBase } [Theory] - [InlineData(HistoryAggregateType.Interpolative)] - [InlineData(HistoryAggregateType.TimeAverage)] [InlineData(HistoryAggregateType.StartBound)] [InlineData(HistoryAggregateType.EndBound)] [InlineData(HistoryAggregateType.DeltaBounds)] @@ -338,6 +340,174 @@ public abstract class ProcessedConformance : ConformanceBase src.GetHealthSnapshot().TotalQueries.ShouldBe(0); } + // ------------------------------------------------------------------------------------------ Time-weighted (#671) + + private const uint GoodRaw = 0x00000000u; + private const uint GoodInterp = 0x00000402u; // Good | InfoType 0x400 | Interpolated 0x2 + private const uint UncertainInterp = 0x40A40402u; // UncertainDataSubNormal | 0x402 + private const uint NoData = 0x809B0000u; + private const uint BadCalc = 0x80000401u; // generic Bad | Calculated (an all-outage interval) + + /// + /// Sparse tag W: −30 = 0 G (bound before the window), +20 = 40 G, +30 = outage marker, +50 = 20 G, +100 = 20 G + /// (bound after the window). Window [t0, t0+80s). Simple-bounding signal: [−30,20) 0→40 sloped Good (24 at +0, + /// 32 at +10); [20,30) 40 held into the outage, Uncertain; [30,50) Bad; [50,100) 20 Good. Interpolated bounding: + /// the same, but [20,50) is sloped 40→20 straight across the outage (100/3 at +30, 80/3 at +40). + /// + private Task SeedWAsync() => B.SeedAsync("W", + (At(-30), HistorianValue.FromDouble(0), Good), + (At(20), HistorianValue.FromDouble(40), Good), + (At(30), HistorianValue.Empty, BadNoCommunication), + (At(50), HistorianValue.FromDouble(20), Good), + (At(100), HistorianValue.FromDouble(20), Good)); + + private async Task> ReadW(HistoryAggregateType agg, double intervalS) => + (await B.CreateDataSource().ReadProcessedAsync("W", At(0), At(80), TimeSpan.FromSeconds(intervalS), agg, Ct)).Samples; + + private static void Values(IReadOnlyList s, params double?[] expected) + { + s.Count.ShouldBe(expected.Length); + for (var i = 0; i < expected.Length; i++) + { + if (expected[i] is { } e) ((double)s[i].Value!).ShouldBe(e, 1e-9, $"bucket {i}"); + else s[i].Value.ShouldBeNull($"bucket {i}"); + } + } + + [Theory] + // 40 s intervals (wider than the sample gaps). B0 [0,40): raw [0,20) 24→40 = 640 Good + [20,30) 40 = 400 Uncertain, + // [30,40) Bad; bridged 640 + [20,40) 40→80/3 = 2000/3. B1 [40,80): raw [50,80) 20 = 600 Good, [40,50) Bad; bridged + // [40,50) 80/3→20 = 700/3 + 600. + [InlineData(HistoryAggregateType.TimeAverage, 98d / 3d, 125d / 6d)] // (3920/3)/40, (2500/3)/40 + [InlineData(HistoryAggregateType.Total, 3920d / 3d, 2500d / 3d)] // TimeAverage × 40 s + [InlineData(HistoryAggregateType.TimeAverage2, 104d / 3d, 20d)] // 1040/30 s, 600/30 s + [InlineData(HistoryAggregateType.Total2, 1040d, 600d)] + public async Task Time_weighted_aggregates_over_a_sparse_tag_with_wide_intervals(HistoryAggregateType agg, double b0, double b1) + { + await SeedWAsync(); + var s = await ReadW(agg, 40); + Values(s, b0, b1); + // Neither interval is Good throughout (an outage in each) ⇒ Uncertain. + s.Select(x => x.StatusCode).ShouldBe([UncertainCalc, UncertainCalc]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(0)), B.Round(At(40))]); + } + + [Fact] + public async Task Interpolative_over_a_sparse_tag_with_wide_intervals() + { + await SeedWAsync(); + var s = await ReadW(HistoryAggregateType.Interpolative, 40); + // +0: from the bound before the window, 0 + 40·(30/50) = 24, Good. +40: inside the outage, bridged 80/3, Uncertain. + Values(s, 24, 80d / 3d); + s.Select(x => x.StatusCode).ShouldBe([GoodInterp, UncertainInterp]); + } + + [Fact] + public async Task TimeAverage2_over_a_sparse_tag_with_narrow_intervals() + { + await SeedWAsync(); + // 10 s intervals (narrower than the gaps): [0,10) 24→32 = 28; [10,20) 32→40 = 36; [20,30) 40 held up to the + // outage (Uncertain time, which counts as Good: TreatUncertainAsBad = false); [30,40) and [40,50) inside the outage + // ⇒ generic Bad (data, but none non-Bad); [50,60) … [70,80) the flat 20 up to the bound after, Good. + var s = await ReadW(HistoryAggregateType.TimeAverage2, 10); + Values(s, 28, 36, 40, null, null, 20, 20, 20); + s.Select(x => x.StatusCode).ShouldBe( + [GoodCalc, GoodCalc, GoodCalc, BadCalc, BadCalc, GoodCalc, GoodCalc, GoodCalc]); + // Total2 of the first interval: 28 × 10 s. + ((double)(await ReadW(HistoryAggregateType.Total2, 10))[0].Value!).ShouldBe(280d, 1e-9); + } + + [Fact] + public async Task Interpolative_over_a_sparse_tag_with_narrow_intervals() + { + await SeedWAsync(); + var s = await ReadW(HistoryAggregateType.Interpolative, 10); + // +0 24, +10 32 (sloped from the bound); +20 exact 40 (raw Good); +30 ON the outage marker ⇒ bridged 100/3; + // +40 bridged 80/3; +50 exact 20; +60, +70 on the Good 20→20 segment to the bound after. + Values(s, 24, 32, 40, 100d / 3d, 80d / 3d, 20, 20, 20); + s.Select(x => x.StatusCode).ShouldBe( + [GoodInterp, GoodInterp, GoodRaw, UncertainInterp, UncertainInterp, GoodRaw, GoodInterp, GoodInterp]); + } + + [Fact] + public async Task A_reverse_Interpolative_read_is_anchored_at_its_later_time() + { + await SeedWAsync(); + // At(80) → At(0), 40 s: points At(80) (on the Good 20→20 segment to the bound at +100 ⇒ 20) and At(40) (inside + // the outage ⇒ bridged 80/3, Uncertain), latest first. A forward read would answer At(0) and At(40) instead. + var s = (await B.CreateDataSource().ReadProcessedAsync("W", At(80), At(0), TimeSpan.FromSeconds(40), + HistoryAggregateType.Interpolative, Ct)).Samples; + s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(80)), B.Round(At(40))]); + Values(s, 20, 80d / 3d); + s.Select(x => x.StatusCode).ShouldBe([GoodInterp, UncertainInterp]); + } + + [Fact] + public async Task The_time_weighted_raw_row_cap_fails_the_read_past_the_cap_only() + { + // Tag A holds five samples in [0,80); W holds three in [0,80) plus two bound rows outside it. + await ConformanceData.SeedAAsync(B); + await SeedWAsync(); + var atCap = B.CreateDataSource(5); + (await atCap.ReadProcessedAsync("A", At(0), At(80), Twenty, HistoryAggregateType.Total, Ct)).Samples.Count.ShouldBe(4); + + var under = B.CreateDataSource(4); + var ex = await Should.ThrowAsync( + () => under.ReadProcessedAsync("A", At(0), At(80), Twenty, HistoryAggregateType.Total, Ct)); + ex.Limit.ShouldBe(4); + // Pushed-down (per-sample) aggregates are not capped; the bound rows outside the window do not count. + (await under.ReadProcessedAsync("A", At(0), At(80), Twenty, HistoryAggregateType.Average, Ct)).Samples.Count.ShouldBe(4); + (await B.CreateDataSource(3).ReadProcessedAsync("W", At(0), At(80), Twenty, HistoryAggregateType.TimeAverage, Ct)) + .Samples.Count.ShouldBe(4); + // A client-request limit, not a backend fault. + under.GetHealthSnapshot().TotalFailures.ShouldBe(0); + // Non-positive = unbounded. + (await B.CreateDataSource(0).ReadProcessedAsync("A", At(0), At(80), Twenty, HistoryAggregateType.Total, Ct)) + .Samples.Count.ShouldBe(4); + } + + [Fact] + public async Task Outages_straddling_both_window_edges_use_the_four_bound_seeks() + { + // −10 = 0 G, −5 outage, +10 = 20 G, +20 outage, +30 = 40 G; window [0,20). The four seeks return −10 (latest + // usable before), −5 (latest before), +20 (first at/after end) and +30 (first usable at/after end). + await B.SeedAsync("O", + (At(-10), HistorianValue.FromDouble(0), Good), + (At(-5), HistorianValue.Empty, BadNoCommunication), + (At(10), HistorianValue.FromDouble(20), Good), + (At(20), HistorianValue.Empty, BadNoCommunication), + (At(30), HistorianValue.FromDouble(40), Good)); + var src = B.CreateDataSource(); + async Task> Read(HistoryAggregateType a, double intervalS) => + (await src.ReadProcessedAsync("O", At(0), At(20), TimeSpan.FromSeconds(intervalS), a, Ct)).Samples; + + // Bridged: −10..10 0→20 ⇒ [0,10) mean 15 (150); 10..30 20→40 ⇒ [10,20) mean 25 (250). 400 / 20 = 20. + // (Without the usable-before seek: 250/10 = 25; without the usable-after seek: [10,20) held 20 ⇒ 17.5.) + var ta = (await Read(HistoryAggregateType.TimeAverage, 0)).Single(); + ((double)ta.Value!).ShouldBe(20d, 1e-9); + ta.StatusCode.ShouldBe(UncertainCalc); + // Simple bounding: [0,10) inside the outage from −5; [10,20) 20 held into the outage at +20 ⇒ 200 / 10 s = 20. + // (Without the any-before seek: (150+200)/20 = 17.5; without the any-after seek: [10,20) 20→30 ⇒ 25.) + var ta2 = (await Read(HistoryAggregateType.TimeAverage2, 0)).Single(); + ((double)ta2.Value!).ShouldBe(20d, 1e-9); + ta2.StatusCode.ShouldBe(UncertainCalc); + // +0 inside the outage ⇒ bridged 10, Uncertain; +10 exact 20, raw Good. + var i = await Read(HistoryAggregateType.Interpolative, 10); + Values(i, 10, 20); + i.Select(x => x.StatusCode).ShouldBe([UncertainInterp, GoodRaw]); + } + + [Fact] + public async Task Time_weighted_aggregates_on_an_unknown_tag_are_BadNoData() + { + var src = B.CreateDataSource(); + (await src.ReadProcessedAsync("nope", At(0), At(40), Twenty, HistoryAggregateType.TimeAverage, Ct)) + .Samples.ShouldAllBe(x => x.StatusCode == NoDataCalc); + (await src.ReadProcessedAsync("nope", At(0), At(40), Twenty, HistoryAggregateType.Interpolative, Ct)) + .Samples.ShouldAllBe(x => x.StatusCode == NoData); + src.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + [Fact] public async Task The_bucket_count_is_capped_at_100000() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index bf863363..1f3be1b3 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -75,6 +75,10 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) => new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock); + /// + public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) => + new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), maxTimeWeightedRawRows: maxTimeWeightedRawRows); + /// /// The T4 live-value writer over a fresh catalog (fresh positive cache) and this instance's WAL wait. public IHistorianValueWriter CreateValueWriter(Func? utcNow = null) => diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs index 6345613b..0f5c52c3 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs @@ -90,8 +90,8 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime max.Value.ShouldBe(50d); max.SourceTimestampUtc.ShouldBe(_b.Round(At(40))); // The extreme DateTime range: MinValue/MaxValue are nudged (never bound as ±infinity, which would match nothing). - (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Total, Ct)) - .Samples.Single().Value.ShouldBe(120d); + (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); // A pre-1970 window holds no data and is not an error. (await src.ReadProcessedAsync("A", start, new DateTime(1969, 1, 1, 0, 0, 0, DateTimeKind.Utc), TimeSpan.FromDays(3650), HistoryAggregateType.Count, Ct)).Samples.ShouldAllBe(x => x.StatusCode == NoDataCalc); @@ -247,7 +247,7 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime await Should.ThrowAsync(() => src.ReadEventsAsync(null, At(0), At(5), 0, Ct)); // Client-request errors are thrown before any I/O and are not backend failures. await Should.ThrowAsync(() => - src.ReadProcessedAsync("A", At(0), At(5), TimeSpan.Zero, HistoryAggregateType.Interpolative, Ct)); + src.ReadProcessedAsync("A", At(0), At(5), TimeSpan.Zero, HistoryAggregateType.StandardDeviationPopulation, Ct)); await Should.ThrowAsync(() => src.ReadProcessedAsync("A", At(0), At(100), TimeSpan.FromTicks(10), HistoryAggregateType.Count, Ct)); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs index 72a4a54b..e7953e8f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs @@ -46,7 +46,6 @@ public sealed class AggregateEvaluatorTests [InlineData(HistoryAggregateType.Average, 15d, 40d, 50d)] [InlineData(HistoryAggregateType.Minimum, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.Maximum, 20d, 40d, 50d)] - [InlineData(HistoryAggregateType.Total, 30d, 40d, 50d)] [InlineData(HistoryAggregateType.Start, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.End, 20d, 40d, 50d)] [InlineData(HistoryAggregateType.Delta, 10d, 0d, 0d)] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs index cce8846f..a12edf4f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs @@ -10,7 +10,11 @@ public sealed class ProcessedReadTests [Fact] public void Every_enum_member_is_either_served_or_refused() { - // Thirteen served; anything appended to the enum later falls to the refusal (total default). - Enum.GetValues().Count(a => AggregateFamilies.FamilyOf(a) is not null).ShouldBe(13); + // Seventeen served (#671 added TimeAverage, TimeAverage2, Interpolative and Total2, and moved Total to the + // time-weighted family); anything appended to the enum later falls to the refusal (total default). + Enum.GetValues().Count(a => AggregateFamilies.FamilyOf(a) is not null).ShouldBe(17); + Enum.GetValues().Where(a => AggregateFamilies.FamilyOf(a) == AggregateFamily.TimeWeighted) + .ShouldBe([HistoryAggregateType.Total, HistoryAggregateType.Interpolative, HistoryAggregateType.TimeAverage, + HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total2]); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/StatusCodeOracleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/StatusCodeOracleTests.cs index aa50f312..a672a621 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/StatusCodeOracleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/StatusCodeOracleTests.cs @@ -39,6 +39,14 @@ public sealed class StatusCodeOracleTests HistorianInfoBits.WithCalculated(interpolatedGood).ShouldBe(0x400u | HistorianInfoBits.HistorianCalculated); } + [Fact] + public void Partial_bit_matches_the_SDK() => + ((uint)Opc.Ua.AggregateBits.Partial).ShouldBe(HistorianInfoBits.HistorianPartial); + + [Fact] + public void Generic_bad_matches_the_SDK() => + HistorianStatusCodes.Bad.ShouldBe(Opc.Ua.StatusCodes.Bad); + [Fact] public void Info_type_is_data_value() => HistorianInfoBits.InfoTypeDataValue.ShouldBe(0x400u); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/TimeWeightedAggregatorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/TimeWeightedAggregatorTests.cs new file mode 100644 index 00000000..e93d14b2 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/TimeWeightedAggregatorTests.cs @@ -0,0 +1,430 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using static ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.Fixture; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests; + +/// +/// The Part 13 time-weighted aggregates (#671) over hand-built sample streams. Every literal is hand-computed in the +/// comment above it (areas in value·seconds; "raw" = the simple-bounding signal, "bridged" = interpolated bounding). +/// +public sealed class TimeWeightedAggregatorTests +{ + private static readonly DateTime Now = At(3600); + private const uint GoodRaw = 0x00000000u; + private const uint NoData = 0x809B0000u; + private const uint InvalidInputs = 0x80D60000u; + private const uint BadCalc = 0x80000401u; // generic Bad | Calculated (an all-outage interval) + private const uint GoodCalcPartial = 0x00000405u; // Good | Calculated | Partial (truncated last interval) + + private static IReadOnlyList Run(HistoryAggregateType agg, double from, double to, double intervalS, + params StoredSample[] samples) => + TimeWeightedAggregator.Evaluate(agg, BucketPlan.Create(At(from), At(to), TimeSpan.FromSeconds(intervalS)), samples, Now); + + private static void Values(IReadOnlyList s, params double?[] expected) + { + s.Count.ShouldBe(expected.Length); + for (var i = 0; i < expected.Length; i++) + { + if (expected[i] is { } e) ((double)s[i].Value!).ShouldBe(e, 1e-9, $"bucket {i}"); + else s[i].Value.ShouldBeNull($"bucket {i}"); + } + } + + // ---------------------------------------------------------------- tag A: +0=10 G, +10=20 G, +20=Bad, +30=40 G, +40=50 G + // Window [0,80), 20 s buckets. Raw: [0,10) 10→20 Good (150); [10,20) 20 held into the outage, Uncertain (200); + // [20,30) Bad; [30,40) 40→50 Good (450); [40,80) 50 extrapolated, Uncertain (2000). + // Bridged: [0,10) 150; [10,30) 20→40 = 250 in B0 + 350 in B1; [30,40) 450; [40,80) 50 held. + + [Fact] + public void TimeAverage_bridges_the_outage_on_tag_A() + { + // B0 (150+250)/20 = 20; B1 (350+450)/20 = 40; B2, B3 = 50. Status (SDK rule, Uncertain time counts as Good): + // B0 is non-Bad throughout (held-into-outage time is Uncertain) ⇒ Good; B1 has the Bad [20,30) ⇒ Uncertain; + // B2, B3 are extrapolated (Uncertain) throughout ⇒ Good. + var s = Run(HistoryAggregateType.TimeAverage, 0, 80, 20, A); + Values(s, 20, 40, 50, 50); + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalc, GoodCalc]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40), At(60)]); + } + + [Fact] + public void Total_is_TimeAverage_times_interval_seconds_on_tag_A() + { + // 20×20, 40×20, 50×20, 50×20. + Values(Run(HistoryAggregateType.Total, 0, 80, 20, A), 400, 800, 1000, 1000); + } + + [Fact] + public void TimeAverage2_excludes_the_bad_interval_on_tag_A() + { + // B0 350/20 = 17.5; B1 450/10 = 45 (the Bad [20,30) excluded); B2, B3 = 50. + var s = Run(HistoryAggregateType.TimeAverage2, 0, 80, 20, A); + Values(s, 17.5, 45, 50, 50); + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalc, GoodCalc]); + } + + [Fact] + public void Total2_integrates_only_the_non_bad_data_on_tag_A() + { + // B0 150+200 = 350; B1 450; B2, B3 1000. + Values(Run(HistoryAggregateType.Total2, 0, 80, 20, A), 350, 450, 1000, 1000); + } + + [Fact] + public void Interpolative_on_tag_A() + { + // +0 exact Good 10 (raw status); +20 sits on the Bad sample ⇒ bridged 20→40 at the midpoint = 30, Uncertain; + // +40 exact Good 50; +60 past the last sample ⇒ 50 extrapolated, Uncertain. + var s = Run(HistoryAggregateType.Interpolative, 0, 80, 20, A); + Values(s, 10, 30, 50, 50); + s.Select(x => x.StatusCode).ShouldBe([GoodRaw, UncertainInterpolated, GoodRaw, UncertainInterpolated]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40), At(60)]); + s.ShouldAllBe(x => x.ServerTimestampUtc == Now); + } + + // ---------------------------------------------------------------- bounding values on both sides, crossing segments + + private static readonly StoredSample[] Bounded = + [ + S(-10, HistorianValue.FromDouble(0), Good), // bound before the window + S(10, HistorianValue.FromDouble(20), Good), + S(30, HistorianValue.FromDouble(20), Good), // bound after the window + ]; + + [Theory] + // Window [0,20), 10 s. Segment -10..10 (0→20) crosses the window start: B0 [0,10) = 10→20, mean 15. + // B1 [10,20) = 20 flat (crossing the window end into the bound). All Good. + [InlineData(HistoryAggregateType.TimeAverage, 15d, 20d)] + [InlineData(HistoryAggregateType.TimeAverage2, 15d, 20d)] + [InlineData(HistoryAggregateType.Total, 150d, 200d)] + [InlineData(HistoryAggregateType.Total2, 150d, 200d)] + public void Bounding_values_before_and_after_the_window_are_used(HistoryAggregateType agg, double b0, double b1) + { + var s = Run(agg, 0, 20, 10, Bounded); + Values(s, b0, b1); + s.ShouldAllBe(x => x.StatusCode == GoodCalc); + } + + [Fact] + public void Interpolative_interpolates_from_the_bound_before_the_window() + { + // +0: 0 + (20−0)·(10/20) = 10, Good both ends; +10 exact Good 20. + var s = Run(HistoryAggregateType.Interpolative, 0, 20, 10, Bounded); + Values(s, 10, 20); + s.Select(x => x.StatusCode).ShouldBe([GoodInterpolated, GoodRaw]); + } + + [Fact] + public void A_segment_spanning_several_buckets_is_split_at_every_edge() + { + // One segment 0..40 (0→40), window [0,40), 10 s: bucket means 5, 15, 25, 35; Total2 50, 150, 250, 350. + StoredSample[] line = [S(0, HistorianValue.FromDouble(0), Good), S(40, HistorianValue.FromDouble(40), Good)]; + Values(Run(HistoryAggregateType.TimeAverage2, 0, 40, 10, line), 5, 15, 25, 35); + Values(Run(HistoryAggregateType.Total2, 0, 40, 10, line), 50, 150, 250, 350); + Values(Run(HistoryAggregateType.Interpolative, 0, 40, 10, line), 0, 10, 20, 30); + } + + // ---------------------------------------------------------------- empty buckets between bounds + + [Fact] + public void An_empty_bucket_with_bounds_on_both_sides_is_interpolated_not_no_data() + { + // -5 = 0, +25 = 30 (slope 1/s); window [0,20), 10 s holds no sample at all. + StoredSample[] sparse = [S(-5, HistorianValue.FromDouble(0), Good), S(25, HistorianValue.FromDouble(30), Good)]; + var interp = Run(HistoryAggregateType.Interpolative, 0, 20, 10, sparse); + Values(interp, 5, 15); + interp.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + // B0 5→15 mean 10; B1 15→25 mean 20. + var avg = Run(HistoryAggregateType.TimeAverage, 0, 20, 10, sparse); + Values(avg, 10, 20); + avg.ShouldAllBe(x => x.StatusCode == GoodCalc); + Values(Run(HistoryAggregateType.Total, 0, 20, 10, sparse), 100, 200); + } + + [Fact] + public void Past_the_last_sample_the_value_is_extrapolated_stepped() + { + // -5 = 7 and nothing later: held at 7 over [0,10). The extrapolated time is Uncertain, which the SDK rule + // (TreatUncertainAsBad = false) counts as Good time ⇒ the interval is Good; the Interpolative point is Uncertain. + StoredSample[] one = [S(-5, HistorianValue.FromDouble(7), Good)]; + var s = Run(HistoryAggregateType.TimeAverage2, 0, 10, 0, one).Single(); + ((double)s.Value!).ShouldBe(7d); + s.StatusCode.ShouldBe(GoodCalc); + var i = Run(HistoryAggregateType.Interpolative, 0, 10, 0, one).Single(); + ((double)i.Value!).ShouldBe(7d); + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.TimeAverage2)] + [InlineData(HistoryAggregateType.Total)] + [InlineData(HistoryAggregateType.Total2)] + public void No_data_at_all_is_BadNoData_with_the_calculated_bit(HistoryAggregateType agg) + { + var s = Run(agg, 0, 20, 10); + s.ShouldAllBe(x => x.StatusCode == NoDataCalc && x.Value == null); + } + + [Fact] + public void Interpolative_with_no_data_is_plain_BadNoData() + { + // The only sample is at +15: nothing at or before either bucket start (+0, +10). + var s = Run(HistoryAggregateType.Interpolative, 0, 20, 10, S(15, HistorianValue.FromDouble(1), Good)); + s.ShouldAllBe(x => x.StatusCode == NoData && x.Value == null); + } + + // ---------------------------------------------------------------- Bad intervals: TimeAverage vs TimeAverage2/Total2 + + private static readonly StoredSample[] Outage = + [ + S(0, HistorianValue.FromDouble(10), Good), + S(5, HistorianValue.Empty, BadNoCommunication), + S(15, HistorianValue.FromDouble(30), Good), + S(25, HistorianValue.FromDouble(30), Good), // bound after the window + ]; + + [Fact] + public void TimeAverage2_and_Total2_exclude_the_bad_interval() + { + // Window [0,20) one bucket. Raw: [0,5) 10 held into the outage (50, Uncertain); [5,15) Bad; [15,20) 30 Good (150). + // TimeAverage2 = 200 / 10 s = 20; Total2 = 200. + var ta2 = Run(HistoryAggregateType.TimeAverage2, 0, 20, 0, Outage).Single(); + ((double)ta2.Value!).ShouldBe(20d); + ta2.StatusCode.ShouldBe(UncertainCalc); + ((double)Run(HistoryAggregateType.Total2, 0, 20, 0, Outage).Single().Value!).ShouldBe(200d); + } + + [Fact] + public void TimeAverage_and_Total_interpolate_across_the_bad_interval() + { + // Bridged: [0,15) 10→30, mean 20 ⇒ 300; [15,20) 30 ⇒ 150. TimeAverage = 450/20 = 22.5; Total = 22.5 × 20 = 450. + var ta = Run(HistoryAggregateType.TimeAverage, 0, 20, 0, Outage).Single(); + ((double)ta.Value!).ShouldBe(22.5d); + ta.StatusCode.ShouldBe(UncertainCalc); + ((double)Run(HistoryAggregateType.Total, 0, 20, 0, Outage).Single().Value!).ShouldBe(450d); + } + + [Fact] + public void An_interval_entirely_inside_an_outage_is_Bad_but_Interpolative_bridges_it() + { + // Window [10,15): raw is Bad throughout ⇒ data but no non-Bad time ⇒ generic Bad (the SDK rule; BadNoData is + // reserved for no data at all). Interpolative at +10 = 10 + 20·(10/15). + var ta = Run(HistoryAggregateType.TimeAverage, 10, 15, 0, Outage).Single(); + ta.StatusCode.ShouldBe(BadCalc); + ta.Value.ShouldBeNull(); + Run(HistoryAggregateType.TimeAverage2, 10, 15, 0, Outage).Single().StatusCode.ShouldBe(BadCalc); + Run(HistoryAggregateType.Total2, 10, 15, 0, Outage).Single().StatusCode.ShouldBe(BadCalc); + var i = Run(HistoryAggregateType.Interpolative, 10, 15, 0, Outage).Single(); + ((double)i.Value!).ShouldBe(10d + 20d * 10d / 15d, 1e-9); + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public void Interpolative_on_a_segment_that_runs_into_an_outage_is_uncertain() + { + // +2 lies between the Good +0 and the Bad +5: bridged 10→30 over [0,15) ⇒ 10 + 20·(2/15); the simple-bounding + // signal there is held into an outage, so the point is Uncertain even though both bridging ends are Good. + var i = Run(HistoryAggregateType.Interpolative, 2, 20, 0, Outage).Single(); + ((double)i.Value!).ShouldBe(10d + 20d * 2d / 15d, 1e-9); + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public void The_latest_usable_bound_is_found_behind_a_bad_bound() + { + // -10 = 0 G, -5 Bad, +10 = 20 G, +30 = 20 G. Window [0,20) one bucket. + // Bridged: -10..10 0→20 ⇒ [0,10) 10→20 (150); [10,20) 20 (200) ⇒ TimeAverage = 17.5. + // Raw: [-5,10) Bad; [10,20) 20 Good (200) ⇒ TimeAverage2 = 20. Good 10 s of 20 ⇒ Uncertain. + StoredSample[] s = + [ + S(-10, HistorianValue.FromDouble(0), Good), S(-5, HistorianValue.Empty, BadNoCommunication), + S(10, HistorianValue.FromDouble(20), Good), S(30, HistorianValue.FromDouble(20), Good), + ]; + ((double)Run(HistoryAggregateType.TimeAverage, 0, 20, 0, s).Single().Value!).ShouldBe(17.5d); + var ta2 = Run(HistoryAggregateType.TimeAverage2, 0, 20, 0, s).Single(); + ((double)ta2.Value!).ShouldBe(20d); + ta2.StatusCode.ShouldBe(UncertainCalc); + var i = Run(HistoryAggregateType.Interpolative, 0, 20, 0, s).Single(); + ((double)i.Value!).ShouldBe(10d); // inside the outage, bridged -10..10 at 0 + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + // ---------------------------------------------------------------- stepped Boolean, Uncertain, invalid inputs + + private static readonly StoredSample[] Switch = + [ + S(0, HistorianValue.FromBoolean(true), Good), + S(5, HistorianValue.FromBoolean(false), Good), + S(15, HistorianValue.FromBoolean(true), Good), + S(25, HistorianValue.FromBoolean(true), Good), // bound after the window + ]; + + [Fact] + public void Booleans_are_stepped_not_sloped() + { + // Window [0,20), 10 s. B0: 1 over [0,5), 0 over [5,10) ⇒ 0.5 (sloped would give 0.375). B1: 0 over [10,15), + // 1 over [15,20) ⇒ 0.5. All Good. Total: 5 and 5. + var s = Run(HistoryAggregateType.TimeAverage, 0, 20, 10, Switch); + Values(s, 0.5, 0.5); + s.ShouldAllBe(x => x.StatusCode == GoodCalc); + Values(Run(HistoryAggregateType.Total, 0, 20, 10, Switch), 5, 5); + // Interpolative at +10: stepped from +5 (false) ⇒ 0, Good. + var i = Run(HistoryAggregateType.Interpolative, 0, 20, 10, Switch); + Values(i, 1, 0); + i.Select(x => x.StatusCode).ShouldBe([GoodRaw, GoodInterpolated]); + } + + [Fact] + public void Uncertain_time_counts_as_good_time() + { + // TreatUncertainAsBad = false (the SDK's time-based status rule): Uncertain time counts toward PercentDataGood. + // 0 = 4 Uncertain, 10 = 6 Good, 20 = 6 Good: [0,10) sloped 4→6 ⇒ mean 5; [10,20) 6. Window [0,20), 10 s. + StoredSample[] s = + [ + S(0, HistorianValue.FromDouble(4), Uncertain), S(10, HistorianValue.FromDouble(6), Good), + S(20, HistorianValue.FromDouble(6), Good), + ]; + var r = Run(HistoryAggregateType.TimeAverage2, 0, 20, 10, s); + Values(r, 5, 6); + r.Select(x => x.StatusCode).ShouldBe([GoodCalc, GoodCalc]); + } + + [Fact] + public void Int64_samples_are_sloped_like_doubles() + { + StoredSample[] s = [S(0, HistorianValue.FromInt64(0), Good), S(10, HistorianValue.FromInt64(10), Good)]; + Values(Run(HistoryAggregateType.TimeAverage2, 0, 10, 0, s), 5); + } + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.TimeAverage2)] + [InlineData(HistoryAggregateType.Total)] + [InlineData(HistoryAggregateType.Total2)] + public void A_usable_string_sample_makes_the_interval_invalid_inputs(HistoryAggregateType agg) + { + StoredSample[] s = [S(0, HistorianValue.FromString("a"), Good), S(10, HistorianValue.FromString("b"), Good)]; + var r = Run(agg, 0, 20, 0, s).Single(); + r.StatusCode.ShouldBe(InvalidInputsCalc); + r.Value.ShouldBeNull(); + } + + [Fact] + public void Interpolative_on_a_string_is_invalid_inputs() + { + StoredSample[] s = [S(0, HistorianValue.FromString("a"), Good), S(10, HistorianValue.FromString("b"), Good)]; + Run(HistoryAggregateType.Interpolative, 0, 20, 10, s).ShouldAllBe(x => x.StatusCode == InvalidInputs); + } + + // ---------------------------------------------------------------- plan shape + + [Fact] + public void A_reverse_window_is_computed_forward_and_returned_descending() + { + var s = TimeWeightedAggregator.Evaluate(HistoryAggregateType.TimeAverage2, + BucketPlan.Create(At(80), At(0), TimeSpan.FromSeconds(20)), A, Now); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(60), At(40), At(20), At(0)]); + Values(s, 50, 50, 45, 17.5); + } + + [Fact] + public void A_truncated_last_bucket_uses_its_own_duration_for_Total_and_is_Partial() + { + // Window [0,50), 20 s: B2 = [40,50) is 10 s of the extrapolated 50 ⇒ TimeAverage 50, Total 500. + var s = Run(HistoryAggregateType.Total, 0, 50, 20, A); + Values(s, 400, 800, 500); + // Only the truncated interval carries the Partial bit. + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalcPartial]); + } + + [Fact] + public void An_outage_with_no_later_sample_is_Bad_to_the_window_end_not_no_data() + { + // 0 = 10 G, then an outage marker at +5 and nothing after: [10,20) lies wholly inside the still-open outage. + StoredSample[] open = [S(0, HistorianValue.FromDouble(10), Good), S(5, HistorianValue.Empty, BadNoCommunication)]; + Run(HistoryAggregateType.TimeAverage2, 10, 20, 0, open).Single().StatusCode.ShouldBe(BadCalc); + } + + [Fact] + public void A_bad_or_no_data_truncated_interval_carries_no_partial_bit() + { + // Window [0,15), 10 s: B1 = [10,15) is inside the outage ⇒ Bad without Partial. + Run(HistoryAggregateType.TimeAverage2, 0, 15, 10, Outage)[1].StatusCode.ShouldBe(BadCalc); + Run(HistoryAggregateType.TimeAverage2, 0, 15, 10)[1].StatusCode.ShouldBe(NoDataCalc); + } + + // ---------------------------------------------------------------- reverse Interpolative (Part 13: anchored at the later time) + + [Fact] + public void A_reverse_Interpolative_read_is_anchored_at_its_later_time() + { + // Request At(20) → At(0), 20 s: one point, at At(20) (not At(0)). Bounded: segment 10..30 is 20 flat, Good. + // A forward read of the same window would give the value at At(0) = 10. + var s = TimeWeightedAggregator.Evaluate(HistoryAggregateType.Interpolative, + BucketPlan.Create(At(20), At(0), TimeSpan.FromSeconds(20)), Bounded, Now); + s.Single().SourceTimestampUtc.ShouldBe(At(20)); + ((double)s.Single().Value!).ShouldBe(20d); + s.Single().StatusCode.ShouldBe(GoodInterpolated); + } + + [Fact] + public void A_reverse_Interpolative_read_steps_back_from_the_later_time_latest_first() + { + // Tag A, At(80) → At(0), 20 s: points 80, 60, 40, 20 (latest first). 80 and 60 are past the last sample ⇒ 50 + // extrapolated, Uncertain; 40 exact Good 50; 20 sits on the Bad sample ⇒ bridged 30, Uncertain. + var s = TimeWeightedAggregator.Evaluate(HistoryAggregateType.Interpolative, + BucketPlan.Create(At(80), At(0), TimeSpan.FromSeconds(20)), A, Now); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(80), At(60), At(40), At(20)]); + Values(s, 50, 50, 50, 30); + s.Select(x => x.StatusCode).ShouldBe([UncertainInterpolated, UncertainInterpolated, GoodRaw, UncertainInterpolated]); + } + + [Fact] + public void A_reverse_Interpolative_read_with_a_partial_interval_ends_inside_the_window() + { + // At(25) → At(0), 10 s: three intervals ⇒ points 25, 15, 5 (never before the window start). + // Bounded: -10..10 slope 1/s from 0 ⇒ +5 = 15; 10..30 flat 20 ⇒ +15 = 20, +25 = 20. + var s = TimeWeightedAggregator.Evaluate(HistoryAggregateType.Interpolative, + BucketPlan.Create(At(25), At(0), TimeSpan.FromSeconds(10)), Bounded, Now); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(25), At(15), At(5)]); + Values(s, 20, 20, 15); + } + + // ---------------------------------------------------------------- raw-row cap helpers + + [Theory] + [InlineData(1_000_000, 1_000_001)] + [InlineData(1, 2)] + [InlineData(0, int.MaxValue)] + [InlineData(-5, int.MaxValue)] + [InlineData(int.MaxValue, int.MaxValue)] + public void The_window_fetch_limit_is_one_past_the_cap(int cap, int expected) => + TimeWeightedAggregator.WindowFetchLimit(cap).ShouldBe(expected); + + [Fact] + public void The_window_row_check_fails_only_past_a_positive_cap() + { + TimeWeightedAggregator.CheckWindowRows(4, 4, "T"); + TimeWeightedAggregator.CheckWindowRows(5_000_000, 0, "T"); + var ex = Should.Throw(() => TimeWeightedAggregator.CheckWindowRows(5, 4, "T")); + ex.Limit.ShouldBe(4); + } + + [Fact] + public void A_duplicate_bound_is_ignored() + { + // The same row fetched as both "latest usable" and "latest any" bound. + var dup = new[] { Bounded[0], Bounded[0], Bounded[1], Bounded[2], Bounded[2] }; + Values(Run(HistoryAggregateType.TimeAverage2, 0, 20, 10, dup), 15, 20); + } + + [Fact] + public void A_non_time_weighted_aggregate_is_refused() + { + Should.Throw(() => + new TimeWeightedAggregator(HistoryAggregateType.Average, BucketPlan.Create(At(0), At(10), TimeSpan.Zero))); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs index 8af5e084..0fd95144 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs @@ -44,8 +44,9 @@ internal sealed class ReadTestDb } /// A fresh data source (fresh catalog cache) over this schema; is its "now". - public SqlServerHistorianDataSource Source(TimeProvider? clock = null) => - new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock); + public SqlServerHistorianDataSource Source(TimeProvider? clock = null, + int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) => + new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock, maxTimeWeightedRawRows: maxTimeWeightedRawRows); /// Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart). public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs index e4c27a8c..79b71ce4 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs @@ -25,6 +25,9 @@ internal sealed class SqlServerConformanceBackend : IConformanceBackend /// public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) => _db.Source(clock); + /// + public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) => _db.Source(null, maxTimeWeightedRawRows); + /// public IHistorianValueWriter CreateValueWriter(Func? utcNow = null) => new SqlServerHistorianValueWriter(_db.Conn, _db.Boot, new SqlServerTagCatalog(_db.Conn, _db.Boot), utcNow: utcNow); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs index a2950a7d..f9eebd88 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -292,6 +292,34 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// #671: the Part 13 TimeAverage2 / Total2 aggregate NodeIds map to their own members. + [Theory] + [InlineData("TimeAverage2")] + [InlineData("Total2")] + public async Task Processed_TimeAverage2_and_Total2_dispatch_to_the_source(string name) + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() }; + nm.HistorianDataSource = fake; + nm.EnsureVariable("eq-1/tw", parentFolderNodeId: null, displayName: "T", dataType: "Float", + writable: false, historianTagname: "WW.Tw", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/tw")!.NodeId; + var (aggregateId, expected) = name == "Total2" + ? (ObjectIds.AggregateFunction_Total2, HistoryAggregateType.Total2) + : (ObjectIds.AggregateFunction_TimeAverage2, HistoryAggregateType.TimeAverage2); + + InvokeHistoryRead(server, nm, new ReadProcessedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), EndTime = DateTime.UtcNow, ProcessingInterval = 10_000.0, + AggregateType = new NodeIdCollection { aggregateId }, + }, nodeId); + + fake.LastCall.ShouldBe("Processed"); + fake.LastAggregate.ShouldBe(expected); + await host.DisposeAsync(); + } + /// #285: a backend that cannot serve a mapped aggregate answers BadAggregateNotSupported /// (not the generic BadHistoryOperationUnsupported a plain throw produces). [Fact] @@ -314,6 +342,28 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// #671: a read past a backend size limit (the time-weighted raw-row cap) answers BadTooManyOperations + /// for that node (not the generic BadHistoryOperationUnsupported, nor a deadline BadTimeout). + [Fact] + public async Task Processed_limit_exceeded_yields_BadTooManyOperations() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.HistorianDataSource = new RecordingHistorianDataSource { LimitExceeded = true }; + nm.EnsureVariable("eq-1/cap", parentFolderNodeId: null, displayName: "C", dataType: "Float", + writable: false, historianTagname: "WW.Cap", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/cap")!.NodeId; + + var (_, errors) = InvokeHistoryRead(server, nm, new ReadProcessedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), EndTime = DateTime.UtcNow, ProcessingInterval = 10_000.0, + AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Total }, + }, nodeId); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations); + await host.DisposeAsync(); + } + /// AtTime read: the requested timestamps reach ReadAtTimeAsync in order. [Fact] public async Task AtTime_dispatches_requested_timestamps() @@ -477,6 +527,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable { public bool ThrowOnRead { get; init; } public bool AggregateNotSupported { get; init; } + public bool LimitExceeded { get; init; } public HistorianRead RawResult { get; set; } = new(Array.Empty(), null); public HistorianRead ProcessedResult { get; set; } = new(Array.Empty(), null); public HistorianRead AtTimeResult { get; set; } = new(Array.Empty(), null); @@ -509,6 +560,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable { if (ThrowOnRead) throw new InvalidOperationException("backend boom"); if (AggregateNotSupported) throw new HistoryAggregateNotSupportedException(aggregate); + if (LimitExceeded) throw new HistoryReadLimitExceededException(4, "too many raw rows"); LastCall = "Processed"; LastTagname = fullReference; LastStart = startUtc; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs index 458070cd..4f38cc8a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs @@ -89,6 +89,16 @@ public sealed class ServerHistorianOptionsTests Assert.Contains(w, m => m.Contains("HistoryReadDeadline")); } + [Fact] + public void MaxTimeWeightedRawRows_defaults_to_one_million_and_warns_when_nonpositive() + { + Assert.Equal(1_000_000, new ServerHistorianOptions().MaxTimeWeightedRawRows); + var w = new ServerHistorianOptions + { Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", MaxTimeWeightedRawRows = 0 } + .Validate(); + Assert.Contains(w, m => m.Contains("MaxTimeWeightedRawRows")); + } + [Fact] public void Disabled_with_nonpositive_HistoryRead_knobs_is_silent() {