PostgreSQL数据库并行查询——工作原理和机制

悠悠 2022-09-13 00:22 871阅读 0赞

大数据时代，人们使用数据库系统处理的数据量越来越大，请求越来越复杂，对数据库系统的大数据处理能力和混合负载能力提出更高的要求。PostgreSQL 作为世界上最先进的开源数据库，在大数据处理方面做了很多工作，如并行和分区。

PostgreSQL 从 2016 年发布的 9.6 开始支持并行，在此之前，PostgreSQL 仅能使用一个进程处理用户的请求，无法充分利用资源，亦无法很好地满足大数据量、复杂查询下的性能需求。2018 年 10 月发布的 PostgreSQL 11，在并行方面做了大量工作，支持了并行哈希连接，并行 Append 以及并行创建索引等特性，对于分区表，支持了 Partition-wise JOIN。

本文从以下三方面介绍 PostgreSQL 的并行查询特性：

*  并行查询基础组件，包括后台工作进程（Background Work Process），动态共享内存（Dynamic Shared Memory）以及后台工作进程间的通信机制和消息传递机制
 *  并行执行算子的实现，包括并行顺序扫描、并行索引扫描等并行扫描算子，三种连接方式的并行执行以及并行 Append
 *  并行查询优化，介绍并行查询引入的两种计划节点，基于规则计算后台工作进程数量以及代价估算

举个例子  
首先，通过一个例子，让我们对 PostgreSQL 的并行查询以及并行计划有一个较宏观的认识。如下查询：统计人员表 people 中参加 2018 PostgreSQL 大会的人数：`SELECT COUNT(*) FROM people WHERE inpgconn2018 = 'Y';`  
没有开并行的情况下（max\_parallel\_workers\_per\_gather=0），查询计划如下：

Aggregate  (cost=169324.73..169324.74 rows=1 width=8) (actual time=983.729..983.730 rows=1 loops=1)
       ->  Seq Scan on people  (cost=0.00..169307.23 rows=7001 width=0) (actual time=981.723..983.051 rows=9999 loops=1)
             Filter: (atpgconn2018 = 'Y'::bpchar)
             Rows Removed by Filter: 9990001
     Planning Time: 0.066 ms
     Execution Time: 983.760 ms

开启并行的情况下（max\_parallel\_workers\_per\_gather=2），查询计划如下：

Finalize Aggregate  (cost=97389.77..97389.78 rows=1 width=8) (actual time=384.848..384.848 rows=1 loops=1)
       ->  Gather  (cost=97389.55..97389.76 rows=2 width=8) (actual time=384.708..386.486 rows=3 loops=1)
             Workers Planned: 2
             Workers Launched: 2
             ->  Partial Aggregate  (cost=96389.55..96389.56 rows=1 width=8) (actual time=379.597..379.597 rows=1 loops=3)
                   ->  Parallel Seq Scan on people  (cost=0.00..96382.26 rows=2917 width=0)
    				 (actual time=378.831..379.341 rows=3333 loops=3)
                         Filter: (atpgconn2018 = 'Y'::bpchar)
                         Rows Removed by Filter: 3330000
     Planning Time: 0.063 ms
     Execution Time: 386.532 ms

max\_parallel\_workers\_per\_gather 参数控制执行节点的最大并行进程数，通过以上并行计划可知，开启并行后，会启动两个 worker 进程（即 Workers Launched: 2）并行执行，且执行时间（Execution Time）仅为不并行的40%。该并行计划可用下图表示：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_18_color_FFFFFF_t_70_g_se_x_16]  
并行查询计划中，我们将处理用户请求的 backend 进程称之为主进程（leader），将执行时动态生成的进程称之为工作进程（worker）。每个 worker 执行 Gather 节点以下计划的一个副本，leader 节点主要负责处理 Gather 及其以上节点的操作，根据 worker 数不同，leader 也可能会执行 Gather 以下计划的副本。

## 原理 ##

PostgreSQL的并行化包含三个重要组件：进程本身(leader进程)、gather、workers。没有开启并行化的时候，进程自身处理所有的数据；一旦计划器决定某个查询或查询中部分可以使用并行的时候，就会在查询的并行化部分添加一个gather节点，将gather节点作为子查询树的根节点。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16]  
查询执行是从leader进程开始。一旦开启了并行或查询中部分支持并行，就会分配一个gather节点和多个worker线程。relation的blocks在各个workers线程之间划分。workers的数量受postgresql的配置参数控制。workers之间使用共享内存相互协调和通信，一旦worker完成了自己的工作，结果就被传给了leader进程。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
workers和leader进程之间使用消息队列(依赖共享内存)进行通信。每个进程有两个队列：一个是error队列；一个是tuples队列。

### 后台工作进程（Background Worker Process） ###

PostgreSQL 是多进程架构，主要包括以下几类进程：

*  守护进程，通常称之为 postmaster 进程，接收用户的连接并 fork 一个子进程处理用户的请求
 *  backend 进程，即 postmaster 创建的用于处理用户请求的进程，每个连接对应一个 backend 进程
 *  辅助进程，用于执行 checkpoint，后台刷脏等操作的进程
 *  **后台工作进程，用于执行特定任务而动态启动的进程，如上文提到的 worker 进程**  
    ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
    上图中 server process 即 postmaster 进程，在内核中，postmaster 与 backend 进程都是 postgres 进程，只是角色不同。对于一个并行查询，其创建 worker 进程的大致流程如下：
 *  client 创建连接，postmaster 为其 fork 一个 backend 进程处理请求
 *  backend 接收用户请求，并生成并行查询计划
 *  **执行器向 backgroudworker 注册 worker 进程**（并没有启动）
 *  执行器通知（kill）postmaster 启动 worker 进程
 *  worker 进程与 leader 进程协调执行，将结果返回 client

