diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 115ebc7925f..838545feb36 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -12365,12 +12365,104 @@ SELECT * FROM result_tbl ORDER BY a; (2 rows) DELETE FROM result_tbl; +-- Merge Append cases +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 count(*) FROM (SELECT * FROM async_pt ORDER BY b, a LIMIT 100000) ss; + QUERY PLAN +--------------------------------------------------------------------------------------------------------------- + Aggregate + Output: count(*) + -> Limit + Output: async_pt.a, async_pt.b, NULL::text + -> 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, NULL::text + Remote SQL: SELECT a, b 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, NULL::text + Remote SQL: SELECT a, b FROM public.base_tbl2 ORDER BY b ASC NULLS LAST, a ASC NULLS LAST +(12 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 +-- Merge Append cases +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 + -> Async Foreign Scan on public.async_p_broken async_pt_3 + Output: async_pt_3.a, async_pt_3.b, async_pt_3.c + Remote SQL: SELECT a, b, c FROM public.non_existent_table ORDER BY b ASC NULLS LAST, a ASC NULLS LAST +(11 rows) + +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); @@ -12474,13 +12566,119 @@ SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WH 3505 (2 rows) +-- Merge Append cases +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 ExecMergeAppendAsyncReset code path that asserts async requests +-- made for subplans aren't pending anymore at ReScan +-- SubPlan case +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) + +-- Nested Loop case +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; -- Test COPY TO when foreign table is partition 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. -DROP FOREIGN TABLE async_p3; -DROP TABLE base_tbl3; -- Check case where the partitioned table has local/remote partitions CREATE TABLE async_p3 PARTITION OF async_pt FOR VALUES FROM (3000) TO (4000); INSERT INTO async_p3 SELECT 3000 + i, i, to_char(i, 'FM0000') FROM generate_series(0, 999, 5) i; @@ -12514,6 +12712,37 @@ SELECT * FROM result_tbl ORDER BY a; (3 rows) DELETE FROM result_tbl; +-- Merge Append cases +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 +12938,21 @@ SELECT * FROM async_pt WHERE a < 2000; Remote SQL: SELECT a, b, c FROM public.base_tbl1 WHERE ((a < 2000)) (3 rows) +-- Merge Append cases +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 +13004,52 @@ SELECT * FROM result_tbl ORDER BY a; (1 row) DELETE FROM result_tbl; +-- Merge Append cases +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 +13586,154 @@ 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; +-- Test joins with async Merge Append +SET enable_partitionwise_join TO ON; +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) + +RESET enable_partitionwise_join; +-- 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; +-- Clean up +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..a570392c948 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 an Append, + * consider skipping configuration of the file descriptor event to + * improve concurrency. If it's a MergeAppend, it cannot make + * progress unless the given request returns a result; in which case + * configure the file descriptor event below. + */ + 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; + } + + /* + * Otherwise, 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 a preceding subplan is currently in-progress. Skip + * the configuration even in the MergeAppend case; in that case just + * difer the configuration until the caller processes the request(s) + * for the preceding subplan(s). */ return; } diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 77c9a25a890..aea72af207e 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -4291,10 +4291,30 @@ 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; +-- Merge Append cases +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 count(*) FROM (SELECT * FROM async_pt ORDER BY b, a LIMIT 100000) ss; +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; +-- Merge Append cases +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM async_pt ORDER BY b, a; +SELECT * FROM async_pt ORDER BY b, a; DROP FOREIGN TABLE async_p_broken; -- Check case where multiple partitions use the same connection @@ -4321,12 +4341,33 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE a = o.x OR (a = 1505 AND o.x = 2505) LIMIT 1) s ORDER BY o.x; SELECT o.x FROM (VALUES (2505), (3505)) o(x), LATERAL (SELECT a FROM async_pt WHERE a = o.x OR (a = 1505 AND o.x = 2505) LIMIT 1) s ORDER BY o.x; --- Test COPY TO when foreign table is partition -COPY async_pt TO stdout; --error +-- Merge Append cases +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 ExecMergeAppendAsyncReset code path that asserts async requests +-- made for subplans aren't pending anymore at ReScan +-- SubPlan case +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); + +-- Nested Loop case +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; +-- Test COPY TO when foreign table is partition +COPY async_pt TO stdout; --error + -- Check case where the partitioned table has local/remote partitions CREATE TABLE async_p3 PARTITION OF async_pt FOR VALUES FROM (3000) TO (4000); INSERT INTO async_p3 SELECT 3000 + i, i, to_char(i, 'FM0000') FROM generate_series(0, 999, 5) i; @@ -4339,6 +4380,11 @@ INSERT INTO result_tbl SELECT * FROM async_pt WHERE b === 505; SELECT * FROM result_tbl ORDER BY a; DELETE FROM result_tbl; +-- Merge Append cases +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 +4425,10 @@ SELECT * FROM async_pt WHERE a < 3000; EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM async_pt WHERE a < 2000; +-- Merge Append cases +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 +4449,18 @@ EXECUTE async_pt_query (2000, 505); SELECT * FROM result_tbl ORDER BY a; DELETE FROM result_tbl; +-- Merge Append cases +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 +4639,70 @@ 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; + +-- Test joins with async Merge Append +SET enable_partitionwise_join TO ON; + +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; + +RESET enable_partitionwise_join; + +-- 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; + +-- Clean up +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/doc/src/sgml/fdwhandler.sgml b/doc/src/sgml/fdwhandler.sgml index 502441fefcb..3f253b8de6a 100644 --- a/doc/src/sgml/fdwhandler.sgml +++ b/doc/src/sgml/fdwhandler.sgml @@ -1632,10 +1632,11 @@ IsForeignPathAsyncCapable(ForeignPath *path); Test whether a given ForeignPath path can scan the underlying foreign relation asynchronously. This function will only be called at the end of query planning when the - given path is a direct child of an AppendPath - path and when the planner believes that asynchronous execution improves - performance, and should return true if the given path is able to scan the - foreign relation asynchronously. + given path is a direct child of an AppendPath or + MergeAppendPath path and when the planner + believes that asynchronous execution improves performance, and should + return true if the given path is able to scan the foreign relation + asynchronously. @@ -1654,7 +1655,8 @@ ForeignAsyncRequest(AsyncRequest *areq); ForeignScan node. areq is the AsyncRequest struct describing the ForeignScan node and the parent - Append node that requested the tuple from it. + Append or MergeAppend + node that requested the tuple from it. This function should store the tuple into the slot specified by areq->result, and set areq->request_complete to true; @@ -1682,10 +1684,14 @@ ForeignAsyncConfigureWait(AsyncRequest *areq); This function will only be called when the ForeignScan node has the areq->callback_pending flag set, and should add - the event to the as_eventset of the parent - Append node described by the - areq. See the comments for - ExecAsyncConfigureWait in + the event to the WaitEventSet associated with + the parent Append or + MergeAppend node described by the + areq. + It's recommended to use + ExecAsyncGetRequestorEventSet to get the + WaitEventSet from the parent node. + See the comments for ExecAsyncConfigureWait in src/backend/executor/execAsync.c for additional information. When the file descriptor event occurs, ForeignAsyncNotify will be called. diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index 01577d8d69b..499d0bb2b10 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -493,8 +493,9 @@ OPTIONS (ADD password_required 'false'); postgres_fdw supports asynchronous execution, which - runs multiple parts of an Append node - concurrently rather than serially to improve performance. + runs multiple parts of an Append or + MergeAppend node concurrently rather than serially + to improve performance. This execution can be controlled using the following option: @@ -523,8 +524,9 @@ OPTIONS (ADD password_required 'false'); Asynchronous execution is applied even when an - Append node contains subplan(s) executed - synchronously as well as subplan(s) executed asynchronously. + Append or MergeAppend + node contains subplan(s) executed synchronously as well as subplan(s) + executed asynchronously. In such a case, if the asynchronous subplans are ones processed using postgres_fdw, tuples from the asynchronous subplans are not returned until after at least one synchronous subplan diff --git a/src/backend/executor/README b/src/backend/executor/README index 54f4782f31b..bf4cac79b91 100644 --- a/src/backend/executor/README +++ b/src/backend/executor/README @@ -419,17 +419,20 @@ at a later time. A process which discovers this type of situation can always handle it simply by blocking, but this may waste time that could be spent executing some other part of the plan tree where progress could be made immediately. This is particularly likely to occur when the plan tree contains -an Append node. Asynchronous execution runs multiple parts of an Append node -concurrently rather than serially to improve performance. +an Append or MergeAppend node. Asynchronous execution runs multiple parts of +an Append or MergeAppend node concurrently rather than serially to improve +performance. For asynchronous execution, an Append node must first request a tuple from an async-capable child node using ExecAsyncRequest. Next, it must execute the asynchronous event loop using ExecAppendAsyncEventWait. Eventually, when a child node to which an asynchronous request has been made produces a tuple, -the Append node will receive it from the event loop via ExecAsyncResponse. In -the current implementation of asynchronous execution, the only node type that -requests tuples from an async-capable child node is an Append, while the only -node type that might be async-capable is a ForeignScan. +the Append node will receive it from the event loop via ExecAsyncResponse. +The same applies to MergeAppend nodes, except that they must execute the event +loop using ExecMergeAppendAsyncEventWait. In the current implementation of +asynchronous execution, the node types that request tuples from an +async-capable child node are an Append and a MergeAppend, while the only node +type that might be async-capable is a ForeignScan. Typically, the ExecAsyncResponse callback is the only one required for nodes that wish to request tuples asynchronously. On the other hand, async-capable 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..7cb0edde5ef 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,32 @@ 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 = palloc0_array(AsyncRequest *, nplans); + + 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 +290,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 +352,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 +464,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 +494,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 ExecAppendAsyncReset, 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 e9d3b168ad9..a957943b827 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.