brin.c

Go to the documentation of this file.

/*
 * brin.c
 *      Implementation of BRIN indexes for Postgres
 *
 * See src/backend/access/brin/README for details.
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *    src/backend/access/brin/brin.c
 *
 * TODO
 *      * ScalarArrayOpExpr (amsearcharray -> SK_SEARCHARRAY)
 */
#include "postgres.h"
 
#include "access/brin.h"
#include "access/brin_page.h"
#include "access/brin_pageops.h"
#include "access/brin_xlog.h"
#include "access/relation.h"
#include "access/reloptions.h"
#include "access/relscan.h"
#include "access/table.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
#include "catalog/index.h"
#include "catalog/pg_am.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/autovacuum.h"
#include "storage/bufmgr.h"
#include "storage/freespace.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/datum.h"
#include "utils/fmgrprotos.h"
#include "utils/guc.h"
#include "utils/index_selfuncs.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/tuplesort.h"
 
/* Magic numbers for parallel state sharing */
#define PARALLEL_KEY_BRIN_SHARED        UINT64CONST(0xB000000000000001)
#define PARALLEL_KEY_TUPLESORT          UINT64CONST(0xB000000000000002)
#define PARALLEL_KEY_QUERY_TEXT         UINT64CONST(0xB000000000000003)
#define PARALLEL_KEY_WAL_USAGE          UINT64CONST(0xB000000000000004)
#define PARALLEL_KEY_BUFFER_USAGE       UINT64CONST(0xB000000000000005)
 
/*
 * Status for index builds performed in parallel.  This is allocated in a
 * dynamic shared memory segment.
 */
typedef struct BrinShared
{
    /*
     * These fields are not modified during the build.  They primarily exist
     * for the benefit of worker processes that need to create state
     * corresponding to that used by the leader.
     */
    Oid         heaprelid;
    Oid         indexrelid;
    bool        isconcurrent;
    BlockNumber pagesPerRange;
    int         scantuplesortstates;
 
    /* Query ID, for report in worker processes */
    int64       queryid;
 
    /*
     * workersdonecv is used to monitor the progress of workers.  All parallel
     * participants must indicate that they are done before leader can use
     * results built by the workers (and before leader can write the data into
     * the index).
     */
    ConditionVariable workersdonecv;
 
    /*
     * mutex protects all fields before heapdesc.
     *
     * These fields contain status information of interest to BRIN index
     * builds that must work just the same when an index is built in parallel.
     */
    slock_t     mutex;
 
    /*
     * Mutable state that is maintained by workers, and reported back to
     * leader at end of the scans.
     *
     * nparticipantsdone is number of worker processes finished.
     *
     * reltuples is the total number of input heap tuples.
     *
     * indtuples is the total number of tuples that made it into the index.
     */
    int         nparticipantsdone;
    double      reltuples;
    double      indtuples;
 
    /*
     * ParallelTableScanDescData data follows. Can't directly embed here, as
     * implementations of the parallel table scan desc interface might need
     * stronger alignment.
     */
} BrinShared;
 
/*
 * Return pointer to a BrinShared's parallel table scan.
 *
 * c.f. shm_toc_allocate as to why BUFFERALIGN is used, rather than just
 * MAXALIGN.
 */
#define ParallelTableScanFromBrinShared(shared) \
    (ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(BrinShared)))
 
/*
 * Status for leader in parallel index build.
 */
typedef struct BrinLeader
{
    /* parallel context itself */
    ParallelContext *pcxt;
 
    /*
     * nparticipanttuplesorts is the exact number of worker processes
     * successfully launched, plus one leader process if it participates as a
     * worker (only DISABLE_LEADER_PARTICIPATION builds avoid leader
     * participating as a worker).
     */
    int         nparticipanttuplesorts;
 
    /*
     * Leader process convenience pointers to shared state (leader avoids TOC
     * lookups).
     *
     * brinshared is the shared state for entire build.  sharedsort is the
     * shared, tuplesort-managed state passed to each process tuplesort.
     * snapshot is the snapshot used by the scan iff an MVCC snapshot is
     * required.
     */
    BrinShared *brinshared;
    Sharedsort *sharedsort;
    Snapshot    snapshot;
    WalUsage   *walusage;
    BufferUsage *bufferusage;
} BrinLeader;
 
/*
 * We use a BrinBuildState during initial construction of a BRIN index.
 * The running state is kept in a BrinMemTuple.
 */
typedef struct BrinBuildState
{
    Relation    bs_irel;
    double      bs_numtuples;
    double      bs_reltuples;
    Buffer      bs_currentInsertBuf;
    BlockNumber bs_pagesPerRange;
    BlockNumber bs_currRangeStart;
    BlockNumber bs_maxRangeStart;
    BrinRevmap *bs_rmAccess;
    BrinDesc   *bs_bdesc;
    BrinMemTuple *bs_dtuple;
 
    BrinTuple  *bs_emptyTuple;
    Size        bs_emptyTupleLen;
    MemoryContext bs_context;
 
    /*
     * bs_leader is only present when a parallel index build is performed, and
     * only in the leader process. (Actually, only the leader process has a
     * BrinBuildState.)
     */
    BrinLeader *bs_leader;
    int         bs_worker_id;
 
    /*
     * The sortstate is used by workers (including the leader). It has to be
     * part of the build state, because that's the only thing passed to the
     * build callback etc.
     */
    Tuplesortstate *bs_sortstate;
} BrinBuildState;
 
/*
 * We use a BrinInsertState to capture running state spanning multiple
 * brininsert invocations, within the same command.
 */
typedef struct BrinInsertState
{
    BrinRevmap *bis_rmAccess;
    BrinDesc   *bis_desc;
    BlockNumber bis_pages_per_range;
} BrinInsertState;
 
/*
 * Struct used as "opaque" during index scans
 */
typedef struct BrinOpaque
{
    BlockNumber bo_pagesPerRange;
    BrinRevmap *bo_rmAccess;
    BrinDesc   *bo_bdesc;
} BrinOpaque;
 
#define BRIN_ALL_BLOCKRANGES    InvalidBlockNumber
 
static BrinBuildState *initialize_brin_buildstate(Relation idxRel,
                                                  BrinRevmap *revmap,
                                                  BlockNumber pagesPerRange,
                                                  BlockNumber tablePages);
static BrinInsertState *initialize_brin_insertstate(Relation idxRel, IndexInfo *indexInfo);
static void terminate_brin_buildstate(BrinBuildState *state);
static void brinsummarize(Relation index, Relation heapRel, BlockNumber pageRange,
                          bool include_partial, double *numSummarized, double *numExisting);
static void form_and_insert_tuple(BrinBuildState *state);
static void form_and_spill_tuple(BrinBuildState *state);
static void union_tuples(BrinDesc *bdesc, BrinMemTuple *a,
                         BrinTuple *b);
static void brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy);
static bool add_values_to_range(Relation idxRel, BrinDesc *bdesc,
                                BrinMemTuple *dtup, const Datum *values, const bool *nulls);
static bool check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys);
static void brin_fill_empty_ranges(BrinBuildState *state,
                                   BlockNumber prevRange, BlockNumber nextRange);
 
/* parallel index builds */
static void _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index,
                                 bool isconcurrent, int request);
static void _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state);
static Size _brin_parallel_estimate_shared(Relation heap, Snapshot snapshot);
static double _brin_parallel_heapscan(BrinBuildState *state);
static double _brin_parallel_merge(BrinBuildState *state);
static void _brin_leader_participate_as_worker(BrinBuildState *buildstate,
                                               Relation heap, Relation index);
static void _brin_parallel_scan_and_build(BrinBuildState *state,
                                          BrinShared *brinshared,
                                          Sharedsort *sharedsort,
                                          Relation heap, Relation index,
                                          int sortmem, bool progress);
 
/*
 * BRIN handler function: return IndexAmRoutine with access method parameters
 * and callbacks.
 */
Datum
brinhandler(PG_FUNCTION_ARGS)
{
    static const IndexAmRoutine amroutine = {
        .type = T_IndexAmRoutine,
        .amstrategies = 0,
        .amsupport = BRIN_LAST_OPTIONAL_PROCNUM,
        .amoptsprocnum = BRIN_PROCNUM_OPTIONS,
        .amcanorder = false,
        .amcanorderbyop = false,
        .amcanhash = false,
        .amconsistentequality = false,
        .amconsistentordering = false,
        .amcanbackward = false,
        .amcanunique = false,
        .amcanmulticol = true,
        .amoptionalkey = true,
        .amsearcharray = false,
        .amsearchnulls = true,
        .amstorage = true,
        .amclusterable = false,
        .ampredlocks = false,
        .amcanparallel = false,
        .amcanbuildparallel = true,
        .amcaninclude = false,
        .amusemaintenanceworkmem = false,
        .amsummarizing = true,
        .amparallelvacuumoptions =
        VACUUM_OPTION_PARALLEL_CLEANUP,
        .amkeytype = InvalidOid,
 
        .ambuild = brinbuild,
        .ambuildempty = brinbuildempty,
        .aminsert = brininsert,
        .aminsertcleanup = brininsertcleanup,
        .ambulkdelete = brinbulkdelete,
        .amvacuumcleanup = brinvacuumcleanup,
        .amcanreturn = NULL,
        .amcostestimate = brincostestimate,
        .amgettreeheight = NULL,
        .amoptions = brinoptions,
        .amproperty = NULL,
        .ambuildphasename = NULL,
        .amvalidate = brinvalidate,
        .amadjustmembers = NULL,
        .ambeginscan = brinbeginscan,
        .amrescan = brinrescan,
        .amgettuple = NULL,
        .amgetbitmap = bringetbitmap,
        .amendscan = brinendscan,
        .ammarkpos = NULL,
        .amrestrpos = NULL,
        .amestimateparallelscan = NULL,
        .aminitparallelscan = NULL,
        .amparallelrescan = NULL,
        .amtranslatestrategy = NULL,
        .amtranslatecmptype = NULL,
    };
 
    PG_RETURN_POINTER(&amroutine);
}
 
/*
 * Initialize a BrinInsertState to maintain state to be used across multiple
 * tuple inserts, within the same command.
 */
static BrinInsertState *
initialize_brin_insertstate(Relation idxRel, IndexInfo *indexInfo)
{
    BrinInsertState *bistate;
    MemoryContext oldcxt;
 
    oldcxt = MemoryContextSwitchTo(indexInfo->ii_Context);
    bistate = palloc0_object(BrinInsertState);
    bistate->bis_desc = brin_build_desc(idxRel);
    bistate->bis_rmAccess = brinRevmapInitialize(idxRel,
                                                 &bistate->bis_pages_per_range);
    indexInfo->ii_AmCache = bistate;
    MemoryContextSwitchTo(oldcxt);
 
    return bistate;
}
 
/*
 * A tuple in the heap is being inserted.  To keep a brin index up to date,
 * we need to obtain the relevant index tuple and compare its stored values
 * with those of the new tuple.  If the tuple values are not consistent with
 * the summary tuple, we need to update the index tuple.
 *
 * If autosummarization is enabled, check if we need to summarize the previous
 * page range.
 *
 * If the range is not currently summarized (i.e. the revmap returns NULL for
 * it), there's nothing to do for this tuple.
 */
