Skip locking fresh dimension rows; halve the Query Store liveness-touch rate (#4249, #4250)

Darling/Darling.Tests/PayloadDimensionLiveTests.cs

@@ -716,6 +716,146 @@ await LiveStoreCleanup.RunAsync(connectionString!, bodySucceeded, async (cleanup
         }
     }
 
+    // ── (e2) #4249: the WHERE NOT EXISTS pre-filter takes no lock on an already-fresh row ──
+
+    /// <summary>
+    /// #4249: before this fix, Postgres locked every conflicting row to evaluate the <c>DO UPDATE
+    /// ... WHERE</c> guard above, even on the digests that guard was about to keep un-updated — a
+    /// <c>Heap/LOCK</c> WAL record and a dirtied page for content that was already fresh. None of
+    /// that is visible from the SQL text alone (a source pin can grep for the <c>WHERE NOT
+    /// EXISTS</c> shape but not for whether Postgres actually took a lock), so this proves it
+    /// against a real server, three ways:
+    ///
+    /// <para>(1) A second upsert of the SAME still-fresh batch, 30 minutes later, leaves every
+    /// row's <c>xmax</c> at 0 (never locked) and advances <c>pg_current_wal_lsn()</c> by under 1
+    /// KB for a 50-row batch — effectively a read-only transaction. This is the pin: reverting the
+    /// <c>WHERE NOT EXISTS</c> pre-filter (restoring the pre-#4249 shape) turns it red. Confirmed
+    /// once by hand with a raw-SQL rehearsal of both shapes against this rig: the OLD shape left a
+    /// real transaction id (781) in <c>xmax</c> on a row whose content and <c>last_seen</c> were
+    /// both unchanged; the NEW shape left <c>xmax</c> at 0. See PR #4288.</para>
+    ///
+    /// <para>(2) A row stamped two hours ago is still refreshed — the pre-filter's own staleness
+    /// read uses the same one-hour boundary the <c>ON CONFLICT ... WHERE</c> guard always used, so
+    /// a genuinely stale row still reaches the <c>UPDATE</c>.</para>
+    ///
+    /// <para>(3) A digest with no existing row at all is still inserted — the pre-filter is an
+    /// anti-join over existing rows, not a blanket skip of the statement.</para>
+    /// </summary>
+    [Fact]
+    public async Task PreFilter_SkipsLockingFreshRows_ButStillRefreshesStaleOnes_AndInsertsNewDigests()
+    {
+        var connectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG");
+        Assert.SkipWhen(string.IsNullOrEmpty(connectionString), SkipReason);
+
+        var ct = TestContext.Current.CancellationToken;
+        var runId = Guid.NewGuid().ToString("N");
+
+        var freshPairs = Enumerable.Range(0, 50)
+            .Select(i =>
+            {
+                var payload = $"<ShowPlanXML prefilter=\"{runId}\" fresh=\"{i}\"/>";
+                return (Digest: PayloadDimensions.Digest(payload), Payload: payload);
+            })
+            .ToArray();
+        var stalePayload = $"<ShowPlanXML prefilter=\"{runId}\" stale=\"0\"/>";
+        var staleDigest = PayloadDimensions.Digest(stalePayload);
+        var newPayload = $"<ShowPlanXML prefilter=\"{runId}\" brandnew=\"0\"/>";
+        var newDigest = PayloadDimensions.Digest(newPayload);
+        var freshDigests = freshPairs.Select(p => p.Digest).ToArray();
+        var everyDigestWritten = freshDigests.Append(staleDigest).Append(newDigest).ToArray();
+
+        var t0 = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Unspecified);
+
+        await using var connection = await OpenMigratedStoreAsync(connectionString!, ct);
+        var bodySucceeded = false;
+        try
+        {
+            async Task FlushAsync(IEnumerable<(byte[] Digest, string Payload)> pairs, DateTime collectionTime)
+            {
+                var batch = new PayloadDimensionBatch();
+                foreach (var (digest, payload) in pairs)
+                {
+                    batch.Add(PayloadDimensions.QueryPlanDimTable, digest, payload);
+                }
+
+                await using var transaction = await connection.BeginTransactionAsync(ct);
+                await PayloadDimensionWriter.FlushAsync(connection, transaction, batch, collectionTime, ct);
+                await transaction.CommitAsync(ct);
+            }
+
+            /* A dedicated ANY($1) command rather than the file's params-object ScalarAsync helper:
+               a byte[][] argument there is covariantly assignable to object[], so C# spreads it
+               across the params array (one digest per parameter) instead of binding it as ONE
+               bytea[] array parameter. Explicit typing sidesteps that entirely. */
+            async Task<long> LockedRowCountAsync(byte[][] digests)
+            {
+                await using var command = new NpgsqlCommand(
+                    "SELECT count(*) FROM query_plan_dim WHERE digest = ANY($1) AND xmax <> 0", connection);
+                command.Parameters.Add(new NpgsqlParameter
+                { NpgsqlDbType = NpgsqlDbType.Array | NpgsqlDbType.Bytea, Value = digests });
+                return (long)(await command.ExecuteScalarAsync(ct))!;
+            }
+
+            async Task<long> ChangedLastSeenCountAsync(byte[][] digests, DateTime expected)
+            {
+                await using var command = new NpgsqlCommand(
+                    "SELECT count(*) FROM query_plan_dim WHERE digest = ANY($1) AND last_seen <> $2", connection);
+                command.Parameters.Add(new NpgsqlParameter
+                { NpgsqlDbType = NpgsqlDbType.Array | NpgsqlDbType.Bytea, Value = digests });
+                command.Parameters.Add(new NpgsqlParameter
+                { NpgsqlDbType = NpgsqlDbType.Timestamp, Value = expected });
+                return (long)(await command.ExecuteScalarAsync(ct))!;
+            }
+
+            async Task<string> CurrentWalLsnAsync()
+                => (string)(await ScalarAsync(connection, "SELECT pg_current_wal_lsn()::text", ct))!;
+
+            /* pg_wal_lsn_diff returns numeric, which Npgsql maps to decimal, not a bigint type. */
+            async Task<long> WalBytesSinceAsync(string beforeLsn)
+                => (long)(decimal)(await ScalarAsync(
+                    connection, "SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), $1::text::pg_lsn)", ct, beforeLsn))!;
+
+            // The initial collection cycle: 50 "hot" plans plus one that will go stale.
+            await FlushAsync(freshPairs.Append((staleDigest, stalePayload)), t0);
+            Assert.Equal(0, await LockedRowCountAsync(freshDigests));
+
+            // (1) Same batch, same digests, 30 minutes later -- still inside the one-hour
+            // freshness window. The pre-filter excludes every one of them before the statement
+            // ever reaches INSERT/ON CONFLICT: no lock taken, no last_seen change, near-zero WAL.
+            var lsnBefore = await CurrentWalLsnAsync();
+            await FlushAsync(freshPairs, t0.AddMinutes(30));
+            var walBytes = await WalBytesSinceAsync(lsnBefore);
+
+            Assert.Equal(0, await LockedRowCountAsync(freshDigests));
+            Assert.Equal(0, await ChangedLastSeenCountAsync(freshDigests, t0));
+            Assert.True(walBytes < 1024, $"expected a near-zero WAL delta for an all-fresh batch, saw {walBytes} bytes");
+
+            // (2) and (3): two hours after t0, the stale row is refreshed and a brand-new digest
+            // is inserted, in the same flush.
+            await FlushAsync([(staleDigest, stalePayload), (newDigest, newPayload)], t0.AddHours(2));
+
+            var refreshedStale = (DateTime)(await ScalarAsync(
+                connection, "SELECT last_seen FROM query_plan_dim WHERE digest = $1", ct, staleDigest))!;
+            Assert.Equal(t0.AddHours(2), refreshedStale);
+
+            var insertedNew = (DateTime)(await ScalarAsync(
+                connection, "SELECT last_seen FROM query_plan_dim WHERE digest = $1", ct, newDigest))!;
+            Assert.Equal(t0.AddHours(2), insertedNew);
+
+            bodySucceeded = true;
+        }
+        finally
+        {
+            await LiveStoreCleanup.RunAsync(connectionString!, bodySucceeded, async (cleanup, cleanupCt) =>
+            {
+                foreach (var digest in everyDigestWritten)
+                {
+                    await DeleteDimRowAsync(cleanup, PayloadDimensions.QueryPlanDimTable, digest, cleanupCt);
+                }
+            });
+        }
+    }
+
     // ── (f) the GC ──
 
     [Fact]

