From f80e7ea77a4faa2abd64782e9bdfc20ad8d0eff3 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Sun, 13 Jul 2025 06:50:40 -0700 Subject: [PATCH 1/2] Refactor CopyToStateData and CopyFromStateData. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch-through: --- src/backend/commands/copyfrom.c | 238 ++++++------ src/backend/commands/copyfromparse.c | 446 ++++++++++++----------- src/backend/commands/copyto.c | 111 +++--- src/include/commands/copy.h | 55 ++- src/include/commands/copyfrom_internal.h | 22 +- 5 files changed, 457 insertions(+), 415 deletions(-) diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index fbbbc09a97b..03779838654 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -168,6 +168,7 @@ CopyFromGetRoutine(const CopyFormatOptions *opts) static void CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc) { + CopyFromExecutionData *edata = cstate->edata; AttrNumber attr_count; /* @@ -175,24 +176,24 @@ CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc) * converted input data. Otherwise, we can just point input_buf to the * same buffer as raw_buf. */ - if (cstate->need_transcoding) + if (cstate->edata->need_transcoding) { - cstate->input_buf = (char *) palloc(INPUT_BUF_SIZE + 1); - cstate->input_buf_index = cstate->input_buf_len = 0; + edata->input_buf = (char *) palloc(INPUT_BUF_SIZE + 1); + edata->input_buf_index = edata->input_buf_len = 0; } else - cstate->input_buf = cstate->raw_buf; - cstate->input_reached_eof = false; + edata->input_buf = edata->raw_buf; + edata->input_reached_eof = false; - initStringInfo(&cstate->line_buf); + initStringInfo(&edata->line_buf); /* * Create workspace for CopyReadAttributes results; used by CSV and text * format. */ attr_count = list_length(cstate->attnumlist); - cstate->max_fields = attr_count; - cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *)); + edata->max_fields = attr_count; + edata->raw_fields = (char **) palloc(attr_count * sizeof(char *)); } /* @@ -254,48 +255,49 @@ void CopyFromErrorCallback(void *arg) { CopyFromState cstate = (CopyFromState) arg; + CopyFromExecutionData *edata = cstate->edata; - if (cstate->relname_only) + if (edata->relname_only) { errcontext("COPY %s", - cstate->cur_relname); + edata->cur_relname); return; } if (cstate->opts.binary) { /* can't usefully display the data */ - if (cstate->cur_attname) + if (edata->cur_attname) errcontext("COPY %s, line %" PRIu64 ", column %s", - cstate->cur_relname, - cstate->cur_lineno, - cstate->cur_attname); + edata->cur_relname, + edata->cur_lineno, + edata->cur_attname); else errcontext("COPY %s, line %" PRIu64, - cstate->cur_relname, - cstate->cur_lineno); + edata->cur_relname, + edata->cur_lineno); } else { - if (cstate->cur_attname && cstate->cur_attval) + if (edata->cur_attname && edata->cur_attval) { /* error is relevant to a particular column */ char *attval; - attval = CopyLimitPrintoutLength(cstate->cur_attval); + attval = CopyLimitPrintoutLength(edata->cur_attval); errcontext("COPY %s, line %" PRIu64 ", column %s: \"%s\"", - cstate->cur_relname, - cstate->cur_lineno, - cstate->cur_attname, + edata->cur_relname, + edata->cur_lineno, + edata->cur_attname, attval); pfree(attval); } - else if (cstate->cur_attname) + else if (edata->cur_attname) { /* error is relevant to a particular column, value is NULL */ errcontext("COPY %s, line %" PRIu64 ", column %s: null input", - cstate->cur_relname, - cstate->cur_lineno, - cstate->cur_attname); + edata->cur_relname, + edata->cur_lineno, + edata->cur_attname); } else { @@ -304,21 +306,21 @@ CopyFromErrorCallback(void *arg) * * If line_buf still contains the correct line, print it. */ - if (cstate->line_buf_valid) + if (edata->line_buf_valid) { char *lineval; - lineval = CopyLimitPrintoutLength(cstate->line_buf.data); + lineval = CopyLimitPrintoutLength(edata->line_buf.data); errcontext("COPY %s, line %" PRIu64 ": \"%s\"", - cstate->cur_relname, - cstate->cur_lineno, lineval); + edata->cur_relname, + edata->cur_lineno, lineval); pfree(lineval); } else { errcontext("COPY %s, line %" PRIu64, - cstate->cur_relname, - cstate->cur_lineno); + edata->cur_relname, + edata->cur_lineno); } } } @@ -448,6 +450,7 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, int64 *processed) { CopyFromState cstate = miinfo->cstate; + CopyFromExecutionData *edata = cstate->edata; EState *estate = miinfo->estate; int nused = buffer->nused; ResultRelInfo *resultRelInfo = buffer->resultRelInfo; @@ -469,8 +472,8 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, * We suppress error context information other than the relation name, * if one of the operations below fails. */ - Assert(!cstate->relname_only); - cstate->relname_only = true; + Assert(!edata->relname_only); + edata->relname_only = true; while (sent < nused) { @@ -514,7 +517,7 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, ExecARInsertTriggers(estate, resultRelInfo, slot, NIL, - cstate->transition_capture); + edata->transition_capture); } } @@ -528,14 +531,14 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, ExecClearTuple(slots[i]); /* reset relname_only */ - cstate->relname_only = false; + edata->relname_only = false; } else { CommandId mycid = miinfo->mycid; int ti_options = miinfo->ti_options; - bool line_buf_valid = cstate->line_buf_valid; - uint64 save_cur_lineno = cstate->cur_lineno; + bool line_buf_valid = edata->line_buf_valid; + uint64 save_cur_lineno = edata->cur_lineno; MemoryContext oldcontext; Assert(buffer->bistate != NULL); @@ -544,7 +547,7 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, * Print error context information correctly, if one of the operations * below fails. */ - cstate->line_buf_valid = false; + edata->line_buf_valid = false; /* * table_multi_insert may leak memory, so switch to short-lived memory @@ -569,14 +572,14 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, { List *recheckIndexes; - cstate->cur_lineno = buffer->linenos[i]; + edata->cur_lineno = buffer->linenos[i]; recheckIndexes = ExecInsertIndexTuples(resultRelInfo, buffer->slots[i], estate, false, false, NULL, NIL, false); ExecARInsertTriggers(estate, resultRelInfo, slots[i], recheckIndexes, - cstate->transition_capture); + edata->transition_capture); list_free(recheckIndexes); } @@ -588,10 +591,10 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, (resultRelInfo->ri_TrigDesc->trig_insert_after_row || resultRelInfo->ri_TrigDesc->trig_insert_new_table)) { - cstate->cur_lineno = buffer->linenos[i]; + edata->cur_lineno = buffer->linenos[i]; ExecARInsertTriggers(estate, resultRelInfo, slots[i], NIL, - cstate->transition_capture); + edata->transition_capture); } ExecClearTuple(slots[i]); @@ -603,8 +606,8 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, *processed); /* reset cur_lineno and line_buf_valid to what they were */ - cstate->line_buf_valid = line_buf_valid; - cstate->cur_lineno = save_cur_lineno; + edata->line_buf_valid = line_buf_valid; + edata->cur_lineno = save_cur_lineno; } /* Mark that all slots are free */ @@ -778,6 +781,7 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, uint64 CopyFrom(CopyFromState cstate) { + CopyFromExecutionData *edata = cstate->edata; ResultRelInfo *resultRelInfo; ResultRelInfo *target_resultRelInfo; ResultRelInfo *prevResultRelInfo = NULL; @@ -801,10 +805,10 @@ CopyFrom(CopyFromState cstate) bool leafpart_use_multi_insert = false; Assert(cstate->rel); - Assert(list_length(cstate->range_table) == 1); + Assert(list_length(edata->range_table) == 1); if (cstate->opts.on_error != COPY_ON_ERROR_STOP) - Assert(cstate->escontext); + Assert(edata->escontext); /* * The target must be a plain, foreign, or partitioned relation, or have @@ -913,7 +917,7 @@ CopyFrom(CopyFromState cstate) * index-entry-making machinery. (There used to be a huge amount of code * here that basically duplicated execUtils.c ...) */ - ExecInitRangeTable(estate, cstate->range_table, cstate->rteperminfos, + ExecInitRangeTable(estate, edata->range_table, edata->rteperminfos, bms_make_singleton(1)); resultRelInfo = target_resultRelInfo = makeNode(ResultRelInfo); ExecInitResultRelation(estate, resultRelInfo, 1); @@ -968,7 +972,7 @@ CopyFrom(CopyFromState cstate) * transition capture is active, we also set it in mtstate, which is * passed to ExecFindPartition() below. */ - cstate->transition_capture = mtstate->mt_transition_capture = + edata->transition_capture = mtstate->mt_transition_capture = MakeTransitionCaptureState(cstate->rel->trigdesc, RelationGetRelid(cstate->rel), CMD_INSERT); @@ -980,9 +984,9 @@ CopyFrom(CopyFromState cstate) if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) proute = ExecSetupPartitionTupleRouting(estate, cstate->rel); - if (cstate->whereClause) - cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause), - &mtstate->ps); + if (edata->whereClause) + edata->qualexpr = ExecInitQual(castNode(List, edata->whereClause), + &mtstate->ps); /* * It's generally more efficient to prepare a bunch of tuples for @@ -1025,7 +1029,7 @@ CopyFrom(CopyFromState cstate) */ insertMethod = CIM_SINGLE; } - else if (cstate->volatile_defexprs) + else if (edata->volatile_defexprs) { /* * Can't support multi-inserts if there are any volatile default @@ -1038,7 +1042,7 @@ CopyFrom(CopyFromState cstate) */ insertMethod = CIM_SINGLE; } - else if (contain_volatile_functions(cstate->whereClause)) + else if (contain_volatile_functions(edata->whereClause)) { /* * Can't support multi-inserts if there are any volatile function @@ -1150,7 +1154,7 @@ CopyFrom(CopyFromState cstate) break; if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE && - cstate->escontext->error_occurred) + edata->escontext->error_occurred) { /* * Soft error occurred, skip this tuple and just make @@ -1158,14 +1162,14 @@ CopyFrom(CopyFromState cstate) * don't set details_wanted and error_data is not to be filled, * just resetting error_occurred is enough. */ - cstate->escontext->error_occurred = false; + edata->escontext->error_occurred = false; /* Report that this tuple was skipped by the ON_ERROR clause */ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED, - cstate->num_errors); + edata->num_errors); if (cstate->opts.reject_limit > 0 && - cstate->num_errors > cstate->opts.reject_limit) + edata->num_errors > cstate->opts.reject_limit) ereport(ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("skipped more than REJECT_LIMIT (%" PRId64 ") rows due to data type incompatibility", @@ -1186,11 +1190,11 @@ CopyFrom(CopyFromState cstate) /* Triggers and stuff need to be invoked in query context. */ MemoryContextSwitchTo(oldcontext); - if (cstate->whereClause) + if (edata->whereClause) { econtext->ecxt_scantuple = myslot; /* Skip items that don't match COPY's WHERE clause */ - if (!ExecQual(cstate->qualexpr, econtext)) + if (!ExecQual(edata->qualexpr, econtext)) { /* * Report that this tuple was filtered out by the WHERE @@ -1266,8 +1270,8 @@ CopyFrom(CopyFromState cstate) * we can just remember the original unconverted tuple to avoid a * needless round trip conversion. */ - if (cstate->transition_capture != NULL) - cstate->transition_capture->tcs_original_insert_tuple = + if (edata->transition_capture != NULL) + edata->transition_capture->tcs_original_insert_tuple = !has_before_insert_row_trig ? myslot : NULL; /* @@ -1379,8 +1383,8 @@ CopyFrom(CopyFromState cstate) /* Add this tuple to the tuple buffer */ CopyMultiInsertInfoStore(&multiInsertInfo, resultRelInfo, myslot, - cstate->line_buf.len, - cstate->cur_lineno); + edata->line_buf.len, + edata->cur_lineno); /* * If enough inserts have queued up, then flush all @@ -1440,7 +1444,7 @@ CopyFrom(CopyFromState cstate) /* AFTER ROW INSERT Triggers */ ExecARInsertTriggers(estate, resultRelInfo, myslot, - recheckIndexes, cstate->transition_capture); + recheckIndexes, edata->transition_capture); list_free(recheckIndexes); } @@ -1468,13 +1472,13 @@ CopyFrom(CopyFromState cstate) error_context_stack = errcallback.previous; if (cstate->opts.on_error != COPY_ON_ERROR_STOP && - cstate->num_errors > 0 && + edata->num_errors > 0 && cstate->opts.log_verbosity >= COPY_LOG_VERBOSITY_DEFAULT) ereport(NOTICE, errmsg_plural("%" PRIu64 " row was skipped due to data type incompatibility", "%" PRIu64 " rows were skipped due to data type incompatibility", - cstate->num_errors, - cstate->num_errors)); + edata->num_errors, + edata->num_errors)); if (bistate != NULL) FreeBulkInsertState(bistate); @@ -1482,7 +1486,7 @@ CopyFrom(CopyFromState cstate) MemoryContextSwitchTo(oldcontext); /* Execute AFTER STATEMENT insertion triggers */ - ExecASInsertTriggers(estate, target_resultRelInfo, cstate->transition_capture); + ExecASInsertTriggers(estate, target_resultRelInfo, edata->transition_capture); /* Handle queued AFTER triggers */ AfterTriggerEndQuery(estate); @@ -1536,6 +1540,7 @@ BeginCopyFrom(ParseState *pstate, List *options) { CopyFromState cstate; + CopyFromExecutionData *edata; bool pipe = (filename == NULL); TupleDesc tupDesc; AttrNumber num_phys_attrs, @@ -1560,15 +1565,18 @@ BeginCopyFrom(ParseState *pstate, /* Allocate workspace and zero all fields */ cstate = (CopyFromStateData *) palloc0(sizeof(CopyFromStateData)); + edata = (CopyFromExecutionData *) palloc0(sizeof(CopyFromExecutionData)); + cstate->edata = edata; + /* * 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); + edata->copycontext = AllocSetContextCreate(CurrentMemoryContext, + "COPY", + ALLOCSET_DEFAULT_SIZES); - oldcontext = MemoryContextSwitchTo(cstate->copycontext); + oldcontext = MemoryContextSwitchTo(edata->copycontext); /* Extract options from the statement node tree */ ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , options); @@ -1617,19 +1625,19 @@ BeginCopyFrom(ParseState *pstate, /* Set up soft error handler for ON_ERROR */ if (cstate->opts.on_error != COPY_ON_ERROR_STOP) { - cstate->escontext = makeNode(ErrorSaveContext); - cstate->escontext->type = T_ErrorSaveContext; - cstate->escontext->error_occurred = false; + edata->escontext = makeNode(ErrorSaveContext); + edata->escontext->type = T_ErrorSaveContext; + edata->escontext->error_occurred = false; /* * Currently we only support COPY_ON_ERROR_IGNORE. We'll add other * options later */ if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE) - cstate->escontext->details_wanted = false; + edata->escontext->details_wanted = false; } else - cstate->escontext = NULL; + edata->escontext = NULL; /* Convert FORCE_NULL name list to per-column flags, check validity */ cstate->opts.force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool)); @@ -1663,7 +1671,7 @@ BeginCopyFrom(ParseState *pstate, List *attnums; ListCell *cur; - cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool)); + edata->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool)); attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.convert_select); @@ -1677,7 +1685,7 @@ BeginCopyFrom(ParseState *pstate, (errcode(ERRCODE_INVALID_COLUMN_REFERENCE), errmsg_internal("selected column \"%s\" not referenced by COPY", NameStr(attr->attname)))); - cstate->convert_select_flags[attnum - 1] = true; + edata->convert_select_flags[attnum - 1] = true; } } @@ -1694,14 +1702,14 @@ BeginCopyFrom(ParseState *pstate, cstate->file_encoding == PG_SQL_ASCII || GetDatabaseEncoding() == PG_SQL_ASCII) { - cstate->need_transcoding = false; + cstate->edata->need_transcoding = false; } else { - cstate->need_transcoding = true; - cstate->conversion_proc = FindDefaultConversionProc(cstate->file_encoding, - GetDatabaseEncoding()); - if (!OidIsValid(cstate->conversion_proc)) + cstate->edata->need_transcoding = true; + cstate->edata->conversion_proc = FindDefaultConversionProc(cstate->file_encoding, + GetDatabaseEncoding()); + if (!OidIsValid(cstate->edata->conversion_proc)) ereport(ERROR, (errcode(ERRCODE_UNDEFINED_FUNCTION), errmsg("default conversion function for encoding \"%s\" to \"%s\" does not exist", @@ -1709,17 +1717,17 @@ BeginCopyFrom(ParseState *pstate, pg_encoding_to_char(GetDatabaseEncoding())))); } - cstate->copy_src = COPY_FILE; /* default */ + edata->copy_src = COPY_FILE; /* default */ - cstate->whereClause = whereClause; + edata->whereClause = whereClause; /* Initialize state variables */ - cstate->eol_type = EOL_UNKNOWN; - cstate->cur_relname = RelationGetRelationName(cstate->rel); - cstate->cur_lineno = 0; - cstate->cur_attname = NULL; - cstate->cur_attval = NULL; - cstate->relname_only = false; + edata->eol_type = EOL_UNKNOWN; + edata->cur_relname = RelationGetRelationName(cstate->rel); + edata->cur_lineno = 0; + edata->cur_attname = NULL; + edata->cur_attval = NULL; + edata->relname_only = false; /* * Allocate buffers for the input pipeline. @@ -1727,17 +1735,17 @@ BeginCopyFrom(ParseState *pstate, * attribute_buf and raw_buf are used in both text and binary modes, but * input_buf and line_buf only in text mode. */ - cstate->raw_buf = palloc(RAW_BUF_SIZE + 1); - cstate->raw_buf_index = cstate->raw_buf_len = 0; - cstate->raw_reached_eof = false; + edata->raw_buf = palloc(RAW_BUF_SIZE + 1); + edata->raw_buf_index = edata->raw_buf_len = 0; + edata->raw_reached_eof = false; - initStringInfo(&cstate->attribute_buf); + initStringInfo(&edata->attribute_buf); /* Assign range table and rteperminfos, we'll need them in CopyFrom. */ if (pstate) { - cstate->range_table = pstate->p_rtable; - cstate->rteperminfos = pstate->p_rteperminfos; + edata->range_table = pstate->p_rtable; + edata->rteperminfos = pstate->p_rteperminfos; } num_defaults = 0; @@ -1818,26 +1826,26 @@ BeginCopyFrom(ParseState *pstate, } } - cstate->defaults = (bool *) palloc0(tupDesc->natts * sizeof(bool)); + edata->defaults = (bool *) palloc0(tupDesc->natts * sizeof(bool)); /* initialize progress */ pgstat_progress_start_command(PROGRESS_COMMAND_COPY, cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid); - cstate->bytes_processed = 0; + edata->bytes_processed = 0; /* We keep those variables in cstate. */ cstate->in_functions = in_functions; cstate->typioparams = typioparams; - cstate->defmap = defmap; - cstate->defexprs = defexprs; - cstate->volatile_defexprs = volatile_defexprs; - cstate->num_defaults = num_defaults; + edata->defmap = defmap; + edata->defexprs = defexprs; + edata->volatile_defexprs = volatile_defexprs; + edata->num_defaults = num_defaults; cstate->is_program = is_program; if (data_source_cb) { progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK; - cstate->copy_src = COPY_CALLBACK; + edata->copy_src = COPY_CALLBACK; cstate->data_source_cb = data_source_cb; } else if (pipe) @@ -1847,7 +1855,7 @@ BeginCopyFrom(ParseState *pstate, if (whereToSendOutput == DestRemote) ReceiveCopyBegin(cstate); else - cstate->copy_file = stdin; + edata->copy_file = stdin; } else { @@ -1856,8 +1864,8 @@ BeginCopyFrom(ParseState *pstate, if (cstate->is_program) { progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM; - cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_R); - if (cstate->copy_file == NULL) + edata->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_R); + if (edata->copy_file == NULL) ereport(ERROR, (errcode_for_file_access(), errmsg("could not execute command \"%s\": %m", @@ -1868,8 +1876,8 @@ BeginCopyFrom(ParseState *pstate, struct stat st; progress_vals[1] = PROGRESS_COPY_TYPE_FILE; - cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R); - if (cstate->copy_file == NULL) + edata->copy_file = AllocateFile(cstate->filename, PG_BINARY_R); + if (edata->copy_file == NULL) { /* copy errno because ereport subfunctions might change it */ int save_errno = errno; @@ -1883,7 +1891,7 @@ BeginCopyFrom(ParseState *pstate, "You may want a client-side facility such as psql's \\copy.") : 0)); } - if (fstat(fileno(cstate->copy_file), &st)) + if (fstat(fileno(edata->copy_file), &st)) ereport(ERROR, (errcode_for_file_access(), errmsg("could not stat file \"%s\": %m", @@ -1923,7 +1931,7 @@ EndCopyFrom(CopyFromState cstate) } else { - if (cstate->filename != NULL && FreeFile(cstate->copy_file)) + if (cstate->filename != NULL && FreeFile(cstate->edata->copy_file)) ereport(ERROR, (errcode_for_file_access(), errmsg("could not close file \"%s\": %m", @@ -1932,7 +1940,7 @@ EndCopyFrom(CopyFromState cstate) pgstat_progress_end_command(); - MemoryContextDelete(cstate->copycontext); + MemoryContextDelete(cstate->edata->copycontext); pfree(cstate); } @@ -1946,7 +1954,7 @@ ClosePipeFromProgram(CopyFromState cstate) Assert(cstate->is_program); - pclose_rc = ClosePipeStream(cstate->copy_file); + pclose_rc = ClosePipeStream(cstate->edata->copy_file); if (pclose_rc == -1) ereport(ERROR, (errcode_for_file_access(), @@ -1959,7 +1967,7 @@ ClosePipeFromProgram(CopyFromState cstate) * should not report that as an error. Otherwise, SIGPIPE indicates a * problem. */ - if (!cstate->raw_reached_eof && + if (!cstate->edata->raw_reached_eof && wait_result_is_signal(pclose_rc, SIGPIPE)) return; diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c index b1ae97b833d..595dc84b172 100644 --- a/src/backend/commands/copyfromparse.c +++ b/src/backend/commands/copyfromparse.c @@ -126,12 +126,12 @@ if (1) \ #define REFILL_LINEBUF \ if (1) \ { \ - if (input_buf_ptr > cstate->input_buf_index) \ + if (input_buf_ptr > edata->input_buf_index) \ { \ - appendBinaryStringInfo(&cstate->line_buf, \ - cstate->input_buf + cstate->input_buf_index, \ - input_buf_ptr - cstate->input_buf_index); \ - cstate->input_buf_index = input_buf_ptr; \ + appendBinaryStringInfo(&edata->line_buf, \ + edata->input_buf + edata->input_buf_index, \ + input_buf_ptr - edata->input_buf_index); \ + edata->input_buf_index = input_buf_ptr; \ } \ } else ((void) 0) @@ -180,8 +180,8 @@ ReceiveCopyBegin(CopyFromState cstate) for (i = 0; i < natts; i++) pq_sendint16(&buf, format); /* per-column formats */ pq_endmessage(&buf); - cstate->copy_src = COPY_FRONTEND; - cstate->fe_msgbuf = makeStringInfo(); + cstate->edata->copy_src = COPY_FRONTEND; + cstate->edata->fe_msgbuf = makeStringInfo(); /* We *must* flush here to ensure FE knows it can send. */ pq_flush(); } @@ -246,23 +246,23 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread) { int bytesread = 0; - switch (cstate->copy_src) + switch (cstate->edata->copy_src) { case COPY_FILE: - bytesread = fread(databuf, 1, maxread, cstate->copy_file); - if (ferror(cstate->copy_file)) + bytesread = fread(databuf, 1, maxread, cstate->edata->copy_file); + if (ferror(cstate->edata->copy_file)) ereport(ERROR, (errcode_for_file_access(), errmsg("could not read from COPY file: %m"))); if (bytesread == 0) - cstate->raw_reached_eof = true; + cstate->edata->raw_reached_eof = true; break; case COPY_FRONTEND: - while (maxread > 0 && bytesread < minread && !cstate->raw_reached_eof) + while (maxread > 0 && bytesread < minread && !cstate->edata->raw_reached_eof) { int avail; - while (cstate->fe_msgbuf->cursor >= cstate->fe_msgbuf->len) + while (cstate->edata->fe_msgbuf->cursor >= cstate->edata->fe_msgbuf->len) { /* Try to receive another message */ int mtype; @@ -297,7 +297,7 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread) break; } /* Now collect the message body */ - if (pq_getmessage(cstate->fe_msgbuf, maxmsglen)) + if (pq_getmessage(cstate->edata->fe_msgbuf, maxmsglen)) ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("unexpected EOF on client connection with an open transaction"))); @@ -309,13 +309,13 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread) break; case PqMsg_CopyDone: /* COPY IN correctly terminated by frontend */ - cstate->raw_reached_eof = true; + cstate->edata->raw_reached_eof = true; return bytesread; case PqMsg_CopyFail: ereport(ERROR, (errcode(ERRCODE_QUERY_CANCELED), errmsg("COPY from stdin failed: %s", - pq_getmsgstring(cstate->fe_msgbuf)))); + pq_getmsgstring(cstate->edata->fe_msgbuf)))); break; case PqMsg_Flush: case PqMsg_Sync: @@ -331,10 +331,10 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread) Assert(false); /* NOT REACHED */ } } - avail = cstate->fe_msgbuf->len - cstate->fe_msgbuf->cursor; + avail = cstate->edata->fe_msgbuf->len - cstate->edata->fe_msgbuf->cursor; if (avail > maxread) avail = maxread; - pq_copymsgbytes(cstate->fe_msgbuf, databuf, avail); + pq_copymsgbytes(cstate->edata->fe_msgbuf, databuf, avail); databuf = (void *) ((char *) databuf + avail); maxread -= avail; bytesread += avail; @@ -399,12 +399,14 @@ CopyGetInt16(CopyFromState cstate, int16 *val) static void CopyConvertBuf(CopyFromState cstate) { + CopyFromExecutionData *edata = cstate->edata; + /* * If the file and server encoding are the same, no encoding conversion is * required. However, we still need to verify that the input is valid for * the encoding. */ - if (!cstate->need_transcoding) + if (!edata->need_transcoding) { /* * When conversion is not required, input_buf and raw_buf are the @@ -412,8 +414,8 @@ CopyConvertBuf(CopyFromState cstate) * input_buf_len tracks how many of those bytes have already been * verified. */ - int preverifiedlen = cstate->input_buf_len; - int unverifiedlen = cstate->raw_buf_len - cstate->input_buf_len; + int preverifiedlen = edata->input_buf_len; + int unverifiedlen = edata->raw_buf_len - edata->input_buf_len; int nverified; if (unverifiedlen == 0) @@ -421,8 +423,8 @@ CopyConvertBuf(CopyFromState cstate) /* * If no more raw data is coming, report the EOF to the caller. */ - if (cstate->raw_reached_eof) - cstate->input_reached_eof = true; + if (edata->raw_reached_eof) + edata->input_reached_eof = true; return; } @@ -431,7 +433,7 @@ CopyConvertBuf(CopyFromState cstate) * previous round. */ nverified = pg_encoding_verifymbstr(cstate->file_encoding, - cstate->raw_buf + preverifiedlen, + edata->raw_buf + preverifiedlen, unverifiedlen); if (nverified == 0) { @@ -444,11 +446,11 @@ CopyConvertBuf(CopyFromState cstate) * least one character, and a failure to do so means that we've * hit an invalid byte sequence. */ - if (cstate->raw_reached_eof || unverifiedlen >= pg_encoding_max_length(cstate->file_encoding)) - cstate->input_reached_error = true; + if (edata->raw_reached_eof || unverifiedlen >= pg_encoding_max_length(cstate->file_encoding)) + edata->input_reached_error = true; return; } - cstate->input_buf_len += nverified; + edata->input_buf_len += nverified; } else { @@ -462,31 +464,31 @@ CopyConvertBuf(CopyFromState cstate) int dstlen; int convertedlen; - if (RAW_BUF_BYTES(cstate) == 0) + if (RAW_BUF_BYTES(edata) == 0) { /* * If no more raw data is coming, report the EOF to the caller. */ - if (cstate->raw_reached_eof) - cstate->input_reached_eof = true; + if (edata->raw_reached_eof) + edata->input_reached_eof = true; return; } /* * First, copy down any unprocessed data. */ - nbytes = INPUT_BUF_BYTES(cstate); - if (nbytes > 0 && cstate->input_buf_index > 0) - memmove(cstate->input_buf, cstate->input_buf + cstate->input_buf_index, + nbytes = INPUT_BUF_BYTES(edata); + if (nbytes > 0 && edata->input_buf_index > 0) + memmove(edata->input_buf, edata->input_buf + edata->input_buf_index, nbytes); - cstate->input_buf_index = 0; - cstate->input_buf_len = nbytes; - cstate->input_buf[nbytes] = '\0'; + edata->input_buf_index = 0; + edata->input_buf_len = nbytes; + edata->input_buf[nbytes] = '\0'; - src = (unsigned char *) cstate->raw_buf + cstate->raw_buf_index; - srclen = cstate->raw_buf_len - cstate->raw_buf_index; - dst = (unsigned char *) cstate->input_buf + cstate->input_buf_len; - dstlen = INPUT_BUF_SIZE - cstate->input_buf_len + 1; + src = (unsigned char *) edata->raw_buf + edata->raw_buf_index; + srclen = edata->raw_buf_len - edata->raw_buf_index; + dst = (unsigned char *) edata->input_buf + edata->input_buf_len; + dstlen = INPUT_BUF_SIZE - edata->input_buf_len + 1; /* * Do the conversion. This might stop short, if there is an invalid @@ -501,7 +503,7 @@ CopyConvertBuf(CopyFromState cstate) * after the end-of-input marker as long as it's valid for the * encoding, but that's harmless. */ - convertedlen = pg_do_encoding_conversion_buf(cstate->conversion_proc, + convertedlen = pg_do_encoding_conversion_buf(cstate->edata->conversion_proc, cstate->file_encoding, GetDatabaseEncoding(), src, srclen, @@ -517,12 +519,12 @@ CopyConvertBuf(CopyFromState cstate) * failure to do so must mean that we've hit a byte sequence * that's invalid. */ - if (cstate->raw_reached_eof || srclen >= MAX_CONVERSION_INPUT_LENGTH) - cstate->input_reached_error = true; + if (edata->raw_reached_eof || srclen >= MAX_CONVERSION_INPUT_LENGTH) + edata->input_reached_error = true; return; } - cstate->raw_buf_index += convertedlen; - cstate->input_buf_len += strlen((char *) dst); + edata->raw_buf_index += convertedlen; + edata->input_buf_len += strlen((char *) dst); } } @@ -532,18 +534,20 @@ CopyConvertBuf(CopyFromState cstate) static void CopyConversionError(CopyFromState cstate) { - Assert(cstate->raw_buf_len > 0); - Assert(cstate->input_reached_error); + CopyFromExecutionData *edata = cstate->edata; + + Assert(edata->raw_buf_len > 0); + Assert(edata->input_reached_error); - if (!cstate->need_transcoding) + if (!edata->need_transcoding) { /* * Everything up to input_buf_len was successfully verified, and * input_buf_len points to the invalid or incomplete character. */ report_invalid_encoding(cstate->file_encoding, - cstate->raw_buf + cstate->input_buf_len, - cstate->raw_buf_len - cstate->input_buf_len); + edata->raw_buf + edata->input_buf_len, + edata->raw_buf_len - edata->input_buf_len); } else { @@ -560,12 +564,12 @@ CopyConversionError(CopyFromState cstate) unsigned char *dst; int dstlen; - src = (unsigned char *) cstate->raw_buf + cstate->raw_buf_index; - srclen = cstate->raw_buf_len - cstate->raw_buf_index; - dst = (unsigned char *) cstate->input_buf + cstate->input_buf_len; - dstlen = INPUT_BUF_SIZE - cstate->input_buf_len + 1; + src = (unsigned char *) edata->raw_buf + edata->raw_buf_index; + srclen = edata->raw_buf_len - edata->raw_buf_index; + dst = (unsigned char *) edata->input_buf + edata->input_buf_len; + dstlen = INPUT_BUF_SIZE - edata->input_buf_len + 1; - (void) pg_do_encoding_conversion_buf(cstate->conversion_proc, + (void) pg_do_encoding_conversion_buf(cstate->edata->conversion_proc, cstate->file_encoding, GetDatabaseEncoding(), src, srclen, @@ -589,6 +593,7 @@ CopyConversionError(CopyFromState cstate) static void CopyLoadRawBuf(CopyFromState cstate) { + CopyFromExecutionData *edata = cstate->edata; int nbytes; int inbytes; @@ -596,45 +601,45 @@ CopyLoadRawBuf(CopyFromState cstate) * In text mode, if encoding conversion is not required, raw_buf and * input_buf point to the same buffer. Their len/index better agree, too. */ - if (cstate->raw_buf == cstate->input_buf) + if (edata->raw_buf == edata->input_buf) { - Assert(!cstate->need_transcoding); - Assert(cstate->raw_buf_index == cstate->input_buf_index); - Assert(cstate->input_buf_len <= cstate->raw_buf_len); + Assert(!edata->need_transcoding); + Assert(edata->raw_buf_index == edata->input_buf_index); + Assert(edata->input_buf_len <= edata->raw_buf_len); } /* * Copy down the unprocessed data if any. */ - nbytes = RAW_BUF_BYTES(cstate); - if (nbytes > 0 && cstate->raw_buf_index > 0) - memmove(cstate->raw_buf, cstate->raw_buf + cstate->raw_buf_index, + nbytes = RAW_BUF_BYTES(edata); + if (nbytes > 0 && edata->raw_buf_index > 0) + memmove(edata->raw_buf, edata->raw_buf + edata->raw_buf_index, nbytes); - cstate->raw_buf_len -= cstate->raw_buf_index; - cstate->raw_buf_index = 0; + edata->raw_buf_len -= edata->raw_buf_index; + edata->raw_buf_index = 0; /* * If raw_buf and input_buf are in fact the same buffer, adjust the * input_buf variables, too. */ - if (cstate->raw_buf == cstate->input_buf) + if (edata->raw_buf == edata->input_buf) { - cstate->input_buf_len -= cstate->input_buf_index; - cstate->input_buf_index = 0; + edata->input_buf_len -= edata->input_buf_index; + edata->input_buf_index = 0; } /* Load more data */ - inbytes = CopyGetData(cstate, cstate->raw_buf + cstate->raw_buf_len, - 1, RAW_BUF_SIZE - cstate->raw_buf_len); + inbytes = CopyGetData(cstate, edata->raw_buf + edata->raw_buf_len, + 1, RAW_BUF_SIZE - edata->raw_buf_len); nbytes += inbytes; - cstate->raw_buf[nbytes] = '\0'; - cstate->raw_buf_len = nbytes; + edata->raw_buf[nbytes] = '\0'; + edata->raw_buf_len = nbytes; - cstate->bytes_processed += inbytes; - pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->bytes_processed); + edata->bytes_processed += inbytes; + pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, edata->bytes_processed); if (inbytes == 0) - cstate->raw_reached_eof = true; + edata->raw_reached_eof = true; } /* @@ -643,24 +648,25 @@ CopyLoadRawBuf(CopyFromState cstate) * On return, at least one more input character is loaded into * input_buf, or input_reached_eof is set. * - * If INPUT_BUF_BYTES(cstate) > 0, the unprocessed bytes are moved to the start + * If INPUT_BUF_BYTES(edata) > 0, the unprocessed bytes are moved to the start * of the buffer and then we load more data after that. */ static void CopyLoadInputBuf(CopyFromState cstate) { - int nbytes = INPUT_BUF_BYTES(cstate); + CopyFromExecutionData *edata = cstate->edata; + int nbytes = INPUT_BUF_BYTES(edata); /* * The caller has updated input_buf_index to indicate how much of the * input has been consumed and isn't needed anymore. If input_buf is the * same physical area as raw_buf, update raw_buf_index accordingly. */ - if (cstate->raw_buf == cstate->input_buf) + if (edata->raw_buf == edata->input_buf) { - Assert(!cstate->need_transcoding); - Assert(cstate->input_buf_index >= cstate->raw_buf_index); - cstate->raw_buf_index = cstate->input_buf_index; + Assert(!edata->need_transcoding); + Assert(edata->input_buf_index >= edata->raw_buf_index); + edata->raw_buf_index = edata->input_buf_index; } for (;;) @@ -669,7 +675,7 @@ CopyLoadInputBuf(CopyFromState cstate) CopyConvertBuf(cstate); /* If we now have some more input bytes ready, return them */ - if (INPUT_BUF_BYTES(cstate) > nbytes) + if (INPUT_BUF_BYTES(edata) > nbytes) return; /* @@ -677,15 +683,15 @@ CopyLoadInputBuf(CopyFromState cstate) * multi-byte character but there is no more raw input data, report * conversion error. */ - if (cstate->input_reached_error) + if (edata->input_reached_error) CopyConversionError(cstate); /* no more input, and everything has been converted */ - if (cstate->input_reached_eof) + if (edata->input_reached_eof) break; /* Try to load more raw data */ - Assert(!cstate->raw_reached_eof); + Assert(!edata->raw_reached_eof); CopyLoadRawBuf(cstate); } } @@ -700,13 +706,14 @@ CopyLoadInputBuf(CopyFromState cstate) static int CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes) { + CopyFromExecutionData *edata = cstate->edata; int copied_bytes = 0; - if (RAW_BUF_BYTES(cstate) >= nbytes) + if (RAW_BUF_BYTES(edata) >= nbytes) { /* Enough bytes are present in the buffer. */ - memcpy(dest, cstate->raw_buf + cstate->raw_buf_index, nbytes); - cstate->raw_buf_index += nbytes; + memcpy(dest, edata->raw_buf + edata->raw_buf_index, nbytes); + edata->raw_buf_index += nbytes; copied_bytes = nbytes; } else @@ -720,17 +727,17 @@ CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes) int copy_bytes; /* Load more data if buffer is empty. */ - if (RAW_BUF_BYTES(cstate) == 0) + if (RAW_BUF_BYTES(edata) == 0) { CopyLoadRawBuf(cstate); - if (cstate->raw_reached_eof) + if (edata->raw_reached_eof) break; /* EOF */ } /* Transfer some bytes. */ - copy_bytes = Min(nbytes - copied_bytes, RAW_BUF_BYTES(cstate)); - memcpy(dest, cstate->raw_buf + cstate->raw_buf_index, copy_bytes); - cstate->raw_buf_index += copy_bytes; + copy_bytes = Min(nbytes - copied_bytes, RAW_BUF_BYTES(edata)); + memcpy(dest, edata->raw_buf + edata->raw_buf_index, copy_bytes); + edata->raw_buf_index += copy_bytes; dest += copy_bytes; copied_bytes += copy_bytes; } while (copied_bytes < nbytes); @@ -770,6 +777,7 @@ NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields) static pg_attribute_always_inline bool NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields, bool is_csv) { + CopyFromExecutionData *edata = cstate->edata; int fldct; bool done = false; @@ -777,7 +785,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields Assert(!cstate->opts.binary); /* on input check that the header line is correct if needed */ - if (cstate->cur_lineno == 0 && cstate->opts.header_line != COPY_HEADER_FALSE) + if (edata->cur_lineno == 0 && cstate->opts.header_line != COPY_HEADER_FALSE) { ListCell *cur; TupleDesc tupDesc; @@ -791,7 +799,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields for (int i = 0; i < lines_to_skip; i++) { - cstate->cur_lineno++; + edata->cur_lineno++; if ((done = CopyReadLine(cstate, is_csv))) break; } @@ -818,9 +826,9 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields char *colName; Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1); - Assert(fldnum < cstate->max_fields); + Assert(fldnum < edata->max_fields); - colName = cstate->raw_fields[fldnum++]; + colName = edata->raw_fields[fldnum++]; if (colName == NULL) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), @@ -841,7 +849,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields return false; } - cstate->cur_lineno++; + edata->cur_lineno++; /* Actually read the line into memory here */ done = CopyReadLine(cstate, is_csv); @@ -851,7 +859,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields * 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) + if (done && edata->line_buf.len == 0) return false; /* Parse the line into de-escaped field values */ @@ -860,7 +868,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields else fldct = CopyReadAttributesText(cstate); - *fields = cstate->raw_fields; + *fields = edata->raw_fields; *nfields = fldct; return true; } @@ -882,10 +890,10 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext, { TupleDesc tupDesc; AttrNumber num_phys_attrs, - num_defaults = cstate->num_defaults; + num_defaults = cstate->edata->num_defaults; int i; - int *defmap = cstate->defmap; - ExprState **defexprs = cstate->defexprs; + int *defmap = cstate->edata->defmap; + ExprState **defexprs = cstate->edata->defexprs; tupDesc = RelationGetDescr(cstate->rel); num_phys_attrs = tupDesc->natts; @@ -893,7 +901,7 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext, /* Initialize all values for row to NULL */ MemSet(values, 0, num_phys_attrs * sizeof(Datum)); MemSet(nulls, true, num_phys_attrs * sizeof(bool)); - MemSet(cstate->defaults, false, num_phys_attrs * sizeof(bool)); + MemSet(cstate->edata->defaults, false, num_phys_attrs * sizeof(bool)); /* Get one row from source */ if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls)) @@ -946,11 +954,12 @@ static pg_attribute_always_inline bool CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values, bool *nulls, bool is_csv) { + CopyFromExecutionData *edata = cstate->edata; TupleDesc tupDesc; AttrNumber attr_count; FmgrInfo *in_functions = cstate->in_functions; Oid *typioparams = cstate->typioparams; - ExprState **defexprs = cstate->defexprs; + ExprState **defexprs = edata->defexprs; char **field_strings; ListCell *cur; int fldct; @@ -986,8 +995,8 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, NameStr(att->attname)))); string = field_strings[fieldno++]; - if (cstate->convert_select_flags && - !cstate->convert_select_flags[m]) + if (edata->convert_select_flags && + !edata->convert_select_flags[m]) { /* ignore input field, leaving column as NULL */ continue; @@ -1017,13 +1026,13 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, } } - cstate->cur_attname = NameStr(att->attname); - cstate->cur_attval = string; + edata->cur_attname = NameStr(att->attname); + edata->cur_attval = string; if (string != NULL) nulls[m] = false; - if (cstate->defaults[m]) + if (edata->defaults[m]) { /* We must have switched into the per-tuple memory context */ Assert(econtext != NULL); @@ -1039,12 +1048,12 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, string, typioparams[m], att->atttypmod, - (Node *) cstate->escontext, + (Node *) edata->escontext, &values[m])) { Assert(cstate->opts.on_error != COPY_ON_ERROR_STOP); - cstate->num_errors++; + edata->num_errors++; if (cstate->opts.log_verbosity == COPY_LOG_VERBOSITY_VERBOSE) { @@ -1053,36 +1062,36 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, * notice message, we suppress error context information other * than the relation name. */ - Assert(!cstate->relname_only); - cstate->relname_only = true; + Assert(!edata->relname_only); + edata->relname_only = true; - if (cstate->cur_attval) + if (edata->cur_attval) { char *attval; - attval = CopyLimitPrintoutLength(cstate->cur_attval); + attval = CopyLimitPrintoutLength(edata->cur_attval); ereport(NOTICE, errmsg("skipping row due to data type incompatibility at line %" PRIu64 " for column \"%s\": \"%s\"", - cstate->cur_lineno, - cstate->cur_attname, + edata->cur_lineno, + edata->cur_attname, attval)); pfree(attval); } else ereport(NOTICE, errmsg("skipping row due to data type incompatibility at line %" PRIu64 " for column \"%s\": null input", - cstate->cur_lineno, - cstate->cur_attname)); + edata->cur_lineno, + edata->cur_attname)); /* reset relname_only */ - cstate->relname_only = false; + edata->relname_only = false; } return true; } - cstate->cur_attname = NULL; - cstate->cur_attval = NULL; + edata->cur_attname = NULL; + edata->cur_attval = NULL; } Assert(fieldno == attr_count); @@ -1105,7 +1114,7 @@ CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values, tupDesc = RelationGetDescr(cstate->rel); attr_count = list_length(cstate->attnumlist); - cstate->cur_lineno++; + cstate->edata->cur_lineno++; if (!CopyGetInt16(cstate, &fld_count)) { @@ -1144,13 +1153,13 @@ CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values, int m = attnum - 1; Form_pg_attribute att = TupleDescAttr(tupDesc, m); - cstate->cur_attname = NameStr(att->attname); + cstate->edata->cur_attname = NameStr(att->attname); values[m] = CopyReadBinaryAttribute(cstate, &in_functions[m], typioparams[m], att->atttypmod, &nulls[m]); - cstate->cur_attname = NULL; + cstate->edata->cur_attname = NULL; } return true; @@ -1166,10 +1175,11 @@ CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values, static bool CopyReadLine(CopyFromState cstate, bool is_csv) { + CopyFromExecutionData *edata = cstate->edata; bool result; - resetStringInfo(&cstate->line_buf); - cstate->line_buf_valid = false; + resetStringInfo(&edata->line_buf); + edata->line_buf_valid = false; /* Parse data and transfer into line_buf */ result = CopyReadLineText(cstate, is_csv); @@ -1181,19 +1191,19 @@ CopyReadLine(CopyFromState cstate, bool is_csv) * after \. up to the protocol end of copy data. (XXX maybe better * not to treat \. as special?) */ - if (cstate->copy_src == COPY_FRONTEND) + if (edata->copy_src == COPY_FRONTEND) { int inbytes; do { - inbytes = CopyGetData(cstate, cstate->input_buf, + inbytes = CopyGetData(cstate, edata->input_buf, 1, INPUT_BUF_SIZE); } while (inbytes > 0); - cstate->input_buf_index = 0; - cstate->input_buf_len = 0; - cstate->raw_buf_index = 0; - cstate->raw_buf_len = 0; + edata->input_buf_index = 0; + edata->input_buf_len = 0; + edata->raw_buf_index = 0; + edata->raw_buf_len = 0; } } else @@ -1202,26 +1212,26 @@ CopyReadLine(CopyFromState cstate, bool is_csv) * If we didn't hit EOF, then we must have transferred the EOL marker * to line_buf along with the data. Get rid of it. */ - switch (cstate->eol_type) + switch (edata->eol_type) { case EOL_NL: - Assert(cstate->line_buf.len >= 1); - Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\n'); - cstate->line_buf.len--; - cstate->line_buf.data[cstate->line_buf.len] = '\0'; + Assert(edata->line_buf.len >= 1); + Assert(edata->line_buf.data[edata->line_buf.len - 1] == '\n'); + edata->line_buf.len--; + edata->line_buf.data[edata->line_buf.len] = '\0'; break; case EOL_CR: - Assert(cstate->line_buf.len >= 1); - Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\r'); - cstate->line_buf.len--; - cstate->line_buf.data[cstate->line_buf.len] = '\0'; + Assert(edata->line_buf.len >= 1); + Assert(edata->line_buf.data[edata->line_buf.len - 1] == '\r'); + edata->line_buf.len--; + edata->line_buf.data[edata->line_buf.len] = '\0'; break; case EOL_CRNL: - Assert(cstate->line_buf.len >= 2); - Assert(cstate->line_buf.data[cstate->line_buf.len - 2] == '\r'); - Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\n'); - cstate->line_buf.len -= 2; - cstate->line_buf.data[cstate->line_buf.len] = '\0'; + Assert(edata->line_buf.len >= 2); + Assert(edata->line_buf.data[edata->line_buf.len - 2] == '\r'); + Assert(edata->line_buf.data[edata->line_buf.len - 1] == '\n'); + edata->line_buf.len -= 2; + edata->line_buf.data[edata->line_buf.len] = '\0'; break; case EOL_UNKNOWN: /* shouldn't get here */ @@ -1231,7 +1241,7 @@ CopyReadLine(CopyFromState cstate, bool is_csv) } /* Now it's safe to use the buffer in error messages */ - cstate->line_buf_valid = true; + edata->line_buf_valid = true; return result; } @@ -1242,6 +1252,7 @@ CopyReadLine(CopyFromState cstate, bool is_csv) static bool CopyReadLineText(CopyFromState cstate, bool is_csv) { + CopyFromExecutionData *edata = cstate->edata; char *copy_input_buf; int input_buf_ptr; int copy_buf_len; @@ -1289,9 +1300,9 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) * For a little extra speed within the loop, we copy input_buf and * input_buf_len into local variables. */ - copy_input_buf = cstate->input_buf; - input_buf_ptr = cstate->input_buf_index; - copy_buf_len = cstate->input_buf_len; + copy_input_buf = edata->input_buf; + input_buf_ptr = edata->input_buf_index; + copy_buf_len = edata->input_buf_len; for (;;) { @@ -1312,15 +1323,15 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) CopyLoadInputBuf(cstate); /* update our local variables */ - hit_eof = cstate->input_reached_eof; - input_buf_ptr = cstate->input_buf_index; - copy_buf_len = cstate->input_buf_len; + hit_eof = edata->input_reached_eof; + input_buf_ptr = edata->input_buf_index; + copy_buf_len = edata->input_buf_len; /* * If we are completely out of data, break out of the loop, * reporting EOF. */ - if (INPUT_BUF_BYTES(cstate) <= 0) + if (INPUT_BUF_BYTES(edata) <= 0) { result = true; break; @@ -1366,16 +1377,16 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) * best we can do. (XXX it's arguable whether we should do this * at all --- is cur_lineno a physical or logical count?) */ - if (in_quote && c == (cstate->eol_type == EOL_NL ? '\n' : '\r')) - cstate->cur_lineno++; + if (in_quote && c == (edata->eol_type == EOL_NL ? '\n' : '\r')) + edata->cur_lineno++; } /* Process \r */ if (c == '\r' && (!is_csv || !in_quote)) { /* Check for \r\n on first line, _and_ handle \r\n. */ - if (cstate->eol_type == EOL_UNKNOWN || - cstate->eol_type == EOL_CRNL) + if (edata->eol_type == EOL_UNKNOWN || + edata->eol_type == EOL_CRNL) { /* * If need more data, go back to loop top to load it. @@ -1391,12 +1402,12 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) if (c == '\n') { input_buf_ptr++; /* eat newline */ - cstate->eol_type = EOL_CRNL; /* in case not set yet */ + edata->eol_type = EOL_CRNL; /* in case not set yet */ } else { /* found \r, but no \n */ - if (cstate->eol_type == EOL_CRNL) + if (edata->eol_type == EOL_CRNL) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), !is_csv ? @@ -1410,10 +1421,10 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) * if we got here, it is the first line and we didn't find * \n, so don't consume the peeked character */ - cstate->eol_type = EOL_CR; + edata->eol_type = EOL_CR; } } - else if (cstate->eol_type == EOL_NL) + else if (edata->eol_type == EOL_NL) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), !is_csv ? @@ -1429,7 +1440,7 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) /* Process \n */ if (c == '\n' && (!is_csv || !in_quote)) { - if (cstate->eol_type == EOL_CR || cstate->eol_type == EOL_CRNL) + if (edata->eol_type == EOL_CR || edata->eol_type == EOL_CRNL) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), !is_csv ? @@ -1438,7 +1449,7 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) !is_csv ? errhint("Use \"\\n\" to represent newline.") : errhint("Use quoted CSV field to represent newline."))); - cstate->eol_type = EOL_NL; /* in case not set yet */ + edata->eol_type = EOL_NL; /* in case not set yet */ /* If reach here, we have found the line terminator */ break; } @@ -1465,7 +1476,7 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) if (c2 == '.') { input_buf_ptr++; /* consume the '.' */ - if (cstate->eol_type == EOL_CRNL) + if (edata->eol_type == EOL_CRNL) { /* Get the next character */ IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(0); @@ -1492,9 +1503,9 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), errmsg("end-of-copy marker is not alone on its line"))); - if ((cstate->eol_type == EOL_NL && c2 != '\n') || - (cstate->eol_type == EOL_CRNL && c2 != '\n') || - (cstate->eol_type == EOL_CR && c2 != '\r')) + if ((edata->eol_type == EOL_NL && c2 != '\n') || + (edata->eol_type == EOL_CRNL && c2 != '\n') || + (edata->eol_type == EOL_CR && c2 != '\r')) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), errmsg("end-of-copy marker does not match previous newline style"))); @@ -1502,8 +1513,8 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) /* * If there is any data on this line before the \., complain. */ - if (cstate->line_buf.len > 0 || - prev_raw_ptr > cstate->input_buf_index) + if (edata->line_buf.len > 0 || + prev_raw_ptr > edata->input_buf_index) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), errmsg("end-of-copy marker is not alone on its line"))); @@ -1511,7 +1522,7 @@ CopyReadLineText(CopyFromState cstate, bool is_csv) /* * Discard the \. and newline, then report EOF. */ - cstate->input_buf_index = input_buf_ptr; + edata->input_buf_index = input_buf_ptr; result = true; /* report EOF */ break; } @@ -1572,6 +1583,7 @@ GetDecimalFromHex(char hex) static int CopyReadAttributesText(CopyFromState cstate) { + CopyFromExecutionData *edata = cstate->edata; char delimc = cstate->opts.delim[0]; int fieldno; char *output_ptr; @@ -1582,31 +1594,31 @@ CopyReadAttributesText(CopyFromState cstate) * We need a special case for zero-column tables: check that the input * line is empty, and return. */ - if (cstate->max_fields <= 0) + if (edata->max_fields <= 0) { - if (cstate->line_buf.len != 0) + if (edata->line_buf.len != 0) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), errmsg("extra data after last expected column"))); return 0; } - resetStringInfo(&cstate->attribute_buf); + resetStringInfo(&edata->attribute_buf); /* * The de-escaped attributes will certainly not be longer than the input * data line, so we can just force attribute_buf to be large enough and * then transfer data without any checks for enough space. We need to do * it this way because enlarging attribute_buf mid-stream would invalidate - * pointers already stored into cstate->raw_fields[]. + * pointers already stored into edata->raw_fields[]. */ - if (cstate->attribute_buf.maxlen <= cstate->line_buf.len) - enlargeStringInfo(&cstate->attribute_buf, cstate->line_buf.len); - output_ptr = cstate->attribute_buf.data; + if (edata->attribute_buf.maxlen <= edata->line_buf.len) + enlargeStringInfo(&edata->attribute_buf, edata->line_buf.len); + output_ptr = edata->attribute_buf.data; /* set pointer variables for loop */ - cur_ptr = cstate->line_buf.data; - line_end_ptr = cstate->line_buf.data + cstate->line_buf.len; + cur_ptr = edata->line_buf.data; + line_end_ptr = edata->line_buf.data + edata->line_buf.len; /* Outer loop iterates over fields */ fieldno = 0; @@ -1619,16 +1631,16 @@ CopyReadAttributesText(CopyFromState cstate) bool saw_non_ascii = false; /* Make sure there is enough space for the next value */ - if (fieldno >= cstate->max_fields) + if (fieldno >= edata->max_fields) { - cstate->max_fields *= 2; - cstate->raw_fields = - repalloc(cstate->raw_fields, cstate->max_fields * sizeof(char *)); + edata->max_fields *= 2; + edata->raw_fields = + repalloc(edata->raw_fields, edata->max_fields * sizeof(char *)); } /* Remember start of field on both input and output sides */ start_ptr = cur_ptr; - cstate->raw_fields[fieldno] = output_ptr; + edata->raw_fields[fieldno] = output_ptr; /* * Scan data for field. @@ -1757,7 +1769,7 @@ CopyReadAttributesText(CopyFromState cstate) input_len = end_ptr - start_ptr; if (input_len == cstate->opts.null_print_len && strncmp(start_ptr, cstate->opts.null_print, input_len) == 0) - cstate->raw_fields[fieldno] = NULL; + edata->raw_fields[fieldno] = NULL; /* Check whether raw input matched default marker */ else if (fieldno < list_length(cstate->attnumlist) && cstate->opts.default_print && @@ -1767,10 +1779,10 @@ CopyReadAttributesText(CopyFromState cstate) /* fieldno is 0-indexed and attnum is 1-indexed */ int m = list_nth_int(cstate->attnumlist, fieldno) - 1; - if (cstate->defexprs[m] != NULL) + if (edata->defexprs[m] != NULL) { /* defaults contain entries for all physical attributes */ - cstate->defaults[m] = true; + edata->defaults[m] = true; } else { @@ -1794,7 +1806,7 @@ CopyReadAttributesText(CopyFromState cstate) */ if (saw_non_ascii) { - char *fld = cstate->raw_fields[fieldno]; + char *fld = edata->raw_fields[fieldno]; pg_verifymbstr(fld, output_ptr - fld, false); } @@ -1812,7 +1824,7 @@ CopyReadAttributesText(CopyFromState cstate) /* Clean up state of attribute_buf */ output_ptr--; Assert(*output_ptr == '\0'); - cstate->attribute_buf.len = (output_ptr - cstate->attribute_buf.data); + edata->attribute_buf.len = (output_ptr - edata->attribute_buf.data); return fieldno; } @@ -1826,6 +1838,7 @@ CopyReadAttributesText(CopyFromState cstate) static int CopyReadAttributesCSV(CopyFromState cstate) { + CopyFromExecutionData *edata = cstate->edata; char delimc = cstate->opts.delim[0]; char quotec = cstate->opts.quote[0]; char escapec = cstate->opts.escape[0]; @@ -1838,31 +1851,31 @@ CopyReadAttributesCSV(CopyFromState cstate) * We need a special case for zero-column tables: check that the input * line is empty, and return. */ - if (cstate->max_fields <= 0) + if (edata->max_fields <= 0) { - if (cstate->line_buf.len != 0) + if (edata->line_buf.len != 0) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), errmsg("extra data after last expected column"))); return 0; } - resetStringInfo(&cstate->attribute_buf); + resetStringInfo(&edata->attribute_buf); /* * The de-escaped attributes will certainly not be longer than the input * data line, so we can just force attribute_buf to be large enough and * then transfer data without any checks for enough space. We need to do * it this way because enlarging attribute_buf mid-stream would invalidate - * pointers already stored into cstate->raw_fields[]. + * pointers already stored into edata->raw_fields[]. */ - if (cstate->attribute_buf.maxlen <= cstate->line_buf.len) - enlargeStringInfo(&cstate->attribute_buf, cstate->line_buf.len); - output_ptr = cstate->attribute_buf.data; + if (edata->attribute_buf.maxlen <= edata->line_buf.len) + enlargeStringInfo(&edata->attribute_buf, edata->line_buf.len); + output_ptr = edata->attribute_buf.data; /* set pointer variables for loop */ - cur_ptr = cstate->line_buf.data; - line_end_ptr = cstate->line_buf.data + cstate->line_buf.len; + cur_ptr = edata->line_buf.data; + line_end_ptr = edata->line_buf.data + edata->line_buf.len; /* Outer loop iterates over fields */ fieldno = 0; @@ -1875,16 +1888,16 @@ CopyReadAttributesCSV(CopyFromState cstate) int input_len; /* Make sure there is enough space for the next value */ - if (fieldno >= cstate->max_fields) + if (fieldno >= edata->max_fields) { - cstate->max_fields *= 2; - cstate->raw_fields = - repalloc(cstate->raw_fields, cstate->max_fields * sizeof(char *)); + edata->max_fields *= 2; + edata->raw_fields = + repalloc(edata->raw_fields, edata->max_fields * sizeof(char *)); } /* Remember start of field on both input and output sides */ start_ptr = cur_ptr; - cstate->raw_fields[fieldno] = output_ptr; + edata->raw_fields[fieldno] = output_ptr; /* * Scan data for field, @@ -1972,7 +1985,7 @@ endfield: input_len = end_ptr - start_ptr; if (!saw_quote && input_len == cstate->opts.null_print_len && strncmp(start_ptr, cstate->opts.null_print, input_len) == 0) - cstate->raw_fields[fieldno] = NULL; + edata->raw_fields[fieldno] = NULL; /* Check whether raw input matched default marker */ else if (fieldno < list_length(cstate->attnumlist) && cstate->opts.default_print && @@ -1982,10 +1995,10 @@ endfield: /* fieldno is 0-index and attnum is 1-index */ int m = list_nth_int(cstate->attnumlist, fieldno) - 1; - if (cstate->defexprs[m] != NULL) + if (edata->defexprs[m] != NULL) { /* defaults contain entries for all physical attributes */ - cstate->defaults[m] = true; + edata->defaults[m] = true; } else { @@ -2009,7 +2022,7 @@ endfield: /* Clean up state of attribute_buf */ output_ptr--; Assert(*output_ptr == '\0'); - cstate->attribute_buf.len = (output_ptr - cstate->attribute_buf.data); + edata->attribute_buf.len = (output_ptr - edata->attribute_buf.data); return fieldno; } @@ -2023,6 +2036,7 @@ CopyReadBinaryAttribute(CopyFromState cstate, FmgrInfo *flinfo, Oid typioparam, int32 typmod, bool *isnull) { + CopyFromExecutionData *edata = cstate->edata; int32 fld_size; Datum result; @@ -2041,24 +2055,24 @@ CopyReadBinaryAttribute(CopyFromState cstate, FmgrInfo *flinfo, errmsg("invalid field size"))); /* reset attribute_buf to empty, and load raw data in it */ - resetStringInfo(&cstate->attribute_buf); + resetStringInfo(&edata->attribute_buf); - enlargeStringInfo(&cstate->attribute_buf, fld_size); - if (CopyReadBinaryData(cstate, cstate->attribute_buf.data, + enlargeStringInfo(&edata->attribute_buf, fld_size); + if (CopyReadBinaryData(cstate, edata->attribute_buf.data, fld_size) != fld_size) ereport(ERROR, (errcode(ERRCODE_BAD_COPY_FILE_FORMAT), errmsg("unexpected EOF in COPY data"))); - cstate->attribute_buf.len = fld_size; - cstate->attribute_buf.data[fld_size] = '\0'; + edata->attribute_buf.len = fld_size; + edata->attribute_buf.data[fld_size] = '\0'; /* Call the column type's binary input converter */ - result = ReceiveFunctionCall(flinfo, &cstate->attribute_buf, + result = ReceiveFunctionCall(flinfo, &edata->attribute_buf, typioparam, typmod); /* Trouble if it didn't eat the whole buffer */ - if (cstate->attribute_buf.cursor != cstate->attribute_buf.len) + if (edata->attribute_buf.cursor != edata->attribute_buf.len) ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION), errmsg("incorrect binary data format"))); diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c index 67b94b91cae..e90fab6d958 100644 --- a/src/backend/commands/copyto.c +++ b/src/backend/commands/copyto.c @@ -62,40 +62,24 @@ typedef enum CopyDest * invoke pg_encoding_mblen() to skip over them. encoding_embeds_ascii is true * when we have to do it the hard way. */ -typedef struct CopyToStateData +typedef struct CopyToExecutionData { - /* format-specific routines */ - const CopyToRoutine *routine; - /* low-level state data */ CopyDest copy_dest; /* type of copy source/destination */ FILE *copy_file; /* used if copy_dest == COPY_FILE */ StringInfo fe_msgbuf; /* used for all dests during COPY TO */ - int file_encoding; /* file or remote side's character encoding */ bool need_transcoding; /* file encoding diff from server? */ bool encoding_embeds_ascii; /* ASCII can be non-first byte? */ - /* parameters from the COPY command */ - Relation rel; /* relation to copy to */ - QueryDesc *queryDesc; /* executable query to copy from */ - List *attnumlist; /* integer list of attnums to copy */ - char *filename; /* filename, or NULL for STDOUT */ - bool is_program; /* is 'filename' a program to popen? */ - copy_data_dest_cb data_dest_cb; /* function for writing data */ - - CopyFormatOptions opts; - Node *whereClause; /* WHERE condition (or NULL) */ - /* * Working state */ MemoryContext copycontext; /* per-copy execution context */ - FmgrInfo *out_functions; /* lookup info for output functions */ MemoryContext rowcontext; /* per-row evaluation context */ uint64 bytes_processed; /* number of bytes processed so far */ -} CopyToStateData; +} CopyToExecutionData; /* DestReceiver for COPY (query) TO */ typedef struct @@ -193,7 +177,7 @@ CopyToTextLikeStart(CopyToState cstate, TupleDesc tupDesc) * For non-binary copy, we need to convert null_print to file encoding, * because it will be sent directly with CopySendString. */ - if (cstate->need_transcoding) + if (cstate->edata->need_transcoding) cstate->opts.null_print_client = pg_server_to_any(cstate->opts.null_print, cstate->opts.null_print_len, cstate->file_encoding); @@ -401,14 +385,14 @@ SendCopyBegin(CopyToState cstate) for (i = 0; i < natts; i++) pq_sendint16(&buf, format); /* per-column formats */ pq_endmessage(&buf); - cstate->copy_dest = COPY_FRONTEND; + cstate->edata->copy_dest = COPY_FRONTEND; } static void SendCopyEnd(CopyToState cstate) { /* Shouldn't have any unsent data */ - Assert(cstate->fe_msgbuf->len == 0); + Assert(cstate->edata->fe_msgbuf->len == 0); /* Send Copy Done message */ pq_putemptymessage(PqMsg_CopyDone); } @@ -426,32 +410,32 @@ SendCopyEnd(CopyToState cstate) static void CopySendData(CopyToState cstate, const void *databuf, int datasize) { - appendBinaryStringInfo(cstate->fe_msgbuf, databuf, datasize); + appendBinaryStringInfo(cstate->edata->fe_msgbuf, databuf, datasize); } static void CopySendString(CopyToState cstate, const char *str) { - appendBinaryStringInfo(cstate->fe_msgbuf, str, strlen(str)); + appendBinaryStringInfo(cstate->edata->fe_msgbuf, str, strlen(str)); } static void CopySendChar(CopyToState cstate, char c) { - appendStringInfoCharMacro(cstate->fe_msgbuf, c); + appendStringInfoCharMacro(cstate->edata->fe_msgbuf, c); } static void CopySendEndOfRow(CopyToState cstate) { - StringInfo fe_msgbuf = cstate->fe_msgbuf; + StringInfo fe_msgbuf = cstate->edata->fe_msgbuf; - switch (cstate->copy_dest) + switch (cstate->edata->copy_dest) { case COPY_FILE: if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1, - cstate->copy_file) != 1 || - ferror(cstate->copy_file)) + cstate->edata->copy_file) != 1 || + ferror(cstate->edata->copy_file)) { if (cstate->is_program) { @@ -492,8 +476,8 @@ CopySendEndOfRow(CopyToState cstate) } /* Update the progress */ - cstate->bytes_processed += fe_msgbuf->len; - pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->bytes_processed); + cstate->edata->bytes_processed += fe_msgbuf->len; + pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->edata->bytes_processed); resetStringInfo(fe_msgbuf); } @@ -505,7 +489,7 @@ CopySendEndOfRow(CopyToState cstate) static inline void CopySendTextLikeEndOfRow(CopyToState cstate) { - switch (cstate->copy_dest) + switch (cstate->edata->copy_dest) { case COPY_FILE: /* Default line termination depends on platform */ @@ -565,7 +549,7 @@ ClosePipeToProgram(CopyToState cstate) Assert(cstate->is_program); - pclose_rc = ClosePipeStream(cstate->copy_file); + pclose_rc = ClosePipeStream(cstate->edata->copy_file); if (pclose_rc == -1) ereport(ERROR, (errcode_for_file_access(), @@ -592,7 +576,7 @@ EndCopy(CopyToState cstate) } else { - if (cstate->filename != NULL && FreeFile(cstate->copy_file)) + if (cstate->filename != NULL && FreeFile(cstate->edata->copy_file)) ereport(ERROR, (errcode_for_file_access(), errmsg("could not close file \"%s\": %m", @@ -601,7 +585,7 @@ EndCopy(CopyToState cstate) pgstat_progress_end_command(); - MemoryContextDelete(cstate->copycontext); + MemoryContextDelete(cstate->edata->copycontext); pfree(cstate); } @@ -688,16 +672,17 @@ BeginCopyTo(ParseState *pstate, /* Allocate workspace and zero all fields */ cstate = (CopyToStateData *) palloc0(sizeof(CopyToStateData)); + cstate->edata = (CopyToExecutionData *) palloc0(sizeof(CopyToExecutionData)); /* * 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); + cstate->edata->copycontext = AllocSetContextCreate(CurrentMemoryContext, + "COPY", + ALLOCSET_DEFAULT_SIZES); - oldcontext = MemoryContextSwitchTo(cstate->copycontext); + oldcontext = MemoryContextSwitchTo(cstate->edata->copycontext); /* Extract options from the statement node tree */ ProcessCopyOptions(pstate, &cstate->opts, false /* is_from */ , options); @@ -895,19 +880,19 @@ BeginCopyTo(ParseState *pstate, */ if (cstate->file_encoding == GetDatabaseEncoding() || cstate->file_encoding == PG_SQL_ASCII) - cstate->need_transcoding = false; + cstate->edata->need_transcoding = false; else - cstate->need_transcoding = true; + cstate->edata->need_transcoding = true; /* See Multibyte encoding comment above */ - cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding); + cstate->edata->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding); - cstate->copy_dest = COPY_FILE; /* default */ + cstate->edata->copy_dest = COPY_FILE; /* default */ if (data_dest_cb) { progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK; - cstate->copy_dest = COPY_CALLBACK; + cstate->edata->copy_dest = COPY_CALLBACK; cstate->data_dest_cb = data_dest_cb; } else if (pipe) @@ -916,7 +901,7 @@ BeginCopyTo(ParseState *pstate, Assert(!is_program); /* the grammar does not allow this */ if (whereToSendOutput != DestRemote) - cstate->copy_file = stdout; + cstate->edata->copy_file = stdout; } else { @@ -926,8 +911,8 @@ BeginCopyTo(ParseState *pstate, if (is_program) { progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM; - cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_W); - if (cstate->copy_file == NULL) + cstate->edata->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_W); + if (cstate->edata->copy_file == NULL) ereport(ERROR, (errcode_for_file_access(), errmsg("could not execute command \"%s\": %m", @@ -952,14 +937,14 @@ BeginCopyTo(ParseState *pstate, oumask = umask(S_IWGRP | S_IWOTH); PG_TRY(); { - cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W); + cstate->edata->copy_file = AllocateFile(cstate->filename, PG_BINARY_W); } PG_FINALLY(); { umask(oumask); } PG_END_TRY(); - if (cstate->copy_file == NULL) + if (cstate->edata->copy_file == NULL) { /* copy errno because ereport subfunctions might change it */ int save_errno = errno; @@ -973,7 +958,7 @@ BeginCopyTo(ParseState *pstate, "You may want a client-side facility such as psql's \\copy.") : 0)); } - if (fstat(fileno(cstate->copy_file), &st)) + if (fstat(fileno(cstate->edata->copy_file), &st)) ereport(ERROR, (errcode_for_file_access(), errmsg("could not stat file \"%s\": %m", @@ -991,7 +976,7 @@ BeginCopyTo(ParseState *pstate, cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid); pgstat_progress_update_multi_param(2, progress_cols, progress_vals); - cstate->bytes_processed = 0; + cstate->edata->bytes_processed = 0; MemoryContextSwitchTo(oldcontext); @@ -1042,8 +1027,8 @@ DoCopyTo(CopyToState cstate) num_phys_attrs = tupDesc->natts; cstate->opts.null_print_client = cstate->opts.null_print; /* default */ - /* We use fe_msgbuf as a per-row buffer regardless of copy_dest */ - cstate->fe_msgbuf = makeStringInfo(); + /* We use fe_msgbuf as a per-row buffer regardless of edata->copy_dest */ + cstate->edata->fe_msgbuf = makeStringInfo(); /* Get info about the columns we need to process. */ cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo)); @@ -1062,9 +1047,9 @@ DoCopyTo(CopyToState cstate) * datatype output routines, and should be faster than retail pfree's * anyway. (We don't need a whole econtext as CopyFrom does.) */ - cstate->rowcontext = AllocSetContextCreate(CurrentMemoryContext, - "COPY TO", - ALLOCSET_DEFAULT_SIZES); + cstate->edata->rowcontext = AllocSetContextCreate(CurrentMemoryContext, + "COPY TO", + ALLOCSET_DEFAULT_SIZES); cstate->routine->CopyToStart(cstate, tupDesc); @@ -1107,7 +1092,7 @@ DoCopyTo(CopyToState cstate) cstate->routine->CopyToEnd(cstate); - MemoryContextDelete(cstate->rowcontext); + MemoryContextDelete(cstate->edata->rowcontext); if (fe_copy) SendCopyEnd(cstate); @@ -1123,8 +1108,8 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot) { MemoryContext oldcontext; - MemoryContextReset(cstate->rowcontext); - oldcontext = MemoryContextSwitchTo(cstate->rowcontext); + MemoryContextReset(cstate->edata->rowcontext); + oldcontext = MemoryContextSwitchTo(cstate->edata->rowcontext); /* Make sure the tuple is fully deconstructed */ slot_getallattrs(slot); @@ -1151,7 +1136,7 @@ CopyAttributeOutText(CopyToState cstate, const char *string) char c; char delimc = cstate->opts.delim[0]; - if (cstate->need_transcoding) + if (cstate->edata->need_transcoding) ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding); else ptr = string; @@ -1170,7 +1155,7 @@ CopyAttributeOutText(CopyToState cstate, const char *string) * it's worth making two copies of it to get the IS_HIGHBIT_SET() test out * of the normal safe-encoding path. */ - if (cstate->encoding_embeds_ascii) + if (cstate->edata->encoding_embeds_ascii) { start = ptr; while ((c = *ptr) != '\0') @@ -1312,7 +1297,7 @@ CopyAttributeOutCSV(CopyToState cstate, const char *string, if (!use_quote && strcmp(string, cstate->opts.null_print) == 0) use_quote = true; - if (cstate->need_transcoding) + if (cstate->edata->need_transcoding) ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding); else ptr = string; @@ -1342,7 +1327,7 @@ CopyAttributeOutCSV(CopyToState cstate, const char *string, use_quote = true; break; } - if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii) + if (IS_HIGHBIT_SET(c) && cstate->edata->encoding_embeds_ascii) tptr += pg_encoding_mblen(cstate->file_encoding, tptr); else tptr++; @@ -1366,7 +1351,7 @@ CopyAttributeOutCSV(CopyToState cstate, const char *string, CopySendChar(cstate, escapec); start = ptr; /* we include char in next run */ } - if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii) + if (IS_HIGHBIT_SET(c) && cstate->edata->encoding_embeds_ascii) ptr += pg_encoding_mblen(cstate->file_encoding, ptr); else ptr++; diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index 541176e1980..1fd44a132be 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -14,6 +14,7 @@ #ifndef COPY_H #define COPY_H +#include "executor/execdesc.h" #include "nodes/execnodes.h" #include "nodes/parsenodes.h" #include "parser/parse_node.h" @@ -87,13 +88,61 @@ typedef struct CopyFormatOptions List *convert_select; /* list of column names (can be NIL) */ } CopyFormatOptions; +typedef int (*copy_data_source_cb) (void *outbuf, int minread, int maxread); +typedef void (*copy_data_dest_cb) (void *data, int len); + +typedef struct CopyToStateData +{ + /* format-specific routines */ + const struct CopyToRoutine *routine; + + struct CopyToExecutionData *edata; + + /* parameters from the COPY command */ + Relation rel; /* relation to copy to */ + QueryDesc *queryDesc; /* executable query to copy from */ + List *attnumlist; /* integer list of attnums to copy */ + char *filename; /* filename, or NULL for STDOUT */ + bool is_program; /* is 'filename' a program to popen? */ + copy_data_dest_cb data_dest_cb; /* function for writing data */ + + int file_encoding; /* file or remote side's character encoding */ + bool need_transcoding; /* file encoding diff from server? */ + bool encoding_embeds_ascii; /* ASCII can be non-first byte? */ + + CopyFormatOptions opts; + + FmgrInfo *out_functions; /* lookup info for output functions */ +} CopyToStateData; + +typedef struct CopyFromStateData +{ + /* format routine */ + const struct CopyFromRoutine *routine; + + struct CopyFromExecutionData *edata; + + /* parameters from the COPY command */ + Relation rel; /* relation to copy from */ + List *attnumlist; /* integer list of attnums to copy */ + char *filename; /* filename, or NULL for STDIN */ + bool is_program; /* is 'filename' a program to popen? */ + copy_data_source_cb data_source_cb; /* function for reading data */ + + int file_encoding; /* file or remote side's character encoding */ + bool need_transcoding; /* file encoding diff from server? */ + Oid conversion_proc; /* encoding conversion function */ + + CopyFormatOptions opts; + + FmgrInfo *in_functions; /* array of input functions for each attrs */ + Oid *typioparams; /* array of element types for in_functions */ +} CopyFromStateData; + /* These are private in commands/copy[from|to].c */ typedef struct CopyFromStateData *CopyFromState; typedef struct CopyToStateData *CopyToState; -typedef int (*copy_data_source_cb) (void *outbuf, int minread, int maxread); -typedef void (*copy_data_dest_cb) (void *data, int len); - extern void DoCopy(ParseState *pstate, const CopyStmt *stmt, int stmt_location, int stmt_len, uint64 *processed); diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h index c8b22af22d8..b91b907e8eb 100644 --- a/src/include/commands/copyfrom_internal.h +++ b/src/include/commands/copyfrom_internal.h @@ -56,29 +56,17 @@ typedef enum CopyInsertMethod * This struct contains all the state variables used throughout a COPY FROM * operation. */ -typedef struct CopyFromStateData +typedef struct CopyFromExecutionData { - /* format routine */ - const struct CopyFromRoutine *routine; - /* low-level state data */ CopySource copy_src; /* type of copy source */ FILE *copy_file; /* used if copy_src == COPY_FILE */ StringInfo fe_msgbuf; /* used if copy_src == COPY_FRONTEND */ EolType eol_type; /* EOL type of input */ - int file_encoding; /* file or remote side's character encoding */ bool need_transcoding; /* file encoding diff from server? */ Oid conversion_proc; /* encoding conversion function */ - /* parameters from the COPY command */ - Relation rel; /* relation to copy from */ - List *attnumlist; /* integer list of attnums to copy */ - char *filename; /* filename, or NULL for STDIN */ - bool is_program; /* is 'filename' a program to popen? */ - copy_data_source_cb data_source_cb; /* function for reading data */ - - CopyFormatOptions opts; bool *convert_select_flags; /* per-column CSV/TEXT CS flags */ Node *whereClause; /* WHERE condition (or NULL) */ @@ -96,8 +84,6 @@ typedef struct CopyFromStateData AttrNumber num_defaults; /* count of att that are missing and have * default value */ - FmgrInfo *in_functions; /* array of input functions for each attrs */ - Oid *typioparams; /* array of element types for in_functions */ ErrorSaveContext *escontext; /* soft error trapped during in_functions * execution */ uint64 num_errors; /* total number of rows which contained soft @@ -164,7 +150,7 @@ typedef struct CopyFromStateData bool input_reached_eof; /* true if we reached EOF */ bool input_reached_error; /* true if a conversion error happened */ /* Shorthand for number of unconsumed bytes available in input_buf */ -#define INPUT_BUF_BYTES(cstate) ((cstate)->input_buf_len - (cstate)->input_buf_index) +#define INPUT_BUF_BYTES(edata) ((edata)->input_buf_len - (edata)->input_buf_index) /* * raw_buf holds raw input data read from the data source (file or client @@ -178,10 +164,10 @@ typedef struct CopyFromStateData bool raw_reached_eof; /* true if we reached EOF */ /* Shorthand for number of unconsumed bytes available in raw_buf */ -#define RAW_BUF_BYTES(cstate) ((cstate)->raw_buf_len - (cstate)->raw_buf_index) +#define RAW_BUF_BYTES(edata) ((edata)->raw_buf_len - (edata)->raw_buf_index) uint64 bytes_processed; /* number of bytes processed so far */ -} CopyFromStateData; +} CopyFromExecutionData; extern void ReceiveCopyBegin(CopyFromState cstate); extern void ReceiveCopyBinaryHeader(CopyFromState cstate); -- 2.43.5