Merge qdb/T4: QuestDB writes (value writer, provisioner, alarm writer)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
+130
@@ -0,0 +1,130 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Npgsql;
|
||||
using NpgsqlTypes;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IAlarmHistorianWriter"/> over the QuestDB event table. Each event carries the deterministic
|
||||
/// <see cref="AlarmEventIds.ComputeEventId"/> and is inserted through the table's <c>DEDUP UPSERT KEYS(ts, event_id)</c>
|
||||
/// (T0 Q3 confirmed VARCHAR works as a key), so the store-and-forward drain's at-least-once redelivery, and both
|
||||
/// pair nodes writing the same event, converge on one row (the redelivered row is identical except
|
||||
/// <c>received_utc</c>). Outcomes: Ack on insert; RetryPlease on transient / unclassified errors and on
|
||||
/// cancellation; PermanentFail only for data the server will always reject (a time before 1970, which is not a
|
||||
/// storable designated timestamp, or a server message naming an inconvertible / out-of-range value). Never throws.
|
||||
/// </summary>
|
||||
internal sealed class QuestDbAlarmHistorianWriter : IAlarmHistorianWriter
|
||||
{
|
||||
private static readonly string[] PermanentMessageFragments =
|
||||
["designated timestamp", "inconvertible", "out of range", "too long", "cannot be null"];
|
||||
|
||||
private readonly QuestDbConnection _conn;
|
||||
private readonly QuestDbSchemaBootstrapper _bootstrap;
|
||||
private readonly QuestDbWalVisibility _wal;
|
||||
private readonly ILogger _log;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
/// <summary>Creates the writer.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
/// <param name="bootstrap">Schema bootstrapper.</param>
|
||||
/// <param name="wal">WAL visibility wait (once per batch, so a following read sees the events).</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="clock">Clock for <c>received_utc</c>.</param>
|
||||
public QuestDbAlarmHistorianWriter(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbWalVisibility wal,
|
||||
ILogger? log = null, TimeProvider? clock = null)
|
||||
{
|
||||
_conn = conn;
|
||||
_bootstrap = bootstrap;
|
||||
_wal = wal;
|
||||
_log = log ?? NullLogger.Instance;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(batch);
|
||||
if (batch.Count == 0) return [];
|
||||
var outcomes = new HistorianWriteOutcome[batch.Count];
|
||||
try
|
||||
{
|
||||
await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogWarning("QuestDB alarm historian bootstrap failed ({Exception}); will retry.", ex.GetType().Name);
|
||||
Array.Fill(outcomes, HistorianWriteOutcome.RetryPlease);
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
var wrote = false;
|
||||
for (var i = 0; i < batch.Count; i++)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested) { outcomes[i] = HistorianWriteOutcome.RetryPlease; continue; }
|
||||
outcomes[i] = await WriteOneAsync(batch[i], cancellationToken).ConfigureAwait(false);
|
||||
wrote |= outcomes[i] == HistorianWriteOutcome.Ack;
|
||||
}
|
||||
|
||||
if (wrote)
|
||||
{
|
||||
try { await _wal.WaitAsync(_conn.EventTable, cancellationToken).ConfigureAwait(false); }
|
||||
catch (Exception ex) { _log.LogDebug("QuestDB alarm visibility wait failed ({Exception}).", ex.GetType().Name); }
|
||||
}
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
private async Task<HistorianWriteOutcome> WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!QuestDbConnection.IsStorable(e.TimestampUtc))
|
||||
{
|
||||
_log.LogWarning("QuestDB alarm event dropped: its time is before 1970-01-01 and cannot be stored.");
|
||||
return HistorianWriteOutcome.PermanentFail;
|
||||
}
|
||||
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
await using var cmd = _conn.CreateCommand(conn,
|
||||
$"INSERT INTO {_conn.EventTable} (ts, event_id, source, alarm_id, alarm_name, alarm_type_name, severity, event_kind, message, user_name, comment, received_utc) " +
|
||||
"VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)");
|
||||
QuestDbConnection.BindTimestamp(cmd, e.TimestampUtc);
|
||||
QuestDbConnection.BindText(cmd, AlarmEventIds.ComputeEventId(e));
|
||||
QuestDbConnection.BindText(cmd, e.EquipmentPath ?? "");
|
||||
QuestDbConnection.BindText(cmd, e.AlarmId ?? "");
|
||||
QuestDbConnection.BindText(cmd, e.AlarmName ?? "");
|
||||
QuestDbConnection.BindText(cmd, e.AlarmTypeName ?? "");
|
||||
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = AlarmEventIds.ToNumericSeverity(e.Severity) });
|
||||
QuestDbConnection.BindText(cmd, e.EventKind ?? "");
|
||||
QuestDbConnection.BindText(cmd, e.Message);
|
||||
QuestDbConnection.BindText(cmd, e.User);
|
||||
QuestDbConnection.BindText(cmd, e.Comment);
|
||||
QuestDbConnection.BindTimestamp(cmd, _clock.GetUtcNow().UtcDateTime);
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
return HistorianWriteOutcome.Ack;
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
return HistorianWriteOutcome.RetryPlease;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
var outcome = Classify(ex);
|
||||
_log.LogWarning("QuestDB alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome);
|
||||
return outcome;
|
||||
}
|
||||
}
|
||||
|
||||
// Only a data error QuestDB will reject forever dead-letters; anything unknown or non-database retries (never lose alarm
|
||||
// history). SQLSTATE is always 00000 on QuestDB, so a server rejection is classified by message text.
|
||||
internal static HistorianWriteOutcome Classify(Exception ex)
|
||||
{
|
||||
if (QuestDbConnection.IsTransient(ex)) return HistorianWriteOutcome.RetryPlease;
|
||||
if (ex is PostgresException pg &&
|
||||
PermanentMessageFragments.Any(f => pg.MessageText.Contains(f, StringComparison.OrdinalIgnoreCase)))
|
||||
return HistorianWriteOutcome.PermanentFail;
|
||||
return HistorianWriteOutcome.RetryPlease;
|
||||
}
|
||||
}
|
||||
+180
@@ -0,0 +1,180 @@
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Npgsql;
|
||||
using NpgsqlTypes;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IHistorianValueWriter"/> over the QuestDB sample table. Redelivery-idempotent with first-wins
|
||||
/// semantics: QuestDB's <c>DEDUP UPSERT KEYS(ts, tag)</c> would let a re-sent sample overwrite the stored one (and a
|
||||
/// re-sent live sample resurrect a tombstone), so the writer reads which timestamps already exist for the tag in
|
||||
/// the batch's range (tombstones count as present, T0 Q11 range form) and inserts only the absent rows.
|
||||
/// Never throws; returns false on any failure so the recorder keeps its entries.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Batch normalisation is <see cref="LiveWritePreparer.Prepare"/> at µs resolution (non-finite doubles become Empty
|
||||
/// + Bad, an in-batch µs collision keeps the first). <b>RC1:</b> a sample whose time is before 1970 cannot be a
|
||||
/// QuestDB designated timestamp and one such row would fail the whole multi-row INSERT, so those samples are dropped
|
||||
/// (Warning once per batch, counted in <see cref="DroppedPre1970"/>) and the call still succeeds for the rest.
|
||||
/// <b>R5:</b> after inserting, the writer waits for WAL visibility so an immediate redelivery sees the rows.
|
||||
/// Inserts are chunked at <see cref="BatchRows"/> rows (8 bound parameters per row, far under the 32,767 limit).
|
||||
/// </remarks>
|
||||
internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter
|
||||
{
|
||||
/// <summary>Rows per multi-row INSERT (T0 Q2: 1,000 rows is reliable).</summary>
|
||||
internal const int BatchRows = 1000;
|
||||
|
||||
private readonly QuestDbConnection _conn;
|
||||
private readonly QuestDbTagCatalog _catalog;
|
||||
private readonly QuestDbWalVisibility _wal;
|
||||
private readonly ILogger _log;
|
||||
private readonly Func<DateTime> _utcNow;
|
||||
private long _droppedPre1970;
|
||||
|
||||
/// <summary>Creates the writer.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
/// <param name="catalog">Tag catalog (the only tag authority).</param>
|
||||
/// <param name="wal">WAL visibility wait.</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="utcNow">Clock for null timestamps and <c>server_ts</c> (tests).</param>
|
||||
public QuestDbHistorianValueWriter(QuestDbConnection conn, QuestDbTagCatalog catalog, QuestDbWalVisibility wal,
|
||||
ILogger? log = null, Func<DateTime>? utcNow = null)
|
||||
{
|
||||
_conn = conn;
|
||||
_catalog = catalog;
|
||||
_wal = wal;
|
||||
_log = log ?? NullLogger.Instance;
|
||||
_utcNow = utcNow ?? (() => DateTime.UtcNow);
|
||||
}
|
||||
|
||||
/// <summary>Samples dropped so far because their time is before 1970 (RC1).</summary>
|
||||
internal long DroppedPre1970 => Interlocked.Read(ref _droppedPre1970);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<bool> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!HistorianTagNames.IsValid(tag))
|
||||
{
|
||||
_log.LogWarning("QuestDB historian live-value write refused: tag name is missing or longer than {Max} characters.",
|
||||
HistorianTagNames.MaxLength);
|
||||
return false;
|
||||
}
|
||||
if (values is null || values.Count == 0) return true;
|
||||
|
||||
var prepared = LiveWritePreparer.Prepare(values, _utcNow, resolutionTicks: 10);
|
||||
var rows = prepared.Where(r => QuestDbConnection.IsStorable(r.TimeUtc)).ToList();
|
||||
var dropped = prepared.Count - rows.Count;
|
||||
if (dropped > 0)
|
||||
{
|
||||
Interlocked.Add(ref _droppedPre1970, dropped);
|
||||
_log.LogWarning("QuestDB historian dropped {Count} sample(s) for a tag because their time is before 1970-01-01 (not storable).", dropped);
|
||||
}
|
||||
if (rows.Count == 0) return true;
|
||||
|
||||
var dataType = HistorianDataTypes.FromKind(rows.Select(r => r.Value.Kind).FirstOrDefault(k => k != HistorianValueKind.Empty));
|
||||
await _catalog.EnsureAsync(tag, dataType, ct).ConfigureAwait(false);
|
||||
|
||||
var present = await ExistingTicksAsync(tag, rows, ct).ConfigureAwait(false);
|
||||
var fresh = rows.Where(r => !present.Contains(QuestDbConnection.Round(r.TimeUtc).Ticks)).ToList();
|
||||
if (fresh.Count == 0) return true;
|
||||
|
||||
var serverTs = _utcNow();
|
||||
for (var i = 0; i < fresh.Count; i += BatchRows)
|
||||
await InsertAsync(tag, fresh.GetRange(i, Math.Min(BatchRows, fresh.Count - i)), serverTs, ct).ConfigureAwait(false);
|
||||
|
||||
await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); // timeout/suspension does not change the result (R5)
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogWarning("QuestDB historian live-value write failed ({Exception}); will retry.", ex.GetType().Name);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A gap between consecutive batch times larger than this starts a new existence-query range.</summary>
|
||||
internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1);
|
||||
|
||||
/// <summary>
|
||||
/// Splits ascending <paramref name="sortedTimes"/> into inclusive <c>[Lo, Hi]</c> ranges: a new range starts where the gap to the
|
||||
/// previous time exceeds <paramref name="gap"/> or the range already holds <paramref name="maxPerCluster"/> times. One outlier
|
||||
/// time therefore costs its own tiny query instead of widening one range over every stored row in between.
|
||||
/// </summary>
|
||||
/// <param name="sortedTimes">Times in ascending order.</param>
|
||||
/// <param name="gap">Largest gap kept inside one range.</param>
|
||||
/// <param name="maxPerCluster">Most times in one range (>= 1).</param>
|
||||
/// <returns>The ranges, ascending.</returns>
|
||||
internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList<DateTime> sortedTimes, TimeSpan gap, int maxPerCluster)
|
||||
{
|
||||
ArgumentOutOfRangeException.ThrowIfLessThan(maxPerCluster, 1);
|
||||
var result = new List<(DateTime, DateTime)>();
|
||||
if (sortedTimes.Count == 0) return result;
|
||||
var lo = sortedTimes[0];
|
||||
var prev = lo;
|
||||
var n = 1;
|
||||
for (var i = 1; i < sortedTimes.Count; i++)
|
||||
{
|
||||
var t = sortedTimes[i];
|
||||
if (t - prev > gap || n >= maxPerCluster)
|
||||
{
|
||||
result.Add((lo, prev));
|
||||
lo = t;
|
||||
n = 0;
|
||||
}
|
||||
prev = t;
|
||||
n++;
|
||||
}
|
||||
result.Add((lo, prev));
|
||||
return result;
|
||||
}
|
||||
|
||||
// T0 Q11 range form, one query per cluster: tombstones are NOT filtered out (a deleted row must not be resurrected).
|
||||
private async Task<HashSet<long>> ExistingTicksAsync(string tag, List<HistorianSampleRow> rows, CancellationToken ct)
|
||||
{
|
||||
var times = rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).OrderBy(t => t).ToList();
|
||||
var set = new HashSet<long>();
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap, BatchRows))
|
||||
{
|
||||
await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3");
|
||||
QuestDbConnection.BindText(cmd, tag);
|
||||
QuestDbConnection.BindTimestamp(cmd, lo);
|
||||
QuestDbConnection.BindTimestamp(cmd, hi);
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
while (await r.ReadAsync(ct).ConfigureAwait(false)) set.Add(QuestDbConnection.ReadUtc(r, 0).Ticks);
|
||||
}
|
||||
return set;
|
||||
}
|
||||
|
||||
private async Task InsertAsync(string tag, List<HistorianSampleRow> chunk, DateTime serverTs, CancellationToken ct)
|
||||
{
|
||||
var sql = new StringBuilder($"INSERT INTO {_conn.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ");
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
await using var cmd = _conn.CreateCommand(conn, "");
|
||||
var n = 1;
|
||||
for (var i = 0; i < chunk.Count; i++)
|
||||
{
|
||||
if (i > 0) sql.Append(',');
|
||||
sql.Append($"(${n},${n + 1},${n + 2},${n + 3},${n + 4},${n + 5},${n + 6},false,${n + 7})");
|
||||
n += 8;
|
||||
var row = chunk[i];
|
||||
var c = SampleColumnMapper.ToColumns(row.Value);
|
||||
QuestDbConnection.BindTimestamp(cmd, row.TimeUtc);
|
||||
QuestDbConnection.BindText(cmd, tag);
|
||||
QuestDbConnection.BindByte(cmd, c.ValueKind);
|
||||
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Double, Value = c.NumValue is { } d ? d : DBNull.Value });
|
||||
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Bigint, Value = c.IntValue is { } l ? l : DBNull.Value });
|
||||
QuestDbConnection.BindText(cmd, c.StrValue);
|
||||
QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(row.StatusCode));
|
||||
QuestDbConnection.BindTimestamp(cmd, serverTs);
|
||||
}
|
||||
cmd.CommandText = sql.ToString();
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IHistorianProvisioning"/> for the QuestDB backend: upserts <c>tag</c> rows (fixed designated timestamp
|
||||
/// 1970-01-01, so the dedup key makes it an upsert by name). QuestDB replaces the WHOLE row on a key hit, so a null
|
||||
/// engineering unit / description in a request would wipe the stored one; per chunk the existing rows are read
|
||||
/// (<c>WHERE name IN (…)</c>) and <c>COALESCE(request, stored)</c> is applied client-side before inserting.
|
||||
/// Every data type is provisionable (Skipped is always 0). Never throws. After each chunk it waits for WAL
|
||||
/// visibility and tells the catalog the names are known (R7).
|
||||
/// </summary>
|
||||
internal sealed class QuestDbTagProvisioner : IHistorianProvisioning
|
||||
{
|
||||
private const int ChunkSize = 400; // 6 parameters per tag on insert, 400 on the IN list
|
||||
private const int MaxUnitLength = 64;
|
||||
private const int MaxDescriptionLength = 400;
|
||||
|
||||
private readonly QuestDbConnection _conn;
|
||||
private readonly QuestDbSchemaBootstrapper _bootstrap;
|
||||
private readonly QuestDbTagCatalog _catalog;
|
||||
private readonly QuestDbWalVisibility _wal;
|
||||
private readonly ILogger _log;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
/// <summary>Creates the provisioner.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
/// <param name="bootstrap">Schema bootstrapper.</param>
|
||||
/// <param name="catalog">Tag catalog (told which names are now known).</param>
|
||||
/// <param name="wal">WAL visibility wait.</param>
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="clock">Clock for <c>updated_utc</c>.</param>
|
||||
public QuestDbTagProvisioner(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog,
|
||||
QuestDbWalVisibility wal, ILogger? log = null, TimeProvider? clock = null)
|
||||
{
|
||||
_conn = conn;
|
||||
_bootstrap = bootstrap;
|
||||
_catalog = catalog;
|
||||
_wal = wal;
|
||||
_log = log ?? NullLogger.Instance;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<HistorianProvisionResult> EnsureTagsAsync(IReadOnlyList<HistorianTagProvisionRequest> requests, CancellationToken ct)
|
||||
{
|
||||
var requested = requests?.Count ?? 0;
|
||||
if (requested == 0) return new HistorianProvisionResult(0, 0, 0, 0);
|
||||
var failed = 0;
|
||||
var ensured = 0;
|
||||
try
|
||||
{
|
||||
await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
|
||||
|
||||
// Distinct names (last request wins); over-long/empty names fail alone. Weight = requests folded into the name.
|
||||
var byName = new Dictionary<string, (HistorianTagProvisionRequest Req, int Count)>(StringComparer.Ordinal);
|
||||
foreach (var r in requests!)
|
||||
{
|
||||
if (r is null || string.IsNullOrWhiteSpace(r.TagName) || !HistorianTagNames.IsValid(r.TagName)) { failed++; continue; }
|
||||
byName[r.TagName] = (r, byName.TryGetValue(r.TagName, out var e) ? e.Count + 1 : 1);
|
||||
}
|
||||
|
||||
var ordered = byName.OrderBy(kv => kv.Key, StringComparer.Ordinal).Select(kv => kv.Value).ToList();
|
||||
for (var i = 0; i < ordered.Count; i += ChunkSize)
|
||||
{
|
||||
var chunk = ordered.Skip(i).Take(ChunkSize).ToList();
|
||||
var weight = chunk.Sum(c => c.Count);
|
||||
if (await TryUpsertAsync(chunk.Select(c => c.Req).ToList(), ct).ConfigureAwait(false)) ensured += weight;
|
||||
else failed += weight;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogWarning("QuestDB historian provisioning failed ({Exception}).", ex.GetType().Name);
|
||||
failed = requested - ensured;
|
||||
}
|
||||
|
||||
return new HistorianProvisionResult(requested, ensured, 0, failed);
|
||||
}
|
||||
|
||||
private async Task<bool> TryUpsertAsync(List<HistorianTagProvisionRequest> chunk, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
var stored = new Dictionary<string, (string? Unit, string? Desc)>(StringComparer.Ordinal);
|
||||
await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
var sel = new StringBuilder($"SELECT name, eng_unit, description FROM {_conn.TagTable} WHERE name IN (");
|
||||
await using var cmd = _conn.CreateCommand(conn, "");
|
||||
for (var i = 0; i < chunk.Count; i++)
|
||||
{
|
||||
sel.Append(i == 0 ? "$" : ",$").Append(i + 1);
|
||||
QuestDbConnection.BindText(cmd, chunk[i].TagName);
|
||||
}
|
||||
cmd.CommandText = sel.Append(')').ToString();
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
while (await r.ReadAsync(ct).ConfigureAwait(false))
|
||||
stored[r.GetString(0)] = (r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2));
|
||||
}
|
||||
|
||||
var now = _clock.GetUtcNow().UtcDateTime;
|
||||
await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
var ins = new StringBuilder($"INSERT INTO {_conn.TagTable} (ts, name, data_type, eng_unit, description, updated_utc) VALUES ");
|
||||
await using var cmd = _conn.CreateCommand(conn, "");
|
||||
var n = 1;
|
||||
for (var i = 0; i < chunk.Count; i++)
|
||||
{
|
||||
var req = chunk[i];
|
||||
stored.TryGetValue(req.TagName, out var old);
|
||||
if (i > 0) ins.Append(',');
|
||||
ins.Append($"(${n},${n + 1},${n + 2},${n + 3},${n + 4},${n + 5})");
|
||||
n += 6;
|
||||
QuestDbConnection.BindTimestamp(cmd, QuestDbTables.TagEpoch);
|
||||
QuestDbConnection.BindText(cmd, req.TagName);
|
||||
QuestDbConnection.BindByte(cmd, (byte)req.DataType);
|
||||
QuestDbConnection.BindText(cmd, Clip(req.EngineeringUnit ?? old.Unit, MaxUnitLength));
|
||||
QuestDbConnection.BindText(cmd, Clip(req.Description ?? old.Desc, MaxDescriptionLength));
|
||||
QuestDbConnection.BindTimestamp(cmd, now);
|
||||
}
|
||||
cmd.CommandText = ins.ToString();
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await _wal.WaitAsync(_conn.TagTable, ct).ConfigureAwait(false);
|
||||
foreach (var c in chunk) _catalog.MarkKnown(c.TagName);
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogWarning("QuestDB historian tag upsert failed ({Exception}).", ex.GetType().Name);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static string? Clip(string? s, int max) => s is null ? null : s.Length <= max ? s : s[..max];
|
||||
}
|
||||
+118
@@ -0,0 +1,118 @@
|
||||
using Npgsql;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
|
||||
|
||||
[Collection(QuestDbCollection.Name)]
|
||||
public sealed class AlarmWriterTests(QuestDbFixture fx)
|
||||
{
|
||||
// µs-aligned: QuestDB stores µs, so the ts column reads back exactly.
|
||||
private static readonly DateTime T = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1230);
|
||||
|
||||
private (QuestDbAlarmHistorianWriter W, QuestDbConnection C) Make(string? connection = null, bool auto = true)
|
||||
{
|
||||
var o = fx.Options(fx.UniquePrefix(), autoCreate: auto);
|
||||
if (connection is not null) o.ConnectionString = connection;
|
||||
var c = new QuestDbConnection(o);
|
||||
return (new QuestDbAlarmHistorianWriter(c, new QuestDbSchemaBootstrapper(c, o), new QuestDbWalVisibility(c, o.VisibilityTimeout)), c);
|
||||
}
|
||||
|
||||
private static AlarmHistorianEvent Ev(string kind = "Activated", string type = "LimitAlarm", string? comment = "c", DateTime? at = null) =>
|
||||
new("alarm-1", "Plant/Line/Eq", "HiTemp", type, AlarmSeverity.High, kind, "msg", "user", comment, at ?? T);
|
||||
|
||||
private static async Task<long> CountAsync(QuestDbConnection c, string? kind = null)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn, $"SELECT count() FROM {c.EventTable}" + (kind is null ? "" : " WHERE event_kind = $1"));
|
||||
if (kind is not null) QuestDbConnection.BindText(cmd, kind);
|
||||
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Event_round_trips_and_a_repeat_is_one_row_with_every_write_Ack()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
(await w.WriteBatchAsync([Ev(), Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]);
|
||||
(await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
|
||||
(await w.WriteBatchAsync([Ev(kind: "Cleared")], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
|
||||
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn,
|
||||
$"SELECT event_id, source, alarm_id, alarm_name, alarm_type_name, severity, message, user_name, comment, ts, received_utc FROM {c.EventTable} WHERE event_kind = 'Activated'");
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
(await r.ReadAsync()).ShouldBeTrue();
|
||||
r.GetString(0).ShouldBe(AlarmEventIds.ComputeEventId(Ev()));
|
||||
r.GetString(1).ShouldBe("Plant/Line/Eq");
|
||||
r.GetString(2).ShouldBe("alarm-1");
|
||||
r.GetString(3).ShouldBe("HiTemp");
|
||||
r.GetString(4).ShouldBe("LimitAlarm");
|
||||
r.GetInt32(5).ShouldBe(AlarmEventIds.ToNumericSeverity(AlarmSeverity.High));
|
||||
r.GetInt32(5).ShouldBe(700);
|
||||
r.GetString(6).ShouldBe("msg");
|
||||
r.GetString(7).ShouldBe("user");
|
||||
r.GetString(8).ShouldBe("c");
|
||||
QuestDbConnection.ReadUtc(r, 9).ShouldBe(T);
|
||||
(DateTime.UtcNow - QuestDbConnection.ReadUtc(r, 10)).ShouldBeLessThan(TimeSpan.FromMinutes(1));
|
||||
(await r.ReadAsync()).ShouldBeFalse();
|
||||
await r.CloseAsync();
|
||||
(await CountAsync(c)).ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_null_comment_and_a_very_long_type_name_are_both_stored_and_Ack()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", type: new string('t', 500)), Ev(kind: "C", comment: null)], CancellationToken.None);
|
||||
r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]);
|
||||
(await CountAsync(c)).ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_event_before_1970_is_PermanentFail_and_siblings_still_Ack()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", at: new DateTime(1960, 1, 1, 0, 0, 0, DateTimeKind.Utc)), Ev(kind: "C")], CancellationToken.None);
|
||||
r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]);
|
||||
(await CountAsync(c)).ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Concurrent_writers_of_the_same_event_all_Ack_with_one_row()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
await w.WriteBatchAsync([Ev(kind: "warm")], CancellationToken.None);
|
||||
var rs = await Task.WhenAll(Enumerable.Range(0, 12).Select(_ => Task.Run(() => w.WriteBatchAsync([Ev()], CancellationToken.None))));
|
||||
rs.ShouldAllBe(x => x[0] == HistorianWriteOutcome.Ack);
|
||||
(await CountAsync(c, "Activated")).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Unreachable_database_is_RetryPlease_not_an_exception_and_empty_batch_is_empty()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, _) = Make("Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2", auto: false);
|
||||
(await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.RetryPlease]);
|
||||
(await w.WriteBatchAsync([], CancellationToken.None)).ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Classify_maps_unknown_and_transport_errors_to_RetryPlease_and_data_rejections_to_PermanentFail()
|
||||
{
|
||||
QuestDbAlarmHistorianWriter.Classify(new InvalidOperationException("boom")).ShouldBe(HistorianWriteOutcome.RetryPlease);
|
||||
QuestDbAlarmHistorianWriter.Classify(new IOException("reset")).ShouldBe(HistorianWriteOutcome.RetryPlease);
|
||||
QuestDbAlarmHistorianWriter.Classify(new TimeoutException()).ShouldBe(HistorianWriteOutcome.RetryPlease);
|
||||
QuestDbAlarmHistorianWriter.Classify(new PostgresException("table is busy", "ERROR", "ERROR", "00000")).ShouldBe(HistorianWriteOutcome.RetryPlease);
|
||||
QuestDbAlarmHistorianWriter.Classify(new PostgresException("designated timestamp before 1970-01-01 is not allowed", "ERROR", "ERROR", "00000"))
|
||||
.ShouldBe(HistorianWriteOutcome.PermanentFail);
|
||||
QuestDbAlarmHistorianWriter.Classify(new PostgresException("inconvertible value: x [VARCHAR -> INT]", "ERROR", "ERROR", "00000"))
|
||||
.ShouldBe(HistorianWriteOutcome.PermanentFail);
|
||||
}
|
||||
}
|
||||
+102
@@ -0,0 +1,102 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
|
||||
|
||||
[Collection(QuestDbCollection.Name)]
|
||||
public sealed class ProvisioningTests(QuestDbFixture fx)
|
||||
{
|
||||
private (QuestDbTagProvisioner P, QuestDbConnection C, QuestDbTagCatalog Cat) Make(string? prefix = null, string? connectionString = null,
|
||||
bool autoCreate = true)
|
||||
{
|
||||
var o = fx.Options(prefix ?? fx.UniquePrefix(), autoCreate: autoCreate);
|
||||
if (connectionString is not null) o.ConnectionString = connectionString;
|
||||
var c = new QuestDbConnection(o);
|
||||
var b = new QuestDbSchemaBootstrapper(c, o);
|
||||
var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
|
||||
var cat = new QuestDbTagCatalog(c, b, wal);
|
||||
return (new QuestDbTagProvisioner(c, b, cat, wal), c, cat);
|
||||
}
|
||||
|
||||
private static async Task<(byte Type, string? Eu, string? Desc)> TagAsync(QuestDbConnection c, string name)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn, $"SELECT data_type, eng_unit, description FROM {c.TagTable} WHERE name = $1");
|
||||
QuestDbConnection.BindText(cmd, name);
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
(await r.ReadAsync()).ShouldBeTrue();
|
||||
return (QuestDbConnection.ReadKind(r, 0), r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2));
|
||||
}
|
||||
|
||||
private static async Task<long> CountAsync(QuestDbConnection c)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn, $"SELECT count() FROM {c.TagTable}");
|
||||
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Upsert_creates_then_changes_DataType_on_rerun_and_every_type_is_accepted()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, c, _) = Make();
|
||||
var r1 = await p.EnsureTagsAsync([new("a", DriverDataType.Boolean, "degC", "desc"), new("s", DriverDataType.String, null, null)], CancellationToken.None);
|
||||
r1.ShouldBe(new HistorianProvisionResult(2, 2, 0, 0));
|
||||
(await TagAsync(c, "a")).ShouldBe(((byte)DriverDataType.Boolean, "degC", "desc"));
|
||||
var r2 = await p.EnsureTagsAsync([new("a", DriverDataType.Float64, null, null)], CancellationToken.None);
|
||||
r2.ShouldBe(new HistorianProvisionResult(1, 1, 0, 0));
|
||||
(await TagAsync(c, "a")).ShouldBe(((byte)DriverDataType.Float64, "degC", "desc")); // null keeps stored metadata
|
||||
(await CountAsync(c)).ShouldBe(2); // upsert by name, not a second row
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_request_with_values_replaces_only_the_fields_it_carries()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, c, _) = Make();
|
||||
await p.EnsureTagsAsync([new("m", DriverDataType.Int32, "A", "first")], CancellationToken.None);
|
||||
await p.EnsureTagsAsync([new("m", DriverDataType.Int32, "B", null)], CancellationToken.None);
|
||||
(await TagAsync(c, "m")).ShouldBe(((byte)DriverDataType.Int32, "B", "first"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Duplicate_names_and_bad_names_are_tallied_not_thrown()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, _, _) = Make();
|
||||
var r = await p.EnsureTagsAsync([
|
||||
new("d", DriverDataType.Int32, null, null), new("d", DriverDataType.Int64, null, null),
|
||||
new("", DriverDataType.Int32, null, null), new(new string('x', 401), DriverDataType.Int32, null, null)], CancellationToken.None);
|
||||
r.ShouldBe(new HistorianProvisionResult(4, 2, 0, 2));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task More_tags_than_one_chunk_all_land()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, c, _) = Make();
|
||||
var reqs = Enumerable.Range(0, 1000).Select(i => new HistorianTagProvisionRequest($"bulk{i}", DriverDataType.Float32, null, null)).ToList();
|
||||
(await p.EnsureTagsAsync(reqs, CancellationToken.None)).ShouldBe(new HistorianProvisionResult(1000, 1000, 0, 0));
|
||||
(await CountAsync(c)).ShouldBe(1000);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_later_catalog_Ensure_of_a_provisioned_tag_does_not_wipe_its_metadata()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, c, cat) = Make();
|
||||
await p.EnsureTagsAsync([new("k", DriverDataType.Float64, "kg", "mass")], CancellationToken.None);
|
||||
(await cat.ExistsAsync("k", CancellationToken.None)).ShouldBeTrue();
|
||||
await cat.EnsureAsync("k", 1, CancellationToken.None);
|
||||
(await TagAsync(c, "k")).ShouldBe(((byte)DriverDataType.Float64, "kg", "mass"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_connection_failure_is_a_Failed_tally_not_an_exception()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, _, _) = Make(connectionString: "Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2");
|
||||
(await p.EnsureTagsAsync([new("x", DriverDataType.Int32, null, null)], CancellationToken.None)).ShouldBe(new HistorianProvisionResult(1, 0, 0, 1));
|
||||
}
|
||||
}
|
||||
+296
@@ -0,0 +1,296 @@
|
||||
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.QuestDb.IntegrationTests;
|
||||
|
||||
[Collection(QuestDbCollection.Name)]
|
||||
public sealed class WriteTests(QuestDbFixture fx)
|
||||
{
|
||||
// µs-aligned (ticks % 10 == 0) so "same µs" offsets of a few ticks stay inside one 10-tick window.
|
||||
private static readonly DateTime T0 = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc);
|
||||
private static readonly DateTime ServerNow = new(2026, 10, 9, 12, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private (QuestDbHistorianValueWriter W, QuestDbConnection C) Make(string? prefix = null, string? connectionString = null)
|
||||
{
|
||||
var o = fx.Options(prefix ?? fx.UniquePrefix());
|
||||
if (connectionString is not null) o.ConnectionString = connectionString;
|
||||
var c = new QuestDbConnection(o);
|
||||
var b = new QuestDbSchemaBootstrapper(c, o);
|
||||
var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
|
||||
return (new QuestDbHistorianValueWriter(c, new QuestDbTagCatalog(c, b, wal), wal, utcNow: () => ServerNow), c);
|
||||
}
|
||||
|
||||
private static async Task<List<(DateTime T, byte Kind, double? N, long? I, string? S, long Status, bool Deleted, DateTime ServerTs)>> RowsAsync(
|
||||
QuestDbConnection c, string tag)
|
||||
{
|
||||
await using var conn = await c.OpenAsync(default);
|
||||
await using var cmd = c.CreateCommand(conn,
|
||||
$"SELECT ts, kind, num_value, int_value, str_value, status, deleted, server_ts FROM {c.SampleTable} WHERE tag = $1 ORDER BY ts");
|
||||
QuestDbConnection.BindText(cmd, tag);
|
||||
var l = new List<(DateTime, byte, double?, long?, string?, long, bool, DateTime)>();
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
while (await r.ReadAsync())
|
||||
l.Add((QuestDbConnection.ReadUtc(r, 0), QuestDbConnection.ReadKind(r, 1), r.IsDBNull(2) ? null : r.GetDouble(2),
|
||||
r.IsDBNull(3) ? null : r.GetInt64(3), r.IsDBNull(4) ? null : r.GetString(4), r.GetInt64(5), r.GetBoolean(6),
|
||||
QuestDbConnection.ReadUtc(r, 7)));
|
||||
return l;
|
||||
}
|
||||
|
||||
private static async Task SeedTombstoneAsync(QuestDbConnection c, QuestDbWalVisibility wal, string tag, DateTime t)
|
||||
{
|
||||
await using (var conn = await c.OpenAsync(default))
|
||||
{
|
||||
await using var cmd = c.CreateCommand(conn,
|
||||
$"INSERT INTO {c.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ($1,$2,$3,NULL,NULL,NULL,$4,true,$5)");
|
||||
QuestDbConnection.BindTimestamp(cmd, t);
|
||||
QuestDbConnection.BindText(cmd, tag);
|
||||
QuestDbConnection.BindByte(cmd, 0);
|
||||
QuestDbConnection.BindLong(cmd, 0);
|
||||
QuestDbConnection.BindTimestamp(cmd, t);
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
await wal.WaitAsync(c.SampleTable, default);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task All_six_kinds_round_trip_with_a_non_Good_status()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
const uint uncertain = HistorianStatusCodes.UncertainDataSubNormal;
|
||||
const uint bad = HistorianStatusCodes.BadOutOfRange;
|
||||
var dt = new DateTime(2025, 5, 6, 7, 8, 9, DateTimeKind.Utc).AddTicks(1230);
|
||||
HistorizationValue[] vals =
|
||||
[
|
||||
new(T0, HistorianValue.FromDouble(1.5), uncertain),
|
||||
new(T0.AddSeconds(1), HistorianValue.FromInt64(-77), uncertain),
|
||||
new(T0.AddSeconds(2), HistorianValue.FromBoolean(true), uncertain),
|
||||
new(T0.AddSeconds(3), HistorianValue.FromString("héllo"), uncertain),
|
||||
new(T0.AddSeconds(4), HistorianValue.FromDateTime(dt), uncertain),
|
||||
new(T0.AddSeconds(5), HistorianValue.Empty, bad),
|
||||
];
|
||||
(await w.WriteLiveValuesAsync("tag.kinds", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
|
||||
var rows = await RowsAsync(c, "tag.kinds");
|
||||
rows.Count.ShouldBe(6);
|
||||
for (var i = 0; i < 6; i++)
|
||||
{
|
||||
rows[i].T.ShouldBe(vals[i].TimestampUtc!.Value);
|
||||
rows[i].Status.ShouldBe(SampleColumnMapper.ToLongStatus(vals[i].StatusCode));
|
||||
rows[i].Deleted.ShouldBeFalse();
|
||||
rows[i].ServerTs.ShouldBe(ServerNow);
|
||||
SampleColumnMapper.FromColumns(rows[i].Kind, rows[i].N, rows[i].I, rows[i].S).ShouldBe(vals[i].Value);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_redelivered_batch_leaves_one_row_and_the_first_value()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
(await w.WriteLiveValuesAsync("tag.redeliver", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
(await w.WriteLiveValuesAsync("tag.redeliver", [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(c, "tag.redeliver");
|
||||
rows.Count.ShouldBe(1);
|
||||
rows[0].N.ShouldBe(1d);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Redelivery_straight_after_the_write_always_sees_it_because_the_writer_waits_for_WAL_visibility()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
// Without the visibility wait ~35 % of immediate re-reads miss the row (T0 Q4) and the second value would replace the first.
|
||||
for (var i = 0; i < 40; i++)
|
||||
{
|
||||
var t = T0.AddSeconds(i);
|
||||
(await w.WriteLiveValuesAsync("tag.rw", [new(t, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
(await w.WriteLiveValuesAsync("tag.rw", [new(t, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
}
|
||||
var rows = await RowsAsync(c, "tag.rw");
|
||||
rows.Count.ShouldBe(40);
|
||||
rows.ShouldAllBe(r => r.N == 1d);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_tombstoned_key_is_not_resurrected_by_redelivery()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5));
|
||||
(await w.WriteLiveValuesAsync("tag.tomb", [new(T0, HistorianValue.FromDouble(9), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
await SeedTombstoneAsync(c, wal, "tag.tomb", T0.AddSeconds(1));
|
||||
|
||||
// Redeliver the tombstoned key plus a fresh neighbour.
|
||||
(await w.WriteLiveValuesAsync("tag.tomb",
|
||||
[new(T0.AddSeconds(1), HistorianValue.FromDouble(5), 0), new(T0.AddSeconds(2), HistorianValue.FromDouble(6), 0)],
|
||||
CancellationToken.None)).ShouldBeTrue();
|
||||
|
||||
var rows = await RowsAsync(c, "tag.tomb");
|
||||
rows.Count.ShouldBe(3);
|
||||
rows.Single(r => r.T == T0.AddSeconds(1)).Deleted.ShouldBeTrue();
|
||||
rows.Single(r => r.T == T0.AddSeconds(2)).N.ShouldBe(6d);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task NaN_is_stored_as_Empty_with_BadOutOfRange()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
(await w.WriteLiveValuesAsync("tag.nan", [new(T0, HistorianValue.FromDouble(double.NaN), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(c, "tag.nan");
|
||||
rows.Count.ShouldBe(1);
|
||||
rows[0].Kind.ShouldBe((byte)HistorianValueKind.Empty);
|
||||
rows[0].Status.ShouldBe((long)HistorianStatusCodes.BadOutOfRange);
|
||||
rows[0].N.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_microsecond_collision_within_a_batch_keeps_the_first()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
// +2 and +7 ticks stay inside one 10-tick (1 µs) window of the aligned T0.
|
||||
(await w.WriteLiveValuesAsync("tag.us",
|
||||
[new(T0.AddTicks(2), HistorianValue.FromDouble(1), 0), new(T0.AddTicks(7), HistorianValue.FromDouble(2), 0)],
|
||||
CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(c, "tag.us");
|
||||
rows.Count.ShouldBe(1);
|
||||
rows[0].N.ShouldBe(1d);
|
||||
rows[0].T.ShouldBe(T0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Samples_before_1970_are_dropped_and_the_rest_still_land()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var old = new DateTime(1960, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
(await w.WriteLiveValuesAsync("tag.old",
|
||||
[new(old, HistorianValue.FromDouble(1), 0), new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
w.DroppedPre1970.ShouldBe(1);
|
||||
var rows = await RowsAsync(c, "tag.old");
|
||||
rows.Count.ShouldBe(1);
|
||||
rows[0].T.ShouldBe(T0);
|
||||
|
||||
// A batch of nothing but unstorable samples succeeds (nothing to retry) and stores nothing.
|
||||
(await w.WriteLiveValuesAsync("tag.old2", [new(old, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
w.DroppedPre1970.ShouldBe(2);
|
||||
(await RowsAsync(c, "tag.old2")).ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_batch_larger_than_one_insert_chunk_lands_completely()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var vals = Enumerable.Range(0, 2500).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromInt64(i), 0)).ToArray();
|
||||
(await w.WriteLiveValuesAsync("tag.big", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(c, "tag.big");
|
||||
rows.Count.ShouldBe(2500);
|
||||
rows[1999].I.ShouldBe(1999);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_over_long_tag_name_returns_false()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, _) = Make();
|
||||
(await w.WriteLiveValuesAsync(new string('x', 401), [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_stopped_port_returns_false_without_throwing()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, _) = Make(connectionString: "Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2");
|
||||
(await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_cancelled_write_returns_false_without_throwing()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, _) = Make();
|
||||
using var cts = new CancellationTokenSource();
|
||||
cts.Cancel();
|
||||
(await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], cts.Token)).ShouldBeFalse();
|
||||
}
|
||||
|
||||
private static async Task SeedRowAsync(QuestDbConnection c, QuestDbWalVisibility wal, string tag, DateTime t, double v)
|
||||
{
|
||||
await using (var conn = await c.OpenAsync(default))
|
||||
{
|
||||
await using var cmd = c.CreateCommand(conn,
|
||||
$"INSERT INTO {c.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ($1,$2,1,$3,NULL,NULL,0,false,$4)");
|
||||
QuestDbConnection.BindTimestamp(cmd, t);
|
||||
QuestDbConnection.BindText(cmd, tag);
|
||||
cmd.Parameters.Add(new Npgsql.NpgsqlParameter { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Double, Value = v });
|
||||
QuestDbConnection.BindTimestamp(cmd, t);
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
await wal.WaitAsync(c.SampleTable, default);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_row_on_another_tag_at_the_same_timestamp_does_not_suppress_the_write()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5));
|
||||
(await w.WriteLiveValuesAsync("tag.A", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
(await w.WriteLiveValuesAsync("tag.B", [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
(await RowsAsync(c, "tag.A")).Single().N.ShouldBe(1d);
|
||||
(await RowsAsync(c, "tag.B")).Single().N.ShouldBe(2d);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_absent_interior_timestamp_is_written_while_both_present_ends_are_left_alone()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5));
|
||||
await w.WriteLiveValuesAsync("tag.ends", [new(T0.AddSeconds(100), HistorianValue.FromDouble(0), 0)], CancellationToken.None); // creates the tag/tables
|
||||
await SeedRowAsync(c, wal, "tag.ends", T0, 10);
|
||||
await SeedRowAsync(c, wal, "tag.ends", T0.AddSeconds(20), 30);
|
||||
(await w.WriteLiveValuesAsync("tag.ends",
|
||||
[new(T0, HistorianValue.FromDouble(-1), 0), new(T0.AddSeconds(10), HistorianValue.FromDouble(20), 0), new(T0.AddSeconds(20), HistorianValue.FromDouble(-3), 0)],
|
||||
CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = (await RowsAsync(c, "tag.ends")).Where(r => r.T <= T0.AddSeconds(20)).ToList();
|
||||
rows.Select(r => r.N).ShouldBe([10d, 20d, 30d]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_outlier_timestamp_does_not_widen_the_existence_read_over_the_stored_rows_between()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, c) = Make();
|
||||
// Behavioural check of the whole path with a far-apart batch: both samples land, a redelivery adds nothing.
|
||||
var far = T0.AddYears(1);
|
||||
HistorizationValue[] vals = [new(T0, HistorianValue.FromDouble(1), 0), new(far, HistorianValue.FromDouble(2), 0)];
|
||||
(await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
(await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
(await RowsAsync(c, "tag.far")).Count.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Cluster_splits_on_a_gap_over_the_limit_and_at_the_per_cluster_cap()
|
||||
{
|
||||
var g = TimeSpan.FromMinutes(1);
|
||||
QuestDbHistorianValueWriter.Cluster([], g, 10).ShouldBeEmpty();
|
||||
QuestDbHistorianValueWriter.Cluster([T0], g, 10).ShouldBe([(T0, T0)]);
|
||||
// exactly the gap stays together; one tick over splits
|
||||
QuestDbHistorianValueWriter.Cluster([T0, T0 + g], g, 10).ShouldBe([(T0, T0 + g)]);
|
||||
QuestDbHistorianValueWriter.Cluster([T0, T0 + g + TimeSpan.FromTicks(10)], g, 10)
|
||||
.ShouldBe([(T0, T0), (T0 + g + TimeSpan.FromTicks(10), T0 + g + TimeSpan.FromTicks(10))]);
|
||||
// outlier gets its own range
|
||||
var far = T0.AddYears(1);
|
||||
QuestDbHistorianValueWriter.Cluster([T0, T0.AddSeconds(1), far], g, 10).ShouldBe([(T0, T0.AddSeconds(1)), (far, far)]);
|
||||
// cap: 5 close times, max 2 per cluster
|
||||
var five = Enumerable.Range(0, 5).Select(i => T0.AddSeconds(i)).ToList();
|
||||
QuestDbHistorianValueWriter.Cluster(five, g, 2)
|
||||
.ShouldBe([(five[0], five[1]), (five[2], five[3]), (five[4], five[4])]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user