/
githubmirror
/
postgres
Обзор
Документация
Войти
/
githubmirror
/
postgres
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/backend/access/brin/brin.c
3 034 строки
90 KB
Melanie Plageman
Thread flags through begin-scan APIs
30 мар 2026, 19:27
30 мар 2026, 19:27
dcd8cc1
Код
Авторство
О чём код?
/* * 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 "executor/instrument.h" #include "miscadmin.h" #include "pgstat.h" #include "postmaster/autovacuum.h" #include "storage/bufmgr.h" #include "storage/condition_variable.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" #include "utils/wait_event.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; /* 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), SO_NONE); 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; } }