src/backend/access/transam/README.parallel  
PG提供了一些简单设施用于更方便地写并行算法。使用ParallelContext数据结构，可以安排启动background worker进程，初始化它们的状态，通过动态共享内存通信，运行可以运行在用户后端或并行worker的复杂代码。

/* List of active parallel contexts. */
    static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list);
    typedef struct ParallelContext { 
    	dlist_node	node;         
    	SubTransactionId subid;   // 调用GetCurrentSubTransactionId获取子事务ID
    	int			nworkers;     // worker的数量
    	int			nworkers_launched;
    	char	   *library_name; // 库名
    	char	   *function_name;// 函数名
    	ErrorContextCallback *error_context_stack; //错误上下文栈
    	shm_toc_estimator estimator;
    	dsm_segment *seg;
    	void	   *private_memory;
    	shm_toc    *toc;
    	ParallelWorkerInfo *worker;
    	int			nknown_attached_workers;
    	bool	   *known_attached_workers;
    } ParallelContext;

并行上下文链表pcxt\_list，用于存放活动的并行上下文。ParallelContext结构体中的node用于插入到并行上下文链表pcxt\_list。

实现惯例：在进入并行操作前，调用EnterParallelMode，完成并行操作后调用ExitParallelMode。使用ParallelContext并行化相应操作。理想情况下，在WaitForParallelWorkersToFinish调用之后，ParallelContext可以重置，这样worker可以使用相同的并行上下文启动。为了达到这样的效果，需要先调用ReinitializeParallelDSM来重新初始化并行上下文状态机管理的状态，然后必要的重置操作，最后就可以重新调用LaunchParallelWorkers。

EnterParallelMode();		/* prohibit unsafe state changes */ // 递增当前事务状态结构体中parallelModeLevel
    	pcxt = CreateParallelContext("library_name", "function_name", nworkers);
    	/* Allow space for application-specific data here. */
    	shm_toc_estimate_chunk(&pcxt->estimator, size);
    	shm_toc_estimate_keys(&pcxt->estimator, keys);
    	InitializeParallelDSM(pcxt);	/* create DSM and copy state to it */
    	/* Store the data for which we reserved space. */
    	space = shm_toc_allocate(pcxt->toc, size);
    	shm_toc_insert(pcxt->toc, key, space);
    	LaunchParallelWorkers(pcxt);
    	/* do parallel stuff */
    	WaitForParallelWorkersToFinish(pcxt);
    	/* read any final results from dynamic shared memory */
    	DestroyParallelContext(pcxt);
    	ExitParallelMode();

#### CreateParallelContext ####

IsInParallelMode函数判断是否处于并行模式

bool IsInParallelMode(void) { 
    	return CurrentTransactionState->parallelModeLevel != 0;
    }

CreateParallelContext函数创建新的并行上下文，进入并行模式下调用该函数。在退出当前子事务subtransaction前需要释放并行上下文。

ParallelContext *CreateParallelContext(const char *library_name, const char *function_name, int nworkers) { 
    	MemoryContext oldcontext;
    	ParallelContext *pcxt;
    	/* It is unsafe to create a parallel context if not in parallel mode. */
    	Assert(IsInParallelMode());
    	/* Number of workers should be non-negative. */
    	Assert(nworkers >= 0);
    	/* We might be running in a short-lived memory context. */
    	oldcontext = MemoryContextSwitchTo(TopTransactionContext);
    	/* Initialize a new ParallelContext. */
    	pcxt = palloc0(sizeof(ParallelContext));
    	pcxt->subid = GetCurrentSubTransactionId();     //调用GetCurrentSubTransactionId获取子事务ID
    	pcxt->nworkers = nworkers;
    	pcxt->library_name = pstrdup(library_name);
    	pcxt->function_name = pstrdup(function_name);
    	pcxt->error_context_stack = error_context_stack; //错误上下文栈
    	shm_toc_initialize_estimator(&pcxt->estimator);  //Allow space for application-specific data
    	dlist_push_head(&pcxt_list, &pcxt->node);
    	/* Restore previous memory context. */
    	MemoryContextSwitchTo(oldcontext);
    	return pcxt;
    }

pcxt->subid = GetCurrentSubTransactionId()调用GetCurrentSubTransactionId获取子事务ID，在退出当前子事务subtransaction前需要释放并行上下文。

搜索PG12.6的源代码，调用CreateParallelContext函数的地方有两处：1. nbsort.c的\_bt\_begin\_parallel函数 2. execParallel.c的ExecInitParallelPlan函数。我们以ExecInitParallelPlan函数为例查看其调用链。  
ExecGather (nodeGather.c) --> ExecInitParallelPlan  
ExecGatherMerge (nodeGatherMerge.c) --> ExecInitParallelPlan

##### ExecGather #####

ExecGather通过多个worker扫描关系表并返回下一个元组：

1.  在第一次执行时初始化并行上下文和worker进程，因为需要分配大量动态共享内存，所以懒加载是有效缓解资源紧张的策略。
2.  初始化或重新初始化worker需要的共享状态node->pei
3.  LaunchParallelWorkers(pcxt)
4.  创建tuples消息队列读取元组ExecParallelCreateReaders(node->pei)
5.  不管是从worker还是本身进程获取元组gather\_getnext(node)

