database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-28 23:42:10 +03:00
Code Activity

Report progress of CREATE INDEX operations

Browse Source 
This uses the progress reporting infrastructure added by c16dc1aca5,
adding support for CREATE INDEX and CREATE INDEX CONCURRENTLY.

There are two pieces to this: one is index-AM-agnostic, and the other is
AM-specific.  The latter is fairly elaborate for btrees, including
reportage for parallel index builds and the separate phases that btree
index creation uses; other index AMs, which are much simpler in their
building procedures, have simplistic reporting only, but that seems
sufficient, at least for non-concurrent builds.

The index-AM-agnostic part is fairly complete, providing insight into
the CONCURRENTLY wait phases as well as block-based progress during the
index validation table scan.  (The index validation index scan requires
patching each AM, which has not been included here.)

Reviewers: Rahila Syed, Pavan Deolasee, Tatsuro Yamada
Discussion: https://postgr.es/m/20181220220022.mg63bhk26zdpvmcj@alvherre.pgsql
This commit is contained in:
Alvaro Herrera
2019-04-02 15:18:08 -03:00
parent 4d0e994eed
commit ab0dfc961b
37 changed files with 768 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/backend/access/brin/brin.c
View File

@ -112,6 +112,7 @@ brinhandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = brincostestimate;
amroutine->amoptions = brinoptions;
amroutine->amproperty = NULL;
amroutine->ambuildphasename = NULL;
amroutine->amvalidate = brinvalidate;
amroutine->ambeginscan = brinbeginscan;
amroutine->amrescan = brinrescan;
@ -719,7 +720,7 @@ brinbuild(Relation heap, Relation index, IndexInfo *indexInfo)
* Now scan the relation. No syncscan allowed here because we want the
* heap blocks in physical order.
*/
reltuples = table_index_build_scan(heap, index, indexInfo, false,
reltuples = table_index_build_scan(heap, index, indexInfo, false, true,
brinbuildCallback, (void *) state, NULL);
/* process the final batch */
@ -1236,7 +1237,7 @@ summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
* cases.
*/
state->bs_currRangeStart = heapBlk;
table_index_build_range_scan(heapRel, state->bs_irel, indexInfo, false, true,
table_index_build_range_scan(heapRel, state->bs_irel, indexInfo, false, true, false,
heapBlk, scanNumBlks,
brinbuildCallback, (void *) state, NULL);

2
src/backend/access/gin/gininsert.c
View File

@ -395,7 +395,7 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo)
* Do the heap scan. We disallow sync scan here because dataPlaceToPage
* prefers to receive tuples in TID order.
*/
reltuples = table_index_build_scan(heap, index, indexInfo, false,
reltuples = table_index_build_scan(heap, index, indexInfo, false, true,
ginBuildCallback, (void *) &buildstate,
NULL);

1
src/backend/access/gin/ginutil.c
View File

@ -64,6 +64,7 @@ ginhandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = gincostestimate;
amroutine->amoptions = ginoptions;
amroutine->amproperty = NULL;
amroutine->ambuildphasename = NULL;
amroutine->amvalidate = ginvalidate;
amroutine->ambeginscan = ginbeginscan;
amroutine->amrescan = ginrescan;

1
src/backend/access/gist/gist.c
View File

@ -86,6 +86,7 @@ gisthandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = gistcostestimate;
amroutine->amoptions = gistoptions;
amroutine->amproperty = gistproperty;
amroutine->ambuildphasename = NULL;
amroutine->amvalidate = gistvalidate;
amroutine->ambeginscan = gistbeginscan;
amroutine->amrescan = gistrescan;

2
src/backend/access/gist/gistbuild.c
View File

@ -205,7 +205,7 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
/*
* Do the heap scan.
*/
reltuples = table_index_build_scan(heap, index, indexInfo, true,
reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
gistBuildCallback,
(void *) &buildstate, NULL);

7
src/backend/access/hash/hash.c
View File

@ -23,9 +23,11 @@
#include "access/relscan.h"
#include "access/tableam.h"
#include "catalog/index.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "optimizer/plancat.h"
#include "pgstat.h"
#include "utils/builtins.h"
#include "utils/index_selfuncs.h"
#include "utils/rel.h"
@ -83,6 +85,7 @@ hashhandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = hashcostestimate;
amroutine->amoptions = hashoptions;
amroutine->amproperty = NULL;
amroutine->ambuildphasename = NULL;
amroutine->amvalidate = hashvalidate;
amroutine->ambeginscan = hashbeginscan;
amroutine->amrescan = hashrescan;
@ -160,9 +163,11 @@ hashbuild(Relation heap, Relation index, IndexInfo *indexInfo)
buildstate.heapRel = heap;
/* do the heap scan */
reltuples = table_index_build_scan(heap, index, indexInfo, true,
reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
hashbuildCallback,
(void *) &buildstate, NULL);
pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_TOTAL,
buildstate.indtuples);
if (buildstate.spool)
{

6
src/backend/access/hash/hashsort.c
View File

@ -26,7 +26,9 @@
#include "postgres.h"
#include "access/hash.h"
#include "commands/progress.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "utils/tuplesort.h"
@ -116,6 +118,7 @@ void
_h_indexbuild(HSpool *hspool, Relation heapRel)
{
IndexTuple itup;
long tups_done = 0;
#ifdef USE_ASSERT_CHECKING
uint32 hashkey = 0;
#endif
@ -141,5 +144,8 @@ _h_indexbuild(HSpool *hspool, Relation heapRel)
#endif
_hash_doinsert(hspool->index, itup, heapRel);
pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_DONE,
++tups_done);
}
}

107
src/backend/access/heap/heapam_handler.c
View File

