nodeGatherMerge.c File Reference

#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"

Include dependency graph for nodeGatherMerge.c:

Go to the source code of this file.

Data Structures
struct	GMReaderTupleBuffer

Macros
#define	MAX_TUPLE_STORE   10

Typedefs
typedef struct GMReaderTupleBuffer	GMReaderTupleBuffer
typedef int32	SlotNumber

Functions
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)
GatherMergeState *	ExecInitGatherMerge (GatherMerge *node, EState *estate, int eflags)
void	ExecEndGatherMerge (GatherMergeState *node)
void	ExecShutdownGatherMerge (GatherMergeState *node)
void	ExecReScanGatherMerge (GatherMergeState *node)

Macro Definition Documentation

MAX_TUPLE_STORE

#define MAX_TUPLE_STORE   10

Definition at line 36 of file nodeGatherMerge.c.

Typedef Documentation

GMReaderTupleBuffer

typedef struct GMReaderTupleBuffer GMReaderTupleBuffer

SlotNumber

typedef int32 SlotNumber

Definition at line 747 of file nodeGatherMerge.c.

Function Documentation

ExecEndGatherMerge()

void ExecEndGatherMerge

(

GatherMergeState *

node

)

Definition at line 289 of file nodeGatherMerge.c.

{
    ExecEndNode(outerPlanState(node));  /* let children clean up first */
    ExecShutdownGatherMerge(node);
    ExecFreeExprContext(&node->ps);
    if (node->ps.ps_ResultTupleSlot)
        ExecClearTuple(node->ps.ps_ResultTupleSlot);
}

References ExecClearTuple(), ExecEndNode(), ExecFreeExprContext(), ExecShutdownGatherMerge(), outerPlanState, GatherMergeState::ps, and PlanState::ps_ResultTupleSlot.

Referenced by ExecEndNode().

ExecGatherMerge()

static TupleTableSlot * ExecGatherMerge ( PlanState *  pstate )

static

Definition at line 188 of file nodeGatherMerge.c.

{
    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);
}

References castNode, CHECK_FOR_INTERRUPTS, ExprContext::ecxt_outertuple, EState::es_use_parallel_mode, ExecInitParallelPlan(), ExecParallelCreateReaders(), ExecParallelReinitialize(), ExecProject(), gather_merge_getnext(), GatherMergeState::initialized, GatherMerge::initParam, LaunchParallelWorkers(), GatherMergeState::need_to_scan_locally, GatherMergeState::nreaders, GatherMerge::num_workers, ParallelContext::nworkers_launched, GatherMergeState::nworkers_launched, outerPlanState, palloc(), parallel_leader_participation, ParallelExecutorInfo::pcxt, GatherMergeState::pei, PlanState::plan, GatherMergeState::ps, PlanState::ps_ExprContext, PlanState::ps_ProjInfo, ParallelExecutorInfo::reader, GatherMergeState::reader, ResetExprContext, PlanState::state, TupIsNull, and GatherMergeState::tuples_needed.

Referenced by ExecInitGatherMerge().

ExecInitGatherMerge()

GatherMergeState* ExecInitGatherMerge	(	GatherMerge *	node,
		EState *	estate,
		int	eflags
	)

Definition at line 72 of file nodeGatherMerge.c.

{
    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;
}

References SortSupportData::abbreviate, Assert(), CurrentMemoryContext, ExecAssignExprContext(), ExecConditionalAssignProjectionInfo(), ExecGatherMerge(), ExecGetResultType(), ExecInitNode(), ExecInitResultTypeTL(), PlanState::ExecProcNode, gather_merge_setup(), GatherMergeState::gm_initialized, GatherMergeState::gm_nkeys, GatherMergeState::gm_sortkeys, i, GatherMergeState::initialized, innerPlan, makeNode, GatherMerge::numCols, OUTER_VAR, PlanState::outeropsfixed, PlanState::outeropsset, outerPlan, outerPlanState, palloc0(), PlanState::plan, GatherMerge::plan, PrepareSortSupportFromOrderingOp(), GatherMergeState::ps, PlanState::ps_ProjInfo, Plan::qual, PlanState::resultopsfixed, PlanState::resultopsset, SortSupportData::ssup_attno, SortSupportData::ssup_collation, SortSupportData::ssup_cxt, SortSupportData::ssup_nulls_first, PlanState::state, GatherMergeState::tupDesc, and GatherMergeState::tuples_needed.

Referenced by ExecInitNode().

ExecReScanGatherMerge()

void ExecReScanGatherMerge

(

GatherMergeState *

node

)

Definition at line 342 of file nodeGatherMerge.c.

{
    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);
}

References bms_add_member(), ExecReScan(), ExecShutdownGatherMergeWorkers(), gather_merge_clear_tuples(), GatherMergeState::gm_initialized, GatherMergeState::initialized, outerPlan, outerPlanState, PlanState::plan, GatherMergeState::ps, and GatherMerge::rescan_param.

Referenced by ExecReScan().

ExecShutdownGatherMerge()

void ExecShutdownGatherMerge

(

GatherMergeState *

node

)

Definition at line 305 of file nodeGatherMerge.c.

{
    ExecShutdownGatherMergeWorkers(node);
 
    /* Now destroy the parallel context. */
    if (node->pei != NULL)
    {
        ExecParallelCleanup(node->pei);
        node->pei = NULL;
    }
}

References ExecParallelCleanup(), ExecShutdownGatherMergeWorkers(), and GatherMergeState::pei.

Referenced by ExecEndGatherMerge(), and ExecShutdownNode_walker().

