pg_backup_mirror_1.diff

diff --git a/src/bin/pg_dump/Makefile b/src/bin/pg_dump/Makefile
index 00501de..8e2e983 100644
*** a/src/bin/pg_dump/Makefile
--- b/src/bin/pg_dump/Makefile
*************** kwlookup.c: % : $(top_srcdir)/src/backen
*** 30,37 ****
  
  all: pg_dump pg_restore pg_dumpall
  
! pg_dump: pg_dump.o common.o pg_dump_sort.o $(OBJS) $(KEYWRDOBJS) | submake-libpq submake-libpgport
! 	$(CC) $(CFLAGS) pg_dump.o common.o pg_dump_sort.o $(KEYWRDOBJS) $(OBJS) $(libpq_pgport) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
  
  pg_restore: pg_restore.o $(OBJS) $(KEYWRDOBJS) | submake-libpq submake-libpgport
  	$(CC) $(CFLAGS) pg_restore.o $(KEYWRDOBJS) $(OBJS) $(libpq_pgport) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
--- 30,37 ----
  
  all: pg_dump pg_restore pg_dumpall
  
! pg_dump: pg_dump.o common.o pg_dump_sort.o pg_backup_mirror.o $(OBJS) $(KEYWRDOBJS) | submake-libpq submake-libpgport
! 	$(CC) $(CFLAGS) pg_dump.o common.o pg_dump_sort.o pg_backup_mirror.o $(KEYWRDOBJS) $(OBJS) $(libpq_pgport) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
  
  pg_restore: pg_restore.o $(OBJS) $(KEYWRDOBJS) | submake-libpq submake-libpgport
  	$(CC) $(CFLAGS) pg_restore.o $(KEYWRDOBJS) $(OBJS) $(libpq_pgport) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c
index bcde24c..b4d0af4 100644
*** a/src/bin/pg_dump/parallel.c
--- b/src/bin/pg_dump/parallel.c
*************** WaitForCommands(ArchiveHandle *AH, int p
*** 921,927 ****
  		}
  		else if (messageStartsWith(command, "RESTORE "))
  		{
! 			Assert(AH->format == archDirectory || AH->format == archCustom);
  			Assert(AH->connection != NULL);
  
  			sscanf(command + strlen("RESTORE "), "%d%n", &dumpId, &nBytes);
--- 921,927 ----
  		}
  		else if (messageStartsWith(command, "RESTORE "))
  		{
! 			Assert(AH->format == archDirectory || AH->format == archCustom || AH->format == archMirror);
  			Assert(AH->connection != NULL);
  
  			sscanf(command + strlen("RESTORE "), "%d%n", &dumpId, &nBytes);
*************** ListenToWorkers(ArchiveHandle *AH, Paral
*** 1004,1010 ****
  	}
  	else if (messageStartsWith(msg, "ERROR "))
  	{
! 		Assert(AH->format == archDirectory || AH->format == archCustom);
  		pstate->parallelSlot[worker].workerStatus = WRKR_TERMINATED;
  		die_horribly(AH, modulename, "%s", msg + strlen("ERROR "));
  	}
--- 1004,1010 ----
  	}
  	else if (messageStartsWith(msg, "ERROR "))
  	{
! 		Assert(AH->format == archDirectory || AH->format == archCustom || AH->format == archMirror);
  		pstate->parallelSlot[worker].workerStatus = WRKR_TERMINATED;
  		die_horribly(AH, modulename, "%s", msg + strlen("ERROR "));
  	}
diff --git a/src/bin/pg_dump/parallel.h b/src/bin/pg_dump/parallel.h
index 4c86b9b..194dea4 100644
*** a/src/bin/pg_dump/parallel.h
--- b/src/bin/pg_dump/parallel.h
*************** typedef enum
*** 32,38 ****
  typedef enum _action
  {
  	ACT_DUMP,
! 	ACT_RESTORE,
  } T_Action;
  
  /* Arguments needed for a worker process */
--- 32,38 ----
  typedef enum _action
  {
  	ACT_DUMP,
! 	ACT_RESTORE
  } T_Action;
  
  /* Arguments needed for a worker process */
diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h
index 767f865..3c90904 100644
*** a/src/bin/pg_dump/pg_backup.h
--- b/src/bin/pg_dump/pg_backup.h
*************** typedef enum _archiveFormat
*** 51,57 ****
  	archFiles = 2,
  	archTar = 3,
  	archNull = 4,
! 	archDirectory = 5
  } ArchiveFormat;
  
  typedef enum _archiveMode
--- 51,58 ----
  	archFiles = 2,
  	archTar = 3,
  	archNull = 4,
! 	archDirectory = 5,
! 	archMirror = 6
  } ArchiveFormat;
  
  typedef enum _archiveMode
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 0c81dfe..68e5307 100644
*** a/src/bin/pg_dump/pg_backup_archiver.c
--- b/src/bin/pg_dump/pg_backup_archiver.c
*************** CloseArchive(Archive *AHX)
*** 154,160 ****
  	int			res = 0;
  	ArchiveHandle *AH = (ArchiveHandle *) AHX;
  
! 	(*AH->ClosePtr) (AH);
  
  	/* Close the output */
  	if (AH->gzOut)
--- 154,161 ----
  	int			res = 0;
  	ArchiveHandle *AH = (ArchiveHandle *) AHX;
  
! 	if (AH->ClosePtr)
! 		(*AH->ClosePtr) (AH);
  
  	/* Close the output */
  	if (AH->gzOut)
*************** _allocAH(const char *FileSpec, const Arc
*** 2015,2020 ****
--- 2016,2025 ----
  			InitArchiveFmt_Tar(AH);
  			break;
  
+ 		case archMirror:
+ 			InitArchiveFmt_Mirror(AH);
+ 			break;
+ 
  		default:
  			die_horribly(AH, modulename, "unrecognized file format \"%d\"\n", fmt);
  	}
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index 3b10384..929a32c 100644
*** a/src/bin/pg_dump/pg_backup_archiver.h
--- b/src/bin/pg_dump/pg_backup_archiver.h
*************** extern void EndRestoreBlob(ArchiveHandle
*** 398,406 ****
  extern void EndRestoreBlobs(ArchiveHandle *AH);
  
  extern void InitArchiveFmt_Custom(ArchiveHandle *AH);
  extern void InitArchiveFmt_Files(ArchiveHandle *AH);
  extern void InitArchiveFmt_Null(ArchiveHandle *AH);
- extern void InitArchiveFmt_Directory(ArchiveHandle *AH);
  extern void InitArchiveFmt_Tar(ArchiveHandle *AH);
  
  extern bool isValidTarHeader(char *header);
--- 398,407 ----
  extern void EndRestoreBlobs(ArchiveHandle *AH);
  
  extern void InitArchiveFmt_Custom(ArchiveHandle *AH);
+ extern void InitArchiveFmt_Directory(ArchiveHandle *AH);
  extern void InitArchiveFmt_Files(ArchiveHandle *AH);
+ extern void InitArchiveFmt_Mirror(ArchiveHandle *AH);
  extern void InitArchiveFmt_Null(ArchiveHandle *AH);
  extern void InitArchiveFmt_Tar(ArchiveHandle *AH);
  
  extern bool isValidTarHeader(char *header);
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 995cf31..5e1e8d9 100644
*** a/src/bin/pg_dump/pg_backup_directory.c
--- b/src/bin/pg_dump/pg_backup_directory.c
*************** static void _DeClone(ArchiveHandle *AH);
*** 89,96 ****
  
  static char *_MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act);
  static int _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, const char *str, T_Action act);
- static char *_WorkerJobRestoreDirectory(ArchiveHandle *AH, TocEntry *te);
- static char *_WorkerJobDumpDirectory(ArchiveHandle *AH, TocEntry *te);
  
  static void createDirectory(const char *dir);
  static char *prependDirectory(ArchiveHandle *AH, char *buf, const char *relativeFilename);