@ -57,6 +57,8 @@ static bool SampleHeapTupleVisible(TableScanDesc scan, Buffer buffer,
HeapTuple tuple,
OffsetNumber tupoffset);
static BlockNumber heapam_scan_get_blocks_done(HeapScanDesc hscan);
static const TableAmRoutine heapam_methods;
@ -1120,6 +1122,7 @@ heapam_index_build_range_scan(Relation heapRelation,
IndexInfo *indexInfo,
bool allow_sync,
bool anyvisible,
bool progress,
BlockNumber start_blockno,
BlockNumber numblocks,
IndexBuildCallback callback,
@ -1140,6 +1143,7 @@ heapam_index_build_range_scan(Relation heapRelation,
Snapshot snapshot;
bool need_unregister_snapshot = false;
TransactionId OldestXmin;
BlockNumber previous_blkno = InvalidBlockNumber;
BlockNumber root_blkno = InvalidBlockNumber;
OffsetNumber root_offsets[MaxHeapTuplesPerPage];
@ -1227,6 +1231,25 @@ heapam_index_build_range_scan(Relation heapRelation,
hscan = (HeapScanDesc) scan;
/* Publish number of blocks to scan */
if (progress)
{
BlockNumber nblocks;
if (hscan->rs_base.rs_parallel != NULL)
{
ParallelBlockTableScanDesc pbscan;
pbscan = (ParallelBlockTableScanDesc) hscan->rs_base.rs_parallel;
nblocks = pbscan->phs_nblocks;
}
else
nblocks = hscan->rs_nblocks;
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_TOTAL,
nblocks);
}
/*
* Must call GetOldestXmin() with SnapshotAny. Should never call
* GetOldestXmin() with MVCC snapshot. (It's especially worth checking
@ -1259,6 +1282,19 @@ heapam_index_build_range_scan(Relation heapRelation,
CHECK_FOR_INTERRUPTS();
/* Report scan progress, if asked to. */
if (progress)
{
BlockNumber blocks_done = heapam_scan_get_blocks_done(hscan);
if (blocks_done != previous_blkno)
{
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_DONE,
blocks_done);
previous_blkno = blocks_done;
}
}
/*
* When dealing with a HOT-chain of updated tuples, we want to index
* the values of the live tuple (if any), but index it under the TID
@ -1600,6 +1636,25 @@ heapam_index_build_range_scan(Relation heapRelation,
}
}
/* Report scan progress one last time. */
if (progress)
{
BlockNumber blks_done;
if (hscan->rs_base.rs_parallel != NULL)
{
ParallelBlockTableScanDesc pbscan;
pbscan = (ParallelBlockTableScanDesc) hscan->rs_base.rs_parallel;
blks_done = pbscan->phs_nblocks;
}
else
blks_done = hscan->rs_nblocks;
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_DONE,
blks_done);
}
table_endscan(scan);
/* we can now forget our snapshot, if set and registered by us */
@ -1636,6 +1691,7 @@ heapam_index_validate_scan(Relation heapRelation,
BlockNumber root_blkno = InvalidBlockNumber;
OffsetNumber root_offsets[MaxHeapTuplesPerPage];
bool in_index[MaxHeapTuplesPerPage];
BlockNumber previous_blkno = InvalidBlockNumber;
/* state variables for the merge */
ItemPointer indexcursor = NULL;
@ -1676,6 +1732,9 @@ heapam_index_validate_scan(Relation heapRelation,
false); /* syncscan not OK */
hscan = (HeapScanDesc) scan;
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_TOTAL,
hscan->rs_nblocks);
/*
* Scan all tuples matching the snapshot.
*/
@ -1689,6 +1748,14 @@ heapam_index_validate_scan(Relation heapRelation,
state->htups += 1;
if ((previous_blkno == InvalidBlockNumber) ||
(hscan->rs_cblock != previous_blkno))
{
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_DONE,
hscan->rs_cblock);
previous_blkno = hscan->rs_cblock;
}
/*
* As commented in table_index_build_scan, we should index heap-only
* tuples under the TIDs of their root tuples; so when we advance onto
@ -1849,6 +1916,46 @@ heapam_index_validate_scan(Relation heapRelation,
indexInfo->ii_PredicateState = NULL;
}
/*
* Return the number of blocks that have been read by this scan since
* starting. This is meant for progress reporting rather than be fully
* accurate: in a parallel scan, workers can be concurrently reading blocks
* further ahead than what we report.
*/
static BlockNumber
heapam_scan_get_blocks_done(HeapScanDesc hscan)
{
ParallelBlockTableScanDesc bpscan = NULL;
BlockNumber startblock;
BlockNumber blocks_done;
if (hscan->rs_base.rs_parallel != NULL)
{
bpscan = (ParallelBlockTableScanDesc) hscan->rs_base.rs_parallel;
startblock = bpscan->phs_startblock;
}
else
startblock = hscan->rs_startblock;
/*
* Might have wrapped around the end of the relation, if startblock was
* not zero.
*/
if (hscan->rs_cblock > startblock)
blocks_done = hscan->rs_cblock - startblock;
else
{
BlockNumber nblocks;
nblocks = bpscan != NULL ? bpscan->phs_nblocks : hscan->rs_nblocks;
blocks_done = nblocks - startblock +
hscan->rs_cblock;
}
return blocks_done;
}
/* ------------------------------------------------------------------------
* Planner related callbacks for the heap AM

9
src/backend/access/nbtree/nbtree.c
View File

@ -22,6 +22,7 @@
#include "access/nbtxlog.h"
#include "access/relscan.h"
#include "access/xlog.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
@ -133,6 +134,7 @@ bthandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = btcostestimate;
amroutine->amoptions = btoptions;
amroutine->amproperty = btproperty;
amroutine->ambuildphasename = btbuildphasename;
amroutine->amvalidate = btvalidate;
amroutine->ambeginscan = btbeginscan;
amroutine->amrescan = btrescan;
@ -1021,6 +1023,10 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
if (needLock)
UnlockRelationForExtension(rel, ExclusiveLock);
if (info->report_progress)
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_TOTAL,
num_pages);
/* Quit if we've scanned the whole relation */
if (blkno >= num_pages)
break;
@ -1028,6 +1034,9 @@ btvacuumscan(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
for (; blkno < num_pages; blkno++)
{
btvacuumpage(&vstate, blkno, blkno);
if (info->report_progress)
pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_DONE,
blkno);
}
}

56
src/backend/access/nbtree/nbtsort.c
View File

