数据库内核月报－ 2022/07 - PostgreSQL · 插件特性 · FDW 异步执行特性 - 《数据库内核月报》

Foreign Data Wrapper（FDW）是 PostgreSQL 提供的一个非常有意思的特性，中文翻译为 外部数据包装器。从字面意思上，PostgreSQL 数据库能够通过 FDW 扩展来操作当前数据库以外的数据。这些外部的数据源可以是：

文件
关系型数据库（PostgreSQL / Oracle / MySQL / …）
非关系型数据库
Git 仓库
网页
大数据平台（Hadoop / Hive / …）
…（尽情遐想）😏

PostgreSQL 内核中定义了 FDW 回调函数接口，这个接口在形式上是一个装满了函数指针的结构体 FdwRoutine。这些函数指针会在 PostgreSQL 内核的优化器和执行器的关键位置上被回调，以完成操纵外部数据。服务于每一种外部数据源的 FDW 都需要提供一个 FdwRoutine 结构体。如果要对一种新的外部数据源实现 FDW 扩展，主要工作就是填充 FdwRoutine 结构体中的函数指针，实现对这种外部数据源的操作逻辑。

由于涉及到操纵数据库外部的数据，一个不可避免的因素出现了：性能。与操作数据库内的数据不同，操纵外部数据可能会带来额外的网络 I/O 开销和计算开销（序列化/反序列化）。受制于目前 PostgreSQL 内核执行器的执行模型，使用 FDW 操纵外部数据使用了与操纵本地数据一致的 同步执行模型，在性能有很大的提升空间。

PostgreSQL 14 首次为 FDW 引入了 异步执行模型。虽然目前支持的场景较为有限，但依旧能够有效并行化一些操纵远程数据的场景，提升执行效率。本文将从原理和实现上对异步执行功能进行分析。源码版本基于 PostgreSQL 14 的稳定分支 REL_14_STABLE，commit 号截止至：


1. commit fb81a93a6442e55d8c7376a01c27cb5d6c062c80
2. Author: Thomas Munro <tmunro@postgresql.org>
3. Date:   Fri Jul 1 12:05:52 2022 +1200

Background

FDW

FDW 是 PostgreSQL 提供的一个特性，它能够让用户在 PostgreSQL 中创建 外部表（Foreign Table）。外部表将被作为代理，用于访问外部数据源。当用户对外部表发起查询时，FDW 会负责把查询进行一定的形式转换后访问外部数据源，并负责将外部数据源返回的数据转换回 PostgreSQL 的结果形式，让用户觉得查询一个外部数据源好像和查询一个数据库内的普通表一样没什么区别。目前 PostgreSQL 官方提供了两个 FDW 实现：

file_fdw：使用户能够创建代表普通文件的外部表
postgres_fdw：使用户能够创建代表另一个 PostgreSQL 数据库表的外部表

其余大量的第三方 FDW 插件及其源码可以参考这里，它们都实现了 FDW 对外暴露出的函数接口 FdwRoutine：


1. /*
2. * FdwRoutine is the struct returned by a foreign-data wrapper's handler
3. * function.  It provides pointers to the callback functions needed by the
4. * planner and executor.
5. *
6. * More function pointers are likely to be added in the future.  Therefore
7. * it's recommended that the handler initialize the struct with
8. * makeNode(FdwRoutine) so that all fields are set to NULL.  This will
9. * ensure that no fields are accidentally left undefined.
10. */
11. typedef struct FdwRoutine
12. {
13. NodeTag     type;

15. /* Functions for scanning foreign tables */
16. GetForeignRelSize_function GetForeignRelSize;
17. GetForeignPaths_function GetForeignPaths;
18. GetForeignPlan_function GetForeignPlan;
19. BeginForeignScan_function BeginForeignScan;
20. IterateForeignScan_function IterateForeignScan;
21. ReScanForeignScan_function ReScanForeignScan;
22. EndForeignScan_function EndForeignScan;

24. /*
25. * Remaining functions are optional.  Set the pointer to NULL for any that
26. * are not provided.
27. */

29. /* Functions for remote-join planning */
30. GetForeignJoinPaths_function GetForeignJoinPaths;

32. /* Functions for remote upper-relation (post scan/join) planning */
33. GetForeignUpperPaths_function GetForeignUpperPaths;

35. /* Functions for updating foreign tables */
36. AddForeignUpdateTargets_function AddForeignUpdateTargets;
37. PlanForeignModify_function PlanForeignModify;
38. BeginForeignModify_function BeginForeignModify;
39. ExecForeignInsert_function ExecForeignInsert;
40. ExecForeignBatchInsert_function ExecForeignBatchInsert;
41. GetForeignModifyBatchSize_function GetForeignModifyBatchSize;
42. ExecForeignUpdate_function ExecForeignUpdate;
43. ExecForeignDelete_function ExecForeignDelete;
44. EndForeignModify_function EndForeignModify;
45. BeginForeignInsert_function BeginForeignInsert;
46. EndForeignInsert_function EndForeignInsert;
47. IsForeignRelUpdatable_function IsForeignRelUpdatable;
48. PlanDirectModify_function PlanDirectModify;
49. BeginDirectModify_function BeginDirectModify;
50. IterateDirectModify_function IterateDirectModify;
51. EndDirectModify_function EndDirectModify;

53. /* Functions for SELECT FOR UPDATE/SHARE row locking */
54. GetForeignRowMarkType_function GetForeignRowMarkType;
55. RefetchForeignRow_function RefetchForeignRow;
56. RecheckForeignScan_function RecheckForeignScan;

58. /* Support functions for EXPLAIN */
59. ExplainForeignScan_function ExplainForeignScan;
60. ExplainForeignModify_function ExplainForeignModify;
61. ExplainDirectModify_function ExplainDirectModify;

63. /* Support functions for ANALYZE */
64. AnalyzeForeignTable_function AnalyzeForeignTable;

66. /* Support functions for IMPORT FOREIGN SCHEMA */
67. ImportForeignSchema_function ImportForeignSchema;

69. /* Support functions for TRUNCATE */
70. ExecForeignTruncate_function ExecForeignTruncate;

72. /* Support functions for parallelism under Gather node */
73. IsForeignScanParallelSafe_function IsForeignScanParallelSafe;
74. EstimateDSMForeignScan_function EstimateDSMForeignScan;
75. InitializeDSMForeignScan_function InitializeDSMForeignScan;
76. ReInitializeDSMForeignScan_function ReInitializeDSMForeignScan;
77. InitializeWorkerForeignScan_function InitializeWorkerForeignScan;
78. ShutdownForeignScan_function ShutdownForeignScan;

80. /* Support functions for path reparameterization. */
81. ReparameterizeForeignPathByChild_function ReparameterizeForeignPathByChild;

83. /* Support functions for asynchronous execution */
84. IsForeignPathAsyncCapable_function IsForeignPathAsyncCapable;
85. ForeignAsyncRequest_function ForeignAsyncRequest;
86. ForeignAsyncConfigureWait_function ForeignAsyncConfigureWait;
87. ForeignAsyncNotify_function ForeignAsyncNotify;
88. } FdwRoutine;

如注释所描述，这些函数指针会在 PostgreSQL 内核的优化器和执行器中被回调，完成 FDW 所需要实现的功能。比如 postgres_fdw 就会在这些回调函数中使用 libpq（PostgreSQL 客户端与服务端进程交互的 API 协议 C 语言库）与一个远程的 PostgreSQL 数据库建立连接并获取数据。FDW 能够在一定程度上体现 PostgreSQL 的可扩展性。

Execution Model of PostgreSQL Executor

PostgreSQL 内核的执行器使用 迭代器模型（又称火山模型，流水线模型）设计。与多数编程语言使用迭代器的方式类似，执行器包含了三个执行阶段：

初始化（打开迭代器）
执行（如果还有更多数据，则获取数据）
清理（关闭迭代器）

对照 Java 迭代器：


1. List<String> items = new ArrayList<>();
2. Iterator<String> iter = items.iterator();
3. while (iter.hasNext()) {
4. String next = iter.next();
5. }
6. // iter close is done by GC

