v8-0005-Tests-for-parallel-copy.patch

From 3d72c381b024c8eb93324e250c1e628aa78965d0 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddy@enterprisedb.com>
Date: Tue, 13 Oct 2020 14:23:10 +0530
Subject: [PATCH v8 5/6] Tests for parallel copy.

This patch has the tests for parallel copy:
---
 contrib/postgres_fdw/expected/postgres_fdw.out |  49 ++++
 contrib/postgres_fdw/sql/postgres_fdw.sql      |  52 ++++
 src/backend/commands/copyparallel.c            | 369 +++++++++++++++++++++++--
 src/test/regress/expected/copy2.out            | 326 +++++++++++++++++++++-
 src/test/regress/input/copy.source             |  31 +++
 src/test/regress/output/copy.source            |  27 ++
 src/test/regress/sql/copy2.sql                 | 368 +++++++++++++++++++++++-
 7 files changed, 1191 insertions(+), 31 deletions(-)

diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 2d88d06..474c5e7 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -9033,5 +9033,54 @@ SELECT 1 FROM ft1 LIMIT 1;    -- should fail
 ERROR:  08006
 \set VERBOSITY default
 COMMIT;
+-- parallel copy related tests.
+CREATE TABLE test_parallel_copy (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) ;
+CREATE FOREIGN TABLE test_parallel_copy_ft (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) SERVER loopback OPTIONS (table_name 'test_parallel_copy') ;
+-- parallel copy into foreign table, parallelism must not be picked up.
+COPY test_parallel_copy_ft FROM stdin WITH (PARALLEL 1);
+SELECT count(*) FROM test_parallel_copy_ft;
+ count 
+-------
+     2
+(1 row)
+
+-- parallel copy into a table with foreign partition.
+CREATE TABLE part_test_parallel_copy (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) PARTITION BY LIST (b);
+CREATE FOREIGN TABLE part_test_parallel_copy_a1 (c TEXT, b INT, a INT, e TEXT, d TEXT) SERVER loopback;
+CREATE TABLE part_test_parallel_copy_a2 (a INT, c TEXT, b INT, d TEXT, e TEXT);
+ALTER TABLE part_test_parallel_copy ATTACH PARTITION part_test_parallel_copy_a1 FOR VALUES IN(1);
+ALTER TABLE part_test_parallel_copy ATTACH PARTITION part_test_parallel_copy_a2 FOR VALUES IN(2);
+COPY part_test_parallel_copy FROM stdin WITH (PARALLEL 1);
+ERROR:  cannot perform PARALLEL COPY if partition has BEFORE/INSTEAD OF triggers, or if the partition is foreign partition
+HINT:  Try COPY without PARALLEL option
+CONTEXT:  COPY part_test_parallel_copy, line 1: "1	1	test_c1	test_d1	test_e1"
+parallel worker
+SELECT count(*) FROM part_test_parallel_copy WHERE b = 2;
+ count 
+-------
+     0
+(1 row)
+
 -- Clean up
 DROP PROCEDURE terminate_backend_and_wait(text);
+DROP FOREIGN TABLE test_parallel_copy_ft;
+DROP TABLE test_parallel_copy;
+DROP TABLE part_test_parallel_copy;
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 7581c54..635fcc2 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -2695,5 +2695,57 @@ SELECT 1 FROM ft1 LIMIT 1;    -- should fail
 \set VERBOSITY default
 COMMIT;
 
+-- parallel copy related tests.
+CREATE TABLE test_parallel_copy (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) ;
+
+CREATE FOREIGN TABLE test_parallel_copy_ft (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) SERVER loopback OPTIONS (table_name 'test_parallel_copy') ;
+
+-- parallel copy into foreign table, parallelism must not be picked up.
+COPY test_parallel_copy_ft FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+SELECT count(*) FROM test_parallel_copy_ft;
+
+-- parallel copy into a table with foreign partition.
+CREATE TABLE part_test_parallel_copy (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) PARTITION BY LIST (b);
+
+CREATE FOREIGN TABLE part_test_parallel_copy_a1 (c TEXT, b INT, a INT, e TEXT, d TEXT) SERVER loopback;
+
+CREATE TABLE part_test_parallel_copy_a2 (a INT, c TEXT, b INT, d TEXT, e TEXT);
+
+ALTER TABLE part_test_parallel_copy ATTACH PARTITION part_test_parallel_copy_a1 FOR VALUES IN(1);
+
+ALTER TABLE part_test_parallel_copy ATTACH PARTITION part_test_parallel_copy_a2 FOR VALUES IN(2);
+
+COPY part_test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+SELECT count(*) FROM part_test_parallel_copy WHERE b = 2;
+
 -- Clean up
 DROP PROCEDURE terminate_backend_and_wait(text);
