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 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);