diff options
Diffstat (limited to 'src/backend')
| -rw-r--r-- | src/backend/catalog/system_functions.sql | 4 | ||||
| -rw-r--r-- | src/backend/catalog/system_views.sql | 27 | ||||
| -rw-r--r-- | src/backend/postmaster/pgstat.c | 656 | ||||
| -rw-r--r-- | src/backend/replication/logical/worker.c | 44 | ||||
| -rw-r--r-- | src/backend/utils/adt/pgstatfuncs.c | 156 |
5 files changed, 389 insertions, 498 deletions
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql index fd1421788e6..758ab6e25a3 100644 --- a/src/backend/catalog/system_functions.sql +++ b/src/backend/catalog/system_functions.sql @@ -639,9 +639,7 @@ REVOKE EXECUTE ON FUNCTION pg_stat_reset_single_function_counters(oid) FROM publ REVOKE EXECUTE ON FUNCTION pg_stat_reset_replication_slot(text) FROM public; -REVOKE EXECUTE ON FUNCTION pg_stat_reset_subscription_worker(oid) FROM public; - -REVOKE EXECUTE ON FUNCTION pg_stat_reset_subscription_worker(oid, oid) FROM public; +REVOKE EXECUTE ON FUNCTION pg_stat_reset_subscription_stats(oid) FROM public; REVOKE EXECUTE ON FUNCTION lo_import(text) FROM public; diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 3cb69b1f87b..40b7bca5a96 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1264,25 +1264,12 @@ GRANT SELECT (oid, subdbid, subname, subowner, subenabled, subbinary, substream, subtwophasestate, subslotname, subsynccommit, subpublications) ON pg_subscription TO public; -CREATE VIEW pg_stat_subscription_workers AS +CREATE VIEW pg_stat_subscription_stats AS SELECT - w.subid, + ss.subid, s.subname, - w.subrelid, - w.last_error_relid, - w.last_error_command, - w.last_error_xid, - w.last_error_count, - w.last_error_message, - w.last_error_time - FROM (SELECT - oid as subid, - NULL as relid - FROM pg_subscription - UNION ALL - SELECT - srsubid as subid, - srrelid as relid - FROM pg_subscription_rel) sr, - LATERAL pg_stat_get_subscription_worker(sr.subid, sr.relid) w - JOIN pg_subscription s ON (w.subid = s.oid); + ss.apply_error_count, + ss.sync_error_count, + ss.stats_reset + FROM pg_subscription as s, + pg_stat_get_subscription_stats(s.oid) as ss; diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 0646f530985..53ddd930e6e 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -106,7 +106,7 @@ #define PGSTAT_DB_HASH_SIZE 16 #define PGSTAT_TAB_HASH_SIZE 512 #define PGSTAT_FUNCTION_HASH_SIZE 512 -#define PGSTAT_SUBWORKER_HASH_SIZE 32 +#define PGSTAT_SUBSCRIPTION_HASH_SIZE 32 #define PGSTAT_REPLSLOT_HASH_SIZE 32 @@ -284,6 +284,7 @@ static PgStat_GlobalStats globalStats; static PgStat_WalStats walStats; static PgStat_SLRUStats slruStats[SLRU_NUM_ELEMENTS]; static HTAB *replSlotStatHash = NULL; +static HTAB *subscriptionStatHash = NULL; /* * List of OIDs of databases we need to write out. If an entry is InvalidOid, @@ -322,14 +323,13 @@ NON_EXEC_STATIC void PgstatCollectorMain(int argc, char *argv[]) pg_attribute_no static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid, bool create); static PgStat_StatTabEntry *pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create); -static PgStat_StatSubWorkerEntry *pgstat_get_subworker_entry(PgStat_StatDBEntry *dbentry, - Oid subid, Oid subrelid, - bool create); +static PgStat_StatSubEntry *pgstat_get_subscription_entry(Oid subid, bool create); +static void pgstat_reset_subscription(PgStat_StatSubEntry *subentry, TimestampTz ts); static void pgstat_write_statsfiles(bool permanent, bool allDbs); static void pgstat_write_db_statsfile(PgStat_StatDBEntry *dbentry, bool permanent); static HTAB *pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep); static void pgstat_read_db_statsfile(Oid databaseid, HTAB *tabhash, HTAB *funchash, - HTAB *subworkerhash, bool permanent); + bool permanent); static void backend_read_statsfile(void); static bool pgstat_write_statsfile_needed(void); @@ -341,7 +341,6 @@ static void pgstat_reset_replslot(PgStat_StatReplSlotEntry *slotstats, Timestamp static void pgstat_send_tabstat(PgStat_MsgTabstat *tsmsg, TimestampTz now); static void pgstat_send_funcstats(void); static void pgstat_send_slru(void); -static void pgstat_send_subscription_purge(PgStat_MsgSubscriptionPurge *msg); static HTAB *pgstat_collect_oids(Oid catalogid, AttrNumber anum_oid); static bool pgstat_should_report_connstat(void); static void pgstat_report_disconnect(Oid dboid); @@ -363,6 +362,7 @@ static void pgstat_recv_resetsharedcounter(PgStat_MsgResetsharedcounter *msg, in static void pgstat_recv_resetsinglecounter(PgStat_MsgResetsinglecounter *msg, int len); static void pgstat_recv_resetslrucounter(PgStat_MsgResetslrucounter *msg, int len); static void pgstat_recv_resetreplslotcounter(PgStat_MsgResetreplslotcounter *msg, int len); +static void pgstat_recv_resetsubcounter(PgStat_MsgResetsubcounter *msg, int len); static void pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len); static void pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len); static void pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len); @@ -380,8 +380,8 @@ static void pgstat_recv_connect(PgStat_MsgConnect *msg, int len); static void pgstat_recv_disconnect(PgStat_MsgDisconnect *msg, int len); static void pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len); static void pgstat_recv_tempfile(PgStat_MsgTempFile *msg, int len); -static void pgstat_recv_subscription_purge(PgStat_MsgSubscriptionPurge *msg, int len); -static void pgstat_recv_subworker_error(PgStat_MsgSubWorkerError *msg, int len); +static void pgstat_recv_subscription_drop(PgStat_MsgSubscriptionDrop *msg, int len); +static void pgstat_recv_subscription_error(PgStat_MsgSubscriptionError *msg, int len); /* ------------------------------------------------------------ * Public functions called from postmaster follow @@ -1188,6 +1188,32 @@ pgstat_vacuum_stat(void) } /* + * Repeat the above steps for subscriptions, if subscription stats are + * being collected. + */ + if (subscriptionStatHash) + { + PgStat_StatSubEntry *subentry; + + /* + * Read pg_subscription and make a list of OIDs of all existing + * subscriptions. + */ + htab = pgstat_collect_oids(SubscriptionRelationId, Anum_pg_subscription_oid); + + hash_seq_init(&hstat, subscriptionStatHash); + while ((subentry = (PgStat_StatSubEntry *) hash_seq_search(&hstat)) != NULL) + { + CHECK_FOR_INTERRUPTS(); + + if (hash_search(htab, (void *) &(subentry->subid), HASH_FIND, NULL) == NULL) + pgstat_report_subscription_drop(subentry->subid); + } + + hash_destroy(htab); + } + + /* * Lookup our own database entry; if not found, nothing more to do. */ dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash, @@ -1311,74 +1337,6 @@ pgstat_vacuum_stat(void) hash_destroy(htab); } - - /* - * Repeat for subscription workers. Similarly, we needn't bother in the - * common case where no subscription workers' stats are being collected. - */ - if (dbentry->subworkers != NULL && - hash_get_num_entries(dbentry->subworkers) > 0) - { - PgStat_StatSubWorkerEntry *subwentry; - PgStat_MsgSubscriptionPurge spmsg; - - /* - * Read pg_subscription and make a list of OIDs of all existing - * subscriptions - */ - htab = pgstat_collect_oids(SubscriptionRelationId, Anum_pg_subscription_oid); - - spmsg.m_databaseid = MyDatabaseId; - spmsg.m_nentries = 0; - - hash_seq_init(&hstat, dbentry->subworkers); - while ((subwentry = (PgStat_StatSubWorkerEntry *) hash_seq_search(&hstat)) != NULL) - { - bool exists = false; - Oid subid = subwentry->key.subid; - - CHECK_FOR_INTERRUPTS(); - - if (hash_search(htab, (void *) &subid, HASH_FIND, NULL) != NULL) - continue; - - /* - * It is possible that we have multiple entries for the - * subscription corresponding to apply worker and tablesync - * workers. In such cases, we don't need to add the same subid - * again. - */ - for (int i = 0; i < spmsg.m_nentries; i++) - { - if (spmsg.m_subids[i] == subid) - { - exists = true; - break; - } - } - - if (exists) - continue; - - /* This subscription is dead, add the subid to the message */ - spmsg.m_subids[spmsg.m_nentries++] = subid; - - /* - * If the message is full, send it out and reinitialize to empty - */ - if (spmsg.m_nentries >= PGSTAT_NUM_SUBSCRIPTIONPURGE) - { - pgstat_send_subscription_purge(&spmsg); - spmsg.m_nentries = 0; - } - } - - /* Send the rest of dead subscriptions */ - if (spmsg.m_nentries > 0) - pgstat_send_subscription_purge(&spmsg); - - hash_destroy(htab); - } } @@ -1551,8 +1509,7 @@ pgstat_reset_shared_counters(const char *target) * ---------- */ void -pgstat_reset_single_counter(Oid objoid, Oid subobjoid, - PgStat_Single_Reset_Type type) +pgstat_reset_single_counter(Oid objoid, PgStat_Single_Reset_Type type) { PgStat_MsgResetsinglecounter msg; @@ -1563,7 +1520,6 @@ pgstat_reset_single_counter(Oid objoid, Oid subobjoid, msg.m_databaseid = MyDatabaseId; msg.m_resettype = type; msg.m_objectid = objoid; - msg.m_subobjectid = subobjoid; pgstat_send(&msg, sizeof(msg)); } @@ -1624,6 +1580,30 @@ pgstat_reset_replslot_counter(const char *name) } /* ---------- + * pgstat_reset_subscription_counter() - + * + * Tell the statistics collector to reset a single subscription + * counter, or all subscription counters (when subid is InvalidOid). + * + * Permission checking for this function is managed through the normal + * GRANT system. + * ---------- + */ +void +pgstat_reset_subscription_counter(Oid subid) +{ + PgStat_MsgResetsubcounter msg; + + if (pgStatSock == PGINVALID_SOCKET) + return; + + msg.m_subid = subid; + pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RESETSUBCOUNTER); + + pgstat_send(&msg, sizeof(msg)); +} + +/* ---------- * pgstat_report_autovac() - * * Called from autovacuum.c to report startup of an autovacuum process. @@ -1949,31 +1929,20 @@ pgstat_report_replslot_drop(const char *slotname) } /* ---------- - * pgstat_report_subworker_error() - + * pgstat_report_subscription_error() - * - * Tell the collector about the subscription worker error. + * Tell the collector about the subscription error. * ---------- */ void -pgstat_report_subworker_error(Oid subid, Oid subrelid, Oid relid, - LogicalRepMsgType command, TransactionId xid, - const char *errmsg) +pgstat_report_subscription_error(Oid subid, bool is_apply_error) { - PgStat_MsgSubWorkerError msg; - int len; + PgStat_MsgSubscriptionError msg; - pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBWORKERERROR); - msg.m_databaseid = MyDatabaseId; + pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONERROR); msg.m_subid = subid; - msg.m_subrelid = subrelid; - msg.m_relid = relid; - msg.m_command = command; - msg.m_xid = xid; - msg.m_timestamp = GetCurrentTimestamp(); - strlcpy(msg.m_message, errmsg, PGSTAT_SUBWORKERERROR_MSGLEN); - - len = offsetof(PgStat_MsgSubWorkerError, m_message) + strlen(msg.m_message) + 1; - pgstat_send(&msg, len); + msg.m_is_apply_error = is_apply_error; + pgstat_send(&msg, sizeof(PgStat_MsgSubscriptionError)); } /* ---------- @@ -1985,12 +1954,11 @@ pgstat_report_subworker_error(Oid subid, Oid subrelid, Oid relid, void pgstat_report_subscription_drop(Oid subid) { - PgStat_MsgSubscriptionPurge msg; + PgStat_MsgSubscriptionDrop msg; - msg.m_databaseid = MyDatabaseId; - msg.m_subids[0] = subid; - msg.m_nentries = 1; - pgstat_send_subscription_purge(&msg); + pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONDROP); + msg.m_subid = subid; + pgstat_send(&msg, sizeof(PgStat_MsgSubscriptionDrop)); } /* ---------- @@ -3000,36 +2968,6 @@ pgstat_fetch_stat_funcentry(Oid func_id) /* * --------- - * pgstat_fetch_stat_subworker_entry() - - * - * Support function for the SQL-callable pgstat* functions. Returns - * the collected statistics for subscription worker or NULL. - * --------- - */ -PgStat_StatSubWorkerEntry * -pgstat_fetch_stat_subworker_entry(Oid subid, Oid subrelid) -{ - PgStat_StatDBEntry *dbentry; - PgStat_StatSubWorkerEntry *wentry = NULL; - - /* Load the stats file if needed */ - backend_read_statsfile(); - - /* - * Lookup our database, then find the requested subscription worker stats. - */ - dbentry = pgstat_fetch_stat_dbentry(MyDatabaseId); - if (dbentry != NULL && dbentry->subworkers != NULL) - { - wentry = pgstat_get_subworker_entry(dbentry, subid, subrelid, - false); - } - - return wentry; -} - -/* - * --------- * pgstat_fetch_stat_archiver() - * * Support function for the SQL-callable pgstat* functions. Returns @@ -3141,6 +3079,23 @@ pgstat_fetch_replslot(NameData slotname) } /* + * --------- + * pgstat_fetch_stat_subscription() - + * + * Support function for the SQL-callable pgstat* functions. Returns + * the collected statistics for one subscription or NULL. + * --------- + */ +PgStat_StatSubEntry * +pgstat_fetch_stat_subscription(Oid subid) +{ + /* Load the stats file if needed */ + backend_read_statsfile(); + + return pgstat_get_subscription_entry(subid, false); +} + +/* * Shut down a single backend's statistics reporting at process exit. * * Flush any remaining statistics counts out to the collector. @@ -3465,24 +3420,6 @@ pgstat_send_slru(void) } } -/* -------- - * pgstat_send_subscription_purge() - - * - * Send a subscription purge message to the collector - * -------- - */ -static void -pgstat_send_subscription_purge(PgStat_MsgSubscriptionPurge *msg) -{ - int len; - - len = offsetof(PgStat_MsgSubscriptionPurge, m_subids[0]) - + msg->m_nentries * sizeof(Oid); - - pgstat_setheader(&msg->m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONPURGE); - pgstat_send(msg, len); -} - /* ---------- * PgstatCollectorMain() - * @@ -3668,6 +3605,10 @@ PgstatCollectorMain(int argc, char *argv[]) len); break; + case PGSTAT_MTYPE_RESETSUBCOUNTER: + pgstat_recv_resetsubcounter(&msg.msg_resetsubcounter, len); + break; + case PGSTAT_MTYPE_AUTOVAC_START: pgstat_recv_autovac(&msg.msg_autovacuum_start, len); break; @@ -3738,12 +3679,12 @@ PgstatCollectorMain(int argc, char *argv[]) pgstat_recv_disconnect(&msg.msg_disconnect, len); break; - case PGSTAT_MTYPE_SUBSCRIPTIONPURGE: - pgstat_recv_subscription_purge(&msg.msg_subscriptionpurge, len); + case PGSTAT_MTYPE_SUBSCRIPTIONDROP: + pgstat_recv_subscription_drop(&msg.msg_subscriptiondrop, len); break; - case PGSTAT_MTYPE_SUBWORKERERROR: - pgstat_recv_subworker_error(&msg.msg_subworkererror, len); + case PGSTAT_MTYPE_SUBSCRIPTIONERROR: + pgstat_recv_subscription_error(&msg.msg_subscriptionerror, len); break; default: @@ -3791,8 +3732,7 @@ PgstatCollectorMain(int argc, char *argv[]) /* * Subroutine to clear stats in a database entry * - * Tables, functions, and subscription workers hashes are initialized - * to empty. + * Tables and functions hashes are initialized to empty. */ static void reset_dbentry_counters(PgStat_StatDBEntry *dbentry) @@ -3845,13 +3785,6 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry) PGSTAT_FUNCTION_HASH_SIZE, &hash_ctl, HASH_ELEM | HASH_BLOBS); - - hash_ctl.keysize = sizeof(PgStat_StatSubWorkerKey); - hash_ctl.entrysize = sizeof(PgStat_StatSubWorkerEntry); - dbentry->subworkers = hash_create("Per-database subscription worker", - PGSTAT_SUBWORKER_HASH_SIZE, - &hash_ctl, - HASH_ELEM | HASH_BLOBS); } /* @@ -3876,7 +3809,7 @@ pgstat_get_db_entry(Oid databaseid, bool create) /* * If not found, initialize the new one. This creates empty hash tables - * for tables, functions, and subscription workers, too. + * for tables and functions, too. */ if (!found) reset_dbentry_counters(result); @@ -3935,48 +3868,6 @@ pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create) } /* ---------- - * pgstat_get_subworker_entry - * - * Return subscription worker entry with the given subscription OID and - * relation OID. If subrelid is InvalidOid, it returns an entry of the - * apply worker otherwise returns an entry of the table sync worker - * associated with subrelid. If no subscription worker entry exists, - * initialize it, if the create parameter is true. Else, return NULL. - * ---------- - */ -static PgStat_StatSubWorkerEntry * -pgstat_get_subworker_entry(PgStat_StatDBEntry *dbentry, Oid subid, Oid subrelid, - bool create) -{ - PgStat_StatSubWorkerEntry *subwentry; - PgStat_StatSubWorkerKey key; - bool found; - HASHACTION action = (create ? HASH_ENTER : HASH_FIND); - - key.subid = subid; - key.subrelid = subrelid; - subwentry = (PgStat_StatSubWorkerEntry *) hash_search(dbentry->subworkers, - (void *) &key, - action, &found); - - if (!create && !found) - return NULL; - - /* If not found, initialize the new one */ - if (!found) - { - subwentry->last_error_relid = InvalidOid; - subwentry->last_error_command = 0; - subwentry->last_error_xid = InvalidTransactionId; - subwentry->last_error_count = 0; - subwentry->last_error_time = 0; - subwentry->last_error_message[0] = '\0'; - } - - return subwentry; -} - -/* ---------- * pgstat_write_statsfiles() - * Write the global statistics file, as well as requested DB files. * @@ -4059,8 +3950,8 @@ pgstat_write_statsfiles(bool permanent, bool allDbs) while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL) { /* - * Write out the table, function, and subscription-worker stats for - * this DB into the appropriate per-DB stat file, if required. + * Write out the table and function stats for this DB into the + * appropriate per-DB stat file, if required. */ if (allDbs || pgstat_db_requested(dbentry->databaseid)) { @@ -4096,6 +3987,22 @@ pgstat_write_statsfiles(bool permanent, bool allDbs) } /* + * Write subscription stats struct + */ + if (subscriptionStatHash) + { + PgStat_StatSubEntry *subentry; + + hash_seq_init(&hstat, subscriptionStatHash); + while ((subentry = (PgStat_StatSubEntry *) hash_seq_search(&hstat)) != NULL) + { + fputc('S', fpout); + rc = fwrite(subentry, sizeof(PgStat_StatSubEntry), 1, fpout); + (void) rc; /* we'll check for error with ferror */ + } + } + + /* * No more output to be done. Close the temp file and replace the old * pgstat.stat with it. The ferror() check replaces testing for error * after each individual fputc or fwrite above. @@ -4174,10 +4081,8 @@ pgstat_write_db_statsfile(PgStat_StatDBEntry *dbentry, bool permanent) { HASH_SEQ_STATUS tstat; HASH_SEQ_STATUS fstat; - HASH_SEQ_STATUS sstat; PgStat_StatTabEntry *tabentry; PgStat_StatFuncEntry *funcentry; - PgStat_StatSubWorkerEntry *subwentry; FILE *fpout; int32 format_id; Oid dbid = dbentry->databaseid; @@ -4233,17 +4138,6 @@ pgstat_write_db_statsfile(PgStat_StatDBEntry *dbentry, bool permanent) } /* - * Walk through the database's subscription worker stats table. - */ - hash_seq_init(&sstat, dbentry->subworkers); - while ((subwentry = (PgStat_StatSubWorkerEntry *) hash_seq_search(&sstat)) != NULL) - { - fputc('S', fpout); - rc = fwrite(subwentry, sizeof(PgStat_StatSubWorkerEntry), 1, fpout); - (void) rc; /* we'll check for error with ferror */ - } - - /* * No more output to be done. Close the temp file and replace the old * pgstat.stat with it. The ferror() check replaces testing for error * after each individual fputc or fwrite above. @@ -4301,9 +4195,8 @@ pgstat_write_db_statsfile(PgStat_StatDBEntry *dbentry, bool permanent) * files after reading; the in-memory status is now authoritative, and the * files would be out of date in case somebody else reads them. * - * If a 'deep' read is requested, table/function/subscription-worker stats are - * read, otherwise the table/function/subscription-worker hash tables remain - * empty. + * If a 'deep' read is requested, table/function stats are read, otherwise + * the table/function hash tables remain empty. * ---------- */ static HTAB * @@ -4482,7 +4375,6 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep) memcpy(dbentry, &dbbuf, sizeof(PgStat_StatDBEntry)); dbentry->tables = NULL; dbentry->functions = NULL; - dbentry->subworkers = NULL; /* * In the collector, disregard the timestamp we read from the @@ -4494,8 +4386,8 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep) dbentry->stats_timestamp = 0; /* - * Don't create tables/functions/subworkers hashtables for - * uninteresting databases. + * Don't create tables/functions hashtables for uninteresting + * databases. */ if (onlydb != InvalidOid) { @@ -4520,14 +4412,6 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep) &hash_ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - hash_ctl.keysize = sizeof(PgStat_StatSubWorkerKey); - hash_ctl.entrysize = sizeof(PgStat_StatSubWorkerEntry); - hash_ctl.hcxt = pgStatLocalContext; - dbentry->subworkers = hash_create("Per-database subscription worker", - PGSTAT_SUBWORKER_HASH_SIZE, - &hash_ctl, - HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - /* * If requested, read the data from the database-specific * file. Otherwise we just leave the hashtables empty. @@ -4536,7 +4420,6 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep) pgstat_read_db_statsfile(dbentry->databaseid, dbentry->tables, dbentry->functions, - dbentry->subworkers, permanent); break; @@ -4580,6 +4463,45 @@ pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep) break; } + /* + * 'S' A PgStat_StatSubEntry struct describing subscription + * statistics. + */ + case 'S': + { + PgStat_StatSubEntry subbuf; + PgStat_StatSubEntry *subentry; + + if (fread(&subbuf, 1, sizeof(PgStat_StatSubEntry), fpin) + != sizeof(PgStat_StatSubEntry)) + { + ereport(pgStatRunningInCollector ? LOG : WARNING, + (errmsg("corrupted statistics file \"%s\"", + statfile))); + goto done; + } + + if (subscriptionStatHash == NULL) + { + HASHCTL hash_ctl; + + hash_ctl.keysize = sizeof(Oid); + hash_ctl.entrysize = sizeof(PgStat_StatSubEntry); + hash_ctl.hcxt = pgStatLocalContext; + subscriptionStatHash = hash_create("Subscription hash", + PGSTAT_SUBSCRIPTION_HASH_SIZE, + &hash_ctl, + HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); + } + + subentry = (PgStat_StatSubEntry *) hash_search(subscriptionStatHash, + (void *) &subbuf.subid, + HASH_ENTER, NULL); + + memcpy(subentry, &subbuf, sizeof(subbuf)); + break; + } + case 'E': goto done; @@ -4614,21 +4536,19 @@ done: * As in pgstat_read_statsfiles, if the permanent file is requested, it is * removed after reading. * - * Note: this code has the ability to skip storing per-table, per-function, or - * per-subscription-worker data, if NULL is passed for the corresponding hashtable. - * That's not used at the moment though. + * Note: this code has the ability to skip storing per-table or per-function + * data, if NULL is passed for the corresponding hashtable. That's not used + * at the moment though. * ---------- */ static void pgstat_read_db_statsfile(Oid databaseid, HTAB *tabhash, HTAB *funchash, - HTAB *subworkerhash, bool permanent) + bool permanent) { PgStat_StatTabEntry *tabentry; PgStat_StatTabEntry tabbuf; PgStat_StatFuncEntry funcbuf; PgStat_StatFuncEntry *funcentry; - PgStat_StatSubWorkerEntry subwbuf; - PgStat_StatSubWorkerEntry *subwentry; FILE *fpin; int32 format_id; bool found; @@ -4743,41 +4663,6 @@ pgstat_read_db_statsfile(Oid databaseid, HTAB *tabhash, HTAB *funchash, break; /* - * 'S' A PgStat_StatSubWorkerEntry struct describing - * subscription worker statistics. - */ - case 'S': - if (fread(&subwbuf, 1, sizeof(PgStat_StatSubWorkerEntry), - fpin) != sizeof(PgStat_StatSubWorkerEntry)) - { - ereport(pgStatRunningInCollector ? LOG : WARNING, - (errmsg("corrupted statistics file \"%s\"", - statfile))); - goto done; - } - - /* - * Skip if subscription worker data not wanted. - */ - if (subworkerhash == NULL) - break; - - subwentry = (PgStat_StatSubWorkerEntry *) hash_search(subworkerhash, - (void *) &subwbuf.key, - HASH_ENTER, &found); - - if (found) - { - ereport(pgStatRunningInCollector ? LOG : WARNING, - (errmsg("corrupted statistics file \"%s\"", - statfile))); - goto done; - } - - memcpy(subwentry, &subwbuf, sizeof(subwbuf)); - break; - - /* * 'E' The EOF marker of a complete stats file. */ case 'E': @@ -4829,6 +4714,7 @@ pgstat_read_db_statsfile_timestamp(Oid databaseid, bool permanent, PgStat_WalStats myWalStats; PgStat_SLRUStats mySLRUStats[SLRU_NUM_ELEMENTS]; PgStat_StatReplSlotEntry myReplSlotStats; + PgStat_StatSubEntry mySubStats; FILE *fpin; int32 format_id; const char *statfile = permanent ? PGSTAT_STAT_PERMANENT_FILENAME : pgstat_stat_filename; @@ -4959,6 +4845,22 @@ pgstat_read_db_statsfile_timestamp(Oid databaseid, bool permanent, } break; + /* + * 'S' A PgStat_StatSubEntry struct describing subscription + * statistics follows. + */ + case 'S': + if (fread(&mySubStats, 1, sizeof(PgStat_StatSubEntry), fpin) + != sizeof(PgStat_StatSubEntry)) + { + ereport(pgStatRunningInCollector ? LOG : WARNING, + (errmsg("corrupted statistics file \"%s\"", + statfile))); + FreeFile(fpin); + return false; + } + break; + case 'E': goto done; @@ -5164,6 +5066,7 @@ pgstat_clear_snapshot(void) pgStatLocalContext = NULL; pgStatDBHash = NULL; replSlotStatHash = NULL; + subscriptionStatHash = NULL; /* * Historically the backend_status.c facilities lived in this file, and @@ -5450,8 +5353,6 @@ pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len) hash_destroy(dbentry->tables); if (dbentry->functions != NULL) hash_destroy(dbentry->functions); - if (dbentry->subworkers != NULL) - hash_destroy(dbentry->subworkers); if (hash_search(pgStatDBHash, (void *) &dbid, @@ -5489,16 +5390,13 @@ pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len) hash_destroy(dbentry->tables); if (dbentry->functions != NULL) hash_destroy(dbentry->functions); - if (dbentry->subworkers != NULL) - hash_destroy(dbentry->subworkers); dbentry->tables = NULL; dbentry->functions = NULL; - dbentry->subworkers = NULL; /* * Reset database-level stats, too. This creates empty hash tables for - * tables, functions, and subscription workers. + * tables and functions. */ reset_dbentry_counters(dbentry); } @@ -5567,14 +5465,6 @@ pgstat_recv_resetsinglecounter(PgStat_MsgResetsinglecounter *msg, int len) else if (msg->m_resettype == RESET_FUNCTION) (void) hash_search(dbentry->functions, (void *) &(msg->m_objectid), HASH_REMOVE, NULL); - else if (msg->m_resettype == RESET_SUBWORKER) - { - PgStat_StatSubWorkerKey key; - - key.subid = msg->m_objectid; - key.subrelid = msg->m_subobjectid; - (void) hash_search(dbentry->subworkers, (void *) &key, HASH_REMOVE, NULL); - } } /* ---------- @@ -5645,6 +5535,51 @@ pgstat_recv_resetreplslotcounter(PgStat_MsgResetreplslotcounter *msg, } } +/* ---------- + * pgstat_recv_resetsubcounter() - + * + * Reset some subscription statistics of the cluster. + * ---------- + */ +static void +pgstat_recv_resetsubcounter(PgStat_MsgResetsubcounter *msg, int len) +{ + PgStat_StatSubEntry *subentry; + TimestampTz ts; + + /* Return if we don't have replication subscription statistics */ + if (subscriptionStatHash == NULL) + return; + + ts = GetCurrentTimestamp(); + if (!OidIsValid(msg->m_subid)) + { + HASH_SEQ_STATUS sstat; + + /* Clear all subscription counters */ + hash_seq_init(&sstat, subscriptionStatHash); + while ((subentry = (PgStat_StatSubEntry *) hash_seq_search(&sstat)) != NULL) + pgstat_reset_subscription(subentry, ts); + } + else + { + /* Get the subscription statistics to reset */ + subentry = pgstat_get_subscription_entry(msg->m_subid, false); + + /* + * Nothing to do if the given subscription entry is not found. This + * could happen when the subscription with the subid is removed and + * the corresponding statistics entry is also removed before receiving + * the reset message. + */ + if (!subentry) + return; + + /* Reset the stats for the requested subscription */ + pgstat_reset_subscription(subentry, ts); + } +} + /* ---------- * pgstat_recv_autovac() - @@ -6118,81 +6053,42 @@ pgstat_recv_funcpurge(PgStat_MsgFuncpurge *msg, int len) } /* ---------- - * pgstat_recv_subscription_purge() - + * pgstat_recv_subscription_drop() - * - * Process a SUBSCRIPTIONPURGE message. + * Process a SUBSCRIPTIONDROP message. * ---------- */ static void -pgstat_recv_subscription_purge(PgStat_MsgSubscriptionPurge *msg, int len) +pgstat_recv_subscription_drop(PgStat_MsgSubscriptionDrop *msg, int len) { - HASH_SEQ_STATUS hstat; - PgStat_StatDBEntry *dbentry; - PgStat_StatSubWorkerEntry *subwentry; - - dbentry = pgstat_get_db_entry(msg->m_databaseid, false); - - /* No need to purge if we don't even know the database */ - if (!dbentry || !dbentry->subworkers) + /* Return if we don't have replication subscription statistics */ + if (subscriptionStatHash == NULL) return; - /* Remove all subscription worker statistics for the given subscriptions */ - hash_seq_init(&hstat, dbentry->subworkers); - while ((subwentry = (PgStat_StatSubWorkerEntry *) hash_seq_search(&hstat)) != NULL) - { - for (int i = 0; i < msg->m_nentries; i++) - { - if (subwentry->key.subid == msg->m_subids[i]) - { - (void) hash_search(dbentry->subworkers, (void *) &(subwentry->key), - HASH_REMOVE, NULL); - break; - } - } - } + /* Remove from hashtable if present; we don't care if it's not */ + (void) hash_search(subscriptionStatHash, (void *) &(msg->m_subid), + HASH_REMOVE, NULL); } /* ---------- - * pgstat_recv_subworker_error() - + * pgstat_recv_subscription_error() - * - * Process a SUBWORKERERROR message. + * Process a SUBSCRIPTIONERROR message. * ---------- */ static void -pgstat_recv_subworker_error(PgStat_MsgSubWorkerError *msg, int len) +pgstat_recv_subscription_error(PgStat_MsgSubscriptionError *msg, int len) { - PgStat_StatDBEntry *dbentry; - PgStat_StatSubWorkerEntry *subwentry; - - dbentry = pgstat_get_db_entry(msg->m_databaseid, true); + PgStat_StatSubEntry *subentry; - /* Get the subscription worker stats */ - subwentry = pgstat_get_subworker_entry(dbentry, msg->m_subid, - msg->m_subrelid, true); - Assert(subwentry); - - if (subwentry->last_error_relid == msg->m_relid && - subwentry->last_error_command == msg->m_command && - subwentry->last_error_xid == msg->m_xid && - strcmp(subwentry->last_error_message, msg->m_message) == 0) - { - /* - * The same error occurred again in succession, just update its - * timestamp and count. - */ - subwentry->last_error_count++; - subwentry->last_error_time = msg->m_timestamp; - return; - } + /* Get the subscription stats */ + subentry = pgstat_get_subscription_entry(msg->m_subid, true); + Assert(subentry); - /* Otherwise, update the error information */ - subwentry->last_error_relid = msg->m_relid; - subwentry->last_error_command = msg->m_command; - subwentry->last_error_xid = msg->m_xid; - subwentry->last_error_count = 1; - subwentry->last_error_time = msg->m_timestamp; - strlcpy(subwentry->last_error_message, msg->m_message, - PGSTAT_SUBWORKERERROR_MSGLEN); + if (msg->m_is_apply_error) + subentry->apply_error_count++; + else + subentry->sync_error_count++; } /* ---------- @@ -6313,6 +6209,68 @@ pgstat_reset_replslot(PgStat_StatReplSlotEntry *slotent, TimestampTz ts) slotent->stat_reset_timestamp = ts; } +/* ---------- + * pgstat_get_subscription_entry + * + * Return the subscription statistics entry with the given subscription OID. + * If no subscription entry exists, initialize it, if the create parameter is + * true. Else, return NULL. + * ---------- + */ +static PgStat_StatSubEntry * +pgstat_get_subscription_entry(Oid subid, bool create) +{ + PgStat_StatSubEntry *subentry; + bool found; + HASHACTION action = (create ? HASH_ENTER : HASH_FIND); + + if (subscriptionStatHash == NULL) + { + HASHCTL hash_ctl; + + /* + * Quick return NULL if the hash table is empty and the caller didn't + * request to create the entry. + */ + if (!create) + return NULL; + + hash_ctl.keysize = sizeof(Oid); + hash_ctl.entrysize = sizeof(PgStat_StatSubEntry); + subscriptionStatHash = hash_create("Subscription hash", + PGSTAT_SUBSCRIPTION_HASH_SIZE, + &hash_ctl, + HASH_ELEM | HASH_BLOBS); + } + + subentry = (PgStat_StatSubEntry *) hash_search(subscriptionStatHash, + (void *) &subid, + action, &found); + + if (!create && !found) + return NULL; + + /* If not found, initialize the new one */ + if (!found) + pgstat_reset_subscription(subentry, 0); + + return subentry; +} + +/* ---------- + * pgstat_reset_subscription + * + * Reset the given subscription stats. + * ---------- + */ +static void +pgstat_reset_subscription(PgStat_StatSubEntry *subentry, TimestampTz ts) +{ + subentry->apply_error_count = 0; + subentry->sync_error_count = 0; + subentry->stat_reset_timestamp = ts; +} + /* * pgstat_slru_index * diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 5d9acc61733..7e267f79607 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -3377,7 +3377,6 @@ void ApplyWorkerMain(Datum main_arg) { int worker_slot = DatumGetInt32(main_arg); - MemoryContext cctx = CurrentMemoryContext; MemoryContext oldctx; char originname[NAMEDATALEN]; XLogRecPtr origin_startpos; @@ -3485,20 +3484,15 @@ ApplyWorkerMain(Datum main_arg) } PG_CATCH(); { - MemoryContext ecxt = MemoryContextSwitchTo(cctx); - ErrorData *errdata = CopyErrorData(); - /* - * Report the table sync error. There is no corresponding message - * type for table synchronization. + * Abort the current transaction so that we send the stats message + * in an idle state. */ - pgstat_report_subworker_error(MyLogicalRepWorker->subid, - MyLogicalRepWorker->relid, - MyLogicalRepWorker->relid, - 0, /* message type */ - InvalidTransactionId, - errdata->message); - MemoryContextSwitchTo(ecxt); + AbortOutOfAnyTransaction(); + + /* Report the worker failed during table synchronization */ + pgstat_report_subscription_error(MySubscription->oid, false); + PG_RE_THROW(); } PG_END_TRY(); @@ -3625,22 +3619,14 @@ ApplyWorkerMain(Datum main_arg) } PG_CATCH(); { - /* report the apply error */ - if (apply_error_callback_arg.command != 0) - { - MemoryContext ecxt = MemoryContextSwitchTo(cctx); - ErrorData *errdata = CopyErrorData(); - - pgstat_report_subworker_error(MyLogicalRepWorker->subid, - MyLogicalRepWorker->relid, - apply_error_callback_arg.rel != NULL - ? apply_error_callback_arg.rel->localreloid - : InvalidOid, - apply_error_callback_arg.command, - apply_error_callback_arg.remote_xid, - errdata->message); - MemoryContextSwitchTo(ecxt); - } + /* + * Abort the current transaction so that we send the stats message in + * an idle state. + */ + AbortOutOfAnyTransaction(); + + /* Report the worker failed while applying changes */ + pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker()); PG_RE_THROW(); } diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index 30e8dfa7c12..fd993d0d5fb 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -2163,7 +2163,7 @@ pg_stat_reset_single_table_counters(PG_FUNCTION_ARGS) { Oid taboid = PG_GETARG_OID(0); - pgstat_reset_single_counter(taboid, InvalidOid, RESET_TABLE); + pgstat_reset_single_counter(taboid, RESET_TABLE); PG_RETURN_VOID(); } @@ -2173,38 +2173,11 @@ pg_stat_reset_single_function_counters(PG_FUNCTION_ARGS) { Oid funcoid = PG_GETARG_OID(0); - pgstat_reset_single_counter(funcoid, InvalidOid, RESET_FUNCTION); + pgstat_reset_single_counter(funcoid, RESET_FUNCTION); PG_RETURN_VOID(); } -Datum -pg_stat_reset_subscription_worker_subrel(PG_FUNCTION_ARGS) -{ - Oid subid = PG_GETARG_OID(0); - Oid relid = PG_ARGISNULL(1) ? InvalidOid : PG_GETARG_OID(1); - - pgstat_reset_single_counter(subid, relid, RESET_SUBWORKER); - - PG_RETURN_VOID(); -} - -/* Reset all subscription worker stats associated with the given subscription */ -Datum -pg_stat_reset_subscription_worker_sub(PG_FUNCTION_ARGS) -{ - Oid subid = PG_GETARG_OID(0); - - /* - * Use subscription drop message to remove statistics of all subscription - * workers. - */ - pgstat_report_subscription_drop(subid); - - PG_RETURN_VOID(); -} - - /* Reset SLRU counters (a specific one or all of them). */ Datum pg_stat_reset_slru(PG_FUNCTION_ARGS) @@ -2258,6 +2231,32 @@ pg_stat_reset_replication_slot(PG_FUNCTION_ARGS) PG_RETURN_VOID(); } +/* Reset subscription stats (a specific one or all of them) */ +Datum +pg_stat_reset_subscription_stats(PG_FUNCTION_ARGS) +{ + Oid subid; + + if (PG_ARGISNULL(0)) + { + /* Clear all subscription stats */ + subid = InvalidOid; + } + else + { + subid = PG_GETARG_OID(0); + + if (!OidIsValid(subid)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("invalid subscription OID %u", subid))); + } + + pgstat_reset_subscription_counter(subid); + + PG_RETURN_VOID(); +} + Datum pg_stat_get_archiver(PG_FUNCTION_ARGS) { @@ -2400,50 +2399,32 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS) } /* - * Get the subscription worker statistics for the given subscription - * (and relation). + * Get the subscription statistics for the given subscription. If the + * subscription statistics is not available, return all-zeros stats. */ Datum -pg_stat_get_subscription_worker(PG_FUNCTION_ARGS) +pg_stat_get_subscription_stats(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_SUBSCRIPTION_WORKER_COLS 8 +#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 4 Oid subid = PG_GETARG_OID(0); - Oid subrelid; TupleDesc tupdesc; - Datum values[PG_STAT_GET_SUBSCRIPTION_WORKER_COLS]; - bool nulls[PG_STAT_GET_SUBSCRIPTION_WORKER_COLS]; - PgStat_StatSubWorkerEntry *wentry; - int i; - - if (PG_ARGISNULL(1)) - subrelid = InvalidOid; - else - subrelid = PG_GETARG_OID(1); + Datum values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS]; + bool nulls[PG_STAT_GET_SUBSCRIPTION_STATS_COLS]; + PgStat_StatSubEntry *subentry; + PgStat_StatSubEntry allzero; - /* Get subscription worker stats */ - wentry = pgstat_fetch_stat_subworker_entry(subid, subrelid); - - /* Return NULL if there is no worker statistics */ - if (wentry == NULL) - PG_RETURN_NULL(); + /* Get subscription stats */ + subentry = pgstat_fetch_stat_subscription(subid); /* Initialise attributes information in the tuple descriptor */ - tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBSCRIPTION_WORKER_COLS); + tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBSCRIPTION_STATS_COLS); TupleDescInitEntry(tupdesc, (AttrNumber) 1, "subid", OIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 2, "subrelid", - OIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 3, "last_error_relid", - OIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 4, "last_error_command", - TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 5, "last_error_xid", - XIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 6, "last_error_count", + TupleDescInitEntry(tupdesc, (AttrNumber) 2, "apply_error_count", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 7, "last_error_message", - TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 8, "last_error_time", + TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 4, "stats_reset", TIMESTAMPTZOID, -1, 0); BlessTupleDesc(tupdesc); @@ -2451,46 +2432,27 @@ pg_stat_get_subscription_worker(PG_FUNCTION_ARGS) MemSet(values, 0, sizeof(values)); MemSet(nulls, 0, sizeof(nulls)); - i = 0; - /* subid */ - values[i++] = ObjectIdGetDatum(subid); - - /* subrelid */ - if (OidIsValid(subrelid)) - values[i++] = ObjectIdGetDatum(subrelid); - else - nulls[i++] = true; - - /* last_error_relid */ - if (OidIsValid(wentry->last_error_relid)) - values[i++] = ObjectIdGetDatum(wentry->last_error_relid); - else - nulls[i++] = true; - - /* last_error_command */ - if (wentry->last_error_command != 0) - values[i++] = - CStringGetTextDatum(logicalrep_message_type(wentry->last_error_command)); - else - nulls[i++] = true; + if (!subentry) + { + /* If the subscription is not found, initialise its stats */ + memset(&allzero, 0, sizeof(PgStat_StatSubEntry)); + subentry = &allzero; + } - /* last_error_xid */ - if (TransactionIdIsValid(wentry->last_error_xid)) - values[i++] = TransactionIdGetDatum(wentry->last_error_xid); - else - nulls[i++] = true; + /* subid */ + values[0] = ObjectIdGetDatum(subid); - /* last_error_count */ - values[i++] = Int64GetDatum(wentry->last_error_count); + /* apply_error_count */ + values[1] = Int64GetDatum(subentry->apply_error_count); - /* last_error_message */ - values[i++] = CStringGetTextDatum(wentry->last_error_message); + /* sync_error_count */ + values[2] = Int64GetDatum(subentry->sync_error_count); - /* last_error_time */ - if (wentry->last_error_time != 0) - values[i++] = TimestampTzGetDatum(wentry->last_error_time); + /* stats_reset */ + if (subentry->stat_reset_timestamp == 0) + nulls[3] = true; else - nulls[i++] = true; + values[3] = TimestampTzGetDatum(subentry->stat_reset_timestamp); /* Returns the record as Datum */ PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls))); |