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.

13 files changed, 423 insertions, 270 deletions

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 3f393eb10de..144d609bdb7 100644
--- a/src/backend/access/transam/xlog.c
+++ b/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 $
  *
  *-------------------------------------------------------------------------
  */
@@ -4914,6 +4914,36 @@ GetRedoRecPtr(void)
 }
 
 /*
+ * 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
  */
 void
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 03783f121e2..bd32c8c841e 100644
--- a/src/backend/commands/analyze.c
+++ b/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 */
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 23b0911e8cc..9db91209448 100644
--- a/src/backend/commands/vacuum.c
+++ b/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);
 }
 
 
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index 64cab8bbca4..179d35b4028 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/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);
 }
 
 
diff --git a/src/backend/libpq/hba.c b/src/backend/libpq/hba.c
index 6033cf57f66..717b398f893 100644
--- a/src/backend/libpq/hba.c
+++ b/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");
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index de6df77031b..10eb9b6da88 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/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 */
@@ -268,22 +272,62 @@ AutoVacMain(int argc, char *argv[])
 	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);
@@ -646,61 +720,6 @@ autovacuum_do_vac_analyze(List *relids, bool dovacuum)
 }
 
 /*
- * 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
  * 		running.
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index e2cc3508b83..4bb0fc60e3f 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/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
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index 5bc59b148ad..091fbeed0b6 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/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);
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index d687c59ec6d..b013eca86cf 100644
--- a/src/backend/utils/init/postinit.c
+++ b/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",
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 726a093d0d7..da6aa1a9c36 100644
--- a/src/backend/utils/misc/guc.c
+++ b/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,
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 8bbc6846de6..c16fdeeebd7 100644
--- a/src/include/access/xlog.h
+++ b/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 */
diff --git a/src/include/libpq/hba.h b/src/include/libpq/hba.h
index d170f303a43..568aaf13c3d 100644
--- a/src/include/libpq/hba.h
+++ b/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 */
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 4df12d77d38..f8d5f02ea18 100644
--- a/src/include/pgstat.h
+++ b/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);