perf(db): keyset pagination for revision prefixes
Keyset по id/ord вместо OFFSET; anti-join для RevisionDiff. Co-authored-by: Cursor <[email protected]>
This commit is contained in:
@@ -789,12 +789,7 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
|
|||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 50
|
limit = 50
|
||||||
}
|
}
|
||||||
off := 0
|
afterID, off, useOffset := store.ParsePrefixPageCursor(cursor)
|
||||||
if cursor != "" {
|
|
||||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
|
||||||
off = n
|
|
||||||
}
|
|
||||||
}
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
var one int
|
var one int
|
||||||
if err := p.pool.QueryRow(ctx, `
|
if err := p.pool.QueryRow(ctx, `
|
||||||
@@ -805,34 +800,52 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
|
|||||||
}
|
}
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
rows, err := p.pool.Query(ctx, `
|
var rows pgx.Rows
|
||||||
SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix
|
var err error
|
||||||
WHERE revision_id=$1 ORDER BY id
|
if useOffset {
|
||||||
LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
|
rows, err = p.pool.Query(ctx, `
|
||||||
|
SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix
|
||||||
|
WHERE revision_id=$1::uuid ORDER BY id
|
||||||
|
LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
|
||||||
|
} else {
|
||||||
|
var afterArg any
|
||||||
|
if afterID != nil {
|
||||||
|
afterArg = *afterID
|
||||||
|
}
|
||||||
|
rows, err = p.pool.Query(ctx, `
|
||||||
|
SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix
|
||||||
|
WHERE revision_id=$1::uuid AND ($2::bigint IS NULL OR id > $2::bigint)
|
||||||
|
ORDER BY id
|
||||||
|
LIMIT $3`, revisionID, afterArg, limit+1)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
var all []store.PrefixRow
|
var all []store.PrefixRow
|
||||||
|
var ids []int64
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
|
var rowID int64
|
||||||
var pr store.PrefixRow
|
var pr store.PrefixRow
|
||||||
var comm *string
|
var comm *string
|
||||||
if err := rows.Scan(&pr.Prefix, &comm, &pr.Source); err != nil {
|
if err := rows.Scan(&rowID, &pr.Prefix, &comm, &pr.Source); err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
pr.CommunityID = comm
|
pr.CommunityID = comm
|
||||||
|
ids = append(ids, rowID)
|
||||||
all = append(all, pr)
|
all = append(all, pr)
|
||||||
}
|
}
|
||||||
agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{
|
agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{
|
||||||
"rows": len(all), "limit": limit, "offset": off,
|
"rows": len(all), "limit": limit, "keyset": !useOffset,
|
||||||
})
|
})
|
||||||
more := len(all) > limit
|
more := len(all) > limit
|
||||||
if more {
|
if more {
|
||||||
all = all[:limit]
|
all = all[:limit]
|
||||||
|
ids = ids[:limit]
|
||||||
}
|
}
|
||||||
next := ""
|
next := ""
|
||||||
if more {
|
if more && len(ids) > 0 {
|
||||||
next = fmt.Sprintf("%d", off+limit)
|
next = store.FormatPrefixPageCursor(ids[len(ids)-1])
|
||||||
}
|
}
|
||||||
if len(all) == 0 {
|
if len(all) == 0 {
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
@@ -896,21 +909,21 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
|||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
var unchanged int
|
var unchanged int
|
||||||
err := p.pool.QueryRow(ctx, `
|
err := p.pool.QueryRow(ctx, `
|
||||||
SELECT COUNT(*)::int FROM (
|
SELECT COUNT(*)::int FROM revision_materialized_prefix b
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
INNER JOIN revision_materialized_prefix a
|
||||||
INTERSECT
|
ON a.revision_id = $1::uuid AND a.prefix = b.prefix
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
WHERE b.revision_id = $2::uuid`, aID, bID).Scan(&unchanged)
|
||||||
) t`, aID, bID).Scan(&unchanged)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
// added: в B, нет в A; removed: в A, нет в B — без загрузки полных снапшотов в память.
|
|
||||||
rowsAdded, err := p.pool.Query(ctx, `
|
rowsAdded, err := p.pool.Query(ctx, `
|
||||||
SELECT prefix::text FROM (
|
SELECT b.prefix::text
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
FROM revision_materialized_prefix b
|
||||||
EXCEPT
|
LEFT JOIN revision_materialized_prefix a
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
ON a.revision_id = $1::uuid AND a.prefix = b.prefix
|
||||||
) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1)
|
WHERE b.revision_id = $2::uuid AND a.prefix IS NULL
|
||||||
|
ORDER BY b.prefix
|
||||||
|
LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -929,11 +942,13 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
|||||||
}
|
}
|
||||||
addedTruncated := len(added) >= maxRevisionDiffRows
|
addedTruncated := len(added) >= maxRevisionDiffRows
|
||||||
rowsRem, err := p.pool.Query(ctx, `
|
rowsRem, err := p.pool.Query(ctx, `
|
||||||
SELECT prefix::text FROM (
|
SELECT a.prefix::text
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
FROM revision_materialized_prefix a
|
||||||
EXCEPT
|
LEFT JOIN revision_materialized_prefix b
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
ON b.revision_id = $2::uuid AND b.prefix = a.prefix
|
||||||
) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
WHERE a.revision_id = $1::uuid AND b.prefix IS NULL
|
||||||
|
ORDER BY a.prefix
|
||||||
|
LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,8 +1,6 @@
|
|||||||
package store
|
package store
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
|
||||||
"strconv"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -795,31 +793,37 @@ func (m *Memory) ListRevisionPrefixes(tenantID, revisionID string, cursor string
|
|||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 50
|
limit = 50
|
||||||
}
|
}
|
||||||
|
afterID, off, useOffset := ParsePrefixPageCursor(cursor)
|
||||||
m.mu.RLock()
|
m.mu.RLock()
|
||||||
defer m.mu.RUnlock()
|
defer m.mu.RUnlock()
|
||||||
if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil {
|
if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil {
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
all := m.revPrefixes[revisionID]
|
allRows := m.revPrefixes[revisionID]
|
||||||
off := 0
|
start := 0
|
||||||
if cursor != "" {
|
if useOffset {
|
||||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
start = off
|
||||||
off = n
|
} else if afterID != nil {
|
||||||
}
|
start = int(*afterID) + 1
|
||||||
}
|
}
|
||||||
end := off + limit
|
if start > len(allRows) {
|
||||||
next := ""
|
|
||||||
more := false
|
|
||||||
if end > len(all) {
|
|
||||||
end = len(all)
|
|
||||||
} else {
|
|
||||||
more = true
|
|
||||||
next = fmt.Sprintf("%d", end)
|
|
||||||
}
|
|
||||||
if off >= len(all) {
|
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
return all[off:end], next, more
|
end := start + limit
|
||||||
|
next := ""
|
||||||
|
more := false
|
||||||
|
if end > len(allRows) {
|
||||||
|
end = len(allRows)
|
||||||
|
} else {
|
||||||
|
more = true
|
||||||
|
next = FormatPrefixPageCursor(int64(end - 1))
|
||||||
|
}
|
||||||
|
if start >= end {
|
||||||
|
return nil, "", false
|
||||||
|
}
|
||||||
|
out := make([]PrefixRow, end-start)
|
||||||
|
copy(out, allRows[start:end])
|
||||||
|
return out, next, more
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *Memory) ListGlobalSettings(tenantID string) (map[string]any, error) {
|
func (m *Memory) ListGlobalSettings(tenantID string) (map[string]any, error) {
|
||||||
|
|||||||
@@ -0,0 +1,30 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ParsePrefixPageCursor decodes opaque cursors for revision prefix pagination.
|
||||||
|
func ParsePrefixPageCursor(cursor string) (afterID *int64, offset int, useOffset bool) {
|
||||||
|
cursor = strings.TrimSpace(cursor)
|
||||||
|
if cursor == "" {
|
||||||
|
return nil, 0, false
|
||||||
|
}
|
||||||
|
if strings.HasPrefix(cursor, "o:") {
|
||||||
|
n, err := strconv.Atoi(strings.TrimPrefix(cursor, "o:"))
|
||||||
|
if err != nil || n < 0 {
|
||||||
|
return nil, 0, false
|
||||||
|
}
|
||||||
|
return nil, n, true
|
||||||
|
}
|
||||||
|
if n, err := strconv.ParseInt(cursor, 10, 64); err == nil && n >= 0 {
|
||||||
|
return &n, 0, false
|
||||||
|
}
|
||||||
|
return nil, 0, false
|
||||||
|
}
|
||||||
|
|
||||||
|
// FormatPrefixPageCursor encodes the keyset cursor (last row id or slice index).
|
||||||
|
func FormatPrefixPageCursor(lastID int64) string {
|
||||||
|
return strconv.FormatInt(lastID, 10)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user