diff --git a/docs/cSpell.json b/docs/cSpell.json index 0d17b0a833..a15696d61e 100644 --- a/docs/cSpell.json +++ b/docs/cSpell.json @@ -3,6 +3,7 @@ "language": "en", "words": [ "jasperfx", + "batchable", "TimescaleDB", "timescaledb", "hypertable", diff --git a/docs/documents/querying/compiled-queries.md b/docs/documents/querying/compiled-queries.md index c4e446b819..bf94c0489b 100644 --- a/docs/documents/querying/compiled-queries.md +++ b/docs/documents/querying/compiled-queries.md @@ -733,7 +733,7 @@ public interface IBatchQueryPlan Task Fetch(IBatchedQuery query); } ``` -snippet source | anchor +snippet source | anchor And because we expect this to be very common, there is convenience base class named `QueryListPlan` for querying lists of `T` data that can be used for both querying directly against an `IQuerySession` and for batch querying. The usage within a batched query is shown below from the Marten tests: @@ -778,3 +778,70 @@ public async Task use_as_batch() ``` snippet source | anchor + +### Query Plans for Event Streams + +Marten ships two concrete query plans for the raw event stream fetches so that they can also be used as batchable +specifications. `FetchStreamStatePlan` fetches the high level `StreamState` metadata about a single stream (yielding +`null` when the stream does not exist), while `FetchStreamPlan` fetches the raw events of a single stream (yielding an +empty list when the stream does not exist) with support for the optional `version`, `timestamp`, and `fromVersion` +filters of `FetchStreamAsync()`. Both plans accept either a `Guid` stream id or a `string` stream key to cover both +stream identity styles. + +Using `FetchStreamPlan` standalone against a session: + + + +```cs +[Fact] +public async Task fetch_stream_by_query_plan() +{ + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam"), + new MembersJoined(2, "Bree", "Aragorn")).Id; + await theSession.SaveChangesAsync(); + + var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(streamId)); + + events.Count.ShouldBe(3); + events[0].Data.ShouldBeOfType(); +} +``` +snippet source | anchor + + +And because both plans also implement `IBatchQueryPlan`, they can be combined with any other registered queries +within a [batched query](/documents/querying/batched-queries) to fetch a stream's state and its raw events in one +database round trip: + + + +```cs +[Fact] +public async Task use_both_plans_in_one_batch() +{ + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id; + await theSession.SaveChangesAsync(); + + // Start a batch query + var batch = theSession.CreateBatchQuery(); + + // Fetching the stream state and the raw events of the same stream + // in one database round trip + var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); + var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); + + // Execute the batch query + await batch.Execute(); + + var state = await stateFetcher; + var events = await eventsFetcher; + + state.ShouldNotBeNull(); + state.Version.ShouldBe(2); + events.Count.ShouldBe(2); +} +``` +snippet source | anchor + diff --git a/docs/events/querying.md b/docs/events/querying.md index 1f5fe63604..ba8dd14b2b 100644 --- a/docs/events/querying.md +++ b/docs/events/querying.md @@ -258,6 +258,47 @@ public class fetching_stream_state: IntegrationContext Furthermore, `StreamState` contains metadata for when the stream was created, `StreamState.Created`, and when the stream was last updated, `StreamState.LastTimestamp`. +## Stream Query Plans + +The stream fetches above are also available as reusable [query plans](/documents/querying/compiled-queries#query-plans-): +`FetchStreamStatePlan` wraps `FetchStreamState()`/`FetchStreamStateAsync()` and `FetchStreamPlan` wraps `FetchStream()`/`FetchStreamAsync()`. +Both plans accept either a `Guid` stream id or a `string` stream key, and `FetchStreamPlan` carries the optional +`version`, `timestamp`, and `fromVersion` filters. Because the plans implement both `IQueryPlan` and +`IBatchQueryPlan`, the same object works standalone with `IQuerySession.QueryByPlanAsync()` or combined with any +other registered queries into a single database round trip within a batched query: + + + +```cs +[Fact] +public async Task use_both_plans_in_one_batch() +{ + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id; + await theSession.SaveChangesAsync(); + + // Start a batch query + var batch = theSession.CreateBatchQuery(); + + // Fetching the stream state and the raw events of the same stream + // in one database round trip + var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); + var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); + + // Execute the batch query + await batch.Execute(); + + var state = await stateFetcher; + var events = await eventsFetcher; + + state.ShouldNotBeNull(); + state.Version.ShouldBe(2); + events.Count.ShouldBe(2); +} +``` +snippet source | anchor + + ## Fetch a Single Event You can fetch the information for a single event by id, including its version number within the stream, by using `IEventStore.LoadAsync()` as shown below: diff --git a/src/EventSourcingTests/fetching_stream_query_plans.cs b/src/EventSourcingTests/fetching_stream_query_plans.cs new file mode 100644 index 0000000000..d5e624c067 --- /dev/null +++ b/src/EventSourcingTests/fetching_stream_query_plans.cs @@ -0,0 +1,134 @@ +using System; +using System.Threading.Tasks; +using JasperFx.Events; +using Marten; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace EventSourcingTests; + +public class fetching_stream_query_plans: OneOffConfigurationsContext +{ + [Fact] + public async Task fetch_stream_state_by_query_plan() + { + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id; + await theSession.SaveChangesAsync(); + + var state = await theSession.QueryByPlanAsync(new FetchStreamStatePlan(streamId)); + + state.ShouldNotBeNull(); + state.Id.ShouldBe(streamId); + state.Version.ShouldBe(2); + } + + [Fact] + public async Task fetch_stream_state_by_query_plan_with_string_identity() + { + StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsString); + + theSession.Events.Append("one-ring", new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam")); + await theSession.SaveChangesAsync(); + + var state = await theSession.QueryByPlanAsync(new FetchStreamStatePlan("one-ring")); + + state.ShouldNotBeNull(); + state.Key.ShouldBe("one-ring"); + state.Version.ShouldBe(2); + } + + [Fact] + public async Task fetch_stream_state_by_query_plan_for_missing_stream_is_null() + { + var state = await theSession.QueryByPlanAsync(new FetchStreamStatePlan(Guid.NewGuid())); + + state.ShouldBeNull(); + } + + #region sample_using_fetch_stream_plan + + [Fact] + public async Task fetch_stream_by_query_plan() + { + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam"), + new MembersJoined(2, "Bree", "Aragorn")).Id; + await theSession.SaveChangesAsync(); + + var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(streamId)); + + events.Count.ShouldBe(3); + events[0].Data.ShouldBeOfType(); + } + + #endregion + + [Fact] + public async Task fetch_stream_by_query_plan_with_string_identity() + { + StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsString); + + theSession.Events.Append("one-ring", new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam")); + await theSession.SaveChangesAsync(); + + var events = await theSession.QueryByPlanAsync(new FetchStreamPlan("one-ring")); + + events.Count.ShouldBe(2); + } + + [Fact] + public async Task fetch_stream_by_query_plan_with_version_cap() + { + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam"), + new MembersJoined(2, "Bree", "Aragorn")).Id; + await theSession.SaveChangesAsync(); + + var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(streamId, version: 2)); + + events.Count.ShouldBe(2); + events[^1].Version.ShouldBe(2); + } + + [Fact] + public async Task fetch_stream_by_query_plan_for_missing_stream_is_empty() + { + var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(Guid.NewGuid())); + + events.ShouldBeEmpty(); + } + + #region sample_fetch_stream_plans_in_batch + + [Fact] + public async Task use_both_plans_in_one_batch() + { + var streamId = theSession.Events.StartStream(new QuestStarted { Name = "Destroy the One Ring" }, + new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id; + await theSession.SaveChangesAsync(); + + // Start a batch query + var batch = theSession.CreateBatchQuery(); + + // Fetching the stream state and the raw events of the same stream + // in one database round trip + var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); + var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); + + // Execute the batch query + await batch.Execute(); + + var state = await stateFetcher; + var events = await eventsFetcher; + + state.ShouldNotBeNull(); + state.Version.ShouldBe(2); + events.Count.ShouldBe(2); + } + + #endregion +} diff --git a/src/Marten/IQueryPlan.cs b/src/Marten/IQueryPlan.cs index cb7ae80efc..b04b424ec5 100644 --- a/src/Marten/IQueryPlan.cs +++ b/src/Marten/IQueryPlan.cs @@ -3,6 +3,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using JasperFx.Events; using Marten.Linq; using Marten.Services.BatchQuerying; @@ -65,3 +66,106 @@ Task> IBatchQueryPlan>.Fetch(IBatchedQuery que return query.AddItem(handler); } } + +/// +/// Query plan to fetch the high level metadata about a single event stream identified by +/// either a Guid stream id or a string stream key. Can be used both individually with +/// IQuerySession.QueryByPlanAsync() and with IBatchedQuery.QueryByPlan(). Yields null +/// if the stream does not exist +/// +public class FetchStreamStatePlan : IQueryPlan, IBatchQueryPlan +{ + private readonly Guid _streamId; + private readonly string? _streamKey; + + /// + /// Fetch the stream state for the stream identified by + /// + /// + public FetchStreamStatePlan(Guid streamId) + { + _streamId = streamId; + } + + /// + /// Fetch the stream state for the stream identified by + /// + /// + public FetchStreamStatePlan(string streamKey) + { + _streamKey = streamKey; + } + + public Task Fetch(IQuerySession session, CancellationToken token) + { + return _streamKey is not null + ? session.Events.FetchStreamStateAsync(_streamKey, token) + : session.Events.FetchStreamStateAsync(_streamId, token); + } + + public async Task Fetch(IBatchedQuery query) + { + return _streamKey is not null + ? await query.Events.FetchStreamState(_streamKey).ConfigureAwait(false) + : await query.Events.FetchStreamState(_streamId).ConfigureAwait(false); + } +} + +/// +/// Query plan to fetch the raw events for a single event stream identified by either a +/// Guid stream id or a string stream key. Can be used both individually with +/// IQuerySession.QueryByPlanAsync() and with IBatchedQuery.QueryByPlan(). Yields an +/// empty list if the stream does not exist +/// +public class FetchStreamPlan : IQueryPlan>, IBatchQueryPlan> +{ + private readonly Guid _streamId; + private readonly string? _streamKey; + private readonly long _version; + private readonly DateTimeOffset? _timestamp; + private readonly long _fromVersion; + + /// + /// Fetch the events for the stream identified by + /// + /// + /// If set, queries for events up to and including this version + /// If set, queries for events captured on or before this timestamp + /// If set, queries for events on or from this version + public FetchStreamPlan(Guid streamId, long version = 0, DateTimeOffset? timestamp = null, long fromVersion = 0) + { + _streamId = streamId; + _version = version; + _timestamp = timestamp; + _fromVersion = fromVersion; + } + + /// + /// Fetch the events for the stream identified by + /// + /// + /// If set, queries for events up to and including this version + /// If set, queries for events captured on or before this timestamp + /// If set, queries for events on or from this version + public FetchStreamPlan(string streamKey, long version = 0, DateTimeOffset? timestamp = null, long fromVersion = 0) + { + _streamKey = streamKey; + _version = version; + _timestamp = timestamp; + _fromVersion = fromVersion; + } + + public Task> Fetch(IQuerySession session, CancellationToken token) + { + return _streamKey is not null + ? session.Events.FetchStreamAsync(_streamKey, _version, _timestamp, _fromVersion, token) + : session.Events.FetchStreamAsync(_streamId, _version, _timestamp, _fromVersion, token); + } + + public Task> Fetch(IBatchedQuery query) + { + return _streamKey is not null + ? query.Events.FetchStream(_streamKey, _version, _timestamp, _fromVersion) + : query.Events.FetchStream(_streamId, _version, _timestamp, _fromVersion); + } +}