Darling/Darling.Tests/PayloadDimensionTests.cs

@@ -445,17 +445,37 @@ public void UpsertSql_IsOneUnnestStatement_WithTheOnConflictStalenessGuard()
             "INSERT INTO query_text_dim (digest, query_text, last_seen)\n" +
             "SELECT u.digest, u.payload, $3\n" +
             "FROM unnest($1::bytea[], $2::text[]) AS u(digest, payload)\n" +
+            "WHERE NOT EXISTS (\n" +
+            "    SELECT 1 FROM query_text_dim d\n" +
+            "    WHERE d.digest = u.digest\n" +
+            "    AND   d.last_seen >= $3 - INTERVAL '1 hour')\n" +
+            "ORDER BY u.digest\n" +
             "ON CONFLICT (digest) DO UPDATE SET last_seen = EXCLUDED.last_seen\n" +
             "WHERE query_text_dim.last_seen < EXCLUDED.last_seen - INTERVAL '1 hour'",
             sql);
 
+        /* #4249: the pre-filter is a plain MVCC read (no lock) that keeps an already-fresh digest from
+           ever reaching INSERT or ON CONFLICT — that is what stops the write, not the conflict guard
+           above, which only ever stopped the UPDATE. */
+        Assert.Contains("WHERE NOT EXISTS (", sql, StringComparison.Ordinal);
+
+        /* #4249: the anti-join gives the planner a row source it can reorder (a hash anti-join returns
+           survivors in hash order), so client-side digest ordering alone no longer fixes the order rows
+           reach ON CONFLICT. Both branches must carry this. */
+        Assert.Contains("ORDER BY u.digest", sql, StringComparison.Ordinal);
+
         /* The plan dimension upserts gzip BYTES into query_plan_gz (#2069) — same statement shape,
            same conflict semantics, bytea payload array instead of text. The text column is never
            written on new rows; pre-V54 rows keep theirs until the GC retires them. */
         Assert.Equal(
             "INSERT INTO query_plan_dim (digest, query_plan_gz, last_seen)\n" +
             "SELECT u.digest, u.payload, $3\n" +
             "FROM unnest($1::bytea[], $2::bytea[]) AS u(digest, payload)\n" +
+            "WHERE NOT EXISTS (\n" +
+            "    SELECT 1 FROM query_plan_dim d\n" +
+            "    WHERE d.digest = u.digest\n" +
+            "    AND   d.last_seen >= $3 - INTERVAL '1 hour')\n" +
+            "ORDER BY u.digest\n" +
             "ON CONFLICT (digest) DO UPDATE SET last_seen = EXCLUDED.last_seen\n" +
             "WHERE query_plan_dim.last_seen < EXCLUDED.last_seen - INTERVAL '1 hour'",
             PayloadDimensions.UpsertSql(PayloadDimensions.QueryPlanDimTable));
