Joseph Doherty
821763a09d
- I-1: an Empty sample is always Bad. The writer stores a non-Bad NaN/Inf as Empty+BadOutOfRange and any other non-Bad Empty as BadNoData; the updater refuses a non-Bad Empty with BadInvalidArgument; reads clamp a legacy non-Bad Empty row to BadNoData (Raw and AtTime). - Tag/source names over 400 chars are refused, never truncated: writer per-tag failure, updater BadInvalidArgument, reads not-found, events none. - HistorianInfoBits masks the historian bits before setting one. - AtTime past now, stepped from the last sample => UncertainDataSubNormal. - Bootstrapper: logger, 5 s negative cache, ClearPool + Warning when sp_releaseapplock fails. - Provider builds its shared state lazily (an invalid Schema no longer throws at Resolve); Kinds is a FrozenSet; Gateway reuses the recorder's DefaultSupportedKinds. - Cleanup: SqlHistorianDataTypes in its own file (updater dup removed), dead Invalidate and unused status consts deleted, named severity masks, alarm writer brace. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
597 lines
33 KiB
C#
597 lines
33 KiB
C#
using System.Data;
|
||
using System.Text;
|
||
using Microsoft.Data.SqlClient;
|
||
using Microsoft.Extensions.Logging;
|
||
using Microsoft.Extensions.Logging.Abstractions;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
|
||
|
||
/// <summary>
|
||
/// SQL Server read side of the historian (spec §4 "Reads"): Raw (forward / reverse / point), AtTime
|
||
/// (one statement, <c>TimeRow</c> TVP + two <c>OUTER APPLY</c>s), Events, and Processed aggregates pushed
|
||
/// down to T-SQL (<see cref="SqlAggregateQueries"/>).
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>Reads never create tags.</b> Tag names resolve through the shared <see cref="SqlServerTagCatalog"/>'s
|
||
/// read-only lookup (positive results cached); an unknown tag is a successful empty read (Raw), an
|
||
/// all-<c>BadNoData</c> read (AtTime / Processed), never an error.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Faults.</b> A SQL / bootstrap exception is recorded in <see cref="GetHealthSnapshot"/> (so a missing
|
||
/// schema with <c>AutoCreateSchema=false</c> surfaces in <c>LastError</c>) and rethrown; the node manager
|
||
/// turns it into a per-node Bad. Client-request errors — an unsupported aggregate
|
||
/// (<see cref="HistoryAggregateNotSupportedException"/> ⇒ <c>BadAggregateNotSupported</c>) and a processed
|
||
/// window over <see cref="MaxProcessedBuckets"/> intervals (<see cref="ArgumentOutOfRangeException"/>, which
|
||
/// the node manager's generic catch surfaces as <c>BadHistoryOperationUnsupported</c> for that node only)
|
||
/// — are thrown before any I/O and are not counted as backend failures.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Bad samples.</b> A stored sample is <i>usable</i> when its status severity is Good or Uncertain and it
|
||
/// carries a value; an <c>Empty</c>-kind sample is an outage marker (comms-loss <c>BadNoCommunication</c>
|
||
/// or a driver's Bad publish) and is Bad whatever its code. Raw returns every sample as stored, outage
|
||
/// markers included. Aggregates exclude Bad samples from the math, report the interval Uncertain when
|
||
/// any non-Good sample is present, and count them in PercentBad. AtTime never interpolates across a Bad
|
||
/// neighbour and reports a time inside an outage with the outage's Bad code (see <see cref="ReadAtTimeAsync"/>).
|
||
/// <b>An Empty sample is always served Bad:</b> a row with no value whose stored status is not Bad (only
|
||
/// possible for rows written before the writer forced Empty samples Bad) is read back as <c>BadNoData</c>,
|
||
/// on Raw and AtTime alike.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Long names.</b> A tag name or event source name longer than <see cref="SqlServerTagCatalog.MaxNameLength"/>
|
||
/// can never be stored, so it is answered as not found (Raw: empty; AtTime / Processed: <c>BadNoData</c>;
|
||
/// Events: none) without being truncated onto another series.
|
||
/// </para>
|
||
/// </remarks>
|
||
public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||
{
|
||
/// <summary>Spec §9.7: a processed read covering more intervals than this is refused.</summary>
|
||
public const int MaxProcessedBuckets = 100_000;
|
||
|
||
private const string SampleColumns = "SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc";
|
||
|
||
private readonly SqlHistorianConnection _conn;
|
||
private readonly SqlServerSchemaBootstrapper _bootstrap;
|
||
private readonly SqlServerTagCatalog _catalog;
|
||
private readonly ILogger _log;
|
||
private readonly TimeProvider _time;
|
||
|
||
private readonly object _healthLock = new();
|
||
private DateTime? _lastSuccessUtc;
|
||
private DateTime? _lastFailureUtc;
|
||
private string? _lastError;
|
||
private long _totalQueries;
|
||
private long _totalSuccesses;
|
||
private long _totalFailures;
|
||
private int _consecutiveFailures;
|
||
|
||
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
|
||
/// <param name="conn">Connection helper.</param>
|
||
/// <param name="bootstrap">Schema bootstrapper (shared with the other SQL seams).</param>
|
||
/// <param name="catalog">Tag catalog (shared; reads use only its read-only lookup).</param>
|
||
/// <param name="log">Optional logger.</param>
|
||
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
|
||
internal SqlServerHistorianDataSource(
|
||
SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null,
|
||
TimeProvider? timeProvider = null)
|
||
{
|
||
_conn = conn ?? throw new ArgumentNullException(nameof(conn));
|
||
_bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap));
|
||
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
|
||
_log = log ?? NullLogger.Instance;
|
||
_time = timeProvider ?? TimeProvider.System;
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ Raw
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// Forward (<c>start < end</c>): <c>[start, end)</c> ascending. Reverse (<c>end < start</c>, Part 11):
|
||
/// <c>(end, start]</c> descending. <c>start == end</c>: the sample exactly at that time (Part 11's point
|
||
/// read; it is also what the node manager's tie-cluster over-fetch asks for). <paramref name="maxValuesPerNode"/>
|
||
/// <c>0</c> is the node manager's "unbounded" sentinel; values above <see cref="int.MaxValue"/> clamp.
|
||
/// The node manager pages by re-reading from the last returned timestamp inclusive and trimming it; the
|
||
/// <c>(TagId, SourceTimeUtc)</c> primary key rules out tie clusters, so that is exact.
|
||
/// </remarks>
|
||
public async Task<HistoryReadResult> ReadRawAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(fullReference);
|
||
var top = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue);
|
||
try
|
||
{
|
||
var tagId = await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
|
||
var samples = new List<DataValueSnapshot>();
|
||
if (tagId is { } id)
|
||
{
|
||
string where, order;
|
||
if (startUtc == endUtc) { where = "SourceTimeUtc = @lo"; order = "ASC"; }
|
||
else if (startUtc < endUtc) { where = "SourceTimeUtc >= @lo AND SourceTimeUtc < @hi"; order = "ASC"; }
|
||
else { where = "SourceTimeUtc > @lo AND SourceTimeUtc <= @hi"; order = "DESC"; }
|
||
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
|
||
|
||
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||
await using var cmd = _conn.CreateCommand(conn,
|
||
$"SELECT TOP (@top) {SampleColumns} FROM {_conn.Q("Sample")} " +
|
||
$"WHERE TagId = @id AND {where} ORDER BY SourceTimeUtc {order}");
|
||
cmd.Parameters.Add(new SqlParameter("@top", SqlDbType.Int) { Value = top });
|
||
cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
|
||
SqlHistorianConnection.AddDateTime2(cmd, "@lo", lo);
|
||
SqlHistorianConnection.AddDateTime2(cmd, "@hi", hi);
|
||
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var s = ReadStored(reader, 0);
|
||
samples.Add(new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc,
|
||
SqlHistorianConnection.GetUtc(reader, 6)));
|
||
}
|
||
}
|
||
|
||
RecordSuccess();
|
||
return new HistoryReadResult(samples, ContinuationPoint: null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordFailure(ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ AtTime
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// One statement: the distinct requested times go in a <c>TimeRow</c> TVP and each picks up its
|
||
/// previous-or-equal (<c>p</c>) and next (<c>n</c>) stored sample via two <c>OUTER APPLY (SELECT TOP 1 …)</c>.
|
||
/// Rules, per requested time <c>T</c> (the result is stamped <c>T</c>):
|
||
/// <list type="bullet">
|
||
/// <item><c>p</c> at exactly <c>T</c> ⇒ the stored value and status, unchanged.</item>
|
||
/// <item>No <c>p</c> ⇒ <c>BadNoData</c>, null value.</item>
|
||
/// <item><c>p</c> Bad (incl. an <c>Empty</c> outage marker) ⇒ <c>T</c> lies inside an outage: null value
|
||
/// carrying <c>p</c>'s Bad status code (info bits stripped) — never better than its source, per the
|
||
/// quality contract (spec §4 as amended after the T10 review).</item>
|
||
/// <item><c>p</c> usable and <c>n</c> Bad ⇒ <c>p</c>'s value held (stepped, NOT interpolated into the outage)
|
||
/// with <c>UncertainDataSubNormal</c> + Interpolated bits.</item>
|
||
/// <item>Double/Int64 with a usable <c>n</c> ⇒ linear interpolation of <c>NumValue</c> (Int64 rounded to the
|
||
/// nearest integer, ties away from zero, keeping the type); status <c>Good</c> + Interpolated when both
|
||
/// neighbours are Good, else <c>UncertainDataSubNormal</c> + Interpolated (never better than the
|
||
/// inputs).</item>
|
||
/// <item>Boolean/String/DateTime, or no <c>n</c> (after the last sample) ⇒ stepped: <c>p</c>'s value with
|
||
/// <c>p</c>'s status code + Interpolated bits — except that a time later than now that is stepped from the
|
||
/// last sample is <c>UncertainDataSubNormal</c> + Interpolated (the value is not known to still hold).</item>
|
||
/// </list>
|
||
/// Unknown tag ⇒ every entry <c>BadNoData</c>.
|
||
/// </remarks>
|
||
public async Task<HistoryReadResult> ReadAtTimeAsync(
|
||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(fullReference);
|
||
ArgumentNullException.ThrowIfNull(timestampsUtc);
|
||
try
|
||
{
|
||
var now = _time.GetUtcNow().UtcDateTime;
|
||
var byTicks = new Dictionary<long, DataValueSnapshot>(timestampsUtc.Count);
|
||
var tagId = timestampsUtc.Count == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
|
||
if (tagId is { } id)
|
||
{
|
||
var distinct = timestampsUtc.Select(t => DateTime.SpecifyKind(t, DateTimeKind.Utc)).Distinct().ToList();
|
||
var cols = "s.SourceTimeUtc, s.ValueKind, s.NumValue, s.IntValue, s.StrValue, s.StatusCode, s.ServerTimeUtc";
|
||
var sql =
|
||
$"SELECT r.T, p.*, n.* FROM @times r " +
|
||
$"OUTER APPLY (SELECT TOP 1 {cols} FROM {_conn.Q("Sample")} s WHERE s.TagId = @id AND s.SourceTimeUtc <= r.T ORDER BY s.SourceTimeUtc DESC) p " +
|
||
$"OUTER APPLY (SELECT TOP 1 {cols} FROM {_conn.Q("Sample")} s WHERE s.TagId = @id AND s.SourceTimeUtc > r.T ORDER BY s.SourceTimeUtc ASC) n";
|
||
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 });
|
||
_conn.AddStructured(cmd, "@times", "TimeRow", SampleRowMapper.ToTimeRecords(distinct));
|
||
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var t = SqlHistorianConnection.GetUtc(reader, 0);
|
||
StoredSample? p = reader.IsDBNull(1) ? null : ReadStored(reader, 1);
|
||
StoredSample? n = reader.IsDBNull(8) ? null : ReadStored(reader, 8);
|
||
var serverTime = p is { } ps && ps.TimeUtc == t ? SqlHistorianConnection.GetUtc(reader, 7) : now;
|
||
byTicks[t.Ticks] = AtTimeValue(t, p, n, serverTime, now);
|
||
}
|
||
}
|
||
|
||
var result = new DataValueSnapshot[timestampsUtc.Count];
|
||
for (var i = 0; i < result.Length; i++)
|
||
{
|
||
var t = DateTime.SpecifyKind(timestampsUtc[i], DateTimeKind.Utc);
|
||
result[i] = byTicks.TryGetValue(t.Ticks, out var v)
|
||
? v
|
||
: new DataValueSnapshot(null, SqlHistorianStatusCodes.BadNoData, t, now);
|
||
}
|
||
|
||
RecordSuccess();
|
||
return new HistoryReadResult(result, ContinuationPoint: null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordFailure(ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>Applies the AtTime rules (see <see cref="ReadAtTimeAsync"/>) to one requested time.</summary>
|
||
private static DataValueSnapshot AtTimeValue(DateTime t, StoredSample? prev, StoredSample? next, DateTime serverTime, DateTime now)
|
||
{
|
||
if (prev is not { } p)
|
||
return new DataValueSnapshot(null, SqlHistorianStatusCodes.BadNoData, t, serverTime);
|
||
if (p.TimeUtc == t)
|
||
return new DataValueSnapshot(p.Value.ToObject(), p.Status, t, serverTime);
|
||
|
||
// Inside an outage: report the outage's own Bad code, never an upgraded Uncertain.
|
||
if (!p.IsUsable)
|
||
return new DataValueSnapshot(null, p.Status & StatusBits.CodeBitsMask, t, serverTime);
|
||
|
||
// Usable previous, Bad next: hold the previous value up to the outage; do not interpolate into it.
|
||
if (next is { IsUsable: false })
|
||
return new DataValueSnapshot(p.Value.ToObject(),
|
||
HistorianInfoBits.WithInterpolated(SqlHistorianStatusCodes.UncertainDataSubNormal), t, serverTime);
|
||
|
||
if (next is { } n && p.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64
|
||
&& n.Value.TryGetDouble(out var nv))
|
||
{
|
||
var fraction = (double)(t.Ticks - p.TimeUtc.Ticks) / (n.TimeUtc.Ticks - p.TimeUtc.Ticks);
|
||
var v = p.Value.Number + (nv - p.Value.Number) * fraction;
|
||
object value = p.Value.Kind == HistorianValueKind.Int64
|
||
? (long)Math.Round(v, MidpointRounding.AwayFromZero)
|
||
: v;
|
||
var status = p.IsGood && n.IsGood ? SqlHistorianStatusCodes.Good : SqlHistorianStatusCodes.UncertainDataSubNormal;
|
||
return new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), t, serverTime);
|
||
}
|
||
|
||
// Beyond the last sample AND beyond now: the value is held into the future, so it is not known to be Good.
|
||
if (next is null && t > now)
|
||
return new DataValueSnapshot(p.Value.ToObject(),
|
||
HistorianInfoBits.WithInterpolated(SqlHistorianStatusCodes.UncertainDataSubNormal), t, serverTime);
|
||
|
||
// Stepped (non-numeric kind, or after the last sample): hold the previous value and its status.
|
||
return new DataValueSnapshot(p.Value.ToObject(), HistorianInfoBits.WithInterpolated(p.Status), t, serverTime);
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ Events
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// <c>[start, end)</c> by <c>EventTimeUtc</c> (a reversed window is swapped), ordered by
|
||
/// <c>(EventTimeUtc, EventId)</c> ascending — the clustered index. <paramref name="maxEvents"/> <= 0 or
|
||
/// <see cref="int.MaxValue"/> (the node manager's "no limit") ⇒ unbounded. Otherwise <c>maxEvents + 1</c>
|
||
/// rows are fetched; when the extra row arrives it is dropped and the continuation point is the UTF-8
|
||
/// of <c>"{EventTimeUtc.Ticks}|{EventId}"</c> of that first not-returned event (opaque to callers; only
|
||
/// its presence is contractual).
|
||
/// </remarks>
|
||
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken)
|
||
{
|
||
// No stored event can carry a source longer than the column (the alarm writer dead-letters it rather than
|
||
// truncating), so an over-long filter matches nothing — never a truncated prefix of someone else's source.
|
||
if (sourceName is not null && sourceName.Length > SqlServerTagCatalog.MaxNameLength)
|
||
return new HistoricalEventsResult([], null);
|
||
try
|
||
{
|
||
await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
|
||
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
|
||
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
|
||
var sql =
|
||
$"SELECT {(bounded ? "TOP (@take) " : "")}EventId, SourceName, EventTimeUtc, ReceivedTimeUtc, Message, Severity " +
|
||
$"FROM {_conn.Q("Event")} WHERE EventTimeUtc >= @lo AND EventTimeUtc < @hi" +
|
||
(sourceName is null ? "" : " AND SourceName = @source") +
|
||
" ORDER BY EventTimeUtc ASC, EventId ASC";
|
||
|
||
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||
await using var cmd = _conn.CreateCommand(conn, sql);
|
||
if (bounded) cmd.Parameters.Add(new SqlParameter("@take", SqlDbType.BigInt) { Value = (long)maxEvents + 1 });
|
||
SqlHistorianConnection.AddDateTime2(cmd, "@lo", lo);
|
||
SqlHistorianConnection.AddDateTime2(cmd, "@hi", hi);
|
||
if (sourceName is not null)
|
||
cmd.Parameters.Add(new SqlParameter("@source", SqlDbType.NVarChar, SqlServerTagCatalog.MaxNameLength) { Value = sourceName });
|
||
|
||
var events = new List<HistoricalEvent>();
|
||
byte[]? continuation = null;
|
||
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var e = new HistoricalEvent(
|
||
reader.GetString(0),
|
||
reader.GetString(1),
|
||
SqlHistorianConnection.GetUtc(reader, 2),
|
||
SqlHistorianConnection.GetUtc(reader, 3),
|
||
reader.IsDBNull(4) ? null : reader.GetString(4),
|
||
(ushort)Math.Clamp(reader.GetInt32(5), 0, ushort.MaxValue));
|
||
if (bounded && events.Count == maxEvents)
|
||
{
|
||
continuation = Encoding.UTF8.GetBytes($"{e.EventTimeUtc.Ticks}|{e.EventId}");
|
||
break;
|
||
}
|
||
events.Add(e);
|
||
}
|
||
|
||
RecordSuccess();
|
||
return new HistoricalEventsResult(events, continuation);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordFailure(ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ Processed
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// <para>
|
||
/// One result per interval of <c>[start, end)</c>, bucket <c>i</c> = <c>[start + i·interval,
|
||
/// start + (i+1)·interval)</c> (start-inclusive, end-exclusive to the tick; the last bucket is truncated at
|
||
/// <c>end</c>). <paramref name="interval"/> <= 0 ⇒ one interval <c>[start, end)</c>. <c>start == end</c>
|
||
/// ⇒ no intervals. <c>end < start</c> (reverse) ⇒ the window is swapped, computed forward, and returned
|
||
/// in descending bucket order (buckets stay anchored at the earlier time).
|
||
/// </para>
|
||
/// <para>
|
||
/// Per interval: no samples ⇒ <c>BadNoData</c>. PercentGood / PercentBad ⇒ <c>100 × good|bad / all</c> over
|
||
/// every sample, status Good. Otherwise no usable sample ⇒ <c>BadNoData</c>; a numeric aggregate with a usable
|
||
/// String/DateTime sample ⇒ <c>BadAggregateInvalidInputs</c>; else the value, Good when every sample is Good,
|
||
/// <c>UncertainDataSubNormal</c> when any is Bad or Uncertain. Every status — including <c>BadNoData</c> —
|
||
/// carries the Calculated info bits. Timestamps are the bucket start, except Minimum/MaximumActualTime
|
||
/// (the sample's time; ties ⇒ earliest).
|
||
/// </para>
|
||
/// <para>
|
||
/// Values: Count is <see cref="int"/>; Start/End keep the sample's type; every other aggregate is
|
||
/// <see cref="double"/> computed over <c>NumValue</c> (Boolean = 0/1). These are the simple per-sample
|
||
/// definitions, not Part 13's interval-boundary (interpolated bound / time-weighted) semantics.
|
||
/// </para>
|
||
/// </remarks>
|
||
/// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not one of the thirteen served.</exception>
|
||
/// <exception cref="ArgumentOutOfRangeException">The window holds more than <see cref="MaxProcessedBuckets"/> intervals.</exception>
|
||
public async Task<HistoryReadResult> ReadProcessedAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(fullReference);
|
||
var family = SqlAggregateQueries.FamilyOf(aggregate) ?? throw new HistoryAggregateNotSupportedException(aggregate);
|
||
|
||
var reverse = endUtc < startUtc;
|
||
var (start, end) = reverse ? (endUtc, startUtc) : (startUtc, endUtc);
|
||
var windowTicks = end.Ticks - start.Ticks;
|
||
// Everything stays in 100 ns ticks (≤ ≈3.2e18 across 0001–9999, inside a long) — no ns scaling that could
|
||
// wrap. An interval wider than the window is one bucket, so clamp it to the window; the ceiling division
|
||
// is written without "+ interval - 1" so a TimeSpan near MaxValue cannot overflow either.
|
||
var intervalTicks = interval > TimeSpan.Zero ? Math.Min(interval.Ticks, windowTicks) : windowTicks;
|
||
long bucketCount = windowTicks == 0 ? 0 : windowTicks / intervalTicks + (windowTicks % intervalTicks == 0 ? 0 : 1);
|
||
if (bucketCount > MaxProcessedBuckets)
|
||
throw new ArgumentOutOfRangeException(nameof(interval), bucketCount,
|
||
$"A processed read may cover at most {MaxProcessedBuckets} intervals; this window holds {bucketCount}.");
|
||
|
||
try
|
||
{
|
||
var now = _time.GetUtcNow().UtcDateTime;
|
||
var rows = new Dictionary<long, BucketRow>();
|
||
var tagId = bucketCount == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
|
||
if (tagId is { } id)
|
||
{
|
||
var sample = _conn.Q("Sample");
|
||
var sql = family switch
|
||
{
|
||
SqlAggregateFamily.Simple => SqlAggregateQueries.Simple(sample),
|
||
SqlAggregateFamily.StartEnd => SqlAggregateQueries.StartEnd(sample),
|
||
_ => SqlAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime),
|
||
};
|
||
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 });
|
||
SqlHistorianConnection.AddDateTime2(cmd, "@start", DateTime.SpecifyKind(start, DateTimeKind.Utc));
|
||
SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(end, DateTimeKind.Utc));
|
||
cmd.Parameters.Add(new SqlParameter("@intervalTicks", SqlDbType.BigInt) { Value = intervalTicks });
|
||
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var row = ReadBucket(reader, family);
|
||
rows[row.Bucket] = row;
|
||
}
|
||
}
|
||
|
||
var result = new List<DataValueSnapshot>((int)bucketCount);
|
||
for (long i = 0; i < bucketCount; i++)
|
||
{
|
||
var bucketStart = DateTime.SpecifyKind(new DateTime(start.Ticks + i * intervalTicks), DateTimeKind.Utc);
|
||
result.Add(rows.TryGetValue(i, out var row)
|
||
? Evaluate(aggregate, row, bucketStart, now)
|
||
: NoData(bucketStart, now));
|
||
}
|
||
if (reverse) result.Reverse();
|
||
|
||
RecordSuccess();
|
||
return new HistoryReadResult(result, ContinuationPoint: null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordFailure(ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
private static DataValueSnapshot NoData(DateTime t, DateTime now) =>
|
||
new(null, HistorianInfoBits.WithCalculated(SqlHistorianStatusCodes.BadNoData), t, now);
|
||
|
||
private static DataValueSnapshot Evaluate(HistoryAggregateType aggregate, BucketRow r, DateTime bucketStart, DateTime now)
|
||
{
|
||
DataValueSnapshot Calc(object? value, uint status, DateTime? t = null) =>
|
||
new(value, HistorianInfoBits.WithCalculated(status), t ?? bucketStart, now);
|
||
|
||
if (r.Total == 0) return NoData(bucketStart, now);
|
||
switch (aggregate)
|
||
{
|
||
case HistoryAggregateType.PercentGood: return Calc(100d * r.Good / r.Total, SqlHistorianStatusCodes.Good);
|
||
case HistoryAggregateType.PercentBad: return Calc(100d * r.Bad / r.Total, SqlHistorianStatusCodes.Good);
|
||
}
|
||
if (r.Usable == 0) return NoData(bucketStart, now);
|
||
|
||
var anyText = aggregate is HistoryAggregateType.Count or HistoryAggregateType.Start or HistoryAggregateType.End;
|
||
if (!anyText && r.NonNumeric > 0)
|
||
return Calc(null, SqlHistorianStatusCodes.BadAggregateInvalidInputs);
|
||
|
||
var status = r.Good == r.Total ? SqlHistorianStatusCodes.Good : SqlHistorianStatusCodes.UncertainDataSubNormal;
|
||
switch (aggregate)
|
||
{
|
||
case HistoryAggregateType.Count: return Calc(r.Usable, status);
|
||
case HistoryAggregateType.Average: return Calc(r.Avg, status);
|
||
case HistoryAggregateType.Minimum: return Calc(r.Min, status);
|
||
case HistoryAggregateType.Maximum: return Calc(r.Max, status);
|
||
case HistoryAggregateType.Total: return Calc(r.Sum, status);
|
||
case HistoryAggregateType.Range: return Calc(r.Max - r.Min, status);
|
||
case HistoryAggregateType.Start: return Calc(r.First.ToObject(), status);
|
||
case HistoryAggregateType.End: return Calc(r.Last.ToObject(), status);
|
||
case HistoryAggregateType.Delta:
|
||
return r.First.TryGetDouble(out var a) && r.Last.TryGetDouble(out var b)
|
||
? Calc(b - a, status)
|
||
: Calc(null, SqlHistorianStatusCodes.BadAggregateInvalidInputs);
|
||
case HistoryAggregateType.MinimumActualTime:
|
||
case HistoryAggregateType.MaximumActualTime:
|
||
return Calc(r.Extreme, status, r.ExtremeTime);
|
||
default:
|
||
throw new HistoryAggregateNotSupportedException(aggregate);
|
||
}
|
||
}
|
||
|
||
private static BucketRow ReadBucket(SqlDataReader reader, SqlAggregateFamily family)
|
||
{
|
||
static double? D(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
|
||
static long? L(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetInt64(i);
|
||
static string? S(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetString(i);
|
||
static HistorianValue V(SqlDataReader r, int i) =>
|
||
r.IsDBNull(i) ? HistorianValue.Empty : SampleRowMapper.FromColumns(r.GetByte(i), D(r, i + 1), L(r, i + 2), S(r, i + 3));
|
||
|
||
var row = new BucketRow
|
||
{
|
||
Bucket = reader.GetInt64(0),
|
||
Total = reader.GetInt32(1),
|
||
Usable = reader.GetInt32(2),
|
||
Good = reader.GetInt32(3),
|
||
Bad = reader.GetInt32(4),
|
||
NonNumeric = reader.GetInt32(5),
|
||
};
|
||
var f = SqlAggregateQueries.FirstFamilyColumn;
|
||
switch (family)
|
||
{
|
||
case SqlAggregateFamily.Simple:
|
||
row.Avg = D(reader, f);
|
||
row.Min = D(reader, f + 1);
|
||
row.Max = D(reader, f + 2);
|
||
row.Sum = D(reader, f + 3);
|
||
break;
|
||
case SqlAggregateFamily.StartEnd:
|
||
row.First = V(reader, f);
|
||
row.Last = V(reader, f + 5);
|
||
break;
|
||
default:
|
||
row.Extreme = D(reader, f);
|
||
row.ExtremeTime = reader.IsDBNull(f + 1) ? null : SqlHistorianConnection.GetUtc(reader, f + 1);
|
||
break;
|
||
}
|
||
return row;
|
||
}
|
||
|
||
/// <summary>One aggregate query row (one non-empty bucket).</summary>
|
||
private sealed class BucketRow
|
||
{
|
||
public long Bucket;
|
||
public int Total, Usable, Good, Bad, NonNumeric;
|
||
public double? Avg, Min, Max, Sum, Extreme;
|
||
public DateTime? ExtremeTime;
|
||
public HistorianValue First, Last;
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ Shared
|
||
|
||
/// <summary>A stored sample read from the <c>{SampleColumns}</c> column run starting at an ordinal.</summary>
|
||
private readonly record struct StoredSample(DateTime TimeUtc, HistorianValue Value, uint Status)
|
||
{
|
||
/// <summary>Severity Good or Uncertain, and a value present (an Empty sample is an outage marker).</summary>
|
||
public bool IsUsable => !StatusBits.IsBad(Status) && Value.Kind != HistorianValueKind.Empty;
|
||
|
||
/// <summary>Usable with severity Good.</summary>
|
||
public bool IsGood => IsUsable && StatusBits.IsGood(Status);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Reads <c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode</c> starting at <paramref name="o"/>.
|
||
/// An Empty (no-value) row whose stored status is not Bad is clamped to <c>BadNoData</c> — defence in depth for
|
||
/// rows written before the writer forced every Empty sample Bad.
|
||
/// </summary>
|
||
private static StoredSample ReadStored(SqlDataReader r, int o)
|
||
{
|
||
var value = SampleRowMapper.FromColumns(
|
||
r.GetByte(o + 1),
|
||
r.IsDBNull(o + 2) ? null : r.GetDouble(o + 2),
|
||
r.IsDBNull(o + 3) ? null : r.GetInt64(o + 3),
|
||
r.IsDBNull(o + 4) ? null : r.GetString(o + 4));
|
||
var status = SampleRowMapper.FromDbStatus(r.GetInt32(o + 5));
|
||
if (value.Kind == HistorianValueKind.Empty && !StatusBits.IsBad(status))
|
||
status = SqlHistorianStatusCodes.BadNoData;
|
||
return new StoredSample(SqlHistorianConnection.GetUtc(r, o), value, status);
|
||
}
|
||
|
||
/// <summary>Read-only tag lookup (never creates a row); runs the schema bootstrap on a cache miss.</summary>
|
||
private Task<int?> ResolveAsync(string name, CancellationToken ct) => _catalog.TryGetTagIdAsync(name, ct);
|
||
|
||
// ------------------------------------------------------------------------------------------ Health
|
||
|
||
/// <inheritdoc />
|
||
public HistorianHealthSnapshot GetHealthSnapshot()
|
||
{
|
||
lock (_healthLock)
|
||
{
|
||
var open = _totalSuccesses > 0 && _consecutiveFailures == 0;
|
||
return new HistorianHealthSnapshot(
|
||
TotalQueries: _totalQueries,
|
||
TotalSuccesses: _totalSuccesses,
|
||
TotalFailures: _totalFailures,
|
||
ConsecutiveFailures: _consecutiveFailures,
|
||
LastSuccessTime: _lastSuccessUtc,
|
||
LastFailureTime: _lastFailureUtc,
|
||
LastError: _lastError,
|
||
ProcessConnectionOpen: open,
|
||
EventConnectionOpen: open,
|
||
ActiveProcessNode: null,
|
||
ActiveEventNode: null,
|
||
Nodes: []);
|
||
}
|
||
}
|
||
|
||
private void RecordSuccess()
|
||
{
|
||
lock (_healthLock)
|
||
{
|
||
_totalQueries++;
|
||
_totalSuccesses++;
|
||
_consecutiveFailures = 0;
|
||
_lastSuccessUtc = DateTime.UtcNow;
|
||
}
|
||
}
|
||
|
||
private void RecordFailure(Exception ex)
|
||
{
|
||
// A cancelled read (deadline / client abort) is not a backend fault.
|
||
if (ex is OperationCanceledException) return;
|
||
lock (_healthLock)
|
||
{
|
||
_totalQueries++;
|
||
_totalFailures++;
|
||
_consecutiveFailures++;
|
||
_lastFailureUtc = DateTime.UtcNow;
|
||
_lastError = $"{ex.GetType().Name}: {ex.Message}";
|
||
}
|
||
_log.LogDebug("SQL historian read failed ({ExceptionType})", ex.GetType().Name);
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
public void Dispose()
|
||
{
|
||
// Connections are per call (pooled by SqlClient); nothing to release.
|
||
}
|
||
}
|