static TupleTableSlot *ExecGather(PlanState *pstate) { 
    	GatherState *node = castNode(GatherState, pstate);
    	TupleTableSlot *slot;
    	ExprContext *econtext;
    	CHECK_FOR_INTERRUPTS();
    	/* Initialize the parallel context and workers on first execution. We do this on first execution rather than during node initialization, as it needs to allocate a large dynamic segment, so it is better to do it only if it is really needed. */
    	if (!node->initialized) { 
    		EState	   *estate = node->ps.state;
    		Gather	   *gather = (Gather *) node->ps.plan;
    		/* Sometimes we might have to run without parallelism; but if parallel mode is active then we can try to fire up some workers. */
    		if (gather->num_workers > 0 && estate->es_use_parallel_mode) { 
    			ParallelContext *pcxt;
    			/* Initialize, or re-initialize, shared state needed by workers. */
    			if (!node->pei)
    				node->pei = ExecInitParallelPlan(node->ps.lefttree,estate,gather->initParam,gather->num_workers,node->tuples_needed);
    			else
    				ExecParallelReinitialize(node->ps.lefttree,node->pei,gather->initParam);
    
    			/* Register backend workers. We might not get as many as we requested, or indeed any at all. */
    			pcxt = node->pei->pcxt;
    			LaunchParallelWorkers(pcxt);
    			/* We save # workers launched for the benefit of EXPLAIN */
    			node->nworkers_launched = pcxt->nworkers_launched;
    
    			/* Set up tuple queue readers to read the results. */
    			if (pcxt->nworkers_launched > 0) { 
    				ExecParallelCreateReaders(node->pei);
    				/* Make a working array showing the active readers */
    				node->nreaders = pcxt->nworkers_launched;
    				node->reader = (TupleQueueReader **)palloc(node->nreaders * sizeof(TupleQueueReader *));
    				memcpy(node->reader, node->pei->reader,node->nreaders * sizeof(TupleQueueReader *));
    			}else{ 
    				/* No workers? Then never mind. */
    				node->nreaders = 0;
    				node->reader = NULL;
    			}
    			node->nextreader = 0;
    		}
    		/* Run plan locally if no workers or enabled and not single-copy. */
    		node->need_to_scan_locally = (node->nreaders == 0) || (!gather->single_copy && parallel_leader_participation);
    		node->initialized = true;
    	}
    	/* Reset per-tuple memory context to free any expression evaluation storage allocated in the previous tuple cycle. */
    	econtext = node->ps.ps_ExprContext;
    	ResetExprContext(econtext);
    	/* Get next tuple, either from one of our workers, or by running the plan ourselves. */
    	slot = gather_getnext(node);
    	if (TupIsNull(slot))
    		return NULL;
    	/* If no projection is required, we're done. */
    	if (node->ps.ps_ProjInfo == NULL)
    		return slot;
    	/* Form the result tuple using ExecProject(), and return it. */
    	econtext->ecxt_outertuple = slot;
    	return ExecProject(node->ps.ps_ProjInfo);
    }

##### ExecGatherMerge #####

#### 动态共享内存（Dynamic Shared Memory）与 IPC ####

PostgreSQL 是多进程架构，进程间通信往往通过共享内存和信号量来实现。对于并行查询而言，执行时创建的 worker 进程与 leader 进程同样通过共享内存实现数据交互。但**这部分内存无法像普通的共享内存那样在系统启动时预先分配**，毕竟直到真正执行时才知道有多少 worker 进程，以及需要分配多少内存。PostgreSQL 实现了动态共享内存，即在执行时动态创建，用于 leader 与 worker 间通信，执行完成后释放。*基于动态共享内存的队列用于进程间传递元组和错误消息*。  
如下图，每个 worker 在动态共享内存中都有对应的元组队列和错误队列，worker 将执行结果放入队列中，leader 会从对应队列获取元组返回给上层算子。动态共享内存的具体实现原理和细节在此不做展开。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_17_color_FFFFFF_t_70_g_se_x_16]

##### InitializeParallelDSM #####

InitializeParallelDSM函数用于建立动态共享内存、拷贝状态、其他bookkeeping的worker需要的信息。

