src/backend/storage/large_object/inv_api.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * inv_api.c
   4  *        routines for manipulating inversion fs large objects. This file
   5  *        contains the user-level large object application interface routines.
   6  *
   7  *
   8  * Note: we access pg_largeobject.data using its C struct declaration.
   9  * This is safe because it immediately follows pageno which is an int4 field,
  10  * and therefore the data field will always be 4-byte aligned, even if it
  11  * is in the short 1-byte-header format.  We have to detoast it since it's
  12  * quite likely to be in compressed or short format.  We also need to check
  13  * for NULLs, since initdb will mark loid and pageno but not data as NOT NULL.
  14  *
  15  * Note: many of these routines leak memory in CurrentMemoryContext, as indeed
  16  * does most of the backend code.  We expect that CurrentMemoryContext will
  17  * be a short-lived context.  Data that must persist across function calls
  18  * is kept either in CacheMemoryContext (the Relation structs) or in the
  19  * memory context given to inv_open (for LargeObjectDesc structs).
  20  *
  21  *
  22  * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
  23  * Portions Copyright (c) 1994, Regents of the University of California
  24  *
  25  *
  26  * IDENTIFICATION
  27  *        $PostgreSQL$
  28  *
  29  *-------------------------------------------------------------------------
  30  */
  31 #include "postgres.h"
  32
  33 #include "access/genam.h"
  34 #include "access/heapam.h"
  35 #include "access/tuptoaster.h"
  36 #include "access/xact.h"
  37 #include "catalog/catalog.h"
  38 #include "catalog/indexing.h"
  39 #include "catalog/pg_largeobject.h"
  40 #include "commands/comment.h"
  41 #include "libpq/libpq-fs.h"
  42 #include "storage/large_object.h"
  43 #include "utils/fmgroids.h"
  44 #include "utils/rel.h"
  45 #include "utils/resowner.h"
  46 #include "utils/snapmgr.h"
  47 #include "utils/tqual.h"
  48
  49
  50 /*
  51  * All accesses to pg_largeobject and its index make use of a single Relation
  52  * reference, so that we only need to open pg_relation once per transaction.
  53  * To avoid problems when the first such reference occurs inside a
  54  * subtransaction, we execute a slightly klugy maneuver to assign ownership of
  55  * the Relation reference to TopTransactionResourceOwner.
  56  */
  57 static Relation lo_heap_r = NULL;
  58 static Relation lo_index_r = NULL;
  59
  60
  61 /*
  62  * Open pg_largeobject and its index, if not already done in current xact
  63  */
  64 static void
  65 open_lo_relation(void)
  66 {
  67         ResourceOwner currentOwner;
  68
  69         if (lo_heap_r && lo_index_r)
  70                 return;                                 /* already open in current xact */
  71
  72         /* Arrange for the top xact to own these relation references */
  73         currentOwner = CurrentResourceOwner;
  74         PG_TRY();
  75         {
  76                 CurrentResourceOwner = TopTransactionResourceOwner;
  77
  78                 /* Use RowExclusiveLock since we might either read or write */
  79                 if (lo_heap_r == NULL)
  80                         lo_heap_r = heap_open(LargeObjectRelationId, RowExclusiveLock);
  81                 if (lo_index_r == NULL)
  82                         lo_index_r = index_open(LargeObjectLOidPNIndexId, RowExclusiveLock);
  83         }
  84         PG_CATCH();
  85         {
  86                 /* Ensure CurrentResourceOwner is restored on error */
  87                 CurrentResourceOwner = currentOwner;
  88                 PG_RE_THROW();
  89         }
  90         PG_END_TRY();
  91         CurrentResourceOwner = currentOwner;
  92 }
  93
  94 /*
  95  * Clean up at main transaction end
  96  */
  97 void
  98 close_lo_relation(bool isCommit)
  99 {
 100         if (lo_heap_r || lo_index_r)
 101         {
 102                 /*
 103                  * Only bother to close if committing; else abort cleanup will handle
 104                  * it
 105                  */
 106                 if (isCommit)
 107                 {
 108                         ResourceOwner currentOwner;
 109
 110                         currentOwner = CurrentResourceOwner;
 111                         PG_TRY();
 112                         {
 113                                 CurrentResourceOwner = TopTransactionResourceOwner;
 114
 115                                 if (lo_index_r)
 116                                         index_close(lo_index_r, NoLock);
 117                                 if (lo_heap_r)
 118                                         heap_close(lo_heap_r, NoLock);
 119                         }
 120                         PG_CATCH();
 121                         {
 122                                 /* Ensure CurrentResourceOwner is restored on error */
 123                                 CurrentResourceOwner = currentOwner;
 124                                 PG_RE_THROW();
 125                         }
 126                         PG_END_TRY();
 127                         CurrentResourceOwner = currentOwner;
 128                 }
 129                 lo_heap_r = NULL;
 130                 lo_index_r = NULL;
 131         }
 132 }
 133
 134
 135 /*
 136  * Same as pg_largeobject.c's LargeObjectExists(), except snapshot to
 137  * read with can be specified.
 138  */
 139 static bool
 140 myLargeObjectExists(Oid loid, Snapshot snapshot)
 141 {
 142         bool            retval = false;
 143         Relation        pg_largeobject;
 144         ScanKeyData skey[1];
 145         SysScanDesc sd;
 146
 147         /*
 148          * See if we can find any tuples belonging to the specified LO
 149          */
 150         ScanKeyInit(&skey[0],
 151                                 Anum_pg_largeobject_loid,
 152                                 BTEqualStrategyNumber, F_OIDEQ,
 153                                 ObjectIdGetDatum(loid));
 154
 155         pg_largeobject = heap_open(LargeObjectRelationId, AccessShareLock);
 156
 157         sd = systable_beginscan(pg_largeobject, LargeObjectLOidPNIndexId, true,
 158                                                         snapshot, 1, skey);
 159
 160         if (systable_getnext(sd) != NULL)
 161                 retval = true;
 162
 163         systable_endscan(sd);
 164
 165         heap_close(pg_largeobject, AccessShareLock);
 166
 167         return retval;
 168 }
 169
 170
 171 static int32
 172 getbytealen(bytea *data)
 173 {
 174         Assert(!VARATT_IS_EXTENDED(data));
 175         if (VARSIZE(data) < VARHDRSZ)
 176                 elog(ERROR, "invalid VARSIZE(data)");
 177         return (VARSIZE(data) - VARHDRSZ);
 178 }
 179
 180
 181 /*
 182  *      inv_create -- create a new large object
 183  *
 184  *      Arguments:
 185  *        lobjId - OID to use for new large object, or InvalidOid to pick one
 186  *
 187  *      Returns:
 188  *        OID of new object
 189  *
 190  * If lobjId is not InvalidOid, then an error occurs if the OID is already
 191  * in use.
 192  */
 193 Oid
 194 inv_create(Oid lobjId)
 195 {
 196         /*
 197          * Allocate an OID to be the LO's identifier, unless we were told what to
 198          * use.  We can use the index on pg_largeobject for checking OID
 199          * uniqueness, even though it has additional columns besides OID.
 200          */
 201         if (!OidIsValid(lobjId))
 202         {
 203                 open_lo_relation();
 204
 205                 lobjId = GetNewOidWithIndex(lo_heap_r, LargeObjectLOidPNIndexId,
 206                                                                         Anum_pg_largeobject_loid);
 207         }
 208
 209         /*
 210          * Create the LO by writing an empty first page for it in pg_largeobject
 211          * (will fail if duplicate)
 212          */
 213         LargeObjectCreate(lobjId);
 214
 215         /*
 216          * Advance command counter to make new tuple visible to later operations.
 217          */
 218         CommandCounterIncrement();
 219
 220         return lobjId;
 221 }
 222
 223 /*
 224  *      inv_open -- access an existing large object.
 225  *
 226  *              Returns:
 227  *                Large object descriptor, appropriately filled in.  The descriptor
 228  *                and subsidiary data are allocated in the specified memory context,
 229  *                which must be suitably long-lived for the caller's purposes.
 230  */
 231 LargeObjectDesc *
 232 inv_open(Oid lobjId, int flags, MemoryContext mcxt)
 233 {
 234         LargeObjectDesc *retval;
 235
 236         retval = (LargeObjectDesc *) MemoryContextAlloc(mcxt,
 237                                                                                                         sizeof(LargeObjectDesc));
 238
 239         retval->id = lobjId;
 240         retval->subid = GetCurrentSubTransactionId();
 241         retval->offset = 0;
 242
 243         if (flags & INV_WRITE)
 244         {
 245                 retval->snapshot = SnapshotNow;
 246                 retval->flags = IFS_WRLOCK | IFS_RDLOCK;
 247         }
 248         else if (flags & INV_READ)
 249         {
 250                 /*
 251                  * We must register the snapshot in TopTransaction's resowner, because
 252                  * it must stay alive until the LO is closed rather than until the
 253                  * current portal shuts down.
 254                  */
 255                 retval->snapshot = RegisterSnapshotOnOwner(GetActiveSnapshot(),
 256                                                                                                 TopTransactionResourceOwner);
 257                 retval->flags = IFS_RDLOCK;
 258         }
 259         else
 260                 elog(ERROR, "invalid flags: %d", flags);
 261
 262         /* Can't use LargeObjectExists here because it always uses SnapshotNow */
 263         if (!myLargeObjectExists(lobjId, retval->snapshot))
 264                 ereport(ERROR,
 265                                 (errcode(ERRCODE_UNDEFINED_OBJECT),
 266                                  errmsg("large object %u does not exist", lobjId)));
 267
 268         return retval;
 269 }
 270
 271 /*
 272  * Closes a large object descriptor previously made by inv_open(), and
 273  * releases the long-term memory used by it.
 274  */
 275 void
 276 inv_close(LargeObjectDesc *obj_desc)
 277 {
 278         Assert(PointerIsValid(obj_desc));
 279
 280         if (obj_desc->snapshot != SnapshotNow)
 281                 UnregisterSnapshotFromOwner(obj_desc->snapshot,
 282                                                                         TopTransactionResourceOwner);
 283
 284         pfree(obj_desc);
 285 }
 286
 287 /*
 288  * Destroys an existing large object (not to be confused with a descriptor!)
 289  *
 290  * returns -1 if failed
 291  */
 292 int
 293 inv_drop(Oid lobjId)
 294 {
 295         LargeObjectDrop(lobjId);
 296
 297         /* Delete any comments on the large object */
 298         DeleteComments(lobjId, LargeObjectRelationId, 0);
 299
 300         /*
 301          * Advance command counter so that tuple removal will be seen by later
 302          * large-object operations in this transaction.
 303          */
 304         CommandCounterIncrement();
 305
 306         return 1;
 307 }
 308
 309 /*
 310  * Determine size of a large object
 311  *
 312  * NOTE: LOs can contain gaps, just like Unix files.  We actually return
 313  * the offset of the last byte + 1.
 314  */
 315 static uint32
 316 inv_getsize(LargeObjectDesc *obj_desc)
 317 {
 318         bool            found = false;
 319         uint32          lastbyte = 0;
 320         ScanKeyData skey[1];
 321         SysScanDesc sd;
 322         HeapTuple       tuple;
 323
 324         Assert(PointerIsValid(obj_desc));
 325
 326         open_lo_relation();
 327
 328         ScanKeyInit(&skey[0],
 329                                 Anum_pg_largeobject_loid,
 330                                 BTEqualStrategyNumber, F_OIDEQ,
 331                                 ObjectIdGetDatum(obj_desc->id));
 332
 333         sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
 334                                                                         obj_desc->snapshot, 1, skey);
 335
 336         /*
 337          * Because the pg_largeobject index is on both loid and pageno, but we
 338          * constrain only loid, a backwards scan should visit all pages of the
 339          * large object in reverse pageno order.  So, it's sufficient to examine
 340          * the first valid tuple (== last valid page).
 341          */
 342         while ((tuple = systable_getnext_ordered(sd, BackwardScanDirection)) != NULL)
 343         {
 344                 Form_pg_largeobject data;
 345                 bytea      *datafield;
 346                 bool            pfreeit;
 347
 348                 found = true;
 349                 if (HeapTupleHasNulls(tuple))   /* paranoia */
 350                         elog(ERROR, "null field found in pg_largeobject");
 351                 data = (Form_pg_largeobject) GETSTRUCT(tuple);
 352                 datafield = &(data->data);              /* see note at top of file */
 353                 pfreeit = false;
 354                 if (VARATT_IS_EXTENDED(datafield))
 355                 {
 356                         datafield = (bytea *)
 357                                 heap_tuple_untoast_attr((struct varlena *) datafield);
 358                         pfreeit = true;
 359                 }
 360                 lastbyte = data->pageno * LOBLKSIZE + getbytealen(datafield);
 361                 if (pfreeit)
 362                         pfree(datafield);
 363                 break;
 364         }
 365
 366         systable_endscan_ordered(sd);
 367
 368         if (!found)
 369                 ereport(ERROR,
 370                                 (errcode(ERRCODE_UNDEFINED_OBJECT),
 371                                  errmsg("large object %u does not exist", obj_desc->id)));
 372         return lastbyte;
 373 }
 374
 375 int
 376 inv_seek(LargeObjectDesc *obj_desc, int offset, int whence)
 377 {
 378         Assert(PointerIsValid(obj_desc));
 379
 380         switch (whence)
 381         {
 382                 case SEEK_SET:
 383                         if (offset < 0)
 384                                 elog(ERROR, "invalid seek offset: %d", offset);
 385                         obj_desc->offset = offset;
 386                         break;
 387                 case SEEK_CUR:
 388                         if (offset < 0 && obj_desc->offset < ((uint32) (-offset)))
 389                                 elog(ERROR, "invalid seek offset: %d", offset);
 390                         obj_desc->offset += offset;
 391                         break;
 392                 case SEEK_END:
 393                         {
 394                                 uint32          size = inv_getsize(obj_desc);
 395
 396                                 if (offset < 0 && size < ((uint32) (-offset)))
 397                                         elog(ERROR, "invalid seek offset: %d", offset);
 398                                 obj_desc->offset = size + offset;
 399                         }
 400                         break;
 401                 default:
 402                         elog(ERROR, "invalid whence: %d", whence);
 403         }
 404         return obj_desc->offset;
 405 }
 406
 407 int
 408 inv_tell(LargeObjectDesc *obj_desc)
 409 {
 410         Assert(PointerIsValid(obj_desc));
 411
 412         return obj_desc->offset;
 413 }
 414
 415 int
 416 inv_read(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 417 {
 418         int                     nread = 0;
 419         int                     n;
 420         int                     off;
 421         int                     len;
 422         int32           pageno = (int32) (obj_desc->offset / LOBLKSIZE);
 423         uint32          pageoff;
 424         ScanKeyData skey[2];
 425         SysScanDesc sd;
 426         HeapTuple       tuple;
 427
 428         Assert(PointerIsValid(obj_desc));
 429         Assert(buf != NULL);
 430
 431         if (nbytes <= 0)
 432                 return 0;
 433
 434         open_lo_relation();
 435
 436         ScanKeyInit(&skey[0],
 437                                 Anum_pg_largeobject_loid,
 438                                 BTEqualStrategyNumber, F_OIDEQ,
 439                                 ObjectIdGetDatum(obj_desc->id));
 440
 441         ScanKeyInit(&skey[1],
 442                                 Anum_pg_largeobject_pageno,
 443                                 BTGreaterEqualStrategyNumber, F_INT4GE,
 444                                 Int32GetDatum(pageno));
 445
 446         sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
 447                                                                         obj_desc->snapshot, 2, skey);
 448
 449         while ((tuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
 450         {
 451                 Form_pg_largeobject data;
 452                 bytea      *datafield;
 453                 bool            pfreeit;
 454
 455                 if (HeapTupleHasNulls(tuple))   /* paranoia */
 456                         elog(ERROR, "null field found in pg_largeobject");
 457                 data = (Form_pg_largeobject) GETSTRUCT(tuple);
 458
 459                 /*
 460                  * We expect the indexscan will deliver pages in order.  However,
 461                  * there may be missing pages if the LO contains unwritten "holes". We
 462                  * want missing sections to read out as zeroes.
 463                  */
 464                 pageoff = ((uint32) data->pageno) * LOBLKSIZE;
 465                 if (pageoff > obj_desc->offset)
 466                 {
 467                         n = pageoff - obj_desc->offset;
 468                         n = (n <= (nbytes - nread)) ? n : (nbytes - nread);
 469                         MemSet(buf + nread, 0, n);
 470                         nread += n;
 471                         obj_desc->offset += n;
 472                 }
 473
 474                 if (nread < nbytes)
 475                 {
 476                         Assert(obj_desc->offset >= pageoff);
 477                         off = (int) (obj_desc->offset - pageoff);
 478                         Assert(off >= 0 && off < LOBLKSIZE);
 479
 480                         datafield = &(data->data);      /* see note at top of file */
 481                         pfreeit = false;
 482                         if (VARATT_IS_EXTENDED(datafield))
 483                         {
 484                                 datafield = (bytea *)
 485                                         heap_tuple_untoast_attr((struct varlena *) datafield);
 486                                 pfreeit = true;
 487                         }
 488                         len = getbytealen(datafield);
 489                         if (len > off)
 490                         {
 491                                 n = len - off;
 492                                 n = (n <= (nbytes - nread)) ? n : (nbytes - nread);
 493                                 memcpy(buf + nread, VARDATA(datafield) + off, n);
 494                                 nread += n;
 495                                 obj_desc->offset += n;
 496                         }
 497                         if (pfreeit)
 498                                 pfree(datafield);
 499                 }
 500
 501                 if (nread >= nbytes)
 502                         break;
 503         }
 504
 505         systable_endscan_ordered(sd);
 506
 507         return nread;
 508 }
 509
 510 int
 511 inv_write(LargeObjectDesc *obj_desc, const char *buf, int nbytes)
 512 {
 513         int                     nwritten = 0;
 514         int                     n;
 515         int                     off;
 516         int                     len;
 517         int32           pageno = (int32) (obj_desc->offset / LOBLKSIZE);
 518         ScanKeyData skey[2];
 519         SysScanDesc sd;
 520         HeapTuple       oldtuple;
 521         Form_pg_largeobject olddata;
 522         bool            neednextpage;
 523         bytea      *datafield;
 524         bool            pfreeit;
 525         struct
 526         {
 527                 bytea           hdr;
 528                 char            data[LOBLKSIZE];        /* make struct big enough */
 529                 int32           align_it;       /* ensure struct is aligned well enough */
 530         }                       workbuf;
 531         char       *workb = VARDATA(&workbuf.hdr);
 532         HeapTuple       newtup;
 533         Datum           values[Natts_pg_largeobject];
 534         bool            nulls[Natts_pg_largeobject];
 535         bool            replace[Natts_pg_largeobject];
 536         CatalogIndexState indstate;
 537
 538         Assert(PointerIsValid(obj_desc));
 539         Assert(buf != NULL);
 540
 541         /* enforce writability because snapshot is probably wrong otherwise */
 542         if ((obj_desc->flags & IFS_WRLOCK) == 0)
 543                 ereport(ERROR,
 544                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 545                                  errmsg("large object %u was not opened for writing",
 546                                                 obj_desc->id)));
 547
 548         if (nbytes <= 0)
 549                 return 0;
 550
 551         open_lo_relation();
 552
 553         indstate = CatalogOpenIndexes(lo_heap_r);
 554
 555         ScanKeyInit(&skey[0],
 556                                 Anum_pg_largeobject_loid,
 557                                 BTEqualStrategyNumber, F_OIDEQ,
 558                                 ObjectIdGetDatum(obj_desc->id));
 559
 560         ScanKeyInit(&skey[1],
 561                                 Anum_pg_largeobject_pageno,
 562                                 BTGreaterEqualStrategyNumber, F_INT4GE,
 563                                 Int32GetDatum(pageno));
 564
 565         sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
 566                                                                         obj_desc->snapshot, 2, skey);
 567
 568         oldtuple = NULL;
 569         olddata = NULL;
 570         neednextpage = true;
 571
 572         while (nwritten < nbytes)
 573         {
 574                 /*
 575                  * If possible, get next pre-existing page of the LO.  We expect the
 576                  * indexscan will deliver these in order --- but there may be holes.
 577                  */
 578                 if (neednextpage)
 579                 {
 580                         if ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
 581                         {
 582                                 if (HeapTupleHasNulls(oldtuple))                /* paranoia */
 583                                         elog(ERROR, "null field found in pg_largeobject");
 584                                 olddata = (Form_pg_largeobject) GETSTRUCT(oldtuple);
 585                                 Assert(olddata->pageno >= pageno);
 586                         }
 587                         neednextpage = false;
 588                 }
 589
 590                 /*
 591                  * If we have a pre-existing page, see if it is the page we want to
 592                  * write, or a later one.
 593                  */
 594                 if (olddata != NULL && olddata->pageno == pageno)
 595                 {
 596                         /*
 597                          * Update an existing page with fresh data.
 598                          *
 599                          * First, load old data into workbuf
 600                          */
 601                         datafield = &(olddata->data);           /* see note at top of file */
 602                         pfreeit = false;
 603                         if (VARATT_IS_EXTENDED(datafield))
 604                         {
 605                                 datafield = (bytea *)
 606                                         heap_tuple_untoast_attr((struct varlena *) datafield);
 607                                 pfreeit = true;
 608                         }
 609                         len = getbytealen(datafield);
 610                         Assert(len <= LOBLKSIZE);
 611                         memcpy(workb, VARDATA(datafield), len);
 612                         if (pfreeit)
 613                                 pfree(datafield);
 614
 615                         /*
 616                          * Fill any hole
 617                          */
 618                         off = (int) (obj_desc->offset % LOBLKSIZE);
 619                         if (off > len)
 620                                 MemSet(workb + len, 0, off - len);
 621
 622                         /*
 623                          * Insert appropriate portion of new data
 624                          */
 625                         n = LOBLKSIZE - off;
 626                         n = (n <= (nbytes - nwritten)) ? n : (nbytes - nwritten);
 627                         memcpy(workb + off, buf + nwritten, n);
 628                         nwritten += n;
 629                         obj_desc->offset += n;
 630                         off += n;
 631                         /* compute valid length of new page */
 632                         len = (len >= off) ? len : off;
 633                         SET_VARSIZE(&workbuf.hdr, len + VARHDRSZ);
 634
 635                         /*
 636                          * Form and insert updated tuple
 637                          */
 638                         memset(values, 0, sizeof(values));
 639                         memset(nulls, false, sizeof(nulls));
 640                         memset(replace, false, sizeof(replace));
 641                         values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
 642                         replace[Anum_pg_largeobject_data - 1] = true;
 643                         newtup = heap_modify_tuple(oldtuple, RelationGetDescr(lo_heap_r),
 644                                                                            values, nulls, replace);
 645                         simple_heap_update(lo_heap_r, &newtup->t_self, newtup);
 646                         CatalogIndexInsert(indstate, newtup);
 647                         heap_freetuple(newtup);
 648
 649                         /*
 650                          * We're done with this old page.
 651                          */
 652                         oldtuple = NULL;
 653                         olddata = NULL;
 654                         neednextpage = true;
 655                 }
 656                 else
 657                 {
 658                         /*
 659                          * Write a brand new page.
 660                          *
 661                          * First, fill any hole
 662                          */
 663                         off = (int) (obj_desc->offset % LOBLKSIZE);
 664                         if (off > 0)
 665                                 MemSet(workb, 0, off);
 666
 667                         /*
 668                          * Insert appropriate portion of new data
 669                          */
 670                         n = LOBLKSIZE - off;
 671                         n = (n <= (nbytes - nwritten)) ? n : (nbytes - nwritten);
 672                         memcpy(workb + off, buf + nwritten, n);
 673                         nwritten += n;
 674                         obj_desc->offset += n;
 675                         /* compute valid length of new page */
 676                         len = off + n;
 677                         SET_VARSIZE(&workbuf.hdr, len + VARHDRSZ);
 678
 679                         /*
 680                          * Form and insert updated tuple
 681                          */
 682                         memset(values, 0, sizeof(values));
 683                         memset(nulls, false, sizeof(nulls));
 684                         values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(obj_desc->id);
 685                         values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(pageno);
 686                         values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
 687                         newtup = heap_form_tuple(lo_heap_r->rd_att, values, nulls);
 688                         simple_heap_insert(lo_heap_r, newtup);
 689                         CatalogIndexInsert(indstate, newtup);
 690                         heap_freetuple(newtup);
 691                 }
 692                 pageno++;
 693         }
 694
 695         systable_endscan_ordered(sd);
 696
 697         CatalogCloseIndexes(indstate);
 698
 699         /*
 700          * Advance command counter so that my tuple updates will be seen by later
 701          * large-object operations in this transaction.
 702          */
 703         CommandCounterIncrement();
 704
 705         return nwritten;
 706 }
 707
 708 void
 709 inv_truncate(LargeObjectDesc *obj_desc, int len)
 710 {
 711         int32           pageno = (int32) (len / LOBLKSIZE);
 712         int                     off;
 713         ScanKeyData skey[2];
 714         SysScanDesc sd;
 715         HeapTuple       oldtuple;
 716         Form_pg_largeobject olddata;
 717         struct
 718         {
 719                 bytea           hdr;
 720                 char            data[LOBLKSIZE];        /* make struct big enough */
 721                 int32           align_it;       /* ensure struct is aligned well enough */
 722         }                       workbuf;
 723         char       *workb = VARDATA(&workbuf.hdr);
 724         HeapTuple       newtup;
 725         Datum           values[Natts_pg_largeobject];
 726         bool            nulls[Natts_pg_largeobject];
 727         bool            replace[Natts_pg_largeobject];
 728         CatalogIndexState indstate;
 729
 730         Assert(PointerIsValid(obj_desc));
 731
 732         /* enforce writability because snapshot is probably wrong otherwise */
 733         if ((obj_desc->flags & IFS_WRLOCK) == 0)
 734                 ereport(ERROR,
 735                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 736                                  errmsg("large object %u was not opened for writing",
 737                                                 obj_desc->id)));
 738
 739         open_lo_relation();
 740
 741         indstate = CatalogOpenIndexes(lo_heap_r);
 742
 743         ScanKeyInit(&skey[0],
 744                                 Anum_pg_largeobject_loid,
 745                                 BTEqualStrategyNumber, F_OIDEQ,
 746                                 ObjectIdGetDatum(obj_desc->id));
 747
 748         ScanKeyInit(&skey[1],
 749                                 Anum_pg_largeobject_pageno,
 750                                 BTGreaterEqualStrategyNumber, F_INT4GE,
 751                                 Int32GetDatum(pageno));
 752
 753         sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
 754                                                                         obj_desc->snapshot, 2, skey);
 755
 756         /*
 757          * If possible, get the page the truncation point is in. The truncation
 758          * point may be beyond the end of the LO or in a hole.
 759          */
 760         olddata = NULL;
 761         if ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
 762         {
 763                 if (HeapTupleHasNulls(oldtuple))                /* paranoia */
 764                         elog(ERROR, "null field found in pg_largeobject");
 765                 olddata = (Form_pg_largeobject) GETSTRUCT(oldtuple);
 766                 Assert(olddata->pageno >= pageno);
 767         }
 768
 769         /*
 770          * If we found the page of the truncation point we need to truncate the
 771          * data in it.  Otherwise if we're in a hole, we need to create a page to
 772          * mark the end of data.
 773          */
 774         if (olddata != NULL && olddata->pageno == pageno)
 775         {
 776                 /* First, load old data into workbuf */
 777                 bytea      *datafield = &(olddata->data);               /* see note at top of
 778                                                                                                                  * file */
 779                 bool            pfreeit = false;
 780                 int                     pagelen;
 781
 782                 if (VARATT_IS_EXTENDED(datafield))
 783                 {
 784                         datafield = (bytea *)
 785                                 heap_tuple_untoast_attr((struct varlena *) datafield);
 786                         pfreeit = true;
 787                 }
 788                 pagelen = getbytealen(datafield);
 789                 Assert(pagelen <= LOBLKSIZE);
 790                 memcpy(workb, VARDATA(datafield), pagelen);
 791                 if (pfreeit)
 792                         pfree(datafield);
 793
 794                 /*
 795                  * Fill any hole
 796                  */
 797                 off = len % LOBLKSIZE;
 798                 if (off > pagelen)
 799                         MemSet(workb + pagelen, 0, off - pagelen);
 800
 801                 /* compute length of new page */
 802                 SET_VARSIZE(&workbuf.hdr, off + VARHDRSZ);
 803
 804                 /*
 805                  * Form and insert updated tuple
 806                  */
 807                 memset(values, 0, sizeof(values));
 808                 memset(nulls, false, sizeof(nulls));
 809                 memset(replace, false, sizeof(replace));
 810                 values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
 811                 replace[Anum_pg_largeobject_data - 1] = true;
 812                 newtup = heap_modify_tuple(oldtuple, RelationGetDescr(lo_heap_r),
 813                                                                    values, nulls, replace);
 814                 simple_heap_update(lo_heap_r, &newtup->t_self, newtup);
 815                 CatalogIndexInsert(indstate, newtup);
 816                 heap_freetuple(newtup);
 817         }
 818         else
 819         {
 820                 /*
 821                  * If the first page we found was after the truncation point, we're in
 822                  * a hole that we'll fill, but we need to delete the later page.
 823                  */
 824                 if (olddata != NULL && olddata->pageno > pageno)
 825                         simple_heap_delete(lo_heap_r, &oldtuple->t_self);
 826
 827                 /*
 828                  * Write a brand new page.
 829                  *
 830                  * Fill the hole up to the truncation point
 831                  */
 832                 off = len % LOBLKSIZE;
 833                 if (off > 0)
 834                         MemSet(workb, 0, off);
 835
 836                 /* compute length of new page */
 837                 SET_VARSIZE(&workbuf.hdr, off + VARHDRSZ);
 838
 839                 /*
 840                  * Form and insert new tuple
 841                  */
 842                 memset(values, 0, sizeof(values));
 843                 memset(nulls, false, sizeof(nulls));
 844                 values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(obj_desc->id);
 845                 values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(pageno);
 846                 values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
 847                 newtup = heap_form_tuple(lo_heap_r->rd_att, values, nulls);
 848                 simple_heap_insert(lo_heap_r, newtup);
 849                 CatalogIndexInsert(indstate, newtup);
 850                 heap_freetuple(newtup);
 851         }
 852
 853         /*
 854          * Delete any pages after the truncation point
 855          */
 856         while ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
 857         {
 858                 simple_heap_delete(lo_heap_r, &oldtuple->t_self);
 859         }
 860
 861         systable_endscan_ordered(sd);
 862
 863         CatalogCloseIndexes(indstate);
 864
 865         /*
 866          * Advance command counter so that tuple updates will be seen by later
 867          * large-object operations in this transaction.
 868          */
 869         CommandCounterIncrement();
 870 }