本节继续介绍聚合函数的实现,主要介绍了不使用Hash算法的情况下聚合函数的实现,在这种情况下会先排序后执行聚合,在ExecAgg节点执行前,已完成排序的操作.下面介绍在已完成排序的情况下聚合的实现,主要实现函数是ExecAgg->agg_retrieve_direct. 下面是不使用HashAggregate情况下GroupAggregate的计划树: ``` ",,,,,"select bh,avg(c1),min(c1),max(c2) from t_agg_simple group by bh;",,,"psql" 2019-05-16 12:04:45.621 CST,"xdb","testdb",1545,"[local]",5cdce11a.609,5,"SELECT",2019-05-16 12:03:38 CST,3/4,0,LOG,00000,"plan:"," {PLANNEDSTMT :commandType 1 :queryId 0 :hasReturning false :hasModifyingCTE false :canSetTag true :transientPlan false :dependsOnRole false :parallelModeNeeded false :jitFlags 0 :planTree {AGG :startup_cost 52.67 :total_cost 64.42 :plan_rows 200 :plan_width 98 :parallel_aware false :parallel_safe true :plan_node_id 0 :targetlist (... ) :qual <> :lefttree {SORT :startup_cost 52.67 :total_cost 54.52 :plan_rows 740 :plan_width 66 :parallel_aware false :parallel_safe true :plan_node_id 1 :targetlist (... ) :qual <> :lefttree {SEQSCAN :startup_cost 0.00 :total_cost 17.40 :plan_rows 740 :plan_width 66 :parallel_aware false :parallel_safe true :plan_node_id 2 :targetlist (... ) :qual <> :lefttree <> :righttree <> :initPlan <> :extParam (b) :allParam (b) :scanrelid 1 } :righttree <> :initPlan <> :extParam (b) :allParam (b) :numCols 1 :sortColIdx 1 :sortOperators 664 :collations 100 :nullsFirst false } :righttree <> :initPlan <> :extParam (b) :allParam (b) :aggstrategy 1 :aggsplit 0 :numCols 1 :grpColIdx 1 :grpOperators 98 :numGroups 200 :aggParams (b) :groupingSets <> :chain <> } :rtable (... ) :resultRelations <> :nonleafResultRelations <> :rootResultRelations <> :subplans <> :rewindPlanIDs (b) :rowMarks <> :relationOids (o 270375) :invalItems <> :paramExecTypes <> :utilityStmt <> :stmt_location 0 :stmt_len 63 } ``` 可以看到,在ExecAgg前会先执行ExecSort. ### 一、数据结构 **AggState** 聚合函数执行时状态结构体,内含AggStatePerAgg等结构体 ``` /* --------------------- * AggState information * * ss.ss_ScanTupleSlot refers to output of underlying plan. * ss.ss_ScanTupleSlot指的是基础计划的输出. * (ss = ScanState,ps = PlanState) * * Note: ss.ps.ps_ExprContext contains ecxt_aggvalues and * ecxt_aggnulls arrays, which hold the computed agg values for the current * input group during evaluation of an Agg node's output tuple(s). We * create a second ExprContext, tmpcontext, in which to evaluate input * expressions and run the aggregate transition functions. * 注意:ss.ps.ps_ExprContext包含了ecxt_aggvalues和ecxt_aggnulls数组, * 这两个数组保存了在计算agg节点的输出元组时当前输入组已计算的agg值. * --------------------- */ /* these structs are private in nodeAgg.c: */ //在nodeAgg.c中私有的结构体 typedef struct AggStatePerAggData *AggStatePerAgg; typedef struct AggStatePerTransData *AggStatePerTrans; typedef struct AggStatePerGroupData *AggStatePerGroup; typedef struct AggStatePerPhaseData *AggStatePerPhase; typedef struct AggStatePerHashData *AggStatePerHash; typedef struct AggState { //第一个字段是NodeTag(继承自ScanState) ScanState ss; /* its first field is NodeTag */ //targetlist和quals中所有的Aggref List *aggs; /* all Aggref nodes in targetlist & quals */ //链表的大小(可以为0) int numaggs; /* length of list (could be zero!) */ //pertrans条目大小 int numtrans; /* number of pertrans items */ //Agg策略模式 AggStrategy aggstrategy; /* strategy mode */ //agg-splitting模式,参见nodes.h AggSplit aggsplit; /* agg-splitting mode, see nodes.h */ //指向当前步骤数据的指针 AggStatePerPhase phase; /* pointer to current phase data */ //步骤数(包括0) int numphases; /* number of phases (including phase 0) */ //当前步骤 int current_phase; /* current phase number */ //per-Aggref信息 AggStatePerAgg peragg; /* per-Aggref information */ //per-Trans状态信息 AggStatePerTrans pertrans; /* per-Trans state information */ //长生命周期数据的ExprContexts(hashtable) ExprContext *hashcontext; /* econtexts for long-lived data (hashtable) */ ////长生命周期数据的ExprContexts(每一个GS使用) ExprContext **aggcontexts; /* econtexts for long-lived data (per GS) */ //输入表达式的ExprContext ExprContext *tmpcontext; /* econtext for input expressions */ #define FIELDNO_AGGSTATE_CURAGGCONTEXT 14 //当前活跃的aggcontext ExprContext *curaggcontext; /* currently active aggcontext */ //当前活跃的aggregate(如存在) AggStatePerAgg curperagg; /* currently active aggregate, if any */ #define FIELDNO_AGGSTATE_CURPERTRANS 16 //当前活跃的trans state AggStatePerTrans curpertrans; /* currently active trans state, if any */ //输入结束? bool input_done; /* indicates end of input */ //Agg扫描结束? bool agg_done; /* indicates completion of Agg scan */ //最后一个grouping set int projected_set; /* The last projected grouping set */ #define FIELDNO_AGGSTATE_CURRENT_SET 20 //将要解析的当前grouping set int current_set; /* The current grouping set being evaluated */ //当前投影操作的分组列 Bitmapset *grouped_cols; /* grouped cols in current projection */ //倒序的分组列链表 List *all_grouped_cols; /* list of all grouped cols in DESC order */ /* These fields are for grouping set phase data */ //-------- 下面的列用于grouping set步骤数据 //所有步骤中最大的sets大小 int maxsets; /* The max number of sets in any phase */ //所有步骤的数组 AggStatePerPhase phases; /* array of all phases */ //对于phases > 1,已排序的输入信息 Tuplesortstate *sort_in; /* sorted input to phases > 1 */ //对于下一个步骤,输入已拷贝 Tuplesortstate *sort_out; /* input is copied here for next phase */ //排序结果的slot TupleTableSlot *sort_slot; /* slot for sort results */ /* these fields are used in AGG_PLAIN and AGG_SORTED modes: */ //------- 下面的列用于AGG_PLAIN和AGG_SORTED模式: //per-group指针的grouping set编号数组 AggStatePerGroup *pergroups; /* grouping set indexed array of per-group * pointers */ //当前组的第一个元组拷贝 HeapTuple grp_firstTuple; /* copy of first tuple of current group */ /* these fields are used in AGG_HASHED and AGG_MIXED modes: */ //--------- 下面的列用于AGG_HASHED和AGG_MIXED模式: //是否已填充hash表? bool table_filled; /* hash table filled yet? */ //hash桶数? int num_hashes; //相应的哈希表数据数组 AggStatePerHash perhash; /* array of per-hashtable data */ //per-group指针的grouping set编号数组 AggStatePerGroup *hash_pergroup; /* grouping set indexed array of * per-group pointers */ /* support for evaluation of agg input expressions: */ //---------- agg输入表达式解析支持 #define FIELDNO_AGGSTATE_ALL_PERGROUPS 34 //首先是->pergroups,然后是hash_pergroup AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than * ->hash_pergroup */ //投影实现机制 ProjectionInfo *combinedproj; /* projection machinery */ } AggState; /* Primitive options supported by nodeAgg.c: */ //nodeag .c支持的基本选项 #define AGGSPLITOP_COMBINE 0x01 /* substitute combinefn for transfn */ #define AGGSPLITOP_SKIPFINAL 0x02 /* skip finalfn, return state as-is */ #define AGGSPLITOP_SERIALIZE 0x04 /* apply serializefn to output */ #define AGGSPLITOP_DESERIALIZE 0x08 /* apply deserializefn to input */ /* Supported operating modes (i.e., useful combinations of these options): */ //支持的操作模式 typedef enum AggSplit { /* Basic, non-split aggregation: */ //基本 : 非split聚合 AGGSPLIT_SIMPLE = 0, /* Initial phase of partial aggregation, with serialization: */ //部分聚合的初始步骤,序列化 AGGSPLIT_INITIAL_SERIAL = AGGSPLITOP_SKIPFINAL | AGGSPLITOP_SERIALIZE, /* Final phase of partial aggregation, with deserialization: */ //部分聚合的最终步骤,反序列化 AGGSPLIT_FINAL_DESERIAL = AGGSPLITOP_COMBINE | AGGSPLITOP_DESERIALIZE } AggSplit; /* Test whether an AggSplit value selects each primitive option: */ //测试AggSplit选择了哪些基本选项 #define DO_AGGSPLIT_COMBINE(as) (((as) & AGGSPLITOP_COMBINE) != 0) #define DO_AGGSPLIT_SKIPFINAL(as) (((as) & AGGSPLITOP_SKIPFINAL) != 0) #define DO_AGGSPLIT_SERIALIZE(as) (((as) & AGGSPLITOP_SERIALIZE) != 0) #define DO_AGGSPLIT_DESERIALIZE(as) (((as) & AGGSPLITOP_DESERIALIZE) != 0) ``` ### 二、源码解读 **agg_retrieve_direct** agg_retrieve_direct计算聚合的最终结果,适用于不使用Hash算法的情况. ``` /* * ExecAgg for non-hashed case * 适用于不使用Hash算法的情况. */ static TupleTableSlot * agg_retrieve_direct(AggState *aggstate) { Agg *node = aggstate->phase->aggnode;//aggstate Node ExprContext *econtext;//表达式解析上下文 ExprContext *tmpcontext;//临时上下文 AggStatePerAgg peragg;//聚合 AggStatePerGroup *pergroups;//分组信息 TupleTableSlot *outerslot;//outer元组slot TupleTableSlot *firstSlot;//第1个slot TupleTableSlot *result;//结果元组 bool hasGroupingSets = aggstate->phase->numsets > 0;//是否有grouping set int numGroupingSets = Max(aggstate->phase->numsets, 1); int currentSet; int nextSetSize; int numReset; int i; /* * get state info from node * 获取状态信息 * * econtext is the per-output-tuple expression context * econtext是per-output-tuple表达式上下文 * * tmpcontext is the per-input-tuple expression context * tmpcontext是per-input-tuple表达式上下文 */ econtext = aggstate->ss.ps.ps_ExprContext; tmpcontext = aggstate->tmpcontext; peragg = aggstate->peragg; pergroups = aggstate->pergroups; firstSlot = aggstate->ss.ss_ScanTupleSlot; /* * We loop retrieving groups until we find one matching * aggstate->ss.ps.qual * 循环检索分组直至找到一个匹配aggstate->ss.ps.qual表达式的分组. * * For grouping sets, we have the invariant that aggstate->projected_set * is either -1 (initial call) or the index (starting from 0) in * gset_lengths for the group we just completed (either by projecting a * row or by discarding it in the qual). * 对于grouping set,aggstate->projected_set是个不变量, * 要么是-1(初始调用),要么是已完成的分组在gset_lengths中的索引编号(从0开始) * (通过投影一行或者在表达式中丢弃一行实现) */ while (!aggstate->agg_done) { //----------- 循环处理 /* * Clear the per-output-tuple context for each group, as well as * aggcontext (which contains any pass-by-ref transvalues of the old * group). Some aggregate functions store working state in child * contexts; those now get reset automatically without us needing to * do anything special. * 跟aggcontext(包含原分组通过引用传递的转换值)一样,每一个分组都会重置per-output-tuple上下文. * 某些聚合函数在子上下文中存储工作状态,这种情况下,不需要做额外的工作,会自动重置. * * We use ReScanExprContext not just ResetExprContext because we want * any registered shutdown callbacks to be called. That allows * aggregate functions to ensure they've cleaned up any non-memory * resources. * 使用ReScanExprContext而不是ResetExprContext是因为我们希望所有已注册的shutdown回调函数可以调用. * 这可以允许聚合函数确保它们已清理了所有非内存类资源. */ ReScanExprContext(econtext); /* * Determine how many grouping sets need to be reset at this boundary. * 确定有多少grouping sets在此边界下需要重置. */ if (aggstate->projected_set >= 0 && aggstate->projected_set < numGroupingSets) numReset = aggstate->projected_set + 1; else numReset = numGroupingSets; /* * numReset can change on a phase boundary, but that's OK; we want to * reset the contexts used in _this_ phase, and later, after possibly * changing phase, initialize the right number of aggregates for the * _new_ phase. * numReset可能在每个阶段的边界处出现变化,但这样也不会出现问题. * 我们希望重置在该阶段的上下文,并在稍后在可能变化的阶段之后,为新的阶段初始化正确的聚合编号. */ for (i = 0; i < numReset; i++) { ReScanExprContext(aggstate->aggcontexts[i]); } /* * Check if input is complete and there are no more groups to project * in this phase; move to next phase or mark as done. * 检查输入是否完成并且没有更多的组在本阶段用于投影. * 移到下一个阶段或者标记为已完成. */ if (aggstate->input_done == true && aggstate->projected_set >= (numGroupingSets - 1)) { if (aggstate->current_phase < aggstate->numphases - 1) { //仍在处理中 initialize_phase(aggstate, aggstate->current_phase + 1); aggstate->input_done = false; aggstate->projected_set = -1; numGroupingSets = Max(aggstate->phase->numsets, 1); node = aggstate->phase->aggnode; numReset = numGroupingSets; } else if (aggstate->aggstrategy == AGG_MIXED) { //照理,不会进入这个分支(AGG_MIXED不是Hash才有吗?) /* * Mixed mode; we've output all the grouped stuff and have * full hashtables, so switch to outputting those. */ initialize_phase(aggstate, 0); aggstate->table_filled = true; ResetTupleHashIterator(aggstate->perhash[0].hashtable, &aggstate->perhash[0].hashiter); select_current_set(aggstate, 0, true); return agg_retrieve_hash_table(aggstate); } else { //已完成处理 aggstate->agg_done = true; break; } } /* * Get the number of columns in the next grouping set after the last * projected one (if any). This is the number of columns to compare to * see if we reached the boundary of that set too. * 在最后一次投影操作后获得下一个grouping set的列数. * 这是要比较的列数，看看我们是否也达到了集合的边界。 */ if (aggstate->projected_set >= 0 && aggstate->projected_set < (numGroupingSets - 1)) nextSetSize = aggstate->phase->gset_lengths[aggstate->projected_set + 1]; else nextSetSize = 0; /*---------- * If a subgroup for the current grouping set is present, project it. * 如果子分组已存在,则执行投影. * * We have a new group if: * - we're out of input but haven't projected all grouping sets * (checked above) * OR * - we already projected a row that wasn't from the last grouping * set * AND * - the next grouping set has at least one grouping column (since * empty grouping sets project only once input is exhausted) * AND * - the previous and pending rows differ on the grouping columns * of the next grouping set * * 如果出现下面情况,则会有新的分组: * - 已完成输入处理,但仍未投影所有的grouping set(上面会执行检查) * - 已投影了一行,但这一行并不是从最后一个grouping set而来的 * 同时 * - 下一个grouping set至少有要一个grouping列(因为空grouping sets投影一次输入就销毁了) * 同时 * - 上一个和接下来的行与下一个grouping set中的分组列不同 *---------- */ tmpcontext->ecxt_innertuple = econtext->ecxt_outertuple; if (aggstate->input_done || (node->aggstrategy != AGG_PLAIN && aggstate->projected_set != -1 && aggstate->projected_set < (numGroupingSets - 1) && nextSetSize > 0 && !ExecQualAndReset(aggstate->phase->eqfunctions[nextSetSize - 1], tmpcontext))) { aggstate->projected_set += 1; Assert(aggstate->projected_set < numGroupingSets); Assert(nextSetSize > 0 || aggstate->input_done); } else { /* * We no longer care what group we just projected, the next * projection will always be the first (or only) grouping set * (unless the input proves to be empty). * 不再关心刚才已投影的分组,下一个投影通常会是第一个grouping set(除非输入已验证为空) */ aggstate->projected_set = 0; /* * If we don't already have the first tuple of the new group, * fetch it from the outer plan. * 如果不再有新分组的第一个元组,则从outer plan中提取一行. */ if (aggstate->grp_firstTuple == NULL) { //提取一行 outerslot = fetch_input_tuple(aggstate); if (!TupIsNull(outerslot)) { //成功提取一行 /* * Make a copy of the first input tuple; we will use this * for comparisons (in group mode) and for projection. * 拷贝之 */ aggstate->grp_firstTuple = ExecCopySlotTuple(outerslot); } else { /* outer plan produced no tuples at all */ //不再产生新行 if (hasGroupingSets) { //----------- 存在grouping set /* * If there was no input at all, we need to project * rows only if there are grouping sets of size 0. * Note that this implies that there can't be any * references to ungrouped Vars, which would otherwise * cause issues with the empty output slot. * 如果根本就不存在输入,只需要在大小为0的grouping set上投影哪些行即可. * 注意这意味着不能依赖未分组的Vars,否则的话会导致输出slot为空. * * XXX: This is no longer true, we currently deal with * this in finalize_aggregates(). * XXX: 这已不复存在,已在finalize_aggregates中进行处理. */ aggstate->input_done = true; while (aggstate->phase->gset_lengths[aggstate->projected_set] > 0) { aggstate->projected_set += 1; if (aggstate->projected_set >= numGroupingSets) { /* * We can't set agg_done here because we might * have more phases to do, even though the * input is empty. So we need to restart the * whole outer loop. * 就算输入为空,但也不能在这里还设置agg_done为T,因为可能还有后续的阶段需要处理. * 因此需要重启整个外循环. */ break; } } if (aggstate->projected_set >= numGroupingSets) continue; } else { aggstate->agg_done = true; /* If we are grouping, we should produce no tuples too */ if (node->aggstrategy != AGG_PLAIN) return NULL; } } } /* * Initialize working state for a new input tuple group. * 为新输入的元组组初始化工作状态. */ initialize_aggregates(aggstate, pergroups, numReset); if (aggstate->grp_firstTuple != NULL) { /* * Store the copied first input tuple in the tuple table slot * reserved for it. The tuple will be deleted when it is * cleared from the slot. * 在元组表slot中拷贝存储第一个输入元组. * 该元组在清理slot时会被删除. */ ExecStoreTuple(aggstate->grp_firstTuple, firstSlot, InvalidBuffer, true); aggstate->grp_firstTuple = NULL; /* 不需要保留双份指针. don't keep two pointers */ /* set up for first advance_aggregates call */ //为第一次advance_aggregates调用设置参数 tmpcontext->ecxt_outertuple = firstSlot; /* * Process each outer-plan tuple, and then fetch the next one, * until we exhaust the outer plan or cross a group boundary. * 处理每一个outer-plan元组,然后提取下一个, * 直至outer plan已消耗完毕或者已跨越分组边界. */ for (;;) { /* * During phase 1 only of a mixed agg, we need to update * hashtables as well in advance_aggregates. * 只有在混合AGG的第一阶段,我们还需要在advance_aggregates中更新哈希表. */ if (aggstate->aggstrategy == AGG_MIXED && aggstate->current_phase == 1) { lookup_hash_entries(aggstate); } /* Advance the aggregates (or combine functions) */ //推动聚合(或者组合函数) advance_aggregates(aggstate); /* Reset per-input-tuple context after each tuple */ //在每一个元组后重置per-input-tuple上下文 ResetExprContext(tmpcontext); outerslot = fetch_input_tuple(aggstate); if (TupIsNull(outerslot)) { /* no more outer-plan tuples available */ //已无更多可用的outer slot if (hasGroupingSets) { aggstate->input_done = true; break; } else { aggstate->agg_done = true; break; } } /* set up for next advance_aggregates call */ //为下一次advance_aggregates调用作准备 tmpcontext->ecxt_outertuple = outerslot; /* * If we are grouping, check whether we've crossed a group * boundary. * 如果是分组,检查是否已跨越分组边界. */ if (node->aggstrategy != AGG_PLAIN) { tmpcontext->ecxt_innertuple = firstSlot; if (!ExecQual(aggstate->phase->eqfunctions[node->numCols - 1], tmpcontext)) { aggstate->grp_firstTuple = ExecCopySlotTuple(outerslot); break; } } } } /* * Use the representative input tuple for any references to * non-aggregated input columns in aggregate direct args, the node * qual, and the tlist. (If we are not grouping, and there are no * input rows at all, we will come here with an empty firstSlot * ... but if not grouping, there can't be any references to * non-aggregated input columns, so no problem.) * 对于聚合直接参数/节点表达式和投影列tlist中的非聚合输入列的引用,使用代表性的输入元组. * (如果不是grouping而且没有输入元组,将使用空的firstSlot,但如果是非grouping, * 不可能存在依赖非聚合输入列,因此不会存在问题) */ econtext->ecxt_outertuple = firstSlot; } Assert(aggstate->projected_set >= 0); currentSet = aggstate->projected_set; //投影处理 prepare_projection_slot(aggstate, econtext->ecxt_outertuple, currentSet); select_current_set(aggstate, currentSet, false); finalize_aggregates(aggstate, peragg, pergroups[currentSet]); /* * If there's no row to project right now, we must continue rather * than returning a null since there might be more groups. * 如不需要马上进行进行投影,必须继续执行而不是返回NULL,因为还需要处理更多的groups. */ result = project_aggregates(aggstate); if (result) return result; } /* No more groups */ //DONE! return NULL; } ``` ### 三、跟踪分析 测试脚本 ``` -- 禁用并行 set max_parallel_workers_per_gather=0; -- 禁用hashagg set enable_hashagg = off; select bh,avg(c1),min(c1),max(c2) from t_agg_simple group by bh; ``` 跟踪分析 ``` (gdb) b agg_retrieve_direct Breakpoint 1 at 0x6ee511: file nodeAgg.c, line 1572. (gdb) c Continuing. Breakpoint 1, agg_retrieve_direct (aggstate=0x268f640) at nodeAgg.c:1572 1572 Agg *node = aggstate->phase->aggnode; ``` 输入参数 ``` (gdb) p *aggstate $1 = {ss = {ps = {type = T_AggState, plan = 0x25af578, state = 0x268f428, ExecProcNode = 0x6ee438 <ExecAgg>, ExecProcNodeReal = 0x6ee438 <ExecAgg>, instrument = 0x0, worker_instrument = 0x0, worker_jit_instrument = 0x0, qual = 0x0, lefttree = 0x268faf0, righttree = 0x0, initPlan = 0x0, subPlan = 0x0, chgParam = 0x0, ps_ResultTupleSlot = 0x2690d50, ps_ExprContext = 0x268fa30, ps_ProjInfo = 0x2690e90, scandesc = 0x26907a0}, ss_currentRelation = 0x0, ss_currentScanDesc = 0x0, ss_ScanTupleSlot = 0x2690a78}, aggs = 0x25d4290, numaggs = 3, numtrans = 3, aggstrategy = AGG_SORTED, aggsplit = AGGSPLIT_SIMPLE, phase = 0x2691290, numphases = 2, current_phase = 1, peragg = 0x2690f28, pertrans = 0x26b24d0, hashcontext = 0x0, aggcontexts = 0x268f858, tmpcontext = 0x268f878, curaggcontext = 0x268f970, curperagg = 0x0, curpertrans = 0x0, input_done = false, agg_done = false, projected_set = -1, current_set = 0, grouped_cols = 0x0, all_grouped_cols = 0x0, maxsets = 1, phases = 0x2691258, sort_in = 0x0, sort_out = 0x0, sort_slot = 0x0, pergroups = 0x25d4da0, grp_firstTuple = 0x0, table_filled = false, num_hashes = 0, perhash = 0x0, hash_pergroup = 0x0, all_pergroups = 0x25d4da0, combinedproj = 0x0} ``` 需要2个阶段,分别是AGG_PLAIN/AGG_SORTED ``` (gdb) p aggstate->phases[0] $2 = {aggstrategy = AGG_PLAIN, numsets = 0, gset_lengths = 0x0, grouped_cols = 0x0, eqfunctions = 0x0, aggnode = 0x0, sortnode = 0x0, evaltrans = 0x0} (gdb) p aggstate->phases[1] $3 = {aggstrategy = AGG_SORTED, numsets = 0, gset_lengths = 0x0, grouped_cols = 0x0, eqfunctions = 0x25d4388, aggnode = 0x25af578, sortnode = 0x0, evaltrans = 0x25d5488} ``` 不存在grouping set. 变量numGroupingSets设置为1 ``` (gdb) n 1580 bool hasGroupingSets = aggstate->phase->numsets > 0; (gdb) p aggstate->phase->numsets $5 = 0 (gdb) n 1581 int numGroupingSets = Max(aggstate->phase->numsets, 1); (gdb) n 1594 econtext = aggstate->ss.ps.ps_ExprContext; (gdb) p numGroupingSets $6 = 1 ``` 设置内存上下文 ``` (gdb) n 1595 tmpcontext = aggstate->tmpcontext; (gdb) 1597 peragg = aggstate->peragg; (gdb) p *econtext $7 = {type = T_ExprContext, ecxt_scantuple = 0x0, ecxt_innertuple = 0x0, ecxt_outertuple = 0x0, ecxt_per_query_memory = 0x268f310, ecxt_per_tuple_memory = 0x26a6370, ecxt_param_exec_vals = 0x0, ecxt_param_list_info = 0x0, ecxt_aggvalues = 0x25d4d48, ecxt_aggnulls = 0x25d4d80, caseValue_datum = 0, caseValue_isNull = true, domainValue_datum = 0, domainValue_isNull = true, ecxt_estate = 0x268f428, ecxt_callbacks = 0x0} (gdb) p *tmpcontext $8 = {type = T_ExprContext, ecxt_scantuple = 0x0, ecxt_innertuple = 0x0, ecxt_outertuple = 0x0, ecxt_per_query_memory = 0x268f310, ecxt_per_tuple_memory = 0x2691320, ecxt_param_exec_vals = 0x0, ecxt_param_list_info = 0x0, ecxt_aggvalues = 0x0, ecxt_aggnulls = 0x0, caseValue_datum = 0, caseValue_isNull = true, domainValue_datum = 0, domainValue_isNull = true, ecxt_estate = 0x268f428, ecxt_callbacks = 0x0} (gdb) ``` 获取聚合信息,一共有3个 ``` (gdb) n 1598 pergroups = aggstate->pergroups; (gdb) 1599 firstSlot = aggstate->ss.ss_ScanTupleSlot; (gdb) p *peragg $9 = {aggref = 0x26a02d0, transno = 0, finalfn_oid = 0, finalfn = {fn_addr = 0x0, fn_oid = 0, fn_nargs = 0, fn_strict = false, fn_retset = false, fn_stats = 0 '\000', fn_extra = 0x0, fn_mcxt = 0x0, fn_expr = 0x0}, numFinalArgs = 1, aggdirectargs = 0x0, resulttypeLen = 4, resulttypeByVal = true, shareable = true} (gdb) p peragg[0] $10 = {aggref = 0x26a02d0, transno = 0, finalfn_oid = 0, finalfn = {fn_addr = 0x0, fn_oid = 0, fn_nargs = 0, fn_strict = false, fn_retset = false, fn_stats = 0 '\000', fn_extra = 0x0, fn_mcxt = 0x0, fn_expr = 0x0}, numFinalArgs = 1, aggdirectargs = 0x0, resulttypeLen = 4, resulttypeByVal = true, shareable = true} (gdb) p peragg[1] $11 = {aggref = 0x26a0048, transno = 1, finalfn_oid = 0, finalfn = {fn_addr = 0x0, fn_oid = 0, fn_nargs = 0, fn_strict = false, fn_retset = false, fn_stats = 0 '\000', fn_extra = 0x0, fn_mcxt = 0x0, fn_expr = 0x0}, numFinalArgs = 1, aggdirectargs = 0x0, resulttypeLen = 4, resulttypeByVal = true, shareable = true} (gdb) p peragg[2] $12 = {aggref = 0x269fdc0, transno = 2, finalfn_oid = 1964, finalfn = {fn_addr = 0x978251 <int8_avg>, fn_oid = 1964, fn_nargs = 1, fn_strict = true, fn_retset = false, fn_stats = 2 '\002', fn_extra = 0x0, fn_mcxt = 0x268f310, fn_expr = 0x25d5190}, numFinalArgs = 1, aggdirectargs = 0x0, resulttypeLen = -1, resulttypeByVal = false, shareable = true} ``` 分组只有一个 ``` (gdb) p pergroups[0] $14 = (AggStatePerGroup) 0x25d4dc0 (gdb) p *pergroups[0] $15 = {transValue = 0, transValueIsNull = false, noTransValue = false} ``` 进入循环 ``` (gdb) n 1610 while (!aggstate->agg_done) (gdb) 1624 ReScanExprContext(econtext); (gdb) ``` 重置内存上下文 ``` (gdb) 1629 if (aggstate->projected_set >= 0 && (gdb) 1633 numReset = numGroupingSets; (gdb) 1642 for (i = 0; i < numReset; i++) (gdb) 1644 ReScanExprContext(aggstate->aggcontexts[i]); (gdb) 1642 for (i = 0; i < numReset; i++) (gdb) ``` 检查输入是否已完成处理/本组已完成投影(实际不满足条件) ``` (gdb) 1651 if (aggstate->input_done == true && (gdb) 1688 if (aggstate->projected_set >= 0 && (gdb) p aggstate->input_done $16 = false (gdb) p aggstate->projected_set $17 = -1 ``` 设置待处理的元组,为NULL ``` (gdb) 1711 tmpcontext->ecxt_innertuple = econtext->ecxt_outertuple; (gdb) 1712 if (aggstate->input_done || (gdb) (gdb) p *tmpcontext->ecxt_innertuple Cannot access memory at address 0x0 ``` 如果子分组已存在,则执行投影(实际不满足条件). ``` (gdb) n 1713 (node->aggstrategy != AGG_PLAIN && (gdb) p node->aggstrategy $18 = AGG_SORTED (gdb) n 1712 if (aggstate->input_done || (gdb) 1714 aggstate->projected_set != -1 && (gdb) 1713 (node->aggstrategy != AGG_PLAIN && (gdb) 1732 aggstate->projected_set = 0; (gdb) p aggstate->input_done $19 = false (gdb) p aggstate->projected_set $20 = -1 (gdb) p node->aggstrategy $21 = AGG_SORTED (gdb) ``` 从outer plan中提取一行,并拷贝为首行 ``` (gdb) n 1738 if (aggstate->grp_firstTuple == NULL) (gdb) p aggstate->grp_firstTuple $22 = (HeapTuple) 0x0 (gdb) n 1740 outerslot = fetch_input_tuple(aggstate); (gdb) 1741 if (!TupIsNull(outerslot)) (gdb) p *outerslot $23 = {type = T_TupleTableSlot, tts_isempty = false, tts_shouldFree = false, tts_shouldFreeMin = false, tts_slow = false, tts_tuple = 0x26909f8, tts_tupleDescriptor = 0x26907a0, tts_mcxt = 0x268f310, tts_buffer = 0, tts_nvalid = 0, tts_values = 0x2690a18, tts_isnull = 0x2690a30, tts_mintuple = 0x26b8ad8, tts_minhdr = {t_len = 40, t_self = {ip_blkid = { bi_hi = 0, bi_lo = 0}, ip_posid = 0}, t_tableOid = 0, t_data = 0x26b8ad0}, tts_off = 0, tts_fixedTupleDescriptor = true} (gdb) (gdb) n 1747 aggstate->grp_firstTuple = ExecCopySlotTuple(outerslot); (gdb) ``` 为新输入的元组组初始化工作状态. ``` (gdb) 1797 initialize_aggregates(aggstate, pergroups, numReset); ``` 把元组拷贝到内存上下文中,并执行聚合运算(advance_aggregates) ``` (gdb) n 1799 if (aggstate->grp_firstTuple != NULL) (gdb) 1806 ExecStoreTuple(aggstate->grp_firstTuple, (gdb) 1810 aggstate->grp_firstTuple = NULL; /* don't keep two pointers */ (gdb) 1813 tmpcontext->ecxt_outertuple = firstSlot; (gdb) 1825 if (aggstate->aggstrategy == AGG_MIXED && (gdb) 1832 advance_aggregates(aggstate); (gdb) 1835 ResetExprContext(tmpcontext); (gdb) ``` 继续提取行,拷贝到内存上下文中 ``` (gdb) n 1837 outerslot = fetch_input_tuple(aggstate); (gdb) 1838 if (TupIsNull(outerslot)) (gdb) 1853 tmpcontext->ecxt_outertuple = outerslot; (gdb) 1859 if (node->aggstrategy != AGG_PLAIN) (gdb) 1861 tmpcontext->ecxt_innertuple = firstSlot; (gdb) 1862 if (!ExecQual(aggstate->phase->eqfunctions[node->numCols - 1], (gdb) 1869 } ``` 执行聚合运算,并继续提取下一行 ``` 825 if (aggstate->aggstrategy == AGG_MIXED && (gdb) 1832 advance_aggregates(aggstate); (gdb) 1835 ResetExprContext(tmpcontext); (gdb) 1837 outerslot = fetch_input_tuple(aggstate); (gdb) 1838 if (TupIsNull(outerslot)) (gdb) 1853 tmpcontext->ecxt_outertuple = outerslot; (gdb) 1859 if (node->aggstrategy != AGG_PLAIN) (gdb) 1861 tmpcontext->ecxt_innertuple = firstSlot; (gdb) ``` 如果是分组,检查是否已跨越分组边界,如已越界在跳出循环. ``` 1862 if (!ExecQual(aggstate->phase->eqfunctions[node->numCols - 1], (gdb) 1865 aggstate->grp_firstTuple = ExecCopySlotTuple(outerslot); (gdb) 1866 break; ``` 已获得一行结果行,返回结果 ``` (gdb) 1880 econtext->ecxt_outertuple = firstSlot; (gdb) n 1883 Assert(aggstate->projected_set >= 0); (gdb) 1885 currentSet = aggstate->projected_set; (gdb) 1887 prepare_projection_slot(aggstate, econtext->ecxt_outertuple, currentSet); (gdb) p aggstate->projected_set $24 = 0 (gdb) n 1889 select_current_set(aggstate, currentSet, false); (gdb) 1893 pergroups[currentSet]); (gdb) 1891 finalize_aggregates(aggstate, (gdb) 1899 result = project_aggregates(aggstate); (gdb) 1900 if (result) (gdb) 1901 return result; (gdb) p *result $25 = {type = T_TupleTableSlot, tts_isempty = false, tts_shouldFree = false, tts_shouldFreeMin = false, tts_slow = false, tts_tuple = 0x0, tts_tupleDescriptor = 0x2690b38, tts_mcxt = 0x268f310, tts_buffer = 0, tts_nvalid = 4, tts_values = 0x2690db0, tts_isnull = 0x2690dd0, tts_mintuple = 0x0, tts_minhdr = {t_len = 0, t_self = {ip_blkid = { bi_hi = 0, bi_lo = 0}, ip_posid = 0}, t_tableOid = 0, t_data = 0x0}, tts_off = 0, tts_fixedTupleDescriptor = true} (gdb) ``` DONE! ### 四、参考资料 [PostgreSQL 源码解读（178）- 查询#95(聚合函数)#1相关数据结构](http://blog.itpub.net/6906/viewspace-2642505/) [PostgreSQL 源码解读（186）- 查询#102(聚合函数#7-advance_aggregates) ](http://blog.itpub.net/6906/viewspace-2643822/)