Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs
T
Joseph Doherty 6669fae7e5 feat(historian): ServerHistorian:MaxRawValuesPerRead caps raw pages and backend reads
Review R08:1.1: an uncapped HistoryRead-Raw (NumValuesPerNode = 0) read a
tag's whole window into memory on every backend (SQL TOP(int.MaxValue),
QuestDB LIMIT, Gateway max_values = int.MaxValue + a second snapshot copy).

New ServerHistorian:MaxRawValuesPerRead (default 10 000; <= 0 unbounded with
a Validate() warning; 1 raised to 2). Each backend data source takes it,
reports it as IHistorianDataSource.MaxRawValuesPerRead (the Gateway: min
with a strict gateway's learned ceiling), so the node manager's existing
RawPageLimit paging serves N = 0 and N > cap as pages of at most the cap
with a continuation point. Independently, a backend raw read asking for 0
or more than the cap reads cap + 1 rows and is REFUSED
(HistoryReadLimitExceededException => BadTooManyOperations) when the window
holds more - never answered truncated, since the server's paging reads a
short answer as the last page. Bound through the Host registry, the
SqlServer/QuestDb providers and GatewayHistorian.CreateDataSource.

Tests: options (default/binding/warnings), provider binding, Gateway unit +
through a real SDK node manager (uncapped -> 999/998/998/605 + CPs), and a
conformance case for the SQL backends.

Refs #736

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:29:10 -04:00

90 lines
4.5 KiB
C#

using Microsoft.Data.SqlClient;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
/// <summary>
/// Seeds a fresh schema with plain INSERTs (independent of the value writer) and builds a data source over it.
/// Shared by the SQL Server-specific tests and by <see cref="SqlServerConformanceBackend"/>. <see cref="T0"/> is deliberately NOT millisecond-aligned.
/// </summary>
internal sealed class ReadTestDb
{
/// <summary>Base time: 2026-10-09 08:00:00.1234567 UTC (sub-millisecond ticks on purpose).</summary>
public static readonly DateTime T0 = Conformance.ConformanceData.T0;
public const uint Good = Conformance.ConformanceData.Good;
public const uint Bad = Conformance.ConformanceData.Bad;
public const uint BadNoCommunication = Conformance.ConformanceData.BadNoCommunication;
public const uint Uncertain = Conformance.ConformanceData.Uncertain;
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) => Conformance.ConformanceData.At(seconds);
public static async Task<ReadTestDb> 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);
}
/// <summary>A fresh data source (fresh catalog cache) over this schema; <paramref name="clock"/> is its "now".</summary>
public SqlServerHistorianDataSource Source(TimeProvider? clock = null,
int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, int maxRawValuesPerRead = 0) =>
new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock, maxTimeWeightedRawRows: maxTimeWeightedRawRows,
maxRawValuesPerRead: maxRawValuesPerRead);
/// <summary>Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart).</summary>
public async Task<int> 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 = SampleColumnMapper.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", SampleColumnMapper.ToDbStatus(status));
SqlHistorianConnection.AddDateTime2(cmd, "@srv", t.AddSeconds(1));
await cmd.ExecuteNonQueryAsync();
}
return id;
}
public async Task<long> 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());
}
}