执行器的每一个物理算子都会实现这三个函数。当 PostgreSQL 的 planner 输出一棵物理计划树以后，在执行器初始化阶段，执行器会从计划树根节点的物理算子开始，调用物理算子的初始化函数；这个物理算子的初始化函数又会递归调用其下层算子的初始化函数，直到执行流到达计划树的叶子节点。执行器执行阶段和清理阶段的逻辑也采用了类似的递归形式实现。

PostgreSQL 内核中的绝大多数物理算子都是一元或二元的，即该算子需要从自己的一个或两个下层物理算子（孩子节点）中获取元组，并完成在当前物理算子中需要完成的事。物理算子是一元还是二元可以通过 EXPLAIN 命令从执行计划中看出来。

比如说 Sort（排序）算子就是一个一元算子，它只需要从它的下层物理算子获取所有元组，并根据排序的 key 完成排序，然后将排序后的第一个元组返回上层物理算子即可：

在 PostgreSQL EXPLAIN 打印的物理计划中，-> 的缩进层次对应了物理计划树的层级关系。


1. ->  Sort  (cost=5726599.94..5789078.79 rows=24991540 width=9)
2. Sort Key: lineitem_1.l_orderkey
3. ->  Parallel Seq Scan on lineitem lineitem_1  (cost=0.00..1374457.40 rows=24991540 width=9)

而 Hash Join（哈希连接）算子就是一个二元算子，它需要先从一个下层物理算子中获取所有元组并构造哈希表，然后从另一个下层物理算子中依次获取元组，并在哈希表中进行哈希探测和连接：


1. ->  Hash Join  (cost=6032162.96..6658785.84 rows=138372 width=24)
2. Hash Cond: (orders.o_orderkey = lineitem_1.l_orderkey)
3. ->  Seq Scan on orders  (cost=0.00..410917.44 rows=15000544 width=20)
4. ->  Hash  (cost=6029892.31..6029892.31 rows=138372 width=4)

Append Operator

Append 是所有算子中的一个奇葩，它破坏了上述计划树类似二叉树的结构。让我们试想，假设要扫描一个带有多于两个子分区的分区表，如果只能产生一棵二叉树，那么会产生一个什么样的计划树呢？形状一定会很奇怪吧。😅

让我们看看目前 PostgreSQL 对分区表扫描的执行计划吧：


1. QUERY PLAN
2. ----------------------------------------------------------------------
3. Append  (cost=0.00..307.30 rows=15820 width=8)
4. ->  Seq Scan on mc2p0  (cost=0.00..32.60 rows=2260 width=8)
5. ->  Seq Scan on mc2p1  (cost=0.00..32.60 rows=2260 width=8)
6. ->  Seq Scan on mc2p2  (cost=0.00..32.60 rows=2260 width=8)
7. ->  Seq Scan on mc2p3  (cost=0.00..32.60 rows=2260 width=8)
8. ->  Seq Scan on mc2p4  (cost=0.00..32.60 rows=2260 width=8)
9. ->  Seq Scan on mc2p5  (cost=0.00..32.60 rows=2260 width=8)
10. ->  Seq Scan on mc2p_default  (cost=0.00..32.60 rows=2260 width=8)

Append 算子就是用来解决这个问题的，它可以有多于两个下层物理算子。事实上，在 Append 算子 src/backend/executor/nodeAppend.c 的注释中就可以了解到，Append 算子并不使用其它物理算子所使用的 左右孩子指针 来引用下层算子，而是持有一个链表，链表中包含了它需要引用的所有下层物理算子，个数不受限。下层物理算子的子树被称为子计划：


1. /*
2. *   NOTES
3. *      Each append node contains a list of one or more subplans which
4. *      must be iteratively processed (forwards or backwards).
5. *      Tuples are retrieved by executing the 'whichplan'th subplan
6. *      until the subplan stops returning tuples, at which point that
7. *      plan is shut down and the next started up.
8. *
9. *      Append nodes don't make use of their left and right
10. *      subtrees, rather they maintain a list of subplans so
11. *      a typical append node looks like this in the plan tree:
12. *
13. *                 ...
14. *                 /
15. *              Append -------+------+------+--- nil
16. *              /   \         |      |      |
17. *            nil   nil      ...    ...    ...
18. *                               subplans
19. */

如注释所示意的，在执行器的执行阶段，Append 算子会依次从它的下层物理算子（子计划）中获取元组并返回上层算子，当一个子计划中的数据枯竭（返回 NULL）后，就开始从下一个子计划开始获取，直到压榨完链表中所有的子计划为止。

因为 Append 算子的特殊性，它被广泛用于需要将来自多方的数据进行融合的场合，包括 FDW。比如说一个分区表的子分区分布在多个 PostgreSQL 实例上，就需要通过 Append 算子来做汇聚：


1. QUERY PLAN
2. ----------------------------------------------------------------------------------------
3. Aggregate  (cost=8882.01..8882.02 rows=1 width=8)
4. Output: count(*)
5. ->  Append  (cost=100.00..8382.00 rows=200001 width=0)
6. ->  Foreign Scan on public.p1 pt_1  (cost=100.00..3641.00 rows=100000 width=0)
7. Remote SQL: SELECT NULL FROM public.loct1
8. ->  Foreign Scan on public.p2 pt_2  (cost=100.00..3641.00 rows=100000 width=0)
9. Remote SQL: SELECT NULL FROM public.loct2
10. ->  Foreign Scan on public.p3 pt_3  (cost=100.00..100.00 rows=1 width=0)
11. Remote SQL: SELECT NULL FROM public.loct3
12. (9 rows)

这里我们就需要注意了：Append 算子是依次 串行同步 地执行每一个子计划的。当其中某个子计划与其在数据库内执行相比需要执行较长时间（通过网络 I/O 外发请求 + 在远程机器上执行）时，FDW 的性能是相当低的。这也是之前 PostgreSQL 执行模型的一个局限：只考虑了如何执行数据库内的操作，并没有为 FDW 这种场景做专门的优化。

Asynchronized Execution

经过上面的分析可以发现，在 FDW 场景中，PostgreSQL 的执行模型有着很大的提升空间。原先，一个子计划需要等待前一个子计划执行完毕以后才可以开始执行。那么能不能一次性让所有的外库子计划全部开始执行，然后开始同步执行一部分库内的计划，同时等待所有外库子计划的执行结果呢？这样所有外库都在并行执行查询，而等待外库执行结果的时间又可以被用于执行库内的计划，从而能够极大提升执行效率。

这就是 PostgreSQL 14 引入的 FDW 异步执行 特性的设计思路。目前这一特性暂时只被实现在了 Append 算子与其直接下游 ForeignScan 算子之间，且暂时仅支持 postgres_fdw 的异步执行。未来或许会再做扩展。其主要的实现思路为：

在 Append 算子的实现 src/backend/executor/nodeAppend.c 中，引入对同步子计划和异步子计划的分别处理
ExecAppendAsyncBegin()
ExecAppendAsyncGetNext()
ExecAppendAsyncRequest()
ExecAppendAsyncEventWait()
ExecAsyncAppendResponse()
在 src/backend/executor/execAsync.c 中，引入了异步执行的抽象中间层
ExecAsyncRequest()
ExecAsyncConfigureWait()
ExecAsyncNotify()
ExecAsyncResponse()
ExecAsyncRequestPending()
ExecAsyncRequestDone()
在外部表扫描层 src/backend/executor/nodeForeignScan.c 中，将执行流引入 FDW 的异步执行函数中
ExecAsyncForeignScanRequest()
ExecAsyncForeignScanConfigureWait()
ExecAsyncForeignScanNotify()
在 postgres_fdw 的实现 contrib/postgres_fdw/postgres_fdw.c 中，实现支持异步执行的接口函数
postgresForeignAsyncRequest()
postgresForeignAsyncConfigureWait()
postgresForeignAsyncNotify()

从上述函数命名可以看出，FDW 的异步执行功能实现在四个层次上。其中，上层的 Append 算子会调用异步抽象层的函数，异步抽象层的函数进而调用 ForeignScan 算子的函数，再进而调用 postgres_fdw 的异步执行函数。整体操作分为三个步骤：

