PostgreSQL源码修改 ——查询优化（三）

//qsort_arg.c

/*ouyang 12.11*/

@H_638_404@/*ouyang*/

char * select_s_k(char * * p,size_t k,voID *a,int64 select_offset);

char * findSmallN(char **p,voID *arg);

#define SELECTSK 5 @H_638_404@//5个为一组

#define SELECTSKMID 3 @H_638_404@//5的中间数为

/*ouyang 12.11*/

//何用select_s_k算法来实现的更复杂而优化的选择“部分排序”算法

voID

my_opt_select_sort(voID *a,int64 select_offset)

{

@H_638_404@/*the arrang of point*/

char ** p;@H_638_404@//指针数组,每个地址都指向待排序的N个项中的某个项.

size_t i;

int swaptype;

char *temp;

@H_638_404@// 检查参数的有效性

if(select_count<=0||

select_count+select_offset > n+1||

select_offset <0)

return;

SWAPINIT(a,es);

@H_638_404@/*指针数组初始化*/

p = (char **)malloc(n*sizeof(char *));

for(i=0;i<n;i++)

{

p[i]=(char*)a+es*i;

}

@H_638_404@/*循环调用select_s_k算法并,并将选择的第K个元素交换到相应的位置*/

for(i=select_offset;i<select_count+select_offset;i++)

{

temp = select_s_k(p,i+1,a,n,es,cmp,arg,select_count,select_offset);

if(temp!=(char *)a+es*(i))

{

swap(temp,(char *)a+es*(i));

}

}

free(p);

}

// select_s_k实现,采用算法课上的select_s_k分置算法

char * select_s_k(char * * p,int64 select_offset)

{

size_t i,j;

size_t t;

int swaptype;

char ** pSmaller;@H_638_404@//比m*小的项的指针数组.m*的意义请参见select_s_k算法思路及示意图.

char ** pLarger;@H_638_404@//比m*大的或等于的指针数组.

char ** pMID;@H_638_404@//按个一组分组后,存入每组的中位数.

char *pTemp;

size_t smallerInder,largerInder;@H_638_404@//pSmaller与pLarger指针的索引

smallerInder = largerInder=0;

if(n<=SELECTSK)

{

return findSmallN(p,k,arg);

}

SWAPINIT(a,es);

pSmaller = (char **)malloc(n*sizeof(char *));

pLarger = (char **)malloc(n*sizeof(char *));

pMID = (char **)malloc((n/SELECTSK+1)*sizeof(char *));

@H_638_404@/*devIDe with SELECTSK and find the mID one,and put it's address to pMID*/

for(i=0,j=0;i<n-SELECTSK;i+=SELECTSK,j++)

{

pMID[j]=findSmallN(&p[i],SELECTSK,SELECTSKMID,arg);

}

@H_638_404@// 递归调用select_s_k求pMID的中位数

pTemp = select_s_k( pMID,(j+1)/2,j,select_offset);

@H_638_404@// 下面的两个循环是将ABCD四个集体分配到pSmaller和pLarger两个集合中.

for(i=0;i<n-SELECTSK;i+=SELECTSK)

{

if(cmp(p[i+SELECTSKMID-1],pTemp,arg)>=0)

{

for(t=0;t<SELECTSKMID-1;t++)

{

if(cmp(p[t+i],arg)>=0)

{

pLarger[largerInder]=p[t+i];

largerInder++;

}

else

{

pSmaller[smallerInder]=p[t+i];

smallerInder++;

}

}

for(t=SELECTSKMID-1;t<SELECTSK;t++)

{

pLarger[largerInder]=p[t+i];

largerInder++;

}

}

else@H_638_404@/*cmp(p[i+SELECTSKMID-1],arg)<0*/

{

for(t=0;t<SELECTSKMID;t++)

{

pSmaller[smallerInder]=p[t+i];

smallerInder++;

}

for(t=SELECTSKMID;t<SELECTSK;t++)

{

if(cmp(p[t+i],arg)>=0)

{

pLarger[largerInder]=p[t+i];

largerInder++;

}

else

{

pSmaller[smallerInder]=p[t+i];

smallerInder++;

}

}

}

}

@H_638_404@// 处理没有除尽SELECTSK时,剩余的元素.

for(;i<n;)

{

if(cmp(p[i],arg)>=0)

{

pLarger[largerInder]=p[i];

largerInder++;

}

else

{

pSmaller[smallerInder]=p[i];

smallerInder++;

}

i++;

}

@H_638_404@// 如果恰好有smallerInder比pTemp小,则pTemp即为解.

if(k==(smallerInder+1))

{

}

@H_638_404@// 否则缩小子问题,递归调用求解.

else if(k<(smallerInder+1))

{

pTemp = select_s_k(pSmaller,smallerInder,select_offset);

}

else

{

if(smallerInder>0)

{

pTemp = select_s_k(pLarger,k-smallerInder,largerInder,select_offset);

}

else@H_638_404@//！！ouyang 12-15找到一个巨大的BUG，当smallerInder为时，即所选的pTemp刚好就是最小无元素之时，如果不加这个分支，会导致死循环！！！！

@H_638_404@//由于largerInder中包含和pTemp相等的元素，故不可能为,不用考虑。

{

for(i=0;i<largerInder;i++)

{

if(pLarger[i]==pTemp)

{

@H_638_404@//swap to position 0

pTemp= pLarger[0];

pLarger[0]=pLarger[i];

pLarger[i]=pTemp;

break;

}

}

pTemp = select_s_k(&pLarger[1],k-smallerInder-1,largerInder-1,select_offset);

}

}

@H_638_404@// 释放内存

free(pMID);

free(pSmaller);

free(pLarger);

return pTemp;

}

