Clean up a number of autovacuum loose ends. Make the stats collector

track shared relations in a separate hashtable, so that operations done
from different databases are counted correctly.  Add proper support for
anti-XID-wraparound vacuuming, even in databases that are never connected
to and so have no stats entries.  Miscellaneous other bug fixes.
Alvaro Herrera, some additional fixes by Tom Lane.

src/backend/access/transam/xlog.c

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/backend/access/transam/xlog.c,v 1.212 2005/07/29 03:25:53 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/access/transam/xlog.c,v 1.213 2005/07/29 19:29:59 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -4913,6 +4913,36 @@ GetRedoRecPtr(void)
 	return RedoRecPtr;
 }
 
+/*
+ * GetRecentNextXid - get the nextXid value saved by the most recent checkpoint
+ *
+ * This is currently used only by the autovacuum daemon.  To check for
+ * impending XID wraparound, autovac needs an approximate idea of the current
+ * XID counter, and it needs it before choosing which DB to attach to, hence
+ * before it sets up a PGPROC, hence before it can take any LWLocks.  But it
+ * has attached to shared memory, and so we can let it reach into the shared
+ * ControlFile structure and pull out the last checkpoint nextXID.
+ *
+ * Since we don't take any sort of lock, we have to assume that reading a
+ * TransactionId is atomic ... but that assumption is made elsewhere, too,
+ * and in any case the worst possible consequence of a bogus result is that
+ * autovac issues an unnecessary database-wide VACUUM.
+ *
+ * Note: we could also choose to read ShmemVariableCache->nextXid in an
+ * unlocked fashion, thus getting a more up-to-date result; but since that
+ * changes far more frequently than the controlfile checkpoint copy, it would
+ * pose a far higher risk of bogus result if we did have a nonatomic-read
+ * problem.
+ *
+ * A (theoretically) completely safe answer is to read the actual pg_control
+ * file into local process memory, but that certainly seems like overkill.
+ */
+TransactionId
+GetRecentNextXid(void)
+{
+	return ControlFile->checkPointCopy.nextXid;
+}
+
 /*
  * This must be called ONCE during postmaster or standalone-backend shutdown
  */

src/backend/commands/analyze.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/analyze.c,v 1.87 2005/07/14 05:13:39 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/analyze.c,v 1.88 2005/07/29 19:30:03 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -317,7 +317,9 @@ analyze_rel(Oid relid, VacuumStmt *vacstmt)
 		 * a zero-column table.
 		 */
 		if (!vacstmt->vacuum)
-			pgstat_report_analyze(RelationGetRelid(onerel), 0, 0);
+			pgstat_report_analyze(RelationGetRelid(onerel),
+								  onerel->rd_rel->relisshared,
+				   				  0, 0);
 
 		vac_close_indexes(nindexes, Irel, AccessShareLock);
 		relation_close(onerel, AccessShareLock);
@@ -436,8 +438,9 @@ analyze_rel(Oid relid, VacuumStmt *vacstmt)
 		}
 
 		/* report results to the stats collector, too */
-		pgstat_report_analyze(RelationGetRelid(onerel), totalrows,
-							  totaldeadrows);
+		pgstat_report_analyze(RelationGetRelid(onerel),
+							  onerel->rd_rel->relisshared,
+			   				  totalrows, totaldeadrows);
 	}
 
 	/* Done with indexes */

src/backend/commands/vacuum.c

@@ -13,7 +13,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/vacuum.c,v 1.311 2005/07/14 05:13:39 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/vacuum.c,v 1.312 2005/07/29 19:30:03 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -41,6 +41,7 @@
 #include "tcop/pquery.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
+#include "utils/flatfiles.h"
 #include "utils/fmgroids.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
@@ -490,7 +491,7 @@ vacuum(VacuumStmt *vacstmt, List *relids)
 		 * If it was a database-wide VACUUM, print FSM usage statistics
 		 * (we don't make you be superuser to see these).
 		 */
-		if (vacstmt->relation == NULL)
+		if (all_rels)
 			PrintFreeSpaceMapStatistics(elevel);
 
 		/*
@@ -712,7 +713,7 @@ vac_update_relstats(Oid relid, BlockNumber num_pages, double num_tuples,
  *	vac_update_dbstats() -- update statistics for one database
  *
  *		Update the whole-database statistics that are kept in its pg_database
- *		row.
+ *		row, and the flat-file copy of pg_database.
  *
  *		We violate no-overwrite semantics here by storing new values for the
  *		statistics columns directly into the tuple that's already on the page.
@@ -721,8 +722,6 @@ vac_update_relstats(Oid relid, BlockNumber num_pages, double num_tuples,
  *
  *		This routine is shared by full and lazy VACUUM.  Note that it is only
  *		applied after a database-wide VACUUM operation.
- *
- *		Note that we don't bother to update the flat-file copy of pg_database.
  */
 static void
 vac_update_dbstats(Oid dbid,
@@ -768,6 +767,9 @@ vac_update_dbstats(Oid dbid,
 	heap_endscan(scan);
 
 	heap_close(relation, RowExclusiveLock);
+
+	/* Mark the flat-file copy of pg_database for update at commit */
+	database_file_update_needed();
 }
 
 
@@ -1165,8 +1167,8 @@ full_vacuum_rel(Relation onerel, VacuumStmt *vacstmt)
 						vacrelstats->rel_tuples, vacrelstats->hasindex);
 
 	/* report results to the stats collector, too */
-	pgstat_report_vacuum(RelationGetRelid(onerel), vacstmt->analyze,
-						 vacrelstats->rel_tuples);
+	pgstat_report_vacuum(RelationGetRelid(onerel), onerel->rd_rel->relisshared,
+		   				 vacstmt->analyze, vacrelstats->rel_tuples);
 }
 
 

src/backend/commands/vacuumlazy.c

@@ -31,7 +31,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/vacuumlazy.c,v 1.55 2005/07/14 05:13:40 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/vacuumlazy.c,v 1.56 2005/07/29 19:30:03 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -182,8 +182,8 @@ lazy_vacuum_rel(Relation onerel, VacuumStmt *vacstmt)
 						hasindex);
 
 	/* report results to the stats collector, too */
-	pgstat_report_vacuum(RelationGetRelid(onerel), vacstmt->analyze,
-						 vacrelstats->rel_tuples);
+	pgstat_report_vacuum(RelationGetRelid(onerel), onerel->rd_rel->relisshared,
+		   				 vacstmt->analyze, vacrelstats->rel_tuples);
 }
 
 

src/backend/libpq/hba.c

@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/libpq/hba.c,v 1.145 2005/07/28 15:30:55 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/libpq/hba.c,v 1.146 2005/07/29 19:30:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -39,6 +39,7 @@
 
 
 #define atooid(x)  ((Oid) strtoul((x), NULL, 10))
+#define atoxid(x)  ((TransactionId) strtoul((x), NULL, 10))
 
 /* Max size of username ident server can return */
 #define IDENT_USERNAME_MAX 512
