TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-04-23 06:41:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(perf): simplify eventstore queries by removing or in projection handlers (#9530)

Browse Source 
# 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 = '<INSTANCE_ID>'
    AND (
        (
            instance_id = '<INSTANCE_ID>'
            AND "position" > <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 = '<INSTANCE_ID>'
            AND "position" > <POSITION>
            AND aggregate_type = 'org'
            AND event_type = 'org.removed'
        ) OR (
            instance_id = '<INSTANCE_ID>'
            AND "position" > <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_<INSTANCE_ID>']) 
            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_<INSTANCE_ID>}'::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 = '<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" > <POSITION>) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) OR ((events2.instance_id = '<INSTANCE_ID>'::text) AND (events2.aggregate_type = 'org'::text) AND (events2.event_type = 'org.removed'::text) AND (events2."position" > <POSITION>) AND (events2."position" > 1741600905.3495) AND (events2."position" < (InitPlan 1).col1)) OR ((events2.instance_id = '<INSTANCE_ID>'::text) AND (events2.aggregate_type = 'instance'::text) AND (events2.event_type = 'instance.removed'::text) AND (events2."position" > <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 = '<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" > <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 = '<INSTANCE_ID>'::text) AND (events2.aggregate_type = 'org'::text) AND (events2.event_type = 'org.removed'::text) AND (events2."position" > <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 = '<INSTANCE_ID>'::text) AND (events2.aggregate_type = 'instance'::text) AND (events2.event_type = 'instance.removed'::text) AND (events2."position" > <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 = '<INSTANCE_ID>'
    AND "position" > <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_<INSTANCE_ID>']) 
            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_<INSTANCE_ID>}'::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 = '<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" > <POSITION>) AND (events2."position" < (InitPlan 1).col1))
 Query Identifier: -8254550537132386499
 Planning Time: 2.864 ms
 Execution Time: 5.414 ms
 ```
This commit is contained in:
Silvan 2025-03-13 16:50:23 +01:00 committed by GitHub
parent 5eb3a543e8
commit e36f402e09
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 50 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/api/oidc/key.go
View File

@ -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(),
)
}

9
internal/api/saml/certificate.go
View File

@ -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(),
)
}

55
internal/command/existing_label_policies_model.go
View File

@ -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()
}

14
internal/eventstore/handler/v2/handler.go
View File

@ -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.

8
internal/eventstore/repository/sql/crdb.go
View File

@ -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}

66
internal/eventstore/repository/sql/query.go
View File

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

8
internal/eventstore/repository/sql/query_test.go
View File

@ -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{