Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * pg_backup_archiver.c
4 : *
5 : * Private implementation of the archiver routines.
6 : *
7 : * See the headers to pg_restore for more details.
8 : *
9 : * Copyright (c) 2000, Philip Warner
10 : * Rights are granted to use this software in any way so long
11 : * as this notice is not removed.
12 : *
13 : * The author is not responsible for loss or damages that may
14 : * result from its use.
15 : *
16 : *
17 : * IDENTIFICATION
18 : * src/bin/pg_dump/pg_backup_archiver.c
19 : *
20 : *-------------------------------------------------------------------------
21 : */
22 : #include "postgres_fe.h"
23 :
24 : #include <ctype.h>
25 : #include <fcntl.h>
26 : #include <unistd.h>
27 : #include <sys/stat.h>
28 : #include <sys/wait.h>
29 : #ifdef WIN32
30 : #include <io.h>
31 : #endif
32 :
33 : #include "catalog/pg_class_d.h"
34 : #include "common/string.h"
35 : #include "compress_io.h"
36 : #include "dumputils.h"
37 : #include "fe_utils/string_utils.h"
38 : #include "lib/binaryheap.h"
39 : #include "lib/stringinfo.h"
40 : #include "libpq/libpq-fs.h"
41 : #include "parallel.h"
42 : #include "pg_backup_archiver.h"
43 : #include "pg_backup_db.h"
44 : #include "pg_backup_utils.h"
45 :
46 : #define TEXT_DUMP_HEADER "--\n-- PostgreSQL database dump\n--\n\n"
47 : #define TEXT_DUMPALL_HEADER "--\n-- PostgreSQL database cluster dump\n--\n\n"
48 :
49 :
50 : static ArchiveHandle *_allocAH(const char *FileSpec, const ArchiveFormat fmt,
51 : const pg_compress_specification compression_spec,
52 : bool dosync, ArchiveMode mode,
53 : SetupWorkerPtrType setupWorkerPtr,
54 : DataDirSyncMethod sync_method);
55 : static void _getObjectDescription(PQExpBuffer buf, const TocEntry *te);
56 : static void _printTocEntry(ArchiveHandle *AH, TocEntry *te, bool isData);
57 : static char *sanitize_line(const char *str, bool want_hyphen);
58 : static void _doSetFixedOutputState(ArchiveHandle *AH);
59 : static void _doSetSessionAuth(ArchiveHandle *AH, const char *user);
60 : static void _reconnectToDB(ArchiveHandle *AH, const char *dbname);
61 : static void _becomeUser(ArchiveHandle *AH, const char *user);
62 : static void _becomeOwner(ArchiveHandle *AH, TocEntry *te);
63 : static void _selectOutputSchema(ArchiveHandle *AH, const char *schemaName);
64 : static void _selectTablespace(ArchiveHandle *AH, const char *tablespace);
65 : static void _selectTableAccessMethod(ArchiveHandle *AH, const char *tableam);
66 : static void _printTableAccessMethodNoStorage(ArchiveHandle *AH,
67 : TocEntry *te);
68 : static void processEncodingEntry(ArchiveHandle *AH, TocEntry *te);
69 : static void processStdStringsEntry(ArchiveHandle *AH, TocEntry *te);
70 : static void processSearchPathEntry(ArchiveHandle *AH, TocEntry *te);
71 : static int _tocEntryRequired(TocEntry *te, teSection curSection, ArchiveHandle *AH);
72 : static RestorePass _tocEntryRestorePass(TocEntry *te);
73 : static bool _tocEntryIsACL(TocEntry *te);
74 : static void _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te);
75 : static void _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te);
76 : static bool is_load_via_partition_root(TocEntry *te);
77 : static void buildTocEntryArrays(ArchiveHandle *AH);
78 : static void _moveBefore(TocEntry *pos, TocEntry *te);
79 : static int _discoverArchiveFormat(ArchiveHandle *AH);
80 :
81 : static int RestoringToDB(ArchiveHandle *AH);
82 : static void dump_lo_buf(ArchiveHandle *AH);
83 : static void dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim);
84 : static void SetOutput(ArchiveHandle *AH, const char *filename,
85 : const pg_compress_specification compression_spec);
86 : static CompressFileHandle *SaveOutput(ArchiveHandle *AH);
87 : static void RestoreOutput(ArchiveHandle *AH, CompressFileHandle *savedOutput);
88 :
89 : static int restore_toc_entry(ArchiveHandle *AH, TocEntry *te, bool is_parallel);
90 : static void restore_toc_entries_prefork(ArchiveHandle *AH,
91 : TocEntry *pending_list);
92 : static void restore_toc_entries_parallel(ArchiveHandle *AH,
93 : ParallelState *pstate,
94 : TocEntry *pending_list);
95 : static void restore_toc_entries_postfork(ArchiveHandle *AH,
96 : TocEntry *pending_list);
97 : static void pending_list_header_init(TocEntry *l);
98 : static void pending_list_append(TocEntry *l, TocEntry *te);
99 : static void pending_list_remove(TocEntry *te);
100 : static int TocEntrySizeCompareQsort(const void *p1, const void *p2);
101 : static int TocEntrySizeCompareBinaryheap(void *p1, void *p2, void *arg);
102 : static void move_to_ready_heap(TocEntry *pending_list,
103 : binaryheap *ready_heap,
104 : RestorePass pass);
105 : static TocEntry *pop_next_work_item(binaryheap *ready_heap,
106 : ParallelState *pstate);
107 : static void mark_dump_job_done(ArchiveHandle *AH,
108 : TocEntry *te,
109 : int status,
110 : void *callback_data);
111 : static void mark_restore_job_done(ArchiveHandle *AH,
112 : TocEntry *te,
113 : int status,
114 : void *callback_data);
115 : static void fix_dependencies(ArchiveHandle *AH);
116 : static bool has_lock_conflicts(TocEntry *te1, TocEntry *te2);
117 : static void repoint_table_dependencies(ArchiveHandle *AH);
118 : static void identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te);
119 : static void reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
120 : binaryheap *ready_heap);
121 : static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
122 : static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
123 :
124 : static void StrictNamesCheck(RestoreOptions *ropt);
125 :
126 :
127 : /*
128 : * Allocate a new DumpOptions block containing all default values.
129 : */
130 : DumpOptions *
131 80 : NewDumpOptions(void)
132 : {
133 80 : DumpOptions *opts = (DumpOptions *) pg_malloc(sizeof(DumpOptions));
134 :
135 80 : InitDumpOptions(opts);
136 80 : return opts;
137 : }
138 :
139 : /*
140 : * Initialize a DumpOptions struct to all default values
141 : */
142 : void
143 466 : InitDumpOptions(DumpOptions *opts)
144 : {
145 466 : memset(opts, 0, sizeof(DumpOptions));
146 : /* set any fields that shouldn't default to zeroes */
147 466 : opts->include_everything = true;
148 466 : opts->cparams.promptPassword = TRI_DEFAULT;
149 466 : opts->dumpSections = DUMP_UNSECTIONED;
150 466 : }
151 :
152 : /*
153 : * Create a freshly allocated DumpOptions with options equivalent to those
154 : * found in the given RestoreOptions.
155 : */
156 : DumpOptions *
157 80 : dumpOptionsFromRestoreOptions(RestoreOptions *ropt)
158 : {
159 80 : DumpOptions *dopt = NewDumpOptions();
160 :
161 : /* this is the inverse of what's at the end of pg_dump.c's main() */
162 80 : dopt->cparams.dbname = ropt->cparams.dbname ? pg_strdup(ropt->cparams.dbname) : NULL;
163 80 : dopt->cparams.pgport = ropt->cparams.pgport ? pg_strdup(ropt->cparams.pgport) : NULL;
164 80 : dopt->cparams.pghost = ropt->cparams.pghost ? pg_strdup(ropt->cparams.pghost) : NULL;
165 80 : dopt->cparams.username = ropt->cparams.username ? pg_strdup(ropt->cparams.username) : NULL;
166 80 : dopt->cparams.promptPassword = ropt->cparams.promptPassword;
167 80 : dopt->outputClean = ropt->dropSchema;
168 80 : dopt->dataOnly = ropt->dataOnly;
169 80 : dopt->schemaOnly = ropt->schemaOnly;
170 80 : dopt->if_exists = ropt->if_exists;
171 80 : dopt->column_inserts = ropt->column_inserts;
172 80 : dopt->dumpSections = ropt->dumpSections;
173 80 : dopt->aclsSkip = ropt->aclsSkip;
174 80 : dopt->outputSuperuser = ropt->superuser;
175 80 : dopt->outputCreateDB = ropt->createDB;
176 80 : dopt->outputNoOwner = ropt->noOwner;
177 80 : dopt->outputNoTableAm = ropt->noTableAm;
178 80 : dopt->outputNoTablespaces = ropt->noTablespace;
179 80 : dopt->disable_triggers = ropt->disable_triggers;
180 80 : dopt->use_setsessauth = ropt->use_setsessauth;
181 80 : dopt->disable_dollar_quoting = ropt->disable_dollar_quoting;
182 80 : dopt->dump_inserts = ropt->dump_inserts;
183 80 : dopt->no_comments = ropt->no_comments;
184 80 : dopt->no_publications = ropt->no_publications;
185 80 : dopt->no_security_labels = ropt->no_security_labels;
186 80 : dopt->no_subscriptions = ropt->no_subscriptions;
187 80 : dopt->lockWaitTimeout = ropt->lockWaitTimeout;
188 80 : dopt->include_everything = ropt->include_everything;
189 80 : dopt->enable_row_security = ropt->enable_row_security;
190 80 : dopt->sequence_data = ropt->sequence_data;
191 :
192 80 : return dopt;
193 : }
194 :
195 :
196 : /*
197 : * Wrapper functions.
198 : *
199 : * The objective is to make writing new formats and dumpers as simple
200 : * as possible, if necessary at the expense of extra function calls etc.
201 : *
202 : */
203 :
204 : /*
205 : * The dump worker setup needs lots of knowledge of the internals of pg_dump,
206 : * so it's defined in pg_dump.c and passed into OpenArchive. The restore worker
207 : * setup doesn't need to know anything much, so it's defined here.
208 : */
209 : static void
210 20 : setupRestoreWorker(Archive *AHX)
211 : {
212 20 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
213 :
214 20 : AH->ReopenPtr(AH);
215 20 : }
216 :
217 :
218 : /* Create a new archive */
219 : /* Public */
220 : Archive *
221 344 : CreateArchive(const char *FileSpec, const ArchiveFormat fmt,
222 : const pg_compress_specification compression_spec,
223 : bool dosync, ArchiveMode mode,
224 : SetupWorkerPtrType setupDumpWorker,
225 : DataDirSyncMethod sync_method)
226 :
227 : {
228 344 : ArchiveHandle *AH = _allocAH(FileSpec, fmt, compression_spec,
229 : dosync, mode, setupDumpWorker, sync_method);
230 :
231 342 : return (Archive *) AH;
232 : }
233 :
234 : /* Open an existing archive */
235 : /* Public */
236 : Archive *
237 76 : OpenArchive(const char *FileSpec, const ArchiveFormat fmt)
238 : {
239 : ArchiveHandle *AH;
240 76 : pg_compress_specification compression_spec = {0};
241 :
242 76 : compression_spec.algorithm = PG_COMPRESSION_NONE;
243 76 : AH = _allocAH(FileSpec, fmt, compression_spec, true,
244 : archModeRead, setupRestoreWorker,
245 : DATA_DIR_SYNC_METHOD_FSYNC);
246 :
247 76 : return (Archive *) AH;
248 : }
249 :
250 : /* Public */
251 : void
252 378 : CloseArchive(Archive *AHX)
253 : {
254 378 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
255 :
256 378 : AH->ClosePtr(AH);
257 :
258 : /* Close the output */
259 378 : errno = 0;
260 378 : if (!EndCompressFileHandle(AH->OF))
261 0 : pg_fatal("could not close output file: %m");
262 378 : }
263 :
264 : /* Public */
265 : void
266 730 : SetArchiveOptions(Archive *AH, DumpOptions *dopt, RestoreOptions *ropt)
267 : {
268 : /* Caller can omit dump options, in which case we synthesize them */
269 730 : if (dopt == NULL && ropt != NULL)
270 80 : dopt = dumpOptionsFromRestoreOptions(ropt);
271 :
272 : /* Save options for later access */
273 730 : AH->dopt = dopt;
274 730 : AH->ropt = ropt;
275 730 : }
276 :
277 : /* Public */
278 : void
279 372 : ProcessArchiveRestoreOptions(Archive *AHX)
280 : {
281 372 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
282 372 : RestoreOptions *ropt = AH->public.ropt;
283 : TocEntry *te;
284 : teSection curSection;
285 :
286 : /* Decide which TOC entries will be dumped/restored, and mark them */
287 372 : curSection = SECTION_PRE_DATA;
288 63436 : for (te = AH->toc->next; te != AH->toc; te = te->next)
289 : {
290 : /*
291 : * When writing an archive, we also take this opportunity to check
292 : * that we have generated the entries in a sane order that respects
293 : * the section divisions. When reading, don't complain, since buggy
294 : * old versions of pg_dump might generate out-of-order archives.
295 : */
296 63064 : if (AH->mode != archModeRead)
297 : {
298 53304 : switch (te->section)
299 : {
300 8896 : case SECTION_NONE:
301 : /* ok to be anywhere */
302 8896 : break;
303 26624 : case SECTION_PRE_DATA:
304 26624 : if (curSection != SECTION_PRE_DATA)
305 0 : pg_log_warning("archive items not in correct section order");
306 26624 : break;
307 8038 : case SECTION_DATA:
308 8038 : if (curSection == SECTION_POST_DATA)
309 0 : pg_log_warning("archive items not in correct section order");
310 8038 : break;
311 9746 : case SECTION_POST_DATA:
312 : /* ok no matter which section we were in */
313 9746 : break;
314 0 : default:
315 0 : pg_fatal("unexpected section code %d",
316 : (int) te->section);
317 : break;
318 : }
319 9760 : }
320 :
321 63064 : if (te->section != SECTION_NONE)
322 52706 : curSection = te->section;
323 :
324 63064 : te->reqs = _tocEntryRequired(te, curSection, AH);
325 : }
326 :
327 : /* Enforce strict names checking */
328 372 : if (ropt->strict_names)
329 0 : StrictNamesCheck(ropt);
330 372 : }
331 :
332 : /* Public */
333 : void
334 314 : RestoreArchive(Archive *AHX)
335 : {
336 314 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
337 314 : RestoreOptions *ropt = AH->public.ropt;
338 : bool parallel_mode;
339 : TocEntry *te;
340 : CompressFileHandle *sav;
341 :
342 314 : AH->stage = STAGE_INITIALIZING;
343 :
344 : /*
345 : * If we're going to do parallel restore, there are some restrictions.
346 : */
347 314 : parallel_mode = (AH->public.numWorkers > 1 && ropt->useDB);
348 314 : if (parallel_mode)
349 : {
350 : /* We haven't got round to making this work for all archive formats */
351 8 : if (AH->ClonePtr == NULL || AH->ReopenPtr == NULL)
352 0 : pg_fatal("parallel restore is not supported with this archive file format");
353 :
354 : /* Doesn't work if the archive represents dependencies as OIDs */
355 8 : if (AH->version < K_VERS_1_8)
356 0 : pg_fatal("parallel restore is not supported with archives made by pre-8.0 pg_dump");
357 :
358 : /*
359 : * It's also not gonna work if we can't reopen the input file, so
360 : * let's try that immediately.
361 : */
362 8 : AH->ReopenPtr(AH);
363 : }
364 :
365 : /*
366 : * Make sure we won't need (de)compression we haven't got
367 : */
368 314 : if (AH->PrintTocDataPtr != NULL)
369 : {
370 38120 : for (te = AH->toc->next; te != AH->toc; te = te->next)
371 : {
372 37990 : if (te->hadDumper && (te->reqs & REQ_DATA) != 0)
373 : {
374 184 : char *errmsg = supports_compression(AH->compression_spec);
375 :
376 184 : if (errmsg)
377 0 : pg_fatal("cannot restore from compressed archive (%s)",
378 : errmsg);
379 : else
380 184 : break;
381 : }
382 : }
383 : }
384 :
385 : /*
386 : * Prepare index arrays, so we can assume we have them throughout restore.
387 : * It's possible we already did this, though.
388 : */
389 314 : if (AH->tocsByDumpId == NULL)
390 310 : buildTocEntryArrays(AH);
391 :
392 : /*
393 : * If we're using a DB connection, then connect it.
394 : */
395 314 : if (ropt->useDB)
396 : {
397 28 : pg_log_info("connecting to database for restore");
398 28 : if (AH->version < K_VERS_1_3)
399 0 : pg_fatal("direct database connections are not supported in pre-1.3 archives");
400 :
401 : /*
402 : * We don't want to guess at whether the dump will successfully
403 : * restore; allow the attempt regardless of the version of the restore
404 : * target.
405 : */
406 28 : AHX->minRemoteVersion = 0;
407 28 : AHX->maxRemoteVersion = 9999999;
408 :
409 28 : ConnectDatabase(AHX, &ropt->cparams, false);
410 :
411 : /*
412 : * If we're talking to the DB directly, don't send comments since they
413 : * obscure SQL when displaying errors
414 : */
415 28 : AH->noTocComments = 1;
416 : }
417 :
418 : /*
419 : * Work out if we have an implied data-only restore. This can happen if
420 : * the dump was data only or if the user has used a toc list to exclude
421 : * all of the schema data. All we do is look for schema entries - if none
422 : * are found then we set the dataOnly flag.
423 : *
424 : * We could scan for wanted TABLE entries, but that is not the same as
425 : * dataOnly. At this stage, it seems unnecessary (6-Mar-2001).
426 : */
427 314 : if (!ropt->dataOnly)
428 : {
429 302 : int impliedDataOnly = 1;
430 :
431 2390 : for (te = AH->toc->next; te != AH->toc; te = te->next)
432 : {
433 2348 : if ((te->reqs & REQ_SCHEMA) != 0)
434 : { /* It's schema, and it's wanted */
435 260 : impliedDataOnly = 0;
436 260 : break;
437 : }
438 : }
439 302 : if (impliedDataOnly)
440 : {
441 42 : ropt->dataOnly = impliedDataOnly;
442 42 : pg_log_info("implied data-only restore");
443 : }
444 : }
445 :
446 : /*
447 : * Setup the output file if necessary.
448 : */
449 314 : sav = SaveOutput(AH);
450 314 : if (ropt->filename || ropt->compression_spec.algorithm != PG_COMPRESSION_NONE)
451 264 : SetOutput(AH, ropt->filename, ropt->compression_spec);
452 :
453 314 : ahprintf(AH, "--\n-- PostgreSQL database dump\n--\n\n");
454 :
455 314 : if (AH->archiveRemoteVersion)
456 314 : ahprintf(AH, "-- Dumped from database version %s\n",
457 : AH->archiveRemoteVersion);
458 314 : if (AH->archiveDumpVersion)
459 314 : ahprintf(AH, "-- Dumped by pg_dump version %s\n",
460 : AH->archiveDumpVersion);
461 :
462 314 : ahprintf(AH, "\n");
463 :
464 314 : if (AH->public.verbose)
465 42 : dumpTimestamp(AH, "Started on", AH->createDate);
466 :
467 314 : if (ropt->single_txn)
468 : {
469 0 : if (AH->connection)
470 0 : StartTransaction(AHX);
471 : else
472 0 : ahprintf(AH, "BEGIN;\n\n");
473 : }
474 :
475 : /*
476 : * Establish important parameter values right away.
477 : */
478 314 : _doSetFixedOutputState(AH);
479 :
480 314 : AH->stage = STAGE_PROCESSING;
481 :
482 : /*
483 : * Drop the items at the start, in reverse order
484 : */
485 314 : if (ropt->dropSchema)
486 : {
487 1964 : for (te = AH->toc->prev; te != AH->toc; te = te->prev)
488 : {
489 1938 : AH->currentTE = te;
490 :
491 : /*
492 : * In createDB mode, issue a DROP *only* for the database as a
493 : * whole. Issuing drops against anything else would be wrong,
494 : * because at this point we're connected to the wrong database.
495 : * (The DATABASE PROPERTIES entry, if any, should be treated like
496 : * the DATABASE entry.)
497 : */
498 1938 : if (ropt->createDB)
499 : {
500 728 : if (strcmp(te->desc, "DATABASE") != 0 &&
501 712 : strcmp(te->desc, "DATABASE PROPERTIES") != 0)
502 700 : continue;
503 : }
504 :
505 : /* Otherwise, drop anything that's selected and has a dropStmt */
506 1238 : if (((te->reqs & (REQ_SCHEMA | REQ_DATA)) != 0) && te->dropStmt)
507 : {
508 584 : bool not_allowed_in_txn = false;
509 :
510 584 : pg_log_info("dropping %s %s", te->desc, te->tag);
511 :
512 : /*
513 : * In --transaction-size mode, we have to temporarily exit our
514 : * transaction block to drop objects that can't be dropped
515 : * within a transaction.
516 : */
517 584 : if (ropt->txn_size > 0)
518 : {
519 24 : if (strcmp(te->desc, "DATABASE") == 0 ||
520 12 : strcmp(te->desc, "DATABASE PROPERTIES") == 0)
521 : {
522 24 : not_allowed_in_txn = true;
523 24 : if (AH->connection)
524 24 : CommitTransaction(AHX);
525 : else
526 0 : ahprintf(AH, "COMMIT;\n");
527 : }
528 : }
529 :
530 : /* Select owner and schema as necessary */
531 584 : _becomeOwner(AH, te);
532 584 : _selectOutputSchema(AH, te->namespace);
533 :
534 : /*
535 : * Now emit the DROP command, if the object has one. Note we
536 : * don't necessarily emit it verbatim; at this point we add an
537 : * appropriate IF EXISTS clause, if the user requested it.
538 : */
539 584 : if (strcmp(te->desc, "BLOB METADATA") == 0)
540 : {
541 : /* We must generate the per-blob commands */
542 8 : if (ropt->if_exists)
543 4 : IssueCommandPerBlob(AH, te,
544 : "SELECT pg_catalog.lo_unlink(oid) "
545 : "FROM pg_catalog.pg_largeobject_metadata "
546 : "WHERE oid = '", "'");
547 : else
548 4 : IssueCommandPerBlob(AH, te,
549 : "SELECT pg_catalog.lo_unlink('",
550 : "')");
551 : }
552 576 : else if (*te->dropStmt != '\0')
553 : {
554 554 : if (!ropt->if_exists ||
555 266 : strncmp(te->dropStmt, "--", 2) == 0)
556 : {
557 : /*
558 : * Without --if-exists, or if it's just a comment (as
559 : * happens for the public schema), print the dropStmt
560 : * as-is.
561 : */
562 290 : ahprintf(AH, "%s", te->dropStmt);
563 : }
564 : else
565 : {
566 : /*
567 : * Inject an appropriate spelling of "if exists". For
568 : * old-style large objects, we have a routine that
569 : * knows how to do it, without depending on
570 : * te->dropStmt; use that. For other objects we need
571 : * to parse the command.
572 : */
573 264 : if (strcmp(te->desc, "BLOB") == 0)
574 : {
575 0 : DropLOIfExists(AH, te->catalogId.oid);
576 : }
577 : else
578 : {
579 264 : char *dropStmt = pg_strdup(te->dropStmt);
580 264 : char *dropStmtOrig = dropStmt;
581 264 : PQExpBuffer ftStmt = createPQExpBuffer();
582 :
583 : /*
584 : * Need to inject IF EXISTS clause after ALTER
585 : * TABLE part in ALTER TABLE .. DROP statement
586 : */
587 264 : if (strncmp(dropStmt, "ALTER TABLE", 11) == 0)
588 : {
589 36 : appendPQExpBufferStr(ftStmt,
590 : "ALTER TABLE IF EXISTS");
591 36 : dropStmt = dropStmt + 11;
592 : }
593 :
594 : /*
595 : * ALTER TABLE..ALTER COLUMN..DROP DEFAULT does
596 : * not support the IF EXISTS clause, and therefore
597 : * we simply emit the original command for DEFAULT
598 : * objects (modulo the adjustment made above).
599 : *
600 : * Likewise, don't mess with DATABASE PROPERTIES.
601 : *
602 : * If we used CREATE OR REPLACE VIEW as a means of
603 : * quasi-dropping an ON SELECT rule, that should
604 : * be emitted unchanged as well.
605 : *
606 : * For other object types, we need to extract the
607 : * first part of the DROP which includes the
608 : * object type. Most of the time this matches
609 : * te->desc, so search for that; however for the
610 : * different kinds of CONSTRAINTs, we know to
611 : * search for hardcoded "DROP CONSTRAINT" instead.
612 : */
613 264 : if (strcmp(te->desc, "DEFAULT") == 0 ||
614 258 : strcmp(te->desc, "DATABASE PROPERTIES") == 0 ||
615 258 : strncmp(dropStmt, "CREATE OR REPLACE VIEW", 22) == 0)
616 6 : appendPQExpBufferStr(ftStmt, dropStmt);
617 : else
618 : {
619 : char buffer[40];
620 : char *mark;
621 :
622 258 : if (strcmp(te->desc, "CONSTRAINT") == 0 ||
623 232 : strcmp(te->desc, "CHECK CONSTRAINT") == 0 ||
624 232 : strcmp(te->desc, "FK CONSTRAINT") == 0)
625 30 : strcpy(buffer, "DROP CONSTRAINT");
626 : else
627 228 : snprintf(buffer, sizeof(buffer), "DROP %s",
628 : te->desc);
629 :
630 258 : mark = strstr(dropStmt, buffer);
631 :
632 258 : if (mark)
633 : {
634 258 : *mark = '\0';
635 258 : appendPQExpBuffer(ftStmt, "%s%s IF EXISTS%s",
636 : dropStmt, buffer,
637 258 : mark + strlen(buffer));
638 : }
639 : else
640 : {
641 : /* complain and emit unmodified command */
642 0 : pg_log_warning("could not find where to insert IF EXISTS in statement \"%s\"",
643 : dropStmtOrig);
644 0 : appendPQExpBufferStr(ftStmt, dropStmt);
645 : }
646 : }
647 :
648 264 : ahprintf(AH, "%s", ftStmt->data);
649 :
650 264 : destroyPQExpBuffer(ftStmt);
651 264 : pg_free(dropStmtOrig);
652 : }
653 : }
654 : }
655 :
656 : /*
657 : * In --transaction-size mode, re-establish the transaction
658 : * block if needed; otherwise, commit after every N drops.
659 : */
660 584 : if (ropt->txn_size > 0)
661 : {
662 24 : if (not_allowed_in_txn)
663 : {
664 24 : if (AH->connection)
665 24 : StartTransaction(AHX);
666 : else
667 0 : ahprintf(AH, "BEGIN;\n");
668 24 : AH->txnCount = 0;
669 : }
670 0 : else if (++AH->txnCount >= ropt->txn_size)
671 : {
672 0 : if (AH->connection)
673 : {
674 0 : CommitTransaction(AHX);
675 0 : StartTransaction(AHX);
676 : }
677 : else
678 0 : ahprintf(AH, "COMMIT;\nBEGIN;\n");
679 0 : AH->txnCount = 0;
680 : }
681 : }
682 : }
683 : }
684 :
685 : /*
686 : * _selectOutputSchema may have set currSchema to reflect the effect
687 : * of a "SET search_path" command it emitted. However, by now we may
688 : * have dropped that schema; or it might not have existed in the first
689 : * place. In either case the effective value of search_path will not
690 : * be what we think. Forcibly reset currSchema so that we will
691 : * re-establish the search_path setting when needed (after creating
692 : * the schema).
693 : *
694 : * If we treated users as pg_dump'able objects then we'd need to reset
695 : * currUser here too.
696 : */
697 26 : free(AH->currSchema);
698 26 : AH->currSchema = NULL;
699 : }
700 :
701 314 : if (parallel_mode)
702 : {
703 : /*
704 : * In parallel mode, turn control over to the parallel-restore logic.
705 : */
706 : ParallelState *pstate;
707 : TocEntry pending_list;
708 :
709 : /* The archive format module may need some setup for this */
710 8 : if (AH->PrepParallelRestorePtr)
711 8 : AH->PrepParallelRestorePtr(AH);
712 :
713 8 : pending_list_header_init(&pending_list);
714 :
715 : /* This runs PRE_DATA items and then disconnects from the database */
716 8 : restore_toc_entries_prefork(AH, &pending_list);
717 : Assert(AH->connection == NULL);
718 :
719 : /* ParallelBackupStart() will actually fork the processes */
720 8 : pstate = ParallelBackupStart(AH);
721 8 : restore_toc_entries_parallel(AH, pstate, &pending_list);
722 8 : ParallelBackupEnd(AH, pstate);
723 :
724 : /* reconnect the leader and see if we missed something */
725 8 : restore_toc_entries_postfork(AH, &pending_list);
726 : Assert(AH->connection != NULL);
727 : }
728 : else
729 : {
730 : /*
731 : * In serial mode, process everything in three phases: normal items,
732 : * then ACLs, then post-ACL items. We might be able to skip one or
733 : * both extra phases in some cases, eg data-only restores.
734 : */
735 306 : bool haveACL = false;
736 306 : bool havePostACL = false;
737 :
738 54190 : for (te = AH->toc->next; te != AH->toc; te = te->next)
739 : {
740 53886 : if ((te->reqs & (REQ_SCHEMA | REQ_DATA)) == 0)
741 2402 : continue; /* ignore if not to be dumped at all */
742 :
743 51484 : switch (_tocEntryRestorePass(te))
744 : {
745 47164 : case RESTORE_PASS_MAIN:
746 47164 : (void) restore_toc_entry(AH, te, false);
747 47162 : break;
748 3764 : case RESTORE_PASS_ACL:
749 3764 : haveACL = true;
750 3764 : break;
751 556 : case RESTORE_PASS_POST_ACL:
752 556 : havePostACL = true;
753 556 : break;
754 : }
755 53884 : }
756 :
757 304 : if (haveACL)
758 : {
759 52144 : for (te = AH->toc->next; te != AH->toc; te = te->next)
760 : {
761 102374 : if ((te->reqs & (REQ_SCHEMA | REQ_DATA)) != 0 &&
762 50366 : _tocEntryRestorePass(te) == RESTORE_PASS_ACL)
763 3764 : (void) restore_toc_entry(AH, te, false);
764 : }
765 : }
766 :
767 304 : if (havePostACL)
768 : {
769 41244 : for (te = AH->toc->next; te != AH->toc; te = te->next)
770 : {
771 81662 : if ((te->reqs & (REQ_SCHEMA | REQ_DATA)) != 0 &&
772 40494 : _tocEntryRestorePass(te) == RESTORE_PASS_POST_ACL)
773 556 : (void) restore_toc_entry(AH, te, false);
774 : }
775 : }
776 : }
777 :
778 : /*
779 : * Close out any persistent transaction we may have. While these two
780 : * cases are started in different places, we can end both cases here.
781 : */
782 312 : if (ropt->single_txn || ropt->txn_size > 0)
783 : {
784 20 : if (AH->connection)
785 20 : CommitTransaction(AHX);
786 : else
787 0 : ahprintf(AH, "COMMIT;\n\n");
788 : }
789 :
790 312 : if (AH->public.verbose)
791 42 : dumpTimestamp(AH, "Completed on", time(NULL));
792 :
793 312 : ahprintf(AH, "--\n-- PostgreSQL database dump complete\n--\n\n");
794 :
795 : /*
796 : * Clean up & we're done.
797 : */
798 312 : AH->stage = STAGE_FINALIZING;
799 :
800 312 : if (ropt->filename || ropt->compression_spec.algorithm != PG_COMPRESSION_NONE)
801 264 : RestoreOutput(AH, sav);
802 :
803 312 : if (ropt->useDB)
804 28 : DisconnectDatabase(&AH->public);
805 312 : }
806 :
807 : /*
808 : * Restore a single TOC item. Used in both parallel and non-parallel restore;
809 : * is_parallel is true if we are in a worker child process.
810 : *
811 : * Returns 0 normally, but WORKER_CREATE_DONE or WORKER_INHIBIT_DATA if
812 : * the parallel parent has to make the corresponding status update.
813 : */
814 : static int
815 51676 : restore_toc_entry(ArchiveHandle *AH, TocEntry *te, bool is_parallel)
816 : {
817 51676 : RestoreOptions *ropt = AH->public.ropt;
818 51676 : int status = WORKER_OK;
819 : int reqs;
820 : bool defnDumped;
821 :
822 51676 : AH->currentTE = te;
823 :
824 : /* Dump any relevant dump warnings to stderr */
825 51676 : if (!ropt->suppressDumpWarnings && strcmp(te->desc, "WARNING") == 0)
826 : {
827 0 : if (!ropt->dataOnly && te->defn != NULL && strlen(te->defn) != 0)
828 0 : pg_log_warning("warning from original dump file: %s", te->defn);
829 0 : else if (te->copyStmt != NULL && strlen(te->copyStmt) != 0)
830 0 : pg_log_warning("warning from original dump file: %s", te->copyStmt);
831 : }
832 :
833 : /* Work out what, if anything, we want from this entry */
834 51676 : reqs = te->reqs;
835 :
836 51676 : defnDumped = false;
837 :
838 : /*
839 : * If it has a schema component that we want, then process that
840 : */
841 51676 : if ((reqs & REQ_SCHEMA) != 0)
842 : {
843 43564 : bool object_is_db = false;
844 :
845 : /*
846 : * In --transaction-size mode, must exit our transaction block to
847 : * create a database or set its properties.
848 : */
849 43564 : if (strcmp(te->desc, "DATABASE") == 0 ||
850 43480 : strcmp(te->desc, "DATABASE PROPERTIES") == 0)
851 : {
852 112 : object_is_db = true;
853 112 : if (ropt->txn_size > 0)
854 : {
855 40 : if (AH->connection)
856 40 : CommitTransaction(&AH->public);
857 : else
858 0 : ahprintf(AH, "COMMIT;\n\n");
859 : }
860 : }
861 :
862 : /* Show namespace in log message if available */
863 43564 : if (te->namespace)
864 41274 : pg_log_info("creating %s \"%s.%s\"",
865 : te->desc, te->namespace, te->tag);
866 : else
867 2290 : pg_log_info("creating %s \"%s\"",
868 : te->desc, te->tag);
869 :
870 43564 : _printTocEntry(AH, te, false);
871 43564 : defnDumped = true;
872 :
873 43564 : if (strcmp(te->desc, "TABLE") == 0)
874 : {
875 9024 : if (AH->lastErrorTE == te)
876 : {
877 : /*
878 : * We failed to create the table. If
879 : * --no-data-for-failed-tables was given, mark the
880 : * corresponding TABLE DATA to be ignored.
881 : *
882 : * In the parallel case this must be done in the parent, so we
883 : * just set the return value.
884 : */
885 0 : if (ropt->noDataForFailedTables)
886 : {
887 0 : if (is_parallel)
888 0 : status = WORKER_INHIBIT_DATA;
889 : else
890 0 : inhibit_data_for_failed_table(AH, te);
891 : }
892 : }
893 : else
894 : {
895 : /*
896 : * We created the table successfully. Mark the corresponding
897 : * TABLE DATA for possible truncation.
898 : *
899 : * In the parallel case this must be done in the parent, so we
900 : * just set the return value.
901 : */
902 9024 : if (is_parallel)
903 0 : status = WORKER_CREATE_DONE;
904 : else
905 9024 : mark_create_done(AH, te);
906 : }
907 : }
908 :
909 : /*
910 : * If we created a DB, connect to it. Also, if we changed DB
911 : * properties, reconnect to ensure that relevant GUC settings are
912 : * applied to our session. (That also restarts the transaction block
913 : * in --transaction-size mode.)
914 : */
915 43564 : if (object_is_db)
916 : {
917 112 : pg_log_info("connecting to new database \"%s\"", te->tag);
918 112 : _reconnectToDB(AH, te->tag);
919 : }
920 : }
921 :
922 : /*
923 : * If it has a data component that we want, then process that
924 : */
925 51676 : if ((reqs & REQ_DATA) != 0)
926 : {
927 : /*
928 : * hadDumper will be set if there is genuine data component for this
929 : * node. Otherwise, we need to check the defn field for statements
930 : * that need to be executed in data-only restores.
931 : */
932 8082 : if (te->hadDumper)
933 : {
934 : /*
935 : * If we can output the data, then restore it.
936 : */
937 7078 : if (AH->PrintTocDataPtr != NULL)
938 : {
939 7078 : _printTocEntry(AH, te, true);
940 :
941 7078 : if (strcmp(te->desc, "BLOBS") == 0 ||
942 6942 : strcmp(te->desc, "BLOB COMMENTS") == 0)
943 : {
944 136 : pg_log_info("processing %s", te->desc);
945 :
946 136 : _selectOutputSchema(AH, "pg_catalog");
947 :
948 : /* Send BLOB COMMENTS data to ExecuteSimpleCommands() */
949 136 : if (strcmp(te->desc, "BLOB COMMENTS") == 0)
950 0 : AH->outputKind = OUTPUT_OTHERDATA;
951 :
952 136 : AH->PrintTocDataPtr(AH, te);
953 :
954 136 : AH->outputKind = OUTPUT_SQLCMDS;
955 : }
956 : else
957 : {
958 : bool use_truncate;
959 :
960 6942 : _disableTriggersIfNecessary(AH, te);
961 :
962 : /* Select owner and schema as necessary */
963 6942 : _becomeOwner(AH, te);
964 6942 : _selectOutputSchema(AH, te->namespace);
965 :
966 6942 : pg_log_info("processing data for table \"%s.%s\"",
967 : te->namespace, te->tag);
968 :
969 : /*
970 : * In parallel restore, if we created the table earlier in
971 : * this run (so that we know it is empty) and we are not
972 : * restoring a load-via-partition-root data item then we
973 : * wrap the COPY in a transaction and precede it with a
974 : * TRUNCATE. If wal_level is set to minimal this prevents
975 : * WAL-logging the COPY. This obtains a speedup similar
976 : * to that from using single_txn mode in non-parallel
977 : * restores.
978 : *
979 : * We mustn't do this for load-via-partition-root cases
980 : * because some data might get moved across partition
981 : * boundaries, risking deadlock and/or loss of previously
982 : * loaded data. (We assume that all partitions of a
983 : * partitioned table will be treated the same way.)
984 : */
985 6974 : use_truncate = is_parallel && te->created &&
986 32 : !is_load_via_partition_root(te);
987 :
988 6942 : if (use_truncate)
989 : {
990 : /*
991 : * Parallel restore is always talking directly to a
992 : * server, so no need to see if we should issue BEGIN.
993 : */
994 20 : StartTransaction(&AH->public);
995 :
996 : /*
997 : * Issue TRUNCATE with ONLY so that child tables are
998 : * not wiped.
999 : */
1000 20 : ahprintf(AH, "TRUNCATE TABLE ONLY %s;\n\n",
1001 20 : fmtQualifiedId(te->namespace, te->tag));
1002 : }
1003 :
1004 : /*
1005 : * If we have a copy statement, use it.
1006 : */
1007 6942 : if (te->copyStmt && strlen(te->copyStmt) > 0)
1008 : {
1009 6804 : ahprintf(AH, "%s", te->copyStmt);
1010 6804 : AH->outputKind = OUTPUT_COPYDATA;
1011 : }
1012 : else
1013 138 : AH->outputKind = OUTPUT_OTHERDATA;
1014 :
1015 6942 : AH->PrintTocDataPtr(AH, te);
1016 :
1017 : /*
1018 : * Terminate COPY if needed.
1019 : */
1020 13742 : if (AH->outputKind == OUTPUT_COPYDATA &&
1021 6802 : RestoringToDB(AH))
1022 18 : EndDBCopyMode(&AH->public, te->tag);
1023 6940 : AH->outputKind = OUTPUT_SQLCMDS;
1024 :
1025 : /* close out the transaction started above */
1026 6940 : if (use_truncate)
1027 20 : CommitTransaction(&AH->public);
1028 :
1029 6940 : _enableTriggersIfNecessary(AH, te);
1030 : }
1031 : }
1032 : }
1033 1004 : else if (!defnDumped)
1034 : {
1035 : /* If we haven't already dumped the defn part, do so now */
1036 1004 : pg_log_info("executing %s %s", te->desc, te->tag);
1037 1004 : _printTocEntry(AH, te, false);
1038 : }
1039 : }
1040 :
1041 : /*
1042 : * If we emitted anything for this TOC entry, that counts as one action
1043 : * against the transaction-size limit. Commit if it's time to.
1044 : */
1045 51674 : if ((reqs & (REQ_SCHEMA | REQ_DATA)) != 0 && ropt->txn_size > 0)
1046 : {
1047 3930 : if (++AH->txnCount >= ropt->txn_size)
1048 : {
1049 2 : if (AH->connection)
1050 : {
1051 2 : CommitTransaction(&AH->public);
1052 2 : StartTransaction(&AH->public);
1053 : }
1054 : else
1055 0 : ahprintf(AH, "COMMIT;\nBEGIN;\n\n");
1056 2 : AH->txnCount = 0;
1057 : }
1058 : }
1059 :
1060 51674 : if (AH->public.n_errors > 0 && status == WORKER_OK)
1061 0 : status = WORKER_IGNORED_ERRORS;
1062 :
1063 51674 : return status;
1064 : }
1065 :
1066 : /*
1067 : * Allocate a new RestoreOptions block.
1068 : * This is mainly so we can initialize it, but also for future expansion,
1069 : */
1070 : RestoreOptions *
1071 436 : NewRestoreOptions(void)
1072 : {
1073 : RestoreOptions *opts;
1074 :
1075 436 : opts = (RestoreOptions *) pg_malloc0(sizeof(RestoreOptions));
1076 :
1077 : /* set any fields that shouldn't default to zeroes */
1078 436 : opts->format = archUnknown;
1079 436 : opts->cparams.promptPassword = TRI_DEFAULT;
1080 436 : opts->dumpSections = DUMP_UNSECTIONED;
1081 436 : opts->compression_spec.algorithm = PG_COMPRESSION_NONE;
1082 436 : opts->compression_spec.level = 0;
1083 :
1084 436 : return opts;
1085 : }
1086 :
1087 : static void
1088 6942 : _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te)
1089 : {
1090 6942 : RestoreOptions *ropt = AH->public.ropt;
1091 :
1092 : /* This hack is only needed in a data-only restore */
1093 6942 : if (!ropt->dataOnly || !ropt->disable_triggers)
1094 6882 : return;
1095 :
1096 60 : pg_log_info("disabling triggers for %s", te->tag);
1097 :
1098 : /*
1099 : * Become superuser if possible, since they are the only ones who can
1100 : * disable constraint triggers. If -S was not given, assume the initial
1101 : * user identity is a superuser. (XXX would it be better to become the
1102 : * table owner?)
1103 : */
1104 60 : _becomeUser(AH, ropt->superuser);
1105 :
1106 : /*
1107 : * Disable them.
1108 : */
1109 60 : ahprintf(AH, "ALTER TABLE %s DISABLE TRIGGER ALL;\n\n",
1110 60 : fmtQualifiedId(te->namespace, te->tag));
1111 : }
1112 :
1113 : static void
1114 6940 : _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te)
1115 : {
1116 6940 : RestoreOptions *ropt = AH->public.ropt;
1117 :
1118 : /* This hack is only needed in a data-only restore */
1119 6940 : if (!ropt->dataOnly || !ropt->disable_triggers)
1120 6880 : return;
1121 :
1122 60 : pg_log_info("enabling triggers for %s", te->tag);
1123 :
1124 : /*
1125 : * Become superuser if possible, since they are the only ones who can
1126 : * disable constraint triggers. If -S was not given, assume the initial
1127 : * user identity is a superuser. (XXX would it be better to become the
1128 : * table owner?)
1129 : */
1130 60 : _becomeUser(AH, ropt->superuser);
1131 :
1132 : /*
1133 : * Enable them.
1134 : */
1135 60 : ahprintf(AH, "ALTER TABLE %s ENABLE TRIGGER ALL;\n\n",
1136 60 : fmtQualifiedId(te->namespace, te->tag));
1137 : }
1138 :
1139 : /*
1140 : * Detect whether a TABLE DATA TOC item is performing "load via partition
1141 : * root", that is the target table is an ancestor partition rather than the
1142 : * table the TOC item is nominally for.
1143 : *
1144 : * In newer archive files this can be detected by checking for a special
1145 : * comment placed in te->defn. In older files we have to fall back to seeing
1146 : * if the COPY statement targets the named table or some other one. This
1147 : * will not work for data dumped as INSERT commands, so we could give a false
1148 : * negative in that case; fortunately, that's a rarely-used option.
1149 : */
1150 : static bool
1151 32 : is_load_via_partition_root(TocEntry *te)
1152 : {
1153 32 : if (te->defn &&
1154 12 : strncmp(te->defn, "-- load via partition root ", 27) == 0)
1155 12 : return true;
1156 20 : if (te->copyStmt && *te->copyStmt)
1157 : {
1158 12 : PQExpBuffer copyStmt = createPQExpBuffer();
1159 : bool result;
1160 :
1161 : /*
1162 : * Build the initial part of the COPY as it would appear if the
1163 : * nominal target table is the actual target. If we see anything
1164 : * else, it must be a load-via-partition-root case.
1165 : */
1166 12 : appendPQExpBuffer(copyStmt, "COPY %s ",
1167 12 : fmtQualifiedId(te->namespace, te->tag));
1168 12 : result = strncmp(te->copyStmt, copyStmt->data, copyStmt->len) != 0;
1169 12 : destroyPQExpBuffer(copyStmt);
1170 12 : return result;
1171 : }
1172 : /* Assume it's not load-via-partition-root */
1173 8 : return false;
1174 : }
1175 :
1176 : /*
1177 : * This is a routine that is part of the dumper interface, hence the 'Archive*' parameter.
1178 : */
1179 :
1180 : /* Public */
1181 : void
1182 3469396 : WriteData(Archive *AHX, const void *data, size_t dLen)
1183 : {
1184 3469396 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
1185 :
1186 3469396 : if (!AH->currToc)
1187 0 : pg_fatal("internal error -- WriteData cannot be called outside the context of a DataDumper routine");
1188 :
1189 3469396 : AH->WriteDataPtr(AH, data, dLen);
1190 3469396 : }
1191 :
1192 : /*
1193 : * Create a new TOC entry. The TOC was designed as a TOC, but is now the
1194 : * repository for all metadata. But the name has stuck.
1195 : *
1196 : * The new entry is added to the Archive's TOC list. Most callers can ignore
1197 : * the result value because nothing else need be done, but a few want to
1198 : * manipulate the TOC entry further.
1199 : */
1200 :
1201 : /* Public */
1202 : TocEntry *
1203 53304 : ArchiveEntry(Archive *AHX, CatalogId catalogId, DumpId dumpId,
1204 : ArchiveOpts *opts)
1205 : {
1206 53304 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
1207 : TocEntry *newToc;
1208 :
1209 53304 : newToc = (TocEntry *) pg_malloc0(sizeof(TocEntry));
1210 :
1211 53304 : AH->tocCount++;
1212 53304 : if (dumpId > AH->maxDumpId)
1213 10998 : AH->maxDumpId = dumpId;
1214 :
1215 53304 : newToc->prev = AH->toc->prev;
1216 53304 : newToc->next = AH->toc;
1217 53304 : AH->toc->prev->next = newToc;
1218 53304 : AH->toc->prev = newToc;
1219 :
1220 53304 : newToc->catalogId = catalogId;
1221 53304 : newToc->dumpId = dumpId;
1222 53304 : newToc->section = opts->section;
1223 :
1224 53304 : newToc->tag = pg_strdup(opts->tag);
1225 53304 : newToc->namespace = opts->namespace ? pg_strdup(opts->namespace) : NULL;
1226 53304 : newToc->tablespace = opts->tablespace ? pg_strdup(opts->tablespace) : NULL;
1227 53304 : newToc->tableam = opts->tableam ? pg_strdup(opts->tableam) : NULL;
1228 53304 : newToc->relkind = opts->relkind;
1229 53304 : newToc->owner = opts->owner ? pg_strdup(opts->owner) : NULL;
1230 53304 : newToc->desc = pg_strdup(opts->description);
1231 53304 : newToc->defn = opts->createStmt ? pg_strdup(opts->createStmt) : NULL;
1232 53304 : newToc->dropStmt = opts->dropStmt ? pg_strdup(opts->dropStmt) : NULL;
1233 53304 : newToc->copyStmt = opts->copyStmt ? pg_strdup(opts->copyStmt) : NULL;
1234 :
1235 53304 : if (opts->nDeps > 0)
1236 : {
1237 17648 : newToc->dependencies = (DumpId *) pg_malloc(opts->nDeps * sizeof(DumpId));
1238 17648 : memcpy(newToc->dependencies, opts->deps, opts->nDeps * sizeof(DumpId));
1239 17648 : newToc->nDeps = opts->nDeps;
1240 : }
1241 : else
1242 : {
1243 35656 : newToc->dependencies = NULL;
1244 35656 : newToc->nDeps = 0;
1245 : }
1246 :
1247 53304 : newToc->dataDumper = opts->dumpFn;
1248 53304 : newToc->dataDumperArg = opts->dumpArg;
1249 53304 : newToc->hadDumper = opts->dumpFn ? true : false;
1250 :
1251 53304 : newToc->formatData = NULL;
1252 53304 : newToc->dataLength = 0;
1253 :
1254 53304 : if (AH->ArchiveEntryPtr != NULL)
1255 9596 : AH->ArchiveEntryPtr(AH, newToc);
1256 :
1257 53304 : return newToc;
1258 : }
1259 :
1260 : /* Public */
1261 : void
1262 8 : PrintTOCSummary(Archive *AHX)
1263 : {
1264 8 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
1265 8 : RestoreOptions *ropt = AH->public.ropt;
1266 : TocEntry *te;
1267 8 : pg_compress_specification out_compression_spec = {0};
1268 : teSection curSection;
1269 : CompressFileHandle *sav;
1270 : const char *fmtName;
1271 : char stamp_str[64];
1272 :
1273 : /* TOC is always uncompressed */
1274 8 : out_compression_spec.algorithm = PG_COMPRESSION_NONE;
1275 :
1276 8 : sav = SaveOutput(AH);
1277 8 : if (ropt->filename)
1278 0 : SetOutput(AH, ropt->filename, out_compression_spec);
1279 :
1280 8 : if (strftime(stamp_str, sizeof(stamp_str), PGDUMP_STRFTIME_FMT,
1281 8 : localtime(&AH->createDate)) == 0)
1282 0 : strcpy(stamp_str, "[unknown]");
1283 :
1284 8 : ahprintf(AH, ";\n; Archive created at %s\n", stamp_str);
1285 16 : ahprintf(AH, "; dbname: %s\n; TOC Entries: %d\n; Compression: %s\n",
1286 8 : sanitize_line(AH->archdbname, false),
1287 : AH->tocCount,
1288 : get_compress_algorithm_name(AH->compression_spec.algorithm));
1289 :
1290 8 : switch (AH->format)
1291 : {
1292 6 : case archCustom:
1293 6 : fmtName = "CUSTOM";
1294 6 : break;
1295 2 : case archDirectory:
1296 2 : fmtName = "DIRECTORY";
1297 2 : break;
1298 0 : case archTar:
1299 0 : fmtName = "TAR";
1300 0 : break;
1301 0 : default:
1302 0 : fmtName = "UNKNOWN";
1303 : }
1304 :
1305 8 : ahprintf(AH, "; Dump Version: %d.%d-%d\n",
1306 8 : ARCHIVE_MAJOR(AH->version), ARCHIVE_MINOR(AH->version), ARCHIVE_REV(AH->version));
1307 8 : ahprintf(AH, "; Format: %s\n", fmtName);
1308 8 : ahprintf(AH, "; Integer: %d bytes\n", (int) AH->intSize);
1309 8 : ahprintf(AH, "; Offset: %d bytes\n", (int) AH->offSize);
1310 8 : if (AH->archiveRemoteVersion)
1311 8 : ahprintf(AH, "; Dumped from database version: %s\n",
1312 : AH->archiveRemoteVersion);
1313 8 : if (AH->archiveDumpVersion)
1314 8 : ahprintf(AH, "; Dumped by pg_dump version: %s\n",
1315 : AH->archiveDumpVersion);
1316 :
1317 8 : ahprintf(AH, ";\n;\n; Selected TOC Entries:\n;\n");
1318 :
1319 8 : curSection = SECTION_PRE_DATA;
1320 2280 : for (te = AH->toc->next; te != AH->toc; te = te->next)
1321 : {
1322 : /* This bit must match ProcessArchiveRestoreOptions' marking logic */
1323 2272 : if (te->section != SECTION_NONE)
1324 1696 : curSection = te->section;
1325 2272 : te->reqs = _tocEntryRequired(te, curSection, AH);
1326 : /* Now, should we print it? */
1327 2272 : if (ropt->verbose ||
1328 2272 : (te->reqs & (REQ_SCHEMA | REQ_DATA)) != 0)
1329 : {
1330 : char *sanitized_name;
1331 : char *sanitized_schema;
1332 : char *sanitized_owner;
1333 :
1334 : /*
1335 : */
1336 2232 : sanitized_name = sanitize_line(te->tag, false);
1337 2232 : sanitized_schema = sanitize_line(te->namespace, true);
1338 2232 : sanitized_owner = sanitize_line(te->owner, false);
1339 :
1340 2232 : ahprintf(AH, "%d; %u %u %s %s %s %s\n", te->dumpId,
1341 : te->catalogId.tableoid, te->catalogId.oid,
1342 : te->desc, sanitized_schema, sanitized_name,
1343 : sanitized_owner);
1344 :
1345 2232 : free(sanitized_name);
1346 2232 : free(sanitized_schema);
1347 2232 : free(sanitized_owner);
1348 : }
1349 2272 : if (ropt->verbose && te->nDeps > 0)
1350 : {
1351 : int i;
1352 :
1353 0 : ahprintf(AH, ";\tdepends on:");
1354 0 : for (i = 0; i < te->nDeps; i++)
1355 0 : ahprintf(AH, " %d", te->dependencies[i]);
1356 0 : ahprintf(AH, "\n");
1357 : }
1358 : }
1359 :
1360 : /* Enforce strict names checking */
1361 8 : if (ropt->strict_names)
1362 0 : StrictNamesCheck(ropt);
1363 :
1364 8 : if (ropt->filename)
1365 0 : RestoreOutput(AH, sav);
1366 8 : }
1367 :
1368 : /***********
1369 : * Large Object Archival
1370 : ***********/
1371 :
1372 : /* Called by a dumper to signal start of a LO */
1373 : int
1374 148 : StartLO(Archive *AHX, Oid oid)
1375 : {
1376 148 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
1377 :
1378 148 : if (!AH->StartLOPtr)
1379 0 : pg_fatal("large-object output not supported in chosen format");
1380 :
1381 148 : AH->StartLOPtr(AH, AH->currToc, oid);
1382 :
1383 148 : return 1;
1384 : }
1385 :
1386 : /* Called by a dumper to signal end of a LO */
1387 : int
1388 148 : EndLO(Archive *AHX, Oid oid)
1389 : {
1390 148 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
1391 :
1392 148 : if (AH->EndLOPtr)
1393 148 : AH->EndLOPtr(AH, AH->currToc, oid);
1394 :
1395 148 : return 1;
1396 : }
1397 :
1398 : /**********
1399 : * Large Object Restoration
1400 : **********/
1401 :
1402 : /*
1403 : * Called by a format handler before a group of LOs is restored
1404 : */
1405 : void
1406 32 : StartRestoreLOs(ArchiveHandle *AH)
1407 : {
1408 32 : RestoreOptions *ropt = AH->public.ropt;
1409 :
1410 : /*
1411 : * LOs must be restored within a transaction block, since we need the LO
1412 : * handle to stay open while we write it. Establish a transaction unless
1413 : * there's one being used globally.
1414 : */
1415 32 : if (!(ropt->single_txn || ropt->txn_size > 0))
1416 : {
1417 32 : if (AH->connection)
1418 0 : StartTransaction(&AH->public);
1419 : else
1420 32 : ahprintf(AH, "BEGIN;\n\n");
1421 : }
1422 :
1423 32 : AH->loCount = 0;
1424 32 : }
1425 :
1426 : /*
1427 : * Called by a format handler after a group of LOs is restored
1428 : */
1429 : void
1430 32 : EndRestoreLOs(ArchiveHandle *AH)
1431 : {
1432 32 : RestoreOptions *ropt = AH->public.ropt;
1433 :
1434 32 : if (!(ropt->single_txn || ropt->txn_size > 0))
1435 : {
1436 32 : if (AH->connection)
1437 0 : CommitTransaction(&AH->public);
1438 : else
1439 32 : ahprintf(AH, "COMMIT;\n\n");
1440 : }
1441 :
1442 32 : pg_log_info(ngettext("restored %d large object",
1443 : "restored %d large objects",
1444 : AH->loCount),
1445 : AH->loCount);
1446 32 : }
1447 :
1448 :
1449 : /*
1450 : * Called by a format handler to initiate restoration of a LO
1451 : */
1452 : void
1453 32 : StartRestoreLO(ArchiveHandle *AH, Oid oid, bool drop)
1454 : {
1455 32 : bool old_lo_style = (AH->version < K_VERS_1_12);
1456 : Oid loOid;
1457 :
1458 32 : AH->loCount++;
1459 :
1460 : /* Initialize the LO Buffer */
1461 32 : if (AH->lo_buf == NULL)
1462 : {
1463 : /* First time through (in this process) so allocate the buffer */
1464 16 : AH->lo_buf_size = LOBBUFSIZE;
1465 16 : AH->lo_buf = (void *) pg_malloc(LOBBUFSIZE);
1466 : }
1467 32 : AH->lo_buf_used = 0;
1468 :
1469 32 : pg_log_info("restoring large object with OID %u", oid);
1470 :
1471 : /* With an old archive we must do drop and create logic here */
1472 32 : if (old_lo_style && drop)
1473 0 : DropLOIfExists(AH, oid);
1474 :
1475 32 : if (AH->connection)
1476 : {
1477 0 : if (old_lo_style)
1478 : {
1479 0 : loOid = lo_create(AH->connection, oid);
1480 0 : if (loOid == 0 || loOid != oid)
1481 0 : pg_fatal("could not create large object %u: %s",
1482 : oid, PQerrorMessage(AH->connection));
1483 : }
1484 0 : AH->loFd = lo_open(AH->connection, oid, INV_WRITE);
1485 0 : if (AH->loFd == -1)
1486 0 : pg_fatal("could not open large object %u: %s",
1487 : oid, PQerrorMessage(AH->connection));
1488 : }
1489 : else
1490 : {
1491 32 : if (old_lo_style)
1492 0 : ahprintf(AH, "SELECT pg_catalog.lo_open(pg_catalog.lo_create('%u'), %d);\n",
1493 : oid, INV_WRITE);
1494 : else
1495 32 : ahprintf(AH, "SELECT pg_catalog.lo_open('%u', %d);\n",
1496 : oid, INV_WRITE);
1497 : }
1498 :
1499 32 : AH->writingLO = true;
1500 32 : }
1501 :
1502 : void
1503 32 : EndRestoreLO(ArchiveHandle *AH, Oid oid)
1504 : {
1505 32 : if (AH->lo_buf_used > 0)
1506 : {
1507 : /* Write remaining bytes from the LO buffer */
1508 16 : dump_lo_buf(AH);
1509 : }
1510 :
1511 32 : AH->writingLO = false;
1512 :
1513 32 : if (AH->connection)
1514 : {
1515 0 : lo_close(AH->connection, AH->loFd);
1516 0 : AH->loFd = -1;
1517 : }
1518 : else
1519 : {
1520 32 : ahprintf(AH, "SELECT pg_catalog.lo_close(0);\n\n");
1521 : }
1522 32 : }
1523 :
1524 : /***********
1525 : * Sorting and Reordering
1526 : ***********/
1527 :
1528 : void
1529 0 : SortTocFromFile(Archive *AHX)
1530 : {
1531 0 : ArchiveHandle *AH = (ArchiveHandle *) AHX;
1532 0 : RestoreOptions *ropt = AH->public.ropt;
1533 : FILE *fh;
1534 : StringInfoData linebuf;
1535 :
1536 : /* Allocate space for the 'wanted' array, and init it */
1537 0 : ropt->idWanted = (bool *) pg_malloc0(sizeof(bool) * AH->maxDumpId);
1538 :
1539 : /* Setup the file */
1540 0 : fh = fopen(ropt->tocFile, PG_BINARY_R);
1541 0 : if (!fh)
1542 0 : pg_fatal("could not open TOC file \"%s\": %m", ropt->tocFile);
1543 :
1544 0 : initStringInfo(&linebuf);
1545 :
1546 0 : while (pg_get_line_buf(fh, &linebuf))
1547 : {
1548 : char *cmnt;
1549 : char *endptr;
1550 : DumpId id;
1551 : TocEntry *te;
1552 :
1553 : /* Truncate line at comment, if any */
1554 0 : cmnt = strchr(linebuf.data, ';');
1555 0 : if (cmnt != NULL)
1556 : {
1557 0 : cmnt[0] = '\0';
1558 0 : linebuf.len = cmnt - linebuf.data;
1559 : }
1560 :
1561 : /* Ignore if all blank */
1562 0 : if (strspn(linebuf.data, " \t\r\n") == linebuf.len)
1563 0 : continue;
1564 :
1565 : /* Get an ID, check it's valid and not already seen */
1566 0 : id = strtol(linebuf.data, &endptr, 10);
1567 0 : if (endptr == linebuf.data || id <= 0 || id > AH->maxDumpId ||
1568 0 : ropt->idWanted[id - 1])
1569 : {
1570 0 : pg_log_warning("line ignored: %s", linebuf.data);
1571 0 : continue;
1572 : }
1573 :
1574 : /* Find TOC entry */
1575 0 : te = getTocEntryByDumpId(AH, id);
1576 0 : if (!te)
1577 0 : pg_fatal("could not find entry for ID %d",
1578 : id);
1579 :
1580 : /* Mark it wanted */
1581 0 : ropt->idWanted[id - 1] = true;
1582 :
1583 : /*
1584 : * Move each item to the end of the list as it is selected, so that
1585 : * they are placed in the desired order. Any unwanted items will end
1586 : * up at the front of the list, which may seem unintuitive but it's
1587 : * what we need. In an ordinary serial restore that makes no
1588 : * difference, but in a parallel restore we need to mark unrestored
1589 : * items' dependencies as satisfied before we start examining
1590 : * restorable items. Otherwise they could have surprising
1591 : * side-effects on the order in which restorable items actually get
1592 : * restored.
1593 : */
1594 0 : _moveBefore(AH->toc, te);
1595 : }
1596 :
1597 0 : pg_free(linebuf.data);
1598 :
1599 0 : if (fclose(fh) != 0)
1600 0 : pg_fatal("could not close TOC file: %m");
1601 0 : }
1602 :
1603 : /**********************
1604 : * Convenience functions that look like standard IO functions
1605 : * for writing data when in dump mode.
1606 : **********************/
1607 :
1608 : /* Public */
1609 : void
1610 43244 : archputs(const char *s, Archive *AH)
1611 : {
1612 43244 : WriteData(AH, s, strlen(s));
1613 43244 : }
1614 :
1615 : /* Public */
1616 : int
1617 6756 : archprintf(Archive *AH, const char *fmt,...)
1618 : {
1619 6756 : int save_errno = errno;
1620 : char *p;
1621 6756 : size_t len = 128; /* initial assumption about buffer size */
1622 : size_t cnt;
1623 :
1624 : for (;;)
1625 0 : {
1626 : va_list args;
1627 :
1628 : /* Allocate work buffer. */
1629 6756 : p = (char *) pg_malloc(len);
1630 :
1631 : /* Try to format the data. */
1632 6756 : errno = save_errno;
1633 6756 : va_start(args, fmt);
1634 6756 : cnt = pvsnprintf(p, len, fmt, args);
1635 6756 : va_end(args);
1636 :
1637 6756 : if (cnt < len)
1638 6756 : break; /* success */
1639 :
1640 : /* Release buffer and loop around to try again with larger len. */
1641 0 : free(p);
1642 0 : len = cnt;
1643 : }
1644 :
1645 6756 : WriteData(AH, p, cnt);
1646 6756 : free(p);
1647 6756 : return (int) cnt;
1648 : }
1649 :
1650 :
1651 : /*******************************
1652 : * Stuff below here should be 'private' to the archiver routines
1653 : *******************************/
1654 :
1655 : static void
1656 264 : SetOutput(ArchiveHandle *AH, const char *filename,
1657 : const pg_compress_specification compression_spec)
1658 : {
1659 : CompressFileHandle *CFH;
1660 : const char *mode;
1661 264 : int fn = -1;
1662 :
1663 264 : if (filename)
1664 : {
1665 264 : if (strcmp(filename, "-") == 0)
1666 0 : fn = fileno(stdout);
1667 : }
1668 0 : else if (AH->FH)
1669 0 : fn = fileno(AH->FH);
1670 0 : else if (AH->fSpec)
1671 : {
1672 0 : filename = AH->fSpec;
1673 : }
1674 : else
1675 0 : fn = fileno(stdout);
1676 :
1677 264 : if (AH->mode == archModeAppend)
1678 86 : mode = PG_BINARY_A;
1679 : else
1680 178 : mode = PG_BINARY_W;
1681 :
1682 264 : CFH = InitCompressFileHandle(compression_spec);
1683 :
1684 264 : if (!CFH->open_func(filename, fn, mode, CFH))
1685 : {
1686 0 : if (filename)
1687 0 : pg_fatal("could not open output file \"%s\": %m", filename);
1688 : else
1689 0 : pg_fatal("could not open output file: %m");
1690 : }
1691 :
1692 264 : AH->OF = CFH;
1693 264 : }
1694 :
1695 : static CompressFileHandle *
1696 322 : SaveOutput(ArchiveHandle *AH)
1697 : {
1698 322 : return (CompressFileHandle *) AH->OF;
1699 : }
1700 :
1701 : static void
1702 264 : RestoreOutput(ArchiveHandle *AH, CompressFileHandle *savedOutput)
1703 : {
1704 264 : errno = 0;
1705 264 : if (!EndCompressFileHandle(AH->OF))
1706 0 : pg_fatal("could not close output file: %m");
1707 :
1708 264 : AH->OF = savedOutput;
1709 264 : }
1710 :
1711 :
1712 :
1713 : /*
1714 : * Print formatted text to the output file (usually stdout).
1715 : */
1716 : int
1717 281870 : ahprintf(ArchiveHandle *AH, const char *fmt,...)
1718 : {
1719 281870 : int save_errno = errno;
1720 : char *p;
1721 281870 : size_t len = 128; /* initial assumption about buffer size */
1722 : size_t cnt;
1723 :
1724 : for (;;)
1725 16158 : {
1726 : va_list args;
1727 :
1728 : /* Allocate work buffer. */
1729 298028 : p = (char *) pg_malloc(len);
1730 :
1731 : /* Try to format the data. */
1732 298028 : errno = save_errno;
1733 298028 : va_start(args, fmt);
1734 298028 : cnt = pvsnprintf(p, len, fmt, args);
1735 298028 : va_end(args);
1736 :
1737 298028 : if (cnt < len)
1738 281870 : break; /* success */
1739 :
1740 : /* Release buffer and loop around to try again with larger len. */
1741 16158 : free(p);
1742 16158 : len = cnt;
1743 : }
1744 :
1745 281870 : ahwrite(p, 1, cnt, AH);
1746 281870 : free(p);
1747 281870 : return (int) cnt;
1748 : }
1749 :
1750 : /*
1751 : * Single place for logic which says 'We are restoring to a direct DB connection'.
1752 : */
1753 : static int
1754 3707098 : RestoringToDB(ArchiveHandle *AH)
1755 : {
1756 3707098 : RestoreOptions *ropt = AH->public.ropt;
1757 :
1758 3707098 : return (ropt && ropt->useDB && AH->connection);
1759 : }
1760 :
1761 : /*
1762 : * Dump the current contents of the LO data buffer while writing a LO
1763 : */
1764 : static void
1765 16 : dump_lo_buf(ArchiveHandle *AH)
1766 : {
1767 16 : if (AH->connection)
1768 : {
1769 : int res;
1770 :
1771 0 : res = lo_write(AH->connection, AH->loFd, AH->lo_buf, AH->lo_buf_used);
1772 0 : pg_log_debug(ngettext("wrote %zu byte of large object data (result = %d)",
1773 : "wrote %zu bytes of large object data (result = %d)",
1774 : AH->lo_buf_used),
1775 : AH->lo_buf_used, res);
1776 : /* We assume there are no short writes, only errors */
1777 0 : if (res != AH->lo_buf_used)
1778 0 : warn_or_exit_horribly(AH, "could not write to large object: %s",
1779 0 : PQerrorMessage(AH->connection));
1780 : }
1781 : else
1782 : {
1783 16 : PQExpBuffer buf = createPQExpBuffer();
1784 :
1785 16 : appendByteaLiteralAHX(buf,
1786 : (const unsigned char *) AH->lo_buf,
1787 : AH->lo_buf_used,
1788 : AH);
1789 :
1790 : /* Hack: turn off writingLO so ahwrite doesn't recurse to here */
1791 16 : AH->writingLO = false;
1792 16 : ahprintf(AH, "SELECT pg_catalog.lowrite(0, %s);\n", buf->data);
1793 16 : AH->writingLO = true;
1794 :
1795 16 : destroyPQExpBuffer(buf);
1796 : }
1797 16 : AH->lo_buf_used = 0;
1798 16 : }
1799 :
1800 :
1801 : /*
1802 : * Write buffer to the output file (usually stdout). This is used for
1803 : * outputting 'restore' scripts etc. It is even possible for an archive
1804 : * format to create a custom output routine to 'fake' a restore if it
1805 : * wants to generate a script (see TAR output).
1806 : */
1807 : void
1808 3702802 : ahwrite(const void *ptr, size_t size, size_t nmemb, ArchiveHandle *AH)
1809 : {
1810 3702802 : int bytes_written = 0;
1811 :
1812 3702802 : if (AH->writingLO)
1813 : {
1814 26 : size_t remaining = size * nmemb;
1815 :
1816 26 : while (AH->lo_buf_used + remaining > AH->lo_buf_size)
1817 : {
1818 0 : size_t avail = AH->lo_buf_size - AH->lo_buf_used;
1819 :
1820 0 : memcpy((char *) AH->lo_buf + AH->lo_buf_used, ptr, avail);
1821 0 : ptr = (const void *) ((const char *) ptr + avail);
1822 0 : remaining -= avail;
1823 0 : AH->lo_buf_used += avail;
1824 0 : dump_lo_buf(AH);
1825 : }
1826 :
1827 26 : memcpy((char *) AH->lo_buf + AH->lo_buf_used, ptr, remaining);
1828 26 : AH->lo_buf_used += remaining;
1829 :
1830 26 : bytes_written = size * nmemb;
1831 : }
1832 3702776 : else if (AH->CustomOutPtr)
1833 3534 : bytes_written = AH->CustomOutPtr(AH, ptr, size * nmemb);
1834 :
1835 : /*
1836 : * If we're doing a restore, and it's direct to DB, and we're connected
1837 : * then send it to the DB.
1838 : */
1839 3699242 : else if (RestoringToDB(AH))
1840 7982 : bytes_written = ExecuteSqlCommandBuf(&AH->public, (const char *) ptr, size * nmemb);
1841 : else
1842 : {
1843 3691260 : CompressFileHandle *CFH = (CompressFileHandle *) AH->OF;
1844 :
1845 3691260 : if (CFH->write_func(ptr, size * nmemb, CFH))
1846 3691260 : bytes_written = size * nmemb;
1847 : }
1848 :
1849 3702802 : if (bytes_written != size * nmemb)
1850 0 : WRITE_ERROR_EXIT;
1851 3702802 : }
1852 :
1853 : /* on some error, we may decide to go on... */
1854 : void
1855 0 : warn_or_exit_horribly(ArchiveHandle *AH, const char *fmt,...)
1856 : {
1857 : va_list ap;
1858 :
1859 0 : switch (AH->stage)
1860 : {
1861 :
1862 0 : case STAGE_NONE:
1863 : /* Do nothing special */
1864 0 : break;
1865 :
1866 0 : case STAGE_INITIALIZING:
1867 0 : if (AH->stage != AH->lastErrorStage)
1868 0 : pg_log_info("while INITIALIZING:");
1869 0 : break;
1870 :
1871 0 : case STAGE_PROCESSING:
1872 0 : if (AH->stage != AH->lastErrorStage)
1873 0 : pg_log_info("while PROCESSING TOC:");
1874 0 : break;
1875 :
1876 0 : case STAGE_FINALIZING:
1877 0 : if (AH->stage != AH->lastErrorStage)
1878 0 : pg_log_info("while FINALIZING:");
1879 0 : break;
1880 : }
1881 0 : if (AH->currentTE != NULL && AH->currentTE != AH->lastErrorTE)
1882 : {
1883 0 : pg_log_info("from TOC entry %d; %u %u %s %s %s",
1884 : AH->currentTE->dumpId,
1885 : AH->currentTE->catalogId.tableoid,
1886 : AH->currentTE->catalogId.oid,
1887 : AH->currentTE->desc ? AH->currentTE->desc : "(no desc)",
1888 : AH->currentTE->tag ? AH->currentTE->tag : "(no tag)",
1889 : AH->currentTE->owner ? AH->currentTE->owner : "(no owner)");
1890 : }
1891 0 : AH->lastErrorStage = AH->stage;
1892 0 : AH->lastErrorTE = AH->currentTE;
1893 :
1894 0 : va_start(ap, fmt);
1895 0 : pg_log_generic_v(PG_LOG_ERROR, PG_LOG_PRIMARY, fmt, ap);
1896 0 : va_end(ap);
1897 :
1898 0 : if (AH->public.exit_on_error)
1899 0 : exit_nicely(1);
1900 : else
1901 0 : AH->public.n_errors++;
1902 0 : }
1903 :
1904 : #ifdef NOT_USED
1905 :
1906 : static void
1907 : _moveAfter(ArchiveHandle *AH, TocEntry *pos, TocEntry *te)
1908 : {
1909 : /* Unlink te from list */
1910 : te->prev->next = te->next;
1911 : te->next->prev = te->prev;
1912 :
1913 : /* and insert it after "pos" */
1914 : te->prev = pos;
1915 : te->next = pos->next;
1916 : pos->next->prev = te;
1917 : pos->next = te;
1918 : }
1919 : #endif
1920 :
1921 : static void
1922 0 : _moveBefore(TocEntry *pos, TocEntry *te)
1923 : {
1924 : /* Unlink te from list */
1925 0 : te->prev->next = te->next;
1926 0 : te->next->prev = te->prev;
1927 :
1928 : /* and insert it before "pos" */
1929 0 : te->prev = pos->prev;
1930 0 : te->next = pos;
1931 0 : pos->prev->next = te;
1932 0 : pos->prev = te;
1933 0 : }
1934 :
1935 : /*
1936 : * Build index arrays for the TOC list
1937 : *
1938 : * This should be invoked only after we have created or read in all the TOC
1939 : * items.
1940 : *
1941 : * The arrays are indexed by dump ID (so entry zero is unused). Note that the
1942 : * array entries run only up to maxDumpId. We might see dependency dump IDs
1943 : * beyond that (if the dump was partial); so always check the array bound
1944 : * before trying to touch an array entry.
1945 : */
1946 : static void
1947 354 : buildTocEntryArrays(ArchiveHandle *AH)
1948 : {
1949 354 : DumpId maxDumpId = AH->maxDumpId;
1950 : TocEntry *te;
1951 :
1952 354 : AH->tocsByDumpId = (TocEntry **) pg_malloc0((maxDumpId + 1) * sizeof(TocEntry *));
1953 354 : AH->tableDataId = (DumpId *) pg_malloc0((maxDumpId + 1) * sizeof(DumpId));
1954 :
1955 63272 : for (te = AH->toc->next; te != AH->toc; te = te->next)
1956 : {
1957 : /* this check is purely paranoia, maxDumpId should be correct */
1958 62918 : if (te->dumpId <= 0 || te->dumpId > maxDumpId)
1959 0 : pg_fatal("bad dumpId");
1960 :
1961 : /* tocsByDumpId indexes all TOCs by their dump ID */
1962 62918 : AH->tocsByDumpId[te->dumpId] = te;
1963 :
1964 : /*
1965 : * tableDataId provides the TABLE DATA item's dump ID for each TABLE
1966 : * TOC entry that has a DATA item. We compute this by reversing the
1967 : * TABLE DATA item's dependency, knowing that a TABLE DATA item has
1968 : * just one dependency and it is the TABLE item.
1969 : */
1970 62918 : if (strcmp(te->desc, "TABLE DATA") == 0 && te->nDeps > 0)
1971 : {
1972 7624 : DumpId tableId = te->dependencies[0];
1973 :
1974 : /*
1975 : * The TABLE item might not have been in the archive, if this was
1976 : * a data-only dump; but its dump ID should be less than its data
1977 : * item's dump ID, so there should be a place for it in the array.
1978 : */
1979 7624 : if (tableId <= 0 || tableId > maxDumpId)
1980 0 : pg_fatal("bad table dumpId for TABLE DATA item");
1981 :
1982 7624 : AH->tableDataId[tableId] = te->dumpId;
1983 : }
1984 : }
1985 354 : }
1986 :
1987 : TocEntry *
1988 20946 : getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)
1989 : {
1990 : /* build index arrays if we didn't already */
1991 20946 : if (AH->tocsByDumpId == NULL)
1992 44 : buildTocEntryArrays(AH);
1993 :
1994 20946 : if (id > 0 && id <= AH->maxDumpId)
1995 20946 : return AH->tocsByDumpId[id];
1996 :
1997 0 : return NULL;
1998 : }
1999 :
2000 : int
2001 20618 : TocIDRequired(ArchiveHandle *AH, DumpId id)
2002 : {
2003 20618 : TocEntry *te = getTocEntryByDumpId(AH, id);
2004 :
2005 20618 : if (!te)
2006 9746 : return 0;
2007 :
2008 10872 : return te->reqs;
2009 : }
2010 :
2011 : size_t
2012 12716 : WriteOffset(ArchiveHandle *AH, pgoff_t o, int wasSet)
2013 : {
2014 : int off;
2015 :
2016 : /* Save the flag */
2017 12716 : AH->WriteBytePtr(AH, wasSet);
2018 :
2019 : /* Write out pgoff_t smallest byte first, prevents endian mismatch */
2020 114444 : for (off = 0; off < sizeof(pgoff_t); off++)
2021 : {
2022 101728 : AH->WriteBytePtr(AH, o & 0xFF);
2023 101728 : o >>= 8;
2024 : }
2025 12716 : return sizeof(pgoff_t) + 1;
2026 : }
2027 :
2028 : int
2029 8220 : ReadOffset(ArchiveHandle *AH, pgoff_t * o)
2030 : {
2031 : int i;
2032 : int off;
2033 : int offsetFlg;
2034 :
2035 : /* Initialize to zero */
2036 8220 : *o = 0;
2037 :
2038 : /* Check for old version */
2039 8220 : if (AH->version < K_VERS_1_7)
2040 : {
2041 : /* Prior versions wrote offsets using WriteInt */
2042 0 : i = ReadInt(AH);
2043 : /* -1 means not set */
2044 0 : if (i < 0)
2045 0 : return K_OFFSET_POS_NOT_SET;
2046 0 : else if (i == 0)
2047 0 : return K_OFFSET_NO_DATA;
2048 :
2049 : /* Cast to pgoff_t because it was written as an int. */
2050 0 : *o = (pgoff_t) i;
2051 0 : return K_OFFSET_POS_SET;
2052 : }
2053 :
2054 : /*
2055 : * Read the flag indicating the state of the data pointer. Check if valid
2056 : * and die if not.
2057 : *
2058 : * This used to be handled by a negative or zero pointer, now we use an
2059 : * extra byte specifically for the state.
2060 : */
2061 8220 : offsetFlg = AH->ReadBytePtr(AH) & 0xFF;
2062 :
2063 8220 : switch (offsetFlg)
2064 : {
2065 8220 : case K_OFFSET_POS_NOT_SET:
2066 : case K_OFFSET_NO_DATA:
2067 : case K_OFFSET_POS_SET:
2068 :
2069 8220 : break;
2070 :
2071 0 : default:
2072 0 : pg_fatal("unexpected data offset flag %d", offsetFlg);
2073 : }
2074 :
2075 : /*
2076 : * Read the bytes
2077 : */
2078 73980 : for (off = 0; off < AH->offSize; off++)
2079 : {
2080 65760 : if (off < sizeof(pgoff_t))
2081 65760 : *o |= ((pgoff_t) (AH->ReadBytePtr(AH))) << (off * 8);
2082 : else
2083 : {
2084 0 : if (AH->ReadBytePtr(AH) != 0)
2085 0 : pg_fatal("file offset in dump file is too large");
2086 : }
2087 : }
2088 :
2089 8220 : return offsetFlg;
2090 : }
2091 :
2092 : size_t
2093 298634 : WriteInt(ArchiveHandle *AH, int i)
2094 : {
2095 : int b;
2096 :
2097 : /*
2098 : * This is a bit yucky, but I don't want to make the binary format very
2099 : * dependent on representation, and not knowing much about it, I write out
2100 : * a sign byte. If you change this, don't forget to change the file
2101 : * version #, and modify ReadInt to read the new format AS WELL AS the old
2102 : * formats.
2103 : */
2104 :
2105 : /* SIGN byte */
2106 298634 : if (i < 0)
2107 : {
2108 62912 : AH->WriteBytePtr(AH, 1);
2109 62912 : i = -i;
2110 : }
2111 : else
2112 235722 : AH->WriteBytePtr(AH, 0);
2113 :
2114 1493170 : for (b = 0; b < AH->intSize; b++)
2115 : {
2116 1194536 : AH->WriteBytePtr(AH, i & 0xFF);
2117 1194536 : i >>= 8;
2118 : }
2119 :
2120 298634 : return AH->intSize + 1;
2121 : }
2122 :
2123 : int
2124 227694 : ReadInt(ArchiveHandle *AH)
2125 : {
2126 227694 : int res = 0;
2127 : int bv,
2128 : b;
2129 227694 : int sign = 0; /* Default positive */
2130 227694 : int bitShift = 0;
2131 :
2132 227694 : if (AH->version > K_VERS_1_0)
2133 : /* Read a sign byte */
2134 227694 : sign = AH->ReadBytePtr(AH);
2135 :
2136 1138470 : for (b = 0; b < AH->intSize; b++)
2137 : {
2138 910776 : bv = AH->ReadBytePtr(AH) & 0xFF;
2139 910776 : if (bv != 0)
2140 213450 : res = res + (bv << bitShift);
2141 910776 : bitShift += 8;
2142 : }
2143 :
2144 227694 : if (sign)
2145 49072 : res = -res;
2146 :
2147 227694 : return res;
2148 : }
2149 :
2150 : size_t
2151 233550 : WriteStr(ArchiveHandle *AH, const char *c)
2152 : {
2153 : size_t res;
2154 :
2155 233550 : if (c)
2156 : {
2157 170638 : int len = strlen(c);
2158 :
2159 170638 : res = WriteInt(AH, len);
2160 170638 : AH->WriteBufPtr(AH, c, len);
2161 170638 : res += len;
2162 : }
2163 : else
2164 62912 : res = WriteInt(AH, -1);
2165 :
2166 233550 : return res;
2167 : }
2168 :
2169 : char *
2170 178242 : ReadStr(ArchiveHandle *AH)
2171 : {
2172 : char *buf;
2173 : int l;
2174 :
2175 178242 : l = ReadInt(AH);
2176 178242 : if (l < 0)
2177 49072 : buf = NULL;
2178 : else
2179 : {
2180 129170 : buf = (char *) pg_malloc(l + 1);
2181 129170 : AH->ReadBufPtr(AH, (void *) buf, l);
2182 :
2183 129170 : buf[l] = '\0';
2184 : }
2185 :
2186 178242 : return buf;
2187 : }
2188 :
2189 : static bool
2190 22 : _fileExistsInDirectory(const char *dir, const char *filename)
2191 : {
2192 : struct stat st;
2193 : char buf[MAXPGPATH];
2194 :
2195 22 : if (snprintf(buf, MAXPGPATH, "%s/%s", dir, filename) >= MAXPGPATH)
2196 0 : pg_fatal("directory name too long: \"%s\"", dir);
2197 :
2198 22 : return (stat(buf, &st) == 0 && S_ISREG(st.st_mode));
2199 : }
2200 :
2201 : static int
2202 56 : _discoverArchiveFormat(ArchiveHandle *AH)
2203 : {
2204 : FILE *fh;
2205 : char sig[6]; /* More than enough */
2206 : size_t cnt;
2207 56 : int wantClose = 0;
2208 :
2209 56 : pg_log_debug("attempting to ascertain archive format");
2210 :
2211 56 : free(AH->lookahead);
2212 :
2213 56 : AH->readHeader = 0;
2214 56 : AH->lookaheadSize = 512;
2215 56 : AH->lookahead = pg_malloc0(512);
2216 56 : AH->lookaheadLen = 0;
2217 56 : AH->lookaheadPos = 0;
2218 :
2219 56 : if (AH->fSpec)
2220 : {
2221 : struct stat st;
2222 :
2223 56 : wantClose = 1;
2224 :
2225 : /*
2226 : * Check if the specified archive is a directory. If so, check if
2227 : * there's a "toc.dat" (or "toc.dat.{gz,lz4,zst}") file in it.
2228 : */
2229 56 : if (stat(AH->fSpec, &st) == 0 && S_ISDIR(st.st_mode))
2230 : {
2231 22 : AH->format = archDirectory;
2232 22 : if (_fileExistsInDirectory(AH->fSpec, "toc.dat"))
2233 22 : return AH->format;
2234 : #ifdef HAVE_LIBZ
2235 0 : if (_fileExistsInDirectory(AH->fSpec, "toc.dat.gz"))
2236 0 : return AH->format;
2237 : #endif
2238 : #ifdef USE_LZ4
2239 0 : if (_fileExistsInDirectory(AH->fSpec, "toc.dat.lz4"))
2240 0 : return AH->format;
2241 : #endif
2242 : #ifdef USE_ZSTD
2243 : if (_fileExistsInDirectory(AH->fSpec, "toc.dat.zst"))
2244 : return AH->format;
2245 : #endif
2246 0 : pg_fatal("directory \"%s\" does not appear to be a valid archive (\"toc.dat\" does not exist)",
2247 : AH->fSpec);
2248 : fh = NULL; /* keep compiler quiet */
2249 : }
2250 : else
2251 : {
2252 34 : fh = fopen(AH->fSpec, PG_BINARY_R);
2253 34 : if (!fh)
2254 0 : pg_fatal("could not open input file \"%s\": %m", AH->fSpec);
2255 : }
2256 : }
2257 : else
2258 : {
2259 0 : fh = stdin;
2260 0 : if (!fh)
2261 0 : pg_fatal("could not open input file: %m");
2262 : }
2263 :
2264 34 : if ((cnt = fread(sig, 1, 5, fh)) != 5)
2265 : {
2266 0 : if (ferror(fh))
2267 0 : pg_fatal("could not read input file: %m");
2268 : else
2269 0 : pg_fatal("input file is too short (read %lu, expected 5)",
2270 : (unsigned long) cnt);
2271 : }
2272 :
2273 : /* Save it, just in case we need it later */
2274 34 : memcpy(&AH->lookahead[0], sig, 5);
2275 34 : AH->lookaheadLen = 5;
2276 :
2277 34 : if (strncmp(sig, "PGDMP", 5) == 0)
2278 : {
2279 : /* It's custom format, stop here */
2280 32 : AH->format = archCustom;
2281 32 : AH->readHeader = 1;
2282 : }
2283 : else
2284 : {
2285 : /*
2286 : * *Maybe* we have a tar archive format file or a text dump ... So,
2287 : * read first 512 byte header...
2288 : */
2289 2 : cnt = fread(&AH->lookahead[AH->lookaheadLen], 1, 512 - AH->lookaheadLen, fh);
2290 : /* read failure is checked below */
2291 2 : AH->lookaheadLen += cnt;
2292 :
2293 2 : if (AH->lookaheadLen >= strlen(TEXT_DUMPALL_HEADER) &&
2294 2 : (strncmp(AH->lookahead, TEXT_DUMP_HEADER, strlen(TEXT_DUMP_HEADER)) == 0 ||
2295 2 : strncmp(AH->lookahead, TEXT_DUMPALL_HEADER, strlen(TEXT_DUMPALL_HEADER)) == 0))
2296 : {
2297 : /*
2298 : * looks like it's probably a text format dump. so suggest they
2299 : * try psql
2300 : */
2301 0 : pg_fatal("input file appears to be a text format dump. Please use psql.");
2302 : }
2303 :
2304 2 : if (AH->lookaheadLen != 512)
2305 : {
2306 0 : if (feof(fh))
2307 0 : pg_fatal("input file does not appear to be a valid archive (too short?)");
2308 : else
2309 0 : READ_ERROR_EXIT(fh);
2310 : }
2311 :
2312 2 : if (!isValidTarHeader(AH->lookahead))
2313 0 : pg_fatal("input file does not appear to be a valid archive");
2314 :
2315 2 : AH->format = archTar;
2316 : }
2317 :
2318 : /* Close the file if we opened it */
2319 34 : if (wantClose)
2320 : {
2321 34 : if (fclose(fh) != 0)
2322 0 : pg_fatal("could not close input file: %m");
2323 : /* Forget lookahead, since we'll re-read header after re-opening */
2324 34 : AH->readHeader = 0;
2325 34 : AH->lookaheadLen = 0;
2326 : }
2327 :
2328 34 : return AH->format;
2329 : }
2330 :
2331 :
2332 : /*
2333 : * Allocate an archive handle
2334 : */
2335 : static ArchiveHandle *
2336 420 : _allocAH(const char *FileSpec, const ArchiveFormat fmt,
2337 : const pg_compress_specification compression_spec,
2338 : bool dosync, ArchiveMode mode,
2339 : SetupWorkerPtrType setupWorkerPtr, DataDirSyncMethod sync_method)
2340 : {
2341 : ArchiveHandle *AH;
2342 : CompressFileHandle *CFH;
2343 420 : pg_compress_specification out_compress_spec = {0};
2344 :
2345 420 : pg_log_debug("allocating AH for %s, format %d",
2346 : FileSpec ? FileSpec : "(stdio)", fmt);
2347 :
2348 420 : AH = (ArchiveHandle *) pg_malloc0(sizeof(ArchiveHandle));
2349 :
2350 420 : AH->version = K_VERS_SELF;
2351 :
2352 : /* initialize for backwards compatible string processing */
2353 420 : AH->public.encoding = 0; /* PG_SQL_ASCII */
2354 420 : AH->public.std_strings = false;
2355 :
2356 : /* sql error handling */
2357 420 : AH->public.exit_on_error = true;
2358 420 : AH->public.n_errors = 0;
2359 :
2360 420 : AH->archiveDumpVersion = PG_VERSION;
2361 :
2362 420 : AH->createDate = time(NULL);
2363 :
2364 420 : AH->intSize = sizeof(int);
2365 420 : AH->offSize = sizeof(pgoff_t);
2366 420 : if (FileSpec)
2367 : {
2368 370 : AH->fSpec = pg_strdup(FileSpec);
2369 :
2370 : /*
2371 : * Not used; maybe later....
2372 : *
2373 : * AH->workDir = pg_strdup(FileSpec); for(i=strlen(FileSpec) ; i > 0 ;
2374 : * i--) if (AH->workDir[i-1] == '/')
2375 : */
2376 : }
2377 : else
2378 50 : AH->fSpec = NULL;
2379 :
2380 420 : AH->currUser = NULL; /* unknown */
2381 420 : AH->currSchema = NULL; /* ditto */
2382 420 : AH->currTablespace = NULL; /* ditto */
2383 420 : AH->currTableAm = NULL; /* ditto */
2384 :
2385 420 : AH->toc = (TocEntry *) pg_malloc0(sizeof(TocEntry));
2386 :
2387 420 : AH->toc->next = AH->toc;
2388 420 : AH->toc->prev = AH->toc;
2389 :
2390 420 : AH->mode = mode;
2391 420 : AH->compression_spec = compression_spec;
2392 420 : AH->dosync = dosync;
2393 420 : AH->sync_method = sync_method;
2394 :
2395 420 : memset(&(AH->sqlparse), 0, sizeof(AH->sqlparse));
2396 :
2397 : /* Open stdout with no compression for AH output handle */
2398 420 : out_compress_spec.algorithm = PG_COMPRESSION_NONE;
2399 420 : CFH = InitCompressFileHandle(out_compress_spec);
2400 420 : if (!CFH->open_func(NULL, fileno(stdout), PG_BINARY_A, CFH))
2401 0 : pg_fatal("could not open stdout for appending: %m");
2402 420 : AH->OF = CFH;
2403 :
2404 : /*
2405 : * On Windows, we need to use binary mode to read/write non-text files,
2406 : * which include all archive formats as well as compressed plain text.
2407 : * Force stdin/stdout into binary mode if that is what we are using.
2408 : */
2409 : #ifdef WIN32
2410 : if ((fmt != archNull || compression_spec.algorithm != PG_COMPRESSION_NONE) &&
2411 : (AH->fSpec == NULL || strcmp(AH->fSpec, "") == 0))
2412 : {
2413 : if (mode == archModeWrite)
2414 : _setmode(fileno(stdout), O_BINARY);
2415 : else
2416 : _setmode(fileno(stdin), O_BINARY);
2417 : }
2418 : #endif
2419 :
2420 420 : AH->SetupWorkerPtr = setupWorkerPtr;
2421 :
2422 420 : if (fmt == archUnknown)
2423 56 : AH->format = _discoverArchiveFormat(AH);
2424 : else
2425 364 : AH->format = fmt;
2426 :
2427 420 : switch (AH->format)
2428 : {
2429 86 : case archCustom:
2430 86 : InitArchiveFmt_Custom(AH);
2431 86 : break;
2432 :
2433 280 : case archNull:
2434 280 : InitArchiveFmt_Null(AH);
2435 280 : break;
2436 :
2437 44 : case archDirectory:
2438 44 : InitArchiveFmt_Directory(AH);
2439 44 : break;
2440 :
2441 10 : case archTar:
2442 10 : InitArchiveFmt_Tar(AH);
2443 8 : break;
2444 :
2445 0 : default:
2446 0 : pg_fatal("unrecognized file format \"%d\"", fmt);
2447 : }
2448 :
2449 418 : return AH;
2450 : }
2451 :
2452 : /*
2453 : * Write out all data (tables & LOs)
2454 : */
2455 : void
2456 62 : WriteDataChunks(ArchiveHandle *AH, ParallelState *pstate)
2457 : {
2458 : TocEntry *te;
2459 :
2460 62 : if (pstate && pstate->numWorkers > 1)
2461 16 : {
2462 : /*
2463 : * In parallel mode, this code runs in the leader process. We
2464 : * construct an array of candidate TEs, then sort it into decreasing
2465 : * size order, then dispatch each TE to a data-transfer worker. By
2466 : * dumping larger tables first, we avoid getting into a situation
2467 : * where we're down to one job and it's big, losing parallelism.
2468 : */
2469 : TocEntry **tes;
2470 : int ntes;
2471 :
2472 16 : tes = (TocEntry **) pg_malloc(AH->tocCount * sizeof(TocEntry *));
2473 16 : ntes = 0;
2474 2028 : for (te = AH->toc->next; te != AH->toc; te = te->next)
2475 : {
2476 : /* Consider only TEs with dataDumper functions ... */
2477 2012 : if (!te->dataDumper)
2478 1776 : continue;
2479 : /* ... and ignore ones not enabled for dump */
2480 236 : if ((te->reqs & REQ_DATA) == 0)
2481 0 : continue;
2482 :
2483 236 : tes[ntes++] = te;
2484 : }
2485 :
2486 16 : if (ntes > 1)
2487 14 : qsort(tes, ntes, sizeof(TocEntry *), TocEntrySizeCompareQsort);
2488 :
2489 252 : for (int i = 0; i < ntes; i++)
2490 236 : DispatchJobForTocEntry(AH, pstate, tes[i], ACT_DUMP,
2491 : mark_dump_job_done, NULL);
2492 :
2493 16 : pg_free(tes);
2494 :
2495 : /* Now wait for workers to finish. */
2496 16 : WaitForWorkers(AH, pstate, WFW_ALL_IDLE);
2497 : }
2498 : else
2499 : {
2500 : /* Non-parallel mode: just dump all candidate TEs sequentially. */
2501 7630 : for (te = AH->toc->next; te != AH->toc; te = te->next)
2502 : {
2503 : /* Must have same filter conditions as above */
2504 7584 : if (!te->dataDumper)
2505 7216 : continue;
2506 368 : if ((te->reqs & REQ_DATA) == 0)
2507 6 : continue;
2508 :
2509 362 : WriteDataChunksForTocEntry(AH, te);
2510 : }
2511 : }
2512 62 : }
2513 :
2514 :
2515 : /*
2516 : * Callback function that's invoked in the leader process after a step has
2517 : * been parallel dumped.
2518 : *
2519 : * We don't need to do anything except check for worker failure.
2520 : */
2521 : static void
2522 236 : mark_dump_job_done(ArchiveHandle *AH,
2523 : TocEntry *te,
2524 : int status,
2525 : void *callback_data)
2526 : {
2527 236 : pg_log_info("finished item %d %s %s",
2528 : te->dumpId, te->desc, te->tag);
2529 :
2530 236 : if (status != 0)
2531 0 : pg_fatal("worker process failed: exit code %d",
2532 : status);
2533 236 : }
2534 :
2535 :
2536 : void
2537 598 : WriteDataChunksForTocEntry(ArchiveHandle *AH, TocEntry *te)
2538 : {
2539 : StartDataPtrType startPtr;
2540 : EndDataPtrType endPtr;
2541 :
2542 598 : AH->currToc = te;
2543 :
2544 598 : if (strcmp(te->desc, "BLOBS") == 0)
2545 : {
2546 32 : startPtr = AH->StartLOsPtr;
2547 32 : endPtr = AH->EndLOsPtr;
2548 : }
2549 : else
2550 : {
2551 566 : startPtr = AH->StartDataPtr;
2552 566 : endPtr = AH->EndDataPtr;
2553 : }
2554 :
2555 598 : if (startPtr != NULL)
2556 598 : (*startPtr) (AH, te);
2557 :
2558 : /*
2559 : * The user-provided DataDumper routine needs to call AH->WriteData
2560 : */
2561 598 : te->dataDumper((Archive *) AH, te->dataDumperArg);
2562 :
2563 598 : if (endPtr != NULL)
2564 598 : (*endPtr) (AH, te);
2565 :
2566 598 : AH->currToc = NULL;
2567 598 : }
2568 :
2569 : void
2570 100 : WriteToc(ArchiveHandle *AH)
2571 : {
2572 : TocEntry *te;
2573 : char workbuf[32];
2574 : int tocCount;
2575 : int i;
2576 :
2577 : /* count entries that will actually be dumped */
2578 100 : tocCount = 0;
2579 16060 : for (te = AH->toc->next; te != AH->toc; te = te->next)
2580 : {
2581 15960 : if ((te->reqs & (REQ_SCHEMA | REQ_DATA | REQ_SPECIAL)) != 0)
2582 15948 : tocCount++;
2583 : }
2584 :
2585 : /* printf("%d TOC Entries to save\n", tocCount); */
2586 :
2587 100 : WriteInt(AH, tocCount);
2588 :
2589 16060 : for (te = AH->toc->next; te != AH->toc; te = te->next)
2590 : {
2591 15960 : if ((te->reqs & (REQ_SCHEMA | REQ_DATA | REQ_SPECIAL)) == 0)
2592 12 : continue;
2593 :
2594 15948 : WriteInt(AH, te->dumpId);
2595 15948 : WriteInt(AH, te->dataDumper ? 1 : 0);
2596 :
2597 : /* OID is recorded as a string for historical reasons */
2598 15948 : sprintf(workbuf, "%u", te->catalogId.tableoid);
2599 15948 : WriteStr(AH, workbuf);
2600 15948 : sprintf(workbuf, "%u", te->catalogId.oid);
2601 15948 : WriteStr(AH, workbuf);
2602 :
2603 15948 : WriteStr(AH, te->tag);
2604 15948 : WriteStr(AH, te->desc);
2605 15948 : WriteInt(AH, te->section);
2606 15948 : WriteStr(AH, te->defn);
2607 15948 : WriteStr(AH, te->dropStmt);
2608 15948 : WriteStr(AH, te->copyStmt);
2609 15948 : WriteStr(AH, te->namespace);
2610 15948 : WriteStr(AH, te->tablespace);
2611 15948 : WriteStr(AH, te->tableam);
2612 15948 : WriteInt(AH, te->relkind);
2613 15948 : WriteStr(AH, te->owner);
2614 15948 : WriteStr(AH, "false");
2615 :
2616 : /* Dump list of dependencies */
2617 38756 : for (i = 0; i < te->nDeps; i++)
2618 : {
2619 22808 : sprintf(workbuf, "%d", te->dependencies[i]);
2620 22808 : WriteStr(AH, workbuf);
2621 : }
2622 15948 : WriteStr(AH, NULL); /* Terminate List */
2623 :
2624 15948 : if (AH->WriteExtraTocPtr)
2625 15948 : AH->WriteExtraTocPtr(AH, te);
2626 : }
2627 100 : }
2628 :
2629 : void
2630 76 : ReadToc(ArchiveHandle *AH)
2631 : {
2632 : int i;
2633 : char *tmp;
2634 : DumpId *deps;
2635 : int depIdx;
2636 : int depSize;
2637 : TocEntry *te;
2638 : bool is_supported;
2639 :
2640 76 : AH->tocCount = ReadInt(AH);
2641 76 : AH->maxDumpId = 0;
2642 :
2643 12108 : for (i = 0; i < AH->tocCount; i++)
2644 : {
2645 12032 : te = (TocEntry *) pg_malloc0(sizeof(TocEntry));
2646 12032 : te->dumpId = ReadInt(AH);
2647 :
2648 12032 : if (te->dumpId > AH->maxDumpId)
2649 2672 : AH->maxDumpId = te->dumpId;
2650 :
2651 : /* Sanity check */
2652 12032 : if (te->dumpId <= 0)
2653 0 : pg_fatal("entry ID %d out of range -- perhaps a corrupt TOC",
2654 : te->dumpId);
2655 :
2656 12032 : te->hadDumper = ReadInt(AH);
2657 :
2658 12032 : if (AH->version >= K_VERS_1_8)
2659 : {
2660 12032 : tmp = ReadStr(AH);
2661 12032 : sscanf(tmp, "%u", &te->catalogId.tableoid);
2662 12032 : free(tmp);
2663 : }
2664 : else
2665 0 : te->catalogId.tableoid = InvalidOid;
2666 12032 : tmp = ReadStr(AH);
2667 12032 : sscanf(tmp, "%u", &te->catalogId.oid);
2668 12032 : free(tmp);
2669 :
2670 12032 : te->tag = ReadStr(AH);
2671 12032 : te->desc = ReadStr(AH);
2672 :
2673 12032 : if (AH->version >= K_VERS_1_11)
2674 : {
2675 12032 : te->section = ReadInt(AH);
2676 : }
2677 : else
2678 : {
2679 : /*
2680 : * Rules for pre-8.4 archives wherein pg_dump hasn't classified
2681 : * the entries into sections. This list need not cover entry
2682 : * types added later than 8.4.
2683 : */
2684 0 : if (strcmp(te->desc, "COMMENT") == 0 ||
2685 0 : strcmp(te->desc, "ACL") == 0 ||
2686 0 : strcmp(te->desc, "ACL LANGUAGE") == 0)
2687 0 : te->section = SECTION_NONE;
2688 0 : else if (strcmp(te->desc, "TABLE DATA") == 0 ||
2689 0 : strcmp(te->desc, "BLOBS") == 0 ||
2690 0 : strcmp(te->desc, "BLOB COMMENTS") == 0)
2691 0 : te->section = SECTION_DATA;
2692 0 : else if (strcmp(te->desc, "CONSTRAINT") == 0 ||
2693 0 : strcmp(te->desc, "CHECK CONSTRAINT") == 0 ||
2694 0 : strcmp(te->desc, "FK CONSTRAINT") == 0 ||
2695 0 : strcmp(te->desc, "INDEX") == 0 ||
2696 0 : strcmp(te->desc, "RULE") == 0 ||
2697 0 : strcmp(te->desc, "TRIGGER") == 0)
2698 0 : te->section = SECTION_POST_DATA;
2699 : else
2700 0 : te->section = SECTION_PRE_DATA;
2701 : }
2702 :
2703 12032 : te->defn = ReadStr(AH);
2704 12032 : te->dropStmt = ReadStr(AH);
2705 :
2706 12032 : if (AH->version >= K_VERS_1_3)
2707 12032 : te->copyStmt = ReadStr(AH);
2708 :
2709 12032 : if (AH->version >= K_VERS_1_6)
2710 12032 : te->namespace = ReadStr(AH);
2711 :
2712 12032 : if (AH->version >= K_VERS_1_10)
2713 12032 : te->tablespace = ReadStr(AH);
2714 :
2715 12032 : if (AH->version >= K_VERS_1_14)
2716 12032 : te->tableam = ReadStr(AH);
2717 :
2718 12032 : if (AH->version >= K_VERS_1_16)
2719 12032 : te->relkind = ReadInt(AH);
2720 :
2721 12032 : te->owner = ReadStr(AH);
2722 12032 : is_supported = true;
2723 12032 : if (AH->version < K_VERS_1_9)
2724 0 : is_supported = false;
2725 : else
2726 : {
2727 12032 : tmp = ReadStr(AH);
2728 :
2729 12032 : if (strcmp(tmp, "true") == 0)
2730 0 : is_supported = false;
2731 :
2732 12032 : free(tmp);
2733 : }
2734 :
2735 12032 : if (!is_supported)
2736 0 : pg_log_warning("restoring tables WITH OIDS is not supported anymore");
2737 :
2738 : /* Read TOC entry dependencies */
2739 12032 : if (AH->version >= K_VERS_1_5)
2740 : {
2741 12032 : depSize = 100;
2742 12032 : deps = (DumpId *) pg_malloc(sizeof(DumpId) * depSize);
2743 12032 : depIdx = 0;
2744 : for (;;)
2745 : {
2746 29818 : tmp = ReadStr(AH);
2747 29818 : if (!tmp)
2748 12032 : break; /* end of list */
2749 17786 : if (depIdx >= depSize)
2750 : {
2751 0 : depSize *= 2;
2752 0 : deps = (DumpId *) pg_realloc(deps, sizeof(DumpId) * depSize);
2753 : }
2754 17786 : sscanf(tmp, "%d", &deps[depIdx]);
2755 17786 : free(tmp);
2756 17786 : depIdx++;
2757 : }
2758 :
2759 12032 : if (depIdx > 0) /* We have a non-null entry */
2760 : {
2761 9604 : deps = (DumpId *) pg_realloc(deps, sizeof(DumpId) * depIdx);
2762 9604 : te->dependencies = deps;
2763 9604 : te->nDeps = depIdx;
2764 : }
2765 : else
2766 : {
2767 2428 : free(deps);
2768 2428 : te->dependencies = NULL;
2769 2428 : te->nDeps = 0;
2770 : }
2771 : }
2772 : else
2773 : {
2774 0 : te->dependencies = NULL;
2775 0 : te->nDeps = 0;
2776 : }
2777 12032 : te->dataLength = 0;
2778 :
2779 12032 : if (AH->ReadExtraTocPtr)
2780 12032 : AH->ReadExtraTocPtr(AH, te);
2781 :
2782 12032 : pg_log_debug("read TOC entry %d (ID %d) for %s %s",
2783 : i, te->dumpId, te->desc, te->tag);
2784 :
2785 : /* link completed entry into TOC circular list */
2786 12032 : te->prev = AH->toc->prev;
2787 12032 : AH->toc->prev->next = te;
2788 12032 : AH->toc->prev = te;
2789 12032 : te->next = AH->toc;
2790 :
2791 : /* special processing immediately upon read for some items */
2792 12032 : if (strcmp(te->desc, "ENCODING") == 0)
2793 76 : processEncodingEntry(AH, te);
2794 11956 : else if (strcmp(te->desc, "STDSTRINGS") == 0)
2795 76 : processStdStringsEntry(AH, te);
2796 11880 : else if (strcmp(te->desc, "SEARCHPATH") == 0)
2797 76 : processSearchPathEntry(AH, te);
2798 : }
2799 76 : }
2800 :
2801 : static void
2802 76 : processEncodingEntry(ArchiveHandle *AH, TocEntry *te)
2803 : {
2804 : /* te->defn should have the form SET client_encoding = 'foo'; */
2805 76 : char *defn = pg_strdup(te->defn);
2806 : char *ptr1;
2807 76 : char *ptr2 = NULL;
2808 : int encoding;
2809 :
2810 76 : ptr1 = strchr(defn, '\'');
2811 76 : if (ptr1)
2812 76 : ptr2 = strchr(++ptr1, '\'');
2813 76 : if (ptr2)
2814 : {
2815 76 : *ptr2 = '\0';
2816 76 : encoding = pg_char_to_encoding(ptr1);
2817 76 : if (encoding < 0)
2818 0 : pg_fatal("unrecognized encoding \"%s\"",
2819 : ptr1);
2820 76 : AH->public.encoding = encoding;
2821 : }
2822 : else
2823 0 : pg_fatal("invalid ENCODING item: %s",
2824 : te->defn);
2825 :
2826 76 : free(defn);
2827 76 : }
2828 :
2829 : static void
2830 76 : processStdStringsEntry(ArchiveHandle *AH, TocEntry *te)
2831 : {
2832 : /* te->defn should have the form SET standard_conforming_strings = 'x'; */
2833 : char *ptr1;
2834 :
2835 76 : ptr1 = strchr(te->defn, '\'');
2836 76 : if (ptr1 && strncmp(ptr1, "'on'", 4) == 0)
2837 76 : AH->public.std_strings = true;
2838 0 : else if (ptr1 && strncmp(ptr1, "'off'", 5) == 0)
2839 0 : AH->public.std_strings = false;
2840 : else
2841 0 : pg_fatal("invalid STDSTRINGS item: %s",
2842 : te->defn);
2843 76 : }
2844 :
2845 : static void
2846 76 : processSearchPathEntry(ArchiveHandle *AH, TocEntry *te)
2847 : {
2848 : /*
2849 : * te->defn should contain a command to set search_path. We just copy it
2850 : * verbatim for use later.
2851 : */
2852 76 : AH->public.searchpath = pg_strdup(te->defn);
2853 76 : }
2854 :
2855 : static void
2856 0 : StrictNamesCheck(RestoreOptions *ropt)
2857 : {
2858 : const char *missing_name;
2859 :
2860 : Assert(ropt->strict_names);
2861 :
2862 0 : if (ropt->schemaNames.head != NULL)
2863 : {
2864 0 : missing_name = simple_string_list_not_touched(&ropt->schemaNames);
2865 0 : if (missing_name != NULL)
2866 0 : pg_fatal("schema \"%s\" not found", missing_name);
2867 : }
2868 :
2869 0 : if (ropt->tableNames.head != NULL)
2870 : {
2871 0 : missing_name = simple_string_list_not_touched(&ropt->tableNames);
2872 0 : if (missing_name != NULL)
2873 0 : pg_fatal("table \"%s\" not found", missing_name);
2874 : }
2875 :
2876 0 : if (ropt->indexNames.head != NULL)
2877 : {
2878 0 : missing_name = simple_string_list_not_touched(&ropt->indexNames);
2879 0 : if (missing_name != NULL)
2880 0 : pg_fatal("index \"%s\" not found", missing_name);
2881 : }
2882 :
2883 0 : if (ropt->functionNames.head != NULL)
2884 : {
2885 0 : missing_name = simple_string_list_not_touched(&ropt->functionNames);
2886 0 : if (missing_name != NULL)
2887 0 : pg_fatal("function \"%s\" not found", missing_name);
2888 : }
2889 :
2890 0 : if (ropt->triggerNames.head != NULL)
2891 : {
2892 0 : missing_name = simple_string_list_not_touched(&ropt->triggerNames);
2893 0 : if (missing_name != NULL)
2894 0 : pg_fatal("trigger \"%s\" not found", missing_name);
2895 : }
2896 0 : }
2897 :
2898 : /*
2899 : * Determine whether we want to restore this TOC entry.
2900 : *
2901 : * Returns 0 if entry should be skipped, or some combination of the
2902 : * REQ_SCHEMA and REQ_DATA bits if we want to restore schema and/or data
2903 : * portions of this TOC entry, or REQ_SPECIAL if it's a special entry.
2904 : */
2905 : static int
2906 65336 : _tocEntryRequired(TocEntry *te, teSection curSection, ArchiveHandle *AH)
2907 : {
2908 65336 : int res = REQ_SCHEMA | REQ_DATA;
2909 65336 : RestoreOptions *ropt = AH->public.ropt;
2910 :
2911 : /* These items are treated specially */
2912 65336 : if (strcmp(te->desc, "ENCODING") == 0 ||
2913 64956 : strcmp(te->desc, "STDSTRINGS") == 0 ||
2914 64576 : strcmp(te->desc, "SEARCHPATH") == 0)
2915 1140 : return REQ_SPECIAL;
2916 :
2917 : /*
2918 : * DATABASE and DATABASE PROPERTIES also have a special rule: they are
2919 : * restored in createDB mode, and not restored otherwise, independently of
2920 : * all else.
2921 : */
2922 64196 : if (strcmp(te->desc, "DATABASE") == 0 ||
2923 64000 : strcmp(te->desc, "DATABASE PROPERTIES") == 0)
2924 : {
2925 252 : if (ropt->createDB)
2926 196 : return REQ_SCHEMA;
2927 : else
2928 56 : return 0;
2929 : }
2930 :
2931 : /*
2932 : * Process exclusions that affect certain classes of TOC entries.
2933 : */
2934 :
2935 : /* If it's an ACL, maybe ignore it */
2936 63944 : if (ropt->aclsSkip && _tocEntryIsACL(te))
2937 0 : return 0;
2938 :
2939 : /* If it's a comment, maybe ignore it */
2940 63944 : if (ropt->no_comments && strcmp(te->desc, "COMMENT") == 0)
2941 0 : return 0;
2942 :
2943 : /*
2944 : * If it's a publication or a table part of a publication, maybe ignore
2945 : * it.
2946 : */
2947 63944 : if (ropt->no_publications &&
2948 0 : (strcmp(te->desc, "PUBLICATION") == 0 ||
2949 0 : strcmp(te->desc, "PUBLICATION TABLE") == 0 ||
2950 0 : strcmp(te->desc, "PUBLICATION TABLES IN SCHEMA") == 0))
2951 0 : return 0;
2952 :
2953 : /* If it's a security label, maybe ignore it */
2954 63944 : if (ropt->no_security_labels && strcmp(te->desc, "SECURITY LABEL") == 0)
2955 0 : return 0;
2956 :
2957 : /* If it's a subscription, maybe ignore it */
2958 63944 : if (ropt->no_subscriptions && strcmp(te->desc, "SUBSCRIPTION") == 0)
2959 0 : return 0;
2960 :
2961 : /* Ignore it if section is not to be dumped/restored */
2962 63944 : switch (curSection)
2963 : {
2964 41638 : case SECTION_PRE_DATA:
2965 41638 : if (!(ropt->dumpSections & DUMP_PRE_DATA))
2966 700 : return 0;
2967 40938 : break;
2968 9380 : case SECTION_DATA:
2969 9380 : if (!(ropt->dumpSections & DUMP_DATA))
2970 172 : return 0;
2971 9208 : break;
2972 12926 : case SECTION_POST_DATA:
2973 12926 : if (!(ropt->dumpSections & DUMP_POST_DATA))
2974 308 : return 0;
2975 12618 : break;
2976 0 : default:
2977 : /* shouldn't get here, really, but ignore it */
2978 0 : return 0;
2979 : }
2980 :
2981 : /* Ignore it if rejected by idWanted[] (cf. SortTocFromFile) */
2982 62764 : if (ropt->idWanted && !ropt->idWanted[te->dumpId - 1])
2983 0 : return 0;
2984 :
2985 : /*
2986 : * Check options for selective dump/restore.
2987 : */
2988 62764 : if (strcmp(te->desc, "ACL") == 0 ||
2989 57940 : strcmp(te->desc, "COMMENT") == 0 ||
2990 52142 : strcmp(te->desc, "SECURITY LABEL") == 0)
2991 : {
2992 : /* Database properties react to createDB, not selectivity options. */
2993 10622 : if (strncmp(te->tag, "DATABASE ", 9) == 0)
2994 : {
2995 118 : if (!ropt->createDB)
2996 38 : return 0;
2997 : }
2998 10504 : else if (ropt->schemaNames.head != NULL ||
2999 10504 : ropt->schemaExcludeNames.head != NULL ||
3000 10504 : ropt->selTypes)
3001 : {
3002 : /*
3003 : * In a selective dump/restore, we want to restore these dependent
3004 : * TOC entry types only if their parent object is being restored.
3005 : * Without selectivity options, we let through everything in the
3006 : * archive. Note there may be such entries with no parent, eg
3007 : * non-default ACLs for built-in objects. Also, we make
3008 : * per-column ACLs additionally depend on the table's ACL if any
3009 : * to ensure correct restore order, so those dependencies should
3010 : * be ignored in this check.
3011 : *
3012 : * This code depends on the parent having been marked already,
3013 : * which should be the case; if it isn't, perhaps due to
3014 : * SortTocFromFile rearrangement, skipping the dependent entry
3015 : * seems prudent anyway.
3016 : *
3017 : * Ideally we'd handle, eg, table CHECK constraints this way too.
3018 : * But it's hard to tell which of their dependencies is the one to
3019 : * consult.
3020 : */
3021 0 : bool dumpthis = false;
3022 :
3023 0 : for (int i = 0; i < te->nDeps; i++)
3024 : {
3025 0 : TocEntry *pte = getTocEntryByDumpId(AH, te->dependencies[i]);
3026 :
3027 0 : if (!pte)
3028 0 : continue; /* probably shouldn't happen */
3029 0 : if (strcmp(pte->desc, "ACL") == 0)
3030 0 : continue; /* ignore dependency on another ACL */
3031 0 : if (pte->reqs == 0)
3032 0 : continue; /* this object isn't marked, so ignore it */
3033 : /* Found a parent to be dumped, so we want to dump this too */
3034 0 : dumpthis = true;
3035 0 : break;
3036 : }
3037 0 : if (!dumpthis)
3038 0 : return 0;
3039 : }
3040 : }
3041 : else
3042 : {
3043 : /* Apply selective-restore rules for standalone TOC entries. */
3044 52142 : if (ropt->schemaNames.head != NULL)
3045 : {
3046 : /* If no namespace is specified, it means all. */
3047 40 : if (!te->namespace)
3048 4 : return 0;
3049 36 : if (!simple_string_list_member(&ropt->schemaNames, te->namespace))
3050 28 : return 0;
3051 : }
3052 :
3053 52110 : if (ropt->schemaExcludeNames.head != NULL &&
3054 76 : te->namespace &&
3055 36 : simple_string_list_member(&ropt->schemaExcludeNames, te->namespace))
3056 8 : return 0;
3057 :
3058 52102 : if (ropt->selTypes)
3059 : {
3060 156 : if (strcmp(te->desc, "TABLE") == 0 ||
3061 116 : strcmp(te->desc, "TABLE DATA") == 0 ||
3062 76 : strcmp(te->desc, "VIEW") == 0 ||
3063 76 : strcmp(te->desc, "FOREIGN TABLE") == 0 ||
3064 76 : strcmp(te->desc, "MATERIALIZED VIEW") == 0 ||
3065 76 : strcmp(te->desc, "MATERIALIZED VIEW DATA") == 0 ||
3066 76 : strcmp(te->desc, "SEQUENCE") == 0 ||
3067 70 : strcmp(te->desc, "SEQUENCE SET") == 0)
3068 : {
3069 92 : if (!ropt->selTable)
3070 60 : return 0;
3071 32 : if (ropt->tableNames.head != NULL &&
3072 32 : !simple_string_list_member(&ropt->tableNames, te->tag))
3073 28 : return 0;
3074 : }
3075 64 : else if (strcmp(te->desc, "INDEX") == 0)
3076 : {
3077 12 : if (!ropt->selIndex)
3078 8 : return 0;
3079 4 : if (ropt->indexNames.head != NULL &&
3080 4 : !simple_string_list_member(&ropt->indexNames, te->tag))
3081 2 : return 0;
3082 : }
3083 52 : else if (strcmp(te->desc, "FUNCTION") == 0 ||
3084 28 : strcmp(te->desc, "AGGREGATE") == 0 ||
3085 28 : strcmp(te->desc, "PROCEDURE") == 0)
3086 : {
3087 24 : if (!ropt->selFunction)
3088 8 : return 0;
3089 16 : if (ropt->functionNames.head != NULL &&
3090 16 : !simple_string_list_member(&ropt->functionNames, te->tag))
3091 12 : return 0;
3092 : }
3093 28 : else if (strcmp(te->desc, "TRIGGER") == 0)
3094 : {
3095 12 : if (!ropt->selTrigger)
3096 8 : return 0;
3097 4 : if (ropt->triggerNames.head != NULL &&
3098 4 : !simple_string_list_member(&ropt->triggerNames, te->tag))
3099 2 : return 0;
3100 : }
3101 : else
3102 16 : return 0;
3103 : }
3104 : }
3105 :
3106 : /*
3107 : * Determine whether the TOC entry contains schema and/or data components,
3108 : * and mask off inapplicable REQ bits. If it had a dataDumper, assume
3109 : * it's both schema and data. Otherwise it's probably schema-only, but
3110 : * there are exceptions.
3111 : */
3112 62542 : if (!te->hadDumper)
3113 : {
3114 : /*
3115 : * Special Case: If 'SEQUENCE SET' or anything to do with LOs, then it
3116 : * is considered a data entry. We don't need to check for BLOBS or
3117 : * old-style BLOB COMMENTS entries, because they will have hadDumper =
3118 : * true ... but we do need to check new-style BLOB ACLs, comments,
3119 : * etc.
3120 : */
3121 54670 : if (strcmp(te->desc, "SEQUENCE SET") == 0 ||
3122 53782 : strcmp(te->desc, "BLOB") == 0 ||
3123 53782 : strcmp(te->desc, "BLOB METADATA") == 0 ||
3124 53590 : (strcmp(te->desc, "ACL") == 0 &&
3125 4824 : strncmp(te->tag, "LARGE OBJECT", 12) == 0) ||
3126 53502 : (strcmp(te->desc, "COMMENT") == 0 &&
3127 5760 : strncmp(te->tag, "LARGE OBJECT", 12) == 0) ||
3128 53388 : (strcmp(te->desc, "SECURITY LABEL") == 0 &&
3129 0 : strncmp(te->tag, "LARGE OBJECT", 12) == 0))
3130 1282 : res = res & REQ_DATA;
3131 : else
3132 53388 : res = res & ~REQ_DATA;
3133 : }
3134 :
3135 : /*
3136 : * If there's no definition command, there's no schema component. Treat
3137 : * "load via partition root" comments as not schema.
3138 : */
3139 62542 : if (!te->defn || !te->defn[0] ||
3140 54694 : strncmp(te->defn, "-- load via partition root ", 27) == 0)
3141 7872 : res = res & ~REQ_SCHEMA;
3142 :
3143 : /*
3144 : * Special case: <Init> type with <Max OID> tag; this is obsolete and we
3145 : * always ignore it.
3146 : */
3147 62542 : if ((strcmp(te->desc, "<Init>") == 0) && (strcmp(te->tag, "Max OID") == 0))
3148 0 : return 0;
3149 :
3150 : /* Mask it if we only want schema */
3151 62542 : if (ropt->schemaOnly)
3152 : {
3153 : /*
3154 : * The sequence_data option overrides schemaOnly for SEQUENCE SET.
3155 : *
3156 : * In binary-upgrade mode, even with schemaOnly set, we do not mask
3157 : * out large objects. (Only large object definitions, comments and
3158 : * other metadata should be generated in binary-upgrade mode, not the
3159 : * actual data, but that need not concern us here.)
3160 : */
3161 4976 : if (!(ropt->sequence_data && strcmp(te->desc, "SEQUENCE SET") == 0) &&
3162 4866 : !(ropt->binary_upgrade &&
3163 4372 : (strcmp(te->desc, "BLOB") == 0 ||
3164 4372 : strcmp(te->desc, "BLOB METADATA") == 0 ||
3165 4366 : (strcmp(te->desc, "ACL") == 0 &&
3166 164 : strncmp(te->tag, "LARGE OBJECT", 12) == 0) ||
3167 4364 : (strcmp(te->desc, "COMMENT") == 0 &&
3168 108 : strncmp(te->tag, "LARGE OBJECT", 12) == 0) ||
3169 4358 : (strcmp(te->desc, "SECURITY LABEL") == 0 &&
3170 0 : strncmp(te->tag, "LARGE OBJECT", 12) == 0))))
3171 4852 : res = res & REQ_SCHEMA;
3172 : }
3173 :
3174 : /* Mask it if we only want data */
3175 62542 : if (ropt->dataOnly)
3176 290 : res = res & REQ_DATA;
3177 :
3178 62542 : return res;
3179 : }
3180 :
3181 : /*
3182 : * Identify which pass we should restore this TOC entry in.
3183 : *
3184 : * See notes with the RestorePass typedef in pg_backup_archiver.h.
3185 : */
3186 : static RestorePass
3187 142724 : _tocEntryRestorePass(TocEntry *te)
3188 : {
3189 : /* "ACL LANGUAGE" was a crock emitted only in PG 7.4 */
3190 142724 : if (strcmp(te->desc, "ACL") == 0 ||
3191 132602 : strcmp(te->desc, "ACL LANGUAGE") == 0 ||
3192 132602 : strcmp(te->desc, "DEFAULT ACL") == 0)
3193 10854 : return RESTORE_PASS_ACL;
3194 131870 : if (strcmp(te->desc, "EVENT TRIGGER") == 0 ||
3195 131656 : strcmp(te->desc, "MATERIALIZED VIEW DATA") == 0)
3196 1652 : return RESTORE_PASS_POST_ACL;
3197 :
3198 : /*
3199 : * Comments need to be emitted in the same pass as their parent objects.
3200 : * ACLs haven't got comments, and neither do matview data objects, but
3201 : * event triggers do. (Fortunately, event triggers haven't got ACLs, or
3202 : * we'd need yet another weird special case.)
3203 : */
3204 130218 : if (strcmp(te->desc, "COMMENT") == 0 &&
3205 11650 : strncmp(te->tag, "EVENT TRIGGER ", 14) == 0)
3206 0 : return RESTORE_PASS_POST_ACL;
3207 :
3208 : /* All else can be handled in the main pass. */
3209 130218 : return RESTORE_PASS_MAIN;
3210 : }
3211 :
3212 : /*
3213 : * Identify TOC entries that are ACLs.
3214 : *
3215 : * Note: it seems worth duplicating some code here to avoid a hard-wired
3216 : * assumption that these are exactly the same entries that we restore during
3217 : * the RESTORE_PASS_ACL phase.
3218 : */
3219 : static bool
3220 52114 : _tocEntryIsACL(TocEntry *te)
3221 : {
3222 : /* "ACL LANGUAGE" was a crock emitted only in PG 7.4 */
3223 52114 : if (strcmp(te->desc, "ACL") == 0 ||
3224 48594 : strcmp(te->desc, "ACL LANGUAGE") == 0 ||
3225 48594 : strcmp(te->desc, "DEFAULT ACL") == 0)
3226 3764 : return true;
3227 48350 : return false;
3228 : }
3229 :
3230 : /*
3231 : * Issue SET commands for parameters that we want to have set the same way
3232 : * at all times during execution of a restore script.
3233 : */
3234 : static void
3235 454 : _doSetFixedOutputState(ArchiveHandle *AH)
3236 : {
3237 454 : RestoreOptions *ropt = AH->public.ropt;
3238 :
3239 : /*
3240 : * Disable timeouts to allow for slow commands, idle parallel workers, etc
3241 : */
3242 454 : ahprintf(AH, "SET statement_timeout = 0;\n");
3243 454 : ahprintf(AH, "SET lock_timeout = 0;\n");
3244 454 : ahprintf(AH, "SET idle_in_transaction_session_timeout = 0;\n");
3245 454 : ahprintf(AH, "SET transaction_timeout = 0;\n");
3246 :
3247 : /* Select the correct character set encoding */
3248 454 : ahprintf(AH, "SET client_encoding = '%s';\n",
3249 : pg_encoding_to_char(AH->public.encoding));
3250 :
3251 : /* Select the correct string literal syntax */
3252 454 : ahprintf(AH, "SET standard_conforming_strings = %s;\n",
3253 454 : AH->public.std_strings ? "on" : "off");
3254 :
3255 : /* Select the role to be used during restore */
3256 454 : if (ropt && ropt->use_role)
3257 0 : ahprintf(AH, "SET ROLE %s;\n", fmtId(ropt->use_role));
3258 :
3259 : /* Select the dump-time search_path */
3260 454 : if (AH->public.searchpath)
3261 454 : ahprintf(AH, "%s", AH->public.searchpath);
3262 :
3263 : /* Make sure function checking is disabled */
3264 454 : ahprintf(AH, "SET check_function_bodies = false;\n");
3265 :
3266 : /* Ensure that all valid XML data will be accepted */
3267 454 : ahprintf(AH, "SET xmloption = content;\n");
3268 :
3269 : /* Avoid annoying notices etc */
3270 454 : ahprintf(AH, "SET client_min_messages = warning;\n");
3271 454 : if (!AH->public.std_strings)
3272 0 : ahprintf(AH, "SET escape_string_warning = off;\n");
3273 :
3274 : /* Adjust row-security state */
3275 454 : if (ropt && ropt->enable_row_security)
3276 0 : ahprintf(AH, "SET row_security = on;\n");
3277 : else
3278 454 : ahprintf(AH, "SET row_security = off;\n");
3279 :
3280 : /*
3281 : * In --transaction-size mode, we should always be in a transaction when
3282 : * we begin to restore objects.
3283 : */
3284 454 : if (ropt && ropt->txn_size > 0)
3285 : {
3286 60 : if (AH->connection)
3287 60 : StartTransaction(&AH->public);
3288 : else
3289 0 : ahprintf(AH, "\nBEGIN;\n");
3290 60 : AH->txnCount = 0;
3291 : }
3292 :
3293 454 : ahprintf(AH, "\n");
3294 454 : }
3295 :
3296 : /*
3297 : * Issue a SET SESSION AUTHORIZATION command. Caller is responsible
3298 : * for updating state if appropriate. If user is NULL or an empty string,
3299 : * the specification DEFAULT will be used.
3300 : */
3301 : static void
3302 2 : _doSetSessionAuth(ArchiveHandle *AH, const char *user)
3303 : {
3304 2 : PQExpBuffer cmd = createPQExpBuffer();
3305 :
3306 2 : appendPQExpBufferStr(cmd, "SET SESSION AUTHORIZATION ");
3307 :
3308 : /*
3309 : * SQL requires a string literal here. Might as well be correct.
3310 : */
3311 2 : if (user && *user)
3312 2 : appendStringLiteralAHX(cmd, user, AH);
3313 : else
3314 0 : appendPQExpBufferStr(cmd, "DEFAULT");
3315 2 : appendPQExpBufferChar(cmd, ';');
3316 :
3317 2 : if (RestoringToDB(AH))
3318 : {
3319 : PGresult *res;
3320 :
3321 0 : res = PQexec(AH->connection, cmd->data);
3322 :
3323 0 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
3324 : /* NOT warn_or_exit_horribly... use -O instead to skip this. */
3325 0 : pg_fatal("could not set session user to \"%s\": %s",
3326 : user, PQerrorMessage(AH->connection));
3327 :
3328 0 : PQclear(res);
3329 : }
3330 : else
3331 2 : ahprintf(AH, "%s\n\n", cmd->data);
3332 :
3333 2 : destroyPQExpBuffer(cmd);
3334 2 : }
3335 :
3336 :
3337 : /*
3338 : * Issue the commands to connect to the specified database.
3339 : *
3340 : * If we're currently restoring right into a database, this will
3341 : * actually establish a connection. Otherwise it puts a \connect into
3342 : * the script output.
3343 : */
3344 : static void
3345 112 : _reconnectToDB(ArchiveHandle *AH, const char *dbname)
3346 : {
3347 112 : if (RestoringToDB(AH))
3348 42 : ReconnectToServer(AH, dbname);
3349 : else
3350 : {
3351 : PQExpBufferData connectbuf;
3352 :
3353 70 : initPQExpBuffer(&connectbuf);
3354 70 : appendPsqlMetaConnect(&connectbuf, dbname);
3355 70 : ahprintf(AH, "%s\n", connectbuf.data);
3356 70 : termPQExpBuffer(&connectbuf);
3357 : }
3358 :
3359 : /*
3360 : * NOTE: currUser keeps track of what the imaginary session user in our
3361 : * script is. It's now effectively reset to the original userID.
3362 : */
3363 112 : free(AH->currUser);
3364 112 : AH->currUser = NULL;
3365 :
3366 : /* don't assume we still know the output schema, tablespace, etc either */
3367 112 : free(AH->currSchema);
3368 112 : AH->currSchema = NULL;
3369 :
3370 112 : free(AH->currTableAm);
3371 112 : AH->currTableAm = NULL;
3372 :
3373 112 : free(AH->currTablespace);
3374 112 : AH->currTablespace = NULL;
3375 :
3376 : /* re-establish fixed state */
3377 112 : _doSetFixedOutputState(AH);
3378 112 : }
3379 :
3380 : /*
3381 : * Become the specified user, and update state to avoid redundant commands
3382 : *
3383 : * NULL or empty argument is taken to mean restoring the session default
3384 : */
3385 : static void
3386 120 : _becomeUser(ArchiveHandle *AH, const char *user)
3387 : {
3388 120 : if (!user)
3389 0 : user = ""; /* avoid null pointers */
3390 :
3391 120 : if (AH->currUser && strcmp(AH->currUser, user) == 0)
3392 118 : return; /* no need to do anything */
3393 :
3394 2 : _doSetSessionAuth(AH, user);
3395 :
3396 : /*
3397 : * NOTE: currUser keeps track of what the imaginary session user in our
3398 : * script is
3399 : */
3400 2 : free(AH->currUser);
3401 2 : AH->currUser = pg_strdup(user);
3402 : }
3403 :
3404 : /*
3405 : * Become the owner of the given TOC entry object. If
3406 : * changes in ownership are not allowed, this doesn't do anything.
3407 : */
3408 : static void
3409 59172 : _becomeOwner(ArchiveHandle *AH, TocEntry *te)
3410 : {
3411 59172 : RestoreOptions *ropt = AH->public.ropt;
3412 :
3413 59172 : if (ropt && (ropt->noOwner || !ropt->use_setsessauth))
3414 59172 : return;
3415 :
3416 0 : _becomeUser(AH, te->owner);
3417 : }
3418 :
3419 :
3420 : /*
3421 : * Issue the commands to select the specified schema as the current schema
3422 : * in the target database.
3423 : */
3424 : static void
3425 59308 : _selectOutputSchema(ArchiveHandle *AH, const char *schemaName)
3426 : {
3427 : PQExpBuffer qry;
3428 :
3429 : /*
3430 : * If there was a SEARCHPATH TOC entry, we're supposed to just stay with
3431 : * that search_path rather than switching to entry-specific paths.
3432 : * Otherwise, it's an old archive that will not restore correctly unless
3433 : * we set the search_path as it's expecting.
3434 : */
3435 59308 : if (AH->public.searchpath)
3436 59308 : return;
3437 :
3438 0 : if (!schemaName || *schemaName == '\0' ||
3439 0 : (AH->currSchema && strcmp(AH->currSchema, schemaName) == 0))
3440 0 : return; /* no need to do anything */
3441 :
3442 0 : qry = createPQExpBuffer();
3443 :
3444 0 : appendPQExpBuffer(qry, "SET search_path = %s",
3445 : fmtId(schemaName));
3446 0 : if (strcmp(schemaName, "pg_catalog") != 0)
3447 0 : appendPQExpBufferStr(qry, ", pg_catalog");
3448 :
3449 0 : if (RestoringToDB(AH))
3450 : {
3451 : PGresult *res;
3452 :
3453 0 : res = PQexec(AH->connection, qry->data);
3454 :
3455 0 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
3456 0 : warn_or_exit_horribly(AH,
3457 : "could not set search_path to \"%s\": %s",
3458 0 : schemaName, PQerrorMessage(AH->connection));
3459 :
3460 0 : PQclear(res);
3461 : }
3462 : else
3463 0 : ahprintf(AH, "%s;\n\n", qry->data);
3464 :
3465 0 : free(AH->currSchema);
3466 0 : AH->currSchema = pg_strdup(schemaName);
3467 :
3468 0 : destroyPQExpBuffer(qry);
3469 : }
3470 :
3471 : /*
3472 : * Issue the commands to select the specified tablespace as the current one
3473 : * in the target database.
3474 : */
3475 : static void
3476 51646 : _selectTablespace(ArchiveHandle *AH, const char *tablespace)
3477 : {
3478 51646 : RestoreOptions *ropt = AH->public.ropt;
3479 : PQExpBuffer qry;
3480 : const char *want,
3481 : *have;
3482 :
3483 : /* do nothing in --no-tablespaces mode */
3484 51646 : if (ropt->noTablespace)
3485 0 : return;
3486 :
3487 51646 : have = AH->currTablespace;
3488 51646 : want = tablespace;
3489 :
3490 : /* no need to do anything for non-tablespace object */
3491 51646 : if (!want)
3492 37874 : return;
3493 :
3494 13772 : if (have && strcmp(want, have) == 0)
3495 13488 : return; /* no need to do anything */
3496 :
3497 284 : qry = createPQExpBuffer();
3498 :
3499 284 : if (strcmp(want, "") == 0)
3500 : {
3501 : /* We want the tablespace to be the database's default */
3502 240 : appendPQExpBufferStr(qry, "SET default_tablespace = ''");
3503 : }
3504 : else
3505 : {
3506 : /* We want an explicit tablespace */
3507 44 : appendPQExpBuffer(qry, "SET default_tablespace = %s", fmtId(want));
3508 : }
3509 :
3510 284 : if (RestoringToDB(AH))
3511 : {
3512 : PGresult *res;
3513 :
3514 22 : res = PQexec(AH->connection, qry->data);
3515 :
3516 22 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
3517 0 : warn_or_exit_horribly(AH,
3518 : "could not set default_tablespace to %s: %s",
3519 0 : fmtId(want), PQerrorMessage(AH->connection));
3520 :
3521 22 : PQclear(res);
3522 : }
3523 : else
3524 262 : ahprintf(AH, "%s;\n\n", qry->data);
3525 :
3526 284 : free(AH->currTablespace);
3527 284 : AH->currTablespace = pg_strdup(want);
3528 :
3529 284 : destroyPQExpBuffer(qry);
3530 : }
3531 :
3532 : /*
3533 : * Set the proper default_table_access_method value for the table.
3534 : */
3535 : static void
3536 50646 : _selectTableAccessMethod(ArchiveHandle *AH, const char *tableam)
3537 : {
3538 50646 : RestoreOptions *ropt = AH->public.ropt;
3539 : PQExpBuffer cmd;
3540 : const char *want,
3541 : *have;
3542 :
3543 : /* do nothing in --no-table-access-method mode */
3544 50646 : if (ropt->noTableAm)
3545 552 : return;
3546 :
3547 50094 : have = AH->currTableAm;
3548 50094 : want = tableam;
3549 :
3550 50094 : if (!want)
3551 41492 : return;
3552 :
3553 8602 : if (have && strcmp(want, have) == 0)
3554 8008 : return;
3555 :
3556 594 : cmd = createPQExpBuffer();
3557 594 : appendPQExpBuffer(cmd, "SET default_table_access_method = %s;", fmtId(want));
3558 :
3559 594 : if (RestoringToDB(AH))
3560 : {
3561 : PGresult *res;
3562 :
3563 22 : res = PQexec(AH->connection, cmd->data);
3564 :
3565 22 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
3566 0 : warn_or_exit_horribly(AH,
3567 : "could not set default_table_access_method: %s",
3568 0 : PQerrorMessage(AH->connection));
3569 :
3570 22 : PQclear(res);
3571 : }
3572 : else
3573 572 : ahprintf(AH, "%s\n\n", cmd->data);
3574 :
3575 594 : destroyPQExpBuffer(cmd);
3576 :
3577 594 : free(AH->currTableAm);
3578 594 : AH->currTableAm = pg_strdup(want);
3579 : }
3580 :
3581 : /*
3582 : * Set the proper default table access method for a table without storage.
3583 : * Currently, this is required only for partitioned tables with a table AM.
3584 : */
3585 : static void
3586 1000 : _printTableAccessMethodNoStorage(ArchiveHandle *AH, TocEntry *te)
3587 : {
3588 1000 : RestoreOptions *ropt = AH->public.ropt;
3589 1000 : const char *tableam = te->tableam;
3590 : PQExpBuffer cmd;
3591 :
3592 : /* do nothing in --no-table-access-method mode */
3593 1000 : if (ropt->noTableAm)
3594 6 : return;
3595 :
3596 994 : if (!tableam)
3597 932 : return;
3598 :
3599 : Assert(te->relkind == RELKIND_PARTITIONED_TABLE);
3600 :
3601 62 : cmd = createPQExpBuffer();
3602 :
3603 62 : appendPQExpBufferStr(cmd, "ALTER TABLE ");
3604 62 : appendPQExpBuffer(cmd, "%s ", fmtQualifiedId(te->namespace, te->tag));
3605 62 : appendPQExpBuffer(cmd, "SET ACCESS METHOD %s;",
3606 : fmtId(tableam));
3607 :
3608 62 : if (RestoringToDB(AH))
3609 : {
3610 : PGresult *res;
3611 :
3612 0 : res = PQexec(AH->connection, cmd->data);
3613 :
3614 0 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
3615 0 : warn_or_exit_horribly(AH,
3616 : "could not alter table access method: %s",
3617 0 : PQerrorMessage(AH->connection));
3618 0 : PQclear(res);
3619 : }
3620 : else
3621 62 : ahprintf(AH, "%s\n\n", cmd->data);
3622 :
3623 62 : destroyPQExpBuffer(cmd);
3624 : }
3625 :
3626 : /*
3627 : * Extract an object description for a TOC entry, and append it to buf.
3628 : *
3629 : * This is used for ALTER ... OWNER TO.
3630 : *
3631 : * If the object type has no owner, do nothing.
3632 : */
3633 : static void
3634 28560 : _getObjectDescription(PQExpBuffer buf, const TocEntry *te)
3635 : {
3636 28560 : const char *type = te->desc;
3637 :
3638 : /* objects that don't require special decoration */
3639 28560 : if (strcmp(type, "COLLATION") == 0 ||
3640 26836 : strcmp(type, "CONVERSION") == 0 ||
3641 26518 : strcmp(type, "DOMAIN") == 0 ||
3642 26266 : strcmp(type, "FOREIGN TABLE") == 0 ||
3643 26200 : strcmp(type, "MATERIALIZED VIEW") == 0 ||
3644 25562 : strcmp(type, "SEQUENCE") == 0 ||
3645 25160 : strcmp(type, "STATISTICS") == 0 ||
3646 24936 : strcmp(type, "TABLE") == 0 ||
3647 15982 : strcmp(type, "TEXT SEARCH DICTIONARY") == 0 ||
3648 15772 : strcmp(type, "TEXT SEARCH CONFIGURATION") == 0 ||
3649 15612 : strcmp(type, "TYPE") == 0 ||
3650 14762 : strcmp(type, "VIEW") == 0 ||
3651 : /* non-schema-specified objects */
3652 14074 : strcmp(type, "DATABASE") == 0 ||
3653 13990 : strcmp(type, "PROCEDURAL LANGUAGE") == 0 ||
3654 13930 : strcmp(type, "SCHEMA") == 0 ||
3655 13630 : strcmp(type, "EVENT TRIGGER") == 0 ||
3656 13560 : strcmp(type, "FOREIGN DATA WRAPPER") == 0 ||
3657 13474 : strcmp(type, "SERVER") == 0 ||
3658 13384 : strcmp(type, "PUBLICATION") == 0 ||
3659 13142 : strcmp(type, "SUBSCRIPTION") == 0)
3660 : {
3661 15602 : appendPQExpBuffer(buf, "%s ", type);
3662 15602 : if (te->namespace && *te->namespace)
3663 14486 : appendPQExpBuffer(buf, "%s.", fmtId(te->namespace));
3664 15602 : appendPQExpBufferStr(buf, fmtId(te->tag));
3665 : }
3666 : /* LOs just have a name, but it's numeric so must not use fmtId */
3667 12958 : else if (strcmp(type, "BLOB") == 0)
3668 : {
3669 0 : appendPQExpBuffer(buf, "LARGE OBJECT %s", te->tag);
3670 : }
3671 :
3672 : /*
3673 : * These object types require additional decoration. Fortunately, the
3674 : * information needed is exactly what's in the DROP command.
3675 : */
3676 12958 : else if (strcmp(type, "AGGREGATE") == 0 ||
3677 12424 : strcmp(type, "FUNCTION") == 0 ||
3678 9346 : strcmp(type, "OPERATOR") == 0 ||
3679 7576 : strcmp(type, "OPERATOR CLASS") == 0 ||
3680 7006 : strcmp(type, "OPERATOR FAMILY") == 0 ||
3681 6520 : strcmp(type, "PROCEDURE") == 0)
3682 : {
3683 : /* Chop "DROP " off the front and make a modifiable copy */
3684 6590 : char *first = pg_strdup(te->dropStmt + 5);
3685 : char *last;
3686 :
3687 : /* point to last character in string */
3688 6590 : last = first + strlen(first) - 1;
3689 :
3690 : /* Strip off any ';' or '\n' at the end */
3691 19770 : while (last >= first && (*last == '\n' || *last == ';'))
3692 13180 : last--;
3693 6590 : *(last + 1) = '\0';
3694 :
3695 6590 : appendPQExpBufferStr(buf, first);
3696 :
3697 6590 : free(first);
3698 6590 : return;
3699 : }
3700 : /* these object types don't have separate owners */
3701 6368 : else if (strcmp(type, "CAST") == 0 ||
3702 6368 : strcmp(type, "CHECK CONSTRAINT") == 0 ||
3703 6318 : strcmp(type, "CONSTRAINT") == 0 ||
3704 4132 : strcmp(type, "DATABASE PROPERTIES") == 0 ||
3705 4126 : strcmp(type, "DEFAULT") == 0 ||
3706 3850 : strcmp(type, "FK CONSTRAINT") == 0 ||
3707 3526 : strcmp(type, "INDEX") == 0 ||
3708 1660 : strcmp(type, "RULE") == 0 ||
3709 1246 : strcmp(type, "TRIGGER") == 0 ||
3710 522 : strcmp(type, "ROW SECURITY") == 0 ||
3711 522 : strcmp(type, "POLICY") == 0 ||
3712 60 : strcmp(type, "USER MAPPING") == 0)
3713 : {
3714 : /* do nothing */
3715 : }
3716 : else
3717 0 : pg_fatal("don't know how to set owner for object type \"%s\"", type);
3718 : }
3719 :
3720 : /*
3721 : * Emit the SQL commands to create the object represented by a TOC entry
3722 : *
3723 : * This now also includes issuing an ALTER OWNER command to restore the
3724 : * object's ownership, if wanted. But note that the object's permissions
3725 : * will remain at default, until the matching ACL TOC entry is restored.
3726 : */
3727 : static void
3728 51646 : _printTocEntry(ArchiveHandle *AH, TocEntry *te, bool isData)
3729 : {
3730 51646 : RestoreOptions *ropt = AH->public.ropt;
3731 :
3732 : /*
3733 : * Select owner, schema, tablespace and default AM as necessary. The
3734 : * default access method for partitioned tables is handled after
3735 : * generating the object definition, as it requires an ALTER command
3736 : * rather than SET.
3737 : */
3738 51646 : _becomeOwner(AH, te);
3739 51646 : _selectOutputSchema(AH, te->namespace);
3740 51646 : _selectTablespace(AH, te->tablespace);
3741 51646 : if (te->relkind != RELKIND_PARTITIONED_TABLE)
3742 50646 : _selectTableAccessMethod(AH, te->tableam);
3743 :
3744 : /* Emit header comment for item */
3745 51646 : if (!AH->noTocComments)
3746 : {
3747 : const char *pfx;
3748 : char *sanitized_name;
3749 : char *sanitized_schema;
3750 : char *sanitized_owner;
3751 :
3752 47554 : if (isData)
3753 7046 : pfx = "Data for ";
3754 : else
3755 40508 : pfx = "";
3756 :
3757 47554 : ahprintf(AH, "--\n");
3758 47554 : if (AH->public.verbose)
3759 : {
3760 1694 : ahprintf(AH, "-- TOC entry %d (class %u OID %u)\n",
3761 : te->dumpId, te->catalogId.tableoid, te->catalogId.oid);
3762 1694 : if (te->nDeps > 0)
3763 : {
3764 : int i;
3765 :
3766 988 : ahprintf(AH, "-- Dependencies:");
3767 2524 : for (i = 0; i < te->nDeps; i++)
3768 1536 : ahprintf(AH, " %d", te->dependencies[i]);
3769 988 : ahprintf(AH, "\n");
3770 : }
3771 : }
3772 :
3773 47554 : sanitized_name = sanitize_line(te->tag, false);
3774 47554 : sanitized_schema = sanitize_line(te->namespace, true);
3775 47554 : sanitized_owner = sanitize_line(ropt->noOwner ? NULL : te->owner, true);
3776 :
3777 47554 : ahprintf(AH, "-- %sName: %s; Type: %s; Schema: %s; Owner: %s",
3778 : pfx, sanitized_name, te->desc, sanitized_schema,
3779 : sanitized_owner);
3780 :
3781 47554 : free(sanitized_name);
3782 47554 : free(sanitized_schema);
3783 47554 : free(sanitized_owner);
3784 :
3785 47554 : if (te->tablespace && strlen(te->tablespace) > 0 && !ropt->noTablespace)
3786 : {
3787 : char *sanitized_tablespace;
3788 :
3789 76 : sanitized_tablespace = sanitize_line(te->tablespace, false);
3790 76 : ahprintf(AH, "; Tablespace: %s", sanitized_tablespace);
3791 76 : free(sanitized_tablespace);
3792 : }
3793 47554 : ahprintf(AH, "\n");
3794 :
3795 47554 : if (AH->PrintExtraTocPtr != NULL)
3796 5800 : AH->PrintExtraTocPtr(AH, te);
3797 47554 : ahprintf(AH, "--\n\n");
3798 : }
3799 :
3800 : /*
3801 : * Actually print the definition. Normally we can just print the defn
3802 : * string if any, but we have three special cases:
3803 : *
3804 : * 1. A crude hack for suppressing AUTHORIZATION clause that old pg_dump
3805 : * versions put into CREATE SCHEMA. Don't mutate the variant for schema
3806 : * "public" that is a comment. We have to do this when --no-owner mode is
3807 : * selected. This is ugly, but I see no other good way ...
3808 : *
3809 : * 2. BLOB METADATA entries need special processing since their defn
3810 : * strings are just lists of OIDs, not complete SQL commands.
3811 : *
3812 : * 3. ACL LARGE OBJECTS entries need special processing because they
3813 : * contain only one copy of the ACL GRANT/REVOKE commands, which we must
3814 : * apply to each large object listed in the associated BLOB METADATA.
3815 : */
3816 51646 : if (ropt->noOwner &&
3817 590 : strcmp(te->desc, "SCHEMA") == 0 && strncmp(te->defn, "--", 2) != 0)
3818 : {
3819 4 : ahprintf(AH, "CREATE SCHEMA %s;\n\n\n", fmtId(te->tag));
3820 : }
3821 51642 : else if (strcmp(te->desc, "BLOB METADATA") == 0)
3822 : {
3823 142 : IssueCommandPerBlob(AH, te, "SELECT pg_catalog.lo_create('", "')");
3824 : }
3825 51500 : else if (strcmp(te->desc, "ACL") == 0 &&
3826 3520 : strncmp(te->tag, "LARGE OBJECTS", 13) == 0)
3827 : {
3828 0 : IssueACLPerBlob(AH, te);
3829 : }
3830 : else
3831 : {
3832 51500 : if (te->defn && strlen(te->defn) > 0)
3833 44434 : ahprintf(AH, "%s\n\n", te->defn);
3834 : }
3835 :
3836 : /*
3837 : * If we aren't using SET SESSION AUTH to determine ownership, we must
3838 : * instead issue an ALTER OWNER command. Schema "public" is special; when
3839 : * a dump emits a comment in lieu of creating it, we use ALTER OWNER even
3840 : * when using SET SESSION for all other objects. We assume that anything
3841 : * without a DROP command is not a separately ownable object.
3842 : */
3843 51646 : if (!ropt->noOwner &&
3844 51056 : (!ropt->use_setsessauth ||
3845 0 : (strcmp(te->desc, "SCHEMA") == 0 &&
3846 0 : strncmp(te->defn, "--", 2) == 0)) &&
3847 51056 : te->owner && strlen(te->owner) > 0 &&
3848 50376 : te->dropStmt && strlen(te->dropStmt) > 0)
3849 : {
3850 28698 : if (strcmp(te->desc, "BLOB METADATA") == 0)
3851 : {
3852 : /* BLOB METADATA needs special code to handle multiple LOs */
3853 138 : char *cmdEnd = psprintf(" OWNER TO %s", fmtId(te->owner));
3854 :
3855 138 : IssueCommandPerBlob(AH, te, "ALTER LARGE OBJECT ", cmdEnd);
3856 138 : pg_free(cmdEnd);
3857 : }
3858 : else
3859 : {
3860 : /* For all other cases, we can use _getObjectDescription */
3861 : PQExpBufferData temp;
3862 :
3863 28560 : initPQExpBuffer(&temp);
3864 28560 : _getObjectDescription(&temp, te);
3865 :
3866 : /*
3867 : * If _getObjectDescription() didn't fill the buffer, then there
3868 : * is no owner.
3869 : */
3870 28560 : if (temp.data[0])
3871 22192 : ahprintf(AH, "ALTER %s OWNER TO %s;\n\n",
3872 22192 : temp.data, fmtId(te->owner));
3873 28560 : termPQExpBuffer(&temp);
3874 : }
3875 : }
3876 :
3877 : /*
3878 : * Select a partitioned table's default AM, once the table definition has
3879 : * been generated.
3880 : */
3881 51646 : if (te->relkind == RELKIND_PARTITIONED_TABLE)
3882 1000 : _printTableAccessMethodNoStorage(AH, te);
3883 :
3884 : /*
3885 : * If it's an ACL entry, it might contain SET SESSION AUTHORIZATION
3886 : * commands, so we can no longer assume we know the current auth setting.
3887 : */
3888 51646 : if (_tocEntryIsACL(te))
3889 : {
3890 3764 : free(AH->currUser);
3891 3764 : AH->currUser = NULL;
3892 : }
3893 51646 : }
3894 :
3895 : /*
3896 : * Sanitize a string to be included in an SQL comment or TOC listing, by
3897 : * replacing any newlines with spaces. This ensures each logical output line
3898 : * is in fact one physical output line, to prevent corruption of the dump
3899 : * (which could, in the worst case, present an SQL injection vulnerability
3900 : * if someone were to incautiously load a dump containing objects with
3901 : * maliciously crafted names).
3902 : *
3903 : * The result is a freshly malloc'd string. If the input string is NULL,
3904 : * return a malloc'ed empty string, unless want_hyphen, in which case return a
3905 : * malloc'ed hyphen.
3906 : *
3907 : * Note that we currently don't bother to quote names, meaning that the name
3908 : * fields aren't automatically parseable. "pg_restore -L" doesn't care because
3909 : * it only examines the dumpId field, but someday we might want to try harder.
3910 : */
3911 : static char *
3912 149442 : sanitize_line(const char *str, bool want_hyphen)
3913 : {
3914 : char *result;
3915 : char *s;
3916 :
3917 149442 : if (!str)
3918 3950 : return pg_strdup(want_hyphen ? "-" : "");
3919 :
3920 145492 : result = pg_strdup(str);
3921 :
3922 1843640 : for (s = result; *s != '\0'; s++)
3923 : {
3924 1698148 : if (*s == '\n' || *s == '\r')
3925 120 : *s = ' ';
3926 : }
3927 :
3928 145492 : return result;
3929 : }
3930 :
3931 : /*
3932 : * Write the file header for a custom-format archive
3933 : */
3934 : void
3935 62 : WriteHead(ArchiveHandle *AH)
3936 : {
3937 : struct tm crtm;
3938 :
3939 62 : AH->WriteBufPtr(AH, "PGDMP", 5); /* Magic code */
3940 62 : AH->WriteBytePtr(AH, ARCHIVE_MAJOR(AH->version));
3941 62 : AH->WriteBytePtr(AH, ARCHIVE_MINOR(AH->version));
3942 62 : AH->WriteBytePtr(AH, ARCHIVE_REV(AH->version));
3943 62 : AH->WriteBytePtr(AH, AH->intSize);
3944 62 : AH->WriteBytePtr(AH, AH->offSize);
3945 62 : AH->WriteBytePtr(AH, AH->format);
3946 62 : AH->WriteBytePtr(AH, AH->compression_spec.algorithm);
3947 62 : crtm = *localtime(&AH->createDate);
3948 62 : WriteInt(AH, crtm.tm_sec);
3949 62 : WriteInt(AH, crtm.tm_min);
3950 62 : WriteInt(AH, crtm.tm_hour);
3951 62 : WriteInt(AH, crtm.tm_mday);
3952 62 : WriteInt(AH, crtm.tm_mon);
3953 62 : WriteInt(AH, crtm.tm_year);
3954 62 : WriteInt(AH, crtm.tm_isdst);
3955 62 : WriteStr(AH, PQdb(AH->connection));
3956 62 : WriteStr(AH, AH->public.remoteVersionStr);
3957 62 : WriteStr(AH, PG_VERSION);
3958 62 : }
3959 :
3960 : void
3961 76 : ReadHead(ArchiveHandle *AH)
3962 : {
3963 : char *errmsg;
3964 : char vmaj,
3965 : vmin,
3966 : vrev;
3967 : int fmt;
3968 :
3969 : /*
3970 : * If we haven't already read the header, do so.
3971 : *
3972 : * NB: this code must agree with _discoverArchiveFormat(). Maybe find a
3973 : * way to unify the cases?
3974 : */
3975 76 : if (!AH->readHeader)
3976 : {
3977 : char tmpMag[7];
3978 :
3979 76 : AH->ReadBufPtr(AH, tmpMag, 5);
3980 :
3981 76 : if (strncmp(tmpMag, "PGDMP", 5) != 0)
3982 0 : pg_fatal("did not find magic string in file header");
3983 : }
3984 :
3985 76 : vmaj = AH->ReadBytePtr(AH);
3986 76 : vmin = AH->ReadBytePtr(AH);
3987 :
3988 76 : if (vmaj > 1 || (vmaj == 1 && vmin > 0)) /* Version > 1.0 */
3989 76 : vrev = AH->ReadBytePtr(AH);
3990 : else
3991 0 : vrev = 0;
3992 :
3993 76 : AH->version = MAKE_ARCHIVE_VERSION(vmaj, vmin, vrev);
3994 :
3995 76 : if (AH->version < K_VERS_1_0 || AH->version > K_VERS_MAX)
3996 0 : pg_fatal("unsupported version (%d.%d) in file header",
3997 : vmaj, vmin);
3998 :
3999 76 : AH->intSize = AH->ReadBytePtr(AH);
4000 76 : if (AH->intSize > 32)
4001 0 : pg_fatal("sanity check on integer size (%lu) failed",
4002 : (unsigned long) AH->intSize);
4003 :
4004 76 : if (AH->intSize > sizeof(int))
4005 0 : pg_log_warning("archive was made on a machine with larger integers, some operations might fail");
4006 :
4007 76 : if (AH->version >= K_VERS_1_7)
4008 76 : AH->offSize = AH->ReadBytePtr(AH);
4009 : else
4010 0 : AH->offSize = AH->intSize;
4011 :
4012 76 : fmt = AH->ReadBytePtr(AH);
4013 :
4014 76 : if (AH->format != fmt)
4015 0 : pg_fatal("expected format (%d) differs from format found in file (%d)",
4016 : AH->format, fmt);
4017 :
4018 76 : if (AH->version >= K_VERS_1_15)
4019 76 : AH->compression_spec.algorithm = AH->ReadBytePtr(AH);
4020 0 : else if (AH->version >= K_VERS_1_2)
4021 : {
4022 : /* Guess the compression method based on the level */
4023 0 : if (AH->version < K_VERS_1_4)
4024 0 : AH->compression_spec.level = AH->ReadBytePtr(AH);
4025 : else
4026 0 : AH->compression_spec.level = ReadInt(AH);
4027 :
4028 0 : if (AH->compression_spec.level != 0)
4029 0 : AH->compression_spec.algorithm = PG_COMPRESSION_GZIP;
4030 : }
4031 : else
4032 0 : AH->compression_spec.algorithm = PG_COMPRESSION_GZIP;
4033 :
4034 76 : errmsg = supports_compression(AH->compression_spec);
4035 76 : if (errmsg)
4036 : {
4037 0 : pg_log_warning("archive is compressed, but this installation does not support compression (%s) -- no data will be available",
4038 : errmsg);
4039 0 : pg_free(errmsg);
4040 : }
4041 :
4042 76 : if (AH->version >= K_VERS_1_4)
4043 : {
4044 : struct tm crtm;
4045 :
4046 76 : crtm.tm_sec = ReadInt(AH);
4047 76 : crtm.tm_min = ReadInt(AH);
4048 76 : crtm.tm_hour = ReadInt(AH);
4049 76 : crtm.tm_mday = ReadInt(AH);
4050 76 : crtm.tm_mon = ReadInt(AH);
4051 76 : crtm.tm_year = ReadInt(AH);
4052 76 : crtm.tm_isdst = ReadInt(AH);
4053 :
4054 : /*
4055 : * Newer versions of glibc have mktime() report failure if tm_isdst is
4056 : * inconsistent with the prevailing timezone, e.g. tm_isdst = 1 when
4057 : * TZ=UTC. This is problematic when restoring an archive under a
4058 : * different timezone setting. If we get a failure, try again with
4059 : * tm_isdst set to -1 ("don't know").
4060 : *
4061 : * XXX with or without this hack, we reconstruct createDate
4062 : * incorrectly when the prevailing timezone is different from
4063 : * pg_dump's. Next time we bump the archive version, we should flush
4064 : * this representation and store a plain seconds-since-the-Epoch
4065 : * timestamp instead.
4066 : */
4067 76 : AH->createDate = mktime(&crtm);
4068 76 : if (AH->createDate == (time_t) -1)
4069 : {
4070 0 : crtm.tm_isdst = -1;
4071 0 : AH->createDate = mktime(&crtm);
4072 0 : if (AH->createDate == (time_t) -1)
4073 0 : pg_log_warning("invalid creation date in header");
4074 : }
4075 : }
4076 :
4077 76 : if (AH->version >= K_VERS_1_4)
4078 : {
4079 76 : AH->archdbname = ReadStr(AH);
4080 : }
4081 :
4082 76 : if (AH->version >= K_VERS_1_10)
4083 : {
4084 76 : AH->archiveRemoteVersion = ReadStr(AH);
4085 76 : AH->archiveDumpVersion = ReadStr(AH);
4086 : }
4087 76 : }
4088 :
4089 :
4090 : /*
4091 : * checkSeek
4092 : * check to see if ftell/fseek can be performed.
4093 : */
4094 : bool
4095 96 : checkSeek(FILE *fp)
4096 : {
4097 : pgoff_t tpos;
4098 :
4099 : /* Check that ftello works on this file */
4100 96 : tpos = ftello(fp);
4101 96 : if (tpos < 0)
4102 2 : return false;
4103 :
4104 : /*
4105 : * Check that fseeko(SEEK_SET) works, too. NB: we used to try to test
4106 : * this with fseeko(fp, 0, SEEK_CUR). But some platforms treat that as a
4107 : * successful no-op even on files that are otherwise unseekable.
4108 : */
4109 94 : if (fseeko(fp, tpos, SEEK_SET) != 0)
4110 0 : return false;
4111 :
4112 94 : return true;
4113 : }
4114 :
4115 :
4116 : /*
4117 : * dumpTimestamp
4118 : */
4119 : static void
4120 84 : dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim)
4121 : {
4122 : char buf[64];
4123 :
4124 84 : if (strftime(buf, sizeof(buf), PGDUMP_STRFTIME_FMT, localtime(&tim)) != 0)
4125 84 : ahprintf(AH, "-- %s %s\n\n", msg, buf);
4126 84 : }
4127 :
4128 : /*
4129 : * Main engine for parallel restore.
4130 : *
4131 : * Parallel restore is done in three phases. In this first phase,
4132 : * we'll process all SECTION_PRE_DATA TOC entries that are allowed to be
4133 : * processed in the RESTORE_PASS_MAIN pass. (In practice, that's all
4134 : * PRE_DATA items other than ACLs.) Entries we can't process now are
4135 : * added to the pending_list for later phases to deal with.
4136 : */
4137 : static void
4138 8 : restore_toc_entries_prefork(ArchiveHandle *AH, TocEntry *pending_list)
4139 : {
4140 : bool skipped_some;
4141 : TocEntry *next_work_item;
4142 :
4143 8 : pg_log_debug("entering restore_toc_entries_prefork");
4144 :
4145 : /* Adjust dependency information */
4146 8 : fix_dependencies(AH);
4147 :
4148 : /*
4149 : * Do all the early stuff in a single connection in the parent. There's no
4150 : * great point in running it in parallel, in fact it will actually run
4151 : * faster in a single connection because we avoid all the connection and
4152 : * setup overhead. Also, pre-9.2 pg_dump versions were not very good
4153 : * about showing all the dependencies of SECTION_PRE_DATA items, so we do
4154 : * not risk trying to process them out-of-order.
4155 : *
4156 : * Stuff that we can't do immediately gets added to the pending_list.
4157 : * Note: we don't yet filter out entries that aren't going to be restored.
4158 : * They might participate in dependency chains connecting entries that
4159 : * should be restored, so we treat them as live until we actually process
4160 : * them.
4161 : *
4162 : * Note: as of 9.2, it should be guaranteed that all PRE_DATA items appear
4163 : * before DATA items, and all DATA items before POST_DATA items. That is
4164 : * not certain to be true in older archives, though, and in any case use
4165 : * of a list file would destroy that ordering (cf. SortTocFromFile). So
4166 : * this loop cannot assume that it holds.
4167 : */
4168 8 : AH->restorePass = RESTORE_PASS_MAIN;
4169 8 : skipped_some = false;
4170 200 : for (next_work_item = AH->toc->next; next_work_item != AH->toc; next_work_item = next_work_item->next)
4171 : {
4172 192 : bool do_now = true;
4173 :
4174 192 : if (next_work_item->section != SECTION_PRE_DATA)
4175 : {
4176 : /* DATA and POST_DATA items are just ignored for now */
4177 92 : if (next_work_item->section == SECTION_DATA ||
4178 60 : next_work_item->section == SECTION_POST_DATA)
4179 : {
4180 92 : do_now = false;
4181 92 : skipped_some = true;
4182 : }
4183 : else
4184 : {
4185 : /*
4186 : * SECTION_NONE items, such as comments, can be processed now
4187 : * if we are still in the PRE_DATA part of the archive. Once
4188 : * we've skipped any items, we have to consider whether the
4189 : * comment's dependencies are satisfied, so skip it for now.
4190 : */
4191 0 : if (skipped_some)
4192 0 : do_now = false;
4193 : }
4194 : }
4195 :
4196 : /*
4197 : * Also skip items that need to be forced into later passes. We need
4198 : * not set skipped_some in this case, since by assumption no main-pass
4199 : * items could depend on these.
4200 : */
4201 192 : if (_tocEntryRestorePass(next_work_item) != RESTORE_PASS_MAIN)
4202 0 : do_now = false;
4203 :
4204 192 : if (do_now)
4205 : {
4206 : /* OK, restore the item and update its dependencies */
4207 100 : pg_log_info("processing item %d %s %s",
4208 : next_work_item->dumpId,
4209 : next_work_item->desc, next_work_item->tag);
4210 :
4211 100 : (void) restore_toc_entry(AH, next_work_item, false);
4212 :
4213 : /* Reduce dependencies, but don't move anything to ready_heap */
4214 100 : reduce_dependencies(AH, next_work_item, NULL);
4215 : }
4216 : else
4217 : {
4218 : /* Nope, so add it to pending_list */
4219 92 : pending_list_append(pending_list, next_work_item);
4220 : }
4221 : }
4222 :
4223 : /*
4224 : * In --transaction-size mode, we must commit the open transaction before
4225 : * dropping the database connection. This also ensures that child workers
4226 : * can see the objects we've created so far.
4227 : */
4228 8 : if (AH->public.ropt->txn_size > 0)
4229 0 : CommitTransaction(&AH->public);
4230 :
4231 : /*
4232 : * Now close parent connection in prep for parallel steps. We do this
4233 : * mainly to ensure that we don't exceed the specified number of parallel
4234 : * connections.
4235 : */
4236 8 : DisconnectDatabase(&AH->public);
4237 :
4238 : /* blow away any transient state from the old connection */
4239 8 : free(AH->currUser);
4240 8 : AH->currUser = NULL;
4241 8 : free(AH->currSchema);
4242 8 : AH->currSchema = NULL;
4243 8 : free(AH->currTablespace);
4244 8 : AH->currTablespace = NULL;
4245 8 : free(AH->currTableAm);
4246 8 : AH->currTableAm = NULL;
4247 8 : }
4248 :
4249 : /*
4250 : * Main engine for parallel restore.
4251 : *
4252 : * Parallel restore is done in three phases. In this second phase,
4253 : * we process entries by dispatching them to parallel worker children
4254 : * (processes on Unix, threads on Windows), each of which connects
4255 : * separately to the database. Inter-entry dependencies are respected,
4256 : * and so is the RestorePass multi-pass structure. When we can no longer
4257 : * make any entries ready to process, we exit. Normally, there will be
4258 : * nothing left to do; but if there is, the third phase will mop up.
4259 : */
4260 : static void
4261 8 : restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
4262 : TocEntry *pending_list)
4263 : {
4264 : binaryheap *ready_heap;
4265 : TocEntry *next_work_item;
4266 :
4267 8 : pg_log_debug("entering restore_toc_entries_parallel");
4268 :
4269 : /* Set up ready_heap with enough room for all known TocEntrys */
4270 8 : ready_heap = binaryheap_allocate(AH->tocCount,
4271 : TocEntrySizeCompareBinaryheap,
4272 : NULL);
4273 :
4274 : /*
4275 : * The pending_list contains all items that we need to restore. Move all
4276 : * items that are available to process immediately into the ready_heap.
4277 : * After this setup, the pending list is everything that needs to be done
4278 : * but is blocked by one or more dependencies, while the ready heap
4279 : * contains items that have no remaining dependencies and are OK to
4280 : * process in the current restore pass.
4281 : */
4282 8 : AH->restorePass = RESTORE_PASS_MAIN;
4283 8 : move_to_ready_heap(pending_list, ready_heap, AH->restorePass);
4284 :
4285 : /*
4286 : * main parent loop
4287 : *
4288 : * Keep going until there is no worker still running AND there is no work
4289 : * left to be done. Note invariant: at top of loop, there should always
4290 : * be at least one worker available to dispatch a job to.
4291 : */
4292 8 : pg_log_info("entering main parallel loop");
4293 :
4294 : for (;;)
4295 : {
4296 : /* Look for an item ready to be dispatched to a worker */
4297 140 : next_work_item = pop_next_work_item(ready_heap, pstate);
4298 140 : if (next_work_item != NULL)
4299 : {
4300 : /* If not to be restored, don't waste time launching a worker */
4301 92 : if ((next_work_item->reqs & (REQ_SCHEMA | REQ_DATA)) == 0)
4302 : {
4303 0 : pg_log_info("skipping item %d %s %s",
4304 : next_work_item->dumpId,
4305 : next_work_item->desc, next_work_item->tag);
4306 : /* Update its dependencies as though we'd completed it */
4307 0 : reduce_dependencies(AH, next_work_item, ready_heap);
4308 : /* Loop around to see if anything else can be dispatched */
4309 0 : continue;
4310 : }
4311 :
4312 92 : pg_log_info("launching item %d %s %s",
4313 : next_work_item->dumpId,
4314 : next_work_item->desc, next_work_item->tag);
4315 :
4316 : /* Dispatch to some worker */
4317 92 : DispatchJobForTocEntry(AH, pstate, next_work_item, ACT_RESTORE,
4318 : mark_restore_job_done, ready_heap);
4319 : }
4320 48 : else if (IsEveryWorkerIdle(pstate))
4321 : {
4322 : /*
4323 : * Nothing is ready and no worker is running, so we're done with
4324 : * the current pass or maybe with the whole process.
4325 : */
4326 24 : if (AH->restorePass == RESTORE_PASS_LAST)
4327 8 : break; /* No more parallel processing is possible */
4328 :
4329 : /* Advance to next restore pass */
4330 16 : AH->restorePass++;
4331 : /* That probably allows some stuff to be made ready */
4332 16 : move_to_ready_heap(pending_list, ready_heap, AH->restorePass);
4333 : /* Loop around to see if anything's now ready */
4334 16 : continue;
4335 : }
4336 : else
4337 : {
4338 : /*
4339 : * We have nothing ready, but at least one child is working, so
4340 : * wait for some subjob to finish.
4341 : */
4342 : }
4343 :
4344 : /*
4345 : * Before dispatching another job, check to see if anything has
4346 : * finished. We should check every time through the loop so as to
4347 : * reduce dependencies as soon as possible. If we were unable to
4348 : * dispatch any job this time through, wait until some worker finishes
4349 : * (and, hopefully, unblocks some pending item). If we did dispatch
4350 : * something, continue as soon as there's at least one idle worker.
4351 : * Note that in either case, there's guaranteed to be at least one
4352 : * idle worker when we return to the top of the loop. This ensures we
4353 : * won't block inside DispatchJobForTocEntry, which would be
4354 : * undesirable: we'd rather postpone dispatching until we see what's
4355 : * been unblocked by finished jobs.
4356 : */
4357 116 : WaitForWorkers(AH, pstate,
4358 : next_work_item ? WFW_ONE_IDLE : WFW_GOT_STATUS);
4359 : }
4360 :
4361 : /* There should now be nothing in ready_heap. */
4362 : Assert(binaryheap_empty(ready_heap));
4363 :
4364 8 : binaryheap_free(ready_heap);
4365 :
4366 8 : pg_log_info("finished main parallel loop");
4367 8 : }
4368 :
4369 : /*
4370 : * Main engine for parallel restore.
4371 : *
4372 : * Parallel restore is done in three phases. In this third phase,
4373 : * we mop up any remaining TOC entries by processing them serially.
4374 : * This phase normally should have nothing to do, but if we've somehow
4375 : * gotten stuck due to circular dependencies or some such, this provides
4376 : * at least some chance of completing the restore successfully.
4377 : */
4378 : static void
4379 8 : restore_toc_entries_postfork(ArchiveHandle *AH, TocEntry *pending_list)
4380 : {
4381 8 : RestoreOptions *ropt = AH->public.ropt;
4382 : TocEntry *te;
4383 :
4384 8 : pg_log_debug("entering restore_toc_entries_postfork");
4385 :
4386 : /*
4387 : * Now reconnect the single parent connection.
4388 : */
4389 8 : ConnectDatabase((Archive *) AH, &ropt->cparams, true);
4390 :
4391 : /* re-establish fixed state */
4392 8 : _doSetFixedOutputState(AH);
4393 :
4394 : /*
4395 : * Make sure there is no work left due to, say, circular dependencies, or
4396 : * some other pathological condition. If so, do it in the single parent
4397 : * connection. We don't sweat about RestorePass ordering; it's likely we
4398 : * already violated that.
4399 : */
4400 8 : for (te = pending_list->pending_next; te != pending_list; te = te->pending_next)
4401 : {
4402 0 : pg_log_info("processing missed item %d %s %s",
4403 : te->dumpId, te->desc, te->tag);
4404 0 : (void) restore_toc_entry(AH, te, false);
4405 : }
4406 8 : }
4407 :
4408 : /*
4409 : * Check if te1 has an exclusive lock requirement for an item that te2 also
4410 : * requires, whether or not te2's requirement is for an exclusive lock.
4411 : */
4412 : static bool
4413 290 : has_lock_conflicts(TocEntry *te1, TocEntry *te2)
4414 : {
4415 : int j,
4416 : k;
4417 :
4418 622 : for (j = 0; j < te1->nLockDeps; j++)
4419 : {
4420 1024 : for (k = 0; k < te2->nDeps; k++)
4421 : {
4422 692 : if (te1->lockDeps[j] == te2->dependencies[k])
4423 2 : return true;
4424 : }
4425 : }
4426 288 : return false;
4427 : }
4428 :
4429 :
4430 : /*
4431 : * Initialize the header of the pending-items list.
4432 : *
4433 : * This is a circular list with a dummy TocEntry as header, just like the
4434 : * main TOC list; but we use separate list links so that an entry can be in
4435 : * the main TOC list as well as in the pending list.
4436 : */
4437 : static void
4438 8 : pending_list_header_init(TocEntry *l)
4439 : {
4440 8 : l->pending_prev = l->pending_next = l;
4441 8 : }
4442 :
4443 : /* Append te to the end of the pending-list headed by l */
4444 : static void
4445 92 : pending_list_append(TocEntry *l, TocEntry *te)
4446 : {
4447 92 : te->pending_prev = l->pending_prev;
4448 92 : l->pending_prev->pending_next = te;
4449 92 : l->pending_prev = te;
4450 92 : te->pending_next = l;
4451 92 : }
4452 :
4453 : /* Remove te from the pending-list */
4454 : static void
4455 92 : pending_list_remove(TocEntry *te)
4456 : {
4457 92 : te->pending_prev->pending_next = te->pending_next;
4458 92 : te->pending_next->pending_prev = te->pending_prev;
4459 92 : te->pending_prev = NULL;
4460 92 : te->pending_next = NULL;
4461 92 : }
4462 :
4463 :
4464 : /* qsort comparator for sorting TocEntries by dataLength */
4465 : static int
4466 1336 : TocEntrySizeCompareQsort(const void *p1, const void *p2)
4467 : {
4468 1336 : const TocEntry *te1 = *(const TocEntry *const *) p1;
4469 1336 : const TocEntry *te2 = *(const TocEntry *const *) p2;
4470 :
4471 : /* Sort by decreasing dataLength */
4472 1336 : if (te1->dataLength > te2->dataLength)
4473 118 : return -1;
4474 1218 : if (te1->dataLength < te2->dataLength)
4475 230 : return 1;
4476 :
4477 : /* For equal dataLengths, sort by dumpId, just to be stable */
4478 988 : if (te1->dumpId < te2->dumpId)
4479 338 : return -1;
4480 650 : if (te1->dumpId > te2->dumpId)
4481 620 : return 1;
4482 :
4483 30 : return 0;
4484 : }
4485 :
4486 : /* binaryheap comparator for sorting TocEntries by dataLength */
4487 : static int
4488 300 : TocEntrySizeCompareBinaryheap(void *p1, void *p2, void *arg)
4489 : {
4490 : /* return opposite of qsort comparator for max-heap */
4491 300 : return -TocEntrySizeCompareQsort(&p1, &p2);
4492 : }
4493 :
4494 :
4495 : /*
4496 : * Move all immediately-ready items from pending_list to ready_heap.
4497 : *
4498 : * Items are considered ready if they have no remaining dependencies and
4499 : * they belong in the current restore pass. (See also reduce_dependencies,
4500 : * which applies the same logic one-at-a-time.)
4501 : */
4502 : static void
4503 24 : move_to_ready_heap(TocEntry *pending_list,
4504 : binaryheap *ready_heap,
4505 : RestorePass pass)
4506 : {
4507 : TocEntry *te;
4508 : TocEntry *next_te;
4509 :
4510 116 : for (te = pending_list->pending_next; te != pending_list; te = next_te)
4511 : {
4512 : /* must save list link before possibly removing te from list */
4513 92 : next_te = te->pending_next;
4514 :
4515 132 : if (te->depCount == 0 &&
4516 40 : _tocEntryRestorePass(te) == pass)
4517 : {
4518 : /* Remove it from pending_list ... */
4519 40 : pending_list_remove(te);
4520 : /* ... and add to ready_heap */
4521 40 : binaryheap_add(ready_heap, te);
4522 : }
4523 : }
4524 24 : }
4525 :
4526 : /*
4527 : * Find the next work item (if any) that is capable of being run now,
4528 : * and remove it from the ready_heap.
4529 : *
4530 : * Returns the item, or NULL if nothing is runnable.
4531 : *
4532 : * To qualify, the item must have no remaining dependencies
4533 : * and no requirements for locks that are incompatible with
4534 : * items currently running. Items in the ready_heap are known to have
4535 : * no remaining dependencies, but we have to check for lock conflicts.
4536 : */
4537 : static TocEntry *
4538 140 : pop_next_work_item(binaryheap *ready_heap,
4539 : ParallelState *pstate)
4540 : {
4541 : /*
4542 : * Search the ready_heap until we find a suitable item. Note that we do a
4543 : * sequential scan through the heap nodes, so even though we will first
4544 : * try to choose the highest-priority item, we might end up picking
4545 : * something with a much lower priority. However, we expect that we will
4546 : * typically be able to pick one of the first few items, which should
4547 : * usually have a relatively high priority.
4548 : */
4549 142 : for (int i = 0; i < binaryheap_size(ready_heap); i++)
4550 : {
4551 94 : TocEntry *te = (TocEntry *) binaryheap_get_node(ready_heap, i);
4552 94 : bool conflicts = false;
4553 :
4554 : /*
4555 : * Check to see if the item would need exclusive lock on something
4556 : * that a currently running item also needs lock on, or vice versa. If
4557 : * so, we don't want to schedule them together.
4558 : */
4559 366 : for (int k = 0; k < pstate->numWorkers; k++)
4560 : {
4561 274 : TocEntry *running_te = pstate->te[k];
4562 :
4563 274 : if (running_te == NULL)
4564 128 : continue;
4565 290 : if (has_lock_conflicts(te, running_te) ||
4566 144 : has_lock_conflicts(running_te, te))
4567 : {
4568 2 : conflicts = true;
4569 2 : break;
4570 : }
4571 : }
4572 :
4573 94 : if (conflicts)
4574 2 : continue;
4575 :
4576 : /* passed all tests, so this item can run */
4577 92 : binaryheap_remove_node(ready_heap, i);
4578 92 : return te;
4579 : }
4580 :
4581 48 : pg_log_debug("no item ready");
4582 48 : return NULL;
4583 : }
4584 :
4585 :
4586 : /*
4587 : * Restore a single TOC item in parallel with others
4588 : *
4589 : * this is run in the worker, i.e. in a thread (Windows) or a separate process
4590 : * (everything else). A worker process executes several such work items during
4591 : * a parallel backup or restore. Once we terminate here and report back that
4592 : * our work is finished, the leader process will assign us a new work item.
4593 : */
4594 : int
4595 92 : parallel_restore(ArchiveHandle *AH, TocEntry *te)
4596 : {
4597 : int status;
4598 :
4599 : Assert(AH->connection != NULL);
4600 :
4601 : /* Count only errors associated with this TOC entry */
4602 92 : AH->public.n_errors = 0;
4603 :
4604 : /* Restore the TOC item */
4605 92 : status = restore_toc_entry(AH, te, true);
4606 :
4607 92 : return status;
4608 : }
4609 :
4610 :
4611 : /*
4612 : * Callback function that's invoked in the leader process after a step has
4613 : * been parallel restored.
4614 : *
4615 : * Update status and reduce the dependency count of any dependent items.
4616 : */
4617 : static void
4618 92 : mark_restore_job_done(ArchiveHandle *AH,
4619 : TocEntry *te,
4620 : int status,
4621 : void *callback_data)
4622 : {
4623 92 : binaryheap *ready_heap = (binaryheap *) callback_data;
4624 :
4625 92 : pg_log_info("finished item %d %s %s",
4626 : te->dumpId, te->desc, te->tag);
4627 :
4628 92 : if (status == WORKER_CREATE_DONE)
4629 0 : mark_create_done(AH, te);
4630 92 : else if (status == WORKER_INHIBIT_DATA)
4631 : {
4632 0 : inhibit_data_for_failed_table(AH, te);
4633 0 : AH->public.n_errors++;
4634 : }
4635 92 : else if (status == WORKER_IGNORED_ERRORS)
4636 0 : AH->public.n_errors++;
4637 92 : else if (status != 0)
4638 0 : pg_fatal("worker process failed: exit code %d",
4639 : status);
4640 :
4641 92 : reduce_dependencies(AH, te, ready_heap);
4642 92 : }
4643 :
4644 :
4645 : /*
4646 : * Process the dependency information into a form useful for parallel restore.
4647 : *
4648 : * This function takes care of fixing up some missing or badly designed
4649 : * dependencies, and then prepares subsidiary data structures that will be
4650 : * used in the main parallel-restore logic, including:
4651 : * 1. We build the revDeps[] arrays of incoming dependency dumpIds.
4652 : * 2. We set up depCount fields that are the number of as-yet-unprocessed
4653 : * dependencies for each TOC entry.
4654 : *
4655 : * We also identify locking dependencies so that we can avoid trying to
4656 : * schedule conflicting items at the same time.
4657 : */
4658 : static void
4659 8 : fix_dependencies(ArchiveHandle *AH)
4660 : {
4661 : TocEntry *te;
4662 : int i;
4663 :
4664 : /*
4665 : * Initialize the depCount/revDeps/nRevDeps fields, and make sure the TOC
4666 : * items are marked as not being in any parallel-processing list.
4667 : */
4668 200 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4669 : {
4670 192 : te->depCount = te->nDeps;
4671 192 : te->revDeps = NULL;
4672 192 : te->nRevDeps = 0;
4673 192 : te->pending_prev = NULL;
4674 192 : te->pending_next = NULL;
4675 : }
4676 :
4677 : /*
4678 : * POST_DATA items that are shown as depending on a table need to be
4679 : * re-pointed to depend on that table's data, instead. This ensures they
4680 : * won't get scheduled until the data has been loaded.
4681 : */
4682 8 : repoint_table_dependencies(AH);
4683 :
4684 : /*
4685 : * Pre-8.4 versions of pg_dump neglected to set up a dependency from BLOB
4686 : * COMMENTS to BLOBS. Cope. (We assume there's only one BLOBS and only
4687 : * one BLOB COMMENTS in such files.)
4688 : */
4689 8 : if (AH->version < K_VERS_1_11)
4690 : {
4691 0 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4692 : {
4693 0 : if (strcmp(te->desc, "BLOB COMMENTS") == 0 && te->nDeps == 0)
4694 : {
4695 : TocEntry *te2;
4696 :
4697 0 : for (te2 = AH->toc->next; te2 != AH->toc; te2 = te2->next)
4698 : {
4699 0 : if (strcmp(te2->desc, "BLOBS") == 0)
4700 : {
4701 0 : te->dependencies = (DumpId *) pg_malloc(sizeof(DumpId));
4702 0 : te->dependencies[0] = te2->dumpId;
4703 0 : te->nDeps++;
4704 0 : te->depCount++;
4705 0 : break;
4706 : }
4707 : }
4708 0 : break;
4709 : }
4710 : }
4711 : }
4712 :
4713 : /*
4714 : * At this point we start to build the revDeps reverse-dependency arrays,
4715 : * so all changes of dependencies must be complete.
4716 : */
4717 :
4718 : /*
4719 : * Count the incoming dependencies for each item. Also, it is possible
4720 : * that the dependencies list items that are not in the archive at all
4721 : * (that should not happen in 9.2 and later, but is highly likely in older
4722 : * archives). Subtract such items from the depCounts.
4723 : */
4724 200 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4725 : {
4726 576 : for (i = 0; i < te->nDeps; i++)
4727 : {
4728 384 : DumpId depid = te->dependencies[i];
4729 :
4730 384 : if (depid <= AH->maxDumpId && AH->tocsByDumpId[depid] != NULL)
4731 384 : AH->tocsByDumpId[depid]->nRevDeps++;
4732 : else
4733 0 : te->depCount--;
4734 : }
4735 : }
4736 :
4737 : /*
4738 : * Allocate space for revDeps[] arrays, and reset nRevDeps so we can use
4739 : * it as a counter below.
4740 : */
4741 200 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4742 : {
4743 192 : if (te->nRevDeps > 0)
4744 104 : te->revDeps = (DumpId *) pg_malloc(te->nRevDeps * sizeof(DumpId));
4745 192 : te->nRevDeps = 0;
4746 : }
4747 :
4748 : /*
4749 : * Build the revDeps[] arrays of incoming-dependency dumpIds. This had
4750 : * better agree with the loops above.
4751 : */
4752 200 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4753 : {
4754 576 : for (i = 0; i < te->nDeps; i++)
4755 : {
4756 384 : DumpId depid = te->dependencies[i];
4757 :
4758 384 : if (depid <= AH->maxDumpId && AH->tocsByDumpId[depid] != NULL)
4759 : {
4760 384 : TocEntry *otherte = AH->tocsByDumpId[depid];
4761 :
4762 384 : otherte->revDeps[otherte->nRevDeps++] = te->dumpId;
4763 : }
4764 : }
4765 : }
4766 :
4767 : /*
4768 : * Lastly, work out the locking dependencies.
4769 : */
4770 200 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4771 : {
4772 192 : te->lockDeps = NULL;
4773 192 : te->nLockDeps = 0;
4774 192 : identify_locking_dependencies(AH, te);
4775 : }
4776 8 : }
4777 :
4778 : /*
4779 : * Change dependencies on table items to depend on table data items instead,
4780 : * but only in POST_DATA items.
4781 : *
4782 : * Also, for any item having such dependency(s), set its dataLength to the
4783 : * largest dataLength of the table data items it depends on. This ensures
4784 : * that parallel restore will prioritize larger jobs (index builds, FK
4785 : * constraint checks, etc) over smaller ones, avoiding situations where we
4786 : * end a restore with only one active job working on a large table.
4787 : */
4788 : static void
4789 8 : repoint_table_dependencies(ArchiveHandle *AH)
4790 : {
4791 : TocEntry *te;
4792 : int i;
4793 : DumpId olddep;
4794 :
4795 200 : for (te = AH->toc->next; te != AH->toc; te = te->next)
4796 : {
4797 192 : if (te->section != SECTION_POST_DATA)
4798 132 : continue;
4799 320 : for (i = 0; i < te->nDeps; i++)
4800 : {
4801 260 : olddep = te->dependencies[i];
4802 260 : if (olddep <= AH->maxDumpId &&
4803 260 : AH->tableDataId[olddep] != 0)
4804 : {
4805 124 : DumpId tabledataid = AH->tableDataId[olddep];
4806 124 : TocEntry *tabledatate = AH->tocsByDumpId[tabledataid];
4807 :
4808 124 : te->dependencies[i] = tabledataid;
4809 124 : te->dataLength = Max(te->dataLength, tabledatate->dataLength);
4810 124 : pg_log_debug("transferring dependency %d -> %d to %d",
4811 : te->dumpId, olddep, tabledataid);
4812 : }
4813 : }
4814 : }
4815 8 : }
4816 :
4817 : /*
4818 : * Identify which objects we'll need exclusive lock on in order to restore
4819 : * the given TOC entry (*other* than the one identified by the TOC entry
4820 : * itself). Record their dump IDs in the entry's lockDeps[] array.
4821 : */
4822 : static void
4823 192 : identify_locking_dependencies(ArchiveHandle *AH, TocEntry *te)
4824 : {
4825 : DumpId *lockids;
4826 : int nlockids;
4827 : int i;
4828 :
4829 : /*
4830 : * We only care about this for POST_DATA items. PRE_DATA items are not
4831 : * run in parallel, and DATA items are all independent by assumption.
4832 : */
4833 192 : if (te->section != SECTION_POST_DATA)
4834 132 : return;
4835 :
4836 : /* Quick exit if no dependencies at all */
4837 60 : if (te->nDeps == 0)
4838 0 : return;
4839 :
4840 : /*
4841 : * Most POST_DATA items are ALTER TABLEs or some moral equivalent of that,
4842 : * and hence require exclusive lock. However, we know that CREATE INDEX
4843 : * does not. (Maybe someday index-creating CONSTRAINTs will fall in that
4844 : * category too ... but today is not that day.)
4845 : */
4846 60 : if (strcmp(te->desc, "INDEX") == 0)
4847 0 : return;
4848 :
4849 : /*
4850 : * We assume the entry requires exclusive lock on each TABLE or TABLE DATA
4851 : * item listed among its dependencies. Originally all of these would have
4852 : * been TABLE items, but repoint_table_dependencies would have repointed
4853 : * them to the TABLE DATA items if those are present (which they might not
4854 : * be, eg in a schema-only dump). Note that all of the entries we are
4855 : * processing here are POST_DATA; otherwise there might be a significant
4856 : * difference between a dependency on a table and a dependency on its
4857 : * data, so that closer analysis would be needed here.
4858 : */
4859 60 : lockids = (DumpId *) pg_malloc(te->nDeps * sizeof(DumpId));
4860 60 : nlockids = 0;
4861 320 : for (i = 0; i < te->nDeps; i++)
4862 : {
4863 260 : DumpId depid = te->dependencies[i];
4864 :
4865 260 : if (depid <= AH->maxDumpId && AH->tocsByDumpId[depid] != NULL &&
4866 260 : ((strcmp(AH->tocsByDumpId[depid]->desc, "TABLE DATA") == 0) ||
4867 136 : strcmp(AH->tocsByDumpId[depid]->desc, "TABLE") == 0))
4868 164 : lockids[nlockids++] = depid;
4869 : }
4870 :
4871 60 : if (nlockids == 0)
4872 : {
4873 0 : free(lockids);
4874 0 : return;
4875 : }
4876 :
4877 60 : te->lockDeps = pg_realloc(lockids, nlockids * sizeof(DumpId));
4878 60 : te->nLockDeps = nlockids;
4879 : }
4880 :
4881 : /*
4882 : * Remove the specified TOC entry from the depCounts of items that depend on
4883 : * it, thereby possibly making them ready-to-run. Any pending item that
4884 : * becomes ready should be moved to the ready_heap, if that's provided.
4885 : */
4886 : static void
4887 192 : reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
4888 : binaryheap *ready_heap)
4889 : {
4890 : int i;
4891 :
4892 192 : pg_log_debug("reducing dependencies for %d", te->dumpId);
4893 :
4894 576 : for (i = 0; i < te->nRevDeps; i++)
4895 : {
4896 384 : TocEntry *otherte = AH->tocsByDumpId[te->revDeps[i]];
4897 :
4898 : Assert(otherte->depCount > 0);
4899 384 : otherte->depCount--;
4900 :
4901 : /*
4902 : * It's ready if it has no remaining dependencies, and it belongs in
4903 : * the current restore pass, and it is currently a member of the
4904 : * pending list (that check is needed to prevent double restore in
4905 : * some cases where a list-file forces out-of-order restoring).
4906 : * However, if ready_heap == NULL then caller doesn't want any list
4907 : * memberships changed.
4908 : */
4909 384 : if (otherte->depCount == 0 &&
4910 148 : _tocEntryRestorePass(otherte) == AH->restorePass &&
4911 148 : otherte->pending_prev != NULL &&
4912 : ready_heap != NULL)
4913 : {
4914 : /* Remove it from pending list ... */
4915 52 : pending_list_remove(otherte);
4916 : /* ... and add to ready_heap */
4917 52 : binaryheap_add(ready_heap, otherte);
4918 : }
4919 : }
4920 192 : }
4921 :
4922 : /*
4923 : * Set the created flag on the DATA member corresponding to the given
4924 : * TABLE member
4925 : */
4926 : static void
4927 9024 : mark_create_done(ArchiveHandle *AH, TocEntry *te)
4928 : {
4929 9024 : if (AH->tableDataId[te->dumpId] != 0)
4930 : {
4931 6722 : TocEntry *ted = AH->tocsByDumpId[AH->tableDataId[te->dumpId]];
4932 :
4933 6722 : ted->created = true;
4934 : }
4935 9024 : }
4936 :
4937 : /*
4938 : * Mark the DATA member corresponding to the given TABLE member
4939 : * as not wanted
4940 : */
4941 : static void
4942 0 : inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te)
4943 : {
4944 0 : pg_log_info("table \"%s\" could not be created, will not restore its data",
4945 : te->tag);
4946 :
4947 0 : if (AH->tableDataId[te->dumpId] != 0)
4948 : {
4949 0 : TocEntry *ted = AH->tocsByDumpId[AH->tableDataId[te->dumpId]];
4950 :
4951 0 : ted->reqs = 0;
4952 : }
4953 0 : }
4954 :
4955 : /*
4956 : * Clone and de-clone routines used in parallel restoration.
4957 : *
4958 : * Enough of the structure is cloned to ensure that there is no
4959 : * conflict between different threads each with their own clone.
4960 : */
4961 : ArchiveHandle *
4962 52 : CloneArchive(ArchiveHandle *AH)
4963 : {
4964 : ArchiveHandle *clone;
4965 :
4966 : /* Make a "flat" copy */
4967 52 : clone = (ArchiveHandle *) pg_malloc(sizeof(ArchiveHandle));
4968 52 : memcpy(clone, AH, sizeof(ArchiveHandle));
4969 :
4970 : /* Likewise flat-copy the RestoreOptions, so we can alter them locally */
4971 52 : clone->public.ropt = (RestoreOptions *) pg_malloc(sizeof(RestoreOptions));
4972 52 : memcpy(clone->public.ropt, AH->public.ropt, sizeof(RestoreOptions));
4973 :
4974 : /* Handle format-independent fields */
4975 52 : memset(&(clone->sqlparse), 0, sizeof(clone->sqlparse));
4976 :
4977 : /* The clone will have its own connection, so disregard connection state */
4978 52 : clone->connection = NULL;
4979 52 : clone->connCancel = NULL;
4980 52 : clone->currUser = NULL;
4981 52 : clone->currSchema = NULL;
4982 52 : clone->currTableAm = NULL;
4983 52 : clone->currTablespace = NULL;
4984 :
4985 : /* savedPassword must be local in case we change it while connecting */
4986 52 : if (clone->savedPassword)
4987 0 : clone->savedPassword = pg_strdup(clone->savedPassword);
4988 :
4989 : /* clone has its own error count, too */
4990 52 : clone->public.n_errors = 0;
4991 :
4992 : /* clones should not share lo_buf */
4993 52 : clone->lo_buf = NULL;
4994 :
4995 : /*
4996 : * Clone connections disregard --transaction-size; they must commit after
4997 : * each command so that the results are immediately visible to other
4998 : * workers.
4999 : */
5000 52 : clone->public.ropt->txn_size = 0;
5001 :
5002 : /*
5003 : * Connect our new clone object to the database, using the same connection
5004 : * parameters used for the original connection.
5005 : */
5006 52 : ConnectDatabase((Archive *) clone, &clone->public.ropt->cparams, true);
5007 :
5008 : /* re-establish fixed state */
5009 52 : if (AH->mode == archModeRead)
5010 20 : _doSetFixedOutputState(clone);
5011 : /* in write case, setupDumpWorker will fix up connection state */
5012 :
5013 : /* Let the format-specific code have a chance too */
5014 52 : clone->ClonePtr(clone);
5015 :
5016 : Assert(clone->connection != NULL);
5017 52 : return clone;
5018 : }
5019 :
5020 : /*
5021 : * Release clone-local storage.
5022 : *
5023 : * Note: we assume any clone-local connection was already closed.
5024 : */
5025 : void
5026 52 : DeCloneArchive(ArchiveHandle *AH)
5027 : {
5028 : /* Should not have an open database connection */
5029 : Assert(AH->connection == NULL);
5030 :
5031 : /* Clear format-specific state */
5032 52 : AH->DeClonePtr(AH);
5033 :
5034 : /* Clear state allocated by CloneArchive */
5035 52 : if (AH->sqlparse.curCmd)
5036 6 : destroyPQExpBuffer(AH->sqlparse.curCmd);
5037 :
5038 : /* Clear any connection-local state */
5039 52 : free(AH->currUser);
5040 52 : free(AH->currSchema);
5041 52 : free(AH->currTablespace);
5042 52 : free(AH->currTableAm);
5043 52 : free(AH->savedPassword);
5044 :
5045 52 : free(AH);
5046 52 : }
|
