fix(historian): live-gate 8 F1 — per-sample counts and picks from one table version (#680 follow-up)
QuestDB Start/End/Delta reads of a tag being written intermittently failed BadHistoryOperationUnsupported (InvalidCastException: Column 'ts_3' is null): the StartEnd query referenced its CTE four times and QuestDB opens a table reader per reference, so a row committed between two of them was counted but not picked. MinimumActualTime/MaximumActualTime had the same race and returned a value-less Good. (a) SQL - QuestDB StartEnd: one scan, one GROUP BY — first()/last() for the any-quality picks, arg_min/arg_max keyed on the usable rows' time for the usable picks (no str_value: arg_min has no VARCHAR form; only Delta reads them). Byte-identical output to the old query on a 1M-row tag, ~15% faster. - SqlServer StartEnd: one windowed pass + conditional MAX (was five CTE references); identical output, 1.4-3x faster. - ActualTime kept on both backends: every single-scan form measured 1.2-2.4x slower; its inconsistency is absorbed by (b). (b) Defence in depth on both backends - ReadBucket takes DbDataReader and reads absent picks as null (ReadStoredOrNull) instead of throwing. - AggregateEvaluator: Delta with a missing first/last usable pick and ActualTime with a missing extreme are BadNoData (were BadAggregateInvalidInputs / a value-less Good); Start/End with no pick already were. - Unit tests (Sql.Common evaluator, QuestDb + SqlServer mappers over a DataTableReader) fail with the guards removed. LiveWriteReadRaceStressTests (QuestDB integration, opt-in via HISTORIAN_QUESTDB_STRESS_SECONDS, Category=SlowUnit): single-row writer vs 4 tight-loop readers. Old code: 6 faults in ~105k reads; new: 0 in 106k. Docs: Historian.md QuestDB processed-read note; F2 (Min/Max on the interval start: SDK raw, ours Calculated) and F3 (integer tags: SDK casts interpolated values to the integer type, ours Double) added to the SDK-parity deviations.
This commit is contained in:
@@ -1172,6 +1172,14 @@ non-analog value is held; the simple bound next to an Uncertain sample takes the
|
|||||||
them as numbers — e.g. an interval that is `false` for its first half and `true` for its second averages 0.5 here, 1 in the SDK, and a
|
them as numbers — e.g. an interval that is `false` for its first half and `true` for its second averages 0.5 here, 1 in the SDK, and a
|
||||||
PercentGood region ahead of an Uncertain Boolean is Good here, Uncertain in the SDK. Each is pinned by `TimeWeightedAggregatorTests` and
|
PercentGood region ahead of an Uncertain Boolean is Good here, Uncertain in the SDK. Each is pinned by `TimeWeightedAggregatorTests` and
|
||||||
by a parity dataset (`OutageLast`, `BoolSwitch`) whose masks are documented in `SdkAggregateParityTests`.
|
by a parity dataset (`OutageLast`, `BoolSwitch`) whose masks are documented in `SdkAggregateParityTests`.
|
||||||
|
Two more, found by the #680 live gate (both cosmetic, both kept deliberately): when an interval's **Minimum/Maximum sample sits exactly
|
||||||
|
on the interval's start**, the SDK reports that result as a raw value (no Calculated/InfoType bits, e.g. `0x00000000` / `0x40A40000`)
|
||||||
|
while ours keeps the Calculated bit (`0x00000401` / `0x40A40401`) — value and timestamp are identical, and ours is consistent with
|
||||||
|
every other interval of the same read. And on an **integer-typed tag** (e.g. a `UInt32`) the SDK casts interpolated and sloped
|
||||||
|
bounding values back to the source integer type, so its `Interpolative` returns `758` where ours returns `758.0459…`, and its
|
||||||
|
`TimeAverage`/`TimeAverage2`/`Total`/`Total2` differ from ours by the truncated bound values (~0.002–0.02 on the gate's data). Ours
|
||||||
|
always computes and returns `Double` (as stated above for every aggregate but Count and Start/End), which is the exact Part 13
|
||||||
|
value; the parity datasets are all `Double`, so this one is not masked there but recorded here.
|
||||||
|
|
||||||
**Pushdown trade-off.** Unlike the per-sample aggregates, the time-weighted family is **not** pushed down to SQL: the backend streams
|
**Pushdown trade-off.** Unlike the per-sample aggregates, the time-weighted family is **not** pushed down to SQL: the backend streams
|
||||||
the window's rows plus the four bounds (one batch of five result sets on SQL Server; a bounds `UNION ALL` + a window select on QuestDB)
|
the window's rows plus the four bounds (one batch of five result sets on SQL Server; a bounds `UNION ALL` + a window select on QuestDB)
|
||||||
@@ -1408,6 +1416,18 @@ Every query filters `NOT deleted`; every value goes through `StoredSample`, so t
|
|||||||
(#677/#680). `tag` is an unindexed SYMBOL, so these are partition scans (a few ms on a live tag; tens of ms cold for a stopped tag on a
|
(#677/#680). `tag` is an unindexed SYMBOL, so these are partition scans (a few ms on a live tag; tens of ms cold for a stopped tag on a
|
||||||
large table — see *Data-range queries* above). Start/End/Delta also return each bucket's first and last sample of any quality (`row_number()` over every row).
|
large table — see *Data-range queries* above). Start/End/Delta also return each bucket's first and last sample of any quality (`row_number()` over every row).
|
||||||
Min/max times use `row_number()` windows with the earliest-timestamp tie-break.
|
Min/max times use `row_number()` windows with the earliest-timestamp tie-break.
|
||||||
|
⚠️ **Live-gate 8 F1 — counts and picks must come from one table version.** QuestDB opens a separate table reader per reference to a
|
||||||
|
CTE, so the earlier Start/End/Delta form (counts CTE + self-joins) could count a row committed between two references without
|
||||||
|
finding it — a NULL first-of-any-quality time that threw and failed the node's read (~1 in 90k reads of a tag being written). That
|
||||||
|
query now reads the table **once**: one `GROUP BY` with `first()`/`last()` for the first/last sample of any quality and
|
||||||
|
`arg_min`/`arg_max` keyed on the usable rows' time for the first/last usable one (~15 % faster than before; the usable picks carry
|
||||||
|
no `str_value` — `arg_min` has no VARCHAR form, and only Delta reads them, never with a usable string in the bucket). On `SqlServer`
|
||||||
|
(READ COMMITTED gives no cross-reference guarantee either, though it was never observed there) Start/End/Delta was reshaped the same
|
||||||
|
way with a windowed pass + conditional `MAX` — 1.4–3× *faster* than the old five-reference form on a 1 M-row tag. Min/MaxActualTime
|
||||||
|
keep their counts + filtered-rank form on both backends: every single-scan variant measured 1.2–2.4× slower, and QuestDB's
|
||||||
|
`arg_max` cannot express the earliest-time tie-break. So, on both backends, a row whose picks are absent anyway maps to "no such
|
||||||
|
sample" — `BadNoData` for that interval, never an exception and never a value-less Good. `LiveWriteReadRaceStressTests` (QuestDB
|
||||||
|
integration, opt-in via `HISTORIAN_QUESTDB_STRESS_SECONDS`) interleaves a single-row writer with tight-loop reads to pin it.
|
||||||
|
|
||||||
### Retention (TTL)
|
### Retention (TTL)
|
||||||
|
|
||||||
|
|||||||
@@ -83,33 +83,54 @@ internal static class QuestDbAggregateQueries
|
|||||||
/// (<c>ts, kind, num_value, int_value, str_value, status</c>) and <c>22..27</c> the last (same order), present for
|
/// (<c>ts, kind, num_value, int_value, str_value, status</c>) and <c>22..27</c> the last (same order), present for
|
||||||
/// every returned (non-empty) bucket.
|
/// every returned (non-empty) bucket.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// <b>One scan, one GROUP BY (live-gate 8 F1).</b> The counts and all four picked samples come from a single read
|
||||||
|
/// of the sample table. The first/last sample of any quality are QuestDB's <c>first()</c>/<c>last()</c> (the
|
||||||
|
/// first/last row of the group in designated-timestamp order — null values included, so every column comes from
|
||||||
|
/// the same row); the first/last <i>usable</i> sample are <c>arg_min</c>/<c>arg_max</c> keyed on <c>uts</c>, the
|
||||||
|
/// row's time when usable and NULL otherwise (a NULL key is skipped, a NULL value is returned as is, and
|
||||||
|
/// <c>(ts, tag)</c> is unique — DEDUP — so there are no ties). <c>arg_min</c> has no VARCHAR form, so the usable
|
||||||
|
/// picks carry no <c>str_value</c>: only Delta reads them, and only when the bucket has no usable non-numeric
|
||||||
|
/// sample.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// The previous form referenced its CTE four times (counts + three self-joins); QuestDB opens a separate table
|
||||||
|
/// reader per reference, so a row committed between two of them produced a bucket whose counts included a sample
|
||||||
|
/// its first/last join could not find (a NULL first-of-any-quality time that threw) — ~1 in 90k reads of a tag
|
||||||
|
/// being written. One reference means one reader at one table version. A windowed (<c>row_number()</c>)
|
||||||
|
/// single-scan form was also correct but ~2× slower; this one is not slower than the old query.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
/// <param name="sample">Prefixed sample table name.</param>
|
/// <param name="sample">Prefixed sample table name.</param>
|
||||||
/// <returns>The SQL text.</returns>
|
/// <returns>The SQL text.</returns>
|
||||||
/// <param name="reverse">True for a reverse read (window <c>($4, $5]</c>).</param>
|
/// <param name="reverse">True for a reverse read (window <c>($4, $5]</c>).</param>
|
||||||
public static string StartEnd(string sample, bool reverse = false) =>
|
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), " +
|
// The usable picks: kind, num_value, int_value, str_value (not available — see remarks), ts.
|
||||||
"f AS (SELECT b, kind, num_value, int_value, str_value, ts, " +
|
static string Usable(string f, string g) =>
|
||||||
"row_number() OVER (PARTITION BY b ORDER BY ts ASC) AS rf, " +
|
$"cast({f}(kind, uts) AS SHORT), {f}(num_value, uts), {f}(int_value, uts), cast(NULL AS VARCHAR), {g}(uts)";
|
||||||
"row_number() OVER (PARTITION BY b ORDER BY ts DESC) AS rl FROM s WHERE u = 1), " +
|
// The any-quality picks: ts, kind, num_value, int_value, str_value, status — all from the group's first / last row.
|
||||||
"a AS (SELECT b, ts, kind, num_value, int_value, str_value, status, " +
|
static string Any(string f) =>
|
||||||
"row_number() OVER (PARTITION BY b ORDER BY ts ASC) AS rf, " +
|
$"{f}(ts), cast({f}(kind) AS SHORT), {f}(num_value), {f}(int_value), {f}(str_value), {f}(status)";
|
||||||
"row_number() OVER (PARTITION BY b ORDER BY ts DESC) AS rl FROM s) " +
|
return "SELECT b, " + CommonCounts() + ", " +
|
||||||
"SELECT st.*, " +
|
Usable("arg_min", "min") + ", " + Usable("arg_max", "max") + ", " + Any("first") + ", " + Any("last") + " " +
|
||||||
"x.kind, x.num_value, x.int_value, x.str_value, x.ts, " +
|
"FROM (SELECT b, ts, kind, num_value, int_value, str_value, status, u, g, CASE WHEN u = 1 THEN ts END AS uts " +
|
||||||
"y.kind, y.num_value, y.int_value, y.str_value, y.ts, " +
|
"FROM (" + Classified(sample, reverse) + ")) GROUP BY b ORDER BY b";
|
||||||
"p.ts, p.kind, p.num_value, p.int_value, p.str_value, p.status, " +
|
}
|
||||||
"q.ts, q.kind, q.num_value, q.int_value, q.str_value, q.status " +
|
|
||||||
"FROM st LEFT JOIN (SELECT * FROM f WHERE rf = 1) x ON x.b = st.b " +
|
|
||||||
"LEFT JOIN (SELECT * FROM f WHERE rl = 1) y ON y.b = st.b " +
|
|
||||||
"LEFT JOIN (SELECT * FROM a WHERE rf = 1) p ON p.b = st.b " +
|
|
||||||
"LEFT JOIN (SELECT * FROM a WHERE rl = 1) q ON q.b = st.b ORDER BY st.b";
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) <c>num_value</c>, ties
|
/// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) <c>num_value</c>, ties
|
||||||
/// broken by the earliest <c>ts</c>. Family columns: <c>6 num_value, 7 ts</c> (NULL when the bucket has no usable
|
/// broken by the earliest <c>ts</c>. Family columns: <c>6 num_value, 7 ts</c> (NULL when the bucket has no usable
|
||||||
/// numeric sample).
|
/// numeric sample).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Deliberately still two references to the classified rows (counts + the ranked extreme), unlike
|
||||||
|
/// <see cref="StartEnd"/>: the single-scan <c>row_number()</c> form measured ~2× slower, and QuestDB's
|
||||||
|
/// <c>arg_max</c>/<c>arg_min</c> cannot express the earliest-time tie-break. Live-gate 8 F1's inconsistency (a row
|
||||||
|
/// committed between the two readers — counted but not ranked) is absorbed by the reader and evaluator instead: a
|
||||||
|
/// NULL extreme in a bucket with usable samples evaluates to <c>BadNoData</c>, never a value-less Good.
|
||||||
|
/// </remarks>
|
||||||
/// <param name="sample">Prefixed sample table name.</param>
|
/// <param name="sample">Prefixed sample table name.</param>
|
||||||
/// <param name="maximum">True for MaximumActualTime.</param>
|
/// <param name="maximum">True for MaximumActualTime.</param>
|
||||||
/// <returns>The SQL text.</returns>
|
/// <returns>The SQL text.</returns>
|
||||||
|
|||||||
@@ -201,14 +201,14 @@ internal sealed class QuestDbConnection : IAsyncDisposable, IDisposable
|
|||||||
/// <param name="r">Reader positioned on a row.</param>
|
/// <param name="r">Reader positioned on a row.</param>
|
||||||
/// <param name="i">Column ordinal.</param>
|
/// <param name="i">Column ordinal.</param>
|
||||||
/// <returns>The timestamp with <c>Kind=Utc</c>.</returns>
|
/// <returns>The timestamp with <c>Kind=Utc</c>.</returns>
|
||||||
public static DateTime ReadUtc(NpgsqlDataReader r, int i) =>
|
public static DateTime ReadUtc(System.Data.Common.DbDataReader r, int i) =>
|
||||||
DateTime.SpecifyKind(r.GetDateTime(i), DateTimeKind.Utc);
|
DateTime.SpecifyKind(r.GetDateTime(i), DateTimeKind.Utc);
|
||||||
|
|
||||||
/// <summary>Reads a <c>BYTE</c> column (QuestDB returns <c>smallint</c>); NULL (never stored) reads as 0 (Empty).</summary>
|
/// <summary>Reads a <c>BYTE</c> column (QuestDB returns <c>smallint</c>); NULL (never stored) reads as 0 (Empty).</summary>
|
||||||
/// <param name="r">Reader positioned on a row.</param>
|
/// <param name="r">Reader positioned on a row.</param>
|
||||||
/// <param name="i">Column ordinal.</param>
|
/// <param name="i">Column ordinal.</param>
|
||||||
/// <returns>The byte value.</returns>
|
/// <returns>The byte value.</returns>
|
||||||
public static byte ReadKind(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? (byte)0 : (byte)r.GetInt16(i);
|
public static byte ReadKind(System.Data.Common.DbDataReader r, int i) => r.IsDBNull(i) ? (byte)0 : (byte)r.GetInt16(i);
|
||||||
|
|
||||||
// ---- error classification --------------------------------------------------------------------------------------
|
// ---- error classification --------------------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|||||||
+24
-7
@@ -1,3 +1,4 @@
|
|||||||
|
using System.Data.Common;
|
||||||
using System.Globalization;
|
using System.Globalization;
|
||||||
using System.Text;
|
using System.Text;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
@@ -574,12 +575,22 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
return await reader.ReadAsync(ct).ConfigureAwait(false) && !reader.IsDBNull(0) ? QuestDbConnection.ReadUtc(reader, 0) : null;
|
return await reader.ReadAsync(ct).ConfigureAwait(false) && !reader.IsDBNull(0) ? QuestDbConnection.ReadUtc(reader, 0) : null;
|
||||||
}
|
}
|
||||||
|
|
||||||
private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family)
|
/// <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(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
|
static double? D(DbDataReader 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);
|
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.
|
// 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) =>
|
static HistorianValue V(DbDataReader r, int i) =>
|
||||||
r.IsDBNull(i + 4)
|
r.IsDBNull(i + 4)
|
||||||
? HistorianValue.Empty
|
? HistorianValue.Empty
|
||||||
: ToValue(QuestDbConnection.ReadKind(r, i), D(r, i + 1), r.IsDBNull(i + 2) ? null : r.GetInt64(i + 2),
|
: ToValue(QuestDbConnection.ReadKind(r, i), D(r, i + 1), r.IsDBNull(i + 2) ? null : r.GetInt64(i + 2),
|
||||||
@@ -606,8 +617,8 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
case AggregateFamily.StartEnd:
|
case AggregateFamily.StartEnd:
|
||||||
row.First = V(reader, f);
|
row.First = V(reader, f);
|
||||||
row.Last = V(reader, f + 5);
|
row.Last = V(reader, f + 5);
|
||||||
row.FirstAny = ReadStored(reader, QuestDbAggregateQueries.FirstAnyColumn);
|
row.FirstAny = ReadStoredOrNull(reader, QuestDbAggregateQueries.FirstAnyColumn);
|
||||||
row.LastAny = ReadStored(reader, QuestDbAggregateQueries.FirstAnyColumn + 6);
|
row.LastAny = ReadStoredOrNull(reader, QuestDbAggregateQueries.FirstAnyColumn + 6);
|
||||||
break;
|
break;
|
||||||
default:
|
default:
|
||||||
row.Extreme = D(reader, f);
|
row.Extreme = D(reader, f);
|
||||||
@@ -623,7 +634,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
/// Reads <c>ts, kind, num_value, int_value, str_value, status</c> starting at <paramref name="o"/> and applies the
|
/// 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"/>).
|
/// shared quality clamp (<see cref="StoredSample.FromStored"/>).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private static StoredSample ReadStored(NpgsqlDataReader r, int o)
|
private static StoredSample ReadStored(DbDataReader r, int o)
|
||||||
{
|
{
|
||||||
var value = ToValue(
|
var value = ToValue(
|
||||||
QuestDbConnection.ReadKind(r, o + 1),
|
QuestDbConnection.ReadKind(r, o + 1),
|
||||||
@@ -635,6 +646,12 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
return StoredSample.FromStored(QuestDbConnection.ReadUtc(r, o), value, status);
|
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>
|
/// <summary>
|
||||||
/// Rebuilds a value from its stored columns via the shared <see cref="SampleColumnMapper.FromColumns"/>, after the
|
/// 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
|
/// RC2 repair: QuestDB reads a <c>LONG</c> holding <see cref="long.MinValue"/> back as NULL, so an Int64 row with
|
||||||
|
|||||||
@@ -25,6 +25,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
|||||||
/// sample is not usable.
|
/// sample is not usable.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
|
/// <b>Inconsistent rows</b> (live-gate 8 F1): a row whose counts say a bucket holds samples while a picked sample
|
||||||
|
/// (first/last of any quality, first/last usable, the extreme) is absent is answered from what is present —
|
||||||
|
/// Start/End/Delta/ActualTime without the sample they need are <c>BadNoData</c> — never an exception. The backends'
|
||||||
|
/// queries read counts and picks in one scan so this should not arise; the evaluator does not rely on it.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
/// Values: Count is <see cref="int"/>; Start/End keep the sample's type; every other aggregate is
|
/// Values: Count is <see cref="int"/>; Start/End keep the sample's type; every other aggregate is
|
||||||
/// <see cref="double"/> computed over the numeric value (Boolean = 0/1). These are the simple per-sample
|
/// <see cref="double"/> computed over the numeric value (Boolean = 0/1). These are the simple per-sample
|
||||||
/// definitions; the Part 13 time-weighted aggregates (TimeAverage, TimeAverage2, Total, Total2, Interpolative)
|
/// definitions; the Part 13 time-weighted aggregates (TimeAverage, TimeAverage2, Total, Total2, Interpolative)
|
||||||
@@ -76,12 +82,20 @@ public static class AggregateEvaluator
|
|||||||
case HistoryAggregateType.Delta:
|
case HistoryAggregateType.Delta:
|
||||||
// #680 (SDK ComputeDelta): last usable − first usable; Uncertain only when the interval's first or last
|
// #680 (SDK ComputeDelta): last usable − first usable; Uncertain only when the interval's first or last
|
||||||
// sample was skipped (not usable) — a Bad sample in the middle does not degrade it.
|
// sample was skipped (not usable) — a Bad sample in the middle does not degrade it.
|
||||||
|
// Live-gate 8 F1: a row whose counts promise a usable sample its picks do not carry (a backend that read
|
||||||
|
// the counts and the picks from two table versions) has no consistent first/last — no data, not a throw
|
||||||
|
// and not a wrong-input verdict.
|
||||||
|
if (r.First.Kind == HistorianValueKind.Empty || r.Last.Kind == HistorianValueKind.Empty)
|
||||||
|
return NoData(bucketStart, now);
|
||||||
if (!r.First.TryGetDouble(out var a) || !r.Last.TryGetDouble(out var b))
|
if (!r.First.TryGetDouble(out var a) || !r.Last.TryGetDouble(out var b))
|
||||||
return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
|
return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
|
||||||
var bounded = r.FirstAny is { IsUsable: true } && r.LastAny is { IsUsable: true };
|
var bounded = r.FirstAny is { IsUsable: true } && r.LastAny is { IsUsable: true };
|
||||||
return Calc(b - a, bounded ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal);
|
return Calc(b - a, bounded ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal);
|
||||||
case HistoryAggregateType.MinimumActualTime:
|
case HistoryAggregateType.MinimumActualTime:
|
||||||
case HistoryAggregateType.MaximumActualTime:
|
case HistoryAggregateType.MaximumActualTime:
|
||||||
|
// Live-gate 8 F1: usable numeric samples counted but no extreme picked (an inconsistent row) is no data,
|
||||||
|
// never a value-less Good.
|
||||||
|
if (r.Extreme is null || r.ExtremeTime is null) return NoData(bucketStart, now);
|
||||||
return Calc(r.Extreme, status, r.ExtremeTime);
|
return Calc(r.Extreme, status, r.ExtremeTime);
|
||||||
default:
|
default:
|
||||||
throw new HistoryAggregateNotSupportedException(aggregate);
|
throw new HistoryAggregateNotSupportedException(aggregate);
|
||||||
|
|||||||
@@ -71,33 +71,53 @@ internal static class SqlAggregateQueries
|
|||||||
/// Family columns: <c>6..10</c> first usable sample (<c>ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc</c>),
|
/// Family columns: <c>6..10</c> first usable sample (<c>ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc</c>),
|
||||||
/// <c>11..15</c> last usable sample (same order) — all NULL when the bucket has no usable sample; then
|
/// <c>11..15</c> last usable sample (same order) — all NULL when the bucket has no usable sample; then
|
||||||
/// <c>16..21</c> the first sample of any quality (<c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue,
|
/// <c>16..21</c> the first sample of any quality (<c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue,
|
||||||
/// StatusCode</c>) and <c>22..27</c> the last (same order) — never NULL for a returned (non-empty) bucket.
|
/// StatusCode</c>) and <c>22..27</c> the last (same order) — present for every returned (non-empty) bucket.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <b>One scan (live-gate 8 F1).</b> The counts and all four picked samples come from one reference to <c>s</c>:
|
||||||
|
/// a windowed pass ranks each row within its bucket (<c>rfa</c>/<c>rla</c> over every sample, <c>rfu</c>/<c>rlu</c>
|
||||||
|
/// within its usable / not-usable partition) and the outer <c>GROUP BY</c> both counts and lifts each picked row's
|
||||||
|
/// columns with a conditional <c>MAX</c> (one row per bucket matches each pick). The previous form referenced the
|
||||||
|
/// CTE five times; SQL Server expands each reference into its own scan, and under READ COMMITTED nothing makes those
|
||||||
|
/// scans see the same rows, so a sample committed between two of them could be counted but not picked (QuestDB hit
|
||||||
|
/// exactly that). It is also fewer scans of the key range.
|
||||||
|
/// </remarks>
|
||||||
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
||||||
/// <returns>The SQL text.</returns>
|
/// <returns>The SQL text.</returns>
|
||||||
/// <param name="reverse">True for a reverse read (window <c>(@start, @end]</c>).</param>
|
/// <param name="reverse">True for a reverse read (window <c>(@start, @end]</c>).</param>
|
||||||
public static string StartEnd(string sample, bool reverse = false) =>
|
public static string StartEnd(string sample, bool reverse = false)
|
||||||
Classified(sample, reverse) +
|
{
|
||||||
", st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b)" +
|
static string Pick(string when, bool any) =>
|
||||||
", f AS (SELECT b, ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc, " +
|
(any ? $"MAX(CASE WHEN {when} THEN SourceTimeUtc END), " : "") +
|
||||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rf, " +
|
$"MAX(CASE WHEN {when} THEN ValueKind END), MAX(CASE WHEN {when} THEN NumValue END), " +
|
||||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rl FROM s WHERE u = 1)" +
|
$"MAX(CASE WHEN {when} THEN IntValue END), MAX(CASE WHEN {when} THEN StrValue END), " +
|
||||||
", a AS (SELECT b, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, " +
|
(any ? $"MAX(CASE WHEN {when} THEN StatusCode END)" : $"MAX(CASE WHEN {when} THEN SourceTimeUtc END)");
|
||||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rf, " +
|
return Classified(sample, reverse) +
|
||||||
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rl FROM s) " +
|
", w AS (SELECT b, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, u, g, " +
|
||||||
"SELECT st.*, " +
|
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rfa, " +
|
||||||
"x.ValueKind, x.NumValue, x.IntValue, x.StrValue, x.SourceTimeUtc, " +
|
"ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rla, " +
|
||||||
"y.ValueKind, y.NumValue, y.IntValue, y.StrValue, y.SourceTimeUtc, " +
|
"ROW_NUMBER() OVER (PARTITION BY b, u ORDER BY SourceTimeUtc ASC) AS rfu, " +
|
||||||
"p.SourceTimeUtc, p.ValueKind, p.NumValue, p.IntValue, p.StrValue, p.StatusCode, " +
|
"ROW_NUMBER() OVER (PARTITION BY b, u ORDER BY SourceTimeUtc DESC) AS rlu FROM s) " +
|
||||||
"q.SourceTimeUtc, q.ValueKind, q.NumValue, q.IntValue, q.StrValue, q.StatusCode " +
|
"SELECT b, " + CommonCounts() + ", " +
|
||||||
"FROM st LEFT JOIN f x ON x.b = st.b AND x.rf = 1 LEFT JOIN f y ON y.b = st.b AND y.rl = 1 " +
|
Pick("u = 1 AND rfu = 1", any: false) + ", " +
|
||||||
"JOIN a p ON p.b = st.b AND p.rf = 1 JOIN a q ON q.b = st.b AND q.rl = 1 ORDER BY st.b;";
|
Pick("u = 1 AND rlu = 1", any: false) + ", " +
|
||||||
|
Pick("rfa = 1", any: true) + ", " +
|
||||||
|
Pick("rla = 1", any: true) + " " +
|
||||||
|
"FROM w GROUP BY b ORDER BY b;";
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) <c>NumValue</c>,
|
/// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) <c>NumValue</c>,
|
||||||
/// ties broken by the earliest <c>SourceTimeUtc</c>. Family columns: <c>6 NumValue, 7 SourceTimeUtc</c>
|
/// ties broken by the earliest <c>SourceTimeUtc</c>. Family columns: <c>6 NumValue, 7 SourceTimeUtc</c>
|
||||||
/// (NULL when the bucket has no usable numeric sample).
|
/// (NULL when the bucket has no usable numeric sample).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Deliberately still two references to <c>s</c> (counts + the ranked extreme), unlike <see cref="StartEnd"/>:
|
||||||
|
/// every single-scan form measured (a <c>ROW_NUMBER</c> or windowed <c>MIN</c>/<c>MAX</c> over every row) was
|
||||||
|
/// 1.2–2× slower on day-plus windows than ranking only the filtered usable rows. Live-gate 8 F1's inconsistency —
|
||||||
|
/// a sample committed between the two scans, counted but not ranked — is instead absorbed by the reader: a
|
||||||
|
/// NULL extreme in a bucket with usable samples evaluates to <c>BadNoData</c> (never a value-less Good).
|
||||||
|
/// </remarks>
|
||||||
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
||||||
/// <param name="maximum">True for MaximumActualTime.</param>
|
/// <param name="maximum">True for MaximumActualTime.</param>
|
||||||
/// <returns>The SQL text.</returns>
|
/// <returns>The SQL text.</returns>
|
||||||
|
|||||||
@@ -106,7 +106,7 @@ internal sealed class SqlHistorianConnection
|
|||||||
/// <param name="reader">Reader positioned on a row.</param>
|
/// <param name="reader">Reader positioned on a row.</param>
|
||||||
/// <param name="ordinal">Column ordinal.</param>
|
/// <param name="ordinal">Column ordinal.</param>
|
||||||
/// <returns>UTC timestamp.</returns>
|
/// <returns>UTC timestamp.</returns>
|
||||||
public static DateTime GetUtc(SqlDataReader reader, int ordinal) =>
|
public static DateTime GetUtc(System.Data.Common.DbDataReader reader, int ordinal) =>
|
||||||
DateTime.SpecifyKind(reader.GetDateTime(ordinal), DateTimeKind.Utc);
|
DateTime.SpecifyKind(reader.GetDateTime(ordinal), DateTimeKind.Utc);
|
||||||
|
|
||||||
/// <summary>True for connection/timeout/deadlock/throttling errors worth retrying.</summary>
|
/// <summary>True for connection/timeout/deadlock/throttling errors worth retrying.</summary>
|
||||||
|
|||||||
+25
-8
@@ -1,4 +1,5 @@
|
|||||||
using System.Data;
|
using System.Data;
|
||||||
|
using System.Data.Common;
|
||||||
using System.Text;
|
using System.Text;
|
||||||
using Microsoft.Data.SqlClient;
|
using Microsoft.Data.SqlClient;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
@@ -524,12 +525,22 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
} while (await reader.NextResultAsync(ct).ConfigureAwait(false));
|
} while (await reader.NextResultAsync(ct).ConfigureAwait(false));
|
||||||
}
|
}
|
||||||
|
|
||||||
private static BucketRow ReadBucket(SqlDataReader reader, AggregateFamily family)
|
/// <summary>
|
||||||
|
/// Maps one per-sample aggregate row (<see cref="SqlAggregateQueries"/> column layout) to a <see cref="BucketRow"/>.
|
||||||
|
/// Every picked sample — first/last usable, first/last of any quality, the extreme — is optional: a NULL reads as
|
||||||
|
/// "no such sample", never an exception (live-gate 8 F1: on QuestDB a row whose counts and picks disagreed threw
|
||||||
|
/// out of <see cref="ReadStored"/> and failed 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(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
|
static double? D(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i);
|
||||||
static long? L(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetInt64(i);
|
static long? L(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetInt64(i);
|
||||||
static string? S(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetString(i);
|
static string? S(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetString(i);
|
||||||
static HistorianValue V(SqlDataReader r, int i) =>
|
static HistorianValue V(DbDataReader r, int i) =>
|
||||||
r.IsDBNull(i) ? HistorianValue.Empty : SampleColumnMapper.FromColumns(r.GetByte(i), D(r, i + 1), L(r, i + 2), S(r, i + 3));
|
r.IsDBNull(i) ? HistorianValue.Empty : SampleColumnMapper.FromColumns(r.GetByte(i), D(r, i + 1), L(r, i + 2), S(r, i + 3));
|
||||||
|
|
||||||
var row = new BucketRow
|
var row = new BucketRow
|
||||||
@@ -553,8 +564,8 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
case AggregateFamily.StartEnd:
|
case AggregateFamily.StartEnd:
|
||||||
row.First = V(reader, f);
|
row.First = V(reader, f);
|
||||||
row.Last = V(reader, f + 5);
|
row.Last = V(reader, f + 5);
|
||||||
row.FirstAny = ReadStored(reader, SqlAggregateQueries.FirstAnyColumn);
|
row.FirstAny = ReadStoredOrNull(reader, SqlAggregateQueries.FirstAnyColumn);
|
||||||
row.LastAny = ReadStored(reader, SqlAggregateQueries.FirstAnyColumn + 6);
|
row.LastAny = ReadStoredOrNull(reader, SqlAggregateQueries.FirstAnyColumn + 6);
|
||||||
break;
|
break;
|
||||||
default:
|
default:
|
||||||
row.Extreme = D(reader, f);
|
row.Extreme = D(reader, f);
|
||||||
@@ -572,7 +583,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
/// (<see cref="StoredSample.FromStored"/>) — defence in depth for rows written before the writer forced every
|
/// (<see cref="StoredSample.FromStored"/>) — defence in depth for rows written before the writer forced every
|
||||||
/// Empty sample Bad.
|
/// Empty sample Bad.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private static StoredSample ReadStored(SqlDataReader r, int o)
|
private static StoredSample ReadStored(DbDataReader r, int o)
|
||||||
{
|
{
|
||||||
var value = SampleColumnMapper.FromColumns(
|
var value = SampleColumnMapper.FromColumns(
|
||||||
r.GetByte(o + 1),
|
r.GetByte(o + 1),
|
||||||
@@ -583,6 +594,12 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
return StoredSample.FromStored(SqlHistorianConnection.GetUtc(r, o), value, status);
|
return StoredSample.FromStored(SqlHistorianConnection.GetUtc(r, o), value, status);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// <see cref="ReadStored"/> for an optional picked sample: a NULL <c>SourceTimeUtc</c> means the row carries no such
|
||||||
|
/// sample (live-gate 8 F1), so the result is null rather than an exception.
|
||||||
|
/// </summary>
|
||||||
|
private static StoredSample? ReadStoredOrNull(DbDataReader r, int o) => r.IsDBNull(o) ? null : ReadStored(r, o);
|
||||||
|
|
||||||
/// <summary>Read-only tag lookup (never creates a row); runs the schema bootstrap on a cache miss.</summary>
|
/// <summary>Read-only tag lookup (never creates a row); runs the schema bootstrap on a cache miss.</summary>
|
||||||
private Task<int?> ResolveAsync(string name, CancellationToken ct) => _catalog.TryGetTagIdAsync(name, ct);
|
private Task<int?> ResolveAsync(string name, CancellationToken ct) => _catalog.TryGetTagIdAsync(name, ct);
|
||||||
|
|
||||||
|
|||||||
+98
@@ -0,0 +1,98 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using Shouldly;
|
||||||
|
using Xunit;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Live-gate 8 F1 (#680 follow-up): per-sample processed reads of a tag that is being written. The StartEnd and
|
||||||
|
/// ActualTime queries reference their classified rows more than once; QuestDB evaluates each reference with its own
|
||||||
|
/// table reader, so a row committed between two of them produced a bucket whose counts included a sample its
|
||||||
|
/// first/last (or extreme) join could not find — a NULL first-of-any-quality time threw <c>InvalidCastException</c>
|
||||||
|
/// and failed the node's whole read (~1 in 90k reads on the rig), and ActualTime returned a value-less Good. StartEnd
|
||||||
|
/// now reads the table once; ActualTime keeps its form and an inconsistent bucket evaluates to <c>BadNoData</c>. This
|
||||||
|
/// test interleaves a continuous single-row writer with tight-loop reads and asserts that no read throws and no
|
||||||
|
/// non-Bad result is value-less.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Opt-in: runs only when <c>HISTORIAN_QUESTDB_STRESS_SECONDS</c> is set (and the fixture's connection is), so it never
|
||||||
|
/// slows the default suite. Category <c>SlowUnit</c> as well, for filter-based exclusion.
|
||||||
|
/// </remarks>
|
||||||
|
[Collection(QuestDbCollection.Name)]
|
||||||
|
[Trait("Category", "SlowUnit")]
|
||||||
|
public sealed class LiveWriteReadRaceStressTests(QuestDbFixture fx)
|
||||||
|
{
|
||||||
|
/// <summary>Seconds to run the race loop for; absent ⇒ the test skips.</summary>
|
||||||
|
public const string SecondsEnvVar = "HISTORIAN_QUESTDB_STRESS_SECONDS";
|
||||||
|
|
||||||
|
private static readonly HistoryAggregateType[] Aggregates =
|
||||||
|
[
|
||||||
|
HistoryAggregateType.Start, HistoryAggregateType.End, HistoryAggregateType.Delta,
|
||||||
|
HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime,
|
||||||
|
];
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Per_sample_reads_of_a_tag_being_written_never_fault_and_never_return_a_valueless_non_bad_result()
|
||||||
|
{
|
||||||
|
if (!int.TryParse(Environment.GetEnvironmentVariable(SecondsEnvVar), out var seconds) || seconds <= 0)
|
||||||
|
Assert.Skip($"{SecondsEnvVar} is not set; the live write/read race stress test is opt-in.");
|
||||||
|
await using var b = await QuestDbConformanceBackend.CreateConcreteAsync(fx);
|
||||||
|
const string tag = "Race";
|
||||||
|
await b.NewCatalog().EnsureAsync(tag, 0, CancellationToken.None);
|
||||||
|
var src = b.CreateDataSource();
|
||||||
|
|
||||||
|
using var stop = new CancellationTokenSource(TimeSpan.FromSeconds(seconds));
|
||||||
|
long written = 0, reads = 0;
|
||||||
|
var faults = new ConcurrentBag<string>();
|
||||||
|
|
||||||
|
var writer = Task.Run(async () =>
|
||||||
|
{
|
||||||
|
var i = 0;
|
||||||
|
while (!stop.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
// One committed row per statement (one WAL transaction each), every 7th Bad, never waiting for visibility.
|
||||||
|
var status = ++i % 7 == 0 ? HistorianStatusCodes.Bad : 0u;
|
||||||
|
await b.InsertRowsNoWaitAsync(tag, [(DateTime.UtcNow, HistorianValue.FromDouble(i), status, false)]);
|
||||||
|
Interlocked.Increment(ref written);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
var readers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
|
||||||
|
{
|
||||||
|
while (!stop.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
var now = DateTime.UtcNow;
|
||||||
|
var start = new DateTime(now.Ticks / TimeSpan.TicksPerSecond / 10 * 10 * TimeSpan.TicksPerSecond, DateTimeKind.Utc)
|
||||||
|
.AddSeconds(-30);
|
||||||
|
var end = start.AddSeconds(70);
|
||||||
|
foreach (var agg in Aggregates)
|
||||||
|
foreach (var reverse in new[] { false, true })
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var r = await src.ReadProcessedAsync(tag, reverse ? end : start, reverse ? start : end,
|
||||||
|
TimeSpan.FromSeconds(10), agg, CancellationToken.None);
|
||||||
|
foreach (var v in r.Samples)
|
||||||
|
if (!StatusBits.IsBad(v.StatusCode) && v.Value is null)
|
||||||
|
faults.Add($"{agg} {(reverse ? "rev" : "fwd")}: non-Bad 0x{v.StatusCode:X8} with no value at {v.SourceTimestampUtc:O}");
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
faults.Add($"{agg} {(reverse ? "rev" : "fwd")}: {ex.GetType().Name}: {ex.Message}");
|
||||||
|
}
|
||||||
|
Interlocked.Increment(ref reads);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})).ToArray();
|
||||||
|
|
||||||
|
await Task.WhenAll(readers.Append(writer));
|
||||||
|
var summary = $"race stress: {seconds}s, {Interlocked.Read(ref written)} rows written, {Interlocked.Read(ref reads)} reads, {faults.Count} faults";
|
||||||
|
var output = TestContext.Current.TestOutputHelper;
|
||||||
|
output?.WriteLine(summary);
|
||||||
|
foreach (var f in faults.Take(20)) output?.WriteLine("fault: " + f);
|
||||||
|
written.ShouldBeGreaterThan(0);
|
||||||
|
faults.ShouldBeEmpty(summary);
|
||||||
|
}
|
||||||
|
}
|
||||||
+13
-1
@@ -107,6 +107,19 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
|
|||||||
/// <param name="tag">The tag.</param>
|
/// <param name="tag">The tag.</param>
|
||||||
/// <param name="rows">Rows: time (already rounded by the caller or not — it is bound through BindTimestamp), value, status, deleted.</param>
|
/// <param name="rows">Rows: time (already rounded by the caller or not — it is bound through BindTimestamp), value, status, deleted.</param>
|
||||||
public async Task InsertRowsAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows)
|
public async Task InsertRowsAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows)
|
||||||
|
{
|
||||||
|
if (rows.Length == 0) return;
|
||||||
|
await InsertRowsNoWaitAsync(tag, rows);
|
||||||
|
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// <see cref="InsertRowsAsync"/> without the WAL visibility wait: the rows become visible whenever QuestDB applies
|
||||||
|
/// them (the live write/read race stress test commits one row per call and never waits).
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="tag">The tag.</param>
|
||||||
|
/// <param name="rows">Rows: time, value, status, deleted.</param>
|
||||||
|
public async Task InsertRowsNoWaitAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows)
|
||||||
{
|
{
|
||||||
if (rows.Length == 0) return;
|
if (rows.Length == 0) return;
|
||||||
await using (var c = await Conn.OpenAsync(CancellationToken.None))
|
await using (var c = await Conn.OpenAsync(CancellationToken.None))
|
||||||
@@ -140,7 +153,6 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
|
|||||||
await cmd.ExecuteNonQueryAsync();
|
await cmd.ExecuteNonQueryAsync();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
|
|||||||
+116
@@ -0,0 +1,116 @@
|
|||||||
|
using System.Data;
|
||||||
|
using Shouldly;
|
||||||
|
using Xunit;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Live-gate 8 F1: the per-sample aggregate row mapper (<see cref="QuestDbHistorianDataSource.ReadBucket"/>) over a
|
||||||
|
/// hand-built row in the <see cref="QuestDbAggregateQueries.StartEnd"/> column layout. A row whose counts include
|
||||||
|
/// samples but whose first/last-of-any-quality picks are NULL used to throw <c>InvalidCastException: Column 'ts' is
|
||||||
|
/// null</c> and fail the node's whole read; it now maps to absent picks the evaluator answers as <c>BadNoData</c>.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class BucketRowMappingTests
|
||||||
|
{
|
||||||
|
private const uint NoDataCalc = 0x809B0401u;
|
||||||
|
private static readonly DateTime T = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc);
|
||||||
|
|
||||||
|
/// <summary>The StartEnd layout: b + 5 counts, two usable picks (kind, num, int, str, ts), two any picks (ts, kind, num, int, str, status).</summary>
|
||||||
|
private static DataTable StartEndTable()
|
||||||
|
{
|
||||||
|
var t = new DataTable();
|
||||||
|
t.Columns.Add("b", typeof(long));
|
||||||
|
foreach (var c in new[] { "total", "usable", "good", "bad", "nonnum" }) t.Columns.Add(c, typeof(long));
|
||||||
|
for (var p = 0; p < 2; p++)
|
||||||
|
{
|
||||||
|
t.Columns.Add($"k{p}", typeof(short));
|
||||||
|
t.Columns.Add($"n{p}", typeof(double));
|
||||||
|
t.Columns.Add($"i{p}", typeof(long));
|
||||||
|
t.Columns.Add($"s{p}", typeof(string));
|
||||||
|
t.Columns.Add($"t{p}", typeof(DateTime));
|
||||||
|
}
|
||||||
|
for (var p = 0; p < 2; p++)
|
||||||
|
{
|
||||||
|
t.Columns.Add($"at{p}", typeof(DateTime));
|
||||||
|
t.Columns.Add($"ak{p}", typeof(short));
|
||||||
|
t.Columns.Add($"an{p}", typeof(double));
|
||||||
|
t.Columns.Add($"ai{p}", typeof(long));
|
||||||
|
t.Columns.Add($"as{p}", typeof(string));
|
||||||
|
t.Columns.Add($"st{p}", typeof(long));
|
||||||
|
}
|
||||||
|
return t;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static DataRow Counts(DataTable t, long bucket, long total)
|
||||||
|
{
|
||||||
|
var row = t.NewRow();
|
||||||
|
row[0] = bucket;
|
||||||
|
row[1] = total; row[2] = total; row[3] = total; row[4] = 0L; row[5] = 0L;
|
||||||
|
return row;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static BucketRow Map(DataTable t, AggregateFamily family)
|
||||||
|
{
|
||||||
|
using var r = t.CreateDataReader();
|
||||||
|
r.Read().ShouldBeTrue();
|
||||||
|
return QuestDbHistorianDataSource.ReadBucket(r, family);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_row_with_counts_but_null_first_and_last_of_any_quality_maps_to_absent_picks_without_throwing()
|
||||||
|
{
|
||||||
|
var t = StartEndTable();
|
||||||
|
var row = Counts(t, 3, 2);
|
||||||
|
// A non-matching BYTE pick reads back as kind 0 (BYTE has no NULL) with every other column NULL.
|
||||||
|
row["k0"] = (short)0;
|
||||||
|
row["k1"] = (short)0;
|
||||||
|
t.Rows.Add(row);
|
||||||
|
|
||||||
|
var b = Map(t, AggregateFamily.StartEnd);
|
||||||
|
b.Bucket.ShouldBe(3);
|
||||||
|
b.Total.ShouldBe(2);
|
||||||
|
b.FirstAny.ShouldBeNull();
|
||||||
|
b.LastAny.ShouldBeNull();
|
||||||
|
b.First.Kind.ShouldBe(HistorianValueKind.Empty);
|
||||||
|
b.Last.Kind.ShouldBe(HistorianValueKind.Empty);
|
||||||
|
|
||||||
|
foreach (var agg in new[] { HistoryAggregateType.Start, HistoryAggregateType.End, HistoryAggregateType.Delta })
|
||||||
|
AggregateEvaluator.Evaluate(agg, b, T, T).StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_consistent_row_still_maps_every_pick()
|
||||||
|
{
|
||||||
|
var t = StartEndTable();
|
||||||
|
var row = Counts(t, 0, 2);
|
||||||
|
row["k0"] = (short)1; row["n0"] = 10d; row["t0"] = T;
|
||||||
|
row["k1"] = (short)1; row["n1"] = 20d; row["t1"] = T.AddSeconds(1);
|
||||||
|
row["at0"] = T; row["ak0"] = (short)1; row["an0"] = 10d; row["st0"] = 0L;
|
||||||
|
row["at1"] = T.AddSeconds(1); row["ak1"] = (short)1; row["an1"] = 20d; row["st1"] = 0L;
|
||||||
|
t.Rows.Add(row);
|
||||||
|
|
||||||
|
var b = Map(t, AggregateFamily.StartEnd);
|
||||||
|
b.First.ShouldBe(HistorianValue.FromDouble(10));
|
||||||
|
b.Last.ShouldBe(HistorianValue.FromDouble(20));
|
||||||
|
b.FirstAny.ShouldNotBeNull().TimeUtc.ShouldBe(T);
|
||||||
|
b.LastAny.ShouldNotBeNull().Value.ShouldBe(HistorianValue.FromDouble(20));
|
||||||
|
AggregateEvaluator.Evaluate(HistoryAggregateType.Delta, b, T, T).Value.ShouldBe(10d);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void An_actual_time_row_with_counts_but_no_extreme_is_no_data()
|
||||||
|
{
|
||||||
|
var t = new DataTable();
|
||||||
|
t.Columns.Add("b", typeof(long));
|
||||||
|
foreach (var c in new[] { "total", "usable", "good", "bad", "nonnum" }) t.Columns.Add(c, typeof(long));
|
||||||
|
t.Columns.Add("num_value", typeof(double));
|
||||||
|
t.Columns.Add("ts", typeof(DateTime));
|
||||||
|
t.Rows.Add(0L, 1L, 1L, 1L, 0L, 0L, DBNull.Value, DBNull.Value);
|
||||||
|
|
||||||
|
var b = Map(t, AggregateFamily.ActualTime);
|
||||||
|
b.Extreme.ShouldBeNull();
|
||||||
|
AggregateEvaluator.Evaluate(HistoryAggregateType.MaximumActualTime, b, T, T).StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
}
|
||||||
|
}
|
||||||
+48
@@ -257,4 +257,52 @@ public sealed class AggregateEvaluatorTests
|
|||||||
avg.Value.ShouldBe(15d);
|
avg.Value.ShouldBe(15d);
|
||||||
avg.StatusCode.ShouldBe(UncertainCalc);
|
avg.StatusCode.ShouldBe(UncertainCalc);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- Live-gate 8 F1: a row whose counts and picks disagree (a backend that read them from two table versions).
|
||||||
|
|
||||||
|
/// <summary>Counts say B0 holds two Good samples, but none of the picked samples came back.</summary>
|
||||||
|
private static BucketRow InconsistentRow() => new()
|
||||||
|
{
|
||||||
|
Bucket = 0, Total = 2, Usable = 2, Good = 2, Bad = 0, NonNumeric = 0, Avg = 15, Min = 10, Max = 20, Sum = 30,
|
||||||
|
First = HistorianValue.Empty, Last = HistorianValue.Empty, Extreme = null, ExtremeTime = null,
|
||||||
|
FirstAny = null, LastAny = null,
|
||||||
|
};
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData(HistoryAggregateType.Start)]
|
||||||
|
[InlineData(HistoryAggregateType.End)]
|
||||||
|
[InlineData(HistoryAggregateType.Delta)]
|
||||||
|
[InlineData(HistoryAggregateType.MinimumActualTime)]
|
||||||
|
[InlineData(HistoryAggregateType.MaximumActualTime)]
|
||||||
|
public void A_bucket_whose_picked_samples_are_missing_is_no_data_not_a_throw_or_a_valueless_good(HistoryAggregateType agg)
|
||||||
|
{
|
||||||
|
var v = AggregateEvaluator.Evaluate(agg, InconsistentRow(), At(0), Now);
|
||||||
|
v.Value.ShouldBeNull();
|
||||||
|
v.StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_delta_whose_first_usable_pick_alone_is_missing_is_no_data_not_invalid_inputs()
|
||||||
|
{
|
||||||
|
var row = InconsistentRow();
|
||||||
|
row.Last = HistorianValue.FromDouble(20);
|
||||||
|
row.FirstAny = A[0];
|
||||||
|
row.LastAny = A[1];
|
||||||
|
AggregateEvaluator.Evaluate(HistoryAggregateType.Delta, row, At(0), Now).StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void An_inconsistent_bucket_does_not_fail_the_whole_read_and_its_neighbours_keep_their_values()
|
||||||
|
{
|
||||||
|
var rows = RowsA();
|
||||||
|
rows[0] = InconsistentRow();
|
||||||
|
var plan = BucketPlan.Create(At(0), At(80), TimeSpan.FromSeconds(20));
|
||||||
|
foreach (var agg in new[] { HistoryAggregateType.Start, HistoryAggregateType.Delta, HistoryAggregateType.MaximumActualTime })
|
||||||
|
{
|
||||||
|
var s = AggregateEvaluator.EvaluateAll(agg, plan, rows.Values, At(0), At(40), Now);
|
||||||
|
s.Count.ShouldBe(4);
|
||||||
|
s[0].StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
s[2].Value.ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+112
@@ -0,0 +1,112 @@
|
|||||||
|
using System.Data;
|
||||||
|
using Shouldly;
|
||||||
|
using Xunit;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Live-gate 8 F1 (defence in depth on SQL Server, where it was not observed): the per-sample aggregate row mapper
|
||||||
|
/// (<see cref="SqlServerHistorianDataSource.ReadBucket"/>) over a hand-built row in the
|
||||||
|
/// <see cref="SqlAggregateQueries.StartEnd"/> column layout. A row whose counts include samples but whose
|
||||||
|
/// first/last-of-any-quality picks are NULL used to throw out of <c>ReadStored</c> and fail the node's whole read; it now
|
||||||
|
/// maps to absent picks the evaluator answers as <c>BadNoData</c>.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class BucketRowMappingTests
|
||||||
|
{
|
||||||
|
private const uint NoDataCalc = 0x809B0401u;
|
||||||
|
private static readonly DateTime T = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc);
|
||||||
|
|
||||||
|
/// <summary>b + 5 counts, two usable picks (kind, num, int, str, ts), two any picks (ts, kind, num, int, str, status).</summary>
|
||||||
|
private static DataTable StartEndTable()
|
||||||
|
{
|
||||||
|
var t = new DataTable();
|
||||||
|
t.Columns.Add("b", typeof(long));
|
||||||
|
foreach (var c in new[] { "total", "usable", "good", "bad", "nonnum" }) t.Columns.Add(c, typeof(int));
|
||||||
|
for (var p = 0; p < 2; p++)
|
||||||
|
{
|
||||||
|
t.Columns.Add($"k{p}", typeof(byte));
|
||||||
|
t.Columns.Add($"n{p}", typeof(double));
|
||||||
|
t.Columns.Add($"i{p}", typeof(long));
|
||||||
|
t.Columns.Add($"s{p}", typeof(string));
|
||||||
|
t.Columns.Add($"t{p}", typeof(DateTime));
|
||||||
|
}
|
||||||
|
for (var p = 0; p < 2; p++)
|
||||||
|
{
|
||||||
|
t.Columns.Add($"at{p}", typeof(DateTime));
|
||||||
|
t.Columns.Add($"ak{p}", typeof(byte));
|
||||||
|
t.Columns.Add($"an{p}", typeof(double));
|
||||||
|
t.Columns.Add($"ai{p}", typeof(long));
|
||||||
|
t.Columns.Add($"as{p}", typeof(string));
|
||||||
|
t.Columns.Add($"st{p}", typeof(int));
|
||||||
|
}
|
||||||
|
return t;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static DataRow Counts(DataTable t, long bucket, int total)
|
||||||
|
{
|
||||||
|
var row = t.NewRow();
|
||||||
|
row[0] = bucket;
|
||||||
|
row[1] = total; row[2] = total; row[3] = total; row[4] = 0; row[5] = 0;
|
||||||
|
return row;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static BucketRow Map(DataTable t, AggregateFamily family)
|
||||||
|
{
|
||||||
|
using var r = t.CreateDataReader();
|
||||||
|
r.Read().ShouldBeTrue();
|
||||||
|
return SqlServerHistorianDataSource.ReadBucket(r, family);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_row_with_counts_but_null_first_and_last_of_any_quality_maps_to_absent_picks_without_throwing()
|
||||||
|
{
|
||||||
|
var t = StartEndTable();
|
||||||
|
t.Rows.Add(Counts(t, 3, 2));
|
||||||
|
|
||||||
|
var b = Map(t, AggregateFamily.StartEnd);
|
||||||
|
b.Bucket.ShouldBe(3);
|
||||||
|
b.Total.ShouldBe(2);
|
||||||
|
b.FirstAny.ShouldBeNull();
|
||||||
|
b.LastAny.ShouldBeNull();
|
||||||
|
b.First.Kind.ShouldBe(HistorianValueKind.Empty);
|
||||||
|
|
||||||
|
foreach (var agg in new[] { HistoryAggregateType.Start, HistoryAggregateType.End, HistoryAggregateType.Delta })
|
||||||
|
AggregateEvaluator.Evaluate(agg, b, T, T).StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void A_consistent_row_still_maps_every_pick()
|
||||||
|
{
|
||||||
|
var t = StartEndTable();
|
||||||
|
var row = Counts(t, 0, 2);
|
||||||
|
row["k0"] = (byte)1; row["n0"] = 10d; row["t0"] = T;
|
||||||
|
row["k1"] = (byte)1; row["n1"] = 20d; row["t1"] = T.AddSeconds(1);
|
||||||
|
row["at0"] = T; row["ak0"] = (byte)1; row["an0"] = 10d; row["st0"] = 0;
|
||||||
|
row["at1"] = T.AddSeconds(1); row["ak1"] = (byte)1; row["an1"] = 20d; row["st1"] = 0;
|
||||||
|
t.Rows.Add(row);
|
||||||
|
|
||||||
|
var b = Map(t, AggregateFamily.StartEnd);
|
||||||
|
b.First.ShouldBe(HistorianValue.FromDouble(10));
|
||||||
|
b.Last.ShouldBe(HistorianValue.FromDouble(20));
|
||||||
|
b.FirstAny.ShouldNotBeNull().TimeUtc.ShouldBe(T);
|
||||||
|
b.LastAny.ShouldNotBeNull().Value.ShouldBe(HistorianValue.FromDouble(20));
|
||||||
|
AggregateEvaluator.Evaluate(HistoryAggregateType.Delta, b, T, T).Value.ShouldBe(10d);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void An_actual_time_row_with_counts_but_no_extreme_is_no_data()
|
||||||
|
{
|
||||||
|
var t = new DataTable();
|
||||||
|
t.Columns.Add("b", typeof(long));
|
||||||
|
foreach (var c in new[] { "total", "usable", "good", "bad", "nonnum" }) t.Columns.Add(c, typeof(int));
|
||||||
|
t.Columns.Add("NumValue", typeof(double));
|
||||||
|
t.Columns.Add("SourceTimeUtc", typeof(DateTime));
|
||||||
|
t.Rows.Add(0L, 1, 1, 1, 0, 0, DBNull.Value, DBNull.Value);
|
||||||
|
|
||||||
|
var b = Map(t, AggregateFamily.ActualTime);
|
||||||
|
b.Extreme.ShouldBeNull();
|
||||||
|
AggregateEvaluator.Evaluate(HistoryAggregateType.MaximumActualTime, b, T, T).StatusCode.ShouldBe(NoDataCalc);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user