diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 115ebc7925f..8f79b5c1c15 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -12365,12 +12365,83 @@ SELECT * FROM result_tbl ORDER BY a;
 (2 rows)
 
 DELETE FROM result_tbl;
+-- Test Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE b % 100 = 0 ORDER BY b, a;
+                                                          QUERY PLAN                                                          
+------------------------------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 WHERE (((b % 100) = 0)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p2 async_pt_2
+         Output: async_pt_2.a, async_pt_2.b, async_pt_2.c
+         Remote SQL: SELECT a, b, c FROM public.base_tbl2 WHERE (((b % 100) = 0)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+(8 rows)
+
+SELECT * FROM async_pt WHERE b % 100 = 0 ORDER BY b, a;
+  a   |  b  |  c   
+------+-----+------
+ 1000 |   0 | 0000
+ 2000 |   0 | 0000
+ 1100 | 100 | 0100
+ 2100 | 100 | 0100
+ 1200 | 200 | 0200
+ 2200 | 200 | 0200
+ 1300 | 300 | 0300
+ 2300 | 300 | 0300
+ 1400 | 400 | 0400
+ 2400 | 400 | 0400
+ 1500 | 500 | 0500
+ 2500 | 500 | 0500
+ 1600 | 600 | 0600
+ 2600 | 600 | 0600
+ 1700 | 700 | 0700
+ 2700 | 700 | 0700
+ 1800 | 800 | 0800
+ 2800 | 800 | 0800
+ 1900 | 900 | 0900
+ 2900 | 900 | 0900
+(20 rows)
+
+-- Test that steady-state execution of an async Merge Append re-fetches from
+-- an async subplan instead of just consuming its initial batch and treating
+-- exhaustion of that batch as end-of-data.  Each partition here has 200
+-- matching rows, more than the FDW's default fetch_size (100), so getting
+-- the right count requires more than one fetch from at least one partition
+-- after the initial heap fill.  The LIMIT keeps the ORDER BY (and therefore
+-- the Merge Append) from being optimized away in the subquery.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt ORDER BY b, a;
+                                              QUERY PLAN                                              
+------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p2 async_pt_2
+         Output: async_pt_2.a, async_pt_2.b, async_pt_2.c
+         Remote SQL: SELECT a, b, c FROM public.base_tbl2 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+(8 rows)
+
+SELECT count(*) FROM (SELECT * FROM async_pt ORDER BY b, a LIMIT 100000) ss;
+ count 
+-------
+   400
+(1 row)
+
 -- Test error handling, if accessing one of the foreign partitions errors out
 CREATE FOREIGN TABLE async_p_broken PARTITION OF async_pt FOR VALUES FROM (10000) TO (10001)
   SERVER loopback OPTIONS (table_name 'non_existent_table');
 SELECT * FROM async_pt;
 ERROR:  relation "public.non_existent_table" does not exist
 CONTEXT:  remote SQL command: SELECT a, b, c FROM public.non_existent_table
+-- Test error handling for async Merge Append
+SELECT * FROM async_pt ORDER BY b, a;
+ERROR:  relation "public.non_existent_table" does not exist
+CONTEXT:  remote SQL command: SELECT a, b, c FROM public.non_existent_table ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
 DROP FOREIGN TABLE async_p_broken;
 -- Check case where multiple partitions use the same connection
 CREATE TABLE base_tbl3 (a int, b int, c text);
@@ -12479,6 +12550,192 @@ COPY async_pt TO stdout; --error
 ERROR:  cannot copy from foreign table "async_p1"
 DETAIL:  Partition "async_p1" is a foreign table in partitioned table "async_pt".
 HINT:  Try the COPY (SELECT ...) TO variant.
+-- Test Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+                                              QUERY PLAN                                              
+------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Filter: (async_pt_1.b === 505)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p2 async_pt_2
+         Output: async_pt_2.a, async_pt_2.b, async_pt_2.c
+         Filter: (async_pt_2.b === 505)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl2 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p3 async_pt_3
+         Output: async_pt_3.a, async_pt_3.b, async_pt_3.c
+         Filter: (async_pt_3.b === 505)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl3 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+(14 rows)
+
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+  a   |  b  |  c   
+------+-----+------
+ 1505 | 505 | 0505
+ 2505 | 505 | 0505
+ 3505 | 505 | 0505
+(3 rows)
+
+-- Test async Merge Append rescan
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT
+	ARRAY(SELECT f.i FROM (SELECT b + g.i FROM async_pt WHERE a > g.i ORDER BY b) f(i) ORDER BY f.i LIMIT 10)
+FROM generate_series(1, 3) g(i);
+                                                            QUERY PLAN                                                            
+----------------------------------------------------------------------------------------------------------------------------------
+ Function Scan on pg_catalog.generate_series g
+   Output: ARRAY(SubPlan array_1)
+   Function Call: generate_series(1, 3)
+   SubPlan array_1
+     ->  Limit
+           Output: f.i
+           ->  Sort
+                 Output: f.i
+                 Sort Key: f.i
+                 ->  Subquery Scan on f
+                       Output: f.i
+                       ->  Merge Append
+                             Sort Key: async_pt.b
+                             ->  Async Foreign Scan on public.async_p1 async_pt_1
+                                   Output: (async_pt_1.b + g.i), async_pt_1.b
+                                   Remote SQL: SELECT b FROM public.base_tbl1 WHERE ((a > $1::integer)) ORDER BY b ASC NULLS LAST
+                             ->  Async Foreign Scan on public.async_p2 async_pt_2
+                                   Output: (async_pt_2.b + g.i), async_pt_2.b
+                                   Remote SQL: SELECT b FROM public.base_tbl2 WHERE ((a > $1::integer)) ORDER BY b ASC NULLS LAST
+                             ->  Async Foreign Scan on public.async_p3 async_pt_3
+                                   Output: (async_pt_3.b + g.i), async_pt_3.b
+                                   Remote SQL: SELECT b FROM public.base_tbl3 WHERE ((a > $1::integer)) ORDER BY b ASC NULLS LAST
+(22 rows)
+
+SELECT
+	ARRAY(SELECT f.i FROM (SELECT b + g.i FROM async_pt WHERE a > g.i ORDER BY b) f(i) ORDER BY f.i LIMIT 10)
+FROM generate_series(1, 3) g(i);
+           array           
+---------------------------
+ {1,1,1,6,6,6,11,11,11,16}
+ {2,2,2,7,7,7,12,12,12,17}
+ {3,3,3,8,8,8,13,13,13,18}
+(3 rows)
+
+-- Test async Merge Append rescan when a LIMIT stops the scan before every
+-- async subplan's request has completed, so on the next rescan (for the
+-- next outer row) some subplan's request is still in flight.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i ORDER BY b LIMIT 1) s;
+                                                     QUERY PLAN                                                     
+--------------------------------------------------------------------------------------------------------------------
+ Nested Loop
+   Output: g.i, async_pt.b
+   ->  Function Scan on pg_catalog.generate_series g
+         Output: g.i
+         Function Call: generate_series(1, 3)
+   ->  Limit
+         Output: async_pt.b
+         ->  Merge Append
+               Sort Key: async_pt.b
+               ->  Async Foreign Scan on public.async_p1 async_pt_1
+                     Output: async_pt_1.b
+                     Remote SQL: SELECT b FROM public.base_tbl1 WHERE ((a > $1::integer)) ORDER BY b ASC NULLS LAST
+               ->  Async Foreign Scan on public.async_p2 async_pt_2
+                     Output: async_pt_2.b
+                     Remote SQL: SELECT b FROM public.base_tbl2 WHERE ((a > $1::integer)) ORDER BY b ASC NULLS LAST
+               ->  Async Foreign Scan on public.async_p3 async_pt_3
+                     Output: async_pt_3.b
+                     Remote SQL: SELECT b FROM public.base_tbl3 WHERE ((a > $1::integer)) ORDER BY b ASC NULLS LAST
+(18 rows)
+
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i ORDER BY b LIMIT 1) s;
+ i | b 
+---+---
+ 1 | 0
+ 2 | 0
+ 3 | 0
+(3 rows)
+
+-- Test async Merge Append rescan when a LIMIT stops the scan before an
+-- async subplan's already-delivered result was ever consumed (because it
+-- didn't win the race to the heap top).  The next rescan (for the next
+-- outer row) must not mistake that leftover result for a fresh one; the
+-- expected result increases each row, so a stale reused value would show
+-- up as the first row's value repeating instead.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i AND b > g.i * 100 ORDER BY b LIMIT 1) s;
+                                                                     QUERY PLAN                                                                     
+----------------------------------------------------------------------------------------------------------------------------------------------------
+ Nested Loop
+   Output: g.i, async_pt.b
+   ->  Function Scan on pg_catalog.generate_series g
+         Output: g.i
+         Function Call: generate_series(1, 3)
+   ->  Limit
+         Output: async_pt.b
+         ->  Merge Append
+               Sort Key: async_pt.b
+               ->  Async Foreign Scan on public.async_p1 async_pt_1
+                     Output: async_pt_1.b
+                     Remote SQL: SELECT b FROM public.base_tbl1 WHERE ((a > $1::integer)) AND ((b > ($1::integer * 100))) ORDER BY b ASC NULLS LAST
+               ->  Async Foreign Scan on public.async_p2 async_pt_2
+                     Output: async_pt_2.b
+                     Remote SQL: SELECT b FROM public.base_tbl2 WHERE ((a > $1::integer)) AND ((b > ($1::integer * 100))) ORDER BY b ASC NULLS LAST
+               ->  Async Foreign Scan on public.async_p3 async_pt_3
+                     Output: async_pt_3.b
+                     Remote SQL: SELECT b FROM public.base_tbl3 WHERE ((a > $1::integer)) AND ((b > ($1::integer * 100))) ORDER BY b ASC NULLS LAST
+(18 rows)
+
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i AND b > g.i * 100 ORDER BY b LIMIT 1) s;
+ i |  b  
+---+-----
+ 1 | 105
+ 2 | 205
+ 3 | 305
+(3 rows)
+
+-- Test the ExecMergeAppendAsyncProcessPending code path: same shared-connection
+-- pruned-out hazard as the Append case earlier, but with ORDER BY forcing a
+-- MergeAppend.  async_p2 and async_p3 share a connection; per outer row exactly
+-- one of them is pruned while async_p1 (a different connection) always matches,
+-- and the inner LIMIT leaves the other shared-connection request outstanding
+-- across the rescan.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT o.x, s.b FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT b FROM async_pt WHERE a = 1505 OR a = o.x ORDER BY b LIMIT 1) s ORDER BY o.x;
+                                                                QUERY PLAN                                                                
+------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: "*VALUES*".column1, async_pt.b
+   Sort Key: "*VALUES*".column1
+   ->  Nested Loop
+         Output: "*VALUES*".column1, async_pt.b
+         ->  Values Scan on "*VALUES*"
+               Output: "*VALUES*".column1
+         ->  Limit
+               Output: async_pt.b
+               ->  Merge Append
+                     Sort Key: async_pt.b
+                     ->  Async Foreign Scan on public.async_p1 async_pt_1
+                           Output: async_pt_1.b
+                           Remote SQL: SELECT b FROM public.base_tbl1 WHERE (((a = 1505) OR (a = $1::integer))) ORDER BY b ASC NULLS LAST
+                     ->  Async Foreign Scan on public.async_p2 async_pt_2
+                           Output: async_pt_2.b
+                           Remote SQL: SELECT b FROM public.base_tbl2 WHERE (((a = 1505) OR (a = $1::integer))) ORDER BY b ASC NULLS LAST
+                     ->  Async Foreign Scan on public.async_p3 async_pt_3
+                           Output: async_pt_3.b
+                           Remote SQL: SELECT b FROM public.base_tbl3 WHERE (((a = 1505) OR (a = $1::integer))) ORDER BY b ASC NULLS LAST
+(20 rows)
+
+SELECT o.x, s.b FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT b FROM async_pt WHERE a = 1505 OR a = o.x ORDER BY b LIMIT 1) s ORDER BY o.x;
+  x   |  b  
+------+-----
+ 2505 | 505
+ 3505 | 505
+(2 rows)
+
 DROP FOREIGN TABLE async_p3;
 DROP TABLE base_tbl3;
 -- Check case where the partitioned table has local/remote partitions
