Carry DCB tags through a bulk event import in hstore mode

docs/events/bulk-appending.md

@@ -281,8 +281,12 @@ The bulk append API intentionally trades off features for throughput:
 - **No optimistic concurrency** -- there is no version checking against existing streams. This API is
   designed for initial data loading, not concurrent writes.
 - **New streams only** -- bulk append creates new streams. It does not support appending to existing streams.
-- **No event tags** -- DCB tag operations are not included in the COPY pipeline. Tags would need to be
-  handled separately after bulk loading.
+- **Event tags: hstore only** -- with `DcbStorageMode.HStore` the DCB tags travel in the same COPY as the
+  events, because they live in a column on `mt_events`; tag a built `IEvent` with `WithTag(...)` before
+  handing its `StreamAction` to the bulk API and a tag query finds it like any appended event. With
+  `DcbStorageMode.TagTables` the tags are rows in per-type tables and are *not* written by the COPY
+  pipeline, so they have to be handled separately after bulk loading. This matters for an imported history:
+  an untagged event is not an error to a tag query, it is simply absent from the answer.
 
 ## Performance
 

src/EventSourcingTests/Dcb/hstore_dcb_tags_survive_a_bulk_import.cs

@@ -0,0 +1,95 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using JasperFx.Events;
+using JasperFx.Events.Tags;
+using Marten;
+using Marten.Events;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+
+namespace EventSourcingTests.Dcb;
+
+/// <summary>
+/// A store whose history arrived through <c>BulkInsertEventsAsync</c> — a migration from another system,
+/// a restore — has to answer DCB tag queries about that history too. The bulk path writes events with a
+/// single COPY into <c>mt_events</c> over a fixed column list, so a tag only survives it if that column
+/// list carries it. In <see cref="DcbStorageMode.HStore" /> the tags are a column on the events table, so
+/// they ride along in the same COPY.
+/// <para>
+/// Without that, a bulk-imported event is invisible to every tag query, and invisible is exactly what a
+/// consistency boundary must not be: the answer looks like "no events" rather than like a failure.
+/// </para>
+/// </summary>
+[Collection("OneOffs")]
+public class hstore_dcb_tags_survive_a_bulk_import: OneOffConfigurationsContext, IAsyncLifetime
+{
+    public override ValueTask InitializeAsync()
+    {
+        StoreOptions(opts =>
+        {
+            opts.Events.AddEventType<StudentEnrolled>();
+            opts.Events.AddEventType<AssignmentSubmitted>();
+
+            opts.Events.DcbStorageMode = DcbStorageMode.HStore;
+            opts.Events.RegisterTagType<StudentId>("student");
+            opts.Events.RegisterTagType<CourseId>("course");
+        });
+
+        return default;
+    }
+
+    public override ValueTask DisposeAsync() => base.DisposeAsync();
+
+    [Fact]
+    public async Task a_tagged_event_is_found_after_a_bulk_import()
+    {
+        var studentId = new StudentId(Guid.NewGuid());
+        var courseId = new CourseId(Guid.NewGuid());
+
+        var action = StreamAction.Start(theStore.Events, Guid.NewGuid(),
+            new StudentEnrolled("Alice", "Math"));
+        foreach (var e in action.Events)
+        {
+            e.WithTag(studentId, courseId);
+        }
+
+        await theStore.BulkInsertEventsAsync(new List<StreamAction> { action });
+
+        var byStudent = await theSession.Events.QueryByTagsAsync(new EventTagQuery().Or<StudentId>(studentId));
+        byStudent.Count.ShouldBe(1);
+        byStudent[0].Data.ShouldBeOfType<StudentEnrolled>().StudentName.ShouldBe("Alice");
+
+        // Both registered tag types land in the one hstore, so either finds it.
+        var byCourse = await theSession.Events.QueryByTagsAsync(new EventTagQuery().Or<CourseId>(courseId));
+        byCourse.Count.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task an_untagged_event_in_the_same_import_is_not_found()
+    {
+        var studentId = new StudentId(Guid.NewGuid());
+
+        var tagged = StreamAction.Start(theStore.Events, Guid.NewGuid(), new StudentEnrolled("Alice", "Math"));
+        foreach (var e in tagged.Events)
+        {
+            e.WithTag(studentId);
+        }
+
+        // No tag at all: the column has to be written as null rather than skipped, or the COPY row goes out
+        // of step with its column list.
+        var untagged = StreamAction.Start(theStore.Events, Guid.NewGuid(), new StudentEnrolled("Bob", "Math"));
+
+        await theStore.BulkInsertEventsAsync(new List<StreamAction> { tagged, untagged });
+
+        var found = await theSession.Events.QueryByTagsAsync(new EventTagQuery().Or<StudentId>(studentId));
+        found.Count.ShouldBe(1);
+        found[0].Data.ShouldBeOfType<StudentEnrolled>().StudentName.ShouldBe("Alice");
+
+        // And both events did land — the import is not what dropped Bob.
+        var all = await theSession.Events.QueryAllRawEvents().ToListAsync(default);
+        all.Count.ShouldBe(2);
+    }
+}

src/Marten/Events/BulkEventAppender.cs

@@ -6,6 +6,7 @@
 using JasperFx;
 using JasperFx.Events;
 using Marten.Events.Daemon.HighWater;
+using Marten.Events.Operations;
 using Marten.Events.Schema;
 using Marten.Storage;
 using Npgsql;
@@ -674,6 +675,14 @@ private List<string> buildEventColumns()
             columns.Add("user_name");
         }
 