@@ -999,13 +1000,14 @@ load_hba(void)
  *	dbname: gets database name (must be of size NAMEDATALEN bytes)
  *	dboid: gets database OID
  *	dbtablespace: gets database's default tablespace's OID
+ *	dbfrozenxid: gets database's frozen XID
  *
  * This is not much related to the other functions in hba.c, but we put it
  * here because it uses the next_token() infrastructure.
  */
 bool
-read_pg_database_line(FILE *fp, char *dbname,
-					  Oid *dboid, Oid *dbtablespace)
+read_pg_database_line(FILE *fp, char *dbname, Oid *dboid,
+					  Oid *dbtablespace, TransactionId *dbfrozenxid)
 {
 	char		buf[MAX_TOKEN];
 
@@ -1024,10 +1026,10 @@ read_pg_database_line(FILE *fp, char *dbname,
 	if (!isdigit((unsigned char) buf[0]))
 		elog(FATAL, "bad data in flat pg_database file");
 	*dbtablespace = atooid(buf);
-	/* discard datfrozenxid */
 	next_token(fp, buf, sizeof(buf));
 	if (!isdigit((unsigned char) buf[0]))
 		elog(FATAL, "bad data in flat pg_database file");
+	*dbfrozenxid = atoxid(buf);
 	/* expect EOL next */
 	if (next_token(fp, buf, sizeof(buf)))
 		elog(FATAL, "bad data in flat pg_database file");

src/backend/postmaster/autovacuum.c

@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/postmaster/autovacuum.c,v 1.1 2005/07/14 05:13:40 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/postmaster/autovacuum.c,v 1.2 2005/07/29 19:30:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -23,9 +23,10 @@
 
 #include "access/genam.h"
 #include "access/heapam.h"
+#include "access/xlog.h"
 #include "catalog/indexing.h"
+#include "catalog/namespace.h"
 #include "catalog/pg_autovacuum.h"
-#include "catalog/pg_database.h"
 #include "commands/vacuum.h"
 #include "libpq/hba.h"
 #include "libpq/pqsignal.h"
@@ -68,7 +69,9 @@ typedef struct autovac_dbase
 {
 	Oid				oid;
 	char		   *name;
+	TransactionId	frozenxid;
 	PgStat_StatDBEntry *entry;
+	int32			age;
 } autovac_dbase;
 
 
@@ -76,8 +79,7 @@ typedef struct autovac_dbase
 static pid_t autovac_forkexec(void);
 #endif
 NON_EXEC_STATIC void AutoVacMain(int argc, char *argv[]);
-static void autovac_check_wraparound(void);
-static void do_autovacuum(PgStat_StatDBEntry *dbentry);
+static void do_autovacuum(bool whole_db, PgStat_StatDBEntry *dbentry);
 static List *autovac_get_database_list(void);
 static void test_rel_for_autovac(Oid relid, PgStat_StatTabEntry *tabentry,
 			 Form_pg_class classForm, Form_pg_autovacuum avForm,
@@ -193,7 +195,9 @@ AutoVacMain(int argc, char *argv[])
 {
 	ListCell	   *cell;
 	List		   *dblist;
+	TransactionId	nextXid;
 	autovac_dbase  *db;
+	bool			whole_db;
 	sigjmp_buf		local_sigjmp_buf;
 
 	/* we are a postmaster subprocess now */
@@ -267,23 +271,63 @@ AutoVacMain(int argc, char *argv[])
 	/* Get a list of databases */
 	dblist = autovac_get_database_list();
 
+	/*
+	 * Get the next Xid that was current as of the last checkpoint.
+	 * We need it to determine whether databases are about to need
+	 * database-wide vacuums.
+	 */
+	nextXid = GetRecentNextXid();
+
 	/*
 	 * Choose a database to connect to.  We pick the database that was least
-	 * recently auto-vacuumed.
+	 * recently auto-vacuumed, or one that needs database-wide vacuum (to
+	 * prevent Xid wraparound-related data loss).
+	 *
+	 * Note that a database with no stats entry is not considered, except
+	 * for Xid wraparound purposes.  The theory is that if no one has ever
+	 * connected to it since the stats were last initialized, it doesn't
+	 * need vacuuming.
 	 *
 	 * XXX This could be improved if we had more info about whether it needs
 	 * vacuuming before connecting to it.  Perhaps look through the pgstats
-	 * data for the database's tables?
-	 *
-	 * XXX it is NOT good that we totally ignore databases that have no
-	 * pgstats entry ...
+	 * data for the database's tables?  One idea is to keep track of the
+	 * number of new and dead tuples per database in pgstats.  However it
+	 * isn't clear how to construct a metric that measures that and not
+	 * cause starvation for less busy databases.
 	 */
 	db = NULL;
+	whole_db = false;
 
 	foreach(cell, dblist)
 	{
-		autovac_dbase	*tmp = lfirst(cell);
+		autovac_dbase  *tmp = lfirst(cell);
+		bool			this_whole_db;
+
+		/*
+		 * We look for the database that most urgently needs a database-wide
+		 * vacuum.  We decide that a database-wide vacuum is needed 100000
+		 * transactions sooner than vacuum.c's vac_truncate_clog() would
+		 * decide to start giving warnings.  If any such db is found, we
+		 * ignore all other dbs.
+		 */
+		tmp->age = (int32) (nextXid - tmp->frozenxid);
+		this_whole_db = (tmp->age > (int32) ((MaxTransactionId >> 3) * 3 - 100000));
+		if (whole_db || this_whole_db)
+		{
+			if (!this_whole_db)
+				continue;
+			if (db == NULL || tmp->age > db->age)
+			{
+				db = tmp;
+				whole_db = true;
+			}
+			continue;
+		}
 
+		/*
+		 * Otherwise, skip a database with no pgstat entry; it means it hasn't
+		 * seen any activity.
+		 */
 		tmp->entry = pgstat_fetch_stat_dbentry(tmp->oid);
 		if (!tmp->entry)
 			continue;
@@ -292,12 +336,18 @@ AutoVacMain(int argc, char *argv[])
 		 * Don't try to access a database that was dropped.  This could only
 		 * happen if we read the pg_database flat file right before it was
 		 * modified, after the database was dropped from the pg_database
-		 * table.
+		 * table.  (This is of course a not-very-bulletproof test, but it's
+		 * cheap to make.  If we do mistakenly choose a recently dropped
+		 * database, InitPostgres will fail and we'll drop out until the
+		 * next autovac run.)
 		 */
 		if (tmp->entry->destroy != 0)
 			continue;
 
-		if (!db ||
+		/*
+		 * Else remember the db with oldest autovac time.
+		 */
+		if (db == NULL ||
 			tmp->entry->last_autovac_time < db->entry->last_autovac_time)
 			db = tmp;
 	}
@@ -316,7 +366,7 @@ AutoVacMain(int argc, char *argv[])
 		/*
 		 * And do an appropriate amount of work on it
 		 */
-		do_autovacuum(db->entry);
+		do_autovacuum(whole_db, db->entry);
 	}
 
 	/* One iteration done, go away */
@@ -338,6 +388,7 @@ autovac_get_database_list(void)
 	FILE   *db_file;
 	Oid		db_id;
 	Oid		db_tablespace;
+	TransactionId db_frozenxid;
 
 	filename = database_getflatfilename();
 	db_file = AllocateFile(filename, "r");
@@ -346,7 +397,8 @@ autovac_get_database_list(void)
 				(errcode_for_file_access(),
 				 errmsg("could not open file \"%s\": %m", filename)));
 
-	while (read_pg_database_line(db_file, thisname, &db_id, &db_tablespace))
+	while (read_pg_database_line(db_file, thisname, &db_id,
+								 &db_tablespace, &db_frozenxid))
 	{
 		autovac_dbase	*db;
 
@@ -354,8 +406,10 @@ autovac_get_database_list(void)
 
 		db->oid = db_id;
 		db->name = pstrdup(thisname);
-		/* this gets set later */
+		db->frozenxid = db_frozenxid;
+		/* these get set later: */
 		db->entry = NULL;
+		db->age = 0;
 
 		dblist = lappend(dblist, db);
 	}
@@ -369,6 +423,12 @@ autovac_get_database_list(void)
 /*
  * Process a database.
  *
+ * If whole_db is true, the database is processed as a whole, and the
+ * dbentry parameter is ignored.  If it's false, dbentry must be a valid
+ * pointer to the database entry in the stats databases' hash table, and
+ * it will be used to determine whether vacuum or analyze is needed on a
+ * per-table basis.
+ *
  * Note that test_rel_for_autovac generates two separate lists, one for
  * vacuum and other for analyze.  This is to facilitate processing all
  * analyzes first, and then all vacuums.
@@ -377,7 +437,7 @@ autovac_get_database_list(void)
  * order not to ignore shutdown commands for too long.
  */
 static void
-do_autovacuum(PgStat_StatDBEntry *dbentry)
+do_autovacuum(bool whole_db, PgStat_StatDBEntry *dbentry)
 {
 	Relation		classRel,
 					avRel;
@@ -387,6 +447,8 @@ do_autovacuum(PgStat_StatDBEntry *dbentry)
 				   *analyze_tables = NIL;
 	MemoryContext	AutovacMemCxt;
 
+	Assert(whole_db || PointerIsValid(dbentry));
+
 	/* Memory context where cross-transaction state is stored */
 	AutovacMemCxt = AllocSetContextCreate(TopMemoryContext,
 										  "Autovacuum context",
@@ -405,81 +467,94 @@ do_autovacuum(PgStat_StatDBEntry *dbentry)
 	 */
 	MemoryContextSwitchTo(AutovacMemCxt);
 
-	/*
-	 * If this database is old enough to need a whole-database VACUUM,
-	 * don't bother checking each table.  If that happens, this function
-	 * will issue the VACUUM command and won't return.
-	 */
-	autovac_check_wraparound();
+	if (whole_db)
+	{
+		elog(DEBUG2, "autovacuum: VACUUM ANALYZE whole database");
+		autovacuum_do_vac_analyze(NIL, true);
+	}
+	else
+	{
+		/* the hash entry where pgstat stores shared relations */
+		PgStat_StatDBEntry *shared = pgstat_fetch_stat_dbentry(InvalidOid);
 
-	CHECK_FOR_INTERRUPTS();
+		classRel = heap_open(RelationRelationId, AccessShareLock);
+		avRel = heap_open(AutovacuumRelationId, AccessShareLock);
 
-	classRel = heap_open(RelationRelationId, AccessShareLock);
-	avRel = heap_open(AutovacuumRelationId, AccessShareLock);
+		relScan = heap_beginscan(classRel, SnapshotNow, 0, NULL);
 
-	relScan = heap_beginscan(classRel, SnapshotNow, 0, NULL);
+		/* Scan pg_class looking for tables to vacuum */
+		while ((tuple = heap_getnext(relScan, ForwardScanDirection)) != NULL)
+		{
+			Form_pg_class classForm = (Form_pg_class) GETSTRUCT(tuple);
+			Form_pg_autovacuum avForm = NULL;
+			PgStat_StatTabEntry *tabentry;
+			SysScanDesc	avScan;
+			HeapTuple	avTup;
+			ScanKeyData	entry[1];
+			Oid			relid;
 
-	/* Scan pg_class looking for tables to vacuum */
-	while ((tuple = heap_getnext(relScan, ForwardScanDirection)) != NULL)
-	{
-		Form_pg_class classForm = (Form_pg_class) GETSTRUCT(tuple);
-		Form_pg_autovacuum avForm = NULL;
-		PgStat_StatTabEntry *tabentry;
-		SysScanDesc	avScan;
-		HeapTuple	avTup;
-		ScanKeyData	entry[1];
-		Oid			relid;
-
-		/* Skip non-table entries. */
-		/* XXX possibly allow RELKIND_TOASTVALUE entries here too? */
-		if (classForm->relkind != RELKIND_RELATION)
-			continue;
-		
-		relid = HeapTupleGetOid(tuple);
+			/* Skip non-table entries. */
+			/* XXX possibly allow RELKIND_TOASTVALUE entries here too? */
+			if (classForm->relkind != RELKIND_RELATION)
+				continue;
 
-		/* See if we have a pg_autovacuum entry for this relation. */
-		ScanKeyInit(&entry[0],
-					Anum_pg_autovacuum_vacrelid,
-					BTEqualStrategyNumber, F_OIDEQ,
-					ObjectIdGetDatum(relid));
+			/*
+			 * Skip temp tables (i.e. those in temp namespaces).  We cannot
+			 * safely process other backends' temp tables.
+			 */
+			if (isTempNamespace(classForm->relnamespace))
+				continue;
 
-		avScan = systable_beginscan(avRel, AutovacuumRelidIndexId, true,
- 									SnapshotNow, 1, entry);
+			relid = HeapTupleGetOid(tuple);
 
-		avTup = systable_getnext(avScan);
+			/* See if we have a pg_autovacuum entry for this relation. */
+			ScanKeyInit(&entry[0],
+						Anum_pg_autovacuum_vacrelid,
+						BTEqualStrategyNumber, F_OIDEQ,
+						ObjectIdGetDatum(relid));
 
-		if (HeapTupleIsValid(avTup))
-			avForm = (Form_pg_autovacuum) GETSTRUCT(avTup);
+			avScan = systable_beginscan(avRel, AutovacuumRelidIndexId, true,
+										SnapshotNow, 1, entry);
 
-		tabentry = hash_search(dbentry->tables, &relid,
-							   HASH_FIND, NULL);
+			avTup = systable_getnext(avScan);
 
-		test_rel_for_autovac(relid, tabentry, classForm, avForm,
-							 &vacuum_tables, &analyze_tables);
+			if (HeapTupleIsValid(avTup))
+				avForm = (Form_pg_autovacuum) GETSTRUCT(avTup);
 
-		systable_endscan(avScan);
-	}
+			if (classForm->relisshared && PointerIsValid(shared))
+				tabentry = hash_search(shared->tables, &relid,
+									   HASH_FIND, NULL);
+			else
+				tabentry = hash_search(dbentry->tables, &relid,
+									   HASH_FIND, NULL);
 
-	heap_endscan(relScan);
-	heap_close(avRel, AccessShareLock);
-	heap_close(classRel, AccessShareLock);
+			test_rel_for_autovac(relid, tabentry, classForm, avForm,
+								 &vacuum_tables, &analyze_tables);
 
-	CHECK_FOR_INTERRUPTS();
+			systable_endscan(avScan);
+		}
 
-	/*
-	 * Perform operations on collected tables.
-	 */  
+		heap_endscan(relScan);
+		heap_close(avRel, AccessShareLock);
+		heap_close(classRel, AccessShareLock);
+
+		CHECK_FOR_INTERRUPTS();
+
+		/*
+		 * Perform operations on collected tables.
+		 */
 
-	if (analyze_tables)
-		autovacuum_do_vac_analyze(analyze_tables, false);
+		if (analyze_tables)
+			autovacuum_do_vac_analyze(analyze_tables, false);
 
-	CHECK_FOR_INTERRUPTS();
+		CHECK_FOR_INTERRUPTS();
 
-	/* get back to proper context */
-	MemoryContextSwitchTo(AutovacMemCxt);
+		/* get back to proper context */
+		MemoryContextSwitchTo(AutovacMemCxt);
 
-	if (vacuum_tables)
-		autovacuum_do_vac_analyze(vacuum_tables, true);
+		if (vacuum_tables)
+			autovacuum_do_vac_analyze(vacuum_tables, true);
+	}
 
 	/* Finally close out the last transaction. */
 	CommitTransactionCommand();
@@ -503,7 +578,9 @@ do_autovacuum(PgStat_StatDBEntry *dbentry)
  * analyze.  This is asymmetric to the VACUUM case.
  *
  * A table whose pg_autovacuum.enabled value is false, is automatically
- * skipped.  Thus autovacuum can be disabled for specific tables.
+ * skipped.  Thus autovacuum can be disabled for specific tables.  Also,
+ * when the stats collector does not have data about a table, it will be
+ * skipped.
  *
  * A table whose vac_base_thresh value is <0 takes the base value from the
  * autovacuum_vacuum_threshold GUC variable.  Similarly, a vac_scale_factor
@@ -534,25 +611,18 @@ test_rel_for_autovac(Oid relid, PgStat_StatTabEntry *tabentry,
 	if (avForm && !avForm->enabled)
 		return;
 
-	rel = RelationIdGetRelation(relid);
-	/* The table was recently dropped? */
-	if (rel == NULL)
+	/*
+	 * Skip a table not found in stat hash.  If it's not acted upon,
+	 * there's no need to vacuum it.  (Note that database-level check
+	 * will take care of Xid wraparound.)
+	 */
+	if (!PointerIsValid(tabentry))
 		return;
 
-	/* Not found in stat hash? */
-	if (tabentry == NULL)
-	{
-		/*
-		 * Analyze this table.  It will emit a stat message for the
-		 * collector that will initialize the entry for the next time
-		 * around, so we won't have to guess again.
-		 */
-		elog(DEBUG2, "table %s not known to stat system, will ANALYZE",
-			 RelationGetRelationName(rel));
-		*analyze_tables = lappend_oid(*analyze_tables, relid);
-		RelationClose(rel);
+	rel = RelationIdGetRelation(relid);
+	/* The table was recently dropped? */
+	if (!PointerIsValid(rel))
 		return;
-	}
 
 	reltuples = rel->rd_rel->reltuples;
 	vactuples = tabentry->n_dead_tuples;
@@ -607,9 +677,13 @@ test_rel_for_autovac(Oid relid, PgStat_StatTabEntry *tabentry,
 	}
 	else if (anltuples > anlthresh)
 	{
-		elog(DEBUG2, "will ANALYZE %s",
-			 RelationGetRelationName(rel));
-		*analyze_tables = lappend_oid(*analyze_tables, relid);
+		/* ANALYZE refuses to work with pg_statistics */
+		if (relid != StatisticRelationId)
+		{
+			elog(DEBUG2, "will ANALYZE %s",
+					RelationGetRelationName(rel));
+			*analyze_tables = lappend_oid(*analyze_tables, relid);
+		}
 	}
 
 	RelationClose(rel);
@@ -645,61 +719,6 @@ autovacuum_do_vac_analyze(List *relids, bool dovacuum)
 	vacuum(vacstmt, relids);
 }
 
-/*
- * autovac_check_wraparound
- *		Check database Xid wraparound
- * 
- * Check pg_database to see if the last database-wide VACUUM was too long ago,
- * and issue one now if so.  If this comes to pass, we do not return, as there
- * is no point in checking individual tables -- they will all get vacuumed
- * anyway.
- */
-static void
-autovac_check_wraparound(void)
-{
-	Relation	relation;
-	ScanKeyData	entry[1];
-	HeapScanDesc scan;
-	HeapTuple	tuple;
-	Form_pg_database dbform;
-	int32		age;
-	bool		whole_db;
-
-	relation = heap_open(DatabaseRelationId, AccessShareLock);
-
-	/* Must use a heap scan, since there's no syscache for pg_database */
-	ScanKeyInit(&entry[0],
-				ObjectIdAttributeNumber,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(MyDatabaseId));
-
-	scan = heap_beginscan(relation, SnapshotNow, 1, entry);
-
-	tuple = heap_getnext(scan, ForwardScanDirection);
-
-	if (!HeapTupleIsValid(tuple))
-		elog(ERROR, "could not find tuple for database %u", MyDatabaseId);
-
-	dbform = (Form_pg_database) GETSTRUCT(tuple);
-
-	/*
-	 * We decide to vacuum at the same point where vacuum.c's
-	 * vac_truncate_clog() would decide to start giving warnings.
-	 */
-	age = (int32) (GetTopTransactionId() - dbform->datfrozenxid);
-	whole_db = (age > (int32) ((MaxTransactionId >> 3) * 3));
-
-	heap_endscan(scan);
-	heap_close(relation, AccessShareLock);
-	
-	if (whole_db)
-	{
-		elog(LOG, "autovacuum: VACUUM ANALYZE whole database");
-		autovacuum_do_vac_analyze(NIL, true);
-		proc_exit(0);
-	}
-}
-
 /*
  * AutoVacuumingActive
  * 		Check GUC vars and report whether the autovacuum process should be

src/backend/postmaster/pgstat.c

@@ -13,7 +13,7 @@
  *
  *	Copyright (c) 2001-2005, PostgreSQL Global Development Group
  *
- *	$PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.101 2005/07/24 00:33:28 tgl Exp $
+ *	$PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.102 2005/07/29 19:30:04 tgl Exp $
  * ----------
  */
 #include "postgres.h"
@@ -119,12 +119,23 @@ static long pgStatNumMessages = 0;
 
 static bool pgStatRunningInCollector = FALSE;
 
-static int	pgStatTabstatAlloc = 0;
-static int	pgStatTabstatUsed = 0;
-static PgStat_MsgTabstat **pgStatTabstatMessages = NULL;
+/*
+ * Place where backends store per-table info to be sent to the collector.
+ * We store shared relations separately from non-shared ones, to be able to
+ * send them in separate messages.
+ */
+typedef struct TabStatArray
+{
+	int		tsa_alloc;					/* num allocated */
+	int		tsa_used;					/* num actually used */
+	PgStat_MsgTabstat **tsa_messages;	/* the array itself */
+} TabStatArray;
 
 #define TABSTAT_QUANTUM		4	/* we alloc this many at a time */
 
+static TabStatArray RegularTabStat = { 0, 0, NULL };
+static TabStatArray SharedTabStat = { 0, 0, NULL }; 
+
 static int	pgStatXactCommit = 0;
 static int	pgStatXactRollback = 0;
 
@@ -158,7 +169,7 @@ static void pgstat_exit(SIGNAL_ARGS);
 static void pgstat_die(SIGNAL_ARGS);
 static void pgstat_beshutdown_hook(int code, Datum arg);
 
-static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid);
+static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid, bool create);
 static int	pgstat_add_backend(PgStat_MsgHdr *msg);
 static void pgstat_sub_backend(int procpid);
 static void pgstat_drop_database(Oid databaseid);
@@ -614,6 +625,7 @@ pgstat_beterm(int pid)
 	if (pgStatSock < 0)
 		return;
 
+	/* can't use pgstat_setheader() because it's not called in a backend */
 	MemSet(&(msg.m_hdr), 0, sizeof(msg.m_hdr));
 	msg.m_hdr.m_type = PGSTAT_MTYPE_BETERM;
 	msg.m_hdr.m_procpid = pid;
@@ -684,7 +696,8 @@ pgstat_bestart(void)
  * ---------
  */
 void
-pgstat_report_vacuum(Oid tableoid, bool analyze, PgStat_Counter tuples)
+pgstat_report_vacuum(Oid tableoid, bool shared,
+					 bool analyze, PgStat_Counter tuples)
 {
 	PgStat_MsgVacuum msg;
 
@@ -692,7 +705,7 @@ pgstat_report_vacuum(Oid tableoid, bool analyze, PgStat_Counter tuples)
 		return;
 
 	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_VACUUM);
-	msg.m_databaseid = MyDatabaseId;
+	msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
 	msg.m_tableoid = tableoid;
 	msg.m_analyze = analyze;
 	msg.m_tuples = tuples;
@@ -706,7 +719,7 @@ pgstat_report_vacuum(Oid tableoid, bool analyze, PgStat_Counter tuples)
  * --------
  */
 void
-pgstat_report_analyze(Oid tableoid, PgStat_Counter livetuples,
+pgstat_report_analyze(Oid tableoid, bool shared, PgStat_Counter livetuples,
 					  PgStat_Counter deadtuples)
 {
 	PgStat_MsgAnalyze msg;
@@ -715,7 +728,7 @@ pgstat_report_analyze(Oid tableoid, PgStat_Counter livetuples,
 		return;
 
 	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_ANALYZE);
-	msg.m_databaseid = MyDatabaseId;
+	msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
 	msg.m_tableoid = tableoid;
 	msg.m_live_tuples = livetuples;
 	msg.m_dead_tuples = deadtuples;
@@ -784,7 +797,8 @@ pgstat_report_tabstat(void)
 		  pgstat_collect_blocklevel))
 	{
 		/* Not reporting stats, so just flush whatever we have */
-		pgStatTabstatUsed = 0;
+		RegularTabStat.tsa_used = 0;
+		SharedTabStat.tsa_used = 0;
 		return;
 	}
 
@@ -792,9 +806,9 @@ pgstat_report_tabstat(void)
 	 * For each message buffer used during the last query set the header
 	 * fields and send it out.
 	 */
-	for (i = 0; i < pgStatTabstatUsed; i++)
+	for (i = 0; i < RegularTabStat.tsa_used; i++)
 	{
-		PgStat_MsgTabstat *tsmsg = pgStatTabstatMessages[i];
+		PgStat_MsgTabstat *tsmsg = RegularTabStat.tsa_messages[i];
 		int			n;
 		int			len;
 
@@ -811,8 +825,28 @@ pgstat_report_tabstat(void)
 		tsmsg->m_databaseid = MyDatabaseId;
 		pgstat_send(tsmsg, len);
 	}
+	RegularTabStat.tsa_used = 0;
+
+	/* Ditto, for shared relations */
+	for (i = 0; i < SharedTabStat.tsa_used; i++)
+	{
+		PgStat_MsgTabstat *tsmsg = SharedTabStat.tsa_messages[i];
+		int			n;
+		int			len;
+
+		n = tsmsg->m_nentries;
+		len = offsetof(PgStat_MsgTabstat, m_entry[0]) +
+			n * sizeof(PgStat_TableEntry);
 
-	pgStatTabstatUsed = 0;
+		/* We don't report transaction commit/abort here */
+		tsmsg->m_xact_commit = 0;
+		tsmsg->m_xact_rollback = 0;
+
+		pgstat_setheader(&tsmsg->m_hdr, PGSTAT_MTYPE_TABSTAT);
+		tsmsg->m_databaseid = InvalidOid;
+		pgstat_send(tsmsg, len);
+	}
+	SharedTabStat.tsa_used = 0;
 }
 
 
@@ -850,14 +884,13 @@ pgstat_vacuum_tabstat(void)
 	backend_read_statsfile();
 
 	/*
-	 * Lookup our own database entry
+	 * Lookup our own database entry; if not found, nothing to do.
 	 */
 	dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
 												 (void *) &MyDatabaseId,
 												 HASH_FIND, NULL);
 	if (dbentry == NULL)
 		return -1;
-
 	if (dbentry->tables == NULL)
 		return 0;
 
@@ -867,7 +900,7 @@ pgstat_vacuum_tabstat(void)
 	msg.m_nentries = 0;
 
 	/*
-	 * Check for all tables if they still exist.
+	 * Check for all tables listed in stats hashtable if they still exist.
 	 */
 	hash_seq_init(&hstat, dbentry->tables);
 	while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&hstat)) != NULL)