--- 89,94 ----
diff --git a/src/bin/pg_dump/pg_backup_mirror.c b/src/bin/pg_dump/pg_backup_mirror.c
index ...100cea1 .
*** a/src/bin/pg_dump/pg_backup_mirror.c
--- b/src/bin/pg_dump/pg_backup_mirror.c
***************
*** 0 ****
--- 1,453 ----
+ /*-------------------------------------------------------------------------
+  *
+  * pg_backup_mirror.c
+  *
+  * Mirror a database from one host to another by doing a parallel
+  * pg_backup/restore.
+  *
+  *	Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+  *	Portions Copyright (c) 1994, Regents of the University of California
+  *	Portions Copyright (c) 2000, Philip Warner
+  *
+  *	Rights are granted to use this software in any way so long
+  *	as this notice is not removed.
+  *
+  *	The author is not responsible for loss or damages that may
+  *	result from it's use.
+  *
+  * IDENTIFICATION
+  *		src/bin/pg_dump/pg_backup_mirror.c
+  *
+  *-------------------------------------------------------------------------
+  */
+ 
+ #include "compress_io.h"
+ #include "dumpmem.h"
+ #include "dumputils.h"
+ #include "parallel.h"
+ 
+ typedef struct
+ {
+ 	PGconn	   *dumpConn;
+ 	PGconn	   *restoreConn;
+ } lclContext;
+ 
+ typedef struct
+ {
+ 	int			status;
+ } lclTocEntry;
+ 
+ static const char *modulename = gettext_noop("mirror archiver");
+ 
+ /* prototypes for private functions */
+ static void _ArchiveEntry(ArchiveHandle *AH, TocEntry *te);
+ static size_t _WriteData(ArchiveHandle *AH, const void *data, size_t dLen);
+ static void _PrintTocData(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
+ 
+ static void _StartBlobs(ArchiveHandle *AH, TocEntry *te);
+ static void _StartBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
+ static void _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
+ static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
+ static void _Clone(ArchiveHandle *AH);
+ static void _DeClone(ArchiveHandle *AH);
+ 
+ static void _ReopenArchive(ArchiveHandle *AH);
+ 
+ static char *_MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act);
+ static int _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, const char *str, T_Action act);
+ static char *_WorkerJobRestoreMirror(ArchiveHandle *AH, TocEntry *te);
+ 
+ /*
+  *	Init routine required by ALL formats. This is a global routine
+  *	and should be declared in pg_backup_archiver.h
+  *
+  *	Its task is to create any extra archive context (using AH->formatData),
+  *	and to initialize the supported function pointers.
+  *
+  *	It should also prepare whatever its input source is for reading/writing,
+  *	and in the case of a read mode connection, it should load the Header & TOC.
+  */
+ void
+ InitArchiveFmt_Mirror(ArchiveHandle *AH)
+ {
+ 	lclContext *ctx;
+ 
+ 	/* Assuming static functions, this can be copied for each format. */
+ 	AH->ArchiveEntryPtr = _ArchiveEntry;
+ 	AH->StartDataPtr = NULL;
+ 	AH->WriteDataPtr = _WriteData;
+ 	AH->EndDataPtr = NULL;
+ 	AH->WriteBytePtr = NULL;
+ 	AH->ReadBytePtr = NULL;
+ 	AH->WriteBufPtr = NULL;
+ 	AH->ReadBufPtr = NULL;
+ 	AH->ClosePtr = NULL;
+ 	AH->ReopenPtr = _ReopenArchive;
+ 	AH->PrintTocDataPtr = _PrintTocData;
+ 	AH->ReadExtraTocPtr = NULL;
+ 	AH->WriteExtraTocPtr = NULL;
+ 	AH->PrintExtraTocPtr = NULL;
+ 
+ 	AH->StartBlobsPtr = _StartBlobs;
+ 	AH->StartBlobPtr = _StartBlob;
+ 	AH->EndBlobPtr = _EndBlob;
+ 	AH->EndBlobsPtr = _EndBlobs;
+ 
+ 	AH->ClonePtr = _Clone;
+ 	AH->DeClonePtr = _DeClone;
+ 
+ 	AH->WorkerJobRestorePtr = _WorkerJobRestoreMirror;
+ 	AH->WorkerJobDumpPtr = NULL;
+ 
+ 	AH->MasterStartParallelItemPtr = _MasterStartParallelItem;
+ 	AH->MasterEndParallelItemPtr = _MasterEndParallelItem;
+ 
+ 	/* Set up our private context */
+ 	ctx = (lclContext *) pg_calloc(1, sizeof(lclContext));
+ 	AH->formatData = (void *) ctx;
+ 
+ 	/* Initialize LO buffering */
+ 	AH->lo_buf_size = LOBBUFSIZE;
+ 	AH->lo_buf = (void *) pg_malloc(LOBBUFSIZE);
+ }
+ 
+ /*
+  * Called by the Archiver when the dumper creates a new TOC entry.
+ */
+ static void
+ _ArchiveEntry(ArchiveHandle *AH, TocEntry *te)
+ {
+ 	lclTocEntry *tctx = (lclTocEntry *) calloc(1, sizeof(lclTocEntry));
+ 	te->formatData = (void *) tctx;
+ }
+ 
+ void
+ setActiveConnection(ArchiveHandle *AH, T_Action act, bool connOnly)
+ {
+ 	lclContext *ctx = (lclContext *) AH->formatData;
+ 
+ 	if (act == ACT_DUMP)
+ 	{
+ 		AH->connection = ctx->dumpConn;
+ 		if (connOnly)
+ 			return;
+ 	}
+ 	else
+ 	{
+ 		AH->connection = ctx->restoreConn;
+ 		if (connOnly)
+ 			return;
+ 
+ 	}
+ }
+ 
+ void
+ MirrorDecoupleDumpConn(ArchiveHandle *AH)
+ {
+ 	lclContext *ctx = (lclContext *) AH->formatData;
+ 
+ 	ctx->dumpConn = AH->connection;
+ 	AH->connection = NULL;
+ }
+ 
+ /*
+  * Called by the archiver when saving TABLE DATA (not schema). This routine
+  * should save whatever format-specific information is needed to read
+  * the archive back.
+  *
+  * It is called just prior to the dumper's 'DataDumper' routine being called.
+  *
+  * We create the data file for writing.
+  *
+  * static void _StartData(ArchiveHandle *AH, TocEntry *te) { }
+  */
+ 
+ /*
+  * Called by archiver when dumper calls WriteData. This routine is
+  * called for both BLOB and TABLE data; it is the responsibility of
+  * the format to manage each kind of data using StartBlob/StartData.
+  *
+  * It should only be called from within a DataDumper routine.
+  */
+ static size_t
+ _WriteData(ArchiveHandle *AH, const void *data, size_t dLen)
+ {
+ 	setActiveConnection(AH, ACT_RESTORE, true);
+ 	ahwrite(data, 1, dLen, AH);
+ 	setActiveConnection(AH, ACT_DUMP, true);
+ 
+ 	return dLen;
+ }
+ 
+ /*
+  * Called by the archiver when a dumper's 'DataDumper' routine has
+  * finished.
+  *
+  * static void _EndData(ArchiveHandle *AH, TocEntry *te) { }
+  */
+ 
+ /*
+  * Print data for a given TOC entry
+ */
+ static void
+ _PrintTocData(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt)
+ {
+ 	setActiveConnection(AH, ACT_DUMP, true);
+ 	WriteDataChunksForTocEntry(AH, te);
+ 	setActiveConnection(AH, ACT_RESTORE, true);
+ }
+ 
+ /*
+  * Write a byte of data to the archive.
+  * Called by the archiver to do integer & byte output to the archive.
+  * These routines are only used to read & write the headers & TOC.
+  *
+  * static int _WriteByte(ArchiveHandle *AH, const int i) { }
+  */
+ 
+ /*
+  * Read a byte of data from the archive.
+  * Called by the archiver to read bytes & integers from the archive.
+  * These routines are only used to read & write headers & TOC.
+  * EOF should be treated as a fatal error.
+  *
+  * static int _ReadByte(ArchiveHandle *AH) { }
+  */
+ 
+ /*
+  * Write a buffer of data to the archive.
+  *
+  * Called by the archiver to write a block of bytes to the TOC or a data file.
+  *
+  * static size_t _WriteBuf(ArchiveHandle *AH, const void *buf, size_t len) { }
+  */
+ 
+ /*
+  * Read a block of bytes from the archive.
+  *
+  * Called by the archiver to read a block of bytes from the archive
+  *
+  * static size_t _ReadBuf(ArchiveHandle *AH, void *buf, size_t len) { }
+  */
+ 
+ /*
+  * Close the archive.
+  *
+  * When writing the archive, this is the routine that actually starts
+  * the process of saving it to files. No data should be written prior
+  * to this point, since the user could sort the TOC after creating it.
+  *
+  * If an archive is to be written, this routine must call:
+  *		WriteHead			to save the archive header
+  *		WriteToc			to save the TOC entries
+  *		WriteDataChunks		to save all DATA & BLOBs.
+  *
+  * static void _CloseArchive(ArchiveHandle *AH) { }
+  */
+ 
+ /*
+  * Reopen the archive's file handle.
+  */
+ static void _ReopenArchive(ArchiveHandle *AH) { }
+ 
+ /*
+  * BLOB support
+  */
+ 
+ /*
+  * Called by the archiver when starting to save all BLOB DATA (not schema).
+  * It is called just prior to the dumper's DataDumper routine.
+  *
+  * We open the large object TOC file here, so that we can append a line to
+  * it for each blob.
+  */
+ static void
+ _StartBlobs(ArchiveHandle *AH, TocEntry *te)
+ {
+ 	setActiveConnection(AH, ACT_RESTORE, true);
+ 	StartRestoreBlobs(AH);
+ 	setActiveConnection(AH, ACT_DUMP, true);
+ }
+ 
+ /*
+  * Called by the archiver when we're about to start dumping a blob.
+  *
+  * We create a file to write the blob to.
+  */
+ static void
+ _StartBlob(ArchiveHandle *AH, TocEntry *te, Oid oid)
+ {
+ 	setActiveConnection(AH, ACT_RESTORE, true);
+ 	StartRestoreBlob(AH, oid, AH->ropt->dropSchema);
+ 	setActiveConnection(AH, ACT_DUMP, true);
+ }
+ 
+ /*
+  * Called by the archiver when the dumper is finished writing a blob.
+  *
+  * We close the blob file and write an entry to the blob TOC file for it.
+  */
+ static void
+ _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid)
+ {
+ 	setActiveConnection(AH, ACT_RESTORE, true);
+ 	EndRestoreBlob(AH, oid);
+ 	setActiveConnection(AH, ACT_DUMP, true);
+ }
+ 
+ /*
+  * Called by the archiver when finishing saving all BLOB DATA.
+  *
+  * We close the blobs TOC file.
+  */
+ static void
+ _EndBlobs(ArchiveHandle *AH, TocEntry *te)
+ {
+ 	setActiveConnection(AH, ACT_RESTORE, true);
+ 	EndRestoreBlobs(AH);
+ 	setActiveConnection(AH, ACT_DUMP, true);
+ }
+ 
+ /*
+  * Clone format-specific fields during parallel restoration.
+  */
+ static void
+ _Clone(ArchiveHandle *AH)
+ {
+ 	char	   *dbname;
+ 	char	   *pghost;
+ 	char	   *pgport;
+ 	char	   *username;
+ 	const char *encname;
+ 
+ 	lclContext *ctx = (lclContext *) AH->formatData;
+ 
+ 	AH->formatData = (lclContext *) pg_malloc(sizeof(lclContext));
+ 	memcpy(AH->formatData, ctx, sizeof(lclContext));
+ 	ctx = (lclContext *) AH->formatData;
+ 
+ 	Assert(ctx->dumpConn != NULL);
+ 
+ 	ctx->restoreConn = AH->connection;
+ 	AH->connection = NULL;
+ 
+ 	/*
+ 	 * Even though we are technically accessing the parent's database object
+ 	 * here, these functions are fine to be called like that because all just
+ 	 * return a pointer and do not actually send/receive any data to/from the
+ 	 * database.
+ 	 */
+ 	dbname = PQdb(ctx->dumpConn);
+ 	pghost = PQhost(ctx->dumpConn);
+ 	pgport = PQport(ctx->dumpConn);
+ 	username = PQuser(ctx->dumpConn);
+ 	/* XXX could we have different encodings ? */
+ 	encname = pg_encoding_to_char(AH->public.encoding);
+ 
+ 	if (AH->savedPassword)
+ 		AH->savedPassword = NULL;
+ 
+ 	/* XXX this does more than we most likely want */
+ 	ConnectDatabase((Archive *) AH, dbname, pghost, pgport, username, TRI_NO);
+ 
+ 	ctx->dumpConn = AH->connection;
+ 
+ 	/* Let the dumpConn be in AH->connection so that we run SetupConnection on
+ 	 * it. */
+ 
+ 
+ 	/*
+ 	 * Note: we do not make a local lo_buf because we expect at most one BLOBS
+ 	 * entry per archive, so no parallelism is possible.  Likewise,
+ 	 * TOC-entry-local state isn't an issue because any one TOC entry is
+ 	 * touched by just one worker child.
+ 	 */
+ 
+ 	/*
+ 	 * We also don't copy the ParallelState pointer (pstate), only the master
+ 	 * process ever writes to it.
+ 	 */
+ }
+ 
+ static void
+ _DeClone(ArchiveHandle *AH)
+ {
+ 	lclContext *ctx = (lclContext *) AH->formatData;
+ 	free(ctx);
+ }
+ 
+ /*
+  * This function is executed in the parent process. Depending on the desired
+  * action (dump or restore) it creates a string that is understood by the
+  * _WorkerJobDumpDirectory/_WorkerJobRestoreDirectory functions of the
+  * respective dump format.
+  */
+ static char *
+ _MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act)
+ {
+ 	/*
+ 	 * A static char is okay here, even on Windows because we call this
+ 	 * function only from one process (the master).
+ 	 */
+ 	static char	buf[64];
+ 	snprintf(buf, sizeof(buf), "RESTORE %d", te->dumpId);
+ 	return buf;
+ }
+ 
+ /*
+  * This function is executed in the child of a parallel backup for the
+  * directory archive and dumps the actual data.
+  */
+ static char *
+ _WorkerJobRestoreMirror(ArchiveHandle *AH, TocEntry *te)
+ {
+ 	/* short fixed-size string + some ID so far, this needs to be malloc'ed
+ 	 * instead of static because we work with threads on windows */
+ 	const int	buflen = 64;
+ 	char	   *buf = (char*) pg_malloc(buflen);
+ 	ParallelArgs pargs;
+ 	int			status;
+ 	lclTocEntry *tctx;
+ 
+ 	tctx = (lclTocEntry *) te->formatData;
+ 
+ 	/* Prepare the item */
+ 	pargs.AH = AH;
+ 	pargs.te = te;
+ 
+ 	/* Prepare AH */
+ 	setActiveConnection(AH, ACT_RESTORE, false);
+ 
+ 	status = parallel_restore(&pargs);
+ 
+ 	tctx->status = status;
+ 	snprintf(buf, buflen, "OK RESTORE %d %d %d", te->dumpId, status,
+ 			 status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0);
+ 
+ 	setActiveConnection(AH, ACT_RESTORE, false);
+ 
+ 	return buf;
+ }
+ /*
+  * This function is executed in the parent process. It analyzes the response of
+  * the _WorkerJobDumpDirectory/_WorkerJobRestoreDirectory functions of the
+  * respective dump format.
+  */
+ static int
+ _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, const char *str, T_Action act)
+ {
+ 	DumpId		dumpId;
+ 	int			nBytes, n_errors;
+ 	int			status = 0;
+ 
+ 	Assert(act == ACT_RESTORE);
+ 
+ 	sscanf(str, "%u %u %u%n", &dumpId, &status, &n_errors, &nBytes);
+ 
+ 	Assert(dumpId == te->dumpId);
+ 	Assert(nBytes == strlen(str));
+ 
+ 	AH->public.n_errors += n_errors;
+ 
+ 	return status;
+ }
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 5a15435..e8b60ff 100644
*** a/src/bin/pg_dump/pg_dump.c
--- b/src/bin/pg_dump/pg_dump.c
*************** static int	no_security_labels = 0;
*** 143,148 ****
--- 143,149 ----
  static int  no_synchronized_snapshots = 0;
  static int	no_unlogged_table_data = 0;
  static int	serializable_deferrable = 0;
