From 8b5fdf0d73e8535e4af14b962d381e73a93aca05 Mon Sep 17 00:00:00 2001
From: Erik Darling <2136037+erikdarlingdata@users.noreply.github.com>
Date: Mon, 12 Oct 2026 15:16:34 -0400
Subject: [PATCH 1/7] Darling: per-collection procedure totals, written with
each procedure_stats batch (part of #5638)
---
.../ProcedureStatsTotalsTests.cs | 194 +++++++++
.../ProcedureStatsTotalsWriterLiveTests.cs | 391 ++++++++++++++++++
.../DarlingCollectorRunner.cs | 20 +-
.../DarlingRetention.cs | 25 ++
.../Mcp/DarlingMcpTrendTools.cs | 18 +-
.../PgMigrations.cs | 14 +
.../ProcedureStatsTotals.cs | 110 +++++
.../ProcedureStatsTotalsWriter.cs | 64 +++
.../StorageVersion.cs | 2 +-
.../ViewerDataService.cs | 16 +-
10 files changed, 847 insertions(+), 7 deletions(-)
create mode 100644 Darling/Darling.Tests/ProcedureStatsTotalsTests.cs
create mode 100644 Darling/Darling.Tests/ProcedureStatsTotalsWriterLiveTests.cs
create mode 100644 Darling/PerformanceMonitor.Darling.Storage/ProcedureStatsTotals.cs
create mode 100644 Darling/PerformanceMonitor.Darling.Storage/ProcedureStatsTotalsWriter.cs
diff --git a/Darling/Darling.Tests/ProcedureStatsTotalsTests.cs b/Darling/Darling.Tests/ProcedureStatsTotalsTests.cs
new file mode 100644
index 0000000000..b7bfa7e313
--- /dev/null
+++ b/Darling/Darling.Tests/ProcedureStatsTotalsTests.cs
@@ -0,0 +1,194 @@
+/*
+ * Copyright (c) 2026 Erik Darling, Darling Data LLC
+ *
+ * This file is part of the SQL Server Performance Monitor.
+ *
+ * Licensed under the MIT License. See LICENSE file in the project root for full license information.
+ */
+
+using System;
+using System.Linq;
+using PerformanceMonitor.Darling.Service;
+using PerformanceMonitor.Darling.Service.Mcp;
+using PerformanceMonitor.Darling.Storage;
+using Xunit;
+
+namespace Darling.Tests;
+
+///
+/// Pins for the per-collection procedure totals (#5638, V176) that need no store: the rung and its text, the upsert statement, the
+/// runner's call site, the retention registration, and the MCP duration trends' tier clock. The live behavior is in
+/// .
+///
+public sealed class ProcedureStatsTotalsTests
+{
+ private static string Normalize(string text) => text.Replace("\r\n", "\n", StringComparison.Ordinal);
+
+ [Fact]
+ public void TheRung_IsV176_AndEmbedsTheCreateText()
+ {
+ Assert.Equal(176, ProcedureStatsTotals.RungVersion);
+ Assert.True(StorageVersion.SchemaVersion >= ProcedureStatsTotals.RungVersion);
+
+ var migrations = RepoFile.ReadRepoFile("Darling", "PerformanceMonitor.Darling.Storage", "PgMigrations.cs");
+ Assert.Contains("new Migration(176, \"procedure-stats-totals\", V176Sql)", migrations, StringComparison.Ordinal);
+ Assert.Contains("private const string V176Sql = ProcedureStatsTotals.CreateSql;", migrations, StringComparison.Ordinal);
+ }
+
+ [Fact]
+ public void TheCreateText_PinsTheShape_NoBackfill_AndTheNullsNotDistinctKey()
+ {
+ var sql = Normalize(ProcedureStatsTotals.CreateSql);
+
+ Assert.Contains("CREATE TABLE IF NOT EXISTS collect.procedure_stats_totals", sql, StringComparison.Ordinal);
+ Assert.Contains("server_id integer NOT NULL,", sql, StringComparison.Ordinal);
+ Assert.Contains("collection_time timestamp NOT NULL,", sql, StringComparison.Ordinal);
+ Assert.Contains("database_name text,", sql, StringComparison.Ordinal);
+ Assert.Contains("total_elapsed_time bigint,", sql, StringComparison.Ordinal);
+ Assert.Contains("total_execution_count bigint,", sql, StringComparison.Ordinal);
+ Assert.Contains("row_count bigint NOT NULL,", sql, StringComparison.Ordinal);
+ Assert.Contains("max_sample_interval_seconds integer", sql, StringComparison.Ordinal);
+ Assert.Contains(
+ "CREATE UNIQUE INDEX IF NOT EXISTS ux_procedure_stats_totals ON collect.procedure_stats_totals (server_id, collection_time, database_name) NULLS NOT DISTINCT;",
+ sql, StringComparison.Ordinal);
+
+ /* No backfill: the rung moves no existing row. */
+ Assert.DoesNotContain("INSERT", sql, StringComparison.OrdinalIgnoreCase);
+ Assert.DoesNotContain("UPDATE", sql, StringComparison.OrdinalIgnoreCase);
+ Assert.DoesNotContain("create_hypertable", sql, StringComparison.OrdinalIgnoreCase);
+ }
+
+ [Fact]
+ public void TheUpsert_ReadsBackTheStoredRows_AndRecomputesEveryMeasureOnConflict()
+ {
+ var sql = Normalize(ProcedureStatsTotals.UpsertSql);
+
+ Assert.Contains("FROM collect.procedure_stats\nWHERE server_id = $1::integer\nAND collection_time = $2::timestamp\nGROUP BY server_id, collection_time, database_name", sql, StringComparison.Ordinal);
+ Assert.Contains("SUM(delta_elapsed_time)::bigint,", sql, StringComparison.Ordinal);
+ Assert.Contains("SUM(delta_execution_count)::bigint,", sql, StringComparison.Ordinal);
+ Assert.Contains("COUNT(*),", sql, StringComparison.Ordinal);
+ Assert.Contains("MAX(sample_interval_seconds)", sql, StringComparison.Ordinal);
+ Assert.Contains("ON CONFLICT (server_id, collection_time, database_name) DO UPDATE SET", sql, StringComparison.Ordinal);
+
+ /* Recompute, never add: every measure is set from EXCLUDED, and no measure is added to its old value. */
+ foreach (var measure in new[] { "total_elapsed_time", "total_execution_count", "row_count", "max_sample_interval_seconds" })
+ {
+ Assert.Contains($"{measure} = EXCLUDED.{measure}", sql, StringComparison.Ordinal);
+ }
+
+ Assert.DoesNotContain("t.row_count +", sql, StringComparison.Ordinal);
+ Assert.DoesNotContain("COALESCE(", sql, StringComparison.Ordinal);
+ }
+
+ [Fact]
+ public void TheRunner_WritesTheTotalsOnTheCopysTransaction_AfterTheCopy_BeforeTheCommit()
+ {
+ var runner = Normalize(RepoFile.ReadRepoFile("Darling", "PerformanceMonitor.Darling.Service", "DarlingCollectorRunner.cs"));
+ var start = runner.IndexOf("private async Task CopyBatchOnceAsync(", StringComparison.Ordinal);
+ Assert.True(start > 0);
+ var body = runner[start..];
+
+ var call = body.IndexOf("ProcedureStatsTotalsWriter.UpsertBatchAsync(", StringComparison.Ordinal);
+ var commit = body.IndexOf("await transaction.CommitAsync(cancellationToken);", StringComparison.Ordinal);
+ var ledger = body.IndexOf("QueryStatsHourLedgerWriter.AddBatchAsync(", StringComparison.Ordinal);
+ Assert.True(ledger > 0 && call > ledger && commit > call, "the totals upsert sits after the COPY block and before the commit");
+
+ /* On the COPY's transaction, for procedure_stats batches only, with no savepoint or catch around it. */
+ Assert.Contains("pgConnection, transaction, server.ServerId, storedCollectionTime,\n rowsWritten", body, StringComparison.Ordinal);
+ Assert.Contains("var totalsBatch = string.Equals(definition.TargetTable, ProcedureStatsCollector.Instance.TargetTable", body, StringComparison.Ordinal);
+ Assert.Contains("|| ledgerBatch || totalsBatch", body, StringComparison.Ordinal);
+ var between = body[(call - 400)..call];
+ Assert.DoesNotContain("catch", between, StringComparison.Ordinal);
+ Assert.DoesNotContain("SAVEPOINT", between, StringComparison.OrdinalIgnoreCase);
+
+ /* collect.procedure_stats has exactly one writer: the runner's COPY. */
+ foreach (var file in new[] { "DarlingWorker.cs", "DarlingRetention.cs" })
+ {
+ var text = RepoFile.ReadRepoFile("Darling", "PerformanceMonitor.Darling.Service", file);
+ Assert.DoesNotContain("INSERT INTO collect.procedure_stats ", text, StringComparison.Ordinal);
+ }
+ }
+
+ [Fact]
+ public void Retention_PurgesTheTotals_AtProcedureStatsDays_NeverBeforeTheRawRows()
+ {
+ var retention = Normalize(RepoFile.ReadRepoFile("Darling", "PerformanceMonitor.Darling.Service", "DarlingRetention.cs"));
+
+ Assert.Contains("TimeSlicedDeleteSql(ProcedureStatsTotals.Table, \"collection_time\", ProcedureStatsTotals.PrunePredicate)", retention, StringComparison.Ordinal);
+ Assert.Contains("EffectivePurgeRetentionDays(\n \"procedure_stats\",", retention, StringComparison.Ordinal);
+
+ var delete = DarlingRetention.TimeSlicedDeleteSql(ProcedureStatsTotals.Table, "collection_time", ProcedureStatsTotals.PrunePredicate);
+ Assert.StartsWith("DELETE FROM collect.procedure_stats_totals WHERE collection_time < $1 AND collection_time < COALESCE((SELECT min(collection_time) FROM collect.procedure_stats), $1)", delete, StringComparison.Ordinal);
+
+ /* The totals table is a plain table outside the collector catalog: no hypertable conversion, no collector schedule entry. */
+ Assert.DoesNotContain(CollectorCatalogTables(), t => string.Equals(t, "procedure_stats_totals", StringComparison.Ordinal));
+ }
+
+ private static string[] CollectorCatalogTables() =>
+ PerformanceMonitor.Collectors.CollectorCatalog.All.Select(c => c.TargetTable).ToArray();
+
+ /* ---- the MCP duration trends' tier clock (#5638) ---- */
+
+ private static readonly DateTime Now = new(2026, 10, 12, 12, 0, 0, DateTimeKind.Utc);
+
+ [Fact]
+ public void TierClock_WithNoAsOf_IsTheWindowEndItself()
+ {
+ Assert.Equal(Now, DarlingMcpTrendTools.TierClock(null, Now, () => Now.AddMinutes(5)));
+ Assert.Equal(Now, DarlingMcpTrendTools.TierClock(" ", Now, () => Now.AddMinutes(5)));
+ }
+
+ [Fact]
+ public void TierClock_WithAnAsOf_IsOneWallClockReading_NeverTheAnchor()
+ {
+ var anchor = Now.AddDays(-30);
+ var wall = Now.AddSeconds(3);
+ var reads = 0;
+
+ var clock = DarlingMcpTrendTools.TierClock("2026-09-12T12:00:00Z", anchor, () => { reads++; return wall; });
+
+ Assert.Equal(wall, clock);
+ Assert.NotEqual(anchor, clock);
+ Assert.Equal(1, reads);
+ }
+
+ ///
+ /// The regression: a 95 h window (start exactly on the raw boundary, 96 h less the 1 h margin) with no as_of routes Raw every time
+ /// when the tier is decided against the instant the window was computed. Deciding it against a clock read a few milliseconds LATER
+ /// (the old default) flips it to the hourly tier. Both procedure and query routes.
+ ///
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(7)]
+ [InlineData(250)]
+ public void A95HourWindowWithNoAsOf_RoutesRawEveryTime(int millisecondsLater)
+ {
+ var windowEnd = Now;
+ var start = windowEnd.AddHours(-95);
+ var coverage = RollupCoverage.Unknown;
+
+ Assert.Equal(TimescaleSupport.RawRetentionSpan - DarlingTrendReader.RawTierMargin, TimeSpan.FromHours(95));
+
+ var pinned = DarlingMcpTrendTools.TierClock(null, windowEnd, () => windowEnd.AddMilliseconds(millisecondsLater));
+ Assert.Equal(RetentionTier.Raw, DarlingTrendReader.ResolveProcedureDurationTrendRoute(start, RollupAvailability.All, coverage, nowUtc: pinned, windowEndUtc: windowEnd).Tier);
+ Assert.Equal(RetentionTier.Raw, DarlingTrendReader.ResolveQueryDurationTrendRoute(start, RollupAvailability.All, coverage, nowUtc: pinned, windowEndUtc: windowEnd).Tier);
+
+ /* The old behaviour, for contrast: a clock read after the window flips the same window to the hourly tier. */
+ if (millisecondsLater > 0)
+ {
+ var late = windowEnd.AddMilliseconds(millisecondsLater);
+ Assert.Equal(RetentionTier.Hourly, DarlingTrendReader.ResolveProcedureDurationTrendRoute(start, RollupAvailability.All, coverage, nowUtc: late, windowEndUtc: windowEnd).Tier);
+ }
+ }
+
+ [Fact]
+ public void TheDurationTrendTools_PassTheirOwnClock_ToBothRoutes()
+ {
+ var tools = Normalize(RepoFile.ReadRepoFile("Darling", "PerformanceMonitor.Darling.Service", "Mcp", "DarlingMcpTrendTools.cs"));
+
+ Assert.Contains("ResolveQueryDurationTrendRoute(startUtc, rollups, coverage, nowUtc: TierClock(as_of, now), windowEndUtc: now)", tools, StringComparison.Ordinal);
+ Assert.Contains("ResolveProcedureDurationTrendRoute(startUtc, rollups, coverage, nowUtc: TierClock(as_of, now), windowEndUtc: now)", tools, StringComparison.Ordinal);
+ Assert.DoesNotContain("coverage, windowEndUtc: now)", tools, StringComparison.Ordinal);
+ }
+}
diff --git a/Darling/Darling.Tests/ProcedureStatsTotalsWriterLiveTests.cs b/Darling/Darling.Tests/ProcedureStatsTotalsWriterLiveTests.cs
new file mode 100644
index 0000000000..4a35a15b17
--- /dev/null
+++ b/Darling/Darling.Tests/ProcedureStatsTotalsWriterLiveTests.cs
@@ -0,0 +1,391 @@
+/*
+ * Copyright (c) 2026 Erik Darling, Darling Data LLC
+ *
+ * This file is part of the SQL Server Performance Monitor.
+ *
+ * Licensed under the MIT License. See LICENSE file in the project root for full license information.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+using Npgsql;
+using PerformanceMonitor.Collectors;
+using PerformanceMonitor.Darling.Service;
+using PerformanceMonitor.Darling.Storage;
+using Xunit;
+
+namespace Darling.Tests;
+
+///
+/// Live pins for the per-collection procedure totals (#5638, V176): the procedure_stats batch's totals are written inside
+/// the COPY's own transaction by DarlingCollectorRunner.CopyBatchOnceAsync, read back from the stored rows, so rows and
+/// totals commit together or not at all. Every batch goes through the product's own WriteBatchAsync with
+/// deltas, so the rows carry exactly the deltas and intervals a fact
+/// names (a restart collection is interval 0 everywhere). The oracle in every fact is a GROUP BY over the stored rows, the
+/// aggregate the trend's CTE takes.
+///
+/// #1776 own-store: every test here creates and drops its own scratch database through
+/// , so it is not in the live-postgres collection.
+///
+public sealed class ProcedureStatsTotalsWriterLiveTests
+{
+ private static string? BaseConnectionString => Environment.GetEnvironmentVariable("DARLING_TEST_PG");
+
+ private const string SkipReason = "Set DARLING_TEST_PG to run the procedure totals writer's live pins (each mints its own scratch database).";
+
+ private static readonly DateTime T0 = new(2026, 1, 5, 10, 0, 0, DateTimeKind.Unspecified);
+
+ private static readonly MethodInfo WriteBatchMethod = typeof(DarlingCollectorRunner)
+ .GetMethod("WriteBatchAsync", BindingFlags.NonPublic | BindingFlags.Instance)!;
+
+ private sealed class Rig : IAsyncDisposable
+ {
+ public required ScratchPostgres Scratch { get; init; }
+
+ public required NpgsqlDataSource DataSource { get; init; }
+
+ public required NpgsqlConnection Connection { get; init; }
+
+ public required DarlingCollectorRunner Runner { get; init; }
+
+ public required CollectorDeltaCalculator Deltas { get; init; }
+
+ public required ServerRuntime Server { get; init; }
+
+ public static async Task OpenAsync(int serverId, string serverName, CancellationToken ct)
+ {
+ var scratch = await ScratchPostgres.CreateAsync(BaseConnectionString!, ct);
+ var connection = new NpgsqlConnection(scratch.ConnectionString);
+ await connection.OpenAsync(ct);
+ await PgMigrations.MigrateAsync(connection, ct);
+ var deltas = new CollectorDeltaCalculator();
+ var dataSource = NpgsqlDataSource.Create(scratch.ConnectionString);
+ return new Rig
+ {
+ Scratch = scratch,
+ DataSource = dataSource,
+ Connection = connection,
+ Deltas = deltas,
+ Runner = new DarlingCollectorRunner(dataSource, deltas),
+ Server = new ServerRuntime
+ {
+ Config = new MonitoredServer { Name = serverName, Host = serverName },
+ ConnectionString = "Server=" + serverName,
+ Target = new CollectorTargetInfo { SqlMajorVersion = 16 },
+ StorageName = serverName,
+ ServerId = serverId,
+ EngineEdition = 3,
+ },
+ };
+ }
+
+ /// One batch through the product's WriteBatchAsync; a fault surfaces as the store's own exception.
+ public async Task WriteAsync(DateTime collectionTime, List rows, CancellationToken ct)
+ {
+ var context = new CollectorContext
+ {
+ ServerId = Server.ServerId,
+ ServerName = Server.StorageName,
+ CollectionTime = collectionTime,
+ Deltas = Deltas,
+ Target = Server.Target,
+ };
+
+ var task = (Task)WriteBatchMethod.MakeGenericMethod(typeof(ProcedureStatsCollector.Row)).Invoke(
+ Runner,
+ new object?[] { Connection, ProcedureStatsCollector.Instance, rows, Server, collectionTime, context, ct })!;
+ await task;
+ return (int)task.GetType().GetProperty("Result")!.GetValue(task)!;
+ }
+
+ public async ValueTask DisposeAsync()
+ {
+ await Connection.DisposeAsync();
+ await DataSource.DisposeAsync();
+ await Scratch.DisposeAsync();
+ }
+ }
+
+ /// One procedure row whose deltas are given, so the interval (0 = restart) is the fact's, not the calculator's.
+ private static ProcedureStatsCollector.Row Row(string database, string name, long elapsed, long executions, int interval) => new(
+ database, "dbo", name, "P ", T0, T0, executions, elapsed, elapsed, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0,
+ "0x" + name, "0xP" + name, null, null)
+ {
+ Precomputed = new ProcedureStatsCollector.ReadTimeDeltas(executions, elapsed, elapsed, 1, 1, 1, 0, interval),
+ };
+
+ private static async Task