0002-Separate-format-specific-fields-from-CopyFromStateDa.patch
application/octet-stream
Filename: 0002-Separate-format-specific-fields-from-CopyFromStateDa.patch
Type: application/octet-stream
Part: 4
Patch
Format: format-patch
Series: patch 0002
Subject: Separate format-specific fields from CopyFromStateData struct.
| File | + | − |
|---|---|---|
| contrib/file_fdw/file_fdw.c | 3 | 2 |
| src/backend/commands/copyfrom.c | 152 | 101 |
| src/backend/commands/copyfromparse.c | 160 | 132 |
| src/include/commands/copyapi.h | 6 | 2 |
| src/include/commands/copyfrom_internal.h | 10 | 94 |
| src/include/commands/copy.h | 8 | 4 |
| src/include/commands/copystate.h | 97 | 0 |
From 54a307c3932d9fb080fdec3ce8b1068138f35f24 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Fri, 7 Nov 2025 15:33:32 -0800
Subject: [PATCH 2/6] Separate format-specific fields from CopyFromStateData
struct.
Previously, CopyFromStateData struct contained all state variables
used throughout COPY FROM operations. To support the upcoming
pluggable COPY format feature, this commit moves fields specific to
built-in formats into their own dedicated struct.
This commit also introduces a new CopyFromRoutine callback named
CopyFromEstimateStateSpace, allowing format routines to estimate and
return the memory size needed for their state data. The format's
state data must embed CopyFromStateData as its first field, and the
returned size must be equal to or larger than
CopyFromStateData. While separate structs for core and
format-specific usage might seem feasible, using contiguous memory
avoids overhead from extra pointer traversal.
This commit mixed many changes but the main goal is to move some
fields that are used only for built-in formats from out of
CopyFromStateData so that format implementation can use their own
state data, similar to what we did for COPY TO. Here is the summary of
changes:
* Introduce a new callback CopyFromEstimateStateSpace.
* Introduce a new struct CopyFromStateBuiltins for built-in format
state data.
* CopyFromStateData has fields used by the core.
* The new struct embeds CopyFromStateData at the first field.
* line_buf, raw_buf, and attribute_buf and their related fields are
now format-specific fields.
* Move some initialization steps (e.g., encoding convesion
preparation) to format's CopyFromStart callback.
Callback refactoring/handling.
==============================
One concern is how to integrate the error callback with
custom format. Currently, COPY FROM's error callback
CopyFromErrorCallback() is set right before the main loop in
CopyFrom() for COPY FROM command and unset after the loop, whereas
COPY command API allows users to call just BeginCopyFrom(),
NextCopyFrom(), and EndCopyFrom() (i.e., without CopyFrom()). I
initially thought we can think that the error callback is
format-specific function so we can set/unset the callback within the
COPY format callbacks (e.g., setting the callback in CopyFromStart
callback etc). However, I found out that it's not straightforward for
example because CopyFromStart can be called multiple times by
file_fdw.c. The current idea is that we allows format-implementations
to update the core's cur_XXX fields so that the core's error callback
can print the details. IOW, the error callback function is still a
core function that allows format-implementation can access and
update. While it looks working, one concern is priting the input row;
now that the input buffer and line buffer are format-speicic fileds,
there is no way for the core to access them to print the actual line
being read. One trick I used is to have the format implementation set
a pointer in the core's CopyFromStateData to the line_buf in format's
CopyFromStateBuiltins.
---
contrib/file_fdw/file_fdw.c | 5 +-
src/backend/commands/copyfrom.c | 253 ++++++++++++--------
src/backend/commands/copyfromparse.c | 292 +++++++++++++----------
src/include/commands/copy.h | 12 +-
src/include/commands/copyapi.h | 8 +-
src/include/commands/copyfrom_internal.h | 104 +-------
src/include/commands/copystate.h | 97 ++++++++
7 files changed, 436 insertions(+), 335 deletions(-)
diff --git a/contrib/file_fdw/file_fdw.c b/contrib/file_fdw/file_fdw.c
index 70564a68b13..af2d86e5b5c 100644
--- a/contrib/file_fdw/file_fdw.c
+++ b/contrib/file_fdw/file_fdw.c
@@ -766,7 +766,8 @@ retry:
*/
ExecClearTuple(slot);
- if (NextCopyFrom(cstate, econtext, slot->tts_values, slot->tts_isnull))
+ if (NextCopyFrom(cstate, econtext, slot->tts_values, slot->tts_isnull,
+ NULL))
{
if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE &&
cstate->escontext->error_occurred)
@@ -1246,7 +1247,7 @@ file_acquire_sample_rows(Relation onerel, int elevel,
MemoryContextReset(tupcontext);
MemoryContextSwitchTo(tupcontext);
- found = NextCopyFrom(cstate, NULL, values, nulls);
+ found = NextCopyFrom(cstate, NULL, values, nulls, NULL);
MemoryContextSwitchTo(oldcontext);
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index 12781963b4f..0c51e5ba5e1 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -30,6 +30,8 @@
#include "catalog/namespace.h"
#include "commands/copyapi.h"
#include "commands/copyfrom_internal.h"
+#include "commands/copystate.h"
+#include "commands/defrem.h"
#include "commands/progress.h"
#include "commands/trigger.h"
#include "executor/execPartition.h"
@@ -129,6 +131,7 @@ static void CopyFromBinaryEnd(CopyFromState cstate);
/* text format */
static const CopyFromRoutine CopyFromRoutineText = {
+ .CopyFromEstimateStateSpace = CopyFromBuiltinsEstimateSpace,
.CopyFromInFunc = CopyFromTextLikeInFunc,
.CopyFromStart = CopyFromTextLikeStart,
.CopyFromOneRow = CopyFromTextOneRow,
@@ -137,6 +140,7 @@ static const CopyFromRoutine CopyFromRoutineText = {
/* CSV format */
static const CopyFromRoutine CopyFromRoutineCSV = {
+ .CopyFromEstimateStateSpace = CopyFromBuiltinsEstimateSpace,
.CopyFromInFunc = CopyFromTextLikeInFunc,
.CopyFromStart = CopyFromTextLikeStart,
.CopyFromOneRow = CopyFromCSVOneRow,
@@ -145,54 +149,129 @@ static const CopyFromRoutine CopyFromRoutineCSV = {
/* binary format */
static const CopyFromRoutine CopyFromRoutineBinary = {
+ .CopyFromEstimateStateSpace = CopyFromBuiltinsEstimateSpace,
.CopyFromInFunc = CopyFromBinaryInFunc,
.CopyFromStart = CopyFromBinaryStart,
.CopyFromOneRow = CopyFromBinaryOneRow,
.CopyFromEnd = CopyFromBinaryEnd,
};
-/* Return a COPY FROM routine for the given options */
-static const CopyFromRoutine *
-CopyFromGetRoutine(const CopyFormatOptions *opts)
+/*
+ * Common routine to initialize CopyFromStateBuiltins data.
+ */
+static void
+initialize_copyfrom_bultins_state(CopyFromStateBuiltins * state, TupleDesc tupDesc)
{
- if (opts->csv_mode)
- return &CopyFromRoutineCSV;
- else if (opts->binary)
- return &CopyFromRoutineBinary;
+ /* Use client encoding when ENCODING option is not specified. */
+ if (state->base.opts.file_encoding < 0)
+ state->file_encoding = pg_get_client_encoding();
+ else
+ state->file_encoding = state->base.opts.file_encoding;
+
+ /*
+ * Look up encoding conversion function.
+ */
+ if (state->file_encoding == GetDatabaseEncoding() ||
+ state->file_encoding == PG_SQL_ASCII ||
+ GetDatabaseEncoding() == PG_SQL_ASCII)
+ {
+ state->need_transcoding = false;
+ }
+ else
+ {
+ state->need_transcoding = true;
+ state->conversion_proc = FindDefaultConversionProc(state->file_encoding,
+ GetDatabaseEncoding());
+ if (!OidIsValid(state->conversion_proc))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_FUNCTION),
+ errmsg("default conversion function for encoding \"%s\" to \"%s\" does not exist",
+ pg_encoding_to_char(state->file_encoding),
+ pg_encoding_to_char(GetDatabaseEncoding()))));
+ }
- /* default is text */
- return &CopyFromRoutineText;
+ state->defaults = (bool *) palloc0(tupDesc->natts * sizeof(bool));
+
+ /* initialize variables */
+ state->eol_type = EOL_UNKNOWN;
+
+ /* Convert convert_selectively name list to per-column flags */
+ if (state->base.opts.convert_selectively)
+ {
+ List *attnums;
+ ListCell *cur;
+ int num_phys_attrs = RelationGetDescr(state->base.rel)->natts;
+
+ state->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
+
+ attnums = CopyGetAttnums(tupDesc, state->base.rel, state->base.opts.convert_select);
+
+ foreach(cur, attnums)
+ {
+ int attnum = lfirst_int(cur);
+ Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
+
+ if (!list_member_int(state->base.attnumlist, attnum))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+ errmsg_internal("selected column \"%s\" not referenced by COPY",
+ NameStr(attr->attname))));
+ state->convert_select_flags[attnum - 1] = true;
+ }
+ }
+
+ /*
+ * Allocate buffers for the input pipeline.
+ *
+ * attribute_buf and raw_buf are used in both text and binary modes, but
+ * input_buf and line_buf only in text mode.
+ */
+ state->raw_buf = palloc(RAW_BUF_SIZE + 1);
+ state->raw_buf_index = state->raw_buf_len = 0;
+
+ initStringInfo(&state->attribute_buf);
+
+ /* Initialize state variables */
+ state->base.cur_relname = RelationGetRelationName(state->base.rel);
+ state->base.cur_lineno = 0;
+ state->base.cur_attname = NULL;
+ state->base.cur_attval = NULL;
+ state->base.relname_only = false;
}
/* Implementation of the start callback for text and CSV formats */
static void
CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc)
{
+ CopyFromStateBuiltins *state = (CopyFromStateBuiltins *) cstate;
AttrNumber attr_count;
+ initialize_copyfrom_bultins_state(state, tupDesc);
+
/*
* If encoding conversion is needed, we need another buffer to hold the
* converted input data. Otherwise, we can just point input_buf to the
* same buffer as raw_buf.
*/
- if (cstate->need_transcoding)
+ if (state->need_transcoding)
{
- cstate->input_buf = (char *) palloc(INPUT_BUF_SIZE + 1);
- cstate->input_buf_index = cstate->input_buf_len = 0;
+ state->input_buf = (char *) palloc(INPUT_BUF_SIZE + 1);
+ state->input_buf_index = state->input_buf_len = 0;
}
else
- cstate->input_buf = cstate->raw_buf;
- cstate->input_reached_eof = false;
+ state->input_buf = state->raw_buf;
+ state->input_reached_eof = false;
- initStringInfo(&cstate->line_buf);
+ initStringInfo(&state->line_buf);
+ state->base.line_buf = &state->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 *));
+ attr_count = list_length(state->base.attnumlist);
+ state->max_fields = attr_count;
+ state->raw_fields = (char **) palloc(attr_count * sizeof(char *));
}
/*
@@ -220,6 +299,8 @@ CopyFromTextLikeEnd(CopyFromState cstate)
static void
CopyFromBinaryStart(CopyFromState cstate, TupleDesc tupDesc)
{
+ initialize_copyfrom_bultins_state((CopyFromStateBuiltins *) cstate, tupDesc);
+
/* Read and verify binary header */
ReceiveCopyBinaryHeader(cstate);
}
@@ -308,7 +389,7 @@ CopyFromErrorCallback(void *arg)
{
char *lineval;
- lineval = CopyLimitPrintoutLength(cstate->line_buf.data);
+ lineval = CopyLimitPrintoutLength(cstate->line_buf->data);
errcontext("COPY %s, line %" PRIu64 ": \"%s\"",
cstate->cur_relname,
cstate->cur_lineno, lineval);
@@ -1113,6 +1194,7 @@ CopyFrom(CopyFromState cstate)
{
TupleTableSlot *myslot;
bool skip_tuple;
+ CopyFromRowInfo rowinfo = {0};
CHECK_FOR_INTERRUPTS();
@@ -1146,7 +1228,8 @@ CopyFrom(CopyFromState cstate)
ExecClearTuple(myslot);
/* Directly store the values/nulls array in the slot */
- if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+ if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull,
+ &rowinfo))
break;
if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE &&
@@ -1379,8 +1462,8 @@ CopyFrom(CopyFromState cstate)
/* Add this tuple to the tuple buffer */
CopyMultiInsertInfoStore(&multiInsertInfo,
resultRelInfo, myslot,
- cstate->line_buf.len,
- cstate->cur_lineno);
+ rowinfo.tuplen,
+ rowinfo.lineno);
/*
* If enough inserts have queued up, then flush all
@@ -1512,6 +1595,50 @@ CopyFrom(CopyFromState cstate)
return processed;
}
+static CopyFromState
+create_copyfrom_state(ParseState *pstate, List *options)
+{
+ const CopyFromRoutine *routine;
+ CopyFromState cstate;
+ Size req_size;
+ bool format_specified = false;
+
+ routine = &CopyFromRoutineText; /* default */
+ foreach_node(DefElem, defel, options)
+ {
+ if (strcmp(defel->defname, "format") == 0)
+ {
+ char *fmt = defGetString(defel);
+
+ if (format_specified)
+ errorConflictingDefElem(defel, pstate);
+ format_specified = true;
+ if (strcmp(fmt, "text") == 0)
+ routine = &CopyFromRoutineText;
+ else if (strcmp(fmt, "csv") == 0)
+ routine = &CopyFromRoutineCSV;
+ else if (strcmp(fmt, "binary") == 0)
+ routine = &CopyFromRoutineBinary;
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("COPY format \"%s\" not recognized", fmt),
+ parser_errposition(pstate, defel->location)));
+
+ break;
+ }
+ }
+
+ req_size = routine->CopyFromEstimateStateSpace();
+ Assert(req_size >= sizeof(CopyFromStateData));
+
+ /* Allocate workspace and zero all fields */
+ cstate = (CopyFromState) palloc0(req_size);
+ cstate->routine = routine;
+
+ return cstate;
+}
+
/*
* Setup to read tuples from a file for COPY FROM.
*
@@ -1558,7 +1685,7 @@ BeginCopyFrom(ParseState *pstate,
};
/* Allocate workspace and zero all fields */
- cstate = (CopyFromStateData *) palloc0(sizeof(CopyFromStateData));
+ cstate = create_copyfrom_state(pstate, options);
/*
* We allocate everything used by a cstate in a new memory context. This
@@ -1573,9 +1700,6 @@ BeginCopyFrom(ParseState *pstate,
/* Extract options from the statement node tree */
ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , options);
- /* Set the format routine */
- cstate->routine = CopyFromGetRoutine(&cstate->opts);
-
/* Process the target relation */
cstate->rel = rel;
@@ -1657,82 +1781,10 @@ BeginCopyFrom(ParseState *pstate,
}
}
- /* Convert convert_selectively name list to per-column flags */
- if (cstate->opts.convert_selectively)
- {
- List *attnums;
- ListCell *cur;
-
- cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
-
- attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.convert_select);
-
- foreach(cur, attnums)
- {
- int attnum = lfirst_int(cur);
- Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
-
- if (!list_member_int(cstate->attnumlist, attnum))
- ereport(ERROR,
- (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
- errmsg_internal("selected column \"%s\" not referenced by COPY",
- NameStr(attr->attname))));
- cstate->convert_select_flags[attnum - 1] = true;
- }
- }
-
- /* Use client encoding when ENCODING option is not specified. */
- if (cstate->opts.file_encoding < 0)
- cstate->file_encoding = pg_get_client_encoding();
- else
- cstate->file_encoding = cstate->opts.file_encoding;
-
- /*
- * Look up encoding conversion function.
- */
- if (cstate->file_encoding == GetDatabaseEncoding() ||
- cstate->file_encoding == PG_SQL_ASCII ||
- GetDatabaseEncoding() == PG_SQL_ASCII)
- {
- cstate->need_transcoding = false;
- }
- else
- {
- cstate->need_transcoding = true;
- cstate->conversion_proc = FindDefaultConversionProc(cstate->file_encoding,
- GetDatabaseEncoding());
- if (!OidIsValid(cstate->conversion_proc))
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_FUNCTION),
- errmsg("default conversion function for encoding \"%s\" to \"%s\" does not exist",
- pg_encoding_to_char(cstate->file_encoding),
- pg_encoding_to_char(GetDatabaseEncoding()))));
- }
-
cstate->copy_src = COPY_FILE; /* default */
cstate->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;
-
- /*
- * Allocate buffers for the input pipeline.
- *
- * 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;
-
- initStringInfo(&cstate->attribute_buf);
-
/* Assign range table and rteperminfos, we'll need them in CopyFrom. */
if (pstate)
{
@@ -1818,8 +1870,6 @@ BeginCopyFrom(ParseState *pstate,
}
}
- cstate->defaults = (bool *) palloc0(tupDesc->natts * sizeof(bool));
-
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
@@ -1833,6 +1883,7 @@ BeginCopyFrom(ParseState *pstate,
cstate->volatile_defexprs = volatile_defexprs;
cstate->num_defaults = num_defaults;
cstate->is_program = is_program;
+ cstate->reached_eof = false;
if (data_source_cb)
{
@@ -1959,7 +2010,7 @@ ClosePipeFromProgram(CopyFromState cstate)
* should not report that as an error. Otherwise, SIGPIPE indicates a
* problem.
*/
- if (!cstate->raw_reached_eof &&
+ if (!cstate->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..669dbfb8459 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -138,21 +138,20 @@ if (1) \
/* NOTE: there's a copy of this in copyto.c */
static const char BinarySignature[11] = "PGCOPY\n\377\r\n\0";
-
/* non-export function prototypes */
-static bool CopyReadLine(CopyFromState cstate, bool is_csv);
-static bool CopyReadLineText(CopyFromState cstate, bool is_csv);
-static int CopyReadAttributesText(CopyFromState cstate);
-static int CopyReadAttributesCSV(CopyFromState cstate);
-static Datum CopyReadBinaryAttribute(CopyFromState cstate, FmgrInfo *flinfo,
+static bool CopyReadLine(CopyFromStateBuiltins * state, bool is_csv);
+static bool CopyReadLineText(CopyFromStateBuiltins * state, bool is_csv);
+static int CopyReadAttributesText(CopyFromStateBuiltins * state);
+static int CopyReadAttributesCSV(CopyFromStateBuiltins * state);
+static Datum CopyReadBinaryAttribute(CopyFromStateBuiltins * state, FmgrInfo *flinfo,
Oid typioparam, int32 typmod,
bool *isnull);
-static pg_attribute_always_inline bool CopyFromTextLikeOneRow(CopyFromState cstate,
+static pg_attribute_always_inline bool CopyFromTextLikeOneRow(CopyFromStateBuiltins * state,
ExprContext *econtext,
- Datum *values,
- bool *nulls,
+ Datum *values, bool *nulls,
+ CopyFromRowInfo * rowinfo,
bool is_csv);
-static pg_attribute_always_inline bool NextCopyFromRawFieldsInternal(CopyFromState cstate,
+static pg_attribute_always_inline bool NextCopyFromRawFieldsInternal(CopyFromStateBuiltins * state,
char ***fields,
int *nfields,
bool is_csv);
@@ -161,10 +160,10 @@ static pg_attribute_always_inline bool NextCopyFromRawFieldsInternal(CopyFromSta
/* Low-level communications functions */
static int CopyGetData(CopyFromState cstate, void *databuf,
int minread, int maxread);
-static inline bool CopyGetInt32(CopyFromState cstate, int32 *val);
-static inline bool CopyGetInt16(CopyFromState cstate, int16 *val);
-static void CopyLoadInputBuf(CopyFromState cstate);
-static int CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes);
+static inline bool CopyGetInt32(CopyFromStateBuiltins * state, int32 *val);
+static inline bool CopyGetInt16(CopyFromStateBuiltins * state, int16 *val);
+static void CopyLoadInputBuf(CopyFromStateBuiltins * state);
+static int CopyReadBinaryData(CopyFromStateBuiltins * state, char *dest, int nbytes);
void
ReceiveCopyBegin(CopyFromState cstate)
@@ -187,8 +186,9 @@ ReceiveCopyBegin(CopyFromState cstate)
}
void
-ReceiveCopyBinaryHeader(CopyFromState cstate)
+ReceiveCopyBinaryHeader(CopyFromState ccstate)
{
+ CopyFromStateBuiltins *cstate = (CopyFromStateBuiltins *) ccstate;
char readSig[11];
int32 tmp;
@@ -255,10 +255,10 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread)
(errcode_for_file_access(),
errmsg("could not read from COPY file: %m")));
if (bytesread == 0)
- cstate->raw_reached_eof = true;
+ cstate->reached_eof = true;
break;
case COPY_FRONTEND:
- while (maxread > 0 && bytesread < minread && !cstate->raw_reached_eof)
+ while (maxread > 0 && bytesread < minread && !cstate->reached_eof)
{
int avail;
@@ -309,7 +309,7 @@ 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->reached_eof = true;
return bytesread;
case PqMsg_CopyFail:
ereport(ERROR,
@@ -359,7 +359,7 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread)
* Returns true if OK, false if EOF
*/
static inline bool
-CopyGetInt32(CopyFromState cstate, int32 *val)
+CopyGetInt32(CopyFromStateBuiltins * cstate, int32 *val)
{
uint32 buf;
@@ -376,7 +376,7 @@ CopyGetInt32(CopyFromState cstate, int32 *val)
* CopyGetInt16 reads an int16 that appears in network byte order
*/
static inline bool
-CopyGetInt16(CopyFromState cstate, int16 *val)
+CopyGetInt16(CopyFromStateBuiltins * cstate, int16 *val)
{
uint16 buf;
@@ -397,7 +397,7 @@ CopyGetInt16(CopyFromState cstate, int16 *val)
* On entry, there must be some data to convert in 'raw_buf'.
*/
static void
-CopyConvertBuf(CopyFromState cstate)
+CopyConvertBuf(CopyFromStateBuiltins * cstate)
{
/*
* If the file and server encoding are the same, no encoding conversion is
@@ -421,7 +421,7 @@ CopyConvertBuf(CopyFromState cstate)
/*
* If no more raw data is coming, report the EOF to the caller.
*/
- if (cstate->raw_reached_eof)
+ if (cstate->base.reached_eof)
cstate->input_reached_eof = true;
return;
}
@@ -444,7 +444,7 @@ 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))
+ if (cstate->base.reached_eof || unverifiedlen >= pg_encoding_max_length(cstate->file_encoding))
cstate->input_reached_error = true;
return;
}
@@ -467,7 +467,7 @@ CopyConvertBuf(CopyFromState cstate)
/*
* If no more raw data is coming, report the EOF to the caller.
*/
- if (cstate->raw_reached_eof)
+ if (cstate->base.reached_eof)
cstate->input_reached_eof = true;
return;
}
@@ -517,7 +517,7 @@ 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)
+ if (cstate->base.reached_eof || srclen >= MAX_CONVERSION_INPUT_LENGTH)
cstate->input_reached_error = true;
return;
}
@@ -530,7 +530,7 @@ CopyConvertBuf(CopyFromState cstate)
* Report an encoding or conversion error.
*/
static void
-CopyConversionError(CopyFromState cstate)
+CopyConversionError(CopyFromStateBuiltins * cstate)
{
Assert(cstate->raw_buf_len > 0);
Assert(cstate->input_reached_error);
@@ -587,7 +587,7 @@ CopyConversionError(CopyFromState cstate)
* beginning of the buffer, and we load new data after that.
*/
static void
-CopyLoadRawBuf(CopyFromState cstate)
+CopyLoadRawBuf(CopyFromStateBuiltins * cstate)
{
int nbytes;
int inbytes;
@@ -624,17 +624,17 @@ CopyLoadRawBuf(CopyFromState cstate)
}
/* Load more data */
- inbytes = CopyGetData(cstate, cstate->raw_buf + cstate->raw_buf_len,
+ inbytes = CopyGetData((CopyFromState) cstate, cstate->raw_buf + cstate->raw_buf_len,
1, RAW_BUF_SIZE - cstate->raw_buf_len);
nbytes += inbytes;
cstate->raw_buf[nbytes] = '\0';
cstate->raw_buf_len = nbytes;
- cstate->bytes_processed += inbytes;
- pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->bytes_processed);
+ cstate->base.bytes_processed += inbytes;
+ pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->base.bytes_processed);
if (inbytes == 0)
- cstate->raw_reached_eof = true;
+ cstate->base.reached_eof = true;
}
/*
@@ -647,7 +647,7 @@ CopyLoadRawBuf(CopyFromState cstate)
* of the buffer and then we load more data after that.
*/
static void
-CopyLoadInputBuf(CopyFromState cstate)
+CopyLoadInputBuf(CopyFromStateBuiltins * cstate)
{
int nbytes = INPUT_BUF_BYTES(cstate);
@@ -685,7 +685,7 @@ CopyLoadInputBuf(CopyFromState cstate)
break;
/* Try to load more raw data */
- Assert(!cstate->raw_reached_eof);
+ Assert(!cstate->base.reached_eof);
CopyLoadRawBuf(cstate);
}
}
@@ -698,7 +698,7 @@ CopyLoadInputBuf(CopyFromState cstate)
* would be less than 'nbytes' only if we reach EOF).
*/
static int
-CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes)
+CopyReadBinaryData(CopyFromStateBuiltins * cstate, char *dest, int nbytes)
{
int copied_bytes = 0;
@@ -723,7 +723,7 @@ CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes)
if (RAW_BUF_BYTES(cstate) == 0)
{
CopyLoadRawBuf(cstate);
- if (cstate->raw_reached_eof)
+ if (cstate->base.reached_eof)
break; /* EOF */
}
@@ -739,15 +739,21 @@ CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes)
return copied_bytes;
}
+Size
+CopyFromBuiltinsEstimateSpace(void)
+{
+ return sizeof(CopyFromStateBuiltins);
+}
+
/*
* This function is exposed for use by extensions that read raw fields in the
* next line. See NextCopyFromRawFieldsInternal() for details.
*/
bool
-NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields)
+NextCopyFromRawFields(CopyFromState ccstate, char ***fields, int *nfields)
{
- return NextCopyFromRawFieldsInternal(cstate, fields, nfields,
- cstate->opts.csv_mode);
+ return NextCopyFromRawFieldsInternal((CopyFromStateBuiltins *) ccstate,
+ fields, nfields, ccstate->opts.csv_mode);
}
/*
@@ -768,35 +774,35 @@ NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields)
* by internal functions such as CopyFromTextLikeOneRow().
*/
static pg_attribute_always_inline bool
-NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields, bool is_csv)
+NextCopyFromRawFieldsInternal(CopyFromStateBuiltins * cstate, char ***fields, int *nfields, bool is_csv)
{
int fldct;
bool done = false;
/* only available for text or csv input */
- Assert(!cstate->opts.binary);
+ Assert(!cstate->base.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 (cstate->base.cur_lineno == 0 && cstate->base.opts.header_line != COPY_HEADER_FALSE)
{
ListCell *cur;
TupleDesc tupDesc;
- int lines_to_skip = cstate->opts.header_line;
+ int lines_to_skip = cstate->base.opts.header_line;
/* If set to "match", one header line is skipped */
- if (cstate->opts.header_line == COPY_HEADER_MATCH)
+ if (cstate->base.opts.header_line == COPY_HEADER_MATCH)
lines_to_skip = 1;
- tupDesc = RelationGetDescr(cstate->rel);
+ tupDesc = RelationGetDescr(cstate->base.rel);
for (int i = 0; i < lines_to_skip; i++)
{
- cstate->cur_lineno++;
+ cstate->base.cur_lineno++;
if ((done = CopyReadLine(cstate, is_csv)))
break;
}
- if (cstate->opts.header_line == COPY_HEADER_MATCH)
+ if (cstate->base.opts.header_line == COPY_HEADER_MATCH)
{
int fldnum;
@@ -805,14 +811,14 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields
else
fldct = CopyReadAttributesText(cstate);
- if (fldct != list_length(cstate->attnumlist))
+ if (fldct != list_length(cstate->base.attnumlist))
ereport(ERROR,
(errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
errmsg("wrong number of fields in header line: got %d, expected %d",
- fldct, list_length(cstate->attnumlist))));
+ fldct, list_length(cstate->base.attnumlist))));
fldnum = 0;
- foreach(cur, cstate->attnumlist)
+ foreach(cur, cstate->base.attnumlist)
{
int attnum = lfirst_int(cur);
char *colName;
@@ -825,7 +831,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields
ereport(ERROR,
(errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
errmsg("column name mismatch in header line field %d: got null value (\"%s\"), expected \"%s\"",
- fldnum, cstate->opts.null_print, NameStr(attr->attname))));
+ fldnum, cstate->base.opts.null_print, NameStr(attr->attname))));
if (namestrcmp(&attr->attname, colName) != 0)
{
@@ -841,7 +847,7 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields
return false;
}
- cstate->cur_lineno++;
+ cstate->base.cur_lineno++;
/* Actually read the line into memory here */
done = CopyReadLine(cstate, is_csv);
@@ -877,8 +883,8 @@ NextCopyFromRawFieldsInternal(CopyFromState cstate, char ***fields, int *nfields
* relation passed to BeginCopyFrom. This function fills the arrays.
*/
bool
-NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls)
+NextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values,
+ bool *nulls, CopyFromRowInfo * rowinfo)
{
TupleDesc tupDesc;
AttrNumber num_phys_attrs,
@@ -893,10 +899,9 @@ 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));
/* Get one row from source */
- if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+ if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls, rowinfo))
return false;
/*
@@ -923,17 +928,19 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
/* Implementation of the per-row callback for text format */
bool
CopyFromTextOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values,
- bool *nulls)
+ bool *nulls, CopyFromRowInfo * rowinfo)
{
- return CopyFromTextLikeOneRow(cstate, econtext, values, nulls, false);
+ return CopyFromTextLikeOneRow((CopyFromStateBuiltins *) cstate, econtext,
+ values, nulls, rowinfo, false);
}
/* Implementation of the per-row callback for CSV format */
bool
CopyFromCSVOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values,
- bool *nulls)
+ bool *nulls, CopyFromRowInfo * rowinfo)
{
- return CopyFromTextLikeOneRow(cstate, econtext, values, nulls, true);
+ return CopyFromTextLikeOneRow((CopyFromStateBuiltins *) cstate, econtext,
+ values, nulls, rowinfo, true);
}
/*
@@ -943,22 +950,25 @@ CopyFromCSVOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values,
* and to help compilers to optimize away the 'is_csv' condition.
*/
static pg_attribute_always_inline bool
-CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls, bool is_csv)
+CopyFromTextLikeOneRow(CopyFromStateBuiltins * cstate, ExprContext *econtext,
+ Datum *values, bool *nulls, CopyFromRowInfo * rowinfo,
+ bool is_csv)
{
TupleDesc tupDesc;
AttrNumber attr_count;
- FmgrInfo *in_functions = cstate->in_functions;
- Oid *typioparams = cstate->typioparams;
- ExprState **defexprs = cstate->defexprs;
+ FmgrInfo *in_functions = cstate->base.in_functions;
+ Oid *typioparams = cstate->base.typioparams;
+ ExprState **defexprs = cstate->base.defexprs;
char **field_strings;
ListCell *cur;
int fldct;
int fieldno;
char *string;
- tupDesc = RelationGetDescr(cstate->rel);
- attr_count = list_length(cstate->attnumlist);
+ tupDesc = RelationGetDescr(cstate->base.rel);
+ attr_count = list_length(cstate->base.attnumlist);
+
+ MemSet(cstate->defaults, false, tupDesc->natts * sizeof(bool));
/* read raw fields in the next line */
if (!NextCopyFromRawFieldsInternal(cstate, &field_strings, &fldct, is_csv))
@@ -973,7 +983,7 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext,
fieldno = 0;
/* Loop to read the user attributes on the line. */
- foreach(cur, cstate->attnumlist)
+ foreach(cur, cstate->base.attnumlist)
{
int attnum = lfirst_int(cur);
int m = attnum - 1;
@@ -996,16 +1006,16 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext,
if (is_csv)
{
if (string == NULL &&
- cstate->opts.force_notnull_flags[m])
+ cstate->base.opts.force_notnull_flags[m])
{
/*
* FORCE_NOT_NULL option is set and column is NULL - convert
* it to the NULL string.
*/
- string = cstate->opts.null_print;
+ string = cstate->base.opts.null_print;
}
- else if (string != NULL && cstate->opts.force_null_flags[m]
- && strcmp(string, cstate->opts.null_print) == 0)
+ else if (string != NULL && cstate->base.opts.force_null_flags[m]
+ && strcmp(string, cstate->base.opts.null_print) == 0)
{
/*
* FORCE_NULL option is set and column matches the NULL
@@ -1017,8 +1027,8 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext,
}
}
- cstate->cur_attname = NameStr(att->attname);
- cstate->cur_attval = string;
+ cstate->base.cur_attname = NameStr(att->attname);
+ cstate->base.cur_attval = string;
if (string != NULL)
nulls[m] = false;
@@ -1039,73 +1049,81 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext,
string,
typioparams[m],
att->atttypmod,
- (Node *) cstate->escontext,
+ (Node *) cstate->base.escontext,
&values[m]))
{
- Assert(cstate->opts.on_error != COPY_ON_ERROR_STOP);
+ Assert(cstate->base.opts.on_error != COPY_ON_ERROR_STOP);
- cstate->num_errors++;
+ cstate->base.num_errors++;
- if (cstate->opts.log_verbosity == COPY_LOG_VERBOSITY_VERBOSE)
+ if (cstate->base.opts.log_verbosity == COPY_LOG_VERBOSITY_VERBOSE)
{
/*
* Since we emit line number and column info in the below
* notice message, we suppress error context information other
* than the relation name.
*/
- Assert(!cstate->relname_only);
- cstate->relname_only = true;
+ Assert(!cstate->base.relname_only);
+ cstate->base.relname_only = true;
- if (cstate->cur_attval)
+ if (cstate->base.cur_attval)
{
char *attval;
- attval = CopyLimitPrintoutLength(cstate->cur_attval);
+ attval = CopyLimitPrintoutLength(cstate->base.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,
+ cstate->base.cur_lineno,
+ cstate->base.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));
+ cstate->base.cur_lineno,
+ cstate->base.cur_attname));
/* reset relname_only */
- cstate->relname_only = false;
+ cstate->base.relname_only = false;
}
return true;
}
- cstate->cur_attname = NULL;
- cstate->cur_attval = NULL;
+ cstate->base.cur_attname = NULL;
+ cstate->base.cur_attval = NULL;
}
Assert(fieldno == attr_count);
+ /* Set output parameters */
+ if (rowinfo)
+ {
+ rowinfo->lineno = cstate->base.cur_lineno;
+ rowinfo->tuplen = cstate->line_buf.len;
+ }
+
return true;
}
/* Implementation of the per-row callback for binary format */
bool
-CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values,
- bool *nulls)
+CopyFromBinaryOneRow(CopyFromState ccstate, ExprContext *econtext, Datum *values,
+ bool *nulls, CopyFromRowInfo * rowinfo)
{
+ CopyFromStateBuiltins *cstate = (CopyFromStateBuiltins *) ccstate;
TupleDesc tupDesc;
AttrNumber attr_count;
- FmgrInfo *in_functions = cstate->in_functions;
- Oid *typioparams = cstate->typioparams;
+ FmgrInfo *in_functions = cstate->base.in_functions;
+ Oid *typioparams = cstate->base.typioparams;
int16 fld_count;
ListCell *cur;
- tupDesc = RelationGetDescr(cstate->rel);
- attr_count = list_length(cstate->attnumlist);
+ tupDesc = RelationGetDescr(cstate->base.rel);
+ attr_count = list_length(cstate->base.attnumlist);
- cstate->cur_lineno++;
+ cstate->base.cur_lineno++;
if (!CopyGetInt16(cstate, &fld_count))
{
@@ -1138,19 +1156,29 @@ CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values,
errmsg("row field count is %d, expected %d",
(int) fld_count, attr_count)));
- foreach(cur, cstate->attnumlist)
+ foreach(cur, cstate->base.attnumlist)
{
int attnum = lfirst_int(cur);
int m = attnum - 1;
Form_pg_attribute att = TupleDescAttr(tupDesc, m);
- cstate->cur_attname = NameStr(att->attname);
+ cstate->base.cur_attname = NameStr(att->attname);
values[m] = CopyReadBinaryAttribute(cstate,
&in_functions[m],
typioparams[m],
att->atttypmod,
&nulls[m]);
- cstate->cur_attname = NULL;
+ cstate->base.cur_attname = NULL;
+ }
+
+ if (rowinfo)
+ {
+ /*
+ * XXX: We used to use line_buf.len but we don't actually use line_buf
+ * in binary format.
+ */
+ rowinfo->lineno = cstate->base.cur_lineno;
+ rowinfo->tuplen = cstate->line_buf.len;
}
return true;
@@ -1164,12 +1192,12 @@ CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext, Datum *values,
* in the final value of line_buf.
*/
static bool
-CopyReadLine(CopyFromState cstate, bool is_csv)
+CopyReadLine(CopyFromStateBuiltins * cstate, bool is_csv)
{
bool result;
resetStringInfo(&cstate->line_buf);
- cstate->line_buf_valid = false;
+ cstate->base.line_buf_valid = false;
/* Parse data and transfer into line_buf */
result = CopyReadLineText(cstate, is_csv);
@@ -1181,13 +1209,13 @@ 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 (cstate->base.copy_src == COPY_FRONTEND)
{
int inbytes;
do
{
- inbytes = CopyGetData(cstate, cstate->input_buf,
+ inbytes = CopyGetData((CopyFromState) cstate, cstate->input_buf,
1, INPUT_BUF_SIZE);
} while (inbytes > 0);
cstate->input_buf_index = 0;
@@ -1231,7 +1259,7 @@ CopyReadLine(CopyFromState cstate, bool is_csv)
}
/* Now it's safe to use the buffer in error messages */
- cstate->line_buf_valid = true;
+ cstate->base.line_buf_valid = true;
return result;
}
@@ -1240,7 +1268,7 @@ CopyReadLine(CopyFromState cstate, bool is_csv)
* CopyReadLineText - inner loop of CopyReadLine for text mode
*/
static bool
-CopyReadLineText(CopyFromState cstate, bool is_csv)
+CopyReadLineText(CopyFromStateBuiltins * cstate, bool is_csv)
{
char *copy_input_buf;
int input_buf_ptr;
@@ -1257,8 +1285,8 @@ CopyReadLineText(CopyFromState cstate, bool is_csv)
if (is_csv)
{
- quotec = cstate->opts.quote[0];
- escapec = cstate->opts.escape[0];
+ quotec = cstate->base.opts.quote[0];
+ escapec = cstate->base.opts.escape[0];
/* ignore special escape processing if it's the same as quotec */
if (quotec == escapec)
escapec = '\0';
@@ -1367,7 +1395,7 @@ CopyReadLineText(CopyFromState cstate, bool is_csv)
* at all --- is cur_lineno a physical or logical count?)
*/
if (in_quote && c == (cstate->eol_type == EOL_NL ? '\n' : '\r'))
- cstate->cur_lineno++;
+ cstate->base.cur_lineno++;
}
/* Process \r */
@@ -1570,9 +1598,9 @@ GetDecimalFromHex(char hex)
* The return value is the number of fields actually read.
*/
static int
-CopyReadAttributesText(CopyFromState cstate)
+CopyReadAttributesText(CopyFromStateBuiltins * cstate)
{
- char delimc = cstate->opts.delim[0];
+ char delimc = cstate->base.opts.delim[0];
int fieldno;
char *output_ptr;
char *cur_ptr;
@@ -1755,26 +1783,26 @@ CopyReadAttributesText(CopyFromState cstate)
/* Check whether raw input matched null marker */
input_len = end_ptr - start_ptr;
- if (input_len == cstate->opts.null_print_len &&
- strncmp(start_ptr, cstate->opts.null_print, input_len) == 0)
+ if (input_len == cstate->base.opts.null_print_len &&
+ strncmp(start_ptr, cstate->base.opts.null_print, input_len) == 0)
cstate->raw_fields[fieldno] = NULL;
/* Check whether raw input matched default marker */
- else if (fieldno < list_length(cstate->attnumlist) &&
- cstate->opts.default_print &&
- input_len == cstate->opts.default_print_len &&
- strncmp(start_ptr, cstate->opts.default_print, input_len) == 0)
+ else if (fieldno < list_length(cstate->base.attnumlist) &&
+ cstate->base.opts.default_print &&
+ input_len == cstate->base.opts.default_print_len &&
+ strncmp(start_ptr, cstate->base.opts.default_print, input_len) == 0)
{
/* fieldno is 0-indexed and attnum is 1-indexed */
- int m = list_nth_int(cstate->attnumlist, fieldno) - 1;
+ int m = list_nth_int(cstate->base.attnumlist, fieldno) - 1;
- if (cstate->defexprs[m] != NULL)
+ if (cstate->base.defexprs[m] != NULL)
{
/* defaults contain entries for all physical attributes */
cstate->defaults[m] = true;
}
else
{
- TupleDesc tupDesc = RelationGetDescr(cstate->rel);
+ TupleDesc tupDesc = RelationGetDescr(cstate->base.rel);
Form_pg_attribute att = TupleDescAttr(tupDesc, m);
ereport(ERROR,
@@ -1824,11 +1852,11 @@ CopyReadAttributesText(CopyFromState cstate)
* "standard" (i.e. common) CSV usage.
*/
static int
-CopyReadAttributesCSV(CopyFromState cstate)
+CopyReadAttributesCSV(CopyFromStateBuiltins * cstate)
{
- char delimc = cstate->opts.delim[0];
- char quotec = cstate->opts.quote[0];
- char escapec = cstate->opts.escape[0];
+ char delimc = cstate->base.opts.delim[0];
+ char quotec = cstate->base.opts.quote[0];
+ char escapec = cstate->base.opts.escape[0];
int fieldno;
char *output_ptr;
char *cur_ptr;
@@ -1970,26 +1998,26 @@ endfield:
/* Check whether raw input matched null marker */
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)
+ if (!saw_quote && input_len == cstate->base.opts.null_print_len &&
+ strncmp(start_ptr, cstate->base.opts.null_print, input_len) == 0)
cstate->raw_fields[fieldno] = NULL;
/* Check whether raw input matched default marker */
- else if (fieldno < list_length(cstate->attnumlist) &&
- cstate->opts.default_print &&
- input_len == cstate->opts.default_print_len &&
- strncmp(start_ptr, cstate->opts.default_print, input_len) == 0)
+ else if (fieldno < list_length(cstate->base.attnumlist) &&
+ cstate->base.opts.default_print &&
+ input_len == cstate->base.opts.default_print_len &&
+ strncmp(start_ptr, cstate->base.opts.default_print, input_len) == 0)
{
/* fieldno is 0-index and attnum is 1-index */
- int m = list_nth_int(cstate->attnumlist, fieldno) - 1;
+ int m = list_nth_int(cstate->base.attnumlist, fieldno) - 1;
- if (cstate->defexprs[m] != NULL)
+ if (cstate->base.defexprs[m] != NULL)
{
/* defaults contain entries for all physical attributes */
cstate->defaults[m] = true;
}
else
{
- TupleDesc tupDesc = RelationGetDescr(cstate->rel);
+ TupleDesc tupDesc = RelationGetDescr(cstate->base.rel);
Form_pg_attribute att = TupleDescAttr(tupDesc, m);
ereport(ERROR,
@@ -2019,7 +2047,7 @@ endfield:
* Read a binary attribute
*/
static Datum
-CopyReadBinaryAttribute(CopyFromState cstate, FmgrInfo *flinfo,
+CopyReadBinaryAttribute(CopyFromStateBuiltins * cstate, FmgrInfo *flinfo,
Oid typioparam, int32 typmod,
bool *isnull)
{
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
index d75a70715a4..30a1d2bff6e 100644
--- a/src/include/commands/copy.h
+++ b/src/include/commands/copy.h
@@ -48,6 +48,12 @@ typedef enum CopyLogVerbosityChoice
COPY_LOG_VERBOSITY_VERBOSE, /* logs additional messages */
} CopyLogVerbosityChoice;
+typedef struct CopyFromRowInfo
+{
+ uint64 lineno;
+ int tuplen;
+} CopyFromRowInfo;
+
/*
* A struct to hold COPY options, in a parsed form. All of these are related
* to formatting, except for 'freeze', which doesn't really belong here, but
@@ -89,8 +95,6 @@ typedef struct CopyFormatOptions
/* defined in copystate.h */
typedef struct CopyToStateData *CopyToState;
-
-/* Private in commands/copyfrom.c */
typedef struct CopyFromStateData *CopyFromState;
typedef int (*copy_data_source_cb) (void *outbuf, int minread, int maxread);
@@ -105,8 +109,8 @@ extern CopyFromState BeginCopyFrom(ParseState *pstate, Relation rel, Node *where
const char *filename,
bool is_program, copy_data_source_cb data_source_cb, List *attnamelist, List *options);
extern void EndCopyFrom(CopyFromState cstate);
-extern bool NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls);
+extern bool NextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values,
+ bool *nulls, CopyFromRowInfo * rowinfo);
extern bool NextCopyFromRawFields(CopyFromState cstate,
char ***fields, int *nfields);
extern void CopyFromErrorCallback(void *arg);
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
index 3b9982d54b8..c3d2199a0b6 100644
--- a/src/include/commands/copyapi.h
+++ b/src/include/commands/copyapi.h
@@ -65,6 +65,11 @@ typedef struct CopyToRoutine
*/
typedef struct CopyFromRoutine
{
+ /*
+ * Estimate and return the memory size required to store the state data.
+ */
+ Size (*CopyFromEstimateStateSpace) (void);
+
/*
* Set input function information. This callback is called once at the
* beginning of COPY FROM.
@@ -99,12 +104,11 @@ typedef struct CopyFromRoutine
* Returns false if there are no more tuples to read.
*/
bool (*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls);
+ Datum *values, bool *nulls, CopyFromRowInfo * rowinfo);
/*
* End a COPY FROM. This callback is called once at the end of COPY FROM.
*/
void (*CopyFromEnd) (CopyFromState cstate);
} CopyFromRoutine;
-
#endif /* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index c8b22af22d8..d2d19536151 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,105 +15,23 @@
#define COPYFROM_INTERNAL_H
#include "commands/copy.h"
-#include "commands/trigger.h"
-#include "nodes/miscnodes.h"
+#include "commands/copystate.h"
/*
- * Represents the different source cases we need to worry about at
- * the bottom level
+ * COPY FROM state data used for builtin formats.
*/
-typedef enum CopySource
+typedef struct CopyFromStateBuiltins
{
- COPY_FILE, /* from file (or a piped program) */
- COPY_FRONTEND, /* from frontend */
- COPY_CALLBACK, /* from callback function */
-} CopySource;
-
-/*
- * Represents the end-of-line terminator type of the input
- */
-typedef enum EolType
-{
- EOL_UNKNOWN,
- EOL_NL,
- EOL_CR,
- EOL_CRNL,
-} EolType;
-
-/*
- * Represents the insert method to be used during COPY FROM.
- */
-typedef enum CopyInsertMethod
-{
- CIM_SINGLE, /* use table_tuple_insert or ExecForeignInsert */
- CIM_MULTI, /* always use table_multi_insert or
- * ExecForeignBatchInsert */
- CIM_MULTI_CONDITIONAL, /* use table_multi_insert or
- * ExecForeignBatchInsert only if valid */
-} CopyInsertMethod;
-
-/*
- * This struct contains all the state variables used throughout a COPY FROM
- * operation.
- */
-typedef struct CopyFromStateData
-{
- /* 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 */
+ CopyFromStateData base;
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) */
-
- /* these are just for error messages, see CopyFromErrorCallback */
- const char *cur_relname; /* table name for error messages */
- uint64 cur_lineno; /* line number for error messages */
- const char *cur_attname; /* current att for error messages */
- const char *cur_attval; /* current att value for error messages */
- bool relname_only; /* don't output line number, att, etc. */
- /*
- * Working state
- */
- MemoryContext copycontext; /* per-copy execution context */
-
- 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
- * errors */
- int *defmap; /* array of default att numbers related to
- * missing att */
- ExprState **defexprs; /* array of default att expressions for all
- * att */
bool *defaults; /* if DEFAULT marker was found for
* corresponding att */
- bool volatile_defexprs; /* is any of defexprs volatile? */
- List *range_table; /* single element list of RangeTblEntry */
- List *rteperminfos; /* single element list of RTEPermissionInfo */
- ExprState *qualexpr;
-
- TransitionCaptureState *transition_capture;
/*
* These variables are used to reduce overhead in COPY FROM.
@@ -141,7 +59,6 @@ typedef struct CopyFromStateData
* appropriate. (In binary mode, line_buf is not used.)
*/
StringInfoData line_buf;
- bool line_buf_valid; /* contains the row being processed? */
/*
* input_buf holds input data, already converted to database encoding.
@@ -175,23 +92,22 @@ typedef struct CopyFromStateData
char *raw_buf;
int raw_buf_index; /* next byte to process */
int raw_buf_len; /* total # of bytes stored */
- 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)
-
- uint64 bytes_processed; /* number of bytes processed so far */
-} CopyFromStateData;
+} CopyFromStateBuiltins;
extern void ReceiveCopyBegin(CopyFromState cstate);
extern void ReceiveCopyBinaryHeader(CopyFromState cstate);
/* One-row callbacks for built-in formats defined in copyfromparse.c */
extern bool CopyFromTextOneRow(CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls);
+ Datum *values, bool *nulls, CopyFromRowInfo * rowinfo);
extern bool CopyFromCSVOneRow(CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls);
+ Datum *values, bool *nulls, CopyFromRowInfo * rowinfo);
extern bool CopyFromBinaryOneRow(CopyFromState cstate, ExprContext *econtext,
- Datum *values, bool *nulls);
+ Datum *values, bool *nulls, CopyFromRowInfo * rowinfo);
+
+extern Size CopyFromBuiltinsEstimateSpace(void);
#endif /* COPYFROM_INTERNAL_H */
diff --git a/src/include/commands/copystate.h b/src/include/commands/copystate.h
index 7561083a323..145dccd0f8f 100644
--- a/src/include/commands/copystate.h
+++ b/src/include/commands/copystate.h
@@ -17,6 +17,8 @@
#include "postgres.h"
#include "commands/copy.h"
#include "executor/execdesc.h"
+#include "commands/trigger.h"
+#include "nodes/miscnodes.h"
/*
* Represents the different dest cases we need to worry about at
@@ -65,4 +67,99 @@ typedef struct CopyToStateData
uint64 bytes_processed; /* number of bytes processed so far */
} CopyToStateData;
+/*
+ * Represents the different source cases we need to worry about at
+ * the bottom level
+ */
+typedef enum CopySource
+{
+ COPY_FILE, /* from file (or a piped program) */
+ COPY_FRONTEND, /* from frontend */
+ COPY_CALLBACK, /* from callback function */
+} CopySource;
+
+/*
+ * Represents the end-of-line terminator type of the input
+ */
+typedef enum EolType
+{
+ EOL_UNKNOWN,
+ EOL_NL,
+ EOL_CR,
+ EOL_CRNL,
+} EolType;
+
+/*
+ * Represents the insert method to be used during COPY FROM.
+ */
+typedef enum CopyInsertMethod
+{
+ CIM_SINGLE, /* use table_tuple_insert or ExecForeignInsert */
+ CIM_MULTI, /* always use table_multi_insert or
+ * ExecForeignBatchInsert */
+ CIM_MULTI_CONDITIONAL, /* use table_multi_insert or
+ * ExecForeignBatchInsert only if valid */
+} CopyInsertMethod;
+
+/*
+ * This struct contains all the state variables used throughout a COPY FROM
+ * operation.
+ */
+typedef struct CopyFromStateData
+{
+ /* 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 */
+ bool reached_eof; /* true if we reached EOF */
+
+ /* 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;
+ Node *whereClause; /* WHERE condition (or NULL) */
+
+ /* these are just for error messages, see CopyFromErrorCallback */
+ const char *cur_relname; /* table name for error messages */
+ uint64 cur_lineno; /* line number for error messages */
+ const char *cur_attname; /* current att for error messages */
+ const char *cur_attval; /* current att value for error messages */
+ bool relname_only; /* don't output line number, att, etc. */
+ StringInfo line_buf;
+ bool line_buf_valid;
+
+ /*
+ * Working state
+ */
+ MemoryContext copycontext; /* per-copy execution context */
+
+ 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
+ * errors */
+ AttrNumber num_defaults; /* count of att that are missing and have
+ * default value */
+ int *defmap; /* array of default att numbers related to
+ * missing att */
+ ExprState **defexprs; /* array of default att expressions for all
+ * att */
+ bool volatile_defexprs; /* is any of defexprs volatile? */
+ List *range_table; /* single element list of RangeTblEntry */
+ List *rteperminfos; /* single element list of RTEPermissionInfo */
+ ExprState *qualexpr;
+
+ TransitionCaptureState *transition_capture;
+
+ uint64 bytes_processed; /* number of bytes processed so far */
+} CopyFromStateData;
+
#endif
--
2.47.3