+ static int  mirror_exit_on_error = false;
  
  
  static void help(const char *progname);
*************** main(int argc, char **argv)
*** 266,271 ****
--- 267,276 ----
  	const char *pghost = NULL;
  	const char *pgport = NULL;
  	const char *username = NULL;
+ 	char	   *restore_dbname = NULL;
+ 	char	   *restore_pghost = NULL;
+ 	char	   *restore_pgport = NULL;
+ 	char	   *restore_username = NULL;
  	const char *dumpencoding = NULL;
  	bool		oids = false;
  	TableInfo  *tblinfo;
*************** main(int argc, char **argv)
*** 288,297 ****
--- 293,304 ----
  	RestoreOptions *ropt;
  	ArchiveFormat archiveFormat = archUnknown;
  	ArchiveMode archiveMode;
+ 	int			exit_code;
  
  	static int	disable_triggers = 0;
  	static int	outputNoTablespaces = 0;
  	static int	use_setsessauth = 0;
+ 	static int	in_restore_options = false;
  
  	static struct option long_options[] = {
  		{"data-only", no_argument, NULL, 'a'},
*************** main(int argc, char **argv)
*** 345,350 ****
--- 352,365 ----
  		{"no-synchronized-snapshots", no_argument, &no_synchronized_snapshots, 1},
  		{"no-unlogged-table-data", no_argument, &no_unlogged_table_data, 1},
  
+ 		/*
+ 		 * the following option switches to target db interpretation for the
+ 		 * pg_backup_mirror options.
+ 		 */
+ 		{"restore", no_argument, &in_restore_options, 1},
+ 		{"dbname", required_argument, NULL, 6},
+ 		{"mirror-exit-on-error", no_argument, &mirror_exit_on_error, 1},
+ 
  		{NULL, 0, NULL, 0}
  	};
  
