From 3d72c381b024c8eb93324e250c1e628aa78965d0 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy 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