diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs
new file mode 100644
index 00000000..afe02efa
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs
@@ -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;
+
+///
+/// Push-down QuestDB SQL for processed (aggregate) reads (spec §3 "Processed", T0 Q8-confirmed forms), one query
+/// per , mirroring the SQL Server backend's SqlAggregateQueries family split and
+/// column order. Every query takes five positional parameters:
+///
+/// - $1 — the (rounded) window start in 100 ns ticks since the Unix epoch
+/// (, Bigint);
+/// - $2 — the bucket width in ticks (, Bigint);
+/// - $3 — the tag (Varchar);
+/// - $4/$5 — the forward window [start, end) (TimestampTz via
+/// ).
+///
+/// Each returns one row per non-empty bucket; the caller fills the empty ones.
+///
+///
+///
+/// Bucketing. b = (cast(ts AS LONG) * 10 − $1) / $2. cast(ts AS LONG) is µs since 1970, so
+/// × 10 is ticks since 1970; both it and $1 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.
+/// SAMPLE BY is deliberately not used: its alignment rules differ from ours.
+///
+///
+/// Sample classes (S6). The status is stored as the unsigned 32-bit code in a LONG, so
+/// "not Bad" is status < 0x80000000 (not T-SQL's sign test). usable = not Bad AND
+/// kind <> 0 (an Empty sample is an outage marker and counts as Bad whatever its code);
+/// good = usable AND status < 0x40000000; bad = not usable; non-numeric = usable
+/// String/DateTime. The literals are generated from and .
+///
+/// Common result columns (ordinals 0–5): b, total_count, usable_count, good_count, bad_count,
+/// non_numeric_count. Family columns follow from .
+///
+internal static class QuestDbAggregateQueries
+{
+ /// Ordinal of the first family-specific column.
+ public const int FirstFamilyColumn = 6;
+
+ /// status < SeverityBadBit, as a decimal LONG literal.
+ internal static readonly string NotBadLimit = StatusBits.SeverityBadBit.ToString(CultureInfo.InvariantCulture);
+
+ /// status < SeverityUncertainBit, as a decimal LONG literal.
+ internal static readonly string GoodLimit = StatusBits.SeverityUncertainBit.ToString(CultureInfo.InvariantCulture);
+
+ /// The kind values of String and DateTime: usable but not numeric.
+ 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);
+
+ ///
+ /// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad.
+ /// Family columns: 6 avg_value, 7 min_value, 8 max_value, 9 sum_value (DOUBLE, NULL when no usable sample).
+ ///
+ /// Prefixed sample table name.
+ /// The SQL text.
+ 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";
+
+ ///
+ /// Start / End / Delta: the first and last usable sample of each bucket by ts.
+ /// Family columns: 6..10 first sample (kind, num_value, int_value, str_value, ts), 11..15 last
+ /// sample (same order). A bucket without a usable sample has NULL ts but kind = 0 (BYTE has no NULL),
+ /// so the reader keys absence on the NULL time.
+ ///
+ /// Prefixed sample table name.
+ /// The SQL text.
+ 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";
+
+ ///
+ /// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) num_value, ties
+ /// broken by the earliest ts. Family columns: 6 num_value, 7 ts (NULL when the bucket has no usable
+ /// numeric sample).
+ ///
+ /// Prefixed sample table name.
+ /// True for MaximumActualTime.
+ /// The SQL text.
+ 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";
+
+ /// The bucket-index expression (exposed for tests).
+ internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2";
+
+ /// The classified subquery s: the window's live samples with bucket index and the usable/good flags.
+ 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";
+
+ /// The five common count columns (after b).
+ 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";
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs
new file mode 100644
index 00000000..f4b97539
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs
@@ -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;
+
+///
+/// QuestDB read side of the historian (spec §3 "Reads"): Raw (forward / reverse / point), AtTime (chunked
+/// LIMIT 1 neighbour seeks, T0 Q9), Events, and Processed aggregates pushed down to QuestDB SQL
+/// (, T0 Q8). Results match the SQL Server data source except that every
+/// timestamp is at microsecond resolution.
+///
+///
+///
+/// Timestamps (R6). Every inbound time is normalised with and every
+/// is bound only through . Read bounds are not
+/// clamped to the epoch (a pre-1970 bound is valid); only MinValue/MaxValue are nudged. Results
+/// carry the rounded times, so an AtTime result is stamped at the rounded request time.
+///
+///
+/// Reads never create tags (R7). A tag resolves only through ;
+/// an unknown tag is a successful empty read (Raw), an all-BadNoData read (AtTime / Processed), never an
+/// error. Every sample query filters NOT deleted, so a DeleteRaw/DeleteAtTime tombstone is invisible.
+///
+///
+/// Values. Every row goes through (an Empty row with a non-Bad status
+/// is served BadNoData). The value kind, not which column is populated, decides the value. QuestDB stores
+/// in a LONG as NULL (its NULL sentinel), so an Int64 row whose
+/// int_value is NULL and whose num_value is exactly −2^63 is served as
+/// (ruling RC2).
+///
+///
+/// Faults. As in the SQL Server data source: a database / bootstrap exception is recorded in
+/// and rethrown, failing only that call. Client-request errors — an unsupported
+/// aggregate () and a processed window over
+/// intervals () — are thrown before
+/// any I/O and are not counted as backend failures.
+///
+///
+public sealed class QuestDbHistorianDataSource : IHistorianDataSource
+{
+ /// Most AtTime request times per statement (two UNION ALL branches each; QuestDB 10.0.1 refuses more than 126 branches).
+ public const int AtTimeChunkSize = 50;
+
+ /// as a double (−2^63, exact): what an Int64 sample of that value leaves in num_value.
+ 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();
+
+ /// Creates the data source over the provider's shared connection, bootstrapper and tag catalog.
+ /// Connection helper.
+ /// Schema bootstrapper (shared with the other QuestDB seams).
+ /// Tag catalog (reads use only its read-only ).
+ /// Optional logger.
+ /// Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.
+ 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
+
+ ///
+ ///
+ /// Forward (start < end): [start, end) ascending. Reverse (end < start):
+ /// (end, start] descending. start == end: the sample exactly at that (rounded) time.
+ /// 0 is "unbounded" (); larger values clamp.
+ /// At most top samples, ContinuationPoint always null (R8): the node manager pages by re-reading
+ /// from the last returned timestamp inclusive, which the (ts, tag) dedup key makes exact.
+ ///
+ public async Task 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();
+ 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
+
+ ///
+ ///
+ /// Each distinct (rounded) requested time picks up its previous-or-equal and next live sample through two
+ /// LIMIT 1 seeks, UNION ALL-ed, at most times per statement (spec §3,
+ /// T0 Q9). then decides each value exactly as for SQL Server (exact hit,
+ /// no previous ⇒ BadNoData, 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 BadNoData.
+ ///
+ public async Task ReadAtTimeAsync(
+ string fullReference, IReadOnlyList 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(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;
+ }
+ }
+
+ ///
+ /// The T0 Q9 statement for request times: $1 = tag, $(k+2) = time k.
+ /// Columns: req, dir (0 = previous-or-equal, 1 = next), then ts, kind, num_value, int_value, str_value,
+ /// status, server_ts. req/dir are generated integers, never user input.
+ ///
+ 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
+
+ ///
+ ///
+ /// [start, end) by event time (a reversed window is swapped), ordered by (ts, event_id) ascending.
+ /// <= 0 or ⇒ unbounded. Otherwise
+ /// maxEvents + 1 rows are fetched; when the extra row arrives it is dropped and the continuation point is
+ /// the UTF-8 of for that first not-returned event.
+ ///
+ public async Task 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();
+ 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
+
+ ///
+ ///
+ /// The (rounded) window is planned by : bucket i =
+ /// [start + i·interval, start + (i+1)·interval), start-inclusive / end-exclusive, the last bucket truncated
+ /// at end, interval <= 0 ⇒ one bucket, start == end ⇒ none, a reversed window computed forward
+ /// and returned descending, at most buckets. One statement per node returns
+ /// the statistics of each non-empty bucket (bucket index
+ /// (cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks, never SAMPLE BY);
+ /// produces each value and fills the
+ /// empty buckets with BadNoData.
+ ///
+ /// The aggregate is not one of the thirteen served.
+ /// The window holds more than intervals.
+ public async Task 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();
+ 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
+
+ ///
+ /// Reads ts, kind, num_value, int_value, str_value, status starting at and applies the
+ /// shared quality clamp ().
+ ///
+ 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);
+ }
+
+ ///
+ /// Rebuilds a value from its stored columns via the shared , after the
+ /// RC2 repair: QuestDB reads a LONG holding back as NULL, so an Int64 row with
+ /// a NULL int_value and a num_value of exactly −2^63 is that value.
+ ///
+ 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);
+
+ /// Read-only tag lookup (R7; never creates a row); runs the schema bootstrap on a cache miss.
+ private Task ResolveAsync(string name, CancellationToken ct) => _catalog.ExistsAsync(name, ct);
+
+ // ------------------------------------------------------------------------------------------ Health
+
+ ///
+ 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);
+ }
+
+ ///
+ public void Dispose()
+ {
+ // Connections are per call (pooled by the shared NpgsqlDataSource, owned by the provider); nothing to release.
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs
new file mode 100644
index 00000000..5dfc6f4c
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs
@@ -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;
+
+///
+/// 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); is , and
+/// rounds the times it is given the same way before storing them.
+///
+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);
+ }
+
+ /// The options of this instance's table set.
+ public QuestDbHistorianOptions Options { get; }
+
+ /// The connection helper.
+ public QuestDbConnection Conn { get; }
+
+ /// The schema bootstrapper (already run).
+ public QuestDbSchemaBootstrapper Boot { get; }
+
+ /// The WAL visibility wait.
+ public QuestDbWalVisibility Wal { get; }
+
+ /// Creates a backend over a fresh table prefix; skips the calling test when the gate is closed.
+ /// The shared fixture.
+ /// The backend.
+ public static async Task CreateAsync(QuestDbFixture fx) => await CreateConcreteAsync(fx);
+
+ /// As , typed (for QuestDB-specific tests that need the connection).
+ /// The shared fixture.
+ /// The backend.
+ public static async Task 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);
+ }
+
+ ///
+ public long ResolutionTicks => 10;
+
+ ///
+ public DateTime Round(DateTime t) => QuestDbConnection.Round(t);
+
+ /// A fresh tag catalog (fresh positive cache) over this table set.
+ public QuestDbTagCatalog NewCatalog() => new(Conn, Boot, Wal);
+
+ ///
+ public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) =>
+ new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock);
+
+ ///
+ /// TODO(T7): build the T4 QuestDbHistorianValueWriter here once T4 is merged.
+ public IHistorianValueWriter CreateValueWriter(Func? utcNow = null) =>
+ throw new NotSupportedException("TODO(T7): the QuestDB value writer (T4) is not merged into this branch yet.");
+
+ ///
+ /// TODO(T7): build the T6 QuestDbHistorianDataUpdater here once T6 is merged.
+ public IHistorianDataUpdater CreateUpdater() =>
+ throw new NotSupportedException("TODO(T7): the QuestDB history updater (T6) is not merged into this branch yet.");
+
+ ///
+ ///
+ /// Times are rounded to the microsecond () 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 (deleted = false);
+ /// server_ts = rounded time + 1 s. Waits for WAL visibility before returning.
+ ///
+ 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());
+ }
+
+ ///
+ /// 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 (ts, tag) replaces that row (DEDUP).
+ ///
+ /// The tag.
+ /// Rows: time (already rounded by the caller or not — it is bound through BindTimestamp), value, status, deleted.
+ 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);
+ }
+
+ /// Runs a statement against this table set (test-only escape hatch for hand-made rows), then waits for WAL visibility of the sample table.
+ /// The statement; parameters are bound positionally by .
+ /// Binds the parameters.
+ public async Task ExecuteAsync(string sql, Action 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);
+ }
+
+ ///
+ public async Task> 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;
+ }
+
+ ///
+ public async Task 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;
+ }
+
+ ///
+ public async Task 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;
+ }
+
+ ///
+ public async ValueTask DisposeAsync()
+ {
+ await Conn.DisposeAsync();
+ await _fx.DropPrefixAsync(_prefix); // early; the fixture also drops every prefix it handed out
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs
new file mode 100644
index 00000000..d0f29d5a
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs
@@ -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;
+
+///
+/// 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.
+///
+[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();
+
+ /// What T6's DeleteAtTime writes: the same (ts, tag), deleted = true, value columns NULL, status BadNoData.
+ 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 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);
+ }
+
+ ///
+ /// The node manager's reverse paging at NumValuesPerNode = 1 (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.
+ ///
+ [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();
+ 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(() => src.ReadRawAsync("A", At(0), At(100), 0, Ct));
+ await Should.ThrowAsync(() => 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(() =>
+ src.ReadProcessedAsync("A", At(0), At(5), TimeSpan.Zero, HistoryAggregateType.Interpolative, Ct));
+ await Should.ThrowAsync(() =>
+ 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);
+ }
+}