@@ -12514,6 +12771,37 @@ SELECT * FROM result_tbl ORDER BY a;
 (3 rows)
 
 DELETE FROM result_tbl;
+-- Test Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+                                              QUERY PLAN                                              
+------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Filter: (async_pt_1.b === 505)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p2 async_pt_2
+         Output: async_pt_2.a, async_pt_2.b, async_pt_2.c
+         Filter: (async_pt_2.b === 505)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl2 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Sort
+         Output: async_pt_3.a, async_pt_3.b, async_pt_3.c
+         Sort Key: async_pt_3.b, async_pt_3.a
+         ->  Seq Scan on public.async_p3 async_pt_3
+               Output: async_pt_3.a, async_pt_3.b, async_pt_3.c
+               Filter: (async_pt_3.b === 505)
+(16 rows)
+
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+  a   |  b  |  c   
+------+-----+------
+ 1505 | 505 | 0505
+ 2505 | 505 | 0505
+ 3505 | 505 | 0505
+(3 rows)
+
 -- partitionwise joins
 SET enable_partitionwise_join TO true;
 CREATE TABLE join_tbl (a1 int, b1 int, c1 text, a2 int, b2 int, c2 text);
@@ -12709,6 +12997,21 @@ SELECT * FROM async_pt WHERE a < 2000;
    Remote SQL: SELECT a, b, c FROM public.base_tbl1 WHERE ((a < 2000))
 (3 rows)
 
