fix(historian): #677 review — start-of-data Partial forward-only, bounded/skipped earliest-sample query, wider SDK parity

F1: the SDK applies its m_startOfData Partial rule to forward reads only; ours
now does too (AggregateEvaluator.EvaluateAll + TimeWeightedAggregator). New
reverse late-start parity tests through the SDK oracle (unit) and real SQL
(conformance, both backends).
F2: AggregateEvaluator.NeedsEarliestSample — the earliest-sample query runs
only for forward reads of aggregates other than Average, and is bounded by
the window end (SqlServer < @end, QuestDB ts < $2).
F3: per-sample SDK parity extended to Range, Start, End, Delta, Min/Max
ActualTime and PercentGood/Bad (oracle maps StatusAggregateCalculator);
pre-existing definitional differences masked and marked #680.
Docs: Historian.md Partial/SDK-parity sections, CLAUDE.md bullet.
This commit is contained in:
Joseph Doherty
2026-10-11 11:47:22 -04:00
parent 9120182314
commit 7349260231
12 changed files with 198 additions and 41 deletions
+1 -1
View File
@@ -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).
+17 -10
View File
@@ -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)
@@ -150,13 +150,14 @@ internal static class QuestDbAggregateQueries
internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2";
/// <summary>
/// The tag's earliest live sample (any quality): the start-of-data Partial rule (#677). Parameter <c>$1</c> tag.
/// Column: <c>ts</c>; 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 <c>$1</c> tag, <c>$2</c> the window
/// end. Column: <c>ts</c>; no row when the tag has none before the end.
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <returns>The SQL text.</returns>
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";
/// <summary>The classified subquery <c>s</c>: the window's live samples with bucket index and the usable/good flags.</summary>
private static string Classified(string sample, bool reverse) =>
@@ -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);
@@ -80,11 +80,26 @@ public static class AggregateEvaluator
/// <returns>Whether the start-of-data Partial rule applies.</returns>
public static bool SetsStartOfDataPartial(HistoryAggregateType aggregate) => aggregate != HistoryAggregateType.Average;
/// <summary>
/// True when a per-sample read must fetch the tag's earliest sample: the start-of-data Partial rule applies only to a
/// <b>forward</b> read (as in the SDK, whose reverse read keys <c>m_startOfData</c> on the first value it is fed — the
/// latest) of an aggregate other than <c>Average</c>. Otherwise the backend skips that query.
/// </summary>
/// <param name="aggregate">Requested aggregate.</param>
/// <param name="plan">The bucket plan.</param>
/// <returns>Whether the earliest sample is needed.</returns>
public static bool NeedsEarliestSample(HistoryAggregateType aggregate, BucketPlan plan)
{
ArgumentNullException.ThrowIfNull(plan);
return !plan.Reversed && plan.BucketCount > 0 && SetsStartOfDataPartial(aggregate);
}
/// <summary>
/// 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 <see cref="BucketPlan.BucketTimestamp"/> (the earlier edge forward, the
/// later edge reverse); empty buckets <c>BadNoData</c>; and — for every aggregate but <c>Average</c> — the Partial
/// bit on a non-Bad result of the bucket the tag's data starts inside (#677, the SDK's <c>m_startOfData</c> rule).
/// later edge reverse); empty buckets <c>BadNoData</c>; and — on a forward read, for every aggregate but
/// <c>Average</c> — the Partial bit on a non-Bad result of the bucket the tag's data starts inside (#677, the SDK's
/// <c>m_startOfData</c> rule; the SDK applies it to forward reads only, <see cref="NeedsEarliestSample"/>).
/// </summary>
/// <param name="aggregate">A per-sample (Simple / StartEnd / ActualTime) aggregate.</param>
/// <param name="plan">The bucket plan.</param>
@@ -98,7 +113,7 @@ public static class AggregateEvaluator
ArgumentNullException.ThrowIfNull(plan);
ArgumentNullException.ThrowIfNull(rows);
var evaluated = new Dictionary<long, DataValueSnapshot>();
var partialRule = SetsStartOfDataPartial(aggregate);
var partialRule = NeedsEarliestSample(aggregate, plan);
foreach (var row in rows)
{
var v = Evaluate(aggregate, row, plan.BucketTimestamp(row.Bucket), now);
@@ -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 ⇒
/// <c>BadAggregateInvalidInputs</c>. 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 <see cref="BucketPlan.BucketTimestamp"/> (its start forward, its
/// within it — forward reads only, as in the SDK, #677); the timestamp is the interval's <see cref="BucketPlan.BucketTimestamp"/> (its start forward, its
/// later edge reverse); the value is a
/// <see cref="double"/> (Boolean = 0/1, Int64 as double).
/// </para>
@@ -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);
@@ -131,12 +131,13 @@ internal static class SqlAggregateQueries
/// <summary>
/// The earliest stored sample of the tag (any quality), one row (<c>NULL</c> when none): the start-of-data Partial
/// rule (#677). A <c>MIN</c> over the <c>(TagId, SourceTimeUtc)</c> key — a single seek.
/// rule (#677; forward reads only). A <c>MIN</c> over the <c>(TagId, SourceTimeUtc)</c> key — a single seek — bounded
/// by the window end (a sample at or past it cannot lie inside an interval).
/// </summary>
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
/// <returns>The SQL text.</returns>
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;";
/// <summary>The <c>s</c> CTE: the window's samples with bucket index and the usable/good flags.</summary>
private static string Classified(string sample, bool reverse) =>
@@ -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);
}
@@ -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);
}
}
/// <summary>Dense: every 2.5 s from −10 to +100, values cycling, one Bad sample at +22.5 (no empty interval).</summary>
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));
}
}
@@ -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
/// <param name="compareCalculatedInterpolatedBits">Also compare the Calculated / Interpolated aggregate bits.</param>
/// <param name="compareCodeBits">Compare the severity + sub-code (false only for a documented status divergence).</param>
/// <param name="compareValues">Compare the values (false only for a documented value divergence).</param>
/// <param name="compareTimestamps">Compare the timestamps (false only for a documented timestamp divergence).</param>
/// <param name="comparePartialBit">Compare the Partial bit (false only for a documented divergence).</param>
public static void ShouldMatch(IReadOnlyList<DataValueSnapshot> ours, IReadOnlyList<Result> 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<string>();
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,
@@ -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()
{
@@ -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<SdkAggregateOracle.Result> 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<string>();
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,
];
/// <summary>
/// pre-existing, #680: ours is the per-SAMPLE percentage (100 × Good samples / all samples); the SDK's
/// <c>StatusAggregateCalculator</c> is the per-DURATION one (Good time / interval time). Different definitions, so the
/// values are not compared; timestamp, status and Partial still are.
/// </summary>
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
}
/// <summary>The per-bucket statistics the backends' SQL computes, built in memory with <see cref="BucketPlan.IndexOf"/>.</summary>
internal static IEnumerable<BucketRow> InMemoryRows(BucketPlan plan, IEnumerable<StoredSample> samples)
internal static IEnumerable<BucketRow> InMemoryRows(BucketPlan plan, IEnumerable<StoredSample> 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<StoredSample?>().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,
};
}
}