本節繼續介紹聚合函數的實現,主要介紹了agg_retrieve_hash_table函數中與投影相關的實現邏輯,包括函數prepare_projection_slot/finalize_aggregates/project_aggregates.

一、數據結構

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)

二、源碼解讀

prepare_projection_slot
prepare_projection_slot函數基于指定的典型元組slot和grouping set準備finalize和project.
比如初始化isnull數組等.

/*
 * Prepare to finalize and project based on the specified representative tuple
 * slot and grouping set.
 * 基于指定的典型元組slot和grouping set準備finalize和project.
 *
 * In the specified tuple slot, force to null all attributes that should be
 * read as null in the context of the current grouping set.  Also stash the
 * current group bitmap where GroupingExpr can get at it.
 * 在指定的元組slot,強制在當前grouping set上下文中應為null的所有屬性值為null.
 * 還可以將當前組位圖保存在GroupingExpr可以獲得的位置.
 *
 * This relies on three conditions:
 * 這取決于下面3個條件:
 *
 * 1) Nothing is ever going to try and extract the whole tuple from this slot,
 * only reference it in evaluations, which will only access individual
 * attributes.
 * 1) 永遠不會嘗試從該slot中提取整個元組,只是在解析中依賴它,這只會訪問單個屬性.
 *
 * 2) No system columns are going to need to be nulled. (If a system column is
 * referenced in a group clause, it is actually projected in the outer plan
 * tlist.)
 * 2) 系統列不需要設置為null.
 *    (如在group語句中依賴系統列,實際上已在outer plan tlist中已完成投影)
 *
 * 3) Within a given phase, we never need to recover the value of an attribute
 * once it has been set to null.
 * 3) 在給定的階段,一旦屬性被設置為null,就不需要恢復屬性值.
 *
 * Poking into the slot this way is a bit ugly, but the consensus is that the
 * alternative was worse.
 * 以這種方法使用slot有點丑陋,但其他方式更糟糕.
 */
static void
prepare_projection_slot(AggState *aggstate, TupleTableSlot *slot, int currentSet)
{
    if (aggstate->phase->grouped_cols)
    {
        Bitmapset  *grouped_cols = aggstate->phase->grouped_cols[currentSet];
        aggstate->grouped_cols = grouped_cols;
        if (slot->tts_isempty)
        {
            /*
             * Force all values to be NULL if working on an empty input tuple
             * (i.e. an empty grouping set for which no input rows were
             * supplied).
             * 如輸入tuple為空,則強制所有值為NULL.
             * (如不提供輸入行的空grouping set)
             */
            ExecStoreAllNullTuple(slot);
        }
        else if (aggstate->all_grouped_cols)
        {
            ListCell   *lc;
            /* all_grouped_cols is arranged in desc order */
            //all_grouped_cols以倒序的方式組織
            slot_getsomeattrs(slot, linitial_int(aggstate->all_grouped_cols));
            foreach(lc, aggstate->all_grouped_cols)
            {
                int            attnum = lfirst_int(lc);
                if (!bms_is_member(attnum, grouped_cols))
                    slot->tts_isnull[attnum - 1] = true;
            }
        }
    }
}

finalize_aggregates
finalize_aggregates函數計算某一組所有聚合的最終值,實現函數是finalize_aggregate,該實現函數下節再行介紹.

/*
 * Compute the final value of all aggregates for one group.
 * 計算某一組所有聚合的最終值
 *
 * This function handles only one grouping set at a time, which the caller must
 * have selected.  It's also the caller's responsibility to adjust the supplied
 * pergroup parameter to point to the current set's transvalues.
 * 該函數一次只會處理一個grouping set(調用者負責選擇).
 * 調用者同樣有職責調整提供的pergroup參數為指向當前集合的transvalues.
 *
 * Results are stored in the output econtext aggvalues/aggnulls.
 */