+-- Test interaction of async Merge Append with plan-time partition pruning
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE a < 3000 ORDER BY b, a;
+                                                       QUERY PLAN                                                        
+-------------------------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 WHERE ((a < 3000)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p2 async_pt_2
+         Output: async_pt_2.a, async_pt_2.b, async_pt_2.c
+         Remote SQL: SELECT a, b, c FROM public.base_tbl2 WHERE ((a < 3000)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+(8 rows)
+
 -- Test interaction of async execution with run-time partition pruning
 SET plan_cache_mode TO force_generic_plan;
 PREPARE async_pt_query (int, int) AS
@@ -12760,6 +13063,52 @@ SELECT * FROM result_tbl ORDER BY a;
 (1 row)
 
 DELETE FROM result_tbl;
+-- Test interaction of async Merge Append with run-time partition pruning
+PREPARE async_pt_merge_query (int, int) AS
+  SELECT * FROM async_pt WHERE a < $1 AND b === $2 ORDER BY b, a;
+EXPLAIN (VERBOSE, COSTS OFF)
+EXECUTE async_pt_merge_query (3000, 505);
+                                                           QUERY PLAN                                                           
+--------------------------------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   Subplans Removed: 1
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Filter: (async_pt_1.b === $2)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 WHERE ((a < $1::integer)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+   ->  Async Foreign Scan on public.async_p2 async_pt_2
+         Output: async_pt_2.a, async_pt_2.b, async_pt_2.c
+         Filter: (async_pt_2.b === $2)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl2 WHERE ((a < $1::integer)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+(11 rows)
+
+EXECUTE async_pt_merge_query (3000, 505);
+  a   |  b  |  c   
+------+-----+------
+ 1505 | 505 | 0505
+ 2505 | 505 | 0505
+(2 rows)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+EXECUTE async_pt_merge_query (2000, 505);
+                                                           QUERY PLAN                                                           
+--------------------------------------------------------------------------------------------------------------------------------
+ Merge Append
+   Sort Key: async_pt.b, async_pt.a
+   Subplans Removed: 2
+   ->  Async Foreign Scan on public.async_p1 async_pt_1
+         Output: async_pt_1.a, async_pt_1.b, async_pt_1.c
+         Filter: (async_pt_1.b === $2)
+         Remote SQL: SELECT a, b, c FROM public.base_tbl1 WHERE ((a < $1::integer)) ORDER BY b ASC NULLS LAST, a ASC NULLS LAST
+(7 rows)
+
+EXECUTE async_pt_merge_query (2000, 505);
+  a   |  b  |  c   
+------+-----+------
+ 1505 | 505 | 0505
+(1 row)
+
 RESET plan_cache_mode;
 CREATE TABLE local_tbl(a int, b int, c text);
 INSERT INTO local_tbl VALUES (1505, 505, 'foo'), (2505, 505, 'bar');
@@ -13296,6 +13645,153 @@ SELECT a FROM base_tbl WHERE (a, random() > 0) IN (SELECT a, random() > 0 FROM f
 DROP FOREIGN TABLE foreign_tbl CASCADE;
 NOTICE:  drop cascades to foreign table foreign_tbl2
 DROP TABLE base_tbl;
+-- Test async Merge Append
+CREATE TABLE distr1 (i int, j int, k text) PARTITION BY HASH (i);
+CREATE TABLE base1 (i int, j int, k text);
+CREATE TABLE base2 (i int, j int, k text);
+CREATE FOREIGN TABLE distr1_p1 PARTITION OF distr1 FOR VALUES WITH (MODULUS 2, REMAINDER 0)
+SERVER loopback OPTIONS (table_name 'base1');
+CREATE FOREIGN TABLE distr1_p2 PARTITION OF distr1 FOR VALUES WITH (MODULUS 2, REMAINDER 1)
+SERVER loopback OPTIONS (table_name 'base2');
+CREATE TABLE distr2 (i int, j int, k text) PARTITION BY HASH (i);
+CREATE TABLE base3 (i int, j int, k text);
+CREATE TABLE base4 (i int, j int, k text);
+CREATE FOREIGN TABLE distr2_p1 PARTITION OF distr2 FOR VALUES WITH (MODULUS 2, REMAINDER 0)
+SERVER loopback OPTIONS (table_name 'base3');
+CREATE FOREIGN TABLE distr2_p2 PARTITION OF distr2 FOR VALUES WITH (MODULUS 2, REMAINDER 1)
+SERVER loopback OPTIONS (table_name 'base4');
+INSERT INTO distr1
+SELECT i, i*10, 'data_' || i FROM generate_series(1, 1000) i;
+INSERT INTO distr2
+SELECT i, i*10, 'data_' || i FROM generate_series(1, 100) i;
+ANALYZE distr1_p1;
+ANALYZE distr1_p2;
+ANALYZE distr2_p1;
+ANALYZE distr2_p2;
+SET enable_partitionwise_join TO ON;
+-- Test joins with async Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM distr1, distr2 WHERE distr1.i=distr2.i AND distr2.j > 90 and distr2.k like 'data%'
+ORDER BY distr2.i LIMIT 10;
+                                                                                                    QUERY PLAN                                                                                                     
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Limit
+   Output: distr1.i, distr1.j, distr1.k, distr2.i, distr2.j, distr2.k
+   ->  Merge Append
+         Sort Key: distr1.i
+         ->  Async Foreign Scan
+               Output: distr1_1.i, distr1_1.j, distr1_1.k, distr2_1.i, distr2_1.j, distr2_1.k
+               Relations: (public.distr1_p1 distr1_1) INNER JOIN (public.distr2_p1 distr2_1)
+               Remote SQL: SELECT r3.i, r3.j, r3.k, r5.i, r5.j, r5.k FROM (public.base1 r3 INNER JOIN public.base3 r5 ON (((r3.i = r5.i)) AND ((r5.j > 90)) AND ((r5.k ~~ 'data%')))) ORDER BY r3.i ASC NULLS LAST
+         ->  Async Foreign Scan
+               Output: distr1_2.i, distr1_2.j, distr1_2.k, distr2_2.i, distr2_2.j, distr2_2.k
+               Relations: (public.distr1_p2 distr1_2) INNER JOIN (public.distr2_p2 distr2_2)
+               Remote SQL: SELECT r4.i, r4.j, r4.k, r6.i, r6.j, r6.k FROM (public.base2 r4 INNER JOIN public.base4 r6 ON (((r4.i = r6.i)) AND ((r6.j > 90)) AND ((r6.k ~~ 'data%')))) ORDER BY r4.i ASC NULLS LAST
+(12 rows)
+
+SELECT * FROM distr1, distr2 WHERE distr1.i=distr2.i AND distr2.j > 90 and distr2.k like 'data%'
+ORDER BY distr2.i LIMIT 10;
+ i  |  j  |    k    | i  |  j  |    k    
+----+-----+---------+----+-----+---------
+ 10 | 100 | data_10 | 10 | 100 | data_10
+ 11 | 110 | data_11 | 11 | 110 | data_11
+ 12 | 120 | data_12 | 12 | 120 | data_12
+ 13 | 130 | data_13 | 13 | 130 | data_13
+ 14 | 140 | data_14 | 14 | 140 | data_14
+ 15 | 150 | data_15 | 15 | 150 | data_15
+ 16 | 160 | data_16 | 16 | 160 | data_16
+ 17 | 170 | data_17 | 17 | 170 | data_17
+ 18 | 180 | data_18 | 18 | 180 | data_18
+ 19 | 190 | data_19 | 19 | 190 | data_19
+(10 rows)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM distr1 LEFT JOIN distr2 ON distr1.i=distr2.i AND distr2.k like 'data%' WHERE  distr1.i > 90
+ORDER BY distr1.i LIMIT 20;
+                                                                                                     QUERY PLAN                                                                                                     
+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Limit
+   Output: distr1.i, distr1.j, distr1.k, distr2.i, distr2.j, distr2.k
+   ->  Merge Append
+         Sort Key: distr1.i
+         ->  Async Foreign Scan
+               Output: distr1_1.i, distr1_1.j, distr1_1.k, distr2_1.i, distr2_1.j, distr2_1.k
+               Relations: (public.distr1_p1 distr1_1) LEFT JOIN (public.distr2_p1 distr2_1)
+               Remote SQL: SELECT r4.i, r4.j, r4.k, r6.i, r6.j, r6.k FROM (public.base1 r4 LEFT JOIN public.base3 r6 ON (((r4.i = r6.i)) AND ((r6.k ~~ 'data%')))) WHERE ((r4.i > 90)) ORDER BY r4.i ASC NULLS LAST
+         ->  Async Foreign Scan
+               Output: distr1_2.i, distr1_2.j, distr1_2.k, distr2_2.i, distr2_2.j, distr2_2.k
+               Relations: (public.distr1_p2 distr1_2) LEFT JOIN (public.distr2_p2 distr2_2)
+               Remote SQL: SELECT r5.i, r5.j, r5.k, r7.i, r7.j, r7.k FROM (public.base2 r5 LEFT JOIN public.base4 r7 ON (((r5.i = r7.i)) AND ((r7.k ~~ 'data%')))) WHERE ((r5.i > 90)) ORDER BY r5.i ASC NULLS LAST
+(12 rows)
+
+SELECT * FROM distr1 LEFT JOIN distr2 ON  distr1.i=distr2.i AND distr2.k like 'data%' WHERE distr1.i > 90
+ORDER BY distr1.i LIMIT 20;
+  i  |  j   |    k     |  i  |  j   |    k     
+-----+------+----------+-----+------+----------
+  91 |  910 | data_91  |  91 |  910 | data_91
+  92 |  920 | data_92  |  92 |  920 | data_92
+  93 |  930 | data_93  |  93 |  930 | data_93
+  94 |  940 | data_94  |  94 |  940 | data_94
+  95 |  950 | data_95  |  95 |  950 | data_95
+  96 |  960 | data_96  |  96 |  960 | data_96
+  97 |  970 | data_97  |  97 |  970 | data_97
+  98 |  980 | data_98  |  98 |  980 | data_98
+  99 |  990 | data_99  |  99 |  990 | data_99
+ 100 | 1000 | data_100 | 100 | 1000 | data_100
+ 101 | 1010 | data_101 |     |      | 
+ 102 | 1020 | data_102 |     |      | 
+ 103 | 1030 | data_103 |     |      | 
+ 104 | 1040 | data_104 |     |      | 
+ 105 | 1050 | data_105 |     |      | 
+ 106 | 1060 | data_106 |     |      | 
+ 107 | 1070 | data_107 |     |      | 
+ 108 | 1080 | data_108 |     |      | 
+ 109 | 1090 | data_109 |     |      | 
+ 110 | 1100 | data_110 |     |      | 
+(20 rows)
+
+-- Test pruning with async Merge Append
+DELETE FROM distr2;
+INSERT INTO distr2
+SELECT i%10, i*10, 'data_' || i FROM generate_series(1, 1000) i;
+DEALLOCATE ALL;
+SET plan_cache_mode TO force_generic_plan;
+PREPARE async_pt_query (int, int) AS
+  SELECT * FROM distr2 WHERE i = ANY(ARRAY[$1, $2])
+  ORDER BY i,j
+  LIMIT 10;
+EXPLAIN (VERBOSE, COSTS OFF)
+	EXECUTE async_pt_query(1, 1);
+                                                                         QUERY PLAN                                                                         
+------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Limit
+   Output: distr2.i, distr2.j, distr2.k
+   ->  Merge Append
+         Sort Key: distr2.i, distr2.j
+         Subplans Removed: 1
+         ->  Async Foreign Scan on public.distr2_p1 distr2_1
+               Output: distr2_1.i, distr2_1.j, distr2_1.k
+               Remote SQL: SELECT i, j, k FROM public.base3 WHERE ((i = ANY (ARRAY[$1::integer, $2::integer]))) ORDER BY i ASC NULLS LAST, j ASC NULLS LAST
+(8 rows)
+
+EXECUTE async_pt_query(1, 1);
+ i |  j  |    k    
+---+-----+---------
+ 1 |  10 | data_1
+ 1 | 110 | data_11
+ 1 | 210 | data_21
+ 1 | 310 | data_31
+ 1 | 410 | data_41
+ 1 | 510 | data_51
+ 1 | 610 | data_61
+ 1 | 710 | data_71
+ 1 | 810 | data_81
+ 1 | 910 | data_91
+(10 rows)
+
+RESET plan_cache_mode;
+RESET enable_partitionwise_join;
+DROP TABLE distr1, distr2, base1, base2, base3, base4;
 ALTER SERVER loopback OPTIONS (DROP async_capable);
 ALTER SERVER loopback2 OPTIONS (DROP async_capable);
 -- ===================================================================
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 6757ee923cb..4c4f106f049 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -1887,7 +1887,8 @@ postgresReScanForeignScan(ForeignScanState *node)
 
 	/*
 	 * If the node is async-capable, any asynchronous fetch made for it should
-	 * have been processed before we get here (see ExecAppendAsyncReset()).
+	 * have been processed before we get here (see ExecAppendAsyncReset() and
+	 * ExecMergeAppendAsyncReset()).
 	 */
 	Assert(!fsstate->async_capable || !fsstate->conn_state->pendingAreq ||
 		   fsstate->conn_state->pendingAreq->requestee != (PlanState *) node);
@@ -8651,11 +8652,13 @@ postgresForeignAsyncConfigureWait(AsyncRequest *areq)
 	ForeignScanState *node = (ForeignScanState *) areq->requestee;
 	PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
 	AsyncRequest *pendingAreq = fsstate->conn_state->pendingAreq;
-	AppendState *requestor = (AppendState *) areq->requestor;
-	WaitEventSet *set = requestor->as_eventset;
+	PlanState  *requestor = areq->requestor;
+	WaitEventSet *set = ExecAsyncGetRequestorEventSet(requestor);
 
 	/* This should not be called unless callback_pending */
 	Assert(areq->callback_pending);
+	/* The requestor should be an Append or MergeAppend node */
+	Assert(IsA(requestor, AppendState) || IsA(requestor, MergeAppendState));
 
 	/*
 	 * If process_pending_request() has been invoked on the given request
@@ -8676,37 +8679,48 @@ postgresForeignAsyncConfigureWait(AsyncRequest *areq)
 	/* The core code would have registered postmaster death event */
 	Assert(GetNumRegisteredWaitEvents(set) >= 1);
 
-	/* Begin an asynchronous data fetch if not already done */
+	/* Begin an asynchronous data fetch if appropriate */
 	if (!pendingAreq)
 		fetch_more_data_begin(areq);
-	else if (pendingAreq->requestor != areq->requestor)
+	else if (pendingAreq->requestor != requestor)
 	{
 		/*
-		 * This is the case when the in-process request was made by another
-		 * Append.  Note that it might be useless to process the request made
-		 * by that Append, because the query might not need tuples from that
-		 * Append anymore; so we avoid processing it to begin a fetch for the
-		 * given request if possible.  If there are any child subplans of the
-		 * same parent that are ready for new requests, skip the given
-		 * request.  Likewise, if there are any configured events other than
-		 * the postmaster death event, skip it.  Otherwise, process the
-		 * in-process request, then begin a fetch to configure the event
-		 * below, because we might otherwise end up with no configured events
-		 * other than the postmaster death event.
+		 * This is the case when an asynchronous request made by another
+		 * requestor is currently in-progress.  If our requestor is Append,
+		 * consider skipping configuration of the file descriptor event to
+		 * improve concurrency.  Note that MergeAppend cannot make progress
+		 * unless the given request returns the result, so we do this only
+		 * in the Append case.
+		 */
+		if (IsA(requestor, AppendState))
+		{
+			/*
+			 * If the Append node already has results that are ready for new
+			 * requests or it has already configured file descriptor events
+			 * for other subplans, skip the configuration.
+			 */
+			if (!bms_is_empty(((AppendState *) requestor)->as_needrequest))
+				return;
+			if (GetNumRegisteredWaitEvents(set) > 1)
+				return;
+		}
+
+		/*
+		 * Otheewise, process the in-progress request, begin an asynchronous
+		 * data fetch for the given request, and configure the file descriptor
+		 * event below.
 		 */
-		if (!bms_is_empty(requestor->as_needrequest))
-			return;
-		if (GetNumRegisteredWaitEvents(set) > 1)
-			return;
 		process_pending_request(pendingAreq);
 		fetch_more_data_begin(areq);
 	}
 	else if (pendingAreq->requestee != areq->requestee)
 	{
 		/*
-		 * This is the case when the in-process request was made by the same
-		 * parent but for a different child.  Since we configure only the
-		 * event for the request made for that child, skip the given request.
+		 * This is the case when an asynchronous request made by the
+		 * requestor for another subplan is currently in-progress.  Skip the
+		 * configuration even in the case where the requestor is MergeAppend;
+		 * in which csse we just defer the configuration until the preceding
+		 * request(s) are processed by ExecMergeAppendAsyncDrain.
 		 */
 		return;
 	}
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 77c9a25a890..7f95ff5a203 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -4291,10 +4291,28 @@ INSERT INTO result_tbl SELECT a, b, 'AAA' || c FROM async_pt WHERE b === 505;
 SELECT * FROM result_tbl ORDER BY a;
 DELETE FROM result_tbl;
 
+-- Test Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE b % 100 = 0 ORDER BY b, a;
+SELECT * FROM async_pt WHERE b % 100 = 0 ORDER BY b, a;
+
+-- Test that steady-state execution of an async Merge Append re-fetches from
+-- an async subplan instead of just consuming its initial batch and treating
+-- exhaustion of that batch as end-of-data.  Each partition here has 200
+-- matching rows, more than the FDW's default fetch_size (100), so getting
+-- the right count requires more than one fetch from at least one partition
+-- after the initial heap fill.  The LIMIT keeps the ORDER BY (and therefore
+-- the Merge Append) from being optimized away in the subquery.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt ORDER BY b, a;
+SELECT count(*) FROM (SELECT * FROM async_pt ORDER BY b, a LIMIT 100000) ss;
+
 -- Test error handling, if accessing one of the foreign partitions errors out
 CREATE FOREIGN TABLE async_p_broken PARTITION OF async_pt FOR VALUES FROM (10000) TO (10001)
   SERVER loopback OPTIONS (table_name 'non_existent_table');
 SELECT * FROM async_pt;
+-- Test error handling for async Merge Append
+SELECT * FROM async_pt ORDER BY b, a;
 DROP FOREIGN TABLE async_p_broken;
 
 -- Check case where multiple partitions use the same connection
@@ -4324,6 +4342,51 @@ SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WH
 -- Test COPY TO when foreign table is partition
 COPY async_pt TO stdout; --error
 
+-- Test Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+
+-- Test async Merge Append rescan
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT
+	ARRAY(SELECT f.i FROM (SELECT b + g.i FROM async_pt WHERE a > g.i ORDER BY b) f(i) ORDER BY f.i LIMIT 10)
+FROM generate_series(1, 3) g(i);
+SELECT
+	ARRAY(SELECT f.i FROM (SELECT b + g.i FROM async_pt WHERE a > g.i ORDER BY b) f(i) ORDER BY f.i LIMIT 10)
+FROM generate_series(1, 3) g(i);
+
+-- Test async Merge Append rescan when a LIMIT stops the scan before every
+-- async subplan's request has completed, so on the next rescan (for the
+-- next outer row) some subplan's request is still in flight.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i ORDER BY b LIMIT 1) s;
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i ORDER BY b LIMIT 1) s;
+
+-- Test async Merge Append rescan when a LIMIT stops the scan before an
+-- async subplan's already-delivered result was ever consumed (because it
+-- didn't win the race to the heap top).  The next rescan (for the next
+-- outer row) must not mistake that leftover result for a fresh one; the
+-- expected result increases each row, so a stale reused value would show
+-- up as the first row's value repeating instead.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i AND b > g.i * 100 ORDER BY b LIMIT 1) s;
+SELECT g.i, s.b FROM generate_series(1, 3) g(i),
+	LATERAL (SELECT b FROM async_pt WHERE a > g.i AND b > g.i * 100 ORDER BY b LIMIT 1) s;
+
+-- Test the ExecMergeAppendAsyncProcessPending code path: same shared-connection
+-- pruned-out hazard as the Append case earlier, but with ORDER BY forcing a
+-- MergeAppend.  async_p2 and async_p3 share a connection; per outer row exactly
+-- one of them is pruned while async_p1 (a different connection) always matches,
+-- and the inner LIMIT leaves the other shared-connection request outstanding
+-- across the rescan.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT o.x, s.b FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT b FROM async_pt WHERE a = 1505 OR a = o.x ORDER BY b LIMIT 1) s ORDER BY o.x;
+SELECT o.x, s.b FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT b FROM async_pt WHERE a = 1505 OR a = o.x ORDER BY b LIMIT 1) s ORDER BY o.x;
+
 DROP FOREIGN TABLE async_p3;
 DROP TABLE base_tbl3;
 