@ -66,6 +66,7 @@
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "catalog/index.h"
#include "commands/progress.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/smgr.h"
@ -298,7 +299,8 @@ static double _bt_parallel_heapscan(BTBuildState *buildstate,
static void _bt_leader_participate_as_worker(BTBuildState *buildstate);
static void _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2,
BTShared *btshared, Sharedsort *sharedsort,
Sharedsort *sharedsort2, int sortmem);
Sharedsort *sharedsort2, int sortmem,
bool progress);
/*
@ -394,6 +396,10 @@ _bt_spools_heapscan(Relation heap, Relation index, BTBuildState *buildstate,
/* Save as primary spool */
buildstate->spool = btspool;
/* Report table scan phase started */
pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE,
PROGRESS_BTREE_PHASE_INDEXBUILD_TABLESCAN);
/* Attempt to launch parallel worker scan when required */
if (indexInfo->ii_ParallelWorkers > 0)
_bt_begin_parallel(buildstate, indexInfo->ii_Concurrent,
@ -480,13 +486,31 @@ _bt_spools_heapscan(Relation heap, Relation index, BTBuildState *buildstate,
/* Fill spool using either serial or parallel heap scan */
if (!buildstate->btleader)
reltuples = table_index_build_scan(heap, index, indexInfo, true,
reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
_bt_build_callback, (void *) buildstate,
NULL);
else
reltuples = _bt_parallel_heapscan(buildstate,
&indexInfo->ii_BrokenHotChain);
/*
* Set the progress target for the next phase. Reset the block number
* values set by table_index_build_scan
*/
{
const int index[] = {
PROGRESS_CREATEIDX_TUPLES_TOTAL,
PROGRESS_SCAN_BLOCKS_TOTAL,
PROGRESS_SCAN_BLOCKS_DONE
};
const int64 val[] = {
buildstate->indtuples,
0, 0
};
pgstat_progress_update_multi_param(3, index, val);
}
/* okay, all heap tuples are spooled */
if (buildstate->spool2 && !buildstate->havedead)
{
@ -535,9 +559,15 @@ _bt_leafbuild(BTSpool *btspool, BTSpool *btspool2)
}
#endif /* BTREE_BUILD_STATS */
pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE,
PROGRESS_BTREE_PHASE_PERFORMSORT_1);
tuplesort_performsort(btspool->sortstate);
if (btspool2)
{
pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE,
PROGRESS_BTREE_PHASE_PERFORMSORT_2);
tuplesort_performsort(btspool2->sortstate);
}
wstate.heap = btspool->heap;
wstate.index = btspool->index;
@ -554,6 +584,8 @@ _bt_leafbuild(BTSpool *btspool, BTSpool *btspool2)
wstate.btws_pages_written = 0;
wstate.btws_zeropage = NULL; /* until needed */
pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE,
PROGRESS_BTREE_PHASE_LEAF_LOAD);
_bt_load(&wstate, btspool, btspool2);
}
@ -1098,6 +1130,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
int i,
keysz = IndexRelationGetNumberOfKeyAttributes(wstate->index);
SortSupport sortKeys;
long tuples_done = 0;
if (merge)
{
@ -1202,6 +1235,10 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
_bt_buildadd(wstate, state, itup2);
itup2 = tuplesort_getindextuple(btspool2->sortstate, true);
}
/* Report progress */
pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_DONE,
++tuples_done);
}
pfree(sortKeys);
}
@ -1216,6 +1253,10 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
state = _bt_pagestate(wstate, 0);
_bt_buildadd(wstate, state, itup);
/* Report progress */
pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_DONE,
++tuples_done);
}
}
@ -1528,7 +1569,7 @@ _bt_leader_participate_as_worker(BTBuildState *buildstate)
/* Perform work common to all participants */
_bt_parallel_scan_and_sort(leaderworker, leaderworker2, btleader->btshared,
btleader->sharedsort, btleader->sharedsort2,
sortmem);
sortmem, true);
#ifdef BTREE_BUILD_STATS
if (log_btree_build_stats)
@ -1619,7 +1660,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
/* Perform sorting of spool, and possibly a spool2 */
sortmem = maintenance_work_mem / btshared->scantuplesortstates;
_bt_parallel_scan_and_sort(btspool, btspool2, btshared, sharedsort,
sharedsort2, sortmem);
sharedsort2, sortmem, false);
#ifdef BTREE_BUILD_STATS
if (log_btree_build_stats)
@ -1648,7 +1689,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
static void
_bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2,
BTShared *btshared, Sharedsort *sharedsort,
Sharedsort *sharedsort2, int sortmem)
Sharedsort *sharedsort2, int sortmem, bool progress)
{
SortCoordinate coordinate;
BTBuildState buildstate;
@ -1705,9 +1746,10 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2,
/* Join parallel scan */
indexInfo = BuildIndexInfo(btspool->index);
indexInfo->ii_Concurrent = btshared->isconcurrent;
scan = table_beginscan_parallel(btspool->heap, ParallelTableScanFromBTShared(btshared));
scan = table_beginscan_parallel(btspool->heap,
ParallelTableScanFromBTShared(btshared));
reltuples = table_index_build_scan(btspool->heap, btspool->index, indexInfo,
true, _bt_build_callback,
true, progress, _bt_build_callback,
(void *) &buildstate, scan);
/*

24
src/backend/access/nbtree/nbtutils.c
View File

@ -20,6 +20,7 @@
#include "access/nbtree.h"
#include "access/reloptions.h"
#include "access/relscan.h"
#include "commands/progress.h"
#include "miscadmin.h"
#include "utils/array.h"
#include "utils/datum.h"
@ -2051,6 +2052,29 @@ btproperty(Oid index_oid, int attno,
}
}
/*
* btbuildphasename() -- Return name of index build phase.
*/
char *
btbuildphasename(int64 phasenum)
{
switch (phasenum)
{
case PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE:
return "initializing";
case PROGRESS_BTREE_PHASE_INDEXBUILD_TABLESCAN:
return "scanning table";
case PROGRESS_BTREE_PHASE_PERFORMSORT_1:
return "sorting live tuples";
case PROGRESS_BTREE_PHASE_PERFORMSORT_2:
return "sorting dead tuples";
case PROGRESS_BTREE_PHASE_LEAF_LOAD:
return "loading tuples in tree";
default:
return NULL;
}
}
/*
* _bt_truncate() -- create tuple without unneeded suffix attributes.
*

2
src/backend/access/spgist/spginsert.c
View File

@ -143,7 +143,7 @@ spgbuild(Relation heap, Relation index, IndexInfo *indexInfo)
"SP-GiST build temporary context",
ALLOCSET_DEFAULT_SIZES);
reltuples = table_index_build_scan(heap, index, indexInfo, true,
reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
spgistBuildCallback, (void *) &buildstate,
NULL);

1
src/backend/access/spgist/spgutils.c
View File

@ -67,6 +67,7 @@ spghandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = spgcostestimate;
amroutine->amoptions = spgoptions;
amroutine->amproperty = spgproperty;
amroutine->ambuildphasename = NULL;
amroutine->amvalidate = spgvalidate;
amroutine->ambeginscan = spgbeginscan;
amroutine->amrescan = spgrescan;

57
src/backend/catalog/index.c
View File

@ -51,9 +51,9 @@
#include "catalog/pg_trigger.h"
#include "catalog/pg_type.h"
#include "catalog/storage.h"
#include "commands/tablecmds.h"
#include "commands/event_trigger.h"
#include "commands/progress.h"
#include "commands/tablecmds.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "miscadmin.h"
@ -2047,7 +2047,7 @@ index_drop(Oid indexId, bool concurrent, bool concurrent_lock_mode)
* to acquire an exclusive lock on our table. The lock code will
* detect deadlock and error out properly.
*/
WaitForLockers(heaplocktag, AccessExclusiveLock);
WaitForLockers(heaplocktag, AccessExclusiveLock, true);
/* Finish invalidation of index and mark it as dead */
index_concurrently_set_dead(heapId, indexId);
@ -2063,7 +2063,7 @@ index_drop(Oid indexId, bool concurrent, bool concurrent_lock_mode)
* Wait till every transaction that saw the old index state has
* finished.
*/
WaitForLockers(heaplocktag, AccessExclusiveLock);
WaitForLockers(heaplocktag, AccessExclusiveLock, true);
/*
* Re-open relations to allow us to complete our actions.
@ -2712,6 +2712,25 @@ index_build(Relation heapRelation,
save_sec_context | SECURITY_RESTRICTED_OPERATION);
save_nestlevel = NewGUCNestLevel();
/* Set up initial progress report status */
{
const int index[] = {
PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_SUBPHASE,
PROGRESS_CREATEIDX_TUPLES_DONE,
PROGRESS_CREATEIDX_TUPLES_TOTAL,
PROGRESS_SCAN_BLOCKS_DONE,
PROGRESS_SCAN_BLOCKS_TOTAL
};
const int64 val[] = {
PROGRESS_CREATEIDX_PHASE_BUILD,
PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE,
0, 0, 0, 0
};
pgstat_progress_update_multi_param(6, index, val);
}
/*
* Call the access method's build procedure
*/
@ -3000,6 +3019,21 @@ validate_index(Oid heapId, Oid indexId, Snapshot snapshot)
int save_sec_context;
int save_nestlevel;
{
const int index[] = {
PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_TUPLES_DONE,
PROGRESS_CREATEIDX_TUPLES_TOTAL,
PROGRESS_SCAN_BLOCKS_DONE,
PROGRESS_SCAN_BLOCKS_TOTAL
};
const int64 val[] = {
PROGRESS_CREATEIDX_PHASE_VALIDATE_IDXSCAN,
0, 0, 0, 0
};
pgstat_progress_update_multi_param(5, index, val);
}
/* Open and lock the parent heap relation */
heapRelation = table_open(heapId, ShareUpdateExclusiveLock);
/* And the target index relation */
@ -3030,6 +3064,7 @@ validate_index(Oid heapId, Oid indexId, Snapshot snapshot)
*/
ivinfo.index = indexRelation;
ivinfo.analyze_only = false;
ivinfo.report_progress = true;
ivinfo.estimated_count = true;
ivinfo.message_level = DEBUG2;
ivinfo.num_heap_tuples = heapRelation->rd_rel->reltuples;
@ -3047,15 +3082,31 @@ validate_index(Oid heapId, Oid indexId, Snapshot snapshot)
NULL, false);
state.htups = state.itups = state.tups_inserted = 0;
/* ambulkdelete updates progress metrics */
(void) index_bulk_delete(&ivinfo, NULL,
validate_index_callback, (void *) &state);
/* Execute the sort */
{
const int index[] = {
PROGRESS_CREATEIDX_PHASE,
PROGRESS_SCAN_BLOCKS_DONE,
PROGRESS_SCAN_BLOCKS_TOTAL
};
const int64 val[] = {
PROGRESS_CREATEIDX_PHASE_VALIDATE_SORT,
0, 0
};
pgstat_progress_update_multi_param(3, index, val);
}
tuplesort_performsort(state.tuplesort);
/*
* Now scan the heap and "merge" it with the index
*/
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_VALIDATE_TABLESCAN);
table_index_validate_scan(heapRelation,
indexRelation,
indexInfo,

27
src/backend/catalog/system_views.sql
View File

@ -934,6 +934,33 @@ CREATE VIEW pg_stat_progress_cluster AS
FROM pg_stat_get_progress_info('CLUSTER') AS S
LEFT JOIN pg_database D ON S.datid = D.oid;
CREATE VIEW pg_stat_progress_create_index AS
SELECT
S.pid AS pid, S.datid AS datid, D.datname AS datname,
S.relid AS relid,
CASE S.param10 WHEN 0 THEN 'initializing'
WHEN 1 THEN 'waiting for old snapshots'
WHEN 2 THEN 'building index' ||
COALESCE((': ' || pg_indexam_progress_phasename(S.param9::oid, S.param11)),
'')
WHEN 3 THEN 'waiting for writer snapshots'
WHEN 4 THEN 'index validation: scanning index'
WHEN 5 THEN 'index validation: sorting tuples'
WHEN 6 THEN 'index validation: scanning table'
WHEN 7 THEN 'waiting for reader snapshots'
END as phase,
S.param4 AS lockers_total,
S.param5 AS lockers_done,
S.param6 AS current_locker_pid,
S.param16 AS blocks_total,
S.param17 AS blocks_done,
S.param12 AS tuples_total,
S.param13 AS tuples_done,
S.param14 AS partitions_total,
S.param15 AS partitions_done
FROM pg_stat_get_progress_info('CREATE INDEX') AS S
LEFT JOIN pg_database D ON S.datid = D.oid;
CREATE VIEW pg_user_mappings AS
SELECT
U.oid AS umid,

72
src/backend/commands/indexcmds.c
View File

@ -36,6 +36,7 @@
#include "commands/dbcommands.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/progress.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "mb/pg_wchar.h"
@ -47,10 +48,12 @@
#include "parser/parse_func.h"
#include "parser/parse_oper.h"
#include "partitioning/partdesc.h"
#include "pgstat.h"
#include "rewrite/rewriteManip.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/sinvaladt.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
@ -334,7 +337,7 @@ CheckIndexCompatible(Oid oldId,
* doesn't show up in the output, we know we can forget about it.
*/
static void
WaitForOlderSnapshots(TransactionId limitXmin)
WaitForOlderSnapshots(TransactionId limitXmin, bool progress)
{
int n_old_snapshots;
int i;
@ -343,6 +346,8 @@ WaitForOlderSnapshots(TransactionId limitXmin)
old_snapshots = GetCurrentVirtualXIDs(limitXmin, true, false,
PROC_IS_AUTOVACUUM | PROC_IN_VACUUM,
&n_old_snapshots);
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_TOTAL, n_old_snapshots);
for (i = 0; i < n_old_snapshots; i++)
{
@ -378,7 +383,19 @@ WaitForOlderSnapshots(TransactionId limitXmin)
}
if (VirtualTransactionIdIsValid(old_snapshots[i]))
{
if (progress)
{
PGPROC *holder = BackendIdGetProc(old_snapshots[i].backendId);
pgstat_progress_update_param(PROGRESS_WAITFOR_CURRENT_PID,
holder->pid);
}
VirtualXactLock(old_snapshots[i], true);
}
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_DONE, i + 1);
}
}
@ -452,6 +469,15 @@ DefineIndex(Oid relationId,
Snapshot snapshot;
int i;
/*
* Start progress report. If we're building a partition, this was already
* done.
*/
if (!OidIsValid(parentIndexId))
pgstat_progress_start_command(PROGRESS_COMMAND_CREATE_INDEX,
relationId);
/*
* count key attributes in index
*/
@ -668,6 +694,9 @@ DefineIndex(Oid relationId,
accessMethodId = accessMethodForm->oid;
amRoutine = GetIndexAmRoutine(accessMethodForm->amhandler);
pgstat_progress_update_param(PROGRESS_CREATEIDX_ACCESS_METHOD_OID,
accessMethodId);
if (stmt->unique && !amRoutine->amcanunique)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
@ -948,6 +977,11 @@ DefineIndex(Oid relationId,
if (!OidIsValid(indexRelationId))
{
table_close(rel, NoLock);
/* If this is the top-level index, we're done */
if (!OidIsValid(parentIndexId))
pgstat_progress_end_command();
return address;
}
@ -973,6 +1007,9 @@ DefineIndex(Oid relationId,
TupleDesc parentDesc;
Oid *opfamOids;
pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_TOTAL,
nparts);
memcpy(part_oids, partdesc->oids, sizeof(Oid) * nparts);
parentDesc = CreateTupleDescCopy(RelationGetDescr(rel));
@ -1122,6 +1159,8 @@ DefineIndex(Oid relationId,
skip_build, quiet);
}
pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_DONE,
i + 1);
pfree(attmap);
}
@ -1156,6 +1195,8 @@ DefineIndex(Oid relationId,
* Indexes on partitioned tables are not themselves built, so we're
* done here.
*/
if (!OidIsValid(parentIndexId))
pgstat_progress_end_command();
return address;
}
@ -1163,6 +1204,11 @@ DefineIndex(Oid relationId,
{
/* Close the heap and we're done, in the non-concurrent case */
table_close(rel, NoLock);
/* If this is the top-level index, we're done. */
if (!OidIsValid(parentIndexId))
pgstat_progress_end_command();
return address;
}
@ -1214,7 +1260,9 @@ DefineIndex(Oid relationId,
* exclusive lock on our table. The lock code will detect deadlock and
* error out properly.
*/
WaitForLockers(heaplocktag, ShareLock);
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_WAIT_1);
WaitForLockers(heaplocktag, ShareLock, true);
/*
* At this moment we are sure that there are no transactions with the
@ -1255,7 +1303,9 @@ DefineIndex(Oid relationId,
* We once again wait until no transaction can have the table open with
* the index marked as read-only for updates.
*/
WaitForLockers(heaplocktag, ShareLock);
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_WAIT_2);
WaitForLockers(heaplocktag, ShareLock, true);
/*
* Now take the "reference snapshot" that will be used by validate_index()
@ -1312,7 +1362,9 @@ DefineIndex(Oid relationId,
* before the reference snap was taken, we have to wait out any
* transactions that might have older snapshots.
*/
WaitForOlderSnapshots(limitXmin);
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_WAIT_3);
WaitForOlderSnapshots(limitXmin, true);
/*
* Index can now be marked valid -- update its pg_index entry
@ -1334,6 +1386,8 @@ DefineIndex(Oid relationId,
*/
UnlockRelationIdForSession(&heaprelid, ShareUpdateExclusiveLock);
pgstat_progress_end_command();
return address;
}
@ -2913,7 +2967,7 @@ ReindexRelationConcurrently(Oid relationOid, int options)
* DefineIndex() for more details.
*/
WaitForLockersMultiple(lockTags, ShareLock);
WaitForLockersMultiple(lockTags, ShareLock, false);
CommitTransactionCommand();
forboth(lc, indexIds, lc2, newIndexIds)
@ -2955,7 +3009,7 @@ ReindexRelationConcurrently(Oid relationOid, int options)
* for more details.
*/
WaitForLockersMultiple(lockTags, ShareLock);
WaitForLockersMultiple(lockTags, ShareLock, false);
CommitTransactionCommand();
foreach(lc, newIndexIds)
@ -3003,7 +3057,7 @@ ReindexRelationConcurrently(Oid relationOid, int options)
* before the reference snap was taken, we have to wait out any
* transactions that might have older snapshots.
*/
WaitForOlderSnapshots(limitXmin);
WaitForOlderSnapshots(limitXmin, false);
CommitTransactionCommand();
}
@ -3074,7 +3128,7 @@ ReindexRelationConcurrently(Oid relationOid, int options)
* index_drop() for more details.
*/
WaitForLockersMultiple(lockTags, AccessExclusiveLock);
WaitForLockersMultiple(lockTags, AccessExclusiveLock, false);
foreach(lc, indexIds)
{
@ -3096,7 +3150,7 @@ ReindexRelationConcurrently(Oid relationOid, int options)
* Drop the old indexes.
*/
WaitForLockersMultiple(lockTags, AccessExclusiveLock);
WaitForLockersMultiple(lockTags, AccessExclusiveLock, false);
PushActiveSnapshot(GetTransactionSnapshot());

2
src/backend/storage/ipc/standby.c
View File

@ -401,7 +401,7 @@ ResolveRecoveryConflictWithLock(LOCKTAG locktag)
*/
VirtualTransactionId *backends;
backends = GetLockConflicts(&locktag, AccessExclusiveLock);
backends = GetLockConflicts(&locktag, AccessExclusiveLock, NULL);
ResolveRecoveryConflictWithVirtualXIDs(backends,
PROCSIG_RECOVERY_CONFLICT_LOCK);
}