@@ -892,7 +925,7 @@ pgstat_vacuum_tabstat(void)
 		nobjects++;
 
 		/*
-		 * If the message is full, send it out and reinitialize ot zero
+		 * If the message is full, send it out and reinitialize to zero
 		 */
 		if (msg.m_nentries >= PGSTAT_NUM_TABPURGE)
 		{
@@ -900,6 +933,7 @@ pgstat_vacuum_tabstat(void)
 				+msg.m_nentries * sizeof(Oid);
 
 			pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
+			msg.m_databaseid = MyDatabaseId;
 			pgstat_send(&msg, len);
 
 			msg.m_nentries = 0;
@@ -961,8 +995,8 @@ pgstat_vacuum_tabstat(void)
 
 		if (dbid != InvalidOid)
 		{
-			nobjects++;
 			pgstat_drop_database(dbid);
+			nobjects++;
 		}
 	}
 
@@ -1045,37 +1079,41 @@ pgstat_ping(void)
 }
 
 /*
- * Create or enlarge the pgStatTabstatMessages array
+ * Enlarge a TabStatArray
  */
 static void
-more_tabstat_space(void)
+more_tabstat_space(TabStatArray *tsarr)
 {
 	PgStat_MsgTabstat *newMessages;
 	PgStat_MsgTabstat **msgArray;
-	int			newAlloc = pgStatTabstatAlloc + TABSTAT_QUANTUM;
+	int			newAlloc;
 	int			i;
 
+	AssertArg(PointerIsValid(tsarr));
+
+	newAlloc = tsarr->tsa_alloc + TABSTAT_QUANTUM;
+
 	/* Create (another) quantum of message buffers */
 	newMessages = (PgStat_MsgTabstat *)
 		MemoryContextAllocZero(TopMemoryContext,
 							   sizeof(PgStat_MsgTabstat) * TABSTAT_QUANTUM);
 
 	/* Create or enlarge the pointer array */
-	if (pgStatTabstatMessages == NULL)
+	if (tsarr->tsa_messages == NULL)
 		msgArray = (PgStat_MsgTabstat **)
 			MemoryContextAlloc(TopMemoryContext,
 							   sizeof(PgStat_MsgTabstat *) * newAlloc);
 	else
 		msgArray = (PgStat_MsgTabstat **)
-			repalloc(pgStatTabstatMessages,
+			repalloc(tsarr->tsa_messages,
 					 sizeof(PgStat_MsgTabstat *) * newAlloc);
 
 	for (i = 0; i < TABSTAT_QUANTUM; i++)
-		msgArray[pgStatTabstatAlloc + i] = newMessages++;
-	pgStatTabstatMessages = msgArray;
-	pgStatTabstatAlloc = newAlloc;
+		msgArray[tsarr->tsa_alloc + i] = newMessages++;
+	tsarr->tsa_messages = msgArray;
+	tsarr->tsa_alloc = newAlloc;
 
-	Assert(pgStatTabstatUsed < pgStatTabstatAlloc);
+	Assert(tsarr->tsa_used < tsarr->tsa_alloc);
 }
 
 /* ----------
@@ -1092,6 +1130,7 @@ pgstat_initstats(PgStat_Info *stats, Relation rel)
 {
 	Oid			rel_id = rel->rd_id;
 	PgStat_TableEntry *useent;
+	TabStatArray	*tsarr;
 	PgStat_MsgTabstat *tsmsg;
 	int			mb;
 	int			i;
@@ -1112,12 +1151,14 @@ pgstat_initstats(PgStat_Info *stats, Relation rel)
 		return;
 	}
 
+	tsarr = rel->rd_rel->relisshared ? &SharedTabStat : &RegularTabStat;
+
 	/*
 	 * Search the already-used message slots for this relation.
 	 */
-	for (mb = 0; mb < pgStatTabstatUsed; mb++)
+	for (mb = 0; mb < tsarr->tsa_used; mb++)
 	{
-		tsmsg = pgStatTabstatMessages[mb];
+		tsmsg = tsarr->tsa_messages[mb];
 
 		for (i = tsmsg->m_nentries; --i >= 0;)
 		{
@@ -1146,14 +1187,14 @@ pgstat_initstats(PgStat_Info *stats, Relation rel)
 	/*
 	 * If we ran out of message buffers, we just allocate more.
 	 */
-	if (pgStatTabstatUsed >= pgStatTabstatAlloc)
-		more_tabstat_space();
+	if (tsarr->tsa_used >= tsarr->tsa_alloc)
+		more_tabstat_space(tsarr);
 
 	/*
 	 * Use the first entry of the next message buffer.
 	 */
-	mb = pgStatTabstatUsed++;
-	tsmsg = pgStatTabstatMessages[mb];
+	mb = tsarr->tsa_used++;
+	tsmsg = tsarr->tsa_messages[mb];
 	tsmsg->m_nentries = 1;
 	useent = &tsmsg->m_entry[0];
 	MemSet(useent, 0, sizeof(PgStat_TableEntry));
@@ -1183,13 +1224,13 @@ pgstat_count_xact_commit(void)
 	 * message buffer used without slots, causing the next report to tell
 	 * new xact-counters.
 	 */
-	if (pgStatTabstatAlloc == 0)
-		more_tabstat_space();
+	if (RegularTabStat.tsa_alloc == 0)
+		more_tabstat_space(&RegularTabStat);
 
-	if (pgStatTabstatUsed == 0)
+	if (RegularTabStat.tsa_used == 0)
 	{
-		pgStatTabstatUsed++;
-		pgStatTabstatMessages[0]->m_nentries = 0;
+		RegularTabStat.tsa_used++;
+		RegularTabStat.tsa_messages[0]->m_nentries = 0;
 	}
 }
 
@@ -1215,13 +1256,13 @@ pgstat_count_xact_rollback(void)
 	 * message buffer used without slots, causing the next report to tell
 	 * new xact-counters.
 	 */
-	if (pgStatTabstatAlloc == 0)
-		more_tabstat_space();
+	if (RegularTabStat.tsa_alloc == 0)
+		more_tabstat_space(&RegularTabStat);
 
-	if (pgStatTabstatUsed == 0)
+	if (RegularTabStat.tsa_used == 0)
 	{
-		pgStatTabstatUsed++;
-		pgStatTabstatMessages[0]->m_nentries = 0;
+		RegularTabStat.tsa_used++;
+		RegularTabStat.tsa_messages[0]->m_nentries = 0;
 	}
 }
 
@@ -1265,6 +1306,7 @@ pgstat_fetch_stat_dbentry(Oid dbid)
 PgStat_StatTabEntry *
 pgstat_fetch_stat_tabentry(Oid relid)
 {
+	Oid			dbid;
 	PgStat_StatDBEntry *dbentry;
 	PgStat_StatTabEntry *tabentry;
 
@@ -1275,26 +1317,38 @@ pgstat_fetch_stat_tabentry(Oid relid)
 	backend_read_statsfile();
 
 	/*
-	 * Lookup our database.
+	 * Lookup our database, then look in its table hash table.
 	 */
+	dbid = MyDatabaseId;
 	dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
-												 (void *) &MyDatabaseId,
+												 (void *) &dbid,
 												 HASH_FIND, NULL);
-	if (dbentry == NULL)
-		return NULL;
+	if (dbentry != NULL && dbentry->tables != NULL)
+	{
+		tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
+													   (void *) &relid,
+													   HASH_FIND, NULL);
+		if (tabentry)
+			return tabentry;
+	}
 
 	/*
-	 * Now inside the DB's table hash table lookup the requested one.
+	 * If we didn't find it, maybe it's a shared table.
 	 */