@@ -4339,6 +4402,11 @@ INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505;
 SELECT * FROM result_tbl ORDER BY a;
 DELETE FROM result_tbl;
 
+-- Test Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+SELECT * FROM async_pt WHERE b === 505 ORDER BY b, a;
+
 -- partitionwise joins
 SET enable_partitionwise_join TO true;
 
@@ -4379,6 +4447,10 @@ SELECT * FROM async_pt WHERE a < 3000;
 EXPLAIN (VERBOSE, COSTS OFF)
 SELECT * FROM async_pt WHERE a < 2000;
 
+-- Test interaction of async Merge Append with plan-time partition pruning
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM async_pt WHERE a < 3000 ORDER BY b, a;
+
 -- Test interaction of async execution with run-time partition pruning
 SET plan_cache_mode TO force_generic_plan;
 
@@ -4399,6 +4471,18 @@ EXECUTE async_pt_query (2000, 505);
 SELECT * FROM result_tbl ORDER BY a;
 DELETE FROM result_tbl;
 
+-- Test interaction of async Merge Append with run-time partition pruning
+PREPARE async_pt_merge_query (int, int) AS
+  SELECT * FROM async_pt WHERE a < $1 AND b === $2 ORDER BY b, a;
+
+EXPLAIN (VERBOSE, COSTS OFF)
+EXECUTE async_pt_merge_query (3000, 505);
+EXECUTE async_pt_merge_query (3000, 505);
+
+EXPLAIN (VERBOSE, COSTS OFF)
+EXECUTE async_pt_merge_query (2000, 505);
+EXECUTE async_pt_merge_query (2000, 505);
+
 RESET plan_cache_mode;
 
 CREATE TABLE local_tbl(a int, b int, c text);
