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
{
    // QuestDB stores microseconds: 10 ticks. HistoryRead paging steps off a drained timestamp by this much.
    /// <inheritdoc />
    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;
    private readonly int _maxRawValuesPerRead;

    /// <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>
    /// <param name="maxRawValuesPerRead">
    ///     The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one
    ///     <see cref="ReadRawAsync"/> may return — reported as <see cref="MaxRawValuesPerRead"/>, so the server pages at it.
    ///     A read asking for more (or for everything) is refused when the window holds more, never truncated.
    ///     &lt;= 0 = unbounded (the pre-#736 behaviour).
    /// </param>
    internal QuestDbHistorianDataSource(
        QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, ILogger? log = null,
        TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows,
        int maxRawValuesPerRead = 0)
    {
        _maxTimeWeightedRawRows = maxTimeWeightedRawRows;
        _maxRawValuesPerRead = Math.Max(0, maxRawValuesPerRead);
        _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. maxValuesPerNode 0 is "unbounded"
    // (int.MaxValue); larger values clamp. At most top samples, ContinuationPoint always null: the node
    // manager pages by re-reading from the last returned timestamp inclusive, which the (ts, tag) dedup
    // key makes exact.
    // The server raw cap (R08:1.1, #736): with a cap, a read asking for everything or for more than the cap reads
    // cap + 1 rows — one more than it may return — so an over-cap window is DETECTED and refused
    // (HistoryReadLimitExceededException ⇒ BadTooManyOperations) rather than answered truncated: the server's paging
    // reads a short answer as the last page, so a silent LIMIT would lose data. The server itself never asks for more
    // than the cap (it pages at MaxRawValuesPerRead); this bounds a caller that bypasses it.
    /// <inheritdoc />
    public async Task<HistoryReadResult> ReadRawAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(fullReference);
        var cap = _maxRawValuesPerRead;
        var probe = cap > 0 && (maxValuesPerNode == 0 || maxValuesPerNode > (uint)cap);
        var top = probe
            ? (int)Math.Min(int.MaxValue, (long)cap + 1)
            : 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)));
                }
            }

            if (probe && samples.Count > cap)
                throw new HistoryReadLimitExceededException(cap,
                    $"The raw read of '{fullReference}' holds more than ServerHistorian:MaxRawValuesPerRead ({cap}) samples; " +
                    "set NumValuesPerNode at or below it to page, or narrow the window.");

            RecordSuccess();
            return new HistoryReadResult(samples, ContinuationPoint: null);
        }
        catch (HistoryReadLimitExceededException)
        {
            // A client-request limit (the server raw cap), not a backend fault: not counted as a failure.
            throw;
        }
        catch (Exception ex)
        {
            RecordFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
            throw;
        }
    }

    // ------------------------------------------------------------------------------------------ Modified (Part 11)

    // Reads {prefix}sample_modified, the append-only trail QuestDbHistorianDataUpdater writes BEFORE
    // each change. Windows follow ReadRawAsync (bounds rounded to microseconds); rows are ordered by
    // (ts, modified_id); maxValues + 1 rows are fetched to report HasMore exactly. A row whose status is
    // NULL had no prior value (an insert) and is served null / BadNoData. 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).
    /// <inheritdoc />
    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);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", 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 AtTimeChunkSize times per statement.
    // AtTimeEvaluator.Evaluate 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 is evaluated at its
    // rounded time but stamped with the requested time itself (UTC-normalised as
    // QuestDbConnection.Round does, not truncated), matching the SQL Server backend. Unknown tag =>
    // every entry BadNoData.
    /// <inheritdoc />
    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.
                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);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", 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>
    /// <param name="sample">The prefixed sample table name.</param>
    /// <param name="count">The number of request times in the statement.</param>
    /// <returns>The SQL text.</returns>
    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

    // Reverse raw reads are served natively ((end, start], ORDER BY ... DESC).
    /// <inheritdoc />
    public bool SupportsReverseRaw => true;

    // The server raw cap (ServerHistorian:MaxRawValuesPerRead; 0 = none): the server pages every raw read at it, so a
    // client's NumValuesPerNode = 0 is served as pages of at most this many values. QuestDB itself has no row ceiling -
    // a capped raw read is one bounded statement.
    /// <inheritdoc />
    public int MaxRawValuesPerRead => _maxRawValuesPerRead;

    // Reverse events reads are served natively (ORDER BY ... DESC).
    /// <inheritdoc />
    public bool SupportsReverseEvents => true;

    // The seventeen AggregateFamilies.Served - shared with the other SQL backend.
    /// <inheritdoc />
    public IReadOnlySet<HistoryAggregateType> SupportedAggregates => AggregateFamilies.Served;


    // [start, end) by event time (a reversed window is swapped), ordered by (ts, event_id) ascending.
    // maxEvents <= 0 or int.MaxValue => unbounded. Otherwise maxEvents + 1 rows are fetched; when the
    // extra row arrives it is dropped and the continuation point is the UTF-8 of EventContinuation.Encode
    // for that first not-returned event.
    /// <inheritdoc />
    public Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
        ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken);

    // Forward is the five-argument read. Reverse is the (rounded) (lo, hi] ordered by (ts, event_id)
    // DESCENDING, so LIMIT keeps the latest events; truncation and the continuation point work as
    // forward.
    /// <inheritdoc />
    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);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
            throw;
        }
    }

    // ------------------------------------------------------------------------------------------ Processed

    // The (rounded) window is planned by BucketPlan.Create: 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, at most BucketPlan.MaxBuckets buckets. A
    // reversed window is anchored at its later time per Part 13 / the SDK: window (end, start], buckets
    // stepping back from start, the partial one earliest, each stamped with its later edge, returned
    // latest first. One statement per node returns the BucketRow statistics of each non-empty bucket
    // (bucket index (cast(ts AS LONG)*10 - BucketBaseUnixTicks) / IntervalTicks, never SAMPLE BY);
    // before it, when AggregateEvaluator.NeedsDataRange, two tag-index seeks fetch the tag's data range
    // (earliest / latest live sample) for the start/end-of-data Partial bits and Count's in-range empty
    // interval; and AggregateEvaluator.EvaluateAll produces the values and fills the empty buckets with
    // BadNoData. The time-weighted family instead streams the window plus four bounding rows into
    // TimeWeightedAggregator, exactly as the SQL Server backend does.
    /// <inheritdoc />
    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);
                // The tag's data range (earliest + latest live sample): the start/end-of-data Partial rules
                // (any per-sample aggregate but Average, either direction) and Count's in-range empty interval; skipped
                // otherwise. Index seeks on tag. Read BEFORE the bucket statement, which is a separate table reader:
                // a sample appended in between (the common race — a live tag read up to now) then reaches the buckets but
                // not the range — at worst a stale end-of-data Partial or a BadNoData where a 0 belongs — instead of a
                // range covering a bucket the rows do not show, which reported a Good Count of 0 for an interval that has
                // samples. (A concurrent HistoryUpdate delete is the mirror case and can still do that, transiently.)
                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);
                }
                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));
                }
            }

            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);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", 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>
    /// <param name="kind">The stored value-kind discriminator.</param>
    /// <param name="num">The stored <c>num_value</c> column.</param>
    /// <param name="integer">The stored <c>int_value</c> column.</param>
    /// <param name="str">The stored <c>str_value</c> column.</param>
    /// <returns>The reconstructed historian value.</returns>
    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();

    /// <summary>Classifies provider transport/timeouts, keeping SQL/query errors distinct.</summary>
    private static bool IsTransportFailure(Exception ex) => ex is NpgsqlException { IsTransient: true } and not PostgresException
        || ex is PostgresException pg && (pg.SqlState.StartsWith("08", StringComparison.Ordinal)
            || pg.SqlState is "57P01" or "57P02" or "57P03");

    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);
    }

    /// <summary>Releases nothing: connections are per call and owned by the shared data source.</summary>
    public void Dispose()
    {
        // Connections are per call (pooled by the shared NpgsqlDataSource, owned by the provider); nothing to release.
    }
}