smarter-parallel-dump-restore-1.patch
text/x-diff
Filename: smarter-parallel-dump-restore-1.patch
Type: text/x-diff
Part: 0
Message:
Re: pg_dump test instability
Patch
Format: unified
| File | + | − |
|---|---|---|
| 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.h | 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 |
diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h
index 42cf441..ba79821 100644
*** a/src/bin/pg_dump/pg_backup.h
--- b/src/bin/pg_dump/pg_backup.h
*************** extern void ConnectDatabase(Archive *AH,
*** 252,269 ****
extern void DisconnectDatabase(Archive *AHX);
extern PGconn *GetConnection(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, bool withOids,
- const char *desc, teSection section,
- const char *defn,
- const char *dropStmt, const char *copyStmt,
- const DumpId *deps, int nDeps,
- DataDumperPtr dumpFn, void *dumpArg);
-
/* Called to write *data* to the archive */
extern void WriteData(Archive *AH, const void *data, size_t dLen);
--- 252,257 ----
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 45a391b..fce014f 100644
*** a/src/bin/pg_dump/pg_backup_archiver.c
--- b/src/bin/pg_dump/pg_backup_archiver.c
*************** typedef struct _outputContext
*** 49,54 ****
--- 49,72 ----
int gzOut;
} OutputContext;
+ /*
+ * State for tracking TocEntrys that are ready to process during a parallel
+ * restore. (This used to be a list, and we still call it that, though now
+ * it's really an array so that we can apply qsort to it.)
+ *
+ * tes[] is sized large enough that we can't overrun it.
+ * The valid entries are indexed first_te .. last_te inclusive.
+ * We periodically sort the array to bring larger-by-dataLength entries to
+ * the front; "sorted" is true if the valid entries are known sorted.
+ */
+ typedef struct _parallelReadyList
+ {
+ TocEntry **tes; /* Ready-to-dump TocEntrys */
+ int first_te; /* index of first valid entry in tes[] */
+ int last_te; /* index of last valid entry in tes[] */
+ bool sorted; /* are valid entries currently sorted? */
+ } ParallelReadyList;
+
/* translator: this is a module name */
static const char *modulename = gettext_noop("archiver");
*************** static void restore_toc_entries_postfork
*** 98,107 ****
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 void move_to_ready_list(TocEntry *pending_list, TocEntry *ready_list,
RestorePass pass);
static TocEntry *get_next_work_item(ArchiveHandle *AH,
! TocEntry *ready_list,
ParallelState *pstate);
static void mark_dump_job_done(ArchiveHandle *AH,
TocEntry *te,
--- 116,131 ----
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 void ready_list_init(ParallelReadyList *ready_list, int tocCount);
! static void ready_list_insert(ParallelReadyList *ready_list, TocEntry *te);
! static void ready_list_remove(ParallelReadyList *ready_list, int i);
! static void ready_list_sort(ParallelReadyList *ready_list);
! static int TocEntrySizeCompare(const void *p1, const void *p2);
! static void move_to_ready_list(TocEntry *pending_list,
! ParallelReadyList *ready_list,
RestorePass pass);
static TocEntry *get_next_work_item(ArchiveHandle *AH,
! ParallelReadyList *ready_list,
ParallelState *pstate);
static void mark_dump_job_done(ArchiveHandle *AH,
TocEntry *te,
*************** static bool has_lock_conflicts(TocEntry
*** 116,122 ****
static void repoint_table_dependencies(ArchiveHandle *AH);
static void identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te);
static void reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
! TocEntry *ready_list);
static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
--- 140,146 ----
static void repoint_table_dependencies(ArchiveHandle *AH);
static void identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te);
static void reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
! ParallelReadyList *ready_list);
static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
*************** RestoreArchive(Archive *AHX)
*** 639,644 ****
--- 663,672 ----
ParallelState *pstate;
TocEntry pending_list;
+ /* The archive format module may need some setup for this */
+ if (AH->PrepParallelRestorePtr)
+ AH->PrepParallelRestorePtr(AH);
+
par_list_header_init(&pending_list);
/* This runs PRE_DATA items and then disconnects from the database */
*************** WriteData(Archive *AHX, const void *data
*** 1039,1048 ****
/*
* Create a new TOC entry. The TOC was designed as a TOC, but is now the
* repository for all metadata. But the name has stuck.
*/
/* Public */
! void
ArchiveEntry(Archive *AHX,
CatalogId catalogId, DumpId dumpId,
const char *tag,
--- 1067,1080 ----
/*
* Create a new TOC entry. The TOC was designed as a TOC, but is now the
* repository for all metadata. But the name has stuck.
+ *
+ * The new entry is added to the Archive's TOC list. Most callers can ignore
+ * the result value because nothing else need be done, but a few want to
+ * manipulate the TOC entry further.
*/
/* Public */
! TocEntry *
ArchiveEntry(Archive *AHX,
CatalogId catalogId, DumpId dumpId,
const char *tag,
*************** ArchiveEntry(Archive *AHX,
*** 1100,1108 ****
--- 1132,1143 ----
newToc->hadDumper = dumpFn ? true : false;
newToc->formatData = NULL;
+ newToc->dataLength = 0;
if (AH->ArchiveEntryPtr != NULL)
AH->ArchiveEntryPtr(AH, newToc);
+
+ return newToc;
}
/* Public */
*************** WriteDataChunks(ArchiveHandle *AH, Paral
*** 2413,2444 ****
{
TocEntry *te;
! for (te = AH->toc->next; te != AH->toc; te = te->next)
{
! if (!te->dataDumper)
! continue;
!
! if ((te->reqs & REQ_DATA) == 0)
! continue;
! if (pstate && pstate->numWorkers > 1)
{
! /*
! * If we are in a parallel backup, then we are always the master
! * process. Dispatch each data-transfer job to a worker.
! */
! DispatchJobForTocEntry(AH, pstate, te, ACT_DUMP,
! mark_dump_job_done, NULL);
}
- else
- WriteDataChunksForTocEntry(AH, te);
- }
! /*
! * If parallel, wait for workers to finish.
! */
! if (pstate && pstate->numWorkers > 1)
WaitForWorkers(AH, pstate, WFW_ALL_IDLE);
}
--- 2448,2506 ----
{
TocEntry *te;
! if (pstate && pstate->numWorkers > 1)
{
! /*
! * In parallel mode, this code runs in the master process. We
! * construct an array of candidate TEs, then sort it into decreasing
! * size order, then dispatch each TE to a data-transfer worker. By
! * dumping larger tables first, we avoid getting into a situation
! * where we're down to one job and it's big, losing parallelism.
! */
! TocEntry **tes;
! int ntes;
! tes = (TocEntry **) pg_malloc(AH->tocCount * sizeof(TocEntry *));
! ntes = 0;
! for (te = AH->toc->next; te != AH->toc; te = te->next)
{
! /* Consider only TEs with dataDumper functions ... */
! if (!te->dataDumper)
! continue;
! /* ... and ignore ones not enabled for dump */
! if ((te->reqs & REQ_DATA) == 0)
! continue;
!
! tes[ntes++] = te;
}
! if (ntes > 1)
! qsort((void *) tes, ntes, sizeof(TocEntry *),
! TocEntrySizeCompare);
!
! for (int i = 0; i < ntes; i++)
! DispatchJobForTocEntry(AH, pstate, tes[i], ACT_DUMP,
! mark_dump_job_done, NULL);
!
! pg_free(tes);
!
! /* Now wait for workers to finish. */
WaitForWorkers(AH, pstate, WFW_ALL_IDLE);
+ }
+ else
+ {
+ /* Non-parallel mode: just dump all candidate TEs sequentially. */
+ for (te = AH->toc->next; te != AH->toc; te = te->next)
+ {
+ /* Must have same filter conditions as above */
+ if (!te->dataDumper)
+ continue;
+ if ((te->reqs & REQ_DATA) == 0)
+ continue;
+
+ WriteDataChunksForTocEntry(AH, te);
+ }
+ }
}
*************** ReadToc(ArchiveHandle *AH)
*** 2690,2695 ****
--- 2752,2758 ----
te->dependencies = NULL;
te->nDeps = 0;
}
+ te->dataLength = 0;
if (AH->ReadExtraTocPtr)
AH->ReadExtraTocPtr(AH, te);
*************** static void
*** 4035,4045 ****
restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
TocEntry *pending_list)
{
! TocEntry ready_list;
TocEntry *next_work_item;
ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
/*
* The pending_list contains all items that we need to restore. Move all
* items that are available to process immediately into the ready_list.
--- 4098,4111 ----
restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
TocEntry *pending_list)
{
! ParallelReadyList ready_list;
TocEntry *next_work_item;
ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
+ /* Set up ready_list with enough room for all known TocEntrys */
+ ready_list_init(&ready_list, AH->tocCount);
+
/*
* The pending_list contains all items that we need to restore. Move all
* items that are available to process immediately into the ready_list.
*************** restore_toc_entries_parallel(ArchiveHand
*** 4048,4054 ****
* contains items that have no remaining dependencies and are OK to
* process in the current restore pass.
*/
- par_list_header_init(&ready_list);
AH->restorePass = RESTORE_PASS_MAIN;
move_to_ready_list(pending_list, &ready_list, AH->restorePass);
--- 4114,4119 ----
*************** restore_toc_entries_parallel(ArchiveHand
*** 4073,4080 ****
ahlog(AH, 1, "skipping item %d %s %s\n",
next_work_item->dumpId,
next_work_item->desc, next_work_item->tag);
! /* Drop it from ready_list, and update its dependencies */
! par_list_remove(next_work_item);
reduce_dependencies(AH, next_work_item, &ready_list);
/* Loop around to see if anything else can be dispatched */
continue;
--- 4138,4144 ----
ahlog(AH, 1, "skipping item %d %s %s\n",
next_work_item->dumpId,
next_work_item->desc, next_work_item->tag);
! /* Update its dependencies as though we'd completed it */
reduce_dependencies(AH, next_work_item, &ready_list);
/* Loop around to see if anything else can be dispatched */
continue;
*************** restore_toc_entries_parallel(ArchiveHand
*** 4084,4092 ****
next_work_item->dumpId,
next_work_item->desc, next_work_item->tag);
! /* Remove it from ready_list, and dispatch to some worker */
! par_list_remove(next_work_item);
!
DispatchJobForTocEntry(AH, pstate, next_work_item, ACT_RESTORE,
mark_restore_job_done, &ready_list);
}
--- 4148,4154 ----
next_work_item->dumpId,
next_work_item->desc, next_work_item->tag);
! /* Dispatch to some worker */
DispatchJobForTocEntry(AH, pstate, next_work_item, ACT_RESTORE,
mark_restore_job_done, &ready_list);
}
*************** restore_toc_entries_parallel(ArchiveHand
*** 4132,4138 ****
}
/* There should now be nothing in ready_list. */
! Assert(ready_list.par_next == &ready_list);
ahlog(AH, 1, "finished main parallel loop\n");
}
--- 4194,4202 ----
}
/* There should now be nothing in ready_list. */
! Assert(ready_list.first_te > ready_list.last_te);
!
! pg_free(ready_list.tes);
ahlog(AH, 1, "finished main parallel loop\n");
}
*************** par_list_remove(TocEntry *te)
*** 4235,4240 ****
--- 4299,4386 ----
/*
+ * Initialize the ready_list with enough room for up to tocCount entries.
+ */
+ static void
+ ready_list_init(ParallelReadyList *ready_list, int tocCount)
+ {
+ ready_list->tes = (TocEntry **)
+ pg_malloc(tocCount * sizeof(TocEntry *));
+ ready_list->first_te = 0;
+ ready_list->last_te = -1;
+ ready_list->sorted = false;
+ }
+
+ /* Add te to the ready_list */
+ static void
+ ready_list_insert(ParallelReadyList *ready_list, TocEntry *te)
+ {
+ ready_list->tes[++ready_list->last_te] = te;
+ ready_list->sorted = false;
+ }
+
+ /* Remove the i'th entry in the ready_list */
+ static void
+ ready_list_remove(ParallelReadyList *ready_list, int i)
+ {
+ Assert(i >= ready_list->first_te && i <= ready_list->last_te);
+ if (i == ready_list->first_te)
+ {
+ /*
+ * In the common case where we dispatch the first available entry, the
+ * list can still be considered sorted.
+ */
+ ready_list->first_te++;
+ }
+ else
+ {
+ /* Must compact the list and mark it unsorted */
+ ready_list->tes[i] = ready_list->tes[ready_list->first_te];
+ ready_list->first_te++;
+ ready_list->sorted = false;
+ }
+ }
+
+ /* Sort the ready_list into the desired order */
+ static void
+ ready_list_sort(ParallelReadyList *ready_list)
+ {
+ if (!ready_list->sorted)
+ {
+ int n = ready_list->last_te - ready_list->first_te + 1;
+
+ if (n > 1)
+ qsort(ready_list->tes + ready_list->first_te, n,
+ sizeof(TocEntry *),
+ TocEntrySizeCompare);
+ ready_list->sorted = true;
+ }
+ }
+
+ /* qsort comparator for sorting TocEntries by dataLength */
+ static int
+ TocEntrySizeCompare(const void *p1, const void *p2)
+ {
+ const TocEntry *te1 = *(const TocEntry *const *) p1;
+ const TocEntry *te2 = *(const TocEntry *const *) p2;
+
+ /* Sort by decreasing dataLength */
+ if (te1->dataLength > te2->dataLength)
+ return -1;
+ if (te1->dataLength < te2->dataLength)
+ return 1;
+
+ /* For equal dataLengths, sort by dumpId, just to be stable */
+ if (te1->dumpId < te2->dumpId)
+ return -1;
+ if (te1->dumpId > te2->dumpId)
+ return 1;
+
+ return 0;
+ }
+
+
+ /*
* Move all immediately-ready items from pending_list to ready_list.
*
* Items are considered ready if they have no remaining dependencies and
*************** par_list_remove(TocEntry *te)
*** 4242,4248 ****
* which applies the same logic one-at-a-time.)
*/
static void
! move_to_ready_list(TocEntry *pending_list, TocEntry *ready_list,
RestorePass pass)
{
TocEntry *te;
--- 4388,4395 ----
* which applies the same logic one-at-a-time.)
*/
static void
! move_to_ready_list(TocEntry *pending_list,
! ParallelReadyList *ready_list,
RestorePass pass)
{
TocEntry *te;
*************** move_to_ready_list(TocEntry *pending_lis
*** 4259,4265 ****
/* Remove it from pending_list ... */
par_list_remove(te);
/* ... and add to ready_list */
! par_list_append(ready_list, te);
}
}
}
--- 4406,4412 ----
/* Remove it from pending_list ... */
par_list_remove(te);
/* ... and add to ready_list */
! ready_list_insert(ready_list, te);
}
}
}
*************** move_to_ready_list(TocEntry *pending_lis
*** 4272,4293 ****
* items currently running. Items in the ready_list are known to have
* no remaining dependencies, but we have to check for lock conflicts.
*
! * Note that the returned item has *not* been removed from ready_list.
! * The caller must do that after successfully dispatching the item.
*
* pref_non_data is for an alternative selection algorithm that gives
* preference to non-data items if there is already a data load running.
* It is currently disabled.
*/
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;
! TocEntry *te;
! int i,
! k;
/*
* Bogus heuristics for pref_non_data
--- 4419,4436 ----
* items currently running. Items in the ready_list are known to have
* no remaining dependencies, but we have to check for lock conflicts.
*
! * The returned item has been removed from the ready_list.
*
* pref_non_data is for an alternative selection algorithm that gives
* preference to non-data items if there is already a data load running.
* It is currently disabled.
*/
static TocEntry *
! get_next_work_item(ArchiveHandle *AH, ParallelReadyList *ready_list,
ParallelState *pstate)
{
bool pref_non_data = false; /* or get from AH->ropt */
! int data_te_index = -1;
/*
* Bogus heuristics for pref_non_data
*************** get_next_work_item(ArchiveHandle *AH, To
*** 4296,4302 ****
{
int count = 0;
! for (k = 0; k < pstate->numWorkers; k++)
{
TocEntry *running_te = pstate->te[k];
--- 4439,4445 ----
{
int count = 0;
! for (int k = 0; k < pstate->numWorkers; k++)
{
TocEntry *running_te = pstate->te[k];
*************** get_next_work_item(ArchiveHandle *AH, To
*** 4309,4318 ****
}
/*
* Search the ready_list until we find a suitable item.
*/
! for (te = ready_list->par_next; te != ready_list; te = te->par_next)
{
bool conflicts = false;
/*
--- 4452,4467 ----
}
/*
+ * Sort the ready_list so that we'll tackle larger jobs first.
+ */
+ ready_list_sort(ready_list);
+
+ /*
* Search the ready_list until we find a suitable item.
*/
! for (int i = ready_list->first_te; i <= ready_list->last_te; i++)
{
+ TocEntry *te = ready_list->tes[i];
bool conflicts = false;
/*
*************** get_next_work_item(ArchiveHandle *AH, To
*** 4320,4328 ****
* 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; i++)
{
! TocEntry *running_te = pstate->te[i];
if (running_te == NULL)
continue;
--- 4469,4477 ----
* that a currently running item also needs lock on, or vice versa. If
* so, we don't want to schedule them together.
*/
! for (int k = 0; k < pstate->numWorkers; k++)
{
! TocEntry *running_te = pstate->te[k];
if (running_te == NULL)
continue;
*************** get_next_work_item(ArchiveHandle *AH, To
*** 4339,4355 ****
if (pref_non_data && te->section == SECTION_DATA)
{
! if (data_te == NULL)
! data_te = te;
continue;
}
/* passed all tests, so this item can run */
return te;
}
! if (data_te != NULL)
return data_te;
ahlog(AH, 2, "no item ready\n");
return NULL;
--- 4488,4510 ----
if (pref_non_data && te->section == SECTION_DATA)
{
! if (data_te_index < 0)
! data_te_index = i;
continue;
}
/* passed all tests, so this item can run */
+ ready_list_remove(ready_list, i);
return te;
}
! if (data_te_index >= 0)
! {
! TocEntry *data_te = ready_list->tes[data_te_index];
!
! ready_list_remove(ready_list, data_te_index);
return data_te;
+ }
ahlog(AH, 2, "no item ready\n");
return NULL;
*************** mark_restore_job_done(ArchiveHandle *AH,
*** 4393,4399 ****
int status,
void *callback_data)
{
! TocEntry *ready_list = (TocEntry *) callback_data;
ahlog(AH, 1, "finished item %d %s %s\n",
te->dumpId, te->desc, te->tag);
--- 4548,4554 ----
int status,
void *callback_data)
{
! ParallelReadyList *ready_list = (ParallelReadyList *) callback_data;
ahlog(AH, 1, "finished item %d %s %s\n",
te->dumpId, te->desc, te->tag);
*************** fix_dependencies(ArchiveHandle *AH)
*** 4551,4556 ****
--- 4706,4717 ----
/*
* Change dependencies on table items to depend on table data items instead,
* but only in POST_DATA items.
+ *
+ * Also, for any item having such dependency(s), set its dataLength to the
+ * largest dataLength of the table data items it depends on. This ensures
+ * that parallel restore will prioritize larger jobs (index builds, FK
+ * constraint checks, etc) over smaller ones, avoiding situations where we
+ * end a restore with only one active job working on a large table.
*/
static void
repoint_table_dependencies(ArchiveHandle *AH)
*************** repoint_table_dependencies(ArchiveHandle
*** 4569,4577 ****
if (olddep <= AH->maxDumpId &&
AH->tableDataId[olddep] != 0)
{
! te->dependencies[i] = AH->tableDataId[olddep];
ahlog(AH, 2, "transferring dependency %d -> %d to %d\n",
! te->dumpId, olddep, AH->tableDataId[olddep]);
}
}
}
--- 4730,4742 ----
if (olddep <= AH->maxDumpId &&
AH->tableDataId[olddep] != 0)
{
! DumpId tabledataid = AH->tableDataId[olddep];
! TocEntry *tabledatate = AH->tocsByDumpId[tabledataid];
!
! te->dependencies[i] = tabledataid;
! te->dataLength = Max(te->dataLength, tabledatate->dataLength);
ahlog(AH, 2, "transferring dependency %d -> %d to %d\n",
! te->dumpId, olddep, tabledataid);
}
}
}
*************** identify_locking_dependencies(ArchiveHan
*** 4639,4645 ****
* becomes ready should be moved to the ready_list, if that's provided.
*/
static void
! reduce_dependencies(ArchiveHandle *AH, TocEntry *te, TocEntry *ready_list)
{
int i;
--- 4804,4811 ----
* becomes ready should be moved to the ready_list, if that's provided.
*/
static void
! reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
! ParallelReadyList *ready_list)
{
int i;
*************** reduce_dependencies(ArchiveHandle *AH, T
*** 4668,4674 ****
/* Remove it from pending list ... */
par_list_remove(otherte);
/* ... and add to ready_list */
! par_list_append(ready_list, otherte);
}
}
}
--- 4834,4840 ----
/* Remove it from pending list ... */
par_list_remove(otherte);
/* ... and add to ready_list */
! ready_list_insert(ready_list, otherte);
}
}
}
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index 8dd1915..3b7b8c0 100644
*** a/src/bin/pg_dump/pg_backup_archiver.h
--- b/src/bin/pg_dump/pg_backup_archiver.h
*************** typedef int (*WriteBytePtrType) (Archive
*** 162,173 ****
typedef int (*ReadBytePtrType) (ArchiveHandle *AH);
typedef void (*WriteBufPtrType) (ArchiveHandle *AH, const void *c, size_t len);
typedef void (*ReadBufPtrType) (ArchiveHandle *AH, void *buf, size_t len);
- typedef void (*SaveArchivePtrType) (ArchiveHandle *AH);
typedef void (*WriteExtraTocPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*ReadExtraTocPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*PrintExtraTocPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*PrintTocDataPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*ClonePtrType) (ArchiveHandle *AH);
typedef void (*DeClonePtrType) (ArchiveHandle *AH);
--- 162,173 ----
typedef int (*ReadBytePtrType) (ArchiveHandle *AH);
typedef void (*WriteBufPtrType) (ArchiveHandle *AH, const void *c, size_t len);
typedef void (*ReadBufPtrType) (ArchiveHandle *AH, void *buf, size_t len);
typedef void (*WriteExtraTocPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*ReadExtraTocPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*PrintExtraTocPtrType) (ArchiveHandle *AH, TocEntry *te);
typedef void (*PrintTocDataPtrType) (ArchiveHandle *AH, TocEntry *te);
+ typedef void (*PrepParallelRestorePtrType) (ArchiveHandle *AH);
typedef void (*ClonePtrType) (ArchiveHandle *AH);
typedef void (*DeClonePtrType) (ArchiveHandle *AH);
*************** struct _archiveHandle
*** 297,302 ****
--- 297,303 ----
WorkerJobDumpPtrType WorkerJobDumpPtr;
WorkerJobRestorePtrType WorkerJobRestorePtr;
+ PrepParallelRestorePtrType PrepParallelRestorePtr;
ClonePtrType ClonePtr; /* Clone format-specific fields */
DeClonePtrType DeClonePtr; /* Clean up cloned fields */
*************** struct _tocEntry
*** 387,392 ****
--- 388,394 ----
void *formatData; /* TOC Entry data specific to file format */
/* working state while dumping/restoring */
+ pgoff_t dataLength; /* item's data size; 0 if none or unknown */
teReqs reqs; /* do we need schema and/or data of object */
bool created; /* set for DATA member if TABLE was created */
*************** extern void on_exit_close_archive(Archiv
*** 405,410 ****
--- 407,424 ----
extern void warn_or_exit_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt,...) pg_attribute_printf(3, 4);
+ /* Called to add a TOC entry */
+ extern TocEntry *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,
+ const DumpId *deps, int nDeps,
+ DataDumperPtr dumpFn, void *dumpArg);
+
extern void WriteTOC(ArchiveHandle *AH);
extern void ReadTOC(ArchiveHandle *AH);
extern void WriteHead(ArchiveHandle *AH);
diff --git a/src/bin/pg_dump/pg_backup_custom.c b/src/bin/pg_dump/pg_backup_custom.c
index ad18a6c..96f44e8 100644
*** a/src/bin/pg_dump/pg_backup_custom.c
--- b/src/bin/pg_dump/pg_backup_custom.c
*************** static void _StartBlob(ArchiveHandle *AH
*** 59,64 ****
--- 59,66 ----
static void _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
static void _LoadBlobs(ArchiveHandle *AH, bool drop);
+
+ static void _PrepParallelRestore(ArchiveHandle *AH);
static void _Clone(ArchiveHandle *AH);
static void _DeClone(ArchiveHandle *AH);
*************** InitArchiveFmt_Custom(ArchiveHandle *AH)
*** 129,134 ****
--- 131,138 ----
AH->StartBlobPtr = _StartBlob;
AH->EndBlobPtr = _EndBlob;
AH->EndBlobsPtr = _EndBlobs;
+
+ AH->PrepParallelRestorePtr = _PrepParallelRestore;
AH->ClonePtr = _Clone;
AH->DeClonePtr = _DeClone;
*************** _ReopenArchive(ArchiveHandle *AH)
*** 776,781 ****
--- 780,845 ----
}
/*
+ * Prepare for parallel restore.
+ *
+ * The main thing that needs to happen here is to fill in TABLE DATA and BLOBS
+ * TOC entries' dataLength fields with appropriate values to guide the
+ * ordering of restore jobs. The source of said data is format-dependent,
+ * as is the exact meaning of the values.
+ *
+ * A format module might also choose to do other setup here.
+ */
+ static void
+ _PrepParallelRestore(ArchiveHandle *AH)
+ {
+ lclContext *ctx = (lclContext *) AH->formatData;
+ TocEntry *prev_te = NULL;
+ lclTocEntry *prev_tctx = NULL;
+ TocEntry *te;
+
+ /*
+ * Knowing that the data items were dumped out in TOC order, we can
+ * reconstruct the length of each item as the delta to the start offset of
+ * the next data item.
+ */
+ for (te = AH->toc->next; te != AH->toc; te = te->next)
+ {
+ lclTocEntry *tctx = (lclTocEntry *) te->formatData;
+
+ /*
+ * Ignore entries without a known data offset; if we were unable to
+ * seek to rewrite the TOC when creating the archive, this'll be all
+ * of them, and we'll end up with no size estimates.
+ */
+ if (tctx->dataState != K_OFFSET_POS_SET)
+ continue;
+
+ /* Compute previous data item's length */
+ if (prev_te)
+ {
+ if (tctx->dataPos > prev_tctx->dataPos)
+ prev_te->dataLength = tctx->dataPos - prev_tctx->dataPos;
+ }
+
+ prev_te = te;
+ prev_tctx = tctx;
+ }
+
+ /* If OK to seek, we can determine the length of the last item */
+ if (prev_te && ctx->hasSeek)
+ {
+ pgoff_t endpos;
+
+ if (fseeko(AH->FH, 0, SEEK_END) != 0)
+ exit_horribly(modulename, "error during file seek: %s\n",
+ strerror(errno));
+ endpos = ftello(AH->FH);
+ if (endpos > prev_tctx->dataPos)
+ prev_te->dataLength = endpos - prev_tctx->dataPos;
+ }
+ }
+
+ /*
* Clone format-specific fields during parallel restoration.
*/
static void
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 4aabb40..cda90b9 100644
*** a/src/bin/pg_dump/pg_backup_directory.c
--- b/src/bin/pg_dump/pg_backup_directory.c
*************** static void _EndBlob(ArchiveHandle *AH,
*** 87,92 ****
--- 87,93 ----
static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
static void _LoadBlobs(ArchiveHandle *AH);
+ static void _PrepParallelRestore(ArchiveHandle *AH);
static void _Clone(ArchiveHandle *AH);
static void _DeClone(ArchiveHandle *AH);
*************** InitArchiveFmt_Directory(ArchiveHandle *
*** 132,137 ****
--- 133,139 ----
AH->EndBlobPtr = _EndBlob;
AH->EndBlobsPtr = _EndBlobs;
+ AH->PrepParallelRestorePtr = _PrepParallelRestore;
AH->ClonePtr = _Clone;
AH->DeClonePtr = _DeClone;
*************** _ArchiveEntry(ArchiveHandle *AH, TocEntr
*** 240,252 ****
char fn[MAXPGPATH];
tctx = (lclTocEntry *) pg_malloc0(sizeof(lclTocEntry));
! if (te->dataDumper)
{
snprintf(fn, MAXPGPATH, "%d.dat", te->dumpId);
tctx->filename = pg_strdup(fn);
}
- else if (strcmp(te->desc, "BLOBS") == 0)
- tctx->filename = pg_strdup("blobs.toc");
else
tctx->filename = NULL;
--- 242,254 ----
char fn[MAXPGPATH];
tctx = (lclTocEntry *) pg_malloc0(sizeof(lclTocEntry));
! if (strcmp(te->desc, "BLOBS") == 0)
! tctx->filename = pg_strdup("blobs.toc");
! else if (te->dataDumper)
{
snprintf(fn, MAXPGPATH, "%d.dat", te->dumpId);
tctx->filename = pg_strdup(fn);
}
else
tctx->filename = NULL;
*************** setFilePath(ArchiveHandle *AH, char *buf
*** 727,732 ****
--- 729,796 ----
}
/*
+ * Prepare for parallel restore.
+ *
+ * The main thing that needs to happen here is to fill in TABLE DATA and BLOBS
+ * TOC entries' dataLength fields with appropriate values to guide the
+ * ordering of restore jobs. The source of said data is format-dependent,
+ * as is the exact meaning of the values.
+ *
+ * A format module might also choose to do other setup here.
+ */
+ static void
+ _PrepParallelRestore(ArchiveHandle *AH)
+ {
+ TocEntry *te;
+
+ for (te = AH->toc->next; te != AH->toc; te = te->next)
+ {
+ lclTocEntry *tctx = (lclTocEntry *) te->formatData;
+ char fname[MAXPGPATH];
+ struct stat st;
+
+ /*
+ * A dumpable object has set tctx->filename, any other object has not.
+ * (see _ArchiveEntry).
+ */
+ if (tctx->filename == NULL)
+ continue;
+
+ /* We may ignore items not due to be restored */
+ if ((te->reqs & REQ_DATA) == 0)
+ continue;
+
+ /*
+ * Stat the file and, if successful, put its size in dataLength. When
+ * using compression, the physical file size might not be a very good
+ * guide to the amount of work involved in restoring the file, but we
+ * only need an approximate indicator of that.
+ */
+ setFilePath(AH, fname, tctx->filename);
+
+ if (stat(fname, &st) == 0)
+ te->dataLength = st.st_size;
+ else
+ {
+ /* It might be compressed */
+ strlcat(fname, ".gz", sizeof(fname));
+ if (stat(fname, &st) == 0)
+ te->dataLength = st.st_size;
+ }
+
+ /*
+ * If this is the BLOBS entry, what we stat'd was blobs.toc, which
+ * most likely is a lot smaller than the actual blob data. We don't
+ * have a cheap way to estimate how much smaller, but fortunately it
+ * doesn't matter too much as long as we get the blobs processed
+ * reasonably early. Arbitrarily scale up by a factor of 1K.
+ */
+ if (strcmp(te->desc, "BLOBS") == 0)
+ te->dataLength *= 1024;
+ }
+ }
+
+ /*
* Clone format-specific fields during parallel restoration.
*/
static void
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 6763a89..0116072 100644
*** a/src/bin/pg_dump/pg_dump.c
--- b/src/bin/pg_dump/pg_dump.c
***************
*** 54,59 ****
--- 54,60 ----
#include "catalog/pg_trigger_d.h"
#include "catalog/pg_type_d.h"
#include "libpq/libpq-fs.h"
+ #include "storage/block.h"
#include "dumputils.h"
#include "parallel.h"
*************** main(int argc, char **argv)
*** 845,854 ****
*/
sortDumpableObjectsByTypeName(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,
boundaryObjs[0].dumpId, boundaryObjs[1].dumpId);
--- 846,851 ----
*************** dumpTableData(Archive *fout, TableDataIn
*** 2156,2168 ****
* See comments for BuildArchiveDependencies.
*/
if (tdinfo->dobj.dump & DUMP_COMPONENT_DATA)
! 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,
! &(tbinfo->dobj.dumpId), 1,
! dumpFn, tdinfo);
destroyPQExpBuffer(copyBuf);
destroyPQExpBuffer(clistBuf);
--- 2153,2180 ----
* See comments for BuildArchiveDependencies.
*/
if (tdinfo->dobj.dump & DUMP_COMPONENT_DATA)
! {
! TocEntry *te;
!
! te = 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,
! &(tbinfo->dobj.dumpId), 1,
! dumpFn, tdinfo);
!
! /*
! * Set the TocEntry's dataLength in case we are doing a parallel dump
! * and want to order dump jobs by table size. We choose to measure
! * dataLength in table pages during dump, so no scaling is needed.
! * However, relpages is declared as "integer" in pg_class, and hence
! * also in TableInfo, but it's really BlockNumber a/k/a unsigned int.
! * Cast so that we get the right interpretation of table sizes
! * exceeding INT_MAX pages.
! */
! te->dataLength = (BlockNumber) tbinfo->relpages;
! }
destroyPQExpBuffer(copyBuf);
destroyPQExpBuffer(clistBuf);
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6759,6766 ****
i_conoid,
i_condef,
i_tablespace,
! i_indreloptions,
! i_relpages;
int ntups;
for (i = 0; i < numTables; i++)
--- 6771,6777 ----
i_conoid,
i_condef,
i_tablespace,
! i_indreloptions;
int ntups;
for (i = 0; i < numTables; i++)
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6807,6813 ****
"i.indnkeyatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "i.indisreplident, t.relpages, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
--- 6818,6824 ----
"i.indnkeyatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "i.indisreplident, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6844,6850 ****
"i.indnatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "i.indisreplident, t.relpages, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
--- 6855,6861 ----
"i.indnatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "i.indisreplident, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6877,6883 ****
"i.indnatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "false AS indisreplident, t.relpages, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
--- 6888,6894 ----
"i.indnatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "false AS indisreplident, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6906,6912 ****
"i.indnatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "false AS indisreplident, t.relpages, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
--- 6917,6923 ----
"i.indnatts AS indnkeyatts, "
"i.indnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "false AS indisreplident, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6938,6944 ****
"t.relnatts AS indnkeyatts, "
"t.relnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "false AS indisreplident, t.relpages, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
--- 6949,6955 ----
"t.relnatts AS indnkeyatts, "
"t.relnatts AS indnatts, "
"i.indkey, i.indisclustered, "
! "false AS indisreplident, "
"c.contype, c.conname, "
"c.condeferrable, c.condeferred, "
"c.tableoid AS contableoid, "
*************** getIndexes(Archive *fout, TableInfo tbli
*** 6974,6980 ****
i_indkey = PQfnumber(res, "indkey");
i_indisclustered = PQfnumber(res, "indisclustered");
i_indisreplident = PQfnumber(res, "indisreplident");
- i_relpages = PQfnumber(res, "relpages");
i_contype = PQfnumber(res, "contype");
i_conname = PQfnumber(res, "conname");
i_condeferrable = PQfnumber(res, "condeferrable");
--- 6985,6990 ----
*************** getIndexes(Archive *fout, TableInfo tbli
*** 7013,7019 ****
indxinfo[j].indisclustered = (PQgetvalue(res, j, i_indisclustered)[0] == 't');
indxinfo[j].indisreplident = (PQgetvalue(res, j, i_indisreplident)[0] == 't');
indxinfo[j].parentidx = atooid(PQgetvalue(res, j, i_parentidx));
- indxinfo[j].relpages = atoi(PQgetvalue(res, j, i_relpages));
contype = *(PQgetvalue(res, j, i_contype));
if (contype == 'p' || contype == 'u' || contype == 'x')
--- 7023,7028 ----
*************** dumpDumpableObject(Archive *fout, Dumpab
*** 9845,9856 ****
break;
case DO_BLOB_DATA:
if (dobj->dump & DUMP_COMPONENT_DATA)
! ArchiveEntry(fout, dobj->catId, dobj->dumpId,
! dobj->name, NULL, NULL, "",
! false, "BLOBS", SECTION_DATA,
! "", "", NULL,
! NULL, 0,
! dumpBlobs, NULL);
break;
case DO_POLICY:
dumpPolicy(fout, (PolicyInfo *) dobj);
--- 9854,9884 ----
break;
case DO_BLOB_DATA:
if (dobj->dump & DUMP_COMPONENT_DATA)
! {
! TocEntry *te;
!
! te = ArchiveEntry(fout, dobj->catId, dobj->dumpId,
! dobj->name, NULL, NULL, "",
! false, "BLOBS", SECTION_DATA,
! "", "", NULL,
! NULL, 0,
! dumpBlobs, NULL);
!
! /*
! * Set the TocEntry's dataLength in case we are doing a
! * parallel dump and want to order dump jobs by table size.
! * (We need some size estimate for every TocEntry with a
! * DataDumper function.) We don't currently have any cheap
! * way to estimate the size of blobs, but it doesn't matter;
! * let's just set the size to a large value so parallel dumps
! * will launch this job first. If there's lots of blobs, we
! * win, and if there aren't, we don't lose much. (If you want
! * to improve on this, really what you should be thinking
! * about is allowing blob dumping to be parallelized, not just
! * getting a smarter estimate for the single TOC entry.)
! */
! te->dataLength = MaxBlockNumber;
! }
break;
case DO_POLICY:
dumpPolicy(fout, (PolicyInfo *) dobj);
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 1448005..685ad78 100644
*** a/src/bin/pg_dump/pg_dump.h
--- b/src/bin/pg_dump/pg_dump.h
*************** typedef struct _indxInfo
*** 370,376 ****
Oid parentidx; /* if partitioned, parent index OID */
/* if there is an associated constraint object, its dumpId: */
DumpId indexconstraint;
- int relpages; /* relpages of the underlying table */
} IndxInfo;
typedef struct _indexAttachInfo
--- 370,375 ----
*************** extern void parseOidArray(const char *st
*** 677,683 ****
extern void sortDumpableObjects(DumpableObject **objs, int numObjs,
DumpId preBoundaryId, DumpId postBoundaryId);
extern void sortDumpableObjectsByTypeName(DumpableObject **objs, int numObjs);
- extern void sortDataAndIndexObjectsBySize(DumpableObject **objs, int numObjs);
/*
* version specific routines
--- 676,681 ----
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 6227a8f..a1d3ced 100644
*** a/src/bin/pg_dump/pg_dump_sort.c
--- b/src/bin/pg_dump/pg_dump_sort.c
*************** static const char *modulename = gettext_
*** 35,44 ****
* pg_dump.c; that is, PRE_DATA objects must sort before DO_PRE_DATA_BOUNDARY,
* POST_DATA objects must sort after DO_POST_DATA_BOUNDARY, and DATA objects
* must sort between them.
- *
- * Note: sortDataAndIndexObjectsBySize wants to have all DO_TABLE_DATA and
- * DO_INDEX objects in contiguous chunks, so do not reuse the values for those
- * for other object types.
*/
static const int dbObjectTypePriority[] =
{
--- 35,40 ----
*************** static void repairDependencyLoop(Dumpabl
*** 111,206 ****
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
--- 107,112 ----