@@ -4577,6 +4661,69 @@ SELECT a FROM base_tbl WHERE (a, random() > 0) IN (SELECT a, random() > 0 FROM f
 DROP FOREIGN TABLE foreign_tbl CASCADE;
 DROP TABLE base_tbl;
 
+-- Test async Merge Append
+CREATE TABLE distr1 (i int, j int, k text) PARTITION BY HASH (i);
+CREATE TABLE base1 (i int, j int, k text);
+CREATE TABLE base2 (i int, j int, k text);
+CREATE FOREIGN TABLE distr1_p1 PARTITION OF distr1 FOR VALUES WITH (MODULUS 2, REMAINDER 0)
+SERVER loopback OPTIONS (table_name 'base1');
+CREATE FOREIGN TABLE distr1_p2 PARTITION OF distr1 FOR VALUES WITH (MODULUS 2, REMAINDER 1)
+SERVER loopback OPTIONS (table_name 'base2');
+
+CREATE TABLE distr2 (i int, j int, k text) PARTITION BY HASH (i);
+CREATE TABLE base3 (i int, j int, k text);
+CREATE TABLE base4 (i int, j int, k text);
+CREATE FOREIGN TABLE distr2_p1 PARTITION OF distr2 FOR VALUES WITH (MODULUS 2, REMAINDER 0)
+SERVER loopback OPTIONS (table_name 'base3');
+CREATE FOREIGN TABLE distr2_p2 PARTITION OF distr2 FOR VALUES WITH (MODULUS 2, REMAINDER 1)
+SERVER loopback OPTIONS (table_name 'base4');
+
+INSERT INTO distr1
+SELECT i, i*10, 'data_' || i FROM generate_series(1, 1000) i;
+
+INSERT INTO distr2
+SELECT i, i*10, 'data_' || i FROM generate_series(1, 100) i;
+
+ANALYZE distr1_p1;
+ANALYZE distr1_p2;
+ANALYZE distr2_p1;
+ANALYZE distr2_p2;
+
+SET enable_partitionwise_join TO ON;
+
+-- Test joins with async Merge Append
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM distr1, distr2 WHERE distr1.i=distr2.i AND distr2.j > 90 and distr2.k like 'data%'
+ORDER BY distr2.i LIMIT 10;
+SELECT * FROM distr1, distr2 WHERE distr1.i=distr2.i AND distr2.j > 90 and distr2.k like 'data%'
+ORDER BY distr2.i LIMIT 10;
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM distr1 LEFT JOIN distr2 ON distr1.i=distr2.i AND distr2.k like 'data%' WHERE  distr1.i > 90
+ORDER BY distr1.i LIMIT 20;
+SELECT * FROM distr1 LEFT JOIN distr2 ON  distr1.i=distr2.i AND distr2.k like 'data%' WHERE distr1.i > 90
+ORDER BY distr1.i LIMIT 20;
+
+-- Test pruning with async Merge Append
+DELETE FROM distr2;
+INSERT INTO distr2
+SELECT i%10, i*10, 'data_' || i FROM generate_series(1, 1000) i;
+
+DEALLOCATE ALL;
+SET plan_cache_mode TO force_generic_plan;
+PREPARE async_pt_query (int, int) AS
+  SELECT * FROM distr2 WHERE i = ANY(ARRAY[$1, $2])
+  ORDER BY i,j
+  LIMIT 10;
+EXPLAIN (VERBOSE, COSTS OFF)
+	EXECUTE async_pt_query(1, 1);
+EXECUTE async_pt_query(1, 1);
+RESET plan_cache_mode;
+
+RESET enable_partitionwise_join;
+
+DROP TABLE distr1, distr2, base1, base2, base3, base4;
+
 ALTER SERVER loopback OPTIONS (DROP async_capable);
 ALTER SERVER loopback2 OPTIONS (DROP async_capable);
 
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index f7db2995c3e..9f2d5b7d746 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5862,6 +5862,20 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-enable-async-merge-append" xreflabel="enable_async_merge_append">
+      <term><varname>enable_async_merge_append</varname> (<type>boolean</type>)
+      <indexterm>
+       <primary><varname>enable_async_merge_append</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Enables or disables the query planner's use of async-aware
+        merge append plan types. The default is <literal>on</literal>.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-enable-bitmapscan" xreflabel="enable_bitmapscan">
       <term><varname>enable_bitmapscan</varname> (<type>boolean</type>)
       <indexterm>
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index cf7ddbb01f4..5c02a9d0322 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -19,6 +19,10 @@
 #include "executor/instrument.h"
 #include "executor/nodeAppend.h"
 #include "executor/nodeForeignscan.h"
+#include "executor/nodeMergeAppend.h"
+#include "miscadmin.h"
+
+#define EVENT_BUFFER_SIZE			16
 
 /*
  * Asynchronously request a tuple from a designed async-capable node.
@@ -122,6 +126,9 @@ ExecAsyncResponse(AsyncRequest *areq)
 		case T_AppendState:
 			ExecAsyncAppendResponse(areq);
 			break;
+		case T_MergeAppendState:
+			ExecAsyncMergeAppendResponse(areq);
+			break;
 		default:
 			/* If the node doesn't support async, caller messed up. */
 			elog(ERROR, "unrecognized node type: %d",
@@ -153,3 +160,126 @@ ExecAsyncRequestPending(AsyncRequest *areq)
 	areq->request_complete = false;
 	areq->result = NULL;
 }
+
+/*
+ * Fetch the WaitEventSet associated with an async-aware plan.
+ */
+WaitEventSet *
+ExecAsyncGetRequestorEventSet(PlanState *requestor)
+{
+	switch (nodeTag(requestor))
+	{
+		case T_AppendState:
+			return ((AppendState *) requestor)->as_eventset;
+		case T_MergeAppendState:
+			return ((MergeAppendState *) requestor)->ms_eventset;
+		default:
+			/* If the node doesn't support async, caller messed up. */
+			elog(ERROR, "unrecognized node type: %d",
+				 (int) nodeTag(requestor));
+			return NULL;
+	}
+}
+
+/*
+ * Workhorse function for Exec{Merge}AppendAsyncEventWait.
+ */
+void
+ExecAsyncEventWait(Bitmapset *asyncplans,
+				   AsyncRequest **asyncrequests,
+				   WaitEventSet *eventset,
+				   int nevents,
+				   long timeout,
+				   uint32 wait_event_info)
+{
+	WaitEvent	occurred_event[EVENT_BUFFER_SIZE];
+	int			noccurred;
+	int			i;
+
+	/*
+	 * Add WL_EXIT_ON_PM_DEATH to the set, to exit immediately if the
+	 * postmaster dies.
+	 */
+	AddWaitEventToSet(eventset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
+					  NULL, NULL);
+
+	/* Give each waiting subplan a chance to add an event. */
+	i = -1;
+	while ((i = bms_next_member(asyncplans, i)) >= 0)
+	{
+		AsyncRequest *areq = asyncrequests[i];
+
+		if (areq->callback_pending)
+			ExecAsyncConfigureWait(areq);
+	}
+
+	/*
+	 * No need for further processing if none of the subplans configured any
+	 * events.
+	 */
+	if (GetNumRegisteredWaitEvents(eventset) == 1)
+		return;
+
+	/*
+	 * Add the process latch to the set, so that we wake up to process the
+	 * standard interrupts with CHECK_FOR_INTERRUPTS().
+	 *
+	 * NOTE: For historical reasons, it's important that this is added to the
+	 * WaitEventSet after the ExecAsyncConfigureWait() calls.  Namely,
+	 * postgres_fdw calls "GetNumRegisteredWaitEvents(set) == 1" to check if
+	 * any other events are in the set.  That's a poor design, it's
+	 * questionable for postgres_fdw to be doing that in the first place, but
+	 * we cannot change it now.  The pattern has possibly been copied to other
+	 * extensions too.
+	 */
+	AddWaitEventToSet(eventset, WL_LATCH_SET, PGINVALID_SOCKET,
+					  MyLatch, NULL);
+
+	/* Return at most EVENT_BUFFER_SIZE events in one call. */
+	if (nevents > EVENT_BUFFER_SIZE)
+		nevents = EVENT_BUFFER_SIZE;
+
+	/*
+	 * If the timeout is -1, wait until at least one event occurs.  If the
+	 * timeout is 0, poll for events, but do not wait at all.
+	 */
+	noccurred = WaitEventSetWait(eventset, timeout, occurred_event, nevents,
+								 wait_event_info);
+	if (noccurred == 0)
+		return;
+
+	/* Deliver notifications. */
+	for (i = 0; i < noccurred; i++)
+	{
+		WaitEvent  *w = &occurred_event[i];
+
+		/*
+		 * Each waiting subplan should have registered its wait event with
+		 * user_data pointing back to its AsyncRequest.
+		 */
+		if ((w->events & WL_SOCKET_READABLE) != 0)
+		{
+			AsyncRequest *areq = (AsyncRequest *) w->user_data;
+
+			if (areq->callback_pending)
+			{
+				/*
+				 * Mark it as no longer needing a callback.  We must do this
+				 * before dispatching the callback in case the callback resets
+				 * the flag.
+				 */
+				areq->callback_pending = false;
+
+				/* Do the actual work. */
+				ExecAsyncNotify(areq);
+			}
+		}
+
+		/* Handle standard interrupts */
+		if ((w->events & WL_LATCH_SET) != 0)
+		{
+			ResetLatch(MyLatch);
+			CHECK_FOR_INTERRUPTS();
+		}
+	}
+}
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index 06f46fe6501..2c5aff61a55 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -64,7 +64,6 @@
 #include "executor/nodeAppend.h"
 #include "miscadmin.h"
 #include "pgstat.h"
-#include "storage/latch.h"
 #include "storage/lwlock.h"
 #include "utils/wait_event.h"
 
@@ -84,7 +83,6 @@ struct ParallelAppendState
 };
 
 #define INVALID_SUBPLAN_INDEX		-1