-	if (dbentry->tables == NULL)
-		return NULL;
-	tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
-												   (void *) &relid,
-												   HASH_FIND, NULL);
-	if (tabentry == NULL)
-		return NULL;
+	dbid = InvalidOid;
+	dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
+												 (void *) &dbid,
+												 HASH_FIND, NULL);
+	if (dbentry != NULL && dbentry->tables != NULL)
+	{
+		tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
+													   (void *) &relid,
+													   HASH_FIND, NULL);
+		if (tabentry)
+			return tabentry;
+	}
 
-	return tabentry;
+	return NULL;
 }
 
 
@@ -2107,18 +2161,23 @@ pgstat_add_backend(PgStat_MsgHdr *msg)
 
 /*
  * Lookup the hash table entry for the specified database. If no hash
- * table entry exists, initialize it.
+ * table entry exists, initialize it, if the create parameter is true.
+ * Else, return NULL.
  */
 static PgStat_StatDBEntry *
-pgstat_get_db_entry(Oid databaseid)
+pgstat_get_db_entry(Oid databaseid, bool create)
 {
 	PgStat_StatDBEntry *result;
 	bool found;
+	HASHACTION action = (create ? HASH_ENTER : HASH_FIND);
 
 	/* Lookup or create the hash table entry for this database */
 	result = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
 												&databaseid,
-												HASH_ENTER, &found);
+												action, &found);
+
+	if (!create && !found)
+		return NULL;
 
 	/* If not found, initialize the new one. */
 	if (!found)
