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

   1 /*-------------------------------------------------------------------------
   2  *
   3  * pg_backup_custom.c
   4  *
   5  *      Implements the custom output format.
   6  *
   7  *      The comments with the routined in this code are a good place to
   8  *      understand how to write a new format.
   9  *
  10  *      See the headers to pg_restore for more details.
  11  *
  12  * Copyright (c) 2000, Philip Warner
  13  *              Rights are granted to use this software in any way so long
  14  *              as this notice is not removed.
  15  *
  16  *      The author is not responsible for loss or damages that may
  17  *      and any liability will be limited to the time taken to fix any
  18  *      related bug.
  19  *
  20  *
  21  * IDENTIFICATION
  22  *              src/bin/pg_dump/pg_backup_custom.c
  23  *
  24  *-------------------------------------------------------------------------
  25  */
  26
  27 #include "compress_io.h"
  28 #include "common.h"
  29
  30 /*--------
  31  * Routines in the format interface
  32  *--------
  33  */
  34
  35 static void _ArchiveEntry(ArchiveHandle *AH, TocEntry *te);
  36 static void _StartData(ArchiveHandle *AH, TocEntry *te);
  37 static size_t _WriteData(ArchiveHandle *AH, const void *data, size_t dLen);
  38 static void _EndData(ArchiveHandle *AH, TocEntry *te);
  39 static int      _WriteByte(ArchiveHandle *AH, const int i);
  40 static int      _ReadByte(ArchiveHandle *);
  41 static size_t _WriteBuf(ArchiveHandle *AH, const void *buf, size_t len);
  42 static size_t _ReadBuf(ArchiveHandle *AH, void *buf, size_t len);
  43 static void _CloseArchive(ArchiveHandle *AH);
  44 static void _ReopenArchive(ArchiveHandle *AH);
  45 static void _PrintTocData(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
  46 static void _WriteExtraToc(ArchiveHandle *AH, TocEntry *te);
  47 static void _ReadExtraToc(ArchiveHandle *AH, TocEntry *te);
  48 static void _PrintExtraToc(ArchiveHandle *AH, TocEntry *te);
  49
  50 static void _PrintData(ArchiveHandle *AH);
  51 static void _skipData(ArchiveHandle *AH);
  52 static void _skipBlobs(ArchiveHandle *AH);
  53
  54 static void _StartBlobs(ArchiveHandle *AH, TocEntry *te);
  55 static void _StartBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
  56 static void _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid);
  57 static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
  58 static void _LoadBlobs(ArchiveHandle *AH, bool drop);
  59 static void _Clone(ArchiveHandle *AH);
  60 static void _DeClone(ArchiveHandle *AH);
  61
  62 typedef struct
  63 {
  64         CompressorState *cs;
  65         int                     hasSeek;
  66         pgoff_t         filePos;
  67         pgoff_t         dataStart;
  68 } lclContext;
  69
  70 typedef struct
  71 {
  72         int                     dataState;
  73         pgoff_t         dataPos;
  74 } lclTocEntry;
  75
  76
  77 /*------
  78  * Static declarations
  79  *------
  80  */
  81 static void _readBlockHeader(ArchiveHandle *AH, int *type, int *id);
  82 static pgoff_t _getFilePos(ArchiveHandle *AH, lclContext *ctx);
  83
  84 static size_t _CustomWriteFunc(ArchiveHandle *AH, const char *buf, size_t len);
  85 static size_t _CustomReadFunc(ArchiveHandle *AH, char **buf, size_t *buflen);
  86
  87 static const char *modulename = gettext_noop("custom archiver");
  88
  89
  90
  91 /*
  92  *      Init routine required by ALL formats. This is a global routine
  93  *      and should be declared in pg_backup_archiver.h
  94  *
  95  *      It's task is to create any extra archive context (using AH->formatData),
  96  *      and to initialize the supported function pointers.
  97  *
  98  *      It should also prepare whatever it's input source is for reading/writing,
  99  *      and in the case of a read mode connection, it should load the Header & TOC.
 100  */
 101 void
 102 InitArchiveFmt_Custom(ArchiveHandle *AH)
 103 {
 104         lclContext *ctx;
 105
 106         /* Assuming static functions, this can be copied for each format. */
 107         AH->ArchiveEntryPtr = _ArchiveEntry;
 108         AH->StartDataPtr = _StartData;
 109         AH->WriteDataPtr = _WriteData;
 110         AH->EndDataPtr = _EndData;
 111         AH->WriteBytePtr = _WriteByte;
 112         AH->ReadBytePtr = _ReadByte;
 113         AH->WriteBufPtr = _WriteBuf;
 114         AH->ReadBufPtr = _ReadBuf;
 115         AH->ClosePtr = _CloseArchive;
 116         AH->ReopenPtr = _ReopenArchive;
 117         AH->PrintTocDataPtr = _PrintTocData;
 118         AH->ReadExtraTocPtr = _ReadExtraToc;
 119         AH->WriteExtraTocPtr = _WriteExtraToc;
 120         AH->PrintExtraTocPtr = _PrintExtraToc;
 121
 122         AH->StartBlobsPtr = _StartBlobs;
 123         AH->StartBlobPtr = _StartBlob;
 124         AH->EndBlobPtr = _EndBlob;
 125         AH->EndBlobsPtr = _EndBlobs;
 126         AH->ClonePtr = _Clone;
 127         AH->DeClonePtr = _DeClone;
 128
 129         /* Set up a private area. */
 130         ctx = (lclContext *) pg_calloc(1, sizeof(lclContext));
 131         AH->formatData = (void *) ctx;
 132
 133         /* Initialize LO buffering */
 134         AH->lo_buf_size = LOBBUFSIZE;
 135         AH->lo_buf = (void *) pg_malloc(LOBBUFSIZE);
 136
 137         ctx->filePos = 0;
 138
 139         /*
 140          * Now open the file
 141          */
 142         if (AH->mode == archModeWrite)
 143         {
 144                 if (AH->fSpec && strcmp(AH->fSpec, "") != 0)
 145                 {
 146                         AH->FH = fopen(AH->fSpec, PG_BINARY_W);
 147                         if (!AH->FH)
 148                                 die_horribly(AH, modulename, "could not open output file \"%s\": %s\n",
 149                                                          AH->fSpec, strerror(errno));
 150                 }
 151                 else
 152                 {
 153                         AH->FH = stdout;
 154                         if (!AH->FH)
 155                                 die_horribly(AH, modulename, "could not open output file: %s\n",
 156                                                          strerror(errno));
 157                 }
 158
 159                 ctx->hasSeek = checkSeek(AH->FH);
 160         }
 161         else
 162         {
 163                 if (AH->fSpec && strcmp(AH->fSpec, "") != 0)
 164                 {
 165                         AH->FH = fopen(AH->fSpec, PG_BINARY_R);
 166                         if (!AH->FH)
 167                                 die_horribly(AH, modulename, "could not open input file \"%s\": %s\n",
 168                                                          AH->fSpec, strerror(errno));
 169                 }
 170                 else
 171                 {
 172                         AH->FH = stdin;
 173                         if (!AH->FH)
 174                                 die_horribly(AH, modulename, "could not open input file: %s\n",
 175                                                          strerror(errno));
 176                 }
 177
 178                 ctx->hasSeek = checkSeek(AH->FH);
 179
 180                 ReadHead(AH);
 181                 ReadToc(AH);
 182                 ctx->dataStart = _getFilePos(AH, ctx);
 183         }
 184
 185 }
 186
 187 /*
 188  * Called by the Archiver when the dumper creates a new TOC entry.
 189  *
 190  * Optional.
 191  *
 192  * Set up extrac format-related TOC data.
 193 */
 194 static void
 195 _ArchiveEntry(ArchiveHandle *AH, TocEntry *te)
 196 {
 197         lclTocEntry *ctx;
 198
 199         ctx = (lclTocEntry *) pg_calloc(1, sizeof(lclTocEntry));
 200         if (te->dataDumper)
 201                 ctx->dataState = K_OFFSET_POS_NOT_SET;
 202         else
 203                 ctx->dataState = K_OFFSET_NO_DATA;
 204
 205         te->formatData = (void *) ctx;
 206 }
 207
 208 /*
 209  * Called by the Archiver to save any extra format-related TOC entry
 210  * data.
 211  *
 212  * Optional.
 213  *
 214  * Use the Archiver routines to write data - they are non-endian, and
 215  * maintain other important file information.
 216  */
 217 static void
 218 _WriteExtraToc(ArchiveHandle *AH, TocEntry *te)
 219 {
 220         lclTocEntry *ctx = (lclTocEntry *) te->formatData;
 221
 222         WriteOffset(AH, ctx->dataPos, ctx->dataState);
 223 }
 224
 225 /*
 226  * Called by the Archiver to read any extra format-related TOC data.
 227  *
 228  * Optional.
 229  *
 230  * Needs to match the order defined in _WriteExtraToc, and sould also
 231  * use the Archiver input routines.
 232  */
 233 static void
 234 _ReadExtraToc(ArchiveHandle *AH, TocEntry *te)
 235 {
 236         lclTocEntry *ctx = (lclTocEntry *) te->formatData;
 237
 238         if (ctx == NULL)
 239         {
 240                 ctx = (lclTocEntry *) pg_calloc(1, sizeof(lclTocEntry));
 241                 te->formatData = (void *) ctx;
 242         }
 243
 244         ctx->dataState = ReadOffset(AH, &(ctx->dataPos));
 245
 246         /*
 247          * Prior to V1.7 (pg7.3), we dumped the data size as an int now we don't
 248          * dump it at all.
 249          */
 250         if (AH->version < K_VERS_1_7)
 251                 ReadInt(AH);
 252 }
 253
 254 /*
 255  * Called by the Archiver when restoring an archive to output a comment
 256  * that includes useful information about the TOC entry.
 257  *
 258  * Optional.
 259  *
 260  */
 261 static void
 262 _PrintExtraToc(ArchiveHandle *AH, TocEntry *te)
 263 {
 264         lclTocEntry *ctx = (lclTocEntry *) te->formatData;
 265
 266         if (AH->public.verbose)
 267                 ahprintf(AH, "-- Data Pos: " INT64_FORMAT "\n",
 268                                  (int64) ctx->dataPos);
 269 }
 270
 271 /*
 272  * Called by the archiver when saving TABLE DATA (not schema). This routine
 273  * should save whatever format-specific information is needed to read
 274  * the archive back.
 275  *
 276  * It is called just prior to the dumper's 'DataDumper' routine being called.
 277  *
 278  * Optional, but strongly recommended.
 279  *
 280  */
 281 static void
 282 _StartData(ArchiveHandle *AH, TocEntry *te)
 283 {
 284         lclContext *ctx = (lclContext *) AH->formatData;
 285         lclTocEntry *tctx = (lclTocEntry *) te->formatData;
 286
 287         tctx->dataPos = _getFilePos(AH, ctx);
 288         tctx->dataState = K_OFFSET_POS_SET;
 289
 290         _WriteByte(AH, BLK_DATA);       /* Block type */
 291         WriteInt(AH, te->dumpId);       /* For sanity check */
 292
 293         ctx->cs = AllocateCompressor(AH->compression, _CustomWriteFunc);
 294 }
 295
 296 /*
 297  * Called by archiver when dumper calls WriteData. This routine is
 298  * called for both BLOB and TABLE data; it is the responsibility of
 299  * the format to manage each kind of data using StartBlob/StartData.
 300  *
 301  * It should only be called from within a DataDumper routine.
 302  *
 303  * Mandatory.
 304  */
 305 static size_t
 306 _WriteData(ArchiveHandle *AH, const void *data, size_t dLen)
 307 {
 308         lclContext *ctx = (lclContext *) AH->formatData;
 309         CompressorState *cs = ctx->cs;
 310
 311         if (dLen == 0)
 312                 return 0;
 313
 314         return WriteDataToArchive(AH, cs, data, dLen);
 315 }
 316
 317 /*
 318  * Called by the archiver when a dumper's 'DataDumper' routine has
 319  * finished.
 320  *
 321  * Optional.
 322  *
 323  */
 324 static void
 325 _EndData(ArchiveHandle *AH, TocEntry *te)
 326 {
 327         lclContext *ctx = (lclContext *) AH->formatData;
 328
 329         EndCompressor(AH, ctx->cs);
 330         /* Send the end marker */
 331         WriteInt(AH, 0);
 332 }
 333
 334 /*
 335  * Called by the archiver when starting to save all BLOB DATA (not schema).
 336  * This routine should save whatever format-specific information is needed
 337  * to read the BLOBs back into memory.
 338  *
 339  * It is called just prior to the dumper's DataDumper routine.
 340  *
 341  * Optional, but strongly recommended.
 342  */
 343 static void
 344 _StartBlobs(ArchiveHandle *AH, TocEntry *te)
 345 {
 346         lclContext *ctx = (lclContext *) AH->formatData;
 347         lclTocEntry *tctx = (lclTocEntry *) te->formatData;
 348
 349         tctx->dataPos = _getFilePos(AH, ctx);
 350         tctx->dataState = K_OFFSET_POS_SET;
 351
 352         _WriteByte(AH, BLK_BLOBS);      /* Block type */
 353         WriteInt(AH, te->dumpId);       /* For sanity check */
 354 }
 355
 356 /*
 357  * Called by the archiver when the dumper calls StartBlob.
 358  *
 359  * Mandatory.
 360  *
 361  * Must save the passed OID for retrieval at restore-time.
 362  */
 363 static void
 364 _StartBlob(ArchiveHandle *AH, TocEntry *te, Oid oid)
 365 {
 366         lclContext *ctx = (lclContext *) AH->formatData;
 367
 368         if (oid == 0)
 369                 die_horribly(AH, modulename, "invalid OID for large object\n");
 370
 371         WriteInt(AH, oid);
 372
 373         ctx->cs = AllocateCompressor(AH->compression, _CustomWriteFunc);
 374 }
 375
 376 /*
 377  * Called by the archiver when the dumper calls EndBlob.
 378  *
 379  * Optional.
 380  */
 381 static void
 382 _EndBlob(ArchiveHandle *AH, TocEntry *te, Oid oid)
 383 {
 384         lclContext *ctx = (lclContext *) AH->formatData;
 385
 386         EndCompressor(AH, ctx->cs);
 387         /* Send the end marker */
 388         WriteInt(AH, 0);
 389 }
 390
 391 /*
 392  * Called by the archiver when finishing saving all BLOB DATA.
 393  *
 394  * Optional.
 395  */
 396 static void
 397 _EndBlobs(ArchiveHandle *AH, TocEntry *te)
 398 {
 399         /* Write out a fake zero OID to mark end-of-blobs. */
 400         WriteInt(AH, 0);
 401 }
 402
 403 /*
 404  * Print data for a given TOC entry
 405  */
 406 static void
 407 _PrintTocData(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt)
 408 {
 409         lclContext *ctx = (lclContext *) AH->formatData;
 410         lclTocEntry *tctx = (lclTocEntry *) te->formatData;
 411         int                     blkType;
 412         int                     id;
 413
 414         if (tctx->dataState == K_OFFSET_NO_DATA)
 415                 return;
 416
 417         if (!ctx->hasSeek || tctx->dataState == K_OFFSET_POS_NOT_SET)
 418         {
 419                 /*
 420                  * We cannot seek directly to the desired block.  Instead, skip over
 421                  * block headers until we find the one we want.  This could fail if we
 422                  * are asked to restore items out-of-order.
 423                  */
 424                 _readBlockHeader(AH, &blkType, &id);
 425
 426                 while (blkType != EOF && id != te->dumpId)
 427                 {
 428                         switch (blkType)
 429                         {
 430                                 case BLK_DATA:
 431                                         _skipData(AH);
 432                                         break;
 433
 434                                 case BLK_BLOBS:
 435                                         _skipBlobs(AH);
 436                                         break;
 437
 438                                 default:                /* Always have a default */
 439                                         die_horribly(AH, modulename,
 440                                                                  "unrecognized data block type (%d) while searching archive\n",
 441                                                                  blkType);
 442                                         break;
 443                         }
 444                         _readBlockHeader(AH, &blkType, &id);
 445                 }
 446         }
 447         else
 448         {
 449                 /* We can just seek to the place we need to be. */
 450                 if (fseeko(AH->FH, tctx->dataPos, SEEK_SET) != 0)
 451                         die_horribly(AH, modulename, "error during file seek: %s\n",
 452                                                  strerror(errno));
 453
 454                 _readBlockHeader(AH, &blkType, &id);
 455         }
 456
 457         /* Produce suitable failure message if we fell off end of file */
 458         if (blkType == EOF)
 459         {
 460                 if (tctx->dataState == K_OFFSET_POS_NOT_SET)
 461                         die_horribly(AH, modulename, "could not find block ID %d in archive -- "
 462                                                  "possibly due to out-of-order restore request, "
 463                                                  "which cannot be handled due to lack of data offsets in archive\n",
 464                                                  te->dumpId);
 465                 else if (!ctx->hasSeek)
 466                         die_horribly(AH, modulename, "could not find block ID %d in archive -- "
 467                                                  "possibly due to out-of-order restore request, "
 468                                   "which cannot be handled due to non-seekable input file\n",
 469                                                  te->dumpId);
 470                 else    /* huh, the dataPos led us to EOF? */
 471                         die_horribly(AH, modulename, "could not find block ID %d in archive -- "
 472                                                  "possibly corrupt archive\n",
 473                                                  te->dumpId);
 474         }
 475
 476         /* Are we sane? */
 477         if (id != te->dumpId)
 478                 die_horribly(AH, modulename, "found unexpected block ID (%d) when reading data -- expected %d\n",
 479                                          id, te->dumpId);
 480
 481         switch (blkType)
 482         {
 483                 case BLK_DATA:
 484                         _PrintData(AH);
 485                         break;
 486
 487                 case BLK_BLOBS:
 488                         _LoadBlobs(AH, ropt->dropSchema);
 489                         break;
 490
 491                 default:                                /* Always have a default */
 492                         die_horribly(AH, modulename, "unrecognized data block type %d while restoring archive\n",
 493                                                  blkType);
 494                         break;
 495         }
 496 }
 497
 498 /*
 499  * Print data from current file position.
 500 */
 501 static void
 502 _PrintData(ArchiveHandle *AH)
 503 {
 504         ReadDataFromArchive(AH, AH->compression, _CustomReadFunc);
 505 }
 506
 507 static void
 508 _LoadBlobs(ArchiveHandle *AH, bool drop)
 509 {
 510         Oid                     oid;
 511
 512         StartRestoreBlobs(AH);
 513
 514         oid = ReadInt(AH);
 515         while (oid != 0)
 516         {
 517                 StartRestoreBlob(AH, oid, drop);
 518                 _PrintData(AH);
 519                 EndRestoreBlob(AH, oid);
 520                 oid = ReadInt(AH);
 521         }
 522
 523         EndRestoreBlobs(AH);
 524 }
 525
 526 /*
 527  * Skip the BLOBs from the current file position.
 528  * BLOBS are written sequentially as data blocks (see below).
 529  * Each BLOB is preceded by it's original OID.
 530  * A zero OID indicated the end of the BLOBS
 531  */
 532 static void
 533 _skipBlobs(ArchiveHandle *AH)
 534 {
 535         Oid                     oid;
 536
 537         oid = ReadInt(AH);
 538         while (oid != 0)
 539         {
 540                 _skipData(AH);
 541                 oid = ReadInt(AH);
 542         }
 543 }
 544
 545 /*
 546  * Skip data from current file position.
 547  * Data blocks are formatted as an integer length, followed by data.
 548  * A zero length denoted the end of the block.
 549 */
 550 static void
 551 _skipData(ArchiveHandle *AH)
 552 {
 553         lclContext *ctx = (lclContext *) AH->formatData;
 554         size_t          blkLen;
 555         char       *buf = NULL;
 556         int                     buflen = 0;
 557         size_t          cnt;
 558
 559         blkLen = ReadInt(AH);
 560         while (blkLen != 0)
 561         {
 562                 if (blkLen > buflen)
 563                 {
 564                         if (buf)
 565                                 free(buf);
 566                         buf = (char *) pg_malloc(blkLen);
 567                         buflen = blkLen;
 568                 }
 569                 cnt = fread(buf, 1, blkLen, AH->FH);
 570                 if (cnt != blkLen)
 571                 {
 572                         if (feof(AH->FH))
 573                                 die_horribly(AH, modulename,
 574                                                          "could not read from input file: end of file\n");
 575                         else
 576                                 die_horribly(AH, modulename,
 577                                         "could not read from input file: %s\n", strerror(errno));
 578                 }
 579
 580                 ctx->filePos += blkLen;
 581
 582                 blkLen = ReadInt(AH);
 583         }
 584
 585         if (buf)
 586                 free(buf);
 587 }
 588
 589 /*
 590  * Write a byte of data to the archive.
 591  *
 592  * Mandatory.
 593  *
 594  * Called by the archiver to do integer & byte output to the archive.
 595  */
 596 static int
 597 _WriteByte(ArchiveHandle *AH, const int i)
 598 {
 599         lclContext *ctx = (lclContext *) AH->formatData;
 600         int                     res;
 601
 602         res = fputc(i, AH->FH);
 603         if (res != EOF)
 604                 ctx->filePos += 1;
 605         else
 606                 die_horribly(AH, modulename, "could not write byte: %s\n", strerror(errno));
 607         return res;
 608 }
 609
 610 /*
 611  * Read a byte of data from the archive.
 612  *
 613  * Mandatory
 614  *
 615  * Called by the archiver to read bytes & integers from the archive.
 616  * EOF should be treated as a fatal error.
 617  */
 618 static int
 619 _ReadByte(ArchiveHandle *AH)
 620 {
 621         lclContext *ctx = (lclContext *) AH->formatData;
 622         int                     res;
 623
 624         res = getc(AH->FH);
 625         if (res == EOF)
 626                 die_horribly(AH, modulename, "unexpected end of file\n");
 627         ctx->filePos += 1;
 628         return res;
 629 }
 630
 631 /*
 632  * Write a buffer of data to the archive.
 633  *
 634  * Mandatory.
 635  *
 636  * Called by the archiver to write a block of bytes to the archive.
 637  */
 638 static size_t
 639 _WriteBuf(ArchiveHandle *AH, const void *buf, size_t len)
 640 {
 641         lclContext *ctx = (lclContext *) AH->formatData;
 642         size_t          res;
 643
 644         res = fwrite(buf, 1, len, AH->FH);
 645
 646         if (res != len)
 647                 die_horribly(AH, modulename,
 648                                          "could not write to output file: %s\n", strerror(errno));
 649
 650         ctx->filePos += res;
 651         return res;
 652 }
 653
 654 /*
 655  * Read a block of bytes from the archive.
 656  *
 657  * Mandatory.
 658  *
 659  * Called by the archiver to read a block of bytes from the archive
 660  */
 661 static size_t
 662 _ReadBuf(ArchiveHandle *AH, void *buf, size_t len)
 663 {
 664         lclContext *ctx = (lclContext *) AH->formatData;
 665         size_t          res;
 666
 667         res = fread(buf, 1, len, AH->FH);
 668         ctx->filePos += res;
 669
 670         return res;
 671 }
 672
 673 /*
 674  * Close the archive.
 675  *
 676  * Mandatory.
 677  *
 678  * When writing the archive, this is the routine that actually starts
 679  * the process of saving it to files. No data should be written prior
 680  * to this point, since the user could sort the TOC after creating it.
 681  *
 682  * If an archive is to be written, this toutine must call:
 683  *              WriteHead                       to save the archive header
 684  *              WriteToc                        to save the TOC entries
 685  *              WriteDataChunks         to save all DATA & BLOBs.
 686  *
 687  */
 688 static void
 689 _CloseArchive(ArchiveHandle *AH)
 690 {
 691         lclContext *ctx = (lclContext *) AH->formatData;
 692         pgoff_t         tpos;
 693
 694         if (AH->mode == archModeWrite)
 695         {
 696                 WriteHead(AH);
 697                 tpos = ftello(AH->FH);
 698                 WriteToc(AH);
 699                 ctx->dataStart = _getFilePos(AH, ctx);
 700                 WriteDataChunks(AH);
 701
 702                 /*
 703                  * If possible, re-write the TOC in order to update the data offset
 704                  * information.  This is not essential, as pg_restore can cope in most
 705                  * cases without it; but it can make pg_restore significantly faster
 706                  * in some situations (especially parallel restore).
 707                  */
 708                 if (ctx->hasSeek &&
 709                         fseeko(AH->FH, tpos, SEEK_SET) == 0)
 710                         WriteToc(AH);
 711         }
 712
 713         if (fclose(AH->FH) != 0)
 714                 die_horribly(AH, modulename, "could not close archive file: %s\n", strerror(errno));
 715
 716         AH->FH = NULL;
 717 }
 718
 719 /*
 720  * Reopen the archive's file handle.
 721  *
 722  * We close the original file handle, except on Windows.  (The difference
 723  * is because on Windows, this is used within a multithreading context,
 724  * and we don't want a thread closing the parent file handle.)
 725  */
 726 static void
 727 _ReopenArchive(ArchiveHandle *AH)
 728 {
 729         lclContext *ctx = (lclContext *) AH->formatData;
 730         pgoff_t         tpos;
 731
 732         if (AH->mode == archModeWrite)
 733                 die_horribly(AH, modulename, "can only reopen input archives\n");
 734
 735         /*
 736          * These two cases are user-facing errors since they represent unsupported
 737          * (but not invalid) use-cases.  Word the error messages appropriately.
 738          */
 739         if (AH->fSpec == NULL || strcmp(AH->fSpec, "") == 0)
 740                 die_horribly(AH, modulename, "parallel restore from stdin is not supported\n");
 741         if (!ctx->hasSeek)
 742                 die_horribly(AH, modulename, "parallel restore from non-seekable file is not supported\n");
 743
 744         errno = 0;
 745         tpos = ftello(AH->FH);
 746         if (errno)
 747                 die_horribly(AH, modulename, "could not determine seek position in archive file: %s\n",
 748                                          strerror(errno));
 749
 750 #ifndef WIN32
 751         if (fclose(AH->FH) != 0)
 752                 die_horribly(AH, modulename, "could not close archive file: %s\n",
 753                                          strerror(errno));
 754 #endif
 755
 756         AH->FH = fopen(AH->fSpec, PG_BINARY_R);
 757         if (!AH->FH)
 758                 die_horribly(AH, modulename, "could not open input file \"%s\": %s\n",
 759                                          AH->fSpec, strerror(errno));
 760
 761         if (fseeko(AH->FH, tpos, SEEK_SET) != 0)
 762                 die_horribly(AH, modulename, "could not set seek position in archive file: %s\n",
 763                                          strerror(errno));
 764 }
 765
 766 /*
 767  * Clone format-specific fields during parallel restoration.
 768  */
 769 static void
 770 _Clone(ArchiveHandle *AH)
 771 {
 772         lclContext *ctx = (lclContext *) AH->formatData;
 773
 774         AH->formatData = (lclContext *) pg_malloc(sizeof(lclContext));
 775         memcpy(AH->formatData, ctx, sizeof(lclContext));
 776         ctx = (lclContext *) AH->formatData;
 777
 778         /* sanity check, shouldn't happen */
 779         if (ctx->cs != NULL)
 780                 die_horribly(AH, modulename, "compressor active\n");
 781
 782         /*
 783          * Note: we do not make a local lo_buf because we expect at most one BLOBS
 784          * entry per archive, so no parallelism is possible.  Likewise,
 785          * TOC-entry-local state isn't an issue because any one TOC entry is
 786          * touched by just one worker child.
 787          */
 788 }
 789
 790 static void
 791 _DeClone(ArchiveHandle *AH)
 792 {
 793         lclContext *ctx = (lclContext *) AH->formatData;
 794
 795         free(ctx);
 796 }
 797
 798 /*--------------------------------------------------
 799  * END OF FORMAT CALLBACKS
 800  *--------------------------------------------------
 801  */
 802
 803 /*
 804  * Get the current position in the archive file.
 805  */
 806 static pgoff_t
 807 _getFilePos(ArchiveHandle *AH, lclContext *ctx)
 808 {
 809         pgoff_t         pos;
 810
 811         if (ctx->hasSeek)
 812         {
 813                 pos = ftello(AH->FH);
 814                 if (pos != ctx->filePos)
 815                 {
 816                         write_msg(modulename, "WARNING: ftell mismatch with expected position -- ftell used\n");
 817
 818                         /*
 819                          * Prior to 1.7 (pg7.3) we relied on the internally maintained
 820                          * pointer. Now we rely on ftello() always, unless the file has
 821                          * been found to not support it.
 822                          */
 823                 }
 824         }
 825         else
 826                 pos = ctx->filePos;
 827         return pos;
 828 }
 829
 830 /*
 831  * Read a data block header. The format changed in V1.3, so we
 832  * centralize the code here for simplicity.  Returns *type = EOF
 833  * if at EOF.
 834  */
 835 static void
 836 _readBlockHeader(ArchiveHandle *AH, int *type, int *id)
 837 {
 838         lclContext *ctx = (lclContext *) AH->formatData;
 839         int                     byt;
 840
 841         /*
 842          * Note: if we are at EOF with a pre-1.3 input file, we'll die_horribly
 843          * inside ReadInt rather than returning EOF.  It doesn't seem worth
 844          * jumping through hoops to deal with that case better, because no such
 845          * files are likely to exist in the wild: only some 7.1 development
 846          * versions of pg_dump ever generated such files.
 847          */
 848         if (AH->version < K_VERS_1_3)
 849                 *type = BLK_DATA;
 850         else
 851         {
 852                 byt = getc(AH->FH);
 853                 *type = byt;
 854                 if (byt == EOF)
 855                 {
 856                         *id = 0;                        /* don't return an uninitialized value */
 857                         return;
 858                 }
 859                 ctx->filePos += 1;
 860         }
 861
 862         *id = ReadInt(AH);
 863 }
 864
 865 /*
 866  * Callback function for WriteDataToArchive. Writes one block of (compressed)
 867  * data to the archive.
 868  */
 869 static size_t
 870 _CustomWriteFunc(ArchiveHandle *AH, const char *buf, size_t len)
 871 {
 872         /* never write 0-byte blocks (this should not happen) */
 873         if (len == 0)
 874                 return 0;
 875
 876         WriteInt(AH, len);
 877         return _WriteBuf(AH, buf, len);
 878 }
 879
 880 /*
 881  * Callback function for ReadDataFromArchive. To keep things simple, we
 882  * always read one compressed block at a time.
 883  */
 884 static size_t
 885 _CustomReadFunc(ArchiveHandle *AH, char **buf, size_t *buflen)
 886 {
 887         size_t          blkLen;
 888         size_t          cnt;
 889
 890         /* Read length */
 891         blkLen = ReadInt(AH);
 892         if (blkLen == 0)
 893                 return 0;
 894
 895         /* If the caller's buffer is not large enough, allocate a bigger one */
 896         if (blkLen > *buflen)
 897         {
 898                 free(*buf);
 899                 *buf = (char *) pg_malloc(blkLen);
 900                 *buflen = blkLen;
 901         }
 902
 903         cnt = _ReadBuf(AH, *buf, blkLen);
 904         if (cnt != blkLen)
 905         {
 906                 if (feof(AH->FH))
 907                         die_horribly(AH, modulename,
 908                                                  "could not read from input file: end of file\n");
 909                 else
 910                         die_horribly(AH, modulename,
 911                                         "could not read from input file: %s\n", strerror(errno));
 912         }
 913         return cnt;
 914 }