Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions input/postgres/backends.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,14 @@ END
const activitySQL string = `
SELECT (extract(epoch from COALESCE(backend_start, pg_catalog.pg_postmaster_start_time()))::int::text || pg_catalog.to_char(pid, 'FM0000000'))::bigint,
datid, datname, usesysid, usename, pid, application_name, client_addr::text, client_port,
backend_start, xact_start, query_start, state_change, COALESCE(wait_event_type, '') = 'Lock' as waiting, backend_xid, backend_xmin, wait_event_type, wait_event, backend_type, %s, state, query
backend_start, xact_start, query_start, state_change, COALESCE(wait_event_type, '') = 'Lock' as waiting,
backend_xid, backend_xmin, wait_event_type, wait_event, backend_type, %s, state, query, %s
FROM %s
WHERE pid IS NOT NULL`

func GetBackends(ctx context.Context, c *Collection, db *sql.DB) ([]state.PostgresBackend, error) {
var blockingPidsField string
var queryIdField string
var sourceTable string

if c.GlobalOpts.CollectPostgresLocks {
Expand All @@ -33,13 +35,19 @@ func GetBackends(ctx context.Context, c *Collection, db *sql.DB) ([]state.Postgr
blockingPidsField = "NULL"
}

if c.PostgresVersion.Numeric >= state.PostgresVersion14 {
queryIdField = "coalesce(query_id, 0)"
} else {
queryIdField = "0"
}

if c.HelperExists("get_stat_activity", nil) {
sourceTable = "pganalyze.get_stat_activity()"
} else {
sourceTable = "pg_catalog.pg_stat_activity"
}

stmt, err := db.PrepareContext(ctx, QueryMarkerSQL+fmt.Sprintf(activitySQL, blockingPidsField, sourceTable))
stmt, err := db.PrepareContext(ctx, QueryMarkerSQL+fmt.Sprintf(activitySQL, blockingPidsField, queryIdField, sourceTable))
if err != nil {
return nil, err
}
Expand All @@ -63,7 +71,7 @@ func GetBackends(ctx context.Context, c *Collection, db *sql.DB) ([]state.Postgr
&row.ClientPort, &row.BackendStart, &row.XactStart, &row.QueryStart,
&row.StateChange, &row.Waiting, &row.BackendXid, &row.BackendXmin,
&row.WaitEventType, &row.WaitEvent, &row.BackendType, pq.Array(&row.BlockedByPids),
&row.State, &row.Query)
&row.State, &row.Query, &row.QueryId)
if err != nil {
return nil, err
}
Expand Down
4 changes: 4 additions & 0 deletions input/postgres/collection.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@ type Collection struct {

// Information that is specific to the current database we're connected to
HelperFunctions map[string][]state.PostgresFunction

Fingerprints *state.Fingerprints
}

func helpersFromFunctions(functions []state.PostgresFunction) map[string][]state.PostgresFunction {
Expand Down Expand Up @@ -81,6 +83,7 @@ func NewCollection(ctx context.Context, logger *util.Logger, server *state.Serve
ConnectedAsSuperUser: connectedAsSuperUser,
ConnectedAsMonitoringRole: connectedAsMonitoringRole,
HelperFunctions: helpersFromFunctions(helperFunctions),
Fingerprints: server.Fingerprints,
}, nil
}

Expand All @@ -95,6 +98,7 @@ func (c *Collection) ForCurrentDatabase(functions []state.PostgresFunction) *Col
ConnectedAsSuperUser: c.ConnectedAsSuperUser,
ConnectedAsMonitoringRole: c.ConnectedAsMonitoringRole,
HelperFunctions: helpersFromFunctions(functions),
Fingerprints: c.Fingerprints,
}
}

Expand Down
18 changes: 13 additions & 5 deletions input/postgres/statements.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"strings"

"github.com/guregu/null"
"github.com/lib/pq"
"github.com/pganalyze/collector/selftest"
"github.com/pganalyze/collector/state"
"github.com/pganalyze/collector/util"
Expand Down Expand Up @@ -41,7 +42,8 @@ SELECT dbid, userid, queryid, %s, calls, %s, rows, shared_blks_hit, shared_blks_

