PostgreSQL 源码解读（194）- 查询#110(排序#3 - 实现)

本节继续介绍排序的实现,上一节介绍了ExecSort,本节介绍该函数中调用的排序的具体实现函数,本节是第一部分,包括tuplesort_begin_heap/tuplesort_puttupleslot.

一、数据结构

SortState
排序运行期状态信息

typedef struct SortState
{
    //基类
    ScanState    ss;                
    //是否需要随机访问排序输出?
    bool        randomAccess;    
    //结果集是否存在边界?
    bool        bounded;        
    //如存在边界,需要多少个元组?
    int64        bound;            
    //是否已完成排序?
    bool        sort_Done;        
    //是否使用有界值?
    bool        bounded_Done;    
    //使用的有界值?
    int64        bound_Done;        
    //tuplesort.c的私有状态
    void       *tuplesortstate; 
    //是否worker?
    bool        am_worker;        
    //每个worker对应一个条目
    SharedSortInfo *shared_info;    
} SortState;

typedef struct SharedSortInfo
{
    //worker个数?
    int            num_workers;
    //排序机制
    TuplesortInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER];
} SharedSortInfo;

TuplesortInstrumentation
报告排序统计的数据结构.

typedef enum
{
    SORT_TYPE_STILL_IN_PROGRESS = 0,//仍然在排序中
    SORT_TYPE_TOP_N_HEAPSORT,//TOP N 堆排序
    SORT_TYPE_QUICKSORT,//快速排序
    SORT_TYPE_EXTERNAL_SORT,//外部排序
    SORT_TYPE_EXTERNAL_MERGE//外部排序后的合并
} TuplesortMethod;//排序方法
typedef enum
{
    SORT_SPACE_TYPE_DISK,//需要用上磁盘
    SORT_SPACE_TYPE_MEMORY//使用内存
} TuplesortSpaceType;
typedef struct TuplesortInstrumentation
{
    //使用的排序算法
    TuplesortMethod sortMethod; 
    //排序使用空间类型
    TuplesortSpaceType spaceType;    
    //空间消耗(以K为单位)
    long        spaceUsed;        
} TuplesortInstrumentation;

二、源码解读

tuplesort_begin_heap和tuplesort_puttupleslot都是准备工作,把tuple放到数组(堆)中,为后续的实际排序实现作准备.
tuplesort_begin_heap

Tuplesortstate *
tuplesort_begin_heap(TupleDesc tupDesc,//元组描述符
                     int nkeys, //排序键个数
                     AttrNumber *attNums,//属性编号
                     Oid *sortOperators, //排序操作符
                     Oid *sortCollations,//排序规则
                     bool *nullsFirstFlags,//标记
                     int workMem, //内存大小
                     SortCoordinate coordinate,//协调器 
                     bool randomAccess)//是否随机访问
{
    //获取排序状态
    Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
                                                   randomAccess);
    MemoryContext oldcontext;
    int            i;
    oldcontext = MemoryContextSwitchTo(state->sortcontext);
    AssertArg(nkeys > 0);
#ifdef TRACE_SORT
    if (trace_sort)
        elog(LOG,
             "begin tuple sort: nkeys = %d, workMem = %d, randomAccess = %c",
             nkeys, workMem, randomAccess ? 't' : 'f');
#endif
    state->nKeys = nkeys;
    TRACE_POSTGRESQL_SORT_START(HEAP_SORT,
                                false,    
                                nkeys,
                                workMem,
                                randomAccess,
                                PARALLEL_SORT(state));
    //设置运行状态
    state->comparetup = comparetup_heap;
    state->copytup = copytup_heap;
    state->writetup = writetup_heap;
    state->readtup = readtup_heap;
    //假定不需要拷贝元组描述符
    state->tupDesc = tupDesc;    
    state->abbrevNext = 10;
    
    //为每一列准备SortSupport数据(分配内存空间)
    state->sortKeys = (SortSupport) palloc0(nkeys * sizeof(SortSupportData));
    for (i = 0; i < nkeys; i++)
    {
        //------- 遍历排序键
        //排序键
        SortSupport sortKey = state->sortKeys + i;
        AssertArg(attNums[i] != 0);
        AssertArg(sortOperators[i] != 0);
        //设置SortSupport
        sortKey->ssup_cxt = CurrentMemoryContext;
        sortKey->ssup_collation = sortCollations[i];
        sortKey->ssup_nulls_first = nullsFirstFlags[i];
        sortKey->ssup_attno = attNums[i];
        
        //缩写优化是否原则上可用?
        sortKey->abbreviate = (i == 0);
        //设置
        PrepareSortSupportFromOrderingOp(sortOperators[i], sortKey);
    }
    
    if (nkeys == 1 && !state->sortKeys->abbrev_converter)
        state->onlyKey = state->sortKeys;
    MemoryContextSwitchTo(oldcontext);
    return state;
}

tuplesort_puttupleslot
接收一个元组(一行)

void
tuplesort_puttupleslot(Tuplesortstate *state, TupleTableSlot *slot)
{
    MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
    SortTuple    stup;
    
    //#define COPYTUP(state,stup,tup) ((*(state)->copytup) (state, stup, tup))
    COPYTUP(state, &stup, (void *) slot);
    puttuple_common(state, &stup);
    MemoryContextSwitchTo(oldcontext);
}

static void
puttuple_common(Tuplesortstate *state, SortTuple *tuple)
{
    Assert(!LEADER(state));
    switch (state->status)
    {
        case TSS_INITIAL://初始化
            
            if (state->memtupcount >= state->memtupsize - 1)
            {
                (void) grow_memtuples(state);
                Assert(state->memtupcount < state->memtupsize);
            }
            state->memtuples[state->memtupcount++] = *tuple;//在数组中保存元组
            
            if (state->bounded &&
                (state->memtupcount > state->bound * 2 ||
                 (state->memtupcount > state->bound && LACKMEM(state))))
            {
#ifdef TRACE_SORT
                if (trace_sort)
                    elog(LOG, "switching to bounded heapsort at %d tuples: %s",
                         state->memtupcount,
                         pg_rusage_show(&state->ru_start));
#endif
                //切换至堆排序
                make_bounded_heap(state);
                return;
            }
            
            if (state->memtupcount < state->memtupsize && !LACKMEM(state))
                return;
            
            inittapes(state, true);
            
            dumptuples(state, false);
            break;
        case TSS_BOUNDED://有界堆排序
            
            if (COMPARETUP(state, tuple, &state->memtuples[0]) <= 0)
            {
                
                // 新元组 <= 堆顶,可以废弃
                free_sort_tuple(state, tuple);
                CHECK_FOR_INTERRUPTS();
            }
            else
            {
                
                //废弃堆顶,使用新元组替换
                free_sort_tuple(state, &state->memtuples[0]);
                tuplesort_heap_replace_top(state, tuple);
            }
            break;
        case TSS_BUILDRUNS://构建运行期信息
            
            state->memtuples[state->memtupcount++] = *tuple;
            
            dumptuples(state, false);
            break;
        default:
            elog(ERROR, "invalid tuplesort state");
            break;
    }
}

三、跟踪分析

N/A

四、参考资料

N/A