From 4ad712c2cbe1c9038bcfa32794b9dbd3f7c4e389 Mon Sep 17 00:00:00 2001
From: Amit Kapila <akapila@postgresql.org>
Date: Sat, 6 Mar 2021 15:23:57 +0530
Subject: [PATCH 1/3] Parallel SELECT for "INSERT INTO ... SELECT ..."
 -advanced tests.

---
 src/test/regress/expected/insert_parallel.out | 633 +++++++++++++++++-
 src/test/regress/sql/insert_parallel.sql      | 302 ++++++++-
 2 files changed, 931 insertions(+), 4 deletions(-)

diff --git a/src/test/regress/expected/insert_parallel.out b/src/test/regress/expected/insert_parallel.out
index 164668e319..2fef07bb6c 100644
--- a/src/test/regress/expected/insert_parallel.out
+++ b/src/test/regress/expected/insert_parallel.out
@@ -11,14 +11,36 @@ create or replace function fullname_parallel_unsafe(f text, l text) returns text
         return f || l;
     end;
 $$ language plpgsql immutable parallel unsafe;
+create or replace function fullname_parallel_safe(f text, l text) returns text as $$
+    begin
+        return f || l;
+    end;
+$$ language plpgsql immutable parallel safe;
 create or replace function fullname_parallel_restricted(f text, l text) returns text as $$
     begin
         return f || l;
     end;
 $$ language plpgsql immutable parallel restricted;
+create or replace function lastname_startswithe_u(last_name text) returns boolean as $$
+	begin
+		return substring(last_name from 1 for 1) = 'e';
+	end;
+$$ language plpgsql immutable parallel unsafe;
+create or replace function lastname_startswithe_s(last_name text) returns boolean as $$
+	begin
+		return substring(last_name from 1 for 1) = 'e';
+	end;
+$$ language plpgsql immutable parallel safe;
+create or replace function lastname_startswithe_r(last_name text) returns boolean as $$
+	begin
+		return substring(last_name from 1 for 1) = 'e';
+	end;
+$$ language plpgsql immutable parallel restricted;
 create table names(index int, first_name text, last_name text);
 create table names2(index int, first_name text, last_name text);
 create index names2_fullname_idx on names2 (fullname_parallel_unsafe(first_name, last_name));
+create table names3(index int, first_name text, last_name text);
+create index names3_fullname_idx on names3 (fullname_parallel_safe(first_name, last_name));
 create table names4(index int, first_name text, last_name text);
 create index names4_fullname_idx on names4 (fullname_parallel_restricted(first_name, last_name));
 insert into names values
@@ -180,6 +202,33 @@ insert into test_data1 select * from test_data where a = 10 returning a as data;
    10
 (1 row)
 
