diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index 8bcd159..6ea8f78 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -16,11 +16,14 @@ #include "access/gin_private.h" #include "access/xloginsert.h" +#include "access/parallel.h" +#include "access/xact.h" #include "catalog/index.h" #include "miscadmin.h" #include "storage/bufmgr.h" #include "storage/smgr.h" #include "storage/indexfsm.h" +#include "storage/spin.h" #include "utils/memutils.h" #include "utils/rel.h" @@ -265,6 +268,10 @@ ginHeapTupleBulkInsert(GinBuildState *buildstate, OffsetNumber attnum, MemoryContextReset(buildstate->funcCtx); } +#define KEY_TASK 42 +#define GIN_MAX_WORKERS 4 +#define GIN_BLOCKS_PER_WORKER 4 + static void ginBuildCallback(Relation index, HeapTuple htup, Datum *values, bool *isnull, bool tupleIsAlive, void *state) @@ -306,17 +313,34 @@ ginBuildCallback(Relation index, HeapTuple htup, Datum *values, MemoryContextSwitchTo(oldCtx); } -Datum -ginbuild(PG_FUNCTION_ARGS) -{ - Relation heap = (Relation) PG_GETARG_POINTER(0); - Relation index = (Relation) PG_GETARG_POINTER(1); - IndexInfo *indexInfo = (IndexInfo *) PG_GETARG_POINTER(2); - IndexBuildResult *result; - double reltuples; +/* + * This structure describes the GIN build task for the parallel workers. We use + * OIDs here because workers are separate processes and pointers may become + * meaningless for them. The "lock" is used to protect the "scanned" field as + * the workers read and increase its value to distribute the task between each + * other. + */ +typedef struct { + int to_scan; + int scanned; + slock_t lock; + Oid heap_oid; + Oid index_oid; +} PGinBuildTask; + +/* + * The idea behind parallel GIN build is letting the parallel workers scan the + * relation and build rbtree in parallel. Each block is scanned by one of the + * workers. + */ +static void pginbuild(dsm_segment *seg, shm_toc *toc) { GinBuildState buildstate; - Buffer RootBuffer, - MetaBuffer; + + Relation heap; + Relation index; + IndexInfo *indexInfo; + + int reltuples = 0; ItemPointerData *list; Datum key; GinNullCategory category; @@ -324,48 +348,16 @@ ginbuild(PG_FUNCTION_ARGS) MemoryContext oldCtx; OffsetNumber attnum; - if (RelationGetNumberOfBlocks(index) != 0) - elog(ERROR, "index \"%s\" already contains data", - RelationGetRelationName(index)); + volatile PGinBuildTask *task = (PGinBuildTask*)shm_toc_lookup(toc, KEY_TASK); + + heap = heap_open(task->heap_oid, NoLock); + index = index_open(task->index_oid, NoLock); + indexInfo = BuildIndexInfo(index); initGinState(&buildstate.ginstate, index); buildstate.indtuples = 0; memset(&buildstate.buildStats, 0, sizeof(GinStatsData)); - /* initialize the meta page */ - MetaBuffer = GinNewBuffer(index); - - /* initialize the root page */ - RootBuffer = GinNewBuffer(index); - - START_CRIT_SECTION(); - GinInitMetabuffer(MetaBuffer); - MarkBufferDirty(MetaBuffer); - GinInitBuffer(RootBuffer, GIN_LEAF); - MarkBufferDirty(RootBuffer); - - if (RelationNeedsWAL(index)) - { - XLogRecPtr recptr; - Page page; - - XLogBeginInsert(); - XLogRegisterBuffer(0, MetaBuffer, REGBUF_WILL_INIT); - XLogRegisterBuffer(1, RootBuffer, REGBUF_WILL_INIT); - - recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_INDEX); - - page = BufferGetPage(RootBuffer); - PageSetLSN(page, recptr); - - page = BufferGetPage(MetaBuffer); - PageSetLSN(page, recptr); - } - - UnlockReleaseBuffer(MetaBuffer); - UnlockReleaseBuffer(RootBuffer); - END_CRIT_SECTION(); - /* count the root as first entry page */ buildstate.buildStats.nEntryPages++; @@ -396,8 +388,27 @@ ginbuild(PG_FUNCTION_ARGS) * Do the heap scan. We disallow sync scan here because dataPlaceToPage * prefers to receive tuples in TID order. */ - reltuples = IndexBuildHeapScan(heap, index, indexInfo, false, - ginBuildCallback, (void *) &buildstate); + while (true) { + int subtuples; + int scan_from, to_scan_this_time; + SpinLockAcquire(&task->lock); + if (task->scanned >= task->to_scan) + { + SpinLockRelease(&task->lock); + break; + } + scan_from = task->scanned; + to_scan_this_time = GIN_BLOCKS_PER_WORKER; + if (to_scan_this_time > task->to_scan - task->scanned) + to_scan_this_time = task->to_scan - task->scanned; + task->scanned += to_scan_this_time; + SpinLockRelease(&task->lock); + + subtuples = IndexBuildHeapRangeScan(heap, index, indexInfo, false, false, + scan_from, to_scan_this_time, + ginBuildCallback, (void *)&buildstate); + reltuples += subtuples; + }; /* dump remaining entries to the index */ oldCtx = MemoryContextSwitchTo(buildstate.tmpCtx); @@ -421,13 +432,94 @@ ginbuild(PG_FUNCTION_ARGS) buildstate.buildStats.nTotalPages = RelationGetNumberOfBlocks(index); ginUpdateStats(index, &buildstate.buildStats); + index_close(index, NoLock); + heap_close(heap, NoLock); +} + +Datum +ginbuild(PG_FUNCTION_ARGS) +{ + Relation heap = (Relation) PG_GETARG_POINTER(0); + Relation index = (Relation) PG_GETARG_POINTER(1); + IndexBuildResult *result; + Buffer RootBuffer, + MetaBuffer; + + if (RelationGetNumberOfBlocks(index) != 0) + elog(ERROR, "index \"%s\" already contains data", + RelationGetRelationName(index)); + + /* initialize the meta page */ + MetaBuffer = GinNewBuffer(index); + + /* initialize the root page */ + RootBuffer = GinNewBuffer(index); + + START_CRIT_SECTION(); + GinInitMetabuffer(MetaBuffer); + MarkBufferDirty(MetaBuffer); + GinInitBuffer(RootBuffer, GIN_LEAF); + MarkBufferDirty(RootBuffer); + + if (RelationNeedsWAL(index)) + { + XLogRecPtr recptr; + Page page; + + XLogBeginInsert(); + XLogRegisterBuffer(0, MetaBuffer, REGBUF_WILL_INIT); + XLogRegisterBuffer(1, RootBuffer, REGBUF_WILL_INIT); + + recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_INDEX); + + page = BufferGetPage(RootBuffer); + PageSetLSN(page, recptr); + + page = BufferGetPage(MetaBuffer); + PageSetLSN(page, recptr); + } + + UnlockReleaseBuffer(MetaBuffer); + UnlockReleaseBuffer(RootBuffer); + END_CRIT_SECTION(); + + EnterParallelMode(); + { + int size = sizeof(PGinBuildTask); + int keys = 1; + PGinBuildTask *task; + ParallelContext *pcxt; + + pcxt = CreateParallelContext(pginbuild, GIN_MAX_WORKERS); + + shm_toc_estimate_chunk(&pcxt->estimator, size); + shm_toc_estimate_keys(&pcxt->estimator, keys); + InitializeParallelDSM(pcxt); + task = (PGinBuildTask*)shm_toc_allocate(pcxt->toc, size); + shm_toc_insert(pcxt->toc, KEY_TASK, (void*)task); + SpinLockInit(&task->lock); + task->to_scan = RelationGetNumberOfBlocks(heap); + task->heap_oid = heap->rd_id; + task->index_oid = index->rd_id; + LaunchParallelWorkers(pcxt); + + WaitForParallelWorkersToFinish(pcxt); + + DestroyParallelContext(pcxt); + } + ExitParallelMode(); + /* * Return statistics */ result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult)); - result->heap_tuples = reltuples; - result->index_tuples = buildstate.indtuples; + /* + * FIXME: implement statistics aggregation from the workers. + * These numbers do not mean anything until then. + */ + result->heap_tuples = 123; + result->index_tuples = 456; PG_RETURN_POINTER(result); }