granicus.if.org Git - postgresql/blob - src/backend/access/gin/ginutil.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * ginutil.c
   4  *        Utility routines for the Postgres inverted index access method.
   5  *
   6  *
   7  * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
   8  * Portions Copyright (c) 1994, Regents of the University of California
   9  *
  10  * IDENTIFICATION
  11  *                      src/backend/access/gin/ginutil.c
  12  *-------------------------------------------------------------------------
  13  */
  14
  15 #include "postgres.h"
  16
  17 #include "access/gin_private.h"
  18 #include "access/reloptions.h"
  19 #include "access/xloginsert.h"
  20 #include "catalog/pg_collation.h"
  21 #include "catalog/pg_type.h"
  22 #include "miscadmin.h"
  23 #include "storage/indexfsm.h"
  24 #include "storage/lmgr.h"
  25 #include "utils/builtins.h"
  26 #include "utils/index_selfuncs.h"
  27 #include "utils/typcache.h"
  28
  29
  30 /*
  31  * GIN handler function: return IndexAmRoutine with access method parameters
  32  * and callbacks.
  33  */
  34 Datum
  35 ginhandler(PG_FUNCTION_ARGS)
  36 {
  37         IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);
  38
  39         amroutine->amstrategies = 0;
  40         amroutine->amsupport = GINNProcs;
  41         amroutine->amcanorder = false;
  42         amroutine->amcanorderbyop = false;
  43         amroutine->amcanbackward = false;
  44         amroutine->amcanunique = false;
  45         amroutine->amcanmulticol = true;
  46         amroutine->amoptionalkey = true;
  47         amroutine->amsearcharray = false;
  48         amroutine->amsearchnulls = false;
  49         amroutine->amstorage = true;
  50         amroutine->amclusterable = false;
  51         amroutine->ampredlocks = false;
  52         amroutine->amkeytype = InvalidOid;
  53
  54         amroutine->ambuild = ginbuild;
  55         amroutine->ambuildempty = ginbuildempty;
  56         amroutine->aminsert = gininsert;
  57         amroutine->ambulkdelete = ginbulkdelete;
  58         amroutine->amvacuumcleanup = ginvacuumcleanup;
  59         amroutine->amcanreturn = NULL;
  60         amroutine->amcostestimate = gincostestimate;
  61         amroutine->amoptions = ginoptions;
  62         amroutine->amproperty = NULL;
  63         amroutine->amvalidate = ginvalidate;
  64         amroutine->ambeginscan = ginbeginscan;
  65         amroutine->amrescan = ginrescan;
  66         amroutine->amgettuple = NULL;
  67         amroutine->amgetbitmap = gingetbitmap;
  68         amroutine->amendscan = ginendscan;
  69         amroutine->ammarkpos = NULL;
  70         amroutine->amrestrpos = NULL;
  71         amroutine->amestimateparallelscan = NULL;
  72         amroutine->aminitparallelscan = NULL;
  73         amroutine->amparallelrescan = NULL;
  74
  75         PG_RETURN_POINTER(amroutine);
  76 }
  77
  78 /*
  79  * initGinState: fill in an empty GinState struct to describe the index
  80  *
  81  * Note: assorted subsidiary data is allocated in the CurrentMemoryContext.
  82  */
  83 void
  84 initGinState(GinState *state, Relation index)
  85 {
  86         TupleDesc       origTupdesc = RelationGetDescr(index);
  87         int                     i;
  88
  89         MemSet(state, 0, sizeof(GinState));
  90
  91         state->index = index;
  92         state->oneCol = (origTupdesc->natts == 1) ? true : false;
  93         state->origTupdesc = origTupdesc;
  94
  95         for (i = 0; i < origTupdesc->natts; i++)
  96         {
  97                 if (state->oneCol)
  98                         state->tupdesc[i] = state->origTupdesc;
  99                 else
 100                 {
 101                         state->tupdesc[i] = CreateTemplateTupleDesc(2, false);
 102
 103                         TupleDescInitEntry(state->tupdesc[i], (AttrNumber) 1, NULL,
 104                                                            INT2OID, -1, 0);
 105                         TupleDescInitEntry(state->tupdesc[i], (AttrNumber) 2, NULL,
 106                                                            origTupdesc->attrs[i]->atttypid,
 107                                                            origTupdesc->attrs[i]->atttypmod,
 108                                                            origTupdesc->attrs[i]->attndims);
 109                         TupleDescInitEntryCollation(state->tupdesc[i], (AttrNumber) 2,
 110                                                                                 origTupdesc->attrs[i]->attcollation);
 111                 }
 112
 113                 /*
 114                  * If the compare proc isn't specified in the opclass definition, look
 115                  * up the index key type's default btree comparator.
 116                  */
 117                 if (index_getprocid(index, i + 1, GIN_COMPARE_PROC) != InvalidOid)
 118                 {
 119                         fmgr_info_copy(&(state->compareFn[i]),
 120                                                    index_getprocinfo(index, i + 1, GIN_COMPARE_PROC),
 121                                                    CurrentMemoryContext);
 122                 }
 123                 else
 124                 {
 125                         TypeCacheEntry *typentry;
 126
 127                         typentry = lookup_type_cache(origTupdesc->attrs[i]->atttypid,
 128                                                                                  TYPECACHE_CMP_PROC_FINFO);
 129                         if (!OidIsValid(typentry->cmp_proc_finfo.fn_oid))
 130                                 ereport(ERROR,
 131                                                 (errcode(ERRCODE_UNDEFINED_FUNCTION),
 132                                 errmsg("could not identify a comparison function for type %s",
 133                                            format_type_be(origTupdesc->attrs[i]->atttypid))));
 134                         fmgr_info_copy(&(state->compareFn[i]),
 135                                                    &(typentry->cmp_proc_finfo),
 136                                                    CurrentMemoryContext);
 137                 }
 138
 139                 /* Opclass must always provide extract procs */
 140                 fmgr_info_copy(&(state->extractValueFn[i]),
 141                                            index_getprocinfo(index, i + 1, GIN_EXTRACTVALUE_PROC),
 142                                            CurrentMemoryContext);
 143                 fmgr_info_copy(&(state->extractQueryFn[i]),
 144                                            index_getprocinfo(index, i + 1, GIN_EXTRACTQUERY_PROC),
 145                                            CurrentMemoryContext);
 146
 147                 /*
 148                  * Check opclass capability to do tri-state or binary logic consistent
 149                  * check.
 150                  */
 151                 if (index_getprocid(index, i + 1, GIN_TRICONSISTENT_PROC) != InvalidOid)
 152                 {
 153                         fmgr_info_copy(&(state->triConsistentFn[i]),
 154                                          index_getprocinfo(index, i + 1, GIN_TRICONSISTENT_PROC),
 155                                                    CurrentMemoryContext);
 156                 }
 157
 158                 if (index_getprocid(index, i + 1, GIN_CONSISTENT_PROC) != InvalidOid)
 159                 {
 160                         fmgr_info_copy(&(state->consistentFn[i]),
 161                                                 index_getprocinfo(index, i + 1, GIN_CONSISTENT_PROC),
 162                                                    CurrentMemoryContext);
 163                 }
 164
 165                 if (state->consistentFn[i].fn_oid == InvalidOid &&
 166                         state->triConsistentFn[i].fn_oid == InvalidOid)
 167                 {
 168                         elog(ERROR, "missing GIN support function (%d or %d) for attribute %d of index \"%s\"",
 169                                  GIN_CONSISTENT_PROC, GIN_TRICONSISTENT_PROC,
 170                                  i + 1, RelationGetRelationName(index));
 171                 }
 172
 173                 /*
 174                  * Check opclass capability to do partial match.
 175                  */
 176                 if (index_getprocid(index, i + 1, GIN_COMPARE_PARTIAL_PROC) != InvalidOid)
 177                 {
 178                         fmgr_info_copy(&(state->comparePartialFn[i]),
 179                                    index_getprocinfo(index, i + 1, GIN_COMPARE_PARTIAL_PROC),
 180                                                    CurrentMemoryContext);
 181                         state->canPartialMatch[i] = true;
 182                 }
 183                 else
 184                 {
 185                         state->canPartialMatch[i] = false;
 186                 }
 187
 188                 /*
 189                  * If the index column has a specified collation, we should honor that
 190                  * while doing comparisons.  However, we may have a collatable storage
 191                  * type for a noncollatable indexed data type (for instance, hstore
 192                  * uses text index entries).  If there's no index collation then
 193                  * specify default collation in case the support functions need
 194                  * collation.  This is harmless if the support functions don't care
 195                  * about collation, so we just do it unconditionally.  (We could
 196                  * alternatively call get_typcollation, but that seems like expensive
 197                  * overkill --- there aren't going to be any cases where a GIN storage
 198                  * type has a nondefault collation.)
 199                  */
 200                 if (OidIsValid(index->rd_indcollation[i]))
 201                         state->supportCollation[i] = index->rd_indcollation[i];
 202                 else
 203                         state->supportCollation[i] = DEFAULT_COLLATION_OID;
 204         }
 205 }
 206
 207 /*
 208  * Extract attribute (column) number of stored entry from GIN tuple
 209  */
 210 OffsetNumber
 211 gintuple_get_attrnum(GinState *ginstate, IndexTuple tuple)
 212 {
 213         OffsetNumber colN;
 214
 215         if (ginstate->oneCol)
 216         {
 217                 /* column number is not stored explicitly */
 218                 colN = FirstOffsetNumber;
 219         }
 220         else
 221         {
 222                 Datum           res;
 223                 bool            isnull;
 224
 225                 /*
 226                  * First attribute is always int16, so we can safely use any tuple
 227                  * descriptor to obtain first attribute of tuple
 228                  */
 229                 res = index_getattr(tuple, FirstOffsetNumber, ginstate->tupdesc[0],
 230                                                         &isnull);
 231                 Assert(!isnull);
 232
 233                 colN = DatumGetUInt16(res);
 234                 Assert(colN >= FirstOffsetNumber && colN <= ginstate->origTupdesc->natts);
 235         }
 236
 237         return colN;
 238 }
 239
 240 /*
 241  * Extract stored datum (and possible null category) from GIN tuple
 242  */
 243 Datum
 244 gintuple_get_key(GinState *ginstate, IndexTuple tuple,
 245                                  GinNullCategory *category)
 246 {
 247         Datum           res;
 248         bool            isnull;
 249
 250         if (ginstate->oneCol)
 251         {
 252                 /*
 253                  * Single column index doesn't store attribute numbers in tuples
 254                  */
 255                 res = index_getattr(tuple, FirstOffsetNumber, ginstate->origTupdesc,
 256                                                         &isnull);
 257         }
 258         else
 259         {
 260                 /*
 261                  * Since the datum type depends on which index column it's from, we
 262                  * must be careful to use the right tuple descriptor here.
 263                  */
 264                 OffsetNumber colN = gintuple_get_attrnum(ginstate, tuple);
 265
 266                 res = index_getattr(tuple, OffsetNumberNext(FirstOffsetNumber),
 267                                                         ginstate->tupdesc[colN - 1],
 268                                                         &isnull);
 269         }
 270
 271         if (isnull)
 272                 *category = GinGetNullCategory(tuple, ginstate);
 273         else
 274                 *category = GIN_CAT_NORM_KEY;
 275
 276         return res;
 277 }
 278
 279 /*
 280  * Allocate a new page (either by recycling, or by extending the index file)
 281  * The returned buffer is already pinned and exclusive-locked
 282  * Caller is responsible for initializing the page by calling GinInitBuffer
 283  */
 284 Buffer
 285 GinNewBuffer(Relation index)
 286 {
 287         Buffer          buffer;
 288         bool            needLock;
 289
 290         /* First, try to get a page from FSM */
 291         for (;;)
 292         {
 293                 BlockNumber blkno = GetFreeIndexPage(index);
 294
 295                 if (blkno == InvalidBlockNumber)
 296                         break;
 297
 298                 buffer = ReadBuffer(index, blkno);
 299
 300                 /*
 301                  * We have to guard against the possibility that someone else already
 302                  * recycled this page; the buffer may be locked if so.
 303                  */
 304                 if (ConditionalLockBuffer(buffer))
 305                 {
 306                         Page            page = BufferGetPage(buffer);
 307
 308                         if (PageIsNew(page))
 309                                 return buffer;  /* OK to use, if never initialized */
 310
 311                         if (GinPageIsDeleted(page))
 312                                 return buffer;  /* OK to use */
 313
 314                         LockBuffer(buffer, GIN_UNLOCK);
 315                 }
 316
 317                 /* Can't use it, so release buffer and try again */
 318                 ReleaseBuffer(buffer);
 319         }
 320
 321         /* Must extend the file */
 322         needLock = !RELATION_IS_LOCAL(index);
 323         if (needLock)
 324                 LockRelationForExtension(index, ExclusiveLock);
 325
 326         buffer = ReadBuffer(index, P_NEW);
 327         LockBuffer(buffer, GIN_EXCLUSIVE);
 328
 329         if (needLock)
 330                 UnlockRelationForExtension(index, ExclusiveLock);
 331
 332         return buffer;
 333 }
 334
 335 void
 336 GinInitPage(Page page, uint32 f, Size pageSize)
 337 {
 338         GinPageOpaque opaque;
 339
 340         PageInit(page, pageSize, sizeof(GinPageOpaqueData));
 341
 342         opaque = GinPageGetOpaque(page);
 343         memset(opaque, 0, sizeof(GinPageOpaqueData));
 344         opaque->flags = f;
 345         opaque->rightlink = InvalidBlockNumber;
 346 }
 347
 348 void
 349 GinInitBuffer(Buffer b, uint32 f)
 350 {
 351         GinInitPage(BufferGetPage(b), f, BufferGetPageSize(b));
 352 }
 353
 354 void
 355 GinInitMetabuffer(Buffer b)
 356 {
 357         GinMetaPageData *metadata;
 358         Page            page = BufferGetPage(b);
 359
 360         GinInitPage(page, GIN_META, BufferGetPageSize(b));
 361
 362         metadata = GinPageGetMeta(page);
 363
 364         metadata->head = metadata->tail = InvalidBlockNumber;
 365         metadata->tailFreeSize = 0;
 366         metadata->nPendingPages = 0;
 367         metadata->nPendingHeapTuples = 0;
 368         metadata->nTotalPages = 0;
 369         metadata->nEntryPages = 0;
 370         metadata->nDataPages = 0;
 371         metadata->nEntries = 0;
 372         metadata->ginVersion = GIN_CURRENT_VERSION;
 373 }
 374
 375 /*
 376  * Compare two keys of the same index column
 377  */
 378 int
 379 ginCompareEntries(GinState *ginstate, OffsetNumber attnum,
 380                                   Datum a, GinNullCategory categorya,
 381                                   Datum b, GinNullCategory categoryb)
 382 {
 383         /* if not of same null category, sort by that first */
 384         if (categorya != categoryb)
 385                 return (categorya < categoryb) ? -1 : 1;
 386
 387         /* all null items in same category are equal */
 388         if (categorya != GIN_CAT_NORM_KEY)
 389                 return 0;
 390
 391         /* both not null, so safe to call the compareFn */
 392         return DatumGetInt32(FunctionCall2Coll(&ginstate->compareFn[attnum - 1],
 393                                                                           ginstate->supportCollation[attnum - 1],
 394                                                                                    a, b));
 395 }
 396
 397 /*
 398  * Compare two keys of possibly different index columns
 399  */
 400 int
 401 ginCompareAttEntries(GinState *ginstate,
 402                                          OffsetNumber attnuma, Datum a, GinNullCategory categorya,
 403                                          OffsetNumber attnumb, Datum b, GinNullCategory categoryb)
 404 {
 405         /* attribute number is the first sort key */
 406         if (attnuma != attnumb)
 407                 return (attnuma < attnumb) ? -1 : 1;
 408
 409         return ginCompareEntries(ginstate, attnuma, a, categorya, b, categoryb);
 410 }
 411
 412
 413 /*
 414  * Support for sorting key datums in ginExtractEntries
 415  *
 416  * Note: we only have to worry about null and not-null keys here;
 417  * ginExtractEntries never generates more than one placeholder null,
 418  * so it doesn't have to sort those.
 419  */
 420 typedef struct
 421 {
 422         Datum           datum;
 423         bool            isnull;
 424 } keyEntryData;
 425
 426 typedef struct
 427 {
 428         FmgrInfo   *cmpDatumFunc;
 429         Oid                     collation;
 430         bool            haveDups;
 431 } cmpEntriesArg;
 432
 433 static int
 434 cmpEntries(const void *a, const void *b, void *arg)
 435 {
 436         const keyEntryData *aa = (const keyEntryData *) a;
 437         const keyEntryData *bb = (const keyEntryData *) b;
 438         cmpEntriesArg *data = (cmpEntriesArg *) arg;
 439         int                     res;
 440
 441         if (aa->isnull)
 442         {
 443                 if (bb->isnull)
 444                         res = 0;                        /* NULL "=" NULL */
 445                 else
 446                         res = 1;                        /* NULL ">" not-NULL */
 447         }
 448         else if (bb->isnull)
 449                 res = -1;                               /* not-NULL "<" NULL */
 450         else
 451                 res = DatumGetInt32(FunctionCall2Coll(data->cmpDatumFunc,
 452                                                                                           data->collation,
 453                                                                                           aa->datum, bb->datum));
 454
 455         /*
 456          * Detect if we have any duplicates.  If there are equal keys, qsort must
 457          * compare them at some point, else it wouldn't know whether one should go
 458          * before or after the other.
 459          */
 460         if (res == 0)
 461                 data->haveDups = true;
 462
 463         return res;
 464 }
 465
 466
 467 /*
 468  * Extract the index key values from an indexable item
 469  *
 470  * The resulting key values are sorted, and any duplicates are removed.
 471  * This avoids generating redundant index entries.
 472  */
 473 Datum *
 474 ginExtractEntries(GinState *ginstate, OffsetNumber attnum,
 475                                   Datum value, bool isNull,
 476                                   int32 *nentries, GinNullCategory **categories)
 477 {
 478         Datum      *entries;
 479         bool       *nullFlags;
 480         int32           i;
 481
 482         /*
 483          * We don't call the extractValueFn on a null item.  Instead generate a
 484          * placeholder.
 485          */
 486         if (isNull)
 487         {
 488                 *nentries = 1;
 489                 entries = (Datum *) palloc(sizeof(Datum));
 490                 entries[0] = (Datum) 0;
 491                 *categories = (GinNullCategory *) palloc(sizeof(GinNullCategory));
 492                 (*categories)[0] = GIN_CAT_NULL_ITEM;
 493                 return entries;
 494         }
 495
 496         /* OK, call the opclass's extractValueFn */
 497         nullFlags = NULL;                       /* in case extractValue doesn't set it */
 498         entries = (Datum *)
 499                 DatumGetPointer(FunctionCall3Coll(&ginstate->extractValueFn[attnum - 1],
 500                                                                           ginstate->supportCollation[attnum - 1],
 501                                                                                   value,
 502                                                                                   PointerGetDatum(nentries),
 503                                                                                   PointerGetDatum(&nullFlags)));
 504
 505         /*
 506          * Generate a placeholder if the item contained no keys.
 507          */
 508         if (entries == NULL || *nentries <= 0)
 509         {
 510                 *nentries = 1;
 511                 entries = (Datum *) palloc(sizeof(Datum));
 512                 entries[0] = (Datum) 0;
 513                 *categories = (GinNullCategory *) palloc(sizeof(GinNullCategory));
 514                 (*categories)[0] = GIN_CAT_EMPTY_ITEM;
 515                 return entries;
 516         }
 517
 518         /*
 519          * If the extractValueFn didn't create a nullFlags array, create one,
 520          * assuming that everything's non-null.  Otherwise, run through the array
 521          * and make sure each value is exactly 0 or 1; this ensures binary
 522          * compatibility with the GinNullCategory representation.
 523          */
 524         if (nullFlags == NULL)
 525                 nullFlags = (bool *) palloc0(*nentries * sizeof(bool));
 526         else
 527         {
 528                 for (i = 0; i < *nentries; i++)
 529                         nullFlags[i] = (nullFlags[i] ? true : false);
 530         }
 531         /* now we can use the nullFlags as category codes */
 532         *categories = (GinNullCategory *) nullFlags;
 533
 534         /*
 535          * If there's more than one key, sort and unique-ify.
 536          *
 537          * XXX Using qsort here is notationally painful, and the overhead is
 538          * pretty bad too.  For small numbers of keys it'd likely be better to use
 539          * a simple insertion sort.
 540          */
 541         if (*nentries > 1)
 542         {
 543                 keyEntryData *keydata;
 544                 cmpEntriesArg arg;
 545
 546                 keydata = (keyEntryData *) palloc(*nentries * sizeof(keyEntryData));
 547                 for (i = 0; i < *nentries; i++)
 548                 {
 549                         keydata[i].datum = entries[i];
 550                         keydata[i].isnull = nullFlags[i];
 551                 }
 552
 553                 arg.cmpDatumFunc = &ginstate->compareFn[attnum - 1];
 554                 arg.collation = ginstate->supportCollation[attnum - 1];
 555                 arg.haveDups = false;
 556                 qsort_arg(keydata, *nentries, sizeof(keyEntryData),
 557                                   cmpEntries, (void *) &arg);
 558
 559                 if (arg.haveDups)
 560                 {
 561                         /* there are duplicates, must get rid of 'em */
 562                         int32           j;
 563
 564                         entries[0] = keydata[0].datum;
 565                         nullFlags[0] = keydata[0].isnull;
 566                         j = 1;
 567                         for (i = 1; i < *nentries; i++)
 568                         {
 569                                 if (cmpEntries(&keydata[i - 1], &keydata[i], &arg) != 0)
 570                                 {
 571                                         entries[j] = keydata[i].datum;
 572                                         nullFlags[j] = keydata[i].isnull;
 573                                         j++;
 574                                 }
 575                         }
 576                         *nentries = j;
 577                 }
 578                 else
 579                 {
 580                         /* easy, no duplicates */
 581                         for (i = 0; i < *nentries; i++)
 582                         {
 583                                 entries[i] = keydata[i].datum;
 584                                 nullFlags[i] = keydata[i].isnull;
 585                         }
 586                 }
 587
 588                 pfree(keydata);
 589         }
 590
 591         return entries;
 592 }
 593
 594 bytea *
 595 ginoptions(Datum reloptions, bool validate)
 596 {
 597         relopt_value *options;
 598         GinOptions *rdopts;
 599         int                     numoptions;
 600         static const relopt_parse_elt tab[] = {
 601                 {"fastupdate", RELOPT_TYPE_BOOL, offsetof(GinOptions, useFastUpdate)},
 602                 {"gin_pending_list_limit", RELOPT_TYPE_INT, offsetof(GinOptions,
 603                                                                                                          pendingListCleanupSize)}
 604         };
 605
 606         options = parseRelOptions(reloptions, validate, RELOPT_KIND_GIN,
 607                                                           &numoptions);
 608
 609         /* if none set, we're done */
 610         if (numoptions == 0)
 611                 return NULL;
 612
 613         rdopts = allocateReloptStruct(sizeof(GinOptions), options, numoptions);
 614
 615         fillRelOptions((void *) rdopts, sizeof(GinOptions), options, numoptions,
 616                                    validate, tab, lengthof(tab));
 617
 618         pfree(options);
 619
 620         return (bytea *) rdopts;
 621 }
 622
 623 /*
 624  * Fetch index's statistical data into *stats
 625  *
 626  * Note: in the result, nPendingPages can be trusted to be up-to-date,
 627  * as can ginVersion; but the other fields are as of the last VACUUM.
 628  */
 629 void
 630 ginGetStats(Relation index, GinStatsData *stats)
 631 {
 632         Buffer          metabuffer;
 633         Page            metapage;
 634         GinMetaPageData *metadata;
 635
 636         metabuffer = ReadBuffer(index, GIN_METAPAGE_BLKNO);
 637         LockBuffer(metabuffer, GIN_SHARE);
 638         metapage = BufferGetPage(metabuffer);
 639         metadata = GinPageGetMeta(metapage);
 640
 641         stats->nPendingPages = metadata->nPendingPages;
 642         stats->nTotalPages = metadata->nTotalPages;
 643         stats->nEntryPages = metadata->nEntryPages;
 644         stats->nDataPages = metadata->nDataPages;
 645         stats->nEntries = metadata->nEntries;
 646         stats->ginVersion = metadata->ginVersion;
 647
 648         UnlockReleaseBuffer(metabuffer);
 649 }
 650
 651 /*
 652  * Write the given statistics to the index's metapage
 653  *
 654  * Note: nPendingPages and ginVersion are *not* copied over
 655  */
 656 void
 657 ginUpdateStats(Relation index, const GinStatsData *stats)
 658 {
 659         Buffer          metabuffer;
 660         Page            metapage;
 661         GinMetaPageData *metadata;
 662
 663         metabuffer = ReadBuffer(index, GIN_METAPAGE_BLKNO);
 664         LockBuffer(metabuffer, GIN_EXCLUSIVE);
 665         metapage = BufferGetPage(metabuffer);
 666         metadata = GinPageGetMeta(metapage);
 667
 668         START_CRIT_SECTION();
 669
 670         metadata->nTotalPages = stats->nTotalPages;
 671         metadata->nEntryPages = stats->nEntryPages;
 672         metadata->nDataPages = stats->nDataPages;
 673         metadata->nEntries = stats->nEntries;
 674
 675         MarkBufferDirty(metabuffer);
 676
 677         if (RelationNeedsWAL(index))
 678         {
 679                 XLogRecPtr      recptr;
 680                 ginxlogUpdateMeta data;
 681
 682                 data.node = index->rd_node;
 683                 data.ntuples = 0;
 684                 data.newRightlink = data.prevTail = InvalidBlockNumber;
 685                 memcpy(&data.metadata, metadata, sizeof(GinMetaPageData));
 686
 687                 XLogBeginInsert();
 688                 XLogRegisterData((char *) &data, sizeof(ginxlogUpdateMeta));
 689                 XLogRegisterBuffer(0, metabuffer, REGBUF_WILL_INIT);
 690
 691                 recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_UPDATE_META_PAGE);
 692                 PageSetLSN(metapage, recptr);
 693         }
 694
 695         UnlockReleaseBuffer(metabuffer);
 696
 697         END_CRIT_SECTION();
 698 }