异步发送请求（只发送请求，不等待结果）
配置等待事件集合，轮询等待事件
等待事件触发后，封装执行结果并返回给上层（Append）算子

函数调用关系为：

接下来从上到下对每一层的代码进行分析。

源代码中的 // 注释是本人添加的说明。由于 PostgreSQL 内核中不允许出现这种风格的注释，我正好使用这种注释风格与原有内核注释加以区别。

Append 算子

首先，FDW 异步执行特性引入了一个新的选项 async_capable。它同时是一个服务器级别和表级别的选项，指示对应的数据库服务器和服务器上的表是否可以被异步执行。表级别的选项将会覆盖服务器级别的选项。只要外部表的这个参数被设置为 true，那么优化器就会产生异步执行的计划树：


1. QUERY PLAN
2. ----------------------------------------------------------------------------------------------
3. Aggregate  (cost=8882.01..8882.02 rows=1 width=8)
4. Output: count(*)
5. ->  Append  (cost=100.00..8382.00 rows=200001 width=0)
6. ->  Async Foreign Scan on public.p1 pt_1  (cost=100.00..3641.00 rows=100000 width=0)
7. Remote SQL: SELECT NULL FROM public.loct1
8. ->  Async Foreign Scan on public.p2 pt_2  (cost=100.00..3641.00 rows=100000 width=0)
9. Remote SQL: SELECT NULL FROM public.loct2
10. ->  Async Foreign Scan on public.p3 pt_3  (cost=100.00..100.00 rows=1 width=0)
11. Remote SQL: SELECT NULL FROM public.loct3
12. (9 rows)

ExecInitAppend

在 Append 算子的初始化阶段，即函数 ExecInitAppend() 中，需要为其下层所有的子计划分配好 PlanState 指针数组以便能够引用下层算子，然后递归调用每一个下层算子的初始化函数。在这个过程中，顺便统计一下哪几个子计划是可以被异步执行的：


1. // 分配指向所有下层算子 PlanState 的指针数组
2. appendplanstates = (PlanState **) palloc(nplans *
3. sizeof(PlanState *));

5. /*
6. * call ExecInitNode on each of the valid plans to be executed and save
7. * the results into the appendplanstates array.
8. *
9. * While at it, find out the first valid partial plan.
10. */
11. j = 0;
12. asyncplans = NULL;
13. nasyncplans = 0;
14. firstvalid = nplans;
15. i = -1;

17. // 遍历每一个合法子计划，递归调用每一个下层算子的初始化函数
18. while ((i = bms_next_member(validsubplans, i)) >= 0)
19. {
20. Plan       *initNode = (Plan *) list_nth(node->appendplans, i);

22. /*
23. * Record async subplans.  When executing EvalPlanQual, we treat them
24. * as sync ones; don't do this when initializing an EvalPlanQual plan
25. * tree.
26. */
27. // 在 bitmap 中记录异步子计划的个数和位置
28. if (initNode->async_capable && estate->es_epq_active == NULL)
29. {
30. asyncplans = bms_add_member(asyncplans, j);
31. nasyncplans++;
32. }

34. /*
35. * Record the lowest appendplans index which is a valid partial plan.
36. */
37. // 记录好第一个将要被执行的下层算子
38. if (i >= node->first_partial_plan && j < firstvalid)
39. firstvalid = j;

41. // 把下层算子构建完成的 PlanState 记录到刚才分配的指针数组中
42. appendplanstates[j++] = ExecInitNode(initNode, estate, eflags);
43. }

在明确哪些子计划需要被异步执行以后，还需要为每一个异步子计划分配一个 AsyncRequest 结构体。这个结构体被用于与下层算子传递请求状态和执行结果：


1. // 如果有异步执行的子计划
2. if (nasyncplans > 0)
3. {
4. // 分配所有子计划的 AsyncRequest 的指针数组
5. appendstate->as_asyncrequests = (AsyncRequest **)
6. palloc0(nplans * sizeof(AsyncRequest *));

8. // 通过 bitmap 找到每一个需要被异步执行的子计划
9. i = -1;
10. while ((i = bms_next_member(asyncplans, i)) >= 0)
11. {
12. AsyncRequest *areq;

14. // 分配 AsyncRequest 结构体内存
15. areq = palloc(sizeof(AsyncRequest));
16. // 数据请求方：Append 算子
17. areq->requestor = (PlanState *) appendstate;
18. // 被请求方：Append 算子的下层算子（ForeignScan 算子）
19. areq->requestee = appendplanstates[i];
20. areq->request_index = i;

22. // AsyncRequest 中保存的状态和结果的初始化
23. areq->callback_pending = false;
24. areq->request_complete = false;
25. areq->result = NULL;

27. appendstate->as_asyncrequests[i] = areq;
28. }

30. // 为异步子计划分配元组缓存槽
31. appendstate->as_asyncresults = (TupleTableSlot **)
32. palloc0(nasyncplans * sizeof(TupleTableSlot *));

34. // 对合法的同步子计划和异步子计划进行分类
35. // 分别保存到不同的 bitmap 中
36. if (appendstate->as_valid_subplans != NULL)
37. classify_matching_subplans(appendstate);
38. }

ExecAppend

当 Append 算子第一次被执行时，将会立刻开启所有异步子计划的执行。当然，由于是异步的，所以只是把请求发出去了，并不阻塞等待结果。通过调用异步抽象层的 ExecAppendAsyncBegin() 完成：


1. /*
2. * If this is the first call after Init or ReScan, we need to do the
3. * initialization work.
4. */
5. // 第一次调用 ExecAppend，需要选择第一个子计划开始执行
6. if (!node->as_begun)
7. {
8. Assert(node->as_whichplan == INVALID_SUBPLAN_INDEX);
9. Assert(!node->as_syncdone);

11. /* Nothing to do if there are no subplans */
12. if (node->as_nplans == 0)
13. return ExecClearTuple(node->ps.ps_ResultTupleSlot);

15. // 如果有异步子计划，立刻开始执行
16. /* If there are any async subplans, begin executing them. */
17. if (node->as_nasyncplans > 0)
18. ExecAppendAsyncBegin(node);

20. // 选择下一个要被同步执行的计划
21. // 如果同步子计划和异步子计划都没有了，就向上层算子返回空槽，Append 的执行结束
22. /*
23. * If no sync subplan has been chosen, we must choose one before
24. * proceeding.
25. */
26. if (!node->choose_next_subplan(node) && node->as_nasyncremain == 0)
27. return ExecClearTuple(node->ps.ps_ResultTupleSlot);

29. Assert(node->as_syncdone ||
30. (node->as_whichplan >= 0 &&
31. node->as_whichplan < node->as_nplans));

33. // 第一次 ExecAppend 的初始化完毕，下次不再进入
34. /* And we're initialized. */
35. node->as_begun = true;
36. }

初始化结束后，就是不断从子计划中递归获取元组了。让我们看看异步执行做了什么样的改造：


