Google Git
Sign in
apache / hawq / bd6196f487c8028f3eca0f56509fe9d718ee31c6 / . / src / backend / cdb / cdbgroup.c
blob: e049bc6b0c054aeda0c0d903a520c57a52deb543 [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.
*/
/*-------------------------------------------------------------------------
*
* cdbgroup.c
* Routines to aid in planning grouping queries for parallel
* execution. This is, essentially, an extension of the file
* optimizer/prep/planner.c, although some functions are not
* externalized.
*
* Portions Copyright (c) 2005-2008, Greenplum inc
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include <limits.h>
#include "catalog/pg_operator.h"
#include "catalog/pg_type.h"
#include "executor/executor.h"
#include "executor/nodeAgg.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/print.h"
#include "optimizer/clauses.h"
#include "optimizer/cost.h"
#include "optimizer/pathnode.h"
#include "optimizer/paths.h"
#include "optimizer/planmain.h"
#include "optimizer/planner.h"
#include "optimizer/planshare.h"
#include "optimizer/prep.h"
#include "optimizer/subselect.h"
#include "optimizer/tlist.h"
#include "optimizer/var.h"
#include "parser/parsetree.h"
#include "parser/parse_clause.h"
#include "parser/parse_coerce.h"
#include "parser/parse_expr.h"
#include "parser/parse_oper.h"
#include "parser/parse_relation.h"
#include "utils/lsyscache.h"
#include "utils/selfuncs.h"
#include "utils/syscache.h"
#include "catalog/pg_aggregate.h"
#include "cdb/cdbllize.h"
#include "cdb/cdbpathtoplan.h" /* cdbpathtoplan_create_flow() */
#include "cdb/cdbpath.h"
#include "cdb/cdbpullup.h"
#include "cdb/cdbvars.h"
#include "cdb/cdbhash.h" /* isGreenplumDbHashable() */
#include "cdb/cdbsetop.h"
#include "cdb/cdbgroup.h"
extern void UpdateScatterClause(Query *query, List *newtlist);
/*
* MppGroupPrep represents a strategy by which to precondition the
* argument to a parallel aggregation plan.
*
* MPP_GRP_PREP_NONE
* Use the input plan as is.
* MPP_GRP_PREP_HASH_GROUPS
* Redistribute the input to collocate groups.
* MPP_GRP_PREP_HASH_DISTINCT
* Redistribute the input on the sole distinct expression used as
* an aggregate argument.
* MPP_GRP_PREP_FOCUS_QE
* Focus a partitioned input on the utility QE.
* MPP_GRP_PREP_FOCUS_QD
* Focus a partitioned input on the QD.
* MPP_GRP_PREP_BROADCAST
* Broadcast to convert a partitioned input into a replicated one.
*/
typedef enum MppGroupPrep
{
MPP_GRP_PREP_NONE = 0,
MPP_GRP_PREP_HASH_GROUPS,
MPP_GRP_PREP_HASH_DISTINCT,
MPP_GRP_PREP_FOCUS_QE,
MPP_GRP_PREP_FOCUS_QD,
MPP_GRP_PREP_BROADCAST
} MppGroupPrep;
/*
* MppGroupType represents a stategy by which to implement parallel
* aggregation on a possibly preconditioned input plan.
*
* MPP_GRP_TYPE_NONE
* No parallel plan found.
* MPP_GRP_TYPE_BASEPLAN
* Use the sequential plan as is.
* MPP_GRP_TYPE_PLAIN_2STAGE,
* Ungrouped, 2-stage aggregation.
* MPP_GRP_TYPE_GROUPED_2STAGE,
* Grouped, 2-stage aggregation.
* MPP_GRP_TYPE_PLAIN_DQA_2STAGE,
* Ungrouped, 2-stage aggregation.
* MPP_GRP_TYPE_GROUPED_DQA_2STAGE,
* Grouped, 2-stage aggregation.
*
* TODO: Add types for min-max optimization, when ready:
*
* MPP_GRP_TYPE_MINMAX
* Use the sequential min-max optimization plan as is.
* MPP_GRP_TYPE_MINMAX_2STAGE
* Use a 2-stage variant of min-max aggregation.
*/
typedef enum MppGroupType
{
MPP_GRP_TYPE_NONE = 0,
MPP_GRP_TYPE_BASEPLAN,
MPP_GRP_TYPE_PLAIN_2STAGE,
MPP_GRP_TYPE_GROUPED_2STAGE,
MPP_GRP_TYPE_PLAIN_DQA_2STAGE,
MPP_GRP_TYPE_GROUPED_DQA_2STAGE,
} MppGroupType;
/* Summary values detailing the post-Motion part of a coplan for
* three-phase aggreation. The code is as follows:
* S = Sort
* G = GroupAgg
* H = HashAgg
* P = PlainAgg
*
* So GSH means (GroupAgg (Sort (HashAgg X))).
*/
typedef enum DqaCoplanType
{
DQACOPLAN_GGS,
DQACOPLAN_GSH,
DQACOPLAN_SHH,
DQACOPLAN_HH,
DQACOPLAN_PGS,
DQACOPLAN_PH
} DqaCoplanType;
typedef enum DqaJoinStrategy
{
DqaJoinUndefined = 0,
DqaJoinNone, /* No join required for solitary DQA argument. */
DqaJoinCross, /* Scalar aggregation uses cross product. */
DqaJoinHash, /* Hash join (possibly with subsequent sort) */
DqaJoinMerge, /* Merge join */
/* These last are abstract and will be replaced
* by DqaJoinHash aor DqaJoinMerge once planning
* is complete.
*/
DqaJoinSorted, /* Sorted output required. */
DqaJoinCheapest, /* No sort requirement. */
} DqaJoinStrategy;
/* DQA coplan information */
typedef struct DqaInfo
{
Node *distinctExpr; /* By reference from agg_counts for convenience. */
AttrNumber base_index; /* Index of attribute in base plan targetlist */
bool can_hash;
double num_rows; /* Estimated cardinality of grouping key, dqa arg */
Plan *coplan; /* Coplan for this (later this and all prior) coplan */
Query *parse; /* Plausible root->parse for the coplan. */
bool distinctkey_collocate; /* Whether the input plan collocates on this
* distinct key */
/* These fields are for costing and planning. Before constructing
* the coplan for this DQA argument, determine cheapest way to get
* the answer and cheapest way to get the answer in grouping key
* order.
*/
bool use_hashed_preliminary;
Cost cost_sorted;
DqaCoplanType coplan_type_sorted;
Cost cost_cheapest;
DqaCoplanType coplan_type_cheapest;
} DqaInfo;
/* Information about the overall plan for a one-, two- or one coplan of
* a three-phase aggregation.
*/
typedef struct AggPlanInfo
{
/*
* The input is either represented as a Path or a Plan and a Path.
* If input_plan is given, use this plan instead of creating one
* through input_path.
* */
Path *input_path;
Plan *input_plan;
/* These are the ordinary fields characterizing an aggregation */
CdbPathLocus input_locus;
MppGroupPrep group_prep;
MppGroupType group_type;
CdbPathLocus output_locus;
bool distinctkey_collocate; /* Whether the input plan collocates on the
* distinct key */
/* These are extra for 3-phase plans */
DqaJoinStrategy join_strategy;
bool use_sharing;
/* These summarize the status of the structure's cost value. */
bool valid;
Cost plan_cost;
} AggPlanInfo;
typedef struct MppGroupContext
{
MppGroupPrep prep;
MppGroupType type;
List *tlist; /* The preprocessed targetlist of the original query. */
Node *havingQual; /* The proprocessed having qual of the original query. */
Path *best_path;
Path *cheapest_path;
Plan *subplan;
AggClauseCounts *agg_counts;
double tuple_fraction;
double *p_dNumGroups; /* Group count estimate shared up the call tree. */
CanonicalGroupingSets *canonical_grpsets;
int64 grouping; /* the GROUPING value */
bool is_grpext; /* identify if this is a grouping extension query */
List *sub_tlist; /* Derived (in cdb_grouping_planner) input targetlist. */
int numGroupCols;
AttrNumber *groupColIdx;
int numDistinctCols;
AttrNumber *distinctColIdx;
DqaInfo *dqaArgs;
bool use_hashed_grouping;
CdbPathLocus input_locus;
CdbPathLocus output_locus;
/* Indicate whether the input plan collocates on the distinct key if any.
* It is used for one or two-phase aggregation. For three-phase aggregation,
* distinctkey_collocate inside DqaInfo is used.
*/
bool distinctkey_collocate;
List *current_pathkeys;
/* Indicate if root->parse has been changed during planning. Carry in pointer
* to root for miscellaneous globals.
*/
bool querynode_changed;
PlannerInfo *root;
/* Work space for aggregate/tlist deconstruction and reconstruction */
Index final_varno; /* input */
bool use_irefs_tlist; /* input */
bool use_dqa_pruning; /* input */
List *prefs_tlist; /* Aggref attributes for prelim_tlist */
List *irefs_tlist; /* Aggref attributes for optional inter_tlist */
List *frefs_tlist; /* Aggref attributes for optional join tlists */
List *dqa_tlist; /* DQA argument attributes for prelim_tlist */
List **dref_tlists; /* Array of DQA Aggref tlists (dqa_tlist order) */
List *grps_tlist; /* Grouping attributes for prelim_tlist */
List *fin_tlist; /* Final tlist cache. */
List *fin_hqual; /* Final having qual cache. */
Index split_aggref_sortgroupref; /* for TargetEntrys made in split_aggref */
Index outer_varno; /* work */
Index inner_varno; /* work */
int *dqa_offsets; /* work */
List *top_tlist; /* work - the target list to finalize */
/* 3-phase DQA decisions */
DqaJoinStrategy join_strategy;
bool use_sharing;
List *wagSortClauses; /* List of List; within-agg multi sort level */
} MppGroupContext;
/* Constants for aggregation approaches.
*/
#define AGG_NONE 0x00
#define AGG_1PHASE 0x01
#define AGG_2PHASE 0x02
#define AGG_2PHASE_DQA 0x04
#define AGG_3PHASE 0x08
#define AGG_SINGLEPHASE (AGG_1PHASE)
#define AGG_MULTIPHASE (AGG_2PHASE | AGG_2PHASE_DQA | AGG_3PHASE)
#define AGG_ALL (AGG_SINGLEPHASE | AGG_MULTIPHASE)
/* Constants for DQA pruning:
*/
static const Index grp_varno = 1; /* var refers to grps_tlist */
static const Index ref_varno = 2; /* var refers to prefs_tlist or relatives */
static const Index dqa_base_varno = 3; /* refers to one of the dref_tlists */
/* Coefficients for cost calculation adjustments: These are candidate GUCs
* or, perhaps, replacements for the gp_eager_... series. We wouldn't
* need these if our statistics and cost calculations were correct, but
* as of 3.2, they not.
*
* Early testing suggested that (1.0, 0.45, 1.7) was about right, but the
* risk of introducing skew in the initial redistribution of a 1-phase plan
* is great (especially given the 3.2 tendency to way underestimate the
* cardinality of joins), so we penalize 1-phase and normalize to the
* 2-phase cost (approximately).
*/
static const double gp_coefficient_1phase_agg = 20.0; /* penalty */
static const double gp_coefficient_2phase_agg = 1.0; /* normalized */
static const double gp_coefficient_3phase_agg = 3.3; /* increase systematic under estimate */
/* Forward declarations */
static Plan * make_one_stage_agg_plan(PlannerInfo *root, MppGroupContext *ctx);
static Plan * make_two_stage_agg_plan(PlannerInfo *root, MppGroupContext *ctx);
static Plan * make_three_stage_agg_plan(PlannerInfo *root, MppGroupContext *ctx);
static Plan * make_plan_for_one_dqa(PlannerInfo *root, MppGroupContext *ctx,
int dqa_index, Plan* result_plan,
Query** coquery_p);
static Plan * join_dqa_coplan(PlannerInfo *root, MppGroupContext *ctx, Plan *plan, int dqa_index);
static int compareDqas(const void *larg, const void *rarg);
static void planDqaJoinOrder(PlannerInfo *root, MppGroupContext *ctx,
double input_rows);
static List *make_subplan_tlist(List *tlist, Node *havingQual,
List *grp_clauses, int *pnum_gkeys, AttrNumber **pcols_gkeys,
List *dqa_args, int *pnum_dqas, AttrNumber **pcols_dqas);
static List *describe_subplan_tlist(List *sub_tlist,
List *tlist, Node *havingQual,
List *grp_clauses, int *pnum_gkeys, AttrNumber **pcols_gkeys,
List *dqa_args, int *pnum_dqas, AttrNumber **pcols_dqas);
static void generate_multi_stage_tlists(MppGroupContext* ctx,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist,
List **p_final_qual);
static void prepare_dqa_pruning_tlists(MppGroupContext *ctx);
static void generate_dqa_pruning_tlists(MppGroupContext *ctx,
int dqa_index,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist,
List **p_final_qual);
static void deconstruct_agg_info(MppGroupContext *ctx);
static void reconstruct_agg_info(MppGroupContext *ctx,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist,
List **p_final_qual);
static void reconstruct_coplan_info(MppGroupContext *ctx,
int dqa_index,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist);
static Expr *deconstruct_expr(Expr *expr, MppGroupContext *ctx);
static Node* deconstruct_expr_mutator(Node *node, MppGroupContext *ctx);
static Node *split_aggref(Aggref *aggref, MppGroupContext *ctx);
static List *make_vars_tlist(List *tlist, Index varno, AttrNumber offset);
static Plan* add_subqueryscan(PlannerInfo* root, List **p_pathkeys,
Index varno, Query *subquery, Plan *subplan);
static List *seq_tlist_concat(List *tlist1, List *tlist2);
static Node *finalize_split_expr(Node *expr, MppGroupContext *ctx);
static Node* finalize_split_expr_mutator(Node *node, MppGroupContext *ctx);
static Oid lookup_agg_transtype(Aggref *aggref);
static bool hash_safe_type(Oid type);
static bool sorting_prefixes_grouping(PlannerInfo *root);
static bool gp_hash_safe_grouping(PlannerInfo *root);
static Cost cost_common_agg(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info, Plan *dummy);
static Cost cost_1phase_aggregation(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info);
static Cost cost_2phase_aggregation(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info);
static Cost cost_3phase_aggregation(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info);
static void set_cost_of_join_strategies(MppGroupContext *ctx, Cost *hashjoin_cost, Cost *mergejoin_cost);
static void initAggPlanInfo(AggPlanInfo *info, Path *input_path, Plan *input_plan);
static void set_coplan_strategies(PlannerInfo *root, MppGroupContext *ctx, DqaInfo *dqaArg, Path *input);
static Cost incremental_sort_cost(double rows, int width, int numKeyCols);
static Cost incremental_agg_cost(double rows, int width, AggStrategy strategy,
int numGroupCols, double numGroups,
int numAggs, int transSpace);
static Cost incremental_motion_cost(double sendrows, double recvrows);
/*---------------------------------------------
* WITHIN/Percentile stuff
*---------------------------------------------*/
/*
* WithinAggContext is a variable set used throughout plan_within_agg_persort().
*/
typedef struct
{
bool use_deduplicate; /* true to choose deduplicate strategy */
AttrNumber pc_pos; /* resno for peer count in outer tlist */
AttrNumber tc_pos; /* resno for total count in inner tlist */
List *current_pathkeys; /* pathkeys tracking */
List *inner_pathkeys; /* pathkeys for inner plan */
List *rtable; /* outer/inner RTE of the output */
} WithinAggContext;
static bool choose_deduplicate(PlannerInfo *root, List *sortExprs,
Plan *input_plan, double *numGroups);
static Plan *wrap_plan_index(PlannerInfo *root, Plan *plan, Query *query,
List **p_pathkeys, Index varno, const char *alias_name, Query **query_p);
static void rebuild_simple_rel_and_rte(PlannerInfo *root);
static Plan *make_parallel_or_sequential_agg(PlannerInfo *root,
AggClauseCounts *agg_count, GroupContext *group_context,
List **current_pathkeys_p);
static Node *deconstruct_within_agg(Node *node, MppGroupContext *ctx);
static Node *deconstruct_within_agg_mutator(Node *node, MppGroupContext *ctx);
static List *fetch_percentiles(Query *parse, List *sortClause);
static Plan *make_deduplicate_plan(PlannerInfo *root, GroupContext *group_context,
List *groupClause, List *sortClause, double numGroups,
AttrNumber *pc_pos_p, List **current_pathkeys_p, Plan *subplan);
static Plan *within_agg_make_baseplan(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
List *sortClause,
Plan *result_plan);
static Plan *within_agg_add_outer_sort(PlannerInfo *root,
WithinAggContext *wag_context,
List *sortClause,
Plan *outer_plan);
static Plan *within_agg_construct_inner(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
Plan *inner_plan);
static Plan *within_agg_join_plans(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
Plan *outer_plan,
Plan *inner_plan);
static Plan *within_agg_final_agg(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
List *sortClause,
Plan *result_plan);
static Plan *plan_within_agg_persort(PlannerInfo *root, GroupContext *group_context,
List *sortClause, List *current_pathkeys, Plan *base_plan);
/*
* add_motion_to_dqa_plan
* Add a Redistribute motion to a dqa child plan if the plan is not already
* distributed on the grouping columns
*/
static Plan *add_motion_to_dqa_child(Plan *plan, PlannerInfo *root, bool *motion_added);
/*
* Function: cdb_grouping_planner
*
* This is basically an extension of the function grouping_planner() from
* planner.c. It (conditionally) replaces the logic in the "normal case"
* (the aggregation/grouping branch) of the main "if" statement.
*
* The result is a Plan for one-, two-, or three-phase grouping/aggregation
* (possibly including a top-level join in case of DQA pruning) or NULL.
*
* A NULL result means that the ordinary sequential planner will produce
* a correct (and preferred) result, so the "normal case" code should be
* used.
*
* A non-NULL result is taken as a partially formed Plan to be processed
* by the trailing sort/distinct/limit logic of grouping_planner(). In
* otherwords, after any associated changes to the local environment (see
* the calling code), the resulting plan should be treated as if from the
* "normal case" or the function optimize_minmax_aggregates().
*
* The pointer at location pcurrent_pathkeys is updated to refer to the
* ordering pathkey or NIL, if none. The parse tree at root->parse may
* be modified in place to reflect changes in the context (e.g. current
* range table).
*/
Plan *
cdb_grouping_planner(PlannerInfo* root,
AggClauseCounts *agg_counts,
GroupContext *group_context)
{
MppGroupContext ctx;
Plan * result_plan = NULL;
List * sub_tlist = NIL;
bool has_groups = root->parse->groupClause != NIL;
bool has_aggs = agg_counts->numAggs > 0;
bool has_ordered_aggs = list_length(agg_counts->aggOrder) > 0;
ListCell *lc;
bool is_grpext = false;
unsigned char consider_agg = AGG_NONE;
AggPlanInfo plan_1p;
AggPlanInfo plan_2p;
AggPlanInfo plan_3p;
AggPlanInfo *plan_info = NULL;
Assert( !has_groups || root->group_pathkeys != NULL );
memset(&ctx, 0, sizeof(ctx));
*(group_context->querynode_changed) = false;
/* We always use sequential plans for distinct-qualified rollup queries,
* so don't waste time working on alternatives.
*/
is_grpext = is_grouping_extension(group_context->canonical_grpsets);
if ( is_grpext && agg_counts->numDistinctAggs > 0)
return NULL;
/*
* First choose a one-stage plan. Since there's always a way to do this,
* it serves as our default choice.
*/
if (group_context->subplan == NULL)
{
Path *input_path = group_context->cheapest_path;
/* Should we prefer the "best" path? Only for vector aggregation
* of input already sorted and collocated on the grouping key.
*/
if ( has_groups &&
pathkeys_contained_in(root->group_pathkeys, group_context->best_path->pathkeys) &&
cdbpathlocus_collocates(group_context->best_path->locus, root->group_pathkeys, false /*exact_match*/) )
{
input_path = group_context->best_path;
}
initAggPlanInfo(&plan_1p, input_path, group_context->subplan);
}
else
{
initAggPlanInfo(&plan_1p, group_context->best_path, group_context->subplan);
plan_1p.input_locus = group_context->best_path->locus;
}
if ( ! CdbPathLocus_IsPartitioned(plan_1p.input_locus) )
{
/* Can use base plan with no motion yielding same locus. */
plan_1p.group_prep = MPP_GRP_PREP_NONE;
plan_1p.output_locus = plan_1p.input_locus;
plan_1p.distinctkey_collocate = true;
}
else if ( has_groups ) /* and not single or replicated */
{
if (root->group_pathkeys != NULL &&
cdbpathlocus_collocates(plan_1p.input_locus, root->group_pathkeys, false /*exact_match*/) )
{
plan_1p.group_prep = MPP_GRP_PREP_NONE;
plan_1p.output_locus = plan_1p.input_locus; /* may be less discriminating that group locus */
plan_1p.distinctkey_collocate = true;
}
else
{
if (gp_hash_safe_grouping(root))
{
plan_1p.group_prep = MPP_GRP_PREP_HASH_GROUPS;
CdbPathLocus_MakeHashed(&plan_1p.output_locus, root->group_pathkeys);
}
else
{
plan_1p.group_prep = MPP_GRP_PREP_FOCUS_QE;
CdbPathLocus_MakeSingleQE(&plan_1p.output_locus);
}
}
}
else if ( has_aggs ) /* and not grouped and not single or replicated */
{
plan_1p.group_prep = MPP_GRP_PREP_FOCUS_QE;
CdbPathLocus_MakeSingleQE(&plan_1p.output_locus);
}
/*
* If the sequential planner can handle the situation with no Motion
* involved, let it do so. Don't bother to investigate the 2-stage
* approach.
*
* If the GUC enable_groupagg is set to off and this is a DQA
* query, we won't use the sequential plan. This is because
* the sequential plan for a DQA query always uses GroupAgg.
*/
if ( plan_1p.group_prep == MPP_GRP_PREP_NONE )
{
if (enable_groupagg || agg_counts->numDistinctAggs == 0)
{
*(group_context->pcurrent_pathkeys) = NIL;
return NULL;
}
}
/*
* When an input plan is given, use it, including its target
* list. When an input target list (and no plan) is given,
* use it for the plan to be created. When neither is given,
* generate a phase 1 target list for the plan to be created.
* Also note the location of any grouping attributes in the
* target list (numGroupCols, groupColIdx).
*
* Also make sure there's a target entry with a non-zero
* sortgroupref for each DQA argument and note the location
* of the attributes (numDistinctCols, distinctColIdx).
*/
if ( group_context->subplan != NULL)
{
sub_tlist = group_context->subplan->targetlist;
}
else if ( group_context->sub_tlist != NULL )
{
sub_tlist = group_context->sub_tlist;
sub_tlist = describe_subplan_tlist(sub_tlist,
group_context->tlist,
root->parse->havingQual,
root->parse->groupClause,
&(group_context->numGroupCols),
&(group_context->groupColIdx),
agg_counts->dqaArgs,
&(group_context->numDistinctCols),
&(group_context->distinctColIdx));
}
else
{
sub_tlist = make_subplan_tlist(group_context->tlist,
root->parse->havingQual,
root->parse->groupClause,
&(group_context->numGroupCols),
&(group_context->groupColIdx),
agg_counts->dqaArgs,
&(group_context->numDistinctCols),
&(group_context->distinctColIdx));
/* Where we need to and we can, add column names to the sub_tlist
* entries to make EXPLAIN output look nice. Note that we could dig
* further than this (if we come up empty handed) by probing the range
* table (root->parse->rtable), but this covers the ordinary cases.
*/
foreach(lc, sub_tlist)
{
TargetEntry *tle = (TargetEntry*)lfirst(lc);
/*
* the first tlist whose expression is equal() to the expression of tle
* may not be the desired one. We should also check if ressortgroupref is
* same if ressortgroupref > 0
*/
if ( IsA(tle->expr, Var) && tle->resname == NULL )
{
TargetEntry *vartle =
tlist_member_with_ressortgroupref((Node*)tle->expr,
group_context->tlist, tle->ressortgroupref);
if ( vartle != NULL && vartle->resname != NULL )
tle->resname = pstrdup(vartle->resname);
}
}
}
/* At this point, we're committed to producing a one- , two- or
* three-stage plan with motion. Determine what aggregation approaches to explore.
* Per MPP-2378, we don't insist on has_aggs for multi-phase
* plans.
*/
{
unsigned char allowed_agg;
unsigned char possible_agg;
allowed_agg = AGG_ALL;
if ( ! root->config->gp_enable_multiphase_agg )
allowed_agg &= AGG_SINGLEPHASE;
/* This prohibition could be relaxed if we tracked missing
* preliminary functions per DQA and were willing to plan
* some DQAs as single and some as multiple phases. Not
* currently, however.
*/
if ( agg_counts->missing_prelimfunc )
allowed_agg &= ~ AGG_MULTIPHASE;
/*
* Ordered aggregates need to run the transition function on the
* values in sorted order, which in turn translates into single
* phase aggregation.
*/
if ( has_ordered_aggs )
allowed_agg &= ~ AGG_MULTIPHASE;
/* We are currently unwilling to redistribute a gathered
* intermediate across the cluster. This might change
* one day.
*/
if ( ! CdbPathLocus_IsPartitioned(plan_1p.input_locus ) )
allowed_agg &= AGG_SINGLEPHASE;
if ( ! root->config->gp_enable_agg_distinct )
allowed_agg &= ~ AGG_2PHASE_DQA;
if ( ! root->config->gp_enable_dqa_pruning )
allowed_agg &= ~ AGG_3PHASE;
possible_agg = AGG_SINGLEPHASE;
if(gp_hash_safe_grouping(root))
{
switch ( list_length(agg_counts->dqaArgs) )
{
case 0:
possible_agg |= AGG_2PHASE;
break;
case 1:
possible_agg |= AGG_2PHASE_DQA | AGG_3PHASE;
break;
default: /* > 1 */
possible_agg |= AGG_3PHASE;
break;
}
}
if ( is_grpext )
possible_agg &= ~ (AGG_2PHASE_DQA | AGG_3PHASE);
consider_agg = allowed_agg & possible_agg;
}
Assert( consider_agg & AGG_1PHASE ); /* Always possible! */
if ( consider_agg & ( AGG_2PHASE | AGG_2PHASE_DQA ) )
{
/* XXX initAggPlanInfo(&plan_2p, group_context->cheapest_path); */
initAggPlanInfo(&plan_2p, group_context->best_path,
group_context->subplan); /* but why? */
/* Common 2-phase setup. */
if ( has_groups )
{
plan_2p.group_type = MPP_GRP_TYPE_GROUPED_2STAGE;
CdbPathLocus_MakeHashed(&plan_2p.output_locus, root->group_pathkeys);
}
else
{
plan_2p.group_type = MPP_GRP_TYPE_PLAIN_2STAGE;
CdbPathLocus_MakeSingleQE(&plan_2p.output_locus);
}
if ( consider_agg & AGG_2PHASE_DQA )
{
List *distinct_pathkeys;
/* Either have DQA or not! */
Assert(! (consider_agg & AGG_2PHASE) );
Insist( IsA(agg_counts->dqaArgs, List) &&
list_length((List*)agg_counts->dqaArgs) == 1 );
distinct_pathkeys = cdb_make_pathkey_for_expr(root,
linitial(agg_counts->dqaArgs),
list_make1(makeString("=")));
distinct_pathkeys = list_make1(distinct_pathkeys);
if ( ! cdbpathlocus_collocates(plan_2p.input_locus, distinct_pathkeys, false /*exact_match*/))
{
plan_2p.group_prep = MPP_GRP_PREP_HASH_DISTINCT;
CdbPathLocus_MakeHashed(&plan_2p.input_locus, distinct_pathkeys);
}
else
{
plan_2p.group_prep = MPP_GRP_PREP_HASH_DISTINCT;
plan_2p.output_locus = plan_2p.input_locus;
plan_2p.distinctkey_collocate = true;
}
}
}
if ( consider_agg & AGG_3PHASE )
{
initAggPlanInfo(&plan_3p, group_context->cheapest_path,
group_context->subplan);
if ( has_groups )
{
plan_3p.group_type = MPP_GRP_TYPE_GROUPED_DQA_2STAGE;
CdbPathLocus_MakeHashed(&plan_3p.output_locus, root->group_pathkeys);
}
else
{
plan_3p.group_type = MPP_GRP_TYPE_PLAIN_DQA_2STAGE;
CdbPathLocus_MakeSingleQE(&plan_3p.output_locus);
}
}
/*
* Package up byproducts for the actual planner.
*/
ctx.prep = plan_1p.group_prep;
ctx.type = plan_1p.group_type;
ctx.tlist = group_context->tlist;
ctx.havingQual = root->parse->havingQual;
ctx.sub_tlist = sub_tlist;
ctx.numGroupCols = group_context->numGroupCols;
ctx.groupColIdx = group_context->groupColIdx;
ctx.numDistinctCols = group_context->numDistinctCols;
ctx.distinctColIdx = group_context->distinctColIdx;
ctx.use_hashed_grouping = group_context->use_hashed_grouping;
ctx.best_path = group_context->best_path;
ctx.cheapest_path = group_context->cheapest_path;
ctx.subplan = group_context->subplan;
ctx.input_locus = plan_1p.input_locus;
ctx.output_locus = plan_1p.output_locus;
ctx.distinctkey_collocate = plan_1p.distinctkey_collocate;
ctx.agg_counts = agg_counts;
ctx.tuple_fraction = group_context->tuple_fraction;
ctx.p_dNumGroups = group_context->p_dNumGroups;
ctx.canonical_grpsets = group_context->canonical_grpsets;
ctx.grouping = group_context->grouping;
ctx.is_grpext = is_grpext;
ctx.current_pathkeys = NIL; /* Initialize to be tidy. */
ctx.querynode_changed = false;
ctx.root = root;
/* If we're to consider 3-phase plans, do some preparation.
*/
if ( ctx.numDistinctCols > 0 && (consider_agg & AGG_3PHASE) )
{
int i;
/* Collect row count estimates and other info for the partial
* results of grouping over combined grouping and distinct (DQA)
* keys. Order the output array of DqaInfo structures (in the
* context) according to how they should be joined.
*/
planDqaJoinOrder(root, &ctx, plan_3p.input_path->parent->rows);
/* Plan the post-Motion portions of each coplan in two ways: one to
* produce the result in the cheapest way and one to produce the
* result ordered by the grouping key in the cheapest way. (For use
* by make_plan_for_one_dqa called by make_three_stage_agg_plan.)
*/
for ( i = 0; i < ctx.numDistinctCols; i++ )
{
List *distinct_pathkeys;
set_coplan_strategies(root, &ctx, &ctx.dqaArgs[i], plan_3p.input_path);
/* Determine if the input plan already collocates on the distinct
* key.
*/
distinct_pathkeys = cdb_make_pathkey_for_expr(root,
ctx.dqaArgs[i].distinctExpr,
list_make1(makeString("=")));
distinct_pathkeys = list_make1(distinct_pathkeys);
if (cdbpathlocus_collocates(plan_3p.input_locus, distinct_pathkeys, false /*exact_match*/))
{
ctx.dqaArgs[i].distinctkey_collocate = true;
}
list_free(distinct_pathkeys);
}
}
plan_info = NULL; /* Most cost-effective, feasible plan. */
if ( consider_agg & AGG_1PHASE )
{
cost_1phase_aggregation(root, &ctx, &plan_1p);
if ( gp_dev_notice_agg_cost )
elog(NOTICE, "1-phase cost: %.6f", plan_1p.plan_cost);
if ( plan_info == NULL || plan_info->plan_cost > plan_1p.plan_cost )
plan_info = &plan_1p;
}
if ( consider_agg & ( AGG_2PHASE | AGG_2PHASE_DQA ) )
{
cost_2phase_aggregation(root, &ctx, &plan_2p);
if ( gp_dev_notice_agg_cost )
elog(NOTICE, "2-phase cost: %.6f", plan_2p.plan_cost);
if ( plan_info == NULL || plan_info->plan_cost > plan_2p.plan_cost )
plan_info = &plan_2p;
}
if ( consider_agg & AGG_3PHASE )
{
cost_3phase_aggregation(root, &ctx, &plan_3p);
if ( gp_dev_notice_agg_cost )
elog(NOTICE, "3-phase cost: %.6f", plan_3p.plan_cost);
if ( plan_info == NULL || !enable_groupagg || plan_info->plan_cost > plan_3p.plan_cost )
plan_info = &plan_3p;
}
Insist( plan_info != NULL );
ctx.prep = plan_info->group_prep;
ctx.type = plan_info->group_type;
ctx.input_locus = plan_info->input_locus;
ctx.output_locus = plan_info->output_locus;
ctx.distinctkey_collocate = plan_info->distinctkey_collocate;
ctx.join_strategy = plan_info->join_strategy;
ctx.use_sharing = plan_info->use_sharing;
/* Call appropriate planner. */
if (ctx.type == MPP_GRP_TYPE_BASEPLAN)
{
if (ctx.prep != MPP_GRP_PREP_NONE)
result_plan = make_one_stage_agg_plan(root, &ctx);
else
result_plan = NULL; /* allow sequential planner to do the work. */
}
else if (ctx.type == MPP_GRP_TYPE_PLAIN_2STAGE ||
ctx.type == MPP_GRP_TYPE_GROUPED_2STAGE)
result_plan = make_two_stage_agg_plan(root, &ctx);
else if (ctx.type == MPP_GRP_TYPE_PLAIN_DQA_2STAGE ||
ctx.type == MPP_GRP_TYPE_GROUPED_DQA_2STAGE)
result_plan = make_three_stage_agg_plan(root, &ctx);
else
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("no parallel plan for aggregation")));
if (!is_grpext && result_plan != NULL &&
IsA(result_plan, Agg))
((Agg *)result_plan)->lastAgg = true;
*(group_context->querynode_changed) = ctx.querynode_changed;
*(group_context->pcurrent_pathkeys) = ctx.current_pathkeys;
return result_plan;
}
/*
* Function make_one_stage_agg_plan
*
* Construct a one-stage aggregation plan by redistributing the result
* of the input plan appropriately
*/
static Plan *
make_one_stage_agg_plan(PlannerInfo *root,
MppGroupContext *ctx)
{
Query *parse = root->parse;
List *tlist = ctx->tlist;
List *sub_tlist = ctx->sub_tlist;
int numGroupCols = ctx->numGroupCols;
AttrNumber *groupColIdx = ctx->groupColIdx;
Path *best_path = ctx->best_path;
Path *cheapest_path = ctx->cheapest_path;
Path *path = NULL;
bool use_hashed_grouping = ctx->use_hashed_grouping;
long numGroups = (*(ctx->p_dNumGroups) < 0) ? 0 :
(*(ctx->p_dNumGroups) > LONG_MAX) ? LONG_MAX :
(long)*(ctx->p_dNumGroups);
List *groupExprs = NIL;
List *current_pathkeys;
QualCost tlist_cost;
int i;
Plan *result_plan;
bool is_sorted;
/*
* The argument to the "lower" Agg node will use a "flattened" tlist
* (having just the (levelsup==0) vars mentioned in the SELECT targetlist
* and HAVING qual plus entries for any GROUP BY expressions that are not
* simple Vars. This is the same sub_tlist as that used for 1-stage
* aggregation in grouping_planner.
*/
/* Create the base plan. If the best path is in grouping key order and
* we don't plan to move it around and this is a vector aggregation, we
* should use best path. In other cases, however, use cheapest.
*/
if (ctx->subplan == NULL)
{
is_sorted = pathkeys_contained_in(root->group_pathkeys, best_path->pathkeys);
path = cheapest_path;
if ( is_sorted && ctx->prep == MPP_GRP_PREP_NONE && numGroupCols > 0 )
path = best_path;
result_plan = create_plan(root, path);
current_pathkeys = path->pathkeys;
/* Instead of the flat target list produced above, use the sub_tlist
* constructed in cdb_grouping_planner. Add a Result node if the
* base plan can't project. (This may be unnecessary, but, if so,
* the Result node will be removed later.)
*/
result_plan = plan_pushdown_tlist(result_plan, sub_tlist);
Assert(result_plan->flow);
/* Account for the cost of evaluation of the sub_tlist. */
cost_qual_eval(&tlist_cost, sub_tlist, root);
result_plan->startup_cost += tlist_cost.startup;
result_plan->total_cost +=
tlist_cost.startup +
tlist_cost.per_tuple * result_plan->plan_rows;
}
else
{
result_plan = ctx->subplan;
current_pathkeys = ctx->current_pathkeys;
}
/* Precondition the input by adjusting its locus prior to adding
* the Agg or Group node to the base plan, if needed.
*/
switch ( ctx->prep )
{
case MPP_GRP_PREP_NONE:
break;
case MPP_GRP_PREP_HASH_GROUPS:
Assert(numGroupCols > 0);
for ( i = 0; i < numGroupCols; i++)
{
TargetEntry *tle = get_tle_by_resno(sub_tlist, groupColIdx[i]);
groupExprs = lappend(groupExprs, copyObject(tle->expr));
}
result_plan = (Plan*)make_motion_hash(root, result_plan, groupExprs);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows);
current_pathkeys = NIL; /* No longer sorted. */
break;
case MPP_GRP_PREP_FOCUS_QE:
result_plan = (Plan*)make_motion_gather_to_QE(result_plan, (current_pathkeys != NIL));
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows * root->config->cdbpath_segments);
break;
case MPP_GRP_PREP_FOCUS_QD:
result_plan = (Plan*)make_motion_gather_to_QD(result_plan, (current_pathkeys != NIL));
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows * root->config->cdbpath_segments);
break;
case MPP_GRP_PREP_HASH_DISTINCT:
case MPP_GRP_PREP_BROADCAST:
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("no parallel plan for aggregation")));
break; /* Never */
}
Assert(result_plan->flow);
/*
* Insert AGG or GROUP node if needed, plus an explicit sort step
* if necessary.
*
* HAVING clause, if any, becomes qual of the Agg or Group node.
*/
if (!ctx->is_grpext && use_hashed_grouping)
{
/* Hashed aggregate plan --- no sort needed */
result_plan = (Plan *) make_agg(root,
tlist,
(List *) parse->havingQual,
AGG_HASHED, false,
numGroupCols,
groupColIdx,
numGroups,
0, /* num_nullcols */
0, /* input_grouping */
ctx->grouping,
0, /* rollup_gs_times */
ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace,
result_plan);
/* Hashed aggregation produces randomly-ordered results */
current_pathkeys = NIL;
}
else if (parse->hasAggs || parse->groupClause)
{
if (!ctx->is_grpext)
{
/* Plain aggregate plan --- sort if needed */
AggStrategy aggstrategy;
if (parse->groupClause)
{
if (!pathkeys_contained_in(root->group_pathkeys,
current_pathkeys))
{
result_plan = (Plan *)
make_sort_from_groupcols(root,
parse->groupClause,
groupColIdx,
false,
result_plan);
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
}
aggstrategy = AGG_SORTED;
/*
* The AGG node will not change the sort ordering of its
* groups, so current_pathkeys describes the result too.
*/
}
else
{
aggstrategy = AGG_PLAIN;
/* Result will be only one row anyway; no sort order */
current_pathkeys = NIL;
}
result_plan = (Plan *) make_agg(root,
tlist,
(List *) parse->havingQual,
aggstrategy, false,
numGroupCols,
groupColIdx,
numGroups,
0, /* num_nullcols */
0, /* input_grouping */
ctx->grouping,
0, /* rollup_gs_times */
ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace,
result_plan);
}
else
{
result_plan = plan_grouping_extension(root, path, ctx->tuple_fraction,
ctx->use_hashed_grouping,
&tlist, sub_tlist,
true, false,
(List *) parse->havingQual,
&numGroupCols,
&groupColIdx,
ctx->agg_counts,
ctx->canonical_grpsets,
ctx->p_dNumGroups,
&(ctx->querynode_changed),
&current_pathkeys,
result_plan);
}
}
else if (root->hasHavingQual)
{
/* No aggregates, and no GROUP BY, but a HAVING qual is a
* degenerate case discussed in grouping_planner. We can
* just throw away the plan-so-far and let the caller handle
* the whole enchilada.
*/
return NULL;
}
/*
* Decorate the top node with a Flow node if it doesn't have one yet.
* (In such cases we require the next-to-top node to have a Flow node
* from which we can obtain the distribution info.)
*/
if (!result_plan->flow)
{
Assert(!IsA(result_plan, Motion));
result_plan->flow = pull_up_Flow(result_plan,
result_plan->lefttree,
(current_pathkeys != NIL));
}
/* Marshal implicit results. Return explicit result. */
ctx->current_pathkeys = current_pathkeys;
return result_plan;
}
/*
* Function make_two_stage_agg_plan
*
* Construct a two-stage aggregation plan.
*/
static Plan *
make_two_stage_agg_plan(PlannerInfo *root,
MppGroupContext *ctx)
{
Query *parse = root->parse;
List *prelim_tlist = NIL;
List *final_tlist = NIL;
List *final_qual = NIL;
List *distinctExpr = NIL;
List *groupExprs = NIL;
List *current_pathkeys;
Plan *result_plan;
QualCost tlist_cost;
AggStrategy aggstrategy;
int i;
int numGroupCols;
AttrNumber *groupColIdx;
AttrNumber *prelimGroupColIdx;
Path *path = ctx->best_path; /* no use for ctx->cheapest_path */
long numGroups = (*(ctx->p_dNumGroups) < 0) ? 0 :
(*(ctx->p_dNumGroups) > LONG_MAX) ? LONG_MAX :
(long)*(ctx->p_dNumGroups);
/* Copy these from context rather than using them directly because we may
* scribble on them in plan_grouping_extension(). It would be good to
* clean this up, but not today.
*/
numGroupCols = ctx->numGroupCols;
groupColIdx = ctx->groupColIdx;
/* Create the base plan which will serve as the outer plan (argument)
* of the partial Agg node.
*/
if (ctx->subplan == NULL)
{
result_plan = create_plan(root, path);
current_pathkeys = path->pathkeys;
/* Instead of the flat target list produced by create_plan above, use
* the sub_tlist constructed in cdb_grouping_planner. This consists
* of just the (levelsup==0) vars mentioned in the SELECT and HAVING
* clauses plus entries for any GROUP BY expressions that are not
* simple Vars. (This is the same sub_tlist as used in 1-stage
* aggregation and in normal aggregation in grouping_planner).
*
* If the base plan is of a type that can't project, add a Result
* node to carry the new target list, else install it directly.
* (Though the result node may not always be necessary, it is safe,
* and superfluous Result nodes are removed later.)
*/
result_plan = plan_pushdown_tlist(result_plan, ctx->sub_tlist);
/* Account for the cost of evaluation of the sub_tlist. */
cost_qual_eval(&tlist_cost, ctx->sub_tlist, root);
result_plan->startup_cost += tlist_cost.startup;
result_plan->total_cost +=
tlist_cost.startup +
tlist_cost.per_tuple * result_plan->plan_rows;
}
else
{
result_plan = ctx->subplan;
current_pathkeys = ctx->current_pathkeys;
}
/* At this point result_plan produces the input relation for two-stage
* aggregation.
*
* Begin by preconditioning the input, if necessary, to collocate on
* non-distinct values of a single DISTINCT argument.
*/
switch ( ctx->prep )
{
case MPP_GRP_PREP_NONE:
break;
case MPP_GRP_PREP_HASH_DISTINCT:
Assert(list_length( ctx->agg_counts->dqaArgs) == 1 );
Assert( ctx->agg_counts->dqaArgs != NIL);
if (!ctx->distinctkey_collocate)
{
distinctExpr = list_make1(linitial(ctx->agg_counts->dqaArgs));
distinctExpr = copyObject(distinctExpr);
result_plan = (Plan*)make_motion_hash(root, result_plan, distinctExpr);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows);
current_pathkeys = NIL; /* No longer sorted. */
}
break;
case MPP_GRP_PREP_FOCUS_QD:
case MPP_GRP_PREP_FOCUS_QE:
case MPP_GRP_PREP_HASH_GROUPS:
case MPP_GRP_PREP_BROADCAST:
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("unexpected call for two-stage aggregation")));
break; /* Never */
}
/*
* Get the target lists for the preliminary and final aggregations and
* the qual (HAVING clause) for the final aggregation based on the target
* list of the base plan. Grouping attributes go on front of preliminary
* target list.
*/
generate_multi_stage_tlists(ctx,
&prelim_tlist,
NULL,
&final_tlist,
&final_qual);
/*
* Since the grouping attributes, if any, are on the front and in order
* on the preliminary targetlist, we need a different vector of grouping
* attribute numbers: (1, 2, 3, ...). Later, we'll need
*/
prelimGroupColIdx = NULL;
if ( numGroupCols > 0 )
{
prelimGroupColIdx = (AttrNumber*)palloc(numGroupCols * sizeof(AttrNumber));
for ( i = 0; i < numGroupCols; i++ )
prelimGroupColIdx[i] = i+1;
}
/*
* Add the Preliminary Agg Node.
*
* When this aggregate is a ROLLUP, we add a sequence of preliminary Agg node.
*/
/* Determine the aggregation strategy to use. */
if ( ctx->use_hashed_grouping )
{
aggstrategy = AGG_HASHED;
current_pathkeys = NIL;
}
else
{
if (parse->groupClause)
{
if (!ctx->is_grpext && !pathkeys_contained_in(root->group_pathkeys,
current_pathkeys))
{
/* TODO -- Investigate WHY we might sort here!
*
* Good reasons would be that one of the grouping
* expressions isn't "hashable" or that too may groups
* are anticipated.
*
* A bad reason would be that the final result will be in
* order of the grouping key. (Redistribution will remove
* the ordering.)
*/
result_plan = (Plan *)
make_sort_from_groupcols(root,
parse->groupClause,
groupColIdx,
false,
result_plan);
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
}
aggstrategy = AGG_SORTED;
/* The AGG node will not change the sort ordering of its
* groups, so current_pathkeys describes the result too.
*/
}
else
{
aggstrategy = AGG_PLAIN;
current_pathkeys = NIL; /* One row, no sort order */
}
}
if (!ctx->is_grpext)
{
result_plan = (Plan *) make_agg(root,
prelim_tlist,
NIL, /* no havingQual */
aggstrategy, root->config->gp_hashagg_streambottom,
numGroupCols,
groupColIdx,
numGroups,
0, /* num_nullcols */
0, /* input_grouping */
0, /* grouping */
0, /* rollup_gs_times */
ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace,
result_plan);
/* May lose useful locus and sort. Unlikely, but could do better. */
mark_plan_strewn(result_plan);
current_pathkeys = NIL;
}
else
{
result_plan = plan_grouping_extension(root, path, ctx->tuple_fraction,
ctx->use_hashed_grouping,
&prelim_tlist, ctx->sub_tlist,
true, true,
NIL, /* no havingQual */
&numGroupCols,
&groupColIdx,
ctx->agg_counts,
ctx->canonical_grpsets,
ctx->p_dNumGroups,
&(ctx->querynode_changed),
&current_pathkeys,
result_plan);
/* Since we add Grouping as an additional grouping column,
* we need to add it into prelimGroupColIdx. */
if (prelimGroupColIdx != NULL)
prelimGroupColIdx = (AttrNumber *)
repalloc(prelimGroupColIdx,
numGroupCols * sizeof(AttrNumber));
else
prelimGroupColIdx = (AttrNumber *)
palloc0(numGroupCols * sizeof(AttrNumber));
Assert(numGroupCols >= 2);
prelimGroupColIdx[numGroupCols-1] = groupColIdx[numGroupCols-1];
prelimGroupColIdx[numGroupCols-2] = groupColIdx[numGroupCols-2];
}
/*
* Add Intermediate Motion to Gather or Hash on Groups
*/
switch ( ctx->type )
{
case MPP_GRP_TYPE_GROUPED_2STAGE:
groupExprs = NIL;
Assert(numGroupCols > 0);
for ( i = 0; i < numGroupCols; i++)
{
TargetEntry *tle;
/* skip Grouping/GroupId columns */
if (ctx->is_grpext && (i == numGroupCols-1 || i == numGroupCols-2))
continue;
tle = get_tle_by_resno(prelim_tlist, prelimGroupColIdx[i]);
groupExprs = lappend(groupExprs, copyObject(tle->expr));
}
result_plan = (Plan*)make_motion_hash(root, result_plan, groupExprs);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows);
break;
case MPP_GRP_TYPE_PLAIN_2STAGE:
result_plan = (Plan*)make_motion_gather_to_QE(result_plan, false);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows * root->config->cdbpath_segments);
break;
case MPP_GRP_TYPE_NONE:
case MPP_GRP_TYPE_BASEPLAN:
case MPP_GRP_TYPE_GROUPED_DQA_2STAGE:
case MPP_GRP_TYPE_PLAIN_DQA_2STAGE:
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("unexpected use of 2-stage aggregation")));
break; /* Never */
}
/*
* Add Sort on Groups if needed for AGG_SORTED strategy
*/
if (aggstrategy == AGG_SORTED)
{
result_plan = (Plan *)
make_sort_from_groupcols(root,
parse->groupClause,
prelimGroupColIdx,
ctx->is_grpext,
result_plan);
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
}
result_plan = add_second_stage_agg(root,
true,
prelim_tlist,
final_tlist,
final_qual,
aggstrategy,
numGroupCols,
prelimGroupColIdx,
0, /* num_nullcols */
0, /* input_grouping */
ctx->grouping,
0, /* rollup_gs_times */
*ctx->p_dNumGroups,
ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace,
"partial_aggregation",
&current_pathkeys,
result_plan,
!ctx->is_grpext,
true);
if (ctx->is_grpext)
{
ListCell *lc;
bool found = false;
((Agg *)result_plan)->inputHasGrouping = true;
/*
* We want to make sure that the targetlist of result plan contains
* either GROUP_ID or a targetentry to represent the value of
* GROUP_ID from the subplans. This is because we may need this
* entry to determine if a tuple will be outputted repeatly, by
* the later Repeat node. In the current grouping extension
* planner, if there is no GROUP_ID entry, then it must be the last
* entry in the targetlist of the subplan.
*/
foreach (lc, result_plan->targetlist)
{
TargetEntry *te = (TargetEntry *)lfirst(lc);
/*
* Find out if GROUP_ID in the final targetlist. It should
* point to the last attribute in the subplan targetlist.
*/
if (IsA(te->expr, Var))
{
Var *var = (Var *)te->expr;
if (var->varattno == list_length(prelim_tlist))
{
found = true;
break;
}
}
}
if (!found)
{
/* Add a new target entry in the targetlist which point to
* GROUP_ID attribute in the subplan. Mark this entry
* as Junk.
*/
TargetEntry *te = get_tle_by_resno(prelim_tlist,
list_length(prelim_tlist));
Expr *expr;
TargetEntry *new_te;
expr = (Expr *)makeVar(1,
te->resno,
exprType((Node *)te->expr),
exprTypmod((Node *)te->expr),
0);
new_te = makeTargetEntry(expr,
list_length(result_plan->targetlist) + 1,
"group_id",
true);
result_plan->targetlist = lappend(result_plan->targetlist,
new_te);
}
}
/* Marshal implicit results. Return explicit result. */
ctx->current_pathkeys = current_pathkeys;
ctx->querynode_changed = true;
return result_plan;
}
/*
* Function make_three_stage_agg_plan
*
* Construct a three-stage aggregation plan involving DQAs (DISTINCT-qualified
* aggregate functions.
*
* Such a plan will always involve the following three aggregation phases:
*
* - preliminary -- remove duplicate (grouping key, DQA argument) values
* from an arbitrarily partitioned input; pre-aggregate plain aggregate
* functions.
*
* - intermediate -- remove duplicate (grouping key, DQA argument) values
* from an input partitioned on the grouping key; pre-aggregate the
* pre-aggregated results of preliminary plain aggregate functions.
*
* - final -- apply ordinary aggregation to DQA arguments (now distinct
* within their group) and final aggregation to the pre-aggregated results
* of the previous phase.
*
* In addition, if there is more than one DQA in the query, the plan will
* join the results of the individual three-phase aggregations into the
* final result.
*
* The preliminary aggregation phase occurs prior to the collocating
* motion and is planned independently on the theory that any ordering
* will be disrupted by the motion. There are cases where this isn't
* necessarily the case, but they are unexploited for now.
*
* The intermediate and final aggregation phases...
*/
static Plan *
make_three_stage_agg_plan(PlannerInfo *root, MppGroupContext *ctx)
{
List *current_pathkeys;
Plan *result_plan;
QualCost tlist_cost;
Path *path = ctx->best_path; /* no use for ctx->cheapest_path */
/* We assume that we are called only when
* - there are no grouping extensions (like ROLLUP),
* - the input is partitioned and needs no preparatory Motion,
* - the required transformation involves DQAs.
*/
Assert ( !is_grouping_extension(ctx->canonical_grpsets) );
Assert ( ctx->prep == MPP_GRP_PREP_NONE );
Assert ( ctx->type == MPP_GRP_TYPE_GROUPED_DQA_2STAGE
|| ctx->type == MPP_GRP_TYPE_PLAIN_DQA_2STAGE );
/* Create the base plan which will serve as the outer plan (argument)
* of the partial Agg node(s).
*/
if (ctx->subplan == NULL)
{
result_plan = create_plan(root, path);
current_pathkeys = path->pathkeys;
/* Instead of the flat target list produced above, use the sub_tlist
* constructed in cdb_grouping_planner. Add a Result node if the
* base plan can't project. (This may be unnecessary, but, if so,
* the Result node will be removed later.)
*/
result_plan = plan_pushdown_tlist(result_plan, ctx->sub_tlist);
Assert(result_plan->flow);
/* Account for the cost of evaluation of the sub_tlist. */
cost_qual_eval(&tlist_cost, ctx->sub_tlist, root);
result_plan->startup_cost += tlist_cost.startup;
result_plan->total_cost +=
tlist_cost.startup +
tlist_cost.per_tuple * result_plan->plan_rows;
}
else
{
result_plan = ctx->subplan;
current_pathkeys = ctx->current_pathkeys;
}
/* Use caller specified join_strategy: None, Cross, Hash, or Merge. */
prepare_dqa_pruning_tlists(ctx);
if ( list_length(ctx->agg_counts->dqaArgs) == 1 )
{
/* Note: single-DQA plans don't require a join and are handled
* specially by make_plan_for_one_dqa so we can return the result
* directly.
*/
Query *query;
result_plan = make_plan_for_one_dqa(root, ctx, 0,
result_plan, &query);
memcpy(root->parse, query, sizeof(Query));
pfree(query);
}
else
{
/* Multi-DQA plans are trickier because of the need to consider input
* sharing and the need to join the coplans back together.
*/
List *share_partners;
int i;
List *rtable = NIL;
if ( ctx->use_sharing )
{
share_partners = share_plan(root, result_plan, ctx->numDistinctCols);
}
else
{
share_partners = NIL;
share_partners = lappend(share_partners, result_plan);
for ( i = 1; i < ctx->numDistinctCols; i++ )
{
share_partners = lappend(share_partners, copyObject(result_plan));
}
}
/* Construct a coplan for each distinct DQA argument. */
for ( i = 0; i < ctx->numDistinctCols; i++ )
{
char buffer[50];
int j;
ListCell *l;
Alias *eref;
Plan *coplan;
Query *coquery;
coplan = (Plan*)list_nth(share_partners,i);
coplan = make_plan_for_one_dqa(root, ctx, i,
coplan, &coquery);
eref = makeNode(Alias);
sprintf(buffer, "dqa_coplan_%d", i+1);
eref->aliasname = pstrdup(buffer);
eref->colnames = NIL;
j = 1;
foreach (l, coplan->targetlist)
{
TargetEntry *tle = (TargetEntry*)lfirst(l);
Value *colname = get_tle_name(tle, coquery->rtable, buffer);
eref->colnames = lappend(eref->colnames, colname);
j++;
}
rtable = lappend(rtable,
package_plan_as_rte(coquery, coplan, eref, NIL));
ctx->dqaArgs[i].coplan = add_subqueryscan(root, NULL, i+1, coquery, coplan);
}
/* Begin with the first coplan, then join in each suceeding coplan. */
result_plan = ctx->dqaArgs[0].coplan;
for ( i = 1; i < ctx->numDistinctCols; i++ )
{
result_plan = join_dqa_coplan(root, ctx, result_plan, i);
}
/* Finalize the last join plan so it has the correct target list
* and having qual.
*/
ctx->top_tlist = result_plan->targetlist;
result_plan->targetlist = (List*) finalize_split_expr((Node*) ctx->fin_tlist, ctx);
result_plan->qual = (List*) finalize_split_expr((Node*) ctx->fin_hqual, ctx);
/*
* Reconstruct the flow since the targetlist for the result_plan may have
* changed.
*/
result_plan->flow = pull_up_Flow(result_plan,
result_plan->lefttree,
true);
/* Need to adjust root. Is this enuf? I think so. */
root->parse->rtable = rtable;
root->parse->targetList = copyObject(result_plan->targetlist);
}
// Rebuild arrays for RelOptInfo and RangeTblEntry for the PlannerInfo
// since the underlying range tables have been transformed
rebuild_simple_rel_and_rte(root);
return result_plan;
}
/* Helper for qsort in planDqaJoinOrder. */
int compareDqas(const void *larg, const void *rarg)
{
double lft = ((DqaInfo*)larg)->num_rows;
double rgt = ((DqaInfo*)rarg)->num_rows;
return (lft < rgt)? -1 : (lft == rgt)? 0 : 1;
}
/* Collect per distinct DQA argument information for use in single- and
* multiple-DQA planning and cache it in the context as a new array of
* DqaInfo structures anchored at ctx->dqaArgs. The order of elements
* in the array determines join order for a multi-DQA plan.
*
* Note: The original list of distinct DQA arguments was collected by
* the count_agg_clauses earlier in planning. Later, make_subplan_tlist
* used it to guarantee that the DQA arguments have target entries with
* non-zero sortgroupref values and to generate vector ctx->distinctColIdx
* to locate those entries. Here, however, we use that vector to locate
* the DQA arguments and reorder the vector to agree with join order.
*/
void planDqaJoinOrder(PlannerInfo *root, MppGroupContext *ctx,
double input_rows)
{
int i;
DqaInfo *args;
Node *distinctExpr;
Assert( ctx->numDistinctCols == list_length(ctx->agg_counts->dqaArgs) );
/* Collect row count estimates for the partial results. */
if ( ctx->numDistinctCols == 0 )
{
ctx->dqaArgs = NULL;
return;
}
args = (DqaInfo*)palloc( ctx->numDistinctCols * sizeof(DqaInfo));
for ( i = 0; i < ctx->numDistinctCols; i++)
{
TargetEntry *dtle;
List *x;
int j;
/* Like PG and the SQL standard, we assume that a DQA may take only
* a single argument -- no REGR_SXY(DISTINCT X,Y). This is what allows
* distinctExpr to be an expression rather than a list of expressions.
*/
dtle = get_tle_by_resno(ctx->sub_tlist, ctx->distinctColIdx[i]);
distinctExpr = (Node*) dtle->expr;
x = NIL;
for ( j = 0; j < ctx->numGroupCols ; j++ )
{
TargetEntry *tle;
tle = get_tle_by_resno(ctx->sub_tlist,ctx->groupColIdx[j]);
x = lappend(x, tle->expr);
}
x = lappend(x, distinctExpr);
args[i].distinctExpr = distinctExpr; /* no copy */
args[i].base_index = dtle->resno;
args[i].num_rows = estimate_num_groups(root, x, input_rows);
args[i].can_hash = hash_safe_type(exprType(distinctExpr));
list_free(x);
}
qsort(args, ctx->numDistinctCols, sizeof(DqaInfo), compareDqas);
/* Reorder ctx->distinctColIdx to agree with join order. */
for ( i = 0; i < ctx->numDistinctCols; i++ )
{
ctx->distinctColIdx[i] = args[i].base_index;
}
ctx->dqaArgs = args;
}
/* Function make_plan_for_one_dqa
*
* Subroutine for make_three_stage_agg_plan constructs a coplan for
* the specified DQA index [0..numDistinctCols-1] which selects a DqaInfo
* entry from the context.
*
* In multi-DQA plans, coplans have minimal targetlists (just grouping
* keys, DQA arguments, and results of single aggregate functions). In
* case this is a single-DQA (join-less) plan, the coplan target list is
* "finalized" to produce the result requested by the user (which may
* include expressions over the minimal list in the targetlist and/or
* having qual).
*
* A Query (including range table) which approximates a query for the
* returned plan is stored back into *coquery_p, if coquery_p is not NULL.
*/
static Plan *
make_plan_for_one_dqa(PlannerInfo *root, MppGroupContext *ctx, int dqa_index,
Plan* result_plan, Query **coquery_p)
{
DqaCoplanType coplan_type;
List *prelim_tlist = NIL;
List *inter_tlist = NIL;
List *final_tlist = NIL;
List *final_qual = NIL;
List *groupExprs = NIL;
List *current_pathkeys;
AggStrategy aggstrategy;
AttrNumber *prelimGroupColIdx;
AttrNumber *inputGroupColIdx;
List *extendedGroupClause;
Query *original_parse;
bool groups_sorted = false;
long numGroups;
int i, n;
DqaInfo *dqaArg = &ctx->dqaArgs[dqa_index];
bool sort_coplans = ( ctx->join_strategy == DqaJoinMerge );
bool groupkeys_collocate = cdbpathlocus_collocates(ctx->input_locus, root->group_pathkeys, false /*exact_match*/);
bool need_inter_agg = false;
bool dqaduphazard = false;
bool stream_bottom_agg = root->config->gp_hashagg_streambottom; /* Take hint */
/* Planning will perturb root->parse, so we copy it's content aside
* so we can restore it later. We flat copy instead of resetting
* because code in the stack may have a local variable set to the
* value of root->parse.
*/
original_parse = makeNode(Query);
memcpy(original_parse, root->parse, sizeof(Query));
/* Our caller, make_three_stage_agg_plan, pushed ctx->sub_tlist onto
* result_plan. This contains all the keys and arguments for the
* whole query. While it would be possible to generate a smaller
* targetlist to use for this single DQA it is probably not worth
* the complexity. Just use sub_tlist as given.
*
* The DQA argument of interest is attribute dqaArg->baseIndex.
*
* Get the target lists for the preliminary, intermediate and final
* aggregations and the qual (HAVING clause) for the final aggregation
* based on the target list of the base plan. Grouping attributes go on
* front of preliminary and intermediate target lists.
*/
generate_dqa_pruning_tlists(ctx,
dqa_index,
&prelim_tlist,
&inter_tlist,
&final_tlist,
&final_qual);
/*
* For the first aggregation phases the original grouping attributes
* (maybe zero of them) must be extended to include the DQA argument
* attribute (exactly one of them) to be pruned.
*
* The grouping attributes and a single DQA argument are on the front and
* in order on the preliminary and intermediate targetlists so we need a
* new vector of grouping attributes, prelimGroupColIdx = (1, 2, 3, ...),
* for use in these aggregations. The vector inputGroupColIdx plays a
* similar role for sub_tlist.
*
* The initial-phase group clause, extendedGroupClause, is the one in
* the query (assumed to have no grouping extensions) augmented by a
* GroupClause node for the DQA argument. This is where the sort
* operator for the DQA argument is selected.
*/
{
GroupClause* gc;
TargetEntry *tle;
prelimGroupColIdx = inputGroupColIdx = NULL;
n = ctx->numGroupCols + 1; /* add the DQA argument as a grouping key */
Assert( n > 0 );
prelimGroupColIdx = (AttrNumber*)palloc(n * sizeof(AttrNumber));
inputGroupColIdx = (AttrNumber*)palloc(n * sizeof(AttrNumber));
for ( i = 0; i < n; i++ )
prelimGroupColIdx[i] = i+1;
for ( i = 0; i < ctx->numGroupCols; i++ )
inputGroupColIdx[i] = ctx->groupColIdx[i];
inputGroupColIdx[ctx->numGroupCols] = dqaArg->base_index;
gc = makeNode(GroupClause);
tle = get_tle_by_resno(ctx->sub_tlist, dqaArg->base_index);
gc->tleSortGroupRef = tle->ressortgroupref;
gc->sortop = ordering_oper_opid(exprType((Node*)dqaArg->distinctExpr));
extendedGroupClause = list_copy(root->parse->groupClause);
extendedGroupClause = lappend(extendedGroupClause,gc);
}
/*
* Determine the first-phase aggregation strategy to use. Prefer hashing
* to sorting because the benefit of the sort will be lost by the Motion
* to follow.
*/
if ( dqaArg->use_hashed_preliminary )
{
aggstrategy = AGG_HASHED;
current_pathkeys = NIL;
}
else
{
/* Here we need to sort! The input pathkeys won't contain the
* DQA argument, so just do it.
*/
result_plan = (Plan *)
make_sort_from_groupcols(root,
extendedGroupClause,
inputGroupColIdx,
false,
result_plan);
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
aggstrategy = AGG_SORTED;
/* The AGG node will not change the sort ordering of its
* groups, so current_pathkeys describes the result too.
*/
}
/*
* Preliminary Aggregation: With the pre-existing distribution, group
* by the combined grouping key and DQA argument. In the case of the
* first coplan, this phase also pre-aggregates any non-DQAs. This
* eliminates duplicate values of the DQA argument on each QE.
*/
numGroups = (dqaArg->num_rows < 0) ? 0 :
(dqaArg->num_rows > LONG_MAX) ? LONG_MAX :
(long)dqaArg->num_rows;
/*
* If the data is distributed on the distinct qualified aggregate's key
* and there is no grouping key, then we prefer to not stream the bottom agg
*/
if (dqaArg->distinctkey_collocate && ctx->numGroupCols == 0)
{
stream_bottom_agg = false;
}
result_plan = (Plan *) make_agg(root,
prelim_tlist,
NIL, /* no havingQual */
aggstrategy, stream_bottom_agg,
ctx->numGroupCols + 1,
inputGroupColIdx,
numGroups,
0, /* num_nullcols */
0, /* input_grouping */
0, /* grouping */
0, /* rollup_gs_times */
ctx->agg_counts->numAggs - ctx->agg_counts->numDistinctAggs + 1,
ctx->agg_counts->transitionSpace, /* worst case */
result_plan);
dqaduphazard = (aggstrategy == AGG_HASHED && stream_bottom_agg);
result_plan->flow = pull_up_Flow(result_plan, result_plan->lefttree, (aggstrategy == AGG_SORTED));
current_pathkeys = NIL;
/*
* Intermediate Motion: Gather or Hash on Groups to get colocation
* on the grouping key. Note that this may bring duplicate values
* of the DQA argument together on the QEs.
*/
switch ( ctx->type )
{
case MPP_GRP_TYPE_GROUPED_DQA_2STAGE:
if (!groupkeys_collocate)
{
groupExprs = NIL;
Assert(ctx->numGroupCols > 0);
for ( i = 0; i < ctx->numGroupCols; i++)
{
TargetEntry *tle;
tle = get_tle_by_resno(prelim_tlist, prelimGroupColIdx[i]);
groupExprs = lappend(groupExprs, copyObject(tle->expr));
}
result_plan = (Plan*)make_motion_hash(root, result_plan, groupExprs);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows);
}
break;
case MPP_GRP_TYPE_PLAIN_DQA_2STAGE:
/* Assert that this is only called for a plain DQA like select count(distinct x) from foo */
Assert(ctx->numGroupCols == 0); /* No group-by */
Assert(n == 1);
/* If already collocated on DQA arg, don't redistribute */
if (!dqaArg->distinctkey_collocate)
{
TargetEntry *tle = get_tle_by_resno(ctx->sub_tlist, dqaArg->base_index);
Assert(tle);
groupExprs = lappend(NIL, copyObject(tle->expr));
result_plan = (Plan*)make_motion_hash(root, result_plan, groupExprs);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows);
}
break;
case MPP_GRP_TYPE_NONE:
case MPP_GRP_TYPE_BASEPLAN:
case MPP_GRP_TYPE_GROUPED_2STAGE:
case MPP_GRP_TYPE_PLAIN_2STAGE:
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("unexpected use of DQA pruned 2-phase aggregation")));
break; /* Never */
}
current_pathkeys = NIL;
groups_sorted = false;
if ( sort_coplans )
{
coplan_type = dqaArg->coplan_type_sorted;
}
else
{
coplan_type = dqaArg->coplan_type_cheapest;
}
if ( dqaduphazard ||
(!dqaArg->distinctkey_collocate && !groupkeys_collocate) )
{
/* Intermediate Aggregation: Grouping key values are colocated so group
* by the combined grouping key and DQA argument while intermediate-
* aggregating any non-DQAs. This once again (and finally) eliminates
* duplicate values of the DQA argument on each QE.
*/
need_inter_agg = true;
switch (coplan_type)
{
case DQACOPLAN_GGS:
case DQACOPLAN_PGS:
aggstrategy = AGG_SORTED;
/* pre-sort required on combined grouping key and DQA argument */
result_plan = (Plan *)
make_sort_from_groupcols(root,
extendedGroupClause,
prelimGroupColIdx,
false,
result_plan);
groups_sorted = true;
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
break;
case DQACOPLAN_GSH:
case DQACOPLAN_SHH:
case DQACOPLAN_HH:
case DQACOPLAN_PH:
aggstrategy = AGG_HASHED;
groups_sorted = false;
break;
}
result_plan = add_second_stage_agg(root,
true,
prelim_tlist,
inter_tlist,
NULL,
aggstrategy,
ctx->numGroupCols + 1,
prelimGroupColIdx,
0, /* num_nullcols */
0, /* input_grouping */
0, /* grouping */
0, /* rollup_gs_times */
dqaArg->num_rows,
ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace,
"partial_aggregation",
&current_pathkeys,
result_plan,
true, false);
}
/* Final Aggregation: Group by the grouping key, aggregate the now
* distinct values of the DQA argument using non-distinct-qualified
* aggregation, final aggregate the intermediate values of any non-DQAs.
*/
switch (coplan_type)
{
case DQACOPLAN_GSH:
/* pre-sort required on grouping key */
result_plan = (Plan *)
make_sort_from_groupcols(root,
root->parse->groupClause,
prelimGroupColIdx,
false,
result_plan);
groups_sorted = true;
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
/* Fall though. */
case DQACOPLAN_GGS:
aggstrategy = AGG_SORTED;
break;
case DQACOPLAN_SHH:
case DQACOPLAN_HH:
aggstrategy = AGG_HASHED;
groups_sorted = false;
break;
case DQACOPLAN_PGS:
case DQACOPLAN_PH:
/* plainagg */
aggstrategy = AGG_PLAIN;
groups_sorted = false;
break;
}
/**
* In the case where there is no grouping key, we need to gather up all the rows in a single segment to compute the final aggregate.
*/
if ( ctx->type == MPP_GRP_TYPE_PLAIN_DQA_2STAGE)
{
/* Assert that this is only called for a plain DQA like select count(distinct x) from foo */
Assert(ctx->numGroupCols == 0); /* No grouping columns */
Assert(n == 1);
result_plan = (Plan*)make_motion_gather_to_QE(result_plan, false);
result_plan->total_cost +=
incremental_motion_cost(result_plan->plan_rows,
result_plan->plan_rows * root->config->cdbpath_segments);
}
result_plan = add_second_stage_agg(root,
true,
need_inter_agg ? inter_tlist : prelim_tlist,
final_tlist,
final_qual,
aggstrategy,
ctx->numGroupCols,
prelimGroupColIdx,
0, /* num_nullcols */
0, /* input_grouping */
ctx->grouping,
0, /* rollup_gs_times */
*ctx->p_dNumGroups,
ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace,
"partial_aggregation",
&current_pathkeys,
result_plan,
true,
false);
/* Final sort */
switch (coplan_type)
{
case DQACOPLAN_SHH:
/* post-sort required */
result_plan = (Plan *)
make_sort_from_groupcols(root,
root->parse->groupClause,
prelimGroupColIdx,
false,
result_plan);
groups_sorted = true;
current_pathkeys = root->group_pathkeys;
mark_sort_locus(result_plan);
break;
case DQACOPLAN_GGS:
case DQACOPLAN_GSH:
case DQACOPLAN_HH:
case DQACOPLAN_PGS:
case DQACOPLAN_PH:
break;
}
/* Marshal implicit results. Return explicit result. */
if ( groups_sorted )
{
/* The following settings work correctly though they seem wrong.
* Though we changed the query tree, we say that we did not so that
* planner.c will notice the useful sort order we have produced.
* We also reset the current pathkeys to the original group keys.
* (Though our target list may differ, its attribute-wise ordering
* is on the group keys.)
*/
ctx->current_pathkeys = root->group_pathkeys; /* current_pathkeys are wrong! */
ctx->querynode_changed = false;
}
else
{
ctx->current_pathkeys = NIL;
ctx->querynode_changed = true;
}
/* If requested, copy our modified Query (at root->parse) for caller. */
if ( coquery_p != NULL )
{
*coquery_p = makeNode(Query);
memcpy(*coquery_p, root->parse, sizeof(Query));
}
/* Restore the original referent of root->parse. */
memcpy(root->parse, original_parse, sizeof(Query));
pfree(original_parse);
return result_plan;
}
static Plan *
join_dqa_coplan(PlannerInfo *root, MppGroupContext *ctx, Plan *outer, int dqa_index)
{
Plan *join_plan = NULL;
Plan *inner = ctx->dqaArgs[dqa_index].coplan;
List *join_tlist = NIL;
List *tlist = NIL;
Index outer_varno = 1;
Index inner_varno = dqa_index + 1;
Index varno = 1;
int i, ng, nd;
/* Make the target list for this join. The outer and inner target lists
* will look like
* (<g'> <D0'> ... <Dn-1'> <F'>) and (<g'> <Dn>)
* or
* (<g'> <D0> <F>) and (<g'> <Dn>)
* The join target list should look like
* (<g'> <D0'> ... <Dn'> <F'>)
*/
/* Use varno 1 for grouping key. */
join_tlist = make_vars_tlist(ctx->grps_tlist, varno, 0);
ng = list_length(join_tlist); /* (<g'>) */
nd = ng + list_length(ctx->dref_tlists[0]);/* (<g'> <D0'>) */
for ( i = 0; i <= dqa_index; i++ )
{
tlist = make_vars_tlist(ctx->dref_tlists[i], varno+i, ng);
join_tlist = seq_tlist_concat(join_tlist, tlist); /* (... <Di'>) */
}
tlist = make_vars_tlist(ctx->frefs_tlist, varno, nd);
join_tlist = seq_tlist_concat(join_tlist, tlist); /* (... <F'>) */
/* Make the join which will be either a cartesian product (in case of
* scalar aggregation) or a merge or hash join (in case of grouped
* aggregation.)
*/
if ( ctx->numGroupCols > 0 ) /* MergeJoin: 1x1 */
{
List *joinclause = NIL;
List *hashclause = NIL;
AttrNumber attrno;
Insist( ctx->join_strategy == DqaJoinMerge || ctx->join_strategy == DqaJoinHash );
/* Make the join clause -- a conjunction of IS NOT DISTINCT FROM
* predicates on the attributes of the grouping key.
*/
for ( attrno = 1; attrno <= ctx->numGroupCols; attrno++ )
{
Expr *qual;
Var *outer_var;
Var *inner_var;
TargetEntry *tle = get_tle_by_resno(outer->targetlist, attrno);
Assert( tle && IsA(tle->expr, Var) );
outer_var = (Var*)copyObject(tle->expr);
outer_var->varno = outer_varno;
outer_var->varnoold = outer_varno;
inner_var = (Var*)copyObject(tle->expr);
inner_var->varno = inner_varno;
inner_var->varnoold = inner_varno;
/* outer should always be on the left */
qual = make_op(NULL, list_make1(makeString("=")),
(Node*) outer_var,
(Node*) inner_var, -1);
if ( ctx->join_strategy == DqaJoinHash )
{
hashclause = lappend(hashclause, copyObject(qual));
}
qual->type = T_DistinctExpr;
qual = make_notclause(qual);
joinclause = lappend(joinclause, qual);
}
if ( ctx->join_strategy == DqaJoinHash )
{
/* Make the hash join. */
bool motion_added_outer = false;
bool motion_added_inner = false;
outer = add_motion_to_dqa_child(outer, root, &motion_added_outer);
inner = add_motion_to_dqa_child(inner, root, &motion_added_inner);
bool prefetch_inner = motion_added_outer || motion_added_inner;
if (motion_added_outer || motion_added_inner)
{
ctx->current_pathkeys = NULL;
}
Hash *hash_plan = make_hash(inner);
join_plan = (Plan*)make_hashjoin(join_tlist,
NIL, /* joinclauses */
NIL, /* otherclauses */
hashclause, /* hashclauses */
joinclause, /* hashqualclauses */
outer, (Plan*)hash_plan,
JOIN_INNER);
((Join *) join_plan)->prefetch_inner = prefetch_inner;
}
else
{
/* Make the merge join noting that the outer plan produces rows
* distinct in the join key. (So does the inner, for that matter,
* but the MJ algorithm is only sensitive to the outer.)
*/
join_plan = (Plan*)make_mergejoin(join_tlist,
NIL, NIL,
joinclause,
outer, inner,
JOIN_INNER);
((MergeJoin*)join_plan)->unique_outer = true;
}
}
else /* NestLoop: Cartesian product: 1x1 */
{
Insist(ctx->join_strategy == DqaJoinCross);
join_plan = (Plan*)make_nestloop(join_tlist,
NIL, NIL,
outer, inner,
JOIN_INNER);
((NestLoop*)join_plan)->singleton_outer = true;
}
join_plan->startup_cost = outer->startup_cost + inner->startup_cost;
join_plan->plan_rows = outer->plan_rows;
join_plan->plan_width = outer->plan_width + inner->plan_width; /* too high for MJ */
join_plan->total_cost = outer->total_cost + inner->total_cost;
join_plan->total_cost += cpu_tuple_cost * join_plan->plan_rows;
join_plan->flow = pull_up_Flow(join_plan, join_plan->lefttree, true);
return join_plan;
}
/*
* Function make_subplan_tlist (for multi-phase aggregation)
*
* The input to the "lower" Agg node will use a "flattened" tlist (having
* just the (levelsup==0) vars mentioned in the targetlist and HAVING qual.
* This is similar to the target list produced by make_subplanTargetList
* and used for 1-stage aggregation in grouping_planner.
*
* The result target list contains entries for all the simple Var attributes
* of the original SELECT and HAVING clauses, plus entries for any GROUP BY
* expressions and DQA arguments that are not simple Vars.
*
* The implicit results are
*
* - the number of grouping attributes and a vector of their positions
* (which are equal to their resno's) in the target list delivered through
* pointers pnum_gkeys and pcols_gkeys, and
*
* - the number of distinct arguments to DISTINCT-qualified aggregate
* function and a vector of their positions (which are equal to their
* resno's) in the target list delivered through pointers pnum_dqas and
* pcols_dqas. These arguments are guaranteed (by the call to function
* augment_subplan_tlist) to appear as attributes of the subplan target
* list.
*
* There are no similar results for sort and distinct attributes since
* they don't necessarily appear in the subplan target list.
*/
List *make_subplan_tlist(List *tlist, Node *havingQual,
List *grp_clauses,
int *pnum_gkeys, AttrNumber **pcols_gkeys,
List *dqa_args,
int *pnum_dqas, AttrNumber **pcols_dqas)
{
List *sub_tlist;
List *extravars;
int num_gkeys;
AttrNumber *cols_gkeys;
Assert( dqa_args != NIL? pnum_dqas != NULL && pcols_dqas != NULL: true );
sub_tlist = flatten_tlist(tlist);
extravars = pull_var_clause(havingQual, false);
sub_tlist = add_to_flat_tlist(sub_tlist, extravars, false /* resjunk */);
list_free(extravars);
num_gkeys = num_distcols_in_grouplist(grp_clauses);
if (num_gkeys > 0)
{
int keyno = 0;
ListCell *l;
List *tles;
cols_gkeys = (AttrNumber*) palloc(sizeof(AttrNumber) * num_gkeys);
tles = get_sortgroupclauses_tles(grp_clauses, tlist);
foreach (l, tles)
{
Node *expr;
TargetEntry *tle, *sub_tle = NULL;
ListCell *sl;
tle = (TargetEntry*) lfirst(l);
expr = (Node*)tle->expr;
/* Find or make a matching sub_tlist entry. */
foreach(sl, sub_tlist)
{
sub_tle = (TargetEntry *) lfirst(sl);
if (equal(expr, sub_tle->expr) && (sub_tle->ressortgroupref == 0))
break;
}
if (!sl)
{
sub_tle = makeTargetEntry((Expr*) expr,
list_length(sub_tlist) + 1,
NULL,
false);
sub_tlist = lappend(sub_tlist, sub_tle);
}
/* Set its group reference and save its resno */
sub_tle->ressortgroupref = tle->ressortgroupref;
cols_gkeys[keyno++] = sub_tle->resno;
}
*pnum_gkeys = num_gkeys;
*pcols_gkeys = cols_gkeys;
}
else
{
*pnum_gkeys = 0;
*pcols_gkeys = NULL;
}
if ( dqa_args != NIL )
sub_tlist = augment_subplan_tlist(sub_tlist, dqa_args, pnum_dqas, pcols_dqas, true);
return sub_tlist; /* Possibly modified by appending expression entries. */
}
/* Function augment_subplan_tlist
*
* Called from make_subplan_tlist, not directly.
*
* Make a target list like the input that includes sortgroupref'd entries
* for the expressions in exprs. Note that the entries in the input expression
* list must be distinct.
*
* New entries corresponding to the expressions in the input exprs list
* (if any) are added to the argument list. Existing entries are modified
* (if necessary) in place.
*
* Return the (modified) input targetlist.
*
* Implicitly return an array of resno values for exprs in (pnum, *pcols), if
* return_resno is true.
*/
List *augment_subplan_tlist(List *tlist, List *exprs, int *pnum, AttrNumber **pcols,
bool return_resno)
{
int num;
AttrNumber *cols = NULL;
num = list_length(exprs); /* Known to be distinct. */
if (num > 0)
{
int keyno = 0;
ListCell *lx, *lt;
TargetEntry *tle, *matched_tle;
Index max_sortgroupref = 0;
foreach (lt, tlist)
{
tle = (TargetEntry*)lfirst(lt);
if ( tle->ressortgroupref > max_sortgroupref )
max_sortgroupref = tle->ressortgroupref;
}
if (return_resno)
cols = (AttrNumber*) palloc(sizeof(AttrNumber) * num);
foreach (lx, exprs)
{
Node *expr = (Node*)lfirst(lx);
matched_tle = NULL;
foreach (lt, tlist)
{
tle = (TargetEntry*)lfirst(lt);
if ( equal(expr, tle->expr) )
{
matched_tle = tle;
break;
}
}
if ( matched_tle == NULL )
{
matched_tle = makeTargetEntry((Expr*) expr,
list_length(tlist) + 1,
NULL,
false);
tlist = lappend(tlist, matched_tle);
}
if ( matched_tle->ressortgroupref == 0 )
matched_tle->ressortgroupref = ++max_sortgroupref;
if (return_resno)
cols[keyno++] = matched_tle->resno;
}
if (return_resno)
{
*pnum = num;
*pcols = cols;
}
}
else
{
if (return_resno)
{
*pnum = 0;
*pcols = NULL;
}
}
/* Note that result is a copy, possibly modified by appending expression
* targetlist entries and/or updating sortgroupref values.
*/
return tlist;
}
/*
* Function describe_subplan_tlist (for single-phase aggregation)
*
* Implicitly return extra information about a supplied target list with having
* qual and and corresponding sub plan target list for single-phase aggregation.
* This does, essentially, what make_subplan_tlist does, but for a precalculated
* subplan target list. In particular
*
* - it constructs the grouping key -> subplan target list resno map.
* - it may extend the subplan targetlist for DQAs and record entries
* in the DQA argument -> subplan target list resno map.
*
* In the later case, the subplan target list may be extended, so return it.
* This function is for the case when a subplan target list (not a whole plan)
* is supplied to cdb_grouping_planner.
*/
List *describe_subplan_tlist(List *sub_tlist,
List *tlist, Node *havingQual,
List *grp_clauses, int *pnum_gkeys, AttrNumber **pcols_gkeys,
List *dqa_args, int *pnum_dqas, AttrNumber **pcols_dqas)
{
int nkeys;
AttrNumber *cols;
nkeys = num_distcols_in_grouplist(grp_clauses);
if ( nkeys > 0 )
{
List *tles;
ListCell *lc;
int keyno = 0;
cols = (AttrNumber*)palloc0(sizeof(AttrNumber)*nkeys);
tles = get_sortgroupclauses_tles(grp_clauses, tlist);
foreach (lc, tles)
{
TargetEntry *tle;
TargetEntry *sub_tle;
tle = (TargetEntry*)lfirst(lc);
sub_tle = tlist_member((Node*)tle->expr, sub_tlist);
Assert(tle->ressortgroupref != 0);
Assert(tle->ressortgroupref == sub_tle->ressortgroupref);
Assert(keyno < nkeys);
cols[keyno++] = sub_tle->resno;
}
*pnum_gkeys = nkeys;
*pcols_gkeys = cols;
}
else
{
*pnum_gkeys = 0;
*pcols_gkeys = NULL;
}
if ( dqa_args != NIL )
sub_tlist = augment_subplan_tlist(sub_tlist, dqa_args, pnum_dqas, pcols_dqas, true);
return sub_tlist;
}
/*
* Generate targetlist for a SubqueryScan node to wrap the stage-one
* Agg node (partial aggregation) of a 2-Stage aggregation sequence.
*
* varno: varno to use in generated Vars
* input_tlist: targetlist of this node's input node
*
* Result is a "flat" (all simple Var node) targetlist in which
* varattno and resno match and are sequential.
*
* This function also returns a map between the original targetlist
* entry to new target list entry using resno values. The index
* positions for resno_map represent the original resnos, while the
* array elements represent the new resnos. The array is allocated
* by the caller, which should have length of list_length(input_tlist).
*/
List *
generate_subquery_tlist(Index varno, List *input_tlist,
bool keep_resjunk, int **p_resno_map)
{
List *tlist = NIL;
int resno = 1;
ListCell *j;
TargetEntry *tle;
Node *expr;
*p_resno_map = (int *)palloc0(list_length(input_tlist) * sizeof(int));
foreach(j, input_tlist)
{
TargetEntry *inputtle = (TargetEntry *) lfirst(j);
Assert(inputtle->resno == resno && inputtle->resno >= 1);
/* Don't pull up constants, always use a Var to reference the input. */
expr = (Node *) makeVar(varno,
inputtle->resno,
exprType((Node *) inputtle->expr),
exprTypmod((Node *) inputtle->expr),
0);
(*p_resno_map)[inputtle->resno - 1] = resno;
tle = makeTargetEntry((Expr *) expr,
(AttrNumber) resno++,
(inputtle->resname == NULL) ?
NULL :
pstrdup(inputtle->resname),
keep_resjunk ? inputtle->resjunk : false);
tle->ressortgroupref = inputtle->ressortgroupref;
tlist = lappend(tlist, tle);
}
return tlist;
}
/*
* Function: cdbpathlocus_collocates
*
* Is a relation with the given locus guaranteed to collocate tuples with
* non-distinct values of the key. The key is a list of pathkeys (each of
* which is a list of PathKeyItem*).
*
* Collocation is guaranteed if the locus specifies a single process or
* if the result is partitioned on a subset of the keys that must be
* collocated.
*
* We ignore onther sorts of collocation, e.g., replication or partitioning
* on a range since these cannot occur at the moment (MPP 2.3).
*/
bool cdbpathlocus_collocates(CdbPathLocus locus, List *pathkeys, bool exact_match)
{
ListCell *lc;
List *exprs = NIL;
if ( CdbPathLocus_IsBottleneck(locus) )
return true;
if ( !CdbPathLocus_IsHashed(locus) )
return false; /* Or would HashedOJ ok, too? */
if (exact_match && list_length(pathkeys) != list_length(locus.partkey))
{
return false;
}
/* Extract a list of expressions from the pathkeys. Since the locus
* presumably knows all about attribute equivalence classes, we use
* only the first item in each input path key.
*/
foreach( lc, pathkeys )
{
PathKeyItem *item;
List *lst = (List*)lfirst(lc);
Assert( list_length(lst) > 0 );
item = (PathKeyItem*)linitial(lst);
exprs = lappend(exprs, item->key);
}
/* Check for containment of locus in exprs. */
return cdbpathlocus_is_hashed_on_exprs(locus, exprs);
}
/*
* Function: cdbpathlocus_from_flow
*
* Generate a locus from a flow. Since the information needed to produce
* canonical path keys is unavailable, this function will never return a
* hashed locus.
*/
CdbPathLocus cdbpathlocus_from_flow(Flow *flow)
{
CdbPathLocus locus;
CdbPathLocus_MakeNull(&locus);
if (!flow)
return locus;
switch (flow->flotype)
{
case FLOW_SINGLETON:
if (flow->segindex == -1)
CdbPathLocus_MakeEntry(&locus);
else
CdbPathLocus_MakeSingleQE(&locus);
break;
case FLOW_REPLICATED:
CdbPathLocus_MakeReplicated(&locus);
break;
case FLOW_PARTITIONED:
CdbPathLocus_MakeStrewn(&locus);
break;
case FLOW_UNDEFINED:
default:
Insist(0);
}
return locus;
}
/*
* Generate 3 target lists for a sequence of consecutive Agg nodes.
*
* This is intended for a sequence of consecutive Agg nodes used in
* a ROLLUP. '*p_tlist3' is for the upper Agg node, and '*p_tlist2' is
* for any Agg node in the middle, and '*p_tlist1' is for the
* bottom Agg node.
*
* '*p_tlist1' and '*p_tlist2' have similar target lists. '*p_tlist3'
* is constructed based on tlist and outputs from *p_tlist2 or
* '*p_tlist1' if twostage is true.
*
* NB This function is called externally (from plangroupext.c) and not
* used in this file! Beware: the API is now legacy here!
*/
void generate_three_tlists(List *tlist,
bool twostage,
List *sub_tlist,
Node *havingQual,
int numGroupCols,
AttrNumber *groupColIdx,
List **p_tlist1,
List **p_tlist2,
List **p_tlist3,
List **p_final_qual)
{
ListCell *lc;
int resno = 1;
MppGroupContext ctx; /* Just for API matching! */
/* Similar to the final tlist entries in two-stage aggregation,
* we use consistent varno in the middle tlist entries.
*/
int middle_varno = 1;
/* Generate the top and bottom tlists by calling the multi-phase
* aggregation code in cdbgroup.c.
*/
ctx.tlist = tlist;
ctx.sub_tlist = sub_tlist;
ctx.havingQual = havingQual;
ctx.numGroupCols = numGroupCols;
ctx.groupColIdx = groupColIdx;
ctx.numDistinctCols = 0;
ctx.distinctColIdx = NULL;
generate_multi_stage_tlists(&ctx,
p_tlist1,
NULL,
p_tlist3,
p_final_qual);
/*
* Read target entries in '*p_tlist1' one by one, and construct
* the entries for '*p_tlist2'.
*/
foreach (lc, *p_tlist1)
{
TargetEntry *tle = (TargetEntry *)lfirst(lc);
Expr *new_expr;
TargetEntry *new_tle;
if (IsA(tle->expr, Aggref))
{
Aggref *aggref = (Aggref *)tle->expr;
Aggref *new_aggref = makeNode(Aggref);
new_aggref->aggfnoid = aggref->aggfnoid;
new_aggref->aggtype = aggref->aggtype;
new_aggref->args =
list_make1((Expr*)makeVar(middle_varno, tle->resno, aggref->aggtype, -1, 0));
new_aggref->agglevelsup = 0;
new_aggref->aggstar = false;
new_aggref->aggdistinct = false; /* handled in preliminary aggregation */
new_aggref->aggstage = AGGSTAGE_INTERMEDIATE;
new_expr = (Expr *)new_aggref;
}
else
{
/* Just make a new Var. */
new_expr = (Expr *)makeVar(middle_varno,
tle->resno,
exprType((Node *)tle->expr),
exprTypmod((Node *)tle->expr),
0);
}
new_tle = makeTargetEntry(new_expr, resno,
(tle->resname == NULL) ?
NULL :
pstrdup(tle->resname),
false);
new_tle->ressortgroupref = tle->ressortgroupref;
*p_tlist2 = lappend(*p_tlist2, new_tle);
resno++;
}
/*
* This may be called inside a two-stage aggregation. In this case,
* We want to make sure all entries in the '*p_tlist3' are visible.
*/
foreach (lc, *p_tlist3)
{
TargetEntry *tle = (TargetEntry *)lfirst(lc);
if (twostage)
tle->resjunk = false;
/* We also set aggstage to AGGSTAGE_INTERMEDIATE if this is in
* a two-stage aggregation, because the agg node in
* the second stage aggregation will do the finalize.
*/
if (twostage && IsA(tle->expr, Aggref))
{
Aggref *aggref = (Aggref *)tle->expr;
aggref->aggstage = AGGSTAGE_INTERMEDIATE;
}
}
}
/*
* Function: generate_multi_stage_tlists
*
* Generate target lists and having qual for multi-stage aggregation.
*
* Input is
*
* From the ctx argument:
* tlist - the preprocessed target list of the original query
* sub_tlist - the reduced target list to use as input to the aggregation
* (If use_dqa_pruning, the all DQA arguments must appear in
* this list and must have non-zero sortgrouprefs.)
* havingQual - the preprocesses having qual of the originaly query
* (in list-of-conjunct-Exprs form)
* numGroupCols - number of grouping attributes (no grouping extensions)
* groupColIdx - resnos (= attr numbers) of the grouping attributes
* numDistinctCols - number of DISTINCT-qualified argument attributes
* distinctColIdx - resnos (= attr numbers) of the DQA argument attributes
*
* From use_dqa_pruning:
* Do we want to construct the tlists to support pruning DQAs?
*
* Output is pointers to
*
* prelim_tlist - the target list of the preliminary Agg node.
* inter_tlist - an optional intermediate target list for an Agg node
* used in multi-phase DQA pruning (p_inter_tlist non-null).
* final_tlist - the target list of the final Agg node.
* final_qual - the qual of the final Agg node.
*/
void generate_multi_stage_tlists(MppGroupContext *ctx,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist,
List **p_final_qual)
{
/* Use consistent varno in final and intermediate tlist entries. It will
* refer to the sole RTE (a Subquery RTE) of a SubqueryScan. */
ctx->final_varno = 1;
/* Do we need to build an intermediate tlist in irefs_tlist? */
ctx->use_irefs_tlist = ( p_inter_tlist != NULL );
/* Don't do DQA pruning. Use prepare/generate_dqa_pruning_tlists! */
ctx->use_dqa_pruning = false;
deconstruct_agg_info(ctx);
reconstruct_agg_info(ctx,
p_prelim_tlist, p_inter_tlist,
p_final_tlist, p_final_qual);
}
/*
* Function: prepare_dqa_pruning_tlists
*
* Performs the first phase of generate_multi_phase_tlist, but waits for
* a subquent call to generate_dqa_pruning_tlists to actually produce the
* target list. (This allows for the fact that DQA pruning may involve
* several "coplans" each with its own target list requirements. This
* function lays the groundwork for all such target lists.
*/
void prepare_dqa_pruning_tlists(MppGroupContext *ctx)
{
/* Use consistent varno in final and intermediate tlist entries. It will
* refer to the sole RTE (a Subquery RTE) of a SubqueryScan. */
ctx->final_varno = 1;
/* Do we need to build an intermediate tlist in irefs_tlist? */
ctx->use_irefs_tlist = true;
/* Do we want to do DQA pruning (in case there are any DISTINCT-qualified
* aggregate functions)? */
ctx->use_dqa_pruning = true;
deconstruct_agg_info(ctx);
}
/*
* Function: generate_dqa_pruning_tlists
*
* Performs the last phase of generate_multi_phase_tlist in the context of
* DQA pruning.
*/
void generate_dqa_pruning_tlists(MppGroupContext *ctx,
int dqa_index,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist,
List **p_final_qual)
{
Assert( p_inter_tlist != NULL ); /* optional elsewhere, required here. */
Assert( ctx->use_dqa_pruning );
if ( ctx->numDistinctCols == 1 )
{
/* Finalized results for single-DQA (join-less) plan. */
reconstruct_agg_info(ctx,
p_prelim_tlist,
p_inter_tlist,
p_final_tlist,
p_final_qual);
}
else
{
/* Minimal results for multi-DQA (join) plan. */
reconstruct_coplan_info(ctx,
dqa_index,
p_prelim_tlist,
p_inter_tlist,
p_final_tlist);
*p_final_qual = NIL;
}
}
/* Function: deconstruct_agg_info
*
* Top-level deconstruction of the target list and having qual of an
* aggregate Query into intermediate structures that will later guide
* reconstruction of the various target lists and expressions involved
* in a multi-phase aggregation plan, possibly with DISTINCT-qualified
* aggregate functions (DQAs).
*/
void deconstruct_agg_info(MppGroupContext *ctx)
{
int i;
ListCell *lc;
/* Initialize temporaries to hold the parts of the preliminary target
* list under construction. */
ctx->grps_tlist = NIL;
ctx->dqa_tlist = NIL;
ctx->prefs_tlist = NIL;
ctx->irefs_tlist = NIL;
ctx->frefs_tlist = NIL;
ctx->dref_tlists = NULL;
ctx->fin_tlist = NIL;
ctx->fin_hqual = NIL;
/*
* Begin constructing the target list for the preliminary Agg node
* by placing targets for the grouping attributes on the grps_tlist
* temporary. Make sure ressortgroupref matches the original. Copying
* the expression may be overkill, but it is safe.
*/
for ( i = 0; i < ctx->numGroupCols; i++ )
{
TargetEntry *sub_tle, *prelim_tle;
sub_tle = get_tle_by_resno(ctx->sub_tlist, ctx->groupColIdx[i]);
prelim_tle = makeTargetEntry(copyObject(sub_tle->expr),
list_length(ctx->grps_tlist) + 1,
(sub_tle->resname == NULL) ?
NULL :
pstrdup(sub_tle->resname),
false);
prelim_tle->ressortgroupref = sub_tle->ressortgroupref;
prelim_tle->resjunk = false;
ctx->grps_tlist = lappend(ctx->grps_tlist, prelim_tle);
}
/*
* Continue to construct the target list for the preliminary Agg node
* by placing targets for the argument attribute of each DQA on the
* dqa_tlist temporary. Make sure ressortgroupref matches the original.
*/
for ( i = 0; i < ctx->numDistinctCols; i++ )
{
TargetEntry *sub_tle, *prelim_tle;
sub_tle = get_tle_by_resno(ctx->sub_tlist, ctx->distinctColIdx[i]);
prelim_tle = makeTargetEntry(copyObject(sub_tle->expr),
list_length(ctx->dqa_tlist) + 1,
(sub_tle->resname == NULL) ?
NULL :
pstrdup(sub_tle->resname),
false);
prelim_tle->ressortgroupref = sub_tle->ressortgroupref;
prelim_tle->resjunk = false;
ctx->dqa_tlist = lappend(ctx->dqa_tlist, prelim_tle);
}
/* Initialize the array of Aggref target lists corresponding to the
* DQA argument target list just constructed.
*/
ctx->dref_tlists = (List **)palloc0(ctx->numDistinctCols * sizeof(List*));
/*
* Derive the final target list with entries corresponding to the input
* target list, but referring to the attributes of the preliminary target
* list rather than to the input attributes. Note that this involves
* augmenting the prefs_tlist temporary as we encounter new Aggref nodes.
*/
foreach (lc, ctx->tlist)
{
TargetEntry *tle, *final_tle;
Expr *expr;
tle = (TargetEntry*)lfirst(lc);
ctx->split_aggref_sortgroupref = tle->ressortgroupref; /* for deconstruction subroutines */
expr = deconstruct_expr(tle->expr, ctx);
ctx->split_aggref_sortgroupref = 0;
final_tle = makeTargetEntry(expr,
tle->resno,
(tle->resname == NULL) ?
NULL :
pstrdup(tle->resname),
tle->resjunk);
final_tle->ressortgroupref = tle->ressortgroupref;
ctx->fin_tlist = lappend(ctx->fin_tlist, final_tle);
}
/*
* Derive the final qual while augmenting the preliminary target list. */
ctx->fin_hqual = (List*)deconstruct_expr((Expr*)ctx->havingQual, ctx);
/* Now cache some values to avoid repeated recalculation by subroutines. */
/* Use consistent varno in final, intermediate an join tlist entries.
* final refers to the sole RTE (a Subquery RTE) of a SubqueryScan.
* outer and inner to the respective inputs to a join.
*/
ctx->final_varno = 1;
ctx->outer_varno = OUTER;
ctx->inner_varno = INNER;
/* Target lists used in multi-phase planning at or above the level
* of individual DQA coplans have one of the forms
*
* [G][D0...Dn][R]
* where
* G represents the grouping key attributes (if any)
* Di represents the results of the DQAs that take the i-th
* unique DQA argument (if any)
* R represents the results of regular aggregate functions (if any)
*
* The offset at index position i is the number of attributes that
* precede the first for the i-th DQA (index origin 0). The last
* is the offset of the first attribute following the DQA attributes.
*/
ctx->dqa_offsets = palloc(sizeof(int) * (1 + ctx->numDistinctCols));
ctx->dqa_offsets[0] = ctx->numGroupCols;
for ( i = 0; i < ctx->numDistinctCols; i++ )
{
ctx->dqa_offsets[i+1] = ctx->dqa_offsets[i]
+ list_length(ctx->dref_tlists[i]);
}
}
/* Function: reconstruct_agg_info
*
* Construct the preliminay, optional intermediate, and final target lists
* and the final having qual for the aggregation plan. If we are doing
* DQA pruning, this function is appropriate only for the cases of 0 or 1
* DQA.
*
* During processing we set ctx->top_tlist to be the flat target list
* containing only the grouping key and the results of individual aggregate
* functions. This list is transient -- it drives the production of the
* final target list and having qual through finalize_split_expression.
*/
void reconstruct_agg_info(MppGroupContext *ctx,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist,
List **p_final_qual)
{
List *prelim_tlist = NIL;
List *inter_tlist = NIL;
List *final_tlist = NIL;
/* Grouping keys */
prelim_tlist = ctx->grps_tlist;
if ( p_inter_tlist != NULL )
inter_tlist = make_vars_tlist(ctx->grps_tlist, ctx->final_varno, 0);
final_tlist = make_vars_tlist(ctx->grps_tlist, ctx->final_varno, 0);
/* If applicable, single DQA argument, corresponding DQAs */
if ( ctx->use_dqa_pruning )
{
if ( list_length(ctx->dqa_tlist) == 1 )
{
int n = list_length(prelim_tlist);
TargetEntry *tle = (TargetEntry*)linitial(ctx->dqa_tlist);
tle->resno = n+1;
prelim_tlist = lappend(prelim_tlist, tle);
if ( p_inter_tlist != NULL )
{
inter_tlist = list_concat(inter_tlist,
make_vars_tlist(ctx->dqa_tlist,
ctx->final_varno, n));
}
final_tlist = seq_tlist_concat(final_tlist, ctx->dref_tlists[0]);
}
else if ( list_length(ctx->dqa_tlist) != 0 )
{
/* Shouldn't use this function for multi-DQA pruning. */
elog(ERROR,"Unexpected use of DISTINCT-qualified aggregate pruning");
}
}
/* Aggrefs */
prelim_tlist = seq_tlist_concat(prelim_tlist, ctx->prefs_tlist);
if ( p_inter_tlist != NULL )
{
inter_tlist = seq_tlist_concat(inter_tlist, ctx->irefs_tlist);
}
final_tlist = seq_tlist_concat(final_tlist, ctx->frefs_tlist);
/* Set implicit results */
*p_prelim_tlist = prelim_tlist;
if ( p_inter_tlist != NULL )
*p_inter_tlist = inter_tlist;
ctx->top_tlist = final_tlist;
*p_final_tlist = (List*) finalize_split_expr((Node*) ctx->fin_tlist, ctx);
*p_final_qual = (List*) finalize_split_expr((Node*) ctx->fin_hqual, ctx);
}
/* Function: reconstruct_coplan_info
*
* Construct the preliminary, intermediate, and final target lists
* for the DQA pruning aggregation coplan specified by dqa_index.
*
* Note: Similar to reconstruct_agg_info but stop short of finalization
* and is sensitive to dqa_index. Ordinarily this function would
* be used only for multiple-DQA planning.
*/
void reconstruct_coplan_info(MppGroupContext *ctx,
int dqa_index,
List **p_prelim_tlist,
List **p_inter_tlist,
List **p_final_tlist)
{
List *prelim_tlist = NIL;
List *inter_tlist = NIL;
List *final_tlist = NIL;
int n;
TargetEntry *tle;
/* Grouping keys */
prelim_tlist = copyObject(ctx->grps_tlist);
if ( p_inter_tlist != NULL )
inter_tlist = make_vars_tlist(ctx->grps_tlist, ctx->final_varno, 0);
final_tlist = make_vars_tlist(ctx->grps_tlist, ctx->final_varno, 0);
/* Single DQA argument, corresponding DQAs */
Assert ( ctx->use_dqa_pruning );
n = list_length(prelim_tlist);
tle = (TargetEntry*)list_nth(ctx->dqa_tlist, dqa_index);
tle->resno = n+1;
prelim_tlist = lappend(prelim_tlist, tle);
if ( p_inter_tlist != NULL )
{
List *x = list_make1(tle);
inter_tlist = list_concat(inter_tlist,
make_vars_tlist(x, ctx->final_varno, n));
list_free(x);
}
final_tlist = seq_tlist_concat(final_tlist, ctx->dref_tlists[dqa_index]);
/* Plain Aggrefs go only on the first coplan! */
if ( dqa_index == 0 )
{
prelim_tlist = seq_tlist_concat(prelim_tlist, ctx->prefs_tlist);
if ( p_inter_tlist != NULL )
{
inter_tlist = seq_tlist_concat(inter_tlist, ctx->irefs_tlist);
}
final_tlist = seq_tlist_concat(final_tlist, ctx->frefs_tlist);
}
/* Set implicit results */
*p_prelim_tlist = prelim_tlist;
if ( p_inter_tlist != NULL )
{
*p_inter_tlist = inter_tlist;
}
*p_final_tlist = final_tlist;
}
/*
* Function: deconstruct_expr
*
* Prepare an expression for execution within 2-stage aggregation.
* This involves adding targets as needed to the target list of the
* first (partial) aggregation and referring to this target list from
* the modified expression for use in the second (final) aggregation.
*/
Expr *deconstruct_expr(Expr *expr, MppGroupContext *ctx)
{
return (Expr*)deconstruct_expr_mutator((Node*)expr, ctx);
}
/*
* Function: deconstruct_expr_mutator
*
* Work for deconstruct_expr.
*/
Node* deconstruct_expr_mutator(Node *node, MppGroupContext *ctx)
{
TargetEntry *tle;
if (node == NULL)
return NULL;
if (IsA(node, Aggref))
{
Aggref *aggref = (Aggref*)node;
return split_aggref(aggref, ctx);
}
/* If the given expression is a grouping expression, replace it with
* a Var node referring to the (lower) preliminary aggregation's
* target list.
*/
tle = tlist_member_with_ressortgroupref(node, ctx->grps_tlist, ctx->split_aggref_sortgroupref);
if ( tle != NULL )
{
return (Node*) makeVar(grp_varno, tle->resno,
exprType((Node*)tle->expr),
exprTypmod((Node*)tle->expr), 0);
}
return expression_tree_mutator(node, deconstruct_expr_mutator, (void*)ctx);
}
/*
* Function: split_aggref
*
* Find or add a partial-stage Aggref expression for the argument in the
* preliminary target list under construction. Return the final-stage
* Aggref with a single Var node argument referring to the partial-stage
* Aggref. In case of a DQA argument reduction, however, there is no
* partial-stage Aggref and the final-stage Aggref differs from the original
* in that (1) it does not specify DISTINCT and (2) it refers to its arguments
* via Vars on the lower range.
*
* For the normal 2-phase case:
*
* Note that he result type of the partial-stage Aggref will be the
* transition type of the aggregate function.
*
* At execution, the function used to compute the transition type in the
* lower aggregation will be the normal accumulator function for the
* aggregate function. The final function, however, will not be used.
* The result will be the ending transition value.
*
* At execution, the function used to compute the transition type in the
* upper aggregation will be the prelim function (known to exist, else
* we would have rejected 2-stage aggregation as a strategy) on input
* values of the transition type. The normal accumulator function for the
* aggregate function will not be used. The normal final function will
* be used to convert the ending transition value to the result type.
* aggregation
*/
Node *split_aggref(Aggref *aggref, MppGroupContext *ctx)
{
ListCell *cell;
Node *final_node;
Oid transtype = InvalidOid;
AttrNumber attrno = OUTER;
TargetEntry *prelim_tle = NULL;
Assert(aggref != NULL && aggref->agglevelsup == 0);
if ( aggref->aggdistinct && ctx->use_dqa_pruning )
{
Index arg_attno;
Index dqa_attno;
TargetEntry *dqa_tle = NULL;
TargetEntry *arg_tle;
List *dref_tlist = NIL;
/* First find the DQA argument. Since this is a DQA, its argument
* list must contain a single expression that matches one of the
* target expressions in ctx->dqa_tlist.
*/
arg_tle = NULL;
if ( list_length(aggref->args) == 1 ) /* safer than Assert */
{
arg_tle = tlist_member(linitial(aggref->args), ctx->dqa_tlist);
}
if (arg_tle == NULL)
elog(ERROR,"Unexpected use of DISTINCT-qualified aggregation");
arg_attno = arg_tle->resno; /* [1..numDistinctCols] */
/* We may have seen a DQA just like this one already. Look for
* one in the distinct Aggref target list to date.
*/
dref_tlist = ctx->dref_tlists[arg_attno - 1];
dqa_attno = 1;
foreach( cell, dref_tlist )
{
TargetEntry *tle = (TargetEntry*)lfirst(cell);
Aggref *ref = (Aggref*)tle->expr;
/* Check similarity, avoid aggtype and aggstage
* (which we control) and don't bother with agglevelsup
* (which is always 0 here) or aggdistinct.
*/
if ( aggref->aggfnoid == ref->aggfnoid
&& aggref->aggstar == ref->aggstar
&& equal(aggref->args, ref->args) )
{
dqa_tle = tle;
break;
}
dqa_attno++;
}
if ( dqa_tle == NULL )
{
/* Didn't find a target for the DQA Aggref so make a new one.
*/
Var *arg_var;
Aggref *dqa_aggref;
arg_var = makeVar(ctx->final_varno, ctx->numGroupCols + 1,
exprType(linitial(aggref->args)),
exprTypmod(linitial(aggref->args)),
0);
dqa_aggref = makeNode(Aggref);
memcpy(dqa_aggref, aggref, sizeof(Aggref)); /* flat copy */
dqa_aggref->args = list_make1(arg_var);
dqa_aggref->aggdistinct = false;
dqa_tle = makeTargetEntry((Expr*)dqa_aggref, dqa_attno, NULL, false);
dref_tlist = lappend(dref_tlist, dqa_tle);
}
ctx->dref_tlists[arg_attno-1] = dref_tlist;
/* Make the "final" target for the DQA case, a reference to the
* DQA Aggref we just found or constructed.
*/
final_node = (Node*) makeVar(dqa_base_varno + arg_attno - 1,
dqa_attno,
exprType((Node*)arg_tle->expr),
exprTypmod((Node*)arg_tle->expr),
0);
}
else /* Ordinary Aggref -or- DQA but ctx->use_dqa_pruning is off. */
{
Aggref *pref;
Aggref *iref;
Aggref *fref;
/*
* We may have seen an Aggref just like this one already. Look for
* the preliminary form of such in the preliminary Aggref target
* list to date.
*/
foreach( cell, ctx->prefs_tlist )
{
TargetEntry *tle = (TargetEntry*)lfirst(cell);
Aggref *ref = (Aggref*)tle->expr;
/* Check similarity, avoid aggtype and aggstage
* (which we control) and don't bother with agglevelsup
* (which is always 0 here).
*/
if ( aggref->aggfnoid == ref->aggfnoid
&& aggref->aggstar == ref->aggstar
&& aggref->aggdistinct == ref->aggdistinct
&& equal(aggref->args, ref->args) )
{
prelim_tle = tle;
transtype = ref->aggtype;
attrno = prelim_tle->resno;
break;
}
}
/*
* If no existing preliminary Aggref target matched, add one that does.
*/
if ( prelim_tle == NULL )
{
TargetEntry *final_tle;
Var *args;
/* Get type information for the Aggref */
transtype = lookup_agg_transtype(aggref);
/* Make a new preliminary Aggref wrapped as a new target entry.
* Like the input Aggref, the preliminary refers to the lower
* range. */
pref = (Aggref*)copyObject(aggref);
pref->aggtype = transtype;
pref->aggstage = AGGSTAGE_PARTIAL;
attrno = 1 + list_length(ctx->prefs_tlist);
prelim_tle = makeTargetEntry((Expr*)pref, attrno, NULL, false);
prelim_tle->ressortgroupref = ctx->split_aggref_sortgroupref;
ctx->prefs_tlist = lappend(ctx->prefs_tlist, prelim_tle);
args = makeVar(ctx->final_varno,
ctx->numGroupCols
+ (ctx->use_dqa_pruning ? 1 : 0)
+ attrno,
transtype, -1, 0);
if ( ctx->use_irefs_tlist )
{
TargetEntry *inter_tle;
iref = makeNode(Aggref);
iref->aggfnoid = pref->aggfnoid;
iref->aggtype = transtype;
iref->args = list_make1((Expr*)copyObject(args));
iref->agglevelsup = 0;
iref->aggstar = false;
iref->aggdistinct = false;
iref->aggstage = AGGSTAGE_INTERMEDIATE;
inter_tle = makeTargetEntry((Expr*)iref, attrno, NULL, false);
inter_tle->ressortgroupref = ctx->split_aggref_sortgroupref;
ctx->irefs_tlist = lappend(ctx->irefs_tlist, inter_tle);
}
/* Make a new final Aggref. */
fref = makeNode(Aggref);
fref->aggfnoid = aggref->aggfnoid;
fref->aggtype = aggref->aggtype;
fref->args = list_make1((Expr*)args);
fref->agglevelsup = 0;
fref->aggstar = false;
fref->aggdistinct = false; /* handled in preliminary aggregation */
fref->aggstage = AGGSTAGE_FINAL;
final_tle = makeTargetEntry((Expr*)fref, attrno, NULL, false);
final_tle->ressortgroupref = ctx->split_aggref_sortgroupref;
ctx->frefs_tlist = lappend(ctx->frefs_tlist, final_tle);
}
final_node = (Node*)makeVar(ref_varno, attrno, aggref->aggtype, -1, 0);
}
return final_node;
}
/* Function: make_vars_tlist
*
* Make a targetlist similar to the given length n tlist but consisting of
* simple Var nodes with the given varno and varattno in offset + [1..N].
*/
List *make_vars_tlist(List *tlist, Index varno, AttrNumber offset)
{
List *new_tlist = NIL;
AttrNumber attno = offset;
ListCell *lc;
foreach (lc, tlist)
{
Var *new_var;
TargetEntry *new_tle;
TargetEntry *tle = (TargetEntry*)lfirst(lc);
attno++;
new_var = makeVar(varno, attno,
exprType((Node*)tle->expr),
exprTypmod((Node*)tle->expr), 0);
new_tle = makeTargetEntry((Expr*)new_var,
attno, /* resno always matches attnr */
(tle->resname == NULL) ? NULL : pstrdup(tle->resname),
false);
new_tle->ressortgroupref = tle->ressortgroupref;
new_tlist = lappend(new_tlist, new_tle);
}
return new_tlist;
}
/* Function: seq_tlist_concat
*
* Concatenates tlist2 to the end of tlist1 adjusting the resno values
* of tlist2 so that the resulting entries have resno = position+1.
* The resno values of tlist1 must be dense from 1 to the length of
* the list. (They are sequential by position, though this is not
* strictly required.
*
* May modify tlist1 in place (to adjust last link and length). Does not
* modify tlist2, but the result shares structure below the TargetEntry
* nodes.
*/
List *seq_tlist_concat(List *tlist1, List *tlist2)
{
ListCell *lc;
AttrNumber high_attno = list_length(tlist1);
foreach (lc, tlist2)
{
TargetEntry *tle= (TargetEntry*)lfirst(lc);
TargetEntry *new_tle = (TargetEntry*)makeNode(TargetEntry);
memcpy(new_tle, tle, sizeof(TargetEntry));
new_tle->resno = ++high_attno;
tlist1 = lappend(tlist1, new_tle);
}
return tlist1;
}
/* Function finalize_split_expr
*
* Note: Only called on the top of the "join" tree, so all D_i are
* included in attribute offset calculations.
*/
Node *finalize_split_expr(Node *expr, MppGroupContext *ctx)
{
return finalize_split_expr_mutator(expr, ctx);
}
/* Mutator subroutine for finalize_split_expr() replaces pseudo Var nodes
* produced by split_aggref() with the similarly typed expression found in
* the top-level targetlist, ctx->top_tlist, being finalized.
*
* For example, a pseudo Var node that represents the 3rd DQA for the
* 2nd DQA argument will be replaced by the targetlist expression that
* corresponds to that DQA.
*/
Node* finalize_split_expr_mutator(Node *node, MppGroupContext *ctx)
{
if (node == NULL)
return NULL;
if (IsA(node, Var))
{
AttrNumber attrno=(AttrNumber)0;
TargetEntry *tle;
Var *pseudoVar = (Var*)node;
if ( pseudoVar->varno == grp_varno )
{
attrno = pseudoVar->varattno;
}
else if ( pseudoVar->varno == ref_varno )
{
if ( ctx->use_dqa_pruning )
{
attrno = ctx->dqa_offsets[ctx->numDistinctCols]
+ pseudoVar->varattno;
}
else
{
attrno = ctx->numGroupCols + pseudoVar->varattno;
}
}
else if ( pseudoVar->varno >= dqa_base_varno && ctx->use_dqa_pruning )
{
int i = pseudoVar->varno - dqa_base_varno;
attrno = ctx->dqa_offsets[i] + pseudoVar->varattno;
}
else
{
elog(ERROR,"Unexpected failure of multi-phase aggregation planning");
}
tle = (TargetEntry*) list_nth(ctx->top_tlist, attrno - 1);
return (Node*) tle->expr;
}
return expression_tree_mutator(node,
finalize_split_expr_mutator,
(void*)ctx);
}
/* Function lookup_agg_transtype
*
* Return the transition type Oid of the given aggregate fuction or throw
* an error, if none.
*/
Oid lookup_agg_transtype(Aggref *aggref)
{
Oid aggid = aggref->aggfnoid;
Oid result;
int fetchCount;
/* XXX: would have been get_agg_transtype() */
result = caql_getoid_plus(
NULL,
&fetchCount,
NULL,
cql("SELECT aggtranstype FROM pg_aggregate "
" WHERE aggfnoid = :1 ",
ObjectIdGetDatum(aggid)));
if (!fetchCount)
elog(ERROR, "cache lookup failed for aggregate %u", aggid);
return result;
}
/* Function: adapt_flow_to_targetlist
*
* Sometimes we replace the targetlist (especially subplan of an Agg).
* In this case, we need to assure that any hash expressions in the
* node's flow remain in the targetlist OR null out the hash leaving,
* in effect, a strewn flow.
*/
void adapt_flow_to_targetlist(Plan *plan)
{
ListCell *c;
Flow *flow;
Assert(plan != NULL);
flow = plan->flow;
Assert(flow);
foreach(c, flow->hashExpr)
{
Node *expr = (Node*)lfirst(c);
if (!tlist_member(expr, plan->targetlist))
{
flow->hashExpr = NIL;
break;
}
}
}
/**
* Update the scatter clause before a query tree's targetlist is about to
* be modified. The scatter clause of a query tree will correspond to
* old targetlist entries. If the query tree is modified and the targetlist
* is to be modified, we must call this method to ensure that the scatter clause
* is kept in sync with the new targetlist.
*/
void UpdateScatterClause(Query *query, List *newtlist)
{
Assert(query);
Assert(query->targetList);
Assert(newtlist);
if (query->scatterClause
&& list_nth(query->scatterClause, 0) != NULL /* scattered randomly */
)
{
Assert(list_length(query->targetList) == list_length(newtlist));
List *scatterClause = NIL;
ListCell *lc = NULL;
foreach (lc, query->scatterClause)
{
Expr *o = (Expr *) lfirst(lc);
Assert(o);
TargetEntry *tle = tlist_member((Node *) o, query->targetList);
Assert(tle);
TargetEntry *ntle = list_nth(newtlist, tle->resno - 1);
scatterClause = lappend(scatterClause, copyObject(ntle->expr));
}
query->scatterClause = scatterClause;
}
}
/*
* Function: add_second_stage_agg
*
* Add an Agg/Group node on top of an Agg/Group node. We add a SubqueryScan
* node on top of the existing Agg/Group node before adding the new Agg/Group
* node.
*
* Params:
* is_agg -- indicate to add an Agg or a Group node.
* prelim_tlist -- the targetlist for the existing Agg/Group node.
* final_tlist -- the targetlist for the new Agg/Group node.
*/
Plan *
add_second_stage_agg(PlannerInfo *root,
bool is_agg,
List *lower_tlist,
List *upper_tlist,
List *upper_qual,
AggStrategy aggstrategy,
int numGroupCols,
AttrNumber *prelimGroupColIdx,
int num_nullcols,
uint64 input_grouping,
uint64 grouping,
int rollup_gs_times,
double numGroups,
int numAggs,
int transSpace,
const char *alias,
List **p_current_pathkeys,
Plan *result_plan,
bool use_root,
bool adjust_scatter)
{
Query *parse = root->parse;
Query *subquery;
List *newrtable;
RangeTblEntry *newrte;
RangeTblRef *newrtref;
Plan *agg_node;
/*
* Add a SubqueryScan node to renumber the range of the query.
*
* The result of the preliminary aggregation (represented by lower_tlist)
* may contain targets with no representatives in the range of its outer
* relation. We resolve this be treating the preliminary aggregation as
* a subquery.
*
* However, this breaks the correspondence between the Plan tree and
* the Query tree that is assumed by the later call to set_plan_references
* as well as by the deparse processing used (e.g.) in EXPLAIN.
*
* So we also push the Query node from the root structure down into a new
* subquery RTE and scribble over the original Query node to make it into
* a simple SELECT * FROM a Subquery RTE.
*
* Note that the Agg phase we add below will refer to the attributes of
* the result of this new SubqueryScan plan node. It is up to the caller
* to set up upper_tlist and upper_qual accordingly.
*/
/* Flat-copy the root query into a newly allocated Query node and adjust
* its target list and having qual to match the lower (existing) Agg
* plan we're about to make into a SubqueryScan.
*/
subquery = copyObject(parse);
subquery->targetList = copyObject(lower_tlist);
subquery->havingQual = NULL;
/* Subquery attributes shouldn't be marked as junk, else they'll be
* skipped by addRangeTableEntryForSubquery. */
{
ListCell *cell;
foreach ( cell, subquery->targetList )
{
TargetEntry *tle = (TargetEntry *)lfirst(cell);
tle->resjunk = false;
if ( tle->resname == NULL )
{
if ( use_root && IsA(tle->expr, Var) )
{
Var *var = (Var*)tle->expr;
RangeTblEntry *rte = rt_fetch(var->varno, root->parse->rtable);
tle->resname = pstrdup(get_rte_attribute_name(rte, var->varattno));
}
else
{
const char *fmt = "unnamed_attr_%d";
char buf[32]; /* big enuf for fmt */
sprintf(buf,fmt,tle->resno);
tle->resname = pstrdup(buf);
}
}
}
}
/* Construct a range table entry referring to it. */
newrte = addRangeTableEntryForSubquery(NULL,
subquery,
makeAlias(alias, NULL),
TRUE);
newrtable = list_make1(newrte);
/* Modify the root query in place to look like its range table is
* a simple Subquery. */
parse->querySource = QSRC_PLANNER; /* but remember it's really ours */
parse->rtable = newrtable;
parse->jointree = makeNode(FromExpr);
newrtref = makeNode(RangeTblRef);
newrtref->rtindex = 1;
parse->jointree->fromlist = list_make1(newrtref);
parse->jointree->quals = NULL;
parse->rowMarks = NIL;
/* <EXECUTE s> uses parse->targetList to derive the portal's tupDesc,
* so when use_root is true, the caller owns the responsibility to make
* sure it ends up in an appropriate form at the end of planning.
*/
if ( use_root )
{
if (adjust_scatter)
{
UpdateScatterClause(parse, upper_tlist);
}
parse->targetList = copyObject(upper_tlist); /* Match range. */
}
result_plan = add_subqueryscan(root, p_current_pathkeys, 1, subquery, result_plan);
/* Add an Agg node */
/* convert current_numGroups to long int */
long lNumGroups = (long) Min(numGroups, (double) LONG_MAX);
agg_node = (Plan *)make_agg(root,
upper_tlist,
upper_qual,
aggstrategy, false,
numGroupCols,
prelimGroupColIdx,
lNumGroups,
num_nullcols,
input_grouping,
grouping,
rollup_gs_times,
numAggs,
transSpace,
result_plan);
/*
* Agg will not change the sort order unless it is hashed.
*/
agg_node->flow = pull_up_Flow(agg_node,
agg_node->lefttree,
(*p_current_pathkeys != NIL)
&& aggstrategy != AGG_HASHED );
/*
* Since the rtable has changed, we had better recreate a RelOptInfo entry for it.
*/
if (root->simple_rel_array)
pfree(root->simple_rel_array);
root->simple_rel_array_size = list_length(parse->rtable) + 1;
root->simple_rel_array = (RelOptInfo **)
palloc0(root->simple_rel_array_size * sizeof(RelOptInfo *));
build_simple_rel(root, 1, RELOPT_BASEREL);
return agg_node;
}
/*
* Add a SubqueryScan node to the input plan and maintain the given
* pathkeys by making adjustments to them and to the equivalence class
* information in root.
*
* Note that submerging a plan into a subquery scan will require changes
* to the range table and to any expressions above the new scan node.
* This is the caller's responsibility since the nature of the changes
* depends on the context in which the subquery is used.
*/
Plan* add_subqueryscan(PlannerInfo* root, List **p_pathkeys,
Index varno, Query *subquery, Plan *subplan)
{
List *subplan_tlist;
int *resno_map;
subplan_tlist = generate_subquery_tlist(varno, subquery->targetList,
false, &resno_map);
subplan = (Plan*)make_subqueryscan(root, subplan_tlist,
NIL,
varno, /* scanrelid (= varno) */
subplan,
subquery->rtable);
mark_passthru_locus(subplan, true, true);
/* Reconstruct equi_key_list since the rtable has changed.
* XXX we leak the old one.
*/
root->equi_key_list = construct_equivalencekey_list(root->equi_key_list,
resno_map,
subquery->targetList,
subplan_tlist);
if ( p_pathkeys != NULL && *p_pathkeys != NULL )
{
*p_pathkeys = reconstruct_pathkeys(root, *p_pathkeys, resno_map,
subquery->targetList, subplan_tlist);
}
pfree(resno_map);
return subplan;
}
/*
* hash_safe_type - is the given type hashable?
*
* Modelled on function hash_safe_grouping in planner.c, we assume
* the type is hashable if its equality operator is marked hashjoinable.
*/
static bool
hash_safe_type(Oid type)
{
Operator optup;
bool oprcanhash;
optup = equality_oper(type, true);
if (!optup)
return false;
oprcanhash = ((Form_pg_operator) GETSTRUCT(optup))->oprcanhash;
ReleaseOperator(optup);
if (!oprcanhash)
return false;
return true;
}
/*
* sorting_prefixes_grouping - is the result ordered on a grouping key prefix?
*
* If so, then we might prefer a pre-ordered grouping result to one that would
* need sorting after the fact.
*/
static bool
sorting_prefixes_grouping(PlannerInfo *root)
{
return root->sort_pathkeys != NIL
&& pathkeys_contained_in(root->sort_pathkeys, root->group_pathkeys);
}
/*
* gp_hash_safe_grouping - are grouping operators GP hashable for
* redistribution motion nodes?
*/
static bool
gp_hash_safe_grouping(PlannerInfo *root)
{
List *grouptles;
ListCell *glc;
grouptles = get_sortgroupclauses_tles(root->parse->groupClause,
root->parse->targetList);
foreach(glc, grouptles)
{
TargetEntry *tle = (TargetEntry *)lfirst(glc);
bool canhash;
canhash = isGreenplumDbHashable(exprType((Node *)tle->expr));
if (!canhash)
return false;
}
return true;
}
/*
* reconstruct_pathkeys
*
* Reconstruct the given pathkeys based on the given mapping from the original
* targetlist to a new targetlist.
*/
List *
reconstruct_pathkeys(PlannerInfo *root, List *pathkeys, int *resno_map,
List *orig_tlist, List *new_tlist)
{
List *new_pathkeys;
ListCell *lc;
foreach (lc, pathkeys)
{
List *keys = (List *)lfirst(lc);
ListCell *key_lc;
TargetEntry *new_tle;
Assert(IsA(keys, List));
foreach(key_lc, keys)
{
PathKeyItem *item = (PathKeyItem *)lfirst(key_lc);
int resno = 0;
ListCell *tle_lc;
Assert(IsA(item, PathKeyItem));
foreach(tle_lc, orig_tlist)
{
TargetEntry *tle = (TargetEntry *)lfirst(tle_lc);
Assert(IsA(tle, TargetEntry));
if (equal(tle->expr, item->key))
{
resno = tle->resno;
break;
}
}
if (resno > 0)
{
new_tle = get_tle_by_resno(new_tlist, resno_map[resno-1]);
Assert(new_tle != NULL);
item->key = (Node *)new_tle->expr;
}
}
}
new_pathkeys = canonicalize_pathkeys(root, pathkeys);
return new_pathkeys;
}
/* cost_common_agg -- Estimate the cost of executing the common subquery
* for an aggregation plan. Assumes that the AggPlanInfo contains the
* correct Path as input_path.
*
* Returns the total cost and, more importantly, populates the given
* dummy Plan node with cost information
*/
Cost cost_common_agg(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info, Plan *dummy)
{
QualCost tlist_cost;
Cost startup_cost;
Cost total_cost;
double input_rows;
int input_width;
int n;
Assert(dummy != NULL);
input_rows = info->input_path->parent->rows;
input_width = info->input_path->parent->width;
/* Path input width isn't correct for ctx->sub_tlist so we guess. */
n = 32 * list_length(ctx->sub_tlist);
input_width = ( input_width < n )? input_width: n;
/* Estimate cost of evaluation of the sub_tlist. */
cost_qual_eval(&tlist_cost, ctx->sub_tlist, root);
startup_cost = info->input_path->startup_cost + tlist_cost.startup;
total_cost = info->input_path->total_cost + tlist_cost.startup +
tlist_cost.per_tuple * input_rows;
memset(dummy, 0, sizeof(Plan));
dummy->type = info->input_path->type;
dummy->startup_cost = startup_cost;
dummy->total_cost = total_cost;
dummy->plan_rows = input_rows;
dummy->plan_width = input_width;
return dummy->total_cost;
}
/* Function cost_1phase_aggregation
*
* May be used for 1 phase aggregation costing with or without DQAs.
* Corresponds to make_one_stage_agg_plan and must be maintained in sync
* with it.
*/
Cost cost_1phase_aggregation(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info)
{
Plan input_dummy;
bool is_sorted;
double input_rows;
long numGroups = (*(ctx->p_dNumGroups) < 0) ? 0 :
(*(ctx->p_dNumGroups) > LONG_MAX) ? LONG_MAX :
(long)*(ctx->p_dNumGroups);
cost_common_agg(root, ctx, info, &input_dummy);
input_rows = input_dummy.plan_rows;
is_sorted = pathkeys_contained_in(root->group_pathkeys, info->input_path->pathkeys);
/* Collocation cost (Motion). */
switch ( info->group_prep )
{
case MPP_GRP_PREP_HASH_GROUPS:
is_sorted = false;
input_dummy.total_cost +=
incremental_motion_cost(input_dummy.plan_rows,
input_dummy.plan_rows);
break;
case MPP_GRP_PREP_FOCUS_QE:
case MPP_GRP_PREP_FOCUS_QD:
input_dummy.total_cost +=
incremental_motion_cost(input_dummy.plan_rows,
input_dummy.plan_rows * root->config->cdbpath_segments);
input_dummy.plan_rows = input_dummy.plan_rows * root->config->cdbpath_segments;
break;
default:
break;
}
/* NB: We don't need to calculate grouping extension costs here because
* grouping extensions are planned elsewhere.
*/
if ( ctx->use_hashed_grouping )
{
/* HashAgg */
Assert( ctx->numDistinctCols == 0 );
add_agg_cost(NULL, &input_dummy,
ctx->sub_tlist, (List*)root->parse->havingQual,
AGG_HASHED, false,
ctx->numGroupCols, ctx->groupColIdx,
numGroups, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
else
{
if ( ctx->numGroupCols == 0 )
{
/* PlainAgg */
add_agg_cost(NULL, &input_dummy,
ctx->sub_tlist, (List*)root->parse->havingQual,
AGG_PLAIN, false,
0, NULL,
1, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
else
{
/* GroupAgg */
if ( ! is_sorted )
{
add_sort_cost(NULL, &input_dummy, ctx->numGroupCols, NULL, NULL);
}
add_agg_cost(NULL, &input_dummy,
ctx->sub_tlist, (List*)root->parse->havingQual,
AGG_SORTED, false,
ctx->numGroupCols, ctx->groupColIdx,
numGroups, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
/* Beware: AGG_PLAIN and AGG_GROUPED may be used with DQAs, however,
* the function cost_agg doesn't distinguish DQAs so it
* consistently under estimates the cost in these cases.
*/
if ( ctx->numDistinctCols > 0 )
{
Path path_dummy;
double ngrps = *(ctx->p_dNumGroups);
double nsorts = ngrps * ctx->numDistinctCols;
double avgsize = input_dummy.plan_rows / ngrps;
cost_sort(&path_dummy, NULL, NIL, 0.0, avgsize, 32);
input_dummy.total_cost += nsorts * path_dummy.total_cost;
}
}
info->plan_cost = root->config->gp_eager_one_phase_agg ? (Cost)0.0 : input_dummy.total_cost;
info->valid = true;
info->join_strategy = DqaJoinNone;
info->use_sharing = false;
info->plan_cost *= gp_coefficient_1phase_agg;
return info->plan_cost;
}
/* Function cost_2phase_aggregation
*
* May be used for 2 phase costing with 0 or 1 DQAs.
* Corresponds to make_two_stage_agg_plan and must be maintained in sync
* with it.
*/
Cost cost_2phase_aggregation(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info)
{
Plan input_dummy;
bool is_sorted;
long numGroups = (*(ctx->p_dNumGroups) < 0) ? 0 :
(*(ctx->p_dNumGroups) > LONG_MAX) ? LONG_MAX :
(long)*(ctx->p_dNumGroups);
double input_rows;
double streaming_fudge = 1.3;
cost_common_agg(root, ctx, info, &input_dummy);
input_rows = input_dummy.plan_rows;
is_sorted = pathkeys_contained_in(root->group_pathkeys, info->input_path->pathkeys);
/* Precondition Input */
switch ( info->group_prep )
{
case MPP_GRP_PREP_HASH_DISTINCT:
input_dummy.total_cost +=
incremental_motion_cost(input_dummy.plan_rows,
input_dummy.plan_rows);
is_sorted = false;
break;
case MPP_GRP_PREP_NONE:
break;
default:
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("unexpected call for two-stage aggregation")));
break; /* Never */
}
/* Preliminary Aggregation */
if ( ctx->use_hashed_grouping )
{
/* Preliminary HashAgg*/
add_agg_cost(NULL, &input_dummy,
NIL, NIL, /* Don't know preliminary tlist, qual IS NIL */
AGG_HASHED, root->config->gp_hashagg_streambottom,
ctx->numGroupCols, ctx->groupColIdx,
numGroups, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
if ( gp_hashagg_streambottom )
{
input_dummy.plan_rows *= streaming_fudge;
}
}
else
{
if ( ctx->numGroupCols == 0 )
{
/* Preliminary PlainAgg*/
add_agg_cost(NULL, &input_dummy,
NIL, NIL, /* Don't know preliminary tlist, qual IS NIL */
AGG_PLAIN, false,
0, NULL,
1, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
else
{
/* Preliminary GroupAgg */
if ( ! is_sorted )
{
add_sort_cost(NULL, &input_dummy, ctx->numGroupCols, NULL, NULL);
}
add_agg_cost(NULL, &input_dummy,
NIL, NIL, /* Don't know preliminary tlist, qual IS NIL */
AGG_SORTED, false,
ctx->numGroupCols, ctx->groupColIdx,
numGroups, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
/* Beware: AGG_PLAIN and AGG_GROUPED may be used with DQAs, however,
* the function cost_agg doesn't distinguish DQAs so it
* consistently under estimates the cost in these cases.
*/
if ( ctx->numDistinctCols > 0 )
{
Path path_dummy;
Cost run_cost;
double ngrps = *(ctx->p_dNumGroups);
double avgsize = input_rows / ngrps;
Assert(ctx->numDistinctCols == 1);
cost_sort(&path_dummy, NULL, NIL, input_dummy.total_cost, avgsize, 32);
run_cost = path_dummy.total_cost - path_dummy.startup_cost;
input_dummy.total_cost += path_dummy.startup_cost + ngrps * run_cost;
}
}
/* Collocate groups */
switch ( info->group_type )
{
case MPP_GRP_TYPE_GROUPED_2STAGE: /* Redistribute */
input_dummy.total_cost +=
incremental_motion_cost(input_dummy.plan_rows,
input_dummy.plan_rows);
break;
case MPP_GRP_TYPE_PLAIN_2STAGE: /* Gather */
input_dummy.total_cost +=
incremental_motion_cost(input_dummy.plan_rows,
input_dummy.plan_rows *root->config->cdbpath_segments);
break;
default:
ereport(ERROR,
(errcode(ERRCODE_CDB_INTERNAL_ERROR),
errmsg("unexpected call for two-stage aggregation")));
break; /* Never */
}
/* Final Aggregation */
if ( ctx->use_hashed_grouping )
{
/* HashAgg*/
add_agg_cost(NULL, &input_dummy,
NIL, NIL, /* Don't know tlist or qual */
AGG_HASHED, false,
ctx->numGroupCols, ctx->groupColIdx,
numGroups, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
else
{
if ( ctx->numGroupCols == 0 )
{
/* PlainAgg*/
add_agg_cost(NULL, &input_dummy,
NIL, NIL, /* Don't know tlist or qual */
AGG_PLAIN, false,
0, NULL,
1, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
else
{
/* GroupAgg */
add_sort_cost(NULL, &input_dummy, ctx->numGroupCols, NULL, NULL);
add_agg_cost(NULL, &input_dummy,
NIL, NIL, /* Don't know tlist or qual */
AGG_SORTED, false,
ctx->numGroupCols, ctx->groupColIdx,
numGroups, 0, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
}
}
info->plan_cost = root->config->gp_eager_two_phase_agg ? (Cost)0.0 : input_dummy.total_cost;
info->valid = true;
info->join_strategy = DqaJoinNone;
info->use_sharing = false;
info->plan_cost *= gp_coefficient_2phase_agg;
return info->plan_cost;
}
/* Function cost_3phase_aggregation
*
* Only used for costing with 2 or more DQAs.
* Corresponds to make_three_stage_agg_plan and must be maintained in sync
* with it.
*
* This function assumes the enviroment established by planDqaJoinOrder()
* and set_coplan_strategies().
*/
Cost cost_3phase_aggregation(PlannerInfo *root, MppGroupContext *ctx, AggPlanInfo *info)
{
Plan dummy;
double input_rows;
Cost total_cost;
Cost share_cost;
Cost reexec_cost;
int i;
bool use_sharing = false;
DqaJoinStrategy join_strategy = DqaJoinUndefined;
Cost input_cost = 0.0;
Cost cost_coplan_cheapest = 0.0;
Cost cost_coplan_sorted = 0.0;
Cost cost_hashjoin = 0.0;
Cost cost_mergejoin = 0.0;
Cost cost_crossjoin = 0.0;
cost_common_agg(root, ctx, info, &dummy);
input_rows = dummy.plan_rows;
total_cost = dummy.total_cost;
Assert( ctx->numDistinctCols == list_length(ctx->agg_counts->dqaArgs) );
/* Note that join order has already been established by an earlier
* call to planDqaJoinOrder. Here we just use that order, but we
* need to decide on the join type.
*/
if ( list_length(ctx->agg_counts->dqaArgs) < 2 )
{
join_strategy = DqaJoinNone;
}
else if ( ctx->numGroupCols == 0 )
{
join_strategy = DqaJoinCross;
}
else if ( sorting_prefixes_grouping(root) )
{
/* Cheapest of merge join of sorted input or sorted hash join */
join_strategy = DqaJoinSorted;
}
else
{
/* Cheapest of merge join of sorted input or hash join */
join_strategy = DqaJoinCheapest;
}
/* Determine whether to use input sharing. */
if ( ctx->numDistinctCols < 2 )
{
reexec_cost = total_cost;
use_sharing = false;
}
else
{
/* Decide based on apparent costs.
* XXX Do we need to override this if there are volatile functions
* in the common plan? Is this known, or do we need to search?
*/
share_cost = cost_share_plan(&dummy, root, ctx->numDistinctCols);
reexec_cost = ctx->numDistinctCols * total_cost;
use_sharing = share_cost < reexec_cost;
}
input_cost = use_sharing ? share_cost : reexec_cost;
/* Get costs for the ctx->numDistinctCols coplans. */
cost_coplan_cheapest = cost_coplan_sorted = 0;
for ( i = 0; i < ctx->numDistinctCols; i++ )
{
DqaInfo *arg = ctx->dqaArgs + i;
cost_coplan_cheapest += arg->cost_cheapest;
cost_coplan_sorted += arg->cost_sorted;
}
/* Get costs to join the coplan results. */
switch ( join_strategy )
{
case DqaJoinNone:
break;
case DqaJoinCross:
cost_crossjoin = (ctx->numDistinctCols - 1) * 2 * cpu_tuple_cost;
break;
case DqaJoinSorted:
case DqaJoinCheapest:
set_cost_of_join_strategies(ctx, &cost_hashjoin, &cost_mergejoin);
if ( join_strategy == DqaJoinSorted )
cost_hashjoin += incremental_sort_cost(*ctx->p_dNumGroups, 100, ctx->numDistinctCols);
cost_hashjoin += cost_coplan_cheapest;
cost_mergejoin += cost_coplan_sorted;
if ( cost_hashjoin > 0.0 && cost_hashjoin <= cost_mergejoin )
{
join_strategy = DqaJoinHash;
}
else
{
join_strategy = DqaJoinMerge;
}
break;
default:
elog(ERROR, "unexpected join strategy during DQA planning");
}
/* Compare costs choose cheapest. */
switch ( join_strategy )
{
case DqaJoinNone:
total_cost = input_cost + cost_coplan_cheapest;
break;
case DqaJoinCross:
total_cost = input_cost + cost_coplan_cheapest + cost_crossjoin;
break;
case DqaJoinHash:
total_cost = input_cost + cost_coplan_cheapest + cost_hashjoin;
break;
case DqaJoinMerge:
total_cost = input_cost + cost_coplan_cheapest + cost_mergejoin;
break;
default:
elog(ERROR, "unexpected join strategy during DQA planning");
}
info->plan_cost = root->config->gp_eager_dqa_pruning ? (Cost)0.0 : total_cost;
info->valid = true;
info->join_strategy = join_strategy;
info->use_sharing = use_sharing;
info->plan_cost *= gp_coefficient_3phase_agg;
return info->plan_cost;
}
/* Estimate the costs of
* 1. HashJoin of cheapest inputs, and
* 2. MergeJoin of sorted input.
*
* If result should be ordered, compare a Sort of 1 with 2.
* Else compare 1 with 2.
*/
void set_cost_of_join_strategies(MppGroupContext *ctx, Cost *hashjoin_cost, Cost *mergejoin_cost)
{
Cost hj_cost;
Cost mj_cost;
List *mergeclauses = NIL;
List *hashclauses = NIL;
double rows;
int gk_width;
int outer_width;
bool try_hashed = true;
AttrNumber attrno;
Index outer_varno = 1;
int i;
rows = *ctx->p_dNumGroups;
/* Widths are wild speculation, but good enough, we hope. */
gk_width = 32 * ctx->numGroupCols;
outer_width = 32; /* DQA transition values for first DQA arg. */
outer_width += 64; /* Ordinary aggregate transition values. */
/* We need join clauses for costing. */
for( i = 0; i < ctx->numGroupCols; i++ )
{
Expr *qual;
Var *outer_var;
Var *inner_var;
AttrNumber resno = ctx->groupColIdx[i];
Index inner_varno = 1 + (i + 1);
TargetEntry *tle = get_tle_by_resno(ctx->sub_tlist, resno);
Assert( tle != NULL );
outer_var = makeVar(outer_varno, resno,
exprType((Node *)tle->expr),
exprTypmod((Node *)tle->expr), 0);
inner_var = makeVar(inner_varno, resno,
exprType((Node *)tle->expr),
exprTypmod((Node *)tle->expr), 0);
/* outer should always be on the left */
qual = make_op(NULL, list_make1(makeString("=")),
(Node*) outer_var,
(Node*) inner_var, -1);
/* If the grouping column is not hashable, do not try hashing. */
if (!hash_safe_type(exprType((Node *)tle->expr)))
try_hashed = false;
if ( try_hashed )
{
hashclauses = lappend(hashclauses, copyObject(qual));
}
qual->type = T_DistinctExpr;
qual = make_notclause(qual);
mergeclauses = lappend(mergeclauses, qual);
}
/* Estimate the incremental join costs. */
hj_cost = mj_cost = 0;
for ( attrno = 1; attrno < ctx->numDistinctCols; attrno++ )
{
int dqa_width = 32;
int inner_width = gk_width + dqa_width;
mj_cost += incremental_mergejoin_cost(rows, mergeclauses, ctx->root);
if ( try_hashed )
hj_cost += incremental_hashjoin_cost(rows, inner_width, outer_width, hashclauses, ctx->root);
outer_width += dqa_width;
}
*mergejoin_cost = mj_cost;
*hashjoin_cost = try_hashed ? hj_cost : 0.0;
}
/* Set up basic structure content. Caller to fill in.
*/
static
void initAggPlanInfo(AggPlanInfo *info, Path *input_path, Plan *input_plan)
{
info->input_path = input_path;
info->input_plan = input_plan;
if (input_path != NULL)
info->input_locus = input_path->locus;
else
CdbPathLocus_MakeNull(&info->input_locus);
info->group_type = MPP_GRP_TYPE_BASEPLAN;
info->group_prep = MPP_GRP_PREP_NONE;
CdbPathLocus_MakeNull(&info->output_locus);
info->distinctkey_collocate = false;
info->valid = false;
info->plan_cost = 0;
info->join_strategy = DqaJoinUndefined;
info->use_sharing = false;
}
/* set_coplan_strategies
*
* Determine and cache in the given DqaInfo structure the cheapest
* strategy that computes the answer and the cheapest strategy that
* computes the answer in grouping key order.
*
* Below, the result cardinalities are shown as <-n- where
*
* x (input_rows) is the input cardinality which is usually about
* equal to #segments * #distinct(grouping key, DQA arg)
*
* d (darg_rows) is #distinct(grouping key, DQA arg)
*
* g (group_rows) is #distinct(grouping key)
*
* The coplan through the Motion that collocates tuples on the
* grouping key is planned independently and will be one of
*
* <- Motion <-x- HashAgg <-t- R
* <- Motion <-x- GroupAgg <- Sort <-t- R
*
* which is encoded in DqaInfo by the flag use_hashed_preliminary.
*
* The possible post-Motion strategies are encoded as enum values of
* type DqaCoplanType and indicate all the required plan nodes.
*
* Vector aggregation strategies that produce a result ordered on the
* grouping key are:
*
* DQACOPLAN_GGS: <-g- GroupAgg <-d- GroupAgg <- Sort <-x-
* DQACOPLAN_GSH: <-g- GroupAgg <- Sort <-d- HashAgg <-x-
* DQACOPLAN_SHH: <- Sort <-g- HashAgg <-d- HashAgg <-x-
*
* In addition, the vector aggreagation strategy
*
* DQACOPLAN_HH: <-g- HashAgg <-d- HashAgg <-x- R
*
* produces an unordered result.
*
* Scalar aggregation strategies are:
*
* DQACOPLAN_PGS: <-1- PlainAgg <-d- GroupAgg <- Sort <-x- R
* DQACOPLAN_PH: <-1- PlainAgg <-d- HashedAgg <-x- R
*
*/
void set_coplan_strategies(PlannerInfo *root, MppGroupContext *ctx, DqaInfo *dqaArg, Path *input)
{
double input_rows = input->parent->rows;
int input_width = input->parent->width;
double darg_rows = dqaArg->num_rows;
double group_rows = *ctx->p_dNumGroups;
long numGroups = (group_rows < 0) ? 0 :
(group_rows > LONG_MAX) ? LONG_MAX :
(long)group_rows;
bool can_hash_group_key = ctx->agg_counts->canHashAgg;
bool can_hash_dqa_arg = dqaArg->can_hash;
bool use_hashed_preliminary = false;
Cost sort_input = incremental_sort_cost(input_rows, input_width,
ctx->numGroupCols+1);
Cost sort_dargs = incremental_sort_cost(darg_rows, input_width,
ctx->numGroupCols);
Cost sort_groups = incremental_sort_cost(group_rows, input_width,
ctx->numGroupCols);
Cost gagg_input = incremental_agg_cost(input_rows, input_width,
AGG_SORTED, ctx->numGroupCols+1,
numGroups, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
Cost gagg_dargs = incremental_agg_cost(darg_rows, input_width,
AGG_SORTED, ctx->numGroupCols,
numGroups, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
Cost hagg_input = incremental_agg_cost(input_rows, input_width,
AGG_HASHED, ctx->numGroupCols+1,
numGroups, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
Cost hagg_dargs = incremental_agg_cost(darg_rows, input_width,
AGG_HASHED, ctx->numGroupCols,
numGroups, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
Cost cost_base;
Cost cost_sorted;
Cost cost_cheapest;
DqaCoplanType type_sorted;
DqaCoplanType type_cheapest;
Cost trial;
/* Preliminary aggregation */
use_hashed_preliminary = ( can_hash_group_key || ctx->numGroupCols == 0 )
&& can_hash_dqa_arg;
if ( use_hashed_preliminary )
{
cost_base = hagg_input;
}
else
{
cost_base = sort_input + gagg_input;
}
/* Collocating motion */
cost_base += incremental_motion_cost(darg_rows, darg_rows);
/* Post-motion processing is more complex. */
if ( ctx->numGroupCols == 0 ) /* scalar agg */
{
Cost pagg_dargs = incremental_agg_cost(darg_rows, input_width,
AGG_PLAIN, 0,
1, ctx->agg_counts->numAggs,
ctx->agg_counts->transitionSpace);
type_sorted = type_cheapest = DQACOPLAN_PGS;
cost_sorted = cost_cheapest = sort_input + gagg_input + pagg_dargs;
trial = hagg_input + pagg_dargs;
if (trial < cost_cheapest )
{
cost_cheapest = trial;
type_cheapest = DQACOPLAN_PH;
}
}
else /* vector agg */
{
type_sorted = type_cheapest = DQACOPLAN_GGS;
cost_sorted = cost_cheapest = sort_input + gagg_input + gagg_dargs;
if ( can_hash_dqa_arg )
{
trial = hagg_input + sort_dargs + gagg_input;
if ( trial < cost_cheapest )
{
cost_cheapest = trial;
type_cheapest = DQACOPLAN_GSH;
}
if ( trial < cost_sorted )
{
cost_sorted = trial;
type_sorted = DQACOPLAN_GSH;
}
}
if ( can_hash_group_key && can_hash_dqa_arg )
{
trial = hagg_input + hagg_dargs;
if ( trial < cost_cheapest )
{
cost_cheapest = trial;
type_cheapest = DQACOPLAN_HH;
}
trial += sort_groups;
if ( trial < cost_sorted )
{
cost_sorted = trial;
type_sorted = DQACOPLAN_SHH;
}
}
}
dqaArg->use_hashed_preliminary = use_hashed_preliminary;
dqaArg->cost_sorted = cost_base + cost_sorted;
dqaArg->coplan_type_sorted = type_sorted;
dqaArg->cost_cheapest = cost_base + cost_cheapest;
dqaArg->coplan_type_cheapest = type_cheapest;
dqaArg->distinctkey_collocate = false;
}
/* incremental_sort_cost -- helper for set_coplan_strategies
*/
Cost incremental_sort_cost(double rows, int width, int numKeyCols)
{
Plan dummy;
memset(&dummy, 0, sizeof(dummy));
dummy.plan_rows = rows;
dummy.plan_width = width;
add_sort_cost(NULL, &dummy, numKeyCols, NULL, NULL);
return dummy.total_cost;
}
/* incremental_agg_cost -- helper for set_coplan_strategies
*/
Cost incremental_agg_cost(double rows, int width, AggStrategy strategy,
int numGroupCols, double numGroups,
int numAggs, int transSpace)
{
Plan dummy;
memset(&dummy, 0, sizeof(dummy));
dummy.plan_rows = rows;
dummy.plan_width = width;
add_agg_cost(NULL, &dummy,
NULL, NULL,
strategy, false,
numGroupCols, NULL,
numGroups, 0, numAggs, transSpace);
return dummy.total_cost;
}
/* incremental_motion_cost -- helper for set_coplan_strategies
*/
Cost incremental_motion_cost(double sendrows, double recvrows)
{
Cost cost_per_row = (gp_motion_cost_per_row > 0.0)
? gp_motion_cost_per_row
: 2.0 * cpu_tuple_cost;
return cost_per_row * 0.5 * (sendrows + recvrows);
}
/*
* choose_deduplicate
* numGroups is an output parameter that is estimated number of unique groups.
*
* Ideally, we could estimate all the costs based on what will happen and
* choose either of a naive or a de-duplicate plan. However, it is nearly
* impossible because there are plans decided by other components such as
* cdb_grouping_planner() and many branches including Redistribute is
* needed or not. In addition, estimated number of groups in multi-column
* case (i.e. GROUP BY + ORDER BY) is far from the truth. We could improve
* the code by adding more cost calculations, but it is reasonable to
* estimate the major costs of initial sort/aggregate for now.
*/
static bool
choose_deduplicate(PlannerInfo *root, List *sortExprs,
Plan *input_plan, double *numGroups)
{
double num_distinct;
double input_rows = input_plan->plan_rows;
Path dummy_path;
Cost naive_cost, dedup_cost;
int32 width;
AggStrategy aggstrategy;
int numGroupCols;
naive_cost = 0;
dedup_cost = 0;
width = input_plan->plan_width;
/* Add int8 column anyway which holds count */
width += get_typavgwidth(INT8OID, -1);
numGroupCols = list_length(sortExprs);
/*
* First, calculate cost of naive case.
*/
cost_sort(&dummy_path, root, NIL,
input_plan->total_cost,
input_plan->plan_rows,
input_plan->plan_width);
naive_cost = dummy_path.total_cost;
/*
* Next, calculate cost of deduplicate.
* The first aggregate calculates number of duplicate for
* each unique sort key, then we add cost of sort after
* the aggregate.
*/
num_distinct = estimate_num_groups(root, sortExprs, input_rows);
aggstrategy = AGG_HASHED;
cost_agg(&dummy_path, root, aggstrategy, 1, numGroupCols, num_distinct,
input_plan->startup_cost, input_plan->total_cost,
input_plan->plan_rows, input_plan->plan_width,
0, 0, false);
dummy_path.total_cost +=
incremental_motion_cost(num_distinct,
num_distinct * root->config->cdbpath_segments);
cost_sort(&dummy_path, root, NIL,
dummy_path.total_cost,
num_distinct,
width);
dedup_cost = dummy_path.total_cost;
if (numGroups)
*numGroups = num_distinct;
/* we need some calculates above even if the flag is off */
if (pg_strcasecmp(gp_idf_deduplicate_str, "force") == 0)
return true;
if (pg_strcasecmp(gp_idf_deduplicate_str, "none") == 0)
return false;
return dedup_cost < naive_cost;
}
/*
* wrap_plan_index
*
* A wrapper of wrap_plan. The column name aliases are retrieved from
* sub plan's target list automatically. The main additional operation here
* is to modify varno on each vars in either the new target list and
* hashExpr (qual is not set by wrap_plan.) While wrap_plan creates
* Query with the initial RangeTblEntry with index 1 and this is right,
* we sometimes need to wrap plan by a subquery scan as a part of upper join.
* In that case, varno is not necessarily 1.
*
* If varno > 1, the vars' references are updated. Thus,
* (*query_p)->rtable is not valid anymore, since rtable should have all
* RangeTblEntry up to varno. Still, we leave it in rtable field
* for the caller to use the rte from rtable.
*
* Note that the target list of the plan and query is identical and
* represents the "one on generated plan," which means the caller still needs
* to update other target list like upper parser->targetList. This routines
* does not handle it.
*
* This is in the middle of wrap_plan() and add_subqueryscan(). We actually
* need a more sophisticated and all-round way.
*
* This also updates locus if given, and root->group_pathkeys to match
* the subquery targetlist.
*/
static Plan *
wrap_plan_index(PlannerInfo *root, Plan *plan, Query *query,
List **p_pathkeys, Index varno, const char *alias_name,
Query **query_p)
{
ListCell *l;
plan = wrap_plan(root, plan, query, p_pathkeys,
alias_name, NIL, query_p);
Assert(varno > 0);
if (varno != 1)
{
foreach (l, plan->flow->hashExpr)
{
Var *var = lfirst(l);
if (IsA(var, Var))
{
/* fix varno, which is set to 1 in wrap_plan */
Assert(var->varno == 1);
var->varno = var->varnoold = varno;
}
}
/*
* Currently, plan and new parse tree shares target list.
* If this breaks, we'll need to update parse's target list as well.
*/
Assert(plan->targetlist == (*query_p)->targetList);
foreach (l, plan->targetlist)
{
TargetEntry *tle = lfirst(l);
Var *var = (Var *) tle->expr;
if (IsA(var, Var))
{
/* fix varno, which is set to 1 in wrap_plan */
Assert(var->varno == 1);
var->varno = var->varnoold = varno;
}
}
Assert(IsA(plan, SubqueryScan));
((SubqueryScan *) plan)->scan.scanrelid = varno;
Assert(plan->qual == NIL);
}
/*
* Update group_pathkeys in order to represent this subquery.
*/
root->group_pathkeys =
make_pathkeys_for_groupclause(root->parse->groupClause, plan->targetlist);
return plan;
}
/*
* rebuilt_simple_rel_and_rte
*
* Rebuild arrays for RelOptInfo and RangeTblEntry for the PlannerInfo when
* the underlying range tables are transformed. They can be rebuilt from
* parse->rtable, and will be used in later than here by processes like
* distinctClause planning. We never pfree the original array, since
* it's potentially used by other PlannerInfo which this is copied from.
*/
static void
rebuild_simple_rel_and_rte(PlannerInfo *root)
{
int i;
int array_size;
ListCell *l;
array_size = list_length(root->parse->rtable) + 1;
root->simple_rel_array_size = array_size;
root->simple_rel_array =
(RelOptInfo **) palloc0(sizeof(RelOptInfo *) * array_size);
root->simple_rte_array =
(RangeTblEntry **) palloc0(sizeof(RangeTblEntry *) * array_size);
i = 1;
foreach (l, root->parse->rtable)
{
(void) build_simple_rel(root, i, RELOPT_BASEREL);
root->simple_rte_array[i] = lfirst(l);
i++;
}
}
/*
* make_parallel_or_sequential_agg
*
* This is the common pattern to create multi-phase aggregate by cdb_grouping_planner
* or single agg by make_agg. The code was stolen from grouping_planner, but a little
* simplified under some assumptions to be used in limited condition. Here we assume
* we don't have GROUPING SETS and there is at least one aggregate. This function
* assumes the caller has already set up subplan as group_context->subplan.
*
* The caller may not pass current_pathkeys_p if it's not interested.
*/
static Plan *
make_parallel_or_sequential_agg(PlannerInfo *root, AggClauseCounts *agg_counts,
GroupContext *group_context, List **current_pathkeys_p)
{
Plan *result_plan;
List *current_pathkeys;
/*
* current_pathkeys_p can be NULL, which means the caller isn't interested in
* the pathkeys. Still, we are.
*/
if (current_pathkeys_p)
current_pathkeys = *current_pathkeys_p;
else
current_pathkeys = NIL;
result_plan = cdb_grouping_planner(root, agg_counts, group_context);
if (!result_plan)
{
/*
* If cdb_grouping_planner doesn't return a plan,
* it means the plan should fall back to sequential.
* In that case, multi-phase aggregate plan is not used.
* Here it's much simpler than grouping_planner,
* since we are sure we have at least one aggregate function
* and no GROUPING SETS.
*/
AggStrategy aggstrategy;
result_plan = group_context->subplan;
if (group_context->use_hashed_grouping)
{
/*
* HashAggregate case
*/
aggstrategy = AGG_HASHED;
current_pathkeys = NIL;
}
else
{
/*
* GroupAggregate case
*/
if (root->parse->groupClause)
{
if (!pathkeys_contained_in(root->group_pathkeys,
current_pathkeys))
{
result_plan = (Plan *)
make_sort_from_groupcols(root,
root->parse->groupClause,
group_context->groupColIdx,
false,
result_plan);
mark_sort_locus(result_plan);
}
aggstrategy = AGG_SORTED;
current_pathkeys =
make_pathkeys_for_groupclause(root->parse->groupClause, result_plan->targetlist);
current_pathkeys = canonicalize_pathkeys(root, current_pathkeys);
}
else
{
/*
* No GROUP BY case
*/
aggstrategy = AGG_PLAIN;
current_pathkeys = NIL;
}
}
/*
* Now make a single Agg node.
*/
result_plan = (Plan *) make_agg(root,
root->parse->targetList,
(List *) root->parse->havingQual,
aggstrategy,
false,
group_context->numGroupCols,
group_context->groupColIdx,
*group_context->p_dNumGroups,
0, /* num_nullcols */
0, /* input_grouping */
0, /* grouping */
0, /* rollup_gs_times */
agg_counts->numAggs,
agg_counts->transitionSpace,
result_plan);
mark_passthru_locus(result_plan, true, current_pathkeys != NIL);
}
else
current_pathkeys = *group_context->pcurrent_pathkeys;
if (current_pathkeys_p)
*current_pathkeys_p = current_pathkeys;
return result_plan;
}
/*
* deconstruct_within_agg
* deconstruct within-aggregate and normal aggregate expressions to sub pieces.
*
* In the split-DQA-join pattern, we want to deconstruct individual aggregate
* or percentile expressions, group them, and relocate them to the appropriate
* sub-plan target list. The information about which expression goes to which
* sub-plan is stored in MppGroupContext structure, and the function returns
* Var that points to the expression on sub-plan which the original expression
* is replaced with.
*
* See also similar deconstruct_expr() for the general aggregate case.
*/
static Node *
deconstruct_within_agg(Node *node, MppGroupContext *ctx)
{
return deconstruct_within_agg_mutator(node, ctx);
}
/*
* deconstruct_within_agg_mutator
*
* The workhorse of deconstruct_within_agg()
*/
static Node *
deconstruct_within_agg_mutator(Node *node, MppGroupContext *ctx)
{
TargetEntry *tle;
if (node == NULL)
return NULL;
if (IsA(node, Aggref) || IsA(node, PercentileExpr))
{
Index tlistno;
AttrNumber attno;
List *sortclauses, *dref_tlist;
ListCell *l;
Node *final_node;
/*
* Here we may see normal aggregates, not only percentiles.
* If normal aggs are involved, ctx->wagSortClauses should have
* NIL elements for it.
*/
if (IsA(node, PercentileExpr))
sortclauses = ((PercentileExpr *) node)->sortClause;
else
sortclauses = NIL;
/*
* Find the right sub-plan which this expression should go.
*/
tlistno = 0;
foreach (l, ctx->wagSortClauses)
{
/* Note NIL can be equal to NIL, too. */
if (equal(sortclauses, lfirst(l)))
break;
tlistno++;
}
/* Not found? Should not happen... */
if (!l)
elog(ERROR, "unexpected use of aggregate");
dref_tlist = ctx->dref_tlists[tlistno];
/*
* If the same expression exists at the same level, recycle it.
* Otherwise, create a new expression.
*/
tle = tlist_member(node, dref_tlist);
attno = list_length(dref_tlist) + 1;
if (!tle)
{
/*
* Don't copy node, share it with tlist, for later operation
* can modify the var reference in tlist.
*/
tle = makeTargetEntry((Expr *) node,
attno, NULL, false);
dref_tlist = lappend(dref_tlist, tle);
}
else
attno = tle->resno;
ctx->dref_tlists[tlistno] = dref_tlist;
final_node = (Node *) makeVar(dqa_base_varno + tlistno,
attno, exprType((Node *) tle->expr),
exprTypmod((Node *) tle->expr), 0);
return final_node;
}
/*
* If the given expression is a grouping expression, replace it with
* a Var node referring to the (lower) preliminary aggregation's
* target list.
*/
tle = tlist_member(node, ctx->grps_tlist);
if (tle != NULL)
return (Node *) makeVar(grp_varno, tle->resno,
exprType((Node *) tle->expr),
exprTypmod((Node *) tle->expr), 0);
return expression_tree_mutator(node, deconstruct_within_agg_mutator, ctx);
}
/*
* Returns percentile expressions associated with the specified sortClause.
*
* Because we want to keep percentile expressions in target list or having list
* to make any changes to them identically, we extract those expressions
* every time we need. It is more valuable here to keep consistent than to reduce
* a few cycles by copying those nodes.
*/
static List *
fetch_percentiles(Query *parse, List *sortClause)
{
List *nodes, *result;
ListCell *l;
nodes = list_concat(extract_nodes(NULL, (Node *) parse->targetList, T_PercentileExpr),
extract_nodes(NULL, (Node *) parse->havingQual, T_PercentileExpr));
nodes = list_concat(nodes,
extract_nodes(NULL, (Node *) parse->scatterClause, T_PercentileExpr));
result = NIL;
foreach (l, nodes)
{
PercentileExpr *p = lfirst(l);
if (equal(sortClause, p->sortClause))
result = lappend(result, p);
}
return result;
}
/*
* make_deduplicate_plan
*
* Build a plan that removes duplicate rows with GROUP BY + ORDER BY (in WITHIN)
* clause. It may or may not use two phase aggregate inside. Note the output
* plan has different target list than non-deduplicate plan, since the aggregate
* process cannot process un-grouped columns.
*/
static Plan *
make_deduplicate_plan(PlannerInfo *root,
GroupContext *group_context,
List *groupClause,
List *sortClause,
double numGroups,
AttrNumber *pc_pos_p,
List **current_pathkeys_p,
Plan *subplan)
{
Plan *result_plan;
Aggref *aggref;
GroupContext ctx;
ListCell *l1, *l2;
TargetEntry *pc_tle;
List *tlist;
int numGroupCols;
AttrNumber *groupColIdx;
List *pathkeys = NIL;
bool querynode_changed = false;
AggClauseCounts agg_counts;
bool use_hashed_grouping;
Query *original_parse, *parse;
List *original_group_pathkeys;
List *sub_tlist = group_context->sub_tlist;
Expr *tvexpr; /* ORDER BY expression */
const Index Outer = 1;
/*
* It is doable to just concatenate groupClause and sortClause,
* but it is more semantic to convert sortClause to groupClause.
* Especially we want to use make_pathkeys_from_grouplcause later where
* sortClause is not handled.
*
* Copy input groupClause, since we change it.
*/
groupClause = copyObject(groupClause);
foreach (l1, sortClause)
{
SortClause *sc = copyObject(lfirst(l1));
sc->type = T_GroupClause;
groupClause = lappend(groupClause, sc);
}
groupColIdx = get_grouplist_colidx(groupClause, sub_tlist, &numGroupCols);
/*
* Make target list derived from sub_tlist. Note that we filter out
* ungrouped columns which will be bogus after the aggregate.
*/
tlist = NIL;
foreach (l1, sub_tlist)
{
TargetEntry *tle = lfirst(l1);
TargetEntry *newtle;
/*
* Check if this target is a part of grouping columns.
*/
foreach (l2, groupClause)
{
GroupClause *gc = lfirst(l2);
if (gc->tleSortGroupRef == tle->ressortgroupref)
break;
}
/* Found, so add it */
if (l2)
{
newtle = flatCopyTargetEntry(tle);
newtle->resno = list_length(tlist) + 1;
tlist = lappend(tlist, newtle);
}
}
/*
* Count ORDER BY expression so that since NULL input should
* be ignored. We still need not to eliminate NULL input since
* the result should be returned per group even if the group
* has nothing but NULL.
*/
tvexpr = (Expr *)
get_sortgroupclause_expr(linitial(sortClause), sub_tlist);
/*
* Append peer count expression to target list.
*/
*pc_pos_p = list_length(tlist) + 1;
aggref = makeAggrefByOid(AGGFNOID_COUNT_ANY, list_make1(tvexpr));
pc_tle = makeTargetEntry((Expr *) aggref,
*pc_pos_p, "peer_count", false);
tlist = lappend(tlist, pc_tle);
MemSet(&agg_counts, 0, sizeof(AggClauseCounts));
count_agg_clauses((Node *) tlist, &agg_counts);
use_hashed_grouping = choose_hashed_grouping(root,
group_context->tuple_fraction,
group_context->cheapest_path,
NULL,
numGroups,
&agg_counts);
use_hashed_grouping = agg_counts.canHashAgg;
ctx.best_path = group_context->best_path;
ctx.cheapest_path = group_context->cheapest_path;
ctx.subplan = subplan;
ctx.sub_tlist = NIL;
ctx.tlist = tlist;
ctx.use_hashed_grouping = use_hashed_grouping;
ctx.tuple_fraction = 0.1;
ctx.canonical_grpsets = NULL;
ctx.grouping = 0;
ctx.numGroupCols = numGroupCols;
ctx.groupColIdx = groupColIdx;
ctx.numDistinctCols = 0;
ctx.distinctColIdx = NULL;
ctx.p_dNumGroups = &numGroups;
ctx.pcurrent_pathkeys = &pathkeys;
ctx.querynode_changed = &querynode_changed;
original_parse = root->parse;
original_group_pathkeys = root->group_pathkeys;
root->parse = parse = copyObject(root->parse);
root->parse->groupClause = groupClause;
root->parse->targetList = tlist;
root->parse->havingQual = NULL;
root->parse->scatterClause = NIL;
root->group_pathkeys = make_pathkeys_for_groupclause(groupClause, tlist);
/*
* Make a multi-phase or simple agg plan.
*/
result_plan = make_parallel_or_sequential_agg(root,
&agg_counts,
&ctx,
current_pathkeys_p);
root->parse = original_parse;
root->group_pathkeys = original_group_pathkeys;
/*
* Add SubqueryScan to wrap this anyway, so that
* the outcome of deduplicate can be treated as a simple subquery relation.
*/
result_plan = wrap_plan_index(root,
result_plan,
parse,
current_pathkeys_p,
Outer,
"deduplicate",
&parse);
/*
* Update the reference based on the aggregate target list. This ensures
* the target list of parse tree is pointing the top SubqueryScan's Vars.
*/
root->parse->targetList = (List *)
cdbpullup_expr((Expr *) root->parse->targetList, tlist, NIL, Outer);
root->parse->havingQual = (Node *)
cdbpullup_expr((Expr *) root->parse->havingQual, tlist, NIL, Outer);
root->parse->scatterClause = (List *)
cdbpullup_expr((Expr *) root->parse->scatterClause, tlist, NIL, Outer);
root->parse->rtable = parse->rtable;
// Rebuild arrays for RelOptInfo and RangeTblEntry for the PlannerInfo
// since the underlying range tables have been transformed
rebuild_simple_rel_and_rte(root);
return result_plan;
}
/*
* within_agg_make_baseplan
*
* Creates the scaffold plan which will be shared between the outer and
* inner plans. It may choose de-duplicate plan based on costs and GUC.
* The output plan is always a closed SubqueryScan. This will set
* pc_pos in wag_context.
*/
static Plan *
within_agg_make_baseplan(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
List *sortClause,
Plan *result_plan)
{
List *sub_tlist = group_context->sub_tlist;
double dedup_numGroups;
List *dedup_key_exprs;
/*
* The GROUP BY keys are the normal grouping keys + sort key.
*/
dedup_key_exprs = list_concat(
get_sortgrouplist_exprs(root->parse->groupClause, sub_tlist),
get_sortgrouplist_exprs(sortClause, sub_tlist));
/*
* Decide whether deduplicate is useful or not.
*/
wag_context->use_deduplicate = choose_deduplicate(root,
dedup_key_exprs,
result_plan,
&dedup_numGroups);
/*
* Create the base subplan for the upper join. We may take
* decuplicate way or not, but anyway the target list of result_plan has
* an extra target entry for the peer count.
*/
if (wag_context->use_deduplicate)
{
/*
* The deduplicate optimization. We reduce identical rows
* and record the number of reduced rows, so that
* percentile function can see the original rows.
* It's similar to the run-length encoding.
*
* root->parse is updated inside to represent this subquery.
*/
result_plan = make_deduplicate_plan(root,
group_context,
root->parse->groupClause,
sortClause,
dedup_numGroups,
&wag_context->pc_pos,
&wag_context->current_pathkeys,
result_plan);
}
else
{
Query *subquery;
Expr *tv_expr;
NullTest *nt;
CaseWhen *casearg;
CaseExpr *pc_expr;
TargetEntry *pc_tle;
/*
* The naive case. Wrapping this plan with SubqueryScan anyway
* is demanded as the underlying plan might be SharedInputScan where
* the target list should not be modified, and in order to align
* the semantics with de-duplicate case.
*/
result_plan = wrap_plan_index(root,
result_plan,
root->parse,
&wag_context->current_pathkeys,
1,
"nondeduplicate",
&subquery);
root->parse->targetList = (List *)
cdbpullup_expr((Expr *) root->parse->targetList, sub_tlist, NIL, 1);
root->parse->havingQual = (Node *)
cdbpullup_expr((Expr *) root->parse->havingQual, sub_tlist, NIL, 1);
root->parse->scatterClause = (List *)
cdbpullup_expr((Expr *) root->parse->scatterClause, sub_tlist, NIL, 1);
root->parse->rtable = subquery->rtable;
/*
* We make zero as the peer count if tv is NULL. The inner
* should count up how many non-NULL there is.
*
* pc = CASE WHEN tv IS NOT NULL THEN 1 ELSE 0 END
*/
tv_expr = (Expr *)
get_sortgroupclause_expr(linitial(sortClause), result_plan->targetlist);
nt = makeNode(NullTest);
nt->arg = tv_expr;
nt->nulltesttype = IS_NOT_NULL;
casearg = makeNode(CaseWhen);
casearg->expr = (Expr *) nt;
casearg->result = (Expr *)
makeConst(INT8OID, -1, 8, Int64GetDatum(1), false, true);
pc_expr = makeNode(CaseExpr);
pc_expr->casetype = INT8OID;
pc_expr->arg = NULL;
pc_expr->args = list_make1(casearg);
pc_expr->defresult = (Expr *)
makeConst(INT8OID, -1, 8, Int64GetDatum(0), false, true);
pc_tle = makeTargetEntry((Expr *) pc_expr,
list_length(result_plan->targetlist) + 1,
pstrdup("peer_count"),
false);
wag_context->pc_pos = pc_tle->resno;
result_plan->targetlist = lappend(result_plan->targetlist, pc_tle);
}
/*
* result_plan is SubqueryScan here whichever we took. Update locus
* in order to represent this subqeury.
*/
Assert(IsA(result_plan, SubqueryScan));
return result_plan;
}
/*
* within_agg_add_outer_sort
*
* Adds Sort to the outer plan, with Motion if desired. This is very
* straightforward and does not contain within-aggregate specific stuff.
* current_pathkeys in wag_context may be updated.
*/
static Plan *
within_agg_add_outer_sort(PlannerInfo *root,
WithinAggContext *wag_context,
List *sortClause,
Plan *outer_plan)
{
List *sort_pathkeys;
Query *outer_parse;
const Index Outer = 1;
if (!root->parse->groupClause)
{
/*
* Plain aggregate case. Gather tuples into QD.
*/
sort_pathkeys =
make_pathkeys_for_sortclauses(sortClause, outer_plan->targetlist);
sort_pathkeys = canonicalize_pathkeys(root, sort_pathkeys);
/*
* Check the sort redundancy.
*/
if (!pathkeys_contained_in(sort_pathkeys, wag_context->current_pathkeys))
{
outer_plan = (Plan *)
make_sort_from_sortclauses(root, sortClause, outer_plan);
mark_sort_locus(outer_plan);
wag_context->current_pathkeys = sort_pathkeys;
}
if (outer_plan->flow->flotype != FLOW_SINGLETON)
{
outer_plan = (Plan *) make_motion_gather_to_QE(outer_plan, true);
outer_plan->total_cost +=
incremental_motion_cost(outer_plan->plan_rows,
outer_plan->plan_rows * root->config->cdbpath_segments);
}
}
else
{
CdbPathLocus current_locus;
List *groupSortClauses;
Assert(root->group_pathkeys);
/*
* Create current locus.
*/
current_locus = cdbpathlocus_from_flow(outer_plan->flow);
if (CdbPathLocus_IsPartitioned(current_locus) &&
outer_plan->flow->hashExpr)
current_locus = cdbpathlocus_from_exprs(root, outer_plan->flow->hashExpr);
/*
* Add a redistribute motion if the group key doesn't collocate.
* group_pathkeys should have been fixed to reflect the latest targetlist.
* best_path->locus is wrong here since we put SubqueryScan already.
*/
if (!cdbpathlocus_collocates(current_locus, root->group_pathkeys, false /*exact_match*/))
{
List *groupExprs;
groupExprs = get_sortgrouplist_exprs(root->parse->groupClause,
outer_plan->targetlist);
outer_plan = (Plan *)
make_motion_hash(root, outer_plan, groupExprs);
outer_plan->total_cost +=
incremental_motion_cost(outer_plan->plan_rows,
outer_plan->plan_rows);
/*
* Invalidate pathkeys; the result is not sorted any more.
*/
wag_context->current_pathkeys = NULL;
}
/*
* Now we can add sort node.
*/
groupSortClauses = list_concat(copyObject(root->parse->groupClause),
sortClause);
sort_pathkeys =
make_pathkeys_for_sortclauses(groupSortClauses, outer_plan->targetlist);
sort_pathkeys = canonicalize_pathkeys(root, sort_pathkeys);
if (!pathkeys_contained_in(sort_pathkeys, wag_context->current_pathkeys))
{
outer_plan = (Plan *) make_sort_from_sortclauses(root,
groupSortClauses,
outer_plan);
mark_sort_locus(outer_plan);
/*
* Update current pathkeys.
*/
wag_context->current_pathkeys = sort_pathkeys;
}
}
/*
* Prepare parse tree for wrapping it by SubqueryScan.
*/
outer_parse = copyObject(root->parse);
root->parse->targetList = (List *)
cdbpullup_expr((Expr *) root->parse->targetList, outer_plan->targetlist, NIL, 1);
root->parse->havingQual = (Node *)
cdbpullup_expr((Expr *) root->parse->havingQual, outer_plan->targetlist, NIL, 1);
root->parse->scatterClause = (List *)
cdbpullup_expr((Expr *) root->parse->scatterClause, outer_plan->targetlist, NIL, 1);
/*
* Wrap plan by subquery as the outer of upcoming join.
*/
outer_plan = wrap_plan_index(root,
outer_plan,
outer_parse,
&wag_context->current_pathkeys,
Outer,
"outer_plan",
&outer_parse);
Assert(list_length(wag_context->rtable) == 0);
wag_context->rtable = list_concat(wag_context->rtable, outer_parse->rtable);
return outer_plan;
}
/*
* within_agg_construct_inner
*
* Constructs the inner plan which calculates the total non-NULL row count
* of the input. The aggregate function is actually not count(), but sum()
* of the peer count which has considered NULL input in both of naive and
* de-duplicate cases. Updates wag_context->tc_pos and the output plan
* should be SubqueryScan ready for the join.
*/
static Plan *
within_agg_construct_inner(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
Plan *inner_plan)
{
ListCell *l;
int idx;
int numGroupCols;
Path input_path;
TargetEntry *pc_tle;
Expr *tc_expr;
GroupContext ctx;
List *tlist;
double numGroups = *group_context->p_dNumGroups;
bool use_hashed_grouping;
bool querynode_changed = false;
List *pathkeys = NIL;
AggClauseCounts agg_counts;
AttrNumber *grpColIdx;
Query *original_parse;
List *original_group_pathkeys;
Query *parse;
const Index Inner = 2;
grpColIdx = get_grouplist_colidx(root->parse->groupClause,
inner_plan->targetlist, &numGroupCols);
/* build grouping key columns */
tlist = NIL;
foreach_with_count (l, root->parse->groupClause, idx)
{
GroupClause *gc = (GroupClause *) lfirst(l);
TargetEntry *tle, *newtle;
tle = get_sortgroupclause_tle(gc, inner_plan->targetlist);
newtle = flatCopyTargetEntry(tle);
newtle->resno = (AttrNumber) idx + 1;
tlist = lappend(tlist, newtle);
}
/*
* Sum up the peer count to count the total number of rows per group.
*/
pc_tle = get_tle_by_resno(inner_plan->targetlist, wag_context->pc_pos);
tc_expr = (Expr *) makeAggrefByOid(AGGFNOID_SUM_BIGINT,
list_make1(pc_tle->expr));
tc_expr = (Expr *) coerce_to_bigint(NULL, (Node *) tc_expr, "sum_to_bigint");
wag_context->tc_pos = list_length(tlist) + 1;
tlist = lappend(tlist, makeTargetEntry((Expr *) tc_expr,
wag_context->tc_pos,
"total_count",
false));
/*
* best_path is not appropriate here after building some SubqueryScan.
* Build up a dummy Path to reflect the underlying plan, but
* needed information is only locus in cdb_grouping_planner.
*/
memcpy(&input_path, group_context->best_path, sizeof(Path));
/*
* Create locus back from flow. Unfortunately cdbpathlocus_from_flow()
* doesn't return hashed locus in repartitioned case, so we need to
* call from_exprs() again if it's available.
*/
input_path.locus = cdbpathlocus_from_flow(inner_plan->flow);
if (CdbPathLocus_IsPartitioned(input_path.locus) &&
inner_plan->flow->hashExpr)
input_path.locus = cdbpathlocus_from_exprs(root, inner_plan->flow->hashExpr);
input_path.pathkeys = wag_context->inner_pathkeys;
MemSet(&agg_counts, 0, sizeof(AggClauseCounts));
count_agg_clauses((Node *) tlist, &agg_counts);
/*
* Evaluate possibility for hash/sort strategy. Things have been changed
* since the last decision in grouping_planner(), as the base plan
* may now be sorted.
*/
use_hashed_grouping = choose_hashed_grouping(root,
group_context->tuple_fraction,
&input_path,
&input_path,
numGroups,
&agg_counts);
ctx.best_path = &input_path;
ctx.cheapest_path = &input_path;
ctx.subplan = inner_plan;
ctx.sub_tlist = NIL;
ctx.tlist = tlist;
ctx.use_hashed_grouping = use_hashed_grouping;
ctx.tuple_fraction = 0.1;
ctx.canonical_grpsets = NULL;
ctx.grouping = 0;
ctx.numGroupCols = numGroupCols;
ctx.groupColIdx = grpColIdx;
ctx.numDistinctCols = 0;
ctx.distinctColIdx = NULL;
ctx.p_dNumGroups = &numGroups;
ctx.pcurrent_pathkeys = &pathkeys;
ctx.querynode_changed = &querynode_changed;
/*
* Save the parse tree, for cdb_grouping_planner will modify it.
*/
original_parse = root->parse;
original_group_pathkeys = root->group_pathkeys;
root->parse = parse = copyObject(root->parse);
root->parse->targetList = tlist;
root->parse->havingQual = NULL;
root->parse->scatterClause = NIL;
root->group_pathkeys =
make_pathkeys_for_sortclauses(parse->groupClause, tlist);
/*
* Make a multi-phase or simple aggregate plan.
*/
inner_plan = make_parallel_or_sequential_agg(root,
&agg_counts,
&ctx,
&wag_context->inner_pathkeys);
/*
* Wrap plan by subquery as the inner of upcoming join.
*/
inner_plan = wrap_plan_index(root,
inner_plan,
parse,
&wag_context->inner_pathkeys,
Inner,
"inner_plan",
&parse);
wag_context->rtable = list_concat(wag_context->rtable, parse->rtable);
/* outer + inner = 2 */
Assert(list_length(wag_context->rtable) == 2);
/*
* Restore the original info. Note that group_pathkeys is updated
* in wrap_plan_index(), so don't move this before it.
*/
root->parse = original_parse;
root->group_pathkeys = original_group_pathkeys;
return inner_plan;
}
/*
* within_agg_join_plans
*
* Joins the outer and inner plans and filters out unrelated rows. We use MergeJoin
* as the outer is already sorted and want to keep the order to the aggregate.
* It requires the inner plan to be able to be re-scannable, so we add sort
* or material node to the inner plan. The join should produce exact same number
* of the outer plan as the join clause is in GROUP BY keys and the inner is
* grouped with them.
*/
static Plan *
within_agg_join_plans(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
Plan *outer_plan,
Plan *inner_plan)
{
Plan *result_plan;
ListCell *l;
int idx;
List *join_tlist;
List *join_clause;
const Index Outer = 1, Inner = 2;
List *extravars;
Var *pc_var, *tc_var;
/*
* Up to now, these should've been prepared.
*/
Assert(wag_context->pc_pos > 0);
Assert(wag_context->tc_pos > 0);
/*
* Build target list for grouping columns.
*/
join_clause = NIL;
foreach_with_count (l, root->parse->groupClause, idx)
{
GroupClause *gc = (GroupClause*) lfirst(l);
TargetEntry *tle;
Var *outer_var, *inner_var;
Expr *qual;
/*
* Construct outer group keys.
*/
tle = get_sortgroupclause_tle(gc, outer_plan->targetlist);
Assert(tle && IsA(tle->expr, Var));
outer_var = makeVar(Outer, tle->resno,
exprType((Node *) tle->expr),
exprTypmod((Node *) tle->expr), 0);
/*
* Construct inner group keys.
*/
tle = get_tle_by_resno(inner_plan->targetlist, idx + 1);
Assert(tle && IsA(tle->expr, Var));
inner_var = makeVar(Inner, tle->resno,
exprType((Node *) tle->expr),
exprTypmod((Node *) tle->expr), 0);
/*
* Make join clause for group keys.
*/
qual = make_op(NULL, list_make1(makeString("=")),
(Node *) outer_var, (Node *) inner_var, -1);
qual->type = T_DistinctExpr;
qual = make_notclause(qual);
join_clause = lappend(join_clause, qual);
}
/*
* This is similar to make_subplanTargetList(), but things are much simpler.
* Note that this makes sure that expressions like SRF are going to be
* in the upper aggregate target list rather than in this join target list.
*/
join_tlist = flatten_tlist(root->parse->targetList);
extravars = pull_var_clause(root->parse->havingQual, false);
join_tlist = add_to_flat_tlist(join_tlist, extravars, false);
foreach (l, root->parse->groupClause)
{
GroupClause *gc = lfirst(l);
TargetEntry *gc_tle, *join_tle;
/*
* We need the grouping expressions in the target list. If they are
* in the taget list already, we remember the grouping reference
* since exracting vars drop those information. Otherwise, we
* simply append the entry to the target list.
*/
gc_tle = get_sortgroupclause_tle(gc, root->parse->targetList);
join_tle = tlist_member((Node *) gc_tle->expr, join_tlist);
if (join_tle)
{
join_tle->ressortgroupref = gc->tleSortGroupRef;
join_tle->resname = gc_tle->resname;
}
else
{
join_tle = flatCopyTargetEntry(gc_tle);
join_tlist = lappend(join_tlist, join_tle);
}
}
/*
* Make sure that the peer count and the total count is in the
* target list of the join. They will be needed in the upper
* final aggregate by the percentile functions.
*/
pc_var = makeVar(Outer, wag_context->pc_pos, INT8OID, -1, 0);
tc_var = makeVar(Inner, wag_context->tc_pos, INT8OID, -1, 0);
join_tlist = add_to_flat_tlist(join_tlist,
list_make2((void *) pc_var, (void *) tc_var),
false);
/*
* It is ideal to tell if the inner plan is fine to merge-join by
* examining it as re-scannable plan, but it seems we don't have
* such infrastructure, so here we assume the inner plan is not
* re-scannable and not sorted. If it is a grouping query,
* we add sort node, otherwise just put a materialize node.
*/
if (root->parse->groupClause && !wag_context->inner_pathkeys)
{
AttrNumber *grpColIdx;
grpColIdx = get_grouplist_colidx(root->parse->groupClause,
inner_plan->targetlist, NULL);
inner_plan = (Plan *)
make_sort_from_groupcols(root,
root->parse->groupClause,
grpColIdx,
false,
inner_plan);
mark_sort_locus(inner_plan);
}
else
{
inner_plan = (Plan *) make_material(inner_plan);
mark_passthru_locus(inner_plan, true, true);
}
/*
* All set. Join two plans.
* We choose cartesian product if there is no join clauses, meaning
* no grouping happens.
*/
if (list_length(join_clause) > 0)
result_plan = (Plan *) make_mergejoin(join_tlist,
NIL,
NIL,
join_clause,
outer_plan,
inner_plan,
JOIN_INNER);
else
result_plan = (Plan *) make_nestloop(join_tlist,
NIL,
NIL,
outer_plan,
inner_plan,
JOIN_INNER);
result_plan->startup_cost = outer_plan->startup_cost + inner_plan->startup_cost;
result_plan->plan_rows = outer_plan->plan_rows;
result_plan->total_cost = outer_plan->total_cost + inner_plan->total_cost;
result_plan->total_cost += cpu_tuple_cost * result_plan->plan_rows;
mark_passthru_locus(result_plan, true, true);
return result_plan;
}
/*
* within_agg_final_agg
*
* The final stage of plan_within_agg_persort(). The input plans are already
* joined and the total count from the inner plan is available now. The
* peer count and the total count is attached to the PercentileExpr so that
* it will get those values in the percentile function in the executor.
*/
static Plan *
within_agg_final_agg(PlannerInfo *root,
GroupContext *group_context,
WithinAggContext *wag_context,
List *sortClause,
Plan *result_plan)
{
ListCell *l;
List *percentiles;
Var *pc_var, *tc_var;
AttrNumber *grpColIdx;
int numGroupCols;
AggClauseCounts agg_counts;
AggStrategy aggstrategy;
const Index Outer = 1, Inner = 2;
/*
* Sanity check. These should've been prepared up to now.
*/
Assert(wag_context->pc_pos > 0);
Assert(wag_context->tc_pos > 0);
/*
* Attach the peer count and the total count expressions to
* PercentileExpr, which will be needed in the executor.
*/
percentiles = fetch_percentiles(root->parse, sortClause);
pc_var = makeVar(Outer, wag_context->pc_pos, INT8OID, -1, 0);
tc_var = makeVar(Inner, wag_context->tc_pos, INT8OID, -1, 0);
foreach (l, percentiles)
{
PercentileExpr *perc = lfirst(l);
perc->pcExpr = (Expr *) pc_var;
perc->tcExpr = (Expr *) tc_var;
}
MemSet(&agg_counts, 0, sizeof(AggClauseCounts));
count_agg_clauses((Node *) root->parse->targetList, &agg_counts);
/*
* Prepare GROUP BY clause for the final aggregate.
* Make sure the column indices point to the topmost target list.
*/
grpColIdx = get_grouplist_colidx(root->parse->groupClause,
result_plan->targetlist, &numGroupCols);
aggstrategy = root->parse->groupClause ? AGG_SORTED : AGG_PLAIN;
result_plan = (Plan *) make_agg(root,
root->parse->targetList,
(List *) root->parse->havingQual,
aggstrategy,
false,
numGroupCols,
grpColIdx,
*group_context->p_dNumGroups,
0, /* num_nullcols */
0, /* input_grouping */
0, /* grouping */
0, /* rollup_gs_times */
1, /* numAggs */
agg_counts.transitionSpace,
result_plan);
/*
* Stop copying sorts in flow, for the targetlist doesn't have them anymore.
*/
mark_passthru_locus(result_plan, true, false);
/*
* current_pathkeys is not needed anymore, but just in case.
*/
wag_context->current_pathkeys = NIL;
return result_plan;
}
/*
* plan_within_agg_persort
* Make a series of plans to calculate percentile per identical sort clause.
*
* The main stream of this planner is:
* - Build the base plan
* - Sort the input by group and target value
* - Split the input into two
* - One for the outer
* - The other for the inner, aggregate to count actual row number
* - Join two trees to get the inner row count to the sorted outer together
* - Aggregate the result of join to perform the final aggregate
*
* The first step is to figure out if we could optimize the input by
* de-duplicate table. If so, we create a sub-plan with aggregate to
* reduce the rows into GROUP + ORDER unit and add the peer count in this unit.
* If we don't take de-duplicate plan, we append constant 1 as the peer count
* to the target list to match the semantics with de-duplicate plan. In any
* case, the input plan is wrapped by a SubqueryScan in order not to confuse
* upper plan. The peer count takes account into NULL, so that the percentile
* function can operate on only non-NULL target value.
*
* The next step is for the outer plan; sort the input tuples by group clause
* and the target value. During this process, the input tuples are distributed
* by the group clause. We need to keep this order until the final aggregate
* so that the final percentile function can tell ordering of the input.
*
* Then we look at the inner side. It basically calculates number of rows
* per group. Since we know the peer count per each distinct target value in
* a group in either of naive or deduplicate approach, we actually take sum()
* rather than count() to calculate the total number of non-NULL target values.
*
* Now, we can join two plans. The inner total count is going to be with
* the target value for each group here. Note that we use Merge Join to
* keep the outer order and the inner should have capability to be rescanned
* per its requirement.
*
* Finally, the rows are aggregated per group. Since PercentileExpr needs
* to know the peer count and the total count in conjunction with its
* original argument and the target value (as the ORDER BY clause,) and we now
* know where those values are, here we attach those expressions to
* PercentileExpr nodes. These additional expressions are similar to
* AggOrder in Aggref, which will be combined with the original argument
* expression in the executor to be evaluated.
*
* The final result of this function is a plan ending Agg node with target
* list including each percentile result. Note this function should not
* change the semantics of input target list as an encapsulation.
*/
static Plan *
plan_within_agg_persort(PlannerInfo *root,
GroupContext *group_context,
List *sortClause,
List *current_pathkeys,
Plan *result_plan)
{
WithinAggContext wag_context;
Plan *outer_plan, *inner_plan;
List *partners;
ListCell *l;
memset(&wag_context, 0, sizeof(WithinAggContext));
wag_context.current_pathkeys = current_pathkeys;
/*
* Group clause expressions should be in ascending order,
* because our MergeJoin is not able to handle descending-ordered
* child plans. It is desirable to improve MergeJoin, but it requires
* amount of work.
*/
foreach (l, root->parse->groupClause)
{
GroupClause *gc = lfirst(l);
Node *gcexpr;
Oid gctype;
/*
* We assume only flattened grouping expressions here.
*/
Assert(IsA(gc, GroupClause));
gcexpr = get_sortgroupclause_expr(gc, group_context->sub_tlist);
gctype = exprType(gcexpr);
gc->sortop = ordering_oper_opid(gctype);
}
/*
* Make a common plan shared by outer and inner plan. It may become
* a de-duplicate plan.
*/
result_plan = within_agg_make_baseplan(root,
group_context,
&wag_context,
sortClause,
result_plan);
Assert(wag_context.pc_pos > 0);
Assert(IsA(result_plan, SubqueryScan));
/*
* Split the tree into outer and inner which will be joined later.
* It comes before Sort, so that the both of outer and inner run
* in parallel. We observed in most cases splitting it here requires
* RedistributeMotion in both sides, which allows more parallel way.
*/
partners = share_plan(root, result_plan, 2);
outer_plan = list_nth(partners, 0);
inner_plan = list_nth(partners, 1);
/*
* Keep the pathkeys for the inner as pre-Subquery one.
*/
wag_context.inner_pathkeys = wag_context.current_pathkeys;
/*
* Add sort if necessary on top of outer plan of join.
*/
outer_plan = within_agg_add_outer_sort(root, &wag_context, sortClause, outer_plan);
Assert(IsA(outer_plan, SubqueryScan));
Assert(list_length(wag_context.rtable) == 1);
/*
* Construct inner plan of join that returns only number of rows.
* The inner side always create target list looking like
* G1, G2, ..., count(*) TP
*/
inner_plan = within_agg_construct_inner(root,
group_context,
&wag_context,
inner_plan);
Assert(IsA(inner_plan, SubqueryScan));
Assert(wag_context.tc_pos > 0);
Assert(list_length(wag_context.rtable) == 2);
/*
* merge join to find interesting one or two rows per group per
* percentile.
*/
result_plan = within_agg_join_plans(root,
group_context,
&wag_context,
outer_plan,
inner_plan);
#ifdef NOT_USED
if (true)
{
/*
* For debug purpose.
* This helps to see what's the intermediate result.
*/
root->parse->targetList = copyObject(result_plan->targetlist);
root->parse->rtable = wag_context.rtable;
return result_plan;
}
#endif
/*
* Finally put Agg node to calculate exact value with filtered few rows.
*/
result_plan = within_agg_final_agg(root,
group_context,
&wag_context,
sortClause,
result_plan);
root->parse->targetList = copyObject(result_plan->targetlist);
root->parse->rtable = wag_context.rtable;
rebuild_simple_rel_and_rte(root);
return result_plan;
}
/*
* within_agg_planner
*
* This is the entry point of within-aggregate planner.
* Makes a specialized plan for within aggregates. If we find the query
* has normal aggregates among the within-aggregates, or within-aggregates
* have different sortClause, we generate multiple sub-plans regarding with
* the sortClause and combine the results by joining those sub-plan trees.
* For the normal aggregates, we call cdb_grouping_planner() to plan the tree,
* and treat it as one of the split sub-plans.
*/
Plan *
within_agg_planner(PlannerInfo *root,
AggClauseCounts *agg_counts,
GroupContext *group_context)
{
List *aggnodes, *percnodes;
ListCell *l;
List **aggreflist;
List **sortlist;
int numsortlist;
int numGroupCols, numDistinctCols;
AttrNumber *grpColIdx, *distinctColIdx;
int i;
List *sub_tlist;
AttrNumber next_resno;
List *current_pathkeys;
Plan *result_plan;
aggnodes = extract_nodes(NULL, (Node *) root->parse->targetList, T_Aggref);
aggnodes = list_concat(aggnodes,
extract_nodes(NULL, (Node *) root->parse->havingQual, T_Aggref));
aggnodes = list_concat(aggnodes,
extract_nodes(NULL, (Node *) root->parse->scatterClause, T_Aggref));
percnodes = extract_nodes(NULL, (Node *) root->parse->targetList, T_PercentileExpr);
percnodes = list_concat(percnodes,
extract_nodes(NULL, (Node *) root->parse->havingQual, T_PercentileExpr));
percnodes = list_concat(percnodes,
extract_nodes(NULL, (Node *) root->parse->scatterClause, T_PercentileExpr));
/* Allocate maximum number of list */
numsortlist = list_length(percnodes) + 1;
/* initialize each element with NIL */
aggreflist = (List **) palloc0(sizeof(List *) * numsortlist);
sortlist = (List **) palloc0(sizeof(List *) * numsortlist);
numsortlist = 0; /* Use this as a counter */
sub_tlist = group_context->sub_tlist;
next_resno = list_length(sub_tlist) + 1;
/*
* WITHIN aggregates are not supported in the grouping extensions.
* However, parse->groupClause may have non-flattened GroupClause list.
* We simply flatten it by reconstruct_group_clause under the assumption
* that we have denied grouping extension cases.
*/
Assert(!is_grouping_extension(group_context->canonical_grpsets));
grpColIdx = get_grouplist_colidx(
root->parse->groupClause, sub_tlist, &numGroupCols);
root->parse->groupClause =
reconstruct_group_clause(root->parse->groupClause,
sub_tlist,
grpColIdx,
numGroupCols);
numDistinctCols = agg_counts->numDistinctAggs;
distinctColIdx = (AttrNumber *) palloc(sizeof(AttrNumber) * numDistinctCols);
numDistinctCols = 0; /* Use this as a counter */
/*
* Collect aggref nodes to process them separately from percentiles.
* Note we represent this special case by NIL for sortClause (sortlist.)
*/
if (aggnodes)
{
sortlist[numsortlist] = NIL;
foreach (l, aggnodes)
{
Aggref *aggref = lfirst(l);
if (aggref->aggdistinct)
{
Node *arg;
TargetEntry *sub_tle;
Assert(list_length(aggref->args) == 1);
arg = linitial(aggref->args);
sub_tle = tlist_member(arg, sub_tlist);
if (!sub_tle)
{
sub_tle = makeTargetEntry((Expr *) arg,
next_resno++,
"<expr>",
true);
sub_tlist = lappend(sub_tlist, sub_tle);
}
distinctColIdx[numDistinctCols++] = sub_tle->resno;
}
aggreflist[numsortlist] = lappend(aggreflist[numsortlist], aggref);
}
numsortlist++;
}
/*
* Collect percentile nodes and classify them into some groups by sortClause.
* During this process, if sub_tlist lacks target entry for the
* sortClause, it is added.
*/
foreach (l, percnodes)
{
PercentileExpr *perc = lfirst(l);
List *sortClause;
ListCell *sl;
sortClause = perc->sortClause;
Assert(sortClause);
/*
* We need to add tlist to the parse's tlist. This is
* basically parser's task, but the list is separated
* to keep away from ungroup'ed columns check and
* this is the right place to add back to the parser's tlist.
*/
foreach (sl, sortClause)
{
SortClause *sc = lfirst(sl);
TargetEntry *tle, *sub_tle;
tle = get_sortgroupclause_tle(sc, perc->sortTargets);
sub_tle = tlist_member((Node *) tle->expr, sub_tlist);
if (!sub_tle)
{
sub_tle = makeTargetEntry(tle->expr,
next_resno++,
tle->resname,
tle->resjunk);
sub_tlist = lappend(sub_tlist, sub_tle);
}
sc->tleSortGroupRef = assignSortGroupRef(sub_tle, sub_tlist);
tle->ressortgroupref = sc->tleSortGroupRef;
}
/* Find identical sortClause. */
for (i = 0; i < numsortlist && !equal(sortClause, sortlist[i]); i++);
if (i == numsortlist)
{
/* Not found, so add a new one. */
sortlist[numsortlist] = sortClause;
numsortlist++;
}
/*
* Add this percentile to the group associated with the sortClause.
*/
aggreflist[i] = lappend(aggreflist[i], perc);
}
/* Make sure group_context->sub_tlist has pointer to what we made */
group_context->sub_tlist = sub_tlist;
/*
* Make the scaffold. We always take best_path here because
* it is not clear which to use for upcoming complex plans.
*/
Assert(sub_tlist != NIL);
result_plan = create_plan(root, group_context->best_path);
result_plan = plan_pushdown_tlist(result_plan, sub_tlist);
Assert(result_plan->flow);
current_pathkeys = group_context->best_path->pathkeys;
/*
* numsortlist is actually the number of sub-plans.
*/
Insist(numsortlist > 0);
/*
* The approach is very close to the one for DQA. If the plan consists
* only one sort group, then construct straightforward plan without
* mangling target list. If more than one, including normal aggregate,
* we split each sort group (normal aggregate goes to NIL sort group) into
* sub pieces and construct separate plans, and join them to get the final
* plan to match the desired target list.
*/
if (numsortlist == 1)
{
/*
* Simply plan a tree and return it. We don't clutter the target list.
*/
result_plan = plan_within_agg_persort(root,
group_context,
*sortlist,
current_pathkeys,
result_plan);
}
else
{
/*
* The pattern on multi-level sort is similar to multi dqa.
* We use its infrastructure much to avoid reinventing wheel.
*/
List *base_plans;
MppGroupContext mgctx;
List *rtable;
base_plans = share_plan(root, result_plan, numsortlist);
Assert(numsortlist > 0);
MemSet(&mgctx, 0, sizeof(MppGroupContext));
/*
* XXX: Need support of DqaJoinMerge for types used in GROUP BY clause
* which are not mergejoinable.
*/
mgctx.join_strategy = numGroupCols > 0 ? DqaJoinHash : DqaJoinCross;
mgctx.numGroupCols = numGroupCols;
/*
* This code is from deconstruct_agg_info. What it does is to
* collect grouping keys and make a simple list which contain
* only those key expressions, which will be used in each individual
* plan tree as leading columns (and later JOIN clause).
*/
mgctx.grps_tlist = NIL;
for (i = 0; i < numGroupCols; i++)
{
TargetEntry *sub_tle, *prelim_tle;
sub_tle = get_tle_by_resno(sub_tlist, grpColIdx[i]);
prelim_tle = flatCopyTargetEntry(sub_tle);
prelim_tle->resno = list_length(mgctx.grps_tlist) + 1;
mgctx.grps_tlist = lappend(mgctx.grps_tlist, prelim_tle);
}
mgctx.dref_tlists = (List **) palloc0(numsortlist * sizeof(List *));
/*
* Within-aggregate special. Used in within_aggregate_expr().
* Each sub-plan tree is identified by the sort clause.
*/
mgctx.wagSortClauses = NIL;
for (i = 0; i < numsortlist; i++)
mgctx.wagSortClauses = lappend(mgctx.wagSortClauses, sortlist[i]);
/*
* Prepare the final tlist to restore the original list. The main work
* goes into deconstruct_within_agg(), which determins which sub-plan tree
* this expression is actually coming from, and store that information in mgctx.
*/
foreach (l, root->parse->targetList)
{
TargetEntry *tle, *final_tle;
tle = (TargetEntry *) lfirst(l);
final_tle = flatCopyTargetEntry(tle);
final_tle->expr =
(Expr *) deconstruct_within_agg((Node *) tle->expr, &mgctx);
mgctx.fin_tlist = lappend(mgctx.fin_tlist, final_tle);
}
/*
* HAVING clause, same as target list. We wish we could optimize this
* as pushing each expression down to the individual plan tree, but
* we don't do it and just follow the same notion of DQA for now.
*/
foreach (l, (List *) root->parse->havingQual)
{
Expr *qual, *fin_hqual;
qual = lfirst(l);
fin_hqual = (Expr *) deconstruct_within_agg((Node *) qual, &mgctx);
mgctx.fin_hqual = lappend(mgctx.fin_hqual, fin_hqual);
}
/*
* Offsets. Used in join_dqa_coplan
*/
mgctx.dqa_offsets = (int *) palloc(sizeof(int) * (numsortlist + 1));
mgctx.dqa_offsets[0] = numGroupCols;
for (i = 0; i < numsortlist; i++)
{
mgctx.dqa_offsets[i + 1] =
mgctx.dqa_offsets[i] + list_length(mgctx.dref_tlists[i]);
}
/*
* Now plan each tree. Store them to array and later join them.
* Don't forget to save rtable representing each subquery.
*/
rtable = NIL;
mgctx.dqaArgs = (DqaInfo *) palloc(numsortlist * sizeof(DqaInfo));
/* keep the original query */
for (i = 0; i < numsortlist; i++)
{
Plan *coplan;
Query *coquery;
char queryname[256];
PlannerInfo root_copy;
size_t sz;
/*
* The base plan is created by best_path.
*/
current_pathkeys = group_context->best_path->pathkeys;
/*
* We use different instance of PlannerInfo for each cycle
* especially cdb_grouping_planner frees simple_rel_array.
* See also plan_append_aggs_with_rewrite.
*/
memcpy(&root_copy, root, sizeof(PlannerInfo));
sz = root->simple_rel_array_size * sizeof(RelOptInfo *);
root_copy.simple_rel_array =
(RelOptInfo **) palloc(sz);
memcpy(root_copy.simple_rel_array, root->simple_rel_array, sz);
/*
* Query should be copied deeply for the planner changes it.
*/
coquery = copyObject(root->parse);
coquery->targetList = seq_tlist_concat(copyObject(mgctx.grps_tlist), mgctx.dref_tlists[i]);
/*
* Clear havingQual and scatterClause, since they will be handled only
* the top of joins, and never in individual aggregate.
*/
coquery->havingQual = NULL;
coquery->scatterClause = NIL;
root_copy.parse = coquery;
if (sortlist[i])
{
/*
* Run the specialized planner for aggregate expressions
* gathered by an identical ORDER BY clause.
*/
coplan = plan_within_agg_persort(&root_copy,
group_context,
sortlist[i],
current_pathkeys,
list_nth(base_plans, i));
}
else
{
/*
* Run normal grouping planner for normal aggs.
*/
GroupContext local_group_context;
memcpy(&local_group_context, group_context, sizeof(GroupContext));
local_group_context.subplan = list_nth(base_plans, i);
local_group_context.tlist = coquery->targetList;
/* These fields are not set in grouping_planner */
local_group_context.numGroupCols = numGroupCols;
local_group_context.groupColIdx = grpColIdx;
local_group_context.numDistinctCols = numDistinctCols;
local_group_context.distinctColIdx = distinctColIdx;
/*
* Make a multi-phase or simple aggregate plan.
*
* agg_counts contain only normal aggregate information
* (without within-aggs information), so it's safe to use it
* as it's passed here.
*/
coplan = make_parallel_or_sequential_agg(&root_copy,
agg_counts,
&local_group_context,
&current_pathkeys);
}
snprintf(queryname, sizeof(queryname), "wag_coplan_%d", i + 1);
mgctx.dqaArgs[i].coplan =
wrap_plan_index(&root_copy, coplan, coquery,
NULL, i + 1, queryname,
&coquery);
rtable = list_concat(rtable, coquery->rtable);
}
/* Begin with the first coplan, then join in each succeeding coplan. */
result_plan = mgctx.dqaArgs[0].coplan;
for (i = 1; i < numsortlist; i++)
result_plan = join_dqa_coplan(root, &mgctx, result_plan, i);
/* Prepare for finalize_split_expr. */
mgctx.top_tlist = result_plan->targetlist;
/* Now to set true. */
mgctx.use_dqa_pruning = true;
result_plan->targetlist = (List *) finalize_split_expr((Node *) mgctx.fin_tlist, &mgctx);
result_plan->qual = (List *) finalize_split_expr((Node *) mgctx.fin_hqual, &mgctx);
UpdateScatterClause(root->parse, result_plan->targetlist);
/*
* Reconstruct the flow since the targetlist for the result_plan may have
* changed.
*/
result_plan->flow = pull_up_Flow(result_plan,
result_plan->lefttree,
true);
/* Need to adjust root->parse for upper plan. */
root->parse->rtable = rtable;
root->parse->targetList = copyObject(result_plan->targetlist);
rebuild_simple_rel_and_rte(root);
}
/* Tell the upper planner that we changed the node */
*(group_context->pcurrent_pathkeys) = NIL;
*(group_context->querynode_changed) = true;
return result_plan;
}
Plan *add_motion_to_dqa_child(Plan *plan, PlannerInfo *root, bool *motion_added)
{
Plan *result = plan;
*motion_added = false;
List *pathkeys = make_pathkeys_for_groupclause(root->parse->groupClause, plan->targetlist);
CdbPathLocus locus = cdbpathlocus_from_flow(plan->flow);
if (CdbPathLocus_IsPartitioned(locus) && NIL != plan->flow->hashExpr)
{
locus = cdbpathlocus_from_exprs(root, plan->flow->hashExpr);
}
if (!cdbpathlocus_collocates(locus, pathkeys, true /*exact_match*/))
{
/* MPP-22413: join requires exact distribution match for collocation purposes,
* which may not be provided by the underlying group by, as computing the
* group by only requires relaxed distribution collocation
*/
List *groupExprs = get_sortgrouplist_exprs(root->parse->groupClause,
plan->targetlist);
result = (Plan *) make_motion_hash(root, plan, groupExprs);
result->total_cost += incremental_motion_cost(plan->plan_rows, plan->plan_rows);
*motion_added = true;
}
return result;
}