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;
///
/// #681: serves the per-sample processed aggregates (the ,
/// and families) on the Gateway backend by
/// streaming the window's RAW samples through the gateway and evaluating them with the shared
/// Driver.Historian.Sql.Common code — for the bucket math (forward and reverse, #677) and
/// for values, status and the Partial/Calculated bits — so the Gateway answers
/// exactly what SqlServer and QuestDB answer for the same stored samples.
///
///
///
/// Forward raw reads, each consumed to completion. The gateway's ReadRaw streams oldest-first over an
/// inclusive [start, end] and refuses end < start, so a reverse read streams the same ascending range
/// and assembles the result newest-first. The window stream is classified sample by sample
/// with ([start, end) forward, (start, end] reverse) — memory is
/// O(buckets), never O(samples). No gateway stream is ever abandoned mid-way: a client-side break cancels the call,
/// and the gateway currently classifies that cancellation as a connection fault and evicts its shared historian
/// session (HistorianGateway HistorianRpcFaults.IsConnectionFault: any RpcException ⇒ true).
///
///
/// The AVEVA start-boundary row. An AVEVA Full raw read emits an initial-value row stamped exactly at
/// the query start (histsdk reverse-engineering: Quality = 133, OpcQuality = 192) carrying the value
/// in effect there, not a stored sample. The window stream is therefore requested from
/// (1 ms) BEFORE the window, so that row lands outside every bucket and can never be counted as a sample. A row
/// before the window counts only as evidence that the tag has earlier data, and only when its OPC quality is not
/// Bad — a "no earlier data" NULL row still arrives WITH a value (histsdk sets NumericValue = 0 on every
/// numeric row, StringValue = "" on a string one) but with a Bad OPC quality (captured: 1, no-data).
///
///
/// The data range. The start/end-of-data Partial rules and Count's in-range empty interval need the tag's
/// earliest sample at or before the window's later edge and latest at or after its earlier edge
/// (; a plain Average needs neither). Earlier data is a non-Bad
/// pre-window row, or — when none arrived — a -row read [,
/// start). Later data is a sample exactly at the window end (forward) from the window read, or else a
/// -row read [end, now + ] (the historian stores
/// nothing further in the future). In both small reads a row stamped exactly at the query start (a boundary row) or
/// at the query end (a possible terminal row) is not a stored sample, whatever its value.
///
///
/// Bounded. The in-window rows are capped by ServerHistorian:MaxTimeWeightedRawRows (the same knob the
/// SQL backends' time-weighted reads use): past it the node fails with
/// (⇒ BadTooManyOperations) once the stream — bounded by the
/// requested row budget — has been drained. The gateway is asked for at most cap + 2 +
/// rows, clamped at while the cap is at
/// or under it (#688, ); a stream that delivers all of them cannot be proven
/// complete and fails the same way rather than answering from a truncated window. A strict gateway whose ceiling is
/// below that budget (a ceiling under 1,000,000, or an unbounded cap) refuses it naming its ceiling; the window is
/// then re-read once with the ceiling as budget (#691.3, ), and the same
/// "delivered the whole budget ⇒ cannot be proven complete" rule applies.
///
///
/// Values and status. A sample's value is its numeric value (Double), else its string value, else Empty; its
/// status is — the same quality mapping a raw HistoryRead serves (C-002: the
/// gateway stores no quality, so AVEVA history is normally all Good). An Empty sample is clamped Bad
/// (), exactly as on the SQL backends.
///
///
internal static class RawAggregateReader
{
/// How far before the window the raw stream starts, so the AVEVA initial-value row lands outside it.
internal static readonly TimeSpan BoundaryMargin = TimeSpan.FromMilliseconds(1);
/// How far past "now" the stream may look for a sample after the window (end-of-data rules).
internal static readonly TimeSpan LookAheadHorizon = TimeSpan.FromDays(1);
/// Rows the stream may spend before the window (the boundary row plus samples inside the margin).
internal const int PreWindowAllowance = 16;
/// Start of the "is there data before the window?" probe (the Unix epoch — before any AVEVA history).
internal static readonly DateTime ProbeFloor = DateTime.UnixEpoch;
/// Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample).
internal const int ProbeRows = 2;
///
/// The ReadRaw start for a window that begins at :
/// earlier, so the AVEVA start-boundary row — stamped exactly at the QUERY start, carrying the value in effect there
/// (Quality = 133) or, with no earlier data, a NULL row (value 0, Bad OPC quality) — lands BEFORE the window
/// (). The one place both the raw-evaluated aggregates (#681) and a raw HistoryRead
/// (#687) decide what the request start is. Never before (the Unix epoch — the start of
/// the historian's time encoding, the same floor the server's reverse walk-back uses, #686): a window starting less
/// than one margin after it is requested from the epoch itself.
///
/// The window's (inclusive) start.
/// The start to request.
public static DateTime RequestStartFor(DateTime windowStartUtc) =>
windowStartUtc - ProbeFloor > BoundaryMargin ? windowStartUtc - BoundaryMargin
: windowStartUtc > ProbeFloor ? ProbeFloor
: windowStartUtc;
///
/// True when a row of a forward raw stream requested from is a stored sample of the raw
/// window [windowStartUtc, windowEndUtc) — the contract
/// (exclusive end, as on the SQL backends; the gateway's own ReadRaw end is inclusive) — or, for a point read
/// (start == end, the server's tie-cluster over-fetch), of that one instant. A row before the start is the AVEVA
/// boundary row (whatever its value or quality — a stored NULL looks the same on the wire, so the timestamp is the
/// evidence) or a stored sample inside the margin; a row without a timestamp is never a sample.
///
/// The gateway row.
/// The window's (inclusive) start.
/// The window's end: exclusive, or inclusive when it equals the start.
/// Whether the row belongs to the window.
public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc, DateTime windowEndUtc) =>
raw.Timestamp is { } ts && ts.ToDateTime() is var t && IsAfterFloor(t) && t >= windowStartUtc
&& (t < windowEndUtc || (t == windowEndUtc && windowStartUtc == windowEndUtc));
///
/// A row stamped at or before (the Unix epoch) is never a stored sample: AVEVA stores nothing
/// there (the server's #686 walk-back assumes the same), and a window starting at or before the epoch gets no 1 ms
/// margin (), so AVEVA's synthetic boundary / no-data row lands exactly at its start.
///
/// The row's timestamp.
/// Whether the row can be a stored sample.
internal static bool IsAfterFloor(DateTime t) => t > ProbeFloor;
/// True for the aggregates this reader serves: the per-sample families of .
/// Requested aggregate.
/// Whether the aggregate is evaluated from raw samples.
public static bool Serves(HistoryAggregateType aggregate) =>
AggregateFamilies.FamilyOf(aggregate) is AggregateFamily.Simple or AggregateFamily.StartEnd or AggregateFamily.ActualTime;
/// The gateway max_values for a configured window cap ( when unbounded).
/// The configured cap (<= 0 = unbounded).
/// The row limit to request.
public static int MaxValuesFor(int maxWindowRows) =>
maxWindowRows <= 0 ? int.MaxValue : ClampToGatewayCeiling((long)maxWindowRows + 2 + PreWindowAllowance, maxWindowRows);
///
/// The HistorianGateway's default Historian:ReadLimits:MaxValuesCeiling. With
/// AllowStreamingBeyondCeiling=false the gateway refuses any max_values above its ceiling.
///
internal const int GatewayDefaultMaxValuesCeiling = 1_000_000;
///
/// #688: a ReadRaw row budget of plus this backend's own allowance
/// () clamped at — but never below
/// itself. The allowance (the boundary row, the margin, the completeness row) must not push a
/// budget that fits the gateway's default ceiling over it: the default MaxTimeWeightedRawRows (1,000,000) + 18
/// used to exceed it, so a gateway with AllowStreamingBeyondCeiling=false refused EVERY per-sample read. A
/// count above the ceiling is the operator's (or client's) deliberate choice for a streaming gateway and keeps its
/// allowance. The cost of clamping is that a window holding within the allowance of the ceiling cannot be proven
/// complete and fails as over the cap — the same outcome it would get one row later.
///
/// The budget including the allowance.
/// The caller's own row count (the cap, or NumValuesPerNode), > 0.
/// The max_values to request.
internal static int ClampToGatewayCeiling(long requested, long rows) =>
(int)Math.Min(int.MaxValue, rows > GatewayDefaultMaxValuesCeiling ? requested : Math.Min(requested, GatewayDefaultMaxValuesCeiling));
/// Maps one gateway raw sample to a stored sample (value + mapped status, Empty clamped Bad).
/// The gateway sample; its timestamp must be set.
/// The stored sample.
public static StoredSample ToStored(HistorianSample sample)
{
var value = sample.HasNumericValue ? HistorianValue.FromDouble(sample.NumericValue)
: sample.HasStringValue ? HistorianValue.FromString(sample.StringValue)
: HistorianValue.Empty;
return StoredSample.FromStored(sample.Timestamp.ToDateTime(), value, SampleMapper.ToStatus(sample));
}
/// Reads and evaluates one per-sample processed read.
/// The gateway client.
/// The data source's learned gateway row ceiling (#691.3).
/// Historian tag name.
/// A per-sample aggregate ().
/// The bucket plan.
/// The in-window raw-row cap (<= 0 = unbounded).
/// Server timestamp, and the anchor of the look-ahead horizon.
/// Cancellation token.
/// One value per interval, in the plan's order.
/// The window holds more than rows,
/// or the stream could not be read to completion within the requested row budget.
public static async Task> ReadAsync(IHistorianGatewayClient client, GatewayRowCeiling ceiling, string tag,
HistoryAggregateType aggregate, BucketPlan plan, int maxWindowRows, DateTime now, CancellationToken ct)
{
if (plan.BucketCount == 0) return [];
var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan);
var requestStart = RequestStartFor(plan.StartUtc);
// A strict gateway that refuses the budget is re-asked once at the ceiling it names (#691.3; e.g. a ceiling below
// the 1,000,000 #688 clamps at, or an unbounded cap) — and a stream that fills that ceiling is never evaluated.
var requested = MaxValuesFor(maxWindowRows);
var (pass, maxValues) = await ceiling.ReadAsync(requested,
budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct),
(r, budget) => !r.OverCap && r.Received >= budget).ConfigureAwait(false);
var (buckets, earliest, latest, received, overCap) = pass;
if (overCap) throw LimitExceeded(tag, maxWindowRows);
if (received >= maxValues) throw BudgetExhausted(tag, maxWindowRows, requested, maxValues);
if (needsRange)
{
// Earlier data: no non-Bad pre-window row ⇒ ask explicitly (normally only when the tag's history really starts
// inside the window).
var earlierSeen = earliest is { } e && (e < plan.StartUtc || (plan.Reversed && e == plan.StartUtc));
if (!earlierSeen && plan.StartUtc.AddTicks(-1) > ProbeFloor)
if (await FirstSampleAsync(client, tag, ProbeFloor, plan.StartUtc.AddTicks(-1), ct).ConfigureAwait(false) is { } before)
earliest = before;
// Later data: the main read covers a sample exactly at the window end (forward: that is already later data);
// otherwise a small separate read from the end to now + horizon.
var laterSeen = latest is { } l && (plan.Reversed ? l > plan.EndUtc : l >= plan.EndUtc);
var horizon = DateTime.MaxValue - now > LookAheadHorizon
? DateTime.SpecifyKind(now + LookAheadHorizon, DateTimeKind.Utc)
: DateTime.MaxValue;
if (!laterSeen && horizon > plan.EndUtc)
if (await FirstSampleAsync(client, tag, plan.EndUtc, horizon, ct).ConfigureAwait(false) is { } after)
latest = after;
}
return AggregateEvaluator.EvaluateAll(aggregate, plan, buckets.Rows(), earliest, latest, now);
}
/// The window read of with one row budget, drained.
private static async Task<(RawBucketAccumulator Buckets, DateTime? Earliest, DateTime? Latest, long Received, bool OverCap)>
ReadWindowAsync(IHistorianGatewayClient client, string tag, HistoryAggregateType aggregate, BucketPlan plan,
DateTime requestStart, int maxValues, int maxWindowRows, CancellationToken ct)
{
var buckets = new RawBucketAccumulator(aggregate == HistoryAggregateType.MaximumActualTime);
DateTime? earliest = null, latest = null;
long windowRows = 0, received = 0;
var overCap = false;
// Consumed to COMPLETION (never broken out of): abandoning a gateway stream cancels the call server-side, and the
// gateway currently treats that cancellation as a connection fault and evicts its shared historian session. Past
// the cap the rest is drained (bounded by maxValues) without being evaluated.
await foreach (var raw in client.ReadRawAsync(tag, requestStart, plan.EndUtc, maxValues, ct).ConfigureAwait(false))
{
received++;
if (overCap || raw.Timestamp is null) continue;
var s = ToStored(raw);
var t = s.TimeUtc;
if (!IsAfterFloor(t)) continue; // the boundary row of a window starting at/before the epoch, never a sample
if (plan.IndexOf(t) is { } bucket)
{
if (maxWindowRows > 0 && ++windowRows > maxWindowRows)
{
overCap = true;
continue;
}
buckets.Add(bucket, s);
}
// The data range as the SQL backends define it: earliest stored sample at or before the window's later edge,
// latest at or after its earlier edge. A row before the window (the AVEVA boundary row, or a stored sample in
// the margin) proves earlier data only when its OPC quality is not Bad: AVEVA's boundary row for a tag with no
// earlier data is a NULL row, which arrives WITH a value (histsdk sets NumericValue = 0 / StringValue "") and
// a Bad OPC quality (captured: 1, no-data).
if (t >= plan.StartUtc || IsNotBadOnTheWire(raw))
{
if (t <= plan.EndUtc && (earliest is null || t < earliest)) earliest = t;
if (t >= plan.StartUtc && (latest is null || t > latest)) latest = t;
}
}
return (buckets, earliest, latest, received, overCap);
}
///
/// The time of the first stored sample (any quality) strictly between and
/// , read as a -row raw read consumed to completion. A row stamped
/// exactly at the query start is the historian's boundary row (the value in effect there, or a NULL row when there is
/// none) and a row exactly at the query end may be a terminal row — neither is a stored sample, whatever its value.
///
private static async Task FirstSampleAsync(IHistorianGatewayClient client, string tag, DateTime startUtc,
DateTime endUtc, CancellationToken ct)
{
DateTime? found = null;
await foreach (var raw in client.ReadRawAsync(tag, startUtc, endUtc, ProbeRows, ct).ConfigureAwait(false))
{
if (found is not null || raw.Timestamp is null) continue; // drain, never break (see ReadAsync)
var t = raw.Timestamp.ToDateTime();
if (t > startUtc && t < endUtc) found = t;
}
return found;
}
///
/// The row's OPC DA quality alone (opc_quality ≥ 64: Good or Uncertain) — deliberately without
/// 's fallback to the AVEVA quality field when opc_quality is 0, which
/// is also the OPC DA Bad value a NULL row carries.
///
private static bool IsNotBadOnTheWire(HistorianSample raw) => (byte)raw.OpcQuality >= 64; // low byte, as SampleMapper.ToStatus reads it
///
/// The stream delivered its whole row budget, so the window cannot be proven complete. Names the limit that actually
/// bound the budget: the gateway's ceiling (a strict gateway refused the requested budget and named it, #691.3), the
/// #688 clamp at the gateway's default ceiling, or the configured cap.
///
private static HistoryReadLimitExceededException BudgetExhausted(string tag, int maxWindowRows, int requested, int budget)
{
if (budget < requested)
return new HistoryReadLimitExceededException(budget,
$"The processed read of '{tag}' covers more raw samples than the HistorianGateway's row ceiling ({budget}, its " +
"Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); the Gateway backend evaluates " +
"per-sample aggregates from raw reads — narrow the window.");
if (maxWindowRows > 0 && budget < (long)maxWindowRows + 2 + PreWindowAllowance)
return new HistoryReadLimitExceededException(budget,
$"The processed read of '{tag}' covers more raw samples than the HistorianGateway's default row ceiling " +
$"({budget}) the request is clamped at; the Gateway backend evaluates per-sample aggregates from raw reads — " +
"narrow the window.");
return LimitExceeded(tag, maxWindowRows);
}
private static HistoryReadLimitExceededException LimitExceeded(string tag, int maxWindowRows) =>
new(maxWindowRows,
$"The processed read of '{tag}' covers more than {maxWindowRows} raw samples " +
"(ServerHistorian:MaxTimeWeightedRawRows); the Gateway backend evaluates per-sample aggregates from raw reads — " +
"narrow the window.");
}
///
/// Builds the per-bucket statistics () the SQL backends' aggregate queries produce, from a stream
/// of raw samples — the C# statement of SqlAggregateQueries' columns: counts by class (usable = non-Bad with a
/// value; good = usable with Good severity; NonNumeric = usable String/DateTime; GoodNonNumeric = good String/DateTime),
/// Avg/Min/Max/Sum over the GOOD numeric values and UncertainMin/UncertainMax over the usable-but-not-Good numeric values
/// (#684), first/last usable value by time, the extreme GOOD numeric value with its time (ties ⇒ earliest), and the
/// first/last sample of any quality.
///
///
/// Keep in step with the SQL backends' per-sample value columns: a change to which samples feed a statistic there
/// (e.g. Uncertain handling) must be repeated here, or the Gateway backend silently diverges.
///
internal sealed class RawBucketAccumulator
{
private readonly bool _maximum;
private readonly Dictionary _buckets = new();
/// Creates an accumulator.
/// Extreme = the maximum (MaximumActualTime); otherwise the minimum.
public RawBucketAccumulator(bool maximum) => _maximum = maximum;
private sealed class State
{
public int Total, Usable, Good, NonNumeric, GoodNonNumeric, GoodNumeric, UncertainNumeric;
public double Sum, Min = double.PositiveInfinity, Max = double.NegativeInfinity;
public double UncertainMin = double.PositiveInfinity, UncertainMax = double.NegativeInfinity;
public StoredSample? First, Last, FirstAny, LastAny, Extreme;
}
/// Adds one in-window sample to its bucket. Order-independent (picks compare by time).
/// Ascending bucket index ().
/// The sample.
public void Add(long bucket, StoredSample s)
{
if (!_buckets.TryGetValue(bucket, out var st)) _buckets[bucket] = st = new State();
st.Total++;
if (st.FirstAny is not { } fa || s.TimeUtc < fa.TimeUtc) st.FirstAny = s;
if (st.LastAny is not { } la || s.TimeUtc >= la.TimeUtc) st.LastAny = s;
if (!s.IsUsable) return;
st.Usable++;
if (s.IsGood) st.Good++;
if (st.First is not { } f || s.TimeUtc < f.TimeUtc) st.First = s;
if (st.Last is not { } l || s.TimeUtc >= l.TimeUtc) st.Last = s;
if (!s.Value.TryGetDouble(out var v))
{
st.NonNumeric++;
if (s.IsGood) st.GoodNonNumeric++;
return;
}
if (!s.IsGood)
{
// #684: an Uncertain value is not part of the value set; it only feeds the Min/Max status rule.
st.UncertainNumeric++;
if (v < st.UncertainMin) st.UncertainMin = v;
if (v > st.UncertainMax) st.UncertainMax = v;
return;
}
st.GoodNumeric++;
st.Sum += v;
if (v < st.Min) st.Min = v;
if (v > st.Max) st.Max = v;
if (st.Extreme is not { } e) st.Extreme = s;
else
{
var ev = e.Value.Number;
var better = _maximum ? v > ev : v < ev;
if (better || (v == ev && s.TimeUtc < e.TimeUtc)) st.Extreme = s;
}
}
/// The non-empty buckets' rows, ascending by bucket index.
/// The rows.
public IEnumerable Rows() =>
_buckets.OrderBy(kv => kv.Key).Select(kv =>
{
var st = kv.Value;
var hasNumeric = st.GoodNumeric > 0;
var hasUncertain = st.UncertainNumeric > 0;
return new BucketRow
{
Bucket = kv.Key,
Total = st.Total,
Usable = st.Usable,
Good = st.Good,
Bad = st.Total - st.Usable,
NonNumeric = st.NonNumeric,
GoodNonNumeric = st.GoodNonNumeric,
Avg = hasNumeric ? st.Sum / st.GoodNumeric : null,
Min = hasNumeric ? st.Min : null,
Max = hasNumeric ? st.Max : null,
Sum = hasNumeric ? st.Sum : null,
UncertainMin = hasUncertain ? st.UncertainMin : null,
UncertainMax = hasUncertain ? st.UncertainMax : null,
First = st.First?.Value ?? HistorianValue.Empty,
Last = st.Last?.Value ?? HistorianValue.Empty,
Extreme = st.Extreme?.Value.Number,
ExtremeTime = st.Extreme?.TimeUtc,
FirstAny = st.FirstAny,
LastAny = st.LastAny,
};
});
}