void InitializeParallelDSM(ParallelContext *pcxt) { 
    	MemoryContext oldcontext;
    	Size		library_len = 0;
    	Size		guc_len = 0;
    	Size		combocidlen = 0;
    	Size		tsnaplen = 0;
    	Size		asnaplen = 0;
    	Size		tstatelen = 0;
    	Size		reindexlen = 0;
    	Size		relmapperlen = 0;
    	Size		enumblacklistlen = 0;
    	Size		segsize = 0;
    	int			i;
    	FixedParallelState *fps;
    	dsm_handle	session_dsm_handle = DSM_HANDLE_INVALID;
    	Snapshot	transaction_snapshot = GetTransactionSnapshot();
    	Snapshot	active_snapshot = GetActiveSnapshot();
    	/* We might be running in a very short-lived memory context. */
    	oldcontext = MemoryContextSwitchTo(TopTransactionContext);
    	/* Allow space to store the fixed-size parallel state. */
    	shm_toc_estimate_chunk(&pcxt->estimator, sizeof(FixedParallelState));
    	shm_toc_estimate_keys(&pcxt->estimator, 1);
    	/* Normally, the user will have requested at least one worker process, but if by chance they have not, we can skip a bunch of things here. */
    	if (pcxt->nworkers > 0) { 
    		/* Get (or create) the per-session DSM segment's handle. */
    		session_dsm_handle = GetSessionDsmHandle();
    		/* If we weren't able to create a per-session DSM segment, then we can * continue but we can't safely launch any workers because their * record typmods would be incompatible so they couldn't exchange * tuples. */
    		if (session_dsm_handle == DSM_HANDLE_INVALID)
    			pcxt->nworkers = 0;
    	}
    
    	if (pcxt->nworkers > 0) { 
    		/* Estimate space for various kinds of state sharing. */
    		library_len = EstimateLibraryStateSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, library_len);
    		guc_len = EstimateGUCStateSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, guc_len);
    		combocidlen = EstimateComboCIDStateSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, combocidlen);
    		tsnaplen = EstimateSnapshotSpace(transaction_snapshot);
    		shm_toc_estimate_chunk(&pcxt->estimator, tsnaplen);
    		asnaplen = EstimateSnapshotSpace(active_snapshot);
    		shm_toc_estimate_chunk(&pcxt->estimator, asnaplen);
    		tstatelen = EstimateTransactionStateSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, tstatelen);
    		shm_toc_estimate_chunk(&pcxt->estimator, sizeof(dsm_handle));
    		reindexlen = EstimateReindexStateSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, reindexlen);
    		relmapperlen = EstimateRelationMapSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, relmapperlen);
    		enumblacklistlen = EstimateEnumBlacklistSpace();
    		shm_toc_estimate_chunk(&pcxt->estimator, enumblacklistlen);
    		/* If you add more chunks here, you probably need to add keys. */
    		shm_toc_estimate_keys(&pcxt->estimator, 10);
    		/* Estimate space need for error queues. */
    		StaticAssertStmt(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) == PARALLEL_ERROR_QUEUE_SIZE, "parallel error queue size not buffer-aligned");
    		shm_toc_estimate_chunk(&pcxt->estimator, mul_size(PARALLEL_ERROR_QUEUE_SIZE, pcxt->nworkers));
    		shm_toc_estimate_keys(&pcxt->estimator, 1);
    		/* Estimate how much we'll need for the entrypoint info. */
    		shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) + strlen(pcxt->function_name) + 2);
    		shm_toc_estimate_keys(&pcxt->estimator, 1);
    	}
    
    	/* Create DSM and initialize with new table of contents. But if the user * didn't request any workers, then don't bother creating a dynamic shared * memory segment; instead, just use backend-private memory. * Also, if we can't create a dynamic shared memory segment because the * maximum number of segments have already been created, then fall back to * backend-private memory, and plan not to use any workers. We hope this * won't happen very often, but it's better to abandon the use of * parallelism than to fail outright. */
    	segsize = shm_toc_estimate(&pcxt->estimator);
    	if (pcxt->nworkers > 0)
    		pcxt->seg = dsm_create(segsize, DSM_CREATE_NULL_IF_MAXSEGMENTS);
    	if (pcxt->seg != NULL)
    		pcxt->toc = shm_toc_create(PARALLEL_MAGIC, dsm_segment_address(pcxt->seg), segsize);
    	else { 
    		pcxt->nworkers = 0;
    		pcxt->private_memory = MemoryContextAlloc(TopMemoryContext, segsize);
    		pcxt->toc = shm_toc_create(PARALLEL_MAGIC, pcxt->private_memory, segsize);
    	}
    
    	/* Initialize fixed-size state in shared memory. */
    	fps = (FixedParallelState *)shm_toc_allocate(pcxt->toc, sizeof(FixedParallelState));
    	fps->database_id = MyDatabaseId;
    	fps->authenticated_user_id = GetAuthenticatedUserId();
    	fps->outer_user_id = GetCurrentRoleId();
    	fps->is_superuser = session_auth_is_superuser;
    	GetUserIdAndSecContext(&fps->current_user_id, &fps->sec_context);
    	GetTempNamespaceState(&fps->temp_namespace_id,&fps->temp_toast_namespace_id);
    	fps->parallel_master_pgproc = MyProc;
    	fps->parallel_master_pid = MyProcPid;
    	fps->parallel_master_backend_id = MyBackendId;
    	fps->xact_ts = GetCurrentTransactionStartTimestamp();
    	fps->stmt_ts = GetCurrentStatementStartTimestamp();
    	fps->serializable_xact_handle = ShareSerializableXact();
    	SpinLockInit(&fps->mutex);
    	fps->last_xlog_end = 0;
    	shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps);
    	/* We can skip the rest of this if we're not budgeting for any workers. */
    	if (pcxt->nworkers > 0){ 
    		char	   *libraryspace;
    		char	   *gucspace;
    		char	   *combocidspace;
    		char	   *tsnapspace;
    		char	   *asnapspace;
    		char	   *tstatespace;
    		char	   *reindexspace;
    		char	   *relmapperspace;
    		char	   *error_queue_space;
    		char	   *session_dsm_handle_space;
    		char	   *entrypointstate;
    		char	   *enumblacklistspace;
    		Size		lnamelen;
    		/* Serialize shared libraries we have loaded. */
    		libraryspace = shm_toc_allocate(pcxt->toc, library_len);
    		SerializeLibraryState(library_len, libraryspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_LIBRARY, libraryspace);
    		/* Serialize GUC settings. */
    		gucspace = shm_toc_allocate(pcxt->toc, guc_len);
    		SerializeGUCState(guc_len, gucspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_GUC, gucspace);
    		/* Serialize combo CID state. */
    		combocidspace = shm_toc_allocate(pcxt->toc, combocidlen);
    		SerializeComboCIDState(combocidlen, combocidspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_COMBO_CID, combocidspace);
    		/* Serialize transaction snapshot and active snapshot. */
    		tsnapspace = shm_toc_allocate(pcxt->toc, tsnaplen);
    		SerializeSnapshot(transaction_snapshot, tsnapspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT, tsnapspace);
    		asnapspace = shm_toc_allocate(pcxt->toc, asnaplen);
    		SerializeSnapshot(active_snapshot, asnapspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, asnapspace);
    		/* Provide the handle for per-session segment. */
    		session_dsm_handle_space = shm_toc_allocate(pcxt->toc,sizeof(dsm_handle));
    		*(dsm_handle *) session_dsm_handle_space = session_dsm_handle;
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_SESSION_DSM,session_dsm_handle_space);
    		/* Serialize transaction state. */
    		tstatespace = shm_toc_allocate(pcxt->toc, tstatelen);
    		SerializeTransactionState(tstatelen, tstatespace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_STATE, tstatespace);
    		/* Serialize reindex state. */
    		reindexspace = shm_toc_allocate(pcxt->toc, reindexlen);
    		SerializeReindexState(reindexlen, reindexspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_REINDEX_STATE, reindexspace);
    		/* Serialize relmapper state. */
    		relmapperspace = shm_toc_allocate(pcxt->toc, relmapperlen);
    		SerializeRelationMap(relmapperlen, relmapperspace);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_RELMAPPER_STATE,relmapperspace);
    		/* Serialize enum blacklist state. */
    		enumblacklistspace = shm_toc_allocate(pcxt->toc, enumblacklistlen);
    		SerializeEnumBlacklist(enumblacklistspace, enumblacklistlen);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENUMBLACKLIST,enumblacklistspace);
    		/* Allocate space for worker information. */
    		pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
    		/* Establish error queues in dynamic shared memory. * These queues should be used only for transmitting ErrorResponse, * NoticeResponse, and NotifyResponse protocol messages. Tuple data * should be transmitted via separate (possibly larger?) queues. */
    		error_queue_space =shm_toc_allocate(pcxt->toc,mul_size(PARALLEL_ERROR_QUEUE_SIZE,pcxt->nworkers));
    		for (i = 0; i < pcxt->nworkers; ++i){ 
    			char	   *start;
    			shm_mq	   *mq;
    			start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
    			mq = shm_mq_create(start, PARALLEL_ERROR_QUEUE_SIZE);
    			shm_mq_set_receiver(mq, MyProc);
    			pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
    		}
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, error_queue_space);
    		/* Serialize entrypoint information. It's unsafe to pass function * pointers across processes, as the function pointer may be different * in each process in EXEC_BACKEND builds, so we always pass library * and function name. (We use library name "postgres" for functions * in the core backend.) */
    		lnamelen = strlen(pcxt->library_name);
    		entrypointstate = shm_toc_allocate(pcxt->toc, lnamelen + strlen(pcxt->function_name) + 2);
    		strcpy(entrypointstate, pcxt->library_name);
    		strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
    		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
    	}
    	/* Restore previous memory context. */
    	MemoryContextSwitchTo(oldcontext);
    }

