fix(historian): #683 review — bound the StartEnd pick joins to the window; usable picks re-check usability

- SqlServer StartEnd join-back: every pick join also carries the window predicate
  (forward >= @start AND < @end, reverse > @start AND <= @end). Keyed on the pick
  time alone a hash/merge/adaptive join read the tag's whole history (review:
  43,657 vs 361 logical reads), and parameter sniffing reused that plan for every
  tag. Forced OPTION (HASH JOIN), 1d@1h on 6M rows: 152,524 -> 6,756 reads.
- Usable picks (x/y) re-check the exact u predicate in the join, so a concurrent
  HistoryUpdate Replace that turned the picked sample Bad/Empty reads as absent
  (BadNoData), never a Bad row's value under Delta.
- StartEndJoinBackTests (integration): 300k history rows either side of a 1 h
  window, forced HASH and MERGE joins, results identical to the unhinted plan and
  Start/End correct, session logical reads < 1,000 (183 bounded; 8,055 with the
  bound removed — the test fails then).
- TagIndexTests: the refused-ADD-INDEX test asserts the Warning naming the ALTER.
- Historian.md: re-measured table incl. long-history windows (output identical by
  EXCEPT both ways in all 7 cases) and the forced-join read counts.

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 (720 buckets) | 19.0–21.7 s | 3.4–3.5 s |
-  | 30 d @ 1 h, reverse | 18.1–29.9 s | 3.4–5.2 s |
+  | 30 d @ 1 h, reverse | 24.7–30.3 s | 3.4 s |
-  | 7 d @ 1 d (7 buckets) | 5.2–6.1 s | 0.79–0.80 s |
+  | 7 d @ 1 d (7 buckets) | 5.5–6.0 s | 0.78–0.81 s |
-  | 1 d @ 1 min (1440 buckets) | 0.31–0.40 s | 0.16 s |
+  | 7 d @ 1 min (10,080 buckets) | 2.1–2.3 s | 1.08–1.11 s |
-  | 1 d @ 7 min, reverse, sub-µs start | 0.46–0.47 s | 0.12 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
+  ⚠️ **Every pick join is bounded to the window** (`@start`/`@end`, the same predicate as the window CTE), not only keyed on the pick
-  statement, so a sample **deleted** by HistoryUpdate between the aggregate and its seek leaves that pick NULL while the counts include
+  time. Keyed alone, the optimizer may choose a hash/merge/adaptive join whose inner input is the tag's **whole history**, and
-  it: the null-tolerant reader below answers that bucket's Start/End/Delta `BadNoData`, transiently. Min/MaxActualTime
+  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

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" (<c>ReadStoredOrNull</c>, live-gate 8 F1)
     ///         and the evaluator answers that bucket's Start/End/Delta <c>BadNoData</c> — 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
+    ///         sample replaced in between is picked with its new value (same time) — except that a usable pick re-checks
-    ///         neither half (the pick times come from the aggregate).
+    ///         usability in its join (the exact <c>u</c> predicate), so a Replace that turned it Bad/Empty reads as absent
+    ///         (<c>BadNoData</c>), 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).
+    ///     </para>
+    ///     <para>
+    ///         Every join is also bounded to the window (the same <c>@start</c>/<c>@end</c> predicate as <c>s</c>): 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.
     ///     </para>
     /// </remarks>
     /// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
@@ -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;";
     }
 
     /// <summary>

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<Exception>(() => 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));
+    }
+
+    /// <summary>Captures the rendered Warning messages.</summary>
+    private sealed class ListLogger : Microsoft.Extensions.Logging.ILogger
+    {
+        private readonly System.Collections.Concurrent.ConcurrentQueue<string> _warnings = new();
+        public IReadOnlyCollection<string> Warnings => _warnings;
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+        public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
+        public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
+            TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            if (logLevel == Microsoft.Extensions.Logging.LogLevel.Warning) _warnings.Enqueue(formatter(state, exception));
+        }
     }
 
     [Fact]

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;
+
+/// <summary>
+///     #683: <see cref="SqlAggregateQueries.StartEnd"/> 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 (<c>OPTION (HASH JOIN)</c>) 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.
+/// </summary>
+[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;
+
+    /// <summary>
+    ///     Tag <c>H</c>: one sample a minute over the hour [+0, +3600) of mixed quality (Good, Uncertain, Bad, an outage
+    ///     marker, a string), plus <see cref="HistoryRowsEachSide"/> rows a second before +0 and after +3600.
+    /// </summary>
+    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);
+    }
+
+    /// <summary>
+    ///     Runs the StartEnd SQL (optionally with a query hint) and returns its rows as strings plus the session's logical
+    ///     reads during the statement (<c>sys.dm_exec_sessions.logical_reads</c> 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.
+    /// </summary>
+    private static async Task<(List<string> 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<long> 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<string>();
+        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));
+    }
+}