bool
brininsert(Relation idxRel, Datum *values, bool *nulls,
           ItemPointer heaptid, Relation heapRel,
           IndexUniqueCheck checkUnique,
           bool indexUnchanged,
           IndexInfo *indexInfo)
{
    BlockNumber pagesPerRange;
    BlockNumber origHeapBlk;
    BlockNumber heapBlk;
    BrinInsertState *bistate = (BrinInsertState *) indexInfo->ii_AmCache;
    BrinRevmap *revmap;
    BrinDesc   *bdesc;
    Buffer      buf = InvalidBuffer;
    MemoryContext tupcxt = NULL;
    MemoryContext oldcxt = CurrentMemoryContext;
    bool        autosummarize = BrinGetAutoSummarize(idxRel);
 
    /*
     * If first time through in this statement, initialize the insert state
     * that we keep for all the inserts in the command.
     */
    if (!bistate)
        bistate = initialize_brin_insertstate(idxRel, indexInfo);
 
    revmap = bistate->bis_rmAccess;
    bdesc = bistate->bis_desc;
    pagesPerRange = bistate->bis_pages_per_range;
 
    /*
     * origHeapBlk is the block number where the insertion occurred.  heapBlk
     * is the first block in the corresponding page range.
     */
    origHeapBlk = ItemPointerGetBlockNumber(heaptid);
    heapBlk = (origHeapBlk / pagesPerRange) * pagesPerRange;
 
    for (;;)
    {
        bool        need_insert = false;
        OffsetNumber off;
        BrinTuple  *brtup;
        BrinMemTuple *dtup;
 
        CHECK_FOR_INTERRUPTS();
 
        /*
         * If auto-summarization is enabled and we just inserted the first
         * tuple into the first block of a new non-first page range, request a
         * summarization run of the previous range.
         */
        if (autosummarize &&
            heapBlk > 0 &&
            heapBlk == origHeapBlk &&
            ItemPointerGetOffsetNumber(heaptid) == FirstOffsetNumber)
        {
            BlockNumber lastPageRange = heapBlk - 1;
            BrinTuple  *lastPageTuple;
 
            lastPageTuple =
                brinGetTupleForHeapBlock(revmap, lastPageRange, &buf, &off,
                                         NULL, BUFFER_LOCK_SHARE);
            if (!lastPageTuple)
            {
                bool        recorded;
 
                recorded = AutoVacuumRequestWork(AVW_BRINSummarizeRange,
                                                 RelationGetRelid(idxRel),
                                                 lastPageRange);
                if (!recorded)
                    ereport(LOG,
                            (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                             errmsg("request for BRIN range summarization for index \"%s\" page %u was not recorded",
                                    RelationGetRelationName(idxRel),
                                    lastPageRange)));
            }
            else
                LockBuffer(buf, BUFFER_LOCK_UNLOCK);
        }
 
        brtup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off,
                                         NULL, BUFFER_LOCK_SHARE);
 
        /* if range is unsummarized, there's nothing to do */
        if (!brtup)
            break;
 
        /* First time through in this brininsert call? */
        if (tupcxt == NULL)
        {
            tupcxt = AllocSetContextCreate(CurrentMemoryContext,
                                           "brininsert cxt",
                                           ALLOCSET_DEFAULT_SIZES);
            MemoryContextSwitchTo(tupcxt);
        }
 
        dtup = brin_deform_tuple(bdesc, brtup, NULL);
 
        need_insert = add_values_to_range(idxRel, bdesc, dtup, values, nulls);
 
        if (!need_insert)
        {
            /*
             * The tuple is consistent with the new values, so there's nothing
             * to do.
             */
            LockBuffer(buf, BUFFER_LOCK_UNLOCK);
        }
        else
        {
            Page        page = BufferGetPage(buf);
            ItemId      lp = PageGetItemId(page, off);
            Size        origsz;
            BrinTuple  *origtup;
            Size        newsz;
            BrinTuple  *newtup;
            bool        samepage;
 
            /*
             * Make a copy of the old tuple, so that we can compare it after
             * re-acquiring the lock.
             */
            origsz = ItemIdGetLength(lp);
            origtup = brin_copy_tuple(brtup, origsz, NULL, NULL);
 
            /*
             * Before releasing the lock, check if we can attempt a same-page
             * update.  Another process could insert a tuple concurrently in
             * the same page though, so downstream we must be prepared to cope
             * if this turns out to not be possible after all.
             */
            newtup = brin_form_tuple(bdesc, heapBlk, dtup, &newsz);
            samepage = brin_can_do_samepage_update(buf, origsz, newsz);
            LockBuffer(buf, BUFFER_LOCK_UNLOCK);
 
            /*
             * Try to update the tuple.  If this doesn't work for whatever
             * reason, we need to restart from the top; the revmap might be
             * pointing at a different tuple for this block now, so we need to
             * recompute to ensure both our new heap tuple and the other
             * inserter's are covered by the combined tuple.  It might be that
             * we don't need to update at all.
             */
            if (!brin_doupdate(idxRel, pagesPerRange, revmap, heapBlk,
                               buf, off, origtup, origsz, newtup, newsz,
                               samepage))
            {
                /* no luck; start over */
                MemoryContextReset(tupcxt);
                continue;
            }
        }
 
        /* success! */
        break;
    }
 
    if (BufferIsValid(buf))
        ReleaseBuffer(buf);
    MemoryContextSwitchTo(oldcxt);
    if (tupcxt != NULL)
        MemoryContextDelete(tupcxt);
 
    return false;
}
 
/*
 * Callback to clean up the BrinInsertState once all tuple inserts are done.
 */
void
brininsertcleanup(Relation index, IndexInfo *indexInfo)
{
    BrinInsertState *bistate = (BrinInsertState *) indexInfo->ii_AmCache;
 
    /* bail out if cache not initialized */
    if (bistate == NULL)
        return;
 
    /* do this first to avoid dangling pointer if we fail partway through */
    indexInfo->ii_AmCache = NULL;
 
    /*
     * Clean up the revmap. Note that the brinDesc has already been cleaned up
     * as part of its own memory context.
     */
    brinRevmapTerminate(bistate->bis_rmAccess);
    pfree(bistate);
}
 
/*
 * Initialize state for a BRIN index scan.
 *
 * We read the metapage here to determine the pages-per-range number that this
 * index was built with.  Note that since this cannot be changed while we're
 * holding lock on index, it's not necessary to recompute it during brinrescan.
 */
IndexScanDesc
brinbeginscan(Relation r, int nkeys, int norderbys)
{
    IndexScanDesc scan;
    BrinOpaque *opaque;
 
    scan = RelationGetIndexScan(r, nkeys, norderbys);
 
    opaque = palloc_object(BrinOpaque);
    opaque->bo_rmAccess = brinRevmapInitialize(r, &opaque->bo_pagesPerRange);
    opaque->bo_bdesc = brin_build_desc(r);
    scan->opaque = opaque;
 
    return scan;
}
 
/*
 * Execute the index scan.
 *
 * This works by reading index TIDs from the revmap, and obtaining the index
 * tuples pointed to by them; the summary values in the index tuples are
 * compared to the scan keys.  We return into the TID bitmap all the pages in
 * ranges corresponding to index tuples that match the scan keys.
 *
 * If a TID from the revmap is read as InvalidTID, we know that range is
 * unsummarized.  Pages in those ranges need to be returned regardless of scan
 * keys.
 */
int64
bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
{
    Relation    idxRel = scan->indexRelation;
    Buffer      buf = InvalidBuffer;
    BrinDesc   *bdesc;
    Oid         heapOid;
    Relation    heapRel;
    BrinOpaque *opaque;
    BlockNumber nblocks;
    int64       totalpages = 0;
    FmgrInfo   *consistentFn;
    MemoryContext oldcxt;
    MemoryContext perRangeCxt;
    BrinMemTuple *dtup;
    BrinTuple  *btup = NULL;
    Size        btupsz = 0;
    ScanKey   **keys,
              **nullkeys;
    int        *nkeys,
               *nnullkeys;
    char       *ptr;
    Size        len;
    char       *tmp PG_USED_FOR_ASSERTS_ONLY;
 
    opaque = (BrinOpaque *) scan->opaque;
    bdesc = opaque->bo_bdesc;
    pgstat_count_index_scan(idxRel);
    if (scan->instrument)
        scan->instrument->nsearches++;
 
    /*
     * We need to know the size of the table so that we know how long to
     * iterate on the revmap.
     */
    heapOid = IndexGetRelation(RelationGetRelid(idxRel), false);
    heapRel = table_open(heapOid, AccessShareLock);
    nblocks = RelationGetNumberOfBlocks(heapRel);
    table_close(heapRel, AccessShareLock);
 
    /*
     * Make room for the consistent support procedures of indexed columns.  We
     * don't look them up here; we do that lazily the first time we see a scan
     * key reference each of them.  We rely on zeroing fn_oid to InvalidOid.
     */
    consistentFn = palloc0_array(FmgrInfo, bdesc->bd_tupdesc->natts);
 
    /*
     * Make room for per-attribute lists of scan keys that we'll pass to the
     * consistent support procedure. We don't know which attributes have scan
     * keys, so we allocate space for all attributes. That may use more memory
     * but it's probably cheaper than determining which attributes are used.
     *
     * We keep null and regular keys separate, so that we can pass just the
     * regular keys to the consistent function easily.
     *
     * To reduce the allocation overhead, we allocate one big chunk and then
     * carve it into smaller arrays ourselves. All the pieces have exactly the
     * same lifetime, so that's OK.
     *
     * XXX The widest index can have 32 attributes, so the amount of wasted
     * memory is negligible. We could invent a more compact approach (with
     * just space for used attributes) but that would make the matching more
     * complex so it's not a good trade-off.
     */
    len =
        MAXALIGN(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts) +    /* regular keys */
        MAXALIGN(sizeof(ScanKey) * scan->numberOfKeys) * bdesc->bd_tupdesc->natts +
        MAXALIGN(sizeof(int) * bdesc->bd_tupdesc->natts) +
        MAXALIGN(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts) +    /* NULL keys */
        MAXALIGN(sizeof(ScanKey) * scan->numberOfKeys) * bdesc->bd_tupdesc->natts +
        MAXALIGN(sizeof(int) * bdesc->bd_tupdesc->natts);
 
    ptr = palloc(len);
    tmp = ptr;
 
    keys = (ScanKey **) ptr;
    ptr += MAXALIGN(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts);
 
    nullkeys = (ScanKey **) ptr;
    ptr += MAXALIGN(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts);
 
    nkeys = (int *) ptr;
    ptr += MAXALIGN(sizeof(int) * bdesc->bd_tupdesc->natts);
 
    nnullkeys = (int *) ptr;
    ptr += MAXALIGN(sizeof(int) * bdesc->bd_tupdesc->natts);
 
    for (int i = 0; i < bdesc->bd_tupdesc->natts; i++)
    {
        keys[i] = (ScanKey *) ptr;
        ptr += MAXALIGN(sizeof(ScanKey) * scan->numberOfKeys);
 
        nullkeys[i] = (ScanKey *) ptr;
        ptr += MAXALIGN(sizeof(ScanKey) * scan->numberOfKeys);
    }
 
    Assert(tmp + len == ptr);
 
    /* zero the number of keys */
    memset(nkeys, 0, sizeof(int) * bdesc->bd_tupdesc->natts);
    memset(nnullkeys, 0, sizeof(int) * bdesc->bd_tupdesc->natts);
 
    /* Preprocess the scan keys - split them into per-attribute arrays. */
    for (int keyno = 0; keyno < scan->numberOfKeys; keyno++)
    {
        ScanKey     key = &scan->keyData[keyno];
        AttrNumber  keyattno = key->sk_attno;
 
        /*
         * The collation of the scan key must match the collation used in the
         * index column (but only if the search is not IS NULL/ IS NOT NULL).
         * Otherwise we shouldn't be using this index ...
         */
        Assert((key->sk_flags & SK_ISNULL) ||
               (key->sk_collation ==
                TupleDescAttr(bdesc->bd_tupdesc,
                              keyattno - 1)->attcollation));
 
        /*
         * First time we see this index attribute, so init as needed.
         *
         * This is a bit of an overkill - we don't know how many scan keys are
         * there for this attribute, so we simply allocate the largest number
         * possible (as if all keys were for this attribute). This may waste a
         * bit of memory, but we only expect small number of scan keys in
         * general, so this should be negligible, and repeated repalloc calls
         * are not free either.
         */
        if (consistentFn[keyattno - 1].fn_oid == InvalidOid)
        {
            FmgrInfo   *tmp;
 
            /* First time we see this attribute, so no key/null keys. */
            Assert(nkeys[keyattno - 1] == 0);
            Assert(nnullkeys[keyattno - 1] == 0);
 
            tmp = index_getprocinfo(idxRel, keyattno,
                                    BRIN_PROCNUM_CONSISTENT);
            fmgr_info_copy(&consistentFn[keyattno - 1], tmp,
                           CurrentMemoryContext);
        }
 
        /* Add key to the proper per-attribute array. */
        if (key->sk_flags & SK_ISNULL)
        {
            nullkeys[keyattno - 1][nnullkeys[keyattno - 1]] = key;
            nnullkeys[keyattno - 1]++;
        }
        else
        {
            keys[keyattno - 1][nkeys[keyattno - 1]] = key;
            nkeys[keyattno - 1]++;
        }
    }
 
    /* allocate an initial in-memory tuple, out of the per-range memcxt */
    dtup = brin_new_memtuple(bdesc);
 
    /*
     * Setup and use a per-range memory context, which is reset every time we
     * loop below.  This avoids having to free the tuples within the loop.
     */
    perRangeCxt = AllocSetContextCreate(CurrentMemoryContext,
                                        "bringetbitmap cxt",
                                        ALLOCSET_DEFAULT_SIZES);
    oldcxt = MemoryContextSwitchTo(perRangeCxt);
 
    /*
     * Now scan the revmap.  We start by querying for heap page 0,
     * incrementing by the number of pages per range; this gives us a full
     * view of the table.  We make use of uint64 for heapBlk as a BlockNumber
     * could wrap for tables with close to 2^32 pages.
     */
    for (uint64 heapBlk = 0; heapBlk < nblocks; heapBlk += opaque->bo_pagesPerRange)
    {
        bool        addrange;
        bool        gottuple = false;
        BrinTuple  *tup;
        OffsetNumber off;
        Size        size;
 
        CHECK_FOR_INTERRUPTS();
 
        MemoryContextReset(perRangeCxt);
 
        tup = brinGetTupleForHeapBlock(opaque->bo_rmAccess, (BlockNumber) heapBlk, &buf,
                                       &off, &size, BUFFER_LOCK_SHARE);
        if (tup)
        {
            gottuple = true;
            btup = brin_copy_tuple(tup, size, btup, &btupsz);
            LockBuffer(buf, BUFFER_LOCK_UNLOCK);
        }
 
        /*
         * For page ranges with no indexed tuple, we must return the whole
         * range; otherwise, compare it to the scan keys.
         */
        if (!gottuple)
        {
            addrange = true;
        }
        else
        {
            dtup = brin_deform_tuple(bdesc, btup, dtup);
            if (dtup->bt_placeholder)
            {
                /*
                 * Placeholder tuples are always returned, regardless of the
                 * values stored in them.
                 */
                addrange = true;
            }
            else
            {
                int         attno;
 
                /*
                 * Compare scan keys with summary values stored for the range.
                 * If scan keys are matched, the page range must be added to
                 * the bitmap.  We initially assume the range needs to be
                 * added; in particular this serves the case where there are
                 * no keys.
                 */
                addrange = true;
                for (attno = 1; attno <= bdesc->bd_tupdesc->natts; attno++)
                {
                    BrinValues *bval;
                    Datum       add;
                    Oid         collation;
 
                    /*
                     * skip attributes without any scan keys (both regular and
                     * IS [NOT] NULL)
                     */
                    if (nkeys[attno - 1] == 0 && nnullkeys[attno - 1] == 0)
                        continue;
 
                    bval = &dtup->bt_columns[attno - 1];
 
                    /*
                     * If the BRIN tuple indicates that this range is empty,
                     * we can skip it: there's nothing to match.  We don't
                     * need to examine the next columns.
                     */
                    if (dtup->bt_empty_range)
                    {
                        addrange = false;
                        break;
                    }
 
                    /*
                     * First check if there are any IS [NOT] NULL scan keys,
                     * and if we're violating them. In that case we can
                     * terminate early, without invoking the support function.
                     *
                     * As there may be more keys, we can only determine
                     * mismatch within this loop.
                     */
                    if (bdesc->bd_info[attno - 1]->oi_regular_nulls &&
                        !check_null_keys(bval, nullkeys[attno - 1],
                                         nnullkeys[attno - 1]))
                    {
                        /*
                         * If any of the IS [NOT] NULL keys failed, the page
                         * range as a whole can't pass. So terminate the loop.
                         */
                        addrange = false;
                        break;
                    }
 
                    /*
                     * So either there are no IS [NOT] NULL keys, or all
                     * passed. If there are no regular scan keys, we're done -
                     * the page range matches. If there are regular keys, but
                     * the page range is marked as 'all nulls' it can't
                     * possibly pass (we're assuming the operators are
                     * strict).
                     */
 
                    /* No regular scan keys - page range as a whole passes. */
                    if (!nkeys[attno - 1])
                        continue;
 
                    Assert((nkeys[attno - 1] > 0) &&
                           (nkeys[attno - 1] <= scan->numberOfKeys));
 
                    /* If it is all nulls, it cannot possibly be consistent. */
                    if (bval->bv_allnulls)
                    {
                        addrange = false;
                        break;
                    }
 
                    /*
                     * Collation from the first key (has to be the same for
                     * all keys for the same attribute).
                     */
                    collation = keys[attno - 1][0]->sk_collation;
 
                    /*
                     * Check whether the scan key is consistent with the page
                     * range values; if so, have the pages in the range added
                     * to the output bitmap.
                     *
                     * The opclass may or may not support processing of
                     * multiple scan keys. We can determine that based on the
                     * number of arguments - functions with extra parameter
                     * (number of scan keys) do support this, otherwise we
                     * have to simply pass the scan keys one by one.
                     */
                    if (consistentFn[attno - 1].fn_nargs >= 4)
                    {
                        /* Check all keys at once */
                        add = FunctionCall4Coll(&consistentFn[attno - 1],
                                                collation,
                                                PointerGetDatum(bdesc),
                                                PointerGetDatum(bval),
                                                PointerGetDatum(keys[attno - 1]),
                                                Int32GetDatum(nkeys[attno - 1]));
                        addrange = DatumGetBool(add);
                    }
                    else
                    {
                        /*
                         * Check keys one by one
                         *
                         * When there are multiple scan keys, failure to meet
                         * the criteria for a single one of them is enough to
                         * discard the range as a whole, so break out of the
                         * loop as soon as a false return value is obtained.
                         */
                        int         keyno;
 
                        for (keyno = 0; keyno < nkeys[attno - 1]; keyno++)
                        {
                            add = FunctionCall3Coll(&consistentFn[attno - 1],
                                                    keys[attno - 1][keyno]->sk_collation,
                                                    PointerGetDatum(bdesc),
                                                    PointerGetDatum(bval),
                                                    PointerGetDatum(keys[attno - 1][keyno]));
                            addrange = DatumGetBool(add);
                            if (!addrange)
                                break;
                        }
                    }
 
                    /*
                     * If we found a scan key eliminating the range, no need
                     * to check additional ones.
                     */
                    if (!addrange)
                        break;
                }
            }
        }
 
        /* add the pages in the range to the output bitmap, if needed */
        if (addrange)
        {
            uint64      pageno;
 
            for (pageno = heapBlk;
                 pageno <= Min(nblocks, heapBlk + opaque->bo_pagesPerRange) - 1;
                 pageno++)
            {
                MemoryContextSwitchTo(oldcxt);
                tbm_add_page(tbm, pageno);
                totalpages++;
                MemoryContextSwitchTo(perRangeCxt);
            }
        }
    }
 
    MemoryContextSwitchTo(oldcxt);
    MemoryContextDelete(perRangeCxt);
 
    if (buf != InvalidBuffer)
        ReleaseBuffer(buf);
 
    /*
     * XXX We have an approximation of the number of *pages* that our scan
     * returns, but we don't have a precise idea of the number of heap tuples
     * involved.
     */
    return totalpages * 10;
}
 
