如何使用 PostgreSQL ExecAgg 函数

1.数据结构

AggState
聚合函数执行时的状态结构，包括AggStatePerAgg等结构

/* ----- --------------------- * AggState 信息 * * ss.ss_ScanTupleSlot 指底层计划的输出。 * ss.ss_ScanTupleSlot 指底层计划的输出。 * (ss = ScanState,ps = PlanState) * * 注意：ss.ps.ps_ExprContext 包含 ecxt_a ggvalues 和 * ecxt_aggnulls 数组，它们在评估 Agg 节点输出期间保存当前 * 输入组的计算 agg 值元组。我们 * 创建第二个 ExprContext、tmpcontext，在其中计算输入 * 表达式并运行聚合转换函数。 * 注：ss.ps.ps_ExprContext 包含 ecxt_aggvalues 和 ecxt_aggnulls 数组， * 这两个数组保存了计算 agg 节点的输出元组时计算出的当前输入组的 agg 值。 *--------------------- *//* 这些结构在 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; /* 它的第一个字段是 NodeTag */ // targetlist 和 quals List 中的所有 Aggrefs *aggs; /* targetlist & quals 中的所有 Aggref 节点 */ //链表的大小（可以为0） int numtrans; /* 列表的长度（可以为零！) */ //pertrans 条目大小 int numtrans; /* pertrans 项目的数量 */ //AggStrategy aggstrategy; /* 策略模式 */ //聚合分割模式，参见nodes.h AggSplit aggsplit; /* agg-splitting模式，参见nodes.h */ //指向当前步数据AggStatePerPhase阶段的指针； /* 指向当前阶段数据的指针 */ //步数（包括0） int numphases; /* 阶段数（包括阶段0） */ //当前步 int​​​​current_phase; /* 当前阶段号 */ //per-Aggref 信息 AggStatePerAgg peragg; /* per-Aggref 信息 */ //per-Trans 状态信息 AggStatePerTrans pertrans; /* per-Trans 状态信息 */ //数据的长生命周期 ExprContexts (hashtable) ExprContext *hashcontext; /* 长期数据的 econtexts (hashtable) */ ////长期数据的 ExprContexts (由每个 GS 使用) ExprContext **aggcontexts; /* 长期数据的 econtexts (per GS) */ //输入表达式的 ExprContext ExprContext *tmpcontext; /* 输入表达式的econtext */#define FIELDNO_AGGSTATE_CURAGGCONTEXT 14 //当前活动的aggcontext ExprContext *curaggcontext; /* 当前活动的aggcontext */ //当前活动的聚合（如果存在） AggStatePerAgg curperagg; /* 当前活动的聚合，如果有的话 */ #define FIELDNO_AGGSTATE_CURPERTRANS 16 //当前活动的跨状态 AggStatePerTrans curpertrans; /* 当前活动的反式状态，如果有的话 */ //输入结束？布尔输入_完成； /* 指示输入结束 */ //聚合扫描结束？布尔 agg_done; /* 指示Agg扫描完成 */ //最后的分组集 intprojected_set; /* 最后投影的分组集 */#define FIELDNO_AGGSTATE_CURRENT_SET 20 //当前要解析的分组集 int current_set; /* 当前正在评估的分组集 */ //当前投影操作的分组列 Bitmapset *grouped_cols; /* 当前投影中的分组列 */ //倒序分组列列表 List   *all_grouped_cols; /* 按 DESC 顺序排列的所有分组列的列表 */ /* 这些 fi字段用于对集合阶段数据进行分组 */ //-------- 以下列用于对集合步骤数据进行分组 // 所有步骤中的最大集合大小 int maxsets; /* 任意阶段的最大集合数 */ //所有步骤的数组 AggStatePerPhase Phases; /* 所有阶段的数组 */ //对于阶段 > 1，已排序的输入信息 Tuplesortstate *sort_in; /* 对输入进行排序 > 1 *///对于下一步，输入已被复制 Tuplesortstate *sort_out; /* 输入被复制到此处用于下一阶段 */ //用于排序结果的槽 TupleTableSlot *sort_slot; /* 排序结果槽 */ /* 这些字段用于 AGG_PLAIN 和 AGG_SORTED 模式： */ //-------- 以下列用于 AGG_PLAIN 和 AGG_SORTED 模式： //grouping set number array of per -组指针 AggStatePerGroup *pergroups; /* 每组的分组集索引数组                                                                     nbsp;*/ //当前组的第一个元组副本 HeapTuple grp_firstTuple; /* 第一个元组 o 的副本f current group */ /* 这些字段用于 AGG_HASHED 和 AGG_MIXED 模式： */ //----- ---- 以下列用于 AGG_HASHED 和 AGG_MIXED 模式： //哈希表是否已填满？布尔表填充； /* 哈希表填满了吗？ */ //哈希桶的数量？ int num_hashes； //对应哈希表数据数组 AggStatePerHash perhash; /* 每个哈希表数据的数组 */ // 每组指针的分组集数数组   AggStatePerGroup *hash_pergroup; /* 分组设置 * 每组指针的索引数组 */ /* 支持计算聚合输入表达式： */ //---------- 聚合输入表达式解析支持 #define FIELDNO_AGGSTATE_ALL_PERGROUPS 34 //First ->pergroups，然后 hash_pergroup AggStatePerGroup *all_pergroups； /* 第一个 ->pergroups 的数组，thannbsp； TOP_COMBINE 0x01 /* 用combinefn 代替transfn */#define AGGSPLITOP_SKIPFINAL 0x02 /* 跳过finalfn，按原样返回状态*/#define AGGSPLITOP_SERIALIZE 0x04 /* 将serializefn 应用到输出* /#define AGGSPLITOP_DESERIALIZE 0x08 /* 对输入应用反序列化*//* 支持的操作模式（即这些选项的有用组合）： */// 支持的操作模式 typedef enum AggSplit{ /* 基本、非分割聚合： */ // 基本: 不分割聚合 AGGSPLIT_SIMPLE = 0, /* 部分聚合的初始阶段，带序列化： */ //部分聚合的初始阶段，序列化 AGGSPLIT_INITIAL_SERIAL = AGGSPLITOP_SKIPFINAL | AGGSPLITOP_SERIALIZE, /* 部分聚合的最后阶段，带反序列化： */ //部分聚合的最后一步，反序列化 AGGSPLIT_FINAL_DESERIAL = AGGSPLITOP_COMBINE | AGGSPLITOP_DESERIALIZE} AggSplit;/* 测试一个 AggSplit 值是否选择每个原始选项： *///测试 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)

