dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(historian): QuestDB reads and aggregates (pushed down, shared evaluation)

Browse Source 
QuestDbHistorianDataSource serves Raw (LIMIT $n, no continuation, R8), AtTime
(chunked LIMIT 1 neighbour seeks, <=50 times per statement), Events and
Processed (one statement per node, bucket = (cast(ts AS LONG)*10 - start)/interval,
shared BucketPlan/AggregateEvaluator/AtTimeEvaluator). Every sample read filters
NOT deleted; every row goes through StoredSample.FromStored; RC2 long.MinValue
read repair. QuestDbConformanceBackend (seed/read-back) for the conformance suite;
writer/updater factories are T7 placeholders.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 02:13:00 -04:00
parent ecc0f914cc
commit c4b4b2e2bc
4 changed files with 1059 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs
+122
View File
@@ -0,0 +1,122 @@
using System.Globalization;
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>
/// Push-down QuestDB SQL for processed (aggregate) reads (spec §3 "Processed", T0 Q8-confirmed forms), one query
/// per <see cref="AggregateFamily"/>, mirroring the SQL Server backend's <c>SqlAggregateQueries</c> family split and
/// column order. Every query takes five positional parameters:
/// <list type="number">
/// <item><c>$1</c> — the (rounded) window start in 100 ns ticks since the Unix epoch
/// (<see cref="BucketPlan.StartUnixTicks"/>, <c>Bigint</c>);</item>
/// <item><c>$2</c> — the bucket width in ticks (<see cref="BucketPlan.IntervalTicks"/>, <c>Bigint</c>);</item>
/// <item><c>$3</c> — the tag (<c>Varchar</c>);</item>
/// <item><c>$4</c>/<c>$5</c> — the forward window <c>[start, end)</c> (<c>TimestampTz</c> via
/// <see cref="QuestDbConnection.BindTimestamp"/>).</item>
/// </list>
/// Each returns one row per <b>non-empty</b> bucket; the caller fills the empty ones.
/// </summary>
/// <remarks>
/// <para>
/// <b>Bucketing.</b> <c>b = (cast(ts AS LONG) * 10 − $1) / $2</c>. <c>cast(ts AS LONG)</c> is µs since 1970, so
/// <c>× 10</c> is ticks since 1970; both it and <c>$1</c> are µs-aligned (R6), so the integer division is
/// start-inclusive / end-exclusive to the tick, exactly like SQL Server's. Every row is at or after the start,
/// so the dividend is never negative (no truncation-toward-zero surprise), and nothing overflows before 9999.
/// <c>SAMPLE BY</c> is deliberately not used: its alignment rules differ from ours.
/// </para>
/// <para>
/// <b>Sample classes (S6).</b> The status is stored as the <b>unsigned</b> 32-bit code in a <c>LONG</c>, so
/// "not Bad" is <c>status &lt; 0x80000000</c> (not T-SQL's sign test). <i>usable</i> = not Bad AND
/// <c>kind &lt;&gt; 0</c> (an Empty sample is an outage marker and counts as Bad whatever its code);
/// <i>good</i> = usable AND <c>status &lt; 0x40000000</c>; <i>bad</i> = not usable; <i>non-numeric</i> = usable
/// String/DateTime. The literals are generated from <see cref="StatusBits"/> and <see cref="HistorianValueKind"/>.
/// </para>
/// <para>Common result columns (ordinals 0–5): <c>b, total_count, usable_count, good_count, bad_count,
/// non_numeric_count</c>. Family columns follow from <see cref="FirstFamilyColumn"/>.</para>
/// </remarks>
internal static class QuestDbAggregateQueries
{
/// <summary>Ordinal of the first family-specific column.</summary>
public const int FirstFamilyColumn = 6;
/// <summary><c>status &lt; SeverityBadBit</c>, as a decimal LONG literal.</summary>
internal static readonly string NotBadLimit = StatusBits.SeverityBadBit.ToString(CultureInfo.InvariantCulture);
/// <summary><c>status &lt; SeverityUncertainBit</c>, as a decimal LONG literal.</summary>
internal static readonly string GoodLimit = StatusBits.SeverityUncertainBit.ToString(CultureInfo.InvariantCulture);
/// <summary>The <c>kind</c> values of String and DateTime: usable but not numeric.</summary>
internal static readonly string NonNumericKinds =
((byte)HistorianValueKind.String).ToString(CultureInfo.InvariantCulture) + ", " +
((byte)HistorianValueKind.DateTime).ToString(CultureInfo.InvariantCulture);
private static readonly string EmptyKind = ((byte)HistorianValueKind.Empty).ToString(CultureInfo.InvariantCulture);
/// <summary>
/// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad.
/// Family columns: <c>6 avg_value, 7 min_value, 8 max_value, 9 sum_value</c> (DOUBLE, NULL when no usable sample).
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <returns>The SQL text.</returns>
public static string Simple(string sample) =>
"SELECT b, " + CommonCounts() + ", " +
"avg(CASE WHEN u = 1 THEN num_value END) AS avg_value, " +
"min(CASE WHEN u = 1 THEN num_value END) AS min_value, " +
"max(CASE WHEN u = 1 THEN num_value END) AS max_value, " +
"sum(CASE WHEN u = 1 THEN num_value END) AS sum_value " +
"FROM (" + Classified(sample) + ") GROUP BY b ORDER BY b";
/// <summary>
/// Start / End / Delta: the first and last usable sample of each bucket by <c>ts</c>.
/// Family columns: <c>6..10</c> first sample (<c>kind, num_value, int_value, str_value, ts</c>), <c>11..15</c> last
/// sample (same order). A bucket without a usable sample has NULL <c>ts</c> but <c>kind = 0</c> (BYTE has no NULL),
/// so the reader keys absence on the NULL time.
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <returns>The SQL text.</returns>
public static string StartEnd(string sample) =>
"WITH s AS (" + Classified(sample) + "), " +
"st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b), " +
"f AS (SELECT b, kind, num_value, int_value, str_value, ts, " +
"row_number() OVER (PARTITION BY b ORDER BY ts ASC) AS rf, " +
"row_number() OVER (PARTITION BY b ORDER BY ts DESC) AS rl FROM s WHERE u = 1) " +
"SELECT st.*, " +
"x.kind, x.num_value, x.int_value, x.str_value, x.ts, " +
"y.kind, y.num_value, y.int_value, y.str_value, y.ts " +
"FROM st LEFT JOIN (SELECT * FROM f WHERE rf = 1) x ON x.b = st.b " +
"LEFT JOIN (SELECT * FROM f WHERE rl = 1) y ON y.b = st.b ORDER BY st.b";
/// <summary>
/// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) <c>num_value</c>, ties
/// broken by the earliest <c>ts</c>. Family columns: <c>6 num_value, 7 ts</c> (NULL when the bucket has no usable
/// numeric sample).
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <param name="maximum">True for MaximumActualTime.</param>
/// <returns>The SQL text.</returns>
public static string ActualTime(string sample, bool maximum) =>
"WITH s AS (" + Classified(sample) + "), " +
"st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b), " +
"r AS (SELECT b, num_value, ts, " +
"row_number() OVER (PARTITION BY b ORDER BY num_value " + (maximum ? "DESC" : "ASC") + ", ts ASC) AS rn " +
"FROM s WHERE u = 1 AND num_value IS NOT NULL) " +
"SELECT st.*, r.num_value, r.ts " +
"FROM st LEFT JOIN (SELECT * FROM r WHERE rn = 1) r ON r.b = st.b ORDER BY st.b";
/// <summary>The bucket-index expression (exposed for tests).</summary>
internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2";
/// <summary>The classified subquery <c>s</c>: the window's live samples with bucket index and the usable/good flags.</summary>
private static string Classified(string sample) =>
"SELECT " + BucketExpression + " AS b, ts, kind, num_value, int_value, str_value, " +
$"CASE WHEN status < {NotBadLimit} AND kind <> {EmptyKind} THEN 1 ELSE 0 END AS u, " +
$"CASE WHEN status < {GoodLimit} AND kind <> {EmptyKind} THEN 1 ELSE 0 END AS g " +
$"FROM {sample} WHERE tag = $3 AND ts >= $4 AND ts < $5 AND NOT deleted";
/// <summary>The five common count columns (after <c>b</c>).</summary>
private static string CommonCounts() =>
"count() AS total_count, sum(u) AS usable_count, sum(g) AS good_count, count() - sum(u) AS bad_count, " +
$"sum(CASE WHEN u = 1 AND kind IN ({NonNumericKinds}) THEN 1 ELSE 0 END) AS non_numeric_count";
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs
+468
View File
@@ -0,0 +1,468 @@
using System.Globalization;
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>
/// QuestDB read side of the historian (spec §3 "Reads"): Raw (forward / reverse / point), AtTime (chunked
/// <c>LIMIT 1</c> neighbour seeks, T0 Q9), Events, and Processed aggregates pushed down to QuestDB SQL
/// (<see cref="QuestDbAggregateQueries"/>, T0 Q8). Results match the SQL Server data source except that every
/// timestamp is at microsecond resolution.
/// </summary>
/// <remarks>
/// <para>
/// <b>Timestamps (R6).</b> Every inbound time is normalised with <see cref="QuestDbConnection.Round"/> and every
/// <see cref="DateTime"/> is bound only through <see cref="QuestDbConnection.BindTimestamp"/>. Read bounds are not
/// clamped to the epoch (a pre-1970 bound is valid); only <c>MinValue</c>/<c>MaxValue</c> are nudged. Results
/// carry the rounded times, so an AtTime result is stamped at the rounded request time.
/// </para>
/// <para>
/// <b>Reads never create tags (R7).</b> A tag resolves only through <see cref="QuestDbTagCatalog.ExistsAsync"/>;
/// an unknown tag is a successful empty read (Raw), an all-<c>BadNoData</c> read (AtTime / Processed), never an
/// error. Every sample query filters <c>NOT deleted</c>, so a DeleteRaw/DeleteAtTime tombstone is invisible.
/// </para>
/// <para>
/// <b>Values.</b> Every row goes through <see cref="StoredSample.FromStored"/> (an Empty row with a non-Bad status
/// is served <c>BadNoData</c>). The value kind, not which column is populated, decides the value. QuestDB stores
/// <see cref="long.MinValue"/> in a <c>LONG</c> as NULL (its NULL sentinel), so an Int64 row whose
/// <c>int_value</c> is NULL and whose <c>num_value</c> is exactly −2^63 is served as <see cref="long.MinValue"/>
/// (ruling RC2).
/// </para>
/// <para>
/// <b>Faults.</b> As in the SQL Server data source: a database / bootstrap exception is recorded in
/// <see cref="GetHealthSnapshot"/> and rethrown, failing only that call. Client-request errors — an unsupported
/// aggregate (<see cref="HistoryAggregateNotSupportedException"/>) and a processed window over
/// <see cref="BucketPlan.MaxBuckets"/> intervals (<see cref="ArgumentOutOfRangeException"/>) — are thrown before
/// any I/O and are not counted as backend failures.
/// </para>
/// </remarks>
public sealed class QuestDbHistorianDataSource : IHistorianDataSource
{
/// <summary>Most AtTime request times per statement (two <c>UNION ALL</c> branches each; QuestDB 10.0.1 refuses more than 126 branches).</summary>
public const int AtTimeChunkSize = 50;
/// <summary><see cref="long.MinValue"/> as a double (−2^63, exact): what an Int64 sample of that value leaves in <c>num_value</c>.</summary>
internal const double LongNullSentinelAsDouble = -9223372036854775808d;
private const string SampleColumns = "ts, kind, num_value, int_value, str_value, status, server_ts";
private readonly QuestDbConnection _conn;
private readonly QuestDbSchemaBootstrapper _bootstrap;
private readonly QuestDbTagCatalog _catalog;
private readonly ILogger _log;
private readonly TimeProvider _time;
private readonly HistorianHealthCounters _health = new();
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
/// <param name="conn">Connection helper.</param>
/// <param name="bootstrap">Schema bootstrapper (shared with the other QuestDB seams).</param>
/// <param name="catalog">Tag catalog (reads use only its read-only <see cref="QuestDbTagCatalog.ExistsAsync"/>).</param>
/// <param name="log">Optional logger.</param>
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
internal QuestDbHistorianDataSource(
QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, ILogger? log = null,
TimeProvider? timeProvider = null)
{
_conn = conn ?? throw new ArgumentNullException(nameof(conn));
_bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap));
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
_log = log ?? NullLogger.Instance;
_time = timeProvider ?? TimeProvider.System;
}
// ------------------------------------------------------------------------------------------ Raw
/// <inheritdoc />
/// <remarks>
/// Forward (<c>start &lt; end</c>): <c>[start, end)</c> ascending. Reverse (<c>end &lt; start</c>):
/// <c>(end, start]</c> descending. <c>start == end</c>: the sample exactly at that (rounded) time.
/// <paramref name="maxValuesPerNode"/> <c>0</c> is "unbounded" (<see cref="int.MaxValue"/>); larger values clamp.
/// At most <c>top</c> samples, <c>ContinuationPoint</c> always null (R8): the node manager pages by re-reading
/// from the last returned timestamp inclusive, which the <c>(ts, tag)</c> dedup key makes exact.
/// </remarks>
public async Task<HistoryReadResult> ReadRawAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReference);
var top = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue);
try
{
var samples = new List<DataValueSnapshot>();
if (await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
{
var start = QuestDbConnection.Round(startUtc);
var end = QuestDbConnection.Round(endUtc);
string where, order;
if (start == end) { where = "ts = $2 AND ts = $3"; order = "ASC"; }
else if (start < end) { where = "ts >= $2 AND ts < $3"; order = "ASC"; }
else { where = "ts > $2 AND ts <= $3"; order = "DESC"; }
var (lo, hi) = start <= end ? (start, end) : (end, start);
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn,
$"SELECT {SampleColumns} FROM {_conn.SampleTable} " +
$"WHERE tag = $1 AND {where} AND NOT deleted ORDER BY ts {order} LIMIT $4");
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, lo);
QuestDbConnection.BindTimestamp(cmd, hi);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = top });
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
var now = _time.GetUtcNow().UtcDateTime;
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
var s = ReadStored(reader, 0);
samples.Add(new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc, ReadServerTime(reader, 6, now)));
}
}
RecordSuccess();
return new HistoryReadResult(samples, ContinuationPoint: null);
}
catch (Exception ex)
{
RecordFailure(ex);
throw;
}
}
// ------------------------------------------------------------------------------------------ AtTime
/// <inheritdoc />
/// <remarks>
/// Each distinct (rounded) requested time picks up its previous-or-equal and next live sample through two
/// <c>LIMIT 1</c> seeks, <c>UNION ALL</c>-ed, at most <see cref="AtTimeChunkSize"/> times per statement (spec §3,
/// T0 Q9). <see cref="AtTimeEvaluator.Evaluate"/> then decides each value exactly as for SQL Server (exact hit,
/// no previous ⇒ <c>BadNoData</c>, inside an outage, held before an outage, linear, stepped, after now). The
/// result matches the request in length and order (duplicates included), each entry stamped at its rounded
/// time. Unknown tag ⇒ every entry <c>BadNoData</c>.
/// </remarks>
public async Task<HistoryReadResult> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReference);
ArgumentNullException.ThrowIfNull(timestampsUtc);
try
{
var now = _time.GetUtcNow().UtcDateTime;
var rounded = timestampsUtc.Select(QuestDbConnection.Round).ToArray();
var byTicks = new Dictionary<long, DataValueSnapshot>(rounded.Length);
if (rounded.Length > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
{
var distinct = rounded.Distinct().ToList();
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
for (var offset = 0; offset < distinct.Count; offset += AtTimeChunkSize)
{
var chunk = distinct.GetRange(offset, Math.Min(AtTimeChunkSize, distinct.Count - offset));
var prev = new StoredSample?[chunk.Count];
var next = new StoredSample?[chunk.Count];
var prevServer = new DateTime?[chunk.Count];
await using (var cmd = _conn.CreateCommand(conn, AtTimeSql(_conn.SampleTable, chunk.Count)))
{
QuestDbConnection.BindText(cmd, fullReference);
foreach (var t in chunk) QuestDbConnection.BindTimestamp(cmd, t);
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
var k = Convert.ToInt32(reader.GetValue(0), CultureInfo.InvariantCulture);
var dir = Convert.ToInt32(reader.GetValue(1), CultureInfo.InvariantCulture);
var s = ReadStored(reader, 2);
if (dir == 0)
{
prev[k] = s;
prevServer[k] = reader.IsDBNull(8) ? null : QuestDbConnection.ReadUtc(reader, 8);
}
else
{
next[k] = s;
}
}
}
for (var k = 0; k < chunk.Count; k++)
{
var t = chunk[k];
var serverTime = prev[k] is { } ps && ps.TimeUtc == t ? prevServer[k] ?? now : now;
byTicks[t.Ticks] = AtTimeEvaluator.Evaluate(t, prev[k], next[k], serverTime, now);
}
}
}
var result = new DataValueSnapshot[rounded.Length];
for (var i = 0; i < result.Length; i++)
{
var t = rounded[i];
result[i] = byTicks.TryGetValue(t.Ticks, out var v)
? v
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, t, now);
}
RecordSuccess();
return new HistoryReadResult(result, ContinuationPoint: null);
}
catch (Exception ex)
{
RecordFailure(ex);
throw;
}
}
/// <summary>
/// The T0 Q9 statement for <paramref name="count"/> request times: <c>$1</c> = tag, <c>$(k+2)</c> = time <c>k</c>.
/// Columns: <c>req, dir</c> (0 = previous-or-equal, 1 = next), then <c>ts, kind, num_value, int_value, str_value,
/// status, server_ts</c>. <c>req</c>/<c>dir</c> are generated integers, never user input.
/// </summary>
internal static string AtTimeSql(string sample, int count)
{
if (count is < 1 or > AtTimeChunkSize) throw new ArgumentOutOfRangeException(nameof(count));
var sb = new StringBuilder();
for (var k = 0; k < count; k++)
{
var p = "$" + (k + 2).ToString(CultureInfo.InvariantCulture);
var req = k.ToString(CultureInfo.InvariantCulture);
if (k > 0) sb.Append(" UNION ALL ");
sb.Append("SELECT * FROM (SELECT ").Append(req).Append(" AS req, 0 AS dir, ").Append(SampleColumns)
.Append(" FROM ").Append(sample).Append(" WHERE tag = $1 AND ts <= ").Append(p)
.Append(" AND NOT deleted ORDER BY ts DESC LIMIT 1)");
sb.Append(" UNION ALL ");
sb.Append("SELECT * FROM (SELECT ").Append(req).Append(" AS req, 1 AS dir, ").Append(SampleColumns)
.Append(" FROM ").Append(sample).Append(" WHERE tag = $1 AND ts > ").Append(p)
.Append(" AND NOT deleted ORDER BY ts ASC LIMIT 1)");
}
return sb.ToString();
}
// ------------------------------------------------------------------------------------------ Events
/// <inheritdoc />
/// <remarks>
/// <c>[start, end)</c> by event time (a reversed window is swapped), ordered by <c>(ts, event_id)</c> ascending.
/// <paramref name="maxEvents"/> &lt;= 0 or <see cref="int.MaxValue"/> ⇒ unbounded. Otherwise
/// <c>maxEvents + 1</c> rows are fetched; when the extra row arrives it is dropped and the continuation point is
/// the UTF-8 of <see cref="EventContinuation.Encode"/> for that first not-returned event.
/// </remarks>
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken)
{
// No stored event can carry a source longer than the limit, so an over-long filter matches nothing.
if (sourceName is not null && sourceName.Length > HistorianTagNames.MaxLength)
return new HistoricalEventsResult([], null);
try
{
await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
var start = QuestDbConnection.Round(startUtc);
var end = QuestDbConnection.Round(endUtc);
var (lo, hi) = start <= end ? (start, end) : (end, start);
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
var sql =
$"SELECT event_id, source, ts, received_utc, message, severity FROM {_conn.EventTable} " +
"WHERE ts >= $1 AND ts < $2" +
(sourceName is null ? "" : " AND source = $3") +
" ORDER BY ts ASC, event_id ASC" +
(bounded ? (sourceName is null ? " LIMIT $3" : " LIMIT $4") : "");
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, sql);
QuestDbConnection.BindTimestamp(cmd, lo);
QuestDbConnection.BindTimestamp(cmd, hi);
if (sourceName is not null) QuestDbConnection.BindText(cmd, sourceName);
if (bounded) QuestDbConnection.BindLong(cmd, (long)maxEvents + 1);
var events = new List<HistoricalEvent>();
byte[]? continuation = null;
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
var eventTime = QuestDbConnection.ReadUtc(reader, 2);
var e = new HistoricalEvent(
reader.IsDBNull(0) ? "" : reader.GetString(0),
reader.IsDBNull(1) ? null : reader.GetString(1),
eventTime,
reader.IsDBNull(3) ? eventTime : QuestDbConnection.ReadUtc(reader, 3),
reader.IsDBNull(4) ? null : reader.GetString(4),
HistoricalEventRows.ClampSeverity(reader.IsDBNull(5) ? 0 : reader.GetInt32(5)));
if (bounded && events.Count == maxEvents)
{
continuation = Encoding.UTF8.GetBytes(EventContinuation.Encode(e.EventTimeUtc, e.EventId));
break;
}
events.Add(e);
}
RecordSuccess();
return new HistoricalEventsResult(events, continuation);
}
catch (Exception ex)
{
RecordFailure(ex);
throw;
}
}
// ------------------------------------------------------------------------------------------ Processed
/// <inheritdoc />
/// <remarks>
/// The (rounded) window is planned by <see cref="BucketPlan.Create"/>: bucket <c>i</c> =
/// <c>[start + i·interval, start + (i+1)·interval)</c>, start-inclusive / end-exclusive, the last bucket truncated
/// at <c>end</c>, interval &lt;= 0 ⇒ one bucket, <c>start == end</c> ⇒ none, a reversed window computed forward
/// and returned descending, at most <see cref="BucketPlan.MaxBuckets"/> buckets. One statement per node returns
/// the <see cref="BucketRow"/> statistics of each non-empty bucket (bucket index
/// <c>(cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks</c>, never <c>SAMPLE BY</c>);
/// <see cref="AggregateEvaluator.Evaluate"/> produces each value and <see cref="BucketPlan.Assemble"/> fills the
/// empty buckets with <c>BadNoData</c>.
/// </remarks>
/// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not one of the thirteen served.</exception>
/// <exception cref="ArgumentOutOfRangeException">The window holds more than <see cref="BucketPlan.MaxBuckets"/> intervals.</exception>
public async Task<HistoryReadResult> ReadProcessedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReference);
var family = AggregateFamilies.FamilyOf(aggregate) ?? throw new HistoryAggregateNotSupportedException(aggregate);
var plan = BucketPlan.Create(QuestDbConnection.Round(startUtc), QuestDbConnection.Round(endUtc), interval);
try
{
var now = _time.GetUtcNow().UtcDateTime;
var evaluated = new Dictionary<long, DataValueSnapshot>();
if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
{
var sample = _conn.SampleTable;
var sql = family switch
{
AggregateFamily.Simple => QuestDbAggregateQueries.Simple(sample),
AggregateFamily.StartEnd => QuestDbAggregateQueries.StartEnd(sample),
_ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime),
};
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, sql);
QuestDbConnection.BindLong(cmd, plan.StartUnixTicks);
QuestDbConnection.BindLong(cmd, plan.IntervalTicks);
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, plan.StartUtc);
QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
var row = ReadBucket(reader, family);
evaluated[row.Bucket] = AggregateEvaluator.Evaluate(aggregate, row, plan.BucketStart(row.Bucket), now);
}
}
var result = plan.Assemble(evaluated, bucketStart => AggregateEvaluator.NoData(bucketStart, now));
RecordSuccess();
return new HistoryReadResult(result, ContinuationPoint: null);
}
catch (Exception ex)
{
RecordFailure(ex);
throw;
}
}
private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family)
{
static double? D(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
static int N(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? 0 : Convert.ToInt32(r.GetValue(i), CultureInfo.InvariantCulture);
// A bucket without a usable sample has NULL times but kind 0 (BYTE has no NULL): absence is keyed on the time.
static HistorianValue V(NpgsqlDataReader r, int i) =>
r.IsDBNull(i + 4)
? HistorianValue.Empty
: ToValue(QuestDbConnection.ReadKind(r, i), D(r, i + 1), r.IsDBNull(i + 2) ? null : r.GetInt64(i + 2),
r.IsDBNull(i + 3) ? null : r.GetString(i + 3));
var row = new BucketRow
{
Bucket = Convert.ToInt64(reader.GetValue(0), CultureInfo.InvariantCulture),
Total = N(reader, 1),
Usable = N(reader, 2),
Good = N(reader, 3),
Bad = N(reader, 4),
NonNumeric = N(reader, 5),
};
var f = QuestDbAggregateQueries.FirstFamilyColumn;
switch (family)
{
case AggregateFamily.Simple:
row.Avg = D(reader, f);
row.Min = D(reader, f + 1);
row.Max = D(reader, f + 2);
row.Sum = D(reader, f + 3);
break;
case AggregateFamily.StartEnd:
row.First = V(reader, f);
row.Last = V(reader, f + 5);
break;
default:
row.Extreme = D(reader, f);
row.ExtremeTime = reader.IsDBNull(f + 1) ? null : QuestDbConnection.ReadUtc(reader, f + 1);
break;
}
return row;
}
// ------------------------------------------------------------------------------------------ Shared
/// <summary>
/// Reads <c>ts, kind, num_value, int_value, str_value, status</c> starting at <paramref name="o"/> and applies the
/// shared quality clamp (<see cref="StoredSample.FromStored"/>).
/// </summary>
private static StoredSample ReadStored(NpgsqlDataReader r, int o)
{
var value = ToValue(
QuestDbConnection.ReadKind(r, o + 1),
r.IsDBNull(o + 2) ? null : r.GetDouble(o + 2),
r.IsDBNull(o + 3) ? null : r.GetInt64(o + 3),
r.IsDBNull(o + 4) ? null : r.GetString(o + 4));
// status is always written; a NULL (only possible for a hand-made row) is served as "no data", never Good.
var status = r.IsDBNull(o + 5) ? HistorianStatusCodes.BadNoData : SampleColumnMapper.FromLongStatus(r.GetInt64(o + 5));
return StoredSample.FromStored(QuestDbConnection.ReadUtc(r, o), value, status);
}
/// <summary>
/// Rebuilds a value from its stored columns via the shared <see cref="SampleColumnMapper.FromColumns"/>, after the
/// RC2 repair: QuestDB reads a <c>LONG</c> holding <see cref="long.MinValue"/> back as NULL, so an Int64 row with
/// a NULL <c>int_value</c> and a <c>num_value</c> of exactly −2^63 is that value.
/// </summary>
internal static HistorianValue ToValue(byte kind, double? num, long? integer, string? str)
{
if (kind == (byte)HistorianValueKind.Int64 && integer is null && num == LongNullSentinelAsDouble)
integer = long.MinValue;
return SampleColumnMapper.FromColumns(kind, num, integer, str);
}
private static DateTime ReadServerTime(NpgsqlDataReader r, int i, DateTime fallback) =>
r.IsDBNull(i) ? fallback : QuestDbConnection.ReadUtc(r, i);
/// <summary>Read-only tag lookup (R7; never creates a row); runs the schema bootstrap on a cache miss.</summary>
private Task<bool> ResolveAsync(string name, CancellationToken ct) => _catalog.ExistsAsync(name, ct);
// ------------------------------------------------------------------------------------------ Health
/// <inheritdoc />
public HistorianHealthSnapshot GetHealthSnapshot() => _health.Snapshot();
private void RecordSuccess() => _health.RecordSuccess();
private void RecordFailure(Exception ex)
{
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
if (ex is OperationCanceledException) return;
_health.RecordFailure(ex);
_log.LogDebug("QuestDB historian read failed ({ExceptionType})", ex.GetType().Name);
}
/// <inheritdoc />
public void Dispose()
{
// Connections are per call (pooled by the shared NpgsqlDataSource, owned by the provider); nothing to release.
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs
+203
View File
@@ -0,0 +1,203 @@
using System.Text;
using Npgsql;
using NpgsqlTypes;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
/// <summary>
/// The QuestDB historian as a conformance backend: a fresh unique table prefix per instance (dropped on dispose),
/// seeded by plain multi-row INSERTs (never through the writer or updater) followed by the WAL visibility wait.
/// Storage resolution is 1 µs (10 ticks); <see cref="Round"/> is <see cref="QuestDbConnection.Round"/>, and
/// <see cref="SeedAsync"/> rounds the times it is given the same way before storing them.
/// </summary>
internal sealed class QuestDbConformanceBackend : IConformanceBackend
{
private const int SeedBatchRows = 1000;
private readonly QuestDbFixture _fx;
private readonly string _prefix;
private QuestDbConformanceBackend(QuestDbFixture fx, string prefix, QuestDbHistorianOptions options,
QuestDbConnection conn, QuestDbSchemaBootstrapper boot)
{
_fx = fx;
_prefix = prefix;
Options = options;
Conn = conn;
Boot = boot;
Wal = new QuestDbWalVisibility(conn, options.VisibilityTimeout);
}
/// <summary>The options of this instance's table set.</summary>
public QuestDbHistorianOptions Options { get; }
/// <summary>The connection helper.</summary>
public QuestDbConnection Conn { get; }
/// <summary>The schema bootstrapper (already run).</summary>
public QuestDbSchemaBootstrapper Boot { get; }
/// <summary>The WAL visibility wait.</summary>
public QuestDbWalVisibility Wal { get; }
/// <summary>Creates a backend over a fresh table prefix; skips the calling test when the gate is closed.</summary>
/// <param name="fx">The shared fixture.</param>
/// <returns>The backend.</returns>
public static async Task<IConformanceBackend> CreateAsync(QuestDbFixture fx) => await CreateConcreteAsync(fx);
/// <summary>As <see cref="CreateAsync"/>, typed (for QuestDB-specific tests that need the connection).</summary>
/// <param name="fx">The shared fixture.</param>
/// <returns>The backend.</returns>
public static async Task<QuestDbConformanceBackend> CreateConcreteAsync(QuestDbFixture fx)
{
fx.SkipIfUnavailable();
var prefix = fx.UniquePrefix();
var o = fx.Options(prefix);
var conn = new QuestDbConnection(o);
var boot = new QuestDbSchemaBootstrapper(conn, o);
await boot.EnsureAsync(CancellationToken.None);
return new QuestDbConformanceBackend(fx, prefix, o, conn, boot);
}
/// <inheritdoc/>
public long ResolutionTicks => 10;
/// <inheritdoc/>
public DateTime Round(DateTime t) => QuestDbConnection.Round(t);
/// <summary>A fresh tag catalog (fresh positive cache) over this table set.</summary>
public QuestDbTagCatalog NewCatalog() => new(Conn, Boot, Wal);
/// <inheritdoc/>
public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) =>
new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock);
/// <inheritdoc/>
/// <remarks>TODO(T7): build the T4 <c>QuestDbHistorianValueWriter</c> here once T4 is merged.</remarks>
public IHistorianValueWriter CreateValueWriter(Func<DateTime>? utcNow = null) =>
throw new NotSupportedException("TODO(T7): the QuestDB value writer (T4) is not merged into this branch yet.");
/// <inheritdoc/>
/// <remarks>TODO(T7): build the T6 <c>QuestDbHistorianDataUpdater</c> here once T6 is merged.</remarks>
public IHistorianDataUpdater CreateUpdater() =>
throw new NotSupportedException("TODO(T7): the QuestDB history updater (T6) is not merged into this branch yet.");
/// <inheritdoc/>
/// <remarks>
/// Times are rounded to the microsecond (<see cref="Round"/>) before they are stored; the tag row is registered
/// through the catalog (data type 0, as the SQL Server seed does); every row is live (<c>deleted = false</c>);
/// <c>server_ts</c> = rounded time + 1 s. Waits for WAL visibility before returning.
/// </remarks>
public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples)
{
await NewCatalog().EnsureAsync(tag, 0, CancellationToken.None);
await InsertRowsAsync(tag, samples.Select(s => (Round(s.T), s.V, s.Status, Deleted: false)).ToArray());
}
/// <summary>
/// Inserts raw rows (the tag must already be registered when reads should find it), including tombstones, then
/// waits for WAL visibility. A tombstone re-inserted at an existing <c>(ts, tag)</c> replaces that row (DEDUP).
/// </summary>
/// <param name="tag">The tag.</param>
/// <param name="rows">Rows: time (already rounded by the caller or not — it is bound through BindTimestamp), value, status, deleted.</param>
public async Task InsertRowsAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows)
{
if (rows.Length == 0) return;
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
for (var offset = 0; offset < rows.Length; offset += SeedBatchRows)
{
var batch = rows.Skip(offset).Take(SeedBatchRows).ToArray();
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 cmd = Conn.CreateCommand(c, "");
var n = 0;
for (var i = 0; i < batch.Length; i++)
{
var (t, v, status, deleted) = batch[i];
var col = SampleColumnMapper.ToColumns(v);
if (i > 0) sql.Append(", ");
sql.Append('(');
for (var j = 0; j < 9; j++) sql.Append(j == 0 ? "" : ", ").Append('$').Append(++n);
sql.Append(')');
QuestDbConnection.BindTimestamp(cmd, t);
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindByte(cmd, col.ValueKind);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Double, Value = (object?)col.NumValue ?? DBNull.Value });
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Bigint, Value = (object?)col.IntValue ?? DBNull.Value });
QuestDbConnection.BindText(cmd, col.StrValue);
QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(status));
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Boolean, Value = deleted });
QuestDbConnection.BindTimestamp(cmd, Round(t).AddSeconds(1));
}
cmd.CommandText = sql.ToString();
await cmd.ExecuteNonQueryAsync();
}
}
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
}
/// <summary>Runs a statement against this table set (test-only escape hatch for hand-made rows), then waits for WAL visibility of the sample table.</summary>
/// <param name="sql">The statement; parameters are bound positionally by <paramref name="bind"/>.</param>
/// <param name="bind">Binds the parameters.</param>
public async Task ExecuteAsync(string sql, Action<NpgsqlCommand> bind)
{
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
await using var cmd = Conn.CreateCommand(c, sql);
bind(cmd);
await cmd.ExecuteNonQueryAsync();
}
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
}
/// <inheritdoc/>
public async Task<IReadOnlyList<(DateTime T, HistorianValue V, uint Status)>> ReadBackAsync(string tag)
{
await using var c = await Conn.OpenAsync(CancellationToken.None);
await using var cmd = Conn.CreateCommand(c,
$"SELECT ts, kind, num_value, int_value, str_value, status FROM {Conn.SampleTable} " +
"WHERE tag = $1 AND NOT deleted ORDER BY ts");
QuestDbConnection.BindText(cmd, tag);
var list = new List<(DateTime, HistorianValue, uint)>();
await using var r = await cmd.ExecuteReaderAsync();
while (await r.ReadAsync())
{
var v = QuestDbHistorianDataSource.ToValue(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));
// Verbatim: no read-side clamp. A NULL status cannot be written by any code path; it reads as the raw 0 bits.
var status = r.IsDBNull(5) ? 0u : SampleColumnMapper.FromLongStatus(r.GetInt64(5));
list.Add((QuestDbConnection.ReadUtc(r, 0), v, status));
}
return list;
}
/// <inheritdoc/>
public async Task<bool> TagExistsAsync(string tag)
{
await using var c = await Conn.OpenAsync(CancellationToken.None);
await using var cmd = Conn.CreateCommand(c, $"SELECT count() FROM {Conn.TagTable} WHERE name = $1");
QuestDbConnection.BindText(cmd, tag);
return Convert.ToInt64(await cmd.ExecuteScalarAsync()) > 0;
}
/// <inheritdoc/>
public async Task<DriverDataType?> TagDataTypeAsync(string tag)
{
await using var c = await Conn.OpenAsync(CancellationToken.None);
await using var cmd = Conn.CreateCommand(c, $"SELECT data_type FROM {Conn.TagTable} WHERE name = $1 LIMIT 1");
QuestDbConnection.BindText(cmd, tag);
await using var r = await cmd.ExecuteReaderAsync();
return await r.ReadAsync() ? (DriverDataType)QuestDbConnection.ReadKind(r, 0) : null;
}
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
await Conn.DisposeAsync();
await _fx.DropPrefixAsync(_prefix); // early; the fixture also drops every prefix it handed out
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs
+266
View File
@@ -0,0 +1,266 @@
using System.Text;
using Npgsql;
using NpgsqlTypes;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
/// <summary>
/// QuestDB-specific read behaviour (the shared Raw/AtTime/Processed literals run through the conformance suite):
/// tombstones, the pre-1970 / whole-range bucket math, reverse paging at one value per page, QuestDB's LONG NULL
/// sentinel, the kind-decides-the-value rule, AtTime chunking, Events and failure accounting.
/// </summary>
[Collection(QuestDbCollection.Name)]
public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime
{
private static readonly CancellationToken Ct = CancellationToken.None;
private const uint Good = ConformanceData.Good;
private const uint Bad = ConformanceData.Bad;
private const uint GoodCalc = 0x00000401u;
private const uint UncertainCalc = 0x40A40401u;
private const uint UncertainInterpolated = 0x40A40402u;
private const uint GoodInterpolated = 0x00000402u;
private const uint NoDataCalc = 0x809B0401u;
private QuestDbConformanceBackend _b = null!;
private static DateTime At(double s) => ConformanceData.At(s);
public async ValueTask InitializeAsync() => _b = await QuestDbConformanceBackend.CreateConcreteAsync(fx);
public async ValueTask DisposeAsync()
{
if (_b is not null) await _b.DisposeAsync();
}
private static double?[] Values(HistoryReadResult r) => r.Samples.Select(s => (double?)s.Value).ToArray();
/// <summary>What T6's DeleteAtTime writes: the same (ts, tag), deleted = true, value columns NULL, status BadNoData.</summary>
private Task TombstoneAsync(string tag, DateTime t) =>
_b.InsertRowsAsync(tag, [(t, HistorianValue.Empty, HistorianStatusCodes.BadNoData, true)]);
[Fact]
public async Task Tombstoned_rows_are_invisible_to_raw_attime_and_processed()
{
await ConformanceData.SeedAAsync(_b); // +0=10, +10=20, +20=Bad 30, +30=40, +40=50
await TombstoneAsync("A", At(10)); // DEDUP replaces the +10 row with a tombstone
(await _b.ReadBackAsync("A")).Count.ShouldBe(4);
var src = _b.CreateDataSource();
var raw = await src.ReadRawAsync("A", At(0), At(100), 0, Ct);
Values(raw).ShouldBe([10d, 30d, 40d, 50d]);
raw.Samples.Select(s => s.StatusCode).ShouldBe([Good, Bad, Good, Good]);
// +10 is no longer an exact hit: previous is +0 = 10 (Good), next is the Bad +20 ⇒ held, Uncertain.
var at = (await src.ReadAtTimeAsync("A", [At(10), At(5)], Ct)).Samples;
at.Select(s => s.Value).ShouldBe([(object?)10d, 10d]);
at.ShouldAllBe(s => s.StatusCode == UncertainInterpolated);
// B0 [0,20) holds only {10}: Good, count 1. A visible tombstone would be a Bad sample ⇒ Uncertain, PercentBad 50.
async Task<DataValueSnapshot> B0(HistoryAggregateType a) =>
(await src.ReadProcessedAsync("A", At(0), At(80), TimeSpan.FromSeconds(20), a, Ct)).Samples[0];
var avg = await B0(HistoryAggregateType.Average);
avg.Value.ShouldBe(10d);
avg.StatusCode.ShouldBe(GoodCalc);
(await B0(HistoryAggregateType.Count)).Value.ShouldBe(1);
(await B0(HistoryAggregateType.PercentBad)).Value.ShouldBe(0d);
(await B0(HistoryAggregateType.End)).Value.ShouldBe(10d);
}
[Fact]
public async Task A_whole_range_processed_read_from_1601_to_9999_with_interval_zero_does_not_overflow()
{
await ConformanceData.SeedAAsync(_b);
var src = _b.CreateDataSource();
var start = new DateTime(1601, 1, 1, 0, 0, 0, DateTimeKind.Utc);
var end = new DateTime(9999, 12, 31, 0, 0, 0, DateTimeKind.Utc);
foreach (var agg in new[] { HistoryAggregateType.Average, HistoryAggregateType.Start, HistoryAggregateType.MaximumActualTime })
{
var s = (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, agg, Ct)).Samples.Single();
s.StatusCode.ShouldBe(UncertainCalc, agg.ToString());
}
(await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.Average, Ct)).Samples.Single().Value.ShouldBe(30d);
(await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.Start, Ct)).Samples.Single().Value.ShouldBe(10d);
var max = (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.MaximumActualTime, Ct)).Samples.Single();
max.Value.ShouldBe(50d);
max.SourceTimestampUtc.ShouldBe(_b.Round(At(40)));
// The extreme DateTime range: MinValue/MaxValue are nudged (never bound as ±infinity, which would match nothing).
(await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Total, Ct))
.Samples.Single().Value.ShouldBe(120d);
// A pre-1970 window holds no data and is not an error.
(await src.ReadProcessedAsync("A", start, new DateTime(1969, 1, 1, 0, 0, 0, DateTimeKind.Utc), TimeSpan.FromDays(3650),
HistoryAggregateType.Count, Ct)).Samples.ShouldAllBe(x => x.StatusCode == NoDataCalc);
src.GetHealthSnapshot().TotalFailures.ShouldBe(0);
}
/// <summary>
/// The node manager's reverse paging at <c>NumValuesPerNode = 1</c> (OtOpcUaNodeManager + HistoryPaging): resume at
/// the last timestamp inclusive and trim it; when the trimmed page stalls on that timestamp, point-read it and step
/// one tick down. One tick below a µs-aligned time rounds to the previous µs, so the walk advances.
/// </summary>
[Fact]
public async Task A_reverse_paging_walk_with_one_value_per_page_returns_every_row()
{
await ConformanceData.SeedAAsync(_b);
var src = _b.CreateDataSource();
var end = At(-1);
DateTime start = At(100);
var resumed = false;
var emitted = new List<object?>();
for (var guard = 0; guard < 50; guard++)
{
var page = (await src.ReadRawAsync("A", start, end, 1, Ct)).Samples;
var trimmed = resumed ? page.Where(s => s.SourceTimestampUtc != start).ToList() : page.ToList();
if (page.Count == 0) break;
if (resumed && (trimmed.Count == 0 || trimmed[^1].SourceTimestampUtc == start))
{
// Stalled on the boundary: the tie cluster at `start` is one sample (already emitted) ⇒ step down a tick.
(await src.ReadRawAsync("A", start, start, 2, Ct)).Samples.Count.ShouldBe(1);
if (start.AddTicks(-1) <= end) break;
start = start.AddTicks(-1);
resumed = false;
continue;
}
emitted.AddRange(trimmed.Select(s => s.Value));
start = page[^1].SourceTimestampUtc!.Value;
resumed = true;
}
emitted.ShouldBe([50d, 40d, 30d, 20d, 10d]);
}
[Fact]
public async Task An_int64_of_long_MinValue_is_served_as_long_MinValue_not_zero()
{
// QuestDB stores long.MinValue in a LONG as NULL (its NULL sentinel); num_value keeps -2^63 exactly (RC2).
await _b.SeedAsync("L",
(At(0), HistorianValue.FromInt64(long.MinValue), Good),
(At(1), HistorianValue.FromInt64(long.MinValue + 1), Good));
var src = _b.CreateDataSource();
(await src.ReadRawAsync("L", At(0), At(10), 0, Ct)).Samples.Select(s => s.Value)
.ShouldBe([(object?)long.MinValue, long.MinValue + 1]);
(await src.ReadAtTimeAsync("L", [At(0)], Ct)).Samples.Single().Value.ShouldBe(long.MinValue);
(await src.ReadProcessedAsync("L", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Start, Ct))
.Samples.Single().Value.ShouldBe(long.MinValue);
}
[Fact]
public async Task The_kind_not_the_populated_column_decides_the_value()
{
// An Int64-kind row whose num_value is NULL (a NaN-free double view lost) still serves its int_value, and a
// Double-kind row carrying a stray int_value still serves num_value.
await _b.SeedAsync("K");
await _b.ExecuteAsync(
$"INSERT INTO {_b.Conn.SampleTable} (ts, tag, kind, num_value, int_value, status, deleted, server_ts) " +
"VALUES ($1, $2, $3, NULL, $4, $5, false, $6), ($7, $2, $8, $9, $10, $5, false, $11)",
cmd =>
{
QuestDbConnection.BindTimestamp(cmd, At(0)); // $1
QuestDbConnection.BindText(cmd, "K"); // $2
QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Int64); // $3
QuestDbConnection.BindLong(cmd, 7); // $4
QuestDbConnection.BindLong(cmd, 0); // $5 status Good
QuestDbConnection.BindTimestamp(cmd, At(1)); // $6
QuestDbConnection.BindTimestamp(cmd, At(2)); // $7
QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Double); // $8
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Double, Value = 2.5 }); // $9
QuestDbConnection.BindLong(cmd, 99); // $10
QuestDbConnection.BindTimestamp(cmd, At(3)); // $11
});
var s = (await _b.CreateDataSource().ReadRawAsync("K", At(0), At(10), 0, Ct)).Samples;
s.Select(x => x.Value).ShouldBe([(object?)7L, 2.5d]);
}
[Fact]
public async Task AtTime_over_several_statement_chunks_keeps_request_length_and_order()
{
await ConformanceData.SeedAAsync(_b);
// 140 distinct times (3 statements of ≤ 50), requested in descending order, plus a duplicate of the first.
var times = Enumerable.Range(0, 140).Select(i => At((390 - i * 3) / 10.0)).ToList();
times.Add(times[0]);
var s = (await _b.CreateDataSource().ReadAtTimeAsync("A", times, Ct)).Samples;
s.Count.ShouldBe(times.Count);
s.Select(x => x.SourceTimestampUtc).ShouldBe(times.Select(t => (DateTime?)_b.Round(t)));
s[0].Value.ShouldBe(49d); // +39: 40 + 10 × 9/10
s[0].StatusCode.ShouldBe(GoodInterpolated);
s[^1].Value.ShouldBe(49d); // the duplicate
s[100].Value.ShouldBe(19d); // +9: 10 + 10 × 9/10, from the third chunk
s[139].Value.ShouldBeNull(); // -2.7 s: no previous sample
s[139].StatusCode.ShouldBe(0x809B0000u); // BadNoData
}
[Fact]
public async Task Events_are_filtered_by_source_and_window_ascending_and_limited_with_a_continuation()
{
await _b.Boot.EnsureAsync(Ct);
async Task Event(DateTime t, string id, string source, int severity, string? message)
{
await _b.ExecuteAsync(
$"INSERT INTO {_b.Conn.EventTable} (ts, event_id, source, severity, message, received_utc) VALUES ($1, $2, $3, $4, $5, $6)",
cmd =>
{
QuestDbConnection.BindTimestamp(cmd, t);
QuestDbConnection.BindText(cmd, id);
QuestDbConnection.BindText(cmd, source);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = severity });
QuestDbConnection.BindText(cmd, message);
QuestDbConnection.BindTimestamp(cmd, t.AddSeconds(2));
});
await _b.Wal.WaitAsync(_b.Conn.EventTable, Ct);
}
await Event(At(3), "e3", "src1", 700, "three");
await Event(At(1), "e1", "src1", 70000, "one");
await Event(At(2), "e2", "src2", 500, null);
await Event(At(9), "e9", "src1", 500, "out of window");
var src = _b.CreateDataSource();
var all = await src.ReadEventsAsync(null, At(0), At(5), 0, Ct);
all.Events.Select(e => e.EventId).ShouldBe(["e1", "e2", "e3"]);
all.ContinuationPoint.ShouldBeNull();
all.Events[0].Severity.ShouldBe((ushort)65535);
all.Events[0].EventTimeUtc.ShouldBe(_b.Round(At(1)));
all.Events[0].ReceivedTimeUtc.ShouldBe(_b.Round(At(1)).AddSeconds(2));
all.Events[1].Message.ShouldBeNull();
var one = await src.ReadEventsAsync("src1", At(5), At(0), 1, Ct); // reversed window is swapped
one.Events.Select(e => e.EventId).ShouldBe(["e1"]);
Encoding.UTF8.GetString(one.ContinuationPoint!).ShouldBe(EventContinuation.Encode(_b.Round(At(3)), "e3"));
(await src.ReadEventsAsync("src1", At(0), At(5), 2, Ct)).ContinuationPoint.ShouldBeNull();
(await src.ReadEventsAsync(new string('x', 401), At(0), At(5), 0, Ct)).Events.ShouldBeEmpty();
}
[Fact]
public async Task A_failed_read_is_rethrown_and_recorded_in_health_without_counting_client_errors()
{
// A table set that does not exist, with AutoCreateSchema off: every read that reaches the database fails.
var prefix = fx.UniquePrefix();
var o = fx.Options(prefix, autoCreate: false);
await using var conn = new QuestDbConnection(o);
var boot = new QuestDbSchemaBootstrapper(conn, o);
using var src = new QuestDbHistorianDataSource(conn, boot, new QuestDbTagCatalog(conn, boot, new QuestDbWalVisibility(conn, o.VisibilityTimeout)));
await Should.ThrowAsync<HistorianSchemaMissingException>(() => src.ReadRawAsync("A", At(0), At(100), 0, Ct));
await Should.ThrowAsync<HistorianSchemaMissingException>(() => src.ReadEventsAsync(null, At(0), At(5), 0, Ct));
// Client-request errors are thrown before any I/O and are not backend failures.
await Should.ThrowAsync<HistoryAggregateNotSupportedException>(() =>
src.ReadProcessedAsync("A", At(0), At(5), TimeSpan.Zero, HistoryAggregateType.Interpolative, Ct));
await Should.ThrowAsync<ArgumentOutOfRangeException>(() =>
src.ReadProcessedAsync("A", At(0), At(100), TimeSpan.FromTicks(10), HistoryAggregateType.Count, Ct));
var h = src.GetHealthSnapshot();
h.TotalQueries.ShouldBe(2);
h.TotalFailures.ShouldBe(2);
h.ConsecutiveFailures.ShouldBe(2);
h.LastError.ShouldNotBeNull();
h.LastError.ShouldStartWith(nameof(HistorianSchemaMissingException));
h.ProcessConnectionOpen.ShouldBeFalse();
// The healthy backend's own data source is unaffected (a failure fails only its call).
await ConformanceData.SeedAAsync(_b);
(await _b.CreateDataSource().ReadRawAsync("A", At(0), At(100), 0, Ct)).Samples.Count.ShouldBe(5);
}
}