+        // Last, so the optional metadata columns above keep the positions they had. In hstore mode the DCB
+        // tags are a column on mt_events, so they ride along in this same COPY; without this a bulk-imported
+        // event is invisible to every tag query, which is silent rather than loud.
+        if (_events.DcbStorageMode == DcbStorageMode.HStore)
+        {
+            columns.Add("tags");
+        }
+
         return columns;
     }
 
@@ -797,6 +806,27 @@ private async Task writeEventRow(NpgsqlBinaryImporter writer, IEvent e, Guid str
                 await writer.WriteNullAsync(cancellation).ConfigureAwait(false);
             }
         }
+
+        // tags (hstore, nullable) — same rule the append path applies, so a bulk-imported event answers a
+        // DCB tag query exactly like an appended one. Null when the event carries no tag, or none of its
+        // tags belongs to a registered tag type.
+        if (_events.DcbStorageMode == DcbStorageMode.HStore)
+        {
+            // Null rather than empty when nothing was ever tagged, so this cannot be a Count check alone.
+            var tags = e.Tags;
+            var hstore = tags is { Count: > 0 }
+                ? EventTagOperations.BuildHstore(_events, tags)
+                : null;
+
+            if (hstore is { Count: > 0 })
+            {
+                await writer.WriteAsync(hstore, NpgsqlDbType.Hstore, cancellation).ConfigureAwait(false);
+            }
+            else
+            {
+                await writer.WriteNullAsync(cancellation).ConfigureAwait(false);
+            }
+        }
     }
 
     private async Task updateHighWaterMark(

src/Marten/Events/Operations/EventTagOperations.cs

@@ -141,7 +141,11 @@ private static void CollectDcbVersionTargets(EventGraph eventGraph,
     /// path). Key is the registered tag's <c>TableSuffix</c>, value is the stringified
     /// tag value (Npgsql maps the dictionary to hstore via <c>NpgsqlDbType.Hstore</c>).
     /// </summary>
-    private static Dictionary<string, string> BuildHstore(EventGraph eventGraph,
+    /// <summary>
+    /// Internal rather than private because the bulk import writes the same value into the same column,
+    /// and a second implementation of this rule would be a second thing to keep in step.
+    /// </summary>
+    internal static Dictionary<string, string> BuildHstore(EventGraph eventGraph,
         IReadOnlyList<EventTag> tags)
     {
         var result = new Dictionary<string, string>(capacity: tags.Count);