From 4bc484dac949c751ef06c7e37f5ba6f4328fad0c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tim=20M=C3=B6hlmann?= Date: Tue, 22 Sep 2026 17:09:42 +0200 Subject: [PATCH] perf(eventstore): read projection events per event type (#12753) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit # Which Problems Are Solved - A projection that subscribes to several event types gets slower the further it falls behind. - The handler reads its events with one query: `aggregate_type = ANY(...) AND event_type = ANY(...)`, ordered by position and limited to the bulk limit. - The `es_projection` index returns each (aggregate type, event type) range in position order, but Postgres cannot merge those ranges. - So each batch reads every matching event after the projection's position, sorts them all, and keeps only the bulk limit. - A projection that is N events behind therefore reads about N rows per batch. Catching up costs about N² in total, and once events arrive faster than such batches can process them, the projection does not recover. - Every query that triggers the projection waits for that catch-up. Examples are `GetSession` and the notification handlers. # How the Problems Are Solved - **New opt-in, `SearchQueryBuilder.ScanEventTypesSeparately()`.** It applies to event queries for a single instance, ascending and limited, whose sub-queries filter only by aggregate and event types. - For such a query, the SQL layer reads each distinct (aggregate type, event type) combination in its own `ORDER BY … LIMIT` subquery, using `UNNEST … CROSS JOIN LATERAL`. - It then merges the results with the same sort key and limit. - Each subquery is one `es_projection` range read in position order, so it stops at the limit. A batch reads at most `limit × combinations` rows, however far behind the projection is. - Every other query renders exactly as before. - **The projection handler** now passes its real combinations, one sub-query per aggregate type instead of the cross product of all types, and sets the opt-in. - **Results are identical to the single query.** - The sort key (`position, in_tx_order, instance_id, aggregate_type, aggregate_id, sequence`) is a total order, and the global top n is contained in the union of each combination's top n. - The cursor from #12703 and the open-transaction cap apply inside every subquery. - A test against a real Postgres compares both shapes from every cursor and every position, for several limits. The test events include positions shared by several transactions, whose `in_tx_order` restarts. # Additional Changes This is a separate commit and can be split into its own PR: - `cmd/defaults.yaml` sets `BulkLimit: 1000` for the `sessions` projection. - `projection_state_latency` gets histogram buckets up to one week. Until now, every projection more than 30 minutes behind was reported in the same top bucket. # Additional Context These numbers come from a local, CPU-pinned sandbox, for comparing builds only. - **Database:** Postgres 17 in a VM with 8 vCPUs and 16 GiB. The `Eventstore.Autovacuum` and `Projections` settings mirror a production-like setup. - **Starting state:** `sessions8` 4.8M events behind, restored byte-identically for every run. Batch query for `sessions8`, `EXPLAIN ANALYZE`, median of 5 runs, parallel workers at the default: | | before | after | | --- | ---: | ---: | | at the backlog, limit 1000 | 832 ms | 4.0 ms | | at the backlog, limit 200 | 832 ms | 1.0 ms | | at the head (idle poll) | 0.14 ms | 0.19 ms | | generic plan (`plan_cache_mode=force_generic_plan`), at the backlog | 1,042 ms | 4.0 ms | Draining `sessions8` from that backlog for 600 s, 3 runs per build (median, with the range in brackets): | | before | after | | --- | ---: | ---: | | events/s | 975 (932-1,032) | 7,811 (7,523-7,955) | | mean batch time, including writes | 1.03 s | 0.13 s | | still behind after 600 s | 4.2M | 17k-277k | A new index `(instance_id, position, in_tx_order, aggregate_type, event_type)` with the old query was evaluated as an alternative and rejected. Under a generic plan, Postgres chose it for projections whose events are rare and walked the whole event stream: 430 ms instead of 0.15 ms. This is the behaviour that led to dropping `es_instance_position` in #10626. - Follow-up for #12703 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- cmd/defaults.yaml | 5 + internal/eventstore/handler/v2/handler.go | 15 +- .../eventstore/handler/v2/handler_test.go | 33 +++ internal/eventstore/handler/v2/metrics.go | 4 +- .../eventstore/repository/search_query.go | 46 ++++ .../repository/search_query_test.go | 67 +++++ .../eventstore/repository/sql/postgres.go | 20 +- internal/eventstore/repository/sql/query.go | 119 +++++++-- .../sql/query_event_type_scans_test.go | 233 ++++++++++++++++++ internal/eventstore/search_query.go | 18 ++ 10 files changed, 516 insertions(+), 44 deletions(-) create mode 100644 internal/eventstore/repository/sql/query_event_type_scans_test.go diff --git a/cmd/defaults.yaml b/cmd/defaults.yaml index 0032b9061..fdfac4b1e 100644 --- a/cmd/defaults.yaml +++ b/cmd/defaults.yaml @@ -549,6 +549,11 @@ Projections: Customizations: custom_texts: BulkLimit: 400 + # The sessions projection processes the highest event volume. + # A higher bulk limit amortizes the fixed cost per batch over more events, + # which shortens catch-up after the projection fell behind. + sessions: + BulkLimit: 1000 # ZITADEL_PROJECTIONS_CUSTOMIZATIONS_SESSIONS_BULKLIMIT project_grant_fields: TransactionDuration: 0s BulkLimit: 2000 diff --git a/internal/eventstore/handler/v2/handler.go b/internal/eventstore/handler/v2/handler.go index 2c423cbf9..d221b2039 100644 --- a/internal/eventstore/handler/v2/handler.go +++ b/internal/eventstore/handler/v2/handler.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "log/slog" + "maps" "math/rand" "slices" "sync" @@ -724,14 +725,24 @@ func (h *Handler) eventQuery(currentState *state, minPosition decimal.Decimal) * return builder } + // Reading each event type separately keeps the cost of a batch bounded by the bulk limit. + // A single query for all event types must read every matching event after the position before it can sort and limit them, + // so a projection that falls behind would get slower the further behind it is. + // Handlers which subscribe to all events of an aggregate (no event types) keep the single query. + if !slices.ContainsFunc(slices.Collect(maps.Values(h.eventTypes)), func(eventTypes []eventstore.EventType) bool { return len(eventTypes) == 0 }) { + builder = builder.ScanEventTypesSeparately() + for _, aggregate := range slices.Sorted(maps.Keys(h.eventTypes)) { + builder = builder.AddQuery().AggregateTypes(aggregate).EventTypes(h.eventTypes[aggregate]...).Builder() + } + return builder + } + aggregateTypes := make([]eventstore.AggregateType, 0, len(h.eventTypes)) eventTypes := make([]eventstore.EventType, 0, len(h.eventTypes)) - for aggregate, events := range h.eventTypes { aggregateTypes = append(aggregateTypes, aggregate) eventTypes = append(eventTypes, events...) } - return builder.AddQuery().AggregateTypes(aggregateTypes...).EventTypes(eventTypes...).Builder() } diff --git a/internal/eventstore/handler/v2/handler_test.go b/internal/eventstore/handler/v2/handler_test.go index 50dbd0066..6eea6345c 100644 --- a/internal/eventstore/handler/v2/handler_test.go +++ b/internal/eventstore/handler/v2/handler_test.go @@ -19,6 +19,39 @@ func TestHandler_eventQuery(t *testing.T) { }, } + t.Run("scans every aggregate's event types separately", func(t *testing.T) { + h := &Handler{ + bulkLimit: 200, + eventTypes: map[eventstore.AggregateType][]eventstore.EventType{ + "user": {"user.human.added", "user.locked"}, + "org": {"org.removed"}, + }, + } + builder := h.eventQuery(&state{instanceID: "inst"}, decimal.Decimal{}) + assert.True(t, builder.GetScanEventTypesSeparately()) + queries := builder.GetQueries() + require.Len(t, queries, 2) + assert.Equal(t, []eventstore.AggregateType{"org"}, queries[0].GetAggregateTypes()) + assert.Equal(t, []eventstore.EventType{"org.removed"}, queries[0].GetEventTypes()) + assert.Equal(t, []eventstore.AggregateType{"user"}, queries[1].GetAggregateTypes()) + assert.Equal(t, []eventstore.EventType{"user.human.added", "user.locked"}, queries[1].GetEventTypes()) + }) + + t.Run("aggregate without event types keeps the single query", func(t *testing.T) { + h := &Handler{ + bulkLimit: 200, + eventTypes: map[eventstore.AggregateType][]eventstore.EventType{ + "org": nil, + }, + } + builder := h.eventQuery(&state{instanceID: "inst"}, decimal.Decimal{}) + assert.False(t, builder.GetScanEventTypesSeparately()) + queries := builder.GetQueries() + require.Len(t, queries, 1) + assert.Equal(t, []eventstore.AggregateType{"org"}, queries[0].GetAggregateTypes()) + assert.Empty(t, queries[0].GetEventTypes()) + }) + t.Run("no cursor on empty position", func(t *testing.T) { builder := h.eventQuery(&state{instanceID: "inst"}, decimal.Decimal{}) assert.Equal(t, uint32(0), builder.GetOffset()) diff --git a/internal/eventstore/handler/v2/metrics.go b/internal/eventstore/handler/v2/metrics.go index 9baf89322..40aa84712 100644 --- a/internal/eventstore/handler/v2/metrics.go +++ b/internal/eventstore/handler/v2/metrics.go @@ -45,7 +45,9 @@ func newProjectionMetrics(ctx context.Context, m metrics.Metrics) *ProjectionMet ProjectionStateLatencyMetric, "When finishing processing a batch of events, this track the age of the last events seen from current time", "s", - []float64{0.1, 0.5, 1, 5, 10, 30, 60, 300, 600, 1800}, + // the upper bounds up to a week make a projection that falls hours or days behind visible, + // instead of reporting every latency above 30 minutes as the 1800s bucket + []float64{0.1, 0.5, 1, 5, 10, 30, 60, 300, 600, 1800, 3600, 21600, 86400, 604800}, ) logging.OnError(ctx, err).Error("failed to register projection state latency metric") return projectionMetrics diff --git a/internal/eventstore/repository/search_query.go b/internal/eventstore/repository/search_query.go index 5a605d6b7..3fab18887 100644 --- a/internal/eventstore/repository/search_query.go +++ b/internal/eventstore/repository/search_query.go @@ -1,7 +1,9 @@ package repository import ( + "cmp" "database/sql" + "slices" "github.com/shopspring/decimal" @@ -32,6 +34,16 @@ type SearchQuery struct { CreatedBefore *Filter ExcludeAggregateIDs []*Filter EventSortKeyAfter *eventstore.EventSortKey + // EventTypeScans contains the combinations of aggregate type and event type to read separately, + // see [eventstore.SearchQueryBuilder.ScanEventTypesSeparately]. + // It is empty if the sub queries cannot be read that way. + EventTypeScans []EventTypeScan +} + +// EventTypeScan is a combination of aggregate type and event type, which is read separately. +type EventTypeScan struct { + AggregateType eventstore.AggregateType + EventType eventstore.EventType } // Filter represents all fields needed to compare a field of an event with a value @@ -135,6 +147,11 @@ func QueryFromBuilder(builder *eventstore.SearchQueryBuilder) (*SearchQuery, err SubQueries: make([][]*Filter, len(builder.GetQueries())), EventSortKeyAfter: builder.GetEventSortKeyAfter(), } + var err error + query.EventTypeScans, err = eventTypeScans(builder) + if err != nil { + return nil, err + } for _, f := range []func(builder *eventstore.SearchQueryBuilder, query *SearchQuery) *Filter{ instanceIDFilter, @@ -192,6 +209,35 @@ func QueryFromBuilder(builder *eventstore.SearchQueryBuilder) (*SearchQuery, err return query, nil } +// eventTypeScans returns the distinct combinations of aggregate type and event type of all sub queries, +// if [eventstore.SearchQueryBuilder.ScanEventTypesSeparately] was set. +// Each sub query must filter a single aggregate type and at least one event type, and nothing else, +// otherwise an error is returned. A single aggregate type is required because event types belong to an aggregate type: +// a sub query with several aggregate types would combine event types with aggregate types they do not belong to. +func eventTypeScans(builder *eventstore.SearchQueryBuilder) ([]EventTypeScan, error) { + if !builder.GetScanEventTypesSeparately() { + return nil, nil + } + var scans []EventTypeScan + for _, query := range builder.GetQueries() { + if len(query.GetAggregateTypes()) != 1 || + len(query.GetEventTypes()) == 0 || + len(query.GetAggregateIDs()) > 0 || + len(query.GetEventData()) > 0 || + !query.GetPositionAfter().IsZero() { + return nil, zerrors.ThrowInvalidArgument(nil, "REPO-Ahx6e", "scanning event types separately requires sub queries with one aggregate type and event types only") + } + for _, eventType := range query.GetEventTypes() { + scans = append(scans, EventTypeScan{AggregateType: query.GetAggregateTypes()[0], EventType: eventType}) + } + } + // sub queries are OR-connected, so a combination listed twice must still be read once + slices.SortFunc(scans, func(a, b EventTypeScan) int { + return cmp.Or(cmp.Compare(a.AggregateType, b.AggregateType), cmp.Compare(a.EventType, b.EventType)) + }) + return slices.Compact(scans), nil +} + func eventSequenceGreaterFilter(builder *eventstore.SearchQueryBuilder, query *SearchQuery) *Filter { if builder.GetEventSequenceGreater() == 0 { return nil diff --git a/internal/eventstore/repository/search_query_test.go b/internal/eventstore/repository/search_query_test.go index a830155e3..d95a4f7fa 100644 --- a/internal/eventstore/repository/search_query_test.go +++ b/internal/eventstore/repository/search_query_test.go @@ -4,7 +4,11 @@ import ( "reflect" "testing" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/zerrors" ) func TestNewFilter(t *testing.T) { @@ -144,3 +148,66 @@ func TestColumns_Validate(t *testing.T) { }) } } + +func TestQueryFromBuilder_eventTypeScans(t *testing.T) { + base := func() *eventstore.SearchQueryBuilder { + return eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).ScanEventTypesSeparately() + } + errScan := zerrors.ThrowInvalidArgument(nil, "REPO-Ahx6e", "") + tests := []struct { + name string + builder *eventstore.SearchQueryBuilder + want []EventTypeScan + wantErr error + }{ + { + name: "not requested", + builder: eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). + AddQuery().AggregateTypes("user").EventTypes("user.locked").Builder(), + want: nil, + }, + { + name: "event types of each sub query, sorted and distinct", + builder: base(). + AddQuery().AggregateTypes("user").EventTypes("user.locked", "user.removed").Builder(). + AddQuery().AggregateTypes("org").EventTypes("org.removed").Builder(). + AddQuery().AggregateTypes("user").EventTypes("user.locked").Builder(), + want: []EventTypeScan{ + {AggregateType: "org", EventType: "org.removed"}, + {AggregateType: "user", EventType: "user.locked"}, + {AggregateType: "user", EventType: "user.removed"}, + }, + }, + { + name: "sub query with several aggregate types", + builder: base().AddQuery().AggregateTypes("user", "org").EventTypes("user.locked", "org.removed").Builder(), + wantErr: errScan, + }, + { + name: "sub query without event types", + builder: base().AddQuery().AggregateTypes("user").Builder(), + wantErr: errScan, + }, + { + name: "sub query with aggregate ids", + builder: base().AddQuery().AggregateTypes("user").AggregateIDs("id").EventTypes("user.locked").Builder(), + wantErr: errScan, + }, + { + name: "sub query with event data", + builder: base().AddQuery().AggregateTypes("user").EventTypes("user.locked").EventData(map[string]any{"key": "value"}).Builder(), + wantErr: errScan, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + query, err := QueryFromBuilder(tt.builder) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, query.EventTypeScans) + }) + } +} diff --git a/internal/eventstore/repository/sql/postgres.go b/internal/eventstore/repository/sql/postgres.go index 8671ea783..99634a784 100644 --- a/internal/eventstore/repository/sql/postgres.go +++ b/internal/eventstore/repository/sql/postgres.go @@ -40,7 +40,10 @@ func NewPostgres(client *database.DB) *Postgres { func (db *Postgres) Health(ctx context.Context) error { return db.Ping() } -const eventSortKeySQL = `"position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence"` +const ( + eventSortKeySQL = `"position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence"` + eventColumnsSQL = `created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order` +) // FilterToReducer finds all events matching the given search query and passes them to the reduce function. func (psql *Postgres) FilterToReducer(ctx context.Context, searchQuery *eventstore.SearchQueryBuilder, reduce eventstore.Reducer) (err error) { @@ -115,20 +118,7 @@ func (db *Postgres) eventQuery(useV1 bool) string { ", aggregate_version" + " FROM eventstore.events" } - return "SELECT" + - " created_at" + - ", event_type" + - `, "sequence"` + - `, "position"` + - ", payload" + - ", creator" + - `, "owner"` + - ", instance_id" + - ", aggregate_type" + - ", aggregate_id" + - ", revision" + - ", in_tx_order" + - " FROM eventstore.events2" + return "SELECT " + eventColumnsSQL + " FROM eventstore.events2" } func (db *Postgres) maxPositionQuery(useV1 bool) string { diff --git a/internal/eventstore/repository/sql/query.go b/internal/eventstore/repository/sql/query.go index 64c670885..147bded79 100644 --- a/internal/eventstore/repository/sql/query.go +++ b/internal/eventstore/repository/sql/query.go @@ -92,39 +92,44 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search where += awaitOpenTransactions(useV1) } - query += where - // instead of using the max function of the database (which doesn't work for postgres) - // we select the most recent row - if q.Columns == eventstore.ColumnsMaxPosition { - q.Limit = 1 - q.Desc = true - } + if canScanEventTypesSeparately(q, useV1) { + query, values = prepareEventTypeScans(q) + } else { + query += where - // if there is only one subquery we can optimize the query ordering by sequence - var shouldOrderBySequence bool - if q.EventSortKeyAfter == nil && len(q.SubQueries) == 1 { - for _, filter := range q.SubQueries[0] { - if filter.Field == repository.FieldAggregateID { - shouldOrderBySequence = filter.Operation == repository.OperationEquals + // instead of using the max function of the database (which doesn't work for postgres) + // we select the most recent row + if q.Columns == eventstore.ColumnsMaxPosition { + q.Limit = 1 + q.Desc = true + } + + // if there is only one subquery we can optimize the query ordering by sequence + var shouldOrderBySequence bool + if q.EventSortKeyAfter == nil && len(q.SubQueries) == 1 { + for _, filter := range q.SubQueries[0] { + if filter.Field == repository.FieldAggregateID { + shouldOrderBySequence = filter.Operation == repository.OperationEquals + } } } - } - switch q.Columns { - case eventstore.ColumnsEvent, - eventstore.ColumnsMaxPosition: - query += criteria.orderByEventSequence(q.Desc, shouldOrderBySequence, useV1) - } + switch q.Columns { + case eventstore.ColumnsEvent, + eventstore.ColumnsMaxPosition: + query += criteria.orderByEventSequence(q.Desc, shouldOrderBySequence, useV1) + } - if q.Limit > 0 { - values = append(values, q.Limit) - query += " LIMIT ?" - } + if q.Limit > 0 { + values = append(values, q.Limit) + query += " LIMIT ?" + } - if q.Offset > 0 { - values = append(values, q.Offset) - query += " OFFSET ?" + if q.Offset > 0 { + values = append(values, q.Offset) + query += " OFFSET ?" + } } query = criteria.placeholder(query) @@ -147,6 +152,68 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search return nil } +// canScanEventTypesSeparately reports if the events of the combinations in [repository.SearchQuery.EventTypeScans] +// can be read one combination at a time: only the instance, the position or sort key and the await of open transactions +// may restrict the query besides the combinations, and it must be ordered ascending with a limit. +func canScanEventTypesSeparately(q *repository.SearchQuery, useV1 bool) bool { + return !useV1 && + len(q.EventTypeScans) > 0 && + q.Columns == eventstore.ColumnsEvent && + q.Limit > 0 && + !q.Desc && + q.Offset == 0 && + q.InstanceID != nil && + q.InstanceIDs == nil && + q.ExcludedInstances == nil && + q.Creator == nil && + q.Owner == nil && + q.Sequence == nil && + q.CreatedAfter == nil && + q.CreatedBefore == nil && + len(q.ExcludeAggregateIDs) == 0 +} + +// prepareEventTypeScans reads the events of every combination of aggregate type and event type +// in sort key order up to the limit, and merges them. +// Each combination is a range of the es_projection index (instance_id, aggregate_type, event_type, position), +// which returns its events in position order, so every scan stops at the limit +// and the query reads at most limit events per combination, however many events match after the position. +func prepareEventTypeScans(q *repository.SearchQuery) (string, []any) { + aggregateTypes := make(database.TextArray[eventstore.AggregateType], len(q.EventTypeScans)) + eventTypes := make(database.TextArray[eventstore.EventType], len(q.EventTypeScans)) + for i, typeScan := range q.EventTypeScans { + aggregateTypes[i] = typeScan.AggregateType + eventTypes[i] = typeScan.EventType + } + values := []any{aggregateTypes, eventTypes, q.InstanceID.Value} + + conditions := "instance_id = ? AND aggregate_type = scans.scan_aggregate_type AND event_type = scans.scan_event_type" + if q.Position != nil { + conditions += ` AND "position" >= ?` + values = append(values, q.Position.Value) + } + if q.EventSortKeyAfter != nil { + conditions += ` AND (` + eventSortKeySQL + `) > (?, ?, ?, ?, ?, ?)` + values = append(values, + q.EventSortKeyAfter.Position, + q.EventSortKeyAfter.InTxOrder, + q.EventSortKeyAfter.InstanceID, + q.EventSortKeyAfter.AggregateType, + q.EventSortKeyAfter.AggregateID, + q.EventSortKeyAfter.Sequence, + ) + } + if q.AwaitOpenTransactions { + conditions += awaitOpenTransactions(false) + } + values = append(values, q.Limit, q.Limit) + + return "SELECT events.* FROM UNNEST(?::TEXT[], ?::TEXT[]) AS scans(scan_aggregate_type, scan_event_type)" + + " CROSS JOIN LATERAL (SELECT " + eventColumnsSQL + " FROM eventstore.events2 WHERE " + conditions + + " ORDER BY " + eventSortKeySQL + " LIMIT ?) AS events" + + " ORDER BY " + eventSortKeySQL + " LIMIT ?", values +} + func prepareColumns(criteria querier, columns eventstore.Columns, useV1 bool) (string, func(s scan, dest interface{}) error) { switch columns { case eventstore.ColumnsMaxPosition: diff --git a/internal/eventstore/repository/sql/query_event_type_scans_test.go b/internal/eventstore/repository/sql/query_event_type_scans_test.go new file mode 100644 index 000000000..89b0e06a7 --- /dev/null +++ b/internal/eventstore/repository/sql/query_event_type_scans_test.go @@ -0,0 +1,233 @@ +package sql + +import ( + "context" + "database/sql/driver" + "fmt" + "regexp" + "testing" + + "github.com/shopspring/decimal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/zitadel/zitadel/internal/database" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/eventstore/repository" + "github.com/zitadel/zitadel/internal/zerrors" +) + +func Test_query_event_type_scans(t *testing.T) { + cursor := eventstore.EventSortKey{ + Position: decimal.NewFromFloat(1.5), + InTxOrder: 5, + InstanceID: "instanceID", + AggregateType: "user", + AggregateID: "agg-a", + Sequence: 5, + } + scanSQL := func(positionCondition string) string { + return `SELECT events.* FROM UNNEST($1::TEXT[], $2::TEXT[]) AS scans(scan_aggregate_type, scan_event_type)` + + ` CROSS JOIN LATERAL (SELECT ` + eventColumnsSQL + ` FROM eventstore.events2` + + ` WHERE instance_id = $3 AND aggregate_type = scans.scan_aggregate_type AND event_type = scans.scan_event_type` + + positionCondition + + ` AND "position" <= EXTRACT(EPOCH FROM now()) ORDER BY ` + eventSortKeySQL + ` LIMIT $%d) AS events` + + ` ORDER BY ` + eventSortKeySQL + ` LIMIT $%d` + } + projectionQuery := func() *eventstore.SearchQueryBuilder { + return eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). + InstanceID("instanceID"). + OrderAsc(). + AwaitOpenTransactions(). + Limit(200). + ScanEventTypesSeparately() + } + aggregateTypes := database.TextArray[eventstore.AggregateType]{"org", "user", "user"} + eventTypes := database.TextArray[eventstore.EventType]{"org.removed", "user.human.added", "user.locked"} + + tests := []struct { + name string + query *eventstore.SearchQueryBuilder + sql string + args []driver.Value + }{ + { + name: "after sort key, one scan per aggregate and event type", + query: projectionQuery(). + AfterEventSortKey(cursor). + AddQuery().AggregateTypes("user").EventTypes("user.locked", "user.human.added").Builder(). + AddQuery().AggregateTypes("org").EventTypes("org.removed").Builder(), + sql: fmt.Sprintf(scanSQL(` AND (`+eventSortKeySQL+`) > ($4, $5, $6, $7, $8, $9)`), 10, 11), + args: []driver.Value{ + aggregateTypes, eventTypes, "instanceID", + cursor.Position, cursor.InTxOrder, cursor.InstanceID, cursor.AggregateType, cursor.AggregateID, cursor.Sequence, + uint64(200), uint64(200), + }, + }, + { + name: "position at least, combinations listed twice are scanned once", + query: projectionQuery(). + PositionAtLeast(decimal.NewFromInt(10)). + AddQuery().AggregateTypes("user").EventTypes("user.locked", "user.human.added").Builder(). + AddQuery().AggregateTypes("org").EventTypes("org.removed").Builder(). + AddQuery().AggregateTypes("user").EventTypes("user.locked").Builder(), + sql: fmt.Sprintf(scanSQL(` AND "position" >= $4`), 5, 6), + args: []driver.Value{ + aggregateTypes, eventTypes, "instanceID", + decimal.NewFromInt(10), + uint64(200), uint64(200), + }, + }, + { + name: "descending order is not scanned separately", + query: eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). + InstanceID("instanceID"). + OrderDesc(). + AwaitOpenTransactions(). + Limit(200). + ScanEventTypesSeparately(). + AddQuery().AggregateTypes("org").EventTypes("org.removed").Builder(), + sql: `SELECT ` + eventColumnsSQL + ` FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND "position" <= EXTRACT(EPOCH FROM now()) ORDER BY (` + eventSortKeySQL + `) DESC LIMIT $4`, + args: []driver.Value{ + "instanceID", eventstore.AggregateType("org"), eventstore.EventType("org.removed"), + uint64(200), + }, + }, + } + + client := NewPostgres(&database.DB{Database: new(testDB)}) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mock := newMockClient(t). + expectExec(regexp.QuoteMeta( + `select pg_advisory_lock('eventstore.events2'::REGCLASS::OID::INTEGER, hashtext($1)), pg_advisory_unlock('eventstore.events2'::REGCLASS::OID::INTEGER, hashtext($1))`), + []driver.Value{"instanceID"}, + ). + expectQuery("^"+regexp.QuoteMeta(tt.sql)+"$", tt.args) + client.DB.DB = mock.client + + require.NoError(t, query(t.Context(), client, tt.query, &[]*repository.Event{}, false)) + assert.NoError(t, mock.mock.ExpectationsWereMet()) + }) + } + + t.Run("sub query filtering by aggregate id returns an error", func(t *testing.T) { + mock := newMockClient(t) + client.DB.DB = mock.client + + err := query(t.Context(), client, + projectionQuery().AddQuery().AggregateTypes("user").AggregateIDs("agg-a").EventTypes("user.locked").Builder(), + &[]*repository.Event{}, false, + ) + require.ErrorIs(t, err, zerrors.ThrowInvalidArgument(nil, "REPO-Ahx6e", "")) + assert.NoError(t, mock.mock.ExpectationsWereMet()) + }) +} + +// Test_query_event_type_scans_postgres checks against a real database that reading every combination of +// aggregate type and event type separately returns exactly the events, in exactly the order, +// of the single query for all event types, from any cursor and for any limit. +// The events include positions shared by several transactions, whose in_tx_order restarts at 1, +// which is the situation where resuming from a cursor has skipped events before. +func Test_query_event_type_scans_postgres(t *testing.T) { + ctx := t.Context() + instanceID := "event-type-scans-" + t.Name() + client := &Postgres{DB: &database.DB{DB: testClient, Database: new(testDB)}} + + t.Cleanup(func() { + _, err := testClient.ExecContext(context.Background(), "DELETE FROM eventstore.events2 WHERE instance_id = $1", instanceID) + assert.NoError(t, err) + }) + + type aggregateEvents struct { + aggregateType eventstore.AggregateType + eventTypes []eventstore.EventType + } + // the last event type of every aggregate is written but not subscribed to + aggregates := []aggregateEvents{ + {"user", []eventstore.EventType{"user.human.added", "user.locked", "user.human.changed"}}, + {"org", []eventstore.EventType{"org.removed", "org.added"}}, + {"session", []eventstore.EventType{"session.added", "session.token.set", "session.terminated"}}, + } + sequences := make(map[string]uint64) + for tx := range 90 { + // every three transactions share a position, each numbering its events from 1 + position := decimal.NewFromInt(1_000_000).Add(decimal.NewFromInt(int64(tx / 3))) + for inTxOrder := range tx%4 + 1 { + agg := aggregates[(tx+inTxOrder)%len(aggregates)] + eventType := agg.eventTypes[(tx*7+inTxOrder)%len(agg.eventTypes)] + aggregateID := fmt.Sprintf("agg-%d", (tx*3+inTxOrder)%5) + key := string(agg.aggregateType) + aggregateID + sequences[key]++ + _, err := testClient.ExecContext(ctx, + `INSERT INTO eventstore.events2 (instance_id, aggregate_type, aggregate_id, event_type, "sequence", revision, created_at, payload, creator, "owner", "position", in_tx_order) + VALUES ($1, $2, $3, $4, $5, 1, now(), NULL, 'creator', 'owner', $6, $7)`, + instanceID, agg.aggregateType, aggregateID, eventType, sequences[key], position, inTxOrder+1, + ) + require.NoError(t, err) + } + } + + subscribe := func(builder *eventstore.SearchQueryBuilder) *eventstore.SearchQueryBuilder { + for _, agg := range aggregates { + builder = builder.AddQuery().AggregateTypes(agg.aggregateType).EventTypes(agg.eventTypes[:len(agg.eventTypes)-1]...).Builder() + } + return builder + } + filter := func(t *testing.T, scanSeparately bool, limit uint64, cursor *eventstore.EventSortKey, minPosition decimal.Decimal) []eventstore.EventSortKey { + builder := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). + InstanceID(instanceID). + AwaitOpenTransactions(). + OrderAsc(). + Limit(limit) + if scanSeparately { + builder = builder.ScanEventTypesSeparately() + } + if cursor != nil { + builder = builder.AfterEventSortKey(*cursor) + } + if !minPosition.IsZero() { + builder = builder.PositionAtLeast(minPosition) + } + keys := []eventstore.EventSortKey{} + err := query(ctx, client, subscribe(builder), eventstore.Reducer(func(event eventstore.Event) error { + keys = append(keys, eventstore.EventSortKey{ + Position: event.Position(), + InTxOrder: event.InTxOrder(), + InstanceID: event.Aggregate().InstanceID, + AggregateType: event.Aggregate().Type, + AggregateID: event.Aggregate().ID, + Sequence: event.Sequence(), + }) + return nil + }), false) + require.NoError(t, err) + return keys + } + + all := filter(t, false, 10_000, nil, decimal.Decimal{}) + require.Len(t, filter(t, true, 10_000, nil, decimal.Decimal{}), len(all)) + require.Greater(t, len(all), 100, "the generated events must include subscribed events") + + for _, limit := range []uint64{1, 2, 3, 7, 50} { + t.Run(fmt.Sprintf("limit %d", limit), func(t *testing.T) { + // from the start + assert.Equal(t, filter(t, false, limit, nil, decimal.Decimal{}), filter(t, true, limit, nil, decimal.Decimal{})) + // from after every event + for i := range all { + cursor := all[i] + want := filter(t, false, limit, &cursor, decimal.Decimal{}) + require.Equal(t, all[i+1:min(len(all), i+1+int(limit))], want, "the single query must page through all events") + assert.Equal(t, want, filter(t, true, limit, &cursor, decimal.Decimal{}), "cursor %d", i) + } + // from every position, inclusive + for i := range all { + assert.Equal(t, + filter(t, false, limit, nil, all[i].Position), + filter(t, true, limit, nil, all[i].Position), + "position %s", all[i].Position, + ) + } + }) + } +} diff --git a/internal/eventstore/search_query.go b/internal/eventstore/search_query.go index 2fe16dcff..1335ab8b9 100644 --- a/internal/eventstore/search_query.go +++ b/internal/eventstore/search_query.go @@ -28,6 +28,7 @@ type SearchQueryBuilder struct { positionAtLeast decimal.Decimal eventSortKeyAfter *EventSortKey awaitOpenTransactions bool + scanEventTypes bool creationDateAfter time.Time creationDateBefore time.Time eventSequenceGreater uint64 @@ -89,6 +90,10 @@ func (b SearchQueryBuilder) GetAwaitOpenTransactions() bool { return b.awaitOpenTransactions } +func (b SearchQueryBuilder) GetScanEventTypesSeparately() bool { + return b.scanEventTypes +} + func (q SearchQueryBuilder) GetEventSequenceGreater() uint64 { return q.eventSequenceGreater } @@ -303,6 +308,19 @@ func (builder *SearchQueryBuilder) AwaitOpenTransactions() *SearchQueryBuilder { return builder } +// ScanEventTypesSeparately reads the events of each combination of aggregate type and event type of the sub queries +// separately, in sort key order up to the limit, and merges the results. +// Without it, a query for multiple types cannot stop reading at the limit: +// it reads every matching event after the position and sorts them, so its cost grows with the number of matching events. +// With it, the cost is bounded by the limit and the number of combinations. +// +// It is only applied to queries for events of a single instance, ordered ascending with a limit, +// whose sub queries filter by aggregate types and event types only. Other queries ignore it. +func (builder *SearchQueryBuilder) ScanEventTypesSeparately() *SearchQueryBuilder { + builder.scanEventTypes = true + return builder +} + // SequenceGreater filters for events with sequence greater the requested sequence func (builder *SearchQueryBuilder) SequenceGreater(sequence uint64) *SearchQueryBuilder { builder.eventSequenceGreater = sequence