Merge qdb/T3: QuestDB historian skeleton
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -27,6 +27,7 @@
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browser/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browser.csproj" />
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Contracts/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Contracts.csproj" />
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj" />
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.csproj" />
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj" />
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj" />
|
||||
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ZB.MOM.WW.OtOpcUa.Driver.Modbus.csproj" />
|
||||
@@ -102,6 +103,8 @@
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browser.Tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browser.Tests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests.csproj" />
|
||||
<Project Path="tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests.csproj" />
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
using System.Net.Sockets;
|
||||
using Npgsql;
|
||||
using NpgsqlTypes;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// Pooled Npgsql access to QuestDB plus every T0-confirmed typing rule, in one place: how timestamps are rounded and
|
||||
/// bound (R6), how BYTE columns are bound and read, and how errors are classified.
|
||||
/// <para>Contract for callers: bind every inbound <see cref="DateTime"/> through <see cref="BindTimestamp"/> (never
|
||||
/// <c>AddWithValue</c>), call <see cref="IsStorable"/> before writing a designated timestamp, and classify
|
||||
/// failures with <see cref="IsTransient"/> / <see cref="IsAlreadyExists"/>. Every QuestDB server error carries
|
||||
/// SQLSTATE <c>00000</c>, so classification is by message text, never by SQLSTATE.</para>
|
||||
/// </summary>
|
||||
internal sealed class QuestDbConnection : IAsyncDisposable
|
||||
{
|
||||
private static readonly long MinSafeTicks = 10; // DateTime.MinValue itself binds as -infinity
|
||||
private static readonly long MaxSafeTicks = DateTime.MaxValue.Ticks - DateTime.MaxValue.Ticks % 10; // MaxValue itself binds as +infinity
|
||||
|
||||
private readonly NpgsqlDataSource _dataSource;
|
||||
|
||||
/// <summary>Creates the connection helper; options must already be valid.</summary>
|
||||
/// <param name="options">Validated options.</param>
|
||||
public QuestDbConnection(QuestDbHistorianOptions options)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
if (!QuestDbHistorianOptions.IsValidPrefix(options.TablePrefix))
|
||||
throw new ArgumentException("Invalid table prefix.", nameof(options));
|
||||
TablePrefix = options.TablePrefix;
|
||||
CommandTimeoutSeconds = Math.Max(1, (int)Math.Ceiling(options.CommandTimeout.TotalSeconds));
|
||||
_dataSource = new NpgsqlDataSourceBuilder(options.ConnectionString).Build();
|
||||
}
|
||||
|
||||
/// <summary>The validated table prefix.</summary>
|
||||
public string TablePrefix { get; }
|
||||
|
||||
/// <summary>Command timeout applied by <see cref="CreateCommand"/>, in seconds.</summary>
|
||||
public int CommandTimeoutSeconds { get; }
|
||||
|
||||
/// <summary>Prefixes a table base name (<c>sample</c>, <c>event</c>, <c>tag</c>) with the configured prefix.</summary>
|
||||
/// <param name="table">Fixed base name from this project (never user input).</param>
|
||||
/// <returns>The full table name.</returns>
|
||||
public string T(string table) => TablePrefix + table;
|
||||
|
||||
/// <summary>The prefixed sample table.</summary>
|
||||
public string SampleTable => T(QuestDbTables.SampleBase);
|
||||
/// <summary>The prefixed event table.</summary>
|
||||
public string EventTable => T(QuestDbTables.EventBase);
|
||||
/// <summary>The prefixed tag table.</summary>
|
||||
public string TagTable => T(QuestDbTables.TagBase);
|
||||
|
||||
/// <summary>Opens a pooled connection (no bootstrap; callers run the bootstrapper first).</summary>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>An open connection the caller disposes.</returns>
|
||||
public async Task<NpgsqlConnection> OpenAsync(CancellationToken ct) =>
|
||||
await _dataSource.OpenConnectionAsync(ct).ConfigureAwait(false);
|
||||
|
||||
/// <summary>Creates a text command with the configured timeout. Parameters are positional (<c>$1</c>, <c>$2</c>…).</summary>
|
||||
/// <param name="conn">Open connection.</param>
|
||||
/// <param name="sql">Command text.</param>
|
||||
/// <returns>The command.</returns>
|
||||
public NpgsqlCommand CreateCommand(NpgsqlConnection conn, string sql)
|
||||
{
|
||||
var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = sql;
|
||||
cmd.CommandTimeout = CommandTimeoutSeconds;
|
||||
return cmd;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public ValueTask DisposeAsync() => _dataSource.DisposeAsync();
|
||||
|
||||
// ---- timestamps (R6) -------------------------------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// Normalises a timestamp for QuestDB (microsecond precision). Rule: <c>Unspecified</c> is treated as UTC
|
||||
/// (<c>SpecifyKind</c>); <c>Local</c> is converted with <c>ToUniversalTime</c>; then ticks are truncated to a
|
||||
/// microsecond (<c>ticks - ticks % 10</c>, because Npgsql truncates relative to 2000-01-01 and so rounds UP
|
||||
/// before 2000); finally the result is nudged into <c>[0001-01-01T00:00:00.000001, 9999-12-31T23:59:59.999999]</c>
|
||||
/// so a value equal to <c>MinValue</c>/<c>MaxValue</c> never reaches Npgsql, which would bind it as ±infinity
|
||||
/// (a <c>[Min, Max)</c> range then silently matches nothing). The result always has <c>Kind=Utc</c>.
|
||||
/// </summary>
|
||||
/// <param name="value">Any timestamp.</param>
|
||||
/// <returns>A UTC, microsecond-aligned, finite timestamp.</returns>
|
||||
public static DateTime Round(DateTime value)
|
||||
{
|
||||
var utc = value.Kind switch
|
||||
{
|
||||
DateTimeKind.Local => value.ToUniversalTime(),
|
||||
DateTimeKind.Utc => value,
|
||||
_ => DateTime.SpecifyKind(value, DateTimeKind.Utc),
|
||||
};
|
||||
var ticks = utc.Ticks - utc.Ticks % 10;
|
||||
ticks = Math.Clamp(ticks, MinSafeTicks, MaxSafeTicks);
|
||||
return new DateTime(ticks, DateTimeKind.Utc);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// True when QuestDB can store the timestamp as a designated timestamp: after <see cref="Round"/> it is on or
|
||||
/// after 1970-01-01 (RC1). Writers drop (live) or refuse (HistoryUpdate) samples for which this is false; reads
|
||||
/// with earlier bounds are fine.
|
||||
/// </summary>
|
||||
/// <param name="value">Candidate sample time.</param>
|
||||
/// <returns>Whether an INSERT with this designated timestamp can succeed.</returns>
|
||||
public static bool IsStorable(DateTime value) => Round(value) >= DateTime.UnixEpoch;
|
||||
|
||||
/// <summary>
|
||||
/// Appends the next positional parameter (<c>$n</c>) as <c>timestamp with time zone</c> carrying
|
||||
/// <see cref="Round"/>'s UTC value (T0-confirmed: <c>Timestamp</c>+<c>Utc</c> throws in Npgsql). It never emits
|
||||
/// <c>MinValue</c>/<c>MaxValue</c>. By default a pre-1970 value is kept as is (a pre-1970 range bound is valid
|
||||
/// and clamping it would move the bound across the epoch row); pass <paramref name="clampToEpoch"/> for a
|
||||
/// designated-timestamp column so a pre-1970 value can only ever become the epoch itself, never a failed batch.
|
||||
/// </summary>
|
||||
/// <param name="cmd">Target command.</param>
|
||||
/// <param name="value">Any timestamp.</param>
|
||||
/// <param name="clampToEpoch">Clamp anything before 1970-01-01 to the epoch (writes only, after <see cref="IsStorable"/>).</param>
|
||||
/// <returns>The added parameter.</returns>
|
||||
public static NpgsqlParameter BindTimestamp(NpgsqlCommand cmd, DateTime value, bool clampToEpoch = false)
|
||||
{
|
||||
var rounded = Round(value);
|
||||
if (clampToEpoch && rounded < DateTime.UnixEpoch) rounded = DateTime.UnixEpoch;
|
||||
var p = new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.TimestampTz, Value = rounded };
|
||||
cmd.Parameters.Add(p);
|
||||
return p;
|
||||
}
|
||||
|
||||
/// <summary>Appends a <c>BYTE</c> parameter (bound as <c>smallint</c>, T0-confirmed).</summary>
|
||||
/// <param name="cmd">Target command.</param>
|
||||
/// <param name="value">The byte.</param>
|
||||
/// <returns>The added parameter.</returns>
|
||||
public static NpgsqlParameter BindByte(NpgsqlCommand cmd, byte value)
|
||||
{
|
||||
var p = new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Smallint, Value = (short)value };
|
||||
cmd.Parameters.Add(p);
|
||||
return p;
|
||||
}
|
||||
|
||||
/// <summary>Appends a <c>LONG</c> parameter (<c>bigint</c>; also the rule for integer arithmetic in tick math).</summary>
|
||||
/// <param name="cmd">Target command.</param>
|
||||
/// <param name="value">The value.</param>
|
||||
/// <returns>The added parameter.</returns>
|
||||
public static NpgsqlParameter BindLong(NpgsqlCommand cmd, long value)
|
||||
{
|
||||
var p = new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Bigint, Value = value };
|
||||
cmd.Parameters.Add(p);
|
||||
return p;
|
||||
}
|
||||
|
||||
/// <summary>Appends a <c>SYMBOL</c>/<c>VARCHAR</c> parameter (<c>varchar</c>); null binds as NULL.</summary>
|
||||
/// <param name="cmd">Target command.</param>
|
||||
/// <param name="value">The string.</param>
|
||||
/// <returns>The added parameter.</returns>
|
||||
public static NpgsqlParameter BindText(NpgsqlCommand cmd, string? value)
|
||||
{
|
||||
var p = new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Varchar, Value = (object?)value ?? DBNull.Value };
|
||||
cmd.Parameters.Add(p);
|
||||
return p;
|
||||
}
|
||||
|
||||
/// <summary>Reads a <c>TIMESTAMP</c> column as UTC (QuestDB returns <c>Kind=Unspecified</c> holding UTC wall clock).</summary>
|
||||
/// <param name="r">Reader positioned on a row.</param>
|
||||
/// <param name="i">Column ordinal.</param>
|
||||
/// <returns>The timestamp with <c>Kind=Utc</c>.</returns>
|
||||
public static DateTime ReadUtc(NpgsqlDataReader r, int i) =>
|
||||
DateTime.SpecifyKind(r.GetDateTime(i), DateTimeKind.Utc);
|
||||
|
||||
/// <summary>Reads a <c>BYTE</c> column (QuestDB returns <c>smallint</c>); NULL (never stored) reads as 0 (Empty).</summary>
|
||||
/// <param name="r">Reader positioned on a row.</param>
|
||||
/// <param name="i">Column ordinal.</param>
|
||||
/// <returns>The byte value.</returns>
|
||||
public static byte ReadKind(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? (byte)0 : (byte)r.GetInt16(i);
|
||||
|
||||
// ---- error classification --------------------------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// True for a transport-level or timeout failure worth retrying. A <see cref="PostgresException"/> is a
|
||||
/// server-side rejection (QuestDB reports every one as SQLSTATE 00000), so it is never transient here; decide
|
||||
/// those by message text.
|
||||
/// </summary>
|
||||
/// <param name="ex">The failure.</param>
|
||||
/// <returns>Whether the failure is transient.</returns>
|
||||
public static bool IsTransient(Exception ex)
|
||||
{
|
||||
for (var e = ex; e is not null; e = e.InnerException)
|
||||
{
|
||||
if (e is PostgresException or OperationCanceledException) return false;
|
||||
if (e is NpgsqlException { IsTransient: true } or SocketException or IOException or TimeoutException) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>True when a server error says the object already exists (message text; SQLSTATE is always 00000).</summary>
|
||||
/// <param name="ex">The failure.</param>
|
||||
/// <returns>Whether it is an "already exists" rejection.</returns>
|
||||
public static bool IsAlreadyExists(Exception ex) =>
|
||||
ex is PostgresException pg && pg.MessageText.Contains("already exists", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>Whether a table exists, per <c>tables()</c>.</summary>
|
||||
/// <param name="conn">Open connection.</param>
|
||||
/// <param name="table">Full table name.</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>True when listed.</returns>
|
||||
public async Task<bool> TableExistsAsync(NpgsqlConnection conn, string table, CancellationToken ct)
|
||||
{
|
||||
await using var cmd = CreateCommand(conn, "SELECT count() FROM tables() WHERE table_name = $1");
|
||||
BindText(cmd, table);
|
||||
return Convert.ToInt64(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) > 0;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
using System.Text.RegularExpressions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>Options for the QuestDB historian backend (the <c>ServerHistorian:QuestDb</c> config block).</summary>
|
||||
public sealed partial class QuestDbHistorianOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Required. An Npgsql connection string for QuestDB's PG-wire port, e.g.
|
||||
/// <c>Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading</c>.
|
||||
/// </summary>
|
||||
public string ConnectionString { get; set; } = "";
|
||||
|
||||
/// <summary>Prefix of the three tables (QuestDB has no schemas). Spliced into DDL/DML, so it is regex-validated.</summary>
|
||||
public string TablePrefix { get; set; } = "hist_";
|
||||
|
||||
/// <summary>Retention in days applied as a TTL to the sample and event tables; <c>null</c> keeps data forever.</summary>
|
||||
public int? RetentionDays { get; set; }
|
||||
|
||||
/// <summary>Per-command timeout. Must be positive.</summary>
|
||||
public TimeSpan CommandTimeout { get; set; } = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>
|
||||
/// <c>true</c> creates the tables (and applies the TTL) on first use; <c>false</c> only verifies they exist and
|
||||
/// fails every historian call with <c>HistorianSchemaMissingException</c> otherwise.
|
||||
/// </summary>
|
||||
public bool AutoCreateSchema { get; set; } = true;
|
||||
|
||||
/// <summary>How long a write waits for its rows to become visible to queries (WAL apply). 0 disables the wait.</summary>
|
||||
public TimeSpan VisibilityTimeout { get; set; } = TimeSpan.FromSeconds(5);
|
||||
|
||||
[GeneratedRegex("^[A-Za-z0-9_]+$")]
|
||||
private static partial Regex PrefixRegex();
|
||||
|
||||
/// <summary>True when <paramref name="prefix"/> is safe to splice into a table name.</summary>
|
||||
/// <param name="prefix">Candidate prefix.</param>
|
||||
/// <returns>Whether it matches <c>^[A-Za-z0-9_]+$</c>.</returns>
|
||||
public static bool IsValidPrefix(string? prefix) => prefix is not null && PrefixRegex().IsMatch(prefix);
|
||||
|
||||
/// <summary>Validates the options without touching QuestDB.</summary>
|
||||
/// <returns>One message per problem (each names its <c>ServerHistorian:QuestDb:<Key></c>); empty when valid.</returns>
|
||||
public IReadOnlyList<string> Validate()
|
||||
{
|
||||
var errors = new List<string>();
|
||||
if (string.IsNullOrWhiteSpace(ConnectionString))
|
||||
errors.Add("ServerHistorian:QuestDb:ConnectionString is required.");
|
||||
if (!IsValidPrefix(TablePrefix))
|
||||
errors.Add("ServerHistorian:QuestDb:TablePrefix must match ^[A-Za-z0-9_]+$.");
|
||||
if (RetentionDays is < 1)
|
||||
errors.Add("ServerHistorian:QuestDb:RetentionDays must be at least 1 when set.");
|
||||
if (CommandTimeout <= TimeSpan.Zero)
|
||||
errors.Add("ServerHistorian:QuestDb:CommandTimeout must be greater than zero.");
|
||||
if (VisibilityTimeout < TimeSpan.Zero)
|
||||
errors.Add("ServerHistorian:QuestDb:VisibilityTimeout must not be negative (0 disables the wait).");
|
||||
return errors;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Npgsql;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// Lazily creates the three tables once per instance (<c>CREATE TABLE IF NOT EXISTS</c>), then reconciles the TTL
|
||||
/// of the sample and event tables with <see cref="QuestDbHistorianOptions.RetentionDays"/>.
|
||||
/// <list type="bullet">
|
||||
/// <item><b>Concurrency:</b> two nodes starting together are safe. If a CREATE still fails with a server error
|
||||
/// whose message says <c>already exists</c>, <c>tables()</c> is re-checked and the error is swallowed only when the
|
||||
/// table is really there; anything else is rethrown (SQLSTATE is always <c>00000</c>, so it cannot classify).</item>
|
||||
/// <item><b>TTL (RC3):</b> <c>tables()</c> normalises units (7 days reads back as <c>1 WEEK</c>) so values are compared
|
||||
/// in hours, and an ALTER becomes visible only after the WAL applies it, so the bootstrapper never re-reads to verify
|
||||
/// its own ALTER. A boot whose TTL already matches issues no ALTER.</item>
|
||||
/// <item><b>Failure:</b> not latched. Within <see cref="BootstrapFailureCache.DefaultDuration"/> the last failure is
|
||||
/// re-thrown without touching QuestDB.</item>
|
||||
/// <item><c>AutoCreateSchema=false</c> only verifies the tables exist (no TTL changes) and throws
|
||||
/// <see cref="HistorianSchemaMissingException"/> otherwise.</item>
|
||||
/// </list>
|
||||
/// </summary>
|
||||
internal sealed class QuestDbSchemaBootstrapper
|
||||
{
|
||||
private readonly QuestDbConnection _conn;
|
||||
private readonly QuestDbHistorianOptions _options;
|
||||
private readonly ILogger _log;
|
||||
private readonly SemaphoreSlim _gate = new(1, 1);
|
||||
private readonly BootstrapFailureCache _failure; // guarded by _gate
|
||||
private volatile bool _ready;
|
||||
|
||||
/// <summary>Creates the bootstrapper.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
/// <param name="options">Validated options.</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="timeProvider">Clock for the failure cache; system by default.</param>
|
||||
public QuestDbSchemaBootstrapper(QuestDbConnection conn, QuestDbHistorianOptions options, ILogger? log = null,
|
||||
TimeProvider? timeProvider = null)
|
||||
{
|
||||
_conn = conn;
|
||||
_options = options;
|
||||
_log = log ?? NullLogger.Instance;
|
||||
_failure = new BootstrapFailureCache(timeProvider ?? TimeProvider.System, BootstrapFailureCache.DefaultDuration);
|
||||
}
|
||||
|
||||
/// <summary>Ensures the schema is ready; no-op after the first success.</summary>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>A task completing when the tables are usable.</returns>
|
||||
/// <exception cref="HistorianSchemaMissingException">AutoCreateSchema is false and a table is absent.</exception>
|
||||
public async Task EnsureAsync(CancellationToken ct)
|
||||
{
|
||||
if (_ready) return;
|
||||
await _gate.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
if (_ready) return;
|
||||
_failure.ThrowIfCached();
|
||||
try
|
||||
{
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
if (_options.AutoCreateSchema)
|
||||
{
|
||||
await CreateTablesAsync(conn, ct).ConfigureAwait(false);
|
||||
await ReconcileTtlAsync(conn, ct).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
await VerifyAsync(conn, ct).ConfigureAwait(false);
|
||||
}
|
||||
_ready = true;
|
||||
_failure.Clear();
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
|
||||
{
|
||||
_failure.Record(ex);
|
||||
_log.LogWarning("QuestDB historian schema bootstrap (prefix '{Prefix}') failed ({Exception}: {Message}); " +
|
||||
"historian calls fail fast for {Seconds} s before retrying.",
|
||||
_conn.TablePrefix, ex.GetType().Name, ex.Message, BootstrapFailureCache.DefaultDuration.TotalSeconds);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_gate.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task VerifyAsync(NpgsqlConnection conn, CancellationToken ct)
|
||||
{
|
||||
foreach (var table in QuestDbTables.All(_conn.TablePrefix))
|
||||
{
|
||||
if (!await _conn.TableExistsAsync(conn, table, ct).ConfigureAwait(false))
|
||||
throw new HistorianSchemaMissingException(
|
||||
$"QuestDB historian table '{table}' is missing and AutoCreateSchema is false. " +
|
||||
"Create the tables (see docs/Historian.md) or enable AutoCreateSchema.");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task CreateTablesAsync(NpgsqlConnection conn, CancellationToken ct)
|
||||
{
|
||||
var statements = QuestDbTables.CreateAll(_conn.TablePrefix, _options.RetentionDays);
|
||||
var tables = QuestDbTables.All(_conn.TablePrefix);
|
||||
for (var i = 0; i < statements.Length; i++)
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var cmd = _conn.CreateCommand(conn, statements[i]);
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (PostgresException ex) when (QuestDbConnection.IsAlreadyExists(ex))
|
||||
{
|
||||
// A concurrent creator won the race - swallow ONLY if the table is really there.
|
||||
if (!await _conn.TableExistsAsync(conn, tables[i], ct).ConfigureAwait(false)) throw;
|
||||
_log.LogDebug("QuestDB table {Table} was created concurrently.", tables[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ReconcileTtlAsync(NpgsqlConnection conn, CancellationToken ct)
|
||||
{
|
||||
var wantHours = _options.RetentionDays is { } d ? d * 24L : 0L;
|
||||
foreach (var table in QuestDbTables.Expiring(_conn.TablePrefix))
|
||||
{
|
||||
long haveHours;
|
||||
await using (var cmd = _conn.CreateCommand(conn, "SELECT ttlValue, ttlUnit FROM tables() WHERE table_name = $1"))
|
||||
{
|
||||
QuestDbConnection.BindText(cmd, table);
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
if (!await r.ReadAsync(ct).ConfigureAwait(false)) continue; // vanished (dropped between create and here)
|
||||
var value = r.IsDBNull(0) ? 0L : Convert.ToInt64(r.GetValue(0));
|
||||
var unit = r.IsDBNull(1) ? null : r.GetString(1);
|
||||
haveHours = QuestDbTables.TtlHours(value, unit);
|
||||
}
|
||||
if (haveHours == wantHours) continue;
|
||||
|
||||
// The ALTER is a WAL transaction: tables() keeps the old value until it applies, so do NOT re-read to verify.
|
||||
await using (var alter = _conn.CreateCommand(conn, QuestDbTables.AlterTtl(table, _options.RetentionDays)))
|
||||
await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
_log.LogInformation("Applied TTL {Days} day(s) to QuestDB table {Table}.", _options.RetentionDays, table);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
using System.Globalization;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// Table names and DDL for the QuestDB backend (T0-confirmed against QuestDB 10.0.1). The TTL clause sits
|
||||
/// between <c>PARTITION BY</c> and <c>WAL</c> and is an inlined integer (QuestDB cannot parameterise it).
|
||||
/// Prefixes are regex-validated by the caller; these builders also refuse an invalid one.
|
||||
/// </summary>
|
||||
internal static class QuestDbTables
|
||||
{
|
||||
/// <summary>Base name of the sample table (prefix + this).</summary>
|
||||
public const string SampleBase = "sample";
|
||||
/// <summary>Base name of the event table.</summary>
|
||||
public const string EventBase = "event";
|
||||
/// <summary>Base name of the tag catalog table.</summary>
|
||||
public const string TagBase = "tag";
|
||||
|
||||
/// <summary>The fixed designated timestamp of every <c>tag</c> row, which makes the dedup key an upsert by name.</summary>
|
||||
public static readonly DateTime TagEpoch = DateTime.UnixEpoch;
|
||||
|
||||
/// <summary>The prefixed sample table name.</summary>
|
||||
public static string Sample(string prefix) => Name(prefix, SampleBase);
|
||||
/// <summary>The prefixed event table name.</summary>
|
||||
public static string Event(string prefix) => Name(prefix, EventBase);
|
||||
/// <summary>The prefixed tag table name.</summary>
|
||||
public static string Tag(string prefix) => Name(prefix, TagBase);
|
||||
|
||||
/// <summary>All three table names in creation order.</summary>
|
||||
public static string[] All(string prefix) => [Sample(prefix), Event(prefix), Tag(prefix)];
|
||||
|
||||
/// <summary>The tables that carry a TTL (sample and event; the tag catalog never expires).</summary>
|
||||
public static string[] Expiring(string prefix) => [Sample(prefix), Event(prefix)];
|
||||
|
||||
/// <summary><c>CREATE TABLE IF NOT EXISTS</c> for the sample table.</summary>
|
||||
public static string CreateSample(string prefix, int? retentionDays) =>
|
||||
$"CREATE TABLE IF NOT EXISTS {Sample(prefix)} (ts TIMESTAMP, tag SYMBOL, kind BYTE, num_value DOUBLE, " +
|
||||
"int_value LONG, str_value VARCHAR, status LONG, deleted BOOLEAN, server_ts TIMESTAMP) " +
|
||||
$"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL DEDUP UPSERT KEYS(ts, tag)";
|
||||
|
||||
/// <summary><c>CREATE TABLE IF NOT EXISTS</c> for the event table.</summary>
|
||||
public static string CreateEvent(string prefix, int? retentionDays) =>
|
||||
$"CREATE TABLE IF NOT EXISTS {Event(prefix)} (ts TIMESTAMP, event_id VARCHAR, source SYMBOL, alarm_id VARCHAR, " +
|
||||
"alarm_name VARCHAR, alarm_type_name VARCHAR, severity INT, event_kind VARCHAR, message VARCHAR, " +
|
||||
"user_name VARCHAR, comment VARCHAR, received_utc TIMESTAMP) " +
|
||||
$"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL DEDUP UPSERT KEYS(ts, event_id)";
|
||||
|
||||
/// <summary><c>CREATE TABLE IF NOT EXISTS</c> for the tag catalog (YEAR partitions, no TTL ever).</summary>
|
||||
public static string CreateTag(string prefix) =>
|
||||
$"CREATE TABLE IF NOT EXISTS {Tag(prefix)} (ts TIMESTAMP, name SYMBOL, data_type BYTE, eng_unit VARCHAR, " +
|
||||
"description VARCHAR, updated_utc TIMESTAMP) TIMESTAMP(ts) PARTITION BY YEAR WAL DEDUP UPSERT KEYS(ts, name)";
|
||||
|
||||
/// <summary>The three CREATE statements in creation order.</summary>
|
||||
public static string[] CreateAll(string prefix, int? retentionDays) =>
|
||||
[CreateSample(prefix, retentionDays), CreateEvent(prefix, retentionDays), CreateTag(prefix)];
|
||||
|
||||
/// <summary><c>ALTER TABLE t SET TTL n DAYS</c>; <c>null</c> removes the TTL (<c>SET TTL 0 DAYS</c>).</summary>
|
||||
public static string AlterTtl(string table, int? retentionDays) =>
|
||||
$"ALTER TABLE {table} SET TTL {(retentionDays ?? 0).ToString(CultureInfo.InvariantCulture)} DAYS";
|
||||
|
||||
/// <summary>
|
||||
/// Converts a <c>tables()</c> <c>ttlValue</c>/<c>ttlUnit</c> pair to hours. QuestDB normalises the unit (7 days
|
||||
/// reads back as <c>1 WEEK</c>), so equality is judged in hours. An unrecognised unit returns -1 (never equal to
|
||||
/// a wanted value, so the bootstrapper re-applies the TTL).
|
||||
/// </summary>
|
||||
public static long TtlHours(long value, string? unit) => (unit ?? "").ToUpperInvariant() switch
|
||||
{
|
||||
"HOUR" => value,
|
||||
"DAY" => value * 24,
|
||||
"WEEK" => value * 168,
|
||||
_ => value == 0 ? 0 : -1,
|
||||
};
|
||||
|
||||
private static string Ttl(int? days) =>
|
||||
days is { } d ? $" TTL {d.ToString(CultureInfo.InvariantCulture)} DAYS" : "";
|
||||
|
||||
private static string Name(string prefix, string baseName)
|
||||
{
|
||||
if (!QuestDbHistorianOptions.IsValidPrefix(prefix))
|
||||
throw new ArgumentException("Invalid table prefix.", nameof(prefix));
|
||||
return prefix + baseName;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// The only "is this tag known" authority for the QuestDB backend (R7): a row in the tag table (fixed designated
|
||||
/// timestamp 1970-01-01, deduplicated by <c>name</c>). Caches positive results only. Both methods run the schema
|
||||
/// bootstrap first. <b>Names longer than <see cref="HistorianTagNames.MaxLength"/> are never truncated</b>:
|
||||
/// <see cref="ExistsAsync"/> answers false and <see cref="EnsureAsync"/> throws.
|
||||
/// <para><see cref="EnsureAsync"/> inserts only when the tag is absent, because the table's dedup key replaces a
|
||||
/// whole row: re-inserting a known tag would erase the engineering unit and description the provisioner wrote.</para>
|
||||
/// </summary>
|
||||
internal sealed class QuestDbTagCatalog
|
||||
{
|
||||
private readonly QuestDbConnection _conn;
|
||||
private readonly QuestDbSchemaBootstrapper _bootstrap;
|
||||
private readonly QuestDbWalVisibility _wal;
|
||||
private readonly TimeProvider _clock;
|
||||
private readonly ConcurrentDictionary<string, byte> _known = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>Creates the catalog.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
/// <param name="bootstrap">Schema bootstrapper.</param>
|
||||
/// <param name="wal">WAL visibility wait.</param>
|
||||
/// <param name="clock">Clock for <c>updated_utc</c>; system by default.</param>
|
||||
public QuestDbTagCatalog(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbWalVisibility wal,
|
||||
TimeProvider? clock = null)
|
||||
{
|
||||
_conn = conn;
|
||||
_bootstrap = bootstrap;
|
||||
_wal = wal;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <summary>Read-only lookup; never creates a row.</summary>
|
||||
/// <param name="name">Historian tag name.</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>True when the tag is in the catalog; false for an unknown or over-long name.</returns>
|
||||
public async Task<bool> ExistsAsync(string name, CancellationToken ct)
|
||||
{
|
||||
if (!HistorianTagNames.IsValid(name)) return false;
|
||||
if (_known.ContainsKey(name)) return true;
|
||||
await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
await using var cmd = _conn.CreateCommand(conn, $"SELECT count() FROM {_conn.TagTable} WHERE name = $1");
|
||||
QuestDbConnection.BindText(cmd, name);
|
||||
var found = Convert.ToInt64(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) > 0;
|
||||
if (found) _known[name] = 0;
|
||||
return found;
|
||||
}
|
||||
|
||||
/// <summary>Registers the tag when absent (no-op when present) and waits for the row to become visible.</summary>
|
||||
/// <param name="name">Historian tag name.</param>
|
||||
/// <param name="dataType">Initial <c>data_type</c> (only written on create; provisioning corrects it later).</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>A task completing when the tag is registered and visible.</returns>
|
||||
/// <exception cref="ArgumentException">The name is longer than <see cref="HistorianTagNames.MaxLength"/>.</exception>
|
||||
public async Task EnsureAsync(string name, byte dataType, CancellationToken ct)
|
||||
{
|
||||
if (!HistorianTagNames.IsValid(name))
|
||||
throw new ArgumentException($"Historian tag name exceeds {HistorianTagNames.MaxLength} characters.", nameof(name));
|
||||
if (await ExistsAsync(name, ct).ConfigureAwait(false)) return;
|
||||
|
||||
await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
await using var cmd = _conn.CreateCommand(conn,
|
||||
$"INSERT INTO {_conn.TagTable} (ts, name, data_type, updated_utc) VALUES ($1, $2, $3, $4)");
|
||||
QuestDbConnection.BindTimestamp(cmd, QuestDbTables.TagEpoch);
|
||||
QuestDbConnection.BindText(cmd, name);
|
||||
QuestDbConnection.BindByte(cmd, dataType);
|
||||
QuestDbConnection.BindTimestamp(cmd, _clock.GetUtcNow().UtcDateTime);
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
await _wal.WaitAsync(_conn.TagTable, ct).ConfigureAwait(false);
|
||||
_known[name] = 0;
|
||||
}
|
||||
|
||||
/// <summary>Marks a tag as known after another component (the provisioner) has written its row and waited.</summary>
|
||||
/// <param name="name">Historian tag name.</param>
|
||||
internal void MarkKnown(string name) => _known[name] = 0;
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// The read-your-writes wait (R5, T0 Q4). QuestDB applies WAL commits asynchronously, so a row just inserted may
|
||||
/// not be visible to the next query. After its INSERT(s) a writer calls <see cref="WaitAsync"/>, which captures
|
||||
/// <c>sequencerTxn</c> from <c>wal_tables()</c> (every autocommitted INSERT is one sequencer transaction; a concurrent
|
||||
/// writer can only raise the target) and polls <c>writerTxn</c> every ~10 ms until it reaches the target, the table
|
||||
/// is suspended, or <see cref="QuestDbHistorianOptions.VisibilityTimeout"/> elapses.
|
||||
/// </summary>
|
||||
internal sealed class QuestDbWalVisibility
|
||||
{
|
||||
private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(10);
|
||||
|
||||
private readonly QuestDbConnection _conn;
|
||||
private readonly TimeSpan _timeout;
|
||||
private readonly ILogger _log;
|
||||
|
||||
/// <summary>Creates the waiter.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
/// <param name="timeout">Maximum wait; <see cref="TimeSpan.Zero"/> (or less) disables the wait.</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
public QuestDbWalVisibility(QuestDbConnection conn, TimeSpan timeout, ILogger? log = null)
|
||||
{
|
||||
_conn = conn;
|
||||
_timeout = timeout;
|
||||
_log = log ?? NullLogger.Instance;
|
||||
}
|
||||
|
||||
/// <summary>True when the wait is disabled (<c>VisibilityTimeout</c> is 0).</summary>
|
||||
public bool Disabled => _timeout <= TimeSpan.Zero;
|
||||
|
||||
/// <summary>True when the applied transaction has caught up with the target, or the table cannot progress.</summary>
|
||||
/// <param name="writerTxn">Last applied transaction.</param>
|
||||
/// <param name="targetTxn">The sequencer transaction captured after the INSERT.</param>
|
||||
/// <returns>Whether polling can stop because the rows are visible.</returns>
|
||||
internal static bool Reached(long writerTxn, long targetTxn) => writerTxn >= targetTxn;
|
||||
|
||||
/// <summary>Waits until everything this caller inserted into <paramref name="table"/> is visible to queries.</summary>
|
||||
/// <param name="table">Full table name written to.</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>
|
||||
/// True when visible (or the wait is disabled, or the table is not a WAL table); false when the table is suspended
|
||||
/// or the timeout elapsed (a Warning is logged; the rows are still committed and become visible later).
|
||||
/// </returns>
|
||||
public async Task<bool> WaitAsync(string table, CancellationToken ct)
|
||||
{
|
||||
if (Disabled) return true;
|
||||
var sw = Stopwatch.StartNew();
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
|
||||
long target;
|
||||
await using (var cmd = _conn.CreateCommand(conn, "SELECT sequencerTxn FROM wal_tables() WHERE name = $1"))
|
||||
{
|
||||
QuestDbConnection.BindText(cmd, table);
|
||||
var o = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
|
||||
if (o is null or DBNull) return true; // not a WAL table: writes are immediately visible
|
||||
target = Convert.ToInt64(o);
|
||||
}
|
||||
|
||||
while (true)
|
||||
{
|
||||
long writer;
|
||||
bool suspended;
|
||||
await using (var cmd = _conn.CreateCommand(conn, "SELECT writerTxn, suspended FROM wal_tables() WHERE name = $1"))
|
||||
{
|
||||
QuestDbConnection.BindText(cmd, table);
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
if (!await r.ReadAsync(ct).ConfigureAwait(false)) return true;
|
||||
writer = r.GetInt64(0);
|
||||
suspended = r.GetBoolean(1);
|
||||
}
|
||||
if (Reached(writer, target)) return true;
|
||||
if (suspended)
|
||||
{
|
||||
_log.LogWarning("QuestDB table {Table} is suspended (writerTxn {Writer} < sequencerTxn {Target}); not waiting.",
|
||||
table, writer, target);
|
||||
return false;
|
||||
}
|
||||
if (sw.Elapsed >= _timeout)
|
||||
{
|
||||
_log.LogWarning("QuestDB table {Table} did not become visible within {Timeout} ms (writerTxn {Writer} < sequencerTxn {Target}).",
|
||||
table, _timeout.TotalMilliseconds, writer, target);
|
||||
return false;
|
||||
}
|
||||
await Task.Delay(PollInterval, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
||||
<GenerateDocumentationFile>true</GenerateDocumentationFile>
|
||||
<NoWarn>$(NoWarn);CS1591</NoWarn>
|
||||
<Platforms>AnyCPU;x64</Platforms>
|
||||
<RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.csproj"/>
|
||||
<!-- The database-independent historian core shared with the SQL Server backend. -->
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj"/>
|
||||
<!-- Runtime owns IHistorianBackendProvider / ServerHistorianOptions (same diamond as the SqlServer backend). -->
|
||||
<ProjectReference Include="..\..\Server\ZB.MOM.WW.OtOpcUa.Runtime\ZB.MOM.WW.OtOpcUa.Runtime.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Npgsql" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests"/>
|
||||
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
+1
@@ -44,6 +44,7 @@
|
||||
puts it in StatusCodeParityTests' scope. -->
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
|
||||
<!-- The SQL historian backends' shared status constants (HistorianStatusCodes) live here. -->
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.csproj"/>
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj"/>
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
+241
@@ -0,0 +1,241 @@
|
||||
using Npgsql;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
|
||||
|
||||
[Collection(QuestDbCollection.Name)]
|
||||
public sealed class BootstrapTests(QuestDbFixture fx)
|
||||
{
|
||||
private (QuestDbConnection Conn, QuestDbSchemaBootstrapper Boot, QuestDbHistorianOptions Opts) Make(
|
||||
string prefix, int? retention = null, bool autoCreate = true)
|
||||
{
|
||||
var o = fx.Options(prefix, retention, autoCreate);
|
||||
var c = new QuestDbConnection(o);
|
||||
return (c, new QuestDbSchemaBootstrapper(c, o), o);
|
||||
}
|
||||
|
||||
private static async Task<List<string>> TablesAsync(QuestDbConnection c, string prefix)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn, "SELECT table_name FROM tables() WHERE starts_with(table_name, $1) ORDER BY table_name");
|
||||
QuestDbConnection.BindText(cmd, prefix);
|
||||
var names = new List<string>();
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
while (await r.ReadAsync()) names.Add(r.GetString(0));
|
||||
return names;
|
||||
}
|
||||
|
||||
private static async Task<(long Value, string Unit)> TtlAsync(QuestDbConnection c, string table)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn, "SELECT ttlValue, ttlUnit FROM tables() WHERE table_name = $1");
|
||||
QuestDbConnection.BindText(cmd, table);
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
(await r.ReadAsync()).ShouldBeTrue();
|
||||
return (Convert.ToInt64(r.GetValue(0)), r.GetString(1));
|
||||
}
|
||||
|
||||
private static async Task<long> SequencerTxnAsync(QuestDbConnection c, string table)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn, "SELECT sequencerTxn FROM wal_tables() WHERE name = $1");
|
||||
QuestDbConnection.BindText(cmd, table);
|
||||
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Tables_are_created()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, _) = Make(p);
|
||||
await boot.EnsureAsync(default);
|
||||
(await TablesAsync(c, p)).ShouldBe([p + "event", p + "sample", p + "tag"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_second_run_is_a_noop_and_issues_no_alter()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, _) = Make(p, retention: 7);
|
||||
await boot.EnsureAsync(default);
|
||||
var before = await SequencerTxnAsync(c, p + "sample");
|
||||
|
||||
var (c2, boot2, _) = Make(p, retention: 7); // a fresh instance on a table whose TTL already matches (stored as 1 WEEK)
|
||||
await boot2.EnsureAsync(default);
|
||||
await Task.Delay(300); // an ALTER would have been applied by now
|
||||
(await SequencerTxnAsync(c2, p + "sample")).ShouldBe(before);
|
||||
(await SequencerTxnAsync(c2, p + "event")).ShouldBe(await SequencerTxnAsync(c, p + "event"));
|
||||
(await TablesAsync(c2, p)).Count.ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Two_concurrent_bootstrappers_both_succeed()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
for (var round = 0; round < 5; round++)
|
||||
{
|
||||
var p = fx.UniquePrefix();
|
||||
var a = Make(p, retention: 7);
|
||||
var b = Make(p, retention: 7);
|
||||
await Task.WhenAll(
|
||||
Task.Run(() => a.Boot.EnsureAsync(default)),
|
||||
Task.Run(() => b.Boot.EnsureAsync(default)));
|
||||
(await TablesAsync(a.Conn, p)).Count.ShouldBe(3);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Ttl_is_in_metadata_when_retention_is_7_days_and_absent_when_unset()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var withTtl = fx.UniquePrefix();
|
||||
var (c1, b1, _) = Make(withTtl, retention: 7);
|
||||
await b1.EnsureAsync(default);
|
||||
QuestDbTables.TtlHours((await TtlAsync(c1, withTtl + "sample")).Value, (await TtlAsync(c1, withTtl + "sample")).Unit).ShouldBe(168);
|
||||
QuestDbTables.TtlHours((await TtlAsync(c1, withTtl + "event")).Value, (await TtlAsync(c1, withTtl + "event")).Unit).ShouldBe(168);
|
||||
(await TtlAsync(c1, withTtl + "tag")).Value.ShouldBe(0); // the tag catalog never expires
|
||||
|
||||
var noTtl = fx.UniquePrefix();
|
||||
var (c2, b2, _) = Make(noTtl);
|
||||
await b2.EnsureAsync(default);
|
||||
(await TtlAsync(c2, noTtl + "sample")).Value.ShouldBe(0);
|
||||
(await TtlAsync(c2, noTtl + "event")).Value.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_changed_retention_is_reapplied_and_removal_clears_it()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, b7, _) = Make(p, retention: 7);
|
||||
await b7.EnsureAsync(default);
|
||||
|
||||
await Make(p, retention: 30).Boot.EnsureAsync(default);
|
||||
await AwaitHoursAsync(c, p + "sample", 720);
|
||||
await AwaitHoursAsync(c, p + "event", 720);
|
||||
|
||||
await Make(p).Boot.EnsureAsync(default); // unset => SET TTL 0
|
||||
await AwaitHoursAsync(c, p + "sample", 0);
|
||||
}
|
||||
|
||||
private static async Task AwaitHoursAsync(QuestDbConnection c, string table, long hours)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddSeconds(10);
|
||||
long got;
|
||||
do
|
||||
{
|
||||
var (v, u) = await TtlAsync(c, table);
|
||||
got = QuestDbTables.TtlHours(v, u);
|
||||
if (got == hours) return;
|
||||
await Task.Delay(50); // the ALTER is a WAL transaction: visible only once applied
|
||||
} while (DateTime.UtcNow < deadline);
|
||||
got.ShouldBe(hours);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AutoCreateSchema_false_against_missing_tables_throws_HistorianSchemaMissingException()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (_, boot, _) = Make(fx.UniquePrefix(), autoCreate: false);
|
||||
var ex = await Should.ThrowAsync<HistorianSchemaMissingException>(() => boot.EnsureAsync(default));
|
||||
ex.Message.ShouldContain("AutoCreateSchema");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AutoCreateSchema_false_passes_once_the_tables_exist()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
await Make(p).Boot.EnsureAsync(default);
|
||||
await Make(p, autoCreate: false).Boot.EnsureAsync(default);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExistsAsync_is_false_for_an_unknown_tag_and_true_after_EnsureAsync()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, o) = Make(p);
|
||||
var catalog = new QuestDbTagCatalog(c, boot, new QuestDbWalVisibility(c, o.VisibilityTimeout));
|
||||
|
||||
(await catalog.ExistsAsync("Plant.Pump1.Speed", default)).ShouldBeFalse();
|
||||
await catalog.EnsureAsync("Plant.Pump1.Speed", 3, default);
|
||||
(await catalog.ExistsAsync("Plant.Pump1.Speed", default)).ShouldBeTrue();
|
||||
// A fresh catalog (no cache) sees it through the table, and a different name is still unknown.
|
||||
var fresh = new QuestDbTagCatalog(c, boot, new QuestDbWalVisibility(c, o.VisibilityTimeout));
|
||||
(await fresh.ExistsAsync("Plant.Pump1.Speed", default)).ShouldBeTrue();
|
||||
(await fresh.ExistsAsync("Plant.Pump1.Other", default)).ShouldBeFalse();
|
||||
await catalog.EnsureAsync("Plant.Pump1.Speed", 3, default); // idempotent
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_over_long_tag_name_is_unknown_and_cannot_be_ensured()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, o) = Make(p);
|
||||
var catalog = new QuestDbTagCatalog(c, boot, new QuestDbWalVisibility(c, o.VisibilityTimeout));
|
||||
var name = new string('x', HistorianTagNames.MaxLength + 1);
|
||||
(await catalog.ExistsAsync(name, default)).ShouldBeFalse();
|
||||
await Should.ThrowAsync<ArgumentException>(() => catalog.EnsureAsync(name, 3, default));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Wait_returns_once_writerTxn_reaches_the_captured_sequencerTxn()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, o) = Make(p);
|
||||
await boot.EnsureAsync(default);
|
||||
var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5));
|
||||
for (var i = 0; i < 20; i++)
|
||||
{
|
||||
var ts = new DateTime(2026, 3, 1, 0, 0, i, DateTimeKind.Utc);
|
||||
await using (var conn = await c.OpenAsync(default))
|
||||
{
|
||||
await using var cmd = c.CreateCommand(conn,
|
||||
$"INSERT INTO {c.SampleTable} (ts, tag, kind, num_value, status, deleted) VALUES ($1, $2, $3, $4, $5, false)");
|
||||
QuestDbConnection.BindTimestamp(cmd, ts);
|
||||
QuestDbConnection.BindText(cmd, "wal-tag");
|
||||
QuestDbConnection.BindByte(cmd, 3);
|
||||
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Double, Value = (double)i });
|
||||
QuestDbConnection.BindLong(cmd, 0);
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
(await wal.WaitAsync(c.SampleTable, default)).ShouldBeTrue();
|
||||
// After the wait the row is visible on the very next query (the T0 gap was 7/20 without it).
|
||||
await using var q = await c.OpenAsync(default);
|
||||
await using var count = c.CreateCommand(q, $"SELECT count() FROM {c.SampleTable} WHERE tag = $1");
|
||||
QuestDbConnection.BindText(count, "wal-tag");
|
||||
Convert.ToInt64(await count.ExecuteScalarAsync()).ShouldBe(i + 1);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Timeout_zero_skips_the_wait()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, _) = Make(p);
|
||||
await boot.EnsureAsync(default);
|
||||
var wal = new QuestDbWalVisibility(c, TimeSpan.Zero);
|
||||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||
(await wal.WaitAsync(c.SampleTable, default)).ShouldBeTrue();
|
||||
sw.ElapsedMilliseconds.ShouldBeLessThan(50); // no round-trip at all
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Waiting_on_a_table_that_is_not_a_WAL_table_returns_immediately()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var p = fx.UniquePrefix();
|
||||
var (c, boot, _) = Make(p);
|
||||
await boot.EnsureAsync(default);
|
||||
var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5));
|
||||
(await wal.WaitAsync(p + "no_such_table", default)).ShouldBeTrue();
|
||||
}
|
||||
}
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Npgsql;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Env-gated QuestDB for the historian integration tests. <c>HISTORIAN_QUESTDB_TEST_CONNECTION</c> absent ⇒
|
||||
/// <see cref="SkipReason"/> is set, nothing connects and every test calls <c>Assert.Skip</c>. Present ⇒ tests isolate
|
||||
/// from each other with <see cref="UniquePrefix"/>; every prefix handed out is dropped on dispose (and a test may
|
||||
/// drop its own early with <see cref="DropPrefixAsync"/>). Never touches a table outside those prefixes.
|
||||
/// </summary>
|
||||
public sealed class QuestDbFixture : IAsyncLifetime
|
||||
{
|
||||
/// <summary>Env var holding an Npgsql connection string for QuestDB's PG-wire port.</summary>
|
||||
public const string ConnectionEnvVar = "HISTORIAN_QUESTDB_TEST_CONNECTION";
|
||||
|
||||
private readonly ConcurrentBag<string> _prefixes = new();
|
||||
|
||||
/// <summary>Why tests must skip, or null when QuestDB is available.</summary>
|
||||
public string? SkipReason { get; private set; }
|
||||
|
||||
/// <summary>The Npgsql connection string. 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 table prefix (<c>"t" + 12 hex + "_"</c>) that nothing uses yet; dropped on dispose.</summary>
|
||||
/// <returns>The prefix.</returns>
|
||||
public string UniquePrefix()
|
||||
{
|
||||
var p = "t" + Guid.NewGuid().ToString("N")[..12] + "_";
|
||||
_prefixes.Add(p);
|
||||
return p;
|
||||
}
|
||||
|
||||
/// <summary>Options targeting this fixture's QuestDB under <paramref name="prefix"/>.</summary>
|
||||
/// <param name="prefix">Table prefix (normally from <see cref="UniquePrefix"/>).</param>
|
||||
/// <param name="retentionDays">Optional TTL.</param>
|
||||
/// <param name="autoCreate">AutoCreateSchema.</param>
|
||||
/// <returns>Valid options.</returns>
|
||||
public QuestDbHistorianOptions Options(string prefix, int? retentionDays = null, bool autoCreate = true) => new()
|
||||
{
|
||||
ConnectionString = ConnectionString,
|
||||
TablePrefix = prefix,
|
||||
RetentionDays = retentionDays,
|
||||
AutoCreateSchema = autoCreate,
|
||||
};
|
||||
|
||||
/// <summary>Drops every table whose name starts with <paramref name="prefix"/> (found through <c>tables()</c>).</summary>
|
||||
/// <param name="prefix">Table prefix.</param>
|
||||
/// <returns>A task completing when they are gone.</returns>
|
||||
public async Task DropPrefixAsync(string prefix)
|
||||
{
|
||||
if (SkipReason is not null) return;
|
||||
if (!QuestDbHistorianOptions.IsValidPrefix(prefix)) throw new ArgumentException("Invalid prefix.", nameof(prefix));
|
||||
await using var ds = NpgsqlDataSource.Create(ConnectionString);
|
||||
await using var conn = await ds.OpenConnectionAsync().ConfigureAwait(false);
|
||||
var names = new List<string>();
|
||||
await using (var cmd = conn.CreateCommand())
|
||||
{
|
||||
cmd.CommandText = "SELECT table_name FROM tables() WHERE starts_with(table_name, $1)"; // never LIKE: '_' is a wildcard
|
||||
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Varchar, Value = prefix });
|
||||
await using var r = await cmd.ExecuteReaderAsync().ConfigureAwait(false);
|
||||
while (await r.ReadAsync().ConfigureAwait(false)) names.Add(r.GetString(0));
|
||||
}
|
||||
foreach (var n in names)
|
||||
{
|
||||
await using var drop = conn.CreateCommand();
|
||||
drop.CommandText = $"DROP TABLE IF EXISTS {n}";
|
||||
await drop.ExecuteNonQueryAsync().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public ValueTask InitializeAsync()
|
||||
{
|
||||
var cs = Environment.GetEnvironmentVariable(ConnectionEnvVar);
|
||||
if (string.IsNullOrWhiteSpace(cs))
|
||||
SkipReason = $"{ConnectionEnvVar} is not set; QuestDB historian integration tests skipped.";
|
||||
else
|
||||
ConnectionString = cs;
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (SkipReason is not null) return;
|
||||
foreach (var p in _prefixes.Distinct())
|
||||
await DropPrefixAsync(p).ConfigureAwait(false);
|
||||
NpgsqlConnection.ClearAllPools();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Shares one fixture across the QuestDB integration tests.</summary>
|
||||
[CollectionDefinition(Name)]
|
||||
public sealed class QuestDbCollection : ICollectionFixture<QuestDbFixture>
|
||||
{
|
||||
/// <summary>Collection name.</summary>
|
||||
public const string Name = "QuestDb";
|
||||
}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
<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.QuestDb.IntegrationTests</RootNamespace>
|
||||
<NoWarn>$(NoWarn);xUnit1051</NoWarn>
|
||||
</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.QuestDb\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.csproj"/>
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- Shared conformance suite (owned by the Conformance task); an empty glob until it lands. -->
|
||||
<Compile Include="..\Historian.Conformance\**\*.cs" LinkBase="Conformance"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
+140
@@ -0,0 +1,140 @@
|
||||
using System.Net.Sockets;
|
||||
using Npgsql;
|
||||
using NpgsqlTypes;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
||||
|
||||
public sealed class QuestDbConnectionTests
|
||||
{
|
||||
private static DateTime Bound(DateTime v, bool clamp = false)
|
||||
{
|
||||
var cmd = new NpgsqlCommand();
|
||||
var p = QuestDbConnection.BindTimestamp(cmd, v, clamp);
|
||||
p.NpgsqlDbType.ShouldBe(NpgsqlDbType.TimestampTz);
|
||||
var bound = (DateTime)p.Value!;
|
||||
bound.Kind.ShouldBe(DateTimeKind.Utc);
|
||||
return bound;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Round_truncates_sub_microsecond_ticks()
|
||||
{
|
||||
var t = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc).AddTicks(1234567);
|
||||
QuestDbConnection.Round(t).ShouldBe(new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc).AddTicks(1234560));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Round_truncates_rather_than_rounds_before_2000()
|
||||
{
|
||||
var t = new DateTime(1990, 5, 5, 0, 0, 0, DateTimeKind.Utc).AddTicks(1234565);
|
||||
QuestDbConnection.Round(t).Ticks.ShouldBe(t.Ticks - 5);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Round_treats_Unspecified_as_UTC()
|
||||
{
|
||||
var u = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Unspecified);
|
||||
var r = QuestDbConnection.Round(u);
|
||||
r.Kind.ShouldBe(DateTimeKind.Utc);
|
||||
r.Ticks.ShouldBe(u.Ticks);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Round_converts_Local_to_UTC()
|
||||
{
|
||||
var local = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Local);
|
||||
var r = QuestDbConnection.Round(local);
|
||||
r.Kind.ShouldBe(DateTimeKind.Utc);
|
||||
r.ShouldBe(local.ToUniversalTime());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Round_keeps_a_UTC_value_that_is_already_aligned()
|
||||
{
|
||||
var t = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
|
||||
QuestDbConnection.Round(t).ShouldBe(t);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)] // MinValue itself
|
||||
[InlineData(5)] // truncates to MinValue
|
||||
public void BindTimestamp_never_emits_MinValue(long ticks)
|
||||
{
|
||||
var b = Bound(new DateTime(ticks, DateTimeKind.Utc));
|
||||
b.ShouldNotBe(DateTime.MinValue);
|
||||
b.Ticks.ShouldBe(10);
|
||||
b.ShouldBe(new DateTime(10, DateTimeKind.Utc)); // 0001-01-01T00:00:00.000001 (T0-confirmed)
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BindTimestamp_never_emits_MaxValue()
|
||||
{
|
||||
var b = Bound(DateTime.MaxValue);
|
||||
b.ShouldNotBe(DateTime.MaxValue);
|
||||
b.ShouldBe(new DateTime(9999, 12, 31, 23, 59, 59, DateTimeKind.Utc).AddTicks(9999990)); // T0-confirmed bound
|
||||
Bound(DateTime.MinValue).ShouldNotBe(DateTime.MinValue);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BindTimestamp_never_emits_either_extreme_for_any_kind()
|
||||
{
|
||||
foreach (var kind in new[] { DateTimeKind.Utc, DateTimeKind.Unspecified, DateTimeKind.Local })
|
||||
foreach (var t in new[] { new DateTime(DateTime.MinValue.Ticks, kind), new DateTime(DateTime.MaxValue.Ticks, kind) })
|
||||
{
|
||||
var b = Bound(t);
|
||||
b.ShouldNotBe(DateTime.MinValue);
|
||||
b.ShouldNotBe(DateTime.MaxValue);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BindTimestamp_keeps_a_pre_1970_range_bound_unless_asked_to_clamp()
|
||||
{
|
||||
var pre = new DateTime(1960, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
Bound(pre).ShouldBe(pre);
|
||||
Bound(pre, clamp: true).ShouldBe(DateTime.UnixEpoch);
|
||||
Bound(new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc), clamp: true).Year.ShouldBe(2026);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void IsStorable_is_false_only_before_1970()
|
||||
{
|
||||
QuestDbConnection.IsStorable(DateTime.UnixEpoch).ShouldBeTrue();
|
||||
QuestDbConnection.IsStorable(DateTime.UnixEpoch.AddTicks(-1)).ShouldBeFalse();
|
||||
QuestDbConnection.IsStorable(new DateTime(1601, 1, 1, 0, 0, 0, DateTimeKind.Utc)).ShouldBeFalse();
|
||||
QuestDbConnection.IsStorable(DateTime.MinValue).ShouldBeFalse();
|
||||
QuestDbConnection.IsStorable(DateTime.MaxValue).ShouldBeTrue();
|
||||
QuestDbConnection.IsStorable(new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc)).ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BindByte_binds_smallint()
|
||||
{
|
||||
var p = QuestDbConnection.BindByte(new NpgsqlCommand(), 200);
|
||||
p.NpgsqlDbType.ShouldBe(NpgsqlDbType.Smallint);
|
||||
p.Value.ShouldBe((short)200);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void IsTransient_classifies_transport_failures_not_server_rejections()
|
||||
{
|
||||
QuestDbConnection.IsTransient(new SocketException()).ShouldBeTrue();
|
||||
QuestDbConnection.IsTransient(new IOException("x")).ShouldBeTrue();
|
||||
QuestDbConnection.IsTransient(new TimeoutException()).ShouldBeTrue();
|
||||
QuestDbConnection.IsTransient(new NpgsqlException("x", new IOException())).ShouldBeTrue();
|
||||
QuestDbConnection.IsTransient(new InvalidOperationException("x")).ShouldBeFalse();
|
||||
QuestDbConnection.IsTransient(new OperationCanceledException()).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void T_prefixes_the_table_base_name()
|
||||
{
|
||||
var c = new QuestDbConnection(new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1", TablePrefix = "p_" });
|
||||
c.T("sample").ShouldBe("p_sample");
|
||||
c.SampleTable.ShouldBe("p_sample");
|
||||
c.EventTable.ShouldBe("p_event");
|
||||
c.TagTable.ShouldBe("p_tag");
|
||||
}
|
||||
}
|
||||
+79
@@ -0,0 +1,79 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
||||
|
||||
public sealed class QuestDbHistorianOptionsTests
|
||||
{
|
||||
private static QuestDbHistorianOptions Valid() => new() { ConnectionString = "Host=h;Port=8812" };
|
||||
|
||||
[Fact]
|
||||
public void Defaults_are_the_documented_ones()
|
||||
{
|
||||
var o = new QuestDbHistorianOptions();
|
||||
o.ConnectionString.ShouldBe("");
|
||||
o.TablePrefix.ShouldBe("hist_");
|
||||
o.RetentionDays.ShouldBeNull();
|
||||
o.CommandTimeout.ShouldBe(TimeSpan.FromSeconds(30));
|
||||
o.AutoCreateSchema.ShouldBeTrue();
|
||||
o.VisibilityTimeout.ShouldBe(TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_valid_configuration_has_no_errors() => Valid().Validate().ShouldBeEmpty();
|
||||
|
||||
[Fact]
|
||||
public void Missing_connection_string_names_the_key()
|
||||
{
|
||||
var o = Valid(); o.ConnectionString = " ";
|
||||
o.Validate().ShouldHaveSingleItem().ShouldContain("ServerHistorian:QuestDb:ConnectionString");
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData("hist-")]
|
||||
[InlineData("hist.")]
|
||||
[InlineData("hist_; DROP")]
|
||||
public void Bad_prefix_names_the_key(string prefix)
|
||||
{
|
||||
var o = Valid(); o.TablePrefix = prefix;
|
||||
o.Validate().ShouldHaveSingleItem().ShouldContain("ServerHistorian:QuestDb:TablePrefix");
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("t0123456789ab_")]
|
||||
[InlineData("A_9")]
|
||||
public void Good_prefixes_pass(string prefix)
|
||||
{
|
||||
var o = Valid(); o.TablePrefix = prefix;
|
||||
o.Validate().ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)]
|
||||
[InlineData(-3)]
|
||||
public void Retention_below_one_names_the_key(int days)
|
||||
{
|
||||
var o = Valid(); o.RetentionDays = days;
|
||||
o.Validate().ShouldHaveSingleItem().ShouldContain("ServerHistorian:QuestDb:RetentionDays");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Retention_of_one_is_valid() { var o = Valid(); o.RetentionDays = 1; o.Validate().ShouldBeEmpty(); }
|
||||
|
||||
[Fact]
|
||||
public void Non_positive_command_timeout_names_the_key()
|
||||
{
|
||||
var o = Valid(); o.CommandTimeout = TimeSpan.Zero;
|
||||
o.Validate().ShouldHaveSingleItem().ShouldContain("ServerHistorian:QuestDb:CommandTimeout");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Negative_visibility_timeout_names_the_key_and_zero_is_valid()
|
||||
{
|
||||
var o = Valid(); o.VisibilityTimeout = TimeSpan.FromMilliseconds(-1);
|
||||
o.Validate().ShouldHaveSingleItem().ShouldContain("ServerHistorian:QuestDb:VisibilityTimeout");
|
||||
o.VisibilityTimeout = TimeSpan.Zero;
|
||||
o.Validate().ShouldBeEmpty();
|
||||
}
|
||||
}
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
||||
|
||||
public sealed class QuestDbSchemaBootstrapperFailureTests
|
||||
{
|
||||
private sealed class ManualClock : TimeProvider
|
||||
{
|
||||
private DateTimeOffset _now = new(2026, 10, 11, 0, 0, 0, TimeSpan.Zero);
|
||||
public override DateTimeOffset GetUtcNow() => _now;
|
||||
public void Advance(TimeSpan by) => _now += by;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_failure_is_cached_for_five_seconds_then_retried()
|
||||
{
|
||||
var opts = new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1;Timeout=1" };
|
||||
var clock = new ManualClock();
|
||||
var boot = new QuestDbSchemaBootstrapper(new QuestDbConnection(opts), opts, timeProvider: clock);
|
||||
|
||||
var first = await Should.ThrowAsync<Exception>(() => boot.EnsureAsync(CancellationToken.None));
|
||||
// Within the window the SAME failure is re-thrown without another connection attempt.
|
||||
var second = await Should.ThrowAsync<Exception>(() => boot.EnsureAsync(CancellationToken.None));
|
||||
second.ShouldBeSameAs(first);
|
||||
|
||||
clock.Advance(BootstrapFailureCache.DefaultDuration + TimeSpan.FromSeconds(1));
|
||||
var third = await Should.ThrowAsync<Exception>(() => boot.EnsureAsync(CancellationToken.None));
|
||||
third.ShouldNotBeSameAs(first); // the window closed: QuestDB was tried again
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
||||
|
||||
public sealed class QuestDbTablesTests
|
||||
{
|
||||
[Fact]
|
||||
public void Sample_ddl_without_ttl_is_pinned() =>
|
||||
QuestDbTables.CreateSample("hist_", null).ShouldBe(
|
||||
"CREATE TABLE IF NOT EXISTS hist_sample (ts TIMESTAMP, tag SYMBOL, kind BYTE, num_value DOUBLE, " +
|
||||
"int_value LONG, str_value VARCHAR, status LONG, deleted BOOLEAN, server_ts TIMESTAMP) " +
|
||||
"TIMESTAMP(ts) PARTITION BY DAY WAL DEDUP UPSERT KEYS(ts, tag)");
|
||||
|
||||
[Fact]
|
||||
public void Sample_ddl_with_ttl_puts_ttl_between_partition_and_wal() =>
|
||||
QuestDbTables.CreateSample("hist_", 7).ShouldBe(
|
||||
"CREATE TABLE IF NOT EXISTS hist_sample (ts TIMESTAMP, tag SYMBOL, kind BYTE, num_value DOUBLE, " +
|
||||
"int_value LONG, str_value VARCHAR, status LONG, deleted BOOLEAN, server_ts TIMESTAMP) " +
|
||||
"TIMESTAMP(ts) PARTITION BY DAY TTL 7 DAYS WAL DEDUP UPSERT KEYS(ts, tag)");
|
||||
|
||||
[Fact]
|
||||
public void Event_ddl_is_pinned()
|
||||
{
|
||||
QuestDbTables.CreateEvent("hist_", null).ShouldBe(
|
||||
"CREATE TABLE IF NOT EXISTS hist_event (ts TIMESTAMP, event_id VARCHAR, source SYMBOL, alarm_id VARCHAR, " +
|
||||
"alarm_name VARCHAR, alarm_type_name VARCHAR, severity INT, event_kind VARCHAR, message VARCHAR, " +
|
||||
"user_name VARCHAR, comment VARCHAR, received_utc TIMESTAMP) " +
|
||||
"TIMESTAMP(ts) PARTITION BY DAY WAL DEDUP UPSERT KEYS(ts, event_id)");
|
||||
QuestDbTables.CreateEvent("hist_", 30).ShouldContain("PARTITION BY DAY TTL 30 DAYS WAL DEDUP");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Tag_ddl_is_pinned_and_never_carries_a_ttl() =>
|
||||
QuestDbTables.CreateTag("hist_").ShouldBe(
|
||||
"CREATE TABLE IF NOT EXISTS hist_tag (ts TIMESTAMP, name SYMBOL, data_type BYTE, eng_unit VARCHAR, " +
|
||||
"description VARCHAR, updated_utc TIMESTAMP) TIMESTAMP(ts) PARTITION BY YEAR WAL DEDUP UPSERT KEYS(ts, name)");
|
||||
|
||||
[Fact]
|
||||
public void The_prefix_is_applied_to_every_table_name()
|
||||
{
|
||||
QuestDbTables.All("t1_").ShouldBe(["t1_sample", "t1_event", "t1_tag"]);
|
||||
QuestDbTables.CreateAll("t1_", 3).ShouldAllBe(s => s.Contains("t1_"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_invalid_prefix_is_refused() =>
|
||||
Should.Throw<ArgumentException>(() => QuestDbTables.CreateSample("a b", null));
|
||||
|
||||
[Fact]
|
||||
public void Alter_ttl_is_pinned_and_null_removes_it()
|
||||
{
|
||||
QuestDbTables.AlterTtl("hist_sample", 30).ShouldBe("ALTER TABLE hist_sample SET TTL 30 DAYS");
|
||||
QuestDbTables.AlterTtl("hist_sample", null).ShouldBe("ALTER TABLE hist_sample SET TTL 0 DAYS");
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(1, "WEEK", 168)] // 7 days reads back as 1 WEEK
|
||||
[InlineData(2, "WEEK", 336)]
|
||||
[InlineData(30, "DAY", 720)]
|
||||
[InlineData(0, "HOUR", 0)] // no TTL
|
||||
[InlineData(5, "hour", 5)]
|
||||
[InlineData(1, "MONTH", -1)] // unknown unit never equals a wanted value
|
||||
public void Ttl_is_compared_in_hours(long value, string unit, long hours) =>
|
||||
QuestDbTables.TtlHours(value, unit).ShouldBe(hours);
|
||||
}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
||||
|
||||
public sealed class QuestDbWalVisibilityTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Timeout_zero_skips_the_wait_without_touching_the_database()
|
||||
{
|
||||
// Port 1 refuses connections: any attempt to query would throw, so returning true proves nothing was opened.
|
||||
var conn = new QuestDbConnection(new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1;Timeout=1" });
|
||||
var wal = new QuestDbWalVisibility(conn, TimeSpan.Zero);
|
||||
wal.Disabled.ShouldBeTrue();
|
||||
(await wal.WaitAsync("hist_sample", CancellationToken.None)).ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_positive_timeout_does_touch_the_database_control()
|
||||
{
|
||||
var conn = new QuestDbConnection(new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1;Timeout=1" });
|
||||
var wal = new QuestDbWalVisibility(conn, TimeSpan.FromSeconds(1));
|
||||
wal.Disabled.ShouldBeFalse();
|
||||
await Should.ThrowAsync<Exception>(() => wal.WaitAsync("hist_sample", CancellationToken.None));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(5, 5, true)]
|
||||
[InlineData(6, 5, true)]
|
||||
[InlineData(4, 5, false)]
|
||||
public void Reached_is_writerTxn_at_or_beyond_the_target(long writer, long target, bool expected) =>
|
||||
QuestDbWalVisibility.Reached(writer, target).ShouldBe(expected);
|
||||
}
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
<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.QuestDb.Tests</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.QuestDb\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.csproj"/>
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user