diff --git a/CLAUDE.md b/CLAUDE.md index dbfdb79c..73b570c4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1345,7 +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. +- ⚠️ **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); on forward reads Partial also marks the interval the data starts inside (not Average — the SDK's rule, forward-only); 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 + QuestDb; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, **Primary-only** since #674 (off-Primary ⇒ `BadInvalidState`, after authorization, same `IDriverPrimaryRoleSource` verdict as the native-ack gate; reads ungated), items with a `SourceTimestamp` past now + `ServerHistorian:HistoryUpdateMaxFutureSkew` (default 1 day, ≤0 disables) ⇒ `BadOutOfRange` in the node manager (every backend), 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`. ⚠️ **SqlServer schema v1 is accepted (#674)** with `AutoCreateSchema=false`: writes/reads/events/retention work; modified reads + HistoryUpdate ⇒ `BadHistoryOperationUnsupported` (nothing sent), retention skips `SampleModified`, one Warning per boot (`SqlServerSchemaBootstrapper.SupportsModifiedTrail`; `RequiredVersion` = 1, `ModifiedTrailVersion` = 2). QuestDB: keep `cairo.ttl.use.wall.clock=true`; a QuestDB trail INSERT retried after a client timeout duplicates (no DEDUP key). diff --git a/docs/Historian.md b/docs/Historian.md index 203cb4ba..2e303d96 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -918,7 +918,7 @@ point (see [Continuation-point paging (Events)](#continuation-point-paging-event > 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 +> from the new (forward-only) 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)`, @@ -933,11 +933,14 @@ the later time and makes the buckets later-edge-inclusive — `BucketPlan.IndexO 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). +**Partial bit (#677).** On a **forward** read, 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 follows the SDK's `m_startOfData` +rule, which the SDK turns off for `Average` and in effect applies to forward reads only (a reverse read is fed latest-first, so its +"start of data" is the latest sample, past the window). A reverse read therefore sets no start-of-data Partial — matching the SDK. Only a +forward per-sample read of an aggregate other than `Average` fetches the tag's earliest sample (one seek, bounded by the window end); +every other read skips that query. 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. @@ -1035,14 +1038,18 @@ value at **10:00**, not 09:00, and no point ever falls before the window. The in `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 +bit. The per-sample comparison covers Average, Minimum, Maximum, Count, Range, Start, End, Delta, Minimum/MaximumActualTime and +PercentGood/Bad; reverse reads over data that starts inside the window are compared too. Known differences that remain (tracked as +#680) — documented and masked in the tests, deliberately not changed here because forward results were to stay as they are — are: +`Start`/`End` (the SDK returns the raw first/last sample, of any quality, with its own timestamp and stored status; ours returns the +first/last usable value stamped with the interval, under the per-sample status rule — the values agree wherever the SDK's is not Bad); +`Delta` status (the SDK's stays Good with a Bad sample in the interval, ours is Uncertain); `PercentGood`/`PercentBad` (ours count +samples, the SDK measures duration — different definitions, so only timestamp and status are compared); 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. +at a point Interpolated (Uncertain if a Bad sample follows) where ours — like the SDK forward — returns the raw sample. **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) 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 bb398d4a..e94efdf7 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 @@ -150,13 +150,14 @@ internal static class QuestDbAggregateQueries 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. + /// The tag's earliest live sample (any quality) before the window end: the start-of-data Partial rule (#677; forward + /// reads only, so the forward window's exclusive end bounds the scan). Parameters $1 tag, $2 the window + /// end. Column: ts; no row when the tag has none before the end. /// /// 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"; + $"SELECT ts FROM {sample} WHERE tag = $1 AND ts < $2 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, bool reverse) => 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 8c0b7f6b..7be1a41f 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 @@ -489,10 +489,13 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource 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))) + // The tag's earliest live sample before the window end: the start-of-data Partial rule (#677) — forward + // reads of an aggregate other than Average only; skipped otherwise. + if (AggregateEvaluator.NeedsEarliestSample(aggregate, plan)) { + await using var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.EarliestSample(sample)); QuestDbConnection.BindText(cmd, fullReference); + QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); 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); 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 053aa67e..1cac4cb7 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 @@ -80,11 +80,26 @@ public static class AggregateEvaluator /// Whether the start-of-data Partial rule applies. public static bool SetsStartOfDataPartial(HistoryAggregateType aggregate) => aggregate != HistoryAggregateType.Average; + /// + /// True when a per-sample read must fetch the tag's earliest sample: the start-of-data Partial rule applies only to a + /// forward read (as in the SDK, whose reverse read keys m_startOfData on the first value it is fed — the + /// latest) of an aggregate other than Average. Otherwise the backend skips that query. + /// + /// Requested aggregate. + /// The bucket plan. + /// Whether the earliest sample is needed. + public static bool NeedsEarliestSample(HistoryAggregateType aggregate, BucketPlan plan) + { + ArgumentNullException.ThrowIfNull(plan); + return !plan.Reversed && plan.BucketCount > 0 && SetsStartOfDataPartial(aggregate); + } + /// /// 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). + /// later edge reverse); empty buckets BadNoData; and — on a forward read, 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; the SDK applies it to forward reads only, ). /// /// A per-sample (Simple / StartEnd / ActualTime) aggregate. /// The bucket plan. @@ -98,7 +113,7 @@ public static class AggregateEvaluator ArgumentNullException.ThrowIfNull(plan); ArgumentNullException.ThrowIfNull(rows); var evaluated = new Dictionary(); - var partialRule = SetsStartOfDataPartial(aggregate); + var partialRule = NeedsEarliestSample(aggregate, plan); foreach (var row in rows) { var v = Evaluate(aggregate, row, plan.BucketTimestamp(row.Bucket), now); 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 47282367..d7a78a98 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 @@ -52,7 +52,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// 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 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 +/// within it — forward reads only, as in the SDK, #677); the timestamp is the interval's (its start forward, its /// later edge reverse); the value is a /// (Boolean = 0/1, Int64 as double). /// @@ -241,8 +241,8 @@ public sealed class TimeWeightedAggregator var stamp = _plan.BucketTimestamp(i); var duration = BucketEnd(i) - BucketStartTicks(i); // 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); + // (no bounding value before it — the SDK's m_startOfData rule, #677), which the SDK applies to forward reads only. + var partial = _plan.IsTruncated(i) || (!_plan.Reversed && _plan.DataStartsInside(i, _firstSampleUtc)); DataValueSnapshot Calc(object? value, uint status) { var code = HistorianInfoBits.WithCalculated(status); 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 430c1fc2..57117811 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 @@ -131,12 +131,13 @@ internal static class SqlAggregateQueries /// /// 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. + /// rule (#677; forward reads only). A MIN over the (TagId, SourceTimeUtc) key — a single seek — bounded + /// by the window end (a sample at or past it cannot lie inside an interval). /// /// Schema-qualified Sample table name. /// The SQL text. public static string EarliestSample(string sample) => - $"SELECT MIN(SourceTimeUtc) FROM {sample} WHERE TagId = @id;"; + $"SELECT MIN(SourceTimeUtc) FROM {sample} WHERE TagId = @id AND SourceTimeUtc < @end;"; /// The s CTE: the window's samples with bucket index and the usable/good flags. private static string Classified(string sample, bool reverse) => 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 8ec2308d..aa86f1d3 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 @@ -449,8 +449,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource 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); + // Second result set: the tag's earliest sample (the start-of-data Partial rule, #677) — forward reads of an + // aggregate other than Average only. + var needsEarliest = AggregateEvaluator.NeedsEarliestSample(aggregate, plan); + if (needsEarliest) 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 }); @@ -461,7 +463,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) rows.Add(ReadBucket(reader, family)); - if (await reader.NextResultAsync(cancellationToken).ConfigureAwait(false) + if (needsEarliest && await reader.NextResultAsync(cancellationToken).ConfigureAwait(false) && await reader.ReadAsync(cancellationToken).ConfigureAwait(false) && !reader.IsDBNull(0)) earliest = DateTime.SpecifyKind(reader.GetDateTime(0), DateTimeKind.Utc); } diff --git a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs index d4503472..ff9731a0 100644 --- a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs +++ b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs @@ -271,6 +271,34 @@ public abstract class ProcessedConformance : ConformanceBase .ShouldAllBe(x => (x.StatusCode & 0x4u) == 0); } + [Fact] + public async Task A_reverse_read_sets_no_start_of_data_Partial_and_matches_the_SDK() + { + // Review F1: the SDK applies the start-of-data rule to FORWARD reads only. Reverse 40 → 0 at 10 s over data that + // starts at +7: (0,10] holds the data's start but is not Partial — checked through the real SQL and the SDK oracle. + var rows = Enumerable.Range(0, 20) + .Select(i => (At(7 + i * 5), HistorianValue.FromDouble(i % 2 == 0 ? 20 : 40), Good)).ToArray(); + await B.SeedAsync("RLS", rows); + var sdkRows = rows.Select(r => (B.Round(r.Item1), r.Item2, r.Item3)).ToArray(); + var src = B.CreateDataSource(); + var ten = TimeSpan.FromSeconds(10); + foreach (var agg in new[] + { + HistoryAggregateType.Count, HistoryAggregateType.Minimum, HistoryAggregateType.Range, + HistoryAggregateType.Delta, HistoryAggregateType.TimeAverage, HistoryAggregateType.TimeAverage2, + HistoryAggregateType.Total2, + }) + { + var ours = (await src.ReadProcessedAsync("RLS", At(40), At(0), ten, agg, Ct)).Samples; + ours.ShouldAllBe(x => (x.StatusCode & 0x4u) == 0, agg.ToString()); + var tw = AggregateFamilies.FamilyOf(agg) == AggregateFamily.TimeWeighted; + SdkAggregateOracle.ShouldMatch(ours, SdkAggregateOracle.Run(agg, B.Round(At(40)), B.Round(At(0)), ten, sdkRows), + $"{agg} reverse late start", compareCalculatedInterpolatedBits: tw, + // pre-existing, #680: the SDK's Delta is Good with a Bad sample present; ours Uncertain (none here). + compareCodeBits: agg != HistoryAggregateType.Delta); + } + } + /// 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()); @@ -299,9 +327,13 @@ public abstract class ProcessedConformance : ConformanceBase HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Count, HistoryAggregateType.TimeAverage, HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total, HistoryAggregateType.Total2, HistoryAggregateType.Interpolative, + HistoryAggregateType.Range, HistoryAggregateType.Start, HistoryAggregateType.End, + HistoryAggregateType.Delta, HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime, + HistoryAggregateType.PercentGood, HistoryAggregateType.PercentBad, }) { var ours = (await src.ReadProcessedAsync("P", At(from), At(to), interval, agg, Ct)).Samples.ToList(); + var startOrEnd = agg is HistoryAggregateType.Start or HistoryAggregateType.End; var tw = AggregateFamilies.FamilyOf(agg) == AggregateFamily.TimeWeighted; if (tw && agg != HistoryAggregateType.Interpolative) { @@ -318,7 +350,12 @@ public abstract class ProcessedConformance : ConformanceBase 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); + // pre-existing, #680: TimeAverage/Total status rule; Start/End raw-sample timestamp + status; Delta status; + // PercentGood/Bad per-sample vs per-duration values (see SdkAggregateParityTests). + compareCodeBits: agg is not (HistoryAggregateType.TimeAverage or HistoryAggregateType.Total + or HistoryAggregateType.Delta) && !reverseInterpolative && !startOrEnd, + compareTimestamps: !startOrEnd, + compareValues: agg is not (HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad)); } } diff --git a/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs b/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs index fbe3b80c..b7dd2f78 100644 --- a/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs +++ b/tests/Drivers/Historian.Conformance/SdkAggregateOracle.cs @@ -52,6 +52,8 @@ public static class SdkAggregateOracle 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.PercentGood or HistoryAggregateType.PercentBad => + new StatusAggregateCalculator(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), @@ -84,8 +86,11 @@ public static class SdkAggregateOracle /// 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). + /// Compare the timestamps (false only for a documented timestamp divergence). + /// Compare the Partial bit (false only for a documented divergence). public static void ShouldMatch(IReadOnlyList ours, IReadOnlyList sdk, string label, - bool compareCalculatedInterpolatedBits = true, bool compareCodeBits = true, bool compareValues = true) + bool compareCalculatedInterpolatedBits = true, bool compareCodeBits = true, bool compareValues = true, + bool compareTimestamps = true, bool comparePartialBit = true) { var problems = new List(); if (ours.Count != sdk.Count) problems.Add($"interval count: ours {ours.Count}, sdk {sdk.Count}"); @@ -93,10 +98,13 @@ public static class SdkAggregateOracle { 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) + var bitMask = (comparePartialBit ? 0x4u : 0u) | (compareCodeBits ? 0xFFFF0000u : 0u) + | (compareCalculatedInterpolatedBits ? 0x3u : 0u); + var same = (!compareTimestamps || o.SourceTimestampUtc == s.Timestamp) + && (o.StatusCode & bitMask) == (s.StatusCode & bitMask); + if (!compareValues || (!compareCodeBits && StatusCode.IsBad(s.StatusCode))) { + // Values not compared (or the SDK answered Bad under a status this comparison does not check). } else if (same && !StatusCode.IsBad(s.StatusCode)) { @@ -139,6 +147,8 @@ public static class SdkAggregateOracle HistoryAggregateType.MinimumActualTime => ObjectIds.AggregateFunction_MinimumActualTime, HistoryAggregateType.MaximumActualTime => ObjectIds.AggregateFunction_MaximumActualTime, HistoryAggregateType.Count => ObjectIds.AggregateFunction_Count, + HistoryAggregateType.PercentGood => ObjectIds.AggregateFunction_PercentGood, + HistoryAggregateType.PercentBad => ObjectIds.AggregateFunction_PercentBad, HistoryAggregateType.Start => ObjectIds.AggregateFunction_Start, HistoryAggregateType.End => ObjectIds.AggregateFunction_End, HistoryAggregateType.Delta => ObjectIds.AggregateFunction_Delta, 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 7359757b..972bb7c2 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 @@ -132,6 +132,19 @@ public sealed class BucketPlanTests p.DataStartsInside(0, At(-5)).ShouldBeFalse(); } + [Fact] + public void The_earliest_sample_is_needed_only_for_forward_reads_of_aggregates_other_than_Average() + { + // Review F2: backends skip the earliest-sample query when its answer cannot change a result. + var fwd = BucketPlan.Create(At(0), At(40), TimeSpan.FromSeconds(10)); + var rev = BucketPlan.Create(At(40), At(0), TimeSpan.FromSeconds(10)); + AggregateEvaluator.NeedsEarliestSample(HistoryAggregateType.Count, fwd).ShouldBeTrue(); + AggregateEvaluator.NeedsEarliestSample(HistoryAggregateType.Average, fwd).ShouldBeFalse(); + AggregateEvaluator.NeedsEarliestSample(HistoryAggregateType.Count, rev).ShouldBeFalse(); + AggregateEvaluator.NeedsEarliestSample(HistoryAggregateType.Count, BucketPlan.Create(At(0), At(0), TimeSpan.Zero)) + .ShouldBeFalse(); + } + [Fact] public void Assemble_fills_missing_buckets_with_the_empty_factory() { 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 index 5d87f2fa..11ca72fd 100644 --- 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 @@ -84,7 +84,8 @@ public sealed class SdkAggregateParityTests 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); + return AggregateEvaluator.EvaluateAll(agg, plan, InMemoryRows(plan, data, agg == HistoryAggregateType.MaximumActualTime), + data.Min(s => s.TimeUtc), Now); } private static IReadOnlyList Sdk(HistoryAggregateType agg, double from, double to, @@ -130,13 +131,75 @@ public sealed class SdkAggregateParityTests [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); + var failures = new List(); + foreach (var agg in PerSample) + { + try + { + SdkAggregateOracle.ShouldMatch(Ours(agg, from, to, intervalS, Dense), Sdk(agg, from, to, intervalS, Dense), + $"{agg} Dense {from}→{to}/{intervalS}s", compareCalculatedInterpolatedBits: false, + compareValues: !PercentAggregate(agg), + // pre-existing, #680: the SDK's Start/End return the raw first/last sample (any quality) with its own + // timestamp and stored status; ours return the first/last USABLE value stamped with the interval, under the + // per-sample status rule. The SDK's Delta is Good even when a Bad sample sits in the interval; ours is + // Uncertain. Values are compared wherever the SDK's answer is not Bad. + compareTimestamps: !StartOrEnd(agg), + compareCodeBits: !StartOrEnd(agg) && agg != HistoryAggregateType.Delta); + } + catch (Exception ex) + { + failures.Add(ex.Message); + } + } + failures.ShouldBeEmpty(string.Join("\n", failures)); + } + + private static readonly HistoryAggregateType[] PerSample = + [ + HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, + HistoryAggregateType.Count, HistoryAggregateType.Range, HistoryAggregateType.Start, HistoryAggregateType.End, + HistoryAggregateType.Delta, HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime, + HistoryAggregateType.PercentGood, HistoryAggregateType.PercentBad, + ]; + + /// + /// pre-existing, #680: ours is the per-SAMPLE percentage (100 × Good samples / all samples); the SDK's + /// StatusAggregateCalculator is the per-DURATION one (Good time / interval time). Different definitions, so the + /// values are not compared; timestamp, status and Partial still are. + /// + private static bool StartOrEnd(HistoryAggregateType agg) => + agg is HistoryAggregateType.Start or HistoryAggregateType.End; + + private static bool PercentAggregate(HistoryAggregateType agg) => + agg is HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad; + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.TimeAverage2)] + [InlineData(HistoryAggregateType.Total)] + [InlineData(HistoryAggregateType.Total2)] + [InlineData(HistoryAggregateType.Minimum)] + [InlineData(HistoryAggregateType.Maximum)] + [InlineData(HistoryAggregateType.Count)] + [InlineData(HistoryAggregateType.Range)] + [InlineData(HistoryAggregateType.Delta)] + [InlineData(HistoryAggregateType.Start)] + [InlineData(HistoryAggregateType.Average)] + public void A_reverse_read_over_data_starting_inside_the_window_sets_no_start_of_data_Partial(HistoryAggregateType agg) + { + // Review F1: the SDK applies its start-of-data rule only on a FORWARD read (a reverse read keys m_startOfData on the + // first value it is fed — the latest — which is past the window here). Reverse 40 → 0 at 10 s, first sample +7: + // (0,10] holds the start of the data but is NOT Partial. + var tw = AggregateFamilies.FamilyOf(agg) == AggregateFamily.TimeWeighted; + var ours = Ours(agg, 40, 0, 10, LateStart); + SdkAggregateOracle.ShouldMatch(ours, Sdk(agg, 40, 0, 10, LateStart), $"{agg} reverse LateStart", + compareCalculatedInterpolatedBits: tw, + // pre-existing, #680: Total over the interval the data starts inside (TimeAverage × interval vs the SDK's + // integral of the covered part); Start's raw-sample timestamp/status (see Per_sample_aggregates_match_the_SDK). + compareValues: agg != HistoryAggregateType.Total, + compareTimestamps: !StartOrEnd(agg), + compareCodeBits: !StartOrEnd(agg) && agg != HistoryAggregateType.Delta); + ours.ShouldAllBe(x => (x.StatusCode & HistorianInfoBits.HistorianPartial) == 0); } [Theory] @@ -162,7 +225,7 @@ public sealed class SdkAggregateParityTests } /// The per-bucket statistics the backends' SQL computes, built in memory with . - internal static IEnumerable InMemoryRows(BucketPlan plan, IEnumerable samples) + internal static IEnumerable InMemoryRows(BucketPlan plan, IEnumerable samples, bool maximum = false) { 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)) @@ -171,6 +234,9 @@ public sealed class SdkAggregateParityTests 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(); + // ActualTime: the extreme usable value, ties ⇒ earliest (the SQL's ROW_NUMBER order). + var extreme = numeric.OrderBy(s => maximum ? -s.Value.Number : s.Value.Number).ThenBy(s => s.TimeUtc) + .Cast().FirstOrDefault(); yield return new BucketRow { Bucket = g.Key, @@ -185,6 +251,8 @@ public sealed class SdkAggregateParityTests 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, + Extreme = extreme?.Value.Number, + ExtremeTime = extreme?.TimeUtc, }; } }