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

Merge feat/677-aggregates: reverse processed reads SDK-anchored, start-of-data Partial (forward), AggregateConfiguration accept-defaults, IntegralDivisor=1, SDK parity tests (#677)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-11 11:47:52 -04:00
parent 50c09e21d5 7349260231
commit 945ec2c069
24 changed files with 1291 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs
+30 -15
View File
@@ -9,21 +9,23 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// per <see cref="AggregateFamily"/>, mirroring the SQL Server backend's <c>SqlAggregateQueries</c> family split and
/// column order. Every query takes five positional parameters:
/// <list type="number">
/// <item><c>$1</c> — the (rounded) window start in 100 ns ticks since the Unix epoch
/// (<see cref="BucketPlan.StartUnixTicks"/>, <c>Bigint</c>);</item>
/// <item><c>$1</c> — the bucket base in 100 ns ticks since the Unix epoch
/// (<see cref="BucketPlan.BucketBaseUnixTicks"/>, <c>Bigint</c>: the rounded window start forward; re-anchored at
/// the later time for a reverse read, #677);</item>
/// <item><c>$2</c> — the bucket width in ticks (<see cref="BucketPlan.IntervalTicks"/>, <c>Bigint</c>);</item>
/// <item><c>$3</c> — the tag (<c>Varchar</c>);</item>
/// <item><c>$4</c>/<c>$5</c> — the forward window <c>[start, end)</c> (<c>TimestampTz</c> via
/// <see cref="QuestDbConnection.BindTimestamp"/>).</item>
/// <item><c>$4</c>/<c>$5</c> — the earlier / later window edges (<c>TimestampTz</c> via
/// <see cref="QuestDbConnection.BindTimestamp"/>): <c>[$4, $5)</c> forward, <c>($4, $5]</c> reverse.</item>
/// </list>
/// Each returns one row per <b>non-empty</b> bucket; the caller fills the empty ones.
/// </summary>
/// <remarks>
/// <para>
/// <b>Bucketing.</b> <c>b = (cast(ts AS LONG) * 10 − $1) / $2</c>. <c>cast(ts AS LONG)</c> is µs since 1970, so
/// <c>× 10</c> is ticks since 1970; both it and <c>$1</c> are µs-aligned (R6), so the integer division is
/// start-inclusive / end-exclusive to the tick, exactly like SQL Server's. Every row is at or after the start,
/// so the dividend is never negative (no truncation-toward-zero surprise), and nothing overflows before 9999.
/// <c>× 10</c> is ticks since 1970, so the integer division is exact to the tick, exactly like SQL Server's
/// (<see cref="BucketPlan.IndexOf"/> is the C# statement of the math). Every in-window row is at or after the base
/// (forward) or strictly after the window start with the base one tick past the grid origin (reverse), so the
/// dividend is never negative (no truncation-toward-zero surprise), and nothing overflows before 9999.
/// <c>SAMPLE BY</c> is deliberately not used: its alignment rules differ from ours.
/// </para>
/// <para>
@@ -60,13 +62,14 @@ internal static class QuestDbAggregateQueries
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <returns>The SQL text.</returns>
public static string Simple(string sample) =>
/// <param name="reverse">True for a reverse read (window <c>($4, $5]</c>).</param>
public static string Simple(string sample, bool reverse = false) =>
"SELECT b, " + CommonCounts() + ", " +
"avg(CASE WHEN u = 1 THEN num_value END) AS avg_value, " +
"min(CASE WHEN u = 1 THEN num_value END) AS min_value, " +
"max(CASE WHEN u = 1 THEN num_value END) AS max_value, " +
"sum(CASE WHEN u = 1 THEN num_value END) AS sum_value " +
"FROM (" + Classified(sample) + ") GROUP BY b ORDER BY b";
"FROM (" + Classified(sample, reverse) + ") GROUP BY b ORDER BY b";
/// <summary>
/// Start / End / Delta: the first and last usable sample of each bucket by <c>ts</c>.
@@ -76,8 +79,9 @@ internal static class QuestDbAggregateQueries
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <returns>The SQL text.</returns>
public static string StartEnd(string sample) =>
"WITH s AS (" + Classified(sample) + "), " +
/// <param name="reverse">True for a reverse read (window <c>($4, $5]</c>).</param>
public static string StartEnd(string sample, bool reverse = false) =>
"WITH s AS (" + Classified(sample, reverse) + "), " +
"st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b), " +
"f AS (SELECT b, kind, num_value, int_value, str_value, ts, " +
"row_number() OVER (PARTITION BY b ORDER BY ts ASC) AS rf, " +
@@ -96,8 +100,9 @@ internal static class QuestDbAggregateQueries
/// <param name="sample">Prefixed sample table name.</param>
/// <param name="maximum">True for MaximumActualTime.</param>
/// <returns>The SQL text.</returns>
public static string ActualTime(string sample, bool maximum) =>
"WITH s AS (" + Classified(sample) + "), " +
/// <param name="reverse">True for a reverse read (window <c>($4, $5]</c>).</param>
public static string ActualTime(string sample, bool maximum, bool reverse = false) =>
"WITH s AS (" + Classified(sample, reverse) + "), " +
"st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b), " +
"r AS (SELECT b, num_value, ts, " +
"row_number() OVER (PARTITION BY b ORDER BY num_value " + (maximum ? "DESC" : "ASC") + ", ts ASC) AS rn " +
@@ -144,12 +149,22 @@ internal static class QuestDbAggregateQueries
/// <summary>The bucket-index expression (exposed for tests).</summary>
internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2";
/// <summary>
/// The tag's earliest live sample (any quality) before the window end: the start-of-data Partial rule (#677; forward
/// reads only, so the forward window's exclusive end bounds the scan). Parameters <c>$1</c> tag, <c>$2</c> the window
/// end. Column: <c>ts</c>; no row when the tag has none before the end.
/// </summary>
/// <param name="sample">Prefixed sample table name.</param>
/// <returns>The SQL text.</returns>
public static string EarliestSample(string sample) =>
$"SELECT ts FROM {sample} WHERE tag = $1 AND ts < $2 AND NOT deleted ORDER BY ts ASC LIMIT 1";
/// <summary>The classified subquery <c>s</c>: the window's live samples with bucket index and the usable/good flags.</summary>
private static string Classified(string sample) =>
private static string Classified(string sample, bool reverse) =>
"SELECT " + BucketExpression + " AS b, ts, kind, num_value, int_value, str_value, " +
$"CASE WHEN status < {NotBadLimit} AND kind <> {EmptyKind} THEN 1 ELSE 0 END AS u, " +
$"CASE WHEN status < {GoodLimit} AND kind <> {EmptyKind} THEN 1 ELSE 0 END AS g " +
$"FROM {sample} WHERE tag = $3 AND ts >= $4 AND ts < $5 AND NOT deleted";
$"FROM {sample} WHERE tag = $3 AND " + (reverse ? "ts > $4 AND ts <= $5" : "ts >= $4 AND ts < $5") + " AND NOT deleted";
/// <summary>The five common count columns (after <c>b</c>).</summary>
private static string CommonCounts() =>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs
+33 -21
View File
@@ -440,12 +440,13 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
/// <remarks>
/// The (rounded) window is planned by <see cref="BucketPlan.Create"/>: bucket <c>i</c> =
/// <c>[start + i·interval, start + (i+1)·interval)</c>, start-inclusive / end-exclusive, the last bucket truncated
/// at <c>end</c>, interval &lt;= 0 ⇒ one bucket, <c>start == end</c> ⇒ none, a reversed window computed forward
/// and returned descending, at most <see cref="BucketPlan.MaxBuckets"/> buckets. One statement per node returns
/// the <see cref="BucketRow"/> statistics of each non-empty bucket (bucket index
/// <c>(cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks</c>, never <c>SAMPLE BY</c>);
/// <see cref="AggregateEvaluator.Evaluate"/> produces each value and <see cref="BucketPlan.Assemble"/> fills the
/// empty buckets with <c>BadNoData</c>. The time-weighted family (#671) instead streams the window plus four bounding
/// 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>
@@ -461,7 +462,6 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
try
{
var now = _time.GetUtcNow().UtcDateTime;
var evaluated = new Dictionary<long, DataValueSnapshot>();
if (family == AggregateFamily.TimeWeighted)
{
var tw = new TimeWeightedAggregator(aggregate, plan);
@@ -471,31 +471,43 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
RecordSuccess();
return new HistoryReadResult(twResult, ContinuationPoint: null);
}
var rows = new List<BucketRow>();
DateTime? earliest = 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),
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.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))
await using (var cmd = _conn.CreateCommand(conn, sql))
{
var row = ReadBucket(reader, family);
evaluated[row.Bucket] = AggregateEvaluator.Evaluate(aggregate, row, plan.BucketStart(row.Bucket), now);
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 earliest live sample before the window end: the start-of-data Partial rule (#677) — forward
// reads of an aggregate other than Average only; skipped otherwise.
if (AggregateEvaluator.NeedsEarliestSample(aggregate, plan))
{
await using var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.EarliestSample(sample));
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, plan.EndUtc);
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false) && !reader.IsDBNull(0))
earliest = QuestDbConnection.ReadUtc(reader, 0);
}
}
var result = plan.Assemble(evaluated, bucketStart => AggregateEvaluator.NoData(bucketStart, now));
var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, now);
RecordSuccess();
return new HistoryReadResult(result, ContinuationPoint: null);