Bulk event insert applies the schema once per database, not per call (#4946)

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`

src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs

@@ -0,0 +1,149 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+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;
+
+/// <summary>
+/// GH-4946: the batch <c>BulkInsertEventsAsync</c> overloads ran
+/// <c>Storage.ApplyAllConfiguredChangesToDatabaseAsync()</c> 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.
+/// </summary>
+public class Bug_4946_bulk_insert_does_not_apply_schema_per_call: OneOffConfigurationsContext
+{
+    private static List<StreamAction> streams(EventGraph events, int count)
+    {
+        var actions = new List<StreamAction>();
+        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 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()
+    {
+        // 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);
+    }
+}

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,84 @@ public async Task BulkInsertDocumentsAsync(string tenantId, IEnumerable<object>
         await bulkInsertion.BulkInsertDocumentsAsync(documents, mode, batchSize, cancellation).ConfigureAwait(false);
     }
 
+    /// <summary>
+    ///     Tracks which databases have already had the schema apply run on behalf of the bulk event
+    ///     insert path, keyed by database identifier. See <see cref="ensureBulkInsertSchemaAsync" />.
+    /// </summary>
+    private readonly ConcurrentDictionary<string, Lazy<Task>> _bulkInsertSchemaApplications = new();
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    internal int BulkInsertSchemaApplicationCount;
+
+    /// <summary>
+    ///     GH-4946: the batch <c>BulkInsertEventsAsync</c> overloads used to call
+    ///     <c>Storage.ApplyAllConfiguredChangesToDatabaseAsync()</c> 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.
+    ///     <para>
+    ///     The apply is now (a) skipped altogether when the effective <see cref="AutoCreate" /> is
+    ///     <see cref="AutoCreate.None" />, 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.
+    ///     </para>
+    ///     <para>
+    ///     Thread safety: batches may be imported concurrently. The per-database entry is a
+    ///     <see cref="Lazy{T}" /> with <see cref="LazyThreadSafetyMode.ExecutionAndPublication" />, 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.
+    ///     </para>
+    /// </summary>
+    private Task ensureBulkInsertSchemaAsync(IMartenDatabase database, CancellationToken cancellation)
+    {
+        if (Options.AutoCreateSchemaObjects == AutoCreate.None)
+        {
+            return Task.CompletedTask;
+        }
+
+        var lazy = _bulkInsertSchemaApplications.GetOrAdd(database.Identifier,
+            _ => new Lazy<Task>(() => applyAsync(database),
+                LazyThreadSafetyMode.ExecutionAndPublication));
+
+        // 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)
+        {
+            try
+            {
+                Interlocked.Increment(ref BulkInsertSchemaApplicationCount);
+
+                // 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
+            {
+                _bulkInsertSchemaApplications.TryRemove(db.Identifier, out _);
+                throw;
+            }
+        }
+    }
+
     public async Task BulkInsertEventsAsync(IReadOnlyList<StreamAction> 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 +289,11 @@ public async Task BulkInsertEventsAsync(IReadOnlyList<StreamAction> streams, int
     public async Task BulkInsertEventsAsync(string tenantId, IReadOnlyList<StreamAction> 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