@@ -2387,7 +2446,7 @@ pgstat_write_statsfile(void)
  * pgstat_read_statsfile() -
  *
  *	Reads in an existing statistics collector and initializes the
- *	databases hash table (who's entries point to the tables hash tables)
+ *	databases' hash table (whose entries point to the tables' hash tables)
  *	and the current backend table.
  * ----------
  */
@@ -2507,10 +2566,15 @@ pgstat_read_statsfile(HTAB **dbhash, Oid onlydb,
 				dbentry->n_backends = 0;
 
 				/*
-				 * Don't collect tables if not the requested DB
+				 * Don't collect tables if not the requested DB (or the
+				 * shared-table info)
 				 */
-				if (onlydb != InvalidOid && onlydb != dbbuf.databaseid)
+				if (onlydb != InvalidOid)
+				{
+					if (dbbuf.databaseid != onlydb &&
+						dbbuf.databaseid != InvalidOid)
 					break;
+				}
 
 				memset(&hash_ctl, 0, sizeof(hash_ctl));
 				hash_ctl.keysize = sizeof(Oid);
@@ -2588,12 +2652,12 @@ pgstat_read_statsfile(HTAB **dbhash, Oid onlydb,
 				 * backend table.
 				 */
 				if (use_mcxt == NULL)
-					*betab = (PgStat_StatBeEntry *) palloc(
-							   sizeof(PgStat_StatBeEntry) * maxbackends);
+					*betab = (PgStat_StatBeEntry *)
+						palloc(sizeof(PgStat_StatBeEntry) * maxbackends);
 				else
-					*betab = (PgStat_StatBeEntry *) MemoryContextAlloc(
-																use_mcxt,
-							   sizeof(PgStat_StatBeEntry) * maxbackends);
+					*betab = (PgStat_StatBeEntry *)
+						MemoryContextAlloc(use_mcxt,
+										   sizeof(PgStat_StatBeEntry) * maxbackends);
 				break;
 
 				/*
@@ -2738,14 +2802,16 @@ pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len)
 	PgStat_StatDBEntry *dbentry;
 
 	/*
-	 * Lookup the database in the hashtable.
-	 *
-	 * XXX this creates the entry if it doesn't exist.  Is this a problem?  (We
-	 * could leak an entry if we send an autovac message and the database is
-	 * later destroyed, _and_ the messages are rearranged.  Doesn't seem very
-	 * likely though.)  Not sure what to do about it.
+	 * Lookup the database in the hashtable.  Don't create the entry if it
+	 * doesn't exist, because autovacuum may be processing a template
+	 * database.  If this isn't the case, the database is most likely to
+	 * have an entry already.  (If it doesn't, not much harm is done
+	 * anyway -- it'll get created as soon as somebody actually uses
+	 * the database.)
 	 */
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
+	if (dbentry == NULL)
+		return;
 
 	/*
 	 * Store the last autovacuum time in the database entry.
@@ -2765,8 +2831,19 @@ pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
 	PgStat_StatDBEntry *dbentry;
 	PgStat_StatTabEntry *tabentry;
 	bool		found;
+	bool		create;
+
+	/*
+	 * If we don't know about the database, ignore the message, because it
+	 * may be autovacuum processing a template database.  But if the message
+	 * is for database InvalidOid, don't ignore it, because we are getting
+	 * a message from vacuuming a shared relation.
+	 */
+	create = (msg->m_databaseid == InvalidOid);
 
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, create);
+	if (dbentry == NULL)
+		return;
 
 	tabentry = hash_search(dbentry->tables, &(msg->m_tableoid),
 						   HASH_ENTER, &found);
@@ -2819,7 +2896,12 @@ pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len)
 	PgStat_StatTabEntry *tabentry;
 	bool		found;
 
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	/*
+	 * Note that we do create the database entry here, as opposed to what
+	 * we do on AutovacStart and Vacuum messages.  This is because
+	 * autovacuum never executes ANALYZE on template databases.
+	 */
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
 
 	tabentry = hash_search(dbentry->tables, &(msg->m_tableoid),
 						   HASH_ENTER, &found);