1. // 接上一段代码的初始化，目前 ExecAppend 的初始化已结束
2. // 应该已经选中了一个子计划准备开始执行
3. // 从当前选中的子计划开始不断调用 ExecProcNode 递归获取元组
4. for (;;)
5. {
6. PlanState  *subnode;

8. CHECK_FOR_INTERRUPTS();

10. // 首先，尝试从刚才已经开启执行的异步子计划中获取一个元组
11. // 如果能够获取到一个元组，那么立刻返回
12. /*
13. * try to get a tuple from an async subplan if any
14. */
15. if (node->as_syncdone || !bms_is_empty(node->as_needrequest))
16. {
17. if (ExecAppendAsyncGetNext(node, &result))
18. return result;
19. Assert(!node->as_syncdone);
20. Assert(bms_is_empty(node->as_needrequest));
21. }

23. // 没能从任何异步子计划中获取到元组
24. // 那么找到目前正在被执行的同步子计划
25. /*
26. * figure out which sync subplan we are currently processing
27. */
28. Assert(node->as_whichplan >= 0 && node->as_whichplan < node->as_nplans);
29. subnode = node->appendplans[node->as_whichplan];

31. // 从同步子计划中获取一个元组
32. /*
33. * get a tuple from the subplan
34. */
35. result = ExecProcNode(subnode);

37. // 同步子计划的结果非空，那么返回结果
38. if (!TupIsNull(result))
39. {
40. /*
41. * If the subplan gave us something then return it as-is. We do
42. * NOT make use of the result slot that was set up in
43. * ExecInitAppend; there's no need for it.
44. */
45. return result;
46. }

48. // 同步子计划结果为空，
49. // 说明这个同步子计划执行结束，需要选择下一个被执行的同步子计划

51. // 在结束这个同步子计划的迭代前
52. // 先对异步子计划的等待事件集合来一次轮询
53. // 这样下一次循环或许可以从异步子计划中获取到元组
54. /*
55. * wait or poll for async events if any. We do this before checking
56. * for the end of iteration, because it might drain the remaining
57. * async subplans.
58. */
59. if (node->as_nasyncremain > 0)
60. // 轮询，并处理触发的事件
61. ExecAppendAsyncEventWait(node);

63. // 如果同步子计划和异步子计划全部执行完毕，那么返回空槽结束 Append 算子的执行
64. // 否则指向下一个将要被执行的同步子计划
65. /* choose new sync subplan; if no sync/async subplans, we're done */
66. if (!node->choose_next_subplan(node) && node->as_nasyncremain == 0)
67. return ExecClearTuple(node->ps.ps_ResultTupleSlot);

69. // 进入下一次循环
70. }

然后我们进一步分析上面的代码中用到的几个异步执行函数。

ExecAppendAsyncBegin

该函数用于向所有合法的异步子计划请求元组。由于是异步执行，因此只管发送请求，不管接收结果。其中将会调用到下一层（也就是异步抽象层）的 ExecAsyncRequest() 函数：


1. /* ----------------------------------------------------------------
2. *      ExecAppendAsyncBegin
3. *
4. *      Begin executing designed async-capable subplans.
5. * ----------------------------------------------------------------
6. */
7. static void
8. ExecAppendAsyncBegin(AppendState *node)
9. {
10. // ...

12. /* Initialize state variables. */
13. node->as_syncdone = bms_is_empty(node->as_valid_subplans);
14. node->as_nasyncremain = bms_num_members(node->as_valid_asyncplans);

16. // 如果没有任何异步子计划需要被执行，那么直接返回
17. /* Nothing to do if there are no valid async subplans. */
18. if (node->as_nasyncremain == 0)
19. return;

21. // 通过 bitmap 遍历每一个合法的异步子计划
22. /* Make a request for each of the valid async subplans. */
23. i = -1;
24. while ((i = bms_next_member(node->as_valid_asyncplans, i)) >= 0)
25. {
26. AsyncRequest *areq = node->as_asyncrequests[i];

28. Assert(areq->request_index == i);
29. Assert(!areq->callback_pending);

31. /* Do the actual work. */
32. // 调用异步抽象层的函数，向异步子计划请求元组
33. ExecAsyncRequest(areq);
34. }
35. }

ExecAppendAsyncGetNext

该函数从任意一个异步子计划中获取下一个元组，并保存到输入参数中的元组缓存槽 result 中。如果函数结束时元组缓存槽中包含了有效结果，那么函数返回 true，否则返回 false。


1. /* ----------------------------------------------------------------
2. *      ExecAppendAsyncGetNext
3. *
4. *      Get the next tuple from any of the asynchronous subplans.
5. * ----------------------------------------------------------------
6. */
7. static bool
8. ExecAppendAsyncGetNext(AppendState *node, TupleTableSlot **result)
9. {
10. *result = NULL;

12. /* We should never be called when there are no valid async subplans. */
13. Assert(node->as_nasyncremain > 0);

15. // 异步地请求子计划
16. // 如果该函数在 result 中产生了有效结果，那么直接返回 true
17. /* Request a tuple asynchronously. */
18. if (ExecAppendAsyncRequest(node, result))
19. return true;

21. // 没有产生有效结果

23. // 还有异步子计划没执行完
24. while (node->as_nasyncremain > 0)
25. {
26. CHECK_FOR_INTERRUPTS();

28. // 轮询等待事件
29. // 如果有就绪事件，并触发回调并保存执行结果
30. /* Wait or poll for async events. */
31. ExecAppendAsyncEventWait(node);

33. // 异步地请求子计划
34. // 如果该函数在 result 中产生了有效结果，那么直接返回 true
35. /* Request a tuple asynchronously. */
36. if (ExecAppendAsyncRequest(node, result))
37. return true;

39. // 没能从异步子计划中获取到任何元组
40. // 如果目前还有同步计划没执行完，那么跳出循环，先去执行同步子计划
41. /* Break from loop if there's any sync subplan that isn't complete. */
42. if (!node->as_syncdone)
43. break;
44. // 如果同步子计划已全部执行完，只剩下异步子计划了
45. // 那么循环从这里绕回，继续等待异步子计划的执行结果
46. }

48. // 至此，异步计划已全部执行完
49. /*
50. * If all sync subplans are complete, we're totally done scanning the
51. * given node.  Otherwise, we're done with the asynchronous stuff but must
52. * continue scanning the sync subplans.
53. */
54. // 如果同步计划也全都执行完了，那么整个 Append 算子的执行就结束了，向上级算子返回空槽
55. if (node->as_syncdone)
56. {
57. Assert(node->as_nasyncremain == 0);
58. *result = ExecClearTuple(node->ps.ps_ResultTupleSlot);
59. return true;
60. }

62. // 这一轮调用没能得到任何有效结果，返回 false
63. return false;
64. }

ExecAppendAsyncRequest

该函数用于异步地请求一个元组，也会调用异步抽象层的 ExecAsyncRequest()。同样，如果函数结束时元组缓存槽 result 中包含了有效结果，那么函数返回 true，否则返回 false。


1. /* ----------------------------------------------------------------
2. *      ExecAppendAsyncRequest
3. *
4. *      Request a tuple asynchronously.
5. * ----------------------------------------------------------------
6. */
7. static bool
8. // 如果调用中包含有效结果，那么返回 true
9. ExecAppendAsyncRequest(AppendState *node, TupleTableSlot **result)
10. {
11. Bitmapset  *needrequest;
12. int         i;

14. // 如果不需要向任何异步子计划发送请求，那么直接返回
15. /* Nothing to do if there are no async subplans needing a new request. */
16. if (bms_is_empty(node->as_needrequest))
17. {
18. Assert(node->as_nasyncresults == 0);
19. return false;
20. }

22. // 如果已有异步计划产生结果了
23. // 那么使返回结果槽指向对应异步子计划的结果槽，然后直接返回
24. /*
25. * If there are any asynchronously-generated results that have not yet
26. * been returned, we have nothing to do; just return one of them.
27. */
28. if (node->as_nasyncresults > 0)
29. {
30. --node->as_nasyncresults;
31. *result = node->as_asyncresults[node->as_nasyncresults];
32. return true;
33. }

35. // 至此，说明所有异步子计划的结果槽都已经空了

37. // 根据 bitmap 为每一个需要被请求的异步子计划发送请求
38. // 并重置 bitmap
39. /* Make a new request for each of the async subplans that need it. */
40. needrequest = node->as_needrequest;
41. node->as_needrequest = NULL;
42. i = -1;
43. while ((i = bms_next_member(needrequest, i)) >= 0)
44. {
45. AsyncRequest *areq = node->as_asyncrequests[i];

47. // 发起请求
48. // 如果异步子计划产生结果
49. // 那么将子计划的元组缓存槽加入到 Append 节点的结果槽数组中
50. /* Do the actual work. */
51. ExecAsyncRequest(areq);
52. }
53. bms_free(needrequest);

55. // 当前已有异步子计划产生了结果
56. // 使返回结果槽指向 Append 节点结果槽中的最后一个槽，并返回
57. /* Return one of the asynchronously-generated results if any. */
58. if (node->as_nasyncresults > 0)
59. {
60. --node->as_nasyncresults;
61. *result = node->as_asyncresults[node->as_nasyncresults];
62. return true;
63. }

65. // 至此，没有获取到任何有效结果，返回 false
66. return false;
67. }

