fix(historian): #691.2 serve reverse native TimeAverage/Total on Gateway via forward reads on the later-anchored #677 grid

2026-10-12 16:17:05 -04:00
parent dbffb6bdb4
commit 0e47e887de
7 changed files with 349 additions and 6 deletions
@@ -249,7 +249,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    ///     <para>
    ///         <b>TimeAverage / Total</b> stay native AVEVA retrieval modes (<see cref="AggregateModeMapper"/>), stamped with
    ///         the interval start and the Calculated bit; AVEVA's extra row for the interval before the window is dropped and
-    ///         an Uncertain interval keeps its value (#685, <see cref="SampleMapper.ToAggregateSnapshots"/>). Everything else is <see cref="HistoryAggregateNotSupportedException"/>
+    ///         an Uncertain interval keeps its value (#685, <see cref="SampleMapper.ToAggregateSnapshots"/>). A reverse read
+    ///         (<c>start &gt; end</c>, which the gateway refuses) is served by forward requests on the later-anchored #677 grid
+    ///         (#691.2, <see cref="NativeReverseAggregatePlan"/>). Everything else is <see cref="HistoryAggregateNotSupportedException"/>
    ///         (see <see cref="ServedAggregates"/>).
    ///     </para>
    /// </remarks>
@@ -266,6 +268,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
        // Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
        // request the server answers BadAggregateNotSupported, not a backend fault to count (#285).
        var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
+        if (startUtc > endUtc)
+            return await ReadNativeReverseAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), mode, cancellationToken)
+                .ConfigureAwait(false);
        try
        {
            // TimeAverage / Total are native gateway retrieval modes (TimeWeightedAverage / Integral) — no
@@ -294,6 +299,44 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
        }
    }

+    /// <summary>
+    ///     A reverse (<c>start &gt; end</c>) native TimeAverage / Total read (#691.2), anchored at its later time as on the SQL
+    ///     backends (#677) — the gateway refuses <c>start &gt; end</c>, so it is served by forward native requests on the
+    ///     later-anchored grid. See <see cref="NativeReverseAggregatePlan"/>.
+    /// </summary>
+    private async Task<HistoryReadResult> ReadNativeReverseAsync(string fullReference, BucketPlan plan, RetrievalMode mode,
+        CancellationToken cancellationToken)
+    {
+        if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null);
+        try
+        {
+            var byBucket = new Dictionary<long, DataValueSnapshot>();
+            foreach (var request in NativeReverseAggregatePlan.Requests(plan))
+            {
+                var buckets = new List<HistorianAggregateSample>();
+                await foreach (var bucket in _client
+                    .ReadAggregateAsync(fullReference, request.StartUtc, request.EndUtc, mode, request.Interval, cancellationToken)
+                    .ConfigureAwait(false))
+                {
+                    buckets.Add(bucket); // drained to completion, mapped afterwards
+                }
+                NativeReverseAggregatePlan.Place(plan, request, buckets, byBucket);
+            }
+
+            RecordOutcome(success: true, error: null);
+            return new HistoryReadResult(NativeReverseAggregatePlan.Assemble(plan, byBucket), ContinuationPoint: null);
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw; // caller-requested, not a backend fault
+        }
+        catch (Exception ex)
+        {
+            RecordReadFailure(ex);
+            throw;
+        }
+    }
+
    /// <summary>Serves a per-sample aggregate from a raw read (#681); see <see cref="RawAggregateReader"/>.</summary>
    private async Task<HistoryReadResult> ReadProcessedFromRawAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