+DROP FOREIGN TABLE test_parallel_copy_ft;
+DROP TABLE test_parallel_copy;
+DROP TABLE part_test_parallel_copy;
diff --git a/src/backend/commands/copyparallel.c b/src/backend/commands/copyparallel.c
index 436df40..add4dbc 100644
--- a/src/backend/commands/copyparallel.c
+++ b/src/backend/commands/copyparallel.c
@@ -102,6 +102,7 @@ SerializeParallelCopyState(ParallelContext *pcxt, CopyState cstate,
 	shared_cstate.convert_selectively = cstate->convert_selectively;
 	shared_cstate.num_defaults = cstate->num_defaults;
 	shared_cstate.relid = cstate->pcdata->relid;
+	shared_cstate.binary = cstate->binary;
 
 	memcpy(shmptr, (char *) &shared_cstate, sizeof(SerializedParallelCopyState));
 	copiedsize = sizeof(SerializedParallelCopyState);
@@ -226,6 +227,7 @@ RestoreParallelCopyState(shm_toc *toc, CopyState cstate, List **attlist)
 	cstate->convert_selectively = shared_cstate.convert_selectively;
 	cstate->num_defaults = shared_cstate.num_defaults;
 	cstate->pcdata->relid = shared_cstate.relid;
+	cstate->binary = shared_cstate.binary;
 
 	cstate->null_print = CopyStringFromSharedMemory(shared_str_val + copiedsize,
 													&copiedsize);
@@ -438,10 +440,6 @@ CheckExprParallelSafety(CopyState cstate)
 bool
 IsParallelCopyAllowed(CopyState cstate)
 {
-	/* Parallel copy not allowed for binary option. */
-	if (cstate->binary)
-		return false;
-
 	/*
 	 * Check if copy is into foreign table. We can not allow parallelism in
 	 * this case because each worker needs to establish FDW connection and
@@ -656,6 +654,7 @@ InitializeParallelCopyInfo(CopyState cstate, List *attnamelist)
 	cstate->cur_lineno = 0;
 	cstate->cur_attname = NULL;
 	cstate->cur_attval = NULL;
+	cstate->pcdata->curr_data_block = NULL;
 
 	/* Set up variables to avoid per-attribute overhead. */
 	initStringInfo(&cstate->attribute_buf);
@@ -878,7 +877,11 @@ return_line:
 
 	/* Mark that encoding conversion hasn't occurred yet. */
 	cstate->line_buf_converted = false;
-	ConvertToServerEncoding(cstate);
+
+	/* For binary format data, we don't need conversion. */
+	if (!cstate->binary)
+		ConvertToServerEncoding(cstate);
+
 	pcdata->worker_line_buf_pos++;
 	return false;
 }
@@ -1034,33 +1037,70 @@ ParallelCopyFrom(CopyState cstate)
 	/* Execute the before statement triggers from the leader */
 	ExecBeforeStmtTrigger(cstate);
 
-	/* On input just throw the header line away. */
-	if (cstate->cur_lineno == 0 && cstate->header_line)
+	if (!cstate->binary)
 	{
-		cstate->cur_lineno++;
-		if (CopyReadLine(cstate))
+		/* On input just throw the header line away. */
+		if (cstate->cur_lineno == 0 && cstate->header_line)
 		{
-			pcshared_info->is_read_in_progress = false;
-			return;				/* done */
+			cstate->cur_lineno++;
+			if (CopyReadLine(cstate))
+			{
+				pcshared_info->is_read_in_progress = false;
+				return;			/* done */
+			}
 		}
-	}
 
-	for (;;)
-	{
-		bool		done;
+		for (;;)
+		{
+			bool		done;
 
-		cstate->cur_lineno++;
+			cstate->cur_lineno++;
 
-		/* Actually read the line into memory here. */
-		done = CopyReadLine(cstate);
+			/* Actually read the line into memory here. */
+			done = CopyReadLine(cstate);
 
+			/*
+			 * EOF at start of line means we're done.  If we see EOF after
+			 * some characters, we act as though it was newline followed by
+			 * EOF, ie, process the line and then exit loop on next iteration.
+			 */
+			if (done && cstate->line_buf.len == 0)
+				break;
+		}
+	}
+	else
+	{
 		/*
-		 * EOF at start of line means we're done.  If we see EOF after some
-		 * characters, we act as though it was newline followed by EOF, ie,
-		 * process the line and then exit loop on next iteration.
+		 * Binary Format Files. In parallel copy leader, fill in the error
+		 * context information here, in case any failures while determining
+		 * tuple offsets, leader would throw the errors with proper context.
 		 */
-		if (done && cstate->line_buf.len == 0)
-			break;
+		ErrorContextCallback errcallback;
+
+		errcallback.callback = CopyFromErrorCallback;
+		errcallback.arg = (void *) cstate;
+		errcallback.previous = error_context_stack;
+		error_context_stack = &errcallback;
+		cstate->pcdata->curr_data_block = NULL;
+		cstate->raw_buf_index = 0;
+		pcshared_info->populated = 0;
+		cstate->cur_lineno = 0;
+		cstate->max_fields = list_length(cstate->attnumlist);
+
+		for (;;)
+		{
+			bool		eof = false;
+
+			cstate->cur_lineno++;
+
+			eof = CopyReadBinaryTupleLeader(cstate);
+
+			if (eof)
+				break;
+		}
+
+		/* Reset the error context. */
+		error_context_stack = errcallback.previous;
 	}
 
 	/*
@@ -1075,6 +1115,289 @@ ParallelCopyFrom(CopyState cstate)
 }
 
 /*
+ * CopyReadBinaryGetDataBlock
+ *
+ * Gets a new block, updates the current offset, calculates the skip bytes.
+ */
+void
+CopyReadBinaryGetDataBlock(CopyState cstate, FieldInfoType field_info)
+{
+	ParallelCopyDataBlock *data_block = NULL;
+	ParallelCopyDataBlock *curr_data_block = cstate->pcdata->curr_data_block;
+	ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info;
+	uint8		move_bytes = 0;
+	uint32		block_pos;
+	uint32		prev_block_pos;
+	int			read_bytes = 0;
+
+	prev_block_pos = pcshared_info->cur_block_pos;
+
+	block_pos = WaitGetFreeCopyBlock(pcshared_info);
+
+	if (field_info == FIELD_SIZE || field_info == FIELD_COUNT)
+		move_bytes = (DATA_BLOCK_SIZE - cstate->raw_buf_index);
+
+	if (curr_data_block != NULL)
+		curr_data_block->skip_bytes = move_bytes;
+
+	data_block = &pcshared_info->data_blocks[block_pos];
+
+	if (move_bytes > 0 && curr_data_block != NULL)
+		memmove(&data_block->data[0], &curr_data_block->data[cstate->raw_buf_index], move_bytes);
+
+	elog(DEBUG1, "LEADER - field info %d is spread across data blocks - moved %d bytes from current block %u to %u block",
+		 field_info, move_bytes, prev_block_pos, block_pos);
+
+	read_bytes = CopyGetData(cstate, &data_block->data[move_bytes], 1, (DATA_BLOCK_SIZE - move_bytes));
+
+	if (field_info == FIELD_NONE && cstate->reached_eof)
+		return;
+
+	if (cstate->reached_eof)
+		EOF_ERROR;
+
+	elog(DEBUG1, "LEADER - bytes read from file %d", read_bytes);
+
+	if (field_info == FIELD_SIZE || field_info == FIELD_DATA)
+	{
+		ParallelCopyDataBlock *prev_data_block = NULL;
+
+		prev_data_block = curr_data_block;
+		prev_data_block->following_block = block_pos;
+
+		if (prev_data_block->curr_blk_completed == false)
+			prev_data_block->curr_blk_completed = true;
+
+		pg_atomic_add_fetch_u32(&prev_data_block->unprocessed_line_parts, 1);
+	}
+
+	cstate->pcdata->curr_data_block = data_block;
+	cstate->raw_buf_index = 0;
+}
+
+/*
+ * CopyReadBinaryTupleLeader
+ *
+ * Leader reads data from binary formatted file to data blocks and identifies
+ * tuple boundaries/offsets so that workers can work on the data blocks data.
+ */
+bool
+CopyReadBinaryTupleLeader(CopyState cstate)
+{
+	ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info;
+	int16		fld_count;
+	uint32		line_size = 0;
+	uint32		start_block_pos;
+	uint32		start_offset;
+
+	if (cstate->pcdata->curr_data_block == NULL)
+	{
+		CopyReadBinaryGetDataBlock(cstate, FIELD_NONE);
+
+		/*
+		 * no data is read from file here. one possibility to be here could be
+		 * that the binary file just has a valid signature but nothing else.
+		 */
+		if (cstate->reached_eof)
+			return true;
+	}
+
+	if ((cstate->raw_buf_index + sizeof(fld_count)) >= DATA_BLOCK_SIZE)
+		CopyReadBinaryGetDataBlock(cstate, FIELD_COUNT);
+
+	memcpy(&fld_count, &cstate->pcdata->curr_data_block->data[cstate->raw_buf_index], sizeof(fld_count));
+	fld_count = (int16) pg_ntoh16(fld_count);
+
+	CHECK_FIELD_COUNT;
+
+	start_offset = cstate->raw_buf_index;
+	cstate->raw_buf_index += sizeof(fld_count);
+	line_size += sizeof(fld_count);
+	start_block_pos = pcshared_info->cur_block_pos;
+
+	CopyReadBinaryFindTupleSize(cstate, &line_size);
+
+	pg_atomic_add_fetch_u32(&cstate->pcdata->curr_data_block->unprocessed_line_parts, 1);
+
+	if (line_size > 0)
+		(void) UpdateSharedLineInfo(cstate, start_block_pos, start_offset,
+									line_size, LINE_LEADER_POPULATED, -1);
+
+	return false;
+}
+
+/*
+ * CopyReadBinaryFindTupleSize
+ *
+ * Leader identifies boundaries/offsets for each attribute/column and finally
+ * results in the tuple/row size. It moves on to next data block if the
+ * attribute/column is spread across data blocks.
+ */
+void
+CopyReadBinaryFindTupleSize(CopyState cstate, uint32 *line_size)
+{
+	int32		fld_size;
+	ListCell   *cur;
+	TupleDesc	tup_desc = RelationGetDescr(cstate->rel);
+
+	foreach(cur, cstate->attnumlist)
+	{
+		int			att_num = lfirst_int(cur);
+		Form_pg_attribute att = TupleDescAttr(tup_desc, (att_num - 1));
+
+		cstate->cur_attname = NameStr(att->attname);
+		fld_size = 0;
+
+		if ((cstate->raw_buf_index + sizeof(fld_size)) >= DATA_BLOCK_SIZE)
+			CopyReadBinaryGetDataBlock(cstate, FIELD_SIZE);
+
+		memcpy(&fld_size, &cstate->pcdata->curr_data_block->data[cstate->raw_buf_index], sizeof(fld_size));
+		cstate->raw_buf_index += sizeof(fld_size);
+		*line_size += sizeof(fld_size);
+		fld_size = (int32) pg_ntoh32(fld_size);
+
+		/* fld_size -1 represents the null value for the field. */
+		if (fld_size == -1)
+			continue;
+
+		CHECK_FIELD_SIZE(fld_size);
+
+		*line_size += fld_size;
+
+		if ((DATA_BLOCK_SIZE - cstate->raw_buf_index) >= fld_size)
+		{
+			cstate->raw_buf_index += fld_size;
+			elog(DEBUG1, "LEADER - tuple lies in he same data block");
+		}
+		else
+		{
+			int32		required_blks = 0;
+			int32		curr_blk_bytes = (DATA_BLOCK_SIZE - cstate->raw_buf_index);
+			int			i = 0;
+
+			GET_REQUIRED_BLOCKS(required_blks, fld_size, curr_blk_bytes);
+
+			i = required_blks;
+
+			while (i > 0)
+			{
+				CopyReadBinaryGetDataBlock(cstate, FIELD_DATA);
+				i--;
+			}
+
+			GET_RAW_BUF_INDEX(cstate->raw_buf_index, fld_size, required_blks, curr_blk_bytes);
+
+			/*
+			 * raw_buf_index should never cross data block size, as the
+			 * required number of data blocks would have been obtained in the
+			 * above while loop.
+			 */
+			Assert(cstate->raw_buf_index <= DATA_BLOCK_SIZE);
+		}
+		cstate->cur_attname = NULL;
+	}
+}
+
+/*
+ * CopyReadBinaryTupleWorker
+ *
+ * Each worker reads data from data blocks caches the tuple data into local
+ * memory.
+ */
+bool
+CopyReadBinaryTupleWorker(CopyState cstate, Datum *values, bool *nulls)
+{
+	int16		fld_count;
+	ListCell   *cur;
+	FmgrInfo   *in_functions = cstate->in_functions;
+	Oid		   *typioparams = cstate->typioparams;
+	TupleDesc	tup_desc = RelationGetDescr(cstate->rel);
+	bool 		done = false;
+
+	done = GetWorkerLine(cstate);
+	cstate->raw_buf_index = 0;
+
+	if (done && cstate->line_buf.len == 0)
+		return true;
+
+	memcpy(&fld_count, &cstate->line_buf.data[cstate->raw_buf_index], sizeof(fld_count));
+	fld_count = (int16) pg_ntoh16(fld_count);
+
+	CHECK_FIELD_COUNT;
+
+	cstate->raw_buf_index += sizeof(fld_count);
+
+	foreach(cur, cstate->attnumlist)
+	{
+		int			att_num = lfirst_int(cur);
+		int			m = att_num - 1;
+		Form_pg_attribute att = TupleDescAttr(tup_desc, m);
+
+		cstate->cur_attname = NameStr(att->attname);
+
+		values[m] = CopyReadBinaryAttributeWorker(cstate,
+												  &in_functions[m],
+												  typioparams[m],
+												  att->atttypmod,
+												  &nulls[m]);
+		cstate->cur_attname = NULL;
+	}
+
+	return false;
+}
+
+/*
+ * CopyReadBinaryAttributeWorker
+ *
+ * Worker identifies and converts each attribute/column data from binary to
+ * the data type of attribute/column.
+ */
+Datum
+CopyReadBinaryAttributeWorker(CopyState cstate, FmgrInfo *flinfo,
+							  Oid typioparam, int32 typmod, bool *isnull)
+{
+	int32		fld_size;
+	Datum		result;
+
+	memcpy(&fld_size, &cstate->line_buf.data[cstate->raw_buf_index], sizeof(fld_size));
+	cstate->raw_buf_index += sizeof(fld_size);
+	fld_size = (int32) pg_ntoh32(fld_size);
+
+	/* fld_size -1 represents the null value for the field. */
+	if (fld_size == -1)
+	{
+		*isnull = true;
+		return ReceiveFunctionCall(flinfo, NULL, typioparam, typmod);
+	}
+
+	CHECK_FIELD_SIZE(fld_size);
+
+	/* Reset attribute_buf to empty, and load raw data in it */
+	resetStringInfo(&cstate->attribute_buf);
+
+	enlargeStringInfo(&cstate->attribute_buf, fld_size);
+
+	memcpy(&cstate->attribute_buf.data[0], &cstate->line_buf.data[cstate->raw_buf_index], fld_size);
+	cstate->raw_buf_index += fld_size;
+
+	cstate->attribute_buf.len = fld_size;
+	cstate->attribute_buf.data[fld_size] = '\0';
+
+	/* Call the column type's binary input converter */
+	result = ReceiveFunctionCall(flinfo, &cstate->attribute_buf,
+								 typioparam, typmod);
+
+	/* Trouble if it didn't eat the whole buffer */
+	if (cstate->attribute_buf.cursor != cstate->attribute_buf.len)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
+				 errmsg("incorrect binary data format")));
+
+	*isnull = false;
+	return result;
+}
+
+/*
  * GetLinePosition
  *
  * Return the line position once the leader has populated the data.
diff --git a/src/test/regress/expected/copy2.out b/src/test/regress/expected/copy2.out
index c64f071..08ce743 100644
--- a/src/test/regress/expected/copy2.out
+++ b/src/test/regress/expected/copy2.out
@@ -301,18 +301,32 @@ It is "perfect".|
 "It is ""perfect"".","	"
 "",
 --test that we read consecutive LFs properly
-CREATE TEMP TABLE testnl (a int, b text, c int);
+CREATE TABLE testnl (a int, b text, c int);
 COPY testnl FROM stdin CSV;
+COPY testnl FROM stdin WITH (FORMAT 'csv', PARALLEL 1);
+SELECT * FROM testnl;
+ a |          b           | c 
+---+----------------------+---
+ 1 | a field with two LFs+| 2
+   |                     +| 
+   | inside               | 
+ 1 | a field with two LFs+| 2
+   |                     +| 
+   | inside               | 
+(2 rows)
+
 -- test end of copy marker
-CREATE TEMP TABLE testeoc (a text);
+CREATE TABLE testeoc (a text);
 COPY testeoc FROM stdin CSV;
+TRUNCATE testeoc;
+COPY testeoc FROM stdin WITH (FORMAT 'csv', PARALLEL 1);
 COPY testeoc TO stdout CSV;
 a\.
 \.b
 c\.d
 "\."
 -- test handling of nonstandard null marker that violates escaping rules
-CREATE TEMP TABLE testnull(a int, b text);
+CREATE TABLE testnull(a int, b text);
 INSERT INTO testnull VALUES (1, E'\\0'), (NULL, NULL);
 COPY testnull TO stdout WITH NULL AS E'\\0';
 1	\\0
@@ -327,6 +341,15 @@ SELECT * FROM testnull;
     | 
 (4 rows)
 
+TRUNCATE testnull;
+COPY testnull FROM stdin WITH (NULL E'\\0', PARALLEL 1);
+SELECT * FROM testnull;
+ a  | b  
+----+----
+ 42 | \0
+    | 
+(2 rows)
+
 BEGIN;
 CREATE TABLE vistest (LIKE testeoc);
 COPY vistest FROM stdin CSV;
@@ -396,6 +419,34 @@ SELECT * FROM vistest;
 
 BEGIN;
 TRUNCATE vistest;
+COPY vistest FROM stdin WITH (FORMAT 'csv', FREEZE, PARALLEL 1);
+SELECT * FROM vistest;
+ a  
+----
+ a2
+ b
+(2 rows)
+
+SAVEPOINT s1;
+TRUNCATE vistest;
+COPY vistest FROM stdin WITH (FORMAT 'csv', FREEZE, PARALLEL 1);
+SELECT * FROM vistest;
+ a  
+----
+ d2
+ e
+(2 rows)
+
+COMMIT;
+SELECT * FROM vistest;
+ a  
+----
+ d2
+ e
+(2 rows)
+
+BEGIN;
+TRUNCATE vistest;
 COPY vistest FROM stdin CSV FREEZE;
 SELECT * FROM vistest;
  a 
@@ -456,7 +507,7 @@ SELECT * FROM vistest;
 (2 rows)
 
 -- Test FORCE_NOT_NULL and FORCE_NULL options
-CREATE TEMP TABLE forcetest (
+CREATE TABLE forcetest (
     a INT NOT NULL,
     b TEXT NOT NULL,
     c TEXT,
@@ -467,6 +518,8 @@ CREATE TEMP TABLE forcetest (
 -- should succeed with no effect ("b" remains an empty string, "c" remains NULL)
 BEGIN;
 COPY forcetest (a, b, c) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(b), FORCE_NULL(c));
+TRUNCATE forcetest;
+COPY forcetest (a, b, c) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(b), FORCE_NULL(c), PARALLEL 1);
 COMMIT;
 SELECT b, c FROM forcetest WHERE a = 1;
  b |  c   
@@ -477,6 +530,8 @@ SELECT b, c FROM forcetest WHERE a = 1;
 -- should succeed, FORCE_NULL and FORCE_NOT_NULL can be both specified
 BEGIN;
 COPY forcetest (a, b, c, d) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(c,d), FORCE_NULL(c,d));
+TRUNCATE forcetest;
+COPY forcetest (a, b, c, d) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(c,d), FORCE_NULL(c,d), PARALLEL 1);
 COMMIT;
 SELECT c, d FROM forcetest WHERE a = 2;
  c |  d   
@@ -533,6 +588,31 @@ select * from check_con_tbl;
    
 (2 rows)
 
+\d+ check_con_tbl
+                               Table "public.check_con_tbl"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ f1     | integer |           |          |         | plain   |              | 
+Check constraints:
+    "check_con_tbl_check" CHECK (check_con_function(check_con_tbl.*))
+
+truncate check_con_tbl;
+copy check_con_tbl from stdin with (parallel 1);
+NOTICE:  input = {"f1":1}
+NOTICE:  input = {"f1":null}
+copy check_con_tbl from stdin with (parallel 1);
+NOTICE:  input = {"f1":0}
+ERROR:  new row for relation "check_con_tbl" violates check constraint "check_con_tbl_check"
+DETAIL:  Failing row contains (0).
+CONTEXT:  COPY check_con_tbl, line 1: "0"
+parallel worker
+select * from check_con_tbl;
+ f1 
+----
+  1
+   
+(2 rows)
+
 -- test with RLS enabled.
 CREATE ROLE regress_rls_copy_user;
 CREATE ROLE regress_rls_copy_user_colperms;
@@ -647,10 +727,248 @@ SELECT * FROM instead_of_insert_tbl;
 (2 rows)
 
 COMMIT;
+-- Parallel copy tests.
+CREATE TABLE test_parallel_copy (
+        a INT,
+        b INT,
+        c TEXT not null default 'stuff',
+        d TEXT,
+        e TEXT
+) ;
+COPY test_parallel_copy (a, b, c, d, e) FROM stdin WITH (PARALLEL 1);
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL 1);
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL 1);
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL 1, FORMAT 'csv', HEADER);
+-- zero workers: should perform non-parallel copy
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL '0');
+ERROR:  value 0 out of bounds for option "parallel"
+DETAIL:  Valid values are between "1" and "1024".
+-- referencing table: should perform non-parallel copy
+CREATE TABLE test_copy_pk(c1 INT PRIMARY KEY);
+INSERT INTO test_copy_pk VALUES(10);
+CREATE TABLE test_copy_ri(c1 INT REFERENCES test_copy_pk(c1));
+COPY test_copy_ri FROM stdin WITH (FORMAT csv, DELIMITER ',', PARALLEL 1);
+-- expressions: should perform non-parallel copy
+CREATE TABLE test_copy_expr (index INT, height REAL, weight REAL);
+COPY test_copy_expr FROM STDIN WITH (FORMAT csv, DELIMITER ',', PARALLEL 1) WHERE height > random() * 65;
+-- serial data: should perform non-parallel copy
+CREATE TABLE testserial (index SERIAL, height REAL);
+COPY testserial(height) FROM STDIN WITH (FORMAT csv, DELIMITER ',', PARALLEL 1);
+-- temporary table copy: should perform non-parallel copy
+CREATE TEMPORARY TABLE temp_test(
+        a int
+) ;
+COPY temp_test (a) FROM stdin WITH (PARALLEL 1);
+-- non-existent column in column list: should fail
+COPY test_parallel_copy (xyz) FROM stdin WITH (PARALLEL 1);
+ERROR:  column "xyz" of relation "test_parallel_copy" does not exist
+-- too many columns in column list: should fail
+COPY test_parallel_copy (a, b, c, d, e, d, c) FROM stdin WITH (PARALLEL 1);
+ERROR:  column "d" specified more than once
+-- missing data: should fail
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+ERROR:  invalid input syntax for type integer: ""
+CONTEXT:  COPY test_parallel_copy, line 0, column a: ""
+parallel worker
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+ERROR:  missing data for column "e"
+CONTEXT:  COPY test_parallel_copy, line 1: "2000	230	23	23"
+parallel worker
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+ERROR:  missing data for column "e"
+CONTEXT:  COPY test_parallel_copy, line 1: "2001	231	\N	\N"
+parallel worker
+-- extra data: should fail
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+ERROR:  extra data after last expected column
+CONTEXT:  COPY test_parallel_copy, line 1: "2002	232	40	50	60	70	80"
+parallel worker
+-- various COPY options: delimiters, oids, NULL string, encoding
+COPY test_parallel_copy (b, c, d, e) FROM stdin WITH (DELIMITER ',', null 'x', PARALLEL 1) ;
+COPY test_parallel_copy FROM stdin WITH (DELIMITER ';', NULL '', PARALLEL 1);
+COPY test_parallel_copy FROM stdin WITH (DELIMITER ':', NULL E'\\X', ENCODING 'sql_ascii', PARALLEL 1);
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a = 50004;
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a > 60003;
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE f > 60003;
+ERROR:  column "f" does not exist
+LINE 1: ..._parallel_copy FROM stdin WITH (PARALLEL 1) WHERE f > 60003;
+                                                             ^
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a = max(x.b);
+ERROR:  missing FROM-clause entry for table "x"
+LINE 1: ...rallel_copy FROM stdin WITH (PARALLEL 1) WHERE a = max(x.b);
+                                                                  ^
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a IN (SELECT 1 FROM x);
+ERROR:  cannot use subquery in COPY FROM WHERE condition
+LINE 1: ...arallel_copy FROM stdin WITH (PARALLEL 1) WHERE a IN (SELECT...
+                                                             ^
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a IN (generate_series(1,5));
+ERROR:  set-returning functions are not allowed in COPY FROM WHERE conditions
+LINE 1: ...lel_copy FROM stdin WITH (PARALLEL 1) WHERE a IN (generate_s...
+                                                             ^
+COPY test_parallel_copy FROM stdin WITH(PARALLEL 1) WHERE a = row_number() over(b);
+ERROR:  window functions are not allowed in COPY FROM WHERE conditions
+LINE 1: ...rallel_copy FROM stdin WITH(PARALLEL 1) WHERE a = row_number...
+                                                             ^
+-- check results of copy in
+SELECT * FROM test_parallel_copy ORDER BY 1;
+   a   | b  |     c      |    d    |    e    
+-------+----+------------+---------+---------
+     1 | 11 | test_c1    | test_d1 | test_e1
+     2 | 12 | test_c2    | test_d2 | test_e2
+  3000 |    | c          |         | 
+  4000 |    | C          |         | 
+  4001 |  1 | empty      |         | 
+  4002 |  2 | null       |         | 
+  4003 |  3 | Backslash  | \       | \
+  4004 |  4 | BackslashX | \X      | \X
+  4005 |  5 | N          | N       | N
+  4006 |  6 | BackslashN | \N      | \N
+  4007 |  7 | XX         | XX      | XX
+  4008 |  8 | Delimiter  | :       | :
+ 50004 | 25 | 35         | 45      | 55
+ 60004 | 25 | 35         | 45      | 55
+ 60005 | 26 | 36         | 46      | 56
+       |  3 | stuff      | test_d3 | 
+       |  4 | stuff      | test_d4 | 
+       |  5 | stuff      | test_d5 | 
+       |    | ,          | \,      | \
+       |    | x          | \x      | \x
+       |    | 45         | 80      | 90
+(21 rows)
+
+-- parallel copy test for unlogged tables. should execute in parallel worker
+CREATE UNLOGGED TABLE test_parallel_copy_unlogged(
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+);
+COPY test_parallel_copy_unlogged FROM stdin WITH (PARALLEL 1);
+SELECT count(*) FROM test_parallel_copy_unlogged;
+ count 
+-------
+     2
+(1 row)
+
+-- parallel copy test for various trigger types
+TRUNCATE test_parallel_copy;
+-- parallel safe trigger function
+CREATE OR REPLACE FUNCTION parallel_copy_trig_func() RETURNS TRIGGER
+LANGUAGE plpgsql PARALLEL SAFE AS $$
+BEGIN
+  RETURN NEW;
+END;
+$$;
+-- before insert row trigger
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy
+FOR EACH ROW
+EXECUTE PROCEDURE parallel_copy_trig_func();
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+-- after insert row trigger
+CREATE TRIGGER parallel_copy_trig
+AFTER INSERT ON test_parallel_copy
+FOR EACH ROW
+EXECUTE PROCEDURE parallel_copy_trig_func();
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+-- before insert statement trigger
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+-- parallelism should be picked, since the trigger function is parallel safe
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+-- after insert statement trigger
+CREATE TRIGGER parallel_copy_trig
+AFTER INSERT ON test_parallel_copy
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+-- transition table is involved
+CREATE TRIGGER parallel_copy_trig
+AFTER INSERT ON test_parallel_copy REFERENCING NEW TABLE AS new_table
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+-- make trigger function parallel unsafe
+ALTER FUNCTION parallel_copy_trig_func PARALLEL UNSAFE;
+-- before statement trigger has a parallel unsafe function
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+SELECT count(*) FROM test_parallel_copy;
+ count 
+-------
+    12
+(1 row)
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+-- instead of insert trigger, no parallelism should be picked
+COPY instead_of_insert_tbl_view FROM stdin WITH(PARALLEL 1);;
+SELECT count(*) FROM instead_of_insert_tbl_view;
+ count 
+-------
+     1
+(1 row)
+
+-- parallel copy test for a partitioned table with a before insert trigger on
+-- one of the partition
+ALTER FUNCTION parallel_copy_trig_func PARALLEL SAFE;
+CREATE TABLE test_parallel_copy_part (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) PARTITION BY LIST (b);
+CREATE TABLE test_parallel_copy_part_a1 (c TEXT, b INT, a INT, e TEXT, d TEXT);
+CREATE TABLE test_parallel_copy_part_a2 (a INT, c TEXT, b INT, d TEXT, e TEXT);
+ALTER TABLE test_parallel_copy_part ATTACH PARTITION test_parallel_copy_part_a1 FOR VALUES IN(1);
+ALTER TABLE test_parallel_copy_part ATTACH PARTITION test_parallel_copy_part_a2 FOR VALUES IN(2);
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy_part_a2
+FOR EACH ROW
+EXECUTE PROCEDURE parallel_copy_trig_func();
+COPY test_parallel_copy_part FROM stdin WITH (PARALLEL 1);
+ERROR:  cannot perform PARALLEL COPY if partition has BEFORE/INSTEAD OF triggers, or if the partition is foreign partition
+HINT:  Try COPY without PARALLEL option
+CONTEXT:  COPY test_parallel_copy_part, line 2: "2	2	test_c2	test_d2	test_e2"
+parallel worker
+SELECT count(*) FROM test_parallel_copy_part;
+ count 
+-------
+     0
+(1 row)
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy_part_a2;
 -- clean up
+DROP TABLE test_copy_ri;
+DROP TABLE test_copy_pk;
+DROP TABLE test_copy_expr;
+DROP TABLE testeoc;
+DROP TABLE testnl;
 DROP TABLE forcetest;
+DROP TABLE test_parallel_copy;
+DROP TABLE test_parallel_copy_unlogged;
+DROP TABLE test_parallel_copy_part;
+DROP TABLE testserial;
+DROP TABLE testnull;
 DROP TABLE vistest;
 DROP FUNCTION truncate_in_subxact();
+DROP FUNCTION parallel_copy_trig_func();
 DROP TABLE x, y;
 DROP TABLE rls_t1 CASCADE;
 DROP ROLE regress_rls_copy_user;
diff --git a/src/test/regress/input/copy.source b/src/test/regress/input/copy.source
index a1d529a..cb39c66 100644
--- a/src/test/regress/input/copy.source
+++ b/src/test/regress/input/copy.source
@@ -15,6 +15,13 @@ DELETE FROM onek;
 
 COPY onek FROM '@abs_builddir@/results/onek.data';
 
+-- test parallel copy
+COPY tenk1 FROM '@abs_srcdir@/data/tenk.data' WITH (parallel 2);
+
+SELECT COUNT(*) FROM tenk1;
+
+TRUNCATE tenk1;
+
 COPY tenk1 FROM '@abs_srcdir@/data/tenk.data';
 
 COPY slow_emp4000 FROM '@abs_srcdir@/data/rect.data';
@@ -159,6 +166,30 @@ truncate parted_copytest;
 
 copy parted_copytest from '@abs_builddir@/results/parted_copytest.csv';
 
+-- Test parallel copy from with a partitioned table.
+truncate parted_copytest;
+
+copy parted_copytest from '@abs_builddir@/results/parted_copytest.csv' with (parallel 2);
+
+-- Test parallel copy from for toast table with csv and binary format file
+create table test_parallel_copy_toast(a1 int, b1 text, c1 text default null);
+
+insert into test_parallel_copy_toast select i, (repeat(md5(i::text), 100000)) FROM generate_series(1,2) AS i;
+
+copy test_parallel_copy_toast to '@abs_builddir@/results/parallelcopytoast.csv' with(format csv);
+
+copy test_parallel_copy_toast to '@abs_builddir@/results/parallelcopytoast.dat' with(format binary);
+
+truncate test_parallel_copy_toast;
+
+copy test_parallel_copy_toast from '@abs_builddir@/results/parallelcopytoast.csv' with(format csv, parallel 2);
+
+copy test_parallel_copy_toast from '@abs_builddir@/results/parallelcopytoast.dat' with(format binary, parallel 2);
+
+select count(*) from test_parallel_copy_toast;
+
+drop table test_parallel_copy_toast;
+
 -- Ensure COPY FREEZE errors for partitioned tables.
 begin;
 truncate parted_copytest;
diff --git a/src/test/regress/output/copy.source b/src/test/regress/output/copy.source
index 938d355..a5dca79 100644
--- a/src/test/regress/output/copy.source
+++ b/src/test/regress/output/copy.source
@@ -9,6 +9,15 @@ COPY onek FROM '@abs_srcdir@/data/onek.data';
 COPY onek TO '@abs_builddir@/results/onek.data';
 DELETE FROM onek;
 COPY onek FROM '@abs_builddir@/results/onek.data';
+-- test parallel copy
+COPY tenk1 FROM '@abs_srcdir@/data/tenk.data' WITH (parallel 2);
+SELECT COUNT(*) FROM tenk1;
+ count 
+-------
+ 10000
+(1 row)
+
+TRUNCATE tenk1;
 COPY tenk1 FROM '@abs_srcdir@/data/tenk.data';
 COPY slow_emp4000 FROM '@abs_srcdir@/data/rect.data';
 COPY person FROM '@abs_srcdir@/data/person.data';
@@ -113,6 +122,24 @@ insert into parted_copytest select x,1,'One' from generate_series(1011,1020) x;
 copy (select * from parted_copytest order by a) to '@abs_builddir@/results/parted_copytest.csv';
 truncate parted_copytest;
 copy parted_copytest from '@abs_builddir@/results/parted_copytest.csv';
+-- Test parallel copy from with a partitioned table.
+truncate parted_copytest;
+copy parted_copytest from '@abs_builddir@/results/parted_copytest.csv' with (parallel 2);
+-- Test parallel copy from for toast table with csv and binary format file
+create table test_parallel_copy_toast(a1 int, b1 text, c1 text default null);
+insert into test_parallel_copy_toast select i, (repeat(md5(i::text), 100000)) FROM generate_series(1,2) AS i;
+copy test_parallel_copy_toast to '@abs_builddir@/results/parallelcopytoast.csv' with(format csv);
+copy test_parallel_copy_toast to '@abs_builddir@/results/parallelcopytoast.dat' with(format binary);
+truncate test_parallel_copy_toast;
+copy test_parallel_copy_toast from '@abs_builddir@/results/parallelcopytoast.csv' with(format csv, parallel 2);
+copy test_parallel_copy_toast from '@abs_builddir@/results/parallelcopytoast.dat' with(format binary, parallel 2);
+select count(*) from test_parallel_copy_toast;
+ count 
+-------
+     4
+(1 row)
+
+drop table test_parallel_copy_toast;
 -- Ensure COPY FREEZE errors for partitioned tables.
 begin;
 truncate parted_copytest;
diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
index b3c16af..b3c9af3 100644
--- a/src/test/regress/sql/copy2.sql
+++ b/src/test/regress/sql/copy2.sql
@@ -171,7 +171,7 @@ COPY y TO stdout (FORMAT CSV, FORCE_QUOTE *);
 
 --test that we read consecutive LFs properly
 
-CREATE TEMP TABLE testnl (a int, b text, c int);
+CREATE TABLE testnl (a int, b text, c int);
 
 COPY testnl FROM stdin CSV;
 1,"a field with two LFs
@@ -179,8 +179,16 @@ COPY testnl FROM stdin CSV;
 inside",2
 \.
 
+COPY testnl FROM stdin WITH (FORMAT 'csv', PARALLEL 1);
+1,"a field with two LFs
+
+inside",2
+\.
+
+SELECT * FROM testnl;
+
 -- test end of copy marker
-CREATE TEMP TABLE testeoc (a text);
+CREATE TABLE testeoc (a text);
 
 COPY testeoc FROM stdin CSV;
 a\.
@@ -189,11 +197,19 @@ c\.d
 "\."
 \.
 
+TRUNCATE testeoc;
+COPY testeoc FROM stdin WITH (FORMAT 'csv', PARALLEL 1);
+a\.
+\.b
+c\.d
+"\."
+\.
+
 COPY testeoc TO stdout CSV;
 
 -- test handling of nonstandard null marker that violates escaping rules
 
-CREATE TEMP TABLE testnull(a int, b text);
+CREATE TABLE testnull(a int, b text);
 INSERT INTO testnull VALUES (1, E'\\0'), (NULL, NULL);
 
 COPY testnull TO stdout WITH NULL AS E'\\0';
@@ -205,6 +221,14 @@ COPY testnull FROM stdin WITH NULL AS E'\\0';
 
 SELECT * FROM testnull;
 
+TRUNCATE testnull;
+COPY testnull FROM stdin WITH (NULL E'\\0', PARALLEL 1);
+42	\\0
+\0	\0
+\.
+
+SELECT * FROM testnull;
+
 BEGIN;
 CREATE TABLE vistest (LIKE testeoc);
 COPY vistest FROM stdin CSV;
@@ -249,6 +273,23 @@ SELECT * FROM vistest;
 
 BEGIN;
 TRUNCATE vistest;
+COPY vistest FROM stdin WITH (FORMAT 'csv', FREEZE, PARALLEL 1);
+a2
+b
+\.
+SELECT * FROM vistest;
+SAVEPOINT s1;
+TRUNCATE vistest;
+COPY vistest FROM stdin WITH (FORMAT 'csv', FREEZE, PARALLEL 1);
+d2
+e
+\.
+SELECT * FROM vistest;
+COMMIT;
+SELECT * FROM vistest;
+
+BEGIN;
+TRUNCATE vistest;
 COPY vistest FROM stdin CSV FREEZE;
 x
 y
@@ -298,7 +339,7 @@ SELECT * FROM vistest;
 COMMIT;
 SELECT * FROM vistest;
 -- Test FORCE_NOT_NULL and FORCE_NULL options
-CREATE TEMP TABLE forcetest (
+CREATE TABLE forcetest (
     a INT NOT NULL,
     b TEXT NOT NULL,
     c TEXT,
@@ -311,6 +352,10 @@ BEGIN;
 COPY forcetest (a, b, c) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(b), FORCE_NULL(c));
 1,,""
 \.
+TRUNCATE forcetest;
+COPY forcetest (a, b, c) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(b), FORCE_NULL(c), PARALLEL 1);
+1,,""
+\.
 COMMIT;
 SELECT b, c FROM forcetest WHERE a = 1;
 -- should succeed, FORCE_NULL and FORCE_NOT_NULL can be both specified
@@ -318,6 +363,10 @@ BEGIN;
 COPY forcetest (a, b, c, d) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(c,d), FORCE_NULL(c,d));
 2,'a',,""
 \.
+TRUNCATE forcetest;
+COPY forcetest (a, b, c, d) FROM STDIN WITH (FORMAT csv, FORCE_NOT_NULL(c,d), FORCE_NULL(c,d), PARALLEL 1);
+2,'a',,""
+\.
 COMMIT;
 SELECT c, d FROM forcetest WHERE a = 2;
 -- should fail with not-null constraint violation
@@ -353,6 +402,16 @@ copy check_con_tbl from stdin;
 0
 \.
 select * from check_con_tbl;
+\d+ check_con_tbl
+truncate check_con_tbl;
+copy check_con_tbl from stdin with (parallel 1);
+1
+\N
+\.
+copy check_con_tbl from stdin with (parallel 1);
+0
+\.
+select * from check_con_tbl;
 
 -- test with RLS enabled.
 CREATE ROLE regress_rls_copy_user;
@@ -454,10 +513,311 @@ test1
 SELECT * FROM instead_of_insert_tbl;
 COMMIT;
 
+-- Parallel copy tests.
+CREATE TABLE test_parallel_copy (
+        a INT,
+        b INT,
+        c TEXT not null default 'stuff',
+        d TEXT,
+        e TEXT
+) ;
+
+COPY test_parallel_copy (a, b, c, d, e) FROM stdin WITH (PARALLEL 1);
+1	11	test_c1	test_d1	test_e1
+2	12	test_c2	test_d2	test_e2
+\.
+
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL 1);
+3	test_d3
+\.
+
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL 1);
+4	test_d4
+5	test_d5
+\.
+
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL 1, FORMAT 'csv', HEADER);
+b	d
+\.
+
+-- zero workers: should perform non-parallel copy
+COPY test_parallel_copy (b, d) FROM stdin WITH (PARALLEL '0');
+
+-- referencing table: should perform non-parallel copy
+CREATE TABLE test_copy_pk(c1 INT PRIMARY KEY);
+INSERT INTO test_copy_pk VALUES(10);
+CREATE TABLE test_copy_ri(c1 INT REFERENCES test_copy_pk(c1));
+COPY test_copy_ri FROM stdin WITH (FORMAT csv, DELIMITER ',', PARALLEL 1);
+10
+\.
+
+-- expressions: should perform non-parallel copy
+CREATE TABLE test_copy_expr (index INT, height REAL, weight REAL);
+
+COPY test_copy_expr FROM STDIN WITH (FORMAT csv, DELIMITER ',', PARALLEL 1) WHERE height > random() * 65;
+60,60,60
+\.
+
+-- serial data: should perform non-parallel copy
+CREATE TABLE testserial (index SERIAL, height REAL);
+
+COPY testserial(height) FROM STDIN WITH (FORMAT csv, DELIMITER ',', PARALLEL 1);
+60
+\.
+
+-- temporary table copy: should perform non-parallel copy
+CREATE TEMPORARY TABLE temp_test(
+        a int
+) ;
+
+COPY temp_test (a) FROM stdin WITH (PARALLEL 1);
+10
+\.
+
+-- non-existent column in column list: should fail
+COPY test_parallel_copy (xyz) FROM stdin WITH (PARALLEL 1);
+
+-- too many columns in column list: should fail
+COPY test_parallel_copy (a, b, c, d, e, d, c) FROM stdin WITH (PARALLEL 1);
+
+-- missing data: should fail
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+
+\.
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+2000	230	23	23
+\.
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+2001	231	\N	\N
+\.
+
+-- extra data: should fail
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+2002	232	40	50	60	70	80
+\.
+
+-- various COPY options: delimiters, oids, NULL string, encoding
+COPY test_parallel_copy (b, c, d, e) FROM stdin WITH (DELIMITER ',', null 'x', PARALLEL 1) ;
+x,45,80,90
+x,\x,\\x,\\\x
+x,\,,\\\,,\\
+\.
+
+COPY test_parallel_copy FROM stdin WITH (DELIMITER ';', NULL '', PARALLEL 1);
+3000;;c;;
+\.
+
+COPY test_parallel_copy FROM stdin WITH (DELIMITER ':', NULL E'\\X', ENCODING 'sql_ascii', PARALLEL 1);
+4000:\X:C:\X:\X
+4001:1:empty::
+4002:2:null:\X:\X
+4003:3:Backslash:\\:\\
+4004:4:BackslashX:\\X:\\X
+4005:5:N:\N:\N
+4006:6:BackslashN:\\N:\\N
+4007:7:XX:\XX:\XX
+4008:8:Delimiter:\::\:
+\.
+
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a = 50004;
+50003	24	34	44	54
+50004	25	35	45 	55
+50005	26	36	46	56
+\.
+
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a > 60003;
+60001	22	32	42	52
+60002	23	33	43	53
+60003	24	34	44	54
+60004	25	35	45	55
+60005	26	36	46	56
+\.
+
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE f > 60003;
+
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a = max(x.b);
+
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a IN (SELECT 1 FROM x);
+
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1) WHERE a IN (generate_series(1,5));
+
+COPY test_parallel_copy FROM stdin WITH(PARALLEL 1) WHERE a = row_number() over(b);
+
+-- check results of copy in
+SELECT * FROM test_parallel_copy ORDER BY 1;
+
+-- parallel copy test for unlogged tables. should execute in parallel worker
+CREATE UNLOGGED TABLE test_parallel_copy_unlogged(
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+);
+
+COPY test_parallel_copy_unlogged FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+SELECT count(*) FROM test_parallel_copy_unlogged;
+
+-- parallel copy test for various trigger types
+TRUNCATE test_parallel_copy;
+
+-- parallel safe trigger function
+CREATE OR REPLACE FUNCTION parallel_copy_trig_func() RETURNS TRIGGER
+LANGUAGE plpgsql PARALLEL SAFE AS $$
+BEGIN
+  RETURN NEW;
+END;
+$$;
+
+-- before insert row trigger
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy
+FOR EACH ROW
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+
+-- after insert row trigger
+CREATE TRIGGER parallel_copy_trig
+AFTER INSERT ON test_parallel_copy
+FOR EACH ROW
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+
+-- before insert statement trigger
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+-- parallelism should be picked, since the trigger function is parallel safe
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+
+-- after insert statement trigger
+CREATE TRIGGER parallel_copy_trig
+AFTER INSERT ON test_parallel_copy
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+
+-- transition table is involved
+CREATE TRIGGER parallel_copy_trig
+AFTER INSERT ON test_parallel_copy REFERENCING NEW TABLE AS new_table
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+
+-- make trigger function parallel unsafe
+ALTER FUNCTION parallel_copy_trig_func PARALLEL UNSAFE;
+
+-- before statement trigger has a parallel unsafe function
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy
+FOR EACH STATEMENT
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+-- no parallelism should be picked
+COPY test_parallel_copy FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+SELECT count(*) FROM test_parallel_copy;
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy;
+
+-- instead of insert trigger, no parallelism should be picked
+COPY instead_of_insert_tbl_view FROM stdin WITH(PARALLEL 1);;
+test1
+\.
+
+SELECT count(*) FROM instead_of_insert_tbl_view;
+
+-- parallel copy test for a partitioned table with a before insert trigger on
+-- one of the partition
+ALTER FUNCTION parallel_copy_trig_func PARALLEL SAFE;
+
+CREATE TABLE test_parallel_copy_part (
+        a INT,
+        b INT,
+        c TEXT default 'stuff',
+        d TEXT,
+        e TEXT
+) PARTITION BY LIST (b);
+
+CREATE TABLE test_parallel_copy_part_a1 (c TEXT, b INT, a INT, e TEXT, d TEXT);
+
+CREATE TABLE test_parallel_copy_part_a2 (a INT, c TEXT, b INT, d TEXT, e TEXT);
+
+ALTER TABLE test_parallel_copy_part ATTACH PARTITION test_parallel_copy_part_a1 FOR VALUES IN(1);
+
+ALTER TABLE test_parallel_copy_part ATTACH PARTITION test_parallel_copy_part_a2 FOR VALUES IN(2);
+
+CREATE TRIGGER parallel_copy_trig
+BEFORE INSERT ON test_parallel_copy_part_a2
+FOR EACH ROW
+EXECUTE PROCEDURE parallel_copy_trig_func();
+
+COPY test_parallel_copy_part FROM stdin WITH (PARALLEL 1);
+1	1	test_c1	test_d1	test_e1
+2	2	test_c2	test_d2	test_e2
+\.
+
+SELECT count(*) FROM test_parallel_copy_part;
+
+DROP TRIGGER parallel_copy_trig ON test_parallel_copy_part_a2;
+
 -- clean up
+DROP TABLE test_copy_ri;
+DROP TABLE test_copy_pk;
+DROP TABLE test_copy_expr;
+DROP TABLE testeoc;
+DROP TABLE testnl;
 DROP TABLE forcetest;
+DROP TABLE test_parallel_copy;
+DROP TABLE test_parallel_copy_unlogged;
+DROP TABLE test_parallel_copy_part;
+DROP TABLE testserial;
+DROP TABLE testnull;
 DROP TABLE vistest;
 DROP FUNCTION truncate_in_subxact();
+DROP FUNCTION parallel_copy_trig_func();
 DROP TABLE x, y;
 DROP TABLE rls_t1 CASCADE;
 DROP ROLE regress_rls_copy_user;
-- 
1.8.3.1