@@ -467,6 +487,11 @@ public void UpsertSql_IsOneUnnestStatement_WithTheOnConflictStalenessGuard()
             "INSERT INTO query_plan_dim (digest, query_plan_xml, last_seen)\n" +
             "SELECT u.digest, u.payload, $3\n" +
             "FROM unnest($1::bytea[], $2::text[]) AS u(digest, payload)\n" +
+            "WHERE NOT EXISTS (\n" +
+            "    SELECT 1 FROM query_plan_dim d\n" +
+            "    WHERE d.digest = u.digest\n" +
+            "    AND   d.last_seen >= $3 - INTERVAL '1 hour')\n" +
+            "ORDER BY u.digest\n" +
             "ON CONFLICT (digest) DO UPDATE SET last_seen = EXCLUDED.last_seen\n" +
             "WHERE query_plan_dim.last_seen < EXCLUDED.last_seen - INTERVAL '1 hour'",
             PayloadDimensions.UpsertSql(PayloadDimensions.QueryPlanDimTable, compressContent: false));
@@ -479,6 +504,30 @@ public void UpsertSql_IsOneUnnestStatement_WithTheOnConflictStalenessGuard()
         Assert.Throws<ArgumentOutOfRangeException>(() => PayloadDimensions.UpsertSql("not_a_dim"));
     }
 
