TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-07-11 17:38:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(eventstore): optimise query hints for event filters (#9497)

Browse Source 
(cherry picked from commit b5781371395a1f4b7ad7ae8c5ff567193f71b4d2)
This commit is contained in:
Silvan 2025-03-12 10:44:06 +01:00 committed by Livio Spring
parent a47f4a30fa
commit e770569413
No known key found for this signature in database
3 changed files with 53 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -124,11 +124,11 @@ type CRDB struct {
func NewCRDB(client *database.DB) *CRDB { func NewCRDB(client *database.DB) *CRDB {
switch client.Type() { switch client.Type() {
case "cockroach": case "cockroach":
awaitOpenTransactionsV1 = " AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))" awaitOpenTransactionsV1 = "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(?))` awaitOpenTransactionsV2 = `hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))`
case "postgres": case "postgres":
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')` 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 = ` 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')` 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')`
} }
return &CRDB{client} return &CRDB{client}

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

@ -32,7 +32,7 @@ type querier interface {
dialect.Database dialect.Database
} }
type scan func(dest ...interface{}) error type scan func(dest ...any) error
type tx struct { type tx struct {
*sql.Tx *sql.Tx
@ -54,7 +54,7 @@ func (t *tx) QueryContext(ctx context.Context, scan func(rows *sql.Rows) error,
return rows.Err() return rows.Err()
} }
func query(ctx context.Context, criteria querier, searchQuery *eventstore.SearchQueryBuilder, dest interface{}, useV1 bool) error { func query(ctx context.Context, criteria querier, searchQuery *eventstore.SearchQueryBuilder, dest any, useV1 bool) error {
q, err := repository.QueryFromBuilder(searchQuery) q, err := repository.QueryFromBuilder(searchQuery)
if err != nil { if err != nil {
return err return err
@ -120,7 +120,7 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search
query = criteria.placeholder(query) query = criteria.placeholder(query)
var contextQuerier interface { var contextQuerier interface {
QueryContext(context.Context, func(rows *sql.Rows) error, string, ...interface{}) error QueryContext(context.Context, func(rows *sql.Rows) error, string, ...any) error
} }
contextQuerier = criteria.Client() contextQuerier = criteria.Client()
if q.Tx != nil { if q.Tx != nil {
@ -145,7 +145,7 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search
return nil return nil
} }
func prepareColumns(criteria querier, columns eventstore.Columns, useV1 bool) (string, func(s scan, dest interface{}) error) { func prepareColumns(criteria querier, columns eventstore.Columns, useV1 bool) (string, func(s scan, dest any) error) {
switch columns { switch columns {
case eventstore.ColumnsMaxSequence: case eventstore.ColumnsMaxSequence:
return criteria.maxSequenceQuery(useV1), maxSequenceScanner return criteria.maxSequenceQuery(useV1), maxSequenceScanner
@ -166,7 +166,7 @@ func prepareTimeTravel(ctx context.Context, criteria querier, allow bool) string
return criteria.Timetravel(took) return criteria.Timetravel(took)
} }
func maxSequenceScanner(row scan, dest interface{}) (err error) { func maxSequenceScanner(row scan, dest any) (err error) {
position, ok := dest.(*sql.NullFloat64) position, ok := dest.(*sql.NullFloat64)
if !ok { if !ok {
return zerrors.ThrowInvalidArgumentf(nil, "SQL-NBjA9", "type must be sql.NullInt64 got: %T", dest) return zerrors.ThrowInvalidArgumentf(nil, "SQL-NBjA9", "type must be sql.NullInt64 got: %T", dest)
@ -178,7 +178,7 @@ func maxSequenceScanner(row scan, dest interface{}) (err error) {
return zerrors.ThrowInternal(err, "SQL-bN5xg", "something went wrong") return zerrors.ThrowInternal(err, "SQL-bN5xg", "something went wrong")
} }
func instanceIDsScanner(scanner scan, dest interface{}) (err error) { func instanceIDsScanner(scanner scan, dest any) (err error) {
ids, ok := dest.(*[]string) ids, ok := dest.(*[]string)
if !ok { if !ok {
return zerrors.ThrowInvalidArgument(nil, "SQL-Begh2", "type must be an array of string") return zerrors.ThrowInvalidArgument(nil, "SQL-Begh2", "type must be an array of string")
@ -194,8 +194,8 @@ func instanceIDsScanner(scanner scan, dest interface{}) (err error) {
return nil return nil
} }
func eventsScanner(useV1 bool) func(scanner scan, dest interface{}) (err error) { func eventsScanner(useV1 bool) func(scanner scan, dest any) (err error) {
return func(scanner scan, dest interface{}) (err error) { return func(scanner scan, dest any) (err error) {
reduce, ok := dest.(eventstore.Reducer) reduce, ok := dest.(eventstore.Reducer)
if !ok { if !ok {
return zerrors.ThrowInvalidArgumentf(nil, "SQL-4GP6F", "events scanner: invalid type %T", dest) return zerrors.ThrowInvalidArgumentf(nil, "SQL-4GP6F", "events scanner: invalid type %T", dest)
@ -243,14 +243,17 @@ func eventsScanner(useV1 bool) func(scanner scan, dest interface{}) (err error)
} }
} }
func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bool) (_ string, args []any) { func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bool) (clauses string, args []any) {
clauses, args := prepareQuery(criteria, useV1, query.InstanceID, query.InstanceIDs, query.ExcludedInstances) if len(query.SubQueries) != 1 {
if clauses != "" && len(query.SubQueries) > 0 { clauses, args = prepareQuery(criteria, useV1, query.InstanceID, query.InstanceIDs, query.ExcludedInstances)
clauses += " AND " if clauses != "" && len(query.SubQueries) > 0 {
clauses += " AND "
}
} }
subClauses := make([]string, len(query.SubQueries)) subClauses := make([]string, len(query.SubQueries))
for i, filters := range query.SubQueries { for i, filters := range query.SubQueries {
var subArgs []any var subArgs []any
filters = append([]*repository.Filter{query.InstanceID, query.InstanceIDs, query.ExcludedInstances, query.Position}, filters...)
subClauses[i], subArgs = prepareQuery(criteria, useV1, filters...) subClauses[i], subArgs = prepareQuery(criteria, useV1, filters...)
// an error is thrown in [query] // an error is thrown in [query]
if subClauses[i] == "" { if subClauses[i] == "" {
@ -267,14 +270,19 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo
clauses += "(" + strings.Join(subClauses, " OR ") + ")" clauses += "(" + strings.Join(subClauses, " OR ") + ")"
} }
additionalClauses, additionalArgs := prepareQuery(criteria, useV1, filters := make([]*repository.Filter, 0, 6)
query.Position, if len(subClauses) != 1 {
filters = append(filters, query.Position)
}
filters = append(filters,
query.Owner, query.Owner,
query.Sequence, query.Sequence,
query.CreatedAfter, query.CreatedAfter,
query.CreatedBefore, query.CreatedBefore,
query.Creator, query.Creator,
) )
additionalClauses, additionalArgs := prepareQuery(criteria, useV1, filters...)
if additionalClauses != "" { if additionalClauses != "" {
if clauses != "" { if clauses != "" {
clauses += " AND " clauses += " AND "
@ -283,20 +291,12 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo
args = append(args, additionalArgs...) args = append(args, additionalArgs...)
} }
excludeAggregateIDs := query.ExcludeAggregateIDs excludeAggregateIDsClause, excludeAggregateIDsArgs := excludeAggregateIDs(criteria, query, useV1)
if len(excludeAggregateIDs) > 0 { if excludeAggregateIDsClause != "" {
excludeAggregateIDs = append(excludeAggregateIDs, query.InstanceID, query.InstanceIDs, query.Position, query.CreatedAfter, query.CreatedBefore)
}
excludeAggregateIDsClauses, excludeAggregateIDsArgs := prepareQuery(criteria, useV1, excludeAggregateIDs...)
if excludeAggregateIDsClauses != "" {
if clauses != "" { if clauses != "" {
clauses += " AND " clauses += " AND "
} }
if useV1 { clauses += excludeAggregateIDsClause
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...) args = append(args, excludeAggregateIDsArgs...)
} }
@ -312,6 +312,9 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo
instanceIDs[i] = "zitadel_es_pusher_" + instanceIDs[i] instanceIDs[i] = "zitadel_es_pusher_" + instanceIDs[i]
} }
if clauses != "" {
clauses += " AND "
}
clauses += awaitOpenTransactions(useV1) clauses += awaitOpenTransactions(useV1)
args = append(args, instanceIDs) args = append(args, instanceIDs)
} }
@ -323,6 +326,23 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo
return " WHERE " + clauses, args 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) { func prepareQuery(criteria querier, useV1 bool, filters ...*repository.Filter) (_ string, args []any) {
clauses := make([]string, 0, len(filters)) clauses := make([]string, 0, len(filters))
args = make([]any, 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{ fields: fields{
mock: newMockClient(t).expectQuery(t, mock: newMockClient(t).expectQuery(t,
regexp.QuoteMeta( 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 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`, `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`,
), ),
[]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)}, []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)},
), ),
}, },
res: res{ res: res{
@ -1051,9 +1051,9 @@ func Test_query_events_mocked(t *testing.T) {
fields: fields{ fields: fields{
mock: newMockClient(t).expectQuery(t, mock: newMockClient(t).expectQuery(t,
regexp.QuoteMeta( 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 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`, `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`,
), ),
[]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)}, []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)},
), ),
}, },
res: res{ res: res{