#### LaunchParallelWorkers ####

src/backend/access/transam/parallel.c Infrastructure for launching parallel workers  
src/backend/executor/execParallel.c Support routines for parallel execution

## 并行支持算子 ##

以上简单介绍了并行查询依赖的两个重要基础组件：后台工作进程和动态共享内存。前者用于动态创建 worker，以并行执行子查询计划；后者用于 leader 和 worker 间通信和数据交互。本节介绍 PostgreSQL 目前支持并行执行的算子的实现原理，包括：

并行扫描，如并行顺序扫描，并行索引扫描等  
并行连接，如并行哈希连接，并行 NestLoop 连接等  
并行 Append

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 3]

### 并行顺序扫描(Parallel sequential scan) ###

在PostgreSQL 9.6中，增加了对并行顺序扫描的支持。顺序扫描是在表上进行的扫描，在该表中一个接一个的块顺序地被评估。就其本质而言，顺序扫描允许并行化。这样，整个表将在多个workers线程之间顺序扫描。

并行顺序扫描快并不是因为可以并行地读，而是将数据分散到了多个cpu。

abce=# explain analyze select work_hour from hh_adds_static_day where create_time <= date '20201010'-interval '10' day;
                                                               QUERY PLAN                                                          
    --------------------------------------------------------------------------------------------------------------------------------
     Seq Scan on hh_adds_static_day  (cost=0.00..261864.36 rows=4241981 width=4) (actual time=0.012..1407.214 rows=4228109 loops=1)
       Filter: (create_time <= '2020-09-30 00:00:00'::timestamp without time zone)
       Rows Removed by Filter: 735600
     Planning Time: 0.108 ms
     Execution Time: 1585.835 ms
    (5 rows)

顺序扫描产生了大量的行，但是没有使用聚合函数。因此，查询使用的是单个cpu核心。

增加一个sum函数后，很明显使用了两个工作线程，从而使得查询加速：

abce=# explain analyze select sum(work_hour) from hh_adds_static_day where create_time <= date '20201010'-interval '10' day;
                                                                            QUERY PLAN                                                                       
    ----------------------------------------------------------------------------------------------------------------------------------------------------------
     Finalize Aggregate  (cost=231089.60..231089.61 rows=1 width=4) (actual time=749.998..751.529 rows=1 loops=1)
       ->  Gather  (cost=231089.38..231089.59 rows=2 width=4) (actual time=749.867..751.515 rows=3 loops=1)
             Workers Planned: 2
             Workers Launched: 2
             ->  Partial Aggregate  (cost=230089.38..230089.39 rows=1 width=4) (actual time=746.463..746.464 rows=1 loops=3)
                   ->  Parallel Seq Scan on hh_adds_static_day  (cost=0.00..225670.65 rows=1767492 width=4) (actual time=0.032..489.501 rows=1409370 loops=3)
                         Filter: (create_time <= '2020-09-30 00:00:00'::timestamp without time zone)
                         Rows Removed by Filter: 245200
     Planning Time: 0.112 ms
     Execution Time: 751.611 ms
    (10 rows)

