parallel_pg_dump_1.diff
text/x-patch
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| src/backend/port/pipe.c | 0 | 0 |
| src/bin/pg_dump/compress_io.c | 0 | 0 |
| src/bin/pg_dump/dumputils.c | 0 | 0 |
| src/bin/pg_dump/dumputils.h | 0 | 0 |
| src/bin/pg_dump/Makefile | 0 | 0 |
| src/bin/pg_dump/parallel.c | 0 | 0 |
| src/bin/pg_dump/parallel.h | 0 | 0 |
| src/bin/pg_dump/pg_backup_archiver.c | 0 | 0 |
| src/bin/pg_dump/pg_backup_archiver.h | 0 | 0 |
| src/bin/pg_dump/pg_backup_custom.c | 0 | 0 |
| src/bin/pg_dump/pg_backup_directory.c | 0 | 0 |
| src/bin/pg_dump/pg_backup_files.c | 0 | 0 |
| src/bin/pg_dump/pg_backup.h | 0 | 0 |
| src/bin/pg_dump/pg_backup_tar.c | 0 | 0 |
| src/bin/pg_dump/pg_dumpall.c | 0 | 0 |
| src/bin/pg_dump/pg_dump.c | 0 | 0 |
| src/bin/pg_dump/pg_dump.h | 0 | 0 |
| src/bin/pg_dump/pg_dump_sort.c | 0 | 0 |
| src/bin/pg_dump/pg_restore.c | 0 | 0 |
diff --git a/src/backend/port/pipe.c b/src/backend/port/pipe.c
index 357f5ec..4e47d13 100644
*** a/src/backend/port/pipe.c
--- b/src/backend/port/pipe.c
***************
*** 15,23 ****
*-------------------------------------------------------------------------
*/
#include "postgres.h"
- #ifdef WIN32
int
pgpipe(int handles[2])
{
--- 15,42 ----
*-------------------------------------------------------------------------
*/
+ #ifdef WIN32
+
+ /*
+ * This pipe implementation is used in both the server and non-server programs.
+ * In the default case we run within the server and use the standard ereport
+ * error reporting.
+ * If the code runs in a non-server program (like pg_dump), then that program
+ * #defines an error routine and includes this .c file.
+ */
+ #ifndef PGPIPE_EREPORT
#include "postgres.h"
+ #define PGPIPE_EREPORT pgpipe_ereport
+ static void
+ pgpipe_ereport(const char *fmt, ...)
+ {
+ va_list args;
+ va_start(args, fmt);
+ ereport(LOG, (errmsg_internal(fmt, args)));
+ va_end(args);
+ }
+ #endif
int
pgpipe(int handles[2])
{
*************** pgpipe(int handles[2])
*** 29,35 ****
if ((s = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
{
! ereport(LOG, (errmsg_internal("pgpipe could not create socket: %ui", WSAGetLastError())));
return -1;
}
--- 48,54 ----
if ((s = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
{
! PGPIPE_EREPORT("pgpipe could not create socket: %ui", WSAGetLastError());
return -1;
}
*************** pgpipe(int handles[2])
*** 39,76 ****
serv_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
if (bind(s, (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR)
{
! ereport(LOG, (errmsg_internal("pgpipe could not bind: %ui", WSAGetLastError())));
closesocket(s);
return -1;
}
if (listen(s, 1) == SOCKET_ERROR)
{
! ereport(LOG, (errmsg_internal("pgpipe could not listen: %ui", WSAGetLastError())));
closesocket(s);
return -1;
}
if (getsockname(s, (SOCKADDR *) &serv_addr, &len) == SOCKET_ERROR)
{
! ereport(LOG, (errmsg_internal("pgpipe could not getsockname: %ui", WSAGetLastError())));
closesocket(s);
return -1;
}
if ((handles[1] = socket(PF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
{
! ereport(LOG, (errmsg_internal("pgpipe could not create socket 2: %ui", WSAGetLastError())));
closesocket(s);
return -1;
}
if (connect(handles[1], (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR)
{
! ereport(LOG, (errmsg_internal("pgpipe could not connect socket: %ui", WSAGetLastError())));
closesocket(s);
return -1;
}
if ((handles[0] = accept(s, (SOCKADDR *) &serv_addr, &len)) == INVALID_SOCKET)
{
! ereport(LOG, (errmsg_internal("pgpipe could not accept socket: %ui", WSAGetLastError())));
closesocket(handles[1]);
handles[1] = INVALID_SOCKET;
closesocket(s);
--- 58,95 ----
serv_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
if (bind(s, (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR)
{
! PGPIPE_EREPORT("pgpipe could not bind: %ui", WSAGetLastError());
closesocket(s);
return -1;
}
if (listen(s, 1) == SOCKET_ERROR)
{
! PGPIPE_EREPORT("pgpipe could not listen: %ui", WSAGetLastError());
closesocket(s);
return -1;
}
if (getsockname(s, (SOCKADDR *) &serv_addr, &len) == SOCKET_ERROR)
{
! PGPIPE_EREPORT("pgpipe could not getsockname: %ui", WSAGetLastError());
closesocket(s);
return -1;
}
if ((handles[1] = socket(PF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
{
! PGPIPE_EREPORT("pgpipe could not create socket 2: %ui", WSAGetLastError());
closesocket(s);
return -1;
}
if (connect(handles[1], (SOCKADDR *) &serv_addr, len) == SOCKET_ERROR)
{
! PGPIPE_EREPORT("pgpipe could not connect socket: %ui", WSAGetLastError());
closesocket(s);
return -1;
}
if ((handles[0] = accept(s, (SOCKADDR *) &serv_addr, &len)) == INVALID_SOCKET)
{
! PGPIPE_EREPORT("pgpipe could not accept socket: %ui", WSAGetLastError());
closesocket(handles[1]);
handles[1] = INVALID_SOCKET;
closesocket(s);
diff --git a/src/bin/pg_dump/Makefile b/src/bin/pg_dump/Makefile
index 033fb1e..00501de 100644
*** a/src/bin/pg_dump/Makefile
--- b/src/bin/pg_dump/Makefile
*************** override CPPFLAGS := -I$(libpq_srcdir) $
*** 20,26 ****
OBJS= pg_backup_archiver.o pg_backup_db.o pg_backup_custom.o \
pg_backup_files.o pg_backup_null.o pg_backup_tar.o \
! pg_backup_directory.o dumpmem.o dumputils.o compress_io.o $(WIN32RES)
KEYWRDOBJS = keywords.o kwlookup.o
--- 20,27 ----
OBJS= pg_backup_archiver.o pg_backup_db.o pg_backup_custom.o \
pg_backup_files.o pg_backup_null.o pg_backup_tar.o \
! pg_backup_directory.o dumpmem.o dumputils.o compress_io.o \
! parallel.o $(WIN32RES)
KEYWRDOBJS = keywords.o kwlookup.o
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index d48b276..3bc224b 100644
*** a/src/bin/pg_dump/compress_io.c
--- b/src/bin/pg_dump/compress_io.c
***************
*** 54,59 ****
--- 54,60 ----
#include "compress_io.h"
#include "dumpmem.h"
+ #include "parallel.h"
/*----------------------
* Compressor API
*************** size_t
*** 181,186 ****
--- 182,190 ----
WriteDataToArchive(ArchiveHandle *AH, CompressorState *cs,
const void *data, size_t dLen)
{
+ /* Are we aborting? */
+ checkAborting(AH);
+
switch (cs->comprAlg)
{
case COMPR_ALG_LIBZ:
*************** ReadDataFromArchiveZlib(ArchiveHandle *A
*** 350,355 ****
--- 354,362 ----
/* no minimal chunk size for zlib */
while ((cnt = readF(AH, &buf, &buflen)))
{
+ /* Are we aborting? */
+ checkAborting(AH);
+
zp->next_in = (void *) buf;
zp->avail_in = cnt;
*************** ReadDataFromArchiveNone(ArchiveHandle *A
*** 410,415 ****
--- 417,425 ----
while ((cnt = readF(AH, &buf, &buflen)))
{
+ /* Are we aborting? */
+ checkAborting(AH);
+
ahwrite(buf, 1, cnt, AH);
}
diff --git a/src/bin/pg_dump/dumputils.c b/src/bin/pg_dump/dumputils.c
index 3493e39..352daba 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 "pg_backup.h"
*************** static bool parseAclItem(const char *ite
*** 36,41 ****
--- 37,43 ----
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)
*** 55,69 ****
}
/*
! * 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
--- 57,67 ----
}
/*
! * 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)
*** 72,80 ****
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 */
--- 70,75 ----
*************** fmtId(const char *rawid)
*** 101,109 ****
#else
s_id_return = id_return;
#endif
-
}
/*
* These checks need to match the identifier production in scan.l. Don't
* use islower() etc.
--- 96,120 ----
#else
s_id_return = id_return;
#endif
}
+ 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)
*** 171,176 ****
--- 182,226 ----
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.
+ */
+ const char *
+ fmtQualifiedId(const char *schema, const char *id, int remoteVersion)
+ {
+ PQExpBuffer id_return;
+ /*
+ * We're using the same PQExpBuffer as fmtId(), that's why we first get all
+ * the values from fmtId and then return them appended in the PQExpBuffer.
+ * The reason we still use the PQExpBuffer to return the string is just for
+ * ease of use in the caller, that doesn't have to free the string
+ * explicitly.
+ */
+ char *schemaBuf, *idBuf;
+
+ /* Suppress schema name if fetching from pre-7.3 DB */
+ if (remoteVersion >= 70300 && schema && *schema)
+ {
+ schemaBuf = pg_strdup(fmtId(schema));
+ } else
+ schemaBuf = pg_strdup("");
+
+ idBuf = pg_strdup(fmtId(id));
+
+ /* this will reset the PQExpBuffer */
+ id_return = getThreadLocalPQExpBuffer();
+ appendPQExpBuffer(id_return, "%s%s%s",
+ schemaBuf,
+ strlen(schemaBuf) > 0 ? "." : "",
+ idBuf);
+ free(schemaBuf);
+ free(idBuf);
+
+ return id_return->data;
+ }
/*
* Convert a string value to an SQL string literal and append it to
diff --git a/src/bin/pg_dump/dumputils.h b/src/bin/pg_dump/dumputils.h
index b4cf730..060c95d 100644
*** a/src/bin/pg_dump/dumputils.h
--- b/src/bin/pg_dump/dumputils.h
*************** extern const char *progname;
*** 24,29 ****
--- 24,30 ----
extern void init_parallel_dump_utils(void);
extern const char *fmtId(const char *identifier);
+ extern const char *fmtQualifiedId(const char *schema, const char *id, int remoteVersion);
extern void appendStringLiteral(PQExpBuffer buf, const char *str,
int encoding, bool std_strings);
extern void appendStringLiteralConn(PQExpBuffer buf, const char *str,
diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c
index ...bcde24c .
*** a/src/bin/pg_dump/parallel.c
--- b/src/bin/pg_dump/parallel.c
***************
*** 0 ****
--- 1,1289 ----
+ /*-------------------------------------------------------------------------
+ *
+ * parallel.c
+ *
+ * Parallel support for the pg_dump archiver
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * The author is not responsible for loss or damages that may
+ * result from its use.
+ *
+ * IDENTIFICATION
+ * src/bin/pg_dump/parallel.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+ #include "pg_backup_db.h"
+
+ #include "dumpmem.h"
+ #include "dumputils.h"
+ #include "parallel.h"
+
+ #ifndef WIN32
+ #include <sys/types.h>
+ #include <sys/wait.h>
+ #include "signal.h"
+ #include <unistd.h>
+ #include <fcntl.h>
+ #endif
+
+ #define PIPE_READ 0
+ #define PIPE_WRITE 1
+ #define SHUTDOWN_GRACE_PERIOD (500)
+
+ /* file-scope variables */
+ #ifdef WIN32
+ static unsigned int tMasterThreadId = 0;
+ static HANDLE termEvent = INVALID_HANDLE_VALUE;
+ #else
+ static volatile sig_atomic_t wantAbort = 0;
+ static bool aborting = false;
+ #endif
+
+ /*
+ * The parallel error handler is called for any die_horribly() in a child or master process.
+ * It then takes control over shutting down the rest of the gang.
+ */
+ void (*volatile vparallel_error_handler)(ArchiveHandle *AH, const char *modulename,
+ const char *fmt, va_list ap)
+ __attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 0))) = NULL;
+
+ /* The actual implementation of the error handler function */
+ static void vparallel_error_handler_imp(ArchiveHandle *AH, const char *modulename,
+ const char *fmt, va_list ap)
+ __attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 0)));
+
+ static const char *modulename = gettext_noop("parallel archiver");
+
+ static int ShutdownConnection(PGconn **conn);
+
+ static void WaitForTerminatingWorkers(ArchiveHandle *AH, ParallelState *pstate);
+ static void ShutdownWorkersHard(ArchiveHandle *AH, ParallelState *pstate);
+ static void ShutdownWorkersSoft(ArchiveHandle *AH, ParallelState *pstate, bool do_wait);
+ static void PrintStatus(ParallelState *pstate);
+ static bool HasEveryWorkerTerminated(ParallelState *pstate);
+
+ static void lockTableNoWait(ArchiveHandle *AH, TocEntry *te);
+ static void WaitForCommands(ArchiveHandle *AH, int pipefd[2]);
+ static char *getMessageFromMaster(ArchiveHandle *AH, int pipefd[2]);
+ static void sendMessageToMaster(ArchiveHandle *AH, int pipefd[2], const char *str);
+ static char *getMessageFromWorker(ArchiveHandle *AH, ParallelState *pstate,
+ bool do_wait, int *worker);
+ static void sendMessageToWorker(ArchiveHandle *AH, ParallelState *pstate,
+ int worker, const char *str);
+ static char *readMessageFromPipe(int fd, bool do_wait);
+
+ static void SetupWorker(ArchiveHandle *AH, int pipefd[2], int worker,
+ RestoreOptions *ropt);
+
+ #define messageStartsWith(msg, prefix) \
+ (strncmp(msg, prefix, strlen(prefix)) == 0)
+ #define messageEquals(msg, pattern) \
+ (strcmp(msg, pattern) == 0)
+
+ /* architecture dependent #defines */
+ #ifdef WIN32
+ /* WIN32 */
+ /* pgpipe implemented in src/backend/port/pipe.c */
+ #define setnonblocking(fd) \
+ do { u_long mode = 1; \
+ ioctlsocket((fd), FIONBIO, &mode); \
+ } while(0);
+ #define setblocking(fd) \
+ do { u_long mode = 0; \
+ ioctlsocket((fd), FIONBIO, &mode); \
+ } while(0);
+ #else /* UNIX */
+ #define setnonblocking(fd) \
+ do { long flags = (long) fcntl((fd), F_GETFL, 0); \
+ fcntl(fd, F_SETFL, flags | O_NONBLOCK); \
+ } while(0);
+ #define setblocking(fd) \
+ do { long flags = (long) fcntl((fd), F_GETFL, 0); \
+ fcntl(fd, F_SETFL, flags & ~O_NONBLOCK); \
+ } while(0);
+ #endif
+
+ #ifdef WIN32
+ /*
+ * On Windows, source in the pgpipe implementation from the backend and provide
+ * an own error reporting routine, the backend usually uses ereport() for that.
+ */
+ static void pgdump_pgpipe_ereport(const char* fmt, ...);
+ #define PGPIPE_EREPORT pgdump_pgpipe_ereport
+ #include "../../backend/port/pipe.c"
+ static void
+ pgdump_pgpipe_ereport(const char* fmt, ...)
+ {
+ va_list args;
+ va_start(args, fmt);
+ vwrite_msg("pgpipe", fmt, args);
+ va_end(args);
+ }
+ #endif
+
+ static int
+ #ifdef WIN32
+ GetSlotOfThread(ParallelState *pstate, unsigned int threadId)
+ #else
+ GetSlotOfProcess(ParallelState *pstate, pid_t pid)
+ #endif
+ {
+ int i;
+
+ for (i = 0; i < pstate->numWorkers; i++)
+ #ifdef WIN32
+ if (pstate->parallelSlot[i].threadId == threadId)
+ #else
+ if (pstate->parallelSlot[i].pid == pid)
+ #endif
+ return i;
+
+ Assert(false);
+ return NO_SLOT;
+ }
+
+ enum escrow_action { GET, SET };
+ static void
+ parallel_error_handler_escrow_data(enum escrow_action act, ParallelState *pstate)
+ {
+ static ParallelState *s_pstate = NULL;
+
+ if (act == SET)
+ s_pstate = pstate;
+ else
+ *pstate = *s_pstate;
+ }
+
+ static void
+ vparallel_error_handler_imp(ArchiveHandle *AH,
+ const char *modulename,
+ const char *fmt, va_list ap)
+ {
+ ParallelState pstate;
+ char buf[512];
+ int pipefd[2];
+ int i;
+
+ if (AH->is_clone)
+ {
+ /* we are the child, get the message out to the parent */
+ parallel_error_handler_escrow_data(GET, &pstate);
+ #ifdef WIN32
+ i = GetSlotOfThread(&pstate, GetCurrentThreadId());
+ #else
+ i = GetSlotOfProcess(&pstate, getpid());
+ #endif
+ if (pstate.parallelSlot[i].inErrorHandling)
+ return;
+
+ pstate.parallelSlot[i].inErrorHandling = true;
+
+ pipefd[PIPE_READ] = pstate.parallelSlot[i].pipeRevRead;
+ pipefd[PIPE_WRITE] = pstate.parallelSlot[i].pipeRevWrite;
+
+ strcpy(buf, "ERROR ");
+ vsnprintf(buf + strlen("ERROR "), sizeof(buf) - strlen("ERROR "), fmt, ap);
+
+ sendMessageToMaster(AH, pipefd, buf);
+ if (AH->connection)
+ ShutdownConnection(&(AH->connection));
+ #ifdef WIN32
+ ExitThread(1);
+ #else
+ exit(1);
+ #endif
+ }
+ else
+ {
+ #ifndef WIN32
+ /*
+ * We are the parent. We need the handling variable to see if we're
+ * already handling an error.
+ */
+ if (aborting)
+ return;
+ aborting = 1;
+
+ signal(SIGPIPE, SIG_IGN);
+ #endif
+ /*
+ * Note that technically we're using a new pstate here, the old one
+ * is copied over and then the old one isn't updated anymore. Only
+ * the new one is, which is okay because control will never return
+ * from this function.
+ */
+ parallel_error_handler_escrow_data(GET, &pstate);
+ ShutdownWorkersHard(AH, &pstate);
+ /* Terminate own connection */
+ if (AH->connection)
+ ShutdownConnection(&(AH->connection));
+ vwrite_msg(NULL, fmt, ap);
+ exit(1);
+ }
+ Assert(false);
+ }
+
+ /*
+ * If we have one worker that terminates for some reason, we'd like the other
+ * threads to terminate as well (and not finish with their 70 GB table dump
+ * first...). Now in UNIX we can just kill these processes, and let the signal
+ * handler set wantAbort to 1 or more. In Windows we set a termEvent and this
+ * serves as the signal for everyone to terminate.
+ */
+ void
+ checkAborting(ArchiveHandle *AH)
+ {
+ #ifdef WIN32
+ if (WaitForSingleObject(termEvent, 0) == WAIT_OBJECT_0)
+ #else
+ if (wantAbort)
+ #endif
+ /*
+ * Terminate, this error will actually never show up somewhere
+ * because if termEvent/wantAbort is set, then we are already in the
+ * process of going down and already have a reason why we're
+ * terminating.
+ */
+ die_horribly(AH, modulename, "worker is terminating");
+ }
+
+ /*
+ * A select loop that repeats calling select until a descriptor in the read set
+ * becomes readable. On Windows we have to check for the termination event from
+ * time to time, on Unix we can just block forever.
+ */
+ #ifdef WIN32
+ static int
+ select_loop(int maxFd, fd_set *workerset)
+ {
+ int i;
+ fd_set saveSet = *workerset;
+
+ /* should always be the master */
+ Assert(tMasterThreadId == GetCurrentThreadId());
+
+ for (;;)
+ {
+ /*
+ * sleep a quarter of a second before checking if we should
+ * terminate.
+ */
+ struct timeval tv = { 0, 250000 };
+ *workerset = saveSet;
+ i = select(maxFd + 1, workerset, NULL, NULL, &tv);
+
+ if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR)
+ continue;
+ if (i)
+ break;
+ }
+
+ return i;
+ }
+ #else /* UNIX */
+ static int
+ select_loop(int maxFd, fd_set *workerset)
+ {
+ int i;
+
+ fd_set saveSet = *workerset;
+ for (;;)
+ {
+ *workerset = saveSet;
+ i = select(maxFd + 1, workerset, NULL, NULL, NULL);
+ Assert(i != 0);
+ if (wantAbort && !aborting) {
+ return NO_SLOT;
+ }
+ if (i < 0 && errno == EINTR)
+ continue;
+ break;
+ }
+
+ return i;
+ }
+ #endif
+
+ /*
+ * Shut down any remaining workers, this has an implicit do_wait == true
+ */
+ static void
+ ShutdownWorkersHard(ArchiveHandle *AH, ParallelState *pstate)
+ {
+ #ifdef WIN32
+ /* The workers monitor this event via checkAborting(). */
+ SetEvent(termEvent);
+ #endif
+ /*
+ * The fastest way we can make them terminate is when they are listening
+ * for new commands and we just tell them to terminate.
+ */
+ ShutdownWorkersSoft(AH, pstate, false);
+
+ #ifndef WIN32
+ {
+ int i;
+ for (i = 0; i < pstate->numWorkers; i++)
+ kill(pstate->parallelSlot[i].pid, SIGTERM);
+
+ /* Reset our signal handler, if we get signaled again, terminate normally */
+ signal(SIGINT, SIG_DFL);
+ signal(SIGTERM, SIG_DFL);
+ signal(SIGQUIT, SIG_DFL);
+ }
+ #endif
+
+ WaitForTerminatingWorkers(AH, pstate);
+ }
+
+ static void
+ WaitForTerminatingWorkers(ArchiveHandle *AH, ParallelState *pstate)
+ {
+ while (!HasEveryWorkerTerminated(pstate))
+ {
+ int worker;
+ char *msg;
+
+ PrintStatus(pstate);
+
+ msg = getMessageFromWorker(AH, pstate, true, &worker);
+ if (!msg || messageStartsWith(msg, "ERROR "))
+ pstate->parallelSlot[worker].workerStatus = WRKR_TERMINATED;
+ if (msg)
+ free(msg);
+ }
+ Assert(HasEveryWorkerTerminated(pstate));
+ }
+
+ #ifndef WIN32
+ /* Signal handling (UNIX only) */
+ static void
+ sigTermHandler(int signum)
+ {
+ wantAbort++;
+ }
+ #endif
+
+ /*
+ * This function is called by both UNIX and Windows variants to set up a
+ * worker process.
+ */
+ static void
+ SetupWorker(ArchiveHandle *AH, int pipefd[2], int worker,
+ RestoreOptions *ropt)
+ {
+ /*
+ * In dump mode (pg_dump) this calls _SetupWorker() as defined in
+ * pg_dump.c, while in restore mode (pg_restore) it calls _SetupWorker() as
+ * defined in pg_restore.c.
+ *
+ * We get the raw connection only for the reason that we can close it
+ * properly when we shut down. This happens only that way when it is
+ * brought down because of an error.
+ */
+ _SetupWorker((Archive *) AH, ropt);
+
+ Assert(AH->connection != NULL);
+
+ WaitForCommands(AH, pipefd);
+
+ closesocket(pipefd[PIPE_READ]);
+ closesocket(pipefd[PIPE_WRITE]);
+ }
+
+ #ifdef WIN32
+ /*
+ * On Windows the _beginthreadex() function allows us to pass one parameter.
+ * Since we need to pass a few values however, we define a structure here
+ * and then pass a pointer to such a structure in _beginthreadex().
+ */
+ typedef struct {
+ ArchiveHandle *AH;
+ RestoreOptions *ropt;
+ int worker;
+ int pipeRead;
+ int pipeWrite;
+ } WorkerInfo;
+
+ static unsigned __stdcall
+ init_spawned_worker_win32(WorkerInfo *wi)
+ {
+ ArchiveHandle *AH;
+ int pipefd[2] = { wi->pipeRead, wi->pipeWrite };
+ int worker = wi->worker;
+ RestoreOptions *ropt = wi->ropt;
+
+ AH = CloneArchive(wi->AH);
+
+ free(wi);
+ SetupWorker(AH, pipefd, worker, ropt);
+
+ DeCloneArchive(AH);
+ _endthreadex(0);
+ return 0;
+ }
+ #endif
+
+ /*
+ * This function starts the parallel dump or restore by spawning off the worker
+ * processes in both Unix and Windows. For Windows, it creates a number of
+ * threads while it does a fork() on Unix.
+ */
+ ParallelState *
+ ParallelBackupStart(ArchiveHandle *AH, RestoreOptions *ropt)
+ {
+ ParallelState *pstate;
+ int i;
+ const size_t slotSize = AH->public.numWorkers * sizeof(ParallelSlot);
+
+ Assert(AH->public.numWorkers > 0);
+
+ /* Ensure stdio state is quiesced before forking */
+ fflush(NULL);
+
+ pstate = (ParallelState *) pg_malloc(sizeof(ParallelState));
+
+ pstate->numWorkers = AH->public.numWorkers;
+ pstate->parallelSlot = NULL;
+
+ if (AH->public.numWorkers == 1)
+ return pstate;
+
+ pstate->parallelSlot = (ParallelSlot *) pg_malloc(slotSize);
+ memset((void *) pstate->parallelSlot, 0, slotSize);
+
+ parallel_error_handler_escrow_data(SET, pstate);
+ vparallel_error_handler = vparallel_error_handler_imp;
+
+ #ifdef WIN32
+ tMasterThreadId = GetCurrentThreadId();
+ termEvent = CreateEvent(NULL, true, false, "Terminate");
+ #else
+ signal(SIGTERM, sigTermHandler);
+ signal(SIGINT, sigTermHandler);
+ signal(SIGQUIT, sigTermHandler);
+ #endif
+
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ #ifdef WIN32
+ WorkerInfo *wi;
+ uintptr_t handle;
+ #else
+ pid_t pid;
+ #endif
+ int pipeMW[2], pipeWM[2];
+
+ if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
+ die_horribly(AH, modulename, "Cannot create communication channels: %s",
+ strerror(errno));
+
+ pstate->parallelSlot[i].workerStatus = WRKR_IDLE;
+ #ifdef WIN32
+ /* Allocate a new structure for every worker */
+ wi = (WorkerInfo *) pg_malloc(sizeof(WorkerInfo));
+
+ wi->ropt = ropt;
+ wi->worker = i;
+ wi->AH = AH;
+ wi->pipeRead = pstate->parallelSlot[i].pipeRevRead = pipeMW[PIPE_READ];
+ wi->pipeWrite = pstate->parallelSlot[i].pipeRevWrite = pipeWM[PIPE_WRITE];
+
+ handle = _beginthreadex(NULL, 0, &init_spawned_worker_win32,
+ wi, 0, &(pstate->parallelSlot[i].threadId));
+ pstate->parallelSlot[i].hThread = handle;
+ #else
+ pid = fork();
+ if (pid == 0)
+ {
+ /* we are the worker */
+ int j;
+ int pipefd[2] = { pipeMW[PIPE_READ], pipeWM[PIPE_WRITE] };
+
+ /*
+ * Store the fds for the reverse communication in pstate. Actually
+ * we only use this in case of an error and don't use pstate
+ * otherwise in the worker process. On Windows we write to the
+ * global pstate, in Unix we write to our process-local copy but
+ * that's also where we'd retrieve this information back from.
+ */
+ pstate->parallelSlot[i].pipeRevRead = pipefd[PIPE_READ];
+ pstate->parallelSlot[i].pipeRevWrite = pipefd[PIPE_WRITE];
+ pstate->parallelSlot[i].pid = getpid();
+
+ /*
+ * Call CloneArchive on Unix as well even though technically we
+ * don't need to because fork() gives us a copy in our own address space
+ * already. But CloneArchive resets the state information, sets is_clone
+ * and also clones the database connection (for parallel dump)
+ * which all seems kinda helpful.
+ */
+ AH = CloneArchive(AH);
+
+ #ifdef HAVE_SETSID
+ /*
+ * If we can, we try to make each process the leader of its own
+ * process group. The reason is that if you hit Ctrl-C and they are
+ * all in the same process group, any termination sequence is
+ * possible, because every process will receive the signal. What
+ * often happens is that a worker receives the signal, terminates
+ * and the master detects that one of the workers had a problem,
+ * even before acting on its own signal. That's still okay because
+ * everyone still terminates but it looks a bit weird.
+ *
+ * With setsid() however, a Ctrl-C is only sent to the master and
+ * he can then cascade it to the worker processes.
+ */
+ setsid();
+ #endif
+
+ closesocket(pipeWM[PIPE_READ]); /* close read end of Worker -> Master */
+ closesocket(pipeMW[PIPE_WRITE]); /* close write end of Master -> Worker */
+
+ /*
+ * Close all inherited fds for communication of the master with
+ * the other workers.
+ */
+ for (j = 0; j < i; j++)
+ {
+ closesocket(pstate->parallelSlot[j].pipeRead);
+ closesocket(pstate->parallelSlot[j].pipeWrite);
+ }
+
+ SetupWorker(AH, pipefd, i, ropt);
+
+ exit(0);
+ }
+ else if (pid < 0)
+ /* fork failed */
+ die_horribly(AH, modulename,
+ "could not create worker process: %s\n",
+ strerror(errno));
+
+ /* we are the Master, pid > 0 here */
+ Assert(pid > 0);
+ closesocket(pipeMW[PIPE_READ]); /* close read end of Master -> Worker */
+ closesocket(pipeWM[PIPE_WRITE]); /* close write end of Worker -> Master */
+
+ pstate->parallelSlot[i].pid = pid;
+ #endif
+
+ pstate->parallelSlot[i].pipeRead = pipeWM[PIPE_READ];
+ pstate->parallelSlot[i].pipeWrite = pipeMW[PIPE_WRITE];
+
+ pstate->parallelSlot[i].args = (ParallelArgs *) pg_malloc(sizeof(ParallelArgs));
+ pstate->parallelSlot[i].args->AH = AH;
+ pstate->parallelSlot[i].args->te = NULL;
+ pstate->parallelSlot[i].workerStatus = WRKR_IDLE;
+ }
+
+ return pstate;
+ }
+
+ /*
+ * Tell all of our workers to terminate.
+ *
+ * Pretty straightforward routine, first we tell everyone to terminate, then we
+ * listen to the workers' replies and finally close the sockets that we have
+ * used for communication.
+ */
+ void
+ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate)
+ {
+ int i;
+
+ if (pstate->numWorkers == 1)
+ return;
+
+ PrintStatus(pstate);
+ Assert(IsEveryWorkerIdle(pstate));
+
+ /* no hard shutdown, let workers exit by themselves and wait for them */
+ ShutdownWorkersSoft(AH, pstate, true);
+
+ PrintStatus(pstate);
+
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ closesocket(pstate->parallelSlot[i].pipeRead);
+ closesocket(pstate->parallelSlot[i].pipeWrite);
+ }
+
+ vparallel_error_handler = NULL;
+
+ free(pstate->parallelSlot);
+ free(pstate);
+ }
+
+
+ /*
+ * The sequence is the following (for dump, similar for restore):
+ *
+ * Master Worker
+ *
+ * enters WaitForCommands()
+ * DispatchJobForTocEntry(...te...)
+ *
+ * [ Worker is IDLE ]
+ *
+ * arg = (MasterStartParallelItemPtr)()
+ * send: DUMP arg
+ * receive: DUMP arg
+ * str = (WorkerJobDumpPtr)(arg)
+ * [ Worker is WORKING ] ... gets te from arg ...
+ * ... dump te ...
+ * send: OK DUMP info
+ *
+ * In ListenToWorkers():
+ *
+ * [ Worker is FINISHED ]
+ * receive: OK DUMP info
+ * status = (MasterEndParallelItemPtr)(info)
+ *
+ * In ReapWorkerStatus(&ptr):
+ * *ptr = status;
+ * [ Worker is IDLE ]
+ */
+ void
+ DispatchJobForTocEntry(ArchiveHandle *AH, ParallelState *pstate, TocEntry *te,
+ T_Action act)
+ {
+ int worker;
+ char *arg;
+
+ /* our caller makes sure that at least one worker is idle */
+ Assert(GetIdleWorker(pstate) != NO_SLOT);
+ worker = GetIdleWorker(pstate);
+ Assert(worker != NO_SLOT);
+
+ arg = (AH->MasterStartParallelItemPtr)(AH, te, act);
+
+ sendMessageToWorker(AH, pstate, worker, arg);
+
+ pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
+ pstate->parallelSlot[worker].args->te = te;
+ PrintStatus(pstate);
+ }
+
+ static void
+ PrintStatus(ParallelState *pstate)
+ {
+ int i;
+ printf("------Status------\n");
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ printf("Status of worker %d: ", i);
+ switch (pstate->parallelSlot[i].workerStatus)
+ {
+ case WRKR_IDLE:
+ printf("IDLE");
+ break;
+ case WRKR_WORKING:
+ printf("WORKING");
+ break;
+ case WRKR_FINISHED:
+ printf("FINISHED");
+ break;
+ case WRKR_TERMINATED:
+ printf("TERMINATED");
+ break;
+ }
+ printf("\n");
+ }
+ printf("------------\n");
+ }
+
+
+ /*
+ * Find the first free parallel slot (if any).
+ */
+ int
+ GetIdleWorker(ParallelState *pstate)
+ {
+ int i;
+ for (i = 0; i < pstate->numWorkers; i++)
+ if (pstate->parallelSlot[i].workerStatus == WRKR_IDLE)
+ return i;
+ return NO_SLOT;
+ }
+
+ /*
+ * Return true iff every worker process is in the WRKR_TERMINATED state.
+ */
+ static bool
+ HasEveryWorkerTerminated(ParallelState *pstate)
+ {
+ int i;
+ for (i = 0; i < pstate->numWorkers; i++)
+ if (pstate->parallelSlot[i].workerStatus != WRKR_TERMINATED)
+ return false;
+ return true;
+ }
+
+ /*
+ * Return true iff every worker is in the WRKR_IDLE state.
+ */
+ bool
+ IsEveryWorkerIdle(ParallelState *pstate)
+ {
+ int i;
+ for (i = 0; i < pstate->numWorkers; i++)
+ if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
+ return false;
+ return true;
+ }
+
+ /*
+ * Performs a soft shutdown and optionally waits for every worker to terminate.
+ * A soft shutdown sends a "TERMINATE" message to every worker only.
+ */
+ static void
+ ShutdownWorkersSoft(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
+ {
+ int i;
+
+ /* soft shutdown */
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ if (pstate->parallelSlot[i].workerStatus != WRKR_TERMINATED)
+ {
+ sendMessageToWorker(AH, pstate, i, "TERMINATE");
+ pstate->parallelSlot[i].workerStatus = WRKR_WORKING;
+ }
+ }
+
+ if (!do_wait)
+ return;
+
+ WaitForTerminatingWorkers(AH, pstate);
+ }
+
+ /*
+ * This routine does some effort to gracefully shut down the database
+ * connection, but not too much, since the parent is waiting for the workers to
+ * terminate. The cancellation of the database connection is done in an
+ * asynchronous way, so we need to wait a bit after sending PQcancel().
+ * Calling PQcancel() first and then and PQfinish() immediately afterwards
+ * would still cancel an active connection because most likely the PQfinish()
+ * has not yet been processed.
+ *
+ * On Windows, when the master process terminates the childrens' database
+ * connections it forks off new threads that do nothing else than close the
+ * connection. These threads only live as long as they are in this function.
+ * And since a thread needs to return a value this function needs to as well.
+ * Hence this function returns an (unsigned) int.
+ */
+ static int
+ ShutdownConnection(PGconn **conn)
+ {
+ PGcancel *cancel;
+ char errbuf[1];
+ int i;
+
+ Assert(conn != NULL);
+ Assert(*conn != NULL);
+
+ if ((cancel = PQgetCancel(*conn)))
+ {
+ PQcancel(cancel, errbuf, sizeof(errbuf));
+ PQfreeCancel(cancel);
+ }
+
+ /* give the server a little while */
+ for (i = 0; i < 10; i++)
+ {
+ PQconsumeInput(*conn);
+ if (!PQisBusy(*conn))
+ break;
+ pg_usleep((SHUTDOWN_GRACE_PERIOD / 10) * 1000);
+ }
+
+ PQfinish(*conn);
+ *conn = NULL;
+ return 0;
+ }
+
+ /*
+ * One danger of the parallel backup is a possible deadlock:
+ *
+ * 1) Master dumps the schema and locks all tables in ACCESS SHARE mode.
+ * 2) Another process requests an ACCESS EXCLUSIVE lock (which is not granted
+ * because the master holds a conflicting ACCESS SHARE lock).
+ * 3) The worker process also requests an ACCESS SHARE lock to read the table.
+ * The worker's not granted that lock but is enqueued behind the ACCESS
+ * EXCLUSIVE lock request.
+ *
+ * Now what we do here is to just request a lock in ACCESS SHARE but with
+ * NOWAIT in the worker prior to touching the table. If we don't get the lock,
+ * then we know that somebody else has requested an ACCESS EXCLUSIVE lock and
+ * are good to just fail the whole backup because we have detected a deadlock.
+ */
+ static void
+ lockTableNoWait(ArchiveHandle *AH, TocEntry *te)
+ {
+ const char *qualId;
+ PQExpBuffer query = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(AH->format == archDirectory);
+ Assert(strcmp(te->desc, "BLOBS") != 0);
+
+ /*
+ * We are only locking tables and thus we can peek at the DROP command
+ * which contains the fully qualified name.
+ *
+ * Additionally, strlen("DROP") == strlen("LOCK").
+ */
+ appendPQExpBuffer(query, "SELECT pg_namespace.nspname,"
+ " pg_class.relname "
+ " FROM pg_class "
+ " JOIN pg_namespace on pg_namespace.oid = relnamespace "
+ " WHERE pg_class.oid = %d", te->catalogId.oid);
+
+ res = PQexec(AH->connection, query->data);
+
+ if (!res || PQresultStatus(res) != PGRES_TUPLES_OK)
+ die_horribly(AH, modulename, "could not get relation name for oid %d: %s",
+ te->catalogId.oid, PQerrorMessage(AH->connection));
+
+ resetPQExpBuffer(query);
+
+ qualId = fmtQualifiedId(PQgetvalue(res, 0, 0),
+ PQgetvalue(res, 0, 1),
+ AH->public.remoteVersion);
+
+ appendPQExpBuffer(query, "LOCK TABLE %s IN ACCESS SHARE MODE NOWAIT", qualId);
+ PQclear(res);
+
+ res = PQexec(AH->connection, query->data);
+
+ if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
+ die_horribly(AH, modulename, "could not obtain lock on relation \"%s\". This "
+ "usually means that someone requested an ACCESS EXCLUSIVE lock "
+ "on the table after the pg_dump parent process has gotten the "
+ "initial ACCESS SHARE lock on the table.", qualId);
+
+ PQclear(res);
+ destroyPQExpBuffer(query);
+ }
+
+ /*
+ * That's the main routine for the worker.
+ * When it starts up it enters this routine and waits for commands from the
+ * master process. After having processed a command it comes back to here to
+ * wait for the next command. Finally it will receive a TERMINATE command and
+ * exit.
+ */
+ static void
+ WaitForCommands(ArchiveHandle *AH, int pipefd[2])
+ {
+ char *command;
+ DumpId dumpId;
+ int nBytes;
+ char *str = NULL;
+ TocEntry *te;
+
+ for(;;)
+ {
+ command = getMessageFromMaster(AH, pipefd);
+
+ if (messageStartsWith(command, "DUMP "))
+ {
+ Assert(AH->format == archDirectory);
+ sscanf(command + strlen("DUMP "), "%d%n", &dumpId, &nBytes);
+ Assert(nBytes == strlen(command) - strlen("DUMP "));
+
+ te = getTocEntryByDumpId(AH, dumpId);
+ Assert(te != NULL);
+
+ /*
+ * Lock the table but with NOWAIT. Note that the parent is already
+ * holding a lock. If we cannot acquire another ACCESS SHARE MODE
+ * lock, then somebody else has requested an exclusive lock in the
+ * meantime. lockTableNoWait dies in this case to prevent a
+ * deadlock.
+ */
+ if (strcmp(te->desc, "BLOBS") != 0)
+ lockTableNoWait(AH, te);
+
+ /*
+ * The message we return here has been pg_malloc()ed and we are
+ * responsible for free()ing it.
+ */
+ str = (AH->WorkerJobDumpPtr)(AH, te);
+ Assert(AH->connection != NULL);
+ sendMessageToMaster(AH, pipefd, str);
+ free(str);
+ }
+ else if (messageStartsWith(command, "RESTORE "))
+ {
+ Assert(AH->format == archDirectory || AH->format == archCustom);
+ Assert(AH->connection != NULL);
+
+ sscanf(command + strlen("RESTORE "), "%d%n", &dumpId, &nBytes);
+ Assert(nBytes == strlen(command) - strlen("RESTORE "));
+
+ te = getTocEntryByDumpId(AH, dumpId);
+ Assert(te != NULL);
+ /*
+ * The message we return here has been pg_malloc()ed and we are
+ * responsible for free()ing it.
+ */
+ str = (AH->WorkerJobRestorePtr)(AH, te);
+ Assert(AH->connection != NULL);
+ sendMessageToMaster(AH, pipefd, str);
+ free(str);
+ }
+ else if (messageEquals(command, "TERMINATE"))
+ {
+ PQfinish(AH->connection);
+ AH->connection = NULL;
+ return;
+ }
+ else
+ {
+ die_horribly(AH, modulename,
+ "Unknown command on communication channel: %s", command);
+ }
+ }
+ }
+
+ /*
+ * Note the status change:
+ *
+ * DispatchJobForTocEntry WRKR_IDLE -> WRKR_WORKING
+ * ListenToWorkers WRKR_WORKING -> WRKR_FINISHED / WRKR_TERMINATED
+ * ReapWorkerStatus WRKR_FINISHED -> WRKR_IDLE
+ *
+ * Just calling ReapWorkerStatus() when all workers are working might or might
+ * not give you an idle worker because you need to call ListenToWorkers() in
+ * between and only thereafter ReapWorkerStatus(). This is necessary in order to
+ * get and deal with the status (=result) of the worker's execution.
+ */
+ void
+ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
+ {
+ int worker;
+ char *msg;
+
+ msg = getMessageFromWorker(AH, pstate, do_wait, &worker);
+
+ if (!msg)
+ {
+ Assert(!do_wait);
+ return;
+ }
+
+ if (messageStartsWith(msg, "OK "))
+ {
+ char *statusString;
+ TocEntry *te;
+
+ pstate->parallelSlot[worker].workerStatus = WRKR_FINISHED;
+ te = pstate->parallelSlot[worker].args->te;
+ if (messageStartsWith(msg, "OK RESTORE "))
+ {
+ statusString = msg + strlen("OK RESTORE ");
+ pstate->parallelSlot[worker].status =
+ (AH->MasterEndParallelItemPtr)
+ (AH, te, statusString, ACT_RESTORE);
+ }
+ else if (messageStartsWith(msg, "OK DUMP "))
+ {
+ statusString = msg + strlen("OK DUMP ");
+ pstate->parallelSlot[worker].status =
+ (AH->MasterEndParallelItemPtr)
+ (AH, te, statusString, ACT_DUMP);
+ }
+ else
+ die_horribly(AH, modulename, "Invalid message received from worker: %s", msg);
+ }
+ 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 "));
+ }
+ else
+ die_horribly(AH, modulename, "Invalid message received from worker: %s", msg);
+
+ PrintStatus(pstate);
+
+ /* both Unix and Win32 return pg_malloc()ed space, so we free it */
+ free(msg);
+ }
+
+ /*
+ * This function is executed in the master process.
+ *
+ * This function is used to get the return value of a terminated worker
+ * process. If a process has terminated, its status is stored in *status and
+ * the id of the worker is returned.
+ */
+ int
+ ReapWorkerStatus(ParallelState *pstate, int *status)
+ {
+ int i;
+
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ if (pstate->parallelSlot[i].workerStatus == WRKR_FINISHED)
+ {
+ *status = pstate->parallelSlot[i].status;
+ pstate->parallelSlot[i].status = 0;
+ pstate->parallelSlot[i].workerStatus = WRKR_IDLE;
+ PrintStatus(pstate);
+ return i;
+ }
+ }
+ return NO_SLOT;
+ }
+
+ /*
+ * This function is executed in the master process.
+ *
+ * It looks for an idle worker process and only returns if there is one.
+ */
+ void
+ EnsureIdleWorker(ArchiveHandle *AH, ParallelState *pstate)
+ {
+ int ret_worker;
+ int work_status;
+
+ for (;;)
+ {
+ int nTerm = 0;
+ while ((ret_worker = ReapWorkerStatus(pstate, &work_status)) != NO_SLOT)
+ {
+ if (work_status != 0)
+ die_horribly(AH, modulename, "Error processing a parallel work item.\n");
+
+ nTerm++;
+ }
+
+ /* We need to make sure that we have an idle worker before dispatching
+ * the next item. If nTerm > 0 we already have that (quick check). */
+ if (nTerm > 0)
+ return;
+
+ /* explicit check for an idle worker */
+ if (GetIdleWorker(pstate) != NO_SLOT)
+ return;
+
+ /*
+ * If we have no idle worker, read the result of one or more
+ * workers and loop the loop to call ReapWorkerStatus() on them
+ */
+ ListenToWorkers(AH, pstate, true);
+ }
+ }
+
+ /*
+ * This function is executed in the master process.
+ *
+ * It waits for all workers to terminate.
+ */
+ void
+ EnsureWorkersFinished(ArchiveHandle *AH, ParallelState *pstate)
+ {
+ int work_status;
+
+ if (!pstate || pstate->numWorkers == 1)
+ return;
+
+ /* Waiting for the remaining worker processes to finish */
+ while (!IsEveryWorkerIdle(pstate))
+ {
+ if (ReapWorkerStatus(pstate, &work_status) == NO_SLOT)
+ ListenToWorkers(AH, pstate, true);
+ else if (work_status != 0)
+ die_horribly(AH, modulename, "Error processing a parallel work item");
+ }
+ }
+
+ /*
+ * This function is executed in the worker process.
+ *
+ * It returns the next message on the communication channel, blocking until it
+ * becomes available.
+ */
+ static char *
+ getMessageFromMaster(ArchiveHandle *AH, int pipefd[2])
+ {
+ return readMessageFromPipe(pipefd[PIPE_READ], true);
+ }
+
+ /*
+ * This function is executed in the worker process.
+ *
+ * It sends a message to the master on the communication channel.
+ */
+ static void
+ sendMessageToMaster(ArchiveHandle *AH, int pipefd[2], const char *str)
+ {
+ int len = strlen(str) + 1;
+
+ if (pipewrite(pipefd[PIPE_WRITE], str, len) != len)
+ die_horribly(AH, modulename,
+ "Error writing to the communication channel: %s",
+ strerror(errno));
+ }
+
+ /*
+ * This function is executed in the master process.
+ *
+ * It returns the next message from the worker on the communication channel,
+ * optionally blocking (do_wait) until it becomes available.
+ *
+ * The id of the worker is returned in *worker.
+ */
+ static char *
+ getMessageFromWorker(ArchiveHandle *AH, ParallelState *pstate, bool do_wait, int *worker)
+ {
+ int i;
+ fd_set workerset;
+ int maxFd = -1;
+ struct timeval nowait = { 0, 0 };
+
+ FD_ZERO(&workerset);
+
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ if (pstate->parallelSlot[i].workerStatus == WRKR_TERMINATED)
+ continue;
+ FD_SET(pstate->parallelSlot[i].pipeRead, &workerset);
+ /* actually WIN32 ignores the first parameter to select()... */
+ if (pstate->parallelSlot[i].pipeRead > maxFd)
+ maxFd = pstate->parallelSlot[i].pipeRead;
+ }
+
+ if (do_wait)
+ {
+ i = select_loop(maxFd, &workerset);
+ Assert(i != 0);
+ }
+ else
+ {
+ if ((i = select(maxFd + 1, &workerset, NULL, NULL, &nowait)) == 0)
+ return NULL;
+ }
+
+ #ifndef WIN32
+ if (wantAbort && !aborting)
+ die_horribly(AH, modulename, "terminated by user\n");
+ #endif
+
+ if (i < 0)
+ {
+ write_msg(NULL, "Error in ListenToWorkers(): %s", strerror(errno));
+ exit(1);
+ }
+
+ for (i = 0; i < pstate->numWorkers; i++)
+ {
+ char *msg;
+
+ if (!FD_ISSET(pstate->parallelSlot[i].pipeRead, &workerset))
+ continue;
+
+ msg = readMessageFromPipe(pstate->parallelSlot[i].pipeRead, false);
+ *worker = i;
+ return msg;
+ }
+ Assert(false);
+ return NULL;
+ }
+
+ /*
+ * This function is executed in the master process.
+ *
+ * It sends a message to a certain worker on the communication channel.
+ */
+ static void
+ sendMessageToWorker(ArchiveHandle *AH, ParallelState *pstate, int worker, const char *str)
+ {
+ int len = strlen(str) + 1;
+
+ if (pipewrite(pstate->parallelSlot[worker].pipeWrite, str, len) != len)
+ die_horribly(AH, modulename,
+ "Error writing to the communication channel: %s",
+ strerror(errno));
+ }
+
+ /*
+ * The underlying function to read a message from the communication channel (fd)
+ * with optional blocking (do_wait).
+ */
+ static char *
+ readMessageFromPipe(int fd, bool do_wait)
+ {
+ char *msg;
+ int msgsize, bufsize;
+ int ret;
+
+ /*
+ * The problem here is that we need to deal with several possibilites:
+ * we could receive only a partial message or several messages at once.
+ * The caller expects us to return exactly one message however.
+ *
+ * We could either read in as much as we can and keep track of what we
+ * delivered back to the caller or we just read byte by byte. Once we see
+ * (char) 0, we know that it's the message's end. This would be quite
+ * inefficient for more data but since we are reading only on the command
+ * channel, the performance loss does not seem worth the trouble of keeping
+ * internal states for different file descriptors.
+ */
+
+ bufsize = 64; /* could be any number */
+ msg = (char *) pg_malloc(bufsize);
+
+ msgsize = 0;
+ for (;;)
+ {
+ Assert(msgsize <= bufsize);
+ /*
+ * If we do non-blocking read, only set the channel non-blocking for
+ * the very first character. We trust in our messages to be
+ * \0-terminated, so if there is any character in the beginning, then
+ * we read the message until we find a \0 somewhere, which indicates
+ * the end of the message.
+ */
+ if (msgsize == 0 && !do_wait) {
+ setnonblocking(fd);
+ }
+
+ ret = piperead(fd, msg + msgsize, 1);
+
+ if (msgsize == 0 && !do_wait)
+ {
+ int saved_errno = errno;
+ setblocking(fd);
+ /* no data has been available */
+ if (ret < 0 && saved_errno == EAGAIN)
+ return NULL;
+ }
+
+ /* worker has closed the connection or another error happened */
+ if (ret <= 0)
+ return NULL;
+
+ Assert(ret == 1);
+
+ if (msg[msgsize] == '\0') {
+ return msg;
+ }
+
+ msgsize++;
+ if (msgsize == bufsize)
+ {
+ /* could be any number */
+ bufsize += 16;
+ msg = (char *) realloc(msg, bufsize);
+ }
+ }
+ }
+
diff --git a/src/bin/pg_dump/parallel.h b/src/bin/pg_dump/parallel.h
index ...4c86b9b .
*** a/src/bin/pg_dump/parallel.h
--- b/src/bin/pg_dump/parallel.h
***************
*** 0 ****
--- 1,91 ----
+ /*-------------------------------------------------------------------------
+ *
+ * parallel.h
+ *
+ * Parallel support header file for the pg_dump archiver
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * The author is not responsible for loss or damages that may
+ * result from its use.
+ *
+ * IDENTIFICATION
+ * src/bin/pg_dump/parallel.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+ #include "pg_backup_db.h"
+
+ struct _archiveHandle;
+ struct _tocEntry;
+
+ typedef enum
+ {
+ WRKR_TERMINATED = 0,
+ WRKR_IDLE,
+ WRKR_WORKING,
+ WRKR_FINISHED
+ } T_WorkerStatus;
+
+ typedef enum _action
+ {
+ ACT_DUMP,
+ ACT_RESTORE,
+ } T_Action;
+
+ /* Arguments needed for a worker process */
+ typedef struct _parallel_args
+ {
+ struct _archiveHandle *AH;
+ struct _tocEntry *te;
+ } ParallelArgs;
+
+ /* State for each parallel activity slot */
+ typedef struct _parallel_slot
+ {
+ ParallelArgs *args;
+ T_WorkerStatus workerStatus;
+ int status;
+ int pipeRead;
+ int pipeWrite;
+ int pipeRevRead;
+ int pipeRevWrite;
+ #ifdef WIN32
+ uintptr_t hThread;
+ unsigned int threadId;
+ #else
+ pid_t pid;
+ #endif
+ bool inErrorHandling;
+ } ParallelSlot;
+
+ #define NO_SLOT (-1)
+
+ typedef struct _parallel_state
+ {
+ int numWorkers;
+ ParallelSlot *parallelSlot;
+ } ParallelState;
+
+ extern int GetIdleWorker(ParallelState *pstate);
+ extern bool IsEveryWorkerIdle(ParallelState *pstate);
+ extern void ListenToWorkers(struct _archiveHandle *AH, ParallelState *pstate, bool do_wait);
+ extern int ReapWorkerStatus(ParallelState *pstate, int *status);
+ extern void EnsureIdleWorker(struct _archiveHandle *AH, ParallelState *pstate);
+ extern void EnsureWorkersFinished(struct _archiveHandle *AH, ParallelState *pstate);
+
+ extern ParallelState *ParallelBackupStart(struct _archiveHandle *AH,
+ RestoreOptions *ropt);
+ extern void DispatchJobForTocEntry(struct _archiveHandle *AH,
+ ParallelState *pstate,
+ struct _tocEntry *te, T_Action act);
+ extern void ParallelBackupEnd(struct _archiveHandle *AH, ParallelState *pstate);
+
+ extern void (* volatile vparallel_error_handler)(struct _archiveHandle *AH,
+ const char *modulename,
+ const char *fmt, va_list ap);
+
+ extern void checkAborting(struct _archiveHandle *AH);
+
diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h
index 8926488..767f865 100644
*** a/src/bin/pg_dump/pg_backup.h
--- b/src/bin/pg_dump/pg_backup.h
*************** typedef struct _Archive
*** 90,95 ****
--- 90,97 ----
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
*** 150,156 ****
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;
--- 152,157 ----
*************** typedef struct _restoreOptions
*** 162,180 ****
/* Lets the archive know we have a DB connection to shutdown if it dies */
! PGconn *ConnectDatabase(Archive *AH,
const char *dbname,
const char *pghost,
const char *pgport,
const char *username,
enum trivalue prompt_password);
/* Called to add a TOC entry */
extern void ArchiveEntry(Archive *AHX,
CatalogId catalogId, DumpId dumpId,
const char *tag,
const char *namespace, const char *tablespace,
! const char *owner, bool withOids,
const char *desc, teSection section,
const char *defn,
const char *dropStmt, const char *copyStmt,
--- 163,183 ----
/* Lets the archive know we have a DB connection to shutdown if it dies */
! PGconn *ConnectDatabase(Archive *AHX,
const char *dbname,
const char *pghost,
const char *pgport,
const char *username,
enum trivalue prompt_password);
+ PGconn *CloneDatabaseConnection(Archive *AHX);
/* Called to add a TOC entry */
extern void ArchiveEntry(Archive *AHX,
CatalogId catalogId, DumpId dumpId,
const char *tag,
const char *namespace, const char *tablespace,
! const char *owner,
! unsigned long int relpages, bool withOids,
const char *desc, teSection section,
const char *defn,
const char *dropStmt, const char *copyStmt,
*************** extern void PrintTOCSummary(Archive *AH,
*** 203,208 ****
--- 206,214 ----
extern RestoreOptions *NewRestoreOptions(void);
+ /* We have one in pg_dump.c and another one in pg_restore.c */
+ extern void _SetupWorker(Archive *AHX, RestoreOptions *ropt);
+
/* Rearrange and filter TOC entries */
extern void SortTocFromFile(Archive *AHX, RestoreOptions *ropt);
extern void InitDummyWantedList(Archive *AHX, RestoreOptions *ropt);
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 234e50f..0c81dfe 100644
*** a/src/bin/pg_dump/pg_backup_archiver.c
--- b/src/bin/pg_dump/pg_backup_archiver.c
***************
*** 23,82 ****
#include "pg_backup_db.h"
#include "dumpmem.h"
#include "dumputils.h"
#include <ctype.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#ifdef WIN32
#include <io.h>
#endif
#include "libpq/libpq-fs.h"
- /*
- * Special exit values from worker children. We reserve 0 for normal
- * success; 1 and other small values should be interpreted as crashes.
- */
- #define WORKER_CREATE_DONE 10
- #define WORKER_INHIBIT_DATA 11
- #define WORKER_IGNORED_ERRORS 12
-
- /*
- * Unix uses exit to return result from worker child, so function is void.
- * Windows thread result comes via function return.
- */
- #ifndef WIN32
- #define parallel_restore_result void
- #else
- #define parallel_restore_result DWORD
- #endif
-
- /* IDs for worker children are either PIDs or thread handles */
- #ifndef WIN32
- #define thandle pid_t
- #else
- #define thandle HANDLE
- #endif
-
- /* Arguments needed for a worker child */
- typedef struct _restore_args
- {
- ArchiveHandle *AH;
- TocEntry *te;
- } RestoreArgs;
-
- /* State for each parallel activity slot */
- typedef struct _parallel_slot
- {
- thandle child_id;
- RestoreArgs *args;
- } ParallelSlot;
-
- #define NO_SLOT (-1)
-
#define TEXT_DUMP_HEADER "--\n-- PostgreSQL database dump\n--\n\n"
#define TEXT_DUMPALL_HEADER "--\n-- PostgreSQL database cluster dump\n--\n\n"
--- 23,46 ----
#include "pg_backup_db.h"
#include "dumpmem.h"
#include "dumputils.h"
+ #include "parallel.h"
#include <ctype.h>
+ #include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
+ static const char *modulename = gettext_noop("archiver");
+
+
#ifdef WIN32
#include <io.h>
#endif
#include "libpq/libpq-fs.h"
#define TEXT_DUMP_HEADER "--\n-- PostgreSQL database dump\n--\n\n"
#define TEXT_DUMPALL_HEADER "--\n-- PostgreSQL database cluster dump\n--\n\n"
*************** typedef struct _outputContext
*** 87,94 ****
int gzOut;
} OutputContext;
- static const char *modulename = gettext_noop("archiver");
-
/* index array created by fix_dependencies -- only used in parallel restore */
static TocEntry **tocsByDumpId; /* index by dumpId - 1 */
static DumpId maxDumpId; /* length of above array */
--- 51,56 ----
*************** static teReqs _tocEntryRequired(TocEntry
*** 115,121 ****
static bool _tocEntryIsACL(TocEntry *te);
static void _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
static void _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
- static TocEntry *getTocEntryByDumpId(ArchiveHandle *AH, DumpId id);
static void _moveBefore(ArchiveHandle *AH, TocEntry *pos, TocEntry *te);
static int _discoverArchiveFormat(ArchiveHandle *AH);
--- 77,82 ----
*************** static void RestoreOutput(ArchiveHandle
*** 132,152 ****
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);
! static int get_next_slot(ParallelSlot *slots, int n_slots);
static void par_list_header_init(TocEntry *l);
static void par_list_append(TocEntry *l, TocEntry *te);
static void par_list_remove(TocEntry *te);
static TocEntry *get_next_work_item(ArchiveHandle *AH,
TocEntry *ready_list,
! ParallelSlot *slots, int n_slots);
! static parallel_restore_result parallel_restore(RestoreArgs *args);
static void mark_work_done(ArchiveHandle *AH, TocEntry *ready_list,
! thandle worker, int status,
! ParallelSlot *slots, int n_slots);
static void fix_dependencies(ArchiveHandle *AH);
static bool has_lock_conflicts(TocEntry *te1, TocEntry *te2);
static void repoint_table_dependencies(ArchiveHandle *AH,
--- 93,111 ----
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, ParallelState *pstate,
! TocEntry *pending_list);
! static void restore_toc_entries_postfork(ArchiveHandle *AH, TocEntry *pending_list);
static void par_list_header_init(TocEntry *l);
static void par_list_append(TocEntry *l, TocEntry *te);
static void par_list_remove(TocEntry *te);
static TocEntry *get_next_work_item(ArchiveHandle *AH,
TocEntry *ready_list,
! ParallelState *pstate);
static void mark_work_done(ArchiveHandle *AH, TocEntry *ready_list,
! int worker, int status,
! ParallelState *pstate);
static void fix_dependencies(ArchiveHandle *AH);
static bool has_lock_conflicts(TocEntry *te1, TocEntry *te2);
static void repoint_table_dependencies(ArchiveHandle *AH,
*************** static void reduce_dependencies(ArchiveH
*** 156,164 ****
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);
-
/*
* Wrapper functions.
--- 115,120 ----
*************** RestoreArchive(Archive *AHX, RestoreOpti
*** 245,251 ****
/*
* 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 */
--- 201,207 ----
/*
* 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, RestoreOpti
*** 411,417 ****
* 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)
--- 367,391 ----
* In parallel mode, turn control over to the parallel-restore logic.
*/
if (parallel_mode)
! {
! ParallelState *pstate;
! 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);
!
! /* ParallelBackupStart() will actually fork the processes */
! pstate = ParallelBackupStart(AH, ropt);
! restore_toc_entries_parallel(AH, pstate, &pending_list);
! ParallelBackupEnd(AH, pstate);
!
! /* 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)
*************** static int
*** 476,482 ****
restore_toc_entry(ArchiveHandle *AH, TocEntry *te,
RestoreOptions *ropt, bool is_parallel)
{
! int retval = 0;
teReqs reqs;
bool defnDumped;
--- 450,456 ----
restore_toc_entry(ArchiveHandle *AH, TocEntry *te,
RestoreOptions *ropt, bool is_parallel)
{
! int status = WORKER_OK;
teReqs reqs;
bool defnDumped;
*************** restore_toc_entry(ArchiveHandle *AH, Toc
*** 518,524 ****
if (ropt->noDataForFailedTables)
{
if (is_parallel)
! retval = WORKER_INHIBIT_DATA;
else
inhibit_data_for_failed_table(AH, te);
}
--- 492,498 ----
if (ropt->noDataForFailedTables)
{
if (is_parallel)
! status = WORKER_INHIBIT_DATA;
else
inhibit_data_for_failed_table(AH, te);
}
*************** restore_toc_entry(ArchiveHandle *AH, Toc
*** 533,539 ****
* just set the return value.
*/
if (is_parallel)
! retval = WORKER_CREATE_DONE;
else
mark_create_done(AH, te);
}
--- 507,513 ----
* just set the return value.
*/
if (is_parallel)
! status = WORKER_CREATE_DONE;
else
mark_create_done(AH, te);
}
*************** restore_toc_entry(ArchiveHandle *AH, Toc
*** 651,657 ****
}
}
! return retval;
}
/*
--- 625,634 ----
}
}
! if (AH->public.n_errors > 0 && status == WORKER_OK)
! status = WORKER_IGNORED_ERRORS;
!
! return status;
}
/*
*************** ArchiveEntry(Archive *AHX,
*** 753,759 ****
const char *tag,
const char *namespace,
const char *tablespace,
! const char *owner, bool withOids,
const char *desc, teSection section,
const char *defn,
const char *dropStmt, const char *copyStmt,
--- 730,737 ----
const char *tag,
const char *namespace,
const char *tablespace,
! const char *owner,
! unsigned long int relpages, bool withOids,
const char *desc, teSection section,
const char *defn,
const char *dropStmt, const char *copyStmt,
*************** static void
*** 1429,1434 ****
--- 1407,1421 ----
vdie_horribly(ArchiveHandle *AH, const char *modulename,
const char *fmt, va_list ap)
{
+ /*
+ * If we have an error handler for the parallel operation, then
+ * control will not come back from there.
+ */
+ if (vparallel_error_handler)
+ vparallel_error_handler(AH, modulename, fmt, ap);
+
+ Assert(!vparallel_error_handler);
+
vwrite_msg(modulename, fmt, ap);
if (AH)
*************** _moveBefore(ArchiveHandle *AH, TocEntry
*** 1534,1540 ****
pos->prev = te;
}
! static TocEntry *
getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)
{
TocEntry *te;
--- 1521,1527 ----
pos->prev = te;
}
! TocEntry *
getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)
{
TocEntry *te;
*************** _allocAH(const char *FileSpec, const Arc
*** 1944,1949 ****
--- 1931,1938 ----
AH->archiveDumpVersion = PG_VERSION;
+ AH->is_clone = false;
+
AH->createDate = time(NULL);
AH->intSize = sizeof(int);
*************** _allocAH(const char *FileSpec, const Arc
*** 2035,2082 ****
void
! WriteDataChunks(ArchiveHandle *AH)
{
TocEntry *te;
- StartDataPtr startPtr;
- EndDataPtr endPtr;
for (te = AH->toc->next; te != AH->toc; te = te->next)
{
! if (te->dataDumper != NULL)
! {
! AH->currToc = te;
! /* printf("Writing data for %d (%x)\n", te->id, te); */
!
! if (strcmp(te->desc, "BLOBS") == 0)
! {
! startPtr = AH->StartBlobsPtr;
! endPtr = AH->EndBlobsPtr;
! }
! else
! {
! startPtr = AH->StartDataPtr;
! endPtr = AH->EndDataPtr;
! }
!
! if (startPtr != NULL)
! (*startPtr) (AH, te);
/*
! * printf("Dumper arg for %d is %x\n", te->id, te->dataDumperArg);
*/
! /*
! * The user-provided DataDumper routine needs to call
! * AH->WriteData
! */
! (*te->dataDumper) ((Archive *) AH, te->dataDumperArg);
! if (endPtr != NULL)
! (*endPtr) (AH, te);
! AH->currToc = NULL;
! }
}
}
void
--- 2024,2088 ----
void
! WriteDataChunks(ArchiveHandle *AH, ParallelState *pstate)
{
TocEntry *te;
for (te = AH->toc->next; te != AH->toc; te = te->next)
{
! if (!te->hadDumper)
! continue;
+ if (pstate && pstate->numWorkers > 1)
+ {
/*
! * If we are in a parallel backup, then we are always the master
! * process.
*/
+ EnsureIdleWorker(AH, pstate);
+ Assert(GetIdleWorker(pstate) != NO_SLOT);
+ DispatchJobForTocEntry(AH, pstate, te, ACT_DUMP);
+ }
+ else
+ {
+ WriteDataChunksForTocEntry(AH, te);
+ }
+ }
+ EnsureWorkersFinished(AH, pstate);
+ }
! void
! WriteDataChunksForTocEntry(ArchiveHandle *AH, TocEntry *te)
! {
! StartDataPtr startPtr;
! EndDataPtr endPtr;
! AH->currToc = te;
!
! if (strcmp(te->desc, "BLOBS") == 0)
! {
! startPtr = AH->StartBlobsPtr;
! endPtr = AH->EndBlobsPtr;
! }
! else
! {
! startPtr = AH->StartDataPtr;
! endPtr = AH->EndDataPtr;
}
+
+ if (startPtr != NULL)
+ (*startPtr) (AH, te);
+
+ /*
+ * The user-provided DataDumper routine needs to call
+ * AH->WriteData
+ */
+ (*te->dataDumper) ((Archive *) AH, te->dataDumperArg);
+
+ if (endPtr != NULL)
+ (*endPtr) (AH, te);
+
+ AH->currToc = NULL;
}
void
*************** WriteToc(ArchiveHandle *AH)
*** 2086,2093 ****
char workbuf[32];
int i;
- /* printf("%d TOC Entries to save\n", AH->tocCount); */
-
WriteInt(AH, AH->tocCount);
for (te = AH->toc->next; te != AH->toc; te = te->next)
--- 2092,2097 ----
*************** dumpTimestamp(ArchiveHandle *AH, const c
*** 3239,3274 ****
ahprintf(AH, "-- %s %s\n\n", msg, buf);
}
-
- /*
- * 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;
! ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
! slots = (ParallelSlot *) pg_calloc(sizeof(ParallelSlot), n_slots);
/* Adjust dependency information */
fix_dependencies(AH);
--- 3243,3264 ----
ahprintf(AH, "-- %s %s\n\n", msg, buf);
}
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");
! /* we haven't got round to making this work for all archive formats */
! if (AH->ClonePtr == NULL || AH->ReopenPtr == NULL)
! die_horribly(AH, modulename, "parallel restore is not supported with this archive file format\n");
!
! /* doesn't work if the archive represents dependencies as OIDs, either */
! if (AH->version < K_VERS_1_8)
! die_horribly(AH, modulename, "parallel restore is not supported with archives made by pre-8.0 pg_dump\n");
/* Adjust dependency information */
fix_dependencies(AH);
*************** restore_toc_entries_parallel(ArchiveHand
*** 3336,3352 ****
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)
--- 3326,3367 ----
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), these fork off and set up their
+ * connections before we call restore_toc_entries_parallel_forked.
+ * Finally we process all the ACL entries in a single connection (that happens
+ * back in RestoreArchive).
+ */
+ static void
+ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
+ TocEntry *pending_list)
+ {
+ int work_status;
+ bool skipped_some;
+ TocEntry ready_list;
+ TocEntry *next_work_item;
+ int ret_child;
+
+ ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
/*
! * Initialize the lists of ready items, the list for pending items has
! * already been initialized in the parent. 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
*** 3371,3377 ****
}
if (next_work_item->depCount > 0)
! par_list_append(&pending_list, next_work_item);
else
par_list_append(&ready_list, next_work_item);
}
--- 3386,3392 ----
}
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
*** 3385,3393 ****
ahlog(AH, 1, "entering main parallel loop\n");
! while ((next_work_item = get_next_work_item(AH, &ready_list,
! slots, n_slots)) != NULL ||
! work_in_progress(slots, n_slots))
{
if (next_work_item != NULL)
{
--- 3400,3407 ----
ahlog(AH, 1, "entering main parallel loop\n");
! while ((next_work_item = get_next_work_item(AH, &ready_list, pstate)) != NULL ||
! !IsEveryWorkerIdle(pstate))
{
if (next_work_item != NULL)
{
*************** restore_toc_entries_parallel(ArchiveHand
*** 3407,3461 ****
continue;
}
! if ((next_slot = get_next_slot(slots, n_slots)) != NO_SLOT)
! {
! /* There is work still to do and a worker slot available */
! thandle child;
! RestoreArgs *args;
!
! ahlog(AH, 1, "launching item %d %s %s\n",
! next_work_item->dumpId,
! next_work_item->desc, next_work_item->tag);
! par_list_remove(next_work_item);
! /* this memory is dealloced in mark_work_done() */
! args = pg_malloc(sizeof(RestoreArgs));
! args->AH = CloneArchive(AH);
! args->te = next_work_item;
! /* run the step in a worker child */
! child = spawn_restore(args);
! slots[next_slot].child_id = child;
! slots[next_slot].args = args;
! continue;
}
- }
! /*
! * If we get here there must be work being done. Either there is no
! * work available to schedule (and work_in_progress returned true) or
! * there are no slots available. So we wait for a worker to finish,
! * and process the result.
! */
! ret_child = reap_child(slots, n_slots, &work_status);
! if (WIFEXITED(work_status))
! {
! mark_work_done(AH, &ready_list,
! ret_child, WEXITSTATUS(work_status),
! slots, n_slots);
! }
! else
! {
! die_horribly(AH, modulename, "worker process crashed: status %d\n",
! work_status);
}
}
ahlog(AH, 1, "finished main parallel loop\n");
/*
* Now reconnect the single parent connection.
--- 3421,3493 ----
continue;
}
! ahlog(AH, 1, "launching item %d %s %s\n",
! next_work_item->dumpId,
! next_work_item->desc, next_work_item->tag);
! par_list_remove(next_work_item);
! Assert(GetIdleWorker(pstate) != NO_SLOT);
! DispatchJobForTocEntry(AH, pstate, next_work_item, ACT_RESTORE);
! }
! else
! {
! /* at least one child is working and we have nothing ready. */
! Assert(!IsEveryWorkerIdle(pstate));
! }
! for (;;)
! {
! int nTerm = 0;
! /*
! * In order to reduce dependencies as soon as possible and
! * especially to reap the status of workers who are working on
! * items that pending items depend on, we do a non-blocking check
! * for ended workers first.
! *
! * However, if we do not have any other work items currently that
! * workers can work on, we do not busy-loop here but instead
! * really wait for at least one worker to terminate. Hence we call
! * ListenToWorkers(..., ..., do_wait = true) in this case.
! */
! ListenToWorkers(AH, pstate, !next_work_item);
! while ((ret_child = ReapWorkerStatus(pstate, &work_status)) != NO_SLOT)
! {
! nTerm++;
! mark_work_done(AH, &ready_list, ret_child, work_status, pstate);
}
! /*
! * We need to make sure that we have an idle worker before re-running the
! * loop. If nTerm > 0 we already have that (quick check).
! */
! if (nTerm > 0)
! break;
! /* if nobody terminated, explicitly check for an idle worker */
! if (GetIdleWorker(pstate) != NO_SLOT)
! break;
!
! /*
! * If we have no idle worker, read the result of one or more
! * workers and loop the loop to call ReapWorkerStatus() on them.
! */
! ListenToWorkers(AH, pstate, true);
}
}
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");
/*
* Now reconnect the single parent connection.
*************** restore_toc_entries_parallel(ArchiveHand
*** 3471,3477 ****
* 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);
--- 3503,3509 ----
* 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);
*************** restore_toc_entries_parallel(ArchiveHand
*** 3482,3602 ****
}
/*
- * create a worker child to perform a restore step in parallel
- */
- static thandle
- spawn_restore(RestoreArgs *args)
- {
- thandle child;
-
- /* Ensure stdio state is quiesced before forking */
- fflush(NULL);
-
- #ifndef WIN32
- child = fork();
- if (child == 0)
- {
- /* in child process */
- parallel_restore(args);
- die_horribly(args->AH, modulename,
- "parallel_restore should not return\n");
- }
- else if (child < 0)
- {
- /* fork failed */
- die_horribly(args->AH, modulename,
- "could not create worker process: %s\n",
- strerror(errno));
- }
- #else
- child = (HANDLE) _beginthreadex(NULL, 0, (void *) parallel_restore,
- args, 0, NULL);
- if (child == 0)
- die_horribly(args->AH, modulename,
- "could not create worker thread: %s\n",
- strerror(errno));
- #endif
-
- return child;
- }
-
- /*
- * collect status from a completed worker child
- */
- static thandle
- reap_child(ParallelSlot *slots, int n_slots, int *work_status)
- {
- #ifndef WIN32
- /* Unix is so much easier ... */
- return wait(work_status);
- #else
- static HANDLE *handles = NULL;
- int hindex,
- snum,
- tnum;
- thandle ret_child;
- DWORD res;
-
- /* first time around only, make space for handles to listen on */
- if (handles == NULL)
- handles = (HANDLE *) pg_calloc(sizeof(HANDLE), n_slots);
-
- /* set up list of handles to listen to */
- for (snum = 0, tnum = 0; snum < n_slots; snum++)
- if (slots[snum].child_id != 0)
- handles[tnum++] = slots[snum].child_id;
-
- /* wait for one to finish */
- hindex = WaitForMultipleObjects(tnum, handles, false, INFINITE);
-
- /* get handle of finished thread */
- ret_child = handles[hindex - WAIT_OBJECT_0];
-
- /* get the result */
- GetExitCodeThread(ret_child, &res);
- *work_status = res;
-
- /* dispose of handle to stop leaks */
- CloseHandle(ret_child);
-
- return ret_child;
- #endif
- }
-
- /*
- * are we doing anything now?
- */
- static bool
- work_in_progress(ParallelSlot *slots, int n_slots)
- {
- int i;
-
- for (i = 0; i < n_slots; i++)
- {
- if (slots[i].child_id != 0)
- return true;
- }
- return false;
- }
-
- /*
- * find the first free parallel slot (if any).
- */
- static int
- get_next_slot(ParallelSlot *slots, int n_slots)
- {
- int i;
-
- for (i = 0; i < n_slots; i++)
- {
- if (slots[i].child_id == 0)
- return i;
- }
- return NO_SLOT;
- }
-
-
- /*
* Check if te1 has an exclusive lock requirement for an item that te2 also
* requires, whether or not te2's requirement is for an exclusive lock.
*/
--- 3514,3519 ----
*************** par_list_remove(TocEntry *te)
*** 3669,3675 ****
*/
static TocEntry *
get_next_work_item(ArchiveHandle *AH, TocEntry *ready_list,
! ParallelSlot *slots, int n_slots)
{
bool pref_non_data = false; /* or get from AH->ropt */
TocEntry *data_te = NULL;
--- 3586,3592 ----
*/
static TocEntry *
get_next_work_item(ArchiveHandle *AH, TocEntry *ready_list,
! ParallelState *pstate)
{
bool pref_non_data = false; /* or get from AH->ropt */
TocEntry *data_te = NULL;
*************** get_next_work_item(ArchiveHandle *AH, To
*** 3684,3694 ****
{
int count = 0;
! for (k = 0; k < n_slots; k++)
! if (slots[k].args->te != NULL &&
! slots[k].args->te->section == SECTION_DATA)
count++;
! if (n_slots == 0 || count * 4 < n_slots)
pref_non_data = false;
}
--- 3601,3611 ----
{
int count = 0;
! for (k = 0; k < pstate->numWorkers; k++)
! if (pstate->parallelSlot[k].args->te != NULL &&
! pstate->parallelSlot[k].args->te->section == SECTION_DATA)
count++;
! if (pstate->numWorkers == 0 || count * 4 < pstate->numWorkers)
pref_non_data = false;
}
*************** get_next_work_item(ArchiveHandle *AH, To
*** 3704,3716 ****
* that a currently running item also needs lock on, or vice versa. If
* so, we don't want to schedule them together.
*/
! for (i = 0; i < n_slots && !conflicts; i++)
{
TocEntry *running_te;
! if (slots[i].args == NULL)
continue;
! running_te = slots[i].args->te;
if (has_lock_conflicts(te, running_te) ||
has_lock_conflicts(running_te, te))
--- 3621,3633 ----
* that a currently running item also needs lock on, or vice versa. If
* so, we don't want to schedule them together.
*/
! for (i = 0; i < pstate->numWorkers && !conflicts; i++)
{
TocEntry *running_te;
! if (pstate->parallelSlot[i].workerStatus != WRKR_WORKING)
continue;
! running_te = pstate->parallelSlot[i].args->te;
if (has_lock_conflicts(te, running_te) ||
has_lock_conflicts(running_te, te))
*************** get_next_work_item(ArchiveHandle *AH, To
*** 3745,3805 ****
/*
* Restore a single TOC item in parallel with others
*
! * this is the procedure run as a thread (Windows) or a
! * separate process (everything else).
*/
! static parallel_restore_result
! parallel_restore(RestoreArgs *args)
{
ArchiveHandle *AH = args->AH;
TocEntry *te = args->te;
RestoreOptions *ropt = AH->ropt;
! int retval;
!
! /*
! * Close and reopen the input file so we have a private file pointer that
! * doesn't stomp on anyone else's file pointer, if we're actually going to
! * need to read from the file. Otherwise, just close it except on Windows,
! * where it will possibly be needed by other threads.
! *
! * Note: on Windows, since we are using threads not processes, the reopen
! * call *doesn't* close the original file pointer but just open a new one.
! */
! if (te->section == SECTION_DATA)
! (AH->ReopenPtr) (AH);
! #ifndef WIN32
! else
! (AH->ClosePtr) (AH);
! #endif
!
! /*
! * We need our own database connection, too
! */
! ConnectDatabase((Archive *) AH, ropt->dbname,
! ropt->pghost, ropt->pgport, ropt->username,
! ropt->promptPassword);
_doSetFixedOutputState(AH);
! /* Restore the TOC item */
! retval = restore_toc_entry(AH, te, ropt, true);
!
! /* And clean up */
! PQfinish(AH->connection);
! AH->connection = NULL;
! /* If we reopened the file, we are done with it, so close it now */
! if (te->section == SECTION_DATA)
! (AH->ClosePtr) (AH);
! if (retval == 0 && AH->public.n_errors)
! retval = WORKER_IGNORED_ERRORS;
! #ifndef WIN32
! exit(retval);
! #else
! return retval;
! #endif
}
--- 3662,3690 ----
/*
* Restore a single TOC item in parallel with others
*
! * this is run in the worker, i.e. in a thread (Windows) or a separate process
! * (everything else). A worker process executes several such work items during
! * a parallel backup or restore. Once we terminate here and report back that
! * our work is finished, the master process will assign us a new work item.
*/
! int
! parallel_restore(ParallelArgs *args)
{
ArchiveHandle *AH = args->AH;
TocEntry *te = args->te;
RestoreOptions *ropt = AH->ropt;
! int status;
_doSetFixedOutputState(AH);
! Assert(AH->connection != NULL);
! AH->public.n_errors = 0;
! /* Restore the TOC item */
! status = restore_toc_entry(AH, te, ropt, true);
! return status;
}
*************** parallel_restore(RestoreArgs *args)
*** 3811,3835 ****
*/
static void
mark_work_done(ArchiveHandle *AH, TocEntry *ready_list,
! thandle worker, int status,
! ParallelSlot *slots, int n_slots)
{
TocEntry *te = NULL;
- int i;
-
- for (i = 0; i < n_slots; i++)
- {
- if (slots[i].child_id == worker)
- {
- slots[i].child_id = 0;
- te = slots[i].args->te;
- DeCloneArchive(slots[i].args->AH);
- free(slots[i].args);
- slots[i].args = NULL;
! break;
! }
! }
if (te == NULL)
die_horribly(AH, modulename, "could not find slot of finished worker\n");
--- 3696,3707 ----
*/
static void
mark_work_done(ArchiveHandle *AH, TocEntry *ready_list,
! int worker, int status,
! ParallelState *pstate)
{
TocEntry *te = NULL;
! te = pstate->parallelSlot[worker].args->te;
if (te == NULL)
die_horribly(AH, modulename, "could not find slot of finished worker\n");
*************** inhibit_data_for_failed_table(ArchiveHan
*** 4184,4193 ****
*
* 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;
--- 4056,4063 ----
*
* 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)
*** 4213,4221 ****
--- 4083,4145 ----
/* clone has its own error count, too */
clone->public.n_errors = 0;
+ /*
+ * Remember that we're a clone, this is used for deciding if we should
+ * install a snapshot.
+ */
+ clone->is_clone = true;
+
+ /*
+ * Connect our new clone object to the database:
+ * In parallel restore the parent is already disconnected.
+ * In parallel backup we clone the parent's existing connection.
+ */
+ if (AH->ropt)
+ {
+ RestoreOptions *ropt = AH->ropt;
+ Assert(AH->connection == NULL);
+ /* this also sets clone->connection */
+ ConnectDatabase((Archive *) clone, ropt->dbname,
+ ropt->pghost, ropt->pgport, ropt->username,
+ ropt->promptPassword);
+ }
+ else
+ {
+ char *dbname;
+ char *pghost;
+ char *pgport;
+ char *username;
+ const char *encname;
+
+ Assert(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(AH->connection);
+ pghost = PQhost(AH->connection);
+ pgport = PQport(AH->connection);
+ username = PQuser(AH->connection);
+ encname = pg_encoding_to_char(AH->public.encoding);
+
+ /* this also sets clone->connection */
+ ConnectDatabase((Archive *) clone, dbname, pghost, pgport, username, TRI_NO);
+
+ /*
+ * Set the same encoding, whatever we set here is what we got from
+ * pg_encoding_to_char(), so we really shouldn't run into an error setting that
+ * very same value. Also see the comment in SetupConnection().
+ */
+ PQsetClientEncoding(clone->connection, encname);
+ }
+
/* Let the format-specific code have a chance too */
(clone->ClonePtr) (clone);
+ Assert(clone->connection != NULL);
return clone;
}
*************** CloneArchive(ArchiveHandle *AH)
*** 4224,4230 ****
*
* Note: we assume any clone-local connection was already closed.
*/
! static void
DeCloneArchive(ArchiveHandle *AH)
{
/* Clear format-specific state */
--- 4148,4154 ----
*
* 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 6dd5158..3b10384 100644
*** a/src/bin/pg_dump/pg_backup_archiver.h
--- b/src/bin/pg_dump/pg_backup_archiver.h
*************** typedef z_stream *z_streamp;
*** 100,108 ****
--- 100,120 ----
#define K_OFFSET_POS_SET 2
#define K_OFFSET_NO_DATA 3
+ /*
+ * Special exit values from worker children. We reserve 0 for normal
+ * success; 1 and other small values should be interpreted as crashes.
+ */
+ #define WORKER_OK 0
+ #define WORKER_CREATE_DONE 10
+ #define WORKER_INHIBIT_DATA 11
+ #define WORKER_IGNORED_ERRORS 12
+
struct _archiveHandle;
struct _tocEntry;
struct _restoreList;
+ struct _parallel_args;
+ struct _parallel_state;
+ enum _action;
typedef void (*ClosePtr) (struct _archiveHandle * AH);
typedef void (*ReopenPtr) (struct _archiveHandle * AH);
*************** typedef void (*PrintTocDataPtr) (struct
*** 130,135 ****
--- 142,154 ----
typedef void (*ClonePtr) (struct _archiveHandle * AH);
typedef void (*DeClonePtr) (struct _archiveHandle * AH);
+ typedef char *(*WorkerJobRestorePtr)(struct _archiveHandle * AH, struct _tocEntry * te);
+ typedef char *(*WorkerJobDumpPtr)(struct _archiveHandle * AH, struct _tocEntry * te);
+ typedef char *(*MasterStartParallelItemPtr)(struct _archiveHandle * AH, struct _tocEntry * te,
+ enum _action act);
+ typedef int (*MasterEndParallelItemPtr)(struct _archiveHandle * AH, struct _tocEntry * te,
+ const char *str, enum _action act);
+
typedef size_t (*CustomOutPtr) (struct _archiveHandle * AH, const void *buf, size_t len);
typedef enum
*************** typedef struct _archiveHandle
*** 188,193 ****
--- 207,213 ----
* Added V1.7 */
ArchiveFormat format; /* Archive format */
+ bool is_clone; /* have we been cloned ? */
sqlparseInfo sqlparse; /* state for parsing INSERT data */
time_t createDate; /* Date archive created */
*************** typedef struct _archiveHandle
*** 228,233 ****
--- 248,259 ----
StartBlobPtr StartBlobPtr;
EndBlobPtr EndBlobPtr;
+ MasterStartParallelItemPtr MasterStartParallelItemPtr;
+ MasterEndParallelItemPtr MasterEndParallelItemPtr;
+
+ WorkerJobDumpPtr WorkerJobDumpPtr;
+ WorkerJobRestorePtr WorkerJobRestorePtr;
+
ClonePtr ClonePtr; /* Clone format-specific fields */
DeClonePtr DeClonePtr; /* Clean up cloned fields */
*************** typedef struct _archiveHandle
*** 237,242 ****
--- 263,271 ----
char *archdbname; /* DB name *read* from archive */
enum trivalue promptPassword;
char *savedPassword; /* password for ropt->username, if known */
+ char *use_role;
+ char *sync_snapshot_id; /* sync snapshot id for parallel
+ operation */
PGconn *connection;
int connectToDB; /* Flag to indicate if direct DB connection is
* required */
*************** typedef struct _tocEntry
*** 324,329 ****
--- 353,359 ----
int nLockDeps; /* number of such dependencies */
} TocEntry;
+ extern int parallel_restore(struct _parallel_args *args);
extern void die_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt,...) __attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 4)));
extern void warn_or_die_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt,...) __attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 4)));
*************** extern void WriteHead(ArchiveHandle *AH)
*** 334,342 ****
extern void ReadHead(ArchiveHandle *AH);
extern void WriteToc(ArchiveHandle *AH);
extern void ReadToc(ArchiveHandle *AH);
! extern void WriteDataChunks(ArchiveHandle *AH);
extern teReqs TocIDRequired(ArchiveHandle *AH, DumpId id, RestoreOptions *ropt);
extern bool checkSeek(FILE *fp);
#define appendStringLiteralAHX(buf,str,AH) \
--- 364,377 ----
extern void ReadHead(ArchiveHandle *AH);
extern void WriteToc(ArchiveHandle *AH);
extern void ReadToc(ArchiveHandle *AH);
! extern void WriteDataChunks(ArchiveHandle *AH, struct _parallel_state *pstate);
! extern void WriteDataChunksForTocEntry(ArchiveHandle *AH, TocEntry *te);
!
! extern ArchiveHandle *CloneArchive(ArchiveHandle *AH);
! extern void DeCloneArchive(ArchiveHandle *AH);
extern teReqs TocIDRequired(ArchiveHandle *AH, DumpId id, RestoreOptions *ropt);
+ TocEntry *getTocEntryByDumpId(ArchiveHandle *AH, DumpId id);
extern bool checkSeek(FILE *fp);
#define appendStringLiteralAHX(buf,str,AH) \
*************** int ahprintf(ArchiveHandle *AH, const
*** 378,381 ****
--- 413,428 ----
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_custom.c b/src/bin/pg_dump/pg_backup_custom.c
index 31fa373..f3070f7 100644
*** a/src/bin/pg_dump/pg_backup_custom.c
--- b/src/bin/pg_dump/pg_backup_custom.c
***************
*** 27,32 ****
--- 27,33 ----
#include "compress_io.h"
#include "dumputils.h"
#include "dumpmem.h"
+ #include "parallel.h"
/*--------
* Routines in the format interface
*************** static void _LoadBlobs(ArchiveHandle *AH
*** 60,65 ****
--- 61,70 ----
static void _Clone(ArchiveHandle *AH);
static void _DeClone(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);
+ char *_WorkerJobRestoreCustom(ArchiveHandle *AH, TocEntry *te);
+
typedef struct
{
CompressorState *cs;
*************** static size_t _CustomReadFunc(ArchiveHan
*** 87,94 ****
static const char *modulename = gettext_noop("custom archiver");
-
-
/*
* Init routine required by ALL formats. This is a global routine
* and should be declared in pg_backup_archiver.h
--- 92,97 ----
*************** InitArchiveFmt_Custom(ArchiveHandle *AH)
*** 127,132 ****
--- 130,142 ----
AH->ClonePtr = _Clone;
AH->DeClonePtr = _DeClone;
+ AH->MasterStartParallelItemPtr = _MasterStartParallelItem;
+ AH->MasterEndParallelItemPtr = _MasterEndParallelItem;
+
+ /* no parallel dump in the custom archive, only parallel restore */
+ AH->WorkerJobDumpPtr = NULL;
+ AH->WorkerJobRestorePtr = _WorkerJobRestoreCustom;
+
/* Set up a private area. */
ctx = (lclContext *) pg_calloc(1, sizeof(lclContext));
AH->formatData = (void *) ctx;
*************** _CloseArchive(ArchiveHandle *AH)
*** 698,704 ****
tpos = ftello(AH->FH);
WriteToc(AH);
ctx->dataStart = _getFilePos(AH, ctx);
! WriteDataChunks(AH);
/*
* If possible, re-write the TOC in order to update the data offset
--- 708,714 ----
tpos = ftello(AH->FH);
WriteToc(AH);
ctx->dataStart = _getFilePos(AH, ctx);
! WriteDataChunks(AH, NULL);
/*
* If possible, re-write the TOC in order to update the data offset
*************** _DeClone(ArchiveHandle *AH)
*** 796,801 ****
--- 806,886 ----
free(ctx);
}
+ /*
+ * This function is executed in the child of a parallel backup for the
+ * custom format archive and dumps the actual data.
+ */
+ char *
+ _WorkerJobRestoreCustom(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;
+
+ pargs.AH = AH;
+ pargs.te = te;
+
+ status = parallel_restore(&pargs);
+
+ snprintf(buf, buflen, "OK RESTORE %d %d %d", te->dumpId, status,
+ status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0);
+
+ return buf;
+ }
+
+ /*
+ * 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]; /* short fixed-size string + number */
+
+ /* no parallel dump in the custom archive format */
+ Assert(act == ACT_RESTORE);
+
+ snprintf(buf, sizeof(buf), "RESTORE %d", te->dumpId);
+
+ 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, status, n_errors;
+
+ /* no parallel dump in the custom archive */
+ Assert(act == ACT_RESTORE);
+
+ sscanf(str, "%u %u %u%n", &dumpId, &status, &n_errors, &nBytes);
+
+ Assert(nBytes == strlen(str));
+ Assert(dumpId == te->dumpId);
+
+ AH->public.n_errors += n_errors;
+
+ return status;
+ }
+
/*--------------------------------------------------
* END OF FORMAT CALLBACKS
*--------------------------------------------------
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 9c6d7c1..995cf31 100644
*** a/src/bin/pg_dump/pg_backup_directory.c
--- b/src/bin/pg_dump/pg_backup_directory.c
***************
*** 35,40 ****
--- 35,42 ----
#include "compress_io.h"
#include "dumpmem.h"
+ #include "dumputils.h"
+ #include "parallel.h"
#include <dirent.h>
#include <sys/stat.h>
*************** typedef struct
*** 50,55 ****
--- 52,58 ----
cfp *dataFH; /* currently open data file */
cfp *blobsTocFH; /* file handle for blobs.toc */
+ ParallelState *pstate; /* for parallel backup / restore */
} lclContext;
typedef struct
*************** static int _ReadByte(ArchiveHandle *);
*** 69,74 ****
--- 72,78 ----
static size_t _WriteBuf(ArchiveHandle *AH, const void *buf, size_t len);
static size_t _ReadBuf(ArchiveHandle *AH, void *buf, size_t len);
static void _CloseArchive(ArchiveHandle *AH);
+ static void _ReopenArchive(ArchiveHandle *AH);
static void _PrintTocData(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
static void _WriteExtraToc(ArchiveHandle *AH, TocEntry *te);
*************** static void _StartBlob(ArchiveHandle *AH
*** 80,90 ****
static void _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
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);
!
/*
* Init routine required by ALL formats. This is a global routine
--- 84,101 ----
static void _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
static void _LoadBlobs(ArchiveHandle *AH, RestoreOptions *ropt);
+ static void _Clone(ArchiveHandle *AH);
+ static void _DeClone(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 *_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);
! static char *_WorkerJobDumpDirectory(ArchiveHandle *AH, TocEntry *te);
! static char *_WorkerJobRestoreDirectory(ArchiveHandle *AH, TocEntry *te);
/*
* Init routine required by ALL formats. This is a global routine
*************** InitArchiveFmt_Directory(ArchiveHandle *
*** 111,117 ****
AH->WriteBufPtr = _WriteBuf;
AH->ReadBufPtr = _ReadBuf;
AH->ClosePtr = _CloseArchive;
! AH->ReopenPtr = NULL;
AH->PrintTocDataPtr = _PrintTocData;
AH->ReadExtraTocPtr = _ReadExtraToc;
AH->WriteExtraTocPtr = _WriteExtraToc;
--- 122,128 ----
AH->WriteBufPtr = _WriteBuf;
AH->ReadBufPtr = _ReadBuf;
AH->ClosePtr = _CloseArchive;
! AH->ReopenPtr = _ReopenArchive;
AH->PrintTocDataPtr = _PrintTocData;
AH->ReadExtraTocPtr = _ReadExtraToc;
AH->WriteExtraTocPtr = _WriteExtraToc;
*************** InitArchiveFmt_Directory(ArchiveHandle *
*** 122,129 ****
AH->EndBlobPtr = _EndBlob;
AH->EndBlobsPtr = _EndBlobs;
! AH->ClonePtr = NULL;
! AH->DeClonePtr = NULL;
/* Set up our private context */
ctx = (lclContext *) pg_calloc(1, sizeof(lclContext));
--- 133,146 ----
AH->EndBlobPtr = _EndBlob;
AH->EndBlobsPtr = _EndBlobs;
! AH->ClonePtr = _Clone;
! AH->DeClonePtr = _DeClone;
!
! AH->WorkerJobRestorePtr = _WorkerJobRestoreDirectory;
! AH->WorkerJobDumpPtr = _WorkerJobDumpDirectory;
!
! AH->MasterStartParallelItemPtr = _MasterStartParallelItem;
! AH->MasterEndParallelItemPtr = _MasterEndParallelItem;
/* Set up our private context */
ctx = (lclContext *) pg_calloc(1, sizeof(lclContext));
*************** InitArchiveFmt_Directory(ArchiveHandle *
*** 152,161 ****
}
else
{ /* Read Mode */
! char *fname;
cfp *tocFH;
! fname = prependDirectory(AH, "toc.dat");
tocFH = cfopen_read(fname, PG_BINARY_R);
if (tocFH == NULL)
--- 169,178 ----
}
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 *
*** 281,289 ****
{
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)
--- 298,306 ----
{
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
*** 372,379 ****
_LoadBlobs(AH, ropt);
else
{
! char *fname = prependDirectory(AH, tctx->filename);
_PrintFileData(AH, fname, ropt);
}
}
--- 389,397 ----
_LoadBlobs(AH, ropt);
else
{
! char fname[MAXPGPATH];
+ prependDirectory(AH, fname, tctx->filename);
_PrintFileData(AH, fname, ropt);
}
}
*************** _LoadBlobs(ArchiveHandle *AH, RestoreOpt
*** 383,394 ****
{
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);
--- 401,412 ----
{
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)
*** 515,521 ****
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);
--- 533,544 ----
if (AH->mode == archModeWrite)
{
cfp *tocFH;
! char fname[MAXPGPATH];
!
! prependDirectory(AH, fname, "toc.dat");
!
! /* this will actually fork the processes for a parallel backup */
! ctx->pstate = ParallelBackupStart(AH, NULL);
/* The TOC is always created uncompressed */
tocFH = cfopen_write(fname, PG_BINARY_W, 0);
*************** _CloseArchive(ArchiveHandle *AH)
*** 536,546 ****
if (cfclose(tocFH) != 0)
die_horribly(AH, modulename, "could not close TOC file: %s\n",
strerror(errno));
! WriteDataChunks(AH);
}
AH->FH = NULL;
}
/*
* BLOB support
--- 559,582 ----
if (cfclose(tocFH) != 0)
die_horribly(AH, modulename, "could not close TOC file: %s\n",
strerror(errno));
! WriteDataChunks(AH, ctx->pstate);
!
! ParallelBackupEnd(AH, ctx->pstate);
}
AH->FH = NULL;
}
+ /*
+ * Reopen the archive's file handle.
+ */
+ static void
+ _ReopenArchive(ArchiveHandle *AH)
+ {
+ /*
+ * Our TOC is in memory, our data files are opened by each child anyway as
+ * they are separate. We support reopening the archive by just doing nothing.
+ */
+ }
/*
* BLOB support
*************** static void
*** 557,565 ****
_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);
--- 593,601 ----
_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);
*************** createDirectory(const char *dir)
*** 652,663 ****
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;
--- 688,703 ----
dir, 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;
*************** prependDirectory(ArchiveHandle *AH, cons
*** 671,673 ****
--- 711,864 ----
return buf;
}
+
+ /*
+ * Clone format-specific fields during parallel restoration.
+ */
+ static void
+ _Clone(ArchiveHandle *AH)
+ {
+ lclContext *ctx = (lclContext *) AH->formatData;
+
+ AH->formatData = (lclContext *) pg_malloc(sizeof(lclContext));
+ if (AH->formatData == NULL)
+ die_horribly(AH, modulename, "out of memory\n");
+ memcpy(AH->formatData, ctx, sizeof(lclContext));
+ ctx = (lclContext *) AH->formatData;
+
+ /*
+ * 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];
+
+ if (act == ACT_DUMP)
+ snprintf(buf, sizeof(buf), "DUMP %d", te->dumpId);
+ else if (act == ACT_RESTORE)
+ 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.
+ *
+ * We are currently returning only the DumpId so theoretically we could
+ * make this function returning an int (or a DumpId). However, to
+ * facilitate further enhancements and because sooner or later we need to
+ * convert this to a string and send it via a message anyway, we stick with
+ * char *. It is parsed on the other side by the _EndMasterParallel()
+ * function of the respective dump format.
+ */
+ static char *
+ _WorkerJobDumpDirectory(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);
+ lclTocEntry *tctx = (lclTocEntry *) te->formatData;
+
+ /* This should never happen */
+ if (!tctx)
+ die_horribly(AH, modulename, "Error during backup\n");
+
+ /*
+ * This function returns void. We either fail and die horribly or succeed...
+ * A failure will be detected by the parent when the child dies unexpectedly.
+ */
+ WriteDataChunksForTocEntry(AH, te);
+
+ snprintf(buf, buflen, "OK DUMP %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 *
+ _WorkerJobRestoreDirectory(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;
+
+ pargs.AH = AH;
+ pargs.te = te;
+
+ status = parallel_restore(&pargs);
+
+ snprintf(buf, buflen, "OK RESTORE %d %d %d", te->dumpId, status,
+ status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0);
+
+ 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;
+
+ if (act == ACT_DUMP)
+ {
+ sscanf(str, "%u%n", &dumpId, &nBytes);
+
+ Assert(dumpId == te->dumpId);
+ Assert(nBytes == strlen(str));
+ }
+ else if (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_backup_files.c b/src/bin/pg_dump/pg_backup_files.c
index ffcbb8f..13e8ed3 100644
*** a/src/bin/pg_dump/pg_backup_files.c
--- b/src/bin/pg_dump/pg_backup_files.c
*************** InitArchiveFmt_Files(ArchiveHandle *AH)
*** 102,107 ****
--- 102,113 ----
AH->ClonePtr = NULL;
AH->DeClonePtr = NULL;
+ AH->MasterStartParallelItemPtr = NULL;
+ AH->MasterEndParallelItemPtr = NULL;
+
+ AH->WorkerJobDumpPtr = NULL;
+ AH->WorkerJobRestorePtr = NULL;
+
/*
* Set up some special context used in compressing data.
*/
*************** _CloseArchive(ArchiveHandle *AH)
*** 455,461 ****
WriteToc(AH);
if (fclose(AH->FH) != 0)
die_horribly(AH, modulename, "could not close TOC file: %s\n", strerror(errno));
! WriteDataChunks(AH);
}
AH->FH = NULL;
--- 461,467 ----
WriteToc(AH);
if (fclose(AH->FH) != 0)
die_horribly(AH, modulename, "could not close TOC file: %s\n", strerror(errno));
! WriteDataChunks(AH, NULL);
}
AH->FH = NULL;
diff --git a/src/bin/pg_dump/pg_backup_tar.c b/src/bin/pg_dump/pg_backup_tar.c
index 39ce417..929083e 100644
*** a/src/bin/pg_dump/pg_backup_tar.c
--- b/src/bin/pg_dump/pg_backup_tar.c
*************** InitArchiveFmt_Tar(ArchiveHandle *AH)
*** 157,162 ****
--- 157,168 ----
AH->ClonePtr = NULL;
AH->DeClonePtr = NULL;
+ AH->MasterStartParallelItemPtr = NULL;
+ AH->MasterEndParallelItemPtr = NULL;
+
+ AH->WorkerJobDumpPtr = NULL;
+ AH->WorkerJobRestorePtr = NULL;
+
/*
* Set up some special context used in compressing data.
*/
*************** _CloseArchive(ArchiveHandle *AH)
*** 835,841 ****
/*
* Now send the data (tables & blobs)
*/
! WriteDataChunks(AH);
/*
* Now this format wants to append a script which does a full restore
--- 841,847 ----
/*
* Now send the data (tables & blobs)
*/
! WriteDataChunks(AH, NULL);
/*
* Now this format wants to append a script which does a full restore
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 13fc667..5a15435 100644
*** a/src/bin/pg_dump/pg_dump.c
--- b/src/bin/pg_dump/pg_dump.c
*************** static int disable_dollar_quoting = 0;
*** 140,145 ****
--- 140,146 ----
static int dump_inserts = 0;
static int column_inserts = 0;
static int no_security_labels = 0;
+ static int no_synchronized_snapshots = 0;
static int no_unlogged_table_data = 0;
static int serializable_deferrable = 0;
*************** static const char *convertTSFunction(Oid
*** 227,235 ****
static Oid findLastBuiltinOid_V71(const char *);
static Oid findLastBuiltinOid_V70(void);
static void selectSourceSchema(const char *schemaName);
static char *getFormattedTypeName(Oid oid, OidOptions opts);
static char *myFormatType(const char *typname, int32 typmod);
- static const char *fmtQualifiedId(const char *schema, const char *id);
static void getBlobs(Archive *AH);
static void dumpBlob(Archive *AH, BlobInfo *binfo);
static int dumpBlobs(Archive *AH, void *arg);
--- 228,237 ----
static Oid findLastBuiltinOid_V71(const char *);
static Oid findLastBuiltinOid_V70(void);
static void selectSourceSchema(const char *schemaName);
+ static void selectSourceSchemaOnAH(ArchiveHandle *AH, const char *schemaName);
+ static void selectSourceSchemaOnConnection(PGconn *conn, const char *schemaName);
static char *getFormattedTypeName(Oid oid, OidOptions opts);
static char *myFormatType(const char *typname, int32 typmod);
static void getBlobs(Archive *AH);
static void dumpBlob(Archive *AH, BlobInfo *binfo);
static int dumpBlobs(Archive *AH, void *arg);
*************** static void binary_upgrade_extension_mem
*** 246,255 ****
DumpableObject *dobj,
const char *objlabel);
static const char *getAttrName(int attrnum, TableInfo *tblInfo);
! static const char *fmtCopyColumnList(const TableInfo *ti);
static void do_sql_command(PGconn *conn, const char *query);
static void check_sql_result(PGresult *res, PGconn *conn, const char *query,
ExecStatusType expected);
int
main(int argc, char **argv)
--- 248,260 ----
DumpableObject *dobj,
const char *objlabel);
static const char *getAttrName(int attrnum, TableInfo *tblInfo);
! static const char *fmtCopyColumnList(const TableInfo *ti, PQExpBuffer buffer);
static void do_sql_command(PGconn *conn, const char *query);
static void check_sql_result(PGresult *res, PGconn *conn, const char *query,
ExecStatusType expected);
+ static void SetupConnection(Archive *AHX, const char *dumpencoding,
+ const char *use_role);
+ static char *get_synchronized_snapshot(ArchiveHandle *AH);
int
main(int argc, char **argv)
*************** main(int argc, char **argv)
*** 262,274 ****
const char *pgport = NULL;
const char *username = NULL;
const char *dumpencoding = NULL;
- const char *std_strings;
bool oids = false;
TableInfo *tblinfo;
int numTables;
DumpableObject **dobjs;
int numObjs;
int i;
enum trivalue prompt_password = TRI_DEFAULT;
int compressLevel = -1;
int plainText = 0;
--- 267,279 ----
const char *pgport = NULL;
const char *username = NULL;
const char *dumpencoding = NULL;
bool oids = false;
TableInfo *tblinfo;
int numTables;
DumpableObject **dobjs;
int numObjs;
int i;
+ int numWorkers = 1;
enum trivalue prompt_password = TRI_DEFAULT;
int compressLevel = -1;
int plainText = 0;
*************** main(int argc, char **argv)
*** 297,302 ****
--- 302,308 ----
{"format", required_argument, NULL, 'F'},
{"host", required_argument, NULL, 'h'},
{"ignore-version", no_argument, NULL, 'i'},
+ {"jobs", 1, NULL, 'j'},
{"no-reconnect", no_argument, NULL, 'R'},
{"oids", no_argument, NULL, 'o'},
{"no-owner", no_argument, NULL, 'O'},
*************** main(int argc, char **argv)
*** 336,341 ****
--- 342,348 ----
{"serializable-deferrable", no_argument, &serializable_deferrable, 1},
{"use-set-session-authorization", no_argument, &use_setsessauth, 1},
{"no-security-labels", no_argument, &no_security_labels, 1},
+ {"no-synchronized-snapshots", no_argument, &no_synchronized_snapshots, 1},
{"no-unlogged-table-data", no_argument, &no_unlogged_table_data, 1},
{NULL, 0, NULL, 0}
*************** main(int argc, char **argv)
*** 373,379 ****
}
}
! while ((c = getopt_long(argc, argv, "abcCE:f:F:h:in:N:oOp:RsS:t:T:U:vwWxZ:",
long_options, &optindex)) != -1)
{
switch (c)
--- 380,386 ----
}
}
! while ((c = getopt_long(argc, argv, "abcCE:f:F:h:ij:n:N:oOp:RsS:t:T:U:vwWxZ:",
long_options, &optindex)) != -1)
{
switch (c)
*************** main(int argc, char **argv)
*** 414,419 ****
--- 421,430 ----
/* ignored, deprecated option */
break;
+ case 'j': /* number of dump jobs */
+ numWorkers = atoi(optarg);
+ break;
+
case 'n': /* include schema(s) */
simple_string_list_append(&schema_include_patterns, optarg);
include_everything = false;
*************** main(int argc, char **argv)
*** 575,580 ****
--- 586,612 ----
compressLevel = 0;
}
+ /*
+ * On Windows we can only have at most MAXIMUM_WAIT_OBJECTS (= 64 usually)
+ * parallel jobs because that's the maximum limit for the
+ * WaitForMultipleObjects() call.
+ */
+ if (numWorkers <= 0
+ #ifdef WIN32
+ || numWorkers > MAXIMUM_WAIT_OBJECTS
+ #endif
+ )
+ {
+ write_msg(NULL, _("%s: invalid number of parallel jobs\n"), progname);
+ exit(1);
+ }
+
+ /* 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);
+ }
+
/* Open the output file */
g_fout = CreateArchive(filename, archiveFormat, compressLevel, archiveMode);
*************** main(int argc, char **argv)
*** 601,606 ****
--- 633,640 ----
g_fout->minRemoteVersion = 70000;
g_fout->maxRemoteVersion = (my_version / 100) * 100 + 99;
+ g_fout->numWorkers = numWorkers;
+
/*
* Open the database using the Archiver, so it knows about it. Errors mean
* death.
*************** main(int argc, char **argv)
*** 608,702 ****
g_conn = ConnectDatabase(g_fout, dbname, pghost, pgport,
username, prompt_password);
! /* Set the client encoding if requested */
! if (dumpencoding)
{
! if (PQsetClientEncoding(g_conn, dumpencoding) < 0)
! {
! write_msg(NULL, "invalid client encoding \"%s\" specified\n",
! dumpencoding);
! exit(1);
! }
}
/*
- * Get the active encoding and the standard_conforming_strings setting, so
- * we know how to escape strings.
- */
- g_fout->encoding = PQclientEncoding(g_conn);
-
- std_strings = PQparameterStatus(g_conn, "standard_conforming_strings");
- g_fout->std_strings = (std_strings && strcmp(std_strings, "on") == 0);
-
- /* Set the role if requested */
- if (use_role && g_fout->remoteVersion >= 80100)
- {
- PQExpBuffer query = createPQExpBuffer();
-
- appendPQExpBuffer(query, "SET ROLE %s", fmtId(use_role));
- do_sql_command(g_conn, query->data);
- destroyPQExpBuffer(query);
- }
-
- /* Set the datestyle to ISO to ensure the dump's portability */
- do_sql_command(g_conn, "SET DATESTYLE = ISO");
-
- /* Likewise, avoid using sql_standard intervalstyle */
- if (g_fout->remoteVersion >= 80400)
- do_sql_command(g_conn, "SET INTERVALSTYLE = POSTGRES");
-
- /*
- * If supported, set extra_float_digits so that we can dump float data
- * exactly (given correctly implemented float I/O code, anyway)
- */
- if (g_fout->remoteVersion >= 90000)
- do_sql_command(g_conn, "SET extra_float_digits TO 3");
- else if (g_fout->remoteVersion >= 70400)
- do_sql_command(g_conn, "SET extra_float_digits TO 2");
-
- /*
- * If synchronized scanning is supported, disable it, to prevent
- * unpredictable changes in row ordering across a dump and reload.
- */
- if (g_fout->remoteVersion >= 80300)
- do_sql_command(g_conn, "SET synchronize_seqscans TO off");
-
- /*
- * Disable timeouts if supported.
- */
- if (g_fout->remoteVersion >= 70300)
- do_sql_command(g_conn, "SET statement_timeout = 0");
-
- /*
- * Quote all identifiers, if requested.
- */
- if (quote_all_identifiers && g_fout->remoteVersion >= 90100)
- do_sql_command(g_conn, "SET quote_all_identifiers = true");
-
- /*
* Disable security label support if server version < v9.1.x (prevents
* access to nonexistent pg_seclabel catalog)
*/
if (g_fout->remoteVersion < 90100)
no_security_labels = 1;
- /*
- * Start transaction-snapshot mode transaction to dump consistent data.
- */
- do_sql_command(g_conn, "BEGIN");
- if (g_fout->remoteVersion >= 90100)
- {
- if (serializable_deferrable)
- do_sql_command(g_conn,
- "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE, "
- "READ ONLY, DEFERRABLE");
- else
- do_sql_command(g_conn,
- "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ");
- }
- else
- do_sql_command(g_conn, "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE");
-
/* Select the appropriate subquery to convert user IDs to names */
if (g_fout->remoteVersion >= 80100)
username_subquery = "SELECT rolname FROM pg_catalog.pg_roles WHERE oid =";
--- 642,665 ----
g_conn = ConnectDatabase(g_fout, dbname, pghost, pgport,
username, prompt_password);
! /* Find the last built-in OID, if needed */
! if (g_fout->remoteVersion < 70300)
{
! if (g_fout->remoteVersion >= 70100)
! g_last_builtin_oid = findLastBuiltinOid_V71(PQdb(g_conn));
! else
! g_last_builtin_oid = findLastBuiltinOid_V70();
! if (g_verbose)
! write_msg(NULL, "last built-in OID is %u\n", g_last_builtin_oid);
}
/*
* Disable security label support if server version < v9.1.x (prevents
* access to nonexistent pg_seclabel catalog)
*/
if (g_fout->remoteVersion < 90100)
no_security_labels = 1;
/* Select the appropriate subquery to convert user IDs to names */
if (g_fout->remoteVersion >= 80100)
username_subquery = "SELECT rolname FROM pg_catalog.pg_roles WHERE oid =";
*************** main(int argc, char **argv)
*** 705,721 ****
else
username_subquery = "SELECT usename FROM pg_user WHERE usesysid =";
! /* Find the last built-in OID, if needed */
! if (g_fout->remoteVersion < 70300)
{
! if (g_fout->remoteVersion >= 70100)
! g_last_builtin_oid = findLastBuiltinOid_V71(PQdb(g_conn));
! else
! g_last_builtin_oid = findLastBuiltinOid_V70();
! if (g_verbose)
! write_msg(NULL, "last built-in OID is %u\n", g_last_builtin_oid);
}
/* Expand schema selection patterns into OID lists */
if (schema_include_patterns.head != NULL)
{
--- 668,687 ----
else
username_subquery = "SELECT usename FROM pg_user WHERE usesysid =";
! if (numWorkers > 1)
{
! /* check the version for the synchronized snapshots feature */
! if (g_fout->remoteVersion < 90200 && !no_synchronized_snapshots)
! {
! write_msg(NULL, "No synchronized snapshots available in this version\n"
! "You might have to run with --no-synchronized-snapshots\n");
! exit(1);
! } else if (g_fout->remoteVersion >= 90200 && no_synchronized_snapshots)
! write_msg(NULL, "Ignoring --no-synchronized-snapshots\n");
}
+ SetupConnection(g_fout, dumpencoding, use_role);
+
/* Expand schema selection patterns into OID lists */
if (schema_include_patterns.head != NULL)
{
*************** main(int argc, char **argv)
*** 797,802 ****
--- 763,772 ----
else
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);
/*
*************** help(const char *progname)
*** 862,867 ****
--- 832,838 ----
printf(_(" -f, --file=FILENAME output file or directory name\n"));
printf(_(" -F, --format=c|d|t|p output file format (custom, directory, tar,\n"
" plain text (default))\n"));
+ printf(_(" -j, --jobs=NUM use this many parallel jobs to dump\n"));
printf(_(" -v, --verbose verbose mode\n"));
printf(_(" -Z, --compress=0-9 compression level for compressed formats\n"));
printf(_(" --lock-wait-timeout=TIMEOUT fail after waiting TIMEOUT for a table lock\n"));
*************** help(const char *progname)
*** 891,896 ****
--- 862,868 ----
printf(_(" --exclude-table-data=TABLE do NOT dump data for the named table(s)\n"));
printf(_(" --inserts dump data as INSERT commands, rather than COPY\n"));
printf(_(" --no-security-labels do not dump security label assignments\n"));
+ printf(_(" --no-synchronized-snapshots parallel processes should not use synchronized snapshots\n"));
printf(_(" --no-tablespaces do not dump tablespace assignments\n"));
printf(_(" --no-unlogged-table-data do not dump unlogged table data\n"));
printf(_(" --quote-all-identifiers quote all identifiers, even if not key words\n"));
*************** exit_nicely(void)
*** 922,927 ****
--- 894,1066 ----
exit(1);
}
+ /*
+ * Initialize the connection for a new worker process.
+ */
+ void
+ _SetupWorker(Archive *AHX, RestoreOptions *ropt)
+ {
+ SetupConnection(AHX, NULL, NULL);
+ }
+
+ static void
+ SetupConnection(Archive *AHX, const char *dumpencoding, const char *use_role)
+ {
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
+ const char *std_strings;
+ PGconn *conn = AH->connection;
+
+ /*
+ * Set the client encoding if requested. If dumpencoding == NULL then
+ * either it hasn't been requested or we're a cloned connection and then this
+ * has already been set in CloneArchive according to the original
+ * connection encoding.
+ */
+ if (dumpencoding)
+ {
+ if (PQsetClientEncoding(AH->connection, dumpencoding) < 0)
+ {
+ write_msg(NULL, "invalid client encoding \"%s\" specified\n",
+ dumpencoding);
+ exit(1);
+ }
+ }
+
+ /*
+ * Get the active encoding and the standard_conforming_strings setting, so
+ * we know how to escape strings.
+ */
+ AHX->encoding = PQclientEncoding(conn);
+
+ std_strings = PQparameterStatus(conn, "standard_conforming_strings");
+ AHX->std_strings = (std_strings && strcmp(std_strings, "on") == 0);
+
+ /* Set the role if requested */
+ if (!use_role && AH->use_role)
+ use_role = AH->use_role;
+
+ if (use_role && AHX->remoteVersion >= 80100)
+ {
+ PQExpBuffer query = createPQExpBuffer();
+
+ appendPQExpBuffer(query, "SET ROLE %s", fmtId(use_role));
+ do_sql_command(conn, query->data);
+ destroyPQExpBuffer(query);
+
+ /* save this for later use on parallel connections */
+ if (!AH->use_role)
+ AH->use_role = strdup(use_role);
+ }
+
+ /* Set the datestyle to ISO to ensure the dump's portability */
+ do_sql_command(conn, "SET DATESTYLE = ISO");
+
+ /* Likewise, avoid using sql_standard intervalstyle */
+ if (AHX->remoteVersion >= 80400)
+ do_sql_command(conn, "SET INTERVALSTYLE = POSTGRES");
+
+ /*
+ * If supported, set extra_float_digits so that we can dump float data
+ * exactly (given correctly implemented float I/O code, anyway)
+ */
+ if (AHX->remoteVersion >= 80500)
+ do_sql_command(conn, "SET extra_float_digits TO 3");
+ else if (AHX->remoteVersion >= 70400)
+ do_sql_command(conn, "SET extra_float_digits TO 2");
+
+ /*
+ * If synchronized scanning is supported, disable it, to prevent
+ * unpredictable changes in row ordering across a dump and reload.
+ */
+ if (AHX->remoteVersion >= 80300)
+ do_sql_command(conn, "SET synchronize_seqscans TO off");
+
+ /*
+ * Quote all identifiers, if requested.
+ */
+ if (quote_all_identifiers && AHX->remoteVersion >= 90100)
+ do_sql_command(conn, "SET quote_all_identifiers = true");
+
+ /*
+ * Disable timeouts if supported.
+ */
+ if (AHX->remoteVersion >= 70300)
+ do_sql_command(conn, "SET statement_timeout = 0");
+
+ /*
+ * Quote all identifiers, if requested.
+ */
+ if (quote_all_identifiers && AHX->remoteVersion >= 90100)
+ do_sql_command(conn, "SET quote_all_identifiers = true");
+
+ /*
+ * Start transaction-snapshot mode transaction to dump consistent data.
+ */
+ do_sql_command(conn, "BEGIN");
+ if (AHX->remoteVersion >= 90100)
+ {
+ if (serializable_deferrable)
+ do_sql_command(conn,
+ "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE, "
+ "READ ONLY, DEFERRABLE");
+ else
+ do_sql_command(conn,
+ "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ");
+ }
+ else
+ do_sql_command(conn, "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE");
+
+ if (AHX->numWorkers > 1 && AHX->remoteVersion >= 90200)
+ {
+ if (AH->is_clone)
+ {
+ PQExpBuffer query = createPQExpBuffer();
+ appendPQExpBuffer(query, "SET TRANSACTION SNAPSHOT ");
+ appendStringLiteralConn(query, AH->sync_snapshot_id, conn);
+ destroyPQExpBuffer(query);
+ }
+ else {
+ /*
+ * If the version is lower and we don't have synchronized snapshots
+ * yet, we will error out earlier already. So either we have the
+ * feature or the user has given the explicit command not to use it.
+ * Note: If we have it, we always use it, you cannot switch it off
+ * then.
+ */
+ if (AHX->remoteVersion >= 90200)
+ AH->sync_snapshot_id = get_synchronized_snapshot(AH);
+ }
+ }
+ }
+
+ static char*
+ get_synchronized_snapshot(ArchiveHandle *AH)
+ {
+ const char *query = "select pg_export_snapshot()";
+ char *result;
+ int ntups;
+ PGconn *conn = AH->connection;
+ PGresult *res = PQexec(conn, query);
+
+ check_sql_result(res, conn, query, PGRES_TUPLES_OK);
+
+ /* Expecting a single result only */
+ ntups = PQntuples(res);
+ if (ntups != 1)
+ {
+ write_msg(NULL, ngettext("query returned %d row instead of one: %s\n",
+ "query returned %d rows instead of one: %s\n",
+ ntups),
+ ntups, query);
+ exit_nicely();
+ }
+
+ result = strdup(PQgetvalue(res, 0, 0));
+ PQclear(res);
+
+ return result;
+ }
+
static ArchiveFormat
parseArchiveFormat(const char *format, ArchiveMode *mode)
{
*************** selectDumpableObject(DumpableObject *dob
*** 1252,1263 ****
--- 1391,1412 ----
static int
dumpTableData_copy(Archive *fout, void *dcontext)
{
+ /*
+ * This is a data dumper routine, executed in a child for parallel backup, so
+ * it must not access the global g_conn but AH->connection instead.
+ */
+ ArchiveHandle *AH = (ArchiveHandle *) fout;
TableDataInfo *tdinfo = (TableDataInfo *) dcontext;
TableInfo *tbinfo = tdinfo->tdtable;
const char *classname = tbinfo->dobj.name;
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();
PGresult *res;
int ret;
char *copybuf;
*************** dumpTableData_copy(Archive *fout, void *
*** 1272,1278 ****
* this ensures reproducible results in case the table contains regproc,
* regclass, etc columns.
*/
! selectSourceSchema(tbinfo->dobj.namespace->dobj.name);
/*
* If possible, specify the column list explicitly so that we have no
--- 1421,1427 ----
* this ensures reproducible results in case the table contains regproc,
* regclass, etc columns.
*/
! selectSourceSchemaOnAH(AH, tbinfo->dobj.namespace->dobj.name);
/*
* If possible, specify the column list explicitly so that we have no
*************** dumpTableData_copy(Archive *fout, void *
*** 1280,1287 ****
* column ordering of COPY will not be what we want in certain corner
* cases involving ADD COLUMN and inheritance.)
*/
! if (g_fout->remoteVersion >= 70300)
! column_list = fmtCopyColumnList(tbinfo);
else
column_list = ""; /* can't select columns in COPY */
--- 1429,1436 ----
* column ordering of COPY will not be what we want in certain corner
* cases involving ADD COLUMN and inheritance.)
*/
! if (AH->public.remoteVersion >= 70300)
! column_list = fmtCopyColumnList(tbinfo, clistBuf);
else
column_list = ""; /* can't select columns in COPY */
*************** dumpTableData_copy(Archive *fout, void *
*** 1289,1295 ****
{
appendPQExpBuffer(q, "COPY %s %s WITH OIDS TO stdout;",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname),
column_list);
}
else if (tdinfo->filtercond)
--- 1438,1445 ----
{
appendPQExpBuffer(q, "COPY %s %s WITH OIDS TO stdout;",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname,
! AH->public.remoteVersion),
column_list);
}
else if (tdinfo->filtercond)
*************** dumpTableData_copy(Archive *fout, void *
*** 1306,1328 ****
appendPQExpBufferStr(q, "* ");
appendPQExpBuffer(q, "FROM %s %s) TO stdout;",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname),
tdinfo->filtercond);
}
else
{
appendPQExpBuffer(q, "COPY %s %s TO stdout;",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname),
column_list);
}
! res = PQexec(g_conn, q->data);
! check_sql_result(res, g_conn, q->data, PGRES_COPY_OUT);
PQclear(res);
for (;;)
{
! ret = PQgetCopyData(g_conn, ©buf, 0);
if (ret < 0)
break; /* done or error */
--- 1456,1481 ----
appendPQExpBufferStr(q, "* ");
appendPQExpBuffer(q, "FROM %s %s) TO stdout;",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname,
! AH->public.remoteVersion),
tdinfo->filtercond);
}
else
{
appendPQExpBuffer(q, "COPY %s %s TO stdout;",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname,
! AH->public.remoteVersion),
column_list);
}
! res = PQexec(AH->connection, q->data);
! check_sql_result(res, AH->connection, q->data, PGRES_COPY_OUT);
PQclear(res);
+ destroyPQExpBuffer(clistBuf);
for (;;)
{
! ret = PQgetCopyData(AH->connection, ©buf, 0);
if (ret < 0)
break; /* done or error */
*************** dumpTableData_copy(Archive *fout, void *
*** 1385,1398 ****
{
/* copy data transfer failed */
write_msg(NULL, "Dumping the contents of table \"%s\" failed: PQgetCopyData() failed.\n", classname);
! write_msg(NULL, "Error message from server: %s", PQerrorMessage(g_conn));
write_msg(NULL, "The command was: %s\n", q->data);
exit_nicely();
}
/* Check command status and return to normal libpq state */
! res = PQgetResult(g_conn);
! check_sql_result(res, g_conn, q->data, PGRES_COMMAND_OK);
PQclear(res);
destroyPQExpBuffer(q);
--- 1538,1551 ----
{
/* copy data transfer failed */
write_msg(NULL, "Dumping the contents of table \"%s\" failed: PQgetCopyData() failed.\n", classname);
! write_msg(NULL, "Error message from server: %s", PQerrorMessage(AH->connection));
write_msg(NULL, "The command was: %s\n", q->data);
exit_nicely();
}
/* Check command status and return to normal libpq state */
! res = PQgetResult(AH->connection);
! check_sql_result(res, AH->connection, q->data, PGRES_COMMAND_OK);
PQclear(res);
destroyPQExpBuffer(q);
*************** dumpTableData_copy(Archive *fout, void *
*** 1410,1415 ****
--- 1563,1573 ----
static int
dumpTableData_insert(Archive *fout, void *dcontext)
{
+ /*
+ * This is a data dumper routine, executed in a child for parallel backup, so
+ * it must not access the global g_conn but AH->connection instead.
+ */
+ ArchiveHandle *AH = (ArchiveHandle *) fout;
TableDataInfo *tdinfo = (TableDataInfo *) dcontext;
TableInfo *tbinfo = tdinfo->tdtable;
const char *classname = tbinfo->dobj.name;
*************** dumpTableData_insert(Archive *fout, void
*** 1425,1458 ****
* this ensures reproducible results in case the table contains regproc,
* regclass, etc columns.
*/
! selectSourceSchema(tbinfo->dobj.namespace->dobj.name);
if (fout->remoteVersion >= 70100)
{
appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
"SELECT * FROM ONLY %s",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname));
}
else
{
appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
"SELECT * FROM %s",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname));
}
if (tdinfo->filtercond)
appendPQExpBuffer(q, " %s", tdinfo->filtercond);
! res = PQexec(g_conn, q->data);
! check_sql_result(res, g_conn, q->data, PGRES_COMMAND_OK);
do
{
PQclear(res);
! res = PQexec(g_conn, "FETCH 100 FROM _pg_dump_cursor");
! check_sql_result(res, g_conn, "FETCH 100 FROM _pg_dump_cursor",
PGRES_TUPLES_OK);
nfields = PQnfields(res);
for (tuple = 0; tuple < PQntuples(res); tuple++)
--- 1583,1618 ----
* this ensures reproducible results in case the table contains regproc,
* regclass, etc columns.
*/
! selectSourceSchemaOnAH(AH, tbinfo->dobj.namespace->dobj.name);
if (fout->remoteVersion >= 70100)
{
appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
"SELECT * FROM ONLY %s",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname,
! AH->public.remoteVersion));
}
else
{
appendPQExpBuffer(q, "DECLARE _pg_dump_cursor CURSOR FOR "
"SELECT * FROM %s",
fmtQualifiedId(tbinfo->dobj.namespace->dobj.name,
! classname,
! AH->public.remoteVersion));
}
if (tdinfo->filtercond)
appendPQExpBuffer(q, " %s", tdinfo->filtercond);
! res = PQexec(AH->connection, q->data);
! check_sql_result(res, AH->connection, q->data, PGRES_COMMAND_OK);
do
{
PQclear(res);
! res = PQexec(AH->connection, "FETCH 100 FROM _pg_dump_cursor");
! check_sql_result(res, AH->connection, "FETCH 100 FROM _pg_dump_cursor",
PGRES_TUPLES_OK);
nfields = PQnfields(res);
for (tuple = 0; tuple < PQntuples(res); tuple++)
*************** dumpTableData_insert(Archive *fout, void
*** 1550,1556 ****
archprintf(fout, "\n\n");
! do_sql_command(g_conn, "CLOSE _pg_dump_cursor");
destroyPQExpBuffer(q);
return 1;
--- 1710,1716 ----
archprintf(fout, "\n\n");
! do_sql_command(AH->connection, "CLOSE _pg_dump_cursor");
destroyPQExpBuffer(q);
return 1;
*************** dumpTableData(Archive *fout, TableDataIn
*** 1568,1573 ****
--- 1728,1734 ----
{
TableInfo *tbinfo = tdinfo->tdtable;
PQExpBuffer copyBuf = createPQExpBuffer();
+ PQExpBuffer clistBuf = createPQExpBuffer();
DataDumperPtr dumpFn;
char *copyStmt;
*************** dumpTableData(Archive *fout, TableDataIn
*** 1583,1589 ****
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;
}
--- 1744,1750 ----
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
*** 1596,1608 ****
ArchiveEntry(fout, tdinfo->dobj.catId, tdinfo->dobj.dumpId,
tbinfo->dobj.name, tbinfo->dobj.namespace->dobj.name,
! NULL, tbinfo->rolname,
false, "TABLE DATA", SECTION_DATA,
"", "", copyStmt,
tdinfo->dobj.dependencies, tdinfo->dobj.nDeps,
dumpFn, tdinfo);
destroyPQExpBuffer(copyBuf);
}
/*
--- 1757,1770 ----
ArchiveEntry(fout, tdinfo->dobj.catId, tdinfo->dobj.dumpId,
tbinfo->dobj.name, tbinfo->dobj.namespace->dobj.name,
! NULL, tbinfo->rolname, tbinfo->relpages,
false, "TABLE DATA", SECTION_DATA,
"", "", copyStmt,
tdinfo->dobj.dependencies, tdinfo->dobj.nDeps,
dumpFn, tdinfo);
destroyPQExpBuffer(copyBuf);
+ destroyPQExpBuffer(clistBuf);
}
/*
*************** dumpDatabase(Archive *AH)
*** 1979,1984 ****
--- 2141,2147 ----
NULL, /* Namespace */
NULL, /* Tablespace */
dba, /* Owner */
+ 0, /* relpages */
false, /* with oids */
"DATABASE", /* Desc */
SECTION_PRE_DATA, /* Section */
*************** dumpDatabase(Archive *AH)
*** 2027,2033 ****
atoi(PQgetvalue(lo_res, 0, i_relfrozenxid)),
LargeObjectRelationId);
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "pg_largeobject", NULL, NULL, "",
false, "pg_largeobject", SECTION_PRE_DATA,
loOutQry->data, "", NULL,
NULL, 0,
--- 2190,2196 ----
atoi(PQgetvalue(lo_res, 0, i_relfrozenxid)),
LargeObjectRelationId);
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "pg_largeobject", NULL, NULL, "", 0,
false, "pg_largeobject", SECTION_PRE_DATA,
loOutQry->data, "", NULL,
NULL, 0,
*************** dumpDatabase(Archive *AH)
*** 2066,2072 ****
atoi(PQgetvalue(lo_res, 0, i_relfrozenxid)),
LargeObjectMetadataRelationId);
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "pg_largeobject_metadata", NULL, NULL, "",
false, "pg_largeobject_metadata", SECTION_PRE_DATA,
loOutQry->data, "", NULL,
NULL, 0,
--- 2229,2235 ----
atoi(PQgetvalue(lo_res, 0, i_relfrozenxid)),
LargeObjectMetadataRelationId);
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "pg_largeobject_metadata", NULL, NULL, "", 0,
false, "pg_largeobject_metadata", SECTION_PRE_DATA,
loOutQry->data, "", NULL,
NULL, 0,
*************** dumpDatabase(Archive *AH)
*** 2101,2107 ****
appendPQExpBuffer(dbQry, ";\n");
ArchiveEntry(AH, dbCatId, createDumpId(), datname, NULL, NULL,
! dba, false, "COMMENT", SECTION_NONE,
dbQry->data, "", NULL,
&dbDumpId, 1, NULL, NULL);
}
--- 2264,2270 ----
appendPQExpBuffer(dbQry, ";\n");
ArchiveEntry(AH, dbCatId, createDumpId(), datname, NULL, NULL,
! dba, 0, false, "COMMENT", SECTION_NONE,
dbQry->data, "", NULL,
&dbDumpId, 1, NULL, NULL);
}
*************** dumpDatabase(Archive *AH)
*** 2128,2134 ****
emitShSecLabels(g_conn, res, seclabelQry, "DATABASE", datname);
if (strlen(seclabelQry->data))
ArchiveEntry(AH, dbCatId, createDumpId(), datname, NULL, NULL,
! dba, false, "SECURITY LABEL", SECTION_NONE,
seclabelQry->data, "", NULL,
&dbDumpId, 1, NULL, NULL);
destroyPQExpBuffer(seclabelQry);
--- 2291,2297 ----
emitShSecLabels(g_conn, res, seclabelQry, "DATABASE", datname);
if (strlen(seclabelQry->data))
ArchiveEntry(AH, dbCatId, createDumpId(), datname, NULL, NULL,
! dba, 0, false, "SECURITY LABEL", SECTION_NONE,
seclabelQry->data, "", NULL,
&dbDumpId, 1, NULL, NULL);
destroyPQExpBuffer(seclabelQry);
*************** dumpEncoding(Archive *AH)
*** 2157,2163 ****
appendPQExpBuffer(qry, ";\n");
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "ENCODING", NULL, NULL, "",
false, "ENCODING", SECTION_PRE_DATA,
qry->data, "", NULL,
NULL, 0,
--- 2320,2326 ----
appendPQExpBuffer(qry, ";\n");
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "ENCODING", NULL, NULL, "", 0,
false, "ENCODING", SECTION_PRE_DATA,
qry->data, "", NULL,
NULL, 0,
*************** dumpStdStrings(Archive *AH)
*** 2184,2190 ****
stdstrings);
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "STDSTRINGS", NULL, NULL, "",
false, "STDSTRINGS", SECTION_PRE_DATA,
qry->data, "", NULL,
NULL, 0,
--- 2347,2353 ----
stdstrings);
ArchiveEntry(AH, nilCatalogId, createDumpId(),
! "STDSTRINGS", NULL, NULL, "", 0,
false, "STDSTRINGS", SECTION_PRE_DATA,
qry->data, "", NULL,
NULL, 0,
*************** dumpBlob(Archive *AH, BlobInfo *binfo)
*** 2296,2302 ****
ArchiveEntry(AH, binfo->dobj.catId, binfo->dobj.dumpId,
binfo->dobj.name,
NULL, NULL,
! binfo->rolname, false,
"BLOB", SECTION_PRE_DATA,
cquery->data, dquery->data, NULL,
binfo->dobj.dependencies, binfo->dobj.nDeps,
--- 2459,2465 ----
ArchiveEntry(AH, binfo->dobj.catId, binfo->dobj.dumpId,
binfo->dobj.name,
NULL, NULL,
! binfo->rolname, 0, false,
"BLOB", SECTION_PRE_DATA,
cquery->data, dquery->data, NULL,
binfo->dobj.dependencies, binfo->dobj.nDeps,
*************** dumpBlob(Archive *AH, BlobInfo *binfo)
*** 2331,2338 ****
* dump the data contents of all large objects
*/
static int
! dumpBlobs(Archive *AH, void *arg)
{
const char *blobQry;
const char *blobFetchQry;
PGresult *res;
--- 2494,2506 ----
* dump the data contents of all large objects
*/
static int
! dumpBlobs(Archive *AHX, void *arg)
{
+ /*
+ * This is a data dumper routine, executed in a child for parallel backup,
+ * so it must not access the global g_conn but AH->connection instead.
+ */
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
const char *blobQry;
const char *blobFetchQry;
PGresult *res;
*************** dumpBlobs(Archive *AH, void *arg)
*** 2345,2365 ****
write_msg(NULL, "saving large objects\n");
/* Make sure we are in proper schema */
! selectSourceSchema("pg_catalog");
/*
* Currently, we re-fetch all BLOB OIDs using a cursor. Consider scanning
* the already-in-memory dumpable objects instead...
*/
! if (AH->remoteVersion >= 90000)
blobQry = "DECLARE bloboid CURSOR FOR SELECT oid FROM pg_largeobject_metadata";
! else if (AH->remoteVersion >= 70100)
blobQry = "DECLARE bloboid CURSOR FOR SELECT DISTINCT loid FROM pg_largeobject";
else
blobQry = "DECLARE bloboid CURSOR FOR SELECT oid FROM pg_class WHERE relkind = 'l'";
! res = PQexec(g_conn, blobQry);
! check_sql_result(res, g_conn, blobQry, PGRES_COMMAND_OK);
/* Command to fetch from cursor */
blobFetchQry = "FETCH 1000 IN bloboid";
--- 2513,2533 ----
write_msg(NULL, "saving large objects\n");
/* Make sure we are in proper schema */
! selectSourceSchemaOnAH(AH, "pg_catalog");
/*
* Currently, we re-fetch all BLOB OIDs using a cursor. Consider scanning
* the already-in-memory dumpable objects instead...
*/
! if (AH->public.remoteVersion >= 90000)
blobQry = "DECLARE bloboid CURSOR FOR SELECT oid FROM pg_largeobject_metadata";
! else if (AH->public.remoteVersion >= 70100)
blobQry = "DECLARE bloboid CURSOR FOR SELECT DISTINCT loid FROM pg_largeobject";
else
blobQry = "DECLARE bloboid CURSOR FOR SELECT oid FROM pg_class WHERE relkind = 'l'";
! res = PQexec(AH->connection, blobQry);
! check_sql_result(res, AH->connection, blobQry, PGRES_COMMAND_OK);
/* Command to fetch from cursor */
blobFetchQry = "FETCH 1000 IN bloboid";
*************** dumpBlobs(Archive *AH, void *arg)
*** 2369,2376 ****
PQclear(res);
/* Do a fetch */
! res = PQexec(g_conn, blobFetchQry);
! check_sql_result(res, g_conn, blobFetchQry, PGRES_TUPLES_OK);
/* Process the tuples, if any */
ntups = PQntuples(res);
--- 2537,2544 ----
PQclear(res);
/* Do a fetch */
! res = PQexec(AH->connection, blobFetchQry);
! check_sql_result(res, AH->connection, blobFetchQry, PGRES_TUPLES_OK);
/* Process the tuples, if any */
ntups = PQntuples(res);
*************** dumpBlobs(Archive *AH, void *arg)
*** 2381,2413 ****
blobOid = atooid(PQgetvalue(res, i, 0));
/* Open the BLOB */
! loFd = lo_open(g_conn, blobOid, INV_READ);
if (loFd == -1)
{
write_msg(NULL, "could not open large object %u: %s",
! blobOid, PQerrorMessage(g_conn));
exit_nicely();
}
! StartBlob(AH, blobOid);
/* Now read it in chunks, sending data to archive */
do
{
! cnt = lo_read(g_conn, loFd, buf, LOBBUFSIZE);
if (cnt < 0)
{
write_msg(NULL, "error reading large object %u: %s",
! blobOid, PQerrorMessage(g_conn));
exit_nicely();
}
! WriteData(AH, buf, cnt);
} while (cnt > 0);
! lo_close(g_conn, loFd);
! EndBlob(AH, blobOid);
}
} while (ntups > 0);
--- 2549,2583 ----
blobOid = atooid(PQgetvalue(res, i, 0));
/* Open the BLOB */
! loFd = lo_open(AH->connection, blobOid, INV_READ);
if (loFd == -1)
{
write_msg(NULL, "could not open large object %u: %s",
! blobOid, PQerrorMessage(AH->connection));
exit_nicely();
}
! StartBlob(AHX, blobOid);
/* Now read it in chunks, sending data to archive */
do
{
! cnt = lo_read(AH->connection, loFd, buf, LOBBUFSIZE);
if (cnt < 0)
{
write_msg(NULL, "error reading large object %u: %s",
! blobOid, PQerrorMessage(AH->connection));
exit_nicely();
}
! /* we try to avoid writing empty chunks */
! if (cnt > 0)
! WriteData(AHX, buf, cnt);
} while (cnt > 0);
! lo_close(AH->connection, loFd);
! EndBlob(AHX, blobOid);
}
} while (ntups > 0);
*************** getTables(int *numTables)
*** 3929,3934 ****
--- 4099,4105 ----
int i_reloptions;
int i_toastreloptions;
int i_reloftype;
+ int i_relpages;
/* Make sure we are in proper schema */
selectSourceSchema("pg_catalog");
*************** getTables(int *numTables)
*** 3968,3973 ****
--- 4139,4145 ----
"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(int *numTables)
*** 4004,4009 ****
--- 4176,4182 ----
"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(int *numTables)
*** 4039,4044 ****
--- 4212,4218 ----
"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(int *numTables)
*** 4074,4079 ****
--- 4248,4254 ----
"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(int *numTables)
*** 4110,4115 ****
--- 4285,4291 ----
"0 AS toid, "
"0 AS tfrozenxid, "
"'p' AS relpersistence, "
+ "relpages, "
"NULL AS reloftype, "
"d.refobjid AS owning_tab, "
"d.refobjsubid AS owning_col, "
*************** getTables(int *numTables)
*** 4145,4150 ****
--- 4321,4327 ----
"0 AS toid, "
"0 AS tfrozenxid, "
"'p' AS relpersistence, "
+ "relpages, "
"NULL AS reloftype, "
"d.refobjid AS owning_tab, "
"d.refobjsubid AS owning_col, "
*************** getTables(int *numTables)
*** 4176,4181 ****
--- 4353,4359 ----
"0 AS toid, "
"0 AS tfrozenxid, "
"'p' AS relpersistence, "
+ "relpages, "
"NULL AS reloftype, "
"NULL::oid AS owning_tab, "
"NULL::int4 AS owning_col, "
*************** getTables(int *numTables)
*** 4202,4207 ****
--- 4380,4386 ----
"0 AS toid, "
"0 AS tfrozenxid, "
"'p' AS relpersistence, "
+ "relpages, "
"NULL AS reloftype, "
"NULL::oid AS owning_tab, "
"NULL::int4 AS owning_col, "
*************** getTables(int *numTables)
*** 4238,4243 ****
--- 4417,4423 ----
"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(int *numTables)
*** 4292,4297 ****
--- 4472,4478 ----
i_reloptions = PQfnumber(res, "reloptions");
i_toastreloptions = PQfnumber(res, "toast_reloptions");
i_reloftype = PQfnumber(res, "reloftype");
+ i_relpages = PQfnumber(res, "relpages");
if (lockWaitTimeout && g_fout->remoteVersion >= 70300)
{
*************** getTables(int *numTables)
*** 4346,4351 ****
--- 4527,4533 ----
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(int *numTables)
*** 4375,4381 ****
appendPQExpBuffer(query,
"LOCK TABLE %s IN ACCESS SHARE MODE",
fmtQualifiedId(tblinfo[i].dobj.namespace->dobj.name,
! tblinfo[i].dobj.name));
do_sql_command(g_conn, query->data);
}
--- 4557,4564 ----
appendPQExpBuffer(query,
"LOCK TABLE %s IN ACCESS SHARE MODE",
fmtQualifiedId(tblinfo[i].dobj.namespace->dobj.name,
! tblinfo[i].dobj.name,
! g_fout->remoteVersion));
do_sql_command(g_conn, query->data);
}
*************** dumpComment(Archive *fout, const char *t
*** 6874,6880 ****
* post-data.
*/
ArchiveEntry(fout, nilCatalogId, createDumpId(),
! target, namespace, NULL, owner,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(dumpId), 1,
--- 7057,7063 ----
* post-data.
*/
ArchiveEntry(fout, nilCatalogId, createDumpId(),
! target, namespace, NULL, owner, 0,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(dumpId), 1,
*************** dumpTableComment(Archive *fout, TableInf
*** 6935,6941 ****
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target->data,
tbinfo->dobj.namespace->dobj.name,
! NULL, tbinfo->rolname,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
--- 7118,7124 ----
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target->data,
tbinfo->dobj.namespace->dobj.name,
! NULL, tbinfo->rolname, 0,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
*************** dumpTableComment(Archive *fout, TableInf
*** 6957,6963 ****
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target->data,
tbinfo->dobj.namespace->dobj.name,
! NULL, tbinfo->rolname,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
--- 7140,7146 ----
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target->data,
tbinfo->dobj.namespace->dobj.name,
! NULL, tbinfo->rolname, 0,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
*************** dumpDumpableObject(Archive *fout, Dumpab
*** 7265,7271 ****
break;
case DO_BLOB_DATA:
ArchiveEntry(fout, dobj->catId, dobj->dumpId,
! dobj->name, NULL, NULL, "",
false, "BLOBS", SECTION_DATA,
"", "", NULL,
dobj->dependencies, dobj->nDeps,
--- 7448,7454 ----
break;
case DO_BLOB_DATA:
ArchiveEntry(fout, dobj->catId, dobj->dumpId,
! dobj->name, NULL, NULL, "", 0,
false, "BLOBS", SECTION_DATA,
"", "", NULL,
dobj->dependencies, dobj->nDeps,
*************** dumpNamespace(Archive *fout, NamespaceIn
*** 7312,7318 ****
ArchiveEntry(fout, nspinfo->dobj.catId, nspinfo->dobj.dumpId,
nspinfo->dobj.name,
NULL, NULL,
! nspinfo->rolname,
false, "SCHEMA", SECTION_PRE_DATA,
q->data, delq->data, NULL,
nspinfo->dobj.dependencies, nspinfo->dobj.nDeps,
--- 7495,7501 ----
ArchiveEntry(fout, nspinfo->dobj.catId, nspinfo->dobj.dumpId,
nspinfo->dobj.name,
NULL, NULL,
! nspinfo->rolname, 0,
false, "SCHEMA", SECTION_PRE_DATA,
q->data, delq->data, NULL,
nspinfo->dobj.dependencies, nspinfo->dobj.nDeps,
*************** dumpExtension(Archive *fout, ExtensionIn
*** 7430,7436 ****
ArchiveEntry(fout, extinfo->dobj.catId, extinfo->dobj.dumpId,
extinfo->dobj.name,
NULL, NULL,
! "",
false, "EXTENSION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
extinfo->dobj.dependencies, extinfo->dobj.nDeps,
--- 7613,7619 ----
ArchiveEntry(fout, extinfo->dobj.catId, extinfo->dobj.dumpId,
extinfo->dobj.name,
NULL, NULL,
! "", 0,
false, "EXTENSION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
extinfo->dobj.dependencies, extinfo->dobj.nDeps,
*************** dumpEnumType(Archive *fout, TypeInfo *ty
*** 7578,7584 ****
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
--- 7761,7767 ----
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, 0, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
*************** dumpRangeType(Archive *fout, TypeInfo *t
*** 7709,7715 ****
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
--- 7892,7898 ----
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, 0, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
*************** dumpBaseType(Archive *fout, TypeInfo *ty
*** 8103,8109 ****
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
--- 8286,8292 ----
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, 0, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
*************** dumpDomain(Archive *fout, TypeInfo *tyin
*** 8270,8276 ****
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, false,
"DOMAIN", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
--- 8453,8459 ----
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, 0, false,
"DOMAIN", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
*************** dumpCompositeType(Archive *fout, TypeInf
*** 8477,8483 ****
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
--- 8660,8666 ----
tyinfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, 0, false,
"TYPE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tyinfo->dobj.dependencies, tyinfo->dobj.nDeps,
*************** dumpCompositeTypeColComments(Archive *fo
*** 8597,8603 ****
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target->data,
tyinfo->dobj.namespace->dobj.name,
! NULL, tyinfo->rolname,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(tyinfo->dobj.dumpId), 1,
--- 8780,8786 ----
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target->data,
tyinfo->dobj.namespace->dobj.name,
! NULL, tyinfo->rolname, 0,
false, "COMMENT", SECTION_NONE,
query->data, "", NULL,
&(tyinfo->dobj.dumpId), 1,
*************** dumpShellType(Archive *fout, ShellTypeIn
*** 8650,8656 ****
stinfo->dobj.name,
stinfo->dobj.namespace->dobj.name,
NULL,
! stinfo->baseType->rolname, false,
"SHELL TYPE", SECTION_PRE_DATA,
q->data, "", NULL,
stinfo->dobj.dependencies, stinfo->dobj.nDeps,
--- 8833,8839 ----
stinfo->dobj.name,
stinfo->dobj.namespace->dobj.name,
NULL,
! stinfo->baseType->rolname, 0, false,
"SHELL TYPE", SECTION_PRE_DATA,
q->data, "", NULL,
stinfo->dobj.dependencies, stinfo->dobj.nDeps,
*************** dumpProcLang(Archive *fout, ProcLangInfo
*** 8824,8830 ****
ArchiveEntry(fout, plang->dobj.catId, plang->dobj.dumpId,
plang->dobj.name,
! lanschema, NULL, plang->lanowner,
false, "PROCEDURAL LANGUAGE", SECTION_PRE_DATA,
defqry->data, delqry->data, NULL,
plang->dobj.dependencies, plang->dobj.nDeps,
--- 9007,9013 ----
ArchiveEntry(fout, plang->dobj.catId, plang->dobj.dumpId,
plang->dobj.name,
! lanschema, NULL, plang->lanowner, 0,
false, "PROCEDURAL LANGUAGE", SECTION_PRE_DATA,
defqry->data, delqry->data, NULL,
plang->dobj.dependencies, plang->dobj.nDeps,
*************** dumpFunc(Archive *fout, FuncInfo *finfo)
*** 9394,9400 ****
funcsig_tag,
finfo->dobj.namespace->dobj.name,
NULL,
! finfo->rolname, false,
"FUNCTION", SECTION_PRE_DATA,
q->data, delqry->data, NULL,
finfo->dobj.dependencies, finfo->dobj.nDeps,
--- 9577,9583 ----
funcsig_tag,
finfo->dobj.namespace->dobj.name,
NULL,
! finfo->rolname, 0, false,
"FUNCTION", SECTION_PRE_DATA,
q->data, delqry->data, NULL,
finfo->dobj.dependencies, finfo->dobj.nDeps,
*************** dumpCast(Archive *fout, CastInfo *cast)
*** 9558,9564 ****
ArchiveEntry(fout, cast->dobj.catId, cast->dobj.dumpId,
labelq->data,
! "pg_catalog", NULL, "",
false, "CAST", SECTION_PRE_DATA,
defqry->data, delqry->data, NULL,
cast->dobj.dependencies, cast->dobj.nDeps,
--- 9741,9747 ----
ArchiveEntry(fout, cast->dobj.catId, cast->dobj.dumpId,
labelq->data,
! "pg_catalog", NULL, "", 0,
false, "CAST", SECTION_PRE_DATA,
defqry->data, delqry->data, NULL,
cast->dobj.dependencies, cast->dobj.nDeps,
*************** dumpOpr(Archive *fout, OprInfo *oprinfo)
*** 9805,9811 ****
oprinfo->dobj.name,
oprinfo->dobj.namespace->dobj.name,
NULL,
! oprinfo->rolname,
false, "OPERATOR", SECTION_PRE_DATA,
q->data, delq->data, NULL,
oprinfo->dobj.dependencies, oprinfo->dobj.nDeps,
--- 9988,9994 ----
oprinfo->dobj.name,
oprinfo->dobj.namespace->dobj.name,
NULL,
! oprinfo->rolname, 0,
false, "OPERATOR", SECTION_PRE_DATA,
q->data, delq->data, NULL,
oprinfo->dobj.dependencies, oprinfo->dobj.nDeps,
*************** dumpOpclass(Archive *fout, OpclassInfo *
*** 10338,10344 ****
opcinfo->dobj.name,
opcinfo->dobj.namespace->dobj.name,
NULL,
! opcinfo->rolname,
false, "OPERATOR CLASS", SECTION_PRE_DATA,
q->data, delq->data, NULL,
opcinfo->dobj.dependencies, opcinfo->dobj.nDeps,
--- 10521,10527 ----
opcinfo->dobj.name,
opcinfo->dobj.namespace->dobj.name,
NULL,
! opcinfo->rolname, 0,
false, "OPERATOR CLASS", SECTION_PRE_DATA,
q->data, delq->data, NULL,
opcinfo->dobj.dependencies, opcinfo->dobj.nDeps,
*************** dumpOpfamily(Archive *fout, OpfamilyInfo
*** 10666,10672 ****
opfinfo->dobj.name,
opfinfo->dobj.namespace->dobj.name,
NULL,
! opfinfo->rolname,
false, "OPERATOR FAMILY", SECTION_PRE_DATA,
q->data, delq->data, NULL,
opfinfo->dobj.dependencies, opfinfo->dobj.nDeps,
--- 10849,10855 ----
opfinfo->dobj.name,
opfinfo->dobj.namespace->dobj.name,
NULL,
! opfinfo->rolname, 0,
false, "OPERATOR FAMILY", SECTION_PRE_DATA,
q->data, delq->data, NULL,
opfinfo->dobj.dependencies, opfinfo->dobj.nDeps,
*************** dumpCollation(Archive *fout, CollInfo *c
*** 10768,10774 ****
collinfo->dobj.name,
collinfo->dobj.namespace->dobj.name,
NULL,
! collinfo->rolname,
false, "COLLATION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
collinfo->dobj.dependencies, collinfo->dobj.nDeps,
--- 10951,10957 ----
collinfo->dobj.name,
collinfo->dobj.namespace->dobj.name,
NULL,
! collinfo->rolname, 0,
false, "COLLATION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
collinfo->dobj.dependencies, collinfo->dobj.nDeps,
*************** dumpConversion(Archive *fout, ConvInfo *
*** 10880,10886 ****
convinfo->dobj.name,
convinfo->dobj.namespace->dobj.name,
NULL,
! convinfo->rolname,
false, "CONVERSION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
convinfo->dobj.dependencies, convinfo->dobj.nDeps,
--- 11063,11069 ----
convinfo->dobj.name,
convinfo->dobj.namespace->dobj.name,
NULL,
! convinfo->rolname, 0,
false, "CONVERSION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
convinfo->dobj.dependencies, convinfo->dobj.nDeps,
*************** dumpAgg(Archive *fout, AggInfo *agginfo)
*** 11129,11135 ****
aggsig_tag,
agginfo->aggfn.dobj.namespace->dobj.name,
NULL,
! agginfo->aggfn.rolname,
false, "AGGREGATE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
agginfo->aggfn.dobj.dependencies, agginfo->aggfn.dobj.nDeps,
--- 11312,11318 ----
aggsig_tag,
agginfo->aggfn.dobj.namespace->dobj.name,
NULL,
! agginfo->aggfn.rolname, 0,
false, "AGGREGATE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
agginfo->aggfn.dobj.dependencies, agginfo->aggfn.dobj.nDeps,
*************** dumpTSParser(Archive *fout, TSParserInfo
*** 11228,11233 ****
--- 11411,11417 ----
prsinfo->dobj.namespace->dobj.name,
NULL,
"",
+ 0,
false, "TEXT SEARCH PARSER", SECTION_PRE_DATA,
q->data, delq->data, NULL,
prsinfo->dobj.dependencies, prsinfo->dobj.nDeps,
*************** dumpTSDictionary(Archive *fout, TSDictIn
*** 11326,11331 ****
--- 11510,11516 ----
dictinfo->dobj.namespace->dobj.name,
NULL,
dictinfo->rolname,
+ 0,
false, "TEXT SEARCH DICTIONARY", SECTION_PRE_DATA,
q->data, delq->data, NULL,
dictinfo->dobj.dependencies, dictinfo->dobj.nDeps,
*************** dumpTSTemplate(Archive *fout, TSTemplate
*** 11392,11397 ****
--- 11577,11583 ----
tmplinfo->dobj.namespace->dobj.name,
NULL,
"",
+ 0,
false, "TEXT SEARCH TEMPLATE", SECTION_PRE_DATA,
q->data, delq->data, NULL,
tmplinfo->dobj.dependencies, tmplinfo->dobj.nDeps,
*************** dumpTSConfig(Archive *fout, TSConfigInfo
*** 11531,11536 ****
--- 11717,11723 ----
cfginfo->dobj.namespace->dobj.name,
NULL,
cfginfo->rolname,
+ 0,
false, "TEXT SEARCH CONFIGURATION", SECTION_PRE_DATA,
q->data, delq->data, NULL,
cfginfo->dobj.dependencies, cfginfo->dobj.nDeps,
*************** dumpForeignDataWrapper(Archive *fout, Fd
*** 11605,11610 ****
--- 11792,11798 ----
NULL,
NULL,
fdwinfo->rolname,
+ 0,
false, "FOREIGN DATA WRAPPER", SECTION_PRE_DATA,
q->data, delq->data, NULL,
fdwinfo->dobj.dependencies, fdwinfo->dobj.nDeps,
*************** dumpForeignServer(Archive *fout, Foreign
*** 11708,11713 ****
--- 11896,11902 ----
NULL,
NULL,
srvinfo->rolname,
+ 0,
false, "SERVER", SECTION_PRE_DATA,
q->data, delq->data, NULL,
srvinfo->dobj.dependencies, srvinfo->dobj.nDeps,
*************** dumpUserMappings(Archive *fout,
*** 11825,11831 ****
tag->data,
namespace,
NULL,
! owner, false,
"USER MAPPING", SECTION_PRE_DATA,
q->data, delq->data, NULL,
&dumpId, 1,
--- 12014,12020 ----
tag->data,
namespace,
NULL,
! owner, 0, false,
"USER MAPPING", SECTION_PRE_DATA,
q->data, delq->data, NULL,
&dumpId, 1,
*************** dumpDefaultACL(Archive *fout, DefaultACL
*** 11896,11901 ****
--- 12085,12091 ----
daclinfo->dobj.namespace ? daclinfo->dobj.namespace->dobj.name : NULL,
NULL,
daclinfo->defaclrole,
+ 0,
false, "DEFAULT ACL", SECTION_NONE,
q->data, "", NULL,
daclinfo->dobj.dependencies, daclinfo->dobj.nDeps,
*************** dumpACL(Archive *fout, CatalogId objCatI
*** 11953,11958 ****
--- 12143,12149 ----
tag, nspname,
NULL,
owner ? owner : "",
+ 0,
false, "ACL", SECTION_NONE,
sql->data, "", NULL,
&(objDumpId), 1,
*************** dumpSecLabel(Archive *fout, const char *
*** 12029,12035 ****
{
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target, namespace, NULL, owner,
! false, "SECURITY LABEL", SECTION_NONE,
query->data, "", NULL,
&(dumpId), 1,
NULL, NULL);
--- 12220,12226 ----
{
ArchiveEntry(fout, nilCatalogId, createDumpId(),
target, namespace, NULL, owner,
! 0, false, "SECURITY LABEL", SECTION_NONE,
query->data, "", NULL,
&(dumpId), 1,
NULL, NULL);
*************** dumpTableSecLabel(Archive *fout, TableIn
*** 12107,12113 ****
target->data,
tbinfo->dobj.namespace->dobj.name,
NULL, tbinfo->rolname,
! false, "SECURITY LABEL", SECTION_NONE,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
NULL, NULL);
--- 12298,12304 ----
target->data,
tbinfo->dobj.namespace->dobj.name,
NULL, tbinfo->rolname,
! 0, false, "SECURITY LABEL", SECTION_NONE,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
NULL, NULL);
*************** dumpTableSchema(Archive *fout, TableInfo
*** 12888,12893 ****
--- 13079,13085 ----
tbinfo->dobj.namespace->dobj.name,
(tbinfo->relkind == RELKIND_VIEW) ? NULL : tbinfo->reltablespace,
tbinfo->rolname,
+ 0,
(strcmp(reltypename, "TABLE") == 0) ? tbinfo->hasoids : false,
reltypename, SECTION_PRE_DATA,
q->data, delq->data, NULL,
*************** dumpAttrDef(Archive *fout, AttrDefInfo *
*** 12961,12966 ****
--- 13153,13159 ----
tbinfo->dobj.namespace->dobj.name,
NULL,
tbinfo->rolname,
+ 0,
false, "DEFAULT", SECTION_PRE_DATA,
q->data, delq->data, NULL,
adinfo->dobj.dependencies, adinfo->dobj.nDeps,
*************** dumpIndex(Archive *fout, IndxInfo *indxi
*** 13062,13068 ****
indxinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
indxinfo->tablespace,
! tbinfo->rolname, false,
"INDEX", SECTION_POST_DATA,
q->data, delq->data, NULL,
indxinfo->dobj.dependencies, indxinfo->dobj.nDeps,
--- 13255,13261 ----
indxinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
indxinfo->tablespace,
! tbinfo->rolname, indxinfo->relpages, false,
"INDEX", SECTION_POST_DATA,
q->data, delq->data, NULL,
indxinfo->dobj.dependencies, indxinfo->dobj.nDeps,
*************** dumpConstraint(Archive *fout, Constraint
*** 13185,13191 ****
coninfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
indxinfo->tablespace,
! tbinfo->rolname, false,
"CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
--- 13378,13384 ----
coninfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
indxinfo->tablespace,
! tbinfo->rolname, 0, false,
"CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
*************** dumpConstraint(Archive *fout, Constraint
*** 13218,13224 ****
coninfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, false,
"FK CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
--- 13411,13417 ----
coninfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, 0, false,
"FK CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
*************** dumpConstraint(Archive *fout, Constraint
*** 13253,13259 ****
coninfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, false,
"CHECK CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
--- 13446,13452 ----
coninfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, 0, false,
"CHECK CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
*************** dumpConstraint(Archive *fout, Constraint
*** 13289,13295 ****
coninfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, false,
"CHECK CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
--- 13482,13488 ----
coninfo->dobj.name,
tyinfo->dobj.namespace->dobj.name,
NULL,
! tyinfo->rolname, 0, false,
"CHECK CONSTRAINT", SECTION_POST_DATA,
q->data, delq->data, NULL,
coninfo->dobj.dependencies, coninfo->dobj.nDeps,
*************** dumpSequence(Archive *fout, TableInfo *t
*** 13579,13585 ****
tbinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname,
false, "SEQUENCE", SECTION_PRE_DATA,
query->data, delqry->data, NULL,
tbinfo->dobj.dependencies, tbinfo->dobj.nDeps,
--- 13772,13778 ----
tbinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, 0,
false, "SEQUENCE", SECTION_PRE_DATA,
query->data, delqry->data, NULL,
tbinfo->dobj.dependencies, tbinfo->dobj.nDeps,
*************** dumpSequence(Archive *fout, TableInfo *t
*** 13615,13621 ****
tbinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname,
false, "SEQUENCE OWNED BY", SECTION_PRE_DATA,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
--- 13808,13814 ----
tbinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, 0,
false, "SEQUENCE OWNED BY", SECTION_PRE_DATA,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
*************** dumpSequence(Archive *fout, TableInfo *t
*** 13645,13650 ****
--- 13838,13844 ----
tbinfo->dobj.namespace->dobj.name,
NULL,
tbinfo->rolname,
+ 0,
false, "SEQUENCE SET", SECTION_PRE_DATA,
query->data, "", NULL,
&(tbinfo->dobj.dumpId), 1,
*************** dumpTrigger(Archive *fout, TriggerInfo *
*** 13845,13851 ****
tginfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, false,
"TRIGGER", SECTION_POST_DATA,
query->data, delqry->data, NULL,
tginfo->dobj.dependencies, tginfo->dobj.nDeps,
--- 14039,14045 ----
tginfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, 0, false,
"TRIGGER", SECTION_POST_DATA,
query->data, delqry->data, NULL,
tginfo->dobj.dependencies, tginfo->dobj.nDeps,
*************** dumpRule(Archive *fout, RuleInfo *rinfo)
*** 13967,13973 ****
rinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, false,
"RULE", SECTION_POST_DATA,
cmd->data, delcmd->data, NULL,
rinfo->dobj.dependencies, rinfo->dobj.nDeps,
--- 14161,14167 ----
rinfo->dobj.name,
tbinfo->dobj.namespace->dobj.name,
NULL,
! tbinfo->rolname, 0, false,
"RULE", SECTION_POST_DATA,
cmd->data, delcmd->data, NULL,
rinfo->dobj.dependencies, rinfo->dobj.nDeps,
*************** getDependencies(void)
*** 14268,14273 ****
--- 14462,14509 ----
destroyPQExpBuffer(query);
}
+ /*
+ * Select the source schema and use a static var to remember what we have set
+ * as the default schema right now. This function is never called in parallel
+ * context, so the static var is okay. The parallel context will call
+ * selectSourceSchemaOnAH, and this remembers the current schema via
+ * AH->currSchema.
+ */
+ static void
+ selectSourceSchema(const char *schemaName)
+ {
+ static char *currSchemaName = NULL;
+
+ if (!schemaName || *schemaName == '\0' ||
+ (currSchemaName && strcmp(currSchemaName, schemaName) == 0))
+ return; /* no need to do anything */
+
+ selectSourceSchemaOnConnection(g_conn, schemaName);
+
+ if (currSchemaName)
+ free(currSchemaName);
+ currSchemaName = pg_strdup(schemaName);
+ }
+
+ /*
+ * This function lets a DataDumper function select a schema on an
+ * ArchiveHandle. These functions can be called from a threaded program for
+ * parallel dump/restore and must therefore not access global variables (read
+ * only access to g_fout->remoteVersion is okay however).
+ */
+ static void
+ selectSourceSchemaOnAH(ArchiveHandle *AH, const char *schemaName)
+ {
+ if (!schemaName || *schemaName == '\0' ||
+ (AH->currSchema && strcmp(AH->currSchema, schemaName) == 0))
+ return; /* no need to do anything */
+
+ selectSourceSchemaOnConnection(AH->connection, schemaName);
+
+ if (AH->currSchema)
+ free(AH->currSchema);
+ AH->currSchema = pg_strdup(schemaName);
+ }
/*
* selectSourceSchema - make the specified schema the active search path
*************** getDependencies(void)
*** 14280,14301 ****
*
* 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(const char *schemaName)
{
- static char *curSchemaName = NULL;
PQExpBuffer query;
/* Not relevant if fetching from pre-7.3 DB */
if (g_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",
--- 14516,14536 ----
*
* 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
! selectSourceSchemaOnConnection(PGconn *conn, 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 (g_fout->remoteVersion < 70300)
return;
query = createPQExpBuffer();
appendPQExpBuffer(query, "SET search_path = %s",
*************** selectSourceSchema(const char *schemaNam
*** 14303,14314 ****
if (strcmp(schemaName, "pg_catalog") != 0)
appendPQExpBuffer(query, ", pg_catalog");
! do_sql_command(g_conn, query->data);
destroyPQExpBuffer(query);
- if (curSchemaName)
- free(curSchemaName);
- curSchemaName = pg_strdup(schemaName);
}
/*
--- 14538,14546 ----
if (strcmp(schemaName, "pg_catalog") != 0)
appendPQExpBuffer(query, ", pg_catalog");
! do_sql_command(conn, query->data);
destroyPQExpBuffer(query);
}
/*
*************** myFormatType(const char *typname, int32
*** 14459,14529 ****
}
/*
- * 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(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 (g_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;
}
/*
--- 14691,14727 ----
}
/*
* 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 11c4d37..dba617b 100644
*** a/src/bin/pg_dump/pg_dump.h
--- b/src/bin/pg_dump/pg_dump.h
*************** typedef struct _tableInfo
*** 257,262 ****
--- 257,263 ----
/* 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
*** 328,333 ****
--- 329,335 ----
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
*** 532,537 ****
--- 534,540 ----
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 4d1ae94..1286124 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_dumpall.c b/src/bin/pg_dump/pg_dumpall.c
index 4c93667..438b8f0 100644
*** a/src/bin/pg_dump/pg_dumpall.c
--- b/src/bin/pg_dump/pg_dumpall.c
*************** doShellQuoting(PQExpBuffer buf, const ch
*** 1913,1915 ****
--- 1913,1918 ----
appendPQExpBufferChar(buf, '"');
#endif /* WIN32 */
}
+
+ /* dummy, no parallel dump/restore for pg_dumpall yet */
+ void _SetupWorker(Archive *AHX, RestoreOptions *ropt) {}
diff --git a/src/bin/pg_dump/pg_restore.c b/src/bin/pg_dump/pg_restore.c
index 6ff1ab8..965e9f2 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)
*** 338,344 ****
}
/* 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);
--- 339,345 ----
}
/* 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)
*** 403,408 ****
--- 404,420 ----
InitDummyWantedList(AH, opts);
}
+ /* See comments in pg_dump.c */
+ #ifdef WIN32
+ if (numWorkers > MAXIMUM_WAIT_OBJECTS)
+ {
+ fprintf(stderr, _("%s: invalid number of parallel jobs\n"), progname);
+ exit(1);
+ }
+ #endif
+
+ AH->numWorkers = numWorkers;
+
if (opts->tocSummary)
PrintTOCSummary(AH, opts);
else
*************** main(int argc, char **argv)
*** 421,426 ****
--- 433,445 ----
return exit_code;
}
+ void
+ _SetupWorker(Archive *AHX, RestoreOptions *ropt)
+ {
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
+ (AH->ReopenPtr) (AH);
+ }
+
static void
usage(const char *progname)
{