feat(historian): SQL Server read side — raw, at-time, events, push-down processed (ph-T10)
SqlServerHistorianDataSource implements IHistorianDataSource over the T8 schema: raw forward/reverse/point with the 0 = unbounded sentinel; AtTime in one statement (TimeRow TVP + two OUTER APPLYs) with linear/stepped/no-prior rules and no interpolation across a Bad neighbour or Empty outage marker; events with a maxEvents+1 truncation probe and a non-null continuation point; processed aggregates pushed down to T-SQL (SqlAggregateQueries) with nanosecond-exact bucket edges, Calculated bits on every result, Uncertain on any non-Good input, BadAggregateInvalidInputs for numeric aggregates over String/DateTime, a 100 000-bucket cap and HistoryAggregateNotSupportedException for the rest. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,127 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
|
||||
|
||||
/// <summary>The SQL shape that serves an aggregate. One query per family.</summary>
|
||||
internal enum SqlAggregateFamily
|
||||
{
|
||||
/// <summary>Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad: one GROUP BY.</summary>
|
||||
Simple,
|
||||
|
||||
/// <summary>Start / End / Delta: first + last usable sample per bucket (<c>ROW_NUMBER</c>).</summary>
|
||||
StartEnd,
|
||||
|
||||
/// <summary>MinimumActualTime / MaximumActualTime: extreme usable sample per bucket, ties ⇒ earliest.</summary>
|
||||
ActualTime,
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Push-down T-SQL for processed (aggregate) reads (spec §4 "Processed"). Every query takes
|
||||
/// <c>@id</c> (int TagId), <c>@start</c>/<c>@end</c> (<c>datetime2(7)</c>, the forward window
|
||||
/// <c>[start, end)</c>) and <c>@intervalNs</c> (bigint bucket width in nanoseconds), and returns one row
|
||||
/// per <b>non-empty</b> bucket. The caller generates every interval and fills the empty ones.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Bucketing.</b> <c>b = DATEDIFF_BIG(nanosecond, @start, SourceTimeUtc) / @intervalNs</c>. The spec
|
||||
/// sketched <c>millisecond</c>, but <c>DATEDIFF</c> counts unit <i>boundaries crossed</i>, not elapsed
|
||||
/// time, so with a non-millisecond-aligned <c>@start</c> a sample one tick before a bucket edge would
|
||||
/// land in the next bucket. <c>datetime2(7)</c> values are whole multiples of 100 ns, so the nanosecond
|
||||
/// boundary count IS the exact elapsed time and the integer division gives start-inclusive /
|
||||
/// end-exclusive buckets to the tick. (Range limit: ≈292 years per window.)
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Sample classes</b> (status stored bit-for-bit in an <c>int</c>, so the severity bits are the sign
|
||||
/// and bit 30): <i>usable</i> = severity Good or Uncertain (<c>StatusCode >= 0</c>) AND a stored value
|
||||
/// (<c>ValueKind <> 0</c>) — an <c>Empty</c> sample is an outage marker and counts as Bad whatever its
|
||||
/// code; <i>good</i> = usable with severity Good (<c>StatusCode < 0x40000000</c>); <i>bad</i> = not
|
||||
/// usable. Every value column below is computed over usable samples only.
|
||||
/// </para>
|
||||
/// Common result columns (ordinals 0–5): <c>Bucket, TotalCount, UsableCount, GoodCount, BadCount,
|
||||
/// NonNumericCount</c> (usable String/DateTime samples). Family columns follow.
|
||||
/// </remarks>
|
||||
internal static class SqlAggregateQueries
|
||||
{
|
||||
/// <summary>Ordinal of the first family-specific column.</summary>
|
||||
public const int FirstFamilyColumn = 6;
|
||||
|
||||
/// <summary><c>ValueKind</c> of String (4) and DateTime (5): usable but not numeric.</summary>
|
||||
private const string NonNumericKinds = "4, 5";
|
||||
|
||||
/// <summary>The family that serves <paramref name="aggregate"/>, or null when this backend does not.</summary>
|
||||
/// <param name="aggregate">Requested aggregate.</param>
|
||||
/// <returns>The family, or null for an unsupported aggregate.</returns>
|
||||
public static SqlAggregateFamily? FamilyOf(HistoryAggregateType aggregate) => aggregate switch
|
||||
{
|
||||
HistoryAggregateType.Average or HistoryAggregateType.Minimum or HistoryAggregateType.Maximum
|
||||
or HistoryAggregateType.Total or HistoryAggregateType.Count or HistoryAggregateType.Range
|
||||
or HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad => SqlAggregateFamily.Simple,
|
||||
HistoryAggregateType.Start or HistoryAggregateType.End or HistoryAggregateType.Delta => SqlAggregateFamily.StartEnd,
|
||||
HistoryAggregateType.MinimumActualTime or HistoryAggregateType.MaximumActualTime => SqlAggregateFamily.ActualTime,
|
||||
_ => null,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad.
|
||||
/// Family columns: <c>6 AvgValue, 7 MinValue, 8 MaxValue, 9 SumValue</c> (float, NULL when no usable
|
||||
/// numeric sample).
|
||||
/// </summary>
|
||||
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
||||
/// <returns>The SQL text.</returns>
|
||||
public static string Simple(string sample) =>
|
||||
Classified(sample) +
|
||||
"SELECT b, " + CommonCounts() + ", " +
|
||||
"AVG(CASE WHEN u = 1 THEN NumValue END) AS AvgValue, " +
|
||||
"MIN(CASE WHEN u = 1 THEN NumValue END) AS MinValue, " +
|
||||
"MAX(CASE WHEN u = 1 THEN NumValue END) AS MaxValue, " +
|
||||
"SUM(CASE WHEN u = 1 THEN NumValue END) AS SumValue " +
|
||||
"FROM s GROUP BY b ORDER BY b;";
|
||||
|
||||
/// <summary>
|
||||
/// Start / End / Delta: the first and last usable sample of each bucket by <c>SourceTimeUtc</c>.
|
||||
/// Family columns: <c>6..10</c> first sample (<c>ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc</c>),
|
||||
/// <c>11..15</c> last sample (same order); all NULL when the bucket has no usable sample.
|
||||
/// </summary>
|
||||
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
||||
/// <returns>The SQL text.</returns>
|
||||
public static string StartEnd(string sample) =>
|
||||
Classified(sample) +
|
||||
", st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b)" +
|
||||
", f AS (SELECT b, ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc, " +
|
||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rf, " +
|
||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rl FROM s WHERE u = 1) " +
|
||||
"SELECT st.*, " +
|
||||
"x.ValueKind, x.NumValue, x.IntValue, x.StrValue, x.SourceTimeUtc, " +
|
||||
"y.ValueKind, y.NumValue, y.IntValue, y.StrValue, y.SourceTimeUtc " +
|
||||
"FROM st LEFT JOIN f x ON x.b = st.b AND x.rf = 1 LEFT JOIN f y ON y.b = st.b AND y.rl = 1 ORDER BY st.b;";
|
||||
|
||||
/// <summary>
|
||||
/// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) <c>NumValue</c>,
|
||||
/// ties broken by the earliest <c>SourceTimeUtc</c>. Family columns: <c>6 NumValue, 7 SourceTimeUtc</c>
|
||||
/// (NULL when the bucket has no usable numeric sample).
|
||||
/// </summary>
|
||||
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
||||
/// <param name="maximum">True for MaximumActualTime.</param>
|
||||
/// <returns>The SQL text.</returns>
|
||||
public static string ActualTime(string sample, bool maximum) =>
|
||||
Classified(sample) +
|
||||
", st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b)" +
|
||||
", r AS (SELECT b, NumValue, SourceTimeUtc, " +
|
||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY NumValue " + (maximum ? "DESC" : "ASC") + ", SourceTimeUtc ASC) AS rn " +
|
||||
"FROM s WHERE u = 1 AND NumValue IS NOT NULL) " +
|
||||
"SELECT st.*, r.NumValue, r.SourceTimeUtc " +
|
||||
"FROM st LEFT JOIN r ON r.b = st.b AND r.rn = 1 ORDER BY st.b;";
|
||||
|
||||
/// <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) =>
|
||||
"WITH s AS (SELECT DATEDIFF_BIG(nanosecond, @start, SourceTimeUtc) / @intervalNs AS b, " +
|
||||
"SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, " +
|
||||
"CASE WHEN StatusCode >= 0 AND ValueKind <> 0 THEN 1 ELSE 0 END AS u, " +
|
||||
"CASE WHEN StatusCode >= 0 AND StatusCode < 1073741824 AND ValueKind <> 0 THEN 1 ELSE 0 END AS g " +
|
||||
$"FROM {sample} WHERE TagId = @id AND SourceTimeUtc >= @start AND SourceTimeUtc < @end) ";
|
||||
|
||||
/// <summary>The five common count columns (after <c>b</c>).</summary>
|
||||
private static string CommonCounts() =>
|
||||
"COUNT(*) AS TotalCount, SUM(u) AS UsableCount, SUM(g) AS GoodCount, COUNT(*) - SUM(u) AS BadCount, " +
|
||||
$"SUM(CASE WHEN u = 1 AND ValueKind IN ({NonNumericKinds}) THEN 1 ELSE 0 END) AS NonNumericCount";
|
||||
}
|
||||
+556
@@ -0,0 +1,556 @@
|
||||
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 (see <see cref="ReadAtTimeAsync"/>).
|
||||
/// </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 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>
|
||||
internal SqlServerHistorianDataSource(
|
||||
SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = 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;
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------------ 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> or <c>n</c> Bad (incl. an <c>Empty</c> outage marker) ⇒ <c>UncertainDataSubNormal</c> +
|
||||
/// Interpolated bits, and NO interpolation across it: the value is <c>p</c>'s held value when <c>p</c> is
|
||||
/// usable (data held up to the outage), else null (inside an outage).</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.</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 = DateTime.UtcNow;
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
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);
|
||||
|
||||
if (!p.IsUsable || next is { IsUsable: false })
|
||||
return new DataValueSnapshot(p.IsUsable ? p.Value.ToObject() : null,
|
||||
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);
|
||||
}
|
||||
|
||||
// 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)
|
||||
{
|
||||
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, 400) { 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;
|
||||
var intervalTicks = interval > TimeSpan.Zero ? interval.Ticks : windowTicks;
|
||||
long bucketCount = windowTicks == 0 ? 0 : (windowTicks + intervalTicks - 1) / intervalTicks;
|
||||
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 = DateTime.UtcNow;
|
||||
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("@intervalNs", SqlDbType.BigInt) { Value = intervalTicks * 100L });
|
||||
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 => (Status & 0x80000000u) == 0 && Value.Kind != HistorianValueKind.Empty;
|
||||
|
||||
/// <summary>Usable with severity Good.</summary>
|
||||
public bool IsGood => IsUsable && (Status & 0xC0000000u) == 0;
|
||||
}
|
||||
|
||||
/// <summary>Reads <c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode</c> starting at <paramref name="o"/>.</summary>
|
||||
private static StoredSample ReadStored(SqlDataReader r, int o) => new(
|
||||
SqlHistorianConnection.GetUtc(r, o),
|
||||
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)),
|
||||
SampleRowMapper.FromDbStatus(r.GetInt32(o + 5)));
|
||||
|
||||
/// <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.
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user