Preserve reader count metadata during concurrent writes

RELEASE_NOTES.md

@@ -13,6 +13,8 @@ These notes cover changes since v4.6.5.
   primary-key ranges through pooled readers and ordinary queries.
 - Preserve reader catalog snapshots across commits and checkpoints, including
   schema metadata used by existing readers.
+- Preserve committed row-count metadata during writes and restore it after
+  publication, avoiding unnecessary full-table scans for reader `COUNT(*)` queries.
 - Reduce table-scan, LIMIT, snapshot-pager, and leaf-prefetch allocations. Forward
   intact row batches while preserving ownership and mixed row/batch traversal.
 - Streamline primary-key mutations and INSERT execution by reusing column ordinals,

docs/releases/changelog-publication.json

@@ -4,7 +4,7 @@
   "source": "https://api.github.com/repos/MaxAkbar/CSharpDB/releases",
   "notesReview": {
     "version": "4.6.6",
-    "sha256": "2927cbf938795999f8e10637e2184e87e9ed3eff6b311fd9acdb90e63b3c0225"
+    "sha256": "0c44953b85d05e17495662f40795fd2704124b9c5fbeba0d0e9fb960218c3628"
   },
   "releases": [
     {

src/CSharpDB.Engine/Database.ReaderCatalog.cs

@@ -11,14 +11,43 @@ public sealed partial class Database
     private int _catalogPublications;
     private bool _readerCatalogCopyUnsafe;
 
-    internal sealed class ReaderCatalogState(WalSnapshot snapshot, CatalogReadSnapshot? catalog)
+    internal sealed class ReaderCatalogState(WalSnapshot snapshot, CatalogReadSnapshot? catalog,
+        bool capturedFromSharedCatalog = false)
     {
         internal readonly WalSnapshot Snapshot = snapshot;
         internal CatalogReadSnapshot? Catalog = catalog;
+        internal readonly bool CapturedFromSharedCatalog = capturedFromSharedCatalog;
         private SemaphoreSlim? _loadGate;
         internal SemaphoreSlim LoadGate => LazyInitializer.EnsureInitialized(ref _loadGate, static () => new(1, 1));
     }
 
+    // The caller owns the legacy write-operation gate, before starting a transaction.
+    // Preserve committed metadata for readers admitted while the live catalog is dirty.
+    // Without this, a reader can lose exact counts and scan the whole table for COUNT(*).
+    private void CaptureReaderCatalogBeforeWrite()
+    {
+        // Databases that have never opened a reader incur no capture work.
+        if (Volatile.Read(ref _readerCatalogState) is null) return;
+        lock (_readerCatalogGate)
+        {
+            if (_catalogPublications != 0 || _readerCatalogCopyUnsafe || !_sharedStateGate.Wait(0)) return;
+            try
+            {
+                WalSnapshot snapshot = _pager.AcquireReaderSnapshot();
+                try
+                {
+                    if (!_inTransaction && _pager.CanCaptureCommittedCatalog(snapshot))
+                    {
+                        var catalog = _catalog.CaptureReadSnapshot(_readerCatalogState?.Catalog);
+                        _readerCatalogState = new ReaderCatalogState(snapshot, catalog, capturedFromSharedCatalog: true);
+                    }
+                }
+                finally { _pager.ReleaseReaderSnapshot(snapshot); }
+            }
+            finally { _sharedStateGate.Release(); }
+        }
+    }
+
     private ReaderSession CreateReaderSessionCore(IReadOnlyDictionary<string, long>? snapshotRowCounts = null)
     {
         lock (_readerCatalogGate)
@@ -27,7 +56,8 @@ private ReaderSession CreateReaderSessionCore(IReadOnlyDictionary<string, long>?
             try
             {
                 ReaderCatalogState? state = _readerCatalogState;
-                if (state is null || !state.Snapshot.SharesPageState(snapshot))
+                bool samePageState = state is not null && state.Snapshot.SharesPageState(snapshot);
+                if (!samePageState || !state!.CapturedFromSharedCatalog)
                 {
                     CatalogReadSnapshot? catalog = null;
                     // Neither gate waits: an active writer uses the snapshot-backed fallback.
@@ -48,11 +78,16 @@ private ReaderSession CreateReaderSessionCore(IReadOnlyDictionary<string, long>?
                         }
                         finally { _writeOperationGate.Release(); }
                     }
-                    state = new ReaderCatalogState(snapshot, catalog);
+                    // A fallback deliberately distrusts persisted exact counts. Retry capture
+                    // after publication finishes, even when the committed pages have not changed.
+                    // Replace the state only on success, retaining the old fallback's load gate
+                    // and catalog for its readers while a writer still prevents safe capture.
+                    if (catalog is not null || !samePageState)
+                        state = new ReaderCatalogState(snapshot, catalog, catalog is not null);
                     _readerCatalogState = state;
                 }
 
-                return new ReaderSession(this, _pager, state, _recordSerializer, snapshot,
+                return new ReaderSession(this, _pager, state!, _recordSerializer, snapshot,
                     _statementCache, _functions, _planner.AdaptiveQueryReoptimization,
                     _planner.WindowExecution, _queryObservability, _queryPlanRuntimeObserver, snapshotRowCounts);
             }

src/CSharpDB.Engine/Database.cs

@@ -540,6 +540,7 @@ private async ValueTask EnsureSharedRowVersionReservationAsync(CancellationToken
             bool transactionStarted = false;
             try
             {
+                CaptureReaderCatalogBeforeWrite();
                 await _pager.BeginTransactionAsync(ct);
                 transactionStarted = true;
                 await _catalog.PersistRowVersionHighWaterAsync(reservedHighWater, ct);
@@ -1487,6 +1488,7 @@ private async ValueTask UpgradeLegacyRowVersionAllocatorAsync(CancellationToken
         bool transactionStarted = false;
         try
         {
+            CaptureReaderCatalogBeforeWrite();
             await _pager.BeginTransactionAsync(ct);
             transactionStarted = true;
             bool regenerated = await _planner.RegenerateLegacyRowVersionTokensAsync(ct);
@@ -2190,6 +2192,7 @@ public async ValueTask BeginTransactionAsync(CancellationToken ct = default)
             await _writeOperationGate.WaitAsync(ct);
             try
             {
+                CaptureReaderCatalogBeforeWrite();
                 await _pager.BeginTransactionAsync(ct);
             }
             catch
@@ -3337,14 +3340,28 @@ private static async ValueTask DelayImplicitConflictRetryAsync(int attempt, Canc
     private async ValueTask<IDisposable> AcquireWriteOperationScopeAsync(CancellationToken ct)
     {
         if (_writeOperationGate.Wait(0))
-            return new WriteOperationScope(_writeOperationGate);
+            return CreateWriteOperationScope();
 
         // Report only a real write-admission wait. The generation-safe lease
         // restores the prior query phase on acquisition/cancellation without
         // inferring time spent by a consumer between streamed rows.
         using IDisposable? waiting = _queryObservability?.EnterWaiting();
         await _writeOperationGate.WaitAsync(ct);
-        return new WriteOperationScope(_writeOperationGate);
+        return CreateWriteOperationScope();
+    }
+
+    private IDisposable CreateWriteOperationScope()
+    {
+        try
+        {
+            CaptureReaderCatalogBeforeWrite();
+            return new WriteOperationScope(_writeOperationGate);
+        }
+        catch
+        {
+            _writeOperationGate.Release();
+            throw;
+        }
     }
 
     private void ReleaseExplicitTransactionWriteGate()