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 ----