static void
finalize_aggregates(AggState *aggstate,
                    AggStatePerAgg peraggs,
                    AggStatePerGroup pergroup)
{
    ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
    Datum       *aggvalues = econtext->ecxt_aggvalues;
    bool       *aggnulls = econtext->ecxt_aggnulls;
    int            aggno;
    int            transno;
    /*
     * If there were any DISTINCT and/or ORDER BY aggregates, sort their
     * inputs and run the transition functions.
     * 如存在DISTINCT或ORDER BY 聚合,排序這些輸入并執行轉換函數.
     */
    //遍歷轉換函數
    for (transno = 0; transno < aggstate->numtrans; transno++)
    {
        //轉換函數
        AggStatePerTrans pertrans = &aggstate->pertrans[transno];
        //pergroup
        AggStatePerGroup pergroupstate;
        pergroupstate = &pergroup[transno];
        if (pertrans->numSortCols > 0)
        {
            //--- 存在DISTINCT/ORDER BY
            //驗證,Hash不需要排序
            Assert(aggstate->aggstrategy != AGG_HASHED &&
                   aggstate->aggstrategy != AGG_MIXED);
            if (pertrans->numInputs == 1)
                //單獨
                process_ordered_aggregate_single(aggstate,
                                                 pertrans,
                                                 pergroupstate);
            else
                //多個
                process_ordered_aggregate_multi(aggstate,
                                                pertrans,
                                                pergroupstate);
        }
    }
    /*
     * Run the final functions.
     * 執行獲取最終值的函數
     */
    //遍歷聚合
    for (aggno = 0; aggno < aggstate->numaggs; aggno++)
    {
        //獲取peragg
        AggStatePerAgg peragg = &peraggs[aggno];
        int            transno = peragg->transno;
        AggStatePerGroup pergroupstate;
        //pergroup
        pergroupstate = &pergroup[transno];
        if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
            //并行處理結果
            finalize_partialaggregate(aggstate, peragg, pergroupstate,
                                      &aggvalues[aggno], &aggnulls[aggno]);
        else
            //調用finalize_aggregate獲取結果
            finalize_aggregate(aggstate, peragg, pergroupstate,
                               &aggvalues[aggno], &aggnulls[aggno]);
    }
}

project_aggregates
project_aggregates函數投影某一組的結果(該組結果已通過finalize_aggregates函數計算得到).

/*
 * Project the result of a group (whose aggs have already been calculated by
 * finalize_aggregates). Returns the result slot, or NULL if no row is
 * projected (suppressed by qual).
 * 投影某一組的結果(該組結果已通過finalize_aggregates函數計算得到).
 * 返回結果slot,如無結果行投影(通過qual處理)則返回NULL.
 */
static TupleTableSlot *
project_aggregates(AggState *aggstate)
{
    ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
    /*
     * Check the qual (HAVING clause); if the group does not match, ignore it.
     * 檢查條件表達式(HAVING子句).如跟group不匹配,則忽略之.
     */
    if (ExecQual(aggstate->ss.ps.qual, econtext))
    {
        /*
         * Form and return projection tuple using the aggregate results and
         * the representative input tuple.
         * 使用聚合結果和相應的輸入tuple組成并返回投影元組.
         */
        return ExecProject(aggstate->ss.ps.ps_ProjInfo);
    }
    else
        InstrCountFiltered1(aggstate, 1);
    return NULL;
}
#define InstrCountFiltered1(node, delta) \
    do { \
        if (((PlanState *)(node))->instrument) \
            ((PlanState *)(node))->instrument->nfiltered1 += (delta); \
    } while(0)

ExecProject
ExecProject函數基于投影信息投影元組并把元組存儲在傳遞給ExecBuildProjectInfo()的slot參數中.

/*
 * ExecProject
 *
 * Projects a tuple based on projection info and stores it in the slot passed
 * to ExecBuildProjectInfo().
 * 基于投影信息投影元組并把元組存儲在傳遞給ExecBuildProjectInfo()的slot參數中.
 *
 * Note: the result is always a virtual tuple; therefore it may reference
 * the contents of the exprContext's scan tuples and/or temporary results
 * constructed in the exprContext.  If the caller wishes the result to be
 * valid longer than that data will be valid, he must call ExecMaterializeSlot
 * on the result slot.
 * 注意:結果通常是虛擬元組.因此該元組可能會依賴exprContext掃描元組的內容和/或在exprContext中構建的臨時結果.
 * 如果調用者希望結果比數據更長久有效,調用者必須調用在結果slot上調用ExecMaterializeSlot(物化).
 */
#ifndef FRONTEND
static inline TupleTableSlot *
ExecProject(ProjectionInfo *projInfo)
{
    ExprContext *econtext = projInfo->pi_exprContext;
    ExprState  *state = &projInfo->pi_state;
    TupleTableSlot *slot = state->resultslot;
    bool        isnull;
    /*
     * Clear any former contents of the result slot.  This makes it safe for
     * us to use the slot's Datum/isnull arrays as workspace.
     * 清理結果slot的形式內容.
     * 這可以確保slot的Datum/isnull數組是OK的.
     */
    ExecClearTuple(slot);
    /* Run the expression, discarding scalar result from the last column. */
    //執行表達式解析,丟棄scalar結果.
    (void) ExecEvalExprSwitchContext(state, econtext, &isnull);
    /*
     * Successfully formed a result row.  Mark the result slot as containing a
     * valid virtual tuple (inlined version of ExecStoreVirtualTuple()).
     * 成功組成一個結果行.
     * 標記結果slot為包含有效虛擬元組(內聯版本的ExecStoreVirtualTuple)
     */
    slot->tts_isempty = false;
    slot->tts_nvalid = slot->tts_tupleDescriptor->natts;
    return slot;
}
#endif

三、跟蹤分析

N/A

四、參考資料

PostgreSQL 源碼解讀（178）- 查詢#95(聚合函數)#1相關數據結構