parallel_pg_dump_8-part1.diff

diff --git a/src/bin/pg_dump/dumputils.c b/src/bin/pg_dump/dumputils.c
index 8574380..c5d2144 100644
*** a/src/bin/pg_dump/dumputils.c
--- b/src/bin/pg_dump/dumputils.c
***************
*** 16,21 ****
--- 16,22 ----
  
  #include <ctype.h>
  
+ #include "dumpmem.h"
  #include "dumputils.h"
  
  #include "parser/keywords.h"
*************** static struct
*** 38,43 ****
--- 39,45 ----
  }	on_exit_nicely_list[MAX_ON_EXIT_NICELY];
  
  static int	on_exit_nicely_index;
+ void (*on_exit_msg_func)(const char *modulename, const char *fmt, va_list ap) = vwrite_msg;
  
  #define supports_grant_options(version) ((version) >= 70400)
  
*************** static bool parseAclItem(const char *ite
*** 48,53 ****
--- 50,56 ----
  static char *copyAclUserName(PQExpBuffer output, char *input);
  static void AddAcl(PQExpBuffer aclbuf, const char *keyword,
  	   const char *subname);
+ static PQExpBuffer getThreadLocalPQExpBuffer(void);
  
  #ifdef WIN32
  static bool parallel_init_done = false;
*************** init_parallel_dump_utils(void)
*** 69,83 ****
  }
  
  /*
!  *	Quotes input string if it's not a legitimate SQL identifier as-is.
!  *
!  *	Note that the returned string must be used before calling fmtId again,
!  *	since we re-use the same return buffer each time.  Non-reentrant but
!  *	reduces memory leakage. (On Windows the memory leakage will be one buffer
!  *	per thread, which is at least better than one per call).
   */
! const char *
! fmtId(const char *rawid)
  {
  	/*
  	 * The Tls code goes awry if we use a static var, so we provide for both
--- 72,82 ----
  }
  
  /*
!  * Non-reentrant but reduces memory leakage. (On Windows the memory leakage
!  * will be one buffer per thread, which is at least better than one per call).
   */
! static PQExpBuffer
! getThreadLocalPQExpBuffer(void)
  {
  	/*
  	 * The Tls code goes awry if we use a static var, so we provide for both
*************** fmtId(const char *rawid)
*** 86,94 ****
  	static PQExpBuffer s_id_return = NULL;
  	PQExpBuffer id_return;
  
- 	const char *cp;
- 	bool		need_quotes = false;
- 
  #ifdef WIN32
  	if (parallel_init_done)
  		id_return = (PQExpBuffer) TlsGetValue(tls_index);		/* 0 when not set */
--- 85,90 ----
*************** fmtId(const char *rawid)
*** 118,123 ****
--- 114,136 ----
  
  	}
  
+ 	return id_return;
+ }
+ 
+ /*
+  *	Quotes input string if it's not a legitimate SQL identifier as-is.
+  *
+  *	Note that the returned string must be used before calling fmtId again,
+  *	since we re-use the same return buffer each time.
+  */
+ const char *
+ fmtId(const char *rawid)
+ {
+ 	PQExpBuffer id_return = getThreadLocalPQExpBuffer();
+ 
+ 	const char *cp;
+ 	bool		need_quotes = false;
+ 
  	/*
  	 * These checks need to match the identifier production in scan.l. Don't
  	 * use islower() etc.
*************** fmtId(const char *rawid)
*** 185,190 ****
--- 198,232 ----
  	return id_return->data;
  }
  
+ /*
+  * fmtQualifiedId - convert a qualified name to the proper format for
+  * the source database.
+  *
+  * Like fmtId, use the result before calling again.
+  *
+  * Since we call fmtId and it also uses getThreadLocalPQExpBuffer() we cannot
+  * use it until we're finished with calling fmtId().
+  */
+ const char *
+ fmtQualifiedId(int remoteVersion, const char *schema, const char *id)
+ {
+ 	PQExpBuffer id_return;
+ 	PQExpBuffer lcl_pqexp = createPQExpBuffer();
+ 
+ 	/* Suppress schema name if fetching from pre-7.3 DB */
+ 	if (remoteVersion >= 70300 && schema && *schema)
+ 	{
+ 		appendPQExpBuffer(lcl_pqexp, "%s.", fmtId(schema));
+ 	}
+ 	appendPQExpBuffer(lcl_pqexp, "%s", fmtId(id));
+ 
+ 	id_return = getThreadLocalPQExpBuffer();
+ 
+ 	appendPQExpBuffer(id_return, "%s", lcl_pqexp->data);
+ 	destroyPQExpBuffer(lcl_pqexp);
+ 
+ 	return id_return->data;
+ }
  
  /*
   * Convert a string value to an SQL string literal and append it to
*************** exit_horribly(const char *modulename, co
*** 1312,1318 ****
  	va_list		ap;
  
  	va_start(ap, fmt);
! 	vwrite_msg(modulename, fmt, ap);
  	va_end(ap);
  
  	exit_nicely(1);
--- 1354,1360 ----
  	va_list		ap;
  
  	va_start(ap, fmt);
! 	on_exit_msg_func(modulename, fmt, ap);
  	va_end(ap);
  
  	exit_nicely(1);
diff --git a/src/bin/pg_dump/dumputils.h b/src/bin/pg_dump/dumputils.h
index 4ef8cb3..fd92970 100644
*** a/src/bin/pg_dump/dumputils.h
--- b/src/bin/pg_dump/dumputils.h
*************** extern const char *progname;
*** 34,39 ****
--- 34,41 ----
  
  extern void init_parallel_dump_utils(void);
  extern const char *fmtId(const char *identifier);
+ extern const char *fmtQualifiedId(int remoteVersion,
+ 								  const char *schema, const char *id);
  extern void appendStringLiteral(PQExpBuffer buf, const char *str,
  					int encoding, bool std_strings);
  extern void appendStringLiteralConn(PQExpBuffer buf, const char *str,
*************** __attribute__((format(PG_PRINTF_ATTRIBUT
*** 72,77 ****
--- 74,81 ----
  extern void
  exit_horribly(const char *modulename, const char *fmt,...)
  __attribute__((format(PG_PRINTF_ATTRIBUTE, 2, 3), noreturn));
+ extern void (*on_exit_msg_func)(const char *modulename, const char *fmt, va_list ap)
+ 				__attribute__((format(PG_PRINTF_ATTRIBUTE, 2, 0)));
  extern void on_exit_nicely(on_exit_nicely_callback function, void *arg);
  extern void exit_nicely(int code) __attribute__((noreturn));
  
diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h
index 3b49395..b82171c 100644
*** a/src/bin/pg_dump/pg_backup.h
--- b/src/bin/pg_dump/pg_backup.h
*************** struct Archive
*** 81,86 ****
--- 81,88 ----
  	int			minRemoteVersion;		/* allowable range */
  	int			maxRemoteVersion;
  
+ 	int			numWorkers;		/* number of parallel processes */
+ 
  	/* info needed for string escaping */
  	int			encoding;		/* libpq code for client_encoding */
  	bool		std_strings;	/* standard_conforming_strings */
*************** typedef struct _restoreOptions
*** 141,147 ****
  	int			suppressDumpWarnings;	/* Suppress output of WARNING entries
  										 * to stderr */
  	bool		single_txn;
- 	int			number_of_jobs;
  
  	bool	   *idWanted;		/* array showing which dump IDs to emit */
  } RestoreOptions;
--- 143,148 ----
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 5826bac..9f3416f 100644
*** a/src/bin/pg_dump/pg_backup_archiver.c
--- b/src/bin/pg_dump/pg_backup_archiver.c
*************** static bool _tocEntryIsACL(TocEntry *te)
*** 136,142 ****
  static void _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
  static void _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
  static void buildTocEntryArrays(ArchiveHandle *AH);