@@ -0,0 +1,103 @@
+using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
+
+/// <summary>
+///     #691.2: serves a REVERSE (<c>start &gt; end</c>) processed read of the AVEVA-native <c>TimeAverage</c> / <c>Total</c>
+///     with the #677 semantics the SQL backends give it — window <c>(end, start]</c>, intervals stepping back from the
+///     request's later time, the clipped interval the EARLIEST, each result stamped with its later edge, newest first
+///     (<see cref="BucketPlan"/>, reversed) — although the gateway refuses <c>start &gt; end</c>.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Forward requests on the later-anchored grid.</b> AVEVA anchors a native read's interval grid at the request
+///         start (live, 2026-10-12: rows at <c>start + j·interval</c>, plus one extra row for the interval before it, #685).
+///         So the whole intervals of the reverse plan — <c>(O + k·I, O + (k+1)·I]</c> with <c>O = later − n·I</c> — are
+///         requested as ONE forward read <c>[O + I·f, later)</c> with interval <c>I</c> (<c>f</c> = 1 when the earliest
+///         interval is clipped, else 0), and the clipped earliest interval <c>(earlier, O + I]</c> as a second forward read
+///         over exactly that span with its own width as the interval — so AVEVA computes it over the clipped span only, never
+///         over time before the window. Neither request is ever reversed.
+///     </para>
+///     <para>
+///         <b>Soundness.</b> A time-weighted average / integral over <c>[a, b)</c> and over <c>(a, b]</c> integrate the same
+///         function over sets differing by two instants (measure zero), so a forward native row IS the value of the
+///         reverse interval with the same edges. The one place the half-open side could matter is AVEVA's status for an
+///         interval whose only non-good sample lies exactly on an edge — accepted, as on every native row. Each row is
+///         mapped exactly as a forward native row (<see cref="SampleMapper.ToAggregateSnapshot"/>: Uncertain keeps its value,
+///         no data ⇒ <c>BadNoData</c>, Calculated bits), restamped with its interval's LATER edge; the clipped interval
+///         carries the Partial bit unless Bad (Part 13 §5.3.3.2 Case 3, as the SQL backends' time-weighted aggregates).
+///     </para>
+///     <para>
+///         <b>Missing rows are not invented.</b> As on a forward native read, an interval AVEVA answers no row for is absent
+///         from the result (not a fabricated <c>BadNoData</c>). Today historiangw#27 (the interval reaches AVEVA in ticks
+///         where it reads milliseconds) makes AVEVA answer ONE row per request — the whole-intervals read lands on its
+///         earliest interval and the rest are absent — exactly as wrong as a forward native read; this mapping needs no
+///         change once the gateway is fixed.
+///     </para>
+/// </remarks>
+internal static class NativeReverseAggregatePlan
+{
+    /// <summary>One forward native request and the ascending bucket index of its first interval.</summary>
+    /// <param name="StartUtc">Forward request start (inclusive).</param>
+    /// <param name="EndUtc">Forward request end.</param>
+    /// <param name="Interval">The request's interval.</param>
+    /// <param name="FirstBucket">Ascending plan bucket of the request's first interval.</param>
+    /// <param name="BucketCount">Number of plan buckets the request covers.</param>
+    internal readonly record struct Request(DateTime StartUtc, DateTime EndUtc, TimeSpan Interval, long FirstBucket, long BucketCount);
+
+    /// <summary>The forward requests serving a reverse plan: the whole intervals first, then the clipped earliest one.</summary>
+    /// <param name="plan">A reverse plan with at least one bucket.</param>
+    /// <returns>One or two forward requests.</returns>
+    public static IReadOnlyList<Request> Requests(BucketPlan plan)
+    {
+        var interval = TimeSpan.FromTicks(plan.IntervalTicks);
+        if (!plan.IsTruncated(0))
+            return [new Request(plan.StartUtc, plan.EndUtc, interval, 0, plan.BucketCount)];
+
+        var firstWholeStart = new DateTime(plan.UpperTicks(0), DateTimeKind.Utc); // O + I
+        var requests = new List<Request>(2);
+        if (plan.BucketCount > 1)
+            requests.Add(new Request(firstWholeStart, plan.EndUtc, interval, 1, plan.BucketCount - 1));
+        requests.Add(new Request(plan.StartUtc, firstWholeStart, firstWholeStart - plan.StartUtc, 0, 1));
+        return requests;
+    }
+
+    /// <summary>
+    ///     Maps a request's rows onto the plan's buckets: rows outside the request window (AVEVA's pre-window row, #685)
+    ///     are dropped; each in-window row is placed by its interval start (or, without one, its end).
+    /// </summary>
+    /// <param name="plan">The reverse plan.</param>
+    /// <param name="request">The forward request the rows answer.</param>
+    /// <param name="rows">The drained rows.</param>
+    /// <param name="byBucket">Results by ascending bucket index (first row per bucket wins).</param>
+    public static void Place(BucketPlan plan, Request request, IEnumerable<HistorianAggregateSample> rows,
+        Dictionary<long, DataValueSnapshot> byBucket)
+    {
+        var interval = request.Interval.Ticks;
+        foreach (var row in rows)
+        {
+            if (!SampleMapper.IsInWindow(row, request.StartUtc, request.EndUtc)) continue;
+            long offset = row.StartTime is { } s
+                ? (s.ToDateTime().Ticks - request.StartUtc.Ticks) / interval
+                : (row.EndTime.ToDateTime().Ticks - request.StartUtc.Ticks - 1) / interval; // IsInWindow ⇒ end in (lo, hi]
+            if (offset < 0 || offset >= request.BucketCount) continue;
+            var index = request.FirstBucket + offset;
+            if (byBucket.ContainsKey(index)) continue;
+
+            var snapshot = SampleMapper.ToAggregateSnapshot(row);
+            var status = snapshot.StatusCode;
+            if (plan.IsTruncated(index) && !StatusBits.IsBad(status)) status |= HistorianInfoBits.HistorianPartial;
+            byBucket[index] = snapshot with { StatusCode = status, SourceTimestampUtc = plan.BucketTimestamp(index) };
+        }
+    }
+
+    /// <summary>The placed results newest first (the reverse read's order).</summary>
+    /// <param name="plan">The reverse plan.</param>
+    /// <param name="byBucket">Results by ascending bucket index.</param>
+    /// <returns>The results, descending.</returns>
+    public static IReadOnlyList<DataValueSnapshot> Assemble(BucketPlan plan, IReadOnlyDictionary<long, DataValueSnapshot> byBucket) =>
+        byBucket.OrderByDescending(kv => kv.Key).Select(kv => kv.Value).ToList();
+}