granicus.if.org Git - postgresql/blob - src/backend/access/heap/tuptoaster.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * tuptoaster.c
   4  *        Support routines for external and compressed storage of
   5  *        variable size attributes.
   6  *
   7  * Copyright (c) 2000-2005, PostgreSQL Global Development Group
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL: pgsql/src/backend/access/heap/tuptoaster.c,v 1.57 2005/12/03 05:50:59 tgl Exp $
  12  *
  13  *
  14  * INTERFACE ROUTINES
  15  *              toast_insert_or_update -
  16  *                      Try to make a given tuple fit into one page by compressing
  17  *                      or moving off attributes
  18  *
  19  *              toast_delete -
  20  *                      Reclaim toast storage when a tuple is deleted
  21  *
  22  *              heap_tuple_untoast_attr -
  23  *                      Fetch back a given value from the "secondary" relation
  24  *
  25  *-------------------------------------------------------------------------
  26  */
  27
  28 #include "postgres.h"
  29
  30 #include <unistd.h>
  31 #include <fcntl.h>
  32
  33 #include "access/heapam.h"
  34 #include "access/genam.h"
  35 #include "access/tuptoaster.h"
  36 #include "catalog/catalog.h"
  37 #include "utils/rel.h"
  38 #include "utils/builtins.h"
  39 #include "utils/fmgroids.h"
  40 #include "utils/pg_lzcompress.h"
  41 #include "utils/typcache.h"
  42
  43
  44 #undef TOAST_DEBUG
  45
  46 static void toast_delete_datum(Relation rel, Datum value);
  47 static Datum toast_save_datum(Relation rel, Datum value);
  48 static varattrib *toast_fetch_datum(varattrib *attr);
  49 static varattrib *toast_fetch_datum_slice(varattrib *attr,
  50                                                 int32 sliceoffset, int32 length);
  51
  52
  53 /* ----------
  54  * heap_tuple_fetch_attr -
  55  *
  56  *      Public entry point to get back a toasted value
  57  *      external storage (possibly still in compressed format).
  58  * ----------
  59  */
  60 varattrib *
  61 heap_tuple_fetch_attr(varattrib *attr)
  62 {
  63         varattrib  *result;
  64
  65         if (VARATT_IS_EXTERNAL(attr))
  66         {
  67                 /*
  68                  * This is an external stored plain value
  69                  */
  70                 result = toast_fetch_datum(attr);
  71         }
  72         else
  73         {
  74                 /*
  75                  * This is a plain value inside of the main tuple - why am I called?
  76                  */
  77                 result = attr;
  78         }
  79
  80         return result;
  81 }
  82
  83
  84 /* ----------
  85  * heap_tuple_untoast_attr -
  86  *
  87  *      Public entry point to get back a toasted value from compression
  88  *      or external storage.
  89  * ----------
  90  */
  91 varattrib *
  92 heap_tuple_untoast_attr(varattrib *attr)
  93 {
  94         varattrib  *result;
  95
  96         if (VARATT_IS_EXTERNAL(attr))
  97         {
  98                 if (VARATT_IS_COMPRESSED(attr))
  99                 {
 100                         /* ----------
 101                          * This is an external stored compressed value
 102                          * Fetch it from the toast heap and decompress.
 103                          * ----------
 104                          */
 105                         varattrib  *tmp;
 106
 107                         tmp = toast_fetch_datum(attr);
 108                         result = (varattrib *) palloc(attr->va_content.va_external.va_rawsize
 109                                                                                   + VARHDRSZ);
 110                         VARATT_SIZEP(result) = attr->va_content.va_external.va_rawsize
 111                                 + VARHDRSZ;
 112                         pglz_decompress((PGLZ_Header *) tmp, VARATT_DATA(result));
 113
 114                         pfree(tmp);
 115                 }
 116                 else
 117                 {
 118                         /*
 119                          * This is an external stored plain value
 120                          */
 121                         result = toast_fetch_datum(attr);
 122                 }
 123         }
 124         else if (VARATT_IS_COMPRESSED(attr))
 125         {
 126                 /*
 127                  * This is a compressed value inside of the main tuple
 128                  */
 129                 result = (varattrib *) palloc(attr->va_content.va_compressed.va_rawsize
 130                                                                           + VARHDRSZ);
 131                 VARATT_SIZEP(result) = attr->va_content.va_compressed.va_rawsize
 132                         + VARHDRSZ;
 133                 pglz_decompress((PGLZ_Header *) attr, VARATT_DATA(result));
 134         }
 135         else
 136
 137                 /*
 138                  * This is a plain value inside of the main tuple - why am I called?
 139                  */
 140                 return attr;
 141
 142         return result;
 143 }
 144
 145
 146 /* ----------
 147  * heap_tuple_untoast_attr_slice -
 148  *
 149  *              Public entry point to get back part of a toasted value
 150  *              from compression or external storage.
 151  * ----------
 152  */
 153 varattrib *
 154 heap_tuple_untoast_attr_slice(varattrib *attr, int32 sliceoffset, int32 slicelength)
 155 {
 156         varattrib  *preslice;
 157         varattrib  *result;
 158         int32           attrsize;
 159
 160         if (VARATT_IS_COMPRESSED(attr))
 161         {
 162                 varattrib  *tmp;
 163
 164                 if (VARATT_IS_EXTERNAL(attr))
 165                         tmp = toast_fetch_datum(attr);
 166                 else
 167                 {
 168                         tmp = attr;                     /* compressed in main tuple */
 169                 }
 170
 171                 preslice = (varattrib *) palloc(attr->va_content.va_external.va_rawsize
 172                                                                                 + VARHDRSZ);
 173                 VARATT_SIZEP(preslice) = attr->va_content.va_external.va_rawsize + VARHDRSZ;
 174                 pglz_decompress((PGLZ_Header *) tmp, VARATT_DATA(preslice));
 175
 176                 if (tmp != attr)
 177                         pfree(tmp);
 178         }
 179         else
 180         {
 181                 /* Plain value */
 182                 if (VARATT_IS_EXTERNAL(attr))
 183                 {
 184                         /* fast path */
 185                         return (toast_fetch_datum_slice(attr, sliceoffset, slicelength));
 186                 }
 187                 else
 188                         preslice = attr;
 189         }
 190
 191         /* slicing of datum for compressed cases and plain value */
 192
 193         attrsize = VARSIZE(preslice) - VARHDRSZ;
 194         if (sliceoffset >= attrsize)
 195         {
 196                 sliceoffset = 0;
 197                 slicelength = 0;
 198         }
 199
 200         if (((sliceoffset + slicelength) > attrsize) || slicelength < 0)
 201                 slicelength = attrsize - sliceoffset;
 202
 203         result = (varattrib *) palloc(slicelength + VARHDRSZ);
 204         VARATT_SIZEP(result) = slicelength + VARHDRSZ;
 205
 206         memcpy(VARDATA(result), VARDATA(preslice) + sliceoffset, slicelength);
 207
 208         if (preslice != attr)
 209                 pfree(preslice);
 210
 211         return result;
 212 }
 213
 214
 215 /* ----------
 216  * toast_raw_datum_size -
 217  *
 218  *      Return the raw (detoasted) size of a varlena datum
 219  * ----------
 220  */
 221 Size
 222 toast_raw_datum_size(Datum value)
 223 {
 224         varattrib  *attr = (varattrib *) DatumGetPointer(value);
 225         Size            result;
 226
 227         if (VARATT_IS_COMPRESSED(attr))
 228         {
 229                 /*
 230                  * va_rawsize shows the original data size, whether the datum is
 231                  * external or not.
 232                  */
 233                 result = attr->va_content.va_compressed.va_rawsize + VARHDRSZ;
 234         }
 235         else if (VARATT_IS_EXTERNAL(attr))
 236         {
 237                 /*
 238                  * an uncompressed external attribute has rawsize including the header
 239                  * (not too consistent!)
 240                  */
 241                 result = attr->va_content.va_external.va_rawsize;
 242         }
 243         else
 244         {
 245                 /* plain untoasted datum */
 246                 result = VARSIZE(attr);
 247         }
 248         return result;
 249 }
 250
 251 /* ----------
 252  * toast_datum_size
 253  *
 254  *      Return the physical storage size (possibly compressed) of a varlena datum
 255  * ----------
 256  */
 257 Size
 258 toast_datum_size(Datum value)
 259 {
 260         varattrib  *attr = (varattrib *) DatumGetPointer(value);
 261         Size            result;
 262
 263         if (VARATT_IS_EXTERNAL(attr))
 264         {
 265                 /*
 266                  * Attribute is stored externally - return the extsize whether
 267                  * compressed or not.  We do not count the size of the toast pointer
 268                  * ... should we?
 269                  */
 270                 result = attr->va_content.va_external.va_extsize;
 271         }
 272         else
 273         {
 274                 /*
 275                  * Attribute is stored inline either compressed or not, just calculate
 276                  * the size of the datum in either case.
 277                  */
 278                 result = VARSIZE(attr);
 279         }
 280         return result;
 281 }
 282
 283
 284 /* ----------
 285  * toast_delete -
 286  *
 287  *      Cascaded delete toast-entries on DELETE
 288  * ----------
 289  */
 290 void
 291 toast_delete(Relation rel, HeapTuple oldtup)
 292 {
 293         TupleDesc       tupleDesc;
 294         Form_pg_attribute *att;
 295         int                     numAttrs;
 296         int                     i;
 297         Datum           toast_values[MaxHeapAttributeNumber];
 298         bool            toast_isnull[MaxHeapAttributeNumber];
 299
 300         /*
 301          * Get the tuple descriptor and break down the tuple into fields.
 302          *
 303          * NOTE: it's debatable whether to use heap_deformtuple() here or just
 304          * heap_getattr() only the varlena columns.  The latter could win if there
 305          * are few varlena columns and many non-varlena ones. However,
 306          * heap_deformtuple costs only O(N) while the heap_getattr way would cost
 307          * O(N^2) if there are many varlena columns, so it seems better to err on
 308          * the side of linear cost.  (We won't even be here unless there's at
 309          * least one varlena column, by the way.)
 310          */
 311         tupleDesc = rel->rd_att;
 312         att = tupleDesc->attrs;
 313         numAttrs = tupleDesc->natts;
 314
 315         Assert(numAttrs <= MaxHeapAttributeNumber);
 316         heap_deform_tuple(oldtup, tupleDesc, toast_values, toast_isnull);
 317
 318         /*
 319          * Check for external stored attributes and delete them from the secondary
 320          * relation.
 321          */
 322         for (i = 0; i < numAttrs; i++)
 323         {
 324                 if (att[i]->attlen == -1)
 325                 {
 326                         Datum           value = toast_values[i];
 327
 328                         if (!toast_isnull[i] && VARATT_IS_EXTERNAL(value))
 329                                 toast_delete_datum(rel, value);
 330                 }
 331         }
 332 }
 333
 334
 335 /* ----------
 336  * toast_insert_or_update -
 337  *
 338  *      Delete no-longer-used toast-entries and create new ones to
 339  *      make the new tuple fit on INSERT or UPDATE
 340  *
 341  * Inputs:
 342  *      newtup: the candidate new tuple to be inserted
 343  *      oldtup: the old row version for UPDATE, or NULL for INSERT
 344  * Result:
 345  *      either newtup if no toasting is needed, or a palloc'd modified tuple
 346  *      that is what should actually get stored
 347  *
 348  * NOTE: neither newtup nor oldtup will be modified.  This is a change
 349  * from the pre-8.1 API of this routine.
 350  * ----------
 351  */
 352 HeapTuple
 353 toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup)
 354 {
 355         HeapTuple       result_tuple;
 356         TupleDesc       tupleDesc;
 357         Form_pg_attribute *att;
 358         int                     numAttrs;
 359         int                     i;
 360
 361         bool            need_change = false;
 362         bool            need_free = false;
 363         bool            need_delold = false;
 364         bool            has_nulls = false;
 365
 366         Size            maxDataLen;
 367
 368         char            toast_action[MaxHeapAttributeNumber];
 369         bool            toast_isnull[MaxHeapAttributeNumber];
 370         bool            toast_oldisnull[MaxHeapAttributeNumber];
 371         Datum           toast_values[MaxHeapAttributeNumber];
 372         Datum           toast_oldvalues[MaxHeapAttributeNumber];
 373         int32           toast_sizes[MaxHeapAttributeNumber];
 374         bool            toast_free[MaxHeapAttributeNumber];
 375         bool            toast_delold[MaxHeapAttributeNumber];
 376
 377         /*
 378          * Get the tuple descriptor and break down the tuple(s) into fields.
 379          */
 380         tupleDesc = rel->rd_att;
 381         att = tupleDesc->attrs;
 382         numAttrs = tupleDesc->natts;
 383
 384         Assert(numAttrs <= MaxHeapAttributeNumber);
 385         heap_deform_tuple(newtup, tupleDesc, toast_values, toast_isnull);
 386         if (oldtup != NULL)
 387                 heap_deform_tuple(oldtup, tupleDesc, toast_oldvalues, toast_oldisnull);
 388
 389         /* ----------
 390          * Then collect information about the values given
 391          *
 392          * NOTE: toast_action[i] can have these values:
 393          *              ' '             default handling
 394          *              'p'             already processed --- don't touch it
 395          *              'x'             incompressible, but OK to move off
 396          *
 397          * NOTE: toast_sizes[i] is only made valid for varlena attributes with
 398          *              toast_action[i] different from 'p'.
 399          * ----------
 400          */
 401         memset(toast_action, ' ', numAttrs * sizeof(char));
 402         memset(toast_free, 0, numAttrs * sizeof(bool));
 403         memset(toast_delold, 0, numAttrs * sizeof(bool));
 404
 405         for (i = 0; i < numAttrs; i++)
 406         {
 407                 varattrib  *old_value;
 408                 varattrib  *new_value;
 409
 410                 if (oldtup != NULL)
 411                 {
 412                         /*
 413                          * For UPDATE get the old and new values of this attribute
 414                          */
 415                         old_value = (varattrib *) DatumGetPointer(toast_oldvalues[i]);
 416                         new_value = (varattrib *) DatumGetPointer(toast_values[i]);
 417
 418                         /*
 419                          * If the old value is an external stored one, check if it has
 420                          * changed so we have to delete it later.
 421                          */
 422                         if (att[i]->attlen == -1 && !toast_oldisnull[i] &&
 423                                 VARATT_IS_EXTERNAL(old_value))
 424                         {
 425                                 if (toast_isnull[i] || !VARATT_IS_EXTERNAL(new_value) ||
 426                                         old_value->va_content.va_external.va_valueid !=
 427                                         new_value->va_content.va_external.va_valueid ||
 428                                         old_value->va_content.va_external.va_toastrelid !=
 429                                         new_value->va_content.va_external.va_toastrelid)
 430                                 {
 431                                         /*
 432                                          * The old external stored value isn't needed any more
 433                                          * after the update
 434                                          */
 435                                         toast_delold[i] = true;
 436                                         need_delold = true;
 437                                 }
 438                                 else
 439                                 {
 440                                         /*
 441                                          * This attribute isn't changed by this update so we reuse
 442                                          * the original reference to the old value in the new
 443                                          * tuple.
 444                                          */
 445                                         toast_action[i] = 'p';
 446                                         toast_sizes[i] = VARATT_SIZE(toast_values[i]);
 447                                         continue;
 448                                 }
 449                         }
 450                 }
 451                 else
 452                 {
 453                         /*
 454                          * For INSERT simply get the new value
 455                          */
 456                         new_value = (varattrib *) DatumGetPointer(toast_values[i]);
 457                 }
 458
 459                 /*
 460                  * Handle NULL attributes
 461                  */
 462                 if (toast_isnull[i])
 463                 {
 464                         toast_action[i] = 'p';
 465                         has_nulls = true;
 466                         continue;
 467                 }
 468
 469                 /*
 470                  * Now look at varlena attributes
 471                  */
 472                 if (att[i]->attlen == -1)
 473                 {
 474                         /*
 475                          * If the table's attribute says PLAIN always, force it so.
 476                          */
 477                         if (att[i]->attstorage == 'p')
 478                                 toast_action[i] = 'p';
 479
 480                         /*
 481                          * We took care of UPDATE above, so any external value we find
 482                          * still in the tuple must be someone else's we cannot reuse.
 483                          * Expand it to plain (and, probably, toast it again below).
 484                          */
 485                         if (VARATT_IS_EXTERNAL(new_value))
 486                         {
 487                                 new_value = heap_tuple_untoast_attr(new_value);
 488                                 toast_values[i] = PointerGetDatum(new_value);
 489                                 toast_free[i] = true;
 490                                 need_change = true;
 491                                 need_free = true;
 492                         }
 493
 494                         /*
 495                          * Remember the size of this attribute
 496                          */
 497                         toast_sizes[i] = VARATT_SIZE(new_value);
 498                 }
 499                 else
 500                 {
 501                         /*
 502                          * Not a varlena attribute, plain storage always
 503                          */
 504                         toast_action[i] = 'p';
 505                 }
 506         }
 507
 508         /* ----------
 509          * Compress and/or save external until data fits into target length
 510          *
 511          *      1: Inline compress attributes with attstorage 'x'
 512          *      2: Store attributes with attstorage 'x' or 'e' external
 513          *      3: Inline compress attributes with attstorage 'm'
 514          *      4: Store attributes with attstorage 'm' external
 515          * ----------
 516          */
 517         maxDataLen = offsetof(HeapTupleHeaderData, t_bits);
 518         if (has_nulls)
 519                 maxDataLen += BITMAPLEN(numAttrs);
 520         maxDataLen = TOAST_TUPLE_TARGET - MAXALIGN(maxDataLen);
 521
 522         /*
 523          * Look for attributes with attstorage 'x' to compress
 524          */
 525         while (MAXALIGN(heap_compute_data_size(tupleDesc,
 526                                                                                    toast_values, toast_isnull)) >
 527                    maxDataLen)
 528         {
 529                 int                     biggest_attno = -1;
 530                 int32           biggest_size = MAXALIGN(sizeof(varattrib));
 531                 Datum           old_value;
 532                 Datum           new_value;
 533
 534                 /*
 535                  * Search for the biggest yet uncompressed internal attribute
 536                  */
 537                 for (i = 0; i < numAttrs; i++)
 538                 {
 539                         if (toast_action[i] != ' ')
 540                                 continue;
 541                         if (VARATT_IS_EXTENDED(toast_values[i]))
 542                                 continue;
 543                         if (att[i]->attstorage != 'x')
 544                                 continue;
 545                         if (toast_sizes[i] > biggest_size)
 546                         {
 547                                 biggest_attno = i;
 548                                 biggest_size = toast_sizes[i];
 549                         }
 550                 }
 551
 552                 if (biggest_attno < 0)
 553                         break;
 554
 555                 /*
 556                  * Attempt to compress it inline
 557                  */
 558                 i = biggest_attno;
 559                 old_value = toast_values[i];
 560                 new_value = toast_compress_datum(old_value);
 561
 562                 if (DatumGetPointer(new_value) != NULL)
 563                 {
 564                         /* successful compression */
 565                         if (toast_free[i])
 566                                 pfree(DatumGetPointer(old_value));
 567                         toast_values[i] = new_value;
 568                         toast_free[i] = true;
 569                         toast_sizes[i] = VARATT_SIZE(toast_values[i]);
 570                         need_change = true;
 571                         need_free = true;
 572                 }
 573                 else
 574                 {
 575                         /*
 576                          * incompressible data, ignore on subsequent compression passes
 577                          */
 578                         toast_action[i] = 'x';
 579                 }
 580         }
 581
 582         /*
 583          * Second we look for attributes of attstorage 'x' or 'e' that are still
 584          * inline.
 585          */
 586         while (MAXALIGN(heap_compute_data_size(tupleDesc,
 587                                                                                    toast_values, toast_isnull)) >
 588                    maxDataLen && rel->rd_rel->reltoastrelid != InvalidOid)
 589         {
 590                 int                     biggest_attno = -1;
 591                 int32           biggest_size = MAXALIGN(sizeof(varattrib));
 592                 Datum           old_value;
 593
 594                 /*------
 595                  * Search for the biggest yet inlined attribute with
 596                  * attstorage equals 'x' or 'e'
 597                  *------
 598                  */
 599                 for (i = 0; i < numAttrs; i++)
 600                 {
 601                         if (toast_action[i] == 'p')
 602                                 continue;
 603                         if (VARATT_IS_EXTERNAL(toast_values[i]))
 604                                 continue;
 605                         if (att[i]->attstorage != 'x' && att[i]->attstorage != 'e')
 606                                 continue;
 607                         if (toast_sizes[i] > biggest_size)
 608                         {
 609                                 biggest_attno = i;
 610                                 biggest_size = toast_sizes[i];
 611                         }
 612                 }
 613
 614                 if (biggest_attno < 0)
 615                         break;
 616
 617                 /*
 618                  * Store this external
 619                  */
 620                 i = biggest_attno;
 621                 old_value = toast_values[i];
 622                 toast_action[i] = 'p';
 623                 toast_values[i] = toast_save_datum(rel, toast_values[i]);
 624                 if (toast_free[i])
 625                         pfree(DatumGetPointer(old_value));
 626
 627                 toast_free[i] = true;
 628                 toast_sizes[i] = VARATT_SIZE(toast_values[i]);
 629
 630                 need_change = true;
 631                 need_free = true;
 632         }
 633
 634         /*
 635          * Round 3 - this time we take attributes with storage 'm' into
 636          * compression
 637          */
 638         while (MAXALIGN(heap_compute_data_size(tupleDesc,
 639                                                                                    toast_values, toast_isnull)) >
 640                    maxDataLen)
 641         {
 642                 int                     biggest_attno = -1;
 643                 int32           biggest_size = MAXALIGN(sizeof(varattrib));
 644                 Datum           old_value;
 645                 Datum           new_value;
 646
 647                 /*
 648                  * Search for the biggest yet uncompressed internal attribute
 649                  */
 650                 for (i = 0; i < numAttrs; i++)
 651                 {
 652                         if (toast_action[i] != ' ')
 653                                 continue;
 654                         if (VARATT_IS_EXTENDED(toast_values[i]))
 655                                 continue;
 656                         if (att[i]->attstorage != 'm')
 657                                 continue;
 658                         if (toast_sizes[i] > biggest_size)
 659                         {
 660                                 biggest_attno = i;
 661                                 biggest_size = toast_sizes[i];
 662                         }
 663                 }
 664
 665                 if (biggest_attno < 0)
 666                         break;
 667
 668                 /*
 669                  * Attempt to compress it inline
 670                  */
 671                 i = biggest_attno;
 672                 old_value = toast_values[i];
 673                 new_value = toast_compress_datum(old_value);
 674
 675                 if (DatumGetPointer(new_value) != NULL)
 676                 {
 677                         /* successful compression */
 678                         if (toast_free[i])
 679                                 pfree(DatumGetPointer(old_value));
 680                         toast_values[i] = new_value;
 681                         toast_free[i] = true;
 682                         toast_sizes[i] = VARATT_SIZE(toast_values[i]);
 683                         need_change = true;
 684                         need_free = true;
 685                 }
 686                 else
 687                 {
 688                         /*
 689                          * incompressible data, ignore on subsequent compression passes
 690                          */
 691                         toast_action[i] = 'x';
 692                 }
 693         }
 694
 695         /*
 696          * Finally we store attributes of type 'm' external
 697          */
 698         while (MAXALIGN(heap_compute_data_size(tupleDesc,
 699                                                                                    toast_values, toast_isnull)) >
 700                    maxDataLen && rel->rd_rel->reltoastrelid != InvalidOid)
 701         {
 702                 int                     biggest_attno = -1;
 703                 int32           biggest_size = MAXALIGN(sizeof(varattrib));
 704                 Datum           old_value;
 705
 706                 /*--------
 707                  * Search for the biggest yet inlined attribute with
 708                  * attstorage = 'm'
 709                  *--------
 710                  */
 711                 for (i = 0; i < numAttrs; i++)
 712                 {
 713                         if (toast_action[i] == 'p')
 714                                 continue;
 715                         if (VARATT_IS_EXTERNAL(toast_values[i]))
 716                                 continue;
 717                         if (att[i]->attstorage != 'm')
 718                                 continue;
 719                         if (toast_sizes[i] > biggest_size)
 720                         {
 721                                 biggest_attno = i;
 722                                 biggest_size = toast_sizes[i];
 723                         }
 724                 }
 725
 726                 if (biggest_attno < 0)
 727                         break;
 728
 729                 /*
 730                  * Store this external
 731                  */
 732                 i = biggest_attno;
 733                 old_value = toast_values[i];
 734                 toast_action[i] = 'p';
 735                 toast_values[i] = toast_save_datum(rel, toast_values[i]);
 736                 if (toast_free[i])
 737                         pfree(DatumGetPointer(old_value));
 738
 739                 toast_free[i] = true;
 740                 toast_sizes[i] = VARATT_SIZE(toast_values[i]);
 741
 742                 need_change = true;
 743                 need_free = true;
 744         }
 745
 746         /*
 747          * In the case we toasted any values, we need to build a new heap tuple
 748          * with the changed values.
 749          */
 750         if (need_change)
 751         {
 752                 HeapTupleHeader olddata = newtup->t_data;
 753                 HeapTupleHeader new_data;
 754                 int32           new_len;
 755
 756                 /*
 757                  * Calculate the new size of the tuple.  Header size should not
 758                  * change, but data size might.
 759                  */
 760                 new_len = offsetof(HeapTupleHeaderData, t_bits);
 761                 if (has_nulls)
 762                         new_len += BITMAPLEN(numAttrs);
 763                 if (olddata->t_infomask & HEAP_HASOID)
 764                         new_len += sizeof(Oid);
 765                 new_len = MAXALIGN(new_len);
 766                 Assert(new_len == olddata->t_hoff);
 767                 new_len += heap_compute_data_size(tupleDesc,
 768                                                                                   toast_values, toast_isnull);
 769
 770                 /*
 771                  * Allocate and zero the space needed, and fill HeapTupleData fields.
 772                  */
 773                 result_tuple = (HeapTuple) palloc0(HEAPTUPLESIZE + new_len);
 774                 result_tuple->t_len = new_len;
 775                 result_tuple->t_self = newtup->t_self;
 776                 result_tuple->t_tableOid = newtup->t_tableOid;
 777                 new_data = (HeapTupleHeader) ((char *) result_tuple + HEAPTUPLESIZE);
 778                 result_tuple->t_data = new_data;
 779
 780                 /*
 781                  * Put the existing tuple header and the changed values into place
 782                  */
 783                 memcpy(new_data, olddata, olddata->t_hoff);
 784
 785                 heap_fill_tuple(tupleDesc,
 786                                                 toast_values,
 787                                                 toast_isnull,
 788                                                 (char *) new_data + olddata->t_hoff,
 789                                                 &(new_data->t_infomask),
 790                                                 has_nulls ? new_data->t_bits : NULL);
 791         }
 792         else
 793                 result_tuple = newtup;
 794
 795         /*
 796          * Free allocated temp values
 797          */
 798         if (need_free)
 799                 for (i = 0; i < numAttrs; i++)
 800                         if (toast_free[i])
 801                                 pfree(DatumGetPointer(toast_values[i]));
 802
 803         /*
 804          * Delete external values from the old tuple
 805          */
 806         if (need_delold)
 807                 for (i = 0; i < numAttrs; i++)
 808                         if (toast_delold[i])
 809                                 toast_delete_datum(rel, toast_oldvalues[i]);
 810
 811         return result_tuple;
 812 }
 813
 814
 815 /* ----------
 816  * toast_flatten_tuple_attribute -
 817  *
 818  *      If a Datum is of composite type, "flatten" it to contain no toasted fields.
 819  *      This must be invoked on any potentially-composite field that is to be
 820  *      inserted into a tuple.  Doing this preserves the invariant that toasting
 821  *      goes only one level deep in a tuple.
 822  * ----------
 823  */
 824 Datum
 825 toast_flatten_tuple_attribute(Datum value,
 826                                                           Oid typeId, int32 typeMod)
 827 {
 828         TupleDesc       tupleDesc;
 829         HeapTupleHeader olddata;
 830         HeapTupleHeader new_data;
 831         int32           new_len;
 832         HeapTupleData tmptup;
 833         Form_pg_attribute *att;
 834         int                     numAttrs;
 835         int                     i;
 836         bool            need_change = false;
 837         bool            has_nulls = false;
 838         Datum           toast_values[MaxTupleAttributeNumber];
 839         bool            toast_isnull[MaxTupleAttributeNumber];
 840         bool            toast_free[MaxTupleAttributeNumber];
 841
 842         /*
 843          * See if it's a composite type, and get the tupdesc if so.
 844          */
 845         tupleDesc = lookup_rowtype_tupdesc_noerror(typeId, typeMod, true);
 846         if (tupleDesc == NULL)
 847                 return value;                   /* not a composite type */
 848
 849         att = tupleDesc->attrs;
 850         numAttrs = tupleDesc->natts;
 851
 852         /*
 853          * Break down the tuple into fields.
 854          */
 855         olddata = DatumGetHeapTupleHeader(value);
 856         Assert(typeId == HeapTupleHeaderGetTypeId(olddata));
 857         Assert(typeMod == HeapTupleHeaderGetTypMod(olddata));
 858         /* Build a temporary HeapTuple control structure */
 859         tmptup.t_len = HeapTupleHeaderGetDatumLength(olddata);
 860         ItemPointerSetInvalid(&(tmptup.t_self));
 861         tmptup.t_tableOid = InvalidOid;
 862         tmptup.t_data = olddata;
 863
 864         Assert(numAttrs <= MaxTupleAttributeNumber);
 865         heap_deform_tuple(&tmptup, tupleDesc, toast_values, toast_isnull);
 866
 867         memset(toast_free, 0, numAttrs * sizeof(bool));
 868
 869         for (i = 0; i < numAttrs; i++)
 870         {
 871                 /*
 872                  * Look at non-null varlena attributes
 873                  */
 874                 if (toast_isnull[i])
 875                         has_nulls = true;
 876                 else if (att[i]->attlen == -1)
 877                 {
 878                         varattrib  *new_value;
 879
 880                         new_value = (varattrib *) DatumGetPointer(toast_values[i]);
 881                         if (VARATT_IS_EXTENDED(new_value))
 882                         {
 883                                 new_value = heap_tuple_untoast_attr(new_value);
 884                                 toast_values[i] = PointerGetDatum(new_value);
 885                                 toast_free[i] = true;
 886                                 need_change = true;
 887                         }
 888                 }
 889         }
 890
 891         /*
 892          * If nothing to untoast, just return the original tuple.
 893          */
 894         if (!need_change)
 895                 return value;
 896
 897         /*
 898          * Calculate the new size of the tuple.  Header size should not change,
 899          * but data size might.
 900          */
 901         new_len = offsetof(HeapTupleHeaderData, t_bits);
 902         if (has_nulls)
 903                 new_len += BITMAPLEN(numAttrs);
 904         if (olddata->t_infomask & HEAP_HASOID)
 905                 new_len += sizeof(Oid);
 906         new_len = MAXALIGN(new_len);
 907         Assert(new_len == olddata->t_hoff);
 908         new_len += heap_compute_data_size(tupleDesc, toast_values, toast_isnull);
 909
 910         new_data = (HeapTupleHeader) palloc0(new_len);
 911
 912         /*
 913          * Put the tuple header and the changed values into place
 914          */
 915         memcpy(new_data, olddata, olddata->t_hoff);
 916
 917         HeapTupleHeaderSetDatumLength(new_data, new_len);
 918
 919         heap_fill_tuple(tupleDesc,
 920                                         toast_values,
 921                                         toast_isnull,
 922                                         (char *) new_data + olddata->t_hoff,
 923                                         &(new_data->t_infomask),
 924                                         has_nulls ? new_data->t_bits : NULL);
 925
 926         /*
 927          * Free allocated temp values
 928          */
 929         for (i = 0; i < numAttrs; i++)
 930                 if (toast_free[i])
 931                         pfree(DatumGetPointer(toast_values[i]));
 932
 933         return PointerGetDatum(new_data);
 934 }
 935
 936
 937 /* ----------
 938  * toast_compress_datum -
 939  *
 940  *      Create a compressed version of a varlena datum
 941  *
 942  *      If we fail (ie, compressed result is actually bigger than original)
 943  *      then return NULL.  We must not use compressed data if it'd expand
 944  *      the tuple!
 945  * ----------
 946  */
 947 Datum
 948 toast_compress_datum(Datum value)
 949 {
 950         varattrib  *tmp;
 951
 952         tmp = (varattrib *) palloc(sizeof(PGLZ_Header) + VARATT_SIZE(value));
 953         pglz_compress(VARATT_DATA(value), VARATT_SIZE(value) - VARHDRSZ,
 954                                   (PGLZ_Header *) tmp,
 955                                   PGLZ_strategy_default);
 956         if (VARATT_SIZE(tmp) < VARATT_SIZE(value))
 957         {
 958                 /* successful compression */
 959                 VARATT_SIZEP(tmp) |= VARATT_FLAG_COMPRESSED;
 960                 return PointerGetDatum(tmp);
 961         }
 962         else
 963         {
 964                 /* incompressible data */
 965                 pfree(tmp);
 966                 return PointerGetDatum(NULL);
 967         }
 968 }
 969
 970
 971 /* ----------
 972  * toast_save_datum -
 973  *
 974  *      Save one single datum into the secondary relation and return
 975  *      a varattrib reference for it.
 976  * ----------
 977  */
 978 static Datum
 979 toast_save_datum(Relation rel, Datum value)
 980 {
 981         Relation        toastrel;
 982         Relation        toastidx;
 983         HeapTuple       toasttup;
 984         TupleDesc       toasttupDesc;
 985         Datum           t_values[3];
 986         bool            t_isnull[3];
 987         varattrib  *result;
 988         struct
 989         {
 990                 struct varlena hdr;
 991                 char            data[TOAST_MAX_CHUNK_SIZE];
 992         }                       chunk_data;
 993         int32           chunk_size;
 994         int32           chunk_seq = 0;
 995         char       *data_p;
 996         int32           data_todo;
 997
 998         /*
 999          * Open the toast relation and its index.  We can use the index to check
1000          * uniqueness of the OID we assign to the toasted item, even though it has
1001          * additional columns besides OID.
1002          */
1003         toastrel = heap_open(rel->rd_rel->reltoastrelid, RowExclusiveLock);
1004         toasttupDesc = toastrel->rd_att;
1005         toastidx = index_open(toastrel->rd_rel->reltoastidxid);
1006
1007         /*
1008          * Create the varattrib reference
1009          */
1010         result = (varattrib *) palloc(sizeof(varattrib));
1011
1012         result->va_header = sizeof(varattrib) | VARATT_FLAG_EXTERNAL;
1013         if (VARATT_IS_COMPRESSED(value))
1014         {
1015                 result->va_header |= VARATT_FLAG_COMPRESSED;
1016                 result->va_content.va_external.va_rawsize =
1017                         ((varattrib *) value)->va_content.va_compressed.va_rawsize;
1018         }
1019         else
1020                 result->va_content.va_external.va_rawsize = VARATT_SIZE(value);
1021
1022         result->va_content.va_external.va_extsize =
1023                 VARATT_SIZE(value) - VARHDRSZ;
1024         result->va_content.va_external.va_valueid =
1025                 GetNewOidWithIndex(toastrel, toastidx);
1026         result->va_content.va_external.va_toastrelid =
1027                 rel->rd_rel->reltoastrelid;
1028
1029         /*
1030          * Initialize constant parts of the tuple data
1031          */
1032         t_values[0] = ObjectIdGetDatum(result->va_content.va_external.va_valueid);
1033         t_values[2] = PointerGetDatum(&chunk_data);
1034         t_isnull[0] = false;
1035         t_isnull[1] = false;
1036         t_isnull[2] = false;
1037
1038         /*
1039          * Get the data to process
1040          */
1041         data_p = VARATT_DATA(value);
1042         data_todo = VARATT_SIZE(value) - VARHDRSZ;
1043
1044         /*
1045          * We must explicitly lock the toast index because we aren't using an
1046          * index scan here.
1047          */
1048         LockRelation(toastidx, RowExclusiveLock);
1049
1050         /*
1051          * Split up the item into chunks
1052          */
1053         while (data_todo > 0)
1054         {
1055                 /*
1056                  * Calculate the size of this chunk
1057                  */
1058                 chunk_size = Min(TOAST_MAX_CHUNK_SIZE, data_todo);
1059
1060                 /*
1061                  * Build a tuple and store it
1062                  */
1063                 t_values[1] = Int32GetDatum(chunk_seq++);
1064                 VARATT_SIZEP(&chunk_data) = chunk_size + VARHDRSZ;
1065                 memcpy(VARATT_DATA(&chunk_data), data_p, chunk_size);
1066                 toasttup = heap_form_tuple(toasttupDesc, t_values, t_isnull);
1067                 if (!HeapTupleIsValid(toasttup))
1068                         elog(ERROR, "failed to build TOAST tuple");
1069
1070                 simple_heap_insert(toastrel, toasttup);
1071
1072                 /*
1073                  * Create the index entry.      We cheat a little here by not using
1074                  * FormIndexDatum: this relies on the knowledge that the index columns
1075                  * are the same as the initial columns of the table.
1076                  *
1077                  * Note also that there had better not be any user-created index on
1078                  * the TOAST table, since we don't bother to update anything else.
1079                  */
1080                 index_insert(toastidx, t_values, t_isnull,
1081                                          &(toasttup->t_self),
1082                                          toastrel, toastidx->rd_index->indisunique);
1083
1084                 /*
1085                  * Free memory
1086                  */
1087                 heap_freetuple(toasttup);
1088
1089                 /*
1090                  * Move on to next chunk
1091                  */
1092                 data_todo -= chunk_size;
1093                 data_p += chunk_size;
1094         }
1095
1096         /*
1097          * Done - close toast relation and return the reference
1098          */
1099         UnlockRelation(toastidx, RowExclusiveLock);
1100         index_close(toastidx);
1101         heap_close(toastrel, RowExclusiveLock);
1102
1103         return PointerGetDatum(result);
1104 }
1105
1106
1107 /* ----------
1108  * toast_delete_datum -
1109  *
1110  *      Delete a single external stored value.
1111  * ----------
1112  */
1113 static void
1114 toast_delete_datum(Relation rel, Datum value)
1115 {
1116         varattrib  *attr = (varattrib *) DatumGetPointer(value);
1117         Relation        toastrel;
1118         Relation        toastidx;
1119         ScanKeyData toastkey;
1120         IndexScanDesc toastscan;
1121         HeapTuple       toasttup;
1122
1123         if (!VARATT_IS_EXTERNAL(attr))
1124                 return;
1125
1126         /*
1127          * Open the toast relation and it's index
1128          */
1129         toastrel = heap_open(attr->va_content.va_external.va_toastrelid,
1130                                                  RowExclusiveLock);
1131         toastidx = index_open(toastrel->rd_rel->reltoastidxid);
1132
1133         /*
1134          * Setup a scan key to fetch from the index by va_valueid (we don't
1135          * particularly care whether we see them in sequence or not)
1136          */
1137         ScanKeyInit(&toastkey,
1138                                 (AttrNumber) 1,
1139                                 BTEqualStrategyNumber, F_OIDEQ,
1140                                 ObjectIdGetDatum(attr->va_content.va_external.va_valueid));
1141
1142         /*
1143          * Find the chunks by index
1144          */
1145         toastscan = index_beginscan(toastrel, toastidx, true,
1146                                                                 SnapshotToast, 1, &toastkey);
1147         while ((toasttup = index_getnext(toastscan, ForwardScanDirection)) != NULL)
1148         {
1149                 /*
1150                  * Have a chunk, delete it
1151                  */
1152                 simple_heap_delete(toastrel, &toasttup->t_self);
1153         }
1154
1155         /*
1156          * End scan and close relations
1157          */
1158         index_endscan(toastscan);
1159         index_close(toastidx);
1160         heap_close(toastrel, RowExclusiveLock);
1161 }
1162
1163
1164 /* ----------
1165  * toast_fetch_datum -
1166  *
1167  *      Reconstruct an in memory varattrib from the chunks saved
1168  *      in the toast relation
1169  * ----------
1170  */
1171 static varattrib *
1172 toast_fetch_datum(varattrib *attr)
1173 {
1174         Relation        toastrel;
1175         Relation        toastidx;
1176         ScanKeyData toastkey;
1177         IndexScanDesc toastscan;
1178         HeapTuple       ttup;
1179         TupleDesc       toasttupDesc;
1180         varattrib  *result;
1181         int32           ressize;
1182         int32           residx,
1183                                 nextidx;
1184         int32           numchunks;
1185         Pointer         chunk;
1186         bool            isnull;
1187         int32           chunksize;
1188
1189         ressize = attr->va_content.va_external.va_extsize;
1190         numchunks = ((ressize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
1191
1192         result = (varattrib *) palloc(ressize + VARHDRSZ);
1193         VARATT_SIZEP(result) = ressize + VARHDRSZ;
1194         if (VARATT_IS_COMPRESSED(attr))
1195                 VARATT_SIZEP(result) |= VARATT_FLAG_COMPRESSED;
1196
1197         /*
1198          * Open the toast relation and its index
1199          */
1200         toastrel = heap_open(attr->va_content.va_external.va_toastrelid,
1201                                                  AccessShareLock);
1202         toasttupDesc = toastrel->rd_att;
1203         toastidx = index_open(toastrel->rd_rel->reltoastidxid);
1204
1205         /*
1206          * Setup a scan key to fetch from the index by va_valueid
1207          */
1208         ScanKeyInit(&toastkey,
1209                                 (AttrNumber) 1,
1210                                 BTEqualStrategyNumber, F_OIDEQ,
1211                                 ObjectIdGetDatum(attr->va_content.va_external.va_valueid));
1212
1213         /*
1214          * Read the chunks by index
1215          *
1216          * Note that because the index is actually on (valueid, chunkidx) we will
1217          * see the chunks in chunkidx order, even though we didn't explicitly ask
1218          * for it.
1219          */
1220         nextidx = 0;
1221
1222         toastscan = index_beginscan(toastrel, toastidx, true,
1223                                                                 SnapshotToast, 1, &toastkey);
1224         while ((ttup = index_getnext(toastscan, ForwardScanDirection)) != NULL)
1225         {
1226                 /*
1227                  * Have a chunk, extract the sequence number and the data
1228                  */
1229                 residx = DatumGetInt32(fastgetattr(ttup, 2, toasttupDesc, &isnull));
1230                 Assert(!isnull);
1231                 chunk = DatumGetPointer(fastgetattr(ttup, 3, toasttupDesc, &isnull));
1232                 Assert(!isnull);
1233                 chunksize = VARATT_SIZE(chunk) - VARHDRSZ;
1234
1235                 /*
1236                  * Some checks on the data we've found
1237                  */
1238                 if (residx != nextidx)
1239                         elog(ERROR, "unexpected chunk number %d (expected %d) for toast value %u",
1240                                  residx, nextidx,
1241                                  attr->va_content.va_external.va_valueid);
1242                 if (residx < numchunks - 1)
1243                 {
1244                         if (chunksize != TOAST_MAX_CHUNK_SIZE)
1245                                 elog(ERROR, "unexpected chunk size %d in chunk %d for toast value %u",
1246                                          chunksize, residx,
1247                                          attr->va_content.va_external.va_valueid);
1248                 }
1249                 else if (residx < numchunks)
1250                 {
1251                         if ((residx * TOAST_MAX_CHUNK_SIZE + chunksize) != ressize)
1252                                 elog(ERROR, "unexpected chunk size %d in chunk %d for toast value %u",
1253                                          chunksize, residx,
1254                                          attr->va_content.va_external.va_valueid);
1255                 }
1256                 else
1257                         elog(ERROR, "unexpected chunk number %d for toast value %u",
1258                                  residx,
1259                                  attr->va_content.va_external.va_valueid);
1260
1261                 /*
1262                  * Copy the data into proper place in our result
1263                  */
1264                 memcpy(((char *) VARATT_DATA(result)) + residx * TOAST_MAX_CHUNK_SIZE,
1265                            VARATT_DATA(chunk),
1266                            chunksize);
1267
1268                 nextidx++;
1269         }
1270
1271         /*
1272          * Final checks that we successfully fetched the datum
1273          */
1274         if (nextidx != numchunks)
1275                 elog(ERROR, "missing chunk number %d for toast value %u",
1276                          nextidx,
1277                          attr->va_content.va_external.va_valueid);
1278
1279         /*
1280          * End scan and close relations
1281          */
1282         index_endscan(toastscan);
1283         index_close(toastidx);
1284         heap_close(toastrel, AccessShareLock);
1285
1286         return result;
1287 }
1288
1289 /* ----------
1290  * toast_fetch_datum_slice -
1291  *
1292  *      Reconstruct a segment of a varattrib from the chunks saved
1293  *      in the toast relation
1294  * ----------
1295  */
1296 static varattrib *
1297 toast_fetch_datum_slice(varattrib *attr, int32 sliceoffset, int32 length)
1298 {
1299         Relation        toastrel;
1300         Relation        toastidx;
1301         ScanKeyData toastkey[3];
1302         int                     nscankeys;
1303         IndexScanDesc toastscan;
1304         HeapTuple       ttup;
1305         TupleDesc       toasttupDesc;
1306         varattrib  *result;
1307         int32           attrsize;
1308         int32           residx;
1309         int32           nextidx;
1310         int                     numchunks;
1311         int                     startchunk;
1312         int                     endchunk;
1313         int32           startoffset;
1314         int32           endoffset;
1315         int                     totalchunks;
1316         Pointer         chunk;
1317         bool            isnull;
1318         int32           chunksize;
1319         int32           chcpystrt;
1320         int32           chcpyend;
1321
1322         attrsize = attr->va_content.va_external.va_extsize;
1323         totalchunks = ((attrsize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
1324
1325         if (sliceoffset >= attrsize)
1326         {
1327                 sliceoffset = 0;
1328                 length = 0;
1329         }
1330
1331         if (((sliceoffset + length) > attrsize) || length < 0)
1332                 length = attrsize - sliceoffset;
1333
1334         result = (varattrib *) palloc(length + VARHDRSZ);
1335         VARATT_SIZEP(result) = length + VARHDRSZ;
1336
1337         if (VARATT_IS_COMPRESSED(attr))
1338                 VARATT_SIZEP(result) |= VARATT_FLAG_COMPRESSED;
1339
1340         if (length == 0)
1341                 return (result);                /* Can save a lot of work at this point! */
1342
1343         startchunk = sliceoffset / TOAST_MAX_CHUNK_SIZE;
1344         endchunk = (sliceoffset + length - 1) / TOAST_MAX_CHUNK_SIZE;
1345         numchunks = (endchunk - startchunk) + 1;
1346
1347         startoffset = sliceoffset % TOAST_MAX_CHUNK_SIZE;
1348         endoffset = (sliceoffset + length - 1) % TOAST_MAX_CHUNK_SIZE;
1349
1350         /*
1351          * Open the toast relation and it's index
1352          */
1353         toastrel = heap_open(attr->va_content.va_external.va_toastrelid,
1354                                                  AccessShareLock);
1355         toasttupDesc = toastrel->rd_att;
1356         toastidx = index_open(toastrel->rd_rel->reltoastidxid);
1357
1358         /*
1359          * Setup a scan key to fetch from the index. This is either two keys or
1360          * three depending on the number of chunks.
1361          */
1362         ScanKeyInit(&toastkey[0],
1363                                 (AttrNumber) 1,
1364                                 BTEqualStrategyNumber, F_OIDEQ,
1365                                 ObjectIdGetDatum(attr->va_content.va_external.va_valueid));
1366
1367         /*
1368          * Use equality condition for one chunk, a range condition otherwise:
1369          */
1370         if (numchunks == 1)
1371         {
1372                 ScanKeyInit(&toastkey[1],
1373                                         (AttrNumber) 2,
1374                                         BTEqualStrategyNumber, F_INT4EQ,
1375                                         Int32GetDatum(startchunk));
1376                 nscankeys = 2;
1377         }
1378         else
1379         {
1380                 ScanKeyInit(&toastkey[1],
1381                                         (AttrNumber) 2,
1382                                         BTGreaterEqualStrategyNumber, F_INT4GE,
1383                                         Int32GetDatum(startchunk));
1384                 ScanKeyInit(&toastkey[2],
1385                                         (AttrNumber) 2,
1386                                         BTLessEqualStrategyNumber, F_INT4LE,
1387                                         Int32GetDatum(endchunk));
1388                 nscankeys = 3;
1389         }
1390
1391         /*
1392          * Read the chunks by index
1393          *
1394          * The index is on (valueid, chunkidx) so they will come in order
1395          */
1396         nextidx = startchunk;
1397         toastscan = index_beginscan(toastrel, toastidx, true,
1398                                                                 SnapshotToast, nscankeys, toastkey);
1399         while ((ttup = index_getnext(toastscan, ForwardScanDirection)) != NULL)
1400         {
1401                 /*
1402                  * Have a chunk, extract the sequence number and the data
1403                  */
1404                 residx = DatumGetInt32(fastgetattr(ttup, 2, toasttupDesc, &isnull));
1405                 Assert(!isnull);
1406                 chunk = DatumGetPointer(fastgetattr(ttup, 3, toasttupDesc, &isnull));
1407                 Assert(!isnull);
1408                 chunksize = VARATT_SIZE(chunk) - VARHDRSZ;
1409
1410                 /*
1411                  * Some checks on the data we've found
1412                  */
1413                 if ((residx != nextidx) || (residx > endchunk) || (residx < startchunk))
1414                         elog(ERROR, "unexpected chunk number %d (expected %d) for toast value %u",
1415                                  residx, nextidx,
1416                                  attr->va_content.va_external.va_valueid);
1417                 if (residx < totalchunks - 1)
1418                 {
1419                         if (chunksize != TOAST_MAX_CHUNK_SIZE)
1420                                 elog(ERROR, "unexpected chunk size %d in chunk %d for toast value %u",
1421                                          chunksize, residx,
1422                                          attr->va_content.va_external.va_valueid);
1423                 }
1424                 else
1425                 {
1426                         if ((residx * TOAST_MAX_CHUNK_SIZE + chunksize) != attrsize)
1427                                 elog(ERROR, "unexpected chunk size %d in chunk %d for toast value %u",
1428                                          chunksize, residx,
1429                                          attr->va_content.va_external.va_valueid);
1430                 }
1431
1432                 /*
1433                  * Copy the data into proper place in our result
1434                  */
1435                 chcpystrt = 0;
1436                 chcpyend = chunksize - 1;
1437                 if (residx == startchunk)
1438                         chcpystrt = startoffset;
1439                 if (residx == endchunk)
1440                         chcpyend = endoffset;
1441
1442                 memcpy(((char *) VARATT_DATA(result)) +
1443                            (residx * TOAST_MAX_CHUNK_SIZE - sliceoffset) + chcpystrt,
1444                            VARATT_DATA(chunk) + chcpystrt,
1445                            (chcpyend - chcpystrt) + 1);
1446
1447                 nextidx++;
1448         }
1449
1450         /*
1451          * Final checks that we successfully fetched the datum
1452          */
1453         if (nextidx != (endchunk + 1))
1454                 elog(ERROR, "missing chunk number %d for toast value %u",
1455                          nextidx,
1456                          attr->va_content.va_external.va_valueid);
1457
1458         /*
1459          * End scan and close relations
1460          */
1461         index_endscan(toastscan);
1462         index_close(toastidx);
1463         heap_close(toastrel, AccessShareLock);
1464
1465         return result;
1466 }