- static TocEntry *getTocEntryByDumpId(ArchiveHandle *AH, DumpId id);
  static void _moveBefore(ArchiveHandle *AH, TocEntry *pos, TocEntry *te);
  static int	_discoverArchiveFormat(ArchiveHandle *AH);
  
--- 136,141 ----
*************** static void RestoreOutput(ArchiveHandle
*** 149,155 ****
  
  static int restore_toc_entry(ArchiveHandle *AH, TocEntry *te,
  				  RestoreOptions *ropt, bool is_parallel);
! static void restore_toc_entries_parallel(ArchiveHandle *AH);
  static thandle spawn_restore(RestoreArgs *args);
  static thandle reap_child(ParallelSlot *slots, int n_slots, int *work_status);
  static bool work_in_progress(ParallelSlot *slots, int n_slots);
--- 148,156 ----
  
  static int restore_toc_entry(ArchiveHandle *AH, TocEntry *te,
  				  RestoreOptions *ropt, bool is_parallel);
! static void restore_toc_entries_prefork(ArchiveHandle *AH);
! static void restore_toc_entries_parallel(ArchiveHandle *AH, TocEntry *pending_list);
! static void restore_toc_entries_postfork(ArchiveHandle *AH, TocEntry *pending_list);
  static thandle spawn_restore(RestoreArgs *args);
  static thandle reap_child(ParallelSlot *slots, int n_slots, int *work_status);
  static bool work_in_progress(ParallelSlot *slots, int n_slots);
*************** static void reduce_dependencies(ArchiveH
*** 172,179 ****
  					TocEntry *ready_list);
  static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
  static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
- static ArchiveHandle *CloneArchive(ArchiveHandle *AH);
- static void DeCloneArchive(ArchiveHandle *AH);
  
  static void setProcessIdentifier(ParallelStateEntry *pse, ArchiveHandle *AH);
  static void unsetProcessIdentifier(ParallelStateEntry *pse);
--- 173,178 ----
*************** RestoreArchive(Archive *AHX)
*** 286,292 ****
  	/*
  	 * If we're going to do parallel restore, there are some restrictions.
  	 */
! 	parallel_mode = (ropt->number_of_jobs > 1 && ropt->useDB);
  	if (parallel_mode)
  	{
  		/* We haven't got round to making this work for all archive formats */
--- 285,291 ----
  	/*
  	 * If we're going to do parallel restore, there are some restrictions.
  	 */
! 	parallel_mode = (AH->public.numWorkers > 1 && ropt->useDB);
  	if (parallel_mode)
  	{
  		/* We haven't got round to making this work for all archive formats */
*************** RestoreArchive(Archive *AHX)
*** 456,462 ****
  	 * In parallel mode, turn control over to the parallel-restore logic.
  	 */
  	if (parallel_mode)
! 		restore_toc_entries_parallel(AH);
  	else
  	{
  		for (te = AH->toc->next; te != AH->toc; te = te->next)
--- 455,476 ----
  	 * In parallel mode, turn control over to the parallel-restore logic.
  	 */
  	if (parallel_mode)
! 	{
! 		TocEntry pending_list;
! 
! 		par_list_header_init(&pending_list);
! 
! 		/* This runs PRE_DATA items and then disconnects from the database */
! 		restore_toc_entries_prefork(AH);
! 		Assert(AH->connection == NULL);
! 
! 		/* This will actually fork the processes */
! 		restore_toc_entries_parallel(AH, &pending_list);
! 
! 		/* reconnect the master and see if we missed something */
! 		restore_toc_entries_postfork(AH, &pending_list);
! 		Assert(AH->connection != NULL);
! 	}
  	else
  	{
  		for (te = AH->toc->next; te != AH->toc; te = te->next)
*************** buildTocEntryArrays(ArchiveHandle *AH)
*** 1594,1600 ****
  	}
  }
  
! static TocEntry *
  getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)
  {
  	/* build index arrays if we didn't already */
--- 1608,1614 ----
  	}
  }
  
! TocEntry *
  getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)
  {
  	/* build index arrays if we didn't already */
*************** on_exit_close_archive(Archive *AHX)
*** 3420,3461 ****
  	on_exit_nicely(archive_close_connection, &shutdown_info);
  }
  
- /*
-  * Main engine for parallel restore.
-  *
-  * Work is done in three phases.
-  * First we process all SECTION_PRE_DATA tocEntries, in a single connection,
-  * just as for a standard restore.	Second we process the remaining non-ACL
-  * steps in parallel worker children (threads on Windows, processes on Unix),
-  * each of which connects separately to the database.  Finally we process all
-  * the ACL entries in a single connection (that happens back in
-  * RestoreArchive).
-  */
  static void
! restore_toc_entries_parallel(ArchiveHandle *AH)
  {
  	RestoreOptions *ropt = AH->ropt;
- 	int			n_slots = ropt->number_of_jobs;
- 	ParallelSlot *slots;
- 	int			work_status;
- 	int			next_slot;
  	bool		skipped_some;
- 	TocEntry	pending_list;
- 	TocEntry	ready_list;
  	TocEntry   *next_work_item;
- 	thandle		ret_child;
- 	TocEntry   *te;
- 	ParallelState *pstate;
- 	int			i;
- 
- 	ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
  
! 	slots = (ParallelSlot *) pg_calloc(n_slots, sizeof(ParallelSlot));
! 	pstate = (ParallelState *) pg_malloc(sizeof(ParallelState));
! 	pstate->pse = (ParallelStateEntry *) pg_calloc(n_slots, sizeof(ParallelStateEntry));
! 	pstate->numWorkers = ropt->number_of_jobs;
! 	for (i = 0; i < pstate->numWorkers; i++)
! 		unsetProcessIdentifier(&(pstate->pse[i]));
  
  	/* Adjust dependency information */
  	fix_dependencies(AH);
--- 3434,3447 ----
  	on_exit_nicely(archive_close_connection, &shutdown_info);
  }
  
  static void