/*
 * Re-initialize state for a BRIN index scan
 */
void
brinrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,
           ScanKey orderbys, int norderbys)
{
    /*
     * Other index AMs preprocess the scan keys at this point, or sometime
     * early during the scan; this lets them optimize by removing redundant
     * keys, or doing early returns when they are impossible to satisfy; see
     * _bt_preprocess_keys for an example.  Something like that could be added
     * here someday, too.
     */
 
    if (scankey && scan->numberOfKeys > 0)
        memcpy(scan->keyData, scankey, scan->numberOfKeys * sizeof(ScanKeyData));
}
 
/*
 * Close down a BRIN index scan
 */
void
brinendscan(IndexScanDesc scan)
{
    BrinOpaque *opaque = (BrinOpaque *) scan->opaque;
 
    brinRevmapTerminate(opaque->bo_rmAccess);
    brin_free_desc(opaque->bo_bdesc);
    pfree(opaque);
}
 
/*
 * Per-heap-tuple callback for table_index_build_scan.
 *
 * Note we don't worry about the page range at the end of the table here; it is
 * present in the build state struct after we're called the last time, but not
 * inserted into the index.  Caller must ensure to do so, if appropriate.
 */
static void
brinbuildCallback(Relation index,
                  ItemPointer tid,
                  Datum *values,
                  bool *isnull,
                  bool tupleIsAlive,
                  void *brstate)
{
    BrinBuildState *state = (BrinBuildState *) brstate;
    BlockNumber thisblock;
 
    thisblock = ItemPointerGetBlockNumber(tid);
 
    /*
     * If we're in a block that belongs to a future range, summarize what
     * we've got and start afresh.  Note the scan might have skipped many
     * pages, if they were devoid of live tuples; make sure to insert index
     * tuples for those too.
     */
    while (thisblock > state->bs_currRangeStart + state->bs_pagesPerRange - 1)
    {
 
        BRIN_elog((DEBUG2,
                   "brinbuildCallback: completed a range: %u--%u",
                   state->bs_currRangeStart,
                   state->bs_currRangeStart + state->bs_pagesPerRange));
 
        /* create the index tuple and insert it */
        form_and_insert_tuple(state);
 
        /* set state to correspond to the next range */
        state->bs_currRangeStart += state->bs_pagesPerRange;
 
        /* re-initialize state for it */
        brin_memtuple_initialize(state->bs_dtuple, state->bs_bdesc);
    }
 
    /* Accumulate the current tuple into the running state */
    (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple,
                               values, isnull);
}
 
/*
 * Per-heap-tuple callback for table_index_build_scan with parallelism.
 *
 * A version of the callback used by parallel index builds. The main difference
 * is that instead of writing the BRIN tuples into the index, we write them
 * into a shared tuplesort, and leave the insertion up to the leader (which may
 * reorder them a bit etc.). The callback also does not generate empty ranges,
 * those will be added by the leader when merging results from workers.
 */
static void
brinbuildCallbackParallel(Relation index,
                          ItemPointer tid,
                          Datum *values,
                          bool *isnull,
                          bool tupleIsAlive,
                          void *brstate)
{
    BrinBuildState *state = (BrinBuildState *) brstate;
    BlockNumber thisblock;
 
    thisblock = ItemPointerGetBlockNumber(tid);
 
    /*
     * If we're in a block that belongs to a different range, summarize what
     * we've got and start afresh.  Note the scan might have skipped many
     * pages, if they were devoid of live tuples; we do not create empty BRIN
     * ranges here - the leader is responsible for filling them in.
     *
     * Unlike serial builds, parallel index builds allow synchronized seqscans
     * (because that's what parallel scans do). This means the block may wrap
     * around to the beginning of the relation, so the condition needs to
     * check for both future and past ranges.
     */
    if ((thisblock < state->bs_currRangeStart) ||
        (thisblock > state->bs_currRangeStart + state->bs_pagesPerRange - 1))
    {
 
        BRIN_elog((DEBUG2,
                   "brinbuildCallbackParallel: completed a range: %u--%u",
                   state->bs_currRangeStart,
                   state->bs_currRangeStart + state->bs_pagesPerRange));
 
        /* create the index tuple and write it into the tuplesort */
        form_and_spill_tuple(state);
 
        /*
         * Set state to correspond to the next range (for this block).
         *
         * This skips ranges that are either empty (and so we don't get any
         * tuples to summarize), or processed by other workers. We can't
         * differentiate those cases here easily, so we leave it up to the
         * leader to fill empty ranges where needed.
         */
        state->bs_currRangeStart
            = state->bs_pagesPerRange * (thisblock / state->bs_pagesPerRange);
 
        /* re-initialize state for it */
        brin_memtuple_initialize(state->bs_dtuple, state->bs_bdesc);
    }
 
    /* Accumulate the current tuple into the running state */
    (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple,
                               values, isnull);
}
 
/*
 * brinbuild() -- build a new BRIN index.
 */