2.源码解读

ExecAgg函数，首先获取AggState运行状态，然后根据每个阶段（aggstate->phase）的策略（aggstrategy）执行相应的逻辑。如果采用Hash聚合，则只有一个节点，但有两种策略。第一个是 AGG_HASHED。该策略根据分组的列值对输入元组进行哈希处理，同时执行转换函数来计算中间结果。 value，缓存到哈希表中；然后执行 AGG_MIXED 策略，从哈希表中获取结果元组并返回结果元组（每个结果是一个结果行）。

/* * ExecAgg - * * ExecAgg 从其外部子计划接收元组并聚合出现在目标列表或质量中的每个聚合函数使用的适当属性（Aggref *节点）节点。要聚合的元组数量取决于选择的是分组聚合还是普通聚合。在分组聚合中，我们为每个组生成一个结果 * 行；在普通聚合中，整个查询只有一个结果行*。无论哪种情况，每个聚合的值都存储在表达式上下文中，以便在 ExecProject 评估结果元组时使用。 * ExecAgg 接收从外部子计划返回的元组，其中包含每个聚合函数（出现在投影列或节点表达式上）的适当属性以执行聚合。 * 需要聚合的元组数量取决于是否已分组或选择普通聚合。 * 分组聚合操作宏中，为每组生成结果行；在普通聚合中，整个查询只有一个结果行。 * 无论哪种情况，每个聚合结果值都将存储在表达式上下文中（ExecProject 将解析e 结果元组) */static TupleTableSlot *ExecAgg(PlanState *pstate){ AggState *node = castNode(AggState, pstate); TupleTableSlot *结果= NULL;检查中断（）； if (!node->agg_done) {  /* 根据策略调度 */ // 根据策略调度 switch (node->phase->aggstrategy) { case AGG_HASHED: if (!node->table_filled) agg_fill_hash_table(node); /* FALLTHROUGH */ 填充后，执行 MIXED case AGG_MIXED:   result = agg_retrieve_hash_table(node);休息;案例 AGG_PLAIN：案例 AGG_SORTED：结果 = agg_ret rieve_direct（节点）；休息; } if (!TupIsNull(result)) 返回结果； } return NULL;}

agg_fill_hash_table
读取输入并构建哈希表。
lookup_hash_entries函数基于inputTuples构建分组列哈希表（搜索或创建新条目），advance_aggregates 调用转换函数计算中间结果并缓存它们。

/* * ExecAgg 用于散列情况：读取输入并构建哈希表 * 读取输入并构建哈希表 */static voidagg_fill_hash_table(AggState *aggstate){ TupleTableSlot *outerslot; ExprContext *tmpcontext = aggstate->tmpcontext; /* * 处理每个外部计划元组，然后获取下一个，直到我们 * 耗尽外部计划。 * 处理outer-plan返回的每个元组，然后继续提取下一个元组，直到处理完所有元组。 */ for (;;) { //--------- 循环直到完成所有元组的处理 //提取输入元组   outerslot = fetch_input_tuple(aggstate); if (TupIsNull(outerslot)) break;//处理完成，退出循环 /* 设置lookup_hash_entries和advance_aggregates */ //配置lookup_hash_entries和advance_aggregates函数 //将tuple放入临时内存上下文中 tmpcontext->ecxt_outertuple =外槽； /* 查找或构建哈希表条目 */ //检索或构建哈希表条目lookup_hash_entries(aggstate); /* 推进聚合（或组合函数）*/// 推送聚合（或组合函数）             advance_aggregates(aggstate); /*      * 在每个元组之后重置每个输入元组上下文，但请注意    * 哈希查找也会执行此操作 * 重置每个输入元组内存上下文，但请注意哈希检索也会执行此操作 */ ResetExprContext(aggstate- >tmp上下文）； } aggstate->table_filled = true; /* 初始化遍历第一个哈希表 */ //初始化用于遍历第一个哈希表 select_current_set(aggstate, 0, true); ResetTupleHashIterator(aggstate->perhash[0].hashtable, &aggstate->perhash[0].hashter);}

agg_retrieve_hash_table
agg_retrieve_hash_table 函数检索哈希中的结果

/* * ExecAgg for hashed case：从哈希表中检索Group * ExecAgg（Hash实现版本）：在哈希表中检索Group */static TupleTableSlot *agg_retrieve_hash_table(聚合状态 *聚合状态){ExprContext *econtext; AggStatePerAgg peragg； AggStatePerGroup 每组； TupleHashEntryData *条目； TupleTableSlot *firstSlot; TupleTableSlot *结果； AggStatePerHash perhash； /* * 从节点获取状态信息。 * 从节点获取状态信息。 * * econtext 是每个输出元组的表达式上下文。 */ econtext = aggstate->ss.ps.ps_ExprContext; peragg = aggstate->peragg; firstSlot = aggstate->ss.ss_ScanTupleSlot; /* * 请注意，当我们在分组集之间进行更改时，perhash（以及因此通过它访问的任何内容）可以在循环内更改。 * 注意，组间切换时，循环中perhash可能会发生变化 */ perhash = &aggstate->perhash[aggstate->current_set]; /* * 我们循环检索组，直到找到一个满足条件的组 * aggstate-> ss.ps.qual * 循环遍历组，直到检索到满足 aggstate->ss.ps.qual 条件的组。 */ while (!aggstate->agg_done) { //------------ --- 选择 OK //获取 Slot TupleTableSlot *hashslot = perhash->hashslot; int int 我； //C哎呀中断 CHECK_FOR_INTERRUPTS(); /* * 查找哈希表中的下一个条目 * 检索哈希表中的下一个条目 */entry = ScanTupleHash Table(perhash->hashtable, &perhash->hashiter); if (entry == NULL) { //条目为NULL，切换到下一个集合 int nextset = aggstate->current_set + 1; iF(NextSet NUM_hashes) {/*******切换到下一组集合，重新初始化，并RESTART*LOOP.*到下一组集合，重新初始化并重新启动后续。 Ring*/select_current_set(aggstate, nextSet, nextSet, nextSet, true); } else else { /* 不再有哈希表，所以完成 */ 搜索完成 , set mark, exitaggstate->agg_done = true;返回空值； /* 清除每个组的每个输出元组上下文 * 清除每个组元组的每个输出元组上下文 * * * 我们故意不在这里使用 ReScanExprContext ；如果任何聚合 * 注册了关闭回调，则还不能调用它们，因为我们 * 可能还没有完成该聚合。 * ReScanExprContext不会在这里使用。如果有aggs注册了关闭回调， * 它不应该被调用，因为我们可能还没有完成对agg的处理。 */ ResetExprCon 文本(econtext); /* * 将代表性元组转换回具有右侧 * 列的元组。 * 将代表性元组转换回具有正确列的元组。 */ ExecStoreMinimalTuple(entry->firstTuple , hashslot, false);slot_getallattrs(hashslot); //清理元组 //重置firstSlot ExecClearTuple(firstSlot); memset(firstSlot->tts_isnull, true, firstSlot->tts_tupleDescriptor->natts * sizeof(bool)); for (i = 0; i < perhash->numhashGrpCols; i++) { //重置firstSlot int varNumber = perhash->hashGrpColIdxInput[i] - 1;firstSlot->tts_values[varNumber] = hashslot->tts_values[i]; firstSlot->tts_isnull[varNumber] = hashslot->tts_isnull[i]; pergroup = (AggStatePerGroup) 条目->附加； /* * 对 qual 和 tlist 中的非聚合输入列的任何引用使用代表性输入元组。*/Econtext-> Ecxt_outertuple= 第一个槽位； // 准备投影槽Prepare_project_slot(aggState, ECONTEXT->ECXT_OUTERTUPLE, AGGSTAT->Current_Set); // 最后，Finalize_aggregates (aggstate, peragg, pergroup); // 投影结果 = project_aggregates(aggstate); if (结果) 返回结果； } /* No more groups */ //No more groups, return NULL return NULL;}

"如何使用"PostgreSQL ExecAgg函数"的介绍就到此结束，谢谢阅读。如果想了解更多行业资讯，可以关注网站，小编将为大家输出更多优质实用文章！