46
src/backend/storage/lmgr/lmgr.c
View File

@ -19,9 +19,12 @@
#include "access/transam.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "commands/progress.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/lmgr.h"
#include "storage/procarray.h"
#include "storage/sinvaladt.h"
#include "utils/inval.h"
@ -857,10 +860,12 @@ XactLockTableWaitErrorCb(void *arg)
* after we obtained our initial list of lockers, we will not wait for them.
*/
void
WaitForLockersMultiple(List *locktags, LOCKMODE lockmode)
WaitForLockersMultiple(List *locktags, LOCKMODE lockmode, bool progress)
{
List *holders = NIL;
ListCell *lc;
int total = 0;
int done = 0;
/* Done if no locks to wait for */
if (list_length(locktags) == 0)
@ -870,10 +875,17 @@ WaitForLockersMultiple(List *locktags, LOCKMODE lockmode)
foreach(lc, locktags)
{
LOCKTAG *locktag = lfirst(lc);
int count;
holders = lappend(holders, GetLockConflicts(locktag, lockmode));
holders = lappend(holders,
GetLockConflicts(locktag, lockmode,
progress ? &count : NULL));
total += count;
}
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_TOTAL, total);
/*
* Note: GetLockConflicts() never reports our own xid, hence we need not
* check for that. Also, prepared xacts are not reported, which is fine
@ -887,10 +899,36 @@ WaitForLockersMultiple(List *locktags, LOCKMODE lockmode)
while (VirtualTransactionIdIsValid(*lockholders))
{
/*
* If requested, publish who we're going to wait for. This is not
* 100% accurate if they're already gone, but we don't care.
*/
if (progress)
{
PGPROC *holder = BackendIdGetProc(lockholders->backendId);
pgstat_progress_update_param(PROGRESS_WAITFOR_CURRENT_PID,
holder->pid);
}
VirtualXactLock(*lockholders, true);
lockholders++;
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_DONE, ++done);
}
}
if (progress)
{
const int index[] = {
PROGRESS_WAITFOR_TOTAL,
PROGRESS_WAITFOR_DONE,
PROGRESS_WAITFOR_CURRENT_PID
};
const int64 values[] = {
0, 0, 0
};
pgstat_progress_update_multi_param(3, index, values);
}
list_free_deep(holders);
}
@ -901,12 +939,12 @@ WaitForLockersMultiple(List *locktags, LOCKMODE lockmode)
* Same as WaitForLockersMultiple, for a single lock tag.
*/
void
WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode)
WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode, bool progress)
{
List *l;
l = list_make1(&heaplocktag);
WaitForLockersMultiple(l, lockmode);
WaitForLockersMultiple(l, lockmode, progress);
list_free(l);
}