IndexBuildResult *
brinbuild(Relation heap, Relation index, IndexInfo *indexInfo)
{
    IndexBuildResult *result;
    double      reltuples;
    double      idxtuples;
    BrinRevmap *revmap;
    BrinBuildState *state;
    Buffer      meta;
    BlockNumber pagesPerRange;
 
    /*
     * We expect to be called exactly once for any index relation.
     */
    if (RelationGetNumberOfBlocks(index) != 0)
        elog(ERROR, "index \"%s\" already contains data",
             RelationGetRelationName(index));
 
    /*
     * Critical section not required, because on error the creation of the
     * whole relation will be rolled back.
     */
 
    meta = ExtendBufferedRel(BMR_REL(index), MAIN_FORKNUM, NULL,
                             EB_LOCK_FIRST | EB_SKIP_EXTENSION_LOCK);
    Assert(BufferGetBlockNumber(meta) == BRIN_METAPAGE_BLKNO);
 
    brin_metapage_init(BufferGetPage(meta), BrinGetPagesPerRange(index),
                       BRIN_CURRENT_VERSION);
    MarkBufferDirty(meta);
 
    if (RelationNeedsWAL(index))
    {
        xl_brin_createidx xlrec;
        XLogRecPtr  recptr;
        Page        page;
 
        xlrec.version = BRIN_CURRENT_VERSION;
        xlrec.pagesPerRange = BrinGetPagesPerRange(index);
 
        XLogBeginInsert();
        XLogRegisterData(&xlrec, SizeOfBrinCreateIdx);
        XLogRegisterBuffer(0, meta, REGBUF_WILL_INIT | REGBUF_STANDARD);
 
        recptr = XLogInsert(RM_BRIN_ID, XLOG_BRIN_CREATE_INDEX);
 
        page = BufferGetPage(meta);
        PageSetLSN(page, recptr);
    }
 
    UnlockReleaseBuffer(meta);
 
    /*
     * Initialize our state, including the deformed tuple state.
     */
    revmap = brinRevmapInitialize(index, &pagesPerRange);
    state = initialize_brin_buildstate(index, revmap, pagesPerRange,
                                       RelationGetNumberOfBlocks(heap));
 
    /*
     * Attempt to launch parallel worker scan when required
     *
     * XXX plan_create_index_workers makes the number of workers dependent on
     * maintenance_work_mem, requiring 32MB for each worker. That makes sense
     * for btree, but not for BRIN, which can do with much less memory. So
     * maybe make that somehow less strict, optionally?
     */
    if (indexInfo->ii_ParallelWorkers > 0)
        _brin_begin_parallel(state, heap, index, indexInfo->ii_Concurrent,
                             indexInfo->ii_ParallelWorkers);
 
    /*
     * If parallel build requested and at least one worker process was
     * successfully launched, set up coordination state, wait for workers to
     * complete. Then read all tuples from the shared tuplesort and insert
     * them into the index.
     *
     * In serial mode, simply scan the table and build the index one index
     * tuple at a time.
     */
    if (state->bs_leader)
    {
        SortCoordinate coordinate;
 
        coordinate = palloc0_object(SortCoordinateData);
        coordinate->isWorker = false;
        coordinate->nParticipants =
            state->bs_leader->nparticipanttuplesorts;
        coordinate->sharedsort = state->bs_leader->sharedsort;
 
        /*
         * Begin leader tuplesort.
         *
         * In cases where parallelism is involved, the leader receives the
         * same share of maintenance_work_mem as a serial sort (it is
         * generally treated in the same way as a serial sort once we return).
         * Parallel worker Tuplesortstates will have received only a fraction
         * of maintenance_work_mem, though.
         *
         * We rely on the lifetime of the Leader Tuplesortstate almost not
         * overlapping with any worker Tuplesortstate's lifetime.  There may
         * be some small overlap, but that's okay because we rely on leader
         * Tuplesortstate only allocating a small, fixed amount of memory
         * here. When its tuplesort_performsort() is called (by our caller),
         * and significant amounts of memory are likely to be used, all
         * workers must have already freed almost all memory held by their
         * Tuplesortstates (they are about to go away completely, too).  The
         * overall effect is that maintenance_work_mem always represents an
         * absolute high watermark on the amount of memory used by a CREATE
         * INDEX operation, regardless of the use of parallelism or any other
         * factor.
         */
        state->bs_sortstate =
            tuplesort_begin_index_brin(maintenance_work_mem, coordinate,
                                       TUPLESORT_NONE);
 
        /* scan the relation and merge per-worker results */
        reltuples = _brin_parallel_merge(state);
 
        _brin_end_parallel(state->bs_leader, state);
    }
    else                        /* no parallel index build */
    {
        /*
         * Now scan the relation.  No syncscan allowed here because we want
         * the heap blocks in physical order (we want to produce the ranges
         * starting from block 0, and the callback also relies on this to not
         * generate summary for the same range twice).
         */
        reltuples = table_index_build_scan(heap, index, indexInfo, false, true,
                                           brinbuildCallback, state, NULL);
 
        /*
         * process the final batch
         *
         * XXX Note this does not update state->bs_currRangeStart, i.e. it
         * stays set to the last range added to the index. This is OK, because
         * that's what brin_fill_empty_ranges expects.
         */
        form_and_insert_tuple(state);
 
        /*
         * Backfill the final ranges with empty data.
         *
         * This saves us from doing what amounts to full table scans when the
         * index with a predicate like WHERE (nonnull_column IS NULL), or
         * other very selective predicates.
         */
        brin_fill_empty_ranges(state,
                               state->bs_currRangeStart,
                               state->bs_maxRangeStart);
    }
 
    /* release resources */
    idxtuples = state->bs_numtuples;
    brinRevmapTerminate(state->bs_rmAccess);
    terminate_brin_buildstate(state);
 
    /*
     * Return statistics
     */
    result = palloc_object(IndexBuildResult);
 
    result->heap_tuples = reltuples;
    result->index_tuples = idxtuples;
 
    return result;
}
 
void
brinbuildempty(Relation index)
{
    Buffer      metabuf;
 
    /* An empty BRIN index has a metapage only. */
    metabuf = ExtendBufferedRel(BMR_REL(index), INIT_FORKNUM, NULL,
                                EB_LOCK_FIRST | EB_SKIP_EXTENSION_LOCK);
 
    /* Initialize and xlog metabuffer. */
    START_CRIT_SECTION();
    brin_metapage_init(BufferGetPage(metabuf), BrinGetPagesPerRange(index),
                       BRIN_CURRENT_VERSION);
    MarkBufferDirty(metabuf);
    log_newpage_buffer(metabuf, true);
    END_CRIT_SECTION();
 
    UnlockReleaseBuffer(metabuf);
}
 
/*
 * brinbulkdelete
 *      Since there are no per-heap-tuple index tuples in BRIN indexes,
 *      there's not a lot we can do here.
 *
 * XXX we could mark item tuples as "dirty" (when a minimum or maximum heap
 * tuple is deleted), meaning the need to re-run summarization on the affected
 * range.  Would need to add an extra flag in brintuples for that.
 */
IndexBulkDeleteResult *
brinbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
               IndexBulkDeleteCallback callback, void *callback_state)
{
    /* allocate stats if first time through, else re-use existing struct */
    if (stats == NULL)
        stats = palloc0_object(IndexBulkDeleteResult);
 
    return stats;
}
 
/*
 * This routine is in charge of "vacuuming" a BRIN index: we just summarize
 * ranges that are currently unsummarized.
 */
IndexBulkDeleteResult *
brinvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats)
{
    Relation    heapRel;
 
    /* No-op in ANALYZE ONLY mode */
    if (info->analyze_only)
        return stats;
 
    if (!stats)
        stats = palloc0_object(IndexBulkDeleteResult);
    stats->num_pages = RelationGetNumberOfBlocks(info->index);
    /* rest of stats is initialized by zeroing */
 
    heapRel = table_open(IndexGetRelation(RelationGetRelid(info->index), false),
                         AccessShareLock);
 
    brin_vacuum_scan(info->index, info->strategy);
 
    brinsummarize(info->index, heapRel, BRIN_ALL_BLOCKRANGES, false,
                  &stats->num_index_tuples, &stats->num_index_tuples);
 
    table_close(heapRel, AccessShareLock);
 
    return stats;
}
 
/*
 * reloptions processor for BRIN indexes
 */
bytea *
brinoptions(Datum reloptions, bool validate)
{
    static const relopt_parse_elt tab[] = {
        {"pages_per_range", RELOPT_TYPE_INT, offsetof(BrinOptions, pagesPerRange)},
        {"autosummarize", RELOPT_TYPE_BOOL, offsetof(BrinOptions, autosummarize)}
    };
 
    return (bytea *) build_reloptions(reloptions, validate,
                                      RELOPT_KIND_BRIN,
                                      sizeof(BrinOptions),
                                      tab, lengthof(tab));
}
 
/*
 * SQL-callable function to scan through an index and summarize all ranges
 * that are not currently summarized.
 */
Datum
brin_summarize_new_values(PG_FUNCTION_ARGS)
{
    Datum       relation = PG_GETARG_DATUM(0);
 
    return DirectFunctionCall2(brin_summarize_range,
                               relation,
                               Int64GetDatum((int64) BRIN_ALL_BLOCKRANGES));
}
 
/*
 * SQL-callable function to summarize the indicated page range, if not already
 * summarized.  If the second argument is BRIN_ALL_BLOCKRANGES, all
 * unsummarized ranges are summarized.
 */
Datum
brin_summarize_range(PG_FUNCTION_ARGS)
{
    Oid         indexoid = PG_GETARG_OID(0);
    int64       heapBlk64 = PG_GETARG_INT64(1);
    BlockNumber heapBlk;
    Oid         heapoid;
    Relation    indexRel;
    Relation    heapRel;
    Oid         save_userid;
    int         save_sec_context;
    int         save_nestlevel;
    double      numSummarized = 0;
 
    if (RecoveryInProgress())
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("recovery is in progress"),
                 errhint("BRIN control functions cannot be executed during recovery.")));
 
    if (heapBlk64 > BRIN_ALL_BLOCKRANGES || heapBlk64 < 0)
        ereport(ERROR,
                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                 errmsg("block number out of range: %" PRId64, heapBlk64)));
    heapBlk = (BlockNumber) heapBlk64;
 
    /*
     * We must lock table before index to avoid deadlocks.  However, if the
     * passed indexoid isn't an index then IndexGetRelation() will fail.
     * Rather than emitting a not-very-helpful error message, postpone
     * complaining, expecting that the is-it-an-index test below will fail.
     */
    heapoid = IndexGetRelation(indexoid, true);
    if (OidIsValid(heapoid))
    {
        heapRel = table_open(heapoid, ShareUpdateExclusiveLock);
 
        /*
         * Autovacuum calls us.  For its benefit, switch to the table owner's
         * userid, so that any index functions are run as that user.  Also
         * lock down security-restricted operations and arrange to make GUC
         * variable changes local to this command.  This is harmless, albeit
         * unnecessary, when called from SQL, because we fail shortly if the
         * user does not own the index.
         */
        GetUserIdAndSecContext(&save_userid, &save_sec_context);
        SetUserIdAndSecContext(heapRel->rd_rel->relowner,
                               save_sec_context | SECURITY_RESTRICTED_OPERATION);
        save_nestlevel = NewGUCNestLevel();
        RestrictSearchPath();
    }
    else
    {
        heapRel = NULL;
        /* Set these just to suppress "uninitialized variable" warnings */
        save_userid = InvalidOid;
        save_sec_context = -1;
        save_nestlevel = -1;
    }
 
    indexRel = index_open(indexoid, ShareUpdateExclusiveLock);
 
    /* Must be a BRIN index */
    if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
        indexRel->rd_rel->relam != BRIN_AM_OID)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not a BRIN index",
                        RelationGetRelationName(indexRel))));
 
    /* User must own the index (comparable to privileges needed for VACUUM) */
    if (heapRel != NULL && !object_ownercheck(RelationRelationId, indexoid, save_userid))
        aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
                       RelationGetRelationName(indexRel));
 
    /*
     * Since we did the IndexGetRelation call above without any lock, it's
     * barely possible that a race against an index drop/recreation could have
     * netted us the wrong table.  Recheck.
     */
    if (heapRel == NULL || heapoid != IndexGetRelation(indexoid, false))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_TABLE),
                 errmsg("could not open parent table of index \"%s\"",
                        RelationGetRelationName(indexRel))));
 
    /* see gin_clean_pending_list() */
    if (indexRel->rd_index->indisvalid)
        brinsummarize(indexRel, heapRel, heapBlk, true, &numSummarized, NULL);
    else
        ereport(DEBUG1,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("index \"%s\" is not valid",
                        RelationGetRelationName(indexRel))));
 
    /* Roll back any GUC changes executed by index functions */
    AtEOXact_GUC(false, save_nestlevel);
 
    /* Restore userid and security context */
    SetUserIdAndSecContext(save_userid, save_sec_context);
 
    index_close(indexRel, ShareUpdateExclusiveLock);
    table_close(heapRel, ShareUpdateExclusiveLock);
 
    PG_RETURN_INT32((int32) numSummarized);
}
 
/*
 * SQL-callable interface to mark a range as no longer summarized
 */