@@ -2902,7 +2984,7 @@ pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
 	if (pgstat_add_backend(&msg->m_hdr) < 0)
 		return;
 
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
 
 	/*
 	 * If the database is marked for destroy, this is a delayed UDP packet
@@ -2994,7 +3076,13 @@ pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len)
 	if (pgstat_add_backend(&msg->m_hdr) < 0)
 		return;
 
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
+
+	/*
+	 * No need to purge if we don't even know the database.
+	 */
+	if (!dbentry || !dbentry->tables)
+		return;
 
 	/*
 	 * If the database is marked for destroy, this is a delayed UDP packet
@@ -3037,12 +3125,13 @@ pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len)
 	/*
 	 * Lookup the database in the hashtable.
 	 */
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
 
 	/*
 	 * Mark the database for destruction.
 	 */
-	dbentry->destroy = PGSTAT_DESTROY_COUNT;
+	if (dbentry)
+		dbentry->destroy = PGSTAT_DESTROY_COUNT;
 }
 
 
@@ -3065,9 +3154,12 @@ pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len)
 		return;
 
 	/*
-	 * Lookup the database in the hashtable.
+	 * Lookup the database in the hashtable.  Nothing to do if not there.
 	 */
-	dbentry = pgstat_get_db_entry(msg->m_databaseid);
+	dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
+
+	if (!dbentry)
+		return;
 
 	/*
 	 * We simply throw away all the database's table entries by

src/backend/postmaster/postmaster.c

@@ -37,7 +37,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/postmaster/postmaster.c,v 1.460 2005/07/21 03:56:11 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/postmaster/postmaster.c,v 1.461 2005/07/29 19:30:04 tgl Exp $
  *
  * NOTES
  *
@@ -1164,13 +1164,13 @@ ServerLoop(void)
 		/*
 		 * Wait for something to happen.
 		 *
-		 * We wait at most one minute, to ensure that the other background
-		 * tasks handled below get done even when no requests are
-		 * arriving.
+		 * We wait at most one minute, or the minimum autovacuum delay, to
+		 * ensure that the other background tasks handled below get done
+		 * even when no requests are arriving.
 		 */
 		memcpy((char *) &rmask, (char *) &readmask, sizeof(fd_set));
 