ExecAppendAsyncEventWait

该函数用于配制所有异步子计划想要监听的事件集合，然后向 OS 内核轮询这个事件集合。如果触发了监听事件，那么回调并处理事件。其中用到了异步抽象层的 ExecAsyncConfigureWait() 以配置监听事件，以及 ExecAsyncNotify() 以处理事件。


1. /* ----------------------------------------------------------------
2. *      ExecAppendAsyncEventWait
3. *
4. *      Wait or poll for file descriptor events and fire callbacks.
5. * ----------------------------------------------------------------
6. */
7. static void
8. ExecAppendAsyncEventWait(AppendState *node)
9. {
10. int         nevents = node->as_nasyncplans + 1;
11. long        timeout = node->as_syncdone ? -1 : 0;
12. WaitEvent   occurred_event[EVENT_BUFFER_SIZE];
13. int         noccurred;
14. int         i;

16. /* We should never be called when there are no valid async subplans. */
17. Assert(node->as_nasyncremain > 0);

19. // 创建一个事件等待集合
20. // 添加监听 PostMaster 退出的事件
21. node->as_eventset = CreateWaitEventSet(CurrentMemoryContext, nevents);
22. AddWaitEventToSet(node->as_eventset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
23. NULL, NULL);

25. // 对每一个异步子计划，如果当前子计划的请求已经发送
26. // 那么将子计划需要监听的文件描述符添加到事件等待集合中
27. /* Give each waiting subplan a chance to add an event. */
28. i = -1;
29. while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
30. {
31. AsyncRequest *areq = node->as_asyncrequests[i];

33. if (areq->callback_pending)
34. ExecAsyncConfigureWait(areq);
35. }

37. // 如果除了监听 PostMaster 进程退出以外没有其它事件需要监听了
38. // 即没有任何异步子计划的事件需要监听，那么直接返回
39. /*
40. * No need for further processing if there are no configured events other
41. * than the postmaster death event.
42. */
43. if (GetNumRegisteredWaitEvents(node->as_eventset) == 1)
44. {
45. FreeWaitEventSet(node->as_eventset);
46. node->as_eventset = NULL;
47. return;
48. }

50. // 最多只监听 EVENT_BUFFER_SIZE 个事件
51. // 因为 WaitEvent 数组的空间有限
52. /* We wait on at most EVENT_BUFFER_SIZE events. */
53. if (nevents > EVENT_BUFFER_SIZE)
54. nevents = EVENT_BUFFER_SIZE;

56. /*
57. * If the timeout is -1, wait until at least one event occurs.  If the
58. * timeout is 0, poll for events, but do not wait at all.
59. */
60. // 向 OS 内核轮询！
61. noccurred = WaitEventSetWait(node->as_eventset, timeout, occurred_event,
62. nevents, WAIT_EVENT_APPEND_READY);
63. FreeWaitEventSet(node->as_eventset);
64. node->as_eventset = NULL;
65. if (noccurred == 0)
66. return;

68. // 处理本轮轮询中触发的事件
69. /* Deliver notifications. */
70. for (i = 0; i < noccurred; i++)
71. {
72. WaitEvent  *w = &occurred_event[i];

74. /*
75. * Each waiting subplan should have registered its wait event with
76. * user_data pointing back to its AsyncRequest.
77. */
78. // 异步子计划的 Socket 读事件被触发
79. if ((w->events & WL_SOCKET_READABLE) != 0)
80. {
81. AsyncRequest *areq = (AsyncRequest *) w->user_data;

83. if (areq->callback_pending)
84. {
85. /*
86. * Mark it as no longer needing a callback.  We must do this
87. * before dispatching the callback in case the callback resets
88. * the flag.
89. */
90. // 复位异步子计划的请求状态
91. areq->callback_pending = false;

93. /* Do the actual work. */
94. // 处理 Socket 读事件
95. ExecAsyncNotify(areq);
96. }
97. }
98. }
99. }

ExecAsyncAppendResponse

这个函数比较特殊。前几个函数都是 Append 算子的函数主动调用下层（异步抽象层）的函数；而这是一个回调函数，由异步抽象层函数获取到结果以后调用，将获取到的执行结果保存到 Append 算子的返回结果槽中。


1. /* ----------------------------------------------------------------
2. *      ExecAsyncAppendResponse
3. *
4. *      Receive a response from an asynchronous request we made.
5. * ----------------------------------------------------------------
6. */
7. void
8. ExecAsyncAppendResponse(AsyncRequest *areq)
9. {
10. AppendState *node = (AppendState *) areq->requestor;
11. TupleTableSlot *slot = areq->result;

13. /* The result should be a TupleTableSlot or NULL. */
14. Assert(slot == NULL || IsA(slot, TupleTableSlot));

16. // 异步子计划的请求还没有获取到结果，那么直接返回
17. /* Nothing to do if the request is pending. */
18. if (!areq->request_complete)
19. {
20. /* The request would have been pending for a callback. */
21. Assert(areq->callback_pending);
22. return;
23. }

25. // 异步子计划已经获取到结果

27. // 如果子计划的返回结果槽为空，那么说明这个异步子计划的执行已经结束
28. // 递减剩余的异步子计划数量，然后返回
29. /* If the result is NULL or an empty slot, there's nothing more to do. */
30. if (TupIsNull(slot))
31. {
32. /* The ending subplan wouldn't have been pending for a callback. */
33. Assert(!areq->callback_pending);
34. --node->as_nasyncremain;
35. return;
36. }

38. // 如果子计划的返回结果槽不为空
39. // 那么将返回结果槽保存到 Append 算子的 as_asyncresults 数组中
40. /* Save result so we can return it. */
41. Assert(node->as_nasyncresults < node->as_nasyncplans);
42. node->as_asyncresults[node->as_nasyncresults++] = slot;

44. // 标记这个异步子计划，表示它准备好进行下一次请求
45. /*
46. * Mark the subplan that returned a result as ready for a new request.  We
47. * don't launch another one here immediately because it might complete.
48. */
49. node->as_needrequest = bms_add_member(node->as_needrequest,
50. areq->request_index);
51. }

异步抽象层

这一层存在的目的应该是为了在两层物理算子之间传递请求状态和执行结果，并方便以后支持其它算子的异步化改造。下面直接开始分析这一层次上的所有函数。

ExecAsyncRequest

该函数被 Append 算子的 ExecAppendAsyncBegin() 和 ExecAppendAsyncRequest() 调用，主要功能是异步地向下层的 ForeignScan 算子请求元组，执行结果不通过返回值返回。如果下层算子产生了有效结果，那么回调上层 Append 算子的函数传递执行结果。


1. /*
2. * Asynchronously request a tuple from a designed async-capable node.
3. */
4. void
5. ExecAsyncRequest(AsyncRequest *areq)
6. {
7. // ...

9. // 目前的被请求者算子仅支持 ForeignScan
10. switch (nodeTag(areq->requestee))
11. {
12. // 向 ForeignScan 算子异步请求结果
13. case T_ForeignScanState:
14. ExecAsyncForeignScanRequest(areq);
15. break;
16. default:
17. /* If the node doesn't support async, caller messed up. */
18. elog(ERROR, "unrecognized node type: %d",
19. (int) nodeTag(areq->requestee));
20. }

22. // 如果子计划产生了结果，那么调用请求者（Append）算子的回调函数
23. ExecAsyncResponse(areq);

25. // ...
26. }

ExecAsyncConfigureWait

该函数被 Append 算子的 ExecAppendAsyncEventWait() 调用，将调用下层 ForeignScan 算子的相应函数以便 FDW 能够配置自己想要监听的文件描述符和及其读事件。


1. /*
2. * Give the asynchronous node a chance to configure the file descriptor event
3. * for which it wishes to wait.  We expect the node-type specific callback to
4. * make a single call of the following form:
5. *
6. * AddWaitEventToSet(set, WL_SOCKET_READABLE, fd, NULL, areq);
7. */
8. void
9. ExecAsyncConfigureWait(AsyncRequest *areq)
10. {
11. // ...

13. switch (nodeTag(areq->requestee))
14. {
15. // 目前唯一合法的下层节点是 ForeignScan
16. case T_ForeignScanState:
17. ExecAsyncForeignScanConfigureWait(areq);
18. break;
19. default:
20. /* If the node doesn't support async, caller messed up. */
21. elog(ERROR, "unrecognized node type: %d",
22. (int) nodeTag(areq->requestee));
23. }

25. // ...
26. }

