dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(historian): extract Driver.Historian.Sql.Common from the SQL Server backend (no behaviour change) (1a)

Browse Source 
Moves the database-independent logic of the SQL Server historian backend into a new
shared library for the upcoming QuestDB backend (#666): status codes and bits, value/column
mapping, StoredSample + quality clamp, the AtTime and aggregate evaluators, BucketPlan,
LiveWritePreparer and HistoryUpdatePlanner (both with a resolutionTicks parameter, default 1),
tag-name rules, the bootstrap failure cache, health counters, alarm event ids, the event
continuation and severity clamp. The SQL Server backend now calls it; its T-SQL, TVP
builders, connection, bootstrapper and catalog stay.

Tests: StatusCodeOracleTests, ValueWriterPrepareTests, the column half of SampleRowMapperTests
and ProcessedReadTests.Every_enum_member_is_either_served_or_refused move to the new
Sql.Common.Tests under the same names. SqlServer integration tests get mechanical renames
only. Test-name set across the three projects is identical to the baseline (160 tests).

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 01:26:49 -04:00
parent cdea173bd1
commit 8c6d6aff6a
49 changed files with 1262 additions and 672 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs
+34 -196
View File
@@ -4,6 +4,7 @@ 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;
@@ -39,7 +40,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// on Raw and AtTime alike.
/// </para>
/// <para>
/// <b>Long names.</b> A tag name or event source name longer than <see cref="SqlServerTagCatalog.MaxNameLength"/>
/// <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>
@@ -47,7 +48,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
public sealed class SqlServerHistorianDataSource : IHistorianDataSource
{
/// <summary>Spec §9.7: a processed read covering more intervals than this is refused.</summary>
public const int MaxProcessedBuckets = 100_000;
public const int MaxProcessedBuckets = BucketPlan.MaxBuckets;
private const string SampleColumns = "SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc";
@@ -57,14 +58,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
private readonly ILogger _log;
private readonly TimeProvider _time;
private readonly object _healthLock = new();
private DateTime? _lastSuccessUtc;
private DateTime? _lastFailureUtc;
private string? _lastError;
private long _totalQueries;
private long _totalSuccesses;
private long _totalFailures;
private int _consecutiveFailures;
private readonly HistorianHealthCounters _health = new();
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
/// <param name="conn">Connection helper.</param>
@@ -192,7 +186,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
StoredSample? p = reader.IsDBNull(1) ? null : ReadStored(reader, 1);
StoredSample? n = reader.IsDBNull(8) ? null : ReadStored(reader, 8);
var serverTime = p is { } ps && ps.TimeUtc == t ? SqlHistorianConnection.GetUtc(reader, 7) : now;
byTicks[t.Ticks] = AtTimeValue(t, p, n, serverTime, now);
byTicks[t.Ticks] = AtTimeEvaluator.Evaluate(t, p, n, serverTime, now);
}
}
@@ -202,7 +196,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
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);
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, t, now);
}
RecordSuccess();
@@ -215,44 +209,6 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
}
}
/// <summary>Applies the AtTime rules (see <see cref="ReadAtTimeAsync"/>) to one requested time.</summary>
private static DataValueSnapshot AtTimeValue(DateTime t, StoredSample? prev, StoredSample? next, DateTime serverTime, DateTime now)
{
if (prev is not { } p)
return new DataValueSnapshot(null, SqlHistorianStatusCodes.BadNoData, t, serverTime);
if (p.TimeUtc == t)
return new DataValueSnapshot(p.Value.ToObject(), p.Status, t, serverTime);
// Inside an outage: report the outage's own Bad code, never an upgraded Uncertain.
if (!p.IsUsable)
return new DataValueSnapshot(null, p.Status & StatusBits.CodeBitsMask, t, serverTime);
// Usable previous, Bad next: hold the previous value up to the outage; do not interpolate into it.
if (next is { IsUsable: false })
return new DataValueSnapshot(p.Value.ToObject(),
HistorianInfoBits.WithInterpolated(SqlHistorianStatusCodes.UncertainDataSubNormal), t, serverTime);
if (next is { } n && p.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64
&& n.Value.TryGetDouble(out var nv))
{
var fraction = (double)(t.Ticks - p.TimeUtc.Ticks) / (n.TimeUtc.Ticks - p.TimeUtc.Ticks);
var v = p.Value.Number + (nv - p.Value.Number) * fraction;
object value = p.Value.Kind == HistorianValueKind.Int64
? (long)Math.Round(v, MidpointRounding.AwayFromZero)
: v;
var status = p.IsGood && n.IsGood ? SqlHistorianStatusCodes.Good : SqlHistorianStatusCodes.UncertainDataSubNormal;
return new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), t, serverTime);
}
// Beyond the last sample AND beyond now: the value is held into the future, so it is not known to be Good.
if (next is null && t > now)
return new DataValueSnapshot(p.Value.ToObject(),
HistorianInfoBits.WithInterpolated(SqlHistorianStatusCodes.UncertainDataSubNormal), t, serverTime);
// Stepped (non-numeric kind, or after the last sample): hold the previous value and its status.
return new DataValueSnapshot(p.Value.ToObject(), HistorianInfoBits.WithInterpolated(p.Status), t, serverTime);
}
// ------------------------------------------------------------------------------------------ Events
/// <inheritdoc />
@@ -269,7 +225,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
{
// No stored event can carry a source longer than the column (the alarm writer dead-letters it rather than
// truncating), so an over-long filter matches nothing — never a truncated prefix of someone else's source.
if (sourceName is not null && sourceName.Length > SqlServerTagCatalog.MaxNameLength)
if (sourceName is not null && sourceName.Length > HistorianTagNames.MaxLength)
return new HistoricalEventsResult([], null);
try
{
@@ -301,10 +257,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
SqlHistorianConnection.GetUtc(reader, 2),
SqlHistorianConnection.GetUtc(reader, 3),
reader.IsDBNull(4) ? null : reader.GetString(4),
(ushort)Math.Clamp(reader.GetInt32(5), 0, ushort.MaxValue));
HistoricalEventRows.ClampSeverity(reader.GetInt32(5)));
if (bounded && events.Count == maxEvents)
{
continuation = Encoding.UTF8.GetBytes($"{e.EventTimeUtc.Ticks}|{e.EventId}");
continuation = Encoding.UTF8.GetBytes(EventContinuation.Encode(e.EventTimeUtc, e.EventId));
break;
}
events.Add(e);
@@ -352,57 +308,38 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
HistoryAggregateType aggregate, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReference);
var family = SqlAggregateQueries.FamilyOf(aggregate) ?? throw new HistoryAggregateNotSupportedException(aggregate);
var reverse = endUtc < startUtc;
var (start, end) = reverse ? (endUtc, startUtc) : (startUtc, endUtc);
var windowTicks = end.Ticks - start.Ticks;
// Everything stays in 100 ns ticks (≤ ≈3.2e18 across 0001–9999, inside a long) — no ns scaling that could
// wrap. An interval wider than the window is one bucket, so clamp it to the window; the ceiling division
// is written without "+ interval - 1" so a TimeSpan near MaxValue cannot overflow either.
var intervalTicks = interval > TimeSpan.Zero ? Math.Min(interval.Ticks, windowTicks) : windowTicks;
long bucketCount = windowTicks == 0 ? 0 : windowTicks / intervalTicks + (windowTicks % intervalTicks == 0 ? 0 : 1);
if (bucketCount > MaxProcessedBuckets)
throw new ArgumentOutOfRangeException(nameof(interval), bucketCount,
$"A processed read may cover at most {MaxProcessedBuckets} intervals; this window holds {bucketCount}.");
var family = AggregateFamilies.FamilyOf(aggregate) ?? throw new HistoryAggregateNotSupportedException(aggregate);
var plan = BucketPlan.Create(startUtc, endUtc, interval);
try
{
var now = _time.GetUtcNow().UtcDateTime;
var rows = new Dictionary<long, BucketRow>();
var tagId = bucketCount == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
var evaluated = new Dictionary<long, DataValueSnapshot>();
var tagId = plan.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),
AggregateFamily.Simple => SqlAggregateQueries.Simple(sample),
AggregateFamily.StartEnd => SqlAggregateQueries.StartEnd(sample),
_ => SqlAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime),
};
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, sql);
cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
SqlHistorianConnection.AddDateTime2(cmd, "@start", DateTime.SpecifyKind(start, DateTimeKind.Utc));
SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(end, DateTimeKind.Utc));
cmd.Parameters.Add(new SqlParameter("@intervalTicks", SqlDbType.BigInt) { Value = intervalTicks });
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 });
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;
evaluated[row.Bucket] = AggregateEvaluator.Evaluate(aggregate, row, plan.BucketStart(row.Bucket), now);
}
}
var result = new List<DataValueSnapshot>((int)bucketCount);
for (long i = 0; i < bucketCount; i++)
{
var bucketStart = DateTime.SpecifyKind(new DateTime(start.Ticks + i * intervalTicks), DateTimeKind.Utc);
result.Add(rows.TryGetValue(i, out var row)
? Evaluate(aggregate, row, bucketStart, now)
: NoData(bucketStart, now));
}
if (reverse) result.Reverse();
var result = plan.Assemble(evaluated, bucketStart => AggregateEvaluator.NoData(bucketStart, now));
RecordSuccess();
return new HistoryReadResult(result, ContinuationPoint: null);
@@ -414,56 +351,13 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
}
}
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)
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 : SampleRowMapper.FromColumns(r.GetByte(i), D(r, i + 1), L(r, i + 2), S(r, i + 3));
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
{
@@ -477,13 +371,13 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
var f = SqlAggregateQueries.FirstFamilyColumn;
switch (family)
{
case SqlAggregateFamily.Simple:
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 SqlAggregateFamily.StartEnd:
case AggregateFamily.StartEnd:
row.First = V(reader, f);
row.Last = V(reader, f + 5);
break;
@@ -495,44 +389,23 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
return row;
}
/// <summary>One aggregate query row (one non-empty bucket).</summary>
private sealed class BucketRow
{
public long Bucket;
public int Total, Usable, Good, Bad, NonNumeric;
public double? Avg, Min, Max, Sum, Extreme;
public DateTime? ExtremeTime;
public HistorianValue First, Last;
}
// ------------------------------------------------------------------------------------------ Shared
/// <summary>A stored sample read from the <c>{SampleColumns}</c> column run starting at an ordinal.</summary>
private readonly record struct StoredSample(DateTime TimeUtc, HistorianValue Value, uint Status)
{
/// <summary>Severity Good or Uncertain, and a value present (an Empty sample is an outage marker).</summary>
public bool IsUsable => !StatusBits.IsBad(Status) && Value.Kind != HistorianValueKind.Empty;
/// <summary>Usable with severity Good.</summary>
public bool IsGood => IsUsable && StatusBits.IsGood(Status);
}
/// <summary>
/// Reads <c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode</c> starting at <paramref name="o"/>.
/// An Empty (no-value) row whose stored status is not Bad is clamped to <c>BadNoData</c> — defence in depth for
/// rows written before the writer forced every Empty sample Bad.
/// 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 = SampleRowMapper.FromColumns(
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 = SampleRowMapper.FromDbStatus(r.GetInt32(o + 5));
if (value.Kind == HistorianValueKind.Empty && !StatusBits.IsBad(status))
status = SqlHistorianStatusCodes.BadNoData;
return new StoredSample(SqlHistorianConnection.GetUtc(r, o), value, status);
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>
@@ -541,50 +414,15 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
// ------------------------------------------------------------------------------------------ Health
/// <inheritdoc />
public HistorianHealthSnapshot GetHealthSnapshot()
{
lock (_healthLock)
{
var open = _totalSuccesses > 0 && _consecutiveFailures == 0;
return new HistorianHealthSnapshot(
TotalQueries: _totalQueries,
TotalSuccesses: _totalSuccesses,
TotalFailures: _totalFailures,
ConsecutiveFailures: _consecutiveFailures,
LastSuccessTime: _lastSuccessUtc,
LastFailureTime: _lastFailureUtc,
LastError: _lastError,
ProcessConnectionOpen: open,
EventConnectionOpen: open,
ActiveProcessNode: null,
ActiveEventNode: null,
Nodes: []);
}
}
public HistorianHealthSnapshot GetHealthSnapshot() => _health.Snapshot();
private void RecordSuccess()
{
lock (_healthLock)
{
_totalQueries++;
_totalSuccesses++;
_consecutiveFailures = 0;
_lastSuccessUtc = DateTime.UtcNow;
}
}
private void RecordSuccess() => _health.RecordSuccess();
private void RecordFailure(Exception ex)
{
// A cancelled read (deadline / client abort) is not a backend fault.
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
if (ex is OperationCanceledException) return;
lock (_healthLock)
{
_totalQueries++;
_totalFailures++;
_consecutiveFailures++;
_lastFailureUtc = DateTime.UtcNow;
_lastError = $"{ex.GetType().Name}: {ex.Message}";
}
_health.RecordFailure(ex);
_log.LogDebug("SQL historian read failed ({ExceptionType})", ex.GetType().Name);
}