### 并行聚合(Parallel Aggregation) ###

在数据库中，计算聚合是非常昂贵的操作。如果以单个进程进行执行，则这将花费相当长的时间。在PostgreSQL 9.6中，通过简单地将它们分成多个块（分而治之策略）来增加了并行计算的能力。多个worker线程执行聚合的部分，然后leader再根据它们的结果计算最终结果。

从技术上讲，将Partial Aggregate节点添加到计划树中，并且每个Partial Aggregate节点包含一个worker线程的输出。然后将这些输出发送到Finalize Aggregate节点，该节点合并来自多个（所有）Partial Aggregate节点的聚合。如此有效的并行部分计划在根部包括一个Finalize Aggregate节点，以及一个将Partial Aggregate节点作为子节点的Gather节点。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 4]  
‘‘Parallel Seq Scan’‘节点生成用于部分聚合(’‘Partial Aggregate’’)的行。

'‘Partial Aggregate’‘节点使用SUM()函数减少这些行。最后，由’‘Gather’'节点收集每个worker的总和计数。

'‘Finalize Aggregate’‘节点计算最后的总和。如果你使用的自己的聚合函数，别忘了将其标记为’‘parallel safe’'的。

worker数量

可以动态调整worker的数量：

abce=# show max_parallel_workers_per_gather;
     max_parallel_workers_per_gather
    ---------------------------------
     2
    (1 row)
     
    abce=# alter system set max_parallel_workers_per_gather=4;
    ALTER SYSTEM
    abce=# select * from pg_reload_conf();
     pg_reload_conf
    ----------------
     t
    (1 row)
     
    abce=# explain analyze select sum(work_hour) from hh_adds_static_day where create_time <= date '20201010'-interval '10' day;
                                                                           QUERY PLAN                                                                       
    ---------------------------------------------------------------------------------------------------------------------------------------------------------
     Finalize Aggregate  (cost=218981.25..218981.26 rows=1 width=4) (actual time=473.424..475.156 rows=1 loops=1)
       ->  Gather  (cost=218980.83..218981.24 rows=4 width=4) (actual time=473.314..475.144 rows=5 loops=1)
             Workers Planned: 4
             Workers Launched: 4
             ->  Partial Aggregate  (cost=217980.83..217980.84 rows=1 width=4) (actual time=468.769..468.770 rows=1 loops=5)
                   ->  Parallel Seq Scan on hh_adds_static_day  (cost=0.00..215329.59 rows=1060495 width=4) (actual time=0.036..306.854 rows=845622 loops=5)
                         Filter: (create_time <= '2020-09-30 00:00:00'::timestamp without time zone)
                         Rows Removed by Filter: 147120
     Planning Time: 0.150 ms
     Execution Time: 475.218 ms
    (10 rows)

我们将数量从2变成了4。

使用多少个worker进程

首先，max\_parallel\_workers\_per\_gather参数定义了worker的最小数量。

其次，查询执行器从池中获取worker的数量受限于max\_parallel\_workers的值。

最后，最顶层的限制是max\_worker\_processes的值，该参数定义了后台worker进程的总数量。

如果分配worker进程失败，就会切换成单进程执行。

查询计划器会根据表或索引的大小，考虑减少worker进程的数量。受参数min\_parallel\_table\_scan\_size、min\_parallel\_index\_scan\_size影响。

参数的默认设置：

abce=# show min_parallel_table_scan_size;
     min_parallel_table_scan_size
    ------------------------------
     8MB
    (1 row)
     
    abce=# show min_parallel_index_scan_size ;
     min_parallel_index_scan_size
    ------------------------------
     512kB
    (1 row)

影响关系：

set min_parallel_table_scan_size='8MB'
    8MB table => 1 worker
    24MB table => 2 workers
    72MB table => 3 workers
    x => log(x / min_parallel_table_scan_size) / log(3) + 1 worker

每当表的大小是min\_parallel(index|table)scan\_size的三倍，postgresql就会增加一个worker进程。worker进程的数量不是基于cost的。

在实践中，这些规则并不总是被遵守的，可以对特定的表执行alter table … set (parallel\_workers=N)进行设置。

为什么并行执行没有被使用？  
除了并行执行的一些限制，postgresql也检查成本(cost)：

·parallel\_setup\_cost避免了对小的查询采用并行执行。它对用于内存设置、进程启动和初始通信的时间进行建模

·parallel\_tuple\_cost：leader进程和worker进程之间的通信会花费很长的时间。时间与worker发送的元组数量成比例。该参数对通信成本进行建模。

嵌套循环连接(Nested loop joins)

从9.6版本开始，postgresql支持对“Nested loop”执行并行操作：

explain (costs off) select c_custkey, count(o_orderkey)
                    from    customer left outer join orders on
                                    c_custkey = o_custkey and o_comment not like '%special%deposits%'
                    group by c_custkey;
                                          QUERY PLAN                                     
    --------------------------------------------------------------------------------------
     Finalize GroupAggregate
       Group Key: customer.c_custkey
       ->  Gather Merge
             Workers Planned: 4
             ->  Partial GroupAggregate
                   Group Key: customer.c_custkey
                   ->  Nested Loop Left Join
                         ->  Parallel Index Only Scan using customer_pkey on customer
                         ->  Index Scan using idx_orders_custkey on orders
                               Index Cond: (customer.c_custkey = o_custkey)
                               Filter: ((o_comment)::text !~~ '%special%deposits%'::text)

