From d1c23f800d31c29b92a6f872262382c662e11453 Mon Sep 17 00:00:00 2001 From: Maximum Code Date: Sun, 11 Oct 2026 20:47:02 -0700 Subject: [PATCH] Preserve reader count metadata during concurrent writes --- RELEASE_NOTES.md | 2 + docs/releases/changelog-publication.json | 2 +- src/CSharpDB.Engine/Database.ReaderCatalog.cs | 43 ++- src/CSharpDB.Engine/Database.cs | 21 +- .../ReaderCatalogWriteAdmissionTests.cs | 314 ++++++++++++++++++ www/changelog.html | 2 +- 6 files changed, 376 insertions(+), 8 deletions(-) create mode 100644 tests/CSharpDB.Tests/ReaderCatalogWriteAdmissionTests.cs diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 7aaf4920..fc196c18 100644 --- a/RELEASE_NOTES.md +++ b/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, diff --git a/docs/releases/changelog-publication.json b/docs/releases/changelog-publication.json index 941ce1ec..7880545c 100644 --- a/docs/releases/changelog-publication.json +++ b/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": [ { diff --git a/src/CSharpDB.Engine/Database.ReaderCatalog.cs b/src/CSharpDB.Engine/Database.ReaderCatalog.cs index bc75b8cc..13551b36 100644 --- a/src/CSharpDB.Engine/Database.ReaderCatalog.cs +++ b/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? snapshotRowCounts = null) { lock (_readerCatalogGate) @@ -27,7 +56,8 @@ private ReaderSession CreateReaderSessionCore(IReadOnlyDictionary? 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? } 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); } diff --git a/src/CSharpDB.Engine/Database.cs b/src/CSharpDB.Engine/Database.cs index 68e6871a..322ff10f 100644 --- a/src/CSharpDB.Engine/Database.cs +++ b/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 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() diff --git a/tests/CSharpDB.Tests/ReaderCatalogWriteAdmissionTests.cs b/tests/CSharpDB.Tests/ReaderCatalogWriteAdmissionTests.cs new file mode 100644 index 00000000..615060f6 --- /dev/null +++ b/tests/CSharpDB.Tests/ReaderCatalogWriteAdmissionTests.cs @@ -0,0 +1,314 @@ +using System.Reflection; +using CSharpDB.Engine; +using CSharpDB.Primitives; +using CSharpDB.Storage.Paging; +using CSharpDB.Storage.StorageEngine; + +namespace CSharpDB.Tests; + +public sealed class ReaderCatalogWriteAdmissionTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + [Theory(Timeout = 30_000)] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task WarmReadersDuringNextLegacyWrite_KeepCommittedCountWithoutScanning(bool fileBacked, bool commit) + { + CancellationToken ct = TestContext.Current.CancellationToken; + string path = Path.Combine(Path.GetTempPath(), $"csharpdb_reader_admission_{Guid.NewGuid():N}.db"); + try + { + await using var db = fileBacked + ? await Database.OpenAsync(path, ct) + : await Database.OpenInMemoryAsync(ct); + await SeedAsync(db); + using (var warm = db.CreateReaderSession()) + await AssertCountAsync(warm, 128); + // Advance the committed pages without opening another reader. The cached + // catalog must be refreshed before the next writer can mutate live metadata. + await db.ExecuteAsync("INSERT INTO t VALUES (128, 'committed')", Ct); + await db.BeginTransactionAsync(Ct); + try + { + await db.ExecuteAsync("ALTER TABLE t ADD COLUMN pending TEXT", Ct); + await db.ExecuteAsync("INSERT INTO t VALUES (129, 'pending', 'uncommitted')", Ct); + using var reader = db.CreateReaderSession(); + AssertMetadataCount(reader, 129); + await AssertCountAsync(reader, 129); + Assert.Null(Field(reader, "_snapshotPager")); + if (commit) + await db.CommitAsync(Ct); + else + await db.RollbackAsync(Ct); + await AssertCountAsync(reader, 129); + await using var rows = await reader.ExecuteReadAsync("SELECT * FROM t WHERE id = 128", Ct); + Assert.Equal(new[] { "id", "label" }, rows.Schema.Select(column => column.Name)); + Assert.Equal("committed", Assert.Single(await rows.ToListAsync(Ct))[1].AsText); + } + finally + { + if (Field(db, "_inTransaction")) await db.RollbackAsync(Ct); + } + using var latest = db.CreateReaderSession(); + await AssertCountAsync(latest, commit ? 130 : 129); + } + finally + { + foreach (string file in new[] { path, path + ".wal", path + ".lock" }) + if (File.Exists(file)) File.Delete(file); + } + } + + [Fact(Timeout = 30_000)] + public async Task WarmReadersDuringImplicitInsert_KeepCommittedCountWithoutScanning() + { + CancellationToken ct = TestContext.Current.CancellationToken; + var interceptor = new PausedCommitInterceptor(); + var options = new DatabaseOptions + { + StorageEngineOptions = new StorageEngineOptions + { + PagerOptions = new PagerOptions { Interceptors = [interceptor] }, + }, + }; + await using var db = await Database.OpenInMemoryAsync(options, ct); + await SeedAsync(db); + using (var warm = db.CreateReaderSession()) await AssertCountAsync(warm, 128); + await db.ExecuteAsync("INSERT INTO t VALUES (128, 'committed')", Ct); + interceptor.Arm(); + Task writer = db.ExecuteAsync("INSERT INTO t VALUES (129, 'pending')", Ct).AsTask(); + try + { + await interceptor.Started.Task.WaitAsync(TimeSpan.FromSeconds(10), Ct); + using var reader = db.CreateReaderSession(); + AssertMetadataCount(reader, 129); + await AssertCountAsync(reader, 129); + Assert.Null(Field(reader, "_snapshotPager")); + interceptor.Release.TrySetResult(); + await writer.WaitAsync(TimeSpan.FromSeconds(10), Ct); + await AssertCountAsync(reader, 129); + } + finally + { + interceptor.Release.TrySetResult(); + await writer.WaitAsync(TimeSpan.FromSeconds(10), Ct); + } + using var latest = db.CreateReaderSession(); + await AssertCountAsync(latest, 130); + } + + [Fact(Timeout = 30_000)] + public async Task CancelledWriteAdmission_ReleasesNoUnownedGates_AndNextWriteProgresses() + { + CancellationToken ct = TestContext.Current.CancellationToken; + await using var db = await Database.OpenInMemoryAsync(ct); + await SeedAsync(db); + using (var warm = db.CreateReaderSession()) await AssertCountAsync(warm, 128); + var writeGate = Field(db, "_writeOperationGate"); + await writeGate.WaitAsync(ct); + using var cancelled = CancellationTokenSource.CreateLinkedTokenSource(ct); + try + { + Task begin = db.BeginTransactionAsync(cancelled.Token).AsTask(); + Assert.False(begin.IsCompleted); + cancelled.Cancel(); + await Assert.ThrowsAnyAsync(() => begin); + Assert.Equal(0, writeGate.CurrentCount); + } + finally { writeGate.Release(); } + await db.ExecuteAsync("INSERT INTO t VALUES (128, 'after cancellation')", ct); + Assert.Equal(1, writeGate.CurrentCount); + Assert.Equal(1, Field(db, "_sharedStateGate").CurrentCount); + Assert.Equal(0, db.ActiveReaderCount); + using var latest = db.CreateReaderSession(); + await AssertCountAsync(latest, 129); + } + + [Fact(Timeout = 30_000)] + public async Task BusySharedCatalog_FallsBackWithoutWaiting_AndCancelledLoadCanRetry() + { + CancellationToken ct = TestContext.Current.CancellationToken; + await using var db = await Database.OpenInMemoryAsync(ct); + await SeedAsync(db); + using (var warm = db.CreateReaderSession()) await AssertCountAsync(warm, 128); + await db.ExecuteAsync("INSERT INTO t VALUES (128, 'committed')", ct); + var sharedGate = Field(db, "_sharedStateGate"); + await sharedGate.WaitAsync(ct); + try + { + // Capture must skip a busy shared gate, even while it owns the writer gate. + await db.BeginTransactionAsync(ct).AsTask().WaitAsync(TimeSpan.FromSeconds(10), ct); + await db.ExecuteAsync("INSERT INTO t VALUES (129, 'pending')", ct); + using var reader = db.CreateReaderSession(); + var state = Field(reader, "_catalogState"); + Assert.Null(state.Catalog); + await state.LoadGate.WaitAsync(ct); + using var cancelled = CancellationTokenSource.CreateLinkedTokenSource(ct); + try + { + Task query = reader.ExecuteReadAsync("SELECT COUNT(*) FROM t", cancelled.Token).AsTask(); + Assert.False(query.IsCompleted); + cancelled.Cancel(); + await Assert.ThrowsAnyAsync(() => query); + Assert.Equal(0, state.LoadGate.CurrentCount); + } + finally { state.LoadGate.Release(); } + await AssertCountAsync(reader, 129); + Assert.Equal(1, state.LoadGate.CurrentCount); + await db.RollbackAsync(ct); + } + finally + { + sharedGate.Release(); + if (Field(db, "_inTransaction")) await db.RollbackAsync(ct); + } + await db.ExecuteAsync("INSERT INTO t VALUES (129, 'after fallback')", ct); + Assert.Equal(0, db.ActiveReaderCount); + using var latest = db.CreateReaderSession(); + await AssertCountAsync(latest, 130); + } + + private static async Task SeedAsync(Database db) + { + await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, label TEXT)", Ct); + await db.BeginTransactionAsync(Ct); + for (int i = 0; i < 128; i++) await db.ExecuteAsync($"INSERT INTO t VALUES ({i}, 'original')", Ct); + await db.CommitAsync(Ct); + } + + [Fact(Timeout = 30_000)] + public async Task ReadersAfterPublication_RecaptureExactCountForTheSameCommittedPages() + { + CancellationToken ct = TestContext.Current.CancellationToken; + var interceptor = new PausedCommitInterceptor(atEnd: true); + var options = new DatabaseOptions + { + StorageEngineOptions = new StorageEngineOptions + { + PagerOptions = new PagerOptions { Interceptors = [interceptor] }, + }, + }; + await using var db = await Database.OpenInMemoryAsync(options, ct); + await SeedAsync(db); + using (var warm = db.CreateReaderSession()) await AssertCountAsync(warm, 128); + interceptor.Arm(); + Task writer = db.ExecuteAsync("INSERT INTO t VALUES (128, 'published')", ct).AsTask(); + try + { + await interceptor.Started.Task.WaitAsync(TimeSpan.FromSeconds(10), ct); + using var during = db.CreateReaderSession(); + var fallback = Field(during, "_catalogState"); + Assert.Null(fallback.Catalog); + await AssertCountAsync(during, 129); + Assert.False(fallback.Catalog!.TryGetExactTableRowCount("t", out _)); + interceptor.Release.TrySetResult(); + await writer.WaitAsync(TimeSpan.FromSeconds(10), ct); + using var after = db.CreateReaderSession(); + var captured = Field(after, "_catalogState"); + Assert.True(fallback.Snapshot.SharesPageState(captured.Snapshot)); + AssertMetadataCount(after, 129); + await AssertCountAsync(after, 129); + Assert.Null(Field(after, "_snapshotPager")); + // Recapture must not mutate a fallback still used by an older reader. + Assert.Same(fallback, Field(during, "_catalogState")); + await db.ExecuteAsync("INSERT INTO t VALUES (129, 'later')", ct); + await AssertCountAsync(during, 129); + await AssertCountAsync(after, 129); + } + finally + { + interceptor.Release.TrySetResult(); + await writer.WaitAsync(TimeSpan.FromSeconds(10), ct); + } + } + + [Fact(Timeout = 30_000)] + public async Task FailedNativePublication_DoesNotRecaptureUnsynchronizedCounts_AndWritesProgress() + { + CancellationToken ct = TestContext.Current.CancellationToken; + var interceptor = new PausedCommitInterceptor(atEnd: true, failAtEnd: true); + var options = new DatabaseOptions + { + StorageEngineOptions = new StorageEngineOptions + { + PagerOptions = new PagerOptions { Interceptors = [interceptor] }, + }, + }; + await using var db = await Database.OpenInMemoryAsync(options, ct); + await SeedAsync(db); + using (var warm = db.CreateReaderSession()) await AssertCountAsync(warm, 128); + await using (var writer = await db.BeginWriteTransactionAsync(ct)) + { + await writer.ExecuteAsync("INSERT INTO t VALUES (128, 'durably published')", ct); + interceptor.Arm(); + await Assert.ThrowsAsync(() => writer.CommitAsync(ct).AsTask()); + } + // The data was published, but the failed completion bypassed shared-count sync. + // Both recapture paths must retain the conservative, snapshot-backed fallback. + using (var reader = db.CreateReaderSession()) + { + Assert.Null(Field(reader, "_catalogState").Catalog); + await AssertCountAsync(reader, 129); + } + await db.ExecuteAsync("INSERT INTO t VALUES (129, 'after failure')", ct); + Assert.Equal(0, db.ActiveReaderCount); + Assert.Equal(1, Field(db, "_writeOperationGate").CurrentCount); + Assert.Equal(1, Field(db, "_sharedStateGate").CurrentCount); + using var latest = db.CreateReaderSession(); + Assert.Null(Field(latest, "_catalogState").Catalog); + await AssertCountAsync(latest, 130); + } + + private static void AssertMetadataCount(Database.ReaderSession reader, long expected) + { + var state = Field(reader, "_catalogState"); + Assert.NotNull(state.Catalog); + Assert.True(state.Catalog.TryGetExactTableRowCount("t", out long count)); + Assert.Equal(expected, count); + } + + private static async Task AssertCountAsync(Database.ReaderSession reader, long expected) + { + await using var result = await reader.ExecuteReadAsync("SELECT COUNT(*) FROM t", Ct); + Assert.Equal(expected, Assert.Single(Assert.Single(await result.ToListAsync(Ct))).AsInteger); + } + + private static T Field(object target, string name) + => (T)target.GetType().GetField(name, BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(target)!; + + private sealed class PausedCommitInterceptor(bool atEnd = false, bool failAtEnd = false) : IPageOperationInterceptor + { + internal readonly TaskCompletionSource Started = new(TaskCreationOptions.RunContinuationsAsynchronously); + internal readonly TaskCompletionSource Release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private bool _armed; + internal void Arm() => _armed = true; + public ValueTask OnCommitStartAsync(int dirtyPageCount, CancellationToken ct = default) + { + if (atEnd || !_armed) return ValueTask.CompletedTask; + Started.TrySetResult(); + return new(Release.Task.WaitAsync(ct)); + } + public ValueTask OnBeforeReadAsync(uint pageId, CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnAfterReadAsync(uint pageId, PageReadSource source, CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnBeforeWriteAsync(uint pageId, CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnAfterWriteAsync(uint pageId, bool succeeded, CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnCommitEndAsync(int dirtyPageCount, bool succeeded, CancellationToken ct = default) + { + if (!atEnd || !_armed || !succeeded) return ValueTask.CompletedTask; + if (failAtEnd) + { + _armed = false; + throw new IOException("Injected failure after durable publication and before catalog sync."); + } + Started.TrySetResult(); + return new(Release.Task.WaitAsync(ct)); + } + public ValueTask OnCheckpointStartAsync(int committedFrameCount, CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnCheckpointEndAsync(int committedFrameCount, bool succeeded, CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnRecoveryStartAsync(CancellationToken ct = default) => ValueTask.CompletedTask; + public ValueTask OnRecoveryEndAsync(bool succeeded, CancellationToken ct = default) => ValueTask.CompletedTask; + } +} diff --git a/www/changelog.html b/www/changelog.html index 3a4d62b2..d1759562 100644 --- a/www/changelog.html +++ b/www/changelog.html @@ -84,7 +84,7 @@

Changelog

Prepared queries, bounded reads, and durable writes

Prepared for v4.6.6, based on changes since v4.6.5. This candidate is not yet published; stable downloads remain on v4.6.5.

    -
  • Queries: reuse prepared SELECT plans, bound integer primary-key ranges, and preserve reader catalog snapshots across commits and checkpoints.
    • +
  • Queries: reuse prepared SELECT plans, bound integer primary-key ranges, preserve reader catalog snapshots, and avoid unnecessary COUNT(*) scans during concurrent writes.
  • Execution: reduce scan, LIMIT, prefetch, and INSERT allocations while preserving row and storage-page ownership.
  • Durable writes: reduce checkpoint stalls and improve WAL lifecycle handling.
  • Storage correctness: prevent checkpoint reset and compaction from invalidating concurrent write-conflict WAL reads, and protect pending durable commit publication.