*************** main(int argc, char **argv)
*** 414,420 ****
  				break;
  
  			case 'h':			/* server host */
! 				pghost = optarg;
  				break;
  
  			case 'i':
--- 429,438 ----
  				break;
  
  			case 'h':			/* server host */
! 				if (!in_restore_options)
! 					pghost = optarg;
! 				else
! 					restore_pghost = optarg;
  				break;
  
  			case 'i':
*************** main(int argc, char **argv)
*** 443,449 ****
  				break;
  
  			case 'p':			/* server port */
! 				pgport = optarg;
  				break;
  
  			case 'R':
--- 461,470 ----
  				break;
  
  			case 'p':			/* server port */
! 				if (!in_restore_options)
! 					pgport = optarg;
! 				else
! 					restore_pgport = optarg;
  				break;
  
  			case 'R':
*************** main(int argc, char **argv)
*** 468,474 ****
  				break;
  
  			case 'U':
! 				username = optarg;
  				break;
  
  			case 'v':			/* verbose */
--- 489,498 ----
  				break;
  
  			case 'U':
! 				if (!in_restore_options)
! 					username = optarg;
! 				else
! 					restore_username = optarg;
  				break;
  
  			case 'v':			/* verbose */
*************** main(int argc, char **argv)
*** 511,516 ****
--- 535,545 ----
  				set_section(optarg, &dumpSections);
  				break;
  