-		timeout.tv_sec = 60;
+		timeout.tv_sec = Min(60, autovacuum_naptime);
 		timeout.tv_usec = 0;
 
 		PG_SETMASK(&UnBlockSig);
@@ -3273,7 +3273,7 @@ SubPostmasterMain(int argc, char *argv[])
 		/* Close the postmaster's sockets */
 		ClosePostmasterPorts(false);
 
-		/* Attached process to shared data structures */
+		/* Attach process to shared data structures */
 		CreateSharedMemoryAndSemaphores(false, 0);
 
 		AutoVacMain(argc - 2, argv + 2);

src/backend/utils/init/postinit.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.153 2005/07/14 05:13:41 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.154 2005/07/29 19:30:05 tgl Exp $
  *
  *
  *-------------------------------------------------------------------------
@@ -78,6 +78,7 @@ FindMyDatabase(const char *name, Oid *db_id, Oid *db_tablespace)
 	char	   *filename;
 	FILE	   *db_file;
 	char		thisname[NAMEDATALEN];
+	TransactionId frozenxid;
 
 	filename = database_getflatfilename();
 	db_file = AllocateFile(filename, "r");
@@ -86,7 +87,8 @@ FindMyDatabase(const char *name, Oid *db_id, Oid *db_tablespace)
 				(errcode_for_file_access(),
 				 errmsg("could not open file \"%s\": %m", filename)));
 
