Joseph Doherty
56bb4855bb
M1: the time-weighted family (incl. Total, formerly a pushed-down SUM) streams raw window rows into the C# evaluator. New shared knob ServerHistorian:MaxTimeWeightedRawRows (default 1,000,000; <=0 unbounded with a Validate warning) flows through HistorianBackendRegistry into both providers. The window query is capped at cap+1 (SQL Server TOP (@maxRows), QuestDB LIMIT $4); one row past the cap throws the new Core.Abstractions HistoryReadLimitExceededException, which the node manager maps to BadTooManyOperations for that node. Not counted as a backend health failure; bound rows are not counted; per-sample aggregates are not capped. M2: reverse Interpolative is anchored at the later time (Part 13): points at End - i*interval, emitted latest first. 10:00->09:00 @1h yields 10:00. L1: Partial bit (0x4, = Opc.Ua.AggregateBits.Partial) on non-Bad results of a truncated last interval (time-weighted family; per-sample family is a follow-up). L2/L3: aligned with the SDK AggregateCalculator.GetTimeBasedStatusCode (decompiled from Opc.Ua.Server 1.5.378.106): TreatUncertainAsBad=false counts Uncertain time (Uncertain samples, held-into-outage, extrapolation) as Good time; data but no non-Bad time => generic Bad (0x80000000); BadNoData only when there is no data at all. Consequence: a live tag's latest interval reads Good (documented). L5: release note no longer claims Total matches the Gateway Integral exactly (AVEVA's divisor is unverified); it notes Total is now evaluated in C# and subject to the cap. Tests: unit (reverse Interpolative x3, Partial, Bad vs BadNoData incl. an open-ended outage, cap helpers, SDK bit oracles); conformance (reverse Interpolative, raw-row cap at/over/0, bounds not counted, health unaffected); node manager BadTooManyOperations; options default + warning. Each new behaviour falsified by mutation (aggregator x9, data-source cap checks on both backends against the real DBs, node-manager catch).
544 lines
30 KiB
C#
544 lines
30 KiB
C#
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 (<= 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 < end</c>): <c>[start, end)</c> ascending. Reverse (<c>end < start</c>):
|
||
/// <c>(end, start]</c> descending. <c>start == end</c>: the sample exactly at that (rounded) time.
|
||
/// <paramref name="maxValuesPerNode"/> <c>0</c> is "unbounded" (<see cref="int.MaxValue"/>); larger values clamp.
|
||
/// At most <c>top</c> samples, <c>ContinuationPoint</c> always null (R8): the node manager pages by re-reading
|
||
/// from the last returned timestamp inclusive, which the <c>(ts, tag)</c> dedup key makes exact.
|
||
/// </remarks>
|
||
public async Task<HistoryReadResult> ReadRawAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(fullReference);
|
||
var top = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue);
|
||
try
|
||
{
|
||
var samples = new List<DataValueSnapshot>();
|
||
if (await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var start = QuestDbConnection.Round(startUtc);
|
||
var end = QuestDbConnection.Round(endUtc);
|
||
string where, order;
|
||
if (start == end) { where = "ts = $2 AND ts = $3"; order = "ASC"; }
|
||
else if (start < end) { where = "ts >= $2 AND ts < $3"; order = "ASC"; }
|
||
else { where = "ts > $2 AND ts <= $3"; order = "DESC"; }
|
||
var (lo, hi) = start <= end ? (start, end) : (end, start);
|
||
|
||
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||
await using var cmd = _conn.CreateCommand(conn,
|
||
$"SELECT {SampleColumns} FROM {_conn.SampleTable} " +
|
||
$"WHERE tag = $1 AND {where} AND NOT deleted ORDER BY ts {order} LIMIT $4");
|
||
QuestDbConnection.BindText(cmd, fullReference);
|
||
QuestDbConnection.BindTimestamp(cmd, lo);
|
||
QuestDbConnection.BindTimestamp(cmd, hi);
|
||
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = top });
|
||
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
||
var now = _time.GetUtcNow().UtcDateTime;
|
||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var s = ReadStored(reader, 0);
|
||
samples.Add(new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc, ReadServerTime(reader, 6, now)));
|
||
}
|
||
}
|
||
|
||
RecordSuccess();
|
||
return new HistoryReadResult(samples, ContinuationPoint: null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordFailure(ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ AtTime
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// Each distinct (rounded) requested time picks up its previous-or-equal and next live sample through two
|
||
/// <c>LIMIT 1</c> seeks, <c>UNION ALL</c>-ed, at most <see cref="AtTimeChunkSize"/> times per statement (spec §3,
|
||
/// T0 Q9). <see cref="AtTimeEvaluator.Evaluate"/> then decides each value exactly as for SQL Server (exact hit,
|
||
/// no previous ⇒ <c>BadNoData</c>, inside an outage, held before an outage, linear, stepped, after now). The
|
||
/// result matches the request in length and order (duplicates included). Each entry 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>
|
||
/// <c>[start, end)</c> by event time (a reversed window is swapped), ordered by <c>(ts, event_id)</c> ascending.
|
||
/// <paramref name="maxEvents"/> <= 0 or <see cref="int.MaxValue"/> ⇒ unbounded. Otherwise
|
||
/// <c>maxEvents + 1</c> rows are fetched; when the extra row arrives it is dropped and the continuation point is
|
||
/// the UTF-8 of <see cref="EventContinuation.Encode"/> for that first not-returned event.
|
||
/// </remarks>
|
||
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken)
|
||
{
|
||
// No stored event can carry a source longer than the limit, so an over-long filter matches nothing.
|
||
if (sourceName is not null && sourceName.Length > HistorianTagNames.MaxLength)
|
||
return new HistoricalEventsResult([], null);
|
||
try
|
||
{
|
||
await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
|
||
var start = QuestDbConnection.Round(startUtc);
|
||
var end = QuestDbConnection.Round(endUtc);
|
||
var (lo, hi) = start <= end ? (start, end) : (end, start);
|
||
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
|
||
var sql =
|
||
$"SELECT event_id, source, ts, received_utc, message, severity FROM {_conn.EventTable} " +
|
||
"WHERE ts >= $1 AND ts < $2" +
|
||
(sourceName is null ? "" : " AND source = $3") +
|
||
" ORDER BY ts ASC, event_id ASC" +
|
||
(bounded ? (sourceName is null ? " LIMIT $3" : " LIMIT $4") : "");
|
||
|
||
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||
await using var cmd = _conn.CreateCommand(conn, sql);
|
||
QuestDbConnection.BindTimestamp(cmd, lo);
|
||
QuestDbConnection.BindTimestamp(cmd, hi);
|
||
if (sourceName is not null) QuestDbConnection.BindText(cmd, sourceName);
|
||
if (bounded) QuestDbConnection.BindLong(cmd, (long)maxEvents + 1);
|
||
|
||
var events = new List<HistoricalEvent>();
|
||
byte[]? continuation = null;
|
||
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
|
||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||
{
|
||
var eventTime = QuestDbConnection.ReadUtc(reader, 2);
|
||
var e = new HistoricalEvent(
|
||
reader.IsDBNull(0) ? "" : reader.GetString(0),
|
||
reader.IsDBNull(1) ? null : reader.GetString(1),
|
||
eventTime,
|
||
reader.IsDBNull(3) ? eventTime : QuestDbConnection.ReadUtc(reader, 3),
|
||
reader.IsDBNull(4) ? null : reader.GetString(4),
|
||
HistoricalEventRows.ClampSeverity(reader.IsDBNull(5) ? 0 : reader.GetInt32(5)));
|
||
if (bounded && events.Count == maxEvents)
|
||
{
|
||
continuation = Encoding.UTF8.GetBytes(EventContinuation.Encode(e.EventTimeUtc, e.EventId));
|
||
break;
|
||
}
|
||
events.Add(e);
|
||
}
|
||
|
||
RecordSuccess();
|
||
return new HistoricalEventsResult(events, continuation);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordFailure(ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ Processed
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// The (rounded) window is planned by <see cref="BucketPlan.Create"/>: bucket <c>i</c> =
|
||
/// <c>[start + i·interval, start + (i+1)·interval)</c>, start-inclusive / end-exclusive, the last bucket truncated
|
||
/// at <c>end</c>, interval <= 0 ⇒ one bucket, <c>start == end</c> ⇒ none, a reversed window computed forward
|
||
/// and returned descending, at most <see cref="BucketPlan.MaxBuckets"/> buckets. One statement per node returns
|
||
/// the <see cref="BucketRow"/> statistics of each non-empty bucket (bucket index
|
||
/// <c>(cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks</c>, never <c>SAMPLE BY</c>);
|
||
/// <see cref="AggregateEvaluator.Evaluate"/> produces each value and <see cref="BucketPlan.Assemble"/> fills the
|
||
/// empty buckets with <c>BadNoData</c>. 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;
|
||
var evaluated = new Dictionary<long, DataValueSnapshot>();
|
||
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);
|
||
}
|
||
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 (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);
|
||
}
|
||
|
||
private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family)
|
||
{
|
||
static double? D(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
|
||
static int N(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? 0 : Convert.ToInt32(r.GetValue(i), CultureInfo.InvariantCulture);
|
||
// A bucket without a usable sample has NULL times but kind 0 (BYTE has no NULL): absence is keyed on the time.
|
||
static HistorianValue V(NpgsqlDataReader r, int i) =>
|
||
r.IsDBNull(i + 4)
|
||
? HistorianValue.Empty
|
||
: ToValue(QuestDbConnection.ReadKind(r, i), D(r, i + 1), r.IsDBNull(i + 2) ? null : r.GetInt64(i + 2),
|
||
r.IsDBNull(i + 3) ? null : r.GetString(i + 3));
|
||
|
||
var row = new BucketRow
|
||
{
|
||
Bucket = Convert.ToInt64(reader.GetValue(0), CultureInfo.InvariantCulture),
|
||
Total = N(reader, 1),
|
||
Usable = N(reader, 2),
|
||
Good = N(reader, 3),
|
||
Bad = N(reader, 4),
|
||
NonNumeric = N(reader, 5),
|
||
};
|
||
var f = QuestDbAggregateQueries.FirstFamilyColumn;
|
||
switch (family)
|
||
{
|
||
case AggregateFamily.Simple:
|
||
row.Avg = D(reader, f);
|
||
row.Min = D(reader, f + 1);
|
||
row.Max = D(reader, f + 2);
|
||
row.Sum = D(reader, f + 3);
|
||
break;
|
||
case AggregateFamily.StartEnd:
|
||
row.First = V(reader, f);
|
||
row.Last = V(reader, f + 5);
|
||
break;
|
||
default:
|
||
row.Extreme = D(reader, f);
|
||
row.ExtremeTime = reader.IsDBNull(f + 1) ? null : QuestDbConnection.ReadUtc(reader, f + 1);
|
||
break;
|
||
}
|
||
return row;
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------------------ Shared
|
||
|
||
/// <summary>
|
||
/// Reads <c>ts, kind, num_value, int_value, str_value, status</c> starting at <paramref name="o"/> and applies the
|
||
/// shared quality clamp (<see cref="StoredSample.FromStored"/>).
|
||
/// </summary>
|
||
private static StoredSample ReadStored(NpgsqlDataReader r, int o)
|
||
{
|
||
var value = ToValue(
|
||
QuestDbConnection.ReadKind(r, o + 1),
|
||
r.IsDBNull(o + 2) ? null : r.GetDouble(o + 2),
|
||
r.IsDBNull(o + 3) ? null : r.GetInt64(o + 3),
|
||
r.IsDBNull(o + 4) ? null : r.GetString(o + 4));
|
||
// status is always written; a NULL (only possible for a hand-made row) is served as "no data", never Good.
|
||
var status = r.IsDBNull(o + 5) ? HistorianStatusCodes.BadNoData : SampleColumnMapper.FromLongStatus(r.GetInt64(o + 5));
|
||
return StoredSample.FromStored(QuestDbConnection.ReadUtc(r, o), value, status);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Rebuilds a value from its stored columns via the shared <see cref="SampleColumnMapper.FromColumns"/>, after the
|
||
/// RC2 repair: QuestDB reads a <c>LONG</c> holding <see cref="long.MinValue"/> back as NULL, so an Int64 row with
|
||
/// a NULL <c>int_value</c> and a <c>num_value</c> of exactly −2^63 is that value.
|
||
/// </summary>
|
||
internal static HistorianValue ToValue(byte kind, double? num, long? integer, string? str)
|
||
{
|
||
if (kind == (byte)HistorianValueKind.Int64 && integer is null && num == LongNullSentinelAsDouble)
|
||
integer = long.MinValue;
|
||
return SampleColumnMapper.FromColumns(kind, num, integer, str);
|
||
}
|
||
|
||
private static DateTime ReadServerTime(NpgsqlDataReader r, int i, DateTime fallback) =>
|
||
r.IsDBNull(i) ? fallback : QuestDbConnection.ReadUtc(r, i);
|
||
|
||
/// <summary>Read-only tag lookup (R7; never creates a row); runs the schema bootstrap on a cache miss.</summary>
|
||
private Task<bool> ResolveAsync(string name, CancellationToken ct) => _catalog.ExistsAsync(name, ct);
|
||
|
||
// ------------------------------------------------------------------------------------------ Health
|
||
|
||
/// <inheritdoc />
|
||
public HistorianHealthSnapshot GetHealthSnapshot() => _health.Snapshot();
|
||
|
||
private void RecordSuccess() => _health.RecordSuccess();
|
||
|
||
private void RecordFailure(Exception ex)
|
||
{
|
||
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
|
||
if (ex is OperationCanceledException) return;
|
||
_health.RecordFailure(ex);
|
||
_log.LogDebug("QuestDB historian read failed ({ExceptionType})", ex.GetType().Name);
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
public void Dispose()
|
||
{
|
||
// Connections are per call (pooled by the shared NpgsqlDataSource, owned by the provider); nothing to release.
|
||
}
|
||
}
|