+ 			case 6:
+ 				/* XXX check that we're in --restore ? */
+ 				restore_dbname = optarg;
+ 				break;
+ 
  			default:
  				fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
  				exit(1);
*************** main(int argc, char **argv)
*** 602,608 ****
  	}
  
  	/* Parallel backup only in the directory archive format so far */
! 	if (archiveFormat != archDirectory && numWorkers > 1) {
  		write_msg(NULL, "parallel backup only supported by the directory format\n");
  		exit(1);
  	}
--- 631,637 ----
  	}
  
  	/* Parallel backup only in the directory archive format so far */
! 	if (archiveFormat != archDirectory && archiveFormat != archMirror && numWorkers > 1) {
  		write_msg(NULL, "parallel backup only supported by the directory format\n");
  		exit(1);
  	}
*************** main(int argc, char **argv)
*** 764,770 ****
  		sortDumpableObjectsByTypeOid(dobjs, numObjs);
  
  	/* If we do a parallel dump, we want the largest tables to go first */
! 	if (archiveFormat == archDirectory && numWorkers > 1)
  		sortDataAndIndexObjectsBySize(dobjs, numObjs);
  
  	sortDumpableObjects(dobjs, numObjs);
--- 793,799 ----
  		sortDumpableObjectsByTypeOid(dobjs, numObjs);
  
  	/* If we do a parallel dump, we want the largest tables to go first */