! restore_toc_entries_prefork(ArchiveHandle *AH)
  {
  	RestoreOptions *ropt = AH->ropt;
  	bool		skipped_some;
  	TocEntry   *next_work_item;
  
! 	ahlog(AH, 2, "entering restore_toc_entries_prefork\n");
  
  	/* Adjust dependency information */
  	fix_dependencies(AH);
*************** restore_toc_entries_parallel(ArchiveHand
*** 3511,3522 ****
  	 */
  	DisconnectDatabase(&AH->public);
  
- 	/*
- 	 * Set the pstate in the shutdown_info. The exit handler uses pstate if
- 	 * set and falls back to AHX otherwise.
- 	 */
- 	shutdown_info.pstate = pstate;
- 
  	/* blow away any transient state from the old connection */
  	if (AH->currUser)
  		free(AH->currUser);
--- 3497,3502 ----
*************** restore_toc_entries_parallel(ArchiveHand
*** 3528,3544 ****
  		free(AH->currTablespace);
  	AH->currTablespace = NULL;
  	AH->currWithOids = -1;
  
  	/*
! 	 * Initialize the lists of pending and ready items.  After this setup, the
! 	 * pending list is everything that needs to be done but is blocked by one
! 	 * or more dependencies, while the ready list contains items that have no
! 	 * remaining dependencies.	Note: we don't yet filter out entries that
! 	 * aren't going to be restored.  They might participate in dependency
! 	 * chains connecting entries that should be restored, so we treat them as
! 	 * live until we actually process them.
  	 */
- 	par_list_header_init(&pending_list);
  	par_list_header_init(&ready_list);
  	skipped_some = false;
  	for (next_work_item = AH->toc->next; next_work_item != AH->toc; next_work_item = next_work_item->next)
--- 3508,3566 ----
  		free(AH->currTablespace);
  	AH->currTablespace = NULL;
  	AH->currWithOids = -1;
+ }
+ 
+ /*
+  * Main engine for parallel restore.
+  *
+  * Work is done in three phases.
+  * First we process all SECTION_PRE_DATA tocEntries, in a single connection,
+  * just as for a standard restore. This is done in restore_toc_entries_prefork().
+  * Second we process the remaining non-ACL steps in parallel worker children
+  * (threads on Windows, processes on Unix), each of which connects separately
+  * to the database.
+  * Finally we process all the ACL entries in a single connection (that happens
+  * back in RestoreArchive).
+  */
+ static void
+ restore_toc_entries_parallel(ArchiveHandle *AH, TocEntry *pending_list)
+ {
+ 	ParallelState *pstate;
+ 	ParallelSlot *slots;
+ 	int			n_slots = AH->public.numWorkers;
+ 	TocEntry   *next_work_item;
+ 	int			next_slot;
+ 	TocEntry	ready_list;
+ 	int			ret_child;
+ 	bool		skipped_some;
+ 	int			work_status;
+ 	int			i;
+ 
+ 	ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
+ 
+ 	slots = (ParallelSlot *) pg_calloc(n_slots, sizeof(ParallelSlot));
+ 	pstate = (ParallelState *) pg_malloc(sizeof(ParallelState));
+ 	pstate->pse = (ParallelStateEntry *) pg_calloc(n_slots, sizeof(ParallelStateEntry));
+ 	pstate->numWorkers = AH->public.numWorkers;
+ 	for (i = 0; i < pstate->numWorkers; i++)
+ 		unsetProcessIdentifier(&(pstate->pse[i]));
  
  	/*
! 	 * Set the pstate in the shutdown_info. The exit handler uses pstate if set
! 	 * and falls back to AHX otherwise.
! 	 */
! 	shutdown_info.pstate = pstate;
! 
! 	/*
! 	 * Initialize the lists of ready items, the list for pending items has
! 	 * already been initialized in the caller.  After this setup, the pending
! 	 * list is everything that needs to be done but is blocked by one or more
! 	 * dependencies, while the ready list contains items that have no remaining
! 	 * dependencies.	Note: we don't yet filter out entries that aren't going
! 	 * to be restored.  They might participate in dependency chains connecting
! 	 * entries that should be restored, so we treat them as live until we
! 	 * actually process them.
  	 */
  	par_list_header_init(&ready_list);
  	skipped_some = false;
  	for (next_work_item = AH->toc->next; next_work_item != AH->toc; next_work_item = next_work_item->next)
*************** restore_toc_entries_parallel(ArchiveHand
*** 3563,3569 ****
  		}
  
  		if (next_work_item->depCount > 0)
! 			par_list_append(&pending_list, next_work_item);
  		else
  			par_list_append(&ready_list, next_work_item);
  	}
--- 3585,3591 ----
  		}
  
  		if (next_work_item->depCount > 0)
! 			par_list_append(pending_list, next_work_item);
  		else
  			par_list_append(&ready_list, next_work_item);
  	}
*************** restore_toc_entries_parallel(ArchiveHand
*** 3647,3652 ****
--- 3669,3683 ----
  	}
  
  	ahlog(AH, 1, "finished main parallel loop\n");
