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