-	while (read_pg_database_line(db_file, thisname, db_id, db_tablespace))
+	while (read_pg_database_line(db_file, thisname, db_id,
+								 db_tablespace, &frozenxid))
 	{
 		if (strcmp(thisname, name) == 0)
 		{
@@ -170,10 +172,11 @@ ReverifyMyDatabase(const char *name)
 	/*
 	 * Also check that the database is currently allowing connections.
 	 * (We do not enforce this in standalone mode, however, so that there is
-	 * a way to recover from "UPDATE pg_database SET datallowconn = false;")
+	 * a way to recover from "UPDATE pg_database SET datallowconn = false;".
+	 * We do not enforce it for the autovacuum process either.)
 	 */
 	dbform = (Form_pg_database) GETSTRUCT(tup);
-	if (IsUnderPostmaster && !dbform->datallowconn)
+	if (IsUnderPostmaster && !IsAutoVacuumProcess() && !dbform->datallowconn)
 		ereport(FATAL,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 		 errmsg("database \"%s\" is not currently accepting connections",

src/backend/utils/misc/guc.c

@@ -10,7 +10,7 @@
  * Written by Peter Eisentraut <peter_e@gmx.net>.
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/misc/guc.c,v 1.278 2005/07/25 22:12:33 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/misc/guc.c,v 1.279 2005/07/29 19:30:07 tgl Exp $
  *
  *--------------------------------------------------------------------
  */
@@ -1418,7 +1418,7 @@ static struct config_int ConfigureNamesInt[] =
 			NULL
 		},
 		&autovacuum_naptime,
-		60, 0, INT_MAX, NULL, NULL
+		60, 1, INT_MAX, NULL, NULL
 	},
 	{
 		{"autovacuum_vacuum_threshold", PGC_SIGHUP, AUTOVACUUM,

src/include/access/xlog.h

@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/access/xlog.h,v 1.66 2005/07/04 04:51:52 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/access/xlog.h,v 1.67 2005/07/29 19:30:08 tgl Exp $
  */
 #ifndef XLOG_H
 #define XLOG_H
@@ -165,5 +165,6 @@ extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(bool shutdown, bool force);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr GetRedoRecPtr(void);
+extern TransactionId GetRecentNextXid(void);
 
 #endif   /* XLOG_H */

src/include/libpq/hba.h

@@ -4,7 +4,7 @@
  *	  Interface to hba.c
  *
  *
- * $PostgreSQL: pgsql/src/include/libpq/hba.h,v 1.38 2005/06/28 05:09:13 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/libpq/hba.h,v 1.39 2005/07/29 19:30:08 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -36,7 +36,7 @@ extern void load_ident(void);
 extern void load_role(void);
 extern int	hba_getauthmethod(hbaPort *port);
 extern int	authident(hbaPort *port);
-extern bool	read_pg_database_line(FILE *fp, char *dbname,
-								  Oid *dboid, Oid *dbtablespace);
+extern bool	read_pg_database_line(FILE *fp, char *dbname, Oid *dboid,
+								  Oid *dbtablespace, TransactionId *dbfrozenxid);
 
 #endif /* HBA_H */

src/include/pgstat.h

@@ -5,7 +5,7 @@
  *
  *	Copyright (c) 2001-2005, PostgreSQL Global Development Group
  *
- *	$PostgreSQL: pgsql/src/include/pgstat.h,v 1.33 2005/07/14 05:13:43 tgl Exp $
+ *	$PostgreSQL: pgsql/src/include/pgstat.h,v 1.34 2005/07/29 19:30:09 tgl Exp $
  * ----------
  */
 #ifndef PGSTAT_H
@@ -384,10 +384,11 @@ extern void pgstat_ping(void);
 extern void pgstat_report_activity(const char *what);
 extern void pgstat_report_tabstat(void);
 extern void pgstat_report_autovac(void);
-extern void pgstat_report_vacuum(Oid tableoid, bool analyze,
-					PgStat_Counter tuples);
-extern void pgstat_report_analyze(Oid tableoid, PgStat_Counter livetuples,
-					PgStat_Counter deadtuples);
+extern void pgstat_report_vacuum(Oid tableoid, bool shared,
+								 bool analyze, PgStat_Counter tuples);
+extern void pgstat_report_analyze(Oid tableoid, bool shared,
+								  PgStat_Counter livetuples,
+								  PgStat_Counter deadtuples);
 extern int	pgstat_vacuum_tabstat(void);
 
 extern void pgstat_reset_counters(void);