+ }
+ 
+ static void
+ restore_toc_entries_postfork(ArchiveHandle *AH, TocEntry *pending_list)
+ {
+ 	RestoreOptions *ropt = AH->ropt;
+ 	TocEntry   *te;
+ 
+ 	ahlog(AH, 2, "entering restore_toc_entries_postfork\n");
  
  	/*
  	 * Remove the pstate again, so the exit handler will now fall back to
*************** restore_toc_entries_parallel(ArchiveHand
*** 3668,3674 ****
  	 * dependencies, or some other pathological condition. If so, do it in the
  	 * single parent connection.
  	 */
! 	for (te = pending_list.par_next; te != &pending_list; te = te->par_next)
  	{
  		ahlog(AH, 1, "processing missed item %d %s %s\n",
  			  te->dumpId, te->desc, te->tag);
--- 3699,3705 ----
  	 * dependencies, or some other pathological condition. If so, do it in the
  	 * single parent connection.
  	 */
! 	for (te = pending_list->par_next; te != pending_list; te = te->par_next)
  	{
  		ahlog(AH, 1, "processing missed item %d %s %s\n",
  			  te->dumpId, te->desc, te->tag);
*************** inhibit_data_for_failed_table(ArchiveHan
*** 4332,4341 ****
   *
   * Enough of the structure is cloned to ensure that there is no
   * conflict between different threads each with their own clone.
-  *
-  * These could be public, but no need at present.
   */
! static ArchiveHandle *
  CloneArchive(ArchiveHandle *AH)
  {
  	ArchiveHandle *clone;
--- 4363,4370 ----
   *
   * Enough of the structure is cloned to ensure that there is no
   * conflict between different threads each with their own clone.
   */
! ArchiveHandle *
  CloneArchive(ArchiveHandle *AH)
  {
  	ArchiveHandle *clone;
*************** CloneArchive(ArchiveHandle *AH)
*** 4372,4378 ****
   *
   * Note: we assume any clone-local connection was already closed.
   */
! static void
  DeCloneArchive(ArchiveHandle *AH)
  {
  	/* Clear format-specific state */
--- 4401,4407 ----
   *
   * Note: we assume any clone-local connection was already closed.
   */
! void
  DeCloneArchive(ArchiveHandle *AH)
  {
  	/* Clear format-specific state */
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index 8859bd9..2c77adc 100644
*** a/src/bin/pg_dump/pg_backup_archiver.h
--- b/src/bin/pg_dump/pg_backup_archiver.h
*************** extern void ReadHead(ArchiveHandle *AH);
*** 338,345 ****
--- 338,348 ----
  extern void WriteToc(ArchiveHandle *AH);
  extern void ReadToc(ArchiveHandle *AH);
  extern void WriteDataChunks(ArchiveHandle *AH);
+ extern ArchiveHandle *CloneArchive(ArchiveHandle *AH);
+ extern void DeCloneArchive(ArchiveHandle *AH);
  
  extern teReqs TocIDRequired(ArchiveHandle *AH, DumpId id);
+ TocEntry *getTocEntryByDumpId(ArchiveHandle *AH, DumpId id);
  extern bool checkSeek(FILE *fp);
  
  #define appendStringLiteralAHX(buf,str,AH) \
*************** int			ahprintf(ArchiveHandle *AH, const
*** 380,383 ****
--- 383,398 ----
  
  void		ahlog(ArchiveHandle *AH, int level, const char *fmt,...) __attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 4)));
  
+ #ifdef USE_ASSERT_CHECKING
+ #define Assert(condition) \
+ 	if (!(condition)) \
+ 	{ \
+ 		write_msg(NULL, "Failed assertion in %s, line %d\n", \
+ 				  __FILE__, __LINE__); \
+ 		abort();\
+ 	}
+ #else
+ #define Assert(condition)
+ #endif
+ 
  #endif
diff --git a/src/bin/pg_dump/pg_backup_db.c b/src/bin/pg_dump/pg_backup_db.c
index d912aaf..8e1d33b 100644
*** a/src/bin/pg_dump/pg_backup_db.c
--- b/src/bin/pg_dump/pg_backup_db.c
*************** ConnectDatabase(Archive *AHX,
*** 308,319 ****
  	PQsetNoticeProcessor(AH->connection, notice_processor, NULL);
  }
  
  void
  DisconnectDatabase(Archive *AHX)
  {
  	ArchiveHandle *AH = (ArchiveHandle *) AHX;
  
! 	PQfinish(AH->connection);	/* noop if AH->connection is NULL */
  	AH->connection = NULL;
  }
  
--- 308,337 ----
  	PQsetNoticeProcessor(AH->connection, notice_processor, NULL);
  }
  
+ /*
+  * Close the connection to the database and also cancel off the query if we
+  * have one running.
+  */
  void
  DisconnectDatabase(Archive *AHX)
  {
  	ArchiveHandle *AH = (ArchiveHandle *) AHX;
+ 	PGcancel   *cancel;
+ 	char		errbuf[1];
  
! 	if (!AH->connection)
! 		return;
! 
! 	if (PQtransactionStatus(AH->connection) == PQTRANS_ACTIVE)
! 	{
! 		if ((cancel = PQgetCancel(AH->connection)))
! 		{
! 			PQcancel(cancel, errbuf, sizeof(errbuf));
! 			PQfreeCancel(cancel);
! 		}
! 	}
! 
! 	PQfinish(AH->connection);
  	AH->connection = NULL;
  }
  
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 32dcb12..3a611d7 100644
*** a/src/bin/pg_dump/pg_backup_directory.c
--- b/src/bin/pg_dump/pg_backup_directory.c
*************** static void _EndBlob(ArchiveHandle *AH,
*** 82,89 ****
  static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
  static void _LoadBlobs(ArchiveHandle *AH, RestoreOptions *ropt);
  
! static char *prependDirectory(ArchiveHandle *AH, const char *relativeFilename);
! 
  static void createDirectory(const char *dir);
  
  
--- 82,88 ----
  static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
  static void _LoadBlobs(ArchiveHandle *AH, RestoreOptions *ropt);
  
! static char *prependDirectory(ArchiveHandle *AH, char *buf, const char *relativeFilename);
  static void createDirectory(const char *dir);
  
  
*************** InitArchiveFmt_Directory(ArchiveHandle *
*** 153,162 ****
  	}
  	else
  	{							/* Read Mode */
! 		char	   *fname;
  		cfp		   *tocFH;
  
! 		fname = prependDirectory(AH, "toc.dat");
  
  		tocFH = cfopen_read(fname, PG_BINARY_R);
  		if (tocFH == NULL)
--- 152,161 ----
  	}
  	else
  	{							/* Read Mode */
! 		char	   fname[MAXPGPATH];
  		cfp		   *tocFH;
  
! 		prependDirectory(AH, fname, "toc.dat");
  
  		tocFH = cfopen_read(fname, PG_BINARY_R);
  		if (tocFH == NULL)
*************** _StartData(ArchiveHandle *AH, TocEntry *
*** 282,290 ****
  {
  	lclTocEntry *tctx = (lclTocEntry *) te->formatData;
  	lclContext *ctx = (lclContext *) AH->formatData;
! 	char	   *fname;
  
! 	fname = prependDirectory(AH, tctx->filename);
  
  	ctx->dataFH = cfopen_write(fname, PG_BINARY_W, AH->compression);
  	if (ctx->dataFH == NULL)
--- 281,289 ----
  {
  	lclTocEntry *tctx = (lclTocEntry *) te->formatData;
  	lclContext *ctx = (lclContext *) AH->formatData;
! 	char		fname[MAXPGPATH];
  
! 	prependDirectory(AH, fname, tctx->filename);
  
  	ctx->dataFH = cfopen_write(fname, PG_BINARY_W, AH->compression);
  	if (ctx->dataFH == NULL)
*************** _PrintTocData(ArchiveHandle *AH, TocEntr
*** 376,383 ****
  		_LoadBlobs(AH, ropt);
  	else
  	{
! 		char	   *fname = prependDirectory(AH, tctx->filename);
  
  		_PrintFileData(AH, fname, ropt);
  	}
  }
--- 375,383 ----
  		_LoadBlobs(AH, ropt);
  	else
  	{
! 		char		fname[MAXPGPATH];
  
+ 		prependDirectory(AH, fname, tctx->filename);
  		_PrintFileData(AH, fname, ropt);
  	}
  }
*************** _LoadBlobs(ArchiveHandle *AH, RestoreOpt
*** 387,398 ****
  {
  	Oid			oid;
  	lclContext *ctx = (lclContext *) AH->formatData;
! 	char	   *fname;
  	char		line[MAXPGPATH];
  
  	StartRestoreBlobs(AH);
  
! 	fname = prependDirectory(AH, "blobs.toc");
  
  	ctx->blobsTocFH = cfopen_read(fname, PG_BINARY_R);
  
--- 387,398 ----
  {
  	Oid			oid;
  	lclContext *ctx = (lclContext *) AH->formatData;
! 	char		fname[MAXPGPATH];
  	char		line[MAXPGPATH];
  
  	StartRestoreBlobs(AH);
  
! 	prependDirectory(AH, fname, "blobs.toc");
  
  	ctx->blobsTocFH = cfopen_read(fname, PG_BINARY_R);
  
*************** _CloseArchive(ArchiveHandle *AH)
*** 519,525 ****
  	if (AH->mode == archModeWrite)
  	{
  		cfp		   *tocFH;
! 		char	   *fname = prependDirectory(AH, "toc.dat");
  
  		/* The TOC is always created uncompressed */
  		tocFH = cfopen_write(fname, PG_BINARY_W, 0);
--- 519,527 ----
  	if (AH->mode == archModeWrite)
  	{
  		cfp		   *tocFH;
! 		char		fname[MAXPGPATH];
! 
! 		prependDirectory(AH, fname, "toc.dat");
  
  		/* The TOC is always created uncompressed */
  		tocFH = cfopen_write(fname, PG_BINARY_W, 0);
*************** static void
*** 561,569 ****
  _StartBlobs(ArchiveHandle *AH, TocEntry *te)
  {
  	lclContext *ctx = (lclContext *) AH->formatData;
! 	char	   *fname;
  
! 	fname = prependDirectory(AH, "blobs.toc");
  
  	/* The blob TOC file is never compressed */
  	ctx->blobsTocFH = cfopen_write(fname, "ab", 0);
--- 563,571 ----
  _StartBlobs(ArchiveHandle *AH, TocEntry *te)
  {
  	lclContext *ctx = (lclContext *) AH->formatData;
! 	char		fname[MAXPGPATH];
  
! 	prependDirectory(AH, fname, "blobs.toc");
  
  	/* The blob TOC file is never compressed */
  	ctx->blobsTocFH = cfopen_write(fname, "ab", 0);
*************** _EndBlobs(ArchiveHandle *AH, TocEntry *t
*** 629,667 ****
  }
  
  static void
! createDirectory(const char *dir)
  {
  	struct stat st;
  
! 	/* the directory must not exist yet. */
! 	if (stat(dir, &st) == 0)
  	{
  		if (S_ISDIR(st.st_mode))
! 			exit_horribly(modulename,
! 						  "cannot create directory %s, it exists already\n",
! 						  dir);
  		else
  			exit_horribly(modulename,
  						  "cannot create directory %s, a file with this name "
! 						  "exists already\n", dir);
  	}
- 
- 	/*
- 	 * Now we create the directory. Note that for some race condition we could
- 	 * also run into the situation that the directory has been created just
- 	 * between our two calls.
- 	 */
- 	if (mkdir(dir, 0700) < 0)
- 		exit_horribly(modulename, "could not create directory %s: %s",
- 					  dir, strerror(errno));
  }
  
! 
  static char *
! prependDirectory(ArchiveHandle *AH, const char *relativeFilename)
  {
  	lclContext *ctx = (lclContext *) AH->formatData;
- 	static char buf[MAXPGPATH];
  	char	   *dname;
  
  	dname = ctx->directory;
--- 631,707 ----
  }
  
  static void
! createDirectory(const char *directory)
  {
  	struct stat st;
  
! 	/* the directory must either not exist or must be empty. */
! 	if (stat(directory, &st) == 0)
  	{
  		if (S_ISDIR(st.st_mode))
! 		{
! 			DIR	   *dir;
! 			struct dirent *entry;
! 			bool	empty = true;
! 
! 			dir = opendir(directory);
! 			if (!dir)
! 				exit_horribly(modulename,
! 							  "cannot create directory %s, it exists already\n",
! 							  directory);
! 
! 			while ((entry = readdir(dir))) {
! 				if (strcmp(entry->d_name, ".") == 0)
! 					continue;
! 				if (strcmp(entry->d_name, "..") == 0)
! 					continue;
! 				empty = false;
! 				break;
! 			}
! 			closedir(dir);
! 
! 			if (!empty)
! 				exit_horribly(modulename,
! 							  "cannot create directory %s, it exists already "
! 							  "and is not empty\n",
! 							  directory);
! 
! 			/*
! 			 * Down here we know that the directory exists and is empty. This
! 			 * doesn't mean that we can create files in it, but we will soon
! 			 * find out. We could test-create a file and delete it again
! 			 * already now but we need to be prepared for failing later
! 			 * anyway...
! 			 */
! 		}
  		else
  			exit_horribly(modulename,
  						  "cannot create directory %s, a file with this name "
! 						  "exists already\n", directory);
! 	}
! 	else
! 	{
! 		/*
! 		 * Now we create the directory. Note that for some race condition we could
! 		 * also run into the situation that the directory has been created just
! 		 * between our two calls.
! 		 */
! 		if (mkdir(directory, 0700) < 0)
! 			exit_horribly(modulename, "could not create directory %s: %s",
! 						  directory, strerror(errno));
  	}
  }
  
! /*
!  * Gets a relative file name and prepends the output directory, writing the
!  * result to buf. The caller needs to make sure that buf is MAXPGPATH bytes
!  * big. Can't use a static char[MAXPGPATH] inside the function because we run
!  * multithreaded on Windows.
!  */
  static char *
! prependDirectory(ArchiveHandle *AH, char* buf, const char *relativeFilename)
  {
  	lclContext *ctx = (lclContext *) AH->formatData;
  	char	   *dname;
  
  	dname = ctx->directory;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index d9aeee3..2e3085d 100644
*** a/src/bin/pg_dump/pg_dump.c
--- b/src/bin/pg_dump/pg_dump.c
*************** static Oid	findLastBuiltinOid_V70(Archiv
*** 231,238 ****
  static void selectSourceSchema(Archive *fout, const char *schemaName);
  static char *getFormattedTypeName(Archive *fout, Oid oid, OidOptions opts);
  static char *myFormatType(const char *typname, int32 typmod);
- static const char *fmtQualifiedId(Archive *fout,
- 			   const char *schema, const char *id);
  static void getBlobs(Archive *fout);
  static void dumpBlob(Archive *fout, BlobInfo *binfo);
  static int	dumpBlobs(Archive *fout, void *arg);
--- 231,236 ----
*************** static void binary_upgrade_extension_mem
*** 250,256 ****
  								DumpableObject *dobj,
  								const char *objlabel);
  static const char *getAttrName(int attrnum, TableInfo *tblInfo);
! static const char *fmtCopyColumnList(const TableInfo *ti);
  static PGresult *ExecuteSqlQueryForSingleRow(Archive *fout, char *query);
  
  
--- 248,254 ----
  								DumpableObject *dobj,
  								const char *objlabel);
  static const char *getAttrName(int attrnum, TableInfo *tblInfo);
! static const char *fmtCopyColumnList(const TableInfo *ti, PQExpBuffer buffer);
  static PGresult *ExecuteSqlQueryForSingleRow(Archive *fout, char *query);
  
  
*************** dumpTableData_copy(Archive *fout, void *
*** 1214,1219 ****
--- 1212,1222 ----
  	const bool	hasoids = tbinfo->hasoids;
  	const bool	oids = tdinfo->oids;
  	PQExpBuffer q = createPQExpBuffer();
+ 	/*
+ 	 * Note: can't use getThreadLocalPQExpBuffer() here, we're calling fmtId which
+ 	 * uses it already.
+ 	 */
+ 	PQExpBuffer clistBuf = createPQExpBuffer();
  	PGconn	   *conn = GetConnection(fout);
  	PGresult   *res;
  	int			ret;
*************** dumpTableData_copy(Archive *fout, void *
*** 1238,1251 ****
  	 * cases involving ADD COLUMN and inheritance.)
  	 */
  	if (fout->remoteVersion >= 70300)
! 		column_list = fmtCopyColumnList(tbinfo);
  	else
  		column_list = "";		/* can't select columns in COPY */
  
  	if (oids && hasoids)
  	{
  		appendPQExpBuffer(q, "COPY %s %s WITH OIDS TO stdout;",
! 						  fmtQualifiedId(fout,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname),
  						  column_list);
--- 1241,1254 ----
  	 * cases involving ADD COLUMN and inheritance.)
  	 */
  	if (fout->remoteVersion >= 70300)
! 		column_list = fmtCopyColumnList(tbinfo, clistBuf);
  	else
  		column_list = "";		/* can't select columns in COPY */
  
  	if (oids && hasoids)
  	{
  		appendPQExpBuffer(q, "COPY %s %s WITH OIDS TO stdout;",
! 						  fmtQualifiedId(fout->remoteVersion,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname),
  						  column_list);
*************** dumpTableData_copy(Archive *fout, void *
*** 1263,1269 ****
  		else
  			appendPQExpBufferStr(q, "* ");
  		appendPQExpBuffer(q, "FROM %s %s) TO stdout;",
! 						  fmtQualifiedId(fout,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname),
  						  tdinfo->filtercond);
--- 1266,1272 ----
  		else
  			appendPQExpBufferStr(q, "* ");
  		appendPQExpBuffer(q, "FROM %s %s) TO stdout;",
! 						  fmtQualifiedId(fout->remoteVersion,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname),
  						  tdinfo->filtercond);
*************** dumpTableData_copy(Archive *fout, void *
*** 1271,1283 ****
  	else
  	{
  		appendPQExpBuffer(q, "COPY %s %s TO stdout;",
! 						  fmtQualifiedId(fout,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname),
  						  column_list);
  	}
  	res = ExecuteSqlQuery(fout, q->data, PGRES_COPY_OUT);
  	PQclear(res);
  
  	for (;;)
  	{
--- 1274,1287 ----
  	else
  	{
  		appendPQExpBuffer(q, "COPY %s %s TO stdout;",
! 						  fmtQualifiedId(fout->remoteVersion,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname),
  						  column_list);
  	}
  	res = ExecuteSqlQuery(fout, q->data, PGRES_COPY_OUT);
  	PQclear(res);
+ 	destroyPQExpBuffer(clistBuf);
  
  	for (;;)
  	{
*************** dumpTableData_insert(Archive *fout, void
*** 1396,1402 ****
  	{
  		appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
  						  "SELECT * FROM ONLY %s",
! 						  fmtQualifiedId(fout,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname));
  	}
--- 1400,1406 ----
  	{
  		appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
  						  "SELECT * FROM ONLY %s",
! 						  fmtQualifiedId(fout->remoteVersion,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname));
  	}
*************** dumpTableData_insert(Archive *fout, void
*** 1404,1410 ****
  	{
  		appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
  						  "SELECT * FROM %s",
! 						  fmtQualifiedId(fout,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname));
  	}
--- 1408,1414 ----
  	{
  		appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
  						  "SELECT * FROM %s",
! 						  fmtQualifiedId(fout->remoteVersion,
  										 tbinfo->dobj.namespace->dobj.name,
  										 classname));
  	}
*************** dumpTableData(Archive *fout, TableDataIn
*** 1536,1541 ****
--- 1540,1546 ----
  {
  	TableInfo  *tbinfo = tdinfo->tdtable;
  	PQExpBuffer copyBuf = createPQExpBuffer();
+ 	PQExpBuffer clistBuf = createPQExpBuffer();
  	DataDumperPtr dumpFn;
  	char	   *copyStmt;
  
*************** dumpTableData(Archive *fout, TableDataIn
*** 1547,1553 ****
  		appendPQExpBuffer(copyBuf, "COPY %s ",
  						  fmtId(tbinfo->dobj.name));
  		appendPQExpBuffer(copyBuf, "%s %sFROM stdin;\n",
! 						  fmtCopyColumnList(tbinfo),
  					  (tdinfo->oids && tbinfo->hasoids) ? "WITH OIDS " : "");
  		copyStmt = copyBuf->data;
  	}
--- 1552,1558 ----
  		appendPQExpBuffer(copyBuf, "COPY %s ",
  						  fmtId(tbinfo->dobj.name));
  		appendPQExpBuffer(copyBuf, "%s %sFROM stdin;\n",
! 						  fmtCopyColumnList(tbinfo, clistBuf),
  					  (tdinfo->oids && tbinfo->hasoids) ? "WITH OIDS " : "");
  		copyStmt = copyBuf->data;
  	}
*************** dumpTableData(Archive *fout, TableDataIn
*** 1567,1572 ****
--- 1572,1578 ----
  				 dumpFn, tdinfo);
  
  	destroyPQExpBuffer(copyBuf);
+ 	destroyPQExpBuffer(clistBuf);
  }
  
  /*
*************** getTables(Archive *fout, int *numTables)
*** 3833,3838 ****
--- 3839,3845 ----
  	int			i_reloptions;
  	int			i_toastreloptions;
  	int			i_reloftype;
+ 	int			i_relpages;
  
  	/* Make sure we are in proper schema */
  	selectSourceSchema(fout, "pg_catalog");
*************** getTables(Archive *fout, int *numTables)
*** 3872,3877 ****
--- 3879,3885 ----
  						  "c.relfrozenxid, tc.oid AS toid, "
  						  "tc.relfrozenxid AS tfrozenxid, "
  						  "c.relpersistence, "
+ 						  "c.relpages, "
  						  "CASE WHEN c.reloftype <> 0 THEN c.reloftype::pg_catalog.regtype ELSE NULL END AS reloftype, "
  						  "d.refobjid AS owning_tab, "
  						  "d.refobjsubid AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 3908,3913 ****
--- 3916,3922 ----
  						  "c.relfrozenxid, tc.oid AS toid, "
  						  "tc.relfrozenxid AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "c.relpages, "
  						  "CASE WHEN c.reloftype <> 0 THEN c.reloftype::pg_catalog.regtype ELSE NULL END AS reloftype, "
  						  "d.refobjid AS owning_tab, "
  						  "d.refobjsubid AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 3943,3948 ****
--- 3952,3958 ----
  						  "c.relfrozenxid, tc.oid AS toid, "
  						  "tc.relfrozenxid AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "c.relpages, "
  						  "NULL AS reloftype, "
  						  "d.refobjid AS owning_tab, "
  						  "d.refobjsubid AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 3978,3983 ****
--- 3988,3994 ----
  						  "c.relfrozenxid, tc.oid AS toid, "
  						  "tc.relfrozenxid AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "c.relpages, "
  						  "NULL AS reloftype, "
  						  "d.refobjid AS owning_tab, "
  						  "d.refobjsubid AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 4014,4019 ****
--- 4025,4031 ----
  						  "0 AS toid, "
  						  "0 AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "relpages, "
  						  "NULL AS reloftype, "
  						  "d.refobjid AS owning_tab, "
  						  "d.refobjsubid AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 4049,4054 ****
--- 4061,4067 ----
  						  "0 AS toid, "
  						  "0 AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "relpages, "
  						  "NULL AS reloftype, "
  						  "d.refobjid AS owning_tab, "
  						  "d.refobjsubid AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 4080,4085 ****
--- 4093,4099 ----
  						  "0 AS toid, "
  						  "0 AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "relpages, "
  						  "NULL AS reloftype, "
  						  "NULL::oid AS owning_tab, "
  						  "NULL::int4 AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 4106,4111 ****
--- 4120,4126 ----
  						  "0 AS toid, "
  						  "0 AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "relpages, "
  						  "NULL AS reloftype, "
  						  "NULL::oid AS owning_tab, "
  						  "NULL::int4 AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 4142,4147 ****
--- 4157,4163 ----
  						  "0 AS toid, "
  						  "0 AS tfrozenxid, "
  						  "'p' AS relpersistence, "
+ 						  "0 AS relpages, "
  						  "NULL AS reloftype, "
  						  "NULL::oid AS owning_tab, "
  						  "NULL::int4 AS owning_col, "
*************** getTables(Archive *fout, int *numTables)
*** 4195,4200 ****
--- 4211,4217 ----
  	i_reloptions = PQfnumber(res, "reloptions");
  	i_toastreloptions = PQfnumber(res, "toast_reloptions");
  	i_reloftype = PQfnumber(res, "reloftype");
+ 	i_relpages = PQfnumber(res, "relpages");
  
  	if (lockWaitTimeout && fout->remoteVersion >= 70300)
  	{
*************** getTables(Archive *fout, int *numTables)
*** 4251,4256 ****
--- 4268,4274 ----
  		tblinfo[i].reltablespace = pg_strdup(PQgetvalue(res, i, i_reltablespace));
  		tblinfo[i].reloptions = pg_strdup(PQgetvalue(res, i, i_reloptions));
  		tblinfo[i].toast_reloptions = pg_strdup(PQgetvalue(res, i, i_toastreloptions));
+ 		tblinfo[i].relpages = atoi(PQgetvalue(res, i, i_relpages));
  
  		/* other fields were zeroed above */
  
*************** getTables(Archive *fout, int *numTables)
*** 4279,4285 ****
  			resetPQExpBuffer(query);
  			appendPQExpBuffer(query,
  							  "LOCK TABLE %s IN ACCESS SHARE MODE",
! 							  fmtQualifiedId(fout,
  										tblinfo[i].dobj.namespace->dobj.name,
  											 tblinfo[i].dobj.name));
  			ExecuteSqlStatement(fout, query->data);
--- 4297,4303 ----
  			resetPQExpBuffer(query);
  			appendPQExpBuffer(query,
  							  "LOCK TABLE %s IN ACCESS SHARE MODE",
! 							  fmtQualifiedId(fout->remoteVersion,
  										tblinfo[i].dobj.namespace->dobj.name,
  											 tblinfo[i].dobj.name));
  			ExecuteSqlStatement(fout, query->data);
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4418,4424 ****
  				i_conoid,
  				i_condef,
  				i_tablespace,
! 				i_options;
  	int			ntups;
  
  	for (i = 0; i < numTables; i++)
--- 4436,4443 ----
  				i_conoid,
  				i_condef,
  				i_tablespace,
! 				i_options,
! 				i_relpages;
  	int			ntups;
  
  	for (i = 0; i < numTables; i++)
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4460,4465 ****
--- 4479,4485 ----
  					 "pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
  							  "t.relnatts AS indnkeys, "
  							  "i.indkey, i.indisclustered, "
+ 							  "t.relpages, "
  							  "c.contype, c.conname, "
  							  "c.condeferrable, c.condeferred, "
  							  "c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4485,4490 ****
--- 4505,4511 ----
  					 "pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
  							  "t.relnatts AS indnkeys, "
  							  "i.indkey, i.indisclustered, "
+ 							  "t.relpages, "
  							  "c.contype, c.conname, "
  							  "c.condeferrable, c.condeferred, "
  							  "c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4513,4518 ****
--- 4534,4540 ----
  					 "pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
  							  "t.relnatts AS indnkeys, "
  							  "i.indkey, i.indisclustered, "
+ 							  "t.relpages, "
  							  "c.contype, c.conname, "
  							  "c.condeferrable, c.condeferred, "
  							  "c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4541,4546 ****
--- 4563,4569 ----
  					 "pg_catalog.pg_get_indexdef(i.indexrelid) AS indexdef, "
  							  "t.relnatts AS indnkeys, "
  							  "i.indkey, i.indisclustered, "
+ 							  "t.relpages, "
  							  "c.contype, c.conname, "
  							  "c.condeferrable, c.condeferred, "
  							  "c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4569,4574 ****
--- 4592,4598 ----
  							  "pg_get_indexdef(i.indexrelid) AS indexdef, "
  							  "t.relnatts AS indnkeys, "
  							  "i.indkey, false AS indisclustered, "
+ 							  "t.relpages, "
  							  "CASE WHEN i.indisprimary THEN 'p'::char "
  							  "ELSE '0'::char END AS contype, "
  							  "t.relname AS conname, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4595,4600 ****
--- 4619,4625 ----
  							  "pg_get_indexdef(i.indexrelid) AS indexdef, "
  							  "t.relnatts AS indnkeys, "
  							  "i.indkey, false AS indisclustered, "
+ 							  "t.relpages, "
  							  "CASE WHEN i.indisprimary THEN 'p'::char "
  							  "ELSE '0'::char END AS contype, "
  							  "t.relname AS conname, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4623,4628 ****
--- 4648,4654 ----
  		i_indnkeys = PQfnumber(res, "indnkeys");
  		i_indkey = PQfnumber(res, "indkey");
  		i_indisclustered = PQfnumber(res, "indisclustered");
+ 		i_relpages = PQfnumber(res, "relpages");
  		i_contype = PQfnumber(res, "contype");
  		i_conname = PQfnumber(res, "conname");
  		i_condeferrable = PQfnumber(res, "condeferrable");
*************** getIndexes(Archive *fout, TableInfo tbli
*** 4665,4670 ****
--- 4691,4697 ----
  			parseOidArray(PQgetvalue(res, j, i_indkey),
  						  indxinfo[j].indkeys, INDEX_MAX_KEYS);
  			indxinfo[j].indisclustered = (PQgetvalue(res, j, i_indisclustered)[0] == 't');
+ 			indxinfo[j].relpages = atoi(PQgetvalue(res, j, i_relpages));
  			contype = *(PQgetvalue(res, j, i_contype));
  
  			if (contype == 'p' || contype == 'u' || contype == 'x')
*************** getDependencies(Archive *fout)
*** 14038,14059 ****
   *
   * Whenever the selected schema is not pg_catalog, be careful to qualify
   * references to system catalogs and types in our emitted commands!
   */
  static void
  selectSourceSchema(Archive *fout, const char *schemaName)
  {
- 	static char *curSchemaName = NULL;
  	PQExpBuffer query;
  
  	/* Not relevant if fetching from pre-7.3 DB */
  	if (fout->remoteVersion < 70300)
  		return;
- 	/* Ignore null schema names */
- 	if (schemaName == NULL || *schemaName == '\0')
- 		return;
- 	/* Optimize away repeated selection of same schema */
- 	if (curSchemaName && strcmp(curSchemaName, schemaName) == 0)
- 		return;
  
  	query = createPQExpBuffer();
  	appendPQExpBuffer(query, "SET search_path = %s",
--- 14065,14085 ----
   *
   * Whenever the selected schema is not pg_catalog, be careful to qualify
   * references to system catalogs and types in our emitted commands!
+  *
+  * This function is called only from selectSourceSchemaOnAH and
+  * selectSourceSchema.
   */
  static void
  selectSourceSchema(Archive *fout, const char *schemaName)
  {
  	PQExpBuffer query;
  
+ 	/* This is checked by the callers already */
+ 	Assert(schemaName != NULL && *schemaName != '\0');
+ 
  	/* Not relevant if fetching from pre-7.3 DB */
  	if (fout->remoteVersion < 70300)
  		return;
  
  	query = createPQExpBuffer();
  	appendPQExpBuffer(query, "SET search_path = %s",
*************** selectSourceSchema(Archive *fout, const
*** 14064,14072 ****
  	ExecuteSqlStatement(fout, query->data);
  
  	destroyPQExpBuffer(query);
- 	if (curSchemaName)
- 		free(curSchemaName);
- 	curSchemaName = pg_strdup(schemaName);
  }
  
  /*
--- 14090,14095 ----
*************** myFormatType(const char *typname, int32
*** 14204,14274 ****
  }
  
  /*
-  * fmtQualifiedId - convert a qualified name to the proper format for
-  * the source database.
-  *
-  * Like fmtId, use the result before calling again.
-  */
- static const char *
- fmtQualifiedId(Archive *fout, const char *schema, const char *id)
- {
- 	static PQExpBuffer id_return = NULL;
- 
- 	if (id_return)				/* first time through? */
- 		resetPQExpBuffer(id_return);
- 	else
- 		id_return = createPQExpBuffer();
- 
- 	/* Suppress schema name if fetching from pre-7.3 DB */
- 	if (fout->remoteVersion >= 70300 && schema && *schema)
- 	{
- 		appendPQExpBuffer(id_return, "%s.",
- 						  fmtId(schema));
- 	}
- 	appendPQExpBuffer(id_return, "%s",
- 					  fmtId(id));
- 
- 	return id_return->data;
- }
- 
- /*
   * Return a column list clause for the given relation.
   *
   * Special case: if there are no undropped columns in the relation, return
   * "", not an invalid "()" column list.
   */
  static const char *
! fmtCopyColumnList(const TableInfo *ti)
  {
- 	static PQExpBuffer q = NULL;
  	int			numatts = ti->numatts;
  	char	  **attnames = ti->attnames;
  	bool	   *attisdropped = ti->attisdropped;
  	bool		needComma;
  	int			i;
  
! 	if (q)						/* first time through? */
! 		resetPQExpBuffer(q);
! 	else
! 		q = createPQExpBuffer();
! 
! 	appendPQExpBuffer(q, "(");
  	needComma = false;
  	for (i = 0; i < numatts; i++)
  	{
  		if (attisdropped[i])
  			continue;
  		if (needComma)
! 			appendPQExpBuffer(q, ", ");
! 		appendPQExpBuffer(q, "%s", fmtId(attnames[i]));
  		needComma = true;
  	}
  
  	if (!needComma)
  		return "";				/* no undropped columns */
  
! 	appendPQExpBuffer(q, ")");
! 	return q->data;
  }
  
  /*
--- 14227,14263 ----
  }
  
  /*
   * Return a column list clause for the given relation.
   *
   * Special case: if there are no undropped columns in the relation, return
   * "", not an invalid "()" column list.
   */
  static const char *
! fmtCopyColumnList(const TableInfo *ti, PQExpBuffer buffer)
  {
  	int			numatts = ti->numatts;
  	char	  **attnames = ti->attnames;
  	bool	   *attisdropped = ti->attisdropped;
  	bool		needComma;
  	int			i;
  
! 	appendPQExpBuffer(buffer, "(");
  	needComma = false;
  	for (i = 0; i < numatts; i++)
  	{
  		if (attisdropped[i])
  			continue;
  		if (needComma)
! 			appendPQExpBuffer(buffer, ", ");
! 		appendPQExpBuffer(buffer, "%s", fmtId(attnames[i]));
  		needComma = true;
  	}
  
  	if (!needComma)
  		return "";				/* no undropped columns */
  
! 	appendPQExpBuffer(buffer, ")");
! 	return buffer->data;
  }
  
  /*
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index f26d8d3..4efafcf 100644
*** a/src/bin/pg_dump/pg_dump.h
--- b/src/bin/pg_dump/pg_dump.h
*************** typedef struct _tableInfo
*** 256,261 ****
--- 256,262 ----
  	/* these two are set only if table is a sequence owned by a column: */
  	Oid			owning_tab;		/* OID of table owning sequence */
  	int			owning_col;		/* attr # of column owning sequence */
+ 	int			relpages;
  
  	bool		interesting;	/* true if need to collect more data */
  
*************** typedef struct _indxInfo
*** 319,324 ****
--- 320,326 ----
  	bool		indisclustered;
  	/* if there is an associated constraint object, its dumpId: */
  	DumpId		indexconstraint;
+ 	int			relpages;		/* relpages of the underlying table */
  } IndxInfo;
  
  typedef struct _ruleInfo
*************** extern void parseOidArray(const char *st
*** 523,528 ****
--- 525,531 ----
  extern void sortDumpableObjects(DumpableObject **objs, int numObjs);
  extern void sortDumpableObjectsByTypeName(DumpableObject **objs, int numObjs);
  extern void sortDumpableObjectsByTypeOid(DumpableObject **objs, int numObjs);
+ extern void sortDataAndIndexObjectsBySize(DumpableObject **objs, int numObjs);
  
  /*
   * version specific routines
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 9a82e4b..bfea171 100644
*** a/src/bin/pg_dump/pg_dump_sort.c
--- b/src/bin/pg_dump/pg_dump_sort.c
*************** static void repairDependencyLoop(Dumpabl
*** 121,126 ****
--- 121,213 ----
  static void describeDumpableObject(DumpableObject *obj,
  					   char *buf, int bufsize);
  
+ static int DOSizeCompare(const void *p1, const void *p2);
+ 
+ static int
+ findFirstEqualType(DumpableObjectType type, DumpableObject **objs, int numObjs)
+ {
+ 	int i;
+ 	for (i = 0; i < numObjs; i++)
+ 		if (objs[i]->objType == type)
+ 			return i;
+ 	return -1;
+ }
+ 
+ static int
+ findFirstDifferentType(DumpableObjectType type, DumpableObject **objs, int numObjs, int start)
+ {
+ 	int i;
+ 	for (i = start; i < numObjs; i++)
+ 		if (objs[i]->objType != type)
+ 			return i;
+ 	return numObjs - 1;
+ }
+ 
+ /*
+  * When we do a parallel dump, we want to start with the largest items first.
+  *
+  * Say we have the objects in this order:
+  * ....DDDDD....III....
+  *
+  * with D = Table data, I = Index, . = other object
+  *
+  * This sorting function now takes each of the D or I blocks and sorts them
+  * according to their size.
+  */
+ void
+ sortDataAndIndexObjectsBySize(DumpableObject **objs, int numObjs)
+ {
+ 	int		startIdx, endIdx;
+ 	void   *startPtr;
+ 
+ 	if (numObjs <= 1)
+ 		return;
+ 
+ 	startIdx = findFirstEqualType(DO_TABLE_DATA, objs, numObjs);
+ 	if (startIdx >= 0)
+ 	{
+ 		endIdx = findFirstDifferentType(DO_TABLE_DATA, objs, numObjs, startIdx);
+ 		startPtr = objs + startIdx;
+ 		qsort(startPtr, endIdx - startIdx, sizeof(DumpableObject *),
+ 			  DOSizeCompare);
+ 	}
+ 
+ 	startIdx = findFirstEqualType(DO_INDEX, objs, numObjs);
+ 	if (startIdx >= 0)
+ 	{
+ 		endIdx = findFirstDifferentType(DO_INDEX, objs, numObjs, startIdx);
+ 		startPtr = objs + startIdx;
+ 		qsort(startPtr, endIdx - startIdx, sizeof(DumpableObject *),
+ 			  DOSizeCompare);
+ 	}
+ }
+ 
+ static int
+ DOSizeCompare(const void *p1, const void *p2)
+ {
+ 	DumpableObject *obj1 = *(DumpableObject **) p1;
+ 	DumpableObject *obj2 = *(DumpableObject **) p2;
+ 	int			obj1_size = 0;
+ 	int			obj2_size = 0;
+ 
+ 	if (obj1->objType == DO_TABLE_DATA)
+ 		obj1_size = ((TableDataInfo *) obj1)->tdtable->relpages;
+ 	if (obj1->objType == DO_INDEX)
+ 		obj1_size = ((IndxInfo *) obj1)->relpages;
+ 
+ 	if (obj2->objType == DO_TABLE_DATA)
+ 		obj2_size = ((TableDataInfo *) obj2)->tdtable->relpages;
+ 	if (obj2->objType == DO_INDEX)
+ 		obj2_size = ((IndxInfo *) obj2)->relpages;
+ 
+ 	/* we want to see the biggest item go first */
+ 	if (obj1_size > obj2_size)
+ 		return -1;
+ 	if (obj2_size > obj1_size)
+ 		return 1;
+ 
+ 	return 0;
+ }
  
  /*
   * Sort the given objects into a type/name-based ordering
diff --git a/src/bin/pg_dump/pg_restore.c b/src/bin/pg_dump/pg_restore.c
index f371527..a644e6a 100644
*** a/src/bin/pg_dump/pg_restore.c
--- b/src/bin/pg_dump/pg_restore.c
*************** main(int argc, char **argv)
*** 72,77 ****
--- 72,78 ----
  	RestoreOptions *opts;
  	int			c;
  	int			exit_code;
+ 	int			numWorkers = 1;
  	Archive    *AH;
  	char	   *inputFileSpec;
  	static int	disable_triggers = 0;
*************** main(int argc, char **argv)
*** 183,189 ****
  				break;
  
  			case 'j':			/* number of restore jobs */
! 				opts->number_of_jobs = atoi(optarg);
  				break;
  
  			case 'l':			/* Dump the TOC summary */
--- 184,190 ----
  				break;
  
  			case 'j':			/* number of restore jobs */
! 				numWorkers = atoi(optarg);
  				break;
  
  			case 'l':			/* Dump the TOC summary */
*************** main(int argc, char **argv)
*** 314,320 ****
  	}
  
  	/* Can't do single-txn mode with multiple connections */
! 	if (opts->single_txn && opts->number_of_jobs > 1)
  	{
  		fprintf(stderr, _("%s: cannot specify both --single-transaction and multiple jobs\n"),
  				progname);
--- 315,321 ----
  	}
  
  	/* Can't do single-txn mode with multiple connections */
! 	if (opts->single_txn && numWorkers > 1)
  	{
  		fprintf(stderr, _("%s: cannot specify both --single-transaction and multiple jobs\n"),
  				progname);
*************** main(int argc, char **argv)
*** 373,378 ****
--- 374,381 ----
  	if (opts->tocFile)
  		SortTocFromFile(AH, opts);
  
+ 	AH->numWorkers = numWorkers;
+ 
  	if (opts->tocSummary)
  		PrintTOCSummary(AH, opts);
  	else