+    /// <summary>
+    /// #4249's source pin: reads <c>PayloadDimensions.cs</c> itself and requires BOTH branches of
+    /// <c>UpsertSql</c> — the compressed-content early return and the shared text/plan-xml tail — to carry
+    /// <c>ORDER BY u.digest</c>. The exact-string assertions above already cover the shipped shape, but they
+    /// pin the STRING; this pins the SOURCE, so a future edit that special-cases one branch and drops the
+    /// other's ORDER BY fails here even if nobody updates the string pins to match.
+    /// </summary>
+    [Fact]
+    public void UpsertSql_SourceCarriesOrderByOnBothBranches()
+    {
+        var root = FindRepoRoot();
+        Assert.True(root is not null, RepoRootNotFound);
+
+        var source = File.ReadAllText(Path.Combine(
+            root!, "Darling", "PerformanceMonitor.Darling.Storage", "PayloadDimensions.cs"));
+
+        var body = MethodBody(source, "public static string UpsertSql(string dimTable, bool compressContent = true)");
+        Assert.False(string.IsNullOrEmpty(body),
+            "could not locate UpsertSql -- this guard must fail rather than silently pass on a parse miss");
+
+        var occurrences = System.Text.RegularExpressions.Regex.Matches(body, "ORDER BY u.digest").Count;
+        Assert.Equal(2, occurrences);
+    }
+
     // ── the gzip codec (#2069) ──
 
     [Fact]

Darling/Darling.Tests/QueryStoreFetchProbeLivePostgresTests.cs

@@ -147,6 +147,102 @@ await LiveStoreCleanup.RunAsync(cs!, bodySucceeded, async (cleanup, cleanupCt) =
         }
     }
 