Datum
brin_desummarize_range(PG_FUNCTION_ARGS)
{
    Oid         indexoid = PG_GETARG_OID(0);
    int64       heapBlk64 = PG_GETARG_INT64(1);
    BlockNumber heapBlk;
    Oid         heapoid;
    Relation    heapRel;
    Relation    indexRel;
    bool        done;
 
    if (RecoveryInProgress())
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("recovery is in progress"),
                 errhint("BRIN control functions cannot be executed during recovery.")));
 
    if (heapBlk64 > MaxBlockNumber || heapBlk64 < 0)
        ereport(ERROR,
                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                 errmsg("block number out of range: %" PRId64,
                        heapBlk64)));
    heapBlk = (BlockNumber) heapBlk64;
 
    /*
     * We must lock table before index to avoid deadlocks.  However, if the
     * passed indexoid isn't an index then IndexGetRelation() will fail.
     * Rather than emitting a not-very-helpful error message, postpone
     * complaining, expecting that the is-it-an-index test below will fail.
     *
     * Unlike brin_summarize_range(), autovacuum never calls this.  Hence, we
     * don't switch userid.
     */
    heapoid = IndexGetRelation(indexoid, true);
    if (OidIsValid(heapoid))
        heapRel = table_open(heapoid, ShareUpdateExclusiveLock);
    else
        heapRel = NULL;
 
    indexRel = index_open(indexoid, ShareUpdateExclusiveLock);
 
    /* Must be a BRIN index */
    if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
        indexRel->rd_rel->relam != BRIN_AM_OID)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not a BRIN index",
                        RelationGetRelationName(indexRel))));
 
    /* User must own the index (comparable to privileges needed for VACUUM) */
    if (!object_ownercheck(RelationRelationId, indexoid, GetUserId()))
        aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
                       RelationGetRelationName(indexRel));
 
    /*
     * Since we did the IndexGetRelation call above without any lock, it's
     * barely possible that a race against an index drop/recreation could have
     * netted us the wrong table.  Recheck.
     */
    if (heapRel == NULL || heapoid != IndexGetRelation(indexoid, false))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_TABLE),
                 errmsg("could not open parent table of index \"%s\"",
                        RelationGetRelationName(indexRel))));
 
    /* see gin_clean_pending_list() */
    if (indexRel->rd_index->indisvalid)
    {
        /* the revmap does the hard work */
        do
        {
            done = brinRevmapDesummarizeRange(indexRel, heapBlk);
        }
        while (!done);
    }
    else
        ereport(DEBUG1,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("index \"%s\" is not valid",
                        RelationGetRelationName(indexRel))));
 
    index_close(indexRel, ShareUpdateExclusiveLock);
    table_close(heapRel, ShareUpdateExclusiveLock);
 
    PG_RETURN_VOID();
}
 
/*
 * Build a BrinDesc used to create or scan a BRIN index
 */
BrinDesc *
brin_build_desc(Relation rel)
{
    BrinOpcInfo **opcinfo;
    BrinDesc   *bdesc;
    TupleDesc   tupdesc;
    int         totalstored = 0;
    int         keyno;
    long        totalsize;
    MemoryContext cxt;
    MemoryContext oldcxt;
 
    cxt = AllocSetContextCreate(CurrentMemoryContext,
                                "brin desc cxt",
                                ALLOCSET_SMALL_SIZES);
    oldcxt = MemoryContextSwitchTo(cxt);
    tupdesc = RelationGetDescr(rel);
 
    /*
     * Obtain BrinOpcInfo for each indexed column.  While at it, accumulate
     * the number of columns stored, since the number is opclass-defined.
     */
    opcinfo = palloc_array(BrinOpcInfo *, tupdesc->natts);
    for (keyno = 0; keyno < tupdesc->natts; keyno++)
    {
        FmgrInfo   *opcInfoFn;
        Form_pg_attribute attr = TupleDescAttr(tupdesc, keyno);
 
        opcInfoFn = index_getprocinfo(rel, keyno + 1, BRIN_PROCNUM_OPCINFO);
 
        opcinfo[keyno] = (BrinOpcInfo *)
            DatumGetPointer(FunctionCall1(opcInfoFn, ObjectIdGetDatum(attr->atttypid)));
        totalstored += opcinfo[keyno]->oi_nstored;
    }
 
    /* Allocate our result struct and fill it in */
    totalsize = offsetof(BrinDesc, bd_info) +
        sizeof(BrinOpcInfo *) * tupdesc->natts;
 
    bdesc = palloc(totalsize);
    bdesc->bd_context = cxt;
    bdesc->bd_index = rel;
    bdesc->bd_tupdesc = tupdesc;
    bdesc->bd_disktdesc = NULL; /* generated lazily */
    bdesc->bd_totalstored = totalstored;
 
    for (keyno = 0; keyno < tupdesc->natts; keyno++)
        bdesc->bd_info[keyno] = opcinfo[keyno];
    pfree(opcinfo);
 
    MemoryContextSwitchTo(oldcxt);
 
    return bdesc;
}
 
void
brin_free_desc(BrinDesc *bdesc)
{
    /* make sure the tupdesc is still valid */
    Assert(bdesc->bd_tupdesc->tdrefcount >= 1);
    /* no need for retail pfree */
    MemoryContextDelete(bdesc->bd_context);
}
 
/*
 * Fetch index's statistical data into *stats
 */
void
brinGetStats(Relation index, BrinStatsData *stats)
{
    Buffer      metabuffer;
    Page        metapage;
    BrinMetaPageData *metadata;
 
    metabuffer = ReadBuffer(index, BRIN_METAPAGE_BLKNO);
    LockBuffer(metabuffer, BUFFER_LOCK_SHARE);
    metapage = BufferGetPage(metabuffer);
    metadata = (BrinMetaPageData *) PageGetContents(metapage);
 
    stats->pagesPerRange = metadata->pagesPerRange;
    stats->revmapNumPages = metadata->lastRevmapPage - 1;
 
    UnlockReleaseBuffer(metabuffer);
}
 
/*
 * Initialize a BrinBuildState appropriate to create tuples on the given index.
 */
static BrinBuildState *
initialize_brin_buildstate(Relation idxRel, BrinRevmap *revmap,
                           BlockNumber pagesPerRange, BlockNumber tablePages)
{
    BrinBuildState *state;
    BlockNumber lastRange = 0;
 
    state = palloc_object(BrinBuildState);
 
    state->bs_irel = idxRel;
    state->bs_numtuples = 0;
    state->bs_reltuples = 0;
    state->bs_currentInsertBuf = InvalidBuffer;
    state->bs_pagesPerRange = pagesPerRange;
    state->bs_currRangeStart = 0;
    state->bs_rmAccess = revmap;
    state->bs_bdesc = brin_build_desc(idxRel);
    state->bs_dtuple = brin_new_memtuple(state->bs_bdesc);
    state->bs_leader = NULL;
    state->bs_worker_id = 0;
    state->bs_sortstate = NULL;
    state->bs_context = CurrentMemoryContext;
    state->bs_emptyTuple = NULL;
    state->bs_emptyTupleLen = 0;
 
    /* Remember the memory context to use for an empty tuple, if needed. */
    state->bs_context = CurrentMemoryContext;
    state->bs_emptyTuple = NULL;
    state->bs_emptyTupleLen = 0;
 
    /*
     * Calculate the start of the last page range. Page numbers are 0-based,
     * so to calculate the index we need to subtract one. The integer division
     * gives us the index of the page range.
     */
    if (tablePages > 0)
        lastRange = ((tablePages - 1) / pagesPerRange) * pagesPerRange;
 
    /* Now calculate the start of the next range. */
    state->bs_maxRangeStart = lastRange + state->bs_pagesPerRange;
 
    return state;
}
 
/*
 * Release resources associated with a BrinBuildState.
 */
static void
terminate_brin_buildstate(BrinBuildState *state)
{
    /*
     * Release the last index buffer used.  We might as well ensure that
     * whatever free space remains in that page is available in FSM, too.
     */
    if (!BufferIsInvalid(state->bs_currentInsertBuf))
    {
        Page        page;
        Size        freespace;
        BlockNumber blk;
 
        page = BufferGetPage(state->bs_currentInsertBuf);
        freespace = PageGetFreeSpace(page);
        blk = BufferGetBlockNumber(state->bs_currentInsertBuf);
        ReleaseBuffer(state->bs_currentInsertBuf);
        RecordPageWithFreeSpace(state->bs_irel, blk, freespace);
        FreeSpaceMapVacuumRange(state->bs_irel, blk, blk + 1);
    }
 
    brin_free_desc(state->bs_bdesc);
    pfree(state->bs_dtuple);
    pfree(state);
}
 
/*
 * On the given BRIN index, summarize the heap page range that corresponds
 * to the heap block number given.
 *
 * This routine can run in parallel with insertions into the heap.  To avoid
 * missing those values from the summary tuple, we first insert a placeholder
 * index tuple into the index, then execute the heap scan; transactions
 * concurrent with the scan update the placeholder tuple.  After the scan, we
 * union the placeholder tuple with the one computed by this routine.  The
 * update of the index value happens in a loop, so that if somebody updates
 * the placeholder tuple after we read it, we detect the case and try again.
 * This ensures that the concurrently inserted tuples are not lost.
 *
 * A further corner case is this routine being asked to summarize the partial
 * range at the end of the table.  heapNumBlocks is the (possibly outdated)
 * table size; if we notice that the requested range lies beyond that size,
 * we re-compute the table size after inserting the placeholder tuple, to
 * avoid missing pages that were appended recently.
 */
static void
summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
                BlockNumber heapBlk, BlockNumber heapNumBlks)
{
    Buffer      phbuf;
    BrinTuple  *phtup;
    Size        phsz;
    OffsetNumber offset;
    BlockNumber scanNumBlks;
 
    /*
     * Insert the placeholder tuple
     */
    phbuf = InvalidBuffer;
    phtup = brin_form_placeholder_tuple(state->bs_bdesc, heapBlk, &phsz);
    offset = brin_doinsert(state->bs_irel, state->bs_pagesPerRange,
                           state->bs_rmAccess, &phbuf,
                           heapBlk, phtup, phsz);
 
    /*
     * Compute range end.  We hold ShareUpdateExclusive lock on table, so it
     * cannot shrink concurrently (but it can grow).
     */
    Assert(heapBlk % state->bs_pagesPerRange == 0);
    if (heapBlk + state->bs_pagesPerRange > heapNumBlks)
    {
        /*
         * If we're asked to scan what we believe to be the final range on the
         * table (i.e. a range that might be partial) we need to recompute our
         * idea of what the latest page is after inserting the placeholder
         * tuple.  Anyone that grows the table later will update the
         * placeholder tuple, so it doesn't matter that we won't scan these
         * pages ourselves.  Careful: the table might have been extended
         * beyond the current range, so clamp our result.
         *
         * Fortunately, this should occur infrequently.
         */
        scanNumBlks = Min(RelationGetNumberOfBlocks(heapRel) - heapBlk,
                          state->bs_pagesPerRange);
    }
    else
    {
        /* Easy case: range is known to be complete */
        scanNumBlks = state->bs_pagesPerRange;
    }
 
    /*
     * Execute the partial heap scan covering the heap blocks in the specified
     * page range, summarizing the heap tuples in it.  This scan stops just
     * short of brinbuildCallback creating the new index entry.
     *
     * Note that it is critical we use the "any visible" mode of
     * table_index_build_range_scan here: otherwise, we would miss tuples
     * inserted by transactions that are still in progress, among other corner
     * cases.
     */
    state->bs_currRangeStart = heapBlk;
    table_index_build_range_scan(heapRel, state->bs_irel, indexInfo, false, true, false,
                                 heapBlk, scanNumBlks,
                                 brinbuildCallback, state, NULL);
 
    /*
     * Now we update the values obtained by the scan with the placeholder
     * tuple.  We do this in a loop which only terminates if we're able to
     * update the placeholder tuple successfully; if we are not, this means
     * somebody else modified the placeholder tuple after we read it.
     */
    for (;;)
    {
        BrinTuple  *newtup;
        Size        newsize;
        bool        didupdate;
        bool        samepage;
 
        CHECK_FOR_INTERRUPTS();
 
        /*
         * Update the summary tuple and try to update.
         */
        newtup = brin_form_tuple(state->bs_bdesc,
                                 heapBlk, state->bs_dtuple, &newsize);
        samepage = brin_can_do_samepage_update(phbuf, phsz, newsize);
        didupdate =
            brin_doupdate(state->bs_irel, state->bs_pagesPerRange,
                          state->bs_rmAccess, heapBlk, phbuf, offset,
                          phtup, phsz, newtup, newsize, samepage);
        brin_free_tuple(phtup);
        brin_free_tuple(newtup);
 
        /* If the update succeeded, we're done. */
        if (didupdate)
            break;
 
        /*
         * If the update didn't work, it might be because somebody updated the
         * placeholder tuple concurrently.  Extract the new version, union it
         * with the values we have from the scan, and start over.  (There are
         * other reasons for the update to fail, but it's simple to treat them
         * the same.)
         */
        phtup = brinGetTupleForHeapBlock(state->bs_rmAccess, heapBlk, &phbuf,
                                         &offset, &phsz, BUFFER_LOCK_SHARE);
        /* the placeholder tuple must exist */
        if (phtup == NULL)
            elog(ERROR, "missing placeholder tuple");
        phtup = brin_copy_tuple(phtup, phsz, NULL, NULL);
        LockBuffer(phbuf, BUFFER_LOCK_UNLOCK);
 
        /* merge it into the tuple from the heap scan */
        union_tuples(state->bs_bdesc, state->bs_dtuple, phtup);
    }
 
    ReleaseBuffer(phbuf);
}
 
