Fix VACUUM's reporting of dead-tuple counts to the stats collector.

Historically, VACUUM has just reported its new_rel_tuples estimate
(the same thing it puts into pg_class.reltuples) to the stats collector.
That number counts both live and dead-but-not-yet-reclaimable tuples.
This behavior may once have been right, but modern versions of the
pgstats code track live and dead tuple counts separately, so putting
the total into n_live_tuples and zero into n_dead_tuples is surely
pretty bogus.  Fix it to report live and dead tuple counts separately.

This doesn't really do much for situations where updating transactions
commit concurrently with a VACUUM scan (possibly causing double-counting or
omission of the tuples they add or delete); but it's clearly an improvement
over what we were doing before.

Hari Babu, reviewed by Amit Kapila

src/backend/commands/vacuumlazy.c

@@ -106,6 +106,7 @@ typedef struct LVRelStats
 	double		scanned_tuples; /* counts only tuples on scanned pages */
 	double		old_rel_tuples; /* previous value of pg_class.reltuples */
 	double		new_rel_tuples; /* new estimated total # of tuples */
+	double		new_dead_tuples;	/* new estimated total # of dead tuples */
 	BlockNumber pages_removed;
 	double		tuples_deleted;
 	BlockNumber nonempty_pages; /* actually, last nonempty page + 1 */
@@ -185,6 +186,7 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 	BlockNumber new_rel_pages;
 	double		new_rel_tuples;
 	BlockNumber new_rel_allvisible;
+	double		new_live_tuples;
 	TransactionId new_frozen_xid;
 	MultiXactId new_min_multi;
 
@@ -307,9 +309,14 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 						new_min_multi);
 
 	/* report results to the stats collector, too */
+	new_live_tuples = new_rel_tuples - vacrelstats->new_dead_tuples;
+	if (new_live_tuples < 0)
+		new_live_tuples = 0;	/* just in case */
+
 	pgstat_report_vacuum(RelationGetRelid(onerel),
 						 onerel->rd_rel->relisshared,
-						 new_rel_tuples);
+						 new_live_tuples,
+						 vacrelstats->new_dead_tuples);
 
 	/* and log the action if appropriate */
 	if (IsAutoVacuumWorkerProcess() && Log_autovacuum_min_duration >= 0)
@@ -334,7 +341,7 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 			ereport(LOG,
 					(errmsg("automatic vacuum of table \"%s.%s.%s\": index scans: %d\n"
 							"pages: %d removed, %d remain\n"
-							"tuples: %.0f removed, %.0f remain\n"
+							"tuples: %.0f removed, %.0f remain, %.0f are dead but not yet removable\n"
 							"buffer usage: %d hits, %d misses, %d dirtied\n"
 					  "avg read rate: %.3f MB/s, avg write rate: %.3f MB/s\n"
 							"system usage: %s",
@@ -346,6 +353,7 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt,
 							vacrelstats->rel_pages,
 							vacrelstats->tuples_deleted,
 							vacrelstats->new_rel_tuples,
+							vacrelstats->new_dead_tuples,
 							VacuumPageHit,
 							VacuumPageMiss,
 							VacuumPageDirty,
@@ -1036,6 +1044,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 	/* save stats for use later */
 	vacrelstats->scanned_tuples = num_tuples;
 	vacrelstats->tuples_deleted = tups_vacuumed;
+	vacrelstats->new_dead_tuples = nkeep;
 
 	/* now we can compute the new value for pg_class.reltuples */
 	vacrelstats->new_rel_tuples = vac_estimate_reltuples(onerel, false,

src/backend/postmaster/pgstat.c

@@ -1327,7 +1327,8 @@ pgstat_report_autovac(Oid dboid)
  * ---------
  */
 void
-pgstat_report_vacuum(Oid tableoid, bool shared, PgStat_Counter tuples)
+pgstat_report_vacuum(Oid tableoid, bool shared,
+					 PgStat_Counter livetuples, PgStat_Counter deadtuples)
 {
 	PgStat_MsgVacuum msg;
 
@@ -1339,7 +1340,8 @@ pgstat_report_vacuum(Oid tableoid, bool shared, PgStat_Counter tuples)
 	msg.m_tableoid = tableoid;
 	msg.m_autovacuum = IsAutoVacuumWorkerProcess();
 	msg.m_vacuumtime = GetCurrentTimestamp();
-	msg.m_tuples = tuples;
+	msg.m_live_tuples = livetuples;
+	msg.m_dead_tuples = deadtuples;
 	pgstat_send(&msg, sizeof(msg));
 }
 
@@ -4809,9 +4811,8 @@ pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
 
 	tabentry = pgstat_get_tab_entry(dbentry, msg->m_tableoid, true);
 
-	tabentry->n_live_tuples = msg->m_tuples;
+	tabentry->n_live_tuples = msg->m_live_tuples;
-	/* Resetting dead_tuples to 0 is an approximation ... */
+	tabentry->n_dead_tuples = msg->m_dead_tuples;
-	tabentry->n_dead_tuples = 0;
 
 	if (msg->m_autovacuum)
 	{

src/include/pgstat.h

@@ -333,7 +333,8 @@ typedef struct PgStat_MsgVacuum
 	Oid			m_tableoid;
 	bool		m_autovacuum;
 	TimestampTz m_vacuumtime;
-	PgStat_Counter m_tuples;
+	PgStat_Counter m_live_tuples;
+	PgStat_Counter m_dead_tuples;
 } PgStat_MsgVacuum;
 
 
@@ -775,7 +776,7 @@ extern void pgstat_reset_single_counter(Oid objectid, PgStat_Single_Reset_Type t
 
 extern void pgstat_report_autovac(Oid dboid);
 extern void pgstat_report_vacuum(Oid tableoid, bool shared,
-					 PgStat_Counter tuples);
+					 PgStat_Counter livetuples, PgStat_Counter deadtuples);
 extern void pgstat_report_analyze(Relation rel,
 					  PgStat_Counter livetuples, PgStat_Counter deadtuples);