Report statistics in logical replication workers

petere · petere · commit 139eb9673cb8 · 2017-04-14T14:37:06.000-04:00
Author: Stas Kelvich &lt;s.kelvich@postgrespro.ru&gt;
Author: Petr Jelinek &lt;petr.jelinek@2ndquadrant.com&gt;
Reported-by: Fujii Masao &lt;masao.fujii@gmail.com&gt;
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
@@ -769,9 +769,10 @@ allow_immediate_pgstat_restart(void)
 /* ----------
  * pgstat_report_stat() -
  *
- *	Called from tcop/postgres.c to send the so far collected per-table
- *	and function usage statistics to the collector.  Note that this is
- *	called only when not within a transaction, so it is fair to use
+ *	Must be called by processes that performs DML: tcop/postgres.c, logical
+ *	receiver processes, SPI worker, etc. to send the so far collected
+ *	per-table and function usage statistics to the collector.  Note that this
+ *	is called only when not within a transaction, so it is fair to use
  *	transaction stop time as an approximation of current time.
  * ----------
  */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
@@ -114,9 +114,15 @@ StringInfo	copybuf = NULL;
 static void pg_attribute_noreturn()
 finish_sync_worker(void)
 {
-	/* Commit any outstanding transaction. */
+	/*
+	 * Commit any outstanding transaction. This is the usual case, unless
+	 * there was nothing to do for the table.
+	 */
 	if (IsTransactionState())
+	{
 		CommitTransactionCommand();
+		pgstat_report_stat(false);
+	}
 
 	/* And flush all writes. */
 	XLogFlush(GetXLogWriteRecPtr());
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
@@ -462,6 +462,7 @@ apply_handle_commit(StringInfo s)
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
+	pgstat_report_stat(false);
 	pgstat_report_activity(STATE_IDLE, NULL);
 }
 

Original file line number	Diff line number	Diff line change
`@@ -462,6 +462,7 @@ apply_handle_commit(StringInfo s)`
`462`	`462`	`/* Process any tables that are being synchronized in parallel. */`
`463`	`463`	`process_syncing_tables(commit_data.end_lsn);`
`464`	`464`
	`465`	`+ pgstat_report_stat(false);`
`465`	`466`	`pgstat_report_activity(STATE_IDLE, NULL);`
`466`	`467`	`}`
`467`	`468`