diff --git a/CLAUDE.md b/CLAUDE.md index 89e33b34..d2cbe1c4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1345,6 +1345,7 @@ historian tagname, which is the tag's driver `FullName`. - **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**; 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". +- ⚠️ **Reverse processed reads are Part 13/SDK-anchored (#677, client-visible):** every aggregate on SqlServer/QuestDB now anchors a reverse read at its later time — window `(end, start]`, partial interval earliest, each stamped with its later edge (was: earlier-anchored, stamped earlier edge); Partial also marks the interval the data starts inside (not Average); a non-default client `AggregateConfiguration` ⇒ `BadAggregateConfigurationRejected`; Gateway provisioning sends `IntegralDivisor = 1`. SDK-parity tests: `SdkAggregateParityTests`. `docs/Historian.md` §Aggregates. - **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. - **HistoryUpdate** (SqlServer only; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, gated in both ACL modes — flag off needs the `HistoryUpdate` role from a **Cluster-scoped** grant (+ key re-check), flag on uses `DataPlaneAuthorizer`'s HistoryUpdate operation; `WriteOperate` never implies it. The `HistoryWrite` AccessLevel bit is set only when an updater is wired at materialise time. `docs/Historian.md` §HistoryUpdate, `docs/security.md`. diff --git a/docs/Historian.md b/docs/Historian.md index f4c1d8e4..eea9544d 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -701,6 +701,12 @@ RetrievalMode.Integral`) — no client-side scaling. (This replaces the retired likewise a native gateway mode. Bucket status codes and timestamps are preserved unchanged; empty / null buckets surface as `BadNoData`. +**Integral unit (#677).** AVEVA's integral is `∫ value dt` (seconds) ÷ the tag's `IntegralDivisor`. The proto3 `double` default for +`HistorianTagDefinition.IntegralDivisor` is `0`, so `GatewayTagProvisioner` now sends **`IntegralDivisor = 1`** explicitly on every +`EnsureTags` — the value·seconds unit the SQL backends' `Total` uses. A tag provisioned earlier keeps its divisor until its next +`EnsureTags` (an upsert). The live magnitude check against AVEVA (a constant 1 integrated over 1 s reads 1) is still open — it needs the +VPN. Reverse processed reads on the `Gateway` backend are answered by AVEVA and are not re-anchored by #677. + ### Aggregate set and `BadAggregateNotSupported` (#285) `HistoryAggregateType` carries the wider Part 13 set (Interpolative, TimeAverage, Start, End, Delta, Range, @@ -860,21 +866,41 @@ point (see [Continuation-point paging (Events)](#continuation-point-paging-event > 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 +> ⚠️ **Release note (#677): reverse processed reads changed — client-visible.** A reverse read (`end < start`) is now anchored at the +> request's start (its *later* time) for **every** aggregate, as OPC UA Part 13 and the SDK's own `AggregateCalculator` do: intervals +> step back from `start`, the window is `(end, start]`, the truncated interval is the **earliest** one, and each result is stamped with its +> interval's **later** edge. Before #677 only `Interpolative` did this; every other aggregate swapped the window, anchored it at the +> earlier time (`[end, start)`, the partial interval at the latest end) and stamped each interval with its earlier edge. So a reverse +> 80 s → 0 s read at 20 s used to answer `[60,80) [40,60) [20,40) [0,20)` stamped 60/40/20/0; it now answers +> `(60,80] (40,60] (20,40] (0,20]` stamped 80/60/40/20 — a sample exactly on an interval edge now belongs to the interval it *ends*, a +> sample at `start` is included and one at `end` is not (the same window a reverse raw read uses). Forward reads are unchanged apart +> from the new start-of-data Partial bit below. A client that paired reverse results with forward ones by timestamp must shift by one +> interval. + +**Bucketing.** Forward: 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`, each stamped with its start. Reverse (`end < start`, +#677): the window is `(end, start]`; result *k* = `(start − (k+1)·interval, start − k·interval]`, clipped at `end` (the **earliest** +interval is the truncated one), stamped with its **later** edge `start − k·interval`, returned latest first. `interval <= 0` ⇒ a single interval `[start, end)`; `start == end` ⇒ no intervals. The bucket index is computed in **100 ns ticks** (not milliseconds): `DATEDIFF_BIG(microsecond, @start, t) × 10` plus a sub-microsecond correction from the two tick digits, divided by the interval in ticks. (`DATEDIFF` counts unit *boundaries crossed*, so a -millisecond formulation misplaces samples near a bucket edge, and a nanosecond one overflows past ≈292 years.) A window of more than -**100 000 intervals** is refused before any I/O and reported `BadHistoryOperationUnsupported` for that node. **Reverse windows** -(`end < start`) are swapped, computed forward **anchored at the earlier time**, and returned in **descending** bucket order. Every returned -status carries the historian *Calculated* info bit, `BadNoData` included. +millisecond formulation misplaces samples near a bucket edge, and a nanosecond one overflows past ≈292 years.) A reverse read adds one +constant to that elapsed-ticks dividend (`BucketPlan.BucketIndexAdjustTicks`; QuestDB folds it into its base) that re-anchors the grid at +the later time and makes the buckets later-edge-inclusive — `BucketPlan.IndexOf` is the C# statement of the math both dialects mirror. A +window of more than **100 000 intervals** is refused before any I/O and reported `BadHistoryOperationUnsupported` for that node. Every +returned status carries the historian *Calculated* info bit, `BadNoData` included. + +**Partial bit (#677).** Every aggregate except plain `Average` sets the **Partial** bit (`0x4`) on a non-Bad result for the interval the +tag's data **starts inside** — its earliest stored sample (any quality) lies strictly between the interval's edges, so there is no +bounding value before it and the data does not cover the whole interval. This is the SDK's `m_startOfData` rule (the SDK turns it off for +`Average` only). Each per-sample read therefore also fetches the tag's earliest sample (one key seek). An earliest sample exactly on an +edge, or before the interval, sets nothing. The time-weighted aggregates also keep setting Partial on the truncated interval (below). **Sample classes.** *Usable* = severity Good or Uncertain **and** a stored value (not `Empty`). *Good* = usable with severity Good. *Bad* = not usable. | Aggregate | Computed over | Value timestamp | |---|---|---| -| `Average`, `Minimum`, `Maximum` | `AVG` / `MIN` / `MAX` of `NumValue` over usable samples | bucket start | +| `Average`, `Minimum`, `Maximum` | `AVG` / `MIN` / `MAX` of `NumValue` over usable samples | bucket start (reverse read: its later edge, as every row marked "bucket start" here) | | `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 | @@ -896,11 +922,16 @@ Status and edge rules: `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. +below). Bucketing (edges, truncated interval, reverse anchoring, the 100 000-interval cap) is the same as above. -**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. +**Server defaults, fixed:** `TreatUncertainAsBad = false`, `PercentDataBad = PercentDataGood = 100`, `UseSlopedExtrapolation = false` +(`ServerAggregateConfiguration` in the OPC UA server project — the SDK's own `AggregateManager` defaults, which a test pins). **A +client's `AggregateConfiguration` (#677):** `UseServerCapabilitiesDefaults = true`, or the four values spelled out equal to these defaults +⇒ served; anything else ⇒ **`BadAggregateConfigurationRejected`** for each node of the request (the source is never called). It used to +be silently ignored, so a client asking for, say, `TreatUncertainAsBad = true` got answers computed without it. The check is server-wide +(it applies on the `Gateway` backend too). A client sending an all-zero structure without `UseServerCapabilitiesDefaults` is asking for +`PercentDataBad = PercentDataGood = 0` and is refused; the SDK's default-constructed `AggregateConfiguration` sets +`UseServerCapabilitiesDefaults = true` and is served. **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 @@ -934,9 +965,9 @@ PercentDataGood. - 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.) +All carry the Calculated bit; a non-Bad result for the **truncated interval** (shorter than the processing interval — the latest one +forward, the earliest one reverse) or for the interval the tag's **data starts inside** (#677, see *Partial bit* above) also carries the +**Partial** bit (`0x4`, `AggregateBits.Partial`). Timestamp = interval start (reverse: its later edge); value `double` (Boolean 0/1). > **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 — @@ -951,11 +982,23 @@ aggregates do not set Partial — a follow-up.) - `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. +**Reverse reads (`end < start`).** Part 13 anchors a reverse read at the request's start, which is the later time — for every aggregate +since #677. `Interpolative`'s points are `start, start − 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 use the intervals +`(start − (k+1)·interval, start − k·interval]`, the earliest one truncated, each stamped with its later edge. + +**SDK parity (#677).** `SdkAggregateParityTests` (unit, `Sql.Common.Tests`) and +`ProcessedConformance.Processed_reads_match_the_SDK_aggregate_calculators` (both SQL backends, through the real SQL) run the pinned SDK's +own calculators (`AverageAggregateCalculator`, `MinMaxAggregateCalculator`, `CountAggregateCalculator`, the base `AggregateCalculator` +for Interpolative) over the same samples, forward and reverse, and compare every interval's timestamp, value, status code and Partial +bit. Known differences that remain — documented in the test, deliberately not changed here because forward results were to stay as +they are — are: the SDK reports `TimeAverage`/`Total` Uncertain whenever any region of the interval is non-Good, where ours applies the +time-based rule (a fully-covered interval bridged across a Bad sample is Good here); `Total` over an interval the data starts inside is +`TimeAverage × interval` here but the integral of the covered part in the SDK; ours flags a truncated interval Partial, the SDK does +not; the SDK's forward end-of-data Partial rule is not implemented; the SDK's per-sample Count answers 0/Good for an empty interval +inside the data and its per-sample aggregates use Good samples only; and on a reverse `Interpolative` read the SDK marks a sample exactly +at a point Interpolated (Uncertain if a Bad sample follows) where ours — like the SDK forward — returns the raw sample. On a reverse read +the SDK's own start-of-data rule keys on the *latest* sample (the first one it is fed); ours keys on the earliest in both directions. **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) @@ -1180,8 +1223,10 @@ Every query filters `NOT deleted`; every value goes through `StoredSample`, so t - **AtTime:** previous-or-equal and next neighbour of each time by `LIMIT 1` seeks, `UNION ALL`-ed, **50 times per statement** (QuestDB 10.0.1 refuses more than 126 branches). `AtTimeEvaluator` decides each value, as on `SqlServer`. - **Events:** `[start, end)` by `ts`, ordered `(ts, event_id)`, `maxEvents + 1` fetched to produce the continuation point. -- **Processed:** one query per node; bucket index `(cast(ts AS LONG) * 10 - startUnixTicks) / intervalTicks` (never `SAMPLE BY`, whose alignment - differs), so bucket edges are identical to `SqlServer` (start-inclusive, end-exclusive, partial last bucket, interval 0 ⇒ one bucket, 100,000-bucket cap). +- **Processed:** one query per node; bucket index `(cast(ts AS LONG) * 10 - bucketBaseUnixTicks) / intervalTicks` (never `SAMPLE BY`, whose alignment + differs), so bucket edges are identical to `SqlServer` (forward start-inclusive / end-exclusive with the partial bucket last; reverse anchored at the + later time, later-edge-inclusive, partial bucket earliest — #677; interval 0 ⇒ one bucket, 100,000-bucket cap). A second `LIMIT 1` statement fetches + the tag's earliest live sample for the start-of-data Partial bit. Min/max times use `row_number()` windows with the earliest-timestamp tie-break. ### Retention (TTL) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagProvisioner.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagProvisioner.cs index b160f04f..d2e06836 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagProvisioner.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagProvisioner.cs @@ -29,6 +29,14 @@ public sealed class GatewayTagProvisioner : IHistorianProvisioning, IAsyncDispos /// private const uint DefaultStorageRateMs = 1000; + /// + /// Integral divisor stamped on every provisioned analog tag (#677). AVEVA's Total integral is + /// ∫ value dt (seconds) ÷ IntegralDivisor; the proto's double default of 0 would reach the gateway as + /// "unset" (or a divide-by-zero), so 1 is sent explicitly — the value·seconds unit the SQL backends' Total + /// uses. (The live magnitude check against AVEVA is still open: it needs the VPN.) + /// + private const double DefaultIntegralDivisor = 1.0; + private readonly IHistorianGatewayClient _client; private readonly ILogger _logger; @@ -70,6 +78,8 @@ public sealed class GatewayTagProvisioner : IHistorianProvisioning, IAsyncDispos Description = request.Description ?? string.Empty, // Required: a 0 storage rate makes the gateway's EnsureTags throw (see the constant). StorageRateMs = DefaultStorageRateMs, + // Explicit: the proto3 double default (0) is not "1" (see the constant). + IntegralDivisor = DefaultIntegralDivisor, }); } 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 85684bd9..bb398d4a 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 @@ -9,21 +9,23 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// per , mirroring the SQL Server backend's SqlAggregateQueries family split and /// column order. Every query takes five positional parameters: /// -/// $1 — the (rounded) window start in 100 ns ticks since the Unix epoch -/// (, Bigint); +/// $1 — the bucket base in 100 ns ticks since the Unix epoch +/// (, Bigint: the rounded window start forward; re-anchored at +/// the later time for a reverse read, #677); /// $2 — the bucket width in ticks (, Bigint); /// $3 — the tag (Varchar); -/// $4/$5 — the forward window [start, end) (TimestampTz via -/// ). +/// $4/$5 — the earlier / later window edges (TimestampTz via +/// ): [$4, $5) forward, ($4, $5] reverse. /// /// Each returns one row per non-empty bucket; the caller fills the empty ones. /// /// /// /// Bucketing. b = (cast(ts AS LONG) * 10 − $1) / $2. cast(ts AS LONG) is µs since 1970, so -/// × 10 is ticks since 1970; both it and $1 are µs-aligned (R6), so the integer division is -/// start-inclusive / end-exclusive to the tick, exactly like SQL Server's. Every row is at or after the start, -/// so the dividend is never negative (no truncation-toward-zero surprise), and nothing overflows before 9999. +/// × 10 is ticks since 1970, so the integer division is exact to the tick, exactly like SQL Server's +/// ( is the C# statement of the math). Every in-window row is at or after the base +/// (forward) or strictly after the window start with the base one tick past the grid origin (reverse), so the +/// dividend is never negative (no truncation-toward-zero surprise), and nothing overflows before 9999. /// SAMPLE BY is deliberately not used: its alignment rules differ from ours. /// /// @@ -60,13 +62,14 @@ internal static class QuestDbAggregateQueries /// /// Prefixed sample table name. /// The SQL text. - public static string Simple(string sample) => + /// True for a reverse read (window ($4, $5]). + public static string Simple(string sample, bool reverse = false) => "SELECT b, " + CommonCounts() + ", " + "avg(CASE WHEN u = 1 THEN num_value END) AS avg_value, " + "min(CASE WHEN u = 1 THEN num_value END) AS min_value, " + "max(CASE WHEN u = 1 THEN num_value END) AS max_value, " + "sum(CASE WHEN u = 1 THEN num_value END) AS sum_value " + - "FROM (" + Classified(sample) + ") GROUP BY b ORDER BY b"; + "FROM (" + Classified(sample, reverse) + ") GROUP BY b ORDER BY b"; /// /// Start / End / Delta: the first and last usable sample of each bucket by ts. @@ -76,8 +79,9 @@ internal static class QuestDbAggregateQueries /// /// Prefixed sample table name. /// The SQL text. - public static string StartEnd(string sample) => - "WITH s AS (" + Classified(sample) + "), " + + /// True for a reverse read (window ($4, $5]). + public static string StartEnd(string sample, bool reverse = false) => + "WITH s AS (" + Classified(sample, reverse) + "), " + "st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b), " + "f AS (SELECT b, kind, num_value, int_value, str_value, ts, " + "row_number() OVER (PARTITION BY b ORDER BY ts ASC) AS rf, " + @@ -96,8 +100,9 @@ internal static class QuestDbAggregateQueries /// Prefixed sample table name. /// True for MaximumActualTime. /// The SQL text. - public static string ActualTime(string sample, bool maximum) => - "WITH s AS (" + Classified(sample) + "), " + + /// True for a reverse read (window ($4, $5]). + public static string ActualTime(string sample, bool maximum, bool reverse = false) => + "WITH s AS (" + Classified(sample, reverse) + "), " + "st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b), " + "r AS (SELECT b, num_value, ts, " + "row_number() OVER (PARTITION BY b ORDER BY num_value " + (maximum ? "DESC" : "ASC") + ", ts ASC) AS rn " + @@ -144,12 +149,21 @@ internal static class QuestDbAggregateQueries /// The bucket-index expression (exposed for tests). internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2"; + /// + /// The tag's earliest live sample (any quality): the start-of-data Partial rule (#677). Parameter $1 tag. + /// Column: ts; no row when the tag has none. + /// + /// Prefixed sample table name. + /// The SQL text. + public static string EarliestSample(string sample) => + $"SELECT ts FROM {sample} WHERE tag = $1 AND NOT deleted ORDER BY ts ASC LIMIT 1"; + /// The classified subquery s: the window's live samples with bucket index and the usable/good flags. - private static string Classified(string sample) => + private static string Classified(string sample, bool reverse) => "SELECT " + BucketExpression + " AS b, ts, kind, num_value, int_value, str_value, " + $"CASE WHEN status < {NotBadLimit} AND kind <> {EmptyKind} THEN 1 ELSE 0 END AS u, " + $"CASE WHEN status < {GoodLimit} AND kind <> {EmptyKind} THEN 1 ELSE 0 END AS g " + - $"FROM {sample} WHERE tag = $3 AND ts >= $4 AND ts < $5 AND NOT deleted"; + $"FROM {sample} WHERE tag = $3 AND " + (reverse ? "ts > $4 AND ts <= $5" : "ts >= $4 AND ts < $5") + " AND NOT deleted"; /// The five common count columns (after b). private static string CommonCounts() => 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 3f6b939b..dc54e4d2 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 @@ -425,12 +425,13 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// /// The (rounded) window is planned by : bucket i = /// [start + i·interval, start + (i+1)·interval), start-inclusive / end-exclusive, the last bucket truncated - /// at end, interval <= 0 ⇒ one bucket, start == end ⇒ none, a reversed window computed forward - /// and returned descending, at most buckets. One statement per node returns - /// 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. The time-weighted family (#671) instead streams the window plus four bounding + /// at end, interval <= 0 ⇒ one bucket, start == end ⇒ none, at most + /// buckets. A reversed window (#677) is anchored at its later time per Part 13 / + /// the SDK: window (end, start], buckets stepping back from start, the partial one earliest, each stamped + /// with its later edge, returned latest first. One statement per node returns the statistics + /// of each non-empty bucket (bucket index (cast(ts AS LONG)*10 − BucketBaseUnixTicks) / IntervalTicks, never + /// SAMPLE BY); a second fetches the tag's earliest sample for the start-of-data Partial bit; and + /// produces the values and fills the 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 seventeen served. @@ -446,7 +447,6 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource try { var now = _time.GetUtcNow().UtcDateTime; - var evaluated = new Dictionary(); if (family == AggregateFamily.TimeWeighted) { var tw = new TimeWeightedAggregator(aggregate, plan); @@ -456,31 +456,40 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource RecordSuccess(); return new HistoryReadResult(twResult, ContinuationPoint: null); } + var rows = new List(); + DateTime? earliest = null; if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false)) { var sample = _conn.SampleTable; var sql = family switch { - AggregateFamily.Simple => QuestDbAggregateQueries.Simple(sample), - AggregateFamily.StartEnd => QuestDbAggregateQueries.StartEnd(sample), - _ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime), + AggregateFamily.Simple => QuestDbAggregateQueries.Simple(sample, plan.Reversed), + AggregateFamily.StartEnd => QuestDbAggregateQueries.StartEnd(sample, plan.Reversed), + _ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed), }; await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false); - await using var cmd = _conn.CreateCommand(conn, sql); - QuestDbConnection.BindLong(cmd, plan.StartUnixTicks); - QuestDbConnection.BindLong(cmd, plan.IntervalTicks); - QuestDbConnection.BindText(cmd, fullReference); - QuestDbConnection.BindTimestamp(cmd, plan.StartUtc); - QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); - await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); - while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + await using (var cmd = _conn.CreateCommand(conn, sql)) { - var row = ReadBucket(reader, family); - evaluated[row.Bucket] = AggregateEvaluator.Evaluate(aggregate, row, plan.BucketStart(row.Bucket), now); + QuestDbConnection.BindLong(cmd, plan.BucketBaseUnixTicks); + QuestDbConnection.BindLong(cmd, plan.IntervalTicks); + QuestDbConnection.BindText(cmd, fullReference); + QuestDbConnection.BindTimestamp(cmd, plan.StartUtc); + QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); + await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + rows.Add(ReadBucket(reader, family)); + } + // The tag's earliest live sample: the start-of-data Partial rule (#677). + await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.EarliestSample(sample))) + { + QuestDbConnection.BindText(cmd, fullReference); + await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false) && !reader.IsDBNull(0)) + earliest = QuestDbConnection.ReadUtc(reader, 0); } } - var result = plan.Assemble(evaluated, bucketStart => AggregateEvaluator.NoData(bucketStart, now)); + var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, now); RecordSuccess(); return new HistoryReadResult(result, ContinuationPoint: null); 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 ff097de9..053aa67e 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 @@ -11,8 +11,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// every sample, status Good. Otherwise no usable sample ⇒ BadNoData; a numeric aggregate with a usable /// String/DateTime sample ⇒ BadAggregateInvalidInputs; else the value, Good when every sample is Good, /// UncertainDataSubNormal when any is Bad or Uncertain. Every status — including BadNoData — -/// carries the Calculated info bits. Timestamps are the bucket start, except Minimum/MaximumActualTime -/// (the sample's time; ties ⇒ earliest). +/// carries the Calculated info bits. Timestamps are the bucket's (its +/// start forward, its later edge reverse), except Minimum/MaximumActualTime (the sample's time; ties ⇒ earliest). +/// Every aggregate but Average also carries the Partial bit on the bucket the data starts inside +/// (). /// /// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is @@ -26,7 +28,7 @@ public static class AggregateEvaluator /// Evaluates over one non-empty bucket. /// Requested aggregate. /// The bucket's statistics. - /// The bucket's start time (the result timestamp). + /// The bucket's result timestamp (). /// Server timestamp. /// The calculated value. /// The aggregate is not one of the per-sample (Simple / @@ -70,6 +72,43 @@ public static class AggregateEvaluator } } + /// + /// True for the per-sample aggregates the SDK sets the start-of-data Partial bit on (every calculator but + /// AverageAggregateCalculator's plain Average, which turns SetPartialBit off). + /// + /// Requested aggregate. + /// Whether the start-of-data Partial rule applies. + public static bool SetsStartOfDataPartial(HistoryAggregateType aggregate) => aggregate != HistoryAggregateType.Average; + + /// + /// Evaluates every non-empty bucket row and assembles one result per interval in the plan's order (descending for a + /// reverse read): each value stamped with (the earlier edge forward, the + /// later edge reverse); empty buckets BadNoData; and — for every aggregate but Average — the Partial + /// bit on a non-Bad result of the bucket the tag's data starts inside (#677, the SDK's m_startOfData rule). + /// + /// A per-sample (Simple / StartEnd / ActualTime) aggregate. + /// The bucket plan. + /// The backend's non-empty bucket rows (ascending bucket index). + /// The tag's earliest stored sample of any quality, or null when none. + /// Server timestamp. + /// results. + public static IReadOnlyList EvaluateAll(HistoryAggregateType aggregate, BucketPlan plan, + IEnumerable rows, DateTime? earliestSampleUtc, DateTime now) + { + ArgumentNullException.ThrowIfNull(plan); + ArgumentNullException.ThrowIfNull(rows); + var evaluated = new Dictionary(); + var partialRule = SetsStartOfDataPartial(aggregate); + foreach (var row in rows) + { + var v = Evaluate(aggregate, row, plan.BucketTimestamp(row.Bucket), now); + if (partialRule && !StatusBits.IsBad(v.StatusCode) && plan.DataStartsInside(row.Bucket, earliestSampleUtc)) + v = v with { StatusCode = v.StatusCode | HistorianInfoBits.HistorianPartial }; + evaluated[row.Bucket] = v; + } + return plan.Assemble(evaluated, t => NoData(t, now)); + } + /// The result for a bucket with no samples (or no usable sample): BadNoData + Calculated. /// The bucket's start time. /// Server timestamp. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/BucketPlan.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/BucketPlan.cs index 29c95bd1..fc46d236 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/BucketPlan.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/BucketPlan.cs @@ -3,12 +3,29 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// -/// The bucket layout of a processed (aggregate) read. Bucket i = [start + i·interval, -/// start + (i+1)·interval) (start-inclusive, end-exclusive to the tick; the last bucket is truncated at -/// end). An interval <= 0 ⇒ one interval [start, end). start == end ⇒ no intervals. -/// end < start (reverse) ⇒ the window is swapped, computed forward, and assembled in descending bucket -/// order (buckets stay anchored at the earlier time). +/// The bucket layout of a processed (aggregate) read, per OPC UA Part 13 / the SDK's AggregateCalculator. /// +/// +/// +/// Forward (start < end): the window is [start, end); bucket i = +/// [start + i·interval, start + (i+1)·interval), the last bucket truncated at end (the partial one is +/// the latest); each bucket's timestamp is its start (earlier edge); results ascend. +/// +/// +/// Reverse (end < start, #677): the read is anchored at the request's start — its later +/// time — and steps back. The window is (end, start] (Part 11: the request's start is inclusive, its end +/// exclusive, so a reverse read includes its later time and excludes its earlier one — the same window as a reverse +/// raw read); result k = (start − (k+1)·interval, start − k·interval], clipped at end (the +/// partial bucket is the earliest); each bucket's timestamp is its later edge start − k·interval; +/// results descend. This is exactly the SDK's CreateSlice / GetTimestamp / backward +/// GetValues. +/// +/// +/// Internally buckets are indexed in ascending time (0 = earliest) for both directions; +/// reverses the list for a reverse read. / are always the earlier / later +/// window edges. An interval <= 0 ⇒ one interval over the whole window. start == end ⇒ no intervals. +/// +/// public sealed class BucketPlan { /// Spec §9.7: a processed read covering more intervals than this is refused. @@ -21,12 +38,15 @@ public sealed class BucketPlan Reversed = reversed; IntervalTicks = intervalTicks; BucketCount = bucketCount; + // Forward buckets are anchored at the earlier edge; reverse ones at the later edge, so the grid origin sits a + // whole number of intervals before it (possibly before the window start — the earliest bucket is clipped). + OriginTicks = reversed ? endUtc.Ticks - bucketCount * intervalTicks : startUtc.Ticks; } - /// The earlier end of the window (the forward start). + /// The earlier edge of the window. public DateTime StartUtc { get; } - /// The later end of the window (the forward, exclusive end). + /// The later edge of the window. public DateTime EndUtc { get; } /// True when the request came in reversed (end < start). @@ -38,9 +58,30 @@ public sealed class BucketPlan /// Number of buckets (0 for an empty window). public long BucketCount { get; } + /// + /// The tick at which the (unclipped) bucket grid starts: forward, EndUtc − n·interval + /// reverse. May precede (and even ), so it is a raw tick count. + /// + public long OriginTicks { get; } + /// as 100 ns ticks since the Unix epoch. public long StartUnixTicks => (StartUtc - DateTime.UnixEpoch).Ticks; + /// + /// The constant a backend adds to elapsed = t − StartUtc (ticks) before the integer division by + /// to get a sample's ascending bucket index: 0 forward ([lo, hi) buckets); + /// (StartUtc − OriginTicks) − 1 reverse, where buckets are (lo, hi] — index ⌈(t − origin)/I⌉ − 1 + /// = (t − origin − 1) / I. Never makes the dividend negative for an in-window sample (reverse excludes + /// t == StartUtc). + /// + public long BucketIndexAdjustTicks => Reversed ? StartUtc.Ticks - OriginTicks - 1 : 0; + + /// + /// minus : the base a backend subtracts from a + /// sample's Unix-epoch ticks to get the same dividend ((t − base) / IntervalTicks). + /// + public long BucketBaseUnixTicks => StartUnixTicks - BucketIndexAdjustTicks; + /// Plans the buckets of a processed read. /// Requested start. /// Requested end (earlier than start for a reverse read). @@ -64,15 +105,59 @@ public sealed class BucketPlan return new BucketPlan(start, end, reverse, intervalTicks, bucketCount); } - /// The start time of bucket (UTC kind). - /// Zero-based bucket index. - /// The bucket start. - public DateTime BucketStart(long index) => - DateTime.SpecifyKind(new DateTime(StartUtc.Ticks + index * IntervalTicks), DateTimeKind.Utc); + /// The earlier edge of bucket (ascending index), in ticks, clipped to the window. + /// Zero-based ascending bucket index. + /// The earlier edge. + public long LowerTicks(long index) => Math.Max(OriginTicks + index * IntervalTicks, StartUtc.Ticks); - /// One snapshot per bucket in window order; missing buckets get ; reversed if the window was. - /// Evaluated snapshots of the non-empty buckets, by bucket index. - /// Builds the snapshot of a bucket with no row, from its start time. + /// The later edge of bucket (ascending index), in ticks, clipped to the window. + /// Zero-based ascending bucket index. + /// The later edge. + public long UpperTicks(long index) => Math.Min(OriginTicks + (index + 1) * IntervalTicks, EndUtc.Ticks); + + /// + /// The result timestamp of bucket (UTC kind): its earlier edge forward, its later edge + /// reverse (the SDK's GetTimestamp). + /// + /// Zero-based ascending bucket index. + /// The bucket's timestamp. + public DateTime BucketTimestamp(long index) => + new(Reversed ? UpperTicks(index) : LowerTicks(index), DateTimeKind.Utc); + + /// True when bucket is shorter than the interval (the clipped one). + /// Zero-based ascending bucket index. + /// Whether the bucket is truncated. + public bool IsTruncated(long index) => UpperTicks(index) - LowerTicks(index) < IntervalTicks; + + /// + /// The ascending bucket index holding a sample at , or null when it is outside the window + /// ([start, end) forward, (start, end] reverse). The C# statement of the bucket math every backend's + /// SQL mirrors. + /// + /// Sample time. + /// The bucket index, or null. + public long? IndexOf(DateTime t) + { + var x = t.Ticks; + if (BucketCount == 0) return null; + var inside = Reversed ? x > StartUtc.Ticks && x <= EndUtc.Ticks : x >= StartUtc.Ticks && x < EndUtc.Ticks; + if (!inside) return null; + return (x - StartUtc.Ticks + BucketIndexAdjustTicks) / IntervalTicks; + } + + /// + /// True when the earliest stored sample of the tag lies strictly inside bucket — the data + /// starts part-way through the interval (no bounding value before it), so Part 13 / the SDK set the Partial bit. + /// + /// Zero-based ascending bucket index. + /// The tag's earliest stored sample (any quality), or null when it has none. + /// Whether the data starts inside the bucket. + public bool DataStartsInside(long index, DateTime? earliestSampleUtc) => + earliestSampleUtc is { } e && e.Ticks > LowerTicks(index) && e.Ticks < UpperTicks(index); + + /// One snapshot per bucket in result order; missing buckets get ; descending if reversed. + /// Evaluated snapshots of the non-empty buckets, by ascending bucket index. + /// Builds the snapshot of a bucket with no row, from its . /// snapshots. public IReadOnlyList Assemble(IReadOnlyDictionary byBucket, Func empty) @@ -81,10 +166,7 @@ public sealed class BucketPlan ArgumentNullException.ThrowIfNull(empty); var result = new List((int)BucketCount); for (long i = 0; i < BucketCount; i++) - { - var bucketStart = BucketStart(i); - result.Add(byBucket.TryGetValue(i, out var v) ? v : empty(bucketStart)); - } + result.Add(byBucket.TryGetValue(i, out var v) ? v : empty(BucketTimestamp(i))); if (Reversed) result.Reverse(); return result; } 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 index a4737fa8..47282367 100644 --- 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 @@ -18,7 +18,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// /// /// The signal (server defaults: TreatUncertainAsBad = false, PercentDataBad = PercentDataGood = 100, -/// UseSlopedExtrapolation = false; a client AggregateConfiguration is not honoured). Between a usable +/// UseSlopedExtrapolation = false; the node manager refuses any other client AggregateConfiguration). Between a usable /// sample p and the next sample s: /// /// s usable, both Double/Int64 ⇒ sloped (linear p → s); quality = worse of the two. @@ -51,7 +51,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// 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 +/// result of the truncated interval or of the interval the data starts inside (the first sample fed lies strictly +/// within it, #677); the timestamp is the interval's (its start forward, its +/// later edge reverse); the value is a /// (Boolean = 0/1, Int64 as double). /// /// @@ -112,6 +114,7 @@ public sealed class TimeWeightedAggregator private StoredSample? _prevAny; private StoredSample? _prevUsable; + private DateTime? _firstSampleUtc; // the earliest sample fed (any quality): where the data starts private long _nextPoint; // next point (ascending time) not yet classified private readonly List _pending = []; // points awaiting the next usable sample private bool _finished; @@ -169,6 +172,7 @@ public sealed class TimeWeightedAggregator if (_finished) throw new InvalidOperationException("The aggregation is already finished."); var t = s.TimeUtc.Ticks; if (_prevAny is { } last && t <= last.TimeUtc.Ticks) return; + _firstSampleUtc ??= s.TimeUtc; // Interpolative: classify every bucket start before this sample by the simple-bounding signal there. ClassifyPointsBefore(t); @@ -234,19 +238,21 @@ public sealed class TimeWeightedAggregator ? point with { ServerTimestampUtc = now } : new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, PointTime(i), now); - var bucketStart = _plan.BucketStart(i); + var stamp = _plan.BucketTimestamp(i); var duration = BucketEnd(i) - BucketStartTicks(i); - var partial = duration < _plan.IntervalTicks; // only the truncated last interval + // Partial: the truncated interval (the latest forward, the earliest reverse), or the one the data starts inside + // (no bounding value before it — the SDK's m_startOfData rule, #677). + var partial = _plan.IsTruncated(i) || _plan.DataStartsInside(i, _firstSampleUtc); 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); + return new DataValueSnapshot(value, code, stamp, 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); + return _badTicks[i] > 0 ? Calc(null, HistorianStatusCodes.Bad) : AggregateEvaluator.NoData(stamp, now); var status = _usableTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal; return _aggregate switch @@ -269,9 +275,12 @@ public sealed class TimeWeightedAggregator 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 BucketStartTicks(long i) => _plan.LowerTicks(i); - private long BucketEnd(long i) => Math.Min(BucketStartTicks(i) + _plan.IntervalTicks, _end); + private long BucketEnd(long i) => _plan.UpperTicks(i); + + /// The ascending index of the bucket whose (half-open) span holds tick (t ≥ window start). + private long FirstBucketAt(long t) => Math.Max(0, (t - _plan.OriginTicks) / _plan.IntervalTicks); private void AddRaw(StoredSample from, long toTicks, double toValue) => Distribute(from, toTicks, toValue, (i, area, ticks) => @@ -286,7 +295,7 @@ public sealed class TimeWeightedAggregator 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++) + for (var i = FirstBucketAt(lo); i < _plan.BucketCount; i++) { var bs = BucketStartTicks(i); if (bs >= hi) break; @@ -320,7 +329,7 @@ public sealed class TimeWeightedAggregator 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++) + for (var i = FirstBucketAt(lo); i < _plan.BucketCount; i++) { var bs = BucketStartTicks(i); if (bs >= hi) break; @@ -335,13 +344,12 @@ public sealed class TimeWeightedAggregator // ------------------------------------------------------------------------------------------ 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 + /// The time of Interpolative point (ascending time): the bucket's timestamp. Forward: the + /// bucket starts start + j·interval. Reverse (Part 13 anchors a reverse read at its later time, the + /// request's start): the later edges 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 long PointTicks(long j) => _plan.BucketTimestamp(j).Ticks; private DateTime PointTime(long j) => new(PointTicks(j), DateTimeKind.Utc); 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 4beb6b15..430c1fc2 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 @@ -5,13 +5,18 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; /// /// Push-down T-SQL for processed (aggregate) reads (spec §4 "Processed"), one query per /// ( picks it). Every query takes -/// @id (int TagId), @start/@end (datetime2(7), the forward window -/// [start, end)) and @intervalTicks (bigint bucket width in 100 ns ticks), and returns one row -/// per non-empty bucket. The caller generates every interval and fills the empty ones. +/// @id (int TagId), @start/@end (datetime2(7), the earlier / later window edges — +/// /), @intervalTicks (bigint bucket width in +/// 100 ns ticks) and @bucketAdjustTicks (bigint, ), and returns +/// one row per non-empty bucket (ascending bucket index). The caller generates every interval and fills the +/// empty ones. /// /// /// -/// Bucketing. b = elapsedTicks / @intervalTicks, where +/// Bucketing. b = (elapsedTicks + @bucketAdjustTicks) / @intervalTicks (#677: the adjust is 0 for a +/// forward read and re-anchors a reverse read's grid at its later time — is the C# +/// statement of the same math). The window is [@start, @end) forward and (@start, @end] reverse (a +/// reverse read includes its later time), selected by the reverse argument. elapsedTicks is /// elapsedTicks = DATEDIFF_BIG(microsecond, @start, t) * 10 + tick-of-microsecond(t) − tick-of-microsecond(@start) /// and tick-of-microsecond(x) = (DATEPART(nanosecond, x) / 100) % 10. The spec sketched /// millisecond, but DATEDIFF counts unit boundaries crossed, not elapsed time, so with a @@ -47,8 +52,9 @@ internal static class SqlAggregateQueries /// /// Schema-qualified Sample table name. /// The SQL text. - public static string Simple(string sample) => - Classified(sample) + + /// True for a reverse read (window (@start, @end]). + public static string Simple(string sample, bool reverse = false) => + Classified(sample, reverse) + "SELECT b, " + CommonCounts() + ", " + "AVG(CASE WHEN u = 1 THEN NumValue END) AS AvgValue, " + "MIN(CASE WHEN u = 1 THEN NumValue END) AS MinValue, " + @@ -63,8 +69,9 @@ internal static class SqlAggregateQueries /// /// Schema-qualified Sample table name. /// The SQL text. - public static string StartEnd(string sample) => - Classified(sample) + + /// True for a reverse read (window (@start, @end]). + public static string StartEnd(string sample, bool reverse = false) => + Classified(sample, reverse) + ", st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b)" + ", f AS (SELECT b, ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc, " + "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rf, " + @@ -82,8 +89,9 @@ internal static class SqlAggregateQueries /// Schema-qualified Sample table name. /// True for MaximumActualTime. /// The SQL text. - public static string ActualTime(string sample, bool maximum) => - Classified(sample) + + /// True for a reverse read (window (@start, @end]). + public static string ActualTime(string sample, bool maximum, bool reverse = false) => + Classified(sample, reverse) + ", st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b)" + ", r AS (SELECT b, NumValue, SourceTimeUtc, " + "ROW_NUMBER() OVER (PARTITION BY b ORDER BY NumValue " + (maximum ? "DESC" : "ASC") + ", SourceTimeUtc ASC) AS rn " + @@ -121,14 +129,25 @@ internal static class SqlAggregateQueries $"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end{usable} ORDER BY SourceTimeUtc ASC;"; } + /// + /// The earliest stored sample of the tag (any quality), one row (NULL when none): the start-of-data Partial + /// rule (#677). A MIN over the (TagId, SourceTimeUtc) key — a single seek. + /// + /// Schema-qualified Sample table name. + /// The SQL text. + public static string EarliestSample(string sample) => + $"SELECT MIN(SourceTimeUtc) FROM {sample} WHERE TagId = @id;"; + /// The s CTE: the window's samples with bucket index and the usable/good flags. - private static string Classified(string sample) => + private static string Classified(string sample, bool reverse) => "WITH s AS (SELECT (DATEDIFF_BIG(microsecond, @start, SourceTimeUtc) * 10 " + - "+ (DATEPART(nanosecond, SourceTimeUtc) / 100) % 10 - (DATEPART(nanosecond, @start) / 100) % 10) / @intervalTicks AS b, " + + "+ (DATEPART(nanosecond, SourceTimeUtc) / 100) % 10 - (DATEPART(nanosecond, @start) / 100) % 10 " + + "+ @bucketAdjustTicks) / @intervalTicks AS b, " + "SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, " + "CASE WHEN StatusCode >= 0 AND ValueKind <> 0 THEN 1 ELSE 0 END AS u, " + "CASE WHEN StatusCode >= 0 AND StatusCode < 1073741824 AND ValueKind <> 0 THEN 1 ELSE 0 END AS g " + - $"FROM {sample} WHERE TagId = @id AND SourceTimeUtc >= @start AND SourceTimeUtc < @end) "; + $"FROM {sample} WHERE TagId = @id AND " + + (reverse ? "SourceTimeUtc > @start AND SourceTimeUtc <= @end) " : "SourceTimeUtc >= @start AND SourceTimeUtc < @end) "); /// The five common count columns (after b). private static string CommonCounts() => 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 99182de6..b74a45a6 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 @@ -378,16 +378,18 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// 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 is truncated at /// end). <= 0 ⇒ one interval [start, end). start == end - /// ⇒ no intervals. end < start (reverse) ⇒ the window is swapped, computed forward, and returned - /// in descending bucket order (buckets stay anchored at the earlier time). + /// ⇒ no intervals. end < start (reverse, #677) ⇒ Part 13 / SDK anchoring: window (end, start], + /// buckets stepping back from start (the partial one is the earliest), each stamped with its later edge, + /// returned latest first (). /// /// /// Per interval: no samples ⇒ BadNoData. PercentGood / PercentBad ⇒ 100 × good|bad / all over /// every sample, status Good. Otherwise no usable sample ⇒ BadNoData; a numeric aggregate with a usable /// String/DateTime sample ⇒ BadAggregateInvalidInputs; else the value, Good when every sample is Good, /// UncertainDataSubNormal when any is Bad or Uncertain. Every status — including BadNoData — - /// carries the Calculated info bits. Timestamps are the bucket start, except Minimum/MaximumActualTime - /// (the sample's time; ties ⇒ earliest). + /// carries the Calculated info bits. Timestamps are the bucket start (its later edge on a reverse read), except + /// Minimum/MaximumActualTime (the sample's time; ties ⇒ earliest). Every aggregate but Average sets the Partial + /// bit on the bucket the tag's earliest sample lies strictly inside (#677). /// /// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is @@ -410,7 +412,6 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource try { 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) { @@ -420,30 +421,35 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource RecordSuccess(); return new HistoryReadResult(twResult, ContinuationPoint: null); } + var rows = new List(); + DateTime? earliest = null; if (tagId is { } id) { var sample = _conn.Q("Sample"); var sql = family switch { - AggregateFamily.Simple => SqlAggregateQueries.Simple(sample), - AggregateFamily.StartEnd => SqlAggregateQueries.StartEnd(sample), - _ => SqlAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime), + AggregateFamily.Simple => SqlAggregateQueries.Simple(sample, plan.Reversed), + AggregateFamily.StartEnd => SqlAggregateQueries.StartEnd(sample, plan.Reversed), + _ => SqlAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed), }; + // Second result set: the tag's earliest sample (the start-of-data Partial rule, #677). + sql += " " + SqlAggregateQueries.EarliestSample(sample); await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false); await using var cmd = _conn.CreateCommand(conn, sql); 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("@intervalTicks", SqlDbType.BigInt) { Value = plan.IntervalTicks }); + cmd.Parameters.Add(new SqlParameter("@bucketAdjustTicks", SqlDbType.BigInt) { Value = plan.BucketIndexAdjustTicks }); await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) - { - var row = ReadBucket(reader, family); - evaluated[row.Bucket] = AggregateEvaluator.Evaluate(aggregate, row, plan.BucketStart(row.Bucket), now); - } + rows.Add(ReadBucket(reader, family)); + if (await reader.NextResultAsync(cancellationToken).ConfigureAwait(false) + && await reader.ReadAsync(cancellationToken).ConfigureAwait(false) && !reader.IsDBNull(0)) + earliest = DateTime.SpecifyKind(reader.GetDateTime(0), DateTimeKind.Utc); } - var result = plan.Assemble(evaluated, bucketStart => AggregateEvaluator.NoData(bucketStart, now)); + var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, now); RecordSuccess(); return new HistoryReadResult(result, ContinuationPoint: null); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 0c2780b8..08c15da2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5009,6 +5009,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // OPC UA ProcessingInterval is a Duration in milliseconds — convert once per batch. var interval = TimeSpan.FromMilliseconds(details.ProcessingInterval); + // #677: a client AggregateConfiguration other than the server defaults is refused per node, never silently ignored. + var configurationServable = ServerAggregateConfiguration.IsServable(details.AggregateConfiguration); nodesToProcess = AuthorizeHistoryRead(context, nodesToProcess, results, errors); WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { @@ -5027,6 +5029,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 continue; } + if (!configurationServable) + { + errors[handle.Index] = StatusCodes.BadAggregateConfigurationRejected; + continue; + } + // Processed is SINGLE-SHOT (no continuation point). Unlike Raw, ReadProcessedDetails carries // NO client count cap (NumValuesPerNode) — the bucket count is deterministic (window / interval) // and the single-shot backend returns every bucket in one read, so there is no "full page ⇒ diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ServerAggregateConfiguration.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ServerAggregateConfiguration.cs new file mode 100644 index 00000000..32187303 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ServerAggregateConfiguration.cs @@ -0,0 +1,62 @@ +using Opc.Ua; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// +/// The Part 13 processed reads are evaluated under (#677), and the rule for a +/// client-supplied one in ReadProcessedDetails. +/// +/// +/// +/// The aggregate evaluators (AggregateEvaluator / TimeWeightedAggregator in +/// Driver.Historian.Sql.Common) are written against fixed settings, not parameterised: Uncertain samples and +/// Uncertain time count as Good ( = false); an interval is Good only when all of +/// it is ( = 100) and Bad only when none of it is non-Bad +/// ( = 100), otherwise UncertainDataSubNormal; the value past the last sample +/// is held (stepped, = false). These are also the SDK's own server defaults +/// (AggregateManager.GetDefaultConfiguration), which a test pins. +/// +/// +/// A request asking for UseServerCapabilitiesDefaults, or spelling out exactly these values, is served; +/// any other configuration is refused per node with BadAggregateConfigurationRejected rather than silently +/// answered under settings the client did not ask for (it used to be ignored). +/// +/// +public static class ServerAggregateConfiguration +{ + /// Uncertain data is not treated as Bad. + public const bool TreatUncertainAsBad = false; + + /// An interval is Bad only when 100 % of it is Bad. + public const byte PercentDataBad = 100; + + /// An interval is Good only when 100 % of it is Good. + public const byte PercentDataGood = 100; + + /// The value past the last sample is held (stepped), not sloped. + public const bool UseSlopedExtrapolation = false; + + /// The server defaults as an (explicit values, not "use defaults"). + public static AggregateConfiguration Defaults => new() + { + UseServerCapabilitiesDefaults = false, + TreatUncertainAsBad = TreatUncertainAsBad, + PercentDataBad = PercentDataBad, + PercentDataGood = PercentDataGood, + UseSlopedExtrapolation = UseSlopedExtrapolation, + }; + + /// + /// True when a processed read with can be served: none supplied, + /// UseServerCapabilitiesDefaults, or every setting equal to the server's. + /// + /// The request's configuration (may be null). + /// Whether to serve; false ⇒ BadAggregateConfigurationRejected. + public static bool IsServable(AggregateConfiguration? requested) => + requested is null + || requested.UseServerCapabilitiesDefaults + || (requested.TreatUncertainAsBad == TreatUncertainAsBad + && requested.PercentDataBad == PercentDataBad + && requested.PercentDataGood == PercentDataGood + && requested.UseSlopedExtrapolation == UseSlopedExtrapolation); +} diff --git a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs index 9ce58c18..d4503472 100644 --- a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs +++ b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs @@ -1,6 +1,7 @@ using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance; @@ -171,7 +172,8 @@ public abstract class ProcessedConformance : ConformanceBase // 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); + // The data starts (+0) strictly inside the one interval ⇒ Partial as well (#677, the SDK's m_startOfData rule). + tw.StatusCode.ShouldBe(UncertainCalc | 0x4u); src.GetHealthSnapshot().TotalFailures.ShouldBe(0); } @@ -196,12 +198,128 @@ public abstract class ProcessedConformance : ConformanceBase } [Fact] - public async Task A_reverse_window_is_computed_forward_and_returned_descending() + public async Task A_reverse_window_is_anchored_at_its_later_time() { + // #677 (Part 13 / SDK): At(80) → At(0) is the window (0,80] in intervals (60,80], (40,60], (20,40], (0,20], each + // stamped with its LATER edge, latest first. A sample on an edge belongs to the interval it ends: +40 is in + // (20,40], +20 (Bad) in (0,20], and +0 is outside the window. (The pre-#677 layout answered [60,80) … [0,20) + // stamped 60/40/20/0 with values null, 50, 40, 15.) await ConformanceData.SeedAAsync(B); var s = (await B.CreateDataSource().ReadProcessedAsync("A", At(80), At(0), Twenty, HistoryAggregateType.Average, Ct)).Samples; - s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(60)), B.Round(At(40)), B.Round(At(20)), B.Round(At(0))]); - s.Select(x => x.Value).ShouldBe([null, 50d, 40d, 15d]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(80)), B.Round(At(60)), B.Round(At(40)), B.Round(At(20))]); + s.Select(x => x.Value).ShouldBe([null, null, 45d, 20d]); + s.Select(x => x.StatusCode).ShouldBe([NoDataCalc, NoDataCalc, GoodCalc, UncertainCalc]); + } + + [Fact] + public async Task A_reverse_window_truncates_its_earliest_interval() + { + // At(50) → At(0), 20 s over tag A (+0=10, +10=20, +20=Bad, +30=40, +40=50): (30,50] = {+40} → 50 (+30 is its + // earlier edge, excluded); (10,30] = {+20 Bad, +30} → 40, Uncertain; (0,10] = {+10} → 20 — the truncated interval + // is the EARLIEST one. (Pre-#677: [0,20), [20,40), [40,50) stamped 40/20/0 with values 50, 40, 15.) + await ConformanceData.SeedAAsync(B); + var s = (await B.CreateDataSource().ReadProcessedAsync("A", At(50), At(0), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(50)), B.Round(At(30)), B.Round(At(10))]); + s.Select(x => x.Value).ShouldBe([50d, 40d, 20d]); + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalc]); + } + + [Fact] + public async Task Reverse_bucket_edges_are_later_edge_inclusive_to_the_tick() + { + // The mirror of the forward edge test: At(40) → At(0), 20 s ⇒ (20,40], (0,20]. The bucket math is pushed down, + // so this pins the SQL, not just the C# plan. + var r = B.ResolutionTicks; + await B.SeedAsync("ER", + (At(0), HistorianValue.FromDouble(1000), Good), // == earlier edge: excluded + (At(0).AddTicks(r), HistorianValue.FromDouble(1), Good), // first tick of (0,20] + (At(20), HistorianValue.FromDouble(2), Good), // == edge: the LATER edge of (0,20] + (At(20).AddTicks(r), HistorianValue.FromDouble(4), Good), // first tick of (20,40] + (At(40), HistorianValue.FromDouble(8), Good), // == request start: included, in (20,40] + (At(40).AddTicks(r), HistorianValue.FromDouble(2000), Good)); // after the window: excluded + var s = (await B.CreateDataSource().ReadProcessedAsync("ER", At(40), At(0), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(40)), B.Round(At(20))]); + s.Select(x => x.Value).ShouldBe([6d, 1.5d]); + } + + [Fact] + public async Task The_interval_the_data_starts_inside_carries_the_Partial_bit() + { + // #677 (SDK m_startOfData): the tag's first sample is +5, inside [0,20) ⇒ Partial there for every aggregate but + // Average; [20,40) is fully covered. With an earlier sample stored (even outside the window) nothing is Partial. + await B.SeedAsync("LS", + (At(5), HistorianValue.FromDouble(10), Good), + (At(15), HistorianValue.FromDouble(30), Good), + (At(25), HistorianValue.FromDouble(20), Good), + (At(60), HistorianValue.FromDouble(20), Good)); + var src = B.CreateDataSource(); + foreach (var agg in new[] { HistoryAggregateType.Count, HistoryAggregateType.Minimum, HistoryAggregateType.TimeAverage2 }) + { + var s = (await src.ReadProcessedAsync("LS", At(0), At(40), Twenty, agg, Ct)).Samples; + (s[0].StatusCode & 0x4u).ShouldBe(0x4u, agg.ToString()); + (s[1].StatusCode & 0x4u).ShouldBe(0u, agg.ToString()); + } + (await src.ReadProcessedAsync("LS", At(0), At(40), Twenty, HistoryAggregateType.Average, Ct)).Samples + .ShouldAllBe(x => (x.StatusCode & 0x4u) == 0); + + await B.SeedAsync("LS2", + (At(-100), HistorianValue.FromDouble(10), Good), + (At(5), HistorianValue.FromDouble(10), Good), + (At(15), HistorianValue.FromDouble(30), Good), + (At(25), HistorianValue.FromDouble(20), Good)); + (await src.ReadProcessedAsync("LS2", At(0), At(40), Twenty, HistoryAggregateType.Count, Ct)).Samples + .ShouldAllBe(x => (x.StatusCode & 0x4u) == 0); + } + + /// Dense: every 2.5 s from −10 to +100, values cycling, one Bad sample at +22.5 (no empty interval). + private Task SeedDenseAsync() => B.SeedAsync("P", DenseRows()); + + private static (DateTime T, HistorianValue V, uint Status)[] DenseRows() => Enumerable.Range(0, 45) + .Select(i => (At(-10 + i * 2.5), HistorianValue.FromDouble(10 + (i * 7 % 13)), i == 13 ? Bad : Good)) + .ToArray(); + + [Theory] + [InlineData(0, 80, 20)] + [InlineData(80, 0, 20)] + [InlineData(77, 3, 20)] + [InlineData(3, 77, 20)] + public async Task Processed_reads_match_the_SDK_aggregate_calculators(double from, double to, double intervalS) + { + // #677: the real backend (pushed-down SQL bucket math + the shared evaluators) against the pinned SDK's own Part 13 + // calculators over the same stored rows — timestamps, values, status code bits and the Partial bit, forward and + // reverse. See SdkAggregateParityTests (Sql.Common.Tests) for the documented divergences this avoids/masks. + await SeedDenseAsync(); + var rows = DenseRows().Select(r => (B.Round(r.T), r.V, r.Status)).ToArray(); + var src = B.CreateDataSource(); + var interval = TimeSpan.FromSeconds(intervalS); + // The SDK sees the window the backend actually serves (QuestDB rounds the bounds to µs). + var (start, end) = (B.Round(At(from)), B.Round(At(to))); + foreach (var agg in new[] + { + HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, + HistoryAggregateType.Count, HistoryAggregateType.TimeAverage, HistoryAggregateType.TimeAverage2, + HistoryAggregateType.Total, HistoryAggregateType.Total2, HistoryAggregateType.Interpolative, + }) + { + var ours = (await src.ReadProcessedAsync("P", At(from), At(to), interval, agg, Ct)).Samples.ToList(); + var tw = AggregateFamilies.FamilyOf(agg) == AggregateFamily.TimeWeighted; + if (tw && agg != HistoryAggregateType.Interpolative) + { + // Documented divergence: ours flags the truncated interval Partial, the SDK does not. + var plan = BucketPlan.Create(start, end, interval); + for (long i = 0; i < plan.BucketCount; i++) + if (plan.IsTruncated(i)) + { + var k = (int)(plan.Reversed ? plan.BucketCount - 1 - i : i); + ours[k] = ours[k] with { StatusCode = ours[k].StatusCode & ~0x4u }; + } + } + var reverseInterpolative = agg == HistoryAggregateType.Interpolative && from > to; + SdkAggregateOracle.ShouldMatch(ours, SdkAggregateOracle.Run(agg, start, end, interval, rows), + $"{agg} {from}→{to}/{intervalS}s", + compareCalculatedInterpolatedBits: tw && !reverseInterpolative, + compareCodeBits: agg is not (HistoryAggregateType.TimeAverage or HistoryAggregateType.Total) && !reverseInterpolative); + } } [Fact] @@ -245,7 +363,8 @@ public abstract class ProcessedConformance : ConformanceBase (await One(HistoryAggregateType.Count)).StatusCode.ShouldBe(NoDataCalc); var pb = await One(HistoryAggregateType.PercentBad); pb.Value.ShouldBe(100d); - pb.StatusCode.ShouldBe(GoodCalc); + // The tag's data starts (+1) inside the interval ⇒ Partial on the non-Bad percentage (#677). + pb.StatusCode.ShouldBe(GoodCalc | 0x4u); } [Fact] diff --git a/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs b/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs new file mode 100644 index 00000000..fbe3b80c --- /dev/null +++ b/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs @@ -0,0 +1,152 @@ +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance; + +/// +/// Drives the pinned OPC UA SDK's own Part 13 aggregate calculators (Opc.Ua.Server — the classes the SDK's +/// AggregateManager registers) in-process over a sample set, so our processed reads can be compared with the +/// reference implementation bucket by bucket (#677). The calculators run under the server's default +/// (TreatUncertainAsBad = false, PercentDataBad = PercentDataGood = 100, +/// UseSlopedExtrapolation = false), sloped interpolation, source timestamps. +/// +/// +/// Shared (source-linked) by Sql.Common.Tests and both SQL backends' integration suites. +/// +public static class SdkAggregateOracle +{ + /// One processed value: value (double / int / null), full status code, timestamp. + public readonly record struct Result(object? Value, uint StatusCode, DateTime Timestamp); + + /// The server-default aggregate configuration (what AggregateManager.GetDefaultConfiguration returns). + public static AggregateConfiguration DefaultConfiguration => new() + { + TreatUncertainAsBad = false, + PercentDataBad = 100, + PercentDataGood = 100, + UseSlopedExtrapolation = false, + UseServerCapabilitiesDefaults = false, + }; + + /// Runs the SDK calculator for over . + /// The aggregate. + /// Request start (later than for a reverse read). + /// Request end. + /// Processing interval. + /// Every stored sample of the tag, any order (sorted here). + /// The SDK's processed values, in the order it returns them. + public static IReadOnlyList Run(HistoryAggregateType aggregate, DateTime start, DateTime end, TimeSpan interval, + IEnumerable<(DateTime Time, HistorianValue Value, uint Status)> samples) + { + var id = AggregateId(aggregate); + var ms = interval.TotalMilliseconds; + var cfg = DefaultConfiguration; + IAggregateCalculator calc = aggregate switch + { + HistoryAggregateType.Average or HistoryAggregateType.TimeAverage or HistoryAggregateType.TimeAverage2 + or HistoryAggregateType.Total or HistoryAggregateType.Total2 => + new AverageAggregateCalculator(id, start, end, ms, false, cfg, null), + HistoryAggregateType.Minimum or HistoryAggregateType.Maximum or HistoryAggregateType.Range + or HistoryAggregateType.MinimumActualTime or HistoryAggregateType.MaximumActualTime => + new MinMaxAggregateCalculator(id, start, end, ms, false, cfg, null), + HistoryAggregateType.Count => new CountAggregateCalculator(id, start, end, ms, false, cfg, null), + HistoryAggregateType.Start or HistoryAggregateType.End or HistoryAggregateType.Delta => + new StartEndAggregateCalculator(id, start, end, ms, false, cfg, null), + HistoryAggregateType.Interpolative => new AggregateCalculator(id, start, end, ms, false, cfg, null), + _ => throw new ArgumentOutOfRangeException(nameof(aggregate), aggregate, "no SDK calculator mapped"), + }; + + // The SDK expects values in the direction of the read: ascending forward, descending reverse. + var ordered = samples.OrderBy(s => s.Time).ToList(); + if (end < start) ordered.Reverse(); + foreach (var s in ordered) calc.QueueRawValue(ToDataValue(s.Time, s.Value, s.Status)); + + var results = new List(); + for (var guard = 0; guard < 100_000; guard++) + { + var dv = calc.GetProcessedValue(true); + if (dv is null) break; + results.Add(new Result(dv.Value, dv.StatusCode.Code, DateTime.SpecifyKind(dv.SourceTimestamp, DateTimeKind.Utc))); + } + return results; + } + + /// + /// Asserts our snapshots equal the SDK's: same count, same timestamps, same status code bits (severity + sub-code) + /// and Partial bit (and optionally the Calculated / Interpolated bits), same value (doubles to 1e-9; ints exactly; + /// null on a Bad status). A failure lists every differing interval side by side. + /// + /// Our processed values. + /// The SDK's. + /// Context for failure messages. + /// Also compare the Calculated / Interpolated aggregate bits. + /// Compare the severity + sub-code (false only for a documented status divergence). + /// Compare the values (false only for a documented value divergence). + public static void ShouldMatch(IReadOnlyList ours, IReadOnlyList sdk, string label, + bool compareCalculatedInterpolatedBits = true, bool compareCodeBits = true, bool compareValues = true) + { + var problems = new List(); + if (ours.Count != sdk.Count) problems.Add($"interval count: ours {ours.Count}, sdk {sdk.Count}"); + for (var i = 0; i < Math.Min(ours.Count, sdk.Count); i++) + { + var o = ours[i]; + var s = sdk[i]; + var bitMask = 0x4u | (compareCodeBits ? 0xFFFF0000u : 0u) | (compareCalculatedInterpolatedBits ? 0x3u : 0u); + var same = o.SourceTimestampUtc == s.Timestamp && (o.StatusCode & bitMask) == (s.StatusCode & bitMask); + if (!compareValues) + { + } + else if (same && !StatusCode.IsBad(s.StatusCode)) + { + same = s.Value switch + { + null => o.Value is null, + int n => o.Value is not null && Convert.ToInt32(o.Value) == n, + _ => o.Value is not null && Math.Abs(Convert.ToDouble(o.Value) - Convert.ToDouble(s.Value)) <= 1e-9, + }; + } + else if (same && compareCodeBits) same = o.Value is null; + if (!same) + problems.Add($"[{i}] ours {o.SourceTimestampUtc:HH:mm:ss.fffffff} {o.Value ?? "null"} 0x{o.StatusCode:X8}" + + $" | sdk {s.Timestamp:HH:mm:ss.fffffff} {s.Value ?? "null"} 0x{s.StatusCode:X8}"); + } + problems.ShouldBeEmpty($"{label} differs from the SDK:\n" + string.Join("\n", problems)); + } + + private static DataValue ToDataValue(DateTime t, HistorianValue v, uint status) => new() + { + WrappedValue = v.Kind switch + { + HistorianValueKind.Empty => Variant.Null, + HistorianValueKind.Int64 => new Variant(v.Integer), + HistorianValueKind.Boolean => new Variant(v.Number != 0), + HistorianValueKind.String => new Variant(v.Text), + _ => new Variant(v.Number), + }, + StatusCode = status, + SourceTimestamp = t, + ServerTimestamp = t, + }; + + private static NodeId AggregateId(HistoryAggregateType a) => a switch + { + HistoryAggregateType.Average => ObjectIds.AggregateFunction_Average, + HistoryAggregateType.Minimum => ObjectIds.AggregateFunction_Minimum, + HistoryAggregateType.Maximum => ObjectIds.AggregateFunction_Maximum, + HistoryAggregateType.Range => ObjectIds.AggregateFunction_Range, + HistoryAggregateType.MinimumActualTime => ObjectIds.AggregateFunction_MinimumActualTime, + HistoryAggregateType.MaximumActualTime => ObjectIds.AggregateFunction_MaximumActualTime, + HistoryAggregateType.Count => ObjectIds.AggregateFunction_Count, + HistoryAggregateType.Start => ObjectIds.AggregateFunction_Start, + HistoryAggregateType.End => ObjectIds.AggregateFunction_End, + HistoryAggregateType.Delta => ObjectIds.AggregateFunction_Delta, + HistoryAggregateType.TimeAverage => ObjectIds.AggregateFunction_TimeAverage, + HistoryAggregateType.TimeAverage2 => ObjectIds.AggregateFunction_TimeAverage2, + HistoryAggregateType.Total => ObjectIds.AggregateFunction_Total, + HistoryAggregateType.Total2 => ObjectIds.AggregateFunction_Total2, + HistoryAggregateType.Interpolative => ObjectIds.AggregateFunction_Interpolative, + _ => throw new ArgumentOutOfRangeException(nameof(a), a, "no SDK aggregate id mapped"), + }; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagProvisionerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagProvisionerTests.cs index 798a0dca..51fba3d5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagProvisionerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagProvisionerTests.cs @@ -78,6 +78,27 @@ public sealed class GatewayTagProvisionerTests Assert.True(defs[0].StorageRateMs > 0, "provisioned tags must carry a positive storage rate (0 fails EnsureTags)."); } + [Fact] + public async Task Definition_carries_an_integral_divisor_of_one() + { + // #677: AVEVA's Total integral is value·seconds ÷ IntegralDivisor. The proto3 double default is 0, which the + // gateway passes through, so every provisioned tag must say 1 explicitly — the unit the SQL backends' Total uses. + var fake = new FakeHistorianGatewayClient { EnsureTagsResult = new TagOperationResults() }; + var p = Provisioner(fake); + + await p.EnsureTagsAsync( + new[] + { + new HistorianTagProvisionRequest("Pump1.Temp", DriverDataType.Float32, null, null), + new HistorianTagProvisionRequest("Pump1.Run", DriverDataType.Boolean, null, null), + }, + TestContext.Current.CancellationToken); + + var defs = fake.LastEnsureDefinitions!; + Assert.Equal(2, defs.Count); + Assert.All(defs, d => Assert.Equal(1.0, d.IntegralDivisor)); + } + [Fact] public async Task Maps_metadata_and_coalesces_null_metadata_to_empty() { 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 0f5c52c3..745343ab 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 @@ -82,7 +82,8 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime foreach (var agg in new[] { HistoryAggregateType.Average, HistoryAggregateType.Start, HistoryAggregateType.MaximumActualTime }) { var s = (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, agg, Ct)).Samples.Single(); - s.StatusCode.ShouldBe(UncertainCalc, agg.ToString()); + // The data starts (+0) strictly inside the one interval ⇒ Partial on every aggregate but Average (#677). + s.StatusCode.ShouldBe(agg == HistoryAggregateType.Average ? UncertainCalc : UncertainCalc | 0x4u, agg.ToString()); } (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.Average, Ct)).Samples.Single().Value.ShouldBe(30d); (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.Start, Ct)).Samples.Single().Value.ShouldBe(10d); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests.csproj index c34ce291..de0bc275 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests.csproj +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests.csproj @@ -14,6 +14,8 @@ + + all runtime; build; native; contentfiles; analyzers; buildtransitive 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 e7953e8f..aa516941 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 @@ -37,9 +37,7 @@ public sealed class AggregateEvaluatorTests private static IReadOnlyList ReadA(HistoryAggregateType agg, bool maximum = false) { var plan = BucketPlan.Create(At(0), At(80), TimeSpan.FromSeconds(20)); - var evaluated = RowsA(maximum).ToDictionary(kv => kv.Key, - kv => AggregateEvaluator.Evaluate(agg, kv.Value, plan.BucketStart(kv.Key), Now)); - return plan.Assemble(evaluated, t => AggregateEvaluator.NoData(t, Now)); + return AggregateEvaluator.EvaluateAll(agg, plan, RowsA(maximum).Values, At(0), Now); } [Theory] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/BucketPlanTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/BucketPlanTests.cs index 71ed9889..7359757b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/BucketPlanTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/BucketPlanTests.cs @@ -28,8 +28,10 @@ public sealed class BucketPlanTests { var p = BucketPlan.Create(At(0), At(50), TimeSpan.FromSeconds(20)); p.BucketCount.ShouldBe(3); - p.BucketStart(2).ShouldBe(At(40)); - p.BucketStart(2).Kind.ShouldBe(DateTimeKind.Utc); + p.BucketTimestamp(2).ShouldBe(At(40)); + p.BucketTimestamp(2).Kind.ShouldBe(DateTimeKind.Utc); + p.IsTruncated(2).ShouldBeTrue(); + p.IsTruncated(1).ShouldBeFalse(); } [Fact] @@ -43,14 +45,91 @@ public sealed class BucketPlanTests } [Fact] - public void A_reverse_window_is_planned_forward_and_assembled_descending() + public void A_reverse_window_is_anchored_at_its_later_time_and_stamped_with_later_edges() { + // #677 (Part 13 / SDK CreateSlice + GetTimestamp): At(80) → At(0), 20 s ⇒ (60,80], (40,60], (20,40], (0,20], + // each stamped with its later edge, latest first. var p = BucketPlan.Create(At(80), At(0), TimeSpan.FromSeconds(20)); p.Reversed.ShouldBeTrue(); p.StartUtc.ShouldBe(At(0)); p.EndUtc.ShouldBe(At(80)); var s = p.Assemble(new Dictionary(), t => new DataValueSnapshot(null, 0, t, t)); - s.Select(x => x.SourceTimestampUtc).ShouldBe([At(60), At(40), At(20), At(0)]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(80), At(60), At(40), At(20)]); + } + + [Fact] + public void A_reverse_window_puts_the_partial_interval_at_its_earliest_end() + { + // At(50) → At(0), 20 s: (30,50], (10,30], (0,10] — the truncated bucket is the EARLIEST (the old layout truncated + // the latest: [0,20), [20,40), [40,50)). + var p = BucketPlan.Create(At(50), At(0), TimeSpan.FromSeconds(20)); + p.BucketCount.ShouldBe(3); + p.LowerTicks(0).ShouldBe(At(0).Ticks); + p.UpperTicks(0).ShouldBe(At(10).Ticks); + p.IsTruncated(0).ShouldBeTrue(); + p.LowerTicks(2).ShouldBe(At(30).Ticks); + p.UpperTicks(2).ShouldBe(At(50).Ticks); + p.IsTruncated(2).ShouldBeFalse(); + var s = p.Assemble(new Dictionary(), t => new DataValueSnapshot(null, 0, t, t)); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(50), At(30), At(10)]); + } + + [Fact] + public void Forward_buckets_are_start_inclusive_and_reverse_buckets_later_edge_inclusive() + { + var fwd = BucketPlan.Create(At(0), At(50), TimeSpan.FromSeconds(20)); + fwd.IndexOf(At(0)).ShouldBe(0); + fwd.IndexOf(At(20)).ShouldBe(1); + fwd.IndexOf(At(20).AddTicks(-1)).ShouldBe(0); + fwd.IndexOf(At(49.9999999)).ShouldBe(2); + fwd.IndexOf(At(50)).ShouldBeNull(); + fwd.IndexOf(At(0).AddTicks(-1)).ShouldBeNull(); + + // (0,10] (10,30] (30,50]: the later time is in the window, the earlier one is not (as a reverse raw read). + var rev = BucketPlan.Create(At(50), At(0), TimeSpan.FromSeconds(20)); + rev.IndexOf(At(50)).ShouldBe(2); + rev.IndexOf(At(30)).ShouldBe(1); + rev.IndexOf(At(30).AddTicks(1)).ShouldBe(2); + rev.IndexOf(At(10)).ShouldBe(0); + rev.IndexOf(At(10).AddTicks(1)).ShouldBe(1); + rev.IndexOf(At(0).AddTicks(1)).ShouldBe(0); + rev.IndexOf(At(0)).ShouldBeNull(); + rev.IndexOf(At(50).AddTicks(1)).ShouldBeNull(); + } + + [Fact] + public void The_backend_bucket_formula_matches_IndexOf_in_both_directions() + { + // b = (t − start + adjust) / interval and b = (unixTicks(t) − base) / interval — what the SQL computes. + foreach (var p in new[] + { + BucketPlan.Create(At(0), At(50), TimeSpan.FromSeconds(20)), + BucketPlan.Create(At(50), At(0), TimeSpan.FromSeconds(20)), + BucketPlan.Create(At(60), At(0), TimeSpan.FromSeconds(20)), + BucketPlan.Create(At(0).AddTicks(70), At(0), TimeSpan.FromTicks(3)), + }) + { + for (var t = p.StartUtc.Ticks - 2; t <= p.EndUtc.Ticks + 2; t += Math.Max(1, (p.EndUtc.Ticks - p.StartUtc.Ticks) / 997)) + { + var at = new DateTime(t, DateTimeKind.Utc); + if (p.IndexOf(at) is not { } idx) continue; + ((t - p.StartUtc.Ticks + p.BucketIndexAdjustTicks) / p.IntervalTicks).ShouldBe(idx); + (((at - DateTime.UnixEpoch).Ticks - p.BucketBaseUnixTicks) / p.IntervalTicks).ShouldBe(idx); + p.LowerTicks(idx).ShouldBeLessThanOrEqualTo(t); + p.UpperTicks(idx).ShouldBeGreaterThanOrEqualTo(t); + } + } + } + + [Fact] + public void Data_starts_inside_only_strictly_between_the_bucket_edges() + { + var p = BucketPlan.Create(At(0), At(40), TimeSpan.FromSeconds(20)); + p.DataStartsInside(0, At(5)).ShouldBeTrue(); + p.DataStartsInside(1, At(5)).ShouldBeFalse(); + p.DataStartsInside(1, At(20)).ShouldBeFalse(); // on the edge: data covers the whole interval + p.DataStartsInside(0, null).ShouldBeFalse(); + p.DataStartsInside(0, At(-5)).ShouldBeFalse(); } [Fact] @@ -64,6 +143,11 @@ public sealed class BucketPlanTests s[2].SourceTimestampUtc.ShouldBe(At(40)); } + [Fact] + public void BucketBaseUnixTicks_is_the_start_forward() + => BucketPlan.Create(DateTime.UnixEpoch.AddTicks(1234567), DateTime.UnixEpoch.AddSeconds(10), TimeSpan.FromSeconds(1)) + .BucketBaseUnixTicks.ShouldBe(1234567L); + [Fact] public void StartUnixTicks_is_ticks_since_the_unix_epoch() => BucketPlan.Create(DateTime.UnixEpoch.AddTicks(1234567), DateTime.UnixEpoch.AddSeconds(10), TimeSpan.Zero) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/SdkAggregateParityTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/SdkAggregateParityTests.cs new file mode 100644 index 00000000..5d87f2fa --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/SdkAggregateParityTests.cs @@ -0,0 +1,191 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance; +using static ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.Fixture; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests; + +/// +/// #677: our processed reads against the pinned SDK's own Part 13 aggregate calculators (), +/// forward AND reverse, bucket by bucket — timestamps, values and status. The per-sample aggregates are bucketed with +/// (the C# statement of the backends' SQL bucket math; the integration suites run the +/// same comparison through the real SQL) and evaluated by ; the +/// time-weighted ones by . +/// +/// +/// Deliberate, documented divergences the datasets avoid (they are not anchoring questions): +/// +/// an empty interval inside the data range: the SDK's Count answers 0/Good, ours BadNoData; +/// Uncertain samples in a per-sample aggregate: the SDK's Average/Count/Min/Max use Good samples only, +/// ours every usable (Good + Uncertain) sample (TreatUncertainAsBad = false applied consistently); +/// the per-sample Calculated / MultipleValues info bits (the SDK's MinMax sets Calculated only when the +/// extreme is not at the interval start, and flags ties) — those comparisons check code bits + Partial only; +/// a truncated interval: ours sets Partial on a non-Bad time-weighted result (the SDK does not), so the +/// non-multiple windows mask that one bucket's Partial bit; +/// the SDK's end-of-data Partial rule (forward, the interval holding the last sample when nothing follows +/// and nothing was extrapolated) is not implemented: every dataset carries data past the window end. +/// +/// Pre-existing forward divergences this parity test surfaced (#671 behaviour, unchanged here — forward results were +/// to stay as they are; recorded for a follow-up rather than fixed in place): +/// +/// TimeAverage / Total status: the SDK (interpolated bounds) reports UncertainDataSubNormal when +/// any region of the interval is non-Good (e.g. a segment bridged across a Bad sample); ours applies the +/// time-based rule (Uncertain time counts as Good) to these two as well, so a bridged-but-fully-covered interval +/// is Good. Values and timestamps agree; the status code bits are not compared for these two. +/// Total over an interval the data starts inside: the SDK integrates only the defined part +/// (value·s over the covered seconds); ours is TimeAverage × interval seconds. The Partial bit and timestamp +/// agree; that value is not compared. +/// Interpolative, reverse, a sample exactly at the point: the SDK's backward UpdateSlice files +/// a sample AT the point as the early bound (>= 0), so it interpolates from it to the next Good sample — +/// same value, but the Interpolated bit, and Uncertain when a Bad sample sits between. Ours returns the raw +/// sample with its own status in both directions (as the SDK itself does forward). Reverse Interpolative +/// therefore compares timestamps and values (the anchoring question) but not status. +/// +/// +public sealed class SdkAggregateParityTests +{ + private static readonly DateTime Now = At(3600); + + /// Dense: every 2.5 s from −10 to +100, values cycling, one Bad sample at +22.5 — no empty interval. + private static readonly StoredSample[] Dense = Enumerable.Range(0, 45) + .Select(i => (t: -10 + i * 2.5, i)) + .Select(x => S(x.t, HistorianValue.FromDouble(10 + (x.i * 7 % 13)), x.t == 22.5 ? Bad : Good)) + .ToArray(); + + /// Tag A with Good bounds on both sides, plus an Uncertain sample and an outage marker further on. + private static readonly StoredSample[] Bounded = + [ + S(-10, HistorianValue.FromDouble(5), Good), + .. A, + S(55, HistorianValue.FromDouble(70), Uncertain), + S(62, HistorianValue.Empty, BadNoCommunication), + S(71, HistorianValue.FromDouble(30), Good), + S(100, HistorianValue.FromDouble(60), Good), + ]; + + /// Data that starts inside the window (no sample before +7) and runs past its end. + private static readonly StoredSample[] LateStart = Enumerable.Range(0, 20) + .Select(i => S(7 + i * 5, HistorianValue.FromDouble(i % 2 == 0 ? 20 : 40), Good)) + .ToArray(); + + public static TheoryData Windows => new() + { + { 0, 80, 20 }, { 80, 0, 20 }, // whole intervals (samples sit on the edges: the inclusivity question) + { 0, 50, 20 }, { 50, 0, 20 }, // a truncated interval: latest forward, EARLIEST reverse + { 5, 80, 15 }, { 80, 5, 15 }, // a misaligned window + { 3, 77, 20 }, { 77, 3, 20 }, // misaligned and truncated + { 0, 60, 0 }, { 60, 0, 0 }, // one interval + }; + + private static IReadOnlyList Ours(HistoryAggregateType agg, double from, double to, double intervalS, + StoredSample[] data) + { + var plan = BucketPlan.Create(At(from), At(to), TimeSpan.FromSeconds(intervalS)); + if (AggregateFamilies.FamilyOf(agg) == AggregateFamily.TimeWeighted) + return TimeWeightedAggregator.Evaluate(agg, plan, data, Now); + return AggregateEvaluator.EvaluateAll(agg, plan, InMemoryRows(plan, data), data.Min(s => s.TimeUtc), Now); + } + + private static IReadOnlyList Sdk(HistoryAggregateType agg, double from, double to, + double intervalS, StoredSample[] data) + { + var interval = intervalS > 0 ? TimeSpan.FromSeconds(intervalS) : (At(to) - At(from)).Duration(); + return SdkAggregateOracle.Run(agg, At(from), At(to), interval, data.Select(s => (s.TimeUtc, s.Value, s.Status))); + } + + /// Masks the Partial bit of our one truncated interval (a documented divergence: ours flags it, the SDK does not). + private static IReadOnlyList MaskTruncation(IReadOnlyList ours, double from, + double to, double intervalS) + { + var plan = BucketPlan.Create(At(from), At(to), TimeSpan.FromSeconds(intervalS)); + var list = ours.ToList(); + for (long i = 0; i < plan.BucketCount; i++) + { + if (!plan.IsTruncated(i)) continue; + var k = (int)(plan.Reversed ? plan.BucketCount - 1 - i : i); + list[k] = list[k] with { StatusCode = list[k].StatusCode & ~HistorianInfoBits.HistorianPartial }; + } + return list; + } + + [Theory] + [MemberData(nameof(Windows))] + public void Time_weighted_aggregates_match_the_SDK(double from, double to, double intervalS) + { + foreach (var agg in new[] + { + HistoryAggregateType.TimeAverage, HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total, + HistoryAggregateType.Total2, HistoryAggregateType.Interpolative, + }) + foreach (var (name, data) in new[] { ("Bounded", Bounded), ("Dense", Dense) }) + SdkAggregateOracle.ShouldMatch(MaskTruncation(Ours(agg, from, to, intervalS, data), from, to, intervalS), + Sdk(agg, from, to, intervalS, data), $"{agg} {name} {from}→{to}/{intervalS}s", + compareCalculatedInterpolatedBits: !(agg == HistoryAggregateType.Interpolative && from > to), + compareCodeBits: agg is not (HistoryAggregateType.TimeAverage or HistoryAggregateType.Total) + && !(agg == HistoryAggregateType.Interpolative && from > to)); + } + + [Theory] + [MemberData(nameof(Windows))] + public void Per_sample_aggregates_match_the_SDK(double from, double to, double intervalS) + { + foreach (var agg in new[] + { + HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, + HistoryAggregateType.Count, + }) + SdkAggregateOracle.ShouldMatch(Ours(agg, from, to, intervalS, Dense), Sdk(agg, from, to, intervalS, Dense), + $"{agg} Dense {from}→{to}/{intervalS}s", compareCalculatedInterpolatedBits: false); + } + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.TimeAverage2)] + [InlineData(HistoryAggregateType.Total)] + [InlineData(HistoryAggregateType.Total2)] + [InlineData(HistoryAggregateType.Minimum)] + [InlineData(HistoryAggregateType.Maximum)] + [InlineData(HistoryAggregateType.Count)] + [InlineData(HistoryAggregateType.Average)] + public void The_interval_the_data_starts_inside_matches_the_SDK_Partial_bit(HistoryAggregateType agg) + { + // Forward [0,40) at 10 s: the data starts at +7, inside [0,10) — the SDK sets Partial there (m_startOfData) for + // every calculator but plain Average; [10,20) … are fully covered. + var tw = AggregateFamilies.FamilyOf(agg) == AggregateFamily.TimeWeighted; + var ours = Ours(agg, 0, 40, 10, LateStart); + SdkAggregateOracle.ShouldMatch(ours, Sdk(agg, 0, 40, 10, LateStart), $"{agg} LateStart", + compareCalculatedInterpolatedBits: tw, compareValues: agg != HistoryAggregateType.Total); + (ours[0].StatusCode & HistorianInfoBits.HistorianPartial).ShouldBe( + agg == HistoryAggregateType.Average ? 0u : HistorianInfoBits.HistorianPartial); + ours.Skip(1).ShouldAllBe(x => (x.StatusCode & HistorianInfoBits.HistorianPartial) == 0); + } + + /// The per-bucket statistics the backends' SQL computes, built in memory with . + internal static IEnumerable InMemoryRows(BucketPlan plan, IEnumerable samples) + { + foreach (var g in samples.Select(s => (s, b: plan.IndexOf(s.TimeUtc))).Where(x => x.b is not null) + .GroupBy(x => x.b!.Value).OrderBy(g => g.Key)) + { + var all = g.Select(x => x.s).OrderBy(s => s.TimeUtc).ToList(); + var usable = all.Where(s => s.IsUsable).ToList(); + var numeric = usable.Where(s => s.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64 + or HistorianValueKind.Boolean).ToList(); + yield return new BucketRow + { + Bucket = g.Key, + Total = all.Count, + Usable = usable.Count, + Good = all.Count(s => s.IsGood), + Bad = all.Count - usable.Count, + NonNumeric = usable.Count - numeric.Count, + Avg = numeric.Count == 0 ? null : numeric.Average(s => s.Value.Number), + Min = numeric.Count == 0 ? null : numeric.Min(s => s.Value.Number), + Max = numeric.Count == 0 ? null : numeric.Max(s => s.Value.Number), + Sum = numeric.Count == 0 ? null : numeric.Sum(s => s.Value.Number), + First = usable.Count == 0 ? HistorianValue.Empty : usable[0].Value, + Last = usable.Count == 0 ? HistorianValue.Empty : usable[^1].Value, + }; + } + } +} 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 index e93d14b2..d9432df5 100644 --- 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 @@ -322,14 +322,46 @@ public sealed class TimeWeightedAggregatorTests // ---------------------------------------------------------------- plan shape [Fact] - public void A_reverse_window_is_computed_forward_and_returned_descending() + public void A_reverse_window_is_returned_latest_first_stamped_with_later_edges() { + // #677: the four whole intervals are the same spans as forward, but each is stamped with its later edge. 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)]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(80), At(60), At(40), At(20)]); Values(s, 50, 50, 45, 17.5); } + [Fact] + public void A_reverse_window_anchors_at_its_later_time_and_truncates_the_earliest_interval() + { + // Tag A, At(50) → At(0), 20 s ⇒ (30,50], (10,30], (0,10]. Raw signal: [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,50) 50 extrapolated (500). + // (30,50]: 950/20 = 47.5 Good; (10,30]: 200/10 = 20, the Bad 10 s ⇒ Uncertain; (0,10]: 150/10 = 15, truncated ⇒ + // Partial. The old layout would have answered [0,20), [20,40), [40,50) stamped 0/20/40, partial at the LATEST end. + var s = TimeWeightedAggregator.Evaluate(HistoryAggregateType.TimeAverage2, + BucketPlan.Create(At(50), At(0), TimeSpan.FromSeconds(20)), A, Now); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(50), At(30), At(10)]); + Values(s, 47.5, 20, 15); + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalcPartial]); + } + + [Fact] + public void The_interval_the_data_starts_inside_is_Partial() + { + // #677 (SDK m_startOfData): no sample before +5 at all ⇒ [0,10) holds the start of the data ⇒ Partial (its first + // 5 s are no-data time ⇒ Uncertain); [10,20) is fully covered ⇒ no Partial. + StoredSample[] late = [S(5, HistorianValue.FromDouble(10), Good), S(30, HistorianValue.FromDouble(10), Good)]; + var s = Run(HistoryAggregateType.TimeAverage, 0, 20, 10, late); + Values(s, 10, 10); + s.Select(x => x.StatusCode).ShouldBe([UncertainCalc | 0x4u, GoodCalc]); + // A bound before the window (the data started earlier) ⇒ no Partial. + StoredSample[] early = [S(-5, HistorianValue.FromDouble(10), Good), S(30, HistorianValue.FromDouble(10), Good)]; + Run(HistoryAggregateType.TimeAverage, 0, 20, 10, early).ShouldAllBe(x => x.StatusCode == GoodCalc); + // Data starting exactly on an interval edge covers that interval ⇒ no Partial. + StoredSample[] edge = [S(10, HistorianValue.FromDouble(10), Good), S(30, HistorianValue.FromDouble(10), Good)]; + Run(HistoryAggregateType.TimeAverage, 0, 20, 10, edge)[1].StatusCode.ShouldBe(GoodCalc); + } + [Fact] public void A_truncated_last_bucket_uses_its_own_duration_for_Total_and_is_Partial() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.csproj index 487e0dd4..f52434f3 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.csproj +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.csproj @@ -15,12 +15,18 @@ + + all runtime; build; native; contentfiles; analyzers; buildtransitive + + + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj index 4da917dd..732de3bb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj @@ -13,6 +13,8 @@ + + all runtime; build; native; contentfiles; analyzers; buildtransitive 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 f148eed7..b23366a3 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -277,6 +277,94 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// + /// #677: a client AggregateConfiguration that differs from the server defaults is refused per node with + /// BadAggregateConfigurationRejected (it used to be silently ignored); the source is never invoked. + /// + [Theory] + [InlineData(true, (byte)100, (byte)100, false)] // TreatUncertainAsBad + [InlineData(false, (byte)50, (byte)100, false)] // PercentDataBad + [InlineData(false, (byte)100, (byte)80, false)] // PercentDataGood + [InlineData(false, (byte)100, (byte)100, true)] // UseSlopedExtrapolation + [InlineData(false, (byte)0, (byte)0, false)] // an all-zero structure without UseServerCapabilitiesDefaults + public async Task Processed_with_a_non_default_aggregate_configuration_is_rejected( + bool treatUncertainAsBad, byte percentBad, byte percentGood, bool sloped) + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() }; + nm.HistorianDataSource = fake; + nm.EnsureVariable("eq-1/cfg", parentFolderNodeId: null, displayName: "Cfg", dataType: "Float", + writable: false, historianTagname: "WW.Cfg", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/cfg")!.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_TimeAverage }, + AggregateConfiguration = new AggregateConfiguration + { + UseServerCapabilitiesDefaults = false, TreatUncertainAsBad = treatUncertainAsBad, + PercentDataBad = percentBad, PercentDataGood = percentGood, UseSlopedExtrapolation = sloped, + }, + }, nodeId); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadAggregateConfigurationRejected); + fake.LastCall.ShouldBeNull(); + + await host.DisposeAsync(); + } + + /// + /// #677: UseServerCapabilitiesDefaults (whatever the other fields say), or a configuration spelling out exactly + /// the server defaults, is served. + /// + [Theory] + [InlineData(true, true, (byte)0, (byte)0, true)] + [InlineData(false, false, (byte)100, (byte)100, false)] + public async Task Processed_with_the_server_default_aggregate_configuration_is_served( + bool useServerDefaults, bool treatUncertainAsBad, byte percentBad, byte percentGood, bool sloped) + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() }; + nm.HistorianDataSource = fake; + nm.EnsureVariable("eq-1/cfg-ok", parentFolderNodeId: null, displayName: "CfgOk", dataType: "Float", + writable: false, historianTagname: "WW.CfgOk", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/cfg-ok")!.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_TimeAverage }, + AggregateConfiguration = new AggregateConfiguration + { + UseServerCapabilitiesDefaults = useServerDefaults, TreatUncertainAsBad = treatUncertainAsBad, + PercentDataBad = percentBad, PercentDataGood = percentGood, UseSlopedExtrapolation = sloped, + }, + }, nodeId); + + StatusCode.IsGood(errors[0].StatusCode).ShouldBeTrue(); + fake.LastCall.ShouldBe("Processed"); + fake.LastAggregate.ShouldBe(HistoryAggregateType.TimeAverage); + + await host.DisposeAsync(); + } + + /// #677: the server's declared aggregate defaults are the SDK's own server defaults. + [Fact] + public void The_server_aggregate_defaults_agree_with_the_SDK_AggregateManager() + { + var sdk = new Opc.Ua.Server.AggregateManager(null!).GetDefaultConfiguration(null); + var ours = ServerAggregateConfiguration.Defaults; + ours.TreatUncertainAsBad.ShouldBe(sdk.TreatUncertainAsBad); + ours.PercentDataBad.ShouldBe(sdk.PercentDataBad); + ours.PercentDataGood.ShouldBe(sdk.PercentDataGood); + ours.UseSlopedExtrapolation.ShouldBe(sdk.UseSlopedExtrapolation); + ServerAggregateConfiguration.IsServable(sdk).ShouldBeTrue(); + ServerAggregateConfiguration.IsServable(null).ShouldBeTrue(); + } + /// #285: a Part 13 aggregate beyond the original five (Interpolative) is forwarded to the /// backend as its own member. [Fact]