const statementTextSQL string = `
SELECT dbid, userid, queryid, %s, query
FROM %s`
FROM %s
WHERE queryid = ANY($1)`

@lfittl lfittl Feb 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As discussed, doing this filtering has three main benefits that I can see:

  1. We avoid sending query text over the network (from the database to the collector) that we've already seen
  2. We avoid running normalize on query texts we've already seen
  3. We avoid sending query text to pganalyze that pganalyze already got

Just for clarity, because of how pg_stat_statements works today, this unfortunately won't reduce the effort done to read the query text file and put the query text and statistics into the tuplestore, since the WHERE condition won't be pushed down into pg_stat_statements_internal.


const statementExtensionVersionSQL string = `
SELECT nspname,
Expand Down Expand Up @@ -135,6 +137,7 @@ func GetStatementStats(ctx context.Context, c *Collection, db *sql.DB) (state.Po

if queryID.Valid {
key.QueryID = queryID.Int64
c.Fingerprints.Add(queryID.Int64, "", "", -1)
} else {
// We can't process this entry, most likely a permission problem with reading the query ID
continue
Expand All @@ -153,6 +156,11 @@ func GetStatementStats(ctx context.Context, c *Collection, db *sql.DB) (state.Po
}

func GetStatementTexts(ctx context.Context, c *Collection, db *sql.DB) (state.PostgresStatementMap, state.PostgresStatementTextMap, error) {
queryIDs := c.Fingerprints.TakeNewQueryIDs()
if len(queryIDs) == 0 {
return nil, nil, nil
}

sourceTable, foundExtMinorVersion, err := getStatementSource(ctx, c, db, true)
if err != nil {
return nil, nil, err
Expand All @@ -170,7 +178,7 @@ func GetStatementTexts(ctx context.Context, c *Collection, db *sql.DB) (state.Po
}
defer stmt.Close()

rows, err := stmt.QueryContext(ctx)
rows, err := stmt.QueryContext(ctx, pq.Array(queryIDs))
if err != nil {
return nil, nil, err
}
Expand Down Expand Up @@ -232,7 +240,7 @@ func GetStatementTexts(ctx context.Context, c *Collection, db *sql.DB) (state.Po
case <-ctx.Done():
return nil, nil, ctx.Err()
default:
fingerprintAndNormalize(c, key, query, statements, statementTextsByFp, ignoreIoTiming)
fingerprintAndNormalize(c, key, key.QueryID, query, statements, statementTextsByFp, ignoreIoTiming)
}
}

Expand Down Expand Up @@ -351,7 +359,7 @@ func ignoreIOTiming(postgresVersion state.PostgresVersion, receivedQuery string)
var collectorQueryFingerprint = util.FingerprintText(util.QueryTextCollector)
var insufficientPrivsQueryFingerprint = util.FingerprintText(util.QueryTextInsufficientPrivs)

func fingerprintAndNormalize(c *Collection, key state.PostgresStatementKey, text string, statements state.PostgresStatementMap, statementTextsByFp state.PostgresStatementTextMap, ignoreIoTiming bool) {
func fingerprintAndNormalize(c *Collection, key state.PostgresStatementKey, queryID int64, text string, statements state.PostgresStatementMap, statementTextsByFp state.PostgresStatementTextMap, ignoreIoTiming bool) {
if insufficientPrivilege(text) {
statements[key] = state.PostgresStatement{
InsufficientPrivilege: true,
Expand All @@ -365,7 +373,7 @@ func fingerprintAndNormalize(c *Collection, key state.PostgresStatementKey, text
IgnoreIoTiming: ignoreIoTiming,
}
} else {
fp := util.FingerprintQuery(text, c.Config.FilterQueryText, -1)
fp := c.Fingerprints.Add(queryID, text, c.Config.FilterQueryText, -1)
statements[key] = state.PostgresStatement{Fingerprint: fp, IgnoreIoTiming: ignoreIoTiming}
_, ok := statementTextsByFp[fp]
if !ok {
Expand Down
2 changes: 1 addition & 1 deletion output/full.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ import (
)

func SendFull(ctx context.Context, server *state.Server, collectionOpts state.CollectionOpts, logger *util.Logger, newState state.PersistedState, diffState state.DiffState, transientState state.TransientState, collectedIntervalSecs uint32) error {
s := transform.StateToSnapshot(newState, diffState, transientState)
s := transform.StateToSnapshot(server, newState, diffState, transientState)
s.CollectedIntervalSecs = collectedIntervalSecs
err := verifyIntegrity(&s)
if err != nil {
Expand Down
1 change: 1 addition & 0 deletions output/transform/activity.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ func ActivityStateToCompactActivitySnapshot(server *state.Server, activityState
b.RoleIdx,
b.DatabaseIdx,
backend.Query.String,
backend.QueryId,
activityState.TrackActivityQuerySize,
)
b.HasQueryIdx = true
Expand Down
2 changes: 2 additions & 0 deletions output/transform/logs.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ func transformPostgresQuerySamples(server *state.Server, s snapshot.CompactLogSn
roleIdx,
databaseIdx,
sampleIn.Query,
0,
-1,
)

Expand Down Expand Up @@ -184,6 +185,7 @@ func transformSystemLogLine(server *state.Server, r *snapshot.CompactSnapshot_Ba
logLine.RoleIdx,
logLine.DatabaseIdx,
logLineIn.Query,
0,
-1,
)
logLine.HasQueryIdx = true
Expand Down
4 changes: 2 additions & 2 deletions output/transform/postgres.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ import (

type OidToIdx map[state.Oid]int32

func transformPostgres(s snapshot.FullSnapshot, newState state.PersistedState, diffState state.DiffState, transientState state.TransientState) snapshot.FullSnapshot {
func transformPostgres(server *state.Server, s snapshot.FullSnapshot, newState state.PersistedState, diffState state.DiffState, transientState state.TransientState) snapshot.FullSnapshot {
s, roleOidToIdx := transformPostgresRoles(s, transientState)
s, databaseOidToIdx := transformPostgresDatabases(s, diffState, transientState, roleOidToIdx)
s, typeOidToIdx := transformPostgresTypes(s, transientState, databaseOidToIdx)
Expand All @@ -19,7 +19,7 @@ func transformPostgres(s snapshot.FullSnapshot, newState state.PersistedState, d
s = transformPostgresConfig(s, transientState)
s = transformPostgresServerStats(s, newState, diffState, transientState)
s = transformPostgresReplication(s, transientState, roleOidToIdx)
s, queryIDKeyToIdx := transformPostgresStatements(s, newState, diffState, transientState, roleOidToIdx, databaseOidToIdx)
s, queryIDKeyToIdx := transformPostgresStatements(server, s, newState, diffState, transientState, roleOidToIdx, databaseOidToIdx)
s = transformPostgresPlans(s, newState, diffState, transientState, queryIDKeyToIdx)
s = transformPostgresRelations(s, newState, diffState, databaseOidToIdx, typeOidToIdx, s.ServerStatistic.CurrentXactId)
s = transformPostgresFunctions(s, newState, diffState, roleOidToIdx, databaseOidToIdx)
Expand Down
11 changes: 7 additions & 4 deletions output/transform/postgres_statements.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,14 @@ import (
"google.golang.org/protobuf/types/known/timestamppb"
)

func groupStatements(statements state.PostgresStatementMap, statsMap state.DiffedPostgresStatementStatsMap) map[statementKey]statementValue {
func groupStatements(server *state.Server, statsMap state.DiffedPostgresStatementStatsMap) map[statementKey]statementValue {
groupedStatements := make(map[statementKey]statementValue)

for sKey, stats := range statsMap {
statement, exist := statements[sKey]
// TODO: what do do about the extra logic in fingerprintAndNormalize? track IO timing, detection of collector queries, etc
// Maybe those should all be implemented on the server side?
fingerprint, exist := server.Fingerprints.Get(sKey.QueryID)
statement := state.PostgresStatement{Fingerprint: fingerprint}
if !exist {
statement = state.PostgresStatement{QueryTextUnavailable: true, Fingerprint: util.FingerprintText(util.QueryTextUnavailable)}
}
Expand Down Expand Up @@ -76,7 +79,7 @@ type queryIDKey struct {
}
type QueryIDKeyToIdx map[queryIDKey]int32

func transformPostgresStatements(s snapshot.FullSnapshot, newState state.PersistedState, diffState state.DiffState, transientState state.TransientState, roleOidToIdx OidToIdx, databaseOidToIdx OidToIdx) (snapshot.FullSnapshot, QueryIDKeyToIdx) {
func transformPostgresStatements(server *state.Server, s snapshot.FullSnapshot, newState state.PersistedState, diffState state.DiffState, transientState state.TransientState, roleOidToIdx OidToIdx, databaseOidToIdx OidToIdx) (snapshot.FullSnapshot, QueryIDKeyToIdx) {
var queryStats []*snapshot.HistoricQueryStatistics
queryIDKeyToIDx := make(QueryIDKeyToIdx)

Expand All @@ -91,7 +94,7 @@ func transformPostgresStatements(s snapshot.FullSnapshot, newState state.Persist
h.CollectedAt = timestamppb.New(timeKey.CollectedAt)
h.CollectedIntervalSecs = timeKey.CollectedIntervalSecs

groupedStatements := groupStatements(transientState.Statements, diffedStats)
groupedStatements := groupStatements(server, diffedStats)
for key, value := range groupedStatements {
idx := upsertQueryReferenceAndInformation(&s, transientState.StatementTexts, roleOidToIdx, databaseOidToIdx, key, value)
// Store the map of QueryIdx (idx here) and databaseOid, userOid, queryID combinations
Expand Down
4 changes: 2 additions & 2 deletions output/transform/transform.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,10 @@ import (
"github.com/pganalyze/collector/state"
)

func StateToSnapshot(newState state.PersistedState, diffState state.DiffState, transientState state.TransientState) snapshot.FullSnapshot {
func StateToSnapshot(server *state.Server, newState state.PersistedState, diffState state.DiffState, transientState state.TransientState) snapshot.FullSnapshot {
var s snapshot.FullSnapshot

s = transformPostgres(s, newState, diffState, transientState)
s = transformPostgres(server, s, newState, diffState, transientState)
s = systemStateToFullSnapshot(s, newState, diffState)
s = transformCollectorStats(s, newState, diffState)
s = transformCollectorPlatform(s, transientState)
Expand Down
28 changes: 16 additions & 12 deletions output/transform/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,12 @@ type statementValue struct {
}

func upsertQueryReferenceAndInformation(s *snapshot.FullSnapshot, statementTexts state.PostgresStatementTextMap, roleOidToIdx OidToIdx, databaseOidToIdx OidToIdx, key statementKey, value statementValue) int32 {
normalizedQuery, exists := statementTexts[key.fingerprint]
if !exists {
return -1 // A query that already exists in the fingerprint cache, so won't be submitted
// TODO: this doesn't work, since the protobuf QueryStatistic uses query index instead of fingerprint
}

fpBuf := make([]byte, 8)
binary.BigEndian.PutUint64(fpBuf, key.fingerprint)
newRef := snapshot.QueryReference{
Expand All @@ -41,16 +47,14 @@ func upsertQueryReferenceAndInformation(s *snapshot.FullSnapshot, statementTexts
s.QueryReferences = append(s.QueryReferences, &newRef)

// Information
normalizedQuery := ""
if value.statement.QueryTextUnavailable {
normalizedQuery = "<query text unavailable>"
} else if value.statement.InsufficientPrivilege {
normalizedQuery = "<insufficient privilege>"
} else if value.statement.Collector {
normalizedQuery = "<pganalyze-collector>"
} else {
normalizedQuery = statementTexts[key.fingerprint]
}
// TODO: does this work now?
// if value.statement.QueryTextUnavailable {
// normalizedQuery = "<query text unavailable>"
// } else if value.statement.InsufficientPrivilege {
// normalizedQuery = "<insufficient privilege>"
// } else if value.statement.Collector {
// normalizedQuery = "<pganalyze-collector>"
// }
queryInformation := snapshot.QueryInformation{
QueryIdx: idx,
NormalizedQuery: normalizedQuery,
Expand All @@ -61,8 +65,8 @@ func upsertQueryReferenceAndInformation(s *snapshot.FullSnapshot, statementTexts
return idx
}

func upsertQueryReferenceAndInformationSimple(server *state.Server, refs []*snapshot.QueryReference, infos []*snapshot.QueryInformation, roleIdx int32, databaseIdx int32, originalQuery string, trackActivityQuerySize int) (int32, []*snapshot.QueryReference, []*snapshot.QueryInformation) {
fingerprint := util.FingerprintQuery(originalQuery, server.Config.FilterQueryText, trackActivityQuerySize)
func upsertQueryReferenceAndInformationSimple(server *state.Server, refs []*snapshot.QueryReference, infos []*snapshot.QueryInformation, roleIdx int32, databaseIdx int32, originalQuery string, queryID int64, trackActivityQuerySize int) (int32, []*snapshot.QueryReference, []*snapshot.QueryInformation) {
fingerprint := server.Fingerprints.Add(queryID, originalQuery, server.Config.FilterQueryText, trackActivityQuerySize)

fpBuf := make([]byte, 8)
binary.BigEndian.PutUint64(fpBuf, fingerprint)
Expand Down
84 changes: 84 additions & 0 deletions state/fingerprints.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
package state

import (
"github.com/pganalyze/collector/util"
"sync"
)

// 500,000 entries use around 12 MB
const MAX_SIZE = 500_000

type Fingerprints struct {
lock sync.RWMutex
cache map[int64]uint64
newQueryIDs []int64
}

func NewFingerprints() *Fingerprints {
return &Fingerprints{
lock: sync.RWMutex{},
cache: make(map[int64]uint64, MAX_SIZE),
}
}

func (c *Fingerprints) Get(queryID int64) (uint64, bool) {
c.lock.RLock()
defer c.lock.RUnlock()
fingerprint, exists := c.cache[queryID]
return fingerprint, exists
}

func (c *Fingerprints) Add(queryID int64, text string, filterQueryText string, trackActivityQuerySize int) uint64 {
if queryID == 0 {
return util.FingerprintQuery(text, filterQueryText, trackActivityQuerySize)
}
fingerprint, exists := c.Get(queryID)
if exists {
return fingerprint
}
fingerprint, virtual := util.TryFingerprintQuery(text, filterQueryText, trackActivityQuerySize)
if virtual {
c.lock.Lock()
c.newQueryIDs = append(c.newQueryIDs, queryID)
c.lock.Unlock()
return fingerprint
}
c.cleanup()
c.lock.Lock()
c.cache[queryID] = fingerprint
c.lock.Unlock()
return fingerprint
}

// Called by GetStatementTexts to only look up query texts for new, unknown query IDs
func (c *Fingerprints) TakeNewQueryIDs() []int64 {
c.lock.Lock()
defer c.lock.Unlock()
newQueryIDs := c.newQueryIDs
c.newQueryIDs = nil
return newQueryIDs
}

func (c *Fingerprints) size() int {
c.lock.RLock()
defer c.lock.RUnlock()
return len(c.cache)
}

// Retains a random 50% sample of entries if the cache grows too large
func (c *Fingerprints) cleanup() {
if c.size() < MAX_SIZE {
return
}
c.lock.Lock()
cache := make(map[int64]uint64, MAX_SIZE)
index := 0
for key, value := range c.cache {
if index%2 == 0 {
cache[key] = value
}
index += 1
}
c.cache = cache
c.lock.Unlock()
}
3 changes: 2 additions & 1 deletion state/postgres_backend.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,8 @@ type PostgresBackend struct {

BackendType null.String // 10+ The process type of this backend

Query null.String // Text of this backend's most recent query
Query null.String // Text of this backend's most recent query
QueryId int64

// Current overall state of this backend. Possible values are:
// - active: The backend is executing a query.
Expand Down
Loading