! 	if (numWorkers > 1 && (archiveFormat == archDirectory || archiveFormat == archMirror))
  		sortDataAndIndexObjectsBySize(dobjs, numObjs);
  
  	sortDumpableObjects(dobjs, numObjs);
*************** main(int argc, char **argv)
*** 789,795 ****
  	/*
  	 * And finally we can do the actual output.
  	 */
! 	if (plainText)
  	{
  		ropt = NewRestoreOptions();
  		ropt->filename = (char *) filename;
--- 818,824 ----
  	/*
  	 * And finally we can do the actual output.
  	 */
! 	if (plainText || archiveFormat == archMirror)
  	{
  		ropt = NewRestoreOptions();
  		ropt->filename = (char *) filename;
*************** main(int argc, char **argv)
*** 810,823 ****
  
  		ropt->suppressDumpWarnings = true;		/* We've already shown them */
  
  		RestoreArchive(g_fout, ropt);
  	}
  
  	CloseArchive(g_fout);
  
  	PQfinish(g_conn);
  
! 	exit(0);
  }
  
  
--- 839,874 ----
  
  		ropt->suppressDumpWarnings = true;		/* We've already shown them */
  
+ 		if (archiveFormat == archMirror)
+ 		{
+ 			ropt->dbname = restore_dbname;
+ 			ropt->pghost = restore_pghost;
+ 			ropt->pgport = restore_pgport;
+ 			ropt->username = restore_username;
+ 			ropt->promptPassword = prompt_password;
+ 			ropt->useDB = 1;
+ 			ropt->exit_on_error = g_fout->exit_on_error = mirror_exit_on_error;
+ 
+ 			/* Disconnect the dump connection */
+ 			MirrorDecoupleDumpConn((ArchiveHandle *) g_fout);
+ 		}
+ 
  		RestoreArchive(g_fout, ropt);
  	}
  
