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 num_sync (
+ enable_async_merge_append (boolean)
+
+ enable_async_merge_append configuration parameter
+
+
+
+
+ Enables or disables the query planner's use of async-aware
+ merge append plan types. The default is on.
+
+
+
+
enable_bitmapscan (boolean)
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 MergeAppend 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.