ExecAsyncNotify

该函数被 Append 算子的 ExecAppendAsyncEventWait() 函数调用，说明下层算子正在监听的文件描述符上触发了读事件。所以该函数将调用下层 ForeignScan 算子的相应函数，从正在监听的文件描述符上获取数据并组装元组；然后回调 Append 算子的 ExecAsyncAppendResponse() 函数传递元组。


1. /*
2. * Call the asynchronous node back when a relevant event has occurred.
3. */
4. void
5. ExecAsyncNotify(AsyncRequest *areq)
6. {
7. // ...

9. switch (nodeTag(areq->requestee))
10. {
11. // 调用 ForeignScan 算子的相应函数
12. // 使其从文件描述符上获取数据并组装元组
13. case T_ForeignScanState:
14. ExecAsyncForeignScanNotify(areq);
15. break;
16. default:
17. /* If the node doesn't support async, caller messed up. */
18. elog(ERROR, "unrecognized node type: %d",
19. (int) nodeTag(areq->requestee));
20. }

22. // 回调，传递元组到 Append 算子中
23. ExecAsyncResponse(areq);

25. // ...
26. }

ExecAsyncResponse

该函数用于触发上层算子的回调函数，以传递从 FDW 获取到的元组。目前支持的上层算子只有 Append。


1. /*
2. * Call the requestor back when an asynchronous node has produced a result.
3. */
4. void
5. ExecAsyncResponse(AsyncRequest *areq)
6. {
7. switch (nodeTag(areq->requestor))
8. {
9. // 目前合法的请求方算子只有 Append
10. // 回调并传递结果元组
11. case T_AppendState:
12. ExecAsyncAppendResponse(areq);
13. break;
14. default:
15. /* If the node doesn't support async, caller messed up. */
16. elog(ERROR, "unrecognized node type: %d",
17. (int) nodeTag(areq->requestor));
18. }
19. }

ExecAsyncRequestPending / ExecAsyncRequestDone

这两个函数用于保存一个异步请求的进行状态和执行结果。


1. /*
2. * A requestee node should call this function to deliver the tuple to its
3. * requestor node.  The requestee node can call this from its ExecAsyncRequest
4. * or ExecAsyncNotify callback.
5. */
6. void
7. ExecAsyncRequestDone(AsyncRequest *areq, TupleTableSlot *result)
8. {
9. // 请求已完成
10. // 状态：完成
11. // 结果：在参数中
12. areq->request_complete = true;
13. areq->result = result;
14. }

16. /*
17. * A requestee node should call this function to indicate that it is pending
18. * for a callback.  The requestee node can call this from its ExecAsyncRequest
19. * or ExecAsyncNotify callback.
20. */
21. void
22. ExecAsyncRequestPending(AsyncRequest *areq)
23. {
24. // 新的请求已经发起
25. // 状态：未完成，进行中
26. // 结果：空
27. areq->callback_pending = true;
28. areq->request_complete = false;
29. areq->result = NULL;
30. }

ForeignScan 算子

从这一层中可以看出，异步执行特性在 FDW API 中加入了四个新的函数指针：


1. typedef struct FdwRoutine
2. {
3. // ...

5. /* Support functions for asynchronous execution */
6. IsForeignPathAsyncCapable_function IsForeignPathAsyncCapable;
7. ForeignAsyncRequest_function ForeignAsyncRequest;
8. ForeignAsyncConfigureWait_function ForeignAsyncConfigureWait;
9. ForeignAsyncNotify_function ForeignAsyncNotify;
10. } FdwRoutine;

其中第一个函数指针被 PostgreSQL 优化器用于判断是否可以产生一个带有异步执行的执行计划。后面三个函数指针被执行器分别用于：异步发起请求、轮询请求结果、回调处理请求结果。

由于目前异步执行仅对 postgres_fdw 做了支持，因此这一层目前的逻辑很简单：就是转而调用 postgres_fdw 对应功能的函数指针。这一层存在的意义是，方便以后对其它的 FDW 插件也支持异步执行。下面简单列出被异步抽象层调用的三个函数：


1. /* ----------------------------------------------------------------
2. *      ExecAsyncForeignScanRequest
3. *
4. *      Asynchronously request a tuple from a designed async-capable node
5. * ----------------------------------------------------------------
6. */
7. void
8. ExecAsyncForeignScanRequest(AsyncRequest *areq)
9. {
10. ForeignScanState *node = (ForeignScanState *) areq->requestee;
11. FdwRoutine *fdwroutine = node->fdwroutine;

13. Assert(fdwroutine->ForeignAsyncRequest != NULL);
14. fdwroutine->ForeignAsyncRequest(areq);
15. }

17. /* ----------------------------------------------------------------
18. *      ExecAsyncForeignScanConfigureWait
19. *
20. *      In async mode, configure for a wait
21. * ----------------------------------------------------------------
22. */
23. void
24. ExecAsyncForeignScanConfigureWait(AsyncRequest *areq)
25. {
26. ForeignScanState *node = (ForeignScanState *) areq->requestee;
27. FdwRoutine *fdwroutine = node->fdwroutine;

29. Assert(fdwroutine->ForeignAsyncConfigureWait != NULL);
30. fdwroutine->ForeignAsyncConfigureWait(areq);
31. }

33. /* ----------------------------------------------------------------
34. *      ExecAsyncForeignScanNotify
35. *
36. *      Callback invoked when a relevant event has occurred
37. * ----------------------------------------------------------------
38. */
39. void
40. ExecAsyncForeignScanNotify(AsyncRequest *areq)
41. {
42. ForeignScanState *node = (ForeignScanState *) areq->requestee;
43. FdwRoutine *fdwroutine = node->fdwroutine;

45. Assert(fdwroutine->ForeignAsyncNotify != NULL);
46. fdwroutine->ForeignAsyncNotify(areq);
47. }

Postgres FDW

在 postgres_fdw 中对 FDW API 中新增的四个函数指针进行了实现。


1. /*
2. * Foreign-data wrapper handler function: return a struct with pointers
3. * to my callback routines.
4. */
5. Datum
6. postgres_fdw_handler(PG_FUNCTION_ARGS)
7. {
8. FdwRoutine *routine = makeNode(FdwRoutine);

10. // ...

12. /* Support functions for asynchronous execution */
13. routine->IsForeignPathAsyncCapable = postgresIsForeignPathAsyncCapable;
14. routine->ForeignAsyncRequest = postgresForeignAsyncRequest;
15. routine->ForeignAsyncConfigureWait = postgresForeignAsyncConfigureWait;
16. routine->ForeignAsyncNotify = postgresForeignAsyncNotify;

18. PG_RETURN_POINTER(routine);
19. }

postgresIsForeignPathAsyncCapable

这个函数会被优化器回调，判断是否可以对一个外部表产生异步执行计划。其逻辑非常简单：判断外部表选项中的 async_capable 是否为 true 即可：


1. /*
2. * postgresIsForeignPathAsyncCapable
3. *      Check whether a given ForeignPath node is async-capable.
4. */
5. static bool
6. postgresIsForeignPathAsyncCapable(ForeignPath *path)
7. {
8. RelOptInfo *rel = ((Path *) path)->parent;
9. PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private;

11. return fpinfo->async_capable;
12. }

postgresForeignAsyncRequest

该函数承接上层算子异步地请求元组的要求。


1. /*
2. * postgresForeignAsyncRequest
3. *      Asynchronously request next tuple from a foreign PostgreSQL table.
4. */
5. static void
6. postgresForeignAsyncRequest(AsyncRequest *areq)
7. {
8. produce_tuple_asynchronously(areq, true);
9. }