/*
 * Summarize page ranges that are not already summarized.  If pageRange is
 * BRIN_ALL_BLOCKRANGES then the whole table is scanned; otherwise, only the
 * page range containing the given heap page number is scanned.
 * If include_partial is true, then the partial range at the end of the table
 * is summarized, otherwise not.
 *
 * For each new index tuple inserted, *numSummarized (if not NULL) is
 * incremented; for each existing tuple, *numExisting (if not NULL) is
 * incremented.
 */
static void
brinsummarize(Relation index, Relation heapRel, BlockNumber pageRange,
              bool include_partial, double *numSummarized, double *numExisting)
{
    BrinRevmap *revmap;
    BrinBuildState *state = NULL;
    IndexInfo  *indexInfo = NULL;
    BlockNumber heapNumBlocks;
    BlockNumber pagesPerRange;
    Buffer      buf;
    BlockNumber startBlk;
 
    revmap = brinRevmapInitialize(index, &pagesPerRange);
 
    /* determine range of pages to process */
    heapNumBlocks = RelationGetNumberOfBlocks(heapRel);
    if (pageRange == BRIN_ALL_BLOCKRANGES)
        startBlk = 0;
    else
    {
        startBlk = (pageRange / pagesPerRange) * pagesPerRange;
        heapNumBlocks = Min(heapNumBlocks, startBlk + pagesPerRange);
    }
    if (startBlk > heapNumBlocks)
    {
        /* Nothing to do if start point is beyond end of table */
        brinRevmapTerminate(revmap);
        return;
    }
 
    /*
     * Scan the revmap to find unsummarized items.
     */
    buf = InvalidBuffer;
    for (; startBlk < heapNumBlocks; startBlk += pagesPerRange)
    {
        BrinTuple  *tup;
        OffsetNumber off;
 
        /*
         * Unless requested to summarize even a partial range, go away now if
         * we think the next range is partial.  Caller would pass true when it
         * is typically run once bulk data loading is done
         * (brin_summarize_new_values), and false when it is typically the
         * result of arbitrarily-scheduled maintenance command (vacuuming).
         */
        if (!include_partial &&
            (startBlk + pagesPerRange > heapNumBlocks))
            break;
 
        CHECK_FOR_INTERRUPTS();
 
        tup = brinGetTupleForHeapBlock(revmap, startBlk, &buf, &off, NULL,
                                       BUFFER_LOCK_SHARE);
        if (tup == NULL)
        {
            /* no revmap entry for this heap range. Summarize it. */
            if (state == NULL)
            {
                /* first time through */
                Assert(!indexInfo);
                state = initialize_brin_buildstate(index, revmap,
                                                   pagesPerRange,
                                                   InvalidBlockNumber);
                indexInfo = BuildIndexInfo(index);
            }
            summarize_range(indexInfo, state, heapRel, startBlk, heapNumBlocks);
 
            /* and re-initialize state for the next range */
            brin_memtuple_initialize(state->bs_dtuple, state->bs_bdesc);
 
            if (numSummarized)
                *numSummarized += 1.0;
        }
        else
        {
            if (numExisting)
                *numExisting += 1.0;
            LockBuffer(buf, BUFFER_LOCK_UNLOCK);
        }
    }
 
    if (BufferIsValid(buf))
        ReleaseBuffer(buf);
 
    /* free resources */
    brinRevmapTerminate(revmap);
    if (state)
    {
        terminate_brin_buildstate(state);
        pfree(indexInfo);
    }
}
 
/*
 * Given a deformed tuple in the build state, convert it into the on-disk
 * format and insert it into the index, making the revmap point to it.
 */
static void
form_and_insert_tuple(BrinBuildState *state)
{
    BrinTuple  *tup;
    Size        size;
 
    tup = brin_form_tuple(state->bs_bdesc, state->bs_currRangeStart,
                          state->bs_dtuple, &size);
    brin_doinsert(state->bs_irel, state->bs_pagesPerRange, state->bs_rmAccess,
                  &state->bs_currentInsertBuf, state->bs_currRangeStart,
                  tup, size);
    state->bs_numtuples++;
 
    pfree(tup);
}
 
/*
 * Given a deformed tuple in the build state, convert it into the on-disk
 * format and write it to a (shared) tuplesort (the leader will insert it
 * into the index later).
 */
static void
form_and_spill_tuple(BrinBuildState *state)
{
    BrinTuple  *tup;
    Size        size;
 
    /* don't insert empty tuples in parallel build */
    if (state->bs_dtuple->bt_empty_range)
        return;
 
    tup = brin_form_tuple(state->bs_bdesc, state->bs_currRangeStart,
                          state->bs_dtuple, &size);
 
    /* write the BRIN tuple to the tuplesort */
    tuplesort_putbrintuple(state->bs_sortstate, tup, size);
 
    state->bs_numtuples++;
 
    pfree(tup);
}
 
/*
 * Given two deformed tuples, adjust the first one so that it's consistent
 * with the summary values in both.
 */
static void
union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b)
{
    int         keyno;
    BrinMemTuple *db;
    MemoryContext cxt;
    MemoryContext oldcxt;
 
    /* Use our own memory context to avoid retail pfree */
    cxt = AllocSetContextCreate(CurrentMemoryContext,
                                "brin union",
                                ALLOCSET_DEFAULT_SIZES);
    oldcxt = MemoryContextSwitchTo(cxt);
    db = brin_deform_tuple(bdesc, b, NULL);
    MemoryContextSwitchTo(oldcxt);
 
    /*
     * Check if the ranges are empty.
     *
     * If at least one of them is empty, we don't need to call per-key union
     * functions at all. If "b" is empty, we just use "a" as the result (it
     * might be empty fine, but that's fine). If "a" is empty but "b" is not,
     * we use "b" as the result (but we have to copy the data into "a" first).
     *
     * Only when both ranges are non-empty, we actually do the per-key merge.
     */
 
    /* If "b" is empty - ignore it and just use "a" (even if it's empty etc.). */
    if (db->bt_empty_range)
    {
        /* skip the per-key merge */
        MemoryContextDelete(cxt);
        return;
    }
 
    /*
     * Now we know "b" is not empty. If "a" is empty, then "b" is the result.
     * But we need to copy the data from "b" to "a" first, because that's how
     * we pass result out.
     *
     * We have to copy all the global/per-key flags etc. too.
     */
    if (a->bt_empty_range)
    {
        for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
        {
            int         i;
            BrinValues *col_a = &a->bt_columns[keyno];
            BrinValues *col_b = &db->bt_columns[keyno];
            BrinOpcInfo *opcinfo = bdesc->bd_info[keyno];
 
            col_a->bv_allnulls = col_b->bv_allnulls;
            col_a->bv_hasnulls = col_b->bv_hasnulls;
 
            /* If "b" has no data, we're done. */
            if (col_b->bv_allnulls)
                continue;
 
            for (i = 0; i < opcinfo->oi_nstored; i++)
                col_a->bv_values[i] =
                    datumCopy(col_b->bv_values[i],
                              opcinfo->oi_typcache[i]->typbyval,
                              opcinfo->oi_typcache[i]->typlen);
        }
 
        /* "a" started empty, but "b" was not empty, so remember that */
        a->bt_empty_range = false;
 
        /* skip the per-key merge */
        MemoryContextDelete(cxt);
        return;
    }
 
    /* Now we know neither range is empty. */
    for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
    {
        FmgrInfo   *unionFn;
        BrinValues *col_a = &a->bt_columns[keyno];
        BrinValues *col_b = &db->bt_columns[keyno];
        BrinOpcInfo *opcinfo = bdesc->bd_info[keyno];
 
        if (opcinfo->oi_regular_nulls)
        {
            /* Does the "b" summary represent any NULL values? */
            bool        b_has_nulls = (col_b->bv_hasnulls || col_b->bv_allnulls);
 
            /* Adjust "hasnulls". */
            if (!col_a->bv_allnulls && b_has_nulls)
                col_a->bv_hasnulls = true;
 
            /* If there are no values in B, there's nothing left to do. */
            if (col_b->bv_allnulls)
                continue;
 
            /*
             * Adjust "allnulls".  If A doesn't have values, just copy the
             * values from B into A, and we're done.  We cannot run the
             * operators in this case, because values in A might contain
             * garbage.  Note we already established that B contains values.
             *
             * Also adjust "hasnulls" in order not to forget the summary
             * represents NULL values. This is not redundant with the earlier
             * update, because that only happens when allnulls=false.
             */
            if (col_a->bv_allnulls)
            {
                int         i;
 
                col_a->bv_allnulls = false;
                col_a->bv_hasnulls = true;
 
                for (i = 0; i < opcinfo->oi_nstored; i++)
                    col_a->bv_values[i] =
                        datumCopy(col_b->bv_values[i],
                                  opcinfo->oi_typcache[i]->typbyval,
                                  opcinfo->oi_typcache[i]->typlen);
 
                continue;
            }
        }
 
        unionFn = index_getprocinfo(bdesc->bd_index, keyno + 1,
                                    BRIN_PROCNUM_UNION);
        FunctionCall3Coll(unionFn,
                          bdesc->bd_index->rd_indcollation[keyno],
                          PointerGetDatum(bdesc),
                          PointerGetDatum(col_a),
                          PointerGetDatum(col_b));
    }
 
    MemoryContextDelete(cxt);
}
 
/*
 * brin_vacuum_scan
 *      Do a complete scan of the index during VACUUM.
 *
 * This routine scans the complete index looking for uncataloged index pages,
 * i.e. those that might have been lost due to a crash after index extension
 * and such.
 */
static void
brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy)
{
    BlockRangeReadStreamPrivate p;
    ReadStream *stream;
    Buffer      buf;
 
    p.current_blocknum = 0;
    p.last_exclusive = RelationGetNumberOfBlocks(idxrel);
 
    /*
     * It is safe to use batchmode as block_range_read_stream_cb takes no
     * locks.
     */
    stream = read_stream_begin_relation(READ_STREAM_MAINTENANCE |
                                        READ_STREAM_FULL |
                                        READ_STREAM_USE_BATCHING,
                                        strategy,
                                        idxrel,
                                        MAIN_FORKNUM,
                                        block_range_read_stream_cb,
                                        &p,
                                        0);
 
    /*
     * Scan the index in physical order, and clean up any possible mess in
     * each page.
     */
    while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
    {
        CHECK_FOR_INTERRUPTS();
 
        brin_page_cleanup(idxrel, buf);
 
        ReleaseBuffer(buf);
    }
 
    read_stream_end(stream);
 
    /*
     * Update all upper pages in the index's FSM, as well.  This ensures not
     * only that we propagate leaf-page FSM updates made by brin_page_cleanup,
     * but also that any pre-existing damage or out-of-dateness is repaired.
     */
    FreeSpaceMapVacuum(idxrel);
}
 
static bool
add_values_to_range(Relation idxRel, BrinDesc *bdesc, BrinMemTuple *dtup,
                    const Datum *values, const bool *nulls)
{
    int         keyno;
 
    /* If the range starts empty, we're certainly going to modify it. */
    bool        modified = dtup->bt_empty_range;
 
    /*
     * Compare the key values of the new tuple to the stored index values; our
     * deformed tuple will get updated if the new tuple doesn't fit the
     * original range (note this means we can't break out of the loop early).
     * Make a note of whether this happens, so that we know to insert the
     * modified tuple later.
     */
    for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
    {
        Datum       result;
        BrinValues *bval;
        FmgrInfo   *addValue;
        bool        has_nulls;
 
        bval = &dtup->bt_columns[keyno];
 
        /*
         * Does the range have actual NULL values? Either of the flags can be
         * set, but we ignore the state before adding first row.
         *
         * We have to remember this, because we'll modify the flags and we
         * need to know if the range started as empty.
         */
        has_nulls = ((!dtup->bt_empty_range) &&
                     (bval->bv_hasnulls || bval->bv_allnulls));
 
        /*
         * If the value we're adding is NULL, handle it locally. Otherwise
         * call the BRIN_PROCNUM_ADDVALUE procedure.
         */
        if (bdesc->bd_info[keyno]->oi_regular_nulls && nulls[keyno])
        {
            /*
             * If the new value is null, we record that we saw it if it's the
             * first one; otherwise, there's nothing to do.
             */
            if (!bval->bv_hasnulls)
            {
                bval->bv_hasnulls = true;
                modified = true;
            }
 
            continue;
        }
 
        addValue = index_getprocinfo(idxRel, keyno + 1,
                                     BRIN_PROCNUM_ADDVALUE);
        result = FunctionCall4Coll(addValue,
                                   idxRel->rd_indcollation[keyno],
                                   PointerGetDatum(bdesc),
                                   PointerGetDatum(bval),
                                   values[keyno],
                                   BoolGetDatum(nulls[keyno]));
        /* if that returned true, we need to insert the updated tuple */
        modified |= DatumGetBool(result);
 
        /*
         * If the range was had actual NULL values (i.e. did not start empty),
         * make sure we don't forget about the NULL values. Either the
         * allnulls flag is still set to true, or (if the opclass cleared it)
         * we need to set hasnulls=true.
         *
         * XXX This can only happen when the opclass modified the tuple, so
         * the modified flag should be set.
         */
        if (has_nulls && !(bval->bv_hasnulls || bval->bv_allnulls))
        {
            Assert(modified);
            bval->bv_hasnulls = true;
        }
    }
 
    /*
     * After updating summaries for all the keys, mark it as not empty.
     *
     * If we're actually changing the flag value (i.e. tuple started as
     * empty), we should have modified the tuple. So we should not see empty
     * range that was not modified.
     */
    Assert(!dtup->bt_empty_range || modified);
    dtup->bt_empty_range = false;
 
    return modified;
}
 
