From 2ceed561ad99c97a71a8ccfaea78845d912823b8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 18:28:29 -0400 Subject: [PATCH] =?UTF-8?q?fix(historian):=20live-gate=208=20F1=20?= =?UTF-8?q?=E2=80=94=20per-sample=20counts=20and=20picks=20from=20one=20ta?= =?UTF-8?q?ble=20version=20(#680=20follow-up)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- docs/Historian.md | 20 +++ .../QuestDbAggregateQueries.cs | 57 ++++++--- .../QuestDbConnection.cs | 4 +- .../QuestDbHistorianDataSource.cs | 31 +++-- .../AggregateEvaluator.cs | 14 +++ .../SqlAggregateQueries.cs | 54 +++++--- .../SqlHistorianConnection.cs | 2 +- .../SqlServerHistorianDataSource.cs | 33 +++-- .../LiveWriteReadRaceStressTests.cs | 98 +++++++++++++++ .../QuestDbConformanceBackend.cs | 14 ++- .../BucketRowMappingTests.cs | 116 ++++++++++++++++++ .../AggregateEvaluatorTests.cs | 48 ++++++++ .../BucketRowMappingTests.cs | 112 +++++++++++++++++ 13 files changed, 549 insertions(+), 54 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/LiveWriteReadRaceStressTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/BucketRowMappingTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs diff --git a/docs/Historian.md b/docs/Historian.md index c8b00913..faa1b876 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -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 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`. +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 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 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. + ⚠️ **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) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs index f4ca88b7..e6380aab 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs @@ -83,33 +83,54 @@ internal static class QuestDbAggregateQueries /// (ts, kind, num_value, int_value, str_value, status) and 22..27 the last (same order), present for /// every returned (non-empty) bucket. /// + /// + /// + /// One scan, one GROUP BY (live-gate 8 F1). 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 first()/last() (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 usable sample are arg_min/arg_max keyed on uts, the + /// row's time when usable and NULL otherwise (a NULL key is skipped, a NULL value is returned as is, and + /// (ts, tag) is unique — DEDUP — so there are no ties). arg_min has no VARCHAR form, so the usable + /// picks carry no str_value: only Delta reads them, and only when the bucket has no usable non-numeric + /// sample. + /// + /// + /// 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 (row_number()) + /// single-scan form was also correct but ~2× slower; this one is not slower than the old query. + /// + /// /// Prefixed sample table name. /// The SQL text. /// True for a reverse read (window ($4, $5]). - 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, " + - "row_number() OVER (PARTITION BY b ORDER BY ts DESC) AS rl FROM s WHERE u = 1), " + - "a AS (SELECT b, ts, kind, num_value, int_value, str_value, status, " + - "row_number() OVER (PARTITION BY b ORDER BY ts ASC) AS rf, " + - "row_number() OVER (PARTITION BY b ORDER BY ts DESC) AS rl FROM s) " + - "SELECT st.*, " + - "x.kind, x.num_value, x.int_value, x.str_value, x.ts, " + - "y.kind, y.num_value, y.int_value, y.str_value, y.ts, " + - "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"; + public static string StartEnd(string sample, bool reverse = false) + { + // The usable picks: kind, num_value, int_value, str_value (not available — see remarks), ts. + static string Usable(string f, string g) => + $"cast({f}(kind, uts) AS SHORT), {f}(num_value, uts), {f}(int_value, uts), cast(NULL AS VARCHAR), {g}(uts)"; + // The any-quality picks: ts, kind, num_value, int_value, str_value, status — all from the group's first / last row. + static string Any(string f) => + $"{f}(ts), cast({f}(kind) AS SHORT), {f}(num_value), {f}(int_value), {f}(str_value), {f}(status)"; + return "SELECT b, " + CommonCounts() + ", " + + Usable("arg_min", "min") + ", " + Usable("arg_max", "max") + ", " + Any("first") + ", " + Any("last") + " " + + "FROM (SELECT b, ts, kind, num_value, int_value, str_value, status, u, g, CASE WHEN u = 1 THEN ts END AS uts " + + "FROM (" + Classified(sample, reverse) + ")) GROUP BY b ORDER BY b"; + } /// /// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) num_value, ties /// broken by the earliest ts. Family columns: 6 num_value, 7 ts (NULL when the bucket has no usable /// numeric sample). /// + /// + /// Deliberately still two references to the classified rows (counts + the ranked extreme), unlike + /// : the single-scan row_number() form measured ~2× slower, and QuestDB's + /// arg_max/arg_min 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 BadNoData, never a value-less Good. + /// /// Prefixed sample table name. /// True for MaximumActualTime. /// The SQL text. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs index d3c7db2d..962886ce 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs @@ -201,14 +201,14 @@ internal sealed class QuestDbConnection : IAsyncDisposable, IDisposable /// Reader positioned on a row. /// Column ordinal. /// The timestamp with Kind=Utc. - 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); /// Reads a BYTE column (QuestDB returns smallint); NULL (never stored) reads as 0 (Empty). /// Reader positioned on a row. /// Column ordinal. /// The byte value. - 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 -------------------------------------------------------------------------------------- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index 638c1611..c09e6361 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -1,3 +1,4 @@ +using System.Data.Common; using System.Globalization; using System.Text; 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; } - private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family) + /// + /// Maps one per-sample aggregate row ( column layout) to a + /// . 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 and fail the node's whole read; the evaluator + /// answers such a bucket from what is present, or BadNoData). + /// + /// Reader positioned on a bucket row. + /// The query's aggregate family. + /// The bucket statistics. + internal static BucketRow ReadBucket(DbDataReader 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); + static double? D(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i); + static int N(DbDataReader r, int i) => r.IsDBNull(i) ? 0 : Convert.ToInt32(r.GetValue(i), CultureInfo.InvariantCulture); // A bucket without a usable sample has NULL times but kind 0 (BYTE has no NULL): absence is keyed on the time. - static HistorianValue V(NpgsqlDataReader r, int i) => + static HistorianValue V(DbDataReader r, int i) => r.IsDBNull(i + 4) ? HistorianValue.Empty : ToValue(QuestDbConnection.ReadKind(r, i), D(r, i + 1), r.IsDBNull(i + 2) ? null : r.GetInt64(i + 2), @@ -606,8 +617,8 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource case AggregateFamily.StartEnd: row.First = V(reader, f); row.Last = V(reader, f + 5); - row.FirstAny = ReadStored(reader, QuestDbAggregateQueries.FirstAnyColumn); - row.LastAny = ReadStored(reader, QuestDbAggregateQueries.FirstAnyColumn + 6); + row.FirstAny = ReadStoredOrNull(reader, QuestDbAggregateQueries.FirstAnyColumn); + row.LastAny = ReadStoredOrNull(reader, QuestDbAggregateQueries.FirstAnyColumn + 6); break; default: row.Extreme = D(reader, f); @@ -623,7 +634,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// Reads ts, kind, num_value, int_value, str_value, status starting at and applies the /// shared quality clamp (). /// - private static StoredSample ReadStored(NpgsqlDataReader r, int o) + private static StoredSample ReadStored(DbDataReader r, int o) { var value = ToValue( QuestDbConnection.ReadKind(r, o + 1), @@ -635,6 +646,12 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource return StoredSample.FromStored(QuestDbConnection.ReadUtc(r, o), value, status); } + /// + /// for an optional picked sample: a NULL ts means the row carries no such sample + /// (live-gate 8 F1), so the result is null rather than an . + /// + private static StoredSample? ReadStoredOrNull(DbDataReader r, int o) => r.IsDBNull(o) ? null : ReadStored(r, o); + /// /// Rebuilds a value from its stored columns via the shared , after the /// RC2 repair: QuestDB reads a LONG holding back as NULL, so an Int64 row with diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs index 00a81408..a3f49514 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs @@ -25,6 +25,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// sample is not usable. /// /// +/// Inconsistent rows (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 BadNoData — 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. +/// +/// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is /// 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) @@ -76,12 +82,20 @@ public static class AggregateEvaluator case HistoryAggregateType.Delta: // #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. + // 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)) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs); var bounded = r.FirstAny is { IsUsable: true } && r.LastAny is { IsUsable: true }; return Calc(b - a, bounded ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal); case HistoryAggregateType.MinimumActualTime: 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); default: throw new HistoryAggregateNotSupportedException(aggregate); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs index 8dcb44dd..8b45164a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs @@ -71,33 +71,53 @@ internal static class SqlAggregateQueries /// Family columns: 6..10 first usable sample (ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc), /// 11..15 last usable sample (same order) — all NULL when the bucket has no usable sample; then /// 16..21 the first sample of any quality (SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, - /// StatusCode) and 22..27 the last (same order) — never NULL for a returned (non-empty) bucket. + /// StatusCode) and 22..27 the last (same order) — present for every returned (non-empty) bucket. /// + /// + /// One scan (live-gate 8 F1). The counts and all four picked samples come from one reference to s: + /// a windowed pass ranks each row within its bucket (rfa/rla over every sample, rfu/rlu + /// within its usable / not-usable partition) and the outer GROUP BY both counts and lifts each picked row's + /// columns with a conditional MAX (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. + /// /// Schema-qualified Sample table name. /// The SQL text. /// True for a reverse read (window (@start, @end]). - public static string StartEnd(string sample, bool reverse = false) => - Classified(sample, reverse) + - ", st AS (SELECT b, " + CommonCounts() + " FROM s GROUP BY b)" + - ", f AS (SELECT b, ValueKind, NumValue, IntValue, StrValue, SourceTimeUtc, " + - "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rf, " + - "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rl FROM s WHERE u = 1)" + - ", a AS (SELECT b, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, " + - "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rf, " + - "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rl FROM s) " + - "SELECT st.*, " + - "x.ValueKind, x.NumValue, x.IntValue, x.StrValue, x.SourceTimeUtc, " + - "y.ValueKind, y.NumValue, y.IntValue, y.StrValue, y.SourceTimeUtc, " + - "p.SourceTimeUtc, p.ValueKind, p.NumValue, p.IntValue, p.StrValue, p.StatusCode, " + - "q.SourceTimeUtc, q.ValueKind, q.NumValue, q.IntValue, q.StrValue, q.StatusCode " + - "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 " + - "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;"; + public static string StartEnd(string sample, bool reverse = false) + { + static string Pick(string when, bool any) => + (any ? $"MAX(CASE WHEN {when} THEN SourceTimeUtc END), " : "") + + $"MAX(CASE WHEN {when} THEN ValueKind END), MAX(CASE WHEN {when} THEN NumValue END), " + + $"MAX(CASE WHEN {when} THEN IntValue END), MAX(CASE WHEN {when} THEN StrValue END), " + + (any ? $"MAX(CASE WHEN {when} THEN StatusCode END)" : $"MAX(CASE WHEN {when} THEN SourceTimeUtc END)"); + return Classified(sample, reverse) + + ", w AS (SELECT b, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, u, g, " + + "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rfa, " + + "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rla, " + + "ROW_NUMBER() OVER (PARTITION BY b, u ORDER BY SourceTimeUtc ASC) AS rfu, " + + "ROW_NUMBER() OVER (PARTITION BY b, u ORDER BY SourceTimeUtc DESC) AS rlu FROM s) " + + "SELECT b, " + CommonCounts() + ", " + + Pick("u = 1 AND rfu = 1", any: false) + ", " + + 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;"; + } /// /// MinimumActualTime / MaximumActualTime: the usable sample with the smallest (largest) NumValue, /// ties broken by the earliest SourceTimeUtc. Family columns: 6 NumValue, 7 SourceTimeUtc /// (NULL when the bucket has no usable numeric sample). /// + /// + /// Deliberately still two references to s (counts + the ranked extreme), unlike : + /// every single-scan form measured (a ROW_NUMBER or windowed MIN/MAX 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 BadNoData (never a value-less Good). + /// /// Schema-qualified Sample table name. /// True for MaximumActualTime. /// The SQL text. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs index 90e1b46e..74a58542 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlHistorianConnection.cs @@ -106,7 +106,7 @@ internal sealed class SqlHistorianConnection /// Reader positioned on a row. /// Column ordinal. /// UTC timestamp. - 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); /// True for connection/timeout/deadlock/throttling errors worth retrying. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index 0cda638e..acef70f7 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -1,4 +1,5 @@ using System.Data; +using System.Data.Common; using System.Text; using Microsoft.Data.SqlClient; using Microsoft.Extensions.Logging; @@ -524,12 +525,22 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource } while (await reader.NextResultAsync(ct).ConfigureAwait(false)); } - private static BucketRow ReadBucket(SqlDataReader reader, AggregateFamily family) + /// + /// Maps one per-sample aggregate row ( column layout) to a . + /// 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 and failed the node's whole read; the evaluator answers such a bucket from what + /// is present, or BadNoData). + /// + /// Reader positioned on a bucket row. + /// The query's aggregate family. + /// The bucket statistics. + internal static BucketRow ReadBucket(DbDataReader reader, AggregateFamily family) { - static double? D(SqlDataReader 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 string? S(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetString(i); - static HistorianValue V(SqlDataReader r, int i) => + static double? D(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i); + static long? L(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetInt64(i); + static string? S(DbDataReader r, int i) => r.IsDBNull(i) ? null : r.GetString(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)); var row = new BucketRow @@ -553,8 +564,8 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource case AggregateFamily.StartEnd: row.First = V(reader, f); row.Last = V(reader, f + 5); - row.FirstAny = ReadStored(reader, SqlAggregateQueries.FirstAnyColumn); - row.LastAny = ReadStored(reader, SqlAggregateQueries.FirstAnyColumn + 6); + row.FirstAny = ReadStoredOrNull(reader, SqlAggregateQueries.FirstAnyColumn); + row.LastAny = ReadStoredOrNull(reader, SqlAggregateQueries.FirstAnyColumn + 6); break; default: row.Extreme = D(reader, f); @@ -572,7 +583,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// () — defence in depth for rows written before the writer forced every /// Empty sample Bad. /// - private static StoredSample ReadStored(SqlDataReader r, int o) + private static StoredSample ReadStored(DbDataReader r, int o) { var value = SampleColumnMapper.FromColumns( r.GetByte(o + 1), @@ -583,6 +594,12 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource return StoredSample.FromStored(SqlHistorianConnection.GetUtc(r, o), value, status); } + /// + /// for an optional picked sample: a NULL SourceTimeUtc means the row carries no such + /// sample (live-gate 8 F1), so the result is null rather than an exception. + /// + private static StoredSample? ReadStoredOrNull(DbDataReader r, int o) => r.IsDBNull(o) ? null : ReadStored(r, o); + /// Read-only tag lookup (never creates a row); runs the schema bootstrap on a cache miss. private Task ResolveAsync(string name, CancellationToken ct) => _catalog.TryGetTagIdAsync(name, ct); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/LiveWriteReadRaceStressTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/LiveWriteReadRaceStressTests.cs new file mode 100644 index 00000000..c6913252 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/LiveWriteReadRaceStressTests.cs @@ -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; + +/// +/// 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 InvalidCastException +/// 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 BadNoData. 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. +/// +/// +/// Opt-in: runs only when HISTORIAN_QUESTDB_STRESS_SECONDS is set (and the fixture's connection is), so it never +/// slows the default suite. Category SlowUnit as well, for filter-based exclusion. +/// +[Collection(QuestDbCollection.Name)] +[Trait("Category", "SlowUnit")] +public sealed class LiveWriteReadRaceStressTests(QuestDbFixture fx) +{ + /// Seconds to run the race loop for; absent ⇒ the test skips. + 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(); + + 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); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index ff4d6aad..6c2b3afd 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -107,6 +107,19 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend /// The tag. /// Rows: time (already rounded by the caller or not — it is bound through BindTimestamp), value, status, deleted. 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); + } + + /// + /// 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). + /// + /// The tag. + /// Rows: time, value, status, deleted. + public async Task InsertRowsNoWaitAsync(string tag, (DateTime T, HistorianValue V, uint Status, bool Deleted)[] rows) { if (rows.Length == 0) return; await using (var c = await Conn.OpenAsync(CancellationToken.None)) @@ -140,7 +153,6 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend await cmd.ExecuteNonQueryAsync(); } } - await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None); } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/BucketRowMappingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/BucketRowMappingTests.cs new file mode 100644 index 00000000..b029702f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/BucketRowMappingTests.cs @@ -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; + +/// +/// Live-gate 8 F1: the per-sample aggregate row mapper () over a +/// hand-built row in the column layout. A row whose counts include +/// samples but whose first/last-of-any-quality picks are NULL used to throw InvalidCastException: Column 'ts' is +/// null and fail the node's whole read; it now maps to absent picks the evaluator answers as BadNoData. +/// +public sealed class BucketRowMappingTests +{ + private const uint NoDataCalc = 0x809B0401u; + private static readonly DateTime T = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc); + + /// The StartEnd layout: b + 5 counts, two usable picks (kind, num, int, str, ts), two any picks (ts, kind, num, int, str, status). + 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); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs index ff5e8458..3c681fc7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs @@ -257,4 +257,52 @@ public sealed class AggregateEvaluatorTests avg.Value.ShouldBe(15d); avg.StatusCode.ShouldBe(UncertainCalc); } + + // ---- Live-gate 8 F1: a row whose counts and picks disagree (a backend that read them from two table versions). + + /// Counts say B0 holds two Good samples, but none of the picked samples came back. + 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(); + } + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs new file mode 100644 index 00000000..9e211022 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs @@ -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; + +/// +/// Live-gate 8 F1 (defence in depth on SQL Server, where it was not observed): the per-sample aggregate row mapper +/// () over a hand-built row in the +/// column layout. A row whose counts include samples but whose +/// first/last-of-any-quality picks are NULL used to throw out of ReadStored and fail the node's whole read; it now +/// maps to absent picks the evaluator answers as BadNoData. +/// +public sealed class BucketRowMappingTests +{ + private const uint NoDataCalc = 0x809B0401u; + private static readonly DateTime T = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc); + + /// b + 5 counts, two usable picks (kind, num, int, str, ts), two any picks (ts, kind, num, int, str, status). + 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); + } +}