11. /*
12. * Asynchronously produce next tuple from a foreign PostgreSQL table.
13. */
14. static void
15. produce_tuple_asynchronously(AsyncRequest *areq, bool fetch)
16. {
17. ForeignScanState *node = (ForeignScanState *) areq->requestee;
18. PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
19. AsyncRequest *pendingAreq = fsstate->conn_state->pendingAreq;
20. TupleTableSlot *result;

22. // 目前请求还没有开始
23. /* This should not be called if the request is currently in-process */
24. Assert(areq != pendingAreq);

26. // 下一个要返回的元组的 index 已经超出元组数组的长度了
27. // 说明 FDW 中缓存的元组已经被消耗殆尽，需要 FETCH 更多元组
28. /* Fetch some more tuples, if we've run out */
29. if (fsstate->next_tuple >= fsstate->num_tuples)
30. {
31. /* No point in another fetch if we already detected EOF, though */
32. // 当前连接还没有到达 EOF
33. // 占用连接，并发送 FETCH 请求
34. if (!fsstate->eof_reached)
35. {
36. /* Mark the request as pending for a callback */
37. ExecAsyncRequestPending(areq);
38. /* Begin another fetch if requested and if no pending request */
39. if (fetch && !pendingAreq)
40. fetch_more_data_begin(areq);
41. }
42. // 当前连接已经到达 EOF
43. // 将连接状态设置为已完成，并向上层传递空的结果
44. else
45. {
46. /* There's nothing more to do; just return a NULL pointer */
47. result = NULL;
48. /* Mark the request as complete */
49. ExecAsyncRequestDone(areq, result);
50. }
51. return;
52. }

54. // FDW 缓存的元组还没有被消耗殆尽
55. // 那么调用 ForeignScan 节点的 next 函数获取一个元组
56. // ForeignScan 节点的 next 函数最终也会调用到 FDW 的 routine 函数中：
57. // IterateDirectModify / IterateForeignScan
58. /* Get a tuple from the ForeignScan node */
59. result = areq->requestee->ExecProcNodeReal(areq->requestee);

61. // 返回元组不为空，本次请求成功
62. // 将本次请求的状态设置为完成，将结果元组保存，返回
63. if (!TupIsNull(result))
64. {
65. /* Mark the request as complete */
66. ExecAsyncRequestDone(areq, result);
67. return;
68. }

70. // 返回元组为空，说明 FDW 中缓存的元组已经消耗殆尽
71. /* We must have run out of tuples */
72. Assert(fsstate->next_tuple >= fsstate->num_tuples);

74. /* Fetch some more tuples, if we've not detected EOF yet */
75. // 当前连接还没有到达 EOF
76. // 占用连接，并发送 FETCH 请求
77. if (!fsstate->eof_reached)
78. {
79. /* Mark the request as pending for a callback */
80. ExecAsyncRequestPending(areq);
81. /* Begin another fetch if requested and if no pending request */
82. if (fetch && !pendingAreq)
83. fetch_more_data_begin(areq);
84. }
85. // 当前连接已经到达 EOF
86. // 将连接状态设置为已完成，并向上层传递空的结果
87. else
88. {
89. /* There's nothing more to do; just return a NULL pointer */
90. result = NULL;
91. /* Mark the request as complete */
92. ExecAsyncRequestDone(areq, result);
93. }
94. }

96. /*
97. * Begin an asynchronous data fetch.
98. *
99. * Note: this function assumes there is no currently-in-progress asynchronous
100. * data fetch.
101. *
102. * Note: fetch_more_data must be called to fetch the result.
103. */
104. static void
105. fetch_more_data_begin(AsyncRequest *areq)
106. {
107. ForeignScanState *node = (ForeignScanState *) areq->requestee;
108. PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
109. char        sql[64];

111. // 此时连接已经被当前这次请求占用
112. Assert(!fsstate->conn_state->pendingAreq);

114. /* Create the cursor synchronously. */
115. // 如果用于 FETCH 的 cursor 还没有被创建
116. // 那么同步地创建 cursor
117. if (!fsstate->cursor_exists)
118. create_cursor(node);

120. /* We will send this query, but not wait for the response. */
121. snprintf(sql, sizeof(sql), "FETCH %d FROM c%u",
122. fsstate->fetch_size, fsstate->cursor_number);

124. // 通过 libpq 库发送 FETCH 请求，不等待结果
125. if (PQsendQuery(fsstate->conn, sql) < 0)
126. pgfdw_report_error(ERROR, NULL, fsstate->conn, false, fsstate->query);

128. /* Remember that the request is in process */
129. fsstate->conn_state->pendingAreq = areq;
130. }

postgresForeignAsyncConfigureWait

该函数主要用于把需要监听的文件描述符（与外部数据库的连接）及其 Socket 读事件添加到等待事件集合中，以便后续的事件轮询。如果当前已经有可以返回的元组，那么直接返回。


1. /*
2. * postgresForeignAsyncConfigureWait
3. *      Configure a file descriptor event for which we wish to wait.
4. */
5. static void
6. postgresForeignAsyncConfigureWait(AsyncRequest *areq)
7. {
8. ForeignScanState *node = (ForeignScanState *) areq->requestee;
9. PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
10. AsyncRequest *pendingAreq = fsstate->conn_state->pendingAreq;
11. AppendState *requestor = (AppendState *) areq->requestor;
12. WaitEventSet *set = requestor->as_eventset;

14. // 连接目前应当已经被一个请求占用了
15. /* This should not be called unless callback_pending */
16. Assert(areq->callback_pending);

18. /*
19. * If process_pending_request() has been invoked on the given request
20. * before we get here, we might have some tuples already; in which case
21. * complete the request
22. */
23. // 如果已经有元组缓存在 FDW 中
24. // 那么将结果返回给请求方（Append 算子），并按需开启下一次异步请求
25. if (fsstate->next_tuple < fsstate->num_tuples)
26. {
27. // 将当前元组返回上层
28. // 如果有需要，异步请求下一个元组
29. complete_pending_request(areq);
30. // 如果请求已经结束，那么直接返回
31. if (areq->request_complete)
32. return;
33. // 如果请求还未结束，那么新一轮请求应该已经被发起了
34. // 此时请求的状态应该是进行中
35. Assert(areq->callback_pending);
36. }

38. // 此时 FDW 中已经没有缓存元组了
39. /* We must have run out of tuples */
40. Assert(fsstate->next_tuple >= fsstate->num_tuples);

42. // 等待事件集合中应该已经注册了监听 PostMaster 退出的事件
43. /* The core code would have registered postmaster death event */
44. Assert(GetNumRegisteredWaitEvents(set) >= 1);

46. /* Begin an asynchronous data fetch if not already done */
47. // 如果 FETCH 请求还没有开始，那么现在开始
48. if (!pendingAreq)
49. fetch_more_data_begin(areq);
50. // ...

52. // 将 Socket 的读事件添加到等待事件集合中
53. AddWaitEventToSet(set, WL_SOCKET_READABLE, PQsocket(fsstate->conn),
54. NULL, areq);
55. }

57. /*
58. * Complete a pending asynchronous request.
59. */
60. static void
61. complete_pending_request(AsyncRequest *areq)
62. {
63. /* The request would have been pending for a callback */
64. // 目前连接应该正被本次请求占用
65. Assert(areq->callback_pending);

67. /* Unlike AsyncNotify, we unset callback_pending ourselves */
68. // 释放当前 FETCH 请求对连接的占用
69. areq->callback_pending = false;

71. /* We begin a fetch afterwards if necessary; don't fetch */
72. // 如果有需要，重新占用连接并发起下一次 FETCH 请求
73. produce_tuple_asynchronously(areq, false);

75. /* Unlike AsyncNotify, we call ExecAsyncResponse ourselves */
76. // 将本轮请求的结果传递给 Append 算子
77. ExecAsyncResponse(areq);

79. /* Also, we do instrumentation ourselves, if required */
80. // 统计元组个数
81. if (areq->requestee->instrument)
82. InstrUpdateTupleCount(areq->requestee->instrument,
83. TupIsNull(areq->result) ? 0.0 : 1.0);
84. }

