From e36f402e093f53b9a8ef614da2e3c77c65cb45f5 Mon Sep 17 00:00:00 2001 From: Silvan <27845747+adlerhurst@users.noreply.github.com> Date: Thu, 13 Mar 2025 16:50:23 +0100 Subject: [PATCH] fix(perf): simplify eventstore queries by removing or in projection handlers (#9530) # Which Problems Are Solved [A recent performance enhancement]((https://github.com/zitadel/zitadel/pull/9497)) aimed at optimizing event store queries, specifically those involving multiple aggregate type filters, has successfully improved index utilization. While the query planner now correctly selects relevant indexes, it employs [bitmap index scans](https://www.postgresql.org/docs/current/indexes-bitmap-scans.html) to retrieve data. This approach, while beneficial in many scenarios, introduces a potential I/O bottleneck. The bitmap index scan first identifies the required database blocks and then utilizes a bitmap to access the corresponding rows from the table's heap. This subsequent "bitmap heap scan" can result in significant I/O overhead, particularly when queries return a substantial number of rows across numerous data pages. ## Impact: Under heavy load or with queries filtering for a wide range of events across multiple aggregate types, this increased I/O activity may lead to: - Increased query latency. - Elevated disk utilization. - Potential performance degradation of the event store and dependent systems. # How the Problems Are Solved To address this I/O bottleneck and further optimize query performance, the projection handler has been modified. Instead of employing multiple OR clauses for each aggregate type, the aggregate and event type filters are now combined using IN ARRAY filters. Technical Details: This change allows the PostgreSQL query planner to leverage [index-only scans](https://www.postgresql.org/docs/current/indexes-index-only-scans.html). By utilizing IN ARRAY filters, the database can efficiently retrieve the necessary data directly from the index, eliminating the need to access the table's heap. This results in: * Reduced I/O: Index-only scans significantly minimize disk I/O operations, as the database avoids reading data pages from the main table. * Improved Query Performance: By reducing I/O, query execution times are substantially improved, leading to lower latency. # Additional Changes - rollback of https://github.com/zitadel/zitadel/pull/9497 # Additional Information ## Query Plan of previous query ```sql SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = '' AND ( ( instance_id = '' AND "position" > AND aggregate_type = 'project' AND event_type = ANY(ARRAY[ 'project.application.added' ,'project.application.changed' ,'project.application.deactivated' ,'project.application.reactivated' ,'project.application.removed' ,'project.removed' ,'project.application.config.api.added' ,'project.application.config.api.changed' ,'project.application.config.api.secret.changed' ,'project.application.config.api.secret.updated' ,'project.application.config.oidc.added' ,'project.application.config.oidc.changed' ,'project.application.config.oidc.secret.changed' ,'project.application.config.oidc.secret.updated' ,'project.application.config.saml.added' ,'project.application.config.saml.changed' ]) ) OR ( instance_id = '' AND "position" > AND aggregate_type = 'org' AND event_type = 'org.removed' ) OR ( instance_id = '' AND "position" > AND aggregate_type = 'instance' AND event_type = 'instance.removed' ) ) AND "position" > 1741600905.3495 AND "position" < ( SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(ARRAY['zitadel_es_pusher_', 'zitadel_es_pusher', 'zitadel_es_pusher_']) AND state <> 'idle' ) ORDER BY "position", in_tx_order LIMIT 200 OFFSET 1; ``` ``` Limit (cost=120.08..120.09 rows=7 width=361) (actual time=2.167..2.172 rows=0 loops=1) Output: events2.created_at, events2.event_type, events2.sequence, events2."position", events2.payload, events2.creator, events2.owner, events2.instance_id, events2.aggregate_type, events2.aggregate_id, events2.revision, events2.in_tx_order InitPlan 1 -> Aggregate (cost=2.74..2.76 rows=1 width=32) (actual time=1.813..1.815 rows=1 loops=1) Output: COALESCE(EXTRACT(epoch FROM min(s.xact_start)), EXTRACT(epoch FROM now())) -> Nested Loop (cost=0.00..2.74 rows=1 width=8) (actual time=1.803..1.805 rows=0 loops=1) Output: s.xact_start Join Filter: (d.oid = s.datid) -> Seq Scan on pg_catalog.pg_database d (cost=0.00..1.07 rows=1 width=4) (actual time=0.016..0.021 rows=1 loops=1) Output: d.oid, d.datname, d.datdba, d.encoding, d.datlocprovider, d.datistemplate, d.datallowconn, d.dathasloginevt, d.datconnlimit, d.datfrozenxid, d.datminmxid, d.dattablespace, d.datcollate, d.datctype, d.datlocale, d.daticurules, d.datcollversion, d.datacl Filter: (d.datname = current_database()) Rows Removed by Filter: 4 -> Function Scan on pg_catalog.pg_stat_get_activity s (cost=0.00..1.63 rows=3 width=16) (actual time=1.781..1.781 rows=0 loops=1) Output: s.datid, s.pid, s.usesysid, s.application_name, s.state, s.query, s.wait_event_type, s.wait_event, s.xact_start, s.query_start, s.backend_start, s.state_change, s.client_addr, s.client_hostname, s.client_port, s.backend_xid, s.backend_xmin, s.backend_type, s.ssl, s.sslversion, s.sslcipher, s.sslbits, s.ssl_client_dn, s.ssl_client_serial, s.ssl_issuer_dn, s.gss_auth, s.gss_princ, s.gss_enc, s.gss_delegation, s.leader_pid, s.query_id Function Call: pg_stat_get_activity(NULL::integer) Filter: ((s.state <> 'idle'::text) AND (s.application_name = ANY ('{zitadel_es_pusher_,zitadel_es_pusher,zitadel_es_pusher_}'::text[]))) Rows Removed by Filter: 49 -> Sort (cost=117.31..117.33 rows=8 width=361) (actual time=2.167..2.168 rows=0 loops=1) Output: events2.created_at, events2.event_type, events2.sequence, events2."position", events2.payload, events2.creator, events2.owner, events2.instance_id, events2.aggregate_type, events2.aggregate_id, events2.revision, events2.in_tx_order Sort Key: events2."position", events2.in_tx_order Sort Method: quicksort Memory: 25kB -> Bitmap Heap Scan on eventstore.events2 (cost=84.92..117.19 rows=8 width=361) (actual time=2.088..2.089 rows=0 loops=1) Output: events2.created_at, events2.event_type, events2.sequence, events2."position", events2.payload, events2.creator, events2.owner, events2.instance_id, events2.aggregate_type, events2.aggregate_id, events2.revision, events2.in_tx_order Recheck Cond: (((events2.instance_id = ''::text) AND (events2.aggregate_type = 'project'::text) AND (events2.event_type = ANY ('{project.application.added,project.application.changed,project.application.deactivated,project.application.reactivated,project.application.removed,project.removed,project.application.config.api.added,project.application.config.api.changed,project.application.config.api.secret.changed,project.application.config.api.secret.updated,project.application.config.oidc.added,project.application.config.oidc.changed,project.application.config.oidc.secret.changed,project.application.config.oidc.secret.updated,project.application.config.saml.added,project.application.config.saml.changed}'::text[])) AND (events2."position" > ) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) OR ((events2.instance_id = ''::text) AND (events2.aggregate_type = 'org'::text) AND (events2.event_type = 'org.removed'::text) AND (events2."position" > ) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) OR ((events2.instance_id = ''::text) AND (events2.aggregate_type = 'instance'::text) AND (events2.event_type = 'instance.removed'::text) AND (events2."position" > ) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1))) -> BitmapOr (cost=84.88..84.88 rows=8 width=0) (actual time=2.080..2.081 rows=0 loops=1) -> Bitmap Index Scan on es_projection (cost=0.00..75.44 rows=8 width=0) (actual time=2.016..2.017 rows=0 loops=1) Index Cond: ((events2.instance_id = ''::text) AND (events2.aggregate_type = 'project'::text) AND (events2.event_type = ANY ('{project.application.added,project.application.changed,project.application.deactivated,project.application.reactivated,project.application.removed,project.removed,project.application.config.api.added,project.application.config.api.changed,project.application.config.api.secret.changed,project.application.config.api.secret.updated,project.application.config.oidc.added,project.application.config.oidc.changed,project.application.config.oidc.secret.changed,project.application.config.oidc.secret.updated,project.application.config.saml.added,project.application.config.saml.changed}'::text[])) AND (events2."position" > ) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) -> Bitmap Index Scan on es_projection (cost=0.00..4.71 rows=1 width=0) (actual time=0.016..0.016 rows=0 loops=1) Index Cond: ((events2.instance_id = ''::text) AND (events2.aggregate_type = 'org'::text) AND (events2.event_type = 'org.removed'::text) AND (events2."position" > ) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) -> Bitmap Index Scan on es_projection (cost=0.00..4.71 rows=1 width=0) (actual time=0.045..0.045 rows=0 loops=1) Index Cond: ((events2.instance_id = ''::text) AND (events2.aggregate_type = 'instance'::text) AND (events2.event_type = 'instance.removed'::text) AND (events2."position" > ) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) Query Identifier: 3194938266011254479 Planning Time: 1.295 ms Execution Time: 2.832 ms ``` ## Query Plan of new query ```sql SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = '' AND "position" > AND aggregate_type = ANY(ARRAY['project', 'instance', 'org']) AND event_type = ANY(ARRAY[ 'project.application.added' ,'project.application.changed' ,'project.application.deactivated' ,'project.application.reactivated' ,'project.application.removed' ,'project.removed' ,'project.application.config.api.added' ,'project.application.config.api.changed' ,'project.application.config.api.secret.changed' ,'project.application.config.api.secret.updated' ,'project.application.config.oidc.added' ,'project.application.config.oidc.changed' ,'project.application.config.oidc.secret.changed' ,'project.application.config.oidc.secret.updated' ,'project.application.config.saml.added' ,'project.application.config.saml.changed' ,'org.removed' ,'instance.removed' ]) AND "position" < ( SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(ARRAY['zitadel_es_pusher_', 'zitadel_es_pusher', 'zitadel_es_pusher_']) AND state <> 'idle' ) ORDER BY "position", in_tx_order LIMIT 200 OFFSET 1; ``` ``` Limit (cost=293.34..293.36 rows=8 width=361) (actual time=4.686..4.689 rows=0 loops=1) Output: events2.created_at, events2.event_type, events2.sequence, events2."position", events2.payload, events2.creator, events2.owner, events2.instance_id, events2.aggregate_type, events2.aggregate_id, events2.revision, events2.in_tx_order InitPlan 1 -> Aggregate (cost=2.74..2.76 rows=1 width=32) (actual time=1.717..1.719 rows=1 loops=1) Output: COALESCE(EXTRACT(epoch FROM min(s.xact_start)), EXTRACT(epoch FROM now())) -> Nested Loop (cost=0.00..2.74 rows=1 width=8) (actual time=1.658..1.659 rows=0 loops=1) Output: s.xact_start Join Filter: (d.oid = s.datid) -> Seq Scan on pg_catalog.pg_database d (cost=0.00..1.07 rows=1 width=4) (actual time=0.026..0.028 rows=1 loops=1) Output: d.oid, d.datname, d.datdba, d.encoding, d.datlocprovider, d.datistemplate, d.datallowconn, d.dathasloginevt, d.datconnlimit, d.datfrozenxid, d.datminmxid, d.dattablespace, d.datcollate, d.datctype, d.datlocale, d.daticurules, d.datcollversion, d.datacl Filter: (d.datname = current_database()) Rows Removed by Filter: 4 -> Function Scan on pg_catalog.pg_stat_get_activity s (cost=0.00..1.63 rows=3 width=16) (actual time=1.628..1.628 rows=0 loops=1) Output: s.datid, s.pid, s.usesysid, s.application_name, s.state, s.query, s.wait_event_type, s.wait_event, s.xact_start, s.query_start, s.backend_start, s.state_change, s.client_addr, s.client_hostname, s.client_port, s.backend_xid, s.backend_xmin, s.backend_type, s.ssl, s.sslversion, s.sslcipher, s.sslbits, s.ssl_client_dn, s.ssl_client_serial, s.ssl_issuer_dn, s.gss_auth, s.gss_princ, s.gss_enc, s.gss_delegation, s.leader_pid, s.query_id Function Call: pg_stat_get_activity(NULL::integer) Filter: ((s.state <> 'idle'::text) AND (s.application_name = ANY ('{zitadel_es_pusher_,zitadel_es_pusher,zitadel_es_pusher_}'::text[]))) Rows Removed by Filter: 42 -> Sort (cost=290.58..290.60 rows=9 width=361) (actual time=4.685..4.685 rows=0 loops=1) Output: events2.created_at, events2.event_type, events2.sequence, events2."position", events2.payload, events2.creator, events2.owner, events2.instance_id, events2.aggregate_type, events2.aggregate_id, events2.revision, events2.in_tx_order Sort Key: events2."position", events2.in_tx_order Sort Method: quicksort Memory: 25kB -> Index Scan using es_projection on eventstore.events2 (cost=0.70..290.43 rows=9 width=361) (actual time=4.616..4.617 rows=0 loops=1) Output: events2.created_at, events2.event_type, events2.sequence, events2."position", events2.payload, events2.creator, events2.owner, events2.instance_id, events2.aggregate_type, events2.aggregate_id, events2.revision, events2.in_tx_order Index Cond: ((events2.instance_id = ''::text) AND (events2.aggregate_type = ANY ('{project,instance,org}'::text[])) AND (events2.event_type = ANY ('{project.application.added,project.application.changed,project.application.deactivated,project.application.reactivated,project.application.removed,project.removed,project.application.config.api.added,project.application.config.api.changed,project.application.config.api.secret.changed,project.application.config.api.secret.updated,project.application.config.oidc.added,project.application.config.oidc.changed,project.application.config.oidc.secret.changed,project.application.config.oidc.secret.updated,project.application.config.saml.added,project.application.config.saml.changed,org.removed,instance.removed}'::text[])) AND (events2."position" > ) AND (events2."position" < (InitPlan 1).col1)) Query Identifier: -8254550537132386499 Planning Time: 2.864 ms Execution Time: 5.414 ms ``` --- internal/api/oidc/key.go | 9 +-- internal/api/saml/certificate.go | 9 +-- .../command/existing_label_policies_model.go | 55 --------------- internal/eventstore/handler/v2/handler.go | 14 ++-- internal/eventstore/repository/sql/crdb.go | 8 +-- internal/eventstore/repository/sql/query.go | 70 +++++++------------ .../eventstore/repository/sql/query_test.go | 8 +-- 7 files changed, 50 insertions(+), 123 deletions(-) delete mode 100644 internal/command/existing_label_policies_model.go diff --git a/internal/api/oidc/key.go b/internal/api/oidc/key.go index 6c0599f556..76f78ab5ab 100644 --- a/internal/api/oidc/key.go +++ b/internal/api/oidc/key.go @@ -419,13 +419,14 @@ func (o *OPStorage) getMaxKeySequence(ctx context.Context) (float64, error) { AwaitOpenTransactions(). AllowTimeTravel(). AddQuery(). - AggregateTypes(keypair.AggregateType). + AggregateTypes( + keypair.AggregateType, + instance.AggregateType, + ). EventTypes( keypair.AddedEventType, + instance.InstanceRemovedEventType, ). - Or(). - AggregateTypes(instance.AggregateType). - EventTypes(instance.InstanceRemovedEventType). Builder(), ) } diff --git a/internal/api/saml/certificate.go b/internal/api/saml/certificate.go index 2eac0e4d36..ff130f7709 100644 --- a/internal/api/saml/certificate.go +++ b/internal/api/saml/certificate.go @@ -157,14 +157,15 @@ func (p *Storage) getMaxKeySequence(ctx context.Context) (float64, error) { ResourceOwner(authz.GetInstance(ctx).InstanceID()). AwaitOpenTransactions(). AddQuery(). - AggregateTypes(keypair.AggregateType). + AggregateTypes( + keypair.AggregateType, + instance.AggregateType, + ). EventTypes( keypair.AddedEventType, keypair.AddedCertificateEventType, + instance.InstanceRemovedEventType, ). - Or(). - AggregateTypes(instance.AggregateType). - EventTypes(instance.InstanceRemovedEventType). Builder(), ) } diff --git a/internal/command/existing_label_policies_model.go b/internal/command/existing_label_policies_model.go deleted file mode 100644 index dda39980f1..0000000000 --- a/internal/command/existing_label_policies_model.go +++ /dev/null @@ -1,55 +0,0 @@ -package command - -import ( - "context" - - "github.com/zitadel/zitadel/internal/eventstore" - "github.com/zitadel/zitadel/internal/repository/instance" - "github.com/zitadel/zitadel/internal/repository/org" -) - -type ExistingLabelPoliciesReadModel struct { - eventstore.WriteModel - - aggregateIDs []string -} - -func NewExistingLabelPoliciesReadModel(ctx context.Context) *ExistingLabelPoliciesReadModel { - return &ExistingLabelPoliciesReadModel{} -} - -func (rm *ExistingLabelPoliciesReadModel) AppendEvents(events ...eventstore.Event) { - rm.WriteModel.AppendEvents(events...) -} - -func (rm *ExistingLabelPoliciesReadModel) Reduce() error { - for _, event := range rm.Events { - switch e := event.(type) { - case *instance.LabelPolicyAddedEvent, - *org.LabelPolicyAddedEvent: - rm.aggregateIDs = append(rm.aggregateIDs, e.Aggregate().ID) - case *org.LabelPolicyRemovedEvent: - for i := len(rm.aggregateIDs) - 1; i >= 0; i-- { - if rm.aggregateIDs[i] == e.Aggregate().ID { - copy(rm.aggregateIDs[i:], rm.aggregateIDs[i+1:]) - rm.aggregateIDs[len(rm.aggregateIDs)-1] = "" - rm.aggregateIDs = rm.aggregateIDs[:len(rm.aggregateIDs)-1] - } - } - } - } - return rm.WriteModel.Reduce() -} - -func (rm *ExistingLabelPoliciesReadModel) Query() *eventstore.SearchQueryBuilder { - return eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). - AddQuery(). - AggregateTypes(instance.AggregateType). - EventTypes(instance.LabelPolicyAddedEventType). - Or(). - AggregateTypes(org.AggregateType). - EventTypes( - org.LabelPolicyAddedEventType, - org.LabelPolicyRemovedEventType). - Builder() -} diff --git a/internal/eventstore/handler/v2/handler.go b/internal/eventstore/handler/v2/handler.go index 03805f360b..052f965e22 100644 --- a/internal/eventstore/handler/v2/handler.go +++ b/internal/eventstore/handler/v2/handler.go @@ -662,15 +662,15 @@ func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder } } - for aggregateType, eventTypes := range h.eventTypes { - builder = builder. - AddQuery(). - AggregateTypes(aggregateType). - EventTypes(eventTypes...). - 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 + return builder.AddQuery().AggregateTypes(aggregateTypes...).EventTypes(eventTypes...).Builder() } // ProjectionName returns the name of the underlying projection. diff --git a/internal/eventstore/repository/sql/crdb.go b/internal/eventstore/repository/sql/crdb.go index 9128b2c144..68610676c3 100644 --- a/internal/eventstore/repository/sql/crdb.go +++ b/internal/eventstore/repository/sql/crdb.go @@ -124,11 +124,11 @@ type CRDB struct { func NewCRDB(client *database.DB) *CRDB { switch client.Type() { case "cockroach": - awaitOpenTransactionsV1 = "creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))" - awaitOpenTransactionsV2 = `hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))` + awaitOpenTransactionsV1 = " AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))" + awaitOpenTransactionsV2 = ` AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))` case "postgres": - awaitOpenTransactionsV1 = `EXTRACT(EPOCH FROM created_at) < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(?) AND state <> 'idle')` - awaitOpenTransactionsV2 = `"position" < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(?) AND state <> 'idle')` + awaitOpenTransactionsV1 = ` AND EXTRACT(EPOCH FROM created_at) < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(?) AND state <> 'idle')` + awaitOpenTransactionsV2 = ` AND "position" < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(?) AND state <> 'idle')` } return &CRDB{client} diff --git a/internal/eventstore/repository/sql/query.go b/internal/eventstore/repository/sql/query.go index 4dc316440a..4e1cc87aff 100644 --- a/internal/eventstore/repository/sql/query.go +++ b/internal/eventstore/repository/sql/query.go @@ -32,7 +32,7 @@ type querier interface { dialect.Database } -type scan func(dest ...any) error +type scan func(dest ...interface{}) error type tx struct { *sql.Tx @@ -54,7 +54,7 @@ func (t *tx) QueryContext(ctx context.Context, scan func(rows *sql.Rows) error, return rows.Err() } -func query(ctx context.Context, criteria querier, searchQuery *eventstore.SearchQueryBuilder, dest any, useV1 bool) error { +func query(ctx context.Context, criteria querier, searchQuery *eventstore.SearchQueryBuilder, dest interface{}, useV1 bool) error { q, err := repository.QueryFromBuilder(searchQuery) if err != nil { return err @@ -120,7 +120,7 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search query = criteria.placeholder(query) var contextQuerier interface { - QueryContext(context.Context, func(rows *sql.Rows) error, string, ...any) error + QueryContext(context.Context, func(rows *sql.Rows) error, string, ...interface{}) error } contextQuerier = criteria.Client() if q.Tx != nil { @@ -145,7 +145,7 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search return nil } -func prepareColumns(criteria querier, columns eventstore.Columns, useV1 bool) (string, func(s scan, dest any) error) { +func prepareColumns(criteria querier, columns eventstore.Columns, useV1 bool) (string, func(s scan, dest interface{}) error) { switch columns { case eventstore.ColumnsMaxSequence: return criteria.maxSequenceQuery(useV1), maxSequenceScanner @@ -166,7 +166,7 @@ func prepareTimeTravel(ctx context.Context, criteria querier, allow bool) string return criteria.Timetravel(took) } -func maxSequenceScanner(row scan, dest any) (err error) { +func maxSequenceScanner(row scan, dest interface{}) (err error) { position, ok := dest.(*sql.NullFloat64) if !ok { return zerrors.ThrowInvalidArgumentf(nil, "SQL-NBjA9", "type must be sql.NullInt64 got: %T", dest) @@ -178,7 +178,7 @@ func maxSequenceScanner(row scan, dest any) (err error) { return zerrors.ThrowInternal(err, "SQL-bN5xg", "something went wrong") } -func instanceIDsScanner(scanner scan, dest any) (err error) { +func instanceIDsScanner(scanner scan, dest interface{}) (err error) { ids, ok := dest.(*[]string) if !ok { return zerrors.ThrowInvalidArgument(nil, "SQL-Begh2", "type must be an array of string") @@ -194,8 +194,8 @@ func instanceIDsScanner(scanner scan, dest any) (err error) { return nil } -func eventsScanner(useV1 bool) func(scanner scan, dest any) (err error) { - return func(scanner scan, dest any) (err error) { +func eventsScanner(useV1 bool) func(scanner scan, dest interface{}) (err error) { + return func(scanner scan, dest interface{}) (err error) { reduce, ok := dest.(eventstore.Reducer) if !ok { return zerrors.ThrowInvalidArgumentf(nil, "SQL-4GP6F", "events scanner: invalid type %T", dest) @@ -243,17 +243,14 @@ func eventsScanner(useV1 bool) func(scanner scan, dest any) (err error) { } } -func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bool) (clauses string, args []any) { - if len(query.SubQueries) != 1 { - clauses, args = prepareQuery(criteria, useV1, query.InstanceID, query.InstanceIDs, query.ExcludedInstances) - if clauses != "" && len(query.SubQueries) > 0 { - clauses += " AND " - } +func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bool) (_ string, args []any) { + clauses, args := prepareQuery(criteria, useV1, query.InstanceID, query.InstanceIDs, query.ExcludedInstances) + if clauses != "" && len(query.SubQueries) > 0 { + clauses += " AND " } subClauses := make([]string, len(query.SubQueries)) for i, filters := range query.SubQueries { var subArgs []any - filters = append([]*repository.Filter{query.InstanceID, query.InstanceIDs, query.ExcludedInstances, query.Position}, filters...) subClauses[i], subArgs = prepareQuery(criteria, useV1, filters...) // an error is thrown in [query] if subClauses[i] == "" { @@ -270,19 +267,14 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo clauses += "(" + strings.Join(subClauses, " OR ") + ")" } - filters := make([]*repository.Filter, 0, 6) - if len(subClauses) != 1 { - filters = append(filters, query.Position) - } - filters = append(filters, + additionalClauses, additionalArgs := prepareQuery(criteria, useV1, + query.Position, query.Owner, query.Sequence, query.CreatedAfter, query.CreatedBefore, query.Creator, ) - - additionalClauses, additionalArgs := prepareQuery(criteria, useV1, filters...) if additionalClauses != "" { if clauses != "" { clauses += " AND " @@ -291,12 +283,20 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo args = append(args, additionalArgs...) } - excludeAggregateIDsClause, excludeAggregateIDsArgs := excludeAggregateIDs(criteria, query, useV1) - if excludeAggregateIDsClause != "" { + excludeAggregateIDs := query.ExcludeAggregateIDs + if len(excludeAggregateIDs) > 0 { + excludeAggregateIDs = append(excludeAggregateIDs, query.InstanceID, query.InstanceIDs, query.Position, query.CreatedAfter, query.CreatedBefore) + } + excludeAggregateIDsClauses, excludeAggregateIDsArgs := prepareQuery(criteria, useV1, excludeAggregateIDs...) + if excludeAggregateIDsClauses != "" { if clauses != "" { clauses += " AND " } - clauses += excludeAggregateIDsClause + if useV1 { + clauses += "aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events WHERE " + excludeAggregateIDsClauses + ")" + } else { + clauses += "aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE " + excludeAggregateIDsClauses + ")" + } args = append(args, excludeAggregateIDsArgs...) } @@ -312,9 +312,6 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo instanceIDs[i] = "zitadel_es_pusher_" + instanceIDs[i] } - if clauses != "" { - clauses += " AND " - } clauses += awaitOpenTransactions(useV1) args = append(args, instanceIDs) } @@ -326,23 +323,6 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo return " WHERE " + clauses, args } -func excludeAggregateIDs(criteria querier, query *repository.SearchQuery, useV1 bool) (clause string, args []any) { - excludeAggregateIDs := query.ExcludeAggregateIDs - if len(excludeAggregateIDs) > 0 { - excludeAggregateIDs = append(excludeAggregateIDs, query.InstanceID, query.InstanceIDs, query.Position, query.CreatedAfter, query.CreatedBefore) - } - excludeAggregateIDsClauses, excludeAggregateIDsArgs := prepareQuery(criteria, useV1, excludeAggregateIDs...) - if excludeAggregateIDsClauses == "" { - return "", nil - } - if useV1 { - clause = "aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events WHERE " + excludeAggregateIDsClauses + ")" - } else { - clause = "aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE " + excludeAggregateIDsClauses + ")" - } - return clause, excludeAggregateIDsArgs -} - func prepareQuery(criteria querier, useV1 bool, filters ...*repository.Filter) (_ string, args []any) { clauses := make([]string, 0, len(filters)) args = make([]any, 0, len(filters)) diff --git a/internal/eventstore/repository/sql/query_test.go b/internal/eventstore/repository/sql/query_test.go index db98b5997f..abac19ead0 100644 --- a/internal/eventstore/repository/sql/query_test.go +++ b/internal/eventstore/repository/sql/query_test.go @@ -1020,9 +1020,9 @@ func Test_query_events_mocked(t *testing.T) { fields: fields{ mock: newMockClient(t).expectQuery(t, regexp.QuoteMeta( - `SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE instance_id = $1 AND "position" > $2 AND aggregate_type = $3 AND event_type = $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND "position" > $8) ORDER BY event_sequence DESC LIMIT $9`, + `SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND "position" > $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND "position" > $8) ORDER BY event_sequence DESC LIMIT $9`, ), - []driver.Value{"instanceID", 123.456, eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", 123.456, uint64(5)}, + []driver.Value{"instanceID", eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), 123.456, eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", 123.456, uint64(5)}, ), }, res: res{ @@ -1051,9 +1051,9 @@ func Test_query_events_mocked(t *testing.T) { fields: fields{ mock: newMockClient(t).expectQuery(t, regexp.QuoteMeta( - `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND "position" > $2 AND aggregate_type = $3 AND event_type = $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND "position" > $8) ORDER BY "position" DESC, in_tx_order DESC LIMIT $9`, + `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND "position" > $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND "position" > $8) ORDER BY "position" DESC, in_tx_order DESC LIMIT $9`, ), - []driver.Value{"instanceID", 123.456, eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", 123.456, uint64(5)}, + []driver.Value{"instanceID", eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), 123.456, eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", 123.456, uint64(5)}, ), }, res: res{