// 当n比较小时,直接调用此函数来找到第k小的数.

char * findSmallN(char **p,voID *arg)

{

size_t i,j;

char *temp;

for (i=1;i<n;i++) @H_638_404@//采用插入排序

{

for(j=i;j>0;j--)

{

if(cmp(p[j-1],p[j],arg) > 0)

{

temp=p[j-1];

p[j-1]=p[j];

p[j]=temp;

}

}

}

return p[k-1];@H_638_404@//返回k-1,是因为第k小,是从开始记数,而数组下标却从开始记数.

}

/*修改文件: tuplesort.c

nodeSort.c

*/

/*ouyang 12-12 copy from tuplesort.c*/

typedef struct

{

voID *tuple; @H_638_404@/* the tuple proper */

Datum datum1; @H_638_404@/* value of first key column */

bool isnull1; @H_638_404@/* is first key column NulL? */

int tupindex; @H_638_404@/* see notes above */

} SortTuple;

typedef enum

{

TSS_INITIAL, @H_638_404@/* Loading tuples; still within memory limit */

TSS_buildrUNS, @H_638_404@/* Loading tuples; writing to tape */

TSS_SORTEDINMEM, @H_638_404@/* Sort completed entirely in memory */

TSS_SORTEDONTAPE, @H_638_404@/* Sort completed,final run is on tape */

TSS_FINALMERGE @H_638_404@/* Performing final merge on-the-fly */

} TupSortStatus;

#define MInorDER 6 @H_638_404@/* minimum merge order */

#define TAPE_BUFFER_OVERhead (BLCKSZ * 3)

#define MERGE_BUFFER_SIZE (BLCKSZ * 32)

struct Tuplesortstate

{

TupSortStatus status; @H_638_404@/* enumerated value as shown above */

int nKeys; @H_638_404@/* number of columns in sort key */

bool randomAccess; @H_638_404@/* dID caller request random access? */

long availMem; @H_638_404@/* remaining memory available,in bytes */

long allowedMem; @H_638_404@/* total memory allowed,in bytes */

int maxTapes; @H_638_404@/* number of tapes (knuth's T) */

int tapeRange; @H_638_404@/* maxTapes-1 (knuth's P) */

MemoryContext sortcontext; @H_638_404@/* memory context holding all sort data */

@H_638_404@/*ouyang 12.10*/

//nodeSort.c

TupletableSlot *ExecSort(SortState *node)

{

EState *estate;

Tuplesortstate *tuplesortstate;

TupletableSlot *slot;

@H_638_404@//获取执行状态

estate = node->ss.ps.state;

tuplesortstate = (Tuplesortstate *) node->tuplesortstate;

@H_638_404@//如果还没有排序，则排序

if (!node->sort_Done)

{

tuplesortstate = tuplesort_begin_heap(tupDesc,node->randomAccess); @H_638_404@//分配空间

@H_638_404@//ouyang 12-12

tuplesortstate -> haslimit = node ->haslimit;

// tuplesort.c

static voID

beginmerge(Tuplesortstate *state)

{

@H_638_404@/*ouyang 12-12*/

int64 j=0;

int64 i;

int64 nSortedCount = state->sort_count+state->sort_offset;

　　@H_860_2404@……

@H_638_404@/*

* Preread as many tuples as possible (and at least one) from each active

* tape

*/

mergepreread(state);

@H_638_404@/* Load the merge heap with the first tuple from each input tape */

for (srcTape = 0; srcTape < state->maxTapes; srcTape++)

{

int tupIndex = state->mergenext[srcTape];

SortTuple *tup;

@H_638_404@/*

ouyang 12-12

每次尽量多的加入元组，如果每组都加入了（sort_count+sort_offset）个，则表示已经足够，算法结束。

*/

if(state -> haslimit &&

state -> sort_count>0&&

state -> sort_offset>=0 &&

(state -> sort_offset+state -> sort_count) < (state->memtupcount +1))

{

for(i=0;i<nSortedCount;i++)

{

tupIndex = state->mergenext[srcTape];

if (tupIndex)

{

tup = &state->memtuples[tupIndex];

state->mergenext[srcTape] = tup->tupindex;

if (state->mergenext[srcTape] == 0)

state->mergelast[srcTape] = 0;

tuplesort_heap_insert(state,tup,srcTape,false);

@H_638_404@/* put the Now-unused memtuples entry on the freeList */

tup->tupindex = state->mergefreeList;

state->mergefreeList = tupIndex;

state->mergeavailslots[srcTape]++;

@H_638_404@/*ouyang 12-12*/

j++;

}

else@H_638_404@//this tape is empty.

{

break;

}

}

}

else@H_638_404@//not modefIEd

{

if (tupIndex)

{

tup = &state->memtuples[tupIndex];

state->mergenext[srcTape] = tup->tupindex;

if (state->mergenext[srcTape] == 0)

state->mergelast[srcTape] = 0;

tuplesort_heap_insert(state,false);

@H_638_404@/* put the Now-unused memtuples entry on the freeList */

tup->tupindex = state->mergefreeList;

state->mergefreeList = tupIndex;

state->mergeavailslots[srcTape]++;

}

}

}

if(j>=nSortedCount)

{

state ->isEnough = true;

}

}

//nodeSort.c

#include <time.h>@H_638_404@//添加此头文件