gather发生在最后阶段，因此“Nested Loop Left Join”是并行操作。从版本10开始提供“Parallel Index Only Scan”。其行为与并行顺序扫描类似。条件c\_custkey = o\_custkey为每个客户行读取一个订单。 因此，它不是并行的。

哈希连接(Hash Join)  
在PostgreSQL 11之前，每个worker都构建自己的哈希表。因此，4个以上的workers进程无法提高性能。新的实现使用一个共享哈希表。每个worker都可以利用WORK\_MEM来构建哈希表。

select
            l_shipmode,
            sum(case
                    when o_orderpriority = '1-URGENT'
                            or o_orderpriority = '2-HIGH'
                            then 1
                    else 0
            end) as high_line_count,
            sum(case
                    when o_orderpriority <> '1-URGENT'
                            and o_orderpriority <> '2-HIGH'
                            then 1
                    else 0
            end) as low_line_count
    from
            orders,
            lineitem
    where
            o_orderkey = l_orderkey
            and l_shipmode in ('MAIL', 'AIR')
            and l_commitdate < l_receiptdate
            and l_shipdate < l_commitdate
            and l_receiptdate >= date '1996-01-01'
            and l_receiptdate < date '1996-01-01' + interval '1' year
    group by
            l_shipmode
    order by
            l_shipmode
    LIMIT 1;

-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
     Limit  (cost=1964755.66..1964961.44 rows=1 width=27) (actual time=7579.592..7922.997 rows=1 loops=1)
       ->  Finalize GroupAggregate  (cost=1964755.66..1966196.11 rows=7 width=27) (actual time=7579.590..7579.591 rows=1 loops=1)
             Group Key: lineitem.l_shipmode
             ->  Gather Merge  (cost=1964755.66..1966195.83 rows=28 width=27) (actual time=7559.593..7922.319 rows=6 loops=1)
                   Workers Planned: 4
                   Workers Launched: 4
                   ->  Partial GroupAggregate  (cost=1963755.61..1965192.44 rows=7 width=27) (actual time=7548.103..7564.592 rows=2 loops=5)
                         Group Key: lineitem.l_shipmode
                         ->  Sort  (cost=1963755.61..1963935.20 rows=71838 width=27) (actual time=7530.280..7539.688 rows=62519 loops=5)
                               Sort Key: lineitem.l_shipmode
                               Sort Method: external merge  Disk: 2304kB
                               Worker 0:  Sort Method: external merge  Disk: 2064kB
                               Worker 1:  Sort Method: external merge  Disk: 2384kB
                               Worker 2:  Sort Method: external merge  Disk: 2264kB
                               Worker 3:  Sort Method: external merge  Disk: 2336kB
                               ->  Parallel Hash Join  (cost=382571.01..1957960.99 rows=71838 width=27) (actual time=7036.917..7499.692 rows=62519 loops=5)
                                     Hash Cond: (lineitem.l_orderkey = orders.o_orderkey)
                                     ->  Parallel Seq Scan on lineitem  (cost=0.00..1552386.40 rows=71838 width=19) (actual time=0.583..4901.063 rows=62519 loops=5)
                                           Filter: ((l_shipmode = ANY ('{MAIL,AIR}'::bpchar[])) AND (l_commitdate < l_receiptdate) AND (l_shipdate < l_commitdate) AND (l_receiptdate >= '1996-01-01'::date) AND (l_receiptdate < '1997-01-01 00:00:00'::timestamp without time zone))
                                           Rows Removed by Filter: 11934691
                                     ->  Parallel Hash  (cost=313722.45..313722.45 rows=3750045 width=20) (actual time=2011.518..2011.518 rows=3000000 loops=5)
                                           Buckets: 65536  Batches: 256  Memory Usage: 3840kB
                                           ->  Parallel Seq Scan on orders  (cost=0.00..313722.45 rows=3750045 width=20) (actual time=0.029..995.948 rows=3000000 loops=5)
     Planning Time: 0.977 ms
     Execution Time: 7923.770 ms

这里每个worker帮助构建一个共享的hash表。

合并连接(Merge Join)  
鉴于合并连接的自身特性，使其不支持并行查询。如果合并连接是查询执行的最后一个阶段-你仍可以看到并行执行。