7
src/backend/storage/lmgr/lock.c
View File

@ -2807,6 +2807,7 @@ FastPathGetRelationLockEntry(LOCALLOCK *locallock)
* xacts merely awaiting such a lock are NOT reported.
*
* The result array is palloc'd and is terminated with an invalid VXID.
* *countp, if not null, is updated to the number of items set.
*
* Of course, the result could be out of date by the time it's returned,
* so use of this function has to be thought about carefully.
@ -2817,7 +2818,7 @@ FastPathGetRelationLockEntry(LOCALLOCK *locallock)
* uses of the result.
*/
VirtualTransactionId *
GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode)
GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
{
static VirtualTransactionId *vxids;
LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
@ -2964,6 +2965,8 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode)
LWLockRelease(partitionLock);
vxids[count].backendId = InvalidBackendId;
vxids[count].localTransactionId = InvalidLocalTransactionId;
if (countp)
*countp = count;
return vxids;
}
@ -3019,6 +3022,8 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode)
vxids[count].backendId = InvalidBackendId;
vxids[count].localTransactionId = InvalidLocalTransactionId;
if (countp)
*countp = count;
return vxids;
}

23
src/backend/utils/adt/amutils.c
View File

@ -445,3 +445,26 @@ pg_index_column_has_property(PG_FUNCTION_ARGS)
return indexam_property(fcinfo, propname, InvalidOid, relid, attno);
}
/*
* Return the name of the given phase, as used for progress reporting by the
* given AM.
*/
Datum
pg_indexam_progress_phasename(PG_FUNCTION_ARGS)
{
Oid amoid = PG_GETARG_OID(0);
int32 phasenum = PG_GETARG_INT32(1);
IndexAmRoutine *routine;
char *name;
routine = GetIndexAmRoutineByAmId(amoid, true);
if (routine == NULL || !routine->ambuildphasename)
PG_RETURN_NULL();
name = routine->ambuildphasename(phasenum);
if (!name)
PG_RETURN_NULL();
PG_RETURN_TEXT_P(CStringGetTextDatum(name));
}