+--
+-- Test INSERT with RETURNING clause (ordered SELECT).
+-- (should create plan with parallel SELECT, GatherMerge parent node)
+--
+truncate test_data1;
+explain (costs off) insert into test_data1 select * from test_data where a <= 5 order by a returning a as data;
+                    QUERY PLAN                    
+--------------------------------------------------
+ Insert on test_data1
+   ->  Gather Merge
+         Workers Planned: 3
+         ->  Sort
+               Sort Key: test_data.a
+               ->  Parallel Seq Scan on test_data
+                     Filter: (a <= 5)
+(7 rows)
+
+insert into test_data1 select * from test_data where a <= 5 order by a returning a as data;
+ data 
+------
+    1
+    2
+    3
+    4
+    5
+(5 rows)
+
 --
 -- Test INSERT into a table with a foreign key.
 -- (Insert into a table with a foreign key is parallel-restricted,
@@ -203,6 +252,86 @@ select count(*), sum(unique1) from para_insert_f1;
  10000 | 49995000
 (1 row)
 
+--
+-- Test INSERT with underlying query, leader participation disabled
+--
+set parallel_leader_participation = off;
+truncate para_insert_p1 cascade;
+NOTICE:  truncate cascades to table "para_insert_f1"
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+               QUERY PLAN                
+-----------------------------------------
+ Insert on para_insert_p1
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+               Filter: (unique1 <= 2500)
+(5 rows)
+
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+select count(*), sum(unique1) from para_insert_p1;
+ count |   sum   
+-------+---------
+  2501 | 3126250
+(1 row)
+
+select * from para_insert_p1 where unique1 >= 2490 order by unique1;
+ unique1 | stringu1 
+---------+----------
+    2490 | URAAAA
+    2491 | VRAAAA
+    2492 | WRAAAA
+    2493 | XRAAAA
+    2494 | YRAAAA
+    2495 | ZRAAAA
+    2496 | ASAAAA
+    2497 | BSAAAA
+    2498 | CSAAAA
+    2499 | DSAAAA
+    2500 | ESAAAA
+(11 rows)
+
+--
+-- Test INSERT with underlying query, leader participation disabled
+--  and no workers available
+set max_parallel_workers=0;
+truncate para_insert_p1 cascade;
+NOTICE:  truncate cascades to table "para_insert_f1"
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+               QUERY PLAN                
+-----------------------------------------
+ Insert on para_insert_p1
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+               Filter: (unique1 <= 2500)
+(5 rows)
+
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+select count(*), sum(unique1) from para_insert_p1;
+ count |   sum   
+-------+---------
+  2501 | 3126250
+(1 row)
+
+select * from para_insert_p1 where unique1 >= 2490 order by unique1;
+ unique1 | stringu1 
+---------+----------
+    2490 | URAAAA
+    2491 | VRAAAA
+    2492 | WRAAAA
+    2493 | XRAAAA
+    2494 | YRAAAA
+    2495 | ZRAAAA
+    2496 | ASAAAA
+    2497 | BSAAAA
+    2498 | CSAAAA
+    2499 | DSAAAA
+    2500 | ESAAAA
+(11 rows)
+
+reset parallel_leader_participation;
+reset max_parallel_workers;
 --
 -- Test INSERT with ON CONFLICT ... DO UPDATE ...
 -- (should not create a parallel plan)
@@ -227,6 +356,208 @@ explain (costs off) insert into test_conflict_table(id, somedata) select a, a fr
    ->  Seq Scan on test_data
 (4 rows)
 
+--
+-- Test INSERT with parallelized aggregate
+--
+create table tenk1_avg_data(count int, avg_unique1 int, avg_stringu1_len int);
+explain (costs off) insert into tenk1_avg_data select count(*), avg(unique1), avg(length(stringu1)) from tenk1;
+                        QUERY PLAN                        
+----------------------------------------------------------
+ Insert on tenk1_avg_data
+   ->  Subquery Scan on "*SELECT*"
+         ->  Finalize Aggregate
+               ->  Gather
+                     Workers Planned: 4
+                     ->  Partial Aggregate
+                           ->  Parallel Seq Scan on tenk1
+(7 rows)
+
+insert into tenk1_avg_data select count(*), avg(unique1), avg(length(stringu1)) from tenk1;
+select * from tenk1_avg_data;
+ count | avg_unique1 | avg_stringu1_len 
+-------+-------------+------------------
+ 10000 |        5000 |                6
+(1 row)
+
+--
+-- Test INSERT with parallel bitmap heap scan
+--
+set enable_seqscan to off;
+set enable_indexscan to off;
+truncate para_insert_p1 cascade;
+NOTICE:  truncate cascades to table "para_insert_f1"
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 7500;
+                      QUERY PLAN                      
+------------------------------------------------------
+ Insert on para_insert_p1
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Bitmap Heap Scan on tenk1
+               Recheck Cond: (unique1 >= 7500)
+               ->  Bitmap Index Scan on tenk1_unique1
+                     Index Cond: (unique1 >= 7500)
+(7 rows)
+
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 7500;
+-- select some values to verify that the insert worked
+select * from para_insert_p1 where unique1 >= 9990 order by unique1;
+ unique1 | stringu1 
+---------+----------
+    9990 | GUAAAA
+    9991 | HUAAAA
+    9992 | IUAAAA
+    9993 | JUAAAA
+    9994 | KUAAAA
+    9995 | LUAAAA
+    9996 | MUAAAA
+    9997 | NUAAAA
+    9998 | OUAAAA
+    9999 | PUAAAA
+(10 rows)
+
+reset enable_seqscan;
+reset enable_indexscan;
+--
+-- Test INSERT with parallel append
+--
+create table a_star_data(aa int);
+explain (costs off) insert into a_star_data select aa from a_star where aa > 10;
+                       QUERY PLAN                       
+--------------------------------------------------------
+ Insert on a_star_data
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Append
+               ->  Parallel Seq Scan on d_star a_star_4
+                     Filter: (aa > 10)
+               ->  Parallel Seq Scan on f_star a_star_6
+                     Filter: (aa > 10)
+               ->  Parallel Seq Scan on e_star a_star_5
+                     Filter: (aa > 10)
+               ->  Parallel Seq Scan on b_star a_star_2
+                     Filter: (aa > 10)
+               ->  Parallel Seq Scan on c_star a_star_3
+                     Filter: (aa > 10)
+               ->  Parallel Seq Scan on a_star a_star_1
+                     Filter: (aa > 10)
+(16 rows)
+
+insert into a_star_data select aa from a_star where aa > 10;
+select count(aa), sum(aa) from a_star_data;
+ count | sum 
+-------+-----
+    16 | 300
+(1 row)
+
+--
+-- Test INSERT with parallel index scan
+--
+set enable_seqscan to off;
+set enable_bitmapscan to off;
+set min_parallel_index_scan_size=0;
+truncate para_insert_p1 cascade;
+NOTICE:  truncate cascades to table "para_insert_f1"
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 500;
+                          QUERY PLAN                          
+--------------------------------------------------------------
+ Insert on para_insert_p1
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Index Scan using tenk1_unique1 on tenk1
+               Index Cond: (unique1 >= 500)
+(5 rows)
+
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 500;
+-- select some values to verify that the parallel insert worked
+select count(*), sum(unique1) from para_insert_p1;
+ count |   sum    
+-------+----------
+  9500 | 49870250
+(1 row)
+
+select * from para_insert_p1 where unique1 >= 9990 order by unique1;
+ unique1 | stringu1 
+---------+----------
+    9990 | GUAAAA
+    9991 | HUAAAA
+    9992 | IUAAAA
+    9993 | JUAAAA
+    9994 | KUAAAA
+    9995 | LUAAAA
+    9996 | MUAAAA
+    9997 | NUAAAA
+    9998 | OUAAAA
+    9999 | PUAAAA
+(10 rows)
+
+--
+-- Test INSERT with parallel index-only scan
+--
+truncate para_insert_p1 cascade;
+NOTICE:  truncate cascades to table "para_insert_f1"
+explain (costs off) insert into para_insert_p1 select unique1 from tenk1 where unique1 >= 500;
+                            QUERY PLAN                             
+-------------------------------------------------------------------
+ Insert on para_insert_p1
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Index Only Scan using tenk1_unique1 on tenk1
+               Index Cond: (unique1 >= 500)
+(5 rows)
+
+insert into para_insert_p1 select unique1 from tenk1 where unique1 >= 500;
+-- select some values to verify that the parallel insert worked
+select count(*), sum(unique1) from para_insert_p1;
+ count |   sum    
+-------+----------
+  9500 | 49870250
+(1 row)
+
+select unique1 from para_insert_p1 where unique1 >= 9990 order by unique1;
+ unique1 
+---------
+    9990
+    9991
+    9992
+    9993
+    9994
+    9995
+    9996
+    9997
+    9998
+    9999
+(10 rows)
+
+reset min_parallel_index_scan_size;
+reset enable_seqscan;
+reset enable_bitmapscan;
+--
+-- Test INSERT with parallel-safe index expression
+-- (should create a parallel plan)
+--
+explain (costs off) insert into names3 select * from names;
+               QUERY PLAN               
+----------------------------------------
+ Insert on names3
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on names
+(4 rows)
+
+insert into names3 select * from names;
+select * from names3 order by fullname_parallel_safe(first_name, last_name);
+ index | first_name |  last_name  
+-------+------------+-------------
+     7 | alan       | turing
+     1 | albert     | einstein
+     3 | erwin      | schrodinger
+     6 | isaac      | newton
+     4 | leonhard   | euler
+     2 | niels      | bohr
+     8 | richard    | feynman
+     5 | stephen    | hawking
+(8 rows)
+
 --
 -- Test INSERT with parallel-unsafe index expression
 -- (should not create a parallel plan)
@@ -323,6 +654,51 @@ insert into names7 select * from names order by last_name returning last_name ||
  turing, alan
 (8 rows)
 
+--
+-- Test INSERT with parallel-safe index predicate
+-- (should create a parallel plan)
+--
+create table names8 (like names);
+create index names8_lastname_partial_idx on names8(index, last_name) where lastname_startswithe_s(last_name);
+explain (costs off) insert into names8 select * from names;
+               QUERY PLAN               
+----------------------------------------
+ Insert on names8
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on names
+(4 rows)
+
+insert into names8 select * from names;
+--
+-- Test INSERT with parallel-unsafe index predicate
+-- (should not create a parallel plan)
+--
+create table names9 (like names);
+create index names9_lastname_partial_idx on names9(index, last_name) where lastname_startswithe_u(last_name);
+explain (costs off) insert into names9 select * from names;
+       QUERY PLAN        
+-------------------------
+ Insert on names9
+   ->  Seq Scan on names
+(2 rows)
+
+--
+-- Test INSERT with parallel-restricted index predicate
+-- (should create a parallel plan)
+--
+create table names10 (like names);
+create index names10_lastname_partial_idx on names10(index, last_name) where lastname_startswithe_r(last_name);
+explain (costs off) insert into names10 select * from names;
+               QUERY PLAN               
+----------------------------------------
+ Insert on names10
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on names
+(4 rows)
+
+insert into names10 select * from names;
 --
 -- Test INSERT into temporary table with underlying query.
 -- (Insert into a temp table is parallel-restricted;
@@ -343,6 +719,40 @@ insert into temp_names select * from names;
 -- Test INSERT with column defaults
 --
 --
+-- a: no default
+-- b: unsafe default
+-- c: restricted default
+-- d: safe default
+--
+--
+-- No column defaults, should use parallel SELECT
+--
+explain (costs off) insert into testdef(a,b,c,d) select a,a*2,a*4,a*8 from test_data;
+                 QUERY PLAN                 
+--------------------------------------------
+ Insert on testdef
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on test_data
+(4 rows)
+
+insert into testdef(a,b,c,d) select a,a*2,a*4,a*8 from test_data;
+select * from testdef order by a;
+ a  | b  | c  | d  
+----+----+----+----
+  1 |  2 |  4 |  8
+  2 |  4 |  8 | 16
+  3 |  6 | 12 | 24
+  4 |  8 | 16 | 32
+  5 | 10 | 20 | 40
+  6 | 12 | 24 | 48
+  7 | 14 | 28 | 56
+  8 | 16 | 32 | 64
+  9 | 18 | 36 | 72
+ 10 | 20 | 40 | 80
+(10 rows)
+
+truncate testdef;
 --
 -- Parallel unsafe column default, should not use a parallel plan
 --
@@ -381,6 +791,35 @@ select * from testdef order by a;
  10 | 20 | 10 | 80
 (10 rows)
 
+truncate testdef;
+--
+-- Parallel safe column default, should use parallel SELECT
+--
+explain (costs off) insert into testdef(a,b,c) select a,a*2,a*4 from test_data;
+                 QUERY PLAN                 
+--------------------------------------------
+ Insert on testdef
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on test_data
+(4 rows)
+
+insert into testdef(a,b,c) select a,a*2,a*4 from test_data;
+select * from testdef order by a;
+ a  | b  | c  | d  
+----+----+----+----
+  1 |  2 |  4 | 20
+  2 |  4 |  8 | 20
+  3 |  6 | 12 | 20
+  4 |  8 | 16 | 20
+  5 | 10 | 20 | 20
+  6 | 12 | 24 | 20
+  7 | 14 | 28 | 20
+  8 | 16 | 32 | 20
+  9 | 18 | 36 | 20
+ 10 | 20 | 40 | 20
+(10 rows)
+
 truncate testdef;
 --
 -- Parallel restricted and unsafe column defaults, should not use a parallel plan
@@ -439,6 +878,64 @@ select count(*) from parttable1_2;
   5000
 (1 row)
 
+--
+-- Test INSERT into partition with parallel-unsafe partition key support function
+-- (should not create a parallel plan)
+--
+create function my_int4_sort(int4,int4) returns int language sql
+  as $$ select case when $1 = $2 then 0 when $1 > $2 then 1 else -1 end; $$;
+create operator class test_int4_ops for type int4 using btree as
+  operator 1 < (int4,int4), operator 2 <= (int4,int4),
+  operator 3 = (int4,int4), operator 4 >= (int4,int4),
+  operator 5 > (int4,int4), function 1 my_int4_sort(int4,int4);
+create table partkey_unsafe_key_supp_fn_t (a int4, b name) partition by range (a test_int4_ops);
+create table partkey_unsafe_key_supp_fn_t_1 partition of partkey_unsafe_key_supp_fn_t for values from (0) to (5000);
+create table partkey_unsafe_key_supp_fn_t_2 partition of partkey_unsafe_key_supp_fn_t for values from (5000) to (10000);
+explain (costs off) insert into partkey_unsafe_key_supp_fn_t select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on partkey_unsafe_key_supp_fn_t
+   ->  Seq Scan on tenk1
+(2 rows)
+
+--
+-- Test INSERT into partition with parallel-unsafe partition key expression
+-- (should not create a parallel plan)
+--
+create table partkey_unsafe_key_expr_t (a int4, b name) partition by range ((fullname_parallel_unsafe('',a::varchar)));
+explain (costs off) insert into partkey_unsafe_key_expr_t select unique1, stringu1 from tenk1;
+             QUERY PLAN              
+-------------------------------------
+ Insert on partkey_unsafe_key_expr_t
+   ->  Seq Scan on tenk1
+(2 rows)
+
+--
+-- Test INSERT into table with parallel-safe check constraint
+-- (should create a parallel plan)
+--
+create or replace function check_a(a int4) returns boolean as $$
+    begin
+        return (a >= 0 and a <= 9999);
+    end;
+$$ language plpgsql parallel safe;
+create table table_check_a(a int4 check (check_a(a)), b name);
+explain (costs off) insert into table_check_a select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on table_check_a
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+insert into table_check_a select unique1, stringu1 from tenk1;
+select count(*), sum(a) from table_check_a;
+ count |   sum    
+-------+----------
+ 10000 | 49995000
+(1 row)
+
 --
 -- Test INSERT into table with parallel-unsafe check constraint
 -- (should not create a parallel plan)
@@ -457,16 +954,24 @@ explain (costs off) insert into table_check_b(a,b,c) select unique1, unique2, st
 (2 rows)
 
 --
--- Test INSERT into table with parallel-safe after stmt-level triggers
+-- Test INSERT into table with parallel-safe before+after stmt-level triggers
 -- (should create a parallel SELECT plan; triggers should fire)
 --
 create table names_with_safe_trigger (like names);
+create or replace function insert_before_trigger_safe() returns trigger as $$
+    begin
+        raise notice 'hello from insert_before_trigger_safe';
+		return new;
+    end;
+$$ language plpgsql parallel safe;
 create or replace function insert_after_trigger_safe() returns trigger as $$
     begin
         raise notice 'hello from insert_after_trigger_safe';
 		return new;
     end;
 $$ language plpgsql parallel safe;
+create trigger insert_before_trigger_safe before insert on names_with_safe_trigger
+    for each statement execute procedure insert_before_trigger_safe();
 create trigger insert_after_trigger_safe after insert on names_with_safe_trigger
     for each statement execute procedure insert_after_trigger_safe();
 explain (costs off) insert into names_with_safe_trigger select * from names;
@@ -479,18 +984,27 @@ explain (costs off) insert into names_with_safe_trigger select * from names;
 (4 rows)
 
 insert into names_with_safe_trigger select * from names;
+NOTICE:  hello from insert_before_trigger_safe
 NOTICE:  hello from insert_after_trigger_safe
 --
--- Test INSERT into table with parallel-unsafe after stmt-level triggers
+-- Test INSERT into table with parallel-unsafe before+after stmt-level triggers
 -- (should not create a parallel plan; triggers should fire)
 --
 create table names_with_unsafe_trigger (like names);
+create or replace function insert_before_trigger_unsafe() returns trigger as $$
+    begin
+        raise notice 'hello from insert_before_trigger_unsafe';
+		return new;
+    end;
+$$ language plpgsql parallel unsafe;
 create or replace function insert_after_trigger_unsafe() returns trigger as $$
     begin
         raise notice 'hello from insert_after_trigger_unsafe';
 		return new;
     end;
 $$ language plpgsql parallel unsafe;
+create trigger insert_before_trigger_unsafe before insert on names_with_unsafe_trigger
+    for each statement execute procedure insert_before_trigger_unsafe();
 create trigger insert_after_trigger_unsafe after insert on names_with_unsafe_trigger
     for each statement execute procedure insert_after_trigger_unsafe();
 explain (costs off) insert into names_with_unsafe_trigger select * from names;
@@ -501,8 +1015,43 @@ explain (costs off) insert into names_with_unsafe_trigger select * from names;
 (2 rows)
 
 insert into names_with_unsafe_trigger select * from names;
+NOTICE:  hello from insert_before_trigger_unsafe
 NOTICE:  hello from insert_after_trigger_unsafe
 --
+-- Test INSERT into table with parallel-restricted before+after stmt-level trigger
+-- (should create a parallel plan with parallel SELECT;
+--  stmt-level before+after triggers should fire)
+--
+create table names_with_restricted_trigger (like names);
+create or replace function insert_before_trigger_restricted() returns trigger as $$
+    begin
+        raise notice 'hello from insert_before_trigger_restricted';
+		return new;
+    end;
+$$ language plpgsql parallel restricted;
+create or replace function insert_after_trigger_restricted() returns trigger as $$
+    begin
+        raise notice 'hello from insert_after_trigger_restricted';
+		return new;
+    end;
+$$ language plpgsql parallel restricted;
+create trigger insert_before_trigger_restricted before insert on names_with_restricted_trigger
+    for each statement execute procedure insert_before_trigger_restricted();
+create trigger insert_after_trigger_restricted after insert on names_with_restricted_trigger
+    for each statement execute procedure insert_after_trigger_restricted();
+explain (costs off) insert into names_with_restricted_trigger select * from names;
+               QUERY PLAN                
+-----------------------------------------
+ Insert on names_with_restricted_trigger
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on names
+(4 rows)
+
+insert into names_with_restricted_trigger select * from names;
+NOTICE:  hello from insert_before_trigger_restricted
+NOTICE:  hello from insert_after_trigger_restricted
+--
 -- Test INSERT into partition with parallel-unsafe trigger
 -- (should not create a parallel plan)
 --
@@ -551,15 +1100,54 @@ explain (costs off) execute q;
          Filter: ((a % 2) = 0)
 (3 rows)
 
+--
+-- Test INSERT into table with TOAST column
+--
+create table insert_toast_table(index int4, data text);
+create table insert_toast_table_data (like insert_toast_table);
+insert into insert_toast_table_data select i, rpad('T', 16384, 'ABCDEFGH') from generate_series(1,20) as i;
+explain (costs off) insert into insert_toast_table select index, data from insert_toast_table_data;
+                        QUERY PLAN                        
+----------------------------------------------------------
+ Insert on insert_toast_table
+   ->  Gather
+         Workers Planned: 3
+         ->  Parallel Seq Scan on insert_toast_table_data
+(4 rows)
+
+insert into insert_toast_table select index, data from insert_toast_table_data;
+select count(*) as row_count, sum(length(data)) as total_data_length from insert_toast_table;
+ row_count | total_data_length 
+-----------+-------------------
+        20 |            327680
+(1 row)
+
+truncate insert_toast_table;
 --
 -- Test INSERT into table having a DOMAIN column with a CHECK constraint
 --
 create function sql_is_distinct_from_u(anyelement, anyelement)
 returns boolean language sql parallel unsafe
 as 'select $1 is distinct from $2 limit 1';
+create or replace function sql_is_distinct_from_r(a anyelement, b anyelement) returns boolean as $$
+    begin
+        return (a <> b);
+    end;
+$$ language plpgsql parallel restricted;
+create or replace function sql_is_distinct_from_s(a anyelement, b anyelement) returns boolean as $$
+    begin
+        return (a <> b);
+    end;
+$$ language plpgsql parallel safe;
 create domain inotnull_u int
   check (sql_is_distinct_from_u(value, null));
+create domain inotnull_r int
+  check (sql_is_distinct_from_r(value, null));
+create domain inotnull_s int
+  check (sql_is_distinct_from_s(value, null));
 create table dom_table_u (x inotnull_u, y int);
+create table dom_table_r (x inotnull_r, y int);
+create table dom_table_s (x inotnull_s, y int);
 -- Test INSERT into table having a DOMAIN column with parallel-unsafe CHECK constraint
 explain (costs off) insert into dom_table_u select unique1, unique2 from tenk1;
        QUERY PLAN        
@@ -568,6 +1156,41 @@ explain (costs off) insert into dom_table_u select unique1, unique2 from tenk1;
    ->  Seq Scan on tenk1
 (2 rows)
 
+-- Test INSERT into table having a DOMAIN column with parallel-restricted CHECK constraint
+explain (costs off) insert into dom_table_r select unique1, unique2 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on dom_table_r
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+insert into dom_table_r select unique1, unique2 from tenk1;
+select count(*), sum(x) as sum_x, sum(y) as sum_y from dom_table_r;
+ count |  sum_x   |  sum_y   
+-------+----------+----------
+ 10000 | 49995000 | 49995000
+(1 row)
+
+-- Test INSERT into table having a DOMAIN column with parallel-safe CHECK constraint
+-- NOTE: Currently max_parallel_hazard() regards CoerceToDomain as parallel-restricted
+explain (costs off) insert into dom_table_s select unique1, unique2 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on dom_table_s
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+insert into dom_table_s select unique1, unique2 from tenk1;
+select count(*), sum(x) as sum_x, sum(y) as sum_y from dom_table_s;
+ count |  sum_x   |  sum_y   
+-------+----------+----------
+ 10000 | 49995000 | 49995000
+(1 row)
+
 rollback;
 --
 -- Clean up anything not created in the transaction
@@ -575,6 +1198,8 @@ rollback;
 drop table names;
 drop index names2_fullname_idx;
 drop table names2;
+drop index names3_fullname_idx;
+drop table names3;
 drop index names4_fullname_idx;
 drop table names4;
 drop table testdef;
@@ -583,4 +1208,8 @@ drop function bdefault_unsafe;
 drop function cdefault_restricted;
 drop function ddefault_safe;
 drop function fullname_parallel_unsafe;
+drop function fullname_parallel_safe;
 drop function fullname_parallel_restricted;
+drop function lastname_startswithe_u;
+drop function lastname_startswithe_s;
+drop function lastname_startswithe_r;
diff --git a/src/test/regress/sql/insert_parallel.sql b/src/test/regress/sql/insert_parallel.sql
index 171d8e5b84..cd8f9ef00e 100644
--- a/src/test/regress/sql/insert_parallel.sql
+++ b/src/test/regress/sql/insert_parallel.sql
@@ -15,15 +15,41 @@ create or replace function fullname_parallel_unsafe(f text, l text) returns text
     end;
 $$ language plpgsql immutable parallel unsafe;
 
+create or replace function fullname_parallel_safe(f text, l text) returns text as $$
+    begin
+        return f || l;
+    end;
+$$ language plpgsql immutable parallel safe;
+
 create or replace function fullname_parallel_restricted(f text, l text) returns text as $$
     begin
         return f || l;
     end;
 $$ language plpgsql immutable parallel restricted;
 
+create or replace function lastname_startswithe_u(last_name text) returns boolean as $$
+	begin
+		return substring(last_name from 1 for 1) = 'e';
+	end;
+$$ language plpgsql immutable parallel unsafe;
+
+create or replace function lastname_startswithe_s(last_name text) returns boolean as $$
+	begin
+		return substring(last_name from 1 for 1) = 'e';
+	end;
+$$ language plpgsql immutable parallel safe;
+
+create or replace function lastname_startswithe_r(last_name text) returns boolean as $$
+	begin
+		return substring(last_name from 1 for 1) = 'e';
+	end;
+$$ language plpgsql immutable parallel restricted;
+
 create table names(index int, first_name text, last_name text);
 create table names2(index int, first_name text, last_name text);
 create index names2_fullname_idx on names2 (fullname_parallel_unsafe(first_name, last_name));
+create table names3(index int, first_name text, last_name text);
+create index names3_fullname_idx on names3 (fullname_parallel_safe(first_name, last_name));
 create table names4(index int, first_name text, last_name text);
 create index names4_fullname_idx on names4 (fullname_parallel_restricted(first_name, last_name));
 
@@ -141,6 +167,14 @@ create table test_data1(like test_data);
 explain (costs off) insert into test_data1 select * from test_data where a = 10 returning a as data;
 insert into test_data1 select * from test_data where a = 10 returning a as data;
 
+--
+-- Test INSERT with RETURNING clause (ordered SELECT).
+-- (should create plan with parallel SELECT, GatherMerge parent node)
+--
+truncate test_data1;
+explain (costs off) insert into test_data1 select * from test_data where a <= 5 order by a returning a as data;
+insert into test_data1 select * from test_data where a <= 5 order by a returning a as data;
+
 --
 -- Test INSERT into a table with a foreign key.
 -- (Insert into a table with a foreign key is parallel-restricted,
@@ -152,6 +186,29 @@ insert into para_insert_f1 select unique1, stringu1 from tenk1;
 -- select some values to verify that the insert worked
 select count(*), sum(unique1) from para_insert_f1;
 
+--
+-- Test INSERT with underlying query, leader participation disabled
+--
+set parallel_leader_participation = off;
+truncate para_insert_p1 cascade;
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+select count(*), sum(unique1) from para_insert_p1;
+select * from para_insert_p1 where unique1 >= 2490 order by unique1;
+
+--
+-- Test INSERT with underlying query, leader participation disabled
+--  and no workers available
+set max_parallel_workers=0;
+truncate para_insert_p1 cascade;
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 <= 2500;
+select count(*), sum(unique1) from para_insert_p1;
+select * from para_insert_p1 where unique1 >= 2490 order by unique1;
+
+reset parallel_leader_participation;
+reset max_parallel_workers;
+
 --
 -- Test INSERT with ON CONFLICT ... DO UPDATE ...
 -- (should not create a parallel plan)
@@ -161,6 +218,70 @@ explain (costs off) insert into test_conflict_table(id, somedata) select a, a fr
 insert into test_conflict_table(id, somedata) select a, a from test_data;
 explain (costs off) insert into test_conflict_table(id, somedata) select a, a from test_data ON CONFLICT(id) DO UPDATE SET somedata = EXCLUDED.somedata + 1;
 
+--
+-- Test INSERT with parallelized aggregate
+--
+create table tenk1_avg_data(count int, avg_unique1 int, avg_stringu1_len int);
+explain (costs off) insert into tenk1_avg_data select count(*), avg(unique1), avg(length(stringu1)) from tenk1;
+insert into tenk1_avg_data select count(*), avg(unique1), avg(length(stringu1)) from tenk1;
+select * from tenk1_avg_data;
+
+--
+-- Test INSERT with parallel bitmap heap scan
+--
+set enable_seqscan to off;
+set enable_indexscan to off;
+truncate para_insert_p1 cascade;
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 7500;
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 7500;
+-- select some values to verify that the insert worked
+select * from para_insert_p1 where unique1 >= 9990 order by unique1;
+reset enable_seqscan;
+reset enable_indexscan;
+
+--
+-- Test INSERT with parallel append
+--
+create table a_star_data(aa int);
+explain (costs off) insert into a_star_data select aa from a_star where aa > 10;
+insert into a_star_data select aa from a_star where aa > 10;
+select count(aa), sum(aa) from a_star_data;
+
+--
+-- Test INSERT with parallel index scan
+--
+set enable_seqscan to off;
+set enable_bitmapscan to off;
+set min_parallel_index_scan_size=0;
+
+truncate para_insert_p1 cascade;
+explain (costs off) insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 500;
+insert into para_insert_p1 select unique1, stringu1 from tenk1 where unique1 >= 500;
+-- select some values to verify that the parallel insert worked
+select count(*), sum(unique1) from para_insert_p1;
+select * from para_insert_p1 where unique1 >= 9990 order by unique1;
+
+--
+-- Test INSERT with parallel index-only scan
+--
+truncate para_insert_p1 cascade;
+explain (costs off) insert into para_insert_p1 select unique1 from tenk1 where unique1 >= 500;
+insert into para_insert_p1 select unique1 from tenk1 where unique1 >= 500;
+-- select some values to verify that the parallel insert worked
+select count(*), sum(unique1) from para_insert_p1;
+select unique1 from para_insert_p1 where unique1 >= 9990 order by unique1;
+
+reset min_parallel_index_scan_size;
+reset enable_seqscan;
+reset enable_bitmapscan;
+
+--
+-- Test INSERT with parallel-safe index expression
+-- (should create a parallel plan)
+--
+explain (costs off) insert into names3 select * from names;
+insert into names3 select * from names;
+select * from names3 order by fullname_parallel_safe(first_name, last_name);
 
 --
 -- Test INSERT with parallel-unsafe index expression
@@ -197,6 +318,31 @@ create table names7 (like names);
 explain (costs off) insert into names7 select * from names order by last_name returning last_name || ', ' || first_name as last_name_then_first_name;
 insert into names7 select * from names order by last_name returning last_name || ', ' || first_name as last_name_then_first_name;
 
+--
+-- Test INSERT with parallel-safe index predicate
+-- (should create a parallel plan)
+--
+create table names8 (like names);
+create index names8_lastname_partial_idx on names8(index, last_name) where lastname_startswithe_s(last_name);
+explain (costs off) insert into names8 select * from names;
+insert into names8 select * from names;
+
+--
+-- Test INSERT with parallel-unsafe index predicate
+-- (should not create a parallel plan)
+--
+create table names9 (like names);
+create index names9_lastname_partial_idx on names9(index, last_name) where lastname_startswithe_u(last_name);
+explain (costs off) insert into names9 select * from names;
+
+--
+-- Test INSERT with parallel-restricted index predicate
+-- (should create a parallel plan)
+--
+create table names10 (like names);
+create index names10_lastname_partial_idx on names10(index, last_name) where lastname_startswithe_r(last_name);
+explain (costs off) insert into names10 select * from names;
+insert into names10 select * from names;
 
 --
 -- Test INSERT into temporary table with underlying query.
@@ -211,6 +357,19 @@ insert into temp_names select * from names;
 -- Test INSERT with column defaults
 --
 --
+-- a: no default
+-- b: unsafe default
+-- c: restricted default
+-- d: safe default
+--
+
+--
+-- No column defaults, should use parallel SELECT
+--
+explain (costs off) insert into testdef(a,b,c,d) select a,a*2,a*4,a*8 from test_data;
+insert into testdef(a,b,c,d) select a,a*2,a*4,a*8 from test_data;
+select * from testdef order by a;
+truncate testdef;
 
 --
 -- Parallel unsafe column default, should not use a parallel plan
@@ -225,6 +384,14 @@ insert into testdef(a,b,d) select a,a*2,a*8 from test_data;
 select * from testdef order by a;
 truncate testdef;
 
+--
+-- Parallel safe column default, should use parallel SELECT
+--
+explain (costs off) insert into testdef(a,b,c) select a,a*2,a*4 from test_data;
+insert into testdef(a,b,c) select a,a*2,a*4 from test_data;
+select * from testdef order by a;
+truncate testdef;
+
 --
 -- Parallel restricted and unsafe column defaults, should not use a parallel plan
 --
@@ -257,6 +424,46 @@ insert into parttable1 select unique1,stringu1 from tenk1;
 select count(*) from parttable1_1;
 select count(*) from parttable1_2;
 
+--
+-- Test INSERT into partition with parallel-unsafe partition key support function
+-- (should not create a parallel plan)
+--
+create function my_int4_sort(int4,int4) returns int language sql
+  as $$ select case when $1 = $2 then 0 when $1 > $2 then 1 else -1 end; $$;
+
+create operator class test_int4_ops for type int4 using btree as
+  operator 1 < (int4,int4), operator 2 <= (int4,int4),
+  operator 3 = (int4,int4), operator 4 >= (int4,int4),
+  operator 5 > (int4,int4), function 1 my_int4_sort(int4,int4);
+
+create table partkey_unsafe_key_supp_fn_t (a int4, b name) partition by range (a test_int4_ops);
+create table partkey_unsafe_key_supp_fn_t_1 partition of partkey_unsafe_key_supp_fn_t for values from (0) to (5000);
+create table partkey_unsafe_key_supp_fn_t_2 partition of partkey_unsafe_key_supp_fn_t for values from (5000) to (10000);
+
+explain (costs off) insert into partkey_unsafe_key_supp_fn_t select unique1, stringu1 from tenk1;
+
+--
+-- Test INSERT into partition with parallel-unsafe partition key expression
+-- (should not create a parallel plan)
+--
+create table partkey_unsafe_key_expr_t (a int4, b name) partition by range ((fullname_parallel_unsafe('',a::varchar)));
+explain (costs off) insert into partkey_unsafe_key_expr_t select unique1, stringu1 from tenk1;
+
+--
+-- Test INSERT into table with parallel-safe check constraint
+-- (should create a parallel plan)
+--
+create or replace function check_a(a int4) returns boolean as $$
+    begin
+        return (a >= 0 and a <= 9999);
+    end;
+$$ language plpgsql parallel safe;
+
+create table table_check_a(a int4 check (check_a(a)), b name);
+explain (costs off) insert into table_check_a select unique1, stringu1 from tenk1;
+insert into table_check_a select unique1, stringu1 from tenk1;
+select count(*), sum(a) from table_check_a;
+
 --
 -- Test INSERT into table with parallel-unsafe check constraint
 -- (should not create a parallel plan)
@@ -271,37 +478,78 @@ create table table_check_b(a int4, b name check (check_b_unsafe(b)), c name);
 explain (costs off) insert into table_check_b(a,b,c) select unique1, unique2, stringu1 from tenk1;
 
 --
--- Test INSERT into table with parallel-safe after stmt-level triggers
+-- Test INSERT into table with parallel-safe before+after stmt-level triggers
 -- (should create a parallel SELECT plan; triggers should fire)
 --
 create table names_with_safe_trigger (like names);
+create or replace function insert_before_trigger_safe() returns trigger as $$
+    begin
+        raise notice 'hello from insert_before_trigger_safe';
+		return new;
+    end;
+$$ language plpgsql parallel safe;
 create or replace function insert_after_trigger_safe() returns trigger as $$
     begin
         raise notice 'hello from insert_after_trigger_safe';
 		return new;
     end;
 $$ language plpgsql parallel safe;
+create trigger insert_before_trigger_safe before insert on names_with_safe_trigger
+    for each statement execute procedure insert_before_trigger_safe();
 create trigger insert_after_trigger_safe after insert on names_with_safe_trigger
     for each statement execute procedure insert_after_trigger_safe();
 explain (costs off) insert into names_with_safe_trigger select * from names;
 insert into names_with_safe_trigger select * from names;
 
 --
--- Test INSERT into table with parallel-unsafe after stmt-level triggers
+-- Test INSERT into table with parallel-unsafe before+after stmt-level triggers
 -- (should not create a parallel plan; triggers should fire)
 --
 create table names_with_unsafe_trigger (like names);
+create or replace function insert_before_trigger_unsafe() returns trigger as $$
+    begin
+        raise notice 'hello from insert_before_trigger_unsafe';
+		return new;
+    end;
+$$ language plpgsql parallel unsafe;
 create or replace function insert_after_trigger_unsafe() returns trigger as $$
     begin
         raise notice 'hello from insert_after_trigger_unsafe';
 		return new;
     end;
 $$ language plpgsql parallel unsafe;
+create trigger insert_before_trigger_unsafe before insert on names_with_unsafe_trigger
+    for each statement execute procedure insert_before_trigger_unsafe();
 create trigger insert_after_trigger_unsafe after insert on names_with_unsafe_trigger
     for each statement execute procedure insert_after_trigger_unsafe();
 explain (costs off) insert into names_with_unsafe_trigger select * from names;
 insert into names_with_unsafe_trigger select * from names;
 
+--
+-- Test INSERT into table with parallel-restricted before+after stmt-level trigger
+-- (should create a parallel plan with parallel SELECT;
+--  stmt-level before+after triggers should fire)
+--
+create table names_with_restricted_trigger (like names);
+create or replace function insert_before_trigger_restricted() returns trigger as $$
+    begin
+        raise notice 'hello from insert_before_trigger_restricted';
+		return new;
+    end;
+$$ language plpgsql parallel restricted;
+create or replace function insert_after_trigger_restricted() returns trigger as $$
+    begin
+        raise notice 'hello from insert_after_trigger_restricted';
+		return new;
+    end;
+$$ language plpgsql parallel restricted;
+create trigger insert_before_trigger_restricted before insert on names_with_restricted_trigger
+    for each statement execute procedure insert_before_trigger_restricted();
+create trigger insert_after_trigger_restricted after insert on names_with_restricted_trigger
+    for each statement execute procedure insert_after_trigger_restricted();
+explain (costs off) insert into names_with_restricted_trigger select * from names;
+insert into names_with_restricted_trigger select * from names;
+
 --
 -- Test INSERT into partition with parallel-unsafe trigger
 -- (should not create a parallel plan)
@@ -335,6 +583,17 @@ function make_table_bar();
 -- should create a non-parallel plan
 explain (costs off) execute q;
 
+--
+-- Test INSERT into table with TOAST column
+--
+create table insert_toast_table(index int4, data text);
+create table insert_toast_table_data (like insert_toast_table);
+insert into insert_toast_table_data select i, rpad('T', 16384, 'ABCDEFGH') from generate_series(1,20) as i;
+explain (costs off) insert into insert_toast_table select index, data from insert_toast_table_data;
+insert into insert_toast_table select index, data from insert_toast_table_data;
+select count(*) as row_count, sum(length(data)) as total_data_length from insert_toast_table;
+truncate insert_toast_table;
+
 --
 -- Test INSERT into table having a DOMAIN column with a CHECK constraint
 --
@@ -342,15 +601,48 @@ create function sql_is_distinct_from_u(anyelement, anyelement)
 returns boolean language sql parallel unsafe
 as 'select $1 is distinct from $2 limit 1';
 
+create or replace function sql_is_distinct_from_r(a anyelement, b anyelement) returns boolean as $$
+    begin
+        return (a <> b);
+    end;
+$$ language plpgsql parallel restricted;
+
+create or replace function sql_is_distinct_from_s(a anyelement, b anyelement) returns boolean as $$
+    begin
+        return (a <> b);
+    end;
+$$ language plpgsql parallel safe;
+
 create domain inotnull_u int
   check (sql_is_distinct_from_u(value, null));
 
+create domain inotnull_r int
+  check (sql_is_distinct_from_r(value, null));
+
+create domain inotnull_s int
+  check (sql_is_distinct_from_s(value, null));
+
 create table dom_table_u (x inotnull_u, y int);
+create table dom_table_r (x inotnull_r, y int);
+create table dom_table_s (x inotnull_s, y int);
 
 
 -- Test INSERT into table having a DOMAIN column with parallel-unsafe CHECK constraint
 explain (costs off) insert into dom_table_u select unique1, unique2 from tenk1;
 
+-- Test INSERT into table having a DOMAIN column with parallel-restricted CHECK constraint
+explain (costs off) insert into dom_table_r select unique1, unique2 from tenk1;
+insert into dom_table_r select unique1, unique2 from tenk1;
+select count(*), sum(x) as sum_x, sum(y) as sum_y from dom_table_r;
+
+-- Test INSERT into table having a DOMAIN column with parallel-safe CHECK constraint
+-- NOTE: Currently max_parallel_hazard() regards CoerceToDomain as parallel-restricted
+explain (costs off) insert into dom_table_s select unique1, unique2 from tenk1;
+insert into dom_table_s select unique1, unique2 from tenk1;
+select count(*), sum(x) as sum_x, sum(y) as sum_y from dom_table_s;
+
+
+
 
 rollback;
 
@@ -361,6 +653,8 @@ rollback;
 drop table names;
 drop index names2_fullname_idx;
 drop table names2;
+drop index names3_fullname_idx;
+drop table names3;
 drop index names4_fullname_idx;
 drop table names4;
 drop table testdef;
@@ -370,4 +664,8 @@ drop function bdefault_unsafe;
 drop function cdefault_restricted;
 drop function ddefault_safe;
 drop function fullname_parallel_unsafe;
+drop function fullname_parallel_safe;
 drop function fullname_parallel_restricted;
+drop function lastname_startswithe_u;
+drop function lastname_startswithe_s;
+drop function lastname_startswithe_r;
-- 
2.27.0