-- Query 2 from TPC-H
    explain (costs off) select s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
    from    part, supplier, partsupp, nation, region
    where
            p_partkey = ps_partkey
            and s_suppkey = ps_suppkey
            and p_size = 36
            and p_type like '%BRASS'
            and s_nationkey = n_nationkey
            and n_regionkey = r_regionkey
            and r_name = 'AMERICA'
            and ps_supplycost = (
                    select
                            min(ps_supplycost)
                    from    partsupp, supplier, nation, region
                    where
                            p_partkey = ps_partkey
                            and s_suppkey = ps_suppkey
                            and s_nationkey = n_nationkey
                            and n_regionkey = r_regionkey
                            and r_name = 'AMERICA'
            )
    order by s_acctbal desc, n_name, s_name, p_partkey
    LIMIT 100;
                                                    QUERY PLAN                                               
    ----------------------------------------------------------------------------------------------------------
     Limit
       ->  Sort
             Sort Key: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
             ->  Merge Join
                   Merge Cond: (part.p_partkey = partsupp.ps_partkey)
                   Join Filter: (partsupp.ps_supplycost = (SubPlan 1))
                   ->  Gather Merge
                         Workers Planned: 4
                         ->  Parallel Index Scan using <strong>part_pkey</strong> on part
                               Filter: (((p_type)::text ~~ '%BRASS'::text) AND (p_size = 36))
                   ->  Materialize
                         ->  Sort
                               Sort Key: partsupp.ps_partkey
                               ->  Nested Loop
                                     ->  Nested Loop
                                           Join Filter: (nation.n_regionkey = region.r_regionkey)
                                           ->  Seq Scan on region
                                                 Filter: (r_name = 'AMERICA'::bpchar)
                                           ->  Hash Join
                                                 Hash Cond: (supplier.s_nationkey = nation.n_nationkey)
                                                 ->  Seq Scan on supplier
                                                 ->  Hash
                                                       ->  Seq Scan on nation
                                     ->  Index Scan using idx_partsupp_suppkey on partsupp
                                           Index Cond: (ps_suppkey = supplier.s_suppkey)
                   SubPlan 1
                     ->  Aggregate
                           ->  Nested Loop
                                 Join Filter: (nation_1.n_regionkey = region_1.r_regionkey)
                                 ->  Seq Scan on region region_1
                                       Filter: (r_name = 'AMERICA'::bpchar)
                                 ->  Nested Loop
                                       ->  Nested Loop
                                             ->  Index Scan using idx_partsupp_partkey on partsupp partsupp_1
                                                   Index Cond: (part.p_partkey = ps_partkey)
                                             ->  Index Scan using supplier_pkey on supplier supplier_1
                                                   Index Cond: (s_suppkey = partsupp_1.ps_suppkey)
                                       ->  Index Scan using nation_pkey on nation nation_1
                                             Index Cond: (n_nationkey = supplier_1.s_nationkey)

“Merge Join”节点在“Gather Merge”上方。 因此，合并不使用并行执行。但是“Parallel Index Scan”节点仍然有助于part\_pkey。

分区智能连接(Partition-wise join)

PostgreSQL 11默认禁用分区智能连接(partition-wise join)功能。分区智能联接的计划成本很高。分区相似的表的联接可以逐分区进行。这允许postgres使用较小的哈希表。每个分区联接操作都可以并行执行。

tpch=# set enable_partitionwise_join=t;
    tpch=# explain (costs off) select * from prt1 t1, prt2 t2
    where t1.a = t2.b and t1.b = 0 and t2.b between 0 and 10000;
                        QUERY PLAN                    
    ---------------------------------------------------
     Append
       ->  Hash Join
             Hash Cond: (t2.b = t1.a)
             ->  Seq Scan on prt2_p1 t2
                   Filter: ((b >= 0) AND (b <= 10000))
             ->  Hash
                   ->  Seq Scan on prt1_p1 t1
                         Filter: (b = 0)
       ->  Hash Join
             Hash Cond: (t2_1.b = t1_1.a)
             ->  Seq Scan on prt2_p2 t2_1
                   Filter: ((b >= 0) AND (b <= 10000))
             ->  Hash
                   ->  Seq Scan on prt1_p2 t1_1
                         Filter: (b = 0)
    tpch=# set parallel_setup_cost = 1;
    tpch=# set parallel_tuple_cost = 0.01;
    tpch=# explain (costs off) select * from prt1 t1, prt2 t2
    where t1.a = t2.b and t1.b = 0 and t2.b between 0 and 10000;
                            QUERY PLAN                        
    -----------------------------------------------------------
     Gather
       Workers Planned: 4
       ->  Parallel Append
             ->  Parallel Hash Join
                   Hash Cond: (t2_1.b = t1_1.a)
                   ->  Parallel Seq Scan on prt2_p2 t2_1
                         Filter: ((b >= 0) AND (b <= 10000))
                   ->  Parallel Hash
                         ->  Parallel Seq Scan on prt1_p2 t1_1
                               Filter: (b = 0)
             ->  Parallel Hash Join
                   Hash Cond: (t2.b = t1.a)
                   ->  Parallel Seq Scan on prt2_p1 t2
                         Filter: ((b >= 0) AND (b <= 10000))
                   ->  Parallel Hash
                         ->  Parallel Seq Scan on prt1_p1 t1
                               Filter: (b = 0)

最重要的是，仅在分区足够大的情况下，分区智能联接才能使用并行执行。

并行追加(Parallel Append)  
通常可以在UNION ALL查询中看到这一点。缺点是并行性差，因为每个work最终都是为单个查询工作。

即使启用了四个worker，也只有两个被启动。

tpch=# explain (costs off) select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day union all select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '2000-12-01' - interval '105' day;
                                               QUERY PLAN                                          
    ------------------------------------------------------------------------------------------------
     Gather
       Workers Planned: 2
       ->  Parallel Append
             ->  Aggregate
                   ->  Seq Scan on lineitem
                         Filter: (l_shipdate <= '2000-08-18 00:00:00'::timestamp without time zone)
             ->  Aggregate
                   ->  Seq Scan on lineitem lineitem_1
                         Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp without time zone)

重要的变量

·work\_mem

·max\_parallel\_workers\_per\_gather：执行器为每个并行执行的计划节点分配的worker数量

·max\_worker\_processes

·max\_parallel\_workers

在9.6版本中，并行查询执行被引入；

在10版本中，默认开启并行执行；

在负载重的oltp系统上，建议关闭并行执行。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 5]  
https://zhuanlan.zhihu.com/p/413491023

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_18_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/59a626d13e6a44d4a587a1722e0bd947.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/b3f07caf622349f2b4f914a36325e569.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/7d6eb45492e945f593266f49982e75a5.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/bf04a5ccb24e4616a316efa970f48930.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_17_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/05969c2681db4c30a2930bf90dc73750.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220828/27a3807e50e548be9a4ff1725de0b098.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220828/95134c2028cc46d3b8c608d98bac569c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220828/a926b5311189446fa947de427604b639.png