+ 	/* done, print a summary of ignored errors */
+ 	if (g_fout->n_errors)
+ 		fprintf(stderr, _("WARNING: errors ignored on restore: %d\n"),
+ 				g_fout->n_errors);
+ 
+ 	/* AH may be freed in CloseArchive? */
+ 	exit_code = g_fout->n_errors ? 1 : 0;
+ 
  	CloseArchive(g_fout);
  
  	PQfinish(g_conn);
  
! 	exit(exit_code);
  }
  
  
*************** parseArchiveFormat(const char *format, A
*** 1089,1094 ****
--- 1140,1149 ----
  		 * documented.
  		 */
  		archiveFormat = archFiles;
+ 	else if (pg_strcasecmp(format, "m") == 0)
+ 		archiveFormat = archMirror;
+ 	else if (pg_strcasecmp(format, "mirror") == 0)
+ 		archiveFormat = archMirror;
  	else if (pg_strcasecmp(format, "p") == 0)
  		archiveFormat = archNull;
  	else if (pg_strcasecmp(format, "plain") == 0)
diff --git a/src/bin/pg_dump/pg_restore.c b/src/bin/pg_dump/pg_restore.c
index 965e9f2..7db2c88 100644
*** a/src/bin/pg_dump/pg_restore.c
--- b/src/bin/pg_dump/pg_restore.c
*************** usage(const char *progname)
*** 498,500 ****
--- 498,504 ----
  	printf(_("\nIf no input file name is supplied, then standard input is used.\n\n"));
  	printf(_("Report bugs to <pgsql-bugs@postgresql.org>.\n"));
  }
+ 
+ /* dummy: while implemented as an archive format, it runs entirely in pg_dump */
+ void InitArchiveFmt_Mirror(ArchiveHandle *AH) {}
+