Google Git
Sign in
apache / hawq / c4da083e0d1c3a4630531307db264d24994d3d65 / . / src / backend / cdb / cdbmutate.c
blob: 1b57be4554e4b5312de0164b048e4dc705b16e30 [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
/*-------------------------------------------------------------------------
*
* cdbmutate.c
* Parallelize a PostgreSQL sequential plan tree.
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "optimizer/clauses.h"
#include "parser/analyze.h" /* for createRandomDistribution() */
#include "parser/parsetree.h" /* for rt_fetch() */
#include "utils/lsyscache.h" /* for getatttypetypmod() */
#include "nodes/makefuncs.h" /* for makeVar() */
#include "parser/parse_expr.h" /* for expr_type() */
#include "parser/parse_oper.h" /* for compatible_oper_opid() */
#include "utils/relcache.h" /* RelationGetPartitioningKey() */
#include "optimizer/tlist.h" /* get_sortgroupclause_tle() */
#include "optimizer/predtest.h"
#include "optimizer/var.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "catalog/catalog.h"
#include "catalog/gp_policy.h"
#include "catalog/pg_type.h"
#include "catalog/pg_proc.h"
#include "utils/syscache.h"
#include "cdb/cdbdisp.h"
#include "cdb/cdbhash.h" /* isGreenplumDbHashable() */
#include "cdb/cdbllize.h"
#include "cdb/cdbmutate.h"
#include "cdb/cdbpartition.h"
#include "cdb/cdbplan.h"
#include "cdb/cdbpullup.h"
#include "cdb/cdbvars.h"
#include "cdb/cdbtargeteddispatch.h"
#include "nodes/print.h"
#include "executor/executor.h"
/* Enable the optimizer */
extern bool optimizer;
/*
* An ApplyMotionState holds state for the recursive apply_motion_mutator().
* It is externalized here to make it shareable by helper code in other
* modules, e.g., cdbaggmutate.c.
*/
typedef struct ApplyMotionState
{
plan_tree_base_prefix base; /* Required prefix for plan_tree_walker/mutator */
int nextMotionID;
int sliceDepth;
int numInitPlanMotionNodes;
List *initPlans;
} ApplyMotionState;
/*
* External functions
*/
bool is_dummy_plan(Plan *plan);
/*
* Forward Declarations
*/
static void assignMotionID(Node *newnode, ApplyMotionState * context, Node *oldnode);
static int *makeDefaultSegIdxArray(int numSegs);
static Motion *
make_motion(Plan *lefttree,
bool sendSorted,
int numSortCols,
AttrNumber *sortColIdx,
Oid *sortOperators,
bool useExecutorVarFormat);
static Node *apply_motion_mutator(Node *node, ApplyMotionState * context);
static void request_explicit_motion(Plan *plan, Index resultRelationIdx, List *rtable);
static void request_explicit_motion_append(Append *append, List *resultRelationsIdx, List *rtable);
static AttrNumber find_segid_column(List *tlist, Index relid);
static bool
doesUpdateAffectPartitionCols(PlannerInfo *root,
Plan *plan,
Query *query);
/*
* Given an expression, return true if it contains anything non-constant.
*
* an expression is considered "constant" if they consist of constant
* expressions and non-volatile functions with constant arguments.
*/
static bool
contains_nonconstant_walker(Node *node, void *context)
{
if (node == NULL)
return false;
if (IsA(node, Const))
return false;
if (contain_volatile_functions(node))
return true;
return expression_tree_walker(node, contains_nonconstant_walker, context);
}
/*
* Is target list all-constant ?
*
* That means that there is no subplan and the elements of the
* targetlist are either constant or guaranteed to produce constants
* (non-volatile functions with all constant arguments).
*/
static bool
allConstantValuesClause(Plan *node)
{
ListCell *cell;
TargetEntry *tle;
Assert(IsA(node, Result));
if (node->lefttree != NULL)
return false;
foreach(cell, node->targetlist)
{
tle = (TargetEntry *)lfirst(cell);
Assert(tle->expr);
if (contains_nonconstant_walker((Node *)tle->expr, NULL))
return false;
}
return true;
}
static void
directDispatchCalculateHash(Plan *plan, GpPolicy *targetPolicy)
{
int i;
CdbHash *h=NULL;
ListCell *cell=NULL;
bool directDispatch;
h = makeCdbHash(GetPlannerSegmentNum(), HASH_FNV_1);
cdbhashinit(h);
/*
* the nested loops here seem scary --
* especially since we've already
* walked them before -- but I think
* this is still required since they
* may not be in the same order.
* (also typically we don't distribute
* by more than a handful of
* attributes).
*/
directDispatch = true;
for (i = 0; i < targetPolicy->nattrs; i++)
{
Const *c;
TargetEntry *tle = NULL;
foreach(cell, plan->targetlist)
{
tle = (TargetEntry *)lfirst(cell);
Assert(tle->expr);
if (tle->resno != targetPolicy->attrs[i])
continue;
if (!IsA(tle->expr, Const))
{
/* the planner could not simplify this */
directDispatch = false;
break;
}
c = (Const *)tle->expr;
if (c->constisnull)
{
cdbhashnull(h);
}
else
{
Assert(isGreenplumDbHashable(c->consttype));
cdbhash(h, c->constvalue, typeIsArrayType(c->consttype) ? ANYARRAYOID : c->consttype);
}
break;
}
if (!directDispatch)
break;
}
/* We now have the hash-partition that this row belong to */
plan->directDispatch.isDirectDispatch = directDispatch;
if (directDispatch)
{
uint32 hashcode = cdbhashreduce(h);
plan->directDispatch.contentIds = list_make1_int(hashcode );
elog(DEBUG1, "sending single row constant insert to content %d", hashcode);
}
}
/* ------------------------------------------------------------------------- *
* Function apply_motion() and apply_motion_mutator() add motion nodes to
* a top-level Plan tree as directed by the Flow nodes in the plan.
*
* In addition, if the argument Plan produces a partitioned flow, a Motion
* appropriate to the command is added to the top. For example, to consolidate
* the tuple streams from a SELECT command into a single tuple stream on the
* dispatcher.
*
* The result is a deep copy of the argument Plan tree with added Motion
* nodes.
*
* TODO Maybe add more context to argument list!
* ------------------------------------------------------------------------- *
*/
Plan *
apply_motion(PlannerInfo *root, Plan *plan, Query *query)
{
Plan *result;
ListCell *cell;
GpPolicy *targetPolicy = NULL;
GpPolicyType targetPolicyType = POLICYTYPE_ENTRY;
ApplyMotionState state;
bool needToAssignDirectDispatchContentIds = false;
/* Initialize mutator context. */
planner_init_plan_tree_base(&state.base, root); /* error on attempt to descend into subplan plan */
state.nextMotionID = 1; /* Start at 1 so zero will mean "unassigned". */
state.sliceDepth = 0;
state.numInitPlanMotionNodes = 0;
state.initPlans = NIL;
Assert(is_plan_node((Node *) plan) && IsA(query, Query));
Assert(plan->flow && plan->flow->flotype != FLOW_REPLICATED);
/* Does query have a target relation? (INSERT/DELETE/UPDATE) */
/*
* NOTE: This code makes the assumption that if we are working on a hierarchy of tables,
* all the tables are distributed, or all are on the entry DB. Any mixture will fail
*/
if (query->resultRelation > 0)
{
RangeTblEntry *rte = rt_fetch(query->resultRelation, query->rtable);
Assert(rte->rtekind == RTE_RELATION);
targetPolicy = GpPolicyFetch(CurrentMemoryContext, rte->relid);
if (targetPolicy)
targetPolicyType = targetPolicy->ptype;
}
switch (query->commandType)
{
case CMD_SELECT:
if (query->intoClause)
{
List *hashExpr;
ListCell *exp1;
int maxattrs = 200;
if (query->intoPolicy != NULL)
{
targetPolicy = query->intoPolicy;
Assert(query->intoPolicy->ptype == POLICYTYPE_PARTITIONED);
Assert(query->intoPolicy->nattrs >= 0);
Assert(query->intoPolicy->nattrs <= 1024);
}
else
{
/* User did not specify a DISTRIBUTED BY clause */
targetPolicy = palloc0(sizeof(GpPolicy)- sizeof(targetPolicy->attrs) + maxattrs * sizeof(targetPolicy->attrs[0]));
targetPolicy->nattrs = 0;
targetPolicy->bucketnum = GetRelOpt_bucket_num_fromRangeVar(query->intoClause->rel, GetDefaultPartitionNum());
targetPolicy->ptype = POLICYTYPE_PARTITIONED;
/* Find out what the flow is partitioned on */
hashExpr = plan->flow->hashExpr;
if(hashExpr)
foreach(exp1, hashExpr)
{
AttrNumber n;
bool found_expr = false;
/* See if this Expr is a column of the result table */
for(n=1;n<=list_length(plan->targetlist);n++)
{
Var *new_var = NULL;
TargetEntry *target = get_tle_by_resno(plan->targetlist, n);
if (!target->resjunk)
{
Expr *var1 = (Expr *) lfirst(exp1);
/* right side variable may be encapsulated by a relabel node. motion, however, does not care about relabel nodes. */
if (IsA(var1, RelabelType))
var1 = ((RelabelType *)var1)->arg;
/* If subplan expr is a Var, copy to preserve its EXPLAIN info. */
if (IsA(target->expr, Var))
{
new_var = copyObject(target->expr);
new_var->varno = OUTER;
new_var->varattno = n;
}
/*
* Make a Var that references the target list entry at this offset,
* using OUTER as the varno
*/
else
new_var = makeVar(OUTER,
n,
exprType((Node *) target->expr),
exprTypmod((Node *) target->expr),
0);
if (equal(var1,new_var))
{
/* If it is, use it to partition the result table, to avoid
* unnecessary redistibution of data */
targetPolicy->attrs[targetPolicy->nattrs++] = n;
found_expr = true;
break;
}
}
}
if (!found_expr)
break;
}
}
query->intoPolicy = targetPolicy;
/*
* Make sure the top level flow is partitioned on the
* partitioning key of the target relation. Since this
* is a SELECT INTO (basically same as an INSERT) command, the target list will correspond
* to the attributes of the target relation in order.
*/
hashExpr = getExprListFromTargetList(plan->targetlist,
targetPolicy->nattrs,
targetPolicy->attrs,
true);
if (!repartitionPlan(plan, false, false, hashExpr))
ereport(ERROR, (errcode(ERRCODE_CDB_FEATURE_NOT_YET),
errmsg("Cannot parallelize that SELECT INTO yet")
));
Assert(query->intoPolicy->ptype == POLICYTYPE_PARTITIONED);
}
if (plan->flow->flotype == FLOW_PARTITIONED && !query->intoClause)
{
/*
* Query result needs to be brought back to the QD.
* Ask for motion to a single QE. Later, apply_motion
* will override that to bring it to the QD instead.
*
* If the query has an ORDER BY clause, use Merge Receive to
* preserve the ordering.
*
* The plan has already been set up to ensure each qExec's
* result is properly ordered according to the ORDER BY
* specification. The existing ordering could be stronger
* than required; if so, omit the extra trailing columns
* from the Merge Receive key.
*
* An unordered result is ok if the ORDER BY ordering is
* degenerate (on constant exprs) or the result is known to
* have at most one row.
*/
if (query->sortClause &&
plan->flow->numSortCols > 0)
{
if (plan->flow->numSortCols > list_length(query->sortClause))
plan->flow->numSortCols = list_length(query->sortClause);
Insist(focusPlan(plan, true, false));
}
/* Use UNION RECEIVE. Does not preserve ordering. */
else
Insist(focusPlan(plan, false, false));
}
needToAssignDirectDispatchContentIds = root->config->gp_enable_direct_dispatch && ! query->intoClause;
break;
case CMD_INSERT:
{
switch (targetPolicyType)
{
case POLICYTYPE_PARTITIONED:
{
List *hashExpr = NIL;
/*
* Make sure the top level flow is partitioned on the
* partitioning key of the target relation. Since this
* is an INSERT command, the target list will correspond
* to the attributes of the target relation in order.
*/
/* Is this plan an all-const insert ? */
if (gp_enable_fast_sri && IsA(plan, Result))
{
ListCell *cell;
bool typesOK=true;
PartitionNode *pn;
RangeTblEntry *rte;
Relation rel;
rte = rt_fetch(query->resultRelation, query->rtable);
rel = relation_open(rte->relid, NoLock);
/* 1: See if it's partitioned */
pn = RelationBuildPartitionDesc(rel, false);
if (pn && !partition_policies_equal(targetPolicy, pn))
{
/* 2: See if partitioning columns are constant */
List *partatts = get_partition_attrs(pn);
ListCell *lc;
bool all_const = true;
foreach(lc, partatts)
{
List *tl = plan->targetlist;
ListCell *cell;
AttrNumber attnum = lfirst_int(lc);
bool found = false;
foreach(cell, tl)
{
TargetEntry *tle = (TargetEntry *)lfirst(cell);
Assert(tle->expr);
if (tle->resno == attnum)
{
found = true;
if (!IsA(tle->expr, Const))
all_const = false;
break;
}
}
Assert(found);
}
/* 3: if not, mutate plan to make constant */
if (!all_const)
plan->targetlist = (List *)
planner_make_plan_constant(root, (Node *)plan->targetlist, true);
/* better be constant now */
if (allConstantValuesClause(plan))
{
bool *nulls;
Datum *values;
EState *estate = CreateExecutorState();
ResultRelInfo *rri;
/* 4: build tuple, look up partitioning key */
nulls = palloc0(sizeof(bool) * rel->rd_att->natts);
values = palloc(sizeof(Datum) * rel->rd_att->natts);
foreach(lc, partatts)
{
AttrNumber attnum = lfirst_int(lc);
List *tl = plan->targetlist;
ListCell *cell;
foreach(cell, tl)
{
TargetEntry *tle = (TargetEntry *)lfirst(cell);
Assert(tle->expr);
if (tle->resno == attnum)
{
Assert(IsA(tle->expr, Const));
nulls[attnum - 1] = ((Const *)tle->expr)->constisnull;
if (!nulls[attnum - 1])
values[attnum - 1] = ((Const *)tle->expr)->constvalue;
}
}
}
estate->es_result_partitions = pn;
estate->es_partition_state =
createPartitionState(estate->es_result_partitions, 1 /*resultPartSize */);
rri = makeNode(ResultRelInfo);
rri->ri_RangeTableIndex = 1; /* dummy */
rri->ri_RelationDesc = rel;
estate->es_result_relations = rri;
estate->es_num_result_relations = 1;
estate->es_result_relation_info = rri;
rri = values_get_partition(values, nulls, RelationGetDescr(rel),
estate);
/* 5: get target policy for destination table */
targetPolicy = RelationGetPartitioningKey(rri->ri_RelationDesc);
if (targetPolicy->ptype != POLICYTYPE_PARTITIONED)
elog(ERROR, "policy must be partitioned");
ExecCloseIndices(rri);
heap_close(rri->ri_RelationDesc, NoLock);
FreeExecutorState(estate);
}
}
relation_close(rel, NoLock);
hashExpr = getExprListFromTargetList(plan->targetlist,
targetPolicy->nattrs,
targetPolicy->attrs,
true);
/* check the types. */
foreach(cell, hashExpr)
{
Expr *elem = NULL;
Oid att_type = InvalidOid;
elem = (Expr *)lfirst(cell);
att_type = exprType((Node *) elem);
Assert(att_type != InvalidOid);
if (!isGreenplumDbHashable(att_type))
{
typesOK=false;
break;
}
}
/* all constants in values clause -- no need to repartition. */
if (typesOK && allConstantValuesClause(plan))
{
Result *rNode = (Result *)plan;
List *hList = NIL;
int i;
/*
* If this table has child tables, we need to
* find out destination partition.
*
* See partition check above.
*/
/* build our list */
for (i = 0; i < targetPolicy->nattrs; i++)
{
Assert(targetPolicy->attrs[i] > 0);
hList = lappend_int(hList, targetPolicy->attrs[i]);
}
if (root->config->gp_enable_direct_dispatch)
{
directDispatchCalculateHash(plan, targetPolicy);
/* we now either have a hash-code, or we've marked the plan non-directed. */
}
rNode->hashFilter = true;
rNode->hashList = hList;
/* Build a partitioned flow */
plan->flow->flotype = FLOW_PARTITIONED;
plan->flow->locustype = CdbLocusType_Hashed;
plan->flow->hashExpr = hashExpr;
}
}
if (!hashExpr)
hashExpr = getExprListFromTargetList(plan->targetlist,
targetPolicy->nattrs,
targetPolicy->attrs,
true);
if (!repartitionPlan(plan, false, false, hashExpr))
ereport(ERROR, (errcode(ERRCODE_CDB_FEATURE_NOT_YET),
errmsg("Cannot parallelize that INSERT yet")));
break;
}
case POLICYTYPE_ENTRY:
/* All's well if query result is already on the QD. */
if (plan->flow->flotype == FLOW_SINGLETON &&
plan->flow->segindex < 0)
break;
/*
* Query result needs to be brought back to the QD.
* Ask for motion to a single QE. Later, apply_motion
* will override that to bring it to the QD instead.
*/
if (!focusPlan(plan, false, false))
ereport(ERROR, (errcode(ERRCODE_CDB_FEATURE_NOT_YET),
errmsg("Cannot parallelize that INSERT yet")));
break;
default:
Insist(0);
}
}
break;
case CMD_UPDATE:
case CMD_DELETE:
{
/* Distributed target table? */
if (targetPolicyType == POLICYTYPE_PARTITIONED)
{
/*
* The planner does not support updating any of the partitioning columns.
* The optimizer supports updating partitioning columns.
*/
if (query->commandType == CMD_UPDATE &&
!optimizer &&
doesUpdateAffectPartitionCols(root, plan,query))
{
ereport(ERROR, (errcode(ERRCODE_CDB_FEATURE_NOT_YET),
errmsg("Cannot parallelize an UPDATE statement that updates the distribution columns"),
errOmitLocation(true)));
}
if (IsA(plan, Append))
{
/* updating a partitioned table */
request_explicit_motion_append((Append *) plan, root->resultRelations, root->glob->finalrtable);
}
else
{
/* updating a non-partitioned table */
Assert(list_length(root->resultRelations) == 1);
int relid = list_nth_int(root->resultRelations, 0);
Assert(relid > 0);
request_explicit_motion(plan, relid, root->glob->finalrtable);
}
needToAssignDirectDispatchContentIds = root->config->gp_enable_direct_dispatch;
}
/* Target table is not distributed. Must be in entry db. */
else
{
if (plan->flow->flotype == FLOW_PARTITIONED ||
plan->flow->flotype == FLOW_REPLICATED ||
(plan->flow->flotype == FLOW_SINGLETON && plan->flow->segindex != -1))
{
/*
* target table is master-only but flow is distributed:
* add a GatherMotion on top
*/
/* create a shallow copy of the plan flow */
Flow *flow = plan->flow;
plan->flow = (Flow *) palloc(sizeof(Flow));
*(plan->flow) = *flow;
/* save original flow information */
plan->flow->flow_before_req_move = flow;
/* request a GatherMotion node */
plan->flow->req_move = MOVEMENT_FOCUS;
plan->flow->hashExpr = NIL;
plan->flow->segindex = 0;
}
else
{
/*
* Source is, presumably, a dispatcher singleton.
*/
plan->flow->req_move = MOVEMENT_NONE;
}
break;
}
}
break;
default:
Insist(0); /* Never see non-DML in here! */
}
result = (Plan *) apply_motion_mutator((Node *) plan, &state);
if ( needToAssignDirectDispatchContentIds )
{
/* figure out if we can run on a reduced set of nodes */
AssignContentIdsToPlanData(query, result, root);
}
Assert(result->nMotionNodes == state.nextMotionID - 1);
Assert(result->nInitPlans == list_length(state.initPlans));
/* Assign slice numbers to the initplans. */
foreach(cell, state.initPlans)
{
SubPlan *subplan = (SubPlan *)lfirst(cell);
Assert(IsA(subplan, SubPlan) && subplan->qDispSliceId == 0);
subplan->qDispSliceId = state.nextMotionID++;
}
#ifdef USE_ASSERT_CHECKING
/* Check whether plan brings back the tuples to their original segments so they
* can be updated */
if (query->commandType == CMD_DELETE ||
query->commandType == CMD_UPDATE)
{
if (targetPolicyType == POLICYTYPE_ENTRY)
{
/* target table is master-only */
Assert(list_length(root->resultRelations) == 1);
Assert(result->flow->flotype == FLOW_SINGLETON && result->flow->segindex == -1);
}
else if (result->nMotionNodes > state.numInitPlanMotionNodes)
{
/* target table is distributed and plan involves motion nodes */
/*
* Currently, we require all DML plans with Motion to have an ExplicitRedistribute
* motion on top or (for partitioned tables) an Append node with ExplicitRedistribute
* child plans
*/
if (list_length(root->resultRelations) == 1)
{
/* non-partitioned table */
Assert(IsA(result, Motion) && ((Motion *) result)->motionType == MOTIONTYPE_EXPLICIT);
}
else
{
/* updating a partitioned table */
Assert(IsA(result, Append));
ListCell *lcAppend;
foreach(lcAppend, ((Append *) result)->appendplans)
{
Plan *appendPlan = (Plan *) lfirst(lcAppend);
Assert(IsA(appendPlan, Motion) &&
((Motion *) appendPlan)->motionType == MOTIONTYPE_EXPLICIT);
}
}
}
}
#endif
/*
* Discard subtrees of Query node that aren't needed for execution.
* Note the targetlist (query->targetList) is used in execution of
* prepared statements, so we leave that alone.
*/
query->jointree = NULL;
query->groupClause = NIL;
query->havingQual = NULL;
query->distinctClause = NIL;
query->sortClause = NIL;
query->limitCount = NULL;
query->limitOffset = NULL;
query->setOperations = NULL;
return result;
}
/*
* Function apply_motion_mutator() is the workhorse for apply_motion().
*/
Node *
apply_motion_mutator(Node *node, ApplyMotionState * context)
{
Node *newnode;
Plan *plan;
Flow *flow;
int saveNextMotionID;
int saveNumInitPlans;
int saveSliceDepth;
if (node == NULL)
return NULL;
/* An expression node might have subtrees containing plans to be mutated. */
if (!is_plan_node(node))
return plan_tree_mutator(node, apply_motion_mutator, context);
plan = (Plan *)node;
flow = plan->flow;
saveNextMotionID = context->nextMotionID;
saveNumInitPlans = list_length(context->initPlans);
/* Descending into a subquery or a new slice? */
saveSliceDepth = context->sliceDepth;
Assert( !IsA(node,Query) );
if (IsA(node, Motion))
context->sliceDepth++;
else if (flow && flow->req_move != MOVEMENT_NONE )
context->sliceDepth++;
/*
* Copy this node and mutate its children. Afterward, this node should be
* an exact image of the input node, except that contained nodes requiring
* parallelization will have had it applied.
*/
newnode = plan_tree_mutator(node, apply_motion_mutator, context);
context->sliceDepth = saveSliceDepth;
plan = (Plan *)newnode;
flow = plan->flow;
/* Make one big list of all of the initplans. */
if (plan->initPlan)
{
ListCell *cell;
SubPlan *subplan;
int nMotion = 0;
foreach(cell, plan->initPlan)
{
PlannerInfo *root = (PlannerInfo *) context->base.node;
subplan = (SubPlan *)lfirst(cell);
Assert(IsA(subplan, SubPlan));
Assert(root);
Assert(planner_subplan_get_plan(root, subplan));
nMotion += planner_subplan_get_plan(root, subplan)->nMotionNodes;
context->initPlans = lappend(context->initPlans, subplan);
}
/* Keep track of the number of Motion nodes found in Init Plans. */
context->numInitPlanMotionNodes += nMotion;
}
/* Pre-existing Motion nodes must be renumbered. */
if (IsA(newnode, Motion))
{
Motion *motion = (Motion *)newnode;
#ifdef USE_ASSERT_CHECKING
/* Sanity check */
/* Sub plan must have flow */
Assert(flow && motion->plan.lefttree->flow);
Assert(flow->req_move == MOVEMENT_NONE && !flow->flow_before_req_move);
#endif
/* Assign unique node number to the new node. */
assignMotionID(newnode, context, node);
/* If top slice marked as singleton, make it a dispatcher singleton. */
if(motion->motionType == MOTIONTYPE_FIXED
&& motion->numOutputSegs == 1
&& motion->outputSegIdx[0] >= 0
&& context->sliceDepth == 0)
{
Assert(flow->flotype == FLOW_SINGLETON &&
flow->segindex == motion->outputSegIdx[0]);
flow->segindex = -1;
motion->outputSegIdx[0] = -1;
}
goto done;
}
/*
* If no Flow node, we're in a portion of the tree that was finished by
* the planner and already contains any needed Motion nodes.
*/
if (!flow)
goto done;
Assert(IsA(flow, Flow));
/* Add motion atop the copied node, if necessary. */
switch (flow->req_move)
{
case MOVEMENT_FOCUS:
/* If top slice is a singleton, let it execute on qDisp. */
if (flow->segindex >= 0 &&
context->sliceDepth == 0)
flow->segindex = -1;
if (flow->numSortCols > 0)
{
newnode = (Node *)make_sorted_union_motion(plan,
flow->segindex,
flow->numSortCols,
flow->sortColIdx,
flow->sortOperators,
true /* useExecutorVarFormat */);
}
else
{
newnode = (Node *)make_union_motion(plan,
flow->segindex,
true /* useExecutorVarFormat */);
}
break;
case MOVEMENT_BROADCAST:
newnode = (Node *) make_broadcast_motion(plan, true /* useExecutorVarFormat */);
break;
case MOVEMENT_REPARTITION:
newnode = (Node *)make_hashed_motion(plan,
flow->hashExpr,
true /* useExecutorVarFormat */
);
break;
case MOVEMENT_EXPLICIT:
/* add an ExplicitRedistribute motion node only if child plan nodes
* have a motion node
*/
if (context->nextMotionID - context->numInitPlanMotionNodes - saveNextMotionID > 0)
{
/* motion node in child nodes: add a ExplicitRedistribute motion */
newnode = (Node *) make_explicit_motion(plan,
flow->segidColIdx,
true /* useExecutorVarFormat */
);
}
else
{
/* no motion nodes in child plan nodes - no need for ExplicitRedistribute:
* restore flow */
flow->req_move = MOVEMENT_NONE;
flow->flow_before_req_move = NULL;
}
break;
case MOVEMENT_NONE:
/* Update flow if reassigning singleton top slice to qDisp. */
if (flow->flotype == FLOW_SINGLETON &&
flow->segindex >= 0 &&
context->sliceDepth == 0)
{
flow->segindex = -1;
}
break;
default:
Insist(0);
break;
}
/*
* After adding Motion node, assign slice id and restore subplan's Flow.
*/
if (flow->flow_before_req_move)
{
Assert(flow->req_move != MOVEMENT_NONE &&
IsA(newnode, Motion) &&
plan == ((Motion *)newnode)->plan.lefttree);
/* Assign unique node number to the new node. */
assignMotionID(newnode, context, NULL);
/* Replace the subplan's modified Flow with the original. */
plan->flow = flow->flow_before_req_move;
}
else
Assert(flow->req_move == MOVEMENT_NONE);
done:
/*
* Label the Plan node with the number of Motion nodes and Init Plans
* in this subtree, inclusive of the node itself. This info is used
* only in the top node of a query or subquery. Someday, find a better
* place to keep it.
*/
plan = (Plan *)newnode;
plan->nMotionNodes = context->nextMotionID - saveNextMotionID;
plan->nInitPlans = list_length(context->initPlans) - saveNumInitPlans;
return newnode;
} /* apply_motion_mutator */
/*
* Helper code for ApplyMotionState -- Assign motion id to new Motion node.
*/
void
assignMotionID(Node *newnode, ApplyMotionState * context, Node *oldnode)
{
Assert(IsA(newnode, Motion) && context != NULL);
Assert(oldnode == NULL || (IsA(oldnode, Motion) && oldnode != newnode));
/* Assign unique node number to the new node. */
((Motion *) newnode)->motionID = context->nextMotionID++;
/* Debugging hint that we have reassigned a previously assigned ID. */
if (oldnode != NULL && ((Motion *) oldnode)->motionID != 0)
{
ereport(DEBUG2, (
errmsg("Motion node renumbered: old=%d new=%d.",
((Motion *) oldnode)->motionID,
((Motion *) newnode)->motionID)
));
}
}
/* --------------------------------------------------------------------
* make_motion -- creates a Motion node.
* Caller must have built the pHashDefn, pFixedDefn,
* and pSortDefn structs already.
* useExecutorVarFormat is true if make_motion is called after setrefs
* This call only make a motion node, without filling in flow info
* After calling this function, caller need to call add_slice_to_motion
* --------------------------------------------------------------------
*/
Motion *
make_motion(Plan *lefttree,
bool sendSorted,
int numSortCols,
AttrNumber *sortColIdx,
Oid *sortOperators,
bool useExecutorVarFormat)
{
Motion *node = makeNode(Motion);
Plan *plan = &node->plan;
Assert(lefttree);
Assert(!IsA(lefttree, Motion));
AssertImply(sendSorted, (numSortCols > 0 && sortColIdx != NULL && sortOperators != NULL));
AssertImply(!sendSorted, (numSortCols == 0 && sortColIdx == NULL && sortOperators == NULL));
plan->startup_cost = lefttree->startup_cost;
plan->total_cost = lefttree->total_cost;
plan->plan_rows = lefttree->plan_rows;
plan->plan_width = lefttree->plan_width;
plan->targetlist = cdbpullup_targetlist(lefttree, useExecutorVarFormat);
plan->qual = NIL;
plan->lefttree = lefttree;
plan->righttree = NULL;
plan->dispatch = DISPATCH_PARALLEL;
node->sendSorted = sendSorted;
node->numSortCols = numSortCols;
node->sortColIdx = sortColIdx;
node->sortOperators = sortOperators;
plan->extParam = bms_copy(lefttree->extParam);
plan->allParam = bms_copy(lefttree->allParam);
plan->flow = NULL;
node->outputSegIdx = NULL;
node->numOutputSegs = 0;
return node;
}
void add_slice_to_motion(Motion *motion,
MotionType motionType, List *hashExpr,
int numOutputSegs, int *outputSegIdx)
{
/* sanity checks */
/* check numOutputSegs and outputSegIdx are in sync. */
AssertEquivalent(numOutputSegs == 0, outputSegIdx == NULL);
/* numOutputSegs are either 0, for hash or broadcast, or 1, for focus */
Assert(numOutputSegs == 0 || numOutputSegs == 1);
AssertImply(motionType == MOTIONTYPE_HASH, numOutputSegs == 0);
AssertImply(numOutputSegs == 1, (outputSegIdx[0] == -1 || outputSegIdx[0] == gp_singleton_segindex));
motion->motionType = motionType;
motion->hashExpr = hashExpr;
motion->hashDataTypes = NULL;
motion->numOutputSegs = numOutputSegs;
motion->outputSegIdx = outputSegIdx;
Assert(motion->plan.lefttree);
/* Build list of hash key expression data types. */
if (hashExpr)
{
List *eq = list_make1(makeString("="));
ListCell *cell;
foreach(cell, hashExpr)
{
Node *expr = (Node *)lfirst(cell);
Oid typeoid = exprType(expr);
Oid eqopoid;
Oid lefttype;
Oid righttype;
/* Get oid of the equality operator for this data type. */
eqopoid = compatible_oper_opid(eq, typeoid, typeoid, true);
if ( eqopoid == InvalidOid )
ereport(ERROR, (errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("no equality operator for typid %d",
typeoid)));
/* Get the equality operator's operand type. */
op_input_types( eqopoid, &lefttype, &righttype );
Assert( lefttype == righttype );
/* If this type is a domain type, get its base type. */
if (get_typtype(lefttype) == 'd')
lefttype = getBaseType(lefttype);
motion->hashDataTypes = lappend_oid(motion->hashDataTypes, lefttype);
}
list_free_deep(eq);
}
/* Attach a descriptive Flow. */
switch(motion->motionType)
{
case MOTIONTYPE_HASH:
motion->plan.flow = makeFlow(FLOW_PARTITIONED);
motion->plan.flow->locustype = CdbLocusType_Hashed;
motion->plan.flow->hashExpr = copyObject(motion->hashExpr);
motion->numOutputSegs = GetPlannerSegmentNum();
motion->outputSegIdx = makeDefaultSegIdxArray(GetPlannerSegmentNum());
break;
case MOTIONTYPE_FIXED:
if (motion->numOutputSegs == 0 )
{
/* broadcast */
motion->plan.flow = makeFlow(FLOW_REPLICATED);
motion->plan.flow->locustype = CdbLocusType_Replicated;
}
else if (motion->numOutputSegs == 1 )
{
/* Focus motion */
motion->plan.flow = makeFlow(FLOW_SINGLETON);
motion->plan.flow->segindex = motion->outputSegIdx[0];
motion->plan.flow->locustype = ( motion->plan.flow->segindex<0 ) ?
CdbLocusType_Entry :
CdbLocusType_SingleQE;
}
else
{
Assert(!"Invalid numoutputSegs for motion type fixed");
motion->plan.flow = NULL;
}
break;
case MOTIONTYPE_EXPLICIT:
motion->plan.flow = makeFlow(FLOW_PARTITIONED);
/* TODO: antova - Nov 18, 2010; add a special locus type for ExplicitRedistribute flows */
motion->plan.flow->locustype = CdbLocusType_Strewn;
break;
default:
Assert(!"Invalid motion type");
motion->plan.flow = NULL;
break;
}
Assert(motion->plan.flow);
/* if the motion has a sort order, preserve it in each route */
if(motion->sendSorted && motion->numSortCols > 0)
{
int i, n;
n = motion->numSortCols;
motion->plan.flow->numSortCols = n;
motion->plan.flow->sortColIdx = palloc(n * sizeof(AttrNumber));
motion->plan.flow->sortOperators = palloc (n * sizeof(Oid));
for ( i = 0; i < n; i++ )
{
motion->plan.flow->sortColIdx[i] = motion->sortColIdx[i];
motion->plan.flow->sortOperators[i] = motion->sortOperators[i];
}
}
}
Motion *
make_union_motion(Plan *lefttree, int destSegIndex, bool useExecutorVarFormat)
{
Motion *motion;
int *outSegIdx = (int *) palloc(sizeof(int));
outSegIdx[0] = destSegIndex;
motion = make_motion(lefttree, false,
0, NULL, NULL, /* numSortCols, sortColIdx, sortOperators */
useExecutorVarFormat);
add_slice_to_motion(motion, MOTIONTYPE_FIXED,
NULL, 1, outSegIdx);
return motion;
}
Motion *
make_sorted_union_motion(Plan *lefttree,
int destSegIndex,
int numSortCols, AttrNumber *sortColIdx, Oid *sortOperators, bool useExecutorVarFormat)
{
Motion *motion;
int *outSegIdx = (int *) palloc(sizeof(int));
outSegIdx[0] = destSegIndex;
motion = make_motion(lefttree, true,
numSortCols, sortColIdx, sortOperators, useExecutorVarFormat);
add_slice_to_motion(motion, MOTIONTYPE_FIXED,
NULL, 1, outSegIdx);
return motion;
}
Motion *
make_hashed_motion(Plan *lefttree,
List *hashExpr, bool useExecutorVarFormat)
{
Motion *motion;
motion = make_motion(lefttree, false,
0, NULL, NULL, useExecutorVarFormat);
add_slice_to_motion(motion, MOTIONTYPE_HASH,
hashExpr,
0, NULL);
return motion;
}
Motion *
make_broadcast_motion(Plan *lefttree, bool useExecutorVarFormat)
{
Motion *motion;
motion = make_motion(lefttree, false,
0, NULL, NULL, useExecutorVarFormat);
add_slice_to_motion(motion, MOTIONTYPE_FIXED,
NULL, 0, NULL);
return motion;
}
Motion *
make_explicit_motion(Plan *lefttree, AttrNumber segidColIdx, bool useExecutorVarFormat)
{
Motion *motion;
motion = make_motion(lefttree, false,
0, NULL, NULL, useExecutorVarFormat); /* numSortCols, sortColIdx, sortOperators */
Assert(segidColIdx > 0 && segidColIdx <= list_length(lefttree->targetlist));
motion->segidColIdx = segidColIdx;
add_slice_to_motion(motion, MOTIONTYPE_EXPLICIT,
NULL, 0, NULL); /* hashExpr, numOutputSegs, outputSegIdx */
return motion;
}
/* --------------------------------------------------------------------
* makeDefaultSegIdxArray -- creates an int array with numSegs elements,
* with values between 0 and numSegs-1
* --------------------------------------------------------------------
*/
int *
makeDefaultSegIdxArray(int numSegs)
{
int *segIdx = palloc(numSegs * sizeof(int));
/*
* The following assumes that the default is to have segindexes
* numbered sequentially starting at 0.
*/
int i;
for (i = 0; i < numSegs; i++)
segIdx[i] = i;
return segIdx;
}
/* --------------------------------------------------------------------
*
* Static Helper routines
* --------------------------------------------------------------------
*/
/* ----------------------------------------------------------------
* getExprListFromTargetList
*
* Creates a VAR that references the indicated entries from the target list,
* sets the restype and restypmod fields from the target list info,
* and puts them into a list.
*
* The AttrNumber indexes actually refer to the 1 based index into the
* target list.
*
* The entries have the varno field replaced by references in OUTER.
* ----------------------------------------------------------------
*/
List *
getExprListFromTargetList(List *tlist,
int numCols,
AttrNumber *colIdx,
bool useExecutorVarFormat)
{
int i;
List *elist = NIL;
for (i = 0; i < numCols; i++)
{
/* Find expr in TargetList */
AttrNumber n = colIdx[i];
TargetEntry *target = get_tle_by_resno(tlist, n);
if (target == NULL)
elog(ERROR, "no tlist entry for key %d", n);
/* After set_plan_references(), make a Var referencing subplan tlist. */
if (useExecutorVarFormat)
elist = lappend(elist, cdbpullup_make_expr(OUTER, n, target->expr, false));
/* Before set_plan_references(), copy the subplan's result expr. */
else
elist = lappend(elist, copyObject(target->expr));
}
return elist;
}
/*
* isAnyColChangedByUpdate
*/
static bool
isAnyColChangedByUpdate(Index targetvarno,
List *targetlist,
int nattrs,
AttrNumber *attrs)
{
/*
* Want to test whether an update statement possibly changes the
* value of any of the partitioning columns.
*/
/*
* For each partitioning column, the TargetListEntry for that varattno
* should be just a Var node with the same varattno.
*/
int i;
for (i = 0; i < nattrs; i++)
{
TargetEntry *tle = get_tle_by_resno(targetlist, attrs[i]);
Var *var;
Insist(tle);
if (!IsA(tle->expr, Var))
return true;
var = (Var *)tle->expr;
if (var->varnoold != targetvarno ||
var->varoattno != attrs[i])
return true;
}
return false;
} /* isAnyColChangedByUpdate */
/*
* Request an ExplicitRedistribute motion node for a plan node
*/
void request_explicit_motion(Plan *plan, Index resultRelationsIdx, List *rtable)
{
/* request a segid redistribute motion */
/* create a shallow copy of the plan flow */
Flow *flow = plan->flow;
plan->flow = (Flow *) palloc(sizeof(*(plan->flow)));
*(plan->flow) = *flow;
/* save original flow information */
plan->flow->flow_before_req_move = flow;
/* request a SegIdRedistribute motion node */
plan->flow->req_move = MOVEMENT_EXPLICIT;
/* find segid column in target list */
AttrNumber segidColIdx = find_segid_column(plan->targetlist, resultRelationsIdx);
Assert(-1 != segidColIdx);
plan->flow->segidColIdx = segidColIdx;
}
/*
* Request SegIdRedistribute motion node for an Append plan updating a partitioned
* table
*/
void request_explicit_motion_append(Append *append, List *resultRelationsIdx, List *rtable)
{
Assert(IsA(append, Append));
Assert(list_length(append->appendplans) == list_length(resultRelationsIdx));
ListCell *lcAppendPlan = NULL;
ListCell *lcResultRel = NULL;
forboth(lcAppendPlan, append->appendplans,
lcResultRel, resultRelationsIdx)
{
/* request a segid redistribute motion node to be put above each
* append plan */
Plan *appendChildPlan = (Plan *) lfirst(lcAppendPlan);
int relid = lfirst_int(lcResultRel);
Assert(relid > 0);
request_explicit_motion(appendChildPlan, relid, rtable);
}
}
/*
* Find the index of the segid column of the requested relation (relid) in the
* target list
*/
AttrNumber find_segid_column(List *tlist, Index relid)
{
if (NIL == tlist)
{
return -1;
}
ListCell *lcte = NULL;
foreach(lcte, tlist)
{
TargetEntry *te = (TargetEntry *) lfirst(lcte);
Var *var = (Var *) te->expr;
if (!IsA(var, Var))
{
continue;
}
if ((var->varno == relid && var->varattno == GpSegmentIdAttributeNumber) ||
(var->varnoold == relid && var->varoattno == GpSegmentIdAttributeNumber))
{
return te->resno;
}
}
// no segid column found
return -1;
}
/*
* doesUpdateAffectPartitionCols
*/
bool
doesUpdateAffectPartitionCols(PlannerInfo *root,
Plan *plan,
Query *query)
{
Append *append;
ListCell *rticell;
ListCell *plancell;
GpPolicy *policy;
if (list_length(root->resultRelations) == 1)
{
RangeTblEntry *rte;
Relation relation;
int resultRelation = linitial_int(root->resultRelations);
rte = rt_fetch(resultRelation, query->rtable);
Assert(rte->rtekind == RTE_RELATION);
/* Get a copy of the rel's GpPolicy from the relcache. */
relation = relation_open(rte->relid, NoLock);
policy = RelationGetPartitioningKey(relation);
relation_close(relation, NoLock);
/*
* Single target relation?
*/
return isAnyColChangedByUpdate(resultRelation,
plan->targetlist,
policy->nattrs,
policy->attrs);
}
/*
* Multiple target relations (inheritance)...
* Plan should have an Append node on top, with a subplan per target.
*/
append = (Append *)plan;
Insist(IsA(append, Append) &&
append->isTarget &&
list_length(append->appendplans) == list_length(root->resultRelations));
forboth(rticell, root->resultRelations,
plancell, append->appendplans)
{
int rti = lfirst_int(rticell);
Plan *subplan = (Plan *)lfirst(plancell);
RangeTblEntry *rte;
Relation relation;
rte = rt_fetch(rti, query->rtable);
Assert(rte->rtekind == RTE_RELATION);
/* Get a copy of the rel's GpPolicy from the relcache. */
relation = relation_open(rte->relid, NoLock);
policy = RelationGetPartitioningKey(relation);
relation_close(relation, NoLock);
if (isAnyColChangedByUpdate(rti, subplan->targetlist, policy->nattrs, policy->attrs))
return true;
}
return false;
} /* doesUpdateAffectPartitionCols */
/* ----------------------------------------------------------------------- *
* cdbmutate_warn_ctid_without_segid() warns the user if the plan refers to a
* partitioned table's ctid column without also referencing its
* gp_segment_id column.
* ----------------------------------------------------------------------- *
*/
typedef struct ctid_inventory_context
{
plan_tree_base_prefix base; /* Required prefix for plan_tree_walker/mutator */
bool uses_ctid;
bool uses_segid;
Index relid;
} ctid_inventory_context;
static bool
ctid_inventory_walker(Node *node, ctid_inventory_context *inv)
{
if (node == NULL)
return false;
if (IsA(node, Var))
{
Var *var = (Var *)node;
if (var->varattno < 0 &&
var->varno == inv->relid &&
var->varlevelsup == 0)
{
if (var->varattno == SelfItemPointerAttributeNumber)
inv->uses_ctid = true;
else if (var->varattno == GpSegmentIdAttributeNumber)
inv->uses_segid = true;
}
return false;
}
return plan_tree_walker(node, ctid_inventory_walker, inv);
}
void
cdbmutate_warn_ctid_without_segid(struct PlannerInfo *root, struct RelOptInfo *rel)
{
ctid_inventory_context inv;
Relids relids_to_ignore;
ListCell *cell;
AttrNumber attno;
int ndx;
planner_init_plan_tree_base(&inv.base, root);
inv.uses_ctid = false;
inv.uses_segid = false;
inv.relid = rel->relid;
/* Rel not distributed? Then segment id doesn't matter. */
if (!rel->cdbpolicy ||
rel->cdbpolicy->ptype != POLICYTYPE_PARTITIONED)
return;
/* Ignore references occurring in the Query's final output targetlist. */
relids_to_ignore = bms_make_singleton(0);
/* Is 'ctid' referenced in join quals? */
attno = SelfItemPointerAttributeNumber;
Assert(attno >= rel->min_attr && attno <= rel->max_attr);
ndx = attno - rel->min_attr;
if (bms_nonempty_difference(rel->attr_needed[ndx], relids_to_ignore))
inv.uses_ctid = true;
/* Is 'gp_segment_id' referenced in join quals? */
attno = GpSegmentIdAttributeNumber;
Assert(attno >= rel->min_attr && attno <= rel->max_attr);
ndx = attno - rel->min_attr;
if (bms_nonempty_difference(rel->attr_needed[ndx], relids_to_ignore))
inv.uses_segid = true;
/* Examine the single-table quals on this rel. */
foreach(cell, rel->baserestrictinfo)
{
RestrictInfo *rinfo = (RestrictInfo *)lfirst(cell);
Assert(IsA(rinfo, RestrictInfo));
ctid_inventory_walker((Node *)rinfo->clause, &inv);
}
/* Notify client if found a reference to ctid only, without gp_segment_id */
if (inv.uses_ctid &&
!inv.uses_segid)
{
RangeTblEntry *rte = rt_fetch(rel->relid, root->parse->rtable);
const char *cmd;
int elevel;
/* Reject if UPDATE or DELETE. Otherwise just give info msg. */
switch (root->parse->commandType)
{
case CMD_UPDATE:
cmd = "UPDATE";
elevel = ERROR;
break;
case CMD_DELETE:
cmd = "DELETE";
elevel = ERROR;
break;
default:
cmd = "SELECT";
elevel = NOTICE;
}
ereport(elevel, (errmsg("%s uses system-defined column \"%s.ctid\" "
"without the necessary companion column "
"\"%s.gp_segment_id\"",
cmd,
rte->eref->aliasname,
rte->eref->aliasname),
errhint("To uniquely identify a row within a "
"distributed table, use the \"gp_segment_id\" "
"column together with the \"ctid\" column.")
));
}
bms_free(relids_to_ignore);
} /* cdbmutate_warn_ctid_without_segid */
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/dependency.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_depend.h"
#include "miscadmin.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
/*
* Code that mutate the tree for share input
*
* After the planner, the plan is really a DAG. SISCs will have valid
* pointer to the underlying share. However, other code (setrefs etc)
* depends on the fact that the plan is a tree. We first mutate the DAG
* to a tree.
*
* Next, we will need to decide if the share is cross slices. If the share
* is not cross slice, we do not need the syncrhonization, and it is possible to
* keep the Material/Sort in memory to save a sort.
*
* It is essential that we walk the tree in the same order as the ExecProcNode start
* execution, otherwise, deadlock may rise.
*/
/* Walk the tree for shareinput.
* Shareinput fix shared_as_id and underlying_share_id of nodes in place. We do not want to use
* the ordinary tree walker as it is unnecessary to make copies etc.
*/
typedef bool (*SHAREINPUT_MUTATOR) (Node *node, ApplyShareInputContext *ctxt, bool fPop);
static void shareinput_walker(SHAREINPUT_MUTATOR f, Node *node, ApplyShareInputContext *ctxt)
{
Plan *plan = NULL;
bool recursive_down;
if(node == NULL)
return;
if(IsA(node, List))
{
List *l = (List *) node;
ListCell *lc;
foreach(lc, l)
{
Node* n = lfirst(lc);
shareinput_walker(f, n, ctxt);
}
return;
}
if(!is_plan_node(node))
return;
plan = (Plan *) node;
recursive_down = (*f)(node, ctxt, false);
if(recursive_down)
{
if(IsA(node, Append))
{
ListCell *cell;
Append *app = (Append *) node;
foreach(cell, app->appendplans)
shareinput_walker(f, (Node *)lfirst(cell), ctxt);
}
else if(IsA(node, SubqueryScan))
{
SubqueryScan *subqscan = (SubqueryScan *) node;
shareinput_walker(f, (Node *) subqscan->subplan, ctxt);
}
else if(IsA(node, BitmapAnd))
{
ListCell *cell;
BitmapAnd *ba = (BitmapAnd *) node;
foreach(cell, ba->bitmapplans)
shareinput_walker(f, (Node *)lfirst(cell), ctxt);
}
else if(IsA(node, BitmapOr))
{
ListCell *cell;
BitmapOr *bo = (BitmapOr *) node;
foreach(cell, bo->bitmapplans)
shareinput_walker(f, (Node *)lfirst(cell), ctxt);
}
else if(IsA(node, NestLoop))
{
/* Nest loop join is strange. Exec order depends on prefetch_inner */
NestLoop *nl = (NestLoop *) node;
if(nl->join.prefetch_inner)
{
shareinput_walker(f, (Node *)plan->righttree, ctxt);
shareinput_walker(f, (Node *)plan->lefttree, ctxt);
}
else
{
shareinput_walker(f, (Node *)plan->lefttree, ctxt);
shareinput_walker(f, (Node *)plan->righttree, ctxt);
}
}
else if(IsA(node, HashJoin))
{
/* Hash join the hash table is at inner */
shareinput_walker(f, (Node *)plan->righttree, ctxt);
shareinput_walker(f, (Node *)plan->lefttree, ctxt);
}
else if (IsA(node, MergeJoin))
{
MergeJoin *mj = (MergeJoin *) node;
if (mj->unique_outer)
{
shareinput_walker(f, (Node *)plan->lefttree, ctxt);
shareinput_walker(f, (Node *)plan->righttree, ctxt);
}
else
{
shareinput_walker(f, (Node *)plan->righttree, ctxt);
shareinput_walker(f, (Node *)plan->lefttree, ctxt);
}
}
else if (IsA(node, Sequence))
{
ListCell *cell = NULL;
Sequence *sequence = (Sequence *) node;
foreach(cell, sequence->subplans)
{
shareinput_walker(f, (Node *)lfirst(cell), ctxt);
}
}
else
{
shareinput_walker(f, (Node *)plan->lefttree, ctxt);
shareinput_walker(f, (Node *)plan->righttree, ctxt);
shareinput_walker(f, (Node *)plan->initPlan, ctxt);
}
}
(*f)(node, ctxt, true);
}
static void leftdeep_walker(SHAREINPUT_MUTATOR f, Node *node, ApplyShareInputContext *ctxt)
{
Plan *plan = NULL;
bool recursive_down;
if(node == NULL)
return;
if(IsA(node, List))
{
List *l = (List *) node;
ListCell *lc;
foreach(lc, l)
{
Node* n = lfirst(lc);
leftdeep_walker(f, n, ctxt);
}
return;
}
if(!is_plan_node(node))
return;
plan = (Plan *) node;
recursive_down = (*f)(node, ctxt, false);
if(recursive_down)
{
if(IsA(node, Append))
{
ListCell *cell;
Append *app = (Append *) node;
foreach(cell, app->appendplans)
leftdeep_walker(f, (Node *)lfirst(cell), ctxt);
}
else if(IsA(node, BitmapAnd))
{
ListCell *cell;
BitmapAnd *ba = (BitmapAnd *) node;
foreach(cell, ba->bitmapplans)
leftdeep_walker(f, (Node *)lfirst(cell), ctxt);
}
else if(IsA(node, BitmapOr))
{
ListCell *cell;
BitmapOr *bo = (BitmapOr *) node;
foreach(cell, bo->bitmapplans)
leftdeep_walker(f, (Node *)lfirst(cell), ctxt);
}
else if(IsA(node, SubqueryScan))
{
SubqueryScan *subqscan = (SubqueryScan *) node;
leftdeep_walker(f, (Node *) subqscan->subplan, ctxt);
}
else
{
leftdeep_walker(f, (Node *)plan->lefttree, ctxt);
leftdeep_walker(f, (Node *)plan->righttree, ctxt);
leftdeep_walker(f, (Node *)plan->initPlan, ctxt);
}
}
(*f)(node, ctxt, true);
}
/*
* DAG to Tree
* Zap children if I am not the first sharer (not recursive down).
* Assign share_id to both ShareInputScan and Material/Sort.
*/
static bool shareinput_mutator_dag_to_tree(Node *node, ApplyShareInputContext* ctxt, bool fPop)
{
Plan *plan = (Plan *) node;
if(fPop)
return true;
if(IsA(plan, ShareInputScan))
{
Plan *subplan = plan->lefttree;
Assert(subplan);
Assert(IsA(subplan, Material) || IsA(subplan, Sort));
Assert(get_plan_share_id(plan) == SHARE_ID_NOT_ASSIGNED);
Assert(plan->righttree == NULL);
Assert(plan->initPlan == NULL);
if(list_member_ptr(ctxt->sharedNodes, subplan))
{
Assert(get_plan_share_id(subplan) >= 0);
set_plan_share_id(plan, get_plan_share_id(subplan));
plan->lefttree = NULL;
return false;
}
else
{
Assert(get_plan_share_id(subplan) == SHARE_ID_NOT_ASSIGNED);
set_plan_share_id(subplan, list_length(ctxt->sharedNodes));
set_plan_share_id(plan, get_plan_share_id(subplan));
ctxt->sharedNodes = lappend(ctxt->sharedNodes, subplan);
return true;
}
}
return true;
}
/*
* Apply the share input mutator.
*/
Plan *apply_shareinput_dag_to_tree(Plan *plan, ApplyShareInputContext *ctxt)
{
shareinput_walker(shareinput_mutator_dag_to_tree, (Node *) plan, ctxt);
return plan;
}
/* Some helper: implements a stack using List. */
static void shareinput_pushmot(ApplyShareInputContext *ctxt, int motid)
{
ctxt->motStack = lcons_int(motid, ctxt->motStack);
}
static void shareinput_popmot(ApplyShareInputContext *ctxt)
{
list_delete_first(ctxt->motStack);
}
static int shareinput_peekmot(ApplyShareInputContext *ctxt)
{
return linitial_int(ctxt->motStack);
}
typedef struct ShareNodeWithSliceMark
{
Plan *plan;
int slice_mark;
} ShareNodeWithSliceMark;
static bool shareinput_find_sharenode(ApplyShareInputContext *ctxt, int share_id, ShareNodeWithSliceMark *result)
{
ListCell *lc;
ListCell *lc_slicemark;
if(ctxt->sharedNodes == NULL)
return false;
forboth(lc, ctxt->sharedNodes, lc_slicemark, ctxt->sliceMarks)
{
Plan *plan = (Plan *) lfirst(lc);
Assert(IsA(plan, Material) || IsA(plan, Sort));
if(get_plan_share_id(plan) == share_id)
{
if(result)
{
result->plan = plan;
result->slice_mark = lfirst_int(lc_slicemark);
}
return true;
}
}
return false;
}
/*
* First walk on shareinput xslice. It does the following:
*
* 1. Build the sharedNodes in context.
* 2. Build the sliceMarks in context.
* 3. Build a list a share on QD
*/
static bool
shareinput_mutator_xslice_1(Node* node, ApplyShareInputContext *ctxt, bool fPop)
{
Plan *plan = (Plan *) node;
if(fPop)
{
if(IsA(plan, Motion))
shareinput_popmot(ctxt);
return false;
}
if(IsA(plan, Motion))
{
Motion *motion = (Motion *) plan;
shareinput_pushmot(ctxt, motion->motionID);
return true;
}
if(IsA(plan, ShareInputScan))
{
ShareInputScan *sisc = (ShareInputScan *) plan;
int motId = shareinput_peekmot(ctxt);
Plan *shared = plan->lefttree;
Assert(sisc->plan.flow);
if(sisc->plan.flow->flotype == FLOW_SINGLETON)
{
if(sisc->plan.flow->segindex < 0)
ctxt->qdShares = list_append_unique_int(ctxt->qdShares, sisc->share_id);
}
if (shared)
{
Assert(shareinput_find_sharenode(ctxt, sisc->share_id, NULL) == false);
Assert(get_plan_share_id(plan) == get_plan_share_id(shared));
set_plan_driver_slice(shared, motId);
ctxt->sharedNodes = lappend(ctxt->sharedNodes, shared);
ctxt->sliceMarks = lappend_int(ctxt->sliceMarks, motId);
}
}
return true;
}
/*
* Second pass on shareinput xslice. It marks the shared node xslice,
* if a 'shared' is cross-slice.
*/
static bool
shareinput_mutator_xslice_2(Node* node, ApplyShareInputContext *ctxt, bool fPop)
{
Plan *plan = (Plan *) node;
if (fPop)
{
if (IsA(plan, Motion))
shareinput_popmot(ctxt);
return false;
}
if (IsA(plan, Motion))
{
Motion *motion = (Motion *) plan;
shareinput_pushmot(ctxt, motion->motionID);
return true;
}
if (IsA(plan, ShareInputScan))
{
ShareInputScan *sisc = (ShareInputScan *) plan;
int motId = shareinput_peekmot(ctxt);
Plan *shared = plan->lefttree;
if(!shared)
{
ShareNodeWithSliceMark plan_slicemark = {NULL /* plan */,0 /* slice_mark */};
int shareSliceId = 0;
shareinput_find_sharenode(ctxt, sisc->share_id, &plan_slicemark);
if (plan_slicemark.plan == NULL)
elog(ERROR, "could not find shared input node with id %d", sisc->share_id);
shareSliceId = get_plan_driver_slice(plan_slicemark.plan);
if(shareSliceId != motId)
{
ShareType stype = get_plan_share_type(plan_slicemark.plan);
if(stype == SHARE_MATERIAL || stype == SHARE_SORT)
set_plan_share_type_xslice(plan_slicemark.plan);
incr_plan_nsharer_xslice(plan_slicemark.plan);
sisc->driver_slice = motId;
}
}
}
return true;
}
/*
* Third pass:
* 1. Mark shareinput scan xslice,
* 2. Bulid a list of QD slices
*/
static bool
shareinput_mutator_xslice_3(Node *node, ApplyShareInputContext *ctxt, bool fPop)
{
Plan *plan = (Plan *) node;
if(fPop)
{
if(IsA(plan, Motion))
shareinput_popmot(ctxt);
return false;
}
if(IsA(plan, Motion))
{
Motion *motion = (Motion *) plan;
shareinput_pushmot(ctxt, motion->motionID);
return true;
}
if(IsA(plan, ShareInputScan))
{
ShareInputScan *sisc = (ShareInputScan *) plan;
int motId = shareinput_peekmot(ctxt);
ShareNodeWithSliceMark plan_slicemark = { NULL, 0 };
ShareType stype = SHARE_NOTSHARED;
shareinput_find_sharenode(ctxt, sisc->share_id, &plan_slicemark);
if(plan_slicemark.plan)
{
stype = get_plan_share_type(plan_slicemark.plan);
}
switch(stype)
{
case SHARE_MATERIAL_XSLICE:
Assert(sisc->share_type == SHARE_MATERIAL);
set_plan_share_type_xslice(plan);
break;
case SHARE_SORT_XSLICE:
Assert(sisc->share_type == SHARE_SORT);
set_plan_share_type_xslice(plan);
break;
default:
Assert(sisc->share_type == stype);
break;
}
if(list_member_int(ctxt->qdShares, sisc->share_id))
{
Assert(sisc->plan.flow);
Assert(sisc->plan.flow->flotype == FLOW_SINGLETON);
ctxt->qdSlices = list_append_unique_int(ctxt->qdSlices, motId);
}
}
return true;
}
/*
* The fourth pass. If a shareinput is running on QD, then all slices in
* this share must be on QD. Move them to QD.
*/
static bool
shareinput_mutator_xslice_4(Node *node, ApplyShareInputContext *ctxt, bool fPop)
{
Plan *plan = (Plan *) node;
int motId = shareinput_peekmot(ctxt);
if(fPop)
{
if(IsA(plan, Motion))
shareinput_popmot(ctxt);
return false;
}
if(IsA(plan, Motion))
{
Motion *motion = (Motion *) plan;
shareinput_pushmot(ctxt, motion->motionID);
/* Do not return. Motion need to be adjusted as well */
}
/* Well, the following test can be optimized if we record the test
* result so we test just once for all node in one slice. But this
* code is not perf critical so be lazy.
*/
if(list_member_int(ctxt->qdSlices, motId))
{
if(plan->flow)
{
Assert(plan->flow->flotype == FLOW_SINGLETON);
plan->flow->segindex = -1;
}
}
return true;
}
static void gpmon_pushplannode(ApplyShareInputContext *ctxt, Plan *plan)
{
ctxt->planNodes = lcons(plan, ctxt->planNodes);
}
static void gpmon_popplannode(ApplyShareInputContext *ctxt)
{
list_delete_first(ctxt->planNodes);
}
static Plan *gpmon_peekplannode(ApplyShareInputContext *ctxt)
{
return ctxt->planNodes == NULL ? NULL : linitial(ctxt->planNodes);
}
static bool assign_plannode_id_walker(Node *node, ApplyShareInputContext *ctxt, bool fPop)
{
Plan *p = (Plan *) node;
Plan *parent;
if(fPop)
{
gpmon_popplannode(ctxt);
return false;
}
p->plan_node_id = ++ctxt->nextPlanId;
parent = gpmon_peekplannode(ctxt);
if(!parent)
p->plan_parent_node_id = -1;
else
p->plan_parent_node_id = parent->plan_node_id;
gpmon_pushplannode(ctxt, p);
return true;
}
Plan *
apply_shareinput_xslice(Plan *plan, PlannerGlobal *glob)
{
ApplyShareInputContext *ctxt = &glob->share;
ListCell *lp;
ctxt->sharedNodes = NULL;
ctxt->sliceMarks = NULL;
ctxt->motStack = NULL;
ctxt->qdShares = NULL;
ctxt->qdSlices = NULL;
ctxt->planNodes = NIL;
ctxt->nextPlanId = 0;
shareinput_pushmot(ctxt, 0);
/*
* Walk the tree. See comment for each pass for what each pass will do.
* The context is used to carry information from one pass to another,
* as well as within a pass.
*/
/*
* A subplan might have a SharedScan consumer while the SharedScan
* producer is in the main plan, or vice versa. So in the first pass, we
* walk through all plans and collect all producer subplans into the
* context, before processing the consumers.
*/
foreach (lp, glob->subplans)
{
Plan *subplan = (Plan *) lfirst(lp);
shareinput_walker(shareinput_mutator_xslice_1, (Node *) subplan, ctxt);
}
shareinput_walker(shareinput_mutator_xslice_1, (Node *) plan, ctxt);
/* Now walk the tree again, and process all the consumers. */
foreach (lp, glob->subplans)
{
Plan *subplan = (Plan *) lfirst(lp);
shareinput_walker(shareinput_mutator_xslice_2, (Node *) subplan, ctxt);
}
shareinput_walker(shareinput_mutator_xslice_2, (Node *) plan, ctxt);
foreach (lp, glob->subplans)
{
Plan *subplan = (Plan *) lfirst(lp);
shareinput_walker(shareinput_mutator_xslice_3, (Node *) subplan, ctxt);
}
shareinput_walker(shareinput_mutator_xslice_3, (Node *) plan, ctxt);
foreach (lp, glob->subplans)
{
Plan *subplan = (Plan *) lfirst(lp);
shareinput_walker(shareinput_mutator_xslice_4, (Node *) subplan, ctxt);
}
shareinput_walker(shareinput_mutator_xslice_4, (Node *) plan, ctxt);
return plan;
}
/*
* assign_plannode_id - Assign an id for each plan node. Used by gpmon.
*/
void assign_plannode_id(PlannedStmt *stmt)
{
ApplyShareInputContext ctxt;
/* We only care about planNodes and nextPlanId, so initialize them. */
ctxt.planNodes = NIL;
ctxt.nextPlanId = 0;
leftdeep_walker(assign_plannode_id_walker, (Node *) stmt->planTree, &ctxt);
ctxt.planNodes = NIL;
ListCell *lc = NULL;
foreach (lc, stmt->subplans)
{
Plan *subplan = lfirst(lc);
Assert(subplan);
ctxt.planNodes = NIL;
leftdeep_walker(assign_plannode_id_walker, (Node *) subplan, &ctxt);
}
}
/*
* We can "zap" a Result node if it has an Append node as its left tree and
* no other complications (initplans, etc.) and its target list is
* composed of nothing but Var nodes.
*/
static bool can_zap_result(Result *res)
{
ListCell *lc;
if(!res->plan.lefttree
|| res->plan.initPlan
|| !IsA(res->plan.lefttree, Append)
|| res->resconstantqual
|| res->hashFilter
)
return false;
foreach(lc, res->plan.targetlist)
{
TargetEntry *te = (TargetEntry *) lfirst(lc);
if(!IsA(te->expr, Var))
return false;
}
return true;
}
/* "Zap" eligible Result nodes in the given plan by pushing their quals and
* targets down onto their Append argument, marking the Append node isZapped,
* and using it in place of the Result node.
*
* XXX Note the call to planner_subplan_put_plan which modifies the
* subplan's plan. This is unfortunate. We should look for a
* better way.
*/
Plan *zap_trivial_result(PlannerInfo *root, Plan *plan)
{
if(plan == NULL)
return NULL;
if(IsA(plan, SubPlan))
{
SubPlan *subplan = (SubPlan *) plan;
Plan *subplan_plan = planner_subplan_get_plan(root, subplan);
planner_subplan_put_plan(root, subplan, zap_trivial_result(root, subplan_plan));
return plan;
}
if(IsA(plan, List))
{
List *l = (List *) plan;
ListCell *lc;
foreach(lc, l)
{
lfirst(lc) = zap_trivial_result(root, lfirst(lc));
}
return plan;
}
if(IsA(plan, Append))
{
Append *app = (Append *) plan;
app->appendplans = (List *) zap_trivial_result(root, (Plan *) app->appendplans);
return plan;
}
if(IsA(plan, SubqueryScan))
{
SubqueryScan *subq = (SubqueryScan *) plan;
subq->subplan = zap_trivial_result(root, subq->subplan);
return plan;
}
if(IsA(plan, Result))
{
Result *res = (Result *) plan;
plan->lefttree = zap_trivial_result(root, plan->lefttree);
if(can_zap_result(res))
{
Append *app = (Append *) plan->lefttree;
app->plan.targetlist = plan->targetlist;
app->plan.qual = plan->qual;
app->isZapped = true;
return (Plan *) app;
}
else
return plan;
}
plan->lefttree = zap_trivial_result(root, plan->lefttree);
plan->righttree = zap_trivial_result(root, plan->righttree);
plan->initPlan = (List *) zap_trivial_result(root, (Plan *) plan->initPlan);
return plan;
}
/*
* Hash a const value with GPDB's hash function
*/
int32
cdbhash_const(Const *pconst, int iSegments)
{
CdbHash *pcdbhash = makeCdbHash(iSegments, HASH_FNV_1);
cdbhashinit(pcdbhash);
if (pconst->constisnull)
{
cdbhashnull(pcdbhash);
}
else
{
Assert(isGreenplumDbHashable(pconst->consttype));
Oid oidType = pconst->consttype;
if (typeIsArrayType(oidType))
{
oidType = ANYARRAYOID;
}
cdbhash(pcdbhash, pconst->constvalue, oidType);
}
return cdbhashreduce(pcdbhash);
}
/*
* Hash a list of const values with GPDB's hash function
*/
int32
cdbhash_const_list(List *plConsts, int iSegments)
{
Assert(0 < list_length(plConsts));
CdbHash *pcdbhash = makeCdbHash(iSegments, HASH_FNV_1);
cdbhashinit(pcdbhash);
ListCell *lc = NULL;
foreach(lc, plConsts)
{
Const *pconst = (Const *) lfirst(lc);
Assert(IsA(pconst, Const));
if (pconst->constisnull)
{
cdbhashnull(pcdbhash);
}
else
{
Assert(isGreenplumDbHashable(pconst->consttype));
Oid oidType = pconst->consttype;
if (typeIsArrayType(oidType))
{
oidType = ANYARRAYOID;
}
cdbhash(pcdbhash, pconst->constvalue, oidType);
}
}
return cdbhashreduce(pcdbhash);
}