ExecShutdownGatherMergeWorkers()

static void ExecShutdownGatherMergeWorkers ( GatherMergeState *  node )

static

Definition at line 324 of file nodeGatherMerge.c.

{
    if (node->pei != NULL)
        ExecParallelFinish(node->pei);
 
    /* Flush local copy of reader array */
    if (node->reader)
        pfree(node->reader);
    node->reader = NULL;
}

References ExecParallelFinish(), GatherMergeState::pei, pfree(), and GatherMergeState::reader.

Referenced by ExecReScanGatherMerge(), and ExecShutdownGatherMerge().

gather_merge_clear_tuples()

static void gather_merge_clear_tuples ( GatherMergeState *  gm_state )

static

Definition at line 527 of file nodeGatherMerge.c.

{
    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]);
    }
}

References ExecClearTuple(), GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, i, GatherMergeState::nreaders, GMReaderTupleBuffer::nTuples, pfree(), GMReaderTupleBuffer::readCounter, and GMReaderTupleBuffer::tuple.

Referenced by ExecReScanGatherMerge(), and gather_merge_getnext().

gather_merge_getnext()

static TupleTableSlot * gather_merge_getnext ( GatherMergeState *  gm_state )

static

Definition at line 548 of file nodeGatherMerge.c.

{
    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];
    }
}

References binaryheap_empty, binaryheap_first(), binaryheap_remove_first(), binaryheap_replace_first(), DatumGetInt32(), gather_merge_clear_tuples(), gather_merge_init(), gather_merge_readnext(), GatherMergeState::gm_heap, GatherMergeState::gm_initialized, GatherMergeState::gm_slots, i, and Int32GetDatum().

Referenced by ExecGatherMerge().

gather_merge_init()

static void gather_merge_init ( GatherMergeState *  gm_state )

static

Definition at line 444 of file nodeGatherMerge.c.

{
    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;
}

References Assert(), binaryheap_add_unordered(), binaryheap_build(), binaryheap_reset(), castNode, CHECK_FOR_INTERRUPTS, GMReaderTupleBuffer::done, ExecClearTuple(), gather_merge_readnext(), GatherMergeState::gm_heap, GatherMergeState::gm_initialized, GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, i, Int32GetDatum(), load_tuple_array(), GatherMergeState::need_to_scan_locally, GatherMergeState::nreaders, GMReaderTupleBuffer::nTuples, PlanState::plan, GatherMergeState::ps, GMReaderTupleBuffer::readCounter, and TupIsNull.

Referenced by gather_merge_getnext().

gather_merge_readnext()

static bool gather_merge_readnext ( GatherMergeState *  gm_state, int  reader, bool  nowait  )

static

Definition at line 637 of file nodeGatherMerge.c.

{
    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;
}

References ParallelExecutorInfo::area, Assert(), GMReaderTupleBuffer::done, EState::es_query_dsa, ExecProcNode(), ExecStoreMinimalTuple(), gm_readnext_tuple(), GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, load_tuple_array(), GatherMergeState::need_to_scan_locally, GMReaderTupleBuffer::nTuples, outerPlan, outerPlanState, GatherMergeState::pei, GatherMergeState::ps, GMReaderTupleBuffer::readCounter, PlanState::state, TupIsNull, and GMReaderTupleBuffer::tuple.

Referenced by gather_merge_getnext(), and gather_merge_init().

gather_merge_setup()

static void gather_merge_setup ( GatherMergeState *  gm_state )

static

Definition at line 396 of file nodeGatherMerge.c.

{
    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);
}

References binaryheap_allocate(), castNode, ExecInitExtraTupleSlot(), GatherMergeState::gm_heap, GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, heap_compare_slots(), i, MAX_TUPLE_STORE, GatherMerge::num_workers, palloc0(), PlanState::plan, GatherMergeState::ps, PlanState::state, TTSOpsMinimalTuple, GatherMergeState::tupDesc, and GMReaderTupleBuffer::tuple.

Referenced by ExecInitGatherMerge().

gm_readnext_tuple()

static MinimalTuple gm_readnext_tuple ( GatherMergeState *  gm_state, int  nreader, bool  nowait, bool *  done  )

static

Definition at line 715 of file nodeGatherMerge.c.

{
    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;
}

References CHECK_FOR_INTERRUPTS, heap_copy_minimal_tuple(), GatherMergeState::reader, and TupleQueueReaderNext().

Referenced by gather_merge_readnext(), and load_tuple_array().

heap_compare_slots()

static int32 heap_compare_slots ( Datum  a, Datum  b, void *  arg  )

static

Definition at line 753 of file nodeGatherMerge.c.

{
    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;
}

References a, ApplySortComparator(), arg, Assert(), b, compare(), DatumGetInt32(), GatherMergeState::gm_nkeys, GatherMergeState::gm_slots, GatherMergeState::gm_sortkeys, INVERT_COMPARE_RESULT, s1, s2, slot_getattr(), SortSupportData::ssup_attno, and TupIsNull.

Referenced by gather_merge_setup().

load_tuple_array()

static void load_tuple_array ( GatherMergeState *  gm_state, int  reader  )

static

Definition at line 598 of file nodeGatherMerge.c.

{
    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++;
    }
}

References GMReaderTupleBuffer::done, gm_readnext_tuple(), GatherMergeState::gm_tuple_buffers, i, MAX_TUPLE_STORE, GMReaderTupleBuffer::nTuples, GMReaderTupleBuffer::readCounter, and GMReaderTupleBuffer::tuple.

Referenced by gather_merge_init(), and gather_merge_readnext().