86. /*
87. * Process a pending asynchronous request.
88. */
89. void
90. process_pending_request(AsyncRequest *areq)
91. {
92. ForeignScanState *node = (ForeignScanState *) areq->requestee;
93. PgFdwScanState *fsstate PG_USED_FOR_ASSERTS_ONLY = (PgFdwScanState *) node->fdw_state;

95. /* The request would have been pending for a callback */
96. // 此时连接应该正被一个请求占用
97. Assert(areq->callback_pending);

99. /* The request should be currently in-process */
100. // 当前正在占用连接的请求应该就是传入参数中的请求
101. Assert(fsstate->conn_state->pendingAreq == areq);

103. // 接收 FETCH 请求的结果，组装并缓存元组
104. fetch_more_data(node);

106. /*
107. * If we didn't get any tuples, must be end of data; complete the request
108. * now.  Otherwise, we postpone completing the request until we are called
109. * from postgresForeignAsyncConfigureWait()/postgresForeignAsyncNotify().
110. */
111. // 如果 FETCH 的结果中没有任何元组
112. if (fsstate->next_tuple >= fsstate->num_tuples)
113. {
114. /* Unlike AsyncNotify, we unset callback_pending ourselves */
115. // 结束占用连接
116. areq->callback_pending = false;
117. /* Mark the request as complete */
118. // 连接已完成，将结果设置为 NULL
119. ExecAsyncRequestDone(areq, NULL);
120. /* Unlike AsyncNotify, we call ExecAsyncResponse ourselves */
121. // 回调 Append 算子的函数，传递执行结果
122. ExecAsyncResponse(areq);
123. }
124. }

postgresForeignAsyncNotify

该函数用于处理监听的文件描述符上触发的 Socket 读事件。读事件被触发后，该函数可以从文件描述符上获取数据。将获取到的数据组装为元组后，保存在 FDW 内用于缓存元组的数组中。其它函数可以调用 ForeignScan 算子的 ForeignNext() 函数进而调用到 FDW 的 IterateForeignScan 函数指针，从而获取 FDW 缓存的元组。


1. /*
2. * postgresForeignAsyncNotify
3. *      Fetch some more tuples from a file descriptor that becomes ready,
4. *      requesting next tuple.
5. */
6. static void
7. postgresForeignAsyncNotify(AsyncRequest *areq)
8. {
9. ForeignScanState *node = (ForeignScanState *) areq->requestee;
10. PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;

12. /* The core code would have initialized the callback_pending flag */
13. // 此时请求状态已经被设置为结束了
14. Assert(!areq->callback_pending);

16. /*
17. * If process_pending_request() has been invoked on the given request
18. * before we get here, we might have some tuples already; in which case
19. * produce the next tuple
20. */
21. // 如果 FDW 中还有缓存的元组，那么异步请求下一个元组并返回
22. if (fsstate->next_tuple < fsstate->num_tuples)
23. {
24. produce_tuple_asynchronously(areq, true);
25. return;
26. }

28. // FDW 缓存的元组已经消耗殆尽
29. /* We must have run out of tuples */
30. Assert(fsstate->next_tuple >= fsstate->num_tuples);

32. /* The request should be currently in-process */
33. Assert(fsstate->conn_state->pendingAreq == areq);

35. /* On error, report the original query, not the FETCH. */
36. if (!PQconsumeInput(fsstate->conn))
37. pgfdw_report_error(ERROR, NULL, fsstate->conn, false, fsstate->query);

39. // 获取数据并组装为元组
40. fetch_more_data(node);
41. // 异步请求下一个元组
42. produce_tuple_asynchronously(areq, true);
43. }

45. /*
46. * Fetch some more rows from the node's cursor.
47. */
48. static void
49. fetch_more_data(ForeignScanState *node)
50. {
51. PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
52. PGresult   *volatile res = NULL;
53. MemoryContext oldcontext;

55. /*
56. * We'll store the tuples in the batch_cxt.  First, flush the previous
57. * batch.
58. */
59. fsstate->tuples = NULL;
60. MemoryContextReset(fsstate->batch_cxt);
61. oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt);

63. /* PGresult must be released before leaving this function. */
64. PG_TRY();
65. {
66. PGconn     *conn = fsstate->conn;
67. int         numrows;
68. int         i;

70. // 异步执行
71. if (fsstate->async_capable)
72. {
73. Assert(fsstate->conn_state->pendingAreq);

75. /*
76. * The query was already sent by an earlier call to
77. * fetch_more_data_begin.  So now we just fetch the result.
78. */
79. // 因为 FETCH 的请求已经在之前被发出去了
80. // 所以这里直接获取 FETCH 的结果
81. res = pgfdw_get_result(conn, fsstate->query);
82. /* On error, report the original query, not the FETCH. */
83. if (PQresultStatus(res) != PGRES_TUPLES_OK)
84. pgfdw_report_error(ERROR, res, conn, false, fsstate->query);

86. /* Reset per-connection state */
87. fsstate->conn_state->pendingAreq = NULL;
88. }
89. else
90. {
91. // ...
92. }

94. // 分配在 FDW 中缓存元组的内存
95. /* Convert the data into HeapTuples */
96. numrows = PQntuples(res);
97. fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple));
98. fsstate->num_tuples = numrows;
99. fsstate->next_tuple = 0;

101. // 遍历获取到的每一行数据
102. for (i = 0; i < numrows; i++)
103. {
104. Assert(IsA(node->ss.ps.plan, ForeignScan));

106. // 组装元组
107. fsstate->tuples[i] =
108. make_tuple_from_result_row(res, i,
109. fsstate->rel,
110. fsstate->attinmeta,
111. fsstate->retrieved_attrs,
112. node,
113. fsstate->temp_cxt);
114. }

116. /* Update fetch_ct_2 */
117. if (fsstate->fetch_ct_2 < 2)
118. fsstate->fetch_ct_2++;

120. /* Must be EOF if we didn't get as many tuples as we asked for. */
121. fsstate->eof_reached = (numrows < fsstate->fetch_size);
122. }
123. PG_FINALLY();
124. {
125. if (res)
126. PQclear(res);
127. }
128. PG_END_TRY();

130. MemoryContextSwitchTo(oldcontext);
131. }

Summary

吃透上面的代码后可以发现，异步执行特性遵循一个很重要的原则：尽量不让任何人闲着 😂。当代码执行到 Append 算子时，它会立刻对所有的外表子计划发起异步请求，然后马上开始执行需要在本地执行的子计划，并不阻塞等待外表子计划的执行结果。之后执行流每次进入 Append 算子时，都会轮询 I/O 事件查看外表子计划的异步执行结果是否已经产生——如果产生，那么获取结果，并立刻发送下一轮的异步请求。这样，无论是当前数据库还是远程数据库，都在尽可能地并行推进一个物理计划的执行。其性能自然会优于原先的同步阻塞执行模型。

这让我想起了一个现实生活中的例子。前段时间我去办理户籍业务，由于当时是周末，只开了一个服务窗口，所以我排了大约一个多小时的队才办理成功。在排队的一个多小时里，我好奇地观察了这个窗口的业务处理流程，发现了一个很有意思的现象：窗口工作人员会把到号的人喊到窗口，然后把一个表格给到号的人现场填写，工作人员等待表格填写完毕后，将表格中的内容录入系统完成办理。绝大部分时候，表格录入的时间远小于表格填写的时间。这里或许有两个可以优化的点：

其它正在排队的人可以利用排队时间填写表格
工作人员等待表格填写的时间，可以用于录入其他人的表格

这里我们不去讨论现实生活中的种种复杂因素，比如表格可能会被填错，需要窗口人员现场指导等。我只是觉得这个过程像极了 FDW：表格填写对应了耗时较久的外部数据操作；表格录入对应了耗时较短的库内数据操作。FDW 的异步执行就是通过上述两个优化点，提升了运行效率。

References

GitHub - postgres/postgres - Add support for asynchronous execution

Parallel execution of postgres_fdw scan’s in PG-14 (Important step forward for horizontal scaling)

PostgreSQL’s Foreign Data Wrapper

PostgreSQL: Documentation - Chapter 34. libpq — C Library

PostgreSQL: Documentation - 57.2. Foreign Data Wrapper Callback Routines

原文：http://mysql.taobao.org/monthly/2022/07/04/