2
src/backend/utils/adt/pgstatfuncs.c
View File

@ -470,6 +470,8 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
cmdtype = PROGRESS_COMMAND_VACUUM;
else if (pg_strcasecmp(cmd, "CLUSTER") == 0)
cmdtype = PROGRESS_COMMAND_CLUSTER;
else if (pg_strcasecmp(cmd, "CREATE INDEX") == 0)
cmdtype = PROGRESS_COMMAND_CREATE_INDEX;
else
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),

4
src/include/access/amapi.h
View File

@ -108,6 +108,9 @@ typedef bool (*amproperty_function) (Oid index_oid, int attno,
IndexAMProperty prop, const char *propname,
bool *res, bool *isnull);
/* name of phase as used in progress reporting */
typedef char *(*ambuildphasename_function) (int64 phasenum);
/* validate definition of an opclass for this AM */
typedef bool (*amvalidate_function) (Oid opclassoid);
@ -213,6 +216,7 @@ typedef struct IndexAmRoutine
amcostestimate_function amcostestimate;
amoptions_function amoptions;
amproperty_function amproperty; /* can be NULL */
ambuildphasename_function ambuildphasename; /* can be NULL */
amvalidate_function amvalidate;
ambeginscan_function ambeginscan;
amrescan_function amrescan;

