From 4b91c890899f6f6df96918c4f7b7b732b9660900 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 13 Jul 2026 19:28:14 -0500 Subject: [PATCH 1/2] Bulk event insert applies the schema once per database, not per call (GH-4946) The batch BulkInsertEventsAsync overloads began with Storage.ApplyAllConfiguredChangesToDatabaseAsync() on every call: a full schema delta -- partition introspection plus information_schema sweeps -- across every database in the store, per batch. A conversion tool importing in 1,000-event batches paid that per batch; measured in a 512-database sharded store it collapsed throughput to ~17 events/s and parked the connection pool on Weasel's partition introspection query. The streaming overload has no such call and is fine, which is the tell that the apply is a leftover, not part of the contract. The apply is now: - skipped entirely when the effective AutoCreate is None, where it is a no-op by contract and only the introspection cost remains, and - otherwise run at most once per *database*, and only for the database being imported into rather than every database in the store. Concurrent batches are safe: the per-database entry is a Lazy with ExecutionAndPublication, so racing callers await a single in-flight apply. A failed apply is evicted so a later call retries instead of caching the failure. Reported by @erdtsieck with production measurements. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/events/bulk-appending.md | 10 ++ ...k_insert_does_not_apply_schema_per_call.cs | 116 ++++++++++++++++++ src/Marten/DocumentStore.cs | 71 ++++++++++- 3 files changed, 193 insertions(+), 4 deletions(-) create mode 100644 src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs diff --git a/docs/events/bulk-appending.md b/docs/events/bulk-appending.md index 977cc0a328..05732c51c9 100644 --- a/docs/events/bulk-appending.md +++ b/docs/events/bulk-appending.md @@ -25,6 +25,16 @@ The bulk append API: This approach is significantly faster than the normal append path because it avoids per-row function calls, version checking, and individual INSERT statements. +### Schema Application + +`BulkInsertEventsAsync` will apply any outstanding schema changes **at most once per database**, and only +for the database the events are actually being imported into. It is *not* re-applied on every call, so a +conversion tool that loops batches over a single store pays the schema introspection once rather than per +batch. When the store's `AutoCreateSchemaObjects` is `AutoCreate.None` — schema managed out of band — the +apply is skipped entirely, since it cannot do anything by contract and would only cost introspection. + +As always with `AutoCreate.None`, the event storage must exist before you import. + ## Basic Usage Build a list of `StreamAction` objects representing new event streams, then call `BulkInsertEventsAsync` diff --git a/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs b/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs new file mode 100644 index 0000000000..4c0c43e699 --- /dev/null +++ b/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs @@ -0,0 +1,116 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using JasperFx; +using JasperFx.Events; +using Marten; +using Marten.Events; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace EventSourcingTests.Bugs; + +/// +/// GH-4946: the batch BulkInsertEventsAsync overloads ran +/// Storage.ApplyAllConfiguredChangesToDatabaseAsync() on *every* call, so a conversion tool +/// importing in 1,000-event batches paid a full schema delta — partition introspection plus +/// information_schema sweeps, across every database in the store — per batch. The regression is +/// *how many times* the apply runs, so that is what these specs assert. +/// +public class Bug_4946_bulk_insert_does_not_apply_schema_per_call: OneOffConfigurationsContext +{ + private static List streams(EventGraph events, int count) + { + var actions = new List(); + for (var i = 0; i < count; i++) + { + actions.Add(StreamAction.Start(events, Guid.NewGuid(), new object[] + { + new QuestStarted { Name = $"Quest {i}" }, new QuestEnded { Name = $"Quest {i}" } + })); + } + + return actions; + } + + [Fact] + public async Task apply_the_schema_at_most_once_across_many_batches() + { + var store = StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsGuid); + + for (var i = 0; i < 5; i++) + { + await store.BulkInsertEventsAsync(streams(store.Events, 2)); + } + + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + + var stats = await store.Advanced.FetchEventStoreStatistics(); + stats.EventCount.ShouldBe(20); + } + + [Fact] + public async Task apply_the_schema_at_most_once_across_many_batches_for_a_tenant() + { + var store = StoreOptions(opts => + { + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + }); + + for (var i = 0; i < 5; i++) + { + await store.BulkInsertEventsAsync("tenant-one", streams(store.Events, 2)); + } + + // Same single database behind every tenant here, so exactly one apply + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + } + + [Fact] + public async Task concurrent_batches_only_apply_the_schema_once() + { + var store = StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsGuid); + + var batches = Enumerable.Range(0, 10) + .Select(_ => store.BulkInsertEventsAsync(streams(store.Events, 2))) + .ToArray(); + + await Task.WhenAll(batches); + + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + } + + [Fact] + public async Task no_schema_apply_at_all_when_auto_create_is_none() + { + // Get the schema in place with a "normal" store first + var creator = StoreOptions(opts => + { + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.Events.AddEventTypes([typeof(QuestStarted), typeof(QuestEnded)]); + }); + + await creator.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); + + // Same schema, but the schema is managed out of band now + var store = SeparateStore(opts => + { + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.AutoCreateSchemaObjects = AutoCreate.None; + }); + + for (var i = 0; i < 3; i++) + { + await store.BulkInsertEventsAsync(streams(store.Events, 2)); + } + + // The apply is a no-op by contract under AutoCreate.None, so it should never be attempted + store.BulkInsertSchemaApplicationCount.ShouldBe(0); + + var stats = await store.Advanced.FetchEventStoreStatistics(); + stats.EventCount.ShouldBe(12); + } +} diff --git a/src/Marten/DocumentStore.cs b/src/Marten/DocumentStore.cs index 42bfd17c4d..49fad004eb 100644 --- a/src/Marten/DocumentStore.cs +++ b/src/Marten/DocumentStore.cs @@ -1,5 +1,6 @@ #nullable enable using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -189,12 +190,73 @@ public async Task BulkInsertDocumentsAsync(string tenantId, IEnumerable await bulkInsertion.BulkInsertDocumentsAsync(documents, mode, batchSize, cancellation).ConfigureAwait(false); } + /// + /// Tracks which databases have already had the schema apply run on behalf of the bulk event + /// insert path, keyed by database identifier. See . + /// + private readonly ConcurrentDictionary> _bulkInsertSchemaApplications = new(); + + /// + /// Test hook (GH-4946): how many times the bulk event insert path has actually executed the + /// schema apply. The regression being guarded is *how many times* the apply runs, not whether + /// the insert works. + /// + internal int BulkInsertSchemaApplicationCount; + + /// + /// GH-4946: the batch BulkInsertEventsAsync overloads used to call + /// Storage.ApplyAllConfiguredChangesToDatabaseAsync() on *every* call — a full schema delta + /// (partition introspection + information_schema sweeps) across *every* database in the store, per + /// batch. A caller importing in 1,000-event batches paid that per batch, which collapsed import + /// throughput and parked the connection pool on Weasel's partition introspection query. + /// + /// The apply is now (a) skipped altogether when the effective is + /// , where it is a no-op by contract and only the introspection cost + /// remains, and (b) otherwise run at most once per *database* — not per call, and not across + /// databases the import never touches. A 512-database sharded store therefore applies at most once + /// per database it actually imports into. + /// + /// + /// Thread safety: batches may be imported concurrently. The per-database entry is a + /// with , so + /// concurrent callers for the same database all await the single in-flight apply. A failed apply is + /// evicted so a later call can retry rather than caching the failure forever. + /// + /// + private Task ensureBulkInsertSchemaAsync(IMartenDatabase database, CancellationToken cancellation) + { + if (Options.AutoCreateSchemaObjects == AutoCreate.None) + { + return Task.CompletedTask; + } + + var lazy = _bulkInsertSchemaApplications.GetOrAdd(database.Identifier, + _ => new Lazy(() => applyAsync(database, cancellation), + LazyThreadSafetyMode.ExecutionAndPublication)); + + return lazy.Value; + + async Task applyAsync(IMartenDatabase db, CancellationToken token) + { + try + { + Interlocked.Increment(ref BulkInsertSchemaApplicationCount); + await db.ApplyAllConfiguredChangesToDatabaseAsync(ct: token).ConfigureAwait(false); + } + catch + { + _bulkInsertSchemaApplications.TryRemove(db.Identifier, out _); + throw; + } + } + } + public async Task BulkInsertEventsAsync(IReadOnlyList streams, int batchSize = 1000, CancellationToken cancellation = default) { - await Storage.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); - var tenant = Tenancy.Default; + await ensureBulkInsertSchemaAsync(tenant.Database, cancellation).ConfigureAwait(false); + var appender = new BulkEventAppender(Events, Options.Serializer()); await using var conn = tenant.Database.CreateConnection(); @@ -216,10 +278,11 @@ public async Task BulkInsertEventsAsync(IReadOnlyList streams, int public async Task BulkInsertEventsAsync(string tenantId, IReadOnlyList streams, int batchSize = 1000, CancellationToken cancellation = default) { - await Storage.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); - var tenant = await Tenancy.GetTenantAsync(Options.TenantIdStyle.MaybeCorrectTenantId(tenantId)) .ConfigureAwait(false); + + await ensureBulkInsertSchemaAsync(tenant.Database, cancellation).ConfigureAwait(false); + var appender = new BulkEventAppender(Events, Options.Serializer()); // Set tenant on all streams From 2ccda097c08810c87c3a855ea8b820a40b2d164a Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 13 Jul 2026 19:31:57 -0500 Subject: [PATCH 2/2] Don't bind the shared bulk-insert schema apply to one caller's token (GH-4946) The per-database apply is memoized, so the first caller to arrive owns the single in-flight task that every concurrent caller for that database awaits. Binding that shared task to the first caller's CancellationToken meant one batch's cancellation would fail sibling batches that were never cancelled. The apply now runs untethered and each caller awaits it under its own token. A schema apply is short and idempotent, so letting it run to completion is the cheaper trade. Co-Authored-By: Claude Opus 4.8 (1M context) --- ...k_insert_does_not_apply_schema_per_call.cs | 33 +++++++++++++++++++ src/Marten/DocumentStore.cs | 19 ++++++++--- 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs b/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs index 4c0c43e699..0771edf9e1 100644 --- a/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs +++ b/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using JasperFx; using JasperFx.Events; @@ -83,6 +84,38 @@ public async Task concurrent_batches_only_apply_the_schema_once() store.BulkInsertSchemaApplicationCount.ShouldBe(1); } + [Fact] + public async Task one_batch_cancelling_does_not_fail_a_concurrent_sibling_batch() + { + // The per-database schema apply is memoized, so the FIRST caller to arrive owns the single + // in-flight task that every concurrent caller for the same database awaits. If that shared + // task ran under the first caller's CancellationToken, then cancelling one batch would fail + // every sibling batch that happened to be waiting on it — batches that were never cancelled. + // The apply therefore runs untethered, and each caller awaits it under its OWN token. + var store = StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsGuid); + + using var doomed = new CancellationTokenSource(); + + var cancelled = store.BulkInsertEventsAsync(streams(store.Events, 2), cancellation: doomed.Token); + var survivor = store.BulkInsertEventsAsync(streams(store.Events, 2)); + + await doomed.CancelAsync(); + + // Whatever becomes of the cancelled batch, the sibling must land on its own merits + try + { + await cancelled; + } + catch (OperationCanceledException) + { + // expected, and immaterial to the assertion below + } + + await Should.NotThrowAsync(async () => await survivor); + + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + } + [Fact] public async Task no_schema_apply_at_all_when_auto_create_is_none() { diff --git a/src/Marten/DocumentStore.cs b/src/Marten/DocumentStore.cs index 49fad004eb..ab07a70191 100644 --- a/src/Marten/DocumentStore.cs +++ b/src/Marten/DocumentStore.cs @@ -231,17 +231,28 @@ private Task ensureBulkInsertSchemaAsync(IMartenDatabase database, CancellationT } var lazy = _bulkInsertSchemaApplications.GetOrAdd(database.Identifier, - _ => new Lazy(() => applyAsync(database, cancellation), + _ => new Lazy(() => applyAsync(database), LazyThreadSafetyMode.ExecutionAndPublication)); - return lazy.Value; + // Await the shared apply under *this* caller's token. The apply itself deliberately does not + // take a caller's token: the first caller to arrive owns the in-flight task that every + // concurrent caller for the same database awaits, so honoring its token here would let one + // batch's cancellation fail a sibling batch that was never cancelled. A schema apply is short + // and idempotent, so letting it run to completion is the cheaper trade. + return lazy.Value.WaitAsync(cancellation); - async Task applyAsync(IMartenDatabase db, CancellationToken token) + async Task applyAsync(IMartenDatabase db) { try { Interlocked.Increment(ref BulkInsertSchemaApplicationCount); - await db.ApplyAllConfiguredChangesToDatabaseAsync(ct: token).ConfigureAwait(false); + + // MA0040 suppressed deliberately: this task is SHARED by every concurrent caller for + // this database, so it must not be bound to any one caller's token. Each caller applies + // its own token at the await site above (`lazy.Value.WaitAsync(cancellation)`). +#pragma warning disable MA0040 + await db.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); +#pragma warning restore MA0040 } catch {