mirror of
https://github.com/postgres/postgres.git
synced 2025-04-27 22:56:53 +03:00
790 lines
22 KiB
C
790 lines
22 KiB
C
/*-------------------------------------------------------------------------
|
|
*
|
|
* nodeGatherMerge.c
|
|
* Scan a plan in multiple workers, and do order-preserving merge.
|
|
*
|
|
* Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
|
|
* Portions Copyright (c) 1994, Regents of the University of California
|
|
*
|
|
* IDENTIFICATION
|
|
* src/backend/executor/nodeGatherMerge.c
|
|
*
|
|
*-------------------------------------------------------------------------
|
|
*/
|
|
|
|
#include "postgres.h"
|
|
|
|
#include "access/relscan.h"
|
|
#include "access/xact.h"
|
|
#include "executor/execdebug.h"
|
|
#include "executor/execParallel.h"
|
|
#include "executor/nodeGatherMerge.h"
|
|
#include "executor/nodeSubplan.h"
|
|
#include "executor/tqueue.h"
|
|
#include "lib/binaryheap.h"
|
|
#include "miscadmin.h"
|
|
#include "optimizer/optimizer.h"
|
|
#include "utils/memutils.h"
|
|
#include "utils/rel.h"
|
|
|
|
/*
|
|
* When we read tuples from workers, it's a good idea to read several at once
|
|
* for efficiency when possible: this minimizes context-switching overhead.
|
|
* But reading too many at a time wastes memory without improving performance.
|
|
* We'll read up to MAX_TUPLE_STORE tuples (in addition to the first one).
|
|
*/
|
|
#define MAX_TUPLE_STORE 10
|
|
|
|
/*
|
|
* Pending-tuple array for each worker. This holds additional tuples that
|
|
* we were able to fetch from the worker, but can't process yet. In addition,
|
|
* this struct holds the "done" flag indicating the worker is known to have
|
|
* no more tuples. (We do not use this struct for the leader; we don't keep
|
|
* any pending tuples for the leader, and the need_to_scan_locally flag serves
|
|
* as its "done" indicator.)
|
|
*/
|
|
typedef struct GMReaderTupleBuffer
|
|
{
|
|
MinimalTuple *tuple; /* array of length MAX_TUPLE_STORE */
|
|
int nTuples; /* number of tuples currently stored */
|
|
int readCounter; /* index of next tuple to extract */
|
|
bool done; /* true if reader is known exhausted */
|
|
} GMReaderTupleBuffer;
|
|
|
|
static TupleTableSlot *ExecGatherMerge(PlanState *pstate);
|
|
static int32 heap_compare_slots(Datum a, Datum b, void *arg);
|
|
static TupleTableSlot *gather_merge_getnext(GatherMergeState *gm_state);
|
|
static MinimalTuple gm_readnext_tuple(GatherMergeState *gm_state, int nreader,
|
|
bool nowait, bool *done);
|
|
static void ExecShutdownGatherMergeWorkers(GatherMergeState *node);
|
|
static void gather_merge_setup(GatherMergeState *gm_state);
|
|
static void gather_merge_init(GatherMergeState *gm_state);
|
|
static void gather_merge_clear_tuples(GatherMergeState *gm_state);
|
|
static bool gather_merge_readnext(GatherMergeState *gm_state, int reader,
|
|
bool nowait);
|
|
static void load_tuple_array(GatherMergeState *gm_state, int reader);
|
|
|
|
/* ----------------------------------------------------------------
|
|
* ExecInitGather
|
|
* ----------------------------------------------------------------
|
|
*/
|
|
GatherMergeState *
|
|
ExecInitGatherMerge(GatherMerge *node, EState *estate, int eflags)
|
|
{
|
|
GatherMergeState *gm_state;
|
|
Plan *outerNode;
|
|
TupleDesc tupDesc;
|
|
|
|
/* Gather merge node doesn't have innerPlan node. */
|
|
Assert(innerPlan(node) == NULL);
|
|
|
|
/*
|
|
* create state structure
|
|
*/
|
|
gm_state = makeNode(GatherMergeState);
|
|
gm_state->ps.plan = (Plan *) node;
|
|
gm_state->ps.state = estate;
|
|
gm_state->ps.ExecProcNode = ExecGatherMerge;
|
|
|
|
gm_state->initialized = false;
|
|
gm_state->gm_initialized = false;
|
|
gm_state->tuples_needed = -1;
|
|
|
|
/*
|
|
* Miscellaneous initialization
|
|
*
|
|
* create expression context for node
|
|
*/
|
|
ExecAssignExprContext(estate, &gm_state->ps);
|
|
|
|
/*
|
|
* GatherMerge doesn't support checking a qual (it's always more efficient
|
|
* to do it in the child node).
|
|
*/
|
|
Assert(!node->plan.qual);
|
|
|
|
/*
|
|
* now initialize outer plan
|
|
*/
|
|
outerNode = outerPlan(node);
|
|
outerPlanState(gm_state) = ExecInitNode(outerNode, estate, eflags);
|
|
|
|
/*
|
|
* Leader may access ExecProcNode result directly (if
|
|
* need_to_scan_locally), or from workers via tuple queue. So we can't
|
|
* trivially rely on the slot type being fixed for expressions evaluated
|
|
* within this node.
|
|
*/
|
|
gm_state->ps.outeropsset = true;
|
|
gm_state->ps.outeropsfixed = false;
|
|
|
|
/*
|
|
* Store the tuple descriptor into gather merge state, so we can use it
|
|
* while initializing the gather merge slots.
|
|
*/
|
|
tupDesc = ExecGetResultType(outerPlanState(gm_state));
|
|
gm_state->tupDesc = tupDesc;
|
|
|
|
/*
|
|
* Initialize result type and projection.
|
|
*/
|
|
ExecInitResultTypeTL(&gm_state->ps);
|
|
ExecConditionalAssignProjectionInfo(&gm_state->ps, tupDesc, OUTER_VAR);
|
|
|
|
/*
|
|
* Without projections result slot type is not trivially known, see
|
|
* comment above.
|
|
*/
|
|
if (gm_state->ps.ps_ProjInfo == NULL)
|
|
{
|
|
gm_state->ps.resultopsset = true;
|
|
gm_state->ps.resultopsfixed = false;
|
|
}
|
|
|
|
/*
|
|
* initialize sort-key information
|
|
*/
|
|
if (node->numCols)
|
|
{
|
|
int i;
|
|
|
|
gm_state->gm_nkeys = node->numCols;
|
|
gm_state->gm_sortkeys =
|
|
palloc0(sizeof(SortSupportData) * node->numCols);
|
|
|
|
for (i = 0; i < node->numCols; i++)
|
|
{
|
|
SortSupport sortKey = gm_state->gm_sortkeys + i;
|
|
|
|
sortKey->ssup_cxt = CurrentMemoryContext;
|
|
sortKey->ssup_collation = node->collations[i];
|
|
sortKey->ssup_nulls_first = node->nullsFirst[i];
|
|
sortKey->ssup_attno = node->sortColIdx[i];
|
|
|
|
/*
|
|
* We don't perform abbreviated key conversion here, for the same
|
|
* reasons that it isn't used in MergeAppend
|
|
*/
|
|
sortKey->abbreviate = false;
|
|
|
|
PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
|
|
}
|
|
}
|
|
|
|
/* Now allocate the workspace for gather merge */
|
|
gather_merge_setup(gm_state);
|
|
|
|
return gm_state;
|
|
}
|
|
|
|
/* ----------------------------------------------------------------
|
|
* ExecGatherMerge(node)
|
|
*
|
|
* Scans the relation via multiple workers and returns
|
|
* the next qualifying tuple.
|
|
* ----------------------------------------------------------------
|
|
*/
|
|
static TupleTableSlot *
|
|
ExecGatherMerge(PlanState *pstate)
|
|
{
|
|
GatherMergeState *node = castNode(GatherMergeState, pstate);
|
|
TupleTableSlot *slot;
|
|
ExprContext *econtext;
|
|
|
|
CHECK_FOR_INTERRUPTS();
|
|
|
|
/*
|
|
* As with Gather, we don't launch workers until this node is actually
|
|
* executed.
|
|
*/
|
|
if (!node->initialized)
|
|
{
|
|
EState *estate = node->ps.state;
|
|
GatherMerge *gm = castNode(GatherMerge, node->ps.plan);
|
|
|
|
/*
|
|
* Sometimes we might have to run without parallelism; but if parallel
|
|
* mode is active then we can try to fire up some workers.
|
|
*/
|
|
if (gm->num_workers > 0 && estate->es_use_parallel_mode)
|
|
{
|
|
ParallelContext *pcxt;
|
|
|
|
/* Initialize, or re-initialize, shared state needed by workers. */
|
|
if (!node->pei)
|
|
node->pei = ExecInitParallelPlan(outerPlanState(node),
|
|
estate,
|
|
gm->initParam,
|
|
gm->num_workers,
|
|
node->tuples_needed);
|
|
else
|
|
ExecParallelReinitialize(outerPlanState(node),
|
|
node->pei,
|
|
gm->initParam);
|
|
|
|
/* Try to launch workers. */
|
|
pcxt = node->pei->pcxt;
|
|
LaunchParallelWorkers(pcxt);
|
|
/* We save # workers launched for the benefit of EXPLAIN */
|
|
node->nworkers_launched = pcxt->nworkers_launched;
|
|
|
|
/* Set up tuple queue readers to read the results. */
|
|
if (pcxt->nworkers_launched > 0)
|
|
{
|
|
ExecParallelCreateReaders(node->pei);
|
|
/* Make a working array showing the active readers */
|
|
node->nreaders = pcxt->nworkers_launched;
|
|
node->reader = (TupleQueueReader **)
|
|
palloc(node->nreaders * sizeof(TupleQueueReader *));
|
|
memcpy(node->reader, node->pei->reader,
|
|
node->nreaders * sizeof(TupleQueueReader *));
|
|
}
|
|
else
|
|
{
|
|
/* No workers? Then never mind. */
|
|
node->nreaders = 0;
|
|
node->reader = NULL;
|
|
}
|
|
}
|
|
|
|
/* allow leader to participate if enabled or no choice */
|
|
if (parallel_leader_participation || node->nreaders == 0)
|
|
node->need_to_scan_locally = true;
|
|
node->initialized = true;
|
|
}
|
|
|
|
/*
|
|
* Reset per-tuple memory context to free any expression evaluation
|
|
* storage allocated in the previous tuple cycle.
|
|
*/
|
|
econtext = node->ps.ps_ExprContext;
|
|
ResetExprContext(econtext);
|
|
|
|
/*
|
|
* Get next tuple, either from one of our workers, or by running the plan
|
|
* ourselves.
|
|
*/
|
|
slot = gather_merge_getnext(node);
|
|
if (TupIsNull(slot))
|
|
return NULL;
|
|
|
|
/* If no projection is required, we're done. */
|
|
if (node->ps.ps_ProjInfo == NULL)
|
|
return slot;
|
|
|
|
/*
|
|
* Form the result tuple using ExecProject(), and return it.
|
|
*/
|
|
econtext->ecxt_outertuple = slot;
|
|
return ExecProject(node->ps.ps_ProjInfo);
|
|
}
|
|
|
|
/* ----------------------------------------------------------------
|
|
* ExecEndGatherMerge
|
|
*
|
|
* frees any storage allocated through C routines.
|
|
* ----------------------------------------------------------------
|
|
*/
|
|
void
|
|
ExecEndGatherMerge(GatherMergeState *node)
|
|
{
|
|
ExecEndNode(outerPlanState(node)); /* let children clean up first */
|
|
ExecShutdownGatherMerge(node);
|
|
ExecFreeExprContext(&node->ps);
|
|
if (node->ps.ps_ResultTupleSlot)
|
|
ExecClearTuple(node->ps.ps_ResultTupleSlot);
|
|
}
|
|
|
|
/* ----------------------------------------------------------------
|
|
* ExecShutdownGatherMerge
|
|
*
|
|
* Destroy the setup for parallel workers including parallel context.
|
|
* ----------------------------------------------------------------
|
|
*/
|
|
void
|
|
ExecShutdownGatherMerge(GatherMergeState *node)
|
|
{
|
|
ExecShutdownGatherMergeWorkers(node);
|
|
|
|
/* Now destroy the parallel context. */
|
|
if (node->pei != NULL)
|
|
{
|
|
ExecParallelCleanup(node->pei);
|
|
node->pei = NULL;
|
|
}
|
|
}
|
|
|
|
/* ----------------------------------------------------------------
|
|
* ExecShutdownGatherMergeWorkers
|
|
*
|
|
* Stop all the parallel workers.
|
|
* ----------------------------------------------------------------
|
|
*/
|
|
static void
|
|
ExecShutdownGatherMergeWorkers(GatherMergeState *node)
|
|
{
|
|
if (node->pei != NULL)
|
|
ExecParallelFinish(node->pei);
|
|
|
|
/* Flush local copy of reader array */
|
|
if (node->reader)
|
|
pfree(node->reader);
|
|
node->reader = NULL;
|
|
}
|
|
|
|
/* ----------------------------------------------------------------
|
|
* ExecReScanGatherMerge
|
|
*
|
|
* Prepare to re-scan the result of a GatherMerge.
|
|
* ----------------------------------------------------------------
|
|
*/
|
|
void
|
|
ExecReScanGatherMerge(GatherMergeState *node)
|
|
{
|
|
GatherMerge *gm = (GatherMerge *) node->ps.plan;
|
|
PlanState *outerPlan = outerPlanState(node);
|
|
|
|
/* Make sure any existing workers are gracefully shut down */
|
|
ExecShutdownGatherMergeWorkers(node);
|
|
|
|
/* Free any unused tuples, so we don't leak memory across rescans */
|
|
gather_merge_clear_tuples(node);
|
|
|
|
/* Mark node so that shared state will be rebuilt at next call */
|
|
node->initialized = false;
|
|
node->gm_initialized = false;
|
|
|
|
/*
|
|
* Set child node's chgParam to tell it that the next scan might deliver a
|
|
* different set of rows within the leader process. (The overall rowset
|
|
* shouldn't change, but the leader process's subset might; hence nodes
|
|
* between here and the parallel table scan node mustn't optimize on the
|
|
* assumption of an unchanging rowset.)
|
|
*/
|
|
if (gm->rescan_param >= 0)
|
|
outerPlan->chgParam = bms_add_member(outerPlan->chgParam,
|
|
gm->rescan_param);
|
|
|
|
/*
|
|
* If chgParam of subnode is not null then plan will be re-scanned by
|
|
* first ExecProcNode. Note: because this does nothing if we have a
|
|
* rescan_param, it's currently guaranteed that parallel-aware child nodes
|
|
* will not see a ReScan call until after they get a ReInitializeDSM call.
|
|
* That ordering might not be something to rely on, though. A good rule
|
|
* of thumb is that ReInitializeDSM should reset only shared state, ReScan
|
|
* should reset only local state, and anything that depends on both of
|
|
* those steps being finished must wait until the first ExecProcNode call.
|
|
*/
|
|
if (outerPlan->chgParam == NULL)
|
|
ExecReScan(outerPlan);
|
|
}
|
|
|
|
/*
|
|
* Set up the data structures that we'll need for Gather Merge.
|
|
*
|
|
* We allocate these once on the basis of gm->num_workers, which is an
|
|
* upper bound for the number of workers we'll actually have. During
|
|
* a rescan, we reset the structures to empty. This approach simplifies
|
|
* not leaking memory across rescans.
|
|
*
|
|
* In the gm_slots[] array, index 0 is for the leader, and indexes 1 to n
|
|
* are for workers. The values placed into gm_heap correspond to indexes
|
|
* in gm_slots[]. The gm_tuple_buffers[] array, however, is indexed from
|
|
* 0 to n-1; it has no entry for the leader.
|
|
*/
|
|
static void
|
|
gather_merge_setup(GatherMergeState *gm_state)
|
|
{
|
|
GatherMerge *gm = castNode(GatherMerge, gm_state->ps.plan);
|
|
int nreaders = gm->num_workers;
|
|
int i;
|
|
|
|
/*
|
|
* Allocate gm_slots for the number of workers + one more slot for leader.
|
|
* Slot 0 is always for the leader. Leader always calls ExecProcNode() to
|
|
* read the tuple, and then stores it directly into its gm_slots entry.
|
|
* For other slots, code below will call ExecInitExtraTupleSlot() to
|
|
* create a slot for the worker's results. Note that during any single
|
|
* scan, we might have fewer than num_workers available workers, in which
|
|
* case the extra array entries go unused.
|
|
*/
|
|
gm_state->gm_slots = (TupleTableSlot **)
|
|
palloc0((nreaders + 1) * sizeof(TupleTableSlot *));
|
|
|
|
/* Allocate the tuple slot and tuple array for each worker */
|
|
gm_state->gm_tuple_buffers = (GMReaderTupleBuffer *)
|
|
palloc0(nreaders * sizeof(GMReaderTupleBuffer));
|
|
|
|
for (i = 0; i < nreaders; i++)
|
|
{
|
|
/* Allocate the tuple array with length MAX_TUPLE_STORE */
|
|
gm_state->gm_tuple_buffers[i].tuple =
|
|
(MinimalTuple *) palloc0(sizeof(MinimalTuple) * MAX_TUPLE_STORE);
|
|
|
|
/* Initialize tuple slot for worker */
|
|
gm_state->gm_slots[i + 1] =
|
|
ExecInitExtraTupleSlot(gm_state->ps.state, gm_state->tupDesc,
|
|
&TTSOpsMinimalTuple);
|
|
}
|
|
|
|
/* Allocate the resources for the merge */
|
|
gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
|
|
heap_compare_slots,
|
|
gm_state);
|
|
}
|
|
|
|
/*
|
|
* Initialize the Gather Merge.
|
|
*
|
|
* Reset data structures to ensure they're empty. Then pull at least one
|
|
* tuple from leader + each worker (or set its "done" indicator), and set up
|
|
* the heap.
|
|
*/
|
|
static void
|
|
gather_merge_init(GatherMergeState *gm_state)
|
|
{
|
|
int nreaders = gm_state->nreaders;
|
|
bool nowait = true;
|
|
int i;
|
|
|
|
/* Assert that gather_merge_setup made enough space */
|
|
Assert(nreaders <= castNode(GatherMerge, gm_state->ps.plan)->num_workers);
|
|
|
|
/* Reset leader's tuple slot to empty */
|
|
gm_state->gm_slots[0] = NULL;
|
|
|
|
/* Reset the tuple slot and tuple array for each worker */
|
|
for (i = 0; i < nreaders; i++)
|
|
{
|
|
/* Reset tuple array to empty */
|
|
gm_state->gm_tuple_buffers[i].nTuples = 0;
|
|
gm_state->gm_tuple_buffers[i].readCounter = 0;
|
|
/* Reset done flag to not-done */
|
|
gm_state->gm_tuple_buffers[i].done = false;
|
|
/* Ensure output slot is empty */
|
|
ExecClearTuple(gm_state->gm_slots[i + 1]);
|
|
}
|
|
|
|
/* Reset binary heap to empty */
|
|
binaryheap_reset(gm_state->gm_heap);
|
|
|
|
/*
|
|
* First, try to read a tuple from each worker (including leader) in
|
|
* nowait mode. After this, if not all workers were able to produce a
|
|
* tuple (or a "done" indication), then re-read from remaining workers,
|
|
* this time using wait mode. Add all live readers (those producing at
|
|
* least one tuple) to the heap.
|
|
*/
|
|
reread:
|
|
for (i = 0; i <= nreaders; i++)
|
|
{
|
|
CHECK_FOR_INTERRUPTS();
|
|
|
|
/* skip this source if already known done */
|
|
if ((i == 0) ? gm_state->need_to_scan_locally :
|
|
!gm_state->gm_tuple_buffers[i - 1].done)
|
|
{
|
|
if (TupIsNull(gm_state->gm_slots[i]))
|
|
{
|
|
/* Don't have a tuple yet, try to get one */
|
|
if (gather_merge_readnext(gm_state, i, nowait))
|
|
binaryheap_add_unordered(gm_state->gm_heap,
|
|
Int32GetDatum(i));
|
|
}
|
|
else
|
|
{
|
|
/*
|
|
* We already got at least one tuple from this worker, but
|
|
* might as well see if it has any more ready by now.
|
|
*/
|
|
load_tuple_array(gm_state, i);
|
|
}
|
|
}
|
|
}
|
|
|
|
/* need not recheck leader, since nowait doesn't matter for it */
|
|
for (i = 1; i <= nreaders; i++)
|
|
{
|
|
if (!gm_state->gm_tuple_buffers[i - 1].done &&
|
|
TupIsNull(gm_state->gm_slots[i]))
|
|
{
|
|
nowait = false;
|
|
goto reread;
|
|
}
|
|
}
|
|
|
|
/* Now heapify the heap. */
|
|
binaryheap_build(gm_state->gm_heap);
|
|
|
|
gm_state->gm_initialized = true;
|
|
}
|
|
|
|
/*
|
|
* Clear out the tuple table slot, and any unused pending tuples,
|
|
* for each gather merge input.
|
|
*/
|
|
static void
|
|
gather_merge_clear_tuples(GatherMergeState *gm_state)
|
|
{
|
|
int i;
|
|
|
|
for (i = 0; i < gm_state->nreaders; i++)
|
|
{
|
|
GMReaderTupleBuffer *tuple_buffer = &gm_state->gm_tuple_buffers[i];
|
|
|
|
while (tuple_buffer->readCounter < tuple_buffer->nTuples)
|
|
pfree(tuple_buffer->tuple[tuple_buffer->readCounter++]);
|
|
|
|
ExecClearTuple(gm_state->gm_slots[i + 1]);
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Read the next tuple for gather merge.
|
|
*
|
|
* Fetch the sorted tuple out of the heap.
|
|
*/
|
|
static TupleTableSlot *
|
|
gather_merge_getnext(GatherMergeState *gm_state)
|
|
{
|
|
int i;
|
|
|
|
if (!gm_state->gm_initialized)
|
|
{
|
|
/*
|
|
* First time through: pull the first tuple from each participant, and
|
|
* set up the heap.
|
|
*/
|
|
gather_merge_init(gm_state);
|
|
}
|
|
else
|
|
{
|
|
/*
|
|
* Otherwise, pull the next tuple from whichever participant we
|
|
* returned from last time, and reinsert that participant's index into
|
|
* the heap, because it might now compare differently against the
|
|
* other elements of the heap.
|
|
*/
|
|
i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
|
|
|
|
if (gather_merge_readnext(gm_state, i, false))
|
|
binaryheap_replace_first(gm_state->gm_heap, Int32GetDatum(i));
|
|
else
|
|
{
|
|
/* reader exhausted, remove it from heap */
|
|
(void) binaryheap_remove_first(gm_state->gm_heap);
|
|
}
|
|
}
|
|
|
|
if (binaryheap_empty(gm_state->gm_heap))
|
|
{
|
|
/* All the queues are exhausted, and so is the heap */
|
|
gather_merge_clear_tuples(gm_state);
|
|
return NULL;
|
|
}
|
|
else
|
|
{
|
|
/* Return next tuple from whichever participant has the leading one */
|
|
i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
|
|
return gm_state->gm_slots[i];
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Read tuple(s) for given reader in nowait mode, and load into its tuple
|
|
* array, until we have MAX_TUPLE_STORE of them or would have to block.
|
|
*/
|
|
static void
|
|
load_tuple_array(GatherMergeState *gm_state, int reader)
|
|
{
|
|
GMReaderTupleBuffer *tuple_buffer;
|
|
int i;
|
|
|
|
/* Don't do anything if this is the leader. */
|
|
if (reader == 0)
|
|
return;
|
|
|
|
tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
|
|
|
|
/* If there's nothing in the array, reset the counters to zero. */
|
|
if (tuple_buffer->nTuples == tuple_buffer->readCounter)
|
|
tuple_buffer->nTuples = tuple_buffer->readCounter = 0;
|
|
|
|
/* Try to fill additional slots in the array. */
|
|
for (i = tuple_buffer->nTuples; i < MAX_TUPLE_STORE; i++)
|
|
{
|
|
MinimalTuple tuple;
|
|
|
|
tuple = gm_readnext_tuple(gm_state,
|
|
reader,
|
|
true,
|
|
&tuple_buffer->done);
|
|
if (!tuple)
|
|
break;
|
|
tuple_buffer->tuple[i] = tuple;
|
|
tuple_buffer->nTuples++;
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Store the next tuple for a given reader into the appropriate slot.
|
|
*
|
|
* Returns true if successful, false if not (either reader is exhausted,
|
|
* or we didn't want to wait for a tuple). Sets done flag if reader
|
|
* is found to be exhausted.
|
|
*/
|
|
static bool
|
|
gather_merge_readnext(GatherMergeState *gm_state, int reader, bool nowait)
|
|
{
|
|
GMReaderTupleBuffer *tuple_buffer;
|
|
MinimalTuple tup;
|
|
|
|
/*
|
|
* If we're being asked to generate a tuple from the leader, then we just
|
|
* call ExecProcNode as normal to produce one.
|
|
*/
|
|
if (reader == 0)
|
|
{
|
|
if (gm_state->need_to_scan_locally)
|
|
{
|
|
PlanState *outerPlan = outerPlanState(gm_state);
|
|
TupleTableSlot *outerTupleSlot;
|
|
EState *estate = gm_state->ps.state;
|
|
|
|
/* Install our DSA area while executing the plan. */
|
|
estate->es_query_dsa = gm_state->pei ? gm_state->pei->area : NULL;
|
|
outerTupleSlot = ExecProcNode(outerPlan);
|
|
estate->es_query_dsa = NULL;
|
|
|
|
if (!TupIsNull(outerTupleSlot))
|
|
{
|
|
gm_state->gm_slots[0] = outerTupleSlot;
|
|
return true;
|
|
}
|
|
/* need_to_scan_locally serves as "done" flag for leader */
|
|
gm_state->need_to_scan_locally = false;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/* Otherwise, check the state of the relevant tuple buffer. */
|
|
tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
|
|
|
|
if (tuple_buffer->nTuples > tuple_buffer->readCounter)
|
|
{
|
|
/* Return any tuple previously read that is still buffered. */
|
|
tup = tuple_buffer->tuple[tuple_buffer->readCounter++];
|
|
}
|
|
else if (tuple_buffer->done)
|
|
{
|
|
/* Reader is known to be exhausted. */
|
|
return false;
|
|
}
|
|
else
|
|
{
|
|
/* Read and buffer next tuple. */
|
|
tup = gm_readnext_tuple(gm_state,
|
|
reader,
|
|
nowait,
|
|
&tuple_buffer->done);
|
|
if (!tup)
|
|
return false;
|
|
|
|
/*
|
|
* Attempt to read more tuples in nowait mode and store them in the
|
|
* pending-tuple array for the reader.
|
|
*/
|
|
load_tuple_array(gm_state, reader);
|
|
}
|
|
|
|
Assert(tup);
|
|
|
|
/* Build the TupleTableSlot for the given tuple */
|
|
ExecStoreMinimalTuple(tup, /* tuple to store */
|
|
gm_state->gm_slots[reader], /* slot in which to
|
|
* store the tuple */
|
|
true); /* pfree tuple when done with it */
|
|
|
|
return true;
|
|
}
|
|
|
|
/*
|
|
* Attempt to read a tuple from given worker.
|
|
*/
|
|
static MinimalTuple
|
|
gm_readnext_tuple(GatherMergeState *gm_state, int nreader, bool nowait,
|
|
bool *done)
|
|
{
|
|
TupleQueueReader *reader;
|
|
MinimalTuple tup;
|
|
|
|
/* Check for async events, particularly messages from workers. */
|
|
CHECK_FOR_INTERRUPTS();
|
|
|
|
/*
|
|
* Attempt to read a tuple.
|
|
*
|
|
* Note that TupleQueueReaderNext will just return NULL for a worker which
|
|
* fails to initialize. We'll treat that worker as having produced no
|
|
* tuples; WaitForParallelWorkersToFinish will error out when we get
|
|
* there.
|
|
*/
|
|
reader = gm_state->reader[nreader - 1];
|
|
tup = TupleQueueReaderNext(reader, nowait, done);
|
|
|
|
/*
|
|
* Since we'll be buffering these across multiple calls, we need to make a
|
|
* copy.
|
|
*/
|
|
return tup ? heap_copy_minimal_tuple(tup) : NULL;
|
|
}
|
|
|
|
/*
|
|
* We have one slot for each item in the heap array. We use SlotNumber
|
|
* to store slot indexes. This doesn't actually provide any formal
|
|
* type-safety, but it makes the code more self-documenting.
|
|
*/
|
|
typedef int32 SlotNumber;
|
|
|
|
/*
|
|
* Compare the tuples in the two given slots.
|
|
*/
|
|
static int32
|
|
heap_compare_slots(Datum a, Datum b, void *arg)
|
|
{
|
|
GatherMergeState *node = (GatherMergeState *) arg;
|
|
SlotNumber slot1 = DatumGetInt32(a);
|
|
SlotNumber slot2 = DatumGetInt32(b);
|
|
|
|
TupleTableSlot *s1 = node->gm_slots[slot1];
|
|
TupleTableSlot *s2 = node->gm_slots[slot2];
|
|
int nkey;
|
|
|
|
Assert(!TupIsNull(s1));
|
|
Assert(!TupIsNull(s2));
|
|
|
|
for (nkey = 0; nkey < node->gm_nkeys; nkey++)
|
|
{
|
|
SortSupport sortKey = node->gm_sortkeys + nkey;
|
|
AttrNumber attno = sortKey->ssup_attno;
|
|
Datum datum1,
|
|
datum2;
|
|
bool isNull1,
|
|
isNull2;
|
|
int compare;
|
|
|
|
datum1 = slot_getattr(s1, attno, &isNull1);
|
|
datum2 = slot_getattr(s2, attno, &isNull2);
|
|
|
|
compare = ApplySortComparator(datum1, isNull1,
|
|
datum2, isNull2,
|
|
sortKey);
|
|
if (compare != 0)
|
|
{
|
|
INVERT_COMPARE_RESULT(compare);
|
|
return compare;
|
|
}
|
|
}
|
|
return 0;
|
|
}
|