From 752e03af7cd04035173ccebc8d3303b0f33764af Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Thu, 20 Aug 2026 16:39:10 +0200 Subject: [PATCH 1/2] Carry DCB tags through a bulk event import in hstore mode BulkInsertEventsAsync writes events with a single COPY into mt_events over a fixed column list, and the DCB tags were not in it - nor did the bulk path write anything to the per-type tag tables. So every event that arrived through a bulk import landed untagged, and a tag query saw nothing of it. For a store whose history came in through a migration that is most of the history, and it fails the wrong way: the answer looks like "no events" rather than like an error. In hstore mode the tags are a column on mt_events itself, so they can travel in the same COPY: one more column at the end of the list - after the optional metadata columns, so those keep their positions - and one more write per row. The value comes from EventTagOperations.BuildHstore, the same rule the append path applies, which is why BuildHstore is internal now rather than private; a second implementation would be a second thing to keep in step. IEvent.Tags is null rather than empty when nothing was tagged, so the write is guarded on that and emits NULL. The second test covers it: a COPY row that skips a column instead of writing null goes out of step with its column list, which would corrupt every later column rather than merely lose a tag. TagTables mode is deliberately untouched: that needs a second COPY per registered tag type keyed on the seq_id, which interacts with the sequence blocks the events COPY draws from. --- .../hstore_dcb_tags_survive_a_bulk_import.cs | 95 +++++++++++++++++++ src/Marten/Events/BulkEventAppender.cs | 30 ++++++ .../Events/Operations/EventTagOperations.cs | 6 +- 3 files changed, 130 insertions(+), 1 deletion(-) create mode 100644 src/EventSourcingTests/Dcb/hstore_dcb_tags_survive_a_bulk_import.cs diff --git a/src/EventSourcingTests/Dcb/hstore_dcb_tags_survive_a_bulk_import.cs b/src/EventSourcingTests/Dcb/hstore_dcb_tags_survive_a_bulk_import.cs new file mode 100644 index 0000000000..b08df2dccb --- /dev/null +++ b/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; + +/// +/// A store whose history arrived through BulkInsertEventsAsync — 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 mt_events over a fixed column list, so a tag only survives it if that column +/// list carries it. In the tags are a column on the events table, so +/// they ride along in the same COPY. +/// +/// 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. +/// +/// +[Collection("OneOffs")] +public class hstore_dcb_tags_survive_a_bulk_import: OneOffConfigurationsContext, IAsyncLifetime +{ + public override ValueTask InitializeAsync() + { + StoreOptions(opts => + { + opts.Events.AddEventType(); + opts.Events.AddEventType(); + + opts.Events.DcbStorageMode = DcbStorageMode.HStore; + opts.Events.RegisterTagType("student"); + opts.Events.RegisterTagType("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 { action }); + + var byStudent = await theSession.Events.QueryByTagsAsync(new EventTagQuery().Or(studentId)); + byStudent.Count.ShouldBe(1); + byStudent[0].Data.ShouldBeOfType().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)); + 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 { tagged, untagged }); + + var found = await theSession.Events.QueryByTagsAsync(new EventTagQuery().Or(studentId)); + found.Count.ShouldBe(1); + found[0].Data.ShouldBeOfType().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); + } +} diff --git a/src/Marten/Events/BulkEventAppender.cs b/src/Marten/Events/BulkEventAppender.cs index 24e8a1001c..f6886079d0 100644 --- a/src/Marten/Events/BulkEventAppender.cs +++ b/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 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( diff --git a/src/Marten/Events/Operations/EventTagOperations.cs b/src/Marten/Events/Operations/EventTagOperations.cs index b664db6d8e..dd32b17ed1 100644 --- a/src/Marten/Events/Operations/EventTagOperations.cs +++ b/src/Marten/Events/Operations/EventTagOperations.cs @@ -141,7 +141,11 @@ private static void CollectDcbVersionTargets(EventGraph eventGraph, /// path). Key is the registered tag's TableSuffix, value is the stringified /// tag value (Npgsql maps the dictionary to hstore via NpgsqlDbType.Hstore). /// - private static Dictionary BuildHstore(EventGraph eventGraph, + /// + /// 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. + /// + internal static Dictionary BuildHstore(EventGraph eventGraph, IReadOnlyList tags) { var result = new Dictionary(capacity: tags.Count); From 0931edf84c60386184ad8bb5d448cbdd501ae68a Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Thu, 20 Aug 2026 16:50:25 +0200 Subject: [PATCH 2/2] Document that bulk-appended events carry DCB tags in hstore mode --- docs/events/bulk-appending.md | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/docs/events/bulk-appending.md b/docs/events/bulk-appending.md index 05732c51c9..07897a8ec0 100644 --- a/docs/events/bulk-appending.md +++ b/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