lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs

using System.Data.Common;
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. Raw and
///         Processed results carry stored (microsecond) times. An AtTime result is evaluated at the rounded request time
///         but stamped with the client's ORIGINAL requested time (as the SQL Server backend does), so a client can match
///         each answer to its request.
///     </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
{
    /// <inheritdoc />
    /// <remarks>QuestDB stores microseconds: 10 ticks. HistoryRead paging steps off a drained timestamp by this much.</remarks>
    public long TimestampResolutionTicks => 10;

    /// <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();
    private readonly int _maxTimeWeightedRawRows;

    /// <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>
    /// <param name="maxTimeWeightedRawRows">Raw in-window row cap of one time-weighted read (&lt;= 0 = unbounded).</param>
    internal QuestDbHistorianDataSource(
        QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, ILogger? log = null,
        TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
    {
        _maxTimeWeightedRawRows = maxTimeWeightedRawRows;
        _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;
        }
    }

    // ------------------------------------------------------------------------------------------ Modified (Part 11, #668)

    /// <inheritdoc />
    /// <remarks>
    ///     Reads <c>{prefix}sample_modified</c>, the append-only trail <see cref="QuestDbHistorianDataUpdater"/> writes
    ///     BEFORE each change. Windows follow <see cref="ReadRawAsync"/> (bounds rounded to µs); rows are ordered by
    ///     <c>(ts, modified_id)</c>; <c>maxValues + 1</c> rows are fetched to report
    ///     <see cref="HistoryModifiedReadResult.HasMore"/> exactly. A row whose <c>status</c> is NULL had no prior value
    ///     (an insert) and is served <c>null</c> / <c>BadNoData</c>. Because the trail row lands first, a crash between it and
    ///     the sample write leaves an orphan row describing a change that never applied (documented).
    /// </remarks>
    public async Task<HistoryModifiedReadResult> ReadModifiedAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(fullReference);
        var bounded = maxValues > 0 && maxValues < int.MaxValue;
        try
        {
            var values = new List<HistoryModifiedValue>();
            var hasMore = false;
            if (await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                var start = QuestDbConnection.Round(startUtc);
                var end = QuestDbConnection.Round(endUtc);
                var reverse = start > end;
                string where;
                if (start == end) where = "ts = $2 AND ts = $3";
                else if (!reverse) where = "ts >= $2 AND ts < $3";
                else where = "ts > $2 AND ts <= $3";
                var next = 4;
                if (resumeAfter is not null)
                {
                    where += reverse
                        ? " AND (ts < $4 OR (ts = $4 AND modified_id < $5))"
                        : " AND (ts > $4 OR (ts = $4 AND modified_id > $5))";
                    next = 6;
                }
                var order = reverse ? "DESC" : "ASC";
                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}, modified_id, modification_ts, update_type, principal FROM {_conn.SampleModifiedTable} " +
                    $"WHERE tag = $1 AND {where} ORDER BY ts {order}, modified_id {order}" + (bounded ? $" LIMIT ${next}" : ""));
                QuestDbConnection.BindText(cmd, fullReference);
                QuestDbConnection.BindTimestamp(cmd, lo);
                QuestDbConnection.BindTimestamp(cmd, hi);
                if (resumeAfter is not null)
                {
                    QuestDbConnection.BindTimestamp(cmd, resumeAfter.SourceTimeUtc);
                    QuestDbConnection.BindLong(cmd, resumeAfter.ModifiedId);
                }
                if (bounded)
                    cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = (int)maxValues + 1 });

                await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
                while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
                {
                    if (bounded && values.Count == maxValues)
                    {
                        hasMore = true;
                        break;
                    }

                    var sourceTime = QuestDbConnection.ReadUtc(reader, 0);
                    var modifiedAt = QuestDbConnection.ReadUtc(reader, 8);
                    DataValueSnapshot prior;
                    if (reader.IsDBNull(5))
                    {
                        prior = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, sourceTime, modifiedAt);
                    }
                    else
                    {
                        var s = ReadStored(reader, 0);
                        prior = new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc, ReadServerTime(reader, 6, modifiedAt));
                    }

                    values.Add(new HistoryModifiedValue(prior, modifiedAt,
                        (HistoryModificationType)QuestDbConnection.ReadKind(reader, 9),
                        reader.IsDBNull(10) ? string.Empty : reader.GetString(10),
                        reader.GetInt64(7), sourceTime));
                }
            }

            RecordSuccess();
            return new HistoryModifiedReadResult(values, hasMore);
        }
        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 is evaluated at its rounded
    ///     time but stamped with the requested time itself (UTC-normalised as <see cref="QuestDbConnection.Round"/> does,
    ///     not truncated), matching the SQL Server backend. 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++)
            {
                // Seek/evaluate at the rounded time; stamp the client's own requested time (M3).
                var requested = AsUtc(timestampsUtc[i]);
                result[i] = byTicks.TryGetValue(rounded[i].Ticks, out var v)
                    ? v with { SourceTimestampUtc = requested }
                    : new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, requested, now);
            }

            RecordSuccess();
            return new HistoryReadResult(result, ContinuationPoint: null);
        }
        catch (Exception ex)
        {
            RecordFailure(ex);
            throw;
        }
    }

    /// <summary>The UTC normalisation of <see cref="QuestDbConnection.Round"/> without its truncation or nudging.</summary>
    private static DateTime AsUtc(DateTime t) => t.Kind switch
    {
        DateTimeKind.Local => t.ToUniversalTime(),
        DateTimeKind.Utc => t,
        _ => DateTime.SpecifyKind(t, DateTimeKind.Utc),
    };

    /// <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>Reverse events reads are served natively (<c>ORDER BY … DESC</c>, Gitea #675).</remarks>
    public bool SupportsReverseEvents => true;

    /// <inheritdoc />
    /// <remarks>The seventeen <see cref="AggregateFamilies.Served"/> — shared with the other SQL backend (#682).</remarks>
    public IReadOnlySet<HistoryAggregateType> SupportedAggregates => AggregateFamilies.Served;


    /// <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 Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
        ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken);

    /// <inheritdoc />
    /// <remarks>
    ///     Forward is the five-argument read. Reverse (Gitea #675) is the (rounded) <c>(lo, hi]</c> ordered by
    ///     <c>(ts, event_id)</c> DESCENDING, so <c>LIMIT</c> keeps the latest events; truncation and the continuation point
    ///     work as forward.
    /// </remarks>
    public async Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, 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} " +
                (reverse ? "WHERE ts > $1 AND ts <= $2" : "WHERE ts >= $1 AND ts < $2") +
                (sourceName is null ? "" : " AND source = $3") +
                (reverse ? " ORDER BY ts DESC, event_id DESC" : " 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, at most
    ///     <see cref="BucketPlan.MaxBuckets"/> buckets. A reversed window (#677) is anchored at its later time per Part 13 /
    ///     the SDK: window <c>(end, start]</c>, buckets stepping back from <c>start</c>, the partial one earliest, each stamped
    ///     with its later edge, returned latest first. One statement per node returns the <see cref="BucketRow"/> statistics
    ///     of each non-empty bucket (bucket index <c>(cast(ts AS LONG)*10 − BucketBaseUnixTicks) / IntervalTicks</c>, never
    ///     <c>SAMPLE BY</c>); a second fetches the tag's earliest sample for the start-of-data Partial bit; and
    ///     <see cref="AggregateEvaluator.EvaluateAll"/> produces the values and fills the empty buckets with <c>BadNoData</c>. The time-weighted family (#671) instead streams the window plus four bounding
    ///     rows into <see cref="TimeWeightedAggregator"/>, exactly as the SQL Server backend does.
    /// </remarks>
    /// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not one of the seventeen 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;
            if (family == AggregateFamily.TimeWeighted)
            {
                var tw = new TimeWeightedAggregator(aggregate, plan);
                if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
                    await StreamTimeWeightedAsync(fullReference, plan, tw, cancellationToken).ConfigureAwait(false);
                var twResult = tw.Finish(now);
                RecordSuccess();
                return new HistoryReadResult(twResult, ContinuationPoint: null);
            }
            var rows = new List<BucketRow>();
            DateTime? earliest = null, latest = null;
            if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                var sample = _conn.SampleTable;
                var sql = family switch
                {
                    AggregateFamily.Simple => QuestDbAggregateQueries.Simple(sample, plan.Reversed),
                    AggregateFamily.StartEnd => QuestDbAggregateQueries.StartEnd(sample, plan.Reversed),
                    _ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed),
                };
                await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
                await using (var cmd = _conn.CreateCommand(conn, sql))
                {
                    QuestDbConnection.BindLong(cmd, plan.BucketBaseUnixTicks);
                    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))
                        rows.Add(ReadBucket(reader, family));
                }
                // The tag's data range (earliest + latest live sample): the start/end-of-data Partial rules (#677/#680 —
                // any per-sample aggregate but Average, either direction) and Count's in-range empty interval (#680); skipped
                // otherwise.
                if (AggregateEvaluator.NeedsDataRange(aggregate, plan))
                {
                    earliest = await ReadBoundAsync(conn, QuestDbAggregateQueries.EarliestSample(sample), fullReference,
                        plan.EndUtc, cancellationToken).ConfigureAwait(false);
                    latest = await ReadBoundAsync(conn, QuestDbAggregateQueries.LatestSample(sample), fullReference,
                        plan.StartUtc, cancellationToken).ConfigureAwait(false);
                }
            }

            var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, latest, now);

            RecordSuccess();
            return new HistoryReadResult(result, ContinuationPoint: null);
        }
        catch (HistoryReadLimitExceededException)
        {
            // A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
            throw;
        }
        catch (Exception ex)
        {
            RecordFailure(ex);
            throw;
        }
    }

    /// <summary>
    ///     Streams the time-weighted bounds (<see cref="QuestDbAggregateQueries.TimeWeightedBounds"/>) and window
    ///     (<see cref="QuestDbAggregateQueries.TimeWeightedWindow"/>) into <paramref name="tw"/> in time order: the two
    ///     before-bounds, the window, the two after-bounds (duplicates are dropped by the aggregator).
    /// </summary>
    private async Task StreamTimeWeightedAsync(string tag, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct)
    {
        var bounds = new StoredSample?[4];
        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
        await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.TimeWeightedBounds(_conn.SampleTable)))
        {
            QuestDbConnection.BindText(cmd, tag);
            QuestDbConnection.BindTimestamp(cmd, plan.StartUtc);
            QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
            await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
            while (await reader.ReadAsync(ct).ConfigureAwait(false))
                bounds[Convert.ToInt32(reader.GetValue(0), CultureInfo.InvariantCulture)] = ReadStored(reader, 1);
        }

        if (bounds[0] is { } usableBefore) tw.Add(usableBefore);
        if (bounds[1] is { } anyBefore) tw.Add(anyBefore);
        await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.TimeWeightedWindow(_conn.SampleTable)))
        {
            QuestDbConnection.BindText(cmd, tag);
            QuestDbConnection.BindTimestamp(cmd, plan.StartUtc);
            QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
            cmd.Parameters.Add(new NpgsqlParameter
                { NpgsqlDbType = NpgsqlDbType.Integer, Value = TimeWeightedAggregator.WindowFetchLimit(_maxTimeWeightedRawRows) });
            await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
            long windowRows = 0;
            while (await reader.ReadAsync(ct).ConfigureAwait(false))
            {
                // LIMIT $4 = cap + 1: one row past the cap fails the node without reading on.
                TimeWeightedAggregator.CheckWindowRows(++windowRows, _maxTimeWeightedRawRows, tag);
                tw.Add(ReadStored(reader, 0));
            }
        }
        if (bounds[2] is { } anyAfter) tw.Add(anyAfter);
        if (bounds[3] is { } usableAfter) tw.Add(usableAfter);
    }

    /// <summary>Runs a one-column <c>ts</c> seek (<see cref="QuestDbAggregateQueries.EarliestSample"/> / <c>LatestSample</c>).</summary>
    private async Task<DateTime?> ReadBoundAsync(NpgsqlConnection conn, string sql, string tag, DateTime edge, CancellationToken ct)
    {
        await using var cmd = _conn.CreateCommand(conn, sql);
        QuestDbConnection.BindText(cmd, tag);
        QuestDbConnection.BindTimestamp(cmd, edge);
        await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
        return await reader.ReadAsync(ct).ConfigureAwait(false) && !reader.IsDBNull(0) ? QuestDbConnection.ReadUtc(reader, 0) : null;
    }

    /// <summary>
    ///     Maps one per-sample aggregate row (<see cref="QuestDbAggregateQueries"/> column layout) to a
    ///     <see cref="BucketRow"/>. Every picked sample — first/last usable, first/last of any quality, the extreme — is
    ///     optional: a NULL time reads as "no such sample", never an exception (live-gate 8 F1: a row whose counts and
    ///     picks disagreed used to throw out of <see cref="ReadStored"/> and fail the node's whole read; the evaluator
    ///     answers such a bucket from what is present, or <c>BadNoData</c>).
    /// </summary>
    /// <param name="reader">Reader positioned on a bucket row.</param>
    /// <param name="family">The query's aggregate family.</param>
    /// <returns>The bucket statistics.</returns>
    internal static BucketRow ReadBucket(DbDataReader reader, AggregateFamily family)
    {
        static double? D(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
        static int N(DbDataReader 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(DbDataReader 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);
                row.UncertainMin = D(reader, f + 4);
                row.UncertainMax = D(reader, f + 5);
                row.GoodNonNumeric = N(reader, f + 6);
                break;
            case AggregateFamily.StartEnd:
                row.First = V(reader, f);
                row.Last = V(reader, f + 5);
                row.FirstAny = ReadStoredOrNull(reader, QuestDbAggregateQueries.FirstAnyColumn);
                row.LastAny = ReadStoredOrNull(reader, QuestDbAggregateQueries.FirstAnyColumn + 6);
                break;
            default:
                row.Extreme = D(reader, f);
                row.ExtremeTime = reader.IsDBNull(f + 1) ? null : QuestDbConnection.ReadUtc(reader, f + 1);
                row.UncertainMin = D(reader, f + 2);
                row.UncertainMax = D(reader, f + 3);
                row.GoodNonNumeric = N(reader, f + 4);
                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(DbDataReader 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>
    ///     <see cref="ReadStored"/> for an optional picked sample: a NULL <c>ts</c> means the row carries no such sample
    ///     (live-gate 8 F1), so the result is null rather than an <see cref="InvalidCastException"/>.
    /// </summary>
    private static StoredSample? ReadStoredOrNull(DbDataReader r, int o) => r.IsDBNull(o) ? null : ReadStored(r, o);

    /// <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.
    }
}