Joseph Doherty
610 lines
36 KiB
C#
610 lines
36 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;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
|
|
|
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 their duration as Bad time 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="HistorianTagNames.MaxLength"/>
|
|
/// 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 = BucketPlan.MaxBuckets;
|
|
|
|
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 HistorianHealthCounters _health = new();
|
|
private readonly int _maxTimeWeightedRawRows;
|
|
|
|
/// <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>
|
|
/// <param name="maxTimeWeightedRawRows">Raw in-window row cap of one time-weighted read (<= 0 = unbounded).</param>
|
|
internal SqlServerHistorianDataSource(
|
|
SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null,
|
|
TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
|
|
{
|
|
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
|
|
_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;
|
|
}
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------------------ Modified (Part 11, #668)
|
|
|
|
/// <inheritdoc />
|
|
/// <remarks>
|
|
/// Reads <c>SampleModified</c> (schema v2), the trail <see cref="SqlServerHistorianDataUpdater"/> writes in the
|
|
/// same statement as each change. Windows follow <see cref="ReadRawAsync"/>; rows are ordered by
|
|
/// <c>(SourceTimeUtc, ModifiedId)</c> (the clustered key after <c>TagId</c>, so a seek). <c>maxValues + 1</c> rows
|
|
/// are fetched to report <see cref="HistoryModifiedReadResult.HasMore"/> exactly. An insert's audit row carries no
|
|
/// prior value and is served <c>null</c> / <c>BadNoData</c>; its server timestamp is the modification time. An
|
|
/// unknown tag is an empty result.
|
|
/// </remarks>
|
|
public async Task<HistoryModifiedReadResult> ReadModifiedAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(fullReference);
|
|
var bounded = maxValues > 0 && maxValues < int.MaxValue;
|
|
try
|
|
{
|
|
// #674: a v1 schema has no SampleModified. NotSupportedException is the interface's "no trail" signal (the
|
|
// node manager reports BadHistoryOperationUnsupported); it is not a backend failure and sends no query.
|
|
await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
|
|
if (!_bootstrap.SupportsModifiedTrail)
|
|
throw new NotSupportedException(
|
|
$"Historian schema [{_conn.Schema}] is below version {SqlServerSchemaBootstrapper.ModifiedTrailVersion}: no modified-value history.");
|
|
var values = new List<HistoryModifiedValue>();
|
|
var hasMore = false;
|
|
var tagId = await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
|
|
if (tagId is { } id)
|
|
{
|
|
var reverse = startUtc > endUtc;
|
|
string where;
|
|
if (startUtc == endUtc) where = "SourceTimeUtc = @lo";
|
|
else if (!reverse) where = "SourceTimeUtc >= @lo AND SourceTimeUtc < @hi";
|
|
else where = "SourceTimeUtc > @lo AND SourceTimeUtc <= @hi";
|
|
if (resumeAfter is not null)
|
|
where += reverse
|
|
? " AND (SourceTimeUtc < @ct OR (SourceTimeUtc = @ct AND ModifiedId < @cid))"
|
|
: " AND (SourceTimeUtc > @ct OR (SourceTimeUtc = @ct AND ModifiedId > @cid))";
|
|
var order = reverse ? "DESC" : "ASC";
|
|
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 {(bounded ? "TOP (@take) " : "")}SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, " +
|
|
"ServerTimeUtc, ModifiedId, ModificationTimeUtc, UpdateType, Principal " +
|
|
$"FROM {_conn.Q("SampleModified")} WHERE TagId = @id AND {where} " +
|
|
$"ORDER BY SourceTimeUtc {order}, ModifiedId {order}");
|
|
if (bounded) cmd.Parameters.Add(new SqlParameter("@take", SqlDbType.BigInt) { Value = (long)maxValues + 1 });
|
|
cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
|
|
SqlHistorianConnection.AddDateTime2(cmd, "@lo", lo);
|
|
SqlHistorianConnection.AddDateTime2(cmd, "@hi", hi);
|
|
if (resumeAfter is not null)
|
|
{
|
|
SqlHistorianConnection.AddDateTime2(cmd, "@ct", resumeAfter.SourceTimeUtc);
|
|
cmd.Parameters.Add(new SqlParameter("@cid", SqlDbType.BigInt) { Value = resumeAfter.ModifiedId });
|
|
}
|
|
|
|
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
|
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
|
{
|
|
if (bounded && values.Count == maxValues)
|
|
{
|
|
hasMore = true;
|
|
break;
|
|
}
|
|
|
|
var sourceTime = SqlHistorianConnection.GetUtc(reader, 0);
|
|
var modifiedAt = SqlHistorianConnection.GetUtc(reader, 8);
|
|
DataValueSnapshot prior;
|
|
if (reader.IsDBNull(1))
|
|
{
|
|
// An insert: there was no prior value.
|
|
prior = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, sourceTime, modifiedAt);
|
|
}
|
|
else
|
|
{
|
|
var s = ReadStored(reader, 0);
|
|
prior = new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc,
|
|
reader.IsDBNull(6) ? modifiedAt : SqlHistorianConnection.GetUtc(reader, 6));
|
|
}
|
|
|
|
values.Add(new HistoryModifiedValue(
|
|
prior, modifiedAt, (HistoryModificationType)reader.GetByte(9), reader.GetString(10), reader.GetInt64(7), sourceTime));
|
|
}
|
|
}
|
|
|
|
RecordSuccess();
|
|
return new HistoryModifiedReadResult(values, hasMore);
|
|
}
|
|
catch (Exception ex) when (ex is not NotSupportedException)
|
|
{
|
|
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] = AtTimeEvaluator.Evaluate(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, HistorianStatusCodes.BadNoData, t, now);
|
|
}
|
|
|
|
RecordSuccess();
|
|
return new HistoryReadResult(result, ContinuationPoint: null);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
RecordFailure(ex);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------------------ Events
|
|
|
|
/// <inheritdoc />
|
|
/// <remarks>Reverse events reads are served natively (<c>ORDER BY … DESC</c>, Gitea #675).</remarks>
|
|
public bool SupportsReverseEvents => true;
|
|
|
|
|
|
/// <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 Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
|
|
ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken);
|
|
|
|
/// <inheritdoc />
|
|
/// <remarks>
|
|
/// Forward is the five-argument read. Reverse (Gitea #675) is <c>(lo, hi]</c> ordered by
|
|
/// <c>(EventTimeUtc, EventId)</c> DESCENDING — the clustered index scanned backwards — so <c>TOP</c> keeps the latest
|
|
/// events; truncation and the continuation point work as forward.
|
|
/// </remarks>
|
|
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, 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 > HistorianTagNames.MaxLength)
|
|
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 " +
|
|
(reverse ? "EventTimeUtc > @lo AND EventTimeUtc <= @hi" : "EventTimeUtc >= @lo AND EventTimeUtc < @hi") +
|
|
(sourceName is null ? "" : " AND SourceName = @source") +
|
|
(reverse ? " ORDER BY EventTimeUtc DESC, EventId DESC" : " 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),
|
|
HistoricalEventRows.ClampSeverity(reader.GetInt32(5)));
|
|
if (bounded && events.Count == maxEvents)
|
|
{
|
|
continuation = Encoding.UTF8.GetBytes(EventContinuation.Encode(e.EventTimeUtc, 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, #677) ⇒ Part 13 / SDK anchoring: window <c>(end, start]</c>,
|
|
/// buckets stepping back from <c>start</c> (the partial one is the earliest), each stamped with its later edge,
|
|
/// returned latest first (<see cref="BucketPlan"/>).
|
|
/// </para>
|
|
/// <para>
|
|
/// Per-sample aggregates, per interval: no samples ⇒ <c>BadNoData</c>. 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 calculated status — including <c>BadNoData</c> — carries the Calculated info bits. Timestamps are the
|
|
/// bucket start (its later edge on a reverse read), except Minimum/MaximumActualTime (the sample's time; ties ⇒
|
|
/// earliest) and Start/End. Every aggregate but Average sets the Partial bit on the bucket the tag's earliest
|
|
/// sample lies strictly inside (forward, #677) or its latest sample lies in (#680; reverse: strictly inside).
|
|
/// PercentGood / PercentBad are duration-based since #680 (time-weighted family, below). #680: Count counts Good samples (value-based status; 0/Good on an empty bucket inside the tag's
|
|
/// data range), Start/End are the raw first/last sample of any quality, Delta is Uncertain only when the bucket's
|
|
/// first or last sample is not usable (see <see cref="AggregateEvaluator"/>).
|
|
/// </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. The Part 13 time-weighted family (TimeAverage, TimeAverage2, Total, Total2, Interpolative, #671) is
|
|
/// not pushed down: the window's rows plus four bounding rows are streamed into the shared
|
|
/// <see cref="TimeWeightedAggregator"/>, which fills every interval (an empty one between bounds gets a value).
|
|
/// </para>
|
|
/// </remarks>
|
|
/// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not one of the seventeen 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 = AggregateFamilies.FamilyOf(aggregate) ?? throw new HistoryAggregateNotSupportedException(aggregate);
|
|
var plan = BucketPlan.Create(startUtc, endUtc, interval);
|
|
|
|
try
|
|
{
|
|
var now = _time.GetUtcNow().UtcDateTime;
|
|
var tagId = plan.BucketCount == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
|
|
if (family == AggregateFamily.TimeWeighted)
|
|
{
|
|
var tw = new TimeWeightedAggregator(aggregate, plan);
|
|
if (tagId is { } twId) await StreamTimeWeightedAsync(twId, fullReference, plan, tw, cancellationToken).ConfigureAwait(false);
|
|
var twResult = tw.Finish(now);
|
|
RecordSuccess();
|
|
return new HistoryReadResult(twResult, ContinuationPoint: null);
|
|
}
|
|
var rows = new List<BucketRow>();
|
|
DateTime? earliest = null, latest = null;
|
|
if (tagId is { } id)
|
|
{
|
|
var sample = _conn.Q("Sample");
|
|
var sql = family switch
|
|
{
|
|
AggregateFamily.Simple => SqlAggregateQueries.Simple(sample, plan.Reversed),
|
|
AggregateFamily.StartEnd => SqlAggregateQueries.StartEnd(sample, plan.Reversed),
|
|
_ => SqlAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed),
|
|
};
|
|
// Second result set: the tag's data range (earliest + latest sample) — the start/end-of-data Partial rules
|
|
// (#677/#680 — any per-sample aggregate but Average, either direction) and Count's in-range empty interval (#680).
|
|
var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan);
|
|
if (needsRange) sql += " " + SqlAggregateQueries.DataRange(sample);
|
|
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(plan.StartUtc, DateTimeKind.Utc));
|
|
SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(plan.EndUtc, DateTimeKind.Utc));
|
|
cmd.Parameters.Add(new SqlParameter("@intervalTicks", SqlDbType.BigInt) { Value = plan.IntervalTicks });
|
|
cmd.Parameters.Add(new SqlParameter("@bucketAdjustTicks", SqlDbType.BigInt) { Value = plan.BucketIndexAdjustTicks });
|
|
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
|
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
|
rows.Add(ReadBucket(reader, family));
|
|
if (needsRange && await reader.NextResultAsync(cancellationToken).ConfigureAwait(false)
|
|
&& await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
|
{
|
|
if (!reader.IsDBNull(0)) earliest = SqlHistorianConnection.GetUtc(reader, 0);
|
|
if (!reader.IsDBNull(1)) latest = SqlHistorianConnection.GetUtc(reader, 1);
|
|
}
|
|
}
|
|
|
|
var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, latest, now);
|
|
|
|
RecordSuccess();
|
|
return new HistoryReadResult(result, ContinuationPoint: null);
|
|
}
|
|
catch (HistoryReadLimitExceededException)
|
|
{
|
|
// A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
|
|
throw;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
RecordFailure(ex);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Streams the time-weighted batch (<see cref="SqlAggregateQueries.TimeWeighted"/>) into <paramref name="tw"/>: the
|
|
/// two before-bounds, the window, the two after-bounds — ascending, with duplicates dropped by the aggregator.
|
|
/// </summary>
|
|
private async Task StreamTimeWeightedAsync(int id, string tag, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct)
|
|
{
|
|
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
|
await using var cmd = _conn.CreateCommand(conn, SqlAggregateQueries.TimeWeighted(_conn.Q("Sample")));
|
|
cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
|
|
SqlHistorianConnection.AddDateTime2(cmd, "@start", DateTime.SpecifyKind(plan.StartUtc, DateTimeKind.Utc));
|
|
SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(plan.EndUtc, DateTimeKind.Utc));
|
|
cmd.Parameters.Add(new SqlParameter("@maxRows", SqlDbType.Int)
|
|
{ Value = TimeWeightedAggregator.WindowFetchLimit(_maxTimeWeightedRawRows) });
|
|
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
|
var resultSet = 0;
|
|
do
|
|
{
|
|
long windowRows = 0;
|
|
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
|
{
|
|
// Result set 2 is the window (TOP (@maxRows) = cap + 1): one row past the cap fails the node.
|
|
if (resultSet == 2) TimeWeightedAggregator.CheckWindowRows(++windowRows, _maxTimeWeightedRawRows, tag);
|
|
tw.Add(ReadStored(reader, 0));
|
|
}
|
|
resultSet++;
|
|
} while (await reader.NextResultAsync(ct).ConfigureAwait(false));
|
|
}
|
|
|
|
private static BucketRow ReadBucket(SqlDataReader reader, AggregateFamily 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 : SampleColumnMapper.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 AggregateFamily.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 AggregateFamily.StartEnd:
|
|
row.First = V(reader, f);
|
|
row.Last = V(reader, f + 5);
|
|
row.FirstAny = ReadStored(reader, SqlAggregateQueries.FirstAnyColumn);
|
|
row.LastAny = ReadStored(reader, SqlAggregateQueries.FirstAnyColumn + 6);
|
|
break;
|
|
default:
|
|
row.Extreme = D(reader, f);
|
|
row.ExtremeTime = reader.IsDBNull(f + 1) ? null : SqlHistorianConnection.GetUtc(reader, f + 1);
|
|
break;
|
|
}
|
|
return row;
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------------------ Shared
|
|
|
|
/// <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>
|
|
/// (<see cref="StoredSample.FromStored"/>) — 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 = SampleColumnMapper.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 = SampleColumnMapper.FromDbStatus(r.GetInt32(o + 5));
|
|
return StoredSample.FromStored(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() => _health.Snapshot();
|
|
|
|
private void RecordSuccess() => _health.RecordSuccess();
|
|
|
|
private void RecordFailure(Exception ex)
|
|
{
|
|
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
|
|
if (ex is OperationCanceledException) return;
|
|
_health.RecordFailure(ex);
|
|
_log.LogDebug("SQL historian read failed ({ExceptionType})", ex.GetType().Name);
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public void Dispose()
|
|
{
|
|
// Connections are per call (pooled by SqlClient); nothing to release.
|
|
}
|
|
}
|