From 43d524094f1f9271521a219856b6e6df2205f17a Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Tue, 10 Nov 2020 18:39:35 +0530 Subject: [PATCH v10 3/6] Allow copy from command to process data from file/STDIN contents to a table in parallel. This feature allows the copy from to leverage multiple CPUs in order to copy data from file/STDIN to a table. This adds a PARALLEL option to COPY FROM command where the user can specify the number of workers that can be used to perform the COPY FROM command. The backend, to which the "COPY FROM" query is submitted acts as leader with the responsibility of reading data from the file/stdin, launching at most n number of workers as specified with PARALLEL 'n' option in the "COPY FROM" query. The leader populates the common data required for the workers execution in the DSM and shares it with the workers. The leader then executes before statement triggers if there exists any. Leader populates DSM lines which includes the start offset and line size, while populating the lines it reads as many blocks as required into the DSM data blocks from the file. Each block is of 64K size. The leader parses the data to identify a line, the existing logic from CopyReadLineText which identifies the lines with some changes was used for this. Leader checks if a free line is available to copy the information, if there is no free line it waits till the required line is freed up by the worker and then copies the identified lines information (offset & line size) into the DSM lines. This process is repeated till the complete file is processed. Simultaneously, the workers cache the lines(50) locally into the local memory and release the lines to the leader for further populating. Each worker processes the lines that it cached and inserts it into the table. The leader does not participate in the insertion of data, leaders only responsibility will be to identify the lines as fast as possible for the workers to do the actual copy operation. The leader waits till all the lines populated are processed by the workers and exits. We have chosen this design based on the reason "that everything stalls if the leader doesn't accept further input data, as well as when there are no available splitted chunks so it doesn't seem like a good idea to have the leader do other work. This is backed by the performance data where we have seen that with 1 worker there is just a 5-10% performance difference". --- src/backend/access/transam/parallel.c | 4 + src/backend/commands/copy.c | 271 +++++-- src/backend/commands/copyparallel.c | 1254 +++++++++++++++++++++++++++++++++ src/bin/psql/tab-complete.c | 2 +- src/include/commands/copy.h | 23 +- src/include/commands/copy_internal.h | 226 +++++- src/tools/pgindent/typedefs.list | 8 + 7 files changed, 1729 insertions(+), 59 deletions(-) diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index b042696..a3cff4b 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -25,6 +25,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/copy.h" #include "executor/execParallel.h" #include "libpq/libpq.h" #include "libpq/pqformat.h" @@ -145,6 +146,9 @@ static const struct }, { "parallel_vacuum_main", parallel_vacuum_main + }, + { + "ParallelCopyMain", ParallelCopyMain } }; diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 3b26fc5..6856ef5 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -44,6 +44,7 @@ #include "parser/parse_expr.h" #include "parser/parse_relation.h" #include "port/pg_bswap.h" +#include "postmaster/bgworker_internals.h" #include "rewrite/rewriteHandler.h" #include "storage/fd.h" #include "tcop/tcopprot.h" @@ -166,9 +167,13 @@ if (1) \ { \ if (raw_buf_ptr > cstate->raw_buf_index) \ { \ - appendBinaryStringInfo(&cstate->line_buf, \ - cstate->raw_buf + cstate->raw_buf_index, \ - raw_buf_ptr - cstate->raw_buf_index); \ + if (!IsParallelCopy()) \ + appendBinaryStringInfo(&cstate->line_buf, \ + cstate->raw_buf + cstate->raw_buf_index, \ + raw_buf_ptr - cstate->raw_buf_index); \ + else \ + line_size += raw_buf_ptr - cstate->raw_buf_index; \ + \ cstate->raw_buf_index = raw_buf_ptr; \ } \ } else ((void) 0) @@ -197,7 +202,6 @@ static void EndCopyTo(CopyState cstate); static uint64 DoCopyTo(CopyState cstate); static uint64 CopyTo(CopyState cstate); static void CopyOneRowTo(CopyState cstate, TupleTableSlot *slot); -static bool CopyReadLine(CopyState cstate); static bool CopyReadLineText(CopyState cstate); static int CopyReadAttributesText(CopyState cstate); static int CopyReadAttributesCSV(CopyState cstate); @@ -227,13 +231,8 @@ static void CopySendInt16(CopyState cstate, int16 val); static bool CopyGetInt16(CopyState cstate, int16 *val); static bool CopyLoadRawBuf(CopyState cstate); static int CopyReadBinaryData(CopyState cstate, char *dest, int nbytes); - -static void PopulateCommonCStateInfo(CopyState cstate, TupleDesc tup_desc, - List *attnamelist); static void ClearEOLFromCopiedData(CopyState cstate, char *copy_line_data, int copy_line_pos, int *copy_line_size); -static void ConvertToServerEncoding(CopyState cstate); - /* * Send copy start/stop messages for frontend copies. These have changed @@ -632,11 +631,11 @@ CopyGetInt16(CopyState cstate, int16 *val) static bool CopyLoadRawBuf(CopyState cstate) { - int nbytes = RAW_BUF_BYTES(cstate); + int nbytes = (!IsParallelCopy()) ? RAW_BUF_BYTES(cstate) : cstate->raw_buf_len; int inbytes; /* Copy down the unprocessed data if any. */ - if (nbytes > 0) + if (nbytes > 0 && !IsParallelCopy()) memmove(cstate->raw_buf, cstate->raw_buf + cstate->raw_buf_index, nbytes); @@ -644,7 +643,9 @@ CopyLoadRawBuf(CopyState cstate) 1, RAW_BUF_SIZE - nbytes); nbytes += inbytes; cstate->raw_buf[nbytes] = '\0'; - cstate->raw_buf_index = 0; + if (!IsParallelCopy()) + cstate->raw_buf_index = 0; + cstate->raw_buf_len = nbytes; return (inbytes > 0); } @@ -943,6 +944,8 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt, if (is_from) { + ParallelContext *pcxt = NULL; + Assert(rel); /* check read-only transaction and parallel mode */ @@ -952,7 +955,46 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt, cstate = BeginCopyFrom(pstate, rel, stmt->filename, stmt->is_program, NULL, stmt->attlist, stmt->options); cstate->whereClause = whereClause; - *processed = CopyFrom(cstate); /* copy from file to database */ + cstate->is_parallel = false; + + /* + * User chosen parallel copy. Determine if the parallel copy is + * actually allowed. If not, go with the non-parallel mode. + */ + if (cstate->nworkers > 0 && IsParallelCopyAllowed(cstate, relid)) + pcxt = BeginParallelCopy(cstate, stmt->attlist, relid); + + if (pcxt) + { + int i; + + ParallelCopyFrom(cstate); + + /* Wait for all copy workers to finish */ + WaitForParallelWorkersToFinish(pcxt); + + /* + * Next, accumulate WAL usage. (This must wait for the workers to + * finish, or we might get incomplete data.) + */ + for (i = 0; i < pcxt->nworkers_launched; i++) + InstrAccumParallelQuery(&cstate->pcdata->bufferusage[i], + &cstate->pcdata->walusage[i]); + + *processed = pg_atomic_read_u64(&cstate->pcdata->pcshared_info->processed); + EndParallelCopy(pcxt); + } + else + { + /* + * Reset nworkers to -1 here. This is useful in cases where user + * specifies parallel workers, but, no worker is picked up, so go + * back to non parallel mode value of nworkers. + */ + cstate->nworkers = -1; + *processed = CopyFrom(cstate); /* copy from file to database */ + } + EndCopyFrom(cstate); } else @@ -1003,6 +1045,7 @@ ProcessCopyOptions(ParseState *pstate, cstate->is_copy_from = is_from; cstate->file_encoding = -1; + cstate->nworkers = -1; /* Extract options from the statement node tree */ foreach(option, options) @@ -1173,6 +1216,31 @@ ProcessCopyOptions(ParseState *pstate, defel->defname), parser_errposition(pstate, defel->location))); } + else if (strcmp(defel->defname, "parallel") == 0) + { + int val; + + if (!cstate->is_copy_from) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("parallel option is supported only for copy from"), + parser_errposition(pstate, defel->location))); + if (cstate->nworkers >= 0) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"), + parser_errposition(pstate, defel->location))); + + val = defGetInt32(defel); + if (val < 1 || val > MAX_PARALLEL_WORKER_LIMIT) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("value %d out of bounds for option \"%s\"", + val, defel->defname), + errdetail("Valid values are between \"%d\" and \"%d\".", + 1, MAX_PARALLEL_WORKER_LIMIT))); + cstate->nworkers = val; + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -1528,9 +1596,9 @@ BeginCopy(ParseState *pstate, * PopulateCommonCStateInfo * * Populates the common variables required for copy from operation. This is a - * helper function for BeginCopy function. + * helper function for BeginCopy & InitializeParallelCopyInfo function. */ -static void +void PopulateCommonCStateInfo(CopyState cstate, TupleDesc tupDesc, List *attnamelist) { int num_phys_attrs; @@ -2547,7 +2615,7 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, * * Check if the relation specified in copy from is valid. */ -static void +void CheckTargetRelValidity(CopyState cstate) { Assert(cstate->rel); @@ -2644,7 +2712,7 @@ CopyFrom(CopyState cstate) PartitionTupleRouting *proute = NULL; ErrorContextCallback errcallback; - CommandId mycid = GetCurrentCommandId(true); + CommandId mycid; int ti_options = 0; /* start with default options for insert */ BulkInsertState bistate = NULL; CopyInsertMethod insertMethod; @@ -2654,7 +2722,18 @@ CopyFrom(CopyState cstate) bool has_instead_insert_row_trig; bool leafpart_use_multi_insert = false; - CheckTargetRelValidity(cstate); + /* + * Perform this check if it is not parallel copy. In case of parallel + * copy, this check is done by the leader, so that if any invalid case + * exist the copy from command will error out from the leader itself, + * avoiding launching workers, just to throw error. + */ + if (!IsParallelCopy()) + CheckTargetRelValidity(cstate); + else + SetCurrentCommandIdUsedForWorker(); + + mycid = GetCurrentCommandId(!IsParallelCopy()); /* * If the target file is new-in-transaction, we assume that checking FSM @@ -2690,7 +2769,8 @@ CopyFrom(CopyState cstate) ExecInitResultRelation(estate, resultRelInfo, 1); /* Verify the named relation is a valid target for INSERT */ - CheckValidResultRel(resultRelInfo, CMD_INSERT); + if (!IsParallelCopy()) + CheckValidResultRel(resultRelInfo, CMD_INSERT); ExecOpenIndices(resultRelInfo, false); @@ -2833,13 +2913,17 @@ CopyFrom(CopyState cstate) has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc && resultRelInfo->ri_TrigDesc->trig_insert_instead_row); - /* - * Check BEFORE STATEMENT insertion triggers. It's debatable whether we - * should do this for COPY, since it's not really an "INSERT" statement as - * such. However, executing these triggers maintains consistency with the - * EACH ROW triggers that we already fire on COPY. - */ - ExecBSInsertTriggers(estate, resultRelInfo); + if (!IsParallelCopy()) + { + /* + * Check BEFORE STATEMENT insertion triggers. It's debatable whether + * we should do this for COPY, since it's not really an "INSERT" + * statement as such. However, executing these triggers maintains + * consistency with the EACH ROW triggers that we already fire on + * COPY. + */ + ExecBSInsertTriggers(estate, resultRelInfo); + } econtext = GetPerTupleExprContext(estate); @@ -3141,7 +3225,10 @@ CopyFrom(CopyState cstate) * or FDW; this is the same definition used by nodeModifyTable.c * for counting tuples inserted by an INSERT command. */ - processed++; + if (!IsParallelCopy()) + processed++; + else + pg_atomic_add_fetch_u64(&cstate->pcdata->pcshared_info->processed, 1); } } @@ -3164,7 +3251,7 @@ CopyFrom(CopyState cstate) * In the old protocol, tell pqcomm that we can process normal protocol * messages again. */ - if (cstate->copy_dest == COPY_OLD_FE) + if (cstate->copy_dest == COPY_OLD_FE && !IsParallelCopy()) pq_endmsgread(); /* Execute AFTER STATEMENT insertion triggers */ @@ -3195,7 +3282,10 @@ CopyFrom(CopyState cstate) FreeExecutorState(estate); - return processed; + if (!IsParallelCopy()) + return processed; + else + return pg_atomic_read_u64(&cstate->pcdata->pcshared_info->processed); } /* @@ -3203,7 +3293,7 @@ CopyFrom(CopyState cstate) * * Populate the cstate catalog information. */ -static void +void PopulateCStateCatalogInfo(CopyState cstate) { TupleDesc tupDesc; @@ -3485,26 +3575,35 @@ NextCopyFromRawFields(CopyState cstate, char ***fields, int *nfields) /* only available for text or csv input */ Assert(!cstate->binary); - /* on input just throw the header line away */ - if (cstate->cur_lineno == 0 && cstate->header_line) + if (IsParallelCopy()) { - cstate->cur_lineno++; - if (CopyReadLine(cstate)) - return false; /* done */ + done = GetWorkerLine(cstate); + if (done && cstate->line_buf.len == 0) + return false; } + else + { + /* on input just throw the header line away */ + if (cstate->cur_lineno == 0 && cstate->header_line) + { + cstate->cur_lineno++; + if (CopyReadLine(cstate)) + return false; /* 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) - return false; + /* + * 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) + return false; + } /* Parse the line into de-escaped field values */ if (cstate->csv_mode) @@ -3729,7 +3828,7 @@ EndCopyFrom(CopyState cstate) * by newline. The terminating newline or EOF marker is not included * in the final value of line_buf. */ -static bool +bool CopyReadLine(CopyState cstate) { bool result; @@ -3752,9 +3851,31 @@ CopyReadLine(CopyState cstate) */ if (cstate->copy_dest == COPY_NEW_FE) { + bool bIsFirst = true; + do { - cstate->raw_buf_index = cstate->raw_buf_len; + if (!IsParallelCopy()) + cstate->raw_buf_index = cstate->raw_buf_len; + else + { + /* + * Get a new block if it is the first time, From the + * subsequent time, reset the index and re-use the same + * block. + */ + if (bIsFirst) + { + ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info; + uint32 block_pos = WaitGetFreeCopyBlock(pcshared_info); + + cstate->raw_buf = pcshared_info->data_blocks[block_pos].data; + bIsFirst = false; + } + + cstate->raw_buf_index = cstate->raw_buf_len = 0; + } + } while (CopyLoadRawBuf(cstate)); } } @@ -3809,11 +3930,11 @@ ClearEOLFromCopiedData(CopyState cstate, char *copy_line_data, * * Convert contents to server encoding. */ -static void +void ConvertToServerEncoding(CopyState cstate) { /* Done reading the line. Convert it to server encoding. */ - if (cstate->need_transcoding) + if (cstate->need_transcoding && (!IsParallelCopy() || IsWorker())) { char *cvt; @@ -3853,6 +3974,11 @@ CopyReadLineText(CopyState cstate) char quotec = '\0'; char escapec = '\0'; + /* For parallel copy */ + int line_size = 0; + uint32 line_pos = 0; + + cstate->eol_type = EOL_UNKNOWN; if (cstate->csv_mode) { quotec = cstate->quote[0]; @@ -3907,6 +4033,10 @@ CopyReadLineText(CopyState cstate) if (raw_buf_ptr >= copy_buf_len || need_data) { REFILL_LINEBUF; + if ((copy_buf_len == DATA_BLOCK_SIZE || copy_buf_len == 0) && + IsParallelCopy()) + SetRawBufForLoad(cstate, line_size, copy_buf_len, raw_buf_ptr, + ©_raw_buf); /* * Try to read some more data. This will certainly reset @@ -3914,14 +4044,14 @@ CopyReadLineText(CopyState cstate) */ if (!CopyLoadRawBuf(cstate)) hit_eof = true; - raw_buf_ptr = 0; + raw_buf_ptr = (IsParallelCopy()) ? cstate->raw_buf_index : 0; copy_buf_len = cstate->raw_buf_len; /* * If we are completely out of data, break out of the loop, * reporting EOF. */ - if (copy_buf_len <= 0) + if (RAW_BUF_BYTES(cstate) <= 0) { result = true; break; @@ -4131,9 +4261,15 @@ CopyReadLineText(CopyState cstate) * discard the data and the \. sequence. */ if (prev_raw_ptr > cstate->raw_buf_index) - appendBinaryStringInfo(&cstate->line_buf, - cstate->raw_buf + cstate->raw_buf_index, - prev_raw_ptr - cstate->raw_buf_index); + { + if (!IsParallelCopy()) + appendBinaryStringInfo(&cstate->line_buf, + cstate->raw_buf + cstate->raw_buf_index, + prev_raw_ptr - cstate->raw_buf_index); + else + line_size += prev_raw_ptr - cstate->raw_buf_index; + } + cstate->raw_buf_index = raw_buf_ptr; result = true; /* report EOF */ break; @@ -4185,6 +4321,22 @@ not_end_of_copy: IF_NEED_REFILL_AND_EOF_BREAK(mblen - 1); raw_buf_ptr += mblen - 1; } + + /* + * Skip the header line. Update the line here, this cannot be done at + * the beginning, as there is a possibility that file contains empty + * lines. + */ + if (IsParallelCopy() && first_char_in_line && !IsHeaderLine()) + { + ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info; + + line_pos = UpdateSharedLineInfo(cstate, + pcshared_info->cur_block_pos, + cstate->raw_buf_index, -1, + LINE_LEADER_POPULATING, -1); + } + first_char_in_line = false; } /* end of outer loop */ @@ -4193,9 +4345,16 @@ not_end_of_copy: */ REFILL_LINEBUF; if (!result && !IsHeaderLine()) - ClearEOLFromCopiedData(cstate, cstate->line_buf.data, - cstate->line_buf.len, &cstate->line_buf.len); + { + if (IsParallelCopy()) + ClearEOLFromCopiedData(cstate, cstate->raw_buf, raw_buf_ptr, + &line_size); + else + ClearEOLFromCopiedData(cstate, cstate->line_buf.data, + cstate->line_buf.len, &cstate->line_buf.len); + } + EndLineParallelCopy(cstate, line_pos, line_size, raw_buf_ptr); return result; } diff --git a/src/backend/commands/copyparallel.c b/src/backend/commands/copyparallel.c index 290d532..f29d63f 100644 --- a/src/backend/commands/copyparallel.c +++ b/src/backend/commands/copyparallel.c @@ -3,6 +3,45 @@ * copyparallel.c * Implements the Parallel COPY utility command * + * Parallel copy allows the copy from to leverage multiple CPUs in order to copy + * data from file/STDIN to a table. This adds a PARALLEL option to COPY FROM + * command where the user can specify the number of workers that can be used + * to perform the COPY FROM command. + * The backend, to which the "COPY FROM" query is submitted acts as leader with + * the responsibility of reading data from the file/stdin, launching at most n + * number of workers as specified with PARALLEL 'n' option in the "COPY FROM" + * query. The leader populates the common data using + * PARALLEL_COPY_KEY_SHARED_INFO, PARALLEL_COPY_KEY_CSTATE, + * PARALLEL_COPY_KEY_WAL_USAGE & PARALLEL_COPY_KEY_BUFFER_USAGE required for + * the workers execution in the DSM and shares it with the workers. The leader + * then executes before statement triggers if there exists any. Leader populates + * DSM lines (ParallelCopyDataBlock & ParallelCopyLineBoundaries data + * structures) which includes the start offset and line size, while populating + * the lines it reads as many blocks as required into the DSM data blocks from + * the file. Each block is of 64K size. The leader parses the data to identify a + * line, the existing logic from CopyReadLineText which identifies the lines + * with some changes was used for this. Leader checks if a free line is + * available (ParallelCopyLineBoundary->line_size will be -1) to copy the + * information, if there is no free line it waits till the required line is + * freed up by the worker and then copies the identified lines information + * (offset & line size) into the DSM lines. Leader will set the state of the + * line to LINE_LEADER_POPULATED to indicate the line is populated by the leader + * and the worker can start processing the line. This process is repeated till + * the complete file is processed. Simultaneously, each worker caches + * WORKER_CHUNK_COUNT lines locally into the local memory and release the lines + * to the leader for further populating. Each worker processes the lines as it + * was done in sequential copy. Refer comments above ParallelCopyLineBoundary + * for more details on leader/workers syncronization. + * The leader does not participate in the insertion of data, leaders only + * responsibility will be to identify the lines as fast as possible for the + * workers to do the actual copy operation. The leader waits till all the lines + * populated are processed by the workers and exits. We have chosen this design + * based on the reason "that everything stalls if the leader doesn't accept + * further input data, as well as when there are no available splitted chunks so + * it doesn't seem like a good idea to have the leader do other work. This is + * backed by the performance data where we have seen that with 1 worker there is + * just a 5-10% performance difference". + * * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * @@ -23,6 +62,326 @@ #include "pgstat.h" #include "utils/lsyscache.h" +/* DSM keys for parallel copy. */ +#define PARALLEL_COPY_KEY_SHARED_INFO 1 +#define PARALLEL_COPY_KEY_CSTATE 2 +#define PARALLEL_COPY_KEY_WAL_USAGE 3 +#define PARALLEL_COPY_KEY_BUFFER_USAGE 4 + +/* + * COPY_WAIT_TO_PROCESS - Wait before continuing to process. + */ +#define COPY_WAIT_TO_PROCESS() \ +{ \ + CHECK_FOR_INTERRUPTS(); \ + (void) WaitLatch(MyLatch, \ + WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, \ + 1L, WAIT_EVENT_PG_SLEEP); \ + ResetLatch(MyLatch); \ +} + +/* + * Estimate1ByteStrSize + * + * Estimate the size required for 1Byte strings in shared memory. + */ +static inline uint32 +Estimate1ByteStrSize(char *src) +{ + return (src) ? sizeof(uint8) + 1 : sizeof(uint8); +} + +/* + * EstimateNodeSize + * + * Estimate the size required for node type in shared memory. + */ +static uint32 +EstimateNodeSize(List *list, char **listStr) +{ + uint32 estsize = sizeof(uint32); + + if ((List *) list != NIL) + { + *listStr = nodeToString(list); + estsize += strlen(*listStr) + 1; + } + + return estsize; +} + +/* + * EstimateCStateSize + * + * Estimate the size of the required cstate variables in the shared memory. + */ +static uint32 +EstimateCStateSize(ParallelContext *pcxt, CopyState cstate, List *attnamelist, + SerializedListToStrCState *list_converted_str) +{ + Size estsize = 0; + uint32 estnodesize; + + estsize = add_size(estsize, sizeof(cstate->copy_dest)); + estsize = add_size(estsize, sizeof(cstate->file_encoding)); + estsize = add_size(estsize, sizeof(cstate->need_transcoding)); + estsize = add_size(estsize, sizeof(cstate->encoding_embeds_ascii)); + estsize = add_size(estsize, sizeof(cstate->csv_mode)); + estsize = add_size(estsize, sizeof(cstate->header_line)); + estsize = add_size(estsize, sizeof(cstate->null_print_len)); + estsize = add_size(estsize, sizeof(cstate->force_quote_all)); + estsize = add_size(estsize, sizeof(cstate->convert_selectively)); + estsize = add_size(estsize, sizeof(cstate->num_defaults)); + estsize = add_size(estsize, sizeof(cstate->pcdata->relid)); + estsize = add_size(estsize, sizeof(cstate->binary)); + + /* Size of null_print is available in null_print_len. */ + estsize = add_size(estsize, sizeof(uint32) + cstate->null_print_len); + + /* delim, quote & escape all uses 1 byte string to store the information. */ + estsize = add_size(estsize, Estimate1ByteStrSize(cstate->delim)); + estsize = add_size(estsize, Estimate1ByteStrSize(cstate->quote)); + estsize = add_size(estsize, Estimate1ByteStrSize(cstate->escape)); + + estnodesize = EstimateNodeSize(attnamelist, &list_converted_str->attnameListStr); + estsize = add_size(estsize, estnodesize); + estnodesize = EstimateNodeSize(cstate->force_notnull, &list_converted_str->notnullListStr); + estsize = add_size(estsize, estnodesize); + estnodesize = EstimateNodeSize(cstate->force_null, &list_converted_str->nullListStr); + estsize = add_size(estsize, estnodesize); + estnodesize = EstimateNodeSize(cstate->convert_select, &list_converted_str->convertListStr); + estsize = add_size(estsize, estnodesize); + estnodesize = EstimateNodeSize((List *) cstate->whereClause, &list_converted_str->whereClauseStr); + estsize = add_size(estsize, estnodesize); + estnodesize = EstimateNodeSize(cstate->range_table, &list_converted_str->rangeTableStr); + estsize = add_size(estsize, estnodesize); + + shm_toc_estimate_chunk(&pcxt->estimator, estsize); + shm_toc_estimate_keys(&pcxt->estimator, 1); + return estsize; +} + +/* + * SerializeStringToSharedMemory + * + * Copy the string to shared memory. + */ +static void +SerializeStringToSharedMemory(char *destptr, char *srcPtr, Size *copiedsize) +{ + uint32 len = srcPtr ? strlen(srcPtr) + 1 : 0; + + memcpy(destptr + *copiedsize, (uint32 *) &len, sizeof(uint32)); + *copiedsize += sizeof(uint32); + if (len) + { + memcpy(destptr + *copiedsize, srcPtr, len); + *copiedsize += len; + } +} + +/* + * Serialize1ByteStr + * + * Copy 1Byte strings to shared memory. + */ +static void +Serialize1ByteStr(char *dest, char *src, Size *copiedsize) +{ + uint8 len = (src) ? 1 : 0; + + memcpy(dest + (*copiedsize), (uint8 *) &len, sizeof(uint8)); + *copiedsize += sizeof(uint8); + if (src) + dest[(*copiedsize)++] = src[0]; +} + +/* + * SerializeParallelCopyState + * + * Serialize the cstate members required by the workers into shared memory. + */ +static void +SerializeParallelCopyState(ParallelContext *pcxt, CopyState cstate, + uint32 estimatedSize, + SerializedListToStrCState *list_converted_str) +{ + char *shmptr = (char *) shm_toc_allocate(pcxt->toc, estimatedSize + 1); + Size copiedsize = 0; + + memcpy(shmptr + copiedsize, (char *) &cstate->copy_dest, sizeof(cstate->copy_dest)); + copiedsize += sizeof(cstate->copy_dest); + memcpy(shmptr + copiedsize, (char *) &cstate->file_encoding, sizeof(cstate->file_encoding)); + copiedsize += sizeof(cstate->file_encoding); + memcpy(shmptr + copiedsize, (char *) &cstate->need_transcoding, sizeof(cstate->need_transcoding)); + copiedsize += sizeof(cstate->need_transcoding); + memcpy(shmptr + copiedsize, (char *) &cstate->encoding_embeds_ascii, sizeof(cstate->encoding_embeds_ascii)); + copiedsize += sizeof(cstate->encoding_embeds_ascii); + memcpy(shmptr + copiedsize, (char *) &cstate->csv_mode, sizeof(cstate->csv_mode)); + copiedsize += sizeof(cstate->csv_mode); + memcpy(shmptr + copiedsize, (char *) &cstate->header_line, sizeof(cstate->header_line)); + copiedsize += sizeof(cstate->header_line); + memcpy(shmptr + copiedsize, (char *) &cstate->null_print_len, sizeof(cstate->null_print_len)); + copiedsize += sizeof(cstate->null_print_len); + memcpy(shmptr + copiedsize, (char *) &cstate->force_quote_all, sizeof(cstate->force_quote_all)); + copiedsize += sizeof(cstate->force_quote_all); + memcpy(shmptr + copiedsize, (char *) &cstate->convert_selectively, sizeof(cstate->convert_selectively)); + copiedsize += sizeof(cstate->convert_selectively); + memcpy(shmptr + copiedsize, (char *) &cstate->num_defaults, sizeof(cstate->num_defaults)); + copiedsize += sizeof(cstate->num_defaults); + memcpy(shmptr + copiedsize, (char *) &cstate->pcdata->relid, sizeof(cstate->pcdata->relid)); + copiedsize += sizeof(cstate->pcdata->relid); + memcpy(shmptr + copiedsize, (char *) &cstate->binary, sizeof(cstate->binary)); + copiedsize += sizeof(cstate->binary); + + memcpy(shmptr + copiedsize, (uint32 *) &cstate->null_print_len, sizeof(uint32)); + copiedsize += sizeof(uint32); + if (cstate->null_print_len) + { + memcpy(shmptr + copiedsize, cstate->null_print, cstate->null_print_len); + copiedsize += cstate->null_print_len; + } + + Serialize1ByteStr(shmptr, cstate->delim, &copiedsize); + Serialize1ByteStr(shmptr, cstate->quote, &copiedsize); + Serialize1ByteStr(shmptr, cstate->escape, &copiedsize); + + SerializeStringToSharedMemory(shmptr, list_converted_str->attnameListStr, &copiedsize); + SerializeStringToSharedMemory(shmptr, list_converted_str->notnullListStr, &copiedsize); + SerializeStringToSharedMemory(shmptr, list_converted_str->nullListStr, &copiedsize); + SerializeStringToSharedMemory(shmptr, list_converted_str->convertListStr, &copiedsize); + SerializeStringToSharedMemory(shmptr, list_converted_str->whereClauseStr, &copiedsize); + SerializeStringToSharedMemory(shmptr, list_converted_str->rangeTableStr, &copiedsize); + + shm_toc_insert(pcxt->toc, PARALLEL_COPY_KEY_CSTATE, shmptr); +} + +/* + * RestoreNodeFromSharedMemory + * + * Copy the node contents which was stored as string format in shared memory & + * convert it into node type. + */ +static void * +RestoreNodeFromSharedMemory(char *srcPtr, Size *copiedsize, bool isSrcStr) +{ + char *destptr = NULL; + List *destList = NIL; + uint32 len; + + memcpy((uint32 *) (&len), srcPtr + *copiedsize, sizeof(uint32)); + *copiedsize += sizeof(uint32); + if (len) + { + destptr = (char *) palloc(len); + memcpy(destptr, srcPtr + *copiedsize, len); + *copiedsize += len; + if (!isSrcStr) + { + destList = (List *) stringToNode(destptr); + pfree(destptr); + return destList; + } + + return destptr; + } + + return NULL; +} + +/* + * Restore1ByteStr + * + * Restore 1Byte strings from shared memory to worker local memory. + */ +static void +Restore1ByteStr(char **dest, char *src, Size *copiedsize) +{ + uint8 len; + + memcpy((uint8 *) (&len), src + (*copiedsize), sizeof(uint8)); + (*copiedsize) += sizeof(uint8); + if (len) + { + *dest = palloc0(sizeof(char) + 1); + (*dest)[0] = src[(*copiedsize)++]; + } +} + +/* + * RestoreParallelCopyState + * + * Retrieve the cstate members which was populated by the leader in the shared + * memory. + */ +static void +RestoreParallelCopyState(shm_toc *toc, CopyState cstate, List **attlist) +{ + char *shared_str_val = (char *) shm_toc_lookup(toc, PARALLEL_COPY_KEY_CSTATE, true); + Size copiedsize = 0; + + memcpy((char *) &cstate->copy_dest, shared_str_val + copiedsize, sizeof(cstate->copy_dest)); + copiedsize += sizeof(cstate->copy_dest); + memcpy((char *) &cstate->file_encoding, shared_str_val + copiedsize, sizeof(cstate->file_encoding)); + copiedsize += sizeof(cstate->file_encoding); + memcpy((char *) &cstate->need_transcoding, shared_str_val + copiedsize, sizeof(cstate->need_transcoding)); + copiedsize += sizeof(cstate->need_transcoding); + memcpy((char *) &cstate->encoding_embeds_ascii, shared_str_val + copiedsize, sizeof(cstate->encoding_embeds_ascii)); + copiedsize += sizeof(cstate->encoding_embeds_ascii); + memcpy((char *) &cstate->csv_mode, shared_str_val + copiedsize, sizeof(cstate->csv_mode)); + copiedsize += sizeof(cstate->csv_mode); + memcpy((char *) &cstate->header_line, shared_str_val + copiedsize, sizeof(cstate->header_line)); + copiedsize += sizeof(cstate->header_line); + memcpy((char *) &cstate->null_print_len, shared_str_val + copiedsize, sizeof(cstate->null_print_len)); + copiedsize += sizeof(cstate->null_print_len); + memcpy((char *) &cstate->force_quote_all, shared_str_val + copiedsize, sizeof(cstate->force_quote_all)); + copiedsize += sizeof(cstate->force_quote_all); + memcpy((char *) &cstate->convert_selectively, shared_str_val + copiedsize, sizeof(cstate->convert_selectively)); + copiedsize += sizeof(cstate->convert_selectively); + memcpy((char *) &cstate->num_defaults, shared_str_val + copiedsize, sizeof(cstate->num_defaults)); + copiedsize += sizeof(cstate->num_defaults); + memcpy((char *) &cstate->pcdata->relid, shared_str_val + copiedsize, sizeof(cstate->pcdata->relid)); + copiedsize += sizeof(cstate->pcdata->relid); + memcpy((char *) &cstate->binary, shared_str_val + copiedsize, sizeof(cstate->binary)); + copiedsize += sizeof(cstate->binary); + + cstate->null_print = (char *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, true); + if (!cstate->null_print) + cstate->null_print = ""; + + Restore1ByteStr(&cstate->delim, shared_str_val, &copiedsize); + Restore1ByteStr(&cstate->quote, shared_str_val, &copiedsize); + Restore1ByteStr(&cstate->escape, shared_str_val, &copiedsize); + + *attlist = (List *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, false); + cstate->force_notnull = (List *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, false); + cstate->force_null = (List *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, false); + cstate->convert_select = (List *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, false); + cstate->whereClause = (Node *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, false); + cstate->range_table = (List *) RestoreNodeFromSharedMemory(shared_str_val, &copiedsize, false); +} + +/* + * PopulateParallelCopyShmInfo + * + * Sets ParallelCopyShmInfo structure members. + */ +static void +PopulateParallelCopyShmInfo(ParallelCopyShmInfo *shared_info_ptr) +{ + uint32 count; + + MemSet(shared_info_ptr, 0, sizeof(ParallelCopyShmInfo)); + shared_info_ptr->is_read_in_progress = true; + shared_info_ptr->cur_block_pos = -1; + for (count = 0; count < RINGSIZE; count++) + { + ParallelCopyLineBoundary *lineInfo = &shared_info_ptr->line_boundaries.ring[count]; + + pg_atomic_init_u32(&(lineInfo->line_size), -1); + } +} + /* * CheckExprParallelSafety * @@ -101,3 +460,898 @@ IsParallelCopyAllowed(CopyState cstate, Oid relid) return true; } + +/* + * BeginParallelCopy - Start parallel copy tasks. + * + * Get the number of workers required to perform the parallel copy. The data + * structures that are required by the parallel workers will be initialized, the + * size required in DSM will be calculated and the necessary keys will be loaded + * in the DSM. The specified number of workers will then be launched. + */ +ParallelContext * +BeginParallelCopy(CopyState cstate, List *attnamelist, Oid relid) +{ + ParallelContext *pcxt; + ParallelCopyShmInfo *shared_info_ptr; + int parallel_workers = 0; + WalUsage *walusage; + BufferUsage *bufferusage; + ParallelCopyData *pcdata; + MemoryContext oldcontext; + uint32 estsize; + SerializedListToStrCState list_converted_str = {0}; + + CheckTargetRelValidity(cstate); + parallel_workers = Min(cstate->nworkers, max_worker_processes); + + /* Can't perform copy in parallel */ + if (parallel_workers == 0) + return NULL; + + oldcontext = MemoryContextSwitchTo(cstate->copycontext); + pcdata = (ParallelCopyData *) palloc0(sizeof(ParallelCopyData)); + MemoryContextSwitchTo(oldcontext); + + cstate->pcdata = pcdata; + pcdata->relid = relid; + (void) GetCurrentFullTransactionId(); + (void) GetCurrentCommandId(true); + + EnterParallelMode(); + pcxt = CreateParallelContext("postgres", "ParallelCopyMain", + parallel_workers); + Assert(pcxt->nworkers > 0); + + /* + * Estimate size for shared information for PARALLEL_COPY_KEY_SHARED_INFO + */ + shm_toc_estimate_chunk(&pcxt->estimator, sizeof(ParallelCopyShmInfo)); + shm_toc_estimate_keys(&pcxt->estimator, 1); + + /* Estimate the size for shared information for PARALLEL_COPY_KEY_CSTATE */ + estsize = EstimateCStateSize(pcxt, cstate, attnamelist, &list_converted_str); + + /* + * Estimate space for WalUsage and BufferUsage -- + * PARALLEL_COPY_KEY_WAL_USAGE and PARALLEL_COPY_KEY_BUFFER_USAGE. + */ + shm_toc_estimate_chunk(&pcxt->estimator, + mul_size(sizeof(WalUsage), pcxt->nworkers)); + shm_toc_estimate_keys(&pcxt->estimator, 1); + shm_toc_estimate_chunk(&pcxt->estimator, + mul_size(sizeof(BufferUsage), pcxt->nworkers)); + shm_toc_estimate_keys(&pcxt->estimator, 1); + + InitializeParallelDSM(pcxt); + + /* If no DSM segment was available, back out (do serial copy) */ + if (pcxt->seg == NULL) + { + EndParallelCopy(pcxt); + pfree(pcdata); + cstate->pcdata = NULL; + return NULL; + } + + /* Allocate shared memory for PARALLEL_COPY_KEY_SHARED_INFO */ + shared_info_ptr = (ParallelCopyShmInfo *) shm_toc_allocate(pcxt->toc, sizeof(ParallelCopyShmInfo)); + PopulateParallelCopyShmInfo(shared_info_ptr); + + shm_toc_insert(pcxt->toc, PARALLEL_COPY_KEY_SHARED_INFO, shared_info_ptr); + pcdata->pcshared_info = shared_info_ptr; + + SerializeParallelCopyState(pcxt, cstate, estsize, &list_converted_str); + + /* + * Allocate space for each worker's WalUsage and BufferUsage; no need to + * initialize. + */ + walusage = shm_toc_allocate(pcxt->toc, + mul_size(sizeof(WalUsage), pcxt->nworkers)); + shm_toc_insert(pcxt->toc, PARALLEL_COPY_KEY_WAL_USAGE, walusage); + pcdata->walusage = walusage; + bufferusage = shm_toc_allocate(pcxt->toc, + mul_size(sizeof(BufferUsage), pcxt->nworkers)); + shm_toc_insert(pcxt->toc, PARALLEL_COPY_KEY_BUFFER_USAGE, bufferusage); + pcdata->bufferusage = bufferusage; + + LaunchParallelWorkers(pcxt); + if (pcxt->nworkers_launched == 0) + { + EndParallelCopy(pcxt); + pfree(pcdata); + cstate->pcdata = NULL; + return NULL; + } + + /* + * Caller needs to wait for all launched workers when we return. Make + * sure that the failure-to-start case will not hang forever. + */ + WaitForParallelWorkersToAttach(pcxt); + + pcdata->is_leader = true; + cstate->is_parallel = true; + return pcxt; +} + +/* + * EndParallelCopy + * + * End the parallel copy tasks. + */ +pg_attribute_always_inline void +EndParallelCopy(ParallelContext *pcxt) +{ + Assert(!IsParallelWorker()); + + DestroyParallelContext(pcxt); + ExitParallelMode(); +} + +/* + * InitializeParallelCopyInfo + * + * Initialize parallel worker. + */ +static void +InitializeParallelCopyInfo(CopyState cstate, List *attnamelist) +{ + uint32 count; + ParallelCopyData *pcdata = cstate->pcdata; + TupleDesc tup_desc = RelationGetDescr(cstate->rel); + + PopulateCommonCStateInfo(cstate, tup_desc, attnamelist); + + /* Initialize state variables. */ + cstate->reached_eof = false; + cstate->eol_type = EOL_UNKNOWN; + cstate->cur_relname = RelationGetRelationName(cstate->rel); + cstate->cur_lineno = 0; + cstate->cur_attname = NULL; + cstate->cur_attval = NULL; + + /* Set up variables to avoid per-attribute overhead. */ + initStringInfo(&cstate->attribute_buf); + + initStringInfo(&cstate->line_buf); + for (count = 0; count < WORKER_CHUNK_COUNT; count++) + initStringInfo(&pcdata->worker_line_buf[count].line_buf); + + cstate->line_buf_converted = false; + cstate->raw_buf = NULL; + cstate->raw_buf_index = cstate->raw_buf_len = 0; + + PopulateCStateCatalogInfo(cstate); + + /* Create workspace for CopyReadAttributes results. */ + if (!cstate->binary) + { + AttrNumber attr_count = list_length(cstate->attnumlist); + + cstate->max_fields = attr_count; + cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *)); + } +} + +/* + * UpdateLineInfo + * + * Update line information & return. + */ +static bool +UpdateLineInfo(ParallelCopyShmInfo *pcshared_info, + ParallelCopyLineBoundary *lineInfo, uint32 write_pos) +{ + elog(DEBUG1, "[Worker] Completed processing line:%u", write_pos); + pg_atomic_write_u32(&lineInfo->line_state, LINE_WORKER_PROCESSED); + pg_atomic_write_u32(&lineInfo->line_size, -1); + pg_atomic_add_fetch_u64(&pcshared_info->total_worker_processed, 1); + return false; +} + +/* + * CacheLineInfo + * + * Cache the line information to local memory. + */ +static bool +CacheLineInfo(CopyState cstate, uint32 buff_count) +{ + ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info; + ParallelCopyData *pcdata = cstate->pcdata; + uint32 write_pos; + ParallelCopyDataBlock *data_blk_ptr; + ParallelCopyLineBoundary *lineInfo; + uint32 offset; + uint32 dataSize; + uint32 copiedSize = 0; + + resetStringInfo(&pcdata->worker_line_buf[buff_count].line_buf); + write_pos = GetLinePosition(cstate); + if (-1 == write_pos) + return true; + + /* Get the current line information. */ + lineInfo = &pcshared_info->line_boundaries.ring[write_pos]; + if (pg_atomic_read_u32(&lineInfo->line_size) == 0) + return UpdateLineInfo(pcshared_info, lineInfo, write_pos); + + /* Get the block information. */ + data_blk_ptr = &pcshared_info->data_blocks[lineInfo->first_block]; + + /* Get the offset information from where the data must be copied. */ + offset = lineInfo->start_offset; + pcdata->worker_line_buf[buff_count].cur_lineno = lineInfo->cur_lineno; + + elog(DEBUG1, "[Worker] Processing - line position:%u, block:%u, unprocessed lines:%u, offset:%u, line size:%u", + write_pos, lineInfo->first_block, + pg_atomic_read_u32(&data_blk_ptr->unprocessed_line_parts), + offset, pg_atomic_read_u32(&lineInfo->line_size)); + + for (;;) + { + uint8 skip_bytes = data_blk_ptr->skip_bytes; + + /* + * There is a possibility that the loop embedded at the bottom of the + * current loop has come out because data_blk_ptr->curr_blk_completed + * is set, but dataSize read might be an old value, if + * data_blk_ptr->curr_blk_completed and the line is completed, + * line_size will be set. Read the line_size again to be sure if it is + * completed or partial block. + */ + dataSize = pg_atomic_read_u32(&lineInfo->line_size); + if (dataSize != -1) + { + uint32 remainingSize = dataSize - copiedSize; + + if (!remainingSize) + break; + + /* Whole line is in current block. */ + if (remainingSize + offset + skip_bytes < DATA_BLOCK_SIZE) + { + appendBinaryStringInfo(&pcdata->worker_line_buf[buff_count].line_buf, + &data_blk_ptr->data[offset], + remainingSize); + pg_atomic_sub_fetch_u32(&data_blk_ptr->unprocessed_line_parts, + 1); + break; + } + else + { + /* Line is spread across the blocks. */ + uint32 lineInCurrentBlock = (DATA_BLOCK_SIZE - skip_bytes) - offset; + + appendBinaryStringInfoNT(&pcdata->worker_line_buf[buff_count].line_buf, + &data_blk_ptr->data[offset], + lineInCurrentBlock); + pg_atomic_sub_fetch_u32(&data_blk_ptr->unprocessed_line_parts, 1); + copiedSize += lineInCurrentBlock; + while (copiedSize < dataSize) + { + uint32 currentBlockCopySize; + ParallelCopyDataBlock *currBlkPtr = &pcshared_info->data_blocks[data_blk_ptr->following_block]; + + skip_bytes = currBlkPtr->skip_bytes; + + /* + * If complete data is present in current block use + * dataSize - copiedSize, or copy the whole block from + * current block. + */ + currentBlockCopySize = Min(dataSize - copiedSize, DATA_BLOCK_SIZE - skip_bytes); + appendBinaryStringInfoNT(&pcdata->worker_line_buf[buff_count].line_buf, + &currBlkPtr->data[0], + currentBlockCopySize); + pg_atomic_sub_fetch_u32(&currBlkPtr->unprocessed_line_parts, 1); + copiedSize += currentBlockCopySize; + data_blk_ptr = currBlkPtr; + } + + break; + } + } + else if (data_blk_ptr->curr_blk_completed) + { + /* Copy this complete block from the current offset. */ + uint32 lineInCurrentBlock = (DATA_BLOCK_SIZE - skip_bytes) - offset; + + appendBinaryStringInfoNT(&pcdata->worker_line_buf[buff_count].line_buf, + &data_blk_ptr->data[offset], + lineInCurrentBlock); + pg_atomic_sub_fetch_u32(&data_blk_ptr->unprocessed_line_parts, 1); + copiedSize += lineInCurrentBlock; + + /* + * Reset the offset. For the first copy, copy from the offset. For + * the subsequent copy the complete block. + */ + offset = 0; + + /* Set data_blk_ptr to the following block. */ + data_blk_ptr = &pcshared_info->data_blocks[data_blk_ptr->following_block]; + } + + for (;;) + { + /* Get the size of this line */ + dataSize = pg_atomic_read_u32(&lineInfo->line_size); + + /* + * If the data is present in current block lineInfo->line_size + * will be updated. If the data is spread across the blocks either + * of lineInfo->line_size or data_blk_ptr->curr_blk_completed can + * be updated. lineInfo->line_size will be updated if the complete + * read is finished. data_blk_ptr->curr_blk_completed will be + * updated if processing of current block is finished and data + * processing is not finished. + */ + if (data_blk_ptr->curr_blk_completed || (dataSize != -1)) + break; + + COPY_WAIT_TO_PROCESS() + } + } + + return UpdateLineInfo(pcshared_info, lineInfo, write_pos); +} + +/* + * GetCachedLine + * + * Return a previously cached line to be processed by the worker. + */ +static bool +GetCachedLine(CopyState cstate, ParallelCopyData *pcdata) +{ + cstate->line_buf = pcdata->worker_line_buf[pcdata->worker_line_buf_pos].line_buf; + cstate->cur_lineno = pcdata->worker_line_buf[pcdata->worker_line_buf_pos].cur_lineno; + cstate->line_buf_valid = true; + + /* Mark that encoding conversion hasn't occurred yet. */ + cstate->line_buf_converted = false; + + /* For binary format data, we don't need conversion. */ + if (!cstate->binary) + ConvertToServerEncoding(cstate); + + pcdata->worker_line_buf_pos++; + return false; +} + +/* + * GetWorkerLine + * + * Returns a line for worker to process. + */ +bool +GetWorkerLine(CopyState cstate) +{ + uint32 buff_count; + ParallelCopyData *pcdata = cstate->pcdata; + + /* + * Copy the line data to line_buf and release the line position so that + * the worker can continue loading data. + */ + if (pcdata->worker_line_buf_pos < pcdata->worker_line_buf_count) + return GetCachedLine(cstate, pcdata); + + pcdata->worker_line_buf_pos = 0; + pcdata->worker_line_buf_count = 0; + + for (buff_count = 0; buff_count < WORKER_CHUNK_COUNT; buff_count++) + { + bool result = CacheLineInfo(cstate, buff_count); + + if (result) + break; + + pcdata->worker_line_buf_count++; + } + + if (pcdata->worker_line_buf_count > 0) + return GetCachedLine(cstate, pcdata); + else + resetStringInfo(&cstate->line_buf); + + return true; +} + +/* + * ParallelCopyMain - Parallel copy worker's code. + * + * Where clause handling, convert tuple to columns, add default null values for + * the missing columns that are not present in that record. Find the partition + * if it is partitioned table, invoke before row insert Triggers, handle + * constraints and insert the tuples. + */ +void +ParallelCopyMain(dsm_segment *seg, shm_toc *toc) +{ + CopyState cstate; + ParallelCopyData *pcdata; + ParallelCopyShmInfo *pcshared_info; + Relation rel = NULL; + MemoryContext oldcontext; + List *attlist = NIL; + WalUsage *walusage; + BufferUsage *bufferusage; + + /* Allocate workspace and zero all fields. */ + cstate = (CopyStateData *) palloc0(sizeof(CopyStateData)); + + /* + * We allocate everything used by a cstate in a new memory context. This + * avoids memory leaks during repeated use of COPY in a query. + */ + cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext, + "COPY", + ALLOCSET_DEFAULT_SIZES); + oldcontext = MemoryContextSwitchTo(cstate->copycontext); + + pcdata = (ParallelCopyData *) palloc0(sizeof(ParallelCopyData)); + cstate->pcdata = pcdata; + pcdata->is_leader = false; + pcdata->worker_processed_pos = -1; + cstate->is_parallel = true; + pcshared_info = (ParallelCopyShmInfo *) shm_toc_lookup(toc, PARALLEL_COPY_KEY_SHARED_INFO, false); + + ereport(DEBUG1, (errmsg("Starting parallel copy worker"))); + + pcdata->pcshared_info = pcshared_info; + RestoreParallelCopyState(toc, cstate, &attlist); + + /* Open and lock the relation, using the appropriate lock type. */ + rel = table_open(cstate->pcdata->relid, RowExclusiveLock); + cstate->rel = rel; + InitializeParallelCopyInfo(cstate, attlist); + + /* Prepare to track buffer usage during parallel execution */ + InstrStartParallelQuery(); + + CopyFrom(cstate); + + if (rel != NULL) + table_close(rel, RowExclusiveLock); + + /* Report WAL/buffer usage during parallel execution */ + bufferusage = shm_toc_lookup(toc, PARALLEL_COPY_KEY_BUFFER_USAGE, false); + walusage = shm_toc_lookup(toc, PARALLEL_COPY_KEY_WAL_USAGE, false); + InstrEndParallelQuery(&bufferusage[ParallelWorkerNumber], + &walusage[ParallelWorkerNumber]); + + MemoryContextSwitchTo(oldcontext); + MemoryContextDelete(cstate->copycontext); + pfree(cstate); + return; +} + +/* + * UpdateSharedLineInfo + * + * Update the line information. + */ +uint32 +UpdateSharedLineInfo(CopyState cstate, uint32 blk_pos, uint32 offset, + uint32 line_size, uint32 line_state, uint32 blk_line_pos) +{ + ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info; + ParallelCopyLineBoundaries *lineBoundaryPtr = &pcshared_info->line_boundaries; + ParallelCopyLineBoundary *lineInfo; + uint32 line_pos; + + /* blk_line_pos will be valid in case line_pos was blocked earlier. */ + if (blk_line_pos == -1) + { + line_pos = lineBoundaryPtr->pos; + + /* Update the line information for the worker to pick and process. */ + lineInfo = &lineBoundaryPtr->ring[line_pos]; + while (pg_atomic_read_u32(&lineInfo->line_size) != -1) + COPY_WAIT_TO_PROCESS() + + lineInfo->first_block = blk_pos; + lineInfo->start_offset = offset; + lineInfo->cur_lineno = cstate->cur_lineno; + lineBoundaryPtr->pos = (lineBoundaryPtr->pos + 1) % RINGSIZE; + } + else + { + line_pos = blk_line_pos; + lineInfo = &lineBoundaryPtr->ring[line_pos]; + } + + if (line_state == LINE_LEADER_POPULATED) + { + elog(DEBUG1, "[Leader] Added line with block:%u, offset:%u, line position:%u, line size:%u", + lineInfo->first_block, lineInfo->start_offset, line_pos, + line_size); + pcshared_info->populated++; + if (line_size == 0 || cstate->binary) + pg_atomic_write_u32(&lineInfo->line_state, line_state); + else + { + uint32 current_line_state = LINE_LEADER_POPULATING; + + /* + * Make sure that no worker has consumed this element, if this + * line is spread across multiple data blocks, worker would have + * started processing, no need to change the state to + * LINE_LEADER_POPULATING in this case. + */ + (void) pg_atomic_compare_exchange_u32(&lineInfo->line_state, + ¤t_line_state, + LINE_LEADER_POPULATED); + } + } + else + { + elog(DEBUG1, "[Leader] Adding - block:%u, offset:%u, line position:%u", + lineInfo->first_block, lineInfo->start_offset, line_pos); + pg_atomic_write_u32(&lineInfo->line_state, line_state); + } + + pg_atomic_write_u32(&lineInfo->line_size, line_size); + + return line_pos; +} + +/* + * ParallelCopyFrom - parallel copy leader's functionality. + * + * Leader executes the before statement for before statement trigger, if before + * statement trigger is present. It will read the table data from the file and + * copy the contents to DSM data blocks. It will then read the input contents + * from the DSM data block and identify the records based on line breaks. This + * information is called line or a record that need to be inserted into a + * relation. The line information will be stored in ParallelCopyLineBoundary DSM + * data structure. Workers will then process this information and insert the + * data in to table. It will repeat this process until the all data is read from + * the file and all the DSM data blocks are processed. While processing if + * leader identifies that DSM Data blocks or DSM ParallelCopyLineBoundary data + * structures is full, leader will wait till the worker frees up some entries + * and repeat the process. It will wait till all the lines populated are + * processed by the workers and exits. + */ +void +ParallelCopyFrom(CopyState cstate) +{ + ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info; + ErrorContextCallback errcallback; + + ereport(DEBUG1, (errmsg("Running parallel copy leader"))); + + /* raw_buf is not used in parallel copy, instead data blocks are used. */ + pfree(cstate->raw_buf); + cstate->raw_buf = NULL; + + /* Execute the before statement triggers from the leader */ + ExecBeforeStmtTrigger(cstate); + + /* Set up callback to identify error line number */ + errcallback.callback = CopyFromErrorCallback; + errcallback.arg = (void *) cstate; + errcallback.previous = error_context_stack; + error_context_stack = &errcallback; + + /* On input just throw the header line away. */ + if (cstate->cur_lineno == 0 && cstate->header_line) + { + cstate->cur_lineno++; + if (CopyReadLine(cstate)) + { + pcshared_info->is_read_in_progress = false; + return; /* done */ + } + } + + for (;;) + { + bool done; + + CHECK_FOR_INTERRUPTS(); + + cstate->cur_lineno++; + + /* 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; + } + + + /* Done, clean up */ + error_context_stack = errcallback.previous; + + /* + * In the old protocol, tell pqcomm that we can process normal protocol + * messages again. + */ + if (cstate->copy_dest == COPY_OLD_FE) + pq_endmsgread(); + + pcshared_info->is_read_in_progress = false; + cstate->cur_lineno = 0; +} + +/* + * GetLinePosition + * + * Return the line position once the leader has populated the data. + */ +uint32 +GetLinePosition(CopyState cstate) +{ + ParallelCopyData *pcdata = cstate->pcdata; + ParallelCopyShmInfo *pcshared_info = pcdata->pcshared_info; + uint32 previous_pos = pcdata->worker_processed_pos; + uint32 write_pos = (previous_pos == -1) ? 0 : (previous_pos + 1) % RINGSIZE; + + for (;;) + { + uint32 dataSize; + bool is_read_in_progress = pcshared_info->is_read_in_progress; + ParallelCopyLineBoundary *lineInfo; + ParallelCopyDataBlock *data_blk_ptr; + uint32 line_state = LINE_LEADER_POPULATED; + ParallelCopyLineState curr_line_state; + + CHECK_FOR_INTERRUPTS(); + + /* File read completed & no elements to process. */ + if (!is_read_in_progress && + (pcshared_info->populated == + pg_atomic_read_u64(&pcshared_info->total_worker_processed))) + { + write_pos = -1; + break; + } + + /* Get the current line information. */ + lineInfo = &pcshared_info->line_boundaries.ring[write_pos]; + curr_line_state = pg_atomic_read_u32(&lineInfo->line_state); + if ((write_pos % WORKER_CHUNK_COUNT == 0) && + (curr_line_state == LINE_WORKER_PROCESSED || + curr_line_state == LINE_WORKER_PROCESSING)) + { + pcdata->worker_processed_pos = write_pos; + write_pos = (write_pos + WORKER_CHUNK_COUNT) % RINGSIZE; + continue; + } + + /* Get the size of this line. */ + dataSize = pg_atomic_read_u32(&lineInfo->line_size); + + if (dataSize != 0) /* If not an empty line. */ + { + /* Get the block information. */ + data_blk_ptr = &pcshared_info->data_blocks[lineInfo->first_block]; + + if (!data_blk_ptr->curr_blk_completed && (dataSize == -1)) + { + /* Wait till the current line or block is added. */ + COPY_WAIT_TO_PROCESS() + continue; + } + } + + /* Make sure that no worker has consumed this element. */ + if (pg_atomic_compare_exchange_u32(&lineInfo->line_state, + &line_state, LINE_WORKER_PROCESSING)) + break; + + line_state = LINE_LEADER_POPULATING; + /* Make sure that no worker has consumed this element. */ + if (pg_atomic_compare_exchange_u32(&lineInfo->line_state, + &line_state, LINE_WORKER_PROCESSING)) + break; + } + + pcdata->worker_processed_pos = write_pos; + return write_pos; +} + +/* + * GetFreeCopyBlock + * + * Get a free block for data to be copied. + */ +static pg_attribute_always_inline uint32 +GetFreeCopyBlock(ParallelCopyShmInfo *pcshared_info) +{ + int count = 0; + uint32 last_free_block = pcshared_info->cur_block_pos; + uint32 block_pos = (last_free_block != -1) ? ((last_free_block + 1) % MAX_BLOCKS_COUNT) : 0; + + /* + * Get a new block for copying data, don't check current block, current + * block will have some unprocessed data. + */ + while (count < (MAX_BLOCKS_COUNT - 1)) + { + ParallelCopyDataBlock *dataBlkPtr = &pcshared_info->data_blocks[block_pos]; + uint32 unprocessed_line_parts = pg_atomic_read_u32(&dataBlkPtr->unprocessed_line_parts); + + if (unprocessed_line_parts == 0) + { + dataBlkPtr->curr_blk_completed = false; + dataBlkPtr->skip_bytes = 0; + dataBlkPtr->following_block = -1; + pcshared_info->cur_block_pos = block_pos; + MemSet(&dataBlkPtr->data[0], 0, DATA_BLOCK_SIZE); + return block_pos; + } + + block_pos = (block_pos + 1) % MAX_BLOCKS_COUNT; + count++; + } + + return -1; +} + +/* + * WaitGetFreeCopyBlock + * + * If there are no blocks available, wait and get a block for copying data. + */ +uint32 +WaitGetFreeCopyBlock(ParallelCopyShmInfo *pcshared_info) +{ + uint32 new_free_pos = -1; + + for (;;) + { + new_free_pos = GetFreeCopyBlock(pcshared_info); + if (new_free_pos != -1) /* We have got one block, break now. */ + break; + + COPY_WAIT_TO_PROCESS() + } + + return new_free_pos; +} + +/* + * SetRawBufForLoad + * + * Set raw_buf to the shared memory where the file data must be read. + */ +void +SetRawBufForLoad(CopyState cstate, uint32 line_size, uint32 copy_buf_len, + uint32 raw_buf_ptr, char **copy_raw_buf) +{ + ParallelCopyShmInfo *pcshared_info; + uint32 cur_block_pos; + uint32 next_block_pos; + ParallelCopyDataBlock *cur_data_blk_ptr = NULL; + ParallelCopyDataBlock *next_data_blk_ptr = NULL; + + Assert(IsParallelCopy()); + + pcshared_info = cstate->pcdata->pcshared_info; + cur_block_pos = pcshared_info->cur_block_pos; + cur_data_blk_ptr = (cstate->raw_buf) ? &pcshared_info->data_blocks[cur_block_pos] : NULL; + next_block_pos = WaitGetFreeCopyBlock(pcshared_info); + next_data_blk_ptr = &pcshared_info->data_blocks[next_block_pos]; + + /* set raw_buf to the data block in shared memory */ + cstate->raw_buf = next_data_blk_ptr->data; + *copy_raw_buf = cstate->raw_buf; + if (cur_data_blk_ptr) + { + if (line_size) + { + /* + * Mark the previous block as completed, worker can start copying + * this data. + */ + cur_data_blk_ptr->following_block = next_block_pos; + pg_atomic_add_fetch_u32(&cur_data_blk_ptr->unprocessed_line_parts, 1); + cur_data_blk_ptr->curr_blk_completed = true; + } + + cur_data_blk_ptr->skip_bytes = copy_buf_len - raw_buf_ptr; + cstate->raw_buf_len = cur_data_blk_ptr->skip_bytes; + + /* Copy the skip bytes to the next block to be processed. */ + if (cur_data_blk_ptr->skip_bytes) + memcpy(cstate->raw_buf, cur_data_blk_ptr->data + raw_buf_ptr, + cur_data_blk_ptr->skip_bytes); + } + else + cstate->raw_buf_len = 0; + + cstate->raw_buf_index = 0; +} + +/* + * EndLineParallelCopy + * + * Update the line information in shared memory. + */ +void +EndLineParallelCopy(CopyState cstate, uint32 line_pos, uint32 line_size, + uint32 raw_buf_ptr) +{ + uint8 new_line_size; + + if (!IsParallelCopy()) + return; + + if (!IsHeaderLine()) + { + ParallelCopyShmInfo *pcshared_info = cstate->pcdata->pcshared_info; + + /* Set the newline size. */ + if (cstate->eol_type == EOL_NL || cstate->eol_type == EOL_CR) + new_line_size = 1; + else if (cstate->eol_type == EOL_CRNL) + new_line_size = 2; + else + new_line_size = 0; + + if (line_size) + { + /* + * If the new_line_size > raw_buf_ptr, then the new block has only + * new line char content. The unprocessed count should not be + * increased in this case. + */ + if (raw_buf_ptr > new_line_size) + { + uint32 cur_block_pos = pcshared_info->cur_block_pos; + ParallelCopyDataBlock *curr_data_blk_ptr = &pcshared_info->data_blocks[cur_block_pos]; + + pg_atomic_add_fetch_u32(&curr_data_blk_ptr->unprocessed_line_parts, 1); + } + + /* + * Update line size & line state, other members are already + * updated. + */ + (void) UpdateSharedLineInfo(cstate, -1, -1, line_size, + LINE_LEADER_POPULATED, line_pos); + } + else if (new_line_size) + /* This means only new line char, empty record should be inserted. */ + (void) UpdateSharedLineInfo(cstate, -1, -1, 0, + LINE_LEADER_POPULATED, -1); + } +} + +/* + * ExecBeforeStmtTrigger + * + * Execute the before statement trigger, this will be executed for parallel copy + * by the leader process. This function code changes has been taken from + * CopyFrom function. Refer to comments section of respective code in CopyFrom + * function for more detailed information. + */ +void +ExecBeforeStmtTrigger(CopyState cstate) +{ + EState *estate = CreateExecutorState(); + ResultRelInfo *resultRelInfo; + + Assert(IsLeader()); + ExecInitRangeTable(estate, cstate->range_table); + resultRelInfo = makeNode(ResultRelInfo); + ExecInitResultRelation(estate, resultRelInfo, 1); + CheckValidResultRel(resultRelInfo, CMD_INSERT); + AfterTriggerBeginQuery(); + ExecBSInsertTriggers(estate, resultRelInfo); + AfterTriggerEndQuery(estate); + ExecCloseResultRelations(estate); + ExecCloseRangeTableRelations(estate); + FreeExecutorState(estate); +} diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index 5238a96..b9da375 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -2381,7 +2381,7 @@ psql_completion(const char *text, int start, int end) /* Complete COPY FROM|TO filename WITH ( */ else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "(")) COMPLETE_WITH("FORMAT", "FREEZE", "DELIMITER", "NULL", - "HEADER", "QUOTE", "ESCAPE", "FORCE_QUOTE", + "HEADER", "PARALLEL", "QUOTE", "ESCAPE", "FORCE_QUOTE", "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING"); /* Complete COPY FROM|TO filename WITH (FORMAT */ diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index fd3a78e..ac92aca 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -14,6 +14,7 @@ #ifndef COPY_H #define COPY_H +#include "access/parallel.h" #include "commands/copy_internal.h" #include "nodes/execnodes.h" #include "nodes/parsenodes.h" @@ -38,6 +39,26 @@ extern uint64 CopyFrom(CopyState cstate); extern DestReceiver *CreateCopyDestReceiver(void); -extern bool IsParallelCopyAllowed(CopyState cstate, Oid relid); +extern void PopulateCommonCStateInfo(CopyState cstate, TupleDesc tup_desc, + List *attnamelist); +extern void ConvertToServerEncoding(CopyState cstate); +extern void ParallelCopyMain(dsm_segment *seg, shm_toc *toc); +extern ParallelContext *BeginParallelCopy(CopyState cstate, List *attnamelist, Oid relid); +extern void ParallelCopyFrom(CopyState cstate); +extern void EndParallelCopy(ParallelContext *pcxt); +extern bool IsParallelCopyAllowed(CopyState cstate, Oid relid); +extern void ExecBeforeStmtTrigger(CopyState cstate); +extern void CheckTargetRelValidity(CopyState cstate); +extern void PopulateCStateCatalogInfo(CopyState cstate); +extern uint32 GetLinePosition(CopyState cstate); +extern bool GetWorkerLine(CopyState cstate); +extern bool CopyReadLine(CopyState cstate); +extern uint32 WaitGetFreeCopyBlock(ParallelCopyShmInfo *pcshared_info); +extern void SetRawBufForLoad(CopyState cstate, uint32 line_size, uint32 copy_buf_len, + uint32 raw_buf_ptr, char **copy_raw_buf); +extern uint32 UpdateSharedLineInfo(CopyState cstate, uint32 blk_pos, uint32 offset, + uint32 line_size, uint32 line_state, uint32 blk_line_pos); +extern void EndLineParallelCopy(CopyState cstate, uint32 line_pos, uint32 line_size, + uint32 raw_buf_ptr); #endif /* COPY_H */ diff --git a/src/include/commands/copy_internal.h b/src/include/commands/copy_internal.h index ea4bfca..30eaa2e 100644 --- a/src/include/commands/copy_internal.h +++ b/src/include/commands/copy_internal.h @@ -17,6 +17,42 @@ #include "commands/trigger.h" #include "executor/executor.h" +#define RAW_BUF_SIZE 65536 /* we palloc RAW_BUF_SIZE+1 bytes */ + +/* + * The macros DATA_BLOCK_SIZE, RINGSIZE & MAX_BLOCKS_COUNT stores the records + * read from the file that need to be inserted into the relation. These values + * help in the handover of multiple records with the significant size of data to + * be processed by each of the workers. This also ensures there is no context + * switch and the work is fairly distributed among the workers. This number + * showed best results in the performance tests. + */ +#define DATA_BLOCK_SIZE RAW_BUF_SIZE + +/* + * It can hold MAX_BLOCKS_COUNT blocks of RAW_BUF_SIZE data in DSM to be + * processed by the worker. + */ +#define MAX_BLOCKS_COUNT 1024 + +/* + * It can hold upto RINGSIZE record information for worker to process. RINGSIZE + * should be a multiple of WORKER_CHUNK_COUNT, as wrap around cases is currently + * not handled while selecting the WORKER_CHUNK_COUNT by the worker. + */ +#define RINGSIZE (10 * 1024) + +/* + * While accessing DSM, each worker will pick the WORKER_CHUNK_COUNT records + * from the DSM data blocks at a time and store them in it's local memory. This + * is to make workers not contend much while getting record information from the + * DSM. Read RINGSIZE comments before changing this value. + */ +#define WORKER_CHUNK_COUNT 64 + +#define IsParallelCopy() (cstate->is_parallel) +#define IsLeader() (cstate->pcdata->is_leader) +#define IsWorker() (IsParallelCopy() && !IsLeader()) #define IsHeaderLine() (cstate->header_line && cstate->cur_lineno == 1) /* @@ -43,6 +79,18 @@ typedef enum EolType } EolType; /* + * State of the line. + */ +typedef enum ParallelCopyLineState +{ + LINE_INIT, /* initial state of line */ + LINE_LEADER_POPULATING, /* leader processing line */ + LINE_LEADER_POPULATED, /* leader completed populating line */ + LINE_WORKER_PROCESSING, /* worker processing line */ + LINE_WORKER_PROCESSED /* worker completed processing line */ +} ParallelCopyLineState; + +/* * Represents the heap insert method to be used during COPY FROM. */ typedef enum CopyInsertMethod @@ -52,7 +100,180 @@ typedef enum CopyInsertMethod CIM_MULTI_CONDITIONAL /* use table_multi_insert only if valid */ } CopyInsertMethod; -#define RAW_BUF_SIZE 65536 /* we palloc RAW_BUF_SIZE+1 bytes */ +/* + * Copy data block information. + * + * These data blocks are created in DSM. Data read from file will be copied in + * these DSM data blocks. The leader process identifies the records and the + * record information will be shared to the workers. The workers will insert the + * records into the table. There can be one or more number of records in each of + * the data block based on the record size. + */ +typedef struct ParallelCopyDataBlock +{ + /* The number of unprocessed lines in the current block. */ + pg_atomic_uint32 unprocessed_line_parts; + + /* + * If the current line data is continued into another block, + * following_block will have the position where the remaining data need to + * be read. + */ + uint32 following_block; + + /* + * This flag will be set, when the leader finds out this block can be read + * safely by the worker. This helps the worker to start processing the + * line early where the line will be spread across many blocks and the + * worker need not wait for the complete line to be processed. + */ + bool curr_blk_completed; + + /* + * Few bytes need to be skipped from this block, this will be set when a + * sequence of characters like \r\n is expected, but end of our block + * contained only \r. In this case we copy the data from \r into the new + * block as they have to be processed together to identify end of line. + * Worker will use skip_bytes to know that this data must be skipped from + * this data block. + */ + uint8 skip_bytes; + char data[DATA_BLOCK_SIZE]; /* data read from file */ +} ParallelCopyDataBlock; + +/* + * Individual line information. + * + * ParallelCopyLineBoundary is common data structure between leader & worker. + * Leader process will be populating data block, data block offset & the size of + * the record in DSM for the workers to copy the data into the relation. + * The leader & worker process access the shared line information by following + * the below steps to avoid any data corruption or hang: + * + * Leader should operate in the following order: + * 1) check if line_size is -1, if line_size is not -1 wait until line_size is + * set to -1 by the worker. If line_size is -1 it means worker is still + * processing. + * 2) set line_state to LINE_LEADER_POPULATING, so that the worker knows that + * leader is populating this line. + * 3) update first_block, start_offset & cur_lineno in any order. + * 4) update line_size. + * 5) update line_state to LINE_LEADER_POPULATED. + * + * Worker should operate in the following order: + * 1) check line_state is LINE_LEADER_POPULATED, if not it means leader is still + * populating the data. + * 2) read line_size to know the size of the data. + * 3) only one worker should choose one line for processing, this is handled by + * using pg_atomic_compare_exchange_u32, worker will change the state to + * LINE_WORKER_PROCESSING only if line_state is LINE_LEADER_POPULATED. + * 4) read first_block, start_offset & cur_lineno in any order. + * 5) process line_size data. + * 6) update line_size to -1. + */ +typedef struct ParallelCopyLineBoundary +{ + /* Position of the first block in data_blocks array. */ + uint32 first_block; + uint32 start_offset; /* start offset of the line */ + + /* + * Size of the current line -1 means line is yet to be filled completely, + * 0 means empty line, >0 means line filled with line size data. + */ + pg_atomic_uint32 line_size; + pg_atomic_uint32 line_state; /* line state */ + uint64 cur_lineno; /* line number for error messages */ +} ParallelCopyLineBoundary; + +/* + * Circular queue used to store the line information. + */ +typedef struct ParallelCopyLineBoundaries +{ + /* Position for the leader to populate a line. */ + uint32 pos; + + /* Data read from the file/stdin by the leader process. */ + ParallelCopyLineBoundary ring[RINGSIZE]; +} ParallelCopyLineBoundaries; + +/* + * Shared information among parallel copy workers. This will be allocated in the + * DSM segment. + */ +typedef struct ParallelCopyShmInfo +{ + bool is_read_in_progress; /* file read status */ + + /* + * Actual lines inserted by worker, will not be same as + * total_worker_processed if where condition is specified along with copy. + * This will be the actual records inserted into the relation. + */ + pg_atomic_uint64 processed; + + /* + * The number of records currently processed by the worker, this will also + * include the number of records that was filtered because of where + * clause. + */ + pg_atomic_uint64 total_worker_processed; + uint64 populated; /* lines populated by leader */ + uint32 cur_block_pos; /* current data block */ + ParallelCopyDataBlock data_blocks[MAX_BLOCKS_COUNT]; /* data block array */ + ParallelCopyLineBoundaries line_boundaries; /* line array */ +} ParallelCopyShmInfo; + +/* + * Parallel copy line buffer information. + */ +typedef struct ParallelCopyLineBuf +{ + StringInfoData line_buf; + uint64 cur_lineno; /* line number for error messages */ +} ParallelCopyLineBuf; + +/* + * This structure helps in storing the common List/Node from CopyStateData that + * are required by the workers. This information will then be copied and stored + * into the DSM for the worker to retrieve and copy it to CopyStateData. + */ +typedef struct SerializedListToStrCState +{ + char *whereClauseStr; + char *rangeTableStr; + char *attnameListStr; + char *notnullListStr; + char *nullListStr; + char *convertListStr; +} SerializedListToStrCState; + +/* + * Parallel copy data information. + */ +typedef struct ParallelCopyData +{ + Oid relid; /* relation id of the table */ + ParallelCopyShmInfo *pcshared_info; /* common info in shared memory */ + bool is_leader; + + /* line position which worker is processing */ + uint32 worker_processed_pos; + + WalUsage *walusage; + BufferUsage *bufferusage; + + /* + * Local line_buf array, workers will copy it here and release the lines + * for the leader to continue. + */ + ParallelCopyLineBuf worker_line_buf[WORKER_CHUNK_COUNT]; + uint32 worker_line_buf_count; /* Number of lines */ + + /* Current position in worker_line_buf */ + uint32 worker_line_buf_pos; +} ParallelCopyData; typedef int (*copy_data_source_cb) (void *outbuf, int minread, int maxread); @@ -188,6 +409,9 @@ typedef struct CopyStateData char *raw_buf; int raw_buf_index; /* next byte to process */ int raw_buf_len; /* total # of bytes stored */ + int nworkers; + bool is_parallel; + ParallelCopyData *pcdata; /* Shorthand for number of unconsumed bytes available in raw_buf */ #define RAW_BUF_BYTES(cstate) ((cstate)->raw_buf_len - (cstate)->raw_buf_index) } CopyStateData; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index fde701b..13c0f53 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1705,6 +1705,13 @@ ParallelBitmapHeapState ParallelBlockTableScanDesc ParallelCompletionPtr ParallelContext +ParallelCopyLineBoundaries +ParallelCopyLineBoundary +ParallelCopyData +ParallelCopyDataBlock +ParallelCopyLineBuf +ParallelCopyLineState +ParallelCopyShmInfo ParallelExecutorInfo ParallelHashGrowth ParallelHashJoinBatch @@ -2227,6 +2234,7 @@ SerCommitSeqNo SerialControl SerializableXactHandle SerializedActiveRelMaps +SerializedListToStrCState SerializedReindexState SerializedSnapshotData SerializedTransactionState -- 1.8.3.1