PostgreSQL数据库查询执行之物化节点——临时tuple存储Tuplestore概述

今天药忘吃喽~ 2022-10-07 05:45 210阅读 0赞

![在这里插入图片描述][resize_m_lfit_w_962_pic_center]

Tuplestore模块处理元组的临时存储，用于Materialize节点、hashjoin批处理文件等。它本质上是tuplesort.c的简化版本；它不对元组进行排序，而只能存储和反刍(regurgitate)一系列元组。但是，因为不需要排序，所以允许在序列全部写入之前开始读取序列。这对游标特别有用，因为它允许在查询的已扫描部分内进行随机访问，而无需处理底层扫描直至完成。此外，还可以**支持多个独立的读指针**。  
如果数据超出调用者指定的空间限制，则使用临时文件来处理数据。元组存储允许的（近似）内存量由调用者以千字节(KB)为单位指定。我们取出元组并将它们简单地存储在内存数组中，只要我们没有超过 maxKBytes。如果我们确实超过了 maxKBytes，我们会将所有元组转储到一个临时文件中，然后在需要时从中读取。

创建时，元组存储支持单个读取指针，编号为 0。可以使用 tuplestore\_alloc\_read\_pointer 创建其他读取指针。复制读取指针支持标记/恢复行为。

当调用者请求向后扫描功能时，我们以允许向前或向后扫描的格式写入临时文件。否则，只允许向前扫描。在将任何元组放入 tuplestore 之前，必须提出反向扫描请求。倒带(Rewind)通常是允许的，但可以通过 tuplestore\_set\_eflags 关闭；关闭所有读取指针的倒带(Rewind)可以在最旧的读取点截断元组存储，以最大限度地减少内存使用。 （调用者必须在适当的时间显式调用 tuplestore\_trim 才能真正发生截断。）

注意：在 TSS\_WRITEFILE 状态下，临时文件的查找位置是当前写入位置，并且元组存储中的写入位置变量不会保持最新。类似地，在 TSS\_READFILE 状态下，临时文件的查找位置是活动读取指针的位置，并且该读取指针不会保持最新。在切换到其他状态或激活不同的读取指针之前，我们使用 ftell() 更新适当的变量。

## Tuplestore对象的状态 ##

src/backend/utils/sort/tuplestore.c  
TSS\_INMEM指示Tuplestore对象所拥有的内存还能存储tuple，TSS\_WRITEFILE指示将tuple写入到临时文件中，TSS\_READFILE从临时文件中读取tuple。

typedef enum { 
    	TSS_INMEM,					/* Tuples still fit in memory */
    	TSS_WRITEFILE,				/* Writing to temp file */
    	TSS_READFILE				/* Reading from temp file */
    } TupStoreStatus;

从tuplestore\_begin\_heap函数的处理逻辑可以看到eflags的取值，如下。如果形参randomAccess为true，则代表支持随机访问，也就是需要设置eflags为EXEC\_FLAG\_BACKWARD | EXEC\_FLAG\_REWIND，如果为false，不支持随机访问，设置eflags为EXEC\_FLAG\_REWIND。

int   eflags; // EXEC_FLAG_BACKWARD EXEC_FLAG_REWIND EXEC_FLAG_REWIND
    // Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
    	eflags = randomAccess ? (EXEC_FLAG_BACKWARD | EXEC_FLAG_REWIND) : (EXEC_FLAG_REWIND);

正如前面概述部分所述：当调用者请求向后扫描功能时，我们以允许向前或向后扫描的格式写入临时文件。否则，只允许向前扫描。在将任何元组放入 tuplestore 之前，必须提出反向扫描请求。倒带(Rewind)通常是允许的，但可以通过 tuplestore\_set\_eflags 关闭；关闭所有读取指针的倒带(Rewind)可以在最旧的读取点截断元组存储，以最大限度地减少内存使用。 （调用者必须在适当的时间显式调用 tuplestore\_trim 才能真正发生截断。）

Tuplestorestate主要分为四个部分，一是内存区域memtuples，二是临时文件的句柄myfile，三是读指针，四是写指针。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]

临时tuple存储Tuplestore读出写入元组的操作函数：

*  copytup，这些函数指针将必须知道我们正在处理的元组类型的例程与不需要知道它的例程分离。 它们由 tuplestore\_begin\_xxx 例程设置。（虽然 tuplestore.c 目前仅支持堆元组，但我已经复制了 tuplesort.c 的这一部分，以便在需要时可以轻松扩展到其他类型的对象。）将**提供的输入元组复制到 palloc 空间的函数**。 （注意：我们假设单个 pfree() 足以稍后释放元组，因此表示必须在一个 palloc 块中“平坦”。） state->availMem 必须减少使用的空间量。
 *  writetup，将存储的元组写入磁带的函数。 磁带上元组的表示不必与内存中的相同； 下面给出了对磁带表示的要求。 写入元组后， pfree() 它，并将 state->availMem 增加由此释放的内存空间量。Function to write a stored tuple onto tape. The representation of the tuple on tape need not be the same as it is in memory; requirements on the tape representation are given below. After writing the tuple, pfree() it, and increase state->availMem by the amount of memory space thereby released.
 *  将存储的元组从磁带读回内存的函数。 ‘len’ 是已存储元组的已读取长度。 创建并返回一个 palloc 的副本，并根据消耗的内存空间量减少 state->availMem。Function to read a stored tuple from tape back into memory. ‘len’ is the already-read length of the stored tuple. Create and return a palloc’d copy, and decrease state->availMem by the amount of memory space consumed.

struct Tuplestorestate
    { 
    	TupStoreStatus status;		
    	int			eflags;			
    	bool		backward;		/* store extra length words in file? */
    	bool		interXact;		/* keep open through transactions? */
    	bool		truncated;		/* tuplestore_trim has removed tuples? */
    	
    	int64		availMem;		/* 内存中剩余空间，以字节为单位 */
    	int64		allowedMem;		/* 内存中大小，以字节为单位 */
    	int64		tuples;			/* 添加的tuples的数量 */
    	
    	BufFile    *myfile;			
    	MemoryContext context;		/* memory context for holding tuples */
    	ResourceOwner resowner;		/* resowner for holding temp files */
    	
    	void	  **memtuples;		/* 存放tuples的内存 */
    	int			memtupdeleted;	/* 目前未使用的第N个slot*/
    	int			memtupcount;	/* 目前tuples的数量*/
    	int			memtupsize;		/* 已分配的内存长度*/
    	bool		growmemtuples;	/* memtuples的增长仍在进行中？ */
    
    
    	TSReadPointer *readptrs;	/* array of read pointers */
    	int			activeptr;		/* index of the active read pointer */
    	int			readptrcount;	/* number of pointers currently valid */
    	int			readptrsize;	/* allocated length of readptrs array */
    	int			writepos_file;	/* file# (valid if READFILE state) */
    	off_t		writepos_offset;	/* offset (valid if READFILE state) */
    
    	void	   *(*copytup) (Tuplestorestate *state, void *tup);
    	void		(*writetup) (Tuplestorestate *state, void *tup);
    	void	   *(*readtup) (Tuplestorestate *state, unsigned int len);
    };

## 构造和析构函数 ##

tuplestore\_begin\_heap函数创建一个新的元组存储Tuplestore； 其他类型的元组存储（除了“堆”元组存储）也是可能的，但目前尚未实现。

*  randomAccess：如果为 true，则允许向前和向后访问元组存储。
 *  interXact：如果为 true，则用于磁盘存储的文件会在当前事务结束后持续存在。 注意：调用者有责任在内存上下文和资源所有者中创建这样一个元组存储，它也将在事务边界中存活，并确保在不再需要元组存储时将其关闭。
 *  maxKBytes：要在内存中存储多少数据（超出此数量的任何数据都将分页到磁盘）。 如有疑问，请使用 work\_mem。

Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes) { 
    	Tuplestorestate *state;
    	int			eflags;
    	/* This interpretation of the meaning of randomAccess is compatible with the pre-8.3 behavior of tuplestores. */
    	eflags = randomAccess ?(EXEC_FLAG_BACKWARD | EXEC_FLAG_REWIND) :(EXEC_FLAG_REWIND);
    	state = tuplestore_begin_common(eflags, interXact, maxKBytes);
    	state->copytup = copytup_heap;
    	state->writetup = writetup_heap;
    	state->readtup = readtup_heap;
    	return state;
    }

tuplestore\_begin\_common函数为Tuplestorestate结构体分配内存，将Tuplestore对象的状态设为TSS\_INMEM，说明Tuplestore对象所拥有的内存还能存储tuple；设置内存大小为maxKBytes \* 1024L比特；设置放置Tuple指针的数组的初始大小为`Max(16384 / sizeof(void *), ALLOCSET_SEPARATE_THRESHOLD / sizeof(void *) + 1)`；设置读指针数组。

static Tuplestorestate * tuplestore_begin_common(int eflags, bool interXact, int maxKBytes) { 
    	Tuplestorestate *state;
    	state = (Tuplestorestate *) palloc0(sizeof(Tuplestorestate));
    	state->status = TSS_INMEM;
    	state->eflags = eflags;
    	state->interXact = interXact;
    	state->truncated = false;
    	state->allowedMem = maxKBytes * 1024L;
    	state->availMem = state->allowedMem;
    	state->myfile = NULL;
    	state->context = CurrentMemoryContext;
    	state->resowner = CurrentResourceOwner;
    	state->memtupdeleted = 0;
    	state->memtupcount = 0;
    	state->tuples = 0;
    
    	/* Initial size of array must be more than ALLOCSET_SEPARATE_THRESHOLD; see comments in grow_memtuples(). */
    	state->memtupsize = Max(16384 / sizeof(void *), ALLOCSET_SEPARATE_THRESHOLD / sizeof(void *) + 1);
    	state->growmemtuples = true;
    	state->memtuples = (void **) palloc(state->memtupsize * sizeof(void *));
    	USEMEM(state, GetMemoryChunkSpace(state->memtuples));
    
    	state->activeptr = 0;
    	state->readptrcount = 1;
    	state->readptrsize = 8;		/* arbitrary */
    	state->readptrs = (TSReadPointer *)palloc(state->readptrsize * sizeof(TSReadPointer));
    	state->readptrs[0].eflags = eflags;
    	state->readptrs[0].eof_reached = false;
    	state->readptrs[0].current = 0;
    	return state;
    }

tuplestore\_end函数释放并清除元组存储Tuplestore

void tuplestore_end(Tuplestorestate *state) { 
    	int			i;
    	if (state->myfile)
    		BufFileClose(state->myfile);
    	if (state->memtuples) { 
    		for (i = state->memtupdeleted; i < state->memtupcount; i++)
    			pfree(state->memtuples[i]);
    		pfree(state->memtuples);
    	}
    	pfree(state->readptrs);
    	pfree(state);
    }

## 读出写入元组的操作函数 ##

copytup\_heap函数从形参tup指向的HeapTuple结构体的t\_data拷贝数据创建MinimalTuple，并返回。The stored form is actually a MinimalTuple, but for largely historical reasons we allow COPYTUP to work from a HeapTuple。

static void * copytup_heap(Tuplestorestate *state, void *tup)
    { 
    	MinimalTuple tuple;
    	tuple = minimal_tuple_from_heap_tuple((HeapTuple) tup);
    	USEMEM(state, GetMemoryChunkSpace(tuple));
    	return (void *) tuple;
    }
    
    // minimal_tuple_from_heap_tuple函数在heaptuple.c中，用于从HeapTuple拷贝数据创建MinimalTuple
    MinimalTuple minimal_tuple_from_heap_tuple(HeapTuple htup) { 
    	MinimalTuple result;
    	uint32		len;
    	Assert(htup->t_len > MINIMAL_TUPLE_OFFSET);
    	len = htup->t_len - MINIMAL_TUPLE_OFFSET;
    	result = (MinimalTuple) palloc(len);
    	memcpy(result, (char *) htup->t_data + MINIMAL_TUPLE_OFFSET, len);
    	result->t_len = len;
    	return result;
    }

### 从myfile读取写入tuple ###

writetup\_heap函数从形参tup指向的MinimalTuple结构体向state->myfile中写入tuplen和tupbody，如果支持反向，则需要再次写入tuplen。

static void writetup_heap(Tuplestorestate *state, void *tup) { 
    	MinimalTuple tuple = (MinimalTuple) tup;
    	/* the part of the MinimalTuple we'll write: */
    	char	   *tupbody = (char *) tuple + MINIMAL_TUPLE_DATA_OFFSET;
    	unsigned int tupbodylen = tuple->t_len - MINIMAL_TUPLE_DATA_OFFSET;
    	/* total on-disk footprint: */
    	unsigned int tuplen = tupbodylen + sizeof(int);
    	BufFileWrite(state->myfile, (void *) &tuplen, sizeof(tuplen));
    	BufFileWrite(state->myfile, (void *) tupbody, tupbodylen);
    	if (state->backward)		/* need trailing length word? */
    		BufFileWrite(state->myfile, (void *) &tuplen, sizeof(tuplen));
    	FREEMEM(state, GetMemoryChunkSpace(tuple));
    	heap_free_minimal_tuple(tuple);
    }

readtup\_heap函数从state->myfile中读取tuple，并使用数据填充MinimalTuple，返回。

static void * readtup_heap(Tuplestorestate *state, unsigned int len)
    { 
    	unsigned int tupbodylen = len - sizeof(int);
    	unsigned int tuplen = tupbodylen + MINIMAL_TUPLE_DATA_OFFSET;
    	MinimalTuple tuple = (MinimalTuple) palloc(tuplen);
    	char	   *tupbody = (char *) tuple + MINIMAL_TUPLE_DATA_OFFSET;
    	size_t		nread;
    	USEMEM(state, GetMemoryChunkSpace(tuple));
    	/* read in the tuple proper */
    	tuple->t_len = tuplen;
    	nread = BufFileRead(state->myfile, (void *) tupbody, tupbodylen);
    	if (nread != (size_t) tupbodylen)
    		ereport(ERROR,(errcode_for_file_access(),errmsg("could not read from tuplestore temporary file: read only %zu of %zu bytes", nread, (size_t) tupbodylen)));
    	if (state->backward)		/* need trailing length word? */{ 
    		nread = BufFileRead(state->myfile, (void *) &tuplen, sizeof(tuplen));
    		if (nread != sizeof(tuplen))
    			ereport(ERROR,(errcode_for_file_access(),errmsg("could not read from tuplestore temporary file: read only %zu of %zu bytes",nread, sizeof(tuplen))));
    	}
    	return (void *) tuple;
    }

### 存储Tuplestore读取写入tuple ###

static void tuplestore\_puttuple\_common(Tuplestorestate \*state, void \*tuple)函数首先将tuples（添加的tuples的数量）增加，然后根据state->status判别tuple存放在何处。比如tuple可存放在内存中，首先更新读指针，按需增加内存slot，将tuple指针放入内存slot中，然后判断内存slot总量是否足够，足够就返回，不够就要提前创建tmp文件，如下代码所示。如果内存不够，需要存放到tmp文件中，需要更新读指针，然后调用WRITETUP函数写入。如果是处于TSS\_READFILE模式下，需要先将reading转为writing，然后更新读指针，调用WRITETUP函数写入。

static void tuplestore_puttuple_common(Tuplestorestate *state, void *tuple) { 
    	TSReadPointer *readptr;
    	int			i;
    	ResourceOwner oldowner;
    	state->tuples++;
    	switch (state->status) { 
    		case TSS_INMEM:
    		    /* Update read pointers as needed; see API spec above. */
    			readptr = state->readptrs;
    			for (i = 0; i < state->readptrcount; readptr++, i++){ 
    				if (readptr->eof_reached && i != state->activeptr){ 
    					readptr->eof_reached = false;
    					readptr->current = state->memtupcount;
    				}
    			}
    			/* Grow the array as needed. */
    			if (state->memtupcount >= state->memtupsize - 1){ 
    				(void) grow_memtuples(state);
    				Assert(state->memtupcount < state->memtupsize);
    			}
    			/* Stash the tuple in the in-memory array */
    			state->memtuples[state->memtupcount++] = tuple;	
    			/* * Done if we still fit in available memory and have array slots. */
    			if (state->memtupcount < state->memtupsize && !LACKMEM(state))
    				return;	
    								
    			/* Nope; time to switch to tape-based operation. Make sure that the temp file(s) are created in suitable temp tablespaces. */
    			PrepareTempTablespaces();
    			/* associate the file with the store's resource owner */
    			oldowner = CurrentResourceOwner;
    			CurrentResourceOwner = state->resowner;
    			state->myfile = BufFileCreateTemp(state->interXact);
    			CurrentResourceOwner = oldowner;
    			/* Freeze the decision about whether trailing length words will be used. We can't change this choice once data is on tape, even though callers might drop the requirement. */
    			state->backward = (state->eflags & EXEC_FLAG_BACKWARD) != 0;
    			state->status = TSS_WRITEFILE;
    			dumptuples(state);
    			break;					    
    		case TSS_WRITEFILE:
    			readptr = state->readptrs;
    			for (i = 0; i < state->readptrcount; readptr++, i++){ 
    				if (readptr->eof_reached && i != state->activeptr){ 
    					readptr->eof_reached = false;
    					BufFileTell(state->myfile,&readptr->file,&readptr->offset);
    				}
    			}
    			WRITETUP(state, tuple);
    			break;		    
    		case TSS_READFILE:
    			/* Switch from reading to writing. */
    			if (!state->readptrs[state->activeptr].eof_reached)
    				BufFileTell(state->myfile,&state->readptrs[state->activeptr].file,&state->readptrs[state->activeptr].offset);
    			if (BufFileSeek(state->myfile,state->writepos_file, state->writepos_offset,SEEK_SET) != 0)
    				ereport(ERROR,(errcode_for_file_access(), errmsg("could not seek in tuplestore temporary file")));
    			state->status = TSS_WRITEFILE;
    			/* Update read pointers as needed; see API spec above. */
    			readptr = state->readptrs;
    			for (i = 0; i < state->readptrcount; readptr++, i++){ 
    				if (readptr->eof_reached && i != state->activeptr){ 
    					readptr->eof_reached = false;
    					readptr->file = state->writepos_file;
    					readptr->offset = state->writepos_offset;
    				}
    			}
    			WRITETUP(state, tuple);
    			break;
    		default:
    			elog(ERROR, "invalid tuplestore state");
    			break;
    	}
    }

下面3个函数就是调用上述函数进行写入操作的定制函数。

void tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple) //通过调用COPYTUP函数拷贝一个HeapTuple，然后将该tuple存入
    void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull) //通过传入的tdesc、values、isnull等参数构造HeapTuple，然后将该tuple存入
    void tuplestore_puttupleslot(Tuplestorestate *state, TupleTableSlot *slot) //从slot上拷贝一个MinimalTuple，然后将该tuple存入

static void \* tuplestore\_gettuple(Tuplestorestate \*state, bool forward, bool \*should\_free)  
bool tuplestore\_gettupleslot(Tuplestorestate \*state, bool forward, bool copy, TupleTableSlot \*slot)

int tuplestore\_alloc\_read\_pointer(Tuplestorestate \*state, int eflags)  
void tuplestore\_select\_read\_pointer(Tuplestorestate \*state, int ptr)  
void tuplestore\_copy\_read\_pointer(Tuplestorestate \*state, int srcptr, int destptr)

static void dumptuples(Tuplestorestate \*state)  
static bool grow\_memtuples(Tuplestorestate \*state)  
static unsigned int getlen(Tuplestorestate \*state, bool eofOK)  
bool tuplestore\_in\_memory(Tuplestorestate \*state)  
int64 tuplestore\_tuple\_count(Tuplestorestate \*state)  
bool tuplestore\_ateof(Tuplestorestate \*state)  
bool tuplestore\_advance(Tuplestorestate \*state, bool forward)  
bool tuplestore\_skiptuples(Tuplestorestate \*state, int64 ntuples, bool forward)  
void tuplestore\_rescan(Tuplestorestate \*state)  
void tuplestore\_trim(Tuplestorestate \*state)

void tuplestore\_set\_eflags(Tuplestorestate \*state, int eflags)

[resize_m_lfit_w_962_pic_center]: /images/20221005/b575539d89bc4351926360ffb9393589.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/69cd9ae033fe4d5f99b69293c0ffee93.png