mirror of
https://github.com/zitadel/zitadel.git
synced 2026-11-03 12:47:09 -06:00
perf(eventstore): order by sort key as column list instead of row constructor (#12792)
# Which Problems Are Solved - #12703 changed the descending default order of events to a row constructor `ORDER BY ("position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence") DESC`. Postgres treats a row constructor as one composite expression, so no btree index can presort it and no incremental sort is possible. Every descending query through the sort key branch reads all matching events and sorts them before applying the `LIMIT`, even where an index matches the leading sort columns. - The default `ORDER BY`, the creation date `ORDER BY` from #12789 and the tuple of the resume cursor were three separate string literals that had to be kept in sync by hand. # How the Problems Are Solved - The `ORDER BY` is written as a column list with the direction on every column. The result order is identical to the row constructor, but the planner can presort from an index again and stop after `LIMIT` rows. - The sort key columns are defined once and drive the default `ORDER BY`, the creation date `ORDER BY` and the cursor tuple, so they can no longer drift apart. - The row-wise `>` comparison of the resume cursor is unchanged. That is the one place where the tuple form is needed. # Additional Changes - Table test for all order-by variants and a test pinning the cursor tuple. - Updated the two SQL fixtures that expected the row constructor. # Additional Context Same query, only the `ORDER BY` form differs (instance filter, `aggregate_type = 'user'`, `LIMIT 1000`, ordered by `created_at` plus the sort key): | ORDER BY form | Plan | Estimated cost | | --- | --- | --- | | row constructor | parallel index scan + full sort of 3.5 M rows | 11.2 M | | column list | incremental sort on index, stops at limit | 20 k, 15 ms | Projection catch-up queries filter several aggregate and event types with `= ANY`, so on Postgres before 17 they still need a sort. The gain there is a top-N sort instead of a full sort where an index prefix matches, and it grows on Postgres 17 which can return ordered output across array conditions. - Follow-up for PR #12703 - Stacked on #12790 (which is stacked on #12789) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Livio SpringandClaude Fable 5.1 authored and 
GitHub committed
1 parent
92dcb84743
commit
1874e58503
4 files changed
+111
-32
No files matched your search
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/shopspring/decimal"
|
||||
@@ -40,19 +41,37 @@ 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"`
|
||||
eventColumnsSQL = `created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order`
|
||||
)
|
||||
const eventColumnsSQL = `created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order`
|
||||
|
||||
// creationDateOrderSQL orders by created_at first and uses the event sort key as tie breaker.
|
||||
// It is deliberately written as a column list and not as a row constructor like the
|
||||
// default order: only a column list allows postgres to presort from an index on created_at
|
||||
// (incremental sort) and stop after LIMIT rows instead of sorting all matching events.
|
||||
var (
|
||||
creationDateOrderAscSQL = ` ORDER BY created_at, ` + eventSortKeySQL
|
||||
creationDateOrderDescSQL = ` ORDER BY created_at DESC, "position" DESC, in_tx_order DESC, instance_id DESC, aggregate_type DESC, aggregate_id DESC, "sequence" DESC`
|
||||
)
|
||||
// eventSortKeyColumns is the lexicographic sort key of events.
|
||||
// It defines the default ORDER BY and the tuple of the resume cursor ([eventstore.EventSortKey]),
|
||||
// both must always use the same columns in the same order.
|
||||
var eventSortKeyColumns = []string{`"position"`, "in_tx_order", "instance_id", "aggregate_type", "aggregate_id", `"sequence"`}
|
||||
|
||||
// eventSortKeySQL is the comma separated [eventSortKeyColumns], used as tuple in the resume cursor
|
||||
// and as ORDER BY of the per event type scans.
|
||||
var eventSortKeySQL = strings.Join(eventSortKeyColumns, ", ")
|
||||
|
||||
// orderBy builds the ORDER BY clause of the given columns, all in the same direction.
|
||||
// It is deliberately written as a column list and not as a row constructor (ORDER BY (a, b) DESC):
|
||||
// only a column list allows postgres to presort from an index (incremental sort) and stop after
|
||||
// LIMIT rows instead of sorting all matching events.
|
||||
func orderBy(desc bool, columns ...string) string {
|
||||
direction := ""
|
||||
if desc {
|
||||
direction = " DESC"
|
||||
}
|
||||
var b strings.Builder
|
||||
b.WriteString(" ORDER BY ")
|
||||
for i, column := range columns {
|
||||
if i > 0 {
|
||||
b.WriteString(", ")
|
||||
}
|
||||
b.WriteString(column)
|
||||
b.WriteString(direction)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// 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) {
|
||||
@@ -94,28 +113,16 @@ func (db *Postgres) Client() *database.DB {
|
||||
|
||||
func (db *Postgres) orderByEventSequence(desc, shouldOrderBySequence, orderByCreationDate, useV1 bool) string {
|
||||
if useV1 {
|
||||
if desc {
|
||||
return ` ORDER BY event_sequence DESC`
|
||||
}
|
||||
return ` ORDER BY event_sequence`
|
||||
return orderBy(desc, "event_sequence")
|
||||
}
|
||||
if shouldOrderBySequence {
|
||||
if desc {
|
||||
return ` ORDER BY "sequence" DESC`
|
||||
}
|
||||
return ` ORDER BY "sequence"`
|
||||
return orderBy(desc, `"sequence"`)
|
||||
}
|
||||
if orderByCreationDate {
|
||||
if desc {
|
||||
return creationDateOrderDescSQL
|
||||
}
|
||||
return creationDateOrderAscSQL
|
||||
// created_at first and the sort key as tie breaker for events created in the same microsecond
|
||||
return orderBy(desc, append([]string{"created_at"}, eventSortKeyColumns...)...)
|
||||
}
|
||||
|
||||
if desc {
|
||||
return ` ORDER BY (` + eventSortKeySQL + `) DESC`
|
||||
}
|
||||
return ` ORDER BY ` + eventSortKeySQL
|
||||
return orderBy(desc, eventSortKeyColumns...)
|
||||
}
|
||||
|
||||
func (db *Postgres) eventQuery(useV1 bool) string {
|
||||
|
||||
@@ -10,6 +10,78 @@ import (
|
||||
"github.com/zitadel/zitadel/internal/eventstore/repository"
|
||||
)
|
||||
|
||||
func TestPostgres_orderByEventSequence(t *testing.T) {
|
||||
type args struct {
|
||||
desc bool
|
||||
shouldOrderBySequence bool
|
||||
orderByCreationDate bool
|
||||
useV1 bool
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "v1 asc",
|
||||
args: args{useV1: true},
|
||||
want: ` ORDER BY event_sequence`,
|
||||
},
|
||||
{
|
||||
name: "v1 desc",
|
||||
args: args{useV1: true, desc: true},
|
||||
want: ` ORDER BY event_sequence DESC`,
|
||||
},
|
||||
{
|
||||
name: "single aggregate asc",
|
||||
args: args{shouldOrderBySequence: true},
|
||||
want: ` ORDER BY "sequence"`,
|
||||
},
|
||||
{
|
||||
name: "single aggregate desc ignores creation date",
|
||||
args: args{shouldOrderBySequence: true, orderByCreationDate: true, desc: true},
|
||||
want: ` ORDER BY "sequence" DESC`,
|
||||
},
|
||||
{
|
||||
name: "sort key asc",
|
||||
args: args{},
|
||||
want: ` ORDER BY "position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence"`,
|
||||
},
|
||||
{
|
||||
name: "sort key desc is a column list, not a row constructor",
|
||||
args: args{desc: true},
|
||||
want: ` ORDER BY "position" DESC, in_tx_order DESC, instance_id DESC, aggregate_type DESC, aggregate_id DESC, "sequence" DESC`,
|
||||
},
|
||||
{
|
||||
name: "creation date asc",
|
||||
args: args{orderByCreationDate: true},
|
||||
want: ` ORDER BY created_at, "position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence"`,
|
||||
},
|
||||
{
|
||||
name: "creation date desc",
|
||||
args: args{orderByCreationDate: true, desc: true},
|
||||
want: ` ORDER BY created_at DESC, "position" DESC, in_tx_order DESC, instance_id DESC, aggregate_type DESC, aggregate_id DESC, "sequence" DESC`,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
db := &Postgres{}
|
||||
got := db.orderByEventSequence(tt.args.desc, tt.args.shouldOrderBySequence, tt.args.orderByCreationDate, tt.args.useV1)
|
||||
if got != tt.want {
|
||||
t.Errorf("Postgres.orderByEventSequence() = %q, want %q", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_eventSortKeySQL_matches_columns(t *testing.T) {
|
||||
// the resume cursor tuple and the ORDER BY must use the same columns in the same order
|
||||
want := `"position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence"`
|
||||
if eventSortKeySQL != want {
|
||||
t.Errorf("eventSortKeySQL = %q, want %q", eventSortKeySQL, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgres_placeholder(t *testing.T) {
|
||||
type args struct {
|
||||
query string
|
||||
|
||||
@@ -87,7 +87,7 @@ func Test_query_event_type_scans(t *testing.T) {
|
||||
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`,
|
||||
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 "position" DESC, in_tx_order DESC, instance_id DESC, aggregate_type DESC, aggregate_id DESC, "sequence" DESC LIMIT $4`,
|
||||
args: []driver.Value{
|
||||
"instanceID", eventstore.AggregateType("org"), eventstore.EventType("org.removed"),
|
||||
uint64(200),
|
||||
|
||||
@@ -904,7 +904,7 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(
|
||||
regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order 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 (`+eventSortKeySQL+`) DESC LIMIT $9`),
|
||||
regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order 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, instance_id DESC, aggregate_type DESC, aggregate_id DESC, "sequence" DESC LIMIT $9`),
|
||||
[]driver.Value{"instanceID", eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), decimal.NewFromFloat(123.456), eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", decimal.NewFromFloat(123.456), uint64(5)},
|
||||
),
|
||||
},
|
||||
@@ -1031,7 +1031,7 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(
|
||||
regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND created_at > $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 created_at > $8) ORDER BY (`+eventSortKeySQL+`) DESC LIMIT $9`),
|
||||
regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND created_at > $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 created_at > $8) ORDER BY "position" DESC, in_tx_order DESC, instance_id DESC, aggregate_type DESC, aggregate_id DESC, "sequence" DESC LIMIT $9`),
|
||||
[]driver.Value{"instanceID", eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), time.Unix(123, 456), eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", time.Unix(123, 456), uint64(5)},
|
||||
),
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user