-#define EVENT_BUFFER_SIZE			16
 
 static TupleTableSlot *ExecAppend(PlanState *pstate);
 static bool choose_next_subplan_locally(AppendState *node);
@@ -1019,103 +1017,20 @@ ExecAppendAsyncEventWait(AppendState *node)
 {
 	int			nevents = node->as_nasyncplans + 2;
 	long		timeout = node->as_syncdone ? -1 : 0;
-	WaitEvent	occurred_event[EVENT_BUFFER_SIZE];
-	int			noccurred;
-	int			i;
 
 	/* We should never be called when there are no valid async subplans. */
 	Assert(node->as_nasyncremain > 0);
 
 	Assert(node->as_eventset == NULL);
 	node->as_eventset = CreateWaitEventSet(CurrentResourceOwner, nevents);
-	AddWaitEventToSet(node->as_eventset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
-					  NULL, NULL);
-
-	/* Give each waiting subplan a chance to add an event. */
-	i = -1;
-	while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
-	{
-		AsyncRequest *areq = node->as_asyncrequests[i];
-
-		if (areq->callback_pending)
-			ExecAsyncConfigureWait(areq);
-	}
-
-	/*
-	 * No need for further processing if none of the subplans configured any
-	 * events.
-	 */
-	if (GetNumRegisteredWaitEvents(node->as_eventset) == 1)
-	{
-		FreeWaitEventSet(node->as_eventset);
-		node->as_eventset = NULL;
-		return;
-	}
-
-	/*
-	 * Add the process latch to the set, so that we wake up to process the
-	 * standard interrupts with CHECK_FOR_INTERRUPTS().
-	 *
-	 * NOTE: For historical reasons, it's important that this is added to the
-	 * WaitEventSet after the ExecAsyncConfigureWait() calls.  Namely,
-	 * postgres_fdw calls "GetNumRegisteredWaitEvents(set) == 1" to check if
-	 * any other events are in the set.  That's a poor design, it's
-	 * questionable for postgres_fdw to be doing that in the first place, but
-	 * we cannot change it now.  The pattern has possibly been copied to other
-	 * extensions too.
-	 */
-	AddWaitEventToSet(node->as_eventset, WL_LATCH_SET, PGINVALID_SOCKET,
-					  MyLatch, NULL);
-
-	/* Return at most EVENT_BUFFER_SIZE events in one call. */
-	if (nevents > EVENT_BUFFER_SIZE)
-		nevents = EVENT_BUFFER_SIZE;
-
-	/*
-	 * If the timeout is -1, wait until at least one event occurs.  If the
-	 * timeout is 0, poll for events, but do not wait at all.
-	 */
-	noccurred = WaitEventSetWait(node->as_eventset, timeout, occurred_event,
-								 nevents, WAIT_EVENT_APPEND_READY);
+	ExecAsyncEventWait(node->as_asyncplans,
+					   node->as_asyncrequests,
+					   node->as_eventset,
+					   nevents,
+					   timeout,
+					   WAIT_EVENT_APPEND_READY);
 	FreeWaitEventSet(node->as_eventset);
 	node->as_eventset = NULL;
-	if (noccurred == 0)
-		return;
-
-	/* Deliver notifications. */
-	for (i = 0; i < noccurred; i++)
-	{
-		WaitEvent  *w = &occurred_event[i];
-
-		/*
-		 * Each waiting subplan should have registered its wait event with
-		 * user_data pointing back to its AsyncRequest.
-		 */
-		if ((w->events & WL_SOCKET_READABLE) != 0)
-		{
-			AsyncRequest *areq = (AsyncRequest *) w->user_data;
-
-			if (areq->callback_pending)
-			{
-				/*
-				 * Mark it as no longer needing a callback.  We must do this
-				 * before dispatching the callback in case the callback resets
-				 * the flag.
-				 */
-				areq->callback_pending = false;
-
-				/* Do the actual work. */
-				ExecAsyncNotify(areq);
-			}
-		}
-
-		/* Handle standard interrupts */
-		if ((w->events & WL_LATCH_SET) != 0)
-		{
-			ResetLatch(MyLatch);
-			CHECK_FOR_INTERRUPTS();
-		}
-	}
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/backend/executor/nodeMergeAppend.c b/src/backend/executor/nodeMergeAppend.c
index 0173074d85d..5e6c459fa3e 100644
--- a/src/backend/executor/nodeMergeAppend.c
+++ b/src/backend/executor/nodeMergeAppend.c
@@ -39,12 +39,14 @@
 
 #include "postgres.h"
 
+#include "executor/execAsync.h"
 #include "executor/executor.h"
 #include "executor/execPartition.h"
 #include "executor/nodeMergeAppend.h"
 #include "lib/binaryheap.h"
 #include "miscadmin.h"
 #include "utils/sortsupport.h"
+#include "utils/wait_event.h"
 
 /*
  * We have one slot for each item in the heap array.  We use SlotNumber
@@ -55,6 +57,12 @@ typedef int32 SlotNumber;
 
 static TupleTableSlot *ExecMergeAppend(PlanState *pstate);
 static int	heap_compare_slots(Datum a, Datum b, void *arg);
+static void ExecMergeAppendAsyncBegin(MergeAppendState *node,
+									   Bitmapset *valid_asyncplans);
+static void ExecMergeAppendAsyncRequest(MergeAppendState *node, SlotNumber i);
+static void ExecMergeAppendAsyncDrain(MergeAppendState *node);
+static void ExecMergeAppendAsyncEventWait(MergeAppendState *node);
+static void ExecMergeAppendAsyncReset(MergeAppendState *node);
 
 
 /* ----------------------------------------------------------------
@@ -135,13 +143,29 @@ ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
 	/*
 	 * call ExecInitNode on each of the valid plans to be executed and save
 	 * the results into the mergeplanstates array.
+	 *
+	 * While at it, find out which subplans can be executed asynchronously.
 	 */
 	j = 0;
 	i = -1;
+	mergestate->ms_asyncplans = NULL;
+	mergestate->ms_nasyncplans = 0;
 	while ((i = bms_next_member(validsubplans, i)) >= 0)
 	{
 		Plan	   *initNode = (Plan *) list_nth(node->mergeplans, i);
 
+		/*
+		 * Record async subplans.  When executing EvalPlanQual, we treat them
+		 * as sync ones; don't do this when initializing an EvalPlanQual plan
+		 * tree.
+		 */
+		if (initNode->async_capable && estate->es_epq_active == NULL)
+		{
+			mergestate->ms_asyncplans =
+				bms_add_member(mergestate->ms_asyncplans, j);
+			mergestate->ms_nasyncplans++;
+		}
+
 		mergeplanstates[j++] = ExecInitNode(initNode, estate, eflags);
 	}
 
@@ -167,6 +191,33 @@ ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
 		mergestate->ps.resultopsfixed = false;
 	}
 
+	/* Initialize async state */
+	mergestate->ms_asyncrequests = NULL;
+	mergestate->ms_asyncpending = NULL;
+	mergestate->ms_eventset = NULL;
+
+	if (mergestate->ms_nasyncplans > 0)
+	{
+		mergestate->ms_asyncrequests = (AsyncRequest **)
+			palloc0(nplans * sizeof(AsyncRequest *));
+
+		i = -1;
+		while ((i = bms_next_member(mergestate->ms_asyncplans, i)) >= 0)
+		{
+			AsyncRequest *areq;
+
+			areq = palloc_object(AsyncRequest);
+			areq->requestor = (PlanState *) mergestate;
+			areq->requestee = mergeplanstates[i];
+			areq->request_index = i;
+			areq->callback_pending = false;
+			areq->request_complete = false;
+			areq->result = NULL;
+
+			mergestate->ms_asyncrequests[i] = areq;
+		}
+	}
+
 	/*
 	 * Miscellaneous initialization
 	 */
