public inbox for [email protected]
help / color / mirror / Atom feedFrom: Tomas Vondra <[email protected]>
To: Matthias van de Meent <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: Parallel CREATE INDEX for GIN indexes
Date: Sun, 5 May 2024 20:49:03 +0200
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<CAEze2WjzVPXKaU4moyqvAhrz8sFVW7_=Q2EjpcCK_8rSE6f28w@mail.gmail.com>
<[email protected]>
Hi,
Here's a slightly improved version, fixing a couple bugs in handling
byval/byref values, causing issues on 32-bit machines (but not only).
And also a couple compiler warnings about string formatting.
Other than that, no changes.
--
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company
Attachments:
[text/x-patch] v20240505-0001-Allow-parallel-create-for-GIN-indexes.patch (56.4K, ../[email protected]/2-v20240505-0001-Allow-parallel-create-for-GIN-indexes.patch)
download | inline diff:
From 510af00802c04b8d6d3982069c96082572a76c72 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:26 +0200
Subject: [PATCH v20240505 1/8] Allow parallel create for GIN indexes
Add support for parallel create of GIN indexes, using an approach and
code very similar to the one used by BRIN indexes.
Each worker reads a subset of the table (from a parallel scan), and
accumulated index entries in memory. But instead of writing the results
into the index (after hitting the memory limit), the data are written
to a shared tuplesort (and sorted by index key).
The leader then reads data from the tuplesort, and combines them into
entries that get inserted into the index.
---
src/backend/access/gin/ginbulk.c | 7 +
src/backend/access/gin/gininsert.c | 1340 +++++++++++++++++++-
src/backend/access/gin/ginutil.c | 2 +-
src/backend/access/transam/parallel.c | 4 +
src/backend/utils/sort/tuplesortvariants.c | 154 +++
src/include/access/gin.h | 4 +
src/include/access/gin_tuple.h | 29 +
src/include/utils/tuplesort.h | 6 +
src/tools/pgindent/typedefs.list | 4 +
9 files changed, 1534 insertions(+), 16 deletions(-)
create mode 100644 src/include/access/gin_tuple.h
diff --git a/src/backend/access/gin/ginbulk.c b/src/backend/access/gin/ginbulk.c
index a522801c2f7..12eeff04a6c 100644
--- a/src/backend/access/gin/ginbulk.c
+++ b/src/backend/access/gin/ginbulk.c
@@ -153,6 +153,13 @@ ginInsertBAEntry(BuildAccumulator *accum,
GinEntryAccumulator *ea;
bool isNew;
+ /*
+ * FIXME prevents writes of uninitialized bytes reported by valgrind in
+ * writetup (likely that build_gin_tuple copies some fields that are only
+ * initialized for a certain category, or something similar)
+ */
+ memset(&eatmp, 0, sizeof(GinEntryAccumulator));
+
/*
* For the moment, fill only the fields of eatmp that will be looked at by
* cmpEntryAccumulator or ginCombineData.
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index 71f38be90c3..b353e155fc6 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -15,14 +15,124 @@
#include "postgres.h"
#include "access/gin_private.h"
+#include "access/gin_tuple.h"
+#include "access/table.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
+#include "catalog/index.h"
+#include "commands/vacuum.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
+#include "pgstat.h"
#include "storage/bufmgr.h"
#include "storage/predicate.h"
+#include "tcop/tcopprot.h" /* pgrminclude ignore */
+#include "utils/datum.h"
#include "utils/memutils.h"
#include "utils/rel.h"
+#include "utils/builtins.h"
+
+
+/* Magic numbers for parallel state sharing */
+#define PARALLEL_KEY_GIN_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 GinShared
+{
+ /*
+ * 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;
+ int scantuplesortstates;
+
+ /*
+ * 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 GIN 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.
+ */
+} GinShared;
+
+/*
+ * Return pointer to a GinShared's parallel table scan.
+ *
+ * c.f. shm_toc_allocate as to why BUFFERALIGN is used, rather than just
+ * MAXALIGN.
+ */
+#define ParallelTableScanFromGinShared(shared) \
+ (ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(GinShared)))
+
+/*
+ * Status for leader in parallel index build.
+ */
+typedef struct GinLeader
+{
+ /* 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).
+ *
+ * GinShared 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.
+ */
+ GinShared *ginshared;
+ Sharedsort *sharedsort;
+ Snapshot snapshot;
+ WalUsage *walusage;
+ BufferUsage *bufferusage;
+} GinLeader;
typedef struct
{
@@ -32,9 +142,49 @@ typedef struct
MemoryContext tmpCtx;
MemoryContext funcCtx;
BuildAccumulator accum;
+
+ /* FIXME likely duplicate with indtuples */
+ double bs_numtuples;
+ double bs_reltuples;
+
+ /*
+ * 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
+ * GinBuildState.)
+ */
+ GinLeader *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;
} GinBuildState;
+/* parallel index builds */
+static void _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index,
+ bool isconcurrent, int request);
+static void _gin_end_parallel(GinLeader *ginleader, GinBuildState *state);
+static Size _gin_parallel_estimate_shared(Relation heap, Snapshot snapshot);
+static double _gin_parallel_heapscan(GinBuildState *buildstate);
+static double _gin_parallel_merge(GinBuildState *buildstate);
+static void _gin_leader_participate_as_worker(GinBuildState *buildstate,
+ Relation heap, Relation index);
+static void _gin_parallel_scan_and_build(GinBuildState *buildstate,
+ GinShared *ginshared,
+ Sharedsort *sharedsort,
+ Relation heap, Relation index,
+ int sortmem, bool progress);
+
+static Datum _gin_parse_tuple(GinTuple *a, ItemPointerData **items);
+static GinTuple *_gin_build_tuple(OffsetNumber attrnum, unsigned char category,
+ Datum key, int16 typlen, bool typbyval,
+ ItemPointerData *items, uint32 nitems,
+ Size *len);
+
/*
* Adds array of item pointers to tuple's posting list, or
* creates posting tree and tuple pointing to tree in case
@@ -313,12 +463,95 @@ ginBuildCallback(Relation index, ItemPointer tid, Datum *values,
MemoryContextSwitchTo(oldCtx);
}
+static void
+ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values,
+ bool *isnull, bool tupleIsAlive, void *state)
+{
+ GinBuildState *buildstate = (GinBuildState *) state;
+ MemoryContext oldCtx;
+ int i;
+
+ oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
+
+ for (i = 0; i < buildstate->ginstate.origTupdesc->natts; i++)
+ ginHeapTupleBulkInsert(buildstate, (OffsetNumber) (i + 1),
+ values[i], isnull[i], tid);
+
+ /*
+ * XXX idea - Instead of writing the entries directly into the shared
+ * tuplesort, write it into a local one, do the sort in the worker, and
+ * combine the results. For large tables with many different keys that's
+ * going to work better than the current approach where we don't get many
+ * matches in work_mem (maybe this should use 32MB, which is what we use
+ * when planning, but even that may not be great). Which means we are
+ * likely to have many entries with a single TID, forcing the leader to do
+ * a qsort() when merging the data, often amounting to ~50% of the serial
+ * part. By doing the qsort() in a worker, leader then can do a mergesort
+ * (likely cheaper). Also, it means the amount of data worker->leader is
+ * going to be lower thanks to deduplication.
+ *
+ * Disadvantage: It needs more disk space, possibly up to 2x, because each
+ * worker creates a tuplestore, then "transforms it" into the shared
+ * tuplestore (hopefully less data, but not guaranteed).
+ *
+ * It's however possible to partition the data into multiple tuplesorts
+ * per worker (by hashing). We don't need perfect sorting, and we can even
+ * live with "equal" keys having multiple hashes (if there are multiple
+ * binary representations of the value).
+ */
+
+ /*
+ * If we've maxed out our available memory, dump everything to the
+ * tuplesort
+ *
+ * XXX probably should use 32MB, not work_mem, as used during planning?
+ */
+ if (buildstate->accum.allocatedMemory >= (Size) work_mem * 1024L)
+ {
+ ItemPointerData *list;
+ Datum key;
+ GinNullCategory category;
+ uint32 nlist;
+ OffsetNumber attnum;
+ TupleDesc tdesc = RelationGetDescr(index);
+
+ ginBeginBAScan(&buildstate->accum);
+ while ((list = ginGetBAEntry(&buildstate->accum,
+ &attnum, &key, &category, &nlist)) != NULL)
+ {
+ /* information about the key */
+ Form_pg_attribute attr = TupleDescAttr(tdesc, (attnum - 1));
+
+ /* GIN tuple and tuple length */
+ GinTuple *tup;
+ Size tuplen;
+
+ /* there could be many entries, so be willing to abort here */
+ CHECK_FOR_INTERRUPTS();
+
+ tup = _gin_build_tuple(attnum, category,
+ key, attr->attlen, attr->attbyval,
+ list, nlist, &tuplen);
+
+ tuplesort_putgintuple(buildstate->bs_sortstate, tup, tuplen);
+
+ pfree(tup);
+ }
+
+ MemoryContextReset(buildstate->tmpCtx);
+ ginInitBA(&buildstate->accum);
+ }
+
+ MemoryContextSwitchTo(oldCtx);
+}
+
IndexBuildResult *
ginbuild(Relation heap, Relation index, IndexInfo *indexInfo)
{
IndexBuildResult *result;
double reltuples;
GinBuildState buildstate;
+ GinBuildState *state = &buildstate;
Buffer RootBuffer,
MetaBuffer;
ItemPointerData *list;
@@ -336,6 +569,14 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo)
buildstate.indtuples = 0;
memset(&buildstate.buildStats, 0, sizeof(GinStatsData));
+ /*
+ * XXX Make sure to initialize a bunch of fields, not to trip valgrind.
+ * Maybe there should be an "init" function for build state?
+ */
+ buildstate.bs_numtuples = 0;
+ buildstate.bs_reltuples = 0;
+ buildstate.bs_leader = NULL;
+
/* initialize the meta page */
MetaBuffer = GinNewBuffer(index);
@@ -376,25 +617,91 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo)
ginInitBA(&buildstate.accum);
/*
- * Do the heap scan. We disallow sync scan here because dataPlaceToPage
- * prefers to receive tuples in TID order.
+ * 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 GIN, which can do with much less memory. So
+ * maybe make that somehow less strict, optionally?
+ */
+ if (indexInfo->ii_ParallelWorkers > 0)
+ _gin_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.
*/
- reltuples = table_index_build_scan(heap, index, indexInfo, false, true,
- ginBuildCallback, (void *) &buildstate,
- NULL);
+ if (state->bs_leader)
+ {
+ SortCoordinate coordinate;
+
+ coordinate = (SortCoordinate) palloc0(sizeof(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_gin(maintenance_work_mem, coordinate,
+ TUPLESORT_NONE);
- /* dump remaining entries to the index */
- oldCtx = MemoryContextSwitchTo(buildstate.tmpCtx);
- ginBeginBAScan(&buildstate.accum);
- while ((list = ginGetBAEntry(&buildstate.accum,
- &attnum, &key, &category, &nlist)) != NULL)
+ /* scan the relation and merge per-worker results */
+ reltuples = _gin_parallel_merge(state);
+
+ _gin_end_parallel(state->bs_leader, state);
+ }
+ else /* no parallel index build */
{
- /* there could be many entries, so be willing to abort here */
- CHECK_FOR_INTERRUPTS();
- ginEntryInsert(&buildstate.ginstate, attnum, key, category,
- list, nlist, &buildstate.buildStats);
+ /*
+ * 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, true,
+ ginBuildCallback, (void *) &buildstate,
+ NULL);
+
+ /* dump remaining entries to the index */
+ oldCtx = MemoryContextSwitchTo(buildstate.tmpCtx);
+ ginBeginBAScan(&buildstate.accum);
+ while ((list = ginGetBAEntry(&buildstate.accum,
+ &attnum, &key, &category, &nlist)) != NULL)
+ {
+ /* there could be many entries, so be willing to abort here */
+ CHECK_FOR_INTERRUPTS();
+ ginEntryInsert(&buildstate.ginstate, attnum, key, category,
+ list, nlist, &buildstate.buildStats);
+ }
+ MemoryContextSwitchTo(oldCtx);
}
- MemoryContextSwitchTo(oldCtx);
MemoryContextDelete(buildstate.funcCtx);
MemoryContextDelete(buildstate.tmpCtx);
@@ -534,3 +841,1006 @@ gininsert(Relation index, Datum *values, bool *isnull,
return false;
}
+
+/*
+ * 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 GinLeader, which caller must use to shut down parallel
+ * mode by passing it to _gin_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
+_gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index,
+ bool isconcurrent, int request)
+{
+ ParallelContext *pcxt;
+ int scantuplesortstates;
+ Snapshot snapshot;
+ Size estginshared;
+ Size estsort;
+ GinShared *ginshared;
+ Sharedsort *sharedsort;
+ GinLeader *ginleader = (GinLeader *) palloc0(sizeof(GinLeader));
+ 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 gin index
+ */
+ EnterParallelMode();
+ Assert(request > 0);
+ pcxt = CreateParallelContext("postgres", "_gin_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_GIN_SHARED workspace.
+ */
+ estginshared = _gin_parallel_estimate_shared(heap, snapshot);
+ shm_toc_estimate_chunk(&pcxt->estimator, estginshared);
+ 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 */
+ ginshared = (GinShared *) shm_toc_allocate(pcxt->toc, estginshared);
+ /* Initialize immutable state */
+ ginshared->heaprelid = RelationGetRelid(heap);
+ ginshared->indexrelid = RelationGetRelid(index);
+ ginshared->isconcurrent = isconcurrent;
+ ginshared->scantuplesortstates = scantuplesortstates;
+
+ ConditionVariableInit(&ginshared->workersdonecv);
+ SpinLockInit(&ginshared->mutex);
+
+ /* Initialize mutable state */
+ ginshared->nparticipantsdone = 0;
+ ginshared->reltuples = 0.0;
+ ginshared->indtuples = 0.0;
+
+ table_parallelscan_initialize(heap,
+ ParallelTableScanFromGinShared(ginshared),
+ 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_GIN_SHARED, ginshared);
+ 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);
+ ginleader->pcxt = pcxt;
+ ginleader->nparticipanttuplesorts = pcxt->nworkers_launched;
+ if (leaderparticipates)
+ ginleader->nparticipanttuplesorts++;
+ ginleader->ginshared = ginshared;
+ ginleader->sharedsort = sharedsort;
+ ginleader->snapshot = snapshot;
+ ginleader->walusage = walusage;
+ ginleader->bufferusage = bufferusage;
+
+ /* If no workers were successfully launched, back out (do serial build) */
+ if (pcxt->nworkers_launched == 0)
+ {
+ _gin_end_parallel(ginleader, NULL);
+ return;
+ }
+
+ /* Save leader state now that it's clear build will be parallel */
+ buildstate->bs_leader = ginleader;
+
+ /* Join heap scan ourselves */
+ if (leaderparticipates)
+ _gin_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
+_gin_end_parallel(GinLeader *ginleader, GinBuildState *state)
+{
+ int i;
+
+ /* Shutdown worker processes */
+ WaitForParallelWorkersToFinish(ginleader->pcxt);
+
+ /*
+ * Next, accumulate WAL usage. (This must wait for the workers to finish,
+ * or we might get incomplete data.)
+ */
+ for (i = 0; i < ginleader->pcxt->nworkers_launched; i++)
+ InstrAccumParallelQuery(&ginleader->bufferusage[i], &ginleader->walusage[i]);
+
+ /* Free last reference to MVCC snapshot, if one was used */
+ if (IsMVCCSnapshot(ginleader->snapshot))
+ UnregisterSnapshot(ginleader->snapshot);
+ DestroyParallelContext(ginleader->pcxt);
+ ExitParallelMode();
+}
+
+/*
+ * Within leader, wait for end of heap scan.
+ *
+ * When called, parallel heap scan started by _gin_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
+_gin_parallel_heapscan(GinBuildState *state)
+{
+ GinShared *ginshared = state->bs_leader->ginshared;
+ int nparticipanttuplesorts;
+
+ nparticipanttuplesorts = state->bs_leader->nparticipanttuplesorts;
+ for (;;)
+ {
+ SpinLockAcquire(&ginshared->mutex);
+ if (ginshared->nparticipantsdone == nparticipanttuplesorts)
+ {
+ /* copy the data into leader state */
+ state->bs_reltuples = ginshared->reltuples;
+ state->bs_numtuples = ginshared->indtuples;
+
+ SpinLockRelease(&ginshared->mutex);
+ break;
+ }
+ SpinLockRelease(&ginshared->mutex);
+
+ ConditionVariableSleep(&ginshared->workersdonecv,
+ WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN);
+ }
+
+ ConditionVariableCancelSleep();
+
+ return state->bs_reltuples;
+}
+
+static int
+tid_cmp(const void *a, const void *b)
+{
+ return ItemPointerCompare((ItemPointer) a, (ItemPointer) b);
+}
+
+/*
+ * State used to combine accumulate TIDs from multiple GinTuples for the same
+ * key value.
+ *
+ * XXX Similar purpose to BuildAccumulator, but much simpler.
+ */
+typedef struct GinBuffer
+{
+ OffsetNumber attnum;
+ GinNullCategory category;
+ Datum key; /* 0 if no key (and keylen == 0) */
+ Size keylen; /* number of bytes (not typlen) */
+
+ /* type info */
+ int16 typlen;
+ bool typbyval;
+
+ /* array of TID values */
+ int nitems;
+ int maxitems;
+ ItemPointerData *items;
+} GinBuffer;
+
+/* XXX should do more checks */
+static void
+AssertCheckGinBuffer(GinBuffer *buffer)
+{
+#ifdef USE_ASSERT_CHECKING
+ Assert(buffer->nitems <= buffer->maxitems);
+#endif
+}
+
+static void
+AssertCheckItemPointers(ItemPointerData *items, int nitems, bool sorted)
+{
+#ifdef USE_ASSERT_CHECKING
+ for (int i = 0; i < nitems; i++)
+ {
+ Assert(ItemPointerIsValid(&items[i]));
+
+ if ((i == 0) || !sorted)
+ continue;
+
+ Assert(ItemPointerCompare(&items[i - 1], &items[i]) < 0);
+ }
+#endif
+}
+
+static GinBuffer *
+GinBufferInit(void)
+{
+ return palloc0(sizeof(GinBuffer));
+}
+
+static bool
+GinBufferIsEmpty(GinBuffer *buffer)
+{
+ return (buffer->nitems == 0);
+}
+
+/*
+ * Compare if the tuple matches the already accumulated data. Compare
+ * scalar fields first, before the actual key.
+ *
+ * XXX The key is compared using memcmp, which means that if a key has
+ * multiple binary representations, we may end up treating them as
+ * different here. But that's OK, the index will merge them anyway.
+ */
+static bool
+GinBufferKeyEquals(GinBuffer *buffer, GinTuple *tup)
+{
+ AssertCheckGinBuffer(buffer);
+
+ if (tup->attrnum != buffer->attnum)
+ return false;
+
+ /* same attribute should have the same type info */
+ Assert(tup->typbyval == buffer->typbyval);
+ Assert(tup->typlen == buffer->typlen);
+
+ if (tup->category != buffer->category)
+ return false;
+
+ if (tup->keylen != buffer->keylen)
+ return false;
+
+ /*
+ * For NULL/empty keys, this means equality, for normal keys we need to
+ * compare the actual key value.
+ */
+ if (buffer->category != GIN_CAT_NORM_KEY)
+ return true;
+
+ /*
+ * Compare the key value, depending on the type information.
+ *
+ * XXX Not sure this works correctly for byval types that don't need the
+ * whole Datum. What if there is garbage in the padding bytes?
+ */
+ if (buffer->typbyval)
+ return (buffer->key == *(Datum *) tup->data);
+
+ /* byref values simply uses memcmp for comparison */
+ return (memcmp(tup->data, DatumGetPointer(buffer->key), buffer->keylen) == 0);
+}
+
+static void
+GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
+{
+ ItemPointerData *items;
+ Datum key;
+
+ AssertCheckGinBuffer(buffer);
+
+ key = _gin_parse_tuple(tup, &items);
+
+ /* if the buffer is empty, set the fields (and copy the key) */
+ if (GinBufferIsEmpty(buffer))
+ {
+ buffer->category = tup->category;
+ buffer->keylen = tup->keylen;
+ buffer->attnum = tup->attrnum;
+
+ buffer->typlen = tup->typlen;
+ buffer->typbyval = tup->typbyval;
+
+ if (tup->category == GIN_CAT_NORM_KEY)
+ buffer->key = datumCopy(key, buffer->typbyval, buffer->typlen);
+ else
+ buffer->key = (Datum) 0;
+ }
+
+ /* enlarge the TID buffer, if needed */
+ if (buffer->nitems + tup->nitems > buffer->maxitems)
+ {
+ /* 64 seems like a good init value */
+ buffer->maxitems = Max(buffer->maxitems, 64);
+
+ while (buffer->nitems + tup->nitems > buffer->maxitems)
+ buffer->maxitems *= 2;
+
+ if (buffer->items == NULL)
+ buffer->items = palloc(buffer->maxitems * sizeof(ItemPointerData));
+ else
+ buffer->items = repalloc(buffer->items,
+ buffer->maxitems * sizeof(ItemPointerData));
+ }
+
+ /* now we should be guaranteed to have enough space for all the TIDs */
+ Assert(buffer->nitems + tup->nitems <= buffer->maxitems);
+
+ /* copy the new TIDs into the buffer */
+ memcpy(&buffer->items[buffer->nitems], items, sizeof(ItemPointerData) * tup->nitems);
+ buffer->nitems += tup->nitems;
+
+ AssertCheckItemPointers(buffer->items, buffer->nitems, false);
+}
+
+static void
+GinBufferSortItems(GinBuffer *buffer)
+{
+ /* we should not have a buffer with no TIDs to sort */
+ Assert(buffer->items != NULL);
+ Assert(buffer->nitems > 0);
+
+ pg_qsort(buffer->items, buffer->nitems, sizeof(ItemPointerData), tid_cmp);
+
+ AssertCheckItemPointers(buffer->items, buffer->nitems, true);
+}
+
+/* XXX probably would be better to have a memory context for the buffer */
+static void
+GinBufferReset(GinBuffer *buffer)
+{
+ Assert(!GinBufferIsEmpty(buffer));
+
+ /* release byref values, do nothing for by-val ones */
+ if ((buffer->category == GIN_CAT_NORM_KEY) && !buffer->typbyval)
+ pfree(DatumGetPointer(buffer->key));
+
+ /* XXX not really needed, but easier to trigger NULL deref etc. */
+ buffer->key = (Datum) 0;
+
+ buffer->attnum = 0;
+ buffer->category = 0;
+ buffer->keylen = 0;
+ buffer->nitems = 0;
+
+ buffer->typlen = 0;
+ buffer->typbyval = 0;
+
+ /* XXX should do something with extremely large array of items? */
+}
+
+/*
+ * XXX Maybe check size of the TID arrays, and return false if it's too
+ * large (more thant maintenance_work_mem or something?).
+ */
+static bool
+GinBufferCanAddKey(GinBuffer *buffer, GinTuple *tup)
+{
+ /* empty buffer can accept data for any key */
+ if (GinBufferIsEmpty(buffer))
+ return true;
+
+ /* otherwise just data for the same key */
+ return GinBufferKeyEquals(buffer, tup);
+}
+
+/*
+ * 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 combinig 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.
+ *
+ * FIXME probably should have local memory contexts similar to what
+ * _brin_parallel_merge does.
+ */
+static double
+_gin_parallel_merge(GinBuildState *state)
+{
+ GinTuple *tup;
+ Size tuplen;
+ double reltuples = 0;
+ GinBuffer *buffer;
+
+ /* wait for workers to scan table and produce partial results */
+ reltuples = _gin_parallel_heapscan(state);
+
+ /* do the actual sort in the leader */
+ tuplesort_performsort(state->bs_sortstate);
+
+ /* initialize buffer to combine entries for the same key */
+ buffer = GinBufferInit();
+
+ /*
+ * Read the GIN tuples from the shared tuplesort, sorted by category and
+ * key. That probably gives us order matching how data is organized in the
+ * index.
+ *
+ * XXX Maybe we should sort by key first, then by category?
+ *
+ * We don't insert the GIN tuples right away, but instead accumulate as
+ * many TIDs for the same key as possible, and then insert that at once.
+ * This way we don't need to decompress/recompress the posting lists, etc.
+ */
+ while ((tup = tuplesort_getgintuple(state->bs_sortstate, &tuplen, true)) != NULL)
+ {
+ CHECK_FOR_INTERRUPTS();
+
+ /*
+ * If the buffer can accept the new GIN tuple, just store it there and
+ * we're done. If it's a different key (or maybe too much data) flush
+ * the current contents into the index first.
+ */
+ if (!GinBufferCanAddKey(buffer, tup))
+ {
+ /*
+ * Buffer is not empty and it's storing a different key - flush
+ * the data into the insert, and start a new entry for current
+ * GinTuple.
+ */
+ GinBufferSortItems(buffer);
+
+ ginEntryInsert(&state->ginstate,
+ buffer->attnum, buffer->key, buffer->category,
+ buffer->items, buffer->nitems, &state->buildStats);
+
+ /* discard the existing data */
+ GinBufferReset(buffer);
+ }
+
+ /* now remember the new key */
+ GinBufferStoreTuple(buffer, tup);
+ }
+
+ /* flush data remaining in the buffer (for the last key) */
+ if (!GinBufferIsEmpty(buffer))
+ {
+ GinBufferSortItems(buffer);
+
+ ginEntryInsert(&state->ginstate,
+ buffer->attnum, buffer->key, buffer->category,
+ buffer->items, buffer->nitems, &state->buildStats);
+
+ /* discard the existing data */
+ GinBufferReset(buffer);
+ }
+
+ tuplesort_end(state->bs_sortstate);
+
+ return reltuples;
+}
+
+/*
+ * Returns size of shared memory required to store state for a parallel
+ * gin index build based on the snapshot its parallel scan will use.
+ */
+static Size
+_gin_parallel_estimate_shared(Relation heap, Snapshot snapshot)
+{
+ /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */
+ return add_size(BUFFERALIGN(sizeof(GinShared)),
+ table_parallelscan_estimate(heap, snapshot));
+}
+
+/*
+ * Within leader, participate as a parallel worker.
+ */
+static void
+_gin_leader_participate_as_worker(GinBuildState *buildstate, Relation heap, Relation index)
+{
+ GinLeader *ginleader = 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 / ginleader->nparticipanttuplesorts;
+
+ /* Perform work common to all participants */
+ _gin_parallel_scan_and_build(buildstate, ginleader->ginshared,
+ ginleader->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
+_gin_parallel_scan_and_build(GinBuildState *state,
+ GinShared *ginshared, 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(sizeof(SortCoordinateData));
+ coordinate->isWorker = true;
+ coordinate->nParticipants = -1;
+ coordinate->sharedsort = sharedsort;
+
+ /* Begin "partial" tuplesort */
+ state->bs_sortstate = tuplesort_begin_index_gin(sortmem, coordinate,
+ TUPLESORT_NONE);
+
+ /* Join parallel scan */
+ indexInfo = BuildIndexInfo(index);
+ indexInfo->ii_Concurrent = ginshared->isconcurrent;
+
+ scan = table_beginscan_parallel(heap,
+ ParallelTableScanFromGinShared(ginshared));
+
+ reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
+ ginBuildCallbackParallel, state, scan);
+
+ /* insert the last item */
+ /* write remaining accumulated entries */
+ {
+ ItemPointerData *list;
+ Datum key;
+ GinNullCategory category;
+ uint32 nlist;
+ OffsetNumber attnum;
+ TupleDesc tdesc = RelationGetDescr(index);
+
+ ginBeginBAScan(&state->accum);
+ while ((list = ginGetBAEntry(&state->accum,
+ &attnum, &key, &category, &nlist)) != NULL)
+ {
+ /* information about the key */
+ Form_pg_attribute attr = TupleDescAttr(tdesc, (attnum - 1));
+
+ GinTuple *tup;
+ Size len;
+
+ /* there could be many entries, so be willing to abort here */
+ CHECK_FOR_INTERRUPTS();
+
+ tup = _gin_build_tuple(attnum, category,
+ key, attr->attlen, attr->attbyval,
+ list, nlist, &len);
+
+ tuplesort_putgintuple(state->bs_sortstate, tup, len);
+
+ pfree(tup);
+ }
+
+ MemoryContextReset(state->tmpCtx);
+ ginInitBA(&state->accum);
+ }
+
+ /* sort the GIN tuples built by this worker */
+ tuplesort_performsort(state->bs_sortstate);
+
+ state->bs_reltuples += reltuples;
+
+ /*
+ * Done. Record ambuild statistics.
+ */
+ SpinLockAcquire(&ginshared->mutex);
+ ginshared->nparticipantsdone++;
+ ginshared->reltuples += state->bs_reltuples;
+ ginshared->indtuples += state->bs_numtuples;
+ SpinLockRelease(&ginshared->mutex);
+
+ /* Notify leader */
+ ConditionVariableSignal(&ginshared->workersdonecv);
+
+ tuplesort_end(state->bs_sortstate);
+}
+
+/*
+ * Perform work within a launched parallel process.
+ */
+void
+_gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
+{
+ char *sharedquery;
+ GinShared *ginshared;
+ Sharedsort *sharedsort;
+ GinBuildState 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 gin shared state */
+ ginshared = shm_toc_lookup(toc, PARALLEL_KEY_GIN_SHARED, false);
+
+ /* Open relations using lock modes known to be obtained by index.c */
+ if (!ginshared->isconcurrent)
+ {
+ heapLockmode = ShareLock;
+ indexLockmode = AccessExclusiveLock;
+ }
+ else
+ {
+ heapLockmode = ShareUpdateExclusiveLock;
+ indexLockmode = RowExclusiveLock;
+ }
+
+ /* Open relations within worker */
+ heapRel = table_open(ginshared->heaprelid, heapLockmode);
+ indexRel = index_open(ginshared->indexrelid, indexLockmode);
+
+ /* initialize the GIN build state */
+ initGinState(&buildstate.ginstate, indexRel);
+ buildstate.indtuples = 0;
+ memset(&buildstate.buildStats, 0, sizeof(GinStatsData));
+
+ /*
+ * create a temporary memory context that is used to hold data not yet
+ * dumped out to the index
+ */
+ buildstate.tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
+ "Gin build temporary context",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /*
+ * create a temporary memory context that is used for calling
+ * ginExtractEntries(), and can be reset after each tuple
+ */
+ buildstate.funcCtx = AllocSetContextCreate(CurrentMemoryContext,
+ "Gin build temporary context for user-defined function",
+ ALLOCSET_DEFAULT_SIZES);
+
+ buildstate.accum.ginstate = &buildstate.ginstate;
+ ginInitBA(&buildstate.accum);
+
+
+ /* 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 / ginshared->scantuplesortstates;
+
+ _gin_parallel_scan_and_build(&buildstate, ginshared, 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);
+}
+
+/*
+ * _gin_build_tuple
+ * Serialize the state for an index key into a tuple for tuplesort.
+ *
+ * The tuple has a number of scalar fields (mostly matching the build state),
+ * and then a data array that stores the key first, and then the TID list.
+ *
+ * For by-reference data types, we store the actual data. For by-val types
+ * we simply copy the whole Datum, so that we don't have to care about stuff
+ * like endianess etc. We could make it a little bit smaller, but it's not
+ * worth it - it's a tiny fraction of the data, and we need to MAXALIGN the
+ * start of the TID list anyway. So we wouldn't save anything.
+ */
+static GinTuple *
+_gin_build_tuple(OffsetNumber attrnum, unsigned char category,
+ Datum key, int16 typlen, bool typbyval,
+ ItemPointerData *items, uint32 nitems,
+ Size *len)
+{
+ GinTuple *tuple;
+ char *ptr;
+
+ Size tuplen;
+ int keylen;
+
+ /*
+ * Calculate how long is the key value. Only keys with GIN_CAT_NORM_KEY
+ * have actual non-empty key. We include varlena headers and \0 bytes for
+ * strings, to make it easier to access the data in-line.
+ *
+ * For byval types we simply copy the whole Datum. We could store just the
+ * necessary bytes, but this is simpler to work with and not worth the
+ * extra complexity. Moreover we still need to do the MAXALIGN to allow
+ * direct access to items pointers.
+ */
+ if (category != GIN_CAT_NORM_KEY)
+ keylen = 0;
+ else if (typbyval)
+ keylen = sizeof(Datum);
+ else if (typlen > 0)
+ keylen = typlen;
+ else if (typlen == -1)
+ keylen = VARSIZE_ANY(key);
+ else if (typlen == -2)
+ keylen = strlen(DatumGetPointer(key)) + 1;
+ else
+ elog(ERROR, "invalid typlen");
+
+ /*
+ * Determine GIN tuple length with all the data included. Be careful about
+ * alignment, to allow direct access to item pointers.
+ */
+ tuplen = MAXALIGN(offsetof(GinTuple, data) + keylen) +
+ (sizeof(ItemPointerData) * nitems);
+
+ *len = tuplen;
+
+ /*
+ * allocate space for the whole GIN tuple
+ *
+ * XXX palloc0 so that valgrind does not complain about uninitialized
+ * bytes in writetup_index_gin, likely because of padding
+ */
+ tuple = palloc0(tuplen);
+
+ tuple->tuplen = tuplen;
+ tuple->attrnum = attrnum;
+ tuple->category = category;
+ tuple->keylen = keylen;
+ tuple->nitems = nitems;
+
+ /* key type info */
+ tuple->typlen = typlen;
+ tuple->typbyval = typbyval;
+
+ /*
+ * Copy the key and items into the tuple. First the key value, which we
+ * can simply copy right at the beginning of the data array.
+ */
+ if (category == GIN_CAT_NORM_KEY)
+ {
+ if (typbyval)
+ {
+ memcpy(tuple->data, &key, sizeof(Datum));
+ }
+ else if (typlen > 0) /* byref, fixed length */
+ {
+ memcpy(tuple->data, DatumGetPointer(key), typlen);
+ }
+ else if (typlen == -1)
+ {
+ memcpy(tuple->data, DatumGetPointer(key), keylen);
+ }
+ else if (typlen == -2)
+ {
+ memcpy(tuple->data, DatumGetPointer(key), keylen);
+ }
+ }
+
+ /* finally, copy the TIDs into the array */
+ ptr = (char *) tuple + MAXALIGN(offsetof(GinTuple, data) + keylen);
+
+ memcpy(ptr, items, sizeof(ItemPointerData) * nitems);
+
+ return tuple;
+}
+
+/*
+ * _gin_parse_tuple
+ * Deserialize the tuple from the tuplestore representation.
+ *
+ * Most of the fields are actually directly accessible, the only thing that
+ * needs more care is the key and the TID list.
+ *
+ * For the key, this returns a regular Datum representing it. It's either the
+ * actual key value, or a pointer to the beginning of the data array (which is
+ * where the data was copied by _gin_build_tuple).
+ *
+ * The pointer to the TID list is returned through 'items' (which is simply
+ * a pointer to the data array).
+ */
+static Datum
+_gin_parse_tuple(GinTuple *a, ItemPointerData **items)
+{
+ Datum key;
+
+ if (items)
+ {
+ char *ptr = (char *) a + MAXALIGN(offsetof(GinTuple, data) + a->keylen);
+
+ *items = (ItemPointerData *) ptr;
+ }
+
+ if (a->category != GIN_CAT_NORM_KEY)
+ return (Datum) 0;
+
+ if (a->typbyval)
+ {
+ memcpy(&key, a->data, a->keylen);
+ return key;
+ }
+
+ return PointerGetDatum(a->data);
+}
+
+/*
+ * _gin_compare_tuples
+ * Compare GIN tuples, used by tuplesort during parallel index build.
+ *
+ * The scalar fields (attrnum, category) are compared first, the key value is
+ * compared last. The comparisons are done simply by "memcmp", based on the
+ * assumption that if we get two keys that are two different representations
+ * of a logically equal value, it'll get merged by the index build.
+ *
+ * FIXME Is the assumption we can just memcmp() actually valid? Won't this
+ * trigger the "could not split GIN page; all old items didn't fit" error
+ * when trying to update the TID list?
+ */
+int
+_gin_compare_tuples(GinTuple *a, GinTuple *b)
+{
+ Datum keya,
+ keyb;
+
+ if (a->attrnum < b->attrnum)
+ return -1;
+
+ if (a->attrnum > b->attrnum)
+ return 1;
+
+ if (a->category < b->category)
+ return -1;
+
+ if (a->category > b->category)
+ return 1;
+
+ if ((a->category == GIN_CAT_NORM_KEY) &&
+ (b->category == GIN_CAT_NORM_KEY))
+ {
+ keya = _gin_parse_tuple(a, NULL);
+ keyb = _gin_parse_tuple(b, NULL);
+
+ /*
+ * works for both byval and byref types with fixed lenght, because for
+ * byval we set keylen to sizeof(Datum)
+ */
+ if (a->typbyval)
+ {
+ return memcmp(&keya, &keyb, a->keylen);
+ }
+ else
+ {
+ if (a->keylen < b->keylen)
+ return -1;
+
+ if (a->keylen > b->keylen)
+ return 1;
+
+ return memcmp(DatumGetPointer(keya), DatumGetPointer(keyb), a->keylen);
+ }
+ }
+
+ return 0;
+}
diff --git a/src/backend/access/gin/ginutil.c b/src/backend/access/gin/ginutil.c
index 5747ae6a4ca..dd22b44aca9 100644
--- a/src/backend/access/gin/ginutil.c
+++ b/src/backend/access/gin/ginutil.c
@@ -53,7 +53,7 @@ ginhandler(PG_FUNCTION_ARGS)
amroutine->amclusterable = false;
amroutine->ampredlocks = true;
amroutine->amcanparallel = false;
- amroutine->amcanbuildparallel = false;
+ amroutine->amcanbuildparallel = true;
amroutine->amcaninclude = false;
amroutine->amusemaintenanceworkmem = true;
amroutine->amsummarizing = false;
diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
index 8613fc6fb54..c9ea769afb5 100644
--- a/src/backend/access/transam/parallel.c
+++ b/src/backend/access/transam/parallel.c
@@ -15,6 +15,7 @@
#include "postgres.h"
#include "access/brin.h"
+#include "access/gin.h"
#include "access/nbtree.h"
#include "access/parallel.h"
#include "access/session.h"
@@ -146,6 +147,9 @@ static const struct
{
"_brin_parallel_build_main", _brin_parallel_build_main
},
+ {
+ "_gin_parallel_build_main", _gin_parallel_build_main
+ },
{
"parallel_vacuum_main", parallel_vacuum_main
}
diff --git a/src/backend/utils/sort/tuplesortvariants.c b/src/backend/utils/sort/tuplesortvariants.c
index 05a853caa36..55cc55969e5 100644
--- a/src/backend/utils/sort/tuplesortvariants.c
+++ b/src/backend/utils/sort/tuplesortvariants.c
@@ -20,6 +20,7 @@
#include "postgres.h"
#include "access/brin_tuple.h"
+#include "access/gin_tuple.h"
#include "access/hash.h"
#include "access/htup_details.h"
#include "access/nbtree.h"
@@ -46,6 +47,8 @@ static void removeabbrev_index(Tuplesortstate *state, SortTuple *stups,
int count);
static void removeabbrev_index_brin(Tuplesortstate *state, SortTuple *stups,
int count);
+static void removeabbrev_index_gin(Tuplesortstate *state, SortTuple *stups,
+ int count);
static void removeabbrev_datum(Tuplesortstate *state, SortTuple *stups,
int count);
static int comparetup_heap(const SortTuple *a, const SortTuple *b,
@@ -74,6 +77,8 @@ static int comparetup_index_hash_tiebreak(const SortTuple *a, const SortTuple *b
Tuplesortstate *state);
static int comparetup_index_brin(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
+static int comparetup_index_gin(const SortTuple *a, const SortTuple *b,
+ Tuplesortstate *state);
static void writetup_index(Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
static void readtup_index(Tuplesortstate *state, SortTuple *stup,
@@ -82,6 +87,10 @@ static void writetup_index_brin(Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
static void readtup_index_brin(Tuplesortstate *state, SortTuple *stup,
LogicalTape *tape, unsigned int len);
+static void writetup_index_gin(Tuplesortstate *state, LogicalTape *tape,
+ SortTuple *stup);
+static void readtup_index_gin(Tuplesortstate *state, SortTuple *stup,
+ LogicalTape *tape, unsigned int len);
static int comparetup_datum(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static int comparetup_datum_tiebreak(const SortTuple *a, const SortTuple *b,
@@ -580,6 +589,35 @@ tuplesort_begin_index_brin(int workMem,
return state;
}
+
+Tuplesortstate *
+tuplesort_begin_index_gin(int workMem, SortCoordinate coordinate,
+ int sortopt)
+{
+ Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
+ sortopt);
+ TuplesortPublic *base = TuplesortstateGetPublic(state);
+
+#ifdef TRACE_SORT
+ if (trace_sort)
+ elog(LOG,
+ "begin index sort: workMem = %d, randomAccess = %c",
+ workMem,
+ sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
+#endif
+
+ base->nKeys = 1; /* Only the index key */
+
+ base->removeabbrev = removeabbrev_index_gin;
+ base->comparetup = comparetup_index_gin;
+ base->writetup = writetup_index_gin;
+ base->readtup = readtup_index_gin;
+ base->haveDatum1 = false;
+ base->arg = NULL;
+
+ return state;
+}
+
Tuplesortstate *
tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
bool nullsFirstFlag, int workMem,
@@ -817,6 +855,37 @@ tuplesort_putbrintuple(Tuplesortstate *state, BrinTuple *tuple, Size size)
MemoryContextSwitchTo(oldcontext);
}
+void
+tuplesort_putgintuple(Tuplesortstate *state, GinTuple *tup, Size size)
+{
+ SortTuple stup;
+ GinTuple *ctup;
+ TuplesortPublic *base = TuplesortstateGetPublic(state);
+ MemoryContext oldcontext = MemoryContextSwitchTo(base->tuplecontext);
+ Size tuplen;
+
+ /* copy the GinTuple into the right memory context */
+ ctup = palloc(size);
+ memcpy(ctup, tup, size);
+
+ stup.tuple = ctup;
+ stup.datum1 = (Datum) 0;
+ stup.isnull1 = false;
+
+ /* GetMemoryChunkSpace is not supported for bump contexts */
+ if (TupleSortUseBumpTupleCxt(base->sortopt))
+ tuplen = MAXALIGN(size);
+ else
+ tuplen = GetMemoryChunkSpace(ctup);
+
+ tuplesort_puttuple_common(state, &stup,
+ base->sortKeys &&
+ base->sortKeys->abbrev_converter &&
+ !stup.isnull1, tuplen);
+
+ MemoryContextSwitchTo(oldcontext);
+}
+
/*
* Accept one Datum while collecting input data for sort.
*
@@ -989,6 +1058,29 @@ tuplesort_getbrintuple(Tuplesortstate *state, Size *len, bool forward)
return &btup->tuple;
}
+GinTuple *
+tuplesort_getgintuple(Tuplesortstate *state, Size *len, bool forward)
+{
+ TuplesortPublic *base = TuplesortstateGetPublic(state);
+ MemoryContext oldcontext = MemoryContextSwitchTo(base->sortcontext);
+ SortTuple stup;
+ GinTuple *tup;
+
+ if (!tuplesort_gettuple_common(state, forward, &stup))
+ stup.tuple = NULL;
+
+ MemoryContextSwitchTo(oldcontext);
+
+ if (!stup.tuple)
+ return false;
+
+ tup = (GinTuple *) stup.tuple;
+
+ *len = tup->tuplen;
+
+ return tup;
+}
+
/*
* Fetch the next Datum in either forward or back direction.
* Returns false if no more datums.
@@ -1777,6 +1869,68 @@ readtup_index_brin(Tuplesortstate *state, SortTuple *stup,
stup->datum1 = tuple->tuple.bt_blkno;
}
+
+/*
+ * Routines specialized for GIN case
+ */
+
+static void
+removeabbrev_index_gin(Tuplesortstate *state, SortTuple *stups, int count)
+{
+ Assert(false);
+ elog(ERROR, "removeabbrev_index_gin not implemented");
+}
+
+static int
+comparetup_index_gin(const SortTuple *a, const SortTuple *b,
+ Tuplesortstate *state)
+{
+ Assert(!TuplesortstateGetPublic(state)->haveDatum1);
+
+ return _gin_compare_tuples((GinTuple *) a->tuple,
+ (GinTuple *) b->tuple);
+}
+
+static void
+writetup_index_gin(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
+{
+ TuplesortPublic *base = TuplesortstateGetPublic(state);
+ GinTuple *tuple = (GinTuple *) stup->tuple;
+ unsigned int tuplen = tuple->tuplen;
+
+ tuplen = tuplen + sizeof(tuplen);
+ LogicalTapeWrite(tape, &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, tuple, tuple->tuplen);
+ if (base->sortopt & TUPLESORT_RANDOMACCESS) /* need trailing length word? */
+ LogicalTapeWrite(tape, &tuplen, sizeof(tuplen));
+}
+
+static void
+readtup_index_gin(Tuplesortstate *state, SortTuple *stup,
+ LogicalTape *tape, unsigned int len)
+{
+ GinTuple *tuple;
+ TuplesortPublic *base = TuplesortstateGetPublic(state);
+ unsigned int tuplen = len - sizeof(unsigned int);
+
+ /*
+ * Allocate space for the GIN sort tuple, which already has the proper
+ * length included in the header.
+ */
+ tuple = (GinTuple *) tuplesort_readtup_alloc(state, tuplen);
+
+ tuple->tuplen = tuplen;
+
+ LogicalTapeReadExact(tape, tuple, tuplen);
+ if (base->sortopt & TUPLESORT_RANDOMACCESS) /* need trailing length word? */
+ LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
+ stup->tuple = (void *) tuple;
+
+ /* no abbreviations (FIXME maybe use attrnum for this?) */
+ stup->datum1 = (Datum) 0;
+}
+
+
/*
* Routines specialized for DatumTuple case
*/
diff --git a/src/include/access/gin.h b/src/include/access/gin.h
index 25983b7a505..be76d8446f4 100644
--- a/src/include/access/gin.h
+++ b/src/include/access/gin.h
@@ -12,6 +12,8 @@
#include "access/xlogreader.h"
#include "lib/stringinfo.h"
+#include "nodes/execnodes.h"
+#include "storage/shm_toc.h"
#include "storage/block.h"
#include "utils/relcache.h"
@@ -88,4 +90,6 @@ extern void ginGetStats(Relation index, GinStatsData *stats);
extern void ginUpdateStats(Relation index, const GinStatsData *stats,
bool is_build);
+extern void _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc);
+
#endif /* GIN_H */
diff --git a/src/include/access/gin_tuple.h b/src/include/access/gin_tuple.h
new file mode 100644
index 00000000000..56aed40fb96
--- /dev/null
+++ b/src/include/access/gin_tuple.h
@@ -0,0 +1,29 @@
+/*--------------------------------------------------------------------------
+ * gin.h
+ * Public header file for Generalized Inverted Index access method.
+ *
+ * Copyright (c) 2006-2024, PostgreSQL Global Development Group
+ *
+ * src/include/access/gin.h
+ *--------------------------------------------------------------------------
+ */
+#ifndef GIN_TUPLE_
+#define GIN_TUPLE_
+
+#include "storage/itemptr.h"
+
+typedef struct GinTuple
+{
+ Size tuplen; /* length of the whole tuple */
+ Size keylen; /* bytes in data for key value */
+ int16 typlen; /* typlen for key */
+ bool typbyval; /* typbyval for key */
+ OffsetNumber attrnum;
+ signed char category; /* category: normal or NULL? */
+ int nitems; /* number of TIDs in the data */
+ char data[FLEXIBLE_ARRAY_MEMBER];
+} GinTuple;
+
+extern int _gin_compare_tuples(GinTuple *a, GinTuple *b);
+
+#endif /* GIN_TUPLE_H */
diff --git a/src/include/utils/tuplesort.h b/src/include/utils/tuplesort.h
index e7941a1f09f..35fa5ae2442 100644
--- a/src/include/utils/tuplesort.h
+++ b/src/include/utils/tuplesort.h
@@ -22,6 +22,7 @@
#define TUPLESORT_H
#include "access/brin_tuple.h"
+#include "access/gin_tuple.h"
#include "access/itup.h"
#include "executor/tuptable.h"
#include "storage/dsm.h"
@@ -443,6 +444,8 @@ extern Tuplesortstate *tuplesort_begin_index_gist(Relation heapRel,
int sortopt);
extern Tuplesortstate *tuplesort_begin_index_brin(int workMem, SortCoordinate coordinate,
int sortopt);
+extern Tuplesortstate *tuplesort_begin_index_gin(int workMem, SortCoordinate coordinate,
+ int sortopt);
extern Tuplesortstate *tuplesort_begin_datum(Oid datumType,
Oid sortOperator, Oid sortCollation,
bool nullsFirstFlag,
@@ -456,6 +459,7 @@ extern void tuplesort_putindextuplevalues(Tuplesortstate *state,
Relation rel, ItemPointer self,
const Datum *values, const bool *isnull);
extern void tuplesort_putbrintuple(Tuplesortstate *state, BrinTuple *tup, Size len);
+extern void tuplesort_putgintuple(Tuplesortstate *state, GinTuple *tup, Size len);
extern void tuplesort_putdatum(Tuplesortstate *state, Datum val,
bool isNull);
@@ -465,6 +469,8 @@ extern HeapTuple tuplesort_getheaptuple(Tuplesortstate *state, bool forward);
extern IndexTuple tuplesort_getindextuple(Tuplesortstate *state, bool forward);
extern BrinTuple *tuplesort_getbrintuple(Tuplesortstate *state, Size *len,
bool forward);
+extern GinTuple *tuplesort_getgintuple(Tuplesortstate *state, Size *len,
+ bool forward);
extern bool tuplesort_getdatum(Tuplesortstate *state, bool forward, bool copy,
Datum *val, bool *isNull, Datum *abbrev);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index eee989bba17..9769f4d6b09 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1004,11 +1004,13 @@ GinBtreeData
GinBtreeDataLeafInsertData
GinBtreeEntryInsertData
GinBtreeStack
+GinBuffer
GinBuildState
GinChkVal
GinEntries
GinEntryAccumulator
GinIndexStat
+GinLeader
GinMetaPageData
GinNullCategory
GinOptions
@@ -1021,9 +1023,11 @@ GinScanEntry
GinScanKey
GinScanOpaque
GinScanOpaqueData
+GinShared
GinState
GinStatsData
GinTernaryValue
+GinTuple
GinTupleCollector
GinVacuumState
GistBuildMode
--
2.44.0
[text/x-patch] v20240505-0002-Use-mergesort-in-the-leader-process.patch (9.6K, ../[email protected]/3-v20240505-0002-Use-mergesort-in-the-leader-process.patch)
download | inline diff:
From f43aeb97f766b24092c3758fa5d6a9f0e6676eaf Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:32 +0200
Subject: [PATCH v20240505 2/8] Use mergesort in the leader process
The leader process (executing the serial part of the index build) spent
a significant part of the time in pg_qsort, after combining the partial
results from the workers. But we can improve this and move some of the
costs to the parallel part in workers - if workers produce sorted TID
lists, the leader can combine them by mergesort.
But to make this really efficient, the mergesort must not be executed
too many times. The workers may easily produce very short TID lists, if
there are many different keys, hitting the memory limit often. So this
adds an intermediate tuplesort pass into each worker, to combine TIDs
for each key and only then write the result into the shared tuplestore.
This means the number of mergesort invocations for each key should be
about the same as the number of workers. We can't really do better, and
it's low enough to keep the mergesort approach efficient.
Note: If we introduce a memory limit on GinBuffer (to not accumulate too
many TIDs in memory), we could end up with more chunks, but it should
not be very common.
---
src/backend/access/gin/gininsert.c | 171 +++++++++++++++++++++++++----
1 file changed, 148 insertions(+), 23 deletions(-)
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index b353e155fc6..cf7a6278914 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -161,6 +161,14 @@ typedef struct
* build callback etc.
*/
Tuplesortstate *bs_sortstate;
+
+ /*
+ * The sortstate is used only within a worker for the first merge pass
+ * that happens in the worker. In principle it doesn't need to be part of
+ * the build state and we could pass it around directly, but it's more
+ * convenient this way.
+ */
+ Tuplesortstate *bs_worker_sort;
} GinBuildState;
@@ -533,7 +541,7 @@ ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values,
key, attr->attlen, attr->attbyval,
list, nlist, &tuplen);
- tuplesort_putgintuple(buildstate->bs_sortstate, tup, tuplen);
+ tuplesort_putgintuple(buildstate->bs_worker_sort, tup, tuplen);
pfree(tup);
}
@@ -1127,7 +1135,6 @@ typedef struct GinBuffer
/* array of TID values */
int nitems;
- int maxitems;
ItemPointerData *items;
} GinBuffer;
@@ -1136,7 +1143,6 @@ static void
AssertCheckGinBuffer(GinBuffer *buffer)
{
#ifdef USE_ASSERT_CHECKING
- Assert(buffer->nitems <= buffer->maxitems);
#endif
}
@@ -1240,28 +1246,22 @@ GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
buffer->key = (Datum) 0;
}
- /* enlarge the TID buffer, if needed */
- if (buffer->nitems + tup->nitems > buffer->maxitems)
+ /* copy the new TIDs into the buffer, combine using merge-sort */
{
- /* 64 seems like a good init value */
- buffer->maxitems = Max(buffer->maxitems, 64);
+ int nnew;
+ ItemPointer new;
- while (buffer->nitems + tup->nitems > buffer->maxitems)
- buffer->maxitems *= 2;
+ new = ginMergeItemPointers(buffer->items, buffer->nitems,
+ items, tup->nitems, &nnew);
- if (buffer->items == NULL)
- buffer->items = palloc(buffer->maxitems * sizeof(ItemPointerData));
- else
- buffer->items = repalloc(buffer->items,
- buffer->maxitems * sizeof(ItemPointerData));
- }
+ Assert(nnew == buffer->nitems + tup->nitems);
- /* now we should be guaranteed to have enough space for all the TIDs */
- Assert(buffer->nitems + tup->nitems <= buffer->maxitems);
+ if (buffer->items)
+ pfree(buffer->items);
- /* copy the new TIDs into the buffer */
- memcpy(&buffer->items[buffer->nitems], items, sizeof(ItemPointerData) * tup->nitems);
- buffer->nitems += tup->nitems;
+ buffer->items = new;
+ buffer->nitems = nnew;
+ }
AssertCheckItemPointers(buffer->items, buffer->nitems, false);
}
@@ -1302,6 +1302,21 @@ GinBufferReset(GinBuffer *buffer)
/* XXX should do something with extremely large array of items? */
}
+/* XXX probably would be better to have a memory context for the buffer */
+static void
+GinBufferFree(GinBuffer *buffer)
+{
+ if (buffer->items)
+ pfree(buffer->items);
+
+ /* release byref values, do nothing for by-val ones */
+ if (!GinBufferIsEmpty(buffer) &&
+ (buffer->category == GIN_CAT_NORM_KEY) && !buffer->typbyval)
+ pfree(DatumGetPointer(buffer->key));
+
+ pfree(buffer);
+}
+
/*
* XXX Maybe check size of the TID arrays, and return false if it's too
* large (more thant maintenance_work_mem or something?).
@@ -1375,7 +1390,7 @@ _gin_parallel_merge(GinBuildState *state)
* the data into the insert, and start a new entry for current
* GinTuple.
*/
- GinBufferSortItems(buffer);
+ AssertCheckItemPointers(buffer->items, buffer->nitems, true);
ginEntryInsert(&state->ginstate,
buffer->attnum, buffer->key, buffer->category,
@@ -1392,7 +1407,7 @@ _gin_parallel_merge(GinBuildState *state)
/* flush data remaining in the buffer (for the last key) */
if (!GinBufferIsEmpty(buffer))
{
- GinBufferSortItems(buffer);
+ AssertCheckItemPointers(buffer->items, buffer->nitems, true);
ginEntryInsert(&state->ginstate,
buffer->attnum, buffer->key, buffer->category,
@@ -1402,6 +1417,9 @@ _gin_parallel_merge(GinBuildState *state)
GinBufferReset(buffer);
}
+ /* relase all the memory */
+ GinBufferFree(buffer);
+
tuplesort_end(state->bs_sortstate);
return reltuples;
@@ -1440,6 +1458,102 @@ _gin_leader_participate_as_worker(GinBuildState *buildstate, Relation heap, Rela
ginleader->sharedsort, heap, index, sortmem, true);
}
+/*
+ * _gin_process_worker_data
+ * First phase of the key merging, happening in the worker.
+ *
+ * Depending on the number of distinct keys, the TID lists produced by the
+ * callback may be very short. But combining many tiny lists is expensive,
+ * so we try to do as much as possible in the workers and only then pass the
+ * results to the leader.
+ *
+ * We read the tuples sorted by the key, and merge them into larger lists.
+ * At the moment there's no memory limit, so this will just produce one
+ * huge (sorted) list per key in each worker. Which means the leader will
+ * do a very limited number of mergesorts, which is good.
+ */
+static void
+_gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
+{
+ GinTuple *tup;
+ Size tuplen;
+
+ GinBuffer *buffer;
+
+ /* initialize buffer to combine entries for the same key */
+ buffer = GinBufferInit();
+
+ /* sort the raw per-worker data */
+ tuplesort_performsort(state->bs_worker_sort);
+
+ /*
+ * Read the GIN tuples from the shared tuplesort, sorted by the key, and
+ * merge them into larger chunks for the leader to combine.
+ */
+ while ((tup = tuplesort_getgintuple(worker_sort, &tuplen, true)) != NULL)
+ {
+
+ CHECK_FOR_INTERRUPTS();
+
+ /*
+ * If the buffer can accept the new GIN tuple, just store it there and
+ * we're done. If it's a different key (or maybe too much data) flush
+ * the current contents into the index first.
+ */
+ if (!GinBufferCanAddKey(buffer, tup))
+ {
+ GinTuple *ntup;
+ Size ntuplen;
+
+ /*
+ * Buffer is not empty and it's storing a different key - flush
+ * the data into the insert, and start a new entry for current
+ * GinTuple.
+ */
+ GinBufferSortItems(buffer);
+
+ ntup = _gin_build_tuple(buffer->attnum, buffer->category,
+ buffer->key, buffer->typlen, buffer->typbyval,
+ buffer->items, buffer->nitems, &ntuplen);
+
+ tuplesort_putgintuple(state->bs_sortstate, ntup, ntuplen);
+
+ pfree(ntup);
+
+ /* discard the existing data */
+ GinBufferReset(buffer);
+ }
+
+ /* now remember the new key */
+ GinBufferStoreTuple(buffer, tup);
+ }
+
+ /* flush data remaining in the buffer (for the last key) */
+ if (!GinBufferIsEmpty(buffer))
+ {
+ GinTuple *ntup;
+ Size ntuplen;
+
+ GinBufferSortItems(buffer);
+
+ ntup = _gin_build_tuple(buffer->attnum, buffer->category,
+ buffer->key, buffer->typlen, buffer->typbyval,
+ buffer->items, buffer->nitems, &ntuplen);
+
+ tuplesort_putgintuple(state->bs_sortstate, ntup, ntuplen);
+
+ pfree(ntup);
+
+ /* discard the existing data */
+ GinBufferReset(buffer);
+ }
+
+ /* relase all the memory */
+ GinBufferFree(buffer);
+
+ tuplesort_end(worker_sort);
+}
+
/*
* Perform a worker's portion of a parallel sort.
*
@@ -1471,6 +1585,10 @@ _gin_parallel_scan_and_build(GinBuildState *state,
state->bs_sortstate = tuplesort_begin_index_gin(sortmem, coordinate,
TUPLESORT_NONE);
+ /* Local per-worker sort of raw-data */
+ state->bs_worker_sort = tuplesort_begin_index_gin(sortmem, NULL,
+ TUPLESORT_NONE);
+
/* Join parallel scan */
indexInfo = BuildIndexInfo(index);
indexInfo->ii_Concurrent = ginshared->isconcurrent;
@@ -1508,7 +1626,7 @@ _gin_parallel_scan_and_build(GinBuildState *state,
key, attr->attlen, attr->attbyval,
list, nlist, &len);
- tuplesort_putgintuple(state->bs_sortstate, tup, len);
+ tuplesort_putgintuple(state->bs_worker_sort, tup, len);
pfree(tup);
}
@@ -1517,6 +1635,13 @@ _gin_parallel_scan_and_build(GinBuildState *state,
ginInitBA(&state->accum);
}
+ /*
+ * Do the first phase of in-worker processing - sort the data produced by
+ * the callback, and combine them into much larger chunks and place that
+ * into the shared tuplestore for leader to process.
+ */
+ _gin_process_worker_data(state, state->bs_worker_sort);
+
/* sort the GIN tuples built by this worker */
tuplesort_performsort(state->bs_sortstate);
--
2.44.0
[text/x-patch] v20240505-0003-Remove-the-explicit-pg_qsort-in-workers.patch (10.9K, ../[email protected]/4-v20240505-0003-Remove-the-explicit-pg_qsort-in-workers.patch)
download | inline diff:
From 45e7f09ec81932c54eef891017d2a10818dd25b6 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:36 +0200
Subject: [PATCH v20240505 3/8] Remove the explicit pg_qsort in workers
We don't need to do the explicit sort before building the GIN tuple,
because the mergesort in GinBufferStoreTuple is already maintaining the
correct order (this was added in an earlier commit).
The comment also adds a field with the first TID, and modifies the
comparator to sort by it (for each key value). This helps workers to
build non-overlapping TID lists and simply append values instead of
having to do the actual mergesort to combine them. This is best-effort,
i.e. it's not guaranteed to eliminate the mergesort - in particular,
parallel scans are synchronized, and thus may start somewhere in the
middle of the table, and wrap around. In which case there may be very
wide list (with low/high TID values).
Note: There's an XXX comment with a couple ideas on how to improve this,
at the cost of more complexity.
---
src/backend/access/gin/gininsert.c | 124 +++++++++++++++++++++--------
src/include/access/gin_tuple.h | 8 ++
2 files changed, 98 insertions(+), 34 deletions(-)
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index cf7a6278914..b2b44066329 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -1110,12 +1110,6 @@ _gin_parallel_heapscan(GinBuildState *state)
return state->bs_reltuples;
}
-static int
-tid_cmp(const void *a, const void *b)
-{
- return ItemPointerCompare((ItemPointer) a, (ItemPointer) b);
-}
-
/*
* State used to combine accumulate TIDs from multiple GinTuples for the same
* key value.
@@ -1147,17 +1141,21 @@ AssertCheckGinBuffer(GinBuffer *buffer)
}
static void
-AssertCheckItemPointers(ItemPointerData *items, int nitems, bool sorted)
+AssertCheckItemPointers(GinBuffer *buffer, bool sorted)
{
#ifdef USE_ASSERT_CHECKING
- for (int i = 0; i < nitems; i++)
+ /* we should not have a buffer with no TIDs to sort */
+ Assert(buffer->items != NULL);
+ Assert(buffer->nitems > 0);
+
+ for (int i = 0; i < buffer->nitems; i++)
{
- Assert(ItemPointerIsValid(&items[i]));
+ Assert(ItemPointerIsValid(&buffer->items[i]));
if ((i == 0) || !sorted)
continue;
- Assert(ItemPointerCompare(&items[i - 1], &items[i]) < 0);
+ Assert(ItemPointerCompare(&buffer->items[i - 1], &buffer->items[i]) < 0);
}
#endif
}
@@ -1220,6 +1218,45 @@ GinBufferKeyEquals(GinBuffer *buffer, GinTuple *tup)
return (memcmp(tup->data, DatumGetPointer(buffer->key), buffer->keylen) == 0);
}
+/*
+ * GinBufferStoreTuple
+ * Add data from a GinTuple into the GinBuffer.
+ *
+ * If the buffer is empty, we simply initialize it with data from the tuple.
+ * Otherwise data from the tuple (the TID list) is added to the TID data in
+ * the buffer, either by simply appending the TIDs or doing merge sort.
+ *
+ * The data (for the same key) is expected to be processed sorted by first
+ * TID. But this does not guarantee the lists do not overlap, especially in
+ * the leader, because the workers process interleaving data. But even in
+ * a single worker, lists can overlap - parallel scans require sync-scans,
+ * and if the scan starts somewhere in the table and then wraps around, it
+ * may contain very wide lists (in terms of TID range).
+ *
+ * But the ginMergeItemPointers() is already smart about detecting cases
+ * where it can simply concatenate the lists, and when full mergesort is
+ * needed. And does the right thing.
+ *
+ * By keeping the first TID in the GinTuple and sorting by that, we make
+ * it more likely the lists won't overlap very often.
+ *
+ * XXX How frequent can the overlaps be? If the scan does not wrap around,
+ * there should be no overlapping lists, and thus no mergesort. After ab
+ * overlap, there probably can be many - the one list will be very wide,
+ * with a very low and high TID, and all other lists will overlap with it.
+ * I'm not sure how much we can do to prevent that, short of disabling sync
+ * scans (which for parallel scans is currently not possible). One option
+ * would be to keep two lists of TIDs, and see if the new list can be
+ * concatenated with one of them. The idea is that there's only one wide
+ * list (because the wraparound happens only once), and then do the
+ * mergesort only once at the very end.
+ *
+ * XXX Alternatively, we could simply detect the case when the lists can't
+ * be appended, and flush the current list out. The wrap around happens only
+ * once, so there's going to be only such wide list, and it'll be sorted
+ * first (because it has the lowest TID for the key). So we'd do this at
+ * most once per key.
+ */
static void
GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
{
@@ -1246,7 +1283,12 @@ GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
buffer->key = (Datum) 0;
}
- /* copy the new TIDs into the buffer, combine using merge-sort */
+ /*
+ * Copy the new TIDs into the buffer, combine with existing data (if any)
+ * using merge-sort. The mergesort is already smart about cases where it
+ * can simply concatenate the two lists, and when it actually needs to
+ * merge the data in an expensive way.
+ */
{
int nnew;
ItemPointer new;
@@ -1261,21 +1303,9 @@ GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
buffer->items = new;
buffer->nitems = nnew;
- }
-
- AssertCheckItemPointers(buffer->items, buffer->nitems, false);
-}
-static void
-GinBufferSortItems(GinBuffer *buffer)
-{
- /* we should not have a buffer with no TIDs to sort */
- Assert(buffer->items != NULL);
- Assert(buffer->nitems > 0);
-
- pg_qsort(buffer->items, buffer->nitems, sizeof(ItemPointerData), tid_cmp);
-
- AssertCheckItemPointers(buffer->items, buffer->nitems, true);
+ AssertCheckItemPointers(buffer, true);
+ }
}
/* XXX probably would be better to have a memory context for the buffer */
@@ -1299,6 +1329,11 @@ GinBufferReset(GinBuffer *buffer)
buffer->typlen = 0;
buffer->typbyval = 0;
+ if (buffer->items)
+ {
+ pfree(buffer->items);
+ buffer->items = NULL;
+ }
/* XXX should do something with extremely large array of items? */
}
@@ -1390,7 +1425,7 @@ _gin_parallel_merge(GinBuildState *state)
* the data into the insert, and start a new entry for current
* GinTuple.
*/
- AssertCheckItemPointers(buffer->items, buffer->nitems, true);
+ AssertCheckItemPointers(buffer, true);
ginEntryInsert(&state->ginstate,
buffer->attnum, buffer->key, buffer->category,
@@ -1400,14 +1435,17 @@ _gin_parallel_merge(GinBuildState *state)
GinBufferReset(buffer);
}
- /* now remember the new key */
+ /*
+ * Remember data for the current tuple (either remember the new key,
+ * or append if to the existing data).
+ */
GinBufferStoreTuple(buffer, tup);
}
/* flush data remaining in the buffer (for the last key) */
if (!GinBufferIsEmpty(buffer))
{
- AssertCheckItemPointers(buffer->items, buffer->nitems, true);
+ AssertCheckItemPointers(buffer, true);
ginEntryInsert(&state->ginstate,
buffer->attnum, buffer->key, buffer->category,
@@ -1510,7 +1548,7 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
* the data into the insert, and start a new entry for current
* GinTuple.
*/
- GinBufferSortItems(buffer);
+ AssertCheckItemPointers(buffer, true);
ntup = _gin_build_tuple(buffer->attnum, buffer->category,
buffer->key, buffer->typlen, buffer->typbyval,
@@ -1524,7 +1562,10 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
GinBufferReset(buffer);
}
- /* now remember the new key */
+ /*
+ * Remember data for the current tuple (either remember the new key,
+ * or append if to the existing data).
+ */
GinBufferStoreTuple(buffer, tup);
}
@@ -1534,7 +1575,7 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
GinTuple *ntup;
Size ntuplen;
- GinBufferSortItems(buffer);
+ AssertCheckItemPointers(buffer, true);
ntup = _gin_build_tuple(buffer->attnum, buffer->category,
buffer->key, buffer->typlen, buffer->typbyval,
@@ -1835,6 +1876,7 @@ _gin_build_tuple(OffsetNumber attrnum, unsigned char category,
tuple->category = category;
tuple->keylen = keylen;
tuple->nitems = nitems;
+ tuple->first = items[0];
/* key type info */
tuple->typlen = typlen;
@@ -1919,6 +1961,12 @@ _gin_parse_tuple(GinTuple *a, ItemPointerData **items)
* assumption that if we get two keys that are two different representations
* of a logically equal value, it'll get merged by the index build.
*
+ * If the key value matches, we compare the first TID value in the TID list,
+ * which means the tuples are merged in an order in which they are most
+ * likely to be simply concatenated. (This "first" TID will also allow us
+ * to determine a point up to which the list is fully determined and can be
+ * written into the index to enforce a memory limit etc.)
+ *
* FIXME Is the assumption we can just memcmp() actually valid? Won't this
* trigger the "could not split GIN page; all old items didn't fit" error
* when trying to update the TID list?
@@ -1953,19 +2001,27 @@ _gin_compare_tuples(GinTuple *a, GinTuple *b)
*/
if (a->typbyval)
{
- return memcmp(&keya, &keyb, a->keylen);
+ int r = memcmp(&keya, &keyb, a->keylen);
+
+ /* if the key is the same, consider the first TID in the array */
+ return (r != 0) ? r : ItemPointerCompare(&a->first, &b->first);
}
else
{
+ int r;
+
if (a->keylen < b->keylen)
return -1;
if (a->keylen > b->keylen)
return 1;
- return memcmp(DatumGetPointer(keya), DatumGetPointer(keyb), a->keylen);
+ r = memcmp(DatumGetPointer(keya), DatumGetPointer(keyb), a->keylen);
+
+ /* if the key is the same, consider the first TID in the array */
+ return (r != 0) ? r : ItemPointerCompare(&a->first, &b->first);
}
}
- return 0;
+ return ItemPointerCompare(&a->first, &b->first);
}
diff --git a/src/include/access/gin_tuple.h b/src/include/access/gin_tuple.h
index 56aed40fb96..8efa33a8d31 100644
--- a/src/include/access/gin_tuple.h
+++ b/src/include/access/gin_tuple.h
@@ -12,6 +12,13 @@
#include "storage/itemptr.h"
+/*
+ * Each worker sees tuples in CTID order, so if we track the first TID and
+ * compare that when combining results in the worker, we would not need to
+ * do an expensive sort in workers (the mergesort is already smart about
+ * detecting this and just concatenating the lists). We'd still need the
+ * full mergesort in the leader, but that's much cheaper.
+ */
typedef struct GinTuple
{
Size tuplen; /* length of the whole tuple */
@@ -20,6 +27,7 @@ typedef struct GinTuple
bool typbyval; /* typbyval for key */
OffsetNumber attrnum;
signed char category; /* category: normal or NULL? */
+ ItemPointerData first; /* first TID in the array */
int nitems; /* number of TIDs in the data */
char data[FLEXIBLE_ARRAY_MEMBER];
} GinTuple;
--
2.44.0
[text/x-patch] v20240505-0004-Compress-TID-lists-before-writing-tuples-t.patch (8.0K, ../[email protected]/5-v20240505-0004-Compress-TID-lists-before-writing-tuples-t.patch)
download | inline diff:
From 08f8ecd7a21370cc452a6185781428729ad58330 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:39 +0200
Subject: [PATCH v20240505 4/8] Compress TID lists before writing tuples to
disk
When serializing GIN tuples to tuplesorts, we can significantly reduce
the amount of data by compressing the TID lists. The GIN opclasses may
produce a lot of data (depending on how many keys are extracted from
each row), and the compression is very effective and efficient.
If the number of different keys is high, the first worker pass may not
benefit from the compression very much - the data will be spilled to
disk before the TID lists can grow long enough for the compression to
actualy help. In the second pass the impact is much more significant.
For real-world data (full-text on mailing list archives), I usually see
the compression to save only about ~15% in the first pass, but ~50% on
the second pass.
---
src/backend/access/gin/gininsert.c | 116 +++++++++++++++++++++++------
src/tools/pgindent/typedefs.list | 1 +
2 files changed, 95 insertions(+), 22 deletions(-)
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index b2b44066329..b84fb8f12b6 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -187,7 +187,9 @@ static void _gin_parallel_scan_and_build(GinBuildState *buildstate,
Relation heap, Relation index,
int sortmem, bool progress);
-static Datum _gin_parse_tuple(GinTuple *a, ItemPointerData **items);
+static ItemPointer _gin_parse_tuple_items(GinTuple *a);
+static Datum _gin_parse_tuple_key(GinTuple *a);
+
static GinTuple *_gin_build_tuple(OffsetNumber attrnum, unsigned char category,
Datum key, int16 typlen, bool typbyval,
ItemPointerData *items, uint32 nitems,
@@ -1265,7 +1267,8 @@ GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
AssertCheckGinBuffer(buffer);
- key = _gin_parse_tuple(tup, &items);
+ key = _gin_parse_tuple_key(tup);
+ items = _gin_parse_tuple_items(tup);
/* if the buffer is empty, set the fields (and copy the key) */
if (GinBufferIsEmpty(buffer))
@@ -1306,6 +1309,9 @@ GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
AssertCheckItemPointers(buffer, true);
}
+
+ /* free the decompressed TID list */
+ pfree(items);
}
/* XXX probably would be better to have a memory context for the buffer */
@@ -1806,6 +1812,15 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
table_close(heapRel, heapLockmode);
}
+/*
+ * Used to keep track of compressed TID lists when building a GIN tuple.
+ */
+typedef struct
+{
+ dlist_node node; /* linked list pointers */
+ GinPostingList *seg;
+} GinSegmentInfo;
+
/*
* _gin_build_tuple
* Serialize the state for an index key into a tuple for tuplesort.
@@ -1818,6 +1833,11 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
* like endianess etc. We could make it a little bit smaller, but it's not
* worth it - it's a tiny fraction of the data, and we need to MAXALIGN the
* start of the TID list anyway. So we wouldn't save anything.
+ *
+ * The TID list is serialized as compressed - it's highly compressible, and
+ * we already have ginCompressPostingList for this purpose. The list may be
+ * pretty long, so we compress it into multiple segments and then copy all
+ * of that into the GIN tuple.
*/
static GinTuple *
_gin_build_tuple(OffsetNumber attrnum, unsigned char category,
@@ -1831,6 +1851,11 @@ _gin_build_tuple(OffsetNumber attrnum, unsigned char category,
Size tuplen;
int keylen;
+ dlist_mutable_iter iter;
+ dlist_head segments;
+ int ncompressed;
+ Size compresslen;
+
/*
* Calculate how long is the key value. Only keys with GIN_CAT_NORM_KEY
* have actual non-empty key. We include varlena headers and \0 bytes for
@@ -1854,12 +1879,34 @@ _gin_build_tuple(OffsetNumber attrnum, unsigned char category,
else
elog(ERROR, "invalid typlen");
+ /* compress the item pointers */
+ ncompressed = 0;
+ compresslen = 0;
+ dlist_init(&segments);
+
+ /* generate compressed segments of TID list chunks */
+ while (ncompressed < nitems)
+ {
+ int cnt;
+ GinSegmentInfo *seginfo = palloc(sizeof(GinSegmentInfo));
+
+ seginfo->seg = ginCompressPostingList(&items[ncompressed],
+ (nitems - ncompressed),
+ UINT16_MAX,
+ &cnt);
+
+ ncompressed += cnt;
+ compresslen += SizeOfGinPostingList(seginfo->seg);
+
+ dlist_push_tail(&segments, &seginfo->node);
+ }
+
/*
* Determine GIN tuple length with all the data included. Be careful about
- * alignment, to allow direct access to item pointers.
+ * alignment, to allow direct access to compressed segments (those require
+ * SHORTALIGN, but we do MAXALING anyway).
*/
- tuplen = MAXALIGN(offsetof(GinTuple, data) + keylen) +
- (sizeof(ItemPointerData) * nitems);
+ tuplen = MAXALIGN(offsetof(GinTuple, data) + keylen) + compresslen;
*len = tuplen;
@@ -1909,37 +1956,40 @@ _gin_build_tuple(OffsetNumber attrnum, unsigned char category,
/* finally, copy the TIDs into the array */
ptr = (char *) tuple + MAXALIGN(offsetof(GinTuple, data) + keylen);
- memcpy(ptr, items, sizeof(ItemPointerData) * nitems);
+ /* copy in the compressed data, and free the segments */
+ dlist_foreach_modify(iter, &segments)
+ {
+ GinSegmentInfo *seginfo = dlist_container(GinSegmentInfo, node, iter.cur);
+
+ memcpy(ptr, seginfo->seg, SizeOfGinPostingList(seginfo->seg));
+
+ ptr += SizeOfGinPostingList(seginfo->seg);
+
+ dlist_delete(&seginfo->node);
+
+ pfree(seginfo->seg);
+ pfree(seginfo);
+ }
return tuple;
}
/*
- * _gin_parse_tuple
- * Deserialize the tuple from the tuplestore representation.
+ * _gin_parse_tuple_key
+ * Return a Datum representing the key stored in the tuple.
*
- * Most of the fields are actually directly accessible, the only thing that
+ * Most of the tuple fields are directly accessible, the only thing that
* needs more care is the key and the TID list.
*
* For the key, this returns a regular Datum representing it. It's either the
* actual key value, or a pointer to the beginning of the data array (which is
* where the data was copied by _gin_build_tuple).
- *
- * The pointer to the TID list is returned through 'items' (which is simply
- * a pointer to the data array).
*/
static Datum
-_gin_parse_tuple(GinTuple *a, ItemPointerData **items)
+_gin_parse_tuple_key(GinTuple *a)
{
Datum key;
- if (items)
- {
- char *ptr = (char *) a + MAXALIGN(offsetof(GinTuple, data) + a->keylen);
-
- *items = (ItemPointerData *) ptr;
- }
-
if (a->category != GIN_CAT_NORM_KEY)
return (Datum) 0;
@@ -1952,6 +2002,28 @@ _gin_parse_tuple(GinTuple *a, ItemPointerData **items)
return PointerGetDatum(a->data);
}
+/*
+* _gin_parse_tuple_items
+ * Return a pointer to a palloc'd array of decompressed TID array.
+ */
+static ItemPointer
+_gin_parse_tuple_items(GinTuple *a)
+{
+ int len;
+ char *ptr;
+ int ndecoded;
+ ItemPointer items;
+
+ len = a->tuplen - MAXALIGN(offsetof(GinTuple, data) + a->keylen);
+ ptr = (char *) a + MAXALIGN(offsetof(GinTuple, data) + a->keylen);
+
+ items = ginPostingListDecodeAllSegments((GinPostingList *) ptr, len, &ndecoded);
+
+ Assert(ndecoded == a->nitems);
+
+ return (ItemPointer) items;
+}
+
/*
* _gin_compare_tuples
* Compare GIN tuples, used by tuplesort during parallel index build.
@@ -1992,8 +2064,8 @@ _gin_compare_tuples(GinTuple *a, GinTuple *b)
if ((a->category == GIN_CAT_NORM_KEY) &&
(b->category == GIN_CAT_NORM_KEY))
{
- keya = _gin_parse_tuple(a, NULL);
- keyb = _gin_parse_tuple(b, NULL);
+ keya = _gin_parse_tuple_key(a);
+ keyb = _gin_parse_tuple_key(b);
/*
* works for both byval and byref types with fixed lenght, because for
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 9769f4d6b09..6b67756ebb9 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1023,6 +1023,7 @@ GinScanEntry
GinScanKey
GinScanOpaque
GinScanOpaqueData
+GinSegmentInfo
GinShared
GinState
GinStatsData
--
2.44.0
[text/x-patch] v20240505-0005-Collect-and-print-compression-stats.patch (5.1K, ../[email protected]/6-v20240505-0005-Collect-and-print-compression-stats.patch)
download | inline diff:
From eaefe4ed07054fd43428f06de35496aecbbadb4a Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:43 +0200
Subject: [PATCH v20240505 5/8] Collect and print compression stats
Allows evaluating the benefits of compressing the TID lists.
---
src/backend/access/gin/gininsert.c | 36 +++++++++++++++++++++++++-----
src/include/access/gin.h | 2 ++
2 files changed, 32 insertions(+), 6 deletions(-)
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index b84fb8f12b6..2206c47dfb1 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -190,7 +190,8 @@ static void _gin_parallel_scan_and_build(GinBuildState *buildstate,
static ItemPointer _gin_parse_tuple_items(GinTuple *a);
static Datum _gin_parse_tuple_key(GinTuple *a);
-static GinTuple *_gin_build_tuple(OffsetNumber attrnum, unsigned char category,
+static GinTuple *_gin_build_tuple(GinBuildState *state,
+ OffsetNumber attrnum, unsigned char category,
Datum key, int16 typlen, bool typbyval,
ItemPointerData *items, uint32 nitems,
Size *len);
@@ -539,7 +540,7 @@ ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values,
/* there could be many entries, so be willing to abort here */
CHECK_FOR_INTERRUPTS();
- tup = _gin_build_tuple(attnum, category,
+ tup = _gin_build_tuple(buildstate, attnum, category,
key, attr->attlen, attr->attbyval,
list, nlist, &tuplen);
@@ -1530,6 +1531,15 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
/* sort the raw per-worker data */
tuplesort_performsort(state->bs_worker_sort);
+ /* print some basic info */
+ elog(LOG, "_gin_parallel_scan_and_build raw %zu compressed %zu ratio %.2f%%",
+ state->buildStats.sizeRaw, state->buildStats.sizeCompressed,
+ (100.0 * state->buildStats.sizeCompressed) / state->buildStats.sizeRaw);
+
+ /* reset before the second phase */
+ state->buildStats.sizeCompressed = 0;
+ state->buildStats.sizeRaw = 0;
+
/*
* Read the GIN tuples from the shared tuplesort, sorted by the key, and
* merge them into larger chunks for the leader to combine.
@@ -1556,7 +1566,7 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
*/
AssertCheckItemPointers(buffer, true);
- ntup = _gin_build_tuple(buffer->attnum, buffer->category,
+ ntup = _gin_build_tuple(state, buffer->attnum, buffer->category,
buffer->key, buffer->typlen, buffer->typbyval,
buffer->items, buffer->nitems, &ntuplen);
@@ -1583,7 +1593,7 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
AssertCheckItemPointers(buffer, true);
- ntup = _gin_build_tuple(buffer->attnum, buffer->category,
+ ntup = _gin_build_tuple(state, buffer->attnum, buffer->category,
buffer->key, buffer->typlen, buffer->typbyval,
buffer->items, buffer->nitems, &ntuplen);
@@ -1598,6 +1608,11 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
/* relase all the memory */
GinBufferFree(buffer);
+ /* print some basic info */
+ elog(LOG, "_gin_process_worker_data raw %zu compressed %zu ratio %.2f%%",
+ state->buildStats.sizeRaw, state->buildStats.sizeCompressed,
+ (100.0 * state->buildStats.sizeCompressed) / state->buildStats.sizeRaw);
+
tuplesort_end(worker_sort);
}
@@ -1669,7 +1684,7 @@ _gin_parallel_scan_and_build(GinBuildState *state,
/* there could be many entries, so be willing to abort here */
CHECK_FOR_INTERRUPTS();
- tup = _gin_build_tuple(attnum, category,
+ tup = _gin_build_tuple(state, attnum, category,
key, attr->attlen, attr->attbyval,
list, nlist, &len);
@@ -1763,6 +1778,7 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
/* initialize the GIN build state */
initGinState(&buildstate.ginstate, indexRel);
buildstate.indtuples = 0;
+ /* XXX shouldn't this initialize the other fiedls, like ginbuild()? */
memset(&buildstate.buildStats, 0, sizeof(GinStatsData));
/*
@@ -1840,7 +1856,8 @@ typedef struct
* of that into the GIN tuple.
*/
static GinTuple *
-_gin_build_tuple(OffsetNumber attrnum, unsigned char category,
+_gin_build_tuple(GinBuildState *state,
+ OffsetNumber attrnum, unsigned char category,
Datum key, int16 typlen, bool typbyval,
ItemPointerData *items, uint32 nitems,
Size *len)
@@ -1971,6 +1988,13 @@ _gin_build_tuple(OffsetNumber attrnum, unsigned char category,
pfree(seginfo);
}
+ /* how large would the tuple be without compression? */
+ state->buildStats.sizeRaw += MAXALIGN(offsetof(GinTuple, data) + keylen) +
+ nitems * sizeof(ItemPointerData);
+
+ /* compressed size */
+ state->buildStats.sizeCompressed += tuplen;
+
return tuple;
}
diff --git a/src/include/access/gin.h b/src/include/access/gin.h
index be76d8446f4..2b6633d068a 100644
--- a/src/include/access/gin.h
+++ b/src/include/access/gin.h
@@ -49,6 +49,8 @@ typedef struct GinStatsData
BlockNumber nDataPages;
int64 nEntries;
int32 ginVersion;
+ Size sizeRaw;
+ Size sizeCompressed;
} GinStatsData;
/*
--
2.44.0
[text/x-patch] v20240505-0006-Enforce-memory-limit-when-combining-tuples.patch (14.8K, ../[email protected]/7-v20240505-0006-Enforce-memory-limit-when-combining-tuples.patch)
download | inline diff:
From fb14d8f86276dc08bec2e93c3191832613a6d56a Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:49 +0200
Subject: [PATCH v20240505 6/8] Enforce memory limit when combining tuples
When combinnig intermediate results during parallel GIN index build, we
want to restrict the memory usage. In ginBuildCallbackParallel() this is
done simply by dumping working state into tuplesort after hitting the
memory limit.
This commit introduces memory limit to the following steps, merging the
intermediate results in both worker and leader. The merge only deals
with one key at a time, and the primary risk is the key might have too
many different TIDs. This is not very likely, because the TID array only
needs 6B per item, it's a potential issue.
We can't simply dump the whole current TID list - the index requires the
TID values to be inserted in the correct order, but if the lists overlap
(as they do between workers), the tail of the list may change during the
mergesort. But thanks to sorting GIN tuples by first TID, we can derive
a safe TID horizon - we know no future tuples will have TIDs from before
this value, so it's safe to output this part of the list.
This commit tracks "frozen" part of the the TID list, which is the part
we know won't change after merging additional TID lists. Then if the TID
list grows too large (more than 64kB), we try to trim it - write out the
frozen part of the list, and discard it from the buffer. We only do the
trimming if there's at least 1024 frozen items - we don't want to write
the data into the index in tiny chunks.
The freezing also allows us to skip the frozen part during mergesort.
The frozen part of the list is known to be fully sorted, so we can just
skip it and mergesort only the rest of the data.
Note: These limites (1024 and 64kB) are mostly arbitrary - but seem high
enough to get good efficiency for compression/batching, but low enough
to release memory early and work in small increments.
---
src/backend/access/gin/gininsert.c | 245 +++++++++++++++++++++++++++--
src/include/access/gin.h | 1 +
2 files changed, 237 insertions(+), 9 deletions(-)
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index 2206c47dfb1..f4a4b8f00e9 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -1130,8 +1130,12 @@ typedef struct GinBuffer
int16 typlen;
bool typbyval;
+ /* Number of TIDs to collect before attempt to write some out. */
+ int maxitems;
+
/* array of TID values */
int nitems;
+ int nfrozen;
ItemPointerData *items;
} GinBuffer;
@@ -1166,7 +1170,21 @@ AssertCheckItemPointers(GinBuffer *buffer, bool sorted)
static GinBuffer *
GinBufferInit(void)
{
- return palloc0(sizeof(GinBuffer));
+ GinBuffer *buffer = (GinBuffer *) palloc0(sizeof(GinBuffer));
+
+ /*
+ * How many items can we fit into the memory limit? 64kB seems more than
+ * enough and we don't want a limit that's too high. OTOH maybe this
+ * should be tied to maintenance_work_mem or something like that?
+ *
+ * XXX This is not enough to prevent repeated merges after a wraparound,
+ * but it should be enough to make the merges cheap because it quickly
+ * finds reaches the end of the second list and can just memcpy the rest
+ * without walking it item by item.
+ */
+ buffer->maxitems = (64 * 1024L) / sizeof(ItemPointerData);
+
+ return buffer;
}
static bool
@@ -1221,6 +1239,54 @@ GinBufferKeyEquals(GinBuffer *buffer, GinTuple *tup)
return (memcmp(tup->data, DatumGetPointer(buffer->key), buffer->keylen) == 0);
}
+/*
+ * GinBufferShouldTrim
+ * Should we trim the list of item pointers?
+ *
+ * By trimming we understand writing out and removing the tuple IDs that
+ * we know can't change by future merges. We can deduce the TID up to which
+ * this is guaranteed from the "first" TID in each GIN tuple, which provides
+ * a "horizon" (for a given key) thanks to the sort.
+ *
+ * We don't want to do this too often - compressing longer TID lists is more
+ * efficient. But we also don't want to accumulate too many TIDs, for two
+ * reasons. First, it consumes memory and we might exceed maintenance_work_mem
+ * (or whatever limit applies), even if that's unlikely because TIDs are very
+ * small so we can fit a lot of them. Second, and more importantly, long TID
+ * lists are an issue if the scan wraps around, because a key may get a very
+ * wide list (with min/max TID for that key), forcing "full" mergesorts for
+ * every list merged into it (instead of the efficient append).
+ *
+ * So we look at two things when deciding if to trim - if the resulting list
+ * (after adding TIDs from the new tuple) would be too long, and if there is
+ * enough TIDs to trim (with values less than "first" TID from the new tuple),
+ * we do the trim. By enough we mean at least 128 TIDs (mostly an arbitrary
+ * number).
+ *
+ * XXX This does help for the wrap around case, because the "wide" TID list
+ * is essentially two ranges - one at the beginning of the table, one at the
+ * end. And all the other ranges (from GIN tuples) come in between, and also
+ * do not overlap. So by trimming up to the range we're about to add, this
+ * guarantees we'll be able to "concatenate" the two lists cheaply.
+ */
+static bool
+GinBufferShouldTrim(GinBuffer *buffer, GinTuple *tup)
+{
+ /* not enough TIDs to trim (1024 is somewhat arbitrary number) */
+ if (buffer->nfrozen < 1024)
+ return false;
+
+ /* We're not to hit the memory limit after adding this tuple. */
+ if ((buffer->nitems + tup->nitems) < buffer->maxitems)
+ return false;
+
+ /*
+ * OK, we have enough frozen TIDs to flush, and we have hit the memory
+ * limit, so it's time to write it out.
+ */
+ return true;
+}
+
/*
* GinBufferStoreTuple
* Add data from a GinTuple into the GinBuffer.
@@ -1259,6 +1325,11 @@ GinBufferKeyEquals(GinBuffer *buffer, GinTuple *tup)
* once, so there's going to be only such wide list, and it'll be sorted
* first (because it has the lowest TID for the key). So we'd do this at
* most once per key.
+ *
+ * XXX Maybe we could/should allocate the buffer once and then keep it
+ * without palloc/pfree. That won't help when just calling the mergesort,
+ * as that does palloc internally, but if we detected the append case,
+ * we could do without it. Not sure how much overhead it is, though.
*/
static void
GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
@@ -1287,26 +1358,82 @@ GinBufferStoreTuple(GinBuffer *buffer, GinTuple *tup)
buffer->key = (Datum) 0;
}
+ /*
+ * Try freeze TIDs at the beginning of the list, i.e. exclude them from
+ * the mergesort. We can do that with TIDs before the first TID in the new
+ * tuple we're about to add into the buffer.
+ *
+ * We do this incrementally when adding data into the in-memory buffer,
+ * and not later (e.g. when hitting a memory limit), because it allows us
+ * to skip the frozen data during the mergesort, making it cheaper.
+ */
+
+ /*
+ * Check if the last TID in the current list is frozen. This is the case
+ * when merging non-overlapping lists, e.g. in each parallel worker.
+ */
+ if ((buffer->nitems > 0) &&
+ (ItemPointerCompare(&buffer->items[buffer->nitems - 1], &tup->first) == 0))
+ buffer->nfrozen = buffer->nitems;
+
+ /*
+ * Now search the list linearly, to find the last frozen TID. If we found
+ * the whole list is frozen, this just does nothing.
+ *
+ * Start with the first not-yet-frozen tuple, and walk until we find the
+ * first TID that's higher.
+ *
+ * XXX Maybe this should do a binary search if the number of "non-frozen"
+ * items is sufficiently high (enough to make linear search slower than
+ * binsearch).
+ */
+ for (int i = buffer->nfrozen; i < buffer->nitems; i++)
+ {
+ /* Is the TID after the first TID of the new tuple? Can't freeze. */
+ if (ItemPointerCompare(&buffer->items[i], &tup->first) > 0)
+ break;
+
+ buffer->nfrozen++;
+ }
+
/*
* Copy the new TIDs into the buffer, combine with existing data (if any)
* using merge-sort. The mergesort is already smart about cases where it
* can simply concatenate the two lists, and when it actually needs to
* merge the data in an expensive way.
+ *
+ * XXX We could check if (buffer->nitems > buffer->nfrozen) and only do
+ * the mergesort in that case. ginMergeItemPointers does some palloc
+ * internally, and this way we could eliminate that. But let's keep the
+ * code simple for now.
*/
{
int nnew;
ItemPointer new;
- new = ginMergeItemPointers(buffer->items, buffer->nitems,
+ /*
+ * Resize the array - we do this first, because we'll dereference the
+ * first unfrozen TID, which would fail if the array is NULL. We'll
+ * still pass 0 as number of elements in that array though.
+ */
+ if (buffer->items == NULL)
+ buffer->items = palloc((buffer->nitems + tup->nitems) * sizeof(ItemPointerData));
+ else
+ buffer->items = repalloc(buffer->items,
+ (buffer->nitems + tup->nitems) * sizeof(ItemPointerData));
+
+ new = ginMergeItemPointers(&buffer->items[buffer->nfrozen], /* first unfronzen */
+ (buffer->nitems - buffer->nfrozen), /* num of unfrozen */
items, tup->nitems, &nnew);
- Assert(nnew == buffer->nitems + tup->nitems);
+ Assert(nnew == (tup->nitems + (buffer->nitems - buffer->nfrozen)));
+
+ memcpy(&buffer->items[buffer->nfrozen], new,
+ nnew * sizeof(ItemPointerData));
- if (buffer->items)
- pfree(buffer->items);
+ pfree(new);
- buffer->items = new;
- buffer->nitems = nnew;
+ buffer->nitems += tup->nitems;
AssertCheckItemPointers(buffer, true);
}
@@ -1332,6 +1459,7 @@ GinBufferReset(GinBuffer *buffer)
buffer->category = 0;
buffer->keylen = 0;
buffer->nitems = 0;
+ buffer->nfrozen = 0;
buffer->typlen = 0;
buffer->typbyval = 0;
@@ -1344,6 +1472,23 @@ GinBufferReset(GinBuffer *buffer)
/* XXX should do something with extremely large array of items? */
}
+/*
+ * GinBufferTrim
+ * Discard the "frozen" part of the TID list (which should have been
+ * written to disk/index before this call).
+ */
+static void
+GinBufferTrim(GinBuffer *buffer)
+{
+ Assert((buffer->nfrozen > 0) && (buffer->nfrozen <= buffer->nitems));
+
+ memmove(&buffer->items[0], &buffer->items[buffer->nfrozen],
+ sizeof(ItemPointerData) * (buffer->nitems - buffer->nfrozen));
+
+ buffer->nitems -= buffer->nfrozen;
+ buffer->nfrozen = 0;
+}
+
/* XXX probably would be better to have a memory context for the buffer */
static void
GinBufferFree(GinBuffer *buffer)
@@ -1402,7 +1547,12 @@ _gin_parallel_merge(GinBuildState *state)
/* do the actual sort in the leader */
tuplesort_performsort(state->bs_sortstate);
- /* initialize buffer to combine entries for the same key */
+ /*
+ * Initialize buffer to combine entries for the same key.
+ *
+ * The leader is allowed to use the whole maintenance_work_mem buffer to
+ * combine data. The parallel workers already completed.
+ */
buffer = GinBufferInit();
/*
@@ -1442,6 +1592,36 @@ _gin_parallel_merge(GinBuildState *state)
GinBufferReset(buffer);
}
+ /*
+ * We're about to add a GIN tuple to the buffer - check the memory
+ * limit first, and maybe write out some of the data into the index
+ * first, if needed (and possible). We only flush the part of the TID
+ * list that we know won't change, and only if there's enough data for
+ * compression to work well.
+ *
+ * XXX The buffer may also be empty, but in that case we skip this.
+ */
+ if (GinBufferShouldTrim(buffer, tup))
+ {
+ Assert(buffer->nfrozen > 0);
+
+ state->buildStats.nTrims++;
+
+ /*
+ * Buffer is not empty and it's storing a different key - flush
+ * the data into the insert, and start a new entry for current
+ * GinTuple.
+ */
+ AssertCheckItemPointers(buffer, true);
+
+ ginEntryInsert(&state->ginstate,
+ buffer->attnum, buffer->key, buffer->category,
+ buffer->items, buffer->nfrozen, &state->buildStats);
+
+ /* truncate the data we've just discarded */
+ GinBufferTrim(buffer);
+ }
+
/*
* Remember data for the current tuple (either remember the new key,
* or append if to the existing data).
@@ -1465,6 +1645,8 @@ _gin_parallel_merge(GinBuildState *state)
/* relase all the memory */
GinBufferFree(buffer);
+ elog(LOG, "_gin_parallel_merge ntrims " INT64_FORMAT, state->buildStats.nTrims);
+
tuplesort_end(state->bs_sortstate);
return reltuples;
@@ -1525,7 +1707,13 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
GinBuffer *buffer;
- /* initialize buffer to combine entries for the same key */
+ /*
+ * Initialize buffer to combine entries for the same key.
+ *
+ * The workers are limited to the same amount of memory as during the sort
+ * in ginBuildCallbackParallel. But this probably should be the 32MB used
+ * during planning, just like there.
+ */
buffer = GinBufferInit();
/* sort the raw per-worker data */
@@ -1578,6 +1766,43 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
GinBufferReset(buffer);
}
+ /*
+ * We're about to add a GIN tuple to the buffer - check the memory
+ * limit first, and maybe write out some of the data into the index
+ * first, if needed (and possible). We only flush the part of the TID
+ * list that we know won't change, and only if there's enough data for
+ * compression to work well.
+ *
+ * XXX The buffer may also be empty, but in that case we skip this.
+ */
+ if (GinBufferShouldTrim(buffer, tup))
+ {
+ GinTuple *ntup;
+ Size ntuplen;
+
+ Assert(buffer->nfrozen > 0);
+
+ state->buildStats.nTrims++;
+
+ /*
+ * Buffer is not empty and it's storing a different key - flush
+ * the data into the insert, and start a new entry for current
+ * GinTuple.
+ */
+ AssertCheckItemPointers(buffer, true);
+
+ ntup = _gin_build_tuple(state, buffer->attnum, buffer->category,
+ buffer->key, buffer->typlen, buffer->typbyval,
+ buffer->items, buffer->nfrozen, &ntuplen);
+
+ tuplesort_putgintuple(state->bs_sortstate, ntup, ntuplen);
+
+ pfree(ntup);
+
+ /* truncate the data we've just discarded */
+ GinBufferTrim(buffer);
+ }
+
/*
* Remember data for the current tuple (either remember the new key,
* or append if to the existing data).
@@ -1613,6 +1838,8 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort)
state->buildStats.sizeRaw, state->buildStats.sizeCompressed,
(100.0 * state->buildStats.sizeCompressed) / state->buildStats.sizeRaw);
+ elog(LOG, "_gin_process_worker_data trims " INT64_FORMAT, state->buildStats.nTrims);
+
tuplesort_end(worker_sort);
}
diff --git a/src/include/access/gin.h b/src/include/access/gin.h
index 2b6633d068a..9381329fac5 100644
--- a/src/include/access/gin.h
+++ b/src/include/access/gin.h
@@ -51,6 +51,7 @@ typedef struct GinStatsData
int32 ginVersion;
Size sizeRaw;
Size sizeCompressed;
+ int64 nTrims;
} GinStatsData;
/*
--
2.44.0
[text/x-patch] v20240505-0007-Detect-wrap-around-in-parallel-callback.patch (5.7K, ../[email protected]/8-v20240505-0007-Detect-wrap-around-in-parallel-callback.patch)
download | inline diff:
From f8af37eb2278d3a97b148458161c30122cdbf9e1 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 2 May 2024 15:21:55 +0200
Subject: [PATCH v20240505 7/8] Detect wrap around in parallel callback
When sync scan during index build wraps around, that may result in some
keys having very long TID lists, requiring "full" merge sort runs when
combining data in workers. It also causes problems with enforcing memory
limit, because we can't just dump the data - the index build requires
append-only posting lists, and violating may result in errors like
ERROR: could not split GIN page; all old items didn't fit
because after the scan wrap around some of the TIDs may belong to the
beginning of the list, affecting the compression.
But we can deal with this in the callback - if we see the TID to jump
back, that must mean a wraparound happened. In that case we simply dump
all the data accumulated in memory, and start from scratch.
This means there won't be any tuples with very wide TID ranges, instead
there'll be one tuple with a range at the end of the table, and another
tuple at the beginning. And all the lists in the worker will be
non-overlapping, and sort nicely based on first TID.
For the leader, we still need to do the full merge - the lists may
overlap and interleave in various ways. But there should be only very
few of those lists, about one per worker, making it not an issue.
---
src/backend/access/gin/gininsert.c | 89 ++++++++++++++++++------------
1 file changed, 55 insertions(+), 34 deletions(-)
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
index f4a4b8f00e9..7705eddfa70 100644
--- a/src/backend/access/gin/gininsert.c
+++ b/src/backend/access/gin/gininsert.c
@@ -142,6 +142,7 @@ typedef struct
MemoryContext tmpCtx;
MemoryContext funcCtx;
BuildAccumulator accum;
+ ItemPointerData tid;
/* FIXME likely duplicate with indtuples */
double bs_numtuples;
@@ -474,6 +475,47 @@ ginBuildCallback(Relation index, ItemPointer tid, Datum *values,
MemoryContextSwitchTo(oldCtx);
}
+static void
+ginFlushBuildState(GinBuildState *buildstate, Relation index)
+{
+ ItemPointerData *list;
+ Datum key;
+ GinNullCategory category;
+ uint32 nlist;
+ OffsetNumber attnum;
+ TupleDesc tdesc = RelationGetDescr(index);
+
+ ginBeginBAScan(&buildstate->accum);
+ while ((list = ginGetBAEntry(&buildstate->accum,
+ &attnum, &key, &category, &nlist)) != NULL)
+ {
+ /* information about the key */
+ Form_pg_attribute attr = TupleDescAttr(tdesc, (attnum - 1));
+
+ /* GIN tuple and tuple length */
+ GinTuple *tup;
+ Size tuplen;
+
+ /* there could be many entries, so be willing to abort here */
+ CHECK_FOR_INTERRUPTS();
+
+ tup = _gin_build_tuple(buildstate, attnum, category,
+ key, attr->attlen, attr->attbyval,
+ list, nlist, &tuplen);
+
+ tuplesort_putgintuple(buildstate->bs_worker_sort, tup, tuplen);
+
+ pfree(tup);
+ }
+
+ MemoryContextReset(buildstate->tmpCtx);
+ ginInitBA(&buildstate->accum);
+}
+
+/*
+ * FIXME Another way to deal with the wrap around of sync scans would be to
+ * detect when tid wraps around and just flush the state.
+ */
static void
ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values,
bool *isnull, bool tupleIsAlive, void *state)
@@ -484,6 +526,16 @@ ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values,
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
+ /* flush contents before wrapping around */
+ if (ItemPointerCompare(tid, &buildstate->tid) < 0)
+ {
+ elog(LOG, "calling ginFlushBuildState");
+ ginFlushBuildState(buildstate, index);
+ }
+
+ /* remember the TID we're about to process */
+ memcpy(&buildstate->tid, tid, sizeof(ItemPointerData));
+
for (i = 0; i < buildstate->ginstate.origTupdesc->natts; i++)
ginHeapTupleBulkInsert(buildstate, (OffsetNumber) (i + 1),
values[i], isnull[i], tid);
@@ -518,40 +570,7 @@ ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values,
* XXX probably should use 32MB, not work_mem, as used during planning?
*/
if (buildstate->accum.allocatedMemory >= (Size) work_mem * 1024L)
- {
- ItemPointerData *list;
- Datum key;
- GinNullCategory category;
- uint32 nlist;
- OffsetNumber attnum;
- TupleDesc tdesc = RelationGetDescr(index);
-
- ginBeginBAScan(&buildstate->accum);
- while ((list = ginGetBAEntry(&buildstate->accum,
- &attnum, &key, &category, &nlist)) != NULL)
- {
- /* information about the key */
- Form_pg_attribute attr = TupleDescAttr(tdesc, (attnum - 1));
-
- /* GIN tuple and tuple length */
- GinTuple *tup;
- Size tuplen;
-
- /* there could be many entries, so be willing to abort here */
- CHECK_FOR_INTERRUPTS();
-
- tup = _gin_build_tuple(buildstate, attnum, category,
- key, attr->attlen, attr->attbyval,
- list, nlist, &tuplen);
-
- tuplesort_putgintuple(buildstate->bs_worker_sort, tup, tuplen);
-
- pfree(tup);
- }
-
- MemoryContextReset(buildstate->tmpCtx);
- ginInitBA(&buildstate->accum);
- }
+ ginFlushBuildState(buildstate, index);
MemoryContextSwitchTo(oldCtx);
}
@@ -587,6 +606,7 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo)
buildstate.bs_numtuples = 0;
buildstate.bs_reltuples = 0;
buildstate.bs_leader = NULL;
+ memset(&buildstate.tid, 0, sizeof(ItemPointerData));
/* initialize the meta page */
MetaBuffer = GinNewBuffer(index);
@@ -2007,6 +2027,7 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
buildstate.indtuples = 0;
/* XXX shouldn't this initialize the other fiedls, like ginbuild()? */
memset(&buildstate.buildStats, 0, sizeof(GinStatsData));
+ memset(&buildstate.tid, 0, sizeof(ItemPointerData));
/*
* create a temporary memory context that is used to hold data not yet
--
2.44.0
view thread (7+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: Re: Parallel CREATE INDEX for GIN indexes
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox