granicus.if.org Git - postgresql/blob - src/bin/pg_dump/pg_backup_archiver.c

   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
  23 #include "pg_backup_db.h"
  24 #include "dumputils.h"
  25
  26 #include <ctype.h>
  27 #include <unistd.h>
  28 #include <sys/types.h>
  29 #include <sys/wait.h>
  30
  31 #ifdef WIN32
  32 #include <io.h>
  33 #endif
  34
  35 #include "libpq/libpq-fs.h"
  36
  37 /*
  38  * Special exit values from worker children.  We reserve 0 for normal
  39  * success; 1 and other small values should be interpreted as crashes.
  40  */
  41 #define WORKER_CREATE_DONE              10
  42 #define WORKER_INHIBIT_DATA             11
  43 #define WORKER_IGNORED_ERRORS   12
  44
  45 /*
  46  * Unix uses exit to return result from worker child, so function is void.
  47  * Windows thread result comes via function return.
  48  */
  49 #ifndef WIN32
  50 #define parallel_restore_result void
  51 #else
  52 #define parallel_restore_result DWORD
  53 #endif
  54
  55 /* IDs for worker children are either PIDs or thread handles */
  56 #ifndef WIN32
  57 #define thandle pid_t
  58 #else
  59 #define thandle HANDLE
  60 #endif
  61
  62 /* Arguments needed for a worker child */
  63 typedef struct _restore_args
  64 {
  65         ArchiveHandle *AH;
  66         TocEntry   *te;
  67 } RestoreArgs;
  68
  69 /* State for each parallel activity slot */
  70 typedef struct _parallel_slot
  71 {
  72         thandle         child_id;
  73         RestoreArgs *args;
  74 } ParallelSlot;
  75
  76 #define NO_SLOT (-1)
  77
  78 const char *progname;
  79
  80 static const char *modulename = gettext_noop("archiver");
  81
  82
  83 static ArchiveHandle *_allocAH(const char *FileSpec, const ArchiveFormat fmt,
  84                  const int compression, ArchiveMode mode);
  85 static void _getObjectDescription(PQExpBuffer buf, TocEntry *te,
  86                                           ArchiveHandle *AH);
  87 static void _printTocEntry(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt, bool isData, bool acl_pass);
  88
  89
  90 static void _doSetFixedOutputState(ArchiveHandle *AH);
  91 static void _doSetSessionAuth(ArchiveHandle *AH, const char *user);
  92 static void _doSetWithOids(ArchiveHandle *AH, const bool withOids);
  93 static void _reconnectToDB(ArchiveHandle *AH, const char *dbname);
  94 static void _becomeUser(ArchiveHandle *AH, const char *user);
  95 static void _becomeOwner(ArchiveHandle *AH, TocEntry *te);
  96 static void _selectOutputSchema(ArchiveHandle *AH, const char *schemaName);
  97 static void _selectTablespace(ArchiveHandle *AH, const char *tablespace);
  98 static void processEncodingEntry(ArchiveHandle *AH, TocEntry *te);
  99 static void processStdStringsEntry(ArchiveHandle *AH, TocEntry *te);
 100 static teReqs _tocEntryRequired(TocEntry *te, RestoreOptions *ropt, bool include_acls);
 101 static bool _tocEntryIsACL(TocEntry *te);
 102 static void _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
 103 static void _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
 104 static TocEntry *getTocEntryByDumpId(ArchiveHandle *AH, DumpId id);
 105 static void _moveBefore(ArchiveHandle *AH, TocEntry *pos, TocEntry *te);
 106 static int      _discoverArchiveFormat(ArchiveHandle *AH);
 107
 108 static void dump_lo_buf(ArchiveHandle *AH);
 109 static void _write_msg(const char *modulename, const char *fmt, va_list ap);
 110 static void _die_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt, va_list ap);
 111
 112 static void dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim);
 113 static OutputContext SetOutput(ArchiveHandle *AH, char *filename, int compression);
 114 static void ResetOutput(ArchiveHandle *AH, OutputContext savedContext);
 115
 116 static int restore_toc_entry(ArchiveHandle *AH, TocEntry *te,
 117                                   RestoreOptions *ropt, bool is_parallel);
 118 static void restore_toc_entries_parallel(ArchiveHandle *AH);
 119 static thandle spawn_restore(RestoreArgs *args);
 120 static thandle reap_child(ParallelSlot *slots, int n_slots, int *work_status);
 121 static bool work_in_progress(ParallelSlot *slots, int n_slots);
 122 static int      get_next_slot(ParallelSlot *slots, int n_slots);
 123 static void par_list_header_init(TocEntry *l);
 124 static void par_list_append(TocEntry *l, TocEntry *te);
 125 static void par_list_remove(TocEntry *te);
 126 static TocEntry *get_next_work_item(ArchiveHandle *AH,
 127                                    TocEntry *ready_list,
 128                                    ParallelSlot *slots, int n_slots);
 129 static parallel_restore_result parallel_restore(RestoreArgs *args);
 130 static void mark_work_done(ArchiveHandle *AH, TocEntry *ready_list,
 131                            thandle worker, int status,
 132                            ParallelSlot *slots, int n_slots);
 133 static void fix_dependencies(ArchiveHandle *AH);
 134 static bool has_lock_conflicts(TocEntry *te1, TocEntry *te2);
 135 static void repoint_table_dependencies(ArchiveHandle *AH,
 136                                                    DumpId tableId, DumpId tableDataId);
 137 static void identify_locking_dependencies(TocEntry *te,
 138                                                           TocEntry **tocsByDumpId,
 139                                                           DumpId maxDumpId);
 140 static void reduce_dependencies(ArchiveHandle *AH, TocEntry *te,
 141                                         TocEntry *ready_list);
 142 static void mark_create_done(ArchiveHandle *AH, TocEntry *te);
 143 static void inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te);
 144 static ArchiveHandle *CloneArchive(ArchiveHandle *AH);
 145 static void DeCloneArchive(ArchiveHandle *AH);
 146
 147
 148 /*
 149  *      Wrapper functions.
 150  *
 151  *      The objective it to make writing new formats and dumpers as simple
 152  *      as possible, if necessary at the expense of extra function calls etc.
 153  *
 154  */
 155
 156
 157 /* Create a new archive */
 158 /* Public */
 159 Archive *
 160 CreateArchive(const char *FileSpec, const ArchiveFormat fmt,
 161                           const int compression, ArchiveMode mode)
 162
 163 {
 164         ArchiveHandle *AH = _allocAH(FileSpec, fmt, compression, mode);
 165
 166         return (Archive *) AH;
 167 }
 168
 169 /* Open an existing archive */
 170 /* Public */
 171 Archive *
 172 OpenArchive(const char *FileSpec, const ArchiveFormat fmt)
 173 {
 174         ArchiveHandle *AH = _allocAH(FileSpec, fmt, 0, archModeRead);
 175
 176         return (Archive *) AH;
 177 }
 178
 179 /* Public */
 180 void
 181 CloseArchive(Archive *AHX)
 182 {
 183         int                     res = 0;
 184         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 185
 186         (*AH->ClosePtr) (AH);
 187
 188         /* Close the output */
 189         if (AH->gzOut)
 190                 res = GZCLOSE(AH->OF);
 191         else if (AH->OF != stdout)
 192                 res = fclose(AH->OF);
 193
 194         if (res != 0)
 195                 die_horribly(AH, modulename, "could not close output file: %s\n",
 196                                          strerror(errno));
 197 }
 198
 199 /* Public */
 200 void
 201 RestoreArchive(Archive *AHX, RestoreOptions *ropt)
 202 {
 203         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 204         TocEntry   *te;
 205         teReqs          reqs;
 206         OutputContext sav;
 207
 208         AH->ropt = ropt;
 209         AH->stage = STAGE_INITIALIZING;
 210
 211         /*
 212          * Check for nonsensical option combinations.
 213          *
 214          * NB: createDB+dropSchema is useless because if you're creating the DB,
 215          * there's no need to drop individual items in it.  Moreover, if we tried
 216          * to do that then we'd issue the drops in the database initially
 217          * connected to, not the one we will create, which is very bad...
 218          */
 219         if (ropt->createDB && ropt->dropSchema)
 220                 die_horribly(AH, modulename, "-C and -c are incompatible options\n");
 221
 222         /*
 223          * -C is not compatible with -1, because we can't create a database inside
 224          * a transaction block.
 225          */
 226         if (ropt->createDB && ropt->single_txn)
 227                 die_horribly(AH, modulename, "-C and -1 are incompatible options\n");
 228
 229         /*
 230          * Make sure we won't need (de)compression we haven't got
 231          */
 232 #ifndef HAVE_LIBZ
 233         if (AH->compression != 0 && AH->PrintTocDataPtr !=NULL)
 234         {
 235                 for (te = AH->toc->next; te != AH->toc; te = te->next)
 236                 {
 237                         reqs = _tocEntryRequired(te, ropt, false);
 238                         if (te->hadDumper && (reqs & REQ_DATA) != 0)
 239                                 die_horribly(AH, modulename, "cannot restore from compressed archive (compression not supported in this installation)\n");
 240                 }
 241         }
 242 #endif
 243
 244         /*
 245          * If we're using a DB connection, then connect it.
 246          */
 247         if (ropt->useDB)
 248         {
 249                 ahlog(AH, 1, "connecting to database for restore\n");
 250                 if (AH->version < K_VERS_1_3)
 251                         die_horribly(AH, modulename, "direct database connections are not supported in pre-1.3 archives\n");
 252
 253                 /* XXX Should get this from the archive */
 254                 AHX->minRemoteVersion = 070100;
 255                 AHX->maxRemoteVersion = 999999;
 256
 257                 ConnectDatabase(AHX, ropt->dbname,
 258                                                 ropt->pghost, ropt->pgport, ropt->username,
 259                                                 ropt->promptPassword);
 260
 261                 /*
 262                  * If we're talking to the DB directly, don't send comments since they
 263                  * obscure SQL when displaying errors
 264                  */
 265                 AH->noTocComments = 1;
 266         }
 267
 268         /*
 269          * Work out if we have an implied data-only restore. This can happen if
 270          * the dump was data only or if the user has used a toc list to exclude
 271          * all of the schema data. All we do is look for schema entries - if none
 272          * are found then we set the dataOnly flag.
 273          *
 274          * We could scan for wanted TABLE entries, but that is not the same as
 275          * dataOnly. At this stage, it seems unnecessary (6-Mar-2001).
 276          */
 277         if (!ropt->dataOnly)
 278         {
 279                 int                     impliedDataOnly = 1;
 280
 281                 for (te = AH->toc->next; te != AH->toc; te = te->next)
 282                 {
 283                         reqs = _tocEntryRequired(te, ropt, true);
 284                         if ((reqs & REQ_SCHEMA) != 0)
 285                         {                                       /* It's schema, and it's wanted */
 286                                 impliedDataOnly = 0;
 287                                 break;
 288                         }
 289                 }
 290                 if (impliedDataOnly)
 291                 {
 292                         ropt->dataOnly = impliedDataOnly;
 293                         ahlog(AH, 1, "implied data-only restore\n");
 294                 }
 295         }
 296
 297         /*
 298          * Setup the output file if necessary.
 299          */
 300         if (ropt->filename || ropt->compression)
 301                 sav = SetOutput(AH, ropt->filename, ropt->compression);
 302
 303         ahprintf(AH, "--\n-- PostgreSQL database dump\n--\n\n");
 304
 305         if (AH->public.verbose)
 306         {
 307                 if (AH->archiveRemoteVersion)
 308                         ahprintf(AH, "-- Dumped from database version %s\n",
 309                                          AH->archiveRemoteVersion);
 310                 if (AH->archiveDumpVersion)
 311                         ahprintf(AH, "-- Dumped by pg_dump version %s\n",
 312                                          AH->archiveDumpVersion);
 313                 dumpTimestamp(AH, "Started on", AH->createDate);
 314         }
 315
 316         if (ropt->single_txn)
 317         {
 318                 if (AH->connection)
 319                         StartTransaction(AH);
 320                 else
 321                         ahprintf(AH, "BEGIN;\n\n");
 322         }
 323
 324         /*
 325          * Establish important parameter values right away.
 326          */
 327         _doSetFixedOutputState(AH);
 328
 329         AH->stage = STAGE_PROCESSING;
 330
 331         /*
 332          * Drop the items at the start, in reverse order
 333          */
 334         if (ropt->dropSchema)
 335         {
 336                 for (te = AH->toc->prev; te != AH->toc; te = te->prev)
 337                 {
 338                         AH->currentTE = te;
 339
 340                         reqs = _tocEntryRequired(te, ropt, false /* needn't drop ACLs */ );
 341                         /* We want anything that's selected and has a dropStmt */
 342                         if (((reqs & (REQ_SCHEMA | REQ_DATA)) != 0) && te->dropStmt)
 343                         {
 344                                 ahlog(AH, 1, "dropping %s %s\n", te->desc, te->tag);
 345                                 /* Select owner and schema as necessary */
 346                                 _becomeOwner(AH, te);
 347                                 _selectOutputSchema(AH, te->namespace);
 348                                 /* Drop it */
 349                                 ahprintf(AH, "%s", te->dropStmt);
 350                         }
 351                 }
 352
 353                 /*
 354                  * _selectOutputSchema may have set currSchema to reflect the effect
 355                  * of a "SET search_path" command it emitted.  However, by now we may
 356                  * have dropped that schema; or it might not have existed in the first
 357                  * place.  In either case the effective value of search_path will not
 358                  * be what we think.  Forcibly reset currSchema so that we will
 359                  * re-establish the search_path setting when needed (after creating
 360                  * the schema).
 361                  *
 362                  * If we treated users as pg_dump'able objects then we'd need to reset
 363                  * currUser here too.
 364                  */
 365                 if (AH->currSchema)
 366                         free(AH->currSchema);
 367                 AH->currSchema = NULL;
 368         }
 369
 370         /*
 371          * In serial mode, we now process each non-ACL TOC entry.
 372          *
 373          * In parallel mode, turn control over to the parallel-restore logic.
 374          */
 375         if (ropt->number_of_jobs > 1 && ropt->useDB)
 376                 restore_toc_entries_parallel(AH);
 377         else
 378         {
 379                 for (te = AH->toc->next; te != AH->toc; te = te->next)
 380                         (void) restore_toc_entry(AH, te, ropt, false);
 381         }
 382
 383         /*
 384          * Scan TOC again to output ownership commands and ACLs
 385          */
 386         for (te = AH->toc->next; te != AH->toc; te = te->next)
 387         {
 388                 AH->currentTE = te;
 389
 390                 /* Work out what, if anything, we want from this entry */
 391                 reqs = _tocEntryRequired(te, ropt, true);
 392
 393                 /* Both schema and data objects might now have ownership/ACLs */
 394                 if ((reqs & (REQ_SCHEMA | REQ_DATA)) != 0)
 395                 {
 396                         ahlog(AH, 1, "setting owner and privileges for %s %s\n",
 397                                   te->desc, te->tag);
 398                         _printTocEntry(AH, te, ropt, false, true);
 399                 }
 400         }
 401
 402         if (ropt->single_txn)
 403         {
 404                 if (AH->connection)
 405                         CommitTransaction(AH);
 406                 else
 407                         ahprintf(AH, "COMMIT;\n\n");
 408         }
 409
 410         if (AH->public.verbose)
 411                 dumpTimestamp(AH, "Completed on", time(NULL));
 412
 413         ahprintf(AH, "--\n-- PostgreSQL database dump complete\n--\n\n");
 414
 415         /*
 416          * Clean up & we're done.
 417          */
 418         AH->stage = STAGE_FINALIZING;
 419
 420         if (ropt->filename || ropt->compression)
 421                 ResetOutput(AH, sav);
 422
 423         if (ropt->useDB)
 424         {
 425                 PQfinish(AH->connection);
 426                 AH->connection = NULL;
 427         }
 428 }
 429
 430 /*
 431  * Restore a single TOC item.  Used in both parallel and non-parallel restore;
 432  * is_parallel is true if we are in a worker child process.
 433  *
 434  * Returns 0 normally, but WORKER_CREATE_DONE or WORKER_INHIBIT_DATA if
 435  * the parallel parent has to make the corresponding status update.
 436  */
 437 static int
 438 restore_toc_entry(ArchiveHandle *AH, TocEntry *te,
 439                                   RestoreOptions *ropt, bool is_parallel)
 440 {
 441         int                     retval = 0;
 442         teReqs          reqs;
 443         bool            defnDumped;
 444
 445         AH->currentTE = te;
 446
 447         /* Work out what, if anything, we want from this entry */
 448         reqs = _tocEntryRequired(te, ropt, false);
 449
 450         /* Dump any relevant dump warnings to stderr */
 451         if (!ropt->suppressDumpWarnings && strcmp(te->desc, "WARNING") == 0)
 452         {
 453                 if (!ropt->dataOnly && te->defn != NULL && strlen(te->defn) != 0)
 454                         write_msg(modulename, "warning from original dump file: %s\n", te->defn);
 455                 else if (te->copyStmt != NULL && strlen(te->copyStmt) != 0)
 456                         write_msg(modulename, "warning from original dump file: %s\n", te->copyStmt);
 457         }
 458
 459         defnDumped = false;
 460
 461         if ((reqs & REQ_SCHEMA) != 0)           /* We want the schema */
 462         {
 463                 ahlog(AH, 1, "creating %s %s\n", te->desc, te->tag);
 464
 465                 _printTocEntry(AH, te, ropt, false, false);
 466                 defnDumped = true;
 467
 468                 if (strcmp(te->desc, "TABLE") == 0)
 469                 {
 470                         if (AH->lastErrorTE == te)
 471                         {
 472                                 /*
 473                                  * We failed to create the table. If
 474                                  * --no-data-for-failed-tables was given, mark the
 475                                  * corresponding TABLE DATA to be ignored.
 476                                  *
 477                                  * In the parallel case this must be done in the parent, so we
 478                                  * just set the return value.
 479                                  */
 480                                 if (ropt->noDataForFailedTables)
 481                                 {
 482                                         if (is_parallel)
 483                                                 retval = WORKER_INHIBIT_DATA;
 484                                         else
 485                                                 inhibit_data_for_failed_table(AH, te);
 486                                 }
 487                         }
 488                         else
 489                         {
 490                                 /*
 491                                  * We created the table successfully.  Mark the corresponding
 492                                  * TABLE DATA for possible truncation.
 493                                  *
 494                                  * In the parallel case this must be done in the parent, so we
 495                                  * just set the return value.
 496                                  */
 497                                 if (is_parallel)
 498                                         retval = WORKER_CREATE_DONE;
 499                                 else
 500                                         mark_create_done(AH, te);
 501                         }
 502                 }
 503
 504                 /* If we created a DB, connect to it... */
 505                 if (strcmp(te->desc, "DATABASE") == 0)
 506                 {
 507                         ahlog(AH, 1, "connecting to new database \"%s\"\n", te->tag);
 508                         _reconnectToDB(AH, te->tag);
 509                         ropt->dbname = strdup(te->tag);
 510                 }
 511         }
 512
 513         /*
 514          * If we have a data component, then process it
 515          */
 516         if ((reqs & REQ_DATA) != 0)
 517         {
 518                 /*
 519                  * hadDumper will be set if there is genuine data component for this
 520                  * node. Otherwise, we need to check the defn field for statements
 521                  * that need to be executed in data-only restores.
 522                  */
 523                 if (te->hadDumper)
 524                 {
 525                         /*
 526                          * If we can output the data, then restore it.
 527                          */
 528                         if (AH->PrintTocDataPtr !=NULL && (reqs & REQ_DATA) != 0)
 529                         {
 530                                 _printTocEntry(AH, te, ropt, true, false);
 531
 532                                 if (strcmp(te->desc, "BLOBS") == 0 ||
 533                                         strcmp(te->desc, "BLOB COMMENTS") == 0)
 534                                 {
 535                                         ahlog(AH, 1, "restoring %s\n", te->desc);
 536
 537                                         _selectOutputSchema(AH, "pg_catalog");
 538
 539                                         (*AH->PrintTocDataPtr) (AH, te, ropt);
 540                                 }
 541                                 else
 542                                 {
 543                                         _disableTriggersIfNecessary(AH, te, ropt);
 544
 545                                         /* Select owner and schema as necessary */
 546                                         _becomeOwner(AH, te);
 547                                         _selectOutputSchema(AH, te->namespace);
 548
 549                                         ahlog(AH, 1, "restoring data for table \"%s\"\n",
 550                                                   te->tag);
 551
 552                                         /*
 553                                          * In parallel restore, if we created the table earlier in
 554                                          * the run then we wrap the COPY in a transaction and
 555                                          * precede it with a TRUNCATE.  If archiving is not on
 556                                          * this prevents WAL-logging the COPY.  This obtains a
 557                                          * speedup similar to that from using single_txn mode in
 558                                          * non-parallel restores.
 559                                          */
 560                                         if (is_parallel && te->created)
 561                                         {
 562                                                 /*
 563                                                  * Parallel restore is always talking directly to a
 564                                                  * server, so no need to see if we should issue BEGIN.
 565                                                  */
 566                                                 StartTransaction(AH);
 567
 568                                                 /*
 569                                                  * If the server version is >= 8.4, make sure we issue
 570                                                  * TRUNCATE with ONLY so that child tables are not
 571                                                  * wiped.
 572                                                  */
 573                                                 ahprintf(AH, "TRUNCATE TABLE %s%s;\n\n",
 574                                                                  (PQserverVersion(AH->connection) >= 80400 ?
 575                                                                   "ONLY " : ""),
 576                                                                  fmtId(te->tag));
 577                                         }
 578
 579                                         /*
 580                                          * If we have a copy statement, use it. As of V1.3, these
 581                                          * are separate to allow easy import from withing a
 582                                          * database connection. Pre 1.3 archives can not use DB
 583                                          * connections and are sent to output only.
 584                                          *
 585                                          * For V1.3+, the table data MUST have a copy statement so
 586                                          * that we can go into appropriate mode with libpq.
 587                                          */
 588                                         if (te->copyStmt && strlen(te->copyStmt) > 0)
 589                                         {
 590                                                 ahprintf(AH, "%s", te->copyStmt);
 591                                                 AH->writingCopyData = true;
 592                                         }
 593
 594                                         (*AH->PrintTocDataPtr) (AH, te, ropt);
 595
 596                                         AH->writingCopyData = false;
 597
 598                                         /* close out the transaction started above */
 599                                         if (is_parallel && te->created)
 600                                                 CommitTransaction(AH);
 601
 602                                         _enableTriggersIfNecessary(AH, te, ropt);
 603                                 }
 604                         }
 605                 }
 606                 else if (!defnDumped)
 607                 {
 608                         /* If we haven't already dumped the defn part, do so now */
 609                         ahlog(AH, 1, "executing %s %s\n", te->desc, te->tag);
 610                         _printTocEntry(AH, te, ropt, false, false);
 611                 }
 612         }
 613
 614         return retval;
 615 }
 616
 617 /*
 618  * Allocate a new RestoreOptions block.
 619  * This is mainly so we can initialize it, but also for future expansion,
 620  */
 621 RestoreOptions *
 622 NewRestoreOptions(void)
 623 {
 624         RestoreOptions *opts;
 625
 626         opts = (RestoreOptions *) calloc(1, sizeof(RestoreOptions));
 627
 628         /* set any fields that shouldn't default to zeroes */
 629         opts->format = archUnknown;
 630         opts->promptPassword = TRI_DEFAULT;
 631
 632         return opts;
 633 }
 634
 635 static void
 636 _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt)
 637 {
 638         /* This hack is only needed in a data-only restore */
 639         if (!ropt->dataOnly || !ropt->disable_triggers)
 640                 return;
 641
 642         ahlog(AH, 1, "disabling triggers for %s\n", te->tag);
 643
 644         /*
 645          * Become superuser if possible, since they are the only ones who can
 646          * disable constraint triggers.  If -S was not given, assume the initial
 647          * user identity is a superuser.  (XXX would it be better to become the
 648          * table owner?)
 649          */
 650         _becomeUser(AH, ropt->superuser);
 651
 652         /*
 653          * Disable them.
 654          */
 655         _selectOutputSchema(AH, te->namespace);
 656
 657         ahprintf(AH, "ALTER TABLE %s DISABLE TRIGGER ALL;\n\n",
 658                          fmtId(te->tag));
 659 }
 660
 661 static void
 662 _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt)
 663 {
 664         /* This hack is only needed in a data-only restore */
 665         if (!ropt->dataOnly || !ropt->disable_triggers)
 666                 return;
 667
 668         ahlog(AH, 1, "enabling triggers for %s\n", te->tag);
 669
 670         /*
 671          * Become superuser if possible, since they are the only ones who can
 672          * disable constraint triggers.  If -S was not given, assume the initial
 673          * user identity is a superuser.  (XXX would it be better to become the
 674          * table owner?)
 675          */
 676         _becomeUser(AH, ropt->superuser);
 677
 678         /*
 679          * Enable them.
 680          */
 681         _selectOutputSchema(AH, te->namespace);
 682
 683         ahprintf(AH, "ALTER TABLE %s ENABLE TRIGGER ALL;\n\n",
 684                          fmtId(te->tag));
 685 }
 686
 687 /*
 688  * This is a routine that is part of the dumper interface, hence the 'Archive*' parameter.
 689  */
 690
 691 /* Public */
 692 size_t
 693 WriteData(Archive *AHX, const void *data, size_t dLen)
 694 {
 695         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 696
 697         if (!AH->currToc)
 698                 die_horribly(AH, modulename, "internal error -- WriteData cannot be called outside the context of a DataDumper routine\n");
 699
 700         return (*AH->WriteDataPtr) (AH, data, dLen);
 701 }
 702
 703 /*
 704  * Create a new TOC entry. The TOC was designed as a TOC, but is now the
 705  * repository for all metadata. But the name has stuck.
 706  */
 707
 708 /* Public */
 709 void
 710 ArchiveEntry(Archive *AHX,
 711                          CatalogId catalogId, DumpId dumpId,
 712                          const char *tag,
 713                          const char *namespace,
 714                          const char *tablespace,
 715                          const char *owner, bool withOids,
 716                          const char *desc, teSection section,
 717                          const char *defn,
 718                          const char *dropStmt, const char *copyStmt,
 719                          const DumpId *deps, int nDeps,
 720                          DataDumperPtr dumpFn, void *dumpArg)
 721 {
 722         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 723         TocEntry   *newToc;
 724
 725         newToc = (TocEntry *) calloc(1, sizeof(TocEntry));
 726         if (!newToc)
 727                 die_horribly(AH, modulename, "out of memory\n");
 728
 729         AH->tocCount++;
 730         if (dumpId > AH->maxDumpId)
 731                 AH->maxDumpId = dumpId;
 732
 733         newToc->prev = AH->toc->prev;
 734         newToc->next = AH->toc;
 735         AH->toc->prev->next = newToc;
 736         AH->toc->prev = newToc;
 737
 738         newToc->catalogId = catalogId;
 739         newToc->dumpId = dumpId;
 740         newToc->section = section;
 741
 742         newToc->tag = strdup(tag);
 743         newToc->namespace = namespace ? strdup(namespace) : NULL;
 744         newToc->tablespace = tablespace ? strdup(tablespace) : NULL;
 745         newToc->owner = strdup(owner);
 746         newToc->withOids = withOids;
 747         newToc->desc = strdup(desc);
 748         newToc->defn = strdup(defn);
 749         newToc->dropStmt = strdup(dropStmt);
 750         newToc->copyStmt = copyStmt ? strdup(copyStmt) : NULL;
 751
 752         if (nDeps > 0)
 753         {
 754                 newToc->dependencies = (DumpId *) malloc(nDeps * sizeof(DumpId));
 755                 memcpy(newToc->dependencies, deps, nDeps * sizeof(DumpId));
 756                 newToc->nDeps = nDeps;
 757         }
 758         else
 759         {
 760                 newToc->dependencies = NULL;
 761                 newToc->nDeps = 0;
 762         }
 763
 764         newToc->dataDumper = dumpFn;
 765         newToc->dataDumperArg = dumpArg;
 766         newToc->hadDumper = dumpFn ? true : false;
 767
 768         newToc->formatData = NULL;
 769
 770         if (AH->ArchiveEntryPtr !=NULL)
 771                 (*AH->ArchiveEntryPtr) (AH, newToc);
 772 }
 773
 774 /* Public */
 775 void
 776 PrintTOCSummary(Archive *AHX, RestoreOptions *ropt)
 777 {
 778         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 779         TocEntry   *te;
 780         OutputContext sav;
 781         char       *fmtName;
 782
 783         if (ropt->filename)
 784                 sav = SetOutput(AH, ropt->filename, 0 /* no compression */ );
 785
 786         ahprintf(AH, ";\n; Archive created at %s", ctime(&AH->createDate));
 787         ahprintf(AH, ";     dbname: %s\n;     TOC Entries: %d\n;     Compression: %d\n",
 788                          AH->archdbname, AH->tocCount, AH->compression);
 789
 790         switch (AH->format)
 791         {
 792                 case archFiles:
 793                         fmtName = "FILES";
 794                         break;
 795                 case archCustom:
 796                         fmtName = "CUSTOM";
 797                         break;
 798                 case archTar:
 799                         fmtName = "TAR";
 800                         break;
 801                 default:
 802                         fmtName = "UNKNOWN";
 803         }
 804
 805         ahprintf(AH, ";     Dump Version: %d.%d-%d\n", AH->vmaj, AH->vmin, AH->vrev);
 806         ahprintf(AH, ";     Format: %s\n", fmtName);
 807         ahprintf(AH, ";     Integer: %d bytes\n", (int) AH->intSize);
 808         ahprintf(AH, ";     Offset: %d bytes\n", (int) AH->offSize);
 809         if (AH->archiveRemoteVersion)
 810                 ahprintf(AH, ";     Dumped from database version: %s\n",
 811                                  AH->archiveRemoteVersion);
 812         if (AH->archiveDumpVersion)
 813                 ahprintf(AH, ";     Dumped by pg_dump version: %s\n",
 814                                  AH->archiveDumpVersion);
 815
 816         ahprintf(AH, ";\n;\n; Selected TOC Entries:\n;\n");
 817
 818         /* We should print DATABASE entries whether or not -C was specified */
 819         ropt->createDB = 1;
 820
 821         for (te = AH->toc->next; te != AH->toc; te = te->next)
 822         {
 823                 if (ropt->verbose || _tocEntryRequired(te, ropt, true) != 0)
 824                         ahprintf(AH, "%d; %u %u %s %s %s %s\n", te->dumpId,
 825                                          te->catalogId.tableoid, te->catalogId.oid,
 826                                          te->desc, te->namespace ? te->namespace : "-",
 827                                          te->tag, te->owner);
 828                 if (ropt->verbose && te->nDeps > 0)
 829                 {
 830                         int                     i;
 831
 832                         ahprintf(AH, ";\tdepends on:");
 833                         for (i = 0; i < te->nDeps; i++)
 834                                 ahprintf(AH, " %d", te->dependencies[i]);
 835                         ahprintf(AH, "\n");
 836                 }
 837         }
 838
 839         if (ropt->filename)
 840                 ResetOutput(AH, sav);
 841 }
 842
 843 /***********
 844  * BLOB Archival
 845  ***********/
 846
 847 /* Called by a dumper to signal start of a BLOB */
 848 int
 849 StartBlob(Archive *AHX, Oid oid)
 850 {
 851         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 852
 853         if (!AH->StartBlobPtr)
 854                 die_horribly(AH, modulename, "large-object output not supported in chosen format\n");
 855
 856         (*AH->StartBlobPtr) (AH, AH->currToc, oid);
 857
 858         return 1;
 859 }
 860
 861 /* Called by a dumper to signal end of a BLOB */
 862 int
 863 EndBlob(Archive *AHX, Oid oid)
 864 {
 865         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 866
 867         if (AH->EndBlobPtr)
 868                 (*AH->EndBlobPtr) (AH, AH->currToc, oid);
 869
 870         return 1;
 871 }
 872
 873 /**********
 874  * BLOB Restoration
 875  **********/
 876
 877 /*
 878  * Called by a format handler before any blobs are restored
 879  */
 880 void
 881 StartRestoreBlobs(ArchiveHandle *AH)
 882 {
 883         if (!AH->ropt->single_txn)
 884         {
 885                 if (AH->connection)
 886                         StartTransaction(AH);
 887                 else
 888                         ahprintf(AH, "BEGIN;\n\n");
 889         }
 890
 891         AH->blobCount = 0;
 892 }
 893
 894 /*
 895  * Called by a format handler after all blobs are restored
 896  */
 897 void
 898 EndRestoreBlobs(ArchiveHandle *AH)
 899 {
 900         if (!AH->ropt->single_txn)
 901         {
 902                 if (AH->connection)
 903                         CommitTransaction(AH);
 904                 else
 905                         ahprintf(AH, "COMMIT;\n\n");
 906         }
 907
 908         ahlog(AH, 1, ngettext("restored %d large object\n",
 909                                                   "restored %d large objects\n",
 910                                                   AH->blobCount),
 911                   AH->blobCount);
 912 }
 913
 914
 915 /*
 916  * Called by a format handler to initiate restoration of a blob
 917  */
 918 void
 919 StartRestoreBlob(ArchiveHandle *AH, Oid oid, bool drop)
 920 {
 921         bool            old_blob_style = (AH->version < K_VERS_1_12);
 922         Oid                     loOid;
 923
 924         AH->blobCount++;
 925
 926         /* Initialize the LO Buffer */
 927         AH->lo_buf_used = 0;
 928
 929         ahlog(AH, 2, "restoring large object with OID %u\n", oid);
 930
 931         /* With an old archive we must do drop and create logic here */
 932         if (old_blob_style && drop)
 933                 DropBlobIfExists(AH, oid);
 934
 935         if (AH->connection)
 936         {
 937                 if (old_blob_style)
 938                 {
 939                         loOid = lo_create(AH->connection, oid);
 940                         if (loOid == 0 || loOid != oid)
 941                                 die_horribly(AH, modulename, "could not create large object %u: %s",
 942                                                          oid, PQerrorMessage(AH->connection));
 943                 }
 944                 AH->loFd = lo_open(AH->connection, oid, INV_WRITE);
 945                 if (AH->loFd == -1)
 946                         die_horribly(AH, modulename, "could not open large object %u: %s",
 947                                                  oid, PQerrorMessage(AH->connection));
 948         }
 949         else
 950         {
 951                 if (old_blob_style)
 952                         ahprintf(AH, "SELECT pg_catalog.lo_open(pg_catalog.lo_create('%u'), %d);\n",
 953                                          oid, INV_WRITE);
 954                 else
 955                         ahprintf(AH, "SELECT pg_catalog.lo_open('%u', %d);\n",
 956                                          oid, INV_WRITE);
 957         }
 958
 959         AH->writingBlob = 1;
 960 }
 961
 962 void
 963 EndRestoreBlob(ArchiveHandle *AH, Oid oid)
 964 {
 965         if (AH->lo_buf_used > 0)
 966         {
 967                 /* Write remaining bytes from the LO buffer */
 968                 dump_lo_buf(AH);
 969         }
 970
 971         AH->writingBlob = 0;
 972
 973         if (AH->connection)
 974         {
 975                 lo_close(AH->connection, AH->loFd);
 976                 AH->loFd = -1;
 977         }
 978         else
 979         {
 980                 ahprintf(AH, "SELECT pg_catalog.lo_close(0);\n\n");
 981         }
 982 }
 983
 984 /***********
 985  * Sorting and Reordering
 986  ***********/
 987
 988 void
 989 SortTocFromFile(Archive *AHX, RestoreOptions *ropt)
 990 {
 991         ArchiveHandle *AH = (ArchiveHandle *) AHX;
 992         FILE       *fh;
 993         char            buf[1024];
 994         char       *cmnt;
 995         char       *endptr;
 996         DumpId          id;
 997         TocEntry   *te;
 998
 999         /* Allocate space for the 'wanted' array, and init it */
1000         ropt->idWanted = (bool *) malloc(sizeof(bool) * AH->maxDumpId);
1001         memset(ropt->idWanted, 0, sizeof(bool) * AH->maxDumpId);
1002
1003         /* Setup the file */
1004         fh = fopen(ropt->tocFile, PG_BINARY_R);
1005         if (!fh)
1006                 die_horribly(AH, modulename, "could not open TOC file \"%s\": %s\n",
1007                                          ropt->tocFile, strerror(errno));
1008
1009         while (fgets(buf, sizeof(buf), fh) != NULL)
1010         {
1011                 /* Truncate line at comment, if any */
1012                 cmnt = strchr(buf, ';');
1013                 if (cmnt != NULL)
1014                         cmnt[0] = '\0';
1015
1016                 /* Ignore if all blank */
1017                 if (strspn(buf, " \t\r\n") == strlen(buf))
1018                         continue;
1019
1020                 /* Get an ID, check it's valid and not already seen */
1021                 id = strtol(buf, &endptr, 10);
1022                 if (endptr == buf || id <= 0 || id > AH->maxDumpId ||
1023                         ropt->idWanted[id - 1])
1024                 {
1025                         write_msg(modulename, "WARNING: line ignored: %s\n", buf);
1026                         continue;
1027                 }
1028
1029                 /* Find TOC entry */
1030                 te = getTocEntryByDumpId(AH, id);
1031                 if (!te)
1032                         die_horribly(AH, modulename, "could not find entry for ID %d\n",
1033                                                  id);
1034
1035                 /* Mark it wanted */
1036                 ropt->idWanted[id - 1] = true;
1037
1038                 /*
1039                  * Move each item to the end of the list as it is selected, so that
1040                  * they are placed in the desired order.  Any unwanted items will end
1041                  * up at the front of the list, which may seem unintuitive but it's
1042                  * what we need.  In an ordinary serial restore that makes no
1043                  * difference, but in a parallel restore we need to mark unrestored
1044                  * items' dependencies as satisfied before we start examining
1045                  * restorable items.  Otherwise they could have surprising
1046                  * side-effects on the order in which restorable items actually get
1047                  * restored.
1048                  */
1049                 _moveBefore(AH, AH->toc, te);
1050         }
1051
1052         if (fclose(fh) != 0)
1053                 die_horribly(AH, modulename, "could not close TOC file: %s\n",
1054                                          strerror(errno));
1055 }
1056
1057 /*
1058  * Set up a dummy ID filter that selects all dump IDs
1059  */
1060 void
1061 InitDummyWantedList(Archive *AHX, RestoreOptions *ropt)
1062 {
1063         ArchiveHandle *AH = (ArchiveHandle *) AHX;
1064
1065         /* Allocate space for the 'wanted' array, and init it to 1's */
1066         ropt->idWanted = (bool *) malloc(sizeof(bool) * AH->maxDumpId);
1067         memset(ropt->idWanted, 1, sizeof(bool) * AH->maxDumpId);
1068 }
1069
1070 /**********************
1071  * 'Convenience functions that look like standard IO functions
1072  * for writing data when in dump mode.
1073  **********************/
1074
1075 /* Public */
1076 int
1077 archputs(const char *s, Archive *AH)
1078 {
1079         return WriteData(AH, s, strlen(s));
1080 }
1081
1082 /* Public */
1083 int
1084 archprintf(Archive *AH, const char *fmt,...)
1085 {
1086         char       *p = NULL;
1087         va_list         ap;
1088         int                     bSize = strlen(fmt) + 256;
1089         int                     cnt = -1;
1090
1091         /*
1092          * This is paranoid: deal with the possibility that vsnprintf is willing
1093          * to ignore trailing null or returns > 0 even if string does not fit. It
1094          * may be the case that it returns cnt = bufsize
1095          */
1096         while (cnt < 0 || cnt >= (bSize - 1))
1097         {
1098                 if (p != NULL)
1099                         free(p);
1100                 bSize *= 2;
1101                 p = (char *) malloc(bSize);
1102                 if (p == NULL)
1103                         exit_horribly(AH, modulename, "out of memory\n");
1104                 va_start(ap, fmt);
1105                 cnt = vsnprintf(p, bSize, fmt, ap);
1106                 va_end(ap);
1107         }
1108         WriteData(AH, p, cnt);
1109         free(p);
1110         return cnt;
1111 }
1112
1113
1114 /*******************************
1115  * Stuff below here should be 'private' to the archiver routines
1116  *******************************/
1117
1118 static OutputContext
1119 SetOutput(ArchiveHandle *AH, char *filename, int compression)
1120 {
1121         OutputContext sav;
1122         int                     fn;
1123
1124         /* Replace the AH output file handle */
1125         sav.OF = AH->OF;
1126         sav.gzOut = AH->gzOut;
1127
1128         if (filename)
1129                 fn = -1;
1130         else if (AH->FH)
1131                 fn = fileno(AH->FH);
1132         else if (AH->fSpec)
1133         {
1134                 fn = -1;
1135                 filename = AH->fSpec;
1136         }
1137         else
1138                 fn = fileno(stdout);
1139
1140         /* If compression explicitly requested, use gzopen */
1141 #ifdef HAVE_LIBZ
1142         if (compression != 0)
1143         {
1144                 char            fmode[10];
1145
1146                 /* Don't use PG_BINARY_x since this is zlib */
1147                 sprintf(fmode, "wb%d", compression);
1148                 if (fn >= 0)
1149                         AH->OF = gzdopen(dup(fn), fmode);
1150                 else
1151                         AH->OF = gzopen(filename, fmode);
1152                 AH->gzOut = 1;
1153         }
1154         else
1155 #endif
1156         {                                                       /* Use fopen */
1157                 if (AH->mode == archModeAppend)
1158                 {
1159                         if (fn >= 0)
1160                                 AH->OF = fdopen(dup(fn), PG_BINARY_A);
1161                         else
1162                                 AH->OF = fopen(filename, PG_BINARY_A);
1163                 }
1164                 else
1165                 {
1166                         if (fn >= 0)
1167                                 AH->OF = fdopen(dup(fn), PG_BINARY_W);
1168                         else
1169                                 AH->OF = fopen(filename, PG_BINARY_W);
1170                 }
1171                 AH->gzOut = 0;
1172         }
1173
1174         if (!AH->OF)
1175         {
1176                 if (filename)
1177                         die_horribly(AH, modulename, "could not open output file \"%s\": %s\n",
1178                                                  filename, strerror(errno));
1179                 else
1180                         die_horribly(AH, modulename, "could not open output file: %s\n",
1181                                                  strerror(errno));
1182         }
1183
1184         return sav;
1185 }
1186
1187 static void
1188 ResetOutput(ArchiveHandle *AH, OutputContext sav)
1189 {
1190         int                     res;
1191
1192         if (AH->gzOut)
1193                 res = GZCLOSE(AH->OF);
1194         else
1195                 res = fclose(AH->OF);
1196
1197         if (res != 0)
1198                 die_horribly(AH, modulename, "could not close output file: %s\n",
1199                                          strerror(errno));
1200
1201         AH->gzOut = sav.gzOut;
1202         AH->OF = sav.OF;
1203 }
1204
1205
1206
1207 /*
1208  *      Print formatted text to the output file (usually stdout).
1209  */
1210 int
1211 ahprintf(ArchiveHandle *AH, const char *fmt,...)
1212 {
1213         char       *p = NULL;
1214         va_list         ap;
1215         int                     bSize = strlen(fmt) + 256;              /* Should be enough */
1216         int                     cnt = -1;
1217
1218         /*
1219          * This is paranoid: deal with the possibility that vsnprintf is willing
1220          * to ignore trailing null
1221          */
1222
1223         /*
1224          * or returns > 0 even if string does not fit. It may be the case that it
1225          * returns cnt = bufsize
1226          */
1227         while (cnt < 0 || cnt >= (bSize - 1))
1228         {
1229                 if (p != NULL)
1230                         free(p);
1231                 bSize *= 2;
1232                 p = (char *) malloc(bSize);
1233                 if (p == NULL)
1234                         die_horribly(AH, modulename, "out of memory\n");
1235                 va_start(ap, fmt);
1236                 cnt = vsnprintf(p, bSize, fmt, ap);
1237                 va_end(ap);
1238         }
1239         ahwrite(p, 1, cnt, AH);
1240         free(p);
1241         return cnt;
1242 }
1243
1244 void
1245 ahlog(ArchiveHandle *AH, int level, const char *fmt,...)
1246 {
1247         va_list         ap;
1248
1249         if (AH->debugLevel < level && (!AH->public.verbose || level > 1))
1250                 return;
1251
1252         va_start(ap, fmt);
1253         _write_msg(NULL, fmt, ap);
1254         va_end(ap);
1255 }
1256
1257 /*
1258  * Single place for logic which says 'We are restoring to a direct DB connection'.
1259  */
1260 static int
1261 RestoringToDB(ArchiveHandle *AH)
1262 {
1263         return (AH->ropt && AH->ropt->useDB && AH->connection);
1264 }
1265
1266 /*
1267  * Dump the current contents of the LO data buffer while writing a BLOB
1268  */
1269 static void
1270 dump_lo_buf(ArchiveHandle *AH)
1271 {
1272         if (AH->connection)
1273         {
1274                 size_t          res;
1275
1276                 res = lo_write(AH->connection, AH->loFd, AH->lo_buf, AH->lo_buf_used);
1277                 ahlog(AH, 5, ngettext("wrote %lu byte of large object data (result = %lu)\n",
1278                                          "wrote %lu bytes of large object data (result = %lu)\n",
1279                                                           AH->lo_buf_used),
1280                           (unsigned long) AH->lo_buf_used, (unsigned long) res);
1281                 if (res != AH->lo_buf_used)
1282                         die_horribly(AH, modulename,
1283                         "could not write to large object (result: %lu, expected: %lu)\n",
1284                                            (unsigned long) res, (unsigned long) AH->lo_buf_used);
1285         }
1286         else
1287         {
1288                 PQExpBuffer buf = createPQExpBuffer();
1289
1290                 appendByteaLiteralAHX(buf,
1291                                                           (const unsigned char *) AH->lo_buf,
1292                                                           AH->lo_buf_used,
1293                                                           AH);
1294
1295                 /* Hack: turn off writingBlob so ahwrite doesn't recurse to here */
1296                 AH->writingBlob = 0;
1297                 ahprintf(AH, "SELECT pg_catalog.lowrite(0, %s);\n", buf->data);
1298                 AH->writingBlob = 1;
1299
1300                 destroyPQExpBuffer(buf);
1301         }
1302         AH->lo_buf_used = 0;
1303 }
1304
1305
1306 /*
1307  *      Write buffer to the output file (usually stdout). This is user for
1308  *      outputting 'restore' scripts etc. It is even possible for an archive
1309  *      format to create a custom output routine to 'fake' a restore if it
1310  *      wants to generate a script (see TAR output).
1311  */
1312 int
1313 ahwrite(const void *ptr, size_t size, size_t nmemb, ArchiveHandle *AH)
1314 {
1315         size_t          res;
1316
1317         if (AH->writingBlob)
1318         {
1319                 size_t          remaining = size * nmemb;
1320
1321                 while (AH->lo_buf_used + remaining > AH->lo_buf_size)
1322                 {
1323                         size_t          avail = AH->lo_buf_size - AH->lo_buf_used;
1324
1325                         memcpy((char *) AH->lo_buf + AH->lo_buf_used, ptr, avail);
1326                         ptr = (const void *) ((const char *) ptr + avail);
1327                         remaining -= avail;
1328                         AH->lo_buf_used += avail;
1329                         dump_lo_buf(AH);
1330                 }
1331
1332                 memcpy((char *) AH->lo_buf + AH->lo_buf_used, ptr, remaining);
1333                 AH->lo_buf_used += remaining;
1334
1335                 return size * nmemb;
1336         }
1337         else if (AH->gzOut)
1338         {
1339                 res = GZWRITE((void *) ptr, size, nmemb, AH->OF);
1340                 if (res != (nmemb * size))
1341                         die_horribly(AH, modulename, "could not write to output file: %s\n", strerror(errno));
1342                 return res;
1343         }
1344         else if (AH->CustomOutPtr)
1345         {
1346                 res = AH->CustomOutPtr (AH, ptr, size * nmemb);
1347
1348                 if (res != (nmemb * size))
1349                         die_horribly(AH, modulename, "could not write to custom output routine\n");
1350                 return res;
1351         }
1352         else
1353         {
1354                 /*
1355                  * If we're doing a restore, and it's direct to DB, and we're
1356                  * connected then send it to the DB.
1357                  */
1358                 if (RestoringToDB(AH))
1359                         return ExecuteSqlCommandBuf(AH, (void *) ptr, size * nmemb);            /* Always 1, currently */
1360                 else
1361                 {
1362                         res = fwrite((void *) ptr, size, nmemb, AH->OF);
1363                         if (res != nmemb)
1364                                 die_horribly(AH, modulename, "could not write to output file: %s\n",
1365                                                          strerror(errno));
1366                         return res;
1367                 }
1368         }
1369 }
1370
1371 /* Common exit code */
1372 static void
1373 _write_msg(const char *modulename, const char *fmt, va_list ap)
1374 {
1375         if (modulename)
1376                 fprintf(stderr, "%s: [%s] ", progname, _(modulename));
1377         else
1378                 fprintf(stderr, "%s: ", progname);
1379         vfprintf(stderr, _(fmt), ap);
1380 }
1381
1382 void
1383 write_msg(const char *modulename, const char *fmt,...)
1384 {
1385         va_list         ap;
1386
1387         va_start(ap, fmt);
1388         _write_msg(modulename, fmt, ap);
1389         va_end(ap);
1390 }
1391
1392
1393 static void
1394 _die_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt, va_list ap)
1395 {
1396         _write_msg(modulename, fmt, ap);
1397
1398         if (AH)
1399         {
1400                 if (AH->public.verbose)
1401                         write_msg(NULL, "*** aborted because of error\n");
1402                 if (AH->connection)
1403                         PQfinish(AH->connection);
1404         }
1405
1406         exit(1);
1407 }
1408
1409 /* External use */
1410 void
1411 exit_horribly(Archive *AH, const char *modulename, const char *fmt,...)
1412 {
1413         va_list         ap;
1414
1415         va_start(ap, fmt);
1416         _die_horribly((ArchiveHandle *) AH, modulename, fmt, ap);
1417         va_end(ap);
1418 }
1419
1420 /* Archiver use (just different arg declaration) */
1421 void
1422 die_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt,...)
1423 {
1424         va_list         ap;
1425
1426         va_start(ap, fmt);
1427         _die_horribly(AH, modulename, fmt, ap);
1428         va_end(ap);
1429 }
1430
1431 /* on some error, we may decide to go on... */
1432 void
1433 warn_or_die_horribly(ArchiveHandle *AH,
1434                                          const char *modulename, const char *fmt,...)
1435 {
1436         va_list         ap;
1437
1438         switch (AH->stage)
1439         {
1440
1441                 case STAGE_NONE:
1442                         /* Do nothing special */
1443                         break;
1444
1445                 case STAGE_INITIALIZING:
1446                         if (AH->stage != AH->lastErrorStage)
1447                                 write_msg(modulename, "Error while INITIALIZING:\n");
1448                         break;
1449
1450                 case STAGE_PROCESSING:
1451                         if (AH->stage != AH->lastErrorStage)
1452                                 write_msg(modulename, "Error while PROCESSING TOC:\n");
1453                         break;
1454
1455                 case STAGE_FINALIZING:
1456                         if (AH->stage != AH->lastErrorStage)
1457                                 write_msg(modulename, "Error while FINALIZING:\n");
1458                         break;
1459         }
1460         if (AH->currentTE != NULL && AH->currentTE != AH->lastErrorTE)
1461         {
1462                 write_msg(modulename, "Error from TOC entry %d; %u %u %s %s %s\n",
1463                                   AH->currentTE->dumpId,
1464                          AH->currentTE->catalogId.tableoid, AH->currentTE->catalogId.oid,
1465                           AH->currentTE->desc, AH->currentTE->tag, AH->currentTE->owner);
1466         }
1467         AH->lastErrorStage = AH->stage;
1468         AH->lastErrorTE = AH->currentTE;
1469
1470         va_start(ap, fmt);
1471         if (AH->public.exit_on_error)
1472                 _die_horribly(AH, modulename, fmt, ap);
1473         else
1474         {
1475                 _write_msg(modulename, fmt, ap);
1476                 AH->public.n_errors++;
1477         }
1478         va_end(ap);
1479 }
1480
1481 #ifdef NOT_USED
1482
1483 static void
1484 _moveAfter(ArchiveHandle *AH, TocEntry *pos, TocEntry *te)
1485 {
1486         /* Unlink te from list */
1487         te->prev->next = te->next;
1488         te->next->prev = te->prev;
1489
1490         /* and insert it after "pos" */
1491         te->prev = pos;
1492         te->next = pos->next;
1493         pos->next->prev = te;
1494         pos->next = te;
1495 }
1496
1497 #endif
1498
1499 static void
1500 _moveBefore(ArchiveHandle *AH, TocEntry *pos, TocEntry *te)
1501 {
1502         /* Unlink te from list */
1503         te->prev->next = te->next;
1504         te->next->prev = te->prev;
1505
1506         /* and insert it before "pos" */
1507         te->prev = pos->prev;
1508         te->next = pos;
1509         pos->prev->next = te;
1510         pos->prev = te;
1511 }
1512
1513 static TocEntry *
1514 getTocEntryByDumpId(ArchiveHandle *AH, DumpId id)
1515 {
1516         TocEntry   *te;
1517
1518         for (te = AH->toc->next; te != AH->toc; te = te->next)
1519         {
1520                 if (te->dumpId == id)
1521                         return te;
1522         }
1523         return NULL;
1524 }
1525
1526 teReqs
1527 TocIDRequired(ArchiveHandle *AH, DumpId id, RestoreOptions *ropt)
1528 {
1529         TocEntry   *te = getTocEntryByDumpId(AH, id);
1530
1531         if (!te)
1532                 return 0;
1533
1534         return _tocEntryRequired(te, ropt, true);
1535 }
1536
1537 size_t
1538 WriteOffset(ArchiveHandle *AH, pgoff_t o, int wasSet)
1539 {
1540         int                     off;
1541
1542         /* Save the flag */
1543         (*AH->WriteBytePtr) (AH, wasSet);
1544
1545         /* Write out pgoff_t smallest byte first, prevents endian mismatch */
1546         for (off = 0; off < sizeof(pgoff_t); off++)
1547         {
1548                 (*AH->WriteBytePtr) (AH, o & 0xFF);
1549                 o >>= 8;
1550         }
1551         return sizeof(pgoff_t) + 1;
1552 }
1553
1554 int
1555 ReadOffset(ArchiveHandle *AH, pgoff_t * o)
1556 {
1557         int                     i;
1558         int                     off;
1559         int                     offsetFlg;
1560
1561         /* Initialize to zero */
1562         *o = 0;
1563
1564         /* Check for old version */
1565         if (AH->version < K_VERS_1_7)
1566         {
1567                 /* Prior versions wrote offsets using WriteInt */
1568                 i = ReadInt(AH);
1569                 /* -1 means not set */
1570                 if (i < 0)
1571                         return K_OFFSET_POS_NOT_SET;
1572                 else if (i == 0)
1573                         return K_OFFSET_NO_DATA;
1574
1575                 /* Cast to pgoff_t because it was written as an int. */
1576                 *o = (pgoff_t) i;
1577                 return K_OFFSET_POS_SET;
1578         }
1579
1580         /*
1581          * Read the flag indicating the state of the data pointer. Check if valid
1582          * and die if not.
1583          *
1584          * This used to be handled by a negative or zero pointer, now we use an
1585          * extra byte specifically for the state.
1586          */
1587         offsetFlg = (*AH->ReadBytePtr) (AH) & 0xFF;
1588
1589         switch (offsetFlg)
1590         {
1591                 case K_OFFSET_POS_NOT_SET:
1592                 case K_OFFSET_NO_DATA:
1593                 case K_OFFSET_POS_SET:
1594
1595                         break;
1596
1597                 default:
1598                         die_horribly(AH, modulename, "unexpected data offset flag %d\n", offsetFlg);
1599         }
1600
1601         /*
1602          * Read the bytes
1603          */
1604         for (off = 0; off < AH->offSize; off++)
1605         {
1606                 if (off < sizeof(pgoff_t))
1607                         *o |= ((pgoff_t) ((*AH->ReadBytePtr) (AH))) << (off * 8);
1608                 else
1609                 {
1610                         if ((*AH->ReadBytePtr) (AH) != 0)
1611                                 die_horribly(AH, modulename, "file offset in dump file is too large\n");
1612                 }
1613         }
1614
1615         return offsetFlg;
1616 }
1617
1618 size_t
1619 WriteInt(ArchiveHandle *AH, int i)
1620 {
1621         int                     b;
1622
1623         /*
1624          * This is a bit yucky, but I don't want to make the binary format very
1625          * dependent on representation, and not knowing much about it, I write out
1626          * a sign byte. If you change this, don't forget to change the file
1627          * version #, and modify readInt to read the new format AS WELL AS the old
1628          * formats.
1629          */
1630
1631         /* SIGN byte */
1632         if (i < 0)
1633         {
1634                 (*AH->WriteBytePtr) (AH, 1);
1635                 i = -i;
1636         }
1637         else
1638                 (*AH->WriteBytePtr) (AH, 0);
1639
1640         for (b = 0; b < AH->intSize; b++)
1641         {
1642                 (*AH->WriteBytePtr) (AH, i & 0xFF);
1643                 i >>= 8;
1644         }
1645
1646         return AH->intSize + 1;
1647 }
1648
1649 int
1650 ReadInt(ArchiveHandle *AH)
1651 {
1652         int                     res = 0;
1653         int                     bv,
1654                                 b;
1655         int                     sign = 0;               /* Default positive */
1656         int                     bitShift = 0;
1657
1658         if (AH->version > K_VERS_1_0)
1659                 /* Read a sign byte */
1660                 sign = (*AH->ReadBytePtr) (AH);
1661
1662         for (b = 0; b < AH->intSize; b++)
1663         {
1664                 bv = (*AH->ReadBytePtr) (AH) & 0xFF;
1665                 if (bv != 0)
1666                         res = res + (bv << bitShift);
1667                 bitShift += 8;
1668         }
1669
1670         if (sign)
1671                 res = -res;
1672
1673         return res;
1674 }
1675
1676 size_t
1677 WriteStr(ArchiveHandle *AH, const char *c)
1678 {
1679         size_t          res;
1680
1681         if (c)
1682         {
1683                 res = WriteInt(AH, strlen(c));
1684                 res += (*AH->WriteBufPtr) (AH, c, strlen(c));
1685         }
1686         else
1687                 res = WriteInt(AH, -1);
1688
1689         return res;
1690 }
1691
1692 char *
1693 ReadStr(ArchiveHandle *AH)
1694 {
1695         char       *buf;
1696         int                     l;
1697
1698         l = ReadInt(AH);
1699         if (l < 0)
1700                 buf = NULL;
1701         else
1702         {
1703                 buf = (char *) malloc(l + 1);
1704                 if (!buf)
1705                         die_horribly(AH, modulename, "out of memory\n");
1706
1707                 if ((*AH->ReadBufPtr) (AH, (void *) buf, l) != l)
1708                         die_horribly(AH, modulename, "unexpected end of file\n");
1709
1710                 buf[l] = '\0';
1711         }
1712
1713         return buf;
1714 }
1715
1716 static int
1717 _discoverArchiveFormat(ArchiveHandle *AH)
1718 {
1719         FILE       *fh;
1720         char            sig[6];                 /* More than enough */
1721         size_t          cnt;
1722         int                     wantClose = 0;
1723
1724 #if 0
1725         write_msg(modulename, "attempting to ascertain archive format\n");
1726 #endif
1727
1728         if (AH->lookahead)
1729                 free(AH->lookahead);
1730
1731         AH->lookaheadSize = 512;
1732         AH->lookahead = calloc(1, 512);
1733         AH->lookaheadLen = 0;
1734         AH->lookaheadPos = 0;
1735
1736         if (AH->fSpec)
1737         {
1738                 wantClose = 1;
1739                 fh = fopen(AH->fSpec, PG_BINARY_R);
1740                 if (!fh)
1741                         die_horribly(AH, modulename, "could not open input file \"%s\": %s\n",
1742                                                  AH->fSpec, strerror(errno));
1743         }
1744         else
1745         {
1746                 fh = stdin;
1747                 if (!fh)
1748                         die_horribly(AH, modulename, "could not open input file: %s\n",
1749                                                  strerror(errno));
1750         }
1751
1752         cnt = fread(sig, 1, 5, fh);
1753
1754         if (cnt != 5)
1755         {
1756                 if (ferror(fh))
1757                         die_horribly(AH, modulename, "could not read input file: %s\n", strerror(errno));
1758                 else
1759                         die_horribly(AH, modulename, "input file is too short (read %lu, expected 5)\n",
1760                                                  (unsigned long) cnt);
1761         }
1762
1763         /* Save it, just in case we need it later */
1764         strncpy(&AH->lookahead[0], sig, 5);
1765         AH->lookaheadLen = 5;
1766
1767         if (strncmp(sig, "PGDMP", 5) == 0)
1768         {
1769                 /*
1770                  * Finish reading (most of) a custom-format header.
1771                  *
1772                  * NB: this code must agree with ReadHead().
1773                  */
1774                 AH->vmaj = fgetc(fh);
1775                 AH->vmin = fgetc(fh);
1776
1777                 /* Save these too... */
1778                 AH->lookahead[AH->lookaheadLen++] = AH->vmaj;
1779                 AH->lookahead[AH->lookaheadLen++] = AH->vmin;
1780
1781                 /* Check header version; varies from V1.0 */
1782                 if (AH->vmaj > 1 || ((AH->vmaj == 1) && (AH->vmin > 0)))                /* Version > 1.0 */
1783                 {
1784                         AH->vrev = fgetc(fh);
1785                         AH->lookahead[AH->lookaheadLen++] = AH->vrev;
1786                 }
1787                 else
1788                         AH->vrev = 0;
1789
1790                 /* Make a convenient integer <maj><min><rev>00 */
1791                 AH->version = ((AH->vmaj * 256 + AH->vmin) * 256 + AH->vrev) * 256 + 0;
1792
1793                 AH->intSize = fgetc(fh);
1794                 AH->lookahead[AH->lookaheadLen++] = AH->intSize;
1795
1796                 if (AH->version >= K_VERS_1_7)
1797                 {
1798                         AH->offSize = fgetc(fh);
1799                         AH->lookahead[AH->lookaheadLen++] = AH->offSize;
1800                 }
1801                 else
1802                         AH->offSize = AH->intSize;
1803
1804                 AH->format = fgetc(fh);
1805                 AH->lookahead[AH->lookaheadLen++] = AH->format;
1806         }
1807         else
1808         {
1809                 /*
1810                  * *Maybe* we have a tar archive format file... So, read first 512
1811                  * byte header...
1812                  */
1813                 cnt = fread(&AH->lookahead[AH->lookaheadLen], 1, 512 - AH->lookaheadLen, fh);
1814                 AH->lookaheadLen += cnt;
1815
1816                 if (AH->lookaheadLen != 512)
1817                         die_horribly(AH, modulename, "input file does not appear to be a valid archive (too short?)\n");
1818
1819                 if (!isValidTarHeader(AH->lookahead))
1820                         die_horribly(AH, modulename, "input file does not appear to be a valid archive\n");
1821
1822                 AH->format = archTar;
1823         }
1824
1825         /* If we can't seek, then mark the header as read */
1826         if (fseeko(fh, 0, SEEK_SET) != 0)
1827         {
1828                 /*
1829                  * NOTE: Formats that use the lookahead buffer can unset this in their
1830                  * Init routine.
1831                  */
1832                 AH->readHeader = 1;
1833         }
1834         else
1835                 AH->lookaheadLen = 0;   /* Don't bother since we've reset the file */
1836
1837         /* Close the file */
1838         if (wantClose)
1839                 if (fclose(fh) != 0)
1840                         die_horribly(AH, modulename, "could not close input file: %s\n",
1841                                                  strerror(errno));
1842
1843         return AH->format;
1844 }
1845
1846
1847 /*
1848  * Allocate an archive handle
1849  */
1850 static ArchiveHandle *
1851 _allocAH(const char *FileSpec, const ArchiveFormat fmt,
1852                  const int compression, ArchiveMode mode)
1853 {
1854         ArchiveHandle *AH;
1855
1856 #if 0
1857         write_msg(modulename, "allocating AH for %s, format %d\n", FileSpec, fmt);
1858 #endif
1859
1860         AH = (ArchiveHandle *) calloc(1, sizeof(ArchiveHandle));
1861         if (!AH)
1862                 die_horribly(AH, modulename, "out of memory\n");
1863
1864         /* AH->debugLevel = 100; */
1865
1866         AH->vmaj = K_VERS_MAJOR;
1867         AH->vmin = K_VERS_MINOR;
1868         AH->vrev = K_VERS_REV;
1869
1870         /* Make a convenient integer <maj><min><rev>00 */
1871         AH->version = ((AH->vmaj * 256 + AH->vmin) * 256 + AH->vrev) * 256 + 0;
1872
1873         /* initialize for backwards compatible string processing */
1874         AH->public.encoding = 0;        /* PG_SQL_ASCII */
1875         AH->public.std_strings = false;
1876
1877         /* sql error handling */
1878         AH->public.exit_on_error = true;
1879         AH->public.n_errors = 0;
1880
1881         AH->archiveDumpVersion = PG_VERSION;
1882
1883         AH->createDate = time(NULL);
1884
1885         AH->intSize = sizeof(int);
1886         AH->offSize = sizeof(pgoff_t);
1887         if (FileSpec)
1888         {
1889                 AH->fSpec = strdup(FileSpec);
1890
1891                 /*
1892                  * Not used; maybe later....
1893                  *
1894                  * AH->workDir = strdup(FileSpec); for(i=strlen(FileSpec) ; i > 0 ;
1895                  * i--) if (AH->workDir[i-1] == '/')
1896                  */
1897         }
1898         else
1899                 AH->fSpec = NULL;
1900
1901         AH->currUser = NULL;            /* unknown */
1902         AH->currSchema = NULL;          /* ditto */
1903         AH->currTablespace = NULL;      /* ditto */
1904         AH->currWithOids = -1;          /* force SET */
1905
1906         AH->toc = (TocEntry *) calloc(1, sizeof(TocEntry));
1907         if (!AH->toc)
1908                 die_horribly(AH, modulename, "out of memory\n");
1909
1910         AH->toc->next = AH->toc;
1911         AH->toc->prev = AH->toc;
1912
1913         AH->mode = mode;
1914         AH->compression = compression;
1915
1916         AH->pgCopyBuf = createPQExpBuffer();
1917         AH->sqlBuf = createPQExpBuffer();
1918
1919         /* Open stdout with no compression for AH output handle */
1920         AH->gzOut = 0;
1921         AH->OF = stdout;
1922
1923         /*
1924          * On Windows, we need to use binary mode to read/write non-text archive
1925          * formats.  Force stdin/stdout into binary mode if that is what we are
1926          * using.
1927          */
1928 #ifdef WIN32
1929         if (fmt != archNull &&
1930                 (AH->fSpec == NULL || strcmp(AH->fSpec, "") == 0))
1931         {
1932                 if (mode == archModeWrite)
1933                         setmode(fileno(stdout), O_BINARY);
1934                 else
1935                         setmode(fileno(stdin), O_BINARY);
1936         }
1937 #endif
1938
1939         if (fmt == archUnknown)
1940                 AH->format = _discoverArchiveFormat(AH);
1941         else
1942                 AH->format = fmt;
1943
1944         AH->promptPassword = TRI_DEFAULT;
1945
1946         switch (AH->format)
1947         {
1948                 case archCustom:
1949                         InitArchiveFmt_Custom(AH);
1950                         break;
1951
1952                 case archFiles:
1953                         InitArchiveFmt_Files(AH);
1954                         break;
1955
1956                 case archNull:
1957                         InitArchiveFmt_Null(AH);
1958                         break;
1959
1960                 case archTar:
1961                         InitArchiveFmt_Tar(AH);
1962                         break;
1963
1964                 default:
1965                         die_horribly(AH, modulename, "unrecognized file format \"%d\"\n", fmt);
1966         }
1967
1968         return AH;
1969 }
1970
1971
1972 void
1973 WriteDataChunks(ArchiveHandle *AH)
1974 {
1975         TocEntry   *te;
1976         StartDataPtr startPtr;
1977         EndDataPtr      endPtr;
1978
1979         for (te = AH->toc->next; te != AH->toc; te = te->next)
1980         {
1981                 if (te->dataDumper != NULL)
1982                 {
1983                         AH->currToc = te;
1984                         /* printf("Writing data for %d (%x)\n", te->id, te); */
1985
1986                         if (strcmp(te->desc, "BLOBS") == 0)
1987                         {
1988                                 startPtr = AH->StartBlobsPtr;
1989                                 endPtr = AH->EndBlobsPtr;
1990                         }
1991                         else
1992                         {
1993                                 startPtr = AH->StartDataPtr;
1994                                 endPtr = AH->EndDataPtr;
1995                         }
1996
1997                         if (startPtr != NULL)
1998                                 (*startPtr) (AH, te);
1999
2000                         /*
2001                          * printf("Dumper arg for %d is %x\n", te->id, te->dataDumperArg);
2002                          */
2003
2004                         /*
2005                          * The user-provided DataDumper routine needs to call
2006                          * AH->WriteData
2007                          */
2008                         (*te->dataDumper) ((Archive *) AH, te->dataDumperArg);
2009
2010                         if (endPtr != NULL)
2011                                 (*endPtr) (AH, te);
2012                         AH->currToc = NULL;
2013                 }
2014         }
2015 }
2016
2017 void
2018 WriteToc(ArchiveHandle *AH)
2019 {
2020         TocEntry   *te;
2021         char            workbuf[32];
2022         int                     i;
2023
2024         /* printf("%d TOC Entries to save\n", AH->tocCount); */
2025
2026         WriteInt(AH, AH->tocCount);
2027
2028         for (te = AH->toc->next; te != AH->toc; te = te->next)
2029         {
2030                 WriteInt(AH, te->dumpId);
2031                 WriteInt(AH, te->dataDumper ? 1 : 0);
2032
2033                 /* OID is recorded as a string for historical reasons */
2034                 sprintf(workbuf, "%u", te->catalogId.tableoid);
2035                 WriteStr(AH, workbuf);
2036                 sprintf(workbuf, "%u", te->catalogId.oid);
2037                 WriteStr(AH, workbuf);
2038
2039                 WriteStr(AH, te->tag);
2040                 WriteStr(AH, te->desc);
2041                 WriteInt(AH, te->section);
2042                 WriteStr(AH, te->defn);
2043                 WriteStr(AH, te->dropStmt);
2044                 WriteStr(AH, te->copyStmt);
2045                 WriteStr(AH, te->namespace);
2046                 WriteStr(AH, te->tablespace);
2047                 WriteStr(AH, te->owner);
2048                 WriteStr(AH, te->withOids ? "true" : "false");
2049
2050                 /* Dump list of dependencies */
2051                 for (i = 0; i < te->nDeps; i++)
2052                 {
2053                         sprintf(workbuf, "%d", te->dependencies[i]);
2054                         WriteStr(AH, workbuf);
2055                 }
2056                 WriteStr(AH, NULL);             /* Terminate List */
2057
2058                 if (AH->WriteExtraTocPtr)
2059                         (*AH->WriteExtraTocPtr) (AH, te);
2060         }
2061 }
2062
2063 void
2064 ReadToc(ArchiveHandle *AH)
2065 {
2066         int                     i;
2067         char       *tmp;
2068         DumpId     *deps;
2069         int                     depIdx;
2070         int                     depSize;
2071         TocEntry   *te;
2072
2073         AH->tocCount = ReadInt(AH);
2074         AH->maxDumpId = 0;
2075
2076         for (i = 0; i < AH->tocCount; i++)
2077         {
2078                 te = (TocEntry *) calloc(1, sizeof(TocEntry));
2079                 te->dumpId = ReadInt(AH);
2080
2081                 if (te->dumpId > AH->maxDumpId)
2082                         AH->maxDumpId = te->dumpId;
2083
2084                 /* Sanity check */
2085                 if (te->dumpId <= 0)
2086                         die_horribly(AH, modulename,
2087                                            "entry ID %d out of range -- perhaps a corrupt TOC\n",
2088                                                  te->dumpId);
2089
2090                 te->hadDumper = ReadInt(AH);
2091
2092                 if (AH->version >= K_VERS_1_8)
2093                 {
2094                         tmp = ReadStr(AH);
2095                         sscanf(tmp, "%u", &te->catalogId.tableoid);
2096                         free(tmp);
2097                 }
2098                 else
2099                         te->catalogId.tableoid = InvalidOid;
2100                 tmp = ReadStr(AH);
2101                 sscanf(tmp, "%u", &te->catalogId.oid);
2102                 free(tmp);
2103
2104                 te->tag = ReadStr(AH);
2105                 te->desc = ReadStr(AH);
2106
2107                 if (AH->version >= K_VERS_1_11)
2108                 {
2109                         te->section = ReadInt(AH);
2110                 }
2111                 else
2112                 {
2113                         /*
2114                          * Rules for pre-8.4 archives wherein pg_dump hasn't classified
2115                          * the entries into sections.  This list need not cover entry
2116                          * types added later than 8.4.
2117                          */
2118                         if (strcmp(te->desc, "COMMENT") == 0 ||
2119                                 strcmp(te->desc, "ACL") == 0 ||
2120                                 strcmp(te->desc, "ACL LANGUAGE") == 0)
2121                                 te->section = SECTION_NONE;
2122                         else if (strcmp(te->desc, "TABLE DATA") == 0 ||
2123                                          strcmp(te->desc, "BLOBS") == 0 ||
2124                                          strcmp(te->desc, "BLOB COMMENTS") == 0)
2125                                 te->section = SECTION_DATA;
2126                         else if (strcmp(te->desc, "CONSTRAINT") == 0 ||
2127                                          strcmp(te->desc, "CHECK CONSTRAINT") == 0 ||
2128                                          strcmp(te->desc, "FK CONSTRAINT") == 0 ||
2129                                          strcmp(te->desc, "INDEX") == 0 ||
2130                                          strcmp(te->desc, "RULE") == 0 ||
2131                                          strcmp(te->desc, "TRIGGER") == 0)
2132                                 te->section = SECTION_POST_DATA;
2133                         else
2134                                 te->section = SECTION_PRE_DATA;
2135                 }
2136
2137                 te->defn = ReadStr(AH);
2138                 te->dropStmt = ReadStr(AH);
2139
2140                 if (AH->version >= K_VERS_1_3)
2141                         te->copyStmt = ReadStr(AH);
2142
2143                 if (AH->version >= K_VERS_1_6)
2144                         te->namespace = ReadStr(AH);
2145
2146                 if (AH->version >= K_VERS_1_10)
2147                         te->tablespace = ReadStr(AH);
2148
2149                 te->owner = ReadStr(AH);
2150                 if (AH->version >= K_VERS_1_9)
2151                 {
2152                         if (strcmp(ReadStr(AH), "true") == 0)
2153                                 te->withOids = true;
2154                         else
2155                                 te->withOids = false;
2156                 }
2157                 else
2158                         te->withOids = true;
2159
2160                 /* Read TOC entry dependencies */
2161                 if (AH->version >= K_VERS_1_5)
2162                 {
2163                         depSize = 100;
2164                         deps = (DumpId *) malloc(sizeof(DumpId) * depSize);
2165                         depIdx = 0;
2166                         for (;;)
2167                         {
2168                                 tmp = ReadStr(AH);
2169                                 if (!tmp)
2170                                         break;          /* end of list */
2171                                 if (depIdx >= depSize)
2172                                 {
2173                                         depSize *= 2;
2174                                         deps = (DumpId *) realloc(deps, sizeof(DumpId) * depSize);
2175                                 }
2176                                 sscanf(tmp, "%d", &deps[depIdx]);
2177                                 free(tmp);
2178                                 depIdx++;
2179                         }
2180
2181                         if (depIdx > 0)         /* We have a non-null entry */
2182                         {
2183                                 deps = (DumpId *) realloc(deps, sizeof(DumpId) * depIdx);
2184                                 te->dependencies = deps;
2185                                 te->nDeps = depIdx;
2186                         }
2187                         else
2188                         {
2189                                 free(deps);
2190                                 te->dependencies = NULL;
2191                                 te->nDeps = 0;
2192                         }
2193                 }
2194                 else
2195                 {
2196                         te->dependencies = NULL;
2197                         te->nDeps = 0;
2198                 }
2199
2200                 if (AH->ReadExtraTocPtr)
2201                         (*AH->ReadExtraTocPtr) (AH, te);
2202
2203                 ahlog(AH, 3, "read TOC entry %d (ID %d) for %s %s\n",
2204                           i, te->dumpId, te->desc, te->tag);
2205
2206                 /* link completed entry into TOC circular list */
2207                 te->prev = AH->toc->prev;
2208                 AH->toc->prev->next = te;
2209                 AH->toc->prev = te;
2210                 te->next = AH->toc;
2211
2212                 /* special processing immediately upon read for some items */
2213                 if (strcmp(te->desc, "ENCODING") == 0)
2214                         processEncodingEntry(AH, te);
2215                 else if (strcmp(te->desc, "STDSTRINGS") == 0)
2216                         processStdStringsEntry(AH, te);
2217         }
2218 }
2219
2220 static void
2221 processEncodingEntry(ArchiveHandle *AH, TocEntry *te)
2222 {
2223         /* te->defn should have the form SET client_encoding = 'foo'; */
2224         char       *defn = strdup(te->defn);
2225         char       *ptr1;
2226         char       *ptr2 = NULL;
2227         int                     encoding;
2228
2229         ptr1 = strchr(defn, '\'');
2230         if (ptr1)
2231                 ptr2 = strchr(++ptr1, '\'');
2232         if (ptr2)
2233         {
2234                 *ptr2 = '\0';
2235                 encoding = pg_char_to_encoding(ptr1);
2236                 if (encoding < 0)
2237                         die_horribly(AH, modulename, "unrecognized encoding \"%s\"\n",
2238                                                  ptr1);
2239                 AH->public.encoding = encoding;
2240         }
2241         else
2242                 die_horribly(AH, modulename, "invalid ENCODING item: %s\n",
2243                                          te->defn);
2244
2245         free(defn);
2246 }
2247
2248 static void
2249 processStdStringsEntry(ArchiveHandle *AH, TocEntry *te)
2250 {
2251         /* te->defn should have the form SET standard_conforming_strings = 'x'; */
2252         char       *ptr1;
2253
2254         ptr1 = strchr(te->defn, '\'');
2255         if (ptr1 && strncmp(ptr1, "'on'", 4) == 0)
2256                 AH->public.std_strings = true;
2257         else if (ptr1 && strncmp(ptr1, "'off'", 5) == 0)
2258                 AH->public.std_strings = false;
2259         else
2260                 die_horribly(AH, modulename, "invalid STDSTRINGS item: %s\n",
2261                                          te->defn);
2262 }
2263
2264 static teReqs
2265 _tocEntryRequired(TocEntry *te, RestoreOptions *ropt, bool include_acls)
2266 {
2267         teReqs          res = REQ_ALL;
2268
2269         /* ENCODING and STDSTRINGS items are dumped specially, so always reject */
2270         if (strcmp(te->desc, "ENCODING") == 0 ||
2271                 strcmp(te->desc, "STDSTRINGS") == 0)
2272                 return 0;
2273
2274         /* If it's an ACL, maybe ignore it */
2275         if ((!include_acls || ropt->aclsSkip) && _tocEntryIsACL(te))
2276                 return 0;
2277
2278         /* Ignore DATABASE entry unless we should create it */
2279         if (!ropt->createDB && strcmp(te->desc, "DATABASE") == 0)
2280                 return 0;
2281
2282         /* Check options for selective dump/restore */
2283         if (ropt->schemaNames)
2284         {
2285                 /* If no namespace is specified, it means all. */
2286                 if (!te->namespace)
2287                         return 0;
2288                 if (strcmp(ropt->schemaNames, te->namespace) != 0)
2289                         return 0;
2290         }
2291
2292         if (ropt->selTypes)
2293         {
2294                 if (strcmp(te->desc, "TABLE") == 0 ||
2295                         strcmp(te->desc, "TABLE DATA") == 0)
2296                 {
2297                         if (!ropt->selTable)
2298                                 return 0;
2299                         if (ropt->tableNames && strcmp(ropt->tableNames, te->tag) != 0)
2300                                 return 0;
2301                 }
2302                 else if (strcmp(te->desc, "INDEX") == 0)
2303                 {
2304                         if (!ropt->selIndex)
2305                                 return 0;
2306                         if (ropt->indexNames && strcmp(ropt->indexNames, te->tag) != 0)
2307                                 return 0;
2308                 }
2309                 else if (strcmp(te->desc, "FUNCTION") == 0)
2310                 {
2311                         if (!ropt->selFunction)
2312                                 return 0;
2313                         if (ropt->functionNames && strcmp(ropt->functionNames, te->tag) != 0)
2314                                 return 0;
2315                 }
2316                 else if (strcmp(te->desc, "TRIGGER") == 0)
2317                 {
2318                         if (!ropt->selTrigger)
2319                                 return 0;
2320                         if (ropt->triggerNames && strcmp(ropt->triggerNames, te->tag) != 0)
2321                                 return 0;
2322                 }
2323                 else
2324                         return 0;
2325         }
2326
2327         /*
2328          * Check if we had a dataDumper. Indicates if the entry is schema or data
2329          */
2330         if (!te->hadDumper)
2331         {
2332                 /*
2333                  * Special Case: If 'SEQUENCE SET' or anything to do with BLOBs, then
2334                  * it is considered a data entry.  We don't need to check for the
2335                  * BLOBS entry or old-style BLOB COMMENTS, because they will have
2336                  * hadDumper = true ... but we do need to check new-style BLOB
2337                  * comments.
2338                  */
2339                 if (strcmp(te->desc, "SEQUENCE SET") == 0 ||
2340                         strcmp(te->desc, "BLOB") == 0 ||
2341                         (strcmp(te->desc, "ACL") == 0 &&
2342                          strncmp(te->tag, "LARGE OBJECT ", 13) == 0) ||
2343                         (strcmp(te->desc, "COMMENT") == 0 &&
2344                          strncmp(te->tag, "LARGE OBJECT ", 13) == 0))
2345                         res = res & REQ_DATA;
2346                 else
2347                         res = res & ~REQ_DATA;
2348         }
2349
2350         /*
2351          * Special case: <Init> type with <Max OID> tag; this is obsolete and we
2352          * always ignore it.
2353          */
2354         if ((strcmp(te->desc, "<Init>") == 0) && (strcmp(te->tag, "Max OID") == 0))
2355                 return 0;
2356
2357         /* Mask it if we only want schema */
2358         if (ropt->schemaOnly)
2359                 res = res & REQ_SCHEMA;
2360
2361         /* Mask it we only want data */
2362         if (ropt->dataOnly)
2363                 res = res & REQ_DATA;
2364
2365         /* Mask it if we don't have a schema contribution */
2366         if (!te->defn || strlen(te->defn) == 0)
2367                 res = res & ~REQ_SCHEMA;
2368
2369         /* Finally, if there's a per-ID filter, limit based on that as well */
2370         if (ropt->idWanted && !ropt->idWanted[te->dumpId - 1])
2371                 return 0;
2372
2373         return res;
2374 }
2375
2376 /*
2377  * Identify TOC entries that are ACLs.
2378  */
2379 static bool
2380 _tocEntryIsACL(TocEntry *te)
2381 {
2382         /* "ACL LANGUAGE" was a crock emitted only in PG 7.4 */
2383         if (strcmp(te->desc, "ACL") == 0 ||
2384                 strcmp(te->desc, "ACL LANGUAGE") == 0 ||
2385                 strcmp(te->desc, "DEFAULT ACL") == 0)
2386                 return true;
2387         return false;
2388 }
2389
2390 /*
2391  * Issue SET commands for parameters that we want to have set the same way
2392  * at all times during execution of a restore script.
2393  */
2394 static void
2395 _doSetFixedOutputState(ArchiveHandle *AH)
2396 {
2397         /* Disable statement_timeout in archive for pg_restore/psql  */
2398         ahprintf(AH, "SET statement_timeout = 0;\n");
2399
2400         /* Select the correct character set encoding */
2401         ahprintf(AH, "SET client_encoding = '%s';\n",
2402                          pg_encoding_to_char(AH->public.encoding));
2403
2404         /* Select the correct string literal syntax */
2405         ahprintf(AH, "SET standard_conforming_strings = %s;\n",
2406                          AH->public.std_strings ? "on" : "off");
2407
2408         /* Select the role to be used during restore */
2409         if (AH->ropt && AH->ropt->use_role)
2410                 ahprintf(AH, "SET ROLE %s;\n", fmtId(AH->ropt->use_role));
2411
2412         /* Make sure function checking is disabled */
2413         ahprintf(AH, "SET check_function_bodies = false;\n");
2414
2415         /* Avoid annoying notices etc */
2416         ahprintf(AH, "SET client_min_messages = warning;\n");
2417         if (!AH->public.std_strings)
2418                 ahprintf(AH, "SET escape_string_warning = off;\n");
2419
2420         ahprintf(AH, "\n");
2421 }
2422
2423 /*
2424  * Issue a SET SESSION AUTHORIZATION command.  Caller is responsible
2425  * for updating state if appropriate.  If user is NULL or an empty string,
2426  * the specification DEFAULT will be used.
2427  */
2428 static void
2429 _doSetSessionAuth(ArchiveHandle *AH, const char *user)
2430 {
2431         PQExpBuffer cmd = createPQExpBuffer();
2432
2433         appendPQExpBuffer(cmd, "SET SESSION AUTHORIZATION ");
2434
2435         /*
2436          * SQL requires a string literal here.  Might as well be correct.
2437          */
2438         if (user && *user)
2439                 appendStringLiteralAHX(cmd, user, AH);
2440         else
2441                 appendPQExpBuffer(cmd, "DEFAULT");
2442         appendPQExpBuffer(cmd, ";");
2443
2444         if (RestoringToDB(AH))
2445         {
2446                 PGresult   *res;
2447
2448                 res = PQexec(AH->connection, cmd->data);
2449
2450                 if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
2451                         /* NOT warn_or_die_horribly... use -O instead to skip this. */
2452                         die_horribly(AH, modulename, "could not set session user to \"%s\": %s",
2453                                                  user, PQerrorMessage(AH->connection));
2454
2455                 PQclear(res);
2456         }
2457         else
2458                 ahprintf(AH, "%s\n\n", cmd->data);
2459
2460         destroyPQExpBuffer(cmd);
2461 }
2462
2463
2464 /*
2465  * Issue a SET default_with_oids command.  Caller is responsible
2466  * for updating state if appropriate.
2467  */
2468 static void
2469 _doSetWithOids(ArchiveHandle *AH, const bool withOids)
2470 {
2471         PQExpBuffer cmd = createPQExpBuffer();
2472
2473         appendPQExpBuffer(cmd, "SET default_with_oids = %s;", withOids ?
2474                                           "true" : "false");
2475
2476         if (RestoringToDB(AH))
2477         {
2478                 PGresult   *res;
2479
2480                 res = PQexec(AH->connection, cmd->data);
2481
2482                 if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
2483                         warn_or_die_horribly(AH, modulename,
2484                                                                  "could not set default_with_oids: %s",
2485                                                                  PQerrorMessage(AH->connection));
2486
2487                 PQclear(res);
2488         }
2489         else
2490                 ahprintf(AH, "%s\n\n", cmd->data);
2491
2492         destroyPQExpBuffer(cmd);
2493 }
2494
2495
2496 /*
2497  * Issue the commands to connect to the specified database.
2498  *
2499  * If we're currently restoring right into a database, this will
2500  * actually establish a connection. Otherwise it puts a \connect into
2501  * the script output.
2502  *
2503  * NULL dbname implies reconnecting to the current DB (pretty useless).
2504  */
2505 static void
2506 _reconnectToDB(ArchiveHandle *AH, const char *dbname)
2507 {
2508         if (RestoringToDB(AH))
2509                 ReconnectToServer(AH, dbname, NULL);
2510         else
2511         {
2512                 PQExpBuffer qry = createPQExpBuffer();
2513
2514                 appendPQExpBuffer(qry, "\\connect %s\n\n",
2515                                                   dbname ? fmtId(dbname) : "-");
2516                 ahprintf(AH, "%s", qry->data);
2517                 destroyPQExpBuffer(qry);
2518         }
2519
2520         /*
2521          * NOTE: currUser keeps track of what the imaginary session user in our
2522          * script is.  It's now effectively reset to the original userID.
2523          */
2524         if (AH->currUser)
2525                 free(AH->currUser);
2526         AH->currUser = NULL;
2527
2528         /* don't assume we still know the output schema, tablespace, etc either */
2529         if (AH->currSchema)
2530                 free(AH->currSchema);
2531         AH->currSchema = NULL;
2532         if (AH->currTablespace)
2533                 free(AH->currTablespace);
2534         AH->currTablespace = NULL;
2535         AH->currWithOids = -1;
2536
2537         /* re-establish fixed state */
2538         _doSetFixedOutputState(AH);
2539 }
2540
2541 /*
2542  * Become the specified user, and update state to avoid redundant commands
2543  *
2544  * NULL or empty argument is taken to mean restoring the session default
2545  */
2546 static void
2547 _becomeUser(ArchiveHandle *AH, const char *user)
2548 {
2549         if (!user)
2550                 user = "";                              /* avoid null pointers */
2551
2552         if (AH->currUser && strcmp(AH->currUser, user) == 0)
2553                 return;                                 /* no need to do anything */
2554
2555         _doSetSessionAuth(AH, user);
2556
2557         /*
2558          * NOTE: currUser keeps track of what the imaginary session user in our
2559          * script is
2560          */
2561         if (AH->currUser)
2562                 free(AH->currUser);
2563         AH->currUser = strdup(user);
2564 }
2565
2566 /*
2567  * Become the owner of the given TOC entry object.      If
2568  * changes in ownership are not allowed, this doesn't do anything.
2569  */
2570 static void
2571 _becomeOwner(ArchiveHandle *AH, TocEntry *te)
2572 {
2573         if (AH->ropt && (AH->ropt->noOwner || !AH->ropt->use_setsessauth))
2574                 return;
2575
2576         _becomeUser(AH, te->owner);
2577 }
2578
2579
2580 /*
2581  * Set the proper default_with_oids value for the table.
2582  */
2583 static void
2584 _setWithOids(ArchiveHandle *AH, TocEntry *te)
2585 {
2586         if (AH->currWithOids != te->withOids)
2587         {
2588                 _doSetWithOids(AH, te->withOids);
2589                 AH->currWithOids = te->withOids;
2590         }
2591 }
2592
2593
2594 /*
2595  * Issue the commands to select the specified schema as the current schema
2596  * in the target database.
2597  */
2598 static void
2599 _selectOutputSchema(ArchiveHandle *AH, const char *schemaName)
2600 {
2601         PQExpBuffer qry;
2602
2603         if (!schemaName || *schemaName == '\0' ||
2604                 (AH->currSchema && strcmp(AH->currSchema, schemaName) == 0))
2605                 return;                                 /* no need to do anything */
2606
2607         qry = createPQExpBuffer();
2608
2609         appendPQExpBuffer(qry, "SET search_path = %s",
2610                                           fmtId(schemaName));
2611         if (strcmp(schemaName, "pg_catalog") != 0)
2612                 appendPQExpBuffer(qry, ", pg_catalog");
2613
2614         if (RestoringToDB(AH))
2615         {
2616                 PGresult   *res;
2617
2618                 res = PQexec(AH->connection, qry->data);
2619
2620                 if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
2621                         warn_or_die_horribly(AH, modulename,
2622                                                                  "could not set search_path to \"%s\": %s",
2623                                                                  schemaName, PQerrorMessage(AH->connection));
2624
2625                 PQclear(res);
2626         }
2627         else
2628                 ahprintf(AH, "%s;\n\n", qry->data);
2629
2630         if (AH->currSchema)
2631                 free(AH->currSchema);
2632         AH->currSchema = strdup(schemaName);
2633
2634         destroyPQExpBuffer(qry);
2635 }
2636
2637 /*
2638  * Issue the commands to select the specified tablespace as the current one
2639  * in the target database.
2640  */
2641 static void
2642 _selectTablespace(ArchiveHandle *AH, const char *tablespace)
2643 {
2644         PQExpBuffer qry;
2645         const char *want,
2646                            *have;
2647
2648         /* do nothing in --no-tablespaces mode */
2649         if (AH->ropt->noTablespace)
2650                 return;
2651
2652         have = AH->currTablespace;
2653         want = tablespace;
2654
2655         /* no need to do anything for non-tablespace object */
2656         if (!want)
2657                 return;
2658
2659         if (have && strcmp(want, have) == 0)
2660                 return;                                 /* no need to do anything */
2661
2662         qry = createPQExpBuffer();
2663
2664         if (strcmp(want, "") == 0)
2665         {
2666                 /* We want the tablespace to be the database's default */
2667                 appendPQExpBuffer(qry, "SET default_tablespace = ''");
2668         }
2669         else
2670         {
2671                 /* We want an explicit tablespace */
2672                 appendPQExpBuffer(qry, "SET default_tablespace = %s", fmtId(want));
2673         }
2674
2675         if (RestoringToDB(AH))
2676         {
2677                 PGresult   *res;
2678
2679                 res = PQexec(AH->connection, qry->data);
2680
2681                 if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
2682                         warn_or_die_horribly(AH, modulename,
2683                                                                  "could not set default_tablespace to %s: %s",
2684                                                                  fmtId(want), PQerrorMessage(AH->connection));
2685
2686                 PQclear(res);
2687         }
2688         else
2689                 ahprintf(AH, "%s;\n\n", qry->data);
2690
2691         if (AH->currTablespace)
2692                 free(AH->currTablespace);
2693         AH->currTablespace = strdup(want);
2694
2695         destroyPQExpBuffer(qry);
2696 }
2697
2698 /*
2699  * Extract an object description for a TOC entry, and append it to buf.
2700  *
2701  * This is not quite as general as it may seem, since it really only
2702  * handles constructing the right thing to put into ALTER ... OWNER TO.
2703  *
2704  * The whole thing is pretty grotty, but we are kind of stuck since the
2705  * information used is all that's available in older dump files.
2706  */
2707 static void
2708 _getObjectDescription(PQExpBuffer buf, TocEntry *te, ArchiveHandle *AH)
2709 {
2710         const char *type = te->desc;
2711
2712         /* Use ALTER TABLE for views and sequences */
2713         if (strcmp(type, "VIEW") == 0 || strcmp(type, "SEQUENCE") == 0)
2714                 type = "TABLE";
2715
2716         /* objects named by a schema and name */
2717         if (strcmp(type, "CONVERSION") == 0 ||
2718                 strcmp(type, "DOMAIN") == 0 ||
2719                 strcmp(type, "TABLE") == 0 ||
2720                 strcmp(type, "TYPE") == 0 ||
2721                 strcmp(type, "TEXT SEARCH DICTIONARY") == 0 ||
2722                 strcmp(type, "TEXT SEARCH CONFIGURATION") == 0)
2723         {
2724                 appendPQExpBuffer(buf, "%s ", type);
2725                 if (te->namespace && te->namespace[0])  /* is null pre-7.3 */
2726                         appendPQExpBuffer(buf, "%s.", fmtId(te->namespace));
2727
2728                 /*
2729                  * Pre-7.3 pg_dump would sometimes (not always) put a fmtId'd name
2730                  * into te->tag for an index. This check is heuristic, so make its
2731                  * scope as narrow as possible.
2732                  */
2733                 if (AH->version < K_VERS_1_7 &&
2734                         te->tag[0] == '"' &&
2735                         te->tag[strlen(te->tag) - 1] == '"' &&
2736                         strcmp(type, "INDEX") == 0)
2737                         appendPQExpBuffer(buf, "%s", te->tag);
2738                 else
2739                         appendPQExpBuffer(buf, "%s", fmtId(te->tag));
2740                 return;
2741         }
2742
2743         /* objects named by just a name */
2744         if (strcmp(type, "DATABASE") == 0 ||
2745                 strcmp(type, "PROCEDURAL LANGUAGE") == 0 ||
2746                 strcmp(type, "SCHEMA") == 0 ||
2747                 strcmp(type, "FOREIGN DATA WRAPPER") == 0 ||
2748                 strcmp(type, "SERVER") == 0 ||
2749                 strcmp(type, "USER MAPPING") == 0)
2750         {
2751                 appendPQExpBuffer(buf, "%s %s", type, fmtId(te->tag));
2752                 return;
2753         }
2754
2755         /* BLOBs just have a name, but it's numeric so must not use fmtId */
2756         if (strcmp(type, "BLOB") == 0)
2757         {
2758                 appendPQExpBuffer(buf, "LARGE OBJECT %s", te->tag);
2759                 return;
2760         }
2761
2762         /*
2763          * These object types require additional decoration.  Fortunately, the
2764          * information needed is exactly what's in the DROP command.
2765          */
2766         if (strcmp(type, "AGGREGATE") == 0 ||
2767                 strcmp(type, "FUNCTION") == 0 ||
2768                 strcmp(type, "OPERATOR") == 0 ||
2769                 strcmp(type, "OPERATOR CLASS") == 0 ||
2770                 strcmp(type, "OPERATOR FAMILY") == 0)
2771         {
2772                 /* Chop "DROP " off the front and make a modifiable copy */
2773                 char       *first = strdup(te->dropStmt + 5);
2774                 char       *last;
2775
2776                 /* point to last character in string */
2777                 last = first + strlen(first) - 1;
2778
2779                 /* Strip off any ';' or '\n' at the end */
2780                 while (last >= first && (*last == '\n' || *last == ';'))
2781                         last--;
2782                 *(last + 1) = '\0';
2783
2784                 appendPQExpBufferStr(buf, first);
2785
2786                 free(first);
2787                 return;
2788         }
2789
2790         write_msg(modulename, "WARNING: don't know how to set owner for object type %s\n",
2791                           type);
2792 }
2793
2794 static void
2795 _printTocEntry(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt, bool isData, bool acl_pass)
2796 {
2797         /* ACLs are dumped only during acl pass */
2798         if (acl_pass)
2799         {
2800                 if (!_tocEntryIsACL(te))
2801                         return;
2802         }
2803         else
2804         {
2805                 if (_tocEntryIsACL(te))
2806                         return;
2807         }
2808
2809         /*
2810          * Avoid dumping the public schema, as it will already be created ...
2811          * unless we are using --clean mode, in which case it's been deleted and
2812          * we'd better recreate it.  Likewise for its comment, if any.
2813          */
2814         if (!ropt->dropSchema)
2815         {
2816                 if (strcmp(te->desc, "SCHEMA") == 0 &&
2817                         strcmp(te->tag, "public") == 0)
2818                         return;
2819                 /* The comment restore would require super-user privs, so avoid it. */
2820                 if (strcmp(te->desc, "COMMENT") == 0 &&
2821                         strcmp(te->tag, "SCHEMA public") == 0)
2822                         return;
2823         }
2824
2825         /* Select owner, schema, and tablespace as necessary */
2826         _becomeOwner(AH, te);
2827         _selectOutputSchema(AH, te->namespace);
2828         _selectTablespace(AH, te->tablespace);
2829
2830         /* Set up OID mode too */
2831         if (strcmp(te->desc, "TABLE") == 0)
2832                 _setWithOids(AH, te);
2833
2834         /* Emit header comment for item */
2835         if (!AH->noTocComments)
2836         {
2837                 const char *pfx;
2838
2839                 if (isData)
2840                         pfx = "Data for ";
2841                 else
2842                         pfx = "";
2843
2844                 ahprintf(AH, "--\n");
2845                 if (AH->public.verbose)
2846                 {
2847                         ahprintf(AH, "-- TOC entry %d (class %u OID %u)\n",
2848                                          te->dumpId, te->catalogId.tableoid, te->catalogId.oid);
2849                         if (te->nDeps > 0)
2850                         {
2851                                 int                     i;
2852
2853                                 ahprintf(AH, "-- Dependencies:");
2854                                 for (i = 0; i < te->nDeps; i++)
2855                                         ahprintf(AH, " %d", te->dependencies[i]);
2856                                 ahprintf(AH, "\n");
2857                         }
2858                 }
2859                 ahprintf(AH, "-- %sName: %s; Type: %s; Schema: %s; Owner: %s",
2860                                  pfx, te->tag, te->desc,
2861                                  te->namespace ? te->namespace : "-",
2862                                  ropt->noOwner ? "-" : te->owner);
2863                 if (te->tablespace && !ropt->noTablespace)
2864                         ahprintf(AH, "; Tablespace: %s", te->tablespace);
2865                 ahprintf(AH, "\n");
2866
2867                 if (AH->PrintExtraTocPtr !=NULL)
2868                         (*AH->PrintExtraTocPtr) (AH, te);
2869                 ahprintf(AH, "--\n\n");
2870         }
2871
2872         /*
2873          * Actually print the definition.
2874          *
2875          * Really crude hack for suppressing AUTHORIZATION clause that old pg_dump
2876          * versions put into CREATE SCHEMA.  We have to do this when --no-owner
2877          * mode is selected.  This is ugly, but I see no other good way ...
2878          */
2879         if (ropt->noOwner && strcmp(te->desc, "SCHEMA") == 0)
2880         {
2881                 ahprintf(AH, "CREATE SCHEMA %s;\n\n\n", fmtId(te->tag));
2882         }
2883         else
2884         {
2885                 if (strlen(te->defn) > 0)
2886                         ahprintf(AH, "%s\n\n", te->defn);
2887         }
2888
2889         /*
2890          * If we aren't using SET SESSION AUTH to determine ownership, we must
2891          * instead issue an ALTER OWNER command.  We assume that anything without
2892          * a DROP command is not a separately ownable object.  All the categories
2893          * with DROP commands must appear in one list or the other.
2894          */
2895         if (!ropt->noOwner && !ropt->use_setsessauth &&
2896                 strlen(te->owner) > 0 && strlen(te->dropStmt) > 0)
2897         {
2898                 if (strcmp(te->desc, "AGGREGATE") == 0 ||
2899                         strcmp(te->desc, "BLOB") == 0 ||
2900                         strcmp(te->desc, "CONVERSION") == 0 ||
2901                         strcmp(te->desc, "DATABASE") == 0 ||
2902                         strcmp(te->desc, "DOMAIN") == 0 ||
2903                         strcmp(te->desc, "FUNCTION") == 0 ||
2904                         strcmp(te->desc, "OPERATOR") == 0 ||
2905                         strcmp(te->desc, "OPERATOR CLASS") == 0 ||
2906                         strcmp(te->desc, "OPERATOR FAMILY") == 0 ||
2907                         strcmp(te->desc, "PROCEDURAL LANGUAGE") == 0 ||
2908                         strcmp(te->desc, "SCHEMA") == 0 ||
2909                         strcmp(te->desc, "TABLE") == 0 ||
2910                         strcmp(te->desc, "TYPE") == 0 ||
2911                         strcmp(te->desc, "VIEW") == 0 ||
2912                         strcmp(te->desc, "SEQUENCE") == 0 ||
2913                         strcmp(te->desc, "TEXT SEARCH DICTIONARY") == 0 ||
2914                         strcmp(te->desc, "TEXT SEARCH CONFIGURATION") == 0 ||
2915                         strcmp(te->desc, "FOREIGN DATA WRAPPER") == 0 ||
2916                         strcmp(te->desc, "SERVER") == 0)
2917                 {
2918                         PQExpBuffer temp = createPQExpBuffer();
2919
2920                         appendPQExpBuffer(temp, "ALTER ");
2921                         _getObjectDescription(temp, te, AH);
2922                         appendPQExpBuffer(temp, " OWNER TO %s;", fmtId(te->owner));
2923                         ahprintf(AH, "%s\n\n", temp->data);
2924                         destroyPQExpBuffer(temp);
2925                 }
2926                 else if (strcmp(te->desc, "CAST") == 0 ||
2927                                  strcmp(te->desc, "CHECK CONSTRAINT") == 0 ||
2928                                  strcmp(te->desc, "CONSTRAINT") == 0 ||
2929                                  strcmp(te->desc, "DEFAULT") == 0 ||
2930                                  strcmp(te->desc, "FK CONSTRAINT") == 0 ||
2931                                  strcmp(te->desc, "INDEX") == 0 ||
2932                                  strcmp(te->desc, "RULE") == 0 ||
2933                                  strcmp(te->desc, "TRIGGER") == 0 ||
2934                                  strcmp(te->desc, "USER MAPPING") == 0)
2935                 {
2936                         /* these object types don't have separate owners */
2937                 }
2938                 else
2939                 {
2940                         write_msg(modulename, "WARNING: don't know how to set owner for object type %s\n",
2941                                           te->desc);
2942                 }
2943         }
2944
2945         /*
2946          * If it's an ACL entry, it might contain SET SESSION AUTHORIZATION
2947          * commands, so we can no longer assume we know the current auth setting.
2948          */
2949         if (acl_pass)
2950         {
2951                 if (AH->currUser)
2952                         free(AH->currUser);
2953                 AH->currUser = NULL;
2954         }
2955 }
2956
2957 void
2958 WriteHead(ArchiveHandle *AH)
2959 {
2960         struct tm       crtm;
2961
2962         (*AH->WriteBufPtr) (AH, "PGDMP", 5);            /* Magic code */
2963         (*AH->WriteBytePtr) (AH, AH->vmaj);
2964         (*AH->WriteBytePtr) (AH, AH->vmin);
2965         (*AH->WriteBytePtr) (AH, AH->vrev);
2966         (*AH->WriteBytePtr) (AH, AH->intSize);
2967         (*AH->WriteBytePtr) (AH, AH->offSize);
2968         (*AH->WriteBytePtr) (AH, AH->format);
2969
2970 #ifndef HAVE_LIBZ
2971         if (AH->compression != 0)
2972                 write_msg(modulename, "WARNING: requested compression not available in this "
2973                                   "installation -- archive will be uncompressed\n");
2974
2975         AH->compression = 0;
2976 #endif
2977
2978         WriteInt(AH, AH->compression);
2979
2980         crtm = *localtime(&AH->createDate);
2981         WriteInt(AH, crtm.tm_sec);
2982         WriteInt(AH, crtm.tm_min);
2983         WriteInt(AH, crtm.tm_hour);
2984         WriteInt(AH, crtm.tm_mday);
2985         WriteInt(AH, crtm.tm_mon);
2986         WriteInt(AH, crtm.tm_year);
2987         WriteInt(AH, crtm.tm_isdst);
2988         WriteStr(AH, PQdb(AH->connection));
2989         WriteStr(AH, AH->public.remoteVersionStr);
2990         WriteStr(AH, PG_VERSION);
2991 }
2992
2993 void
2994 ReadHead(ArchiveHandle *AH)
2995 {
2996         char            tmpMag[7];
2997         int                     fmt;
2998         struct tm       crtm;
2999
3000         /*
3001          * If we haven't already read the header, do so.
3002          *
3003          * NB: this code must agree with _discoverArchiveFormat().      Maybe find a
3004          * way to unify the cases?
3005          */
3006         if (!AH->readHeader)
3007         {
3008                 if ((*AH->ReadBufPtr) (AH, tmpMag, 5) != 5)
3009                         die_horribly(AH, modulename, "unexpected end of file\n");
3010
3011                 if (strncmp(tmpMag, "PGDMP", 5) != 0)
3012                         die_horribly(AH, modulename, "did not find magic string in file header\n");
3013
3014                 AH->vmaj = (*AH->ReadBytePtr) (AH);
3015                 AH->vmin = (*AH->ReadBytePtr) (AH);
3016
3017                 if (AH->vmaj > 1 || ((AH->vmaj == 1) && (AH->vmin > 0)))                /* Version > 1.0 */
3018                         AH->vrev = (*AH->ReadBytePtr) (AH);
3019                 else
3020                         AH->vrev = 0;
3021
3022                 AH->version = ((AH->vmaj * 256 + AH->vmin) * 256 + AH->vrev) * 256 + 0;
3023
3024                 if (AH->version < K_VERS_1_0 || AH->version > K_VERS_MAX)
3025                         die_horribly(AH, modulename, "unsupported version (%d.%d) in file header\n",
3026                                                  AH->vmaj, AH->vmin);
3027
3028                 AH->intSize = (*AH->ReadBytePtr) (AH);
3029                 if (AH->intSize > 32)
3030                         die_horribly(AH, modulename, "sanity check on integer size (%lu) failed\n",
3031                                                  (unsigned long) AH->intSize);
3032
3033                 if (AH->intSize > sizeof(int))
3034                         write_msg(modulename, "WARNING: archive was made on a machine with larger integers, some operations might fail\n");
3035
3036                 if (AH->version >= K_VERS_1_7)
3037                         AH->offSize = (*AH->ReadBytePtr) (AH);
3038                 else
3039                         AH->offSize = AH->intSize;
3040
3041                 fmt = (*AH->ReadBytePtr) (AH);
3042
3043                 if (AH->format != fmt)
3044                         die_horribly(AH, modulename, "expected format (%d) differs from format found in file (%d)\n",
3045                                                  AH->format, fmt);
3046         }
3047
3048         if (AH->version >= K_VERS_1_2)
3049         {
3050                 if (AH->version < K_VERS_1_4)
3051                         AH->compression = (*AH->ReadBytePtr) (AH);
3052                 else
3053                         AH->compression = ReadInt(AH);
3054         }
3055         else
3056                 AH->compression = Z_DEFAULT_COMPRESSION;
3057
3058 #ifndef HAVE_LIBZ
3059         if (AH->compression != 0)
3060                 write_msg(modulename, "WARNING: archive is compressed, but this installation does not support compression -- no data will be available\n");
3061 #endif
3062
3063         if (AH->version >= K_VERS_1_4)
3064         {
3065                 crtm.tm_sec = ReadInt(AH);
3066                 crtm.tm_min = ReadInt(AH);
3067                 crtm.tm_hour = ReadInt(AH);
3068                 crtm.tm_mday = ReadInt(AH);
3069                 crtm.tm_mon = ReadInt(AH);
3070                 crtm.tm_year = ReadInt(AH);
3071                 crtm.tm_isdst = ReadInt(AH);
3072
3073                 AH->archdbname = ReadStr(AH);
3074
3075                 AH->createDate = mktime(&crtm);
3076
3077                 if (AH->createDate == (time_t) -1)
3078                         write_msg(modulename, "WARNING: invalid creation date in header\n");
3079         }
3080
3081         if (AH->version >= K_VERS_1_10)
3082         {
3083                 AH->archiveRemoteVersion = ReadStr(AH);
3084                 AH->archiveDumpVersion = ReadStr(AH);
3085         }
3086 }
3087
3088
3089 /*
3090  * checkSeek
3091  *        check to see if ftell/fseek can be performed.
3092  */
3093 bool
3094 checkSeek(FILE *fp)
3095 {
3096         pgoff_t         tpos;
3097
3098         /*
3099          * If pgoff_t is wider than long, we must have "real" fseeko and not an
3100          * emulation using fseek.  Otherwise report no seek capability.
3101          */
3102 #ifndef HAVE_FSEEKO
3103         if (sizeof(pgoff_t) > sizeof(long))
3104                 return false;
3105 #endif
3106
3107         /* Check that ftello works on this file */
3108         errno = 0;
3109         tpos = ftello(fp);
3110         if (errno)
3111                 return false;
3112
3113         /*
3114          * Check that fseeko(SEEK_SET) works, too.      NB: we used to try to test
3115          * this with fseeko(fp, 0, SEEK_CUR).  But some platforms treat that as a
3116          * successful no-op even on files that are otherwise unseekable.
3117          */
3118         if (fseeko(fp, tpos, SEEK_SET) != 0)
3119                 return false;
3120
3121         return true;
3122 }
3123
3124
3125 /*
3126  * dumpTimestamp
3127  */
3128 static void
3129 dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim)
3130 {
3131         char            buf[256];
3132
3133         /*
3134          * We don't print the timezone on Win32, because the names are long and
3135          * localized, which means they may contain characters in various random
3136          * encodings; this has been seen to cause encoding errors when reading the
3137          * dump script.
3138          */
3139         if (strftime(buf, sizeof(buf),
3140 #ifndef WIN32
3141                                  "%Y-%m-%d %H:%M:%S %Z",
3142 #else
3143                                  "%Y-%m-%d %H:%M:%S",
3144 #endif
3145                                  localtime(&tim)) != 0)
3146                 ahprintf(AH, "-- %s %s\n\n", msg, buf);
3147 }
3148
3149
3150 /*
3151  * Main engine for parallel restore.
3152  *
3153  * Work is done in three phases.
3154  * First we process tocEntries until we come to one that is marked
3155  * SECTION_DATA or SECTION_POST_DATA, in a single connection, just as for a
3156  * standard restore.  Second we process the remaining non-ACL steps in
3157  * parallel worker children (threads on Windows, processes on Unix), each of
3158  * which connects separately to the database.  Finally we process all the ACL
3159  * entries in a single connection (that happens back in RestoreArchive).
3160  */
3161 static void
3162 restore_toc_entries_parallel(ArchiveHandle *AH)
3163 {
3164         RestoreOptions *ropt = AH->ropt;
3165         int                     n_slots = ropt->number_of_jobs;
3166         ParallelSlot *slots;
3167         int                     work_status;
3168         int                     next_slot;
3169         TocEntry        pending_list;
3170         TocEntry        ready_list;
3171         TocEntry   *next_work_item;
3172         thandle         ret_child;
3173         TocEntry   *te;
3174
3175         ahlog(AH, 2, "entering restore_toc_entries_parallel\n");
3176
3177         /* we haven't got round to making this work for all archive formats */
3178         if (AH->ClonePtr == NULL || AH->ReopenPtr == NULL)
3179                 die_horribly(AH, modulename, "parallel restore is not supported with this archive file format\n");
3180
3181         /* doesn't work if the archive represents dependencies as OIDs, either */
3182         if (AH->version < K_VERS_1_8)
3183                 die_horribly(AH, modulename, "parallel restore is not supported with archives made by pre-8.0 pg_dump\n");
3184
3185         slots = (ParallelSlot *) calloc(sizeof(ParallelSlot), n_slots);
3186
3187         /* Adjust dependency information */
3188         fix_dependencies(AH);
3189
3190         /*
3191          * Do all the early stuff in a single connection in the parent. There's no
3192          * great point in running it in parallel, in fact it will actually run
3193          * faster in a single connection because we avoid all the connection and
3194          * setup overhead.  Also, pg_dump is not currently very good about
3195          * showing all the dependencies of SECTION_PRE_DATA items, so we do not
3196          * risk trying to process them out-of-order.
3197          */
3198         for (next_work_item = AH->toc->next; next_work_item != AH->toc; next_work_item = next_work_item->next)
3199         {
3200                 /* Non-PRE_DATA items are just ignored for now */
3201                 if (next_work_item->section == SECTION_DATA ||
3202                         next_work_item->section == SECTION_POST_DATA)
3203                         continue;
3204
3205                 ahlog(AH, 1, "processing item %d %s %s\n",
3206                           next_work_item->dumpId,
3207                           next_work_item->desc, next_work_item->tag);
3208
3209                 (void) restore_toc_entry(AH, next_work_item, ropt, false);
3210
3211                 /* there should be no touch of ready_list here, so pass NULL */
3212                 reduce_dependencies(AH, next_work_item, NULL);
3213         }
3214
3215         /*
3216          * Now close parent connection in prep for parallel steps.      We do this
3217          * mainly to ensure that we don't exceed the specified number of parallel
3218          * connections.
3219          */
3220         PQfinish(AH->connection);
3221         AH->connection = NULL;
3222
3223         /* blow away any transient state from the old connection */
3224         if (AH->currUser)
3225                 free(AH->currUser);
3226         AH->currUser = NULL;
3227         if (AH->currSchema)
3228                 free(AH->currSchema);
3229         AH->currSchema = NULL;
3230         if (AH->currTablespace)
3231                 free(AH->currTablespace);
3232         AH->currTablespace = NULL;
3233         AH->currWithOids = -1;
3234
3235         /*
3236          * Initialize the lists of pending and ready items.  After this setup, the
3237          * pending list is everything that needs to be done but is blocked by one
3238          * or more dependencies, while the ready list contains items that have no
3239          * remaining dependencies.      Note: we don't yet filter out entries that
3240          * aren't going to be restored.  They might participate in dependency
3241          * chains connecting entries that should be restored, so we treat them as
3242          * live until we actually process them.
3243          */
3244         par_list_header_init(&pending_list);
3245         par_list_header_init(&ready_list);
3246         for (next_work_item = AH->toc->next; next_work_item != AH->toc; next_work_item = next_work_item->next)
3247         {
3248                 /* All PRE_DATA items were dealt with above */
3249                 if (next_work_item->section == SECTION_DATA ||
3250                         next_work_item->section == SECTION_POST_DATA)
3251                 {
3252                         if (next_work_item->depCount > 0)
3253                                 par_list_append(&pending_list, next_work_item);
3254                         else
3255                                 par_list_append(&ready_list, next_work_item);
3256                 }
3257         }
3258
3259         /*
3260          * main parent loop
3261          *
3262          * Keep going until there is no worker still running AND there is no work
3263          * left to be done.
3264          */
3265
3266         ahlog(AH, 1, "entering main parallel loop\n");
3267
3268         while ((next_work_item = get_next_work_item(AH, &ready_list,
3269                                                                                                 slots, n_slots)) != NULL ||
3270                    work_in_progress(slots, n_slots))
3271         {
3272                 if (next_work_item != NULL)
3273                 {
3274                         teReqs          reqs;
3275
3276                         /* If not to be dumped, don't waste time launching a worker */
3277                         reqs = _tocEntryRequired(next_work_item, AH->ropt, false);
3278                         if ((reqs & (REQ_SCHEMA | REQ_DATA)) == 0)
3279                         {
3280                                 ahlog(AH, 1, "skipping item %d %s %s\n",
3281                                           next_work_item->dumpId,
3282                                           next_work_item->desc, next_work_item->tag);
3283
3284                                 par_list_remove(next_work_item);
3285                                 reduce_dependencies(AH, next_work_item, &ready_list);
3286
3287                                 continue;
3288                         }
3289
3290                         if ((next_slot = get_next_slot(slots, n_slots)) != NO_SLOT)
3291                         {
3292                                 /* There is work still to do and a worker slot available */
3293                                 thandle         child;
3294                                 RestoreArgs *args;
3295
3296                                 ahlog(AH, 1, "launching item %d %s %s\n",
3297                                           next_work_item->dumpId,
3298                                           next_work_item->desc, next_work_item->tag);
3299
3300                                 par_list_remove(next_work_item);
3301
3302                                 /* this memory is dealloced in mark_work_done() */
3303                                 args = malloc(sizeof(RestoreArgs));
3304                                 args->AH = CloneArchive(AH);
3305                                 args->te = next_work_item;
3306
3307                                 /* run the step in a worker child */
3308                                 child = spawn_restore(args);
3309
3310                                 slots[next_slot].child_id = child;
3311                                 slots[next_slot].args = args;
3312
3313                                 continue;
3314                         }
3315                 }
3316
3317                 /*
3318                  * If we get here there must be work being done.  Either there is no
3319                  * work available to schedule (and work_in_progress returned true) or
3320                  * there are no slots available.  So we wait for a worker to finish,
3321                  * and process the result.
3322                  */
3323                 ret_child = reap_child(slots, n_slots, &work_status);
3324
3325                 if (WIFEXITED(work_status))
3326                 {
3327                         mark_work_done(AH, &ready_list,
3328                                                    ret_child, WEXITSTATUS(work_status),
3329                                                    slots, n_slots);
3330                 }
3331                 else
3332                 {
3333                         die_horribly(AH, modulename, "worker process crashed: status %d\n",
3334                                                  work_status);
3335                 }
3336         }
3337
3338         ahlog(AH, 1, "finished main parallel loop\n");
3339
3340         /*
3341          * Now reconnect the single parent connection.
3342          */
3343         ConnectDatabase((Archive *) AH, ropt->dbname,
3344                                         ropt->pghost, ropt->pgport, ropt->username,
3345                                         ropt->promptPassword);
3346
3347         _doSetFixedOutputState(AH);
3348
3349         /*
3350          * Make sure there is no non-ACL work left due to, say, circular
3351          * dependencies, or some other pathological condition. If so, do it in the
3352          * single parent connection.
3353          */
3354         for (te = pending_list.par_next; te != &pending_list; te = te->par_next)
3355         {
3356                 ahlog(AH, 1, "processing missed item %d %s %s\n",
3357                           te->dumpId, te->desc, te->tag);
3358                 (void) restore_toc_entry(AH, te, ropt, false);
3359         }
3360
3361         /* The ACLs will be handled back in RestoreArchive. */
3362 }
3363
3364 /*
3365  * create a worker child to perform a restore step in parallel
3366  */
3367 static thandle
3368 spawn_restore(RestoreArgs *args)
3369 {
3370         thandle         child;
3371
3372         /* Ensure stdio state is quiesced before forking */
3373         fflush(NULL);
3374
3375 #ifndef WIN32
3376         child = fork();
3377         if (child == 0)
3378         {
3379                 /* in child process */
3380                 parallel_restore(args);
3381                 die_horribly(args->AH, modulename,
3382                                          "parallel_restore should not return\n");
3383         }
3384         else if (child < 0)
3385         {
3386                 /* fork failed */
3387                 die_horribly(args->AH, modulename,
3388                                          "could not create worker process: %s\n",
3389                                          strerror(errno));
3390         }
3391 #else
3392         child = (HANDLE) _beginthreadex(NULL, 0, (void *) parallel_restore,
3393                                                                         args, 0, NULL);
3394         if (child == 0)
3395                 die_horribly(args->AH, modulename,
3396                                          "could not create worker thread: %s\n",
3397                                          strerror(errno));
3398 #endif
3399
3400         return child;
3401 }
3402
3403 /*
3404  *      collect status from a completed worker child
3405  */
3406 static thandle
3407 reap_child(ParallelSlot *slots, int n_slots, int *work_status)
3408 {
3409 #ifndef WIN32
3410         /* Unix is so much easier ... */
3411         return wait(work_status);
3412 #else
3413         static HANDLE *handles = NULL;
3414         int                     hindex,
3415                                 snum,
3416                                 tnum;
3417         thandle         ret_child;
3418         DWORD           res;
3419
3420         /* first time around only, make space for handles to listen on */
3421         if (handles == NULL)
3422                 handles = (HANDLE *) calloc(sizeof(HANDLE), n_slots);
3423
3424         /* set up list of handles to listen to */
3425         for (snum = 0, tnum = 0; snum < n_slots; snum++)
3426                 if (slots[snum].child_id != 0)
3427                         handles[tnum++] = slots[snum].child_id;
3428
3429         /* wait for one to finish */
3430         hindex = WaitForMultipleObjects(tnum, handles, false, INFINITE);
3431
3432         /* get handle of finished thread */
3433         ret_child = handles[hindex - WAIT_OBJECT_0];
3434
3435         /* get the result */
3436         GetExitCodeThread(ret_child, &res);
3437         *work_status = res;
3438
3439         /* dispose of handle to stop leaks */
3440         CloseHandle(ret_child);
3441
3442         return ret_child;
3443 #endif
3444 }
3445
3446 /*
3447  * are we doing anything now?
3448  */
3449 static bool
3450 work_in_progress(ParallelSlot *slots, int n_slots)
3451 {
3452         int                     i;
3453
3454         for (i = 0; i < n_slots; i++)
3455         {
3456                 if (slots[i].child_id != 0)
3457                         return true;
3458         }
3459         return false;
3460 }
3461
3462 /*
3463  * find the first free parallel slot (if any).
3464  */
3465 static int
3466 get_next_slot(ParallelSlot *slots, int n_slots)
3467 {
3468         int                     i;
3469
3470         for (i = 0; i < n_slots; i++)
3471         {
3472                 if (slots[i].child_id == 0)
3473                         return i;
3474         }
3475         return NO_SLOT;
3476 }
3477
3478
3479 /*
3480  * Check if te1 has an exclusive lock requirement for an item that te2 also
3481  * requires, whether or not te2's requirement is for an exclusive lock.
3482  */
3483 static bool
3484 has_lock_conflicts(TocEntry *te1, TocEntry *te2)
3485 {
3486         int                     j,
3487                                 k;
3488
3489         for (j = 0; j < te1->nLockDeps; j++)
3490         {
3491                 for (k = 0; k < te2->nDeps; k++)
3492                 {
3493                         if (te1->lockDeps[j] == te2->dependencies[k])
3494                                 return true;
3495                 }
3496         }
3497         return false;
3498 }
3499
3500
3501 /*
3502  * Initialize the header of a parallel-processing list.
3503  *
3504  * These are circular lists with a dummy TocEntry as header, just like the
3505  * main TOC list; but we use separate list links so that an entry can be in
3506  * the main TOC list as well as in a parallel-processing list.
3507  */
3508 static void
3509 par_list_header_init(TocEntry *l)
3510 {
3511         l->par_prev = l->par_next = l;
3512 }
3513
3514 /* Append te to the end of the parallel-processing list headed by l */
3515 static void
3516 par_list_append(TocEntry *l, TocEntry *te)
3517 {
3518         te->par_prev = l->par_prev;
3519         l->par_prev->par_next = te;
3520         l->par_prev = te;
3521         te->par_next = l;
3522 }
3523
3524 /* Remove te from whatever parallel-processing list it's in */
3525 static void
3526 par_list_remove(TocEntry *te)
3527 {
3528         te->par_prev->par_next = te->par_next;
3529         te->par_next->par_prev = te->par_prev;
3530         te->par_prev = NULL;
3531         te->par_next = NULL;
3532 }
3533
3534
3535 /*
3536  * Find the next work item (if any) that is capable of being run now.
3537  *
3538  * To qualify, the item must have no remaining dependencies
3539  * and no requirements for locks that are incompatible with
3540  * items currently running.  Items in the ready_list are known to have
3541  * no remaining dependencies, but we have to check for lock conflicts.
3542  *
3543  * Note that the returned item has *not* been removed from ready_list.
3544  * The caller must do that after successfully dispatching the item.
3545  *
3546  * pref_non_data is for an alternative selection algorithm that gives
3547  * preference to non-data items if there is already a data load running.
3548  * It is currently disabled.
3549  */
3550 static TocEntry *
3551 get_next_work_item(ArchiveHandle *AH, TocEntry *ready_list,
3552                                    ParallelSlot *slots, int n_slots)
3553 {
3554         bool            pref_non_data = false;  /* or get from AH->ropt */
3555         TocEntry   *data_te = NULL;
3556         TocEntry   *te;
3557         int                     i,
3558                                 k;
3559
3560         /*
3561          * Bogus heuristics for pref_non_data
3562          */
3563         if (pref_non_data)
3564         {
3565                 int                     count = 0;
3566
3567                 for (k = 0; k < n_slots; k++)
3568                         if (slots[k].args->te != NULL &&
3569                                 slots[k].args->te->section == SECTION_DATA)
3570                                 count++;
3571                 if (n_slots == 0 || count * 4 < n_slots)
3572                         pref_non_data = false;
3573         }
3574
3575         /*
3576          * Search the ready_list until we find a suitable item.
3577          */
3578         for (te = ready_list->par_next; te != ready_list; te = te->par_next)
3579         {
3580                 bool            conflicts = false;
3581
3582                 /*
3583                  * Check to see if the item would need exclusive lock on something
3584                  * that a currently running item also needs lock on, or vice versa. If
3585                  * so, we don't want to schedule them together.
3586                  */
3587                 for (i = 0; i < n_slots && !conflicts; i++)
3588                 {
3589                         TocEntry   *running_te;
3590
3591                         if (slots[i].args == NULL)
3592                                 continue;
3593                         running_te = slots[i].args->te;
3594
3595                         if (has_lock_conflicts(te, running_te) ||
3596                                 has_lock_conflicts(running_te, te))
3597                         {
3598                                 conflicts = true;
3599                                 break;
3600                         }
3601                 }
3602
3603                 if (conflicts)
3604                         continue;
3605
3606                 if (pref_non_data && te->section == SECTION_DATA)
3607                 {
3608                         if (data_te == NULL)
3609                                 data_te = te;
3610                         continue;
3611                 }
3612
3613                 /* passed all tests, so this item can run */
3614                 return te;
3615         }
3616
3617         if (data_te != NULL)
3618                 return data_te;
3619
3620         ahlog(AH, 2, "no item ready\n");
3621         return NULL;
3622 }
3623
3624
3625 /*
3626  * Restore a single TOC item in parallel with others
3627  *
3628  * this is the procedure run as a thread (Windows) or a
3629  * separate process (everything else).
3630  */
3631 static parallel_restore_result
3632 parallel_restore(RestoreArgs *args)
3633 {
3634         ArchiveHandle *AH = args->AH;
3635         TocEntry   *te = args->te;
3636         RestoreOptions *ropt = AH->ropt;
3637         int                     retval;
3638
3639         /*
3640          * Close and reopen the input file so we have a private file pointer that
3641          * doesn't stomp on anyone else's file pointer, if we're actually going to
3642          * need to read from the file. Otherwise, just close it except on Windows,
3643          * where it will possibly be needed by other threads.
3644          *
3645          * Note: on Windows, since we are using threads not processes, the reopen
3646          * call *doesn't* close the original file pointer but just open a new one.
3647          */
3648         if (te->section == SECTION_DATA)
3649                 (AH->ReopenPtr) (AH);
3650 #ifndef WIN32
3651         else
3652                 (AH->ClosePtr) (AH);
3653 #endif
3654
3655         /*
3656          * We need our own database connection, too
3657          */
3658         ConnectDatabase((Archive *) AH, ropt->dbname,
3659                                         ropt->pghost, ropt->pgport, ropt->username,
3660                                         ropt->promptPassword);
3661
3662         _doSetFixedOutputState(AH);
3663
3664         /* Restore the TOC item */
3665         retval = restore_toc_entry(AH, te, ropt, true);
3666
3667         /* And clean up */
3668         PQfinish(AH->connection);
3669         AH->connection = NULL;
3670
3671         /* If we reopened the file, we are done with it, so close it now */
3672         if (te->section == SECTION_DATA)
3673                 (AH->ClosePtr) (AH);
3674
3675         if (retval == 0 && AH->public.n_errors)
3676                 retval = WORKER_IGNORED_ERRORS;
3677
3678 #ifndef WIN32
3679         exit(retval);
3680 #else
3681         return retval;
3682 #endif
3683 }
3684
3685
3686 /*
3687  * Housekeeping to be done after a step has been parallel restored.
3688  *
3689  * Clear the appropriate slot, free all the extra memory we allocated,
3690  * update status, and reduce the dependency count of any dependent items.
3691  */
3692 static void
3693 mark_work_done(ArchiveHandle *AH, TocEntry *ready_list,
3694                            thandle worker, int status,
3695                            ParallelSlot *slots, int n_slots)
3696 {
3697         TocEntry   *te = NULL;
3698         int                     i;
3699
3700         for (i = 0; i < n_slots; i++)
3701         {
3702                 if (slots[i].child_id == worker)
3703                 {
3704                         slots[i].child_id = 0;
3705                         te = slots[i].args->te;
3706                         DeCloneArchive(slots[i].args->AH);
3707                         free(slots[i].args);
3708                         slots[i].args = NULL;
3709
3710                         break;
3711                 }
3712         }
3713
3714         if (te == NULL)
3715                 die_horribly(AH, modulename, "could not find slot of finished worker\n");
3716
3717         ahlog(AH, 1, "finished item %d %s %s\n",
3718                   te->dumpId, te->desc, te->tag);
3719
3720         if (status == WORKER_CREATE_DONE)
3721                 mark_create_done(AH, te);
3722         else if (status == WORKER_INHIBIT_DATA)
3723         {
3724                 inhibit_data_for_failed_table(AH, te);
3725                 AH->public.n_errors++;
3726         }
3727         else if (status == WORKER_IGNORED_ERRORS)
3728                 AH->public.n_errors++;
3729         else if (status != 0)
3730                 die_horribly(AH, modulename, "worker process failed: exit code %d\n",
3731                                          status);
3732
3733         reduce_dependencies(AH, te, ready_list);
3734 }
3735
3736
3737 /*
3738  * Process the dependency information into a form useful for parallel restore.
3739  *
3740  * We set up depCount fields that are the number of as-yet-unprocessed
3741  * dependencies for each TOC entry.
3742  *
3743  * We also identify locking dependencies so that we can avoid trying to
3744  * schedule conflicting items at the same time.
3745  */
3746 static void
3747 fix_dependencies(ArchiveHandle *AH)
3748 {
3749         TocEntry  **tocsByDumpId;
3750         TocEntry   *te;
3751         DumpId          maxDumpId;
3752         int                     i;
3753
3754         /*
3755          * For some of the steps here, it is convenient to have an array that
3756          * indexes the TOC entries by dump ID, rather than searching the TOC list
3757          * repeatedly.  Entries for dump IDs not present in the TOC will be NULL.
3758          *
3759          * NOTE: because maxDumpId is just the highest dump ID defined in the
3760          * archive, there might be dependencies for IDs > maxDumpId.  All uses of
3761          * this array must guard against out-of-range dependency numbers.
3762          *
3763          * Also, initialize the depCount fields, and make sure all the TOC items
3764          * are marked as not being in any parallel-processing list.
3765          */
3766         maxDumpId = AH->maxDumpId;
3767         tocsByDumpId = (TocEntry **) calloc(maxDumpId, sizeof(TocEntry *));
3768         for (te = AH->toc->next; te != AH->toc; te = te->next)
3769         {
3770                 tocsByDumpId[te->dumpId - 1] = te;
3771                 te->depCount = te->nDeps;
3772                 te->par_prev = NULL;
3773                 te->par_next = NULL;
3774         }
3775
3776         /*
3777          * POST_DATA items that are shown as depending on a table need to be
3778          * re-pointed to depend on that table's data, instead.  This ensures they
3779          * won't get scheduled until the data has been loaded.  We handle this by
3780          * first finding TABLE/TABLE DATA pairs and then scanning all the
3781          * dependencies.
3782          *
3783          * Note: currently, a TABLE DATA should always have exactly one
3784          * dependency, on its TABLE item.  So we don't bother to search, but look
3785          * just at the first dependency.  We do trouble to make sure that it's a
3786          * TABLE, if possible.  However, if the dependency isn't in the archive
3787          * then just assume it was a TABLE; this is to cover cases where the table
3788          * was suppressed but we have the data and some dependent post-data items.
3789          */
3790         for (te = AH->toc->next; te != AH->toc; te = te->next)
3791         {
3792                 if (strcmp(te->desc, "TABLE DATA") == 0 && te->nDeps > 0)
3793                 {
3794                         DumpId          tableId = te->dependencies[0];
3795
3796                         if (tableId > maxDumpId ||
3797                                 tocsByDumpId[tableId - 1] == NULL ||
3798                                 strcmp(tocsByDumpId[tableId - 1]->desc, "TABLE") == 0)
3799                         {
3800                                 repoint_table_dependencies(AH, tableId, te->dumpId);
3801                         }
3802                 }
3803         }
3804
3805         /*
3806          * Pre-8.4 versions of pg_dump neglected to set up a dependency from BLOB
3807          * COMMENTS to BLOBS.  Cope.  (We assume there's only one BLOBS and only
3808          * one BLOB COMMENTS in such files.)
3809          */
3810         if (AH->version < K_VERS_1_11)
3811         {
3812                 for (te = AH->toc->next; te != AH->toc; te = te->next)
3813                 {
3814                         if (strcmp(te->desc, "BLOB COMMENTS") == 0 && te->nDeps == 0)
3815                         {
3816                                 TocEntry   *te2;
3817
3818                                 for (te2 = AH->toc->next; te2 != AH->toc; te2 = te2->next)
3819                                 {
3820                                         if (strcmp(te2->desc, "BLOBS") == 0)
3821                                         {
3822                                                 te->dependencies = (DumpId *) malloc(sizeof(DumpId));
3823                                                 te->dependencies[0] = te2->dumpId;
3824                                                 te->nDeps++;
3825                                                 te->depCount++;
3826                                                 break;
3827                                         }
3828                                 }
3829                                 break;
3830                         }
3831                 }
3832         }
3833
3834         /*
3835          * It is possible that the dependencies list items that are not in the
3836          * archive at all.      Subtract such items from the depCounts.
3837          */
3838         for (te = AH->toc->next; te != AH->toc; te = te->next)
3839         {
3840                 for (i = 0; i < te->nDeps; i++)
3841                 {
3842                         DumpId          depid = te->dependencies[i];
3843
3844                         if (depid > maxDumpId || tocsByDumpId[depid - 1] == NULL)
3845                                 te->depCount--;
3846                 }
3847         }
3848
3849         /*
3850          * Lastly, work out the locking dependencies.
3851          */
3852         for (te = AH->toc->next; te != AH->toc; te = te->next)
3853         {
3854                 te->lockDeps = NULL;
3855                 te->nLockDeps = 0;
3856                 identify_locking_dependencies(te, tocsByDumpId, maxDumpId);
3857         }
3858
3859         free(tocsByDumpId);
3860 }
3861
3862 /*
3863  * Change dependencies on tableId to depend on tableDataId instead,
3864  * but only in POST_DATA items.
3865  */
3866 static void
3867 repoint_table_dependencies(ArchiveHandle *AH,
3868                                                    DumpId tableId, DumpId tableDataId)
3869 {
3870         TocEntry   *te;
3871         int                     i;
3872
3873         for (te = AH->toc->next; te != AH->toc; te = te->next)
3874         {
3875                 if (te->section != SECTION_POST_DATA)
3876                         continue;
3877                 for (i = 0; i < te->nDeps; i++)
3878                 {
3879                         if (te->dependencies[i] == tableId)
3880                         {
3881                                 te->dependencies[i] = tableDataId;
3882                                 ahlog(AH, 2, "transferring dependency %d -> %d to %d\n",
3883                                           te->dumpId, tableId, tableDataId);
3884                         }
3885                 }
3886         }
3887 }
3888
3889 /*
3890  * Identify which objects we'll need exclusive lock on in order to restore
3891  * the given TOC entry (*other* than the one identified by the TOC entry
3892  * itself).  Record their dump IDs in the entry's lockDeps[] array.
3893  * tocsByDumpId[] is a convenience array (of size maxDumpId) to avoid
3894  * searching the TOC for each dependency.
3895  */
3896 static void
3897 identify_locking_dependencies(TocEntry *te,
3898                                                           TocEntry **tocsByDumpId,
3899                                                           DumpId maxDumpId)
3900 {
3901         DumpId     *lockids;
3902         int                     nlockids;
3903         int                     i;
3904
3905         /* Quick exit if no dependencies at all */
3906         if (te->nDeps == 0)
3907                 return;
3908
3909         /* Exit if this entry doesn't need exclusive lock on other objects */
3910         if (!(strcmp(te->desc, "CONSTRAINT") == 0 ||
3911                   strcmp(te->desc, "CHECK CONSTRAINT") == 0 ||
3912                   strcmp(te->desc, "FK CONSTRAINT") == 0 ||
3913                   strcmp(te->desc, "RULE") == 0 ||
3914                   strcmp(te->desc, "TRIGGER") == 0))
3915                 return;
3916
3917         /*
3918          * We assume the item requires exclusive lock on each TABLE DATA item
3919          * listed among its dependencies.  (This was originally a dependency on
3920          * the TABLE, but fix_dependencies repointed it to the data item. Note
3921          * that all the entry types we are interested in here are POST_DATA, so
3922          * they will all have been changed this way.)
3923          */
3924         lockids = (DumpId *) malloc(te->nDeps * sizeof(DumpId));
3925         nlockids = 0;
3926         for (i = 0; i < te->nDeps; i++)
3927         {
3928                 DumpId          depid = te->dependencies[i];
3929
3930                 if (depid <= maxDumpId && tocsByDumpId[depid - 1] &&
3931                         strcmp(tocsByDumpId[depid - 1]->desc, "TABLE DATA") == 0)
3932                         lockids[nlockids++] = depid;
3933         }
3934
3935         if (nlockids == 0)
3936         {
3937                 free(lockids);
3938                 return;
3939         }
3940
3941         te->lockDeps = realloc(lockids, nlockids * sizeof(DumpId));
3942         te->nLockDeps = nlockids;
3943 }
3944
3945 /*
3946  * Remove the specified TOC entry from the depCounts of items that depend on
3947  * it, thereby possibly making them ready-to-run.  Any pending item that
3948  * becomes ready should be moved to the ready list.
3949  */
3950 static void
3951 reduce_dependencies(ArchiveHandle *AH, TocEntry *te, TocEntry *ready_list)
3952 {
3953         DumpId          target = te->dumpId;
3954         int                     i;
3955
3956         ahlog(AH, 2, "reducing dependencies for %d\n", target);
3957
3958         /*
3959          * We must examine all entries, not only the ones after the target item,
3960          * because if the user used a -L switch then the original dependency-
3961          * respecting order has been destroyed by SortTocFromFile.
3962          */
3963         for (te = AH->toc->next; te != AH->toc; te = te->next)
3964         {
3965                 for (i = 0; i < te->nDeps; i++)
3966                 {
3967                         if (te->dependencies[i] == target)
3968                         {
3969                                 te->depCount--;
3970                                 if (te->depCount == 0 && te->par_prev != NULL)
3971                                 {
3972                                         /* It must be in the pending list, so remove it ... */
3973                                         par_list_remove(te);
3974                                         /* ... and add to ready_list */
3975                                         par_list_append(ready_list, te);
3976                                 }
3977                         }
3978                 }
3979         }
3980 }
3981
3982 /*
3983  * Set the created flag on the DATA member corresponding to the given
3984  * TABLE member
3985  */
3986 static void
3987 mark_create_done(ArchiveHandle *AH, TocEntry *te)
3988 {
3989         TocEntry   *tes;
3990
3991         for (tes = AH->toc->next; tes != AH->toc; tes = tes->next)
3992         {
3993                 if (strcmp(tes->desc, "TABLE DATA") == 0 &&
3994                         strcmp(tes->tag, te->tag) == 0 &&
3995                         strcmp(tes->namespace ? tes->namespace : "",
3996                                    te->namespace ? te->namespace : "") == 0)
3997                 {
3998                         tes->created = true;
3999                         break;
4000                 }
4001         }
4002 }
4003
4004 /*
4005  * Mark the DATA member corresponding to the given TABLE member
4006  * as not wanted
4007  */
4008 static void
4009 inhibit_data_for_failed_table(ArchiveHandle *AH, TocEntry *te)
4010 {
4011         RestoreOptions *ropt = AH->ropt;
4012         TocEntry   *tes;
4013
4014         ahlog(AH, 1, "table \"%s\" could not be created, will not restore its data\n",
4015                   te->tag);
4016
4017         for (tes = AH->toc->next; tes != AH->toc; tes = tes->next)
4018         {
4019                 if (strcmp(tes->desc, "TABLE DATA") == 0 &&
4020                         strcmp(tes->tag, te->tag) == 0 &&
4021                         strcmp(tes->namespace ? tes->namespace : "",
4022                                    te->namespace ? te->namespace : "") == 0)
4023                 {
4024                         /* mark it unwanted; we assume idWanted array already exists */
4025                         ropt->idWanted[tes->dumpId - 1] = false;
4026                         break;
4027                 }
4028         }
4029 }
4030
4031
4032 /*
4033  * Clone and de-clone routines used in parallel restoration.
4034  *
4035  * Enough of the structure is cloned to ensure that there is no
4036  * conflict between different threads each with their own clone.
4037  *
4038  * These could be public, but no need at present.
4039  */
4040 static ArchiveHandle *
4041 CloneArchive(ArchiveHandle *AH)
4042 {
4043         ArchiveHandle *clone;
4044
4045         /* Make a "flat" copy */
4046         clone = (ArchiveHandle *) malloc(sizeof(ArchiveHandle));
4047         if (clone == NULL)
4048                 die_horribly(AH, modulename, "out of memory\n");
4049         memcpy(clone, AH, sizeof(ArchiveHandle));
4050
4051         /* Handle format-independent fields */
4052         clone->pgCopyBuf = createPQExpBuffer();
4053         clone->sqlBuf = createPQExpBuffer();
4054         clone->sqlparse.tagBuf = NULL;
4055
4056         /* The clone will have its own connection, so disregard connection state */
4057         clone->connection = NULL;
4058         clone->currUser = NULL;
4059         clone->currSchema = NULL;
4060         clone->currTablespace = NULL;
4061         clone->currWithOids = -1;
4062
4063         /* savedPassword must be local in case we change it while connecting */
4064         if (clone->savedPassword)
4065                 clone->savedPassword = strdup(clone->savedPassword);
4066
4067         /* clone has its own error count, too */
4068         clone->public.n_errors = 0;
4069
4070         /* Let the format-specific code have a chance too */
4071         (clone->ClonePtr) (clone);
4072
4073         return clone;
4074 }
4075
4076 /*
4077  * Release clone-local storage.
4078  *
4079  * Note: we assume any clone-local connection was already closed.
4080  */
4081 static void
4082 DeCloneArchive(ArchiveHandle *AH)
4083 {
4084         /* Clear format-specific state */
4085         (AH->DeClonePtr) (AH);
4086
4087         /* Clear state allocated by CloneArchive */
4088         destroyPQExpBuffer(AH->pgCopyBuf);
4089         destroyPQExpBuffer(AH->sqlBuf);
4090         if (AH->sqlparse.tagBuf)
4091                 destroyPQExpBuffer(AH->sqlparse.tagBuf);
4092
4093         /* Clear any connection-local state */
4094         if (AH->currUser)
4095                 free(AH->currUser);
4096         if (AH->currSchema)
4097                 free(AH->currSchema);
4098         if (AH->currTablespace)
4099                 free(AH->currTablespace);
4100         if (AH->savedPassword)
4101                 free(AH->savedPassword);
4102
4103         free(AH);
4104 }