+    /// <summary>
+    /// #4250: the touch guard's freshness gate is the <c>last_seen</c> stored in the map, dimension and
+    /// text rows themselves — nothing about it lives in the writer's memory, because this design has none;
+    /// <see cref="QueryStoreFetchProbe"/> is static and every call is connection-scoped. So a process
+    /// restart — the nearest thing a live rig can rehearse is a brand-new connection, since there is no
+    /// other per-process state to reset — must not re-touch rows the guard already stamped fresh: probing
+    /// the SAME references again from a DIFFERENT connection, still inside the guard window, updates zero
+    /// rows anywhere.
+    /// </summary>
+    [Fact]
+    public async Task Restart_ANewConnectionRetouchesNothingFresh_TheGuardReadsStoredLastSeen()
+    {
+        var cs = ConnectionString;
+        Assert.SkipWhen(string.IsNullOrEmpty(cs), "Set DARLING_TEST_PG to a Postgres connection string to run the live fetch-probe test.");
+
+        var ct = TestContext.Current.CancellationToken;
+        using var connection = new NpgsqlConnection(cs);
+        await connection.OpenAsync(ct);
+        await PgMigrations.MigrateAsync(connection, ct);
+        await DeleteRowsAsync(connection, ct);
+
+        var bodySucceeded = false;
+        try
+        {
+            /* Older than the guard, so the FIRST touch below actually fires (same reasoning as the sibling
+               test above: a hard-coded age would silently stop exercising the touch once the width moves). */
+            var landedAt = DateTime.UtcNow.AddHours(-(QueryStoreLivenessTouchGuard.GuardHours + 1));
+
+            await QueryStorePlanWriter.WriteAsync(
+                connection, ServerId, Db,
+                new[] { new FetchedPlan(201, "<plan restart/>", "0xR1") },
+                landedAt, TestTimeoutSeconds, ct);
+            await QueryStoreTextWriter.WriteAsync(
+                connection, ServerId, Db,
+                new[] { new FetchedQueryText(202, "SELECT 'restart'", "0xR2") },
+                landedAt, TestTimeoutSeconds, ct);
+
+            async Task<(long Map, long Dim, long Text)> FreshnessCountsAsync(DateTime stamp)
+            {
+                using var freshness = new NpgsqlCommand(@"
+SELECT
+    (SELECT COUNT(*) FROM collect.query_store_plan_map
+     WHERE server_id = $1 AND database_name = $2 AND last_seen = $3),
+    (SELECT COUNT(*) FROM collect.query_plan_dim d
+     JOIN collect.query_store_plan_map m ON m.digest = d.digest
+     WHERE m.server_id = $1 AND m.database_name = $2 AND d.last_seen = $3),
+    (SELECT COUNT(*) FROM collect.query_store_text
+     WHERE server_id = $1 AND database_name = $2 AND last_seen = $3)", connection);
+                freshness.Parameters.AddWithValue(ServerId);
+                freshness.Parameters.AddWithValue(Db);
+                freshness.Parameters.AddWithValue(stamp);
+                await using var reader = await freshness.ExecuteReaderAsync(ct);
+                await reader.ReadAsync(ct);
+                return (reader.GetInt64(0), reader.GetInt64(1), reader.GetInt64(2));
+            }
+
+            // The first touch, from the original connection: the rows landed older than the guard, so
+            // this fires and stamps all three relations with firstTouch.
+            var firstTouch = DateTime.UtcNow;
+            await QueryStoreFetchProbe.TouchAndProbePlansAsync(
+                connection, ServerId, Db, new[] { (201L, (string?)"0xR1") }, firstTouch, TestTimeoutSeconds, ct);
+            await QueryStoreFetchProbe.TouchAndProbeTextsAsync(
+                connection, ServerId, Db, new[] { (202L, (string?)"0xR2") }, firstTouch, TestTimeoutSeconds, ct);
+
+            var firstTouchStamp = QueryStorePlanMap.Naive(firstTouch);
+            Assert.Equal((1L, 1L, 1L), await FreshnessCountsAsync(firstTouchStamp));
+
+            // "Restart": a brand-new connection. Same references, five minutes later — still well inside
+            // the guard window, so nothing here should qualify for a re-touch.
+            using var freshConnection = new NpgsqlConnection(cs);
+            await freshConnection.OpenAsync(ct);
+            var secondTouch = firstTouch.AddMinutes(5);
+
+            var planVerdicts = await QueryStoreFetchProbe.TouchAndProbePlansAsync(
+                freshConnection, ServerId, Db, new[] { (201L, (string?)"0xR1") }, secondTouch, TestTimeoutSeconds, ct);
+            var textVerdicts = await QueryStoreFetchProbe.TouchAndProbeTextsAsync(
+                freshConnection, ServerId, Db, new[] { (202L, (string?)"0xR2") }, secondTouch, TestTimeoutSeconds, ct);
+
+            // Zero rows updated: every row still carries the FIRST touch's stamp, none carry the second.
+            Assert.Equal((1L, 1L, 1L), await FreshnessCountsAsync(firstTouchStamp));
+            Assert.Equal((0L, 0L, 0L), await FreshnessCountsAsync(QueryStorePlanMap.Naive(secondTouch)));
+
+            // The fresh connection's verdicts still resolve correctly — the restart cost nothing but the
+            // write it correctly skipped.
+            Assert.Equal(new FetchProbeVerdict(201, Resolved: true, HashStale: false), planVerdicts.Single());
+            Assert.Equal(new FetchProbeVerdict(202, Resolved: true, HashStale: false), textVerdicts.Single());
+
+            bodySucceeded = true;
+        }
+        finally
+        {
+            await LiveStoreCleanup.RunAsync(cs!, bodySucceeded, async (cleanup, cleanupCt) =>
+                await DeleteRowsAsync(cleanup, cleanupCt));
+        }
+    }
+
     private static async Task DeleteRowsAsync(NpgsqlConnection connection, CancellationToken ct)
     {
         var sql =