1
src/include/access/genam.h
View File

@ -45,6 +45,7 @@ typedef struct IndexVacuumInfo
{
Relation index; /* the index being vacuumed */
bool analyze_only; /* ANALYZE (without any actual vacuum) */
bool report_progress; /* emit progress.h status reports */
bool estimated_count; /* num_heap_tuples is an estimate */
int message_level; /* ereport level for progress messages */
double num_heap_tuples; /* tuples remaining in heap */

11
src/include/access/nbtree.h
View File

@ -671,6 +671,16 @@ typedef BTScanOpaqueData *BTScanOpaque;
#define SK_BT_DESC (INDOPTION_DESC << SK_BT_INDOPTION_SHIFT)
#define SK_BT_NULLS_FIRST (INDOPTION_NULLS_FIRST << SK_BT_INDOPTION_SHIFT)
/*
* Constant definition for progress reporting. Phase numbers must match
* btbuildphasename.
*/
/* PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE is 1 (see progress.h) */
#define PROGRESS_BTREE_PHASE_INDEXBUILD_TABLESCAN 2
#define PROGRESS_BTREE_PHASE_PERFORMSORT_1 3
#define PROGRESS_BTREE_PHASE_PERFORMSORT_2 4
#define PROGRESS_BTREE_PHASE_LEAF_LOAD 5
/*
* external entry points for btree, in nbtree.c
*/
@ -784,6 +794,7 @@ extern bytea *btoptions(Datum reloptions, bool validate);
extern bool btproperty(Oid index_oid, int attno,
IndexAMProperty prop, const char *propname,
bool *res, bool *isnull);
extern char *btbuildphasename(int64 phasenum);
extern IndexTuple _bt_truncate(Relation rel, IndexTuple lastleft,
IndexTuple firstright, BTScanInsert itup_key);
extern int _bt_keep_natts_fast(Relation rel, IndexTuple lastleft,

7
src/include/access/tableam.h
View File

@ -507,6 +507,7 @@ typedef struct TableAmRoutine
struct IndexInfo *index_nfo,
bool allow_sync,
bool anyvisible,
bool progress,
BlockNumber start_blockno,
BlockNumber end_blockno,
IndexBuildCallback callback,
@ -1369,6 +1370,8 @@ table_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
* so here because the AM might reject some of the tuples for its own reasons,
* such as being unable to store NULLs.
*
* If 'progress', the PROGRESS_SCAN_BLOCKS_TOTAL counter is updated when
* starting the scan, and PROGRESS_SCAN_BLOCKS_DONE is updated as we go along.
*
* A side effect is to set indexInfo->ii_BrokenHotChain to true if we detect
* any potentially broken HOT chains. Currently, we set this if there are any
@ -1382,6 +1385,7 @@ table_index_build_scan(Relation heap_rel,
Relation index_rel,
struct IndexInfo *index_nfo,
bool allow_sync,
bool progress,
IndexBuildCallback callback,
void *callback_state,
TableScanDesc scan)
@ -1391,6 +1395,7 @@ table_index_build_scan(Relation heap_rel,
index_nfo,
allow_sync,
false,
progress,
0,
InvalidBlockNumber,
callback,
@ -1414,6 +1419,7 @@ table_index_build_range_scan(Relation heap_rel,
struct IndexInfo *index_nfo,
bool allow_sync,
bool anyvisible,
bool progress,
BlockNumber start_blockno,
BlockNumber numblocks,
IndexBuildCallback callback,
@ -1425,6 +1431,7 @@ table_index_build_range_scan(Relation heap_rel,
index_nfo,
allow_sync,
anyvisible,
progress,
start_blockno,
numblocks,
callback,

2
src/include/catalog/catversion.h
View File

@ -53,6 +53,6 @@
*/
/* yyyymmddN */
#define CATALOG_VERSION_NO 201904011
#define CATALOG_VERSION_NO 201904021
#endif

10
src/include/catalog/pg_proc.dat
View File

@ -924,6 +924,10 @@
proname => 'pg_index_column_has_property', provolatile => 's',
prorettype => 'bool', proargtypes => 'regclass int4 text',
prosrc => 'pg_index_column_has_property' },
{ oid => '676', descr => 'return name of given index build phase',
proname => 'pg_indexam_progress_phasename', provolatile => 'i',
prorettype => 'text', proargtypes => 'oid int8',
prosrc => 'pg_indexam_progress_phasename' },
{ oid => '339',
proname => 'poly_same', prorettype => 'bool',
@ -5122,9 +5126,9 @@
proname => 'pg_stat_get_progress_info', prorows => '100', proretset => 't',
provolatile => 's', proparallel => 'r', prorettype => 'record',
proargtypes => 'text',
proallargtypes => '{text,int4,oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8}',
proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o}',
proargnames => '{cmdtype,pid,datid,relid,param1,param2,param3,param4,param5,param6,param7,param8,param9,param10}',
proallargtypes => '{text,int4,oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8}',
proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
proargnames => '{cmdtype,pid,datid,relid,param1,param2,param3,param4,param5,param6,param7,param8,param9,param10,param11,param12,param13,param14,param15,param16,param17,param18,param19,param20}',
prosrc => 'pg_stat_get_progress_info' },
{ oid => '3099',
descr => 'statistics: information about currently active replication',

37
src/include/commands/progress.h
View File

@ -44,7 +44,7 @@
#define PROGRESS_CLUSTER_HEAP_BLKS_SCANNED 6
#define PROGRESS_CLUSTER_INDEX_REBUILD_COUNT 7
/* Phases of cluster (as dvertised via PROGRESS_CLUSTER_PHASE) */
/* Phases of cluster (as advertised via PROGRESS_CLUSTER_PHASE) */
#define PROGRESS_CLUSTER_PHASE_SEQ_SCAN_HEAP 1
#define PROGRESS_CLUSTER_PHASE_INDEX_SCAN_HEAP 2
#define PROGRESS_CLUSTER_PHASE_SORT_TUPLES 3
@ -57,4 +57,39 @@
#define PROGRESS_CLUSTER_COMMAND_CLUSTER 1
#define PROGRESS_CLUSTER_COMMAND_VACUUM_FULL 2
/* Progress parameters for CREATE INDEX */
/* 3, 4 and 5 reserved for "waitfor" metrics */
#define PROGRESS_CREATEIDX_ACCESS_METHOD_OID 8
#define PROGRESS_CREATEIDX_PHASE 9 /* AM-agnostic phase # */
#define PROGRESS_CREATEIDX_SUBPHASE 10 /* phase # filled by AM */
#define PROGRESS_CREATEIDX_TUPLES_TOTAL 11
#define PROGRESS_CREATEIDX_TUPLES_DONE 12
#define PROGRESS_CREATEIDX_PARTITIONS_TOTAL 13
#define PROGRESS_CREATEIDX_PARTITIONS_DONE 14
/* 15 and 16 reserved for "block number" metrics */
/* Phases of CREATE INDEX (as advertised via PROGRESS_CREATEIDX_PHASE) */
#define PROGRESS_CREATEIDX_PHASE_WAIT_1 1
#define PROGRESS_CREATEIDX_PHASE_BUILD 2
#define PROGRESS_CREATEIDX_PHASE_WAIT_2 3
#define PROGRESS_CREATEIDX_PHASE_VALIDATE_IDXSCAN 4
#define PROGRESS_CREATEIDX_PHASE_VALIDATE_SORT 5
#define PROGRESS_CREATEIDX_PHASE_VALIDATE_TABLESCAN 6
#define PROGRESS_CREATEIDX_PHASE_WAIT_3 7
/*
* Subphases of CREATE INDEX, for index_build.
*/
#define PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE 1
/* Additional phases are defined by each AM */
/* Lock holder wait counts */
#define PROGRESS_WAITFOR_TOTAL 3
#define PROGRESS_WAITFOR_DONE 4
#define PROGRESS_WAITFOR_CURRENT_PID 5
/* Block numbers in a generic relation scan */
#define PROGRESS_SCAN_BLOCKS_TOTAL 15
#define PROGRESS_SCAN_BLOCKS_DONE 16
#endif

5
src/include/pgstat.h
View File

@ -951,10 +951,11 @@ typedef enum ProgressCommandType
{
PROGRESS_COMMAND_INVALID,
PROGRESS_COMMAND_VACUUM,
PROGRESS_COMMAND_CLUSTER
PROGRESS_COMMAND_CLUSTER,
PROGRESS_COMMAND_CREATE_INDEX
} ProgressCommandType;
#define PGSTAT_NUM_PROGRESS_PARAM 10
#define PGSTAT_NUM_PROGRESS_PARAM 20
/* ----------
* Shared-memory data structures

4
src/include/storage/lmgr.h
View File

@ -78,8 +78,8 @@ extern void XactLockTableWait(TransactionId xid, Relation rel,
extern bool ConditionalXactLockTableWait(TransactionId xid);
/* Lock VXIDs, specified by conflicting locktags */
extern void WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode);
extern void WaitForLockersMultiple(List *locktags, LOCKMODE lockmode);
extern void WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode, bool progress);
extern void WaitForLockersMultiple(List *locktags, LOCKMODE lockmode, bool progress);
/* Lock an XID for tuple insertion (used to wait for an insertion to finish) */
extern uint32 SpeculativeInsertionLockAcquire(TransactionId xid);

2
src/include/storage/lock.h
View File

@ -544,7 +544,7 @@ extern bool LockHeldByMe(const LOCKTAG *locktag, LOCKMODE lockmode);
extern bool LockHasWaiters(const LOCKTAG *locktag,
LOCKMODE lockmode, bool sessionLock);
extern VirtualTransactionId *GetLockConflicts(const LOCKTAG *locktag,
LOCKMODE lockmode);
LOCKMODE lockmode, int *countp);
extern void AtPrepare_Locks(void);
extern void PostPrepare_Locks(TransactionId xid);
extern int LockCheckConflicts(LockMethod lockMethodTable,

30
src/test/regress/expected/rules.out
View File

@ -1856,7 +1856,33 @@ pg_stat_progress_cluster| SELECT s.pid,
s.param6 AS heap_blks_total,
s.param7 AS heap_blks_scanned,
s.param8 AS index_rebuild_count
FROM (pg_stat_get_progress_info('CLUSTER'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10)
FROM (pg_stat_get_progress_info('CLUSTER'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20)
LEFT JOIN pg_database d ON ((s.datid = d.oid)));
pg_stat_progress_create_index| SELECT s.pid,
s.datid,
d.datname,
s.relid,
CASE s.param10
WHEN 0 THEN 'initializing'::text
WHEN 1 THEN 'waiting for old snapshots'::text
WHEN 2 THEN ('building index'::text || COALESCE((': '::text || pg_indexam_progress_phasename((s.param9)::oid, s.param11)), ''::text))
WHEN 3 THEN 'waiting for writer snapshots'::text
WHEN 4 THEN 'index validation: scan index'::text
WHEN 5 THEN 'index validation: sort index scan results'::text
WHEN 6 THEN 'index validation: scan heap'::text
WHEN 7 THEN 'waiting for reader snapshots'::text
ELSE NULL::text
END AS phase,
s.param4 AS lockers_total,
s.param5 AS lockers_done,
s.param6 AS current_locker_pid,
s.param16 AS blocks_total,
s.param17 AS blocks_done,
s.param12 AS tuples_total,
s.param13 AS tuples_done,
s.param14 AS partitions_total,
s.param15 AS partitions_done
FROM (pg_stat_get_progress_info('CREATE INDEX'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20)
LEFT JOIN pg_database d ON ((s.datid = d.oid)));
pg_stat_progress_vacuum| SELECT s.pid,
s.datid,
@ -1878,7 +1904,7 @@ pg_stat_progress_vacuum| SELECT s.pid,
s.param5 AS index_vacuum_count,
s.param6 AS max_dead_tuples,
s.param7 AS num_dead_tuples
FROM (pg_stat_get_progress_info('VACUUM'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10)
FROM (pg_stat_get_progress_info('VACUUM'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20)
LEFT JOIN pg_database d ON ((s.datid = d.oid)));
pg_stat_replication| SELECT s.pid,
s.usesysid,