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; /// /// Gitea #693: HistoryRead AtTime on the Gateway backend, served from RAW gateway reads and evaluated with the SQL backends' /// shared (Part 11 §6.5.4) — as #681 did for the per-sample aggregates — so the Gateway /// answers exactly what SqlServer and QuestDB answer for the same stored samples. AVEVA's own interpolated answer (the /// gateway's ReadAtTime) is not used: live, it returned the previous value as Good at a stored NULL's own time, /// null + Bad at a Good sample right after a NULL, a drifted value (99.99999999944) at an exact stored 100, and no /// Interpolated bit. /// /// /// /// What each requested time needs. prev = the latest stored sample at or before it, next = the /// first stored sample after it. next is only read when the evaluator uses it (a usable prev that is not /// an exact hit). /// /// /// Read plan — O(1) reads for a dense batch, never two round-trips per timestamp by default. The distinct /// requested times are walked in ascending order with ONE forward window read [t − 1 ms, max + 1 ms] of at most /// rows, which yields every in-window neighbour. The sample before the first time comes from a /// walk-back (below), the sample after the last time from one small forward probe. A window read that fills its budget /// (dense data, widely spread times) resolves the times before its last row; the next unresolved time either /// continues from that row (when the rows in between are expected to fit one read) or jumps there with a read that /// starts a density-sized margin before it (so its prev is normally in the same read). Ten times one second /// apart on a 1 Hz tag cost 1 read (all exact hits) or 3 (interpolated); ten times an hour apart about one read each. /// /// /// Walk-back (the gateway reads forward only), in two phases. Gallop: segments (lo, hi] stepping /// back from hi, each one forward read, the first sized from the density the window read showed (else /// ), an empty one growing the next ×4; after two empty segments the whole remainder down /// to the epoch is asked once — a complete answer ends the walk, a FULL one seeds the bisection with its last row. /// Bisect (once a stored row is known below hi — a full segment's last row, or the last row of an earlier /// window read): read (mid, hi]; empty ⇒ hi = mid; full ⇒ the known row moves up to its last row; /// complete with rows ⇒ its last row is the answer. Right after a full read the next segment is aimed at about a /// quarter of a read at the density it showed (one try — an empty aimed segment falls back to halving), and below /// 2 ms the segment is the whole stretch from the known row (never a 100 ns step). /// /// /// Bounded. One request may take at most 64 + 4 × distinct times raw reads (, /// ); past it — or with a strict-gateway ceiling below rows — the /// node fails BadTooManyOperations with a message naming the limit. A jump read (to a time far past the previous /// read) asks for only rows, so a thousand times spread over ten years cost about a thousand /// small reads, not a thousand 1,024-row reads. /// /// /// Wire shape. Every request bound is a whole millisecond (AVEVA stores milliseconds and truncates request /// times); a row counts only strictly INSIDE its request — AVEVA's synthetic boundary row at the query start (the value /// in effect there, Quality = 133, or a NULL row) can never be taken for a stored sample, nor a possible row at /// the query end; the epoch and before hold nothing. A Bad row (a stored NULL arrives as value 0, q=1 opc=0, /// #690) is value-less. Every stream is drained, never abandoned (a cancelled gateway stream evicts its pooled /// historian session). Nothing is asked past now + (the /// historian stores nothing further ahead, and the gateway's AVEVA may refuse a far-future end); a strict gateway's /// row ceiling (#691.3) is learned and honoured — a short read is ordinary paging here, never re-probed. /// /// /// Values. A gateway value is a Double or a String whatever the node's data type (AVEVA stores every historized /// type as a numeric analog — a Boolean as a 0/1 Int2). The node's type hint converts each row BEFORE the /// shared evaluator runs, so its own type rules apply exactly as on the SQL backends: Boolean (value ≠ 0) and String /// are stepped; an integer type is an Int64 — interpolated, rounded half away from zero — and the result is returned /// as the node's integer type; Float / Double (and no hint) interpolate linearly as Double. /// /// internal static class GatewayAtTimeReader { /// The row budget of a window read (the first, or one continuing from a full one) and a walk-back segment. internal const int ReadRows = 1024; /// The row budget of a jump read (a window read restarting at a time far past the previous read's last row). internal const int JumpRows = 64; /// The row budget of the forward probe for the sample after the last time (boundary row + ties + one). internal const int ProbeRows = 2 + RawAggregateReader.PreWindowAllowance; /// The smallest row budget a read can work with (the boundary row, a tie, one sample beyond). internal const int MinBudget = 4; /// Raw reads one request may always take … internal const int BaseReads = 64; /// … plus this many per distinct requested time. internal const int ReadsPerTime = 4; /// The first walk-back segment's width when the window read showed no density. internal static readonly TimeSpan InitialWalkWidth = TimeSpan.FromMinutes(1); /// The widest walk-back segment sized from density, before ×4 growth. private static readonly TimeSpan MaxSeededWidth = TimeSpan.FromDays(1); private const long Ms = TimeSpan.TicksPerMillisecond; private static readonly TimeSpan OneMs = TimeSpan.FromMilliseconds(1); private static readonly DateTime Floor = RawAggregateReader.ProbeFloor; /// One drained read: the rows strictly inside the request (ascending) and whether the stream ended short. private readonly record struct Read(List Rows, bool Complete); /// Evaluates AtTime for (same length and order, duplicates included). /// The gateway client. /// The data source's learned gateway row ceiling (#691.3). /// Historian tag name. /// Requested times (any kind; treated as UTC, as on the SQL backends). /// Server timestamp, the beyond-now rule and the look-ahead horizon. /// Cancellation. /// The node's scalar value type (the hinted IHistorianDataSource.ReadAtTimeAsync), or null. /// The request's raw-read cap; null = + per distinct time. /// One value per requested time, stamped with it. /// More than rows within one millisecond, the /// read cap, or a strict gateway ceiling below . public static async Task> ReadAsync(IHistorianGatewayClient client, GatewayRowCeiling ceiling, string tag, IReadOnlyList timestampsUtc, DateTime now, CancellationToken ct, TypeCode? valueType = null, int? maxReads = null) { var requested = timestampsUtc.Select(t => DateTime.SpecifyKind(t, DateTimeKind.Utc)).ToArray(); if (requested.Length == 0) return []; var times = requested.Distinct().Order().ToArray(); var ctx = new Ctx(client, ceiling, tag, ct, valueType, maxReads ?? BaseReads + ReadsPerTime * times.Length); var neighbours = await ResolveAsync(ctx, times, now).ConfigureAwait(false); var result = new DataValueSnapshot[requested.Length]; for (var i = 0; i < requested.Length; i++) { var t = requested[i]; var (prev, next) = neighbours[t.Ticks]; // A DateTime node is stepped (as on the SQL backends): a usable non-numeric next sends the evaluator down its // stepped branch, with the next's own status. if (valueType == TypeCode.DateTime && next is { IsUsable: true } n) next = n with { Value = HistorianValue.FromString(string.Empty) }; result[i] = ToNodeType(AtTimeEvaluator.Evaluate(t, prev, next, now, now), valueType); } return result; } /// One request's read context: the client, the type hint, and the read count against its cap. private sealed class Ctx(IHistorianGatewayClient client, GatewayRowCeiling ceiling, string tag, CancellationToken ct, TypeCode? valueType, int maxReads) { public IHistorianGatewayClient Client { get; } = client; public GatewayRowCeiling Ceiling { get; } = ceiling; public string Tag { get; } = tag; public CancellationToken Ct { get; } = ct; public TypeCode? ValueType { get; } = valueType; public int MaxReads { get; } = maxReads; public int Reads { get; set; } /// The smallest budget a read was served with (a strict gateway's ceiling shows here). public int SmallestBudget { get; set; } = int.MaxValue; } /// True for the integer type codes (evaluated as Int64, returned as the node's own type). private static bool IsInteger(TypeCode? t) => t is TypeCode.SByte or TypeCode.Byte or TypeCode.Int16 or TypeCode.UInt16 or TypeCode.Int32 or TypeCode.UInt32 or TypeCode.Int64 or TypeCode.UInt64; /// /// An evaluated Int64 (integer hint) returned as the node's integer type. A value the type cannot hold (only possible /// for a stored value already outside the node's range) stays the evaluator's Int64. /// private static DataValueSnapshot ToNodeType(DataValueSnapshot v, TypeCode? valueType) { if (v.Value is not long l || !IsInteger(valueType) || valueType == TypeCode.Int64) return v; try { return v with { Value = Convert.ChangeType(l, valueType!.Value, System.Globalization.CultureInfo.InvariantCulture) }; } catch (OverflowException) { return v; } } /// prev / next for every time of (distinct, ascending). private static async Task> ResolveAsync( Ctx ctx, DateTime[] times, DateTime now) { var result = new Dictionary(times.Length); var horizon = MsFloor(DateTime.MaxValue - now > RawAggregateReader.LookAheadHorizon ? DateTime.SpecifyKind(now + RawAggregateReader.LookAheadHorizon, DateTimeKind.Utc) : DateTime.MaxValue.AddTicks(-Ms)); // Nothing is stored at or before the epoch. var i = 0; for (; i < times.Length && times[i] <= Floor; i++) result[times[i].Ticks] = (null, null); if (i == times.Length) return result; // Reads are planned on times clamped to the horizon (nothing is stored beyond it, and no read ends past it). var qe = Min(MsFloor(Min(times[^1], horizon)) + OneMs, horizon); var qs = RequestStart(Min(times[i], horizon)); var budget = ReadRows; StoredSample? before = null; // the latest stored sample at or before qs … var beforeKnown = false; // … once known DateTime? walkKnown = null; // a stored sample exists here (at or before qs): the walk-back bisects above it var walkWidth = InitialWalkWidth; var probeFrom = -1; // first index whose next lies at or after qe (complete last read) while (true) { var read = await ReadTrustedAsync(ctx, qs, qe, budget).ConfigureAwait(false); var rows = read.Rows; if (!read.Complete && rows.Count == 0) throw TooDense(ctx, qs); if (rows.Count >= 2 && !beforeKnown && walkKnown is null) walkWidth = Clamp(TimeSpan.FromTicks(4 * (rows[^1].TimeUtc - rows[0].TimeUtc).Ticks / (rows.Count - 1))); // A short stream saw every row up to qe; a full one every row before its last row's time x. var x = read.Complete ? qe : rows[^1].TimeUtc; var j = -1; // index of the last row at or before the current time var k = i; for (; k < times.Length; k++) { var t = times[k]; var tr = Min(t, horizon); if (!read.Complete && tr >= x) break; while (j + 1 < rows.Count && rows[j + 1].TimeUtc <= tr) j++; StoredSample? prev; if (j >= 0) prev = rows[j]; else { if (!beforeKnown) { before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false); beforeKnown = true; } prev = before; } StoredSample? next = null; if (NeedsNext(t, prev) && t < horizon) { if (j + 1 < rows.Count) next = rows[j + 1]; else if (probeFrom < 0) probeFrom = k; // after every row of a complete read: the probe answers it } result[t.Ticks] = (prev, next); } if (k == times.Length) break; // Truncated before times[k]: continue from the last row, or jump to times[k] with a density-sized margin and a // small budget (it needs that time's neighbours, and whatever later times follow closely). var tn = Min(times[k], horizon); var spanTicks = Math.Max(1, (x - rows[0].TimeUtc).Ticks); var density = (double)rows.Count / spanTicks; // rows per tick var gapRows = density * (tn - x).Ticks; var margin = Clamp(TimeSpan.FromTicks((long)Math.Min(TimeSpan.MaxValue.Ticks / 2.0, JumpRows / 4.0 / Math.Max(density, 1e-18)))); var resume = MsFloor(x) - OneMs; var jump = gapRows >= ReadRows / 2.0; var nextQs = jump ? Max(RequestStart(tn - Min(margin, tn - Floor)), resume) : resume; if (nextQs <= qs) throw TooDense(ctx, x); if (nextQs < x) { // Every row before x was seen: the latest at or before nextQs is known. var lastIndex = rows.FindLastIndex(r => r.TimeUtc <= nextQs); if (lastIndex >= 0) { before = rows[lastIndex]; beforeKnown = true; } walkKnown = null; } else { beforeKnown = false; before = null; walkKnown = x; walkWidth = margin; } budget = jump ? JumpRows : ReadRows; qs = nextQs; i = k; } if (probeFrom >= 0 && qe < horizon) { var after = await FirstAfterAsync(ctx, qe - OneMs, horizon, Min(times[^1], horizon)).ConfigureAwait(false); if (after is not null) for (var k = probeFrom; k < times.Length; k++) { var t = times[k]; var (prev, next) = result[t.Ticks]; if (next is null && t < horizon && NeedsNext(t, prev) && after.Value.TimeUtc > t) result[t.Ticks] = (prev, after); } } return result; } /// The evaluator reads next only for a usable prev that is not an exact hit. private static bool NeedsNext(DateTime t, StoredSample? prev) => prev is { IsUsable: true } p && p.TimeUtc != t; /// One walk-back segment (lo, hi]: its oldest and newest rows, and whether the read was complete. private static async Task<(StoredSample? First, StoredSample? Last, bool Complete)> SegmentAsync( Ctx ctx, DateTime lo, DateTime hi) { var read = await ReadTrustedAsync(ctx, MsFloor(lo), MsFloor(hi) + OneMs, ReadRows).ConfigureAwait(false); StoredSample? first = null, last = null; foreach (var r in read.Rows) if (r.TimeUtc > lo && r.TimeUtc <= hi) { first ??= r; last = r; } return (first, last, read.Complete); } /// /// The latest stored sample with time in (Floor, hi] (see the type remarks: gallop, then bisect). /// : a stored sample exists at that time (≤ ) — the walk then bisects /// above it straight away, its first segment aimed at . /// private static async Task WalkBackAsync(Ctx ctx, DateTime hi, DateTime? known, TimeSpan width) { TimeSpan? aim = known is null ? null : width; // Gallop: no stored row known below hi yet. var empties = 0; while (known is null) { if (hi <= Floor) return null; var rest = empties >= 2; var lo = rest || (hi - Floor).Ticks <= width.Ticks ? Floor : hi - width; var seg = await SegmentAsync(ctx, lo, hi).ConfigureAwait(false); if (seg.Complete) { if (seg.Last is { } found) return found; if (lo <= Floor) return null; hi = lo; empties++; width = Grow(width); continue; } // Full — a whole-remainder read of deep history too: its last row is a stored row below hi. Bisect above it. if (seg.Last is not { } seen) throw TooDense(ctx, lo); known = seen.TimeUtc; aim = AimFrom(seg.First!.Value.TimeUtc, seen.TimeUtc); } // Bisect: a stored row exists at `known`, nothing newer than `hi` matters; the answer is the newest row in [known, hi]. var aimMissed = false; while (true) { var k = known.Value; DateTime lo; var aimed = false; if (hi - k < TimeSpan.FromMilliseconds(2)) lo = k.AddTicks(-1); // the whole stretch from the known row: never a sub-millisecond step else { lo = k + TimeSpan.FromTicks((hi - k).Ticks / 2); if (aim is { } a && hi - a > lo) { lo = hi - a; // an aimed try right after a full read — until one comes back empty (a gap: halve from then on) aimed = true; } } aim = null; var seg = await SegmentAsync(ctx, lo, hi).ConfigureAwait(false); if (seg.Complete) { if (seg.Last is { } found) return found; if (lo < k) return null; // the known row is gone (deleted meanwhile): nothing is left to find hi = lo; aimMissed |= aimed; continue; } if (seg.Last is not { } seen || seen.TimeUtc <= k) throw TooDense(ctx, k); known = seen.TimeUtc; if (!aimMissed) aim = AimFrom(seg.First!.Value.TimeUtc, seen.TimeUtc); } } /// About a quarter of a read at the density a full read showed (its rows from to ). private static TimeSpan AimFrom(DateTime first, DateTime last) => TimeSpan.FromTicks(Math.Max(Ms, (last - first).Ticks / 4)); /// The first stored sample after in (qs, end), from a small forward read. private static async Task FirstAfterAsync(Ctx ctx, DateTime qs, DateTime end, DateTime after) { foreach (var budget in new[] { ProbeRows, ReadRows }) { var read = await ReadTrustedAsync(ctx, qs, end, budget).ConfigureAwait(false); foreach (var r in read.Rows) if (r.TimeUtc > after) return r; if (read.Complete) return null; } throw TooDense(ctx, qs); } /// /// One forward read [qs, qe] (whole milliseconds), drained, keeping only rows strictly inside it and after the /// epoch — counted against the request's read cap. A strict gateway's ceiling is learned on refusal and then asked for /// directly; a full read is not re-probed (the caller pages on, it does not need the rest of this read). /// private static async Task ReadTrustedAsync(Ctx ctx, DateTime qs, DateTime qe, int budget) { if (++ctx.Reads > ctx.MaxReads) throw new HistoryReadLimitExceededException(ctx.MaxReads, $"The AtTime read of '{ctx.Tag}' needed more than {ctx.MaxReads} raw reads ({BaseReads} + {ReadsPerTime} per " + "requested time)" + CeilingNote(ctx) + "; the Gateway backend serves AtTime from raw reads — request fewer times."); var (read, used) = await ctx.Ceiling.ReadAsync(budget, b => PassAsync(ctx, qs, qe, b), (_, _) => false) .ConfigureAwait(false); if (used < ctx.SmallestBudget) ctx.SmallestBudget = used; if (used < MinBudget) throw new HistoryReadLimitExceededException(used, $"The HistorianGateway's row ceiling ({used}, its Historian:ReadLimits:MaxValuesCeiling with " + $"AllowStreamingBeyondCeiling=false) is below the {MinBudget} rows an AtTime raw read of '{ctx.Tag}' needs."); return read; } private static string CeilingNote(Ctx ctx) => ctx.SmallestBudget < JumpRows ? $", with reads bounded by the HistorianGateway's row ceiling of {ctx.SmallestBudget} (Historian:ReadLimits:MaxValuesCeiling)" : string.Empty; private static async Task PassAsync(Ctx ctx, DateTime qs, DateTime qe, int budget) { var rows = new List(); var received = 0; await foreach (var raw in ctx.Client.ReadRawAsync(ctx.Tag, qs, qe, budget, ctx.Ct).ConfigureAwait(false)) { received++; // drained to completion, never broken out of if (raw.Timestamp is null) continue; var t = raw.Timestamp.ToDateTime(); if (t > qs && t < qe && t > Floor) rows.Add(ToStored(raw, t, ctx.ValueType)); } return new Read(rows, received < budget); } /// /// A gateway row as a stored sample of the node's type (see the type remarks); a Bad row carries no value (#690: a /// stored NULL arrives as value 0). /// private static StoredSample ToStored(HistorianSample raw, DateTime t, TypeCode? valueType) { var status = SampleMapper.ToStatus(raw); var value = StatusBits.IsBad(status) ? HistorianValue.Empty : raw.HasNumericValue ? Typed(raw.NumericValue, valueType) : raw.HasStringValue ? HistorianValue.FromString(raw.StringValue) : HistorianValue.Empty; return StoredSample.FromStored(t, value, status); } /// A gateway number as the hinted kind, so the shared evaluator's own type branch decides stepped vs linear. private static HistorianValue Typed(double v, TypeCode? valueType) => valueType switch { TypeCode.Boolean => HistorianValue.FromBoolean(v != 0), TypeCode.String => HistorianValue.FromString(v.ToString("R", System.Globalization.CultureInfo.InvariantCulture)), // In Int64 range only (2^63 is exactly representable); NaN, ±∞ or out of range is no integer at all — Empty, which // StoredSample.FromStored makes BadNoData (unusable), never a saturated Good value. _ when IsInteger(valueType) => v >= -9.2233720368547758E18 && v < 9.2233720368547758E18 ? HistorianValue.FromInt64((long)Math.Round(v, MidpointRounding.AwayFromZero)) : HistorianValue.Empty, _ => HistorianValue.FromDouble(v), }; /// The window read's start for a time: a millisecond before its millisecond (never before the epoch). private static DateTime RequestStart(DateTime t) { var s = MsFloor(t) - TimeSpan.FromMilliseconds(1); return s < Floor ? Floor : s; } private static DateTime MsFloor(DateTime t) => new(t.Ticks - t.Ticks % Ms, DateTimeKind.Utc); private static DateTime Min(DateTime a, DateTime b) => a <= b ? a : b; private static DateTime Max(DateTime a, DateTime b) => a >= b ? a : b; private static TimeSpan Min(TimeSpan a, TimeSpan b) => a <= b ? a : b; private static TimeSpan Clamp(TimeSpan w) => w < TimeSpan.FromMilliseconds(1) ? TimeSpan.FromMilliseconds(1) : w > MaxSeededWidth ? MaxSeededWidth : w; private static TimeSpan Grow(TimeSpan w) => w.Ticks > TimeSpan.MaxValue.Ticks / 4 ? TimeSpan.MaxValue : w * 4; private static HistoryReadLimitExceededException TooDense(Ctx ctx, DateTime at) { var rows = Math.Min(ReadRows, ctx.SmallestBudget); return new(rows, $"The AtTime read of '{ctx.Tag}' met more than {rows} stored samples within about one millisecond of {at:O}" + (rows < ReadRows ? $" (reads are bounded by the HistorianGateway's row ceiling of {rows})" : string.Empty) + "; the Gateway backend serves AtTime from raw reads."); } }