static bool
check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys)
{
    int         keyno;
 
    /*
     * First check if there are any IS [NOT] NULL scan keys, and if we're
     * violating them.
     */
    for (keyno = 0; keyno < nnullkeys; keyno++)
    {
        ScanKey     key = nullkeys[keyno];
 
        Assert(key->sk_attno == bval->bv_attno);
 
        /* Handle only IS NULL/IS NOT NULL tests */
        if (!(key->sk_flags & SK_ISNULL))
            continue;
 
        if (key->sk_flags & SK_SEARCHNULL)
        {
            /* IS NULL scan key, but range has no NULLs */
            if (!bval->bv_allnulls && !bval->bv_hasnulls)
                return false;
        }
        else if (key->sk_flags & SK_SEARCHNOTNULL)
        {
            /*
             * For IS NOT NULL, we can only skip ranges that are known to have
             * only nulls.
             */
            if (bval->bv_allnulls)
                return false;
        }
        else
        {
            /*
             * Neither IS NULL nor IS NOT NULL was used; assume all indexable
             * operators are strict and thus return false with NULL value in
             * the scan key.
             */
            return false;
        }
    }
 
    return true;
}
 
/*
 * Create parallel context, and launch workers for leader.
 *
 * buildstate argument should be initialized (with the exception of the
 * tuplesort states, which may later be created based on shared
 * state initially set up here).
 *
 * isconcurrent indicates if operation is CREATE INDEX CONCURRENTLY.
 *
 * request is the target number of parallel worker processes to launch.
 *
 * Sets buildstate's BrinLeader, which caller must use to shut down parallel
 * mode by passing it to _brin_end_parallel() at the very end of its index
 * build.  If not even a single worker process can be launched, this is
 * never set, and caller should proceed with a serial index build.
 */
static void
_brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index,
                     bool isconcurrent, int request)
{
    ParallelContext *pcxt;
    int         scantuplesortstates;
    Snapshot    snapshot;
    Size        estbrinshared;
    Size        estsort;
    BrinShared *brinshared;
    Sharedsort *sharedsort;
    BrinLeader *brinleader = palloc0_object(BrinLeader);
    WalUsage   *walusage;
    BufferUsage *bufferusage;
    bool        leaderparticipates = true;
    int         querylen;
 
#ifdef DISABLE_LEADER_PARTICIPATION
    leaderparticipates = false;
#endif
 
    /*
     * Enter parallel mode, and create context for parallel build of brin
     * index
     */
    EnterParallelMode();
    Assert(request > 0);
    pcxt = CreateParallelContext("postgres", "_brin_parallel_build_main",
                                 request);
 
    scantuplesortstates = leaderparticipates ? request + 1 : request;
 
    /*
     * Prepare for scan of the base relation.  In a normal index build, we use
     * SnapshotAny because we must retrieve all tuples and do our own time
     * qual checks (because we have to index RECENTLY_DEAD tuples).  In a
     * concurrent build, we take a regular MVCC snapshot and index whatever's
     * live according to that.
     */
    if (!isconcurrent)
        snapshot = SnapshotAny;
    else
        snapshot = RegisterSnapshot(GetTransactionSnapshot());
 
    /*
     * Estimate size for our own PARALLEL_KEY_BRIN_SHARED workspace.
     */
    estbrinshared = _brin_parallel_estimate_shared(heap, snapshot);
    shm_toc_estimate_chunk(&pcxt->estimator, estbrinshared);
    estsort = tuplesort_estimate_shared(scantuplesortstates);
    shm_toc_estimate_chunk(&pcxt->estimator, estsort);
 
    shm_toc_estimate_keys(&pcxt->estimator, 2);
 
    /*
     * Estimate space for WalUsage and BufferUsage -- PARALLEL_KEY_WAL_USAGE
     * and PARALLEL_KEY_BUFFER_USAGE.
     *
     * If there are no extensions loaded that care, we could skip this.  We
     * have no way of knowing whether anyone's looking at pgWalUsage or
     * pgBufferUsage, so do it unconditionally.
     */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
    if (debug_query_string)
    {
        querylen = strlen(debug_query_string);
        shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
        shm_toc_estimate_keys(&pcxt->estimator, 1);
    }
    else
        querylen = 0;           /* keep compiler quiet */
 
    /* Everyone's had a chance to ask for space, so now create the DSM */
    InitializeParallelDSM(pcxt);
 
    /* If no DSM segment was available, back out (do serial build) */
    if (pcxt->seg == NULL)
    {
        if (IsMVCCSnapshot(snapshot))
            UnregisterSnapshot(snapshot);
        DestroyParallelContext(pcxt);
        ExitParallelMode();
        return;
    }
 
    /* Store shared build state, for which we reserved space */
    brinshared = (BrinShared *) shm_toc_allocate(pcxt->toc, estbrinshared);
    /* Initialize immutable state */
    brinshared->heaprelid = RelationGetRelid(heap);
    brinshared->indexrelid = RelationGetRelid(index);
    brinshared->isconcurrent = isconcurrent;
    brinshared->scantuplesortstates = scantuplesortstates;
    brinshared->pagesPerRange = buildstate->bs_pagesPerRange;
    brinshared->queryid = pgstat_get_my_query_id();
    ConditionVariableInit(&brinshared->workersdonecv);
    SpinLockInit(&brinshared->mutex);
 
    /* Initialize mutable state */
    brinshared->nparticipantsdone = 0;
    brinshared->reltuples = 0.0;
    brinshared->indtuples = 0.0;
 
    table_parallelscan_initialize(heap,
                                  ParallelTableScanFromBrinShared(brinshared),
                                  snapshot);
 
    /*
     * Store shared tuplesort-private state, for which we reserved space.
     * Then, initialize opaque state using tuplesort routine.
     */
    sharedsort = (Sharedsort *) shm_toc_allocate(pcxt->toc, estsort);
    tuplesort_initialize_shared(sharedsort, scantuplesortstates,
                                pcxt->seg);
 
    /*
     * Store shared tuplesort-private state, for which we reserved space.
     * Then, initialize opaque state using tuplesort routine.
     */
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_BRIN_SHARED, brinshared);
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_TUPLESORT, sharedsort);
 
    /* Store query string for workers */
    if (debug_query_string)
    {
        char       *sharedquery;
 
        sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
        memcpy(sharedquery, debug_query_string, querylen + 1);
        shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery);
    }
 
    /*
     * Allocate space for each worker's WalUsage and BufferUsage; no need to
     * initialize.
     */
    walusage = shm_toc_allocate(pcxt->toc,
                                mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_WAL_USAGE, walusage);
    bufferusage = shm_toc_allocate(pcxt->toc,
                                   mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_BUFFER_USAGE, bufferusage);
 
    /* Launch workers, saving status for leader/caller */
    LaunchParallelWorkers(pcxt);
    brinleader->pcxt = pcxt;
    brinleader->nparticipanttuplesorts = pcxt->nworkers_launched;
    if (leaderparticipates)
        brinleader->nparticipanttuplesorts++;
    brinleader->brinshared = brinshared;
    brinleader->sharedsort = sharedsort;
    brinleader->snapshot = snapshot;
    brinleader->walusage = walusage;
    brinleader->bufferusage = bufferusage;
 
    /* If no workers were successfully launched, back out (do serial build) */
    if (pcxt->nworkers_launched == 0)
    {
        _brin_end_parallel(brinleader, NULL);
        return;
    }
 
    /* Save leader state now that it's clear build will be parallel */
    buildstate->bs_leader = brinleader;
 
    /* Join heap scan ourselves */
    if (leaderparticipates)
        _brin_leader_participate_as_worker(buildstate, heap, index);
 
    /*
     * Caller needs to wait for all launched workers when we return.  Make
     * sure that the failure-to-start case will not hang forever.
     */
    WaitForParallelWorkersToAttach(pcxt);
}
 
/*
 * Shut down workers, destroy parallel context, and end parallel mode.
 */
static void
_brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state)
{
    int         i;
 
    /* Shutdown worker processes */
    WaitForParallelWorkersToFinish(brinleader->pcxt);
 
    /*
     * Next, accumulate WAL usage.  (This must wait for the workers to finish,
     * or we might get incomplete data.)
     */
    for (i = 0; i < brinleader->pcxt->nworkers_launched; i++)
        InstrAccumParallelQuery(&brinleader->bufferusage[i], &brinleader->walusage[i]);
 
    /* Free last reference to MVCC snapshot, if one was used */
    if (IsMVCCSnapshot(brinleader->snapshot))
        UnregisterSnapshot(brinleader->snapshot);
    DestroyParallelContext(brinleader->pcxt);
    ExitParallelMode();
}
 
/*
 * Within leader, wait for end of heap scan.
 *
 * When called, parallel heap scan started by _brin_begin_parallel() will
 * already be underway within worker processes (when leader participates
 * as a worker, we should end up here just as workers are finishing).
 *
 * Returns the total number of heap tuples scanned.
 */
static double
_brin_parallel_heapscan(BrinBuildState *state)
{
    BrinShared *brinshared = state->bs_leader->brinshared;
    int         nparticipanttuplesorts;
 
    nparticipanttuplesorts = state->bs_leader->nparticipanttuplesorts;
    for (;;)
    {
        SpinLockAcquire(&brinshared->mutex);
        if (brinshared->nparticipantsdone == nparticipanttuplesorts)
        {
            /* copy the data into leader state */
            state->bs_reltuples = brinshared->reltuples;
            state->bs_numtuples = brinshared->indtuples;
 
            SpinLockRelease(&brinshared->mutex);
            break;
        }
        SpinLockRelease(&brinshared->mutex);
 
        ConditionVariableSleep(&brinshared->workersdonecv,
                               WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN);
    }
 
    ConditionVariableCancelSleep();
 
    return state->bs_reltuples;
}
 
/*
 * Within leader, wait for end of heap scan and merge per-worker results.
 *
 * After waiting for all workers to finish, merge the per-worker results into
 * the complete index. The results from each worker are sorted by block number
 * (start of the page range). While combining the per-worker results we merge
 * summaries for the same page range, and also fill-in empty summaries for
 * ranges without any tuples.
 *
 * Returns the total number of heap tuples scanned.
 */
