diff --git a/docs/Historian.md b/docs/Historian.md index d9041c78..71eda810 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -1607,19 +1607,32 @@ Every query filters `NOT deleted`; every value goes through `StoredSample`, so t the five-reference form on long windows. It is now **aggregate, then join back on the key**: one `GROUP BY` computes the counts and each bucket's four pick *times* (`MIN`/`MAX(SourceTimeUtc)`, any quality and usable), and each pick's columns are fetched by a seek on the clustered key `(TagId, SourceTimeUtc)`. Output is identical (`EXCEPT` both ways empty on 6 M rows, forward and reverse); measured on - 6 M rows / 30 days, one tag, average of 3 runs over 2 rounds: + 6 M rows / 30 days, one tag (so a 1 d or 7 d window leaves up to 5.8 M rows of the tag's history outside it), average of 3 runs + over 2 rounds, plans compiled by the 30 d read and reused (sniffed) by the rest: - | Window | windowed form | join-back | + | Window | windowed form (before) | join-back | |---|---|---| - | 30 d @ 1 h (720 buckets) | 18.6–31.7 s | 3.4 s | - | 30 d @ 1 h, reverse | 18.1–29.9 s | 3.4–5.2 s | - | 7 d @ 1 d (7 buckets) | 5.2–6.1 s | 0.79–0.80 s | - | 1 d @ 1 min (1440 buckets) | 0.31–0.40 s | 0.16 s | - | 1 d @ 7 min, reverse, sub-µs start | 0.46–0.47 s | 0.12 s | + | 30 d @ 1 h (720 buckets) | 19.0–21.7 s | 3.4–3.5 s | + | 30 d @ 1 h, reverse | 24.7–30.3 s | 3.4 s | + | 7 d @ 1 d (7 buckets) | 5.5–6.0 s | 0.78–0.81 s | + | 7 d @ 1 min (10,080 buckets) | 2.1–2.3 s | 1.08–1.11 s | + | 1 d @ 1 h (24 buckets) | 0.55–0.60 s | 0.12–0.13 s | + | 1 d @ 1 h, reverse | 0.52–0.55 s | 0.11 s | + | 1 d @ 1 min (1440 buckets) | 0.29–0.32 s | 0.16 s | - (The dev SQL container is slow in absolute terms — the ratio is the point.) ⚠️ The seeks are separate reads of the table within one - statement, so a sample **deleted** by HistoryUpdate between the aggregate and its seek leaves that pick NULL while the counts include - it: the null-tolerant reader below answers that bucket's Start/End/Delta `BadNoData`, transiently. Min/MaxActualTime + ⚠️ **Every pick join is bounded to the window** (`@start`/`@end`, the same predicate as the window CTE), not only keyed on the pick + time. Keyed alone, the optimizer may choose a hash/merge/adaptive join whose inner input is the tag's **whole history**, and + parameter sniffing then reuses that plan for every tag — a review measured 43,657 logical reads vs 361 for a 1 d @ 1 h read of a tag + with 3 M rows outside the window. Forcing that shape here, 1 d @ 1 h: **152,524** session logical reads unbounded vs **6,756** bounded + under `OPTION (HASH JOIN)` (6,655 under `MERGE JOIN`; the unforced plan reads 1,846 either way). `StartEndJoinBackTests` pins it: + a tag with 300 k rows either side of a 1 h window, forced hash and merge joins, < 1,000 reads (183 bounded, 8,055 with the bound + removed). Wider windows trade reads for time — 7 d @ 1 min reads 174 k pages (four key seeks per bucket) against the windowed form's + 46 k, yet runs in half the time. (The dev SQL container is slow in absolute terms — the ratios are the point.) + + ⚠️ The seeks are separate reads of the table within one statement, so a sample **deleted** by HistoryUpdate between the aggregate + and its seek leaves that pick NULL while the counts include it: the null-tolerant reader below answers that bucket's Start/End/Delta + `BadNoData`, transiently. A usable pick's join also re-checks usability (the exact `u` predicate), so a HistoryUpdate **Replace** + that turned the picked sample Bad/Empty in between reads the same way — never a Bad row's value under Delta. 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 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 50bbdf0d..c2d953cc 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 @@ -93,8 +93,15 @@ internal static class SqlAggregateQueries /// deleted (HistoryUpdate DeleteRaw/DeleteAtTime) between the aggregate and its seek leaves that pick NULL while /// the counts still include it. The reader maps a NULL pick to "absent" (ReadStoredOrNull, live-gate 8 F1) /// and the evaluator answers that bucket's Start/End/Delta BadNoData — transient, until the next read. A - /// sample replaced in between is picked with its new value (same time); a sample inserted in between is in - /// neither half (the pick times come from the aggregate). + /// sample replaced in between is picked with its new value (same time) — except that a usable pick re-checks + /// usability in its join (the exact u predicate), so a Replace that turned it Bad/Empty reads as absent + /// (BadNoData), never as a Bad row's value under Delta. A sample inserted in between is in neither half + /// (the pick times come from the aggregate). + /// + /// + /// Every join is also bounded to the window (the same @start/@end predicate as s): keyed + /// on the pick time alone, the optimizer may choose a hash/merge/adaptive join whose inner input is the tag's + /// whole history, and parameter sniffing then reuses that plan for every tag. /// /// /// Schema-qualified Sample table name. @@ -106,14 +113,24 @@ internal static class SqlAggregateQueries // quality picks (time, kind, num, int, str, status). static string Usable(string a) => $"{a}.ValueKind, {a}.NumValue, {a}.IntValue, {a}.StrValue, {a}.SourceTimeUtc"; static string Any(string a) => $"{a}.SourceTimeUtc, {a}.ValueKind, {a}.NumValue, {a}.IntValue, {a}.StrValue, {a}.StatusCode"; - string Join(string a, string time) => $"LEFT JOIN {sample} {a} ON {a}.TagId = @id AND {a}.SourceTimeUtc = st.{time} "; + // Each seek is bounded to the window as well as keyed on the pick time: without the bounds the optimizer may pick a + // hash/merge/adaptive join whose inner input is the tag's WHOLE history (TagId = @id only), and parameter sniffing + // then reuses that plan for every tag (review of 910befc8: 43,657 logical reads vs 361 on a tag with 3M rows of + // history outside a 1 d window). A usable pick also re-checks usability (the exact `u` predicate), so a sample a + // HistoryUpdate Replace turned Bad/Empty in between joins nothing ⇒ absent ⇒ BadNoData, never a Bad row's value. + string Window(string a) => reverse + ? $"{a}.SourceTimeUtc > @start AND {a}.SourceTimeUtc <= @end" + : $"{a}.SourceTimeUtc >= @start AND {a}.SourceTimeUtc < @end"; + string Join(string a, string time, bool usable) => + $"LEFT JOIN {sample} {a} ON {a}.TagId = @id AND {a}.SourceTimeUtc = st.{time} AND {Window(a)}" + + (usable ? $" AND {a}.StatusCode >= 0 AND {a}.ValueKind <> 0" : "") + " "; return Classified(sample, reverse) + ", st AS (SELECT b, " + CommonCounts() + ", " + "MIN(CASE WHEN u = 1 THEN SourceTimeUtc END) AS fu, MAX(CASE WHEN u = 1 THEN SourceTimeUtc END) AS lu, " + "MIN(SourceTimeUtc) AS fa, MAX(SourceTimeUtc) AS la FROM s GROUP BY b) " + "SELECT st.b, st.TotalCount, st.UsableCount, st.GoodCount, st.BadCount, st.NonNumericCount, " + Usable("x") + ", " + Usable("y") + ", " + Any("p") + ", " + Any("q") + " FROM st " + - Join("x", "fu") + Join("y", "lu") + Join("p", "fa") + Join("q", "la") + "ORDER BY st.b;"; + Join("x", "fu", usable: true) + Join("y", "lu", usable: true) + Join("p", "fa", usable: false) + Join("q", "la", usable: false) + "ORDER BY st.b;"; } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs index 2e71c621..bc5ac9da 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs @@ -100,7 +100,24 @@ public sealed class TagIndexTests(QuestDbFixture fx) await using var alter = conn.CreateCommand(c, QuestDbTables.AddTagIndex(conn.SampleTable)); await Should.ThrowAsync(() => alter.ExecuteNonQueryAsync(Ct)); // precondition: really refused } - await new QuestDbSchemaBootstrapper(conn, o).EnsureAsync(Ct); + var log = new ListLogger(); + await new QuestDbSchemaBootstrapper(conn, o, log).EnsureAsync(Ct); + log.Warnings.ShouldContain(m => m.Contains("Could not add the tag index", StringComparison.Ordinal) + && m.Contains(QuestDbTables.AddTagIndex(conn.SampleTable), StringComparison.Ordinal)); + } + + /// Captures the rendered Warning messages. + private sealed class ListLogger : Microsoft.Extensions.Logging.ILogger + { + private readonly System.Collections.Concurrent.ConcurrentQueue _warnings = new(); + public IReadOnlyCollection Warnings => _warnings; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true; + public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, + TState state, Exception? exception, Func formatter) + { + if (logLevel == Microsoft.Extensions.Logging.LogLevel.Warning) _warnings.Enqueue(formatter(state, exception)); + } } [Fact] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/StartEndJoinBackTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/StartEndJoinBackTests.cs new file mode 100644 index 00000000..04e40336 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/StartEndJoinBackTests.cs @@ -0,0 +1,123 @@ +using System.Data; +using Microsoft.Data.SqlClient; +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.IntegrationTests; + +/// +/// #683: aggregates the pick times, then fetches each pick by a key join. +/// The joins must be bounded to the window: keyed on the pick time alone, a hash/merge/adaptive join reads the tag's +/// whole history (review of 910befc8: 43,657 logical reads vs 361 on a tag with 3M rows outside a 1 d window). The +/// plan choice depends on estimates, so the test FORCES the worst shape (OPTION (HASH JOIN)) on a tag with long +/// history either side of the window and asserts the statement reads only window-sized pages — and returns exactly +/// what the unhinted plan and the data source return. +/// +[Collection(SqlHistorianCollection.Name)] +public sealed class StartEndJoinBackTests(SqlHistorianFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static DateTime At(double s) => ReadTestDb.At(s); + private const int HistoryRowsEachSide = 150_000; + + /// + /// Tag H: one sample a minute over the hour [+0, +3600) of mixed quality (Good, Uncertain, Bad, an outage + /// marker, a string), plus rows a second before +0 and after +3600. + /// + private static async Task<(ReadTestDb Db, int Id, List<(DateTime T, HistorianValue V, uint S)> Window)> SeedAsync(SqlHistorianFixture fx) + { + var db = await ReadTestDb.CreateAsync(fx); + var window = new List<(DateTime T, HistorianValue V, uint S)>(); + for (var i = 0; i < 60; i++) + { + var (v, s) = (i % 7) switch + { + 3 => (HistorianValue.FromDouble(i), ReadTestDb.Bad), + 4 => (HistorianValue.FromDouble(i), ReadTestDb.Uncertain), + 5 => (HistorianValue.Empty, ReadTestDb.BadNoCommunication), + 6 => (HistorianValue.FromString("s" + i), ReadTestDb.Good), + _ => (HistorianValue.FromDouble(i), ReadTestDb.Good), + }; + window.Add((At(i * 60 + 7), v, s)); + } + var id = await db.SeedAsync("H", window.ToArray()); + // The history: inserted set-based (a per-row INSERT of 300k rows would dominate the test). + await db.ScalarAsync( + $"WITH n AS (SELECT TOP ({HistoryRowsEachSide}) ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS x " + + "FROM sys.all_objects a CROSS JOIN sys.all_objects b) " + + $"INSERT {db.Conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " + + $"SELECT {id}, DATEADD(second, -x, CAST('{At(0):yyyy-MM-dd HH:mm:ss.fffffff}' AS datetime2(7))), 3, x, NULL, NULL, 0, SYSUTCDATETIME() FROM n " + + $"UNION ALL SELECT {id}, DATEADD(second, x, CAST('{At(3600):yyyy-MM-dd HH:mm:ss.fffffff}' AS datetime2(7))), 3, x, NULL, NULL, 0, SYSUTCDATETIME() FROM n; " + + "SELECT 0"); + await db.ScalarAsync($"UPDATE STATISTICS {db.Conn.Q("Sample")}; SELECT 0"); + return (db, id, window); + } + + /// + /// Runs the StartEnd SQL (optionally with a query hint) and returns its rows as strings plus the session's logical + /// reads during the statement (sys.dm_exec_sessions.logical_reads before/after — STATISTICS IO messages do + /// not reach SqlClient's InfoMessage here). The two probes add a few reads of their own; the bound is far apart. + /// + private static async Task<(List Rows, long LogicalReads)> RunAsync(ReadTestDb db, int id, BucketPlan plan, string? hint) + { + var sql = SqlAggregateQueries.StartEnd(db.Conn.Q("Sample"), plan.Reversed); + if (hint is not null) sql = sql.TrimEnd().TrimEnd(';') + $" OPTION ({hint});"; + await using var c = await db.Conn.OpenAsync(Ct); + async Task SessionReadsAsync() + { + await using var probe = db.Conn.CreateCommand(c, "SELECT logical_reads FROM sys.dm_exec_sessions WHERE session_id = @@SPID"); + return Convert.ToInt64(await probe.ExecuteScalarAsync(Ct)); + } + var before = await SessionReadsAsync(); + await using var cmd = db.Conn.CreateCommand(c, sql); + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id }); + SqlHistorianConnection.AddDateTime2(cmd, "@start", plan.StartUtc); + SqlHistorianConnection.AddDateTime2(cmd, "@end", plan.EndUtc); + cmd.Parameters.Add(new SqlParameter("@intervalTicks", SqlDbType.BigInt) { Value = plan.IntervalTicks }); + cmd.Parameters.Add(new SqlParameter("@bucketAdjustTicks", SqlDbType.BigInt) { Value = plan.BucketIndexAdjustTicks }); + var rows = new List(); + await using (var r = await cmd.ExecuteReaderAsync(Ct)) + while (await r.ReadAsync(Ct)) + rows.Add(string.Join("|", Enumerable.Range(0, r.FieldCount).Select(i => r.IsDBNull(i) ? "NULL" : Convert.ToString(r.GetValue(i), System.Globalization.CultureInfo.InvariantCulture)))); + return (rows, await SessionReadsAsync() - before); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task The_pick_joins_read_only_the_window_even_under_a_forced_hash_join(bool reverse) + { + fx.SkipIfUnavailable(); + var (db, id, window) = await SeedAsync(fx); + var plan = reverse + ? BucketPlan.Create(At(3600), At(0), TimeSpan.FromMinutes(10)) + : BucketPlan.Create(At(0), At(3600), TimeSpan.FromMinutes(10)); + + var (plain, _) = await RunAsync(db, id, plan, hint: null); + var (hashed, hashReads) = await RunAsync(db, id, plan, hint: "HASH JOIN"); + var (merged, mergeReads) = await RunAsync(db, id, plan, hint: "MERGE JOIN"); + + plain.Count.ShouldBe(6); + hashed.ShouldBe(plain); + merged.ShouldBe(plain); + // Measured: 183 session logical reads bounded vs 8,055 with the window bound removed from the joins (300k history + // rows either side; the threshold sits ~5x above the one and ~8x below the other). Every join + // bounded ⇒ each input is a window-range seek, whatever the join algorithm. + hashReads.ShouldBeGreaterThan(0L, "the session read counter must move (positive control)"); + hashReads.ShouldBeLessThan(1000, "a hash join's inner input must be bounded to the window"); + mergeReads.ShouldBeLessThan(1000, "a merge join's inner input must be bounded to the window"); + + // And the picks are right: Start / End are each bucket's first / last sample of any quality. + var src = db.Source(); + var start = await src.ReadProcessedAsync("H", reverse ? At(3600) : At(0), reverse ? At(0) : At(3600), + TimeSpan.FromMinutes(10), HistoryAggregateType.Start, Ct); + var end = await src.ReadProcessedAsync("H", reverse ? At(3600) : At(0), reverse ? At(0) : At(3600), + TimeSpan.FromMinutes(10), HistoryAggregateType.End, Ct); + var buckets = Enumerable.Range(0, 6).Select(b => window.Skip(b * 10).Take(10).ToList()).ToList(); + if (reverse) buckets.Reverse(); + start.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(buckets.Select(b => (DateTime?)b[0].T)); + end.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(buckets.Select(b => (DateTime?)b[^1].T)); + } +}