@@ -240,14 +291,56 @@ ExecMergeAppend(PlanState *pstate)
 		/*
 		 * First time through: pull the first tuple from each valid subplan,
 		 * and set up the heap.
+		 *
+		 * When there are async-capable subplans, fire off all their requests
+		 * up front so that they can run concurrently (for example, on
+		 * different foreign servers).  We then pull the first tuple from each
+		 * synchronous subplan while the asynchronous ones are still in flight,
+		 * and finally collect the asynchronous results.
 		 */
-		i = -1;
-		while ((i = bms_next_member(node->ms_valid_subplans, i)) >= 0)
+		if (node->ms_nasyncplans > 0)
 		{
-			node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
-			if (!TupIsNull(node->ms_slots[i]))
-				binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
+			Bitmapset  *valid_asyncplans;
+
+			/* Determine the valid async subplans and begin executing them. */
+			valid_asyncplans = bms_intersect(node->ms_valid_subplans,
+											 node->ms_asyncplans);
+			ExecMergeAppendAsyncBegin(node, valid_asyncplans);
+
+			/* Pull the first tuple from each valid synchronous subplan. */
+			i = -1;
+			while ((i = bms_next_member(node->ms_valid_subplans, i)) >= 0)
+			{
+				if (bms_is_member(i, node->ms_asyncplans))
+					continue;
+
+				node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
+				if (!TupIsNull(node->ms_slots[i]))
+					binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
+			}
+
+			/* Now wait for the asynchronous subplans to deliver their tuples. */
+			ExecMergeAppendAsyncDrain(node);
+			i = -1;
+			while ((i = bms_next_member(valid_asyncplans, i)) >= 0)
+			{
+				if (!TupIsNull(node->ms_slots[i]))
+					binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
+			}
+
+			bms_free(valid_asyncplans);
 		}
+		else
+		{
+			i = -1;
+			while ((i = bms_next_member(node->ms_valid_subplans, i)) >= 0)
+			{
+				node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
+				if (!TupIsNull(node->ms_slots[i]))
+					binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
+			}
+		}
+
 		binaryheap_build(node->ms_heap);
 		node->ms_initialized = true;
 	}
@@ -260,9 +353,23 @@ ExecMergeAppend(PlanState *pstate)
 		 * elements of the heap.  (We could perhaps simplify the logic a bit
 		 * by doing this before returning from the prior call, but it's better
 		 * to not pull tuples until necessary.)
+		 *
+		 * If the subplan is async-capable, we fire a fresh request for it and
+		 * wait for the result.  Unlike a plain Append, the merge cannot make
+		 * progress until this particular subplan's next tuple is known (it
+		 * might be the new heap minimum), so there is at most one outstanding
+		 * async request during this phase.
 		 */
 		i = DatumGetInt32(binaryheap_first(node->ms_heap));
-		node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
+
+		if (bms_is_member(i, node->ms_asyncplans))
+		{
+			ExecMergeAppendAsyncRequest(node, i);
+			ExecMergeAppendAsyncDrain(node);
+		}
+		else
+			node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
+
 		if (!TupIsNull(node->ms_slots[i]))
 			binaryheap_replace_first(node->ms_heap, Int32GetDatum(i));
 		else
@@ -358,6 +465,10 @@ ExecReScanMergeAppend(MergeAppendState *node)
 {
 	int			i;
 
+	/* If there are any async subplans, reset async requests made for them. */
+	if (node->ms_nasyncplans > 0)
+		ExecMergeAppendAsyncReset(node);
+
 	/*
 	 * If any PARAM_EXEC Params used in pruning expressions have changed, then
 	 * we'd better unset the valid subplans so that they are reselected for
@@ -384,11 +495,186 @@ ExecReScanMergeAppend(MergeAppendState *node)
 
 		/*
 		 * If chgParam of subnode is not null then plan will be re-scanned by
-		 * first ExecProcNode.
+		 * first ExecProcNode or by first ExecAsyncRequest.
 		 */
 		if (subnode->chgParam == NULL)
 			ExecReScan(subnode);
 	}
+
 	binaryheap_reset(node->ms_heap);
 	node->ms_initialized = false;
 }
