Merge ph-T8 (pluggable historian)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 13:00:51 -04:00
parent 1447a50140 17962d4233
commit 65e6352dad
19 changed files with 1187 additions and 0 deletions
@@ -43,6 +43,7 @@
             DriverTypeNames guard skips it — but it does hard-code OPC UA status constants, which
             puts it in StatusCodeParityTests' scope. -->
        <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
+        <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/>
    </ItemGroup>

 </Project>
@@ -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<long> 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<HistorianSchemaMissingException>(
+            () => 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<HistorianSchemaMissingException>(() => 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);
+    }
+}
@@ -0,0 +1,79 @@
+using Microsoft.Data.SqlClient;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
+
+/// <summary>
+///     Env-gated throwaway database for the SQL historian backend. <c>HISTORIAN_SQL_TEST_CONNECTION</c> absent ⇒
+///     <see cref="SkipReason"/> is set, nothing connects, and every test calls <c>Assert.Skip</c>. Present ⇒
+///     creates <c>OtOpcUaHistorian_T_&lt;guid&gt;</c> (never touches another database) and drops it on dispose.
+///     Tests isolate from each other with <see cref="UniqueSchema"/>.
+/// </summary>
+public sealed class SqlHistorianFixture : IAsyncLifetime
+{
+    /// <summary>Env var holding a connection string with CREATE/DROP DATABASE rights.</summary>
+    public const string ConnectionEnvVar = "HISTORIAN_SQL_TEST_CONNECTION";
+
+    private string? _master;
+    private string? _dbName;
+
+    /// <summary>Why tests must skip, or null when the database is available.</summary>
+    public string? SkipReason { get; private set; }
+
+    /// <summary>Connection string targeting the throwaway database. Valid only when <see cref="SkipReason"/> is null.</summary>
+    public string ConnectionString { get; private set; } = "";
+
+    /// <summary>Skips the calling test when the gate is closed.</summary>
+    public void SkipIfUnavailable()
+    {
+        if (SkipReason is not null) Assert.Skip(SkipReason);
+    }
+
+    /// <summary>A fresh, valid schema name (nothing exists under it yet).</summary>
+    /// <returns>A schema identifier unique to this call.</returns>
+    public static string UniqueSchema() => "h" + Guid.NewGuid().ToString("N")[..16];
+
+    /// <inheritdoc/>
+    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;
+    }
+
+    /// <inheritdoc/>
+    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);
+    }
+}
+
+/// <summary>Shares one throwaway database across the integration tests.</summary>
+[CollectionDefinition(Name)]
+public sealed class SqlHistorianCollection : ICollectionFixture<SqlHistorianFixture>
+{
+    /// <summary>Collection name.</summary>
+    public const string Name = "SqlHistorian";
+}
@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net10.0</TargetFramework>
+        <Nullable>enable</Nullable>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <IsPackable>false</IsPackable>
+        <IsTestProject>true</IsTestProject>
+        <RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests</RootNamespace>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="xunit.v3"/>
+        <PackageReference Include="Shouldly"/>
+        <PackageReference Include="Microsoft.NET.Test.Sdk"/>
+        <PackageReference Include="xunit.runner.visualstudio">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageReference>
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/>
+    </ItemGroup>
+
+</Project>
@@ -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<HistorianValue> 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);
+    }
+}
@@ -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);
+    }
+}
@@ -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<ArgumentException>(() => 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]");
+    }
+}
@@ -0,0 +1,28 @@
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests;
+
+/// <summary>The pinned SDK is the oracle for the info-bit constants (StatusCodeParityTests covers the named codes).</summary>
+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);
+}
@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net10.0</TargetFramework>
+        <Nullable>enable</Nullable>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <IsPackable>false</IsPackable>
+        <IsTestProject>true</IsTestProject>
+        <RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests</RootNamespace>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="xunit.v3"/>
+        <PackageReference Include="Shouldly"/>
+        <PackageReference Include="Microsoft.NET.Test.Sdk"/>
+        <!-- SDK oracle for the status-code / aggregate-bit tests. -->
+        <PackageReference Include="OPCFoundation.NetStandard.Opc.Ua.Core"/>
+        <PackageReference Include="xunit.runner.visualstudio">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageReference>
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/>
+    </ItemGroup>
+
+</Project>