static double
_brin_parallel_merge(BrinBuildState *state)
{
    BrinTuple  *btup;
    BrinMemTuple *memtuple = NULL;
    Size        tuplen;
    BlockNumber prevblkno = InvalidBlockNumber;
    MemoryContext rangeCxt,
                oldCxt;
    double      reltuples;
 
    /* wait for workers to scan table and produce partial results */
    reltuples = _brin_parallel_heapscan(state);
 
    /* do the actual sort in the leader */
    tuplesort_performsort(state->bs_sortstate);
 
    /*
     * Initialize BrinMemTuple we'll use to union summaries from workers (in
     * case they happened to produce parts of the same page range).
     */
    memtuple = brin_new_memtuple(state->bs_bdesc);
 
    /*
     * Create a memory context we'll reset to combine results for a single
     * page range (received from the workers). We don't expect huge number of
     * overlaps under regular circumstances, because for large tables the
     * chunk size is likely larger than the BRIN page range), but it can
     * happen, and the union functions may do all kinds of stuff. So we better
     * reset the context once in a while.
     */
    rangeCxt = AllocSetContextCreate(CurrentMemoryContext,
                                     "brin union",
                                     ALLOCSET_DEFAULT_SIZES);
    oldCxt = MemoryContextSwitchTo(rangeCxt);
 
    /*
     * Read the BRIN tuples from the shared tuplesort, sorted by block number.
     * That probably gives us an index that is cheaper to scan, thanks to
     * mostly getting data from the same index page as before.
     */
    while ((btup = tuplesort_getbrintuple(state->bs_sortstate, &tuplen, true)) != NULL)
    {
        /* Ranges should be multiples of pages_per_range for the index. */
        Assert(btup->bt_blkno % state->bs_leader->brinshared->pagesPerRange == 0);
 
        /*
         * Do we need to union summaries for the same page range?
         *
         * If this is the first brin tuple we read, then just deform it into
         * the memtuple, and continue with the next one from tuplesort. We
         * however may need to insert empty summaries into the index.
         *
         * If it's the same block as the last we saw, we simply union the brin
         * tuple into it, and we're done - we don't even need to insert empty
         * ranges, because that was done earlier when we saw the first brin
         * tuple (for this range).
         *
         * Finally, if it's not the first brin tuple, and it's not the same
         * page range, we need to do the insert and then deform the tuple into
         * the memtuple. Then we'll insert empty ranges before the new brin
         * tuple, if needed.
         */
        if (prevblkno == InvalidBlockNumber)
        {
            /* First brin tuples, just deform into memtuple. */
            memtuple = brin_deform_tuple(state->bs_bdesc, btup, memtuple);
 
            /* continue to insert empty pages before thisblock */
        }
        else if (memtuple->bt_blkno == btup->bt_blkno)
        {
            /*
             * Not the first brin tuple, but same page range as the previous
             * one, so we can merge it into the memtuple.
             */
            union_tuples(state->bs_bdesc, memtuple, btup);
            continue;
        }
        else
        {
            BrinTuple  *tmp;
            Size        len;
 
            /*
             * We got brin tuple for a different page range, so form a brin
             * tuple from the memtuple, insert it, and re-init the memtuple
             * from the new brin tuple.
             */
            tmp = brin_form_tuple(state->bs_bdesc, memtuple->bt_blkno,
                                  memtuple, &len);
 
            brin_doinsert(state->bs_irel, state->bs_pagesPerRange, state->bs_rmAccess,
                          &state->bs_currentInsertBuf, tmp->bt_blkno, tmp, len);
 
            /*
             * Reset the per-output-range context. This frees all the memory
             * possibly allocated by the union functions, and also the BRIN
             * tuple we just formed and inserted.
             */
            MemoryContextReset(rangeCxt);
 
            memtuple = brin_deform_tuple(state->bs_bdesc, btup, memtuple);
 
            /* continue to insert empty pages before thisblock */
        }
 
        /* Fill empty ranges for all ranges missing in the tuplesort. */
        brin_fill_empty_ranges(state, prevblkno, btup->bt_blkno);
 
        prevblkno = btup->bt_blkno;
    }
 
    tuplesort_end(state->bs_sortstate);
 
    /* Fill the BRIN tuple for the last page range with data. */
    if (prevblkno != InvalidBlockNumber)
    {
        BrinTuple  *tmp;
        Size        len;
 
        tmp = brin_form_tuple(state->bs_bdesc, memtuple->bt_blkno,
                              memtuple, &len);
 
        brin_doinsert(state->bs_irel, state->bs_pagesPerRange, state->bs_rmAccess,
                      &state->bs_currentInsertBuf, tmp->bt_blkno, tmp, len);
 
        pfree(tmp);
    }
 
    /* Fill empty ranges at the end, for all ranges missing in the tuplesort. */
    brin_fill_empty_ranges(state, prevblkno, state->bs_maxRangeStart);
 
    /*
     * Switch back to the original memory context, and destroy the one we
     * created to isolate the union_tuple calls.
     */
    MemoryContextSwitchTo(oldCxt);
    MemoryContextDelete(rangeCxt);
 
    return reltuples;
}
 
/*
 * Returns size of shared memory required to store state for a parallel
 * brin index build based on the snapshot its parallel scan will use.
 */
static Size
_brin_parallel_estimate_shared(Relation heap, Snapshot snapshot)
{
    /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */
    return add_size(BUFFERALIGN(sizeof(BrinShared)),
                    table_parallelscan_estimate(heap, snapshot));
}
 
/*
 * Within leader, participate as a parallel worker.
 */
static void
_brin_leader_participate_as_worker(BrinBuildState *buildstate, Relation heap, Relation index)
{
    BrinLeader *brinleader = buildstate->bs_leader;
    int         sortmem;
 
    /*
     * Might as well use reliable figure when doling out maintenance_work_mem
     * (when requested number of workers were not launched, this will be
     * somewhat higher than it is for other workers).
     */
    sortmem = maintenance_work_mem / brinleader->nparticipanttuplesorts;
 
    /* Perform work common to all participants */
    _brin_parallel_scan_and_build(buildstate, brinleader->brinshared,
                                  brinleader->sharedsort, heap, index, sortmem, true);
}
 
/*
 * Perform a worker's portion of a parallel sort.
 *
 * This generates a tuplesort for the worker portion of the table.
 *
 * sortmem is the amount of working memory to use within each worker,
 * expressed in KBs.
 *
 * When this returns, workers are done, and need only release resources.
 */
static void
_brin_parallel_scan_and_build(BrinBuildState *state,
                              BrinShared *brinshared, Sharedsort *sharedsort,
                              Relation heap, Relation index,
                              int sortmem, bool progress)
{
    SortCoordinate coordinate;
    TableScanDesc scan;
    double      reltuples;
    IndexInfo  *indexInfo;
 
    /* Initialize local tuplesort coordination state */
    coordinate = palloc0_object(SortCoordinateData);
    coordinate->isWorker = true;
    coordinate->nParticipants = -1;
    coordinate->sharedsort = sharedsort;
 
    /* Begin "partial" tuplesort */
    state->bs_sortstate = tuplesort_begin_index_brin(sortmem, coordinate,
                                                     TUPLESORT_NONE);
 
    /* Join parallel scan */
    indexInfo = BuildIndexInfo(index);
    indexInfo->ii_Concurrent = brinshared->isconcurrent;
 
    scan = table_beginscan_parallel(heap,
                                    ParallelTableScanFromBrinShared(brinshared));
 
    reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
                                       brinbuildCallbackParallel, state, scan);
 
    /* insert the last item */
    form_and_spill_tuple(state);
 
    /* sort the BRIN ranges built by this worker */
    tuplesort_performsort(state->bs_sortstate);
 
    state->bs_reltuples += reltuples;
 
    /*
     * Done.  Record ambuild statistics.
     */
    SpinLockAcquire(&brinshared->mutex);
    brinshared->nparticipantsdone++;
    brinshared->reltuples += state->bs_reltuples;
    brinshared->indtuples += state->bs_numtuples;
    SpinLockRelease(&brinshared->mutex);
 
    /* Notify leader */
    ConditionVariableSignal(&brinshared->workersdonecv);
 
    tuplesort_end(state->bs_sortstate);
}
 
/*
 * Perform work within a launched parallel process.
 */
void
_brin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
{
    char       *sharedquery;
    BrinShared *brinshared;
    Sharedsort *sharedsort;
    BrinBuildState *buildstate;
    Relation    heapRel;
    Relation    indexRel;
    LOCKMODE    heapLockmode;
    LOCKMODE    indexLockmode;
    WalUsage   *walusage;
    BufferUsage *bufferusage;
    int         sortmem;
 
    /*
     * The only possible status flag that can be set to the parallel worker is
     * PROC_IN_SAFE_IC.
     */
    Assert((MyProc->statusFlags == 0) ||
           (MyProc->statusFlags == PROC_IN_SAFE_IC));
 
    /* Set debug_query_string for individual workers first */
    sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true);
    debug_query_string = sharedquery;
 
    /* Report the query string from leader */
    pgstat_report_activity(STATE_RUNNING, debug_query_string);
 
    /* Look up brin shared state */
    brinshared = shm_toc_lookup(toc, PARALLEL_KEY_BRIN_SHARED, false);
 
    /* Open relations using lock modes known to be obtained by index.c */
    if (!brinshared->isconcurrent)
    {
        heapLockmode = ShareLock;
        indexLockmode = AccessExclusiveLock;
    }
    else
    {
        heapLockmode = ShareUpdateExclusiveLock;
        indexLockmode = RowExclusiveLock;
    }
 
    /* Track query ID */
    pgstat_report_query_id(brinshared->queryid, false);
 
    /* Open relations within worker */
    heapRel = table_open(brinshared->heaprelid, heapLockmode);
    indexRel = index_open(brinshared->indexrelid, indexLockmode);
 
    buildstate = initialize_brin_buildstate(indexRel, NULL,
                                            brinshared->pagesPerRange,
                                            InvalidBlockNumber);
 
    /* Look up shared state private to tuplesort.c */
    sharedsort = shm_toc_lookup(toc, PARALLEL_KEY_TUPLESORT, false);
    tuplesort_attach_shared(sharedsort, seg);
 
    /* Prepare to track buffer usage during parallel execution */
    InstrStartParallelQuery();
 
    /*
     * Might as well use reliable figure when doling out maintenance_work_mem
     * (when requested number of workers were not launched, this will be
     * somewhat higher than it is for other workers).
     */
    sortmem = maintenance_work_mem / brinshared->scantuplesortstates;
 
    _brin_parallel_scan_and_build(buildstate, brinshared, sharedsort,
                                  heapRel, indexRel, sortmem, false);
 
    /* Report WAL/buffer usage during parallel execution */
    bufferusage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
    walusage = shm_toc_lookup(toc, PARALLEL_KEY_WAL_USAGE, false);
    InstrEndParallelQuery(&bufferusage[ParallelWorkerNumber],
                          &walusage[ParallelWorkerNumber]);
 
    index_close(indexRel, indexLockmode);
    table_close(heapRel, heapLockmode);
}
 
/*
 * brin_build_empty_tuple
 *      Maybe initialize a BRIN tuple representing empty range.
 *
 * Returns a BRIN tuple representing an empty page range starting at the
 * specified block number. The empty tuple is initialized only once, when it's
 * needed for the first time, stored in the memory context bs_context to ensure
 * proper life span, and reused on following calls. All empty tuples are
 * exactly the same except for the bt_blkno field, which is set to the value
 * in blkno parameter.
 */
static void
brin_build_empty_tuple(BrinBuildState *state, BlockNumber blkno)
{
    /* First time an empty tuple is requested? If yes, initialize it. */
    if (state->bs_emptyTuple == NULL)
    {
        MemoryContext oldcxt;
        BrinMemTuple *dtuple = brin_new_memtuple(state->bs_bdesc);
 
        /* Allocate the tuple in context for the whole index build. */
        oldcxt = MemoryContextSwitchTo(state->bs_context);
 
        state->bs_emptyTuple = brin_form_tuple(state->bs_bdesc, blkno, dtuple,
                                               &state->bs_emptyTupleLen);
 
        MemoryContextSwitchTo(oldcxt);
    }
    else
    {
        /* If we already have an empty tuple, just update the block. */
        state->bs_emptyTuple->bt_blkno = blkno;
    }
}
 
/*
 * brin_fill_empty_ranges
 *      Add BRIN index tuples representing empty page ranges.
 *
 * prevRange/nextRange determine for which page ranges to add empty summaries.
 * Both boundaries are exclusive, i.e. only ranges starting at blkno for which
 * (prevRange < blkno < nextRange) will be added to the index.
 *
 * If prevRange is InvalidBlockNumber, this means there was no previous page
 * range (i.e. the first empty range to add is for blkno=0).
 *
 * The empty tuple is built only once, and then reused for all future calls.
 */
static void
brin_fill_empty_ranges(BrinBuildState *state,
                       BlockNumber prevRange, BlockNumber nextRange)
{
    BlockNumber blkno;
 
    /*
     * If we already summarized some ranges, we need to start with the next
     * one. Otherwise start from the first range of the table.
     */
    blkno = (prevRange == InvalidBlockNumber) ? 0 : (prevRange + state->bs_pagesPerRange);
 
    /* Generate empty ranges until we hit the next non-empty range. */
    while (blkno < nextRange)
    {
        /* Did we already build the empty tuple? If not, do it now. */
        brin_build_empty_tuple(state, blkno);
 
        brin_doinsert(state->bs_irel, state->bs_pagesPerRange, state->bs_rmAccess,
                      &state->bs_currentInsertBuf,
                      blkno, state->bs_emptyTuple, state->bs_emptyTupleLen);
 
        /* try next page range */
        blkno += state->bs_pagesPerRange;
    }
}