0001-Refactor-CopyToStateData-and-CopyFromStateData.patch
application/octet-stream
Filename: 0001-Refactor-CopyToStateData-and-CopyFromStateData.patch
Type: application/octet-stream
Part: 0
Patch
Format: format-patch
Series: patch 0001
Subject: Refactor CopyToStateData and CopyFromStateData.
| File | + | − |
|---|---|---|
| src/backend/commands/copyfrom.c | 123 | 115 |
| src/backend/commands/copyfromparse.c | 230 | 216 |
| src/backend/commands/copyto.c | 48 | 63 |
| src/include/commands/copyfrom_internal.h | 4 | 18 |
| src/include/commands/copy.h | 52 | 3 |
From f80e7ea77a4faa2abd64782e9bdfc20ad8d0eff3 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
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