From 38d37f5c230aaa0cd05acd0d9c428ba182c8a605 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 13:13:07 -0400 Subject: [PATCH] =?UTF-8?q?feat(historian):=20SQL=20Server=20read=20side?= =?UTF-8?q?=20=E2=80=94=20raw,=20at-time,=20events,=20push-down=20processe?= =?UTF-8?q?d=20(ph-T10)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../SqlAggregateQueries.cs | 127 ++++ .../SqlServerHistorianDataSource.cs | 556 ++++++++++++++++++ .../AtTimeReadTests.cs | 184 ++++++ .../EventReadTests.cs | 92 +++ .../ProcessedReadTests.cs | 337 +++++++++++ .../RawReadTests.cs | 257 ++++++++ 6 files changed, 1553 insertions(+) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AtTimeReadTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/EventReadTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProcessedReadTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RawReadTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs new file mode 100644 index 00000000..a099a168 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs @@ -0,0 +1,127 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// The SQL shape that serves an aggregate. One query per family. +internal enum SqlAggregateFamily +{ + /// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad: one GROUP BY. + Simple, + + /// Start / End / Delta: first + last usable sample per bucket (ROW_NUMBER). + StartEnd, + + /// MinimumActualTime / MaximumActualTime: extreme usable sample per bucket, ties ⇒ earliest. + ActualTime, +} + +/// +/// Push-down T-SQL for processed (aggregate) reads (spec §4 "Processed"). Every query takes +/// @id (int TagId), @start/@end (datetime2(7), the forward window +/// [start, end)) and @intervalNs (bigint bucket width in nanoseconds), and returns one row +/// per non-empty bucket. The caller generates every interval and fills the empty ones. +/// +/// +/// +/// Bucketing. b = DATEDIFF_BIG(nanosecond, @start, SourceTimeUtc) / @intervalNs. The spec +/// sketched millisecond, but DATEDIFF counts unit boundaries crossed, not elapsed +/// time, so with a non-millisecond-aligned @start a sample one tick before a bucket edge would +/// land in the next bucket. datetime2(7) 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.) +/// +/// +/// Sample classes (status stored bit-for-bit in an int, so the severity bits are the sign +/// and bit 30): usable = severity Good or Uncertain (StatusCode >= 0) AND a stored value +/// (ValueKind <> 0) — an Empty sample is an outage marker and counts as Bad whatever its +/// code; good = usable with severity Good (StatusCode < 0x40000000); bad = not +/// usable. Every value column below is computed over usable samples only. +/// +/// Common result columns (ordinals 0–5): Bucket, TotalCount, UsableCount, GoodCount, BadCount, +/// NonNumericCount (usable String/DateTime samples). Family columns follow. +/// +internal static class SqlAggregateQueries +{ + /// Ordinal of the first family-specific column. + public const int FirstFamilyColumn = 6; + + /// ValueKind of String (4) and DateTime (5): usable but not numeric. + private const string NonNumericKinds = "4, 5"; + + /// The family that serves , or null when this backend does not. + /// Requested aggregate. + /// The family, or null for an unsupported aggregate. + 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, + }; + + /// + /// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad. + /// Family columns: 6 AvgValue, 7 MinValue, 8 MaxValue, 9 SumValue (float, NULL when no usable + /// numeric sample). + /// + /// Schema-qualified Sample table name. + /// The SQL text. + 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;"; + + /// + /// Start / End / Delta: the first and last usable sample of each bucket by SourceTimeUtc. + /// Family columns: 6..10 first sample (ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc), + /// 11..15 last sample (same order); all NULL when the bucket has no usable sample. + /// + /// Schema-qualified Sample table name. + /// The SQL text. + 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;"; + + /// + /// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) NumValue, + /// ties broken by the earliest SourceTimeUtc. Family columns: 6 NumValue, 7 SourceTimeUtc + /// (NULL when the bucket has no usable numeric sample). + /// + /// Schema-qualified Sample table name. + /// True for MaximumActualTime. + /// The SQL text. + 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;"; + + /// The s CTE: the window's samples with bucket index and the usable/good flags. + 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) "; + + /// The five common count columns (after b). + 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"; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs new file mode 100644 index 00000000..5ef1d4e7 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -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; + +/// +/// SQL Server read side of the historian (spec §4 "Reads"): Raw (forward / reverse / point), AtTime +/// (one statement, TimeRow TVP + two OUTER APPLYs), Events, and Processed aggregates pushed +/// down to T-SQL (). +/// +/// +/// +/// Reads never create tags. Tag names resolve through the shared 's +/// read-only lookup (positive results cached); an unknown tag is a successful empty read (Raw), an +/// all-BadNoData read (AtTime / Processed), never an error. +/// +/// +/// Faults. A SQL / bootstrap exception is recorded in (so a missing +/// schema with AutoCreateSchema=false surfaces in LastError) and rethrown; the node manager +/// turns it into a per-node Bad. Client-request errors — an unsupported aggregate +/// ( ⇒ BadAggregateNotSupported) and a processed +/// window over intervals (, which +/// the node manager's generic catch surfaces as BadHistoryOperationUnsupported for that node only) +/// — are thrown before any I/O and are not counted as backend failures. +/// +/// +/// Bad samples. A stored sample is usable when its status severity is Good or Uncertain and it +/// carries a value; an Empty-kind sample is an outage marker (comms-loss BadNoCommunication +/// 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 ). +/// +/// +public sealed class SqlServerHistorianDataSource : IHistorianDataSource +{ + /// Spec §9.7: a processed read covering more intervals than this is refused. + 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; + + /// Creates the data source over the provider's shared connection, bootstrapper and tag catalog. + /// Connection helper. + /// Schema bootstrapper (shared with the other SQL seams). + /// Tag catalog (shared; reads use only its read-only lookup). + /// Optional logger. + 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 + + /// + /// + /// Forward (start < end): [start, end) ascending. Reverse (end < start, Part 11): + /// (end, start] descending. start == end: 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). + /// 0 is the node manager's "unbounded" sentinel; values above clamp. + /// The node manager pages by re-reading from the last returned timestamp inclusive and trimming it; the + /// (TagId, SourceTimeUtc) primary key rules out tie clusters, so that is exact. + /// + public async Task 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(); + 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 + + /// + /// + /// One statement: the distinct requested times go in a TimeRow TVP and each picks up its + /// previous-or-equal (p) and next (n) stored sample via two OUTER APPLY (SELECT TOP 1 …). + /// Rules, per requested time T (the result is stamped T): + /// + /// p at exactly T ⇒ the stored value and status, unchanged. + /// No p ⇒ BadNoData, null value. + /// p or n Bad (incl. an Empty outage marker) ⇒ UncertainDataSubNormal + + /// Interpolated bits, and NO interpolation across it: the value is p's held value when p is + /// usable (data held up to the outage), else null (inside an outage). + /// Double/Int64 with a usable n ⇒ linear interpolation of NumValue (Int64 rounded to the + /// nearest integer, ties away from zero, keeping the type); status Good + Interpolated when both + /// neighbours are Good, else UncertainDataSubNormal + Interpolated (never better than the + /// inputs). + /// Boolean/String/DateTime, or no n (after the last sample) ⇒ stepped: p's value with + /// p's status code + Interpolated bits. + /// + /// Unknown tag ⇒ every entry BadNoData. + /// + public async Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(fullReference); + ArgumentNullException.ThrowIfNull(timestampsUtc); + try + { + var now = DateTime.UtcNow; + var byTicks = new Dictionary(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; + } + } + + /// Applies the AtTime rules (see ) to one requested time. + 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 + + /// + /// + /// [start, end) by EventTimeUtc (a reversed window is swapped), ordered by + /// (EventTimeUtc, EventId) ascending — the clustered index. <= 0 or + /// (the node manager's "no limit") ⇒ unbounded. Otherwise maxEvents + 1 + /// rows are fetched; when the extra row arrives it is dropped and the continuation point is the UTF-8 + /// of "{EventTimeUtc.Ticks}|{EventId}" of that first not-returned event (opaque to callers; only + /// its presence is contractual). + /// + public async Task 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(); + 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 + + /// + /// + /// + /// One result per interval of [start, end), bucket i = [start + i·interval, + /// start + (i+1)·interval) (start-inclusive, end-exclusive to the tick; the last bucket is truncated at + /// end). <= 0 ⇒ one interval [start, end). start == end + /// ⇒ no intervals. end < start (reverse) ⇒ the window is swapped, computed forward, and returned + /// in descending bucket order (buckets stay anchored at the earlier time). + /// + /// + /// Per interval: no samples ⇒ BadNoData. PercentGood / PercentBad ⇒ 100 × good|bad / all over + /// every sample, status Good. Otherwise no usable sample ⇒ BadNoData; a numeric aggregate with a usable + /// String/DateTime sample ⇒ BadAggregateInvalidInputs; else the value, Good when every sample is Good, + /// UncertainDataSubNormal when any is Bad or Uncertain. Every status — including BadNoData — + /// carries the Calculated info bits. Timestamps are the bucket start, except Minimum/MaximumActualTime + /// (the sample's time; ties ⇒ earliest). + /// + /// + /// Values: Count is ; Start/End keep the sample's type; every other aggregate is + /// computed over NumValue (Boolean = 0/1). These are the simple per-sample + /// definitions, not Part 13's interval-boundary (interpolated bound / time-weighted) semantics. + /// + /// + /// The aggregate is not one of the thirteen served. + /// The window holds more than intervals. + public async Task 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(); + 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((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; + } + + /// One aggregate query row (one non-empty bucket). + 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 + + /// A stored sample read from the {SampleColumns} column run starting at an ordinal. + private readonly record struct StoredSample(DateTime TimeUtc, HistorianValue Value, uint Status) + { + /// Severity Good or Uncertain, and a value present (an Empty sample is an outage marker). + public bool IsUsable => (Status & 0x80000000u) == 0 && Value.Kind != HistorianValueKind.Empty; + + /// Usable with severity Good. + public bool IsGood => IsUsable && (Status & 0xC0000000u) == 0; + } + + /// Reads SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode starting at . + 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))); + + /// Read-only tag lookup (never creates a row); runs the schema bootstrap on a cache miss. + private Task ResolveAsync(string name, CancellationToken ct) => _catalog.TryGetTagIdAsync(name, ct); + + // ------------------------------------------------------------------------------------------ Health + + /// + 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); + } + + /// + public void Dispose() + { + // Connections are per call (pooled by SqlClient); nothing to release. + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AtTimeReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AtTimeReadTests.cs new file mode 100644 index 00000000..90c28571 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AtTimeReadTests.cs @@ -0,0 +1,184 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +/// AtTime rules (spec §4): exact, linear, stepped, no-prior, Bad neighbour (no bridging). +[Collection(SqlHistorianCollection.Name)] +public sealed class AtTimeReadTests(SqlHistorianFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static DateTime At(double s) => ReadTestDb.At(s); + + // Hand-computed status literals: Interpolated = InfoType DataValue 0x400 | Interpolated 0x2. + private const uint GoodInterpolated = 0x00000402u; + private const uint UncertainInterpolated = 0x40A40402u; // UncertainDataSubNormal | 0x402 + private const uint BadNoData = 0x809B0000u; + + [Fact] + public async Task Interpolated_bits_are_the_SDK_AggregateBits_Interpolated() + { + var code = new Opc.Ua.StatusCode(GoodInterpolated); + code.AggregateBits.ShouldBe(Opc.Ua.AggregateBits.Interpolated); + Opc.Ua.StatusCode.IsGood(code).ShouldBeTrue(); + new Opc.Ua.StatusCode(UncertainInterpolated).CodeBits.ShouldBe(Opc.Ua.StatusCodes.UncertainDataSubNormal); + await Task.CompletedTask; + } + + [Fact] + public async Task Exact_hit_returns_the_stored_value_and_status() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(10), At(20)], Ct)).Samples; + s[0].Value.ShouldBe(20d); + s[0].StatusCode.ShouldBe(0u); + s[0].SourceTimestampUtc.ShouldBe(At(10)); + s[0].ServerTimestampUtc.ShouldBe(At(11)); // the stored ServerTimeUtc + s[1].Value.ShouldBe(30d); // a stored Bad sample comes back as stored + s[1].StatusCode.ShouldBe(0x80000000u); + } + + [Fact] + public async Task Between_two_good_doubles_is_linearly_interpolated_good() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(5), At(2.5), At(32.5)], Ct)).Samples; + s[0].Value.ShouldBe(15d); // 10 + (20-10) * 5/10 + s[1].Value.ShouldBe(12.5d); // 10 + 10 * 2.5/10 + s[2].Value.ShouldBe(42.5d); // 40 + 10 * 2.5/10 + s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + s[0].SourceTimestampUtc.ShouldBe(At(5)); // stamped at the requested (sub-ms) time + } + + [Fact] + public async Task No_previous_sample_is_BadNoData() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(0).AddTicks(-1)], Ct)).Samples.Single(); + s.Value.ShouldBeNull(); + s.StatusCode.ShouldBe(BadNoData); + s.SourceTimestampUtc.ShouldBe(At(0).AddTicks(-1)); + } + + [Fact] + public async Task A_bad_next_neighbour_holds_the_previous_value_uncertain() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(15)], Ct)).Samples.Single(); + s.Value.ShouldBe(20d); // NOT 25 (interpolating toward the Bad 30) + s.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public async Task A_bad_previous_neighbour_yields_no_value_uncertain() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(25)], Ct)).Samples.Single(); + s.Value.ShouldBeNull(); // NOT 35 (30→40) and NOT 30 (a Bad value) + s.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public async Task Interpolation_never_bridges_an_outage_marker() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("M", + (At(0), HistorianValue.FromDouble(10), ReadTestDb.Good), + (At(5), HistorianValue.Empty, ReadTestDb.BadNoCommunication), + (At(10), HistorianValue.FromDouble(30), ReadTestDb.Good)); + var s = (await db.Source().ReadAtTimeAsync("M", [At(2.5), At(7.5), At(5)], Ct)).Samples; + s[0].Value.ShouldBe(10d); // bridging 10→30 would give 15 + s[0].StatusCode.ShouldBe(UncertainInterpolated); + s[1].Value.ShouldBeNull(); // bridging would give 25 + s[1].StatusCode.ShouldBe(UncertainInterpolated); + s[2].Value.ShouldBeNull(); // exact hit on the marker: as stored + s[2].StatusCode.ShouldBe(0x80310000u); + } + + [Fact] + public async Task After_the_last_sample_the_value_is_stepped() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(45)], Ct)).Samples.Single(); + s.Value.ShouldBe(50d); + s.StatusCode.ShouldBe(GoodInterpolated); + } + + [Fact] + public async Task An_uncertain_neighbour_makes_the_interpolation_uncertain() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("U", + (At(0), HistorianValue.FromDouble(10), ReadTestDb.Uncertain), + (At(10), HistorianValue.FromDouble(20), ReadTestDb.Good)); + var s = (await db.Source().ReadAtTimeAsync("U", [At(5)], Ct)).Samples.Single(); + s.Value.ShouldBe(15d); + s.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public async Task Int64_is_interpolated_and_rounded_keeping_its_type() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("I", + (At(0), HistorianValue.FromInt64(10), ReadTestDb.Good), + (At(10), HistorianValue.FromInt64(15), ReadTestDb.Good)); + var s = (await db.Source().ReadAtTimeAsync("I", [At(5), At(2)], Ct)).Samples; + s[0].Value.ShouldBe(13L); // 12.5 ⇒ 13 (away from zero) + s[1].Value.ShouldBe(11L); // 11.0 + s[0].StatusCode.ShouldBe(GoodInterpolated); + } + + [Fact] + public async Task Boolean_string_and_datetime_are_stepped_from_the_previous_sample() + { + var db = await ReadTestDb.CreateAsync(fx); + var d0 = new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddTicks(3); + await db.SeedAsync("B", + (At(0), HistorianValue.FromBoolean(true), ReadTestDb.Good), + (At(10), HistorianValue.FromBoolean(false), ReadTestDb.Good)); + await db.SeedAsync("S", + (At(0), HistorianValue.FromString("alpha"), ReadTestDb.Uncertain), + (At(10), HistorianValue.FromString("beta"), ReadTestDb.Good)); + await db.SeedAsync("D", + (At(0), HistorianValue.FromDateTime(d0), ReadTestDb.Good), + (At(10), HistorianValue.FromDateTime(d0.AddDays(1)), ReadTestDb.Good)); + var src = db.Source(); + var b = (await src.ReadAtTimeAsync("B", [At(9)], Ct)).Samples.Single(); + b.Value.ShouldBe(true); + b.StatusCode.ShouldBe(GoodInterpolated); + var s = (await src.ReadAtTimeAsync("S", [At(5)], Ct)).Samples.Single(); + s.Value.ShouldBe("alpha"); + s.StatusCode.ShouldBe(0x40000402u); // the previous sample's code (Uncertain) + Interpolated + (await src.ReadAtTimeAsync("D", [At(5)], Ct)).Samples.Single().Value.ShouldBe(d0); + } + + [Fact] + public async Task Result_matches_request_length_and_order_including_duplicates() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadAtTimeAsync("A", [At(5), At(-1), At(5), At(10)], Ct)).Samples; + s.Count.ShouldBe(4); + s.Select(x => x.Value).ShouldBe([15d, null, 15d, 20d]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(5), At(-1), At(5), At(10)]); + } + + [Fact] + public async Task Empty_request_and_unknown_tag() + { + var db = await ReadTestDb.CreateAsync(fx); + var src = db.Source(); + (await src.ReadAtTimeAsync("A", [], Ct)).Samples.ShouldBeEmpty(); + var s = (await src.ReadAtTimeAsync("nope", [At(1), At(2)], Ct)).Samples; + s.Count.ShouldBe(2); + s.ShouldAllBe(x => x.StatusCode == BadNoData && x.Value == null); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/EventReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/EventReadTests.cs new file mode 100644 index 00000000..e12c480b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/EventReadTests.cs @@ -0,0 +1,92 @@ +using System.Text; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +[Collection(SqlHistorianCollection.Name)] +public sealed class EventReadTests(SqlHistorianFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static DateTime At(double s) => ReadTestDb.At(s); + + private static async Task InsertEventAsync(ReadTestDb db, string id, string source, DateTime t, string? message, int severity) + { + await using var c = await db.Conn.OpenAsync(Ct); + await using var cmd = db.Conn.CreateCommand(c, + $"INSERT {db.Conn.Q("Event")} (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, " + + "UserName, Comment, EventTimeUtc, ReceivedTimeUtc) VALUES (@id, @src, 'a', 'n', 'T', @sev, 'Activated', @msg, NULL, NULL, @t, @r)"); + cmd.Parameters.AddWithValue("@id", id); + cmd.Parameters.AddWithValue("@src", source); + cmd.Parameters.AddWithValue("@sev", severity); + cmd.Parameters.AddWithValue("@msg", (object?)message ?? DBNull.Value); + SqlHistorianConnection.AddDateTime2(cmd, "@t", t); + SqlHistorianConnection.AddDateTime2(cmd, "@r", t.AddTicks(3)); + await cmd.ExecuteNonQueryAsync(); + } + + private static async Task SeedAsync(SqlHistorianFixture fx) + { + var db = await ReadTestDb.CreateAsync(fx); + await InsertEventAsync(db, "e1", "Eq/1", At(0), "first", 500); + await InsertEventAsync(db, "e2", "Eq/2", At(1), null, 700); + await InsertEventAsync(db, "e3", "Eq/1", At(2), "third", 900); + await InsertEventAsync(db, "e4", "Eq/1", At(3), "at end", 100); + return db; + } + + [Fact] + public async Task Source_filter_window_order_and_field_mapping() + { + var db = await SeedAsync(fx); + var r = await db.Source().ReadEventsAsync("Eq/1", At(0), At(3), 0, Ct); + r.Events.Select(e => e.EventId).ShouldBe(["e1", "e3"]); // e4 at end excluded, e2 other source + var e = r.Events[0]; + e.SourceName.ShouldBe("Eq/1"); + e.EventTimeUtc.ShouldBe(At(0)); + e.EventTimeUtc.Kind.ShouldBe(DateTimeKind.Utc); + e.ReceivedTimeUtc.ShouldBe(At(0).AddTicks(3)); + e.Message.ShouldBe("first"); + e.Severity.ShouldBe((ushort)500); + r.ContinuationPoint.ShouldBeNull(); + } + + [Fact] + public async Task Null_source_returns_every_source_and_null_messages() + { + var db = await SeedAsync(fx); + var r = await db.Source().ReadEventsAsync(null, At(0), At(10), 0, Ct); + r.Events.Select(e => e.EventId).ShouldBe(["e1", "e2", "e3", "e4"]); + r.Events[1].Message.ShouldBeNull(); + } + + [Fact] + public async Task Truncation_drops_the_extra_row_and_returns_a_continuation_point() + { + var db = await SeedAsync(fx); + var r = await db.Source().ReadEventsAsync(null, At(0), At(3), 2, Ct); // 3 matching rows + r.Events.Select(e => e.EventId).ShouldBe(["e1", "e2"]); + r.ContinuationPoint.ShouldNotBeNull(); + Encoding.UTF8.GetString(r.ContinuationPoint!).ShouldBe($"{At(2).Ticks}|e3"); + } + + [Fact] + public async Task An_exact_fit_or_an_unbounded_read_has_no_continuation_point() + { + var db = await SeedAsync(fx); + var src = db.Source(); + var exact = await src.ReadEventsAsync(null, At(0), At(3), 3, Ct); + exact.Events.Count.ShouldBe(3); + exact.ContinuationPoint.ShouldBeNull(); + (await src.ReadEventsAsync(null, At(0), At(10), int.MaxValue, Ct)).Events.Count.ShouldBe(4); // node manager's "no limit" + (await src.ReadEventsAsync(null, At(0), At(10), -1, Ct)).ContinuationPoint.ShouldBeNull(); + } + + [Fact] + public async Task A_reversed_window_is_swapped() + { + var db = await SeedAsync(fx); + (await db.Source().ReadEventsAsync(null, At(3), At(0), 0, Ct)).Events.Select(e => e.EventId) + .ShouldBe(["e1", "e2", "e3"]); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProcessedReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProcessedReadTests.cs new file mode 100644 index 00000000..837577fd --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProcessedReadTests.cs @@ -0,0 +1,337 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +/// +/// Push-down aggregates over tag A (+0=10, +10=20, +20=Bad(30), +30=40, +40=50), 20 s intervals from the +/// sub-millisecond , window [t0, t0+80s) ⇒ four buckets: +/// B0 [0,20) = {10, 20}; B1 [20,40) = {Bad 30, 40}; B2 [40,60) = {50}; B3 [60,80) = {} (empty). +/// Every expected value below is hand-computed. +/// +[Collection(SqlHistorianCollection.Name)] +public sealed class ProcessedReadTests(SqlHistorianFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static readonly TimeSpan Twenty = TimeSpan.FromSeconds(20); + private static DateTime At(double s) => ReadTestDb.At(s); + + // Calculated = InfoType DataValue 0x400 | Calculated 0x1, OR-ed onto the code. + private const uint GoodCalc = 0x00000401u; + private const uint UncertainCalc = 0x40A40401u; // UncertainDataSubNormal | 0x401 + private const uint NoDataCalc = 0x809B0401u; // BadNoData | 0x401 + private const uint InvalidInputsCalc = 0x80D60401u; // BadAggregateInvalidInputs | 0x401 + + private static async Task ReadA(ReadTestDb db, HistoryAggregateType agg) => + await db.Source().ReadProcessedAsync("A", At(0), At(80), Twenty, agg, Ct); + + [Fact] + public void The_status_literals_carry_the_SDK_Calculated_aggregate_bits() + { + new Opc.Ua.StatusCode(GoodCalc).AggregateBits.ShouldBe(Opc.Ua.AggregateBits.Calculated); + new Opc.Ua.StatusCode(UncertainCalc).AggregateBits.ShouldBe(Opc.Ua.AggregateBits.Calculated); + new Opc.Ua.StatusCode(UncertainCalc).CodeBits.ShouldBe(Opc.Ua.StatusCodes.UncertainDataSubNormal); + new Opc.Ua.StatusCode(NoDataCalc).CodeBits.ShouldBe(Opc.Ua.StatusCodes.BadNoData); + new Opc.Ua.StatusCode(NoDataCalc).AggregateBits.ShouldBe(Opc.Ua.AggregateBits.Calculated); + new Opc.Ua.StatusCode(InvalidInputsCalc).CodeBits.ShouldBe(Opc.Ua.StatusCodes.BadAggregateInvalidInputs); + } + + [Theory] + [InlineData(HistoryAggregateType.Average, 15d, 40d, 50d)] + [InlineData(HistoryAggregateType.Minimum, 10d, 40d, 50d)] + [InlineData(HistoryAggregateType.Maximum, 20d, 40d, 50d)] + [InlineData(HistoryAggregateType.Total, 30d, 40d, 50d)] + [InlineData(HistoryAggregateType.Start, 10d, 40d, 50d)] + [InlineData(HistoryAggregateType.End, 20d, 40d, 50d)] + [InlineData(HistoryAggregateType.Delta, 10d, 0d, 0d)] + [InlineData(HistoryAggregateType.Range, 10d, 0d, 0d)] + public async Task Numeric_aggregates_over_the_fixture(HistoryAggregateType agg, double b0, double b1, double b2) + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await ReadA(db, agg)).Samples; + + s.Count.ShouldBe(4); + s.Select(x => x.Value).ShouldBe([b0, b1, b2, null]); + // B1 holds a Bad sample alongside a usable one ⇒ Uncertain; the empty B3 ⇒ BadNoData. + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalc, NoDataCalc]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40), At(60)]); // bucket starts + } + + [Fact] + public async Task Count_counts_usable_samples_as_int() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await ReadA(db, HistoryAggregateType.Count)).Samples; + s.Select(x => x.Value).ShouldBe([(object?)2, 1, 1, null]); + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalc, NoDataCalc]); + } + + [Theory] + [InlineData(HistoryAggregateType.PercentGood, 100d, 50d, 100d)] + [InlineData(HistoryAggregateType.PercentBad, 0d, 50d, 0d)] + public async Task Percent_aggregates_count_every_sample(HistoryAggregateType agg, double b0, double b1, double b2) + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await ReadA(db, agg)).Samples; + s.Select(x => x.Value).ShouldBe([b0, b1, b2, null]); + // A percentage is a statement about quality, so it is not itself degraded by the Bad input. + s.Select(x => x.StatusCode).ShouldBe([GoodCalc, GoodCalc, GoodCalc, NoDataCalc]); + } + + [Fact] + public async Task Actual_time_aggregates_stamp_the_sample_time() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var min = (await ReadA(db, HistoryAggregateType.MinimumActualTime)).Samples; + min.Select(x => x.Value).ShouldBe([10d, 40d, 50d, null]); + min.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(30), At(40), At(60)]); + min.Select(x => x.StatusCode).ShouldBe([GoodCalc, UncertainCalc, GoodCalc, NoDataCalc]); + var max = (await ReadA(db, HistoryAggregateType.MaximumActualTime)).Samples; + max.Select(x => x.Value).ShouldBe([20d, 40d, 50d, null]); + max.Select(x => x.SourceTimestampUtc).ShouldBe([At(10), At(30), At(40), At(60)]); + } + + [Fact] + public async Task Actual_time_ties_resolve_to_the_earliest_sample() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("T", + (At(0), HistorianValue.FromDouble(5), ReadTestDb.Good), + (At(2), HistorianValue.FromDouble(3), ReadTestDb.Good), + (At(4), HistorianValue.FromDouble(3), ReadTestDb.Good), + (At(6), HistorianValue.FromDouble(7), ReadTestDb.Good), + (At(8), HistorianValue.FromDouble(7), ReadTestDb.Good)); + var src = db.Source(); + var min = (await src.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.MinimumActualTime, Ct)).Samples.Single(); + min.Value.ShouldBe(3d); + min.SourceTimestampUtc.ShouldBe(At(2)); + var max = (await src.ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.MaximumActualTime, Ct)).Samples.Single(); + max.Value.ShouldBe(7d); + max.SourceTimestampUtc.ShouldBe(At(6)); + } + + [Fact] + public async Task Bucket_edges_are_start_inclusive_end_exclusive_to_the_tick() + { + // T0 is not millisecond-aligned: a DATEDIFF(millisecond) bucket formula would push the sample one tick + // before the +20 s edge into bucket 1 (it counts millisecond boundaries crossed, not elapsed time). + var db = await ReadTestDb.CreateAsync(fx); + var edge = At(20); + await db.SeedAsync("E", + (At(0).AddTicks(-1), HistorianValue.FromDouble(1000), ReadTestDb.Good), // before start: excluded + (At(0), HistorianValue.FromDouble(1), ReadTestDb.Good), // == start: B0 + (edge.AddTicks(-1), HistorianValue.FromDouble(2), ReadTestDb.Good), // last tick of B0 + (edge, HistorianValue.FromDouble(4), ReadTestDb.Good), // first tick of B1 + (At(40).AddTicks(-1), HistorianValue.FromDouble(8), ReadTestDb.Good), // last tick of B1 + (At(40), HistorianValue.FromDouble(2000), ReadTestDb.Good)); // == end: excluded + var s = (await db.Source().ReadProcessedAsync("E", At(0), At(40), Twenty, HistoryAggregateType.Total, Ct)).Samples; + s.Select(x => x.Value).ShouldBe([3d, 12d]); + s.ShouldAllBe(x => x.StatusCode == GoodCalc); + } + + [Fact] + public async Task A_partial_last_interval_is_truncated_at_end() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadProcessedAsync("A", At(0), At(50), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40)]); + s.Select(x => x.Value).ShouldBe([15d, 40d, 50d]); + } + + [Fact] + public async Task A_non_positive_interval_is_one_interval_over_the_window() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var src = db.Source(); + var avg = (await src.ReadProcessedAsync("A", At(0), At(50), TimeSpan.Zero, HistoryAggregateType.Average, Ct)).Samples.Single(); + avg.Value.ShouldBe(30d); // usable {10, 20, 40, 50} + avg.StatusCode.ShouldBe(UncertainCalc); + avg.SourceTimestampUtc.ShouldBe(At(0)); + (await src.ReadProcessedAsync("A", At(0), At(50), TimeSpan.FromSeconds(-1), HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); + } + + [Fact] + public async Task A_reverse_window_is_computed_forward_and_returned_descending() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var s = (await db.Source().ReadProcessedAsync("A", At(80), At(0), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(60), At(40), At(20), At(0)]); + s.Select(x => x.Value).ShouldBe([null, 50d, 40d, 15d]); + } + + [Fact] + public async Task Equal_start_and_end_yields_no_intervals() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + (await db.Source().ReadProcessedAsync("A", At(10), At(10), Twenty, HistoryAggregateType.Average, Ct)) + .Samples.ShouldBeEmpty(); + } + + [Fact] + public async Task Outage_markers_are_bad_samples_in_aggregates() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("M", + (At(0), HistorianValue.FromDouble(10), ReadTestDb.Good), + (At(5), HistorianValue.Empty, ReadTestDb.BadNoCommunication), + (At(10), HistorianValue.FromDouble(30), ReadTestDb.Good)); + var src = db.Source(); + async Task One(HistoryAggregateType a) => + (await src.ReadProcessedAsync("M", At(0), At(20), Twenty, a, Ct)).Samples.Single(); + + var avg = await One(HistoryAggregateType.Average); + avg.Value.ShouldBe(20d); + avg.StatusCode.ShouldBe(UncertainCalc); + (await One(HistoryAggregateType.Count)).Value.ShouldBe(2); + (await One(HistoryAggregateType.Start)).Value.ShouldBe(10d); + ((double)(await One(HistoryAggregateType.PercentBad)).Value!).ShouldBe(33.333333333333336, 1e-9); // 100 × 1/3 + } + + [Fact] + public async Task An_interval_of_only_bad_samples_is_BadNoData_except_the_percentages() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("Z", + (At(1), HistorianValue.Empty, ReadTestDb.BadNoCommunication), + (At(2), HistorianValue.FromDouble(5), ReadTestDb.Bad)); + var src = db.Source(); + async Task One(HistoryAggregateType a) => + (await src.ReadProcessedAsync("Z", At(0), At(20), Twenty, a, Ct)).Samples.Single(); + + (await One(HistoryAggregateType.Average)).StatusCode.ShouldBe(NoDataCalc); + (await One(HistoryAggregateType.Count)).StatusCode.ShouldBe(NoDataCalc); + var pb = await One(HistoryAggregateType.PercentBad); + pb.Value.ShouldBe(100d); + pb.StatusCode.ShouldBe(GoodCalc); + } + + [Fact] + public async Task Uncertain_samples_are_used_but_make_the_interval_uncertain() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("U", + (At(1), HistorianValue.FromDouble(4), ReadTestDb.Uncertain), + (At(2), HistorianValue.FromDouble(6), ReadTestDb.Good)); + var s = (await db.Source().ReadProcessedAsync("U", At(0), At(20), Twenty, HistoryAggregateType.Average, Ct)).Samples.Single(); + s.Value.ShouldBe(5d); + s.StatusCode.ShouldBe(UncertainCalc); + } + + [Fact] + public async Task String_tags_serve_count_start_end_and_refuse_numeric_aggregates() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("S", + (At(0), HistorianValue.FromString("a"), ReadTestDb.Good), + (At(10), HistorianValue.FromString("b"), ReadTestDb.Good)); + var src = db.Source(); + async Task One(HistoryAggregateType a) => + (await src.ReadProcessedAsync("S", At(0), At(20), Twenty, a, Ct)).Samples.Single(); + + (await One(HistoryAggregateType.Count)).Value.ShouldBe(2); + (await One(HistoryAggregateType.Start)).Value.ShouldBe("a"); + (await One(HistoryAggregateType.End)).Value.ShouldBe("b"); + (await One(HistoryAggregateType.PercentGood)).Value.ShouldBe(100d); + foreach (var a in new[] + { + HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, + HistoryAggregateType.Total, HistoryAggregateType.Delta, HistoryAggregateType.Range, + HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime, + }) + { + var r = await One(a); + r.StatusCode.ShouldBe(InvalidInputsCalc, a.ToString()); + r.Value.ShouldBeNull(); + } + } + + [Fact] + public async Task DateTime_tags_refuse_numeric_aggregates() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("D", (At(0), HistorianValue.FromDateTime(At(-100)), ReadTestDb.Good)); + (await db.Source().ReadProcessedAsync("D", At(0), At(20), Twenty, HistoryAggregateType.Average, Ct)) + .Samples.Single().StatusCode.ShouldBe(InvalidInputsCalc); + } + + [Fact] + public async Task Boolean_tags_aggregate_as_zero_or_one() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("B", + (At(0), HistorianValue.FromBoolean(true), ReadTestDb.Good), + (At(5), HistorianValue.FromBoolean(false), ReadTestDb.Good), + (At(10), HistorianValue.FromBoolean(true), ReadTestDb.Good), + (At(15), HistorianValue.FromBoolean(true), ReadTestDb.Good)); + var src = db.Source(); + async Task One(HistoryAggregateType a) => + (await src.ReadProcessedAsync("B", At(0), At(20), Twenty, a, Ct)).Samples.Single(); + (await One(HistoryAggregateType.Average)).Value.ShouldBe(0.75d); + (await One(HistoryAggregateType.Total)).Value.ShouldBe(3d); + (await One(HistoryAggregateType.Start)).Value.ShouldBe(true); + (await One(HistoryAggregateType.Count)).Value.ShouldBe(4); + } + + [Fact] + public async Task Unknown_tag_yields_every_interval_as_BadNoData() + { + var db = await ReadTestDb.CreateAsync(fx); + var s = (await db.Source().ReadProcessedAsync("nope", At(0), At(60), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Count.ShouldBe(3); + s.ShouldAllBe(x => x.StatusCode == NoDataCalc && x.Value == null); + } + + [Theory] + [InlineData(HistoryAggregateType.Interpolative)] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.StartBound)] + [InlineData(HistoryAggregateType.EndBound)] + [InlineData(HistoryAggregateType.DeltaBounds)] + [InlineData(HistoryAggregateType.DurationGood)] + [InlineData(HistoryAggregateType.DurationBad)] + [InlineData(HistoryAggregateType.WorstQuality)] + [InlineData(HistoryAggregateType.NumberOfTransitions)] + [InlineData(HistoryAggregateType.AnnotationCount)] + [InlineData(HistoryAggregateType.StandardDeviationPopulation)] + [InlineData(HistoryAggregateType.VariancePopulation)] + public async Task Unsupported_aggregates_throw_not_supported_and_are_not_health_failures(HistoryAggregateType agg) + { + var db = await ReadTestDb.CreateAsync(fx); + var src = db.Source(); + var ex = await Should.ThrowAsync( + () => src.ReadProcessedAsync("A", At(0), At(20), Twenty, agg, Ct)); + ex.Aggregate.ShouldBe(agg); + src.GetHealthSnapshot().TotalQueries.ShouldBe(0); + } + + [Fact] + public void Every_enum_member_is_either_served_or_refused() + { + // Thirteen served; anything appended to the enum later falls to the refusal (total default). + Enum.GetValues().Count(a => SqlAggregateQueries.FamilyOf(a) is not null).ShouldBe(13); + } + + [Fact] + public async Task The_bucket_count_is_capped_at_100000() + { + var db = await ReadTestDb.CreateAsync(fx); + var src = db.Source(); + var oneTick = TimeSpan.FromTicks(1); + (await src.ReadProcessedAsync("nope", At(0), At(0).AddTicks(100_000), oneTick, HistoryAggregateType.Count, Ct)) + .Samples.Count.ShouldBe(100_000); + await Should.ThrowAsync( + () => src.ReadProcessedAsync("nope", At(0), At(0).AddTicks(100_001), oneTick, HistoryAggregateType.Count, Ct)); + var h = src.GetHealthSnapshot(); + h.TotalFailures.ShouldBe(0); // a client-request error, not a backend fault + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RawReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RawReadTests.cs new file mode 100644 index 00000000..727f9b0f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RawReadTests.cs @@ -0,0 +1,257 @@ +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +/// +/// Seeds a fresh schema with plain INSERTs (independent of the value writer) and builds a data source over it. +/// Shared by the four read test classes. is deliberately NOT millisecond-aligned. +/// +internal sealed class ReadTestDb +{ + /// Base time: 2026-10-09 08:00:00.1234567 UTC (sub-millisecond ticks on purpose). + public static readonly DateTime T0 = new DateTime(2026, 10, 9, 8, 0, 0, DateTimeKind.Utc).AddTicks(1_234_567); + + public const uint Good = 0x00000000u; + public const uint Bad = 0x80000000u; + public const uint BadNoCommunication = 0x80310000u; + public const uint Uncertain = 0x40000000u; + + private ReadTestDb(SqlServerHistorianOptions options, SqlHistorianConnection conn, SqlServerSchemaBootstrapper boot) + { + Options = options; + Conn = conn; + Boot = boot; + } + + public SqlServerHistorianOptions Options { get; } + public SqlHistorianConnection Conn { get; } + public SqlServerSchemaBootstrapper Boot { get; } + + public static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond)); + + public static async Task CreateAsync(SqlHistorianFixture fx) + { + fx.SkipIfUnavailable(); + var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = SqlHistorianFixture.UniqueSchema() }; + var conn = new SqlHistorianConnection(o); + var boot = new SqlServerSchemaBootstrapper(conn, o); + await boot.EnsureAsync(CancellationToken.None); + return new ReadTestDb(o, conn, boot); + } + + /// A fresh data source (fresh catalog cache) over this schema. + public SqlServerHistorianDataSource Source() => new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot)); + + /// Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart). + public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples) + { + await using var c = await Conn.OpenAsync(CancellationToken.None); + int id; + await using (var cmd = Conn.CreateCommand(c, + $"INSERT {Conn.Q("Tag")} (Name, DataType, CreatedUtc, UpdatedUtc) OUTPUT inserted.TagId VALUES (@n, 0, SYSUTCDATETIME(), SYSUTCDATETIME())")) + { + cmd.Parameters.AddWithValue("@n", tag); + id = (int)(await cmd.ExecuteScalarAsync())!; + } + foreach (var (t, v, status) in samples) + { + var col = SampleRowMapper.ToColumns(v); + await using var cmd = Conn.CreateCommand(c, + $"INSERT {Conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " + + "VALUES (@id, @t, @k, @num, @int, @str, @st, @srv)"); + cmd.Parameters.AddWithValue("@id", id); + SqlHistorianConnection.AddDateTime2(cmd, "@t", t); + cmd.Parameters.AddWithValue("@k", col.ValueKind); + cmd.Parameters.Add(new SqlParameter("@num", System.Data.SqlDbType.Float) { Value = (object?)col.NumValue ?? DBNull.Value }); + cmd.Parameters.Add(new SqlParameter("@int", System.Data.SqlDbType.BigInt) { Value = (object?)col.IntValue ?? DBNull.Value }); + cmd.Parameters.Add(new SqlParameter("@str", System.Data.SqlDbType.NVarChar, -1) { Value = (object?)col.StrValue ?? DBNull.Value }); + cmd.Parameters.AddWithValue("@st", SampleRowMapper.ToDbStatus(status)); + SqlHistorianConnection.AddDateTime2(cmd, "@srv", t.AddSeconds(1)); + await cmd.ExecuteNonQueryAsync(); + } + return id; + } + + /// The fixture tag A of the brief: +0=10, +10=20, +20=Bad(30), +30=40, +40=50 (Double). + public Task SeedAAsync(string name = "A") => SeedAsync(name, + (At(0), HistorianValue.FromDouble(10), Good), + (At(10), HistorianValue.FromDouble(20), Good), + (At(20), HistorianValue.FromDouble(30), Bad), + (At(30), HistorianValue.FromDouble(40), Good), + (At(40), HistorianValue.FromDouble(50), Good)); + + public async Task ScalarAsync(string sql) + { + await using var c = await Conn.OpenAsync(CancellationToken.None); + await using var cmd = Conn.CreateCommand(c, sql); + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } +} + +[Collection(SqlHistorianCollection.Name)] +public sealed class RawReadTests(SqlHistorianFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static DateTime At(double s) => ReadTestDb.At(s); + + private static double?[] Values(HistoryReadResult r) => r.Samples.Select(s => (double?)s.Value).ToArray(); + + [Fact] + public async Task Forward_read_is_start_inclusive_end_exclusive_ascending_with_exact_ticks() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var r = await db.Source().ReadRawAsync("A", At(10), At(40), 0, Ct); + + Values(r).ShouldBe([20d, 30d, 40d]); // +10 included, +40 (== end) excluded + r.Samples.Select(s => s.SourceTimestampUtc!.Value.Ticks).ShouldBe([At(10).Ticks, At(20).Ticks, At(30).Ticks]); + r.Samples.Select(s => s.StatusCode).ShouldBe([ReadTestDb.Good, ReadTestDb.Bad, ReadTestDb.Good]); + r.Samples[0].SourceTimestampUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc); + r.Samples[0].ServerTimestampUtc.Ticks.ShouldBe(At(11).Ticks); // stored ServerTimeUtc, not "now" + r.ContinuationPoint.ShouldBeNull(); + } + + [Fact] + public async Task Forward_read_honours_the_limit() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + Values(await db.Source().ReadRawAsync("A", At(0), At(100), 2, Ct)).ShouldBe([10d, 20d]); + } + + [Fact] + public async Task Zero_max_values_is_unbounded() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + Values(await db.Source().ReadRawAsync("A", At(-1), At(100), 0, Ct)).ShouldBe([10d, 20d, 30d, 40d, 50d]); + Values(await db.Source().ReadRawAsync("A", At(-1), At(100), uint.MaxValue, Ct)).Length.ShouldBe(5); + } + + [Fact] + public async Task Reverse_read_is_descending_over_end_exclusive_start_inclusive() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + // start=+30, end=+10 ⇒ (+10, +30] descending. + Values(await db.Source().ReadRawAsync("A", At(30), At(10), 0, Ct)).ShouldBe([40d, 30d]); + Values(await db.Source().ReadRawAsync("A", At(100), At(-1), 2, Ct)).ShouldBe([50d, 40d]); + } + + [Fact] + public async Task Equal_start_and_end_is_a_point_read() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + Values(await db.Source().ReadRawAsync("A", At(10), At(10), 0, Ct)).ShouldBe([20d]); + (await db.Source().ReadRawAsync("A", At(10).AddTicks(1), At(10).AddTicks(1), 0, Ct)).Samples.ShouldBeEmpty(); + } + + [Fact] + public async Task A_resumed_page_from_the_last_timestamp_inclusive_continues_without_gaps() + { + // The node manager's paging: read cap N, resume at the last returned timestamp INCLUSIVE, trim it. + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var src = db.Source(); + var p1 = await src.ReadRawAsync("A", At(0), At(100), 2, Ct); + var p2 = await src.ReadRawAsync("A", p1.Samples[^1].SourceTimestampUtc!.Value, At(100), 2, Ct); + Values(p1).ShouldBe([10d, 20d]); + Values(p2).ShouldBe([20d, 30d]); // first is the boundary sample the node manager trims + } + + [Fact] + public async Task Unknown_tag_is_an_empty_result_and_reads_never_create_tags() + { + var db = await ReadTestDb.CreateAsync(fx); + var src = db.Source(); + (await src.ReadRawAsync("nope", At(0), At(10), 0, Ct)).Samples.ShouldBeEmpty(); + await src.ReadAtTimeAsync("nope", [At(1)], Ct); + await src.ReadProcessedAsync("nope", At(0), At(10), TimeSpan.FromSeconds(5), HistoryAggregateType.Average, Ct); + (await db.ScalarAsync($"SELECT COUNT(*) FROM {db.Conn.Q("Tag")}")).ShouldBe(0); + src.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task Outage_markers_and_bad_samples_come_back_with_their_bad_status() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAsync("M", + (At(0), HistorianValue.FromDouble(10), ReadTestDb.Good), + (At(5), HistorianValue.Empty, ReadTestDb.BadNoCommunication), + (At(7), HistorianValue.FromDouble(99), ReadTestDb.Bad)); + var r = await db.Source().ReadRawAsync("M", At(0), At(10), 0, Ct); + r.Samples.Count.ShouldBe(3); + r.Samples[1].Value.ShouldBeNull(); + r.Samples[1].StatusCode.ShouldBe(0x80310000u); + r.Samples[2].Value.ShouldBe(99d); + r.Samples[2].StatusCode.ShouldBe(0x80000000u); + } + + [Fact] + public async Task Every_kind_round_trips_to_its_CLR_type() + { + var db = await ReadTestDb.CreateAsync(fx); + var dt = new DateTime(2025, 1, 2, 3, 4, 5, DateTimeKind.Utc).AddTicks(7); + await db.SeedAsync("K", + (At(0), HistorianValue.FromInt64(-9_000_000_000L), ReadTestDb.Good), + (At(1), HistorianValue.FromBoolean(true), ReadTestDb.Good), + (At(2), HistorianValue.FromString("héllo"), ReadTestDb.Good), + (At(3), HistorianValue.FromDateTime(dt), ReadTestDb.Good), + (At(4), HistorianValue.FromDouble(1.5), ReadTestDb.Uncertain)); + var s = (await db.Source().ReadRawAsync("K", At(0), At(5), 0, Ct)).Samples; + s[0].Value.ShouldBe(-9_000_000_000L); + s[1].Value.ShouldBe(true); + s[2].Value.ShouldBe("héllo"); + s[3].Value.ShouldBe(dt); + ((DateTime)s[3].Value!).Kind.ShouldBe(DateTimeKind.Utc); + s[4].Value.ShouldBe(1.5); + s[4].StatusCode.ShouldBe(0x40000000u); + } + + [Fact] + public async Task Successful_reads_are_counted_in_health() + { + var db = await ReadTestDb.CreateAsync(fx); + await db.SeedAAsync(); + var src = db.Source(); + await src.ReadRawAsync("A", At(0), At(10), 0, Ct); + await src.ReadRawAsync("A", At(0), At(10), 0, Ct); + var h = src.GetHealthSnapshot(); + h.TotalQueries.ShouldBe(2); + h.TotalSuccesses.ShouldBe(2); + h.ConsecutiveFailures.ShouldBe(0); + h.LastSuccessTime.ShouldNotBeNull(); + h.ProcessConnectionOpen.ShouldBeTrue(); + } + + [Fact] + public async Task A_missing_schema_surfaces_in_health_LastError_and_is_not_latched() + { + fx.SkipIfUnavailable(); + var o = new SqlServerHistorianOptions + { + ConnectionString = fx.ConnectionString, Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false, + }; + var conn = new SqlHistorianConnection(o); + var boot = new SqlServerSchemaBootstrapper(conn, o); + var src = new SqlServerHistorianDataSource(conn, boot, new SqlServerTagCatalog(conn, boot)); + + await Should.ThrowAsync(() => src.ReadRawAsync("A", At(0), At(1), 0, Ct)); + await Should.ThrowAsync(() => src.ReadEventsAsync(null, At(0), At(1), 0, Ct)); + var h = src.GetHealthSnapshot(); + h.TotalFailures.ShouldBe(2); + h.ConsecutiveFailures.ShouldBe(2); + h.LastError!.ShouldContain("missing version 1"); + h.ProcessConnectionOpen.ShouldBeFalse(); + + // Another node creates the schema; the same data source then succeeds (bootstrap failure is not latched). + var creator = new SqlServerHistorianOptions { ConnectionString = o.ConnectionString, Schema = o.Schema }; + await new SqlServerSchemaBootstrapper(new SqlHistorianConnection(creator), creator).EnsureAsync(Ct); + (await src.ReadRawAsync("A", At(0), At(1), 0, Ct)).Samples.ShouldBeEmpty(); + src.GetHealthSnapshot().ConsecutiveFailures.ShouldBe(0); + } +}