+
+/* ----------------------------------------------------------------
+ *						Asynchronous MergeAppend Support
+ * ----------------------------------------------------------------
+ */
+
+/* ----------------------------------------------------------------
+ *		ExecMergeAppendAsyncBegin
+ *
+ *		Begin executing the designated valid async-capable subplans.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecMergeAppendAsyncBegin(MergeAppendState *node, Bitmapset *valid_asyncplans)
+{
+	int			i;
+
+	/* Backward scan is not supported by async-aware MergeAppends. */
+	Assert(ScanDirectionIsForward(node->ps.state->es_direction));
+
+	/* We should never be called when there are no async subplans. */
+	Assert(node->ms_nasyncplans > 0);
+
+	/* Make a request for each of the valid async subplans. */
+	i = -1;
+	while ((i = bms_next_member(valid_asyncplans, i)) >= 0)
+		ExecMergeAppendAsyncRequest(node, i);
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMergeAppendAsyncRequest
+ *
+ *		Request the next tuple from a single async-capable subplan.
+ *
+ *		The subplan is marked as having an outstanding request.  If the
+ *		request happens to complete synchronously, ExecAsyncMergeAppendResponse
+ *		clears that marker (and stores the result slot) before this returns.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecMergeAppendAsyncRequest(MergeAppendState *node, SlotNumber i)
+{
+	AsyncRequest *areq = node->ms_asyncrequests[i];
+
+	Assert(areq != NULL);
+	Assert(areq->request_index == i);
+	Assert(!areq->callback_pending);
+	Assert(!bms_is_member(i, node->ms_asyncpending));
+
+	node->ms_asyncpending = bms_add_member(node->ms_asyncpending, i);
+
+	/* Do the actual work. */
+	ExecAsyncRequest(areq);
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMergeAppendAsyncDrain
+ *
+ *		Wait until every outstanding async request has delivered its tuple
+ *		(or reached end-of-scan) into the corresponding ms_slots entry.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecMergeAppendAsyncDrain(MergeAppendState *node)
+{
+	while (!bms_is_empty(node->ms_asyncpending))
+	{
+		CHECK_FOR_INTERRUPTS();
+
+		ExecMergeAppendAsyncEventWait(node);
+	}
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMergeAppendAsyncEventWait
+ *
+ *		Wait for file descriptor events and fire callbacks.
+ *
+ *		Unlike an async Append, a MergeAppend cannot make progress until the
+ *		outstanding tuples arrive, so we always wait (indefinitely) for at
+ *		least one event.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecMergeAppendAsyncEventWait(MergeAppendState *node)
+{
+	int			nevents = node->ms_nasyncplans + 2;
+
+	/* We should never be called when there are no outstanding requests. */
+	Assert(!bms_is_empty(node->ms_asyncpending));
+
+	Assert(node->ms_eventset == NULL);
+	node->ms_eventset = CreateWaitEventSet(CurrentResourceOwner, nevents);
+	ExecAsyncEventWait(node->ms_asyncplans,
+					   node->ms_asyncrequests,
+					   node->ms_eventset,
+					   nevents,
+					   -1,
+					   WAIT_EVENT_MERGE_APPEND_READY);
+	FreeWaitEventSet(node->ms_eventset);
+	node->ms_eventset = NULL;
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMergeAppendAsyncReset
+ *
+ *		Reset asynchronous requests made for async-capable subplans.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecMergeAppendAsyncReset(MergeAppendState *node)
+{
+	int			i;
+
+	/* We should never be called when there are no async subplans. */
+	Assert(node->ms_nasyncplans > 0);
+
+	/* Reset async requests. */
+	i = -1;
+	while ((i = bms_next_member(node->ms_asyncplans, i)) >= 0)
+	{
+		AsyncRequest *areq = node->ms_asyncrequests[i];
+
+		/*
+		 * Unlike in the async-Append case, the request wouldn't have been
+		 * pending anymore at ReScan.
+		 */
+		Assert(!areq->callback_pending);
+		areq->request_complete = false;
+		areq->result = NULL;
+	}
+
+	/* Reset state variables. */
+	bms_free(node->ms_asyncpending);
+	node->ms_asyncpending = NULL;
+}
+
+/* ----------------------------------------------------------------
+ *		ExecAsyncMergeAppendResponse
+ *
+ *		Receive a response from an asynchronous request we made.
+ * ----------------------------------------------------------------
+ */
+void
+ExecAsyncMergeAppendResponse(AsyncRequest *areq)
+{
+	MergeAppendState *node = (MergeAppendState *) areq->requestor;
+	TupleTableSlot *slot = areq->result;
+	SlotNumber	i = areq->request_index;
+
+	/* The result should be a TupleTableSlot or NULL. */
+	Assert(slot == NULL || IsA(slot, TupleTableSlot));
+
+	/* Nothing to do if the request is still pending. */
+	if (!areq->request_complete)
+	{
+		/* The request would have been pending for a callback. */
+		Assert(areq->callback_pending);
+		return;
+	}
+
+	/*
+	 * Store the delivered result (a tuple, or an empty/NULL slot at
+	 * end-of-scan) as this subplan's current slot, and mark the request as no
+	 * longer outstanding.
+	 *
+	 * Note that the binary heap is manipulated by ExecMergeAppend, not here,
+	 * because the required operation differs between the initial heap build
+	 * and later single-subplan refills.
+	 */
+	Assert(bms_is_member(i, node->ms_asyncpending));
+	node->ms_slots[i] = slot;
+	node->ms_asyncpending = bms_del_member(node->ms_asyncpending, i);
+}
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 7bbddb8bee4..f760ef9aa3a 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -165,6 +165,7 @@ bool		enable_parallel_hash = true;
 bool		enable_partition_pruning = true;
 bool		enable_presorted_aggregate = true;
 bool		enable_async_append = true;
+bool		enable_async_merge_append = true;
 
 typedef struct
 {
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 5bc3a8e76d5..73438e76945 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -1475,6 +1475,19 @@ create_merge_append_plan(PlannerInfo *root, MergeAppendPath *best_path,
 	List	   *subplans = NIL;
 	ListCell   *subpaths;
 	RelOptInfo *rel = best_path->path.parent;
+	bool		consider_async;
+
+	/*
+	 * If appropriate, consider async execution of the children.  Unlike a
+	 * plain Append (which returns tuples in arrival order and hence disables
+	 * async whenever the output must be ordered), a MergeAppend always has
+	 * pathkeys but still preserves ordering by merging the per-child streams.
+	 * A child is only eligible if it is directly async-capable, i.e. we don't
+	 * have to insert a Sort node on top of it below; that is checked per child.
+	 */
+	consider_async = (enable_async_merge_append &&
+					  !best_path->path.parallel_safe &&
+					  list_length(best_path->subpaths) > 1);
 
 	/*
 	 * We don't have the actual creation of the MergeAppend node split out
@@ -1589,6 +1602,18 @@ create_merge_append_plan(PlannerInfo *root, MergeAppendPath *best_path,
 
 			subplan = sort_plan;
 		}
+		else if (consider_async)
+		{
+			/*
+			 * The subplan is sufficiently ordered on its own, so no Sort node
+			 * was inserted on top of it.  Only in this case is the child still
+			 * directly async-capable, so check for that here.  (If a Sort node
+			 * had been inserted, it would be the MergeAppend's child, and Sort
+			 * nodes are not async-capable.)
+			 */
+			if (mark_async_capable_plan(subplan, subpath))
+				Assert(subplan->async_capable);
+		}
 
 		subplans = lappend(subplans, subplan);
 	}
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 3d366fd1114..112873dc46b 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -138,6 +138,7 @@ LOGICAL_APPLY_SEND_DATA	"Waiting for a logical replication leader apply process
 LOGICAL_PARALLEL_APPLY_STATE_CHANGE	"Waiting for a logical replication parallel apply process to change state."
 LOGICAL_SYNC_DATA	"Waiting for a logical replication remote server to send data for initial table synchronization."
 LOGICAL_SYNC_STATE_CHANGE	"Waiting for a logical replication remote server to change state."
+MERGE_APPEND_READY	"Waiting for subplan nodes of a <literal>MergeAppend</literal> plan node to be ready."
 MESSAGE_QUEUE_INTERNAL	"Waiting for another process to be attached to a shared message queue."
 MESSAGE_QUEUE_PUT_MESSAGE	"Waiting to write a protocol message to a shared message queue."
 MESSAGE_QUEUE_RECEIVE	"Waiting to receive bytes from a shared message queue."
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 380679a76c0..d3ce67ffcdc 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -862,6 +862,13 @@
   boot_val => 'true',
 },
 
+{ name => 'enable_async_merge_append', type => 'bool', context => 'PGC_USERSET', group => 'QUERY_TUNING_METHOD',
+  short_desc => 'Enables the planner\'s use of async merge append plans.',
+  flags => 'GUC_EXPLAIN',
+  variable => 'enable_async_merge_append',
+  boot_val => 'true',
+},
+
 { name => 'enable_bitmapscan', type => 'bool', context => 'PGC_USERSET', group => 'QUERY_TUNING_METHOD',
   short_desc => 'Enables the planner\'s use of bitmap-scan plans.',
   flags => 'GUC_EXPLAIN',
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 01f98de8a9f..b492b3af06c 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -427,6 +427,7 @@
 # - Planner Method Configuration -
 
 #enable_async_append = on
+#enable_async_merge_append = on
 #enable_bitmapscan = on
 #enable_gathermerge = on
 #enable_groupagg = on
diff --git a/src/include/executor/execAsync.h b/src/include/executor/execAsync.h
index 402ffe49994..edb964bde72 100644
--- a/src/include/executor/execAsync.h
+++ b/src/include/executor/execAsync.h
@@ -14,6 +14,7 @@
 #define EXECASYNC_H
 
 #include "nodes/execnodes.h"
+#include "storage/latch.h"
 
 extern void ExecAsyncRequest(AsyncRequest *areq);
 extern void ExecAsyncConfigureWait(AsyncRequest *areq);
@@ -21,5 +22,12 @@ extern void ExecAsyncNotify(AsyncRequest *areq);
 extern void ExecAsyncResponse(AsyncRequest *areq);
 extern void ExecAsyncRequestDone(AsyncRequest *areq, TupleTableSlot *result);
 extern void ExecAsyncRequestPending(AsyncRequest *areq);
+extern WaitEventSet *ExecAsyncGetRequestorEventSet(PlanState *requestor);
+extern void ExecAsyncEventWait(Bitmapset *asyncplans,
+							   AsyncRequest **asyncrequests,
+							   WaitEventSet *eventset,
+							   int nevents,
+							   long timeout,
+							   uint32 wait_event_info);
 
 #endif							/* EXECASYNC_H */
diff --git a/src/include/executor/nodeMergeAppend.h b/src/include/executor/nodeMergeAppend.h
index dfcf45099ba..2255cc68b21 100644
--- a/src/include/executor/nodeMergeAppend.h
+++ b/src/include/executor/nodeMergeAppend.h
@@ -19,5 +19,6 @@
 extern MergeAppendState *ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags);
 extern void ExecEndMergeAppend(MergeAppendState *node);
 extern void ExecReScanMergeAppend(MergeAppendState *node);
+extern void ExecAsyncMergeAppendResponse(AsyncRequest *areq);
 
 #endif							/* NODEMERGEAPPEND_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index fe8681ba642..a6f1300d846 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1565,6 +1565,12 @@ typedef struct MergeAppendState
 	bool		ms_initialized; /* are subplans started? */
 	struct PartitionPruneState *ms_prune_state;
 	Bitmapset  *ms_valid_subplans;
+	Bitmapset  *ms_asyncplans;	/* asynchronous plans indexes */
+	int			ms_nasyncplans; /* # of asynchronous plans */
+	AsyncRequest **ms_asyncrequests;	/* array of AsyncRequests */
+	Bitmapset  *ms_asyncpending;	/* async plans with an outstanding request */
+	struct WaitEventSet *ms_eventset;	/* WaitEventSet used to configure file
+										 * descriptor wait events */
 } MergeAppendState;
 
 /* ----------------
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index 14255c900fc..19f8753e2d8 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -71,6 +71,7 @@ extern PGDLLIMPORT bool enable_parallel_hash;
 extern PGDLLIMPORT bool enable_partition_pruning;
 extern PGDLLIMPORT bool enable_presorted_aggregate;
 extern PGDLLIMPORT bool enable_async_append;
+extern PGDLLIMPORT bool enable_async_merge_append;
 extern PGDLLIMPORT int constraint_exclusion;
 
 extern double index_pages_fetched(double tuples_fetched, BlockNumber pages,
diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out
index 1e327c2afa4..f2bb803d1d5 100644
--- a/src/test/regress/expected/sysviews.out
+++ b/src/test/regress/expected/sysviews.out
@@ -156,6 +156,7 @@ select name, setting from pg_settings where name like 'enable%';
               name              | setting 
 --------------------------------+---------
  enable_async_append            | on
+ enable_async_merge_append      | on
  enable_bitmapscan              | on
  enable_distinct_reordering     | on
  enable_eager_aggregate         | on
@@ -181,7 +182,7 @@ select name, setting from pg_settings where name like 'enable%';
  enable_seqscan                 | on
  enable_sort                    | on
  enable_tidscan                 | on
-(26 rows)
+(27 rows)
 
 -- There are always wait event descriptions for various types.  InjectionPoint
 -- may be present or absent, depending on history since last postmaster start.