From 17962d4233c5a8bfe908fd9e048a335346183159 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 12:52:15 -0400 Subject: [PATCH] feat(historian): SQL Server backend skeleton - schema DDL, locked bootstrapper, options, connection/mapper/tag-catalog helpers, status consts, env-gated fixture Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- ZB.MOM.WW.OtOpcUa.slnx | 3 + docs/historian/sqlserver/001_initial.sql | 69 ++++++++ .../SampleRowMapper.cs | 105 ++++++++++++ .../Schema/001_initial.sql | 69 ++++++++ .../SqlHistorianConnection.cs | 130 +++++++++++++++ .../SqlHistorianStatusCodes.cs | 44 +++++ .../SqlServerHistorianOptions.cs | 54 ++++++ .../SqlServerSchemaBootstrapper.cs | 155 ++++++++++++++++++ .../SqlServerTagCatalog.cs | 80 +++++++++ ....OtOpcUa.Driver.Historian.SqlServer.csproj | 34 ++++ ....WW.OtOpcUa.Core.Abstractions.Tests.csproj | 1 + .../SchemaBootstrapIntegrationTests.cs | 139 ++++++++++++++++ .../SqlHistorianFixture.cs | 79 +++++++++ ...istorian.SqlServer.IntegrationTests.csproj | 26 +++ .../SampleRowMapperTests.cs | 65 ++++++++ .../SchemaScriptTests.cs | 32 ++++ .../SqlServerHistorianOptionsTests.cs | 46 ++++++ .../StatusCodeOracleTests.cs | 28 ++++ ...Ua.Driver.Historian.SqlServer.Tests.csproj | 28 ++++ 19 files changed, 1187 insertions(+) create mode 100644 docs/historian/sqlserver/001_initial.sql create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SampleRowMapper.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/Schema/001_initial.sql create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianStatusCodes.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerSchemaBootstrapper.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagCatalog.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SchemaBootstrapIntegrationTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlHistorianFixture.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SampleRowMapperTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SchemaScriptTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianOptionsTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/StatusCodeOracleTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests.csproj diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx index fd2ce393..959ac9ac 100644 --- a/ZB.MOM.WW.OtOpcUa.slnx +++ b/ZB.MOM.WW.OtOpcUa.slnx @@ -27,6 +27,7 @@ + @@ -100,6 +101,8 @@ + + diff --git a/docs/historian/sqlserver/001_initial.sql b/docs/historian/sqlserver/001_initial.sql new file mode 100644 index 00000000..d4051334 --- /dev/null +++ b/docs/historian/sqlserver/001_initial.sql @@ -0,0 +1,69 @@ +-- OtOpcUa SQL Server historian backend - schema version 1. +-- $(Schema) is substituted by SqlServerSchemaBootstrapper (identifier-validated) BEFORE the script is split on GO lines. +-- Every statement is guarded so a re-run is a no-op. +IF SCHEMA_ID(N'$(Schema)') IS NULL EXEC(N'CREATE SCHEMA [$(Schema)]') +GO +IF OBJECT_ID(N'[$(Schema)].[SchemaVersion]', N'U') IS NULL +CREATE TABLE [$(Schema)].[SchemaVersion] ( + Version int NOT NULL PRIMARY KEY, + AppliedUtc datetime2(7) NOT NULL) +GO +IF OBJECT_ID(N'[$(Schema)].[Tag]', N'U') IS NULL +CREATE TABLE [$(Schema)].[Tag] ( + TagId int IDENTITY(1,1) NOT NULL PRIMARY KEY, + Name nvarchar(400) NOT NULL CONSTRAINT UQ_Tag_Name UNIQUE, + DataType tinyint NOT NULL, + EngineeringUnit nvarchar(64) NULL, + Description nvarchar(400) NULL, + CreatedUtc datetime2(7) NOT NULL, + UpdatedUtc datetime2(7) NOT NULL) +GO +IF OBJECT_ID(N'[$(Schema)].[Sample]', N'U') IS NULL +CREATE TABLE [$(Schema)].[Sample] ( + TagId int NOT NULL, + SourceTimeUtc datetime2(7) NOT NULL, + ValueKind tinyint NOT NULL, + NumValue float NULL, + IntValue bigint NULL, + StrValue nvarchar(max) NULL, + StatusCode int NOT NULL, + ServerTimeUtc datetime2(7) NOT NULL, + CONSTRAINT PK_Sample PRIMARY KEY CLUSTERED (TagId, SourceTimeUtc), + CONSTRAINT FK_Sample_Tag FOREIGN KEY (TagId) REFERENCES [$(Schema)].[Tag](TagId)) +GO +IF OBJECT_ID(N'[$(Schema)].[Event]', N'U') IS NULL +CREATE TABLE [$(Schema)].[Event] ( + EventId nvarchar(64) NOT NULL CONSTRAINT PK_Event PRIMARY KEY NONCLUSTERED, + SourceName nvarchar(400) NOT NULL, + AlarmId nvarchar(400) NOT NULL, + AlarmName nvarchar(400) NOT NULL, + AlarmTypeName nvarchar(100) NOT NULL, + Severity int NOT NULL, + EventKind nvarchar(50) NOT NULL, + Message nvarchar(max) NULL, + UserName nvarchar(200) NULL, + Comment nvarchar(max) NULL, + EventTimeUtc datetime2(7) NOT NULL, + ReceivedTimeUtc datetime2(7) NOT NULL) +GO +IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'CX_Event_Time' AND object_id = OBJECT_ID(N'[$(Schema)].[Event]')) +CREATE CLUSTERED INDEX CX_Event_Time ON [$(Schema)].[Event] (EventTimeUtc, EventId) +GO +IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'IX_Event_Source_Time' AND object_id = OBJECT_ID(N'[$(Schema)].[Event]')) +CREATE INDEX IX_Event_Source_Time ON [$(Schema)].[Event] (SourceName, EventTimeUtc) +GO +IF TYPE_ID(N'[$(Schema)].[SampleRow]') IS NULL +CREATE TYPE [$(Schema)].[SampleRow] AS TABLE ( + SourceTimeUtc datetime2(7) NOT NULL PRIMARY KEY, + ValueKind tinyint NOT NULL, + NumValue float NULL, + IntValue bigint NULL, + StrValue nvarchar(max) NULL, + StatusCode int NOT NULL) +GO +IF TYPE_ID(N'[$(Schema)].[TimeRow]') IS NULL +CREATE TYPE [$(Schema)].[TimeRow] AS TABLE (T datetime2(7) NOT NULL PRIMARY KEY) +GO +IF NOT EXISTS (SELECT 1 FROM [$(Schema)].[SchemaVersion] WHERE Version = 1) +INSERT [$(Schema)].[SchemaVersion] (Version, AppliedUtc) VALUES (1, SYSUTCDATETIME()) +GO diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SampleRowMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SampleRowMapper.cs new file mode 100644 index 00000000..e9e20ac2 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SampleRowMapper.cs @@ -0,0 +1,105 @@ +using System.Data; +using Microsoft.Data.SqlClient.Server; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// One sample bound for a SampleRow TVP. +/// Source timestamp (UTC, full tick precision). +/// Typed value. +/// OPC UA status code (stored bit-for-bit as int). +internal readonly record struct SqlSampleRow(DateTime TimeUtc, HistorianValue Value, uint StatusCode); + +/// The (ValueKind, NumValue, IntValue, StrValue) column quad for a value. +internal readonly record struct SampleColumns(byte ValueKind, double? NumValue, long? IntValue, string? StrValue); + +/// +/// HistorianValue ⇄ column mapping (spec §4) and the SqlDataRecord TVP builders. Mapping: +/// Double → Num; Int64 → Int + Num(double); Boolean → Int 0/1 + Num 0/1; String → Str; DateTime → Int (ticks); +/// Empty → all NULL. +/// +internal static class SampleRowMapper +{ + private static readonly SqlMetaData[] SampleMeta = + [ + new("SourceTimeUtc", SqlDbType.DateTime2, 0, 7), + new("ValueKind", SqlDbType.TinyInt), + new("NumValue", SqlDbType.Float), + new("IntValue", SqlDbType.BigInt), + new("StrValue", SqlDbType.NVarChar, SqlMetaData.Max), + new("StatusCode", SqlDbType.Int), + ]; + + private static readonly SqlMetaData[] TimeMeta = [new("T", SqlDbType.DateTime2, 0, 7)]; + + /// Splits a value into its column quad. + /// Typed value. + /// The columns to store. + public static SampleColumns ToColumns(HistorianValue v) => v.Kind switch + { + HistorianValueKind.Double => new((byte)v.Kind, v.Number, null, null), + HistorianValueKind.Int64 => new((byte)v.Kind, v.Number, v.Integer, null), + HistorianValueKind.Boolean => new((byte)v.Kind, v.Integer != 0 ? 1d : 0d, v.Integer != 0 ? 1L : 0L, null), + HistorianValueKind.String => new((byte)v.Kind, null, null, v.Text ?? ""), + HistorianValueKind.DateTime => new((byte)v.Kind, null, v.Integer, null), + _ => new((byte)HistorianValueKind.Empty, null, null, null), + }; + + /// Rebuilds a value from stored columns. + /// ValueKind column. + /// NumValue column (null = NULL). + /// IntValue column (null = NULL). + /// StrValue column (null = NULL). + /// The typed value; an unknown kind yields . + public static HistorianValue FromColumns(byte valueKind, double? num, long? integer, string? str) => (HistorianValueKind)valueKind switch + { + HistorianValueKind.Double => HistorianValue.FromDouble(num ?? 0d), + HistorianValueKind.Int64 => HistorianValue.FromInt64(integer ?? 0L), + HistorianValueKind.Boolean => HistorianValue.FromBoolean((integer ?? 0L) != 0), + HistorianValueKind.String => HistorianValue.FromString(str ?? ""), + HistorianValueKind.DateTime => new HistorianValue(HistorianValueKind.DateTime, 0d, integer ?? 0L, null), + _ => HistorianValue.Empty, + }; + + /// Stores a uint status bit-for-bit in the int column. + /// Status code. + /// Same bits as int. + public static int ToDbStatus(uint status) => unchecked((int)status); + + /// Reads the int status column back to a uint code. + /// Column value. + /// Same bits as uint. + public static uint FromDbStatus(int dbStatus) => unchecked((uint)dbStatus); + + /// Builds SampleRow TVP records (datetime2(7); never a DataTable). Lazy; enumerate once. + /// Samples; must be time-unique (the TVP has a PK on time) and non-empty when bound. + /// Records for . + public static IEnumerable ToSampleRecords(IEnumerable rows) + { + foreach (var r in rows) + { + var c = ToColumns(r.Value); + var rec = new SqlDataRecord(SampleMeta); + rec.SetDateTime(0, DateTime.SpecifyKind(r.TimeUtc, DateTimeKind.Utc)); + rec.SetByte(1, c.ValueKind); + if (c.NumValue is { } n) rec.SetDouble(2, n); else rec.SetDBNull(2); + if (c.IntValue is { } i) rec.SetInt64(3, i); else rec.SetDBNull(3); + if (c.StrValue is { } s) rec.SetString(4, s); else rec.SetDBNull(4); + rec.SetInt32(5, ToDbStatus(r.StatusCode)); + yield return rec; + } + } + + /// Builds TimeRow TVP records (datetime2(7)). Lazy; enumerate once. + /// Distinct UTC times; non-empty when bound. + /// Records for . + public static IEnumerable ToTimeRecords(IEnumerable timesUtc) + { + foreach (var t in timesUtc) + { + var rec = new SqlDataRecord(TimeMeta); + rec.SetDateTime(0, DateTime.SpecifyKind(t, DateTimeKind.Utc)); + yield return rec; + } + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/Schema/001_initial.sql b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/Schema/001_initial.sql new file mode 100644 index 00000000..d4051334 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/Schema/001_initial.sql @@ -0,0 +1,69 @@ +-- OtOpcUa SQL Server historian backend - schema version 1. +-- $(Schema) is substituted by SqlServerSchemaBootstrapper (identifier-validated) BEFORE the script is split on GO lines. +-- Every statement is guarded so a re-run is a no-op. +IF SCHEMA_ID(N'$(Schema)') IS NULL EXEC(N'CREATE SCHEMA [$(Schema)]') +GO +IF OBJECT_ID(N'[$(Schema)].[SchemaVersion]', N'U') IS NULL +CREATE TABLE [$(Schema)].[SchemaVersion] ( + Version int NOT NULL PRIMARY KEY, + AppliedUtc datetime2(7) NOT NULL) +GO +IF OBJECT_ID(N'[$(Schema)].[Tag]', N'U') IS NULL +CREATE TABLE [$(Schema)].[Tag] ( + TagId int IDENTITY(1,1) NOT NULL PRIMARY KEY, + Name nvarchar(400) NOT NULL CONSTRAINT UQ_Tag_Name UNIQUE, + DataType tinyint NOT NULL, + EngineeringUnit nvarchar(64) NULL, + Description nvarchar(400) NULL, + CreatedUtc datetime2(7) NOT NULL, + UpdatedUtc datetime2(7) NOT NULL) +GO +IF OBJECT_ID(N'[$(Schema)].[Sample]', N'U') IS NULL +CREATE TABLE [$(Schema)].[Sample] ( + TagId int NOT NULL, + SourceTimeUtc datetime2(7) NOT NULL, + ValueKind tinyint NOT NULL, + NumValue float NULL, + IntValue bigint NULL, + StrValue nvarchar(max) NULL, + StatusCode int NOT NULL, + ServerTimeUtc datetime2(7) NOT NULL, + CONSTRAINT PK_Sample PRIMARY KEY CLUSTERED (TagId, SourceTimeUtc), + CONSTRAINT FK_Sample_Tag FOREIGN KEY (TagId) REFERENCES [$(Schema)].[Tag](TagId)) +GO +IF OBJECT_ID(N'[$(Schema)].[Event]', N'U') IS NULL +CREATE TABLE [$(Schema)].[Event] ( + EventId nvarchar(64) NOT NULL CONSTRAINT PK_Event PRIMARY KEY NONCLUSTERED, + SourceName nvarchar(400) NOT NULL, + AlarmId nvarchar(400) NOT NULL, + AlarmName nvarchar(400) NOT NULL, + AlarmTypeName nvarchar(100) NOT NULL, + Severity int NOT NULL, + EventKind nvarchar(50) NOT NULL, + Message nvarchar(max) NULL, + UserName nvarchar(200) NULL, + Comment nvarchar(max) NULL, + EventTimeUtc datetime2(7) NOT NULL, + ReceivedTimeUtc datetime2(7) NOT NULL) +GO +IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'CX_Event_Time' AND object_id = OBJECT_ID(N'[$(Schema)].[Event]')) +CREATE CLUSTERED INDEX CX_Event_Time ON [$(Schema)].[Event] (EventTimeUtc, EventId) +GO +IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'IX_Event_Source_Time' AND object_id = OBJECT_ID(N'[$(Schema)].[Event]')) +CREATE INDEX IX_Event_Source_Time ON [$(Schema)].[Event] (SourceName, EventTimeUtc) +GO +IF TYPE_ID(N'[$(Schema)].[SampleRow]') IS NULL +CREATE TYPE [$(Schema)].[SampleRow] AS TABLE ( + SourceTimeUtc datetime2(7) NOT NULL PRIMARY KEY, + ValueKind tinyint NOT NULL, + NumValue float NULL, + IntValue bigint NULL, + StrValue nvarchar(max) NULL, + StatusCode int NOT NULL) +GO +IF TYPE_ID(N'[$(Schema)].[TimeRow]') IS NULL +CREATE TYPE [$(Schema)].[TimeRow] AS TABLE (T datetime2(7) NOT NULL PRIMARY KEY) +GO +IF NOT EXISTS (SELECT 1 FROM [$(Schema)].[SchemaVersion] WHERE Version = 1) +INSERT [$(Schema)].[SchemaVersion] (Version, AppliedUtc) VALUES (1, SYSUTCDATETIME()) +GO diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs new file mode 100644 index 00000000..90e1b46e --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs @@ -0,0 +1,130 @@ +using System.Data; +using Microsoft.Data.SqlClient; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// Connection + command helpers shared by every SQL historian component. +/// Usage contract for writers/readers: call +/// once (it is cheap after the first success), then . Build SQL with +/// for every object name and bind every timestamp through or a TVP built by +/// — never a DataTable (it infers 3.33 ms datetime). +/// +internal sealed class SqlHistorianConnection +{ + private static readonly int[] TransientNumbers = + [1205, -2, 40613, 40197, 40501, 49918, 4060, 233, 10053, 10054, 10060, 11001]; + + private readonly string _connectionString; + + /// Creates the helper; options must already be valid. + /// Validated options. + public SqlHistorianConnection(SqlServerHistorianOptions options) + { + ArgumentNullException.ThrowIfNull(options); + if (!SqlServerHistorianOptions.IsValidSchemaName(options.Schema)) + throw new ArgumentException("Invalid schema name.", nameof(options)); + _connectionString = options.ConnectionString; + Schema = options.Schema; + CommandTimeoutSeconds = Math.Max(1, (int)Math.Ceiling(options.CommandTimeout.TotalSeconds)); + } + + /// The validated schema name. + public string Schema { get; } + + /// Command timeout applied by , in seconds. + public int CommandTimeoutSeconds { get; } + + /// Schema-qualifies an object name: [schema].[obj]. + /// Unquoted object name (a fixed literal from this project, never user input). + /// The quoted two-part name. + public string Q(string obj) => $"[{Schema}].[{obj}]"; + + /// Opens a new connection (no bootstrap; see the class remarks). + /// Cancellation. + /// An open connection the caller disposes. + public async Task OpenAsync(CancellationToken ct) + { + var conn = new SqlConnection(_connectionString); + try + { + await conn.OpenAsync(ct).ConfigureAwait(false); + return conn; + } + catch + { + await conn.DisposeAsync().ConfigureAwait(false); + throw; + } + } + + /// Creates a text command with the configured timeout. + /// Open connection. + /// Command text. + /// Optional transaction. + /// The command. + public SqlCommand CreateCommand(SqlConnection conn, string sql, SqlTransaction? tx = null) + { + var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + cmd.CommandTimeout = CommandTimeoutSeconds; + cmd.Transaction = tx; + return cmd; + } + + /// Adds a datetime2(7) scalar parameter (full 100 ns tick precision, UTC kind). + /// Target command. + /// Parameter name including @. + /// Timestamp; treated as UTC. + /// The added parameter. + public static SqlParameter AddDateTime2(SqlCommand cmd, string name, DateTime value) + { + var p = new SqlParameter(name, SqlDbType.DateTime2) + { + Scale = 7, + Value = DateTime.SpecifyKind(value, DateTimeKind.Utc), + }; + cmd.Parameters.Add(p); + return p; + } + + /// Adds a table-valued parameter of one of this schema's table types. + /// Target command. + /// Parameter name including @. + /// Unqualified type name (SampleRow or TimeRow). + /// Non-empty SqlDataRecord sequence (SqlClient rejects an empty one — skip the call). + /// The added parameter. + public SqlParameter AddStructured(SqlCommand cmd, string name, string typeName, + IEnumerable rows) + { + var p = new SqlParameter(name, SqlDbType.Structured) { TypeName = Q(typeName), Value = rows }; + cmd.Parameters.Add(p); + return p; + } + + /// Reads a datetime2 column as a UTC (the driver returns Unspecified). + /// Reader positioned on a row. + /// Column ordinal. + /// UTC timestamp. + public static DateTime GetUtc(SqlDataReader reader, int ordinal) => + DateTime.SpecifyKind(reader.GetDateTime(ordinal), DateTimeKind.Utc); + + /// True for connection/timeout/deadlock/throttling errors worth retrying. + /// Exception from a SQL call. + /// Whether a retry may succeed. + public static bool IsTransient(Exception ex) + { + if (ex is TimeoutException) return true; + if (ex is not SqlException sql) return false; + if (sql.IsTransient) return true; + foreach (SqlError e in sql.Errors) + if (Array.IndexOf(TransientNumbers, e.Number) >= 0) return true; + return Array.IndexOf(TransientNumbers, sql.Number) >= 0; + } + + /// True when is a unique-key violation (2627/2601): "already present". + /// Exception from a SQL call. + /// Whether it is a duplicate-key error. + public static bool IsDuplicateKey(Exception ex) => + ex is SqlException sql && sql.Errors.Cast().Any(e => e.Number is 2627 or 2601); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianStatusCodes.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianStatusCodes.cs new file mode 100644 index 00000000..540667b5 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianStatusCodes.cs @@ -0,0 +1,44 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// Every OPC UA status this backend emits, as named constants (inline literals are invisible to +/// StatusCodeParityTests). Values mirror Opc.Ua.StatusCodes. +/// +internal static class SqlHistorianStatusCodes +{ + public const uint Good = 0x00000000u; + public const uint GoodNoData = 0x00A50000u; + public const uint GoodEntryInserted = 0x00A20000u; + public const uint GoodEntryReplaced = 0x00A30000u; + public const uint BadNoData = 0x809B0000u; + public const uint BadEntryExists = 0x809F0000u; + public const uint BadNoEntryExists = 0x80A00000u; + public const uint BadInvalidTimestamp = 0x80230000u; + public const uint BadTypeMismatch = 0x80740000u; + public const uint BadAggregateInvalidInputs = 0x80D60000u; + public const uint BadHistoryOperationUnsupported = 0x80720000u; + public const uint UncertainDataSubNormal = 0x40A40000u; +} + +/// Part 4 §7.39 historian info bits OR-ed onto a status code. +internal static class HistorianInfoBits +{ + /// InfoType = DataValue. + public const uint InfoTypeDataValue = 0x00000400u; + + /// The value was produced by an aggregate calculation. + public const uint HistorianCalculated = 0x1u; + + /// The value was interpolated. + public const uint HistorianInterpolated = 0x2u; + + /// Marks as a Calculated data value. + /// Base status code. + /// Status with InfoType and Calculated bits set. + public static uint WithCalculated(uint status) => status | InfoTypeDataValue | HistorianCalculated; + + /// Marks as an Interpolated data value. + /// Base status code. + /// Status with InfoType and Interpolated bits set. + public static uint WithInterpolated(uint status) => status | InfoTypeDataValue | HistorianInterpolated; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs new file mode 100644 index 00000000..ed23c993 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs @@ -0,0 +1,54 @@ +using System.Text.RegularExpressions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// Options for the SQL Server historian backend (the ServerHistorian:SqlServer config block). +/// +public sealed partial class SqlServerHistorianOptions +{ + /// Required. ADO.NET connection string for the historian database (never the ConfigDb). + public string ConnectionString { get; set; } = ""; + + /// SQL schema holding the historian objects. Spliced into DDL/DML, so it is regex-validated. + public string Schema { get; set; } = "hist"; + + /// Per-command timeout. Must be positive. + public TimeSpan CommandTimeout { get; set; } = TimeSpan.FromSeconds(30); + + /// + /// true creates the schema and applies missing scripts on first use; false only verifies + /// that schema version 1 exists and fails every historian call with a clear error otherwise. + /// + public bool AutoCreateSchema { get; set; } = true; + + [GeneratedRegex("^[A-Za-z_][A-Za-z0-9_]{0,127}$")] + private static partial Regex IdentifierRegex(); + + /// True when is a safe SQL identifier. + /// Candidate schema name. + /// Whether it matches ^[A-Za-z_][A-Za-z0-9_]{0,127}$. + public static bool IsValidSchemaName(string? schema) => schema is not null && IdentifierRegex().IsMatch(schema); + + /// Validates the options without touching SQL. + /// One message per problem; empty when valid. + public IReadOnlyList Validate() + { + var errors = new List(); + if (string.IsNullOrWhiteSpace(ConnectionString)) + errors.Add("ServerHistorian:SqlServer:ConnectionString is required."); + if (!IsValidSchemaName(Schema)) + errors.Add("ServerHistorian:SqlServer:Schema must be a SQL identifier matching ^[A-Za-z_][A-Za-z0-9_]{0,127}$."); + if (CommandTimeout <= TimeSpan.Zero) + errors.Add("ServerHistorian:SqlServer:CommandTimeout must be greater than zero."); + return errors; + } +} + +/// Thrown when AutoCreateSchema=false and the historian schema is absent or behind. +public sealed class HistorianSchemaMissingException : Exception +{ + /// Creates the exception. + /// Message naming the missing version. + public HistorianSchemaMissingException(string message) : base(message) { } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerSchemaBootstrapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerSchemaBootstrapper.cs new file mode 100644 index 00000000..c91efc30 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerSchemaBootstrapper.cs @@ -0,0 +1,155 @@ +using System.Data; +using System.Reflection; +using System.Text.RegularExpressions; +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// Lazily applies the embedded, idempotent schema scripts once per instance, under a SQL application lock +/// so concurrent nodes/instances serialise. A failure is NOT latched: the next retries. +/// With AutoCreateSchema=false it only verifies MAX(Version) >= 1. +/// +internal sealed partial class SqlServerSchemaBootstrapper +{ + /// The schema version this build requires. + public const int RequiredVersion = 1; + + private const string LockResource = "OtOpcUaHistorianSchema"; + private const int LockTimeoutMs = 30_000; + + private readonly SqlHistorianConnection _conn; + private readonly bool _autoCreate; + private readonly ILogger _log; + private readonly SemaphoreSlim _gate = new(1, 1); + private volatile bool _ready; + + /// Creates the bootstrapper. + /// Connection helper. + /// Validated options. + /// Optional logger. + public SqlServerSchemaBootstrapper(SqlHistorianConnection conn, SqlServerHistorianOptions options, ILogger? log = null) + { + _conn = conn; + _autoCreate = options.AutoCreateSchema; + _log = log ?? NullLogger.Instance; + } + + /// Ensures the schema is ready; no-op after the first success. + /// Cancellation. + /// A task completing when the schema is usable. + /// AutoCreateSchema is false and version 1 is absent. + public async Task EnsureAsync(CancellationToken ct) + { + if (_ready) return; + await _gate.WaitAsync(ct).ConfigureAwait(false); + try + { + if (_ready) return; + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + if (_autoCreate) await ApplyAsync(conn, ct).ConfigureAwait(false); + else await VerifyAsync(conn, ct).ConfigureAwait(false); + _ready = true; + } + finally + { + _gate.Release(); + } + } + + private async Task ReadVersionAsync(SqlConnection conn, CancellationToken ct) + { + await using var cmd = _conn.CreateCommand(conn, + $"IF OBJECT_ID(N'{_conn.Q("SchemaVersion")}', N'U') IS NULL SELECT 0 ELSE SELECT ISNULL(MAX(Version), 0) FROM {_conn.Q("SchemaVersion")}"); + return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)); + } + + private async Task VerifyAsync(SqlConnection conn, CancellationToken ct) + { + var v = await ReadVersionAsync(conn, ct).ConfigureAwait(false); + if (v < RequiredVersion) + throw new HistorianSchemaMissingException( + $"Historian schema [{_conn.Schema}] is missing version {RequiredVersion} (found {v}) and AutoCreateSchema is false. " + + "Apply docs/historian/sqlserver/001_initial.sql or enable AutoCreateSchema."); + } + + private async Task ApplyAsync(SqlConnection conn, CancellationToken ct) + { + await AcquireLockAsync(conn, ct).ConfigureAwait(false); + try + { + var current = await ReadVersionAsync(conn, ct).ConfigureAwait(false); + foreach (var (version, name, text) in LoadScripts()) + { + if (version <= current) continue; + foreach (var batch in SplitBatches(text.Replace("$(Schema)", _conn.Schema, StringComparison.Ordinal))) + { + await using var cmd = _conn.CreateCommand(conn, batch); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + _log.LogInformation("Applied historian schema script {Script} to [{Schema}]", name, _conn.Schema); + } + await VerifyAsync(conn, ct).ConfigureAwait(false); + } + finally + { + await ReleaseLockAsync(conn).ConfigureAwait(false); + } + } + + private async Task AcquireLockAsync(SqlConnection conn, CancellationToken ct) + { + await using var cmd = _conn.CreateCommand(conn, "sp_getapplock"); + cmd.CommandType = CommandType.StoredProcedure; + cmd.CommandTimeout = Math.Max(cmd.CommandTimeout, LockTimeoutMs / 1000 + 15); + cmd.Parameters.AddWithValue("@Resource", LockResource); + cmd.Parameters.AddWithValue("@LockMode", "Exclusive"); + cmd.Parameters.AddWithValue("@LockOwner", "Session"); + cmd.Parameters.AddWithValue("@LockTimeout", LockTimeoutMs); + var rc = new SqlParameter("@rc", SqlDbType.Int) { Direction = ParameterDirection.ReturnValue }; + cmd.Parameters.Add(rc); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + if ((int)rc.Value < 0) + throw new TimeoutException($"Could not acquire historian schema applock within {LockTimeoutMs} ms (sp_getapplock returned {rc.Value})."); + } + + private async Task ReleaseLockAsync(SqlConnection conn) + { + try + { + await using var cmd = _conn.CreateCommand(conn, "sp_releaseapplock"); + cmd.CommandType = CommandType.StoredProcedure; + cmd.Parameters.AddWithValue("@Resource", LockResource); + cmd.Parameters.AddWithValue("@LockOwner", "Session"); + await cmd.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + // Closing the session releases a Session-owned lock anyway. + _log.LogDebug(ex, "sp_releaseapplock failed; the lock dies with the connection"); + } + } + + [GeneratedRegex(@"^\s*GO\s*$", RegexOptions.Multiline | RegexOptions.IgnoreCase)] + private static partial Regex GoLine(); + + internal static IEnumerable SplitBatches(string script) => + GoLine().Split(script).Select(b => b.Trim()).Where(b => b.Length > 0); + + private static List<(int Version, string Name, string Text)> LoadScripts() + { + var asm = typeof(SqlServerSchemaBootstrapper).Assembly; + var list = new List<(int, string, string)>(); + foreach (var res in asm.GetManifestResourceNames()) + { + var m = Regex.Match(res, @"^Schema\.(\d+)_.*\.sql$"); + if (!m.Success) continue; + using var s = asm.GetManifestResourceStream(res)!; + using var r = new StreamReader(s); + list.Add((int.Parse(m.Groups[1].Value), res, r.ReadToEnd())); + } + return list.OrderBy(x => x.Item1).ToList(); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagCatalog.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagCatalog.cs new file mode 100644 index 00000000..6f62d86a --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagCatalog.cs @@ -0,0 +1,80 @@ +using System.Collections.Concurrent; +using Microsoft.Data.SqlClient; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// Tag name → TagId resolution shared by writes, reads, provisioning and updates. Caches positive +/// results only. Both methods run the schema bootstrap first, so callers need not. +/// +internal sealed class SqlServerTagCatalog +{ + private readonly SqlHistorianConnection _conn; + private readonly SqlServerSchemaBootstrapper _bootstrap; + private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); + + /// Creates the catalog. + /// Connection helper. + /// Schema bootstrapper. + public SqlServerTagCatalog(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap) + { + _conn = conn; + _bootstrap = bootstrap; + } + + /// Read-only lookup; never creates a row. + /// Historian tag name. + /// Cancellation. + /// The TagId, or null when the tag does not exist. + public async Task TryGetTagIdAsync(string name, CancellationToken ct) + { + if (_cache.TryGetValue(name, out var cached)) return cached; + await _bootstrap.EnsureAsync(ct).ConfigureAwait(false); + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + var id = await SelectIdAsync(conn, name, ct).ConfigureAwait(false); + if (id is { } v) _cache[name] = v; + return id; + } + + /// Returns the TagId, creating the row on a miss. Safe under concurrency (HOLDLOCK MERGE; 2627/2601 re-select). + /// Historian tag name. + /// Initial Tag.DataType (only used on create; provisioning corrects it later). + /// Cancellation. + /// The TagId. + public async Task GetOrCreateTagIdAsync(string name, byte dataType, CancellationToken ct) + { + if (_cache.TryGetValue(name, out var cached)) return cached; + await _bootstrap.EnsureAsync(ct).ConfigureAwait(false); + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + var sql = + $"MERGE {_conn.Q("Tag")} WITH (HOLDLOCK) AS t USING (SELECT @name AS Name) AS s ON t.Name = s.Name " + + "WHEN NOT MATCHED THEN INSERT (Name, DataType, CreatedUtc, UpdatedUtc) VALUES (@name, @dt, SYSUTCDATETIME(), SYSUTCDATETIME());"; + try + { + await using var cmd = _conn.CreateCommand(conn, sql); + cmd.Parameters.Add(new SqlParameter("@name", System.Data.SqlDbType.NVarChar, 400) { Value = name }); + cmd.Parameters.Add(new SqlParameter("@dt", System.Data.SqlDbType.TinyInt) { Value = dataType }); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + catch (SqlException ex) when (SqlHistorianConnection.IsDuplicateKey(ex)) + { + // A concurrent creator won; fall through to the re-select. + } + var id = await SelectIdAsync(conn, name, ct).ConfigureAwait(false) + ?? throw new InvalidOperationException($"Tag '{name}' vanished after create."); + _cache[name] = id; + return id; + } + + /// Drops a cached id (e.g. after a tag row is deleted). + /// Historian tag name. + public void Invalidate(string name) => _cache.TryRemove(name, out _); + + private async Task SelectIdAsync(SqlConnection conn, string name, CancellationToken ct) + { + await using var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")} WHERE Name = @name"); + cmd.Parameters.Add(new SqlParameter("@name", System.Data.SqlDbType.NVarChar, 400) { Value = name }); + var o = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false); + return o is null or DBNull ? null : Convert.ToInt32(o); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj new file mode 100644 index 00000000..ff5e29c5 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj @@ -0,0 +1,34 @@ + + + + true + true + $(NoWarn);CS1591 + AnyCPU;x64 + ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.csproj b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.csproj index c8ed6243..7c38ec29 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.csproj +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.csproj @@ -43,6 +43,7 @@ DriverTypeNames guard skips it — but it does hard-code OPC UA status constants, which puts it in StatusCodeParityTests' scope. --> + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SchemaBootstrapIntegrationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SchemaBootstrapIntegrationTests.cs new file mode 100644 index 00000000..b5eb4812 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SchemaBootstrapIntegrationTests.cs @@ -0,0 +1,139 @@ +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +[Collection(SqlHistorianCollection.Name)] +public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx) +{ + private SqlServerHistorianOptions Opts(string schema, bool auto = true) => new() + { + ConnectionString = fx.ConnectionString, Schema = schema, AutoCreateSchema = auto, + }; + + private static SqlServerSchemaBootstrapper Boot(SqlServerHistorianOptions o) => + new(new SqlHistorianConnection(o), o); + + private async Task ScalarAsync(string sql) + { + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = sql; + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } + + [Fact] + public async Task Bootstrapping_twice_is_a_no_op_with_one_version_row() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + await Boot(Opts(s)).EnsureAsync(CancellationToken.None); + await Boot(Opts(s)).EnsureAsync(CancellationToken.None); // fresh instance => re-runs the guarded path + (await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(1); + (await ScalarAsync($"SELECT COUNT(*) FROM sys.indexes WHERE name='CX_Event_Time' AND object_id=OBJECT_ID('[{s}].Event')")).ShouldBe(1); + } + + [Fact] + public async Task A_re_run_of_the_raw_script_is_idempotent() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + await Boot(Opts(s)).EnsureAsync(CancellationToken.None); + var script = new StreamReader(typeof(SqlServerSchemaBootstrapper).Assembly.GetManifestResourceStream("Schema.001_initial.sql")!).ReadToEnd(); + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + foreach (var b in SqlServerSchemaBootstrapper.SplitBatches(script.Replace("$(Schema)", s))) + { + await using var cmd = c.CreateCommand(); + cmd.CommandText = b; + await cmd.ExecuteNonQueryAsync(); + } + (await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(1); + } + + [Fact] + public async Task Concurrent_bootstrappers_both_succeed_with_one_version_row() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + var tasks = Enumerable.Range(0, 4).Select(_ => Task.Run(() => Boot(Opts(s)).EnsureAsync(CancellationToken.None))).ToArray(); + await Task.WhenAll(tasks); + (await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(1); + } + + [Fact] + public async Task AutoCreateSchema_false_on_an_empty_schema_throws_missing() + { + fx.SkipIfUnavailable(); + var ex = await Should.ThrowAsync( + () => Boot(Opts(SqlHistorianFixture.UniqueSchema(), auto: false)).EnsureAsync(CancellationToken.None)); + ex.Message.ShouldContain("missing version 1"); + } + + [Fact] + public async Task AutoCreateSchema_false_passes_once_the_schema_exists_and_failure_is_not_latched() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + var checker = Boot(Opts(s, auto: false)); + await Should.ThrowAsync(() => checker.EnsureAsync(CancellationToken.None)); + await Boot(Opts(s)).EnsureAsync(CancellationToken.None); + await checker.EnsureAsync(CancellationToken.None); // same instance retries + } + + [Fact] + public async Task Concurrent_get_or_create_of_one_new_tag_yields_one_row_and_one_id() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + var o = Opts(s); + var conn = new SqlHistorianConnection(o); + var boot = new SqlServerSchemaBootstrapper(conn, o); + await boot.EnsureAsync(CancellationToken.None); + // Separate catalogs => no shared cache, so every task really races the MERGE. Several rounds on fresh + // names, all callers released together, raise the odds of hitting the create window. + for (var round = 0; round < 10; round++) + { + var name = $"Plant/Tag{round}"; + using var start = new ManualResetEventSlim(false); + var ids = Enumerable.Range(0, 32).Select(_ => Task.Run(async () => + { + var cat = new SqlServerTagCatalog(conn, boot); + await cat.TryGetTagIdAsync("warm", CancellationToken.None); // opens the pool + bootstrap before the race + start.Wait(); + return await cat.GetOrCreateTagIdAsync(name, 1, CancellationToken.None); + })).ToArray(); + await Task.Delay(200); + start.Set(); + (await Task.WhenAll(ids)).Distinct().Count().ShouldBe(1); + (await ScalarAsync($"SELECT COUNT(*) FROM [{s}].Tag WHERE Name = N'{name}'")).ShouldBe(1); + } + (await new SqlServerTagCatalog(conn, boot).TryGetTagIdAsync("nope", CancellationToken.None)).ShouldBeNull(); + } + + [Fact] + public async Task Sample_tvp_keeps_sub_millisecond_ticks_end_to_end() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + var o = Opts(s); + var conn = new SqlHistorianConnection(o); + var boot = new SqlServerSchemaBootstrapper(conn, o); + var tagId = await new SqlServerTagCatalog(conn, boot).GetOrCreateTagIdAsync("T", 1, CancellationToken.None); + var t = new DateTime(2026, 10, 9, 0, 0, 0, DateTimeKind.Utc).AddTicks(1234567); + await using var c = await conn.OpenAsync(CancellationToken.None); + await using var cmd = conn.CreateCommand(c, + $"INSERT {conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " + + "SELECT @id, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, SYSUTCDATETIME() FROM @rows"); + cmd.Parameters.AddWithValue("@id", tagId); + conn.AddStructured(cmd, "@rows", "SampleRow", + SampleRowMapper.ToSampleRecords([new SqlSampleRow(t, ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistorianValue.FromDouble(1), 0u)])); + await cmd.ExecuteNonQueryAsync(); + await using var q = conn.CreateCommand(c, $"SELECT SourceTimeUtc FROM {conn.Q("Sample")}"); + await using var r = await q.ExecuteReaderAsync(); + (await r.ReadAsync()).ShouldBeTrue(); + SqlHistorianConnection.GetUtc(r, 0).Ticks.ShouldBe(t.Ticks); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlHistorianFixture.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlHistorianFixture.cs new file mode 100644 index 00000000..4610341d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlHistorianFixture.cs @@ -0,0 +1,79 @@ +using Microsoft.Data.SqlClient; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +/// +/// Env-gated throwaway database for the SQL historian backend. HISTORIAN_SQL_TEST_CONNECTION absent ⇒ +/// is set, nothing connects, and every test calls Assert.Skip. Present ⇒ +/// creates OtOpcUaHistorian_T_<guid> (never touches another database) and drops it on dispose. +/// Tests isolate from each other with . +/// +public sealed class SqlHistorianFixture : IAsyncLifetime +{ + /// Env var holding a connection string with CREATE/DROP DATABASE rights. + public const string ConnectionEnvVar = "HISTORIAN_SQL_TEST_CONNECTION"; + + private string? _master; + private string? _dbName; + + /// Why tests must skip, or null when the database is available. + public string? SkipReason { get; private set; } + + /// Connection string targeting the throwaway database. Valid only when is null. + public string ConnectionString { get; private set; } = ""; + + /// Skips the calling test when the gate is closed. + public void SkipIfUnavailable() + { + if (SkipReason is not null) Assert.Skip(SkipReason); + } + + /// A fresh, valid schema name (nothing exists under it yet). + /// A schema identifier unique to this call. + public static string UniqueSchema() => "h" + Guid.NewGuid().ToString("N")[..16]; + + /// + public async ValueTask InitializeAsync() + { + var cs = Environment.GetEnvironmentVariable(ConnectionEnvVar); + if (string.IsNullOrWhiteSpace(cs)) + { + SkipReason = $"{ConnectionEnvVar} is not set; SQL Server historian integration tests skipped."; + return; + } + + var b = new SqlConnectionStringBuilder(cs) { InitialCatalog = "master" }; + _master = b.ConnectionString; + _dbName = "OtOpcUaHistorian_T_" + Guid.NewGuid().ToString("N"); + await ExecAsync(_master, $"CREATE DATABASE [{_dbName}]").ConfigureAwait(false); + b.InitialCatalog = _dbName; + ConnectionString = b.ConnectionString; + } + + /// + public async ValueTask DisposeAsync() + { + if (_master is null || _dbName is null) return; + SqlConnection.ClearAllPools(); + await ExecAsync(_master, + $"ALTER DATABASE [{_dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; DROP DATABASE [{_dbName}]").ConfigureAwait(false); + } + + private static async Task ExecAsync(string cs, string sql) + { + await using var c = new SqlConnection(cs); + await c.OpenAsync().ConfigureAwait(false); + await using var cmd = c.CreateCommand(); + cmd.CommandText = sql; + await cmd.ExecuteNonQueryAsync().ConfigureAwait(false); + } +} + +/// Shares one throwaway database across the integration tests. +[CollectionDefinition(Name)] +public sealed class SqlHistorianCollection : ICollectionFixture +{ + /// Collection name. + public const string Name = "SqlHistorian"; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj new file mode 100644 index 00000000..74dbd900 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj @@ -0,0 +1,26 @@ + + + + net10.0 + enable + enable + false + true + ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SampleRowMapperTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SampleRowMapperTests.cs new file mode 100644 index 00000000..c71aefe5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SampleRowMapperTests.cs @@ -0,0 +1,65 @@ +using Microsoft.Data.SqlClient.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +public sealed class SampleRowMapperTests +{ + public static TheoryData Values => new() + { + HistorianValue.FromDouble(12.5), + HistorianValue.FromInt64(long.MinValue + 3), + HistorianValue.FromBoolean(true), + HistorianValue.FromBoolean(false), + HistorianValue.FromString("héllo"), + HistorianValue.FromString(""), + HistorianValue.FromDateTime(new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1234567)), + HistorianValue.Empty, + }; + + [Theory] + [MemberData(nameof(Values))] + public void Every_kind_round_trips_through_the_column_quad(HistorianValue v) + { + var c = SampleRowMapper.ToColumns(v); + SampleRowMapper.FromColumns(c.ValueKind, c.NumValue, c.IntValue, c.StrValue).ShouldBe(v); + } + + [Fact] + public void Columns_follow_the_spec_mapping() + { + SampleRowMapper.ToColumns(HistorianValue.FromInt64(7)).ShouldBe(new SampleColumns(2, 7d, 7L, null)); + SampleRowMapper.ToColumns(HistorianValue.FromBoolean(true)).ShouldBe(new SampleColumns(3, 1d, 1L, null)); + SampleRowMapper.ToColumns(HistorianValue.FromDouble(1.5)).ShouldBe(new SampleColumns(1, 1.5, null, null)); + SampleRowMapper.ToColumns(HistorianValue.Empty).ShouldBe(new SampleColumns(0, null, null, null)); + } + + [Fact] + public void Status_is_stored_bit_for_bit() + { + SampleRowMapper.ToDbStatus(0x80050000u).ShouldBeLessThan(0); + SampleRowMapper.FromDbStatus(SampleRowMapper.ToDbStatus(0x80050000u)).ShouldBe(0x80050000u); + } + + [Fact] + public void Sample_records_keep_sub_millisecond_ticks() + { + var t = new DateTime(2026, 10, 9, 0, 0, 0, DateTimeKind.Utc).AddTicks(1234567); + var rec = SampleRowMapper.ToSampleRecords([new SqlSampleRow(t, HistorianValue.FromDouble(2), 0x80050000u)]).Single(); + rec.GetDateTime(0).Ticks.ShouldBe(t.Ticks); + rec.GetByte(1).ShouldBe((byte)1); + rec.GetDouble(2).ShouldBe(2d); + rec.IsDBNull(3).ShouldBeTrue(); + rec.IsDBNull(4).ShouldBeTrue(); + rec.GetInt32(5).ShouldBe(unchecked((int)0x80050000u)); + } + + [Fact] + public void Time_records_keep_sub_millisecond_ticks() + { + var t = new DateTime(2026, 10, 9, 0, 0, 0, DateTimeKind.Utc).AddTicks(1234567); + SampleRowMapper.ToTimeRecords([t]).Single().GetDateTime(0).Ticks.ShouldBe(t.Ticks); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SchemaScriptTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SchemaScriptTests.cs new file mode 100644 index 00000000..649dd550 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SchemaScriptTests.cs @@ -0,0 +1,32 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +public sealed class SchemaScriptTests +{ + [Fact] + public void Go_splitting_ignores_go_inside_other_text_and_drops_empties() + { + var parts = SqlServerSchemaBootstrapper.SplitBatches("A\nGO\n go \nB GO C\nGO\n").ToArray(); + parts.ShouldBe(["A", "B GO C"]); + } + + [Fact] + public void Embedded_script_is_present_and_matches_the_docs_copy() + { + var asm = typeof(SqlServerSchemaBootstrapper).Assembly; + using var s = asm.GetManifestResourceStream("Schema.001_initial.sql"); + s.ShouldNotBeNull(); + var text = new StreamReader(s!).ReadToEnd(); + text.ShouldContain("$(Schema)"); + text.ShouldContain("PRIMARY KEY NONCLUSTERED"); + text.ShouldContain("CREATE CLUSTERED INDEX CX_Event_Time"); + text.ShouldNotContain("IX_Event_Time"); + + var dir = new DirectoryInfo(AppContext.BaseDirectory); + while (dir is not null && !File.Exists(Path.Combine(dir.FullName, "ZB.MOM.WW.OtOpcUa.slnx"))) dir = dir.Parent; + dir.ShouldNotBeNull(); + File.ReadAllText(Path.Combine(dir!.FullName, "docs/historian/sqlserver/001_initial.sql")).ShouldBe(text); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianOptionsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianOptionsTests.cs new file mode 100644 index 00000000..a21800ac --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianOptionsTests.cs @@ -0,0 +1,46 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +public sealed class SqlServerHistorianOptionsTests +{ + private static SqlServerHistorianOptions Valid() => new() { ConnectionString = "Server=x" }; + + [Fact] + public void Defaults_with_a_connection_string_are_valid() => Valid().Validate().ShouldBeEmpty(); + + [Fact] + public void Missing_connection_string_is_rejected() => + new SqlServerHistorianOptions().Validate().ShouldContain(e => e.Contains("ConnectionString")); + + [Theory] + [InlineData("hist]; DROP TABLE x;--")] + [InlineData("1abc")] + [InlineData("a b")] + [InlineData("")] + [InlineData("a.b")] + public void Invalid_schema_names_are_rejected_before_any_sql_runs(string schema) + { + var o = Valid(); + o.Schema = schema; + o.Validate().ShouldContain(e => e.Contains("Schema")); + Should.Throw(() => new SqlHistorianConnection(o)); + } + + [Fact] + public void Non_positive_timeout_is_rejected() + { + var o = Valid(); + o.CommandTimeout = TimeSpan.Zero; + o.Validate().ShouldContain(e => e.Contains("CommandTimeout")); + } + + [Fact] + public void Q_schema_qualifies_and_brackets() + { + var o = Valid(); + o.Schema = "my_s"; + new SqlHistorianConnection(o).Q("Tag").ShouldBe("[my_s].[Tag]"); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/StatusCodeOracleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/StatusCodeOracleTests.cs new file mode 100644 index 00000000..bfab0cab --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/StatusCodeOracleTests.cs @@ -0,0 +1,28 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +/// The pinned SDK is the oracle for the info-bit constants (StatusCodeParityTests covers the named codes). +public sealed class StatusCodeOracleTests +{ + [Fact] + public void Calculated_bits_match_the_SDK() + { + var sdk = new Opc.Ua.StatusCode(Opc.Ua.StatusCodes.Good).SetAggregateBits(Opc.Ua.AggregateBits.Calculated); + sdk.Code.ShouldBe(HistorianInfoBits.WithCalculated(SqlHistorianStatusCodes.Good)); + ((uint)Opc.Ua.AggregateBits.Calculated).ShouldBe(HistorianInfoBits.HistorianCalculated); + } + + [Fact] + public void Interpolated_bits_match_the_SDK() + { + var sdk = new Opc.Ua.StatusCode(Opc.Ua.StatusCodes.Good).SetAggregateBits(Opc.Ua.AggregateBits.Interpolated); + sdk.Code.ShouldBe(HistorianInfoBits.WithInterpolated(SqlHistorianStatusCodes.Good)); + ((uint)Opc.Ua.AggregateBits.Interpolated).ShouldBe(HistorianInfoBits.HistorianInterpolated); + } + + [Fact] + public void Info_type_is_data_value() => + HistorianInfoBits.InfoTypeDataValue.ShouldBe(0x400u); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests.csproj new file mode 100644 index 00000000..dbcbb595 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests.csproj @@ -0,0 +1,28 @@ + + + + net10.0 + enable + enable + false + true + ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests + + + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + +