PostgreSQL数据库查询执行——PORTAL_ONE_SELECT策略一

太过爱你忘了你带给我的痛 2022-10-17 13:42 179阅读 0赞

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

## 最外层查询执行流程 ##

查询执行的流程函数：

1.  portal = CreatePortal("", true, true) 创建未命名的portal用于运行查询，如果已经有了，就丢掉
2.  portal->visible = false Don’t display the portal in pg\_cursors
3.  PortalDefineQuery(portal, NULL, query\_string, commandTag, plantree\_list, NULL)
4.  PortalStart(portal, NULL, 0, InvalidSnapshot)
5.  PortalSetResultFormat(portal, 1, &format)
6.  receiver = CreateDestReceiver(dest)
7.  SetRemoteDestReceiverParams(receiver, portal)
8.  (void) PortalRun(portal, FETCH\_ALL, true, true, receiver, receiver, completionTag)
9.  receiver->rDestroy(receiver)
10. PortalDrop(portal, false)

/* Run through the raw parsetree(s) and process each one. */
    	foreach(parsetree_item, parsetree_list) { 
            RawStmt    *parsetree = lfirst_node(RawStmt, parsetree_item);
            ....
      		oldcontext = MemoryContextSwitchTo(MessageContext);
    		querytree_list = pg_analyze_and_rewrite(parsetree, query_string, NULL, 0, NULL);
    		plantree_list = pg_plan_queries(querytree_list, CURSOR_OPT_PARALLEL_OK, NULL);   
    		....
    		portal = CreatePortal("", true, true);
    		portal->visible = false;
    		/* We don't have to copy anything into the portal, because everything * we are passing here is in MessageContext, which will outlive the * portal anyway. */
    		PortalDefineQuery(portal, NULL, query_string, commandTag, plantree_list, NULL);
    		/* Start the portal. No parameters here. */
    		PortalStart(portal, NULL, 0, InvalidSnapshot);	
    		/* Select the appropriate output format: text unless we are doing a * FETCH from a binary cursor. (Pretty grotty to have to do this here * --- but it avoids grottiness in other places. Ah, the joys of * backward compatibility...) */
    		format = 0;				/* TEXT is default */
    		if (IsA(parsetree->stmt, FetchStmt)){ 
    			FetchStmt  *stmt = (FetchStmt *) parsetree->stmt;
    			if (!stmt->ismove){ 
    				Portal		fportal = GetPortalByName(stmt->portalname);
    				if (PortalIsValid(fportal) &&(fportal->cursorOptions & CURSOR_OPT_BINARY))
    					format = 1; /* BINARY */
    			}
    		}
    		PortalSetResultFormat(portal, 1, &format);
    		/* Now we can create the destination receiver object. */
    		receiver = CreateDestReceiver(dest);
    		if (dest == DestRemote)
    			SetRemoteDestReceiverParams(receiver, portal);
    		/* Switch back to transaction context for execution. */
    		MemoryContextSwitchTo(oldcontext);
    		/* Run the portal to completion, and then drop it (and the receiver). */
    		(void) PortalRun(portal, FETCH_ALL, true,	/* always top level */ true, receiver, receiver, completionTag);
    		receiver->rDestroy(receiver);
    		PortalDrop(portal, false);	
    		....		   
        }

### CreatePortal和PortalDefineQuery函数 ###

CreatePortal函数返回以name命名的新portal，如果allowDup为true，自动drop任何已经存在的名称和name相同的portal，如果为false，将会raise一个错误；如果dupSilent为true，drop portal时甚至连警告都不会发出。函数流程为：先查看是否有同名的portal，有就drop掉，然后直接调用MemoryContextAllocZero创建portal结构，为portal创建resource owner，之后初始化一些参数。

Portal CreatePortal(const char *name, bool allowDup, bool dupSilent) { 
    	Portal		portal;
    	AssertArg(PointerIsValid(name));
    	portal = GetPortalByName(name);
    	if (PortalIsValid(portal)) { 
    		if (!allowDup)
    			ereport(ERROR,(errcode(ERRCODE_DUPLICATE_CURSOR), errmsg("cursor \"%s\" already exists", name)));
    		if (!dupSilent)
    			ereport(WARNING,(errcode(ERRCODE_DUPLICATE_CURSOR), errmsg("closing existing cursor \"%s\"", name)));
    		PortalDrop(portal, false);
    	}
    	/* make new portal structure */
    	portal = (Portal) MemoryContextAllocZero(TopPortalContext, sizeof *portal);
    	/* initialize portal context; typically it won't store much */
    	portal->portalContext = AllocSetContextCreate(TopPortalContext, "PortalContext", ALLOCSET_SMALL_SIZES);
    	/* create a resource owner for the portal */
    	portal->resowner = ResourceOwnerCreate(CurTransactionResourceOwner, "Portal");
    	/* initialize portal fields that don't start off zero */
    	portal->status = PORTAL_NEW;
    	portal->cleanup = PortalCleanup;
    	portal->createSubid = GetCurrentSubTransactionId();
    	portal->activeSubid = portal->createSubid;
    	portal->strategy = PORTAL_MULTI_QUERY;
    	portal->cursorOptions = CURSOR_OPT_NO_SCROLL;
    	portal->atStart = true;
    	portal->atEnd = true;		/* disallow fetches until query is set */
    	portal->visible = true;
    	portal->creation_time = GetCurrentStatementStartTimestamp();
    	/* put portal in table (sets portal->name) */
    	PortalHashTableInsert(portal, name);
    	/* reuse portal->name copy */
    	MemoryContextSetIdentifier(portal->portalContext, portal->name);
    	return portal;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]  
PortalDefineQuery函数（PortalDefineQuery(portal, NULL, query\_string, commandTag, plantree\_list, NULL);）用于建立portal查询。  
注意：从 PG 8.4 开始，调用者必须提供一个 sourceText 字符串；不再允许传递 NULL。 （如果你真的没有源文本，你可以传递一个常量字符串，也许是“（查询不可用）”。）当且仅当原始查询字符串（重写前）为空字符串时，commandTag 应为 NULL。此外，传递的 commandTag 必须是指向常量字符串的指针，因为它不会被复制。  
如果提供了 cplan，那么它是一个包含 stmts 的缓存计划，并且调用者必须已经完成了 GetCachedPlan()，导致引用计数增加。当传送门被摧毁时，引用计数将被释放。  
如果 cplan 为 NULL，则调用者有责任确保传递的计划树具有足够的生命周期。通常这是通过将它们复制到门户的上下文来完成的。  
调用者还负责确保传递的 prepStmtName（如果不是 NULL）和 sourceText 具有足够的生命周期。  
注意：除了存储传递的值之外，此函数不能做太多事情；特别是不要做任何有elog（ERROR）风险的事情。如果在存储 cplan 引用之前发生这种情况，我们将泄漏调用者试图传递给我们的 plancache 引用计数。

void PortalDefineQuery(Portal portal, const char *prepStmtName, const char *sourceText, const char *commandTag, List *stmts, CachedPlan *cplan) { 
    	AssertArg(PortalIsValid(portal));
    	AssertState(portal->status == PORTAL_NEW);
    	AssertArg(sourceText != NULL);
    	AssertArg(commandTag != NULL || stmts == NIL);
    	portal->prepStmtName = prepStmtName;
    	portal->sourceText = sourceText;
    	portal->commandTag = commandTag;
    	portal->stmts = stmts;
    	portal->cplan = cplan;
    	portal->status = PORTAL_DEFINED;
    }

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

## Portal执行过程 ##

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

### PortalStart函数 ###

PortalStart函数（PortalStart(portal, NULL, 0, InvalidSnapshot)），调用者必须已经创建了portal，完成PortalDefineQuery()，并在需要时调整portal选项。如果查询需要参数，它们必须在“params”中传递（调用者负责为它们提供适当的生命周期）。调用者还可以提供一组初始“eflags”以传递给 ExecutorStart（但请注意，这些可以在内部进行修改，目前仅适用于 PORTAL\_ONE\_SELECT portal）。 大多数调用者应该简单地传递零。调用者可以选择传递要使用的快照； 为设置新快照的正常行为传递 InvalidSnapshot。 对于非 PORTAL\_ONE\_SELECT 门户，此参数目前被忽略（它仅用于游标）。  
返回时，portal已准备好接受 PortalRun() 调用，并且结果 tupdesc（如果有）是已知的。

解析ChoosePortalStrategy(portal->stmts)函数的逻辑如下所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 3]

这里我们只看PORTAL\_ONE\_SELECT策略：

*  先将ActivePortal、CurrentResourceOwner、PortalContext进行保存，然后将ActivePortal切换到形参给定的portal，CurrentResourceOwner切换为形参portal->resowner，PortalContext切换为形参portal->portalContext。并切换内存上下文。
 *  设置portal参数列表，即portalParams成员
 *  通过portal->stmts计划树判定执行策略，这里只看PORTAL\_ONE\_SELECT策略
 *  如果形参snapshot不为NULL，则PushActiveSnapshot(snapshot)，否则PushActiveSnapshot(GetTransactionSnapshot())。
 *  创建QueryDesc
 *  如果portal->cursorOptions指定了CURSOR\_OPT\_SCROLL，也就是scrollable游标，执行器需要支持向前向后双向扫描，应该设置myeflags = eflags | EXEC\_FLAG\_REWIND | EXEC\_FLAG\_BACKWARD。
 *  调用ExecutorStart(queryDesc, myeflags)为执行准备plan
 *  将创建的QueryDesc赋值给portal->queryDesc，设置portal->tupDesc = queryDesc->tupDesc
 *  将游标位置设置为查询开始
 *  PopActiveSnapshot()
 *  恢复内存上下文、ActivePortal、CurrentResourceOwner、PortalContext

void PortalStart(Portal portal, ParamListInfo params, int eflags, Snapshot snapshot) { 
    	Portal		saveActivePortal;
    	ResourceOwner saveResourceOwner;
    	MemoryContext savePortalContext;
    	MemoryContext oldContext;
    	QueryDesc  *queryDesc;
    	int			myeflags;
    	AssertArg(PortalIsValid(portal));
    	AssertState(portal->status == PORTAL_DEFINED);
    	/* Set up global portal context pointers. */
    	saveActivePortal = ActivePortal;
    	saveResourceOwner = CurrentResourceOwner;
    	savePortalContext = PortalContext;
    	PG_TRY();
    	{ 
    		ActivePortal = portal;
    		if (portal->resowner)
    			CurrentResourceOwner = portal->resowner;
    		PortalContext = portal->portalContext;
    		oldContext = MemoryContextSwitchTo(PortalContext);
    		/* Must remember portal param list, if any */
    		portal->portalParams = params;
    		/* Determine the portal execution strategy */
    		portal->strategy = ChoosePortalStrategy(portal->stmts);
    		/* Fire her up according to the strategy */
    		switch (portal->strategy) { 
    			case PORTAL_ONE_SELECT:
    				/* Must set snapshot before starting executor. */
    				if (snapshot)
    					PushActiveSnapshot(snapshot);
    				else
    					PushActiveSnapshot(GetTransactionSnapshot());
    				/* Create QueryDesc in portal's context; for the moment, set the destination to DestNone. */
    				queryDesc = CreateQueryDesc(linitial_node(PlannedStmt, portal->stmts), portal->sourceText, GetActiveSnapshot(), InvalidSnapshot, None_Receiver, params, portal->queryEnv, 0);
    				/* If it's a scrollable cursor, executor needs to support REWIND and backwards scan, as well as whatever the caller might've asked for. */
    				if (portal->cursorOptions & CURSOR_OPT_SCROLL)
    					myeflags = eflags | EXEC_FLAG_REWIND | EXEC_FLAG_BACKWARD;
    				else
    					myeflags = eflags;
    				/* Call ExecutorStart to prepare the plan for execution */
    				ExecutorStart(queryDesc, myeflags);
    				/* This tells PortalCleanup to shut down the executor */
    				portal->queryDesc = queryDesc;
    				/* Remember tuple descriptor (computed by ExecutorStart) */
    				portal->tupDesc = queryDesc->tupDesc;
    				/* Reset cursor position data to "start of query" */
    				portal->atStart = true;
    				portal->atEnd = false;	/* allow fetches */
    				portal->portalPos = 0;
    				PopActiveSnapshot();
    				break;          
    			case PORTAL_ONE_RETURNING:
    			case PORTAL_ONE_MOD_WITH:
                    ...   
    			case PORTAL_UTIL_SELECT:
                    ...
    			case PORTAL_MULTI_QUERY:
                    ...
    		}
    	}
    	PG_CATCH();
    	{ 
            ...
    	}
    	PG_END_TRY();
    	MemoryContextSwitchTo(oldContext);
    	ActivePortal = saveActivePortal;
    	CurrentResourceOwner = saveResourceOwner;
    	PortalContext = savePortalContext;
    	portal->status = PORTAL_READY;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]  
这里要解析Portal执行过程图中的第一个红框中涉及的两个函数：CreateQueryDesc(linitial\_node(PlannedStmt, portal->stmts), portal->sourceText, GetActiveSnapshot(), InvalidSnapshot, None\_Receiver, params, portal->queryEnv, 0)和ExecutorStart(queryDesc, myeflags)。  
先看CreateQueryDesc函数的参数，第一个是linitial\_node(PlannedStmt, portal->stmts)，PlannedStmt数据结构参见https://feishujun.blog.csdn.net/article/details/117388099，linitial\_node宏定义为`#define linitial_node(type,l) castNode(type, linitial(l))`，linitial宏定义为`#define linitial(l) lfirst(list_head(l))`，那么linitial\_node(PlannedStmt, portal->stmts)转换为`castNode(PlannedStmt, linitial(portal->stmts))`–>`PlannedStmt * linitial(portal->stmts)`–>`PlannedStmt * lfirst(list_head(portal->stmts))`–>`PlannedStmt * lfirst(portal->stmts->head)`–>`(PlannedStmt *) portal->stmts->head->data.ptr_value`最终转换为`(PlannedStmt *) portal->plantree_list->head->data.ptr_value`，这里的portal->stmts等同于portal->plantree\_list。  
CreateQueryDesc(linitial\_node(PlannedStmt, portal->stmts), portal->sourceText, GetActiveSnapshot(), InvalidSnapshot, None\_Receiver, params, portal->queryEnv, 0)的各个形参的值为：

*  `(PlannedStmt *) portal->plantree_list->head->data.ptr_value`
 *  portal->sourceText --> query\_string
 *  GetActiveSnapshot()
 *  InvalidSnapshot
 *  None\_Receiver
 *  NULL
 *  portal->queryEnv
 *  0

QueryDesc *CreateQueryDesc(PlannedStmt *plannedstmt, const char *sourceText, Snapshot snapshot, Snapshot crosscheck_snapshot, DestReceiver *dest, ParamListInfo params, QueryEnvironment *queryEnv, int instrument_options) { 
    	QueryDesc  *qd = (QueryDesc *) palloc(sizeof(QueryDesc));
    	qd->operation = plannedstmt->commandType;	/* operation */
    	qd->plannedstmt = plannedstmt;	/* plan */
    	qd->sourceText = sourceText;	/* query text */
    	qd->snapshot = RegisterSnapshot(snapshot);	/* snapshot */
    	/* RI check snapshot */
    	qd->crosscheck_snapshot = RegisterSnapshot(crosscheck_snapshot);
    	qd->dest = dest;			/* output dest */
    	qd->params = params;		/* parameter values passed into query */
    	qd->queryEnv = queryEnv;
    	qd->instrument_options = instrument_options;	/* instrumentation wanted? */
    	/* null these fields until set by ExecutorStart */
    	qd->tupDesc = NULL;
    	qd->estate = NULL;
    	qd->planstate = NULL;
    	qd->totaltime = NULL;
    	/* not yet executed */
    	qd->already_executed = false;
    	return qd;
    }

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

ExecutorStart(queryDesc, myeflags)函数采用先前由 CreateQueryDesc 创建的 QueryDesc （这是分开的，只是因为有些地方将 QueryDescs 用于实用程序命令）。 **QueryDesc 的 tupDesc 字段用来描述将返回的元组**，内部字段（estate 和 planstate）会被设置。eflags 包含 executor.h 中描述的标志位。注意：调用此方法时的 CurrentMemoryContext 将成为用于此 Executor 调用的每个查询上下文的父级。这里PG提供了一个函数钩子变量，让可加载插件在调用 ExecutorStart 时获得控制。 这样的插件通常会调用standard\_ExecutorStart()。

void ExecutorStart(QueryDesc *queryDesc, int eflags) { 
    	if (ExecutorStart_hook)
    		(*ExecutorStart_hook) (queryDesc, eflags);
    	else
    		standard_ExecutorStart(queryDesc, eflags);
    }

standard\_ExecutorStart函数进行如下工作：

1.  如果事务是只读的，我们需要检查是否有计划写入非临时表。 EXPLAIN 被认为是只读的。
2.  初始化EState，切换到per-query内存上下文
3.  从queryDesc填充内部参数、查询环境queryEnv，为内部参数分配空间
4.  如果queryDesc->operation为CMD\_SELECT，如果为SELECT FOR \[KEY\] UPDATE/SHARE和modifying CTEs need to mark tuples，则将estate->es\_output\_cid设置为GetCurrentCommandId(true)，如果queryDesc->operation为CMD\_INSERT、CMD\_DELETE、CMD\_UPDATE时，将estate->es\_output\_cid设置为GetCurrentCommandId(true)
5.  将一些重要信息拷贝到EState中
6.  Set up an AFTER-trigger statement context, unless told not to, or unless it’s EXPLAIN-only mode (when ExecutorFinish won’t be called)
7.  调用InitPlan(queryDesc, eflags)函数初始化计划树

void standard_ExecutorStart(QueryDesc *queryDesc, int eflags) { 
    	EState	   *estate;
    	MemoryContext oldcontext;
    	Assert(queryDesc != NULL);
    	Assert(queryDesc->estate == NULL);
    	if ((XactReadOnly || IsInParallelMode()) && !(eflags & EXEC_FLAG_EXPLAIN_ONLY))
    		ExecCheckXactReadOnly(queryDesc->plannedstmt);
    	/* Build EState, switch into per-query memory context for startup. */
    	estate = CreateExecutorState();
    	queryDesc->estate = estate;
    	oldcontext = MemoryContextSwitchTo(estate->es_query_cxt);
    	/* Fill in external parameters, if any, from queryDesc; and allocate workspace for internal parameters */
    	estate->es_param_list_info = queryDesc->params;
    	if (queryDesc->plannedstmt->paramExecTypes != NIL) { 
    		int			nParamExec;
    		nParamExec = list_length(queryDesc->plannedstmt->paramExecTypes);
    		estate->es_param_exec_vals = (ParamExecData *) palloc0(nParamExec * sizeof(ParamExecData));
    	}
    	estate->es_sourceText = queryDesc->sourceText;
    	/* Fill in the query environment, if any, from queryDesc. */
    	estate->es_queryEnv = queryDesc->queryEnv;
    	/* If non-read-only query, set the command ID to mark output tuples with */
    	switch (queryDesc->operation){ 
    		case CMD_SELECT:
    			if (queryDesc->plannedstmt->rowMarks != NIL || queryDesc->plannedstmt->hasModifyingCTE)
    				estate->es_output_cid = GetCurrentCommandId(true);
    			if (!queryDesc->plannedstmt->hasModifyingCTE)
    				eflags |= EXEC_FLAG_SKIP_TRIGGERS;
    			break;
    		case CMD_INSERT:
    		case CMD_DELETE:
    		case CMD_UPDATE:
    			estate->es_output_cid = GetCurrentCommandId(true);
    			break;
    		default:
    			elog(ERROR, "unrecognized operation code: %d", (int) queryDesc->operation);
    			break;
    	}
    	/* Copy other important information into the EState */
    	estate->es_snapshot = RegisterSnapshot(queryDesc->snapshot);
    	estate->es_crosscheck_snapshot = RegisterSnapshot(queryDesc->crosscheck_snapshot);
    	estate->es_top_eflags = eflags;
    	estate->es_instrument = queryDesc->instrument_options;
    	estate->es_jit_flags = queryDesc->plannedstmt->jitFlags;
    	/* Set up an AFTER-trigger statement context, unless told not to, or unless it's EXPLAIN-only mode (when ExecutorFinish won't be called). */
    	if (!(eflags & (EXEC_FLAG_SKIP_TRIGGERS | EXEC_FLAG_EXPLAIN_ONLY)))
    		AfterTriggerBeginQuery();
    	/* Initialize the plan state tree */
    	InitPlan(queryDesc, eflags);
    	MemoryContextSwitchTo(oldcontext);
    }

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

InitPlan(queryDesc, eflags)函数参见[PostgreSQL数据库查询执行——InitPlan函数分析][PostgreSQL_InitPlan]

### PortalSetResultFormat和SetRemoteDestReceiverParams函数 ###

/* Select the appropriate output format: text unless we are doing a * FETCH from a binary cursor. (Pretty grotty to have to do this here * --- but it avoids grottiness in other places. Ah, the joys of * backward compatibility...) */
    		format = 0;				/* TEXT is default */
    		if (IsA(parsetree->stmt, FetchStmt)){ 
    			FetchStmt  *stmt = (FetchStmt *) parsetree->stmt;
    			if (!stmt->ismove){ 
    				Portal		fportal = GetPortalByName(stmt->portalname);
    				if (PortalIsValid(fportal) && (fportal->cursorOptions & CURSOR_OPT_BINARY))
    					format = 1; /* BINARY */
    			}
    		}
    		PortalSetResultFormat(portal, 1, &format);
    		/* Now we can create the destination receiver object. */
    		receiver = CreateDestReceiver(dest);
    		if (dest == DestRemote)
    			SetRemoteDestReceiverParams(receiver, portal);

PortalSetResultFormat函数为portal输出选择format code。formats\[\]是客户端格式化请求。主要是设置portal->format。

void PortalSetResultFormat(Portal portal, int nFormats, int16 *formats) { 
    	int			natts;
    	int			i;
    	/* Do nothing if portal won't return tuples */
    	if (portal->tupDesc == NULL)
    		return;
    	natts = portal->tupDesc->natts;
    	portal->formats = (int16 *)MemoryContextAlloc(portal->portalContext,natts * sizeof(int16));
    	if (nFormats > 1) { 
    		/* format specified for each column */
    		if (nFormats != natts)
    			ereport(ERROR,(errcode(ERRCODE_PROTOCOL_VIOLATION),errmsg("bind message has %d result formats but query has %d columns",nFormats, natts)));
    		memcpy(portal->formats, formats, natts * sizeof(int16));
    	}else if (nFormats > 0){ 
    		/* single format specified, use for all columns */
    		int16		format1 = formats[0];
    		for (i = 0; i < natts; i++)
    			portal->formats[i] = format1;
    	}else{ 
    		/* use default format for all columns */
    		for (i = 0; i < natts; i++)
    			portal->formats[i] = 0;
    	}
    }

SetRemoteDestReceiverParams函数为DestRemote receiver设置参数

void SetRemoteDestReceiverParams(DestReceiver *self, Portal portal){ 
    	DR_printtup *myState = (DR_printtup *) self;
    	Assert(myState->pub.mydest == DestRemote || myState->pub.mydest == DestRemoteExecute);
    	myState->portal = portal;
    	if (PG_PROTOCOL_MAJOR(FrontendProtocol) < 3) { 
    		if (portal->formats && portal->formats[0] != 0)
    			myState->pub.receiveSlot = printtup_internal_20;
    		else
    			myState->pub.receiveSlot = printtup_20;
    	}
    }

### PortalRun函数 ###

`(void) PortalRun(portal,FETCH_ALL,true,true,receiver,receiver,completionTag);`  
count <= 0 被解释为无操作：目标启动和关闭，但没有其他任何事情发生。 此外，count == FETCH\_ALL 被解释为“所有行”。 请注意，在多查询情况下会忽略计数，在这种情况下，我们始终运行门户直至完成。  
isTopLevel：如果查询正在后端“顶级”执行（即直接从客户端命令消息执行），则为 true  
dest：将主要（canSetTag）查询的输出发送到哪里  
altdest：将非主要查询的输出发送到哪里  
completionTag：指向大小为 COMPLETION\_TAG\_BUFSIZE 的缓冲区

*  在其中存储命令完成状态字符串。
 *  如果调用者不想要状态字符串，则可能为 NULL。

如果portal的执行完成，则返回 true，如果由于计数参数耗尽而暂停，则返回 false。

bool PortalRun(Portal portal, long count, bool isTopLevel, bool run_once, DestReceiver *dest, DestReceiver *altdest, char *completionTag) { 
    	bool		result;
    	uint64		nprocessed;
    	ResourceOwner saveTopTransactionResourceOwner;
    	MemoryContext saveTopTransactionContext;
    	Portal		saveActivePortal;
    	ResourceOwner saveResourceOwner;
    	MemoryContext savePortalContext;
    	MemoryContext saveMemoryContext;
    	AssertArg(PortalIsValid(portal));
    	TRACE_POSTGRESQL_QUERY_EXECUTE_START();
    	/* Initialize completion tag to empty string */
    	if (completionTag)
    		completionTag[0] = '\0';
    	if (log_executor_stats && portal->strategy != PORTAL_MULTI_QUERY) { 
    		elog(DEBUG3, "PortalRun");
    		/* PORTAL_MULTI_QUERY logs its own stats per query */
    		ResetUsage();
    	}
    	/* Check for improper portal use, and mark portal active. */
    	MarkPortalActive(portal);
    	/* Set run_once flag. Shouldn't be clear if previously set. */
    	Assert(!portal->run_once || run_once);
    	portal->run_once = run_once;
    
    	/* Set up global portal context pointers. */
    	saveTopTransactionResourceOwner = TopTransactionResourceOwner;
    	saveTopTransactionContext = TopTransactionContext;
    	saveActivePortal = ActivePortal;
    	saveResourceOwner = CurrentResourceOwner;
    	savePortalContext = PortalContext;
    	saveMemoryContext = CurrentMemoryContext;
    	PG_TRY();
    	{ 
    		ActivePortal = portal;
    		if (portal->resowner)
    			CurrentResourceOwner = portal->resowner;
    		PortalContext = portal->portalContext;
    		MemoryContextSwitchTo(PortalContext);
    		switch (portal->strategy){ 
    			case PORTAL_ONE_SELECT:
    			case PORTAL_ONE_RETURNING:
    			case PORTAL_ONE_MOD_WITH:
    			case PORTAL_UTIL_SELECT:
    				if (portal->strategy != PORTAL_ONE_SELECT && !portal->holdStore)
    					FillPortalStore(portal, isTopLevel);
    				/* Now fetch desired portion of results. */
    				nprocessed = PortalRunSelect(portal, true, count, dest);
    				/* If the portal result contains a command tag and the caller * gave us a pointer to store it, copy it. Patch the "SELECT" * tag to also provide the rowcount. */
    				if (completionTag && portal->commandTag)
    				{ 
    					if (strcmp(portal->commandTag, "SELECT") == 0)
    						snprintf(completionTag, COMPLETION_TAG_BUFSIZE, "SELECT " UINT64_FORMAT, nprocessed);
    					else
    						strcpy(completionTag, portal->commandTag);
    				}
    				/* Mark portal not active */
    				portal->status = PORTAL_READY;
    				/ Since it's a forward fetch, say DONE iff atEnd is now true. */
    				result = portal->atEnd;
    				break;
    			case PORTAL_MULTI_QUERY:
    				...
    				break;
    			default:
    				elog(ERROR, "unrecognized portal strategy: %d",(int) portal->strategy);
    				result = false; /* keep compiler quiet */
    				break;
    		}
    	}
    	PG_CATCH();
    	{ 
            ...
    		PG_RE_THROW();
    	}
    	PG_END_TRY();
    
    	if (saveMemoryContext == saveTopTransactionContext)
    		MemoryContextSwitchTo(TopTransactionContext);
    	else
    		MemoryContextSwitchTo(saveMemoryContext);
    	ActivePortal = saveActivePortal;
    	if (saveResourceOwner == saveTopTransactionResourceOwner)
    		CurrentResourceOwner = TopTransactionResourceOwner;
    	else
    		CurrentResourceOwner = saveResourceOwner;
    	PortalContext = savePortalContext;
    	if (log_executor_stats && portal->strategy != PORTAL_MULTI_QUERY)
    		ShowUsage("EXECUTOR STATISTICS");
    	TRACE_POSTGRESQL_QUERY_EXECUTE_DONE();
    	return result;
    }

针对PORTAL\_ONE\_SELECT策略，主要执行函数为`nprocessed = PortalRunSelect(portal, true, count, dest)`，

static uint64 PortalRunSelect(Portal portal, bool forward, long count, DestReceiver *dest)
    { 
    	QueryDesc  *queryDesc;
    	ScanDirection direction;
    	uint64		nprocessed;
    	/* NB: queryDesc will be NULL if we are fetching from a held cursor or a completed utility query; can't use it in that path. */
    	queryDesc = portal->queryDesc;
    	/* Caller messed up if we have neither a ready query nor held data. */
    	Assert(queryDesc || portal->holdStore);
    	/* Force the queryDesc destination to the right thing. This supports * MOVE, for example, which will pass in dest = DestNone. This is okay to * change as long as we do it on every fetch. (The Executor must not * assume that dest never changes.) */
    	if (queryDesc)
    		queryDesc->dest = dest;
    	/* * Determine which direction to go in, and check to see if we're already * at the end of the available tuples in that direction. If so, set the * direction to NoMovement to avoid trying to fetch any tuples. (This * check exists because not all plan node types are robust about being * called again if they've already returned NULL once.) Then call the * executor (we must not skip this, because the destination needs to see a * setup and shutdown even if no tuples are available). Finally, update * the portal position state depending on the number of tuples that were * retrieved. */
    	if (forward){ 
    		if (portal->atEnd || count <= 0){ 
    			direction = NoMovementScanDirection;
    			count = 0;			/* don't pass negative count to executor */
    		}else
    			direction = ForwardScanDirection;
    		/* In the executor, zero count processes all rows */
    		if (count == FETCH_ALL)
    			count = 0;
    		if (portal->holdStore)
    			nprocessed = RunFromStore(portal, direction, (uint64) count, dest);
    		else{ 
    			PushActiveSnapshot(queryDesc->snapshot);
    			ExecutorRun(queryDesc, direction, (uint64) count,portal->run_once);
    			nprocessed = queryDesc->estate->es_processed;
    			PopActiveSnapshot();
    		}
    		if (!ScanDirectionIsNoMovement(direction)){ 
    			if (nprocessed > 0)
    				portal->atStart = false;	/* OK to go backward now */
    			if (count == 0 || nprocessed < (uint64) count)
    				portal->atEnd = true;	/* we retrieved 'em all */
    			portal->portalPos += nprocessed;
    		}
    	}else{ 
    		if (portal->cursorOptions & CURSOR_OPT_NO_SCROLL)
    			ereport(ERROR,(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),errmsg("cursor can only scan forward"),errhint("Declare it with SCROLL option to enable backward scan.")));
    		if (portal->atStart || count <= 0){ 
    			direction = NoMovementScanDirection;
    			count = 0;			/* don't pass negative count to executor */
    		}else
    			direction = BackwardScanDirection;
    		/* In the executor, zero count processes all rows */
    		if (count == FETCH_ALL)
    			count = 0;
    		if (portal->holdStore)
    			nprocessed = RunFromStore(portal, direction, (uint64) count, dest);
    		else{ 
    			PushActiveSnapshot(queryDesc->snapshot);
    			ExecutorRun(queryDesc, direction, (uint64) count,portal->run_once);
    			nprocessed = queryDesc->estate->es_processed;
    			PopActiveSnapshot();
    		}
    
    		if (!ScanDirectionIsNoMovement(direction)){ 
    			if (nprocessed > 0 && portal->atEnd){ 
    				portal->atEnd = false;	/* OK to go forward now */
    				portal->portalPos++;	/* adjust for endpoint case */
    			}
    			if (count == 0 || nprocessed < (uint64) count){ 
    				portal->atStart = true; /* we retrieved 'em all */
    				portal->portalPos = 0;
    			}else{ 
    				portal->portalPos -= nprocessed;
    			}
    		}
    	}
    	return nprocessed;
    }

### PortalDrop函数 ###

`PortalDrop(portal, false)`函数，函数逻辑如下：

1.  调用portal->cleanup函数
2.  调用PortalHashTableDelete(portal)函数
3.  调用PortalReleaseCachedPlan(portal)函数

void PortalDrop(Portal portal, bool isTopCommit) { 
    	AssertArg(PortalIsValid(portal));
    	/* Don't allow dropping a pinned portal, it's still needed by whoever pinned it. */
    	if (portal->portalPinned)
    		ereport(ERROR, (errcode(ERRCODE_INVALID_CURSOR_STATE), errmsg("cannot drop pinned portal \"%s\"", portal->name)));
    	/* Not sure if the PORTAL_ACTIVE case can validly happen or not... */
    	if (portal->status == PORTAL_ACTIVE)
    		ereport(ERROR,(errcode(ERRCODE_INVALID_CURSOR_STATE), errmsg("cannot drop active portal \"%s\"", portal->name)));
    	if (PointerIsValid(portal->cleanup)) { 
    		portal->cleanup(portal);
    		portal->cleanup = NULL;
    	}
    	PortalHashTableDelete(portal);
    	PortalReleaseCachedPlan(portal);
    	if (portal->holdSnapshot){ 
    		if (portal->resowner)
    			UnregisterSnapshotFromOwner(portal->holdSnapshot,portal->resowner);
    		portal->holdSnapshot = NULL;
    	}
    	if (portal->resowner && (!isTopCommit || portal->status == PORTAL_FAILED)) { 
    		bool		isCommit = (portal->status != PORTAL_FAILED);
    		ResourceOwnerRelease(portal->resowner,RESOURCE_RELEASE_BEFORE_LOCKS,isCommit, false);
    		ResourceOwnerRelease(portal->resowner,RESOURCE_RELEASE_LOCKS,isCommit, false);
    		ResourceOwnerRelease(portal->resowner,RESOURCE_RELEASE_AFTER_LOCKS, isCommit, false);
    		ResourceOwnerDelete(portal->resowner);
    	}
    	portal->resowner = NULL;
    	if (portal->holdStore){ 
    		MemoryContext oldcontext;
    		oldcontext = MemoryContextSwitchTo(portal->holdContext);
    		tuplestore_end(portal->holdStore);
    		MemoryContextSwitchTo(oldcontext);
    		portal->holdStore = NULL;
    	}
    	/* delete tuplestore storage, if any */
    	if (portal->holdContext)
    		MemoryContextDelete(portal->holdContext);
    	/* release subsidiary storage */
    	MemoryContextDelete(portal->portalContext);
    	/* release portal struct (it's in TopPortalContext) */
    	pfree(portal);
    }

[resize_m_lfit_w_962_pic_center]: /images/20221014/c28d0d5ec9ae459db20b2343b529c498.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]: /images/20221014/afc08d8a97ff4d3791668ba29bf8dac6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]: /images/20221014/b6112441cae745d3862fc0a116a54cf6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 2]: /images/20221014/0a82bede87134fe191f5cd8f2674b608.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 3]: /images/20221014/8286e313b6eb4d128a1256be3d6e2799.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221014/b86062ce39e746d29a1ef460ae918b63.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221014/1a80fc3815164e068f80f04871f57c4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221014/d4a40ae929fa4adfb1535a75fe5fc9cd.png
[PostgreSQL_InitPlan]: https://feishujun.blog.csdn.net/article/details/118052184